jpmschweitzerandClaude Sonnet 5
0453d08206
The Office (workspace D-73) now signs every visitor in through Authentik with one OAuth client per consumer, and every token carries a signed tatlock_user claim. Caller gains consumer (X-Office-Consumer, the Authentik application slug) and tenant (X-Office-Tenant, the tatlock_user claim). For kind "office" both are required on every call, reads included — missing is 400 consumer_required/tenant_required, malformed is 400 consumer_invalid/tenant_invalid. Every other kind ignores both headers outright: tatlock and scheduler carry no tenant, and a "user" caller's tenant comes only from their own JWT's tatlock_user claim, never from a header — a header must not be able to put a non-Office caller into a tenant. X-Office-Client (informational) is retired in favour of X-Office-Consumer. book_events.client keeps its name (no migration this round) and now records the consumer instead. Book routes do not start filtering by tenant this round — that is core-api T-2. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
412 lines
16 KiB
Python
412 lines
16 KiB
Python
"""
|
|
Tests for require_caller (T-1, extended by T-4).
|
|
|
|
Builds a tiny standalone FastAPI app rather than reusing src.main's real
|
|
routes, so these are pure caller-resolution tests with no database and no
|
|
coupling to a particular domain's routing.
|
|
"""
|
|
import pytest
|
|
from fastapi import Depends, FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import patch
|
|
|
|
from src.domains.auth.callers import Caller, require_caller
|
|
from src.domains.auth.oidc import oidc_config
|
|
from src.shared.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
VALID_CONSUMER = "office-claude"
|
|
VALID_TENANT = "jpmschweitzer"
|
|
|
|
|
|
def _reset_service_keys():
|
|
settings.office_service_key = ""
|
|
settings.tatlock_service_key = ""
|
|
settings.scheduler_service_key = ""
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_caller_state():
|
|
"""Service keys and OIDC state are process-global; isolate every test."""
|
|
original_keys = (
|
|
settings.office_service_key,
|
|
settings.tatlock_service_key,
|
|
settings.scheduler_service_key,
|
|
)
|
|
original_oidc = (oidc_config.enabled, oidc_config.issuers, oidc_config.audiences)
|
|
_reset_service_keys()
|
|
yield
|
|
(
|
|
settings.office_service_key,
|
|
settings.tatlock_service_key,
|
|
settings.scheduler_service_key,
|
|
) = original_keys
|
|
oidc_config.enabled, oidc_config.issuers, oidc_config.audiences = original_oidc
|
|
|
|
|
|
# Named with a leading underscore so pytest does not try to collect it as
|
|
# a test module attribute (it previously warned: "cannot collect 'test_app'
|
|
# because it is not a function").
|
|
_caller_app = FastAPI()
|
|
|
|
|
|
@_caller_app.get("/only-office")
|
|
async def only_office(caller: Caller = Depends(require_caller("office"))):
|
|
return {
|
|
"kind": caller.kind,
|
|
"identity": caller.identity,
|
|
"consumer": caller.consumer,
|
|
"tenant": caller.tenant,
|
|
}
|
|
|
|
|
|
@_caller_app.get("/office-tatlock-user")
|
|
async def office_tatlock_user(
|
|
caller: Caller = Depends(require_caller("office", "tatlock", "user")),
|
|
):
|
|
return {"kind": caller.kind, "identity": caller.identity, "tenant": caller.tenant}
|
|
|
|
|
|
@_caller_app.post("/office-user-write")
|
|
async def office_user_write(
|
|
caller: Caller = Depends(require_caller("office", "user", identity_required=True)),
|
|
):
|
|
return {"kind": caller.kind, "identity": caller.identity}
|
|
|
|
|
|
@_caller_app.get("/all-kinds")
|
|
async def all_kinds(
|
|
caller: Caller = Depends(require_caller("office", "tatlock", "scheduler", "user")),
|
|
):
|
|
return {"kind": caller.kind, "identity": caller.identity, "tenant": caller.tenant}
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return TestClient(_caller_app)
|
|
|
|
|
|
class TestNoAuth:
|
|
def test_no_token_defaults_to_local_user_when_oidc_disabled(self, client):
|
|
"""Matches existing housekeeping/dashboard tests: OIDC off -> local user."""
|
|
response = client.get("/office-tatlock-user")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "user", "identity": "local", "tenant": None}
|
|
|
|
def test_no_token_401_when_oidc_enabled(self, client):
|
|
oidc_config.configure(enabled=True, issuers=[], audiences=[])
|
|
response = client.get("/office-tatlock-user")
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestOfficeKey:
|
|
def test_office_key_with_identity_resolves(self, client):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Identity": "jeroen",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "office", "identity": "jeroen", "tenant": VALID_TENANT}
|
|
|
|
def test_office_key_consumer_and_tenant_are_set_on_caller(self, client):
|
|
"""Dossier verification: both present -> Caller.consumer and .tenant set."""
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/only-office",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Identity": "jeroen",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"kind": "office",
|
|
"identity": "jeroen",
|
|
"consumer": VALID_CONSUMER,
|
|
"tenant": VALID_TENANT,
|
|
}
|
|
|
|
def test_office_key_x_office_client_header_is_ignored(self, client):
|
|
"""X-Office-Client is retired (T-4) in favour of X-Office-Consumer."""
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/only-office",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Identity": "jeroen",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
"X-Office-Client": "claude-desktop",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["consumer"] == VALID_CONSUMER # not "claude-desktop"
|
|
assert "client" not in body
|
|
|
|
def test_office_key_missing_identity_on_read_defaults_empty(self, client):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "office", "identity": "", "tenant": VALID_TENANT}
|
|
|
|
def test_office_key_missing_identity_on_write_is_400(self, client):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.post(
|
|
"/office-user-write",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "identity_required"
|
|
|
|
@pytest.mark.parametrize("identity", ["Jeroen", "group-home", "tatlock", "office", "book", "scheduler", "-abc", "a_b"])
|
|
def test_office_key_malformed_or_reserved_identity_is_400(self, client, identity):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.post(
|
|
"/office-user-write",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Identity": identity,
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "identity_invalid"
|
|
|
|
def test_office_key_valid_identity_on_write_succeeds(self, client):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.post(
|
|
"/office-user-write",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Identity": "jeroen",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "office", "identity": "jeroen"}
|
|
|
|
|
|
class TestOfficeConsumerAndTenant:
|
|
"""T-4: both are required on every office call, reads included — the
|
|
tenant scopes reads too. Missing -> 400 */_required; malformed -> 400
|
|
*/_invalid."""
|
|
|
|
def test_missing_consumer_is_400(self, client):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "consumer_required"
|
|
|
|
@pytest.mark.parametrize("consumer", ["Office-Claude", "office_claude", "-office"])
|
|
def test_malformed_consumer_is_400(self, client, consumer):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Consumer": consumer,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "consumer_invalid"
|
|
|
|
def test_missing_tenant_is_400_even_on_a_read(self, client):
|
|
"""Dossier verification: an Office read without a tenant is 400 —
|
|
/office-tatlock-user carries no identity_required, and this still
|
|
fails before reaching the handler."""
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "tenant_required"
|
|
|
|
@pytest.mark.parametrize("tenant", ["Jeroen", "-abc", "a b"])
|
|
def test_malformed_tenant_is_400(self, client, tenant):
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": tenant,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "tenant_invalid"
|
|
|
|
def test_tenant_allows_underscores(self, client):
|
|
"""tatlock's own tenant ids contain underscores (e.g. llm_tester)."""
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": "llm_tester",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["tenant"] == "llm_tester"
|
|
|
|
def test_missing_both_reports_consumer_first(self, client):
|
|
"""Lock down the check order: consumer before tenant when both are absent."""
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user", headers={"Authorization": "Bearer office-secret"}
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "consumer_required"
|
|
|
|
|
|
class TestTatlockAndSchedulerKeys:
|
|
def test_tatlock_key_resolves_and_ignores_identity_header(self, client):
|
|
settings.tatlock_service_key = "tatlock-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer tatlock-secret",
|
|
"X-Office-Identity": "jeroen",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "tatlock", "identity": "tatlock", "tenant": None}
|
|
|
|
def test_scheduler_key_forbidden_on_a_route_that_excludes_it(self, client):
|
|
settings.scheduler_service_key = "scheduler-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user", headers={"Authorization": "Bearer scheduler-secret"}
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"]["code"] == "forbidden"
|
|
|
|
def test_scheduler_key_resolves_and_ignores_consumer_and_tenant_headers(self, client):
|
|
"""Every kind but 'office' ignores X-Office-Consumer/X-Office-Tenant outright."""
|
|
settings.scheduler_service_key = "scheduler-secret"
|
|
response = client.get(
|
|
"/all-kinds",
|
|
headers={
|
|
"Authorization": "Bearer scheduler-secret",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "scheduler", "identity": "scheduler", "tenant": None}
|
|
|
|
|
|
class TestServiceKeyBypassesJWT:
|
|
def test_matched_service_key_never_calls_get_current_user(self, client):
|
|
"""Static keys are checked before any JWT parsing (dossier T-1 §1)."""
|
|
settings.office_service_key = "office-secret"
|
|
with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
|
|
mock_get_current_user.side_effect = AssertionError(
|
|
"get_current_user must not run when a service key matched"
|
|
)
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer office-secret",
|
|
"X-Office-Identity": "jeroen",
|
|
"X-Office-Consumer": VALID_CONSUMER,
|
|
"X-Office-Tenant": VALID_TENANT,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
mock_get_current_user.assert_not_called()
|
|
|
|
def test_unmatched_token_falls_through_to_get_current_user(self, client):
|
|
"""A wrong-looking token with no matching key must still reach the JWT path."""
|
|
settings.office_service_key = "office-secret"
|
|
response = client.get(
|
|
"/office-tatlock-user", headers={"Authorization": "Bearer not-the-office-key"}
|
|
)
|
|
# OIDC disabled -> get_current_user ignores the bad bearer and returns the local user.
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "user", "identity": "local", "tenant": None}
|
|
|
|
|
|
class TestUserTenant:
|
|
"""A JWT user's tenant comes only from their own token's `tatlock_user`
|
|
claim — never from a header (T-4). This is the tenant-spoofing guard."""
|
|
|
|
def test_user_tenant_comes_from_jwt_claim(self, client):
|
|
with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
|
|
mock_get_current_user.return_value = {
|
|
"preferred_username": "jpmschweitzer",
|
|
"tatlock_user": "jpmschweitzer",
|
|
}
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={"Authorization": "Bearer some-jwt"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "user", "identity": "jpmschweitzer", "tenant": "jpmschweitzer"}
|
|
|
|
def test_user_tenant_ignores_x_office_tenant_header_spoofing_attempt(self, client):
|
|
"""The header must not be able to put a non-Office caller into a tenant:
|
|
a JWT carrying the real claim wins over a forged header every time."""
|
|
with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
|
|
mock_get_current_user.return_value = {
|
|
"preferred_username": "jpmschweitzer",
|
|
"tatlock_user": "jpmschweitzer",
|
|
}
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer some-jwt",
|
|
"X-Office-Tenant": "someone-elses-tenant",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "user", "identity": "jpmschweitzer", "tenant": "jpmschweitzer"}
|
|
|
|
def test_user_tenant_is_none_when_jwt_carries_no_claim_even_with_header_sent(self, client):
|
|
with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
|
|
mock_get_current_user.return_value = {"preferred_username": "jpmschweitzer"}
|
|
response = client.get(
|
|
"/office-tatlock-user",
|
|
headers={
|
|
"Authorization": "Bearer some-jwt",
|
|
"X-Office-Tenant": "someone-elses-tenant",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"kind": "user", "identity": "jpmschweitzer", "tenant": None}
|