feat(a2a-bridge): wire auth callout client and outbound upstream forwarder (#380)

yordis · web-flow · commit e1ab68589ab0 · 2026-06-21T16:41:30.000-04:00
Signed-off-by: Yordis Prieto &lt;yordis.prieto@gmail.com&gt;
diff --git a/rsworkspace/Cargo.lock b/rsworkspace/Cargo.lock
diff --git a/rsworkspace/crates/a2a-bridge/Cargo.toml b/rsworkspace/crates/a2a-bridge/Cargo.toml
@@ -16,4 +16,14 @@ workspace = true
 [dependencies]
 a2a-auth-callout = { workspace = true }
 a2a-nats = { workspace = true }
+async-nats = { workspace = true }
+async-trait = { workspace = true }
+bytes = { workspace = true }
+reqwest = { workspace = true, features = ["json", "rustls-tls"] }
 serde_json = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "sync"] }
+
+[dev-dependencies]
+nkeys = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
diff --git a/rsworkspace/crates/a2a-bridge/src/auth.rs b/rsworkspace/crates/a2a-bridge/src/auth.rs
@@ -0,0 +1,160 @@
+mod callout_mint;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+
+use a2a_auth_callout::{BridgeMintRequest, BridgeMintResponse};
+
+pub use callout_mint::{BridgeTenantAccount, InProcessCalloutDispatcherMintWire};
+
+use crate::error::BridgeError;
+use crate::identity::{BridgeUserJwt, CallerHttpsAuth};
+
+#[async_trait]
+pub trait AuthCalloutClient: Send + Sync {
+    async fn mint(&self, caller_auth: &CallerHttpsAuth) -> Result<BridgeUserJwt, BridgeError>;
+}
+
+#[derive(Clone)]
+pub struct StubAuthCalloutClient;
+
+#[async_trait]
+impl AuthCalloutClient for StubAuthCalloutClient {
+    async fn mint(&self, _caller_auth: &CallerHttpsAuth) -> Result<BridgeUserJwt, BridgeError> {
+        Err(BridgeError::Mint(
+            "auth callout not configured for this deployment".into(),
+        ))
+    }
+}
+
+/// Test/dev mint that returns a fixed Account User JWT without a live auth-callout deployment.
+#[derive(Clone, Debug)]
+pub struct StubAuthCalloutMint {
+    jwt: BridgeUserJwt,
+}
+
+impl StubAuthCalloutMint {
+    pub fn new(jwt: impl Into<String>) -> Result<Self, BridgeError> {
+        Ok(Self {
+            jwt: BridgeUserJwt::new(jwt.into())?,
+        })
+    }
+
+    pub fn fixture() -> Result<Self, BridgeError> {
+        // A real compact JWT shape (header.payload.signature) so the
+        // dev/test stub satisfies the same gate production tokens go
+        // through — keeps the smuggling-prevention invariant intact
+        // even on fixture paths.
+        Self::new("eyJhbGciOiJub25lIn0.eyJzdWIiOiJoYXJuZXNzIn0.sig")
+    }
+}
+
+#[async_trait]
+impl AuthCalloutClient for StubAuthCalloutMint {
+    async fn mint(&self, _caller_auth: &CallerHttpsAuth) -> Result<BridgeUserJwt, BridgeError> {
+        Ok(self.jwt.clone())
+    }
+}
+
+#[async_trait]
+pub trait AuthMintWire: Send + Sync {
+    async fn roundtrip_message(&self, subject: String, payload: BytesPayload) -> Result<Vec<u8>, BridgeError>;
+}
+
+#[derive(Clone, Debug)]
+pub struct BytesPayload(pub Vec<u8>);
+
+#[derive(Clone)]
+pub struct AsyncNatsAuthMintWire {
+    client: Arc<async_nats::Client>,
+    request_timeout: Duration,
+}
+
+impl AsyncNatsAuthMintWire {
+    pub fn new(client: Arc<async_nats::Client>, request_timeout: Duration) -> Self {
+        Self {
+            client,
+            request_timeout,
+        }
+    }
+
+    #[must_use]
+    pub fn from_client(client: async_nats::Client, request_timeout: Duration) -> Self {
+        Self::new(Arc::new(client), request_timeout)
+    }
+}
+
+#[async_trait]
+impl AuthMintWire for AsyncNatsAuthMintWire {
+    async fn roundtrip_message(&self, subject: String, payload: BytesPayload) -> Result<Vec<u8>, BridgeError> {
+        tokio::time::timeout(self.request_timeout, self.client.request(subject, payload.0.into()))
+            .await
+            .map_err(|_| BridgeError::Mint("auth mint NATS roundtrip exceeded timeout".into()))?
+            .map_err(|e| BridgeError::Mint(e.to_string()))
+            .map(|m| m.payload.to_vec())
+    }
+}
+
+pub struct AuthCalloutJsonMintClient<W: AuthMintWire> {
+    wire: Arc<W>,
+    mint_subject: Arc<str>,
+    tenant_account: Option<BridgeTenantAccount>,
+}
+
+impl<W: AuthMintWire + 'static> AuthCalloutJsonMintClient<W> {
+    #[must_use]
+    pub fn new(wire: Arc<W>, mint_subject: impl Into<Arc<str>>) -> Self {
+        Self::with_tenant_account(wire, mint_subject, None)
+    }
+
+    #[must_use]
+    pub fn with_tenant_account(
+        wire: Arc<W>,
+        mint_subject: impl Into<Arc<str>>,
+        tenant_account: Option<BridgeTenantAccount>,
+    ) -> Self {
+        Self {
+            wire,
+            mint_subject: mint_subject.into(),
+            tenant_account,
+        }
+    }
+
+    #[must_use]
+    pub fn default_mint_subject() -> &'static str {
+        "a2a.bridge.auth.callout.request"
+    }
+}
+
+#[async_trait]
+impl<W: AuthMintWire + 'static> AuthCalloutClient for AuthCalloutJsonMintClient<W> {
+    async fn mint(&self, caller_auth: &CallerHttpsAuth) -> Result<BridgeUserJwt, BridgeError> {
+        let bearer_jwt = caller_auth
+            .as_str()
+            .strip_prefix("Bearer ")
+            .or_else(|| caller_auth.as_str().strip_prefix("bearer "))
+            .map(str::to_owned);
+        let envelope = BridgeMintRequest {
+            user_nkey: None,
+            user_jwt: bearer_jwt,
+            account: self.tenant_account.as_ref().map(|a| a.as_str().to_owned()),
+            client_info: None,
+            connect_opts: None,
+        };
+        let bytes = serde_json::to_vec(&envelope).map_err(|e: serde_json::Error| BridgeError::Serialize(e))?;
+        // The wire impl owns the deadline (e.g. AsyncNatsAuthMintWire's
+        // request_timeout) — wrapping a fixed 30s timeout here would
+        // mask longer wire configs and surface a different error
+        // message on short timeouts.
+        let reply = self
+            .wire
+            .roundtrip_message(self.mint_subject.to_string(), BytesPayload(bytes))
+            .await?;
+
+        let response: BridgeMintResponse =
+            serde_json::from_slice(&reply).map_err(|e: serde_json::Error| BridgeError::Deserialize(e))?;
+        BridgeUserJwt::new(response.user_jwt)
+    }
+}
diff --git a/rsworkspace/crates/a2a-bridge/src/auth/callout_mint.rs b/rsworkspace/crates/a2a-bridge/src/auth/callout_mint.rs
@@ -0,0 +1,160 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use a2a_auth_callout::dispatcher::AuthDispatcher;
+use a2a_auth_callout::wire::ServerAuthRequestClaims;
+use a2a_auth_callout::{BridgeAuthScheme, BridgeConnectOpts, BridgeMintRequest, BridgeMintResponse};
+use async_trait::async_trait;
+
+use super::{AuthMintWire, BytesPayload};
+use crate::error::BridgeError;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct BridgeTenantAccount(String);
+
+impl BridgeTenantAccount {
+    pub fn new(account: impl Into<String>) -> Result<Self, BridgeError> {
+        let raw = account.into();
+        let trimmed = raw.trim();
+        if trimmed.is_empty() {
+            return Err(BridgeError::Mint("bridge tenant account must be non-empty".into()));
+        }
+        Ok(Self(trimmed.to_owned()))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+/// In-process [`AuthMintWire`] that delegates to [`CalloutDispatcher`] (no live NATS).
+pub struct InProcessCalloutDispatcherMintWire {
+    dispatcher: Arc<dyn AuthDispatcher>,
+    tenant: BridgeTenantAccount,
+    mint_count: Arc<AtomicUsize>,
+}
+
+impl InProcessCalloutDispatcherMintWire {
+    pub fn new(dispatcher: Arc<dyn AuthDispatcher>, tenant: BridgeTenantAccount) -> Self {
+        Self {
+            dispatcher,
+            tenant,
+            mint_count: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+
+    #[must_use]
+    pub fn mint_count(&self) -> usize {
+        self.mint_count.load(Ordering::SeqCst)
+    }
+}
+
+#[async_trait]
+impl AuthMintWire for InProcessCalloutDispatcherMintWire {
+    async fn roundtrip_message(&self, _subject: String, payload: BytesPayload) -> Result<Vec<u8>, BridgeError> {
+        self.mint_count.fetch_add(1, Ordering::SeqCst);
+        let mut request: BridgeMintRequest =
+            serde_json::from_slice(&payload.0).map_err(|e: serde_json::Error| BridgeError::Deserialize(e))?;
+        if request.account.is_none() {
+            request.account = Some(self.tenant.as_str().to_owned());
+        }
+        if request.connect_opts.is_none() && request.user_jwt.is_some() {
+            request.connect_opts = Some(BridgeConnectOpts {
+                auth_scheme: Some(BridgeAuthScheme::Oidc),
+                api_key: None,
+            });
+        }
+        let claims =
+            ServerAuthRequestClaims::from_bridge_mint(request).map_err(|e| BridgeError::Mint(e.to_string()))?;
+        let user_jwt = self
+            .dispatcher
+            .dispatch(claims)
+            .await
+            .map_err(|e| BridgeError::Mint(e.to_string()))?;
+        let response = BridgeMintResponse {
+            user_jwt: user_jwt.as_str().to_owned(),
+        };
+        serde_json::to_vec(&response).map_err(|e: serde_json::Error| BridgeError::Serialize(e))
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn harness_callout_dispatcher(caller_id: &str) -> a2a_auth_callout::dispatcher::CalloutDispatcher {
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use a2a_auth_callout::StaticAccountResolver;
+    use a2a_auth_callout::credentials::oidc::{BearerToken, OidcVerifier};
+    use a2a_auth_callout::dispatcher::{CalloutDispatcher, CalloutDispatcherConfig};
+    use a2a_auth_callout::jwt::{AudienceAccount, CallerId, ExternalSubject, SpiceDbPrincipal, UserJwtClaims};
+    use a2a_auth_callout::permissions::IssuedPermissions;
+    use a2a_auth_callout::signing_key_source::{KeyVersion, SigningKeySource, StaticSigningKeySource};
+    use serde_json::json;
+
+    struct HarnessOidcVerifier {
+        caller_id: CallerId,
+    }
+
+    #[async_trait::async_trait]
+    impl OidcVerifier for HarnessOidcVerifier {
+        async fn verify(
+            &self,
+            _token: &BearerToken,
+            account: &AudienceAccount,
+        ) -> Result<UserJwtClaims, a2a_auth_callout::AuthCalloutError> {
+            Ok(UserJwtClaims {
+                kid: KeyVersion::new("pending").expect("fixture kid"),
+                sub: ExternalSubject::new("harness-sub").expect("fixture sub"),
+                aud: account.clone(),
+                data: SpiceDbPrincipal(json!({"spicedb_subject": "harness-sub"})),
+                nats_permissions: IssuedPermissions::default_for_caller(&self.caller_id),
+                caller_id: self.caller_id.clone(),
+            })
+        }
+    }
+
+    let caller = CallerId::new(caller_id).expect("harness caller_id");
+    let oidc: Arc<dyn OidcVerifier> = Arc::new(HarnessOidcVerifier { caller_id: caller });
+    let resolver: Arc<dyn a2a_auth_callout::AccountResolver> =
+        Arc::new(StaticAccountResolver::new(["tenant-harness".to_string()]));
+    let issuer = nkeys::KeyPair::new_account();
+    let issuer_seed = issuer.seed().expect("issuer seed");
+    let signing_key_source: Arc<dyn SigningKeySource> = Arc::new(
+        StaticSigningKeySource::new(&issuer_seed, KeyVersion::new("test").expect("harness version"))
+            .expect("harness signing source"),
+    );
+    CalloutDispatcher::new(CalloutDispatcherConfig {
+        signing_key_source,
+        user_jwt_ttl: Duration::from_secs(60),
+        account_resolver: resolver,
+        oidc: Some(oidc),
+        mtls: None,
+        api_key: None,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use a2a_auth_callout::caller_id_from_minted_jwt;
+
+    use super::*;
+    use crate::auth::{AuthCalloutClient, AuthCalloutJsonMintClient};
+    use crate::identity::CallerHttpsAuth;
+
+    #[tokio::test]
+    async fn harness_mint_wire_returns_deterministic_caller_id_jwt() {
+        let tenant = BridgeTenantAccount::new("tenant-harness").unwrap();
+        let dispatcher = Arc::new(harness_callout_dispatcher("bridge-harness-caller"));
+        let wire = Arc::new(InProcessCalloutDispatcherMintWire::new(dispatcher, tenant.clone()));
+        let client =
+            AuthCalloutJsonMintClient::with_tenant_account(wire, "a2a.bridge.auth.callout.request", Some(tenant));
+        let jwt = client
+            .mint(&CallerHttpsAuth::new("Bearer fixture-token"))
+            .await
+            .expect("harness mint");
+        let caller = caller_id_from_minted_jwt(jwt.as_str()).expect("caller_id claim");
+        assert_eq!(caller.as_str(), "bridge-harness-caller");
+    }
+}
diff --git a/rsworkspace/crates/a2a-bridge/src/lib.rs b/rsworkspace/crates/a2a-bridge/src/lib.rs
@@ -1,8 +1,15 @@
 #![doc = include_str!("../README.md")]
 #![cfg_attr(test, allow(clippy::expect_used, clippy::panic, clippy::unwrap_used))]
 
+pub mod auth;
 pub mod error;
 pub mod identity;
+pub mod outbound;
 
+pub use auth::{
+    AsyncNatsAuthMintWire, AuthCalloutClient, AuthCalloutJsonMintClient, BridgeTenantAccount,
+    InProcessCalloutDispatcherMintWire, StubAuthCalloutClient, StubAuthCalloutMint,
+};
 pub use error::BridgeError;
 pub use identity::{BridgeAgentId, BridgeUserJwt, CallerHttpsAuth, MintedCallerId};
+pub use outbound::{AgentRegistrationId, MethodSegment, OutboundHttpsAgentUpstream, StubOutboundForwarder, forward};
diff --git a/rsworkspace/crates/a2a-bridge/src/outbound.rs b/rsworkspace/crates/a2a-bridge/src/outbound.rs
