diff --git a/Cargo.lock b/Cargo.lock index 24c6f8e9..d422051e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -35,7 +35,7 @@ checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", "cipher", - "cpufeatures", + "cpufeatures 0.2.17", ] [[package]] @@ -466,7 +466,7 @@ checksum = "3c3610892ee6e0cbce8ae2700349fcf8f98adb0dbfbee85aec3c9179d29cc072" dependencies = [ "base64ct", "blake2", - "cpufeatures", + "cpufeatures 0.2.17", "password-hash", ] @@ -1113,7 +1113,7 @@ dependencies = [ "cc", "cfg-if", "constant_time_eq", - "cpufeatures", + "cpufeatures 0.2.17", ] [[package]] @@ -1565,7 +1565,18 @@ checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" dependencies = [ "cfg-if", "cipher", - "cpufeatures", + "cpufeatures 0.2.17", +] + +[[package]] +name = "chacha20" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", ] [[package]] @@ -1575,7 +1586,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" dependencies = [ "aead", - "chacha20", + "chacha20 0.9.1", "cipher", "poly1305", "zeroize", @@ -1871,6 +1882,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "cranelift-assembler-x64" version = "0.128.4" @@ -2144,7 +2164,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "curve25519-dalek-derive", "digest 0.10.7", "fiat-crypto", @@ -3559,6 +3579,7 @@ dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", + "rand_core 0.10.1", "wasip2", "wasip3", ] @@ -4408,7 +4429,7 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb26cec98cce3a3d96cbb7bced3c4b16e3d13f27ec56dbd62cbc8f39cfb9d653" dependencies = [ - "cpufeatures", + "cpufeatures 0.2.17", ] [[package]] @@ -5574,7 +5595,7 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8159bd90725d2df49889a078b54f4f79e87f1f8a8444194cdca81d38f5393abf" dependencies = [ - "cpufeatures", + "cpufeatures 0.2.17", "opaque-debug", "universal-hash", ] @@ -5586,7 +5607,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "opaque-debug", "universal-hash", ] @@ -5878,6 +5899,17 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rand" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" +dependencies = [ + "chacha20 0.10.0", + "getrandom 0.4.2", + "rand_core 0.10.1", +] + [[package]] name = "rand_chacha" version = "0.3.1" @@ -5916,6 +5948,12 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "rand_distr" version = "0.4.3" @@ -6303,19 +6341,27 @@ checksum = "0810a9f717d9828f475fe1f629f4c305c8464b7f496c3a854b58d29e65f4058e" dependencies = [ "async-trait", "base64", + "bytes", "chrono", "futures", + "http", + "http-body", + "http-body-util", "pastey", "pin-project-lite", + "rand 0.10.1", "rmcp-macros", "schemars", "serde", "serde_json", + "sse-stream", "thiserror 2.0.18", "tokio", "tokio-stream", "tokio-util", + "tower-service", "tracing", + "uuid", ] [[package]] @@ -6856,7 +6902,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest 0.10.7", ] @@ -6867,7 +6913,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest 0.10.7", ] @@ -6878,7 +6924,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c5f3b1e2dc8aad28310d8410bd4d7e180eca65fca176c52ab00d364475d0024" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest 0.11.1", ] @@ -7106,6 +7152,19 @@ dependencies = [ "der", ] +[[package]] +name = "sse-stream" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3962b63f038885f15bce2c6e02c0e7925c072f1ac86bb60fd44c5c6b762fb72" +dependencies = [ + "bytes", + "futures-util", + "http-body", + "http-body-util", + "pin-project-lite", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" diff --git a/README.md b/README.md index 87ca4ac5..271eb2cc 100644 --- a/README.md +++ b/README.md @@ -433,6 +433,46 @@ Replace `--db ` with `--memory` for an ephemeral, in-memory graph. > stdout is reserved for the JSON-RPC stream; logs are written to stderr. +### Remote (HTTP) connector + +Build with the HTTP transport and run cortex normally; the MCP endpoint is served at `/mcp`: + +```bash +cargo build -p aingle_cortex --features "mcp dag mcp-http" --release + +AINGLE_MCP_HTTP_TOKEN=your-secret AINGLE_PUBLIC_HOST=your.domain \ + aingle-cortex --db ./data/graph.sled +# MCP available at http://localhost:19090/mcp +# Clients send: Authorization: Bearer your-secret +``` + +- The `/mcp` route is **only mounted** when a bearer token is set (`--mcp-http-token` / `AINGLE_MCP_HTTP_TOKEN`) or `--mcp-http-allow-anonymous` is passed — it is never exposed unintentionally. +- `AINGLE_PUBLIC_HOST` (comma-separated) must list the public hostname(s) for a remote deployment (rmcp rejects non-loopback `Host` headers otherwise). +- `--mcp-http-allow-anonymous` serves `/mcp` without auth (test only). + +> Note: claude.ai's connector UI cannot attach a static bearer header; secured remote use from claude.ai needs OAuth (planned). Verify the deployed endpoint with `curl`/MCP Inspector using the bearer token. + +#### OAuth (secured remote access) + +Build with `--features "mcp dag mcp-http mcp-oauth"` and set an issuer; cortex then acts as an OAuth 2.0 +Resource Server for `/mcp` (e.g. for claude.ai remote connectors): + +```bash +AINGLE_OAUTH_ISSUER=https://auth.example/realms/aingle \ +AINGLE_OAUTH_RESOURCE=https://mcp.example/mcp \ + aingle-cortex --db ./data/graph.sled +``` + +- Serves `GET /.well-known/oauth-protected-resource` (RFC 9728); a request to `/mcp` without a valid token + gets `401` + `WWW-Authenticate: Bearer resource_metadata="…"` so clients can discover the authorization server. +- `/mcp` accepts a Bearer **JWT** signed by the issuer — validated via its JWKS, algorithm pinned to RS256, + with `iss`, `aud` (must equal the resource), and `exp` all required. +- The Phase-1 static bearer (`AINGLE_MCP_HTTP_TOKEN`) is still accepted alongside OAuth (handy for `curl`). + This dual-credential behavior is intentional; a leaked static token bypasses the JWT checks, so use it only + where appropriate. +- For non-Keycloak issuers, set `AINGLE_OAUTH_JWKS_URL` explicitly (the default derives the Keycloak certs path). + The Authorization Server (login, PKCE, client registration) is external — see the private deploy repo. + --- ## Contributing diff --git a/crates/aingle_cortex/Cargo.toml b/crates/aingle_cortex/Cargo.toml index f22e1d50..7ceecf18 100644 --- a/crates/aingle_cortex/Cargo.toml +++ b/crates/aingle_cortex/Cargo.toml @@ -23,7 +23,9 @@ p2p-mdns = ["p2p", "dep:mdns-sd", "dep:if-addrs"] cluster = ["p2p", "dep:aingle_wal", "dep:aingle_raft", "dep:openraft", "dep:tokio-rustls", "dep:rustls-pemfile"] dag = ["cluster", "aingle_graph/dag", "aingle_graph/dag-sign", "aingle_raft/dag"] mcp = ["dep:rmcp", "dep:schemars"] -full = ["rest", "graphql", "sparql", "auth", "dag"] +mcp-http = ["mcp", "rmcp/transport-streamable-http-server", "rmcp/server-side-http"] +mcp-oauth = ["mcp-http", "dep:jsonwebtoken"] +full =["rest", "graphql", "sparql", "auth", "dag"] [[bin]] name = "aingle-cortex" @@ -119,3 +121,6 @@ tokio-test = "0.4" # Enable the rmcp `client` feature for the in-process MCP integration test. # This is dev-only; the production `mcp` feature uses the server side only. rmcp = { version = "1.7", features = ["client", "transport-io"] } +# OAuth integration test: derive a JWK from the test public key + base64url encode. +rsa = { version = "0.9", features = ["pem"] } +base64 = "0.22" diff --git a/crates/aingle_cortex/src/main.rs b/crates/aingle_cortex/src/main.rs index fb7b71d1..d37b9439 100644 --- a/crates/aingle_cortex/src/main.rs +++ b/crates/aingle_cortex/src/main.rs @@ -74,6 +74,33 @@ async fn main() -> Result<(), Box> { "--mcp" => { config.mcp_mode = true; } + "--mcp-http-token" => { + if i + 1 < args.len() { + config.mcp_http_token = Some(args[i + 1].clone()); + i += 1; + } + } + "--mcp-http-allow-anonymous" => { + config.mcp_http_allow_anonymous = true; + } + "--mcp-oauth-issuer" => { + if i + 1 < args.len() { + config.mcp_oauth_issuer = Some(args[i + 1].clone()); + i += 1; + } + } + "--mcp-oauth-resource" => { + if i + 1 < args.len() { + config.mcp_oauth_resource = Some(args[i + 1].clone()); + i += 1; + } + } + "--mcp-oauth-jwks-url" => { + if i + 1 < args.len() { + config.mcp_oauth_jwks_url = Some(args[i + 1].clone()); + i += 1; + } + } "--flush-interval" => { if i + 1 < args.len() { config.flush_interval_secs = args[i + 1].parse().unwrap_or(300); @@ -89,6 +116,20 @@ async fn main() -> Result<(), Box> { i += 1; } + // Fall back to the environment for the MCP HTTP bearer token if not given as a flag. + if config.mcp_http_token.is_none() { + config.mcp_http_token = std::env::var("AINGLE_MCP_HTTP_TOKEN").ok(); + } + if config.mcp_oauth_issuer.is_none() { + config.mcp_oauth_issuer = std::env::var("AINGLE_OAUTH_ISSUER").ok(); + } + if config.mcp_oauth_resource.is_none() { + config.mcp_oauth_resource = std::env::var("AINGLE_OAUTH_RESOURCE").ok(); + } + if config.mcp_oauth_jwks_url.is_none() { + config.mcp_oauth_jwks_url = std::env::var("AINGLE_OAUTH_JWKS_URL").ok(); + } + // If --mcp was requested but the binary was built without the `mcp` feature, // fail loudly instead of silently falling through to the TCP REST server. #[cfg(not(feature = "mcp"))] @@ -299,6 +340,21 @@ fn print_help() { println!(" --memory Use volatile in-memory storage (no persistence)"); println!(" --flush-interval Periodic flush interval in seconds (default: 300, 0=off)"); println!(" --mcp Serve MCP over stdio (requires --features mcp)"); + println!( + " --mcp-http-token Bearer token for the /mcp HTTP endpoint (requires --features mcp-http)" + ); + println!( + " --mcp-http-allow-anonymous Serve /mcp without auth (test mode; requires --features mcp-http)" + ); + println!( + " --mcp-oauth-issuer OAuth issuer URL; enables OAuth on /mcp (requires --features mcp-oauth)" + ); + println!( + " --mcp-oauth-resource OAuth protected-resource id = expected JWT audience (requires --features mcp-oauth)" + ); + println!( + " --mcp-oauth-jwks-url Explicit JWKS URL; derived from issuer if omitted (requires --features mcp-oauth)" + ); println!(" -V, --version Print version and exit"); println!(" --help Print this help message"); println!(); diff --git a/crates/aingle_cortex/src/mcp/http.rs b/crates/aingle_cortex/src/mcp/http.rs new file mode 100644 index 00000000..fd4a2926 --- /dev/null +++ b/crates/aingle_cortex/src/mcp/http.rs @@ -0,0 +1,172 @@ +// Copyright 2019-2026 Apilium Technologies OÜ. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR Commercial + +//! Streamable HTTP transport for the MCP server, mounted at `/mcp`. + +/// Constant-time comparison of the presented bearer token against the expected one. +/// Returns true only when the `Authorization` header is exactly `Bearer `. +pub(crate) fn bearer_ok(expected: &str, header: Option<&str>) -> bool { + let presented = match header.and_then(|h| h.strip_prefix("Bearer ")) { + Some(t) => t, + None => return false, + }; + let a = expected.as_bytes(); + let b = presented.as_bytes(); + if a.len() != b.len() { + return false; + } + let mut diff = 0u8; + for (x, y) in a.iter().zip(b.iter()) { + diff |= x ^ y; + } + diff == 0 +} + +use crate::mcp::AingleMcp; +use crate::state::AppState; +use axum::response::IntoResponse; +use axum::Router; + +/// Given an OAuth resource URL like `https://mcp.example/mcp`, return its +/// `:///.well-known/oauth-protected-resource`. +#[cfg(feature = "mcp-oauth")] +fn metadata_url_from_resource(resource: &str) -> Option { + // resource is ":///"; take scheme + authority. + let rest = resource.split("://").nth(1)?; // "/" + let authority = rest.split('/').next()?; // "" + let scheme = resource.split("://").next()?; // "https" / "http" + if authority.is_empty() { + return None; + } + Some(format!( + "{scheme}://{authority}/.well-known/oauth-protected-resource" + )) +} + +/// Build the `/mcp` sub-router. Returns None when neither a token nor anonymous +/// mode is configured (so the endpoint is never exposed unintentionally). +pub fn mcp_http_router( + state: AppState, + token: Option, + allow_anonymous: bool, + public_hosts: Vec, + #[cfg(feature = "mcp-oauth")] oauth: Option<( + crate::mcp::oauth::OAuthConfig, + crate::mcp::oauth::JwksCache, + )>, +) -> Option { + use rmcp::transport::streamable_http_server::{ + session::local::LocalSessionManager, StreamableHttpServerConfig, StreamableHttpService, + }; + use std::sync::Arc; + + #[cfg(feature = "mcp-oauth")] + let oauth_enabled = oauth.is_some(); + #[cfg(not(feature = "mcp-oauth"))] + let oauth_enabled = false; + if token.is_none() && !allow_anonymous && !oauth_enabled { + return None; + } + + let factory_state = state; + // Default loopback hosts plus any public host(s) for remote deployment. + // `::1` is included by StreamableHttpServerConfig::default(), but we rebuild + // the list explicitly so we control exactly which hosts are accepted. + let mut allowed_hosts = vec![ + "localhost".to_string(), + "127.0.0.1".to_string(), + "::1".to_string(), + ]; + allowed_hosts.extend(public_hosts.clone()); + + // StreamableHttpServerConfig is #[non_exhaustive]; build via Default + builder. + let config = StreamableHttpServerConfig::default().with_allowed_hosts(allowed_hosts); + let service = StreamableHttpService::new( + move || Ok(AingleMcp::new(factory_state.clone())), + Arc::new(LocalSessionManager::default()), + config, + ); + + // The StreamableHttpService is a tower Service over `http::Request`; + // axum's body satisfies the `Body` bound, so it can be mounted directly as + // the router's fallback service. Nested under `/mcp` in build_router, the + // full path served is `/mcp`. + let mut router = Router::new().fallback_service(service); + + if !allow_anonymous { + let static_token = token.clone(); + #[cfg(feature = "mcp-oauth")] + let oauth_for_layer = oauth.clone(); + let resource_metadata_url = { + #[cfg(feature = "mcp-oauth")] + let from_oauth = oauth + .as_ref() + .and_then(|(cfg, _)| metadata_url_from_resource(&cfg.resource)); + #[cfg(not(feature = "mcp-oauth"))] + let from_oauth: Option = None; + from_oauth + .or_else(|| { + public_hosts + .first() + .map(|h| format!("https://{h}/.well-known/oauth-protected-resource")) + }) + .unwrap_or_default() + }; + router = router.layer(axum::middleware::from_fn( + move |req: axum::extract::Request, next: axum::middleware::Next| { + let static_token = static_token.clone(); + #[cfg(feature = "mcp-oauth")] + let oauth_for_layer = oauth_for_layer.clone(); + let rmu = resource_metadata_url.clone(); + async move { + let hdr = req + .headers() + .get(axum::http::header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()); + if let Some(ref t) = static_token { + if bearer_ok(t, hdr) { + return next.run(req).await; + } + } + #[cfg(feature = "mcp-oauth")] + if let Some((ref cfg, ref jwks)) = oauth_for_layer { + if let Some(raw) = hdr.and_then(|h| h.strip_prefix("Bearer ")) { + if let Some(kid) = crate::mcp::oauth::token_kid(raw) { + if let Some(key) = jwks.key_for(&kid).await { + if crate::mcp::oauth::validate_jwt(raw, &key, cfg).is_ok() { + return next.run(req).await; + } + } + } + } + } + let www = if rmu.is_empty() { + "Bearer".to_string() + } else { + format!("Bearer resource_metadata=\"{rmu}\", scope=\"mcp\"") + }; + ( + axum::http::StatusCode::UNAUTHORIZED, + [(axum::http::header::WWW_AUTHENTICATE, www)], + "unauthorized", + ) + .into_response() + } + }, + )); + } + Some(router) +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn bearer_check() { + assert!(bearer_ok("secret", Some("Bearer secret"))); + assert!(!bearer_ok("secret", Some("Bearer wrong"))); + assert!(!bearer_ok("secret", Some("secret"))); // missing prefix + assert!(!bearer_ok("secret", None)); // missing header + assert!(!bearer_ok("secret", Some("Bearer sec"))); // length mismatch + } +} diff --git a/crates/aingle_cortex/src/mcp/mod.rs b/crates/aingle_cortex/src/mcp/mod.rs index 48963423..532726d4 100644 --- a/crates/aingle_cortex/src/mcp/mod.rs +++ b/crates/aingle_cortex/src/mcp/mod.rs @@ -10,6 +10,10 @@ //! stdout is reserved for the JSON-RPC stream; all logging must go to stderr. mod convert; +#[cfg(feature = "mcp-http")] +pub mod http; +#[cfg(feature = "mcp-oauth")] +pub mod oauth; mod server; pub use server::AingleMcp; diff --git a/crates/aingle_cortex/src/mcp/oauth.rs b/crates/aingle_cortex/src/mcp/oauth.rs new file mode 100644 index 00000000..b16feb07 --- /dev/null +++ b/crates/aingle_cortex/src/mcp/oauth.rs @@ -0,0 +1,267 @@ +// Copyright 2019-2026 Apilium Technologies OÜ. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR Commercial + +//! OAuth 2.0 Resource Server support for the MCP HTTP endpoint: +//! RFC 9728 protected-resource metadata + Bearer JWT validation against a JWKS. + +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::RwLock; + +use jsonwebtoken::{decode, decode_header, Algorithm, DecodingKey, Validation}; +use serde::Deserialize; + +#[derive(Clone, Debug)] +pub struct OAuthConfig { + pub issuer: String, + pub resource: String, // expected `aud` + pub jwks_url: String, +} + +#[derive(Debug, Deserialize)] +pub struct Claims { + pub sub: Option, + #[serde(default)] + pub scope: Option, +} + +/// RS256-only; `iss`/`aud`/`exp` enforced. +pub fn validate_jwt( + token: &str, + key: &DecodingKey, + cfg: &OAuthConfig, +) -> Result { + let mut v = Validation::new(Algorithm::RS256); + v.set_required_spec_claims(&["exp", "iss", "aud"]); + v.set_issuer(&[cfg.issuer.as_str()]); + v.set_audience(&[cfg.resource.as_str()]); + v.validate_exp = true; + Ok(decode::(token, key, &v)?.claims) +} + +#[derive(Clone)] +pub struct JwksCache { + jwks_url: String, + keys: Arc>>, + last_refresh: Arc>>, + client: reqwest::Client, +} + +#[derive(Deserialize)] +struct Jwk { + kid: String, + n: String, + e: String, + #[serde(default)] + kty: String, +} +#[derive(Deserialize)] +struct JwkSet { + keys: Vec, +} + +impl JwksCache { + pub fn new(jwks_url: impl Into) -> Self { + Self { + jwks_url: jwks_url.into(), + keys: Arc::new(RwLock::new(HashMap::new())), + last_refresh: Arc::new(RwLock::new(None)), + client: reqwest::Client::new(), + } + } + /// Test-only: pre-seed a kid -> key (no network). + pub fn with_key(jwks_url: impl Into, kid: impl Into, key: DecodingKey) -> Self { + let mut m = HashMap::new(); + m.insert(kid.into(), key); + Self { + jwks_url: jwks_url.into(), + keys: Arc::new(RwLock::new(m)), + last_refresh: Arc::new(RwLock::new(None)), + client: reqwest::Client::new(), + } + } + /// Decoding key for `kid`, refreshing from JWKS on a miss (handles key rotation). + /// + /// Refreshes are debounced to at most one per 30s so that an attacker spamming + /// forged, unknown `kid` headers cannot trigger unbounded outbound JWKS fetches. + pub async fn key_for(&self, kid: &str) -> Option { + if let Some(k) = self.keys.read().await.get(kid).cloned() { + return Some(k); + } + // Debounce: at most one JWKS refresh per 30s, regardless of unknown-kid spam. + { + let last = *self.last_refresh.read().await; + if let Some(t) = last { + if t.elapsed() < std::time::Duration::from_secs(30) { + return None; + } + } + } + *self.last_refresh.write().await = Some(std::time::Instant::now()); + let _ = self.refresh().await; + self.keys.read().await.get(kid).cloned() + } + async fn refresh(&self) -> Result<(), String> { + let set: JwkSet = self + .client + .get(&self.jwks_url) + .send() + .await + .map_err(|e| e.to_string())? + .json() + .await + .map_err(|e| e.to_string())?; + let mut g = self.keys.write().await; + for jwk in set.keys { + if jwk.kty == "RSA" || jwk.kty.is_empty() { + if let Ok(k) = DecodingKey::from_rsa_components(&jwk.n, &jwk.e) { + g.insert(jwk.kid, k); + } + } + } + Ok(()) + } +} + +pub fn token_kid(token: &str) -> Option { + decode_header(token).ok().and_then(|h| h.kid) +} + +pub fn protected_resource_metadata(cfg: &OAuthConfig) -> serde_json::Value { + serde_json::json!({ "resource": cfg.resource, "authorization_servers": [cfg.issuer] }) +} + +#[cfg(test)] +mod tests { + use super::*; + use jsonwebtoken::{encode, EncodingKey, Header}; + + const PRIV: &str = include_str!("../../tests/fixtures/test_rsa_priv.pem"); + const PUB: &str = include_str!("../../tests/fixtures/test_rsa_pub.pem"); + fn cfg() -> OAuthConfig { + OAuthConfig { + issuer: "https://auth.test/realms/aingle".into(), + resource: "https://mcp.test/mcp".into(), + jwks_url: "https://auth.test/jwks".into(), + } + } + fn dkey() -> DecodingKey { + DecodingKey::from_rsa_pem(PUB.as_bytes()).unwrap() + } + fn ekey() -> EncodingKey { + EncodingKey::from_rsa_pem(PRIV.as_bytes()).unwrap() + } + fn sign(iss: &str, aud: &str, exp: i64) -> String { + #[derive(serde::Serialize)] + struct C<'a> { + iss: &'a str, + aud: &'a str, + exp: i64, + sub: &'a str, + } + let mut h = Header::new(Algorithm::RS256); + h.kid = Some("test-kid".into()); + encode( + &h, + &C { + iss, + aud, + exp, + sub: "u", + }, + &ekey(), + ) + .unwrap() + } + #[test] + fn valid_accepted() { + let c = cfg(); + assert!(validate_jwt(&sign(&c.issuer, &c.resource, 4_000_000_000), &dkey(), &c).is_ok()); + } + #[test] + fn wrong_aud_rejected() { + let c = cfg(); + assert!(validate_jwt( + &sign(&c.issuer, "https://evil/mcp", 4_000_000_000), + &dkey(), + &c + ) + .is_err()); + } + #[test] + fn wrong_iss_rejected() { + let c = cfg(); + assert!(validate_jwt( + &sign("https://evil/realm", &c.resource, 4_000_000_000), + &dkey(), + &c + ) + .is_err()); + } + #[test] + fn expired_rejected() { + let c = cfg(); + assert!(validate_jwt(&sign(&c.issuer, &c.resource, 1_000_000_000), &dkey(), &c).is_err()); + } + #[test] + fn missing_aud_rejected() { + let c = cfg(); + #[derive(serde::Serialize)] + struct C<'a> { + iss: &'a str, + exp: i64, + sub: &'a str, + } + let mut h = Header::new(Algorithm::RS256); + h.kid = Some("test-kid".into()); + let tok = encode( + &h, + &C { + iss: &c.issuer, + exp: 4_000_000_000, + sub: "u", + }, + &EncodingKey::from_rsa_pem(PRIV.as_bytes()).unwrap(), + ) + .unwrap(); + assert!( + validate_jwt(&tok, &dkey(), &c).is_err(), + "token without aud must be rejected" + ); + } + #[test] + fn missing_iss_rejected() { + let c = cfg(); + #[derive(serde::Serialize)] + struct C<'a> { + aud: &'a str, + exp: i64, + sub: &'a str, + } + let mut h = Header::new(Algorithm::RS256); + h.kid = Some("test-kid".into()); + let tok = encode( + &h, + &C { + aud: &c.resource, + exp: 4_000_000_000, + sub: "u", + }, + &EncodingKey::from_rsa_pem(PRIV.as_bytes()).unwrap(), + ) + .unwrap(); + assert!( + validate_jwt(&tok, &dkey(), &c).is_err(), + "token without iss must be rejected" + ); + } + #[test] + fn metadata_shape() { + let m = protected_resource_metadata(&cfg()); + assert_eq!(m["resource"], "https://mcp.test/mcp"); + assert_eq!( + m["authorization_servers"][0], + "https://auth.test/realms/aingle" + ); + } +} diff --git a/crates/aingle_cortex/src/server.rs b/crates/aingle_cortex/src/server.rs index 8720662e..22df5559 100644 --- a/crates/aingle_cortex/src/server.rs +++ b/crates/aingle_cortex/src/server.rs @@ -47,6 +47,16 @@ pub struct CortexConfig { pub db_path: Option, /// If `true`, serve MCP over stdio instead of binding a TCP listener. pub mcp_mode: bool, + /// Bearer token required on the `/mcp` HTTP endpoint. None = not configured. + pub mcp_http_token: Option, + /// Serve `/mcp` without auth (test mode, pre-OAuth). Default false. + pub mcp_http_allow_anonymous: bool, + /// OAuth issuer URL (e.g. https://auth.example/realms/aingle). Enables OAuth on /mcp when set. + pub mcp_oauth_issuer: Option, + /// OAuth protected-resource id = expected JWT audience (e.g. https://mcp.example/mcp). + pub mcp_oauth_resource: Option, + /// Optional explicit JWKS URL; if None, derived from the issuer (Keycloak certs path). + pub mcp_oauth_jwks_url: Option, } impl Default for CortexConfig { @@ -65,6 +75,11 @@ impl Default for CortexConfig { flush_interval_secs: 300, db_path: None, mcp_mode: false, + mcp_http_token: None, + mcp_http_allow_anonymous: false, + mcp_oauth_issuer: None, + mcp_oauth_resource: None, + mcp_oauth_jwks_url: None, } } } @@ -163,6 +178,88 @@ impl CortexServer { // Add the shared state to the router. let app = app.with_state(self.state.clone()); + // Mount the MCP-over-HTTP endpoint at `/mcp` (self-contained sub-router). + // Only mounted when a bearer token or anonymous mode is configured. + #[cfg(feature = "mcp-http")] + let app = { + #[allow(unused_mut)] + let mut app = app; + let public_hosts = std::env::var("AINGLE_PUBLIC_HOST") + .ok() + .map(|s| { + s.split(',') + .map(|x| x.trim().to_string()) + .filter(|x| !x.is_empty()) + .collect::>() + }) + .unwrap_or_default(); + + // Build the OAuth resource-server validator when issuer + resource are set. + #[cfg(feature = "mcp-oauth")] + let oauth_validator = match ( + self.config.mcp_oauth_issuer.clone(), + self.config.mcp_oauth_resource.clone(), + ) { + (Some(issuer), Some(resource)) => { + let jwks_url = self.config.mcp_oauth_jwks_url.clone().unwrap_or_else(|| { + format!( + "{}/protocol/openid-connect/certs", + issuer.trim_end_matches('/') + ) + }); + if jwks_url.starts_with("http://") + && !jwks_url.contains("127.0.0.1") + && !jwks_url.contains("localhost") + && !jwks_url.contains("[::1]") + { + tracing::warn!(jwks_url = %jwks_url, "OAuth JWKS URL is not HTTPS — keys could be MITM'd; use https in production"); + } + let cfg = crate::mcp::oauth::OAuthConfig { + issuer, + resource, + jwks_url: jwks_url.clone(), + }; + Some((cfg, crate::mcp::oauth::JwksCache::new(jwks_url))) + } + _ => None, + }; + + // RFC 9728 protected-resource metadata. + #[cfg(feature = "mcp-oauth")] + if let Some((ref cfg, _)) = oauth_validator { + let meta = crate::mcp::oauth::protected_resource_metadata(cfg); + app = app.route( + "/.well-known/oauth-protected-resource", + axum::routing::get(move || { + let meta = meta.clone(); + async move { axum::Json(meta) } + }), + ); + } + + #[cfg(feature = "mcp-oauth")] + let mcp_router = crate::mcp::http::mcp_http_router( + self.state.clone(), + self.config.mcp_http_token.clone(), + self.config.mcp_http_allow_anonymous, + public_hosts, + oauth_validator, + ); + #[cfg(not(feature = "mcp-oauth"))] + let mcp_router = crate::mcp::http::mcp_http_router( + self.state.clone(), + self.config.mcp_http_token.clone(), + self.config.mcp_http_allow_anonymous, + public_hosts, + ); + + if let Some(mcp_router) = mcp_router { + tracing::info!("MCP HTTP endpoint mounted at /mcp"); + app = app.nest("/mcp", mcp_router); + } + app + }; + // Add middleware layers (note: layers are applied in reverse order of definition). // Rate limiting layer. diff --git a/crates/aingle_cortex/tests/fixtures/test_rsa_priv.pem b/crates/aingle_cortex/tests/fixtures/test_rsa_priv.pem new file mode 100644 index 00000000..9b36a15c --- /dev/null +++ b/crates/aingle_cortex/tests/fixtures/test_rsa_priv.pem @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC4ZRZt+6H0c0jH +sPGUcJis+XMgTN/xRa2e2/B7RXwJSWddBt/Zob7zXqq30sWqyKREEDM3gXexdKjm +vDayQzxbEemKDGMSK5BIAHbwzd/MsMnZ+wxXDyX+TuLW4KxakTneEzxDLZPCbZE7 +chUM5/slItxuWUx4sHFb2nuFw+d+HJmsY87Fc7IGpzEXupj463QYYTqVSBuA9w+S +LjmGULfkU5ryrVXWLPFlYN0DPNh9/g5fwsEl3AFhOkfrQFcnS4f3G848IUU5N2BX +5vqZY19O0InOQqXi2h6iLge3Zm8+ubm3n/NwP82CUJibgahWI1Ay13BVe7Npcena +Sc2P5OCtAgMBAAECggEAMNr3umQ1YOM6oU4Mc9vxV8Mv8Zrsqqxfd/umF8MtPfio +3kj3/l2QjkLC0LmTjdBjVXVFXKt83xOdvKSZiVmvICH60oy9ow7Px4P3/41MgptE +n/CYMGjfFCYqi+wzPjvELUbDbLkisRz+odV0JltAe3JlHQej70Ywgrm8iJCDQTI3 +We1wPpGhekVOErFwwAD4qUTYC50/AMZmdCzxjanb2rCntLAqQTIrarAFRXVnq17v +sFbThlrWw3PYeqRDLaq7+4zY+Ig25auNnyAMC6fXycD//uIOwtHbk3mqQYY0CbJn +ocThzSVBzhKy7Cb8d1zWbzyjHlTJz3EY1Hf3QvwLsQKBgQDMGxzAeUifxBdZtSyy +in8ToaZM+3w7fwBKmbHr3AdS0l4QHzH2n52zubhspmOIzpkNNEM7gCU09gVKdeW+ +zcWjtj+aYbCotA1ENDyAXr4yvD1kSfoKkOXsUm3bZyAEjJPQ7W+4epMB+gWEayYu +Id+V2JQhs9Z66QSpgI7UCnx+mwKBgQDnRwT5eufmnyqyw1XiZmCwwF1g05ILFjrV +zIA6ozf5u2E/+Fd5JBhIvqfKc3RzlodXKaIF5TFTuiIAfQJnffCbkMF69DGE50I9 +enLwAHDfphVF9BC8Rdr6245vBgWVN3L3h8cWqWDalI1ZcIzjJAci+Ac9A4qQIayG +YwlLVOIuVwKBgBmIErKo+UKy/tDD7xFubbMA95KNqt6uZlTuoOkGHwxnMEkH/fIB +yXJf+wfsUGsenNqf/QEGaKEVXVgRI2oYx25RL+eZ7MVNsJMljaPpNhWWon9XGaYU +tg43yXI2ljS7eQobiWwkORt4MRR9ass+hX9zFiSZIG1DlMe3QyaXITedAoGBAL9S +kfvsP8EZtNHTI845kQ1G5Th/bVY8molZNk7LzTeLWkT4JSyyvmAXUGj7H7+rK+Tg +LQehdS/lT9GFmKKSnaOjmVskfX6LbNG85S5DvxxUoU6QO/Pz+dYCIQ8ZNS8egr3g +lcLadaP8tf3iTLpuiHTWJ+7CajMXmEhHz12BJ+bbAoGBALwVGASOeTN6Nfgr7Qaw +iMlRI63RVpnL8FkB0GpNU0z1dkj9zdGlNvP1bCun/aoV9jbSl6oq5uenkWO7VMbH +se9xZaY42VfYko68Ix6t8ocz73ifDeDHFAKKhhrEUuj3XbX897jvnGFxtIWfCAWd +8qBuAm66z2qxWn/8P9ZhWwG+ +-----END PRIVATE KEY----- diff --git a/crates/aingle_cortex/tests/fixtures/test_rsa_pub.pem b/crates/aingle_cortex/tests/fixtures/test_rsa_pub.pem new file mode 100644 index 00000000..2cc3baa9 --- /dev/null +++ b/crates/aingle_cortex/tests/fixtures/test_rsa_pub.pem @@ -0,0 +1,9 @@ +-----BEGIN PUBLIC KEY----- +MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAuGUWbfuh9HNIx7DxlHCY +rPlzIEzf8UWtntvwe0V8CUlnXQbf2aG+816qt9LFqsikRBAzN4F3sXSo5rw2skM8 +WxHpigxjEiuQSAB28M3fzLDJ2fsMVw8l/k7i1uCsWpE53hM8Qy2Twm2RO3IVDOf7 +JSLcbllMeLBxW9p7hcPnfhyZrGPOxXOyBqcxF7qY+Ot0GGE6lUgbgPcPki45hlC3 +5FOa8q1V1izxZWDdAzzYff4OX8LBJdwBYTpH60BXJ0uH9xvOPCFFOTdgV+b6mWNf +TtCJzkKl4toeoi4Ht2ZvPrm5t5/zcD/NglCYm4GoViNQMtdwVXuzaXHp2knNj+Tg +rQIDAQAB +-----END PUBLIC KEY----- diff --git a/crates/aingle_cortex/tests/mcp_http_integration.rs b/crates/aingle_cortex/tests/mcp_http_integration.rs new file mode 100644 index 00000000..fe9bfd7c --- /dev/null +++ b/crates/aingle_cortex/tests/mcp_http_integration.rs @@ -0,0 +1,157 @@ +// Copyright 2019-2026 Apilium Technologies OÜ. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR Commercial +//! HTTP integration test for the MCP Streamable endpoint at /mcp. +//! +//! This test exercises `/mcp` over real HTTP (it does not call internals): +//! 1. POST /mcp with NO auth -> 401 +//! 2. POST /mcp with WRONG bearer -> 401 +//! 3. POST /mcp with CORRECT bearer + MCP `initialize` -> 2xx + body contains "serverInfo" +//! +//! Approach: raw `reqwest`. rmcp's Streamable HTTP transport answers a plain POST +//! `initialize` with the JSON-RPC result as an SSE `text/event-stream` body +//! (`event: message\ndata: {...serverInfo...}`), so `body.contains("serverInfo")` +//! holds whether the body is plain JSON or SSE. +#![cfg(feature = "mcp-http")] + +use aingle_cortex::{CortexConfig, CortexServer}; + +async fn boot(token: Option, anon: bool) -> (u16, tokio::task::JoinHandle<()>) { + // pick a free port + let port = { + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let p = l.local_addr().unwrap().port(); + drop(l); + p + }; + let mut config = CortexConfig::default() + .with_host("127.0.0.1") + .with_port(port); + config.db_path = Some(":memory:".to_string()); + config.mcp_http_token = token; + config.mcp_http_allow_anonymous = anon; + let server = CortexServer::new(config).unwrap(); + let handle = tokio::spawn(async move { + let _ = server.run().await; + }); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + (port, handle) +} + +#[tokio::test] +async fn mcp_http_auth_and_initialize() { + let (port, h) = boot(Some("test-token-123".into()), false).await; + let client = reqwest::Client::new(); + let url = format!("http://127.0.0.1:{port}/mcp"); + let init = serde_json::json!({ + "jsonrpc":"2.0","id":1,"method":"initialize", + "params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"0"}} + }); + + // 1) no auth -> 401 + let r = client + .post(&url) + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + assert_eq!( + r.status(), + reqwest::StatusCode::UNAUTHORIZED, + "no-auth should be 401" + ); + + // 2) wrong token -> 401 + let r = client + .post(&url) + .bearer_auth("nope") + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + assert_eq!( + r.status(), + reqwest::StatusCode::UNAUTHORIZED, + "wrong token should be 401" + ); + + // 3) correct token -> 2xx + serverInfo + let r = client + .post(&url) + .bearer_auth("test-token-123") + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + let status = r.status(); + let body = r.text().await.unwrap(); + assert!( + status.is_success(), + "expected 2xx, got {status}; body={body}" + ); + assert!( + body.contains("serverInfo"), + "body lacked serverInfo: {body}" + ); + + h.abort(); +} + +#[tokio::test] +async fn mcp_http_not_mounted_without_token() { + // Neither a token nor anonymous mode -> /mcp must NOT exist. + let (port, h) = boot(None, false).await; + let client = reqwest::Client::new(); + let url = format!("http://127.0.0.1:{port}/mcp"); + let init = serde_json::json!({ + "jsonrpc":"2.0","id":1,"method":"initialize", + "params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"0"}} + }); + let r = client + .post(&url) + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + // Route absent -> 404 (NOT 401, which would mean it IS mounted+guarded; NOT 2xx). + assert_eq!( + r.status(), + reqwest::StatusCode::NOT_FOUND, + "/mcp must be absent without token/anon, got {}", + r.status() + ); + h.abort(); +} + +#[tokio::test] +async fn mcp_http_anonymous_serves_without_auth() { + let (port, h) = boot(None, true).await; // allow_anonymous = true + let client = reqwest::Client::new(); + let url = format!("http://127.0.0.1:{port}/mcp"); + let init = serde_json::json!({ + "jsonrpc":"2.0","id":1,"method":"initialize", + "params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"0"}} + }); + // No Authorization header at all -> still 2xx + serverInfo because anonymous. + let r = client + .post(&url) + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + let status = r.status(); + let body = r.text().await.unwrap(); + assert!( + status.is_success(), + "anonymous should serve, got {status}; body={body}" + ); + assert!( + body.contains("serverInfo"), + "anonymous body lacked serverInfo: {body}" + ); + h.abort(); +} diff --git a/crates/aingle_cortex/tests/mcp_oauth_integration.rs b/crates/aingle_cortex/tests/mcp_oauth_integration.rs new file mode 100644 index 00000000..15bfd4a8 --- /dev/null +++ b/crates/aingle_cortex/tests/mcp_oauth_integration.rs @@ -0,0 +1,160 @@ +// Copyright 2019-2026 Apilium Technologies OÜ. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR Commercial +//! OAuth resource-server integration tests for /mcp. +#![cfg(feature = "mcp-oauth")] + +use aingle_cortex::{CortexConfig, CortexServer}; +use base64::Engine; +use jsonwebtoken::{encode, Algorithm, EncodingKey, Header}; +use rsa::pkcs8::DecodePublicKey; +use rsa::traits::PublicKeyParts; + +const PRIV: &str = include_str!("fixtures/test_rsa_priv.pem"); +const PUB: &str = include_str!("fixtures/test_rsa_pub.pem"); + +fn b64u(bytes: &[u8]) -> String { + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes) +} + +fn jwks_json() -> serde_json::Value { + let pk = rsa::RsaPublicKey::from_public_key_pem(PUB).unwrap(); + let n = b64u(&pk.n().to_bytes_be()); + let e = b64u(&pk.e().to_bytes_be()); + serde_json::json!({ "keys": [ {"kty":"RSA","alg":"RS256","use":"sig","kid":"test-kid","n":n,"e":e} ] }) +} + +fn sign(iss: &str, aud: &str, exp: i64) -> String { + #[derive(serde::Serialize)] + struct C<'a> { + iss: &'a str, + aud: &'a str, + exp: i64, + sub: &'a str, + } + let mut h = Header::new(Algorithm::RS256); + h.kid = Some("test-kid".into()); + encode( + &h, + &C { + iss, + aud, + exp, + sub: "u", + }, + &EncodingKey::from_rsa_pem(PRIV.as_bytes()).unwrap(), + ) + .unwrap() +} + +async fn free_port() -> u16 { + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let p = l.local_addr().unwrap().port(); + drop(l); + p +} + +#[tokio::test] +async fn oauth_resource_server_end_to_end() { + // 1) tiny JWKS server + let jwks_port = free_port().await; + let jwks = jwks_json(); + let jwks_app = axum::Router::new().route( + "/certs", + axum::routing::get(move || { + let j = jwks.clone(); + async move { axum::Json(j) } + }), + ); + let jwks_listener = tokio::net::TcpListener::bind(("127.0.0.1", jwks_port)) + .await + .unwrap(); + tokio::spawn(async move { + axum::serve(jwks_listener, jwks_app).await.unwrap(); + }); + + // 2) cortex with OAuth + let cortex_port = free_port().await; + let issuer = "https://auth.test/realms/aingle".to_string(); + let resource = format!("http://127.0.0.1:{cortex_port}/mcp"); + let mut config = CortexConfig::default() + .with_host("127.0.0.1") + .with_port(cortex_port); + config.db_path = Some(":memory:".into()); + config.mcp_oauth_issuer = Some(issuer.clone()); + config.mcp_oauth_resource = Some(resource.clone()); + config.mcp_oauth_jwks_url = Some(format!("http://127.0.0.1:{jwks_port}/certs")); + // The /mcp unauthorized challenge derives its `resource_metadata` URL from the + // OAuth `resource` (the canonical public `/mcp` URL) when OAuth is configured. + // We deliberately do NOT set `AINGLE_PUBLIC_HOST` here so this test exercises + // the resource-derived path of the RFC 9728 `WWW-Authenticate` challenge. + let server = CortexServer::new(config).unwrap(); + tokio::spawn(async move { + let _ = server.run().await; + }); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + + let client = reqwest::Client::new(); + let base = format!("http://127.0.0.1:{cortex_port}"); + let init = serde_json::json!({"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"0"}}}); + + // metadata + let r = client + .get(format!("{base}/.well-known/oauth-protected-resource")) + .send() + .await + .unwrap(); + assert!(r.status().is_success(), "metadata status {}", r.status()); + let meta: serde_json::Value = r.json().await.unwrap(); + assert_eq!(meta["resource"], resource); + assert_eq!(meta["authorization_servers"][0], issuer); + + // no token -> 401 + WWW-Authenticate w/ resource_metadata + let r = client + .post(format!("{base}/mcp")) + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + assert_eq!(r.status(), reqwest::StatusCode::UNAUTHORIZED); + let www = r + .headers() + .get("www-authenticate") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + assert!(www.contains("resource_metadata"), "WWW-Authenticate: {www}"); + + // valid JWT -> 2xx + serverInfo + let good = sign(&issuer, &resource, 4_000_000_000); + let r = client + .post(format!("{base}/mcp")) + .bearer_auth(&good) + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + let status = r.status(); + let body = r.text().await.unwrap(); + assert!(status.is_success(), "valid jwt status {status} body {body}"); + assert!( + body.contains("serverInfo"), + "valid jwt body lacked serverInfo: {body}" + ); + + // wrong audience -> 401 + let bad = sign(&issuer, "https://evil/mcp", 4_000_000_000); + let r = client + .post(format!("{base}/mcp")) + .bearer_auth(&bad) + .header("Accept", "application/json, text/event-stream") + .json(&init) + .send() + .await + .unwrap(); + assert_eq!( + r.status(), + reqwest::StatusCode::UNAUTHORIZED, + "wrong-aud must be 401" + ); +}