diff --git a/docs/leo-reproducible-identity.md b/docs/leo-reproducible-identity.md index ff49a67..74b31f4 100644 --- a/docs/leo-reproducible-identity.md +++ b/docs/leo-reproducible-identity.md @@ -170,6 +170,12 @@ python -m scripts.leo_identity_profile compile \ All commands require a clean Git worktree because a commit plus an unretained dirty source tree is not reproducible. +Child processes execute the real interpreter and temporary-profile paths only +in process-local memory. Retained receipts persist a repo-relative +`logical_command`, the stable `` and `` placeholders, +and an explicit `profile_role`; they never serialize the checkout, interpreter, +or temporary directory path. + ## State inventory and transitions | State | Meaning | Next valid transition | diff --git a/scripts/run_leo_identity_reconstruction_canary.py b/scripts/run_leo_identity_reconstruction_canary.py index 2850804..907f1fe 100644 --- a/scripts/run_leo_identity_reconstruction_canary.py +++ b/scripts/run_leo_identity_reconstruction_canary.py @@ -13,7 +13,7 @@ import sys import tempfile import time from datetime import UTC, datetime -from pathlib import Path +from pathlib import Path, PurePosixPath, PureWindowsPath from typing import Any if __package__ in {None, ""}: @@ -24,6 +24,7 @@ from scripts import leo_identity_manifest as identity from scripts import leo_identity_profile as profile_runtime SCHEMA = "livingip.leoIdentityReconstructionReceipt.v1" +PRIVATE_PATH_PREFIXES = tuple("/" + value for value in ("Users/", "private/tmp/", "var/folders/")) class CanaryError(RuntimeError): @@ -35,6 +36,25 @@ def _require(condition: bool, message: str) -> None: raise CanaryError(message) +def _receipt_strings(value: object) -> list[str]: + if isinstance(value, dict): + keys = [key for key in value if isinstance(key, str)] + return keys + [text for item in value.values() for text in _receipt_strings(item)] + if isinstance(value, list): + return [text for item in value for text in _receipt_strings(item)] + return [value] if isinstance(value, str) else [] + + +def _validate_portable_receipt(value: object) -> None: + for text in _receipt_strings(value): + if ( + any(prefix in text for prefix in PRIVATE_PATH_PREFIXES) + or PurePosixPath(text).is_absolute() + or PureWindowsPath(text).is_absolute() + ): + raise CanaryError("receipt contains a private or absolute filesystem path") + + def _process_group_absent(process_group: int) -> bool: try: os.killpg(process_group, 0) @@ -72,15 +92,55 @@ def _cleanup_remaining_process_group(process_group: int) -> str | None: return "SIGKILL" +def _logical_repo_path(path: Path, *, deployment_root: Path, placeholder: str) -> str: + """Return a stable repo-relative path or a non-identifying placeholder.""" + + try: + relative = path.resolve().relative_to(deployment_root.resolve()) + except (OSError, ValueError): + return placeholder + value = relative.as_posix() + return value if value != "." else "." + + +def _logical_query_command( + *, + deployment_root: Path, + source_root: Path, + hermes_root: Path, +) -> list[str]: + return [ + "", + "-m", + "scripts.leo_identity_profile", + "query", + "--profile", + "", + "--source-root", + _logical_repo_path(source_root, deployment_root=deployment_root, placeholder=""), + "--hermes-root", + _logical_repo_path(hermes_root, deployment_root=deployment_root, placeholder=""), + "--deployment-root", + ".", + "--query", + profile_runtime.FIXED_QUERY, + ] + + def _run_query_child( *, deployment_root: Path, profile: Path, source_root: Path, hermes_root: Path, + profile_role: str, timeout_seconds: float = 60, ) -> dict[str, Any]: - command = [ + deployment_root = deployment_root.resolve() + profile = profile.resolve() + source_root = source_root.resolve() + hermes_root = hermes_root.resolve() + actual_argv = [ sys.executable, "-m", "scripts.leo_identity_profile", @@ -97,7 +157,7 @@ def _run_query_child( profile_runtime.FIXED_QUERY, ] process = subprocess.Popen( - command, + actual_argv, cwd=deployment_root, env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}, text=True, @@ -143,7 +203,14 @@ def _run_query_child( except json.JSONDecodeError as exc: raise CanaryError("identity child returned invalid JSON") from exc return { - "command": command, + "logical_command": _logical_query_command( + deployment_root=deployment_root, + source_root=source_root, + hermes_root=hermes_root, + ), + "command_serialization": "logical_redacted_v1", + "profile_role": profile_role, + "actual_argv_persisted": False, "launcher_pid": process.pid, "returncode": process.returncode, "timed_out": timed_out, @@ -181,6 +248,7 @@ def _drift_attempt( profile=drift_profile, source_root=source_root, hermes_root=hermes_root, + profile_role="drift_negative", ) child["fail_closed"] = bool( child["returncode"] == 3 @@ -250,6 +318,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]: profile=compiled_profile, source_root=args.source_root, hermes_root=args.hermes_root, + profile_role="first_start", ) report["first_start"] = first _require(_query_passed(first), "first disposable identity process did not answer successfully") @@ -271,6 +340,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]: profile=compiled_profile, source_root=args.source_root, hermes_root=args.hermes_root, + profile_role="pre_restart_drift", ) report["pre_restart_drift"] = rejected report["drift_target"] = "compiled SOUL.md generated view" @@ -315,6 +385,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]: profile=restarted_profile, source_root=args.source_root, hermes_root=args.hermes_root, + profile_role="restart", ) report["restart"] = second _require(_query_passed(second), "restarted disposable identity process did not answer successfully") @@ -370,7 +441,13 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]: exit_code = 0 except (CanaryError, identity.IdentityManifestError, OSError, subprocess.SubprocessError) as exc: report["status"] = "fail" - report["errors"].append({"type": type(exc).__name__, "message": str(exc)}) + report["errors"].append( + { + "type": type(exc).__name__, + "message": "details_redacted", + "detail_sha256": behavior.sha256_bytes(str(exc).encode("utf-8")), + } + ) finally: shutil.rmtree(temporary_root, ignore_errors=True) report["cleanup"] = { @@ -402,6 +479,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]: else "Local first-process and pre-restart drift-refusal component only; no successful restart proof, " "GatewayRunner, hosted model, production database, VPS, or T3 claim." ) + _validate_portable_receipt(report) stable = {key: value for key, value in report.items() if key != "receipt_sha256"} report["receipt_sha256"] = behavior.canonical_sha256(stable) identity.write_json(output, report) diff --git a/tests/test_leo_identity_manifest.py b/tests/test_leo_identity_manifest.py index 35dd4d6..e421ffb 100644 --- a/tests/test_leo_identity_manifest.py +++ b/tests/test_leo_identity_manifest.py @@ -6,7 +6,7 @@ import json import os import subprocess import time -from pathlib import Path +from pathlib import Path, PureWindowsPath import pytest @@ -16,6 +16,10 @@ from scripts import leo_identity_profile as profile_runtime from scripts import run_leo_identity_reconstruction_canary as canary ROOT = Path(__file__).resolve().parents[1] +COMMITTED_RECEIPTS = ( + ROOT / "docs/reports/leo-working-state-20260709/leo-reproducible-identity-t2-current.json", + ROOT / "docs/reports/leo-working-state-20260709/leo-reproducible-identity-drift-negative-current.json", +) def _write(path: Path, value: str) -> None: @@ -27,6 +31,21 @@ def _write_json(path: Path, value: dict) -> None: _write(path, json.dumps(value, indent=2, sort_keys=True) + "\n") +def _assert_path_portable_receipt(value: object) -> None: + canary._validate_portable_receipt(value) + + +def _assert_logical_child_execution(child: dict, *, profile_role: str) -> None: + logical_command = child["logical_command"] + assert "command" not in child + assert child["profile_role"] == profile_role + assert child["command_serialization"] == "logical_redacted_v1" + assert child["actual_argv_persisted"] is False + assert logical_command[0] == "" + assert logical_command[logical_command.index("--profile") + 1] == "" + _assert_path_portable_receipt(child) + + def _fixture(tmp_path: Path) -> dict[str, Path | dict]: repo = tmp_path / "repo" profile = repo / "profile-template" @@ -224,6 +243,50 @@ def _canary_args(fixture: dict[str, Path | dict], output: Path, *, inject_drift: ) +@pytest.mark.parametrize( + "private_path", + [ + "/" + "Users/operator/checkout/profile", + "/" + "private/tmp/lane/.venv/bin/python", + "/" + "var/folders/cache/temporary-profile", + "/" + "tmp/temporary-profile", + "/" + "opt/checkout/interpreter", + "/" + "workspace/checkout/temporary-profile", + str(PureWindowsPath("C:/operator/checkout/temporary-profile")), + "\\\\" + "server\\share\\checkout\\temporary-profile", + ], +) +def test_receipt_path_guard_recursively_rejects_private_and_absolute_paths(private_path: str) -> None: + with pytest.raises(canary.CanaryError, match="private or absolute filesystem path"): + _assert_path_portable_receipt({"outer": [{"logical_command": ["", private_path]}]}) + + +def test_receipt_path_guard_rejects_an_absolute_path_in_a_nested_key() -> None: + private_key = "/" + "tmp/checkout/temporary-profile" + with pytest.raises(canary.CanaryError, match="private or absolute filesystem path"): + _assert_path_portable_receipt({"outer": [{private_key: ""}]}) + + +def test_logical_command_uses_placeholders_for_paths_outside_the_deployment_root(tmp_path: Path) -> None: + logical_command = canary._logical_query_command( + deployment_root=tmp_path / "checkout", + source_root=tmp_path / "outside-source", + hermes_root=tmp_path / "outside-hermes", + ) + + assert logical_command[logical_command.index("--source-root") + 1] == "" + assert logical_command[logical_command.index("--hermes-root") + 1] == "" + _assert_path_portable_receipt(logical_command) + + +@pytest.mark.parametrize("receipt_path", COMMITTED_RECEIPTS, ids=lambda path: path.name) +def test_committed_identity_receipts_are_path_portable_and_self_hashed(receipt_path: Path) -> None: + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + _assert_path_portable_receipt(receipt) + stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"} + assert behavior.canonical_sha256(stable) == receipt["receipt_sha256"] + + def test_manifest_compiles_generated_views_and_reproduces_identity_across_queries(tmp_path: Path) -> None: fixture = _fixture(tmp_path) manifest = fixture["manifest"] @@ -493,7 +556,8 @@ def test_fully_rehashed_model_claim_must_match_observed_runtime(tmp_path: Path) def test_restart_canary_uses_distinct_processes_and_cleans_up(tmp_path: Path) -> None: fixture = _fixture(tmp_path) - report, exit_code = canary.run_canary(_canary_args(fixture, tmp_path / "restart-receipt.json")) + output = tmp_path / "restart-receipt.json" + report, exit_code = canary.run_canary(_canary_args(fixture, output)) assert exit_code == 0 assert report["status"] == "pass" @@ -501,9 +565,10 @@ def test_restart_canary_uses_distinct_processes_and_cleans_up(tmp_path: Path) -> assert report["gates"]["restart_profile_recompiled_from_manifest"] is True assert report["restart_compile"]["session_directories_started_empty"] is True assert report["drift_compile"]["session_directories_started_empty"] is True - first_profile = report["first_start"]["command"][report["first_start"]["command"].index("--profile") + 1] - restart_profile = report["restart"]["command"][report["restart"]["command"].index("--profile") + 1] - assert first_profile != restart_profile + _assert_logical_child_execution(report["first_start"], profile_role="first_start") + _assert_logical_child_execution(report["restart"], profile_role="restart") + _assert_logical_child_execution(report["drift_negative"], profile_role="drift_negative") + _assert_path_portable_receipt(report) assert report["first_start"]["launcher_pid"] != report["restart"]["launcher_pid"] assert report["first_start"]["process_group_absent_after_wait"] is True assert report["restart"]["process_group_absent_after_wait"] is True @@ -513,22 +578,28 @@ def test_restart_canary_uses_distinct_processes_and_cleans_up(tmp_path: Path) -> assert report["current_tier"] == "T2_runtime" assert report["cleanup"]["temporary_root_removed"] is True assert behavior.git_state(fixture["repo"])["worktree_clean"] is True + assert json.loads(output.read_text(encoding="utf-8")) == report def test_restart_canary_rejects_drift_before_second_start(tmp_path: Path) -> None: fixture = _fixture(tmp_path) - report, exit_code = canary.run_canary(_canary_args(fixture, tmp_path / "drift-receipt.json", inject_drift=True)) + output = tmp_path / "drift-receipt.json" + report, exit_code = canary.run_canary(_canary_args(fixture, output, inject_drift=True)) assert exit_code == 0 assert report["status"] == "pass" assert report["gates"]["drift_detected_before_restart"] is True assert report["second_start_attempted"] is False assert "restart" not in report + _assert_logical_child_execution(report["first_start"], profile_role="first_start") + _assert_logical_child_execution(report["pre_restart_drift"], profile_role="pre_restart_drift") + _assert_path_portable_receipt(report) assert report["pre_restart_drift"]["process_group_absent_after_wait"] is True assert all(report["gates"].values()) assert report["goal_tier_satisfied"] is False assert report["current_tier"] == "T1_model" assert report["cleanup"]["temporary_root_removed"] is True + assert json.loads(output.read_text(encoding="utf-8")) == report def test_query_child_timeout_terminates_process_group(tmp_path: Path) -> None: @@ -550,6 +621,7 @@ time.sleep(60) profile=fixture["compiled"], source_root=fixture["repo"], hermes_root=fixture["hermes"], + profile_role="timeout_test", timeout_seconds=0.5, ) @@ -594,6 +666,7 @@ print(json.dumps({query_payload!r})) profile=fixture["compiled"], source_root=fixture["repo"], hermes_root=fixture["hermes"], + profile_role="orphan_cleanup_test", timeout_seconds=5, )