feat(auth): add opaque bearer sessions

server/tests/test_auth_session_endpoints.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.api.deps import CurrentUserContext, get_db, require_platform_admin_user
+from app.db.base import Base
+from app.main import create_app
+from app.models.auth_session import AuthSession
+from app.models.user_session_metric import UserSessionMetric
+from app.schemas.settings import SettingsWrite
+from app.services.auth_sessions import AuthSessionService
+from app.services.settings import SettingsService
+
+
+def build_client() -> tuple[TestClient, sessionmaker[Session]]:
+    engine = create_engine(
+        "sqlite+pysqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine)
+    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+    with session_factory() as db:
+        settings_service = SettingsService(db)
+        payload = settings_service.get_settings_snapshot().model_dump()
+        payload["adminForm"]["adminAccount"] = "auth-admin"
+        payload["adminForm"]["newPassword"] = "safe-admin-password"
+        payload["adminForm"]["confirmPassword"] = "safe-admin-password"
+        settings_service.save_settings_snapshot(SettingsWrite(**payload))
+
+    app = create_app()
+
+    def override_db() -> Generator[Session, None, None]:
+        with session_factory() as db:
+            yield db
+
+    app.dependency_overrides[get_db] = override_db
+    return TestClient(app), session_factory
+
+
+def login_admin(client: TestClient) -> dict:
+    response = client.post(
+        "/api/v1/auth/login",
+        json={"username": "auth-admin", "password": "safe-admin-password"},
+    )
+    assert response.status_code == 200
+    return response.json()
+
+
+def test_login_issues_opaque_bearer_token_and_me_uses_server_session() -> None:
+    client, session_factory = build_client()
+    payload = login_admin(client)
+    token = payload["accessToken"]
+
+    assert payload["tokenType"] == "Bearer"
+    assert payload["expiresAt"]
+    with session_factory() as db:
+        auth_session = db.scalars(select(AuthSession)).one()
+        assert auth_session.token_hash == AuthSessionService.hash_token(token)
+        assert auth_session.token_hash != token
+
+    response = client.get(
+        "/api/v1/auth/me",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert response.status_code == 200
+    assert response.json()["username"] == "auth-admin"
+    assert response.json()["isAdmin"] is True
+
+
+def test_forged_identity_headers_no_longer_authenticate() -> None:
+    client, _ = build_client()
+    response = client.get(
+        "/api/v1/auth/me",
+        headers={
+            "X-Auth-Username": "superadmin",
+            "X-Auth-Role-Codes": "manager",
+            "X-Auth-Is-Admin": "true",
+        },
+    )
+
+    assert response.status_code == 401
+    assert response.headers["www-authenticate"] == "Bearer"
+
+
+def test_expired_and_revoked_tokens_are_rejected() -> None:
+    client, session_factory = build_client()
+    first_payload = login_admin(client)
+    first_token = first_payload["accessToken"]
+
+    with session_factory() as db:
+        auth_session = db.scalars(
+            select(AuthSession).where(
+                AuthSession.token_hash == AuthSessionService.hash_token(first_token)
+            )
+        ).one()
+        auth_session.expires_at = datetime.now(UTC) - timedelta(seconds=1)
+        db.commit()
+
+    expired_response = client.get(
+        "/api/v1/auth/me",
+        headers={"Authorization": f"Bearer {first_token}"},
+    )
+    assert expired_response.status_code == 401
+
+    second_payload = login_admin(client)
+    second_token = second_payload["accessToken"]
+    logout_response = client.post(
+        "/api/v1/auth/logout",
+        headers={"Authorization": f"Bearer {second_token}"},
+        json={"sessionId": second_payload["sessionId"], "reason": "manual"},
+    )
+    assert logout_response.status_code == 200
+    with session_factory() as db:
+        metric_session = db.scalars(
+            select(UserSessionMetric).where(
+                UserSessionMetric.session_id == second_payload["sessionId"]
+            )
+        ).one()
+        assert metric_session.status == "closed"
+
+    revoked_response = client.get(
+        "/api/v1/auth/me",
+        headers={"Authorization": f"Bearer {second_token}"},
+    )
+    assert revoked_response.status_code == 401
+
+
+def test_manager_role_is_not_platform_admin() -> None:
+    manager = CurrentUserContext(
+        username="manager@example.com",
+        name="Manager",
+        role_codes=["manager"],
+        is_admin=False,
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        require_platform_admin_user(manager)
+
+    assert exc_info.value.status_code == 403