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 ✓
65 lines
1.6 KiB
Python
65 lines
1.6 KiB
Python
from typing import Any
|
|
from urllib.parse import urlencode
|
|
|
|
import httpx
|
|
|
|
from src.config import Settings
|
|
|
|
|
|
def build_login_redirect_url(
|
|
*,
|
|
settings: Settings,
|
|
redirect_uri: str,
|
|
state: str,
|
|
) -> str:
|
|
query = urlencode(
|
|
{
|
|
"response_type": "code",
|
|
"client_id": settings.authentik_client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"scope": "openid profile email",
|
|
"state": state,
|
|
}
|
|
)
|
|
return f"{settings.resolved_authentik_authorize_url}?{query}"
|
|
|
|
|
|
async def exchange_code_for_tokens(
|
|
*,
|
|
settings: Settings,
|
|
code: str,
|
|
redirect_uri: str,
|
|
client: httpx.AsyncClient,
|
|
) -> dict[str, str]:
|
|
response = await client.post(
|
|
settings.resolved_authentik_token_url,
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
"client_id": settings.authentik_client_id,
|
|
"client_secret": settings.authentik_client_secret,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
return {
|
|
"access_token": payload["access_token"],
|
|
"refresh_token": payload.get("refresh_token"),
|
|
}
|
|
|
|
|
|
async def fetch_user_info(
|
|
*,
|
|
settings: Settings,
|
|
access_token: str,
|
|
client: httpx.AsyncClient,
|
|
) -> dict[str, Any]:
|
|
"""Fetch user info from Authentik userinfo endpoint."""
|
|
response = await client.get(
|
|
f"{settings.authentik_base_url}/application/o/userinfo/",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|