refactor: 替换密码哈希实现为原生bcrypt并更新依赖
重构CryptUtils类,替换原有的passlib密码哈希逻辑为直接使用bcrypt库;添加密码字节长度限制以符合bcrypt原生72字节上限;更新pyproject.toml配置,新增bcrypt依赖并将passlib设为可选依赖;统一代码缩进格式
This commit is contained in:
+39
-30
@@ -1,48 +1,57 @@
|
|||||||
from datetime import timedelta, datetime, timezone
|
from datetime import timedelta, datetime, timezone
|
||||||
from typing import TypedDict, Literal, TypeAlias
|
from typing import TypedDict, Literal, TypeAlias
|
||||||
|
|
||||||
|
import bcrypt
|
||||||
import jwt
|
import jwt
|
||||||
from passlib.context import CryptContext
|
|
||||||
|
|
||||||
from app.config.env import env
|
from app.config.env import env
|
||||||
|
|
||||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
BCRYPT_MAX_BYTES = 72
|
||||||
|
|
||||||
# token的类型,access用于接口认证,refresh用于刷新access token,verify用于激活用户账号
|
|
||||||
AccessTokenType: TypeAlias = Literal["access", "refresh", "verify", "api"]
|
AccessTokenType: TypeAlias = Literal["access", "refresh", "verify", "api"]
|
||||||
|
|
||||||
|
|
||||||
class TokenInfo(TypedDict):
|
class TokenInfo(TypedDict):
|
||||||
# 用户名信息
|
username: str
|
||||||
username: str
|
exp: datetime
|
||||||
# token过期时间
|
type: AccessTokenType
|
||||||
exp: datetime
|
|
||||||
type: AccessTokenType
|
|
||||||
|
def _safe_password_bytes(password: str) -> bytes:
|
||||||
|
data = password.encode("utf-8")
|
||||||
|
if len(data) > BCRYPT_MAX_BYTES:
|
||||||
|
data = data[:BCRYPT_MAX_BYTES]
|
||||||
|
return data
|
||||||
|
|
||||||
|
|
||||||
class CryptUtils:
|
class CryptUtils:
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_password_hash(password: str):
|
def get_password_hash(password: str):
|
||||||
return pwd_context.hash(password)
|
return bcrypt.hashpw(_safe_password_bytes(password), bcrypt.gensalt()).decode(
|
||||||
|
"utf-8"
|
||||||
|
)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def verify_password(plain_password: str, hashed_password: str):
|
def verify_password(plain_password: str, hashed_password: str):
|
||||||
return pwd_context.verify(plain_password, hashed_password)
|
try:
|
||||||
|
return bcrypt.checkpw(
|
||||||
|
_safe_password_bytes(plain_password), hashed_password.encode("utf-8")
|
||||||
|
)
|
||||||
|
except ValueError:
|
||||||
|
return False
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def create_token(
|
def create_token(username: str, type: AccessTokenType, expires_delta: timedelta):
|
||||||
username: str,
|
data: TokenInfo = {
|
||||||
type: AccessTokenType,
|
"username": username,
|
||||||
expires_delta: timedelta
|
"type": type,
|
||||||
):
|
"exp": datetime.now(timezone.utc) + expires_delta,
|
||||||
data: TokenInfo = {
|
}
|
||||||
"username": username,
|
return jwt.encode(data, env.jwt_secret_key, env.jwt_algorithm)
|
||||||
"type": type,
|
|
||||||
"exp": datetime.now(timezone.utc) + expires_delta
|
|
||||||
}
|
|
||||||
return jwt.encode(data, env.jwt_secret_key, env.jwt_algorithm)
|
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_token_info(token: str) -> TokenInfo:
|
def get_token_info(token: str) -> TokenInfo:
|
||||||
data = jwt.decode(token, env.jwt_secret_key, algorithms=[env.jwt_algorithm])
|
data = jwt.decode(token, env.jwt_secret_key, algorithms=[env.jwt_algorithm])
|
||||||
return data
|
return data
|
||||||
|
|||||||
+2
-1
@@ -20,7 +20,8 @@ psutil = "^7.0.0"
|
|||||||
asyncmy = "^0.2.10"
|
asyncmy = "^0.2.10"
|
||||||
sqlmodel = "^0.0.24"
|
sqlmodel = "^0.0.24"
|
||||||
greenlet = "^3.2.3"
|
greenlet = "^3.2.3"
|
||||||
passlib = {extras = ["bcrypt"], version = "^1.7.4"}
|
passlib = {extras = ["bcrypt"], version = "^1.7.4", optional = true}
|
||||||
|
bcrypt = "^4.0.0"
|
||||||
pyjwt = "^2.10.1"
|
pyjwt = "^2.10.1"
|
||||||
python-multipart = "^0.0.20"
|
python-multipart = "^0.0.20"
|
||||||
langgraph = "^0.6.3"
|
langgraph = "^0.6.3"
|
||||||
|
|||||||
Reference in New Issue
Block a user