Onur Özkan
62b18d5b2e
Replace argument recovery with direct build targets. Run host, target, board, coverage, Miri and selected Kani checks while retaining source identities, logs and artifacts, including failed attempts. Keep the PID proof diagnostic exception explicit and match its exact identity alongside passing contract assertions and covers. Validation: final-source verification matrix, assembled workflow tests and documentation build.
444 lines
17 KiB
Python
444 lines
17 KiB
Python
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
SCRIPT = Path(__file__).with_name("verify.py")
|
|
SPEC = importlib.util.spec_from_file_location("verify", SCRIPT)
|
|
verify = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(verify)
|
|
|
|
|
|
class VerificationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.directory = Path(self.temporary.name)
|
|
|
|
def test_contract_cannot_remove_whole_proof_or_interpreter_gates(self):
|
|
contract, _ = verify.load_contract()
|
|
loader = verify.tomllib.load
|
|
for section, field, value in (
|
|
(None, "properties", []),
|
|
("miri", "packages", []),
|
|
("miri", "packages", ["kaptan-log", "kaptan-log"]),
|
|
("miri", "seeds", []),
|
|
("miri", "seeds", [0, 0]),
|
|
("miri", "seeds", [-1]),
|
|
("miri", "seeds", [True]),
|
|
):
|
|
damaged = copy.deepcopy(contract)
|
|
(damaged[section] if section else damaged)[field] = value
|
|
|
|
def load(source):
|
|
if Path(source.name).name == "verification.toml":
|
|
return damaged
|
|
return loader(source)
|
|
|
|
with (
|
|
self.subTest(field=field, value=value),
|
|
mock.patch.object(verify.tomllib, "load", side_effect=load),
|
|
self.assertRaises(ValueError),
|
|
):
|
|
verify.load_contract()
|
|
|
|
def proof(self):
|
|
prop = {
|
|
"id": "example",
|
|
"harnesses": ["proof"],
|
|
"covers": [1],
|
|
"package": "example",
|
|
"source": "proof.rs",
|
|
}
|
|
report = {
|
|
"metadata": {
|
|
"version": "1.0",
|
|
"kani_version": "0.68.0",
|
|
"target": "x86_64-unknown-linux-gnu",
|
|
},
|
|
"tools": {"rustc": "pinned rustc"},
|
|
"harness_metadata": [
|
|
{
|
|
"pretty_name": "proof",
|
|
"crate_name": "example",
|
|
"source": {"file": "proof.rs"},
|
|
}
|
|
],
|
|
"verification_results": {
|
|
"summary": {
|
|
"total_harnesses": 1,
|
|
"executed": 1,
|
|
"status": "completed",
|
|
"successful": 1,
|
|
"failed": 0,
|
|
},
|
|
"results": [
|
|
{
|
|
"harness_id": "proof",
|
|
"status": "Success",
|
|
"checks": [
|
|
{
|
|
"category": "assertion",
|
|
"status": "Success",
|
|
"description": "assertion failed: invariant",
|
|
"function": "proof",
|
|
},
|
|
{"category": "cover", "status": "Satisfied"},
|
|
],
|
|
}
|
|
],
|
|
},
|
|
}
|
|
return prop, report
|
|
|
|
def validate(self, prop, report, exit_code=0):
|
|
path = self.directory / "proof.json"
|
|
path.write_text(json.dumps(report))
|
|
return verify.validate_kani(
|
|
path, prop, "0.68.0", "x86_64-unknown-linux-gnu", exit_code, "pinned rustc"
|
|
)
|
|
|
|
def test_failed_missing_vacuous_and_unknown_proofs_cannot_pass(self):
|
|
prop, report = self.proof()
|
|
self.assertTrue(self.validate(prop, report)["raw_verifier_passed"])
|
|
for status in (
|
|
"Failure",
|
|
"Undetermined",
|
|
"SolverError",
|
|
"Unreachable",
|
|
"new-status",
|
|
):
|
|
damaged = copy.deepcopy(report)
|
|
damaged["verification_results"]["results"][0]["checks"][0]["status"] = (
|
|
status
|
|
)
|
|
with self.subTest(status=status), self.assertRaises(ValueError):
|
|
self.validate(prop, damaged)
|
|
for status in ("Unreachable", "Unsatisfiable", "Undetermined"):
|
|
damaged = copy.deepcopy(report)
|
|
damaged["verification_results"]["results"][0]["checks"][1]["status"] = (
|
|
status
|
|
)
|
|
with self.subTest(cover=status), self.assertRaises(ValueError):
|
|
self.validate(prop, damaged)
|
|
damaged = copy.deepcopy(report)
|
|
damaged["verification_results"]["results"][0]["checks"].pop()
|
|
with self.assertRaises(ValueError):
|
|
self.validate(prop, damaged)
|
|
damaged["verification_results"]["results"] = []
|
|
with self.assertRaises(ValueError):
|
|
self.validate(prop, damaged)
|
|
with self.assertRaises(ValueError):
|
|
self.validate(prop, report, exit_code=1)
|
|
damaged = copy.deepcopy(report)
|
|
damaged["harness_metadata"][0]["source"]["file"] = "another.rs"
|
|
with self.assertRaises(ValueError):
|
|
self.validate(prop, damaged)
|
|
damaged = copy.deepcopy(report)
|
|
damaged["tools"]["rustc"] = "different compiler"
|
|
with self.assertRaises(ValueError):
|
|
self.validate(prop, damaged)
|
|
|
|
def test_pid_exception_preserves_raw_failure_and_rejects_other_failures(self):
|
|
prop, report = self.proof()
|
|
prop.update(
|
|
id="control.pid-state",
|
|
harnesses=["kani_proofs::pid_update_preserves_bounds_or_state"],
|
|
intermediate_nan_line=196,
|
|
)
|
|
result = report["verification_results"]["results"][0]
|
|
result.update(harness_id=prop["harnesses"][0], status="Failure")
|
|
report["harness_metadata"][0]["pretty_name"] = prop["harnesses"][0]
|
|
report["verification_results"]["summary"].update(successful=0, failed=1)
|
|
diagnostic = {
|
|
"category": "NaN",
|
|
"status": "Failure",
|
|
"description": "NaN on addition",
|
|
"function": "Pid::update_with_valid_time",
|
|
"location": {
|
|
"file": "libs/kaptan-control/src/lib.rs",
|
|
"line": "196",
|
|
"column": "13",
|
|
},
|
|
}
|
|
result["checks"].append(diagnostic)
|
|
accepted = self.validate(prop, report, exit_code=1)
|
|
self.assertFalse(accepted["raw_verifier_passed"])
|
|
self.assertTrue(accepted["contract_checks_passed"])
|
|
for field, value in (
|
|
("category", "overflow"),
|
|
("description", "NaN on division"),
|
|
("function", "different"),
|
|
):
|
|
damaged = copy.deepcopy(report)
|
|
damaged["verification_results"]["results"][0]["checks"][-1][field] = value
|
|
with self.subTest(field=field), self.assertRaises(ValueError):
|
|
self.validate(prop, damaged, exit_code=1)
|
|
result["checks"].append(copy.deepcopy(diagnostic))
|
|
with self.assertRaises(ValueError):
|
|
self.validate(prop, report, exit_code=1)
|
|
|
|
def test_miri_cannot_pass_with_empty_ignored_or_filtered_suites(self):
|
|
path = self.directory / "miri.log"
|
|
template = "test result: ok. {} passed; 0 failed; {} ignored; 0 measured; {} filtered out"
|
|
for passed, ignored, filtered in ((0, 0, 0), (1, 1, 0), (1, 0, 1)):
|
|
path.write_text(template.format(passed, ignored, filtered))
|
|
with self.assertRaises(ValueError):
|
|
verify.validate_miri(path, 1)
|
|
path.write_text(template.format(4, 0, 0))
|
|
self.assertEqual(verify.validate_miri(path, 1), 4)
|
|
with self.assertRaises(ValueError):
|
|
verify.validate_miri(path, 2)
|
|
|
|
def test_snapshot_includes_untracked_bytes_modes_and_deletions(self):
|
|
path = self.directory / "untracked.rs"
|
|
path.write_text("original")
|
|
with mock.patch.object(
|
|
verify.subprocess,
|
|
"check_output",
|
|
return_value=b"untracked.rs\0deleted.rs\0",
|
|
):
|
|
before = verify.snapshot(self.directory)
|
|
path.write_text("edited")
|
|
self.assertNotEqual(before, verify.snapshot(self.directory))
|
|
path.write_text("original")
|
|
path.chmod(0o755)
|
|
self.assertNotEqual(before, verify.snapshot(self.directory))
|
|
path.unlink()
|
|
self.assertTrue(
|
|
all(
|
|
entry["kind"] == "absent"
|
|
for entry in verify.snapshot(self.directory)["entries"]
|
|
)
|
|
)
|
|
|
|
def test_compiler_and_target_environment_overrides_are_rejected(self):
|
|
for name in (
|
|
"RUSTFLAGS",
|
|
"MIRIFLAGS",
|
|
"CARGO_BUILD_RUSTFLAGS",
|
|
"CARGO_BUILD_RUSTC_WRAPPER",
|
|
"CARGO_ENCODED_RUSTDOCFLAGS",
|
|
"CARGO_PROFILE_RELEASE_OVERFLOW_CHECKS",
|
|
"CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS",
|
|
):
|
|
with self.subTest(name=name), self.assertRaises(ValueError):
|
|
verify.reject_overrides({name: "override"})
|
|
verify.reject_overrides({"PATH": "/bin", "CARGO_TARGET_DIR": "/tmp/build"})
|
|
|
|
def test_make_controls_cannot_skip_gate_recipes_or_override_commands(self):
|
|
(self.directory / "Makefile").write_text(
|
|
"COMMAND = touch executed\n.PHONY: gate\ngate:\n\t$(COMMAND)\n"
|
|
)
|
|
injected = self.directory / "injected.mk"
|
|
injected.write_text("override COMMAND = true\n")
|
|
for controls in (
|
|
{"MAKEFLAGS": "n"},
|
|
{"MAKEFLAGS": "t"},
|
|
{"MFLAGS": "-n", "GNUMAKEFLAGS": "-n"},
|
|
{"MAKEFLAGS": "-- COMMAND=true", "MAKEOVERRIDES": "COMMAND=true"},
|
|
{"MAKEFILES": str(injected)},
|
|
):
|
|
with self.subTest(controls=controls):
|
|
marker = self.directory / "executed"
|
|
marker.unlink(missing_ok=True)
|
|
result = verify.run(
|
|
["make", "-C", str(self.directory), "gate"],
|
|
self.directory / "make.log",
|
|
verify.make_environment(os.environ | controls),
|
|
5,
|
|
)
|
|
self.assertTrue(result["passed"])
|
|
self.assertTrue(marker.exists())
|
|
|
|
def test_optimized_final_link_rejects_invalid_outcomes(self):
|
|
fixture = (
|
|
SCRIPT.parents[1] / "libs/kaptan-err-reportable-macros/tests/final_link.py"
|
|
)
|
|
harness = """
|
|
import importlib.util,subprocess,sys
|
|
spec=importlib.util.spec_from_file_location('final_link',sys.argv[1])
|
|
fixture=importlib.util.module_from_spec(spec); spec.loader.exec_module(fixture)
|
|
failure=sys.argv[3]
|
|
def run(arguments, cwd):
|
|
linking=arguments[0]=='c++' or arguments[-1]=='error-id-app'
|
|
duplicate=any('used-duplicate' in arg for arg in arguments)
|
|
code=int(linking and duplicate and failure=='diagnostic')
|
|
return subprocess.CompletedProcess(arguments,code,
|
|
'host: test-host\\n' if arguments[0]=='rustc' else '',
|
|
'unrelated link failure' if code else '')
|
|
fixture.run=run
|
|
sys.argv=[sys.argv[1],'--work-dir',sys.argv[2]]
|
|
fixture.main()
|
|
"""
|
|
for case, diagnostic in (
|
|
("success", "duplicate unexpectedly linked"),
|
|
("diagnostic", "unrelated link failure"),
|
|
):
|
|
with self.subTest(case=case):
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-O",
|
|
"-c",
|
|
harness,
|
|
str(fixture),
|
|
str(self.directory / case),
|
|
case,
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn(diagnostic, result.stderr)
|
|
|
|
def test_source_intake_drift_and_finalization_failures_leave_failed_receipts(self):
|
|
baseline = {"sha256": "baseline", "entries": []}
|
|
changed = {"sha256": "changed", "entries": []}
|
|
for stage, snapshots in (
|
|
("intake", [ValueError("unavailable source")]),
|
|
("archive", [baseline, baseline]),
|
|
("drift", [baseline, baseline, changed, changed]),
|
|
("finalization", [baseline, baseline, baseline, OSError("final snapshot")]),
|
|
):
|
|
output = self.directory / stage
|
|
result = {
|
|
"completed": True,
|
|
"passed": True,
|
|
"exit_code": 0,
|
|
"log": "unused",
|
|
}
|
|
with (
|
|
self.subTest(stage=stage),
|
|
mock.patch.object(
|
|
sys, "argv", [str(SCRIPT), "host", "--output", str(output)]
|
|
),
|
|
mock.patch.object(verify.signal, "signal"),
|
|
mock.patch.object(verify, "snapshot", side_effect=snapshots),
|
|
mock.patch.object(verify, "capture", return_value="identity"),
|
|
mock.patch.object(verify, "run", return_value=result),
|
|
mock.patch.object(
|
|
verify.tarfile,
|
|
"open",
|
|
wraps=verify.tarfile.open,
|
|
side_effect=OSError("archive") if stage == "archive" else None,
|
|
),
|
|
):
|
|
self.assertEqual(verify.main(), 1)
|
|
receipt = json.loads((output / "receipt.json").read_text())
|
|
self.assertFalse(receipt["passed"])
|
|
if stage != "archive":
|
|
self.assertFalse(receipt["source_stable"])
|
|
|
|
def test_launch_failure_and_timeout_are_recorded(self):
|
|
failed = verify.run(
|
|
["/no-such-verification-program"],
|
|
self.directory / "missing.log",
|
|
os.environ,
|
|
1,
|
|
)
|
|
self.assertFalse(failed["completed"])
|
|
timed = verify.run(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"],
|
|
self.directory / "timeout.log",
|
|
os.environ,
|
|
0.05,
|
|
)
|
|
self.assertFalse(timed["passed"])
|
|
self.assertFalse(timed["completed"])
|
|
|
|
def test_repeated_termination_kills_the_owned_process_group(self):
|
|
pid_file = self.directory / "child.pid"
|
|
receipt = self.directory / "result.json"
|
|
child = (
|
|
"import os,time,pathlib; pathlib.Path(%r).write_text(str(os.getpid())); time.sleep(30)"
|
|
% str(pid_file)
|
|
)
|
|
harness = """
|
|
import importlib.util,json,os,pathlib,signal,sys
|
|
spec=importlib.util.spec_from_file_location('verify',sys.argv[1])
|
|
v=importlib.util.module_from_spec(spec); spec.loader.exec_module(v)
|
|
for s in (signal.SIGINT,signal.SIGTERM): signal.signal(s,v.interrupt)
|
|
r=v.run([sys.executable,'-c',sys.argv[2]],pathlib.Path(sys.argv[3]),os.environ,30)
|
|
pathlib.Path(sys.argv[4]).write_text(json.dumps(r))
|
|
"""
|
|
parent = subprocess.Popen(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
harness,
|
|
str(SCRIPT),
|
|
child,
|
|
str(self.directory / "signal.log"),
|
|
str(receipt),
|
|
]
|
|
)
|
|
|
|
def cleanup():
|
|
if pid_file.exists():
|
|
try:
|
|
os.killpg(int(pid_file.read_text()), signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
if parent.poll() is None:
|
|
parent.kill()
|
|
parent.wait()
|
|
|
|
self.addCleanup(cleanup)
|
|
deadline = time.monotonic() + 5
|
|
while not pid_file.exists() and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
self.assertTrue(pid_file.exists())
|
|
child_pid = int(pid_file.read_text())
|
|
os.kill(parent.pid, signal.SIGTERM)
|
|
os.kill(parent.pid, signal.SIGTERM)
|
|
self.assertEqual(parent.wait(timeout=5), 0)
|
|
self.assertFalse(json.loads(receipt.read_text())["completed"])
|
|
with self.assertRaises(ProcessLookupError):
|
|
os.kill(child_pid, 0)
|
|
|
|
def test_cancellation_during_spawn_successful_wait_or_cleanup_cannot_pass(self):
|
|
for stage in ("spawn", "wait", "cleanup"):
|
|
process = mock.Mock(pid=999999)
|
|
|
|
def wait(**_kwargs):
|
|
if stage == "wait":
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
return 0
|
|
|
|
def spawn(*_args, **_kwargs):
|
|
if stage == "spawn":
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
return process
|
|
|
|
def kill(*_args):
|
|
if stage == "cleanup":
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
|
|
process.wait.side_effect = wait
|
|
with (
|
|
self.subTest(stage=stage),
|
|
mock.patch.object(verify.subprocess, "Popen", side_effect=spawn),
|
|
mock.patch.object(verify.os, "killpg", side_effect=kill),
|
|
):
|
|
result = verify.run(
|
|
["fake-child"], self.directory / f"{stage}.log", os.environ, 1
|
|
)
|
|
self.assertFalse(result["passed"])
|
|
self.assertFalse(result["completed"])
|
|
process.wait.assert_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|