diff --git a/examples/v3_reference_seller/alembic/versions/0003_unique_buyer_api_key.py b/examples/v3_reference_seller/alembic/versions/0003_unique_buyer_api_key.py new file mode 100644 index 00000000..a02a161d --- /dev/null +++ b/examples/v3_reference_seller/alembic/versions/0003_unique_buyer_api_key.py @@ -0,0 +1,69 @@ +"""require globally unique buyer-agent bearer identifiers + +Revision ID: 0003 +Revises: 0002 +Create Date: 2026-07-29 + +``api_key_id`` is a bearer credential, so it must identify exactly one +commercial identity and tenant. The preflight deliberately stops the +migration before changing indexes when legacy duplicates exist; operators +must rotate or remove duplicates rather than letting the database choose an +arbitrary owner. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import context, op + +revision: str = "0003" +down_revision: str | None = "0002" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + if not context.is_offline_mode(): + connection = op.get_bind() + duplicate_count = connection.execute( + sa.text( + "SELECT COUNT(*) FROM (" + "SELECT api_key_id FROM buyer_agents " + "WHERE api_key_id IS NOT NULL " + "GROUP BY api_key_id HAVING COUNT(*) > 1" + ") AS duplicate_credentials" + ) + ).scalar_one() + if duplicate_count: + raise RuntimeError( + "Cannot enforce buyer_agents.api_key_id uniqueness: " + f"found {duplicate_count} duplicated credential identifier(s). " + "Rotate or remove duplicate bearer credentials, then rerun the migration." + ) + + # Create first so a concurrent/legacy duplicate makes the migration fail + # while the old lookup index remains available. PostgreSQL then rolls the + # transaction back; offline SQL retains the same safe ordering. + op.create_index( + "buyer_agents_api_key_uidx", + "buyer_agents", + ["api_key_id"], + unique=True, + postgresql_where=sa.text("api_key_id IS NOT NULL"), + sqlite_where=sa.text("api_key_id IS NOT NULL"), + ) + op.drop_index("buyer_agents_api_key_idx", table_name="buyer_agents") + + +def downgrade() -> None: + op.drop_index("buyer_agents_api_key_uidx", table_name="buyer_agents") + op.create_index( + "buyer_agents_api_key_idx", + "buyer_agents", + ["api_key_id"], + unique=False, + postgresql_where=sa.text("api_key_id IS NOT NULL"), + sqlite_where=sa.text("api_key_id IS NOT NULL"), + ) diff --git a/examples/v3_reference_seller/src/app.py b/examples/v3_reference_seller/src/app.py index 4b105119..2de9cca1 100644 --- a/examples/v3_reference_seller/src/app.py +++ b/examples/v3_reference_seller/src/app.py @@ -104,11 +104,18 @@ def _build_context_factory(): def build(meta: RequestMetadata) -> ToolContext: ctx = auth_context_factory(meta) - # Pin tenant from SubdomainTenantMiddleware. Subdomain wins for - # tenant routing; the validator's tenant_id is only the token's - # home tenant and may not match the host the request came in on. + # A bearer token is valid only on its home tenant's host. Letting + # the subdomain silently replace the authenticated tenant would + # allow the same credential value to be rebound to a buyer row in + # another tenant. tenant = current_tenant() if tenant is not None: + if ctx.tenant_id is not None and ctx.tenant_id != tenant.id: + raise AdcpError( + "PERMISSION_DENIED", + message="Bearer credential is not valid for this tenant.", + recovery="terminal", + ) ctx = replace(ctx, tenant_id=tenant.id) # Upgrade bearer-flow auth_info with a typed ApiKeyCredential @@ -151,6 +158,11 @@ async def _load_token_map(sessionmaker) -> dict[str, Principal]: select(BuyerAgentRow).where(BuyerAgentRow.api_key_id.is_not(None)) ) for row in result.scalars(): + if row.api_key_id in token_map: + raise RuntimeError( + "Duplicate buyer-agent api_key_id detected; bearer credentials " + "must identify exactly one tenant." + ) token_map[row.api_key_id] = Principal( caller_identity=row.agent_url, tenant_id=row.tenant_id, diff --git a/examples/v3_reference_seller/src/models.py b/examples/v3_reference_seller/src/models.py index 119e49c5..252641e7 100644 --- a/examples/v3_reference_seller/src/models.py +++ b/examples/v3_reference_seller/src/models.py @@ -194,9 +194,11 @@ class BuyerAgent(Base): UniqueConstraint("tenant_id", "agent_url", name="buyer_agents_tenant_agent_uk"), Index("buyer_agents_tenant_idx", "tenant_id"), Index( - "buyer_agents_api_key_idx", + "buyer_agents_api_key_uidx", "api_key_id", - postgresql_where=(api_key_id.is_not(None)), # type: ignore[has-type] + unique=True, + postgresql_where=(api_key_id.is_not(None)), + sqlite_where=(api_key_id.is_not(None)), ), ) diff --git a/examples/v3_reference_seller/src/platform.py b/examples/v3_reference_seller/src/platform.py index 7a81043c..ed885768 100644 --- a/examples/v3_reference_seller/src/platform.py +++ b/examples/v3_reference_seller/src/platform.py @@ -287,11 +287,38 @@ async def resolve( if ref is not None: account_id = ref.get("account_id") + principal: str | None = None + if auth_info is not None: + principal = getattr(auth_info, "principal", None) + if not principal: + raise AdcpError( + "AUTH_REQUIRED", + message="Account resolution requires an authenticated buyer-agent principal.", + recovery="terminal", + ) + async with sessionmaker() as session: + ba_result = await session.execute( + select(BuyerAgentRow).where( + BuyerAgentRow.tenant_id == tenant.id, + BuyerAgentRow.agent_url == principal, + ) + ) + buyer_agent = ba_result.scalar_one_or_none() + if buyer_agent is None: + raise AdcpError( + "ACCOUNT_NOT_FOUND", + message=( + f"No buyer agent matches principal {principal!r} " + f"under tenant {tenant.id!r}." + ), + recovery="terminal", + ) if account_id: result = await session.execute( select(AccountRow).where( AccountRow.tenant_id == tenant.id, + AccountRow.buyer_agent_id == buyer_agent.id, AccountRow.account_id == account_id, AccountRow.status == "active", ) @@ -313,39 +340,6 @@ async def resolve( # buyer agent. The buyer-agent's agent_url is the # `principal` field on auth_info — populated by the # framework's auth middleware from the validated bearer. - principal: str | None = None - if auth_info is not None: - principal = getattr(auth_info, "principal", None) - if not principal: - raise AdcpError( - "ACCOUNT_NOT_FOUND", - message=( - "Request did not include `account.account_id` " - "and no authenticated buyer-agent principal was " - "available to resolve a brand-shaped reference. " - "Send `account.account_id` explicitly, or " - "authenticate with a bearer token bound to a " - "seeded buyer agent." - ), - recovery="correctable", - field="account.account_id", - ) - ba_result = await session.execute( - select(BuyerAgentRow).where( - BuyerAgentRow.tenant_id == tenant.id, - BuyerAgentRow.agent_url == principal, - ) - ) - buyer_agent = ba_result.scalar_one_or_none() - if buyer_agent is None: - raise AdcpError( - "ACCOUNT_NOT_FOUND", - message=( - f"No buyer agent matches principal {principal!r} " - f"under tenant {tenant.id!r}." - ), - recovery="terminal", - ) acct_result = await session.execute( select(AccountRow) .where( @@ -447,6 +441,16 @@ async def upsert( session.add(new_row) action: str = "created" else: + if existing.buyer_agent_id != buyer_agent_row.id: + raise AdcpError( + "ACCOUNT_NOT_FOUND", + message=( + f"Account {natural_account_id!r} is not visible " + "to the authenticated buyer agent." + ), + recovery="terminal", + field="accounts", + ) existing.billing = billing_value existing.billing_entity = billing_entity_payload existing.sandbox = bool(incoming.sandbox) @@ -826,8 +830,8 @@ def __init__( # ``client_request_id``, so the seller has to track the mapping # to (a) echo the buyer's id in ``list_creatives`` and (b) # translate before calling ``attach_creative`` upstream. - self._creative_id_map: dict[str, str] = {} # buyer_id → upstream_id - self._creative_id_reverse: dict[str, str] = {} # upstream_id → buyer_id + self._creative_id_map: dict[tuple[str, str], str] = {} + self._creative_id_reverse: dict[tuple[str, str], str] = {} # AccountStore is always wired. ``app.main`` passes the # MOCK_AD_SERVER_URL env so resolved accounts route at the JS # mock-server fixture. Tests that bypass the AccountStore (by @@ -861,6 +865,35 @@ def _client(self, ctx: RequestContext) -> UpstreamHttpClient: treat_404_as_none=False, ) + @staticmethod + def _account_scope(ctx: RequestContext) -> str: + """Return the stable account boundary for seller-local state.""" + tenant_id = str(ctx.account.metadata.get("tenant_id") or "") + return f"{tenant_id}:{ctx.account.id}" + + async def _get_owned_order( + self, + ctx: RequestContext, + client: UpstreamHttpClient, + *, + network_code: str, + order_id: str, + ) -> dict[str, Any]: + """Fetch an order and fail closed unless it belongs to the account.""" + order = await upstream_helpers.get_order( + client, network_code=network_code, order_id=order_id + ) + expected_advertiser = ctx.account.metadata["advertiser_id"] + if order.get("advertiser_id") != expected_advertiser: + # Keep foreign and nonexistent ids indistinguishable. + raise AdcpError( + "MEDIA_BUY_NOT_FOUND", + message=f"Media buy {order_id!r} was not found.", + recovery="terminal", + field="media_buy_id", + ) + return order + def _record(self, method: str, args: dict[str, Any]) -> None: """Record an outbound upstream call on the wired :class:`MockAdServer`, if any. @@ -1074,6 +1107,9 @@ async def create_media_buy(self, req: CreateMediaBuyRequest, ctx: RequestContext ) order_id: str = order["order_id"] + self._buy_state.setdefault(order_id, {"packages": {}, "canceled": False})[ + "account_scope" + ] = self._account_scope(ctx) approval_task_id: str | None = order.get("approval_task_id") # Sync fast path — the upstream may auto-approve on creation # for non-guaranteed delivery (rare, but possible). @@ -1358,7 +1394,7 @@ async def update_media_buy( # Validate the media buy exists upstream. The SDK maps a 404 onto # ``MEDIA_BUY_NOT_FOUND`` automatically. - await upstream_helpers.get_order(client, network_code=network_code, order_id=media_buy_id) + await self._get_owned_order(ctx, client, network_code=network_code, order_id=media_buy_id) # Validate referenced packages exist on the order. The mock's # ``serializeOrder`` strips ``line_items`` from ``GET /orders/{id}`` @@ -1441,7 +1477,8 @@ async def update_media_buy( # before issuing attach_creative. Pass through unchanged # when no mapping is known (the upstream will surface # a 404 → CREATIVE_NOT_FOUND). - upstream_creative_id = self._creative_id_map.get(creative_id, creative_id) + creative_key = (self._account_scope(ctx), creative_id) + upstream_creative_id = self._creative_id_map.get(creative_key, creative_id) await upstream_helpers.attach_creative( client, network_code=network_code, @@ -1524,7 +1561,8 @@ async def sync_creatives( # the seller treats the second call as "creative already # known, just acknowledge". The buyer's intent for the new # placement flows through the ``assignments`` field below. - if creative.creative_id in self._creative_id_map: + creative_key = (self._account_scope(ctx), creative.creative_id) + if creative_key in self._creative_id_map: results.append( SyncCreativeResult.model_validate( { @@ -1555,8 +1593,10 @@ async def sync_creatives( ) upstream_id = str(upstream_resp.get("creative_id") or "") if upstream_id: - self._creative_id_map[creative.creative_id] = upstream_id - self._creative_id_reverse[upstream_id] = creative.creative_id + self._creative_id_map[creative_key] = upstream_id + self._creative_id_reverse[(self._account_scope(ctx), upstream_id)] = ( + creative.creative_id + ) results.append( SyncCreativeResult.model_validate( { @@ -1575,14 +1615,16 @@ async def sync_creatives( package_id = getattr(assignment, "package_id", None) if not buyer_creative_id or not package_id: continue - upstream_creative_id = self._creative_id_map.get(buyer_creative_id, buyer_creative_id) + creative_key = (self._account_scope(ctx), buyer_creative_id) + upstream_creative_id = self._creative_id_map.get(creative_key, buyer_creative_id) # Find the owning order via the shadow store: package_ids are # globally unique (upstream line_item ids). owning_order_id = next( ( oid for oid, state in self._buy_state.items() - if package_id in state.get("packages", {}) + if state.get("account_scope") == self._account_scope(ctx) + and package_id in state.get("packages", {}) ), None, ) @@ -1640,6 +1682,9 @@ async def get_media_buy_delivery( client = self._client(ctx) for order_id in media_buy_ids: try: + order_meta = await self._get_owned_order( + ctx, client, network_code=network_code, order_id=order_id + ) upstream_row = await upstream_helpers.get_delivery( client, network_code=network_code, order_id=order_id ) @@ -1654,19 +1699,7 @@ async def get_media_buy_delivery( # the order so we project the correct AdCP MediaBuyStatus # — completed / canceled / rejected buys would otherwise # all surface as 'active' to the buyer. - try: - order_meta = await upstream_helpers.get_order( - client, network_code=network_code, order_id=order_id - ) - upstream_status = order_meta.get("status", "") - except AdcpError as exc: - if exc.code == "MEDIA_BUY_NOT_FOUND": - # Delivery row exists but order is gone — odd, - # surface as 'active' so the row is at least - # well-formed; the operator's audit log will catch it. - upstream_status = "" - else: - raise + upstream_status = order_meta.get("status", "") wire_status = _DELIVERY_STATUS_MAP.get(upstream_status, "active") buy_state = self._buy_state.get(order_id, {}) if buy_state.get("canceled"): @@ -1876,6 +1909,9 @@ async def provide_performance_feedback( ], } client = self._client(ctx) + await self._get_owned_order( + ctx, client, network_code=network_code, order_id=req.media_buy_id + ) await upstream_helpers.post_conversions( client, network_code=network_code, @@ -1960,7 +1996,10 @@ async def list_creatives( filters = getattr(req, "filters", None) wanted_ids = list(getattr(filters, "creative_ids", None) or []) if filters else [] if wanted_ids: - upstream_wanted = {self._creative_id_map.get(cid, cid) for cid in wanted_ids} + account_scope = self._account_scope(ctx) + upstream_wanted = { + self._creative_id_map.get((account_scope, cid), cid) for cid in wanted_ids + } upstream_creatives = [ c for c in upstream_creatives if c.get("creative_id") in upstream_wanted ] @@ -1971,7 +2010,9 @@ async def list_creatives( # Surface the buyer's original creative_id when the seller # owns the mapping; falls back to the upstream id when the # creative was synced outside this seller instance. - "creative_id": self._creative_id_reverse.get(c["creative_id"], c["creative_id"]), + "creative_id": self._creative_id_reverse.get( + (self._account_scope(ctx), c["creative_id"]), c["creative_id"] + ), "name": c["name"], "format_id": {"agent_url": agent_url, "id": c.get("format_id", "")}, "status": _project_creative_status(c.get("status", "active")), diff --git a/examples/v3_reference_seller/tests/test_smoke.py b/examples/v3_reference_seller/tests/test_smoke.py index b6502247..f93ffc48 100644 --- a/examples/v3_reference_seller/tests/test_smoke.py +++ b/examples/v3_reference_seller/tests/test_smoke.py @@ -11,6 +11,7 @@ import sys from pathlib import Path +from unittest.mock import AsyncMock, MagicMock import pytest @@ -32,6 +33,11 @@ def test_models_import_and_declare_tables() -> None: for cls in (Tenant, BuyerAgent, Account): assert cls.__tablename__ in table_names + api_key_index = next( + index for index in BuyerAgent.__table__.indexes if index.name == "buyer_agents_api_key_uidx" + ) + assert api_key_index.unique is True + def test_platform_satisfies_decisioning_protocol() -> None: """The platform impl exists and can be inspected without an @@ -188,3 +194,55 @@ def test_platform_advertises_webhook_signing_when_alg_passed() -> None: assert ws.profile == "adcp/webhook-signing/v1" assert ws.algorithms is not None assert [a.value for a in ws.algorithms] == ["ed25519"] + + +def test_bearer_context_rejects_cross_tenant_rebinding( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The request host cannot replace the tenant authenticated by the token.""" + import src.app as app_module + + from adcp.decisioning import AdcpError + from adcp.server import ToolContext + + class _Tenant: + id = "tenant-b" + + monkeypatch.setattr( + app_module, + "auth_context_factory", + lambda _meta: ToolContext(tenant_id="tenant-a"), + ) + monkeypatch.setattr(app_module, "current_tenant", lambda: _Tenant()) + + with pytest.raises(AdcpError) as excinfo: + app_module._build_context_factory()(object()) + assert excinfo.value.code == "PERMISSION_DENIED" + + +@pytest.mark.asyncio +async def test_token_loader_rejects_duplicate_bearer_identifiers() -> None: + """A bearer value must resolve to exactly one buyer and tenant.""" + from src.app import _load_token_map + + rows = [ + MagicMock( + api_key_id="duplicate", + agent_url="https://a.example/", + tenant_id="tenant-a", + ), + MagicMock( + api_key_id="duplicate", + agent_url="https://b.example/", + tenant_id="tenant-b", + ), + ] + result = MagicMock() + result.scalars.return_value = rows + session = MagicMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=None) + session.execute = AsyncMock(return_value=result) + + with pytest.raises(RuntimeError, match="Duplicate buyer-agent api_key_id"): + await _load_token_map(MagicMock(return_value=session)) diff --git a/examples/v3_reference_seller/tests/test_smoke_broadening.py b/examples/v3_reference_seller/tests/test_smoke_broadening.py index b0f2105f..5a589bc7 100644 --- a/examples/v3_reference_seller/tests/test_smoke_broadening.py +++ b/examples/v3_reference_seller/tests/test_smoke_broadening.py @@ -342,6 +342,96 @@ class _Tenant: ), f"bank details leaked through to_wire_account projection: {wire}" +@pytest.mark.asyncio +async def test_account_store_explicit_id_is_bound_to_authenticated_buyer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An account id owned by another buyer is indistinguishable from missing.""" + import src.platform as platform_module + from src.platform import _make_account_store + + from adcp.decisioning import AdcpError, AuthInfo + + buyer_result = MagicMock() + buyer_result.scalar_one_or_none.return_value = MagicMock(id="ba_caller") + missing_account_result = MagicMock() + missing_account_result.scalar_one_or_none.return_value = None + session = MagicMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=None) + session.execute = AsyncMock(side_effect=[buyer_result, missing_account_result]) + + class _Tenant: + id = "t_acme" + + monkeypatch.setattr(platform_module, "current_tenant", lambda: _Tenant()) + store = _make_account_store(MagicMock(return_value=session), mock_upstream_url="http://up.test") + auth = AuthInfo(kind="anonymous", principal="https://caller.example/") + with pytest.raises(AdcpError) as excinfo: + await store.resolve({"account_id": "foreign-account"}, auth) + assert excinfo.value.code == "ACCOUNT_NOT_FOUND" + + +@pytest.mark.asyncio +async def test_account_store_upsert_cannot_overwrite_another_buyers_account( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import src.platform as platform_module + from src.models import Account as AccountRow + from src.platform import V3ReferenceSeller + + from adcp.decisioning import AdcpError, AuthInfo + from adcp.decisioning.accounts import ResolveContext + from adcp.types import SyncAccountsRequest + + buyer_result = MagicMock() + buyer_result.scalar_one_or_none.return_value = MagicMock(id="ba_caller") + foreign = AccountRow( + id="a_foreign", + tenant_id="t_acme", + buyer_agent_id="ba_other", + account_id="acme.example::operator.example", + name="Foreign", + status="active", + sandbox=False, + ) + existing_result = MagicMock() + existing_result.scalar_one_or_none.return_value = foreign + session = MagicMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=None) + session.begin = MagicMock(return_value=session) + session.execute = AsyncMock(side_effect=[buyer_result, existing_result]) + + class _Tenant: + id = "t_acme" + + monkeypatch.setattr(platform_module, "current_tenant", lambda: _Tenant()) + platform = V3ReferenceSeller( + sessionmaker=MagicMock(return_value=session), upstream_api_key="test-key" + ) + req = SyncAccountsRequest.model_validate( + { + "idempotency_key": "k_" + "a" * 18, + "accounts": [ + { + "brand": {"domain": "acme.example"}, + "operator": "operator.example", + "billing": "operator", + } + ], + } + ) + ctx = ResolveContext( + auth_info=AuthInfo(kind="anonymous", principal="https://caller.example/"), + tool_name="sync_accounts", + ) + with pytest.raises(AdcpError) as excinfo: + await platform.accounts.upsert(list(req.accounts), ctx) + assert excinfo.value.code == "ACCOUNT_NOT_FOUND" + assert foreign.buyer_agent_id == "ba_other" + + # --------------------------------------------------------------------------- # Translator-pattern HTTP plumbing — upstream is called via upstream_for() # --------------------------------------------------------------------------- @@ -898,7 +988,11 @@ async def test_update_media_buy_cancel_marks_local_state(respx_mock: Any) -> Non respx_mock.get("/v1/orders/ord_test").mock( return_value=httpx.Response( 200, - json={"order_id": "ord_test", "status": "active"}, + json={ + "order_id": "ord_test", + "status": "active", + "advertiser_id": "adv_volta_motors", + }, ) ) respx_mock.get("/v1/orders/ord_test/lineitems").mock( @@ -979,7 +1073,11 @@ async def test_update_media_buy_unknown_package_id_raises_not_found( respx_mock.get("/v1/orders/ord_test").mock( return_value=httpx.Response( 200, - json={"order_id": "ord_test", "status": "active"}, + json={ + "order_id": "ord_test", + "status": "active", + "advertiser_id": "adv_volta_motors", + }, ) ) respx_mock.get("/v1/orders/ord_test/lineitems").mock( @@ -1016,7 +1114,11 @@ async def test_update_media_buy_affected_packages_echo_list_agent_urls( respx_mock.get("/v1/orders/ord_test").mock( return_value=httpx.Response( 200, - json={"order_id": "ord_test", "status": "delivering"}, + json={ + "order_id": "ord_test", + "status": "delivering", + "advertiser_id": "adv_volta_motors", + }, ) ) respx_mock.get("/v1/orders/ord_test/lineitems").mock( @@ -1284,6 +1386,12 @@ async def test_provide_performance_feedback_posts_capi_conversion( json={"order_id": "ord_1", "events_received": 1, "events_deduplicated": 0}, ) ) + respx_mock.get("/v1/orders/ord_1").mock( + return_value=httpx.Response( + 200, + json={"order_id": "ord_1", "advertiser_id": "adv_volta_motors"}, + ) + ) platform = _platform_with_upstream() ctx = _build_ctx() req = ProvidePerformanceFeedbackRequest.model_validate( @@ -1350,7 +1458,7 @@ async def test_provide_performance_feedback_404_translates_to_media_buy_not_foun from adcp.decisioning import AdcpError from adcp.types import ProvidePerformanceFeedbackRequest - respx_mock.post("/v1/orders/ord_missing/conversions").mock( + respx_mock.get("/v1/orders/ord_missing").mock( return_value=httpx.Response(404, json={"code": "ORDER_NOT_FOUND", "message": "missing"}) ) platform = _platform_with_upstream() @@ -1426,6 +1534,124 @@ async def test_list_creative_formats_is_static_no_upstream_call() -> None: assert respx_mock.calls.call_count == 0 +@pytest.mark.asyncio +@respx.mock(base_url=_RESPX_BASE_URL) +async def test_update_media_buy_rejects_foreign_advertiser_order(respx_mock: Any) -> None: + from adcp.decisioning import AdcpError + from adcp.types import UpdateMediaBuyRequest + + respx_mock.get("/v1/orders/ord_foreign").mock( + return_value=httpx.Response( + 200, + json={"order_id": "ord_foreign", "advertiser_id": "adv_other"}, + ) + ) + platform = _platform_with_upstream() + req = UpdateMediaBuyRequest.model_validate( + { + "account": {"account_id": "signed-buyer-main"}, + "media_buy_id": "ord_foreign", + "idempotency_key": "k_" + "u" * 18, + "paused": True, + } + ) + with pytest.raises(AdcpError) as excinfo: + await platform.update_media_buy("ord_foreign", req, _build_ctx()) + assert excinfo.value.code == "MEDIA_BUY_NOT_FOUND" + assert not any(call.request.url.path.endswith("/lineitems") for call in respx_mock.calls) + + +@pytest.mark.asyncio +@respx.mock(base_url=_RESPX_BASE_URL) +async def test_delivery_omits_foreign_advertiser_order(respx_mock: Any) -> None: + from adcp.types import GetMediaBuyDeliveryRequest + + respx_mock.get("/v1/orders/ord_foreign").mock( + return_value=httpx.Response( + 200, + json={"order_id": "ord_foreign", "advertiser_id": "adv_other"}, + ) + ) + platform = _platform_with_upstream() + req = GetMediaBuyDeliveryRequest.model_validate({"media_buy_ids": ["ord_foreign"]}) + response = await platform.get_media_buy_delivery(req, _build_ctx()) + assert response.media_buy_deliveries == [] + assert not any(call.request.url.path.endswith("/delivery") for call in respx_mock.calls) + + +@pytest.mark.asyncio +@respx.mock(base_url=_RESPX_BASE_URL) +async def test_performance_feedback_rejects_foreign_advertiser_order( + respx_mock: Any, +) -> None: + from adcp.decisioning import AdcpError + from adcp.types import ProvidePerformanceFeedbackRequest + + respx_mock.get("/v1/orders/ord_foreign").mock( + return_value=httpx.Response( + 200, + json={"order_id": "ord_foreign", "advertiser_id": "adv_other"}, + ) + ) + platform = _platform_with_upstream() + req = ProvidePerformanceFeedbackRequest.model_validate( + { + "idempotency_key": "k_" + "f" * 18, + "media_buy_id": "ord_foreign", + "metric_type": "conversion_rate", + "performance_index": 1.0, + "measurement_period": { + "start": "2026-04-01T00:00:00Z", + "end": "2026-04-30T23:59:59Z", + }, + } + ) + with pytest.raises(AdcpError) as excinfo: + await platform.provide_performance_feedback(req, _build_ctx()) + assert excinfo.value.code == "MEDIA_BUY_NOT_FOUND" + assert not any(call.request.method == "POST" for call in respx_mock.calls) + + +@pytest.mark.asyncio +@respx.mock(base_url=_RESPX_BASE_URL) +async def test_creative_id_mapping_is_scoped_to_account(respx_mock: Any) -> None: + from adcp.types import SyncCreativesRequest + + uploads = iter(["up_account_a", "up_account_b"]) + + def upload(_: httpx.Request) -> httpx.Response: + return httpx.Response(201, json={"creative_id": next(uploads)}) + + route = respx_mock.post("/v1/creatives").mock(side_effect=upload) + platform = _platform_with_upstream() + req = SyncCreativesRequest.model_validate( + { + "account": {"account_id": "account"}, + "idempotency_key": "k_" + "c" * 18, + "creatives": [ + { + "creative_id": "shared-id", + "name": "Creative", + "format_id": { + "agent_url": "https://reference.adcp.org", + "id": "display_300x250", + }, + "assets": {}, + } + ], + } + ) + ctx_a = _build_ctx() + ctx_b = _build_ctx() + ctx_b.account.id = "a_acme_2" + ctx_b.account.metadata["account_id"] = "other-account" + ctx_b.account.metadata["advertiser_id"] = "adv_other" + + await platform.sync_creatives(req, ctx_a) + await platform.sync_creatives(req, ctx_b) + assert route.call_count == 2 + + @pytest.mark.asyncio async def test_account_loader_rejects_account_missing_upstream_routing( monkeypatch: pytest.MonkeyPatch, @@ -1439,7 +1665,7 @@ async def test_account_loader_rejects_account_missing_upstream_routing( from src.models import Account as AccountRow from src.platform import _make_account_store - from adcp.decisioning import AdcpError + from adcp.decisioning import AdcpError, AuthInfo bad_row = AccountRow( id="a_bad", @@ -1452,12 +1678,14 @@ async def test_account_loader_rejects_account_missing_upstream_routing( sandbox=False, ext=None, ) + buyer_result = MagicMock() + buyer_result.scalar_one_or_none = MagicMock(return_value=MagicMock(id="ba_x")) result = MagicMock() result.scalar_one_or_none = MagicMock(return_value=bad_row) session = MagicMock() session.__aenter__ = AsyncMock(return_value=session) session.__aexit__ = AsyncMock(return_value=None) - session.execute = AsyncMock(return_value=result) + session.execute = AsyncMock(side_effect=[buyer_result, result]) sessionmaker = MagicMock(return_value=session) class _Tenant: @@ -1467,7 +1695,10 @@ class _Tenant: store = _make_account_store(sessionmaker, mock_upstream_url="http://up.test") with pytest.raises(AdcpError) as excinfo: - await store.resolve({"account_id": "bad-acct"}) + await store.resolve( + {"account_id": "bad-acct"}, + AuthInfo(kind="anonymous", principal="https://signed-buyer.example/"), + ) assert excinfo.value.code == "SERVICE_UNAVAILABLE" assert excinfo.value.recovery == "transient" @@ -1485,6 +1716,8 @@ async def test_account_loader_returns_mock_mode_with_upstream_url( from src.models import Account as AccountRow from src.platform import _make_account_store + from adcp.decisioning import AuthInfo + good_row = AccountRow( id="a_good", tenant_id="t_acme", @@ -1496,12 +1729,14 @@ async def test_account_loader_returns_mock_mode_with_upstream_url( sandbox=False, ext={"network_code": "net_premium_us", "advertiser_id": "adv_volta_motors"}, ) + buyer_result = MagicMock() + buyer_result.scalar_one_or_none = MagicMock(return_value=MagicMock(id="ba_x")) result = MagicMock() result.scalar_one_or_none = MagicMock(return_value=good_row) session = MagicMock() session.__aenter__ = AsyncMock(return_value=session) session.__aexit__ = AsyncMock(return_value=None) - session.execute = AsyncMock(return_value=result) + session.execute = AsyncMock(side_effect=[buyer_result, result]) sessionmaker = MagicMock(return_value=session) class _Tenant: @@ -1510,7 +1745,10 @@ class _Tenant: monkeypatch.setattr(platform_module, "current_tenant", lambda: _Tenant()) store = _make_account_store(sessionmaker, mock_upstream_url="http://127.0.0.1:4503") - account = await store.resolve({"account_id": "good-acct"}) + account = await store.resolve( + {"account_id": "good-acct"}, + AuthInfo(kind="anonymous", principal="https://signed-buyer.example/"), + ) assert account.mode == "mock" assert account.metadata["mock_upstream_url"] == "http://127.0.0.1:4503" # Routing data still flows through metadata so platform methods diff --git a/examples/v3_reference_seller/tests/test_unique_api_key_migration.py b/examples/v3_reference_seller/tests/test_unique_api_key_migration.py new file mode 100644 index 00000000..142f857e --- /dev/null +++ b/examples/v3_reference_seller/tests/test_unique_api_key_migration.py @@ -0,0 +1,71 @@ +"""Regression tests for the bearer-credential uniqueness migration.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +_MIGRATION = ( + Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0003_unique_buyer_api_key.py" +) + + +def _load_migration(): + spec = importlib.util.spec_from_file_location("unique_buyer_api_key_migration", _MIGRATION) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_upgrade_rejects_legacy_duplicates_before_index_changes() -> None: + migration = _load_migration() + result = MagicMock() + result.scalar_one.return_value = 2 + connection = MagicMock() + connection.execute.return_value = result + + with ( + patch.object(migration.context, "is_offline_mode", return_value=False), + patch.object(migration.op, "get_bind", return_value=connection), + patch.object(migration.op, "drop_index") as drop_index, + ): + with pytest.raises(RuntimeError, match="duplicated credential identifier"): + migration.upgrade() + drop_index.assert_not_called() + + +def test_upgrade_replaces_legacy_index_with_unique_index() -> None: + migration = _load_migration() + result = MagicMock() + result.scalar_one.return_value = 0 + connection = MagicMock() + connection.execute.return_value = result + + with ( + patch.object(migration.context, "is_offline_mode", return_value=False), + patch.object(migration.op, "get_bind", return_value=connection), + patch.object(migration.op, "drop_index") as drop_index, + patch.object(migration.op, "create_index") as create_index, + ): + migration.upgrade() + + drop_index.assert_called_once_with("buyer_agents_api_key_idx", table_name="buyer_agents") + assert create_index.call_args.kwargs["unique"] is True + + +def test_offline_upgrade_emits_index_changes_without_querying_data() -> None: + migration = _load_migration() + with ( + patch.object(migration.context, "is_offline_mode", return_value=True), + patch.object(migration.op, "get_bind") as get_bind, + patch.object(migration.op, "drop_index"), + patch.object(migration.op, "create_index") as create_index, + ): + migration.upgrade() + + get_bind.assert_not_called() + assert create_index.call_args.kwargs["unique"] is True diff --git a/src/adcp/decisioning/account_projection.py b/src/adcp/decisioning/account_projection.py index a07385f1..af42af18 100644 --- a/src/adcp/decisioning/account_projection.py +++ b/src/adcp/decisioning/account_projection.py @@ -374,6 +374,8 @@ def _scrub_dict(value: dict[str, Any]) -> dict[str, Any]: * ``governance_agents[i].authentication`` — write-only credential. * ``billing_entity.bank`` — write-only bank coordinates. + * ``notification_configs[i].authentication.credentials`` — legacy + webhook bearer/HMAC secret. Walks recursively into nested dicts and lists. Returns a NEW dict — the input is not mutated, so callers (idempotency replay cache, @@ -383,6 +385,11 @@ def _scrub_dict(value: dict[str, Any]) -> dict[str, Any]: for key, sub in value.items(): if key == "governance_agents" and isinstance(sub, list): out[key] = [_scrub_governance_agent_dict(a) if isinstance(a, dict) else a for a in sub] + elif key == "notification_configs" and isinstance(sub, list): + out[key] = [ + _scrub_notification_config_dict(config) if isinstance(config, dict) else config + for config in sub + ] elif key == "billing_entity" and isinstance(sub, dict): out[key] = {k: v for k, v in _scrub_value(sub).items() if k != "bank"} elif key == "authorization": @@ -398,6 +405,17 @@ def _scrub_dict(value: dict[str, Any]) -> dict[str, Any]: return out +def _scrub_notification_config_dict(config: dict[str, Any]) -> dict[str, Any]: + """Strip legacy webhook credentials while preserving the auth scheme.""" + out = {key: _scrub_value(value) for key, value in config.items()} + authentication = out.get("authentication") + if isinstance(authentication, dict): + out["authentication"] = { + key: value for key, value in authentication.items() if key != "credentials" + } + return out + + def _scrub_error_dict(error: dict[str, Any]) -> dict[str, Any]: """Strip private fields from code-specific error details.""" out = {k: _scrub_value(v) for k, v in error.items() if k != "details"} diff --git a/src/adcp/decisioning/pg/buyer_agent_registry.py b/src/adcp/decisioning/pg/buyer_agent_registry.py index c3672490..cd569ec7 100644 --- a/src/adcp/decisioning/pg/buyer_agent_registry.py +++ b/src/adcp/decisioning/pg/buyer_agent_registry.py @@ -190,7 +190,7 @@ def __init__( ) self._sql_select_by_api_key_id = ( f"SELECT {cols} FROM {self._table} " # noqa: S608 - f"WHERE api_key_id = %s" + f"WHERE api_key_id = %s LIMIT 2" ) self._sql_upsert = ( f"INSERT INTO {self._table} (" # noqa: S608 @@ -242,7 +242,7 @@ def create_schema(self) -> None: f" created_at TIMESTAMPTZ NOT NULL DEFAULT now()," f" updated_at TIMESTAMPTZ NOT NULL DEFAULT now()" f");" - f"CREATE INDEX IF NOT EXISTS {table}_api_key_id_idx " # noqa: S608 + f"CREATE UNIQUE INDEX IF NOT EXISTS {table}_api_key_id_uidx " # noqa: S608 f" ON {table} (api_key_id) WHERE api_key_id IS NOT NULL;" f"CREATE INDEX IF NOT EXISTS {table}_status_idx " # noqa: S608 f" ON {table} (status) WHERE status <> 'active';" @@ -514,8 +514,17 @@ def _sync_lookup_by_agent_url(self, agent_url: str) -> BuyerAgent | None: def _sync_lookup_by_api_key_id(self, key: str) -> BuyerAgent | None: with self._pool.connection() as conn, conn.cursor() as cur: cur.execute(self._sql_select_by_api_key_id, (key,)) - row = cur.fetchone() - return _row_to_agent(row) if row else None + rows = cur.fetchall() + if len(rows) > 1: + # Defense in depth for deployments that have not yet applied + # the unique-index migration. Never select an arbitrary + # commercial identity for an ambiguous credential. + logger.error( + "PgBuyerAgentRegistry rejected an ambiguous credential mapping; " + "apply the unique api_key_id index migration" + ) + return None + return _row_to_agent(rows[0]) if rows else None def _row_to_agent(row: Any) -> BuyerAgent: diff --git a/src/adcp/decisioning/pg/buyer_agent_registry.sql b/src/adcp/decisioning/pg/buyer_agent_registry.sql index db8a4f54..d6215b22 100644 --- a/src/adcp/decisioning/pg/buyer_agent_registry.sql +++ b/src/adcp/decisioning/pg/buyer_agent_registry.sql @@ -57,10 +57,11 @@ CREATE TABLE IF NOT EXISTS adcp_buyer_agents ( updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); --- Bearer-credential lookup index. Partial — only rows with an --- api_key_id pay the index cost. Signing-only adopters store no --- bearer keys and the index stays empty. -CREATE INDEX IF NOT EXISTS adcp_buyer_agents_api_key_id_idx +-- A credential identifier must resolve to exactly one commercial identity. +-- Partial — signing-only adopters store NULL and do not occupy the index. +-- Existing deployments should resolve any duplicates before applying this +-- migration; index creation intentionally fails closed when duplicates exist. +CREATE UNIQUE INDEX IF NOT EXISTS adcp_buyer_agents_api_key_id_uidx ON adcp_buyer_agents (api_key_id) WHERE api_key_id IS NOT NULL; diff --git a/src/adcp/decisioning/platform.py b/src/adcp/decisioning/platform.py index 295b36d0..c4921953 100644 --- a/src/adcp/decisioning/platform.py +++ b/src/adcp/decisioning/platform.py @@ -11,7 +11,9 @@ from __future__ import annotations +import asyncio import warnings +from collections import OrderedDict from collections.abc import Awaitable from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any @@ -49,6 +51,10 @@ from adcp.types import GetAdcpCapabilitiesRequest +_NO_AUTH = NoAuth() +_DEFAULT_UPSTREAM_CLIENT_CACHE_SIZE = 128 + + @dataclass class DecisioningCapabilities: """What a platform claims to support. @@ -421,6 +427,12 @@ def create_media_buy(self, req, ctx): #: this attribute. upstream_url: str | None = None + #: Maximum per-platform upstream client pools retained by + #: :meth:`upstream_for`. Least-recently-used entries are closed on + #: eviction. Increase only when one platform intentionally uses more + #: than 128 stable URL/auth/header/transport combinations. + upstream_client_cache_size: int = _DEFAULT_UPSTREAM_CLIENT_CACHE_SIZE + def get_adcp_capabilities_for_request( self, params: GetAdcpCapabilitiesRequest | dict[str, Any] | None = None, @@ -540,7 +552,7 @@ def upstream_for( return self._cached_upstream_client( base_url=base_url, - auth=auth or NoAuth(), + auth=auth or _NO_AUTH, default_headers=default_headers, timeout=timeout, treat_404_as_none=treat_404_as_none, @@ -557,29 +569,48 @@ def _cached_upstream_client( ) -> UpstreamHttpClient: """Per-instance cached :class:`UpstreamHttpClient` factory. - Cache key is ``(base_url, id(auth))``. Pooling correctness + Cache identity includes URL, auth identity, default headers, and + transport behavior. Pooling correctness requires keying on the auth instance — different ``DynamicBearer`` closures for different tenants need distinct clients so the token resolver doesn't get accidentally shared, and the ``UpstreamHttpClient`` itself owns the underlying ``httpx.AsyncClient`` connection pool. - Cache lives on the platform instance (``__dict__`` lazy init); + Cache lives on the platform instance (``__dict__`` lazy init) and + is bounded to prevent adopter-generated option combinations from + retaining connection pools for the platform lifetime; multi-platform processes don't cross-pollute. Adopter code does not mutate the cache; lifecycle is "create once, reuse for the platform instance's lifetime." """ - cache: dict[tuple[str, int], UpstreamHttpClient] | None + cache: OrderedDict[tuple[Any, ...], UpstreamHttpClient] | None cache = getattr(self, "_upstream_client_cache", None) if cache is None: - cache = {} + cache = OrderedDict() self._upstream_client_cache = cache - key = (base_url, id(auth)) + header_key = tuple( + sorted((name.lower(), value) for name, value in (default_headers or {}).items()) + ) + key = ( + base_url, + id(auth), + header_key, + float(timeout), + bool(treat_404_as_none), + ) existing = cache.get(key) if existing is not None: + cache.move_to_end(key) return existing + max_size = int( + getattr(self, "upstream_client_cache_size", _DEFAULT_UPSTREAM_CLIENT_CACHE_SIZE) + ) + if max_size < 1: + raise ValueError("upstream_client_cache_size must be at least 1") + client = UpstreamHttpClient( base_url=base_url, auth=auth, @@ -588,4 +619,31 @@ def _cached_upstream_client( treat_404_as_none=treat_404_as_none, ) cache[key] = client + while len(cache) > max_size: + _, evicted = cache.popitem(last=False) + self._close_evicted_upstream_client(evicted) return client + + def _close_evicted_upstream_client(self, client: UpstreamHttpClient) -> None: + """Close an evicted pool on the active loop, or defer to shutdown.""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + pending = getattr(self, "_upstream_clients_pending_close", None) + if pending is None: + pending = [] + self._upstream_clients_pending_close = pending + pending.append(client) + return + loop.create_task(client.aclose()) + + async def aclose_upstream_clients(self) -> None: + """Release all cached and previously evicted upstream pools.""" + cache = getattr(self, "_upstream_client_cache", None) + clients = list(cache.values()) if cache is not None else [] + if cache is not None: + cache.clear() + clients.extend(getattr(self, "_upstream_clients_pending_close", [])) + self._upstream_clients_pending_close = [] + if clients: + await asyncio.gather(*(client.aclose() for client in clients)) diff --git a/src/adcp/decisioning/proposal_store.py b/src/adcp/decisioning/proposal_store.py index a1ca68a6..7229e584 100644 --- a/src/adcp/decisioning/proposal_store.py +++ b/src/adcp/decisioning/proposal_store.py @@ -407,7 +407,7 @@ def __init__( ``lambda: datetime.now(timezone.utc)``. Tests pin a deterministic clock to validate eviction. """ - self._records: dict[str, ProposalRecord] = {} + self._records: dict[tuple[str, str], ProposalRecord] = {} # Reverse index keyed by (account_id, media_buy_id). Tenant scoping # in the key prevents collisions when adopter media_buy_ids overlap # across tenants (sequential IDs, deterministic test fixtures, etc.) — @@ -417,24 +417,33 @@ def __init__( self._draft_ttl = draft_ttl self._committed_grace = committed_grace self._clock = clock or (lambda: datetime.now(timezone.utc)) - self._creation_times: dict[str, datetime] = {} + self._creation_times: dict[tuple[str, str], datetime] = {} + + def _record_key( + self, proposal_id: str, expected_account_id: str | None + ) -> tuple[str, str] | None: + """Resolve a composite key, refusing ambiguous unscoped lookups.""" + if expected_account_id is not None: + return (expected_account_id, proposal_id) + matches = [key for key in self._records if key[1] == proposal_id] + return matches[0] if len(matches) == 1 else None def _evict_expired_locked(self) -> None: """Remove records past their TTL. Must be called under the lock.""" now = self._clock() - to_remove: list[str] = [] - for proposal_id, record in self._records.items(): - created = self._creation_times.get(proposal_id, now) + to_remove: list[tuple[str, str]] = [] + for key, record in self._records.items(): + created = self._creation_times.get(key, now) if record.state == ProposalState.DRAFT: if now - created > self._draft_ttl: - to_remove.append(proposal_id) + to_remove.append(key) elif record.expires_at is not None: deadline = record.expires_at + self._committed_grace if now > deadline: - to_remove.append(proposal_id) - for proposal_id in to_remove: - removed = self._records.pop(proposal_id, None) - self._creation_times.pop(proposal_id, None) + to_remove.append(key) + for key in to_remove: + removed = self._records.pop(key, None) + self._creation_times.pop(key, None) if removed is not None and removed.media_buy_id is not None: self._media_buy_index.pop((removed.account_id, removed.media_buy_id), None) @@ -448,7 +457,8 @@ async def put_draft( ) -> None: async with self._lock: self._evict_expired_locked() - existing = self._records.get(proposal_id) + key = (account_id, proposal_id) + existing = self._records.get(key) if existing is not None and existing.state != ProposalState.DRAFT: raise AdcpError( "INTERNAL_ERROR", @@ -467,13 +477,13 @@ async def put_draft( recipes=dict(recipes), proposal_payload=dict(proposal_payload), ) - self._records[proposal_id] = record + self._records[key] = record # Track creation time only for fresh records — refine # iterations preserve the original creation time so the # 24h draft TTL is anchored to the start of the buyer's # session, not the most recent iteration. - if proposal_id not in self._creation_times: - self._creation_times[proposal_id] = self._clock() + if key not in self._creation_times: + self._creation_times[key] = self._clock() async def get( self, @@ -483,7 +493,8 @@ async def get( ) -> ProposalRecord | None: async with self._lock: self._evict_expired_locked() - record = self._records.get(proposal_id) + key = self._record_key(proposal_id, expected_account_id) + record = self._records.get(key) if key is not None else None if record is None: return None if expected_account_id is not None and record.account_id != expected_account_id: @@ -501,8 +512,9 @@ async def commit( ) -> None: async with self._lock: self._evict_expired_locked() - record = self._records.get(proposal_id) - if record is None or record.account_id != expected_account_id: + key = (expected_account_id, proposal_id) + record = self._records.get(key) + if record is None: # Cross-tenant probe collapses to "not in store" — same # principal-enumeration defence as :meth:`get`. raise AdcpError( @@ -539,7 +551,7 @@ async def commit( ), recovery="terminal", ) - self._records[proposal_id] = replace( + self._records[key] = replace( record, state=ProposalState.COMMITTED, expires_at=expires_at, @@ -554,10 +566,11 @@ async def try_reserve_consumption( ) -> ProposalRecord: async with self._lock: self._evict_expired_locked() - record = self._records.get(proposal_id) + key = (expected_account_id, proposal_id) + record = self._records.get(key) # Cross-tenant probe collapses to PROPOSAL_NOT_FOUND — same # principal-enumeration defense as :meth:`get`. - if record is None or record.account_id != expected_account_id: + if record is None: raise AdcpError( "PROPOSAL_NOT_FOUND", message=(f"Proposal {proposal_id!r} not found."), @@ -577,7 +590,7 @@ async def try_reserve_consumption( field="proposal_id", ) reserved = replace(record, state=ProposalState.CONSUMING) - self._records[proposal_id] = reserved + self._records[key] = reserved return reserved async def finalize_consumption( @@ -588,8 +601,9 @@ async def finalize_consumption( expected_account_id: str, ) -> None: async with self._lock: - record = self._records.get(proposal_id) - if record is None or record.account_id != expected_account_id: + key = (expected_account_id, proposal_id) + record = self._records.get(key) + if record is None: raise AdcpError( "INTERNAL_ERROR", message=( @@ -622,7 +636,7 @@ async def finalize_consumption( ), recovery="terminal", ) - self._records[proposal_id] = replace( + self._records[key] = replace( record, state=ProposalState.CONSUMED, media_buy_id=media_buy_id, @@ -636,8 +650,9 @@ async def release_consumption( expected_account_id: str, ) -> None: async with self._lock: - record = self._records.get(proposal_id) - if record is None or record.account_id != expected_account_id: + key = (expected_account_id, proposal_id) + record = self._records.get(key) + if record is None: # Idempotent — releasing an unknown id is a no-op so the # adapter-failure rollback path can be unconditional. return @@ -654,7 +669,7 @@ async def release_consumption( ), recovery="terminal", ) - self._records[proposal_id] = replace( + self._records[key] = replace( record, state=ProposalState.COMMITTED, ) @@ -671,8 +686,9 @@ async def mark_consumed( # two-phase methods directly. async with self._lock: self._evict_expired_locked() - record = self._records.get(proposal_id) - if record is None or record.account_id != expected_account_id: + key = (expected_account_id, proposal_id) + record = self._records.get(key) + if record is None: raise AdcpError( "INTERNAL_ERROR", message=( @@ -704,7 +720,7 @@ async def mark_consumed( ), recovery="terminal", ) - self._records[proposal_id] = replace( + self._records[key] = replace( record, state=ProposalState.CONSUMED, media_buy_id=media_buy_id, @@ -718,12 +734,13 @@ async def discard( expected_account_id: str, ) -> None: async with self._lock: - record = self._records.get(proposal_id) - if record is None or record.account_id != expected_account_id: + key = (expected_account_id, proposal_id) + record = self._records.get(key) + if record is None: # Idempotent — unknown id or cross-tenant probe is a no-op. return - self._records.pop(proposal_id, None) - self._creation_times.pop(proposal_id, None) + self._records.pop(key, None) + self._creation_times.pop(key, None) if record.media_buy_id is not None: self._media_buy_index.pop((record.account_id, record.media_buy_id), None) @@ -738,7 +755,7 @@ async def get_by_media_buy_id( proposal_id = self._media_buy_index.get((expected_account_id, media_buy_id)) if proposal_id is None: return None - record = self._records.get(proposal_id) + record = self._records.get((expected_account_id, proposal_id)) if record is None: # Index drift — clean up. self._media_buy_index.pop((expected_account_id, media_buy_id), None) diff --git a/src/adcp/decisioning/registry_cache.py b/src/adcp/decisioning/registry_cache.py index 76b1b7e3..004934ad 100644 --- a/src/adcp/decisioning/registry_cache.py +++ b/src/adcp/decisioning/registry_cache.py @@ -12,8 +12,8 @@ an enumeration probe walking a million ``agent_url`` strings would otherwise hit the DB once per probe; with negative caching it hits the DB once per ``(tenant, agent_url)`` pair within the TTL window. -* :class:`RateLimitedBuyerAgentRegistry` — per-(tenant, lookup-key) - token bucket. On exhaustion, raises ``PERMISSION_DENIED`` with no +* :class:`RateLimitedBuyerAgentRegistry` — aggregate per-tenant plus + per-(tenant, lookup-key) token buckets. On exhaustion, raises ``PERMISSION_DENIED`` with no ``details`` so the wire shape matches every other denied path (registry miss, suspended, blocked) — preserves the spec's omit-on-unestablished-identity rule from PR #393. A distinct @@ -458,7 +458,7 @@ class _Bucket: class RateLimitedBuyerAgentRegistry: - """Per-tenant token-bucket rate limiter wrapping a + """Aggregate per-tenant and per-lookup token-bucket rate limiter wrapping a :class:`BuyerAgentRegistry`. Sized for the credential-stuffing oracle: the registry's @@ -470,7 +470,7 @@ class RateLimitedBuyerAgentRegistry: :param inner: The wrapped :class:`BuyerAgentRegistry`. :param rps_per_tenant: Steady-state requests per second per - ``(tenant_id, lookup_key)`` bucket. Default 100 — high + tenant aggregate and each ``(tenant_id, lookup_key)`` bucket. Default 100 — high enough to absorb a real buyer's storyboard burst, low enough that an enumeration probe at line rate gets cut off. :param burst: Maximum bucket capacity (tokens). Default @@ -484,6 +484,11 @@ class RateLimitedBuyerAgentRegistry: probing). :param time_source: Override for tests — defaults to :func:`time.monotonic`. + :param max_buckets: Hard cap across aggregate and per-lookup bucket state. + When full and no idle bucket can be reclaimed, new lookup identities + fail closed instead of allocating memory. + :param bucket_idle_ttl_seconds: Idle bucket retention. Discarding an idle + bucket is safe because it would have refilled to its full burst. Failure mode ------------ @@ -506,18 +511,28 @@ def __init__( audit_sink: AuditSink | None = None, sink_timeout_seconds: float = 5.0, time_source: Callable[[], float] = time.monotonic, + max_buckets: int = 10_000, + bucket_idle_ttl_seconds: float = 300.0, ) -> None: if rps_per_tenant <= 0: raise ValueError(f"rps_per_tenant must be > 0, got {rps_per_tenant!r}") if burst is not None and burst <= 0: raise ValueError(f"burst must be > 0, got {burst!r}") + if max_buckets < 2: + raise ValueError(f"max_buckets must be >= 2, got {max_buckets!r}") + if bucket_idle_ttl_seconds <= 0: + raise ValueError( + "bucket_idle_ttl_seconds must be > 0, " f"got {bucket_idle_ttl_seconds!r}" + ) self._inner = inner self._rate = rps_per_tenant self._burst = burst if burst is not None else rps_per_tenant self._sink = audit_sink self._sink_timeout = sink_timeout_seconds self._now = time_source - self._buckets: dict[tuple[str | None, str], _Bucket] = {} + self._max_buckets = max_buckets + self._bucket_idle_ttl = bucket_idle_ttl_seconds + self._buckets: OrderedDict[tuple[str | None, str], _Bucket] = OrderedDict() self._lock = asyncio.Lock() async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None: @@ -551,22 +566,11 @@ async def _charge( exhaustion.""" now = self._now() async with self._lock: - bucket = self._buckets.get(key) - if bucket is None: - # New bucket — start full so a fresh tenant gets the - # burst allowance immediately. - bucket = _Bucket(tokens=self._burst, last_refill=now) - self._buckets[key] = bucket - else: - # Refill at ``rate`` tokens/sec, capped at ``burst``. - elapsed = now - bucket.last_refill - bucket.tokens = min(self._burst, bucket.tokens + elapsed * self._rate) - bucket.last_refill = now - if bucket.tokens < 1.0: - exhausted = True - else: - bucket.tokens -= 1.0 - exhausted = False + self._prune_idle(now) + aggregate_key = (tenant_id, "__tenant_aggregate__") + exhausted = not self._spend_locked(aggregate_key, now) + if not exhausted: + exhausted = not self._spend_locked(key, now) if exhausted: # Audit emission OUTSIDE the lock — the sink may be slow. await _emit_audit( @@ -579,6 +583,31 @@ async def _charge( ) raise _denied_error() + def _spend_locked(self, key: tuple[str | None, str], now: float) -> bool: + bucket = self._buckets.get(key) + if bucket is None: + if len(self._buckets) >= self._max_buckets: + return False + bucket = _Bucket(tokens=self._burst, last_refill=now) + self._buckets[key] = bucket + else: + elapsed = max(0.0, now - bucket.last_refill) + bucket.tokens = min(self._burst, bucket.tokens + elapsed * self._rate) + bucket.last_refill = now + self._buckets.move_to_end(key) + if bucket.tokens < 1.0: + return False + bucket.tokens -= 1.0 + return True + + def _prune_idle(self, now: float) -> None: + """Discard up to 16 least-recently-used buckets that are safely idle.""" + for _ in range(min(16, len(self._buckets))): + key, bucket = next(iter(self._buckets.items())) + if now - bucket.last_refill < self._bucket_idle_ttl: + return + self._buckets.pop(key) + # ----- Audit-emitting terminal wrapper ----------------------------- diff --git a/src/adcp/decisioning/roster_store.py b/src/adcp/decisioning/roster_store.py index 1f5d42a9..e3370b1c 100644 --- a/src/adcp/decisioning/roster_store.py +++ b/src/adcp/decisioning/roster_store.py @@ -25,10 +25,9 @@ Design notes ------------ -* **Roster IS the allowlist.** Auth-based filtering happens upstream of - this layer — the framework's account-resolution gate enforces - principal-vs-account scope. The store does not consult ``ctx`` to - filter ``list``. +* **Roster membership is not authorization.** A caller-supplied + ``authorize`` callback binds verified auth to each account. It is a + required factory argument so missing policy fails visibly at boot. * **Immutable post-construction.** The input dict is copied into an internal :class:`MappingProxyType` so external mutation of the caller's dict cannot widen the allowlist after the fact. Adopters @@ -53,15 +52,23 @@ store = create_roster_account_store( roster={ - "acct_alpha": Account(id="acct_alpha", name="Alpha", status="active"), - "acct_beta": Account(id="acct_beta", name="Beta", status="active"), + "acct_alpha": Account( + id="acct_alpha", name="Alpha", status="active", + metadata={"principals": {"https://buyer-alpha.example/"}}, + ), + "acct_beta": Account( + id="acct_beta", name="Beta", status="active", + metadata={"principals": {"https://buyer-beta.example/"}}, + ), }, + authorize=lambda account, auth: auth.principal in account.metadata["principals"], ) """ from __future__ import annotations -from collections.abc import Mapping +import inspect +from collections.abc import Awaitable, Callable, Mapping from types import MappingProxyType from typing import TYPE_CHECKING, Any, Generic, Literal @@ -106,7 +113,11 @@ class _RosterAccountStore(Generic[TMeta]): resolution: Literal["explicit"] = "explicit" - def __init__(self, roster: Mapping[str, Account[TMeta]]) -> None: + def __init__( + self, + roster: Mapping[str, Account[TMeta]], + authorize: Callable[[Account[TMeta], AuthInfo], bool | Awaitable[bool]], + ) -> None: # Copy into a plain dict, then wrap in MappingProxyType so the # store's view is decoupled from the caller's input. Two layers # of protection: external mutation of the input dict can't @@ -121,6 +132,20 @@ def __init__(self, roster: Mapping[str, Account[TMeta]]) -> None: f"its key" ) self._roster: Mapping[str, Account[TMeta]] = MappingProxyType(copied) + self._authorize = authorize + + async def _is_authorized(self, account: Account[TMeta], auth_info: AuthInfo | None) -> bool: + if auth_info is None: + return False + try: + result = self._authorize(account, auth_info) + if inspect.isawaitable(result): + result = await result + return result is True + except Exception: + # Authorization callbacks are security controls: callback + # failures deny access and never widen the roster. + return False async def resolve( self, @@ -136,14 +161,17 @@ async def resolve( Signature mirrors the :class:`AccountStore` Protocol's ``resolve(ref, auth_info=None)`` — the framework dispatcher passes ``auth_info`` as a keyword argument. ``auth_info`` is - accepted for Protocol parity but unused: the roster IS the - allowlist, no auth-based filtering at this layer. + required for access. Missing auth, callback failure, and callback + denial all collapse to ``None`` so a foreign id is + indistinguishable from an unknown id. """ - del auth_info # roster is the allowlist; no per-principal filtering account_id = ref_account_id(ref) if account_id is None: return None - return self._roster.get(account_id) + account = self._roster.get(account_id) + if account is None or not await self._is_authorized(account, auth_info): + return None + return account async def upsert( self, @@ -222,17 +250,18 @@ async def list( filter: dict[str, Any] | None = None, ctx: ResolveContext | None = None, ) -> list[Account[TMeta]]: - """Return every roster entry. - - Adopters who need filtering (status, sandbox, pagination) wrap - ``list`` and post-filter the returned list — the roster store - does not interpret ``filter`` because the typical roster - cardinality (single-digit to low-thousands of accounts per - publisher) is small enough that in-memory filtering at the - adopter layer is fine. + """Return roster entries authorized for the verified principal. + + Missing auth returns ``[]``. + The optional wire ``filter`` remains adopter-defined. """ - del filter, ctx - return list(self._roster.values()) + del filter + auth_info = ctx.auth_info if ctx is not None else None + return [ + account + for account in self._roster.values() + if await self._is_authorized(account, auth_info) + ] def _ref_brand(ref: AccountReference | None) -> dict[str, Any]: @@ -267,10 +296,16 @@ def _ref_operator(ref: AccountReference | None) -> str: def create_roster_account_store( *, roster: Mapping[str, Account[TMeta]], + authorize: Callable[[Account[TMeta], AuthInfo], bool | Awaitable[bool]], ) -> _RosterAccountStore[TMeta]: """Build an :class:`AccountStore` backed by a fixed publisher- curated roster. + ``authorize`` is the required security boundary binding a verified + principal to an account. It may be synchronous or asynchronous. + The callback is required so a roster cannot be mistaken for an + authorization policy. Possession of an account id never grants access. + The returned object conforms to the :class:`AccountStore` Protocol plus the optional :class:`AccountStoreList`, :class:`AccountStoreUpsert`, and :class:`AccountStoreSyncGovernance` @@ -284,12 +319,16 @@ def create_roster_account_store( :class:`ValueError` at construction. The mapping is copied into an internal immutable view, so subsequent mutation of the caller's dict does not affect the store. + :param authorize: Principal/account authorization callback. Receives + the candidate account and verified auth info. Return exactly + ``True`` to grant access. May be async. ``False`` or an exception + denies access. :returns: An :class:`AccountStore` whose: - * :meth:`resolve` returns the roster entry for an + * :meth:`resolve` returns an authorized roster entry for an ``account_id``-arm ref, ``None`` otherwise. - * :meth:`list` returns every roster entry. + * :meth:`list` returns only authorized roster entries. * :meth:`upsert` rejects every input entry with ``PERMISSION_DENIED``. * :meth:`sync_governance` rejects every input entry with @@ -298,4 +337,4 @@ def create_roster_account_store( :raises ValueError: When any roster value's ``id`` does not match its dict key. """ - return _RosterAccountStore(roster) + return _RosterAccountStore(roster, authorize) diff --git a/src/adcp/decisioning/tenant_store.py b/src/adcp/decisioning/tenant_store.py index c346080c..79d54dfc 100644 --- a/src/adcp/decisioning/tenant_store.py +++ b/src/adcp/decisioning/tenant_store.py @@ -325,9 +325,9 @@ async def upsert( 1. Compute the entry's tenant via ``resolve_by_ref``. 2. Compare against the auth principal's tenant (``resolve_from_auth(ctx)``, computed once per call). - 3. Unknown ref → ``ACCOUNT_NOT_FOUND``. - 4. Auth tenant ``None`` OR auth tenant != entry tenant → - ``PERMISSION_DENIED`` (fail-closed). + 3. Unknown ref OR an account outside the authenticated tenant → + ``ACCOUNT_NOT_FOUND`` (existence-hiding). + 4. Auth tenant ``None`` → ``PERMISSION_DENIED`` (fail-closed). 5. Otherwise, dispatch to ``upsert_row`` (or no-op ``action='unchanged'`` if no hook). @@ -381,13 +381,20 @@ async def upsert( ) ) continue - if auth_tid is None or auth_tid != entry_tid: + if auth_tid is None: rows.append( _build_failed_sync_accounts_row( ref, "PERMISSION_DENIED", _permission_denied_message(ref) ) ) continue + if auth_tid != entry_tid: + rows.append( + _build_failed_sync_accounts_row( + ref, "ACCOUNT_NOT_FOUND", _account_not_found_message(ref) + ) + ) + continue if self._upsert_row is None: rows.append(_default_unchanged_row(ref)) else: diff --git a/src/adcp/decisioning/upstream.py b/src/adcp/decisioning/upstream.py index 4c184d50..a122b9c2 100644 --- a/src/adcp/decisioning/upstream.py +++ b/src/adcp/decisioning/upstream.py @@ -106,12 +106,9 @@ def _project_status( not_found_code: str, method: str, path: str, - body_text: str, ) -> AdcpError: - """Project an upstream non-2xx status to a spec-conformant AdcpError.""" - snippet = body_text[:200] if body_text else "" - suffix = f" — {snippet}" if snippet else "" - base = f"upstream {method} {path} failed: {status_code}{suffix}" + """Project an upstream status without exposing its untrusted response body.""" + base = f"upstream {method} {path} failed: {status_code}" if status_code == 401: return AdcpError( "AUTH_REQUIRED", @@ -270,16 +267,11 @@ async def _request( if response.status_code == 404 and self._treat_404_as_none: return None if response.status_code >= 300: - try: - body_text = response.text - except Exception: # pragma: no cover — defensive - body_text = "" raise _project_status( response.status_code, not_found_code=not_found_code, method=method, path=path, - body_text=body_text, ) if response.status_code == 204 or not response.content: return {} diff --git a/src/adcp/server/auth.py b/src/adcp/server/auth.py index e9853244..7121b0a2 100644 --- a/src/adcp/server/auth.py +++ b/src/adcp/server/auth.py @@ -437,14 +437,14 @@ async def dispatch(self, request: Request, call_next: Any) -> Any: tenant_token = None metadata_token = None try: - if self.is_discovery_request(method, tool): + bearer = self._extract_bearer(request) + if self.is_discovery_request(method, tool) and not bearer: principal_token = current_principal.set(None) tenant_token = current_tenant.set(None) metadata_token = current_principal_metadata.set(None) _set_request_state(request, None, None, None) return await call_next(request) - bearer = self._extract_bearer(request) if not bearer: if self._allow_unauthenticated: # Network-trust deployment: no bearer is expected on this @@ -495,6 +495,25 @@ async def dispatch(self, request: Request, call_next: Any) -> Any: principal.tenant_id, principal_metadata, ) + # The MCP stateful transport binds Mcp-Session-Id to scope['user']. + # Mirror the validated ADCP principal into MCP's standard shape so + # sessions created through this middleware cannot be reused by a + # differently authenticated caller. The raw token remains + # request-scoped; the session manager stores only its derived + # (client_id, issuer, subject) authorization context. + from mcp.server.auth.middleware.bearer_auth import ( # noqa: PLC0415 + AuthenticatedUser, + ) + from mcp.server.auth.provider import AccessToken # noqa: PLC0415 + + request.scope["user"] = AuthenticatedUser( + AccessToken( + token=bearer, + client_id=principal.caller_identity, + scopes=[], + subject=principal.tenant_id, + ) + ) return await call_next(request) finally: # Reset unconditionally so a later task sharing this context diff --git a/src/adcp/server/responses.py b/src/adcp/server/responses.py index 08889d58..a06af4a9 100644 --- a/src/adcp/server/responses.py +++ b/src/adcp/server/responses.py @@ -126,12 +126,12 @@ def _strip_write_only_fields(value: Any) -> Any: * ``governance_agents[i].authentication`` — write-only credential. * ``billing_entity.bank`` — write-only bank coordinates. + * ``notification_configs[i].authentication.credentials`` — legacy + webhook bearer/HMAC secret. - Pydantic models are passed through unchanged — adopters using - typed response models are responsible for the strip via - :func:`adcp.decisioning.project_account_for_response` or - equivalent. Loose dicts (the more common case for hand-built - builder calls) get the recursive walk. + Pydantic models are dumped before the same recursive strip. Response + builders are a security boundary and cannot assume callers selected a + response-only projection rather than a request model carrying secrets. """ if isinstance(value, dict): out: dict[str, Any] = {} @@ -152,6 +152,20 @@ def _strip_write_only_fields(value: Any) -> Any: out[key] = projected elif key == "billing_entity" and isinstance(sub, dict): out[key] = {k: _strip_write_only_fields(v) for k, v in sub.items() if k != "bank"} + elif key == "notification_configs" and isinstance(sub, list): + projected_configs: list[Any] = [] + for config in sub: + if not isinstance(config, dict): + projected_configs.append(config) + continue + projected_config = {k: _strip_write_only_fields(v) for k, v in config.items()} + authentication = projected_config.get("authentication") + if isinstance(authentication, dict): + projected_config["authentication"] = { + k: v for k, v in authentication.items() if k != "credentials" + } + projected_configs.append(projected_config) + out[key] = projected_configs else: out[key] = _strip_write_only_fields(sub) return out @@ -163,21 +177,19 @@ def _strip_write_only_fields(value: Any) -> Any: def _serialize(items: list[Any]) -> list[Any]: """Serialize a list of dicts or Pydantic models to plain dicts. - Loose-dict items (adopters returning ``{**db_record, ...}`` from - a hand-built response builder) get a recursive write-only-field - strip via :func:`_strip_write_only_fields` so + Every item gets a recursive write-only-field strip via + :func:`_strip_write_only_fields` so ``governance_agents[i].authentication`` and ``billing_entity.bank`` can't smuggle through, followed by :func:`_strip_none_values` to remove ``null``-valued keys that the bundled JSON schemas declare as - non-nullable (e.g. ``ImageAsset.format``). Pydantic models are - passed through their own ``model_dump(exclude_none=True)`` — the - typed projections at :mod:`adcp.decisioning.account_projection` are - responsible for the write-only strip on that path. + non-nullable (e.g. ``ImageAsset.format``). Pydantic models are first + converted through ``model_dump(exclude_none=True)`` and then scrubbed. """ out: list[Any] = [] for p in items: if hasattr(p, "model_dump"): - out.append(p.model_dump(mode="json", exclude_none=True)) + dumped = p.model_dump(mode="json", exclude_none=True) + out.append(_strip_write_only_fields(dumped)) elif isinstance(p, dict): out.append(_strip_none_values(_strip_write_only_fields(p))) else: diff --git a/src/adcp/types/projections.py b/src/adcp/types/projections.py index 2b432695..b1e429e9 100644 --- a/src/adcp/types/projections.py +++ b/src/adcp/types/projections.py @@ -2,11 +2,10 @@ The AdCP spec marks certain fields as ``writeOnly: true`` — present in requests so adopters can populate them, but MUST NOT be echoed in -responses. The clearest case is ``BusinessEntity.bank``: IBANs, BICs, -routing numbers, and account numbers flow into the seller during account -setup and stay there. Pydantic's default serialization round-trips -everything, so an adopter who reuses an internal ``Account`` model on -the response path can leak bank details without realizing it. +responses. ``BusinessEntity.bank`` and notification authentication credentials +flow into the seller during account setup and stay there. Pydantic's default +serialization round-trips everything, so an adopter who reuses an internal +``Account`` model on the response path can leak secrets without realizing it. The projections here type-narrow the write-only fields to ``None``: construction with a non-None value raises ``ValidationError``, and the @@ -28,11 +27,13 @@ from collections.abc import Iterator, Mapping from typing import Any -from pydantic import Field, field_validator +from pydantic import ConfigDict, Field, field_validator from adcp._version import normalize_to_release_precision -from adcp.types import Account, BusinessEntity +from adcp.types import Account, AuthenticationScheme, BusinessEntity, NotificationConfig +from adcp.types.base import AdCPBaseModel from adcp.types.capabilities import GeoPostalAreas, LegacyPostalCodeSystem +from adcp.types.variants import SchemaVariant _NATIVE_TO_LEGACY_POSTAL: dict[tuple[str, str], LegacyPostalCodeSystem] = { ("US", "zip"): LegacyPostalCodeSystem.us_zip, @@ -165,6 +166,33 @@ def _reject_bank(cls, v: Any) -> None: return None +class _NotificationAuthenticationResponse(AdCPBaseModel): + """Response projection of legacy notification authentication.""" + + model_config = ConfigDict(extra="forbid") + + schemes: list[AuthenticationScheme] = Field(min_length=1, max_length=1) + credentials: Any = Field(default=None, exclude=True) + + @field_validator("credentials", mode="before") + @classmethod + def _reject_credentials(cls, v: Any) -> None: + if v is not None: + raise ValueError( + "Notification authentication credentials are write-only and " + "must not be included in an AccountResponse. Drop the field " + "before constructing a response, or use to_account_response() " + "to strip it." + ) + return None + + +class _NotificationConfigResponse(NotificationConfig): + """Account notification config with write-only credentials stripped.""" + + authentication: SchemaVariant[_NotificationAuthenticationResponse | None] = None + + class AccountResponse(Account): """Response projection of :class:`Account` — billing_entity is the bank-stripped variant. @@ -177,12 +205,14 @@ class AccountResponse(Account): """ billing_entity: BusinessEntityResponse | None = None + notification_configs: SchemaVariant[list[_NotificationConfigResponse] | None] = None def to_account_response(account: Account) -> AccountResponse: """Project an internal ``Account`` to its response shape. - Strips ``billing_entity.bank`` and returns an :class:`AccountResponse`. + Strips ``billing_entity.bank`` and notification authentication credentials, + then returns an :class:`AccountResponse`. The remaining fields (legal_name, tax_id, address, contacts, vat_id, registration_number, ext) round-trip unchanged. ``reporting_bucket``, ``governance_agents``, and other non-write-only fields are preserved. @@ -194,6 +224,9 @@ def to_account_response(account: Account) -> AccountResponse: payload = account.model_dump(mode="python") if isinstance(payload.get("billing_entity"), dict): payload["billing_entity"].pop("bank", None) + for config in payload.get("notification_configs") or []: + if isinstance(config, dict) and isinstance(config.get("authentication"), dict): + config["authentication"].pop("credentials", None) return AccountResponse.model_validate(payload) diff --git a/tests/conformance/decisioning/test_pg_buyer_agent_registry.py b/tests/conformance/decisioning/test_pg_buyer_agent_registry.py index 91f3fe1a..0df374d5 100644 --- a/tests/conformance/decisioning/test_pg_buyer_agent_registry.py +++ b/tests/conformance/decisioning/test_pg_buyer_agent_registry.py @@ -181,6 +181,52 @@ def test_resolve_by_credential_returns_none_for_unknown_key(isolated_pool) -> No assert result is None +def test_upsert_rejects_duplicate_credential_identifier(isolated_pool) -> None: + registry = _registry(isolated_pool) + registry.upsert( + BuyerAgent( + agent_url="https://first-buyer/", + display_name="First Buyer", + status="active", + ), + api_key_id="shared-credential", + ) + + with pytest.raises(psycopg.errors.UniqueViolation): + registry.upsert( + BuyerAgent( + agent_url="https://second-buyer/", + display_name="Second Buyer", + status="active", + ), + api_key_id="shared-credential", + ) + + +def test_legacy_duplicate_credentials_fail_closed(isolated_pool) -> None: + """Lookup remains safe before an existing deployment applies the index.""" + pool, table = isolated_pool + registry = _registry(isolated_pool) + with pool.connection() as conn, conn.cursor() as cur: + cur.execute(f"DROP INDEX {table}_api_key_id_uidx") + for suffix in ("one", "two"): + registry.upsert( + BuyerAgent( + agent_url=f"https://buyer-{suffix}/", + display_name=f"Buyer {suffix}", + status="active", + ), + api_key_id="legacy-duplicate", + ) + + result = asyncio.run( + registry.resolve_by_credential( + ApiKeyCredential(kind="api_key", key_id="legacy-duplicate"), + ) + ) + assert result is None + + # ----- upsert (admin path) ----------------------------------------------- diff --git a/tests/test_account_projections.py b/tests/test_account_projections.py index dfee0930..4deb8c25 100644 --- a/tests/test_account_projections.py +++ b/tests/test_account_projections.py @@ -168,3 +168,63 @@ def test_to_account_response_preserves_reporting_bucket() -> None: assert response.reporting_bucket is not None assert response.reporting_bucket.bucket == "reports-acme-prod" assert response.reporting_bucket.file_retention_days == 30 + + +# ---- notification_configs — write-only authentication credentials guard ---- + + +def test_account_response_rejects_notification_credentials() -> None: + """Response-shaped accounts cannot be constructed with webhook secrets.""" + with pytest.raises(ValidationError) as excinfo: + AccountResponse.model_validate( + { + "account_id": "acct-1", + "name": "Acme", + "status": "active", + "notification_configs": [ + { + "subscriber_id": "buyer-primary", + "url": "https://buyer.example/webhooks", + "event_types": ["creative.status_changed"], + "authentication": { + "schemes": ["Bearer"], + "credentials": "secret-token-that-is-at-least-32-chars", + }, + } + ], + } + ) + + assert any( + "credentials" in str(loc) for error in excinfo.value.errors() for loc in error["loc"] + ) + + +def test_to_account_response_strips_notification_credentials() -> None: + """Projection keeps subscription state and scheme but drops its secret.""" + internal = Account.model_validate( + { + "account_id": "acct-1", + "name": "Acme", + "status": "active", + "notification_configs": [ + { + "subscriber_id": "buyer-primary", + "url": "https://buyer.example/webhooks", + "event_types": ["creative.status_changed"], + "authentication": { + "schemes": ["Bearer"], + "credentials": "secret-token-that-is-at-least-32-chars", + }, + "active": True, + } + ], + } + ) + + dumped = to_account_response(internal).model_dump(mode="json", exclude_none=True) + config = dumped["notification_configs"][0] + assert config["subscriber_id"] == "buyer-primary" + assert config["active"] is True + assert config["authentication"]["schemes"] == ["Bearer"] + assert "credentials" not in config["authentication"] diff --git a/tests/test_account_v3_wire.py b/tests/test_account_v3_wire.py index 22b544eb..9de1b528 100644 --- a/tests/test_account_v3_wire.py +++ b/tests/test_account_v3_wire.py @@ -317,6 +317,31 @@ def test_account_authorization_projection_strips_accidental_secrets() -> None: assert "conn_private" not in str(scrubbed) +def test_loose_account_projection_strips_notification_credentials() -> None: + secret = "notification-secret-that-must-never-echo" + payload = { + "accounts": [ + { + "account_id": "acct_1", + "notification_configs": [ + { + "subscriber_id": "buyer-primary", + "authentication": { + "schemes": ["HMAC-SHA256"], + "credentials": secret, + }, + } + ], + } + ] + } + + scrubbed = strip_credentials_from_wire_result("list_accounts", payload) + authentication = scrubbed["accounts"][0]["notification_configs"][0]["authentication"] + assert authentication == {"schemes": ["HMAC-SHA256"]} + assert secret not in str(scrubbed) + + @pytest.mark.asyncio async def test_handler_list_accounts_returns_ads_and_publisher_identity_grants() -> None: class _TikTokAccountStore: diff --git a/tests/test_buyer_agent_registry_cache.py b/tests/test_buyer_agent_registry_cache.py index 793880e9..d22daea0 100644 --- a/tests/test_buyer_agent_registry_cache.py +++ b/tests/test_buyer_agent_registry_cache.py @@ -314,9 +314,8 @@ async def test_rate_limit_refills_over_time() -> None: @pytest.mark.asyncio -async def test_rate_limit_isolates_distinct_lookup_keys() -> None: - """Each ``(tenant, lookup_key)`` gets its own bucket — exhausting - one does not affect another.""" +async def test_rate_limit_aggregate_blocks_rotating_lookup_keys() -> None: + """Fresh identifiers cannot bypass the tenant's aggregate budget.""" inner = FakeRegistry() clock = FakeClock(start=0.0) limiter = RateLimitedBuyerAgentRegistry( @@ -327,10 +326,41 @@ async def test_rate_limit_isolates_distinct_lookup_keys() -> None: await limiter.resolve_by_agent_url("https://agent-A/") with pytest.raises(AdcpError): - await limiter.resolve_by_agent_url("https://agent-A/") + await limiter.resolve_by_agent_url("https://agent-B/") - # Different key — fresh bucket. + +@pytest.mark.asyncio +async def test_rate_limit_bucket_state_is_bounded() -> None: + limiter = RateLimitedBuyerAgentRegistry( + FakeRegistry(), + rps_per_tenant=100.0, + burst=100.0, + max_buckets=4, + time_source=FakeClock(start=0.0), + ) + for suffix in ("A", "B", "C"): + await limiter.resolve_by_agent_url(f"https://agent-{suffix}/") + with pytest.raises(AdcpError): + await limiter.resolve_by_agent_url("https://agent-D/") + assert len(limiter._buckets) == 4 + + +@pytest.mark.asyncio +async def test_rate_limit_idle_buckets_expire() -> None: + clock = FakeClock(start=0.0) + limiter = RateLimitedBuyerAgentRegistry( + FakeRegistry(), + rps_per_tenant=100.0, + max_buckets=4, + bucket_idle_ttl_seconds=10.0, + time_source=clock, + ) + await limiter.resolve_by_agent_url("https://agent-A/") await limiter.resolve_by_agent_url("https://agent-B/") + assert len(limiter._buckets) == 3 + clock.advance(11.0) + await limiter.resolve_by_agent_url("https://agent-C/") + assert len(limiter._buckets) == 2 # ----- Audit emission: every outcome fires an event -------------------- diff --git a/tests/test_credential_leak_strip.py b/tests/test_credential_leak_strip.py index a908b853..b3765909 100644 --- a/tests/test_credential_leak_strip.py +++ b/tests/test_credential_leak_strip.py @@ -611,6 +611,17 @@ def test_sync_accounts_response_builder_round_trip_strip() -> None: }, } ], + "notification_configs": [ + { + "subscriber_id": "buyer-primary", + "url": "https://buyer.example/webhooks", + "event_types": ["creative.status_changed"], + "authentication": { + "schemes": ["Bearer"], + "credentials": _BEARER, + }, + } + ], } ], ) @@ -619,9 +630,41 @@ def test_sync_accounts_response_builder_round_trip_strip() -> None: assert _BEARER not in str(response) assert "bank" not in serialized["billing_entity"] assert "authentication" not in serialized["governance_agents"][0] + notification_auth = serialized["notification_configs"][0]["authentication"] + assert notification_auth == {"schemes": ["Bearer"]} assert serialized["billing_entity"]["legal_name"] == "Acme Inc." +def test_response_builder_scrubs_notification_credentials_from_pydantic_models() -> None: + """Request-capable Pydantic models are not trusted on a response edge.""" + from adcp.server.responses import sync_accounts_response + from adcp.types import Account + + account = Account.model_validate( + { + "account_id": "acct_1", + "name": "Acme", + "status": "active", + "notification_configs": [ + { + "subscriber_id": "buyer-primary", + "url": "https://buyer.example/webhooks", + "event_types": ["creative.status_changed"], + "authentication": { + "schemes": ["Bearer"], + "credentials": _BEARER, + }, + } + ], + } + ) + + response = sync_accounts_response([account]) # type: ignore[list-item] + authentication = response["accounts"][0]["notification_configs"][0]["authentication"] + assert authentication == {"schemes": ["Bearer"]} + assert _BEARER not in str(response) + + # --------------------------------------------------------------------------- # M3 — ctx_metadata fail-closed gate # --------------------------------------------------------------------------- diff --git a/tests/test_mcp_stateful_session.py b/tests/test_mcp_stateful_session.py index 90415760..918cee56 100644 --- a/tests/test_mcp_stateful_session.py +++ b/tests/test_mcp_stateful_session.py @@ -401,6 +401,84 @@ async def get_products( assert received["tenant_id"] == "t-acme" +@pytest.mark.asyncio +async def test_stateful_session_is_bound_to_authenticated_principal() -> None: + """A second valid bearer principal cannot attach to another session.""" + from adcp.server import ( + BearerTokenAuthMiddleware, + Principal, + create_mcp_server, + validator_from_token_map, + ) + + mcp = create_mcp_server( + _BareHandler(), + name="t", + advertise_all=True, + allowed_hosts=["localhost", "127.0.0.1"], + ) + app = mcp.streamable_http_app() + app.add_middleware( + BearerTokenAuthMiddleware, + validate_token=validator_from_token_map( + { + "token-alice": Principal(caller_identity="alice", tenant_id="tenant-a"), + "token-bob": Principal(caller_identity="bob", tenant_id="tenant-b"), + } + ), + ) + base_headers = { + "content-type": "application/json", + "accept": "application/json, text/event-stream", + } + + async with LifespanManager(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://localhost", + follow_redirects=True, + ) as client: + init = await client.post( + "/mcp/", + json={ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "alice", "version": "1"}, + }, + }, + headers={**base_headers, "authorization": "Bearer token-alice"}, + ) + assert init.status_code == 200, init.text + session_id = init.headers["mcp-session-id"] + + hijack = await client.post( + "/mcp/", + json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}, + headers={ + **base_headers, + "authorization": "Bearer token-bob", + "mcp-session-id": session_id, + }, + ) + assert hijack.status_code == 404 + assert "Session not found" in hijack.text + + owner = await client.post( + "/mcp/", + json={"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}, + headers={ + **base_headers, + "authorization": "Bearer token-alice", + "mcp-session-id": session_id, + }, + ) + assert owner.status_code == 200, owner.text + + @pytest.mark.asyncio async def test_stateful_rejects_request_without_session_id() -> None: """Inverse of the above — without ``Mcp-Session-Id`` the upstream diff --git a/tests/test_pg_buyer_agent_registry_unit.py b/tests/test_pg_buyer_agent_registry_unit.py new file mode 100644 index 00000000..5db8dfda --- /dev/null +++ b/tests/test_pg_buyer_agent_registry_unit.py @@ -0,0 +1,76 @@ +"""Database-independent security tests for PgBuyerAgentRegistry SQL paths.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from adcp.decisioning.pg import buyer_agent_registry as registry_module +from adcp.decisioning.pg.buyer_agent_registry import PgBuyerAgentRegistry + + +class _Cursor: + def __init__(self, rows: list[tuple[Any, ...]] | None = None) -> None: + self.rows = rows or [] + self.queries: list[str] = [] + + def __enter__(self) -> _Cursor: + return self + + def __exit__(self, *args: object) -> None: + return None + + def execute(self, query: str, params: object = None) -> None: + del params + self.queries.append(query) + + def fetchall(self) -> list[tuple[Any, ...]]: + return self.rows + + +class _Connection: + def __init__(self, cursor: _Cursor) -> None: + self._cursor = cursor + + def __enter__(self) -> _Connection: + return self + + def __exit__(self, *args: object) -> None: + return None + + def cursor(self) -> _Cursor: + return self._cursor + + +class _Pool: + def __init__(self, cursor: _Cursor) -> None: + self._connection = _Connection(cursor) + + def connection(self) -> _Connection: + return self._connection + + +@pytest.fixture(autouse=True) +def _enable_optional_pg_module(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(registry_module, "PG_AVAILABLE", True) + + +def test_ambiguous_legacy_credential_mapping_fails_closed() -> None: + cursor = _Cursor(rows=[("first",), ("second",)]) + registry = PgBuyerAgentRegistry(pool=_Pool(cursor)) # type: ignore[arg-type] + + assert registry._sync_lookup_by_api_key_id("shared") is None + assert "LIMIT 2" in cursor.queries[0] + + +def test_schema_bootstrap_creates_partial_unique_credential_index() -> None: + cursor = _Cursor() + registry = PgBuyerAgentRegistry(pool=_Pool(cursor)) # type: ignore[arg-type] + + registry.create_schema() + + ddl = cursor.queries[0] + assert "CREATE UNIQUE INDEX IF NOT EXISTS" in ddl + assert "api_key_id_uidx" in ddl + assert "WHERE api_key_id IS NOT NULL" in ddl diff --git a/tests/test_proposal_store.py b/tests/test_proposal_store.py index d3e5c001..74ceb29a 100644 --- a/tests/test_proposal_store.py +++ b/tests/test_proposal_store.py @@ -363,6 +363,30 @@ async def test_get_cross_tenant_returns_none(store: InMemoryProposalStore) -> No assert await store.get("p1", expected_account_id="acct_a") is not None +@pytest.mark.asyncio +async def test_same_proposal_id_is_isolated_by_account(store: InMemoryProposalStore) -> None: + """Two accounts may use the same buyer-generated proposal id safely.""" + await store.put_draft( + proposal_id="shared", + account_id="acct_a", + recipes={}, + proposal_payload={"owner": "a"}, + ) + await store.put_draft( + proposal_id="shared", + account_id="acct_b", + recipes={}, + proposal_payload={"owner": "b"}, + ) + + record_a = await store.get("shared", expected_account_id="acct_a") + record_b = await store.get("shared", expected_account_id="acct_b") + assert record_a is not None and record_a.proposal_payload == {"owner": "a"} + assert record_b is not None and record_b.proposal_payload == {"owner": "b"} + # An unscoped lookup cannot safely choose between colliding accounts. + assert await store.get("shared") is None + + @pytest.mark.asyncio async def test_get_by_media_buy_id_cross_tenant_returns_none( store: InMemoryProposalStore, diff --git a/tests/test_roster_store.py b/tests/test_roster_store.py index 95cf69ae..5696897b 100644 --- a/tests/test_roster_store.py +++ b/tests/test_roster_store.py @@ -1,12 +1,12 @@ """Tests for :func:`adcp.decisioning.create_roster_account_store`. Shape C ``AccountStore`` factory for publisher-curated rosters where the -adopter has a fixed allowlist of accounts. Pairs with +adopter has a fixed roster plus a principal/account authorization callback. Pairs with :class:`SingletonAccounts` (Shape derived) and :class:`ExplicitAccounts` (Shape explicit, loader-driven). -The roster IS the allowlist — auth-based filtering happens upstream of -this layer. Write paths (``upsert`` / ``sync_governance``) fail closed +Roster membership is not authorization. Write paths +(``upsert`` / ``sync_governance``) fail closed with ``PERMISSION_DENIED`` per-entry; the roster is read-only by design. """ @@ -45,6 +45,14 @@ def _make_roster() -> dict[str, Account]: } +_AUTH = AuthInfo(kind="derived", principal="agent_foo", credential=None) + + +def _allow_all(account: Account, auth: AuthInfo) -> bool: + del account + return auth.principal == "agent_foo" + + # --------------------------------------------------------------------------- # resolve # --------------------------------------------------------------------------- @@ -52,8 +60,8 @@ def _make_roster() -> dict[str, Account]: def test_resolve_hit_returns_account() -> None: """ref carrying a known ``account_id`` returns the roster entry.""" - store = create_roster_account_store(roster=_make_roster()) - result = asyncio.run(store.resolve(_by_id("acct_alpha"))) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) + result = asyncio.run(store.resolve(_by_id("acct_alpha"), auth_info=_AUTH)) assert result is not None assert result.id == "acct_alpha" assert result.name == "Alpha" @@ -62,8 +70,8 @@ def test_resolve_hit_returns_account() -> None: def test_resolve_miss_returns_none() -> None: """ref carrying an unknown ``account_id`` returns ``None`` — fall-through path the framework projects to ``ACCOUNT_NOT_FOUND``.""" - store = create_roster_account_store(roster=_make_roster()) - result = asyncio.run(store.resolve(_by_id("acct_unknown"))) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) + result = asyncio.run(store.resolve(_by_id("acct_unknown"), auth_info=_AUTH)) assert result is None @@ -71,8 +79,10 @@ def test_resolve_natural_key_returns_none() -> None: """``{brand, operator}``-shaped refs return ``None`` — publisher- curated rosters are queried by explicit id only. Adopters wanting natural-key resolution wrap ``resolve``.""" - store = create_roster_account_store(roster=_make_roster()) - result = asyncio.run(store.resolve(_by_natural_key("alpha.example.com", "alpha.example.com"))) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) + result = asyncio.run( + store.resolve(_by_natural_key("alpha.example.com", "alpha.example.com"), auth_info=_AUTH) + ) assert result is None @@ -81,8 +91,8 @@ def test_resolve_none_ref_returns_none() -> None: ``list_creative_formats``, ``preview_creative``) pass ``ref=None``; the helper returns ``None`` and adopters wrap to synthesize a publisher singleton when needed.""" - store = create_roster_account_store(roster=_make_roster()) - result = asyncio.run(store.resolve(None)) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) + result = asyncio.run(store.resolve(None, auth_info=_AUTH)) assert result is None @@ -90,22 +100,19 @@ def test_resolve_accepts_auth_info_kwarg() -> None: """The framework dispatcher calls ``accounts.resolve(ref_dict, auth_info=auth_info)`` — i.e. ``auth_info`` is a keyword argument on every dispatch path. Verify the roster store accepts that exact - call shape (and ignores ``auth_info`` because the roster IS the - allowlist).""" - store = create_roster_account_store(roster=_make_roster()) + call shape and uses it for the authorization callback.""" + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) auth = AuthInfo(kind="signed_request", principal="agent_foo", scopes=["read"]) result = asyncio.run(store.resolve(_by_id("acct_alpha"), auth_info=auth)) assert result is not None assert result.id == "acct_alpha" -def test_resolve_positional_no_auth_info() -> None: - """Positional single-arg calls (no ``auth_info``) keep working — - matches the Protocol's ``auth_info=None`` default.""" - store = create_roster_account_store(roster=_make_roster()) +def test_resolve_without_auth_info_fails_closed() -> None: + """Possession of a known account id is not authorization.""" + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) result = asyncio.run(store.resolve(_by_id("acct_beta"))) - assert result is not None - assert result.id == "acct_beta" + assert result is None def test_store_conforms_to_account_store_protocol() -> None: @@ -113,7 +120,7 @@ def test_store_conforms_to_account_store_protocol() -> None: boot-time platform validator calls ``isinstance(store, AccountStore)``. Any structural drift between the roster store's ``resolve`` signature and the Protocol breaks that check.""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) assert isinstance(store, AccountStore) @@ -121,7 +128,7 @@ def test_resolution_literal_is_explicit() -> None: """Boot-time platform validation reads ``store.resolution`` to fail fast on misconfigured deployments. Roster stores are ``'explicit'`` — wire ref drives lookup.""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) assert store.resolution == "explicit" @@ -131,11 +138,10 @@ def test_resolution_literal_is_explicit() -> None: def test_list_returns_full_roster() -> None: - """``list_accounts`` returns every roster entry. Auth-based - filtering is upstream — the roster IS the allowlist.""" + """An authorizer that grants both accounts returns the full roster.""" roster = _make_roster() - store = create_roster_account_store(roster=roster) - result = asyncio.run(store.list(ctx=ResolveContext())) + store = create_roster_account_store(roster=roster, authorize=_allow_all) + result = asyncio.run(store.list(ctx=ResolveContext(auth_info=_AUTH))) assert len(result) == 2 ids = {a.id for a in result} assert ids == {"acct_alpha", "acct_beta"} @@ -143,11 +149,28 @@ def test_list_returns_full_roster() -> None: def test_list_empty_roster() -> None: """Empty roster lists empty — not an error.""" - store = create_roster_account_store(roster={}) - result = asyncio.run(store.list(ctx=ResolveContext())) + store = create_roster_account_store(roster={}, authorize=_allow_all) + result = asyncio.run(store.list(ctx=ResolveContext(auth_info=_AUTH))) assert result == [] +def test_authorization_callback_filters_direct_lookup_and_list() -> None: + def authorize(account: Account, auth: AuthInfo) -> bool: + return account.id == "acct_alpha" and auth.principal == "agent_alpha" + + store = create_roster_account_store(roster=_make_roster(), authorize=authorize) + auth = AuthInfo(kind="derived", principal="agent_alpha", credential=None) + assert asyncio.run(store.resolve(_by_id("acct_alpha"), auth_info=auth)) is not None + assert asyncio.run(store.resolve(_by_id("acct_beta"), auth_info=auth)) is None + listed = asyncio.run(store.list(ctx=ResolveContext(auth_info=auth))) + assert [account.id for account in listed] == ["acct_alpha"] + + +def test_omitted_authorization_callback_fails_at_construction() -> None: + with pytest.raises(TypeError, match="authorize"): + create_roster_account_store(roster=_make_roster()) # type: ignore[call-arg] + + # --------------------------------------------------------------------------- # upsert — fail-closed PERMISSION_DENIED per entry # --------------------------------------------------------------------------- @@ -159,7 +182,7 @@ def test_upsert_denies_every_entry() -> None: ``failed`` row with ``PERMISSION_DENIED`` so the wire response surfaces the rejection per-entry instead of operation-level raising (which would fail the whole batch).""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) refs = [ _by_natural_key("acme.com", "acme.com"), _by_natural_key("globex.com", "globex.com"), @@ -176,7 +199,7 @@ def test_upsert_denies_every_entry() -> None: def test_upsert_empty_refs_returns_empty() -> None: """An empty refs list returns an empty result list — not an error.""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) rows = asyncio.run(store.upsert([], ctx=ResolveContext())) assert rows == [] @@ -189,7 +212,7 @@ def test_upsert_denies_by_id_refs_with_conformant_row_shape() -> None: shape conforms to :class:`SyncAccountsResultRow` (instance type + required fields populated, so the framework's wire projector won't crash on a missing field).""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) rows = asyncio.run( store.upsert([_by_id("acct_alpha"), _by_id("acct_unknown")], ctx=ResolveContext()) ) @@ -210,7 +233,7 @@ def test_upsert_echoes_brand_operator_for_natural_key_refs() -> None: """``SyncAccountsResultRow.brand`` and ``operator`` are required on the wire. For natural-key refs we echo them back so the buyer can correlate the rejection to their request entry.""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) rows = asyncio.run( store.upsert([_by_natural_key("acme.com", "acme.com")], ctx=ResolveContext()) ) @@ -228,7 +251,7 @@ def test_sync_governance_denies_every_entry() -> None: roster-backed store — the adopter doesn't model buyer-supplied governance bindings. Per-entry rejection (not operation-level) so a multi-account batch sees explicit rejection per row.""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) entries = [ SyncGovernanceEntry( account=_by_id("acct_alpha"), @@ -251,7 +274,7 @@ def test_sync_governance_denies_every_entry() -> None: def test_sync_governance_echoes_account_ref() -> None: """``SyncGovernanceResultRow.account`` echoes the request ref so the buyer can correlate the rejection.""" - store = create_roster_account_store(roster=_make_roster()) + store = create_roster_account_store(roster=_make_roster(), authorize=_allow_all) ref = _by_id("acct_alpha") rows = asyncio.run( store.sync_governance( @@ -276,7 +299,7 @@ def test_construction_rejects_key_id_mismatch() -> None: "acct_beta": Account(id="acct_WRONG", name="Beta"), } with pytest.raises(ValueError) as exc_info: - create_roster_account_store(roster=bad_roster) + create_roster_account_store(roster=bad_roster, authorize=_allow_all) msg = str(exc_info.value) assert "acct_beta" in msg assert "acct_WRONG" in msg @@ -285,7 +308,7 @@ def test_construction_rejects_key_id_mismatch() -> None: def test_construction_accepts_empty_roster() -> None: """An empty roster is legal — adopter can ship an empty allowlist (every resolve misses, every list returns empty).""" - store = create_roster_account_store(roster={}) + store = create_roster_account_store(roster={}, authorize=_allow_all) assert store.resolution == "explicit" @@ -300,7 +323,7 @@ def test_external_mutation_does_not_leak_into_store() -> None: store's view — adopters who reuse the input dict for other purposes don't accidentally widen the allowlist.""" roster = _make_roster() - store = create_roster_account_store(roster=roster) + store = create_roster_account_store(roster=roster, authorize=_allow_all) # Buyer-side mutation: adopter clears their map after handing it # to the store. @@ -309,10 +332,10 @@ def test_external_mutation_does_not_leak_into_store() -> None: # Store still sees the original two entries; the injected attacker # entry is invisible. - listed = asyncio.run(store.list(ctx=ResolveContext())) + listed = asyncio.run(store.list(ctx=ResolveContext(auth_info=_AUTH))) ids = {a.id for a in listed} assert ids == {"acct_alpha", "acct_beta"} assert "acct_attacker" not in ids - attacker = asyncio.run(store.resolve(_by_id("acct_attacker"))) + attacker = asyncio.run(store.resolve(_by_id("acct_attacker"), auth_info=_AUTH)) assert attacker is None diff --git a/tests/test_tenant_store.py b/tests/test_tenant_store.py index 0f323723..900e84ba 100644 --- a/tests/test_tenant_store.py +++ b/tests/test_tenant_store.py @@ -301,7 +301,7 @@ def test_cross_tenant_entry_rejected_before_adopter_code(self) -> None: assert rows[0].action == "failed" assert rows[0].status == "rejected" assert rows[0].errors is not None - assert rows[0].errors[0]["code"] == "PERMISSION_DENIED" + assert rows[0].errors[0]["code"] == "ACCOUNT_NOT_FOUND" def test_unknown_ref_rejected_with_account_not_found(self) -> None: """ACCOUNT_NOT_FOUND is the right code for "ref points @@ -354,7 +354,7 @@ def test_mixed_batch_partitions_correctly(self) -> None: assert len(writes) == 1, "only the in-tenant entry should reach upsert_row" assert rows[0].action == "created" assert rows[1].errors is not None - assert rows[1].errors[0]["code"] == "PERMISSION_DENIED" + assert rows[1].errors[0]["code"] == "ACCOUNT_NOT_FOUND" assert rows[2].errors is not None assert rows[2].errors[0]["code"] == "ACCOUNT_NOT_FOUND" diff --git a/tests/test_upstream_for.py b/tests/test_upstream_for.py index 8e23ed3d..ae386352 100644 --- a/tests/test_upstream_for.py +++ b/tests/test_upstream_for.py @@ -24,7 +24,9 @@ from __future__ import annotations +import asyncio from typing import Any +from unittest.mock import AsyncMock import pytest @@ -235,6 +237,52 @@ def test_repeated_call_same_url_returns_same_client_instance() -> None: assert client_1 is client_2 +def test_repeated_default_no_auth_reuses_client() -> None: + """Omitting auth must not allocate a fresh cache identity per request.""" + platform = _Platform() + client_1 = platform.upstream_for(_ctx(mode="live")) + client_2 = platform.upstream_for(_ctx(mode="live")) + assert client_1 is client_2 + + +def test_default_headers_are_part_of_cache_identity() -> None: + """Tenant routing headers cannot bleed through a shared cached client.""" + platform = _Platform() + auth = StaticBearer(token="shared") + client_a = platform.upstream_for( + _ctx(mode="live"), auth=auth, default_headers={"X-Tenant": "tenant-a"} + ) + client_b = platform.upstream_for( + _ctx(mode="live"), auth=auth, default_headers={"X-Tenant": "tenant-b"} + ) + assert client_a is not client_b + assert client_a._default_headers == {"X-Tenant": "tenant-a"} + assert client_b._default_headers == {"X-Tenant": "tenant-b"} + + +def test_transport_options_are_part_of_cache_identity() -> None: + platform = _Platform() + auth = StaticBearer(token="shared") + default = platform.upstream_for(_ctx(mode="live"), auth=auth) + different_timeout = platform.upstream_for(_ctx(mode="live"), auth=auth, timeout=5.0) + different_404 = platform.upstream_for(_ctx(mode="live"), auth=auth, treat_404_as_none=False) + assert len({id(default), id(different_timeout), id(different_404)}) == 3 + + +@pytest.mark.asyncio +async def test_bounded_cache_closes_evicted_client() -> None: + platform = _Platform() + platform.upstream_client_cache_size = 1 + first = platform.upstream_for(_ctx(mode="live"), auth=StaticBearer(token="first")) + first.aclose = AsyncMock() # type: ignore[method-assign] + + platform.upstream_for(_ctx(mode="live"), auth=StaticBearer(token="second")) + await asyncio.sleep(0) + + first.aclose.assert_awaited_once() + assert len(platform._upstream_client_cache) == 1 + + def test_distinct_auth_strategies_get_distinct_clients() -> None: """Different auth instances ⇒ different clients. The auth is injected at construction; the framework can't swap it on a cached diff --git a/tests/test_upstream_helpers.py b/tests/test_upstream_helpers.py index 1ecb21cf..cef09a7e 100644 --- a/tests/test_upstream_helpers.py +++ b/tests/test_upstream_helpers.py @@ -344,6 +344,21 @@ async def test_500_raises_service_unavailable() -> None: await client.aclose() +@respx.mock +async def test_error_response_body_is_not_exposed() -> None: + secret = "upstream-secret-token" + respx.get(f"{BASE}/x").mock( + return_value=httpx.Response(500, text=f"database failed Authorization=Bearer {secret}") + ) + client = create_upstream_http_client(BASE) + with pytest.raises(AdcpError) as exc_info: + await client.get("/x") + + assert secret not in str(exc_info.value) + assert "database failed" not in str(exc_info.value) + await client.aclose() + + @respx.mock async def test_400_raises_invalid_request_correctable() -> None: respx.get(f"{BASE}/x").mock(return_value=httpx.Response(400, text="bad"))