diff --git a/.gitignore b/.gitignore index 5a8798e59..77c17e3d2 100644 --- a/.gitignore +++ b/.gitignore @@ -52,6 +52,7 @@ container/state/ *.premigrate .limen.env .limen-private/ +/.limen-workstream/ MASTER-PLAN.md # git worktree dirs (isolation checkouts — never tracked) .worktrees/ diff --git a/cli/tests/test_private_vault.py b/cli/tests/test_private_vault.py new file mode 100644 index 000000000..b03da40c7 --- /dev/null +++ b/cli/tests/test_private_vault.py @@ -0,0 +1,1705 @@ +"""Focused contracts for scripts/private-vault.py.""" + +from __future__ import annotations + +import importlib.util +import json +import shutil +import stat +import subprocess +import tempfile +import threading +from contextlib import contextmanager +from pathlib import Path +from types import SimpleNamespace + +import pytest + +SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "private-vault.py" + + +@pytest.fixture +def vault(monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + spec = importlib.util.spec_from_file_location("private_vault_under_test", SCRIPT) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + root = tmp_path / "repo" + vault_dir = root / "institutio" / "vault" + vault_dir.mkdir(parents=True) + public_key = root / "docs" / "keys" / "synthetic-public-key.asc" + public_key.parent.mkdir(parents=True) + public_key.write_text("synthetic public key\n", encoding="utf-8") + monkeypatch.setattr(module, "ROOT", root) + monkeypatch.setattr(module, "VAULT_DIR", vault_dir) + monkeypatch.setattr(module, "MANIFEST", vault_dir / "manifest.jsonl") + monkeypatch.setattr(module, "PUBKEY", public_key) + monkeypatch.setattr(module, "BOOTSTRAP_ARTIFACT_IDS", frozenset({"artifact-001"})) + module._real_tracked_files = module._tracked_files + monkeypatch.setattr(module, "_tracked_files", lambda: set()) + module._real_index_entry_matches_worktree = module._index_entry_matches_worktree + monkeypatch.setattr(module, "_index_entry_matches_worktree", lambda _relative: True) + module._real_historical_artifacts = module._historical_artifacts + + def synthetic_historical_artifacts(): + if not module.MANIFEST.is_file(): + return {} + return { + row["artifact_id"]: module._custody_metadata(row) + for row in module._read_manifest() + if row.get("schema") == module.SCHEMA and isinstance(row.get("artifact_id"), str) + } + + monkeypatch.setattr(module, "_historical_artifacts", synthetic_historical_artifacts) + module._real_validate_committed_pubkey = module._validate_committed_pubkey + monkeypatch.setattr(module, "_validate_committed_pubkey", lambda: None) + module._real_encrypt_file = module._encrypt_file + module._real_decrypt_file = module._decrypt_file + module._real_ciphertext_recipient_keyids = module._ciphertext_recipient_keyids + module._real_openpgp_packet_tags = module._openpgp_packet_tags + monkeypatch.setattr(module, "_encrypt_file", lambda source, destination: shutil.copyfile(source, destination)) + monkeypatch.setattr(module, "_decrypt_file", lambda source, destination: shutil.copyfile(source, destination)) + + def copy_descriptor(source_fd: int, destination_fd: int) -> None: + module.os.lseek(source_fd, 0, module.os.SEEK_SET) + module.os.ftruncate(destination_fd, 0) + module.os.lseek(destination_fd, 0, module.os.SEEK_SET) + with ( + module.os.fdopen(module.os.dup(source_fd), "rb") as source_handle, + module.os.fdopen(module.os.dup(destination_fd), "wb") as destination_handle, + ): + shutil.copyfileobj(source_handle, destination_handle) + + monkeypatch.setattr(module, "_decrypt_descriptors", copy_descriptor) + monkeypatch.setattr( + module, + "_ciphertext_recipient_keyids", + lambda _ciphertext: {module.ENCRYPTION_SUBKEY_ID}, + ) + monkeypatch.setattr( + module, + "_ciphertext_recipient_keyids_fd", + lambda _ciphertext_fd: {module.ENCRYPTION_SUBKEY_ID}, + ) + monkeypatch.setattr(module, "_openpgp_packet_tags", lambda _ciphertext: [1, 20]) + monkeypatch.setattr(module, "_openpgp_packet_tags_fd", lambda _ciphertext_fd: [1, 20]) + return module + + +def _add(vault, source: Path, artifact_id: str = "artifact-001") -> int: + return vault.cmd_add(SimpleNamespace(file=str(source), artifact_id=artifact_id, apply=True)) + + +def _tracked_paths(vault, artifact_id: str = "artifact-001") -> set[str]: + return { + "docs/keys/synthetic-public-key.asc", + "institutio/vault/manifest.jsonl", + f"institutio/vault/{artifact_id}.gpg", + } + + +def _read_descriptor(vault, file_descriptor: int) -> bytes: + vault.os.lseek(file_descriptor, 0, vault.os.SEEK_SET) + with vault.os.fdopen(vault.os.dup(file_descriptor), "rb") as handle: + return handle.read() + + +def _write_descriptor(vault, file_descriptor: int, payload: bytes) -> None: + vault.os.ftruncate(file_descriptor, 0) + vault.os.lseek(file_descriptor, 0, vault.os.SEEK_SET) + with vault.os.fdopen(vault.os.dup(file_descriptor), "wb") as handle: + handle.write(payload) + + +def test_add_writes_public_safe_manifest_and_rejects_duplicate_id(vault, tmp_path: Path): + source = tmp_path / "private-research.md" + source.write_text("private evidence\n", encoding="utf-8") + + assert _add(vault, source) == 0 + with pytest.raises(vault.VaultError, match="already vaulted"): + _add(vault, source) + + rows = vault._read_manifest() + assert len(rows) == 1 + row = rows[0] + assert set(row) == vault.PUBLIC_FIELDS + assert row["artifact_id"] == "artifact-001" + assert row["ciphertext"] == "artifact-001.gpg" + encoded = vault.MANIFEST.read_text(encoding="utf-8") + assert str(source) not in encoded + assert source.name not in encoded + assert "plaintext_sha256" not in encoded + assert "source_path" not in encoded + + +def test_manifest_rejects_duplicate_json_fields(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + encoded = vault.MANIFEST.read_text(encoding="utf-8") + vault.MANIFEST.write_text( + encoded.replace( + '"artifact_id":"artifact-001"', + '"artifact_id":"artifact-999","artifact_id":"artifact-001"', + ), + encoding="utf-8", + ) + + with pytest.raises(vault.VaultError, match="duplicate field"): + vault._read_manifest() + + +def test_manifest_rejects_noncanonical_vaulted_at(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + row = vault._read_manifest()[0] + row["vaulted_at"] = "not-a-canonical-timestamp" + + assert any("invalid vaulted_at" in error for error in vault._validate_public_row(row, 1)) + with pytest.raises(vault.VaultError, match="immutable custody metadata"): + vault._custody_metadata(row) + + +def test_add_requires_apply_before_any_write(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + + with pytest.raises(vault.VaultError, match="--apply"): + vault.cmd_add(SimpleNamespace(file=str(source), artifact_id="artifact-001", apply=False)) + assert list(vault.VAULT_DIR.iterdir()) == [] + + +def test_add_rejects_unattested_id_before_any_write(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + + with pytest.raises(vault.VaultError, match="recovery admission proof"): + _add(vault, source, "artifact-002") + assert list(vault.VAULT_DIR.iterdir()) == [] + + +@pytest.mark.parametrize("failure", [OSError("synthetic hash failure"), KeyboardInterrupt()]) +def test_add_rolls_back_published_ciphertext_when_metadata_construction_fails( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, failure: BaseException +): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + + def fail_metadata(_path: Path) -> str: + raise failure + + monkeypatch.setattr(vault, "_sha256", fail_metadata) + + with pytest.raises(type(failure)): + _add(vault, source) + + assert not (vault.VAULT_DIR / "artifact-001.gpg").exists() + assert not vault.MANIFEST.exists() + assert [path for path in vault.VAULT_DIR.iterdir() if path.name.startswith(".artifact-001.")] == [] + + +@pytest.mark.parametrize("artifact_id", ["../escape", "nested/path", "/absolute", "Uppercase", "descriptive-name"]) +def test_add_rejects_traversal_and_non_neutral_ids(vault, tmp_path: Path, artifact_id: str): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + with pytest.raises(vault.VaultError): + _add(vault, source, artifact_id) + assert list(vault.VAULT_DIR.iterdir()) == [] + + +@pytest.mark.parametrize("private_root", [".limen-private", ".agent-runtime", ".limen-workstream", None]) +def test_add_rejects_hardlink_to_tracked_plaintext_alias( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, private_root: str | None +): + tracked_alias = vault.ROOT / "tracked-alias" + tracked_alias.write_text("synthetic", encoding="utf-8") + if private_root is None: + source = tmp_path / "outside-private" + else: + source = vault.ROOT / private_root / "private" + source.parent.mkdir(parents=True) + vault.os.link(tracked_alias, source) + monkeypatch.setattr(vault, "_tracked_files", lambda: {"tracked-alias"}) + + with pytest.raises(vault.VaultError, match="file object is already git-tracked"): + _add(vault, source) + + assert not vault.MANIFEST.exists() + + +def test_add_rejects_tracked_hardlink_swapped_in_before_snapshot( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +): + source = tmp_path / "private" + source.write_text("initial", encoding="utf-8") + tracked_alias = vault.ROOT / "tracked-alias" + tracked_alias.write_text("tracked", encoding="utf-8") + monkeypatch.setattr(vault, "_tracked_files", lambda: {"tracked-alias"}) + real_snapshot = vault._snapshot_source + + def swap_then_snapshot(source_path: Path, destination: Path, expected_identity: tuple[int, int]): + source_path.unlink() + vault.os.link(tracked_alias, source_path) + return real_snapshot(source_path, destination, expected_identity) + + monkeypatch.setattr(vault, "_snapshot_source", swap_then_snapshot) + + with pytest.raises(vault.VaultError, match="changed before snapshot"): + _add(vault, source) + + assert not vault.MANIFEST.exists() + + +def test_add_rechecks_tracked_aliases_after_snapshot(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private" + source.write_text("private", encoding="utf-8") + tracked_alias = vault.ROOT / "tracked-alias" + tracked_alias.write_text("initial tracked object", encoding="utf-8") + monkeypatch.setattr(vault, "_tracked_files", lambda: {"tracked-alias"}) + real_snapshot = vault._snapshot_source + + def relink_tracked_path_then_snapshot(source_path: Path, destination: Path, expected_identity: tuple[int, int]): + tracked_alias.unlink() + vault.os.link(source_path, tracked_alias) + return real_snapshot(source_path, destination, expected_identity) + + monkeypatch.setattr(vault, "_snapshot_source", relink_tracked_path_then_snapshot) + + with pytest.raises(vault.VaultError, match="file object is already git-tracked"): + _add(vault, source) + + assert not vault.MANIFEST.exists() + assert not (vault.VAULT_DIR / "artifact-001.gpg").exists() + + +def test_verify_accepts_coherent_public_safe_custody(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault)) + assert vault.cmd_verify(SimpleNamespace()) == 0 + + +@pytest.mark.parametrize( + ("target", "relative", "expected"), + [ + ("manifest", "institutio/vault/manifest.jsonl", "manifest Git index content differs"), + ("public_key", "docs/keys/synthetic-public-key.asc", "public-key Git index content differs"), + ("ciphertext", "institutio/vault/artifact-001.gpg", "ciphertext Git index content differs"), + ], +) +def test_verify_rejects_staged_custody_file_that_differs_from_validated_worktree( + vault, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + target: str, + relative: str, + expected: str, +): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + subprocess.run( + ["git", "-C", str(vault.ROOT), "init", "-b", "main"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + subprocess.run( + ["git", "-C", str(vault.ROOT), "add", "institutio/vault", "docs/keys/synthetic-public-key.asc"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + subprocess.run( + [ + "git", + "-C", + str(vault.ROOT), + "-c", + "user.email=vault-test@example.invalid", + "-c", + "user.name=Vault Test", + "-c", + "commit.gpgsign=false", + "commit", + "-m", + "commit coherent custody", + ], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + target_path = { + "manifest": vault.MANIFEST, + "public_key": vault.PUBKEY, + "ciphertext": vault.VAULT_DIR / "artifact-001.gpg", + }[target] + validated_content = target_path.read_bytes() + target_path.write_bytes(validated_content + b"synthetic staged content\n") + subprocess.run( + ["git", "-C", str(vault.ROOT), "add", relative], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + target_path.write_bytes(validated_content) + monkeypatch.setattr(vault, "_tracked_files", vault._real_tracked_files) + monkeypatch.setattr(vault, "_index_entry_matches_worktree", vault._real_index_entry_matches_worktree) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + assert expected in capsys.readouterr().out + + +def test_verify_rejects_copied_ciphertext_under_distinct_ids( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + monkeypatch.setattr(vault, "BOOTSTRAP_ARTIFACT_IDS", frozenset({"artifact-001", "artifact-002"})) + _add(vault, source) + first_row = vault._read_manifest()[0] + second_cipher = vault.VAULT_DIR / "artifact-002.gpg" + shutil.copyfile(vault.VAULT_DIR / "artifact-001.gpg", second_cipher) + second_row = dict(first_row, artifact_id="artifact-002", ciphertext="artifact-002.gpg") + vault._write_manifest([first_row, second_row]) + monkeypatch.setattr( + vault, + "_tracked_files", + lambda: _tracked_paths(vault) | {"institutio/vault/artifact-002.gpg"}, + ) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + assert "duplicate ciphertext digest across distinct artifact ids" in capsys.readouterr().out + + +def test_verify_allows_independently_encrypted_equal_plaintext(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + first = tmp_path / "first.md" + second = tmp_path / "second.md" + first.write_text("equal plaintext", encoding="utf-8") + second.write_text("equal plaintext", encoding="utf-8") + monkeypatch.setattr(vault, "BOOTSTRAP_ARTIFACT_IDS", frozenset({"artifact-001", "artifact-002"})) + + def distinct_encryption(envelope: Path, destination: Path) -> None: + destination.write_bytes(destination.name.encode("utf-8") + envelope.read_bytes()) + + monkeypatch.setattr(vault, "_encrypt_file", distinct_encryption) + _add(vault, first, "artifact-001") + _add(vault, second, "artifact-002") + monkeypatch.setattr( + vault, + "_tracked_files", + lambda: _tracked_paths(vault) | {"institutio/vault/artifact-002.gpg"}, + ) + + assert vault.cmd_verify(SimpleNamespace()) == 0 + + +def test_verify_rejects_missing_required_manifest(vault): + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_deletion_from_committed_custody(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault)) + monkeypatch.setattr( + vault, + "_historical_artifacts", + lambda: { + "artifact-002": ( + "artifact-002.gpg", + "0" * 64, + 1, + vault.FINGERPRINT, + "2026-08-09T00:00:00+00:00", + ) + }, + ) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_historical_baseline_is_monotonic_across_commits(vault, monkeypatch: pytest.MonkeyPatch): + def run_git(*args: str) -> None: + subprocess.run( + ["git", "-C", str(vault.ROOT), *args], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + + run_git("init", "-b", "main") + run_git("config", "user.email", "vault-test@example.invalid") + run_git("config", "user.name", "Vault Test") + + def history_row(artifact_id: str, digest: str) -> dict: + return { + "schema": vault.SCHEMA, + "artifact_id": artifact_id, + "ciphertext": f"{artifact_id}.gpg", + "ciphertext_sha256": digest, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + } + + first_rows = [ + history_row("artifact-001", "1" * 64), + history_row("artifact-002", "2" * 64), + {"schema": vault.SCHEMA, "artifact_id": "descriptive-name"}, + ] + vault.MANIFEST.write_text("".join(json.dumps(row) + "\n" for row in first_rows), encoding="utf-8") + run_git("add", "institutio/vault/manifest.jsonl") + run_git("-c", "commit.gpgsign=false", "commit", "-m", "admit custody") + vault.MANIFEST.write_text(json.dumps(history_row("artifact-001", "1" * 64)) + "\n", encoding="utf-8") + run_git("add", "institutio/vault/manifest.jsonl") + run_git("-c", "commit.gpgsign=false", "commit", "-m", "attempt deletion") + safe_root = subprocess.run( + ["git", "-C", str(vault.ROOT), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + timeout=60, + ).stdout.strip() + monkeypatch.setattr(vault, "PUBLIC_SAFE_HISTORY_ROOT", safe_root) + + assert set(vault._real_historical_artifacts()) == {"artifact-001", "artifact-002"} + + +def test_historical_baseline_survives_fixed_path_replacement(vault): + def run_git(*args: str) -> None: + subprocess.run( + ["git", "-C", str(vault.ROOT), *args], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + + def history_row(artifact_id: str, digest: str) -> dict: + return { + "schema": vault.SCHEMA, + "artifact_id": artifact_id, + "ciphertext": f"{artifact_id}.gpg", + "ciphertext_sha256": digest, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + } + + run_git("init", "-b", "main") + run_git("config", "user.email", "vault-test@example.invalid") + run_git("config", "user.name", "Vault Test") + vault.MANIFEST.write_text(json.dumps(history_row("artifact-001", "1" * 64)) + "\n", encoding="utf-8") + run_git("add", "institutio/vault/manifest.jsonl") + run_git("-c", "commit.gpgsign=false", "commit", "-m", "admit original custody") + run_git("mv", "institutio/vault/manifest.jsonl", "institutio/vault/original.jsonl") + run_git("-c", "commit.gpgsign=false", "commit", "-m", "rename fixed manifest away") + replacement = vault.ROOT / "replacement" / "manifest.jsonl" + replacement.parent.mkdir() + replacement.write_text(json.dumps(history_row("artifact-002", "2" * 64)) + "\n", encoding="utf-8") + run_git("add", "replacement/manifest.jsonl") + run_git("-c", "commit.gpgsign=false", "commit", "-m", "add replacement elsewhere") + run_git("mv", "replacement/manifest.jsonl", "institutio/vault/manifest.jsonl") + run_git("-c", "commit.gpgsign=false", "commit", "-m", "restore fixed manifest path") + + assert set(vault._real_historical_artifacts()) == {"artifact-001", "artifact-002"} + + +def test_historical_manifest_ignores_git_replace_objects(vault, monkeypatch: pytest.MonkeyPatch): + def run_git(*args: str, input_text: str | None = None) -> str: + return subprocess.run( + ["git", "-C", str(vault.ROOT), *args], + input=input_text, + check=True, + capture_output=True, + text=True, + timeout=60, + ).stdout.strip() + + run_git("init", "-b", "main") + run_git("config", "user.email", "vault-test@example.invalid") + run_git("config", "user.name", "Vault Test") + row = { + "schema": vault.SCHEMA, + "artifact_id": "artifact-001", + "ciphertext": "artifact-001.gpg", + "ciphertext_sha256": "1" * 64, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + } + vault.MANIFEST.write_text(json.dumps(row) + "\n", encoding="utf-8") + run_git("add", "institutio/vault/manifest.jsonl") + run_git("-c", "commit.gpgsign=false", "commit", "-m", "admit custody") + admitted = run_git("rev-parse", "HEAD") + empty_tree = run_git("mktree", input_text="") + replacement = run_git("-c", "commit.gpgsign=false", "commit-tree", empty_tree, input_text="synthetic\n") + run_git("replace", admitted, replacement) + monkeypatch.setattr(vault, "PUBLIC_SAFE_HISTORY_ROOT", admitted) + + assert set(vault._real_historical_artifacts()) == {"artifact-001"} + + +def test_historical_manifest_rejects_legacy_grafts(vault): + subprocess.run( + ["git", "-C", str(vault.ROOT), "init", "-b", "main"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + grafts = vault.ROOT / ".git" / "info" / "grafts" + grafts.write_text("synthetic\n", encoding="utf-8") + + with pytest.raises(vault.VaultError, match="rewritten Git custody history"): + vault._real_historical_artifacts() + + +def test_historical_manifest_rejects_duplicate_json_fields(vault): + subprocess.run( + ["git", "-C", str(vault.ROOT), "init", "-b", "main"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + row = { + "schema": vault.SCHEMA, + "artifact_id": "artifact-001", + "ciphertext": "artifact-001.gpg", + "ciphertext_sha256": "1" * 64, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + } + encoded = json.dumps(row, separators=(",", ":")).replace( + '"artifact_id":"artifact-001"', + '"artifact_id":"artifact-999","artifact_id":"artifact-001"', + ) + vault.MANIFEST.write_text(encoded + "\n", encoding="utf-8") + subprocess.run( + ["git", "-C", str(vault.ROOT), "add", "institutio/vault/manifest.jsonl"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + subprocess.run( + [ + "git", + "-C", + str(vault.ROOT), + "-c", + "user.email=vault-test@example.invalid", + "-c", + "user.name=Vault Test", + "-c", + "commit.gpgsign=false", + "commit", + "-m", + "commit duplicate manifest field", + ], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + + with pytest.raises(vault.VaultError, match="duplicate field"): + vault._real_historical_artifacts() + + +def test_historical_manifest_fails_when_present_snapshot_is_unreadable(vault, monkeypatch: pytest.MonkeyPatch): + manifest_relative = "institutio/vault/manifest.jsonl" + + def missing_snapshot(args, *, env=None): + del env + if "--is-shallow-repository" in args: + return subprocess.CompletedProcess(args, 0, stdout="false\n", stderr="") + if "rev-list" in args: + return subprocess.CompletedProcess(args, 0, stdout="a" * 40 + "\n", stderr="") + if "cat-file" in args: + return subprocess.CompletedProcess(args, 1, stdout="", stderr="") + if "ls-tree" in args: + return subprocess.CompletedProcess(args, 0, stdout=manifest_relative + "\0", stderr="") + if "show" in args: + return subprocess.CompletedProcess(args, 128, stdout="", stderr="missing blob") + raise AssertionError(args) + + monkeypatch.setattr(vault, "_run_command", missing_snapshot) + + with pytest.raises(vault.VaultError, match="cannot read a committed manifest history snapshot"): + vault._real_historical_artifacts() + + +def test_historical_manifest_rejects_shallow_repository(vault, monkeypatch: pytest.MonkeyPatch): + def shallow_repository(args, *, env=None): + del env + assert "--is-shallow-repository" in args + return subprocess.CompletedProcess(args, 0, stdout="true\n", stderr="") + + monkeypatch.setattr(vault, "_run_command", shallow_repository) + + with pytest.raises(vault.VaultError, match="non-shallow repository"): + vault._real_historical_artifacts() + + +def test_historical_manifest_rejects_unprovable_repository_depth(vault, monkeypatch: pytest.MonkeyPatch): + def unavailable_depth(args, *, env=None): + del env + assert "--is-shallow-repository" in args + return subprocess.CompletedProcess(args, 128, stdout="", stderr="unavailable") + + monkeypatch.setattr(vault, "_run_command", unavailable_depth) + + with pytest.raises(vault.VaultError, match="cannot prove complete"): + vault._real_historical_artifacts() + + +def test_historical_manifest_rejects_non_public_safe_rows(vault): + subprocess.run( + ["git", "-C", str(vault.ROOT), "init", "-b", "main"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + row = { + "schema": vault.SCHEMA, + "artifact_id": "artifact-001", + "ciphertext": "artifact-001.gpg", + "ciphertext_sha256": "1" * 64, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + "unsupported": "synthetic", + } + vault.MANIFEST.write_text(json.dumps(row) + "\n", encoding="utf-8") + subprocess.run( + ["git", "-C", str(vault.ROOT), "add", "institutio/vault/manifest.jsonl"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + subprocess.run( + [ + "git", + "-C", + str(vault.ROOT), + "-c", + "user.email=vault-test@example.invalid", + "-c", + "user.name=Vault Test", + "-c", + "commit.gpgsign=false", + "commit", + "-m", + "commit unsupported public field", + ], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + + with pytest.raises(vault.VaultError, match="non-public-safe"): + vault._real_historical_artifacts() + + +@pytest.mark.parametrize("schema", [None, "unsupported-schema"]) +def test_historical_manifest_rejects_non_v2_post_boundary_rows( + vault, monkeypatch: pytest.MonkeyPatch, schema: str | None +): + manifest_relative = "institutio/vault/manifest.jsonl" + row = {"artifact_id": "artifact-001"} + if schema is not None: + row["schema"] = schema + + def non_v2_snapshot(args, *, env=None): + del env + if "--is-shallow-repository" in args: + return subprocess.CompletedProcess(args, 0, stdout="false\n", stderr="") + if "rev-list" in args: + return subprocess.CompletedProcess(args, 0, stdout="a" * 40 + "\n", stderr="") + if "cat-file" in args: + return subprocess.CompletedProcess(args, 1, stdout="", stderr="") + if "ls-tree" in args: + return subprocess.CompletedProcess(args, 0, stdout=manifest_relative + "\0", stderr="") + if "show" in args: + return subprocess.CompletedProcess(args, 0, stdout=json.dumps(row) + "\n", stderr="") + raise AssertionError(args) + + monkeypatch.setattr(vault, "_run_command", non_v2_snapshot) + + with pytest.raises(vault.VaultError, match="non-public-safe"): + vault._real_historical_artifacts() + + +def test_historical_manifest_validates_side_branches_incomparable_with_boundary(vault, monkeypatch: pytest.MonkeyPatch): + manifest_relative = "institutio/vault/manifest.jsonl" + safe_root = "a" * 40 + side_revision = "b" * 40 + monkeypatch.setattr(vault, "PUBLIC_SAFE_HISTORY_ROOT", safe_root) + row = { + "schema": vault.SCHEMA, + "artifact_id": "artifact-001", + "ciphertext": "artifact-001.gpg", + "ciphertext_sha256": "1" * 64, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + "private_field": "synthetic", + } + + def side_branch_snapshot(args, *, env=None): + del env + if "--is-shallow-repository" in args: + return subprocess.CompletedProcess(args, 0, stdout="false\n", stderr="") + if "rev-list" in args: + return subprocess.CompletedProcess(args, 0, stdout=side_revision + "\n", stderr="") + if "cat-file" in args: + return subprocess.CompletedProcess(args, 0, stdout="", stderr="") + if "merge-base" in args and args[-2:] == [safe_root, "HEAD"]: + return subprocess.CompletedProcess(args, 0, stdout="", stderr="") + if "merge-base" in args and args[-2:] == [side_revision, safe_root]: + return subprocess.CompletedProcess(args, 1, stdout="", stderr="") + if "ls-tree" in args: + return subprocess.CompletedProcess(args, 0, stdout=manifest_relative + "\0", stderr="") + if "show" in args: + return subprocess.CompletedProcess(args, 0, stdout=json.dumps(row) + "\n", stderr="") + raise AssertionError(args) + + monkeypatch.setattr(vault, "_run_command", side_branch_snapshot) + + with pytest.raises(vault.VaultError, match="non-public-safe"): + vault._real_historical_artifacts() + + +def test_tracked_files_preserve_non_ascii_private_paths(vault): + subprocess.run( + ["git", "-C", str(vault.ROOT), "init", "-b", "main"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + private_path = vault.ROOT / ".limen-private" / "résumé.md" + private_path.parent.mkdir() + private_path.write_text("synthetic", encoding="utf-8") + subprocess.run( + ["git", "-C", str(vault.ROOT), "add", "-f", ".limen-private/résumé.md"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + + assert ".limen-private/résumé.md" in vault._real_tracked_files() + + +def test_accepted_repository_private_namespaces_are_gitignored(vault, tmp_path: Path): + repository = tmp_path / "ignore-repository" + repository.mkdir() + shutil.copyfile(SCRIPT.parents[1] / ".gitignore", repository / ".gitignore") + subprocess.run( + ["git", "-C", str(repository), "init", "-b", "main"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + subprocess.run( + ["git", "-C", str(repository), "config", "core.excludesFile", "/dev/null"], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + for prefix in vault.PRIVATE_TRACKING_PREFIXES: + probe = f"{prefix}vault-ignore-probe" + ignored = subprocess.run( + ["git", "-C", str(repository), "check-ignore", "--no-index", "--quiet", probe], + check=False, + timeout=60, + ) + assert ignored.returncode == 0, probe + + +def test_verify_rejects_changed_historical_ciphertext_metadata(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault)) + current = vault._custody_metadata(vault._read_manifest()[0]) + historical = (current[0], "0" * 64, current[2], current[3], current[4]) + monkeypatch.setattr(vault, "_historical_artifacts", lambda: {"artifact-001": historical}) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_non_string_artifact_id(vault): + row = { + "schema": vault.SCHEMA, + "artifact_id": 123, + "ciphertext": "123.gpg", + "ciphertext_sha256": "0" * 64, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + } + + errors = vault._validate_public_row(row, 1) + + assert any("artifact id must be a string" in error for error in errors) + + +def test_verify_rejects_symlinked_manifest(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + target = tmp_path / "manifest-target.jsonl" + target.write_text("{}\n", encoding="utf-8") + vault.MANIFEST.symlink_to(target) + monkeypatch.setattr(vault, "_tracked_files", lambda: {"institutio/vault/manifest.jsonl"}) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_invalid_committed_public_key(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault)) + + def reject_key() -> None: + raise vault.VaultError("pinned identity mismatch") + + monkeypatch.setattr(vault, "_validate_committed_pubkey", reject_key) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_import_rejects_symlinked_public_key(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + target = tmp_path / "public-key.asc" + target.write_text("synthetic public key", encoding="utf-8") + symlink = tmp_path / "committed-key.asc" + symlink.symlink_to(target) + monkeypatch.setattr(vault, "PUBKEY", symlink) + + with pytest.raises(vault.VaultError, match="non-symlink"): + vault._import_pubkey(str(tmp_path / "gnupg")) + + +def test_import_rejects_secret_key_material(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + public_key = tmp_path / "committed-key.asc" + canonical_armor = "synthetic canonical public-key armor\n" + public_key.write_text(canonical_armor, encoding="utf-8") + monkeypatch.setattr(vault, "PUBKEY", public_key) + + def expose_secret(args, *, env=None): + del env + if "--import" in args: + return subprocess.CompletedProcess(args, 0, stdout="", stderr="") + if "--export" in args: + return subprocess.CompletedProcess(args, 0, stdout=canonical_armor, stderr="") + return subprocess.CompletedProcess(args, 0, stdout="sec:u:255:22:SECRET\n", stderr="") + + monkeypatch.setattr(vault, "_run_command", expose_secret) + + with pytest.raises(vault.VaultError, match="secret-key material"): + vault._import_pubkey(str(tmp_path / "gnupg")) + + +def test_import_rejects_non_key_bytes(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + canonical_armor = "synthetic canonical public-key armor\n" + public_key = tmp_path / "committed-key.asc" + public_key.write_text(canonical_armor + "synthetic trailing bytes\n", encoding="utf-8") + monkeypatch.setattr(vault, "PUBKEY", public_key) + + def canonical_export(args, *, env=None): + del env + if "--import" in args: + return subprocess.CompletedProcess(args, 0, stdout="", stderr="") + if "--export" in args: + return subprocess.CompletedProcess(args, 0, stdout=canonical_armor, stderr="") + raise AssertionError(args) + + monkeypatch.setattr(vault, "_run_command", canonical_export) + + with pytest.raises(vault.VaultError, match="only canonical public-key armor"): + vault._import_pubkey(str(tmp_path / "gnupg")) + + +def test_committed_public_key_validation_rejects_unusable_subkey(vault, monkeypatch: pytest.MonkeyPatch): + primary = [""] * 12 + primary[0] = "pub" + primary[4] = vault.FINGERPRINT[-16:] + primary[11] = "scESC" + primary_fingerprint = [""] * 10 + primary_fingerprint[0] = "fpr" + primary_fingerprint[9] = vault.FINGERPRINT + subkey = [""] * 12 + subkey[0] = "sub" + subkey[4] = vault.ENCRYPTION_SUBKEY_ID + subkey[11] = "e" + subkey_fingerprint = [""] * 10 + subkey_fingerprint[0] = "fpr" + subkey_fingerprint[9] = "0" * 24 + vault.ENCRYPTION_SUBKEY_ID + listing = "\n".join(":".join(fields) for fields in (primary, primary_fingerprint, subkey, subkey_fingerprint)) + + monkeypatch.setattr(vault, "_import_pubkey", lambda _gnupghome: None) + + def reject_probe(args, *, env=None): + del env + if "--list-keys" in args: + return subprocess.CompletedProcess(args, 0, stdout=listing, stderr="") + return subprocess.CompletedProcess(args, 2, stdout="", stderr="Unusable public key") + + monkeypatch.setattr(vault, "_run_command", reject_probe) + + with pytest.raises(vault.VaultError, match="unusable"): + vault._real_validate_committed_pubkey() + + +def test_verify_rejects_tracked_private_namespace(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = vault.ROOT / ".limen-private" / "private.md" + source.parent.mkdir(parents=True) + source.write_text("secret", encoding="utf-8") + _add(vault, source) + tracked = _tracked_paths(vault) | {".limen-private/private.md"} + monkeypatch.setattr(vault, "_tracked_files", lambda: tracked) + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +@pytest.mark.parametrize("private_root", [".limen-private", ".agent-runtime", ".limen-workstream"]) +def test_verify_rejects_tracked_private_namespace_root( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, private_root: str +): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault) | {private_root}) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_non_bootstrap_artifact_without_recovery_proof( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + with monkeypatch.context() as allow_synthetic_admission: + allow_synthetic_admission.setattr(vault, "BOOTSTRAP_ARTIFACT_IDS", frozenset({"artifact-001", "artifact-002"})) + _add(vault, source, "artifact-002") + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault, "artifact-002")) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_nested_vault_content(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + nested = vault.VAULT_DIR / "import" + nested.mkdir() + (nested / "unmanifested.gpg").write_bytes(b"ciphertext") + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault)) + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_wrong_recipient(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault)) + monkeypatch.setattr(vault, "_ciphertext_recipient_keyids", lambda _ciphertext: {"0" * 16}) + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_openpgp_framing_rejects_appended_bytes(vault, tmp_path: Path): + ciphertext = tmp_path / "synthetic.gpg" + ciphertext.write_bytes(bytes([0xC1, 0x01]) + b"x" + bytes([0xD4, 0x02]) + b"yz") + + assert vault._real_openpgp_packet_tags(ciphertext) == [1, 20] + + with ciphertext.open("ab") as handle: + handle.write(b"synthetic trailing bytes") + + with pytest.raises(vault.VaultError, match="outside OpenPGP packet framing"): + vault._real_openpgp_packet_tags(ciphertext) + + +def test_verify_rejects_symlinked_ciphertext(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + ciphertext = vault.VAULT_DIR / "artifact-001.gpg" + outside = tmp_path / "outside.gpg" + outside.write_bytes(ciphertext.read_bytes()) + ciphertext.unlink() + ciphertext.symlink_to(outside) + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault)) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_dangling_vault_symlink(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + (vault.VAULT_DIR / "alias").symlink_to(tmp_path / "missing") + monkeypatch.setattr(vault, "_tracked_files", lambda: _tracked_paths(vault) | {"institutio/vault/alias"}) + + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_verify_rejects_manifest_ciphertext_traversal(vault, monkeypatch: pytest.MonkeyPatch): + row = { + "schema": vault.SCHEMA, + "artifact_id": "artifact-001", + "ciphertext": "../outside.gpg", + "ciphertext_sha256": "0" * 64, + "ciphertext_bytes": 1, + "recipient_fpr": vault.FINGERPRINT, + "vaulted_at": "2026-08-09T00:00:00+00:00", + } + vault.MANIFEST.write_text(json.dumps(row) + "\n", encoding="utf-8") + monkeypatch.setattr(vault, "_tracked_files", lambda: {"institutio/vault/manifest.jsonl"}) + assert vault.cmd_verify(SimpleNamespace()) == 1 + + +def test_failed_restore_removes_all_plaintext_temporaries(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + + def corrupt_decrypt(ciphertext_fd: int, envelope_fd: int) -> None: + _write_descriptor(vault, envelope_fd, _read_descriptor(vault, ciphertext_fd) + b"tamper") + + monkeypatch.setattr(vault, "_decrypt_descriptors", corrupt_decrypt) + with pytest.raises(vault.VaultError, match="integrity mismatch"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + assert not destination.exists() + + +def test_failed_restore_cleans_up_after_decrypt_failure(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + + def reject_decrypt(_ciphertext_fd: int, _envelope_fd: int) -> None: + raise vault.VaultError("decryption failed: no secret key") + + monkeypatch.setattr(vault, "_decrypt_descriptors", reject_decrypt) + with pytest.raises(vault.VaultError, match="no secret key"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + assert not destination.exists() + + +def test_restore_rejects_unpinned_ciphertext_before_writing(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + (vault.VAULT_DIR / "artifact-001.gpg").write_bytes(b"substituted") + destination = tmp_path / "restore" + + with pytest.raises(vault.VaultError, match="ciphertext"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + assert not destination.exists() + + +def test_restore_rejects_ciphertext_replaced_before_snapshot(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("trusted", encoding="utf-8") + _add(vault, source) + ciphertext = vault.VAULT_DIR / "artifact-001.gpg" + destination = tmp_path / "restore" + real_snapshot = vault._snapshot_ciphertext + + def replace_before_snapshot(source_path: Path, snapshot_fd: int) -> None: + source_path.write_bytes(b"replacement") + real_snapshot(source_path, snapshot_fd) + + monkeypatch.setattr(vault, "_snapshot_ciphertext", replace_before_snapshot) + + with pytest.raises(vault.VaultError, match="ciphertext"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert ciphertext.read_bytes() == b"replacement" + assert not destination.exists() + + +def test_restore_decrypts_the_validated_ciphertext_snapshot(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("trusted", encoding="utf-8") + _add(vault, source) + ciphertext = vault.VAULT_DIR / "artifact-001.gpg" + expected_ciphertext = ciphertext.read_bytes() + destination = tmp_path / "restore" + + def replace_original_then_decrypt(snapshot_fd: int, envelope_fd: int) -> None: + assert _read_descriptor(vault, snapshot_fd) == expected_ciphertext + ciphertext.write_bytes(b"replacement") + _write_descriptor(vault, envelope_fd, _read_descriptor(vault, snapshot_fd)) + + monkeypatch.setattr(vault, "_decrypt_descriptors", replace_original_then_decrypt) + + assert ( + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + == 0 + ) + assert (destination / "artifact-001--private.md").read_text(encoding="utf-8") == "trusted" + assert ciphertext.read_bytes() == b"replacement" + + +def test_restore_rejects_substitution_against_committed_custody(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + original_row = vault._read_manifest()[0] + expected_metadata = vault._custody_metadata(original_row) + ciphertext = vault.VAULT_DIR / "artifact-001.gpg" + ciphertext.write_bytes(b"synthetic replacement ciphertext") + substituted_row = dict(original_row) + substituted_row["ciphertext_sha256"] = vault._sha256(ciphertext) + substituted_row["ciphertext_bytes"] = ciphertext.stat().st_size + vault._write_manifest([substituted_row]) + monkeypatch.setattr(vault, "_historical_artifacts", lambda: {"artifact-001": expected_metadata}) + destination = tmp_path / "restore" + + with pytest.raises(vault.VaultError, match="committed custody history"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert not destination.exists() + + +def test_restore_requires_apply_before_plaintext_write(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + + with pytest.raises(vault.VaultError, match="--apply"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=False)) + assert not destination.exists() + + +def test_restore_rejects_unprotected_repository_destination(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + destination = vault.ROOT / "restored" + + with pytest.raises(vault.VaultError, match="private namespace"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert not destination.exists() + + +@pytest.mark.parametrize("private_root", [".limen-private", ".agent-runtime", ".limen-workstream"]) +def test_restore_allows_each_repository_private_destination(vault, tmp_path: Path, private_root: str): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + destination = vault.ROOT / private_root / "restore" + + assert ( + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + == 0 + ) + assert (destination / "artifact-001--private.md").read_text(encoding="utf-8") == "synthetic" + + +def test_restore_rejects_private_destination_redirected_into_public_repository_path(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + private_root = vault.ROOT / ".limen-private" + private_root.mkdir() + public_root = vault.ROOT / "public" + public_root.mkdir() + (private_root / "redirect").symlink_to(public_root, target_is_directory=True) + destination = private_root / "redirect" / "nested" + + with pytest.raises(vault.VaultError, match="remain in a private namespace"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert not (public_root / "nested").exists() + assert list(public_root.iterdir()) == [] + + +def test_restore_rejects_external_alias_into_public_repository_path(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + public_root = vault.ROOT / "public" + public_root.mkdir() + external_alias = tmp_path / "external-alias" + external_alias.symlink_to(public_root, target_is_directory=True) + destination = external_alias / "nested" + + with pytest.raises(vault.VaultError, match="remain in a private namespace"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert not (public_root / "nested").exists() + assert list(public_root.iterdir()) == [] + + +def test_successful_restore_is_verified_atomic_and_owner_only(vault, tmp_path: Path): + source = tmp_path / "private notes.md" + payload = b"valuable private research\n" + source.write_bytes(payload) + _add(vault, source) + destination = tmp_path / "restore" + + assert ( + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + == 0 + ) + restored = destination / "artifact-001--private notes.md" + assert restored.read_bytes() == payload + assert stat.S_IMODE(destination.stat().st_mode) == 0o700 + assert stat.S_IMODE(restored.stat().st_mode) == 0o600 + assert [path for path in destination.iterdir() if path.name.startswith(".artifact-001.")] == [] + + +def test_restore_preserves_existing_directory_permissions(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + destination.mkdir(mode=0o750) + destination.chmod(0o750) + + with pytest.raises(vault.VaultError, match="owner-only"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert stat.S_IMODE(destination.stat().st_mode) == 0o750 + + +def test_restore_refuses_dangling_target(vault, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("synthetic", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + destination.mkdir(mode=0o700) + final_path = destination / "artifact-001--private.md" + final_path.symlink_to(destination / "missing") + + with pytest.raises(vault.VaultError, match="already exists"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert final_path.is_symlink() + + +def test_restore_rejects_destination_swap_before_temp_creation(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("trusted", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + destination.mkdir(mode=0o700) + original_destination = tmp_path / "restore-original" + redirect = tmp_path / "redirect" + redirect.mkdir(mode=0o700) + real_temporary_file_at = vault._temporary_file_at + swapped = False + + def swap_then_create(directory_fd: int, artifact_id: str, suffix: str): + nonlocal swapped + if not swapped: + swapped = True + destination.rename(original_destination) + destination.symlink_to(redirect, target_is_directory=True) + return real_temporary_file_at(directory_fd, artifact_id, suffix) + + monkeypatch.setattr(vault, "_temporary_file_at", swap_then_create) + + with pytest.raises(vault.VaultError, match="destination changed"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert destination.is_symlink() + assert list(redirect.iterdir()) == [] + assert list(original_destination.iterdir()) == [] + + +def test_restore_rejects_destination_swap_during_publication(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("trusted", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + destination.mkdir(mode=0o700) + original_destination = tmp_path / "restore-original" + redirect = tmp_path / "redirect" + redirect.mkdir(mode=0o700) + real_link = vault._link_no_replace_at + swapped = False + + def swap_then_link(directory_fd: int, source_name: str, destination_name: str): + nonlocal swapped + if not swapped: + swapped = True + destination.rename(original_destination) + destination.symlink_to(redirect, target_is_directory=True) + return real_link(directory_fd, source_name, destination_name) + + monkeypatch.setattr(vault, "_link_no_replace_at", swap_then_link) + + with pytest.raises(vault.VaultError, match="destination changed"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert destination.is_symlink() + assert list(redirect.iterdir()) == [] + assert list(original_destination.iterdir()) == [] + + +def test_restore_atomic_publish_does_not_replace_concurrent_target( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +): + source = tmp_path / "private.md" + source.write_text("trusted", encoding="utf-8") + _add(vault, source) + destination = tmp_path / "restore" + destination.mkdir(mode=0o700) + final_path = destination / "artifact-001--private.md" + real_link = vault.os.link + + def concurrent_link( + source_path, + destination_path, + *, + src_dir_fd=None, + dst_dir_fd=None, + follow_symlinks=True, + ): + concurrent_fd = vault.os.open( + destination_path, + vault.os.O_WRONLY | vault.os.O_CREAT | vault.os.O_EXCL, + 0o600, + dir_fd=dst_dir_fd, + ) + try: + vault.os.write(concurrent_fd, b"concurrent") + finally: + vault.os.close(concurrent_fd) + return real_link( + source_path, + destination_path, + src_dir_fd=src_dir_fd, + dst_dir_fd=dst_dir_fd, + follow_symlinks=follow_symlinks, + ) + + monkeypatch.setattr(vault.os, "link", concurrent_link) + + with pytest.raises(vault.VaultError, match="already exists"): + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + + assert final_path.read_text(encoding="utf-8") == "concurrent" + assert [path for path in destination.iterdir() if path.name.startswith(".artifact-001.")] == [] + + +def test_restore_all_rolls_back_earlier_links_on_late_concurrent_target( + vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +): + first = tmp_path / "first.md" + first.write_text("first", encoding="utf-8") + second = tmp_path / "second.md" + second.write_text("second", encoding="utf-8") + monkeypatch.setattr(vault, "BOOTSTRAP_ARTIFACT_IDS", frozenset({"artifact-001", "artifact-002"})) + _add(vault, first, "artifact-001") + _add(vault, second, "artifact-002") + destination = tmp_path / "restore" + destination.mkdir(mode=0o700) + first_path = destination / "artifact-001--first.md" + second_path = destination / "artifact-002--second.md" + real_link = vault.os.link + links = 0 + + def race_second_link( + source_path, + destination_path, + *, + src_dir_fd=None, + dst_dir_fd=None, + follow_symlinks=True, + ): + nonlocal links + links += 1 + if links == 2: + concurrent_fd = vault.os.open( + destination_path, + vault.os.O_WRONLY | vault.os.O_CREAT | vault.os.O_EXCL, + 0o600, + dir_fd=dst_dir_fd, + ) + try: + vault.os.write(concurrent_fd, b"concurrent") + finally: + vault.os.close(concurrent_fd) + return real_link( + source_path, + destination_path, + src_dir_fd=src_dir_fd, + dst_dir_fd=dst_dir_fd, + follow_symlinks=follow_symlinks, + ) + + monkeypatch.setattr(vault.os, "link", race_second_link) + + with pytest.raises(vault.VaultError, match="already exists"): + vault.cmd_restore(SimpleNamespace(all=True, artifact_id=None, dest=str(destination), apply=True)) + + assert not first_path.exists() + assert second_path.read_text(encoding="utf-8") == "concurrent" + assert [path for path in destination.iterdir() if path.name.startswith(".artifact-")] == [] + + +def test_restore_all_preflights_every_target_before_publish(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + first = tmp_path / "first.md" + first.write_text("first", encoding="utf-8") + second = tmp_path / "second.md" + second.write_text("second", encoding="utf-8") + monkeypatch.setattr(vault, "BOOTSTRAP_ARTIFACT_IDS", frozenset({"artifact-001", "artifact-002"})) + _add(vault, first, "artifact-001") + _add(vault, second, "artifact-002") + destination = tmp_path / "restore" + destination.mkdir(mode=0o700) + existing = destination / "artifact-002--second.md" + existing.write_text("existing", encoding="utf-8") + + with pytest.raises(vault.VaultError, match="already exists"): + vault.cmd_restore(SimpleNamespace(all=True, artifact_id=None, dest=str(destination), apply=True)) + + assert not (destination / "artifact-001--first.md").exists() + assert existing.read_text(encoding="utf-8") == "existing" + + +def test_restore_bounds_output_name(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / ("n" * 240) + source.write_text("secret", encoding="utf-8") + artifact_id = "artifact-" + "1" * 55 + monkeypatch.setattr(vault, "BOOTSTRAP_ARTIFACT_IDS", frozenset({artifact_id})) + _add(vault, source, artifact_id) + destination = tmp_path / "restore" + + assert ( + vault.cmd_restore(SimpleNamespace(all=False, artifact_id=artifact_id, dest=str(destination), apply=True)) == 0 + ) + assert (destination / f"{artifact_id}--restored").read_text(encoding="utf-8") == "secret" + + +def test_add_stages_ciphertext_on_vault_filesystem(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + + def assert_vault_staging(envelope: Path, destination: Path) -> None: + assert destination.parent == vault.VAULT_DIR + shutil.copyfile(envelope, destination) + + monkeypatch.setattr(vault, "_encrypt_file", assert_vault_staging) + assert _add(vault, source) == 0 + + +def test_add_encrypts_one_immutable_snapshot(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("initial", encoding="utf-8") + original_write_envelope = vault._write_envelope + + def mutate_source_after_snapshot(*args, **kwargs): + source.write_text("changed-after-snapshot", encoding="utf-8") + original_write_envelope(*args, **kwargs) + + monkeypatch.setattr(vault, "_write_envelope", mutate_source_after_snapshot) + _add(vault, source) + destination = tmp_path / "restore" + vault.cmd_restore(SimpleNamespace(all=False, artifact_id="artifact-001", dest=str(destination), apply=True)) + assert (destination / "artifact-001--private.md").read_text(encoding="utf-8") == "initial" + + +def test_add_rejects_source_mutation_during_snapshot(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + original = b"a" * (2 << 20) + replacement = b"b" * len(original) + source.write_bytes(original) + begin_write = threading.Event() + write_finished = threading.Event() + + def writer() -> None: + assert begin_write.wait(5) + with source.open("r+b") as handle: + handle.write(replacement) + handle.flush() + vault.os.fsync(handle.fileno()) + write_finished.set() + + writer_thread = threading.Thread(target=writer) + writer_thread.start() + + def hybrid_copy(input_handle, output_handle): + digest = vault.hashlib.sha256() + first = input_handle.read(1) + output_handle.write(first) + digest.update(first) + begin_write.set() + assert write_finished.wait(5) + count = len(first) + for chunk in iter(lambda: input_handle.read(1 << 20), b""): + output_handle.write(chunk) + digest.update(chunk) + count += len(chunk) + return digest.hexdigest(), count + + monkeypatch.setattr(vault, "_copy_file_and_hash", hybrid_copy) + try: + with pytest.raises(vault.VaultError, match="source changed while creating its snapshot"): + _add(vault, source) + finally: + writer_thread.join(timeout=5) + + assert not writer_thread.is_alive() + assert not vault.MANIFEST.exists() + + +def test_add_stages_plaintext_on_source_filesystem(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "source-custody" / "private.md" + source.parent.mkdir() + source.write_text("synthetic", encoding="utf-8") + real_temporary_directory = tempfile.TemporaryDirectory + observed_directories: list[Path] = [] + + def custody_temporary_directory(*args, **kwargs): + observed_directories.append(Path(kwargs["dir"])) + return real_temporary_directory(*args, **kwargs) + + monkeypatch.setattr(vault.tempfile, "TemporaryDirectory", custody_temporary_directory) + + _add(vault, source) + + assert observed_directories == [source.parent] + + +def test_add_holds_lock_through_manifest_publication(vault, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + source = tmp_path / "private.md" + source.write_text("secret", encoding="utf-8") + events: list[str] = [] + original_write_manifest = vault._write_manifest + + @contextmanager + def observed_lock(): + events.append("lock-enter") + yield + events.append("lock-exit") + + def observed_write_manifest(rows): + events.append("manifest-write") + original_write_manifest(rows) + + monkeypatch.setattr(vault, "_vault_lock", observed_lock) + monkeypatch.setattr(vault, "_write_manifest", observed_write_manifest) + _add(vault, source) + assert events == ["lock-enter", "manifest-write", "lock-exit"] + + +def test_restore_selectors_are_mutually_exclusive(vault, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr( + vault.sys, + "argv", + ["private-vault.py", "restore", "--artifact-id", "artifact-001", "--all", "--apply"], + ) + with pytest.raises(SystemExit) as exc_info: + vault.main() + assert exc_info.value.code == 2 + + +def test_recovery_check_round_trips_only_synthetic_content(vault): + assert vault.cmd_recovery_check(SimpleNamespace(apply=True)) == 0 + + +def test_recovery_check_requires_apply(vault): + with pytest.raises(vault.VaultError, match="--apply"): + vault.cmd_recovery_check(SimpleNamespace(apply=False)) + + +def test_real_gpg_round_trip_with_scratch_key( + vault, + monkeypatch: pytest.MonkeyPatch, + request: pytest.FixtureRequest, + tmp_path: Path, +): + if shutil.which("gpg") is None: + pytest.skip("gpg is unavailable on this host") + # Keep the GnuPG agent socket path short while respecting the configured temp directory. + gnupghome = Path(tempfile.mkdtemp(prefix="limen-vault-gpg-")) + request.addfinalizer(lambda: shutil.rmtree(gnupghome, ignore_errors=True)) + gnupghome.chmod(0o700) + identity = "Limen Vault Recovery Test " + common = [ + "gpg", + "--batch", + "--homedir", + str(gnupghome), + "--pinentry-mode", + "loopback", + "--passphrase", + "", + ] + subprocess.run( + [*common, "--quick-generate-key", identity, "ed25519", "sign", "1d"], + check=True, + capture_output=True, + text=True, + ) + listing = subprocess.run( + ["gpg", "--batch", "--homedir", str(gnupghome), "--with-colons", "--list-secret-keys"], + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + fingerprint = next(line.split(":")[9] for line in listing if line.startswith("fpr:")) + subprocess.run( + [*common, "--quick-add-key", fingerprint, "cv25519", "encrypt", "1d"], + check=True, + capture_output=True, + text=True, + ) + listing = subprocess.run( + ["gpg", "--batch", "--homedir", str(gnupghome), "--with-colons", "--list-secret-keys"], + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + encryption_key_id = next(line.split(":")[4] for line in listing if line.startswith("ssb:")) + public_key = tmp_path / "scratch-public-key.asc" + public_key.write_text( + subprocess.run( + ["gpg", "--batch", "--homedir", str(gnupghome), "--armor", "--export", fingerprint], + check=True, + capture_output=True, + text=True, + ).stdout, + encoding="utf-8", + ) + monkeypatch.setattr(vault, "PUBKEY", public_key) + monkeypatch.setattr(vault, "FINGERPRINT", fingerprint) + monkeypatch.setattr(vault, "ENCRYPTION_SUBKEY_ID", encryption_key_id) + monkeypatch.setenv("GNUPGHOME", str(gnupghome)) + source = tmp_path / "synthetic-source" + ciphertext = tmp_path / "synthetic-source.gpg" + restored = tmp_path / "synthetic-restored" + source.write_bytes(vault.RECOVERY_CANARY) + + vault._real_encrypt_file(source, ciphertext) + assert vault._real_ciphertext_recipient_keyids(ciphertext) == {encryption_key_id} + vault._real_decrypt_file(ciphertext, restored) + assert restored.read_bytes() == vault.RECOVERY_CANARY diff --git a/institutio/governance/gates.yaml b/institutio/governance/gates.yaml index 3df3fa514..5c66a614e 100644 --- a/institutio/governance/gates.yaml +++ b/institutio/governance/gates.yaml @@ -467,6 +467,12 @@ gates: ci_job: "pr-gate.yml:pr-gate" owner: gitvs note: "A governed measurement document is written only by its keeper: every commit touching it is a keeper ship (A), every production toucher is declared (B), every committed row is a distinct correctly-ordered census (C). tasks.yaml had this via task-writer-audit.py; the debt ledger had a convention and nothing enforcing it — five of its six observations rode in as passengers on unrelated feature PRs. Registry: institutio/governance/ledger-custody.yaml; adding a governed document is one entry." + private-vault: + command: "python3 scripts/private-vault.py verify && bash scripts/run-pytest-hermetic.sh cli/tests/test_private_vault.py -q" + paths: ["institutio/vault/**", ".limen-private", ".limen-private/**", ".agent-runtime", ".agent-runtime/**", ".limen-workstream", ".limen-workstream/**", ".gitignore", "scripts/private-vault.py", "cli/tests/test_private_vault.py", "docs/keys/anthony-padavano-gpg.asc"] + ci_job: "pr-gate.yml:pr-gate" + owner: custody + note: "Git-tracked ciphertext custody for high-value private artifacts (research dossiers, strategy memos): every manifest row's .gpg exists, sha-matches, and is tracked; no plaintext source is ever tracked (leak check); no unmanifested ciphertext. .gitignore is secrecy, not custody — a gitignored dossier bought with real research spend has zero replication. Encrypt-only to the committed pubkey; decrypt requires Anthony's private key." nomenclator: command: "python3 scripts/nomenclator.py" paths: ["spec/**", "organs/**", "organ-ladder.json", "scripts/nomenclator.py", "scripts/heartbeat-loop.sh"] diff --git a/institutio/vault/artifact-001.gpg b/institutio/vault/artifact-001.gpg new file mode 100644 index 000000000..9fb5adad3 Binary files /dev/null and b/institutio/vault/artifact-001.gpg differ diff --git a/institutio/vault/artifact-002.gpg b/institutio/vault/artifact-002.gpg new file mode 100644 index 000000000..eb0d86d2a Binary files /dev/null and b/institutio/vault/artifact-002.gpg differ diff --git a/institutio/vault/artifact-003.gpg b/institutio/vault/artifact-003.gpg new file mode 100644 index 000000000..161208c24 Binary files /dev/null and b/institutio/vault/artifact-003.gpg differ diff --git a/institutio/vault/artifact-004.gpg b/institutio/vault/artifact-004.gpg new file mode 100644 index 000000000..015ac42a8 Binary files /dev/null and b/institutio/vault/artifact-004.gpg differ diff --git a/institutio/vault/manifest.jsonl b/institutio/vault/manifest.jsonl new file mode 100644 index 000000000..41f46a9a5 --- /dev/null +++ b/institutio/vault/manifest.jsonl @@ -0,0 +1,4 @@ +{"artifact_id":"artifact-001","ciphertext":"artifact-001.gpg","ciphertext_bytes":23259,"ciphertext_sha256":"abfa13c43ed4212bf5083a49696c88c28f3b2aa8c2c1f048076dbd13170b4692","recipient_fpr":"205A566A5FFE43D2E28E05A4C5B98FFAF8ED000E","schema":"private-vault-manifest-v2","vaulted_at":"2026-08-09T10:29:23+00:00"} +{"artifact_id":"artifact-002","ciphertext":"artifact-002.gpg","ciphertext_bytes":2616,"ciphertext_sha256":"5dc15ecb821778a2d40eb2f12357dcb5d84acbdb397dcc0486ff03548b27b856","recipient_fpr":"205A566A5FFE43D2E28E05A4C5B98FFAF8ED000E","schema":"private-vault-manifest-v2","vaulted_at":"2026-08-09T10:29:23+00:00"} +{"artifact_id":"artifact-003","ciphertext":"artifact-003.gpg","ciphertext_bytes":7206,"ciphertext_sha256":"b27449a0ea819371045dd2de4f4a2683f3ef735729de19b6934491415c514fb0","recipient_fpr":"205A566A5FFE43D2E28E05A4C5B98FFAF8ED000E","schema":"private-vault-manifest-v2","vaulted_at":"2026-08-09T10:29:23+00:00"} +{"artifact_id":"artifact-004","ciphertext":"artifact-004.gpg","ciphertext_bytes":1056,"ciphertext_sha256":"d5189f16a003781b60ae4c7fa2e591e03e1352d1c72eca6850c5f2e4f6d2c46e","recipient_fpr":"205A566A5FFE43D2E28E05A4C5B98FFAF8ED000E","schema":"private-vault-manifest-v2","vaulted_at":"2026-08-09T18:09:49+00:00"} diff --git a/scripts/private-vault.py b/scripts/private-vault.py new file mode 100755 index 000000000..b0e1a1bdf --- /dev/null +++ b/scripts/private-vault.py @@ -0,0 +1,1461 @@ +#!/usr/bin/env python3 +"""PRIVATE-VAULT — git-tracked ciphertext custody for private artifacts. + +Ciphertext and a deliberately minimal manifest are tracked. Plaintext names, source paths, +content hashes, sizes, and descriptions live only inside the encrypted envelope. The public +manifest contains only a neutral artifact id plus ciphertext custody metadata. + + add encrypt a file into a v2 envelope and append a public-safe manifest row + verify validate manifest schema, containment, ciphertext integrity, and git custody + restore decrypt, verify, and atomically publish one artifact (or --all) + recovery-check prove the real private key can round-trip a synthetic canary + list list neutral artifact ids, newest first + +The committed public key is the encryption source of truth. Decryption uses the operator's +normal GPG keyring and therefore still requires the private key. +""" + +from __future__ import annotations + +import argparse +import fcntl +import hashlib +import json +import os +import re +import secrets +import shutil +import stat +import subprocess +import sys +import tempfile +import time +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import BinaryIO, Iterator + +ROOT = Path(__file__).resolve().parent.parent +PUBKEY = ROOT / "docs" / "keys" / "anthony-padavano-gpg.asc" +VAULT_DIR = ROOT / "institutio" / "vault" +MANIFEST = VAULT_DIR / "manifest.jsonl" +FINGERPRINT = "205A566A5FFE43D2E28E05A4C5B98FFAF8ED000E" +ENCRYPTION_SUBKEY_ID = "7C99B54C1ED4B555" + +SCHEMA = "private-vault-manifest-v2" +MAGIC = b"LIMEN-PRIVATE-VAULT-V2\n" +MAX_HEADER_BYTES = 64 * 1024 +ARTIFACT_ID_RE = re.compile(r"^artifact-[0-9]{3,55}$") +SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +PUBLIC_FIELDS = { + "schema", + "artifact_id", + "ciphertext", + "ciphertext_sha256", + "ciphertext_bytes", + "recipient_fpr", + "vaulted_at", +} +BOOTSTRAP_ARTIFACT_IDS = frozenset( + { + "artifact-001", + "artifact-002", + "artifact-003", + "artifact-004", + } +) +PRIVATE_TRACKING_PREFIXES = (".limen-private/", ".agent-runtime/", ".limen-workstream/") +PRIVATE_TRACKING_ROOTS = tuple(prefix.rstrip("/") for prefix in PRIVATE_TRACKING_PREFIXES) +COMMAND_TIMEOUT_SECONDS = 120 +LOCK_TIMEOUT_SECONDS = 30 +DIAGNOSTIC_LIMIT = 4096 +RECOVERY_CANARY = b"LIMEN-PRIVATE-VAULT-RECOVERY-CANARY-V1\n" +PUBLIC_SAFE_HISTORY_ROOT = "eeaaa85b7e7270e1b9e9140b78f7ff2360e2524f" + + +class VaultError(RuntimeError): + """A user-facing vault contract failure.""" + + +def _diagnostic(run: subprocess.CompletedProcess[str]) -> str: + value = (run.stderr or run.stdout or "").strip() + if len(value) > DIAGNOSTIC_LIMIT: + return value[:DIAGNOSTIC_LIMIT] + "... [truncated]" + return value + + +def _run_command(args: list[str], *, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: + try: + return subprocess.run( + args, + env=env, + capture_output=True, + text=True, + errors="surrogateescape", + timeout=COMMAND_TIMEOUT_SECONDS, + ) + except FileNotFoundError as exc: + raise VaultError(f"required executable is unavailable: {args[0]}") from exc + except subprocess.TimeoutExpired as exc: + raise VaultError(f"{args[0]} exceeded the {COMMAND_TIMEOUT_SECONDS}s command deadline") from exc + + +def _git_common_directory() -> Path | None: + marker = ROOT / ".git" + if marker.is_dir(): + git_directory = marker + elif marker.is_file() and not marker.is_symlink(): + try: + header = marker.read_text(encoding="utf-8").strip() + except (OSError, UnicodeDecodeError) as exc: + raise VaultError("cannot establish authentic Git history") from exc + if not header.startswith("gitdir: "): + raise VaultError("cannot establish authentic Git history") + git_directory = Path(header.removeprefix("gitdir: ")) + if not git_directory.is_absolute(): + git_directory = marker.parent / git_directory + git_directory = git_directory.resolve() + else: + return None + common_marker = git_directory / "commondir" + if not common_marker.exists(): + return git_directory + try: + common_value = common_marker.read_text(encoding="utf-8").strip() + except (OSError, UnicodeDecodeError) as exc: + raise VaultError("cannot establish authentic Git history") from exc + if not common_value: + raise VaultError("cannot establish authentic Git history") + common_directory = Path(common_value) + if not common_directory.is_absolute(): + common_directory = git_directory / common_directory + return common_directory.resolve() + + +def _reject_legacy_grafts() -> None: + common_directory = _git_common_directory() + if common_directory is not None and os.path.lexists(common_directory / "info" / "grafts"): + raise VaultError("refusing rewritten Git custody history") + + +def _git_command(*args: str) -> subprocess.CompletedProcess[str]: + _reject_legacy_grafts() + return _run_command(["git", "--no-replace-objects", "-C", str(ROOT), *args]) + + +@contextmanager +def _vault_lock() -> Iterator[None]: + identity = hashlib.sha256(str(ROOT.resolve()).encode("utf-8")).hexdigest()[:16] + lock_path = Path(tempfile.gettempdir()) / f"limen-private-vault-{os.getuid()}-{identity}.lock" + flags = os.O_CREAT | os.O_RDWR | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(lock_path, flags, 0o600) + except OSError as exc: + raise VaultError(f"cannot open the bounded vault lock ({exc.errno})") from exc + lock_info = os.fstat(descriptor) + if ( + not stat.S_ISREG(lock_info.st_mode) + or lock_info.st_uid != os.getuid() + or stat.S_IMODE(lock_info.st_mode) != 0o600 + ): + os.close(descriptor) + raise VaultError("vault lock must be an owner-only regular file") + deadline = time.monotonic() + LOCK_TIMEOUT_SECONDS + try: + while True: + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + if time.monotonic() >= deadline: + raise VaultError(f"vault lock exceeded the {LOCK_TIMEOUT_SECONDS}s deadline") + time.sleep(0.05) + yield + finally: + try: + fcntl.flock(descriptor, fcntl.LOCK_UN) + finally: + os.close(descriptor) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _safe_artifact_id(value: str) -> str: + if not ARTIFACT_ID_RE.fullmatch(value or ""): + raise VaultError("artifact id must use the neutral artifact-NNN form") + return value + + +def _reject_duplicate_fields(pairs: list[tuple[str, object]]) -> dict: + value: dict = {} + for key, item in pairs: + if key in value: + raise VaultError("manifest object contains a duplicate field") + value[key] = item + return value + + +def _canonical_vaulted_at(value: object) -> bool: + if not isinstance(value, str): + return False + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return False + return parsed.tzinfo == timezone.utc and value == parsed.isoformat(timespec="seconds") + + +def _contained_file(base: Path, name: str) -> Path: + if not name or Path(name).name != name or Path(name).is_absolute(): + raise VaultError(f"unsafe vault filename: {name!r}") + base_resolved = base.resolve() + candidate = base / name + if candidate.parent.resolve() != base_resolved: + raise VaultError(f"vault filename escapes custody root: {name!r}") + return candidate + + +def _cipher_path(row: dict) -> Path: + artifact_id_value = row.get("artifact_id") + if not isinstance(artifact_id_value, str): + raise VaultError("artifact id must be a string") + artifact_id = _safe_artifact_id(artifact_id_value) + expected = f"{artifact_id}.gpg" + name = str(row.get("ciphertext") or "") + if name != expected: + raise VaultError(f"ciphertext for {artifact_id} must be named {expected}") + return _contained_file(VAULT_DIR, name) + + +def _read_manifest() -> list[dict]: + if MANIFEST.is_symlink(): + raise VaultError("manifest must be a regular non-symlink file") + if not MANIFEST.exists(): + return [] + if not MANIFEST.is_file(): + raise VaultError("manifest must be a regular non-symlink file") + rows: list[dict] = [] + for line_number, raw in enumerate(MANIFEST.read_text(encoding="utf-8").splitlines(), 1): + if not raw.strip(): + continue + try: + row = json.loads(raw, object_pairs_hook=_reject_duplicate_fields) + except json.JSONDecodeError as exc: + raise VaultError(f"manifest line {line_number} is invalid JSON: {exc.msg}") from exc + except VaultError as exc: + raise VaultError(f"manifest line {line_number} contains a duplicate field") from exc + if not isinstance(row, dict): + raise VaultError(f"manifest line {line_number} is not an object") + rows.append(row) + return rows + + +def _validate_public_row(row: dict, line_number: int) -> list[str]: + errors: list[str] = [] + extra = sorted(set(row) - PUBLIC_FIELDS) + missing = sorted(PUBLIC_FIELDS - set(row)) + if extra: + errors.append(f"manifest line {line_number} exposes unsupported fields") + if missing: + errors.append(f"manifest line {line_number} misses fields: {', '.join(missing)}") + if row.get("schema") != SCHEMA: + errors.append(f"manifest line {line_number} has unsupported schema") + try: + _cipher_path(row) + except VaultError as exc: + errors.append(f"manifest line {line_number}: {exc}") + cipher_sha = str(row.get("ciphertext_sha256") or "") + if not SHA256_RE.fullmatch(cipher_sha): + errors.append(f"manifest line {line_number} has invalid ciphertext sha256") + if not isinstance(row.get("ciphertext_bytes"), int) or row.get("ciphertext_bytes", -1) < 0: + errors.append(f"manifest line {line_number} has invalid ciphertext byte count") + if row.get("recipient_fpr") != FINGERPRINT: + errors.append(f"manifest line {line_number} has unexpected recipient fingerprint") + if not _canonical_vaulted_at(row.get("vaulted_at")): + errors.append(f"manifest line {line_number} has invalid vaulted_at") + return errors + + +def _write_manifest(rows: list[dict]) -> None: + VAULT_DIR.mkdir(parents=True, exist_ok=True) + fd, temporary_name = tempfile.mkstemp(prefix=".private-vault-manifest.", dir=ROOT) + temporary = Path(temporary_name) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") + handle.flush() + os.fsync(handle.fileno()) + os.chmod(temporary, 0o644) + os.replace(temporary, MANIFEST) + finally: + temporary.unlink(missing_ok=True) + + +def _tracked_files() -> set[str]: + run = _git_command("ls-files", "-z") + if run.returncode != 0: + raise VaultError(f"cannot inspect git custody: {_diagnostic(run)}") + return {path for path in run.stdout.split("\0") if path} + + +def _index_entry_matches_worktree(relative: str) -> bool: + staged = _git_command("ls-files", "--stage", "-z", "--", relative) + if staged.returncode != 0: + return False + entries = [entry for entry in staged.stdout.split("\0") if entry] + if len(entries) != 1 or "\t" not in entries[0]: + return False + metadata, staged_path = entries[0].split("\t", 1) + fields = metadata.split() + if len(fields) != 3 or fields[0] != "100644" or fields[2] != "0" or staged_path != relative: + return False + worktree = _git_command("hash-object", f"--path={relative}", "--", relative) + return worktree.returncode == 0 and worktree.stdout.strip() == fields[1] + + +def _custody_metadata(row: dict) -> tuple[str, str, int, str, str]: + ciphertext = row.get("ciphertext") + ciphertext_sha256 = row.get("ciphertext_sha256") + ciphertext_bytes = row.get("ciphertext_bytes") + recipient_fpr = row.get("recipient_fpr") + vaulted_at = row.get("vaulted_at") + if ( + not isinstance(ciphertext, str) + or not isinstance(ciphertext_sha256, str) + or not SHA256_RE.fullmatch(ciphertext_sha256) + or not isinstance(ciphertext_bytes, int) + or ciphertext_bytes < 0 + or not isinstance(recipient_fpr, str) + or not _canonical_vaulted_at(vaulted_at) + ): + raise VaultError("manifest contains invalid immutable custody metadata") + return ciphertext, ciphertext_sha256, ciphertext_bytes, recipient_fpr, vaulted_at + + +def _historical_artifacts() -> dict[str, tuple[str, str, int, str, str]]: + """Return immutable metadata for every neutral v2 artifact admitted to custody.""" + shallow = _git_command("rev-parse", "--is-shallow-repository") + if shallow.returncode != 0: + raise VaultError("cannot prove complete committed custody history") + if shallow.stdout.strip() != "false": + raise VaultError("committed custody history requires a non-shallow repository") + manifest_relative = MANIFEST.relative_to(ROOT).as_posix() + history = _git_command("rev-list", "--full-history", "HEAD", "--", manifest_relative) + if history.returncode != 0: + raise VaultError(f"cannot inspect manifest custody history: {_diagnostic(history)}") + safe_root_object = _git_command("cat-file", "-e", f"{PUBLIC_SAFE_HISTORY_ROOT}^{{commit}}") + safe_root_is_reachable = False + if safe_root_object.returncode == 0: + safe_root = _git_command("merge-base", "--is-ancestor", PUBLIC_SAFE_HISTORY_ROOT, "HEAD") + if safe_root.returncode not in {0, 1}: + raise VaultError("cannot inspect the public-safe manifest history boundary") + safe_root_is_reachable = safe_root.returncode == 0 + + artifacts: dict[str, tuple[str, str, int, str, str]] = {} + for revision in history.stdout.splitlines(): + presence = _git_command("ls-tree", "--name-only", "-z", revision, "--", manifest_relative) + if presence.returncode != 0: + raise VaultError("cannot inspect a committed manifest history tree") + historical_paths = [path for path in presence.stdout.split("\0") if path] + if not historical_paths: + # A deletion commit is part of the fixed-path history but has no file at that revision. + continue + if historical_paths != [manifest_relative]: + raise VaultError("committed manifest history tree returned an unexpected path") + snapshot = _git_command("show", f"{revision}:{manifest_relative}") + if snapshot.returncode != 0: + raise VaultError("cannot read a committed manifest history snapshot") + require_public_safe = True + if safe_root_is_reachable: + before_safe_root = _git_command("merge-base", "--is-ancestor", revision, PUBLIC_SAFE_HISTORY_ROOT) + if before_safe_root.returncode not in {0, 1}: + raise VaultError("cannot classify a committed manifest history revision") + require_public_safe = revision == PUBLIC_SAFE_HISTORY_ROOT or before_safe_root.returncode == 1 + for line_number, raw in enumerate(snapshot.stdout.splitlines(), 1): + if not raw.strip(): + continue + try: + row = json.loads(raw, object_pairs_hook=_reject_duplicate_fields) + except json.JSONDecodeError as exc: + raise VaultError(f"committed manifest history contains invalid JSON at line {line_number}") from exc + except VaultError as exc: + raise VaultError( + f"committed manifest history contains a duplicate field at line {line_number}" + ) from exc + if not isinstance(row, dict): + if require_public_safe: + raise VaultError("committed manifest history contains a non-public-safe row") + continue + if require_public_safe and _validate_public_row(row, line_number): + raise VaultError("committed manifest history contains a non-public-safe row") + if row.get("schema") != SCHEMA: + continue + artifact_id = row.get("artifact_id") + if not isinstance(artifact_id, str): + raise VaultError("committed manifest history contains a non-string artifact id") + if ARTIFACT_ID_RE.fullmatch(artifact_id): + metadata = _custody_metadata(row) + previous = artifacts.get(artifact_id) + if previous is not None and previous != metadata: + raise VaultError("committed custody history changes immutable artifact metadata") + artifacts[artifact_id] = metadata + return artifacts + + +def _require_committed_custody(rows: list[dict]) -> None: + historical_artifacts = _historical_artifacts() + for row in rows: + artifact_id_value = row.get("artifact_id") + if not isinstance(artifact_id_value, str): + raise VaultError("restore target has an invalid artifact id") + artifact_id = _safe_artifact_id(artifact_id_value) + expected_metadata = historical_artifacts.get(artifact_id) + if expected_metadata is None: + raise VaultError(f"restore target is not admitted to committed custody: {artifact_id}") + if _custody_metadata(row) != expected_metadata: + raise VaultError(f"restore target differs from committed custody history: {artifact_id}") + + +def _reject_tracked_plaintext(source: Path) -> tuple[int, int]: + source_info = source.stat() + source_identity = source_info.st_dev, source_info.st_ino + tracked = _tracked_files() + for tracked_path in tracked: + try: + tracked_info = (ROOT / tracked_path).stat(follow_symlinks=False) + except (FileNotFoundError, NotADirectoryError, OSError): + continue + if stat.S_ISREG(tracked_info.st_mode) and (tracked_info.st_dev, tracked_info.st_ino) == source_identity: + raise VaultError("refusing to vault plaintext whose file object is already git-tracked") + try: + relative = source.relative_to(ROOT.resolve()).as_posix() + except ValueError: + return source_identity + if relative in tracked: + raise VaultError("refusing to vault plaintext that is already git-tracked") + if not any(relative.startswith(prefix) for prefix in PRIVATE_TRACKING_PREFIXES): + raise VaultError("repository-local plaintext must remain under a gitignored private namespace") + return source_identity + + +def _require_untracked_source_identity(source: Path, expected_identity: tuple[int, int]) -> None: + current_identity = _reject_tracked_plaintext(source) + if current_identity != expected_identity: + raise VaultError("private source changed during custody admission") + + +def _gpg_env(gnupghome: str) -> dict: + env = dict(os.environ) + env["GNUPGHOME"] = gnupghome + return env + + +def _import_pubkey(gnupghome: str) -> None: + if not PUBKEY.is_file() or PUBKEY.is_symlink(): + raise VaultError("committed public key must be a regular non-symlink file") + run = _run_command( + ["gpg", "--batch", "--import", str(PUBKEY)], + env=_gpg_env(gnupghome), + ) + if run.returncode != 0: + raise VaultError(f"public-key import failed: {_diagnostic(run)}") + exported = _run_command( + ["gpg", "--batch", "--armor", "--export", FINGERPRINT], + env=_gpg_env(gnupghome), + ) + if exported.returncode != 0 or not exported.stdout: + raise VaultError("canonical public-key export failed") + try: + committed_armor = PUBKEY.read_text(encoding="ascii") + except (OSError, UnicodeDecodeError) as exc: + raise VaultError("committed public-key file is not canonical ASCII armor") from exc + if committed_armor != exported.stdout: + raise VaultError("committed public-key file must contain only canonical public-key armor") + secret_listing = _run_command( + ["gpg", "--batch", "--with-colons", "--list-secret-keys"], + env=_gpg_env(gnupghome), + ) + if secret_listing.returncode != 0: + raise VaultError(f"secret-key inspection failed: {_diagnostic(secret_listing)}") + if any(line.startswith(("sec:", "ssb:")) for line in secret_listing.stdout.splitlines()): + raise VaultError("committed public-key file contains secret-key material") + + +def _validate_committed_pubkey() -> None: + with tempfile.TemporaryDirectory() as gnupghome: + os.chmod(gnupghome, 0o700) + _import_pubkey(gnupghome) + listing = _run_command( + ["gpg", "--batch", "--with-colons", "--fingerprint", "--fingerprint", "--list-keys"], + env=_gpg_env(gnupghome), + ) + if listing.returncode != 0: + raise VaultError(f"committed public-key inspection failed: {_diagnostic(listing)}") + + primary_fingerprints: set[str] = set() + encryption_subkeys: set[str] = set() + pending_key: tuple[str, str, str] | None = None + for line in listing.stdout.splitlines(): + fields = line.split(":") + record_type = fields[0] if fields else "" + if record_type in {"pub", "sub"} and len(fields) > 11: + pending_key = (record_type, fields[4].upper(), fields[11].lower()) + elif record_type == "fpr" and len(fields) > 9 and pending_key is not None: + key_type, key_id, capabilities = pending_key + if key_type == "pub": + primary_fingerprints.add(fields[9].upper()) + elif "e" in capabilities: + encryption_subkeys.add(key_id) + pending_key = None + + if primary_fingerprints != {FINGERPRINT}: + raise VaultError("committed public key does not match the pinned primary fingerprint") + if ENCRYPTION_SUBKEY_ID not in encryption_subkeys: + raise VaultError("committed public key lacks the pinned encryption subkey") + + canary = Path(gnupghome) / "synthetic-canary" + ciphertext = Path(gnupghome) / "synthetic-canary.gpg" + canary.write_bytes(RECOVERY_CANARY) + os.chmod(canary, 0o600) + probe = _run_command( + [ + "gpg", + "--batch", + "--yes", + "--trust-model", + "always", + "--recipient", + f"{ENCRYPTION_SUBKEY_ID}!", + "--output", + str(ciphertext), + "--encrypt", + str(canary), + ], + env=_gpg_env(gnupghome), + ) + if probe.returncode != 0 or not ciphertext.is_file(): + raise VaultError("pinned encryption subkey is unusable") + + +def _encrypt_file(source: Path, destination: Path) -> None: + with tempfile.TemporaryDirectory() as gnupghome: + os.chmod(gnupghome, 0o700) + _import_pubkey(gnupghome) + run = _run_command( + [ + "gpg", + "--batch", + "--yes", + "--trust-model", + "always", + "--recipient", + f"{ENCRYPTION_SUBKEY_ID}!", + "--output", + str(destination), + "--encrypt", + str(source), + ], + env=_gpg_env(gnupghome), + ) + if run.returncode != 0 or not destination.is_file(): + raise VaultError(f"encryption failed: {_diagnostic(run)}") + + +def _decrypt_file(source: Path, destination: Path) -> None: + run = _run_command(["gpg", "--batch", "--yes", "--output", str(destination), "--decrypt", str(source)]) + if run.returncode != 0 or not destination.is_file(): + raise VaultError(f"decryption failed (private key required): {_diagnostic(run)}") + + +def _run_gpg_from_fd( + args: list[str], source_fd: int, destination_fd: int | None = None +) -> subprocess.CompletedProcess[str]: + os.lseek(source_fd, 0, os.SEEK_SET) + if destination_fd is not None: + os.ftruncate(destination_fd, 0) + os.lseek(destination_fd, 0, os.SEEK_SET) + with os.fdopen(os.dup(source_fd), "rb") as source_handle: + destination_handle = os.fdopen(os.dup(destination_fd), "wb") if destination_fd is not None else None + try: + return subprocess.run( + args, + stdin=source_handle, + stdout=destination_handle if destination_handle is not None else subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + errors="surrogateescape", + timeout=COMMAND_TIMEOUT_SECONDS, + ) + except FileNotFoundError as exc: + raise VaultError("required executable is unavailable: gpg") from exc + except subprocess.TimeoutExpired as exc: + raise VaultError(f"gpg exceeded the {COMMAND_TIMEOUT_SECONDS}s command deadline") from exc + finally: + if destination_handle is not None: + destination_handle.close() + + +def _decrypt_descriptors(source_fd: int, destination_fd: int) -> None: + run = _run_gpg_from_fd(["gpg", "--batch", "--yes", "--decrypt"], source_fd, destination_fd) + if run.returncode != 0: + raise VaultError(f"decryption failed (private key required): {_diagnostic(run)}") + os.fsync(destination_fd) + + +def _ciphertext_recipient_keyids_fd(ciphertext_fd: int) -> set[str]: + run = _run_gpg_from_fd( + ["gpg", "--batch", "--list-only", "--status-fd", "1", "--decrypt"], + ciphertext_fd, + ) + if run.returncode != 0: + raise VaultError(f"cannot inspect ciphertext recipient: {_diagnostic(run)}") + recipients = { + fields[2].upper() + for line in run.stdout.splitlines() + if line.startswith("[GNUPG:] ENC_TO ") and len(fields := line.split()) >= 3 + } + if not recipients: + raise VaultError("ciphertext has no inspectable OpenPGP recipient") + return recipients + + +def _ciphertext_recipient_keyids(ciphertext: Path) -> set[str]: + run = _run_command(["gpg", "--batch", "--list-only", "--status-fd", "1", "--decrypt", str(ciphertext)]) + if run.returncode != 0: + raise VaultError(f"cannot inspect ciphertext recipient: {_diagnostic(run)}") + recipients = { + fields[2].upper() + for line in run.stdout.splitlines() + if line.startswith("[GNUPG:] ENC_TO ") and len(fields := line.split()) >= 3 + } + if not recipients: + raise VaultError("ciphertext has no inspectable OpenPGP recipient") + return recipients + + +def _openpgp_packet_tags_handle(handle: BinaryIO, size: int) -> list[int]: + tags: list[int] = [] + + def read_octet(handle) -> int: + raw = handle.read(1) + if len(raw) != 1: + raise VaultError("ciphertext has truncated OpenPGP framing") + return raw[0] + + def skip_body(handle, length: int) -> None: + if length < 0 or handle.tell() + length > size: + raise VaultError("ciphertext has invalid OpenPGP packet length") + handle.seek(length, os.SEEK_CUR) + + def read_new_length(handle) -> tuple[int, bool]: + first = read_octet(handle) + if first < 192: + return first, False + if first < 224: + second = read_octet(handle) + return ((first - 192) << 8) + second + 192, False + if first == 255: + raw = handle.read(4) + if len(raw) != 4: + raise VaultError("ciphertext has truncated OpenPGP packet length") + return int.from_bytes(raw, "big"), False + return 1 << (first & 0x1F), True + + while handle.tell() < size: + header = read_octet(handle) + if not header & 0x80: + raise VaultError("ciphertext contains bytes outside OpenPGP packet framing") + if header & 0x40: + tags.append(header & 0x3F) + length, partial = read_new_length(handle) + skip_body(handle, length) + while partial: + length, partial = read_new_length(handle) + skip_body(handle, length) + continue + + tags.append((header >> 2) & 0x0F) + length_type = header & 0x03 + if length_type == 3: + raise VaultError("ciphertext uses indeterminate OpenPGP packet framing") + length_octets = (1, 2, 4)[length_type] + raw_length = handle.read(length_octets) + if len(raw_length) != length_octets: + raise VaultError("ciphertext has truncated OpenPGP packet length") + skip_body(handle, int.from_bytes(raw_length, "big")) + + if tags not in ([1, 18], [1, 20]): + raise VaultError("ciphertext has an unexpected OpenPGP packet sequence") + return tags + + +def _openpgp_packet_tags(ciphertext: Path) -> list[int]: + """Parse complete OpenPGP packet framing without decrypting packet bodies.""" + with ciphertext.open("rb") as handle: + return _openpgp_packet_tags_handle(handle, ciphertext.stat().st_size) + + +def _openpgp_packet_tags_fd(ciphertext_fd: int) -> list[int]: + os.lseek(ciphertext_fd, 0, os.SEEK_SET) + with os.fdopen(os.dup(ciphertext_fd), "rb") as handle: + return _openpgp_packet_tags_handle(handle, os.fstat(ciphertext_fd).st_size) + + +def _ciphertext_failures(row: dict, path: Path) -> list[str]: + name = str(row.get("ciphertext") or "") + failures: list[str] = [] + if not path.is_file() or path.is_symlink(): + return [f"missing or unsafe ciphertext: {name}"] + if _sha256(path) != row.get("ciphertext_sha256"): + failures.append(f"ciphertext sha mismatch: {name}") + if path.stat().st_size != row.get("ciphertext_bytes"): + failures.append(f"ciphertext byte count mismatch: {name}") + if not failures: + try: + _openpgp_packet_tags(path) + recipients = _ciphertext_recipient_keyids(path) + except VaultError as exc: + failures.append(f"ciphertext recipient inspection failed: {name}: {exc}") + else: + if recipients != {ENCRYPTION_SUBKEY_ID}: + failures.append(f"ciphertext recipient mismatch: {name}") + return failures + + +def _sha256_fd(file_descriptor: int) -> str: + os.lseek(file_descriptor, 0, os.SEEK_SET) + digest = hashlib.sha256() + with os.fdopen(os.dup(file_descriptor), "rb") as handle: + for chunk in iter(lambda: handle.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _ciphertext_failures_fd(row: dict, file_descriptor: int) -> list[str]: + name = str(row.get("ciphertext") or "") + failures: list[str] = [] + info = os.fstat(file_descriptor) + if not stat.S_ISREG(info.st_mode): + return [f"missing or unsafe ciphertext: {name}"] + if _sha256_fd(file_descriptor) != row.get("ciphertext_sha256"): + failures.append(f"ciphertext sha mismatch: {name}") + if info.st_size != row.get("ciphertext_bytes"): + failures.append(f"ciphertext byte count mismatch: {name}") + if not failures: + try: + _openpgp_packet_tags_fd(file_descriptor) + recipients = _ciphertext_recipient_keyids_fd(file_descriptor) + except VaultError as exc: + failures.append(f"ciphertext recipient inspection failed: {name}: {exc}") + else: + if recipients != {ENCRYPTION_SUBKEY_ID}: + failures.append(f"ciphertext recipient mismatch: {name}") + return failures + + +def _copy_file_and_hash(source: BinaryIO, destination: BinaryIO) -> tuple[str, int]: + digest = hashlib.sha256() + count = 0 + for chunk in iter(lambda: source.read(1 << 20), b""): + destination.write(chunk) + digest.update(chunk) + count += len(chunk) + return digest.hexdigest(), count + + +def _snapshot_source(source: Path, destination: Path, expected_identity: tuple[int, int]) -> tuple[str, int]: + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + source_fd = os.open(source, flags) + except OSError as exc: + raise VaultError("cannot open a stable private source snapshot") from exc + try: + before = os.fstat(source_fd) + if not stat.S_ISREG(before.st_mode): + raise VaultError("private snapshot source is not a regular file") + if (before.st_dev, before.st_ino) != expected_identity: + raise VaultError("private source changed before snapshot creation") + with os.fdopen(source_fd, "rb", closefd=False) as input_handle, destination.open("wb") as output: + result = _copy_file_and_hash(input_handle, output) + output.flush() + os.fsync(output.fileno()) + after = os.fstat(source_fd) + finally: + os.close(source_fd) + before_identity = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) + after_identity = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns) + if before_identity != after_identity or result[1] != before.st_size: + raise VaultError("private source changed while creating its snapshot") + os.chmod(destination, 0o600) + return result + + +def _write_envelope( + snapshot: Path, + artifact_id: str, + original_name: str, + plaintext_sha256: str, + plaintext_bytes: int, + destination: Path, +) -> None: + header = { + "artifact_id": artifact_id, + "original_name": original_name, + "plaintext_sha256": plaintext_sha256, + "plaintext_bytes": plaintext_bytes, + } + encoded_header = json.dumps(header, sort_keys=True, separators=(",", ":")).encode("utf-8") + if len(encoded_header) > MAX_HEADER_BYTES: + raise VaultError("encrypted envelope header is too large") + with destination.open("wb") as output, snapshot.open("rb") as input_handle: + output.write(MAGIC) + output.write(encoded_header + b"\n") + shutil.copyfileobj(input_handle, output, length=1 << 20) + + +def _extract_envelope(envelope: Path, artifact_id: str, destination: Path) -> str: + digest = hashlib.sha256() + count = 0 + with envelope.open("rb") as source: + if source.readline(len(MAGIC) + 1) != MAGIC: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid magic") + raw_header = source.readline(MAX_HEADER_BYTES + 1) + if not raw_header.endswith(b"\n") or len(raw_header) > MAX_HEADER_BYTES: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid header") + try: + header = json.loads(raw_header) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid metadata") from exc + if header.get("artifact_id") != artifact_id: + raise VaultError(f"decrypted envelope identity mismatch for {artifact_id}") + original_name = str(header.get("original_name") or "") + if not original_name or Path(original_name).name != original_name: + raise VaultError(f"decrypted envelope for {artifact_id} has unsafe output name") + expected_sha = str(header.get("plaintext_sha256") or "") + expected_bytes = header.get("plaintext_bytes") + if not SHA256_RE.fullmatch(expected_sha): + raise VaultError(f"decrypted envelope for {artifact_id} has invalid plaintext hash") + if not isinstance(expected_bytes, int) or expected_bytes < 0: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid plaintext size") + with destination.open("wb") as output: + for chunk in iter(lambda: source.read(1 << 20), b""): + output.write(chunk) + digest.update(chunk) + count += len(chunk) + if digest.hexdigest() != expected_sha or count != expected_bytes: + raise VaultError(f"restored plaintext integrity mismatch for {artifact_id}") + return original_name + + +def _extract_envelope_descriptors(envelope_fd: int, artifact_id: str, destination_fd: int) -> str: + digest = hashlib.sha256() + count = 0 + os.lseek(envelope_fd, 0, os.SEEK_SET) + os.ftruncate(destination_fd, 0) + os.lseek(destination_fd, 0, os.SEEK_SET) + with ( + os.fdopen(os.dup(envelope_fd), "rb") as source, + os.fdopen(os.dup(destination_fd), "wb") as output, + ): + if source.readline(len(MAGIC) + 1) != MAGIC: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid magic") + raw_header = source.readline(MAX_HEADER_BYTES + 1) + if not raw_header.endswith(b"\n") or len(raw_header) > MAX_HEADER_BYTES: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid header") + try: + header = json.loads(raw_header) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid metadata") from exc + if header.get("artifact_id") != artifact_id: + raise VaultError(f"decrypted envelope identity mismatch for {artifact_id}") + original_name = str(header.get("original_name") or "") + if not original_name or Path(original_name).name != original_name: + raise VaultError(f"decrypted envelope for {artifact_id} has unsafe output name") + expected_sha = str(header.get("plaintext_sha256") or "") + expected_bytes = header.get("plaintext_bytes") + if not SHA256_RE.fullmatch(expected_sha): + raise VaultError(f"decrypted envelope for {artifact_id} has invalid plaintext hash") + if not isinstance(expected_bytes, int) or expected_bytes < 0: + raise VaultError(f"decrypted envelope for {artifact_id} has invalid plaintext size") + for chunk in iter(lambda: source.read(1 << 20), b""): + output.write(chunk) + digest.update(chunk) + count += len(chunk) + output.flush() + os.fsync(output.fileno()) + if digest.hexdigest() != expected_sha or count != expected_bytes: + raise VaultError(f"restored plaintext integrity mismatch for {artifact_id}") + return original_name + + +def cmd_add(args: argparse.Namespace) -> int: + if not getattr(args, "apply", False): + raise VaultError("add is mutating; rerun with --apply") + artifact_id = _safe_artifact_id(args.artifact_id) + if artifact_id not in BOOTSTRAP_ARTIFACT_IDS: + raise VaultError("artifact id has no public recovery admission proof") + source = Path(args.file).expanduser().resolve() + if not source.is_file(): + raise VaultError("source is not a file") + source_identity = _reject_tracked_plaintext(source) + with _vault_lock(): + rows = _read_manifest() + for line_number, row in enumerate(rows, 1): + errors = _validate_public_row(row, line_number) + if errors: + raise VaultError("; ".join(errors)) + if row["artifact_id"] == artifact_id: + raise VaultError(f"artifact id is already vaulted: {artifact_id}") + + VAULT_DIR.mkdir(parents=True, exist_ok=True) + cipher_name = f"{artifact_id}.gpg" + cipher_path = _contained_file(VAULT_DIR, cipher_name) + if cipher_path.exists(): + raise VaultError(f"ciphertext already exists without a matching manifest row: {cipher_name}") + + temporary_cipher = _temporary_file(VAULT_DIR, artifact_id, ".ciphertext.gpg") + cipher_published = False + try: + try: + temporary_context = tempfile.TemporaryDirectory( + prefix=".limen-vault-plaintext-", + dir=source.parent, + ) + except OSError as exc: + raise VaultError("cannot create a secure temporary directory on the source filesystem") from exc + with temporary_context as temporary_directory: + temporary_root = Path(temporary_directory) + os.chmod(temporary_root, 0o700) + snapshot = temporary_root / "snapshot" + envelope = temporary_root / "envelope" + plaintext_sha256, plaintext_bytes = _snapshot_source(source, snapshot, source_identity) + _require_untracked_source_identity(source, source_identity) + _write_envelope( + snapshot, + artifact_id, + source.name, + plaintext_sha256, + plaintext_bytes, + envelope, + ) + _encrypt_file(envelope, temporary_cipher) + _require_untracked_source_identity(source, source_identity) + os.chmod(temporary_cipher, 0o644) + os.replace(temporary_cipher, cipher_path) + cipher_published = True + row = { + "schema": SCHEMA, + "artifact_id": artifact_id, + "ciphertext": cipher_name, + "ciphertext_sha256": _sha256(cipher_path), + "ciphertext_bytes": cipher_path.stat().st_size, + "recipient_fpr": FINGERPRINT, + "vaulted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + } + _write_manifest([*rows, row]) + except BaseException: + if cipher_published: + cipher_path.unlink(missing_ok=True) + raise + finally: + temporary_cipher.unlink(missing_ok=True) + print(f"OK: vaulted {artifact_id} -> institutio/vault/{cipher_name}") + print(" next: git add the ciphertext and public-safe manifest") + return 0 + + +def cmd_verify(_args: argparse.Namespace) -> int: + tracked = _tracked_files() + failures: list[str] = [] + manifest_relative = MANIFEST.relative_to(ROOT).as_posix() + manifest_is_safe = MANIFEST.is_file() and not MANIFEST.is_symlink() + if not manifest_is_safe: + failures.append(f"required manifest is missing: {manifest_relative}") + rows: list[dict] = [] + else: + rows = _read_manifest() + if manifest_relative not in tracked: + failures.append(f"manifest not git-tracked (custody gap): {manifest_relative}") + elif not _index_entry_matches_worktree(manifest_relative): + failures.append("manifest Git index content differs from the validated worktree file") + pubkey_relative = PUBKEY.relative_to(ROOT).as_posix() + if pubkey_relative not in tracked or not _index_entry_matches_worktree(pubkey_relative): + failures.append("public-key Git index content differs from the validated worktree file") + try: + _validate_committed_pubkey() + except VaultError as exc: + failures.append(f"committed public-key validation failed: {exc}") + for prefix in PRIVATE_TRACKING_PREFIXES: + root = prefix.rstrip("/") + if any(path == root or path.startswith(prefix) for path in tracked): + failures.append(f"private plaintext namespace contains git-tracked content: {prefix}") + seen_ids: set[str] = set() + seen_ciphers: set[str] = set() + digest_owners: dict[str, str] = {} + expected_vault_files = {manifest_relative} + current_artifacts: dict[str, tuple[str, str, int, str, str]] = {} + for line_number, row in enumerate(rows, 1): + failures.extend(_validate_public_row(row, line_number)) + artifact_id = str(row.get("artifact_id") or "") + name = str(row.get("ciphertext") or "") + if artifact_id in seen_ids: + failures.append(f"duplicate artifact id: {artifact_id}") + if name in seen_ciphers: + failures.append(f"duplicate ciphertext: {name}") + digest = row.get("ciphertext_sha256") + if isinstance(digest, str) and SHA256_RE.fullmatch(digest): + previous_owner = digest_owners.get(digest) + if previous_owner is not None and previous_owner != artifact_id: + failures.append("duplicate ciphertext digest across distinct artifact ids") + else: + digest_owners[digest] = artifact_id + seen_ids.add(artifact_id) + seen_ciphers.add(name) + if ARTIFACT_ID_RE.fullmatch(artifact_id): + try: + current_artifacts[artifact_id] = _custody_metadata(row) + except VaultError: + pass + try: + path = _cipher_path(row) + except VaultError: + continue + failures.extend(_ciphertext_failures(row, path)) + if not path.is_file() or path.is_symlink(): + continue + relative = path.relative_to(ROOT).as_posix() + expected_vault_files.add(relative) + if relative not in tracked: + failures.append(f"ciphertext not git-tracked (custody gap): {relative}") + elif not _index_entry_matches_worktree(relative): + failures.append(f"ciphertext Git index content differs from the validated worktree file: {name}") + try: + historical_artifacts = _historical_artifacts() + custody_baseline = BOOTSTRAP_ARTIFACT_IDS | set(historical_artifacts) + except VaultError as exc: + failures.append(str(exc)) + historical_artifacts = {} + custody_baseline = BOOTSTRAP_ARTIFACT_IDS + missing_required = sorted(custody_baseline - seen_ids) + if missing_required: + failures.append(f"required custody baseline is missing neutral ids: {', '.join(missing_required)}") + unattested_ids = sorted(seen_ids - BOOTSTRAP_ARTIFACT_IDS) + if unattested_ids: + failures.append("non-bootstrap ciphertext lacks a public recovery admission proof") + for artifact_id, expected_metadata in historical_artifacts.items(): + current_metadata = current_artifacts.get(artifact_id) + if current_metadata is not None and current_metadata != expected_metadata: + failures.append(f"immutable custody metadata changed: {artifact_id}") + if any(path.startswith("institutio/vault/") and path not in expected_vault_files for path in tracked): + failures.append("Git index contains unmanifested vault content") + for stray in VAULT_DIR.iterdir() if VAULT_DIR.exists() else []: + if stray.is_symlink(): + failures.append(f"unsupported vault symlink: {stray.name}") + elif stray.is_dir(): + failures.append(f"unsupported vault directory: {stray.name}") + elif stray.suffix == ".gpg" and stray.name not in seen_ciphers: + failures.append(f"unmanifested ciphertext: {stray.name}") + elif stray.is_file() and stray != MANIFEST and stray.suffix != ".gpg": + failures.append(f"unsupported vault file: {stray.name}") + elif not stray.is_file(): + failures.append(f"unsupported vault entry: {stray.name}") + if failures: + print("FAIL: private-vault custody:") + for failure in failures: + print(f" - {failure}") + return 1 + print(f"OK: private-vault custody ({len(rows)} row(s); ciphertext tracked; manifest public-safe)") + return 0 + + +def _temporary_file(directory: Path, artifact_id: str, suffix: str) -> Path: + fd, name = tempfile.mkstemp(prefix=f".{artifact_id}.", suffix=suffix, dir=directory) + os.close(fd) + path = Path(name) + os.chmod(path, 0o600) + return path + + +def _temporary_file_at(directory_fd: int, artifact_id: str, suffix: str) -> tuple[str, int]: + flags = os.O_RDWR | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + for _attempt in range(128): + name = f".{artifact_id}.{secrets.token_hex(8)}{suffix}" + try: + file_descriptor = os.open(name, flags, 0o600, dir_fd=directory_fd) + except FileExistsError: + continue + os.fchmod(file_descriptor, 0o600) + return name, file_descriptor + raise VaultError("cannot allocate a private restore temporary") + + +def _restore_name(destination_fd: int, artifact_id: str, original_name: str) -> str: + proposed = f"{artifact_id}--{original_name}" + try: + name_max = os.fpathconf(destination_fd, "PC_NAME_MAX") + except (OSError, ValueError): + name_max = 255 + if len(os.fsencode(proposed)) <= name_max: + return proposed + return f"{artifact_id}--restored" + + +def _snapshot_ciphertext(source: Path, destination_fd: int) -> None: + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + source_fd = os.open(source, flags) + except OSError as exc: + raise VaultError("cannot open a stable ciphertext snapshot") from exc + try: + source_info = os.fstat(source_fd) + if not stat.S_ISREG(source_info.st_mode): + raise VaultError("ciphertext snapshot source is not a regular file") + os.ftruncate(destination_fd, 0) + os.lseek(destination_fd, 0, os.SEEK_SET) + with ( + os.fdopen(source_fd, "rb", closefd=False) as input_handle, + os.fdopen(destination_fd, "wb", closefd=False) as output_handle, + ): + shutil.copyfileobj(input_handle, output_handle, length=1 << 20) + output_handle.flush() + os.fsync(output_handle.fileno()) + finally: + os.close(source_fd) + os.fchmod(destination_fd, 0o600) + + +def _link_no_replace_at(directory_fd: int, source: str, destination: str) -> tuple[int, int]: + source_info = os.stat(source, dir_fd=directory_fd, follow_symlinks=False) + identity = source_info.st_dev, source_info.st_ino + try: + os.link( + source, + destination, + src_dir_fd=directory_fd, + dst_dir_fd=directory_fd, + follow_symlinks=False, + ) + except FileExistsError as exc: + raise VaultError(f"restore target already exists: {destination}") from exc + return identity + + +def _rollback_link_at(directory_fd: int, destination: str, identity: tuple[int, int]) -> None: + try: + current = os.stat(destination, dir_fd=directory_fd, follow_symlinks=False) + except FileNotFoundError: + return + if (current.st_dev, current.st_ino) == identity: + os.unlink(destination, dir_fd=directory_fd) + + +def _destination_identity(directory_fd: int) -> tuple[int, int]: + info = os.fstat(directory_fd) + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077: + raise VaultError("restore destination must be an owner-only directory") + return info.st_dev, info.st_ino + + +def _is_private_repository_destination(relative: Path) -> bool: + value = relative.as_posix() + return any( + value == root or value.startswith(prefix) + for root, prefix in zip(PRIVATE_TRACKING_ROOTS, PRIVATE_TRACKING_PREFIXES, strict=True) + ) + + +def _require_destination_route(destination_root: Path, *, repository_local_requested: bool) -> None: + try: + resolved_destination = destination_root.resolve(strict=False) + resolved_relative = resolved_destination.relative_to(ROOT.resolve()) + except (OSError, RuntimeError, ValueError) as exc: + if repository_local_requested: + raise VaultError("repository-local restore destination must remain in a private namespace") from exc + return + if not _is_private_repository_destination(resolved_relative): + raise VaultError("repository-local restore destination must remain in a private namespace") + + +def _require_destination_identity( + destination_root: Path, + identity: tuple[int, int], + *, + repository_local_requested: bool, +) -> None: + _require_destination_route(destination_root, repository_local_requested=repository_local_requested) + try: + current = destination_root.stat(follow_symlinks=False) + except OSError as exc: + raise VaultError("restore destination changed during operation") from exc + if not stat.S_ISDIR(current.st_mode) or (current.st_dev, current.st_ino) != identity: + raise VaultError("restore destination changed during operation") + _require_destination_route(destination_root, repository_local_requested=repository_local_requested) + + +def _entry_exists_at(directory_fd: int, name: str) -> bool: + try: + os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + except FileNotFoundError: + return False + return True + + +def cmd_restore(args: argparse.Namespace) -> int: + rows = _read_manifest() + for line_number, row in enumerate(rows, 1): + errors = _validate_public_row(row, line_number) + if errors: + raise VaultError("; ".join(errors)) + if args.all: + targets = rows + else: + artifact_id = _safe_artifact_id(args.artifact_id or "") + targets = [row for row in rows if row.get("artifact_id") == artifact_id] + if not targets: + raise VaultError("no matching vault entry (use list)") + _require_committed_custody(targets) + for row in targets: + failures = _ciphertext_failures(row, _cipher_path(row)) + if failures: + raise VaultError("; ".join(failures)) + if not getattr(args, "apply", False): + raise VaultError("restore is mutating; rerun with --apply") + destination_root = Path(os.path.abspath(os.fspath(Path(args.dest).expanduser()))) + try: + repository_destination = destination_root.relative_to(ROOT.resolve()) + except ValueError: + repository_destination = None + repository_local_requested = repository_destination is not None + if repository_destination is not None and not _is_private_repository_destination(repository_destination): + raise VaultError("repository-local restore destination must use a private namespace") + _require_destination_route(destination_root, repository_local_requested=repository_local_requested) + try: + destination_root.mkdir(parents=True, mode=0o700) + except FileExistsError: + created_destination = False + else: + created_destination = True + try: + initial_destination = destination_root.stat(follow_symlinks=False) + except OSError as exc: + raise VaultError("cannot establish the restore destination") from exc + if not stat.S_ISDIR(initial_destination.st_mode): + raise VaultError("restore destination must be an owner-only directory") + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + destination_fd = os.open(destination_root, directory_flags) + except OSError as exc: + raise VaultError("cannot pin the restore destination directory") from exc + try: + if created_destination: + os.fchmod(destination_fd, 0o700) + destination_identity = _destination_identity(destination_fd) + except Exception: + os.close(destination_fd) + raise + if destination_identity != (initial_destination.st_dev, initial_destination.st_ino): + os.close(destination_fd) + raise VaultError("restore destination changed during operation") + prepared: list[tuple[str, str, int, str, int, str, int, str]] = [] + temporaries: list[tuple[str, int]] = [] + final_names: set[str] = set() + published: list[tuple[str, tuple[int, int]]] = [] + succeeded = False + try: + _require_destination_identity( + destination_root, + destination_identity, + repository_local_requested=repository_local_requested, + ) + for row in targets: + artifact_id = _safe_artifact_id(str(row.get("artifact_id") or "")) + cipher_path = _cipher_path(row) + _require_destination_identity( + destination_root, + destination_identity, + repository_local_requested=repository_local_requested, + ) + cipher_name, cipher_fd = _temporary_file_at(destination_fd, artifact_id, ".ciphertext.gpg") + temporaries.append((cipher_name, cipher_fd)) + _require_destination_identity( + destination_root, + destination_identity, + repository_local_requested=repository_local_requested, + ) + envelope_name, envelope_fd = _temporary_file_at(destination_fd, artifact_id, ".envelope") + temporaries.append((envelope_name, envelope_fd)) + _require_destination_identity( + destination_root, + destination_identity, + repository_local_requested=repository_local_requested, + ) + plaintext_name, plaintext_fd = _temporary_file_at(destination_fd, artifact_id, ".plaintext") + temporaries.append((plaintext_name, plaintext_fd)) + _snapshot_ciphertext(cipher_path, cipher_fd) + failures = _ciphertext_failures_fd(row, cipher_fd) + if failures: + raise VaultError("; ".join(failures)) + _decrypt_descriptors(cipher_fd, envelope_fd) + original_name = _extract_envelope_descriptors(envelope_fd, artifact_id, plaintext_fd) + final_name = _restore_name(destination_fd, artifact_id, original_name) + if _entry_exists_at(destination_fd, final_name) or final_name in final_names: + raise VaultError(f"restore target already exists: {final_name}") + final_names.add(final_name) + prepared.append( + ( + artifact_id, + cipher_name, + cipher_fd, + envelope_name, + envelope_fd, + plaintext_name, + plaintext_fd, + final_name, + ) + ) + + _require_destination_identity( + destination_root, + destination_identity, + repository_local_requested=repository_local_requested, + ) + for ( + _artifact_id, + _cipher_name, + _cipher_fd, + _envelope_name, + _envelope_fd, + plaintext, + _plaintext_fd, + final, + ) in prepared: + published.append((final, _link_no_replace_at(destination_fd, plaintext, final))) + _require_destination_identity( + destination_root, + destination_identity, + repository_local_requested=repository_local_requested, + ) + for temporary_name, _file_descriptor in temporaries: + os.unlink(temporary_name, dir_fd=destination_fd) + succeeded = True + for ( + artifact_id, + _cipher_name, + _cipher_fd, + _envelope_name, + _envelope_fd, + _plaintext, + _plaintext_fd, + _final, + ) in prepared: + print(f"OK: restored {artifact_id} (pinned ciphertext and plaintext verified)") + finally: + if not succeeded: + for final_name, identity in reversed(published): + _rollback_link_at(destination_fd, final_name, identity) + for temporary_name, file_descriptor in reversed(temporaries): + try: + os.unlink(temporary_name, dir_fd=destination_fd) + except FileNotFoundError: + pass + finally: + os.close(file_descriptor) + if created_destination and not succeeded: + try: + _require_destination_identity( + destination_root, + destination_identity, + repository_local_requested=repository_local_requested, + ) + except OSError: + pass + except VaultError: + pass + else: + try: + destination_root.rmdir() + except OSError: + pass + os.close(destination_fd) + return 0 + + +def cmd_list(_args: argparse.Namespace) -> int: + rows = _read_manifest() + if not rows: + print("(vault empty)") + return 0 + for row in sorted(rows, key=lambda item: item.get("vaulted_at", ""), reverse=True): + print(f"{row.get('vaulted_at', '?')} {row.get('artifact_id', '?'):24s} {row.get('ciphertext', '?')}") + return 0 + + +def cmd_recovery_check(args: argparse.Namespace) -> int: + if not getattr(args, "apply", False): + raise VaultError("recovery-check writes a temporary plaintext canary; rerun with --apply") + with tempfile.TemporaryDirectory() as temporary_directory: + temporary_root = Path(temporary_directory) + source = temporary_root / "synthetic-canary" + ciphertext = temporary_root / "synthetic-canary.gpg" + restored = temporary_root / "synthetic-canary.restored" + source.write_bytes(RECOVERY_CANARY) + os.chmod(source, 0o600) + _encrypt_file(source, ciphertext) + if _ciphertext_recipient_keyids(ciphertext) != {ENCRYPTION_SUBKEY_ID}: + raise VaultError("synthetic recovery canary has the wrong recipient") + _decrypt_file(ciphertext, restored) + if restored.read_bytes() != RECOVERY_CANARY: + raise VaultError("synthetic recovery canary content mismatch") + os.chmod(restored, 0o600) + print("OK: real-key recovery canary passed (synthetic content only)") + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + sub = parser.add_subparsers(dest="cmd", required=True) + + add = sub.add_parser("add", help="encrypt a file into the vault") + add.add_argument("file") + add.add_argument("--artifact-id", required=True, help="neutral public id (lowercase letters, digits, hyphens)") + add.add_argument("--apply", action="store_true", help="authorize ciphertext and manifest writes") + add.set_defaults(fn=cmd_add) + + verify = sub.add_parser("verify", help="validate public-safe ciphertext custody") + verify.set_defaults(fn=cmd_verify) + + restore = sub.add_parser("restore", help="decrypt entries (requires private key)") + selectors = restore.add_mutually_exclusive_group(required=True) + selectors.add_argument("--artifact-id", help="neutral id to restore") + selectors.add_argument("--all", action="store_true") + restore.add_argument("--dest", default=str(Path.home() / ".limen-restore")) + restore.add_argument("--apply", action="store_true", help="authorize plaintext restoration") + restore.set_defaults(fn=cmd_restore) + + recovery = sub.add_parser("recovery-check", help="round-trip a synthetic canary with the real private key") + recovery.add_argument("--apply", action="store_true", help="authorize temporary plaintext canary writes") + recovery.set_defaults(fn=cmd_recovery_check) + + listing = sub.add_parser("list", help="list neutral artifact ids, newest first") + listing.set_defaults(fn=cmd_list) + + args = parser.parse_args() + try: + return args.fn(args) + except VaultError as exc: + print(f"FAIL: {exc}", file=sys.stderr) + return 1 + except OSError as exc: + print(f"FAIL: filesystem operation failed ({exc.errno})", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/verify-ci-hardening.test.sh b/scripts/tests/verify-ci-hardening.test.sh index 737354a8d..8ee93e149 100755 --- a/scripts/tests/verify-ci-hardening.test.sh +++ b/scripts/tests/verify-ci-hardening.test.sh @@ -44,7 +44,7 @@ flunk() { printf 'FAIL %s\n %s\n' "$1" "$2"; fails=$((fails + 1)); } make_sandbox() { local dir dir="$(mktemp -d "${TMPDIR:-/tmp}/verify-ci-hardening.XXXXXX")" - mkdir -p "$dir/scripts" "$dir/institutio/governance" "$dir/src" "$dir/web/app" "$dir/webish" + mkdir -p "$dir/scripts" "$dir/institutio/governance" "$dir/institutio/vault" "$dir/docs/keys" "$dir/src" "$dir/web/app" "$dir/webish" cp "$ROOT/scripts/verify.py" "$dir/scripts/verify.py" cat >"$dir/institutio/governance/gates.yaml" <<'YAML' schema_version: 0.1 @@ -70,8 +70,13 @@ gates: ci_job: "ci.yml:web" owner: verify note: "fixture gate mirrored in another workflow — must defer under --skip-ci-covered, never run" + deleted-custody: + command: "touch ran-deleted-custody" + paths: ["institutio/vault/**", "docs/keys/anthony-padavano-gpg.asc", ".limen-private", ".limen-private/**", ".agent-runtime", ".agent-runtime/**", ".limen-workstream", ".limen-workstream/**"] + owner: custody + note: "deleted custody paths must remain eligible for scoped gate selection" YAML - touch "$dir/src/.keep" "$dir/web/app/.keep" "$dir/webish/.keep" + touch "$dir/src/.keep" "$dir/institutio/vault/artifact.gpg" "$dir/docs/keys/anthony-padavano-gpg.asc" "$dir/web/app/.keep" "$dir/webish/.keep" git -C "$dir" init -q -b main git -C "$dir" -c user.email=t@t -c user.name=t add -A git -C "$dir" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm base @@ -110,7 +115,276 @@ out="$(python3 "$sb/scripts/verify.py" --changed --base HEAD 2>&1)" \ || flunk empty-diff-local "missing nothing-to-verify message: $out"; } \ || flunk empty-diff-local "non-zero exit without --require-base: $out" -# ── 4: deploy-trigger diff escalates to the whole matrix (seam) ──────────────── +# ── 4: deleting every custody file still selects its scoped gate ────────────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +rm "$sb/institutio/vault/artifact.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add -u institutio/vault/artifact.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "delete custody" +out_file="$sb/verify.out" +if python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base >"$out_file" 2>&1 +then + : +else + out="$(<"$out_file")" + flunk deleted-path-selects-gate "deleted-path run exited non-zero: $out" +fi +if [[ -f "$sb/ran-deleted-custody" ]] +then + pass deleted-path-selects-gate +else + out="$(<"$out_file")" + flunk deleted-path-selects-gate "deleted custody path was filtered out: $out" +fi + +# ── 4a: renaming every custody file still selects its scoped gate ──────────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +mkdir "$sb/elsewhere" +git -C "$sb" mv institutio/vault/artifact.gpg elsewhere/artifact.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "rename custody away" +out_file="$sb/verify.out" +if python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base >"$out_file" 2>&1 +then + : +else + out="$(<"$out_file")" + flunk renamed-path-selects-gate "renamed-path run exited non-zero: $out" +fi +if [[ -f "$sb/ran-deleted-custody" ]] +then + pass renamed-path-selects-gate +else + out="$(<"$out_file")" + flunk renamed-path-selects-gate "renamed custody source path was filtered out: $out" +fi + +# ── 4b: every private namespace selects the custody gate ───────────────────── +for private_path in \ + ".limen-private" \ + ".limen-private/probe" \ + ".limen-private/résumé.md" \ + ".agent-runtime" \ + ".agent-runtime/probe" \ + ".limen-workstream" \ + ".limen-workstream/probe" +do + sb="$(make_sandbox)" + base_sha="$(git -C "$sb" rev-parse HEAD)" + mkdir -p "$(dirname "$sb/$private_path")" + printf 'private\n' >"$sb/$private_path" + git -C "$sb" -c user.email=t@t -c user.name=t add -f "$private_path" + git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "add private namespace" + out_file="$sb/verify.out" + if python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base >"$out_file" 2>&1 + then + : + else + out="$(<"$out_file")" + flunk private-namespace-selects-gate "private namespace run exited non-zero: $out" + fi + if [[ -f "$sb/ran-deleted-custody" ]] + then + pass "private-namespace-selects-gate:$private_path" + else + flunk private-namespace-selects-gate "private namespace did not select custody gate: $private_path" + fi +done + +# ── 4bc: one final public custody version is eligible for vault validation ───── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +printf 'ciphertext\n' >"$sb/institutio/vault/artifact-new.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/artifact-new.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "add final custody version" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + || flunk final-custody-version "single final custody version exited non-zero: $out" +[[ -f "$sb/ran-deleted-custody" ]] \ + && pass final-custody-version \ + || flunk final-custody-version "single final custody version did not select its gate: $out" + +# ── 4c: add-then-delete private content is rejected without naming it ───────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +mkdir -p "$sb/.limen-private" +printf 'private\n' >"$sb/.limen-private/sensitive-probe" +git -C "$sb" -c user.email=t@t -c user.name=t add -f .limen-private/sensitive-probe +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "add transient private content" +git -C "$sb" rm -q .limen-private/sensitive-probe +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "delete transient private content" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk transient-private-history "exit 0 despite committed transient private content" \ + || { grep -q "refusing to expose or certify transient private content" <<<"$out" \ + && ! grep -q "sensitive-probe" <<<"$out" \ + && pass transient-private-history \ + || flunk transient-private-history "missing neutral refusal or leaked path: $out"; } + +# ── 4d: historical-only custody names fail without log leakage ──────────────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +printf 'ciphertext\n' >"$sb/institutio/vault/sensitive-probe.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/sensitive-probe.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "add transient custody path" +git -C "$sb" rm -q institutio/vault/sensitive-probe.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "delete transient custody path" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk historical-custody-redaction "exit 0 despite deleted transient custody" \ + || { grep -q "refusing to certify an unvalidated intermediate version" <<<"$out" \ + && ! grep -q "sensitive-probe" <<<"$out" \ + && pass historical-custody-redaction \ + || flunk historical-custody-redaction "missing neutral refusal or leaked path: $out"; } + +# ── 4e: reverted versions of tracked custody files fail without path leakage ── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +printf 'transient bytes\n' >>"$sb/institutio/vault/artifact.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/artifact.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "mutate tracked custody" +git -C "$sb" restore --source "$base_sha" -- institutio/vault/artifact.gpg +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/artifact.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "restore tracked custody" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk transient-custody-reversion "exit 0 despite an unvalidated intermediate custody version" \ + || { grep -q "refusing to certify an unvalidated intermediate version" <<<"$out" \ + && ! grep -q "artifact.gpg" <<<"$out" \ + && pass transient-custody-reversion \ + || flunk transient-custody-reversion "missing neutral refusal or leaked path: $out"; } + +# ── 4e1: superseded custody versions fail even when the final path changed ───── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +printf 'first version\n' >"$sb/institutio/vault/artifact.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/artifact.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "first custody version" +printf 'final version\n' >"$sb/institutio/vault/artifact.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/artifact.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "supersede custody version" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk superseded-custody-version "exit 0 despite an unvalidated superseded custody version" \ + || { grep -q "refusing to certify an unvalidated intermediate version" <<<"$out" \ + && ! grep -q "artifact.gpg" <<<"$out" \ + && pass superseded-custody-version \ + || flunk superseded-custody-version "missing neutral refusal or leaked path: $out"; } + +# ── 4e2: reverted public-key versions receive the same neutral refusal ───────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +printf 'transient key bytes\n' >>"$sb/docs/keys/anthony-padavano-gpg.asc" +git -C "$sb" -c user.email=t@t -c user.name=t add docs/keys/anthony-padavano-gpg.asc +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "mutate public key" +git -C "$sb" restore --source "$base_sha" -- docs/keys/anthony-padavano-gpg.asc +git -C "$sb" -c user.email=t@t -c user.name=t add docs/keys/anthony-padavano-gpg.asc +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "restore public key" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk transient-public-key "exit 0 despite an unvalidated intermediate public-key version" \ + || { grep -q "refusing to certify an unvalidated intermediate version" <<<"$out" \ + && ! grep -q "anthony-padavano" <<<"$out" \ + && pass transient-public-key \ + || flunk transient-public-key "missing neutral refusal or leaked path: $out"; } + +# ── 4ea: add-then-delete public custody files fail without path leakage ──────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +printf 'transient bytes\n' >"$sb/institutio/vault/résumé-cipher.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/résumé-cipher.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "add transient custody blob" +git -C "$sb" rm -q institutio/vault/résumé-cipher.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "delete transient custody blob" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk deleted-transient-custody "exit 0 despite a deleted transient custody blob" \ + || { grep -q "refusing to certify an unvalidated intermediate version" <<<"$out" \ + && ! grep -q "résumé-cipher" <<<"$out" \ + && pass deleted-transient-custody \ + || flunk deleted-transient-custody "missing neutral refusal or leaked path: $out"; } + +# ── 4f: synthetic merge inventory excludes paths changed only on the base ────── +sb="$(make_sandbox)" +git -C "$sb" switch -q -c feature +commit_touch "$sb" src/feature.txt +git -C "$sb" switch -q main +commit_touch "$sb" webish/base-only.txt +base_sha="$(git -C "$sb" rev-parse HEAD)" +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false merge -q --no-ff feature -m "synthetic PR merge" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + || flunk merge-base-path-exclusion "synthetic merge run exited non-zero: $out" +[[ -f "$sb/ran-runs-here" ]] \ + && ! grep -q "base-only" <<<"$out" \ + && pass merge-base-path-exclusion \ + || flunk merge-base-path-exclusion "feature gate missing or base-only path leaked into scope: $out" + +# ── 4g: merge-resolution-only private content remains in the inventory ───────── +sb="$(make_sandbox)" +git -C "$sb" switch -q -c feature +commit_touch "$sb" src/feature.txt +git -C "$sb" switch -q main +commit_touch "$sb" webish/base-only.txt +base_sha="$(git -C "$sb" rev-parse HEAD)" +git -C "$sb" switch -q feature +git -C "$sb" -c user.email=t@t -c user.name=t merge -q --no-ff --no-commit main +mkdir -p "$sb/.limen-private" +printf 'private\n' >"$sb/.limen-private/merge-only-probe" +git -C "$sb" -c user.email=t@t -c user.name=t add -f .limen-private/merge-only-probe +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "merge with private resolution" +git -C "$sb" rm -q .limen-private/merge-only-probe +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "delete merge-only private content" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk merge-resolution-private "exit 0 despite merge-created private history" \ + || { grep -q "refusing to expose or certify transient private content" <<<"$out" \ + && ! grep -q "merge-only-probe" <<<"$out" \ + && pass merge-resolution-private \ + || flunk merge-resolution-private "missing neutral refusal or leaked path: $out"; } + +# ── 4h: merge-resolution-only public custody content is also rejected ────────── +sb="$(make_sandbox)" +git -C "$sb" switch -q -c feature +commit_touch "$sb" src/feature.txt +git -C "$sb" switch -q main +commit_touch "$sb" webish/base-only.txt +base_sha="$(git -C "$sb" rev-parse HEAD)" +git -C "$sb" switch -q feature +git -C "$sb" -c user.email=t@t -c user.name=t merge -q --no-ff --no-commit main +printf 'ciphertext\n' >"$sb/institutio/vault/merge-only-cipher.gpg" +git -C "$sb" -c user.email=t@t -c user.name=t add institutio/vault/merge-only-cipher.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "merge with custody resolution" +git -C "$sb" rm -q institutio/vault/merge-only-cipher.gpg +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "delete merge-only custody content" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk merge-resolution-custody "exit 0 despite merge-created custody history" \ + || { grep -q "refusing to certify an unvalidated intermediate version" <<<"$out" \ + && ! grep -q "merge-only-cipher" <<<"$out" \ + && pass merge-resolution-custody \ + || flunk merge-resolution-custody "missing neutral refusal or leaked path: $out"; } + +# ── 4i: replacement refs cannot rewrite the committed path inventory ────────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +mkdir -p "$sb/.limen-private" +printf 'private\n' >"$sb/.limen-private/replace-probe" +git -C "$sb" -c user.email=t@t -c user.name=t add -f .limen-private/replace-probe +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "add replace probe" +private_commit="$(git -C "$sb" rev-parse HEAD)" +git -C "$sb" rm -q .limen-private/replace-probe +git -C "$sb" -c user.email=t@t -c user.name=t -c commit.gpgsign=false commit -qm "delete replace probe" +git -C "$sb" replace "$private_commit" "$base_sha" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk replace-objects-disabled "exit 0 after replacement history hid private content" \ + || { grep -q "refusing to expose or certify transient private content" <<<"$out" \ + && ! grep -q "replace-probe" <<<"$out" \ + && pass replace-objects-disabled \ + || flunk replace-objects-disabled "replacement object influenced inventory or leaked a path: $out"; } + +# ── 4j: legacy grafts are an explicit fail-closed history state ──────────────── +sb="$(make_sandbox)" +base_sha="$(git -C "$sb" rev-parse HEAD)" +commit_touch "$sb" src/graft-probe.txt +printf '%s\n' "$(git -C "$sb" rev-parse HEAD)" >"$sb/.git/info/grafts" +out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-base 2>&1)" \ + && flunk legacy-grafts-rejected "exit 0 with a legacy graft installed" \ + || { grep -q "refusing to verify rewritten Git history" <<<"$out" \ + && pass legacy-grafts-rejected \ + || flunk legacy-grafts-rejected "missing neutral graft refusal: $out"; } + +# ── 5: deploy-trigger diff escalates to the whole matrix (seam) ──────────────── sb="$(make_sandbox)" base_sha="$(git -C "$sb" rev-parse HEAD)" commit_touch "$sb" web/app/page.txt @@ -129,7 +403,7 @@ out="$(LIMEN_VERIFY_WHOLE_CMD="$sb/whole-marker.sh" \ && pass deploy-no-escalation-local \ || flunk deploy-no-escalation-local "escalated without --require-base" -# ── 5: queue integration reuses head matrix and runs scoped composition ──────── +# ── 6: queue integration reuses head matrix and runs scoped composition ──────── rm -f "$sb/whole-ran" out="$(LIMEN_VERIFY_WHOLE_CMD="$sb/whole-marker.sh" \ python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --integration 2>&1)" \ @@ -164,7 +438,7 @@ out="$(python3 "$sb/scripts/verify.py" --changed --base "$competing_sha" --integ && pass integration-rejects-common-ancestor \ || flunk integration-rejects-common-ancestor "missing exact-base refusal: $out"; } -# ── 6: --skip-ci-covered defers foreign-job mirrors, runs everything else ────── +# ── 7: --skip-ci-covered defers foreign-job mirrors, runs everything else ────── sb="$(make_sandbox)" base_sha="$(git -C "$sb" rev-parse HEAD)" commit_touch "$sb" webish/x.txt @@ -188,7 +462,7 @@ out="$(python3 "$sb/scripts/verify.py" --changed --base "$base_sha" --require-ba && pass own-job-still-runs \ || flunk own-job-still-runs "unmirrored/own-job gates were skipped: $out" -# ── 7: PR-lane opt-out — deploy diff stays scoped, never execs the matrix ────── +# ── 8: PR-lane opt-out — deploy diff stays scoped, never execs the matrix ────── sb="$(make_sandbox)" base_sha="$(git -C "$sb" rev-parse HEAD)" commit_touch "$sb" web/app/page.txt diff --git a/scripts/verify.py b/scripts/verify.py index e0f1ddf07..03acbf0ff 100755 --- a/scripts/verify.py +++ b/scripts/verify.py @@ -6,8 +6,8 @@ being two scripts and become two selections over the same data: verify.py --changed [--base REF] the scoped push gate: compute the changed set - (merge-base vs origin/main + staged + unstaged + - untracked), run exactly the implicated gates — + (every commit after the merge-base + staged + + unstaged + untracked), run exactly the implicated gates — each independent cheap/heavy tier runs as one parallel wave, then the explicitly serialized tail runs under the machine-wide flock verify-whole.sh @@ -72,6 +72,9 @@ ROOT = Path(__file__).resolve().parent.parent REGISTRY = ROOT / "institutio" / "governance" / "gates.yaml" +PRIVATE_CUSTODY_ROOTS = (".limen-private", ".agent-runtime", ".limen-workstream") +PUBLIC_CUSTODY_HISTORY_ROOT = "eeaaa85b7e7270e1b9e9140b78f7ff2360e2524f" +_GRAFT_PATH: Path | None = None class HostAdmissionFailure(RuntimeError): @@ -131,8 +134,38 @@ def load_registry() -> dict: return yaml.safe_load(REGISTRY.read_text()) +def _reject_legacy_grafts() -> None: + global _GRAFT_PATH + if _GRAFT_PATH is None: + probe = subprocess.run( + ["git", "--no-replace-objects", "rev-parse", "--git-path", "info/grafts"], + cwd=ROOT, + capture_output=True, + text=True, + errors="surrogateescape", + check=True, + ) + candidate = Path(probe.stdout.strip()) + _GRAFT_PATH = candidate if candidate.is_absolute() else ROOT / candidate + if os.path.lexists(_GRAFT_PATH): + raise RuntimeError("refusing to verify rewritten Git history") + + def git(*args: str) -> str: - return subprocess.run(["git", *args], cwd=ROOT, capture_output=True, text=True, check=True).stdout + _reject_legacy_grafts() + return subprocess.run( + ["git", "--no-replace-objects", *args], + cwd=ROOT, + capture_output=True, + text=True, + errors="surrogateescape", + check=True, + ).stdout + + +def git_paths(*args: str) -> list[str]: + """Return a NUL-delimited Git path list without core.quotePath rewriting.""" + return [path for path in git(*args).split("\0") if path] def resolve_merge_base(base: str | None) -> str: @@ -166,17 +199,129 @@ def integration_base(base: str | None) -> str: return supplied +def committed_path_changes(merge_base: str) -> list[tuple[str, str]]: + """Return PR-side non-merge status/path pairs after ``merge_base``. + + Endpoint diffs omit a path added in one PR commit and deleted in another. The + non-merge inventory keeps those paths visible to gate selection, while the + endpoint diff covers merge results. Merge commits are not expanded against + every parent because that misclassifies base-only paths as PR changes. Rename + detection stays disabled so both custody source and destination remain present. + """ + if not merge_base: + return [] + fields = [ + field + for field in git( + "log", + "--no-merges", + "--format=", + "--name-status", + "--no-renames", + "-z", + f"{merge_base}..HEAD", + ).split("\0") + if field + ] + if len(fields) % 2: + raise RuntimeError("git returned a malformed committed path inventory") + changes = list(zip(fields[::2], fields[1::2], strict=True)) + for revision in git("rev-list", "--merges", f"{merge_base}..HEAD").splitlines(): + merge_fields = [ + field + for field in git( + "diff-tree", + "--cc", + "--no-commit-id", + "--name-status", + "--no-renames", + "-r", + "-z", + revision, + ).split("\0") + if field + ] + if len(merge_fields) % 2: + raise RuntimeError("git returned a malformed merge-resolution path inventory") + for status, path in zip(merge_fields[::2], merge_fields[1::2], strict=True): + changes.append(("D" if status and set(status) == {"D"} else "M", path)) + return changes + + +def private_history_leak(base: str | None) -> bool: + """Detect transient committed private-namespace content without naming it.""" + merge_base = resolve_merge_base(base) + if not merge_base: + return False + tracked = set(git_paths("ls-files", "-z")) + return any( + status != "D" + and path not in tracked + and any(path == root or path.startswith(root + "/") for root in PRIVATE_CUSTODY_ROOTS) + for status, path in committed_path_changes(merge_base) + ) + + +def _is_public_custody_path(path: str) -> bool: + return path == "docs/keys/anthony-padavano-gpg.asc" or path.startswith("institutio/vault/") + + +def public_custody_history_start(base: str | None) -> str: + """Use the neutralization boundary when reachable, otherwise the PR base.""" + merge_base = resolve_merge_base(base) + safe_root = resolve_commit(PUBLIC_CUSTODY_HISTORY_ROOT) + if safe_root: + try: + git("merge-base", "--is-ancestor", safe_root, "HEAD") + except subprocess.CalledProcessError: + pass + else: + return safe_root + return merge_base + + +def transient_custody_reversion(base: str | None) -> bool: + """Detect deleted, reverted, or superseded unvalidated custody versions.""" + merge_base = resolve_merge_base(base) + if not merge_base: + return False + endpoint = set(git_paths("diff", "--name-only", "--no-renames", "-z", merge_base, "HEAD")) + versions: dict[str, int] = {} + for status, path in committed_path_changes(public_custody_history_start(base)): + if status != "D" and _is_public_custody_path(path): + versions[path] = versions.get(path, 0) + 1 + return any(path not in endpoint or count > 1 for path, count in versions.items()) + + def changed_set(base: str | None) -> list[str]: - """Branch diff vs merge-base + staged + unstaged + untracked, existing-or-tracked only.""" + """Per-commit branch paths plus staged, unstaged, and untracked paths. + + Per-commit paths keep add-then-delete changes visible. Deleted paths stay in + the set so removing every file matched by a custody or security gate still + selects that gate in PR and merge-group verification. + """ + paths: set[str] = set() + merge_base = resolve_merge_base(base) + if merge_base: + paths.update(path for _status, path in committed_path_changes(merge_base)) + paths.update(endpoint_changed_set(base)) + return sorted(p for p in paths if p) + + +def endpoint_changed_set(base: str | None) -> list[str]: + """Return only paths visible in the final branch diff or local checkout. + + Historical-only paths are intentionally excluded from display because a + transient private filename must not be copied into public CI logs. + """ paths: set[str] = set() merge_base = resolve_merge_base(base) if merge_base: - paths.update(git("diff", "--name-only", merge_base, "HEAD").splitlines()) - paths.update(git("diff", "--name-only").splitlines()) - paths.update(git("diff", "--name-only", "--cached").splitlines()) - paths.update(git("ls-files", "--others", "--exclude-standard").splitlines()) - tracked = set(git("ls-files").splitlines()) - return sorted(p for p in paths if p and ((ROOT / p).exists() or p in tracked)) + paths.update(git_paths("diff", "--name-only", "--no-renames", "-z", merge_base, "HEAD")) + paths.update(git_paths("diff", "--name-only", "--no-renames", "-z")) + paths.update(git_paths("diff", "--name-only", "--no-renames", "-z", "--cached")) + paths.update(git_paths("ls-files", "--others", "--exclude-standard", "-z")) + return sorted(p for p in paths if p) def gate_paths(gate_id: str, gate: dict, file_sets: dict) -> list[str]: @@ -246,7 +391,7 @@ def deploy_hits(registry: dict, changed: list[str]) -> list[str]: def expand_file_set(registry: dict, name: str) -> list[str]: spec = (registry.get("file_sets") or {})[name] - tracked = git("ls-files").splitlines() + tracked = git_paths("ls-files", "-z") excluded = {e.get("path") if isinstance(e, dict) else e for e in spec.get("exclude") or []} files: list[str] = [] for pattern in spec.get("include") or []: @@ -652,6 +797,20 @@ def cmd_changed( file=sys.stderr, ) return 1 + if private_history_leak(base): + print( + "private-history: a committed private namespace entry is absent from HEAD; " + "refusing to expose or certify transient private content.", + file=sys.stderr, + ) + return 1 + if transient_custody_reversion(base): + print( + "custody-history: a public custody path has an unvalidated intermediate version; " + "refusing to certify an unvalidated intermediate version.", + file=sys.stderr, + ) + return 1 changed = changed_set(base) if not changed: if require_base: @@ -663,9 +822,13 @@ def cmd_changed( return 1 print("No changes vs the base and no local modifications — nothing to verify.") return 0 - print(f"Changed paths ({len(changed)}):") - for p in changed: + display_paths = endpoint_changed_set(base) + hidden_history_count = len(set(changed) - set(display_paths)) + print(f"Changed paths ({len(display_paths)}):") + for p in display_paths: print(f" {p}") + if hidden_history_count: + print(f"Historical-only paths ({hidden_history_count}): [redacted; retained internally for gate selection]") if require_base and not integration and deploy_hits(registry, changed): if os.environ.get("LIMEN_VERIFY_NO_DEPLOY_ESCALATION") == "1":