diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 7bfe92f200..7a309c8eab 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -967,6 +967,17 @@ jobs: tests/control_plane/test_document_io.py::test_ordinary_windows_long_path_publish_preserves_bytes_and_create_only tests/control_plane/test_todo_completion_validation.py::test_private_validation_store_is_owner_only_and_detects_tampering + - name: Qualify portable file-flag probes on native Windows + # read_native_trace_file, the recovery readback and checkpoint retention + # used to reference os.O_NOFOLLOW, os.O_DIRECTORY and os.getuid + # directly, so a host that does not define one raised AttributeError + # before the reader could refuse. Only a native Windows host takes that + # branch for real; the simulated cases cover it on POSIX. + run: >- + python -m pytest -q + tests/architecture/test_posix_flag_portability.py + tests/control_plane/test_posix_flag_fallback.py + - name: Check Windows collaboration lock sidecars run: python -m pytest -q --basetemp .t tests/test_collaboration_sidecar_scan.py diff --git a/loopx/control_plane/state_recovery_io.py b/loopx/control_plane/state_recovery_io.py index f0207b5225..ebe24345fc 100644 --- a/loopx/control_plane/state_recovery_io.py +++ b/loopx/control_plane/state_recovery_io.py @@ -105,14 +105,34 @@ def _audit_owners(request: dict[str, Any]) -> dict[str, Any]: @contextmanager def _regular_input(path: Path, limit: int) -> Iterator[BinaryIO]: + # O_NOFOLLOW and os.getuid are POSIX-only. A bare reference to either raises + # AttributeError on a host that does not define it, so this reader would + # surface an untyped crash instead of its own refusal -- and the readback is + # reachable without the platform gate verify_state_backup applies. Ask for + # what this host has, and refuse a symlink here when the open cannot. + getuid = getattr(os, "getuid", None) + if getuid is None: + raise StateRecoveryError("recovery_input_unavailable") try: - fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) except OSError as exc: raise StateRecoveryError("recovery_input_unavailable") from exc + try: + # Where the host has no no-follow flag the open cannot refuse a link, and + # inspecting the path first only moves the race to the open below. Bind the + # descriptor to the entry before anything is read: a link opens to its + # target, whose device and inode differ from the link's own. + opened, entry = os.fstat(fd), os.lstat(path) + _require(not stat.S_ISLNK(entry.st_mode) + and (entry.st_dev, entry.st_ino) == (opened.st_dev, opened.st_ino), + "recovery_input_unavailable") + except BaseException: + os.close(fd) + raise with os.fdopen(fd, "rb") as handle: info = os.fstat(handle.fileno()) _require(stat.S_ISREG(info.st_mode), "recovery_input_not_regular") - _require(info.st_uid == os.getuid() and not info.st_mode & 0o077, "recovery_input_not_private") + _require(info.st_uid == getuid() and not info.st_mode & 0o077, "recovery_input_not_private") _require(0 < info.st_size <= limit, "recovery_capacity_exceeded") yield handle diff --git a/loopx/control_plane/turn_driver/codex_input_trace.py b/loopx/control_plane/turn_driver/codex_input_trace.py index 4fd5ecfe5c..5f211f7541 100644 --- a/loopx/control_plane/turn_driver/codex_input_trace.py +++ b/loopx/control_plane/turn_driver/codex_input_trace.py @@ -13,10 +13,41 @@ from .decision_capture import capture_page_range as _capture_page_range +def _open_verified(path: Path) -> int: + """Open the entry at `path` without following a link, on any host. + + `O_NOFOLLOW` and `O_NONBLOCK` are POSIX-only attributes, so a bare reference + raises AttributeError on a host that does not define them -- here, before the + bounded-regular-file check below can refuse anything. + + Where the host has no no-follow flag, the open cannot refuse a link for us, + and checking the path first would only move the race: the entry can be + replaced between the check and the open. Bind the descriptor to the entry + instead, and do it before the caller reads one byte: a link opens to its + target, whose device and inode differ from the link's own, so a swapped or + linked entry is refused rather than followed. + """ + descriptor = os.open( + path, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0), + ) + try: + opened = os.fstat(descriptor) + entry = os.lstat(path) + if stat.S_ISLNK(entry.st_mode): + raise ValueError("native observation must not be a symlink") + if (entry.st_dev, entry.st_ino) != (opened.st_dev, opened.st_ino): + raise ValueError("native observation changed while it was opened") + except BaseException: + os.close(descriptor) + raise + return descriptor + + def read_native_trace_file(path: Path, *, tail: bool = False) -> bytes: from ..effect_runtime import MAX_REQUEST_BYTES - descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + descriptor = _open_verified(path) with os.fdopen(descriptor, "rb") as stream: metadata = os.fstat(stream.fileno()) if not stat.S_ISREG(metadata.st_mode) or (metadata.st_size > MAX_REQUEST_BYTES and not tail): diff --git a/loopx/hosts/checkpoint_retention.py b/loopx/hosts/checkpoint_retention.py index 7b12f3fe0d..610887d8e9 100644 --- a/loopx/hosts/checkpoint_retention.py +++ b/loopx/hosts/checkpoint_retention.py @@ -108,12 +108,25 @@ def retention_plan(bank: Path, policy: dict | None, *, now: float | None = None) def _sync(path: Path) -> None: - if os.name == "posix": - fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(fd) - finally: - os.close(fd) + # `_state` qualifies this transport on POSIX and the rotation moves recovery + # points only after that, so skipping the flush here would report a completed + # rotation that never made it durable. O_DIRECTORY is POSIX-only and only + # hardens the open, so take the flag this host has and keep the guarantee by + # refusing anything that is not the directory itself -- the pattern + # `file_lock.write_json_atomic` already uses. + if os.name != "posix": + return + fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + _require_directory(os.fstat(fd).st_mode, path) + os.fsync(fd) + finally: + os.close(fd) + + +def _require_directory(mode: int, path: Path) -> None: + if not stat.S_ISDIR(mode): + raise ValueError(f"checkpoint sync target is not a directory: {path}") def _receipt(path: Path, record: dict) -> None: diff --git a/tests/architecture/posix_flag_ceilings.json b/tests/architecture/posix_flag_ceilings.json new file mode 100644 index 0000000000..8a69273a3d --- /dev/null +++ b/tests/architecture/posix_flag_ceilings.json @@ -0,0 +1,12 @@ +{ + "schema": "loopx_posix_flag_ceilings_v0", + "baseline_commit": "b4216d8c79f7f31785cf3bb6845f6847fcf0a6a6", + "note": "Per-file ceiling on bare `os.O_NOFOLLOW` / `os.O_DIRECTORY` / `os.O_NONBLOCK` references under loopx/. A ceiling is a maximum, not a target: the guard fails when a file exceeds it or a new file appears, and accepts a lower measurement, so retiring one of these sites needs no edit here. Lower or drop an entry whenever a file is fixed, so the ceiling does not describe an older tree than the code.", + "ceiling": { + "loopx/cli_commands/quota_capture.py": 2, + "loopx/control_plane/collaboration/result_files.py": 6, + "loopx/control_plane/effect_runtime.py": 2, + "loopx/control_plane/turn_driver/codex_context.py": 4, + "loopx/control_plane/turn_driver/codex_context_history.py": 2 + } +} diff --git a/tests/architecture/test_posix_flag_portability.py b/tests/architecture/test_posix_flag_portability.py new file mode 100644 index 0000000000..86a8da7427 --- /dev/null +++ b/tests/architecture/test_posix_flag_portability.py @@ -0,0 +1,159 @@ +"""No module grows a bare reference to a POSIX-only `os` flag. + +`os.O_NOFOLLOW`, `os.O_DIRECTORY` and `os.O_NONBLOCK` are POSIX-only +attributes. Spelling one directly raises `AttributeError` on a host that does +not define it, and it raises it before the bounded-regular-file check on the +next line can refuse anything: the caller reports an untyped crash in place of +its own refusal, or the diagnostic it was opened to produce never appears. +`from os import O_NOFOLLOW` is worse still: the module fails at import time. + +The repository already reads these flags with `getattr(os, "", 0)` in +the peer, completion, decision-capture and effect-runtime readers, and in +`codex_input_trace`, `state_recovery_io` and `checkpoint_retention` after +this change. This guard keeps a new bare site out — including the sites in +files another change is reworking. + +Ceilings live in `posix_flag_ceilings.json`, following +`test_package_kernel_import_ratchet`: an increase needs a recorded decision, +and a decrease is accepted as-is so that fixing one of these files does not +have to edit a fixture that would then describe a tree it no longer matches. +""" + +from __future__ import annotations + +import ast +import json +import re +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[2] +LOOPX_ROOT = REPO_ROOT / "loopx" +CEILINGS_FIXTURE = Path(__file__).with_name("posix_flag_ceilings.json") +CEILINGS_SCHEMA = "loopx_posix_flag_ceilings_v0" +POSIX_ONLY_FLAGS = frozenset({"O_NOFOLLOW", "O_DIRECTORY", "O_NONBLOCK"}) +_COMMIT_SHA = re.compile(r"^[0-9a-f]{40}$") + + +def bare_flag_uses(source: str) -> list[tuple[int, str]]: + """Every direct `os.` or `from os import ` in one module.""" + tree = ast.parse(source) + aliases = { + alias.asname or alias.name + for node in ast.walk(tree) + if isinstance(node, ast.Import) + for alias in node.names + if alias.name == "os" + } + uses: list[tuple[int, str]] = [] + for node in ast.walk(tree): + if ( + isinstance(node, ast.Attribute) + and isinstance(node.value, ast.Name) + and node.value.id in aliases + and node.attr in POSIX_ONLY_FLAGS + ): + uses.append((node.lineno, f"{node.value.id}.{node.attr}")) + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module == "os": + uses.extend( + (node.lineno, f"from os import {alias.name}") + for alias in node.names + if alias.name in POSIX_ONLY_FLAGS + ) + return sorted(uses) + + +def observed_ceiling(loopx_root: Path | None = None) -> dict[str, list[tuple[int, str]]]: + root = loopx_root or LOOPX_ROOT + observed: dict[str, list[tuple[int, str]]] = {} + for path in sorted(root.rglob("*.py")): + uses = bare_flag_uses(path.read_text(encoding="utf-8")) + if uses: + observed[path.relative_to(REPO_ROOT).as_posix()] = uses + return observed + + +def _ceiling() -> dict[str, Any]: + payload = json.loads(CEILINGS_FIXTURE.read_text(encoding="utf-8")) + assert payload["schema"] == CEILINGS_SCHEMA, ( + f"{CEILINGS_FIXTURE.name} schema is {payload['schema']!r}, expected {CEILINGS_SCHEMA!r}" + ) + assert _COMMIT_SHA.match(str(payload["baseline_commit"])), ( + "baseline_commit must be the full 40-character commit the ceilings were measured at" + ) + ceiling = payload["ceiling"] + assert isinstance(ceiling, dict) and ceiling, "ceiling must map module path -> positive count" + assert all( + isinstance(path, str) and isinstance(count, int) and count > 0 + for path, count in ceiling.items() + ), "every ceiling entry needs a module path and a positive count" + assert list(ceiling) == sorted(ceiling), "ceiling entries must be sorted by module path" + return payload + + +def test_no_module_exceeds_its_bare_posix_flag_ceiling() -> None: + ceiling = _ceiling()["ceiling"] + observed = observed_ceiling() + + exceeded = { + path: {"ceiling": ceiling.get(path, 0), "observed": uses} + for path, uses in observed.items() + if len(uses) > ceiling.get(path, 0) + } + assert not exceeded, ( + "bare POSIX-only os flags grew: " + + json.dumps( + {path: [f"{line}: {what}" for line, what in uses] for path, uses in exceeded.items()}, + indent=2, + ) + + ". Read the flag with `getattr(os, \"\", 0)` the way the peer, completion " + "and decision-capture readers do, so a host without it refuses or degrades on its " + "own terms instead of raising AttributeError first. Raising a ceiling in " + f"{CEILINGS_FIXTURE.name} needs a recorded reason." + ) + + +def test_fixed_readers_keep_no_bare_posix_flag() -> None: + """The readers whose fallback this guard protects stay at zero.""" + observed = observed_ceiling() + fixed = ( + "loopx/control_plane/turn_driver/codex_input_trace.py", + "loopx/control_plane/state_recovery_io.py", + "loopx/hosts/checkpoint_retention.py", + ) + remaining = {path: observed[path] for path in fixed if path in observed} + assert not remaining, ( + f"these readers spell a POSIX-only flag directly again: {remaining}. They are " + "reachable on hosts that do not define the flag, so the open must request it " + "with `getattr(os, ...)`." + ) + + +def test_scan_sees_bare_alias_and_from_import_but_not_getattr() -> None: + source = ( + "import os\n" + "import os as host_os\n" + "from os import O_DIRECTORY\n" + "def f(path):\n" + " first = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)\n" + " second = host_os.open(path, host_os.O_RDONLY | host_os.O_NOFOLLOW)\n" + " third = getattr(host_os, \"O_NONBLOCK\", 0)\n" + " return first, second, third, os.O_RDONLY\n" + ) + assert bare_flag_uses(source) == [ + (3, "from os import O_DIRECTORY"), + (5, "os.O_NOFOLLOW"), + (6, "host_os.O_NOFOLLOW"), + ] + + +def test_scan_ignores_a_name_that_is_not_the_os_module() -> None: + """A local `os` binding is not the module, and a flag name alone is not a use.""" + source = ( + "class Holder:\n" + " O_NOFOLLOW = 1\n" + "def f(holder):\n" + " flag = \"O_DIRECTORY\"\n" + " return holder.O_NONBLOCK, Holder.O_NOFOLLOW, flag\n" + ) + assert bare_flag_uses(source) == [] diff --git a/tests/control_plane/test_checkpoint_retention.py b/tests/control_plane/test_checkpoint_retention.py index 09a01470af..ae7e981a82 100644 --- a/tests/control_plane/test_checkpoint_retention.py +++ b/tests/control_plane/test_checkpoint_retention.py @@ -1,6 +1,7 @@ """Real local artifact owner, retention CLI and filesystem recovery negatives.""" import json import os +import stat import threading from types import SimpleNamespace @@ -59,6 +60,25 @@ def test_preview_is_read_only_and_execute_preserves_pins_latest_and_unsealed(ban assert transport.rotate_bank(bank, policy, execute=True)["deleted"] == [] +def test_rotation_flushes_directories_without_the_directory_flag(bank, policy, monkeypatch): + # A host can report os.name == "posix" without defining O_DIRECTORY. The + # rotation still moves recovery points, so it must still flush the + # directories the moves changed; skipping that would report a completed + # rotation that was never made durable. + policy["keep_latest"] = 1 + flushed, real_fsync = [], os.fsync + + def recording(descriptor): + flushed.append(stat.S_ISDIR(os.fstat(descriptor).st_mode)) + return real_fsync(descriptor) + + monkeypatch.setattr(os, "fsync", recording) + monkeypatch.delattr(os, "O_DIRECTORY", raising=False) + result = transport.rotate_bank(bank, policy, execute=True) + assert result["deleted"] + assert sum(flushed) >= 2, "the bank and the trash parent must be flushed after the moves" + + def test_stale_preview_and_changed_pin_cannot_delete(bank, policy): preview = transport.rotate_bank(bank, policy) os.utime(bank / "p2/checkpoint.json", (9000, 9000)) diff --git a/tests/control_plane/test_codex_input_retention.py b/tests/control_plane/test_codex_input_retention.py index 4febe3f3f1..ad2a523485 100644 --- a/tests/control_plane/test_codex_input_retention.py +++ b/tests/control_plane/test_codex_input_retention.py @@ -3,6 +3,7 @@ import hashlib import json +import os from pathlib import Path import pytest @@ -188,6 +189,30 @@ def test_unknown_stale_lossy_or_unbound_observation_never_reuses(tmp_path, failu assert not codex_context_in_active_input(observation, baseline=baseline, session_id="session") +def test_trace_link_swapped_after_containment_is_refused_before_the_read(tmp_path, monkeypatch): + """The path is replaced at open time, past the checks the bundle passed.""" + observation, baseline, request, rows = fixture(tmp_path) + trace = Path(observation["trace_bundle"]) / "trace.jsonl" + outside = tmp_path / "outside.jsonl" + outside.write_text(trace.read_text(encoding="utf-8"), encoding="utf-8") + # The reader resolves the bundle first, so compare against the resolved entry. + resolved = trace.resolve() + real_open = os.open + + def swapping(target, flags, *args, **kwargs): + if os.fspath(target) == os.fspath(resolved) and not trace.is_symlink(): + trace.unlink() + try: + trace.symlink_to(outside) + except (OSError, NotImplementedError): + pytest.skip("this host cannot create a file symlink") + return real_open(target, flags, *args, **kwargs) + + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + monkeypatch.setattr(os, "open", swapping) + assert not codex_context_in_active_input(observation, baseline=baseline, session_id="session") + + @pytest.mark.parametrize("pragma", ["", '// @exec: {"max_output_tokens":12000}\n', ' \t// @exec: {"yield_time_ms":1000}\r\n']) @pytest.mark.parametrize("failure", [None, "parent_missing", "parent_changed", "parent_ended", "parent_wrong_call", diff --git a/tests/control_plane/test_posix_flag_fallback.py b/tests/control_plane/test_posix_flag_fallback.py new file mode 100644 index 0000000000..9ddd8d2fca --- /dev/null +++ b/tests/control_plane/test_posix_flag_fallback.py @@ -0,0 +1,187 @@ +"""Readers keep working where the host does not define a POSIX-only attribute. + +`os.O_NOFOLLOW`, `os.O_NONBLOCK`, `os.O_DIRECTORY` and `os.getuid` are POSIX +attributes. Removing one for the duration of a single call is what +`getattr(os, "", ...)` sees on a host that lacks it, and it is the same +branch a native Windows host takes. Each reader must refuse on its own terms +there: an untyped AttributeError, or an open that quietly follows a symlink, +would replace a diagnosis the caller asked for. +""" + +from __future__ import annotations + +import os +import stat + +import pytest + +from loopx.control_plane.state_recovery_io import StateRecoveryError, _regular_input +from loopx.control_plane.turn_driver.codex_input_trace import read_native_trace_file +from loopx.hosts.checkpoint_retention import _sync + +OWNER_MODEL = pytest.mark.skipif( + getattr(os, "getuid", None) is None, + reason="this host has no owner model to check the private regular file against", +) +POSIX_SYNC = pytest.mark.skipif( + os.name != "posix", + reason="_sync only qualifies a POSIX bank; other hosts are claimed by the no-op case", +) + + +def _symlink(path, target) -> None: + try: + path.symlink_to(target) + except (OSError, NotImplementedError): + pytest.skip("this host cannot create a file symlink") + + +def test_native_trace_read_survives_a_host_without_the_posix_flags(tmp_path, monkeypatch): + trace = tmp_path / "trace.jsonl" + trace.write_bytes(b'{"seq": 1}\n') + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + monkeypatch.delattr(os, "O_NONBLOCK", raising=False) + assert read_native_trace_file(trace) == b'{"seq": 1}\n' + + +def test_native_trace_read_refuses_a_symlink_without_the_flag(tmp_path, monkeypatch): + original = tmp_path / "original.jsonl" + original.write_bytes(b'{"seq": 1}\n') + trace = tmp_path / "trace.jsonl" + _symlink(trace, original) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + with pytest.raises(ValueError, match="symlink"): + read_native_trace_file(trace) + + +@OWNER_MODEL +def test_recovery_readback_refuses_a_host_without_an_owner_model(tmp_path, monkeypatch): + payload = tmp_path / "recovery-audit.json" + payload.write_bytes(b'{"ok": false}') + payload.chmod(0o600) + monkeypatch.delattr(os, "getuid", raising=False) + with pytest.raises(StateRecoveryError, match="recovery_input_unavailable"): + with _regular_input(payload, 1024): + pass + + +@OWNER_MODEL +def test_recovery_readback_survives_a_host_without_the_no_follow_flag(tmp_path, monkeypatch): + payload = tmp_path / "recovery-audit.json" + payload.write_bytes(b'{"ok": false}') + payload.chmod(0o600) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + with _regular_input(payload, 1024) as handle: + assert handle.read() == b'{"ok": false}' + + +@OWNER_MODEL +def test_recovery_readback_refuses_a_symlink_without_the_flag(tmp_path, monkeypatch): + original = tmp_path / "original.json" + original.write_bytes(b'{"ok": false}') + original.chmod(0o600) + payload = tmp_path / "recovery-audit.json" + _symlink(payload, original) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + with pytest.raises(StateRecoveryError, match="recovery_input_unavailable"): + with _regular_input(payload, 1024): + pass + + +@POSIX_SYNC +def test_checkpoint_sync_flushes_the_directory_without_the_flag(tmp_path, monkeypatch): + flushed, real = [], os.fsync + + def recording(descriptor): + flushed.append(stat.S_ISDIR(os.fstat(descriptor).st_mode)) + return real(descriptor) + + monkeypatch.setattr(os, "fsync", recording) + monkeypatch.delattr(os, "O_DIRECTORY", raising=False) + _sync(tmp_path) + assert flushed == [True], "the directory must be flushed, not silently skipped" + + +@POSIX_SYNC +def test_checkpoint_sync_refuses_a_target_that_is_not_a_directory(tmp_path, monkeypatch): + target = tmp_path / "bank.json" + target.write_text("{}") + monkeypatch.delattr(os, "O_DIRECTORY", raising=False) + with pytest.raises(ValueError, match="not a directory"): + _sync(target) + + +def test_checkpoint_sync_claims_nothing_off_posix(tmp_path, monkeypatch): + calls = [] + monkeypatch.setattr(os, "fsync", lambda descriptor: calls.append(descriptor)) + monkeypatch.setattr(os, "name", "nt") + target = tmp_path / "bank.json" + target.write_text("{}") + _sync(tmp_path) + _sync(target) + assert calls == [], "a host outside the POSIX contract must not report a flush" + + +def test_checkpoint_sync_still_syncs_where_the_directory_flag_exists(tmp_path, monkeypatch): + if not getattr(os, "O_DIRECTORY", 0): + pytest.skip("this host does not define O_DIRECTORY") + synced, real = [], os.fsync + monkeypatch.setattr(os, "fsync", lambda descriptor: synced.append(descriptor) or real(descriptor)) + _sync(tmp_path) + assert len(synced) == 1 + + +def _swap_at_open(monkeypatch, path, replacement): + """Replace `path` with a link to `replacement` the instant it is opened.""" + real_open = os.open + + def swapping(target, flags, *args, **kwargs): + if os.fspath(target) == os.fspath(path) and not os.path.islink(path): + path.unlink() + try: + path.symlink_to(replacement) + except (OSError, NotImplementedError): + pytest.skip("this host cannot create a file symlink") + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", swapping) + + +def test_native_trace_read_refuses_a_link_swapped_in_at_the_open(tmp_path, monkeypatch): + """Binding the descriptor refuses the entry the check-then-open window admits.""" + outside = tmp_path / "outside.jsonl" + outside.write_bytes(b'{"external": true}\n') + trace = tmp_path / "trace.jsonl" + trace.write_bytes(b'{"seq": 1}\n') + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + _swap_at_open(monkeypatch, trace, outside) + with pytest.raises(ValueError, match="symlink"): + read_native_trace_file(trace) + + +def test_native_trace_read_refuses_the_same_swap_with_the_flag(tmp_path, monkeypatch): + """The native flag refuses it in the open, so the no-flag branch must match.""" + if not getattr(os, "O_NOFOLLOW", 0): + pytest.skip("this host has no no-follow flag to compare against") + outside = tmp_path / "outside.jsonl" + outside.write_bytes(b'{"external": true}\n') + trace = tmp_path / "trace.jsonl" + trace.write_bytes(b'{"seq": 1}\n') + _swap_at_open(monkeypatch, trace, outside) + with pytest.raises(OSError): + read_native_trace_file(trace) + + +@OWNER_MODEL +def test_recovery_readback_refuses_a_link_swapped_in_at_the_open(tmp_path, monkeypatch): + outside = tmp_path / "outside.json" + outside.write_bytes(b'{"external": true}') + outside.chmod(0o600) + payload = tmp_path / "recovery-audit.json" + payload.write_bytes(b'{"ok": false}') + payload.chmod(0o600) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + _swap_at_open(monkeypatch, payload, outside) + with pytest.raises(StateRecoveryError, match="recovery_input_unavailable"): + with _regular_input(payload, 1024): + pass