fix: webhook 加固 + 异步任务/JSON 解析健壮性 (monorepo 同迁) (#1)

* chore(service): update paths after monorepo colocation

* chore(config): extra=forbid + 新增 WEBHOOK_HMAC_SECRET

- SettingsConfigDict.extra 从 "ignore" 改成 "forbid"，typo 的 env key
  启动就报错，而不是静默跑错配置
- 新增可选 WEBHOOK_HMAC_SECRET（SecretStr | None），给 /alert/flagged
  做可选 HMAC 签名校验用；_empty_to_none validator 一并覆盖

* fix(alerts): webhook X-Internal-Key 用常量时间比较 + 可选 HMAC 签名

- X-Internal-Key 原本是裸 != 比较，换 hmac.compare_digest 防 timing
  attack 通过响应时间猜前缀
- 新增 X-Signature: sha256=<hex> 校验层，key 是 WEBHOOK_HMAC_SECRET。
  向后兼容：没配这把密钥就跳过（启动时 warn 一次），配了就强校验，
  缺失/格式错/不匹配统一 401
- HMAC 对 raw bytes 算，先 req.read() 再 parse JSON，避免序列化漂移
- bind 仍然 0.0.0.0（Docker container → host 的既有需求不动）
- 补 4 个 HMAC 分支 test，_FakeSettings 加 hmac_secret 参数

* fix(listener): 后台轮询任务包一层 _safe，异常不再静默丢

asyncio.create_task 原本裸调 self._notify_final_status，task 抛异常
后 future 没人 await，log 里看不到任何东西。包 _safe(..., name=...)
把 exception 打到 structlog，并给 create_task 传 name=poll_<id> 方便
在 debugger / asyncio.all_tasks() 里定位。

* fix(api_client): resp.json() 包 try/except 防后端吐非 JSON

后端偶尔挂掉的时候会返回 gateway HTML 或纯文本错误页，原代码直接
resp.json() 会抛 JSONDecodeError/ValueError，上层没接住就冒泡成裸异常。
新增 _safe_json() helper，统一：
- warn 级别记一条含 status + body 前 200 字符的日志
- 转抛 InternalAPIError，上层既有的 try/except 分支能兜住

submit_internal / fetch_link / fetch_summary 三处 resp.json() 全部过一遍。
httpx.AsyncClient 每次 new 的复用问题是单独的 perf 项，这 PR 不动。

* revert(config): extra 改回 ignore，避免 systemd 启动期集体 validation error

上一版把 extra 改成 forbid 想做 typo 检测，但 ChatBot 复用的是
/home/ubuntu/involution-hell-project/backend/.env——里面有 ~30 个
backend-only 的 key（PG*, SPRING_*, GITHUB_TOKEN, OPENAI_*, GA4_* 等），
pydantic 会在 Settings() 实例化时抛一堆 validation error，bot 根本启动
不起来。改回 ignore，保持 backward compat。

WEBHOOK_HMAC_SECRET 字段和 _empty_to_none validator 保留不动。
未来若把 ChatBot 切到独立 .env（不再共用 backend 那份），可以重新开
forbid 吃 typo 检测红利。

* chore(listener): 给 _safe 的 coro 参数补全 type hint

_safe(coro, *, name) 原本 coro 是裸参数，mypy / reviewer 一眼看不出意图。
改成 Coroutine[Any, Any, Any] 明确表达"接一个 awaitable coroutine"。
另外在 docstring 里补一句说明 except Exception 为什么不会吞
CancelledError（3.12 起继承自 BaseException），避免后续 reviewer
再踩同一个坑。

* test(alerts): 加一个鉴权顺序 case——HMAC 对但 key 错应返回 403

当前 _handle_flagged 的鉴权分两层：先 X-Internal-Key，再 HMAC 签名。
之前的 test 没覆盖"签名对、key 错"这条路径，reviewer 指出缺了 auth
precedence 的文档化。补一条：签名合法但 X-Internal-Key=WRONG-KEY
必须返回 403（走 key 校验的分支）而不是 401（HMAC 分支）。

* chore(alerts): HMAC 未配置时的 warning 文案改得更 actionable

之前的 "建议后端上线签名后尽快补上这把密钥" 不够清楚——没说明这是
过渡模式，也没说要做什么具体动作。改成：
- 明确只做 X-Internal-Key 校验、不验签
- 说明这是 backward-compat / 过渡模式
- 指出开启姿势：backend 上线签名后把密钥同步过来即可

log level 保持 warning（不是 error）：optional HMAC 是设计如此，给
backend 分阶段 rollout 留窗口。

* docs(alerts,config): 纠正 bind address 注释——实际是 0.0.0.0 不是 127.0.0.1

两处 docstring/注释都还在说 "只绑 127.0.0.1 / loopback 内网"，但上一轮
commit 011a8e3 已经把 bind 改成 0.0.0.0（让 docker bridge 里的 backend
能打过来）。reviewer 看到文档和代码对不上会以为是 bug。

- alerts.py 模块 docstring：说明绑 0.0.0.0 的原因（Docker bridge），
  把三层防护（X-Internal-Key / 可选 HMAC / 上游 firewall）列清楚
- config.py chatbot_alert_port 字段注释同步修掉"只绑 127.0.0.1"

不改行为，只改文档。

* fix(api_client): _safe_json body preview 走 bytes 切片，不再整体 decode

之前 preview = resp.text[:200]，resp.text 会对整个 body 做 UTF-8 decode
再切——对大 body 是白白多跑一遍，对含非 UTF-8 字节的 gateway 错误页还
可能在错误处理路径里再抛一次 UnicodeDecodeError，把原始 JSONDecodeError
的栈盖掉。

改成 resp.content[:200].decode("utf-8", errors="replace")：只切前 200
字节，坏字节走 U+FFFD fallback，日志里能看清到底是什么东西吐回来。
行为语义不变，上层 InternalAPIError 照常抛。

* fix(service): 把 uv cache 指到项目内，绕开 ProtectHome=read-only

* fix(alerts): 把 log.warning 的 msg= 改成 note=，避开 LogRecord 保留字段冲突

* docs(api_client,tests): 把 _safe_json 和 HMAC 分支纳入 docstring

api_client: submit_internal / fetch_link / fetch_summary 三个 caller 现在
走 _safe_json，2xx 但 body 非 JSON 也会抛 InternalAPIError，把这层加进
docstring 避免误导（之前 fetch_summary 根本没写异常，submit_internal 只写
"其它 4xx/5xx"）。

test_alert_server: 模块 docstring 的覆盖清单补上 HMAC 签名分支和鉴权先后
顺序 case（这轮新增 5 个 test case 但 docstring 还是旧的三条）。

* docs(readme): 跑起来路径改成 monorepo 下的 ChatBot 子目录

Author: longsizhuo

README.md

@@ -96,7 +96,7 @@ INTERNAL_API_KEY=...（openssl rand -hex 32）
 ### 3. 跑起来
 
 ```bash
-cd /home/ubuntu/ChatBot
+cd /home/ubuntu/involution-hell-project/ChatBot
 uv sync
 uv run chat-bot
 ```

chat-bot.service

@@ -7,11 +7,13 @@ Wants=network-online.target
 Type=simple
 User=ubuntu
 Group=ubuntu
-WorkingDirectory=/home/ubuntu/ChatBot
+WorkingDirectory=/home/ubuntu/involution-hell-project/ChatBot
 # 直接吃 involution-hell 后端的 .env（单一来源）
 EnvironmentFile=/home/ubuntu/involution-hell-project/backend/.env
 # uv 在 PATH 里（~/.local/bin），systemd 默认 PATH 偏窄，显式给一下
 Environment=PATH=/home/ubuntu/.local/bin:/usr/local/bin:/usr/bin:/bin
+# ProtectHome=read-only 会把 ~/.cache/uv 锁住，uv 起不来。把 cache 挪进项目（ReadWritePaths 已覆盖）。
+Environment=UV_CACHE_DIR=/home/ubuntu/involution-hell-project/ChatBot/.uv-cache
 ExecStart=/home/ubuntu/.local/bin/uv run --no-sync chat-bot
 Restart=on-failure
 RestartSec=5
@@ -24,7 +26,7 @@ NoNewPrivileges=true
 PrivateTmp=true
 ProtectSystem=strict
 ProtectHome=read-only
-ReadWritePaths=/home/ubuntu/ChatBot
+ReadWritePaths=/home/ubuntu/involution-hell-project/ChatBot
 
 [Install]
 WantedBy=multi-user.target

src/chat_bot/api_client.py

@@ -6,9 +6,13 @@
 
 from __future__ import annotations
 
+import json
 from dataclasses import dataclass
 
 import httpx
+import structlog
+
+log = structlog.get_logger(__name__)
 
 
 class DuplicateURL(Exception):
@@ -54,6 +58,30 @@ class AdminSummary:
     pending_samples: list[dict]  # {id, host, url}
 
 
+def _safe_json(resp: httpx.Response, *, endpoint: str) -> dict:
+    """安全解析 resp.json()：失败不让调用方吃 raw ValueError，统一抛 InternalAPIError。
+
+    - 后端异常时偶尔会吐 HTML 错误页 / gateway 的纯文本 → JSONDecodeError
+    - 记录 status + body 前 200 字节（不走 resp.text，避免解码整个大 body
+      或在错误处理路径上再抛一次 UnicodeDecodeError 把原始 JSON 错因盖掉）
+    """
+    try:
+        return resp.json()
+    except (json.JSONDecodeError, ValueError) as e:
+        # 只切前 200 字节，再用 replace 策略解码——坏字节变成 U+FFFD 而不是抛
+        preview = resp.content[:200].decode("utf-8", errors="replace")
+        log.warning(
+            "api_bad_json",
+            endpoint=endpoint,
+            status=resp.status_code,
+            body_preview=preview,
+        )
+        raise InternalAPIError(
+            resp.status_code,
+            f"invalid JSON from {endpoint}: {e}; body[:200]={preview!r}",
+        ) from e
+
+
 async def submit_internal(
     base_url: str,
     internal_key: str,
@@ -66,7 +94,7 @@ async def submit_internal(
 
     异常：
     - DuplicateURL：后端 409，URL 已被提交过
-    - InternalAPIError：其它 4xx/5xx
+    - InternalAPIError：其它 4xx/5xx，或 2xx 但 body 不是合法 JSON（_safe_json 抛）
     - httpx 原生的网络异常不包装，直接向上抛
     """
     payload = {
@@ -84,7 +112,7 @@ async def submit_internal(
     if resp.status_code >= 400:
         raise InternalAPIError(resp.status_code, resp.text)
 
-    body = resp.json()
+    body = _safe_json(resp, endpoint="submit_internal")
     data = body.get("data") or {}
     return SubmitResult(
         link_id=data.get("id", 0),
@@ -102,7 +130,7 @@ async def fetch_link(
 ) -> LinkDetail | None:
     """GET /api/community/links/internal/{id}。用于轮询 async 富化后的最终状态。
 
-    404 时返回 None；其它错误抛 InternalAPIError。
+    404 时返回 None；其它 4xx/5xx 或 body 非 JSON 一律抛 InternalAPIError。
     """
     url = base_url.rstrip("/") + f"/{link_id}"
     headers = {"X-Internal-Key": internal_key}
@@ -112,7 +140,7 @@ async def fetch_link(
         return None
     if resp.status_code >= 400:
         raise InternalAPIError(resp.status_code, resp.text)
-    data = (resp.json() or {}).get("data") or {}
+    data = (_safe_json(resp, endpoint="fetch_link") or {}).get("data") or {}
     return LinkDetail(
         link_id=data.get("id", 0),
         status=data.get("status", "UNKNOWN"),
@@ -131,14 +159,17 @@ async def fetch_summary(
     sample_limit: int = 5,
     timeout: float = 10.0,
 ) -> AdminSummary:
-    """GET /api/community/links/internal/summary。"""
+    """GET /api/community/links/internal/summary。
+
+    4xx/5xx 或 body 非 JSON 都抛 InternalAPIError。
+    """
     url = base_url.rstrip("/") + "/summary"
     headers = {"X-Internal-Key": internal_key}
     async with httpx.AsyncClient(timeout=timeout) as client:
         resp = await client.get(url, headers=headers, params={"sampleLimit": sample_limit})
     if resp.status_code >= 400:
         raise InternalAPIError(resp.status_code, resp.text)
-    data = (resp.json() or {}).get("data") or {}
+    data = (_safe_json(resp, endpoint="fetch_summary") or {}).get("data") or {}
     return AdminSummary(
         pending_manual=data.get("pendingManual", 0),
         flagged=data.get("flagged", 0),

src/chat_bot/cogs/alerts.py

@@ -1,11 +1,15 @@
 """FLAGGED 告警即时推送 Cog。
 
-Bot 内嵌一个 aiohttp 服务器（127.0.0.1:CHATBOT_ALERT_PORT），接收后端
+Bot 内嵌一个 aiohttp 服务器（0.0.0.0:CHATBOT_ALERT_PORT），接收后端
 SharedLinkEnrichmentWorker 在判定 status=FLAGGED 时 fire 的 webhook POST。
 收到后立即推送 Discord 管理员频道 + 邮件，不走每日 digest。
 
-鉴权：X-Internal-Key header，和后端共用同一把密钥。
-loopback 端口：和后端同机，不经 Caddy 不开公网，纯内网通信。
+为什么绑 0.0.0.0 而不是 127.0.0.1：backend 跑在 Docker 容器里，从容器看
+宿主机是 docker bridge (host.docker.internal)，只绑 loopback 接不到。
+对外暴露面由三层兜：
+  (a) X-Internal-Key 常量时间比较，防 timing 猜解
+  (b) 可选 HMAC-SHA256 签名（WEBHOOK_HMAC_SECRET 配了就强校验）
+  (c) 上游 Oracle VCN ingress / Docker networking 决定哪些 IP 能打过来
 
 payload 形如：
 {
@@ -21,6 +25,9 @@
 
 from __future__ import annotations
 
+import hashlib
+import hmac
+import json
 from datetime import datetime
 from zoneinfo import ZoneInfo
 
@@ -37,6 +44,11 @@
 _CST = ZoneInfo("Asia/Shanghai")
 
 
+def _loads_json(raw: bytes) -> dict:
+    """把 raw bytes 解析为 dict（上层用 try/except 兜 JSONDecodeError）。"""
+    return json.loads(raw.decode("utf-8"))
+
+
 class AlertServer(commands.Cog):
     def __init__(self, bot: commands.Bot, settings: Settings) -> None:
         self.bot = bot
@@ -53,26 +65,69 @@ async def cog_load(self) -> None:
         await self._runner.setup()
         # 绑 0.0.0.0：Backend 跑在 Docker 容器里，从容器内看 127.0.0.1 是容器自己
         # 而不是宿主机，因此必须监听所有接口才能接 docker bridge (host.docker.internal)。
-        # 公网侧安全性由 Oracle VCN ingress + X-Internal-Key header 双重保证。
+        # 公网侧安全性由 Oracle VCN ingress + X-Internal-Key header + 可选 HMAC 签名保证。
         site = web.TCPSite(
             self._runner, host="0.0.0.0", port=self.settings.chatbot_alert_port  # noqa: S104
         )
         await site.start()
         log.info("alert_server_listening", port=self.settings.chatbot_alert_port)
+        # HMAC 没配：允许向前兼容（后端可能还没部署签名逻辑），但必须显式警告
+        if self.settings.webhook_hmac_secret is None:
+            log.warning(
+                "alert_webhook_hmac_disabled",
+                note="WEBHOOK_HMAC_SECRET 未配置：/alert/flagged 只做 X-Internal-Key"
+                " 校验，不验签。属于过渡模式（backend 尚未发出签名）。backend 上线"
+                " 签名后，把密钥同步到本服务 env 即可启用。",
+            )
 
     async def cog_unload(self) -> None:
         if self._runner:
             await self._runner.cleanup()
 
+    @staticmethod
+    def _verify_hmac(secret: str, raw_body: bytes, sig_header: str) -> bool:
+        """校验 X-Signature: sha256=<hex> 格式的签名。
+
+        - header 缺失 / 格式不对 / digest 不匹配 → False
+        - 用 hmac.compare_digest 做常量时间比较
+        """
+        if not sig_header or not sig_header.startswith("sha256="):
+            return False
+        provided_hex = sig_header[len("sha256="):].strip()
+        if not provided_hex:
+            return False
+        expected_hex = hmac.new(
+            secret.encode("utf-8"), raw_body, hashlib.sha256
+        ).hexdigest()
+        # 大小写无关比较——避免后端大写/小写差异导致误拒
+        return hmac.compare_digest(provided_hex.lower(), expected_hex.lower())
+
     async def _handle_flagged(self, req: web.Request) -> web.Response:
-        # 鉴权
+        # 鉴权 1：X-Internal-Key（常量时间比较，避免 timing 泄露 key 前缀）
         provided = req.headers.get("X-Internal-Key", "")
         expected = self.settings.internal_api_key.get_secret_value()
-        if not expected or provided != expected:
+        if not expected or not hmac.compare_digest(provided, expected):
             return web.json_response({"ok": False, "msg": "forbidden"}, status=403)
 
+        # 先把原始 body 读出来——HMAC 必须对 raw bytes 算，JSON parse 之后再序列化会漂
+        raw_body = await req.read()
+
+        # 鉴权 2：HMAC-SHA256 签名（可选）。配了 secret 就强校验，没配就跳过。
+        hmac_secret = self.settings.webhook_hmac_secret
+        if hmac_secret is not None:
+            sig_header = req.headers.get("X-Signature", "")
+            if not self._verify_hmac(hmac_secret.get_secret_value(), raw_body, sig_header):
+                log.warning(
+                    "alert_hmac_reject",
+                    has_header=bool(sig_header),
+                    body_len=len(raw_body),
+                )
+                return web.json_response(
+                    {"ok": False, "msg": "invalid signature"}, status=401
+                )
+
         try:
-            payload = await req.json()
+            payload = _loads_json(raw_body)
         except Exception:
             return web.json_response({"ok": False, "msg": "bad json"}, status=400)
 

src/chat_bot/cogs/listener.py

@@ -15,6 +15,8 @@
 
 import asyncio
 import re
+from collections.abc import Coroutine
+from typing import Any
 from urllib.parse import urlparse
 
 import discord
@@ -67,6 +69,18 @@ def _should_skip(url: str) -> bool:
 log = structlog.get_logger(__name__)
 
 
+async def _safe(coro: Coroutine[Any, Any, Any], *, name: str) -> None:
+    """包装 background coroutine：异常打进 log，不让 fire-and-forget task 静默失败。
+
+    注意：except Exception 不会捕到 CancelledError（3.12 起 CancelledError
+    继承自 BaseException），所以 bot 优雅退出时取消 task 的行为不会被这里吞掉。
+    """
+    try:
+        await coro
+    except Exception:
+        log.exception("background_task_failed", task=name)
+
+
 class ShareListener(commands.Cog):
     def __init__(self, bot: commands.Bot, settings: Settings) -> None:
         self.bot = bot
@@ -140,8 +154,10 @@ async def _handle_one_url(self, message: discord.Message, url: str) -> None:
         )
 
         # 后台轮询拿最终状态，拿到了再发第二条
+        task_name = f"poll_{result.link_id}"
         asyncio.create_task(
-            self._notify_final_status(message, result.link_id)
+            _safe(self._notify_final_status(message, result.link_id), name=task_name),
+            name=task_name,
         )
 
     async def _notify_final_status(self, message: discord.Message, link_id: int) -> None:

src/chat_bot/config.py

@@ -26,6 +26,10 @@ class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         env_file=ENV_FILE if Path(ENV_FILE).exists() else None,
         env_file_encoding="utf-8",
+        # ChatBot 读取的是 backend 共享 .env，里头含大量非本服务的 key（PG*/
+        # SPRING_*/OPENAI_* 等），开 forbid 会在启动期直接炸出 ~28 个
+        # validation error。改用 ignore 兜底。
+        # 未来若把 ChatBot 切到独立 .env，可以回到 forbid 以获得 typo 检测。
         extra="ignore",
         case_sensitive=False,
     )
@@ -47,16 +51,27 @@ class Settings(BaseSettings):
     digest_time_cst: str = Field("09:00", alias="DIGEST_TIME_CST")
 
     # ---------- FLAGGED 实时 alert ----------
-    # 后端 webhook → Bot 内嵌 aiohttp server 接收端口，只绑 127.0.0.1
+    # 后端 webhook → Bot 内嵌 aiohttp server 接收端口。实际绑 0.0.0.0:<port>
+    # （backend 在 Docker 容器里，只绑 127.0.0.1 的话容器过不来）。暴露面靠
+    # X-Internal-Key 常量时间比较 + 可选 HMAC 签名 + 上游 firewall / Docker
+    # networking 共同兜。
     chatbot_alert_port: int = Field(6200, alias="CHATBOT_ALERT_PORT")
+    # HMAC-SHA256 共享密钥（可选）。配了之后 /alert/flagged 会强校验
+    # X-Signature: sha256=<hex> = HMAC(secret, raw_body)。没配时跳过这层。
+    webhook_hmac_secret: SecretStr | None = Field(None, alias="WEBHOOK_HMAC_SECRET")
 
     # ---------- Gmail SMTP ----------
     # 未填时不发邮件（但 Discord 推送仍走）
     gmail_user: str = Field("", alias="GMAIL_USER")
     gmail_app_password: SecretStr = Field(SecretStr(""), alias="GMAIL_APP_PASSWORD")
     digest_email_to: str = Field("", alias="DIGEST_EMAIL_TO")
 
-    @field_validator("discord_guild_id", "discord_admin_channel_id", mode="before")
+    @field_validator(
+        "discord_guild_id",
+        "discord_admin_channel_id",
+        "webhook_hmac_secret",
+        mode="before",
+    )
     @classmethod
     def _empty_to_none(cls, v: object) -> object:
         if isinstance(v, str) and v.strip() == "":