diff --git a/sanguo_api/auth.py b/sanguo_api/auth.py index 1f30c73..2db600e 100644 --- a/sanguo_api/auth.py +++ b/sanguo_api/auth.py @@ -3,11 +3,10 @@ import os from datetime import datetime, timedelta, timezone import jwt -from passlib.context import CryptContext +import bcrypt from fastapi import HTTPException, status _CONFIG = {"secret": "change-me", "expire_minutes": 60, "algorithm": "HS256"} -_pwd = CryptContext(schemes=["bcrypt"], deprecated="auto") def set_jwt_config(secret: str, expire_minutes: int, algorithm: str = "HS256"): @@ -15,11 +14,14 @@ def set_jwt_config(secret: str, expire_minutes: int, algorithm: str = "HS256"): def hash_password(password: str) -> str: - return _pwd.hash(password) + return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(password: str, password_hash: str) -> bool: - return _pwd.verify(password, password_hash) + try: + return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8")) + except (ValueError, TypeError): + return False def create_token(username: str) -> str: diff --git a/scripts/smoke_phase3a.py b/scripts/smoke_phase3a.py index 218445c..e0fda1d 100644 --- a/scripts/smoke_phase3a.py +++ b/scripts/smoke_phase3a.py @@ -35,7 +35,7 @@ def test_jwt_login(): try: from fastapi.testclient import TestClient from sanguo_api.app import create_app - from unittest.mock import patch + from sanguo_api.auth import hash_password # Create temporary directory for test tmpdir = tempfile.mkdtemp() @@ -43,9 +43,9 @@ def test_jwt_login(): file_dir = tmpdir try: - # Use pre-hashed bcrypt password to avoid compatibility issues - # Password: "password123" hashed with bcrypt - password_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyY9Wt3KpPqm" + # Generate real password hash using bcrypt directly + test_password = "password123" + password_hash = hash_password(test_password) # Create app with auth config app = create_app( @@ -60,28 +60,26 @@ def test_jwt_login(): max_workers=1 ) - # Mock the password verification to avoid bcrypt compatibility issues - with patch('sanguo_api.auth.verify_password', return_value=True): - # Test login with TestClient - client = TestClient(app) - response = client.post("/api/v1/auth/login", json={ - "username": "admin", - "password": "password123" - }) + # Test login with TestClient - REAL password verification + client = TestClient(app) + response = client.post("/api/v1/auth/login", json={ + "username": "admin", + "password": test_password + }) - assert response.status_code == 200, f"Expected 200, got {response.status_code}" - data = response.json() - assert "token" in data, "Token not in response" - print(f" Login successful, token: {data['token'][:20]}...") - print("=== PASS: JWT login ===") - return data['token'] + assert response.status_code == 200, f"Expected 200, got {response.status_code}" + data = response.json() + assert "token" in data, "Token not in response" + print(f" Login successful, token: {data['token'][:20]}...") + print("=== PASS: JWT login ===") + return data['token'] finally: shutil.rmtree(tmpdir, ignore_errors=True) except Exception as e: - print(f" === SKIP: JWT login test failed due to bcrypt compatibility: {type(e).__name__} ===") - print(" This is a known issue with passlib/bcrypt compatibility in container environment") - print(" Core JWT functionality is tested in local environment") - print("=== SKIP: JWT login ===") + print(f" === FAIL: JWT login test failed: {type(e).__name__}: {e} ===") + import traceback + traceback.print_exc() + print("=== FAIL: JWT login ===") return None @@ -91,15 +89,16 @@ def test_protected_route_auth(): try: from fastapi.testclient import TestClient from sanguo_api.app import create_app - from unittest.mock import patch + from sanguo_api.auth import hash_password tmpdir = tempfile.mkdtemp() db_path = os.path.join(tmpdir, "test.db") file_dir = tmpdir try: - # Use pre-hashed bcrypt password to avoid compatibility issues - password_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyY9Wt3KpPqm" + # Generate real password hash using bcrypt directly + test_password = "password123" + password_hash = hash_password(test_password) app = create_app( db_path=db_path, @@ -120,14 +119,12 @@ def test_protected_route_auth(): assert response.status_code == 401, f"Expected 401 without token, got {response.status_code}" print(" No token: 401 Unauthorized ✓") - # Mock the password verification to avoid bcrypt compatibility issues - with patch('sanguo_api.auth.verify_password', return_value=True): - # Get valid token - login_response = client.post("/api/v1/auth/login", json={ - "username": "admin", - "password": "password123" - }) - token = login_response.json()["token"] + # Get valid token using REAL password verification + login_response = client.post("/api/v1/auth/login", json={ + "username": "admin", + "password": test_password + }) + token = login_response.json()["token"] # Test with token but non-existent task - should get 404 response = client.get( @@ -140,10 +137,10 @@ def test_protected_route_auth(): finally: shutil.rmtree(tmpdir, ignore_errors=True) except Exception as e: - print(f" === SKIP: Protected route auth test failed due to bcrypt compatibility: {type(e).__name__} ===") - print(" This is a known issue with passlib/bcrypt compatibility in container environment") - print(" Core auth functionality is tested in local environment") - print("=== SKIP: protected route auth ===") + print(f" === FAIL: Protected route auth test failed: {type(e).__name__}: {e} ===") + import traceback + traceback.print_exc() + print("=== FAIL: protected route auth ===") async def test_orchestrator_async(): @@ -192,13 +189,15 @@ async def test_ws_stage_wiring(): from sanguo_api.app import create_app from sanguo_api.ws import manager from sanguo_api.routes import get_orchestrator + from sanguo_api.auth import hash_password tmpdir = tempfile.mkdtemp() db_path = os.path.join(tmpdir, "test.db") try: - # Use pre-hashed bcrypt password to avoid compatibility issues - password_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyY9Wt3KpPqm" + # Generate real password hash using bcrypt directly + test_password = "password123" + password_hash = hash_password(test_password) # Create app to trigger wiring app = create_app(