fix(api): auth 用 bcrypt 直调替代 passlib(修复容器 __about__ 缺失致登录失败)

- 替换 passlib.context.CryptContext 为直接 bcrypt 调用
- hash_password: bcrypt.hashpw + gensalt
- verify_password: bcrypt.checkpw + 异常处理
- 保持公共接口不变(hash_password/verify_password/create_token/verify_token)
- 移除 passlib 导入,直接使用 bcrypt 模块
- 现有 $2b$12$... bcrypt hash 仍可验证通过

修复问题:
- passlib 1.7.4 探测 bcrypt.__about__.__version__ 导致 AttributeError
- 现代 bcrypt 移除了 __about__ 属性
- 致使容器内密码验证失败,登录跳过

验证结果:
- 本地 test_auth.py: 3/3 PASS
- 容器 test_auth.py: 3/3 PASS
- 容器 smoke JWT LOGIN: PASS(之前 SKIP)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2026-07-06 19:55:06 +08:00
parent effd3caa4a
commit 4723554e29
2 changed files with 44 additions and 43 deletions
+6 -4
View File
@@ -3,11 +3,10 @@
import os import os
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
import jwt import jwt
from passlib.context import CryptContext import bcrypt
from fastapi import HTTPException, status from fastapi import HTTPException, status
_CONFIG = {"secret": "change-me", "expire_minutes": 60, "algorithm": "HS256"} _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"): 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: 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: 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: def create_token(username: str) -> str:
+38 -39
View File
@@ -35,7 +35,7 @@ def test_jwt_login():
try: try:
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sanguo_api.app import create_app from sanguo_api.app import create_app
from unittest.mock import patch from sanguo_api.auth import hash_password
# Create temporary directory for test # Create temporary directory for test
tmpdir = tempfile.mkdtemp() tmpdir = tempfile.mkdtemp()
@@ -43,9 +43,9 @@ def test_jwt_login():
file_dir = tmpdir file_dir = tmpdir
try: try:
# Use pre-hashed bcrypt password to avoid compatibility issues # Generate real password hash using bcrypt directly
# Password: "password123" hashed with bcrypt test_password = "password123"
password_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyY9Wt3KpPqm" password_hash = hash_password(test_password)
# Create app with auth config # Create app with auth config
app = create_app( app = create_app(
@@ -60,28 +60,26 @@ def test_jwt_login():
max_workers=1 max_workers=1
) )
# Mock the password verification to avoid bcrypt compatibility issues # Test login with TestClient - REAL password verification
with patch('sanguo_api.auth.verify_password', return_value=True): client = TestClient(app)
# Test login with TestClient response = client.post("/api/v1/auth/login", json={
client = TestClient(app) "username": "admin",
response = client.post("/api/v1/auth/login", json={ "password": test_password
"username": "admin", })
"password": "password123"
})
assert response.status_code == 200, f"Expected 200, got {response.status_code}" assert response.status_code == 200, f"Expected 200, got {response.status_code}"
data = response.json() data = response.json()
assert "token" in data, "Token not in response" assert "token" in data, "Token not in response"
print(f" Login successful, token: {data['token'][:20]}...") print(f" Login successful, token: {data['token'][:20]}...")
print("=== PASS: JWT login ===") print("=== PASS: JWT login ===")
return data['token'] return data['token']
finally: finally:
shutil.rmtree(tmpdir, ignore_errors=True) shutil.rmtree(tmpdir, ignore_errors=True)
except Exception as e: except Exception as e:
print(f" === SKIP: JWT login test failed due to bcrypt compatibility: {type(e).__name__} ===") print(f" === FAIL: JWT login test failed: {type(e).__name__}: {e} ===")
print(" This is a known issue with passlib/bcrypt compatibility in container environment") import traceback
print(" Core JWT functionality is tested in local environment") traceback.print_exc()
print("=== SKIP: JWT login ===") print("=== FAIL: JWT login ===")
return None return None
@@ -91,15 +89,16 @@ def test_protected_route_auth():
try: try:
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sanguo_api.app import create_app from sanguo_api.app import create_app
from unittest.mock import patch from sanguo_api.auth import hash_password
tmpdir = tempfile.mkdtemp() tmpdir = tempfile.mkdtemp()
db_path = os.path.join(tmpdir, "test.db") db_path = os.path.join(tmpdir, "test.db")
file_dir = tmpdir file_dir = tmpdir
try: try:
# Use pre-hashed bcrypt password to avoid compatibility issues # Generate real password hash using bcrypt directly
password_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyY9Wt3KpPqm" test_password = "password123"
password_hash = hash_password(test_password)
app = create_app( app = create_app(
db_path=db_path, 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}" assert response.status_code == 401, f"Expected 401 without token, got {response.status_code}"
print(" No token: 401 Unauthorized ✓") print(" No token: 401 Unauthorized ✓")
# Mock the password verification to avoid bcrypt compatibility issues # Get valid token using REAL password verification
with patch('sanguo_api.auth.verify_password', return_value=True): login_response = client.post("/api/v1/auth/login", json={
# Get valid token "username": "admin",
login_response = client.post("/api/v1/auth/login", json={ "password": test_password
"username": "admin", })
"password": "password123" token = login_response.json()["token"]
})
token = login_response.json()["token"]
# Test with token but non-existent task - should get 404 # Test with token but non-existent task - should get 404
response = client.get( response = client.get(
@@ -140,10 +137,10 @@ def test_protected_route_auth():
finally: finally:
shutil.rmtree(tmpdir, ignore_errors=True) shutil.rmtree(tmpdir, ignore_errors=True)
except Exception as e: except Exception as e:
print(f" === SKIP: Protected route auth test failed due to bcrypt compatibility: {type(e).__name__} ===") print(f" === FAIL: Protected route auth test failed: {type(e).__name__}: {e} ===")
print(" This is a known issue with passlib/bcrypt compatibility in container environment") import traceback
print(" Core auth functionality is tested in local environment") traceback.print_exc()
print("=== SKIP: protected route auth ===") print("=== FAIL: protected route auth ===")
async def test_orchestrator_async(): 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.app import create_app
from sanguo_api.ws import manager from sanguo_api.ws import manager
from sanguo_api.routes import get_orchestrator from sanguo_api.routes import get_orchestrator
from sanguo_api.auth import hash_password
tmpdir = tempfile.mkdtemp() tmpdir = tempfile.mkdtemp()
db_path = os.path.join(tmpdir, "test.db") db_path = os.path.join(tmpdir, "test.db")
try: try:
# Use pre-hashed bcrypt password to avoid compatibility issues # Generate real password hash using bcrypt directly
password_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyY9Wt3KpPqm" test_password = "password123"
password_hash = hash_password(test_password)
# Create app to trigger wiring # Create app to trigger wiring
app = create_app( app = create_app(