fix: 安全与运行时正确性修复(H1/H2 + M2/M4/M5/L4/L5)

app/api/websocket.py

@@ -500,7 +500,12 @@ async def audio_processor(
                     segment_start_time=speech_start_sample / sample_rate,
                 )
                 speech_buffer = np.array([], dtype=np.float32)
-                ctx.last_full_text = ""  # 超时后重置上下文
+                # Bug(M4): 超时只是"队列暂空",状态机仍在 SPEECH、语义上
+                # 上下文连续。若在此重置 last_full_text,下一段 ASR 返回的
+                # 含旧前缀文本(0.5s 上下文 / 短间隔重复语气词)会因基准清空
+                # 被当增量重发 → 前端看到重复字。故超时 flush 后**不重置**
+                # last_full_text,也不切 state;静音触发的 reset(见下文
+                # STATE_SILENCE 切换)才重置上下文。
                 state = STATE_SILENCE
                 silence_frame_count = 0
                 silence_sample_count = 0

app/middleware/security.py

@@ -9,7 +9,20 @@
 
 
 def is_trusted_browser_origin(origin: str | None, allowed_origins=()) -> bool:
-    """Allow native/no-Origin clients and explicitly trusted browser origins."""
+    """Allow native/no-Origin clients and explicitly trusted browser origins.
+
+    无 Origin 视为可信是**有意为之**:浏览器对同源 fetch/XHR 在部分实现
+    (如 Firefox)下不携带 Origin 头,本地模式的 SPA 在这些浏览器里靠这个
+    放行。强制要求 Origin 会直接打断 SPA。
+
+    CSRF 防护不靠"必须有 Origin",而是靠"Origin 存在时必须是可信的":
+    不可信的跨站 Origin(如 evil.example)在此返回 False,中间件会拒绝
+    bypass 并要求 Bearer token。
+
+    注意:本机攻击者可任意伪造 Origin/Sec-Fetch-* 头,header 层无法区分
+    "合法本机工具"与"恶意本机进程"。真正的本机隔离需走 DEPLOYMENT_MODE=lan
+    或 LOCAL_AUTH_DISABLED=false(强制 token),这是文档化的威胁模型。
+    """
     if not origin:
         return True
     if origin in allowed_origins:

app/runtime.py

@@ -108,19 +108,34 @@ def diagnose_audio_dependencies() -> AudioDependencyReport:
 
 
 class InferenceCoordinator:
-    """Serialize shared-device inference while giving live audio first access."""
+    """Serialize shared-device inference while giving live audio first access.
+
+    公平性(M2):live 享有优先权,但持续涌入的 live 会无限期饿死已排队的
+    offline(refinement job)。offline 等待超过 _OFFLINE_STARVE_TIMEOUT 秒后
+    放宽进入条件为 `not active`(不再要求"无 live 在排队"),这样 live 运行
+    完释放槽的瞬间,等了很久的 offline 能抢到。live 的实时优先权在常规情况
+    下不受影响(只有 offline 饿了很久才让步一次)。
+    """
+
+    # offline 饿死超时:超过后放宽进入条件,保证 refinement job 终被调度
+    _OFFLINE_STARVE_TIMEOUT = 30.0
 
     def __init__(self) -> None:
         self._condition = asyncio.Condition()
         self._active = False
         self._live_waiters = 0
+        # 有 offline 已过饿死超时、正等待让位。live 见此会谦让一次,
+        # 保证 starved offline 在下一个槽释放时优先进入。
+        self._offline_starving = False
 
     @asynccontextmanager
     async def live(self):
         async with self._condition:
             self._live_waiters += 1
             try:
-                await self._condition.wait_for(lambda: not self._active)
+                await self._condition.wait_for(
+                    lambda: not self._active and not self._offline_starving
+                )
                 self._active = True
             finally:
                 self._live_waiters -= 1
@@ -132,9 +147,23 @@ async def live(self):
     @asynccontextmanager
     async def offline(self):
         async with self._condition:
-            await self._condition.wait_for(
-                lambda: not self._active and self._live_waiters == 0
-            )
+            starved = False  # 是否已过饿死超时,过则放宽进入条件
+            while True:
+                # 常规条件:无运行 + 无 live 排队(让 live 优先)
+                if not self._active and self._live_waiters == 0:
+                    break
+                # 已饿死:放宽为仅"无运行",并置 starving 标志让 live 谦让
+                if starved and not self._active:
+                    break
+                # 单层 wait_for 超时:仅首次用饿死超时,标记 starved 后转为
+                # 无超时等待 live 释放(避免 wait_for 嵌套 wait_for 的锁陷阱)
+                timeout = self._OFFLINE_STARVE_TIMEOUT if not starved else None
+                try:
+                    await asyncio.wait_for(self._condition.wait(), timeout=timeout)
+                except asyncio.TimeoutError:
+                    starved = True
+                    self._offline_starving = True
+            self._offline_starving = False
             self._active = True
         try:
             yield

app/services/llm_gateway.py

@@ -31,18 +31,25 @@ class LLMModelMissingError(LLMUnavailableError):
     pass
 
 
-def _validate_endpoint(endpoint: str, allowed_hosts: tuple, allow_public: bool = False) -> None:
-    """解析 endpoint 并校验 host 是私有/本地地址
-    allow_public=True 时跳过 IP 校验（用户已显式开公网）。
+def _validate_endpoint(
+    endpoint: str, allowed_hosts: tuple, allow_public: bool = False
+) -> Optional[str]:
+    """解析 endpoint 并校验 host 是私有/本地地址。
+
+    返回 init 时刻解析到的 IP(供 DNS pinning 缓存,防 init→request 之间的
+    DNS rebinding)。返回 None 表示:命中 allowed_hosts / allow_public=True /
+    host 是 IP literal — 这些情况不需要缓存 IP 做 pinning 基准。
+
+    allow_public=True 时跳过 IP 校验(用户已显式开公网)。
     """
     parsed = urlparse(endpoint)
     host = parsed.hostname
     if not host:
         raise EndpointSecurityError(f"无法解析 endpoint: {endpoint}")
     if host in allowed_hosts:
-        return
+        return None
     if allow_public:
-        return
+        return None
     try:
         ip = ipaddress.ip_address(socket.gethostbyname(host))
     except (socket.gaierror, ValueError) as e:
@@ -54,12 +61,37 @@ def _validate_endpoint(endpoint: str, allowed_hosts: tuple, allow_public: bool =
             f"如确实需要,设置环境变量 LLM_ALLOW_PUBLIC=true 显式开公网,"
             f"并通过 LLM_API_KEY 配置 Bearer token。"
         )
+    return str(ip)
 
 
 class LLMGateway:
     def __init__(self, config: LLMConfig):
+        # DNS rebinding 防御基准:host 与 IP 在 init 时刻就被校验并缓存,
+        # 请求时 pin 到这个缓存 IP(而非每次重新解析),并在请求前再校验
+        # 一次当前 DNS 与缓存一致(见 _assert_no_dns_rebind)。
+        self._pinned_host: Optional[str] = None
+        self._pinned_ip: Optional[str] = None
         if config.enabled:
-            _validate_endpoint(config.endpoint, config.allowed_hosts, config.allow_public)
+            validated_ip = _validate_endpoint(
+                config.endpoint, config.allowed_hosts, config.allow_public
+            )
+            parsed = urlparse(config.endpoint)
+            host = parsed.hostname
+            self._pinned_host = host
+            if validated_ip is not None:
+                # strict 私网模式:init 已解析出私网 IP,缓存作为 pinning 基准。
+                self._pinned_ip = validated_ip
+            elif not config.allow_public and host is not None:
+                # allowed_hosts 里的域名(如 "localhost"):validate 没解析,
+                # 这里补一次解析用于 pinning。IP literal 不需要 pin。
+                try:
+                    ipaddress.ip_address(host)
+                except ValueError:
+                    try:
+                        self._pinned_ip = socket.gethostbyname(host)
+                    except socket.gaierror:
+                        self._pinned_ip = None
+                # allow_public=True 不缓存 → 请求时跟随 DNS 轮询,无 rebinding 威胁
         self.config = config
         self._available_cache: Optional[bool] = None
         self._cache_time: float = 0.0
@@ -246,6 +278,30 @@ async def _call_llm(self, prompt: str) -> str:
         except httpx.HTTPError as e:
             raise LLMUnavailableError(f"LLM HTTP 错误: {e}") from e
 
+    def _assert_no_dns_rebind(self) -> None:
+        """请求前再校验一次:当前 DNS 解析结果与 init 缓存的 IP 是否一致。
+
+        仅 strict 私网模式(有缓存 IP 且未开 allow_public)生效。allow_public
+        用户已显式接受公网,无 rebinding 威胁,跟随 DNS 轮询即可。
+        不一致 → 抛 EndpointSecurityError,调用方降级到 extractive 兜底,
+        绝不把连接重定向到攻击者 rebind 出的公网 IP。
+        """
+        if self.config.allow_public:
+            return
+        if not self._pinned_host or not self._pinned_ip:
+            return
+        try:
+            fresh = socket.gethostbyname(self._pinned_host)
+        except socket.gaierror:
+            # DNS 暂时不可用:pinning 仍用缓存 IP(安全默认),不阻断
+            return
+        if fresh != self._pinned_ip:
+            raise EndpointSecurityError(
+                f"DNS rebinding 检测: {self._pinned_host} 启动时解析为 "
+                f"{self._pinned_ip},当前解析为 {fresh}。"
+                f"拒绝 LLM 调用,降级到本地摘要。"
+            )
+
     async def _post_with_dns_guard(
         self,
         url: str,
@@ -260,6 +316,8 @@ async def _post_with_dns_guard(
         in a worker thread so concurrent async callers do not block the event
         loop, and the context manager always restores the socket function.
         """
+        # 先做 rebinding 校验(可能抛 EndpointSecurityError → 调用方降级)
+        self._assert_no_dns_rebind()
         pinned_ip = self._resolve_pinned_ip(url)
         async with self._dns_pin_guard(url, pinned_ip):
             async with httpx.AsyncClient(timeout=timeout) as client:
@@ -291,9 +349,17 @@ async def _dns_pin_guard(cls, url: str, pinned_ip: Optional[str]):
             cls._uninstall_socket_patch()
             cls._socket_patch_lock.release()
 
-    @classmethod
-    def _resolve_pinned_ip(cls, url: str) -> Optional[str]:
-        """从 URL 解析域名,提前解析成 IP 备用。返回 None 表示跳过 pinning。"""
+    def _resolve_pinned_ip(self, url: str) -> Optional[str]:
+        """返回用于 socket pinning 的 IP。None 表示跳过 pinning。
+
+        - IP literal host:不需要 pin(URL 里的 host 就是 IP)
+        - allow_public=True:跟随当前 DNS(用户显式开公网,无 rebinding 威胁,
+          且公网 endpoint 常做 DNS 轮询/CDN,固定 IP 反而会连不上)
+        - strict 私网模式:返回 init 时缓存的 IP,不再重新解析 —— 这是防
+          DNS rebinding 的关键:连接目标锁死到 init 校验过的私网 IP,
+          请求时刻即使 DNS 被 rebind 到公网也不受影响(_assert_no_dns_rebind
+          还会再校验一次并拒绝)。
+        """
         try:
             parsed = urlparse(url)
             host = parsed.hostname
@@ -305,7 +371,15 @@ def _resolve_pinned_ip(cls, url: str) -> Optional[str]:
                 return None
             except ValueError:
                 pass
-            return socket.gethostbyname(host)
+            if self.config.allow_public:
+                try:
+                    return socket.gethostbyname(host)
+                except socket.gaierror:
+                    return None
+            # strict 模式:用 init 缓存的 IP,绝不重新解析作为连接目标
+            if host == self._pinned_host:
+                return self._pinned_ip
+            return None
         except (socket.gaierror, ValueError):
             return None
 

app/services/realtime_auth.py

@@ -1,23 +1,59 @@
 """Authentication boundary for the realtime WebSocket protocol."""
 import asyncio
+import collections
 import json
 import logging
 import os
+import time
 
 from app.config import config
 from app.middleware.security import is_trusted_browser_origin
 
 
 logger = logging.getLogger("Matrix_Core")
 
+# L5: 进程级 WebSocket 连接速率限制。HTTP 有 RateLimitMiddleware,但 WS 不
+# 走 HTTP 中间件,攻击者可空打 WS 触发 5s receive 超时占用 fd/协程。这里在
+# 鉴权前做 per-IP 滑动窗口:窗口内连接数超阈值则 close(4401)。
+_WS_CONNECT_WINDOW = 60.0          # 滑动窗口(秒)
+_WS_CONNECT_MAX = 20               # 窗口内每 IP 最大连接数
+_ws_connect_log: dict[str, collections.deque] = {}
+
+
+def _ws_rate_limited(client_host: str) -> bool:
+    """返回 True 表示该 IP 在窗口内 WS 连接数超限。"""
+    if not client_host:
+        return False
+    now = time.time()
+    dq = _ws_connect_log.setdefault(client_host, collections.deque())
+    cutoff = now - _WS_CONNECT_WINDOW
+    while dq and dq[0] < cutoff:
+        dq.popleft()
+    if len(dq) >= _WS_CONNECT_MAX:
+        return True
+    dq.append(now)
+    return False
+
 
 async def authenticate_websocket(websocket, client_id: str) -> bool:
     """Authenticate the first WebSocket message, or allow trusted local mode."""
     client_host = websocket.client.host if websocket.client else ""
+    # L5: WS 连接限流(在鉴权之前),防空打 WS 占 fd/协程
+    if _ws_rate_limited(client_host):
+        logger.warning("[WS] %s 连接被限流 (%.0fs 内超 %d)", client_host,
+                       _WS_CONNECT_WINDOW, _WS_CONNECT_MAX)
+        try:
+            await websocket.close(code=4401, reason="连接过于频繁")
+        except Exception:
+            pass
+        return False
+    # 注意:loopback 集合只含真实本机地址。"testclient"(Starlette TestClient
+    # 的固定 host)不得放进生产鉴权路径 —— 那等于为测试开后门,真实客户端
+    # 不会用它。WS 测试如需 bypass,走 TEST_AUTH_BYPASS=1 或带真实 token。
     local_bypass = (
         config.deployment.mode == "local"
         and config.auth.local_auth_disabled
-        and client_host in ("127.0.0.1", "::1", "localhost", "testclient")
+        and client_host in ("127.0.0.1", "::1", "localhost")
         and is_trusted_browser_origin(
             websocket.headers.get("origin"), config.cors.allowed_origins
         )

engine/asr_engine.py

@@ -111,13 +111,17 @@ def _load_with_fallback(self, preferred: str) -> bool:
         for device in order:
             logger.info(f"[ASR] 尝试加载到 {device}（超时 {timeout}s）")
             t0 = time.time()
-            result: dict = {"ok": False, "err": None, "asr_model": None}
+            result: dict = {"ok": False, "err": None, "asr_model": None, "cancelled": False}
 
             def _worker():
                 try:
+                    if result["cancelled"]:
+                        return
                     model_dir = snapshot_download(
                         "Qwen/Qwen3-ASR-0.6B", revision=QWEN_ASR_REVISION
                     )
+                    if result["cancelled"]:
+                        return
                     dtype = torch.bfloat16 if device in ("cuda", "mps") else torch.float32
                     # 可选加载 forced aligner (Qwen3-ForcedAligner-0.6B, ~600MB)
                     # ⚠️ qwen_asr 的 Qwen3ASRModel.from_pretrained 把 forced_aligner 当 str(repo id) 处理
@@ -126,15 +130,23 @@ def _worker():
                     # (否则会触发 "Repo id must use alphanumeric chars" 错误)
                     forced_aligner_id = None
                     if self.config.asr_word_timestamps:
+                        if result["cancelled"]:
+                            return
                         logger.info("[ASR] 加载 forced aligner (Qwen3-ForcedAligner-0.6B)")
                         forced_aligner_id = snapshot_download(
                             "Qwen/Qwen3-ForcedAligner-0.6B",
                             revision=QWEN_ALIGNER_REVISION,
                         )
+                    if result["cancelled"]:
+                        return
                     result["asr_model"] = Qwen3ASRModel.from_pretrained(
                         model_dir, dtype=dtype, device_map=device,
                         forced_aligner=forced_aligner_id,
                     )
+                    if result["cancelled"]:
+                        # 已被取消:刚加载的模型无人持有,清引用让 GC 回收
+                        result["asr_model"] = None
+                        return
                     result["ok"] = True
                 except Exception as e:
                     result["err"] = e
@@ -148,7 +160,10 @@ def _worker():
                 logger.warning(
                     f"[ASR] {device} 加载超时 ({elapsed:.0f}s > {timeout}s)，放弃此设备"
                 )
-                # daemon=True 会随主进程退出；这里尽力释放 worker 内部引用
+                # M5: 置 cancel flag,daemon worker 在下一个阶段边界会主动 return,
+                # 减少与下一个设备加载线程并发占显存/内存(daemon=True 随主进程退出,
+                # 但服务长跑时仍值得尽早收尾)。
+                result["cancelled"] = True
                 if result.get("asr_model") is not None:
                     result["asr_model"] = None
                 continue

engine/speaker/campplus_engine.py

@@ -182,12 +182,14 @@ def compare_and_identify(
         Returns:
             tuple[str, float]: (说话人ID, 置信度 0.0-1.0)
                 - 短段(<MIN_USABLE_DURATION) → ("Spk_unknown", 0.0)
-                - embedding=None → ("Unknown", 0.0)
+                - embedding=None → ("Spk_unknown", 0.0)
                 - 命中已有 Spk(高/边缘阈值) → (spk_id, 1.0 - best_dist)
                 - 新建 Spk → (new_id, 1.0 - best_dist)  (新 Spk 通常置信度较低)
         """
         if current_emb is None:
-            return "Unknown", 0.0
+            # L4: 用 Spk_unknown(符合 ^Spk_ 格式),与下游默认值及
+            # speaker_id 校验 pattern 一致,避免 "Unknown" 不匹配格式。
+            return "Spk_unknown", 0.0
 
         # Bug-68: 用纯函数分类段类型(supports testable + 复用)
         segment_class = self._classify_segment_duration(audio_duration)

engine/speaker/eres2net_engine.py

@@ -122,7 +122,8 @@ def compare_and_identify(
             tuple[str, float]: (说话人ID, 置信度 0.0-1.0)
         """
         if current_emb is None:
-            return "Unknown", 0.0
+            # L4: Spk_unknown 符合 ^Spk_ 格式,与下游默认值一致
+            return "Spk_unknown", 0.0
 
         # 判断是否为可靠样本
         is_reliable = audio_duration >= 1.5