fix: 一批低危问题修复

- alert_store: 删除 delete_one 里不可达的 return count 死代码;
  list_recent/count 持锁读, 防与 prune/clear 整文件重写并发读到截断内容
- ai_generator: import 校验从黑名单改为白名单 (仅 polars/__future__),
  黑名单挡不住 ctypes/importlib/builtins/pickle 等未列出模块;
  _extract_meta 改用 ast.literal_eval 替代 compile+eval
- frontend api.ts: request() 合并调用方 headers (此前被整体覆盖丢弃)
- VERSION 同步至 v0.1.70 (与 pyproject/package.json 一致)

VERSION

@@ -1 +1 @@
-v0.1.64
+v0.1.70

backend/app/services/alert_store.py

@@ -72,7 +72,10 @@ def list_recent(
     source: str | None = None,
     type: str | None = None,
 ) -> list[dict]:
-    """读取近 N 天记录,按时间倒序,支持按 source/type 过滤。"""
+    """读取近 N 天记录,按时间倒序,支持按 source/type 过滤。
+
+    持锁读: prune/delete/clear 会整文件重写, 无锁读可能读到截断内容。
+    """
     import time
     cutoff = (time.time() - days * 86400) * 1000  # 毫秒
     out: list[dict] = []
@@ -80,7 +83,7 @@ def list_recent(
     if not p.exists():
         return []
     try:
-        with p.open("r", encoding="utf-8") as f:
+        with _lock, p.open("r", encoding="utf-8") as f:
             for line in f:
                 line = line.strip()
                 if not line:
@@ -159,16 +162,15 @@ def delete_one(data_dir: Path, ts: int) -> bool:
             logger.warning("alert_store delete_one write failed: %s", e)
             return False
         return True
-        return count
 
 
 def count(data_dir: Path) -> int:
-    """返回当前记录总数。"""
+    """返回当前记录总数。持锁读, 防与整文件重写并发。"""
     p = _path(data_dir)
     if not p.exists():
         return 0
     try:
-        with p.open("r", encoding="utf-8") as f:
+        with _lock, p.open("r", encoding="utf-8") as f:
             return sum(1 for line in f if line.strip())
     except Exception:
         return 0

backend/app/strategy/ai_generator.py

@@ -95,13 +95,15 @@ class AIStrategyGenerator:
             content = content.split("```", 1)[1].split("```", 1)[0].strip()
         return content
 
-    @staticmethod
-    def _validate_safety(code: str) -> None:
-        """AST 级安全检查"""
+    # import 白名单: 策略文件只允许 polars (见 strategy-guide.md「只 import polars」)。
+    # 白名单而非黑名单 — 黑名单挡不住 ctypes/importlib/builtins/pickle 等未列出的危险模块。
+    _ALLOWED_IMPORT_MODULES = frozenset({"polars", "__future__"})
+
+    @classmethod
+    def _validate_safety(cls, code: str) -> None:
+        """AST 级安全检查: import 白名单 + 危险内建调用拦截。"""
         tree = ast.parse(code)
 
-        forbidden_modules = {"os", "sys", "subprocess", "socket", "shutil",
-                             "pathlib", "http", "urllib", "requests", "httpx"}
         forbidden_calls = {"open", "exec", "eval", "compile", "__import__",
                            "globals", "locals", "vars", "dir", "getattr",
                            "setattr", "delattr", "type", "input"}
@@ -109,28 +111,26 @@ class AIStrategyGenerator:
         for node in ast.walk(tree):
             if isinstance(node, ast.Import):
                 for alias in node.names:
-                    if alias.name.split(".")[0] not in ("polars",):
-                        if alias.name.split(".")[0] in forbidden_modules:
-                            raise ValueError(f"禁止 import {alias.name}")
+                    if alias.name.split(".")[0] not in cls._ALLOWED_IMPORT_MODULES:
+                        raise ValueError(f"禁止 import {alias.name} (策略只允许 import polars)")
             if isinstance(node, ast.ImportFrom):
-                if node.module and node.module.split(".")[0] not in ("polars",):
-                    if node.module.split(".")[0] in forbidden_modules:
-                        raise ValueError(f"禁止 from {node.module} import")
+                mod = (node.module or "").split(".")[0]
+                if mod not in cls._ALLOWED_IMPORT_MODULES:
+                    raise ValueError(f"禁止 from {node.module} import (策略只允许 import polars)")
             if isinstance(node, ast.Call):
                 if isinstance(node.func, ast.Name) and node.func.id in forbidden_calls:
                     raise ValueError(f"禁止调用 {node.func.id}()")
 
     @staticmethod
     def _extract_meta(code: str) -> dict:
-        """从代码字符串中提取 META 字典（不执行代码）"""
+        """从代码字符串中提取 META 字典（不执行代码, 仅接受字面量）"""
         tree = ast.parse(code)
         for node in ast.walk(tree):
             if isinstance(node, ast.Assign):
                 for target in node.targets:
                     if isinstance(target, ast.Name) and target.id == "META":
-                        # 找到 META 赋值，用 compile+eval 安全提取
-                        # 只允许字面量
-                        meta_node = node.value
-                        code_obj = compile(ast.Expression(meta_node), "<meta>", "eval")
-                        return eval(code_obj, {"__builtins__": {}})  # noqa: S307
+                        try:
+                            return ast.literal_eval(node.value)
+                        except (ValueError, SyntaxError) as e:
+                            raise ValueError(f"META 必须是纯字面量字典: {e}") from e
         return {}

frontend/src/lib/api.ts

@@ -11,6 +11,8 @@ async function request<T>(path: string, init?: RequestInit): Promise<T> {
   const isFormData = init?.body instanceof FormData
   const headers: Record<string, string> = {}
   if (!isFormData) headers['Content-Type'] = 'application/json'
+  // 合并调用方传入的 headers (此前会被整体覆盖丢弃)
+  Object.assign(headers, init?.headers as Record<string, string> | undefined)
   const res = await fetch(`${BASE}${path}`, { ...init, headers })
   if (!res.ok) {
     let detail = ''