fix(recovery): bind external evidence reads

yuqei-infra/deploy/plan-first-release-snapshot-restore.py

@@ -183,14 +183,26 @@ def _absolute(path: Path) -> Path:
     return Path(os.path.abspath(os.fspath(path)))
 
 
+def _metadata_is_reparse(metadata: os.stat_result) -> bool:
+    reparse_attribute = int(getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400))
+    return bool(
+        stat.S_ISLNK(metadata.st_mode)
+        or int(getattr(metadata, "st_file_attributes", 0)) & reparse_attribute
+        or int(getattr(metadata, "st_reparse_tag", 0)) != 0
+    )
+
+
 def _reject_symlink_components(path: Path, *, label: str) -> None:
     absolute = _absolute(path)
     current = Path(absolute.anchor)
     for part in absolute.parts[1:]:
         current /= part
         try:
-            if current.exists() and current.is_symlink():
+            metadata = os.lstat(current)
+            if _metadata_is_reparse(metadata):
                 raise SnapshotRestorePlanError(f"{label} contains a symbolic link")
+        except FileNotFoundError:
+            continue
         except OSError as exc:
             raise SnapshotRestorePlanError(f"cannot inspect {label} path") from exc
 
@@ -241,6 +253,135 @@ def load_json(path: Path, *, label: str) -> dict[str, Any]:
     return payload
 
 
+def _metadata_state(metadata: os.stat_result) -> tuple[int, ...]:
+    """Return metadata that must stay stable for one already-open file view."""
+
+    return (
+        metadata.st_dev,
+        metadata.st_ino,
+        metadata.st_mode,
+        metadata.st_nlink,
+        metadata.st_size,
+        metadata.st_mtime_ns,
+        metadata.st_ctime_ns,
+        int(getattr(metadata, "st_file_attributes", 0)),
+        int(getattr(metadata, "st_reparse_tag", 0)),
+    )
+
+
+def _assert_external_file_metadata(
+    metadata: os.stat_result,
+    *,
+    label: str,
+    max_bytes: int,
+) -> None:
+    if (
+        not stat.S_ISREG(metadata.st_mode)
+        or metadata.st_nlink != 1
+        or metadata.st_ino == 0
+        or _metadata_is_reparse(metadata)
+    ):
+        raise SnapshotRestorePlanError(
+            f"{label} must be a single-link non-reparse regular file"
+        )
+    if metadata.st_size <= 0 or metadata.st_size > max_bytes:
+        raise SnapshotRestorePlanError(f"{label} has an invalid size")
+
+
+def _strict_json_object_from_bytes(raw: bytes, *, label: str) -> dict[str, Any]:
+    try:
+        payload = json.loads(
+            raw.decode("utf-8-sig"),
+            object_pairs_hook=_reject_duplicate_pairs,
+            parse_constant=_reject_constant,
+        )
+    except SnapshotRestorePlanError:
+        raise
+    except (UnicodeError, ValueError) as exc:
+        raise SnapshotRestorePlanError(f"{label} is not strict UTF-8 JSON") from exc
+    if not isinstance(payload, dict):
+        raise SnapshotRestorePlanError(f"{label} must be a JSON object")
+    return payload
+
+
+def _read_stable_external_json(
+    path: Path,
+    *,
+    label: str,
+    expected_sha256: str,
+    max_bytes: int = MAX_JSON_BYTES,
+) -> tuple[dict[str, Any], Path]:
+    """Read, hash, parse, and revalidate an external record via one descriptor."""
+
+    lexical = _absolute(path)
+    _reject_symlink_components(lexical, label=label)
+    try:
+        path_before = os.lstat(lexical)
+    except OSError as exc:
+        raise SnapshotRestorePlanError(f"{label} is missing") from exc
+    _assert_external_file_metadata(path_before, label=label, max_bytes=max_bytes)
+
+    flags = (
+        os.O_RDONLY
+        | getattr(os, "O_BINARY", 0)
+        | getattr(os, "O_NOFOLLOW", 0)
+    )
+    try:
+        descriptor = os.open(lexical, flags)
+    except OSError as exc:
+        raise SnapshotRestorePlanError(f"{label} cannot be opened safely") from exc
+    try:
+        opened_before = os.fstat(descriptor)
+        _assert_external_file_metadata(opened_before, label=label, max_bytes=max_bytes)
+        if (
+            not os.path.samestat(path_before, opened_before)
+            or path_before.st_size != opened_before.st_size
+            or path_before.st_mtime_ns != opened_before.st_mtime_ns
+        ):
+            raise SnapshotRestorePlanError(f"{label} path identity changed before reading")
+
+        chunks: list[bytes] = []
+        total = 0
+        while True:
+            try:
+                chunk = os.read(descriptor, min(1024 * 1024, max_bytes + 1 - total))
+            except OSError as exc:
+                raise SnapshotRestorePlanError(f"{label} cannot be read safely") from exc
+            if not chunk:
+                break
+            chunks.append(chunk)
+            total += len(chunk)
+            if total > max_bytes:
+                raise SnapshotRestorePlanError(f"{label} has an invalid size")
+        raw = b"".join(chunks)
+        if len(raw) != opened_before.st_size:
+            raise SnapshotRestorePlanError(f"{label} size changed while reading")
+        if hashlib.sha256(raw).hexdigest() != expected_sha256:
+            raise SnapshotRestorePlanError(f"{label} SHA-256 mismatch")
+        payload = _strict_json_object_from_bytes(raw, label=label)
+
+        opened_after = os.fstat(descriptor)
+        _assert_external_file_metadata(opened_after, label=label, max_bytes=max_bytes)
+        if _metadata_state(opened_after) != _metadata_state(opened_before):
+            raise SnapshotRestorePlanError(f"{label} metadata changed while reading")
+        try:
+            path_after = os.lstat(lexical)
+        except OSError as exc:
+            raise SnapshotRestorePlanError(f"{label} path changed while reading") from exc
+        _assert_external_file_metadata(path_after, label=label, max_bytes=max_bytes)
+        if (
+            _metadata_state(path_after) != _metadata_state(path_before)
+            or not os.path.samestat(path_after, opened_after)
+            or path_after.st_size != opened_after.st_size
+            or path_after.st_mtime_ns != opened_after.st_mtime_ns
+        ):
+            raise SnapshotRestorePlanError(f"{label} path identity changed while reading")
+        _reject_symlink_components(lexical, label=label)
+        return payload, lexical
+    finally:
+        os.close(descriptor)
+
+
 def _exact(value: object, fields: set[str], *, label: str) -> dict[str, Any]:
     if not isinstance(value, dict) or set(value) != fields:
         raise SnapshotRestorePlanError(f"{label} fields are missing or unknown")
@@ -317,10 +458,12 @@ def _external_record(value: object, *, label: str) -> tuple[dict[str, Any], str,
     if not isinstance(raw_path, str) or not Path(raw_path).is_absolute():
         raise SnapshotRestorePlanError(f"{label} path must be absolute")
     expected = _sha(record.get("sha256"), label=f"{label} external SHA-256")
-    path = _regular_file(Path(raw_path), label=label)
-    if file_sha256(path) != expected:
-        raise SnapshotRestorePlanError(f"{label} SHA-256 mismatch")
-    return load_json(path, label=label), expected, path
+    payload, path = _read_stable_external_json(
+        Path(raw_path),
+        label=label,
+        expected_sha256=expected,
+    )
+    return payload, expected, path
 
 
 def _validate_source_closure(

yuqei-infra/tests/test_first_release_snapshot_restore_planner.py

@@ -5,6 +5,7 @@ import hashlib
 import importlib.util
 import inspect
 import json
+import os
 import sys
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -792,6 +793,40 @@ def test_formal_backup_rejects_drifted_source_artifacts_manifest(tmp_path: Path)
         _direct_backup_validation(case)
 
 
+@pytest.mark.parametrize("target_kind", ("formal_v3", "source_artifacts_v2"))
+def test_formal_backup_rejects_atomic_external_manifest_replacement(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    target_kind: str,
+) -> None:
+    case = _case(tmp_path)
+    target = (
+        case["backup_path"]
+        if target_kind == "formal_v3"
+        else case["source_backup_path"]
+    )
+    target_absolute = Path(os.path.abspath(target))
+    attacker = target.with_name(f"{target.stem}.replacement.json")
+    attacker.write_bytes(target.read_bytes() + b"\n")
+    real_open = planner.os.open
+    replaced = False
+
+    def replace_after_path_check(path: object, flags: int, *args: object) -> int:
+        nonlocal replaced
+        if not replaced and Path(os.path.abspath(os.fspath(path))) == target_absolute:
+            os.replace(attacker, target)
+            replaced = True
+        return real_open(path, flags, *args)
+
+    monkeypatch.setattr(planner.os, "open", replace_after_path_check)
+    with pytest.raises(
+        planner.SnapshotRestorePlanError,
+        match="path identity changed before reading",
+    ):
+        _direct_backup_validation(case)
+    assert replaced is True
+
+
 def test_formal_backup_rejects_application_runtime_identity_outside_candidate(
     tmp_path: Path,
 ) -> None: