Start Magent beta overhaul

backend/app/security.py

@@ -0,0 +1,73 @@
+from datetime import datetime, timedelta, timezone
+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=["pbkdf2_sha256"], deprecated="auto")
+_ALGORITHM = "HS256"
+MIN_PASSWORD_LENGTH = 8
+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:
+    return _pwd_context.verify(plain_password, hashed_password)
+
+
+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",
+) -> str:
+    payload: Dict[str, Any] = {
+        "sub": subject,
+        "role": role,
+        "typ": token_type,
+        "exp": expires_at,
+    }
+    return jwt.encode(payload, settings.jwt_secret, algorithm=_ALGORITHM)
+
+def create_access_token(subject: str, role: str, expires_minutes: Optional[int] = None) -> 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")
+
+
+def create_stream_token(subject: str, role: str, expires_seconds: int = 120) -> 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")
+
+
+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])
+
+
+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