fix(mcp): replay permit_wallet_mismatch 403 with signed denial receipt and details (SEL-482)

gateway/app/routers/mcp.py

@@ -1364,6 +1364,7 @@ async def _execute_registered_tool_inner(
                         wallet_id=wallet_id,
                         tool_name=tool_name,
                         key_id=auth.key_id,
+                        replay=replay,
                     )
                 await _raise_replayed_error(replay)
                 return replay.response_json
@@ -1635,6 +1636,7 @@ async def _execute_registered_tool_inner(
                 wallet_id=wallet_id,
                 tool_name=tool_name,
                 key_id=auth.key_id,
+                replay=replay,
             )
             await _raise_replayed_error(replay)
             return replay.response_json
@@ -1688,6 +1690,7 @@ async def _execute_registered_tool_inner(
                     reason_code=reason,
                     outcome="denied",
                     status_code=403,
+                    response_extra_data={"details": permit_validation.details},
                 )
             elif permit_validation.reason:
                 await _complete_governed_denial_idempotency(
@@ -4064,15 +4067,34 @@ async def _assert_governed_replay_access(
     wallet_id: str,
     tool_name: str,
     key_id: str | None,
+    replay: Any,
 ) -> None:
-    """Keep terminal replay bound to the permit's stable caller identity."""
+    """Keep terminal replay bound to its original caller and permit."""
     validation = await get_permit_service().validate_replay_access(
         permit_id=permit_id or "",
         wallet_id=wallet_id,
         tool_name=tool_name,
         key_id=key_id,
     )
     if not validation.allowed:
+        response = replay.response_json if replay is not None else None
+        receipt = response.get("receipt") if isinstance(response, dict) else None
+        if (
+            validation.reason == "permit_wallet_mismatch"
+            and replay is not None
+            and replay.status_code == 403
+            and response is not None
+            and response.get("error") == validation.reason
+            and isinstance(receipt, dict)
+            and receipt.get("wallet_id") == wallet_id
+            and receipt.get("key_id") == key_id
+            and receipt.get("permit_id") == permit_id
+            and receipt.get("tool") == tool_name
+            and receipt.get("outcome") == "denied"
+        ):
+            # The original denial is evidence for this wallet and key, even
+            # though the foreign permit cannot authorize an execution.
+            return
         raise ToolPermissionDenied(validation.reason or "permit_denied")
 
 
@@ -4258,7 +4280,12 @@ async def _raise_replayed_error(replay: Any) -> None:
     if isinstance(dispatch, dict):
         extra_data["dispatch"] = dispatch
     if replay.status_code == 403:
-        raise ToolPermissionDenied(reason, receipt=receipt)
+        details = replay.response_json.get("details")
+        raise ToolPermissionDenied(
+            reason,
+            receipt=receipt,
+            details=details if isinstance(details, dict) else None,
+        )
     raise GovernedToolError(
         reason,
         receipt=receipt,

gateway/tests/test_mcp_permit_denial_replay.py

@@ -0,0 +1,100 @@
+"""A stored cross-tenant permit denial replays its signed response."""
+
+from __future__ import annotations
+
+import json
+from uuid import uuid4
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import select
+
+from app.db.database import get_session_factory
+from app.db.models import IdempotencyRecordModel
+from app.main import app
+from app.services.receipts import get_receipt_service
+from tests.test_mcp_idempotency_key_validation import counted_tool  # noqa: F401
+from tests.test_trust_helpers import (
+    BOOTSTRAP_HEADERS,
+    create_tool_permit,
+    provision_agent_wallet,
+)
+
+
+@pytest.mark.anyio
+async def test_foreign_permit_denial_replays_original_403(
+    clean_database,
+    counted_tool,  # noqa: F811
+):
+    async with AsyncClient(
+        transport=ASGITransport(app=app), base_url="http://test"
+    ) as client:
+        tenant_a = await provision_agent_wallet(client)
+        tenant_b = await provision_agent_wallet(client)
+        permit_a = await create_tool_permit(
+            client,
+            wallet_id=tenant_a["agent_wallet_id"],
+            key_id=tenant_a["key_id"],
+            tool_name=counted_tool.name,
+        )
+        key = f"call_sel482_{uuid4().hex}"
+        request = {
+            "name": counted_tool.name,
+            "arguments": {"text": "one"},
+            "mcp_context": {
+                "wallet_id": tenant_b["agent_wallet_id"],
+                "permit_id": permit_a["permit_id"],
+                "idempotency_key": key,
+            },
+        }
+        headers = {**tenant_b["agent_headers"], "Idempotency-Key": key}
+        path = f"/mcp/tools/{counted_tool.name}/invoke"
+
+        first = await client.post(path, json=request, headers=headers)
+        async with get_session_factory()() as session:
+            record = (
+                await session.execute(
+                    select(IdempotencyRecordModel).where(
+                        IdempotencyRecordModel.idempotency_key == key
+                    )
+                )
+            ).scalar_one()
+        stored = json.loads(record.response_json)
+        assert set(stored) >= {"error", "receipt", "details"}
+        replay = await client.post(path, json=request, headers=headers)
+        other_key = await client.post(
+            "/v1/api-keys",
+            json={
+                "wallet_id": tenant_b["agent_wallet_id"],
+                "key_name": "replay-other-key",
+                "expires_in_days": 30,
+            },
+            headers=BOOTSTRAP_HEADERS,
+        )
+        assert other_key.status_code == 201
+        refused_other_key = await client.post(
+            path,
+            json=request,
+            headers={
+                "X-API-Key": other_key.json()["api_key"],
+                "Idempotency-Key": key,
+            },
+        )
+
+    assert first.status_code == replay.status_code == 403
+    detail = first.json()["detail"]
+    assert detail["error"] == "permit_wallet_mismatch"
+    assert detail["details"] == {"bound_to": "subject_wallet_id"}
+    receipt = detail["receipt"]
+    assert receipt["wallet_id"] == tenant_b["agent_wallet_id"]
+    assert receipt["outcome"] == "denied"
+    assert receipt["credits_charged"] == "0"
+    assert set(replay.json()["detail"]) == set(detail)
+    assert replay.content == first.content
+    assert refused_other_key.status_code == 403
+    assert refused_other_key.json()["detail"] == {"error": "permit_wallet_mismatch"}
+    assert tenant_a["agent_wallet_id"] not in first.text
+    assert counted_tool.effects == []
+
+    valid, reason, _ = await get_receipt_service().verify_receipt(receipt["receipt_id"])
+    assert (valid, reason) == (True, None)