Pine
26fa546f67
- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
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
|