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 ✓
72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
import hmac
|
|
import hashlib
|
|
import json
|
|
import base64
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from src.config import Settings
|
|
|
|
|
|
def _base64url_encode(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def _base64url_decode(data: str) -> bytes:
|
|
padding = 4 - len(data) % 4
|
|
if padding != 4:
|
|
data += "=" * padding
|
|
return base64.urlsafe_b64decode(data)
|
|
|
|
|
|
def create_session_cookie(*, settings: Settings, user_id: str) -> str:
|
|
"""Create a signed session cookie value."""
|
|
payload = {
|
|
"user_id": user_id,
|
|
"exp": int((datetime.now(UTC) + timedelta(hours=settings.session_ttl_hours)).timestamp()),
|
|
}
|
|
|
|
header = _base64url_encode(json.dumps({"alg": "HS256", "typ": "session"}).encode())
|
|
payload_encoded = _base64url_encode(json.dumps(payload).encode())
|
|
message = f"{header}.{payload_encoded}"
|
|
|
|
signature = hmac.new(
|
|
settings.session_secret.encode(),
|
|
message.encode(),
|
|
hashlib.sha256,
|
|
).digest()
|
|
signature_encoded = _base64url_encode(signature)
|
|
|
|
return f"{message}.{signature_encoded}"
|
|
|
|
|
|
def decode_session_cookie(*, settings: Settings, cookie_value: str) -> dict[str, Any]:
|
|
"""Decode and verify a session cookie. Returns payload or raises ValueError."""
|
|
parts = cookie_value.split(".")
|
|
if len(parts) != 3:
|
|
raise ValueError("invalid session format")
|
|
|
|
header, payload_encoded, signature_encoded = parts
|
|
message = f"{header}.{payload_encoded}"
|
|
|
|
# Verify signature
|
|
expected_signature = hmac.new(
|
|
settings.session_secret.encode(),
|
|
message.encode(),
|
|
hashlib.sha256,
|
|
).digest()
|
|
expected_signature_encoded = _base64url_encode(expected_signature)
|
|
|
|
if not hmac.compare_digest(signature_encoded, expected_signature_encoded):
|
|
raise ValueError("invalid session signature")
|
|
|
|
# Decode payload
|
|
payload_bytes = _base64url_decode(payload_encoded)
|
|
payload = json.loads(payload_bytes)
|
|
|
|
# Check expiry
|
|
if payload.get("exp", 0) < int(datetime.now(UTC).timestamp()):
|
|
raise ValueError("session expired")
|
|
|
|
return payload
|