document-management/backend/dms/security/jwt_tokens.py

"""JWT 签发与严格校验。"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from uuid import uuid4

import jwt
from flask import current_app

from dms.common.errors import AuthenticationError, InternalError, TokenExpiredError
from dms.models import User


ALGORITHM = "HS256"
REQUIRED_CLAIMS = ("sub", "username", "roleCode", "authVersion", "iat", "exp", "jti", "iss")


def _secret() -> str:
    secret = str(current_app.config.get("DMS_JWT_SECRET", "")).strip()
    if not secret:
        raise InternalError("JWT签名密钥未配置")
    return secret


def issue_access_token(user: User, *, keep_signed_in: bool) -> tuple[str, int]:
    now = datetime.now(timezone.utc)
    duration_key = (
        "DMS_JWT_KEEP_SIGNED_IN_EXPIRES_SECONDS"
        if keep_signed_in
        else "DMS_JWT_EXPIRES_SECONDS"
    )
    expires_in = int(current_app.config[duration_key])
    claims = {
        "sub": str(user.id),
        "username": user.username,
        "roleCode": user.role_code,
        "authVersion": user.auth_version,
        "iat": now,
        "exp": now + timedelta(seconds=expires_in),
        "jti": str(uuid4()),
        "iss": current_app.config["DMS_JWT_ISSUER"],
    }
    return jwt.encode(claims, _secret(), algorithm=ALGORITHM), expires_in


def decode_access_token(token: str) -> dict[str, object]:
    try:
        return jwt.decode(
            token,
            _secret(),
            algorithms=[ALGORITHM],
            issuer=current_app.config["DMS_JWT_ISSUER"],
            options={"require": list(REQUIRED_CLAIMS)},
        )
    except jwt.ExpiredSignatureError as exc:
        raise TokenExpiredError() from exc
    except jwt.PyJWTError as exc:
        raise AuthenticationError() from exc