feat(auth): rfc 0029 green (ingest binding) — async auth layer on both listeners

crates/ourios-ingester/Cargo.toml

@@ -85,6 +85,10 @@ serde_json = { version = "1", default-features = false, features = ["std"] }
 # (`State`, `Bytes`, `HeaderMap`) are core. On hyper 1.x, already in the
 # tree.
 axum = { version = "0.8", default-features = false, features = ["tokio", "http1"] }
+# The RFC 0029 §3.3 async auth gate on the gRPC listener is a tower
+# Layer/Service (the sync tonic interceptor cannot await a JWKS refetch).
+tower = { version = "0.5", default-features = false }
+http = { version = "1", default-features = false, features = ["std"] }
 # gzip decode for OTLP/HTTP `Content-Encoding: gzip` (an OTLP MUST,
 # RFC0003.13). Pure-Rust backend — no zlib C dependency (CLAUDE.md).
 flate2 = { version = "1", default-features = false, features = ["rust_backend"] }
@@ -98,6 +102,10 @@ tonic = { version = "0.14", default-features = false, features = ["codegen"] }
 # §6.6). Same major as ourios-wal's, so the types unify.
 uuid = { version = "1", default-features = false, features = ["std", "v7"] }
 
+[features]
+# RFC 0029: OIDC bearer resolution in front of the RFC 0026 enforcement.
+oidc = ["ourios-core/oidc"]
+
 [dev-dependencies]
 # The RFC0026.2 served-stack arm: serve `LogsServiceServer::with_interceptor`
 # over a local socket and drive it with a real tonic client, so the

crates/ourios-ingester/src/receiver.rs

@@ -40,7 +40,7 @@ pub mod materialize;
 pub mod pipeline;
 pub mod tenant;
 
-pub use auth::{AuthBinding, Unauthenticated, authenticate_bearer};
+pub use auth::{AuthBinding, AuthResolver, Unauthenticated, authenticate_bearer};
 pub use commit::CommitCoordinator;
 pub use decode::{DecodeError, decode_json, decode_protobuf};
 pub use materialize::{materialize_record, materialize_resource_logs};

crates/ourios-ingester/src/receiver/auth.rs

@@ -16,6 +16,8 @@
 //! Nothing here carries or renders a token value: the binding holds the
 //! token's audit *name* and its tenant set only.
 
+use std::sync::Arc;
+
 use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
 use ourios_core::auth::{TenantSet, TokenStore};
 
@@ -93,6 +95,109 @@ fn parse_bearer(value: &str) -> Option<&str> {
     (!token.is_empty()).then_some(token)
 }
 
+/// The full request resolution in front of the RFC 0026 enforcement:
+/// the constant-time static store first, then — when configured and the
+/// static store does not match — RFC 0029 OIDC verification (`oidc`
+/// feature). Open mode (§3.1) only when *nothing* is configured. Async
+/// because an OIDC unseen-`kid` miss may refetch the JWKS; the static
+/// path never awaits.
+#[derive(Clone)]
+pub struct AuthResolver {
+    store: Option<Arc<TokenStore>>,
+    #[cfg(feature = "oidc")]
+    oidc: Option<Arc<ourios_core::auth::oidc::OidcVerifier>>,
+}
+
+impl std::fmt::Debug for AuthResolver {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let mut d = f.debug_struct("AuthResolver");
+        d.field("static_store", &self.store.is_some());
+        #[cfg(feature = "oidc")]
+        d.field("oidc", &self.oidc.is_some());
+        d.finish()
+    }
+}
+
+impl AuthResolver {
+    /// A resolver over the static store only (`None` = open mode) — the
+    /// RFC 0026 shape, and the whole story when the `oidc` feature is
+    /// off or unconfigured.
+    #[must_use]
+    pub fn static_only(store: Option<Arc<TokenStore>>) -> Self {
+        Self {
+            store,
+            #[cfg(feature = "oidc")]
+            oidc: None,
+        }
+    }
+
+    /// A resolver with an OIDC verifier alongside the (optional) static
+    /// store — RFC 0029 §3.3 coexistence: each credential authenticates
+    /// via its own path, carrying its own tenant binding.
+    #[cfg(feature = "oidc")]
+    #[must_use]
+    pub fn with_oidc(
+        store: Option<Arc<TokenStore>>,
+        oidc: Arc<ourios_core::auth::oidc::OidcVerifier>,
+    ) -> Self {
+        Self {
+            store,
+            oidc: Some(oidc),
+        }
+    }
+
+    /// Whether every request passes unbound (§3.1 open mode).
+    #[must_use]
+    pub fn is_open(&self) -> bool {
+        #[cfg(feature = "oidc")]
+        {
+            self.store.is_none() && self.oidc.is_none()
+        }
+        #[cfg(not(feature = "oidc"))]
+        {
+            self.store.is_none()
+        }
+    }
+
+    /// Resolve a request's `Authorization` value (RFC 0026 §3.2 /
+    /// RFC 0029 §3.3). Same contract as [`authenticate_bearer`]:
+    /// `Ok(None)` in open mode, one undifferentiated error otherwise.
+    ///
+    /// # Errors
+    ///
+    /// [`Unauthenticated`] on a missing, malformed, or unknown credential
+    /// — including a JWT that fails verification.
+    pub async fn authenticate(
+        &self,
+        authorization: Option<&str>,
+    ) -> Result<Option<AuthBinding>, Unauthenticated> {
+        if self.is_open() {
+            return Ok(None);
+        }
+        let token = authorization
+            .and_then(parse_bearer)
+            .ok_or(Unauthenticated)?;
+        if let Some(store) = self.store.as_deref()
+            && let Some(entry) = store.authenticate(token)
+        {
+            return Ok(Some(AuthBinding {
+                token_name: entry.name().to_string(),
+                tenants: entry.tenants().clone(),
+            }));
+        }
+        #[cfg(feature = "oidc")]
+        if let Some(oidc) = &self.oidc
+            && let Some(identity) = oidc.verify(token).await
+        {
+            return Ok(Some(AuthBinding {
+                token_name: identity.name,
+                tenants: identity.tenants,
+            }));
+        }
+        Err(Unauthenticated)
+    }
+}
+
 /// Enforce the §3.2 per-batch tenant binding: derive every `ResourceLogs`
 /// group's tenant (the same rule and error surface as the fan-out — RFC
 /// 0003 §6.3 derivation is unchanged) and require each to fall inside the

crates/ourios-ingester/src/receiver/grpc.rs

@@ -18,70 +18,122 @@
 //! coordinator — RFC0008.8 — which offloads the blocking `sync` itself),
 //! so the handler simply `.await`s it; the handler never panics.
 
+use std::future::Future;
+use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};
 
 use opentelemetry_proto::tonic::collector::logs::v1::logs_service_server::LogsService;
 use opentelemetry_proto::tonic::collector::logs::v1::{
     ExportLogsServiceRequest, ExportLogsServiceResponse,
 };
-use ourios_core::auth::TokenStore;
 use tonic::{Request, Response, Status};
 
-use crate::receiver::auth::{AuthBinding, authenticate_bearer};
+use crate::receiver::auth::{AuthBinding, AuthResolver};
 use crate::receiver::pipeline::{ReceiveError, SharedPipeline};
 
-/// The RFC 0026 §3.2 authentication gate for the gRPC listener, installed
-/// via `LogsServiceServer::with_interceptor`. It runs **before the message
-/// decode** (an interceptor sees only metadata), rejecting a missing or
-/// unknown bearer with `UNAUTHENTICATED`; on success it attaches the
-/// resolved [`AuthBinding`] as a request extension for the handler's
-/// tenant-binding check. With no store configured it passes every request
-/// through unbound (open mode, §3.1) — one service type either way.
+/// The authentication gate for the gRPC listener (RFC 0026 §3.2 /
+/// RFC 0029 §3.3), applied as a tower layer on the tonic server. A tower
+/// service (unlike a sync tonic interceptor) can await the resolver — an
+/// OIDC unseen-`kid` miss refetches the JWKS — while still running
+/// **before the message decode** (it sees only the HTTP envelope). A
+/// rejection is a trailers-only `UNAUTHENTICATED` (grpc-status 16)
+/// response; on success the resolved [`AuthBinding`] rides the request
+/// extensions into the handler's tenant-binding check. With nothing
+/// configured every request passes through unbound (open mode, §3.1).
 #[derive(Clone)]
-pub struct AuthInterceptor {
-    store: Option<Arc<TokenStore>>,
+pub struct AuthLayer {
+    resolver: AuthResolver,
     /// Rejection telemetry (RFC 0026 §3.4). The instruments resolve by
     /// name through the global meter, so this instance aggregates with
     /// the pipeline's.
     metrics: Arc<crate::metrics::IngestMetrics>,
 }
 
-impl AuthInterceptor {
-    /// An interceptor over `store` (`None` = open mode pass-through).
+impl AuthLayer {
+    /// A layer over `resolver` (see [`AuthResolver`] for open mode).
     #[must_use]
-    pub fn new(store: Option<Arc<TokenStore>>) -> Self {
+    pub fn new(resolver: AuthResolver) -> Self {
         Self {
-            store,
+            resolver,
             metrics: Arc::new(crate::metrics::IngestMetrics::new()),
         }
     }
 }
 
-impl tonic::service::Interceptor for AuthInterceptor {
-    fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
-        let authorization = request
-            .metadata()
-            .get("authorization")
-            .and_then(|value| value.to_str().ok());
-        match authenticate_bearer(self.store.as_deref(), authorization) {
-            Ok(None) => Ok(request),
-            Ok(Some(binding)) => {
-                request.extensions_mut().insert(binding);
-                Ok(request)
-            }
-            // One undifferentiated message: missing vs malformed vs unknown
-            // would be a probing oracle (RFC 0026 §3.2). §3.4: the
-            // rejection counts on `ourios.ingest.batches`
-            // (`error.type = unauthenticated`).
-            Err(_) => {
-                self.metrics
-                    .record_rejected_batch(crate::metrics::ERROR_TYPE_UNAUTHENTICATED);
-                Err(Status::unauthenticated("a valid bearer token is required"))
-            }
+impl<S> tower::Layer<S> for AuthLayer {
+    type Service = AuthService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        AuthService {
+            inner,
+            resolver: self.resolver.clone(),
+            metrics: Arc::clone(&self.metrics),
         }
     }
 }
 
+/// The [`AuthLayer`] service: authenticate, then delegate.
+#[derive(Clone)]
+pub struct AuthService<S> {
+    inner: S,
+    resolver: AuthResolver,
+    metrics: Arc<crate::metrics::IngestMetrics>,
+}
+
+impl<S, ReqBody, ResBody> tower::Service<http::Request<ReqBody>> for AuthService<S>
+where
+    S: tower::Service<http::Request<ReqBody>, Response = http::Response<ResBody>>
+        + Clone
+        + Send
+        + 'static,
+    S::Future: Send,
+    ReqBody: Send + 'static,
+    ResBody: Default,
+{
+    type Response = http::Response<ResBody>;
+    type Error = S::Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut request: http::Request<ReqBody>) -> Self::Future {
+        // The tower readiness dance: `poll_ready` reserved capacity on
+        // `self.inner`, so that instance (not a fresh clone) must serve
+        // this call; the clone waits for its own `poll_ready` next time.
+        let clone = self.inner.clone();
+        let mut inner = std::mem::replace(&mut self.inner, clone);
+        let resolver = self.resolver.clone();
+        let metrics = Arc::clone(&self.metrics);
+        Box::pin(async move {
+            let authorization = request
+                .headers()
+                .get(http::header::AUTHORIZATION)
+                .and_then(|value| value.to_str().ok())
+                .map(str::to_owned);
+            match resolver.authenticate(authorization.as_deref()).await {
+                Ok(None) => {}
+                Ok(Some(binding)) => {
+                    request.extensions_mut().insert(binding);
+                }
+                // One undifferentiated message: missing vs malformed vs
+                // unknown would be a probing oracle (RFC 0026 §3.2). §3.4:
+                // the rejection counts on `ourios.ingest.batches`
+                // (`error.type = unauthenticated`).
+                Err(_) => {
+                    metrics.record_rejected_batch(crate::metrics::ERROR_TYPE_UNAUTHENTICATED);
+                    return Ok(
+                        Status::unauthenticated("a valid bearer token is required").into_http()
+                    );
+                }
+            }
+            inner.call(request).await
+        })
+    }
+}
+
 /// The gRPC `LogsService` over a shared `IngestPipeline`.
 pub struct LogsReceiver {
     pipeline: SharedPipeline,