diff --git a/Cargo.lock b/Cargo.lock index 26524b1..0e384ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -199,6 +199,7 @@ dependencies = [ "anyhow", "axum", "clap", + "futures-util", "http", "reqwest", "serde_json", diff --git a/crates/aver-scope-shim/Cargo.toml b/crates/aver-scope-shim/Cargo.toml index 3507065..44ca4d1 100644 --- a/crates/aver-scope-shim/Cargo.toml +++ b/crates/aver-scope-shim/Cargo.toml @@ -23,10 +23,11 @@ path = "src/lib.rs" anyhow.workspace = true axum = "0.8" clap.workspace = true +futures-util = "0.3" http = "1" reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream"] } sha2 = "0.10" -tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "signal", "process"] } +tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "signal", "process", "time", "sync"] } tower = "0.5" [dev-dependencies] diff --git a/crates/aver-scope-shim/src/lib.rs b/crates/aver-scope-shim/src/lib.rs index 034ac6d..2d1f0e4 100644 --- a/crates/aver-scope-shim/src/lib.rs +++ b/crates/aver-scope-shim/src/lib.rs @@ -9,14 +9,16 @@ //! //! Layout: //! - [`derive_scope`] / [`scope_from_git`] — slug derivation rules. -//! - The HTTP proxy lives in `main.rs`; this module exports the building -//! blocks for unit tests. +//! - [`proxy`] — the streaming HTTP proxy (router, config, URL joining); +//! `main.rs` is only CLI parsing and startup wiring. use std::path::Path; use std::process::Command; use sha2::{Digest, Sha256}; +pub mod proxy; + /// Per-startup scope decision: either a derived `proj/...` scope, an env-var /// override, or the hardcoded "global" fallback. #[derive(Debug, Clone, PartialEq, Eq)] @@ -55,7 +57,8 @@ pub fn slug_hash(input: &str) -> String { /// Derive the scope a shim should inject for a given working directory. /// /// Precedence: -/// 1. `cli_override` if supplied. +/// 1. `cli_override` if supplied (empty/whitespace-only values are ignored, +/// matching the `AVER_DEFAULT_SCOPE` handling below). /// 2. Git remote origin URL hash, if `cwd` is inside a git repo with origin. /// 3. Git toplevel absolute path hash (per ADR-0022 amendment / council /// risk #3), if inside a git repo without origin. @@ -66,7 +69,7 @@ pub fn derive_scope( cli_override: Option<&str>, env_default: Option<&str>, ) -> DerivedScope { - if let Some(s) = cli_override { + if let Some(s) = cli_override.filter(|v| !v.trim().is_empty()) { return DerivedScope { scope: s.to_string(), source: ScopeSource::CliOverride, diff --git a/crates/aver-scope-shim/src/main.rs b/crates/aver-scope-shim/src/main.rs index 535041f..98ab244 100644 --- a/crates/aver-scope-shim/src/main.rs +++ b/crates/aver-scope-shim/src/main.rs @@ -2,20 +2,18 @@ //! //! Binds to `127.0.0.1:0` (ephemeral TCP per council verdict 2026-05-10), //! prints the bound URL on stdout, and forwards every HTTP request to the -//! upstream aver-server with `X-Aver-Scope` injected. +//! upstream aver-server with `X-Aver-Scope` injected. The proxy itself lives +//! in [`aver_scope_shim::proxy`]; this binary is CLI parsing and startup +//! wiring only. use std::net::SocketAddr; use std::path::PathBuf; use anyhow::Context; -use axum::body::Body; -use axum::extract::State; -use axum::http::{HeaderMap, HeaderName, HeaderValue, Request, Response, StatusCode, Uri}; -use axum::response::IntoResponse; -use axum::routing::any; use clap::Parser; use aver_scope_shim::derive_scope; +use aver_scope_shim::proxy::{DEFAULT_UPSTREAM_TIMEOUT, ShimConfig, router, scope_header_value}; #[derive(Debug, Parser)] #[command( @@ -23,7 +21,9 @@ use aver_scope_shim::derive_scope; about = "Per-workspace HTTP MCP proxy that injects X-Aver-Scope (ADR-0022)" )] struct Cli { - /// Upstream aver-server MCP URL. + /// Upstream aver-server MCP URL. Request paths are appended to this base; + /// a duplicated `/mcp` prefix is collapsed so the shim URL works with or + /// without a trailing `/mcp`. #[arg( long, env = "AVER_UPSTREAM_URL", @@ -41,13 +41,6 @@ struct Cli { bind: SocketAddr, } -#[derive(Clone)] -struct AppState { - upstream: String, - scope: String, - client: reqwest::Client, -} - #[tokio::main] async fn main() -> anyhow::Result<()> { let cli = Cli::parse(); @@ -58,6 +51,17 @@ async fn main() -> anyhow::Result<()> { let env_default = std::env::var("AVER_DEFAULT_SCOPE").ok(); let derived = derive_scope(&cwd, cli.scope.as_deref(), env_default.as_deref()); + // Validate the scope as a legal header value once at startup so a + // misconfigured scope fails fast instead of producing a 502 per request. + let scope = scope_header_value(&derived.scope) + .with_context(|| format!("scope {:?} is not a legal HTTP header value", derived.scope))?; + + let app = router(ShimConfig { + upstream: cli.upstream.clone(), + scope: Some(scope), + upstream_timeout: DEFAULT_UPSTREAM_TIMEOUT, + })?; + let listener = tokio::net::TcpListener::bind(cli.bind).await?; let bound = listener.local_addr()?; eprintln!( @@ -66,121 +70,6 @@ async fn main() -> anyhow::Result<()> { ); println!("http://{bound}"); - let state = AppState { - upstream: cli.upstream, - scope: derived.scope, - client: reqwest::Client::builder() - .pool_idle_timeout(std::time::Duration::from_secs(30)) - .build() - .context("building reqwest client")?, - }; - - let app = axum::Router::new() - .route("/{*rest}", any(forward)) - .route("/", any(forward)) - .with_state(state); - axum::serve(listener, app.into_make_service()).await?; Ok(()) } - -async fn forward(State(state): State, request: Request) -> Response { - match forward_inner(state, request).await { - Ok(resp) => resp, - Err(err) => { - eprintln!("aver-scope-shim: forward error: {err:#}"); - (StatusCode::BAD_GATEWAY, format!("upstream error: {err}")).into_response() - } - } -} - -async fn forward_inner(state: AppState, request: Request) -> anyhow::Result> { - let (parts, body) = request.into_parts(); - let method = parts.method.clone(); - let uri: Uri = parts.uri.clone(); - let path_and_query = uri.path_and_query().map(|p| p.as_str()).unwrap_or(""); - - // Build target URL: upstream base + path/query of incoming request. - // ADR-0022: every forwarded request is rewritten to land under upstream. - let upstream_base: Uri = state.upstream.parse().context("parsing upstream URL")?; - let upstream_str = if path_and_query.is_empty() || path_and_query == "/" { - state.upstream.clone() - } else { - // If upstream path has a tail like `/mcp`, append the request's tail. - // Path prefixing: upstream_base.path() + path_and_query — but axum's - // captured `*rest` is the suffix after `/`. Simpler: concat strings. - let base = upstream_base.to_string(); - let base = base.trim_end_matches('/'); - format!("{base}{path_and_query}") - }; - - let body_bytes = axum::body::to_bytes(body, usize::MAX) - .await - .context("buffering request body")?; - let mut req_builder = state - .client - .request(method, &upstream_str) - .body(body_bytes.to_vec()); - - let mut forward_headers = HeaderMap::new(); - for (name, value) in parts.headers.iter() { - // Don't forward hop-by-hop or host headers. - let lower = name.as_str().to_ascii_lowercase(); - if matches!( - lower.as_str(), - "host" - | "content-length" - | "connection" - | "keep-alive" - | "proxy-authenticate" - | "proxy-authorization" - | "te" - | "trailer" - | "transfer-encoding" - | "upgrade" - | "x-aver-scope" - ) { - continue; - } - forward_headers.insert(name.clone(), value.clone()); - } - forward_headers.insert( - HeaderName::from_static("x-aver-scope"), - HeaderValue::from_str(&state.scope).context("constructing X-Aver-Scope header value")?, - ); - req_builder = req_builder.headers(forward_headers); - - let upstream_resp = req_builder - .send() - .await - .context("sending request to upstream")?; - let status = upstream_resp.status(); - let resp_headers = upstream_resp.headers().clone(); - let resp_bytes = upstream_resp - .bytes() - .await - .context("buffering upstream response body")?; - let mut response = Response::builder().status(status); - if let Some(hs) = response.headers_mut() { - for (name, value) in resp_headers.iter() { - let lower = name.as_str().to_ascii_lowercase(); - if matches!( - lower.as_str(), - "connection" - | "keep-alive" - | "proxy-authenticate" - | "proxy-authorization" - | "te" - | "trailer" - | "transfer-encoding" - | "upgrade" - ) { - continue; - } - hs.insert(name.clone(), value.clone()); - } - } - response - .body(Body::from(resp_bytes)) - .context("building forwarded response") -} diff --git a/crates/aver-scope-shim/src/proxy.rs b/crates/aver-scope-shim/src/proxy.rs new file mode 100644 index 0000000..49a455d --- /dev/null +++ b/crates/aver-scope-shim/src/proxy.rs @@ -0,0 +1,271 @@ +//! Streaming HTTP proxy: forwards every request to the upstream +//! aver-server with `X-Aver-Scope` injected. +//! +//! Bodies stream in both directions — no store-and-forward — so long-lived +//! SSE responses (MCP streamable HTTP) reach the client incrementally and +//! shim memory stays bounded. Request bodies are hard-capped at +//! [`MAX_REQUEST_BODY_BYTES`] (413 beyond); response bodies are never +//! capped. The upstream deadline covers connect through response headers; +//! once headers arrive the body streams without a total timeout. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use anyhow::Context; +use axum::body::{Body, HttpBody}; +use axum::extract::State; +use axum::http::{HeaderMap, HeaderName, HeaderValue, Request, Response, StatusCode}; +use axum::response::IntoResponse; +use axum::routing::any; +use futures_util::StreamExt as _; + +/// Request bodies larger than this are rejected with 413. Responses are +/// never capped: SSE streams are long-lived by design. +pub const MAX_REQUEST_BODY_BYTES: u64 = 16 * 1024 * 1024; + +/// Default deadline for the upstream to produce response headers. +pub const DEFAULT_UPSTREAM_TIMEOUT: Duration = Duration::from_secs(30); + +/// Configuration for the shim's HTTP proxy. +#[derive(Clone)] +pub struct ShimConfig { + /// Upstream base URL, e.g. `http://127.0.0.1:3317/mcp`. + pub upstream: String, + /// Scope to inject as `X-Aver-Scope`. `None` injects nothing (any + /// client-supplied header is still stripped, never trusted). + pub scope: Option, + /// Deadline for the upstream to produce response headers. + pub upstream_timeout: Duration, +} + +/// Validate a scope string as a legal HTTP header value. Called once at +/// startup so a misconfigured scope fails fast instead of producing a 502 +/// on every request. +pub fn scope_header_value( + scope: &str, +) -> Result { + HeaderValue::from_str(scope) +} + +/// Build the shim's axum router. +pub fn router(config: ShimConfig) -> anyhow::Result { + let state = AppState { + upstream: config.upstream, + scope: config.scope, + upstream_timeout: config.upstream_timeout, + // No total `timeout()` here: it would also cap response-body + // streaming and kill long-lived SSE channels. The deadline is + // applied to `send()` (headers) in `forward_inner` instead. + client: reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .pool_idle_timeout(Duration::from_secs(30)) + .build() + .context("building reqwest client")?, + }; + Ok(axum::Router::new() + .route("/{*rest}", any(forward)) + .route("/", any(forward)) + .with_state(state)) +} + +/// Join the upstream base URL with an incoming path-and-query. +/// +/// The default upstream already ends in `/mcp`; a harness that appends +/// `/mcp` to the shim URL must not be forwarded to `/mcp/mcp`, so a +/// duplicated `/mcp` prefix is collapsed. +pub fn join_upstream(base: &str, path_and_query: &str) -> String { + let base = base.trim_end_matches('/'); + if path_and_query.is_empty() || path_and_query == "/" { + return base.to_string(); + } + let path_only = path_and_query.split('?').next().unwrap_or(""); + let suffix = + if base.ends_with("/mcp") && (path_only == "/mcp" || path_only.starts_with("/mcp/")) { + let rest = &path_and_query["/mcp".len()..]; + // "/mcp/" collapses onto the base itself, not a trailing-slash variant. + if rest == "/" { "" } else { rest } + } else { + path_and_query + }; + format!("{base}{suffix}") +} + +#[derive(Clone)] +struct AppState { + upstream: String, + scope: Option, + upstream_timeout: Duration, + client: reqwest::Client, +} + +async fn forward(State(state): State, request: Request) -> Response { + match forward_inner(state, request).await { + Ok(resp) => resp, + Err(err) => { + eprintln!("aver-scope-shim: forward error: {err:#}"); + (StatusCode::BAD_GATEWAY, format!("upstream error: {err}")).into_response() + } + } +} + +async fn forward_inner(state: AppState, request: Request) -> anyhow::Result> { + let (parts, body) = request.into_parts(); + + // Reject oversized uploads up front when the length is known. + if let Some(len) = parts + .headers + .get(axum::http::header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + && len > MAX_REQUEST_BODY_BYTES + { + return Ok(too_large_response()); + } + + let method = parts.method.clone(); + let path_and_query = parts.uri.path_and_query().map(|p| p.as_str()).unwrap_or(""); + let target = join_upstream(&state.upstream, path_and_query); + + let mut req_builder = state.client.request(method, &target); + + // Stream the request body with a hard cap. Chunked bodies have no + // upfront length, so a mid-stream trip flags the cap and the aborted + // `send()` below is mapped to 413. + let too_large = Arc::new(AtomicBool::new(false)); + if !body.is_end_stream() { + let flag = Arc::clone(&too_large); + let mut used = 0u64; + let stream = body.into_data_stream().map(move |chunk| { + chunk + .map_err(|err| -> Box { err.into() }) + .and_then(|bytes: axum::body::Bytes| { + used = used.saturating_add(bytes.len() as u64); + if used > MAX_REQUEST_BODY_BYTES { + flag.store(true, Ordering::Relaxed); + Err("request body too large".into()) + } else { + Ok(bytes) + } + }) + }); + req_builder = req_builder.body(reqwest::Body::wrap_stream(stream)); + } + + let mut forward_headers = HeaderMap::new(); + for (name, value) in parts.headers.iter() { + // Don't forward hop-by-hop, host, or framing headers (reqwest + // re-computes those), nor a client-supplied X-Aver-Scope (never + // trusted — the shim is the authority). + if is_hop_by_hop(name) + || matches!(name.as_str(), "host" | "content-length" | "x-aver-scope") + { + continue; + } + forward_headers.insert(name.clone(), value.clone()); + } + if let Some(scope) = &state.scope { + forward_headers.insert(HeaderName::from_static("x-aver-scope"), scope.clone()); + } + req_builder = req_builder.headers(forward_headers); + + // Deadline covers connect + request upload + response headers only; + // the response body then streams without a total timeout so SSE + // channels stay open. + let sent = tokio::time::timeout(state.upstream_timeout, req_builder.send()).await; + let upstream_resp = match sent { + Ok(Ok(resp)) => resp, + Ok(Err(err)) => { + if too_large.load(Ordering::Relaxed) { + return Ok(too_large_response()); + } + return Err(err).context("sending request to upstream"); + } + Err(_) => { + return Ok(( + StatusCode::GATEWAY_TIMEOUT, + format!( + "upstream produced no response headers within {:?}", + state.upstream_timeout + ), + ) + .into_response()); + } + }; + + let status = upstream_resp.status(); + let resp_headers = upstream_resp.headers().clone(); + let mut response = Response::builder().status(status); + if let Some(hs) = response.headers_mut() { + for (name, value) in resp_headers.iter() { + if is_hop_by_hop(name) { + continue; + } + hs.insert(name.clone(), value.clone()); + } + } + // Stream the response body through: SSE channels stay open and chunk + // timing is preserved. + response + .body(Body::from_stream(upstream_resp.bytes_stream())) + .context("building forwarded response") +} + +/// Hop-by-hop headers that must never cross the proxy in either direction +/// (RFC 9110 §7.6.1). `HeaderName::as_str` is lowercase by invariant. +fn is_hop_by_hop(name: &HeaderName) -> bool { + matches!( + name.as_str(), + "connection" + | "keep-alive" + | "proxy-authenticate" + | "proxy-authorization" + | "te" + | "trailer" + | "transfer-encoding" + | "upgrade" + ) +} + +fn too_large_response() -> Response { + ( + StatusCode::PAYLOAD_TOO_LARGE, + format!("request body exceeds {MAX_REQUEST_BODY_BYTES} byte limit"), + ) + .into_response() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn join_strips_duplicated_mcp_prefix() { + let base = "http://127.0.0.1:3317/mcp"; + assert_eq!(join_upstream(base, "/mcp"), base); + assert_eq!(join_upstream(base, "/mcp/"), base); + assert_eq!(join_upstream(base, "/mcp/foo"), format!("{base}/foo")); + assert_eq!(join_upstream(base, "/mcp?x=1"), format!("{base}?x=1")); + } + + #[test] + fn join_appends_other_paths_under_upstream() { + let base = "http://127.0.0.1:3317/mcp"; + assert_eq!(join_upstream(base, "/foo"), format!("{base}/foo")); + assert_eq!(join_upstream(base, "/"), base); + assert_eq!(join_upstream(base, ""), base); + } + + #[test] + fn join_does_not_strip_mcp_when_base_has_no_mcp_tail() { + let base = "http://127.0.0.1:3317"; + assert_eq!(join_upstream(base, "/mcp"), format!("{base}/mcp")); + } + + #[test] + fn scope_header_value_rejects_control_characters() { + assert!(scope_header_value("proj/abc").is_ok()); + assert!(scope_header_value("bad\nvalue").is_err()); + assert!(scope_header_value("bad\rvalue").is_err()); + } +} diff --git a/crates/aver-scope-shim/tests/proxy_integration.rs b/crates/aver-scope-shim/tests/proxy_integration.rs new file mode 100644 index 0000000..720f0e7 --- /dev/null +++ b/crates/aver-scope-shim/tests/proxy_integration.rs @@ -0,0 +1,269 @@ +//! HTTP-path integration tests for the shim proxy. +//! +//! Each test spawns a real upstream axum server and the shim router on +//! ephemeral localhost ports, then drives the shim with a reqwest client. +//! Everything is in-process and offline. + +use std::sync::Arc; +use std::time::Duration; + +use axum::body::Body; +use axum::extract::Request; +use axum::http::{HeaderMap, HeaderValue, StatusCode, header}; +use axum::response::IntoResponse; +use axum::routing::any; +use futures_util::StreamExt as _; +use tokio::sync::Notify; + +use aver_scope_shim::proxy::{MAX_REQUEST_BODY_BYTES, ShimConfig, router}; + +/// Spawn `app` on an ephemeral localhost port; return its base URL. +async fn spawn(app: axum::Router) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") +} + +/// Spawn the shim proxying to `upstream` with the given scope. +async fn spawn_shim(upstream: &str, scope: Option<&str>, timeout: Duration) -> String { + let app = router(ShimConfig { + upstream: upstream.to_string(), + scope: scope.map(|s| HeaderValue::from_str(s).unwrap()), + upstream_timeout: timeout, + }) + .unwrap(); + spawn(app).await +} + +/// Upstream that reports which `X-Aver-Scope` value (if any) it received. +fn echo_upstream() -> axum::Router { + async fn echo(headers: HeaderMap) -> String { + headers + .get("x-aver-scope") + .map(|v| v.to_str().unwrap().to_string()) + .unwrap_or_else(|| "".to_string()) + } + axum::Router::new().route("/{*rest}", any(echo)) +} + +#[tokio::test] +async fn injects_scope_header() { + let upstream = spawn(echo_upstream()).await; + let shim = spawn_shim(&upstream, Some("proj/abc123"), Duration::from_secs(5)).await; + + let body = reqwest::get(format!("{shim}/mcp")).await.unwrap(); + assert_eq!(body.status(), StatusCode::OK); + assert_eq!(body.text().await.unwrap(), "proj/abc123"); +} + +#[tokio::test] +async fn no_scope_configured_passes_through_without_header() { + let upstream = spawn(echo_upstream()).await; + let shim = spawn_shim(&upstream, None, Duration::from_secs(5)).await; + + // Even a client-supplied header must be stripped: the shim never trusts + // downstream scope claims. + let resp = reqwest::Client::new() + .get(format!("{shim}/mcp")) + .header("x-aver-scope", "proj/forged") + .send() + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(resp.text().await.unwrap(), ""); +} + +#[tokio::test] +async fn client_scope_header_is_replaced_by_configured_scope() { + let upstream = spawn(echo_upstream()).await; + let shim = spawn_shim(&upstream, Some("proj/real"), Duration::from_secs(5)).await; + + let resp = reqwest::Client::new() + .get(format!("{shim}/mcp")) + .header("x-aver-scope", "proj/forged") + .send() + .await + .unwrap(); + assert_eq!(resp.text().await.unwrap(), "proj/real"); +} + +#[tokio::test] +async fn sse_response_streams_incrementally() { + // Upstream emits chunk 1, parks until the test releases it, then emits + // chunk 2. If the shim buffered the whole body, the client could never + // observe chunk 1 while the upstream is still parked. + let release = Arc::new(Notify::new()); + let release_in_handler = Arc::clone(&release); + let upstream = spawn(axum::Router::new().route( + "/mcp", + any(move || { + let release = Arc::clone(&release_in_handler); + async move { + let stream = futures_util::stream::once(async { + Ok::<_, std::io::Error>(axum::body::Bytes::from("data: one\n\n")) + }) + .chain(futures_util::stream::once(async move { + release.notified().await; + Ok(axum::body::Bytes::from("data: two\n\n")) + })); + ( + [(header::CONTENT_TYPE, "text/event-stream")], + Body::from_stream(stream), + ) + .into_response() + } + }), + )) + .await; + let shim = spawn_shim(&upstream, Some("proj/sse"), Duration::from_secs(5)).await; + + let resp = reqwest::get(format!("{shim}/mcp")).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!( + resp.headers().get(header::CONTENT_TYPE).unwrap(), + "text/event-stream" + ); + + let mut stream = resp.bytes_stream(); + // Chunk 1 must arrive while the upstream is still parked on `release`. + let first = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("first chunk never arrived — shim is buffering the stream") + .unwrap() + .unwrap(); + assert_eq!(&first[..], b"data: one\n\n"); + + release.notify_one(); + let second = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("second chunk never arrived") + .unwrap() + .unwrap(); + assert_eq!(&second[..], b"data: two\n\n"); +} + +#[tokio::test] +async fn request_body_over_cap_is_rejected_413() { + let upstream = spawn(echo_upstream()).await; + let shim = spawn_shim(&upstream, Some("proj/cap"), Duration::from_secs(5)).await; + + // Known Content-Length above the cap: rejected up front. + let body = vec![b'x'; MAX_REQUEST_BODY_BYTES as usize + 1]; + let resp = reqwest::Client::new() + .post(format!("{shim}/mcp")) + .body(body) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE); +} + +/// Upstream that drains the request body before answering. +fn draining_upstream() -> axum::Router { + async fn drain(req: Request) -> StatusCode { + let mut stream = req.into_body().into_data_stream(); + while stream.next().await.is_some() {} + StatusCode::OK + } + axum::Router::new().route("/{*rest}", any(drain)) +} + +#[tokio::test] +async fn chunked_request_body_over_cap_is_rejected_413() { + // The upstream must actually consume the body; otherwise it can answer + // before the shim's cap trips and the race is unobservable. + let upstream = spawn(draining_upstream()).await; + let shim = spawn_shim(&upstream, Some("proj/cap"), Duration::from_secs(5)).await; + + // Unknown length (chunked): the cap trips mid-stream. + let half = MAX_REQUEST_BODY_BYTES as usize / 2; + let chunks: Vec> = vec![ + Ok(axum::body::Bytes::from(vec![b'x'; half])), + Ok(axum::body::Bytes::from(vec![b'x'; half])), + Ok(axum::body::Bytes::from(vec![b'x'; 1])), + ]; + let resp = reqwest::Client::new() + .post(format!("{shim}/mcp")) + .body(reqwest::Body::wrap_stream(futures_util::stream::iter( + chunks, + ))) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE); +} + +#[tokio::test] +async fn request_body_under_cap_passes_through() { + async fn echo_body(req: Request) -> String { + let bytes = axum::body::to_bytes(req.into_body(), usize::MAX) + .await + .unwrap(); + String::from_utf8(bytes.to_vec()).unwrap() + } + let upstream = spawn(axum::Router::new().route("/{*rest}", any(echo_body))).await; + let shim = spawn_shim(&upstream, Some("proj/ok"), Duration::from_secs(5)).await; + + let payload = "x".repeat(1024 * 1024); + let resp = reqwest::Client::new() + .post(format!("{shim}/mcp")) + .body(payload.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(resp.text().await.unwrap(), payload); +} + +#[tokio::test] +async fn upstream_hang_yields_504_gateway_timeout() { + // Upstream never answers within the shim's (short) header deadline. + let upstream = spawn(axum::Router::new().route( + "/{*rest}", + any(|| async { + tokio::time::sleep(Duration::from_secs(60)).await; + "too late" + }), + )) + .await; + let shim = spawn_shim(&upstream, Some("proj/slow"), Duration::from_millis(150)).await; + + let resp = reqwest::get(format!("{shim}/mcp")).await.unwrap(); + assert_eq!(resp.status(), StatusCode::GATEWAY_TIMEOUT); +} + +#[tokio::test] +async fn duplicated_mcp_path_is_collapsed() { + // Upstream only serves exactly `/mcp`; anything else is 404. A harness + // appending `/mcp` to the shim URL must still reach it. + async fn mcp(req: Request) -> StatusCode { + if req.uri().path() == "/mcp" { + StatusCode::OK + } else { + StatusCode::NOT_FOUND + } + } + let upstream = spawn( + axum::Router::new() + .route("/{*rest}", any(mcp)) + .route("/", any(mcp)), + ) + .await; + // Shim upstream base already ends in /mcp. + let shim = spawn_shim( + &format!("{upstream}/mcp"), + Some("proj/p"), + Duration::from_secs(5), + ) + .await; + + for path in ["/mcp", "/mcp/", ""] { + let resp = reqwest::get(format!("{shim}{path}")).await.unwrap(); + assert_eq!( + resp.status(), + StatusCode::OK, + "path {path:?} must reach upstream /mcp" + ); + } +} diff --git a/crates/aver-scope-shim/tests/scope_derivation.rs b/crates/aver-scope-shim/tests/scope_derivation.rs index e51b99e..a4f2639 100644 --- a/crates/aver-scope-shim/tests/scope_derivation.rs +++ b/crates/aver-scope-shim/tests/scope_derivation.rs @@ -32,6 +32,25 @@ fn cli_override_short_circuits() { assert_eq!(derived.source, ScopeSource::CliOverride); } +#[test] +fn empty_cli_override_falls_through_like_empty_env() { + let dir = tempfile::tempdir().unwrap(); + // An empty/whitespace `--scope` must be ignored, matching the env-var + // path — otherwise an empty `X-Aver-Scope` header would be injected. + for empty in ["", " ", "\t"] { + let derived = derive_scope(dir.path(), Some(empty), Some("proj/env")); + if scope_from_git(dir.path()).is_none() { + assert_eq!(derived.source, ScopeSource::EnvDefault, "input {empty:?}"); + assert_eq!(derived.scope, "proj/env"); + } + let derived = derive_scope(dir.path(), Some(empty), None); + if scope_from_git(dir.path()).is_none() { + assert_eq!(derived.source, ScopeSource::HardcodedGlobal); + assert_eq!(derived.scope, "global"); + } + } +} + #[test] fn git_origin_drives_scope_when_present() { let dir = tempfile::tempdir().unwrap();