diff --git a/src/adapters/forward.rs b/src/adapters/forward.rs index 0996a08..a8f397e 100644 --- a/src/adapters/forward.rs +++ b/src/adapters/forward.rs @@ -26,6 +26,26 @@ use url::Url; use crate::client::SharedHttpClient; +/// Maximum size, in bytes, of a buffered upstream response body on any +/// non-streaming read path in [`forward`] (JSON, text, or binary). A +/// hostile upstream cannot grow caller memory past this budget; a larger +/// body fails the read with [`BodyReadError::TooLarge`], surfaced as an +/// `HTTP_413` error envelope. +pub(crate) const RESPONSE_BODY_CAP: usize = 16 * 1024 * 1024; + +/// Maximum size, in bytes, of the error-body echo surfaced on a non-2xx +/// upstream response ([`forward`] and [`forward_stream`]). The echo is +/// bounded, not logged, and carries no request credential material — it +/// is the upstream's diagnostics (validation details, rate-limit info) +/// that would otherwise be discarded (FWD-10). +pub(crate) const ERROR_BODY_ECHO_CAP: usize = 4096; + +/// Upper bound on how much of a non-2xx upstream body is read, both for +/// the bounded echo and for connection reuse. A larger error body is +/// truncated in the surfaced message and the connection is dropped — +/// a deliberate trade of pool reuse against unbounded drain time (FWD-10). +const STATUS_BODY_DRAIN: usize = 64 * 1024; + #[derive(Clone)] pub enum HttpAuthScheme { Bearer, @@ -63,7 +83,7 @@ pub(crate) fn build_request( if key == "body" { body = Some(value.clone()); } else { - query_params.push((key.clone(), value_to_path_segment(value))); + query_params.push((key.clone(), value_to_query(value))); } } } @@ -182,26 +202,38 @@ const PATH_AFTER_PERCENT_ENCODE_SET: &AsciiSet = &CONTROLS .add(b'/') .add(b'\\'); -/// Percent-encoding scheme used by [`render_path_template`]: a value -/// containing `/`, `?`, `#`, or `\` — and any raw `%` it carries — is -/// encoded so that a rendered value stays one literal path segment -/// (FWD-01 traversal/ smuggled-segments gate). Path parameters are never -/// rejected: they are rendered safely instead. -/// Raw string form of a scalar input value, used for both path-template -/// rendering and query-pair emission (FWD-12: the two were byte-identical -/// helpers; they are one function now, and percent-encoding is applied by -/// the path renderer and `url::query_pairs_mut` respectively). -pub(crate) fn value_to_path_segment(value: &Value) -> String { - let raw = match value { +/// Raw string form of a scalar input value (FWD-12: `value_to_path_segment` +/// and `value_to_query` were byte-identical helpers; the shared scalar +/// extraction is this one function, and the encoding layers mount on it). +fn scalar_value_to_string(value: &Value) -> String { + match value { Value::String(s) => s.clone(), Value::Number(n) => n.to_string(), Value::Bool(b) => b.to_string(), Value::Null => String::new(), other => other.to_string(), - }; + } +} + +/// Percent-encoding scheme used by [`render_path_template`]: a value +/// containing `/`, `?`, `#`, or `\` — and any raw `%` it carries — is +/// encoded so that a rendered value stays one literal path segment +/// (FWD-01 traversal/ smuggled-segments gate). Path parameters are never +/// rejected: they are rendered safely instead. +pub(crate) fn value_to_path_segment(value: &Value) -> String { + let raw = scalar_value_to_string(value); utf8_percent_encode(&raw, PATH_VALUE_ENCODE_SET).to_string() } +/// Raw form of a scalar input value for query emission: `&`/`=` +/// separation, escaping, and space encoding are handled by +/// `url::query_pairs_mut` downstream, so the value itself must be raw — +/// a path-rendered value (`%20` for space) would be double-encoded into +/// `%2520` by `append_pair`. +fn value_to_query(value: &Value) -> String { + scalar_value_to_string(value) +} + /// Single-pass template renderer. Two invariants instead of the old /// iterative `replace` (FWD-01): /// @@ -756,26 +788,6 @@ pub(crate) enum SseParseError { BufferOverflow, } -/// Maximum size, in bytes, of a buffered upstream response body on any -/// non-streaming read path in [`forward`] (JSON, text, or binary). A -/// hostile upstream cannot grow caller memory past this budget; a larger -/// body fails the read with [`BodyReadError::TooLarge`], surfaced as an -/// `HTTP_413` error envelope. -pub(crate) const RESPONSE_BODY_CAP: usize = 16 * 1024 * 1024; - -/// Maximum size, in bytes, of the error-body echo surfaced on a non-2xx -/// upstream response ([`forward`] and [`forward_stream`]). The echo is -/// bounded, not logged, and carries no request credential material — it -/// is the upstream's diagnostics (validation details, rate-limit info) -/// that would otherwise be discarded (FWD-10). -pub(crate) const ERROR_BODY_ECHO_CAP: usize = 4096; - -/// Upper bound on how much of a non-2xx upstream body is read, both for -/// the bounded echo and for connection reuse. A larger error body is -/// truncated in the surfaced message and the connection is dropped — -/// a deliberate trade of pool reuse against unbounded drain time (FWD-10). -const STATUS_BODY_DRAIN: usize = 64 * 1024; - /// Maximum size, in bytes, of the SSE parser's internal reassembly /// buffer. A single event (all `data:` lines plus framing) must fit /// within this budget; a stream emitting a longer partial line — or an @@ -908,6 +920,15 @@ mod tests { use std::sync::Arc as TestArc; use std::time::Duration; + struct CapturedRequest { + #[allow(dead_code)] + method: String, + #[allow(dead_code)] + target: String, + #[allow(dead_code)] + headers: TestHashMap, + } + fn noop_context() -> OperationContext { struct NoopEnv; #[async_trait::async_trait] @@ -1094,4 +1115,436 @@ mod tests { .expect("request builds"); assert_eq!(url.query(), Some("lang=en&q=a%26b%3Dc+d")); } + + fn ctx_with_capability(namespace: &str, value: String) -> OperationContext { + let mut ctx = noop_context(); + ctx.capabilities = Capabilities::new().with_http_token(namespace, value); + ctx + } + + fn minimal_client() -> TestArc { + TestArc::new( + SharedHttpClient::new(crate::client::HttpClientConfig::default()) + .expect("client builds"), + ) + } + + type ServerResponder = Arc http::Response> + Send + Sync>; + + async fn spawn_responder(responder: ServerResponder) -> String { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local addr"); + tokio::spawn(async move { + loop { + let Ok((mut sock, _)) = listener.accept().await else { + break; + }; + let mut buf = vec![0u8; 8192]; + let mut n = 0; + loop { + let read = sock.read(&mut buf[n..]).await.unwrap_or(0); + if read == 0 { + break; + } + n += read; + if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") { + break; + } + } + let request = String::from_utf8_lossy(&buf[..n]); + let mut lines = request.lines(); + let request_line = lines.next().unwrap_or_default(); + let mut parts_iter = request_line.split_whitespace(); + let method = parts_iter.next().unwrap_or("GET").to_string(); + let target = parts_iter.next().unwrap_or("/").to_string(); + let mut headers = TestHashMap::new(); + for line in lines { + if line.is_empty() { + break; + } + if let Some((k, v)) = line.split_once(':') { + headers.insert(k.trim().to_lowercase(), v.trim().to_string()); + } + } + let response = (responder)(&CapturedRequest { + method, + target, + headers, + }); + let mut head = format!("HTTP/1.1 {}\r\n", response.status()); + for (name, value) in response.headers() { + head.push_str(&format!( + "{}: {}\r\n", + name, + value.to_str().unwrap_or_default() + )); + } + head.push_str(&format!( + "content-length: {}\r\n\r\n", + response.body().len() + )); + let _ = sock.write_all(head.as_bytes()).await; + let _ = sock.write_all(response.body()).await; + let _ = sock.flush().await; + let _ = sock.shutdown().await; + } + }); + format!("http://{addr}") + } + + async fn collect_stream(mut stream: ResponseStream) -> Vec { + let mut out = Vec::new(); + while let Some(envelope) = stream.next().await { + out.push(envelope); + } + out + } + + fn http_response(status: u16, content_type: &str, body: Vec) -> http::Response> { + let mut builder = http::Response::builder().status(status); + if !content_type.is_empty() { + builder = builder.header("content-type", content_type); + } + builder.body(body).expect("static response builds") + } + + async fn call_forward(base_url: &str, ctx: OperationContext) -> ResponseEnvelope { + call_forward_authed(base_url, ctx, &None).await + } + + async fn call_forward_authed( + base_url: &str, + ctx: OperationContext, + auth_scheme: &Option, + ) -> ResponseEnvelope { + forward( + &minimal_client(), + base_url, + "/x", + "GET", + auth_scheme, + &TestHashMap::new(), + "svc", + &[], + json!({}), + ctx, + ) + .await + } + + #[tokio::test] + async fn vendor_json_content_types_decode_as_json() { + for content_type in [ + "application/vnd.api+json", + "application/problem+json", + "application/hal+json; charset=utf-8", + "application/json", + "APPLICATION/JSON", + ] { + let base = spawn_responder(TestArc::new(move |_parts| { + http_response(200, content_type, br#"{"ok":true}"#.to_vec()) + })) + .await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Ok(Value::Object(map)) => assert_eq!(map["ok"], json!(true), "{content_type}"), + other => panic!("{content_type}: expected JSON object, got {other:?}"), + } + } + } + + #[tokio::test] + async fn oversized_json_body_trips_the_response_cap() { + let response = http_response(200, "application/json", vec![b'['; RESPONSE_BODY_CAP + 1]); + let base = spawn_responder(TestArc::new(move |_| response.clone())).await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Err(err) => { + assert_eq!(err.code, "HTTP_413"); + assert!(err.message.contains("response cap")); + } + other => panic!("expected cap error, got {other:?}"), + } + } + + #[tokio::test] + async fn oversized_text_body_trips_the_response_cap() { + let response = http_response(200, "text/plain", vec![b'a'; RESPONSE_BODY_CAP + 1]); + let base = spawn_responder(TestArc::new(move |_| response.clone())).await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Err(err) => assert_eq!(err.code, "HTTP_413"), + other => panic!("expected cap error, got {other:?}"), + } + } + + #[tokio::test] + async fn oversized_binary_body_trips_the_response_cap() { + let response = http_response( + 200, + "application/octet-stream", + vec![0u8; RESPONSE_BODY_CAP + 1], + ); + let base = spawn_responder(TestArc::new(move |_| response.clone())).await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Err(err) => assert_eq!(err.code, "HTTP_413"), + other => panic!("expected cap error, got {other:?}"), + } + } + + #[tokio::test] + async fn non_sse_content_type_on_a_sub_stream_errors_loudly() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response(200, "text/html", b"hello".to_vec()) + })) + .await; + let stream = forward_stream( + &minimal_client(), + &base, + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &[], + json!({}), + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!( + envelopes.len(), + 1, + "exactly one loud error, not an empty stream" + ); + match &envelopes[0].result { + Err(err) => { + assert_eq!(err.code, "INVALID_RESPONSE_TYPE"); + assert!(err.message.contains("text/html")); + assert!(err.message.contains("text/event-stream")); + } + other => panic!("expected content-type error, got {other:?}"), + } + } + + #[tokio::test] + async fn sse_content_type_still_streams() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response( + 200, + "text/event-stream; charset=utf-8", + b"data: {\"n\":1}\n\ndata: done\n\n".to_vec(), + ) + })) + .await; + let stream = forward_stream( + &minimal_client(), + &base, + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &[], + json!({}), + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!(envelopes.len(), 2); + assert!(envelopes[0].result.is_ok()); + assert!(envelopes[1].result.is_ok()); + } + + #[tokio::test] + async fn bearer_credential_with_control_character_fails_loudly() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response(200, "application/json", b"{}".to_vec()) + })) + .await; + let ctx = ctx_with_capability("svc", "tok\u{0007}en-secret-marker".to_string()); + let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Bearer)).await; + match envelope.result { + Err(err) => { + assert!(err + .message + .contains("refusing to send the request unauthenticated")); + assert!( + !err.message.contains("secret-marker"), + "error must not echo credential material" + ); + assert!( + !err.message.contains("tok\u{0007}en"), + "error must not echo credential material" + ); + } + other => panic!("expected loud credential error, got {other:?}"), + } + } + + #[tokio::test] + async fn api_key_with_invalid_header_name_fails_loudly() { + let base = "https://api.example.com".to_string(); + let ctx = ctx_with_capability("svc", "key-value".to_string()); + let mut ctx = ctx; + ctx.capabilities = Capabilities::new().with_http_token("svc", "key-value".to_string()); + let result = build_request( + &base, + "/x", + "GET", + &Some(HttpAuthScheme::ApiKey { + header_name: "bad header name".to_string(), + }), + &TestHashMap::new(), + "svc", + &json!({}), + &ctx, + ); + match result { + Ok(_) => panic!("invalid API-key header name must fail loudly"), + Err(err) => { + assert!(err.message.contains("invalid header name")); + assert!(!err.message.contains("key-value")); + } + } + } + + #[tokio::test] + async fn default_header_with_invalid_value_fails_loudly() { + let mut defaults = TestHashMap::new(); + defaults.insert("X-Trace".to_string(), "bad\u{0000}value".to_string()); + let err = build_request( + "https://api.example.com", + "/x", + "GET", + &None, + &defaults, + "svc", + &json!({}), + &noop_context(), + ) + .expect_err("invalid default-header value must fail loudly"); + assert!(err.message.contains("X-Trace")); + assert!(err.message.contains("invalid value")); + } + + #[tokio::test] + async fn default_header_with_invalid_name_fails_loudly() { + let mut defaults = TestHashMap::new(); + defaults.insert("bad header".to_string(), "v".to_string()); + let err = build_request( + "https://api.example.com", + "/x", + "GET", + &None, + &defaults, + "svc", + &json!({}), + &noop_context(), + ) + .expect_err("invalid default-header name must fail loudly"); + assert!(err.message.contains("bad header")); + } + + #[tokio::test] + async fn error_body_is_echoed_bounded_in_the_error_envelope() { + let body = vec![b'x'; ERROR_BODY_ECHO_CAP * 3]; + let base = spawn_responder(TestArc::new(move |_| { + http_response(429, "text/plain", body.clone()) + })) + .await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Err(err) => { + assert_eq!(err.code, "HTTP_429", "message was: {}", err.message); + assert!( + err.message.contains("HTTP 429: Too Many Requests"), + "message was: {}", + err.message + ); + assert!( + err.message.contains("[truncated]"), + "message was: {}", + err.message + ); + assert!( + err.message.len() < ERROR_BODY_ECHO_CAP * 2, + "echo must stay bounded, was {}", + err.message.len() + ); + } + other => panic!("expected HTTP_429 error, got {other:?}"), + } + } + + #[tokio::test] + async fn small_error_body_is_echoed_in_full() { + let base = spawn_responder(TestArc::new(|_| { + http_response(404, "text/plain", b"no such widget: id=42".to_vec()) + })) + .await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Err(err) => { + assert_eq!(err.code, "HTTP_404"); + assert!(err.message.contains("no such widget: id=42")); + } + other => panic!("expected HTTP_404, got {other:?}"), + } + } + + #[test] + fn json_detection_covers_vendor_suffix_and_rejects_lookalikes() { + for positive in [ + "application/json", + "application/json; charset=utf-8", + "application/vnd.api+json", + "application/problem+json", + "Application/Vnd.Api+JSON", + ] { + assert!(is_json_content_type(positive), "{positive} must be JSON"); + } + for negative in [ + "text/json", + "application/jsonx", + "xapplication/json", + "text/html", + ] { + assert!( + !is_json_content_type(negative), + "{negative} must not be JSON" + ); + } + } + + #[test] + fn sse_detection_requires_exact_essence() { + assert!(is_sse_content_type("text/event-stream")); + assert!(is_sse_content_type("text/event-stream; charset=utf-8")); + assert!(!is_sse_content_type("text/html")); + assert!(!is_sse_content_type("text/event-streamx")); + } + + #[tokio::test] + async fn oversized_non2xx_error_body_is_not_echoed_unbounded() { + let body = vec![b'e'; STATUS_BODY_DRAIN + 1]; + let base = spawn_responder(TestArc::new(move |_| { + http_response(500, "text/plain", body.clone()) + })) + .await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Err(err) => { + assert_eq!(err.code, "HTTP_500"); + assert!( + err.message.len() < STATUS_BODY_DRAIN, + "echo must stay far below the drain budget" + ); + assert!(err.message.contains("too large to echo")); + } + other => panic!("expected HTTP_500, got {other:?}"), + } + } }