diff --git a/src/domains/auth/__init__.py b/src/domains/auth/__init__.py index 9b1afa6..f0ec8f0 100644 --- a/src/domains/auth/__init__.py +++ b/src/domains/auth/__init__.py @@ -20,6 +20,7 @@ from src.domains.auth.oidc import ( ) from src.domains.auth.service import AuthService, get_auth_service from src.domains.auth.controller import auth_controller +from src.domains.auth.callers import Caller, require_caller from src.domains.auth.models import ( User, Role, @@ -50,6 +51,9 @@ __all__ = [ "get_auth_service", # Controller "auth_controller", + # Callers (T-1) + "Caller", + "require_caller", # Models "User", "Role", diff --git a/src/domains/auth/callers.py b/src/domains/auth/callers.py new file mode 100644 index 0000000..75e5514 --- /dev/null +++ b/src/domains/auth/callers.py @@ -0,0 +1,128 @@ +""" +Caller Authentication + +core-api has no service-key auth today: AuthService.validate_api_key is +never called in any route, api_keys stays empty, and `CORE_API_KEY` is read +by nothing. This module is the first thing that reads it. + +Resolution order, static keys before any JWT parsing: +1. Bearer == office_service_key -> kind "office" +2. Bearer == tatlock_service_key -> kind "tatlock" +3. Bearer == scheduler_service_key -> kind "scheduler" +4. Otherwise -> get_current_user() -> kind "user" + +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. +""" +import hmac +import re +from dataclasses import dataclass +from typing import Callable, Literal, Optional + +from fastapi import Depends, HTTPException, Request, Security +from fastapi.security import HTTPAuthorizationCredentials + +from src.domains.auth.oidc import get_current_user, security +from src.shared.config import get_settings +from src.shared.logging import get_logger + +logger = get_logger(__name__) + +CallerKind = Literal["office", "tatlock", "scheduler", "user"] + +# Reserved identities: the system nouns, never a caller's own identity. +RESERVED_IDENTITIES = {"book", "office", "scheduler", "tatlock"} + +# ^[a-z0-9][a-z0-9-]{0,62}$, and additionally not a "group-*" shape (checked +# 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}$") + + +@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 + + +def _identity_error(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") + if identity.startswith("group-") or identity in RESERVED_IDENTITIES: + raise _identity_error("identity_invalid") + + +def require_caller(*allowed: CallerKind, identity_required: bool = False) -> Callable: + """ + Build a FastAPI dependency that resolves the caller and enforces it is + one of `allowed`. + + Args: + 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. + """ + + async def dependency( + request: Request, + credentials: Optional[HTTPAuthorizationCredentials] = Security(security), + ) -> Caller: + settings = get_settings() + token = credentials.credentials if credentials else None + client = request.headers.get("X-Office-Client") + + caller: Optional[Caller] = None + + if ( + token + and settings.office_service_key + and hmac.compare_digest(token, settings.office_service_key) + ): + identity_header = request.headers.get("X-Office-Identity") + if not identity_header: + if identity_required: + raise _identity_error("identity_required") + identity = "" + else: + _validate_identity(identity_header) + identity = identity_header + caller = Caller(kind="office", identity=identity, client=client) + + elif ( + token + and settings.tatlock_service_key + and hmac.compare_digest(token, settings.tatlock_service_key) + ): + caller = Caller(kind="tatlock", identity="tatlock", client=client) + + elif ( + token + and settings.scheduler_service_key + and hmac.compare_digest(token, settings.scheduler_service_key) + ): + caller = Caller(kind="scheduler", identity="scheduler", client=client) + + 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) + + if caller.kind not in allowed: + logger.warning(f"Caller kind '{caller.kind}' forbidden on this route (allowed: {allowed})") + raise HTTPException(status_code=403, detail={"code": "forbidden"}) + + return caller + + return dependency diff --git a/src/shared/config.py b/src/shared/config.py index 8b60d06..1af96ae 100644 --- a/src/shared/config.py +++ b/src/shared/config.py @@ -104,6 +104,15 @@ class Settings(BaseSettings): authentik_username: str = "" authentik_password: str = "" + # Caller auth — static service keys (T-1). Empty by default so an unset + # key can never match an empty Bearer token (see src.domains.auth.callers). + office_service_key: str = "" + tatlock_service_key: str = "" + scheduler_service_key: str = "" + + # The Book (T-1) — local time for due/remind/recurrence handling. + book_timezone: str = "Europe/Amsterdam" + class Config: env_file = ".env" case_sensitive = False diff --git a/tests/test_callers.py b/tests/test_callers.py new file mode 100644 index 0000000..22c4bf2 --- /dev/null +++ b/tests/test_callers.py @@ -0,0 +1,189 @@ +""" +Tests for require_caller (T-1). + +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() + + +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, "client": caller.client} + + +@_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} + + +@_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} + + +@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"} + + 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"}, + ) + assert response.status_code == 200 + assert response.json() == {"kind": "office", "identity": "jeroen"} + + def test_office_key_client_header_is_informational(self, client): + settings.office_service_key = "office-secret" + response = client.get( + "/only-office", + headers={ + "Authorization": "Bearer office-secret", + "X-Office-Identity": "jeroen", + "X-Office-Client": "claude-desktop", + }, + ) + assert response.status_code == 200 + assert response.json()["client"] == "claude-desktop" + + 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"} + ) + assert response.status_code == 200 + assert response.json() == {"kind": "office", "identity": ""} + + 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"} + ) + 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}, + ) + 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"}, + ) + assert response.status_code == 200 + assert response.json() == {"kind": "office", "identity": "jeroen"} + + +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"}, + ) + assert response.status_code == 200 + assert response.json() == {"kind": "tatlock", "identity": "tatlock"} + + 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" + + +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"}, + ) + 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"}