diff --git a/.secrets.baseline b/.secrets.baseline index 4d8c4e22..38a1c0a6 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -1,9 +1,9 @@ { "exclude": { - "files": "(?x)(Cargo\\.lock$|\\.lock$)|^\\.secrets\\.baseline$", + "files": "(?x)(Cargo\\.lock$|\\.lock$)|^\\.secrets\\.baseline$|^.secrets.baseline$", "lines": null }, - "generated_at": "2026-09-02T15:42:38Z", + "generated_at": "2026-09-03T17:12:20Z", "plugins_used": [ { "name": "AWSKeyDetector" @@ -80,371 +80,373 @@ "assets/contextforgeCA/contextforge-client.key.pem": [ { "hashed_secret": "1348b145fa1a555461c1b790a2f66614781091e9", + "is_secret": false, "is_verified": false, "line_number": 1, "type": "Private Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "assets/contextforgeCA/contextforge-server.key.pem": [ { "hashed_secret": "1348b145fa1a555461c1b790a2f66614781091e9", + "is_secret": false, "is_verified": false, "line_number": 1, "type": "Private Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "assets/contextforgeCA/contextforge.ca.key.pem": [ { "hashed_secret": "1348b145fa1a555461c1b790a2f66614781091e9", + "is_secret": false, "is_verified": false, "line_number": 1, "type": "Private Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "assets/contextforgeCA/contextforge.intermediate.key.pem": [ { "hashed_secret": "1348b145fa1a555461c1b790a2f66614781091e9", + "is_secret": false, "is_verified": false, "line_number": 1, "type": "Private Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "assets/jwt.key": [ { "hashed_secret": "be4fc4886bd949b369d5e092eb87494f12e57e5b", + "is_secret": false, "is_verified": false, "line_number": 1, "type": "Private Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "assets/tls_key.pem": [ { "hashed_secret": "1348b145fa1a555461c1b790a2f66614781091e9", + "is_secret": false, "is_verified": false, "line_number": 1, "type": "Private Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "crates/contextforge-data-plane-lib/src/common.rs": [ { - "hashed_secret": "4a4645604f0b9e29503be96a87f6f47a6e4a7890", - "is_verified": false, - "line_number": 154, - "type": "Secret Keyword", - "verified_result": null, - "is_secret": false - }, - { - "hashed_secret": "427f5e1b530d4a544883308d876a11d724060c86", + "hashed_secret": "bfc6000db1195a9522813fc405c666dd4ce669ad", + "is_secret": false, "is_verified": false, - "line_number": 157, + "line_number": 242, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { - "hashed_secret": "bfc6000db1195a9522813fc405c666dd4ce669ad", + "hashed_secret": "4a4645604f0b9e29503be96a87f6f47a6e4a7890", + "is_secret": false, "is_verified": false, - "line_number": 263, + "line_number": 271, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "crates/contextforge-data-plane-lib/src/telemetry.rs": [ { "hashed_secret": "0a24796d4c71ce722a92f450f69dc36c60b21de4", + "is_secret": false, "is_verified": false, "line_number": 87, "type": "Hex High Entropy String", - "verified_result": null, - "is_secret": false + "verified_result": null + } + ], + "crates/contextforge-data-plane-lib/tests/gateway_pagination.rs": [ + { + "hashed_secret": "2eceb4c6d4592232a10c3b018108b6b974226001", + "is_secret": false, + "is_verified": false, + "line_number": 69, + "type": "Secret Keyword", + "verified_result": null } ], "crates/contextforge-data-plane-lib/tests/support/client.rs": [ { "hashed_secret": "a453c8b2640819a451ce875ac1e04d0dbab7b403", + "is_secret": false, "is_verified": false, "line_number": 12, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "crates/contextforge-data-plane-lib/tests/support/mod.rs": [ { "hashed_secret": "a453c8b2640819a451ce875ac1e04d0dbab7b403", + "is_secret": false, "is_verified": false, - "line_number": 17, + "line_number": 20, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "crates/contextforge-data-plane/Cargo.toml": [ { "hashed_secret": "58e7dc38ba3a7d4a720006d2f3cc4cda774d89dc", + "is_secret": false, "is_verified": false, "line_number": 20, "type": "Hex High Entropy String", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "crates/plugins/cpex-secrets-detection/src/lib.rs": [ { "hashed_secret": "86de8c52637ec530fe39b0a8471da9b8764d5242", + "is_secret": false, "is_verified": false, "line_number": 610, "type": "AWS Access Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "crates/plugins/cpex-secrets-detection/src/scanner.rs": [ { "hashed_secret": "9249e2590f5d19742260cb5296cb76fe0677f147", + "is_secret": false, "is_verified": false, "line_number": 238, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "199da8f71b7dced64f82cf6e96483134cace9b14", + "is_secret": false, "is_verified": false, "line_number": 239, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "c0026c4c848882618c987859077ffbae92130625", + "is_secret": false, "is_verified": false, "line_number": 242, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "078553dc10635837abb80f404302c70cba91b879", + "is_secret": false, "is_verified": false, "line_number": 278, "type": "Base64 High Entropy String", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "e175c6f5f2a92e8623bd9a4820edb4e8c1b0fd10", + "is_secret": false, "is_verified": false, "line_number": 278, "type": "GitHub Token", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "97d99a51e5ac827bb36fe6273facfda35245917a", + "is_secret": false, "is_verified": false, "line_number": 279, "type": "Base64 High Entropy String", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "be4fc4886bd949b369d5e092eb87494f12e57e5b", + "is_secret": false, "is_verified": false, "line_number": 282, "type": "Private Key", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "b1775a785f09a6ebaf2dc33d6eaeb98974d9cdb8", + "is_secret": false, "is_verified": false, "line_number": 284, "type": "Hex High Entropy String", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "eae9124e42e2ef05ba727bd1a1c0c6fa61a05b9e", + "is_secret": false, "is_verified": false, "line_number": 302, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "86de8c52637ec530fe39b0a8471da9b8764d5242", + "is_secret": false, "is_verified": false, "line_number": 401, "type": "AWS Access Key", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "9d7235fe33b6612ed7ebca4b63afd00d4adf5d66", + "is_secret": false, "is_verified": false, "line_number": 410, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "27a39044bff80a4c196689dfa8dcf129cb27fef8", + "is_secret": false, "is_verified": false, "line_number": 431, "type": "Base64 High Entropy String", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "crates/plugins/cpex-secrets-detection/tests/plugin_manager.rs": [ { "hashed_secret": "436da7d4d22c39c0165ab0d5b40073d0f2fc11c5", + "is_secret": false, "is_verified": false, "line_number": 197, "type": "AWS Access Key", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "8b4510a576d82f38bd2730436bf5e20c4e15b30e", + "is_secret": false, "is_verified": false, "line_number": 198, "type": "AWS Access Key", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "e4ea017859bcad962c8ab551fe29da9147877eee", + "is_secret": false, "is_verified": false, "line_number": 199, "type": "AWS Access Key", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "86de8c52637ec530fe39b0a8471da9b8764d5242", + "is_secret": false, "is_verified": false, "line_number": 268, "type": "AWS Access Key", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "docker/docker-compose-langfuse.yaml": [ { "hashed_secret": "cb1fde0682fbd1ac0faf2a9f297167ac9d06434b", + "is_secret": false, "is_verified": false, "line_number": 16, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "cb58df830a45cc33df1a313e616ecad78cd796c5", + "is_secret": false, "is_verified": false, "line_number": 77, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "2e0c522bfe4e7885492862df2e0b987c0ca02623", + "is_secret": false, "is_verified": false, "line_number": 100, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "d9d007c8de197b3f36a3a0ba4f13c0f7df175d5a", + "is_secret": false, "is_verified": false, "line_number": 255, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "docker/docker-compose.yml": [ { "hashed_secret": "2a8bfc0ce436d55ca907d0162989481bcb7677b4", + "is_secret": false, "is_verified": false, "line_number": 189, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "fdda45b7f6d2ead95d9991fc4678640c3bab0d84", + "is_secret": false, "is_verified": false, "line_number": 363, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "093d378410a5cfa4bd5088f3fef62fbdb8a95665", + "is_secret": false, "is_verified": false, "line_number": 369, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "c3de40d5e3fc71ed62771c2127a8e42585026c97", + "is_secret": false, "is_verified": false, "line_number": 371, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "4d4acd9b084d13f5fdb23807d857e1c48a1cfd0f", + "is_secret": false, "is_verified": false, "line_number": 460, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "bd0160c2cf35d950843c88f3be2b9412ed71f485", + "is_secret": false, "is_verified": false, "line_number": 495, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null }, { "hashed_secret": "293324f6824bb3a6db5c4dc42a60ddd4a9851c99", + "is_secret": false, "is_verified": false, "line_number": 658, "type": "Hex High Entropy String", - "verified_result": null, - "is_secret": false + "verified_result": null } ], "scripts/git/resolve-secrets-baseline-conflict.sh": [ { "hashed_secret": "44ffd1bfb94772d5f91d528e7aca703990edbbd7", + "is_secret": false, "is_verified": false, "line_number": 31, "type": "Secret Keyword", - "verified_result": null, - "is_secret": false + "verified_result": null } ] }, @@ -453,4 +455,4 @@ "file": null, "hash": null } -} \ No newline at end of file +} diff --git a/Cargo.lock b/Cargo.lock index 64485c8b..5ec2daff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -623,7 +623,6 @@ dependencies = [ "rmp-serde", "rustls", "rustls-pki-types", - "secret-string", "serde", "serde_json", "test-log", @@ -2489,12 +2488,6 @@ dependencies = [ "syn 3.0.3", ] -[[package]] -name = "secret-string" -version = "0.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "068d960589a13890bdb2a9ea5e7ae0974b89249101117f2e9a5e166bf3af44b9" - [[package]] name = "security-framework" version = "3.7.0" diff --git a/_context/wiki/security.md b/_context/wiki/security.md index 551eb11c..ff261aa5 100644 --- a/_context/wiki/security.md +++ b/_context/wiki/security.md @@ -119,5 +119,5 @@ These routes are registered **outside the authentication middleware** — unauth ## Secrets Handling -- The HMAC secret is held as a `SecretString`; key and certificate material is read from disk paths at startup. +- JWT validation keys are fetched from a remote JWKS endpoint; TLS certificate material is read from disk paths at startup. - Never log: tokens, authorization headers, secrets, Redis key/value bytes, full `UserConfig` documents, or backend credentials. diff --git a/crates/contextforge-data-plane-lib/Cargo.toml b/crates/contextforge-data-plane-lib/Cargo.toml index 681dabab..c601175c 100644 --- a/crates/contextforge-data-plane-lib/Cargo.toml +++ b/crates/contextforge-data-plane-lib/Cargo.toml @@ -46,7 +46,8 @@ rustls-pki-types = { version = "1.14.1", features = ["std", "alloc"] } tokio-rustls = "0.26.4" typed-builder = "0.23.2" url = { workspace = true, features = ["serde"] } -secret-string = "0.0.2" + + [features] @@ -54,6 +55,7 @@ default = [] with_tools = [] + [dev-dependencies] opentelemetry_sdk.workspace = true cpex.workspace = true @@ -62,6 +64,7 @@ cpex-secrets-detection.workspace = true test-log = "0.2.20" axum-server = { version = "0.8.0", features = ["tls-rustls"] } + [lints] workspace = true diff --git a/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks.rs b/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks.rs new file mode 100644 index 00000000..33e16fc9 --- /dev/null +++ b/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks.rs @@ -0,0 +1,182 @@ +use std::time::Duration; + +use futures::StreamExt as _; +use jsonwebtoken::{ + Algorithm, AlgorithmFamily, DecodingKey, Header, Validation, decode, + jwk::{Jwk, JwkSet, KeyOperations, PublicKeyUse}, +}; +use lru_time_cache::LruCache; + +use reqwest::Url; +use serde_json::Value; +use tokio::sync::RwLock; +use tracing::debug; +use typed_builder::TypedBuilder; + +use crate::authorization::{ + AuthorizationClaims, AuthorizationError, + jwks::principal::{DefaultPrincipalExtractor, PrincipalExtractor}, +}; + +pub const JWKS_CACHE_TTL: Duration = Duration::from_mins(5); +pub const JWKS_CACHE_KEY: &str = "jwks"; + +const JWKS_MAX_RESPONSE_BYTES: usize = 1024 * 1024; + +#[derive(TypedBuilder)] +pub(super) struct Jwks +where + T: PrincipalExtractor, +{ + client: reqwest::Client, + url: Url, + #[builder(default = RwLock::new(LruCache::with_expiry_duration(JWKS_CACHE_TTL)))] + cache: RwLock>>, + #[builder(default = false)] + validate_audience: bool, + #[builder(default = true)] + validate_expiry: bool, + #[builder(default = true)] + validate_not_before: bool, + principal_extractor: T, +} + +impl Jwks { + fn validation(&self, alg: Algorithm) -> Validation { + let mut validation = Validation::new(alg); + validation.required_spec_claims.clear(); + validation.validate_aud = self.validate_audience; + validation.validate_exp = self.validate_expiry; + validation.validate_nbf = self.validate_not_before; + validation + } + + pub async fn validate(&self, token: &str, header: &Header) -> Option { + { + let cache = self.cache.read().await; + if let Some(keys) = cache.peek(JWKS_CACHE_KEY) + && keys.iter().any(|key| key.matches(header)) + { + return self.validate_with_keys(keys, token, header, &self.validation(header.alg)); + } + } + + match fetch_jwks(&self.client, &self.url).await { + Ok(keys) => { + let key_count = keys.len(); + let claims = self.validate_with_keys(&keys, token, header, &self.validation(header.alg)); + self.cache.write().await.insert(JWKS_CACHE_KEY.to_owned(), keys); + tracing::info!("validate: SaaS JWKS cache refreshed {key_count}"); + + claims + }, + Err(error) => { + tracing::info!("validate: unable to refresh SaaS JWKS {error:?}"); + None + }, + } + } + + fn validate_with_keys( + &self, + keys: &[VerificationKey], + token: &str, + header: &Header, + validation: &Validation, + ) -> Option { + keys.iter() + .filter(|key| key.matches(header)) + .find_map(|key| self.validate_and_decode_claims(token, &key.decoding_key, validation)) + } + + fn validate_and_decode_claims( + &self, + token: &str, + key: &DecodingKey, + validation: &Validation, + ) -> Option { + let claims = decode::(token, key, validation) + .inspect_err(|e| { + debug!("validate_and_decode_claims: problem {e:?}"); + }) + .ok()? + .claims; + let claims = claims.as_object()?; + let user_id = self.principal_extractor.user_id(claims)?; + let tenant_id = self.principal_extractor.tenant_id(claims)?; + + Some(AuthorizationClaims::new(user_id, tenant_id)) + } +} + +pub struct VerificationKey { + pub(crate) key_id: Option, + pub(crate) decoding_key: DecodingKey, +} + +impl VerificationKey { + fn from_jwk(jwk: Jwk) -> Result, AuthorizationError> { + if jwk.common.public_key_use.as_ref().is_some_and(|key_use| key_use != &PublicKeyUse::Signature) + || jwk.common.key_operations.as_ref().is_some_and(|operations| !operations.contains(&KeyOperations::Verify)) + { + return Ok(None); + } + + let decoding_key = DecodingKey::from_jwk(&jwk).map_err(AuthorizationError::InvalidKey)?; + if !matches!(decoding_key.family(), AlgorithmFamily::Rsa | AlgorithmFamily::Ec) { + return Ok(None); + } + + Ok(Some(Self { key_id: jwk.common.key_id, decoding_key })) + } + + pub(super) fn matches(&self, header: &Header) -> bool { + self.decoding_key.family() == header.alg.family() + && header + .kid + .as_ref() + .is_none_or(|header_key_id| self.key_id.as_ref().is_none_or(|key_id| key_id == header_key_id)) + } +} + +async fn fetch_jwks(client: &reqwest::Client, url: &Url) -> Result, AuthorizationError> { + let response = client + .get(url.clone()) + .send() + .await + .and_then(reqwest::Response::error_for_status) + .map_err(AuthorizationError::JwksRequest)?; + if response + .content_length() + .is_some_and(|length| length > u64::try_from(JWKS_MAX_RESPONSE_BYTES).unwrap_or(u64::MAX)) + { + return Err(AuthorizationError::JwksResponseTooLarge); + } + let mut body = Vec::new(); + let mut chunks = response.bytes_stream(); + while let Some(chunk) = chunks.next().await { + let chunk = chunk.map_err(AuthorizationError::JwksRequest)?; + if chunk.len() > JWKS_MAX_RESPONSE_BYTES.saturating_sub(body.len()) { + return Err(AuthorizationError::JwksResponseTooLarge); + } + body.extend_from_slice(&chunk); + } + let jwks = serde_json::from_slice::(&body).map_err(AuthorizationError::InvalidJson)?; + if jwks.keys.is_empty() { Ok(Vec::new()) } else { validated_json_web_keys(jwks.keys) } +} + +pub(super) fn validated_json_web_keys( + jwks: impl IntoIterator, +) -> Result, AuthorizationError> { + let mut keys = Vec::new(); + for jwk in jwks { + if let Some(key) = VerificationKey::from_jwk(jwk)? { + keys.push(key); + } + } + + if keys.is_empty() { + return Err(AuthorizationError::NoSupportedKeys); + } + Ok(keys) +} diff --git a/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs b/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs new file mode 100644 index 00000000..392f0166 --- /dev/null +++ b/crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs @@ -0,0 +1,377 @@ +use crate::authorization::jwks::jwks::Jwks; +use crate::authorization::jwks::principal::DefaultPrincipalExtractor; +use crate::authorization::{AuthorizationClaims, AuthorizationError, AuthorizationService}; +use async_trait::async_trait; +use jsonwebtoken::decode_header; +use std::fmt; +use std::net::IpAddr; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use url::Url; + +const JWKS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const JWKS_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const JWKS_READ_TIMEOUT: Duration = Duration::from_secs(5); + +pub struct JwtAuthorizationService { + jwks: Jwks, +} + +impl JwtAuthorizationService { + pub fn from_jwks_url(jwks_url: Url, ca_cert_path: Option<&PathBuf>) -> Result { + let url = parse_jwks_url(jwks_url)?; + let mut client = reqwest::Client::builder() + .tls_backend_rustls() + .connect_timeout(JWKS_CONNECT_TIMEOUT) + .read_timeout(JWKS_READ_TIMEOUT) + .timeout(JWKS_REQUEST_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) + .user_agent(concat!("mcp-ops/", env!("CARGO_PKG_VERSION"))); + if let Some(ca_cert_path) = ca_cert_path { + client = client.tls_certs_only(load_ca_certificates(ca_cert_path)?); + } + let client = client.build().map_err(AuthorizationError::JwksRequest)?; + Ok(Self { + jwks: Jwks::builder().client(client).url(url).principal_extractor(DefaultPrincipalExtractor {}).build(), + }) + } + + async fn authorize_token(&self, token: &str) -> Option { + let header = decode_header(token).ok()?; + self.jwks.validate(token, &header).await + } +} + +impl fmt::Debug for JwtAuthorizationService { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("JwtAuthorizationService") + .field("verification_source", &"remote JWKS") + .finish_non_exhaustive() + } +} + +#[async_trait] +impl AuthorizationService for JwtAuthorizationService { + async fn authorize(&self, authorization_token: &http::HeaderValue) -> Option { + let token = authorization_token.as_bytes().strip_prefix(b"Bearer ")?; + let token = str::from_utf8(token).ok()?; + let claims = self.authorize_token(token).await; + + if claims.is_none() { + tracing::debug!("validate_saas_jwt SaaS JWT was rejected"); + } + + claims + } +} + +fn parse_jwks_url(url: Url) -> Result { + let secure = url.scheme() == "https"; + let local_http = url.scheme() == "http" + && url.host_str().is_some_and(|host| { + host.eq_ignore_ascii_case("localhost") || host.parse::().is_ok_and(|address| address.is_loopback()) + }); + if !secure && !local_http { + return Err(AuthorizationError::InsecureJwksUrl); + } + Ok(url) +} + +fn load_ca_certificates(path: &Path) -> Result, AuthorizationError> { + let pem = std::fs::read(path) + .map_err(|source| AuthorizationError::ReadJwksCaCertificate { path: path.to_owned(), source })?; + let certificates = reqwest::Certificate::from_pem_bundle(&pem) + .map_err(|source| AuthorizationError::InvalidJwksCaCertificate { path: path.to_owned(), source })?; + if certificates.is_empty() { + return Err(AuthorizationError::EmptyJwksCaCertificate { path: path.to_owned() }); + } + Ok(certificates) +} + +#[cfg(test)] +mod test { + use crate::authorization::{ + AuthorizationError, + jwks::{ + JwtAuthorizationService, + jwks::{JWKS_CACHE_KEY, Jwks, VerificationKey}, + jwks_authorization::{JWKS_CONNECT_TIMEOUT, JWKS_READ_TIMEOUT, JWKS_REQUEST_TIMEOUT}, + principal::DefaultPrincipalExtractor, + }, + }; + use crate::{ + Config, + authorization::{AuthorizationClaims, Scopes}, + common::ContextForgeDataPlaneAppState, + layers::claims_id::claims_layer, + user_config_store::{ConfigStoreError, UserConfigStore}, + }; + use async_trait::async_trait; + use axum::{Router, body::Body, middleware, response::Response, routing::get}; + + use contextforge_data_plane_apis::{User, user_store::UserConfig}; + use http::{HeaderMap, Request, StatusCode}; + use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, encode}; + use lru_time_cache::LruCache; + + use std::sync::{Arc, Once}; + use std::{str::FromStr, time::Duration}; + use tokio::sync::RwLock; + use tower::ServiceExt; + + use url::Url; + use uuid::Uuid; + + const GATEWAY_AUDIENCE: &str = "audience"; + const GATEWAY_ISSUER: &str = "issuer"; + + impl VerificationKey { + pub fn new(id: Option, decoding_key: DecodingKey) -> Self { + Self { key_id: id, decoding_key } + } + } + + impl JwtAuthorizationService { + pub async fn from_keys(verification_keys: Vec) -> Result { + let url: Url = Url::from_str("http://127.0.0.1:0/").expect("this should work"); + let client = reqwest::Client::builder() + .tls_backend_rustls() + .connect_timeout(JWKS_CONNECT_TIMEOUT) + .read_timeout(JWKS_READ_TIMEOUT) + .timeout(JWKS_REQUEST_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) + .user_agent(concat!("mcp-ops/", env!("CARGO_PKG_VERSION"))); + + let client = client.build().map_err(AuthorizationError::JwksRequest)?; + + let cache = RwLock::new(LruCache::with_expiry_duration(Duration::from_hours(100))); + let mut guard = cache.write().await; + guard.insert(JWKS_CACHE_KEY.to_owned(), verification_keys); + drop(guard); + + Ok(Self { + jwks: Jwks::builder() + .cache(cache) + .client(client) + .url(url) + .principal_extractor(DefaultPrincipalExtractor {}) + .build(), + }) + } + } + + static CRYPTO: Once = Once::new(); + const HMAC_SECRET: &[u8] = b"my-test-key-but-now-longer-than-32-bytes"; + + fn now_epoch_seconds() -> u64 { + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).expect("Time went backwards").as_secs() + } + + fn active_test_claims() -> AuthorizationClaims { + let now = now_epoch_seconds(); + let user_id = "11111111-1111-1111-1111-111111111111".to_owned(); + + AuthorizationClaims { + iss: GATEWAY_ISSUER.to_owned(), + sub: user_id.clone(), + aud: GATEWAY_AUDIENCE.to_owned(), + exp: now + Duration::from_hours(1).as_secs(), + nbf: Some(now - Duration::from_mins(1).as_secs()), + iat: Some(now), + jti: Uuid::new_v4().to_string(), + token_use: Some("api".to_owned()), + teams: Some(vec!["team_awesome".to_owned()]), + user: Some( + crate::authorization::User::builder() + .tenant_id("team_awesome".to_owned()) + .user_id(user_id.clone()) + .build(), + ), + scopes: Some( + Scopes::builder() + .server_id(Some("my_id".to_owned())) + .ip_restrictions(vec!["192.169.1.0/24".to_owned()]) + .permissions(vec!["tools.read".to_owned(), "servers.use".to_owned()]) + .time_restrictions(None) + .build(), + ), + tenant_id: "tenant".to_owned(), + ..Default::default() + } + } + + fn get_hmac_token_for_claims(claims: &AuthorizationClaims) -> String { + let key = EncodingKey::from_secret(HMAC_SECRET); + let header = Header::new(Algorithm::HS256); + + encode::(&header, claims, &key).expect("Expecting this to work") + } + + struct MockedUserConfigStore; + #[async_trait] + impl UserConfigStore for MockedUserConfigStore { + async fn get_config<'a>(&self, _: &'a User) -> Result { + Err(ConfigStoreError::InvalidConnection) + } + + async fn set_config<'a>(&self, _: &'a User, _: &'a UserConfig) -> Result<(), ConfigStoreError> { + Err(ConfigStoreError::InvalidConnection) + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[allow(clippy::items_after_statements)] + #[test_log::test] + async fn claim_test_valid_hmac() { + CRYPTO.call_once(|| { + _ = rustls::crypto::ring::default_provider().install_default(); + }); + + async fn handle(_: HeaderMap) -> Response { + Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") + } + + let token = get_hmac_token_for_claims(&active_test_claims()); + + let decoding_key = DecodingKey::from_secret(HMAC_SECRET); + let verfication_key = VerificationKey::new(Some("HS256".to_owned()), decoding_key); + + let state = ContextForgeDataPlaneAppState { + authorization_service: Arc::new( + JwtAuthorizationService::from_keys(vec![verfication_key]).await.expect("this should work"), + ), + config_store: Arc::new(MockedUserConfigStore {}), + config: Config::default(), + }; + let http_requst = Request::builder() + .header("Authorization", format!("Bearer {token}")) + .method("GET") + .body(Body::empty()) + .expect("This should work"); + + let app = + Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); + + let res = app.oneshot(http_requst).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[allow(clippy::items_after_statements)] + async fn claim_test_missing_scopes_is_allowed() { + CRYPTO.call_once(|| { + _ = rustls::crypto::ring::default_provider().install_default(); + }); + + async fn handle(_: HeaderMap) -> Response { + Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") + } + + let mut claims = active_test_claims(); + claims.scopes = None; + let token = get_hmac_token_for_claims(&claims); + + let decoding_key = DecodingKey::from_secret(HMAC_SECRET); + let verfication_key = VerificationKey::new(Some("HS256".to_owned()), decoding_key); + let state = ContextForgeDataPlaneAppState { + authorization_service: Arc::new( + JwtAuthorizationService::from_keys(vec![verfication_key]).await.expect("this should work"), + ), + config_store: Arc::new(MockedUserConfigStore {}), + config: Config::default(), + }; + let http_requst = Request::builder() + .header("Authorization", format!("Bearer {token}")) + .method("GET") + .body(Body::empty()) + .expect("This should work"); + + let app = + Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); + + let res = app.oneshot(http_requst).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[allow(clippy::items_after_statements)] + async fn claim_test_missing_token_use_and_full_name_is_allowed() { + CRYPTO.call_once(|| { + _ = rustls::crypto::ring::default_provider().install_default(); + }); + let user_id = "11111111-1111-1111-1111-111111111111".to_owned(); + + async fn handle(_: HeaderMap) -> Response { + Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") + } + + let mut claims = active_test_claims(); + claims.token_use = None; + + claims.user = Some( + crate::authorization::User::builder().tenant_id("team_awesome".to_owned()).user_id(user_id.clone()).build(), + ); + let token = get_hmac_token_for_claims(&claims); + + let decoding_key = DecodingKey::from_secret(HMAC_SECRET); + let verfication_key = VerificationKey::new(Some("HS256".to_owned()), decoding_key); + + let state = ContextForgeDataPlaneAppState { + authorization_service: Arc::new( + JwtAuthorizationService::from_keys(vec![verfication_key]).await.expect("this should work"), + ), + config_store: Arc::new(MockedUserConfigStore {}), + config: Config::default(), + }; + let http_requst = Request::builder() + .header("Authorization", format!("Bearer {token}")) + .method("GET") + .body(Body::empty()) + .expect("This should work"); + + let app = + Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); + + let res = app.oneshot(http_requst).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[allow(clippy::items_after_statements)] + async fn claim_test_expired_token() { + CRYPTO.call_once(|| { + _ = rustls::crypto::ring::default_provider().install_default(); + }); + + async fn handle(_: HeaderMap) -> Response { + Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") + } + + let mut claims = active_test_claims(); + claims.exp = 0; + let token = get_hmac_token_for_claims(&claims); + + let decoding_key = DecodingKey::from_secret(HMAC_SECRET); + let verfication_key = VerificationKey::new(Some("HS256".to_owned()), decoding_key); + + let state = ContextForgeDataPlaneAppState { + authorization_service: Arc::new( + JwtAuthorizationService::from_keys(vec![verfication_key]).await.expect("this should work"), + ), + config_store: Arc::new(MockedUserConfigStore {}), + config: Config::default(), + }; + let http_requst = Request::builder() + .header("Authorization", format!("Bearer {token}")) + .method("GET") + .body(Body::empty()) + .expect("This should work"); + + let app = + Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); + + let res = app.oneshot(http_requst).await.unwrap(); + assert_eq!(res.status(), StatusCode::UNAUTHORIZED); + } +} diff --git a/crates/contextforge-data-plane-lib/src/authorization/jwks/mod.rs b/crates/contextforge-data-plane-lib/src/authorization/jwks/mod.rs new file mode 100644 index 00000000..02cf598f --- /dev/null +++ b/crates/contextforge-data-plane-lib/src/authorization/jwks/mod.rs @@ -0,0 +1,6 @@ +#[allow(clippy::module_inception)] +mod jwks; +mod jwks_authorization; +mod principal; + +pub use jwks_authorization::JwtAuthorizationService; diff --git a/crates/contextforge-data-plane-lib/src/authorization/jwks/principal.rs b/crates/contextforge-data-plane-lib/src/authorization/jwks/principal.rs new file mode 100644 index 00000000..dc7575da --- /dev/null +++ b/crates/contextforge-data-plane-lib/src/authorization/jwks/principal.rs @@ -0,0 +1,19 @@ +use serde_json::Value; + +pub struct DefaultPrincipalExtractor {} + +pub trait PrincipalExtractor { + fn user_id<'a>(&self, claims: &'a serde_json::Map) -> Option<&'a str> { + ["sub", "user_id", "UserId"].into_iter().find_map(|claim| claims.get(claim)).and_then(non_empty_string) + } + + fn tenant_id<'a>(&self, claims: &'a serde_json::Map) -> Option<&'a str> { + ["tenantId", "tenant_id"].into_iter().find_map(|claim| claims.get(claim).and_then(non_empty_string)) + } +} + +impl PrincipalExtractor for DefaultPrincipalExtractor {} + +fn non_empty_string(value: &Value) -> Option<&str> { + value.as_str().filter(|value| !value.trim().is_empty()) +} diff --git a/crates/contextforge-data-plane-lib/src/authorization/mod.rs b/crates/contextforge-data-plane-lib/src/authorization/mod.rs new file mode 100644 index 00000000..6d00c433 --- /dev/null +++ b/crates/contextforge-data-plane-lib/src/authorization/mod.rs @@ -0,0 +1,135 @@ +use std::{path::PathBuf, sync::Arc}; + +use async_trait::async_trait; +use chrono::Duration; +use http::HeaderValue; +use serde::{Deserialize, Serialize}; +use typed_builder::TypedBuilder; + +use crate::Config; + +mod jwks; + +pub const AUDIENCE: &str = "audience"; +pub const ISSUER: &str = "issuer"; + +pub fn get_authorization_service( + config: &Config, +) -> Result, AuthorizationError> { + let service = + jwks::JwtAuthorizationService::from_jwks_url(config.jwks_url.clone(), config.jwks_ca_cert_path.as_ref())?; + Ok(Arc::new(service) as Arc) +} + +#[async_trait] +pub trait AuthorizationService: std::fmt::Debug { + async fn authorize(&self, authorization_token: &HeaderValue) -> Option; +} + +#[derive(Debug, thiserror::Error)] +#[allow(dead_code)] +pub enum AuthorizationError { + #[error("SaaS JWKS contains no supported signing keys")] + NoSupportedKeys, + #[error("SaaS JWKS is invalid")] + InvalidJson(#[source] serde_json::Error), + #[error("SaaS JWKS is invalid")] + InvalidKey(#[source] jsonwebtoken::errors::Error), + + #[error("MCPOPS_JWKS_URL must use HTTPS (HTTP is allowed only for loopback testing)")] + InsecureJwksUrl, + #[error("unable to retrieve SaaS JWKS")] + JwksRequest(#[source] reqwest::Error), + #[error("unable to read JWKS CA certificate `{path}`")] + ReadJwksCaCertificate { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("JWKS CA certificate `{path}` is invalid")] + InvalidJwksCaCertificate { + path: PathBuf, + #[source] + source: reqwest::Error, + }, + #[error("JWKS CA certificate `{path}` contains no certificates")] + EmptyJwksCaCertificate { path: PathBuf }, + #[error("SaaS JWKS response exceeds 1 MiB")] + JwksResponseTooLarge, +} + +#[derive(Clone, Debug, Serialize, Deserialize, TypedBuilder, PartialEq)] +pub struct User { + pub user_id: String, + pub tenant_id: String, +} + +impl From for User { + fn from(claims: AuthorizationClaims) -> Self { + Self { user_id: claims.idp_unique_id.clone(), tenant_id: claims.tenant_id.clone() } + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, TypedBuilder, PartialEq)] +pub struct Scopes { + server_id: Option, + permissions: Vec, + ip_restrictions: Vec, + time_restrictions: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize, TypedBuilder, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct Idp { + real_name: String, + iss: String, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Default, TypedBuilder)] +#[serde(rename_all = "camelCase")] +pub struct AuthorizationClaims { + pub iss: String, + pub jti: String, + pub aud: String, + pub exp: u64, + pub iat: Option, + pub nbf: Option, + pub tenant_id: String, + pub subscription_id: String, + pub sub: String, + pub entity_type: String, + pub email: Option, + pub name: Option, + pub displayname: Option, + pub idp: Option, + pub groups: Option>, + pub roles: Option>, + pub idp_unique_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub teams: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub user: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scopes: Option, + pub token_use: Option, +} + +impl AuthorizationClaims { + pub fn new(user_id: &str, tenant_id: &str) -> Self { + let audience = AUDIENCE.to_owned(); + let start = std::time::SystemTime::now(); + let now = start.duration_since(std::time::UNIX_EPOCH).expect("Time went backwards").as_secs(); + Self { + iss: ISSUER.to_owned(), + sub: user_id.to_owned(), + aud: audience, + exp: now + Duration::hours(1).num_seconds().cast_unsigned(), + iat: Some(now), + nbf: Some(now - Duration::minutes(5).num_seconds().cast_unsigned()), + idp_unique_id: user_id.to_owned(), + tenant_id: tenant_id.to_owned(), + groups: Some(vec!["team_awesome".to_owned()]), + ..Default::default() + } + } +} diff --git a/crates/contextforge-data-plane-lib/src/common.rs b/crates/contextforge-data-plane-lib/src/common.rs index 064f8edf..eed6b8f4 100644 --- a/crates/contextforge-data-plane-lib/src/common.rs +++ b/crates/contextforge-data-plane-lib/src/common.rs @@ -3,7 +3,6 @@ use http::uri::Authority; use jsonwebtoken::DecodingKey; use redis::{ConnectionAddr, IntoConnectionInfo, RedisError}; use rustls_pki_types::{CertificateDer, PrivatePkcs8KeyDer, pem::PemObject}; -use secret_string::SecretString; use serde::{Deserialize, Serialize}; use std::{ fs::{self, File}, @@ -16,7 +15,7 @@ use thiserror::Error; use typed_builder::TypedBuilder; use url::Url; -use crate::user_config_store::UserConfigStore; +use crate::{authorization::AuthorizationService, user_config_store::UserConfigStore}; #[derive(Clone)] pub struct JwtTokenDecoders { @@ -27,7 +26,7 @@ pub struct JwtTokenDecoders { #[allow(unused)] #[derive(Clone)] pub struct ContextForgeDataPlaneAppState { - pub(crate) jwt_token_decoding_keys: JwtTokenDecoders, + pub(crate) authorization_service: Arc, pub(crate) config_store: Arc, pub(crate) config: Config, } @@ -49,22 +48,6 @@ pub struct Scopes { time_restrictions: Option, } -#[derive(Clone, Debug, Serialize, Deserialize, TypedBuilder)] -pub struct ContextForgeClaims { - pub sub: String, - pub jti: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub token_use: Option, - pub iat: Option, - pub iss: String, - pub aud: String, - pub exp: u64, - pub teams: Option>, - pub user: User, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub scopes: Option, -} - pub type RedisClient = redis::Client; #[derive(Debug, Clone)] @@ -139,22 +122,18 @@ pub enum OtlpProtocol { HttpProtobuf, } -#[derive(Debug, Clone, Parser, Default)] +#[derive(Debug, Clone, Parser)] #[command(name = "contextforge-data-plane")] #[command(about = "Minimal, fast, experimental data plane for ContextForge")] pub struct Config { #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_ADDRESS")] pub address: Option, - #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_TOKEN_VERIFICATION_PUBLIC_KEY")] - pub token_verification_public_key: Option, - - #[cfg(feature = "with_tools")] - #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_TOKEN_VERIFICATION_PRIVATE_KEY")] - pub token_verification_private_key: PathBuf, + #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_JWKS_URL")] + pub jwks_url: url::Url, - #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_TOKEN_SECRET")] - pub token_verification_secret: Option>, + #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_JWKS_CA_PATH")] + pub jwks_ca_cert_path: Option, #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_ENABLE_OPEN_TELEMETRY")] pub enable_open_telemetry: Option, @@ -286,6 +265,10 @@ pub struct Config { num_args = 1.. )] pub mcp_allowed_hosts: Option>, + + #[cfg(feature = "with_tools")] + #[arg(long)] + pub token_verification_private_key: PathBuf, } pub const DEFAULT_MCP_STANDARD_HEADER_MAX_COUNT: usize = 32; @@ -432,3 +415,55 @@ fn extract_identity(config: &Config) -> crate::Result { _ => Err("Invalid/missing configuration".into()), } } + +#[cfg(test)] +mod tests { + #[cfg(feature = "with_tools")] + use std::{path::PathBuf, str::FromStr}; + + use crate::Config; + + impl Default for Config { + fn default() -> Self { + Self { + address: None, + jwks_url: "http://127.0.0.1:8080/".parse().expect("should work"), + jwks_ca_cert_path: None, + enable_open_telemetry: None, + otlp_endpoint: None, + otlp_protocol: None, + otlp_headers: None, + otlp_service_name: None, + enable_otel_metrics: None, + otlp_metrics_endpoint: None, + mcp_standard_header_max_count: 10, + mcp_standard_header_max_value_bytes: 4096, + mcp_standard_header_max_total_bytes: 4096, + number_of_cpus: None, + single_runtime: None, + runtime_plugins_enabled: None, + tls_address: None, + server_private_key: None, + server_certificate: None, + upstream_connection_mode: None, + upstream_private_key: None, + upstream_certificate: None, + upstream_trust_bundle: None, + user_config_cache_expiry_seconds: 10, + redis_address: String::new(), + redis_port: 0, + redis_mode: super::RedisConnectionMode::PlainText, + redis_tls_trust_bundle: None, + redis_tls_client_private_key: None, + redis_tls_client_certificate: None, + log_name: None, + log_rotation: None, + mcp_allowed_origins: None, + mcp_allowed_hosts: None, + + #[cfg(feature = "with_tools")] + token_verification_private_key: PathBuf::from_str("./assets/jwt.key").expect("This should work"), + } + } + } +} diff --git a/crates/contextforge-data-plane-lib/src/const_values.rs b/crates/contextforge-data-plane-lib/src/const_values.rs index 3f5e83b4..1e7a1bea 100644 --- a/crates/contextforge-data-plane-lib/src/const_values.rs +++ b/crates/contextforge-data-plane-lib/src/const_values.rs @@ -1,4 +1,3 @@ pub const LRU_CACHE_ENTRIES: usize = 50_000; -pub const CONTEXT_FORGE_GATEWAY_AUDIENCE: &str = "mcpgateway-api"; -pub const CONTEXT_FORGE_GATEWAY_ISSUER: &str = "mcpgateway"; + pub const REDIS_RETRIES: usize = 1000; // keep re-trying forver diff --git a/crates/contextforge-data-plane-lib/src/errors.rs b/crates/contextforge-data-plane-lib/src/errors.rs new file mode 100644 index 00000000..2c3ec1c2 --- /dev/null +++ b/crates/contextforge-data-plane-lib/src/errors.rs @@ -0,0 +1,22 @@ +use axum::response::Response; +use http::{StatusCode, header}; + +pub(crate) fn unauthorized_response(message: &str) -> Response { + custom_error(StatusCode::UNAUTHORIZED, message) +} + +pub(crate) fn bad_request(message: &str) -> Response { + custom_error(StatusCode::BAD_REQUEST, message) +} + +pub(crate) fn internal_server_error(message: &str) -> Response { + custom_error(StatusCode::INTERNAL_SERVER_ERROR, message) +} + +pub(crate) fn custom_error(status: StatusCode, message: &str) -> Response { + Response::builder() + .status(status) + .header(header::CONTENT_TYPE, "text/plain") + .body(message.to_owned().into()) + .expect("Expecting this to work") +} diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs index fd435f83..0c3aff69 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs @@ -3,7 +3,7 @@ use http::request::Parts; use rmcp::{ErrorData, RoleServer, model::ErrorCode, service::RequestContext}; use tracing::debug; -use crate::{common::ContextForgeClaims, layers::virtual_host_id::VirtualHostId}; +use crate::{authorization::AuthorizationClaims, layers::virtual_host_id::VirtualHostId}; pub struct AuthorizedCallValidator<'a> { call_name: &'a str, @@ -15,10 +15,10 @@ impl<'a> AuthorizedCallValidator<'a> { Self { call_name, ctx } } - pub fn validate_stateless(self) -> Result<(&'a VirtualHost, &'a ContextForgeClaims), ErrorData> { + pub fn validate_stateless(self) -> Result<(&'a VirtualHost, &'a AuthorizationClaims), ErrorData> { let maybe_parts = self.ctx.extensions.get::(); let maybe_user_config = maybe_parts.and_then(|parts| parts.extensions.get::()); - let maybe_claims = maybe_parts.and_then(|parts| parts.extensions.get::()); + let maybe_claims = maybe_parts.and_then(|parts| parts.extensions.get::()); let maybe_virtual_host_id = maybe_parts.and_then(|parts| parts.extensions.get::()); let call_name = self.call_name; let has_user_config = maybe_user_config.is_some(); diff --git a/crates/contextforge-data-plane-lib/src/layers/claims_id.rs b/crates/contextforge-data-plane-lib/src/layers/claims_id.rs index 7cdd9875..4f4cc9ed 100644 --- a/crates/contextforge-data-plane-lib/src/layers/claims_id.rs +++ b/crates/contextforge-data-plane-lib/src/layers/claims_id.rs @@ -3,300 +3,23 @@ use axum::{ middleware::Next, response::Response, }; -use http::{StatusCode, header}; -use jsonwebtoken::Validation; -use crate::{ - common::{ContextForgeClaims, ContextForgeDataPlaneAppState}, - const_values::{CONTEXT_FORGE_GATEWAY_AUDIENCE, CONTEXT_FORGE_GATEWAY_ISSUER}, -}; - -fn unauthorized_response(message: &str) -> Response { - Response::builder() - .status(StatusCode::UNAUTHORIZED) - .header(header::CONTENT_TYPE, "text/plain") - .body(message.to_owned().into()) - .expect("Expecting this to work") -} +use crate::{common::ContextForgeDataPlaneAppState, errors::unauthorized_response}; pub async fn claims_layer( State(state): State, request: http::Request, next: Next, ) -> Response { - let decoding_keys = state.jwt_token_decoding_keys; let (mut parts, body) = request.into_parts(); let Some(authorization) = parts.headers.get("Authorization") else { return unauthorized_response("No header") }; - let Some(token) = authorization.as_bytes().strip_prefix(b"Bearer ") else { + let Some(claims) = state.authorization_service.authorize(authorization).await else { return unauthorized_response("Invalid token"); }; - let Ok(raw_token) = str::from_utf8(token) else { return unauthorized_response("Invalid token encoding") }; - - let Ok(header) = jsonwebtoken::decode_header(raw_token) else { return unauthorized_response("Invalid header") }; - - let mut validation = Validation::new(header.alg); - validation.set_audience(&[CONTEXT_FORGE_GATEWAY_AUDIENCE]); - validation.set_issuer(&[CONTEXT_FORGE_GATEWAY_ISSUER]); - let claims = match header.alg { - jsonwebtoken::Algorithm::RS256 | jsonwebtoken::Algorithm::RS384 | jsonwebtoken::Algorithm::RS512 => { - let Some(decoding_key) = decoding_keys.rs.as_ref() else { - return Response::builder() - .status(StatusCode::UNAUTHORIZED) - .header(header::CONTENT_TYPE, "text/plain") - .body("Invalid key".into()) - .expect("Expecting this to work"); - }; - let maybe_valid = jsonwebtoken::decode::(raw_token, decoding_key, &validation); - let Ok(claims) = maybe_valid else { - return unauthorized_response(&format!("Invalid claims {maybe_valid:?}")); - }; - claims - }, - jsonwebtoken::Algorithm::HS256 | jsonwebtoken::Algorithm::HS384 | jsonwebtoken::Algorithm::HS512 => { - let Some(decoding_key) = decoding_keys.hmac_sha.as_ref() else { - return unauthorized_response("Invalid decoding key"); - }; - let maybe_valid = jsonwebtoken::decode::(raw_token, decoding_key, &validation); - let Ok(claims) = maybe_valid else { return unauthorized_response("Invalid claims") }; - - claims - }, - - _ => { - return unauthorized_response("Invalid algorithm"); - }, - }; - - let claims: ContextForgeClaims = claims.claims; parts.extensions.insert(claims.clone()); let request = Request::from_parts(parts, body); next.run(request).await } - -#[cfg(test)] -mod test { - - use std::sync::{Arc, Once}; - - use async_trait::async_trait; - use axum::{Router, body::Body, middleware, response::Response, routing::get}; - use chrono::Duration; - use contextforge_data_plane_apis::{User, user_store::UserConfig}; - use http::{HeaderMap, Request, StatusCode}; - use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, encode}; - use tower::ServiceExt; - use uuid::Uuid; - - use crate::{ - Config, - common::{self, ContextForgeClaims, ContextForgeDataPlaneAppState, JwtTokenDecoders, Scopes}, - const_values::{CONTEXT_FORGE_GATEWAY_AUDIENCE, CONTEXT_FORGE_GATEWAY_ISSUER}, - layers::claims_id::claims_layer, - user_config_store::{ConfigStoreError, UserConfigStore}, - }; - - static CRYPTO: Once = Once::new(); - const HMAC_SECRET: &[u8] = b"my-test-key-but-now-longer-than-32-bytes"; - - fn now_epoch_seconds() -> u64 { - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).expect("Time went backwards").as_secs() - } - - fn active_test_claims() -> ContextForgeClaims { - let now = now_epoch_seconds(); - let user_id = "11111111-1111-1111-1111-111111111111".to_owned(); - let user_email = "admin@example.com".to_owned(); - - ContextForgeClaims { - iss: CONTEXT_FORGE_GATEWAY_ISSUER.to_owned(), - sub: user_id.clone(), - aud: CONTEXT_FORGE_GATEWAY_AUDIENCE.to_owned(), - exp: now + Duration::hours(1).num_seconds().cast_unsigned(), - iat: Some(now), - jti: Uuid::new_v4().to_string(), - token_use: Some("api".to_owned()), - teams: Some(vec!["team_awesome".to_owned()]), - user: common::User::builder() - .email(user_email) - .auth_provider("api_token".to_owned()) - .full_name(Some("API Token User".to_owned())) - .is_admin(true) - .build(), - scopes: Some( - Scopes::builder() - .server_id(Some("my_id".to_owned())) - .ip_restrictions(vec!["192.169.1.0/24".to_owned()]) - .permissions(vec!["tools.read".to_owned(), "servers.use".to_owned()]) - .time_restrictions(None) - .build(), - ), - } - } - - fn get_hmac_token_for_claims(claims: &ContextForgeClaims) -> String { - let key = EncodingKey::from_secret(HMAC_SECRET); - let header = Header::new(Algorithm::HS256); - - encode::(&header, claims, &key).expect("Expecting this to work") - } - - struct MockedUserConfigStore; - #[async_trait] - impl UserConfigStore for MockedUserConfigStore { - async fn get_config<'a>(&self, _: &'a User) -> Result { - Err(ConfigStoreError::InvalidConnection) - } - - async fn set_config<'a>(&self, _: &'a User, _: &'a UserConfig) -> Result<(), ConfigStoreError> { - Err(ConfigStoreError::InvalidConnection) - } - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] - #[allow(clippy::items_after_statements)] - async fn claim_test_valid_hmac() { - CRYPTO.call_once(|| { - _ = rustls::crypto::ring::default_provider().install_default(); - }); - - async fn handle(_: HeaderMap) -> Response { - Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") - } - - let token = get_hmac_token_for_claims(&active_test_claims()); - - let decoding_key = DecodingKey::from_secret(HMAC_SECRET); - - let state = ContextForgeDataPlaneAppState { - jwt_token_decoding_keys: JwtTokenDecoders { rs: None, hmac_sha: Some(decoding_key) }, - config_store: Arc::new(MockedUserConfigStore {}), - config: Config::default(), - }; - let http_requst = Request::builder() - .header("Authorization", format!("Bearer {token}")) - .method("GET") - .body(Body::empty()) - .expect("This should work"); - - let app = - Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); - - let res = app.oneshot(http_requst).await.unwrap(); - assert_eq!(res.status(), StatusCode::OK); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] - #[allow(clippy::items_after_statements)] - async fn claim_test_missing_scopes_is_allowed() { - CRYPTO.call_once(|| { - _ = rustls::crypto::ring::default_provider().install_default(); - }); - - async fn handle(_: HeaderMap) -> Response { - Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") - } - - let mut claims = active_test_claims(); - claims.scopes = None; - let token = get_hmac_token_for_claims(&claims); - - let decoding_key = DecodingKey::from_secret(HMAC_SECRET); - - let state = ContextForgeDataPlaneAppState { - jwt_token_decoding_keys: JwtTokenDecoders { rs: None, hmac_sha: Some(decoding_key) }, - config_store: Arc::new(MockedUserConfigStore {}), - config: Config::default(), - }; - let http_requst = Request::builder() - .header("Authorization", format!("Bearer {token}")) - .method("GET") - .body(Body::empty()) - .expect("This should work"); - - let app = - Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); - - let res = app.oneshot(http_requst).await.unwrap(); - assert_eq!(res.status(), StatusCode::OK); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] - #[allow(clippy::items_after_statements)] - async fn claim_test_missing_token_use_and_full_name_is_allowed() { - CRYPTO.call_once(|| { - _ = rustls::crypto::ring::default_provider().install_default(); - }); - - async fn handle(_: HeaderMap) -> Response { - Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") - } - - let user_email = "admin@example.com".to_owned(); - let mut claims = active_test_claims(); - claims.token_use = None; - claims.user = common::User::builder() - .email(user_email) - .auth_provider("local".to_owned()) - .full_name(None) - .is_admin(true) - .build(); - let token = get_hmac_token_for_claims(&claims); - - let decoding_key = DecodingKey::from_secret(HMAC_SECRET); - - let state = ContextForgeDataPlaneAppState { - jwt_token_decoding_keys: JwtTokenDecoders { rs: None, hmac_sha: Some(decoding_key) }, - config_store: Arc::new(MockedUserConfigStore {}), - config: Config::default(), - }; - let http_requst = Request::builder() - .header("Authorization", format!("Bearer {token}")) - .method("GET") - .body(Body::empty()) - .expect("This should work"); - - let app = - Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); - - let res = app.oneshot(http_requst).await.unwrap(); - assert_eq!(res.status(), StatusCode::OK); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] - #[allow(clippy::items_after_statements)] - async fn claim_test_expired_token() { - CRYPTO.call_once(|| { - _ = rustls::crypto::ring::default_provider().install_default(); - }); - - async fn handle(_: HeaderMap) -> Response { - Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") - } - - let mut claims = active_test_claims(); - claims.exp = 0; - let token = get_hmac_token_for_claims(&claims); - - let decoding_key = DecodingKey::from_secret(HMAC_SECRET); - - let state = ContextForgeDataPlaneAppState { - jwt_token_decoding_keys: JwtTokenDecoders { rs: None, hmac_sha: Some(decoding_key) }, - config_store: Arc::new(MockedUserConfigStore {}), - config: Config::default(), - }; - let http_requst = Request::builder() - .header("Authorization", format!("Bearer {token}")) - .method("GET") - .body(Body::empty()) - .expect("This should work"); - - let app = - Router::new().route("/", get(handle)).layer(middleware::from_fn_with_state(state.clone(), claims_layer)); - - let res = app.oneshot(http_requst).await.unwrap(); - assert_eq!(res.status(), StatusCode::UNAUTHORIZED); - } -} diff --git a/crates/contextforge-data-plane-lib/src/layers/mcp_header_limits.rs b/crates/contextforge-data-plane-lib/src/layers/mcp_header_limits.rs index d5e7b9bb..8a4c2b79 100644 --- a/crates/contextforge-data-plane-lib/src/layers/mcp_header_limits.rs +++ b/crates/contextforge-data-plane-lib/src/layers/mcp_header_limits.rs @@ -1,21 +1,22 @@ -use axum::{body::Body, extract::State, middleware::Next, response::Response}; -use http::{StatusCode, header}; +use axum::{extract::State, middleware::Next, response::Response}; +use http::StatusCode; use tracing::debug; use crate::common::{ Config, DEFAULT_MCP_STANDARD_HEADER_MAX_COUNT, DEFAULT_MCP_STANDARD_HEADER_MAX_TOTAL_BYTES, DEFAULT_MCP_STANDARD_HEADER_MAX_VALUE_BYTES, }; +use crate::errors::custom_error; use crate::mcp_standard_headers; #[derive(Clone, Debug, PartialEq, Eq)] -pub(crate) struct McpStandardHeaderLimits { +pub(crate) struct StandardHeaderLimits { pub(crate) count: usize, pub(crate) value_bytes: usize, pub(crate) total_bytes: usize, } -impl From<&Config> for McpStandardHeaderLimits { +impl From<&Config> for StandardHeaderLimits { fn from(config: &Config) -> Self { Self { count: configured_or_default(config.mcp_standard_header_max_count, DEFAULT_MCP_STANDARD_HEADER_MAX_COUNT), @@ -36,14 +37,14 @@ fn configured_or_default(configured: usize, default: usize) -> usize { } #[derive(Clone, Copy, Debug, PartialEq, Eq)] -struct McpStandardHeaderUsage { +struct StandardHeaderUsage { count: usize, value_bytes: usize, total_bytes: usize, } pub(crate) async fn mcp_header_limits_layer( - State(limits): State, + State(limits): State, request: http::Request, next: Next, ) -> Response { @@ -54,24 +55,20 @@ pub(crate) async fn mcp_header_limits_layer( debug!( "mcp_header_limits_layer - rejecting request count = {count} value_bytes = {value_bytes} total_bytes = {total_bytes}" ); - return Response::builder() - .status(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE) - .header(header::CONTENT_TYPE, "text/plain") - .body(Body::from("MCP standard header limits exceeded")) - .expect("Expecting this to work"); + return custom_error(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE, "MCP standard header limits exceeded"); } next.run(request).await } -fn exceeded_limits(headers: &http::HeaderMap, limits: &McpStandardHeaderLimits) -> Option { +fn exceeded_limits(headers: &http::HeaderMap, limits: &StandardHeaderLimits) -> Option { let usage = mcp_standard_header_usage(headers); usage.exceeds(limits).then_some(usage) } -fn mcp_standard_header_usage(headers: &http::HeaderMap) -> McpStandardHeaderUsage { - let mut usage = McpStandardHeaderUsage { count: 0, value_bytes: 0, total_bytes: 0 }; +fn mcp_standard_header_usage(headers: &http::HeaderMap) -> StandardHeaderUsage { + let mut usage = StandardHeaderUsage { count: 0, value_bytes: 0, total_bytes: 0 }; for (name, value) in headers.iter().filter(|(name, _)| mcp_standard_headers::is_limited(name)) { let value_bytes = value.as_bytes().len(); @@ -85,8 +82,8 @@ fn mcp_standard_header_usage(headers: &http::HeaderMap) -> McpStandardHeaderUsag usage } -impl McpStandardHeaderUsage { - fn exceeds(self, limits: &McpStandardHeaderLimits) -> bool { +impl StandardHeaderUsage { + fn exceeds(self, limits: &StandardHeaderLimits) -> bool { self.count > limits.count || self.value_bytes > limits.value_bytes || self.total_bytes > limits.total_bytes } } @@ -98,15 +95,16 @@ mod tests { use async_trait::async_trait; use axum::{Router, body::Body, middleware, response::Response, routing::get}; use contextforge_data_plane_apis::{User, user_store::UserConfig}; - use http::{Request, StatusCode}; + use http::{HeaderValue, Request, StatusCode}; use tower::ServiceExt; use crate::{ Config, - common::{ContextForgeDataPlaneAppState, JwtTokenDecoders}, + authorization::{AuthorizationClaims, AuthorizationService}, + common::ContextForgeDataPlaneAppState, layers::{ claims_id::claims_layer, - mcp_header_limits::{McpStandardHeaderLimits, mcp_header_limits_layer}, + mcp_header_limits::{StandardHeaderLimits, mcp_header_limits_layer}, }, user_config_store::{ConfigStoreError, UserConfigStore}, }; @@ -115,7 +113,7 @@ mod tests { Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work") } - fn app(limits: McpStandardHeaderLimits) -> Router { + fn app(limits: StandardHeaderLimits) -> Router { Router::new().route("/", get(ok)).layer(middleware::from_fn_with_state(limits, mcp_header_limits_layer)) } @@ -129,7 +127,7 @@ mod tests { #[tokio::test] async fn rejects_too_many_mcp_headers() { - let limits = McpStandardHeaderLimits { count: 2, value_bytes: 1024, total_bytes: 4096 }; + let limits = StandardHeaderLimits { count: 2, value_bytes: 1024, total_bytes: 4096 }; let response = app(limits) .oneshot(request_with_headers(&[ ("Mcp-Method", "tools/call"), @@ -144,7 +142,7 @@ mod tests { #[tokio::test] async fn rejects_oversized_mcp_header_value() { - let limits = McpStandardHeaderLimits { count: 32, value_bytes: 4, total_bytes: 4096 }; + let limits = StandardHeaderLimits { count: 32, value_bytes: 4, total_bytes: 4096 }; let response = app(limits) .oneshot(request_with_headers(&[("Mcp-Param-User", "alice")])) .await @@ -155,7 +153,7 @@ mod tests { #[tokio::test] async fn rejects_excessive_total_mcp_header_bytes() { - let limits = McpStandardHeaderLimits { count: 32, value_bytes: 16, total_bytes: 24 }; + let limits = StandardHeaderLimits { count: 32, value_bytes: 16, total_bytes: 24 }; let response = app(limits) .oneshot(request_with_headers(&[("Mcp-Method", "tools/call"), ("Mcp-Name", "example")])) .await @@ -166,7 +164,7 @@ mod tests { #[tokio::test] async fn counts_mcp_headers_case_insensitively() { - let limits = McpStandardHeaderLimits { count: 1, value_bytes: 1024, total_bytes: 4096 }; + let limits = StandardHeaderLimits { count: 1, value_bytes: 1024, total_bytes: 4096 }; let response = app(limits) .oneshot(request_with_headers(&[("McP-MeThOd", "tools/call"), ("mCp-PaRaM-User", "alice")])) .await @@ -177,7 +175,7 @@ mod tests { #[tokio::test] async fn ignores_non_mcp_headers_for_mcp_specific_budget() { - let limits = McpStandardHeaderLimits { count: 1, value_bytes: 1024, total_bytes: 4096 }; + let limits = StandardHeaderLimits { count: 1, value_bytes: 1024, total_bytes: 4096 }; let response = app(limits) .oneshot(request_with_headers(&[ ("X-One", "1"), @@ -205,11 +203,21 @@ mod tests { } } + #[derive(Debug)] + pub struct Noop; + + #[async_trait] + impl AuthorizationService for Noop { + async fn authorize(&self, _: &HeaderValue) -> Option { + None + } + } + #[tokio::test] async fn rejects_excessive_mcp_headers_before_auth() { - let limits = McpStandardHeaderLimits { count: 1, value_bytes: 1024, total_bytes: 4096 }; + let limits = StandardHeaderLimits { count: 1, value_bytes: 1024, total_bytes: 4096 }; let state = ContextForgeDataPlaneAppState { - jwt_token_decoding_keys: JwtTokenDecoders { rs: None, hmac_sha: None }, + authorization_service: Arc::new(Noop {}), config_store: Arc::new(UnusedConfigStore), config: Config::default(), }; diff --git a/crates/contextforge-data-plane-lib/src/layers/mod.rs b/crates/contextforge-data-plane-lib/src/layers/mod.rs index 97164fef..995a8a8c 100644 --- a/crates/contextforge-data-plane-lib/src/layers/mod.rs +++ b/crates/contextforge-data-plane-lib/src/layers/mod.rs @@ -1,6 +1,9 @@ pub mod claims_id; pub mod mcp_header_limits; pub mod mcp_origin; +pub mod principal_extractor; pub mod user_config_store; pub mod virtual_host_config; pub mod virtual_host_id; + +pub(crate) use principal_extractor::AuthorizedPrincipal; diff --git a/crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs b/crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs new file mode 100644 index 00000000..4a84d627 --- /dev/null +++ b/crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs @@ -0,0 +1,46 @@ +use axum::{extract::Request, middleware::Next, response::Response}; + +use contextforge_data_plane_apis::User; +use tracing::debug; +use typed_builder::TypedBuilder; + +use crate::{AuthorizationClaims, errors::unauthorized_response}; + +#[derive(Debug, Clone, TypedBuilder)] +#[allow(dead_code)] +pub struct AuthorizedPrincipal { + user_id: String, + tenant_id: String, + scopes: Vec, +} + +impl<'a> From<&'a AuthorizedPrincipal> for User<'a> { + fn from(value: &'a AuthorizedPrincipal) -> Self { + Self::new(&value.user_id) + } +} + +impl TryFrom<&AuthorizationClaims> for AuthorizedPrincipal { + type Error = Box; + + fn try_from(value: &AuthorizationClaims) -> Result { + Ok(AuthorizedPrincipal::builder() + .user_id(value.sub.clone()) + .tenant_id(value.tenant_id.clone()) + .scopes(vec![]) + .build()) + } +} + +pub async fn principal_extractor_layer(request: http::Request, next: Next) -> Response { + let maybe_claims = request.extensions().get::(); + let Some(Ok(authorized_principal)) = maybe_claims.map(|claims| { + AuthorizedPrincipal::try_from(claims).inspect_err(|e| debug!("Can't extract the principal {e:?}")) + }) else { + return unauthorized_response("Invalid token. Unable to extract the principal from claims"); + }; + let (mut parts, body) = request.into_parts(); + parts.extensions.insert(authorized_principal); + let request = Request::from_parts(parts, body); + next.run(request).await +} diff --git a/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs b/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs index 4d5804c2..facea51e 100644 --- a/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs +++ b/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs @@ -1,11 +1,11 @@ -use axum::{body::Body, extract::State, middleware::Next, response::Response}; +use axum::{extract::State, middleware::Next, response::Response}; use contextforge_data_plane_apis::User; -use http::{StatusCode, header}; -//use openid::Claims; + use tracing::{debug, info, warn}; use crate::{ - common::{ContextForgeClaims, ContextForgeDataPlaneAppState}, + common::ContextForgeDataPlaneAppState, + errors::{bad_request, internal_server_error}, user_config_store::ConfigStoreError, }; @@ -16,17 +16,17 @@ pub async fn user_config_store_layer( ) -> Response { let method = request.method().clone(); let path = request.uri().path().to_owned(); - let maybe_claims = request.extensions().get::(); - if let Some(claims) = maybe_claims { - let subject = claims.sub.clone(); + let maybe_principal = request.extensions().get::(); + if let Some(principal) = maybe_principal { debug!( - "user_config_store_layer - getting user config for request subject = {subject} method = {method} path = {path}" + "user_config_store_layer - getting user config for principal {principal:?} method = {method} path = {path}" ); - match state.config_store.get_config(&User::new(&subject)).await { + let user = User::from(principal); + match state.config_store.get_config(&user).await { Ok(user_config) => { let virtual_hosts = user_config.virtual_hosts.len(); info!( - "user_config_store_layer - loaded user config subject = {subject} virtual_hosts = {virtual_hosts}" + "user_config_store_layer - loaded user config principal = {principal:?} virtual_hosts = {virtual_hosts}" ); request.extensions_mut().insert(user_config); next.run(request).await @@ -34,32 +34,20 @@ pub async fn user_config_store_layer( Err(ConfigStoreError::NoDataForKey) => { debug!( - "user_config_store_layer - user config lookup returned no data subject = {subject} method = {method} path = {path}" + "user_config_store_layer - user config lookup returned no data principal = {principal:?} method = {method} path = {path}" ); - Response::builder() - .status(StatusCode::BAD_REQUEST) - .header(header::CONTENT_TYPE, "text/plain") - .body(Body::from("Problem occurred retrieving the configuration")) - .expect("Expecting this to work") + bad_request("Problem occurred retrieving the configuration") }, Err(error) => { debug!( - "user_config_store_layer - user config lookup failed subject = {subject} method = {method} path = {path} error = {error}" + "user_config_store_layer - user config lookup failed principal = {principal:?} method = {method} path = {path} error = {error}" ); - Response::builder() - .status(StatusCode::INTERNAL_SERVER_ERROR) - .header(header::CONTENT_TYPE, "text/plain") - .body(Body::from("Problem occurred retrieving the configuration")) - .expect("Expecting this to work") + internal_server_error("Problem occurred retrieving the configuration") }, } } else { warn!("user_config_store_layer - no claims found in request extensions method = {method} path = {path}"); - Response::builder() - .status(StatusCode::BAD_REQUEST) - .header(header::CONTENT_TYPE, "text/plain") - .body(Body::from("No claims in the token")) - .expect("Expecting this to work") + bad_request("No claims in the token") } } diff --git a/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs b/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs index e83752f8..ccd002ae 100644 --- a/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs +++ b/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs @@ -1,7 +1,9 @@ -use axum::{body::Body, middleware::Next, response::Response}; -use http::{StatusCode, header}; +use axum::{middleware::Next, response::Response}; + use tracing::debug; +use crate::errors::bad_request; + #[derive(Clone, Debug, PartialEq, PartialOrd)] pub struct VirtualHostId { value: String, @@ -26,11 +28,7 @@ pub async fn virtual_host_id_layer(mut request: http::Request, next.run(request).await } else { debug!("virtual_host_id_layer - failed to extract virtual host id from request path path = {path}"); - Response::builder() - .status(StatusCode::BAD_REQUEST) - .header(header::CONTENT_TYPE, "text/plain") - .body(Body::from("Problem occured retrieving the configuration")) - .expect("Expecting this to work") + bad_request("Problem occured retrieving the configuration") } } diff --git a/crates/contextforge-data-plane-lib/src/lib.rs b/crates/contextforge-data-plane-lib/src/lib.rs index 96a7508c..48397b18 100644 --- a/crates/contextforge-data-plane-lib/src/lib.rs +++ b/crates/contextforge-data-plane-lib/src/lib.rs @@ -1,17 +1,19 @@ -use std::{fs, sync::Arc}; +use std::sync::Arc; use axum::middleware; use axum_otel_metrics::HttpMetricsLayerBuilder; use contextforge_data_plane_cpex::GatewayPluginRuntimeHandle; use futures::FutureExt; use http::uri::Authority; -use jsonwebtoken::DecodingKey; + use rmcp::transport::{ StreamableHttpServerConfig, streamable_http_server::{session::local::LocalSessionManager, tower::StreamableHttpService}, }; +mod authorization; mod common; mod const_values; +mod errors; mod gateway; mod layers; mod mcp_standard_headers; @@ -32,22 +34,21 @@ use typed_builder::TypedBuilder; pub use user_config_store::RedisUserConfigStore; pub use user_config_store::{ConfigStoreError, UserConfigStore}; -pub use crate::common::{Config, LogRotation, OtlpProtocol}; +pub use crate::common::*; pub type Error = Box; pub type Result = std::result::Result; -use crate::{ - common::{ContextForgeDataPlaneAppState, JwtTokenDecoders}, - layers::{ - claims_id::claims_layer, - mcp_header_limits::{McpStandardHeaderLimits, mcp_header_limits_layer}, - mcp_origin::mcp_origin_layer, - user_config_store::user_config_store_layer, - virtual_host_config::virtual_host_config_layer, - virtual_host_id::virtual_host_id_layer, - }, +use crate::layers::{ + claims_id::claims_layer, + mcp_header_limits::{StandardHeaderLimits, mcp_header_limits_layer}, + mcp_origin::mcp_origin_layer, + principal_extractor::principal_extractor_layer, + user_config_store::user_config_store_layer, + virtual_host_config::virtual_host_config_layer, + virtual_host_id::virtual_host_id_layer, }; +pub use authorization::{AuthorizationClaims, AuthorizationService, get_authorization_service}; #[derive(Clone)] pub enum UserConfigStoreType { @@ -63,6 +64,7 @@ pub struct Gateway { user_config_store_type: UserConfigStoreType, #[builder(default)] plugin_runtime: Option, + authorization_service: Arc, } impl Gateway { @@ -97,7 +99,7 @@ impl Gateway { } async fn build_app(self) -> Result { - let Gateway { config, session_manager, user_config_store_type, plugin_runtime } = self; + let Gateway { config, session_manager, user_config_store_type, plugin_runtime, authorization_service } = self; let user_config_store = match user_config_store_type { UserConfigStoreType::Redis => Arc::new(get_config_store(&config).await?), UserConfigStoreType::Test(store) => store, @@ -130,38 +132,19 @@ impl Gateway { let cors_layer = CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any).expose_headers(Any); - let rs_decoding_key = config.token_verification_public_key.as_ref().map(|path| { - let Ok(key) = - fs::read(path).map_err(|e| format!("Error when creating local decoder {e:?} {}", path.display())) - else { - return Err(format!("Error when creating local decoder. Can't read path {}", path.display())); - }; - - let Ok(key) = DecodingKey::from_rsa_pem(&key) else { - return Err(format!("Error when creating local decoder. Can't read the key {}", path.display())); - }; - Ok(key) - }); - - let mcp_add_state: ContextForgeDataPlaneAppState = ContextForgeDataPlaneAppState { - jwt_token_decoding_keys: JwtTokenDecoders { - rs: rs_decoding_key.transpose()?, - hmac_sha: config - .token_verification_secret - .as_ref() - .map(|token| DecodingKey::from_secret(token.value().as_bytes())), - }, - + let mcp_gateway_state: ContextForgeDataPlaneAppState = ContextForgeDataPlaneAppState { + authorization_service, config_store: Arc::clone(&user_config_store), config: config.clone(), }; - let mcp_standard_header_limits = McpStandardHeaderLimits::from(&config); + let mcp_standard_header_limits = StandardHeaderLimits::from(&config); let app = axum::Router::new() .nest_service("/servers/{virtual_host_name}/mcp", mcp_service) .layer(middleware::from_fn(virtual_host_config_layer)) - .layer(middleware::from_fn_with_state(mcp_add_state.clone(), user_config_store_layer)) - .layer(middleware::from_fn_with_state(mcp_add_state.clone(), claims_layer)) + .layer(middleware::from_fn_with_state(mcp_gateway_state.clone(), user_config_store_layer)) + .layer(middleware::from_fn(principal_extractor_layer)) + .layer(middleware::from_fn_with_state(mcp_gateway_state.clone(), claims_layer)) .layer(middleware::from_fn(virtual_host_id_layer)) // Keep this outside auth/config/RMCP work so oversized MCP headers // are rejected before JWT validation or body parsing. @@ -174,7 +157,7 @@ impl Gateway { #[cfg(feature = "with_tools")] let app = tools::add_tools(app); - let app = app.with_state(mcp_add_state); + let app = app.with_state(mcp_gateway_state); let app = axum::Router::new() .nest("/contextforge-rs", app) .layer(TraceLayer::new_for_http().make_span_with(telemetry::ExtractingMakeSpan)) @@ -202,7 +185,7 @@ mod tests { use tower::ServiceExt; use crate::{ - Config, Gateway, UserConfigStoreType, + Config, Gateway, UserConfigStoreType, get_authorization_service, user_config_store::{ConfigStoreError, UserConfigStore}, }; @@ -224,6 +207,7 @@ mod tests { async fn production_router_rejects_excessive_mcp_headers_before_auth() { let config = Config { mcp_standard_header_max_count: 1, ..Config::default() }; let app = Gateway::builder() + .with_authorization_service(get_authorization_service(&config).expect("this should not fail")) .with_config(config) .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(UnusedConfigStore))) diff --git a/crates/contextforge-data-plane-lib/src/tools.rs b/crates/contextforge-data-plane-lib/src/tools.rs index 87ab7f41..13741b69 100644 --- a/crates/contextforge-data-plane-lib/src/tools.rs +++ b/crates/contextforge-data-plane-lib/src/tools.rs @@ -1,5 +1,3 @@ -use std::fs; - use axum::{ Json, body::Body, @@ -7,61 +5,50 @@ use axum::{ response::{IntoResponse, Response}, routing::{Router, get, post}, }; -use chrono::Duration; use contextforge_data_plane_apis::{User as CFUser, user_store::UserConfig}; -use http::{StatusCode, header}; +use http::{ + StatusCode, + header::{self, CACHE_CONTROL}, +}; +use jsonwebtoken::jwk::{Jwk, JwkSet}; use serde::Deserialize; -use uuid::Uuid; +use std::fs; -//use tracing::debug; -use crate::{ - common::{ContextForgeClaims, ContextForgeDataPlaneAppState, Scopes, User}, - const_values::{CONTEXT_FORGE_GATEWAY_AUDIENCE, CONTEXT_FORGE_GATEWAY_ISSUER}, -}; +use crate::{authorization::AuthorizationClaims, common::ContextForgeDataPlaneAppState}; const DEFAULT_TOKEN_EMAIL: &str = "admin@example.com"; +const JWKS_CACHE_CONTROL: &str = "public, max-age=300, must-revalidate"; +const TOKEN_PATH: &str = "/admin/tokens/{tenant_id}/{user_id}"; +const JWKS_PATH: &str = "/admin/.well-known/jwks.json"; +const CONFIGURE_USER_PATH: &str = "/admin/userconfigs/{user_id}"; #[derive(Debug, Deserialize)] pub struct TokenQuery { email: Option, } -impl ContextForgeClaims { - pub fn new(user_id: &str, user_email: &str) -> Self { - let audience = CONTEXT_FORGE_GATEWAY_AUDIENCE.to_owned(); - let start = std::time::SystemTime::now(); - let now = start.duration_since(std::time::UNIX_EPOCH).expect("Time went backwards").as_secs(); - Self { - iss: CONTEXT_FORGE_GATEWAY_ISSUER.to_owned(), - sub: user_id.to_owned(), - aud: audience, - exp: now + Duration::hours(1).num_seconds().cast_unsigned(), - iat: Some(now), - jti: Uuid::new_v4().to_string(), - token_use: Some("api".to_owned()), - teams: Some(vec!["team_awesome".to_owned()]), - user: User::builder() - .email(user_email.to_owned()) - .auth_provider("api_token".to_owned()) - .full_name(Some("API Token User".to_owned())) - .is_admin(true) - .build(), - scopes: Some( - Scopes::builder() - .server_id(Some("my_id".to_owned())) - .ip_restrictions(vec!["192.169.1.0/24".to_owned()]) - .permissions(vec!["tools.read".to_owned(), "servers.use".to_owned()]) - .time_restrictions(None) - .build(), - ), - } - } +async fn get_jwks(State(state): State) -> Response { + let Ok(key) = jsonwebtoken::EncodingKey::from_rsa_pem( + &fs::read(&state.config.token_verification_private_key).expect("Expecting this to work"), + ) else { + return (StatusCode::INTERNAL_SERVER_ERROR, "Can't find the encoding key or the format is wrong") + .into_response(); + }; + + let Ok(key) = Jwk::from_encoding_key(&key, jsonwebtoken::Algorithm::RS256) else { + return (StatusCode::INTERNAL_SERVER_ERROR, "Can't find the encoding key or the format is wrong") + .into_response(); + }; + + let keys = vec![key]; + (StatusCode::OK, [(CACHE_CONTROL, JWKS_CACHE_CONTROL)], Json(JwkSet { keys })).into_response() } pub fn add_tools(router: Router) -> Router { router - .route("/admin/tokens/{user_id}", get(get_token)) - .route("/admin/userconfigs/{user_id}", post(configure_user)) + .route(TOKEN_PATH, get(get_token)) + .route(JWKS_PATH, get(get_jwks)) + .route(CONFIGURE_USER_PATH, post(configure_user)) .route("/health", get(health)) } @@ -75,7 +62,7 @@ pub async fn health() -> Response { pub async fn get_token( State(state): State, - Path(user_id): Path, + Path((tenant_id, user_id)): Path<(String, String)>, Query(query): Query, ) -> Response { let key = jsonwebtoken::EncodingKey::from_rsa_pem( @@ -84,10 +71,11 @@ pub async fn get_token( .expect("Expecting this to work"); let user_email = query.email.as_deref().unwrap_or(DEFAULT_TOKEN_EMAIL); - let claims = ContextForgeClaims::new(&user_id, user_email); + let mut claims = AuthorizationClaims::new(&user_id, user_email); + claims.tenant_id = tenant_id; let mut header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::RS256); header.kid = Some("test".to_owned()); - let token = jsonwebtoken::encode::(&header, &claims, &key).expect("Expecting this to work"); + let token = jsonwebtoken::encode::(&header, &claims, &key).expect("Expecting this to work"); token.into_response() } @@ -112,20 +100,3 @@ pub async fn configure_user( .expect("Expecting this to work") } } - -#[cfg(test)] -mod tests { - use serde_json::Value; - - use super::ContextForgeClaims; - - #[test] - fn new_claims_keeps_subject_and_email_metadata_separate() { - let claims = ContextForgeClaims::new("11111111-1111-1111-1111-111111111111", "admin@example.com"); - - let payload = serde_json::to_value(claims).expect("claims should serialize"); - - assert_eq!(payload["sub"], Value::String("11111111-1111-1111-1111-111111111111".to_owned())); - assert_eq!(payload["user"]["email"], Value::String("admin@example.com".to_owned())); - } -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_call_tools.rs b/crates/contextforge-data-plane-lib/tests/gateway_call_tools.rs index cf543ab5..3f8ba965 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_call_tools.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_call_tools.rs @@ -6,7 +6,10 @@ use tracing::{info, warn}; use support::{ListToolsGatewaySettings, TEST_USER_ID, create_client, create_gateway_with_four_counters, create_ports}; -use crate::support::{connect_client_with_protocol, connect_modern_client, create_gateway_with_four_legacy_counters}; +use crate::support::{ + connect_client_with_protocol, connect_modern_client, create_default_config, + create_gateway_with_four_legacy_counters, +}; #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] @@ -15,9 +18,8 @@ async fn plaintext_call_prefixed_backend_tools_modern_modern() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -53,9 +55,8 @@ async fn plaintext_call_prefixed_backend_tools_modern_legacy() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -91,9 +92,8 @@ async fn plaintext_call_prefixed_backend_tools_legacy_modern() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -172,9 +172,8 @@ async fn plaintext_call_invalid_backend_tools() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs b/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs index 89c7f96b..97c9adcd 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs @@ -12,6 +12,8 @@ use support::{ create_ports, create_tls_client, create_tls_gateway_with_four_tls_counters, }; +use crate::support::create_default_config; + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] #[ignore = "Fan out list tools is not supported at the moment. This should be enabled in 2.x"] @@ -20,9 +22,8 @@ async fn plaintext_lists_prefixed_backend_tools() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -58,13 +59,12 @@ async fn tls_lists_prefixed_backend_tools() -> Result<()> { format!("127.0.0.1:{gateway_port}").parse().expect("This should work"); let config = Config { - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), tls_address: Some(server_socket_addr), server_private_key: Some("../../assets/contextforgeCA/contextforge-server.key.pem".into()), server_certificate: Some("../../assets/contextforgeCA/contextforge-server.cert.pem".into()), upstream_trust_bundle: Some("../../assets/contextforgeCA/contextforge.intermediate.ca-chain.cert.pem".into()), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs index 246ce100..fa06f477 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs @@ -66,6 +66,9 @@ async fn start_gateway(config: Config, virtual_host_id: &str, user_config: UserC .with_config(config) .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(store))) + .with_authorization_service(Arc::new(support::auth::AlwaysAllowAuthorizatioService::new( + TEST_USER_ID.to_owned(), + ))) .build(); tokio::spawn(async move { diff --git a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs b/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs index 70921aca..213df7e8 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs @@ -10,7 +10,7 @@ use support::{ create_ports, }; -use crate::support::connect_modern_client; +use crate::support::{connect_modern_client, create_default_config}; #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] @@ -20,9 +20,8 @@ async fn plaintext_lists_prefixed_backend_prompts() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -46,9 +45,8 @@ async fn plaintext_gets_prompt_from_prefixed_backend_name() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_resource_read.rs b/crates/contextforge-data-plane-lib/tests/gateway_resource_read.rs index d6648b1d..b02b8515 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_resource_read.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_resource_read.rs @@ -6,7 +6,10 @@ use tracing::{info, warn}; use support::{ListToolsGatewaySettings, TEST_USER_ID, create_client, create_gateway_with_four_counters, create_ports}; -use crate::support::{connect_client_with_protocol, connect_modern_client, create_gateway_with_four_legacy_counters}; +use crate::support::{ + connect_client_with_protocol, connect_modern_client, create_default_config, + create_gateway_with_four_legacy_counters, +}; #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] @@ -15,9 +18,8 @@ async fn plaintext_call_prefixed_read_resources_modern_modern() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -59,9 +61,8 @@ async fn plaintext_call_prefixed_read_resources_modern_legacy() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -103,9 +104,8 @@ async fn plaintext_call_prefixed_read_resources_legacy_modern() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -196,9 +196,8 @@ async fn plaintext_call_invalid_backend_tools() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs b/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs index c289efbd..d99ceec4 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs @@ -9,6 +9,8 @@ use support::{ create_ports, }; +use crate::support::create_default_config; + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] #[ignore = "Fan out list resources is not supported at the moment. This should be enabled in 2.x"] @@ -17,9 +19,8 @@ async fn plaintext_lists_prefixed_backend_resource_templates() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; @@ -55,9 +56,8 @@ async fn plaintext_reads_resource_from_prefixed_template() -> Result<()> { let config = Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() }; let user = TEST_USER_ID; diff --git a/crates/contextforge-data-plane-lib/tests/support/auth.rs b/crates/contextforge-data-plane-lib/tests/support/auth.rs index 9311f0a3..a73fdc7d 100644 --- a/crates/contextforge-data-plane-lib/tests/support/auth.rs +++ b/crates/contextforge-data-plane-lib/tests/support/auth.rs @@ -3,6 +3,9 @@ use std::{ time::{SystemTime, UNIX_EPOCH}, }; +use async_trait::async_trait; +use contextforge_data_plane_lib::{AuthorizationClaims, AuthorizationService}; +use http::HeaderValue; use jsonwebtoken::{Algorithm, EncodingKey, Header, encode}; use serde_json::json; @@ -39,3 +42,22 @@ pub(crate) fn token(user_id: &str) -> String { }); encode(&header, &claims, &key).expect("jwt token") } + +#[derive(Debug)] +pub struct AlwaysAllowAuthorizatioService { + user: String, +} + +impl AlwaysAllowAuthorizatioService { + pub fn new(user: String) -> AlwaysAllowAuthorizatioService { + Self { user } + } +} +#[async_trait] +impl AuthorizationService for AlwaysAllowAuthorizatioService { + async fn authorize(&self, _: &HeaderValue) -> Option { + let mut claims = AuthorizationClaims::default(); + claims.sub.clone_from(&self.user); + Some(claims) + } +} diff --git a/crates/contextforge-data-plane-lib/tests/support/mod.rs b/crates/contextforge-data-plane-lib/tests/support/mod.rs index e40c84e6..e04e427c 100644 --- a/crates/contextforge-data-plane-lib/tests/support/mod.rs +++ b/crates/contextforge-data-plane-lib/tests/support/mod.rs @@ -1,6 +1,6 @@ #![allow(dead_code, unused_imports, reason = "shared CPEX test fixture is used by separate integration test targets")] -mod auth; +pub mod auth; mod client; pub(crate) mod mock_counter; pub(crate) mod paginating_mock; @@ -14,11 +14,15 @@ mod user_config_store; pub(crate) const TEST_USER_ID: &str = "11111111-1111-1111-1111-111111111111"; pub(crate) const TEST_USER_EMAIL: &str = "admin@example.com"; +#[cfg(feature = "with_tools")] +use std::{path::PathBuf, str::FromStr}; + pub(crate) use auth::token; pub(crate) use client::{ CLIENT_CONNECT_TIMEOUT, TEST_POLL_INTERVAL, connect_client, connect_client_with_handler, connect_client_with_protocol, connect_modern_client, create_client, create_tls_client, modern_client_info, }; +use contextforge_data_plane_lib::{Config, RedisConnectionMode}; pub(crate) use plugin::{ POST_DENY_ERROR_CODE, PRE_DENY_ERROR_CODE, PROMPT_ERROR_MESSAGE, PROMPT_POST_DENY_ERROR_CODE, PromptBehavior, PromptTestPlugin, REWRITTEN_PROMPT_RESOURCE, REWRITTEN_PROMPT_TEXT, REWRITTEN_PROMPT_TOPIC, REWRITTEN_SUM_A, @@ -35,3 +39,45 @@ pub(crate) use test_gateways::{ }; pub(crate) use tool::{error_code, error_parts, sum_request, text}; pub(crate) use user_config_store::MemoryUserConfigStore; + +pub fn create_default_config() -> Config { + Config { + address: None, + jwks_url: "http://127.0.0.1:8080/".parse().expect("should work"), + jwks_ca_cert_path: None, + enable_open_telemetry: None, + otlp_endpoint: None, + otlp_protocol: None, + otlp_headers: None, + otlp_service_name: None, + enable_otel_metrics: None, + otlp_metrics_endpoint: None, + mcp_standard_header_max_count: 10, + mcp_standard_header_max_value_bytes: 4096, + mcp_standard_header_max_total_bytes: 4096, + number_of_cpus: None, + single_runtime: None, + runtime_plugins_enabled: None, + tls_address: None, + server_private_key: None, + server_certificate: None, + upstream_connection_mode: None, + upstream_private_key: None, + upstream_certificate: None, + upstream_trust_bundle: None, + user_config_cache_expiry_seconds: 10, + redis_address: String::new(), + redis_port: 0, + redis_mode: RedisConnectionMode::PlainText, + redis_tls_trust_bundle: None, + redis_tls_client_private_key: None, + redis_tls_client_certificate: None, + log_name: None, + log_rotation: None, + mcp_allowed_origins: None, + mcp_allowed_hosts: None, + + #[cfg(feature = "with_tools")] + token_verification_private_key: PathBuf::from_str("./assets/jwt.key").expect("This should work"), + } +} diff --git a/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs b/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs index 14677d97..4442f46f 100644 --- a/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs +++ b/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs @@ -29,7 +29,7 @@ use rmcp::{ use serde_json::{Map, Value, json}; use tokio::sync::Mutex as TokioMutex; -use crate::support::test_gateways::construct_services; +use crate::support::{self, create_default_config, test_gateways::construct_services}; use super::{MemoryUserConfigStore, token}; @@ -417,14 +417,14 @@ async fn start_gateway_with_state( let gateway = Gateway::builder() .with_config(Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), runtime_plugins_enabled: Some(runtime_plugins_enabled), - ..Default::default() + ..create_default_config() }) .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(user_store))) .with_plugin_runtime(runtime_plugins_enabled.then(|| plugin_runtime.handle())) + .with_authorization_service(Arc::new(support::auth::AlwaysAllowAuthorizatioService::new(user.to_owned()))) .build(); let gateway = async move { gateway.run_gateway().await }.boxed(); diff --git a/crates/contextforge-data-plane-lib/tests/support/test_gateways.rs b/crates/contextforge-data-plane-lib/tests/support/test_gateways.rs index 64fabaa2..b074bf83 100644 --- a/crates/contextforge-data-plane-lib/tests/support/test_gateways.rs +++ b/crates/contextforge-data-plane-lib/tests/support/test_gateways.rs @@ -1,19 +1,24 @@ use std::{collections::HashMap, sync::Arc}; +use async_trait::async_trait; use contextforge_data_plane_apis::{ User, user_store::{BackendMCPGateway, ServiceRoute, UserConfig, VirtualHost}, }; use contextforge_data_plane_lib::{ - Config, Gateway, Result, UpstreamConnectionMode, UserConfigStore, UserConfigStoreType, + AuthorizationClaims, AuthorizationService, Config, Gateway, Result, UpstreamConnectionMode, UserConfigStore, + UserConfigStoreType, }; use futures::{FutureExt, future::BoxFuture}; +use http::HeaderValue; use rmcp::transport::{ StreamableHttpServerConfig, StreamableHttpService, streamable_http_server::session::local::LocalSessionManager, }; use rustls::ProtocolVersion; use tracing::warn; +use crate::support::{self, create_default_config}; + use super::{MemoryUserConfigStore, mock_counter}; const MOCK_COUNTER_TOOL_NAMES: &[&str] = @@ -37,9 +42,8 @@ pub(crate) struct ListToolsGatewaySettings { pub(crate) fn plaintext_config(gateway_port: u16) -> Config { Config { address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - token_verification_public_key: Some("../../assets/jwt.key.pub".into()), upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..Default::default() + ..create_default_config() } } @@ -156,6 +160,7 @@ async fn create_gateway_with_four_counters_and_custom_config( .with_config(config.clone()) .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(mocked_user_config_store))) + .with_authorization_service(Arc::new(support::auth::AlwaysAllowAuthorizatioService::new(user.to_owned()))) .build(); let gateway = async move { diff --git a/crates/contextforge-data-plane/Cargo.toml b/crates/contextforge-data-plane/Cargo.toml index 6de6816e..e8096061 100644 --- a/crates/contextforge-data-plane/Cargo.toml +++ b/crates/contextforge-data-plane/Cargo.toml @@ -37,6 +37,7 @@ rustls.workspace = true [features] plugins = ["dep:cpex-secrets-detection"] test-plugins = ["dep:cpex-payload-marker", "dep:cpex-text-prefixer", "dep:cpex-tool-namespace"] +with_tools = ["contextforge-data-plane-lib/with_tools"] [dev-dependencies] axum.workspace = true diff --git a/crates/contextforge-data-plane/src/logging.rs b/crates/contextforge-data-plane/src/logging.rs index a695f7f8..08d5b8e3 100644 --- a/crates/contextforge-data-plane/src/logging.rs +++ b/crates/contextforge-data-plane/src/logging.rs @@ -33,6 +33,8 @@ const DEFAULT_HTTP_TRACES_ENDPOINT: &str = "http://127.0.0.1:4318/v1/traces"; const DEFAULT_HTTP_METRICS_ENDPOINT: &str = "http://127.0.0.1:4318/v1/metrics"; const METRICS_EXPORT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30); +const DEFAULT_LOGGING: &str = "debug,hyper_util=OFF,tower_http=OFF,rmcp=warn,reqwest=warn,rustls=WARN"; + pub fn init_tracing_logging(configuration: &Config) -> Result> { let registry = Registry::default(); @@ -46,12 +48,14 @@ pub fn init_tracing_logging(configuration: &Config) -> Result Result<(), Box> { }; let plugin_runtime = plugin_registry.as_ref().map(|runtime| runtime.handle()); + let authorization_service = get_authorization_service(&config)?; + let gateway = Gateway::builder() .with_config(config) .with_user_config_store_type(UserConfigStoreType::Redis) .with_session_manager(Arc::new(LocalSessionManager::default())) .with_plugin_runtime(plugin_runtime.clone()) + .with_authorization_service(authorization_service) .build(); runtime.execute(gateway, plugin_registry)