hankin
423baff73b
- Docker bridge 网络隔离（8000 端口封死） - Gunicorn 4 Worker 多进程 - Alembic 数据库迁移基线 - 日志轮转 20m×3 - JWT 密钥 + DB 密码 + CORS 收紧 - 3-2-1 备份链路（NAS + R740-B 冷备） - 连接池 pool_pre_ping + pool_recycle=3600
65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
安全模块：JWT 令牌签发/验证 + 密码哈希
|
|
使用 python-jose 进行 JWT 操作，bcrypt 直接进行密码哈希。
|
|
注意：passlib 已不再维护，与 bcrypt>=5.0 不兼容，故直接使用 bcrypt 库。
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import bcrypt
|
|
from jose import JWTError, jwt
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def hash_password(plain_password: str) -> str:
|
|
"""将明文密码哈希为 bcrypt 格式存储"""
|
|
return bcrypt.hashpw(
|
|
plain_password.encode("utf-8"), bcrypt.gensalt()
|
|
).decode("utf-8")
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""校验明文密码与数据库中的哈希值是否匹配"""
|
|
return bcrypt.checkpw(
|
|
plain_password.encode("utf-8"),
|
|
hashed_password.encode("utf-8"),
|
|
)
|
|
|
|
|
|
def create_access_token(
|
|
subject: str,
|
|
role: str,
|
|
expires_delta: timedelta | None = None,
|
|
) -> str:
|
|
"""
|
|
签发 JWT 访问令牌。
|
|
:param subject: 用户标识 (通常是 username 或 user_id)
|
|
:param role: 用户角色 (admin / user)，嵌入 claims 供前端和后端鉴权
|
|
:param expires_delta: 自定义过期时间，默认使用配置值
|
|
"""
|
|
expire = datetime.now(timezone.utc) + (
|
|
expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
)
|
|
payload = {
|
|
"sub": subject,
|
|
"role": role,
|
|
"exp": expire,
|
|
}
|
|
return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
|
|
|
|
|
|
def decode_access_token(token: str) -> dict | None:
|
|
"""
|
|
解码并验证 JWT 令牌。
|
|
:return: payload 字典，失败返回 None
|
|
"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token, settings.SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]
|
|
)
|
|
return payload
|
|
except JWTError:
|
|
return None
|