74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""JWT 访问令牌：签发与验签。
|
|||
|
|
|
|||
|
|
- 无状态验签（HS256），claims 携带角色/组织/数据范围/权限。
|
|||
|
|
- 撤销由 ``sessions`` 表（按 jti）+ ``users.token_version`` 兜底，
|
|||
|
|
``get_current_user`` 每请求一次 DB 读校验。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import uuid
|
|||
|
|
from datetime import datetime, timedelta, timezone
|
|||
|
|
|
|||
|
|
import jwt as pyjwt
|
|||
|
|
|
|||
|
|
from . import config
|
|||
|
|
|
|||
|
|
JTI_TTL_SECONDS = config.JWT_ACCESS_EXPIRY_SECONDS
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _now_ts() -> int:
|
|||
|
|
return int(datetime.now(timezone.utc).timestamp())
|
|||
|
|
|
|||
|
|
|
|||
|
|
def create_access_token(
|
|||
|
|
user: dict,
|
|||
|
|
*,
|
|||
|
|
permissions: list[str],
|
|||
|
|
scope_region_ids: list[str],
|
|||
|
|
scope_level: str | None,
|
|||
|
|
expiry_seconds: int,
|
|||
|
|
identity_id: str | None = None,
|
|||
|
|
) -> tuple[str, str, str]:
|
|||
|
|
"""签发 JWT，返回 (token, jti, expires_at_iso)。
|
|||
|
|
|
|||
|
|
``user`` 为来自 UserRepository 的用户 dict（含 id/username/role/
|
|||
|
|
sub_role/org_id/region_id/token_version）。``identity_id`` 非空时表示
|
|||
|
|
该令牌绑定到指定端口身份（get_current_user 据此解析当前身份）。
|
|||
|
|
"""
|
|||
|
|
jti = str(uuid.uuid4())
|
|||
|
|
now = _now_ts()
|
|||
|
|
exp_ts = now + (expiry_seconds if expiry_seconds else config.JWT_ACCESS_EXPIRY_SECONDS)
|
|||
|
|
payload = {
|
|||
|
|
"sub": user["id"],
|
|||
|
|
"username": user.get("username", ""),
|
|||
|
|
"role": user.get("role", "opc_member"),
|
|||
|
|
"sub_role": user.get("sub_role"),
|
|||
|
|
"org_id": user.get("org_id"),
|
|||
|
|
"region_id": user.get("region_id"),
|
|||
|
|
"identity_id": identity_id,
|
|||
|
|
"scope_level": scope_level,
|
|||
|
|
"scope_region_ids": scope_region_ids,
|
|||
|
|
"perms": permissions,
|
|||
|
|
"ver": user.get("token_version", 0),
|
|||
|
|
"jti": jti,
|
|||
|
|
"iat": now,
|
|||
|
|
"exp": exp_ts,
|
|||
|
|
}
|
|||
|
|
token = pyjwt.encode(payload, config.JWT_SECRET, algorithm=config.JWT_ALGORITHM)
|
|||
|
|
expires_at = datetime.fromtimestamp(exp_ts, tz=timezone.utc).isoformat()
|
|||
|
|
return token, jti, expires_at
|
|||
|
|
|
|||
|
|
|
|||
|
|
def decode_access_token(token: str) -> dict | None:
|
|||
|
|
"""验签并返回 claims；无效/过期返回 None。"""
|
|||
|
|
try:
|
|||
|
|
payload = pyjwt.decode(
|
|||
|
|
token,
|
|||
|
|
config.JWT_SECRET,
|
|||
|
|
algorithms=[config.JWT_ALGORITHM],
|
|||
|
|
)
|
|||
|
|
except pyjwt.PyJWTError:
|
|||
|
|
return None
|
|||
|
|
return payload
|