code/ai_finance_agent/backend/app/infrastructure/security.py

"""密码使用随机盐 scrypt；访问令牌和刷新凭据职责分离以支持退出撤销。"""
import hashlib
import hmac
import re
import secrets
from datetime import timedelta
import jwt
from sqlalchemy import select
from app.domain.models import RefreshSession, Role, User
from app.shared.common import BusinessError, utcnow
from app.shared.config import settings


def hash_password(password: str) -> str:
    """内存硬化哈希降低离线暴力破解成本，不存可逆密码。"""
    salt = secrets.token_bytes(16)
    key = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1)
    return "scrypt$" + salt.hex() + "$" + key.hex()


def validate_password(password: str):
    if len(password) < 10 or not all(re.search(pattern, password) for pattern in (r"[A-Z]", r"[a-z]", r"\d", r"[^A-Za-z0-9]")):
        raise BusinessError("密码至少10位，包含大小写字母、数字和符号", code="WEAK_PASSWORD")


def check_password(password: str, encoded: str) -> bool:
    """恒定时间比较避免因哈希首个匹配位置不同产生时间侧信道。"""
    try:
        kind, salt, key = encoded.split("$")
        actual = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), n=16384, r=8, p=1)
        return kind == "scrypt" and hmac.compare_digest(actual.hex(), key)
    except (ValueError, TypeError):
        return False


async def public_user(session, user: User) -> dict:
    role = await session.get(Role, user.role_id)
    return {"id": user.id, "username": user.username, "real_name": user.real_name,
            "department_id": user.department_id, "role": role.code, "active": user.active}


async def issue_tokens(session, user: User) -> dict:
    """刷新令牌只保存哈希；轮换在同事务中撤销旧凭据并创建新凭据。"""
    from datetime import timezone
    now = utcnow()
    expires = now + timedelta(minutes=settings.access_minutes)
    access = jwt.encode({"sub": str(user.id), "type": "access", "jti": secrets.token_hex(16),
                         "iat": int(now.replace(tzinfo=timezone.utc).timestamp()),
                         "exp": int(expires.replace(tzinfo=timezone.utc).timestamp())}, settings.secret(), algorithm="HS256")
    refresh = secrets.token_urlsafe(48)
    session.add(RefreshSession(user_id=user.id, token_hash=hashlib.sha256(refresh.encode()).hexdigest(),
                               expires_at=now + timedelta(days=settings.refresh_days)))
    return {"access_token": access, "refresh_token": refresh, "expires_in": settings.access_minutes * 60,
            "token_type": "Bearer", "user": await public_user(session, user)}


def decode_access(token: str) -> int:
    """只接受访问类型和指定算法，不能用刷新凭据冒充访问令牌。"""
    try:
        payload = jwt.decode(token, settings.secret(), algorithms=["HS256"], options={"require": ["sub", "exp", "type"]})
        if payload["type"] != "access":
            raise ValueError()
        return int(payload["sub"])
    except (jwt.PyJWTError, ValueError, KeyError):
        raise BusinessError("登录已失效，请重新登录", 401, "TOKEN_INVALID")


async def refresh_record(session, raw: str):
    item = await session.scalar(select(RefreshSession).where(RefreshSession.token_hash == hashlib.sha256(raw.encode()).hexdigest()))
    if not item or item.revoked_at or item.expires_at <= utcnow():
        raise BusinessError("刷新会话已失效", 401, "REFRESH_INVALID")
    return item