diff --git a/Cargo.lock b/Cargo.lock index 9a4ea7d1..66327c5c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3225,6 +3225,7 @@ version = "0.1.1" dependencies = [ "axum", "flate2", + "http", "opentelemetry", "opentelemetry-proto", "opentelemetry_sdk", @@ -3323,7 +3324,9 @@ name = "ourios-server" version = "0.1.1" dependencies = [ "axum", + "base64", "clap", + "jsonwebtoken", "opentelemetry", "opentelemetry-proto", "opentelemetry_sdk", @@ -3336,8 +3339,10 @@ dependencies = [ "ourios-semconv", "ourios-telemetry", "ourios-wal", + "p256", "proptest", "prost", + "rand 0.8.6", "rmcp", "schemars 1.2.1", "serde", diff --git a/crates/ourios-ingester/Cargo.toml b/crates/ourios-ingester/Cargo.toml index 106cb76c..31b9c87e 100644 --- a/crates/ourios-ingester/Cargo.toml +++ b/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 diff --git a/crates/ourios-ingester/src/receiver.rs b/crates/ourios-ingester/src/receiver.rs index 5b81a790..1118a4bc 100644 --- a/crates/ourios-ingester/src/receiver.rs +++ b/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}; diff --git a/crates/ourios-ingester/src/receiver/auth.rs b/crates/ourios-ingester/src/receiver/auth.rs index 461a6e68..c22a72d2 100644 --- a/crates/ourios-ingester/src/receiver/auth.rs +++ b/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>, + #[cfg(feature = "oidc")] + oidc: Option>, +} + +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>) -> 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>, + oidc: Arc, + ) -> 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, 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 diff --git a/crates/ourios-ingester/src/receiver/grpc.rs b/crates/ourios-ingester/src/receiver/grpc.rs index 52d4a4dd..7752425b 100644 --- a/crates/ourios-ingester/src/receiver/grpc.rs +++ b/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>, +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, } -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>) -> 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, 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 tower::Layer for AuthLayer { + type Service = AuthService; + + 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 { + inner: S, + resolver: AuthResolver, + metrics: Arc, +} + +impl tower::Service> for AuthService +where + S: tower::Service, Response = http::Response> + + Clone + + Send + + 'static, + S::Future: Send, + ReqBody: Send + 'static, + ResBody: Default, +{ + type Response = http::Response; + type Error = S::Error; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, mut request: http::Request) -> 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, diff --git a/crates/ourios-ingester/src/receiver/http.rs b/crates/ourios-ingester/src/receiver/http.rs index ee1dad0b..9c1a9aae 100644 --- a/crates/ourios-ingester/src/receiver/http.rs +++ b/crates/ourios-ingester/src/receiver/http.rs @@ -15,8 +15,6 @@ //! §3.1) while letting concurrent requests batch their fsyncs //! (RFC0008.8). `ingest` is async, so the handler simply `.await`s it. -use std::sync::Arc; - use axum::Router; use axum::body::Bytes; use axum::extract::{DefaultBodyLimit, State}; @@ -24,10 +22,9 @@ use axum::http::{HeaderMap, StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::routing::post; use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceResponse; -use ourios_core::auth::TokenStore; use prost::Message; -use crate::receiver::auth::authenticate_bearer; +use crate::receiver::auth::AuthResolver; use crate::receiver::decode::{decode_json, decode_protobuf}; use crate::receiver::pipeline::{ReceiveError, SharedPipeline}; @@ -40,11 +37,12 @@ pub struct HttpConfig { /// Maximum request body size in bytes; a larger body is rejected with /// 413 (RFC0003.11). pub max_body_bytes: usize, - /// The RFC 0026 token store; `None` (the default) is open mode - /// (§3.1). With a store, every request must carry a known - /// `Authorization: Bearer` credential (→ 401) and its batch is bound - /// to the token's tenant set (→ 403). - pub auth: Option>, + /// The RFC 0026 / RFC 0029 credential resolver; the default + /// (`AuthResolver::static_only(None)`) is open mode (§3.1). Otherwise + /// every request must carry a resolvable `Authorization: Bearer` + /// credential (→ 401) and its batch is bound to the resolved tenant + /// set (→ 403). + pub auth: AuthResolver, } impl Default for HttpConfig { @@ -52,7 +50,7 @@ impl Default for HttpConfig { Self { path: "/v1/logs".to_owned(), max_body_bytes: 4 * 1024 * 1024, - auth: None, + auth: AuthResolver::static_only(None), } } } @@ -65,7 +63,7 @@ impl Default for HttpConfig { struct AppState { pipeline: SharedPipeline, max_decompressed_bytes: usize, - auth: Option>, + auth: AuthResolver, } /// Build the OTLP/HTTP router over `pipeline`. @@ -101,7 +99,7 @@ async fn handle_logs(State(state): State, headers: HeaderMap, body: By let authorization = headers .get(header::AUTHORIZATION) .and_then(|value| value.to_str().ok()); - let Ok(binding) = authenticate_bearer(state.auth.as_deref(), authorization) else { + let Ok(binding) = state.auth.authenticate(authorization).await else { // RFC 0026 §3.4: the rejection counts on `ourios.ingest.batches` (`error.type = unauthenticated`). state.pipeline.record_unauthenticated(); return StatusCode::UNAUTHORIZED.into_response(); diff --git a/crates/ourios-ingester/tests/it/rfc0026_auth.rs b/crates/ourios-ingester/tests/it/rfc0026_auth.rs index 0eeb4279..0ae01272 100644 --- a/crates/ourios-ingester/tests/it/rfc0026_auth.rs +++ b/crates/ourios-ingester/tests/it/rfc0026_auth.rs @@ -20,11 +20,11 @@ use std::sync::Arc; use crate::ingest_support::{capturing_pipeline, post_request, request, resource_logs, send}; use opentelemetry_proto::tonic::collector::logs::v1::logs_service_server::LogsService; use ourios_core::auth::{TokenSpec, TokenStore, build_token_store}; -use ourios_ingester::receiver::grpc::{AuthInterceptor, LogsReceiver}; +use ourios_ingester::receiver::AuthResolver; +use ourios_ingester::receiver::grpc::{AuthLayer, LogsReceiver}; use ourios_ingester::receiver::http::{HttpConfig, router}; use ourios_ingester::receiver::{AuthBinding, ReceiveError, authenticate_bearer}; use prost::Message; -use tonic::service::Interceptor; /// A store with one token bound to `tenants`. fn store(tenants: &[&str]) -> Arc { @@ -82,7 +82,7 @@ async fn rfc0026_2_ingest_authentication() { // and nothing reaches the WAL — the journal records no append. let (pipeline, captured) = capturing_pipeline(); let config = HttpConfig { - auth: Some(store(&["checkout"])), + auth: AuthResolver::static_only(Some(store(&["checkout"]))), ..HttpConfig::default() }; for bearer in [None, Some("Bearer tok-unknown"), Some("Basic dXNlcg==")] { @@ -115,41 +115,26 @@ async fn rfc0026_2_ingest_authentication() { "the authenticated batch is appended", ); - // gRPC: the interceptor — which `LogsServiceServer::with_interceptor` - // runs before the message decode — rejects the same credentials with - // UNAUTHENTICATED, so the decoding service (and everything behind it, - // WAL included) never runs. - let mut interceptor = AuthInterceptor::new(Some(store(&["checkout"]))); - for metadata in [None, Some("Bearer tok-unknown"), Some("Basic dXNlcg==")] { - let mut request = tonic::Request::new(()); - if let Some(value) = metadata { - request - .metadata_mut() - .insert("authorization", value.parse().expect("metadata")); - } - let status = interceptor.call(request).expect_err("rejected"); - assert_eq!( - status.code(), - tonic::Code::Unauthenticated, - "{metadata:?} must be rejected", + // gRPC: the auth layer — which the server installs on the tonic + // stack — resolves before the message decode; the same credentials + // are rejected as one undifferentiated failure, so the decoding + // service (and everything behind it, WAL included) never runs. The + // served-stack arm below asserts the wire-level UNAUTHENTICATED. + let resolver = AuthResolver::static_only(Some(store(&["checkout"]))); + for authorization in [None, Some("Bearer tok-unknown"), Some("Basic dXNlcg==")] { + assert!( + resolver.authenticate(authorization).await.is_err(), + "{authorization:?} must be rejected", ); } - // A known bearer passes the interceptor and attaches the binding the - // handler enforces with. - let mut request = tonic::Request::new(()); - request - .metadata_mut() - .insert("authorization", "Bearer tok-edge".parse().expect("md")); - let passed = interceptor.call(request).expect("authenticated"); - assert_eq!( - passed - .extensions() - .get::() - .expect("binding attached") - .token_name(), - "edge-collector", - ); + // A known bearer resolves to the binding the handler enforces with. + let passed = resolver + .authenticate(Some("Bearer tok-edge")) + .await + .expect("authenticated") + .expect("bound"); + assert_eq!(passed.token_name(), "edge-collector"); } /// Scenario RFC0026.2 (served gRPC stack) — the metadata → interceptor → @@ -164,16 +149,15 @@ async fn rfc0026_2_served_grpc_stack_authenticates() { use opentelemetry_proto::tonic::collector::logs::v1::logs_service_server::LogsServiceServer; let (pipeline, captured) = capturing_pipeline(); - let service = LogsServiceServer::with_interceptor( - LogsReceiver::new(pipeline), - AuthInterceptor::new(Some(store(&["checkout"]))), - ); + let service = LogsServiceServer::new(LogsReceiver::new(pipeline)); + let layer = AuthLayer::new(AuthResolver::static_only(Some(store(&["checkout"])))); let incoming = tonic::transport::server::TcpIncoming::bind("127.0.0.1:0".parse().expect("addr")) .expect("bind"); let addr = incoming.local_addr().expect("local addr"); let server = tokio::spawn(async move { tonic::transport::Server::builder() + .layer(layer) .add_service(service) .serve_with_incoming(incoming) .await @@ -264,7 +248,7 @@ async fn rfc0026_3_ingest_tenant_binding() { router( pipeline.clone(), &HttpConfig { - auth: Some(store(&["tenant-a", "tenant-b"])), + auth: AuthResolver::static_only(Some(store(&["tenant-a", "tenant-b"]))), ..HttpConfig::default() }, ), diff --git a/crates/ourios-ingester/tests/rfc0026_telemetry.rs b/crates/ourios-ingester/tests/rfc0026_telemetry.rs index 3e8865ea..2909eaa0 100644 --- a/crates/ourios-ingester/tests/rfc0026_telemetry.rs +++ b/crates/ourios-ingester/tests/rfc0026_telemetry.rs @@ -16,9 +16,9 @@ use ingest_support::{request, resource_logs}; use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData, ResourceMetrics}; use ourios_core::audit::{AuditPayload, SharedAuditSink}; use ourios_core::auth::{TokenSpec, build_token_store}; -use ourios_ingester::receiver::grpc::AuthInterceptor; +use ourios_ingester::receiver::AuthResolver; +use ourios_ingester::receiver::grpc::AuthLayer; use ourios_ingester::receiver::{ReceiveError, authenticate_bearer}; -use tonic::service::Interceptor as _; /// The exported `ourios.ingest.batches` datapoint value for `error.type == /// wanted`, across all resource metrics. @@ -58,13 +58,33 @@ async fn rfc0026_7_rejection_telemetry_and_audit() { .expect("valid") .expect("enabled"); - // Authn rejection through the gRPC interceptor (the transport that - // owns the 401 surface): error.type = unauthenticated. - let mut interceptor = AuthInterceptor::new(Some(Arc::new(store.clone()))); - let status = interceptor - .call(tonic::Request::new(())) - .expect_err("no bearer"); - assert_eq!(status.code(), tonic::Code::Unauthenticated); + // Authn rejection through the gRPC auth layer (the transport's own + // rejection surface — UNAUTHENTICATED / grpc-status 16; HTTP's is the + // 401): error.type = unauthenticated. The layer + // answers a bearer-less request itself with a trailers-only + // UNAUTHENTICATED response — the inner service never runs. + let layer = AuthLayer::new(AuthResolver::static_only(Some(Arc::new(store.clone())))); + let service = tower::Layer::layer( + &layer, + tower::service_fn(|_req: http::Request<()>| async { + panic!("the inner service must not run for a rejected request"); + #[allow(unreachable_code)] + Ok::, std::convert::Infallible>(http::Response::new( + tonic::body::Body::default(), + )) + }), + ); + let response = tower::ServiceExt::oneshot(service, http::Request::new(())) + .await + .expect("infallible"); + assert_eq!( + response + .headers() + .get("grpc-status") + .and_then(|v| v.to_str().ok()), + Some("16"), + "trailers-only grpc-status UNAUTHENTICATED" + ); // Authz rejection through the pipeline: error.type = permission_denied // + the ingest_denied audit event. diff --git a/crates/ourios-server/Cargo.toml b/crates/ourios-server/Cargo.toml index 0f7e8fef..fce8d0d4 100644 --- a/crates/ourios-server/Cargo.toml +++ b/crates/ourios-server/Cargo.toml @@ -24,7 +24,7 @@ ourios-config = { path = "../ourios-config" } # The background compaction daemon (RFC 0009 §3.2) + the OTLP receiver # building blocks (RFC 0003 §6.2): the ingest pipeline, the axum HTTP # router, and the tonic `LogsService` impl the receiver role serves. -ourios-ingester = { path = "../ourios-ingester" } +ourios-ingester = { path = "../ourios-ingester", features = ["oidc"] } # Durable compaction-audit sink (RFC 0005 §3.7) + the compaction policy. ourios-parquet = { path = "../ourios-parquet" } # OTel SDK + OTLP push MeterProvider bootstrap (RFC 0001 §6.8). @@ -38,7 +38,7 @@ ourios-semconv = { path = "../ourios-semconv" } # over a single `Wal`, RFC 0003 §6.5 / RFC 0008 §3.1). ourios-wal = { path = "../ourios-wal" } ourios-miner = { path = "../ourios-miner" } -ourios-core = { path = "../ourios-core" } +ourios-core = { path = "../ourios-core", features = ["oidc"] } # The query engine (RFC 0007) the querier role (RFC 0016) serves: the DSL # front-ends (`dsl::parse_statement` / `parse_structured_statement`) + the # `Querier` (`run_query` / `run_drift`). Its types are kept internal — no @@ -107,6 +107,13 @@ opentelemetry_sdk = { version = "=0.32.1", default-features = false, features = # `Store::s3`. `#[ignore]`d — run only by the `s3-integration` CI job (Docker + # AWS_* env); the default `cargo test` compiles but skips them. testcontainers-modules = { version = "=0.15.0", features = ["localstack"] } +# RFC 0029 §5 served arms: a loopback fixture issuer (axum, already a lib +# dep) plus ES256 minting — the same runtime-generated-key policy as the +# ourios-core fixture (no committed private keys). +jsonwebtoken = { version = "10", default-features = false, features = ["rust_crypto", "use_pem"] } +p256 = { version = "0.13", default-features = false, features = ["ecdsa", "pkcs8", "pem", "arithmetic"] } +rand = "0.8" +base64 = "0.22" [lints] workspace = true diff --git a/crates/ourios-server/src/main.rs b/crates/ourios-server/src/main.rs index 9552848e..3995bedc 100644 --- a/crates/ourios-server/src/main.rs +++ b/crates/ourios-server/src/main.rs @@ -590,6 +590,34 @@ fn enforcement_store( .map(|auth| std::sync::Arc::new(auth.enforcement_store())) } +/// Build the ingest listeners' RFC 0026/0029 credential resolver from the +/// resolved auth config. OIDC discovery contacts the issuer once, here at +/// startup — a failure (unreachable issuer, issuer mismatch, unusable +/// JWKS) is a startup error, not a degraded mode (§3.2: with no cached +/// keys nothing could ever verify). +async fn ingest_resolver( + config: &ServerConfig, +) -> Result { + use ourios_ingester::receiver::AuthResolver; + let static_store = config + .auth + .as_ref() + .and_then(|auth| auth.static_tokens.clone()) + .map(std::sync::Arc::new); + match config.auth.as_ref().and_then(|auth| auth.oidc.clone()) { + Some(oidc) => { + let verifier = ourios_core::auth::oidc::OidcVerifier::discover(oidc) + .await + .map_err(|e| format!("auth.oidc: {e}"))?; + Ok(AuthResolver::with_oidc( + static_store, + std::sync::Arc::new(verifier), + )) + } + None => Ok(AuthResolver::static_only(static_store)), + } +} + /// Resolve the configuration (file or env, RFC 0020 §3.2) and pre-create /// a local store root (`Store::local` canonicalises it and errors on a /// missing dir; an S3 backend needs no such step — mirrors the querier @@ -651,7 +679,7 @@ async fn main() -> Result<(), Box> { // handle is cheap to share, the compactor keeps the original). store: store.clone(), promoted: config.promoted.clone(), - auth: enforcement_store(&config), + auth: ingest_resolver(&config).await?, }) .await?; println!("receiver gRPC listening on {}", handle.grpc_addr); diff --git a/crates/ourios-server/src/receiver.rs b/crates/ourios-server/src/receiver.rs index d2a25ac5..ce767d19 100644 --- a/crates/ourios-server/src/receiver.rs +++ b/crates/ourios-server/src/receiver.rs @@ -20,7 +20,8 @@ use opentelemetry_proto::tonic::collector::logs::v1::logs_service_server::LogsSe use ourios_config::MinerConfig; use ourios_ingester::audit_sink::{BufferingAuditSink, SharedParquetAuditSink}; use ourios_ingester::publish::PublishCoordinator; -use ourios_ingester::receiver::grpc::{AuthInterceptor, LogsReceiver}; +use ourios_ingester::receiver::AuthResolver; +use ourios_ingester::receiver::grpc::{AuthLayer, LogsReceiver}; use ourios_ingester::receiver::http::{HttpConfig, router}; use ourios_ingester::receiver::pipeline::RotationHook; use ourios_ingester::receiver::{CommitCoordinator, IngestPipeline, SharedPipeline, TenantRule}; @@ -211,11 +212,12 @@ pub struct ReceiverConfig { /// The RFC 0022 promoted attribute set every flushed data file projects /// (`storage.promoted_attributes`, §3.2). pub promoted: PromotedAttributes, - /// The RFC 0026 token store; `None` is open mode (§3.1). Applied to - /// both listeners: the gRPC interceptor and the HTTP handler - /// authenticate before decode, and the pipeline binds each batch to - /// the token's tenant set before the WAL append (§3.2). - pub auth: Option>, + /// The RFC 0026 / RFC 0029 credential resolver (static store and/or + /// OIDC verifier; `AuthResolver::static_only(None)` is open mode, + /// §3.1). Applied to both listeners: the gRPC auth layer and the HTTP + /// handler authenticate before decode, and the pipeline binds each + /// batch to the resolved tenant set before the WAL append (§3.2). + pub auth: AuthResolver, } /// A running receiver role: the **resolved** bound addresses (so a `:0` @@ -472,17 +474,18 @@ pub async fn serve(config: ReceiverConfig) -> Result { shutdown_rx.clone(), ); - // RFC 0026 §3.2: the interceptor authenticates before the message - // decode (open mode passes through unbound); the handler's pipeline - // enforces the tenant binding it attaches. - let grpc_service = LogsServiceServer::with_interceptor( - LogsReceiver::new(pipeline.clone()), - AuthInterceptor::new(config.auth.clone()), - ); + // RFC 0026 §3.2 / RFC 0029 §3.3: the auth layer resolves before the + // message decode (open mode passes through unbound); the handler's + // pipeline enforces the tenant binding it attaches. A tower layer + // rather than a sync interceptor because OIDC resolution may await a + // JWKS refetch. + let grpc_service = LogsServiceServer::new(LogsReceiver::new(pipeline.clone())); + let auth_layer = AuthLayer::new(config.auth.clone()); let grpc = tokio::spawn({ let mut rx = shutdown_rx.clone(); async move { Server::builder() + .layer(auth_layer) .add_service(grpc_service) .serve_with_incoming_shutdown(grpc_incoming, async move { let _ = rx.changed().await; @@ -746,7 +749,7 @@ mod tests { wal: test_wal_config(wal_dir.path()), store, promoted: PromotedAttributes::default(), - auth: None, + auth: AuthResolver::static_only(None), }) .await .expect("serve"); @@ -865,7 +868,7 @@ mod tests { wal: test_wal_config(wal_dir.path()), store, promoted: PromotedAttributes::default(), - auth: None, + auth: AuthResolver::static_only(None), }) .await .expect("serve"); diff --git a/crates/ourios-server/tests/it/rfc0029_oidc.rs b/crates/ourios-server/tests/it/rfc0029_oidc.rs index cdc116dd..65750d02 100644 --- a/crates/ourios-server/tests/it/rfc0029_oidc.rs +++ b/crates/ourios-server/tests/it/rfc0029_oidc.rs @@ -235,3 +235,298 @@ fn rfc0029_7_dex_end_to_end() { value; no JWT material on any surface" ); } + +// --- RFC 0029 §3.3 ingest-binding arms (the verifier + tower-layer slice): +// a loopback fixture issuer, ES256 minting with a runtime-generated key +// (the ourios-core fixture policy — no committed private keys), and the +// spawned binary enforcing OIDC-resolved bindings on both listeners. + +mod ingest_binding { + use std::io::Write as _; + use std::time::Duration; + + use base64::Engine as _; + use base64::engine::general_purpose::URL_SAFE_NO_PAD; + use jsonwebtoken::EncodingKey; + use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; + use opentelemetry_proto::tonic::collector::logs::v1::logs_service_client::LogsServiceClient; + use opentelemetry_proto::tonic::common::v1::any_value::Value; + use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue}; + use opentelemetry_proto::tonic::logs::v1::{LogRecord, ResourceLogs, ScopeLogs}; + use opentelemetry_proto::tonic::resource::v1::Resource; + use p256::ecdsa::SigningKey; + use p256::pkcs8::EncodePrivateKey as _; + use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; + use tokio::process::Command; + use tokio::time::timeout; + + /// A fresh ES256 keypair (runtime-generated) and its public JWK. + fn make_key(kid: &str) -> (EncodingKey, serde_json::Value) { + let signing = SigningKey::random(&mut rand::rngs::OsRng); + let pem = signing + .to_pkcs8_pem(p256::pkcs8::LineEnding::LF) + .expect("pkcs8 pem"); + let encoding = EncodingKey::from_ec_pem(pem.as_bytes()).expect("encoding key"); + let point = signing.verifying_key().to_encoded_point(false); + let jwk = serde_json::json!({ + "kty": "EC", "crv": "P-256", "use": "sig", "alg": "ES256", "kid": kid, + "x": URL_SAFE_NO_PAD.encode(point.x().expect("x")), + "y": URL_SAFE_NO_PAD.encode(point.y().expect("y")), + }); + (encoding, jwk) + } + + /// A loopback issuer serving discovery + a fixed JWKS. + async fn serve_issuer(jwk: serde_json::Value) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fixture issuer"); + let issuer = format!("http://{}", listener.local_addr().expect("addr")); + let discovery = serde_json::json!({ + "issuer": issuer, + "jwks_uri": format!("{issuer}/jwks"), + }); + let jwks = serde_json::json!({ "keys": [jwk] }); + // Plain-string JSON responses: the server's axum feature set has + // no `json` (the fixture doesn't need it). + let json = |body: String| ([("content-type", "application/json")], body); + let app = axum::Router::new() + .route( + "/.well-known/openid-configuration", + axum::routing::get({ + let discovery = discovery.to_string(); + move || async move { json(discovery) } + }), + ) + .route( + "/jwks", + axum::routing::get({ + let jwks = jwks.to_string(); + move || async move { json(jwks) } + }), + ); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve issuer"); + }); + issuer + } + + fn mint(encoding: &EncodingKey, kid: &str, issuer: &str, tenants: &[&str]) -> String { + let now = i64::try_from( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("epoch") + .as_secs(), + ) + .expect("fits"); + let claims = serde_json::json!({ + "iss": issuer, "aud": "ourios", "exp": now + 600, + "sub": "edge-collector", "ourios_tenants": tenants, + }); + let mut header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256); + header.kid = Some(kid.to_string()); + jsonwebtoken::encode(&header, &claims, encoding).expect("mint") + } + + /// One `ResourceLogs` batch whose tenant derives from `service.name`. + fn batch(tenant: &str) -> ExportLogsServiceRequest { + ExportLogsServiceRequest { + resource_logs: vec![ResourceLogs { + resource: Some(Resource { + attributes: vec![KeyValue { + key: "service.name".to_string(), + value: Some(AnyValue { + value: Some(Value::StringValue(tenant.to_string())), + }), + ..Default::default() + }], + ..Default::default() + }), + scope_logs: vec![ScopeLogs { + log_records: vec![LogRecord { + time_unix_nano: 1_775_127_480_000_000_000, + body: Some(AnyValue { + value: Some(Value::StringValue("user logged in".to_string())), + }), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }], + } + } + + /// The §3.3 ingest binding on the served binary: startup discovery + /// against the fixture issuer; a bearer-less gRPC export is + /// UNAUTHENTICATED before decode; a verified JWT ingests within its + /// tenant claim and is whole-batch denied outside it; the HTTP + /// listener 401s a bearer-less POST through the same resolver. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn oidc_ingest_binding_enforces_on_the_served_binary() { + let (encoding, jwk) = make_key("key-1"); + let issuer = serve_issuer(jwk).await; + + let tmp = tempfile::TempDir::new().expect("temp"); + let wal = tmp.path().join("wal"); + std::fs::create_dir_all(&wal).expect("wal dir"); + let config_path = tmp.path().join("ourios.yaml"); + let mut file = std::fs::File::create(&config_path).expect("create config"); + write!( + file, + "storage:\n local:\n bucket_root: {}\n\ + receiver:\n enabled: true\n grpc_addr: 127.0.0.1:0\n http_addr: 127.0.0.1:0\n wal_root: {}\n\ + auth:\n oidc:\n issuer: {}\n audience: ourios\n tenant_claim: ourios_tenants\n", + tmp.path().display(), + wal.display(), + issuer, + ) + .expect("write config"); + + let mut child = Command::new(env!("CARGO_BIN_EXE_ourios-server")) + .arg("--config") + .arg(&config_path) + .env("RUST_LOG", "info") + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::null()) + .kill_on_drop(true) + .spawn() + .expect("spawn ourios-server"); + + let stdout = child.stdout.take().expect("stdout piped"); + let mut out_lines = BufReader::new(stdout).lines(); + let (grpc_addr, http_addr) = timeout(Duration::from_secs(15), async { + let mut grpc = None; + let mut http = None; + while let Some(line) = out_lines.next_line().await.expect("read stdout") { + if let Some(addr) = line.strip_prefix("receiver gRPC listening on ") { + grpc = Some(addr.trim().to_string()); + } + if let Some(addr) = line.strip_prefix("receiver HTTP listening on ") { + http = Some(addr.trim().to_string()); + } + if let (Some(g), Some(h)) = (&grpc, &http) { + return (g.clone(), h.clone()); + } + } + panic!("receiver lines never appeared — discovery must succeed at startup"); + }) + .await + .expect("server ready before timeout"); + + let mut client = LogsServiceClient::connect(format!("http://{grpc_addr}")) + .await + .expect("grpc connect"); + + // Bearer-less: UNAUTHENTICATED from the auth layer, pre-decode. + let status = client + .export(tonic::Request::new(batch("acme"))) + .await + .expect_err("no bearer is rejected"); + assert_eq!(status.code(), tonic::Code::Unauthenticated); + + // A verified JWT bound to ["acme"]: in-set ingests… + let token = mint(&encoding, "key-1", &issuer, &["acme"]); + let authorization: tonic::metadata::MetadataValue<_> = + format!("Bearer {token}").parse().expect("metadata"); + let mut request = tonic::Request::new(batch("acme")); + request + .metadata_mut() + .insert("authorization", authorization.clone()); + client.export(request).await.expect("in-set batch acks"); + + // …and an out-of-set tenant is whole-batch denied (§3.2 — + // RFC 0026 semantics with the OIDC-resolved binding). + let mut request = tonic::Request::new(batch("globex")); + request + .metadata_mut() + .insert("authorization", authorization); + let status = client + .export(request) + .await + .expect_err("out-of-set tenant is denied"); + assert_eq!(status.code(), tonic::Code::PermissionDenied); + + // The HTTP listener runs the same resolver: bearer-less POST → 401. + let mut stream = tokio::net::TcpStream::connect(&http_addr) + .await + .expect("http connect"); + stream + .write_all( + b"POST /v1/logs HTTP/1.1\r\nHost: 127.0.0.1\r\n\ + Content-Type: application/json\r\nContent-Length: 2\r\n\ + Connection: close\r\n\r\n{}", + ) + .await + .expect("write request"); + let mut response = String::new(); + timeout( + Duration::from_secs(15), + stream.read_to_string(&mut response), + ) + .await + .expect("response before timeout") + .expect("read response"); + assert!( + response.starts_with("HTTP/1.1 401 "), + "bearer-less HTTP ingest is 401: {response}", + ); + + child.kill().await.expect("kill the server"); + } + + /// §3.2: OIDC discovery failure is a startup error, not a degraded + /// mode — a receiver configured against an unreachable issuer exits + /// nonzero naming `auth.oidc`. + #[tokio::test] + async fn oidc_unreachable_issuer_fails_receiver_startup() { + // A held loopback port that deterministically fails every + // connection (accept-then-close). Holding the listener for the + // test's lifetime avoids the race where a dropped port is re-bound + // by another local process before the child runs discovery; no + // DNS or egress dependency either way. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let unreachable = format!("http://{}", listener.local_addr().expect("addr")); + tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + break; + }; + drop(stream); + } + }); + let tmp = tempfile::TempDir::new().expect("temp"); + let wal = tmp.path().join("wal"); + std::fs::create_dir_all(&wal).expect("wal dir"); + let config_path = tmp.path().join("ourios.yaml"); + let mut file = std::fs::File::create(&config_path).expect("create config"); + write!( + file, + "storage:\n local:\n bucket_root: {}\n\ + receiver:\n enabled: true\n grpc_addr: 127.0.0.1:0\n http_addr: 127.0.0.1:0\n wal_root: {}\n\ + auth:\n oidc:\n issuer: {}\n audience: ourios\n tenant_claim: ourios_tenants\n", + tmp.path().display(), + wal.display(), + unreachable, + ) + .expect("write config"); + + let output = Command::new(env!("CARGO_BIN_EXE_ourios-server")) + .arg("--config") + .arg(&config_path) + .output() + .await + .expect("run ourios-server"); + assert!( + !output.status.success(), + "an unreachable issuer must fail startup" + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("auth.oidc"), + "names the failing key: {stderr}" + ); + } +} diff --git a/deny.toml b/deny.toml index dc77b4b1..f69a5baa 100644 --- a/deny.toml +++ b/deny.toml @@ -56,7 +56,14 @@ allow = [ "CC0-1.0", ] confidence-threshold = 0.8 -exceptions = [] +exceptions = [ + # CDLA-Permissive-2.0 covers the Mozilla CA-certificate *data* bundled by + # webpki-roots (the standard rustls root store, via reqwest's rustls-tls — + # the RFC 0029 verifier's HTTPS to the issuer). A permissive data license + # (attribution-free redistribution); scoped to this one crate rather than + # allowed globally. + { crate = "webpki-roots", allow = ["CDLA-Permissive-2.0"] }, +] # The allow-list is a stable permissive baseline; the set of licenses actually # encountered varies by target/platform, so don't warn on allowed-but-unused # entries (they are not policy violations).