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
|