python端登录完成
This commit is contained in:
@@ -4,6 +4,7 @@ JWT 工具:解析 Java 端签发的 token、从请求中提取 token。
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from flask import request
|
||||
@@ -25,6 +26,14 @@ except ImportError as _e:
|
||||
_DEFAULT_SECRET = "please-change-this-secret-please-rotate-at-least-32-bytes"
|
||||
COOKIE_NAME = os.getenv("AIIMAGE_AUTH_COOKIE_NAME", "aiimage_token")
|
||||
|
||||
# 桌面端与 Java 服务器时钟可能漂移(笔记本休眠、用户手动改时间、跨时区等),
|
||||
# 给 JWT exp/nbf 校验留出容差,避免 /api/auth/sync 因为时间不同步而 401。
|
||||
# 默认 5 分钟,可通过 AIIMAGE_JWT_LEEWAY_SECONDS 覆盖。
|
||||
try:
|
||||
_JWT_LEEWAY_SECONDS = int(os.getenv("AIIMAGE_JWT_LEEWAY_SECONDS", "300"))
|
||||
except ValueError:
|
||||
_JWT_LEEWAY_SECONDS = 300
|
||||
|
||||
|
||||
def _signing_key() -> bytes:
|
||||
"""与 Java JwtService.signingKey 保持一致:UTF-8 字节,不足 32 字节右侧补 0。"""
|
||||
@@ -35,6 +44,26 @@ def _signing_key() -> bytes:
|
||||
return key_bytes
|
||||
|
||||
|
||||
def _describe_clock_skew(token: str) -> str:
|
||||
"""解析 token 里的 iat/exp,与本地时钟比较,返回 ' iat=.. exp=.. now=.. skew=..s' 字符串。
|
||||
仅供日志使用,不做安全决策;解析失败返回空串,不影响主流程。"""
|
||||
if pyjwt is None:
|
||||
return ""
|
||||
try:
|
||||
unverified = pyjwt.decode(token, options={"verify_signature": False, "verify_exp": False, "verify_iat": False, "verify_nbf": False})
|
||||
except Exception:
|
||||
return ""
|
||||
iat = unverified.get("iat")
|
||||
exp = unverified.get("exp")
|
||||
now = int(time.time())
|
||||
parts = [f"now={now}"]
|
||||
if isinstance(iat, (int, float)):
|
||||
parts.append(f"iat={int(iat)} skew_iat={now - int(iat)}s")
|
||||
if isinstance(exp, (int, float)):
|
||||
parts.append(f"exp={int(exp)} skew_exp={now - int(exp)}s")
|
||||
return ", " + " ".join(parts)
|
||||
|
||||
|
||||
def parse_token_with_reason(token: str) -> Tuple[Optional[dict], Optional[str]]:
|
||||
"""解析 JWT,返回 (payload, error_reason)。
|
||||
payload 命中时 error_reason 为 None;失败时 payload 为 None,error_reason 描述根因。"""
|
||||
@@ -49,10 +78,17 @@ def parse_token_with_reason(token: str) -> Tuple[Optional[dict], Optional[str]]:
|
||||
return None, f"无法解析 token header: {type(e).__name__}: {e}"
|
||||
alg = header.get("alg") or "?"
|
||||
try:
|
||||
payload = pyjwt.decode(token, _signing_key(), algorithms=["HS256", "HS384", "HS512"])
|
||||
payload = pyjwt.decode(
|
||||
token,
|
||||
_signing_key(),
|
||||
algorithms=["HS256", "HS384", "HS512"],
|
||||
leeway=_JWT_LEEWAY_SECONDS,
|
||||
)
|
||||
except Exception as e:
|
||||
# 常见:ExpiredSignatureError / InvalidSignatureError / DecodeError / InvalidAlgorithmError
|
||||
return None, f"{type(e).__name__}: {e} (token alg={alg})"
|
||||
# 把 token 内 exp/iat 与本地时钟一起打出来,定位"时间漂移"类失败更直接
|
||||
skew_info = _describe_clock_skew(token)
|
||||
return None, f"{type(e).__name__}: {e} (token alg={alg}{skew_info})"
|
||||
sub = payload.get("sub")
|
||||
try:
|
||||
user_id = int(sub) if sub is not None else None
|
||||
|
||||
Reference in New Issue
Block a user