117 lines
3.3 KiB
Python
117 lines
3.3 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
import uuid
|
|
from typing import Any, Dict, Optional
|
|
|
|
from passlib.context import CryptContext
|
|
import jwt
|
|
from jwt import InvalidTokenError
|
|
|
|
from .config import settings
|
|
|
|
_pwd_context = CryptContext(
|
|
schemes=["argon2", "pbkdf2_sha256"],
|
|
deprecated=["pbkdf2_sha256"],
|
|
argon2__memory_cost=65536,
|
|
argon2__time_cost=3,
|
|
argon2__parallelism=4,
|
|
)
|
|
_ALGORITHM = "HS256"
|
|
MIN_PASSWORD_LENGTH = 12
|
|
PASSWORD_POLICY_MESSAGE = f"Password must be at least {MIN_PASSWORD_LENGTH} characters."
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return _pwd_context.hash(password)
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
try:
|
|
return _pwd_context.verify(plain_password, hashed_password)
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
def verify_and_update_password(plain_password: str, hashed_password: str) -> tuple[bool, Optional[str]]:
|
|
try:
|
|
return _pwd_context.verify_and_update(plain_password, hashed_password)
|
|
except (TypeError, ValueError):
|
|
return False, None
|
|
|
|
|
|
def validate_password_policy(password: str) -> str:
|
|
candidate = password.strip()
|
|
if len(candidate) < MIN_PASSWORD_LENGTH:
|
|
raise ValueError(PASSWORD_POLICY_MESSAGE)
|
|
return candidate
|
|
|
|
|
|
def _create_token(
|
|
subject: str,
|
|
role: str,
|
|
*,
|
|
expires_at: datetime,
|
|
token_type: str = "access",
|
|
auth_version: int = 1,
|
|
) -> str:
|
|
issued_at = datetime.now(timezone.utc)
|
|
payload: Dict[str, Any] = {
|
|
"sub": subject,
|
|
"role": role,
|
|
"typ": token_type,
|
|
"exp": expires_at,
|
|
"iat": issued_at,
|
|
"jti": uuid.uuid4().hex,
|
|
"iss": settings.jwt_issuer,
|
|
"aud": settings.jwt_audience,
|
|
"ver": max(1, int(auth_version or 1)),
|
|
}
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm=_ALGORITHM)
|
|
|
|
def create_access_token(
|
|
subject: str,
|
|
role: str,
|
|
expires_minutes: Optional[int] = None,
|
|
*,
|
|
auth_version: int = 1,
|
|
) -> str:
|
|
if not settings.jwt_secret:
|
|
raise ValueError("JWT_SECRET is not configured")
|
|
minutes = expires_minutes or settings.jwt_exp_minutes
|
|
expires = datetime.now(timezone.utc) + timedelta(minutes=minutes)
|
|
return _create_token(subject, role, expires_at=expires, token_type="access", auth_version=auth_version)
|
|
|
|
|
|
def create_stream_token(
|
|
subject: str,
|
|
role: str,
|
|
expires_seconds: int = 120,
|
|
*,
|
|
auth_version: int = 1,
|
|
) -> str:
|
|
expires = datetime.now(timezone.utc) + timedelta(seconds=max(30, int(expires_seconds or 120)))
|
|
return _create_token(subject, role, expires_at=expires, token_type="sse", auth_version=auth_version)
|
|
|
|
|
|
def decode_token(token: str) -> Dict[str, Any]:
|
|
if not settings.jwt_secret:
|
|
raise ValueError("JWT_SECRET is not configured")
|
|
return jwt.decode(
|
|
token,
|
|
settings.jwt_secret,
|
|
algorithms=[_ALGORITHM],
|
|
audience=settings.jwt_audience,
|
|
issuer=settings.jwt_issuer,
|
|
options={"require": ["exp", "iat", "jti", "iss", "aud", "sub", "typ", "ver"]},
|
|
)
|
|
|
|
|
|
class TokenError(Exception):
|
|
pass
|
|
|
|
|
|
def safe_decode_token(token: str) -> Dict[str, Any]:
|
|
try:
|
|
return decode_token(token)
|
|
except InvalidTokenError as exc:
|
|
raise TokenError("Invalid token") from exc
|