1use std::sync::Arc;
15use std::task::{Context, Poll};
16
17use futures_util::future::BoxFuture;
18use http::{Request, Response};
19use tonic::Status;
20use tonic::body::Body;
21use tower::{Layer, Service};
22
23use super::{AuthCtx, AuthError, CURRENT_AUTH, TokenExtractor, TokenVerifier};
24
25#[derive(Clone)]
27pub struct AuthLayer {
28 extractor: Arc<dyn TokenExtractor>,
29 verifier: Arc<dyn TokenVerifier>,
30 optional: bool,
32}
33
34impl AuthLayer {
35 pub fn new<E, V>(extractor: E, verifier: V) -> Self
36 where
37 E: TokenExtractor,
38 V: TokenVerifier,
39 {
40 Self {
41 extractor: Arc::new(extractor),
42 verifier: Arc::new(verifier),
43 optional: false,
44 }
45 }
46
47 pub fn optional(mut self) -> Self {
50 self.optional = true;
51 self
52 }
53}
54
55impl<S> Layer<S> for AuthLayer {
56 type Service = AuthService<S>;
57 fn layer(&self, inner: S) -> Self::Service {
58 AuthService {
59 inner,
60 extractor: self.extractor.clone(),
61 verifier: self.verifier.clone(),
62 optional: self.optional,
63 }
64 }
65}
66
67#[derive(Clone)]
68pub struct AuthService<S> {
69 inner: S,
70 extractor: Arc<dyn TokenExtractor>,
71 verifier: Arc<dyn TokenVerifier>,
72 optional: bool,
73}
74
75impl<S> Service<Request<Body>> for AuthService<S>
76where
77 S: Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
78 S::Error: Send + 'static,
79 S::Future: Send + 'static,
80{
81 type Response = Response<Body>;
82 type Error = S::Error;
83 type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
84
85 fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
86 self.inner.poll_ready(cx)
87 }
88
89 fn call(&mut self, mut req: Request<Body>) -> Self::Future {
90 if req.uri().path().starts_with("/grpc.health.v1.Health/") {
94 let mut inner = self.inner.clone();
95 return Box::pin(async move {
96 CURRENT_AUTH
97 .scope(AuthCtx::anonymous(), inner.call(req))
98 .await
99 });
100 }
101
102 let mut inner = self.inner.clone();
103 let extractor = self.extractor.clone();
104 let verifier = self.verifier.clone();
105 let optional = self.optional;
106
107 Box::pin(async move {
108 let metadata = metadata_from_headers(req.headers());
112
113 let ctx = match extractor.extract(&metadata) {
114 Ok(token) => match verifier.verify(&token).await {
115 Ok(ctx) => ctx,
116 Err(e) => return Ok(error_response(e)),
117 },
118 Err(AuthError::MissingToken) if optional => AuthCtx::anonymous(),
119 Err(e) => return Ok(error_response(e)),
120 };
121
122 let otel_cx = opentelemetry::Context::current();
123 let ctx = super::bridge_baggage_to_authctx(&ctx, &otel_cx);
124
125 req.extensions_mut().insert(ctx.clone());
127
128 CURRENT_AUTH.scope(ctx, inner.call(req)).await
131 })
132 }
133}
134
135fn metadata_from_headers(h: &http::HeaderMap) -> tonic::metadata::MetadataMap {
138 tonic::metadata::MetadataMap::from_headers(h.clone())
139}
140
141fn error_response(e: AuthError) -> Response<Body> {
144 let status: Status = e.into();
145 status.into_http()
146}