Upload app/core/security.py

app/core/security.py

@@ -0,0 +1,101 @@
+"""Password hashing and JWT encoding/decoding.
+
+Uses python-jose[cryptography] (per 02-architecture.md Section 13.1) and
+passlib[bcrypt] with bcrypt 4.0.1 (per Section 13.6, 03a-patterns R-6).
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+
+from app.core.config import get_settings
+
+_settings = get_settings()
+
+# === Password Hashing ===
+
+# CryptContext auto-upgrades old hashes when verified. bcrypt 4.0.1 is pinned
+# because 4.1+ breaks passlib (per 03a-patterns-summary R-6).
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_password(plain: str) -> str:
+    """Hash a plaintext password using bcrypt with the configured rounds."""
+    return pwd_context.hash(plain)
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    """Verify a plaintext password against a bcrypt hash."""
+    try:
+        return pwd_context.verify(plain, hashed)
+    except (ValueError, TypeError):
+        return False
+
+
+# === JWT ===
+
+
+def create_access_token(
+    user_id: int,
+    org_id: int,
+    role: str,
+    expires_delta: timedelta | None = None,
+) -> str:
+    """Create a JWT access token with the given user, org, and role claims.
+
+    The token contains:
+      - sub: user id (string)
+      - org_id: organization id
+      - role: user role
+      - exp: expiry timestamp
+      - iat: issued-at timestamp
+    """
+    now = datetime.now(UTC)
+    if expires_delta is None:
+        expires_delta = timedelta(hours=_settings.JWT_EXPIRY_HOURS)
+    expire = now + expires_delta
+
+    payload: dict[str, Any] = {
+        "sub": str(user_id),
+        "org_id": org_id,
+        "role": role,
+        "iat": int(now.timestamp()),
+        "exp": int(expire.timestamp()),
+    }
+    return jwt.encode(payload, _settings.AUTH_SECRET, algorithm=_settings.JWT_ALGORITHM)
+
+
+def decode_access_token(token: str) -> dict[str, Any] | None:
+    """Decode and validate a JWT access token.
+
+    Returns the payload dict on success, or None on any error
+    (invalid signature, malformed token, expired).
+    """
+    try:
+        payload = jwt.decode(
+            token,
+            _settings.AUTH_SECRET,
+            algorithms=[_settings.JWT_ALGORITHM],
+        )
+        return payload
+    except JWTError:
+        return None
+
+
+def is_token_expired(token: str) -> bool:
+    """Check whether a token is expired without raising on other errors."""
+    try:
+        jwt.get_unverified_claims(token)
+        # If decode succeeds, it's not expired.
+        jwt.decode(
+            token, _settings.AUTH_SECRET, algorithms=[_settings.JWT_ALGORITHM]
+        )
+        return False
+    except JWTError as e:
+        return "expired" in str(e).lower() or "exp" in str(e).lower()
+    except Exception:
+        return True