diff --git a/crates/openshell-server/src/grpc/provider.rs b/crates/openshell-server/src/grpc/provider.rs index cc2395fe7d..fb22dc698d 100644 --- a/crates/openshell-server/src/grpc/provider.rs +++ b/crates/openshell-server/src/grpc/provider.rs @@ -2127,6 +2127,7 @@ fn provider_credential_config_key_collision( struct DynamicTokenGrantBinding { provider_name: String, credential_name: String, + header_name: String, host: String, port: u32, path: String, @@ -2182,7 +2183,7 @@ fn push_dynamic_token_grant_bindings_for_endpoint( push_dynamic_token_grant_binding( bindings, provider_name, - &credential.name, + credential, endpoint_host, endpoint_port, endpoint_path, @@ -2214,7 +2215,7 @@ fn push_dynamic_token_grant_bindings_for_endpoint( push_dynamic_token_grant_binding( bindings, provider_name, - &credential.name, + credential, override_host, override_port, override_path, @@ -2225,14 +2226,21 @@ fn push_dynamic_token_grant_bindings_for_endpoint( fn push_dynamic_token_grant_binding( bindings: &mut Vec, provider_name: &str, - credential_name: &str, + credential: &ProviderProfileCredential, host: &str, port: u32, path: &str, ) { let candidate = DynamicTokenGrantBinding { provider_name: provider_name.to_string(), - credential_name: credential_name.to_string(), + credential_name: credential.name.clone(), + // The supervisor selects one grant per case-insensitive header, using + // Authorization when bearer placement omits an explicit destination. + header_name: if credential.header_name.trim().is_empty() { + "authorization".to_string() + } else { + credential.header_name.trim().to_ascii_lowercase() + }, host: host.to_ascii_lowercase(), port, path: path.to_string(), @@ -2253,7 +2261,8 @@ fn validate_dynamic_token_grant_bindings_unambiguous( { continue; } - if first.port == second.port + if first.header_name == second.header_name + && first.port == second.port && first.score == second.score && host_patterns_can_overlap(&first.host, &second.host) && path_patterns_can_overlap(&first.path, &second.path) @@ -5456,9 +5465,28 @@ mod tests { host: &str, port: u32, path: &str, + ) { + import_token_grant_profile_with_credentials( + state, + id, + host, + port, + path, + vec![token_grant_credential("access_token")], + ) + .await; + } + + async fn import_token_grant_profile_with_credentials( + state: &Arc, + id: &str, + host: &str, + port: u32, + path: &str, + credentials: Vec, ) { let mut profile = custom_profile(id); - profile.credentials = vec![token_grant_credential("access_token")]; + profile.credentials = credentials; profile.endpoints = vec![NetworkEndpoint { host: host.to_string(), port, @@ -5515,6 +5543,102 @@ mod tests { .unwrap() } + #[tokio::test] + async fn dynamic_token_grants_allow_distinct_headers_in_one_profile() { + let state = test_server_state().await; + let store = state.store.as_ref(); + let service = token_grant_credential("service"); + let mut identity = token_grant_credential("identity"); + identity.auth_style = "header".into(); + identity.header_name = "X-Workload-Jwt".into(); + let grant = identity.token_grant.as_mut().unwrap(); + grant.token_endpoint = "https://identity.example.com/token".into(); + grant.jwt_svid_audience = "identity-proxy".into(); + grant.audience = "workload".into(); + grant.scopes = vec!["identity.read".into()]; + grant.cache_ttl = Some(prost_types::Duration { + seconds: 45, + nanos: 0, + }); + import_token_grant_profile_with_credentials( + &state, + "grant-pair", + "api.example.com", + 443, + "/v1/**", + vec![service.clone(), identity.clone()], + ) + .await; + create_empty_token_grant_provider(store, "provider", "grant-pair").await; + validate_provider_environment_keys_unique(store, "default", &["provider".into()]) + .await + .expect("distinct headers must compose in one provider"); + + let catalog = ProviderProfileSources::with_default_sources() + .snapshot_catalog(store, "default") + .await + .unwrap(); + let profile = get_provider_type_profile_for_scope(&catalog, "grant-pair", "default") + .unwrap() + .to_proto(); + let mut credentials = HashMap::new(); + insert_dynamic_credentials_for_profile(&mut credentials, &profile, "provider"); + assert_eq!(credentials.len(), 2); + for expected in [service, identity] { + let key = dynamic_credential_key( + "api.example.com", + 443, + "/v1/**", + "provider", + &expected.name, + ); + assert_eq!(credentials[&key].token_grant, expected.token_grant); + assert_eq!(credentials[&key].header_name, expected.header_name); + assert!(credentials[&key].env_vars.is_empty()); + } + } + + #[test] + fn dynamic_token_grants_reject_normalized_header_collisions() { + for header in ["Authorization", " authorization ", ""] { + let mut profile = custom_profile("grant-pair"); + let mut second = token_grant_credential("second"); + second.header_name = header.into(); + profile.credentials = vec![token_grant_credential("first"), second]; + profile.endpoints = vec![NetworkEndpoint { + host: "api.example.com".into(), + port: 443, + path: "/v1/**".into(), + ..Default::default() + }]; + let bindings = dynamic_token_grant_bindings_for_profile("provider", &profile); + assert_eq!(bindings.len(), 2); + assert!(validate_dynamic_token_grant_bindings_unambiguous(&bindings).is_err()); + } + } + + #[test] + fn dynamic_token_grants_allow_distinct_headers_across_providers() { + let mut first = custom_profile("grant-a"); + first.credentials = vec![token_grant_credential("service")]; + first.endpoints = vec![NetworkEndpoint { + host: "api.example.com".into(), + port: 443, + path: "/v1/**".into(), + ..Default::default() + }]; + let mut second = first.clone(); + second.credentials[0].auth_style = "header".into(); + second.credentials[0].header_name = "X-Workload-Jwt".into(); + let mut bindings = dynamic_token_grant_bindings_for_profile("provider-a", &first); + bindings.extend(dynamic_token_grant_bindings_for_profile( + "provider-b", + &second, + )); + validate_dynamic_token_grant_bindings_unambiguous(&bindings) + .expect("different headers must not be treated as alternatives"); + } + #[tokio::test] async fn dynamic_token_grants_reject_equal_specificity_overlap() { let state = test_server_state().await; diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index 0aa453258f..7615005720 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -4994,6 +4994,130 @@ network_policies: ); } + #[tokio::test] + async fn l7_rest_tls_relay_injects_multiple_grants() { + assert_multiple_grants_tls_relay(Ok("identity-token")).await; + } + + #[tokio::test] + async fn l7_rest_tls_relay_second_grant_failure_forwards_nothing() { + assert_multiple_grants_tls_relay(Err("issuer echoed identity-secret")).await; + } + + async fn token_grant_tls_pair() -> ( + tokio_rustls::client::TlsStream, + tokio_rustls::server::TlsStream, + ) { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let key = rcgen::KeyPair::generate().unwrap(); + let cert = rcgen::CertificateParams::new(vec!["api.example.test".into()]) + .unwrap() + .self_signed(&key) + .unwrap(); + let server_config = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![cert.der().clone()], + rustls::pki_types::PrivatePkcs8KeyDer::from(key.serialize_der()).into(), + ) + .unwrap(); + let mut roots = rustls::RootCertStore::empty(); + roots.add(cert.der().clone()).unwrap(); + let client_config = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + let connector = tokio_rustls::TlsConnector::from(Arc::new(client_config)); + let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config)); + let (client, server) = tokio::io::duplex(16384); + let (client, server) = tokio::join!( + connector.connect("api.example.test".try_into().unwrap(), client), + acceptor.accept(server), + ); + (client.unwrap(), server.unwrap()) + } + + async fn assert_multiple_grants_tls_relay(identity_result: std::result::Result<&str, &str>) { + let (config, tunnel_engine, ctx, fixture) = + rest_token_grant_relay_context(Ok("service-token")); + let service_key = "api.example.test\t8080\t/v1/**\tprovider:access_token"; + let identity_key = "api.example.test\t8080\t/v1/**\tprovider:identity"; + let mut identity = fixture.dynamic_credentials().read().unwrap()[service_key].clone(); + identity.name = "identity".into(); + identity.auth_style = "header".into(); + identity.header_name = "X-Workload-Jwt".into(); + fixture.add_credential(identity_key, identity, identity_result); + // Both sides verify a synthetic certificate: the test exercises encrypted + // application traffic, inspection and credential injection, then upstream TLS. + let (mut app, mut relay_client) = token_grant_tls_pair().await; + let (mut relay_upstream, mut upstream) = token_grant_tls_pair().await; + let relay = tokio::spawn(async move { + relay_with_inspection( + &config, + tunnel_engine, + &mut relay_client, + &mut relay_upstream, + &ctx, + ) + .await + }); + app.write_all(b"GET /v1/projects HTTP/1.1\r\nHost: api.example.test\r\nAuthorization: Bearer agent-token\r\nX-Workload-Jwt: agent-identity\r\nConnection: close\r\n\r\n") + .await.unwrap(); + if identity_result.is_ok() { + let mut request = [0u8; 2048]; + let n = tokio::time::timeout( + std::time::Duration::from_secs(2), + upstream.read(&mut request), + ) + .await + .unwrap() + .unwrap(); + let request = String::from_utf8_lossy(&request[..n]); + assert!(request.contains("Authorization: Bearer service-token\r\n")); + assert!(request.contains("X-Workload-Jwt: identity-token\r\n")); + assert!(!request.contains("agent-token")); + assert!(!request.contains("agent-identity")); + upstream + .write_all( + b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ) + .await + .unwrap(); + } + let mut response = [0u8; 1024]; + let n = tokio::time::timeout(std::time::Duration::from_secs(2), app.read(&mut response)) + .await + .unwrap() + .unwrap(); + let response = String::from_utf8_lossy(&response[..n]); + assert!(response.contains(if identity_result.is_ok() { + "204 No Content" + } else { + "502 Bad Gateway" + })); + assert!(!response.contains("service-token")); + assert!(!response.contains("identity-secret")); + drop(app); + tokio::time::timeout(std::time::Duration::from_secs(2), relay) + .await + .unwrap() + .unwrap() + .unwrap(); + if identity_result.is_err() { + let mut request = [0u8; 128]; + match tokio::time::timeout( + std::time::Duration::from_secs(2), + upstream.read(&mut request), + ) + .await + .unwrap() + { + Ok(n) => assert_eq!(n, 0, "failed grant must send no request bytes"), + Err(error) => assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof), + } + } + fixture.assert_requested_keys(&[service_key, identity_key]); + } + #[tokio::test] async fn l7_rest_relay_injects_token_grant_authorization_header() { let (config, tunnel_engine, ctx, fixture) = diff --git a/crates/openshell-supervisor-network/src/l7/token_grant_injection.rs b/crates/openshell-supervisor-network/src/l7/token_grant_injection.rs index 1813ddcf4b..98d22c4854 100644 --- a/crates/openshell-supervisor-network/src/l7/token_grant_injection.rs +++ b/crates/openshell-supervisor-network/src/l7/token_grant_injection.rs @@ -3,6 +3,10 @@ //! Endpoint-bound dynamic token grant injection for HTTP relay paths. +#[cfg(test)] +mod multiple_grants; + +use std::collections::BTreeMap; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -68,13 +72,18 @@ pub fn default_resolver() -> Arc { Arc::new(SpiffeTokenGrantResolver) } -/// Checks for endpoint-bound token grant credentials and injects an -/// Authorization header before forwarding the request upstream. +/// Resolves one endpoint-bound grant per protected header before rewriting a request. +/// +/// Each header independently uses its most-specific matching binding. Every selected +/// grant must succeed; callers must not forward the request when this returns an error. pub async fn inject_if_needed(req: L7Request, ctx: &L7EvalContext) -> Result { let request_path = req.target.split('?').next().unwrap_or(req.target.as_str()); - let token_grant_credential = ctx.dynamic_credentials.as_ref().and_then(|dyn_creds| { - dyn_creds.read().map_or(None, |creds_guard| { - creds_guard + let credentials = match ctx.dynamic_credentials.as_ref() { + Some(dynamic_credentials) => { + let credentials = dynamic_credentials + .read() + .map_err(|_| miette!("dynamic credential snapshot unavailable"))?; + let candidates = credentials .iter() .filter_map(|(key, cred)| { let score = @@ -83,83 +92,117 @@ pub async fn inject_if_needed(req: L7Request, ctx: &L7EvalContext) -> Result Vec::new(), + }; + if credentials.is_empty() { + return Ok(req); + } + let resolver = ctx + .token_grant_resolver + .as_ref() + .ok_or_else(|| miette!("token grant resolver unavailable"))?; + + // Release the credential snapshot lock before acquisition. Keep all token values + // local until every selected grant and header has been validated, so a later + // failure cannot expose a partially authenticated request to a relay caller. + let mut headers = Vec::with_capacity(credentials.len()); + for (provider_key, cred) in &credentials { + let token_grant = cred + .token_grant .as_ref() - .ok_or_else(|| miette!("token grant resolver unavailable"))?; - let request = token_grant_request(&provider_key, token_grant)?; - - match resolver.obtain(request).await { - Ok(access_token) => { - let modified_raw_header = - inject_token_grant_header(&req.raw_header, &cred, &access_token)?; - let provider_key = ocsf_message_field(&provider_key); - ocsf_emit!( - HttpActivityBuilder::new(ocsf_ctx()) - .activity(ActivityId::Other) - .action(ActionId::Allowed) - .disposition(DispositionId::Allowed) - .severity(SeverityId::Informational) - .http_request(HttpRequest::new( - &req.action, - OcsfUrl::new("http", &ctx.host, request_path, ctx.port), - )) - .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) - .message(format!( - "Token grant successful for {} to {}:{}", - provider_key, ctx.host, ctx.port - )) - .build() - ); - return Ok(L7Request { - action: req.action, - target: req.target, - query_params: req.query_params, - raw_header: modified_raw_header, - body_length: req.body_length, - }); - } - Err(e) => { - warn!( - host = %ctx.host, - port = ctx.port, - provider = %provider_key, - error = %e, - "Token grant failed: {e}" - ); - let provider_key = ocsf_message_field(&provider_key); - ocsf_emit!( - HttpActivityBuilder::new(ocsf_ctx()) - .activity(ActivityId::Fail) - .action(ActionId::Denied) - .disposition(DispositionId::Blocked) - .severity(SeverityId::Medium) - .status(StatusId::Failure) - .http_request(HttpRequest::new( - &req.action, - OcsfUrl::new("http", &ctx.host, request_path, ctx.port), - )) - .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) - .message(format!( - "Token grant failed for {} to {}:{}: {}", - provider_key, ctx.host, ctx.port, e - )) - .build() - ); - return Err(miette!("Token grant failed: {}", e)); - } + .ok_or_else(|| miette!("selected credential has no token grant"))?; + let request = token_grant_request(provider_key, token_grant)?; + if let Ok(access_token) = resolver.obtain(request).await { + crate::token_grant::validate_access_token(&access_token)?; + headers.push(token_grant_header(cred, &access_token)?); + } else { + // An issuer may echo credentials in its error description. Only the + // binding identity is safe to include in diagnostics or relay errors. + let provider_key = ocsf_message_field(provider_key); + warn!( + host = %ctx.host, + port = ctx.port, + provider = %provider_key, + "Token grant failed" + ); + ocsf_emit!( + HttpActivityBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .http_request(HttpRequest::new( + &req.action, + OcsfUrl::new("http", &ctx.host, request_path, ctx.port), + )) + .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) + .message(format!( + "Token grant failed for {} to {}:{}", + provider_key, ctx.host, ctx.port + )) + .build() + ); + return Err(miette!("Token grant failed")); } } - Ok(req) + let mut raw_header = req.raw_header; + for (name, value) in headers { + raw_header = inject_header(&raw_header, &name, &value)?; + } + for (provider_key, _) in credentials { + ocsf_emit!( + HttpActivityBuilder::new(ocsf_ctx()) + .activity(ActivityId::Other) + .action(ActionId::Allowed) + .disposition(DispositionId::Allowed) + .severity(SeverityId::Informational) + .http_request(HttpRequest::new( + &req.action, + OcsfUrl::new("http", &ctx.host, request_path, ctx.port), + )) + .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) + .message(format!( + "Token grant successful for {} to {}:{}", + ocsf_message_field(&provider_key), + ctx.host, + ctx.port + )) + .build() + ); + } + Ok(L7Request { raw_header, ..req }) +} + +fn select_token_grants( + mut candidates: Vec<(u32, String, ProviderProfileCredential)>, +) -> Result> { + // Examine the strongest bindings first. The complete binding key stays with + // its grant because the resolver uses it to separate cache entries and revisions. + candidates.sort_unstable_by(|a, b| b.0.cmp(&a.0).then_with(|| b.1.cmp(&a.1))); + let mut selected = BTreeMap::::new(); + for (score, key, credential) in candidates { + let header = token_grant_header_name(&credential)?.to_ascii_lowercase(); + if let Some((selected_score, selected_key, _)) = selected.get(&header) { + // Equal-specificity selectors of the same credential can overlap. A + // different credential cannot win a tie for the same protected header. + if score == *selected_score + && key.rsplit('\t').next() != selected_key.rsplit('\t').next() + { + return Err(miette!("ambiguous dynamic token grants for one header")); + } + continue; + } + selected.insert(header, (score, key, credential)); + } + Ok(selected + .into_values() + .map(|(_, key, cred)| (key, cred)) + .collect()) } fn ocsf_message_field(value: &str) -> String { @@ -246,6 +289,7 @@ fn count_as_u32(count: usize) -> u32 { u32::try_from(count).unwrap_or(u32::MAX) } +#[cfg(test)] fn inject_token_grant_header( raw_header: &[u8], credential: &ProviderProfileCredential, @@ -260,6 +304,16 @@ fn token_grant_header( credential: &ProviderProfileCredential, access_token: &str, ) -> Result<(String, String)> { + let header_name = token_grant_header_name(credential)?; + let header_value = if credential.auth_style.trim().eq_ignore_ascii_case("header") { + access_token.to_string() + } else { + format!("Bearer {access_token}") + }; + Ok((header_name.to_string(), header_value)) +} + +fn token_grant_header_name(credential: &ProviderProfileCredential) -> Result<&str> { match credential.auth_style.trim().to_ascii_lowercase().as_str() { "" | "bearer" => { let header_name = if credential.header_name.trim().is_empty() { @@ -268,7 +322,7 @@ fn token_grant_header( credential.header_name.trim() }; validate_header_name(header_name)?; - Ok((header_name.to_string(), format!("Bearer {access_token}"))) + Ok(header_name) } "header" => { let header_name = credential.header_name.trim(); @@ -278,7 +332,7 @@ fn token_grant_header( )); } validate_header_name(header_name)?; - Ok((header_name.to_string(), access_token.to_string())) + Ok(header_name) } other => Err(miette!( "token grant auth_style '{other}' is not supported; use bearer or header" @@ -371,7 +425,7 @@ pub mod test_support { struct FakeTokenGrantResolver { requests: Arc>>, - response: std::result::Result, + responses: Arc>>>, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -391,6 +445,7 @@ pub mod test_support { dynamic_credentials: Arc>>, resolver: Arc, requests: Arc>>, + responses: Arc>>>, } impl TokenGrantTestFixture { @@ -420,9 +475,13 @@ pub mod test_support { token_grant: ProviderCredentialTokenGrant, ) -> Self { let requests = Arc::new(Mutex::new(Vec::new())); + let responses = Arc::new(Mutex::new(HashMap::from([( + key.to_string(), + response.map(str::to_string).map_err(str::to_string), + )]))); let resolver = Arc::new(FakeTokenGrantResolver { requests: requests.clone(), - response: response.map(str::to_string).map_err(str::to_string), + responses: responses.clone(), }); let mut dynamic_credentials = HashMap::new(); @@ -441,9 +500,65 @@ pub mod test_support { dynamic_credentials: Arc::new(std::sync::RwLock::new(dynamic_credentials)), resolver, requests, + responses, } } + pub fn add_credential( + &self, + key: &str, + credential: ProviderProfileCredential, + response: std::result::Result<&str, &str>, + ) { + self.dynamic_credentials + .write() + .unwrap() + .insert(key.to_string(), credential); + self.responses.lock().unwrap().insert( + key.to_string(), + response.map(str::to_string).map_err(str::to_string), + ); + } + + pub fn assert_requested_keys(&self, expected: &[&str]) { + let requests = self.requests.lock().unwrap(); + assert_eq!( + requests + .iter() + .map(|r| r.provider_key.as_str()) + .collect::>(), + expected + ); + } + + pub fn assert_request_configuration( + &self, + key: &str, + grant: &ProviderCredentialTokenGrant, + ) { + let requests = self.requests.lock().unwrap(); + let request = requests + .iter() + .find(|r| r.provider_key == key) + .expect("grant was requested"); + assert_eq!(request.token_endpoint, grant.token_endpoint); + assert_eq!(request.jwt_svid_audience, grant.jwt_svid_audience); + assert_eq!(request.client_assertion_type, grant.client_assertion_type); + assert_eq!(request.audience, grant.audience); + assert_eq!(request.scopes, grant.scopes); + assert_eq!( + request.cache_ttl, + grant + .cache_ttl + .as_ref() + .map(openshell_core::time::duration_to_std) + .transpose() + .unwrap() + ); + assert_eq!(request.grant_type, grant.grant_type); + assert_eq!(request.requested_token_type, grant.requested_token_type); + } + pub fn dynamic_credentials( &self, ) -> Arc>> { @@ -571,7 +686,16 @@ pub mod test_support { .lock() .expect("fake token grant requests lock poisoned") .push(owned); - self.response.clone().map_err(|err| miette!("{err}")) + // Keep acquisition pending once so joined request tests exercise + // overlapping grants instead of completing each branch in one poll. + tokio::task::yield_now().await; + self.responses + .lock() + .unwrap() + .get(request.provider_key) + .expect("fake grant response configured") + .clone() + .map_err(|err| miette!("{err}")) }) } } diff --git a/crates/openshell-supervisor-network/src/l7/token_grant_injection/multiple_grants.rs b/crates/openshell-supervisor-network/src/l7/token_grant_injection/multiple_grants.rs new file mode 100644 index 0000000000..cb46fa883a --- /dev/null +++ b/crates/openshell-supervisor-network/src/l7/token_grant_injection/multiple_grants.rs @@ -0,0 +1,257 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use super::test_support::TokenGrantTestFixture; +use super::*; +use crate::l7::provider::BodyLength; +use std::collections::HashMap; +use std::sync::Mutex; +use tracing::instrument::WithSubscriber; +use tracing_subscriber::layer::SubscriberExt; + +const SERVICE: &str = "api.example.com\t443\t/v1/**\trev:7\tprovider:service"; +const IDENTITY: &str = "api.example.com\t443\t/v1/**\trev:7\tprovider:identity"; + +fn request() -> L7Request { + L7Request { + action: "POST".into(), + target: "/v1/projects?view=full".into(), + query_params: HashMap::default(), + raw_header: b"POST /v1/projects?view=full HTTP/1.1\r\nHost: api.example.com\r\nAuthorization: Bearer agent-token\r\nauthorization : duplicate\r\nX-Workload-Jwt: agent-identity\r\nx-workload-jwt: duplicate\r\nX-Static: openshell:placeholder\r\nContent-Length: 4\r\n\r\nbody".to_vec(), + body_length: BodyLength::ContentLength(4), + } +} + +fn fixture( + identity_result: std::result::Result<&str, &str>, +) -> (TokenGrantTestFixture, L7EvalContext) { + let fixture = TokenGrantTestFixture::success(SERVICE, "service-token"); + let mut identity = fixture.dynamic_credentials().read().unwrap()[SERVICE].clone(); + identity.name = "identity".into(); + identity.auth_style = "header".into(); + identity.header_name = "X-Workload-Jwt".into(); + let grant = identity.token_grant.as_mut().unwrap(); + grant.token_endpoint = "https://identity.example.com/token".into(); + grant.jwt_svid_audience = "identity-proxy".into(); + grant.audience = "workload".into(); + grant.scopes = vec!["identity.read".into()]; + grant.cache_ttl = Some(prost_types::Duration { + seconds: 45, + nanos: 0, + }); + fixture.add_credential(IDENTITY, identity, identity_result); + let ctx = L7EvalContext { + host: "api.example.com".into(), + port: 443, + dynamic_credentials: Some(fixture.dynamic_credentials()), + token_grant_resolver: Some(fixture.resolver()), + ..Default::default() + }; + (fixture, ctx) +} + +#[tokio::test] +async fn injects_independent_grants_and_replaces_all_protected_headers() { + let (fixture, ctx) = fixture(Ok("identity-token")); + let rewritten = inject_if_needed(request(), &ctx).await.unwrap(); + let bytes = String::from_utf8(rewritten.raw_header).unwrap(); + assert_eq!( + bytes + .matches("Authorization: Bearer service-token\r\n") + .count(), + 1 + ); + assert_eq!( + bytes.matches("X-Workload-Jwt: identity-token\r\n").count(), + 1 + ); + assert!(!bytes.contains("agent-token")); + assert!(!bytes.contains("agent-identity")); + assert!(!bytes.contains("duplicate")); + assert!(bytes.contains("X-Static: openshell:placeholder\r\n")); + assert!(bytes.ends_with("\r\n\r\nbody")); + fixture.assert_requested_keys(&[SERVICE, IDENTITY]); + for key in [SERVICE, IDENTITY] { + let credentials = fixture.dynamic_credentials(); + let credentials = credentials.read().unwrap(); + fixture.assert_request_configuration(key, credentials[key].token_grant.as_ref().unwrap()); + } +} + +#[tokio::test] +async fn chooses_specific_binding_independently_for_each_header() { + let (fixture, ctx) = fixture(Ok("identity-token")); + let specific_key = "api.example.com\t443\t/v1/projects\trev:7\tother:service"; + let mut specific = fixture.dynamic_credentials().read().unwrap()[SERVICE].clone(); + specific.token_grant.as_mut().unwrap().audience = "projects-only".into(); + fixture.add_credential(specific_key, specific, Ok("specific-service")); + let rewritten = inject_if_needed(request(), &ctx).await.unwrap(); + let bytes = String::from_utf8(rewritten.raw_header).unwrap(); + assert!(bytes.contains("Authorization: Bearer specific-service\r\n")); + assert!(bytes.contains("X-Workload-Jwt: identity-token\r\n")); + fixture.assert_requested_keys(&[specific_key, IDENTITY]); +} + +#[tokio::test] +async fn audience_override_changes_only_its_own_credential() { + let (fixture, ctx) = fixture(Ok("identity-token")); + let override_key = "api.example.com\t443\t/v1/projects\trev:7\tprovider:identity"; + let mut identity = fixture.dynamic_credentials().read().unwrap()[IDENTITY].clone(); + identity.token_grant.as_mut().unwrap().audience = "project-identity".into(); + fixture.add_credential(override_key, identity.clone(), Ok("project-identity-token")); + let rewritten = inject_if_needed(request(), &ctx).await.unwrap(); + let bytes = String::from_utf8(rewritten.raw_header).unwrap(); + assert!(bytes.contains("Authorization: Bearer service-token\r\n")); + assert!(bytes.contains("X-Workload-Jwt: project-identity-token\r\n")); + fixture.assert_requested_keys(&[SERVICE, override_key]); + fixture.assert_request_configuration(override_key, identity.token_grant.as_ref().unwrap()); +} + +#[tokio::test] +async fn rejects_tied_credentials_for_same_header_before_acquisition() { + for header in ["Authorization", " authorization ", ""] { + let (fixture, ctx) = fixture(Ok("identity-token")); + let mut collision = fixture.dynamic_credentials().read().unwrap()[SERVICE].clone(); + collision.name = "collision".into(); + collision.header_name = header.into(); + fixture.add_credential( + "api.example.com\t443\t/v1/**\trev:7\tother:collision", + collision, + Ok("collision-token"), + ); + let error = inject_if_needed(request(), &ctx).await.unwrap_err(); + assert_eq!( + error.to_string(), + "ambiguous dynamic token grants for one header" + ); + fixture.assert_no_requests(); + } +} + +#[tokio::test] +async fn rejects_invalid_second_header_before_acquisition() { + let (fixture, ctx) = fixture(Ok("identity-token")); + let mut identity = fixture.dynamic_credentials().read().unwrap()[IDENTITY].clone(); + identity.header_name = "Content-Length".into(); + fixture.add_credential(IDENTITY, identity, Ok("identity-token")); + assert!(inject_if_needed(request(), &ctx).await.is_err()); + fixture.assert_no_requests(); +} + +#[tokio::test] +async fn rejects_malformed_second_token_after_first_grant_succeeds() { + let (fixture, ctx) = fixture(Ok("identity-token\r\nInjected: yes")); + let error = inject_if_needed(request(), &ctx).await.unwrap_err(); + assert_eq!( + error.to_string(), + "token grant returned a malformed access token" + ); + fixture.assert_requested_keys(&[SERVICE, IDENTITY]); +} + +#[derive(Clone)] +struct Capture(Arc>>); + +impl std::io::Write for Capture { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[tokio::test] +async fn second_grant_failure_redacts_errors_and_emits_no_success() { + // OCSF callsite interest is process-global. Run capture alone so unrelated + // parallel tests cannot disable the event while its subscriber is installed. + const CAPTURE_CHILD: &str = "OPENSHELL_MULTIPLE_GRANTS_CAPTURE_CHILD"; + if std::env::var_os(CAPTURE_CHILD).is_none() { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "l7::token_grant_injection::multiple_grants::second_grant_failure_redacts_errors_and_emits_no_success", + "--nocapture", + ]) + .env(CAPTURE_CHILD, "1") + .output() + .unwrap(); + assert!( + output.status.success(), + "isolated capture failed: {}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + return; + } + let (fixture, ctx) = fixture(Err("issuer echoed service-token and identity-secret")); + let logs = Arc::new(Mutex::new(Vec::new())); + let events = Arc::new(Mutex::new(Vec::new())); + let log_writer = Capture(logs.clone()); + let subscriber = tracing_subscriber::registry() + .with( + tracing_subscriber::fmt::layer() + .with_ansi(false) + .without_time() + .with_writer(move || log_writer.clone()), + ) + .with(openshell_ocsf::OcsfJsonlLayer::new(Capture(events.clone()))); + let error = inject_if_needed(request(), &ctx) + .with_subscriber(subscriber) + .await + .unwrap_err(); + assert_eq!(error.to_string(), "Token grant failed"); + let logs = String::from_utf8(logs.lock().unwrap().clone()).unwrap(); + let events = String::from_utf8(events.lock().unwrap().clone()).unwrap(); + assert!(logs.contains("Token grant failed")); + assert!(events.contains("Token grant failed")); + for output in [&logs, &events] { + assert!(!output.contains("service-token")); + assert!(!output.contains("identity-secret")); + assert!(!output.contains("Token grant successful")); + } + fixture.assert_requested_keys(&[SERVICE, IDENTITY]); +} + +#[tokio::test] +async fn does_not_acquire_grants_outside_endpoint_bindings() { + for (host, port, path) in [ + ("other.example.com", 443, "/v1/projects"), + ("api.example.com", 8443, "/v1/projects"), + ("api.example.com", 443, "/private/projects"), + ] { + let (fixture, mut ctx) = fixture(Ok("identity-token")); + ctx.host = host.into(); + ctx.port = port; + let mut req = request(); + req.target = path.into(); + let original = req.raw_header.clone(); + assert_eq!( + inject_if_needed(req, &ctx).await.unwrap().raw_header, + original + ); + fixture.assert_no_requests(); + } +} + +#[tokio::test] +async fn concurrent_requests_keep_credential_snapshots_separate() { + let (first, first_ctx) = fixture(Ok("first-identity")); + let (second, second_ctx) = fixture(Ok("second-identity")); + let (first_result, second_result) = tokio::join!( + inject_if_needed(request(), &first_ctx), + inject_if_needed(request(), &second_ctx), + ); + for (result, expected, absent) in [ + (first_result, "first-identity", "second-identity"), + (second_result, "second-identity", "first-identity"), + ] { + let bytes = String::from_utf8(result.unwrap().raw_header).unwrap(); + assert!(bytes.contains(expected)); + assert!(!bytes.contains(absent)); + } + first.assert_requested_keys(&[SERVICE, IDENTITY]); + second.assert_requested_keys(&[SERVICE, IDENTITY]); +} diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 4f273817d7..7adbc31b73 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -12805,8 +12805,7 @@ network_policies: .await .expect_err("forward token grant failure should stop request rewriting"); - assert!(err.to_string().contains("Token grant failed")); - assert!(err.to_string().contains("oauth unavailable")); + assert_eq!(err.to_string(), "Token grant failed"); fixture.assert_one_request("api.example.test\t8080\t/v1/**\tprovider:access_token"); } @@ -12819,8 +12818,7 @@ network_policies: .await .expect_err("forward token exchange failure should stop request rewriting"); - assert!(err.to_string().contains("Token grant failed")); - assert!(err.to_string().contains("oauth unavailable")); + assert_eq!(err.to_string(), "Token grant failed"); fixture.assert_one_token_exchange_request( "api.example.test\t8080\t/v1/**\tprovider:access_token", ); diff --git a/crates/openshell-supervisor-network/src/token_grant.rs b/crates/openshell-supervisor-network/src/token_grant.rs index 5d03012f61..f522b0a3d3 100644 --- a/crates/openshell-supervisor-network/src/token_grant.rs +++ b/crates/openshell-supervisor-network/src/token_grant.rs @@ -900,6 +900,69 @@ mod tests { ); } + #[tokio::test] + async fn multiple_credentials_cache_and_expire_independently() { + let cache = TokenCache::new(); + let calls = Arc::new(AtomicUsize::new(0)); + let scopes = vec!["read".to_string()]; + let service = "api.example.test\t443\t/v1/**\trev:1\tprovider:service"; + let identity = "api.example.test\t443\t/v1/**\trev:1\tprovider:identity"; + for (key, expected) in [(service, "token-1"), (identity, "token-2")] { + let token = obtain_counted_test_token(CountedTokenGrantInput { + cache: &cache, + provider_name: key, + token_endpoint: "https://auth.example.com/token", + jwt_svid_audience: "https://auth.example.com", + audience: "api://resource", + scopes: &scopes, + cache_ttl_override: None, + expires_in: 120, + grant_calls: calls.clone(), + }) + .await + .unwrap(); + assert_eq!(token, expected); + } + let identity_cache_key = token_cache_key(TokenCacheKeyInput { + provider_name: identity, + token_endpoint: "https://auth.example.com/token", + jwt_svid_audience: "https://auth.example.com", + client_assertion_type: DEFAULT_CLIENT_ASSERTION_TYPE, + audience: "api://resource", + scopes: &scopes, + grant_type: ProviderCredentialTokenGrantType::ClientCredentials, + requested_token_type: ACCESS_TOKEN_TYPE, + }); + cache.set(identity_cache_key, "expired".into(), current_time_ms() - 1); + let service_token = obtain_token_without_grant_call( + &cache, + service, + "https://auth.example.com/token", + "https://auth.example.com", + "api://resource", + &scopes, + None, + ) + .await + .unwrap(); + assert_eq!(service_token, "token-1"); + let identity_token = obtain_counted_test_token(CountedTokenGrantInput { + cache: &cache, + provider_name: identity, + token_endpoint: "https://auth.example.com/token", + jwt_svid_audience: "https://auth.example.com", + audience: "api://resource", + scopes: &scopes, + cache_ttl_override: None, + expires_in: 120, + grant_calls: calls.clone(), + }) + .await + .unwrap(); + assert_eq!(identity_token, "token-3"); + assert_eq!(calls.load(Ordering::SeqCst), 3); + } + #[tokio::test] async fn obtain_provider_token_uses_cache_for_same_key() { let cache = TokenCache::new(); diff --git a/docs/how-it-works/providers/profiles.mdx b/docs/how-it-works/providers/profiles.mdx index 11c5438d3c..7a68c6aec3 100644 --- a/docs/how-it-works/providers/profiles.mdx +++ b/docs/how-it-works/providers/profiles.mdx @@ -555,7 +555,45 @@ The refresh attaches to the primary credential (`access_key_id`). Each reference ### Dynamic Token Grants -`token_grant` belongs to one credential declaration. When a sandbox with the provider attached sends HTTP traffic to a matching profile endpoint, the supervisor resolves the dynamic credential, caches the returned access token, and injects it before forwarding the request upstream. Use `auth_style: bearer` to inject `Authorization: Bearer `, or `auth_style: header` with `header_name` to inject the raw access token into a custom header. Token grants do not support `query` or `path` placement. +`token_grant` belongs to one credential declaration. When a sandbox with the provider attached sends HTTP traffic to a matching profile endpoint, the supervisor resolves the dynamic credentials, caches each returned access token separately, and injects them before forwarding the request upstream. Use `auth_style: bearer` to inject `Authorization: Bearer `, or `auth_style: header` with `header_name` to inject the raw access token into a custom header. Token grants do not support `query` or `path` placement. + +Declare multiple credentials with different destination headers when one request needs independently issued tokens. Each credential keeps its own token endpoint, audience, scopes, and cache lifetime. For example, this profile sends a service bearer token and a separate workload token on the same inspected request: + +```yaml +id: protected-api +display_name: Protected API +category: data +credentials: + - name: service_access + required: true + auth_style: bearer + header_name: Authorization + token_grant: + token_endpoint: https://service-issuer.example.com/token + audience: https://api.example.com + scopes: [resource.read] + cache_ttl: 60s + - name: workload_identity + required: true + auth_style: header + header_name: x-workload-jwt + token_grant: + token_endpoint: https://identity-issuer.example.com/token + jwt_svid_audience: identity-proxy + cache_ttl: 45s +endpoints: + - host: api.example.com + port: 443 + path: /v1/** + protocol: rest + access: full +``` + +The custom header carries the identity issuer's returned access token. To carry a JWT-SVID there, that issuer must return the JWT-SVID as its access token. Neither granted value is placed in the sandbox environment. + +The supervisor chooses the most-specific matching binding independently for each destination header. Bindings that target the same header are alternatives; equal-specificity matches from different credentials are rejected. Header names are case-insensitive. Omitting both `auth_style` and `header_name` selects `Authorization`; an explicit `bearer` or `header` style requires `header_name`. A more-specific audience override for one credential does not suppress grants for other headers. + +All selected grants must succeed and return valid header values before the request is rewritten. A failed grant returns HTTP 502 without forwarding the request. An earlier successful grant may remain in its own cache for a later attempt. Injection replaces every agent-supplied instance of each selected header, including any static credential placeholder in that header, and preserves unrelated headers. Issuer error descriptions are excluded from grant-failure logs and client errors because they may contain credentials. OpenShell supports two dynamic grant types: @@ -1080,7 +1118,7 @@ For a static provider, the sequence is attach, wait, then launch client A; updat An acknowledged detachment removes its provider policy layer from the active effective policy, revokes future resolution for its existing placeholders, and removes its credential placeholders from future process environments. It does not remove strings from already-running process environments or undo requests already forwarded upstream. -OpenShell rejects provider updates and refresh configuration when they would make two providers attached to the same sandbox expose the same active credential environment key. It also rejects attached provider sets with ambiguous dynamic token grants at equal host/path specificity. Use provider-specific credential names and make one dynamic grant selector more specific when one sandbox needs multiple providers with overlapping upstream concepts. +OpenShell rejects provider updates and refresh configuration when they would make two providers attached to the same sandbox expose the same active credential environment key. It also rejects attached provider sets whose dynamic token grants target the same header at equal host/path specificity. Grants for distinct headers can share an endpoint. Use provider-specific credential names and make one same-header selector more specific when the bindings are intended as alternatives. ## Next Steps diff --git a/e2e/rust/tests/provider_token_exchange.rs b/e2e/rust/tests/provider_token_exchange.rs index e845938e70..97d8e2c587 100644 --- a/e2e/rust/tests/provider_token_exchange.rs +++ b/e2e/rust/tests/provider_token_exchange.rs @@ -11,20 +11,20 @@ use std::net::{Ipv4Addr, SocketAddr}; use std::os::unix::fs::PermissionsExt as _; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::task::{Context, Poll}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use base64::Engine as _; use futures_util::future::BoxFuture; -use jsonwebtoken::{Algorithm, EncodingKey, Header}; +use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation}; use openshell_e2e::harness::binary::openshell_cmd; use openshell_e2e::harness::port::find_free_port; use openshell_e2e::harness::sandbox::SandboxGuard; use serde_json::json; use tempfile::NamedTempFile; use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; -use tokio::net::{TcpListener, UnixListener}; -use tokio::process::Command; +use tokio::net::{TcpListener, TcpStream, UnixListener}; use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream, UnixListenerStream}; use tonic::body::Body as TonicBody; use tonic::codegen::{Body, http}; @@ -38,6 +38,30 @@ const INTERMEDIATE_TOKEN: &str = "intermediate-token"; const FINAL_ACCESS_TOKEN: &str = "final-access-token"; const TOKEN_TYPE_ACCESS_TOKEN: &str = "urn:ietf:params:oauth:token-type:access_token"; const CLIENT_ASSERTION_TYPE: &str = "urn:ietf:params:oauth:client-assertion-type:jwt-spiffe"; +const IDENTITY_AUDIENCE: &str = "identity-proxy"; +const IDENTITY_JWT_AUDIENCE: &str = "https://identity.openshell-e2e.test"; +const IDENTITY_ASSERTION_TYPE: &str = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"; + +#[derive(Default)] +struct GrantObservations { + service_issued: [AtomicUsize; 2], + identity_issued: [AtomicUsize; 2], + service_denied: AtomicUsize, + identity_denied: AtomicUsize, + target_requests: AtomicUsize, + target_rejected: AtomicUsize, +} + +impl GrantObservations { + fn issued(&self) -> [(usize, usize); 2] { + std::array::from_fn(|index| { + ( + self.service_issued[index].load(Ordering::SeqCst), + self.identity_issued[index].load(Ordering::SeqCst), + ) + }) + } +} const TEST_RSA_PRIVATE_KEY: &str = r"-----BEGIN PRIVATE KEY----- MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCvCoZ0mVHpCHsF @@ -118,7 +142,7 @@ struct SpiffeWorkloadApi { } impl SpiffeWorkloadApi { - fn jwt_svid(&self, audience: Vec) -> Result { + fn jwt_svid(&self, audience: &[String]) -> Result { let now = unix_timestamp(); let mut header = Header::new(Algorithm::RS256); header.kid = Some(KEY_ID.to_string()); @@ -173,7 +197,7 @@ where let inner = Arc::clone(&self.0); Box::pin(async move { let request = request.into_inner(); - let svid = inner.jwt_svid(request.audience)?; + let svid = inner.jwt_svid(&request.audience)?; Ok(Response::new(JwtsvidResponse { svids: vec![Jwtsvid { spiffe_id: inner.subject.to_string(), @@ -326,7 +350,81 @@ async fn start_spiffe_workload_api(path: &Path, subject: &str) -> FixtureHandle FixtureHandle { task } } -async fn start_gateway_token_endpoint(port: u16) -> FixtureHandle { +// Forms and JWT headers can span TCP reads. Read the bounded, Content-Length +// framed request completely so packet boundaries cannot change fixture results. +async fn read_http_request(stream: &mut TcpStream) -> Option { + tokio::time::timeout(Duration::from_secs(5), async { + let mut request = Vec::new(); + let mut buffer = [0_u8; 4096]; + loop { + let length = stream.read(&mut buffer).await.ok()?; + if length == 0 || request.len() + length > 32 * 1024 { + return None; + } + request.extend_from_slice(&buffer[..length]); + if let Some(end) = request.windows(4).position(|bytes| bytes == b"\r\n\r\n") { + let headers = std::str::from_utf8(&request[..end]).ok()?; + let content_length = headers + .lines() + .filter_map(|line| line.split_once(':')) + .find(|(name, _)| name.eq_ignore_ascii_case("content-length")) + .map_or(Some(0), |(_, value)| value.trim().parse::().ok())?; + if request.len() >= end.checked_add(4)?.checked_add(content_length)? { + return String::from_utf8(request).ok(); + } + } + } + }) + .await + .ok() + .flatten() +} + +fn verifies_jwt_svid(token: &str, instance: &str) -> bool { + let keys: jsonwebtoken::jwk::JwkSet = + serde_json::from_slice(&jwks()).expect("fixture JWKS should parse"); + let key = DecodingKey::from_jwk(keys.find(KEY_ID).expect("fixture signing key exists")) + .expect("fixture verification key should parse"); + let mut validation = Validation::new(Algorithm::RS256); + validation.set_issuer(&[ISSUER]); + validation.set_audience(&[format!("{IDENTITY_JWT_AUDIENCE}/{instance}")]); + validation.sub = Some(format!("spiffe://{TRUST_DOMAIN}/openshell/sandbox/e2e")); + validation.set_required_spec_claims(&["exp", "iat", "iss", "sub", "aud"]); + jsonwebtoken::decode::(token, &key, &validation) + .is_ok_and(|verified| verified.header.kid.as_deref() == Some(KEY_ID)) +} + +fn token_form(request: &str) -> HashMap { + let (_, body) = request.split_once("\r\n\r\n").unwrap_or_default(); + url::form_urlencoded::parse(body.as_bytes()) + .into_owned() + .collect() +} + +async fn write_token_response(stream: &mut TcpStream, access_token: Option<&str>, ttl: u64) { + let (status, body) = if let Some(access_token) = access_token { + ( + "HTTP/1.1 200 OK", + json!({"access_token": access_token, "token_type": "Bearer", "expires_in": ttl}) + .to_string(), + ) + } else { + ( + "HTTP/1.1 400 Bad Request", + json!({"error": "invalid_grant"}).to_string(), + ) + }; + let response = format!( + "{status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = stream.write_all(response.as_bytes()).await; +} + +async fn start_gateway_token_endpoint( + port: u16, + observations: Arc, +) -> FixtureHandle { let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, port))) .await .expect("bind gateway token endpoint"); @@ -335,51 +433,91 @@ async fn start_gateway_token_endpoint(port: u16) -> FixtureHandle { let Ok((mut stream, _peer)) = listener.accept().await else { break; }; + let observations = Arc::clone(&observations); tokio::spawn(async move { - let mut buf = vec![0_u8; 8192]; - let n = stream.read(&mut buf).await.unwrap_or(0); - let request = String::from_utf8_lossy(&buf[..n]); - let access_token = if request.starts_with("POST /token ") - && request.contains("subject_token=stored-user-token") - && request.contains("client_assertion=") - { - Some(INTERMEDIATE_TOKEN) - } else if request.starts_with("POST /token ") - && request.contains("subject_token=intermediate-token") - && request.contains("client_assertion=") - { - Some(FINAL_ACCESS_TOKEN) - } else { - None + let Some(request) = read_http_request(&mut stream).await else { + return; }; - let (status, body) = if let Some(access_token) = access_token { - ( - "HTTP/1.1 200 OK", - json!({ - "access_token": access_token, - "token_type": "Bearer", - "expires_in": 300 - }) - .to_string(), - ) - } else { - ( - "HTTP/1.1 400 Bad Request", - json!({"error": "unexpected_token_exchange"}).to_string(), - ) + let form = token_form(&request); + let field = |name: &str| form.get(name).map_or("", String::as_str); + let valid = request.starts_with("POST /token ") + && field("grant_type") == "urn:ietf:params:oauth:grant-type:token-exchange" + && field("client_assertion_type") == CLIENT_ASSERTION_TYPE + && field("subject_token_type") == TOKEN_TYPE_ACCESS_TOKEN + && field("requested_token_type") == TOKEN_TYPE_ACCESS_TOKEN + && !field("client_assertion").is_empty(); + let mut access_token = None; + for (index, instance) in ["a", "b"].into_iter().enumerate() { + if valid + && field("subject_token") == format!("{USER_SUBJECT_TOKEN}-{instance}") + && field("audience") + == format!("spiffe://{TRUST_DOMAIN}/openshell/sandbox/e2e") + && field("scope").is_empty() + { + access_token = Some(format!("{INTERMEDIATE_TOKEN}-{instance}")); + } else if valid + && field("subject_token") == format!("{INTERMEDIATE_TOKEN}-{instance}") + && field("scope") == "service.read" + { + if field("audience") == format!("service-{instance}") { + observations.service_issued[index].fetch_add(1, Ordering::SeqCst); + access_token = Some(format!("{FINAL_ACCESS_TOKEN}-{instance}")); + } else if field("audience") == format!("denied-service-{instance}") { + observations.service_denied.fetch_add(1, Ordering::SeqCst); + } + } + } + write_token_response(&mut stream, access_token.as_deref(), 600).await; + }); + } + }); + FixtureHandle { task } +} + +// This issuer returns the actual Workload API JWT-SVID after checking its +// signature and audience. No new production grant type or sandbox-visible +// credential is needed to exercise custom-header injection. +async fn start_identity_token_endpoint( + port: u16, + observations: Arc, +) -> FixtureHandle { + let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, port))) + .await + .expect("bind identity token endpoint"); + let task = tokio::spawn(async move { + loop { + let Ok((mut stream, _peer)) = listener.accept().await else { + break; + }; + let observations = Arc::clone(&observations); + tokio::spawn(async move { + let Some(request) = read_http_request(&mut stream).await else { + return; }; - let response = format!( - "{status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", - body.len() - ); - let _ = stream.write_all(response.as_bytes()).await; + let form = token_form(&request); + let field = |name: &str| form.get(name).map_or("", String::as_str); + let mut access_token = None; + for (index, instance) in ["a", "b"].into_iter().enumerate() { + let valid = request.starts_with("POST /identity-token ") + && field("grant_type") == "client_credentials" + && field("client_assertion_type") == IDENTITY_ASSERTION_TYPE + && field("scope") == "identity.read" + && verifies_jwt_svid(field("client_assertion"), instance); + if valid && field("audience") == format!("{IDENTITY_AUDIENCE}-{instance}") { + observations.identity_issued[index].fetch_add(1, Ordering::SeqCst); + access_token = Some(field("client_assertion")); + } else if valid && field("audience") == format!("denied-identity-{instance}") { + observations.identity_denied.fetch_add(1, Ordering::SeqCst); + } + } + write_token_response(&mut stream, access_token, 900).await; }); } }); FixtureHandle { task } } -async fn start_protected_target(port: u16) -> FixtureHandle { +async fn start_protected_target(port: u16, observations: Arc) -> FixtureHandle { let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::UNSPECIFIED, port))) .await .expect("bind protected target"); @@ -388,19 +526,49 @@ async fn start_protected_target(port: u16) -> FixtureHandle { let Ok((mut stream, _peer)) = listener.accept().await else { break; }; + let observations = Arc::clone(&observations); tokio::spawn(async move { - let mut buf = vec![0_u8; 8192]; - let n = stream.read(&mut buf).await.unwrap_or(0); - let request = String::from_utf8_lossy(&buf[..n]); - let ok = request.lines().any(|line| { - line.eq_ignore_ascii_case(&format!( - "authorization: Bearer {FINAL_ACCESS_TOKEN}" - )) + // Count any request bytes, including an incomplete header. A + // failed grant may open a TCP stream but must send no request. + let mut byte = [0_u8; 1]; + if !matches!( + tokio::time::timeout(Duration::from_secs(5), stream.peek(&mut byte)).await, + Ok(Ok(1)) + ) { + return; + } + observations.target_requests.fetch_add(1, Ordering::SeqCst); + let Some(request) = read_http_request(&mut stream).await else { + return; + }; + let headers = request.split("\r\n\r\n").next().unwrap_or_default(); + let header_values = |name: &str| { + headers + .lines() + .filter_map(|line| line.split_once(':')) + .filter(|(key, _)| key.eq_ignore_ascii_case(name)) + .map(|(_, value)| value.trim()) + .collect::>() + }; + let bearer = header_values("authorization"); + let identity = header_values("x-workload-jwt"); + let ok = ["a", "b"].into_iter().any(|instance| { + request.starts_with(&format!("GET /resource/{instance} ")) + && bearer.len() == 1 + && bearer[0] == format!("Bearer {FINAL_ACCESS_TOKEN}-{instance}") + && identity.len() == 1 + && verifies_jwt_svid(identity[0], instance) }); let (status, body) = if ok { - ("HTTP/1.1 200 OK", "token-exchange-ok") + ("HTTP/1.1 200 OK", "independent-grants-ok") } else { - ("HTTP/1.1 401 Unauthorized", "missing-final-token") + // Keep rejection evidence even if the workload retries and + // a later request happens to receive the correct credentials. + observations.target_rejected.fetch_add(1, Ordering::SeqCst); + ( + "HTTP/1.1 401 Unauthorized", + "credential-verification-failed", + ) }; let response = format!( "{status}\r\ncontent-type: text/plain\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", @@ -413,21 +581,19 @@ async fn start_protected_target(port: u16) -> FixtureHandle { FixtureHandle { task } } -async fn run_cli(args: &[&str]) -> Result { +async fn run_cli(args: &[&str]) -> Result<(), String> { let output = openshell_cmd() .args(args) .output() .await - .map_err(|err| format!("spawn openshell: {err}"))?; - let stdout = String::from_utf8_lossy(&output.stdout); - let stderr = String::from_utf8_lossy(&output.stderr); - let combined = format!("{stdout}{stderr}"); + .map_err(|_| "could not spawn openshell CLI".to_string())?; if output.status.success() { - Ok(combined) + Ok(()) } else { + // Provider creation arguments contain the stored subject token. + // Keep command arguments and raw diagnostics out of test failures. Err(format!( - "openshell {:?} failed with {:?}:\n{combined}", - args, + "openshell command failed; exit={:?}", output.status.code() )) } @@ -437,104 +603,13 @@ async fn run_cli_ignore_error(args: &[&str]) { let _ = openshell_cmd().args(args).output().await; } -async fn sandbox_logs(sandbox_name: &str) -> String { - run_cli(&["logs", sandbox_name]) - .await - .unwrap_or_else(|err| format!("failed to collect sandbox logs: {err}")) -} - -async fn podman_exec_capture(container_name: &str, args: &[&str]) -> String { - let Ok(socket) = std::env::var("OPENSHELL_PODMAN_SOCKET") else { - return "OPENSHELL_PODMAN_SOCKET is not set".to_string(); - }; - let mut cmd = Command::new("podman"); - cmd.arg("--url") - .arg(format!("unix://{socket}")) - .arg("exec") - .arg(container_name) - .args(args); - apply_podman_config_env(&mut cmd); - match cmd.output().await { - Ok(output) => format!( - "exit={:?}\nstdout:\n{}\nstderr:\n{}", - output.status.code(), - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ), - Err(err) => format!("failed to run podman exec {:?}: {err}", args), - } -} - -async fn podman_logs_capture(container_name: &str) -> String { - let Ok(socket) = std::env::var("OPENSHELL_PODMAN_SOCKET") else { - return "OPENSHELL_PODMAN_SOCKET is not set".to_string(); - }; - let mut cmd = Command::new("podman"); - cmd.arg("--url").arg(format!("unix://{socket}")).args([ - "logs", - "--tail", - "200", - container_name, - ]); - apply_podman_config_env(&mut cmd); - match cmd.output().await { - Ok(output) => format!( - "exit={:?}\nstdout:\n{}\nstderr:\n{}", - output.status.code(), - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ), - Err(err) => format!("failed to run podman logs: {err}"), - } -} - -async fn provider_token_debug(sandbox_name: &str, target_port: u16) -> String { - let sandbox_logs = sandbox_logs(sandbox_name).await; - let Ok(socket) = std::env::var("OPENSHELL_PODMAN_SOCKET") else { - return format!("Sandbox logs:\n{sandbox_logs}\nOPENSHELL_PODMAN_SOCKET is not set"); - }; - let container_name = match podman_container_name_for_sandbox(&socket, sandbox_name).await { - Ok(name) => name, - Err(err) => return format!("Sandbox logs:\n{sandbox_logs}\n{err}"), - }; - let env = podman_exec_capture(&container_name, &["env"]).await; - let hosts = podman_exec_capture(&container_name, &["cat", "/etc/hosts"]).await; - let processes = podman_exec_capture(&container_name, &["ps", "-ef"]).await; - let resolve_host = podman_exec_capture( - &container_name, - &[ - "python3", - "-c", - "import socket; print(socket.getaddrinfo('host.openshell.internal', 0, type=socket.SOCK_STREAM))", - ], - ) - .await; - let target_probe = podman_exec_capture( - &container_name, - &[ - "python3", - "-c", - &format!( - "import socket; s=socket.create_connection(('host.openshell.internal', {target_port}), 2); print('connected', s.getpeername()); s.close()" - ), - ], - ) - .await; - let container_logs = podman_logs_capture(&container_name).await; - - format!( - "Sandbox logs:\n{sandbox_logs}\n\ - Container: {container_name}\n\ - --- podman env ---\n{env}\n\ - --- /etc/hosts ---\n{hosts}\n\ - --- ps -ef ---\n{processes}\n\ - --- resolve host.openshell.internal ---\n{resolve_host}\n\ - --- protected target TCP probe ---\n{target_probe}\n\ - --- podman logs ---\n{container_logs}" - ) -} - -fn write_profile(profile_type: &str, token_port: u16, target_port: u16) -> NamedTempFile { +fn write_profile( + profile_type: &str, + token_port: u16, + identity_port: u16, + target_port: u16, + instance: &str, +) -> NamedTempFile { let token_endpoint = format!("http://127.0.0.1:{token_port}/token"); let mut file = tempfile::Builder::new() .suffix(".yaml") @@ -543,7 +618,7 @@ fn write_profile(profile_type: &str, token_port: u16, target_port: u16) -> Named let profile = format!( r"id: {profile_type} display_name: Podman token exchange e2e -description: Podman e2e provider profile for two-stage token exchange +description: Independent bearer and JWT-SVID grants for one request category: other credentials: - name: subject_token @@ -557,18 +632,39 @@ credentials: token_grant: grant_type: token_exchange token_endpoint: {token_endpoint} - audience: final-audience + audience: service-{instance} + scopes: [service.read] jwt_svid_audience: {token_endpoint} client_assertion_type: {CLIENT_ASSERTION_TYPE} requested_token_type: {TOKEN_TYPE_ACCESS_TOKEN} - cache_ttl_seconds: 30 + cache_ttl_seconds: 300 + audience_overrides: + - path: /deny-service/{instance} + audience: denied-service-{instance} subject_token: source: provider_credential credential: subject_token subject_token_type: {TOKEN_TYPE_ACCESS_TOKEN} + - name: workload_identity + description: Signed workload identity for the identity proxy + required: false + auth_style: header + header_name: X-Workload-Jwt + token_grant: + grant_type: client_credentials + token_endpoint: http://127.0.0.1:{identity_port}/identity-token + audience: {IDENTITY_AUDIENCE}-{instance} + scopes: [identity.read] + jwt_svid_audience: {IDENTITY_JWT_AUDIENCE}/{instance} + client_assertion_type: {IDENTITY_ASSERTION_TYPE} + cache_ttl_seconds: 600 + audience_overrides: + - path: /deny-identity/{instance} + audience: denied-identity-{instance} endpoints: - host: host.openshell.internal port: {target_port} + path: /** protocol: rest access: read-write enforcement: enforce @@ -595,64 +691,56 @@ while true; do sleep 60; done .to_string() } -async fn podman_container_name_for_sandbox( - socket: &str, +async fn sandbox_exec_http( sandbox_name: &str, + target_port: u16, + path: &str, + expect_denied: bool, ) -> Result { - let mut cmd = Command::new("podman"); - cmd.arg("--url") - .arg(format!("unix://{socket}")) - .arg("ps") - .arg("--filter") - .arg(format!("label=openshell.ai/sandbox-name={sandbox_name}")) - .arg("--filter") - .arg("label=openshell.io/isolation-role=sandbox") - .arg("--format") - .arg("{{.Names}}"); - apply_podman_config_env(&mut cmd); - let output = cmd - .output() - .await - .map_err(|err| format!("spawn podman ps for sandbox container: {err}"))?; - if !output.status.success() { - return Err(format!( - "podman ps for sandbox container failed: {}{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - )); - } - let names = String::from_utf8_lossy(&output.stdout) - .lines() - .map(str::trim) - .filter(|line| !line.is_empty()) - .map(str::to_string) - .collect::>(); - match names.as_slice() { - [name] => Ok(name.clone()), - [] => Err(format!( - "no running Podman container found for sandbox '{sandbox_name}'" - )), - _ => Err(format!( - "multiple running Podman containers found for sandbox '{sandbox_name}': {}", - names.join(", ") - )), - } -} - -fn apply_podman_config_env(cmd: &mut Command) { - if std::env::var_os("OPENSHELL_E2E_CONTAINER_ENGINE_UNSET_XDG_CONFIG_HOME").is_some() { - cmd.env_remove("XDG_CONFIG_HOME"); - } else if let Some(value) = std::env::var_os("OPENSHELL_E2E_CONTAINER_ENGINE_XDG_CONFIG_HOME") { - cmd.env("XDG_CONFIG_HOME", value); - } -} - -async fn sandbox_exec_http(sandbox_name: &str, target_port: u16) -> Result { - let url = format!("http://host.openshell.internal:{target_port}/resource"); + let url = format!("http://host.openshell.internal:{target_port}{path}"); + let expect_denied_python = if expect_denied { "True" } else { "False" }; let script = format!( - "import urllib.request; print(urllib.request.urlopen({url:?}, timeout=5).read().decode())" + r#"import base64, json, os, re, urllib.error, urllib.request + +def contains_credential(value): + if any(marker in value for marker in ( + "stored-user-token", "intermediate-token", "final-access-token", "openshell:resolve:" + )): + return True + for token in re.findall(r"[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+", value): + try: + payload = token.split(".")[1] + claims = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4))) + if claims.get("iss") == "{ISSUER}": + return True + except (ValueError, UnicodeError): + pass + return False + +if any(contains_credential(value) for value in os.environ.values()): + raise RuntimeError("provider credential material reached the workload environment") +request = urllib.request.Request({url:?}, headers={{ + "Authorization": "Bearer agent-supplied", + "X-Workload-Jwt": "agent-supplied", +}}) +try: + with urllib.request.urlopen(request, timeout=10) as response: + body = response.read().decode() + if contains_credential(body): + raise RuntimeError("provider credential material reached the workload response") + if {expect_denied_python}: + raise RuntimeError("failed grant unexpectedly reached the target") + print(body) +except urllib.error.HTTPError as error: + body = error.read().decode() + if contains_credential(body): + raise RuntimeError("provider credential material reached the workload error") + if not {expect_denied_python} or error.code != 502: + raise RuntimeError("unexpected HTTP response") from None + print("grant-denied") +"# ); - let mut last_output = String::new(); + let mut last_status = None; for _ in 0..20 { let output = openshell_cmd() .args([ @@ -668,26 +756,61 @@ async fn sandbox_exec_http(sandbox_name: &str, target_port: u16) -> Result String { + let name = format!("podman-grants-{instance}-{}", std::process::id()); + run_cli_ignore_error(&["provider", "delete", &name, "--yes"]).await; + run_cli_ignore_error(&["profile", "delete", &name, "--yes"]).await; + let profile = write_profile(&name, token_port, identity_port, target_port, instance); + let profile_path = profile + .path() + .to_str() + .expect("profile path should be UTF-8"); + run_cli(&["profile", "import", "-f", profile_path]) + .await + .expect("import independent-grant profile"); + run_cli(&[ + "provider", + "create", + "--name", + &name, + "--type", + &name, + "--credential", + &format!("subject_token={USER_SUBJECT_TOKEN}-{instance}"), + ]) + .await + .expect("create provider instance"); + name } #[tokio::test] -async fn podman_provider_token_exchange_injects_bearer_header() { +async fn podman_provider_token_exchange_injects_independent_grants_across_sandboxes() { let gateway_socket = PathBuf::from( std::env::var("OPENSHELL_E2E_GATEWAY_SPIFFE_SOCKET") .expect("OPENSHELL_E2E_GATEWAY_SPIFFE_SOCKET must be set by e2e-podman.sh"), @@ -696,71 +819,102 @@ async fn podman_provider_token_exchange_injects_bearer_header() { std::env::var("OPENSHELL_E2E_PROVIDER_SPIFFE_SOCKET") .expect("OPENSHELL_E2E_PROVIDER_SPIFFE_SOCKET must be set by e2e-podman.sh"), ); - - let profile_type = format!("podman-token-exchange-e2e-{}", std::process::id()); - let provider_name = format!("podman-token-exchange-e2e-{}", std::process::id()); let token_port = find_free_port(); + let identity_port = find_free_port(); let target_port = find_free_port(); - let token_endpoint = format!("http://127.0.0.1:{token_port}/token"); let gateway_subject = format!("spiffe://{TRUST_DOMAIN}/openshell/gateway"); + // The mock shares a workload subject. Distinct providers and JWT audiences + // test cross-sandbox credential isolation, not production SPIRE attestation. let supervisor_subject = format!("spiffe://{TRUST_DOMAIN}/openshell/sandbox/e2e"); - + let observations = Arc::new(GrantObservations::default()); let _gateway_spiffe = start_spiffe_workload_api(&gateway_socket, &gateway_subject).await; let _provider_spiffe = start_spiffe_workload_api(&provider_socket, &supervisor_subject).await; - let _gateway_token = start_gateway_token_endpoint(token_port).await; - let _target = start_protected_target(target_port).await; - - run_cli_ignore_error(&["provider", "delete", &provider_name, "--yes"]).await; - run_cli_ignore_error(&["profile", "delete", &profile_type, "--yes"]).await; - - let profile = write_profile(&profile_type, token_port, target_port); - let profile_path = profile - .path() - .to_str() - .expect("profile path should be UTF-8"); - run_cli(&["profile", "import", "-f", profile_path]) - .await - .expect("import provider profile"); - run_cli(&[ - "provider", - "create", - "--name", - &provider_name, - "--type", - &profile_type, - "--credential", - &format!("subject_token={USER_SUBJECT_TOKEN}"), - ]) - .await - .expect("create provider"); + let _gateway_token = start_gateway_token_endpoint(token_port, Arc::clone(&observations)).await; + let _identity_token = + start_identity_token_endpoint(identity_port, Arc::clone(&observations)).await; + let _target = start_protected_target(target_port, Arc::clone(&observations)).await; + + let mut provider_names = Vec::new(); + for instance in ["a", "b"] { + provider_names + .push(create_provider_instance(token_port, identity_port, target_port, instance).await); + } let script = sandbox_script(); - let mut sandbox = SandboxGuard::create_keep_with_args( - &["--provider", &provider_name], + let mut sandbox_a = SandboxGuard::create_keep_with_args( + &["--provider", &provider_names[0]], &["sh", "-lc", &script], "token-server-ready", ) .await - .unwrap_or_else(|err| { - panic!( - "sandbox should complete token exchange against {token_endpoint} and protected target port {target_port}:\n{err}" - ) - }); - let request_output = match sandbox_exec_http(&sandbox.name, target_port).await { - Ok(output) => output, - Err(err) => { - let debug = provider_token_debug(&sandbox.name, target_port).await; - panic!("request protected target from kept sandbox: {err}\n{debug}"); - } - }; + .expect("create sandbox A"); + let mut sandbox_b = SandboxGuard::create_keep_with_args( + &["--provider", &provider_names[1]], + &["sh", "-lc", &script], + "token-server-ready", + ) + .await + .expect("create sandbox B"); + + // Both supervisors request the same host/port concurrently. The target + // verifies each path's distinct bearer and signed JWT audience together. + let initial = tokio::join!( + sandbox_exec_http(&sandbox_a.name, target_port, "/resource/a", false), + sandbox_exec_http(&sandbox_b.name, target_port, "/resource/b", false), + ); + let before_cache = observations.issued(); + let cached = tokio::join!( + sandbox_exec_http(&sandbox_a.name, target_port, "/resource/a", false), + sandbox_exec_http(&sandbox_b.name, target_port, "/resource/b", false), + ); + let after_cache = observations.issued(); + let target_requests = observations.target_requests.load(Ordering::SeqCst); + // An audience override forces acquisition of only the failing credential; + // the other credential remains cached from the successful request. + let denied = tokio::join!( + sandbox_exec_http(&sandbox_a.name, target_port, "/deny-service/a", true), + sandbox_exec_http(&sandbox_b.name, target_port, "/deny-identity/b", true), + ); - run_cli_ignore_error(&["provider", "delete", &provider_name, "--yes"]).await; - run_cli_ignore_error(&["profile", "delete", &profile_type, "--yes"]).await; - sandbox.cleanup().await; + sandbox_a.cleanup().await; + sandbox_b.cleanup().await; + for name in &provider_names { + run_cli_ignore_error(&["provider", "delete", name, "--yes"]).await; + run_cli_ignore_error(&["profile", "delete", name, "--yes"]).await; + } + for outcome in [initial.0, initial.1, cached.0, cached.1] { + assert_eq!( + outcome.expect("independent grant request should succeed"), + "independent-grants-ok" + ); + } + assert_eq!( + observations.target_rejected.load(Ordering::SeqCst), + 0, + "no request may present mismatched or untrusted credentials, even before a retry" + ); assert!( - request_output.contains("token-exchange-ok"), - "protected target should receive the final exchanged bearer token:\n{}", - request_output + before_cache + .iter() + .all(|(service, identity)| *service > 0 && *identity > 0), + "each provider instance must acquire both credentials" + ); + assert_eq!( + after_cache, before_cache, + "repeat requests should use both caches" + ); + for outcome in [denied.0, denied.1] { + assert_eq!( + outcome.expect("failed grant should return 502"), + "grant-denied" + ); + } + assert!(observations.service_denied.load(Ordering::SeqCst) > 0); + assert!(observations.identity_denied.load(Ordering::SeqCst) > 0); + assert_eq!( + observations.target_requests.load(Ordering::SeqCst), + target_requests, + "neither failed grant may forward any request bytes" ); }