jpmschweitzer/core-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat(auth): T-4 — the caller carries consumer and tenant from the Office

Browse Source 
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>
This commit is contained in:
jpmschweitzer
2026-10-19 13:43:36 +02:00
co-authored by Claude Sonnet 5
parent 1da057bea6
commit 0453d08206
4 changed files with 414 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/domains/auth/callers.py
+65 -12
View File
@@ -15,6 +15,25 @@ Each static key is matched with hmac.compare_digest, and an unconfigured
(empty-string) key never matches — checked explicitly rather than trusting
compare_digest("", "") to be the right thing to compare, which would make an
empty configured key match an empty bearer.
core-api T-4 (workspace D-73): the Office now signs every visitor in through
Authentik with one OAuth client per consumer, and every token carries a
signed `tatlock_user` claim. The Office verifies the token and forwards the
result in two headers, trusted only alongside the office service key:
- `X-Office-Consumer` — the Authentik application slug the token came from
(`office-claude`, `office-gemini`, ...).
- `X-Office-Tenant` — the token's `tatlock_user` claim (the tatlock tenant).
For kind "office" both are required on **every** call, reads included — the
tenant scopes reads too. For every other kind they are ignored outright:
"tatlock" and "scheduler" carry no tenant (`None`); a "user" caller gets its
tenant from its own JWT's `tatlock_user` claim if it has one, else `None` —
never from a header, so a header can never put a non-Office caller into a
tenant.
`X-Office-Client` (informational, pre-T-4) is retired in favour of
`X-Office-Consumer`; nothing reads it any more.
"""
import hmac
import re
@@ -39,26 +58,54 @@ RESERVED_IDENTITIES = {"book", "office", "scheduler", "tatlock"}
# separately below, since the character class alone can't exclude a prefix).
_IDENTITY_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$")
# X-Office-Consumer: the Authentik application slug, same shape as an identity.
_CONSUMER_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$")
# X-Office-Tenant: tatlock's own tenant ids contain underscores (e.g.
# `llm_tester`), so the identity/consumer shape doesn't fit.
_TENANT_RE = re.compile(r"^[a-z0-9_][a-z0-9_.-]{0,62}$")
@dataclass(frozen=True)
class Caller:
"""Who is calling, resolved once per request."""
kind: CallerKind
identity: str # stamped on every write and event
client: Optional[str] # X-Office-Client, informational
identity: str # stamped on every write and event
consumer: Optional[str] # X-Office-Consumer; office only, else None
tenant: Optional[str] # the tatlock tenant; office header or a user's own JWT claim
def _identity_error(code: str) -> HTTPException:
def _bad_request(code: str) -> HTTPException:
return HTTPException(status_code=400, detail={"code": code})
def _validate_identity(identity: str) -> None:
"""Raise 400 identity_invalid for a malformed or reserved identity."""
if not _IDENTITY_RE.match(identity):
raise _identity_error("identity_invalid")
raise _bad_request("identity_invalid")
if identity.startswith("group-") or identity in RESERVED_IDENTITIES:
raise _identity_error("identity_invalid")
raise _bad_request("identity_invalid")
def _require_office_consumer(request: Request) -> str:
"""X-Office-Consumer is required on every office call; raise if absent/malformed."""
value = request.headers.get("X-Office-Consumer")
if not value:
raise _bad_request("consumer_required")
if not _CONSUMER_RE.match(value):
raise _bad_request("consumer_invalid")
return value
def _require_office_tenant(request: Request) -> str:
"""X-Office-Tenant is required on every office call, reads included."""
value = request.headers.get("X-Office-Tenant")
if not value:
raise _bad_request("tenant_required")
if not _TENANT_RE.match(value):
raise _bad_request("tenant_invalid")
return value
def require_caller(*allowed: CallerKind, identity_required: bool = False) -> Callable:
@@ -70,7 +117,8 @@ def require_caller(*allowed: CallerKind, identity_required: bool = False) -> Cal
allowed: caller kinds permitted on this route.
identity_required: for an "office" caller, whether X-Office-Identity
must be present (true on writes). Other kinds carry a fixed
identity and ignore this.
identity and ignore this. Unrelated to consumer/tenant, which an
"office" caller must carry on every call regardless of this flag.
"""
async def dependency(
@@ -79,7 +127,6 @@ def require_caller(*allowed: CallerKind, identity_required: bool = False) -> Cal
) -> Caller:
settings = get_settings()
token = credentials.credentials if credentials else None
client = request.headers.get("X-Office-Client")
caller: Optional[Caller] = None
@@ -88,36 +135,42 @@ def require_caller(*allowed: CallerKind, identity_required: bool = False) -> Cal
and settings.office_service_key
and hmac.compare_digest(token, settings.office_service_key)
):
consumer = _require_office_consumer(request)
tenant = _require_office_tenant(request)
identity_header = request.headers.get("X-Office-Identity")
if not identity_header:
if identity_required:
raise _identity_error("identity_required")
raise _bad_request("identity_required")
identity = ""
else:
_validate_identity(identity_header)
identity = identity_header
caller = Caller(kind="office", identity=identity, client=client)
caller = Caller(kind="office", identity=identity, consumer=consumer, tenant=tenant)
elif (
token
and settings.tatlock_service_key
and hmac.compare_digest(token, settings.tatlock_service_key)
):
caller = Caller(kind="tatlock", identity="tatlock", client=client)
caller = Caller(kind="tatlock", identity="tatlock", consumer=None, tenant=None)
elif (
token
and settings.scheduler_service_key
and hmac.compare_digest(token, settings.scheduler_service_key)
):
caller = Caller(kind="scheduler", identity="scheduler", client=client)
caller = Caller(kind="scheduler", identity="scheduler", consumer=None, tenant=None)
else:
# No static key matched (or none configured) — fall through to
# the existing OIDC/JWT path. A static key never reaches here.
user = await get_current_user(credentials)
identity = (user or {}).get("preferred_username") or "user"
caller = Caller(kind="user", identity=identity, client=client)
# The tenant comes only from the JWT's own claim — never from a
# header, which would let any caller spoof another tenant.
tenant = (user or {}).get("tatlock_user") or None
caller = Caller(kind="user", identity=identity, consumer=None, tenant=tenant)
if caller.kind not in allowed:
logger.warning(f"Caller kind '{caller.kind}' forbidden on this route (allowed: {allowed})")
src/domains/book/service.py
+1 -1
View File
@@ -70,7 +70,7 @@ class BookService:
event = BookEvent(
record_id=record_id,
identity=caller.identity,
client=caller.client,
client=caller.consumer, # column name unchanged (no migration this round); T-4 retires X-Office-Client and records the consumer here instead
action=action,
diff=diff,
)
tests/test_book.py
+109 -8
View File
@@ -60,11 +60,21 @@ def client(clean_book_tables, test_database_url):
app.dependency_overrides.pop(get_async_session, None)
def office_headers(identity: str = "jeroen", client_name: str | None = None) -> dict:
headers = {"Authorization": f"Bearer {OFFICE_KEY}", "X-Office-Identity": identity}
if client_name:
headers["X-Office-Client"] = client_name
return headers
VALID_CONSUMER = "office-claude"
VALID_TENANT = "jpmschweitzer"
def office_headers(
identity: str = "jeroen",
consumer: str = VALID_CONSUMER,
tenant: str = VALID_TENANT,
) -> dict:
return {
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Identity": identity,
"X-Office-Consumer": consumer,
"X-Office-Tenant": tenant,
}
class TestCallerMatrixOnBookRoutes:
@@ -82,11 +92,78 @@ class TestCallerMatrixOnBookRoutes:
response = client.post(
"/book/records",
json={"body": "hi"},
headers={"Authorization": f"Bearer {OFFICE_KEY}"},
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Consumer": VALID_CONSUMER,
"X-Office-Tenant": VALID_TENANT,
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "identity_required"
def test_office_write_without_consumer_is_400(self, client):
response = client.post(
"/book/records",
json={"body": "hi"},
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Identity": "jeroen",
"X-Office-Tenant": VALID_TENANT,
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "consumer_required"
def test_office_write_without_tenant_is_400(self, client):
response = client.post(
"/book/records",
json={"body": "hi"},
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Identity": "jeroen",
"X-Office-Consumer": VALID_CONSUMER,
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "tenant_required"
def test_office_read_without_tenant_is_400(self, client):
"""The tenant scopes reads too — a list call with no tenant is 400,
even though list carries no identity_required."""
response = client.get(
"/book/records",
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Consumer": VALID_CONSUMER,
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "tenant_required"
def test_office_malformed_consumer_is_400(self, client):
response = client.get(
"/book/records",
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Consumer": "Not Valid",
"X-Office-Tenant": VALID_TENANT,
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "consumer_invalid"
def test_office_malformed_tenant_is_400(self, client):
response = client.get(
"/book/records",
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Consumer": VALID_CONSUMER,
"X-Office-Tenant": "Not Valid",
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "tenant_invalid"
def test_scheduler_key_is_403_on_book_records(self, client):
settings.scheduler_service_key = "test-scheduler-key"
try:
@@ -104,7 +181,12 @@ class TestCallerMatrixOnBookRoutes:
response = client.post(
"/book/records",
json={"body": "hi"},
headers={"Authorization": f"Bearer {OFFICE_KEY}", "X-Office-Identity": "Not Valid"},
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Identity": "Not Valid",
"X-Office-Consumer": VALID_CONSUMER,
"X-Office-Tenant": VALID_TENANT,
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "identity_invalid"
@@ -238,7 +320,11 @@ class TestChange:
response = client.patch(
f"/book/records/{record['id']}",
json={"priority": 1},
headers={"Authorization": f"Bearer {OFFICE_KEY}"},
headers={
"Authorization": f"Bearer {OFFICE_KEY}",
"X-Office-Consumer": VALID_CONSUMER,
"X-Office-Tenant": VALID_TENANT,
},
)
assert response.status_code == 400
assert response.json()["detail"]["code"] == "identity_required"
@@ -384,3 +470,18 @@ class TestHistory:
history = client.get(f"/book/records/{record['id']}/history", headers=office_headers()).json()
assert len(history["events"]) == 1
assert history["events"][0]["action"] == "add"
def test_a_write_s_event_records_the_consumer(self, client):
"""T-4: X-Office-Client is retired; book_events.client (no migration
this round) now carries the consumer instead."""
record = client.post(
"/book/records",
json={"body": "x"},
headers=office_headers(consumer="office-gemini"),
).json()
history = client.get(
f"/book/records/{record['id']}/history", headers=office_headers()
).json()
assert history["events"][0]["action"] == "add"
assert history["events"][0]["client"] == "office-gemini"
tests/test_callers.py
+239 -17
View File
@@ -1,5 +1,5 @@
"""
Tests for require_caller (T-1).
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
@@ -16,6 +16,9 @@ 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 = ""
@@ -50,14 +53,19 @@ _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, "client": caller.client}
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}
return {"kind": caller.kind, "identity": caller.identity, "tenant": caller.tenant}
@_caller_app.post("/office-user-write")
@@ -67,6 +75,13 @@ async def office_user_write(
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)
@@ -77,7 +92,7 @@ class TestNoAuth:
"""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"}
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=[])
@@ -90,36 +105,76 @@ class TestOfficeKey:
settings.office_service_key = "office-secret"
response = client.get(
"/office-tatlock-user",
headers={"Authorization": "Bearer office-secret", "X-Office-Identity": "jeroen"},
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"}
assert response.json() == {"kind": "office", "identity": "jeroen", "tenant": VALID_TENANT}
def test_office_key_client_header_is_informational(self, client):
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
assert response.json()["client"] == "claude-desktop"
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"}
"/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": ""}
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"}
"/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"
@@ -129,7 +184,12 @@ class TestOfficeKey:
settings.office_service_key = "office-secret"
response = client.post(
"/office-user-write",
headers={"Authorization": "Bearer office-secret", "X-Office-Identity": identity},
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"
@@ -138,21 +198,115 @@ class TestOfficeKey:
settings.office_service_key = "office-secret"
response = client.post(
"/office-user-write",
headers={"Authorization": "Bearer office-secret", "X-Office-Identity": "jeroen"},
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"},
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"}
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"
@@ -162,6 +316,20 @@ class TestTatlockAndSchedulerKeys:
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):
@@ -173,7 +341,12 @@ class TestServiceKeyBypassesJWT:
)
response = client.get(
"/office-tatlock-user",
headers={"Authorization": "Bearer office-secret", "X-Office-Identity": "jeroen"},
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()
@@ -186,4 +359,53 @@ class TestServiceKeyBypassesJWT:
)
# 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"}
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}