From df580d1e6416100e0f1954c3480697934a926da0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 16 Oct 2026 02:07:58 +0800 Subject: [PATCH] fix: harden isolated preview bundle readback Co-Authored-By: Codex --- src/quant_advisor_research/preview_bundle.py | 65 +++++++++++++++++--- tests/test_preview_bundle.py | 63 +++++++++++++++++-- 2 files changed, 117 insertions(+), 11 deletions(-) diff --git a/src/quant_advisor_research/preview_bundle.py b/src/quant_advisor_research/preview_bundle.py index 5d71fc1..93bc1d3 100644 --- a/src/quant_advisor_research/preview_bundle.py +++ b/src/quant_advisor_research/preview_bundle.py @@ -1,10 +1,14 @@ """Concrete daily report-to-preview bundle with no legacy or runtime integration.""" from __future__ import annotations +import ctypes +import errno import hashlib import html import json import os +import stat +import sys import tempfile from collections.abc import Mapping from dataclasses import dataclass @@ -42,12 +46,52 @@ class PreviewBundleEvidence: def __post_init__(self) -> None: if not isinstance(self.report, Mapping) or not isinstance(self.manifest, Mapping): raise PreviewBundleError("readback_invalid") + object.__setattr__(self, "report", _freeze(self.report)) + object.__setattr__(self, "manifest", _freeze(self.manifest)) @property def bundle_contract(self) -> str: return BUNDLE_CONTRACT +def _freeze(value: object) -> object: + if isinstance(value, Mapping): + return MappingProxyType({key: _freeze(item) for key, item in value.items()}) + if isinstance(value, (list, tuple)): + return tuple(_freeze(item) for item in value) + return value + + +def _install_no_replace(source: str, destination: Path) -> None: + """Atomically publish without replacing even an existing empty directory.""" + if sys.platform == "win32": + os.rename(source, destination) + return + libc = ctypes.CDLL(None, use_errno=True) + if sys.platform == "darwin": + operation = libc.renamex_np + operation.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint] + args = (os.fsencode(source), os.fsencode(destination), 4) # RENAME_EXCL + elif sys.platform.startswith("linux") and hasattr(libc, "renameat2"): + operation = libc.renameat2 + operation.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] + args = (-100, os.fsencode(source), -100, os.fsencode(destination), 1) # RENAME_NOREPLACE + else: + raise OSError(errno.ENOTSUP, "exclusive directory publication unavailable") + operation.restype = ctypes.c_int + if operation(*args) != 0: + code = ctypes.get_errno() + raise OSError(code, os.strerror(code)) + + +def _read_regular(directory: int, name: str) -> bytes: + descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) + with os.fdopen(descriptor, "rb") as stream: + if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): + raise _error("readback_file_set_invalid") + return stream.read() + + def _error(code: str) -> PreviewBundleError: return PreviewBundleError(code) @@ -138,7 +182,7 @@ def build_preview_bundle(report: Mapping[str, Any], output_dir: str | Path) -> P for name, content in files.items(): Path(staging_dir, name).write_bytes(content) read_preview_bundle(staging_dir) - os.rename(staging_dir, output) + _install_no_replace(staging_dir, output) staging_dir = None except FileExistsError: raise _error("output_exists") from None @@ -163,11 +207,15 @@ def _parse_json_bytes(content: bytes) -> object: def read_preview_bundle(output_dir: str | Path) -> PreviewBundleEvidence: output = Path(output_dir) try: - if not output.is_dir() or {item.name for item in output.iterdir()} != _FIXED_FILES: - raise _error("readback_file_set_invalid") - report_bytes = (output / "report.json").read_bytes() - html_bytes = (output / "report.html").read_bytes() - manifest_bytes = (output / "manifest.json").read_bytes() + descriptor = os.open(output, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + if set(os.listdir(descriptor)) != _FIXED_FILES: + raise _error("readback_file_set_invalid") + report_bytes = _read_regular(descriptor, "report.json") + html_bytes = _read_regular(descriptor, "report.html") + manifest_bytes = _read_regular(descriptor, "manifest.json") + finally: + os.close(descriptor) except PreviewBundleError: raise except (OSError, TypeError, ValueError): @@ -182,7 +230,10 @@ def read_preview_bundle(output_dir: str | Path) -> PreviewBundleEvidence: manifest = _parse_json_bytes(manifest_bytes) if not isinstance(manifest, Mapping): raise _error("manifest_invalid") - if set(manifest) != _MANIFEST_KEYS or set(manifest.get("source", {})) != _SOURCE_KEYS: + if _canonical_json(manifest) != manifest_bytes: + raise _error("manifest_bytes_noncanonical") + if (set(manifest) != _MANIFEST_KEYS or not isinstance(manifest.get("source"), Mapping) + or set(manifest["source"]) != _SOURCE_KEYS): raise _error("manifest_shape_invalid") artifacts = manifest.get("artifacts") if not isinstance(artifacts, Mapping) or set(artifacts) != {"report.json", "report.html"}: diff --git a/tests/test_preview_bundle.py b/tests/test_preview_bundle.py index 7d2497e..f9717d6 100644 --- a/tests/test_preview_bundle.py +++ b/tests/test_preview_bundle.py @@ -162,18 +162,18 @@ def inspect(path): def test_concurrent_destination_winner_is_not_overwritten_or_cleaned(tmp_path, monkeypatch): output = tmp_path / "preview" - real_rename = preview_bundle.os.rename + real_rename = preview_bundle._install_no_replace def concurrent_winner(source, destination): Path(destination).mkdir() - raise FileExistsError(destination) + real_rename(source, destination) - monkeypatch.setattr(preview_bundle.os, "rename", concurrent_winner) + monkeypatch.setattr(preview_bundle, "_install_no_replace", concurrent_winner) with pytest.raises(PreviewBundleError, match="output_exists"): build_preview_bundle(report(), output) assert output.is_dir() assert not list(tmp_path.glob(".preview.staging-*")) - monkeypatch.setattr(preview_bundle.os, "rename", real_rename) + monkeypatch.setattr(preview_bundle, "_install_no_replace", real_rename) def test_staging_directory_is_cleaned_when_install_fails(tmp_path, monkeypatch): @@ -187,3 +187,58 @@ def fail_readback(_path): build_preview_bundle(report(), output) assert not output.exists() assert not list(tmp_path.glob(".preview.staging-*")) + + +@pytest.mark.parametrize("name", ["report.json", "report.html", "manifest.json"]) +def test_readback_rejects_symlink_entries(tmp_path, name): + output = tmp_path / "preview" + build_preview_bundle(report(), output) + target = tmp_path / name + (output / name).rename(target) + (output / name).symlink_to(target) + with pytest.raises(PreviewBundleError): + read_preview_bundle(output) + + +def test_readback_rejects_symlink_root_and_nonregular_entry(tmp_path): + output = tmp_path / "preview" + build_preview_bundle(report(), output) + link = tmp_path / "link" + link.symlink_to(output, target_is_directory=True) + with pytest.raises(PreviewBundleError): + read_preview_bundle(link) + (output / "report.html").unlink() + (output / "report.html").mkdir() + with pytest.raises(PreviewBundleError): + read_preview_bundle(output) + + +def test_readback_requires_canonical_manifest_and_mapping_source(tmp_path): + output = tmp_path / "preview" + build_preview_bundle(report(), output) + manifest_path = output / "manifest.json" + manifest = json.loads(manifest_path.read_bytes()) + manifest_path.write_text(json.dumps(manifest, indent=2)) + with pytest.raises(PreviewBundleError, match="manifest_bytes_noncanonical"): + read_preview_bundle(output) + manifest["source"] = None + manifest_path.write_bytes(preview_bundle._canonical_json(manifest)) + with pytest.raises(PreviewBundleError, match="manifest_shape_invalid"): + read_preview_bundle(output) + + +def test_returned_evidence_is_recursively_immutable(tmp_path): + output = tmp_path / "preview" + for evidence in (build_preview_bundle(report(), output), read_preview_bundle(output)): + with pytest.raises(TypeError): + evidence.manifest["source"]["cadence"] = "monthly" + with pytest.raises(TypeError): + evidence.report["source_artifacts"]["political_events"] = "changed" + assert isinstance(evidence.report["recommendations"], tuple) + + +def test_v5_explicit_contract_field_is_rejected(tmp_path): + value = report() + value.update(schema_version="5", contract_version="model_recommendations.v5") + with pytest.raises(PreviewBundleError): + build_preview_bundle(value, tmp_path / "preview")