tegwick
adb74d2443
- authz.py: WriteAuthorizer Protocol + DefaultDenyWriteAuthorizer. Every
write endpoint calls it before touching the store; denial maps to
403 write_denied via an exception handler.
- app.py: GET /tenants/{id}/roles (cache-read, key-cape) and
GET /tenants/{id}/roles/live (live-lookup, flex-auth) share one handler
that fails closed (503) on StoreUnavailableError -- deliberately made
identical rather than giving cache-read weaker guarantees than the task
strictly required. POST /tenants, /roles/grant, /roles/revoke, /plan --
all four gated by the WriteAuthorizer seam, domain/store errors mapped to
400/404/409 after authorization passes.
- store.py: new StoreUnavailableError for the fail-closed test double.
43 tests passing: default-deny on every write endpoint, an
_AllowAllAuthorizer test double proving the seam actually gates (full
create->grant->read->revoke->read->assign-plan lifecycle over real HTTP),
and a _BrokenStore double proving outage never looks like "zero roles".
Verified live over real HTTP, not just TestClient.
TEN-WP-0002 closed: all 7 tasks done, boundary-contract ownership checked
against the implementation with no drift found. Follow-ups recorded in the
closure note (real flex-auth WriteAuthorizer, key-cape wiring, guardrail
policy design, Binky as first real tenant record, durable persistence).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from tenant_engine.app import create_app
|
|
from tenant_engine.authz import WriteAuthorizer
|
|
from tenant_engine.store import InMemoryTenantStore
|
|
|
|
|
|
class _AllowAllAuthorizer(WriteAuthorizer):
|
|
def authorize(self, *, action: str, tenant_id: str, actor: str) -> None:
|
|
return None
|
|
|
|
|
|
def _client(*, allow: bool = False) -> TestClient:
|
|
store = InMemoryTenantStore()
|
|
authorizer = _AllowAllAuthorizer() if allow else None
|
|
return TestClient(create_app(store=store, authorizer=authorizer))
|
|
|
|
|
|
def test_create_tenant_denied_by_default() -> None:
|
|
client = _client()
|
|
response = client.post(
|
|
"/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["error_code"] == "write_denied"
|
|
|
|
|
|
def test_grant_role_denied_by_default() -> None:
|
|
client = _client()
|
|
response = client.post(
|
|
"/tenants/t-1/roles/grant",
|
|
json={
|
|
"grant_id": "g-1",
|
|
"role": "CUS",
|
|
"grant_reason": "manual_grant",
|
|
"granted_by": "ops",
|
|
"correlation_id": "corr-1",
|
|
"actor": "ops",
|
|
},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_revoke_role_denied_by_default() -> None:
|
|
client = _client()
|
|
response = client.post("/tenants/t-1/roles/revoke", json={"grant_id": "g-1", "actor": "ops"})
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_assign_plan_denied_by_default() -> None:
|
|
client = _client()
|
|
response = client.post("/tenants/t-1/plan", json={"plan_id": "plan-x", "actor": "ops"})
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_full_write_lifecycle_succeeds_when_authorizer_allows() -> None:
|
|
client = _client(allow=True)
|
|
|
|
created = client.post(
|
|
"/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
|
|
)
|
|
assert created.status_code == 201
|
|
assert created.json()["grouping"] == "friendly"
|
|
|
|
granted = client.post(
|
|
"/tenants/t-1/roles/grant",
|
|
json={
|
|
"grant_id": "g-1",
|
|
"role": "CUS",
|
|
"grant_reason": "manual_grant",
|
|
"granted_by": "ops",
|
|
"correlation_id": "corr-1",
|
|
"actor": "ops",
|
|
},
|
|
)
|
|
assert granted.status_code == 201
|
|
|
|
roles = client.get("/tenants/t-1/roles")
|
|
assert roles.json()["roles"] == ["CUS"]
|
|
|
|
revoked = client.post("/tenants/t-1/roles/revoke", json={"grant_id": "g-1", "actor": "ops"})
|
|
assert revoked.status_code == 200
|
|
|
|
roles_after = client.get("/tenants/t-1/roles")
|
|
assert roles_after.json()["roles"] == []
|
|
|
|
plan = client.post("/tenants/t-1/plan", json={"plan_id": "plan-x", "actor": "ops"})
|
|
assert plan.status_code == 200
|
|
assert plan.json()["plan_id"] == "plan-x"
|
|
|
|
|
|
def test_create_tenant_rejects_invalid_identifier_after_authorization() -> None:
|
|
client = _client(allow=True)
|
|
response = client.post(
|
|
"/tenants", json={"tenant_id": "t-1", "identifier": "tenant:unknown:binky", "actor": "ops"}
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_create_tenant_duplicate_is_409() -> None:
|
|
client = _client(allow=True)
|
|
client.post("/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"})
|
|
response = client.post(
|
|
"/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_grant_role_plan_assignment_without_plan_id_is_400() -> None:
|
|
client = _client(allow=True)
|
|
client.post("/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"})
|
|
|
|
response = client.post(
|
|
"/tenants/t-1/roles/grant",
|
|
json={
|
|
"grant_id": "g-1",
|
|
"role": "IAM",
|
|
"grant_reason": "plan_assignment",
|
|
"granted_by": "ops",
|
|
"correlation_id": "corr-1",
|
|
"actor": "ops",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_grant_role_unknown_tenant_is_404() -> None:
|
|
client = _client(allow=True)
|
|
response = client.post(
|
|
"/tenants/does-not-exist/roles/grant",
|
|
json={
|
|
"grant_id": "g-1",
|
|
"role": "CUS",
|
|
"grant_reason": "manual_grant",
|
|
"granted_by": "ops",
|
|
"correlation_id": "corr-1",
|
|
"actor": "ops",
|
|
},
|
|
)
|
|
assert response.status_code == 404
|