TEN-WP-0006-T05: cross-tenant denial and write-path outage conformance

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/test_api_guardrails.py

@@ -27,11 +27,34 @@ class _ScopedAuthorizer(WriteAuthorizer):
             raise WriteAuthorizationDeniedError(action, "not permitted")
 
 
+class _TenantScopedAuthorizer(WriteAuthorizer):
+    """Permits guardrail work on exactly one tenant.
+
+    Stands in for a flex-auth policy that scopes an operator to their own
+    tenant -- the case where a caller is authenticated and permitted in
+    general, but not for *this* tenant.
+    """
+
+    def __init__(self, permitted_tenant: str) -> None:
+        self._permitted = permitted_tenant
+
+    def authorize(self, *, action: str, tenant_id: str, actor: str) -> None:
+        if action == "tenant.create":
+            return
+        if tenant_id != self._permitted:
+            raise WriteAuthorizationDeniedError(action, "not permitted for this tenant")
+
+
 class _BrokenStore(InMemoryTenantStore):
     def get_tenant(self, tenant_id: str):
         raise StoreUnavailableError("connection to /var/lib/tenant-engine/tenant.db refused")
 
 
+class _BrokenWriteStore(InMemoryTenantStore):
+    def set_guardrail_override(self, **kwargs):
+        raise StoreUnavailableError("connection to /var/lib/tenant-engine/tenant.db refused")
+
+
 def make_client(authorizer=None, store=None) -> TestClient:
     app = create_app(
         store=store or InMemoryTenantStore(), authorizer=authorizer or _AllowAllAuthorizer()
@@ -276,6 +299,43 @@ def test_tightening_a_retired_tenants_guardrail_is_allowed(client):
     assert response.status_code == 200
 
 
+# --- Cross-tenant denial and write-path outage ----------------------------
+
+
+def test_a_caller_cannot_reach_across_tenants():
+    client = make_client(_TenantScopedAuthorizer("t-own"))
+    client.post(
+        "/tenants",
+        json={"tenant_id": "t-own", "identifier": "tenant:small:own", "actor": "ops"},
+    )
+    assert client.get("/tenants/t-own/guardrails", params={"actor": "ops"}).status_code == 200
+    assert read(client, actor="ops").status_code == 403
+    assert put(client).status_code == 403
+
+
+def test_a_cross_tenant_write_changes_nothing():
+    client = make_client(_TenantScopedAuthorizer("t-own"))
+    assert put(client).status_code == 403
+    # the store was never touched: the version is untouched
+    assert client.get("/tenants/t-1").json()["version"] == 1
+
+
+def test_a_store_outage_fails_closed_on_write():
+    client = make_client(store=_BrokenWriteStore())
+    response = put(client)
+    assert response.status_code == 503
+    assert response.json()["error_code"] == "tenant_authority_unavailable"
+    assert "tenant.db" not in response.text
+    assert response.json()["correlation_id"] == "corr-1"
+
+
+def test_errors_never_reflect_policy_internals():
+    client = make_client(_ScopedAuthorizer("tenant.create"))
+    body = put(client).json()
+    assert "tenant.db" not in str(body)
+    assert body["error_code"] == "write_denied"
+
+
 # --- Compatibility --------------------------------------------------------