glm-5.3-flash
0837e942ab
- new tests/retry_policy_wire.rs: POST+500 = 1 upstream hit (method gate), GET 500/503/200 = 3 hits (retry-to-success), budget exhaustion stops retries under a 50-attempt cap (wall-time + hit-count bounds) - forward_stream terminal arms on the wire: oversized SSE line -> one INTERNAL terminal envelope; mid-stream transport abort (staged via a notify gate so the abort is genuinely mid-stream) -> terminal envelope after the delivered frame; pending event EOF-flush; dead port through both forward() and forward_stream() - Once-path decode arms: malformed application/json 200 -> INTERNAL decode envelope; application/octet-stream 200 -> byte-array envelope - COV-12 credential arms: ApiKey and Basic malformed values fail loudly without echoing secret material; declared header-param invalid name/value rejections - new tests/client_config_reload.rs: config() reflects a reloaded config (FWD-12 atomicity half)
3341 lines
129 KiB
Rust
3341 lines
129 KiB
Rust
//! Shared HTTP forwarding core for the HTTP-backed adapters
|
|
//! (`from_openapi`, `from_jsonschema`): request construction
|
|
//! (path templates, query split, body, credential injection), the
|
|
//! reqwest forwarding handlers (`forward` / `forward_stream`), and the
|
|
//! SSE frame parser for streaming (SSE → `ResponseEnvelope`) handlers
|
|
//! (ADR-049).
|
|
//!
|
|
//! # Timeout / byte-bound policy for streaming forwards (FWD-15, FWD-14)
|
|
//!
|
|
//! The two send paths impose deliberately different bounds, mirroring
|
|
//! the gateway's dispatch contract (alkcall ADR-021: Once-op invokes
|
|
//! carry a 30 s deadline; `HandlerKind::Stream` invokes set
|
|
//! `deadline: None` — subscriptions are unbounded in *time* by design):
|
|
//!
|
|
//! - **`forward`** (request/response) sends through the shared client's
|
|
//! total request timeout (30 s default), so a forwarded call fails
|
|
//! before its caller does.
|
|
//! - **`forward_stream`** (subscriptions) sends through
|
|
//! [`SharedHttpClient::stream_client`] — the same config minus the
|
|
//! total request timeout, connect + read timeouts retained. A healthy
|
|
//! subscription longer than 30 s survives; the read timeout remains
|
|
//! the stall guard on upstream staleness (it resets on every body
|
|
//! byte, so a keep-alive-emitting source lives as long as it keeps
|
|
//! the connection warm — and quiet silence past the read timeout
|
|
//! still terminates it). reqwest 0.13's per-request timeout override
|
|
//! can lengthen but never clear a client-level total timeout, hence
|
|
//! the derived client rather than an extension override.
|
|
//!
|
|
//! Unbounded time without a byte bound would let a hostile upstream
|
|
//! stream well-formed 1-MiB-line events forever (the 1 MiB SSE line cap
|
|
//! bounds one line, not the stream), so the streaming branch also
|
|
//! enforces a total streamed-bytes cap per subscription —
|
|
//! [`crate::client::HttpClientConfig::stream_total_byte_cap`], 1 GiB by
|
|
//! default, accumulated across every chunk fed to the SSE parser.
|
|
//! Exceeding it terminates the stream with a single terminal error
|
|
//! envelope (the stream-ends semantics: one error frame, then end —
|
|
//! matching the other terminal arms). The line-cap check runs before
|
|
//! the buffer grows, so the reassembly buffer can never exceed the
|
|
//! line cap.
|
|
//!
|
|
//! # Streaming payload contract (FWD-17)
|
|
//!
|
|
//! Each parsed SSE frame becomes one success envelope:
|
|
//!
|
|
//! - A `data:` payload that is valid JSON surfaces as the decoded value
|
|
//! itself — `123` as a number, `"123"` as a string (the raw-text
|
|
//! fallback this replaces erased that distinction). This holds even
|
|
//! when the frame carries an `event:` name.
|
|
//! - Any other payload surfaces as the
|
|
//! `{"data": <raw payload>, "event": <name|null>}` wrapper: the raw
|
|
//! text stays field-addressable and an upstream's named-event
|
|
//! conventions (`event: error`, `event: ping`, …) are visible on
|
|
//! non-JSON frames. The name is the frame's `event:` field (WHATWG
|
|
//! last-wins); `null` when the upstream sent none — the spec's
|
|
//! implicit `message` default is *not* substituted, so "upstream
|
|
//! named it" and "default name" remain distinguishable.
|
|
//! - An empty payload is `Null`.
|
|
//!
|
|
//! Known limitation: a JSON body under a named event surfaces as the
|
|
//! decoded value only — the event name is not carried on JSON frames;
|
|
//! a non-JSON body under a named event is the only combination where
|
|
//! both are visible.
|
|
//!
|
|
//! # Credentials and input routing
|
|
//!
|
|
//! The forwarding handler is the no-env-vars credential injection point
|
|
//! (ADR-014): it reads `OperationContext.capabilities`, never
|
|
//! `std::env::var`. Imported error codes are `HTTP_<status>` to avoid
|
|
//! collision with the protocol-level codes (ADR-023).
|
|
//!
|
|
//! The loud-missing credential matrix: when an operation declares an
|
|
//! auth scheme, a request is sent only when the capability is present
|
|
//! AND well-formed. A malformed credential name/value fails loudly
|
|
//! (FWD-08), and an absent capability — the registry holds neither
|
|
//! `api_key:{namespace}` nor `http_token:{namespace}` — fails loudly too
|
|
//! (FWD-16): the request is refused with an `INTERNAL` error naming the
|
|
//! missing key rather than sent unauthenticated to produce corrupted
|
|
//! upstream 401s with no local diagnostic.
|
|
//!
|
|
//! Input-schema enforcement (review 001 OAI-02): every input key must be
|
|
//! declared by the operation's `input_schema` or consumed by a path
|
|
//! template placeholder; undeclared keys are rejected with `INVALID_INPUT`
|
|
//! at call time. There is no pass-through knob — a request the forwarder
|
|
//! sends must match what `/schema` advertises, so peer-supplied input
|
|
//! cannot add upstream query parameters, headers, or craft a request body
|
|
//! the contract does not declare. Declared keys route by designation: a
|
|
//! property marked with the `HEADER_PARAM_IN_MARKER` marker key set to "header" is sent as a
|
|
//! request header, the declared `GATEWAY_BODY_KEY` property becomes the
|
|
//! request body, and every other declared key becomes an upstream query
|
|
//! parameter.
|
|
//!
|
|
//! # Input routing and path placeholders (FWD-18)
|
|
//!
|
|
//! A key that matches a `{placeholder}` in the path template is consumed
|
|
//! by the path and never also emits as a query parameter, regardless of
|
|
//! its value's shape — the placeholder check precedes query routing in
|
|
//! the request builder. A placeholder renders exactly one literal path
|
|
//! segment, so its value must be a scalar: object/array values fail with
|
|
//! `INVALID_INPUT` (structural values have no faithful single-segment
|
|
//! rendering; splicing the minified JSON into the path was the
|
|
//! pre-decision behavior and is rejected now).
|
|
//!
|
|
//! # Percent handling in the rendered path (FWD-19)
|
|
//!
|
|
//! Two different contracts apply, by design:
|
|
//!
|
|
//! - A `%` arriving inside a *value* is always encoded (`%` → `%25`),
|
|
//! so values carrying `%2F` cannot be mistaken for this crate's own
|
|
//! escapes and a value can never inject URL structure.
|
|
//! - A `%` in *template/base text* survives verbatim. Templates and
|
|
//! base URLs are assembly-supplied (ADR-066 trust boundary), so an
|
|
//! assembly that writes `/s3%2Fkeys` is presumed to mean a
|
|
//! pre-encoded segment for upstreams that route `%2F` differently
|
|
//! from `/` — that upstream-semantics choice belongs to the assembly,
|
|
//! not this crate. No injection results: the surviving `%2F` still
|
|
//! forms a single segment (the origin check plus the two-pass
|
|
//! percent-encoding over template text see to that), and the
|
|
//! value-side rule above means every bare `%` in a rendered path
|
|
//! traces to template text the assembly author wrote.
|
|
//!
|
|
|
|
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
|
|
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
|
|
use alkcall::registry::context::OperationContext;
|
|
use alkcall::registry::registration::ResponseStream;
|
|
use futures::stream;
|
|
use futures::StreamExt;
|
|
use percent_encoding::{percent_decode_str, utf8_percent_encode, AsciiSet, CONTROLS};
|
|
use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE};
|
|
use reqwest::Method;
|
|
use serde_json::Value;
|
|
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;
|
|
|
|
// Fractional-path helpers and body echo caps follow; the gateway body key
|
|
// and the header-param wire marker used by the input routing live with the
|
|
// other request-construction defaults here.
|
|
|
|
/// 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;
|
|
|
|
/// The input key that carries the request body on the gateway shape
|
|
/// (ADR-047). Declared by `from_openapi`'s generated input schemas
|
|
/// (OAI-07 rejects a spec parameter with the same name at import) and
|
|
/// consumed by [`build_request`] ahead of the query-parameter routing.
|
|
pub(crate) const GATEWAY_BODY_KEY: &str = "body";
|
|
|
|
/// Property-level marker inside an operation's `input_schema` properties
|
|
/// that routes the declared value into an HTTP request header instead of
|
|
/// the query string (review 001 OAI-03). `from_openapi` stamps
|
|
/// `"wire": "header"` on properties generated from `in: header`
|
|
/// parameters; `from_jsonschema` callers can mark their declared
|
|
/// properties the same way. Unmarked properties default to query
|
|
/// placement, preserving the pre-OAI-03 wire behavior for
|
|
/// `in: query`/`in: path` parameters.
|
|
pub(crate) const HEADER_PARAM_IN_MARKER: &str = "wire";
|
|
pub(crate) const HEADER_PARAM_MARKER_VALUE: &str = "header";
|
|
|
|
/// The credential scheme forwarded handlers apply to outbound requests
|
|
/// (ADR-014). The credential value itself flows through
|
|
/// `OperationContext.capabilities` at call time — never through this
|
|
/// config.
|
|
#[derive(Clone)]
|
|
pub enum HttpAuthScheme {
|
|
/// `Authorization: Bearer <token>` from the caller's `Bearer`
|
|
/// capability.
|
|
Bearer,
|
|
/// A named API-key header (e.g. `x-api-key`) carrying the caller's
|
|
/// `ApiKey` capability value.
|
|
ApiKey {
|
|
/// The upstream header the key is sent in.
|
|
header_name: String,
|
|
},
|
|
/// HTTP Basic auth from the caller's `username`/`password`
|
|
/// capability pair.
|
|
Basic,
|
|
}
|
|
|
|
/// Assembly-time configuration for one imported HTTP service: the
|
|
/// registry namespace its operations land under, where traffic goes,
|
|
/// and how credentials attach.
|
|
pub struct HttpServiceConfig {
|
|
/// Registry namespace for the imported operations (the
|
|
/// `<namespace>/<operationId>` op names).
|
|
pub namespace: String,
|
|
/// Outbound base URL every path template is resolved against.
|
|
pub base_url: String,
|
|
/// Credential scheme for outbound requests; `None` sends an
|
|
/// unauthenticated request.
|
|
pub auth: Option<HttpAuthScheme>,
|
|
/// Static headers attached to every outbound request (e.g. a
|
|
/// required `User-Agent`).
|
|
pub default_headers: HashMap<String, String>,
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(crate) fn build_request(
|
|
base_url: &str,
|
|
path_template: &str,
|
|
method: &str,
|
|
auth_scheme: &Option<HttpAuthScheme>,
|
|
default_headers: &HashMap<String, String>,
|
|
namespace: &str,
|
|
input_schema: &Value,
|
|
input: &Value,
|
|
context: &OperationContext,
|
|
) -> Result<(Method, Url, Option<Value>, HeaderMap), CallError> {
|
|
let inputs = input.as_object().ok_or_else(|| {
|
|
CallError::invalid_input(format!(
|
|
"input must be a JSON object (got {}); adapter operations take a named-key input",
|
|
type_name_of(input)
|
|
))
|
|
})?;
|
|
|
|
enforce_input_schema(input_schema, inputs)?;
|
|
|
|
let mut query_params: Vec<(String, String)> = Vec::new();
|
|
let mut header_params: Vec<(String, String)> = Vec::new();
|
|
let param_locations = param_locations(input_schema);
|
|
let mut body: Option<Value> = None;
|
|
for (key, value) in inputs {
|
|
if is_path_placeholder(key, path_template) {
|
|
continue;
|
|
}
|
|
if key == GATEWAY_BODY_KEY {
|
|
body = Some(value.clone());
|
|
continue;
|
|
}
|
|
if param_locations.get(key.as_str()) == Some(&ParamLocation::Header) {
|
|
header_params.push((key.clone(), value_to_query(value)));
|
|
} else {
|
|
query_params.push((key.clone(), value_to_query(value)));
|
|
}
|
|
}
|
|
|
|
let rendered_path = render_path_template(path_template, Some(inputs))?;
|
|
let mut url = assemble_request_url(base_url, &rendered_path)?;
|
|
if !query_params.is_empty() {
|
|
let mut pairs = url.query_pairs_mut();
|
|
for (k, v) in &query_params {
|
|
pairs.append_pair(k, v);
|
|
}
|
|
}
|
|
|
|
let mut headers = HeaderMap::new();
|
|
for (k, v) in &header_params {
|
|
let name = HeaderName::try_from(k.as_str()).map_err(|_| {
|
|
CallError::internal(format!(
|
|
"declared header parameter `{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!(
|
|
"declared header parameter `{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);
|
|
}
|
|
for (k, v) in default_headers {
|
|
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() {
|
|
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
|
|
}
|
|
|
|
if let Some(scheme) = auth_scheme {
|
|
let secret = context.capabilities.get(namespace).ok_or_else(|| {
|
|
CallError::internal(format!(
|
|
"capability for namespace `{namespace}` is absent (the registry holds neither `api_key:{namespace}` nor `http_token:{namespace}`); refusing to send the request unauthenticated"
|
|
))
|
|
})?;
|
|
let credential = secret.expose_secret().clone();
|
|
match scheme {
|
|
HttpAuthScheme::Bearer => {
|
|
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 } => {
|
|
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 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);
|
|
}
|
|
}
|
|
}
|
|
|
|
let http_method = Method::from_bytes(method.as_bytes())
|
|
.map_err(|_| CallError::internal(format!("invalid HTTP method `{method}`")))?;
|
|
Ok((http_method, url, body, headers))
|
|
}
|
|
|
|
/// Upstream placement of a declared input property (review 001 OAI-03).
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
enum ParamLocation {
|
|
Query,
|
|
Header,
|
|
}
|
|
|
|
/// Map of declared property name → upstream placement, read from the
|
|
/// `input_schema`'s `properties` entries via [`HEADER_PARAM_IN_MARKER`]
|
|
/// (query placement is the default).
|
|
fn param_locations(input_schema: &Value) -> HashMap<&str, ParamLocation> {
|
|
let mut out = HashMap::new();
|
|
let Some(properties) = input_schema.get("properties").and_then(|p| p.as_object()) else {
|
|
return out;
|
|
};
|
|
for (name, schema) in properties {
|
|
let is_header = schema
|
|
.get(HEADER_PARAM_IN_MARKER)
|
|
.and_then(|w| w.as_str())
|
|
.is_some_and(|w| w == HEADER_PARAM_MARKER_VALUE);
|
|
out.insert(
|
|
name.as_str(),
|
|
if is_header {
|
|
ParamLocation::Header
|
|
} else {
|
|
ParamLocation::Query
|
|
},
|
|
);
|
|
}
|
|
out
|
|
}
|
|
|
|
fn type_name_of(value: &Value) -> &'static str {
|
|
match value {
|
|
Value::Null => "null",
|
|
Value::Bool(_) => "a boolean",
|
|
Value::Number(_) => "a number",
|
|
Value::String(_) => "a string",
|
|
Value::Array(_) => "an array",
|
|
Value::Object(_) => "an object",
|
|
}
|
|
}
|
|
|
|
/// Input-schema enforcement at call time (review 001 OAI-02).
|
|
///
|
|
/// Every input key must be declared by the input schema's `properties` —
|
|
/// including the gateway body property, which `from_openapi` declares as
|
|
/// `body` whenever the operation has a requestBody, and including the
|
|
/// path-consumed placeholders, which `from_openapi` also declares. An
|
|
/// explicit `"additionalProperties": true` opts the operation into
|
|
/// catch-all input (JSON Schema semantics: the schema advertises that
|
|
/// extra properties are valid), which `from_jsonschema` callers can use
|
|
/// for open-shaped endpoints. Anything else undeclared is a rejected
|
|
/// `INVALID_INPUT` rather than a silently-added upstream query parameter,
|
|
/// so peer input like `{debug: true}` or `{impersonate_id: …}` cannot
|
|
/// decorate an upstream request the contract does not advertise.
|
|
fn enforce_input_schema(
|
|
input_schema: &Value,
|
|
inputs: &serde_json::Map<String, Value>,
|
|
) -> Result<(), CallError> {
|
|
let declared = input_schema
|
|
.get("properties")
|
|
.and_then(|p| p.as_object())
|
|
.map(|p| p.keys().cloned().collect::<Vec<_>>())
|
|
.unwrap_or_default();
|
|
let catch_all = input_schema.get("additionalProperties") == Some(&Value::Bool(true));
|
|
let unknown: Vec<String> = inputs
|
|
.keys()
|
|
.filter(|key| !catch_all && !declared.iter().any(|d| d == *key))
|
|
.cloned()
|
|
.collect();
|
|
if let Some(first) = unknown.first() {
|
|
let declared_list = if declared.is_empty() {
|
|
"none".to_string()
|
|
} else {
|
|
declared.join(", ")
|
|
};
|
|
return Err(CallError::invalid_input(format!(
|
|
"input key `{first}` is not declared by the operation's input schema \
|
|
(declared: {declared_list}); undeclared keys are rejected so peer \
|
|
input cannot shape the upstream request beyond the advertised contract"
|
|
)));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Percent-encode set for a spliced path-parameter value: the WHATWG
|
|
/// path set (controls, space, `"`, `<`, `>`, `` ` ``, `#`, `?`, `{`, `}`)
|
|
/// plus `/` so a value stays one literal segment, plus `%` `?` `#`
|
|
/// belt-and-braces, plus `\` so a Windows-style separator cannot smuggle
|
|
/// a backslash segment on special-scheme URLs.
|
|
///
|
|
/// Because `%` is in this set, percent-encoding with it is idempotent:
|
|
/// every `%` in a value becomes `%25`, so the result contains `%` only
|
|
/// as the lead byte of an escape this crate itself produced.
|
|
const PATH_VALUE_ENCODE_SET: &AsciiSet = &CONTROLS
|
|
.add(b' ')
|
|
.add(b'"')
|
|
.add(b'<')
|
|
.add(b'>')
|
|
.add(b'`')
|
|
.add(b'#')
|
|
.add(b'?')
|
|
.add(b'{')
|
|
.add(b'}')
|
|
.add(b'/')
|
|
.add(b'%')
|
|
.add(b'\\');
|
|
|
|
/// Encode set for the second pass over a rendered segment in
|
|
/// [`request_path`]: applied to the text *between* `%` characters, so
|
|
/// anything that still looks like a separator is encoded, while the `%`
|
|
/// itself is left untouched to preserve this crate's own `%2F`-style
|
|
/// escapes.
|
|
const PATH_AFTER_PERCENT_ENCODE_SET: &AsciiSet = &CONTROLS
|
|
.add(b' ')
|
|
.add(b'"')
|
|
.add(b'<')
|
|
.add(b'>')
|
|
.add(b'`')
|
|
.add(b'#')
|
|
.add(b'?')
|
|
.add(b'{')
|
|
.add(b'}')
|
|
.add(b'/')
|
|
.add(b'\\');
|
|
|
|
/// 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). Scalars are never
|
|
/// rejected: they are rendered safely instead — with the lone-dot
|
|
/// exception below (FWD-13). Object/array values under a placeholder
|
|
/// key are rejected with `INVALID_INPUT` (FWD-18): a path placeholder
|
|
/// designates exactly one literal segment, so a structural value has no
|
|
/// faithful rendering — the pre-decision behavior spliced the minified
|
|
/// JSON into the path *and* emitted the key as a query parameter
|
|
/// (double-routed, neither faithful).
|
|
///
|
|
/// # Lone-dot values (FWD-13)
|
|
///
|
|
/// A scalar whose *decoded* form is exactly `.` or `..` — including the
|
|
/// `%2e`/`%2E` spellings, which `Url::set_path` normalizes away after
|
|
/// decoding regardless of what the encode set preserves — is rejected
|
|
/// with `INVALID_INPUT`. `Url::set_path` removes lone dot segments
|
|
/// case-insensitively (`/../x` → `/x`, `/./x` → `/x`, `/%2e%2e/x` →
|
|
/// `/x`), so no encode-set fix can keep such a value faithful: the
|
|
/// upstream would receive a *different* endpoint than the template
|
|
/// describes, with the namespace's credentials attached. The rejection
|
|
/// is exact-match on the full decoded segment; values like `v1.2.3` or
|
|
/// `.hidden-file` render normally. The parameter name is quoted but the
|
|
/// value is never echoed.
|
|
pub(crate) fn value_to_path_segment(value: &Value) -> Result<String, CallError> {
|
|
let raw = match value {
|
|
Value::String(_) | Value::Number(_) | Value::Bool(_) | Value::Null => {
|
|
scalar_value_to_string(value)
|
|
}
|
|
other => {
|
|
return Err(CallError::invalid_input(format!(
|
|
"path placeholder value must be a scalar (string, number, boolean, or null); got {}: structural values have no faithful single-segment rendering",
|
|
type_name_of(other)
|
|
)))
|
|
}
|
|
};
|
|
reject_lone_dot_value(value, &raw)?;
|
|
Ok(utf8_percent_encode(&raw, PATH_VALUE_ENCODE_SET).to_string())
|
|
}
|
|
|
|
/// FWD-13 gate: a decoded path-value of exactly `.` or `..` cannot be
|
|
/// rendered faithfully — `Url::set_path` silently normalizes lone dot
|
|
/// segments away, so the upstream would receive a different endpoint
|
|
/// than the template describes. The parameter is named but the value is
|
|
/// not echoed. Called with both the original [`Value`] (for type-naming)
|
|
/// and its raw scalar string.
|
|
fn reject_lone_dot_value(value: &Value, raw: &str) -> Result<(), CallError> {
|
|
let type_label = type_name_of(value);
|
|
let is_lone_dot = matches!(raw, "." | "..")
|
|
|| raw.eq_ignore_ascii_case("%2e")
|
|
|| raw.eq_ignore_ascii_case("%2e%2e");
|
|
if is_lone_dot {
|
|
return Err(CallError::invalid_input(format!(
|
|
"path placeholder value of type {type_label} decodes to a lone dot segment, which cannot appear in a rendered path (Url::set_path would silently normalize it away); pass a concrete non-dot value"
|
|
)));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// 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):
|
|
///
|
|
/// 1. A rendered value is never re-scanned — each template placeholder
|
|
/// is replaced exactly once, so a `{repo}` value that itself contains
|
|
/// a later placeholder string cannot trigger a second substitution.
|
|
/// 2. The whole template and every input key are walked in one pass, so
|
|
/// `?` and `#` in a value split into percent-encoded octets (`%3F`,
|
|
/// `%23`) rather than injecting URL structure.
|
|
pub(crate) fn render_path_template(
|
|
template: &str,
|
|
inputs: Option<&serde_json::Map<String, Value>>,
|
|
) -> Result<String, CallError> {
|
|
let mut out = String::with_capacity(template.len());
|
|
let mut rest = template;
|
|
let mut unresolved: Vec<String> = Vec::new();
|
|
while let Some(start) = rest.find('{') {
|
|
let (head, tail) = rest.split_at(start);
|
|
out.push_str(head);
|
|
let Some(end) = tail.find('}') else {
|
|
return Err(CallError::internal(format!(
|
|
"invalid path template `{template}`: unterminated placeholder"
|
|
)));
|
|
};
|
|
let name = &tail[1..end];
|
|
let raw_value = match inputs.and_then(|map| map.get(name)) {
|
|
Some(v) => v,
|
|
None => {
|
|
unresolved.push(format!("{{{name}}}"));
|
|
rest = &tail[end + 1..];
|
|
continue;
|
|
}
|
|
};
|
|
out.push_str(&value_to_path_segment(raw_value)?);
|
|
rest = &tail[end + 1..];
|
|
}
|
|
out.push_str(rest);
|
|
if !unresolved.is_empty() {
|
|
return Err(CallError::internal(format!(
|
|
"path template `{template}` references unbound placeholder(s): {}",
|
|
unresolved.join(", ")
|
|
)));
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
/// True when `input_name` is consumed by a `{input_name}` placeholder in
|
|
/// the path template. Drives input routing in [`build_request`]; the
|
|
/// value itself is rendered by [`render_path_template`].
|
|
pub(crate) fn is_path_placeholder(input_name: &str, template: &str) -> bool {
|
|
let placeholder = format!("{{{input_name}}}");
|
|
template.contains(&placeholder)
|
|
}
|
|
|
|
fn parse_base_url(base_url: &str) -> Result<Url, CallError> {
|
|
let parsed = Url::parse(base_url)
|
|
.map_err(|e| CallError::internal(format!("invalid base_url `{base_url}`: {e}")))?;
|
|
let scheme = parsed.scheme();
|
|
if scheme != "https" && scheme != "http" {
|
|
return Err(CallError::internal(format!(
|
|
"base_url `{base_url}` must use https (or http for plain non-TLS origins); `{scheme}` is not an HTTP scheme"
|
|
)));
|
|
}
|
|
let host = parsed.host_str().unwrap_or_default();
|
|
if host.is_empty() {
|
|
return Err(CallError::internal(format!(
|
|
"base_url `{base_url}` must include an explicit host"
|
|
)));
|
|
}
|
|
if !parsed.username().is_empty() || parsed.password().is_some() {
|
|
return Err(CallError::internal(format!(
|
|
"base_url `{base_url}` must not embed userinfo; credentials are injected per-operation from Capabilities"
|
|
)));
|
|
}
|
|
Ok(parsed)
|
|
}
|
|
|
|
fn request_path(rendered_path: &str) -> Result<String, CallError> {
|
|
let trimmed = rendered_path.trim_matches('/');
|
|
if trimmed.is_empty() {
|
|
return Err(CallError::internal(
|
|
"path template resolves to an empty request path; at least one segment is required",
|
|
));
|
|
}
|
|
let mut path = String::new();
|
|
for segment in trimmed.split('/') {
|
|
path.push('/');
|
|
let mut pieces = segment.split('%');
|
|
utf8_percent_encode_into(&mut path, pieces.next().unwrap_or_default());
|
|
for piece in pieces {
|
|
path.push('%');
|
|
utf8_percent_encode_into(&mut path, piece);
|
|
}
|
|
}
|
|
Ok(path)
|
|
}
|
|
|
|
fn utf8_percent_encode_into(out: &mut String, text: &str) {
|
|
for piece in utf8_percent_encode(text, PATH_AFTER_PERCENT_ENCODE_SET) {
|
|
out.push_str(piece);
|
|
}
|
|
}
|
|
|
|
fn assemble_request_url(base_url: &str, rendered_path: &str) -> Result<Url, CallError> {
|
|
let base = parse_base_url(base_url)?;
|
|
let request_path = request_path(rendered_path)?;
|
|
let base_path = base.path();
|
|
let base_dir = match base_path.strip_suffix('/') {
|
|
Some(stripped) => stripped,
|
|
None => base_path,
|
|
};
|
|
let mut full_path = String::with_capacity(base_dir.len() + request_path.len() + 1);
|
|
full_path.push_str(base_dir);
|
|
full_path.push_str(&request_path);
|
|
let mut url = base.clone();
|
|
url.set_path(&full_path);
|
|
assert_untouched_by_normalization(&url, base_dir, rendered_path, base_url)?;
|
|
let same_origin = url.scheme() == base.scheme()
|
|
&& url.host() == base.host()
|
|
&& url.port_or_known_default() == base.port_or_known_default();
|
|
if !same_origin {
|
|
return Err(CallError::internal(format!(
|
|
"request path `{rendered_path}` resolved against `{base_url}` changed the target origin: {} != {}",
|
|
url.origin().ascii_serialization(),
|
|
base.origin().ascii_serialization()
|
|
)));
|
|
}
|
|
Ok(url)
|
|
}
|
|
|
|
/// Post-`set_path` invariant (FWD-13): the decoded segments of the
|
|
/// final URL path must equal `base_dir`'s segments plus the decoded
|
|
/// rendered segments, byte-identical. `set_path` is the one step this
|
|
/// crate does not control, and its parser normalizes lone dot segments
|
|
/// away (`/../x` → `/x`); this check turns any future normalizer
|
|
/// surprise into a loud `INTERNAL` error instead of a silently
|
|
/// re-routed authenticated request.
|
|
fn assert_untouched_by_normalization(
|
|
url: &Url,
|
|
base_dir: &str,
|
|
rendered_path: &str,
|
|
base_url: &str,
|
|
) -> Result<(), CallError> {
|
|
let rendered_segments = rendered_path
|
|
.trim_matches('/')
|
|
.split('/')
|
|
.filter(|s| !s.is_empty())
|
|
.map(|s| percent_decode_str(s).decode_utf8_lossy().into_owned());
|
|
let base_segments: Vec<&str> = base_dir.split('/').filter(|s| !s.is_empty()).collect();
|
|
let expected: Vec<String> = base_segments
|
|
.iter()
|
|
.copied()
|
|
.map(str::to_string)
|
|
.chain(rendered_segments)
|
|
.collect();
|
|
let actual: Vec<String> = url
|
|
.path()
|
|
.split('/')
|
|
.filter(|s| !s.is_empty())
|
|
.map(|s| percent_decode_str(s).decode_utf8_lossy().into_owned())
|
|
.collect();
|
|
let matches = actual.len() == expected.len()
|
|
&& std::iter::Iterator::zip(actual.iter(), expected.iter()).all(|(a, e)| a == e);
|
|
if matches {
|
|
Ok(())
|
|
} else {
|
|
Err(CallError::internal(format!(
|
|
"request path `{rendered_path}` resolved against `{base_url}` was rewritten by URL normalization: expected segments {expected:?}, got {actual:?}"
|
|
)))
|
|
}
|
|
}
|
|
|
|
#[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: axum::body::Bytes) -> Option<String> {
|
|
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<axum::body::Bytes, BodyReadError> {
|
|
let mut stream = response.bytes_stream();
|
|
let mut buf: Vec<u8> = 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::<Value>(&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<Value> = 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}")),
|
|
),
|
|
}
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(crate) async fn forward(
|
|
http_client: &Arc<SharedHttpClient>,
|
|
base_url: &str,
|
|
path_template: &str,
|
|
method: &str,
|
|
auth_scheme: &Option<HttpAuthScheme>,
|
|
default_headers: &HashMap<String, String>,
|
|
namespace: &str,
|
|
input_schema: &Value,
|
|
error_status_codes: &[(u16, String)],
|
|
input: Value,
|
|
context: OperationContext,
|
|
) -> ResponseEnvelope {
|
|
let request_id = context.request_id.clone();
|
|
|
|
let (http_method, url, body, headers) = match build_request(
|
|
base_url,
|
|
path_template,
|
|
method,
|
|
auth_scheme,
|
|
default_headers,
|
|
namespace,
|
|
input_schema,
|
|
&input,
|
|
&context,
|
|
) {
|
|
Ok(parts) => parts,
|
|
Err(err) => return ResponseEnvelope::error(request_id, err),
|
|
};
|
|
|
|
let http_client = http_client.client();
|
|
|
|
let request_builder = http_client
|
|
.request(http_method, url.as_str())
|
|
.headers(headers)
|
|
.header(ACCEPT, "*/*");
|
|
|
|
let request_builder = match body.as_ref() {
|
|
Some(b) => {
|
|
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,
|
|
};
|
|
|
|
let response: reqwest::Response = match request_builder.send().await {
|
|
Ok(r) => r,
|
|
Err(err) => {
|
|
return ResponseEnvelope::error(
|
|
request_id,
|
|
CallError::internal(format!("HTTP request failed: {err}")),
|
|
);
|
|
}
|
|
};
|
|
|
|
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_<status>` 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();
|
|
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(BodyReadError::TooLarge) => {
|
|
message.push_str(": [error body too large to echo]");
|
|
}
|
|
Err(_) => {}
|
|
}
|
|
ResponseEnvelope::error(request_id, CallError::new(code, message, false))
|
|
}
|
|
|
|
/// Maps a parsed SSE frame to a response envelope under the FWD-17
|
|
/// payload contract: a payload that is valid JSON surfaces as the
|
|
/// decoded value itself (a number payload stays a number, a
|
|
/// quoted-string payload stays a string — the distinction a raw-text
|
|
/// fallback would erase); any other payload surfaces as the
|
|
/// `{"data": <raw payload>, "event": <event-name|null>}` wrapper, so
|
|
/// a non-JSON stream is field-addressable and an upstream's
|
|
/// named-event conventions are visible on it. An empty payload is
|
|
/// `Null`. JSON payloads intentionally carry no `event` binding: a
|
|
/// JSON frame surfaces as itself (shape stability for JSON-first
|
|
/// upstreams, and the spec's implicit `event: message` default never
|
|
/// wraps a payload).
|
|
fn sse_event_envelope(event: SseEvent, request_id: &str) -> ResponseEnvelope {
|
|
let parsed = if event.data.trim().is_empty() {
|
|
Value::Null
|
|
} else {
|
|
match serde_json::from_str::<Value>(&event.data) {
|
|
Ok(value) => value,
|
|
Err(_) => serde_json::json!({
|
|
"data": event.data,
|
|
"event": event.event,
|
|
}),
|
|
}
|
|
};
|
|
ResponseEnvelope::ok(request_id, parsed)
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(crate) fn forward_stream(
|
|
http_client: &Arc<SharedHttpClient>,
|
|
base_url: &str,
|
|
path_template: &str,
|
|
method: &str,
|
|
auth_scheme: &Option<HttpAuthScheme>,
|
|
default_headers: &HashMap<String, String>,
|
|
namespace: &str,
|
|
input_schema: &Value,
|
|
error_status_codes: &[(u16, String)],
|
|
input: Value,
|
|
context: OperationContext,
|
|
) -> ResponseStream {
|
|
let request_id = context.request_id.clone();
|
|
|
|
let (http_method, url, body, headers) = match build_request(
|
|
base_url,
|
|
path_template,
|
|
method,
|
|
auth_scheme,
|
|
default_headers,
|
|
namespace,
|
|
input_schema,
|
|
&input,
|
|
&context,
|
|
) {
|
|
Ok(parts) => parts,
|
|
Err(err) => {
|
|
return Box::pin(stream::once(async move {
|
|
ResponseEnvelope::error(request_id, err)
|
|
}));
|
|
}
|
|
};
|
|
|
|
let http_client = Arc::clone(http_client);
|
|
let error_status_codes = error_status_codes.to_vec();
|
|
|
|
let request_id_stream = request_id.clone();
|
|
let error_status_codes_stream = error_status_codes.clone();
|
|
let stream_byte_cap = http_client.config().stream_total_byte_cap;
|
|
|
|
let init = async move {
|
|
let request_builder = http_client
|
|
.stream_client()
|
|
.request(http_method, url.as_str())
|
|
.headers(headers)
|
|
.header(ACCEPT, "text/event-stream");
|
|
let request_builder = match body.as_ref() {
|
|
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
|
|
.map_err(|err| CallError::internal(format!("HTTP request failed: {err}")))
|
|
};
|
|
|
|
let sse = stream::once(init).flat_map(move |result| {
|
|
let request_id = request_id_stream.clone();
|
|
let error_status_codes = error_status_codes_stream.clone();
|
|
match result {
|
|
Err(err) => Box::pin(stream::once(async move {
|
|
ResponseEnvelope::error(request_id, err)
|
|
})) as ResponseStream,
|
|
Ok(response) => {
|
|
let status = response.status();
|
|
if !status.is_success() {
|
|
let request_id = request_id.clone();
|
|
Box::pin(stream::once(async move {
|
|
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(
|
|
(
|
|
response.bytes_stream(),
|
|
SseParser::new(),
|
|
false,
|
|
0u64,
|
|
),
|
|
move |(mut bytes, mut parser, broken, mut total_bytes)| {
|
|
let request_id = request_id_inner.clone();
|
|
async move {
|
|
if broken {
|
|
return None;
|
|
}
|
|
match bytes.next().await {
|
|
Some(Ok(chunk)) => {
|
|
let chunk_len = chunk.len() as u64;
|
|
if stream_byte_cap > 0
|
|
&& total_bytes.saturating_add(chunk_len)
|
|
> stream_byte_cap
|
|
{
|
|
let error = CallError::new(
|
|
"HTTP_413",
|
|
format!(
|
|
"upstream SSE stream exceeded the {stream_byte_cap}-byte total streamed-bytes cap on a subscription operation"
|
|
),
|
|
false,
|
|
);
|
|
return Some((
|
|
vec![ResponseEnvelope::error(
|
|
request_id, error,
|
|
)],
|
|
(bytes, parser, true, total_bytes),
|
|
));
|
|
}
|
|
total_bytes = total_bytes.saturating_add(chunk_len);
|
|
match parser.feed(&chunk, false) {
|
|
Ok(events) => {
|
|
let envelopes: Vec<ResponseEnvelope> =
|
|
events
|
|
.into_iter()
|
|
.map(|e| {
|
|
sse_event_envelope(
|
|
e, &request_id,
|
|
)
|
|
})
|
|
.collect();
|
|
Some((
|
|
envelopes,
|
|
(bytes, parser, false, total_bytes),
|
|
))
|
|
}
|
|
Err(err) => {
|
|
let error = CallError::internal(format!(
|
|
"SSE parse error: {err}"
|
|
));
|
|
Some((
|
|
vec![ResponseEnvelope::error(
|
|
request_id, error,
|
|
)],
|
|
(bytes, parser, true, total_bytes),
|
|
))
|
|
}
|
|
}
|
|
},
|
|
Some(Err(err)) => {
|
|
let error = CallError::internal(format!(
|
|
"SSE stream error: {err}"
|
|
));
|
|
Some((
|
|
vec![ResponseEnvelope::error(request_id, error)],
|
|
(bytes, parser, true, total_bytes),
|
|
))
|
|
}
|
|
None => match parser.feed(&[], true) {
|
|
Ok(events) if !events.is_empty() => {
|
|
let envelopes: Vec<ResponseEnvelope> = events
|
|
.into_iter()
|
|
.map(|e| sse_event_envelope(e, &request_id))
|
|
.collect();
|
|
Some((
|
|
envelopes,
|
|
(bytes, parser, true, total_bytes),
|
|
))
|
|
}
|
|
_ => None,
|
|
},
|
|
}
|
|
}
|
|
},
|
|
)
|
|
.flat_map(stream::iter),
|
|
) as ResponseStream
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
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`, plus the
|
|
/// frame's `event:` field name when the upstream sent one (WHATWG
|
|
/// event-stream semantics: the last `event:` line before the blank
|
|
/// line wins; absent → `None`, never the implicit `message` default —
|
|
/// consumers distinguish "upstream named it" from "default name").
|
|
pub(crate) struct SseEvent {
|
|
pub(crate) data: String,
|
|
pub(crate) event: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub(crate) enum SseParseError {
|
|
#[error("SSE event buffer exceeded {SSE_EVENT_BUFFER_CAP} bytes without a complete event")]
|
|
BufferOverflow,
|
|
}
|
|
|
|
/// 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
|
|
/// unterminated event — trips `SseParseError::BufferOverflow` instead
|
|
/// of buffering without bound. The check runs *before* the buffer
|
|
/// takes a chunk's bytes, so the reassembly buffer can never exceed
|
|
/// the cap (FWD-14). This bounds one *line*, not the
|
|
/// stream; the per-subscription total is
|
|
/// [`HttpClientConfig::stream_total_byte_cap`], enforced by the
|
|
/// `forward_stream` unfold across every `feed`.
|
|
pub(crate) const SSE_EVENT_BUFFER_CAP: usize = 1024 * 1024;
|
|
|
|
/// Incremental byte-level SSE frame parser.
|
|
///
|
|
/// Holds raw bytes across `feed` calls so a frame split across TCP
|
|
/// chunks is reassembled, and only decodes UTF-8 once a complete line
|
|
/// (or EOF) bounds the decode window — a multi-byte character split at
|
|
/// a chunk boundary is therefore not corrupted. Framing follows the
|
|
/// WHATWG `text/event-stream` draft semantics for what the call
|
|
/// protocol needs: lines split on `\n` with optional trailing `\r`;
|
|
/// `data:` fields accumulate and join with `\n`; `event:` is captured
|
|
/// per frame (last one before the blank line wins) and surfaced on
|
|
/// [`SseEvent::event`] for non-JSON payloads (FWD-17); `id:` and
|
|
/// `retry:` fields are accepted and ignored; a blank line dispatches
|
|
/// the pending event; a pending event with data is dispatched at EOF.
|
|
pub(crate) struct SseParser {
|
|
buf: Vec<u8>,
|
|
data_lines: Vec<String>,
|
|
data_seen: bool,
|
|
event_name: Option<String>,
|
|
bom_stripped: bool,
|
|
}
|
|
|
|
impl SseParser {
|
|
pub(crate) fn new() -> Self {
|
|
Self {
|
|
buf: Vec::new(),
|
|
data_lines: Vec::new(),
|
|
data_seen: false,
|
|
event_name: None,
|
|
bom_stripped: false,
|
|
}
|
|
}
|
|
|
|
/// Feeds one chunk and drains every complete event (a blank line
|
|
/// dispatches; the last line stays buffered unless `eof`). With
|
|
/// `eof`, also dispatches a pending event if it carries data
|
|
/// lines, and flushes the buffer.
|
|
pub(crate) fn feed(&mut self, chunk: &[u8], eof: bool) -> Result<Vec<SseEvent>, SseParseError> {
|
|
if self.buf.len().saturating_add(chunk.len()) > SSE_EVENT_BUFFER_CAP {
|
|
return Err(SseParseError::BufferOverflow);
|
|
}
|
|
self.buf.extend_from_slice(chunk);
|
|
let mut events = Vec::new();
|
|
let mut start = 0usize;
|
|
while let Some(nl) = self.buf[start..].iter().position(|&b| b == b'\n') {
|
|
let end = start + nl;
|
|
let line_end = if end > start && self.buf[end - 1] == b'\r' {
|
|
end - 1
|
|
} else {
|
|
end
|
|
};
|
|
let line = self.buf[start..line_end].to_vec();
|
|
if let Some(event) = self.parse_line(&line) {
|
|
events.push(event);
|
|
}
|
|
start = end + 1;
|
|
}
|
|
if eof {
|
|
if start < self.buf.len() {
|
|
let line = self.buf[start..].to_vec();
|
|
if let Some(event) = self.parse_line(&line) {
|
|
events.push(event);
|
|
}
|
|
}
|
|
if self.data_seen {
|
|
if let Some(event) = self.complete_event() {
|
|
events.push(event);
|
|
}
|
|
}
|
|
self.buf.clear();
|
|
} else {
|
|
self.buf.drain(..start);
|
|
}
|
|
Ok(events)
|
|
}
|
|
|
|
fn parse_line(&mut self, line: &[u8]) -> Option<SseEvent> {
|
|
if !self.bom_stripped {
|
|
self.bom_stripped = true;
|
|
let bom = b"\xef\xbb\xbf";
|
|
let line = if line.starts_with(bom) {
|
|
&line[bom.len()..]
|
|
} else {
|
|
line
|
|
};
|
|
return self.parse_line(line);
|
|
}
|
|
let text = match std::str::from_utf8(line) {
|
|
Ok(t) => t,
|
|
Err(_) => return None,
|
|
};
|
|
if text.is_empty() {
|
|
return if self.data_seen {
|
|
self.complete_event()
|
|
} else {
|
|
self.discard_event();
|
|
None
|
|
};
|
|
}
|
|
if text.starts_with(':') {
|
|
return None;
|
|
}
|
|
let (field, value) = match text.split_once(':') {
|
|
Some((f, v)) => (f, v.strip_prefix(' ').unwrap_or(v)),
|
|
None => (text, ""),
|
|
};
|
|
if field == "data" {
|
|
self.data_lines.push(value.to_string());
|
|
self.data_seen = true;
|
|
} else if field == "event" && !value.is_empty() {
|
|
self.event_name = Some(value.to_string());
|
|
}
|
|
None
|
|
}
|
|
|
|
fn complete_event(&mut self) -> Option<SseEvent> {
|
|
self.data_seen = false;
|
|
Some(SseEvent {
|
|
data: std::mem::take(&mut self.data_lines).join("\n"),
|
|
event: self.event_name.take(),
|
|
})
|
|
}
|
|
|
|
fn discard_event(&mut self) {
|
|
self.data_seen = false;
|
|
self.event_name = None;
|
|
self.data_lines.clear();
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use alkcall::core::types::Capabilities;
|
|
use alkcall::registry::context::{AbortPolicy, ScopedPeerEnv};
|
|
use serde_json::json;
|
|
use std::collections::HashMap as TestHashMap;
|
|
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]
|
|
impl alkcall::registry::env::OperationEnv for NoopEnv {
|
|
async fn invoke_with_policy(
|
|
&self,
|
|
_ns: &str,
|
|
_op: &str,
|
|
_input: Value,
|
|
parent: &OperationContext,
|
|
_policy: AbortPolicy,
|
|
) -> ResponseEnvelope {
|
|
ResponseEnvelope::ok(parent.request_id.clone(), Value::Null)
|
|
}
|
|
fn contains(&self, _name: &str) -> bool {
|
|
false
|
|
}
|
|
}
|
|
OperationContext {
|
|
request_id: "req-fwd".to_string(),
|
|
parent_request_id: None,
|
|
identity: None,
|
|
handler_identity: None,
|
|
forwarded_for: None,
|
|
capabilities: Capabilities::new(),
|
|
metadata: TestHashMap::new(),
|
|
scoped_env: ScopedPeerEnv::empty(),
|
|
env: TestArc::new(NoopEnv),
|
|
abort_policy: AbortPolicy::default(),
|
|
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
|
|
internal: true,
|
|
ownership: None,
|
|
}
|
|
}
|
|
|
|
fn request_url(base_url: &str, template: &str, input: Value) -> Result<url::Url, CallError> {
|
|
let ctx = noop_context();
|
|
let (_, url, _, _) = build_request(
|
|
base_url,
|
|
template,
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object", "additionalProperties": true}),
|
|
&input,
|
|
&ctx,
|
|
)?;
|
|
Ok(url)
|
|
}
|
|
|
|
#[test]
|
|
fn traversal_value_cannot_escape_template_path() {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/repos/{owner}/{repo}/issues",
|
|
json!({"owner": "../../admin", "repo": "x"}),
|
|
)
|
|
.expect("request builds");
|
|
assert_eq!(url.host_str(), Some("api.example.com"));
|
|
assert_eq!(url.path(), "/repos/..%2F..%2Fadmin/x/issues");
|
|
assert!(!url.path().contains("/admin"));
|
|
}
|
|
|
|
#[test]
|
|
fn structural_characters_cannot_split_or_inject_url_parts() {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/files/{name}",
|
|
json!({"name": "a?via=query#frag"}),
|
|
)
|
|
.expect("request builds");
|
|
assert_eq!(url.query(), None, "`?` in a value must be encoded");
|
|
assert_eq!(url.fragment(), None, "`#` in a value must be encoded");
|
|
assert_eq!(url.path(), "/files/a%3Fvia=query%23frag");
|
|
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/files/{name}",
|
|
json!({"name": "a b/c\\d"}),
|
|
)
|
|
.expect("request builds");
|
|
assert_eq!(url.path_segments().map(|s| s.count()), Some(2));
|
|
assert_eq!(url.path(), "/files/a%20b%2Fc%5Cd");
|
|
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/files/{name}",
|
|
json!({"name": "héllo→世界"}),
|
|
)
|
|
.expect("request builds");
|
|
assert_eq!(url.path(), "/files/h%C3%A9llo%E2%86%92%E4%B8%96%E7%95%8C");
|
|
}
|
|
|
|
#[test]
|
|
fn rendering_is_single_pass_and_never_re_substitutes() {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/x/{a}/{b}",
|
|
json!({"a": "{b}", "b": "second"}),
|
|
)
|
|
.expect("request builds");
|
|
assert_eq!(url.path(), "/x/%7Bb%7D/second");
|
|
|
|
let rendered = render_path_template(
|
|
"/x/{a}",
|
|
Some(&json!({"a": "../../{b}"}).as_object().unwrap().clone()),
|
|
)
|
|
.expect("renders");
|
|
assert_eq!(rendered, "/x/..%2F..%2F%7Bb%7D");
|
|
}
|
|
|
|
#[test]
|
|
fn base_path_prefix_is_preserved() {
|
|
let url = request_url("https://api.openai.com/v1", "/chat/completions", json!({}))
|
|
.expect("request builds");
|
|
assert_eq!(url.path(), "/v1/chat/completions");
|
|
|
|
let url =
|
|
request_url("https://api.example.com", "/data", json!({})).expect("request builds");
|
|
assert_eq!(url.path(), "/data");
|
|
}
|
|
|
|
#[test]
|
|
fn absolute_url_in_input_cannot_change_origin_or_hop_paths() {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/fetch/{url}",
|
|
json!({"url": "http://169.254.169.254/latest/meta-data"}),
|
|
)
|
|
.expect("absolute URL in a path value stays an encoded segment");
|
|
assert_eq!(url.host_str(), Some("api.example.com"));
|
|
assert_eq!(
|
|
url.path(),
|
|
"/fetch/http:%2F%2F169.254.169.254%2Flatest%2Fmeta-data"
|
|
);
|
|
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/fetch/{target}",
|
|
json!({"target": "https://evil.example.com/x"}),
|
|
)
|
|
.expect("https absolute URL also stays an encoded segment");
|
|
assert_eq!(url.host_str(), Some("api.example.com"));
|
|
assert_eq!(url.path(), "/fetch/https:%2F%2Fevil.example.com%2Fx");
|
|
}
|
|
|
|
#[test]
|
|
fn unbound_and_malformed_templates_error_loudly() {
|
|
let err = request_url("https://api.example.com", "/x/{missing}", json!({}))
|
|
.expect_err("unbound placeholder must error");
|
|
assert!(err.message.contains("unbound placeholder"));
|
|
|
|
let err = request_url("https://api.example.com", "/x/{open", json!({}))
|
|
.expect_err("unterminated placeholder must error");
|
|
assert!(err.message.contains("unterminated"));
|
|
|
|
let err = request_url("https://api.example.com", "/x{missing}/a/b", json!({}))
|
|
.expect_err("partial render without the placeholder is still loud");
|
|
assert!(err.message.contains("unbound placeholder"));
|
|
}
|
|
|
|
#[test]
|
|
fn base_url_validation_rejects_bad_inputs() {
|
|
let err = request_url("ftp://api.example.com", "/x", json!({}))
|
|
.expect_err("non-http scheme must be rejected");
|
|
assert!(err.message.contains("not an HTTP scheme"));
|
|
|
|
let err = request_url("https://u:p@api.example.com", "/x", json!({}))
|
|
.expect_err("userinfo must be rejected");
|
|
assert!(err.message.contains("userinfo"));
|
|
|
|
let err = request_url("not a url at all", "/x", json!({}))
|
|
.expect_err("unparseable base must be rejected");
|
|
assert!(err.message.contains("invalid base_url"));
|
|
}
|
|
|
|
#[test]
|
|
fn query_values_remain_encoded_via_query_pairs_mut() {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/search",
|
|
json!({"q": "a&b=c d", "lang": "en"}),
|
|
)
|
|
.expect("request builds");
|
|
assert_eq!(url.query(), Some("lang=en&q=a%26b%3Dc+d"));
|
|
}
|
|
|
|
#[test]
|
|
fn undeclared_input_keys_are_rejected_not_sent_upstream() {
|
|
let ctx = noop_context();
|
|
let schema = json!({
|
|
"type": "object",
|
|
"properties": {"owner": {"type": "string"}, "body": {"type": "object"}},
|
|
});
|
|
for input in [
|
|
json!({"owner": "a", "debug": "true"}),
|
|
json!({"owner": "a", "impersonate_id": "x"}),
|
|
json!({"debug": "true"}),
|
|
] {
|
|
let err = build_request(
|
|
"https://api.example.com",
|
|
"/x/{owner}",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&schema,
|
|
&input,
|
|
&ctx,
|
|
)
|
|
.expect_err("undeclared peer input must be rejected");
|
|
assert_eq!(err.code, "INVALID_INPUT", "input was: {input}");
|
|
assert!(err.message.contains("not declared"), "input was: {input}");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn declared_body_and_header_params_route_off_the_query_string() {
|
|
let ctx = noop_context();
|
|
let schema = json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"q": {"type": "string"},
|
|
"X-Trace-Id": {"type": "string", "wire": "header"},
|
|
"body": {"type": "object"},
|
|
},
|
|
});
|
|
let (_, url, body, headers) = build_request(
|
|
"https://api.example.com",
|
|
"/search",
|
|
"POST",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&schema,
|
|
&json!({"q": "rust", "X-Trace-Id": "t-1", "body": {"page": 2}}),
|
|
&ctx,
|
|
)
|
|
.expect("declared input builds");
|
|
assert_eq!(url.query(), Some("q=rust"));
|
|
assert_eq!(
|
|
headers
|
|
.get("x-trace-id")
|
|
.expect("header param sent as header")
|
|
.to_str()
|
|
.expect("ascii header"),
|
|
"t-1"
|
|
);
|
|
assert_eq!(body, Some(json!({"page": 2})));
|
|
}
|
|
|
|
#[test]
|
|
fn non_object_input_is_rejected() {
|
|
let ctx = noop_context();
|
|
let schema = json!({"type": "object", "properties": {"q": {"type": "string"}}});
|
|
for input in [json!(null), json!([1]), json!("str"), json!(42)] {
|
|
let err = build_request(
|
|
"https://api.example.com",
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&schema,
|
|
&input,
|
|
&ctx,
|
|
)
|
|
.expect_err("non-object input must be rejected");
|
|
assert_eq!(err.code, "INVALID_INPUT");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn additional_properties_true_opts_into_catch_all_input() {
|
|
let ctx = noop_context();
|
|
let schema = json!({
|
|
"type": "object",
|
|
"properties": {"q": {"type": "string"}},
|
|
"additionalProperties": true,
|
|
});
|
|
let (_, url, _, _) = build_request(
|
|
"https://api.example.com",
|
|
"/search",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&schema,
|
|
&json!({"q": "rust", "debug": "true"}),
|
|
&ctx,
|
|
)
|
|
.expect("catch-all input builds");
|
|
let q = url.query().expect("query present");
|
|
assert!(q.contains("q=rust") && q.contains("debug=true"), "{q}");
|
|
}
|
|
|
|
/// FWD-18, routing half pin: a key that matched a placeholder never
|
|
/// also emits as a query parameter — the skip in `build_request`
|
|
/// precedes query routing and is value-shape-agnostic. The scalar
|
|
/// case renders into the path only; the object-value case is
|
|
/// rejected (see the structural-value test below), so here the
|
|
/// companion non-placeholder key proves query routing stays intact
|
|
/// around the placeholder skip.
|
|
#[test]
|
|
fn placeholder_keys_never_double_route_as_query_params() {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/repos/{owner}",
|
|
json!({"owner": "octocat", "extra": "q-val"}),
|
|
)
|
|
.expect("scalar placeholder builds");
|
|
assert_eq!(url.path(), "/repos/octocat");
|
|
assert_eq!(
|
|
url.query(),
|
|
Some("extra=q-val"),
|
|
"only the non-placeholder key routes to query"
|
|
);
|
|
|
|
let ctx = noop_context();
|
|
let schema = json!({
|
|
"type": "object",
|
|
"properties": {"id": {"type": "string"}},
|
|
});
|
|
let err = build_request(
|
|
"https://api.example.com",
|
|
"/items/{id}",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&schema,
|
|
&json!({"id": "x", "undeclared": "v"}),
|
|
&ctx,
|
|
)
|
|
.expect_err("undeclared key still rejected with a placeholder present");
|
|
assert_eq!(err.code, "INVALID_INPUT");
|
|
}
|
|
|
|
/// FWD-18, structural-value decision: an object/array value under a
|
|
/// placeholder key is an `INVALID_INPUT` error, not a spliced JSON
|
|
/// segment. A path placeholder designates exactly one literal
|
|
/// segment; the pre-decision behavior embedded the minified JSON
|
|
/// into the path (and, under an additionalProperties schema, also
|
|
/// emitted the key as a query param — the double-route).
|
|
#[test]
|
|
fn object_or_array_path_values_error_instead_of_splicing_json_into_the_path() {
|
|
for value in [json!({"a": 1}), json!([1, 2])] {
|
|
let err = request_url(
|
|
"https://api.example.com",
|
|
"/things/{id}",
|
|
json!({"id": value}),
|
|
)
|
|
.expect_err("structural path values must be rejected");
|
|
assert_eq!(err.code, "INVALID_INPUT", "value was: {value}");
|
|
assert!(
|
|
err.message.contains("must be a scalar"),
|
|
"value was: {value}"
|
|
);
|
|
assert!(
|
|
!err.message.contains("{\"a\":1}") && !err.message.contains("[1,2]"),
|
|
"error must not echo the structural value: {value}"
|
|
);
|
|
}
|
|
|
|
for value in [json!("text"), json!(42), json!(true)] {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/things/{id}",
|
|
json!({ "id": value }),
|
|
)
|
|
.expect("scalar path values still render");
|
|
assert!(
|
|
url.path().starts_with("/things/"),
|
|
"scalar {value} renders into the path"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// FWD-19 pin (documented trade-off, not a rejection): literal `%`
|
|
/// in template text is preserved as-is, so an assembly-supplied
|
|
/// pre-encoded template like `/s3%2Fkeys` reaches the upstream
|
|
/// verbatim (most stacks route `%2F` differently from `/` — the
|
|
/// assembly layer owns that choice, ADR-066). A `%` arriving in a
|
|
/// *value* is always encoded to `%25`, so the only bare `%` in a
|
|
/// rendered path is one the template author placed.
|
|
#[test]
|
|
fn percent_in_template_text_survives_and_percent_in_values_are_always_encoded() {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/s3%2Fkeys/{name}",
|
|
json!({"name": "x"}),
|
|
)
|
|
.expect("template text may carry literal percent escapes");
|
|
assert_eq!(url.path(), "/s3%2Fkeys/x");
|
|
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/files/{name}",
|
|
json!({"name": "a%2Fb"}),
|
|
)
|
|
.expect("value percent is encoded, not preserved");
|
|
assert_eq!(url.path(), "/files/a%252Fb");
|
|
}
|
|
|
|
/// Empirical pin of the failure mode FWD-13 describes, against the
|
|
/// locked `url` crate: lone `.`/`..` (and their percent-escaped
|
|
/// spellings) are silently *normalized away* by `Url::set_path`, so
|
|
/// the encode set alone cannot keep the rendered path faithful.
|
|
#[test]
|
|
fn url_set_path_normalizes_lone_dot_and_dot_dot_path_values() {
|
|
let mut url = Url::parse("https://api.example.com").expect("parses");
|
|
for (spliced, normalized) in [
|
|
("/tenants/../resources", "/resources"),
|
|
("/tenants/./resources", "/tenants/resources"),
|
|
("/files/..", "/"),
|
|
("/repos/%2E%2E/x", "/x"),
|
|
] {
|
|
url.set_path(spliced);
|
|
assert_eq!(
|
|
url.path(),
|
|
normalized,
|
|
"set_path silently normalized `{spliced}`"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn lone_dot_dot_path_value_is_rejected() {
|
|
let err = request_url(
|
|
"https://api.example.com",
|
|
"/tenants/{tenant}/resources",
|
|
json!({"tenant": ".."}),
|
|
)
|
|
.expect_err("lone `..` value must be rejected");
|
|
assert_eq!(err.code, "INVALID_INPUT");
|
|
assert!(
|
|
err.message.contains("lone dot segment"),
|
|
"message must explain the lone-dot rejection: {}",
|
|
err.message
|
|
);
|
|
assert!(
|
|
!err.message.contains(".."),
|
|
"error must not echo the raw value: {}",
|
|
err.message
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn lone_dot_path_value_is_rejected() {
|
|
let err = request_url(
|
|
"https://api.example.com",
|
|
"/tenants/{tenant}/resources",
|
|
json!({"tenant": "."}),
|
|
)
|
|
.expect_err("lone `.` value must be rejected");
|
|
assert_eq!(err.code, "INVALID_INPUT");
|
|
}
|
|
|
|
#[test]
|
|
fn percent_escapes_spellings_of_lone_dots_are_rejected() {
|
|
for value in ["\u{2e}\u{2e}", "%2e%2e", "%2E%2e", "%2e%2E"] {
|
|
let err = request_url(
|
|
"https://api.example.com",
|
|
"/tenants/{tenant}/resources",
|
|
json!({ "tenant": value }),
|
|
)
|
|
.expect_err("escaped lone-dot spellings must be rejected");
|
|
assert_eq!(err.code, "INVALID_INPUT", "value was: {value}");
|
|
}
|
|
|
|
for value in ["%2e", "%2E"] {
|
|
let err = request_url(
|
|
"https://api.example.com",
|
|
"/tenants/{tenant}/resources",
|
|
json!({ "tenant": value }),
|
|
)
|
|
.expect_err("escaped lone-dot spellings must be rejected");
|
|
assert_eq!(err.code, "INVALID_INPUT", "value was: {value}");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn dotted_but_not_lone_dot_values_still_render() {
|
|
for value in [
|
|
"v1.2.3",
|
|
".hidden-file",
|
|
"..hidden",
|
|
"hidden..",
|
|
"a..b",
|
|
".a.b.",
|
|
"...",
|
|
] {
|
|
let url = request_url(
|
|
"https://api.example.com",
|
|
"/files/{name}",
|
|
json!({ "name": value }),
|
|
)
|
|
.unwrap_or_else(|e| panic!("value `{value}` must render: {e:?}"));
|
|
assert!(
|
|
url.path().starts_with("/files/"),
|
|
"value `{value}` rendered into the path"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// FWD-13 belt-and-braces: across a dot/percent/binary corpus the
|
|
/// post-`set_path` invariant holds — the decoded URL path segments
|
|
/// equal the base dir plus the decoded rendered segments,
|
|
/// byte-identical — so a normalizer rewrite (today's lone-dot
|
|
/// removal, tomorrow's regression in the `url` crate) can never
|
|
/// silently re-route an authenticated request. Lone-dot values are
|
|
/// rejected outright; values containing `/` or `\` are rejected as
|
|
/// smuggled separators only if the encode-set layer ever regressed.
|
|
#[test]
|
|
fn post_set_path_invariant_holds_across_dot_percent_binary_corpus() {
|
|
let corpus = [
|
|
"v1.2.3",
|
|
".hidden-file",
|
|
"..",
|
|
".",
|
|
"%2e",
|
|
"%2E",
|
|
"%2e%2e",
|
|
"%2E%2E",
|
|
"%252e",
|
|
"hidden..",
|
|
"a..b",
|
|
"...",
|
|
"a%2Fb",
|
|
"a b",
|
|
"h\\éllo→世界",
|
|
"line\nbreak",
|
|
"tab\tchar",
|
|
"\u{7f}\u{1f600}",
|
|
];
|
|
for value in corpus {
|
|
let outcome = request_url(
|
|
"https://api.example.com/v1",
|
|
"/files/{name}",
|
|
json!({ "name": value }),
|
|
);
|
|
match outcome {
|
|
Ok(url) => {
|
|
assert_eq!(url.host_str(), Some("api.example.com"));
|
|
let segments: Vec<String> = url
|
|
.path_segments()
|
|
.map(|s| {
|
|
s.map(|seg| percent_decode_str(seg).decode_utf8_lossy().into_owned())
|
|
.collect()
|
|
})
|
|
.unwrap_or_default();
|
|
assert_eq!(
|
|
segments.len(),
|
|
3,
|
|
"value `{value:?}` must render as one literal segment under /v1/files/"
|
|
);
|
|
assert_eq!(
|
|
segments.get(2).map(String::as_str),
|
|
Some(value),
|
|
"value `{value:?}` must survive byte-identical as the final segment"
|
|
);
|
|
assert!(
|
|
!segments.iter().any(|s| s == "." || s == ".."),
|
|
"value `{value:?}` must not leave lone dot segments: {:?}",
|
|
segments
|
|
);
|
|
}
|
|
Err(err) => {
|
|
assert_eq!(
|
|
err.code, "INVALID_INPUT",
|
|
"value `{value:?}` may only fail as INVALID_INPUT"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
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")
|
|
}
|
|
|
|
/// Spawns a raw-TCP responder whose response head is written from a
|
|
/// status/content-type pair, then hands the socket to an async
|
|
/// writer so a test can trickle SSE frames over time (the
|
|
/// wire-level seam the FWD-15/FWD-14 tests need: a stream that stays
|
|
/// open past a deadline, or dribbles bytes toward a cap).
|
|
async fn spawn_sse_responder_with_writer<F, Fut>(head: &str, writer: F) -> String
|
|
where
|
|
F: FnOnce(tokio::net::tcp::OwnedWriteHalf) -> Fut + Send + 'static,
|
|
Fut: std::future::Future<Output = ()> + Send,
|
|
{
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
|
.await
|
|
.expect("bind");
|
|
let addr = listener.local_addr().expect("local addr");
|
|
let head = head.to_string();
|
|
tokio::spawn(async move {
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
let Ok((mut sock, _)) = listener.accept().await else {
|
|
return;
|
|
};
|
|
let mut buf = vec![0u8; 8192];
|
|
loop {
|
|
let read = sock.read(&mut buf).await.unwrap_or(0);
|
|
if read == 0 || String::from_utf8_lossy(&buf).contains("\r\n\r\n") {
|
|
break;
|
|
}
|
|
}
|
|
let _ = sock.write_all(head.as_bytes()).await;
|
|
let _ = sock.flush().await;
|
|
let (_, write_half) = sock.into_split();
|
|
writer(write_half).await;
|
|
});
|
|
format!("http://{addr}")
|
|
}
|
|
|
|
fn streaming_client(total_byte_cap: u64) -> TestArc<SharedHttpClient> {
|
|
TestArc::new(
|
|
SharedHttpClient::new(crate::client::HttpClientConfig {
|
|
stream_total_byte_cap: total_byte_cap,
|
|
..crate::client::HttpClientConfig::default()
|
|
})
|
|
.expect("client builds"),
|
|
)
|
|
}
|
|
|
|
/// Builds a client whose configured request/read timeout is the
|
|
/// (test-scaled) old 30 s deadline: the FWD-15 wire test sends
|
|
/// through it and asserts the stream outlives that deadline.
|
|
fn client_with_timeout(timeout: Duration) -> TestArc<SharedHttpClient> {
|
|
TestArc::new(
|
|
SharedHttpClient::new(crate::client::HttpClientConfig {
|
|
request_timeout: Some(timeout),
|
|
connect_timeout: Some(Duration::from_secs(5)),
|
|
read_timeout: Some(timeout),
|
|
..crate::client::HttpClientConfig::default()
|
|
})
|
|
.expect("client 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",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
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",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
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",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
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());
|
|
}
|
|
|
|
/// FWD-17: a valid-JSON payload surfaces as the decoded value
|
|
/// itself; a non-JSON payload surfaces as the
|
|
/// `{"data", "event"}` wrapper. Numbers-vs-quoted-strings are both
|
|
/// valid JSON, so both decode (the raw-text fallback would have
|
|
/// erased that distinction).
|
|
#[tokio::test]
|
|
async fn sse_payload_contract_decodes_json_and_wraps_non_json() {
|
|
let base = spawn_responder(TestArc::new(|_parts| {
|
|
http_response(
|
|
200,
|
|
"text/event-stream",
|
|
b"data: 123\n\ndata: \"123\"\n\ndata: plain text\n\ndata: {\"n\":1}\n\n".to_vec(),
|
|
)
|
|
}))
|
|
.await;
|
|
let stream = forward_stream(
|
|
&minimal_client(),
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
let envelopes = collect_stream(stream).await;
|
|
assert_eq!(envelopes.len(), 4);
|
|
assert_eq!(envelopes[0].result.clone().unwrap(), json!(123));
|
|
assert_eq!(envelopes[1].result.clone().unwrap(), json!("123"));
|
|
assert_eq!(
|
|
envelopes[2].result.clone().unwrap(),
|
|
json!({"data": "plain text", "event": null})
|
|
);
|
|
assert_eq!(envelopes[3].result.clone().unwrap(), json!({"n": 1}));
|
|
}
|
|
|
|
/// FWD-17: an upstream `event:` field on a non-JSON payload is
|
|
/// carried in the wrapper; JSON payloads surface as themselves even
|
|
/// under a named event, and the name is reset after dispatch (the
|
|
/// second frame must not inherit the first frame's name).
|
|
#[tokio::test]
|
|
async fn sse_event_field_carrys_on_the_non_json_wrapper_and_resets() {
|
|
let base = spawn_responder(TestArc::new(|_parts| {
|
|
http_response(
|
|
200,
|
|
"text/event-stream",
|
|
b"event: error\ndata: upstream exploded\n\nevent: custom\ndata: {\"n\":1}\n\ndata: after\n\n"
|
|
.to_vec(),
|
|
)
|
|
}))
|
|
.await;
|
|
let stream = forward_stream(
|
|
&minimal_client(),
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
let envelopes = collect_stream(stream).await;
|
|
assert_eq!(envelopes.len(), 3);
|
|
assert_eq!(
|
|
envelopes[0].result.clone().unwrap(),
|
|
json!({"data": "upstream exploded", "event": "error"})
|
|
);
|
|
assert_eq!(envelopes[1].result.clone().unwrap(), json!({"n": 1}));
|
|
assert_eq!(
|
|
envelopes[2].result.clone().unwrap(),
|
|
json!({"data": "after", "event": null}),
|
|
"the name must not leak across frames"
|
|
);
|
|
}
|
|
|
|
/// FWD-17 parser pin: WHATWG last-wins for repeated `event:` lines
|
|
/// in one frame; an `event:`-only frame (no data) dispatches
|
|
/// nothing but must not leak its name into a later frame.
|
|
#[test]
|
|
fn sse_parser_last_event_wins_and_name_does_not_leak_across_frames() {
|
|
let mut parser = SseParser::new();
|
|
let events = parser
|
|
.feed(b"event: a\nevent: b\ndata: x\n\n", false)
|
|
.expect("ascii only");
|
|
assert_eq!(events.len(), 1);
|
|
assert_eq!(events[0].event.as_deref(), Some("b"));
|
|
assert_eq!(events[0].data, "x");
|
|
|
|
let events = parser
|
|
.feed(b"event: orphan\n\n", false)
|
|
.expect("ascii only");
|
|
assert!(events.is_empty(), "no data field, nothing dispatched");
|
|
|
|
let events = parser.feed(b"data: next\n\n", false).expect("ascii only");
|
|
assert_eq!(events.len(), 1);
|
|
assert_eq!(
|
|
events[0].event.as_deref(),
|
|
None,
|
|
"an event:-only frame must not leak its name"
|
|
);
|
|
}
|
|
|
|
#[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 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",
|
|
&serde_json::json!({"type": "object"}),
|
|
&json!({}),
|
|
&noop_context(),
|
|
)
|
|
.expect_err("invalid default-header name must fail loudly");
|
|
assert!(err.message.contains("bad header"));
|
|
}
|
|
|
|
#[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 result = build_request(
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&Some(HttpAuthScheme::ApiKey {
|
|
header_name: "bad header name".to_string(),
|
|
}),
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&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"));
|
|
}
|
|
}
|
|
}
|
|
|
|
/// FWD-16, loud-missing matrix: an authed operation whose registry
|
|
/// capability is entirely absent (`api_key:` and `http_token:` both
|
|
/// missing) is refused with an `INTERNAL` error naming the missing
|
|
/// capability keys — the request is never sent unauthenticated.
|
|
#[test]
|
|
fn authed_op_with_absent_capability_fails_loudly_in_build_request() {
|
|
for scheme in [
|
|
HttpAuthScheme::Bearer,
|
|
HttpAuthScheme::ApiKey {
|
|
header_name: "x-api-key".to_string(),
|
|
},
|
|
HttpAuthScheme::Basic,
|
|
] {
|
|
let err = build_request(
|
|
"https://api.example.com",
|
|
"/x",
|
|
"GET",
|
|
&Some(scheme),
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&json!({}),
|
|
&noop_context(),
|
|
)
|
|
.expect_err("absent capability must fail loudly");
|
|
assert_eq!(err.code, "INTERNAL");
|
|
assert!(
|
|
err.message
|
|
.contains("capability for namespace `svc` is absent"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
assert!(
|
|
err.message.contains("api_key:svc") && err.message.contains("http_token:svc"),
|
|
"message must name the missing capability keys: {}",
|
|
err.message
|
|
);
|
|
assert!(
|
|
err.message
|
|
.contains("refusing to send the request unauthenticated"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
}
|
|
|
|
/// FWD-16, unchanged arm: `auth_scheme: None` stays unauthenticated
|
|
/// even with empty capabilities — no error, no credential headers.
|
|
#[test]
|
|
fn unauthed_op_with_empty_capabilities_is_unchanged() {
|
|
let (_, _, _, headers) = build_request(
|
|
"https://api.example.com",
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&json!({}),
|
|
&noop_context(),
|
|
)
|
|
.expect("unauthed op builds without capabilities");
|
|
assert!(headers.get(AUTHORIZATION).is_none());
|
|
}
|
|
|
|
/// FWD-16 wire test: an authed op with empty capabilities returns
|
|
/// an error envelope and the upstream receives zero requests (the
|
|
/// responder fails the test if any connection arrives).
|
|
#[tokio::test]
|
|
async fn authed_op_with_empty_capabilities_sends_zero_requests() {
|
|
let upstream_hit = TestArc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let flag = TestArc::clone(&upstream_hit);
|
|
let base = spawn_responder(TestArc::new(move |_| {
|
|
flag.store(true, std::sync::atomic::Ordering::SeqCst);
|
|
http_response(200, "application/json", b"{}".to_vec())
|
|
}))
|
|
.await;
|
|
let envelope =
|
|
call_forward_authed(&base, noop_context(), &Some(HttpAuthScheme::Bearer)).await;
|
|
match envelope.result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
|
|
assert!(
|
|
err.message
|
|
.contains("capability for namespace `svc` is absent"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
assert!(
|
|
!err.message.to_lowercase().contains("token")
|
|
|| err.message.contains("http_token:svc"),
|
|
"error must carry no credential material, only key names: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected loud missing-capability error, got {other:?}"),
|
|
}
|
|
assert!(
|
|
!upstream_hit.load(std::sync::atomic::Ordering::SeqCst),
|
|
"upstream must receive zero requests when the capability is absent"
|
|
);
|
|
}
|
|
|
|
#[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",
|
|
&serde_json::json!({"type": "object"}),
|
|
&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 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:?}"),
|
|
}
|
|
}
|
|
|
|
/// FWD-15 wire test (deadline-scaled): the responder trickles
|
|
/// `: keepalive` comments and one `data:` event past the configured
|
|
/// total request timeout (a scaled stand-in for the 30 s default
|
|
/// the streaming path used to inherit); the subscription must still
|
|
/// be delivering events after that deadline has passed. With the
|
|
/// fix, `forward_stream` sends through the derived no-total-timeout
|
|
/// client, so the stream survives; without it, reqwest 0.13's total
|
|
/// timeout rides into the body stream and kills the subscription at
|
|
/// the deadline.
|
|
#[tokio::test]
|
|
async fn stream_survives_past_the_total_request_timeout() {
|
|
let timeout = Duration::from_millis(500);
|
|
let keepalive = Duration::from_millis(200);
|
|
let total_keepalives = 6u32;
|
|
let start = std::time::Instant::now();
|
|
let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
|
|
let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
|
|
use tokio::io::AsyncWriteExt;
|
|
for _ in 0..total_keepalives {
|
|
tokio::time::sleep(keepalive).await;
|
|
let _ = sock.write_all(b": keepalive\n\n").await;
|
|
let _ = sock.flush().await;
|
|
}
|
|
let _ = sock.write_all(b"data: {\"late\":true}\n\n").await;
|
|
let _ = sock.flush().await;
|
|
let _ = sock.shutdown().await;
|
|
})
|
|
.await;
|
|
let client = client_with_timeout(timeout);
|
|
let stream = forward_stream(
|
|
&client,
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
tokio::pin!(stream);
|
|
let crossed = tokio::time::timeout(Duration::from_secs(5), stream.next())
|
|
.await
|
|
.expect("the stream must deliver within the test budget")
|
|
.expect("stream must not end without the data event");
|
|
assert!(
|
|
start.elapsed() > timeout,
|
|
"the event must arrive after the old total-request deadline (elapsed {:?}, timeout {:?})",
|
|
start.elapsed(),
|
|
timeout
|
|
);
|
|
match crossed.result {
|
|
Ok(value) => assert_eq!(value, json!({"late": true})),
|
|
other => panic!("expected the post-deadline data event, got {other:?}"),
|
|
}
|
|
assert!(
|
|
tokio::time::timeout(Duration::from_millis(500), stream.next())
|
|
.await
|
|
.unwrap_or(None)
|
|
.is_none(),
|
|
"responder closed after the data event; stream must end"
|
|
);
|
|
}
|
|
|
|
/// FWD-14 wire test: a stream whose total bytes exceed the
|
|
/// configured per-subscription cap terminates with exactly one
|
|
/// terminal error envelope (the stream-ends semantics: error frame,
|
|
/// then end).
|
|
#[tokio::test]
|
|
async fn stream_exceeding_total_byte_cap_terminates_with_one_error() {
|
|
let cap = 4096u64;
|
|
let chunk = vec![b'a'; 1024];
|
|
let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
|
|
let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
|
|
use tokio::io::AsyncWriteExt;
|
|
for _ in 0..64 {
|
|
let _ = sock.write_all(b"data: ").await;
|
|
let _ = sock.write_all(&chunk).await;
|
|
let _ = sock.write_all(b"\n\n").await;
|
|
let _ = sock.flush().await;
|
|
}
|
|
let _ = sock.shutdown().await;
|
|
})
|
|
.await;
|
|
let client = streaming_client(cap);
|
|
let stream = forward_stream(
|
|
&client,
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
let envelopes = collect_stream(stream).await;
|
|
assert!(!envelopes.is_empty(), "events before the cap still flow");
|
|
assert!(
|
|
envelopes[..envelopes.len() - 1]
|
|
.iter()
|
|
.all(|e| e.result.is_ok()),
|
|
"every envelope before the terminal one is an event"
|
|
);
|
|
let terminal = envelopes.last().expect("terminal envelope present");
|
|
match &terminal.result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "HTTP_413");
|
|
assert!(err.message.contains("total streamed-bytes cap"));
|
|
}
|
|
other => panic!("expected the terminal cap error, got {other:?}"),
|
|
}
|
|
let ok_count = envelopes[..envelopes.len() - 1].len();
|
|
assert_eq!(
|
|
envelopes.len(),
|
|
ok_count + 1,
|
|
"exactly one terminal error envelope after the last event"
|
|
);
|
|
}
|
|
|
|
/// FWD-14 parser-boundary test: for a never-newline feed the
|
|
/// cap check fires *before* the buffer takes the chunk's bytes, so
|
|
/// the buffered bytes stay at or below the cap — the pre-fix shape
|
|
/// (extend first, check after) buffered past the cap before erroring.
|
|
/// A full-cap buffer remains legal (the cap is inclusive) as long as
|
|
/// the arriving chunk completes a line.
|
|
#[test]
|
|
fn line_cap_trips_before_the_buffer_takes_the_overshooting_chunk() {
|
|
let mut parser = SseParser::new();
|
|
let seed = vec![b'x'; SSE_EVENT_BUFFER_CAP + 1];
|
|
let oversized = parser.feed(&seed, false);
|
|
assert!(
|
|
matches!(oversized, Err(SseParseError::BufferOverflow)),
|
|
"a single over-cap line trips at the pre-extend check"
|
|
);
|
|
let mut parser = SseParser::new();
|
|
let half = vec![b'x'; SSE_EVENT_BUFFER_CAP / 2];
|
|
let first = parser.feed(&half, false);
|
|
assert!(first.is_ok(), "partial line under the cap buffers fine");
|
|
let second = parser.feed(&seed, false);
|
|
assert!(
|
|
matches!(second, Err(SseParseError::BufferOverflow)),
|
|
"the chunk that would push past the cap is rejected before extend"
|
|
);
|
|
let mut parser = SseParser::new();
|
|
let at_cap = vec![b'x'; SSE_EVENT_BUFFER_CAP - 2];
|
|
let ok = parser.feed(&at_cap, false);
|
|
assert!(ok.is_ok(), "a partial line under the cap is legal");
|
|
let framing = parser.feed(b"\n\n", false);
|
|
assert!(
|
|
framing.is_ok(),
|
|
"the newline pair completes the event without tripping the cap"
|
|
);
|
|
let next = parser.feed(&vec![b'y'; SSE_EVENT_BUFFER_CAP], false);
|
|
assert!(
|
|
next.is_ok(),
|
|
"the dispatched event drained the buffer; a fresh full-cap line is legal again"
|
|
);
|
|
let over = parser.feed(b"z", false);
|
|
assert!(
|
|
matches!(over, Err(SseParseError::BufferOverflow)),
|
|
"one byte past a full buffer still trips before extend"
|
|
);
|
|
}
|
|
|
|
/// FWD-07 Once-path decode arm: a `200` with `application/json` and
|
|
/// a body that does not parse surfaces a single `INTERNAL` decode
|
|
/// error envelope — never a partial or fabricated success.
|
|
#[tokio::test]
|
|
async fn malformed_json_body_200_decodes_to_an_internal_error_envelope() {
|
|
let base = spawn_responder(TestArc::new(|_parts| {
|
|
http_response(200, "application/json", b"{not json".to_vec())
|
|
}))
|
|
.await;
|
|
let envelope = call_forward(&base, noop_context()).await;
|
|
match envelope.result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
|
|
assert!(
|
|
err.message.contains("failed to decode response body"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected INTERNAL decode envelope, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// FWD-07 Once-path binary arm: a `200` with
|
|
/// `application/octet-stream` surfaces the body as a JSON byte
|
|
/// array (bounded by the response cap like every other read path).
|
|
#[tokio::test]
|
|
async fn binary_octet_stream_200_surfaces_as_a_byte_array_envelope() {
|
|
let base = spawn_responder(TestArc::new(|_parts| {
|
|
http_response(
|
|
200,
|
|
"application/octet-stream",
|
|
vec![0x00, 0xFF, 0x10, 0x42],
|
|
)
|
|
}))
|
|
.await;
|
|
let envelope = call_forward(&base, noop_context()).await;
|
|
match envelope.result {
|
|
Ok(value) => assert_eq!(value, json!([0, 255, 16, 66])),
|
|
other => panic!("expected a byte-array envelope, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// COV-10 (a): an upstream SSE line longer than the 1 MiB line cap
|
|
/// with no newline terminates `forward_stream` with a single
|
|
/// `INTERNAL` terminal envelope, then the stream ends.
|
|
#[tokio::test]
|
|
async fn oversized_upstream_sse_line_terminates_with_one_internal_envelope() {
|
|
let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
|
|
let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
|
|
use tokio::io::AsyncWriteExt;
|
|
let line = vec![b'a'; SSE_EVENT_BUFFER_CAP + 1];
|
|
let _ = sock.write_all(b"data: ").await;
|
|
let _ = sock.write_all(&line).await;
|
|
let _ = sock.write_all(b"\n\n").await;
|
|
let _ = sock.flush().await;
|
|
})
|
|
.await;
|
|
let stream = forward_stream(
|
|
&minimal_client(),
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
let envelopes = collect_stream(stream).await;
|
|
assert_eq!(
|
|
envelopes.len(),
|
|
1,
|
|
"the parse-overflow terminal envelope is the only output"
|
|
);
|
|
match &envelopes[0].result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
|
|
assert!(
|
|
err.message.contains("SSE parse error"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected one INTERNAL terminal envelope, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// COV-10 (b): the responder emits one complete `data:` frame, lets
|
|
/// the client consume it, then kills the connection before the
|
|
/// declared body length is met — a genuine mid-stream transport
|
|
/// abort, not a graceful EOF. The delivered event passes, a single
|
|
/// `INTERNAL` terminal error envelope follows, the stream ends.
|
|
#[tokio::test]
|
|
async fn aborted_socket_mid_stream_emits_a_terminal_error_envelope() {
|
|
let head =
|
|
"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: 512\r\n\r\n";
|
|
let gate = TestArc::new(tokio::sync::Notify::new());
|
|
let notify = TestArc::clone(&gate);
|
|
let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
|
|
use tokio::io::AsyncWriteExt;
|
|
let _ = sock.write_all(b"data: {\"n\":1}\n\n").await;
|
|
let _ = sock.flush().await;
|
|
notify.notified().await;
|
|
drop(sock);
|
|
})
|
|
.await;
|
|
let stream = forward_stream(
|
|
&minimal_client(),
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
tokio::pin!(stream);
|
|
let first = tokio::time::timeout(Duration::from_secs(5), stream.next())
|
|
.await
|
|
.expect("the complete frame must deliver within the test budget")
|
|
.expect("stream must deliver the frame before the abort");
|
|
assert_eq!(first.result.clone().unwrap(), json!({"n": 1}));
|
|
gate.notify_one();
|
|
let terminal = tokio::time::timeout(Duration::from_secs(5), stream.next())
|
|
.await
|
|
.expect("the terminal envelope must follow within the test budget")
|
|
.expect("the stream must end with the terminal envelope");
|
|
match terminal.result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
|
|
assert!(
|
|
err.message.contains("SSE stream error"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected the terminal error envelope, got {other:?}"),
|
|
}
|
|
let ended = tokio::time::timeout(Duration::from_secs(1), stream.next()).await;
|
|
assert!(
|
|
matches!(ended, Ok(None) | Err(_)),
|
|
"the stream ends after the terminal envelope"
|
|
);
|
|
}
|
|
|
|
/// COV-10 (c): the responder ends the stream with a pending event
|
|
/// (data lines received, no trailing blank line) — the WHATWG
|
|
/// EOF-flush dispatches that pending event as a final success
|
|
/// envelope at EOF.
|
|
#[tokio::test]
|
|
async fn pending_event_flushes_at_eof_without_a_trailing_blank_line() {
|
|
let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
|
|
let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
|
|
use tokio::io::AsyncWriteExt;
|
|
let _ = sock
|
|
.write_all(b"data: {\"n\":1}\n\ndata: {\"final\":true}\n")
|
|
.await;
|
|
let _ = sock.flush().await;
|
|
let _ = sock.shutdown().await;
|
|
})
|
|
.await;
|
|
let stream = forward_stream(
|
|
&minimal_client(),
|
|
&base,
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
let envelopes = collect_stream(stream).await;
|
|
assert_eq!(envelopes.len(), 2);
|
|
assert_eq!(envelopes[0].result.clone().unwrap(), json!({"n": 1}));
|
|
assert_eq!(
|
|
envelopes[1].result.clone().unwrap(),
|
|
json!({"final": true}),
|
|
"EOF-flush dispatches the pending event"
|
|
);
|
|
}
|
|
|
|
/// COV-10 (d), Once-path transport arm: a connect-refused upstream
|
|
/// surfaces a single `INTERNAL` error envelope through `forward`.
|
|
#[tokio::test]
|
|
async fn dead_port_forward_yields_an_internal_envelope() {
|
|
let envelope = forward(
|
|
&minimal_client(),
|
|
"http://127.0.0.1:9",
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
)
|
|
.await;
|
|
match envelope.result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
|
|
assert!(
|
|
err.message.contains("HTTP request failed"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected INTERNAL transport envelope, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// COV-10 (d), streaming-path transport arm: the same dead upstream
|
|
/// through `forward_stream` surfaces a single terminal `INTERNAL`
|
|
/// error envelope (the streaming build/send-failure terminal arm).
|
|
#[tokio::test]
|
|
async fn dead_port_forward_stream_yields_a_terminal_internal_envelope() {
|
|
let stream = forward_stream(
|
|
&minimal_client(),
|
|
"http://127.0.0.1:9",
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&serde_json::json!({"type": "object"}),
|
|
&[],
|
|
json!({}),
|
|
noop_context(),
|
|
);
|
|
let envelopes = collect_stream(stream).await;
|
|
assert_eq!(
|
|
envelopes.len(),
|
|
1,
|
|
"exactly one terminal envelope, then the stream ends"
|
|
);
|
|
match &envelopes[0].result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
|
|
assert!(
|
|
err.message.contains("HTTP request failed"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected terminal INTERNAL envelope, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// COV-12, loud-credential arm (ApiKey): an ApiKey credential value
|
|
/// carrying invalid HTTP header bytes fails loudly and never echoes
|
|
/// the credential material (FWD-08 family).
|
|
#[tokio::test]
|
|
async fn api_key_with_invalid_value_fails_loudly_without_echoing_secrets() {
|
|
let base = spawn_responder(TestArc::new(|_parts| {
|
|
http_response(200, "application/json", b"{}".to_vec())
|
|
}))
|
|
.await;
|
|
let ctx = ctx_with_capability("svc", "key\u{0003}-secret-marker".to_string());
|
|
let envelope = call_forward_authed(
|
|
&base,
|
|
ctx,
|
|
&Some(HttpAuthScheme::ApiKey {
|
|
header_name: "x-api-key".to_string(),
|
|
}),
|
|
)
|
|
.await;
|
|
match envelope.result {
|
|
Err(err) => {
|
|
assert!(
|
|
err.message
|
|
.contains("refusing to send the request unauthenticated"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
assert!(
|
|
!err.message.contains("secret-marker") && !err.message.contains("key\u{0003}"),
|
|
"error must not echo credential material: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected loud credential error, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// COV-12, loud-credential arm (Basic): a Basic credential pair
|
|
/// with invalid HTTP header bytes fails loudly like Bearer/ApiKey
|
|
/// (FWD-08 family).
|
|
#[tokio::test]
|
|
async fn basic_credential_with_invalid_value_fails_loudly_without_echoing_secrets() {
|
|
let base = spawn_responder(TestArc::new(|_parts| {
|
|
http_response(200, "application/json", b"{}".to_vec())
|
|
}))
|
|
.await;
|
|
let ctx = ctx_with_capability("svc", "basic\u{0001}-secret-marker".to_string());
|
|
let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Basic)).await;
|
|
match envelope.result {
|
|
Err(err) => {
|
|
assert!(
|
|
err.message
|
|
.contains("refusing to send the request unauthenticated"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
assert!(
|
|
!err.message.contains("secret-marker")
|
|
&& !err.message.contains("basic\u{0001}"),
|
|
"error must not echo credential material: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("expected loud credential error, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// COV-12: a declared header parameter with an invalid *value*
|
|
/// (control bytes) is rejected loudly at build time rather than
|
|
/// silently dropped from the request (FWD-08 family).
|
|
#[test]
|
|
fn declared_header_param_with_invalid_value_is_rejected() {
|
|
let ctx = noop_context();
|
|
let schema = json!({
|
|
"type": "object",
|
|
"properties": {"X-Trace": {"type": "string", "wire": "header"}},
|
|
});
|
|
let err = build_request(
|
|
"https://api.example.com",
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&schema,
|
|
&json!({"X-Trace": "bad\u{0000}value"}),
|
|
&ctx,
|
|
)
|
|
.expect_err("invalid header-param value must fail loudly");
|
|
assert!(
|
|
err.message.contains("X-Trace") && err.message.contains("invalid value"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
|
|
/// COV-12: a declared header parameter whose name is not a valid
|
|
/// HTTP header name is rejected loudly at build time (FWD-08
|
|
/// family, the header-param sibling of the default-header arm).
|
|
#[test]
|
|
fn declared_header_param_with_invalid_name_is_rejected() {
|
|
let ctx = noop_context();
|
|
let schema = json!({
|
|
"type": "object",
|
|
"properties": {"bad header": {"type": "string", "wire": "header"}},
|
|
});
|
|
let err = build_request(
|
|
"https://api.example.com",
|
|
"/x",
|
|
"GET",
|
|
&None,
|
|
&TestHashMap::new(),
|
|
"svc",
|
|
&schema,
|
|
&json!({"bad header": "v"}),
|
|
&ctx,
|
|
)
|
|
.expect_err("invalid header-param name must fail loudly");
|
|
assert!(
|
|
err.message.contains("valid HTTP header name"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
|
|
/// CLI-02 (a): a same-host 302 is followed, and the credential
|
|
/// header is forwarded to the followed hop — the FWD-03 policy only
|
|
/// scrubs headers across *cross-host* hops, so a same-origin hop
|
|
/// must stay authenticated end to end.
|
|
#[tokio::test]
|
|
async fn same_host_redirect_is_followed_with_credential_forwarding() {
|
|
let hits = TestArc::new(std::sync::atomic::AtomicU32::new(0));
|
|
let hit_counter = TestArc::clone(&hits);
|
|
let base = spawn_responder(TestArc::new(move |parts| {
|
|
hit_counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
|
if parts.target == "/x" {
|
|
http::Response::builder()
|
|
.status(302)
|
|
.header("location", "/final")
|
|
.body(Vec::new())
|
|
.expect("redirect response builds")
|
|
} else {
|
|
let auth = parts
|
|
.headers
|
|
.get("authorization")
|
|
.cloned()
|
|
.unwrap_or_default();
|
|
let body = format!(r#"{{"auth":"{auth}"}}"#);
|
|
http_response(200, "application/json", body.into_bytes())
|
|
}
|
|
}))
|
|
.await;
|
|
let ctx = ctx_with_capability("svc", "tok-secret-marker".to_string());
|
|
let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Bearer)).await;
|
|
match envelope.result {
|
|
Ok(value) => assert_eq!(
|
|
value,
|
|
json!({"auth": "Bearer tok-secret-marker"}),
|
|
"the followed hop must receive the credential header"
|
|
),
|
|
other => panic!("same-host redirect must be followed to success, got {other:?}"),
|
|
}
|
|
assert_eq!(
|
|
hits.load(std::sync::atomic::Ordering::SeqCst),
|
|
2,
|
|
"origin request plus exactly one followed hop"
|
|
);
|
|
}
|
|
|
|
/// CLI-02 (b), the FWD-03 load-bearing property: a cross-host 302 is
|
|
/// surfaced to the caller as `HTTP_302` and the redirected-to host
|
|
/// receives zero requests — credential-bearing or not, no second
|
|
/// request ever leaves for another host.
|
|
#[tokio::test]
|
|
async fn cross_host_redirect_is_surfaced_and_the_target_receives_zero_requests() {
|
|
let attacker_hits = TestArc::new(std::sync::atomic::AtomicU32::new(0));
|
|
let attacker_counter = TestArc::clone(&attacker_hits);
|
|
let attacker = spawn_responder(TestArc::new(move |_| {
|
|
attacker_counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
|
http_response(200, "text/plain", b"stolen".to_vec())
|
|
}))
|
|
.await;
|
|
let base = spawn_responder(TestArc::new(move |_| {
|
|
http::Response::builder()
|
|
.status(302)
|
|
.header("location", format!("{attacker}/steal"))
|
|
.body(Vec::new())
|
|
.expect("redirect response builds")
|
|
}))
|
|
.await;
|
|
let ctx = ctx_with_capability("svc", "tok-secret-marker".to_string());
|
|
let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Bearer)).await;
|
|
match envelope.result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "HTTP_302", "message was: {}", err.message);
|
|
assert!(
|
|
!err.message.contains("tok-secret-marker"),
|
|
"error must carry no credential material: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("cross-host redirect must surface the 3xx, got {other:?}"),
|
|
}
|
|
assert_eq!(
|
|
attacker_hits.load(std::sync::atomic::Ordering::SeqCst),
|
|
0,
|
|
"the cross-host target must receive zero requests"
|
|
);
|
|
}
|
|
|
|
/// CLI-02 (c): a same-host redirect loop runs into the hop cap and
|
|
/// fails loudly as an `INTERNAL` transport error naming the
|
|
/// redirect machinery — never silently looping or surfacing a
|
|
/// partial body.
|
|
#[tokio::test]
|
|
async fn redirect_hop_cap_errors_loudly() {
|
|
let hits = TestArc::new(std::sync::atomic::AtomicU32::new(0));
|
|
let hit_counter = TestArc::clone(&hits);
|
|
let base = spawn_responder(TestArc::new(move |_| {
|
|
hit_counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
|
http::Response::builder()
|
|
.status(302)
|
|
.header("location", "/next")
|
|
.body(Vec::new())
|
|
.expect("redirect response builds")
|
|
}))
|
|
.await;
|
|
let envelope = call_forward(&base, noop_context()).await;
|
|
match envelope.result {
|
|
Err(err) => {
|
|
assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
|
|
assert!(
|
|
err.message.contains("HTTP request failed"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
assert!(
|
|
err.message.contains("redirect"),
|
|
"message was: {}",
|
|
err.message
|
|
);
|
|
}
|
|
other => panic!("hop-cap redirect must error loudly, got {other:?}"),
|
|
}
|
|
let hit_count = hits.load(std::sync::atomic::Ordering::SeqCst);
|
|
assert!(
|
|
hit_count >= 2,
|
|
"the chain followed before capping: {hit_count}"
|
|
);
|
|
assert!(
|
|
hit_count <= 12,
|
|
"the hop cap bounded the redirect chain: {hit_count}"
|
|
);
|
|
}
|
|
}
|