"""SessionAuthenticationMiddleware: populates request.state.principal from a verified session cookie, and only ever adds one -- never rejects a request itself, matching get_current_session's own null-on-any-failure behavior.""" from __future__ import annotations import secrets from datetime import UTC, datetime, timedelta from uuid import UUID, uuid4 from fastapi import FastAPI, Request from fastapi.testclient import TestClient from ragcore.application.auth import SignedTokenCodec, encode_session from ragcore.domain.auth import Session, principal_id_for_subject from ragcore.domain.authorization import PrincipalType from ragcore.domain.control import Role from ragcore.web.auth import SessionAuthenticationMiddleware from ragcore.web.auth.dependencies import SESSION_COOKIE_NAME SECRET = b"0" * 32 def _session( *, subject: str = "alice", workspace_id: UUID | None = None, roles: frozenset[Role] = frozenset({Role.EDITOR}), issued_at: datetime | None = None, expires_at: datetime | None = None, ) -> Session: now = issued_at or datetime.now(UTC) return Session( id=uuid4(), subject=subject, workspace_id=workspace_id or uuid4(), issued_at=now, expires_at=expires_at or (now + timedelta(hours=1)), csrf_token=secrets.token_urlsafe(32), roles=roles, ) def _app(*, codec: SignedTokenCodec | None) -> FastAPI: app = FastAPI() @app.get("/whoami") async def whoami(request: Request) -> dict[str, object]: principal = getattr(request.state, "principal", None) if principal is None: return {"authenticated": False} return { "authenticated": True, "id": str(principal.id), "workspace_id": str(principal.workspace_id), "principal_type": principal.principal_type.value, "roles": sorted(role.value for role in principal.roles), } app.add_middleware(SessionAuthenticationMiddleware) if codec is not None: app.state.auth_session_codec = codec return app def test_valid_session_cookie_populates_a_matching_principal() -> None: codec = SignedTokenCodec(secret=SECRET) session = _session(subject="alice", roles=frozenset({Role.EDITOR, Role.VIEWER})) client = TestClient(_app(codec=codec)) client.cookies.set(SESSION_COOKIE_NAME, encode_session(session, codec=codec)) response = client.get("/whoami") assert response.status_code == 200 body = response.json() assert body["authenticated"] is True assert body["id"] == str(principal_id_for_subject("alice")) assert body["workspace_id"] == str(session.workspace_id) assert body["principal_type"] == PrincipalType.HUMAN.value assert body["roles"] == sorted(role.value for role in session.roles) def test_no_cookie_leaves_the_principal_unset() -> None: codec = SignedTokenCodec(secret=SECRET) client = TestClient(_app(codec=codec)) response = client.get("/whoami") assert response.json() == {"authenticated": False} def test_tampered_cookie_leaves_the_principal_unset_rather_than_erroring() -> None: codec = SignedTokenCodec(secret=SECRET) other_codec = SignedTokenCodec(secret=b"1" * 32) session = _session() client = TestClient(_app(codec=codec)) client.cookies.set(SESSION_COOKIE_NAME, encode_session(session, codec=other_codec)) response = client.get("/whoami") assert response.status_code == 200 assert response.json() == {"authenticated": False} def test_expired_session_cookie_leaves_the_principal_unset() -> None: codec = SignedTokenCodec(secret=SECRET) now = datetime.now(UTC) session = _session(issued_at=now - timedelta(hours=2), expires_at=now - timedelta(hours=1)) client = TestClient(_app(codec=codec)) client.cookies.set(SESSION_COOKIE_NAME, encode_session(session, codec=codec)) response = client.get("/whoami") assert response.json() == {"authenticated": False} def test_missing_codec_is_a_silent_noop_even_with_a_cookie_present() -> None: """OIDC not configured (no auth_session_codec on app.state) must not crash.""" codec = SignedTokenCodec(secret=SECRET) session = _session() client = TestClient(_app(codec=None)) client.cookies.set(SESSION_COOKIE_NAME, encode_session(session, codec=codec)) response = client.get("/whoami") assert response.status_code == 200 assert response.json() == {"authenticated": False} def test_malformed_cookie_value_leaves_the_principal_unset() -> None: codec = SignedTokenCodec(secret=SECRET) client = TestClient(_app(codec=codec)) client.cookies.set(SESSION_COOKIE_NAME, "not-a-valid-token") response = client.get("/whoami") assert response.json() == {"authenticated": False}