jpmschweitzerandClaude Sonnet 5
99ee994023
A tenant has been an unchecked string since T-4/T-7: the Office's X-Office-Tenant header or a user JWT's tatlock_user claim, with Book and Scribe each running their own _require_tenant that checked only presence. A typo'd claim opened a fresh, empty namespace rather than being refused. src/domains/tenants/ (controller/models/schemas/service) adds tenants and tenant_consumers (migration 007, down_revision="006"): tenants.id is the slug a tatlock tenant carries, kind (household|person|test) and status (active|suspended) CHECK-constrained; tenant_consumers is the grant pairing a consumer to a tenant. Migration 007 seeds the two tenants in use today — jpmschweitzer (household, "Jeroen") and llm_tester (test, "Playground") — each granted to office-claude, the only Office consumer configured in Authentik. Every existing owner_tenant column (book_records, book_tenant_counters, scribe_addresses, scribe_messages, scribe_mailbox_entries — five, grepped and confirmed) now carries a foreign key to tenants.id, ON DELETE RESTRICT, at both the migration and the ORM model. TenantsService.require_tenant is the one enforcement point: no tenant is 400 tenant_required (unchanged); an unknown tenant is 403 tenant_unknown; a suspended one is 403 tenant_suspended; an Office caller whose consumer holds no grant in that tenant is 403 consumer_not_granted. One query — a LEFT JOIN against the grant — so presence, status and the grant are all read in a single round trip. Book's and Scribe's own _require_tenant now delegate to it, translating TenantError to their own exception type so every call site and each controller's single except clause keep working unchanged. Read-only this round: GET /tenants/me answers the caller's own tenant and its granted consumers (the cheap "am I allowed here" probe); GET /tenants answers every tenant for a tatlock/scheduler caller, or a list of one for a tenant-bearing caller — another tenant's existence is never revealed. No write routes; provisioning a tenant is workspace T-195. Tests: tests/test_tenants.py and tests/test_tenants_migration.py (new); a TestTenantRegistryEnforcement class added to both tests/test_book.py and tests/test_scribe.py proving all four cases (unknown, suspended, ungranted-consumer, and the control) on a live Book and Scribe route, not only on /tenants itself. tests/conftest.py gains a tenant_registry schema fixture and a suspended_tenant fixture (inserted and deleted per test; no FK references it); book_schema and scribe_schema each independently grant a second, test-environment-only consumer (office-gemini) in both seeded tenants, since test_book.py and test_scribe.py have used it as a stand-in second consumer since before this grant check existed — the migration itself still seeds only office-claude, matching the dossier. Mutation-checked: dropping the grant check fails the three consumer_not_granted tests (200 instead of 403); dropping the status check fails the three tenant_suspended tests; making Book bypass the helper fails its unknown-tenant test (200 instead of 403). All three restored and the full suite re-verified green (508 + 35 = 543 passed). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
621 lines
26 KiB
Python
621 lines
26 KiB
Python
"""
|
|
Tests for the Scribe domain (dossier core-api T-5 / D-66, D-68, D-73)
|
|
against `core_api_test`.
|
|
|
|
Follows tests/test_book.py's TestClient pattern: `get_async_session` is
|
|
overridden to the test database for every test, building a fresh
|
|
engine/session per call because TestClient drives the ASGI app on a
|
|
different event loop than pytest-asyncio's fixture setup.
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.domains.auth.callers import _IDENTITY_RE, RESERVED_IDENTITIES
|
|
from src.domains.scribe.models import ADDRESS_NAME_PATTERN, RESERVED_ADDRESS_NAMES
|
|
from src.main import app
|
|
from src.shared.config import get_settings
|
|
from src.shared.database import get_async_session
|
|
|
|
settings = get_settings()
|
|
|
|
OFFICE_KEY = "test-office-key"
|
|
TENANT = "jpmschweitzer"
|
|
OTHER_TENANT = "llm_tester"
|
|
|
|
|
|
def test_address_name_rule_matches_callers_identity_rule():
|
|
"""models.py deliberately duplicates callers.py's identity shape rather
|
|
than importing its private constants (see models.py's comment) — assert
|
|
the two never drift apart instead."""
|
|
assert ADDRESS_NAME_PATTERN == _IDENTITY_RE.pattern
|
|
assert RESERVED_ADDRESS_NAMES == RESERVED_IDENTITIES
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def office_key():
|
|
"""Every Scribe test authenticates as the office caller unless it says otherwise."""
|
|
original = settings.office_service_key
|
|
settings.office_service_key = OFFICE_KEY
|
|
yield
|
|
settings.office_service_key = original
|
|
|
|
|
|
@pytest.fixture
|
|
def client(clean_scribe_tables, test_database_url):
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
|
|
async def _override():
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
session_factory = async_sessionmaker(bind=engine, expire_on_commit=False)
|
|
async with session_factory() as session:
|
|
yield session
|
|
await engine.dispose()
|
|
|
|
app.dependency_overrides[get_async_session] = _override
|
|
yield TestClient(app)
|
|
app.dependency_overrides.pop(get_async_session, None)
|
|
|
|
|
|
def office_headers(identity: str, consumer: str, tenant: str = TENANT) -> dict:
|
|
return {
|
|
"Authorization": f"Bearer {OFFICE_KEY}",
|
|
"X-Office-Identity": identity,
|
|
"X-Office-Consumer": consumer,
|
|
"X-Office-Tenant": tenant,
|
|
}
|
|
|
|
|
|
def register(client, identity: str, consumer: str, tenant: str = TENANT):
|
|
return client.post(
|
|
"/scribe/addresses",
|
|
json={"name": identity},
|
|
headers=office_headers(identity, consumer, tenant),
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Caller matrix — a representative slice; the full matrix is test_callers.py
|
|
# and test_book.py's TestCallerMatrixOnBookRoutes already exercises
|
|
# require_caller itself end to end on a sibling domain.
|
|
# =============================================================================
|
|
|
|
|
|
class TestCallerMatrixOnScribeRoutes:
|
|
def test_no_auth_is_401_on_scribe_routes(self, client):
|
|
from src.domains.auth.oidc import oidc_config
|
|
|
|
oidc_config.configure(enabled=True, issuers=[], audiences=[])
|
|
try:
|
|
response = client.post("/scribe/addresses", json={"name": "jeroen"})
|
|
finally:
|
|
oidc_config.configure(enabled=False, issuers=[], audiences=[])
|
|
assert response.status_code == 401
|
|
|
|
def test_office_write_without_identity_is_400(self, client):
|
|
response = client.post(
|
|
"/scribe/addresses",
|
|
json={"name": "jeroen"},
|
|
headers={
|
|
"Authorization": f"Bearer {OFFICE_KEY}",
|
|
"X-Office-Consumer": "office-claude",
|
|
"X-Office-Tenant": TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "identity_required"
|
|
|
|
def test_office_write_without_tenant_is_400(self, client):
|
|
response = client.post(
|
|
"/scribe/addresses",
|
|
json={"name": "jeroen"},
|
|
headers={
|
|
"Authorization": f"Bearer {OFFICE_KEY}",
|
|
"X-Office-Identity": "jeroen",
|
|
"X-Office-Consumer": "office-claude",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "tenant_required"
|
|
|
|
def test_scheduler_key_is_403_on_scribe_addresses(self, client):
|
|
settings.scheduler_service_key = "test-scheduler-key"
|
|
try:
|
|
response = client.post(
|
|
"/scribe/addresses",
|
|
json={"name": "jeroen"},
|
|
headers={"Authorization": "Bearer test-scheduler-key"},
|
|
)
|
|
finally:
|
|
settings.scheduler_service_key = ""
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "forbidden"
|
|
|
|
|
|
# =============================================================================
|
|
# register / list addresses
|
|
# =============================================================================
|
|
|
|
|
|
class TestRegisterAndListAddresses:
|
|
def test_register_creates_address_owned_by_caller(self, client):
|
|
response = register(client, "jeroen", "office-claude")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["name"] == "jeroen"
|
|
assert data["owner_consumer"] == "office-claude"
|
|
assert data["created_by"] == "jeroen"
|
|
assert data["owned"] is True
|
|
|
|
def test_register_is_idempotent_for_the_same_consumer(self, client):
|
|
first = register(client, "jeroen", "office-claude")
|
|
second = register(client, "jeroen", "office-claude")
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
assert second.json()["owned"] is True
|
|
|
|
def test_register_by_another_consumer_is_409_address_taken(self, client):
|
|
register(client, "jeroen", "office-claude")
|
|
response = register(client, "jeroen", "office-gemini")
|
|
assert response.status_code == 409
|
|
assert response.json()["detail"]["code"] == "address_taken"
|
|
|
|
@pytest.mark.parametrize("name", ["book", "office", "scheduler", "tatlock", "group-home"])
|
|
def test_register_reserved_or_group_name_is_400(self, client, name):
|
|
"""The registering identity ("jeroen") is valid and distinct from
|
|
the name being registered, so this exercises the service's own
|
|
reserved/group- check, not the caller header's identity check."""
|
|
response = client.post(
|
|
"/scribe/addresses",
|
|
json={"name": name},
|
|
headers=office_headers("jeroen", "office-claude"),
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "address_invalid"
|
|
|
|
@pytest.mark.parametrize("name", ["Jeroen", "-abc", "a_b", ""])
|
|
def test_register_malformed_name_is_422(self, client, name):
|
|
"""Shape alone is a pydantic concern (422), not a ScribeError (400)."""
|
|
response = client.post(
|
|
"/scribe/addresses",
|
|
json={"name": name},
|
|
headers=office_headers("jeroen", "office-claude"),
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
def test_list_addresses_marks_ownership(self, client):
|
|
register(client, "jeroen", "office-claude")
|
|
register(client, "tower-control", "office-gemini")
|
|
|
|
as_claude = client.get(
|
|
"/scribe/addresses", headers=office_headers("jeroen", "office-claude")
|
|
).json()
|
|
owned = {a["name"]: a["owned"] for a in as_claude["addresses"]}
|
|
assert owned == {"jeroen": True, "tower-control": False}
|
|
|
|
def test_list_addresses_is_scoped_to_the_callers_tenant(self, client):
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "jeroen", "office-claude", tenant=OTHER_TENANT)
|
|
|
|
as_other_tenant = client.get(
|
|
"/scribe/addresses", headers=office_headers("jeroen", "office-claude", OTHER_TENANT)
|
|
).json()
|
|
assert len(as_other_tenant["addresses"]) == 1
|
|
|
|
|
|
# =============================================================================
|
|
# The flow (dossier verification #2): register, send, mailbox, mark,
|
|
# history, find, from shows identity and consumer.
|
|
# =============================================================================
|
|
|
|
|
|
class TestMessagingFlow:
|
|
def test_full_flow(self, client):
|
|
register(client, "jeroen", "office-claude")
|
|
register(client, "tower-control", "office-gemini")
|
|
|
|
sent = client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "subject": "status", "body": "all green"},
|
|
headers=office_headers("jeroen", "office-claude"),
|
|
)
|
|
assert sent.status_code == 201
|
|
message = sent.json()
|
|
assert message["from_identity"] == "jeroen"
|
|
assert message["from_consumer"] == "office-claude"
|
|
assert message["topics"] == ["tower-control"]
|
|
|
|
mailbox = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
|
|
).json()
|
|
assert mailbox["total"] == 1
|
|
entry = mailbox["entries"][0]
|
|
assert entry["read_at"] is None
|
|
assert entry["message"]["id"] == message["id"]
|
|
assert entry["message"]["body"] == "all green"
|
|
|
|
marked = client.patch(
|
|
f"/scribe/mailbox/{entry['id']}",
|
|
json={"read": True},
|
|
headers=office_headers("tower-control", "office-gemini"),
|
|
)
|
|
assert marked.status_code == 200
|
|
assert marked.json()["read_at"] is not None
|
|
|
|
mailbox_after = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
|
|
).json()
|
|
assert mailbox_after["entries"][0]["read_at"] is not None
|
|
|
|
history = client.get(
|
|
"/scribe/addresses/tower-control/history",
|
|
headers=office_headers("tower-control", "office-gemini"),
|
|
).json()
|
|
assert history["address"] == "tower-control"
|
|
assert len(history["messages"]) == 1
|
|
assert history["messages"][0]["id"] == message["id"]
|
|
|
|
found = client.get(
|
|
"/scribe/messages",
|
|
params={"q": "green"},
|
|
headers=office_headers("tower-control", "office-gemini"),
|
|
).json()
|
|
assert found["total"] == 1
|
|
assert found["entries"][0]["message"]["id"] == message["id"]
|
|
|
|
def test_history_reading_never_marks_or_consumes(self, client):
|
|
register(client, "jeroen", "office-claude")
|
|
register(client, "tower-control", "office-gemini")
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "hi"},
|
|
headers=office_headers("jeroen", "office-claude"),
|
|
)
|
|
|
|
client.get(
|
|
"/scribe/addresses/tower-control/history",
|
|
headers=office_headers("tower-control", "office-gemini"),
|
|
)
|
|
mailbox = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
|
|
).json()
|
|
assert mailbox["entries"][0]["read_at"] is None
|
|
|
|
def test_mailbox_lists_unread_first(self, client):
|
|
register(client, "jeroen", "office-claude")
|
|
register(client, "tower-control", "office-gemini")
|
|
for body in ("first", "second", "third"):
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": body},
|
|
headers=office_headers("jeroen", "office-claude"),
|
|
)
|
|
|
|
mailbox = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
|
|
).json()
|
|
first_entry_id = mailbox["entries"][0]["id"]
|
|
client.patch(
|
|
f"/scribe/mailbox/{first_entry_id}",
|
|
json={"read": True},
|
|
headers=office_headers("tower-control", "office-gemini"),
|
|
)
|
|
|
|
mailbox_after = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
|
|
).json()
|
|
read_states = [e["read_at"] is None for e in mailbox_after["entries"]]
|
|
# unread (True) entries sort before the one read (False) entry
|
|
assert read_states == sorted(read_states, reverse=True)
|
|
|
|
def test_tag_then_untag_mailbox_entry(self, client):
|
|
register(client, "jeroen", "office-claude")
|
|
register(client, "tower-control", "office-gemini")
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "hi"},
|
|
headers=office_headers("jeroen", "office-claude"),
|
|
)
|
|
entry_id = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
|
|
).json()["entries"][0]["id"]
|
|
|
|
tagged = client.patch(
|
|
f"/scribe/mailbox/{entry_id}",
|
|
json={"add_tags": ["urgent"]},
|
|
headers=office_headers("tower-control", "office-gemini"),
|
|
)
|
|
assert tagged.json()["tags"] == ["urgent"]
|
|
|
|
untagged = client.patch(
|
|
f"/scribe/mailbox/{entry_id}",
|
|
json={"remove_tags": ["urgent"]},
|
|
headers=office_headers("tower-control", "office-gemini"),
|
|
)
|
|
assert untagged.json()["tags"] == []
|
|
|
|
|
|
# =============================================================================
|
|
# send validation
|
|
# =============================================================================
|
|
|
|
|
|
class TestSendValidation:
|
|
def test_send_to_unknown_address_is_404_naming_missing_ones(self, client):
|
|
register(client, "jeroen", "office-claude")
|
|
response = client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["nobody", "tower-control"], "body": "hi"},
|
|
headers=office_headers("jeroen", "office-claude"),
|
|
)
|
|
assert response.status_code == 404
|
|
detail = response.json()["detail"]
|
|
assert detail["code"] == "unknown_address"
|
|
assert set(detail["names"]) == {"nobody", "tower-control"}
|
|
|
|
def test_send_from_identity_not_registered_is_403(self, client):
|
|
register(client, "tower-control", "office-gemini")
|
|
response = client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "hi"},
|
|
headers=office_headers("jeroen", "office-claude"), # jeroen never registered
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "identity_not_owned"
|
|
|
|
def test_cannot_send_as_an_address_another_consumer_owns(self, client):
|
|
"""dossier verification #3: a consumer cannot send *as* an address
|
|
another consumer owns."""
|
|
register(client, "jeroen", "office-claude")
|
|
register(client, "tower-control", "office-gemini")
|
|
response = client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["jeroen"], "body": "impersonating"},
|
|
# office-gemini claiming to BE jeroen, which office-claude owns
|
|
headers=office_headers("jeroen", "office-gemini"),
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "identity_not_owned"
|
|
|
|
def test_send_with_reply_to_in_another_tenant_is_404(self, client):
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
sent = client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "original"},
|
|
headers=office_headers("jeroen", "office-claude", tenant=TENANT),
|
|
).json()
|
|
|
|
register(client, "jeroen", "office-claude", tenant=OTHER_TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT)
|
|
reply = client.post(
|
|
"/scribe/messages",
|
|
json={
|
|
"to": ["tower-control"],
|
|
"body": "cross-tenant reply",
|
|
"reply_to": sent["id"],
|
|
},
|
|
headers=office_headers("jeroen", "office-claude", tenant=OTHER_TENANT),
|
|
)
|
|
assert reply.status_code == 404
|
|
assert reply.json()["detail"]["code"] == "not_found"
|
|
|
|
|
|
# =============================================================================
|
|
# Isolation (dossier verification #3): the hard tenant boundary, both
|
|
# directions — every cross-tenant answer is indistinguishable from "does
|
|
# not exist".
|
|
# =============================================================================
|
|
|
|
|
|
class TestTenantIsolation:
|
|
def test_llm_tester_cannot_list_jpmschweitzer_addresses(self, client):
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
as_other = client.get(
|
|
"/scribe/addresses", headers=office_headers("jeroen", "office-claude", OTHER_TENANT)
|
|
).json()
|
|
assert as_other["addresses"] == []
|
|
|
|
def test_llm_tester_cannot_send_to_a_jpmschweitzer_address(self, client):
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT) # TENANT only
|
|
register(client, "jeroen", "office-claude", tenant=OTHER_TENANT)
|
|
response = client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "leaking?"},
|
|
headers=office_headers("jeroen", "office-claude", OTHER_TENANT),
|
|
)
|
|
# From OTHER_TENANT's point of view, "tower-control" does not exist.
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"]["code"] == "unknown_address"
|
|
|
|
def test_llm_tester_cannot_read_mailbox_of_jpmschweitzer_address(self, client):
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "secret"},
|
|
headers=office_headers("jeroen", "office-claude", tenant=TENANT),
|
|
)
|
|
|
|
# llm_tester has never registered tower-control, so it cannot even
|
|
# bind the identity -> 403, indistinguishable from "no such address".
|
|
response = client.get(
|
|
"/scribe/mailbox",
|
|
headers=office_headers("tower-control", "office-gemini", OTHER_TENANT),
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "identity_not_owned"
|
|
|
|
def test_llm_tester_cannot_see_history_of_a_jpmschweitzer_address(self, client):
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "secret"},
|
|
headers=office_headers("jeroen", "office-claude", tenant=TENANT),
|
|
)
|
|
|
|
response = client.get(
|
|
"/scribe/addresses/tower-control/history",
|
|
headers=office_headers("jeroen", "office-claude", OTHER_TENANT),
|
|
)
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"]["code"] == "unknown_address"
|
|
|
|
def test_llm_tester_cannot_find_a_message_of_a_jpmschweitzer_address(self, client):
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "unique-secret-word"},
|
|
headers=office_headers("jeroen", "office-claude", tenant=TENANT),
|
|
)
|
|
|
|
response = client.get(
|
|
"/scribe/messages",
|
|
params={"q": "unique-secret-word"},
|
|
headers=office_headers("tower-control", "office-gemini", OTHER_TENANT),
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "identity_not_owned"
|
|
|
|
def test_mark_on_an_entry_in_another_tenant_is_404(self, client):
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "secret"},
|
|
headers=office_headers("jeroen", "office-claude", tenant=TENANT),
|
|
)
|
|
entry_id = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini", TENANT)
|
|
).json()["entries"][0]["id"]
|
|
|
|
register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT)
|
|
response = client.patch(
|
|
f"/scribe/mailbox/{entry_id}",
|
|
json={"read": True},
|
|
headers=office_headers("tower-control", "office-gemini", OTHER_TENANT),
|
|
)
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"]["code"] == "not_found"
|
|
|
|
def test_mark_on_someone_elses_entry_within_the_same_tenant_is_404(self, client):
|
|
"""Not a tenant crossing, but the same non-disclosure principle:
|
|
an entry addressed to another identity answers like it doesn't
|
|
exist, rather than confirming it belongs to someone else."""
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["tower-control"], "body": "secret"},
|
|
headers=office_headers("jeroen", "office-claude", tenant=TENANT),
|
|
)
|
|
entry_id = client.get(
|
|
"/scribe/mailbox", headers=office_headers("tower-control", "office-gemini", TENANT)
|
|
).json()["entries"][0]["id"]
|
|
|
|
response = client.patch(
|
|
f"/scribe/mailbox/{entry_id}",
|
|
json={"read": True},
|
|
headers=office_headers("jeroen", "office-claude", TENANT), # not tower-control
|
|
)
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"]["code"] == "not_found"
|
|
|
|
def test_mailbox_does_not_leak_entries_from_another_tenants_same_named_address(self, client):
|
|
"""The identity-ownership precheck alone is not the isolation
|
|
guarantee: a caller legitimately owns "jeroen" in their own tenant,
|
|
and a *different* "jeroen" (same name, different owner) exists in
|
|
another tenant with mail of its own. The mailbox query's own tenant
|
|
filter is what keeps that mail out — this is what mutation check #1
|
|
(dossier) targets."""
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["jeroen"], "body": "for the TENANT jeroen"},
|
|
headers=office_headers("tower-control", "office-gemini", tenant=TENANT),
|
|
)
|
|
|
|
register(client, "jeroen", "office-claude", tenant=OTHER_TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["jeroen"], "body": "for the OTHER_TENANT jeroen"},
|
|
headers=office_headers("tower-control", "office-gemini", tenant=OTHER_TENANT),
|
|
)
|
|
|
|
mailbox = client.get(
|
|
"/scribe/mailbox", headers=office_headers("jeroen", "office-claude", TENANT)
|
|
).json()
|
|
assert mailbox["total"] == 1
|
|
assert mailbox["entries"][0]["message"]["body"] == "for the TENANT jeroen"
|
|
|
|
def test_find_does_not_leak_messages_from_another_tenants_same_named_address(self, client):
|
|
"""Same concern as above, for `find` (mutation check #1's other half)."""
|
|
register(client, "jeroen", "office-claude", tenant=TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["jeroen"], "body": "shared-search-term in TENANT"},
|
|
headers=office_headers("tower-control", "office-gemini", tenant=TENANT),
|
|
)
|
|
|
|
register(client, "jeroen", "office-claude", tenant=OTHER_TENANT)
|
|
register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT)
|
|
client.post(
|
|
"/scribe/messages",
|
|
json={"to": ["jeroen"], "body": "shared-search-term in OTHER_TENANT"},
|
|
headers=office_headers("tower-control", "office-gemini", tenant=OTHER_TENANT),
|
|
)
|
|
|
|
found = client.get(
|
|
"/scribe/messages",
|
|
params={"q": "shared-search-term"},
|
|
headers=office_headers("jeroen", "office-claude", TENANT),
|
|
).json()
|
|
assert found["total"] == 1
|
|
assert found["entries"][0]["message"]["body"] == "shared-search-term in TENANT"
|
|
|
|
|
|
class TestTenantRegistryEnforcement:
|
|
"""core-api T-8 / workspace D-73: Scribe's own `_require_tenant` now
|
|
delegates to the tenant registry's one enforcement point, so a Scribe
|
|
route gains three refusal paths it did not have before, alongside the
|
|
control that a properly granted, active tenant still succeeds."""
|
|
|
|
def test_unknown_tenant_is_403(self, client):
|
|
response = client.get(
|
|
"/scribe/addresses",
|
|
headers=office_headers("jeroen", "office-claude", tenant="no-such-tenant"),
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "tenant_unknown"
|
|
|
|
def test_suspended_tenant_is_403(self, client, suspended_tenant):
|
|
response = client.get(
|
|
"/scribe/addresses",
|
|
headers=office_headers("jeroen", "office-claude", tenant=suspended_tenant),
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "tenant_suspended"
|
|
|
|
def test_consumer_without_a_grant_is_403(self, client):
|
|
response = client.get(
|
|
"/scribe/addresses",
|
|
headers=office_headers("jeroen", "office-ungranted", tenant=TENANT),
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "consumer_not_granted"
|
|
|
|
def test_granted_consumer_in_an_active_tenant_succeeds(self, client):
|
|
"""The control: office-claude, granted in jpmschweitzer (active),
|
|
still reaches the route."""
|
|
response = client.get(
|
|
"/scribe/addresses", headers=office_headers("jeroen", "office-claude", tenant=TENANT)
|
|
)
|
|
assert response.status_code == 200
|