feat(a2a): add TOTP second-factor authentication for A2A server/client [WIP - untested]

[rjmendez]

Motivation

In agent mesh deployments (multiple A2A agents communicating peer-to-peer), a stolen bearer token alone should not be sufficient to impersonate an agent or issue dangerous commands. TOTP as a second factor ensures that even if a token is compromised, an attacker cannot authenticate without the shared TOTP seed.

What This Adds

TOTPServerAuthScheme (server-side validation)

• Wraps an existing ServerAuthScheme (e.g., SimpleTokenAuth) as a delegate for bearer token validation
• Requires an X-TOTP header with a valid 6-digit TOTP code on every request
• Per-peer seeds: Map bearer tokens → peer names → individual TOTP seeds ({"mrpink": "SEED1", "charlie": "SEED2"})
• Shared seed mode: Single seed for simpler deployments
• Validates via pyotp.TOTP(seed).verify(code, valid_window=1) — allows ±30s clock drift
• Fails closed: missing header = 401, wrong/expired code = 401

TOTPClientAuthScheme (client-side injection)

• Wraps an existing ClientAuthScheme (e.g., BearerTokenAuth) as a delegate
• Automatically generates and injects X-TOTP: <code> header on every outgoing request
• Reads seed from config or A2A_TOTP_SEED env var

Usage Example

from crewai.a2a.auth import TOTPServerAuthScheme, SimpleTokenAuth

# Server with per-peer TOTP
server_auth = TOTPServerAuthScheme(
    delegate=SimpleTokenAuth(token="bearer-token"),
    peer_seeds={"mrpink": "JBSWY3DPEHPK3PXP", "charlie": "KZQW6ZTBNFXHG33N"},
    token_to_peer={"bearer-token-mrpink": "mrpink", "bearer-token-charlie": "charlie"},
)

# Client with TOTP
from crewai.a2a.auth import TOTPClientAuthScheme, BearerTokenAuth

client_auth = TOTPClientAuthScheme(
    delegate=BearerTokenAuth(token="my-bearer-token"),
    seed="JBSWY3DPEHPK3PXP",
)

Design Notes

• valid_window=1 allows ±30 seconds of clock drift between agents — sufficient for well-configured systems while limiting replay attack windows
• Delegate pattern: wraps existing auth schemes rather than replacing them, so TOTP composes with any ServerAuthScheme/ClientAuthScheme
• authenticate_with_totp() is a separate method from authenticate() since the base ABC signature only passes the token — server middleware needs to extract and pass the X-TOTP header separately
• Seeds use SecretStr (via CoercedSecretStr) to prevent accidental logging

Changes

• src/crewai/a2a/auth/totp_scheme.py — new file with both schemes
• src/crewai/a2a/auth/__init__.py — exports updated
• pyproject.toml — added pyotp>=2.9.0 dependency
• tests/a2a/test_totp_auth.py — comprehensive tests

Tests Cover

• Valid token + valid TOTP → success
• Valid token + missing TOTP → 401
• Valid token + wrong TOTP → 401
• Valid token + expired TOTP (5 min old) → 401
• Invalid bearer token → 401
• Per-peer seed lookup (correct and incorrect peer)
• Client header injection with validation
• Client env var fallback
• Client missing seed → ValueError

[rjmendez]

Status: Work in Progress — untested

This PR is not yet ready for merge. Outstanding items before review:

• Server middleware integration (A2A request handler needs to extract X-TOTP header and call authenticate_with_totp())
• Agent card security_schemes auto-advertising for TOTP
• Tests verified against full dev environment with all dependencies

Marking as draft pending those fixes.

[rjmendez]

Update: Middleware Integration Complete

Added TOTPCallContextBuilder — the request-level integration point for TOTP validation in the A2A protocol.

What's new in this commit:

• TOTPCallContextBuilder: Implements CallContextBuilder from the A2A SDK. Extracts bearer token from Authorization header and TOTP code from X-TOTP header, validates both via TOTPServerAuthScheme, and returns an authenticated ServerCallContext.
• Async→sync boundary: The build() method is sync (per SDK interface) but auth is async. Handled via asyncio.run() when no event loop is running, with a thread-pool fallback for existing async contexts.
• _TOTPAuthenticatedUser: Implements the SDK's User ABC to wrap AuthenticatedUser for ServerCallContext.
• 4 new tests for the builder: valid request, missing auth header, missing TOTP header, invalid TOTP code. All 14 tests pass.
• Exports updated in __init__.py with graceful ImportError handling when a2a-sdk isn't installed.

The TOTP auth scheme is now fully integrated at both the authentication layer and the request middleware layer.