Skip to content

Commit 79d70ca

Browse files
committed
test: add adversarial gate-bypass tests, document the confirmation gate's trust boundary
1 parent 3035b68 commit 79d70ca

3 files changed

Lines changed: 124 additions & 1 deletion

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,14 @@
44

55
## [Unreleased]
66

7-
_本分支(`refactor/runner-fixes-docs-consolidation`)的改动记录见下文;发布时再并入正式版本号。_
7+
_本区记录 `[1.0.0]` 之后、尚未发布的所有改动;发布时把本区内容并入新的版本号。_
88

99
### Added
1010

1111
- `tests/test_runner.py`:覆盖 pipeline runner 的参数转发、门禁拦截、运行状态文件等行为
1212
- `tests/test_fetch_aihot.py`:覆盖 `fetch_aihot.py` 的 CLI 与 `fetch_json` 重试逻辑
13+
- `tests/test_gate_adversarial.py`:8 个对抗式用例,固化“试图绕过用户确认闸门”的几种走法(跳过步骤 03、只传 `--confirm`、过期确认、过期活人感门禁、损坏的运行状态文件),任一步被改坏都会红
14+
- `skills/tryworld-paper/references/workflow.md` §1.4:写明闸门的技术边界——`.prose_pass` / `.confirmed` 只绑 draft 的 sha256,防漂移不防同用户伪造,因此“等用户确认”必须是对话里真实发生的动作
1315

1416
### Changed
1517

‎skills/tryworld-paper/references/workflow.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,8 @@ python -X utf8 "$env:USERPROFILE\.agents\skills\tryworld-paper\scripts\check_pro
8686
- 全程无数据堆砌:每个数据/概念都要有"人话翻译"和"与观众的关系"。
8787
4. **未获用户确认,禁止进入 ## 2 配音及后续所有制作步骤**;用户确认后再继续。
8888

89+
闸门的技术边界(必须知道):`.prose_pass` / `.confirmed` 是 `work/` 下绑定了 draft 内容 sha256 的普通 JSON 文件,由 `pipeline/state.py` 读写。它能防住"草稿改了但没重新过门禁"这类漂移与误操作,**但不能防住同一用户/同一进程伪造标记**——所以"停下来等用户确认"必须是真实发生在对话里的动作,不能由 agent 自己补一个 `--confirm` 或手写标记文件来代替。tests/test_gate_adversarial.py 固化了一组绕过尝试(跳过步骤 03、只传 `--confirm`、过期确认等),回归时一并跑。
90+
8991
口播稿约定(保证配音无生硬停顿/卡顿):
9092

9193
- 句子以句号/问号/感叹号/省略号结束;句内不要为排版硬换行(脚本会自动把句子中间的换行合并,不会产生停顿)。

‎tests/test_gate_adversarial.py‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
"""Adversarial tests for the pipeline gates: can a run skip user confirmation?
2+
3+
These encode the attacks a careless operator (or an over-eager agent) would try,
4+
so a future refactor cannot quietly turn the confirmation gate into a no-op.
5+
6+
Offline only: every case here fails *before* edge-tts / ffmpeg / hyperframes is
7+
invoked, because the marker checks run first.
8+
9+
Known limitation (by design, not covered here): `.confirmed` / `.prose_pass` are
10+
plain JSON files under `work/` bound to the draft's sha256, so a process running
11+
as the same user can forge them. The gate therefore assumes the *human* really
12+
did confirm outside the agent; it protects against accidents and drift, not
13+
against a determined same-user forgery.
14+
"""
15+
import json
16+
import subprocess
17+
import sys
18+
from pathlib import Path
19+
20+
REPO = Path(__file__).resolve().parents[1]
21+
CLEAN_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "clean_prose.md"
22+
23+
24+
def run_runner(project_dir: Path, *extra: str) -> subprocess.CompletedProcess:
25+
return subprocess.run(
26+
[sys.executable, "-X", "utf8", "-m", "pipeline.runner",
27+
"--project-dir", str(project_dir), *extra],
28+
cwd=REPO, capture_output=True, text=True, encoding="utf-8", errors="replace",
29+
)
30+
31+
32+
def fresh_project(tmp_path: Path, name: str) -> Path:
33+
import shutil
34+
p = tmp_path / name
35+
if p.exists():
36+
shutil.rmtree(p)
37+
(p / "work").mkdir(parents=True)
38+
return p
39+
40+
41+
def steps_1_2(project_dir: Path) -> subprocess.CompletedProcess:
42+
return run_runner(project_dir, "--steps", "1,2", "--script", str(CLEAN_FIXTURE))
43+
44+
45+
class TestGateBypassAttempts:
46+
def test_step_04_alone_is_blocked(self, tmp_path):
47+
p = fresh_project(tmp_path, "a")
48+
r = run_runner(p, "--steps", "4")
49+
assert r.returncode != 0
50+
assert "confirm" in (r.stdout + r.stderr).lower()
51+
assert not (p / "work" / ".confirmed").exists()
52+
53+
def test_step_05_alone_is_blocked(self, tmp_path):
54+
p = fresh_project(tmp_path, "b")
55+
r = run_runner(p, "--steps", "5")
56+
assert r.returncode != 0
57+
assert "confirm" in (r.stdout + r.stderr).lower()
58+
59+
def test_skipping_step_03_with_confirm_still_blocks_step_04(self, tmp_path):
60+
"""--confirm on the runner must not substitute for running step 03."""
61+
p = fresh_project(tmp_path, "c")
62+
r = run_runner(p, "--steps", "1,2,4", "--confirm", "--script", str(CLEAN_FIXTURE))
63+
assert r.returncode != 0
64+
combined = (r.stdout + r.stderr).lower()
65+
assert "confirm" in combined
66+
# steps 1-2 still did their job; only the gate stopped the run
67+
assert (p / "work" / "draft.md").exists()
68+
assert (p / "work" / ".prose_pass").exists()
69+
assert not (p / "work" / ".confirmed").exists()
70+
71+
def test_step_03_without_a_draft_fails(self, tmp_path):
72+
p = fresh_project(tmp_path, "d")
73+
r = run_runner(p, "--steps", "3", "--confirm")
74+
assert r.returncode != 0
75+
assert "draft" in (r.stdout + r.stderr).lower()
76+
77+
def test_step_03_failure_stops_the_run_before_step_04(self, tmp_path):
78+
p = fresh_project(tmp_path, "e")
79+
r = run_runner(p, "--steps", "3,4", "--confirm")
80+
assert r.returncode != 0
81+
assert "[03]" in r.stdout
82+
assert "[04]" not in r.stdout
83+
84+
def test_stale_confirmation_blocks_step_04(self, tmp_path):
85+
p = fresh_project(tmp_path, "f")
86+
assert steps_1_2(p).returncode == 0
87+
assert run_runner(p, "--steps", "3", "--confirm").returncode == 0
88+
draft = p / "work" / "draft.md"
89+
draft.write_text(draft.read_text(encoding="utf-8") + "\n未确认的新增内容。\n", encoding="utf-8")
90+
91+
r = run_runner(p, "--steps", "4")
92+
assert r.returncode != 0
93+
assert "stale" in (r.stdout + r.stderr).lower()
94+
95+
def test_stale_prose_marker_blocks_step_03(self, tmp_path):
96+
"""Editing the draft after the prose gate must force step 02 again."""
97+
p = fresh_project(tmp_path, "g")
98+
assert steps_1_2(p).returncode == 0
99+
draft = p / "work" / "draft.md"
100+
draft.write_text(draft.read_text(encoding="utf-8") + "\n未过门禁的新增内容。\n", encoding="utf-8")
101+
102+
r = run_runner(p, "--steps", "3", "--confirm")
103+
assert r.returncode != 0
104+
assert "prose" in (r.stdout + r.stderr).lower() or "门禁" in (r.stdout + r.stderr)
105+
assert not (p / "work" / ".confirmed").exists()
106+
107+
def test_corrupt_run_state_does_not_affect_any_gate(self, tmp_path):
108+
p = fresh_project(tmp_path, "h")
109+
assert steps_1_2(p).returncode == 0
110+
(p / "work" / "pipeline_run.json").write_text("{ not json", encoding="utf-8")
111+
112+
r = run_runner(p, "--steps", "3")
113+
assert r.returncode != 0
114+
assert "confirm" in (r.stdout + r.stderr).lower()
115+
116+
r = run_runner(p, "--steps", "3", "--confirm")
117+
assert r.returncode == 0
118+
state = json.loads((p / "work" / "pipeline_run.json").read_text(encoding="utf-8"))
119+
assert state["result"] == "complete"

0 commit comments

Comments
 (0)