Update auth/security.py

auth/security.py

@@ -3,55 +3,52 @@ from datetime import datetime, timedelta, timezone
 from typing import Optional, Dict, Any
 
 from jose import jwt, JWTError
-from passlib.context import CryptContext
 from fastapi import HTTPException, status, Depends
 from fastapi.security import OAuth2PasswordBearer
 
+from passlib.context import CryptContext
+
 
 # =========================================================
-# CONFIG (ENV-FIRST, NO HARDCODED SECRETS)
+# CONFIG (ENV-FIRST SECURITY MODEL)
 # =========================================================
 
-SECRET_KEY = os.getenv("SECRET_KEY", "")
+SECRET_KEY = os.getenv("SECRET_KEY")
 ALGORITHM = os.getenv("ALGORITHM", "HS256")
-ACCESS_TOKEN_EXPIRE_MINUTES = int(
-    os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "1440")
-)
+ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "1440"))
 
 if not SECRET_KEY:
     raise RuntimeError("SECRET_KEY is not set in environment variables")
 
 
 # =========================================================
-# PASSWORD HASHING (FIXED BCRYPT ISSUE)
+# PASSWORD HASHING (FIXED - NO BCRYPT)
 # =========================================================
-# IMPORTANT:
-# bcrypt_sha256 avoids:
-# - 72-byte password limit crash
-# - broken bcrypt backend detection (__about__ error)
+# Why this change:
+# - bcrypt is broken in your runtime (passlib backend + __about__ bug)
+# - bcrypt has 72-byte limitation causing signup crashes
+# - argon2 is modern, memory-hard, and production safe
 
 pwd_context = CryptContext(
-    schemes=["bcrypt_sha256"],
+    schemes=["argon2"],
     deprecated="auto",
+    argon2__time_cost=2,
+    argon2__memory_cost=102400,
+    argon2__parallelism=8,
 )
 
 
 def hash_password(password: str) -> str:
-    """
-    Enterprise-safe password hashing.
-    - avoids bcrypt 72-byte limitation
-    - avoids broken bcrypt backend in passlib
-    """
-    if not isinstance(password, str) or not password:
-        raise ValueError("Password must be a non-empty string")
+    if not isinstance(password, str) or not password.strip():
+        raise ValueError("Invalid password input")
 
     return pwd_context.hash(password)
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
-    """
-    Constant-time password verification.
-    """
+    if not plain_password or not hashed_password:
+        return False
+
     try:
         return pwd_context.verify(plain_password, hashed_password)
     except Exception:
@@ -59,7 +56,7 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
 
 
 # =========================================================
-# JWT TOKEN HANDLING
+# JWT HANDLING
 # =========================================================
 
 def create_access_token(
@@ -70,36 +67,21 @@ def create_access_token(
     to_encode = data.copy()
 
     expire = datetime.now(timezone.utc) + (
-        expires_delta
-        if expires_delta
-        else timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     )
 
     to_encode.update({"exp": expire})
 
-    encoded_jwt = jwt.encode(
-        to_encode,
-        SECRET_KEY,
-        algorithm=ALGORITHM
-    )
-
-    return encoded_jwt
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 
 
 def decode_token(token: str) -> Dict[str, Any]:
-    """
-    Decodes JWT and validates signature.
-    """
     try:
-        return jwt.decode(
-            token,
-            SECRET_KEY,
-            algorithms=[ALGORITHM]
-        )
+        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
     except JWTError:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid authentication token",
+            detail="Invalid or expired authentication token",
             headers={"WWW-Authenticate": "Bearer"},
         )
 
@@ -112,12 +94,9 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
 
 
 def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict[str, Any]:
-    """
-    Extracts authenticated user from JWT.
-    """
     payload = decode_token(token)
 
-    user_id: Optional[str] = payload.get("sub")
+    user_id = payload.get("sub")
 
     if not user_id:
         raise HTTPException(
@@ -129,15 +108,15 @@ def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict[str, Any]:
 
 
 # =========================================================
-# OPTIONAL: PASSWORD POLICY (ENTERPRISE HARDENING)
+# PASSWORD POLICY (HARDENED)
 # =========================================================
 
 def validate_password_strength(password: str) -> None:
-    """
-    Enforces baseline security policy.
-    """
     if len(password) < 8:
         raise ValueError("Password must be at least 8 characters")
 
     if password.isdigit():
-        raise ValueError("Password cannot be only numeric")
+        raise ValueError("Password cannot be only numeric")
+
+    if password.lower() in {"password", "12345678", "qwerty"}:
+        raise ValueError("Password is too weak")