feat(security): T14 implement JWT utilities

- Add create_access_token with custom/default expiration
- Add decode_access_token with signature verification
- Add verify_token returning TokenData dataclass
- Support HS256 algorithm with config.SECRET_KEY
- Payload includes exp, iat, sub claims
- 19 comprehensive tests with 100% coverage
- Handle expired tokens, invalid signatures, missing claims
This commit is contained in:
Luca Sacchi Ricciardi
2026-04-07 12:10:04 +02:00
parent 54e81162df
commit 781e564ea0
3 changed files with 447 additions and 3 deletions

View File

@@ -0,0 +1,129 @@
"""JWT utilities for authentication.
This module provides functions for creating, decoding, and verifying
JWT tokens using the HS256 algorithm.
"""
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Optional
from jose import JWTError, jwt
# Default algorithm
ALGORITHM = "HS256"
DEFAULT_EXPIRE_HOURS = 24
@dataclass
class TokenData:
"""Data extracted from a verified JWT token."""
user_id: str
exp: datetime
iat: datetime
def create_access_token(
data: dict,
expires_delta: Optional[timedelta] = None,
secret_key: str = None,
) -> str:
"""Create a JWT access token.
Args:
data: Dictionary containing claims to encode (e.g., {"sub": user_id}).
expires_delta: Optional custom expiration time. Defaults to 24 hours.
secret_key: Secret key for signing. If None, uses config.SECRET_KEY.
Returns:
The encoded JWT token string.
Raises:
ValueError: If secret_key is not provided and config.SECRET_KEY is not set.
"""
# Import config here to avoid circular imports
if secret_key is None:
from openrouter_monitor.config import get_settings
settings = get_settings()
secret_key = settings.secret_key
to_encode = data.copy()
# Calculate expiration time
now = datetime.now(timezone.utc)
if expires_delta:
expire = now + expires_delta
else:
expire = now + timedelta(hours=DEFAULT_EXPIRE_HOURS)
# Add standard claims
to_encode.update({
"exp": expire,
"iat": now,
})
# Encode token
encoded_jwt = jwt.encode(to_encode, secret_key, algorithm=ALGORITHM)
return encoded_jwt
def decode_access_token(token: str, secret_key: str = None) -> dict:
"""Decode and validate a JWT token.
Args:
token: The JWT token string to decode.
secret_key: Secret key for verification. If None, uses config.SECRET_KEY.
Returns:
Dictionary containing the decoded payload.
Raises:
JWTError: If token is invalid, expired, or signature verification fails.
ValueError: If secret_key is not provided and config.SECRET_KEY is not set.
"""
# Import config here to avoid circular imports
if secret_key is None:
from openrouter_monitor.config import get_settings
settings = get_settings()
secret_key = settings.secret_key
payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM])
return payload
def verify_token(token: str, secret_key: str = None) -> TokenData:
"""Verify a JWT token and extract user data.
Args:
token: The JWT token string to verify.
secret_key: Secret key for verification. If None, uses config.SECRET_KEY.
Returns:
TokenData object containing user_id, exp, and iat.
Raises:
JWTError: If token is invalid, expired, or missing required claims.
ValueError: If secret_key is not provided and config.SECRET_KEY is not set.
"""
payload = decode_access_token(token, secret_key=secret_key)
# Extract required claims
user_id = payload.get("sub")
if user_id is None:
raise JWTError("Token missing 'sub' claim")
exp_timestamp = payload.get("exp")
iat_timestamp = payload.get("iat")
if exp_timestamp is None or iat_timestamp is None:
raise JWTError("Token missing exp or iat claim")
# Convert timestamps to datetime
exp = datetime.fromtimestamp(exp_timestamp, tz=timezone.utc)
iat = datetime.fromtimestamp(iat_timestamp, tz=timezone.utc)
return TokenData(user_id=user_id, exp=exp, iat=iat)