micromegas_auth/tower.rs
1//! Tower service layer for async authentication with tonic/gRPC.
2//!
3//! This module provides a tower service wrapper that integrates authentication
4//! into tonic gRPC services. It extracts request parts from gRPC metadata,
5//! validates them using an AuthProvider, and injects the AuthContext into request extensions.
6
7use crate::client_ip::resolve_client_ip;
8use crate::types::{AuthProvider, GrpcRequestParts, ProviderUnavailable, RequestParts};
9use futures::future::BoxFuture;
10use micromegas_tracing::prelude::*;
11use std::sync::Arc;
12use tonic::Status;
13use tower::Service;
14
15/// Async authentication service wrapper for tonic/gRPC.
16///
17/// This service wraps another tower service and adds authentication:
18/// 1. Extracts request parts from gRPC metadata
19/// 2. Validates request using the configured AuthProvider
20/// 3. Injects AuthContext into request extensions
21/// 4. Logs authentication success/failure
22///
23/// If no auth_provider is configured, requests pass through without authentication.
24///
25/// # Example
26///
27/// ```rust,no_run
28/// use micromegas_auth::api_key::{ApiKeyAuthProvider, parse_key_ring};
29/// use micromegas_auth::tower::AuthService;
30/// use std::sync::Arc;
31///
32/// # async fn example<S>(inner_service: S) -> anyhow::Result<()>
33/// # where
34/// # S: tower::Service<http::Request<tonic::body::Body>> + Clone + Send + 'static,
35/// # S::Response: 'static,
36/// # S::Future: Send + 'static,
37/// # S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
38/// # {
39/// // Create auth provider
40/// let keyring = parse_key_ring(r#"[{"name": "test", "key": "secret"}]"#)?;
41/// let auth_provider = Arc::new(ApiKeyAuthProvider::new(keyring));
42///
43/// // Wrap your service with authentication
44/// let auth_service = AuthService {
45/// inner: inner_service,
46/// auth_provider: Some(auth_provider as Arc<dyn micromegas_auth::types::AuthProvider>),
47/// };
48/// # Ok(())
49/// # }
50/// ```
51#[derive(Clone)]
52pub struct AuthService<S> {
53 /// The inner service to wrap
54 pub inner: S,
55 /// Optional authentication provider (None = no auth required)
56 pub auth_provider: Option<Arc<dyn AuthProvider>>,
57}
58
59impl<S> Service<http::Request<tonic::body::Body>> for AuthService<S>
60where
61 S: Service<http::Request<tonic::body::Body>> + Clone + Send + 'static,
62 S::Response: 'static,
63 S::Future: Send + 'static,
64 S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
65{
66 type Response = S::Response;
67 type Error = Box<dyn std::error::Error + Send + Sync>;
68 type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
69
70 fn poll_ready(
71 &mut self,
72 cx: &mut std::task::Context<'_>,
73 ) -> std::task::Poll<Result<(), Self::Error>> {
74 self.inner.poll_ready(cx).map_err(Into::into)
75 }
76
77 fn call(&mut self, req: http::Request<tonic::body::Body>) -> Self::Future {
78 let clone = self.inner.clone();
79 let mut inner = std::mem::replace(&mut self.inner, clone);
80 let auth_provider = self.auth_provider.clone();
81
82 Box::pin(async move {
83 if let Some(provider) = auth_provider {
84 let (mut parts, body) = req.into_parts();
85
86 // Resolved right after `req.into_parts()`, before headers are cloned into
87 // `GrpcRequestParts` -- for gRPC, the `SocketAddr` extension is already present
88 // on `parts.extensions` by the time `AuthService::call` runs
89 // (`ConnectedIncoming`/`ConnectedStream` attach it at TCP-accept time, below
90 // every custom tower layer).
91 let client_ip = resolve_client_ip(&parts.headers, &parts.extensions);
92
93 // Extract request parts for validation
94 let request_parts = GrpcRequestParts {
95 metadata: tonic::metadata::MetadataMap::from_headers(parts.headers.clone()),
96 client_ip,
97 };
98
99 // Validate request
100 match provider
101 .validate_request(&request_parts as &dyn RequestParts)
102 .await
103 {
104 Ok(auth_ctx) => {
105 info!(
106 "authenticated: subject={} email={:?} issuer={} audience={:?} admin={}",
107 auth_ctx.subject,
108 auth_ctx.email,
109 auth_ctx.issuer,
110 auth_ctx.audience,
111 auth_ctx.is_admin()
112 );
113
114 // SECURITY: Remove any client-provided auth headers to prevent spoofing
115 // These headers are only set by the authentication layer
116 parts.headers.remove("x-auth-subject");
117 parts.headers.remove("x-auth-email");
118 parts.headers.remove("x-auth-issuer");
119 parts.headers.remove("x-allow-delegation");
120 parts.headers.remove("x-auth-is-admin");
121
122 // Inject auth context into gRPC metadata headers
123 parts.headers.insert(
124 "x-auth-subject",
125 http::HeaderValue::from_str(&auth_ctx.subject)
126 .expect("valid user id header"),
127 );
128 if let Some(email) = &auth_ctx.email {
129 parts.headers.insert(
130 "x-auth-email",
131 http::HeaderValue::from_str(email).expect("valid email header"),
132 );
133 }
134 parts.headers.insert(
135 "x-auth-issuer",
136 http::HeaderValue::from_str(&auth_ctx.issuer)
137 .expect("valid issuer header"),
138 );
139 parts.headers.insert(
140 "x-allow-delegation",
141 http::HeaderValue::from_str(&auth_ctx.allow_delegation.to_string())
142 .expect("valid allow_delegation header"),
143 );
144 parts.headers.insert(
145 "x-auth-is-admin",
146 http::HeaderValue::from_str(&auth_ctx.is_admin().to_string())
147 .expect("valid is_admin header"),
148 );
149
150 parts.extensions.insert(auth_ctx);
151 let req = http::Request::from_parts(parts, body);
152 inner.call(req).await.map_err(Into::into)
153 }
154 Err(e) => {
155 let status = if e.downcast_ref::<ProviderUnavailable>().is_some() {
156 debug!("authentication failed: {e}");
157 Status::unavailable("auth provider unavailable")
158 } else {
159 warn!("authentication failed: {e}");
160 Status::unauthenticated("invalid token")
161 };
162 Err(Box::new(status) as Box<dyn std::error::Error + Send + Sync>)
163 }
164 }
165 } else {
166 inner.call(req).await.map_err(Into::into)
167 }
168 })
169 }
170}