diff --git a/loopx/control_plane/coordination/cold_source_backup.py b/loopx/control_plane/coordination/cold_source_backup.py index 2d0cb1613d..4bbd3390e6 100644 --- a/loopx/control_plane/coordination/cold_source_backup.py +++ b/loopx/control_plane/coordination/cold_source_backup.py @@ -14,6 +14,7 @@ import json import os from pathlib import Path, PurePosixPath +import posixpath import shutil import sqlite3 import stat @@ -85,9 +86,13 @@ def _open_regular(path: Path) -> tuple[BinaryIO, os.stat_result]: def _same_file(before: os.stat_result, after: os.stat_result) -> bool: + identity_fields: tuple[str, ...] = ("st_dev", "st_ino", "st_size", "st_mtime_ns") + # Only a complete Windows volume/file identity can tolerate ctime skew. + if os.name != "nt" or not all(info.st_dev and info.st_ino for info in (before, after)): + identity_fields += ("st_ctime_ns",) return all( getattr(before, field) == getattr(after, field) - for field in ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns") + for field in identity_fields ) @@ -114,7 +119,7 @@ def _safe_link(name: str, target: str, *, hardlink: bool) -> str: raise ColdSourceBackupError("cold_import_backup_link_unsafe") if hardlink: return _member_name(target) - resolved = os.path.normpath(str(PurePosixPath(name).parent / target)) + resolved = posixpath.normpath(str(PurePosixPath(name).parent / target)) if resolved == ".." or resolved.startswith("../"): raise ColdSourceBackupError("cold_import_backup_link_unsafe") return _member_name(resolved) @@ -190,8 +195,8 @@ def _matches(name: str, prefixes: list[str] | None) -> bool: def _scan_archive( - archive: BinaryIO, *, selected: list[str] | None, workspace: Path -) -> tuple[dict[str, Any], list[dict[str, Any]], int, int, int]: + archive: BinaryIO, *, execution: dict[str, Any], selected: list[str] | None, workspace: Path +) -> tuple[dict[str, Any], list[dict[str, Any]], int, int, int, str, int]: workspace.mkdir(mode=0o700, parents=True, exist_ok=True) workspace_info = workspace.lstat() if not stat.S_ISDIR(workspace_info.st_mode): @@ -205,9 +210,20 @@ def _scan_archive( archive.seek(0) # One disposable transaction avoids per-prefix durable writes. # This scratch index is not a Goal authority or recovery store. - with closing(sqlite3.connect(database_path)) as database: + with ( + closing(sqlite3.connect(database_path)) as database, + (Path(temporary) / "archive.tar.gz").open("w+b") as snapshot, + ): + # Hash and scan the same private bytes even if the source is overwritten and restored. + shutil.copyfileobj(archive, snapshot, _CHUNK) + original_sha, archive_size = _sha(snapshot) + if original_sha != execution.get("archive_sha256"): + raise ColdSourceBackupError("cold_import_backup_archive_changed") + declared_size = execution.get("archive_size_bytes") + if declared_size is not None and declared_size != archive_size: + raise ColdSourceBackupError("cold_import_backup_archive_changed") database.execute("CREATE TABLE members (name BLOB PRIMARY KEY, marker BLOB NOT NULL) WITHOUT ROWID") - with tarfile.open(fileobj=archive, mode="r|gz") as tar: + with tarfile.open(fileobj=snapshot, mode="r|gz") as tar: # r|gz streams bytes but still caches every TarInfo on # supported Python versions. No random member lookup is # needed: hardlink coverage already lives in the disk DB. @@ -282,7 +298,7 @@ def _scan_archive( raise if embedded is None: raise ColdSourceBackupError("cold_import_backup_manifest_missing") - return embedded, members, member_count, logical_bytes, workspace_bytes + return embedded, members, member_count, logical_bytes, workspace_bytes, original_sha, archive_size def read_cold_source_backup( @@ -319,15 +335,10 @@ def read_cold_source_backup( archive_path = Path(str(execution.get("archive_path") or "")).expanduser().absolute() archive, before = _open_regular(archive_path) with archive: - original_sha, archive_size = _sha(archive) - if original_sha != execution.get("archive_sha256"): - raise ColdSourceBackupError("cold_import_backup_archive_changed") - declared_size = execution.get("archive_size_bytes") - if declared_size is not None and declared_size != archive_size: - raise ColdSourceBackupError("cold_import_backup_archive_changed") scan_workspace = (workspace or manifest_path.parent / ".cold-import-work").expanduser().absolute() - embedded, members, member_count, logical_bytes, workspace_bytes = _scan_archive( - archive, selected=selected, workspace=scan_workspace + (embedded, members, member_count, logical_bytes, workspace_bytes, + original_sha, archive_size) = _scan_archive( + archive, execution=execution, selected=selected, workspace=scan_workspace ) after = os.fstat(archive.fileno()) if not _same_file(before, after): diff --git a/loopx/materials.py b/loopx/materials.py index e4b3196f4e..dc81accc40 100644 --- a/loopx/materials.py +++ b/loopx/materials.py @@ -222,7 +222,12 @@ def read_bounded_material_file(path: Path, max_bytes: int) -> tuple[Path, bytes, after = os.fstat(handle.fileno()) resolved = path.resolve() current = resolved.stat() - identity_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns") + identity_fields: tuple[str, ...] = ("st_dev", "st_ino", "st_size", "st_mtime_ns") + # Tolerate Windows ctime skew only with a volume and file index in every observation. + if os.name != "nt" or not all( + observed.st_dev and observed.st_ino for observed in (before, after, current) + ): + identity_fields += ("st_ctime_ns",) identity = tuple(getattr(before, field) for field in identity_fields) if len(content) != before.st_size or any( tuple(getattr(observed, field) for field in identity_fields) != identity diff --git a/tests/capabilities/test_decision_context_profile.py b/tests/capabilities/test_decision_context_profile.py index 7a8c9ac8d0..1964d1211a 100644 --- a/tests/capabilities/test_decision_context_profile.py +++ b/tests/capabilities/test_decision_context_profile.py @@ -411,6 +411,78 @@ def stale_preflight(path, *args, **kwargs): assert grew +@pytest.mark.parametrize("platform", ["nt", "posix"]) +@pytest.mark.parametrize("descriptor_change", [False, True]) +def test_local_file_ctime_skew(tmp_path, monkeypatch, platform, descriptor_change): + import os + from types import SimpleNamespace + + from loopx import materials + + authority = tmp_path / "authority.txt" + authority.write_text("stable source", encoding="utf-8") + profile = normalize_decision_context_profile(profile_payload(authority)) + provider = build_decision_source_provider(profile.provider_binding_map()["local-authority"]) + original_fstat = os.fstat + calls = 0 + + def descriptor_stat(fd): + nonlocal calls + observed = original_fstat(fd) + calls += 1 + fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")} + fields["st_ctime_ns"] = authority.stat().st_ctime_ns + 499_700 + if descriptor_change and calls == 2: + fields["st_ctime_ns"] += 1 + return SimpleNamespace(**fields) + + monkeypatch.setattr(materials, "os", SimpleNamespace(**{ + **vars(os), "name": platform, "fstat": descriptor_stat, + })) + args = dict(source=profile.sources[0], after_cursor=None, + before="2100-01-01T00:00:00+00:00", limit=8, + timeout_seconds=2, observed_at=OBSERVED_AT) + if platform == "posix": + with pytest.raises(ValueError, match="changed during bounded read"): + provider.scan(**args) + else: + scan = provider.scan(**args) + exact = provider.exact_read(source=profile.sources[0], item=scan.items[0], + timeout_seconds=2, observed_at=OBSERVED_AT) + assert scan.status == "completed" + assert exact.content == "stable source" + + +@pytest.mark.parametrize("change", ["mtime", "size"]) +def test_local_file_path_change_after_descriptor_snapshot(tmp_path, monkeypatch, change): + import os + + authority = tmp_path / "authority.txt" + authority.write_bytes(b"before") + profile = normalize_decision_context_profile(profile_payload(authority)) + provider = build_decision_source_provider(profile.provider_binding_map()["local-authority"]) + original_fstat = os.fstat + calls = 0 + + def change_after_snapshot(fd): + nonlocal calls + observed = original_fstat(fd) + calls += 1 + if calls == 2: + if change == "size": + authority.write_bytes(b"longer source") + os.utime(authority, ns=(observed.st_atime_ns, + observed.st_mtime_ns + (1_000_000_000 if change == "mtime" else 0))) + return observed + + monkeypatch.setattr(os, "fstat", change_after_snapshot) + with pytest.raises(ValueError, match="changed during bounded read"): + provider.scan(source=profile.sources[0], after_cursor=None, + before="2100-01-01T00:00:00+00:00", limit=8, + timeout_seconds=2, observed_at=OBSERVED_AT) + assert calls == 2 + + def test_local_file_growth_during_read_is_bounded(tmp_path, monkeypatch): import os @@ -460,6 +532,86 @@ def read(self, size=-1): assert read_sizes == [17] +@pytest.mark.parametrize("missing_identity", [None, "st_ino", "st_dev"]) +def test_local_file_same_metadata_replacement_has_no_verified_read(tmp_path, monkeypatch, missing_identity): + import os + import subprocess + from types import SimpleNamespace + + from loopx import materials + + original_dir, replacement_dir = tmp_path / "original", tmp_path / "replacement" + original_dir.mkdir() + replacement_dir.mkdir() + original = original_dir / "authority.txt" + original.write_bytes(b"original") + os.utime(original, (1_700_000_000, 1_700_000_000)) + before = original.stat() + replacement = replacement_dir / "authority.txt" + replacement.write_bytes(b"replaced") + os.utime(replacement, ns=(before.st_atime_ns, before.st_mtime_ns)) + assert replacement.stat().st_ino != before.st_ino + assert replacement.stat().st_size == before.st_size + assert replacement.stat().st_mtime_ns == before.st_mtime_ns + alias = tmp_path / "alias" + + def link_to(target): + if os.name == "nt": + created = subprocess.run( + ["cmd.exe", "/d", "/c", "mklink", "/J", str(alias), str(target)], + capture_output=True, + ) + assert created.returncode == 0, created.stderr or created.stdout + else: + alias.symlink_to(target, target_is_directory=True) + + link_to(original_dir) + authority = alias / "authority.txt" + profile = normalize_decision_context_profile(profile_payload(authority)) + provider = build_decision_source_provider(profile.provider_binding_map()["local-authority"]) + original_fstat, original_stat = os.fstat, Path.stat + observations = 0 + + def windows_stat(observed): + fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")} + fields["st_ctime_ns"] = before.st_ctime_ns + (observed.st_ino != before.st_ino) + if missing_identity: + fields[missing_identity] = 0 + if missing_identity == "st_dev": + # File indexes may collide across volumes; inject only the OS metadata. + fields["st_ino"] = before.st_ino + return SimpleNamespace(**fields) + + def descriptor_stat(fd): + nonlocal observations + observed = original_fstat(fd) + observations += 1 + if observations == 2: + # Retarget the directory while the original file is open on Windows. + if os.name == "nt": + alias.rmdir() + else: + alias.unlink() + link_to(replacement_dir) + assert authority.read_bytes() == b"replaced" + return windows_stat(observed) + + def path_stat(path, *args, **kwargs): + observed = original_stat(path, *args, **kwargs) + return windows_stat(observed) if path in (authority, original, replacement) else observed + + monkeypatch.setattr(materials, "os", SimpleNamespace(**{ + **vars(os), "name": "nt", "fstat": descriptor_stat, + })) + monkeypatch.setattr(Path, "stat", path_stat) + with pytest.raises(ValueError, match="changed during bounded read"): + provider.scan(source=profile.sources[0], after_cursor=None, + before="2100-01-01T00:00:00+00:00", limit=8, + timeout_seconds=2, observed_at=OBSERVED_AT) + assert observations == 2 + assert authority.read_bytes() == b"replaced" + + def test_local_file_replacement_after_open_has_no_verified_read(tmp_path, monkeypatch): import os diff --git a/tests/control_plane/test_cold_source_backup_scope.py b/tests/control_plane/test_cold_source_backup_scope.py index d34f33cc17..61ddb94e4c 100644 --- a/tests/control_plane/test_cold_source_backup_scope.py +++ b/tests/control_plane/test_cold_source_backup_scope.py @@ -4,9 +4,11 @@ import hashlib import io import json +import os from pathlib import Path import sqlite3 import tarfile +from types import SimpleNamespace import pytest @@ -140,6 +142,66 @@ def test_scoping_does_not_hide_unsafe_unrelated_members_or_changed_artifact(tmp_ read_cold_source_backup(manifest, source_paths=[tmp_path / "runtime/goals/example"]) +@pytest.mark.parametrize("unavailable_identity", [None, "st_dev", "st_ino"]) +@pytest.mark.parametrize("platform", ["nt", "posix"]) +@pytest.mark.parametrize("field", ["st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns"]) +def test_archive_identity_change_policy(tmp_path, monkeypatch, platform, field, unavailable_identity): + from loopx.control_plane.coordination import cold_source_backup + + manifest = backup(tmp_path) + archive = tmp_path / "state.tar.gz" + archive_stat = archive.stat() + assert archive_stat.st_dev and archive_stat.st_ino + original_fstat = os.fstat + original_lstat = Path.lstat + observations = 0 + + def partial_identity(observed): + if unavailable_identity is None or (observed.st_dev, observed.st_ino) != ( + archive_stat.st_dev, archive_stat.st_ino + ): + return observed + # Unavailable volume/file indexes are injected only at the stat boundary. + fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")} + fields[unavailable_identity] = 0 + return SimpleNamespace(**fields) + + def path_stat(path, *args, **kwargs): + return partial_identity(original_lstat(path, *args, **kwargs)) + + def descriptor_stat(fd): + nonlocal observations + observed = original_fstat(fd) + if (observed.st_dev, observed.st_ino) != (archive_stat.st_dev, archive_stat.st_ino): + return observed + observations += 1 + observed = partial_identity(observed) + if observations != 2: + return observed + fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")} + fields[field] += 1 + return SimpleNamespace(**fields) + + monkeypatch.setattr(Path, "lstat", path_stat) + monkeypatch.setattr(cold_source_backup, "os", SimpleNamespace(**{ + **vars(os), "name": platform, "fstat": descriptor_stat, + })) + if platform == "nt" and field == "st_ctime_ns" and unavailable_identity is None: + witness = read_cold_source_backup(manifest) + assert witness["archive_sha256"] == hashlib.sha256(archive.read_bytes()).hexdigest() + assert witness["archive_size_bytes"] == archive_stat.st_size + assert witness["members"] == [ + {"archive_path": "runtime/goals/example/state.md", + "sha256": hashlib.sha256(b"retained source").hexdigest()}, + {"archive_path": "runtime/goals/example/task-leases/settled.json", + "sha256": hashlib.sha256(b"retained lease").hexdigest()}, + ] + else: + with pytest.raises(ColdSourceBackupError, match="cold_import_backup_changed_retry"): + read_cold_source_backup(manifest) + assert observations == 2 + + def test_streaming_witness_has_no_old_expanded_byte_ceiling(tmp_path): archive = tmp_path / "large-expanded.tar.gz" source = tmp_path / "source.bin" @@ -200,6 +262,8 @@ def unsafe_manifest(name: str, members: list[tarfile.TarInfo]) -> Path: link = tarfile.TarInfo("root/link") link.type, link.linkname = tarfile.SYMTYPE, "target" + assert read_cold_source_backup(unsafe_manifest("first-link", [link]), + source_paths=[tmp_path])["members"] == [] descendant = tarfile.TarInfo("root/link/file") descendant.size = 1 with pytest.raises(ValueError, match="member_unsafe"): diff --git a/tests/control_plane/test_cold_source_import_cli.py b/tests/control_plane/test_cold_source_import_cli.py index 5071d6bbd7..c49eaf5d30 100644 --- a/tests/control_plane/test_cold_source_import_cli.py +++ b/tests/control_plane/test_cold_source_import_cli.py @@ -6,6 +6,8 @@ """ from __future__ import annotations +from contextlib import contextmanager +import gzip import hashlib import io import json @@ -15,6 +17,7 @@ import subprocess import sys import tarfile +from types import SimpleNamespace import pytest @@ -94,6 +97,75 @@ def cli(action, *args, success=True, output_format="json", runtime_override=None return cli, state, backup, body, runtime, receiver, env +def test_prepare_rejects_archive_swap_and_restore_during_member_scan(tmp_path, monkeypatch, capsys): + from loopx.cli import main + from loopx.control_plane.coordination import cold_source_backup + + _, state, backup, _, runtime, _, _ = workspace(tmp_path, monkeypatch) + manifest_path = Path(backup["manifest_path"]) + manifest = json.loads(manifest_path.read_text()) + archive = Path(manifest["execution"]["archive_path"]) + original = archive.read_bytes() + source_a = state.read_bytes() + source_b = source_a.replace(b"Complete cold source requirement", b"Tampered cold source requirement") + assert source_a != source_b and len(source_a) == len(source_b) + expanded = gzip.decompress(original) + assert source_a in expanded + variant = gzip.compress(expanded.replace(source_a, source_b), mtime=0) + size = max(len(original), len(variant)) + original, variant = original.ljust(size, b"\0"), variant.ljust(size, b"\0") + archive.write_bytes(original) + manifest["execution"].update(archive_sha256=hashlib.sha256(original).hexdigest(), archive_size_bytes=size) + manifest_path.write_text(json.dumps(manifest)) + state.write_bytes(source_b) + os.utime(archive, (1_700_000_000, 1_700_000_000)) + before = archive.stat() + original_open, original_fstat = tarfile.open, os.fstat + scans = 0 + + def overwrite(data): + with archive.open("r+b") as handle: + handle.write(data) + os.utime(archive, ns=(before.st_atime_ns, before.st_mtime_ns)) + observed = archive.stat() + assert (observed.st_dev, observed.st_ino, observed.st_size, observed.st_mtime_ns) == ( + before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) + + @contextmanager + def swap_during_scan(*args, **kwargs): + nonlocal scans + scans += 1 + overwrite(variant) + try: + with original_open(*args, **kwargs) as tar: + yield tar + finally: + overwrite(original) + + def descriptor_stat(fd): + observed = original_fstat(fd) + if (observed.st_dev, observed.st_ino) != (before.st_dev, before.st_ino): + return observed + fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")} + fields["st_ctime_ns"] = before.st_ctime_ns + scans + return SimpleNamespace(**fields) + + monkeypatch.setattr(tarfile, "open", swap_during_scan) + monkeypatch.setattr(cold_source_backup, "os", SimpleNamespace(**{ + **vars(os), "name": "nt", "fstat": descriptor_stat, + })) + code = main(["--registry", str(tmp_path / "project/.loopx/registry.json"), + "--runtime-root", str(runtime), "--format", "json", "coordination-shadow", "prepare-import", + "--goal-id", "cold", "--operation-id", "cold-original", "--backup-manifest", str(manifest_path), + "--provider", "file", "--target-handoff-mode", "hard_lease"]) + result = json.loads(capsys.readouterr().out)["cold_import"] + assert scans == 1 and archive.read_bytes() == original + assert code == 1, result + assert result["reason_code"] == "cold_import_backup_source_missing_or_changed" + assert result.get("coordination_source_backup_verified") is not True + assert not list((runtime / "authority-transition").rglob("writer-fence.json")) + + def test_coordination_command_help_and_rejections_match_full_cli(tmp_path): # Independently compare the existing compatibility parser/handler with the # selected dispatcher, including diagnostics that deliberately fall back.