73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
"""认证服务：登录业务 + JWT 签发/校验（HS256）+ 密码哈希/校验。
|
|
|
|
密码用 stdlib PBKDF2（免 bcrypt 依赖），存储格式自描述：
|
|
pbkdf2_sha256$600000$<salt_hex>$<digest_hex>
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import time
|
|
|
|
import jwt
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from config.settings import settings
|
|
from model.sys_user import SysUser
|
|
from repositories.sys_user import SysUserRepo
|
|
from utils.exceptions import ForbiddenError, ParamError
|
|
|
|
_ITERATIONS = 600_000
|
|
|
|
|
|
# ---- JWT ----
|
|
def create_token(user_id: int, *, extra: dict | None = None) -> str:
|
|
"""签发 token：仅存 user_id + iat/exp，用户信息校验时按 id 回查 sys_user。"""
|
|
now = int(time.time())
|
|
payload = {"user_id": user_id, "iat": now, "exp": now + settings.jwt.expire_seconds}
|
|
if extra:
|
|
payload.update(extra)
|
|
return jwt.encode(payload, settings.jwt.secret, algorithm=settings.jwt.algorithm)
|
|
|
|
|
|
def decode_token(token: str) -> dict:
|
|
"""校验签名 + 过期，返回 claims；任何失败抛异常（调用方转 401）。"""
|
|
return jwt.decode(token, settings.jwt.secret, algorithms=[settings.jwt.algorithm])
|
|
|
|
|
|
# ---- 密码 ----
|
|
def hash_password(password: str) -> str:
|
|
salt = secrets.token_hex(16)
|
|
digest = hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), _ITERATIONS)
|
|
return f"pbkdf2_sha256${_ITERATIONS}${salt}${digest.hex()}"
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
_, iters, salt, digest = stored.split("$")
|
|
except ValueError:
|
|
return False
|
|
calc = hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), int(iters))
|
|
return hmac.compare_digest(calc.hex(), digest)
|
|
|
|
|
|
# ---- 登录业务（路由层只编排，不碰数据/逻辑） ----
|
|
def user_payload(u: SysUser) -> dict:
|
|
"""对外暴露的用户信息（不含密码等敏感字段）。"""
|
|
return {
|
|
"id": u.id,
|
|
"username": u.username,
|
|
"user_type": u.user_type,
|
|
"employee_role": u.employee_role,
|
|
"customer_level": u.customer_level,
|
|
"status": u.status,
|
|
}
|
|
|
|
|
|
async def login(db: AsyncSession, username: str, password: str) -> dict:
|
|
"""校验账号密码与状态，成功签发 token 并返回用户信息；失败抛业务异常。"""
|
|
user = await SysUserRepo(db).get_by_username(username)
|
|
if user is None or not verify_password(password, user.password_hash):
|
|
raise ParamError("用户名或密码错误")
|
|
if user.status != "正常":
|
|
raise ForbiddenError("账号状态异常，无法登录")
|
|
return {"token": create_token(user.id), "user": user_payload(user)} |