🛡️ Sentinel: Fix Information Disclosure in Authentication Decorator

auth/utils.py

@@ -1,39 +1,46 @@
-import bcrypt
-import jwt
-import os
-import secrets
+"""Authentication and authorization utilities."""
+
 from datetime import datetime, timedelta
 from functools import wraps
-from flask import request, jsonify, g
+import logging
+import os
+import secrets
+from typing import Optional, Tuple
+
+import bcrypt
+from flask import g, jsonify, request
 from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request
-from cache_db.redis_client import redis_client
-from cache_db.models import User, RefreshToken
-from typing import Tuple, Optional
+import jwt
+
+from cache_db.models import User  # pylint: disable=import-error
+from cache_db.redis_client import redis_client  # pylint: disable=import-error
+
+logger = logging.getLogger(__name__)
 
 
 class PasswordUtils:
     """Password hashing and verification utilities"""
-    
+
     @staticmethod
     def hash_password(password: str) -> str:
         """Hash password using bcrypt"""
         if not password or len(password) < 8:
             raise ValueError("Password must be at least 8 characters")
         salt = bcrypt.gensalt(rounds=12)
         return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8')
-    
+
     @staticmethod
     def verify_password(password: str, hash_: str) -> bool:
         """Verify password against hash"""
         try:
             return bcrypt.checkpw(password.encode('utf-8'), hash_.encode('utf-8'))
-        except Exception:
+        except Exception:  # pylint: disable=broad-exception-caught
             return False
 
 
 class JWTUtils:
     """JWT token utilities"""
-    
+
     @staticmethod
     def create_tokens(user_id: str, username: str) -> Tuple[str, str]:
         """Create access and refresh tokens"""
@@ -47,7 +54,7 @@ def create_tokens(user_id: str, username: str) -> Tuple[str, str]:
             os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
             algorithm='HS256'
         )
-        
+
         refresh_token = jwt.encode(
             {
                 'user_id': user_id,
@@ -58,9 +65,9 @@ def create_tokens(user_id: str, username: str) -> Tuple[str, str]:
             os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
             algorithm='HS256'
         )
-        
+
         return access_token, refresh_token
-    
+
     @staticmethod
     def decode_token(token: str) -> Optional[dict]:
         """Decode and verify token"""
@@ -79,26 +86,26 @@ def decode_token(token: str) -> Optional[dict]:
 
 class SessionUtils:
     """Session management utilities"""
-    
+
     @staticmethod
     def generate_session_id() -> str:
         """Generate unique session ID"""
         return secrets.token_urlsafe(32)
-    
+
     @staticmethod
     def get_device_info(user_agent: str = None) -> dict:
         """Extract device info from user agent"""
         if not user_agent:
             user_agent = request.headers.get('User-Agent', 'Unknown')
-        
+
         # Simple device detection
         if 'Mobile' in user_agent or 'Android' in user_agent:
             device_type = 'mobile'
         elif 'Tablet' in user_agent or 'iPad' in user_agent:
             device_type = 'tablet'
         else:
             device_type = 'desktop'
-        
+
         return {
             'user_agent': user_agent,
             'device_type': device_type,
@@ -115,7 +122,7 @@ def decorated_function(*args, **kwargs):
         try:
             verify_jwt_in_request()
             user_id = get_jwt_identity()
-            
+
             # Try to get user from cache first
             user_data = redis_client.get_cached_user(user_id)
             if not user_data:
@@ -124,12 +131,14 @@ def decorated_function(*args, **kwargs):
                     return jsonify({'error': 'User not found or inactive'}), 401
                 user_data = user.to_dict()
                 redis_client.cache_user(user_id, user_data)
-            
+
             g.user_id = user_id
             g.user = user_data
             return f(*args, **kwargs)
-        except Exception as e:
-            return jsonify({'error': 'Unauthorized', 'details': str(e)}), 401
+        except Exception as e:  # pylint: disable=broad-exception-caught
+            # Log security exception internally without leaking details to response
+            logger.error("Authentication failed: %s", e)
+            return jsonify({'error': 'Unauthorized'}), 401
     return decorated_function
 
 

tests/test_auth.py

@@ -0,0 +1,81 @@
+"""Tests for authentication utilities."""
+
+import sys
+from unittest.mock import MagicMock, patch
+
+# Mock cache_db module dependencies prior to importing auth.utils
+mock_cache_db = MagicMock()
+mock_redis_client = MagicMock()
+mock_models = MagicMock()
+
+sys.modules['cache_db'] = mock_cache_db
+sys.modules['cache_db.redis_client'] = mock_redis_client
+sys.modules['cache_db.models'] = mock_models
+
+from flask import Flask, jsonify  # pylint: disable=wrong-import-position
+import pytest  # pylint: disable=wrong-import-position
+
+from auth.utils import (  # pylint: disable=wrong-import-position
+    JWTUtils,
+    PasswordUtils,
+    SessionUtils,
+    login_required,
+)
+
+
+@pytest.fixture
+def app():
+    """Create Flask test app."""
+    test_app = Flask(__name__)
+    test_app.config['TESTING'] = True
+    test_app.config['JWT_SECRET_KEY'] = 'test-secret'
+
+    @test_app.route('/protected')
+    @login_required
+    def protected():
+        return jsonify({'message': 'success'})
+
+    return test_app
+
+
+def test_password_utils():
+    """Test password hashing and verification."""
+    password = "securePassword123"
+    hashed = PasswordUtils.hash_password(password)
+    assert hashed != password
+    assert PasswordUtils.verify_password(password, hashed) is True
+    assert PasswordUtils.verify_password("wrongPassword", hashed) is False
+
+    with pytest.raises(ValueError):
+        PasswordUtils.hash_password("short")
+
+
+def test_jwt_utils():
+    """Test JWT creation and decoding."""
+    access, refresh = JWTUtils.create_tokens("user-1", "testuser")
+    assert access is not None
+    assert refresh is not None
+
+    decoded = JWTUtils.decode_token(access)
+    assert decoded is not None
+    assert decoded['user_id'] == "user-1"
+    assert decoded['username'] == "testuser"
+
+
+def test_session_utils():
+    """Test session ID generation."""
+    session_id = SessionUtils.generate_session_id()
+    assert isinstance(session_id, str)
+    assert len(session_id) > 0
+
+
+def test_login_required_sanitizes_error_details(app):
+    """Verify login_required does not leak sensitive internal exception details."""
+    with app.test_client() as client:
+        with patch('auth.utils.verify_jwt_in_request', side_effect=Exception("Sensitive DB error connection string: postgresql://user:pass@localhost:5432/db")):
+            response = client.get('/protected')
+            assert response.status_code == 401
+            data = response.get_json()
+            assert data == {'error': 'Unauthorized'}
+            assert 'details' not in data
+            assert 'Sensitive DB error' not in str(data)