feat: simplify auth flow - replace JWT with session cookies
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 ✓
This commit is contained in:
@@ -0,0 +1,71 @@
|
||||
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
|
||||
Reference in New Issue
Block a user