diff --git a/skills/verification-ladder/bin/verify.py b/skills/verification-ladder/bin/verify.py index 2285963..3ec9396 100644 --- a/skills/verification-ladder/bin/verify.py +++ b/skills/verification-ladder/bin/verify.py @@ -191,15 +191,39 @@ def _hash_path(digest: "hashlib._Hash", repo: Path, relative: str) -> None: # content that lives outside the state being identified. digest.update(b"symlink:" + os.readlink(target).encode()) elif target.is_dir(): - for child in sorted(p for p in target.rglob("*") if p.is_file()): - digest.update(child.relative_to(repo).as_posix().encode()) - digest.update(hashlib.sha256(child.read_bytes()).digest()) + # The same rule as the top level, applied inside. `git status` reports an + # untracked nested repository as one directory entry, and hashing it + # through `rglob` and `is_file` followed any symlink within it, so the + # contents of a file outside the repository became part of its state id. + # An explicit walk that never follows links, rather than `rglob`, whose + # symlink handling changed between the interpreters CI runs. + children = [] + for root, dirs, files in os.walk(target, followlinks=False): + children += [Path(root) / name for name in (*dirs, *files)] + for child in sorted(children): + if child.is_symlink(): + digest.update(child.relative_to(repo).as_posix().encode()) + digest.update(b"symlink:" + os.readlink(child).encode()) + elif child.is_file(): + digest.update(child.relative_to(repo).as_posix().encode()) + digest.update(hashlib.sha256(child.read_bytes()).digest()) elif target.is_file(): digest.update(hashlib.sha256(target.read_bytes()).digest()) else: digest.update(b"absent") +def _excluded(path: str, exclude: frozenset[str]) -> bool: + """Whether a `git status` path is evidence rather than repository state. + + An entry ending in "/" excludes everything beneath it; any other entry is + one exact path. The directory form is what makes the evidence directory + excluded whole: a list of the files it held when a command started cannot + cover a file a gate writes there while it runs. + """ + return path in exclude or any(path.startswith(entry) for entry in exclude if entry.endswith("/")) + + def state_id(repo: Path, exclude: frozenset[str] = frozenset()) -> str: """Identify the exact state under verification: HEAD plus every deviation from it. @@ -215,7 +239,7 @@ def state_id(repo: Path, exclude: frozenset[str] = frozenset()) -> str: digest = hashlib.sha256() digest.update(head_sha(repo).encode()) for status, path in _status_entries(git(repo, "status", "--porcelain=v1", "-z", "--untracked-files=all")): - if path in exclude: + if _excluded(path, exclude): continue digest.update(status.encode()) digest.update(path.encode()) @@ -267,12 +291,12 @@ def state_paths(repo: Path, exclude: frozenset[str] = frozenset()) -> set[tuple[ reported, just without the path list. """ entries = _status_entries(git(repo, "status", "--porcelain=v1", "-z", "--untracked-files=all")) - return {(status, path) for status, path in entries if path not in exclude} + return {(status, path) for status, path in entries if not _excluded(path, exclude)} def is_dirty(repo: Path, exclude: frozenset[str] = frozenset()) -> bool: entries = _status_entries(git(repo, "status", "--porcelain=v1", "-z", "--untracked-files=all")) - return any(path not in exclude for _, path in entries) + return any(not _excluded(path, exclude) for _, path in entries) def blocked(message: str) -> "SystemExit": @@ -1946,8 +1970,18 @@ def command_compose_v3(args) -> int: baseline_path = baseline_path_for(paths[0]) if not baseline_path.exists(): - raise blocked("GOVERNING BASELINE MISSING: run `verify.py baseline` before the task and keep " - ".verification/shadow/baseline.v3.json with the evidence") + # Named exactly. The baseline is read beside the FIRST record, so a first + # record kept anywhere else reported the baseline missing while it sat + # beside the others - sending the reader to recreate something present. + elsewhere = sorted({str(baseline_path_for(path)) for path in paths[1:] + if baseline_path_for(path).exists()}) + raise blocked( + f"GOVERNING BASELINE MISSING: no baseline at {baseline_path}, which is where compose " + f"reads it: beside the first record given ({paths[0]}). Run `verify.py baseline` before " + f"the task and keep it with the evidence." + + (f" One exists beside another record given ({', '.join(elsewhere)}); pass a record " + f"from that directory first." if elsewhere else "") + ) baseline = load_baseline(baseline_path) applies, why = baseline_applies(repo, baseline) if not applies: @@ -2270,10 +2304,22 @@ def command_compare(args) -> int: def load_json(path: Path) -> dict: + """A JSON object from disk, or BLOCKED naming the file. + + Every record, payload and baseline this verifier reads is an object, and + every caller reads it with `.get`. Valid JSON of another type - `[1]`, + `"evidence"`, `5` - parsed, reached that `.get`, and ended in an + AttributeError traceback where a refusal belonged. Checked once, here, + rather than at each of the call sites that assumed it. + """ try: - return json.loads(path.read_text()) + value = json.loads(path.read_text()) except (OSError, json.JSONDecodeError) as error: raise blocked(f"{path}: unreadable ({type(error).__name__})") from None + if not isinstance(value, dict): + raise blocked(f"{path}: holds a JSON {type(value).__name__}, not an object; " + f"no verification record or payload has that shape") + return value def load_record(path: Path) -> dict: @@ -2454,10 +2500,11 @@ def relative_inside(repo: Path, paths) -> frozenset[str]: correctness depend on onboarding being remembered. """ named = {p.relative_to(repo).as_posix() for p in paths if p is not None and p.is_relative_to(repo)} - evidence = repo / EVIDENCE_DIR - if evidence.is_dir(): - named |= {f.relative_to(repo).as_posix() for f in evidence.rglob("*") if f.is_file()} - return frozenset(named) + # By prefix, not by listing what is there now. The listing was taken before + # the gates ran, so an artifact a gate wrote into the evidence directory - + # where evidence/3 tells behavioural gates to put them - was not in it, and + # read as the gate moving the repository. Only a `.gitignore` entry hid that. + return frozenset(named | {f"{EVIDENCE_DIR}/"}) def new_record(repo: Path, authority: str, gates: list[dict], exclude: frozenset[str], diff --git a/tests/test_input_boundaries.py b/tests/test_input_boundaries.py new file mode 100644 index 0000000..08b4b8e --- /dev/null +++ b/tests/test_input_boundaries.py @@ -0,0 +1,209 @@ +"""What the verifier accepts from disk and from the worktree, at the edges. + +Four defects, each reproduced against the CLI on main before it was fixed: + +* Valid JSON that is not an object reached `.get` and ended in a traceback. +* The evidence directory was excluded by the files it held when a command + started, so a file a gate created there read as repository drift. +* A symlink inside an untracked nested repository was followed, folding bytes + from outside the repository into its state id. +* A missing baseline was reported without saying where it had been looked for. +""" + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).resolve().parents[1] / "skills" / "verification-ladder" / "bin" / "verify.py" + +POLICY = '''required_gates = ["unit"] +judgment_rungs = ["diff"] + +[gates.local] +unit = "true" +''' + + +def cli(repo, *args): + done = subprocess.run([sys.executable, str(SCRIPT), "--repo", str(repo), *args], + capture_output=True, text=True, check=False) + return done.returncode, done.stdout + done.stderr + + +def make_repo(root, policy=POLICY, gitignore=".verification/\n"): + root.mkdir() + if gitignore is not None: + (root / ".gitignore").write_text(gitignore) + (root / "verification.toml").write_text(policy) + (root / "src.txt").write_text("src\n") + subprocess.run(["git", "init", "-q", "-b", "main", str(root)], check=True) + subprocess.run(["git", "add", "-A"], cwd=root, check=True) + subprocess.run(["git", "-c", "user.email=t@t", "-c", "user.name=t", "-c", "commit.gpgsign=false", + "commit", "-qm", "base"], cwd=root, check=True) + return root + + +@pytest.fixture +def repo(tmp_path): + return make_repo(tmp_path / "repo") + + +# -------------------------------------------------------------------------- +# Valid JSON of the wrong shape is refused, not a traceback. +# -------------------------------------------------------------------------- + +WRONG_SHAPES = [pytest.param("[1]", "list", id="a-list"), + pytest.param('"evidence"', "str", id="a-string"), + pytest.param("5", "int", id="a-number"), + pytest.param("null", "NoneType", id="null")] + + +@pytest.mark.parametrize(("text", "kind"), WRONG_SHAPES) +@pytest.mark.parametrize("command", ["check", "compose"]) +def test_a_record_that_is_not_an_object_is_refused(repo, text, kind, command): + record = repo / ".verification" / "bad.json" + record.parent.mkdir() + record.write_text(text) + code, output = cli(repo, command, str(record)) + assert "Traceback" not in output, output + assert code == 2, output + assert f"holds a JSON {kind}, not an object" in output + + +@pytest.mark.parametrize(("text", "kind"), WRONG_SHAPES) +@pytest.mark.parametrize("which", ["run", "job"]) +def test_a_ci_payload_that_is_not_an_object_is_refused(repo, tmp_path, text, kind, which): + payloads = {"run": tmp_path / "run.json", "job": tmp_path / "job.json"} + for name, path in payloads.items(): + path.write_text(text if name == which else "{}") + code, output = cli(repo, "import-ci", "--run", str(payloads["run"]), "--job", str(payloads["job"])) + assert "Traceback" not in output, output + assert code == 2, output + assert f"{payloads[which].name}: holds a JSON {kind}, not an object" in output + + +# -------------------------------------------------------------------------- +# The evidence directory is excluded whole, including what a gate writes there. +# -------------------------------------------------------------------------- + +WRITES_ARTIFACT = POLICY.replace( + 'unit = "true"', '''unit = "sh -c 'mkdir -p .verification/artifacts && echo x > .verification/artifacts/new.json'"''') +WRITES_SOURCE = POLICY.replace('unit = "true"', '''unit = "sh -c 'echo x > generated.txt'"''') + + +def test_an_artifact_a_gate_writes_into_the_evidence_directory_is_not_drift(tmp_path): + """The exclusion was the files the directory held when `run` started. + + A gate writing an artifact there - where evidence/3 tells behavioural gates to + put them - produced a file that list did not contain, and the run read it as + the gate moving the repository. Only a `.gitignore` entry hid it, and the + exclusion exists precisely so correctness does not depend on one. + """ + root = make_repo(tmp_path / "repo", WRITES_ARTIFACT, gitignore=None) + (root / ".verification").mkdir() + (root / ".verification" / "seed.txt").write_text("present before the run\n") + code, output = cli(root, "run", "--output", str(root / ".verification" / "local.json")) + assert (root / ".verification" / "artifacts" / "new.json").exists(), "premise: the gate wrote it" + assert code == 0, output + assert "drift" not in output, output + assert json.loads((root / ".verification" / "local.json").read_text())["drift"] is False + + +def test_a_gate_writing_outside_the_evidence_directory_is_still_drift(tmp_path): + """The other side: excluding by prefix must not stop drift being seen at all.""" + root = make_repo(tmp_path / "repo", WRITES_SOURCE, gitignore=None) + code, output = cli(root, "run", "--output", str(root / ".verification" / "local.json")) + assert code == 2, output + assert "moved generated.txt" in output + + +# -------------------------------------------------------------------------- +# Symlinks are never followed, however deep in an untracked directory. +# -------------------------------------------------------------------------- + +def _state(root): + import importlib.util + spec = importlib.util.spec_from_file_location("ladder_verify_boundaries", SCRIPT) + verify = importlib.util.module_from_spec(spec) + spec.loader.exec_module(verify) + return verify.state_id(root) + + +def test_a_symlink_inside_an_untracked_nested_repository_is_not_followed(tmp_path): + """Only the top level refused to follow links. + + `git status` reports an untracked nested repository as one directory entry, + and hashing that directory followed the links inside it, so bytes outside the + repository decided its state id. A link's identity is where it points. + """ + root = make_repo(tmp_path / "repo") + outside = tmp_path / "outside" + (outside / "dir").mkdir(parents=True) + (outside / "file.txt").write_text("one\n") + (outside / "dir" / "inner.txt").write_text("one\n") + nested = root / "nested" + nested.mkdir() + subprocess.run(["git", "init", "-q", str(nested)], check=True) + (nested / "real.txt").write_text("real\n") + (nested / "to-file").symlink_to(outside / "file.txt") + (nested / "to-dir").symlink_to(outside / "dir") + assert "?? nested/" in subprocess.run(["git", "status", "--porcelain"], cwd=root, + capture_output=True, text=True, check=True).stdout, \ + "premise: git reports the nested repository as a single directory entry" + + before = _state(root) + (outside / "file.txt").write_text("two\n") + (outside / "dir" / "inner.txt").write_text("two\n") + assert _state(root) == before, "bytes outside the repository moved its state id" + + # Narrower is not blind: what the repository does hold still binds. + (nested / "to-file").unlink() + (nested / "to-file").symlink_to(outside / "dir") + retargeted = _state(root) + assert retargeted != before, "a link that points somewhere else is a different state" + (nested / "real.txt").write_text("changed\n") + assert _state(root) != retargeted, "a real file inside the nested repository still binds" + + +# -------------------------------------------------------------------------- +# A missing baseline is reported where compose actually looked. +# -------------------------------------------------------------------------- + +def _activated(root): + """An evidence/3 repository: adopted in the policy, activated in the checkout.""" + (root / "verification.toml").write_text('evidence = "verification.ladder.evidence/3"\n' + POLICY) + subprocess.run(["git", "-c", "user.email=t@t", "-c", "user.name=t", "-c", "commit.gpgsign=false", + "commit", "-qam", "adopt"], cwd=root, check=True) + (root / ".verification").mkdir(exist_ok=True) + (root / ".verification" / "authority.json").write_text(json.dumps( + {"schema": "verification.ladder.authority/1", "authority": "verification.ladder.evidence/3"})) + assert cli(root, "baseline")[0] == 0 + record = root / ".verification" / "local.json" + code, output = cli(root, "run", "--output", str(record)) + assert code == 0 and json.loads(record.read_text())["schema"] == "verification.ladder.evidence/3", output + return record + + +def test_compose_names_the_baseline_it_looked_for(repo, tmp_path): + """Read beside the FIRST record, and said to be missing without saying where. + + With the first record kept outside the evidence directory, the report sent + the reader to recreate a baseline that sat beside the other records. + """ + record = _activated(repo) + elsewhere = tmp_path / "copies" + elsewhere.mkdir() + (elsewhere / "local.json").write_text(record.read_text()) + + code, output = cli(repo, "compose", str(elsewhere / "local.json"), str(record)) + assert code == 2, output + assert f"no baseline at {elsewhere / 'shadow' / 'baseline.v3.json'}" in output, output + assert f"One exists beside another record given ({repo / '.verification' / 'shadow' / 'baseline.v3.json'})" in output + + code, output = cli(repo, "compose", str(elsewhere / "local.json")) + assert code == 2, output + assert f"no baseline at {elsewhere / 'shadow' / 'baseline.v3.json'}" in output + assert "One exists beside" not in output, "nothing to point at when no given record has one"