From f680c08cac0f55d651afbc9b94ab24d0c564f5c5 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Wed, 29 Jul 2026 09:32:17 -0400 Subject: [PATCH] fix(security): harden external sinks and callbacks --- docs/handler-authoring.md | 39 ++- examples/a2a_db_tasks.py | 147 +++++++---- examples/a2a_sqlalchemy_tasks.py | 122 ++++++--- src/adcp/audit_sink.py | 11 +- src/adcp/decisioning/property_list.py | 83 ++++-- src/adcp/server/a2a_push_security.py | 145 +++++++++++ src/adcp/server/a2a_server.py | 13 +- tests/test_a2a_push_security.py | 321 ++++++++++++++++++++++++ tests/test_a2a_server.py | 278 +++++++++++++++++++- tests/test_audit_sink.py | 35 ++- tests/test_decisioning_property_list.py | 56 +++-- 11 files changed, 1088 insertions(+), 162 deletions(-) create mode 100644 src/adcp/server/a2a_push_security.py create mode 100644 tests/test_a2a_push_security.py diff --git a/docs/handler-authoring.md b/docs/handler-authoring.md index e6d9730e3..d62315458 100644 --- a/docs/handler-authoring.md +++ b/docs/handler-authoring.md @@ -1076,25 +1076,40 @@ serve( transport="a2a", task_store=SqliteTaskStore("/var/lib/myagent/tasks.db"), push_config_store=SqlitePushNotificationConfigStore( - "/var/lib/myagent/push_configs.db" + "/var/lib/myagent/push_configs.db", + allowed_destination_hosts=None, # public-HTTPS mode ), ) ``` +Choose the destination policy explicitly: + +| Mode | Wiring | Behavior | +|---|---|---| +| Disabled | Omit `push_config_store` | Agent card does not advertise push support; registration is unsupported. | +| Public HTTPS | Pass a store with `allowed_destination_hosts=None` | Accept any HTTPS hostname that resolves only to public, non-reserved addresses. | +| Allowlist | Pass a non-empty `frozenset` | Apply the public HTTPS/SSRF checks, then require an exact canonical hostname match. | + +The reference examples expose the same modes through `A2A_PUSH_MODE` set to +`disabled` (the default), `public_https`, or `allowlist`. Allowlist mode also +requires `A2A_PUSH_ALLOWED_HOSTS=buyer.example,another.example`. + **Three things a durable push-notification config store MUST do — beyond the four from the TaskStore section above:** -1. **Validate the client-supplied `url` against an allowlist before - persisting.** a2a-sdk's push-notif sender POSTs full task JSON to +1. **Validate the client-supplied `url` before persisting.** a2a-sdk's + push-notif sender POSTs full task JSON to whatever URL is stored, with no built-in validation. An attacker registering `url=http://169.254.169.254/…` (cloud metadata) or `http://localhost:5432/` (internal services) gets SSRF + exfiltration in one call — the task JSON that lands on the attacker's server includes `history` and `artifacts`. The - reference impl does NOT validate URLs; the seller's store (or - a pre-persist hook) must. Reject non-https, reject RFC 1918 / - IPv6 link-local, and require the host match an egress allowlist - before `set_info` writes anything. + reference stores reject non-HTTPS destinations and DNS results in private, + reserved, metadata, or special-use ranges before `set_info` writes anything. + An exact hostname allowlist is an optional additional policy for closed + deployments; open buyer ecosystems normally use public-HTTPS mode. Repeat + the DNS/SSRF validation at delivery and pin the connection to the validated + address so DNS rebinding cannot bypass the registration-time decision. 2. **Treat `PushNotificationConfig.authentication.credentials` and `PushNotificationConfig.token` as secrets at rest.** Clients pass bearer tokens / shared secrets so the agent's callbacks can @@ -1105,11 +1120,11 @@ beyond the four from the TaskStore section above:** Production stores should envelope-encrypt those fields, or persist opaque references and keep the secrets in a dedicated backend (Vault, AWS KMS, GCP Secret Manager). -3. **Scope by principal, not just by tenant.** a2a-sdk's ABC doesn't - pass a `ServerCallContext` to push-config methods, so scoping has - to happen out-of-band. The reference `SqlitePushNotificationConfigStore` - reads a `ContextVar` your auth middleware populates and writes a - `scope` column on every row. Cross-scope isolation works; **within +3. **Scope by principal, not just by tenant.** Current a2a-sdk handler calls + pass `ServerCallContext` to push-config methods, and the reference store + derives its scope from the authenticated principal. A `ContextVar` remains + only as a compatibility fallback for context-free/background calls. The + store writes that scope on every row. Cross-scope isolation works; **within a scope, multiple principals can still overwrite each other's configs** (same `(scope, task_id)`, client omits `config_id`, PK collision). For multi-principal-per-tenant deployments, widen the diff --git a/examples/a2a_db_tasks.py b/examples/a2a_db_tasks.py index f0a035052..29e3cdb12 100644 --- a/examples/a2a_db_tasks.py +++ b/examples/a2a_db_tasks.py @@ -27,15 +27,18 @@ **Security model — push-notification config store adds two threats tenant-scoping alone does NOT address:** -1. **SSRF via unvalidated webhook URLs.** Clients supply +1. **SSRF via webhook URLs.** Clients supply ``PushNotificationConfig.url`` when subscribing to task progress; a2a-sdk's push-notif sender POSTs the full task JSON to that URL with no built-in validation. An attacker can register ``url=http://169.254.169.254/…`` (cloud metadata), ``http://localhost:5432/`` (internal services), link-local IPs, - etc. The store persists URLs verbatim — URL validation is the - seller's responsibility. Reject non-https, reject RFC 1918 / IPv6 - link-local, check against an egress allowlist before persisting. + etc. The store rejects URLs unless they use HTTPS and their canonical + hostname is publicly routable. Closed deployments can additionally require + it to appear in ``allowed_destination_hosts``. This storage-time gate does + not replace sender-side DNS resolution checks and IP-pinned connections on + every send; implement a custom ``PushNotificationSender`` for that + production boundary. 2. **Webhook secrets stored plaintext.** ``PushNotificationConfig.authentication.credentials`` and ``PushNotificationConfig.token`` are bearer tokens / shared @@ -72,8 +75,17 @@ Run:: - uv run python examples/a2a_db_tasks.py - # or: python -m adcp.examples.a2a_db_tasks + A2A_PUSH_MODE=public_https \ + uv run python examples/a2a_db_tasks.py + + A2A_PUSH_MODE=allowlist \ + A2A_PUSH_ALLOWED_HOSTS=callback.example \ + uv run python examples/a2a_db_tasks.py + +The default mode is ``disabled``: the example omits the push-config store and +does not advertise push support. ``public_https`` accepts any HTTPS callback +that passes DNS and reserved-range SSRF validation. ``allowlist`` adds an exact +hostname restriction using the comma-separated canonical hostnames. """ from __future__ import annotations @@ -83,7 +95,7 @@ import sqlite3 import uuid import warnings -from collections.abc import Callable +from collections.abc import AsyncIterator, Callable from contextlib import asynccontextmanager from contextvars import ContextVar from pathlib import Path @@ -106,6 +118,12 @@ from google.protobuf.json_format import MessageToJson, Parse from adcp.server import ADCPHandler, serve +from adcp.server.a2a_push_security import ( + normalize_allowed_push_hosts, + resolve_push_destination_settings, + scope_from_server_context, + validate_a2a_push_notification_url, +) from adcp.server.responses import capabilities_response, products_response _ANONYMOUS_SCOPE = "__anonymous__" @@ -115,6 +133,11 @@ is part of every WHERE clause.""" +def _scope_from_server_context(context: ServerCallContext | None) -> str: + """Derive a verified principal scope from an a2a-sdk call context.""" + return scope_from_server_context(context) or _ANONYMOUS_SCOPE + + # ---------------------------------------------------------------------- # SQLite-backed TaskStore # ---------------------------------------------------------------------- @@ -171,17 +194,10 @@ def _scope_from_context(self, context: ServerCallContext | None) -> str: key on every read/write; anything you don't include here *cannot* be enforced by the store. """ - user = getattr(context, "user", None) if context is not None else None - if user is None: - return _ANONYMOUS_SCOPE - user_name = getattr(user, "user_name", None) - is_authenticated = getattr(user, "is_authenticated", False) - if is_authenticated and isinstance(user_name, str) and user_name: - return user_name - return _ANONYMOUS_SCOPE + return _scope_from_server_context(context) @asynccontextmanager - async def _conn(self): + async def _conn(self) -> AsyncIterator[sqlite3.Connection]: # SQLite connections aren't safe across threads. Open a fresh # connection per operation and commit-on-success / rollback-on-error # so a port to psycopg / aiomysql doesn't silently leak partial @@ -278,22 +294,19 @@ async def list( # a single-user host but loses that guarantee across backups, # Docker bind mounts with wrong umask, and DB migrations. Either # encrypt those fields or move them to a secrets backend. -# 3. Isolate by principal, not just by scope. Within a single auth -# scope (e.g. "tenant-acme") multiple principals may share access -# to the same task. The reference impl keys on ``(scope, task_id, -# config_id)`` and falls ``config_id`` back to ``task_id`` when -# the client omits it — two principals registering without a -# ``config_id`` overwrite each other silently. Either require an -# explicit ``config_id`` from the client, or widen the scope key to -# include the principal. +# 3. Isolate by principal, not just by a coarse organization scope. The +# normal SDK path below uses the authenticated ``user_name`` directly. +# Adopters replacing it with a custom provider that groups principals +# must widen that key or authorize each config row explicitly. _current_push_config_scope: ContextVar[str | None] = ContextVar( "adcp_push_config_scope", default=None ) -"""Default ContextVar used by ``SqlitePushNotificationConfigStore`` when -no ``scope_provider`` is supplied. HTTP auth middleware sets it per -request; the store reads it on every op. Exposed at module level so -a seller with their own auth middleware can pair it with this -reference impl without subclassing.""" +"""Fallback ContextVar for context-less direct/background store calls. + +Normal a2a-sdk handler calls carry ``ServerCallContext`` and do not consult +this value. Exposed so a custom sender can restore the owning scope when it +later reads configs without a request context. +""" def _default_push_config_scope_provider() -> str | None: @@ -306,17 +319,13 @@ def _default_push_config_scope_provider() -> str | None: class SqlitePushNotificationConfigStore(PushNotificationConfigStore): """Durable A2A ``PushNotificationConfigStore`` backed by a single SQLite file, scoped by an authenticated principal resolved at - set/get/delete time via a ``scope_provider`` callable. - - a2a-sdk's ``PushNotificationConfigStore`` ABC does **not** pass a - ``ServerCallContext`` to ``set_info`` / ``get_info`` / - ``delete_info`` (unlike the ``TaskStore`` ABC), so scoping has to - happen out-of-band. The canonical pattern is a ``ContextVar`` the - seller's HTTP auth middleware populates per request — the - ``_default_push_config_scope_provider()`` factory below reads the - module-level ``_current_push_config_scope``. Sellers who already - maintain their own ContextVar (or prefer thread-locals, Starlette - ``request.state``, etc.) inject a custom provider. + set/get/delete time from the a2a-sdk ``ServerCallContext``. + + a2a-sdk 1.0 passes ``ServerCallContext`` to all three store methods; + normal handler calls therefore bind directly to the authenticated + ``user.user_name`` just like :class:`SqliteTaskStore`. A ContextVar + ``scope_provider`` remains as a fallback for background sender and direct + calls that genuinely lack a context. Example — wiring the default ContextVar from auth middleware:: @@ -353,7 +362,7 @@ async def dispatch(self, request, call_next): scope_provider=lambda: my_scope.get(default=None), ) - **Fails closed on anonymous requests.** If the provider returns + **Fails loudly on anonymous fallback.** If a context-less call's provider returns ``None``, a ``UserWarning`` is emitted once per store instance and the store falls through to ``__anonymous__`` — unauthenticated requests end up sharing one giant scope. Operators should reject @@ -361,9 +370,13 @@ async def dispatch(self, request, call_next): before the store is touched; the warning is the signal they forgot to. + ``allowed_destination_hosts=None`` accepts any public HTTPS destination + that passes the shared DNS/SSRF checks. Pass a concrete ``frozenset`` for + an additional exact-host allowlist; an explicitly empty set denies all. + **Background-task caveat — sender path.** a2a-sdk's push-notif - sender calls ``get_info()`` from a background ``asyncio.Task`` - spawned by ``DefaultRequestHandler``. That task inherits the + sender may call ``get_info()`` from a background ``asyncio.Task`` + without a ``ServerCallContext``. That task inherits the ContextVar snapshot captured at task-creation time; if the seller's auth middleware has already reset the ContextVar before the background task reads it, ``get_info()`` will return an empty @@ -381,9 +394,15 @@ def __init__( db_path: str | Path = "a2a_push_configs.db", *, scope_provider: Callable[[], str | None] | None = None, + allowed_destination_hosts: frozenset[str] | None = None, ) -> None: self._db_path = str(db_path) self._scope_provider = scope_provider or _default_push_config_scope_provider + self._allowed_destination_hosts = ( + normalize_allowed_push_hosts(allowed_destination_hosts) + if allowed_destination_hosts is not None + else None + ) self._init_schema() self._warned_anonymous = False @@ -409,8 +428,12 @@ def _init_schema(self) -> None: with contextlib.suppress(OSError): os.chmod(self._db_path, 0o600) - def _scope(self) -> str: - scope = self._scope_provider() + def _scope(self, context: ServerCallContext | None) -> str: + # An explicit context is authoritative. Never let an unauthenticated + # request inherit an ambient tenant from a ContextVar/provider. + scope = ( + scope_from_server_context(context) if context is not None else self._scope_provider() + ) if not scope: if not self._warned_anonymous: self._warned_anonymous = True @@ -429,7 +452,7 @@ def _scope(self) -> str: return scope @asynccontextmanager - async def _conn(self): + async def _conn(self) -> AsyncIterator[sqlite3.Connection]: conn = sqlite3.connect(self._db_path) try: yield conn @@ -447,7 +470,11 @@ async def set_info( notification_config: PushNotificationConfig, context: ServerCallContext | None = None, ) -> None: - scope = self._scope() + scope = self._scope(context) + validate_a2a_push_notification_url( + str(notification_config.url), + allowed_hosts=self._allowed_destination_hosts, + ) # PushNotificationConfig.id is optional on the wire; when the # client didn't supply one we synthesise a UUID so two clients # registering on the same task without explicit ids don't @@ -471,7 +498,7 @@ async def get_info( task_id: str, context: ServerCallContext | None = None, ) -> list[PushNotificationConfig]: - scope = self._scope() + scope = self._scope(context) async with self._conn() as conn: rows = conn.execute( "SELECT config_json FROM a2a_push_configs WHERE scope = ? AND task_id = ?", @@ -485,11 +512,11 @@ async def delete_info( context: ServerCallContext | None = None, config_id: str | None = None, ) -> None: - scope = self._scope() + scope = self._scope(context) async with self._conn() as conn: if config_id is None: - # a2a-sdk's ABC semantic: ``delete_info(task_id, None)`` - # removes every config for the task. Within a scope + # a2a-sdk's ABC semantic: ``config_id=None`` removes every + # config for the task. Within a scope # with multiple principals, this lets any principal # wipe every other principal's subscriptions — a # tenant-local DoS. Production stores that admit @@ -514,7 +541,7 @@ async def delete_info( # ---------------------------------------------------------------------- -class DemoAgent(ADCPHandler): +class DemoAgent(ADCPHandler[Any]): async def get_adcp_capabilities(self, params: Any, context: Any = None) -> dict[str, Any]: return capabilities_response(["media_buy"]) @@ -529,7 +556,21 @@ async def get_products(self, params: Any, context: Any = None) -> dict[str, Any] def main() -> None: task_store = SqliteTaskStore(db_path="a2a_tasks.db") - push_store = SqlitePushNotificationConfigStore(db_path="a2a_push_configs.db") + configured_push_hosts = frozenset( + host for host in os.environ.get("A2A_PUSH_ALLOWED_HOSTS", "").split(",") if host + ) + push_settings = resolve_push_destination_settings( + os.environ.get("A2A_PUSH_MODE", "disabled"), + configured_push_hosts, + ) + push_store = ( + SqlitePushNotificationConfigStore( + db_path="a2a_push_configs.db", + allowed_destination_hosts=push_settings.allowed_hosts, + ) + if push_settings.enabled + else None + ) serve( DemoAgent(), name="a2a-db-tasks-demo", diff --git a/examples/a2a_sqlalchemy_tasks.py b/examples/a2a_sqlalchemy_tasks.py index 3af2fa29c..9313c187a 100644 --- a/examples/a2a_sqlalchemy_tasks.py +++ b/examples/a2a_sqlalchemy_tasks.py @@ -21,10 +21,11 @@ etc.) and the wrapper just maps those that the protocol cares about. **Security model — same as the SQLite reference.** Tenant-scoped -lookups via ``ServerCallContext.user.user_name``; SSRF-vulnerable -``PushNotificationConfig.url`` MUST be validated by the seller before -persistence (this example does NOT validate — see the SQLite reference -docstring for the egress-allowlist pattern). Webhook secrets in +lookups via ``ServerCallContext.user.user_name``; push-notification URLs +must use HTTPS and match the store's explicit destination-host +allowlist before persistence. Allowlist only operator-owned, stable hosts; +storage-time validation does not replace sender-side DNS resolution checks +and IP-pinned connections on every delivery. Webhook secrets in ``authentication.credentials`` / ``token`` should be envelope-encrypted or moved to a secrets backend in production; this example persists them plaintext for runnability. @@ -44,17 +45,29 @@ Run:: - uv run python examples/a2a_sqlalchemy_tasks.py + A2A_PUSH_MODE=public_https \ + uv run python examples/a2a_sqlalchemy_tasks.py + + A2A_PUSH_MODE=allowlist \ + A2A_PUSH_ALLOWED_HOSTS=callback.example \ + uv run python examples/a2a_sqlalchemy_tasks.py + +The default mode is ``disabled`` and does not advertise push support. +``public_https`` accepts any HTTPS callback that passes DNS/reserved-range +validation; ``allowlist`` adds the exact hostname restriction. Then connect any A2A client to ``http://localhost:3001/`` — ``message/send`` carries a ``configuration.push_notification_config`` -that lands in the ``a2a_push_configs`` SQLite table; ``tasks/get`` -reads from ``a2a_tasks``. Tear down by deleting ``a2a_sqlalchemy.db``. +whose URL must pass the selected policy before it lands in the +``a2a_push_configs`` SQLite table; ``tasks/get`` reads from ``a2a_tasks``. +Tear down by deleting ``a2a_sqlalchemy.db``. """ from __future__ import annotations import contextlib +import os +import uuid import warnings from contextvars import ContextVar from datetime import datetime, timezone @@ -76,6 +89,13 @@ from a2a.types import TaskPushNotificationConfig as PushNotificationConfig from google.protobuf.json_format import MessageToJson, Parse +from adcp.server.a2a_push_security import ( + normalize_allowed_push_hosts, + resolve_push_destination_settings, + scope_from_server_context, + validate_a2a_push_notification_url, +) + try: from sqlalchemy import ( Boolean, @@ -184,18 +204,14 @@ def _scope_from_context(context: ServerCallContext | None) -> str: context never falls through to a "no filter" query that would leak other tenants' tasks. """ - if context is None or context.user is None: - return _NO_AUTH_SCOPE - return context.user.user_name or _NO_AUTH_SCOPE + return scope_from_server_context(context) or _NO_AUTH_SCOPE # Adopter-side push-notif scope hook. The -# ``PushNotificationConfigStore`` Protocol does NOT receive the -# ``ServerCallContext`` (a2a-sdk caveat — see the SQLite reference's -# docstring); adopters compose with their tenant-scoped ``TaskStore`` -# to derive scope by walking from ``task_id`` to the owning row. This -# example uses a contextvar that the surrounding handler populates -# from its own auth middleware. +# a2a-sdk 1.0 passes ``ServerCallContext`` to the push-config store. +# The ContextVar remains a compatibility fallback for direct calls and +# older surrounding middleware, while normal handler calls derive scope +# directly from the authenticated context. _push_config_scope: ContextVar[str | None] = ContextVar("_push_config_scope", default=None) @@ -294,20 +310,31 @@ async def list( class SqlAlchemyPushNotificationConfigStore(PushNotificationConfigStore): """Tenant-scoped, SQLAlchemy-backed push-notification config store. - URL validation is the seller's responsibility. This example does - NOT validate ``config.push_notification_config.url`` before - persisting — production deployments MUST reject non-https, - RFC-1918, link-local IPv6, and the cloud metadata service URL - before this method runs. See the SQLite reference's module - docstring for the SSRF threat model. + ``allowed_destination_hosts=None`` accepts any public HTTPS destination + that passes the shared DNS/SSRF checks. Pass a concrete ``frozenset`` for + an additional exact-host allowlist; an explicitly empty set denies all. """ - def __init__(self, session_factory: sessionmaker[Session]) -> None: + def __init__( + self, + session_factory: sessionmaker[Session], + *, + allowed_destination_hosts: frozenset[str] | None = None, + ) -> None: self._session_factory = session_factory + self._allowed_destination_hosts = ( + normalize_allowed_push_hosts(allowed_destination_hosts) + if allowed_destination_hosts is not None + else None + ) @staticmethod - def _scope() -> str: - scope = _push_config_scope.get() + def _scope(context: ServerCallContext | None) -> str: + # An explicit context is authoritative. Never let an unauthenticated + # request inherit an ambient tenant from the ContextVar. + scope = ( + scope_from_server_context(context) if context is not None else _push_config_scope.get() + ) if scope is None: warnings.warn( "PushNotificationConfigStore scope contextvar unset — " @@ -324,12 +351,14 @@ async def set_info( self, task_id: str, notification_config: PushNotificationConfig, + context: ServerCallContext | None = None, ) -> None: - scope = self._scope() - config_id = ( - notification_config.push_notification_config.id - or notification_config.push_notification_config.url + scope = self._scope(context) + validate_a2a_push_notification_url( + str(notification_config.url), + allowed_hosts=self._allowed_destination_hosts, ) + config_id = notification_config.id or f"auto-{uuid.uuid4()}" with self._session_factory() as session: row = A2APushConfigRow( scope=scope, @@ -342,8 +371,12 @@ async def set_info( session.merge(row) session.commit() - async def get_info(self, task_id: str) -> list[PushNotificationConfig]: - scope = self._scope() + async def get_info( + self, + task_id: str, + context: ServerCallContext | None = None, + ) -> list[PushNotificationConfig]: + scope = self._scope(context) with self._session_factory() as session: rows = session.execute( select(A2APushConfigRow).where( @@ -354,8 +387,13 @@ async def get_info(self, task_id: str) -> list[PushNotificationConfig]: ).scalars() return [Parse(row.payload, PushNotificationConfig()) for row in rows] - async def delete_info(self, task_id: str, config_id: str | None = None) -> None: - scope = self._scope() + async def delete_info( + self, + task_id: str, + context: ServerCallContext | None = None, + config_id: str | None = None, + ) -> None: + scope = self._scope(context) with self._session_factory() as session: stmt = delete(A2APushConfigRow).where( A2APushConfigRow.scope == scope, @@ -404,7 +442,7 @@ def build_engine_and_sessions( # ---------------------------------------------------------------------- -class DemoAgent(ADCPHandler): +class DemoAgent(ADCPHandler[Any]): async def get_adcp_capabilities(self, params: Any, context: Any = None) -> dict[str, Any]: return capabilities_response(["media_buy"]) @@ -420,7 +458,21 @@ async def get_products(self, params: Any, context: Any = None) -> dict[str, Any] def main() -> None: session_factory = build_engine_and_sessions() task_store = SqlAlchemyTaskStore(session_factory) - push_store = SqlAlchemyPushNotificationConfigStore(session_factory) + configured_push_hosts = frozenset( + host for host in os.environ.get("A2A_PUSH_ALLOWED_HOSTS", "").split(",") if host + ) + push_settings = resolve_push_destination_settings( + os.environ.get("A2A_PUSH_MODE", "disabled"), + configured_push_hosts, + ) + push_store = ( + SqlAlchemyPushNotificationConfigStore( + session_factory, + allowed_destination_hosts=push_settings.allowed_hosts, + ) + if push_settings.enabled + else None + ) serve( DemoAgent(), name="a2a-sqlalchemy-demo", diff --git a/src/adcp/audit_sink.py b/src/adcp/audit_sink.py index 14e639284..7cb306626 100644 --- a/src/adcp/audit_sink.py +++ b/src/adcp/audit_sink.py @@ -226,6 +226,10 @@ class SlackAlertSink: operation/identity/error summary. Prevents accidental egress of financial fields (budgets, credit limits), PII (contact info), or buyer-supplied free text. + :param include_error_message: Include raw exception text in Slack alerts. + Defaults to ``False`` because exception messages can contain request + values, upstream response fragments, or credentials. Enable only when + Slack is inside the same trusted logging boundary. :param timeout_seconds: Per-call HTTP timeout. The middleware also applies its own ``sink_timeout_seconds`` ceiling; the tighter of the two governs. @@ -243,6 +247,7 @@ def __init__( *, sensitive_operations: frozenset[str] | None = None, allowed_fields: frozenset[str] = frozenset(), + include_error_message: bool = False, timeout_seconds: float = 5.0, allow_private_destinations: bool = False, allowed_destination_ports: frozenset[int] | None = None, @@ -255,6 +260,7 @@ def __init__( self._webhook_url = webhook_url self._sensitive_operations = sensitive_operations self._allowed_fields = allowed_fields + self._include_error_message = include_error_message self._timeout = timeout_seconds self._allow_private = allow_private_destinations self._allowed_ports = allowed_destination_ports @@ -313,7 +319,7 @@ def _format(self, event: AuditEvent) -> str: parts.append(f"request_id={event.request_id}") if not event.success and event.error_type: parts.append(f"error={event.error_type}") - if event.error_message: + if self._include_error_message and event.error_message: parts.append(f"msg={event.error_message}") if self._allowed_fields: filtered = {k: v for k, v in event.details.items() if k in self._allowed_fields} @@ -329,6 +335,7 @@ def make_audit_middleware( sinks: Sequence[AuditSink], *, sink_timeout_seconds: float = 5.0, + include_error_message: bool = False, ) -> SkillMiddleware: """Compose one or more :class:`AuditSink` instances into a :data:`~adcp.server.SkillMiddleware`. @@ -396,7 +403,7 @@ async def audit_middleware( tenant_id=context.tenant_id, request_id=context.request_id, error_type=type(exc).__name__, - error_message=str(exc)[:200], + error_message=str(exc)[:200] if include_error_message else None, ), timeout_seconds=sink_timeout_seconds, ) diff --git a/src/adcp/decisioning/property_list.py b/src/adcp/decisioning/property_list.py index be696e0b8..b03ebeb9a 100644 --- a/src/adcp/decisioning/property_list.py +++ b/src/adcp/decisioning/property_list.py @@ -21,7 +21,9 @@ from __future__ import annotations import logging +import re from typing import Any, Protocol, runtime_checkable +from urllib.parse import urlsplit logger = logging.getLogger(__name__) @@ -35,12 +37,10 @@ class PropertyListFetcher(Protocol): reference. Adopters plug in their own HTTP client — the framework ships no hidden HTTP dependency. - Typical implementation:: + Implementations MUST pin delivery to the validated IP, disable redirects + and environment proxies, and URL-encode ``list_id``. A safe httpx pattern:: class MyFetcher: - def __init__(self, client: httpx.AsyncClient) -> None: - self._client = client - async def fetch( self, agent_url: str, @@ -48,13 +48,17 @@ async def fetch( *, auth_token: str | None = None, ) -> list[str]: + url = f"{agent_url.rstrip('/')}/property-lists/{quote(list_id, safe='')}" + transport = build_async_ip_pinned_transport(url) headers = {"Authorization": f"Bearer {auth_token}"} if auth_token else {} - resp = await self._client.get( - f"{agent_url}/property-lists/{list_id}", - headers=headers, - ) - resp.raise_for_status() - return resp.json()["property_ids"] + async with httpx.AsyncClient( + transport=transport, + follow_redirects=False, + trust_env=False, + ) as client: + resp = await client.get(url, headers=headers) + resp.raise_for_status() + return resp.json()["property_ids"] Wire the fetcher via:: @@ -95,33 +99,64 @@ async def resolve_property_list( ``auth_token`` is never included in the error details. """ from adcp.decisioning.types import AdcpError + from adcp.webhooks import ( + WebhookDestinationPolicy, + WebhookDestinationValidationError, + validate_webhook_destination_url, + ) list_id: str = ref.list_id agent_url: str = str(ref.agent_url) auth_token: str | None = getattr(ref, "auth_token", None) + if not re.fullmatch(r"[A-Za-z0-9._~-]+", list_id): + raise AdcpError( + "INVALID_REQUEST", + message="Property list_id must be a single URL-safe path segment", + recovery="correctable", + details={"list_id": list_id}, + ) + try: - ids = await fetcher.fetch(agent_url, list_id, auth_token=auth_token) + validation = validate_webhook_destination_url( + agent_url, + policy=WebhookDestinationPolicy.production(), + field="property_list.agent_url", + ) + except WebhookDestinationValidationError as exc: + raise AdcpError( + "INVALID_REQUEST", + message="Property list agent_url failed destination policy", + recovery="correctable", + details={"reason": exc.reason}, + ) from None + + parsed = urlsplit(validation.original_url) + safe_origin = f"{parsed.scheme}://{parsed.hostname or ''}" + if parsed.port is not None: + safe_origin += f":{parsed.port}" + + try: + ids = await fetcher.fetch(validation.original_url, list_id, auth_token=auth_token) return set(ids) except Exception as exc: - # Log the raw exception server-side; never include it in the wire - # error message — the exception repr may carry auth_token or other - # credential-shaped values from the upstream HTTP response. + # Exception text may carry auth_token or credential-shaped upstream + # values. Log only the class and deliberately omit the exception chain. logger.warning( - "[adcp.property_list] fetch failed for list_id=%r agent_url=%r: %s", + "[adcp.property_list] fetch failed for list_id=%r agent_origin=%r (%s)", list_id, - agent_url, - exc, + safe_origin, + type(exc).__name__, ) raise AdcpError( "SERVICE_UNAVAILABLE", message=( f"Property list fetch failed for list_id={list_id!r} " - f"from agent_url={agent_url!r}" + f"from agent_origin={safe_origin!r}" ), recovery="transient", - details={"list_id": list_id, "agent_url": agent_url}, - ) from exc + details={"list_id": list_id, "agent_origin": safe_origin}, + ) from None def filter_products_by_property_list( @@ -177,9 +212,7 @@ def _product_matches(product: Any, allowed: set[str]) -> bool: if st == "by_id": raw_ids: list[Any] = list(getattr(pp, "property_ids", None) or []) - product_ids = { - (pid.root if hasattr(pid, "root") else str(pid)) for pid in raw_ids - } + product_ids = {(pid.root if hasattr(pid, "root") else str(pid)) for pid in raw_ids} if permissive: if product_ids & allowed: logger.debug( @@ -262,9 +295,7 @@ async def maybe_apply_property_list_filter( products: list[Any] = list(getattr(response, "products", None) or []) filtered = filter_products_by_property_list(products, allowed) - return response.model_copy( - update={"products": filtered, "property_list_applied": True} - ) + return response.model_copy(update={"products": filtered, "property_list_applied": True}) def validate_property_list_config( diff --git a/src/adcp/server/a2a_push_security.py b/src/adcp/server/a2a_push_security.py new file mode 100644 index 000000000..6f727d70f --- /dev/null +++ b/src/adcp/server/a2a_push_security.py @@ -0,0 +1,145 @@ +"""Shared A2A push-notification destination and identity policy.""" + +from __future__ import annotations + +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Any, Literal + +from a2a.utils.errors import InvalidParamsError + +from adcp.signing._idna_canonicalize import canonicalize_host +from adcp.webhooks import WebhookDestinationPolicy, WebhookDestinationValidationError +from adcp.webhooks import validate_webhook_destination_url as validate_destination + +PushDestinationMode = Literal["disabled", "public_https", "allowlist"] + + +@dataclass(frozen=True) +class PushDestinationSettings: + """Resolved operator policy for accepting A2A push destinations.""" + + enabled: bool + allowed_hosts: frozenset[str] | None + + +def normalize_allowed_push_hosts(hosts: Iterable[str]) -> frozenset[str]: + """Canonicalize an operator-managed set of stable destination hosts.""" + try: + return frozenset(canonicalize_host(host.strip()) for host in hosts if host.strip()) + except (UnicodeError, ValueError) as exc: + raise WebhookDestinationValidationError( + "allowed push-notification hostname is invalid", + reason="invalid_allowed_hostname", + field="allowed_destination_hosts", + ) from exc + + +def resolve_push_destination_settings( + mode: str, + allowed_hosts: Iterable[str] = (), +) -> PushDestinationSettings: + """Resolve explicit disabled/public/allowlist operator configuration. + + ``public_https`` retains the shared DNS and reserved-range SSRF checks; it + disables only the additional hostname allowlist. ``disabled`` is expressed + by omitting the push-config store entirely so the agent card does not claim + push support. + """ + normalized_mode = mode.strip().lower() + normalized_hosts = normalize_allowed_push_hosts(allowed_hosts) + if normalized_mode == "disabled": + if normalized_hosts: + raise ValueError( + "A2A_PUSH_ALLOWED_HOSTS is set while A2A_PUSH_MODE=disabled; " + "choose allowlist or remove the hosts" + ) + return PushDestinationSettings(enabled=False, allowed_hosts=None) + if normalized_mode == "public_https": + if normalized_hosts: + raise ValueError( + "allowed push hosts require A2A_PUSH_MODE=allowlist; " + "public_https accepts any SSRF-safe public HTTPS destination" + ) + return PushDestinationSettings(enabled=True, allowed_hosts=None) + if normalized_mode == "allowlist": + if not normalized_hosts: + raise ValueError( + "A2A_PUSH_MODE=allowlist requires at least one allowed destination host" + ) + return PushDestinationSettings(enabled=True, allowed_hosts=normalized_hosts) + raise ValueError("A2A_PUSH_MODE must be one of: disabled, public_https, allowlist") + + +def scope_from_server_context(context: Any | None) -> str | None: + """Return a verified authenticated principal name, or ``None``.""" + user = getattr(context, "user", None) if context is not None else None + if user is None or not getattr(user, "is_authenticated", False): + return None + user_name = getattr(user, "user_name", None) + return user_name if isinstance(user_name, str) and user_name else None + + +def validate_push_notification_url( + url: str, + *, + allowed_hosts: frozenset[str] | None = None, + allowed_ports: frozenset[int] | None = None, +) -> None: + """Validate a callback through the SDK's shared SSRF classifier.""" + canonical_allowed_hosts = ( + normalize_allowed_push_hosts(allowed_hosts) if allowed_hosts is not None else None + ) + validation = validate_destination( + url, + policy=WebhookDestinationPolicy.production( + allowed_destination_ports=allowed_ports, + ), + field="push_notification_config.url", + ) + if canonical_allowed_hosts is not None and validation.hostname not in canonical_allowed_hosts: + raise WebhookDestinationValidationError( + "push notification URL hostname is not in allowed_destination_hosts", + reason="hostname_not_allowed", + field="push_notification_config.url", + url=url, + effective_url=validation.effective_url, + policy=validation.policy, + ) + + +def validate_a2a_push_notification_url( + url: str, + *, + allowed_hosts: frozenset[str] | None = None, + allowed_ports: frozenset[int] | None = None, +) -> None: + """Validate a callback and map policy failures to the A2A wire error.""" + try: + validate_push_notification_url( + url, + allowed_hosts=allowed_hosts, + allowed_ports=allowed_ports, + ) + except WebhookDestinationValidationError as exc: + # Do not reflect the rejected URL, its resolved address, or policy + # internals onto the public JSON-RPC surface. In particular, the + # shared SSRF classifier's diagnostic can contain a private IP. + data = {"code": exc.code, "reason": exc.reason} + if exc.field is not None: + data["field"] = exc.field + raise InvalidParamsError( + message="push notification destination failed validation", + data=data, + ) from None + + +__all__ = [ + "PushDestinationMode", + "PushDestinationSettings", + "normalize_allowed_push_hosts", + "resolve_push_destination_settings", + "scope_from_server_context", + "validate_a2a_push_notification_url", + "validate_push_notification_url", +] diff --git a/src/adcp/server/a2a_server.py b/src/adcp/server/a2a_server.py index 25194cfc0..8972de398 100644 --- a/src/adcp/server/a2a_server.py +++ b/src/adcp/server/a2a_server.py @@ -1019,13 +1019,12 @@ def create_a2a_server( ``examples/a2a_db_tasks.py`` for a reference SQLite-backed implementation that pairs with the ``SqliteTaskStore`` there. - Security note: unlike ``TaskStore``, a2a-sdk's - ``PushNotificationConfigStore`` ABC does not pass a - ``ServerCallContext`` to ``set_info`` / ``get_info`` / - ``delete_info``. Scoping by principal has to happen out-of-band - (via a ``ContextVar`` your auth middleware populates) or by - composition with a tenant-scoped ``TaskStore`` — the reference - impl shows the ContextVar pattern. + Security note: a2a-sdk 1.0 passes ``ServerCallContext`` to + ``set_info`` / ``get_info`` / ``delete_info``. Stores should + scope normal request-path access by the authenticated principal + in that context. A ``ContextVar`` is only needed as a fallback + for direct or background sender calls that lack a context; the + reference implementation demonstrates both paths. middleware: Optional sequence of :data:`~adcp.server.SkillMiddleware` callables wrapping every A2A skill dispatch. Composes outermost-first (first entry sees the call before later diff --git a/tests/test_a2a_push_security.py b/tests/test_a2a_push_security.py new file mode 100644 index 000000000..f7e93ab93 --- /dev/null +++ b/tests/test_a2a_push_security.py @@ -0,0 +1,321 @@ +"""A2A push destination policy and SQLAlchemy example parity tests.""" + +from __future__ import annotations + +import socket + +import pytest +from a2a.auth.user import UnauthenticatedUser, User +from a2a.server.context import ServerCallContext +from a2a.types import TaskPushNotificationConfig + +from adcp.server.a2a_push_security import ( + normalize_allowed_push_hosts, + resolve_push_destination_settings, + validate_push_notification_url, +) + + +class _AuthenticatedUser(User): + def __init__(self, user_name: str) -> None: + self._user_name = user_name + + @property + def is_authenticated(self) -> bool: + return True + + @property + def user_name(self) -> str: + return self._user_name + + +@pytest.fixture(autouse=True) +def _resolve_example_hosts(monkeypatch: pytest.MonkeyPatch): + original = socket.getaddrinfo + + def resolve(host: str, port: object, *args: object, **kwargs: object): + if host.rstrip(".").endswith(".example") or host.rstrip(".") == "example.com": + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] + return original(host, port, *args, **kwargs) + + monkeypatch.setattr(socket, "getaddrinfo", resolve) + + +def test_push_url_canonicalizes_idna_case_and_trailing_dot() -> None: + allowed = normalize_allowed_push_hosts(["BÜCHER.Example."]) + assert allowed == frozenset({"xn--bcher-kva.example"}) + validate_push_notification_url( + "https://xn--bcher-kva.EXAMPLE./callback", + allowed_hosts=frozenset({"BÜCHER.Example."}), + ) + + +@pytest.mark.parametrize( + ("url", "message"), + [ + ("https://user@example.com/hook", "userinfo"), + ("https://user:password@example.com/hook", "userinfo"), + ("https://example.com:not-a-port/hook", "Port could not be cast"), + ("http://example.com/hook", "https"), + ], +) +def test_push_url_rejects_unsafe_authority_forms(url: str, message: str) -> None: + with pytest.raises(ValueError, match=message): + validate_push_notification_url( + url, + allowed_hosts=normalize_allowed_push_hosts(["example.com"]), + ) + + +@pytest.mark.parametrize("host", ["127.0.0.1", "10.0.0.1", "::1", "fe80::1"]) +def test_push_url_rejects_private_ipv4_and_ipv6_literals(host: str) -> None: + rendered_host = f"[{host}]" if ":" in host else host + with pytest.raises(ValueError, match="blocked|private|SSRF"): + validate_push_notification_url( + f"https://{rendered_host}/hook", + allowed_hosts=normalize_allowed_push_hosts([host]), + ) + + +@pytest.mark.parametrize( + "host", + [ + "192.88.99.1", + "192.31.196.1", + "192.52.193.1", + "192.175.48.1", + "2001:20::1", + "64:ff9b::7f00:1", + ], +) +def test_push_url_rejects_globally_classified_reserved_literals(host: str) -> None: + """Shared SSRF policy covers ranges ``ipaddress.is_global`` misses.""" + rendered_host = f"[{host}]" if ":" in host else host + with pytest.raises(ValueError, match="blocked|reserved|SSRF"): + validate_push_notification_url( + f"https://{rendered_host}/hook", + allowed_hosts=normalize_allowed_push_hosts([host]), + ) + + +def test_push_url_allows_nonstandard_tls_port_by_default() -> None: + validate_push_notification_url( + "https://example.com:9443/hook", + allowed_hosts=frozenset({"example.com"}), + ) + + +def test_public_https_mode_accepts_unlisted_public_destination() -> None: + validate_push_notification_url("https://buyer-callback.example/hook") + + +def test_public_https_mode_still_rejects_private_destination() -> None: + with pytest.raises(ValueError, match="blocked|private|SSRF"): + validate_push_notification_url("https://127.0.0.1/hook") + + +def test_explicit_empty_allowlist_denies_every_destination() -> None: + with pytest.raises(ValueError, match="allowed_destination_hosts"): + validate_push_notification_url( + "https://buyer-callback.example/hook", + allowed_hosts=frozenset(), + ) + + +@pytest.mark.parametrize( + ("mode", "hosts", "enabled", "allowed"), + [ + ("disabled", (), False, None), + ("public_https", (), True, None), + ("allowlist", ("CALLBACK.Example.",), True, frozenset({"callback.example"})), + ], +) +def test_push_destination_modes( + mode: str, + hosts: tuple[str, ...], + enabled: bool, + allowed: frozenset[str] | None, +) -> None: + settings = resolve_push_destination_settings(mode, hosts) + assert settings.enabled is enabled + assert settings.allowed_hosts == allowed + + +@pytest.mark.parametrize( + ("mode", "hosts", "message"), + [ + ("allowlist", (), "requires at least one"), + ("public_https", ("callback.example",), "require A2A_PUSH_MODE=allowlist"), + ("disabled", ("callback.example",), "set while A2A_PUSH_MODE=disabled"), + ("anything", (), "must be one of"), + ], +) +def test_invalid_push_destination_mode_configuration( + mode: str, + hosts: tuple[str, ...], + message: str, +) -> None: + with pytest.raises(ValueError, match=message): + resolve_push_destination_settings(mode, hosts) + + +def test_sqlite_example_disabled_mode_omits_push_store( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import examples.a2a_db_tasks as example + + captured: dict[str, object] = {} + monkeypatch.delenv("A2A_PUSH_MODE", raising=False) + monkeypatch.delenv("A2A_PUSH_ALLOWED_HOSTS", raising=False) + monkeypatch.setattr(example, "SqliteTaskStore", lambda **_kwargs: object()) + monkeypatch.setattr( + example, + "SqlitePushNotificationConfigStore", + lambda **_kwargs: pytest.fail("disabled mode must not construct a push store"), + ) + monkeypatch.setattr( + example, + "serve", + lambda *_args, **kwargs: captured.update(kwargs), + ) + + example.main() + + assert captured["push_config_store"] is None + + +@pytest.mark.asyncio +async def test_sqlite_push_store_uses_a2a_context_for_tenant_isolation(tmp_path) -> None: + import examples.a2a_db_tasks as example + + store = example.SqlitePushNotificationConfigStore( + tmp_path / "push.db", + allowed_destination_hosts=frozenset({"callback.example"}), + ) + tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a")) + tenant_b = ServerCallContext(user=_AuthenticatedUser("tenant-b")) + config = TaskPushNotificationConfig( + id="cfg-1", + task_id="task-shared", + url="https://callback.example/hook", + ) + + await store.set_info("task-shared", config, tenant_a) + assert [item.id for item in await store.get_info("task-shared", tenant_a)] == ["cfg-1"] + assert await store.get_info("task-shared", tenant_b) == [] + + await store.delete_info("task-shared", tenant_b) + assert [item.id for item in await store.get_info("task-shared", tenant_a)] == ["cfg-1"] + + +@pytest.mark.asyncio +async def test_sqlite_explicit_unauthenticated_context_cannot_inherit_ambient_scope( + tmp_path, +) -> None: + import examples.a2a_db_tasks as example + + store = example.SqlitePushNotificationConfigStore( + tmp_path / "push.db", + scope_provider=lambda: "tenant-a", + allowed_destination_hosts=frozenset({"callback.example"}), + ) + tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a")) + unauthenticated = ServerCallContext(user=UnauthenticatedUser()) + config = TaskPushNotificationConfig( + id="cfg-1", + task_id="task-shared", + url="https://callback.example/hook", + ) + + await store.set_info("task-shared", config, tenant_a) + assert await store.get_info("task-shared", unauthenticated) == [] + + +@pytest.mark.asyncio +async def test_sqlalchemy_push_store_matches_a2a_v1_set_get_delete_contract() -> None: + import examples.a2a_sqlalchemy_tasks as example + + session_factory = example.build_engine_and_sessions(database_url="sqlite:///:memory:") + store = example.SqlAlchemyPushNotificationConfigStore( + session_factory, + allowed_destination_hosts=frozenset({"callback.example"}), + ) + tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a")) + tenant_b = ServerCallContext(user=_AuthenticatedUser("tenant-b")) + first = TaskPushNotificationConfig( + id="cfg-1", + task_id="task-1", + url="https://callback.example/first", + ) + second = TaskPushNotificationConfig( + id="cfg-2", + task_id="task-1", + url="https://callback.example/second", + ) + await store.set_info("task-1", first, tenant_a) + await store.set_info("task-1", second, tenant_a) + + stored = await store.get_info("task-1", tenant_a) + assert {config.id for config in stored} == {"cfg-1", "cfg-2"} + assert {config.url for config in stored} == { + "https://callback.example/first", + "https://callback.example/second", + } + assert await store.get_info("task-1", tenant_b) == [] + + await store.delete_info("task-1", tenant_b) + assert len(await store.get_info("task-1", tenant_a)) == 2 + + await store.delete_info("task-1", tenant_a, "cfg-1") + remaining = await store.get_info("task-1", tenant_a) + assert [config.id for config in remaining] == ["cfg-2"] + + await store.delete_info("task-1", tenant_a) + assert await store.get_info("task-1", tenant_a) == [] + + +@pytest.mark.asyncio +async def test_sqlalchemy_explicit_unauthenticated_context_cannot_inherit_ambient_scope() -> None: + import examples.a2a_sqlalchemy_tasks as example + + session_factory = example.build_engine_and_sessions(database_url="sqlite:///:memory:") + store = example.SqlAlchemyPushNotificationConfigStore( + session_factory, + allowed_destination_hosts=frozenset({"callback.example"}), + ) + tenant_a = ServerCallContext(user=_AuthenticatedUser("tenant-a")) + unauthenticated = ServerCallContext(user=UnauthenticatedUser()) + config = TaskPushNotificationConfig( + id="cfg-1", + task_id="task-shared", + url="https://callback.example/hook", + ) + token = example._push_config_scope.set("tenant-a") + try: + await store.set_info("task-shared", config, tenant_a) + assert [item.id for item in await store.get_info("task-shared", None)] == ["cfg-1"] + assert await store.get_info("task-shared", unauthenticated) == [] + finally: + example._push_config_scope.reset(token) + + +@pytest.mark.asyncio +async def test_sqlalchemy_push_store_defaults_to_public_https_destinations() -> None: + import examples.a2a_sqlalchemy_tasks as example + + session_factory = example.build_engine_and_sessions(database_url="sqlite:///:memory:") + store = example.SqlAlchemyPushNotificationConfigStore(session_factory) + scope_token = example._push_config_scope.set("tenant-a") + try: + await store.set_info( + "task-1", + TaskPushNotificationConfig( + task_id="task-1", + url="https://callback.example/hook", + ), + ServerCallContext(user=UnauthenticatedUser()), + ) + stored = await store.get_info("task-1", ServerCallContext(user=UnauthenticatedUser())) + assert [item.url for item in stored] == ["https://callback.example/hook"] + finally: + example._push_config_scope.reset(scope_token) diff --git a/tests/test_a2a_server.py b/tests/test_a2a_server.py index a2f7a4198..a7c5052aa 100644 --- a/tests/test_a2a_server.py +++ b/tests/test_a2a_server.py @@ -4,6 +4,7 @@ import contextlib import json +import socket import sys from typing import Any @@ -1106,7 +1107,10 @@ async def test_sqlite_push_config_store_isolates_scopes_by_contextvar(): with tempfile.TemporaryDirectory() as tmp: db = Path(tmp) / "push.db" - store = mod.SqlitePushNotificationConfigStore(db_path=db) + store = mod.SqlitePushNotificationConfigStore( + db_path=db, + allowed_destination_hosts=frozenset({"callback.tenant-a.example"}), + ) scope_var = mod._current_push_config_scope cfg = PushNotificationConfig(id="cfg-1", url="https://callback.tenant-a.example/webhook") @@ -1138,13 +1142,64 @@ async def test_sqlite_push_config_store_isolates_scopes_by_contextvar(): tok_a2 = scope_var.set("tenant-a") try: still_a = await store.get_info("task-shared") - assert len(still_a) == 1, ( - "SqlitePushNotificationConfigStore cross-scope delete " "removed tenant A's config." - ) + assert ( + len(still_a) == 1 + ), "SqlitePushNotificationConfigStore cross-scope delete removed tenant A's config." finally: scope_var.reset(tok_a2) +async def test_sqlite_push_config_store_rejects_untrusted_destinations(): + """Push callback URLs fail closed before attacker-controlled storage.""" + import importlib.util + import tempfile + from pathlib import Path + + from a2a.types import TaskPushNotificationConfig as PushNotificationConfig + from a2a.utils.errors import InvalidParamsError + + example_path = Path(__file__).parent.parent / "examples" / "a2a_db_tasks.py" + spec = importlib.util.spec_from_file_location("_a2a_db_tasks_ex_ssrf", example_path) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + with tempfile.TemporaryDirectory() as tmp: + store = mod.SqlitePushNotificationConfigStore( + db_path=Path(tmp) / "push.db", + scope_provider=lambda: "tenant-a", + allowed_destination_hosts=frozenset({"trusted.example"}), + ) + with pytest.raises(InvalidParamsError) as disallowed: + await store.set_info( + "task-1", + PushNotificationConfig(url="https://attacker.example/hook"), + ) + assert disallowed.value.message == "push notification destination failed validation" + assert disallowed.value.data == { + "code": "INVALID_REQUEST", + "reason": "hostname_not_allowed", + "field": "push_notification_config.url", + } + + private_store = mod.SqlitePushNotificationConfigStore( + db_path=Path(tmp) / "private.db", + scope_provider=lambda: "tenant-a", + allowed_destination_hosts=frozenset({"127.0.0.1"}), + ) + with pytest.raises(InvalidParamsError) as private: + await private_store.set_info( + "task-1", + PushNotificationConfig(url="https://127.0.0.1/hook"), + ) + assert private.value.message == "push notification destination failed validation" + assert private.value.data == { + "code": "INVALID_REQUEST", + "reason": "ssrf_rejected", + "field": "push_notification_config.url", + } + + @pytest.mark.skipif( sys.version_info < (3, 11), reason="a2a-sdk starlette integration requires Python 3.11+", @@ -1197,6 +1252,197 @@ async def test_custom_push_config_store_receives_sets_from_handler(): ) +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="a2a-sdk starlette integration requires Python 3.11+", +) +async def test_push_config_destination_policy_reaches_jsonrpc_error_envelope(tmp_path): + """Production A2A dispatch rejects unsafe callbacks as invalid params.""" + import httpx + + import examples.a2a_db_tasks as example + + task_store = _RecordingTaskStore() + await task_store.save( + Task(id="task-1", context_id="ctx-1", status=pb.TaskStatus(state="working")), + _empty_call_context(), + ) + push_store = example.SqlitePushNotificationConfigStore( + tmp_path / "push.db", + allowed_destination_hosts=frozenset({"127.0.0.1"}), + ) + app = create_a2a_server( + _TestHandler(), + name="push-policy-wire", + task_store=task_store, + push_config_store=push_store, + ) + + transport = httpx.ASGITransport(app=app, raise_app_exceptions=True) + async with httpx.AsyncClient( + transport=transport, + base_url="http://testserver", + ) as client: + response = await client.post( + "/", + headers={"A2A-Version": "1.0"}, + json={ + "jsonrpc": "2.0", + "id": "push-1", + "method": "CreateTaskPushNotificationConfig", + "params": { + "taskId": "task-1", + "url": "https://127.0.0.1/hook", + }, + }, + ) + + assert response.status_code == 200 + error = response.json()["error"] + assert error["code"] == -32602 + assert "push_notification_config.url" in str(error) + assert "127.0.0.1" not in str(error) + assert await push_store.get_info("task-1", _empty_call_context()) == [] + + +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="a2a-sdk starlette integration requires Python 3.11+", +) +async def test_push_config_wire_dispatch_isolates_principals_and_deletes( + tmp_path, monkeypatch: pytest.MonkeyPatch +): + """A2A 1.0 dispatch passes context and delete arguments in SDK order.""" + import httpx + from a2a.auth.user import User + from a2a.server.context import ServerCallContext + + import examples.a2a_db_tasks as example + + class _AuthenticatedUser(User): + def __init__(self, name: str) -> None: + self._name = name + + @property + def is_authenticated(self) -> bool: + return True + + @property + def user_name(self) -> str: + return self._name + + class _HeaderContextBuilder: + def build(self, request: Any) -> ServerCallContext: + return ServerCallContext( + user=_AuthenticatedUser(request.headers["x-test-principal"]), + state={"headers": dict(request.headers)}, + ) + + real_getaddrinfo = socket.getaddrinfo + + def public_callback_dns(host: str, *args: Any, **kwargs: Any) -> Any: + if host == "callback.example": + host = "93.184.216.34" + return real_getaddrinfo(host, *args, **kwargs) + + monkeypatch.setattr(socket, "getaddrinfo", public_callback_dns) + + task_store = _RecordingTaskStore() + await task_store.save( + Task(id="task-1", context_id="ctx-1", status=pb.TaskStatus(state="working")), + _empty_call_context(), + ) + push_store = example.SqlitePushNotificationConfigStore( + tmp_path / "push-isolation.db", + allowed_destination_hosts=frozenset({"callback.example"}), + ) + app = create_a2a_server( + _TestHandler(), + name="push-isolation-wire", + task_store=task_store, + push_config_store=push_store, + context_builder=_HeaderContextBuilder(), + ) + + request_id = 0 + + async def rpc(client: httpx.AsyncClient, method: str, params: dict, principal: str) -> dict: + nonlocal request_id + request_id += 1 + response = await client.post( + "/", + headers={"A2A-Version": "1.0", "x-test-principal": principal}, + json={ + "jsonrpc": "2.0", + "id": request_id, + "method": method, + "params": params, + }, + ) + assert response.status_code == 200 + return response.json() + + transport = httpx.ASGITransport(app=app, raise_app_exceptions=True) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + created = await rpc( + client, + "CreateTaskPushNotificationConfig", + { + "taskId": "task-1", + "id": "cfg-1", + "url": "https://callback.example/hook", + }, + "tenant-a", + ) + assert "result" in created, created + assert created["result"]["id"] == "cfg-1" + + tenant_b_list = await rpc( + client, + "ListTaskPushNotificationConfigs", + {"taskId": "task-1"}, + "tenant-b", + ) + assert "error" not in tenant_b_list + assert tenant_b_list["result"].get("configs", []) == [] + + # This exercises delete_info(task_id, context, config_id) through + # the real dispatcher. Tenant B cannot bind its context as config_id + # or remove Tenant A's row. + deleted_by_b = await rpc( + client, + "DeleteTaskPushNotificationConfig", + {"taskId": "task-1", "id": "cfg-1"}, + "tenant-b", + ) + assert "error" not in deleted_by_b + + tenant_a_get = await rpc( + client, + "GetTaskPushNotificationConfig", + {"taskId": "task-1", "id": "cfg-1"}, + "tenant-a", + ) + assert tenant_a_get["result"]["id"] == "cfg-1" + + deleted_by_a = await rpc( + client, + "DeleteTaskPushNotificationConfig", + {"taskId": "task-1", "id": "cfg-1"}, + "tenant-a", + ) + assert "error" not in deleted_by_a + + after_delete = await rpc( + client, + "ListTaskPushNotificationConfigs", + {"taskId": "task-1"}, + "tenant-a", + ) + assert "error" not in after_delete + assert after_delete["result"].get("configs", []) == [] + + async def test_sqlite_push_config_store_warns_once_on_anonymous_scope(): """Reference impl must fail LOUD when the scope_provider returns None — silent fall-through to the anonymous bucket is the @@ -1220,7 +1466,11 @@ async def test_sqlite_push_config_store_warns_once_on_anonymous_scope(): db = Path(tmp) / "anon.db" # Force the anonymous path by supplying a provider that always # returns None. - store = mod.SqlitePushNotificationConfigStore(db_path=db, scope_provider=lambda: None) + store = mod.SqlitePushNotificationConfigStore( + db_path=db, + scope_provider=lambda: None, + allowed_destination_hosts=frozenset({"x.example"}), + ) cfg = PushNotificationConfig(url="https://x.example/hook") with _warnings.catch_warnings(record=True) as caught: @@ -1255,7 +1505,11 @@ async def test_sqlite_push_config_store_synthesises_config_id_when_omitted(): with tempfile.TemporaryDirectory() as tmp: db = Path(tmp) / "uuid.db" - store = mod.SqlitePushNotificationConfigStore(db_path=db, scope_provider=lambda: "tenant-a") + store = mod.SqlitePushNotificationConfigStore( + db_path=db, + scope_provider=lambda: "tenant-a", + allowed_destination_hosts=frozenset({"first.example", "second.example"}), + ) await store.set_info( "shared-task", @@ -1686,3 +1940,15 @@ def composed(ctx: RequestContext) -> tuple[str | None, dict[str, Any]]: event = await queue.dequeue_event() assert isinstance(event, Task) assert event.status.state == pb.TaskState.TASK_STATE_COMPLETED + + +@pytest.fixture(autouse=True) +def _resolve_a2a_example_hosts(monkeypatch: pytest.MonkeyPatch): + original = socket.getaddrinfo + + def resolve(host: str, port: object, *args: object, **kwargs: object): + if host.rstrip(".").endswith(".example"): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] + return original(host, port, *args, **kwargs) + + monkeypatch.setattr(socket, "getaddrinfo", resolve) diff --git a/tests/test_audit_sink.py b/tests/test_audit_sink.py index 2cf5699e6..d7133473a 100644 --- a/tests/test_audit_sink.py +++ b/tests/test_audit_sink.py @@ -334,6 +334,35 @@ def _handler(request: httpx.Request) -> httpx.Response: assert "ops@buyer.example" not in text +@pytest.mark.asyncio +async def test_slack_alert_sink_omits_exception_message_by_default() -> None: + captured: dict[str, Any] = {} + + def _handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, text="ok") + + sink = SlackAlertSink("https://hooks.slack.com/services/T/B/x") + event = AuditEvent( + operation="create_media_buy", + success=False, + occurred_at=datetime.now(UTC), + error_type="RuntimeError", + error_message="Authorization=Bearer secret-token", + ) + + with patch( + "adcp.signing.ip_pinned_transport.build_async_ip_pinned_transport", + return_value=httpx.MockTransport(_handler), + ): + await sink.record(event) + + text = captured["body"]["text"] + assert "RuntimeError" in text + assert "secret-token" not in text + assert "Authorization" not in text + + @pytest.mark.asyncio async def test_slack_alert_sink_emits_only_allowlisted_details() -> None: captured: dict[str, Any] = {} @@ -398,7 +427,7 @@ def _handler(request: httpx.Request) -> httpx.Response: @pytest.mark.asyncio async def test_middleware_records_on_success() -> None: sink = _RecordingSink() - middleware = make_audit_middleware([sink]) + middleware = make_audit_middleware([sink], include_error_message=True) async def handler() -> dict[str, str]: return {"ok": "yes"} @@ -420,7 +449,7 @@ async def handler() -> dict[str, str]: @pytest.mark.asyncio async def test_middleware_records_failure_and_reraises() -> None: sink = _RecordingSink() - middleware = make_audit_middleware([sink]) + middleware = make_audit_middleware([sink], include_error_message=True) class _BoomError(RuntimeError): pass @@ -441,7 +470,7 @@ async def handler() -> Any: @pytest.mark.asyncio async def test_middleware_truncates_long_error_messages() -> None: sink = _RecordingSink() - middleware = make_audit_middleware([sink]) + middleware = make_audit_middleware([sink], include_error_message=True) async def handler() -> Any: raise RuntimeError("x" * 500) diff --git a/tests/test_decisioning_property_list.py b/tests/test_decisioning_property_list.py index 53a56cccb..34256df7f 100644 --- a/tests/test_decisioning_property_list.py +++ b/tests/test_decisioning_property_list.py @@ -2,6 +2,7 @@ from __future__ import annotations +import socket from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -17,6 +18,19 @@ from adcp.decisioning.types import AdcpError +@pytest.fixture(autouse=True) +def _resolve_example_hosts(monkeypatch: pytest.MonkeyPatch) -> None: + """Give documentation-only example hosts a public test address.""" + original = socket.getaddrinfo + + def fake_getaddrinfo(host: str, port: int, *args: Any, **kwargs: Any) -> Any: + if host.endswith(".example.com"): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port))] + return original(host, port, *args, **kwargs) + + monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo) + + # --------------------------------------------------------------------------- # Helpers — minimal wire-shape stubs # --------------------------------------------------------------------------- @@ -139,7 +153,8 @@ def test_by_id_strict_partial_match_excluded(self) -> None: property_targeting_allowed=False, ) result = filter_products_by_property_list( - [product], allowed_property_ids={"home"} # missing "sports" + [product], + allowed_property_ids={"home"}, # missing "sports" ) assert result == [] @@ -160,9 +175,7 @@ def test_by_id_permissive_any_intersection_sufficient(self) -> None: [_make_pp_by_id(["home", "sports"])], property_targeting_allowed=True, ) - result = filter_products_by_property_list( - [product], allowed_property_ids={"sports"} - ) + result = filter_products_by_property_list([product], allowed_property_ids={"sports"}) assert result == [product] def test_by_id_permissive_no_intersection_excluded(self) -> None: @@ -201,9 +214,7 @@ def test_mixed_by_id_and_by_tag_respects_by_id(self) -> None: [_make_pp_by_tag(["ctv"]), _make_pp_by_id(["home"])], property_targeting_allowed=True, ) - result = filter_products_by_property_list( - [product], allowed_property_ids={"home"} - ) + result = filter_products_by_property_list([product], allowed_property_ids={"home"}) assert result == [product] def test_multiple_products_filtered_correctly(self) -> None: @@ -252,9 +263,7 @@ def test_by_id_empty_property_ids_excluded_permissive(self) -> None: [_make_pp_by_id([])], property_targeting_allowed=True, ) - result = filter_products_by_property_list( - [product], allowed_property_ids={"home"} - ) + result = filter_products_by_property_list([product], allowed_property_ids={"home"}) assert result == [] def test_property_targeting_allowed_none_treated_as_false(self) -> None: @@ -279,9 +288,7 @@ class TestResolvePropertyList: async def test_returns_set_from_fetcher(self) -> None: fetcher = AsyncMock(spec=PropertyListFetcher) fetcher.fetch = AsyncMock(return_value=["home", "sports", "news"]) - ref = _make_property_list_ref( - agent_url="https://agent.example.com", list_id="list_1" - ) + ref = _make_property_list_ref(agent_url="https://agent.example.com", list_id="list_1") result = await resolve_property_list(ref, fetcher=fetcher) @@ -329,13 +336,28 @@ async def test_error_details_do_not_include_auth_token(self) -> None: await resolve_property_list(ref, fetcher=fetcher) err = exc_info.value - # details should include list_id and agent_url but NOT auth_token + # details include only a sanitized origin, never the full URL or token. assert err.details is not None assert "list_id" in err.details - assert "agent_url" in err.details + assert "agent_origin" in err.details assert "auth_token" not in err.details assert "secret_bearer_token" not in str(err.details) + @pytest.mark.asyncio + async def test_fetch_failure_log_omits_exception_text( + self, caplog: pytest.LogCaptureFixture + ) -> None: + secret = "secret_bearer_token" + fetcher = AsyncMock(spec=PropertyListFetcher) + fetcher.fetch = AsyncMock(side_effect=RuntimeError(f"Authorization=Bearer {secret}")) + ref = _make_property_list_ref(auth_token=secret) + + with caplog.at_level("WARNING"), pytest.raises(AdcpError): + await resolve_property_list(ref, fetcher=fetcher) + + assert "RuntimeError" in caplog.text + assert secret not in caplog.text + # --------------------------------------------------------------------------- # validate_property_list_config @@ -471,9 +493,7 @@ async def test_model_copy_used_not_in_place_mutation(self) -> None: """Response is updated via model_copy, not in-place mutation.""" p = _make_product("p1", [_make_pp_all()]) original_response = _make_response([p]) - original_response.model_copy = MagicMock( - side_effect=original_response.model_copy - ) + original_response.model_copy = MagicMock(side_effect=original_response.model_copy) params = MagicMock() params.property_list = _make_property_list_ref()