60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
"""JWT token handling implementation (application layer)."""
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Any, Dict, Optional
|
|
|
|
import jwt
|
|
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
|
|
|
|
|
|
def create_access_token(
|
|
payload: Dict[str, Any],
|
|
secret_key: str,
|
|
algorithm: str = "HS256",
|
|
expires_minutes: int = 30,
|
|
) -> str:
|
|
"""Create a JWT access token."""
|
|
now = datetime.utcnow()
|
|
expires = now + timedelta(minutes=expires_minutes)
|
|
payload["iat"] = now
|
|
payload["exp"] = expires
|
|
return jwt.encode(payload, secret_key, algorithm=algorithm)
|
|
|
|
|
|
def create_refresh_token(
|
|
payload: Dict[str, Any],
|
|
secret_key: str,
|
|
algorithm: str = "HS256",
|
|
expires_days: int = 30,
|
|
) -> str:
|
|
"""Create a JWT refresh token."""
|
|
now = datetime.utcnow()
|
|
expires = now + timedelta(days=expires_days)
|
|
payload["iat"] = now
|
|
payload["exp"] = expires
|
|
return jwt.encode(payload, secret_key, algorithm=algorithm)
|
|
|
|
|
|
def decode_token(
|
|
token: str,
|
|
secret_key: str,
|
|
algorithms: list[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Decode and verify a JWT token."""
|
|
if algorithms is None:
|
|
algorithms = ["HS256"]
|
|
try:
|
|
return jwt.decode(token, secret_key, algorithms=algorithms)
|
|
except ExpiredSignatureError:
|
|
raise ValueError("Token has expired")
|
|
except InvalidTokenError as e:
|
|
raise ValueError(f"Invalid token: {e}")
|
|
|
|
|
|
def decode_token_unsafe(token: str) -> Optional[Dict[str, Any]]:
|
|
"""Decode a JWT token without signature verification."""
|
|
try:
|
|
return jwt.decode(token, options={"verify_signature": False})
|
|
except Exception:
|
|
return None
|