139 lines
4.7 KiB
Python
139 lines
4.7 KiB
Python
"""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}
|