test(adapters): cover FWD-07/08/10 response-decode paths (vendor JSON, caps, loud auth errors, bounded error echo, non-SSE Sub stream)

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
+/// Raw string form of a scalar input value (FWD-12: `value_to_path_segment`
-/// containing `/`, `?`, `#`, or `\` — and any raw `%` it carries — is
+/// and `value_to_query` were byte-identical helpers; the shared scalar
-/// encoded so that a rendered value stays one literal path segment
+/// extraction is this one function, and the encoding layers mount on it).
-/// (FWD-01 traversal/ smuggled-segments gate). Path parameters are never
+fn scalar_value_to_string(value: &Value) -> String {
-/// rejected: they are rendered safely instead.
+    match value {
-/// 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 {
         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<String, String>,
+    }
+
     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<SharedHttpClient> {
+        TestArc::new(
+            SharedHttpClient::new(crate::client::HttpClientConfig::default())
+                .expect("client builds"),
+        )
+    }
+
+    type ServerResponder = Arc<dyn Fn(&CapturedRequest) -> http::Response<Vec<u8>> + 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<ResponseEnvelope> {
+        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<u8>) -> http::Response<Vec<u8>> {
+        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<HttpAuthScheme>,
+    ) -> 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"<html>hello</html>".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:?}"),
+        }
+    }
 }