diff --git a/services/core/src/independent_966_log_tests.rs b/services/core/src/independent_966_log_tests.rs new file mode 100644 index 000000000..4960f67ac --- /dev/null +++ b/services/core/src/independent_966_log_tests.rs @@ -0,0 +1,328 @@ +use crate::log_redact::{access_log_line, sanitize_url_for_log}; + +fn log_target(target: &str) -> String { + access_log_line("GET", 401, target) + .expect("a query-bearing request still has an access diagnostic") +} + +fn assert_no_secret(output: &str, secrets: &[&str]) { + for secret in secrets { + assert!( + !output.contains(secret), + "credential {secret:?} escaped redaction: {output}" + ); + } + let target = output + .split_once(" target=") + .map_or(output, |(_, target)| target); + let absolute = if target.starts_with('/') { + format!("https://example.invalid{target}") + } else { + target.to_owned() + }; + if let Ok(url) = reqwest::Url::parse(&absolute) { + for (key, value) in url.query_pairs() { + for secret in secrets { + assert!( + !key.contains(secret) && !value.contains(secret), + "credential remained after normal URL decoding: {output}" + ); + } + } + } +} + +fn assert_single_line(output: &str) { + assert!( + !output + .chars() + .any(|c| c.is_control() || matches!(c, '\u{2028}' | '\u{2029}')), + "log output contains a line/control separator: {output:?}" + ); +} + +#[test] +fn independent_966_actual_and_other_oauth_paths_never_log_code_or_state() { + for path in [ + "/core/v1/auth/oauth/google/callback", + "/api/oauth2/authorize", + "/oauth/callback", + "/oauth", + "/api/oauth", + "/unrecognized-login/callback", + ] { + for status in [200, 401, 404, 500] { + let target = format!("{path}?code=synthetic-code-966&state=synthetic-state-966"); + let line = access_log_line("GET", status, &target).expect("access diagnostic"); + assert_no_secret(&line, &["synthetic-code-966", "synthetic-state-966"]); + assert_single_line(&line); + } + } +} + +#[test] +fn independent_966_decoded_and_duplicate_credential_values_are_hidden() { + let line = log_target( + "/v1/models?%6bey=synthetic-first-966&KEY=synthetic-second-966&api%5Fkey=synthetic-third-966&+token+=synthetic-fourth-966&authorization=synthetic-fifth-966", + ); + assert_no_secret( + &line, + &[ + "synthetic-first-966", + "synthetic-second-966", + "synthetic-third-966", + "synthetic-fourth-966", + "synthetic-fifth-966", + ], + ); +} + +#[test] +fn independent_966_unknown_query_values_and_bare_query_credentials_are_hidden() { + for target in [ + "/v1?credential=synthetic-credential-966", + "/v1?authz=synthetic-authz-966", + "/v1?unrecognized=synthetic-value-966", + "/v1?synthetic-bare-bearer-966", + "/v1?synthetic-secret-key-966=unused", + ] { + let line = log_target(target); + assert_no_secret( + &line, + &[ + "synthetic-credential-966", + "synthetic-authz-966", + "synthetic-value-966", + "synthetic-bare-bearer-966", + "synthetic-secret-key-966", + ], + ); + } +} + +#[test] +fn independent_966_invalid_encoding_and_parse_failures_do_not_return_credentials() { + for target in [ + "/v1?key=synthetic-bad-966%ZZ", + "/v1?credential=synthetic-bad-966%", + "/v1?%ZZ=synthetic-bad-966", + "/v1?key=synthetic-bad-966;alt=sse", + "/bad%ZZ?key=synthetic-bad-966", + "/v1?key=synthetic-bad-966%FF", + ] { + assert_no_secret(&log_target(target), &["synthetic-bad-966"]); + let absolute = format!("https://example.invalid{target}"); + assert_no_secret(&sanitize_url_for_log(&absolute), &["synthetic-bad-966"]); + } + for malformed in [ + "https://example.invalid/v1?key=synthetic-bad-966\n", + "not-a-valid-url synthetic-bad-966", + ] { + let output = sanitize_url_for_log(malformed); + assert_no_secret(&output, &["synthetic-bad-966"]); + assert_single_line(&output); + } +} + +#[test] +fn independent_966_absolute_url_userinfo_and_fragment_are_not_log_data() { + let output = sanitize_url_for_log( + "https://synthetic-user-966:synthetic-password-966@example.invalid/v1?alt=sse#access_token=synthetic-fragment-966", + ); + assert_no_secret( + &output, + &[ + "synthetic-user-966", + "synthetic-password-966", + "synthetic-fragment-966", + ], + ); + if let Ok(url) = reqwest::Url::parse(&output) { + assert!(url.username().is_empty()); + assert!(url.password().is_none()); + assert!(url.fragment().is_none()); + } + assert_single_line(&output); +} + +#[test] +fn independent_966_paths_and_methods_cannot_inject_a_log_record() { + for target in [ + "/oauth/a%0D%0Ainjected?key=synthetic-key-966", + "/oauth/a%E2%80%A8injected?code=synthetic-key-966", + "/oauth/a%E2%80%A9injected?code=synthetic-key-966", + "/v1/a%20b?key=synthetic-key-966", + ] { + let output = access_log_line("GET\r\ninjected", 404, target).expect("access diagnostic"); + assert_single_line(&output); + assert_no_secret(&output, &["synthetic-key-966"]); + assert!(!output.contains("a b")); + } + let output = sanitize_url_for_log( + "https://example.invalid/oauth/a%E2%80%A8injected?code=synthetic-key-966", + ); + assert_single_line(&output); + assert_no_secret(&output, &["synthetic-key-966"]); +} + +#[test] +fn independent_966_useful_bounded_operational_values_remain_visible() { + let output = log_target( + "/v1/models?model=gpt-test-966&alt=sse&api-version=2024-02-01&stream=true&key=synthetic-key-966", + ); + for operational in ["gpt-test-966", "sse", "2024-02-01", "true"] { + assert!( + output.contains(operational), + "lost operational value: {output}" + ); + } + assert_no_secret(&output, &["synthetic-key-966"]); + assert!(output.starts_with("core_access method=GET status=401 target=")); + assert!(access_log_line("GET", 200, "/health/live").is_none()); +} + +#[test] +fn independent_966_truncation_is_bounded_after_redaction() { + let target = format!( + "/{}?key=synthetic-tail-key-966&unrecognized=synthetic-tail-value-966", + "δΈ­".repeat(800) + ); + let line = log_target(&target); + assert_no_secret( + &line, + &["synthetic-tail-key-966", "synthetic-tail-value-966"], + ); + assert_single_line(&line); + // The original target contract was 1024 characters. A byte-bounded target + // plus its fixed method/status prefix fits within this conservative margin. + assert!(line.len() <= 1152, "unbounded log bytes: {}", line.len()); +} + +#[test] +fn independent_966_real_middleware_redacts_loopback_access_logs() { + const CHILD_ENV: &str = "LMM_TEST_20261027_PR966_MIDDLEWARE_CHILD"; + const CHILD_MARKER: &str = "real-axum-loopback-fixture"; + const COMPLETED: &str = "independent_966_real_middleware_http_requests_completed"; + + if std::env::var(CHILD_ENV).as_deref() == Ok(CHILD_MARKER) { + use axum::{Router, http::StatusCode, middleware, routing::get}; + use std::time::Duration; + + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let address = listener.local_addr().unwrap(); + let app = Router::new() + .route( + "/core/v1/auth/oauth/google/callback", + get(|| async { (StatusCode::UNAUTHORIZED, "fixture callback denied") }), + ) + .route("/v1/fixture-ok", get(|| async { "fixture success" })) + .route("/health/live", get(|| async { "fixture ready" })) + .layer(middleware::from_fn( + crate::log_redact::log_redacted_request_target, + )); + let (shutdown, stopped) = tokio::sync::oneshot::channel::<()>(); + let server = tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(async move { + stopped.await.unwrap(); + }) + .await + .unwrap(); + }); + let client = reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .timeout(Duration::from_secs(3)) + .build() + .unwrap(); + for (path, status, body) in [ + ( + "/core/v1/auth/oauth/google/callback?code=synthetic-runtime-code-401&state=synthetic-runtime-state-401&unknown=synthetic-runtime-value-401&synthetic-runtime-key-401=unused&model=gpt-runtime-966", + StatusCode::UNAUTHORIZED, + Some("fixture callback denied"), + ), + ( + "/unregistered-966?code=synthetic-runtime-code-404&state=synthetic-runtime-state-404&unknown=synthetic-runtime-value-404&synthetic-runtime-key-404=unused&model=gpt-runtime-966", + StatusCode::NOT_FOUND, + None, + ), + ( + "/v1/fixture-ok?code=synthetic-runtime-code-200&state=synthetic-runtime-state-200&unknown=synthetic-runtime-value-200&synthetic-runtime-key-200=unused&model=gpt-runtime-966", + StatusCode::OK, + Some("fixture success"), + ), + ("/health/live", StatusCode::OK, Some("fixture ready")), + ] { + let response = client + .get(format!("http://{address}{path}")) + .header(reqwest::header::CONNECTION, "close") + .send() + .await + .unwrap(); + assert_eq!(response.status(), status); + let actual_body = response.text().await.unwrap(); + if let Some(expected_body) = body { + assert_eq!(actual_body, expected_body); + } + } + drop(client); + shutdown.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(3), server) + .await + .unwrap() + .unwrap(); + }); + println!("\n{COMPLETED}"); + return; + } + + // Libtest omits the crate name from its exact test path. The completion + // marker below also prevents a mistyped filter from yielding a false pass. + let full_name = concat!( + module_path!(), + "::independent_966_real_middleware_redacts_loopback_access_logs" + ); + let (_, exact_name) = full_name.split_once("::").unwrap(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", exact_name, "--nocapture", "--test-threads=1"]) + .env_clear() + .env(CHILD_ENV, CHILD_MARKER) + .output() + .unwrap(); + let stdout = String::from_utf8(output.stdout).unwrap(); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!( + output.status.success(), + "middleware child failed: {}\nstdout:\n{stdout}\nstderr:\n{stderr}", + output.status + ); + assert!( + stdout.lines().any(|line| line == COMPLETED), + "child did not execute the HTTP fixture:\n{stdout}\n{stderr}" + ); + assert_no_secret(&stderr, &["synthetic-runtime-"]); + let lines: Vec<_> = stderr + .lines() + .filter(|line| line.starts_with("core_access ")) + .collect(); + assert_eq!( + lines, + [ + "core_access method=GET status=401 target=/core/v1/auth/oauth/google/callback?model=gpt-runtime-966", + "core_access method=GET status=404 target=/unregistered-966?model=gpt-runtime-966", + "core_access method=GET status=200 target=/v1/fixture-ok?model=gpt-runtime-966", + ], + "query-less health request must remain silent; each real query request needs its final status" + ); + for line in lines { + assert_single_line(line); + } + eprintln!("captured real middleware child stderr:\n{stderr}"); +} diff --git a/services/core/src/lib.rs b/services/core/src/lib.rs index fcc3d41ff..be9d6ad9a 100644 --- a/services/core/src/lib.rs +++ b/services/core/src/lib.rs @@ -29,6 +29,7 @@ pub mod identity_http; pub mod internal_rpc; pub mod legacy_auth; pub mod lifecycle; +pub mod log_redact; pub mod market_oauth; pub mod model_http; pub mod native_key_read; @@ -53,3 +54,6 @@ pub mod usage; pub mod wallet; pub mod humangrant; + +#[cfg(test)] +mod independent_966_log_tests; diff --git a/services/core/src/log_redact.rs b/services/core/src/log_redact.rs new file mode 100644 index 000000000..42c75f5d1 --- /dev/null +++ b/services/core/src/log_redact.rs @@ -0,0 +1,340 @@ +//! A conservative request-target log derived from QuantumNous/new-api +//! `30748c6558bc97de49ff74871cfec717e50f3322`. +//! +//! Only bounded `model`, `alt`, `api-version` and `stream` values are retained. +//! Every other query member, including its key, is omitted: OAuth codes, +//! state, unknown credential names and bare query tokens are never logged. +//! Parsing uses the existing reqwest::Url implementation. Invalid input loses +//! the complete target rather than returning unredacted bytes. URL userinfo +//! and fragments are always removed, and path escaping is preserved. + +use axum::{extract::Request, http::Uri, middleware::Next, response::Response}; +use reqwest::Url; + +const MAX_URL_BYTES: usize = 16_384; +const MAX_TARGET_BYTES: usize = 1024; +const REDACTED: &str = "[redacted-target]"; +const ORIGIN: &str = "https://redacted.invalid"; + +/// Known credential names; the logger also omits all unlisted query members. +pub fn is_sensitive_query_key(key: &str) -> bool { + let key = key.trim().to_ascii_lowercase(); + matches!( + key.as_str(), + "key" + | "api_key" + | "api-key" + | "apikey" + | "x-api-key" + | "authorization" + | "auth" + | "password" + | "passwd" + | "sig" + | "awsaccesskeyid" + | "x-amz-credential" + | "code" + | "state" + ) || key.contains("token") + || key.contains("secret") + || key.contains("signature") +} + +/// Remove userinfo, fragments and all query members outside the finite +/// operational allowlist. Malformed or oversized input returns a fixed marker. +pub fn sanitize_url_for_log(raw: &str) -> String { + if raw.is_empty() { + return String::new(); + } + if !valid_input(raw) { + return REDACTED.to_owned(); + } + if raw.starts_with('?') { + return sanitize(raw, Form::Query); + } + if raw.starts_with("//") { + return sanitize(raw, Form::Network); + } + if raw.starts_with('/') { + return sanitize(raw, Form::Target); + } + sanitize(raw, Form::Absolute) +} + +/// Log an origin-form HTTP target. Unknown/credential query keys and values +/// are omitted on every path, including OAuth callbacks and unknown routes. +pub fn logged_request_target(target: &str) -> String { + if target == "*" { + return "*".to_owned(); + } + if !target.starts_with('/') || !valid_input(target) { + return REDACTED.to_owned(); + } + sanitize(target, Form::Target) +} + +/// A single bounded stderr line; ordinary query-less requests remain quiet. +pub fn access_log_line(method: &str, status: u16, target: &str) -> Option { + if !target.contains('?') { + return None; + } + Some(format_line(method, status, &logged_request_target(target))) +} + +/// Redact before moving the request into the handler. Only the already-safe +/// target is retained while waiting for the final response status. +pub async fn log_redacted_request_target(request: Request, next: Next) -> Response { + let method = safe_method(request.method().as_str()).to_owned(); + let target = request.uri().path_and_query().and_then(|target| { + request + .uri() + .query() + .map(|_| logged_request_target(target.as_str())) + }); + let response = next.run(request).await; + if let Some(target) = target { + eprintln!( + "{}", + format_line(&method, response.status().as_u16(), &target) + ); + } + response +} + +fn safe_method(method: &str) -> &str { + match method { + "GET" | "POST" | "PUT" | "PATCH" | "DELETE" | "HEAD" | "OPTIONS" | "CONNECT" | "TRACE" => { + method + } + _ => "OTHER", + } +} + +fn format_line(method: &str, status: u16, target: &str) -> String { + let method = safe_method(method); + format!("core_access method={method} status={status} target={target}") +} + +#[derive(Clone, Copy)] +enum Form { + Target, + Absolute, + Network, + Query, +} + +fn valid_input(raw: &str) -> bool { + if raw.len() > MAX_URL_BYTES || raw.chars().any(char::is_control) { + return false; + } + let bytes = raw.as_bytes(); + let mut at = 0; + while at < bytes.len() { + if bytes[at] == b'%' { + if bytes.get(at + 1).is_none_or(|b| !b.is_ascii_hexdigit()) + || bytes.get(at + 2).is_none_or(|b| !b.is_ascii_hexdigit()) + { + return false; + } + at += 3; + } else { + at += 1; + } + } + true +} + +fn sanitize(raw: &str, form: Form) -> String { + let original_path = if matches!(form, Form::Target) { + let Ok(uri) = raw.parse::() else { + return REDACTED.to_owned(); + }; + if raw.contains('#') { + return REDACTED.to_owned(); + } + Some(uri.path().to_owned()) + } else { + None + }; + let input = match form { + Form::Target => format!("{ORIGIN}{raw}"), + Form::Query => format!("{ORIGIN}/{raw}"), + Form::Network => format!("https:{raw}"), + Form::Absolute => raw.to_owned(), + }; + let Ok(mut url) = Url::parse(&input) else { + return REDACTED.to_owned(); + }; + if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() { + return REDACTED.to_owned(); + } + if url.set_username("").is_err() || url.set_password(None).is_err() { + return REDACTED.to_owned(); + } + url.set_fragment(None); + let mut pairs: Vec<_> = url + .query_pairs() + .filter(|(key, value)| allowed_pair(key, value)) + .map(|(key, value)| (key.into_owned(), value.into_owned())) + .collect(); + pairs.sort_by(|left, right| left.0.cmp(&right.0)); + url.set_query(None); + if !pairs.is_empty() { + url.query_pairs_mut().extend_pairs(pairs); + } + let target = match form { + Form::Absolute => url.to_string(), + Form::Network => url + .as_str() + .strip_prefix("https:") + .unwrap_or(REDACTED) + .to_owned(), + Form::Target => { + // Keep the actual escaped HTTP path, including dot segments; + // the URL parser's normalization must not rewrite log evidence. + let mut target = original_path + .filter(|path| path.is_ascii()) + .unwrap_or_else(|| url.path().to_owned()); + if let Some(query) = url.query() { + target.push('?'); + target.push_str(query); + } + target + } + Form::Query => url + .query() + .map(|query| format!("?{query}")) + .unwrap_or_default(), + }; + bounded(target) +} + +fn allowed_pair(key: &str, value: &str) -> bool { + if is_sensitive_query_key(key) { + return false; + } + match key { + "alt" => matches!(value, "json" | "sse" | "media"), + "stream" => matches!(value, "true" | "false" | "0" | "1"), + "model" => { + !value.is_empty() + && value.len() <= 128 + && value + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"-_.:/".contains(&b)) + } + "api-version" => { + let bytes = value.as_bytes(); + bytes.len() >= 10 + && matches!(bytes.len(), 10 | 18) + && bytes[..10].iter().enumerate().all(|(index, b)| { + if matches!(index, 4 | 7) { + *b == b'-' + } else { + b.is_ascii_digit() + } + }) + && (bytes.len() == 10 || &value[10..] == "-preview") + } + _ => false, + } +} + +fn bounded(mut target: String) -> String { + const SUFFIX: &str = "...[truncated]"; + if target.len() <= MAX_TARGET_BYTES { + return target; + } + let mut end = MAX_TARGET_BYTES - SUFFIX.len(); + while !target.is_char_boundary(end) { + end -= 1; + } + if target.as_bytes()[end - 1] == b'%' { + end -= 1; + } else if target.as_bytes()[end - 2] == b'%' { + end -= 2; + } + target.truncate(end); + target.push_str(SUFFIX); + target +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn credentials_and_unknown_members_are_omitted_on_every_path() { + for path in [ + "/v1/models", + "/oauth", + "/oauth2/callback", + "/core/v1/auth/oauth/google/callback", + "/unknown", + ] { + let target = format!( + "{path}?key=synthetic-key&code=synthetic-code&state=synthetic-state&unknown=synthetic-unknown&synthetic-bare-token&model=gpt-5&alt=sse" + ); + for status in [200, 401, 404, 500] { + let line = access_log_line("GET", status, &target).unwrap(); + assert!(!line.contains("synthetic"), "{line}"); + assert!(line.contains("alt=sse&model=gpt-5"), "{line}"); + } + } + assert!(access_log_line("GET", 200, "/health/live").is_none()); + } + + #[test] + fn absolute_urls_drop_userinfo_fragments_and_invalid_input() { + assert_eq!( + sanitize_url_for_log( + "https://synthetic-user:synthetic-pass@example.test/v1?key=synthetic-key&stream=true#synthetic-fragment" + ), + "https://example.test/v1?stream=true" + ); + for raw in [ + "https://example.test/?key=synthetic%ZZ", + "/v1?key=synthetic\n", + "/v1?model=synthetic%", + "not a url?key=synthetic", + ] { + assert_eq!(sanitize_url_for_log(raw), REDACTED); + } + assert_eq!(sanitize_url_for_log("?state=synthetic"), ""); + assert_eq!( + sanitize_url_for_log("//user:synthetic@example.test/v1?password=synthetic"), + "//example.test/v1" + ); + } + + #[test] + fn operational_values_are_bounded_and_path_remains_escaped() { + assert_eq!( + logged_request_target( + "/core/v1/auth/oauth/google/callback%20%E2%80%A8?code=synthetic&state=synthetic&api-version=2024-02-01" + ), + "/core/v1/auth/oauth/google/callback%20%E2%80%A8?api-version=2024-02-01" + ); + assert_eq!( + logged_request_target("//v1%2Fmodels?alt=sse"), + "//v1%2Fmodels?alt=sse" + ); + assert_eq!( + logged_request_target( + "/v1?model=one%26key%3Dsynthetic&stream=maybe&alt=other&client_id=synthetic" + ), + "/v1" + ); + let line = + access_log_line("GET", 401, &format!("/{}?code=synthetic", "a".repeat(2000))).unwrap(); + assert!(line.ends_with("...[truncated]")); + assert!(line.len() <= MAX_TARGET_BYTES + 64); + assert!(!line.contains("synthetic")); + for separator in ['\u{2028}', '\u{2029}'] { + let line = + access_log_line("GET", 200, &format!("/oauth/{separator}?code=synthetic")).unwrap(); + assert!(!line.contains(separator)); + assert!(!line.contains("synthetic")); + } + } +} diff --git a/services/core/src/main.rs b/services/core/src/main.rs index 31ca9c18d..deee212b8 100644 --- a/services/core/src/main.rs +++ b/services/core/src/main.rs @@ -631,6 +631,10 @@ async fn run_core() -> Result<(), Box> { lmm_core::remote_control::preadmit, )); } + // Outermost: see the final status, and mask query credentials before stderr. + app = app.layer(axum::middleware::from_fn( + lmm_core::log_redact::log_redacted_request_target, + )); // All enabled components and listener binds have succeeded. The health body // lists capabilities; model readiness has its own explicit endpoint. assert!(lifecycle.mark_ready()); diff --git a/services/core/src/relay/honeypot.rs b/services/core/src/relay/honeypot.rs index b216a6d80..e0173c016 100644 --- a/services/core/src/relay/honeypot.rs +++ b/services/core/src/relay/honeypot.rs @@ -565,9 +565,9 @@ impl Worker { } else { None }; - let payload = self.request.upstream_to(route.protocol, &route.model)?; - let bytes = serde_json::to_vec(&payload) - .map_err(|_| RelayError::Invalid("request encoding failed"))?; + let bytes = self + .request + .upstream_body_to(route.protocol, &route.model)?; if bytes.len() > self.limits.request_bytes { return Err(RelayError::Limit("encoded request")); } diff --git a/services/core/src/relay/independent_966_schema_tests.rs b/services/core/src/relay/independent_966_schema_tests.rs new file mode 100644 index 000000000..f04d46e65 --- /dev/null +++ b/services/core/src/relay/independent_966_schema_tests.rs @@ -0,0 +1,490 @@ +use super::{ClientProtocol, Limits, RelayError, Request, Route, UpstreamProtocol}; +use serde::{ + Deserialize, + de::{IgnoredAny, MapAccess, Visitor}, +}; +use serde_json::{Value, json, value::RawValue}; +use sha2::{Digest, Sha256}; +use std::{collections::BTreeMap, time::Instant}; + +const ZA: &str = + r#"{"type":"object","properties":{"zeta":{"type":"string"},"alpha":{"type":"string"}}}"#; +const AZ: &str = + r#"{"type":"object","properties":{"alpha":{"type":"string"},"zeta":{"type":"string"}}}"#; + +struct ObjectKeys(Vec); +impl<'de> Deserialize<'de> for ObjectKeys { + fn deserialize>(d: D) -> Result { + struct KeysVisitor; + impl<'de> Visitor<'de> for KeysVisitor { + type Value = ObjectKeys; + fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("an object whose key order is observable") + } + fn visit_map>(self, mut map: A) -> Result { + let mut keys = Vec::new(); + while let Some(key) = map.next_key::()? { + keys.push(key); + map.next_value::()?; + } + Ok(ObjectKeys(keys)) + } + } + d.deserialize_map(KeysVisitor) + } +} + +fn raw_at(body: &[u8], path: &[&str]) -> String { + let mut raw = std::str::from_utf8(body).unwrap().to_owned(); + for segment in path { + if raw.trim_start().starts_with('[') { + let values: Vec> = serde_json::from_str(&raw).unwrap(); + raw = values[segment.parse::().unwrap()].get().to_owned(); + } else { + let values: BTreeMap> = serde_json::from_str(&raw).unwrap(); + raw = values[*segment].get().to_owned(); + } + } + raw +} + +fn keys_at(body: &[u8], path: &[&str]) -> Vec { + serde_json::from_str::(&raw_at(body, path)) + .unwrap() + .0 +} + +fn route(protocol: UpstreamProtocol) -> Route { + Route::new( + "synthetic-966".into(), + protocol, + "https://provider.invalid/v1", + "wire-model".into(), + "synthetic-key", + 1, + ) + .unwrap() +} + +fn tool_path(protocol: UpstreamProtocol, index: usize) -> Vec { + let mut path: Vec = match protocol { + UpstreamProtocol::Chat => vec!["tools", "", "function", "parameters"], + UpstreamProtocol::Responses => vec!["tools", "", "parameters"], + UpstreamProtocol::Anthropic => vec!["tools", "", "input_schema"], + UpstreamProtocol::Gemini => { + return vec![ + "tools".into(), + "0".into(), + "functionDeclarations".into(), + index.to_string(), + "parametersJsonSchema".into(), + ]; + } + } + .into_iter() + .map(str::to_owned) + .collect(); + path[1] = index.to_string(); + path +} + +fn schema_keys(body: &[u8], protocol: UpstreamProtocol, index: usize) -> Vec { + let mut path = tool_path(protocol, index); + path.push("properties".into()); + keys_at(body, &path.iter().map(String::as_str).collect::>()) +} + +fn client_body(client: ClientProtocol, schemas: &[&str]) -> Vec { + let tools = schemas + .iter() + .enumerate() + .map(|(index, schema)| match client { + ClientProtocol::Chat => format!( + r#"{{"type":"function","function":{{"name":"tool_{index}","parameters":{schema}}}}}"# + ), + ClientProtocol::Responses => { + format!(r#"{{"type":"function","name":"tool_{index}","parameters":{schema}}}"#) + } + ClientProtocol::Anthropic => { + format!(r#"{{"name":"tool_{index}","input_schema":{schema}}}"#) + } + ClientProtocol::Gemini => { + format!(r#"{{"name":"tool_{index}","parametersJsonSchema":{schema}}}"#) + } + }) + .collect::>() + .join(","); + match client { + ClientProtocol::Chat => format!( + r#"{{"model":"m","messages":[{{"role":"user","content":"hello"}}],"tools":[{tools}]}}"# + ), + ClientProtocol::Responses => { + format!(r#"{{"model":"m","input":"hello","tools":[{tools}]}}"#) + } + ClientProtocol::Anthropic => format!( + r#"{{"model":"m","max_tokens":16,"messages":[{{"role":"user","content":"hello"}}],"tools":[{tools}]}}"# + ), + ClientProtocol::Gemini => format!( + r#"{{"contents":[{{"parts":[{{"text":"hello"}}]}}],"tools":[{{"functionDeclarations":[{tools}]}}]}}"# + ), + } + .into_bytes() +} + +fn parse(client: ClientProtocol, body: &[u8]) -> Request { + match client { + ClientProtocol::Gemini => { + super::native_gemini::parse("m", false, body, &Limits::default()).unwrap() + } + _ => Request::parse(client, body, &Limits::default()).unwrap(), + } +} + +fn encoded(request: &Request, route: &Route) -> Vec { + let body = request.upstream_body(route).unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + request.upstream(route).unwrap(), + "schema ordering must not alter the authorized payload's JSON value" + ); + body +} + +#[test] +fn independent_966_all_plain_protocol_directions_preserve_each_tool_schema_order() { + for client in [ + ClientProtocol::Chat, + ClientProtocol::Responses, + ClientProtocol::Anthropic, + ClientProtocol::Gemini, + ] { + let request = parse(client, &client_body(client, &[ZA, AZ])); + for upstream in [ + UpstreamProtocol::Chat, + UpstreamProtocol::Responses, + UpstreamProtocol::Anthropic, + UpstreamProtocol::Gemini, + ] { + let body = encoded(&request, &route(upstream)); + assert_eq!(schema_keys(&body, upstream, 0), ["zeta", "alpha"]); + assert_eq!(schema_keys(&body, upstream, 1), ["alpha", "zeta"]); + } + } +} + +#[test] +fn independent_966_anthropic_same_value_tool_input_cannot_steal_schema_order() { + let body = format!( + r#"{{"model":"m","max_tokens":16,"messages":[{{"role":"assistant","content":[{{"type":"tool_use","id":"call_1","name":"tool_0","input":{ZA}}}]}},{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"call_1","content":"done"}}]}}],"tools":[{{"name":"tool_0","input_schema":{ZA}}}]}}"# + ); + let request = parse(ClientProtocol::Anthropic, body.as_bytes()); + let body = encoded(&request, &route(UpstreamProtocol::Anthropic)); + assert_eq!( + keys_at( + &body, + &["messages", "0", "content", "0", "input", "properties"] + ), + ["alpha", "zeta"], + "ordinary tool input keeps the existing canonical encoder behavior" + ); + assert_eq!( + schema_keys(&body, UpstreamProtocol::Anthropic, 0), + ["zeta", "alpha"] + ); +} + +#[test] +fn independent_966_gemini_same_value_call_args_cannot_steal_schema_order() { + let body = format!( + r#"{{"contents":[{{"role":"model","parts":[{{"functionCall":{{"id":"call_1","name":"tool_0","args":{ZA}}}}}]}},{{"role":"user","parts":[{{"functionResponse":{{"id":"call_1","name":"tool_0","response":{{"ok":true}}}}}}]}}],"tools":[{{"functionDeclarations":[{{"name":"tool_0","parametersJsonSchema":{ZA}}}]}}]}}"# + ); + let request = parse(ClientProtocol::Gemini, body.as_bytes()); + let body = encoded(&request, &route(UpstreamProtocol::Gemini)); + assert_eq!( + keys_at( + &body, + &[ + "contents", + "0", + "parts", + "0", + "functionCall", + "args", + "properties" + ] + ), + ["alpha", "zeta"] + ); + assert_eq!( + schema_keys(&body, UpstreamProtocol::Gemini, 0), + ["zeta", "alpha"] + ); +} + +#[test] +fn independent_966_gemini_openapi_conversion_preserves_order_and_normalizes_types() { + let body = br#"{"contents":[{"parts":[{"text":"hello"}]}],"tools":[{"functionDeclarations":[{"name":"tool_0","parameters":{"type":"OBJECT","properties":{"zeta":{"type":"STRING"},"alpha":{"type":"INTEGER"}}}}]}]}"#; + let request = parse(ClientProtocol::Gemini, body); + let body = encoded(&request, &route(UpstreamProtocol::Gemini)); + assert_eq!( + schema_keys(&body, UpstreamProtocol::Gemini, 0), + ["zeta", "alpha"] + ); + let actual: Value = serde_json::from_str(&raw_at( + &body, + &[ + "tools", + "0", + "functionDeclarations", + "0", + "parametersJsonSchema", + ], + )) + .unwrap(); + assert_eq!(actual["type"], "object"); + assert_eq!(actual["properties"]["alpha"]["type"], "integer"); +} + +#[test] +fn independent_966_deferred_parser_retains_order_and_route_authority() { + let mut body = String::from_utf8(client_body(ClientProtocol::Chat, &[ZA])).unwrap(); + body.pop(); + body.push_str(r#","speed":"fast"}"#); + let request = + Request::parse_deferred_openai(ClientProtocol::Chat, body.as_bytes(), &Limits::default()) + .unwrap(); + let ordinary = route(UpstreamProtocol::Chat); + assert!( + request.upstream_body(&ordinary).is_err(), + "raw body must not bypass route policy" + ); + let allowed = route(UpstreamProtocol::Chat).with_payload_policy( + crate::routing::PayloadPolicy::LegacyOpenai { + allow_service_tier: false, + allow_inference_geo: false, + allow_speed: true, + allow_safety_identifier: false, + allow_include_obfuscation: false, + }, + ); + let encoded = encoded(&request, &allowed); + assert_eq!( + schema_keys(&encoded, UpstreamProtocol::Chat, 0), + ["zeta", "alpha"] + ); + assert_eq!( + serde_json::from_slice::(&encoded).unwrap()["speed"], + "fast" + ); +} + +#[test] +fn independent_966_empty_and_mixed_schema_objects_do_not_touch_other_fields() { + let request = parse( + ClientProtocol::Chat, + &client_body(ClientProtocol::Chat, &["{}", ZA, r#"{"type":"object"}"#]), + ); + for upstream in [ + UpstreamProtocol::Chat, + UpstreamProtocol::Responses, + UpstreamProtocol::Anthropic, + UpstreamProtocol::Gemini, + ] { + let body = encoded(&request, &route(upstream)); + let p0 = tool_path(upstream, 0); + assert_eq!( + serde_json::from_str::(&raw_at( + &body, + &p0.iter().map(String::as_str).collect::>() + )) + .unwrap(), + json!({}) + ); + assert_eq!(schema_keys(&body, upstream, 1), ["zeta", "alpha"]); + } +} + +#[test] +fn independent_966_uncharged_strict_transport_keeps_order_without_route_authority() { + let plain = parse( + ClientProtocol::Chat, + &client_body(ClientProtocol::Chat, &[ZA]), + ); + let mut body = String::from_utf8(client_body(ClientProtocol::Chat, &[ZA])).unwrap(); + body.pop(); + body.push_str(r#","speed":"fast"}"#); + let deferred = + Request::parse_deferred_openai(ClientProtocol::Chat, body.as_bytes(), &Limits::default()) + .unwrap(); + for protocol in [ + UpstreamProtocol::Chat, + UpstreamProtocol::Responses, + UpstreamProtocol::Anthropic, + UpstreamProtocol::Gemini, + ] { + let output = plain.upstream_body_to(protocol, "wire-model").unwrap(); + assert_eq!( + serde_json::from_slice::(&output).unwrap(), + plain.upstream_to(protocol, "wire-model").unwrap() + ); + assert_eq!(schema_keys(&output, protocol, 0), ["zeta", "alpha"]); + assert!( + deferred.upstream_body_to(protocol, "wire-model").is_err(), + "uncharged transport has no selected-route compatibility authority" + ); + } +} + +#[test] +fn independent_966_duplicate_schema_keys_keep_last_value_and_single_output_key() { + let schema = r#"{"type":"object","properties":{"zeta":{"type":"number"},"alpha":{"type":"string"},"\u007aeta":{"type":"integer"}}}"#; + let request = parse( + ClientProtocol::Chat, + &client_body(ClientProtocol::Chat, &[schema]), + ); + let body = encoded(&request, &route(UpstreamProtocol::Chat)); + assert_eq!( + schema_keys(&body, UpstreamProtocol::Chat, 0), + ["zeta", "alpha"] + ); + let actual: Value = + serde_json::from_str(&raw_at(&body, &["tools", "0", "function", "parameters"])).unwrap(); + assert_eq!(actual["properties"]["zeta"]["type"], "integer"); +} + +#[test] +fn independent_966_large_numbers_unicode_and_nested_property_order_round_trip() { + let schema = r#"{"type":"object","properties":{"zeta":{"type":"number","default":1234567890123456789012345678901234567890},"alpha":{"type":"object","description":"δΈ­ζ–‡πŸ˜€\ud83d\ude00","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}}}"#; + let request = parse( + ClientProtocol::Chat, + &client_body(ClientProtocol::Chat, &[schema]), + ); + let body = encoded(&request, &route(UpstreamProtocol::Chat)); + assert_eq!( + schema_keys(&body, UpstreamProtocol::Chat, 0), + ["zeta", "alpha"] + ); + assert_eq!( + keys_at( + &body, + &[ + "tools", + "0", + "function", + "parameters", + "properties", + "alpha", + "properties" + ] + ), + ["reasoning", "answer"] + ); + let text = std::str::from_utf8(&body).unwrap(); + assert!(text.contains("1234567890123456789012345678901234567890")); + let actual: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!( + actual["tools"][0]["function"]["parameters"]["properties"]["alpha"]["description"], + "δΈ­ζ–‡πŸ˜€πŸ˜€" + ); +} + +#[test] +fn independent_966_capture_does_not_relax_ambiguous_controls_or_depth_limit() { + let valid = String::from_utf8(client_body(ClientProtocol::Chat, &[ZA])).unwrap(); + let repeated = valid.replacen(r#""model":"m""#, r#""model":"m","MODEL":"other""#, 1); + assert!( + Request::parse( + ClientProtocol::Chat, + repeated.as_bytes(), + &Limits::default() + ) + .is_err() + ); + let mut body = valid; + body.pop(); + body.push_str(r#","stream_options":{"include_usage":true,"INCLUDE_USAGE":false}}"#); + assert!( + Request::parse_deferred_openai(ClientProtocol::Chat, body.as_bytes(), &Limits::default()) + .is_err() + ); + let deep = format!( + r#"{{"type":"object","properties":{{"x":{}0{}}}}}"#, + "[".repeat(160), + "]".repeat(160) + ); + assert!( + Request::parse( + ClientProtocol::Chat, + &client_body(ClientProtocol::Chat, &[&deep]), + &Limits::default() + ) + .is_err() + ); +} + +#[test] +fn independent_966_input_byte_limit_remains_exact() { + let body = client_body(ClientProtocol::Chat, &[ZA]); + for offset in [0, 1] { + let limits = Limits { + request_bytes: body.len() + offset, + ..Limits::default() + }; + assert!(Request::parse(ClientProtocol::Chat, &body, &limits).is_ok()); + } + let limits = Limits { + request_bytes: body.len() - 1, + ..Limits::default() + }; + assert!(matches!( + Request::parse(ClientProtocol::Chat, &body, &limits), + Err(RelayError::Limit(_)) + )); +} + +#[test] +fn independent_966_global_canonical_json_and_digest_stay_sorted() { + let mut map = serde_json::Map::new(); + map.insert("z".into(), json!(1)); + map.insert("a".into(), json!(2)); + let actual = serde_json::to_vec(&Value::Object(map)).unwrap(); + let expected = br#"{"a":2,"z":1}"#; + assert_eq!(actual, expected); + assert_eq!( + format!("{:x}", Sha256::digest(&actual)), + "c2985c5ba6f7d2a55e768f92490ca09388e95bc4cccb9fdf11b15f4d42f93e73" + ); +} + +#[test] +fn independent_966_legal_wide_and_unicode_capacity_probe() { + for n in [1024, 2048, 4096] { + let properties = (0..n) + .rev() + .map(|i| format!(r#""p{i:05}":{{"type":"string"}}"#)) + .collect::>() + .join(","); + let schema = format!( + r#"{{"type":"object","description":"{}","properties":{{{properties}}}}}"#, + "δΈ­".repeat(n) + ); + let body = client_body(ClientProtocol::Chat, &[&schema]); + assert!(body.len() < Limits::default().request_bytes); + let started = Instant::now(); + let request = parse(ClientProtocol::Chat, &body); + let output = encoded(&request, &route(UpstreamProtocol::Chat)); + let elapsed = started.elapsed(); + let keys = schema_keys(&output, UpstreamProtocol::Chat, 0); + assert_eq!(keys.len(), n); + assert_eq!(keys.first().unwrap(), &format!("p{:05}", n - 1)); + assert_eq!(keys.last().unwrap(), "p00000"); + eprintln!( + "independent_966_capacity n={n} input_bytes={} output_bytes={} elapsed_ns={}", + body.len(), + output.len(), + elapsed.as_nanos() + ); + } + // Functional capacity exercise only. The diagnostic timing is not a + // release-mode, isolated-resource performance acceptance result. +} diff --git a/services/core/src/relay/mod.rs b/services/core/src/relay/mod.rs index 4f08354be..e3541d14f 100644 --- a/services/core/src/relay/mod.rs +++ b/services/core/src/relay/mod.rs @@ -12,6 +12,7 @@ pub(crate) mod native_anthropic; pub(crate) mod native_gemini; pub(crate) mod provider_network; mod request; +mod schema_order; mod sse; #[cfg(test)] mod tests; @@ -680,9 +681,7 @@ impl Worker { .await?; self.report.route_id = Some(route.id.clone()); self.report.price_version = Some(route.price_version); - let payload = self.request.upstream(&route)?; - let body = serde_json::to_vec(&payload) - .map_err(|_| RelayError::Invalid("request encoding failed"))?; + let body = self.request.upstream_body(&route)?; if body.len() > self.inner.limits.request_bytes { return Err(RelayError::Limit("encoded request")); } @@ -893,3 +892,6 @@ impl Worker { }); } } + +#[cfg(test)] +mod independent_966_schema_tests; diff --git a/services/core/src/relay/native_anthropic.rs b/services/core/src/relay/native_anthropic.rs index 4b14f1eb0..426d3ae21 100644 --- a/services/core/src/relay/native_anthropic.rs +++ b/services/core/src/relay/native_anthropic.rs @@ -258,6 +258,9 @@ pub(crate) fn parse(body: &[u8], limits: &Limits) -> Result &serde_json::to_vec(&canonical).map_err(|_| RelayError::Invalid("request encoding"))?, limits, )?; + if let Some(captured) = super::schema_order::anthropic_schemas(body) { + request.assign_ordered_schemas(&captured); + } let mut messages = Vec::new(); if let Some(s) = v.get("system") { messages.push(Message { diff --git a/services/core/src/relay/native_gemini.rs b/services/core/src/relay/native_gemini.rs index a73670263..dc7b87915 100644 --- a/services/core/src/relay/native_gemini.rs +++ b/services/core/src/relay/native_gemini.rs @@ -376,6 +376,9 @@ pub(crate) fn parse( .map_err(|_| RelayError::Invalid("Gemini conversion failed"))?; let mut request = Request::parse(ClientProtocol::Chat, &bytes, limits)?; request.continuation_token = continuation_token(v.get("continuationToken"), limits)?; + if let Some(captured) = super::schema_order::gemini_schemas(body) { + request.assign_ordered_schemas(&captured); + } Ok(request) } fn continuation_token(v: Option<&Value>, limits: &Limits) -> Result, RelayError> { diff --git a/services/core/src/relay/request.rs b/services/core/src/relay/request.rs index 912350f02..a310d35d9 100644 --- a/services/core/src/relay/request.rs +++ b/services/core/src/relay/request.rs @@ -66,6 +66,9 @@ pub struct FunctionTool { pub description: Option, pub parameters: Value, pub strict: Option, + /// Compact JSON for `parameters`, using the client's object-key order when + /// that text still parses as the validated value. + pub(crate) ordered_parameters: String, } #[derive(Clone, Debug)] pub enum ToolChoice { @@ -372,6 +375,9 @@ impl Request { serde_json::to_vec(&fields).map_err(|_| RelayError::Invalid("invalid request body"))?; let mut request = Self::parse(client, &bytes, limits)?; request.compatibility = compatibility; + // The compatibility parser re-encodes fields; capture only after its + // original duplicate/depth checks, from the original ordered schemas. + request.capture_ordered_schemas(client, body); Ok(request) } @@ -539,11 +545,14 @@ impl Request { if !parameters.is_object() { return Err(RelayError::Invalid("function parameters must be an object")); } + let ordered_parameters = serde_json::to_string(¶meters) + .map_err(|_| RelayError::Invalid("function schema encoding failed"))?; tools.push(FunctionTool { name, parameters, description: f.get("description").map(string).transpose()?, strict: boolean(f.get("strict"))?, + ordered_parameters, }); } } @@ -810,7 +819,7 @@ impl Request { Ok(options.clone()) }) .transpose()?; - Ok(Self { + let mut request = Self { client, model, stream, @@ -828,7 +837,73 @@ impl Request { image_config, continuation_token: None, compatibility: Map::new(), - }) + }; + request.capture_ordered_schemas(client, body); + Ok(request) + } + + fn capture_ordered_schemas(&mut self, client: ClientProtocol, body: &[u8]) { + let captured = match client { + ClientProtocol::Chat => super::schema_order::chat_schemas(body), + ClientProtocol::Responses => super::schema_order::responses_schemas(body), + ClientProtocol::Anthropic | ClientProtocol::Gemini => None, + }; + if let Some(captured) = captured { + self.assign_ordered_schemas(&captured); + } + } + + pub(crate) fn assign_ordered_schemas( + &mut self, + captured: &[Option], + ) { + if captured.len() != self.tools.len() { + return; + } + for (tool, item) in self.tools.iter_mut().zip(captured) { + let Some(item) = item else { continue }; + let ordered = match item { + super::schema_order::RawSchema::Json(raw) => { + super::schema_order::ordered_json_matching(raw, &tool.parameters) + } + super::schema_order::RawSchema::GeminiOpenApi(raw) => { + super::schema_order::ordered_gemini_schema(raw, &tool.parameters) + } + }; + if let Some(text) = ordered { + tool.ordered_parameters = text; + } + } + } + + /// Encode only after applying the immutable selected-route payload policy. + pub(crate) fn upstream_body(&self, route: &Route) -> Result, RelayError> { + let payload = self.upstream(route)?; + self.ordered_payload_body(&payload, route.protocol) + } + + /// Strict protocol/model encoding for the isolated uncharged transport. + /// It cannot authorize deferred compatibility fields or accept a Route. + pub(crate) fn upstream_body_to( + &self, + protocol: UpstreamProtocol, + model: &str, + ) -> Result, RelayError> { + let payload = self.upstream_to(protocol, model)?; + self.ordered_payload_body(&payload, protocol) + } + + fn ordered_payload_body( + &self, + payload: &Value, + protocol: UpstreamProtocol, + ) -> Result, RelayError> { + let schemas: Vec = self + .tools + .iter() + .map(|tool| tool.ordered_parameters.clone()) + .collect(); + super::schema_order::encode_ordered_schemas(payload, protocol, &schemas) } pub(crate) fn upstream(&self, route: &Route) -> Result { diff --git a/services/core/src/relay/schema_order.rs b/services/core/src/relay/schema_order.rs new file mode 100644 index 000000000..2e69ae462 --- /dev/null +++ b/services/core/src/relay/schema_order.rs @@ -0,0 +1,672 @@ +//! Keep client object order only in the selected upstream tool-schema fields. +//! +//! Behavior follows QuantumNous/new-api `d0b840e75ff36d731b04ff981ee361819acd593f`. +//! Parsing and raw-field serialization reuse the already locked serde_json +//! 1.0.151 (`serde-rs/json` v1.0.151, MIT OR Apache-2.0). Global JSON maps remain +//! sorted, so billing and policy digests keep their existing representation. + +use super::{RelayError, UpstreamProtocol}; +use serde::{ + Deserialize, Serialize, + de::{MapAccess, Visitor}, + ser::SerializeMap, +}; +use serde_json::{Value, value::RawValue}; +use std::collections::BTreeMap; + +#[derive(Clone, Debug)] +enum Node { + Scalar(Value), + Array(Vec), + Object(Vec<(String, Node)>), +} + +impl Serialize for Node { + fn serialize(&self, serializer: S) -> Result { + match self { + Self::Scalar(value) => value.serialize(serializer), + Self::Array(items) => items.serialize(serializer), + Self::Object(items) => { + let mut map = serializer.serialize_map(Some(items.len()))?; + for (key, value) in items { + map.serialize_entry(key, value)?; + } + map.end() + } + } + } +} + +#[derive(Clone, Debug)] +pub(crate) enum RawSchema { + Json(String), + GeminiOpenApi(String), +} + +pub(crate) fn ordered_json_matching(raw: &str, expected: &Value) -> Option { + let text = serde_json::to_string(&parse_json(raw)?).ok()?; + (serde_json::from_str::(&text).ok().as_ref() == Some(expected)).then_some(text) +} + +pub(crate) fn ordered_gemini_schema(raw: &str, expected: &Value) -> Option { + let node = convert_gemini(&parse_json(raw)?)?; + let text = serde_json::to_string(&node).ok()?; + (serde_json::from_str::(&text).ok().as_ref() == Some(expected)).then_some(text) +} + +pub(crate) fn chat_schemas(body: &[u8]) -> Option>> { + #[derive(Deserialize)] + struct Body { + #[serde(default)] + tools: Vec, + } + #[derive(Deserialize)] + struct Tool { + #[serde(default)] + function: Option, + } + #[derive(Deserialize)] + struct Function { + #[serde(default)] + parameters: Option>, + } + let body: Body = serde_json::from_slice(body).ok()?; + Some( + body.tools + .into_iter() + .map(|tool| { + tool.function.and_then(|function| { + function + .parameters + .map(|raw| RawSchema::Json(raw.get().to_owned())) + }) + }) + .collect(), + ) +} + +pub(crate) fn responses_schemas(body: &[u8]) -> Option>> { + #[derive(Deserialize)] + struct Body { + #[serde(default)] + tools: Vec, + } + #[derive(Deserialize)] + struct Tool { + #[serde(default)] + parameters: Option>, + } + let body: Body = serde_json::from_slice(body).ok()?; + Some( + body.tools + .into_iter() + .map(|tool| { + tool.parameters + .map(|raw| RawSchema::Json(raw.get().to_owned())) + }) + .collect(), + ) +} + +pub(crate) fn anthropic_schemas(body: &[u8]) -> Option>> { + #[derive(Deserialize)] + struct Body { + #[serde(default)] + tools: Vec, + } + #[derive(Deserialize)] + struct Tool { + #[serde(default)] + input_schema: Option>, + } + let body: Body = serde_json::from_slice(body).ok()?; + Some( + body.tools + .into_iter() + .map(|tool| { + tool.input_schema + .map(|raw| RawSchema::Json(raw.get().to_owned())) + }) + .collect(), + ) +} + +pub(crate) fn gemini_schemas(body: &[u8]) -> Option>> { + #[derive(Deserialize)] + struct Body { + #[serde(default)] + tools: Vec, + } + #[derive(Deserialize)] + struct Tool { + #[serde(default, rename = "functionDeclarations")] + function_declarations: Vec, + } + #[derive(Deserialize)] + struct Function { + #[serde(default)] + parameters: Option>, + #[serde(default, rename = "parametersJsonSchema")] + parameters_json_schema: Option>, + } + let body: Body = serde_json::from_slice(body).ok()?; + Some( + body.tools + .into_iter() + .flat_map(|tool| tool.function_declarations) + .map( + |function| match (function.parameters, function.parameters_json_schema) { + (Some(raw), None) => Some(RawSchema::GeminiOpenApi(raw.get().to_owned())), + (None, Some(raw)) => Some(RawSchema::Json(raw.get().to_owned())), + (None, None) => None, + (Some(_), Some(_)) => None, + }, + ) + .collect(), + ) +} + +/// Serialize a verified payload, replacing only the protocol's actual schema +/// fields. A same-valued message/tool-result object can never consume an update. +pub(crate) fn encode_ordered_schemas( + payload: &Value, + protocol: UpstreamProtocol, + schemas: &[String], +) -> Result, RelayError> { + let canonical = + || serde_json::to_vec(payload).map_err(|_| RelayError::Invalid("request encoding failed")); + let Some(paths) = schema_paths(payload, protocol) else { + return canonical(); + }; + if paths.len() != schemas.len() { + return canonical(); + } + let mut output = WireValue::Value(payload); + for (path, schema) in paths.iter().zip(schemas) { + let Some(actual) = value_at(payload, path) else { + continue; + }; + if serde_json::from_str::(schema).ok().as_ref() != Some(actual) { + continue; + } + let Ok(raw) = RawValue::from_string(schema.clone()) else { + continue; + }; + replace_at(&mut output, path, raw); + } + let body = + serde_json::to_vec(&output).map_err(|_| RelayError::Invalid("request encoding failed"))?; + if serde_json::from_slice::(&body).ok().as_ref() == Some(payload) { + Ok(body) + } else { + canonical() + } +} + +#[derive(Serialize)] +#[serde(untagged)] +enum WireValue<'a> { + Value(&'a Value), + Raw(Box), + Object(BTreeMap<&'a str, WireValue<'a>>), + Array(Vec>), +} + +#[derive(Clone, Copy)] +enum Member { + Key(&'static str), + Index(usize), +} + +fn schema_paths(payload: &Value, protocol: UpstreamProtocol) -> Option>> { + use Member::{Index, Key}; + let Some(tools) = payload.get("tools") else { + return Some(Vec::new()); + }; + let mut paths = Vec::new(); + for (index, tool) in tools.as_array()?.iter().enumerate() { + let mut path = vec![Key("tools"), Index(index)]; + match protocol { + UpstreamProtocol::Chat => { + path.extend([Key("function"), Key("parameters")]); + paths.push(path); + } + UpstreamProtocol::Responses => { + path.push(Key("parameters")); + paths.push(path); + } + UpstreamProtocol::Anthropic => { + path.push(Key("input_schema")); + paths.push(path); + } + UpstreamProtocol::Gemini => { + let declarations = tool.get("functionDeclarations")?.as_array()?; + for declaration in 0..declarations.len() { + let mut path = path.clone(); + path.extend([ + Key("functionDeclarations"), + Index(declaration), + Key("parametersJsonSchema"), + ]); + paths.push(path); + } + } + } + } + Some(paths) +} + +fn value_at<'a>(mut value: &'a Value, path: &[Member]) -> Option<&'a Value> { + for member in path { + value = match member { + Member::Key(key) => value.get(*key)?, + Member::Index(index) => value.get(*index)?, + }; + } + Some(value) +} + +fn replace_at<'a>(value: &mut WireValue<'a>, path: &[Member], raw: Box) { + let Some((head, tail)) = path.split_first() else { + *value = WireValue::Raw(raw); + return; + }; + if let WireValue::Value(original) = value { + *value = match original { + Value::Object(items) => WireValue::Object( + items + .iter() + .map(|(key, value)| (key.as_str(), WireValue::Value(value))) + .collect(), + ), + Value::Array(items) => WireValue::Array(items.iter().map(WireValue::Value).collect()), + _ => return, + }; + } + let next = match (value, head) { + (WireValue::Object(items), Member::Key(key)) => items.get_mut(key), + (WireValue::Array(items), Member::Index(index)) => items.get_mut(*index), + _ => None, + }; + if let Some(next) = next { + replace_at(next, tail, raw); + } +} + +const GEMINI_KEYS: &[&str] = &[ + "type", + "properties", + "required", + "description", + "items", + "enum", + "format", + "minimum", + "maximum", + "minItems", + "maxItems", +]; + +fn convert_gemini(node: &Node) -> Option { + let Node::Object(items) = node else { + return None; + }; + if items + .iter() + .any(|(key, _)| !GEMINI_KEYS.contains(&key.as_str())) + { + return None; + } + let mut out = Vec::with_capacity(items.len()); + for (key, value) in items { + let converted = match key.as_str() { + "type" => { + let Node::Scalar(Value::String(kind)) = value else { + return None; + }; + let mapped = match kind.as_str() { + "OBJECT" => "object", + "ARRAY" => "array", + "STRING" => "string", + "NUMBER" => "number", + "INTEGER" => "integer", + "BOOLEAN" => "boolean", + _ => return None, + }; + Node::Scalar(Value::String(mapped.to_owned())) + } + "properties" => { + let Node::Object(properties) = value else { + return None; + }; + let mut converted = Vec::with_capacity(properties.len()); + for (name, schema) in properties { + converted.push((name.clone(), convert_gemini(schema)?)); + } + Node::Object(converted) + } + "items" => convert_gemini(value)?, + _ => value.clone(), + }; + out.push((key.clone(), converted)); + } + Some(Node::Object(out)) +} + +// RawValue and MapAccess delegate JSON syntax, UTF-8, numbers and escapes to +// serde_json. The bounded tree is only for local object order and Gemini type +// conversion; it is never a replacement for the main request validator. +struct Members<'a>(Vec<(String, &'a RawValue)>); +impl<'de> Deserialize<'de> for Members<'de> { + fn deserialize>(deserializer: D) -> Result { + struct MembersVisitor; + impl<'de> Visitor<'de> for MembersVisitor { + type Value = Members<'de>; + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("a JSON object") + } + fn visit_map>(self, mut map: A) -> Result { + let mut positions: BTreeMap = BTreeMap::new(); + let mut items: Vec<(String, &'de RawValue)> = Vec::new(); + while let Some(key) = map.next_key::()? { + let value = map.next_value::<&'de RawValue>()?; + if let Some(index) = positions.get(&key).copied() { + items[index].1 = value; + } else { + positions.insert(key.clone(), items.len()); + items.push((key, value)); + } + } + Ok(Members(items)) + } + } + deserializer.deserialize_map(MembersVisitor) + } +} + +fn parse_json(raw: &str) -> Option { + parse_node(serde_json::from_str::<&RawValue>(raw).ok()?, 0) +} + +fn parse_node(raw: &RawValue, depth: usize) -> Option { + if depth > 64 { + return None; + } + match raw.get().as_bytes().first()? { + b'{' => { + let members: Members<'_> = serde_json::from_str(raw.get()).ok()?; + let mut items = Vec::with_capacity(members.0.len()); + for (key, value) in members.0 { + items.push((key, parse_node(value, depth + 1)?)); + } + Some(Node::Object(items)) + } + b'[' => { + let values: Vec<&RawValue> = serde_json::from_str(raw.get()).ok()?; + let mut items = Vec::with_capacity(values.len()); + for value in values { + items.push(parse_node(value, depth + 1)?); + } + Some(Node::Array(items)) + } + _ => Some(Node::Scalar(serde_json::from_str(raw.get()).ok()?)), + } +} + +#[cfg(test)] +mod tests { + use super::super::{ + ClientProtocol, Limits, Route, UpstreamProtocol, native_anthropic, native_gemini, request, + }; + use super::*; + use serde_json::json; + + fn route(protocol: UpstreamProtocol) -> Route { + Route::new( + "test-route".into(), + protocol, + "https://example.test/model", + "upstream-model".into(), + "synthetic-key", + 1, + ) + .unwrap() + } + + const SCHEMA: &str = r#"{"type":"object","properties":{"reasoning":{"type":"string","description":"why"},"answer":{"type":"string"}},"required":["reasoning","answer"]}"#; + + #[test] + fn schema_properties_keep_insertion_order_across_a_marshal_round_trip() { + let ordered = + ordered_json_matching(SCHEMA, &serde_json::from_str::(SCHEMA).unwrap()) + .expect("schema matches its parsed value"); + assert!( + ordered.contains( + r#""properties":{"reasoning":{"type":"string","description":"why"},"answer":{"type":"string"}}"# + ), + "{ordered}" + ); + let again = + ordered_json_matching(&ordered, &serde_json::from_str::(&ordered).unwrap()) + .unwrap(); + assert_eq!(again, ordered); + let sorted = + serde_json::to_string(&serde_json::from_str::(SCHEMA).unwrap()).unwrap(); + assert!(sorted.find("\"answer\"").unwrap() < sorted.find("\"reasoning\"").unwrap()); + assert_ne!(ordered, sorted); + } + + #[test] + fn chat_responses_and_anthropic_upstream_bodies_keep_property_order() { + let chat = br#"{"model":"m","messages":[{"role":"user","content":"hi"}],"tools":[{"type":"function","function":{"name":"answer","description":"d","parameters":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}},"required":["reasoning","answer"]}}}]}"#; + let request = + request::Request::parse(ClientProtocol::Chat, chat, &Limits::default()).unwrap(); + let body = String::from_utf8( + request + .upstream_body(&route(UpstreamProtocol::Chat)) + .unwrap(), + ) + .unwrap(); + assert!( + body.contains( + r#""parameters":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}},"required":["reasoning","answer"]}"# + ), + "{body}" + ); + let responses = br#"{"model":"m","input":"hi","tools":[{"type":"function","name":"answer","parameters":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}}]}"#; + let request = + request::Request::parse(ClientProtocol::Responses, responses, &Limits::default()) + .unwrap(); + let body = String::from_utf8( + request + .upstream_body(&route(UpstreamProtocol::Responses)) + .unwrap(), + ) + .unwrap(); + assert!( + body.contains( + r#""parameters":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}"# + ), + "{body}" + ); + let anthropic = br#"{"model":"m","max_tokens":16,"messages":[{"role":"user","content":"hi"}],"tools":[{"name":"answer","input_schema":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}}]}"#; + let request = native_anthropic::parse(anthropic, &Limits::default()).unwrap(); + let body = String::from_utf8( + request + .upstream_body(&route(UpstreamProtocol::Anthropic)) + .unwrap(), + ) + .unwrap(); + assert!( + body.contains( + r#""input_schema":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}"# + ), + "{body}" + ); + } + + #[test] + fn gemini_openapi_parameters_keep_property_order_while_types_are_lowercased() { + let body = br#"{"contents":[{"parts":[{"text":"hi"}]}],"tools":[{"functionDeclarations":[{"name":"answer","parameters":{"type":"OBJECT","properties":{"reasoning":{"type":"STRING"},"answer":{"type":"STRING"}}}}]}]}"#; + let request = native_gemini::parse("gemini-test", false, body, &Limits::default()).unwrap(); + let encoded = request + .upstream_body(&route(UpstreamProtocol::Gemini)) + .unwrap(); + let text = String::from_utf8(encoded).unwrap(); + assert!( + text.contains( + r#""parametersJsonSchema":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}"# + ), + "{text}" + ); + let value: Value = serde_json::from_slice(body).unwrap(); + assert_eq!( + value["tools"][0]["functionDeclarations"][0]["parameters"]["type"], + json!("OBJECT") + ); + } + + #[test] + fn gemini_json_schema_parameters_keep_client_key_order() { + let body = br#"{"contents":[{"parts":[{"text":"hi"}]}],"tools":[{"functionDeclarations":[{"name":"answer","parametersJsonSchema":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}}]}]}"#; + let request = native_gemini::parse("gemini-test", false, body, &Limits::default()).unwrap(); + let text = String::from_utf8( + request + .upstream_body(&route(UpstreamProtocol::Gemini)) + .unwrap(), + ) + .unwrap(); + assert!( + text.contains( + r#""parametersJsonSchema":{"type":"object","properties":{"reasoning":{"type":"string"},"answer":{"type":"string"}}}"# + ), + "{text}" + ); + } + + #[test] + fn replacing_schema_order_cannot_rewrite_a_same_value_tool_input() { + let raw = r#"{"type":"object","properties":{"z":{"type":"string"},"a":{"type":"string"}}}"#; + let body = format!( + r#"{{"model":"m","max_tokens":16,"messages":[{{"role":"assistant","content":[{{"type":"tool_use","id":"call_1","name":"echo","input":{raw}}}]}},{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"call_1","content":"done"}}]}}],"tools":[{{"name":"echo","input_schema":{raw}}}]}}"# + ); + let request = request::Request::parse( + ClientProtocol::Anthropic, + body.as_bytes(), + &Limits::default(), + ) + .unwrap(); + let route = route(UpstreamProtocol::Anthropic); + let payload = request.upstream(&route).unwrap(); + let canonical = + serde_json::to_string(&payload["messages"][0]["content"][0]["input"]).unwrap(); + let bytes = request.upstream_body(&route).unwrap(); + assert_eq!(serde_json::from_slice::(&bytes).unwrap(), payload); + let text = String::from_utf8(bytes).unwrap(); + assert!(text.contains(&format!("\"input_schema\":{raw}")), "{text}"); + assert!(text.contains(&format!("\"input\":{canonical}")), "{text}"); + } + + #[test] + fn equal_tool_schemas_keep_their_own_order_by_index() { + let first = + r#"{"type":"object","properties":{"z":{"type":"string"},"a":{"type":"string"}}}"#; + let second = + r#"{"properties":{"a":{"type":"string"},"z":{"type":"string"}},"type":"object"}"#; + let body = format!( + r#"{{"model":"m","messages":[{{"role":"user","content":"hi"}}],"tools":[{{"type":"function","function":{{"name":"one","parameters":{first}}}}},{{"type":"function","function":{{"name":"two","parameters":{second}}}}}]}}"# + ); + let request = + request::Request::parse(ClientProtocol::Chat, body.as_bytes(), &Limits::default()) + .unwrap(); + let route = route(UpstreamProtocol::Chat); + let bytes = request.upstream_body(&route).unwrap(); + assert_eq!( + serde_json::from_slice::(&bytes).unwrap(), + request.upstream(&route).unwrap() + ); + let text = String::from_utf8(bytes).unwrap(); + assert!(text.contains(&format!("\"parameters\":{first}")), "{text}"); + assert!(text.contains(&format!("\"parameters\":{second}")), "{text}"); + } + + #[test] + fn deferred_capture_and_body_encoding_keep_the_selected_route_policy() { + let body = br#"{"model":"m","service_tier":"priority","messages":[{"role":"user","content":"hi"}],"tools":[{"type":"function","function":{"name":"answer","parameters":{"type":"object","properties":{"z":{"type":"string"},"a":{"type":"string"}}}}}]}"#; + let request = + request::Request::parse_deferred_openai(ClientProtocol::Chat, body, &Limits::default()) + .unwrap(); + assert!( + request + .upstream_body(&route(UpstreamProtocol::Chat)) + .is_err() + ); + for allowed in [false, true] { + let route = route(UpstreamProtocol::Chat).with_payload_policy( + crate::routing::PayloadPolicy::LegacyOpenai { + allow_service_tier: allowed, + allow_inference_geo: false, + allow_speed: false, + allow_safety_identifier: false, + allow_include_obfuscation: false, + }, + ); + let bytes = request.upstream_body(&route).unwrap(); + let value: Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value, request.upstream(&route).unwrap()); + assert_eq!(value.get("service_tier").is_some(), allowed); + let text = String::from_utf8(bytes).unwrap(); + assert!( + text.contains(r#""properties":{"z":{"type":"string"},"a":{"type":"string"}}"#), + "{text}" + ); + } + } + + #[test] + fn duplicate_escaped_keys_use_last_value_at_first_position() { + let raw = r#"{"z":1,"a":{"description":"δΈ­ζ–‡πŸ˜„"},"\u007a":1234567890123456789012345678901234567890123}"#; + let expected: Value = serde_json::from_str(raw).unwrap(); + let text = ordered_json_matching(raw, &expected).unwrap(); + assert_eq!(serde_json::from_str::(&text).unwrap(), expected); + assert!(text.find("\"z\"").unwrap() < text.find("\"a\"").unwrap()); + assert_eq!(text.matches("\"z\"").count(), 1); + } + + #[test] + fn strict_transport_encoding_preserves_schema_and_rejects_deferred_fields() { + let strict = br#"{"model":"m","messages":[{"role":"user","content":"hi"}],"tools":[{"type":"function","function":{"name":"answer","parameters":{"type":"object","properties":{"z":{"type":"string"},"a":{"type":"string"}}}}}]}"#; + let request = + request::Request::parse(ClientProtocol::Chat, strict, &Limits::default()).unwrap(); + for protocol in [ + UpstreamProtocol::Chat, + UpstreamProtocol::Responses, + UpstreamProtocol::Anthropic, + UpstreamProtocol::Gemini, + ] { + let bytes = request + .upstream_body_to(protocol, "upstream-model") + .unwrap(); + assert_eq!( + serde_json::from_slice::(&bytes).unwrap(), + request.upstream_to(protocol, "upstream-model").unwrap() + ); + let text = String::from_utf8(bytes).unwrap(); + assert!( + text.contains(r#""properties":{"z":{"type":"string"},"a":{"type":"string"}}"#), + "{text}" + ); + } + let deferred = br#"{"model":"m","service_tier":"priority","messages":[{"role":"user","content":"hi"}]}"#; + let request = request::Request::parse_deferred_openai( + ClientProtocol::Chat, + deferred, + &Limits::default(), + ) + .unwrap(); + assert!(matches!( + request.upstream_body_to(UpstreamProtocol::Chat, "upstream-model"), + Err(RelayError::Unsupported(_)) + )); + } +}