100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
|
"""轻量 access token 签发与校验。
|
|||
|
|
|
|||
|
|
说明：
|
|||
|
|
- 当前实现不是标准 JWT（避免引入额外依赖），而是“base64url(payload) + HMAC 签名”的轻量令牌。
|
|||
|
|
- 适用于内部系统的最小化认证需求；如需与第三方兼容，可替换为 JWT。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import base64
|
|||
|
|
import hashlib
|
|||
|
|
import hmac
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import time
|
|||
|
|
from dataclasses import dataclass
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(frozen=True)
|
|||
|
|
class TokenPayload:
|
|||
|
|
"""解析后的 token 载荷。"""
|
|||
|
|
|
|||
|
|
user_id: str
|
|||
|
|
username: str
|
|||
|
|
role_id: str
|
|||
|
|
exp: int
|
|||
|
|
iat: int
|
|||
|
|
|
|||
|
|
|
|||
|
|
def issue_access_token(*, user_id: str, username: str, role_id: str, expires_in_seconds: int = 3600) -> str:
|
|||
|
|
"""签发 access token。"""
|
|||
|
|
now = int(time.time())
|
|||
|
|
payload = {
|
|||
|
|
"sub": user_id,
|
|||
|
|
"username": username,
|
|||
|
|
"role_id": role_id,
|
|||
|
|
"iat": now,
|
|||
|
|
"exp": now + int(expires_in_seconds),
|
|||
|
|
"v": 1,
|
|||
|
|
}
|
|||
|
|
payload_bytes = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|||
|
|
payload_b64 = _b64url_encode(payload_bytes)
|
|||
|
|
sig = _sign(payload_b64.encode("ascii"))
|
|||
|
|
sig_b64 = _b64url_encode(sig)
|
|||
|
|
return f"v1.{payload_b64}.{sig_b64}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def verify_access_token(token: str) -> TokenPayload:
|
|||
|
|
"""校验 access token 并返回载荷。
|
|||
|
|
|
|||
|
|
Raises:
|
|||
|
|
ValueError: token 非法或已过期。
|
|||
|
|
"""
|
|||
|
|
if not token or not isinstance(token, str):
|
|||
|
|
raise ValueError("invalid token")
|
|||
|
|
parts = token.split(".")
|
|||
|
|
if len(parts) != 3 or parts[0] != "v1":
|
|||
|
|
raise ValueError("invalid token")
|
|||
|
|
payload_b64, sig_b64 = parts[1], parts[2]
|
|||
|
|
expected_sig = _sign(payload_b64.encode("ascii"))
|
|||
|
|
actual_sig = _b64url_decode(sig_b64)
|
|||
|
|
if not hmac.compare_digest(expected_sig, actual_sig):
|
|||
|
|
raise ValueError("invalid token")
|
|||
|
|
payload_raw = _b64url_decode(payload_b64)
|
|||
|
|
payload = json.loads(payload_raw.decode("utf-8"))
|
|||
|
|
exp = int(payload.get("exp"))
|
|||
|
|
if int(time.time()) >= exp:
|
|||
|
|
raise ValueError("token expired")
|
|||
|
|
return TokenPayload(
|
|||
|
|
user_id=str(payload.get("sub")),
|
|||
|
|
username=str(payload.get("username")),
|
|||
|
|
role_id=str(payload.get("role_id")),
|
|||
|
|
exp=exp,
|
|||
|
|
iat=int(payload.get("iat")),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def access_token_secret() -> bytes:
|
|||
|
|
"""获取 token 签名密钥（来自环境变量 SMARTEDT_AUTH_SECRET）。"""
|
|||
|
|
secret = os.getenv("SMARTEDT_AUTH_SECRET", "").strip()
|
|||
|
|
if not secret:
|
|||
|
|
secret = "smartedt-dev-secret-change-me"
|
|||
|
|
return secret.encode("utf-8")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _sign(message: bytes) -> bytes:
|
|||
|
|
"""对 message 做 HMAC-SHA256 签名。"""
|
|||
|
|
return hmac.new(access_token_secret(), message, hashlib.sha256).digest()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _b64url_encode(data: bytes) -> str:
|
|||
|
|
"""编码为不带 padding 的 base64url 字符串。"""
|
|||
|
|
return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _b64url_decode(value: str) -> bytes:
|
|||
|
|
"""解码不带 padding 的 base64url 字符串。"""
|
|||
|
|
padded = value + "=" * (-len(value) % 4)
|
|||
|
|
return base64.urlsafe_b64decode(padded.encode("ascii"))
|