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 / "docs" / "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 / "docs" / "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 / "docs" / "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 / "docs" / "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 / "docs" / "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 / "docs" / "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, "docs/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 / "docs" / "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()