diff --git a/auth/utils.py b/auth/utils.py index 4bded89..977c2f7 100644 --- a/auth/utils.py +++ b/auth/utils.py @@ -1,19 +1,26 @@ -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""" @@ -21,19 +28,19 @@ def hash_password(password: str) -> str: 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,18 +86,18 @@ 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' @@ -98,7 +105,7 @@ def get_device_info(user_agent: str = None) -> dict: 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 diff --git a/src/handoff/__init__.py b/src/handoff/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..cbb62c1 --- /dev/null +++ b/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)