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>
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""T5.5 表权限沙箱单测（ARCH §5.6：声明式权限 + apply 层白名单，非进程沙箱）。
|
||
|
||
纯函数、无 IO：
|
||
- `filter_reads`：注入时只把 skill 声明 `reads` 的表数据喂给它（越权读 → 丢弃，不喂）。
|
||
- `partition_writes`：写库时只应用 skill 声明 `writes` 的产出字段（越权写 → 丢弃 + 审计），
|
||
返回 `(allowed, rejected)` 两份，调用方据此落库 allowed、log rejected。
|
||
- `validate_declaration`：加载/注册 skill 时校验其声明的 `reads/writes` 全在已知表白名单内
|
||
（杜绝声明指向不存在的表）；越界 → AppError(VALIDATION)。
|
||
不变量 #3：自定义 skill 产出仍过验收 gate——本层只做白名单裁剪，不开写后门。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from ww_agents import AgentSpec
|
||
from ww_shared import AppError, ErrorCode
|
||
from ww_skills import (
|
||
KNOWN_TABLES,
|
||
filter_reads,
|
||
partition_writes,
|
||
validate_declaration,
|
||
)
|
||
|
||
|
||
def _spec(*, reads: list[str], writes: list[str], scope: str = "custom") -> AgentSpec:
|
||
return AgentSpec(
|
||
name="custom_skill",
|
||
tier="writer",
|
||
system_prompt="x",
|
||
reads=reads,
|
||
writes=writes,
|
||
scope=scope,
|
||
)
|
||
|
||
|
||
# ---- filter_reads ----
|
||
|
||
|
||
def test_filter_reads_keeps_only_declared_tables() -> None:
|
||
spec = _spec(reads=["world_entities"], writes=[])
|
||
available = {
|
||
"world_entities": [{"name": "灵根"}],
|
||
"characters": [{"name": "主角"}], # 未声明 → 必须丢弃
|
||
}
|
||
|
||
fed = filter_reads(spec, available)
|
||
|
||
assert fed == {"world_entities": [{"name": "灵根"}]}
|
||
|
||
|
||
def test_filter_reads_ignores_declared_table_absent_from_available() -> None:
|
||
spec = _spec(reads=["world_entities", "characters"], writes=[])
|
||
available = {"world_entities": [{"name": "灵根"}]}
|
||
|
||
fed = filter_reads(spec, available)
|
||
|
||
assert fed == {"world_entities": [{"name": "灵根"}]}
|
||
|
||
|
||
def test_filter_reads_empty_declaration_feeds_nothing() -> None:
|
||
spec = _spec(reads=[], writes=[])
|
||
available = {"world_entities": [{"name": "灵根"}]}
|
||
|
||
assert filter_reads(spec, available) == {}
|
||
|
||
|
||
# ---- partition_writes ----
|
||
|
||
|
||
def test_partition_writes_drops_over_permission_fields() -> None:
|
||
spec = _spec(reads=[], writes=["world_entities"])
|
||
produced = {
|
||
"world_entities": [{"name": "新势力"}],
|
||
"characters": [{"name": "越权角色"}], # 越权 → 丢弃 + 审计
|
||
}
|
||
|
||
allowed, rejected = partition_writes(spec, produced)
|
||
|
||
assert allowed == {"world_entities": [{"name": "新势力"}]}
|
||
assert rejected == ["characters"]
|
||
|
||
|
||
def test_partition_writes_all_allowed_yields_empty_rejected() -> None:
|
||
spec = _spec(reads=[], writes=["world_entities", "characters"])
|
||
produced: dict[str, object] = {"world_entities": [], "characters": []}
|
||
|
||
allowed, rejected = partition_writes(spec, produced)
|
||
|
||
assert set(allowed) == {"world_entities", "characters"}
|
||
assert rejected == []
|
||
|
||
|
||
def test_partition_writes_no_declared_writes_rejects_all() -> None:
|
||
spec = _spec(reads=[], writes=[])
|
||
produced = {"world_entities": [{"name": "x"}]}
|
||
|
||
allowed, rejected = partition_writes(spec, produced)
|
||
|
||
assert allowed == {}
|
||
assert rejected == ["world_entities"]
|
||
|
||
|
||
# ---- validate_declaration ----
|
||
|
||
|
||
def test_validate_declaration_passes_for_known_tables() -> None:
|
||
spec = _spec(reads=["world_entities"], writes=["world_entities"])
|
||
# 不抛即通过
|
||
validate_declaration(spec)
|
||
|
||
|
||
def test_validate_declaration_rejects_unknown_read_table() -> None:
|
||
spec = _spec(reads=["secret_table"], writes=[])
|
||
|
||
with pytest.raises(AppError) as exc:
|
||
validate_declaration(spec)
|
||
|
||
assert exc.value.code is ErrorCode.VALIDATION
|
||
assert "secret_table" in str(exc.value.details)
|
||
|
||
|
||
def test_validate_declaration_rejects_unknown_write_table() -> None:
|
||
# 系统表 users 不在创作表白名单 → 越权写声明被拒。
|
||
assert "users" not in KNOWN_TABLES
|
||
spec = _spec(reads=[], writes=["users"])
|
||
|
||
with pytest.raises(AppError) as exc:
|
||
validate_declaration(spec)
|
||
|
||
assert exc.value.code is ErrorCode.VALIDATION
|