765dbdfbd4
M4（文风）: style-auditor 双轨（提取指纹/漂移第四审）+ jobs 长任务框架(zombie reaper) + 回炉 refine + GET /style read-back。 M5（生成+扩展）: worldbuilder/character-gen（入库 continuity 409 gate + partition_writes 白名单 + schema→JSONB 形变）； 网关多 provider 回退链/熔断/能力降级（Anthropic/Gemini 适配器）；Skill registry + 表权限沙箱 + 规则； 前端 角色生成器/世界观/Codex/规则页/技能库/⌘K 命令面板。 K1（Kimi Code 订阅接入）: OAuth device-flow（kimi-code）+ 静态 Console key（kimi-code-key）两路径； coding 端点 KimiCLI 伪造头（实测 UA allow-list 门禁，缺则 403）+ JSON 模式结构化（thinking ⊥ tool_choice）。 本地联调修复: CORS 中间件；assemble 注入 premise+「写第N章」指令（修空 prompt 400）； GET /outline·/draft read-back + 大纲/工作台/审稿页重载；写页 client/server 常量边界 + notFound 健壮化； 字数 toLocaleString locale 水合；审稿页终稿从已存草稿 seed（修 accept 422）。 门禁: backend ruff/mypy(157)/alembic 无漂移/pytest 451 · frontend lint/tsc/vitest/build。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
"""Skill 表权限沙箱（ARCH §5.6 / PRODUCT_SPEC §5.5）——纯函数，无 IO。
|
||
|
||
这**不是进程级沙箱**：声明式 Skill 不执行代码。强制点在编排/写库层（"声明式权限 +
|
||
apply 层白名单"）：
|
||
- `filter_reads`：注入时只把 skill 声明 `reads` 的表数据喂给它（未声明的表丢弃，不喂）。
|
||
- `partition_writes`：写库时只应用 skill 声明 `writes` 的产出（越权字段丢弃 + 审计），
|
||
返回 `(allowed, rejected)`——调用方落 `allowed`、log `rejected`。不变量 #3：
|
||
allowed 仍须经验收 gate 才真入库，本层只裁剪白名单，不开写后门。
|
||
- `validate_declaration`：注册/加载 skill 时校验 `reads/writes` 全在已知创作表白名单内
|
||
（杜绝声明指向不存在/系统表）；越界 → AppError(VALIDATION)。
|
||
|
||
不可变：所有函数返回新 dict/list，从不原地改入参（全局 immutability 约定）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from ww_agents import AgentSpec
|
||
from ww_shared import AppError, ErrorCode
|
||
|
||
# 声明式 Skill 允许声明读/写的创作表白名单（ARCH §3.1 创作表 + rules；
|
||
# 系统/运营表 users/jobs/usage_ledger/provider_credentials/tier_routing/skills 不在此列）。
|
||
KNOWN_TABLES: frozenset[str] = frozenset(
|
||
{
|
||
"projects",
|
||
"characters",
|
||
"world_entities",
|
||
"outline",
|
||
"chapter_digests",
|
||
"foreshadow",
|
||
"style_fingerprint",
|
||
"rules",
|
||
"chapters",
|
||
"chapter_reviews",
|
||
}
|
||
)
|
||
|
||
|
||
def filter_reads(spec: AgentSpec, available: dict[str, Any]) -> dict[str, Any]:
|
||
"""注入白名单：只保留 skill 声明 `reads` 且 `available` 里确有的表（新 dict）。"""
|
||
declared = set(spec.reads)
|
||
return {table: data for table, data in available.items() if table in declared}
|
||
|
||
|
||
def partition_writes(spec: AgentSpec, produced: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
|
||
"""写库白名单：把产出分成 (allowed, rejected)。
|
||
|
||
`allowed`：产出表 ∈ skill 声明 `writes`（新 dict，调用方落库）。
|
||
`rejected`：越权产出表名（排序，调用方审计 log + 丢弃）。
|
||
"""
|
||
declared = set(spec.writes)
|
||
allowed: dict[str, Any] = {}
|
||
rejected: list[str] = []
|
||
for table, data in produced.items():
|
||
if table in declared:
|
||
allowed[table] = data
|
||
else:
|
||
rejected.append(table)
|
||
return allowed, sorted(rejected)
|
||
|
||
|
||
def validate_declaration(spec: AgentSpec) -> None:
|
||
"""校验 skill 的 `reads/writes` 全在 `KNOWN_TABLES`；越界 → AppError(VALIDATION)。"""
|
||
unknown = sorted((set(spec.reads) | set(spec.writes)) - KNOWN_TABLES)
|
||
if unknown:
|
||
raise AppError(
|
||
ErrorCode.VALIDATION,
|
||
f"Skill「{spec.name}」声明了未知表：{unknown}",
|
||
{"skill": spec.name, "unknown_tables": unknown},
|
||
)
|