laily
f3cd56b78e
Use ~/.pouch, the pouch CLI, and .pouch.yaml as the SSOT container. Keep the inner skills/ packages, and store ACK project state in .pouch/ack instead of docs/ack.
592 lines
22 KiB
Python
592 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import yaml
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
RUNNER = REPO_ROOT / "skills" / "ack" / "scripts" / "run_verification.py"
|
|
sys.path.insert(0, str(RUNNER.parent))
|
|
RUNNER_SPEC = importlib.util.spec_from_file_location("ack_run_verification", RUNNER)
|
|
assert RUNNER_SPEC is not None and RUNNER_SPEC.loader is not None
|
|
RUNNER_MODULE = importlib.util.module_from_spec(RUNNER_SPEC)
|
|
RUNNER_SPEC.loader.exec_module(RUNNER_MODULE)
|
|
|
|
|
|
def knowledge_with_target(path: str, args: list[str]) -> dict:
|
|
return {
|
|
"version": 1,
|
|
"updatedAt": "2026-07-31T12:00:00+08:00",
|
|
"project": {"name": "demo"},
|
|
"verificationRegistry": {
|
|
"reviewed-check": {
|
|
"path": path,
|
|
"args": args,
|
|
}
|
|
},
|
|
"entries": [],
|
|
}
|
|
|
|
|
|
class AckVerificationRunnerTests(unittest.TestCase):
|
|
def run_check(
|
|
self,
|
|
project: Path,
|
|
knowledge: dict,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
knowledge_path = project / ".pouch" / "ack" / "knowledge.yaml"
|
|
knowledge_path.parent.mkdir(parents=True, exist_ok=True)
|
|
knowledge_path.write_text(
|
|
yaml.safe_dump(knowledge, allow_unicode=True),
|
|
encoding="utf-8",
|
|
)
|
|
return subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(RUNNER),
|
|
str(knowledge_path),
|
|
"reviewed-check",
|
|
"--project-root",
|
|
str(project),
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_executes_reviewed_target_with_structured_args(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
target = project / "checks" / "write-marker"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(
|
|
"#!/bin/sh\nprintf 'passed' > \"$1\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
|
|
result = self.run_check(
|
|
project,
|
|
knowledge_with_target(
|
|
"checks/write-marker",
|
|
["verification-marker.txt"],
|
|
),
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(
|
|
(project / "verification-marker.txt").read_text(encoding="utf-8"),
|
|
"passed",
|
|
)
|
|
|
|
def test_exposes_project_rooted_execution_contract(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
target = project / "checks" / "show-context"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(
|
|
"#!/bin/sh\n"
|
|
"printf '%s\\n%s\\n%s\\n%s\\n' "
|
|
'"$PWD" "$ACK_PROJECT_ROOT" "$ACK_VERIFICATION_REF" '
|
|
'"$ACK_VERIFICATION_PATH" > "$1"\n',
|
|
encoding="utf-8",
|
|
)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
|
|
result = self.run_check(
|
|
project,
|
|
knowledge_with_target(
|
|
"checks/show-context",
|
|
["verification-context.txt"],
|
|
),
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
context = (
|
|
(project / "verification-context.txt")
|
|
.read_text(encoding="utf-8")
|
|
.splitlines()
|
|
)
|
|
self.assertEqual(context[0], str(project))
|
|
self.assertRegex(context[1], r"^/(?:proc/self|dev)/fd/[0-9]+$")
|
|
self.assertEqual(context[2:], ["reviewed-check", "checks/show-context"])
|
|
|
|
def test_rejects_missing_or_non_executable_target(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
project.mkdir()
|
|
|
|
missing = self.run_check(
|
|
project,
|
|
knowledge_with_target("checks/missing", []),
|
|
)
|
|
self.assertEqual(missing.returncode, 1)
|
|
self.assertIn("不存在", missing.stderr)
|
|
|
|
target = project / "checks" / "not-executable"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
target.chmod(target.stat().st_mode & ~0o111)
|
|
not_executable = self.run_check(
|
|
project,
|
|
knowledge_with_target("checks/not-executable", []),
|
|
)
|
|
|
|
self.assertEqual(not_executable.returncode, 1)
|
|
self.assertIn("不可执行", not_executable.stderr)
|
|
|
|
def test_rejects_project_root_that_is_wider_than_knowledge_project(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
outside = base / "outside"
|
|
outside.mkdir()
|
|
target = outside / "unsafe"
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
knowledge_path = project / ".pouch" / "ack" / "knowledge.yaml"
|
|
knowledge_path.parent.mkdir(parents=True)
|
|
knowledge_path.write_text(
|
|
yaml.safe_dump(
|
|
knowledge_with_target("outside/unsafe", []),
|
|
allow_unicode=True,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(RUNNER),
|
|
str(knowledge_path),
|
|
"reviewed-check",
|
|
"--project-root",
|
|
str(base),
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertIn("推断的项目根目录", result.stderr)
|
|
|
|
def test_rejects_duplicate_registry_ids_before_execution(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
checks = project / "checks"
|
|
checks.mkdir(parents=True)
|
|
marker = project / "unsafe-marker"
|
|
for name in ("safe", "unsafe"):
|
|
target = checks / name
|
|
target.write_text(
|
|
f"#!/bin/sh\nprintf '{name}' > {marker}\n",
|
|
encoding="utf-8",
|
|
)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
knowledge_path = project / ".pouch" / "ack" / "knowledge.yaml"
|
|
knowledge_path.parent.mkdir(parents=True)
|
|
knowledge_path.write_text(
|
|
"""\
|
|
version: 1
|
|
updatedAt: "2026-07-31T12:00:00+08:00"
|
|
project:
|
|
name: demo
|
|
verificationRegistry:
|
|
reviewed-check:
|
|
path: checks/safe
|
|
args: []
|
|
reviewed-check:
|
|
path: checks/unsafe
|
|
args: []
|
|
entries: []
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(RUNNER),
|
|
str(knowledge_path),
|
|
"reviewed-check",
|
|
"--project-root",
|
|
str(project),
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("duplicate key", result.stderr)
|
|
self.assertFalse(marker.exists())
|
|
|
|
@unittest.skipUnless(hasattr(os, "symlink"), "requires symlink support")
|
|
def test_rejects_non_authoritative_knowledge_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
ack_dir = project / ".pouch" / "ack"
|
|
ack_dir.mkdir(parents=True)
|
|
document = yaml.safe_dump(
|
|
knowledge_with_target("checks/reviewed", []),
|
|
allow_unicode=True,
|
|
)
|
|
|
|
for label, knowledge_path in (
|
|
("external", base / "attacker-knowledge.yaml"),
|
|
("alternate", ack_dir / "alternate.yaml"),
|
|
):
|
|
with self.subTest(label=label):
|
|
knowledge_path.write_text(document, encoding="utf-8")
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(RUNNER),
|
|
str(knowledge_path),
|
|
"reviewed-check",
|
|
"--project-root",
|
|
str(project),
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertIn("权威知识库", result.stderr)
|
|
|
|
outside = base / "outside-knowledge.yaml"
|
|
outside.write_text(document, encoding="utf-8")
|
|
canonical = ack_dir / "knowledge.yaml"
|
|
canonical.symlink_to(outside)
|
|
symlinked = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(RUNNER),
|
|
str(canonical),
|
|
"reviewed-check",
|
|
"--project-root",
|
|
str(project),
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(symlinked.returncode, 2)
|
|
self.assertIn("路径包含软链接", symlinked.stderr)
|
|
|
|
def test_authoritative_knowledge_is_read_from_an_opened_snapshot(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
canonical = project / ".pouch" / "ack" / "knowledge.yaml"
|
|
canonical.parent.mkdir(parents=True)
|
|
safe_document = knowledge_with_target("checks/safe", [])
|
|
forged_document = knowledge_with_target("checks/danger", [])
|
|
canonical.write_text(
|
|
yaml.safe_dump(safe_document, allow_unicode=True),
|
|
encoding="utf-8",
|
|
)
|
|
forged_path = base / "forged.yaml"
|
|
forged_path.write_text(
|
|
yaml.safe_dump(forged_document, allow_unicode=True),
|
|
encoding="utf-8",
|
|
)
|
|
original_reader = RUNNER_MODULE._read_stable_bytes
|
|
|
|
def swap_after_open(
|
|
source_fd: int,
|
|
*,
|
|
maximum: int,
|
|
) -> tuple[bytes | None, str | None]:
|
|
canonical.unlink()
|
|
canonical.symlink_to(forged_path)
|
|
return original_reader(source_fd, maximum=maximum)
|
|
|
|
with mock.patch.object(
|
|
RUNNER_MODULE,
|
|
"_read_stable_bytes",
|
|
side_effect=swap_after_open,
|
|
):
|
|
data, error = RUNNER_MODULE.load_authoritative_knowledge(project)
|
|
|
|
self.assertIsNone(error)
|
|
self.assertEqual(
|
|
data["verificationRegistry"]["reviewed-check"]["path"],
|
|
"checks/safe",
|
|
)
|
|
|
|
def test_authoritative_knowledge_rejects_same_inode_change_during_read(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
canonical = project / ".pouch" / "ack" / "knowledge.yaml"
|
|
canonical.parent.mkdir(parents=True)
|
|
canonical.write_text(
|
|
yaml.safe_dump(
|
|
knowledge_with_target("checks/safe", []),
|
|
allow_unicode=True,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
source_fd, error = RUNNER_MODULE._open_regular_beneath(
|
|
project,
|
|
".pouch/ack/knowledge.yaml",
|
|
require_executable=False,
|
|
)
|
|
self.assertIsNone(error)
|
|
self.assertIsNotNone(source_fd)
|
|
assert source_fd is not None
|
|
original_read = os.read
|
|
changed = False
|
|
|
|
def mutate_during_read(
|
|
file_descriptor: int,
|
|
size: int,
|
|
) -> bytes:
|
|
nonlocal changed
|
|
chunk = original_read(file_descriptor, size)
|
|
if not changed:
|
|
changed = True
|
|
canonical.write_text(
|
|
yaml.safe_dump(
|
|
knowledge_with_target("checks/danger", []),
|
|
allow_unicode=True,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return chunk
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
RUNNER_MODULE.os,
|
|
"read",
|
|
side_effect=mutate_during_read,
|
|
):
|
|
content, read_error = RUNNER_MODULE._read_stable_bytes(
|
|
source_fd,
|
|
maximum=RUNNER_MODULE.MAX_KNOWLEDGE_BYTES,
|
|
)
|
|
finally:
|
|
os.close(source_fd)
|
|
|
|
self.assertIsNone(content)
|
|
self.assertIn("读取期间发生变化", read_error)
|
|
|
|
@unittest.skipUnless(hasattr(os, "symlink"), "requires symlink support")
|
|
def test_rejects_target_that_resolves_outside_project(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
outside = base / "outside"
|
|
project.mkdir()
|
|
outside.mkdir()
|
|
target = outside / "unsafe"
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
(project / "checks").symlink_to(outside, target_is_directory=True)
|
|
|
|
result = self.run_check(
|
|
project,
|
|
knowledge_with_target("checks/unsafe", []),
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("不能包含 symlink", result.stderr)
|
|
|
|
@unittest.skipUnless(
|
|
hasattr(os, "O_NOFOLLOW") and Path("/proc/self/fd").is_dir(),
|
|
"requires fd-based POSIX execution",
|
|
)
|
|
def test_opened_target_cannot_be_swapped_before_execution(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
target = project / "checks" / "reviewed"
|
|
outside = base / "outside"
|
|
marker = project / "marker.txt"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(
|
|
"#!/bin/sh\nprintf 'reviewed' > \"$1\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
outside.write_text(
|
|
"#!/bin/sh\nprintf 'swapped' > \"$1\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
outside.chmod(outside.stat().st_mode | 0o111)
|
|
data = knowledge_with_target("checks/reviewed", [str(marker)])
|
|
|
|
target_fd, target_args, error = RUNNER_MODULE.open_target(
|
|
data,
|
|
"reviewed-check",
|
|
project,
|
|
)
|
|
self.assertIsNone(error)
|
|
self.assertIsNotNone(target_fd)
|
|
self.assertIsNotNone(target_args)
|
|
assert target_fd is not None and target_args is not None
|
|
target.unlink()
|
|
target.symlink_to(outside)
|
|
executable = RUNNER_MODULE._fd_executable_path(target_fd)
|
|
self.assertIsNotNone(executable)
|
|
try:
|
|
completed = subprocess.run(
|
|
[str(executable), *target_args],
|
|
cwd=project,
|
|
pass_fds=(target_fd,),
|
|
check=False,
|
|
)
|
|
finally:
|
|
os.close(target_fd)
|
|
|
|
self.assertEqual(completed.returncode, 0)
|
|
self.assertEqual(marker.read_text(encoding="utf-8"), "reviewed")
|
|
|
|
@unittest.skipUnless(
|
|
hasattr(os, "O_NOFOLLOW") and Path("/proc/self/fd").is_dir(),
|
|
"requires fd-based POSIX execution",
|
|
)
|
|
def test_opened_target_snapshot_ignores_same_inode_rewrite(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
target = project / "checks" / "reviewed"
|
|
marker = project / "marker.txt"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(
|
|
"#!/bin/sh\nprintf 'reviewed' > \"$1\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
data = knowledge_with_target("checks/reviewed", [str(marker)])
|
|
|
|
target_fd, target_args, error = RUNNER_MODULE.open_target(
|
|
data,
|
|
"reviewed-check",
|
|
project,
|
|
)
|
|
self.assertIsNone(error)
|
|
self.assertIsNotNone(target_fd)
|
|
self.assertIsNotNone(target_args)
|
|
assert target_fd is not None and target_args is not None
|
|
target.write_text(
|
|
"#!/bin/sh\nprintf 'mutated' > \"$1\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
executable = RUNNER_MODULE._fd_executable_path(target_fd)
|
|
self.assertIsNotNone(executable)
|
|
try:
|
|
completed = subprocess.run(
|
|
[str(executable), *target_args],
|
|
cwd=project,
|
|
pass_fds=(target_fd,),
|
|
check=False,
|
|
)
|
|
finally:
|
|
os.close(target_fd)
|
|
|
|
self.assertEqual(completed.returncode, 0)
|
|
self.assertEqual(marker.read_text(encoding="utf-8"), "reviewed")
|
|
|
|
@unittest.skipUnless(
|
|
hasattr(os, "O_NOFOLLOW") and Path("/proc/self/fd").is_dir(),
|
|
"requires fd-based POSIX execution",
|
|
)
|
|
def test_project_root_replacement_cannot_swap_target_or_cwd(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
original = base / "original-project"
|
|
target = project / "checks" / "reviewed"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(
|
|
"#!/bin/sh\nprintf 'reviewed' > marker.txt\n",
|
|
encoding="utf-8",
|
|
)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
knowledge_path = project / ".pouch" / "ack" / "knowledge.yaml"
|
|
knowledge_path.parent.mkdir(parents=True)
|
|
knowledge_path.write_text(
|
|
yaml.safe_dump(
|
|
knowledge_with_target("checks/reviewed", []),
|
|
allow_unicode=True,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
original_loader = RUNNER_MODULE.load_authoritative_knowledge
|
|
|
|
def replace_root_after_knowledge(
|
|
project_root: Path | int,
|
|
) -> tuple[dict | None, str | None]:
|
|
data, error = original_loader(project_root)
|
|
project.rename(original)
|
|
replacement = project / "checks" / "reviewed"
|
|
replacement.parent.mkdir(parents=True)
|
|
replacement.write_text(
|
|
"#!/bin/sh\nprintf 'swapped' > marker.txt\n",
|
|
encoding="utf-8",
|
|
)
|
|
replacement.chmod(replacement.stat().st_mode | 0o111)
|
|
return data, error
|
|
|
|
with mock.patch.object(
|
|
RUNNER_MODULE,
|
|
"load_authoritative_knowledge",
|
|
side_effect=replace_root_after_knowledge,
|
|
):
|
|
result = RUNNER_MODULE.main(
|
|
[
|
|
str(knowledge_path),
|
|
"reviewed-check",
|
|
"--project-root",
|
|
str(project),
|
|
]
|
|
)
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertEqual(
|
|
(original / "marker.txt").read_text(encoding="utf-8"),
|
|
"reviewed",
|
|
)
|
|
self.assertFalse((project / "marker.txt").exists())
|
|
|
|
def test_rejects_oversized_verification_target_before_snapshot(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
target = project / "checks" / "oversized"
|
|
target.parent.mkdir(parents=True)
|
|
with target.open("wb") as target_file:
|
|
target_file.truncate(RUNNER_MODULE.MAX_TARGET_BYTES + 1)
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
|
|
target_fd, target_args, error = RUNNER_MODULE.open_target(
|
|
knowledge_with_target("checks/oversized", []),
|
|
"reviewed-check",
|
|
project,
|
|
)
|
|
|
|
self.assertIsNone(target_fd)
|
|
self.assertIsNone(target_args)
|
|
self.assertIn("超过大小上限", error)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|