From 9f33cb5e0fb1c5f661838ca311f505f8ed9e7a4e Mon Sep 17 00:00:00 2001 From: Jarvis Date: Mon, 10 Aug 2026 11:55:48 +0800 Subject: [PATCH 1/3] fix(a2a): announce the pinned wire version and find path-hosted agent cards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two faults on the upstream-facing side of the A2A bridge, hit in that order by the same agent. The gateway never sent `A2A-Version`. `protocol_version` was stored, validated and exposed through the Admin API, but nothing in the data plane read it. The spec makes an agent read an absent version as `0.3`, so every 1.0-pinned agent answered `VersionNotSupportedError` and the field was inert. The pinned version now rides on `A2aUpstream` and is announced on both the JSON-RPC call and the card fetch — a card fetch is an A2A request like any other. Agent-card discovery resolved the well-known URI at the origin, because `Url::set_path` replaces a path rather than extending it. An agent published under a path prefix — any platform multiplexing tenants, any agent behind an ingress path — was asked for a card it does not serve there, and the platform's catch-all `405` came back as the agent's answer. There was no configuration workaround: one `url` field feeds both the card fetch and the RPC endpoint. Discovery now walks candidates, the registered path before the origin, and both spec paths (`agent-card.json`, then the previous `agent.json`). Any failure moves to the next candidate, not just a 404, since an unrouted URI answers with whatever the upstream's catch-all returns. The origin stays a candidate, so agents registered while it was the only one keep resolving. Also closes the sibling gap found while reproducing the first: the served card rewrote only its top-level `url`. A 1.0 caller picks its endpoint out of `supportedInterfaces`, so it read the upstream address off the card and called the agent directly — past auth, quota and usage, with the internal address handed over on the way. Every advertised URL is now rewritten. Tests: the A2A MVP deferred endpoint-level e2e, so this adds it — `a2a-gateway-e2e` drives a real gateway + etcd + real upstream agents (one origin-hosted, one path-hosted) and pins the version announcement, path-hosted discovery, the full card rewrite, credential containment and the ACL gates. Each fix was confirmed to fail its test when reverted, at the crate level and against a pre-fix binary. Fixes #911 Fixes #913 --- crates/aisix-a2a/src/bridge.rs | 182 +++++++++++-- crates/aisix-a2a/tests/upstream_roundtrip.rs | 171 ++++++++++-- crates/aisix-core/src/models/a2a_agent.rs | 28 +- crates/aisix-proxy/src/a2a.rs | 73 ++++- schemas/resources/a2a_agent.schema.json | 4 +- tests/e2e/src/cases/a2a-gateway-e2e.test.ts | 263 +++++++++++++++++++ tests/e2e/src/harness/index.ts | 6 + tests/e2e/src/harness/upstream-a2a.ts | 167 ++++++++++++ 8 files changed, 832 insertions(+), 62 deletions(-) create mode 100644 tests/e2e/src/cases/a2a-gateway-e2e.test.ts create mode 100644 tests/e2e/src/harness/upstream-a2a.ts diff --git a/crates/aisix-a2a/src/bridge.rs b/crates/aisix-a2a/src/bridge.rs index bb2def0b..fcd1cdb7 100644 --- a/crates/aisix-a2a/src/bridge.rs +++ b/crates/aisix-a2a/src/bridge.rs @@ -11,9 +11,14 @@ //! exposed to the calling client, which presents only its AISIX key. //! //! Wire references (verified against the A2A specification): -//! - Agent card discovery: `https://{domain}/.well-known/agent-card.json`, -//! an RFC 8615 well-known URI resolved at the domain origin. +//! - Agent card discovery: the RFC 8615 well-known URI +//! `https://{domain}/.well-known/agent-card.json`. The spec resolves it at the +//! origin, but real deployments also publish it under the agent's own path +//! prefix, so both bases are tried — see [`HttpBridge::agent_card_urls`]. //! +//! - Every request announces the wire version it speaks in the `A2A-Version` +//! header; an agent that receives none must assume `0.3`. The version is +//! pinned per agent on the `A2aAgent` resource. //! - `message/send` is a JSON-RPC 2.0 method whose envelope differs between the //! A2A 0.3 and 1.0 wire formats. This bridge forwards the caller's request //! verbatim and does not translate between versions, so the method name and @@ -23,7 +28,7 @@ use std::sync::OnceLock; use std::time::Duration; -use aisix_core::{A2aAgent, A2aAuthType}; +use aisix_core::{A2aAgent, A2aAuthType, A2aProtocolVersion}; use async_trait::async_trait; use futures::StreamExt; use serde::{Deserialize, Serialize}; @@ -39,8 +44,17 @@ pub const DEFAULT_UPSTREAM_TIMEOUT: Duration = Duration::from_secs(30); /// Header carrying the gateway-held key for `api_key` upstream auth. const API_KEY_HEADER: &str = "x-api-key"; -/// Standard RFC 8615 well-known path for an A2A agent card. -const AGENT_CARD_PATH: &str = "/.well-known/agent-card.json"; +/// Header naming the A2A wire version the client speaks. The spec requires a +/// client to send it on every request and requires an agent to read an absent +/// value as `0.3`, so an unlabelled call to a 1.0 agent is not merely untidy — +/// the agent answers `VersionNotSupportedError` and the call never lands. +const VERSION_HEADER: &str = "A2A-Version"; + +/// Well-known agent-card paths, current spec first. RFC 8615 resolves a +/// well-known URI at the origin, but platforms that multiplex tenants under a +/// path prefix publish the card relative to the agent's own path instead, so +/// each of these is tried against both bases — see [`HttpBridge::agent_card_urls`]. +const AGENT_CARD_PATHS: [&str; 2] = ["/.well-known/agent-card.json", "/.well-known/agent.json"]; /// Hard cap on an upstream response body the gateway will buffer. A registered /// agent is semi-trusted, but a compromised or misbehaving one must not be able @@ -129,10 +143,13 @@ impl std::fmt::Debug for A2aAuth { pub struct A2aUpstream { /// The agent's A2A service endpoint, where JSON-RPC requests are sent, e.g. /// `https://agents.example.com/a2a`. The agent card is discovered at the - /// well-known path relative to this URL's origin. + /// well-known paths relative to this URL, then to its origin. pub url: String, /// Upstream authentication, held gateway-side. pub auth: A2aAuth, + /// The wire version this agent speaks, announced to it on every request in + /// the `A2A-Version` header. + pub protocol_version: A2aProtocolVersion, /// Per-operation deadline. Defaults to [`DEFAULT_UPSTREAM_TIMEOUT`]. pub timeout: Duration, } @@ -144,6 +161,7 @@ impl std::fmt::Debug for A2aUpstream { f.debug_struct("A2aUpstream") .field("url", &self.url) .field("auth", &self.auth) + .field("protocol_version", &self.protocol_version) .field("timeout", &self.timeout) .finish() } @@ -199,6 +217,7 @@ pub fn upstream_from_a2a_agent(agent: &A2aAgent) -> A2aUpstream { A2aUpstream { url: agent.url.clone(), auth, + protocol_version: agent.protocol_version, timeout, } } @@ -249,8 +268,12 @@ impl HttpBridge { } } - /// Apply the gateway-held upstream credential to an outgoing request. - fn apply_auth(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + /// Apply the gateway-held upstream credential and announce the wire version + /// this agent is pinned to. Both belong on every outgoing request: the + /// credential because the gateway is the one holding it, the version + /// because an agent that receives no `A2A-Version` must assume `0.3`. + fn prepare(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + let req = req.header(VERSION_HEADER, self.upstream.protocol_version.as_wire_str()); match &self.upstream.auth { A2aAuth::None => req, A2aAuth::Bearer(token) => req.bearer_auth(token), @@ -258,23 +281,43 @@ impl HttpBridge { } } - /// Resolve the agent-card well-known URI from the service endpoint's origin - /// (RFC 8615): scheme + host + port, with the well-known path. - fn agent_card_url(&self) -> Result { - let mut url = reqwest::Url::parse(&self.upstream.url) + /// Candidate agent-card URIs, most specific first: each well-known path + /// resolved against the registered URL's own path, then against its origin. + /// + /// Both bases are real deployments. An agent that owns its domain publishes + /// at the RFC 8615 origin URI — the only shape this bridge used to try, and + /// what every agent registered before now still serves. A platform that + /// multiplexes tenants under a path prefix (or a self-hosted agent behind an + /// ingress path) publishes relative to its own path instead — the origin + /// there belongs to the platform, not to any one agent. Resolving only one + /// of the two locks out the other, so both are tried, specific first. + /// + /// A registered URL that is already at the origin yields the origin + /// candidates alone — the two bases coincide and probing twice is waste. + fn agent_card_urls(&self) -> Result, A2aError> { + let base = reqwest::Url::parse(&self.upstream.url) .map_err(|e| A2aError::Connect(format!("invalid upstream url: {e}")))?; - url.set_path(AGENT_CARD_PATH); - url.set_query(None); - Ok(url) + let prefix = base.path().trim_end_matches('/').to_string(); + let mut urls = Vec::with_capacity(AGENT_CARD_PATHS.len() * 2); + for path in AGENT_CARD_PATHS { + if !prefix.is_empty() { + let mut relative = base.clone(); + relative.set_path(&format!("{prefix}{path}")); + relative.set_query(None); + urls.push(relative); + } + let mut origin = base.clone(); + origin.set_path(path); + origin.set_query(None); + urls.push(origin); + } + Ok(urls) } -} -#[async_trait] -impl A2aBridge for HttpBridge { - async fn fetch_agent_card(&self) -> Result { - let url = self.agent_card_url()?; + /// Fetch and parse the card at exactly one candidate URI. + async fn fetch_card_at(&self, url: reqwest::Url) -> Result { let resp = self - .apply_auth(self.client.get(url).timeout(self.upstream.timeout)) + .prepare(self.client.get(url).timeout(self.upstream.timeout)) .send() .await .map_err(|e| A2aError::Connect(e.to_string()))?; @@ -288,10 +331,31 @@ impl A2aBridge for HttpBridge { serde_json::from_slice::(&bytes) .map_err(|e| A2aError::Request(format!("malformed agent card: {e}"))) } +} + +#[async_trait] +impl A2aBridge for HttpBridge { + async fn fetch_agent_card(&self) -> Result { + // Any failure moves to the next candidate, not just a 404: an upstream + // that does not route the URI answers with whatever its catch-all + // returns (the 405 in #913), and a prefix that happens to resolve can + // still hand back something that is not a card. + let mut last_err = None; + for url in self.agent_card_urls()? { + match self.fetch_card_at(url.clone()).await { + Ok(card) => return Ok(card), + Err(err) => { + tracing::debug!(%url, error = %err, "A2A agent card candidate did not answer"); + last_err = Some(err); + } + } + } + Err(last_err.expect("agent_card_urls always yields at least one candidate")) + } async fn send(&self, request: &serde_json::Value) -> Result { let resp = self - .apply_auth( + .prepare( self.client .post(&self.upstream.url) .timeout(self.upstream.timeout) @@ -368,22 +432,82 @@ mod tests { let up = A2aUpstream { url: "https://x/a2a".into(), auth: A2aAuth::Bearer("super-secret".into()), + protocol_version: A2aProtocolVersion::V1_0, timeout: DEFAULT_UPSTREAM_TIMEOUT, }; assert!(!format!("{up:?}").contains("super-secret")); } - #[test] - fn agent_card_url_is_origin_well_known() { - let bridge = HttpBridge::new(A2aUpstream { - url: "https://agents.example.com/a2a/v1".into(), + fn bridge_at(url: &str) -> HttpBridge { + HttpBridge::new(A2aUpstream { + url: url.into(), auth: A2aAuth::None, + protocol_version: A2aProtocolVersion::V1_0, timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + }) + } + + fn card_candidates(url: &str) -> Vec { + bridge_at(url) + .agent_card_urls() + .unwrap() + .iter() + .map(|u| u.as_str().to_string()) + .collect() + } + + #[test] + fn card_candidates_try_the_registered_path_before_the_origin() { + // #913: `set_path` used to replace the path outright, so a path-hosted + // agent was asked for a card URI it never publishes. The prefix now + // survives, and the origin URI stays as the fallback so agents + // registered under the old behaviour keep resolving. + assert_eq!( + card_candidates("https://agents.example.com/v3/a2a/serve/abc"), + vec![ + "https://agents.example.com/v3/a2a/serve/abc/.well-known/agent-card.json", + "https://agents.example.com/.well-known/agent-card.json", + "https://agents.example.com/v3/a2a/serve/abc/.well-known/agent.json", + "https://agents.example.com/.well-known/agent.json", + ] + ); + } + + #[test] + fn card_candidates_collapse_when_the_agent_owns_the_origin() { + // The two bases coincide here, so the relative candidate would be a + // byte-identical second request. A trailing slash is the same case. + let expected = vec![ + "https://agents.example.com/.well-known/agent-card.json", + "https://agents.example.com/.well-known/agent.json", + ]; + assert_eq!(card_candidates("https://agents.example.com"), expected); + assert_eq!(card_candidates("https://agents.example.com/"), expected); + } + + #[test] + fn card_candidates_drop_the_query_and_keep_the_port() { + assert_eq!( + card_candidates("http://127.0.0.1:8080/a2a?tenant=acme"), + vec![ + "http://127.0.0.1:8080/a2a/.well-known/agent-card.json", + "http://127.0.0.1:8080/.well-known/agent-card.json", + "http://127.0.0.1:8080/a2a/.well-known/agent.json", + "http://127.0.0.1:8080/.well-known/agent.json", + ] + ); + } + + #[test] + fn upstream_carries_the_pinned_protocol_version() { + let mut pinned_03 = agent("none"); + pinned_03.protocol_version = A2aProtocolVersion::V0_3; assert_eq!( - bridge.agent_card_url().unwrap().as_str(), - "https://agents.example.com/.well-known/agent-card.json" + upstream_from_a2a_agent(&pinned_03).protocol_version, + A2aProtocolVersion::V0_3 ); + assert_eq!(A2aProtocolVersion::V1_0.as_wire_str(), "1.0"); + assert_eq!(A2aProtocolVersion::V0_3.as_wire_str(), "0.3"); } #[test] diff --git a/crates/aisix-a2a/tests/upstream_roundtrip.rs b/crates/aisix-a2a/tests/upstream_roundtrip.rs index 39c54464..650338ef 100644 --- a/crates/aisix-a2a/tests/upstream_roundtrip.rs +++ b/crates/aisix-a2a/tests/upstream_roundtrip.rs @@ -11,23 +11,44 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use aisix_a2a::{A2aAuth, A2aBridge, A2aError, A2aUpstream, HttpBridge, DEFAULT_UPSTREAM_TIMEOUT}; +use aisix_core::A2aProtocolVersion; use axum::http::header::LOCATION; use axum::http::{HeaderMap, StatusCode}; use axum::response::IntoResponse; -use axum::routing::{get, post}; +use axum::routing::{any, get, post}; use axum::{Json, Router}; use serde_json::{json, Value}; +/// An upstream pinned to A2A 1.0 — the default for a registered agent. +fn upstream(url: String, auth: A2aAuth) -> A2aUpstream { + A2aUpstream { + url, + auth, + protocol_version: A2aProtocolVersion::V1_0, + timeout: DEFAULT_UPSTREAM_TIMEOUT, + } +} + +/// Read back the `A2A-Version` an inbound request carried, or `null`. +fn seen_version(headers: &HeaderMap) -> Value { + headers + .get("a2a-version") + .and_then(|v| v.to_str().ok()) + .map(|v| Value::String(v.to_string())) + .unwrap_or(Value::Null) +} + /// A minimal upstream A2A agent: serves its card at the well-known URI and /// answers JSON-RPC by echoing back the request id and the credentials it saw, /// so the test can assert what the gateway forwarded. async fn spawn_agent() -> SocketAddr { - async fn card() -> Json { + async fn card(headers: HeaderMap) -> Json { Json(json!({ "name": "Test Agent", "url": "https://upstream.example.com/a2a", "version": "1.0.0", - "skills": [{"id": "echo", "name": "Echo"}] + "skills": [{"id": "echo", "name": "Echo"}], + "echoed_version": seen_version(&headers), })) } @@ -49,6 +70,7 @@ async fn spawn_agent() -> SocketAddr { "status": {"state": "completed"}, "echoed_auth": auth, "echoed_api_key": api_key, + "echoed_version": seen_version(&headers), } })) } @@ -84,11 +106,10 @@ fn message_send(id: &str) -> Value { #[tokio::test] async fn fetches_card_and_forwards_bearer() { let addr = spawn_agent().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/a2a"), - auth: A2aAuth::Bearer("tok-123".into()), - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream( + format!("http://{addr}/a2a"), + A2aAuth::Bearer("tok-123".into()), + )); let card = bridge.fetch_agent_card().await.unwrap(); assert_eq!(card.name, "Test Agent"); @@ -105,11 +126,10 @@ async fn fetches_card_and_forwards_bearer() { #[tokio::test] async fn forwards_api_key_header() { let addr = spawn_agent().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/a2a"), - auth: A2aAuth::ApiKey("k-secret".into()), - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream( + format!("http://{addr}/a2a"), + A2aAuth::ApiKey("k-secret".into()), + )); let resp = bridge.send(&message_send("req-2")).await.unwrap(); assert_eq!(resp["result"]["echoed_api_key"], "k-secret"); @@ -120,11 +140,7 @@ async fn forwards_api_key_header() { #[tokio::test] async fn sends_no_credential_when_none() { let addr = spawn_agent().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/a2a"), - auth: A2aAuth::None, - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream(format!("http://{addr}/a2a"), A2aAuth::None)); let resp = bridge.send(&message_send("req-3")).await.unwrap(); assert!(resp["result"]["echoed_auth"].is_null()); @@ -165,11 +181,7 @@ async fn spawn_redirect_probe() -> (SocketAddr, Arc) { #[tokio::test] async fn refuses_to_follow_upstream_redirect() { let (addr, secret_hits) = spawn_redirect_probe().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/redirect"), - auth: A2aAuth::None, - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream(format!("http://{addr}/redirect"), A2aAuth::None)); let err = bridge.send(&message_send("r")).await.unwrap_err(); // The 302 surfaces as a non-success status error — it is NOT followed. @@ -183,3 +195,116 @@ async fn refuses_to_follow_upstream_redirect() { "the redirect target must NOT be fetched — the gateway must not chase upstream redirects" ); } + +/// An upstream that publishes its card ONLY under the agent's own path prefix +/// and answers every unrouted path with the catch-all `405` the real one +/// returns. This is the shape of any platform that multiplexes tenants under a +/// path, and of any self-hosted agent behind an ingress path. +async fn spawn_path_hosted_agent() -> SocketAddr { + async fn card(headers: HeaderMap) -> Json { + Json(json!({ + "name": "Path Hosted Agent", + "url": "https://upstream.example.com/v3/a2a/serve/agent-42", + "protocolVersion": "0.3.0", + "echoed_version": seen_version(&headers), + })) + } + + async fn rpc(Json(body): Json) -> Json { + Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": {"kind": "task", "id": "task-path", "status": {"state": "completed"}} + })) + } + + let app = Router::new() + .route( + "/v3/a2a/serve/agent-42/.well-known/agent-card.json", + get(card), + ) + .route("/v3/a2a/serve/agent-42", post(rpc)) + .fallback(any(|| async { StatusCode::METHOD_NOT_ALLOWED })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app.into_make_service()) + .await + .unwrap(); + }); + addr +} + +#[tokio::test] +async fn discovers_a_card_hosted_under_the_agent_path() { + // #913: the registered path used to be discarded, so the bridge asked the + // ORIGIN for a card this agent publishes only under its prefix, and took + // the catch-all 405 as the agent's answer. No configuration could unblock + // it — the one `url` field feeds both the card fetch and the RPC endpoint. + let addr = spawn_path_hosted_agent().await; + let bridge = HttpBridge::new(upstream( + format!("http://{addr}/v3/a2a/serve/agent-42"), + A2aAuth::None, + )); + + let card = bridge.fetch_agent_card().await.unwrap(); + assert_eq!(card.name, "Path Hosted Agent"); + // And the endpoint itself still resolves off the same registered URL. + let resp = bridge.send(&message_send("p-1")).await.unwrap(); + assert_eq!(resp["result"]["id"], "task-path"); +} + +#[tokio::test] +async fn still_discovers_a_card_hosted_at_the_origin() { + // The origin URI remains a candidate, so every agent registered while it + // was the ONLY candidate keeps resolving. `spawn_agent` serves its card + // there and nowhere else. + let addr = spawn_agent().await; + let bridge = HttpBridge::new(upstream(format!("http://{addr}/a2a"), A2aAuth::None)); + + assert_eq!(bridge.fetch_agent_card().await.unwrap().name, "Test Agent"); +} + +#[tokio::test] +async fn reports_a_failure_when_no_candidate_serves_a_card() { + let addr = spawn_path_hosted_agent().await; + let bridge = HttpBridge::new(upstream(format!("http://{addr}/nope"), A2aAuth::None)); + + let err = bridge.fetch_agent_card().await.unwrap_err(); + assert!( + matches!(err, A2aError::Connect(_)), + "exhausting every candidate must surface the upstream failure, got {err:?}" + ); +} + +#[tokio::test] +async fn announces_the_pinned_wire_version_on_every_upstream_call() { + // #911: the gateway sent no `A2A-Version` at all. The spec makes an agent + // read an absent value as 0.3, so a 1.0-pinned agent rejected every call + // with VersionNotSupportedError and `protocol_version` was inert. + let addr = spawn_agent().await; + + let pinned_10 = HttpBridge::new(upstream(format!("http://{addr}/a2a"), A2aAuth::None)); + assert_eq!( + pinned_10.fetch_agent_card().await.unwrap().rest["echoed_version"], + "1.0", + "the card fetch must announce the pinned version too" + ); + assert_eq!( + pinned_10.send(&message_send("v-1")).await.unwrap()["result"]["echoed_version"], + "1.0" + ); + + let pinned_03 = HttpBridge::new(A2aUpstream { + protocol_version: A2aProtocolVersion::V0_3, + ..upstream(format!("http://{addr}/a2a"), A2aAuth::None) + }); + assert_eq!( + pinned_03.fetch_agent_card().await.unwrap().rest["echoed_version"], + "0.3" + ); + assert_eq!( + pinned_03.send(&message_send("v-0")).await.unwrap()["result"]["echoed_version"], + "0.3" + ); +} diff --git a/crates/aisix-core/src/models/a2a_agent.rs b/crates/aisix-core/src/models/a2a_agent.rs index 2ea8d5be..6601d6ca 100644 --- a/crates/aisix-core/src/models/a2a_agent.rs +++ b/crates/aisix-core/src/models/a2a_agent.rs @@ -40,15 +40,19 @@ pub struct A2aAgent { )] pub name: String, - /// The upstream agent's base URL, such as `https://agents.example.com/a2a`. - /// AISIX reaches this URL over HTTP with the A2A JSON-RPC 2.0 protocol and - /// discovers the agent card relative to it. + /// The upstream agent's A2A service endpoint, such as + /// `https://agents.example.com/a2a`, where AISIX sends JSON-RPC 2.0 requests + /// over HTTP. AISIX looks for the agent card at the well-known path under + /// this URL's own path first, then under its origin, so both an agent that + /// owns its domain and one published under a path prefix are reachable + /// without extra configuration. #[schemars(length(min = 1))] pub url: String, - /// The A2A wire-format version AISIX uses for this agent. AISIX pins the - /// version explicitly so the served agent card and accepted requests stay - /// consistent. + /// The A2A wire-format version this agent speaks. AISIX announces it to the + /// agent in the `A2A-Version` header on every request, so it must match what + /// the agent actually serves: an agent reads an absent or mismatched version + /// as a protocol error and rejects the call. #[serde(default)] pub protocol_version: A2aProtocolVersion, @@ -106,6 +110,18 @@ pub enum A2aProtocolVersion { V0_3, } +impl A2aProtocolVersion { + /// The value this version carries on the wire, in the `A2A-Version` header + /// and in the `protocolVersion` field of an agent card. Kept in lockstep + /// with the `serde` renames above, which are the same strings. + pub fn as_wire_str(self) -> &'static str { + match self { + Self::V1_0 => "1.0", + Self::V0_3 => "0.3", + } + } +} + /// How the gateway authenticates to an upstream A2A agent. #[derive( Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs index 67eea0e1..144cc821 100644 --- a/crates/aisix-proxy/src/a2a.rs +++ b/crates/aisix-proxy/src/a2a.rs @@ -251,16 +251,42 @@ pub async fn a2a_agent_card( return (StatusCode::BAD_GATEWAY, err.to_string()).into_response(); } }; - // Rewrite the advertised service endpoint to the gateway so downstream + // Rewrite the advertised service endpoints to the gateway so downstream // callers route subsequent requests through `/a2a/`. Derived from // the request's Host header (and forwarded scheme) since the gateway's // public URL is not otherwise known here. if let Some(base) = gateway_base(&headers) { - card.url = format!("{base}/a2a/{agent}"); + rewrite_card_urls(&mut card, &format!("{base}/a2a/{agent}")); } axum::Json(card).into_response() } +/// Point every service URL the card advertises at the gateway. +/// +/// The top-level `url` is what a 0.3 caller reads. A 1.0 caller instead picks +/// its endpoint out of `supportedInterfaces` (`additionalInterfaces` on a 0.3 +/// card), so rewriting only the top level leaves it reading the upstream +/// address off the card and calling the agent directly — no auth, no quota, no +/// usage, and the internal address handed to the caller on the way past. +/// +/// Entries are rewritten in place rather than filtered out: this gateway serves +/// JSON-RPC over HTTP only, so an entry naming a transport it does not proxy +/// now points somewhere that will reject the caller. That is the intended +/// trade — failing loudly beats silently bypassing governance. +fn rewrite_card_urls(card: &mut aisix_a2a::AgentCard, gateway_url: &str) { + card.url = gateway_url.to_string(); + for key in ["supportedInterfaces", "additionalInterfaces"] { + let Some(serde_json::Value::Array(interfaces)) = card.rest.get_mut(key) else { + continue; + }; + for interface in interfaces { + if let Some(url) = interface.get_mut("url") { + *url = serde_json::Value::String(gateway_url.to_string()); + } + } + } +} + /// Reconstruct the gateway's public base (`scheme://host`) from request /// headers: the `Host` header, and `X-Forwarded-Proto` when a proxy set it /// (defaulting to `https`). Returns `None` when no Host header is present, in @@ -369,6 +395,49 @@ mod tests { assert_eq!(gateway_base(&HeaderMap::new()), None); } + #[test] + fn card_rewrite_leaves_no_upstream_url_for_a_caller_to_follow() { + // Only the top-level `url` used to be rewritten. A 1.0 caller reads its + // endpoint out of `supportedInterfaces` instead, so it went straight to + // the upstream — past auth, quota and usage — and read the internal + // address off the card on the way (#911). + let mut card: aisix_a2a::AgentCard = serde_json::from_str( + r#"{ + "name": "Agent", + "url": "https://internal.upstream/a2a", + "supportedInterfaces": [ + {"url": "https://internal.upstream/a2a", "protocolBinding": "JSONRPC"}, + {"url": "https://internal.upstream/grpc", "protocolBinding": "GRPC"} + ], + "additionalInterfaces": [ + {"url": "https://internal.upstream/rest", "transport": "HTTP+JSON"} + ], + "skills": [{"id": "s1"}] + }"#, + ) + .unwrap(); + + rewrite_card_urls(&mut card, "https://gw.example.com/a2a/billing"); + + let served = serde_json::to_string(&card).unwrap(); + assert!( + !served.contains("internal.upstream"), + "no upstream address may survive anywhere in the served card:\n{served}" + ); + assert_eq!(card.url, "https://gw.example.com/a2a/billing"); + for key in ["supportedInterfaces", "additionalInterfaces"] { + for interface in card.rest[key].as_array().unwrap() { + assert_eq!(interface["url"], "https://gw.example.com/a2a/billing"); + } + } + // Everything the gateway does not own is passed through untouched. + assert_eq!(card.rest["skills"][0]["id"], "s1"); + assert_eq!( + card.rest["supportedInterfaces"][1]["protocolBinding"], + "GRPC" + ); + } + // ---- endpoint integration tests: drive the real router via oneshot ---- use crate::build_router; use aisix_core::{A2aAgent, AisixSnapshot, ApiKey, ProxyConfig, ResourceEntry, SnapshotHandle}; diff --git a/schemas/resources/a2a_agent.schema.json b/schemas/resources/a2a_agent.schema.json index d7c95a8d..050465a9 100644 --- a/schemas/resources/a2a_agent.schema.json +++ b/schemas/resources/a2a_agent.schema.json @@ -149,7 +149,7 @@ } ], "default": "1.0", - "description": "The A2A wire-format version AISIX uses for this agent. AISIX pins the version explicitly so the served agent card and accepted requests stay consistent." + "description": "The A2A wire-format version this agent speaks. AISIX announces it to the agent in the `A2A-Version` header on every request, so it must match what the agent actually serves: an agent reads an absent or mismatched version as a protocol error and rejects the call." }, "secret": { "description": "Credential AISIX uses to authenticate to the upstream agent. For `bearer`, AISIX sends it as `Authorization: Bearer `; for `api_key`, AISIX sends it as `x-api-key: `. Leave unset for `none`.", @@ -168,7 +168,7 @@ ] }, "url": { - "description": "The upstream agent's base URL, such as `https://agents.example.com/a2a`. AISIX reaches this URL over HTTP with the A2A JSON-RPC 2.0 protocol and discovers the agent card relative to it.", + "description": "The upstream agent's A2A service endpoint, such as `https://agents.example.com/a2a`, where AISIX sends JSON-RPC 2.0 requests over HTTP. AISIX looks for the agent card at the well-known path under this URL's own path first, then under its origin, so both an agent that owns its domain and one published under a path prefix are reachable without extra configuration.", "minLength": 1, "type": "string" } diff --git a/tests/e2e/src/cases/a2a-gateway-e2e.test.ts b/tests/e2e/src/cases/a2a-gateway-e2e.test.ts new file mode 100644 index 00000000..56453334 --- /dev/null +++ b/tests/e2e/src/cases/a2a-gateway-e2e.test.ts @@ -0,0 +1,263 @@ +import { createHash, randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + EtcdClient, + SeedClient, + spawnApp, + startA2aUpstream, + waitConfigPropagation, + type A2aUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E: the `/a2a/{agent}` gateway endpoint against a real gateway + etcd + real +// upstream A2A agents. The A2A MVP (#717) deferred endpoint-level e2e; this is +// it, written around the two faults that made the endpoint unusable in +// practice. +// +// Pinned contract: +// - every upstream call announces the agent's pinned wire version in +// `A2A-Version`, on the card fetch as well as the JSON-RPC call (#911) — +// without it an agent must read the call as 0.3 and reject a 1.0 body; +// - a `0.3`-pinned agent is announced as 0.3, so the pin is what is sent +// rather than a constant; +// - an agent whose card is published under its own path prefix resolves, and +// the catch-all 405 its platform returns at the origin is not mistaken for +// the agent's answer (#913); +// - the served card carries NO upstream address: the top-level `url` and +// every `supportedInterfaces[].url` point back at the gateway, so a 1.0 +// caller (which reads its endpoint out of `supportedInterfaces`) cannot +// route around the gateway; +// - the upstream credential is presented by the gateway and never reaches +// the caller; the caller's own key never reaches the upstream; +// - per-agent ACL and unknown/disabled agents still gate the endpoint. + +const KEY_ALLOWED = "sk-a2a-e2e-allowed"; +const KEY_NO_AGENTS = "sk-a2a-e2e-no-agents"; + +const sha256 = (s: string) => createHash("sha256").update(s).digest("hex"); + +const messageSend = (id: string) => ({ + jsonrpc: "2.0", + id, + method: "message/send", + params: { + message: { + role: "user", + parts: [{ kind: "text", text: "invoice 42" }], + messageId: "m-1", + }, + }, +}); + +describe("a2a gateway e2e: /a2a/{agent}", () => { + let app: SpawnedApp | undefined; + let rootHosted: A2aUpstream | undefined; + let pathHosted: A2aUpstream | undefined; + let etcdReachable = false; + let seed: SeedClient; + + // The gate responses (401 / 403 / 404) are plain text, not JSON-RPC + // envelopes, so the body is parsed opportunistically. + const readBody = (text: string): Record | undefined => { + if (!text) return undefined; + try { + return JSON.parse(text) as Record; + } catch { + return undefined; + } + }; + + const call = async (path: string, token: string, body: unknown) => { + const res = await fetch(`${app!.proxyUrl}${path}`, { + method: "POST", + headers: { + authorization: `Bearer ${token}`, + "content-type": "application/json", + }, + body: JSON.stringify(body), + }); + const text = await res.text(); + return { status: res.status, raw: text, json: readBody(text) }; + }; + + const fetchCard = async (agent: string, token: string) => { + const res = await fetch( + `${app!.proxyUrl}/a2a/${agent}/.well-known/agent-card.json`, + { headers: { authorization: `Bearer ${token}` } }, + ); + const text = await res.text(); + return { status: res.status, raw: text, json: readBody(text) }; + }; + + beforeAll(async () => { + const etcd = new EtcdClient(); + etcdReachable = await etcd.ping(); + if (!etcdReachable) return; + + rootHosted = await startA2aUpstream({ + cardMount: "origin", + token: "upstream-secret-tok", + }); + pathHosted = await startA2aUpstream({ cardMount: "path" }); + app = await spawnApp(); + seed = new SeedClient(etcd, app.etcdPrefix); + + await seed.update("a2a_agents", randomUUID(), { + name: "invoices", + url: rootHosted.url, + protocol_version: "1.0", + auth_type: "bearer", + secret: "upstream-secret-tok", + enabled: true, + }); + await seed.update("a2a_agents", randomUUID(), { + name: "legacy", + url: rootHosted.url, + protocol_version: "0.3", + auth_type: "bearer", + secret: "upstream-secret-tok", + enabled: true, + }); + await seed.update("a2a_agents", randomUUID(), { + name: "tenant", + url: pathHosted.url, + protocol_version: "1.0", + auth_type: "none", + enabled: true, + }); + await seed.update("a2a_agents", randomUUID(), { + name: "retired", + url: rootHosted.url, + protocol_version: "1.0", + auth_type: "none", + enabled: false, + }); + + await seed.createApiKey({ + key_hash: sha256(KEY_ALLOWED), + allowed_models: [], + allowed_agents: ["*"], + }); + await seed.createApiKey({ + key_hash: sha256(KEY_NO_AGENTS), + allowed_models: [], + }); + + await waitConfigPropagation(async () => { + const card = await fetchCard("invoices", KEY_ALLOWED); + if (card.status !== 200) return false; + const denied = await fetchCard("invoices", KEY_NO_AGENTS); + return denied.status === 403; + }); + }, 60_000); + + afterAll(async () => { + await app?.exit(); + await rootHosted?.close(); + await pathHosted?.close(); + }); + + test("announces the pinned wire version on every upstream call", async (ctx) => { + if (!etcdReachable || !app || !rootHosted) return ctx.skip(); + rootHosted.requests.length = 0; + + await fetchCard("invoices", KEY_ALLOWED); + const reply = await call("/a2a/invoices", KEY_ALLOWED, messageSend("e2e-1")); + + expect(reply.status).toBe(200); + expect(reply.json?.result?.sawVersion).toBe("1.0"); + expect(reply.json?.result?.sawMethod).toBe("message/send"); + // A card fetch is an A2A request like any other, so EVERY hop carries the + // version — including the candidate probes that miss. + expect(rootHosted.requests.map((r) => r.version)).not.toContain(null); + expect(rootHosted.requests.every((r) => r.version === "1.0")).toBe(true); + expect(rootHosted.requests.some((r) => r.httpMethod === "GET")).toBe(true); + expect(rootHosted.requests.some((r) => r.httpMethod === "POST")).toBe(true); + }); + + test("announces 0.3 for an agent pinned to 0.3", async (ctx) => { + if (!etcdReachable || !app || !rootHosted) return ctx.skip(); + rootHosted.requests.length = 0; + + const reply = await call("/a2a/legacy", KEY_ALLOWED, messageSend("e2e-2")); + + expect(reply.status).toBe(200); + expect(reply.json?.result?.sawVersion).toBe("0.3"); + expect(rootHosted.requests.at(-1)?.version).toBe("0.3"); + }); + + test("reaches an agent whose card is published under a path prefix", async (ctx) => { + if (!etcdReachable || !app || !pathHosted) return ctx.skip(); + pathHosted.requests.length = 0; + + const card = await fetchCard("tenant", KEY_ALLOWED); + expect(card.status).toBe(200); + expect(card.json?.name).toBe("Invoice Processor"); + + // Resolved on the first candidate: the agent's own path prefix is tried + // before the origin, so the platform's catch-all 405 at the origin is never + // even reached — which is exactly what used to be mistaken for the agent's + // answer when the origin was the ONLY candidate. + const cardPaths = pathHosted.requests + .filter((r) => r.httpMethod === "GET") + .map((r) => r.path); + expect(cardPaths).toEqual([ + "/v3/agents/serve/tenant-42/.well-known/agent-card.json", + ]); + + const reply = await call("/a2a/tenant", KEY_ALLOWED, messageSend("e2e-3")); + expect(reply.status).toBe(200); + expect(reply.json?.result?.id).toBe("task-e2e-1"); + }); + + test("the served card points every endpoint back at the gateway", async (ctx) => { + if (!etcdReachable || !app) return ctx.skip(); + + const card = await fetchCard("invoices", KEY_ALLOWED); + + expect(card.status).toBe(200); + expect(card.raw).not.toContain("internal.upstream.invalid"); + expect(card.json?.url).toMatch(/\/a2a\/invoices$/); + for (const iface of card.json?.supportedInterfaces ?? []) { + expect(iface.url).toMatch(/\/a2a\/invoices$/); + } + // Everything the gateway does not own survives the rewrite. + expect(card.json?.skills?.[0]?.id).toBe("invoice"); + expect(card.json?.version).toBe("2.1.0"); + }); + + test("the gateway holds the upstream credential and the caller never sees it", async (ctx) => { + if (!etcdReachable || !app || !rootHosted) return ctx.skip(); + rootHosted.requests.length = 0; + + const reply = await call("/a2a/invoices", KEY_ALLOWED, messageSend("e2e-4")); + + expect(reply.status).toBe(200); + const seen = rootHosted.requests.at(-1); + expect(seen?.authorization).toBe("Bearer upstream-secret-tok"); + // The caller's own AISIX key must never be forwarded as the upstream token. + expect(seen?.authorization).not.toContain(KEY_ALLOWED); + expect(JSON.stringify(reply.json)).not.toContain("upstream-secret-tok"); + }); + + test("gates on the per-agent ACL and on agent existence", async (ctx) => { + if (!etcdReachable || !app) return ctx.skip(); + + const denied = await call("/a2a/invoices", KEY_NO_AGENTS, messageSend("x")); + expect(denied.status).toBe(403); + + const unknown = await call("/a2a/ghost", KEY_ALLOWED, messageSend("x")); + expect(unknown.status).toBe(404); + + const disabled = await call("/a2a/retired", KEY_ALLOWED, messageSend("x")); + expect(disabled.status).toBe(404); + + const unauthenticated = await fetch(`${app.proxyUrl}/a2a/invoices`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(messageSend("x")), + }); + expect(unauthenticated.status).toBe(401); + }); +}); diff --git a/tests/e2e/src/harness/index.ts b/tests/e2e/src/harness/index.ts index 7830d1fb..4b28316e 100644 --- a/tests/e2e/src/harness/index.ts +++ b/tests/e2e/src/harness/index.ts @@ -5,6 +5,12 @@ export { EtcdClient } from "./etcd.js"; export { SeedClient } from "./seed.js"; export { startOpenAiUpstream, type OpenAiUpstream, type ReceivedRequest } from "./upstream-openai.js"; export { startMcpUpstream, type McpUpstream } from "./upstream-mcp.js"; +export { + startA2aUpstream, + type A2aUpstream, + type A2aCardMount, + type A2aReceivedRequest, +} from "./upstream-a2a.js"; export { startRestUpstream, type RestUpstream } from "./upstream-rest.js"; export { pickFreePort, pickFreePorts } from "./ports.js"; export { diff --git a/tests/e2e/src/harness/upstream-a2a.ts b/tests/e2e/src/harness/upstream-a2a.ts new file mode 100644 index 00000000..db89b707 --- /dev/null +++ b/tests/e2e/src/harness/upstream-a2a.ts @@ -0,0 +1,167 @@ +import { + createServer, + type IncomingMessage, + type Server as HttpServer, + type ServerResponse, +} from "node:http"; + +/** One request the stub agent received, as the gateway sent it. */ +export interface A2aReceivedRequest { + httpMethod: string; + path: string; + /** The `A2A-Version` header, or `null` when the gateway sent none. */ + version: string | null; + authorization: string | null; + apiKey: string | null; + body?: Record; +} + +/** + * Where the agent publishes its card. + * + * - `origin` — at the RFC 8615 origin URI, the shape of an agent that owns its + * whole domain. This is the only shape the gateway used to be able to reach. + * - `path` — under the service endpoint's own path prefix, with every other + * path answering `405` from a catch-all. This is the shape of any platform + * that multiplexes tenants under a prefix, and of a self-hosted agent behind + * an ingress path (api7/aisix#913). + */ +export type A2aCardMount = "origin" | "path"; + +export interface A2aUpstream { + /** Register this as the agent's `url`: the A2A service endpoint. */ + url: string; + /** Every request the agent received, in arrival order. */ + requests: A2aReceivedRequest[]; + close(): Promise; +} + +export interface A2aUpstreamOptions { + cardMount?: A2aCardMount; + /** When set, the agent is registered with `auth_type: bearer` and this token. */ + token?: string; +} + +const PATH_PREFIX = "/v3/agents/serve/tenant-42"; + +/** + * A stub upstream A2A agent: serves an agent card and answers JSON-RPC, while + * recording what the gateway actually sent it — the wire version it announced + * and the credential it presented. + * + * The card it serves deliberately advertises an unreachable `https://` address + * both at the top level and inside `supportedInterfaces`, so a test can prove + * the gateway rewrote every one of them before handing the card to a caller. + */ +export async function startA2aUpstream( + options: A2aUpstreamOptions = {}, +): Promise { + const mount: A2aCardMount = options.cardMount ?? "origin"; + const requests: A2aReceivedRequest[] = []; + const servicePath = mount === "path" ? PATH_PREFIX : "/a2a"; + const cardPath = + mount === "path" + ? `${PATH_PREFIX}/.well-known/agent-card.json` + : "/.well-known/agent-card.json"; + + const httpServer: HttpServer = createServer((req, res) => { + void handle(req, res, { requests, servicePath, cardPath, token: options.token }); + }); + await new Promise((resolve) => + httpServer.listen(0, "127.0.0.1", resolve), + ); + const address = httpServer.address(); + if (address === null || typeof address === "string") { + throw new Error("a2a upstream: no listen address"); + } + + return { + url: `http://127.0.0.1:${address.port}${servicePath}`, + requests, + close: () => new Promise((resolve) => httpServer.close(() => resolve())), + }; +} + +async function handle( + req: IncomingMessage, + res: ServerResponse, + ctx: { + requests: A2aReceivedRequest[]; + servicePath: string; + cardPath: string; + token?: string; + }, +): Promise { + const path = new URL(req.url ?? "/", "http://127.0.0.1").pathname; + const header = (name: string): string | null => { + const value = req.headers[name]; + return typeof value === "string" ? value : null; + }; + const send = (status: number, body: unknown): void => { + res.writeHead(status, { "content-type": "application/json" }); + res.end(JSON.stringify(body)); + }; + + let body: Record | undefined; + if (req.method === "POST") { + const chunks: Buffer[] = []; + for await (const chunk of req) chunks.push(chunk as Buffer); + const raw = Buffer.concat(chunks).toString("utf8"); + body = raw ? (JSON.parse(raw) as Record) : undefined; + } + + ctx.requests.push({ + httpMethod: req.method ?? "GET", + path, + version: header("a2a-version"), + authorization: header("authorization"), + apiKey: header("x-api-key"), + body, + }); + + if (ctx.token !== undefined && header("authorization") !== `Bearer ${ctx.token}`) { + send(401, { error: "unauthorized" }); + return; + } + + if (req.method === "GET" && path === ctx.cardPath) { + send(200, { + name: "Invoice Processor", + description: "Stub agent for the A2A gateway e2e.", + protocolVersion: "1.0", + version: "2.1.0", + url: "https://internal.upstream.invalid/a2a", + supportedInterfaces: [ + { + url: "https://internal.upstream.invalid/a2a", + protocolBinding: "JSONRPC", + protocolVersion: "1.0", + }, + ], + skills: [{ id: "invoice", name: "Process invoice", tags: ["billing"] }], + }); + return; + } + + if (req.method === "POST" && path === ctx.servicePath) { + send(200, { + jsonrpc: "2.0", + id: body?.id ?? null, + result: { + kind: "task", + id: "task-e2e-1", + status: { state: "completed" }, + // Echoed so the gateway's forwarding can be asserted from the caller + // side as well as from `requests`. + sawVersion: header("a2a-version"), + sawMethod: body?.method ?? null, + }, + }); + return; + } + + // The catch-all a real path-hosting agent platform answers with: not a 404, + // which is what made the original report look like a missing card rather + // than a mis-built URL. + send(405, { error: "method not allowed" }); +} From b4d3bcab58e37f2bd539c2c9a9f6f6c3579c89c7 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Mon, 10 Aug 2026 14:08:51 +0800 Subject: [PATCH 2/3] fix(a2a): bound the card-fetch walk and never serve an unrewritten card MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the two fixes. The candidate walk gave each candidate the full `timeout_ms`, so a hung agent stretched a card fetch to `candidates × timeout_ms` — 120s on the default — against a field documented as bounding one upstream operation. The walk now shares a single deadline and passes each request only what is left of it. `gateway_base` returned `None` when no `Host` header was present, and the card was then served unrewritten — advertising the upstream's own address, which is precisely what the rewrite exists to prevent. That is reachable: axum-server sets ALPN to `h2, http/1.1`, so this listener negotiates HTTP/2 whenever `proxy.tls` is configured, and an h2 request carries its authority as the `:authority` pseudo-header with no `Host` at all. The authority now falls back to the request URI's, and a request with neither is refused with a 500 rather than answered with a card that leaks the upstream. The base is resolved before the upstream is contacted, so the failure costs no round trip. The e2e readiness gate probed the card endpoint, which is behaviour under test: a regression in discovery or the ACL would have surfaced as a 60s propagation timeout in `beforeAll` instead of naming the defect. It now probes only that the seeded rows reached the snapshot — not 404, not 401 — and asserts nothing else. --- crates/aisix-a2a/src/bridge.rs | 34 +++++++-- crates/aisix-a2a/tests/upstream_roundtrip.rs | 42 +++++++++++ crates/aisix-proxy/src/a2a.rs | 79 ++++++++++++++++---- tests/e2e/src/cases/a2a-gateway-e2e.test.ts | 12 ++- 4 files changed, 141 insertions(+), 26 deletions(-) diff --git a/crates/aisix-a2a/src/bridge.rs b/crates/aisix-a2a/src/bridge.rs index fcd1cdb7..3385ab91 100644 --- a/crates/aisix-a2a/src/bridge.rs +++ b/crates/aisix-a2a/src/bridge.rs @@ -26,7 +26,7 @@ //! use std::sync::OnceLock; -use std::time::Duration; +use std::time::{Duration, Instant}; use aisix_core::{A2aAgent, A2aAuthType, A2aProtocolVersion}; use async_trait::async_trait; @@ -314,10 +314,15 @@ impl HttpBridge { Ok(urls) } - /// Fetch and parse the card at exactly one candidate URI. - async fn fetch_card_at(&self, url: reqwest::Url) -> Result { + /// Fetch and parse the card at exactly one candidate URI, within whatever + /// is left of the fetch's overall budget. + async fn fetch_card_at( + &self, + url: reqwest::Url, + budget: Duration, + ) -> Result { let resp = self - .prepare(self.client.get(url).timeout(self.upstream.timeout)) + .prepare(self.client.get(url).timeout(budget)) .send() .await .map_err(|e| A2aError::Connect(e.to_string()))?; @@ -340,9 +345,24 @@ impl A2aBridge for HttpBridge { // that does not route the URI answers with whatever its catch-all // returns (the 405 in #913), and a prefix that happens to resolve can // still hand back something that is not a card. + // + // `timeout_ms` bounds the card fetch as ONE upstream operation, so the + // whole walk shares a single deadline instead of handing each candidate + // a fresh one. Otherwise a slow or hung upstream stretches the fetch to + // `candidates × timeout_ms` — four times what the operator configured — + // and pins a gateway request for the duration. + let deadline = Instant::now() + self.upstream.timeout; let mut last_err = None; for url in self.agent_card_urls()? { - match self.fetch_card_at(url.clone()).await { + let budget = deadline.saturating_duration_since(Instant::now()); + if budget.is_zero() { + tracing::debug!( + agent_url = %self.upstream.url, + "A2A agent card fetch exhausted its deadline before trying every candidate" + ); + break; + } + match self.fetch_card_at(url.clone(), budget).await { Ok(card) => return Ok(card), Err(err) => { tracing::debug!(%url, error = %err, "A2A agent card candidate did not answer"); @@ -350,7 +370,9 @@ impl A2aBridge for HttpBridge { } } } - Err(last_err.expect("agent_card_urls always yields at least one candidate")) + Err(last_err.unwrap_or_else(|| { + A2aError::Connect("agent card fetch exceeded its timeout".to_string()) + })) } async fn send(&self, request: &serde_json::Value) -> Result { diff --git a/crates/aisix-a2a/tests/upstream_roundtrip.rs b/crates/aisix-a2a/tests/upstream_roundtrip.rs index 650338ef..dc2c7e83 100644 --- a/crates/aisix-a2a/tests/upstream_roundtrip.rs +++ b/crates/aisix-a2a/tests/upstream_roundtrip.rs @@ -9,6 +9,7 @@ use std::net::SocketAddr; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::time::{Duration, Instant}; use aisix_a2a::{A2aAuth, A2aBridge, A2aError, A2aUpstream, HttpBridge, DEFAULT_UPSTREAM_TIMEOUT}; use aisix_core::A2aProtocolVersion; @@ -308,3 +309,44 @@ async fn announces_the_pinned_wire_version_on_every_upstream_call() { "0.3" ); } + +/// An upstream that accepts the connection and never answers, so every +/// candidate URI burns its whole deadline. +async fn spawn_black_hole() -> SocketAddr { + let app = Router::new().fallback(any(|| async { + std::future::pending::<()>().await; + StatusCode::OK + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app.into_make_service()) + .await + .unwrap(); + }); + addr +} + +#[tokio::test] +async fn the_card_fetch_deadline_covers_the_whole_candidate_walk() { + // `timeout_ms` bounds the card fetch as ONE upstream operation. Walking + // several candidates must not hand each a fresh deadline, or a hung agent + // pins a gateway request for `candidates × timeout_ms`. + let addr = spawn_black_hole().await; + let bridge = HttpBridge::new(A2aUpstream { + timeout: Duration::from_millis(600), + ..upstream(format!("http://{addr}/a2a"), A2aAuth::None) + }); + + let started = Instant::now(); + let err = bridge.fetch_agent_card().await.unwrap_err(); + let elapsed = started.elapsed(); + + assert!(matches!(err, A2aError::Connect(_)), "got {err:?}"); + // Four candidates would be 2.4s if each restarted the clock. The bound is + // loose enough for a slow CI box while staying far below that. + assert!( + elapsed < Duration::from_millis(1500), + "card fetch took {elapsed:?}; the candidate walk must share ONE deadline" + ); +} diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs index 144cc821..26b5ae36 100644 --- a/crates/aisix-proxy/src/a2a.rs +++ b/crates/aisix-proxy/src/a2a.rs @@ -227,6 +227,7 @@ pub async fn a2a_agent_card( auth: AuthenticatedKey, AisixPath(agent): AisixPath, State(state): State, + uri: axum::http::Uri, headers: HeaderMap, ) -> Response { let snapshot = state.snapshot.load(); @@ -241,6 +242,21 @@ pub async fn a2a_agent_card( ) .into_response(); } + // Resolved BEFORE the upstream is contacted: without a public base there is + // no card this gateway can serve, and finding that out after the fetch only + // wastes an upstream round trip. + let Some(base) = gateway_base(&uri, &headers) else { + tracing::warn!( + agent = %agent, + "cannot derive the gateway's public base for an A2A agent card; refusing to serve one" + ); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "cannot determine the gateway's public address for this agent card", + ) + .into_response(); + }; + let upstream = upstream_from_a2a_agent(&entry.value); let bridge = HttpBridge::new(upstream); @@ -252,12 +268,8 @@ pub async fn a2a_agent_card( } }; // Rewrite the advertised service endpoints to the gateway so downstream - // callers route subsequent requests through `/a2a/`. Derived from - // the request's Host header (and forwarded scheme) since the gateway's - // public URL is not otherwise known here. - if let Some(base) = gateway_base(&headers) { - rewrite_card_urls(&mut card, &format!("{base}/a2a/{agent}")); - } + // callers route subsequent requests through `/a2a/`. + rewrite_card_urls(&mut card, &format!("{base}/a2a/{agent}")); axum::Json(card).into_response() } @@ -287,12 +299,24 @@ fn rewrite_card_urls(card: &mut aisix_a2a::AgentCard, gateway_url: &str) { } } -/// Reconstruct the gateway's public base (`scheme://host`) from request -/// headers: the `Host` header, and `X-Forwarded-Proto` when a proxy set it -/// (defaulting to `https`). Returns `None` when no Host header is present, in -/// which case the card's `url` is left as the upstream advertised it. -fn gateway_base(headers: &HeaderMap) -> Option { - let host = headers.get(header::HOST)?.to_str().ok()?; +/// Reconstruct the gateway's public base (`scheme://host`) for a request: the +/// authority, and `X-Forwarded-Proto` when a proxy set it (defaulting to +/// `https`). +/// +/// The authority comes from the `Host` header, falling back to the request +/// URI's own authority. HTTP/2 carries it there as the `:authority` +/// pseudo-header and sends no `Host` at all — and this listener negotiates h2 +/// whenever `proxy.tls` is set, so header-only lookup finds nothing on exactly +/// the deployments most likely to be in production. +/// +/// `None` means no card can be served: the caller must fail rather than hand +/// back a card still advertising the upstream's own address. +fn gateway_base(uri: &axum::http::Uri, headers: &HeaderMap) -> Option { + let host = headers + .get(header::HOST) + .and_then(|v| v.to_str().ok()) + .filter(|v| !v.is_empty()) + .or_else(|| uri.authority().map(|a| a.as_str()))?; let scheme = headers .get("x-forwarded-proto") .and_then(|v| v.to_str().ok()) @@ -377,22 +401,45 @@ mod tests { #[test] fn gateway_base_uses_forwarded_proto_then_defaults_https() { + let uri = axum::http::Uri::from_static("/a2a/x/.well-known/agent-card.json"); let mut headers = HeaderMap::new(); headers.insert(header::HOST, "gw.example.com".parse().unwrap()); assert_eq!( - gateway_base(&headers).as_deref(), + gateway_base(&uri, &headers).as_deref(), Some("https://gw.example.com") ); headers.insert("x-forwarded-proto", "http".parse().unwrap()); assert_eq!( - gateway_base(&headers).as_deref(), + gateway_base(&uri, &headers).as_deref(), Some("http://gw.example.com") ); } #[test] - fn gateway_base_is_none_without_host() { - assert_eq!(gateway_base(&HeaderMap::new()), None); + fn gateway_base_falls_back_to_the_uri_authority() { + // HTTP/2 sends no `Host` — the authority arrives as the `:authority` + // pseudo-header, which lands on the URI. This listener negotiates h2 + // whenever `proxy.tls` is set, so a header-only lookup finds nothing + // there and the card would otherwise be served still advertising the + // upstream's address. + let h2_uri = axum::http::Uri::from_static("https://gw.example.com/a2a/x"); + assert_eq!( + gateway_base(&h2_uri, &HeaderMap::new()).as_deref(), + Some("https://gw.example.com") + ); + // An empty Host does not shadow a usable authority either. + let mut empty_host = HeaderMap::new(); + empty_host.insert(header::HOST, "".parse().unwrap()); + assert_eq!( + gateway_base(&h2_uri, &empty_host).as_deref(), + Some("https://gw.example.com") + ); + } + + #[test] + fn gateway_base_is_none_without_any_authority() { + let origin_form = axum::http::Uri::from_static("/a2a/x/.well-known/agent-card.json"); + assert_eq!(gateway_base(&origin_form, &HeaderMap::new()), None); } #[test] diff --git a/tests/e2e/src/cases/a2a-gateway-e2e.test.ts b/tests/e2e/src/cases/a2a-gateway-e2e.test.ts index 56453334..3f6961d5 100644 --- a/tests/e2e/src/cases/a2a-gateway-e2e.test.ts +++ b/tests/e2e/src/cases/a2a-gateway-e2e.test.ts @@ -144,11 +144,15 @@ describe("a2a gateway e2e: /a2a/{agent}", () => { allowed_models: [], }); + // The gate proves only that the seeded rows reached the DP snapshot, and + // deliberately asserts none of the behaviour under test: a 404 would mean + // the agent row has not landed, a 401 that the key row has not. Anything + // else — including a 403 or a 502 — means both are present, and any defect + // in discovery, the version header, the card rewrite or the ACL then fails + // its own test by name instead of surfacing as a 60s propagation timeout. await waitConfigPropagation(async () => { - const card = await fetchCard("invoices", KEY_ALLOWED); - if (card.status !== 200) return false; - const denied = await fetchCard("invoices", KEY_NO_AGENTS); - return denied.status === 403; + const probe = await call("/a2a/invoices", KEY_ALLOWED, messageSend("gate")); + return probe.status !== 404 && probe.status !== 401; }); }, 60_000); From 9bc12e879bfda784b6dab6ecdb45fb70cf1f5753 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Mon, 10 Aug 2026 14:10:40 +0800 Subject: [PATCH 3/3] test(a2a): answer on the wire when the e2e stub agent faults `void handle(...)` discarded a promise that rejects on a malformed body or a write to a closed socket. Node terminates the test process on an unhandled rejection, and the request never got a response either way, so a stub fault surfaced as a gateway timeout rather than as itself. --- tests/e2e/src/harness/upstream-a2a.ts | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/tests/e2e/src/harness/upstream-a2a.ts b/tests/e2e/src/harness/upstream-a2a.ts index db89b707..4cf6704b 100644 --- a/tests/e2e/src/harness/upstream-a2a.ts +++ b/tests/e2e/src/harness/upstream-a2a.ts @@ -65,7 +65,21 @@ export async function startA2aUpstream( : "/.well-known/agent-card.json"; const httpServer: HttpServer = createServer((req, res) => { - void handle(req, res, { requests, servicePath, cardPath, token: options.token }); + handle(req, res, { + requests, + servicePath, + cardPath, + token: options.token, + }).catch((err: unknown) => { + // `handle` rejects on a malformed body, and on a write to an already + // closed socket. Unhandled, that terminates the test process; worse, the + // request never gets a response, so a stub fault reads as a gateway + // timeout instead of what it is. Answer on the wire either way. + if (!res.headersSent) { + res.writeHead(500, { "content-type": "application/json" }); + } + res.end(JSON.stringify({ error: `a2a stub failed: ${String(err)}` })); + }); }); await new Promise((resolve) => httpServer.listen(0, "127.0.0.1", resolve),