Fusion
2ce7862058
Replace complex JWT + refresh token authentication with simple session-based auth using signed cookies. **Removed:** - JWT token service (jwt_service.py) - Refresh token store (refresh_store.py) - Refresh token model and database table - JWKS fetching and OIDC token verification - python-jose dependency **Added:** - Session service (session.py) with HMAC-SHA256 signed cookies - Auth dependencies module for shared auth logic - Session-based auth endpoints **Updated:** - All API endpoints to use session-based auth - Config: removed JWT settings, added SESSION_SECRET/SESSION_TTL_HOURS - Tests: rewritten for session-based flow - Frontend: no changes needed (already uses cookies) Quality gates: ruff ✓, mypy ✓, typecheck ✓, lint ✓
131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
import uuid
|
|
import asyncio
|
|
import importlib
|
|
|
|
from fastapi.testclient import TestClient
|
|
import pytest
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from src.auth.session import create_session_cookie
|
|
from src.config import Settings, build_database_url
|
|
from src.models import Base
|
|
from src.models.user import User
|
|
|
|
|
|
def _prepare_auth_test_db() -> None:
|
|
async def _run() -> None:
|
|
engine = create_async_engine(
|
|
build_database_url(
|
|
user="headquarter",
|
|
password="headquarter",
|
|
host="localhost",
|
|
port=5432,
|
|
database="headquarter",
|
|
)
|
|
)
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
await connection.execute(text("TRUNCATE TABLE users RESTART IDENTITY CASCADE"))
|
|
await engine.dispose()
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
def _load_app():
|
|
import src.database as database_module
|
|
import src.api.auth as auth_module
|
|
import src.main as main_module
|
|
|
|
importlib.reload(database_module)
|
|
importlib.reload(auth_module)
|
|
importlib.reload(main_module)
|
|
return main_module.app
|
|
|
|
|
|
def _insert_test_user(user_id: str) -> None:
|
|
async def _run() -> None:
|
|
engine = create_async_engine(
|
|
build_database_url(
|
|
user="headquarter",
|
|
password="headquarter",
|
|
host="localhost",
|
|
port=5432,
|
|
database="headquarter",
|
|
)
|
|
)
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with session_factory() as session:
|
|
user = User(
|
|
id=uuid.UUID(user_id),
|
|
email="test@headquarter.local",
|
|
name="Test User",
|
|
authentik_id="test-user",
|
|
avatar_url=None,
|
|
)
|
|
await session.merge(user)
|
|
await session.commit()
|
|
await engine.dispose()
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_login_redirects_to_authentik_authorize_endpoint() -> None:
|
|
_prepare_auth_test_db()
|
|
app = _load_app()
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/auth/login", follow_redirects=False)
|
|
|
|
assert response.status_code == 307
|
|
assert "response_type=code" in response.headers["location"]
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_me_returns_401_without_session_cookie() -> None:
|
|
_prepare_auth_test_db()
|
|
app = _load_app()
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/auth/me")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_me_returns_user_with_valid_session() -> None:
|
|
user_id = "11111111-1111-1111-1111-111111111111"
|
|
_prepare_auth_test_db()
|
|
_insert_test_user(user_id)
|
|
app = _load_app()
|
|
|
|
settings = Settings()
|
|
session_cookie = create_session_cookie(settings=settings, user_id=user_id)
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/auth/me", cookies={"session": session_cookie})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["email"] == "test@headquarter.local"
|
|
assert data["name"] == "Test User"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_logout_clears_session_cookie() -> None:
|
|
_prepare_auth_test_db()
|
|
app = _load_app()
|
|
|
|
client = TestClient(app)
|
|
response = client.post("/auth/logout")
|
|
|
|
assert response.status_code == 200
|
|
# Check that session cookie is deleted
|
|
set_cookie = response.headers.get("set-cookie", "")
|
|
assert "session=" in set_cookie or "session=\"\"" in set_cookie
|