From f23a8f0360ae1fef3e89a9e87d8d450ad0cda3f5 Mon Sep 17 00:00:00 2001 From: yuqei Date: Sun, 27 Sep 2026 17:25:37 +0800 Subject: [PATCH] fix(recovery): bind external evidence reads --- .../plan-first-release-snapshot-restore.py | 153 +++++++++++++++++- ..._first_release_snapshot_restore_planner.py | 35 ++++ 2 files changed, 183 insertions(+), 5 deletions(-) diff --git a/yuqei-infra/deploy/plan-first-release-snapshot-restore.py b/yuqei-infra/deploy/plan-first-release-snapshot-restore.py index 00b8b8e1..99f193f2 100644 --- a/yuqei-infra/deploy/plan-first-release-snapshot-restore.py +++ b/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( diff --git a/yuqei-infra/tests/test_first_release_snapshot_restore_planner.py b/yuqei-infra/tests/test_first_release_snapshot_restore_planner.py index da7d9746..0b77f21e 100644 --- a/yuqei-infra/tests/test_first_release_snapshot_restore_planner.py +++ b/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: