Leopoldadmin
d89304845a
- Backend: FastAPI, JWT auth (HS256), bcrypt, RBAC middleware - User CRUD (admin-only), soft-delete, pagination - Pydantic BaseSettings config, async SQLAlchemy 2.0 - 50 backend tests, 88% coverage - Frontend: Next.js 14 App Router, Tailwind, design tokens - Login page, auth context, API client with auto-refresh - i18n: next-intl, DE/EN (31 keys each) - 6 base UI components (Button, Input, Card, Table, Modal, Toast) - 12 frontend tests, npm build success
84 lines
2.1 KiB
Python
84 lines
2.1 KiB
Python
"""JWT utility functions for token creation and verification."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from jose import JWTError, jwt
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def create_access_token(
|
|
user_id: str,
|
|
role: str,
|
|
email: str,
|
|
lang: str = "de",
|
|
) -> str:
|
|
"""Create a short-lived access JWT."""
|
|
now = datetime.now(timezone.utc)
|
|
expire = now + timedelta(seconds=settings.access_token_ttl_seconds)
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"role": role,
|
|
"email": email,
|
|
"lang": lang,
|
|
"exp": expire,
|
|
"iat": now,
|
|
"type": "access",
|
|
}
|
|
return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
|
|
|
|
|
|
def create_refresh_token(
|
|
user_id: str,
|
|
role: str,
|
|
email: str,
|
|
lang: str = "de",
|
|
) -> str:
|
|
"""Create a long-lived refresh JWT."""
|
|
now = datetime.now(timezone.utc)
|
|
expire = now + timedelta(seconds=settings.refresh_token_ttl_seconds)
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"role": role,
|
|
"email": email,
|
|
"lang": lang,
|
|
"exp": expire,
|
|
"iat": now,
|
|
"type": "refresh",
|
|
}
|
|
return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> Optional[dict]:
|
|
"""Decode and verify a JWT token. Returns payload or None."""
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
settings.JWT_SECRET,
|
|
algorithms=[settings.JWT_ALGORITHM],
|
|
)
|
|
return payload
|
|
except JWTError:
|
|
return None
|
|
|
|
|
|
def verify_access_token(token: str) -> Optional[dict]:
|
|
"""Verify an access token specifically (checks type claim)."""
|
|
payload = decode_token(token)
|
|
if payload is None:
|
|
return None
|
|
if payload.get("type") != "access":
|
|
return None
|
|
return payload
|
|
|
|
|
|
def verify_refresh_token(token: str) -> Optional[dict]:
|
|
"""Verify a refresh token specifically (checks type claim)."""
|
|
payload = decode_token(token)
|
|
if payload is None:
|
|
return None
|
|
if payload.get("type") != "refresh":
|
|
return None
|
|
return payload
|