fix(identity)!: enforce tenant and credential isolation

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"),
+    )

examples/v3_reference_seller/src/app.py

@@ -62,9 +62,14 @@
 from adcp.server import (
     SubdomainTenantMiddleware,
     ToolContext,
-    current_tenant,
 )
-from adcp.server.auth import BearerTokenAuth, Principal, auth_context_factory
+from adcp.server.auth import (
+    ROUTED_TENANT_METADATA_KEY,
+    BearerTokenAuth,
+    Principal,
+    auth_context_factory,
+    enforce_authenticated_tenant,
+)
 from adcp.validation import ValidationHookConfig
 from adcp.webhook_sender import WebhookSender
 from adcp.webhook_supervisor import InMemoryWebhookDeliverySupervisor
@@ -104,12 +109,12 @@ 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.
-        tenant = current_tenant()
-        if tenant is not None:
-            ctx = replace(ctx, tenant_id=tenant.id)
+        # ``auth_context_factory`` preserves both identities in metadata.
+        # Pin the business context to the routed host; the skill middleware
+        # below compares it with the authenticated identity from the token at
+        # a boundary where both MCP and A2A project AdcpError consistently.
+        if ROUTED_TENANT_METADATA_KEY in ctx.metadata:
+            ctx = replace(ctx, tenant_id=ctx.metadata[ROUTED_TENANT_METADATA_KEY])
 
         # Upgrade bearer-flow auth_info with a typed ApiKeyCredential
         # when the validator stashed the raw token in principal metadata.
@@ -151,6 +156,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,
@@ -366,6 +376,7 @@ def main() -> None:
         # registry with credential=None and returns PERMISSION_DENIED.
         auth=BearerTokenAuth(validate_token=_make_validate_token(token_map)),
         context_factory=_build_context_factory(),
+        middleware=[enforce_authenticated_tenant],
         asgi_middleware=[
             (SubdomainTenantMiddleware, {"router": router}),
         ],

examples/v3_reference_seller/src/buyer_registry.py

@@ -100,13 +100,23 @@ async def resolve_by_credential(
             key = credential.client_id
         async with self._sessionmaker() as session:
             result = await session.execute(
-                select(BuyerAgentRow).where(
+                select(BuyerAgentRow)
+                .where(
                     BuyerAgentRow.tenant_id == tenant.id,
                     BuyerAgentRow.api_key_id == key,
                 )
+                .limit(2)
             )
-            row = result.scalar_one_or_none()
-        return _row_to_agent(row) if row else None
+            rows = list(result.scalars().all())
+        if len(rows) > 1:
+            logger.error(
+                "ambiguous buyer credential within tenant; denying lookup "
+                "(tenant_id=%s, credential_kind=%s)",
+                tenant.id,
+                credential.kind,
+            )
+            return None
+        return _row_to_agent(rows[0]) if rows else None
 
 
 def _row_to_agent(row: BuyerAgentRow) -> BuyerAgent:

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)),
         ),
     )