From 43a3d22e371f1c2b85e2df651171b6093f21ca7e Mon Sep 17 00:00:00 2001 From: "Andrei G." Date: Mon, 20 Jul 2026 23:27:39 +0200 Subject: [PATCH] fix(memory,a2a)!: harden Qdrant search limits and A2A trust defaults Address four low-severity defense-in-depth findings from a security audit: clamp caller-supplied search limits inside zeph-memory itself instead of relying on the sole external MCP-tool clamp, warn when a non-loopback Qdrant endpoint lacks TLS/an API key, warn (in logs and the TUI chat pane) when CardTrustPolicy resolves to the permissive ignore default, and require A2aClient::new callers to pass an explicit SecurityPolicy instead of silently inheriting a permissive one. BREAKING CHANGE: A2aClient::new now requires a SecurityPolicy argument. Callers of A2aClient::new(client) must migrate to A2aClient::new(client, policy) or A2aClient::new_insecure(client) for the previous permissive-by-default behavior. --- CHANGELOG.md | 32 +++ Cargo.lock | 1 + crates/zeph-a2a/src/client.rs | 224 +++++++++++-------- crates/zeph-a2a/src/lib.rs | 2 +- crates/zeph-config/Cargo.toml | 1 + crates/zeph-config/src/loader.rs | 92 ++++++++ crates/zeph-memory/src/embedding_registry.rs | 37 +++ crates/zeph-memory/src/embedding_store.rs | 94 ++++++++ crates/zeph-memory/src/lib.rs | 51 +++++ crates/zeph-memory/src/reasoning.rs | 49 ++++ specs/014-a2a/spec.md | 7 +- src/tui_remote.rs | 33 ++- 12 files changed, 529 insertions(+), 94 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ce81db81..3f34ef937 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -97,6 +97,38 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `fetch_document_rag`, `fetch_summaries`, `fetch_cross_session`); `fetch_graph_facts` was left untouched since its three-way timeout/error/success split is not a fit for the shared helper. +### Security + +- `zeph-memory`: added a crate-level `MAX_SEARCH_LIMIT` (100) enforced directly inside + `EmbeddingStore::search`/`search_collection`, `EmbeddingRegistry::search_raw`, and + `ReasoningMemory::retrieve_by_embedding` — every caller-supplied `limit`/`top_k` is now + clamped to `[1, MAX_SEARCH_LIMIT]` inside the crate itself instead of relying on the sole + existing external clamp (`crates/zeph-core/src/memory_tools.rs`'s MCP tool clamp of + `[1, 20]`, unaffected). Defense-in-depth: no exploitable path today, but closes a gap where + a future caller forwarding an externally-supplied limit without its own clamp could trigger + an oversized Qdrant result-set allocation (issue #6553). +- `zeph-config`: `Config::validate` now logs a `tracing::warn!` when `memory.qdrant_url` + resolves to a non-loopback host without TLS (`https://`) and/or `memory.qdrant_api_key` + configured — memory content would otherwise travel in plaintext with no server + authentication. Non-fatal (does not block startup) and skipped entirely for loopback + targets (`localhost`, `127.0.0.1`, `::1`), matching the existing `A2aClientConfig` carve-out + (issue #6553). +- `zeph-a2a`: `zeph --connect ` now logs a `tracing::warn!` at discovery time when + `a2a_client.card_trust_policy` resolves to `ignore` (the default), noting that the peer's + AgentCard is accepted without signature/URL-origin verification. The default itself is + unchanged (`ignore`, byte-identical to pre-#5928 behavior) — impact was already bounded + because the A2A session connects to the CLI/config-supplied URL, never the discovered + card's self-declared URL (issue #6553). +- **BREAKING** `zeph-a2a`: `A2aClient::new` now requires an explicit `SecurityPolicy` argument + instead of defaulting to `SecurityPolicy::permissive()` — a call site can no longer silently + inherit an insecure (no TLS enforcement, no SSRF protection) client by omission. Added + `A2aClient::new_insecure` for callers that want the previous permissive-by-default behavior + explicitly (local/dev use). The one production call site (`src/tui_remote.rs`) already + computed and applied a config-driven `SecurityPolicy` via `.with_security`, and now passes + it directly to `new` (issue #6553). `zeph-a2a` is a published crate — existing callers of + `A2aClient::new(client)` must migrate to either `A2aClient::new(client, policy)` or + `A2aClient::new_insecure(client)`. + ### Added - `zeph-channels`/`zeph-config`: adopt Telegram Bot API 10.1 expandable blockquotes diff --git a/Cargo.lock b/Cargo.lock index a8aec0308..9e6acc60a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11465,6 +11465,7 @@ dependencies = [ "toml 1.1.2+spec-1.1.0", "toml_edit", "tracing", + "tracing-test", "url", "zeph-common", ] diff --git a/crates/zeph-a2a/src/client.rs b/crates/zeph-a2a/src/client.rs index 1ea08333e..495fbb05c 100644 --- a/crates/zeph-a2a/src/client.rs +++ b/crates/zeph-a2a/src/client.rs @@ -53,8 +53,9 @@ pub enum TaskEvent { /// ```rust /// use zeph_a2a::{A2aClient, SecurityPolicy}; /// -/// // Recommended for production: reject HTTP and private/loopback targets. -/// let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy::hardened()); +/// // Recommended for production: reject HTTP and private/loopback targets. Pass the +/// // policy directly to `new` so it can never be silently inherited by omission. +/// let client = A2aClient::new(reqwest::Client::new(), SecurityPolicy::hardened()); /// /// // Partial policy via named fields — no ambiguity about which flag is which. /// let tls_only = SecurityPolicy { @@ -132,11 +133,15 @@ struct PinnedTarget { /// /// # Security /// -/// Use [`with_security`](A2aClient::with_security) to harden the client for -/// production deployments — see [`SecurityPolicy`]. When either flag is enabled, -/// each request is sent through a dedicated per-request `reqwest::Client` with -/// redirects disabled and, when `ssrf_protection` is on, the connection pinned to -/// the exact addresses that were validated (no re-resolution at connect time). +/// [`new`](Self::new) requires an explicit [`SecurityPolicy`] so a call site can never +/// silently inherit a permissive default by omission (issue #6553) — use +/// [`SecurityPolicy::hardened()`] for production, or [`new_insecure`](Self::new_insecure) +/// to opt into [`SecurityPolicy::permissive()`] explicitly for local/dev use. +/// [`with_security`](A2aClient::with_security) can still override the policy after +/// construction. When either flag is enabled, each request is sent through a dedicated +/// per-request `reqwest::Client` with redirects disabled and, when `ssrf_protection` is +/// on, the connection pinned to the exact addresses that were validated (no re-resolution +/// at connect time). /// /// # Examples /// @@ -144,8 +149,7 @@ struct PinnedTarget { /// use zeph_a2a::{A2aClient, SecurityPolicy, SendMessageParams, Message}; /// /// # async fn example() -> Result<(), Box> { -/// let client = A2aClient::new(reqwest::Client::new()) -/// .with_security(SecurityPolicy::hardened()); +/// let client = A2aClient::new(reqwest::Client::new(), SecurityPolicy::hardened()); /// /// let params = SendMessageParams { /// message: Message::user_text("Summarize this page."), @@ -175,32 +179,63 @@ pub struct A2aClient { } impl A2aClient { - /// Create a new `A2aClient` with no security restrictions. + /// Create a new `A2aClient` with an explicit [`SecurityPolicy`]. /// - /// Security features are disabled by default for local/dev usage. Enable them - /// with [`with_security`](Self::with_security) for production deployments. + /// The policy is a mandatory argument (issue #6553) so a call site states its + /// security intent at construction time instead of silently inheriting a + /// permissive default by omission. Use [`SecurityPolicy::hardened()`] for + /// production deployments, or [`new_insecure`](Self::new_insecure) to opt into + /// [`SecurityPolicy::permissive()`] explicitly for local/dev use. + /// + /// # Examples + /// + /// ```rust + /// use zeph_a2a::{A2aClient, SecurityPolicy}; + /// + /// let client = A2aClient::new(reqwest::Client::new(), SecurityPolicy::hardened()); + /// ``` #[must_use] - pub fn new(client: reqwest::Client) -> Self { + pub fn new(client: reqwest::Client, security: SecurityPolicy) -> Self { Self { client, - security: SecurityPolicy::permissive(), + security, request_timeout: Duration::from_secs(30), ibct_key: None, ibct_ttl: Duration::from_mins(5), } } + /// Create a new `A2aClient` with [`SecurityPolicy::permissive()`] — no TLS + /// enforcement or SSRF protection. + /// + /// Suitable only for local development against trusted, non-adversarial endpoints + /// (e.g. `http://localhost`). Production code should call [`new`](Self::new) with + /// an explicit policy instead of relying on this insecure default. + /// + /// # Examples + /// + /// ```rust + /// use zeph_a2a::A2aClient; + /// + /// let client = A2aClient::new_insecure(reqwest::Client::new()); + /// ``` + #[must_use] + pub fn new_insecure(client: reqwest::Client) -> Self { + Self::new(client, SecurityPolicy::permissive()) + } + /// Configure the [`SecurityPolicy`] for this client. /// - /// Defaults to [`SecurityPolicy::permissive()`] (no restrictions). This method - /// uses the builder pattern and can be chained directly after [`new`](Self::new). + /// Overrides whatever policy was set at construction time ([`new`](Self::new) or + /// [`new_insecure`](Self::new_insecure)). This method uses the builder pattern and + /// can be chained directly after either constructor. /// /// # Examples /// /// ```rust /// use zeph_a2a::{A2aClient, SecurityPolicy}; /// - /// let client = A2aClient::new(reqwest::Client::new()) + /// let client = A2aClient::new_insecure(reqwest::Client::new()) /// .with_security(SecurityPolicy::hardened()); /// ``` #[must_use] @@ -238,7 +273,7 @@ impl A2aClient { /// use zeph_a2a::{A2aClient, IbctKey}; /// /// let key = IbctKey { key_id: "k1".into(), key_bytes: b"secret".to_vec() }; - /// let client = A2aClient::new(reqwest::Client::new()).with_ibct_key(key); + /// let client = A2aClient::new_insecure(reqwest::Client::new()).with_ibct_key(key); /// ``` #[must_use] pub fn with_ibct_key(mut self, key: IbctKey) -> Self { @@ -601,7 +636,7 @@ mod tests { #[test] fn a2a_client_construction() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); drop(client); } @@ -637,10 +672,11 @@ mod tests { #[tokio::test] async fn tls_enforcement_rejects_http() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: true, - ssrf_protection: false, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: true, + ssrf_protection: false, + }); let result = client.validate_endpoint("http://example.com/rpc").await; assert!(result.is_err()); let err = result.unwrap_err(); @@ -650,20 +686,22 @@ mod tests { #[tokio::test] async fn tls_enforcement_allows_https() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: true, - ssrf_protection: false, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: true, + ssrf_protection: false, + }); let result = client.validate_endpoint("https://example.com/rpc").await; assert!(result.is_ok()); } #[tokio::test] async fn ssrf_protection_rejects_localhost() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: false, - ssrf_protection: true, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: false, + ssrf_protection: true, + }); let result = client.validate_endpoint("http://127.0.0.1:8080/rpc").await; assert!(result.is_err()); assert!(result.unwrap_err().to_string().contains("SSRF")); @@ -671,7 +709,7 @@ mod tests { #[tokio::test] async fn no_security_allows_http_localhost() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let result = client.validate_endpoint("http://127.0.0.1:8080/rpc").await; assert!(result.is_ok()); } @@ -727,7 +765,7 @@ mod tests { #[tokio::test] async fn send_message_connection_error() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = SendMessageParams { message: Message::user_text("hello"), configuration: None, @@ -741,7 +779,7 @@ mod tests { #[tokio::test] async fn get_task_connection_error() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = TaskIdParams { id: "t-1".into(), history_length: None, @@ -755,7 +793,7 @@ mod tests { #[tokio::test] async fn cancel_task_connection_error() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = TaskIdParams { id: "t-1".into(), history_length: None, @@ -769,7 +807,7 @@ mod tests { #[tokio::test] async fn stream_message_connection_error() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = SendMessageParams { message: Message::user_text("stream me"), configuration: None, @@ -782,10 +820,11 @@ mod tests { #[tokio::test] async fn stream_message_tls_required_rejects_http() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: true, - ssrf_protection: false, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: true, + ssrf_protection: false, + }); let params = SendMessageParams { message: Message::user_text("hello"), configuration: None, @@ -801,10 +840,11 @@ mod tests { #[tokio::test] async fn send_message_tls_required_rejects_http() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: true, - ssrf_protection: false, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: true, + ssrf_protection: false, + }); let params = SendMessageParams { message: Message::user_text("hello"), configuration: None, @@ -818,10 +858,11 @@ mod tests { #[tokio::test] async fn get_task_tls_required_rejects_http() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: true, - ssrf_protection: false, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: true, + ssrf_protection: false, + }); let params = TaskIdParams { id: "t-1".into(), history_length: None, @@ -835,10 +876,11 @@ mod tests { #[tokio::test] async fn cancel_task_tls_required_rejects_http() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: true, - ssrf_protection: false, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: true, + ssrf_protection: false, + }); let params = TaskIdParams { id: "t-1".into(), history_length: None, @@ -852,10 +894,11 @@ mod tests { #[tokio::test] async fn validate_endpoint_invalid_url_with_ssrf() { - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: false, - ssrf_protection: true, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: false, + ssrf_protection: true, + }); let result = client.validate_endpoint("not-a-url").await; assert!(result.is_err()); assert_matches!(result.unwrap_err(), A2aError::Security(_)); @@ -863,24 +906,24 @@ mod tests { #[test] fn with_security_returns_configured_client() { - let client = - A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy::hardened()); + let client = A2aClient::new_insecure(reqwest::Client::new()) + .with_security(SecurityPolicy::hardened()); assert!(client.security.require_tls); assert!(client.security.ssrf_protection); } #[test] fn default_client_no_security() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); assert!(!client.security.require_tls); assert!(!client.security.ssrf_protection); } #[test] fn needs_hardened_client_reflects_policy() { - assert!(!A2aClient::new(reqwest::Client::new()).needs_hardened_client()); + assert!(!A2aClient::new_insecure(reqwest::Client::new()).needs_hardened_client()); assert!( - A2aClient::new(reqwest::Client::new()) + A2aClient::new_insecure(reqwest::Client::new()) .with_security(SecurityPolicy { require_tls: true, ssrf_protection: false, @@ -888,7 +931,7 @@ mod tests { .needs_hardened_client() ); assert!( - A2aClient::new(reqwest::Client::new()) + A2aClient::new_insecure(reqwest::Client::new()) .with_security(SecurityPolicy { require_tls: false, ssrf_protection: true, @@ -896,7 +939,7 @@ mod tests { .needs_hardened_client() ); assert!( - A2aClient::new(reqwest::Client::new()) + A2aClient::new_insecure(reqwest::Client::new()) .with_security(SecurityPolicy::hardened()) .needs_hardened_client() ); @@ -994,7 +1037,7 @@ mod tests { key_id: "k1".into(), key_bytes: b"secret".to_vec(), }; - let client = A2aClient::new(reqwest::Client::new()).with_ibct_key(key); + let client = A2aClient::new_insecure(reqwest::Client::new()).with_ibct_key(key); assert!(client.ibct_key.is_some()); assert_eq!(client.ibct_ttl, Duration::from_mins(5)); } @@ -1005,7 +1048,7 @@ mod tests { key_id: "k1".into(), key_bytes: b"secret".to_vec(), }; - let client = A2aClient::new(reqwest::Client::new()) + let client = A2aClient::new_insecure(reqwest::Client::new()) .with_ibct_key(key) .with_ibct_ttl(Duration::from_mins(1)); assert_eq!(client.ibct_ttl, Duration::from_mins(1)); @@ -1013,7 +1056,7 @@ mod tests { #[test] fn no_ibct_key_configured_yields_no_header() { - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); assert!( client .ibct_header_value("https://agent.example.com", "task-1") @@ -1028,7 +1071,7 @@ mod tests { key_id: "k1".into(), key_bytes: b"secret".to_vec(), }; - let client = A2aClient::new(reqwest::Client::new()).with_ibct_key(key); + let client = A2aClient::new_insecure(reqwest::Client::new()).with_ibct_key(key); let header = client .ibct_header_value("https://agent.example.com", "task-1") .expect("header should be issued when ibct feature is enabled"); @@ -1081,7 +1124,7 @@ mod tests { key_id: "k1".into(), key_bytes: b"secret".to_vec(), }; - let client = A2aClient::new(reqwest::Client::new()).with_ibct_key(key); + let client = A2aClient::new_insecure(reqwest::Client::new()).with_ibct_key(key); let header_a2a = client .ibct_header_value("http://127.0.0.1:8080/a2a", "task-1") @@ -1141,7 +1184,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = SendMessageParams { message: Message::user_text("hello"), configuration: None, @@ -1162,7 +1205,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = SendMessageParams { message: Message::user_text("hi"), configuration: None, @@ -1185,7 +1228,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = SendMessageParams { message: Message::user_text("secure"), configuration: None, @@ -1210,7 +1253,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = TaskIdParams { id: "task-get".into(), history_length: None, @@ -1231,7 +1274,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = TaskIdParams { id: "task-cancel".into(), history_length: None, @@ -1252,7 +1295,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = SendMessageParams { message: Message::user_text("stream"), configuration: None, @@ -1274,7 +1317,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()); + let client = A2aClient::new_insecure(reqwest::Client::new()); let params = SendMessageParams { message: Message::user_text("fail"), configuration: None, @@ -1306,7 +1349,7 @@ mod wiremock_tests { .mount(&server) .await; - let client = A2aClient::new(reqwest::Client::new()) + let client = A2aClient::new_insecure(reqwest::Client::new()) .with_request_timeout(std::time::Duration::from_millis(100)); let params = SendMessageParams { message: Message::user_text("hello"), @@ -1337,10 +1380,11 @@ mod wiremock_tests { let addr = *server.address(); let fake_host = "zeph-a2a-pin-test.invalid"; - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: false, - ssrf_protection: true, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: false, + ssrf_protection: true, + }); let pinned = PinnedTarget { host: fake_host.to_owned(), addrs: vec![addr], @@ -1372,10 +1416,11 @@ mod wiremock_tests { let addr = *server.address(); let fake_host = "zeph-a2a-redirect-test.invalid"; - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: false, - ssrf_protection: true, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: false, + ssrf_protection: true, + }); let pinned = PinnedTarget { host: fake_host.to_owned(), addrs: vec![addr], @@ -1408,10 +1453,11 @@ mod wiremock_tests { let addr = *server.address(); let fake_host = "zeph-a2a-tls-test.invalid"; - let client = A2aClient::new(reqwest::Client::new()).with_security(SecurityPolicy { - require_tls: true, - ssrf_protection: true, - }); + let client = + A2aClient::new_insecure(reqwest::Client::new()).with_security(SecurityPolicy { + require_tls: true, + ssrf_protection: true, + }); let pinned = PinnedTarget { host: fake_host.to_owned(), addrs: vec![addr], @@ -1445,7 +1491,7 @@ mod wiremock_tests { key_id: "k1".into(), key_bytes: b"secret".to_vec(), }; - let client = A2aClient::new(reqwest::Client::new()).with_ibct_key(key); + let client = A2aClient::new_insecure(reqwest::Client::new()).with_ibct_key(key); let params = SendMessageParams { message: Message::user_text("hello"), configuration: None, diff --git a/crates/zeph-a2a/src/lib.rs b/crates/zeph-a2a/src/lib.rs index 80f19e32d..a79ed59dd 100644 --- a/crates/zeph-a2a/src/lib.rs +++ b/crates/zeph-a2a/src/lib.rs @@ -53,7 +53,7 @@ //! let peer_card = registry.discover("http://peer-agent.example.com").await?; //! //! // Send a message to the peer agent. -//! let client = A2aClient::new(reqwest::Client::new()); +//! let client = A2aClient::new_insecure(reqwest::Client::new()); //! let params = SendMessageParams { //! message: Message::user_text("Hello, peer agent!"), //! configuration: None, diff --git a/crates/zeph-config/Cargo.toml b/crates/zeph-config/Cargo.toml index 840ced843..129d330a9 100644 --- a/crates/zeph-config/Cargo.toml +++ b/crates/zeph-config/Cargo.toml @@ -40,6 +40,7 @@ zeph-common.workspace = true [dev-dependencies] insta.workspace = true tempfile.workspace = true +tracing-test.workspace = true [lints] workspace = true diff --git a/crates/zeph-config/src/loader.rs b/crates/zeph-config/src/loader.rs index c83c8f2bf..b5bbc75a2 100644 --- a/crates/zeph-config/src/loader.rs +++ b/crates/zeph-config/src/loader.rs @@ -114,9 +114,50 @@ impl Config { .validate() .map_err(ConfigError::Validation)?; } + self.warn_insecure_qdrant_endpoint(); Ok(()) } + /// Log a warning when `memory.qdrant_url` points at a non-loopback host without TLS or + /// an API key configured (issue #6553). + /// + /// Deliberately non-fatal — repointing at a remote/managed Qdrant cluster without TLS or + /// auth is a real deployment (e.g. an internal network already trusted by other means), + /// so this warns instead of hard-failing like the bound checks in [`Self::validate`] above. + /// Skipped entirely for loopback targets (`localhost`, `127.0.0.1`, `::1`): connecting to + /// your own machine is definitionally not the plaintext-over-the-wire risk this guards + /// against, matching the same carve-out `A2aClientConfig` documents for `--connect`. + fn warn_insecure_qdrant_endpoint(&self) { + let Ok(url) = url::Url::parse(&self.memory.qdrant_url) else { + return; + }; + let Some(host) = url.host_str() else { + return; + }; + if zeph_common::net::is_loopback_host(host) { + return; + } + + let has_tls = url.scheme().eq_ignore_ascii_case("https"); + let has_api_key = self + .memory + .qdrant_api_key + .as_ref() + .is_some_and(|k| !k.expose().trim().is_empty()); + + if !has_tls || !has_api_key { + tracing::warn!( + qdrant_url = %self.memory.qdrant_url, + tls = has_tls, + api_key_configured = has_api_key, + "memory.qdrant_url points at a non-loopback host without TLS and/or an API key \ + configured — memory content would travel in plaintext with no server \ + authentication; set qdrant_url to an https:// endpoint and configure \ + memory.qdrant_api_key (vault key ZEPH_QDRANT_API_KEY) for remote/managed Qdrant" + ); + } + } + /// Validate scalar bounds for memory, agent, a2a, and gateway fields. fn validate_scalar_bounds(&self) -> Result<(), ConfigError> { if self.memory.history_limit > 10_000 { @@ -1880,4 +1921,55 @@ weight = 0.3 cfg.orchestration.ensemble.ema_alpha = 5.0; assert!(cfg.validate().is_ok()); } + + // ── warn_insecure_qdrant_endpoint (issue #6553) ─────────────────────────── + + #[test] + #[tracing_test::traced_test] + fn qdrant_loopback_url_never_warns() { + let mut cfg = Config::default(); + cfg.memory.qdrant_url = "http://localhost:6334".into(); + assert!(cfg.validate().is_ok()); + assert!(!logs_contain("memory.qdrant_url")); + } + + #[test] + #[tracing_test::traced_test] + fn qdrant_non_loopback_plaintext_no_key_warns() { + let mut cfg = Config::default(); + cfg.memory.qdrant_url = "http://qdrant.example.com:6334".into(); + assert!(cfg.validate().is_ok(), "must warn, not hard-fail"); + assert!(logs_contain("memory.qdrant_url")); + } + + #[test] + #[tracing_test::traced_test] + fn qdrant_non_loopback_https_with_key_does_not_warn() { + let mut cfg = Config::default(); + cfg.memory.qdrant_url = "https://qdrant.example.com:6334".into(); + cfg.memory.qdrant_api_key = Some(zeph_common::secret::Secret::new("test-key")); + assert!(cfg.validate().is_ok()); + assert!(!logs_contain("memory.qdrant_url")); + } + + #[test] + #[tracing_test::traced_test] + fn qdrant_non_loopback_https_without_key_still_warns() { + let mut cfg = Config::default(); + cfg.memory.qdrant_url = "https://qdrant.example.com:6334".into(); + assert!(cfg.validate().is_ok()); + assert!(logs_contain("memory.qdrant_url")); + } + + #[test] + #[tracing_test::traced_test] + fn qdrant_non_loopback_plaintext_with_key_still_warns() { + // TLS is still required even with an API key — a key sent over plaintext HTTP is + // itself exposed on the wire. + let mut cfg = Config::default(); + cfg.memory.qdrant_url = "http://qdrant.example.com:6334".into(); + cfg.memory.qdrant_api_key = Some(zeph_common::secret::Secret::new("test-key")); + assert!(cfg.validate().is_ok()); + assert!(logs_contain("memory.qdrant_url")); + } } diff --git a/crates/zeph-memory/src/embedding_registry.rs b/crates/zeph-memory/src/embedding_registry.rs index 144bc2903..ea6f66f9d 100644 --- a/crates/zeph-memory/src/embedding_registry.rs +++ b/crates/zeph-memory/src/embedding_registry.rs @@ -295,6 +295,11 @@ impl EmbeddingRegistry { /// /// Consumers map the payloads to their domain types. /// + /// `limit` is clamped to `[1, `[`MAX_SEARCH_LIMIT`](crate::MAX_SEARCH_LIMIT)`]` before + /// being forwarded to Qdrant (issue #6553) — the bound is enforced here rather than + /// relying on every caller to clamp before calling. A one-shot `tracing::warn!` fires + /// the first time this actually reduces the requested value. + /// /// # Errors /// /// Returns [`EmbeddingRegistryError::DimensionProbe`] when the query vector dimension does not @@ -307,6 +312,10 @@ impl EmbeddingRegistry { limit: usize, embed_fn: impl Fn(&str) -> EmbedFuture, ) -> Result, EmbeddingRegistryError> { + static CLAMP_WARNED: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(false); + crate::warn_if_search_limit_clamped("EmbeddingRegistry::search_raw", limit, &CLAMP_WARNED); + let limit = limit.clamp(1, crate::MAX_SEARCH_LIMIT); let query_vec = embed_fn(query) .await .map_err(|e| EmbeddingRegistryError::Embedding(e.to_string()))?; @@ -819,6 +828,34 @@ mod tests { ); } + /// Issue #6553: an oversized `limit` must be clamped to `MAX_SEARCH_LIMIT` before it + /// reaches the `u64` conversion / Qdrant call, rather than forwarded as-is. `QdrantOps` + /// has no test seam (concrete gRPC client, not a trait object), so this cannot assert the + /// clamped *count* end-to-end the way `EmbeddingStore::search`/`search_collection` and + /// `ReasoningMemory::retrieve_by_embedding` do (those accept an injectable `VectorStore`). + /// It does assert, via the one-shot `tracing::warn!`, that the clamp logic actually ran + /// with the oversized value rather than being skipped or short-circuited — a stronger + /// signal than "did not panic" alone (critic finding M5). + #[tokio::test] + #[tracing_test::traced_test] + async fn search_raw_oversized_limit_does_not_panic() { + let ops = QdrantOps::new("http://127.0.0.1:1", None).unwrap(); // unreachable — forces network error + let ns = uuid::Uuid::from_bytes([0u8; 16]); + let reg = EmbeddingRegistry::new(ops, "test_oversized_limit", ns); + reg.set_cached_dim(2).await; + + let embed_fn = |_: &str| -> EmbedFuture { Box::pin(async { Ok(vec![1.0_f32, 0.0]) }) }; + let result = reg.search_raw("query", usize::MAX, embed_fn).await; + assert!( + !matches!(result, Err(EmbeddingRegistryError::DimensionProbe(_))), + "clamp must not corrupt the dimension guard" + ); + assert!( + logs_contain("requested search limit exceeds MAX_SEARCH_LIMIT"), + "expected the one-shot clamp warning to fire for an oversized limit" + ); + } + #[tokio::test] async fn get_vectors_by_keys_empty_input_short_circuits() { // Unreachable Qdrant instance: if this didn't short-circuit it would error/hang. diff --git a/crates/zeph-memory/src/embedding_store.rs b/crates/zeph-memory/src/embedding_store.rs index ac6328af9..403cfb159 100644 --- a/crates/zeph-memory/src/embedding_store.rs +++ b/crates/zeph-memory/src/embedding_store.rs @@ -464,6 +464,11 @@ impl EmbeddingStore { /// Search for similar vectors in Qdrant, returning up to `limit` results. /// + /// `limit` is clamped to `[1, `[`MAX_SEARCH_LIMIT`](crate::MAX_SEARCH_LIMIT)`]` before + /// being forwarded to Qdrant (issue #6553) — the bound is enforced here rather than + /// relying on every caller to clamp before calling. A one-shot `tracing::warn!` fires + /// the first time this actually reduces the requested value. + /// /// # Errors /// /// Returns an error if the Qdrant search fails. @@ -474,6 +479,10 @@ impl EmbeddingStore { limit: usize, filter: Option, ) -> Result, MemoryError> { + static CLAMP_WARNED: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(false); + crate::warn_if_search_limit_clamped("EmbeddingStore::search", limit, &CLAMP_WARNED); + let limit = limit.clamp(1, crate::MAX_SEARCH_LIMIT); let limit_u64 = u64::try_from(limit)?; let vector_filter = filter.as_ref().and_then(|f| { @@ -640,6 +649,11 @@ impl EmbeddingStore { /// Search a named Qdrant collection, returning scored points with payloads. /// + /// `limit` is clamped to `[1, `[`MAX_SEARCH_LIMIT`](crate::MAX_SEARCH_LIMIT)`]` before + /// being forwarded to Qdrant (issue #6553) — the bound is enforced here rather than + /// relying on every caller to clamp before calling. A one-shot `tracing::warn!` fires + /// the first time this actually reduces the requested value. + /// /// # Errors /// /// Returns an error if the Qdrant search fails. @@ -651,6 +665,14 @@ impl EmbeddingStore { limit: usize, filter: Option, ) -> Result, MemoryError> { + static CLAMP_WARNED: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(false); + crate::warn_if_search_limit_clamped( + "EmbeddingStore::search_collection", + limit, + &CLAMP_WARNED, + ); + let limit = limit.clamp(1, crate::MAX_SEARCH_LIMIT); let limit_u64 = u64::try_from(limit)?; let results = self .ops @@ -1047,6 +1069,47 @@ mod tests { assert!((results[0].score - 1.0).abs() < 0.001); } + /// Issue #6553: `search` must clamp an oversized `limit` to `MAX_SEARCH_LIMIT` internally, + /// rather than relying on every caller to clamp before calling, and must log a one-shot + /// warning so a config-driven candidate pool silently shrunk by the clamp is observable + /// (critic finding S1). + #[tokio::test] + #[tracing_test::traced_test] + async fn embedding_store_search_clamps_oversized_limit() { + let (store, sqlite) = setup_with_store().await; + let cid = sqlite.create_conversation().await.unwrap(); + + for i in 0..(crate::MAX_SEARCH_LIMIT + 10) { + let msg_id = sqlite + .save_message(cid, "user", &format!("message {i}")) + .await + .unwrap(); + store + .store( + msg_id, + cid, + "user", + vec![1.0, 0.0, 0.0, 0.0], + MessageKind::Regular, + "test-model", + 0, + None, + ) + .await + .unwrap(); + } + + let results = store + .search(&[1.0, 0.0, 0.0, 0.0], usize::MAX, None) + .await + .unwrap(); + assert_eq!(results.len(), crate::MAX_SEARCH_LIMIT); + assert!( + logs_contain("requested search limit exceeds MAX_SEARCH_LIMIT"), + "expected a one-shot warn when the clamp actually reduces the requested limit" + ); + } + /// #5742 regression: two independent databases (own `conversation_id` counters, each starting /// at 1) share one backing vector store. Before the fix, `search()`'s `conversation_id`-only /// filter would let a message stored under db-b leak into db-a's scoped search and vice versa. @@ -1209,6 +1272,37 @@ mod tests { assert!(!results[0].payload.contains_key("category")); } + /// Issue #6553: `search_collection` must clamp an oversized `limit` to `MAX_SEARCH_LIMIT` + /// internally, rather than relying on every caller to clamp before calling, and must log a + /// one-shot warning so a config-driven candidate pool silently shrunk by the clamp is + /// observable (critic finding S1). + #[tokio::test] + #[tracing_test::traced_test] + async fn embedding_store_search_collection_clamps_oversized_limit() { + let (store, _sqlite) = setup_with_store().await; + + for _ in 0..(crate::MAX_SEARCH_LIMIT + 10) { + store + .store_to_collection( + COLLECTION_NAME, + serde_json::json!({}), + vec![1.0, 0.0, 0.0, 0.0], + ) + .await + .unwrap(); + } + + let results = store + .search_collection(COLLECTION_NAME, &[1.0, 0.0, 0.0, 0.0], usize::MAX, None) + .await + .unwrap(); + assert_eq!(results.len(), crate::MAX_SEARCH_LIMIT); + assert!( + logs_contain("requested search limit exceeds MAX_SEARCH_LIMIT"), + "expected a one-shot warn when the clamp actually reduces the requested limit" + ); + } + /// `store_with_tool_context` must write `tool_name`, `exit_code`, and `timestamp` payload /// fields (#5486 shared `store_impl` helper must preserve this variant's fields). #[tokio::test] diff --git a/crates/zeph-memory/src/lib.rs b/crates/zeph-memory/src/lib.rs index c143cfda7..7b80af9b4 100644 --- a/crates/zeph-memory/src/lib.rs +++ b/crates/zeph-memory/src/lib.rs @@ -65,6 +65,57 @@ //! | `pdf` | Enable `PdfLoader` for PDF ingestion. | //! | `postgres` | Enable PostgreSQL support via `zeph-db`. | +/// Upper bound clamped onto every caller-supplied `limit`/`top_k` search parameter +/// inside this crate (issue #6553). +/// +/// [`embedding_store::EmbeddingStore::search`], [`embedding_store::EmbeddingStore::search_collection`], +/// [`embedding_registry::EmbeddingRegistry::search_raw`], and +/// [`reasoning::ReasoningMemory::retrieve_by_embedding`] all enforce this bound directly, so the +/// safety guarantee does not depend on every external caller (MCP tools, plugins, future +/// call sites) remembering to clamp before forwarding a value here. +/// +/// `100` bounds the Qdrant result set Zeph will allocate/deserialize in a single call. Note +/// this **can** clamp a legitimate config-driven candidate pool: `memory.retrieval.depth` and +/// `memory.session.recall_limit` have no upper-bound validation today, and `retrieval.depth`'s +/// own doc actively recommends raising it ("higher for better MMR diversity") with no ceiling +/// — an operator who does so past `100` gets a silently smaller ANN pool than configured. Each +/// clamping call site logs a `tracing::warn!` the first time this happens so the degradation is +/// observable rather than silent; there is deliberately no config knob to raise this ceiling, +/// since doing so would reopen the oversized-result-set `DoS` this constant exists to close. +/// +/// # Examples +/// +/// ``` +/// use zeph_memory::MAX_SEARCH_LIMIT; +/// +/// let requested = 5_000_usize; +/// assert_eq!(requested.clamp(1, MAX_SEARCH_LIMIT), MAX_SEARCH_LIMIT); +/// ``` +pub const MAX_SEARCH_LIMIT: usize = 100; + +/// Log a one-shot `tracing::warn!` the first time a caller's requested search limit is +/// actually reduced by [`MAX_SEARCH_LIMIT`] (issue #6553 follow-up). +/// +/// `warned` is a call-site-local flag (a `static AtomicBool`) so each of the four clamping +/// functions warns independently and at most once per process — a config-driven candidate +/// pool that regularly exceeds the ceiling would otherwise log on every search call. +pub(crate) fn warn_if_search_limit_clamped( + site: &'static str, + requested: usize, + warned: &std::sync::atomic::AtomicBool, +) { + if requested > MAX_SEARCH_LIMIT && !warned.swap(true, std::sync::atomic::Ordering::Relaxed) { + tracing::warn!( + site, + requested, + max = MAX_SEARCH_LIMIT, + "requested search limit exceeds MAX_SEARCH_LIMIT and was clamped — a config-driven \ + candidate pool (e.g. memory.retrieval.depth) larger than this will silently return \ + a smaller ANN pool than configured" + ); + } +} + pub mod admission; pub mod anchored_summary; pub mod compaction_probe; diff --git a/crates/zeph-memory/src/reasoning.rs b/crates/zeph-memory/src/reasoning.rs index bf3470d67..41aa9f356 100644 --- a/crates/zeph-memory/src/reasoning.rs +++ b/crates/zeph-memory/src/reasoning.rs @@ -277,6 +277,11 @@ impl ReasoningMemory { /// /// Returns an empty vec when no vector store is configured. /// + /// `top_k` is clamped to `[1, `[`MAX_SEARCH_LIMIT`](crate::MAX_SEARCH_LIMIT)`]` before + /// being forwarded to Qdrant (issue #6553) — the bound is enforced here rather than + /// relying on every caller to clamp before calling. A one-shot `tracing::warn!` fires + /// the first time this actually reduces the requested value. + /// /// # Errors /// /// Returns an error if the Qdrant search or `SQLite` fetch fails. @@ -294,6 +299,16 @@ impl ReasoningMemory { return Ok(Vec::new()); }; + static CLAMP_WARNED: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(false); + if let Ok(requested) = usize::try_from(top_k) { + crate::warn_if_search_limit_clamped( + "ReasoningMemory::retrieve_by_embedding", + requested, + &CLAMP_WARNED, + ); + } + let top_k = top_k.clamp(1, crate::MAX_SEARCH_LIMIT as u64); let scored = vs .search(REASONING_COLLECTION, embedding.to_vec(), top_k, None) .await?; @@ -1011,6 +1026,40 @@ mod tests { assert_eq!(rows[0].outcome, Outcome::Success); } + /// Issue #6553: `retrieve_by_embedding` must clamp an oversized `top_k` to + /// `MAX_SEARCH_LIMIT` internally, rather than relying on every caller to clamp, and must + /// log a one-shot warning so a config-driven candidate pool silently shrunk by the clamp + /// is observable (critic finding S1). + #[cfg(feature = "sqlite")] + #[tokio::test] + #[tracing_test::traced_test] + async fn retrieve_by_embedding_clamps_oversized_top_k() { + let pool = make_test_pool().await; + let vector_store = std::sync::Arc::new(crate::in_memory_store::InMemoryVectorStore::new()); + vector_store + .ensure_collection(REASONING_COLLECTION, 4) + .await + .unwrap(); + let mem = ReasoningMemory::new(pool, Some(vector_store)); + + for i in 0..(crate::MAX_SEARCH_LIMIT + 10) { + let id = format!("strat-{i}"); + mem.insert(&make_strategy(&id), vec![1.0, 0.0, 0.0, 0.0]) + .await + .unwrap(); + } + + let results = mem + .retrieve_by_embedding(&[1.0, 0.0, 0.0, 0.0], u64::MAX) + .await + .unwrap(); + assert_eq!(results.len(), crate::MAX_SEARCH_LIMIT); + assert!( + logs_contain("requested search limit exceeds MAX_SEARCH_LIMIT"), + "expected a one-shot warn when the clamp actually reduces the requested limit" + ); + } + #[cfg(feature = "sqlite")] #[tokio::test] async fn mark_used_increments_count() { diff --git a/specs/014-a2a/spec.md b/specs/014-a2a/spec.md index 735f0e035..e83946bcc 100644 --- a/specs/014-a2a/spec.md +++ b/specs/014-a2a/spec.md @@ -174,8 +174,11 @@ Terminal states: `completed | failed | canceled | rejected` `A2aClient` accepts a [`SecurityPolicy`] (`crates/zeph-a2a/src/client.rs`) with two independent flags: `require_tls` and `ssrf_protection`. `SecurityPolicy::hardened()` enables both; -`SecurityPolicy::permissive()` (the `A2aClient::new` default) disables both, for local/dev use -against trusted endpoints. +`SecurityPolicy::permissive()` disables both, for local/dev use against trusted endpoints. +`A2aClient::new` requires the policy as an explicit constructor argument — it has no default — +so a call site can never silently inherit a permissive client by omission (issue #6553). +`A2aClient::new_insecure` is the explicit opt-in to `SecurityPolicy::permissive()` for +local/dev callers that want the previous default-constructor behavior. ### SSRF invariant: validated address == connected address diff --git a/src/tui_remote.rs b/src/tui_remote.rs index ce6799149..bc5f3c8d2 100644 --- a/src/tui_remote.rs +++ b/src/tui_remote.rs @@ -212,6 +212,17 @@ pub(crate) async fn run_tui_remote( let discovery_base_url = discovery_origin(&url); let discovery_client = hardened_discovery_client(&discovery_base_url, security).await?; let trust_policy = convert_card_trust_policy(config.a2a_client.card_trust_policy); + if trust_policy == zeph_a2a::CardTrustPolicy::Ignore { + // Default policy (issue #6553): peer AgentCards are accepted without signature or + // URL-origin verification. Bounded risk — the A2A session below connects to the + // CLI/config-supplied `url`, never to the discovered card's self-declared `url` — but + // still worth a visible signal since it means capability metadata is fully trusted. + tracing::warn!( + "a2a discovery: card_trust_policy=ignore — peer AgentCard accepted without \ + signature/origin verification; set a2a_client.card_trust_policy to \"prefer\" or \ + \"require\" to verify peer identity" + ); + } let registry = zeph_a2a::AgentRegistry::new(discovery_client, Duration::from_mins(5)) .with_trust( trust_policy, @@ -237,8 +248,7 @@ pub(crate) async fn run_tui_remote( } } - let client = - zeph_a2a::A2aClient::new(zeph_core::http::default_client()).with_security(security); + let client = zeph_a2a::A2aClient::new(zeph_core::http::default_client(), security); // Cloned before `url` is moved into the `async move` SSE pump block below. let remote_daemon_url = url.clone(); @@ -248,6 +258,25 @@ pub(crate) async fn run_tui_remote( let (user_tx, mut user_rx) = tokio::sync::mpsc::channel::(32); let (agent_tx, agent_rx) = tokio::sync::mpsc::channel::(256); + if trust_policy == zeph_a2a::CardTrustPolicy::Ignore { + // Mirrors the `tracing::warn!` above, but the `tracing::warn!` alone is not a + // reliable signal once the TUI screen is up: `init_tracing` only keeps the stderr + // layer when `--tui` was NOT passed (`tracing_init.rs`), yet this function renders a + // ratatui screen unconditionally — so `zeph --tui --connect ` would otherwise + // suppress the warning to the file-only log layer with no on-screen trace at all + // (CLAUDE.md TUI Rules: implicit/security-relevant state needs a visible indicator). + // Queued on `agent_tx` before `App::new` even takes `agent_rx`, so it is guaranteed to + // be the first system message the chat pane renders once the TUI starts. + let _ = agent_tx + .send(zeph_tui::AgentEvent::FullMessage( + "Warning: card_trust_policy=ignore — peer AgentCard accepted without \ + signature/origin verification. Set a2a_client.card_trust_policy to \ + \"prefer\" or \"require\" to verify peer identity." + .into(), + )) + .await; + } + let tui_cancel = tokio_util::sync::CancellationToken::new(); let tui_supervisor = zeph_common::TaskSupervisor::new(tui_cancel.clone());