diff --git a/src/adapters/forward.rs b/src/adapters/forward.rs index 77afe76..0996a08 100644 --- a/src/adapters/forward.rs +++ b/src/adapters/forward.rs @@ -16,7 +16,6 @@ use std::sync::Arc; use alkcall::protocol::wire::{CallError, ResponseEnvelope}; use alkcall::registry::context::OperationContext; use alkcall::registry::registration::ResponseStream; -use alkcall::registry::spec::OperationType; use futures::stream; use futures::StreamExt; use percent_encoding::{utf8_percent_encode, AsciiSet, CONTROLS}; @@ -64,7 +63,7 @@ pub(crate) fn build_request( if key == "body" { body = Some(value.clone()); } else { - query_params.push((key.clone(), value_to_query(value))); + query_params.push((key.clone(), value_to_path_segment(value))); } } } @@ -80,12 +79,17 @@ pub(crate) fn build_request( let mut headers = HeaderMap::new(); for (k, v) in default_headers { - if let (Ok(name), Ok(value)) = ( - HeaderName::try_from(k.as_str()), - HeaderValue::try_from(v.as_str()), - ) { - headers.insert(name, value); - } + let name = HeaderName::try_from(k.as_str()).map_err(|_| { + CallError::internal(format!( + "default header `{k}` is not a valid HTTP header name; refusing to build a request that silently drops it" + )) + })?; + let value = HeaderValue::try_from(v.as_str()).map_err(|_| { + CallError::internal(format!( + "default header `{k}` has an invalid value (control or non-ASCII bytes are not permitted); refusing to build a request that silently drops it" + )) + })?; + headers.insert(name, value); } if body.is_some() { @@ -97,24 +101,36 @@ pub(crate) fn build_request( let credential = secret.expose_secret().clone(); match scheme { HttpAuthScheme::Bearer => { - let header_value = format!("Bearer {credential}"); - if let Ok(value) = HeaderValue::try_from(header_value) { - headers.insert(AUTHORIZATION, value); - } + let value = + HeaderValue::try_from(format!("Bearer {credential}")).map_err(|_| { + CallError::internal(format!( + "credential for namespace `{namespace}` contains invalid HTTP header characters (control or non-ASCII bytes); refusing to send the request unauthenticated" + )) + })?; + headers.insert(AUTHORIZATION, value); } HttpAuthScheme::ApiKey { header_name } => { - if let (Ok(name), Ok(value)) = ( - HeaderName::try_from(header_name.as_str()), - HeaderValue::try_from(credential.as_str()), - ) { - headers.insert(name, value); - } + let name = + HeaderName::try_from(header_name.as_str()).map_err(|_| { + CallError::internal(format!( + "API-key auth for namespace `{namespace}` declares invalid header name `{header_name}`; refusing to send the request unauthenticated" + )) + })?; + let value = HeaderValue::try_from(credential.as_str()).map_err(|_| { + CallError::internal(format!( + "credential for namespace `{namespace}` contains invalid HTTP header characters (control or non-ASCII bytes); refusing to send the request unauthenticated" + )) + })?; + headers.insert(name, value); } HttpAuthScheme::Basic => { - let header_value = format!("Basic {credential}"); - if let Ok(value) = HeaderValue::try_from(header_value) { - headers.insert(AUTHORIZATION, value); - } + let value = + HeaderValue::try_from(format!("Basic {credential}")).map_err(|_| { + CallError::internal(format!( + "credential for namespace `{namespace}` contains invalid HTTP header characters (control or non-ASCII bytes); refusing to send the request unauthenticated" + )) + })?; + headers.insert(AUTHORIZATION, value); } } } @@ -171,6 +187,10 @@ const PATH_AFTER_PERCENT_ENCODE_SET: &AsciiSet = &CONTROLS /// 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 { Value::String(s) => s.clone(), @@ -311,13 +331,133 @@ fn assemble_request_url(base_url: &str, rendered_path: &str) -> Result 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(), +#[derive(Debug, thiserror::Error)] +pub(crate) enum BodyReadError { + #[error("upstream response body exceeds the {RESPONSE_BODY_CAP}-byte response cap")] + TooLarge, + #[error("transport error reading response body: {0}")] + Transport(reqwest::Error), + #[error("malformed response body: {0}")] + Decode(serde_json::Error), +} + +/// True when a response `Content-Type` header is JSON by mime-essence +/// semantics: type `application`, subtype `json` or a `+json` structured +/// suffix (`application/vnd.api+json`, `application/problem+json`, …). +/// Parameters (`; charset=…`) are ignored. +pub(crate) fn is_json_content_type(content_type: &str) -> bool { + let essence = content_type + .split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + match essence.split_once('/') { + Some(("application", subtype)) => subtype == "json" || subtype.ends_with("+json"), + _ => false, + } +} + +/// Bounded string form of an upstream error body for the error envelope +/// (FWD-10): capped at [`ERROR_BODY_ECHO_CAP`] bytes, lossily decoded, +/// control characters (which could forge log or display framing) elided, +/// and truncated with a marker. The echo is never logged by this crate. +fn bounded_error_body(bytes: bytes::Bytes) -> Option { + if bytes.is_empty() { + return None; + } + let truncated = bytes.len() >= ERROR_BODY_ECHO_CAP; + let text = String::from_utf8_lossy(&bytes); + let printable: String = text + .chars() + .filter(|c| !c.is_control() || *c == '\n' || *c == '\t') + .take(ERROR_BODY_ECHO_CAP) + .collect(); + if printable.is_empty() { + None + } else if truncated { + Some(format!("{printable}\n[truncated]")) + } else { + Some(printable) + } +} + +/// Reads the response body, byte-capped: any read that would push the +/// accumulated bytes past `cap` returns [`BodyReadError::TooLarge`], so +/// a hostile upstream cannot grow caller memory past the cap (FWD-07). +async fn read_body_capped( + response: reqwest::Response, + cap: usize, +) -> Result { + let mut stream = response.bytes_stream(); + let mut buf: Vec = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(BodyReadError::Transport)?; + if buf.len().saturating_add(chunk.len()) > cap { + return Err(BodyReadError::TooLarge); + } + buf.extend_from_slice(&chunk); + } + Ok(buf.into()) +} + +/// Reads a `200`-class response into a [`ResponseEnvelope`], dispatching +/// on the mime essence of its `Content-Type` (FWD-07): `application/json` +/// and `application/*+json` decode as JSON, `text/*` as a string, and +/// everything else as a byte array — every path byte-capped at +/// [`RESPONSE_BODY_CAP`]. +async fn success_envelope(response: reqwest::Response, request_id: &str) -> ResponseEnvelope { + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok()) + .unwrap_or_default() + .to_ascii_lowercase(); + let essence = content_type.split(';').next().unwrap_or_default().trim(); + + let read = if is_json_content_type(&content_type) { + read_body_capped(response, RESPONSE_BODY_CAP) + .await + .and_then(|bytes| { + serde_json::from_slice::(&bytes) + .map(|v| ResponseEnvelope::ok(request_id, v)) + .map_err(BodyReadError::Decode) + }) + } else if essence.starts_with("text/") { + read_body_capped(response, RESPONSE_BODY_CAP) + .await + .map(|bytes| { + ResponseEnvelope::ok( + request_id, + Value::String(String::from_utf8_lossy(&bytes).into_owned()), + ) + }) + } else { + read_body_capped(response, RESPONSE_BODY_CAP) + .await + .map(|bytes| { + let arr: Vec = bytes + .iter() + .map(|byte| Value::Number((*byte).into())) + .collect(); + ResponseEnvelope::ok(request_id, Value::Array(arr)) + }) + }; + + match read { + Ok(envelope) => envelope, + Err(BodyReadError::TooLarge) => ResponseEnvelope::error( + request_id, + CallError::new( + "HTTP_413", + format!("upstream response body exceeds the {RESPONSE_BODY_CAP}-byte response cap"), + false, + ), + ), + Err(err) => ResponseEnvelope::error( + request_id, + CallError::internal(format!("failed to decode response body: {err}")), + ), } } @@ -331,7 +471,6 @@ pub(crate) async fn forward( default_headers: &HashMap, namespace: &str, error_status_codes: &[(u16, String)], - op_type: OperationType, input: Value, context: OperationContext, ) -> ResponseEnvelope { @@ -355,17 +494,20 @@ pub(crate) async fn forward( let request_builder = http_client .request(http_method, url.as_str()) - .headers(headers); - - let request_builder = if op_type == OperationType::Sub { - request_builder.header(ACCEPT, "text/event-stream") - } else { - request_builder.header(ACCEPT, "*/*") - }; + .headers(headers) + .header(ACCEPT, "*/*"); let request_builder = match body.as_ref() { Some(b) => { - let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null")); + let serialized = match serde_json::to_string(b) { + Ok(s) => s, + Err(err) => { + return ResponseEnvelope::error( + request_id, + CallError::internal(format!("failed to serialize request body: {err}")), + ); + } + }; request_builder.body(serialized) } None => request_builder, @@ -381,57 +523,49 @@ pub(crate) async fn forward( } }; + if !response.status().is_success() { + return error_envelope(response, &request_id, error_status_codes).await; + } + + success_envelope(response, &request_id).await +} + +/// Builds the non-2xx error envelope for an upstream response (FWD-10): +/// `HTTP_` mapping per ADR-023, plus a bounded, control-stripped +/// echo of the upstream error body woven into the message. The body is +/// read exactly once, capped at [`STATUS_BODY_DRAIN`] bytes — larger +/// error bodies are truncated for the echo and the connection is +/// dropped rather than drained further, so a firehose upstream cannot +/// pin a worker to connection cleanup. +async fn error_envelope( + response: reqwest::Response, + request_id: &str, + error_status_codes: &[(u16, String)], +) -> ResponseEnvelope { let status = response.status(); - - if !status.is_success() { - let code = error_status_codes - .iter() - .find(|(s, _)| *s == status.as_u16()) - .map(|(_, code)| code.clone()) - .unwrap_or_else(|| format!("HTTP_{}", status.as_u16())); - let message = format!( - "HTTP {}: {}", - status.as_u16(), - status.canonical_reason().unwrap_or("") - ); - return ResponseEnvelope::error(request_id, CallError::new(code, message, false)); - } - - let content_type = response - .headers() - .get(reqwest::header::CONTENT_TYPE) - .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok()) - .unwrap_or("") - .to_string(); - - if content_type.contains("application/json") { - match response.json::().await { - Ok(v) => ResponseEnvelope::ok(request_id, v), - Err(err) => ResponseEnvelope::error( - request_id, - CallError::internal(format!("failed to decode JSON body: {err}")), - ), - } - } else if content_type.starts_with("text/") { - match response.text().await { - Ok(t) => ResponseEnvelope::ok(request_id, Value::String(t)), - Err(err) => ResponseEnvelope::error( - request_id, - CallError::internal(format!("failed to decode text body: {err}")), - ), - } - } else { - match response.bytes().await { - Ok(b) => { - let arr: Vec = b.iter().map(|byte| Value::Number((*byte).into())).collect(); - ResponseEnvelope::ok(request_id, Value::Array(arr)) + let code = error_status_codes + .iter() + .find(|(s, _)| *s == status.as_u16()) + .map(|(_, c)| c.clone()) + .unwrap_or_else(|| format!("HTTP_{}", status.as_u16())); + let mut message = format!( + "HTTP {}: {}", + status.as_u16(), + status.canonical_reason().unwrap_or("") + ); + match read_body_capped(response, STATUS_BODY_DRAIN).await { + Ok(bytes) => { + if let Some(echo) = bounded_error_body(bytes) { + message.push_str(": "); + message.push_str(&echo); } - Err(err) => ResponseEnvelope::error( - request_id, - CallError::internal(format!("failed to read body: {err}")), - ), } + Err(BodyReadError::TooLarge) => { + message.push_str(": [error body too large to echo]"); + } + Err(_) => {} } + ResponseEnvelope::error(request_id, CallError::new(code, message, false)) } /// Converts a parsed SSE event into a response envelope, JSON-decoding @@ -491,13 +625,20 @@ pub(crate) fn forward_stream( .headers(headers) .header(ACCEPT, "text/event-stream"); let request_builder = match body.as_ref() { - Some(b) => { - let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null")); - request_builder.body(serialized) - } + Some(b) => match serde_json::to_string(b) { + Ok(serialized) => request_builder.body(serialized), + Err(err) => { + return Err(CallError::internal(format!( + "failed to serialize request body: {err}" + ))); + } + }, None => request_builder, }; - request_builder.send().await + request_builder + .send() + .await + .map_err(|err| CallError::internal(format!("HTTP request failed: {err}"))) }; let sse = stream::once(init).flat_map(move |result| { @@ -505,28 +646,33 @@ pub(crate) fn forward_stream( let error_status_codes = error_status_codes_stream.clone(); match result { Err(err) => Box::pin(stream::once(async move { - ResponseEnvelope::error( - request_id, - CallError::internal(format!("HTTP request failed: {err}")), - ) + ResponseEnvelope::error(request_id, err) })) as ResponseStream, Ok(response) => { let status = response.status(); if !status.is_success() { - let code = error_status_codes - .iter() - .find(|(s, _)| *s == status.as_u16()) - .map(|(_, c)| c.clone()) - .unwrap_or_else(|| format!("HTTP_{}", status.as_u16())); - let message = format!( - "HTTP {}: {}", - status.as_u16(), - status.canonical_reason().unwrap_or("") - ); + let request_id = request_id.clone(); Box::pin(stream::once(async move { - ResponseEnvelope::error(request_id, CallError::new(code, message, false)) + error_envelope(response, &request_id, &error_status_codes).await })) as ResponseStream } else { + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok()) + .unwrap_or_default() + .to_ascii_lowercase(); + if !is_sse_content_type(&content_type) { + let message = format!( + "upstream returned Content-Type `{content_type}` on a subscription operation; expected `text/event-stream`" + ); + return Box::pin(stream::once(async move { + ResponseEnvelope::error( + request_id, + CallError::new("INVALID_RESPONSE_TYPE", message, false), + ) + })) as ResponseStream; + } let request_id_inner = request_id.clone(); Box::pin( stream::unfold( @@ -591,6 +737,14 @@ pub(crate) fn forward_stream( Box::pin(sse) } +/// True when a response `Content-Type` header is `text/event-stream` by +/// mime-essence semantics; parameters (`; charset=…`) are ignored. A +/// subscription forwarder that receives anything else surfaces a loud +/// error rather than an indefinitely empty stream (FWD-07). +fn is_sse_content_type(content_type: &str) -> bool { + content_type.split(';').next().unwrap_or_default().trim() == "text/event-stream" +} + /// A parsed SSE event: the `data:` lines joined with `\n`. pub(crate) struct SseEvent { pub(crate) data: String, @@ -602,6 +756,26 @@ 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 diff --git a/src/adapters/from_jsonschema.rs b/src/adapters/from_jsonschema.rs index 13abfb8..6695ddd 100644 --- a/src/adapters/from_jsonschema.rs +++ b/src/adapters/from_jsonschema.rs @@ -105,7 +105,6 @@ impl OperationAdapter for FromJsonSchema { let namespace = namespace.clone(); let http_client = Arc::clone(&http_client); let error_status_codes = error_status_codes.clone(); - let op_type = op_type; async move { forward( &http_client, @@ -116,7 +115,6 @@ impl OperationAdapter for FromJsonSchema { &default_headers, &namespace, &error_status_codes, - op_type, input, context, ) diff --git a/src/adapters/from_openapi.rs b/src/adapters/from_openapi.rs index d069650..5fdf574 100644 --- a/src/adapters/from_openapi.rs +++ b/src/adapters/from_openapi.rs @@ -302,7 +302,6 @@ impl FromOpenAPI { let namespace = namespace.clone(); let http_client = Arc::clone(&http_client); let error_status_codes = error_status_codes.clone(); - let op_type = op_type; async move { forward( &http_client, @@ -313,7 +312,6 @@ impl FromOpenAPI { &default_headers, &namespace, &error_status_codes, - op_type, input, context, ) diff --git a/src/client/http_client.rs b/src/client/http_client.rs index a49973c..7295f49 100644 --- a/src/client/http_client.rs +++ b/src/client/http_client.rs @@ -175,14 +175,22 @@ pub enum HttpClientBuildError { } pub struct SharedHttpClient { - inner: ArcSwap, - config: ArcSwap, + inner: ArcSwap, +} + +/// Joint holder for the client and its config so a reload swaps both in +/// one atomic `ArcSwap::store` — a reader can never observe the new +/// config paired with the previous client (FWD-12). +#[derive(Clone)] +struct SharedHttpInner { + client: Arc, + config: Arc, } impl std::fmt::Debug for SharedHttpClient { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("SharedHttpClient") - .field("config", &self.config.load()) + .field("config", &self.inner.load().config) .finish_non_exhaustive() } } @@ -196,27 +204,32 @@ impl SharedHttpClient { pub fn new(config: HttpClientConfig) -> Result { let client = build_client_sync(&config)?; Ok(Self { - inner: ArcSwap::from_pointee(client), - config: ArcSwap::from_pointee(config), + inner: ArcSwap::from_pointee(SharedHttpInner { + client: Arc::new(client), + config: Arc::new(config), + }), }) } pub fn client(&self) -> Arc { - self.inner.load_full() + Arc::clone(&self.inner.load().client) } pub fn config(&self) -> Arc { - self.config.load_full() + Arc::clone(&self.inner.load().config) } /// Rebuild the underlying client and swap it in for new callers /// (in-flight requests complete on the previous client). PEM reads /// use `tokio::fs`, so this is safe to call from async contexts - /// without blocking a worker. + /// without blocking a worker. Client and config swap together in a + /// single atomic store (FWD-12). pub async fn reload(&self, config: HttpClientConfig) -> Result<(), HttpClientBuildError> { let client = build_client(&config).await?; - self.config.store(Arc::new(config)); - self.inner.store(Arc::new(client)); + self.inner.store(Arc::new(SharedHttpInner { + client: Arc::new(client), + config: Arc::new(config), + })); Ok(()) } }