53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
"""
|
|
内部服务鉴权：验证来自 java-mock 的请求签名。
|
|
|
|
签名算法（与 java-mock/middleware/internalSign.js 保持一致）：
|
|
message = "java-gateway:{userId}:{毫秒时间戳}"
|
|
signature = HMAC-SHA256(INTERNAL_SERVICE_SECRET, message) # hex
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import time
|
|
|
|
# 允许的时钟偏差（毫秒），防重放攻击
|
|
_ALLOWED_SKEW_MS = 5 * 60 * 1000 # 5 分钟
|
|
|
|
|
|
def verify_internal_request(headers) -> bool:
|
|
"""
|
|
验证内部服务请求签名。
|
|
未配置 INTERNAL_SERVICE_SECRET 时直接放行（开发/测试环境兼容）。
|
|
"""
|
|
secret = os.environ.get("INTERNAL_SERVICE_SECRET", "")
|
|
if not secret:
|
|
return True
|
|
|
|
service = headers.get("X-Internal-Service", "")
|
|
user_id = headers.get("X-Internal-User-Id", "")
|
|
timestamp = headers.get("X-Internal-Timestamp", "")
|
|
signature = headers.get("X-Internal-Signature", "")
|
|
|
|
# 基础字段校验
|
|
if service != "java-gateway" or not user_id or not timestamp or not signature:
|
|
return False
|
|
|
|
# 时间窗口校验（防重放）
|
|
try:
|
|
ts_ms = int(timestamp)
|
|
if abs(int(time.time() * 1000) - ts_ms) > _ALLOWED_SKEW_MS:
|
|
return False
|
|
except ValueError:
|
|
return False
|
|
|
|
# 重新计算签名，使用常量时间比较防时序攻击
|
|
message = f"java-gateway:{user_id}:{timestamp}"
|
|
expected = hmac.new(
|
|
secret.encode("utf-8"),
|
|
message.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
return hmac.compare_digest(expected, signature)
|