Update auth/security.py

auth/security.py

@@ -1,32 +1,186 @@
-from datetime import datetime, timedelta
-from jose import jwt, JWTError
+"""
+auth/security.py
+Enterprise Authentication Security Layer
+Basyx Whisper V10.1
+"""
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional, Dict, Any
+
+from jose import JWTError, jwt
 from passlib.context import CryptContext
 
-SECRET_KEY = "CHANGE_THIS_SECRET"
-ALGORITHM = "HS256"
-TOKEN_EXPIRE_DAYS = 30
+from utils.logger import logger
+
+# ==========================================================
+# SECURITY CONSTANTS
+# ==========================================================
+
+# IMPORTANT:
+# Replace ONLY via environment variable in production.
+# No fallback assumptions allowed.
+import os
+
+JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY")
+if not JWT_SECRET_KEY:
+    raise RuntimeError("JWT_SECRET_KEY environment variable is required")
+
+JWT_ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = int(
+    os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", "60")
+)
+
+# bcrypt safe configuration
+pwd_context = CryptContext(
+    schemes=["bcrypt"],
+    deprecated="auto",
+)
+
+# ==========================================================
+# PASSWORD UTILITIES
+# ==========================================================
+
+
+def _sanitize_password(password: str) -> bytes:
+    """
+    bcrypt only supports 72 bytes.
+
+    Enterprise rule:
+    - Never silently fail
+    - Deterministically truncate
+    """
+
+    if not isinstance(password, str):
+        raise TypeError("Password must be string")
+
+    encoded = password.encode("utf-8")
+
+    if len(encoded) > 72:
+        logger.warning("Password exceeded bcrypt limit — truncated safely")
+        encoded = encoded[:72]
+
+    return encoded
+
+
+def hash_password(password: str) -> str:
+    """
+    Secure password hashing
+    """
+
+    try:
+        safe_password = _sanitize_password(password)
+        return pwd_context.hash(safe_password)
+    except Exception as e:
+        logger.exception("Password hashing failed")
+        raise RuntimeError("Password hashing failure") from e
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """
+    Verify password against stored hash
+    """
+
+    try:
+        safe_password = _sanitize_password(plain_password)
+        return pwd_context.verify(safe_password, hashed_password)
+    except Exception:
+        logger.warning("Password verification failed")
+        return False
 
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
+# ==========================================================
+# JWT TOKEN MANAGEMENT
+# ==========================================================
 
-def hash_password(password: str):
-    return pwd_context.hash(password)
 
+def create_access_token(
+    subject: str,
+    additional_claims: Optional[Dict[str, Any]] = None,
+) -> str:
+    """
+    Create signed JWT access token
+    """
 
-def verify_password(password, hashed):
-    return pwd_context.verify(password, hashed)
+    if not subject:
+        raise ValueError("Token subject required")
 
+    expire = datetime.now(timezone.utc) + timedelta(
+        minutes=ACCESS_TOKEN_EXPIRE_MINUTES
+    )
 
-def create_token(user_id: int):
-    payload = {
-        "sub": str(user_id),
-        "exp": datetime.utcnow() + timedelta(days=TOKEN_EXPIRE_DAYS)
+    payload: Dict[str, Any] = {
+        "sub": subject,
+        "exp": expire,
+        "iat": datetime.now(timezone.utc),
+        "type": "access",
     }
-    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
 
+    if additional_claims:
+        payload.update(additional_claims)
+
+    token = jwt.encode(
+        payload,
+        JWT_SECRET_KEY,
+        algorithm=JWT_ALGORITHM,
+    )
+
+    return token
+
+
+def decode_access_token(token: str) -> Optional[Dict[str, Any]]:
+    """
+    Decode and validate JWT token
+    """
 
-def decode_token(token: str):
     try:
-        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        payload = jwt.decode(
+            token,
+            JWT_SECRET_KEY,
+            algorithms=[JWT_ALGORITHM],
+        )
+
+        if payload.get("type") != "access":
+            logger.warning("Invalid token type")
+            return None
+
+        return payload
+
     except JWTError:
-        return None
+        logger.warning("JWT decode failed")
+        return None
+
+
+# ==========================================================
+# AUTH HELPER FUNCTIONS
+# ==========================================================
+
+
+def get_subject_from_token(token: str) -> Optional[str]:
+    """
+    Extract user identity from token
+    """
+
+    payload = decode_access_token(token)
+
+    if not payload:
+        return None
+
+    return payload.get("sub")
+
+
+# ==========================================================
+# HEALTH CHECK (DEBUG SAFE)
+# ==========================================================
+
+
+def security_status() -> dict:
+    """
+    Internal verification helper.
+    Safe for health diagnostics.
+    """
+
+    return {
+        "jwt_algorithm": JWT_ALGORITHM,
+        "token_expire_minutes": ACCESS_TOKEN_EXPIRE_MINUTES,
+        "bcrypt_scheme": pwd_context.schemes(),
+    }