security.py

src/core/security.py

@@ -105,44 +105,64 @@
 #             headers={"WWW-Authenticate": "Bearer"}
 #         )
 
+"""
+Security utilities for authentication and authorization.
+"""
 
+from datetime import datetime, timedelta
+from typing import Dict, Any
 
-"""Security utilities for authentication and authorization."""
 import jwt
-from datetime import datetime, timedelta
 from passlib.context import CryptContext
 from fastapi import HTTPException, status, Depends
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
-from typing import Dict, Any
+
 from src.core.config import settings
 
+# =========================
+# Password hashing (bcrypt-safe)
+# =========================
+
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 security = HTTPBearer()
-import hashlib
+
 MAX_BCRYPT_BYTES = 72
 
 
-def _normalize_password(password: str) -> bytes:
+def _bcrypt_safe(password: str) -> bytes:
     """
     bcrypt only supports 72 bytes.
-    This guarantees no runtime crash in any environment.
+    Truncate safely to avoid runtime crashes.
     """
-    password_bytes = password.encode("utf-8")
-    if len(password_bytes) > MAX_BCRYPT_BYTES:
-        password_bytes = password_bytes[:MAX_BCRYPT_BYTES]
-    return password_bytes
+    return password.encode("utf-8")[:MAX_BCRYPT_BYTES]
 
 
 def hash_password(password: str) -> str:
-    # SHA-256 produces 64 hex chars = 64 bytes < 72-byte limit
-    pre_hashed = hashlib.sha256(password.encode("utf-8")).hexdigest()
-    return pwd_context.hash(pre_hashed)
+    """
+    Hash a password using bcrypt (safe for all environments).
+    """
+    return pwd_context.hash(_bcrypt_safe(password))
 
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    pre_hashed = hashlib.sha256(plain_password.encode("utf-8")).hexdigest()
-    return pwd_context.verify(pre_hashed, hashed_password)
 
-def create_jwt_token(user_id: int, email: str, secret: str, expiration_days: int = 7) -> str:
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """
+    Verify a password against its bcrypt hash.
+    """
+    return pwd_context.verify(
+        _bcrypt_safe(plain_password),
+        hashed_password
+    )
+
+# =========================
+# JWT utilities
+# =========================
+
+def create_jwt_token(
+    user_id: int,
+    email: str,
+    secret: str,
+    expiration_days: int = 7
+) -> str:
     now = datetime.utcnow()
     payload = {
         "sub": str(user_id),
@@ -162,57 +182,59 @@ def verify_jwt_token(token: str, secret: str) -> dict:
             algorithms=["HS256"],
             options={"verify_exp": True},
         )
+
         if payload.get("iss") != "better-auth":
-            raise HTTPException(status_code=401, detail="Invalid token issuer")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid token issuer"
+            )
+
         return payload
+
     except jwt.ExpiredSignatureError:
-        raise HTTPException(status_code=401, detail="Token expired")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token expired"
+        )
     except jwt.InvalidTokenError:
-        raise HTTPException(status_code=401, detail="Invalid token")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid token"
+        )
 
+# =========================
+# FastAPI dependency
+# =========================
 
 def get_current_user(
     credentials: HTTPAuthorizationCredentials = Depends(security)
 ) -> Dict[str, Any]:
     """
-    FastAPI dependency to extract and validate JWT token from Authorization header.
-
-    Args:
-        credentials: HTTP Bearer token credentials from request header
-
-    Returns:
-        Dictionary containing user information from token payload:
-        - id: User ID (parsed from 'sub' claim)
-        - email: User email
-        - iat: Token issued at timestamp
-        - exp: Token expiration timestamp
-
-    Raises:
-        HTTPException 401: If token is missing, invalid, or expired
+    Extract and validate JWT token from Authorization header.
     """
     token = credentials.credentials
 
     try:
         payload = verify_jwt_token(token, settings.BETTER_AUTH_SECRET)
 
-        # Extract user ID from 'sub' claim and convert to integer
         user_id = int(payload.get("sub"))
 
         return {
             "id": user_id,
             "email": payload.get("email"),
             "iat": payload.get("iat"),
-            "exp": payload.get("exp")
+            "exp": payload.get("exp"),
         }
+
     except ValueError:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid user ID in token",
-            headers={"WWW-Authenticate": "Bearer"}
+            headers={"WWW-Authenticate": "Bearer"},
         )
     except Exception as e:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail=f"Authentication failed: {str(e)}",
-            headers={"WWW-Authenticate": "Bearer"}
+            headers={"WWW-Authenticate": "Bearer"},
         )