diff --git a/app/api/endpoints/user.py b/app/api/endpoints/user.py index 1b8fd466e..11828e7d7 100644 --- a/app/api/endpoints/user.py +++ b/app/api/endpoints/user.py @@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app import schemas from app.api.response import ResponseAPIRouter -from app.application.security.access import get_password_hash +from app.application.security.access import PasswordTooLongError, get_password_hash from app.db import get_async_db from app.db.models.user import User from app.api.deps import get_current_active_superuser_async, get_current_active_user_async, get_current_active_user @@ -42,7 +42,10 @@ async def create_user( return schemas.Response(success=False, message="用户已存在") user_info = user_in.model_dump() if user_info.get("password"): - user_info["hashed_password"] = get_password_hash(user_info["password"]) + try: + user_info["hashed_password"] = get_password_hash(user_info["password"]) + except PasswordTooLongError as error: + return schemas.Response(success=False, message=str(error)) user_info.pop("password") user = await User(**user_info).async_create(db) return schemas.Response(success=True if user else False) @@ -67,7 +70,10 @@ async def update_user( success=False, message="密码需要同时包含字母、数字、特殊字符中的至少两项，且长度大于6位", ) - user_info["hashed_password"] = get_password_hash(user_info["password"]) + try: + user_info["hashed_password"] = get_password_hash(user_info["password"]) + except PasswordTooLongError as error: + return schemas.Response(success=False, message=str(error)) user_info.pop("password") user = await current_user.async_get_by_id(db, user_id=user_info["id"]) user_name = user_info.get("name") diff --git a/app/application/security/access.py b/app/application/security/access.py index 5e86c2836..be72f1ce0 100644 --- a/app/application/security/access.py +++ b/app/application/security/access.py @@ -8,25 +8,43 @@ import traceback from datetime import timedelta from typing import Any, Union, Annotated, Optional, Callable +import bcrypt import jwt from Crypto.Cipher import AES from Crypto.Util.Padding import pad from cryptography.fernet import Fernet from fastapi import HTTPException, status, Security, Request, Response from fastapi.security import OAuth2PasswordBearer, APIKeyHeader, APIKeyQuery, APIKeyCookie, HTTPBearer -from passlib.context import CryptContext - from app import schemas from app.runtime.cache import cached from app.runtime.config import settings from app.runtime.log import logger -pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +BCRYPT_PASSWORD_MAX_BYTES = 72 +BCRYPT_ROUNDS = 12 ALGORITHM = "HS256" SuperuserTokenPayloadProvider = Callable[[], schemas.TokenPayload] _superuser_token_payload_provider: Optional[SuperuserTokenPayloadProvider] = None +class PasswordTooLongError(ValueError): + """密码的 UTF-8 字节长度超过 bcrypt 可安全处理的上限。""" + + +def _encode_bcrypt_password( + password: str, *, allow_legacy_truncation: bool = False +) -> bytes: + """编码 bcrypt 密码；仅验证既有哈希时允许按历史语义截断。""" + password_bytes = password.encode("utf-8") + if len(password_bytes) > BCRYPT_PASSWORD_MAX_BYTES: + if allow_legacy_truncation: + return password_bytes[:BCRYPT_PASSWORD_MAX_BYTES] + raise PasswordTooLongError( + f"密码 UTF-8 编码后不能超过 {BCRYPT_PASSWORD_MAX_BYTES} 字节" + ) + return password_bytes + + def set_superuser_token_payload_provider( provider: SuperuserTokenPayloadProvider, ) -> None: @@ -343,13 +361,24 @@ def verify_apikey(apikey: Annotated[str | None, Security(__get_api_key)]) -> str def verify_password(plain_password: str, hashed_password: str) -> bool: - """校验明文密码是否匹配已保存的密码摘要。""" - return pwd_context.verify(plain_password, hashed_password) + """验证既有 bcrypt 哈希，并保留超长历史密码的截断语义。""" + try: + return bcrypt.checkpw( + _encode_bcrypt_password( + plain_password, allow_legacy_truncation=True + ), + hashed_password.encode("ascii"), + ) + except (UnicodeEncodeError, ValueError): + return False def get_password_hash(password: str) -> str: - """生成适合持久化保存的密码摘要。""" - return pwd_context.hash(password) + """使用 $2b$ 前缀和 cost 12 生成可持久化的 bcrypt 密码哈希。""" + return bcrypt.hashpw( + _encode_bcrypt_password(password), + bcrypt.gensalt(rounds=BCRYPT_ROUNDS, prefix=b"2b"), + ).decode("ascii") def decrypt(data: bytes, key: bytes) -> Optional[bytes]: diff --git a/requirements.in b/requirements.in index f8803b682..21dc863c8 100644 --- a/requirements.in +++ b/requirements.in @@ -4,7 +4,6 @@ pydantic-settings>=2.14.2,<3.0.0 SQLAlchemy~=2.0.50 uvicorn~=0.49.0 fastapi~=0.136.3 -passlib~=1.7.4 PyJWT~=2.13.0 python-multipart~=0.0.32 aiofiles~=25.1.0 diff --git a/tests/test_password_hashing.py b/tests/test_password_hashing.py new file mode 100644 index 000000000..575d74d86 --- /dev/null +++ b/tests/test_password_hashing.py @@ -0,0 +1,127 @@ +import asyncio +from types import SimpleNamespace + +import bcrypt +import pytest + +from app.api.endpoints import user as user_endpoint +from app.application.security.access import ( + PasswordTooLongError, + get_password_hash, + verify_password, +) + + +PASSLIB_BCRYPT_HASH = "$2b$12$6QiVIML7x3T.F/p6cuFjLuMvFumE1V4OZpvhGVgCwaSoBE7lHlMle" + + +def test_password_hash_uses_existing_bcrypt_contract(): + """新密码保持 $2b$、cost 12，并可由同一包装正确验证。""" + hashed_password = get_password_hash("new-password") + + assert hashed_password.startswith("$2b$12$") + assert verify_password("new-password", hashed_password) is True + assert verify_password("wrong-password", hashed_password) is False + + +def test_verify_password_accepts_existing_passlib_bcrypt_hash(): + """不依赖 Passlib 时仍能验证既有 bcrypt 密码哈希。""" + assert verify_password("existing-passlib-password", PASSLIB_BCRYPT_HASH) is True + + +def test_get_password_hash_rejects_more_than_72_utf8_bytes(): + """bcrypt 不得静默截断 UTF-8 编码后超过 72 字节的新密码。""" + with pytest.raises(PasswordTooLongError, match="72 字节"): + get_password_hash("a" * 73) + + with pytest.raises(PasswordTooLongError, match="72 字节"): + get_password_hash("密" * 25) + + +def test_get_password_hash_accepts_exactly_72_utf8_bytes(): + """UTF-8 编码后恰好 72 字节的密码仍属于有效输入。""" + password = "密" * 24 + + assert verify_password(password, get_password_hash(password)) is True + + +def test_verify_password_preserves_legacy_long_password_access(): + """既有超长密码即使在多字节字符中间截断也应保持可登录。""" + password = "a" * 70 + "密" + hashed_password = bcrypt.hashpw( + password.encode("utf-8")[:72], bcrypt.gensalt(rounds=4) + ).decode("ascii") + + assert verify_password(password, hashed_password) is True + + +def test_verify_password_rejects_malformed_hash(): + """损坏的数据库哈希应按认证失败处理。""" + assert verify_password("password", "not-a-bcrypt-hash") is False + + +class _CreateUserInput: + """提供新增用户接口所需的最小输入契约。""" + + name = "new-user" + + @staticmethod + def model_dump(): + """返回包含超长多字节密码的用户数据。""" + return { + "name": "new-user", + "email": None, + "password": "Ab1!" + "密" * 23, + "is_active": True, + "is_superuser": False, + "avatar": None, + "is_otp": False, + "permissions": {}, + "settings": {}, + } + + +class _CurrentUser: + """提供用户接口长度校验前需要的最小查询契约。""" + + @staticmethod + async def async_get_by_name(_db, name): + """模拟用户名尚未被使用。""" + assert name == "new-user" + return None + + +def test_create_user_returns_business_error_for_password_over_72_bytes(): + """新增用户遇到超长密码时应返回可读业务错误。""" + response = asyncio.run( + user_endpoint.create_user( + db=SimpleNamespace(), + user_in=_CreateUserInput(), + current_user=_CurrentUser(), + ) + ) + + assert response.success is False + assert response.message == "密码 UTF-8 编码后不能超过 72 字节" + + +def test_update_user_returns_business_error_for_password_over_72_bytes(): + """修改用户遇到超长密码时应返回可读业务错误。""" + user_in = SimpleNamespace( + model_dump=lambda: { + "id": 1, + "name": "user", + "password": "Ab1!" + "密" * 23, + } + ) + + response = asyncio.run( + user_endpoint.update_user( + db=SimpleNamespace(), + user_in=user_in, + current_user=SimpleNamespace(), + ) + ) + + assert response.success is False + assert response.message == "密码 UTF-8 编码后不能超过 72 字节"