rebase

app/core/security.py

@@ -0,0 +1,85 @@
+"""Security: JWT creation/verification and password hashing."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+
+from app.config import get_settings
+
+_pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_password(plain: str) -> str:
+    """Hash a password using bcrypt."""
+    return _pwd_context.hash(plain)
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    """Verify a password against its bcrypt hash."""
+    try:
+        return _pwd_context.verify(plain, hashed)
+    except (ValueError, TypeError):
+        return False
+
+
+def create_access_token(
+    subject: str | uuid.UUID,
+    extra_claims: dict[str, Any] | None = None,
+    expires_in_minutes: int | None = None,
+) -> str:
+    """Create a signed JWT access token."""
+    cfg = get_settings()
+    minutes = expires_in_minutes or cfg.access_token_expire_minutes
+    now = datetime.now(timezone.utc)
+    payload: dict[str, Any] = {
+        "sub": str(subject),
+        "iat": int(now.timestamp()),
+        "exp": int((now + timedelta(minutes=minutes)).timestamp()),
+        "type": "access",
+    }
+    if extra_claims:
+        payload.update(extra_claims)
+    return jwt.encode(payload, cfg.secret_key, algorithm=cfg.jwt_algorithm)
+
+
+def create_refresh_token(
+    subject: str | uuid.UUID, expires_in_minutes: int | None = None
+) -> str:
+    """Create a signed JWT refresh token."""
+    cfg = get_settings()
+    minutes = expires_in_minutes or cfg.refresh_token_expire_minutes
+    now = datetime.now(timezone.utc)
+    payload = {
+        "sub": str(subject),
+        "iat": int(now.timestamp()),
+        "exp": int((now + timedelta(minutes=minutes)).timestamp()),
+        "type": "refresh",
+    }
+    return jwt.encode(payload, cfg.secret_key, algorithm=cfg.jwt_algorithm)
+
+
+def decode_token(token: str) -> dict[str, Any]:
+    """Decode and verify a JWT. Raises JWTError on failure."""
+    cfg = get_settings()
+    return jwt.decode(token, cfg.secret_key, algorithms=[cfg.jwt_algorithm])
+
+
+def validate_password_strength(password: str) -> list[str]:
+    """Return a list of validation errors (empty list = valid password)."""
+    errors: list[str] = []
+    if len(password) < 8:
+        errors.append("Password must be at least 8 characters long")
+    if not any(c.isalpha() for c in password):
+        errors.append("Password must contain at least one letter")
+    if not any(c.isdigit() for c in password):
+        errors.append("Password must contain at least one digit")
+    # Tiny blacklist of trivial passwords
+    blacklist = {"password", "12345678", "qwerty12", "password1", "abcdefgh"}
+    if password.lower() in blacklist:
+        errors.append("Password is too common")
+    return errors