diff --git a/packages/application/auth/jwt_handler.py b/packages/application/auth/jwt_handler.py new file mode 100644 index 000000000..830cd1212 --- /dev/null +++ b/packages/application/auth/jwt_handler.py @@ -0,0 +1,59 @@ +"""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