Update app.py

app.py

@@ -14,6 +14,8 @@ FSD Simulator 缓存中间层 (Hugging Face Space + Docker)
 import os
 import time
 import random
+import hashlib
+import hmac
 from typing import Any, Optional
 
 import httpx
@@ -30,6 +32,9 @@ CACHE_TTL = 12 * 60 * 60
 # 抖动范围 ±30 分钟，避免所有 key 同一刻过期造成"缓存雪崩"
 CACHE_JITTER = 30 * 60
 
+# 签名有效时间窗（秒）：容忍 Roblox 与服务端 5 分钟时钟差，同时限制重放窗口
+SIGNATURE_WINDOW = 300
+
 app = FastAPI(title="FSD Cache Layer")
 
 # ---------- 内存缓存（进程一重启就清空，所以只当加速层，不当唯一数据源）----------
@@ -113,11 +118,29 @@ def parse_rows(execute_response: dict) -> list[dict]:
     return rows_out
 
 
-# ---------- 鉴权：校验 Roblox 传来的口令 ----------
-def check_app_key(x_app_key: Optional[str]):
-    # 没设 APP_SECRET 就跳过校验（方便本地测试）；线上务必设置！
-    if APP_SECRET and x_app_key != APP_SECRET:
-        raise HTTPException(401, "X-App-Key 无效")
+# ---------- 鉴权：时间戳签名校验 ----------
+# Roblox 端发送两个请求头：
+#   X-Timestamp = 当前 Unix 秒（Luau 的 os.time()）
+#   X-Signature = sha256(APP_SECRET + X-Timestamp) 的十六进制小写
+# 好处：不再明文传 APP_SECRET；签名随时间变化，超出时间窗即失效。
+def check_signature(x_timestamp: Optional[str], x_signature: Optional[str]):
+    # 没设 APP_SECRET 就跳过（方便本地测试）；线上务必设置！
+    if not APP_SECRET:
+        return
+    if not x_timestamp or not x_signature:
+        raise HTTPException(401, "缺少 X-Timestamp 或 X-Signature")
+    try:
+        t = int(x_timestamp)
+    except ValueError:
+        raise HTTPException(401, "X-Timestamp 格式错误")
+    # 超出时间窗 → 视为过期或重放，拒绝
+    if abs(int(time.time()) - t) > SIGNATURE_WINDOW:
+        raise HTTPException(401, "签名已过期（检查两端时钟）")
+    # 用同样的 APP_SECRET + 原始时间戳字符串重算，必须和客户端逐字符一致
+    expected = hashlib.sha256((APP_SECRET + x_timestamp).encode()).hexdigest()
+    # 恒定时间比较，防时序攻击（timing attack）
+    if not hmac.compare_digest(expected, x_signature):
+        raise HTTPException(401, "签名无效")
 
 
 # ==================== 路由 ====================
@@ -141,10 +164,11 @@ class LogPayload(BaseModel):
 
 @app.post("/log")
 async def log_events(payload: LogPayload,
-                     x_app_key: Optional[str] = Header(default=None)):
+                     x_timestamp: Optional[str] = Header(default=None),
+                     x_signature: Optional[str] = Header(default=None)):
     """【写勤】Roblox 批量上报驾驶行为，直通写入 Turso（不缓存）。
     玩家数据必须尽快落库，不能压在缓存里等 12 小时——Space 一重启就丢了。"""
-    check_app_key(x_app_key)
+    check_signature(x_timestamp, x_signature)
     if not payload.events:
         return {"written": 0}
 
@@ -159,10 +183,11 @@ async def log_events(payload: LogPayload,
 
 @app.get("/stats")
 async def scene_stats(scene: str = Query(...),
-                      x_app_key: Optional[str] = Header(default=None)):
+                      x_timestamp: Optional[str] = Header(default=None),
+                      x_signature: Optional[str] = Header(default=None)):
     """【读懒】查某个场景下各操作的人气统计，结果缓存 12 小时。
     NPC 决策大量调这个，能命中缓存就绝不打扰 Turso。"""
-    check_app_key(x_app_key)
+    check_signature(x_timestamp, x_signature)
 
     cache_key = f"stats:{scene}"
     cached = cache_get(cache_key)