Update src/utils/security.py

src/utils/security.py

@@ -2,122 +2,56 @@ from passlib.context import CryptContext
 from datetime import datetime, timedelta
 from typing import Optional, Union
 import uuid
-import logging
 from jose import JWTError, jwt
 from ..config import settings
 
-# Configure logger
-logger = logging.getLogger(__name__)
-logger.setLevel(logging.DEBUG)
-
-# Fallback values for JWT settings
-_FALLBACK_JWT_SECRET_KEY = "fallback_secret_key_for_development_only"
-_FALLBACK_JWT_ALGORITHM = "HS256"
-_FALLBACK_ACCESS_TOKEN_EXPIRE_DAYS = 7
-
 # Password hashing context
-# Handle bcrypt backend issues by specifying a fallback
-try:
-    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-except Exception as e:
-    logger.warning(f"Failed to initialize bcrypt context: {e}, using plaintext (NOT FOR PRODUCTION)")
-    pwd_context = CryptContext(schemes=["plaintext"], deprecated="auto")
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 
 def hash_password(password: str) -> str:
-    """Hash a password using bcrypt or fallback method."""
-    try:
-        # Truncate password to 72 bytes to avoid bcrypt limitation
-        if len(password.encode('utf-8')) > 72:
-            logger.warning("Password exceeds 72 bytes, truncating")
-            # Properly truncate UTF-8 bytes
-            password_bytes = password.encode('utf-8')[:72]
-            password = password_bytes.decode('utf-8', errors='ignore')
-        return pwd_context.hash(password)
-    except ValueError as ve:
-        if "72 bytes" in str(ve):
-            logger.error(f"Password too long even after truncation: {str(ve)}")
-            # Force truncate to exactly 72 bytes and try again
-            password = password.encode('utf-8')[:72].decode('utf-8', errors='ignore')
-            return pwd_context.hash(password)
-        else:
-            logger.error(f"ValueError hashing password: {str(ve)}")
-            raise
-    except Exception as e:
-        logger.error(f"Error hashing password: {str(e)}")
-        raise
+    """Hash a password using bcrypt."""
+    return pwd_context.hash(password)
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
     """Verify a plain password against its hash."""
-    try:
-        # Truncate password to 72 bytes to match hashing behavior
-        if len(plain_password.encode('utf-8')) > 72:
-            logger.warning("Password exceeds 72 bytes during verification, truncating")
-            plain_password = plain_password.encode('utf-8')[:72].decode('utf-8', errors='ignore')
-        return pwd_context.verify(plain_password, hashed_password)
-    except Exception as e:
-        logger.error(f"Error verifying password: {str(e)}")
-        return False
+    return pwd_context.verify(plain_password, hashed_password)
 
 
 def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
     """Create a JWT access token."""
-    try:
-        to_encode = data.copy()
+    to_encode = data.copy()
 
-        if expires_delta:
-            expire = datetime.utcnow() + expires_delta
-        else:
-            # Default to 7 days if no expiration is provided
-            expire = datetime.utcnow() + timedelta(days=getattr(settings, 'ACCESS_TOKEN_EXPIRE_DAYS', _FALLBACK_ACCESS_TOKEN_EXPIRE_DAYS))
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        # Default to 7 days if no expiration is provided
+        expire = datetime.utcnow() + timedelta(days=settings.ACCESS_TOKEN_EXPIRE_DAYS)
 
-        to_encode.update({"exp": expire, "iat": datetime.utcnow()})
-        
-        # Use fallback values if settings are not properly configured
-        secret_key = getattr(settings, 'JWT_SECRET_KEY', _FALLBACK_JWT_SECRET_KEY)
-        algorithm = getattr(settings, 'JWT_ALGORITHM', _FALLBACK_JWT_ALGORITHM)
-        
-        logger.debug(f"Creating token with settings: SECRET_KEY={secret_key}, ALGORITHM={algorithm}")
+    to_encode.update({"exp": expire, "iat": datetime.utcnow()})
 
-        encoded_jwt = jwt.encode(to_encode, secret_key, algorithm=algorithm)
-        return encoded_jwt
-    except Exception as e:
-        logger.error(f"Error creating access token: {str(e)}")
-        raise
+    encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
+    return encoded_jwt
 
 
 def verify_token(token: str) -> Optional[dict]:
     """Verify a JWT token and return the payload if valid."""
     try:
-        # Use fallback values if settings are not properly configured
-        secret_key = getattr(settings, 'JWT_SECRET_KEY', _FALLBACK_JWT_SECRET_KEY)
-        algorithm = getattr(settings, 'JWT_ALGORITHM', _FALLBACK_JWT_ALGORITHM)
-        
-        logger.debug(f"Verifying token with settings: SECRET_KEY={secret_key}, ALGORITHM={algorithm}")
-        payload = jwt.decode(token, secret_key, algorithms=[algorithm])
+        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
         return payload
-    except JWTError as e:
-        logger.warning(f"JWT Error during token verification: {str(e)}")
-        return None
-    except Exception as e:
-        logger.error(f"Unexpected error during token verification: {str(e)}")
+    except JWTError:
         return None
 
 
 def verify_user_id_from_token(token: str) -> Optional[uuid.UUID]:
     """Extract user_id from JWT token."""
-    try:
-        payload = verify_token(token)
-        if payload:
-            user_id_str = payload.get("sub")
-            if user_id_str:
-                try:
-                    return uuid.UUID(user_id_str)
-                except ValueError as e:
-                    logger.warning(f"Invalid UUID format in token payload: {user_id_str} - {str(e)}")
-                    return None
-        return None
-    except Exception as e:
-        logger.error(f"Error extracting user ID from token: {str(e)}")
-        return None
+    payload = verify_token(token)
+    if payload:
+        user_id_str = payload.get("sub")
+        if user_id_str:
+            try:
+                return uuid.UUID(user_id_str)
+            except ValueError:
+                return None
+    return None