diff --git a/docs/how-to/formal-core.md b/docs/how-to/formal-core.md index d8a5a7d4..9905b3a5 100644 --- a/docs/how-to/formal-core.md +++ b/docs/how-to/formal-core.md @@ -64,6 +64,20 @@ JSON-отчёты, логи, воспроизводимые контрприме артефактом задачи `formal core (Kani)` внутри обязательного CI. Публикация требует эту задачу в успешном canonical push-run точного коммита. Старый результат не подменяет новый запуск. Нечистый checkout не получает `source_commit` в квитанции. +При проверке чистоты локальный Git fsmonitor отключается, целостность достижимых +объектов Git проверяется через `git fsck --strict`, а сырые байты всех +отслеживаемых файлов сравниваются с blob указанного коммита. В том числе +`verified_source_sha256` формальных исходников привязаны к этим blob. +Физическое поддерево `crates/labcolors-core` дополнительно сверяется с Git-деревом, +включая игнорируемые файлы и пустые каталоги; workspace `target/` остаётся допустимым. +Отслеживаемые символьные ссылки внутри Core и среди формальных входов вне Core +отклоняются: Git связывает текст ссылки, а компилятор или верификатор может прочитать +изменяемую игнорируемую цель. Посторонние ссылки вне Core остаются допустимыми. +Это обнаруживает правку, скрытую настройками статуса или индексом Git. Checkout, +меняющий байты исходников при извлечении, не получает `source_commit`. Для Linux CI +нужен checkout с LF: перевод LF в CRLF при извлечении также не совпадёт с blob. Для этой +проверки нужен Git 2.35.2 или новее: более ранние версии трактуют +`core.fsmonitor=false` как путь к hook, поэтому верификатор завершится отказом. ## Обновление и пределы diff --git a/proof/artifact/tests.json b/proof/artifact/tests.json index dfad25bf..0f566aaa 100644 --- a/proof/artifact/tests.json +++ b/proof/artifact/tests.json @@ -600,7 +600,7 @@ "start": "proof/region/v1/tests" }, { - "count": 272, + "count": 278, "load_errors": [], "names": [ "ci_workflow_binding.TestCiWorkflowBinding.test_accepts_exact_local_call_at_workflow_sha", @@ -850,6 +850,12 @@ "test_verify_clean_set_receipt.ReceiptHostileTests.test_unknown_receipt_field_is_rejected", "test_verify_clean_set_receipt.ReceiptHostileTests.test_valid_receipt_binds_committed_research_and_product_codec", "test_verify_clean_set_receipt.ReceiptHostileTests.test_wrong_research_commit_is_rejected_before_blob_lookup", + "test_verify_core_formal.FormalReportTests.test_assembly_rejects_index_hidden_source_changes", + "test_verify_core_formal.FormalReportTests.test_checkout_ignores_foreign_git_repository_environment", + "test_verify_core_formal.FormalReportTests.test_checkout_rejects_corrupt_blob_even_when_status_and_bytes_match", + "test_verify_core_formal.FormalReportTests.test_checkout_rejects_ignored_core_extras_but_allows_workspace_target", + "test_verify_core_formal.FormalReportTests.test_checkout_rejects_symlinks_for_each_formal_input_outside_core", + "test_verify_core_formal.FormalReportTests.test_checkout_rejects_tracked_core_symlink_to_ignored_target", "test_verify_core_formal.FormalReportTests.test_complete_result_passes", "test_verify_core_formal.FormalReportTests.test_completed_wrapper_still_cleans_its_solver_group", "test_verify_core_formal.FormalReportTests.test_deleted_semantic_assertion_or_cover_is_rejected", @@ -882,7 +888,7 @@ "start": "scripts" } ], - "record_sha256": "93ed27c874673dea462679f7d54b9d1dc091cbca376436f5506ee4b334a2739c", + "record_sha256": "2e62131734917268357705f8bacdea4216290ee6fe15e36171165748286627be", "rust": { "enabled": [ "actual_process_cannot_succeed_when_stdout_refuses_bytes: test", diff --git a/proof/artifact/tree.json b/proof/artifact/tree.json index 8d8c4bfd..ab85fad5 100644 --- a/proof/artifact/tree.json +++ b/proof/artifact/tree.json @@ -478,18 +478,18 @@ "scripts/test_program_public_surface.py": "cd04ede56d9d500850250577bcae9e992701185c", "scripts/test_transport_distribution.mjs": "c532d55b086dc324ef6477b0d7b0352d16d4e395", "scripts/test_verify_clean_set_receipt.py": "a8051388a28b68fc4fb984609814e79ddb45be53", - "scripts/test_verify_core_formal.py": "f92c6c6a290e9413a3048e63714e0ec40219fc0a", + "scripts/test_verify_core_formal.py": "85372064d897d0bb7048b8aa7436817eb6a74e67", "scripts/test_wcag22_source_binding.py": "381b2c5685cb4022367f0465d02b044f100ea7f8", "scripts/transport-distribution.mjs": "77ad862e3ea464cb0824e519b6583e3090b85f2a", "scripts/verify-package-release.mjs": "8b0815717dc02e6af23341e7f2e83b02f4c1a120", "scripts/verify_clean_set_receipt.py": "05fbd7254d592b8cb4d87bc53297ab4f47429e77", - "scripts/verify_core_formal.py": "2a9ce2e87a6647c98ef1f894d6ab36517c3e4a0d", + "scripts/verify_core_formal.py": "fb5fb76667a4fc9fd5c9e1654002a416c280acb5", "scripts/verify_point_support_surplus.py": "24785f2d6cc56c88e12b0c50643f8f25365b3f6c", "scripts/verify_program_public_surface.py": "613740af9951edf658817a530e216a3ddec739e5", "scripts/verify_wcag22_neutral_axis.py": "13e7a1aa7621ff96435849d2f340ab128377a867", "scripts/verify_wcag22_q55.py": "39d7a4ae3355ae0907f5745c612f9b928cd1ff01" }, - "record_sha256": "885529180af1868482db541f60b9eb8b1dc40e698e7fb43b0a3ee4c26e5c3325", + "record_sha256": "7766ec8c003140a6d46f2abe9dafd030f68fb9d7a7b2df1e3bce1c66ad299274", "roots": [ ".github/workflows", "Cargo.toml", diff --git a/scripts/test_verify_core_formal.py b/scripts/test_verify_core_formal.py index f92c6c6a..85372064 100644 --- a/scripts/test_verify_core_formal.py +++ b/scripts/test_verify_core_formal.py @@ -3,16 +3,20 @@ import copy import json +import os import unittest from unittest import mock import signal import subprocess import tempfile +import zlib from pathlib import Path from verify_core_formal import ( CONTRACTS, KANI_VERSION, - assemble_fragments, digest, selected_mutants, selected_positive_harnesses, + assemble_fragments, assert_snapshot, checkout_is_clean, checkout_snapshot, digest, formal_source_files, + fragment_base, + selected_mutants, selected_positive_harnesses, validate_report, validate_mutant, run_kani, ) @@ -80,6 +84,279 @@ def valid_subset_report(harnesses): class FormalReportTests(unittest.TestCase): + def test_checkout_rejects_corrupt_blob_even_when_status_and_bytes_match(self): + with tempfile.TemporaryDirectory() as temporary: + checkout = Path(temporary) + source = checkout / "formal.rs" + source.write_bytes(b"trusted\n") + + def git(*arguments): + return subprocess.check_output(["git", *arguments], cwd=checkout) + + git("init", "-q") + git("config", "core.autocrlf", "false") + git("config", "commit.gpgsign", "false") + git("add", "--", source.name) + git("-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", "initial") + fixed_time = 1_600_000_000_000_000_000 + os.utime(source, ns=(fixed_time, fixed_time)) + git("add", "--", source.name) + git("config", "core.trustctime", "false") + git("update-index", "--refresh") + commit = git("rev-parse", "HEAD").decode("ascii").strip() + + with mock.patch("verify_core_formal.ROOT", checkout), \ + mock.patch("verify_core_formal.core_tree_has_no_physical_extras", return_value=True): + identities = {source.name: digest(source)} + self.assertTrue(checkout_is_clean(identities, commit)) + + original = source.stat() + source.write_bytes(b"trusteX\n") + os.utime(source, ns=(original.st_atime_ns, original.st_mtime_ns)) + oid = git("rev-parse", f"HEAD:{source.name}").decode("ascii").strip() + object_path = checkout / ".git" / "objects" / oid[:2] / oid[2:] + self.assertTrue(object_path.is_file()) + object_path.chmod(0o600) + payload = source.read_bytes() + object_path.write_bytes(zlib.compress(b"blob " + str(len(payload)).encode("ascii") + b"\0" + payload)) + + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + self.assertNotEqual(subprocess.run( + ["git", "fsck", "--strict", "--no-reflogs", "--no-progress", "--no-dangling", commit], + cwd=checkout, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ).returncode, 0) + self.assertFalse(checkout_is_clean({source.name: digest(source)}, commit)) + + def test_checkout_ignores_foreign_git_repository_environment(self): + root = Path(__file__).resolve().parent.parent + sources = {path.relative_to(root).as_posix(): path.read_bytes() for path in formal_source_files()} + oids = {path: f"{index:040x}".encode("ascii") for index, path in enumerate(sources, 1)} + foreign = { + "GIT_DIR": "/foreign/.git", + "GIT_WORK_TREE": "/foreign", + "GIT_INDEX_FILE": "/foreign/index", + "GIT_OBJECT_DIRECTORY": "/foreign/objects", + "GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": "core.repositoryformatversion", + "GIT_CONFIG_VALUE_0": "99", + } + def git_output(command, **kwargs): + if command[3] == "version": + return "git version 2.51.0" + if command[3] == "ls-files": + return "H source.rs\0" + if command[3] == "status": + return "" + if command[3] == "ls-tree": + return b"".join((b"100755" if (root / path).stat().st_mode & 0o100 else b"100644") + + b" blob " + oid + b"\t" + path.encode() + b"\0" + for path, oid in oids.items()) + if command[3] == "cat-file": + self.assertEqual(kwargs["input"], b"".join(oid + b"\n" for oid in oids.values())) + return b"".join(oid + b" blob " + str(len(sources[path])).encode() + b"\n" + + sources[path] + b"\n" for path, oid in oids.items()) + return "local-head" + + with mock.patch.dict(os.environ, foreign), \ + mock.patch("verify_core_formal.subprocess.check_output", side_effect=git_output) as command, \ + mock.patch("verify_core_formal.core_tree_has_no_physical_extras", return_value=True), \ + mock.patch("verify_core_formal.subprocess.run"): + identities, commit, clean = checkout_snapshot() + self.assertTrue(clean) + self.assertTrue(assert_snapshot(identities, commit)) + self.assertEqual(command.call_count, 14) + for call in command.call_args_list: + environment = call.kwargs["env"] + self.assertFalse(set(foreign) & set(environment)) + self.assertEqual(environment["GIT_NO_REPLACE_OBJECTS"], "1") + self.assertEqual(call.args[0][1:3], ["-c", "core.fsmonitor=false"]) + + with mock.patch("verify_core_formal.git_checkout_output", return_value="git version 2.35.1"): + with self.assertRaisesRegex(ValueError, "Git 2.35.2"): + checkout_is_clean() + + def test_checkout_rejects_ignored_core_extras_but_allows_workspace_target(self): + with tempfile.TemporaryDirectory() as temporary: + checkout = Path(temporary) + core = checkout / "crates/labcolors-core" + source = core / "src/lib.rs" + source.parent.mkdir(parents=True) + source.write_bytes(b"trusted source\n") + (checkout / ".gitignore").write_text("target/\n", encoding="ascii") + + def git(*arguments): + return subprocess.check_output(["git", *arguments], cwd=checkout) + + git("init", "-q") + git("config", "core.autocrlf", "false") + git("add", "--", ".gitignore", "crates/labcolors-core/src/lib.rs") + git("-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", "initial") + (checkout / "target").mkdir() + (checkout / "target/workspace-build.txt").write_bytes(b"ordinary build output\n") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + + real_run = subprocess.run + + def run_without_cargo(command, *args, **kwargs): + if command[:2] == ["cargo", "metadata"]: + return subprocess.CompletedProcess(command, 0) + return real_run(command, *args, **kwargs) + + with mock.patch("verify_core_formal.ROOT", checkout), \ + mock.patch("verify_core_formal.formal_source_files", return_value=[source]), \ + mock.patch("verify_core_formal.subprocess.run", side_effect=run_without_cargo): + identities, commit, clean = checkout_snapshot() + self.assertTrue(clean) + self.assertTrue(assert_snapshot(identities, commit)) + + ignored_dir = core / "target" + ignored_dir.mkdir() + (ignored_dir / "ignored-marker.txt").write_bytes(b"changes compiled source identity\n") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + self.assertFalse(checkout_snapshot()[2]) + self.assertFalse(assert_snapshot(identities, commit)) + + (ignored_dir / "ignored-marker.txt").unlink() + self.assertFalse(checkout_snapshot()[2], "even an empty extra directory changes the Core tree") + ignored_dir.rmdir() + self.assertTrue(checkout_snapshot()[2]) + + @unittest.skipUnless(os.name == "posix", "requires POSIX symlink semantics") + def test_checkout_rejects_tracked_core_symlink_to_ignored_target(self): + with tempfile.TemporaryDirectory() as temporary: + checkout = Path(temporary) + source = checkout / "crates/labcolors-core/src/numerics.rs" + source.parent.mkdir(parents=True) + source.write_bytes(b"regular source\n") + (checkout / ".gitignore").write_text("target/\n", encoding="ascii") + target = checkout / "target/numerics.rs" + target.parent.mkdir() + target.write_bytes(b"mutable source\n") + + def git(*arguments): + return subprocess.check_output(["git", *arguments], cwd=checkout) + + git("init", "-q") + git("config", "core.autocrlf", "false") + git("config", "core.symlinks", "true") + git("add", "--", ".gitignore", "crates/labcolors-core/src/numerics.rs") + git("-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", "regular source") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + + real_run = subprocess.run + + def run_without_cargo(command, *args, **kwargs): + if command[:2] == ["cargo", "metadata"]: + return subprocess.CompletedProcess(command, 0) + return real_run(command, *args, **kwargs) + + with mock.patch("verify_core_formal.ROOT", checkout), \ + mock.patch("verify_core_formal.formal_source_files", return_value=[source]), \ + mock.patch("verify_core_formal.subprocess.run", side_effect=run_without_cargo): + self.assertTrue(checkout_snapshot()[2], "regular Core source must remain admissible") + + source.unlink() + source.symlink_to("../../../target/numerics.rs") + git("add", "--", "crates/labcolors-core/src/numerics.rs") + git("-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", "tracked symlink") + mode = git("ls-files", "-s", "--", "crates/labcolors-core/src/numerics.rs").split(b" ", 1)[0] + self.assertEqual(mode, b"120000") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + before, _, clean = checkout_snapshot() + self.assertFalse(clean) + + target.write_bytes(b"changed while Git remains clean\n") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + after, _, clean = checkout_snapshot() + self.assertNotEqual(after, before) + self.assertFalse(clean) + + @unittest.skipUnless(os.name == "posix", "requires POSIX symlink semantics") + def test_checkout_rejects_symlinks_for_each_formal_input_outside_core(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + origin = root / "origin" + checkout = root / "checkout" + origin.mkdir() + manifest_bytes = (b'[package]\nname = "formal-source-link"\n' + b'version = "0.1.0"\nedition = "2021"\n') + script_bytes = b"CONTRACTS = {}\n" + (origin / "Cargo.toml").write_bytes(manifest_bytes) + (origin / ".gitignore").write_text("target/\n", encoding="ascii") + (origin / "src").mkdir() + (origin / "src/lib.rs").write_bytes(b"pub fn witness() {}\n") + (origin / "crates/labcolors-core").mkdir(parents=True) + (origin / "crates/labcolors-core/README.md").write_bytes(b"Core fixture\n") + (origin / "scripts").mkdir() + (origin / "scripts/formal_contracts.py").write_bytes(script_bytes) + (origin / "docs").mkdir() + (origin / "docs/guide.md").symlink_to("../target/guide.md") + + def git(directory, *arguments): + return subprocess.check_output(["git", *arguments], cwd=directory) + + git(origin, "init", "-q") + git(origin, "config", "core.autocrlf", "false") + git(origin, "config", "core.symlinks", "true") + subprocess.run(["cargo", "generate-lockfile", "--offline"], cwd=origin, + stdout=subprocess.DEVNULL, check=True) + git(origin, "add", "-A") + git(origin, "-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", "regular formal inputs") + git(root, "clone", "-q", str(origin), str(checkout)) + git(checkout, "config", "core.autocrlf", "false") + git(checkout, "config", "core.symlinks", "true") + manifest = checkout / "Cargo.toml" + script = checkout / "scripts/formal_contracts.py" + ignored = checkout / "target" + (ignored / "src").mkdir(parents=True) + (ignored / "Cargo.toml").write_bytes(manifest_bytes) + (ignored / "src/lib.rs").write_bytes(b"pub fn witness() {}\n") + (ignored / "formal_contracts.py").write_bytes(script_bytes) + (ignored / "guide.md").write_bytes(b"unrelated documentation\n") + + def commit(message, path): + git(checkout, "add", "--", path) + git(checkout, "-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", message) + + with mock.patch("verify_core_formal.ROOT", checkout), \ + mock.patch("verify_core_formal.formal_source_files", return_value=[manifest, script]): + identities, head, clean = checkout_snapshot() + self.assertTrue(clean, "regular formal inputs and unrelated docs link remain admissible") + self.assertEqual(fragment_base(identities, head, clean)["source_commit"], head) + + manifest.unlink() + manifest.symlink_to("target/Cargo.toml") + commit("linked manifest", "Cargo.toml") + self.assertEqual(git(checkout, "ls-files", "-s", "--", "Cargo.toml").split(b" ", 1)[0], b"120000") + self.assertEqual(git(checkout, "status", "--porcelain", "--untracked-files=all").strip(), b"") + before, head, clean = checkout_snapshot() + self.assertFalse(clean) + self.assertIsNone(fragment_base(before, head, clean)["source_commit"]) + + (ignored / "Cargo.toml").write_bytes(manifest_bytes + b'description = "mutable"\n') + self.assertEqual(git(checkout, "status", "--porcelain", "--untracked-files=all").strip(), b"") + after, head, clean = checkout_snapshot() + self.assertNotEqual(after, before) + self.assertFalse(clean) + + manifest.unlink() + manifest.write_bytes(manifest_bytes) + commit("regular manifest", "Cargo.toml") + self.assertTrue(checkout_snapshot()[2]) + + script.unlink() + script.symlink_to("../target/formal_contracts.py") + commit("linked formal contract", "scripts/formal_contracts.py") + self.assertEqual(git(checkout, "ls-files", "-s", "--", "scripts/formal_contracts.py").split(b" ", 1)[0], b"120000") + self.assertEqual(git(checkout, "status", "--porcelain", "--untracked-files=all").strip(), b"") + self.assertFalse(checkout_snapshot()[2]) + def test_complete_result_passes(self): validate_report(valid_report()) @@ -281,6 +558,11 @@ def test_fragment_assembly_requires_exact_complete_evidence(self): "verified_source_sha256": {"source.rs": "source-digest"}, } + def assemble_claimed_source(): + with mock.patch("verify_core_formal.checkout_snapshot", return_value=(common["verified_source_sha256"], common["checkout_commit"], True)), \ + mock.patch("verify_core_formal.assert_snapshot", return_value=True): + assemble_fragments(root, root) + for index in range(5): harnesses = selected_positive_harnesses(index, 5) path = root / f"positive-{index}.json" @@ -310,29 +592,179 @@ def test_fragment_assembly_requires_exact_complete_evidence(self): "negative_sha256": negatives, }), encoding="utf-8") - assemble_fragments(root, root) + with mock.patch("verify_core_formal.checkout_snapshot", return_value=({"source.rs": "actual-digest"}, common["checkout_commit"], True)): + with self.assertRaisesRegex(ValueError, "source state"): + assemble_fragments(root, root) + with mock.patch("verify_core_formal.checkout_snapshot", return_value=(common["verified_source_sha256"], "actual-head", True)): + with self.assertRaisesRegex(ValueError, "source state"): + assemble_fragments(root, root) + assemble_claimed_source() receipt = json.loads((root / "receipt.json").read_text(encoding="utf-8")) self.assertEqual(set(receipt["negative_sha256"]), {mutant[0] for mutant in MUTANTS}) self.assertEqual(receipt["harnesses"], sorted(CONTRACTS)) self.assertTrue(receipt["semantic_mutant_rejected"]) validate_report(json.loads((root / "positive.json").read_text(encoding="utf-8"))) + with mock.patch("verify_core_formal.checkout_snapshot", return_value=(common["verified_source_sha256"], common["checkout_commit"], False)): + with self.assertRaisesRegex(ValueError, "clean source state"): + assemble_fragments(root, root) + with mock.patch("verify_core_formal.checkout_snapshot", return_value=(common["verified_source_sha256"], common["checkout_commit"], True)), \ + mock.patch("verify_core_formal.assert_snapshot", return_value=False): + with self.assertRaisesRegex(ValueError, "became dirty"): + assemble_fragments(root, root) + unknown_fragment = root / "fragment-unknown.json" unknown_fragment.write_text(json.dumps({**common, "kind": "future"}), encoding="utf-8") with self.assertRaisesRegex(ValueError, "unknown formal fragment kind"): - assemble_fragments(root, root) + assemble_claimed_source() unknown_fragment.unlink() positive_fragment = root / "fragment-positive-4.json" saved_positive = positive_fragment.read_bytes() positive_fragment.unlink() with self.assertRaisesRegex(ValueError, "missing or duplicate positive shard"): - assemble_fragments(root, root) + assemble_claimed_source() positive_fragment.write_bytes(saved_positive) (root / "fragment-mutants-4.json").unlink() with self.assertRaisesRegex(ValueError, "missing or duplicate mutant shard"): - assemble_fragments(root, root) + assemble_claimed_source() + + def test_assembly_rejects_index_hidden_source_changes(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + checkout = root / "checkout" + evidence = root / "evidence" + checkout.mkdir() + evidence.mkdir() + source = checkout / "formal.rs" + source.write_bytes(b"trusted\n") + other = checkout / "other.txt" + other.write_bytes(b"trusted other\n") + + def git(*arguments): + return subprocess.check_output(["git", *arguments], cwd=checkout) + + git("init", "-q") + git("config", "core.autocrlf", "false") + git("add", "--", source.name, other.name) + git("-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "commit", "-qm", "initial") + + positive = evidence / "positive.json" + positive.write_text(json.dumps(valid_report()), encoding="utf-8") + negatives = {} + for mutant in MUTANTS: + name = mutant[0] + path = evidence / f"negative-{name}.json" + path.write_text(json.dumps({"mutant": name}), encoding="utf-8") + negatives[name] = digest(path) + + commit = git("rev-parse", "HEAD").decode("ascii").strip() + + def write_fragments(source_digest): + common = { + "fragment_schema": 1, + "kani_version": KANI_VERSION, + "checkout_commit": commit, + "source_commit": commit, + "working_tree_clean": True, + "verified_source_sha256": {source.name: source_digest}, + } + (evidence / "fragment-positive-0.json").write_text(json.dumps({ + **common, "kind": "positive", "shard_index": 0, "shard_count": 1, + "positive_sha256": digest(positive), "harnesses": sorted(CONTRACTS), + }), encoding="utf-8") + (evidence / "fragment-mutants-0.json").write_text(json.dumps({ + **common, "kind": "mutants", "shard_index": 0, "shard_count": 1, + "negative_sha256": negatives, + }), encoding="utf-8") + + real_run = subprocess.run + + def run_without_cargo(command, *args, **kwargs): + if command[:2] == ["cargo", "metadata"]: + return subprocess.CompletedProcess(command, 0) + return real_run(command, *args, **kwargs) + + with mock.patch("verify_core_formal.ROOT", checkout), \ + mock.patch("verify_core_formal.formal_source_files", return_value=[source]), \ + mock.patch("verify_core_formal.core_tree_has_no_physical_extras", return_value=True), \ + mock.patch("verify_core_formal.subprocess.run", side_effect=run_without_cargo): + write_fragments(digest(source)) + assemble_fragments(evidence, evidence) + receipt = evidence / "receipt.json" + self.assertEqual(json.loads(receipt.read_text(encoding="utf-8"))[ + "verified_source_sha256"], {source.name: digest(source)}) + receipt.unlink() + + monitor = checkout / ".git" / "fsmonitor.sh" + monitor.write_text("#!/bin/sh\nprintf 'token\\0'\n", encoding="ascii") + monitor.chmod(0o755) + git("config", "core.fsmonitor", ".git/fsmonitor.sh") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + source.write_bytes(b"tampered\n") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + self.assertEqual(git("ls-files", "-v", "--", source.name).strip(), b"H formal.rs") + write_fragments(digest(source)) + with self.assertRaisesRegex(ValueError, "clean source state"): + assemble_fragments(evidence, evidence) + self.assertFalse(receipt.exists()) + source.write_bytes(b"trusted\n") + other.write_bytes(b"trusteX other\n") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + write_fragments(digest(source)) + with self.assertRaisesRegex(ValueError, "clean source state"): + assemble_fragments(evidence, evidence) + self.assertFalse(receipt.exists()) + other.write_bytes(b"trusted other\n") + git("config", "--unset", "core.fsmonitor") + + for flag, clear in (("--assume-unchanged", "--no-assume-unchanged"), + ("--skip-worktree", "--no-skip-worktree")): + with self.subTest(flag=flag): + git("update-index", flag, "--", source.name) + source.write_bytes(b"tampered\n") + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + self.assertNotEqual(source.read_bytes(), git("show", f"HEAD:{source.name}")) + write_fragments(digest(source)) + with self.assertRaisesRegex(ValueError, "clean source state"): + assemble_fragments(evidence, evidence) + self.assertFalse(receipt.exists()) + git("update-index", clear, "--", source.name) + source.write_bytes(b"trusted\n") + + fixed_time = 1_600_000_000_000_000_000 + os.utime(source, ns=(fixed_time, fixed_time)) + os.utime(other, ns=(fixed_time, fixed_time)) + git("add", "--", source.name, other.name) + git("config", "core.trustctime", "false") + try: + git("update-index", "--refresh") + original = source.stat() + other_original = other.stat() + source.write_bytes(b"trusteX\n") + os.utime(source, ns=(original.st_atime_ns, original.st_mtime_ns)) + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + self.assertEqual(git("ls-files", "-v", "--", source.name).strip(), b"H formal.rs") + self.assertNotEqual(source.read_bytes(), git("show", f"HEAD:{source.name}")) + write_fragments(digest(source)) + with self.assertRaisesRegex(ValueError, "clean source state"): + assemble_fragments(evidence, evidence) + self.assertFalse(receipt.exists()) + source.write_bytes(b"trusted\n") + os.utime(source, ns=(original.st_atime_ns, original.st_mtime_ns)) + other.write_bytes(b"trusteX other\n") + os.utime(other, ns=(other_original.st_atime_ns, other_original.st_mtime_ns)) + self.assertEqual(git("status", "--porcelain", "--untracked-files=all").strip(), b"") + write_fragments(digest(source)) + with self.assertRaisesRegex(ValueError, "clean source state"): + assemble_fragments(evidence, evidence) + self.assertFalse(receipt.exists()) + finally: + git("config", "--unset", "core.trustctime") + source.write_bytes(b"trusted\n") + other.write_bytes(b"trusted other\n") if __name__ == "__main__": diff --git a/scripts/verify_core_formal.py b/scripts/verify_core_formal.py index 2a9ce2e8..fb5fb766 100644 --- a/scripts/verify_core_formal.py +++ b/scripts/verify_core_formal.py @@ -8,6 +8,7 @@ import json import os import signal +import stat from pathlib import Path import subprocess @@ -228,13 +229,143 @@ def formal_source_files() -> list[Path]: return files +def git_checkout_environment() -> dict[str, str]: + environment = {key: value for key, value in os.environ.items() if not key.startswith("GIT_")} + environment.update({ + "GIT_NO_REPLACE_OBJECTS": "1", + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_SYSTEM": os.devnull, + "GIT_TERMINAL_PROMPT": "0", + "GIT_OPTIONAL_LOCKS": "0", + "LC_ALL": "C", + }) + return environment + + +def git_checkout_output(*args: str) -> str: + return subprocess.check_output( + ["git", "-c", "core.fsmonitor=false", *args], + cwd=ROOT, text=True, env=git_checkout_environment(), + ).strip() + + +def core_tree_has_no_physical_extras(paths: set[str]) -> bool: + core_relative = Path("crates/labcolors-core") + core_root = ROOT / core_relative + prefix = core_relative.as_posix() + "/" + expected_files = {path for path in paths if path.startswith(prefix)} + if not expected_files or core_root.is_symlink() or not core_root.is_dir(): + return False + + expected_directories: set[str] = set() + for file in expected_files: + parent = Path(file).parent + while parent != core_relative: + expected_directories.add(parent.as_posix()) + parent = parent.parent + expected = expected_files | expected_directories + + # Cargo сверяет физическое поддерево Core, включая игнорируемые Git записи. + observed: set[str] = set() + pending = [core_root] + while pending: + for entry in pending.pop().iterdir(): + relative = entry.relative_to(ROOT).as_posix() + if relative not in expected: + return False + observed.add(relative) + if relative in expected_directories: + if entry.is_symlink() or not entry.is_dir(): + return False + pending.append(entry) + return observed == expected + + +def tracked_tree_matches_head(commit: str, identities: dict[str, str]) -> bool: + # Сначала удостоверяем достижимые объекты Git: cat-file может отдать повреждённый blob под прежним OID. + environment = git_checkout_environment() + command = ["git", "-c", "core.fsmonitor=false"] + try: + subprocess.check_output( + [*command, "fsck", "--strict", "--no-reflogs", "--no-progress", "--no-dangling", commit], + cwd=ROOT, env=environment, stderr=subprocess.DEVNULL, + ) + tree = subprocess.check_output( + [*command, "ls-tree", "-rz", "--full-tree", commit], cwd=ROOT, env=environment, + ) + entries = [] + paths = set() + for record in tree.split(b"\0"): + if not record: + continue + metadata, path_bytes = record.split(b"\t", 1) + mode, kind, oid = metadata.split(b" ") + if kind != b"blob" or mode not in {b"100644", b"100755", b"120000"}: + return False + relative = os.fsdecode(path_bytes) + entries.append((mode, oid, ROOT / relative)) + paths.add(Path(relative).as_posix()) + formal_paths = {Path(path).as_posix() for path in identities} + if formal_paths - paths: + return False + blobs = subprocess.check_output( + [*command, "cat-file", "--batch"], + input=b"".join(oid + b"\n" for _, oid, _ in entries), cwd=ROOT, env=environment, + ) + cursor = 0 + for mode, oid, path in entries: + if mode == b"120000": + # Git связывает только текст ссылки; формальный вход может читать изменяемую цель. + if (path.is_relative_to(ROOT / "crates/labcolors-core") + or path.relative_to(ROOT).as_posix() in formal_paths): + return False + if not path.is_symlink(): + return False + actual = os.fsencode(os.readlink(path)) + else: + if path.is_symlink() or not path.is_file(): + return False + actual = path.read_bytes() + if os.name != "nt" and bool(path.stat().st_mode & stat.S_IXUSR) != (mode == b"100755"): + return False + end = blobs.find(b"\n", cursor) + if end < 0 or blobs[cursor:end] != oid + b" blob " + str(len(actual)).encode("ascii"): + return False + cursor = end + 1 + if blobs[cursor:cursor + len(actual)] != actual or blobs[cursor + len(actual):cursor + len(actual) + 1] != b"\n": + return False + cursor += len(actual) + 1 + return cursor == len(blobs) and core_tree_has_no_physical_extras(paths) + except (OSError, ValueError, subprocess.CalledProcessError): + return False + + +def checkout_is_clean(identities: dict[str, str] | None = None, commit: str | None = None) -> bool: + # Git status скрывает правки с флагами индекса; для связи квитанции с HEAD этого недостаточно. + version_text = git_checkout_output("version").split() + try: + version = tuple(int(part) for part in version_text[2].split(".")[:3]) + except (IndexError, ValueError) as error: + raise ValueError("cannot verify Git version for formal checkout") from error + if len(version) != 3 or version < (2, 35, 2): + raise ValueError("Git 2.35.2 or newer is required for formal checkout") + records = git_checkout_output("ls-files", "-v", "-z").split("\0") + status = git_checkout_output("status", "--porcelain", "--untracked-files=all") + if status or any(not record.startswith("H ") for record in records if record): + return False + if identities is None: + identities = {str(path.relative_to(ROOT)): digest(path) for path in formal_source_files()} + if commit is None: + commit = git_checkout_output("rev-parse", "HEAD") + return tracked_tree_matches_head(commit, identities) + + def checkout_snapshot() -> tuple[dict[str, str], str, bool]: files = formal_source_files() identities = {str(path.relative_to(ROOT)): digest(path) for path in files} - commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() - clean = not subprocess.check_output( - ["git", "status", "--porcelain", "--untracked-files=all"], cwd=ROOT, text=True - ) + commit = git_checkout_output("rev-parse", "HEAD") + clean = checkout_is_clean(identities, commit) subprocess.run( ["cargo", "metadata", "--locked", "--format-version", "1", "--no-deps"], cwd=ROOT, stdout=subprocess.DEVNULL, check=True, @@ -246,12 +377,10 @@ def assert_snapshot(identities: dict[str, str], commit: str) -> bool: current = {str(path.relative_to(ROOT)): digest(path) for path in formal_source_files()} if identities != current: raise ValueError("source or dependency identity changed during verification") - actual_commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() + actual_commit = git_checkout_output("rev-parse", "HEAD") if commit != actual_commit: raise ValueError("checkout changed during verification") - return not subprocess.check_output( - ["git", "status", "--porcelain", "--untracked-files=all"], cwd=ROOT, text=True - ) + return checkout_is_clean(current, commit) def fragment_base(identities: dict[str, str], commit: str, clean: bool) -> dict: @@ -441,6 +570,13 @@ def assemble_fragments(output: Path, input_dir: Path) -> None: if any(fragment.get(key) != reference.get(key) for key in common): raise ValueError("formal fragments describe different source states") + identities, commit, clean = checkout_snapshot() + if (not clean or reference.get("working_tree_clean") is not True + or reference.get("checkout_commit") != commit + or reference.get("source_commit") != commit + or reference.get("verified_source_sha256") != identities): + raise ValueError("formal fragments do not bind to the current clean source state") + positive_counts = {item.get("shard_count") for item in positives} if len(positive_counts) != 1: raise ValueError("positive shard-count mismatch") @@ -519,6 +655,8 @@ def assemble_fragments(output: Path, input_dir: Path) -> None: "harnesses": sorted(CONTRACTS), "semantic_mutant_rejected": True, } + if not assert_snapshot(identities, commit): + raise ValueError("source state became dirty during formal assembly") (output / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8")