diff --git a/src/research_signal_context_pipelines/__init__.py b/src/research_signal_context_pipelines/__init__.py index 99b98fe..6fd08d0 100644 --- a/src/research_signal_context_pipelines/__init__.py +++ b/src/research_signal_context_pipelines/__init__.py @@ -1,5 +1,6 @@ """Research signal context artifact helpers.""" +from .artifact_io import DEFAULT_MAX_ARTIFACT_BYTES, read_bounded_artifact_bytes from .overlay_backtest import OverlayPolicy, backtest_overlay from .context_bundle import DEFAULT_UNIVERSE, build_context_bundle, build_context_from_source from .price_history import PriceExtractionSummary, write_filtered_price_history @@ -16,6 +17,7 @@ "PriceExtractionSummary", "SignalValidationError", "DEFAULT_UNIVERSE", + "DEFAULT_MAX_ARTIFACT_BYTES", "backtest_overlay", "build_context_bundle", "build_context_from_source", @@ -23,6 +25,7 @@ "build_theme_momentum_snapshot", "load_symbol_theme_exposure", "load_theme_taxonomy", + "read_bounded_artifact_bytes", "validate_signal", "validate_theme_momentum_snapshot", "write_filtered_price_history", diff --git a/src/research_signal_context_pipelines/artifact_io.py b/src/research_signal_context_pipelines/artifact_io.py new file mode 100644 index 0000000..a2b0cde --- /dev/null +++ b/src/research_signal_context_pipelines/artifact_io.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import os +import stat +from pathlib import Path + +from .schema import SignalValidationError + + +DEFAULT_MAX_ARTIFACT_BYTES = 4 * 1024 * 1024 +_OPEN_SUPPORTS_DIR_FD = os.open in os.supports_dir_fd + + +def read_bounded_artifact_bytes( + source_path: str | Path, + *, + base_dir: str | Path, + max_bytes: int = DEFAULT_MAX_ARTIFACT_BYTES, +) -> bytes: + """Read one regular artifact beneath a directory without following symlinks.""" + if not isinstance(max_bytes, int) or isinstance(max_bytes, bool) or max_bytes <= 0: + raise SignalValidationError("artifact size limit must be a positive integer") + _require_safe_open_support() + + raw_base = Path(os.path.abspath(os.fspath(base_dir))) + try: + canonical_base = raw_base.resolve(strict=True) + except OSError as exc: + raise SignalValidationError("artifact base directory is unavailable") from exc + if not canonical_base.is_dir(): + raise SignalValidationError("artifact base must be a directory") + + parts = _relative_source_parts(source_path, raw_base=raw_base, canonical_base=canonical_base) + opened_fds: list[int] = [] + try: + base_fd = os.open( + canonical_base, + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, + ) + opened_fds.append(base_fd) + parent_fd = base_fd + for part in parts[:-1]: + parent_fd = os.open( + part, + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, + dir_fd=parent_fd, + ) + opened_fds.append(parent_fd) + artifact_fd = os.open( + parts[-1], + os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW | os.O_CLOEXEC, + dir_fd=parent_fd, + ) + opened_fds.append(artifact_fd) + metadata = os.fstat(artifact_fd) + if not stat.S_ISREG(metadata.st_mode): + raise SignalValidationError("artifact source must be a regular file") + if metadata.st_size > max_bytes: + raise SignalValidationError("artifact exceeds the configured size limit") + with os.fdopen(artifact_fd, "rb", closefd=False) as stream: + payload = stream.read(max_bytes + 1) + if len(payload) > max_bytes: + raise SignalValidationError("artifact exceeds the configured size limit") + return payload + except SignalValidationError: + raise + except (OSError, ValueError) as exc: + raise SignalValidationError("artifact source must be a readable regular file beneath base") from exc + finally: + for fd in reversed(opened_fds): + try: + os.close(fd) + except OSError: + pass + + +def _require_safe_open_support() -> None: + required_flags = ("O_CLOEXEC", "O_DIRECTORY", "O_NOFOLLOW", "O_NONBLOCK") + if any(not hasattr(os, flag) for flag in required_flags) or not _OPEN_SUPPORTS_DIR_FD: + raise SignalValidationError("safe descriptor-relative artifact reads are unsupported") + + +def _relative_source_parts( + source_path: str | Path, + *, + raw_base: Path, + canonical_base: Path, +) -> tuple[str, ...]: + source = Path(source_path) + if source.is_absolute(): + relative = _relative_to_either_base(source, raw_base=raw_base, canonical_base=canonical_base) + else: + relative = source + parts = relative.parts + if not parts or any(part in {"", ".", ".."} for part in parts): + raise SignalValidationError("artifact source must remain beneath base") + return parts + + +def _relative_to_either_base(source: Path, *, raw_base: Path, canonical_base: Path) -> Path: + for base in (raw_base, canonical_base): + try: + return source.relative_to(base) + except ValueError: + continue + raise SignalValidationError("artifact source must remain beneath base") diff --git a/tests/test_artifact_io.py b/tests/test_artifact_io.py new file mode 100644 index 0000000..e27ca2f --- /dev/null +++ b/tests/test_artifact_io.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from research_signal_context_pipelines import SignalValidationError, read_bounded_artifact_bytes + + +def test_reads_relative_source_once_from_base(tmp_path: Path) -> None: + source = tmp_path / "artifact.json" + source.write_bytes(b'{"ok": true}') + + assert read_bounded_artifact_bytes("artifact.json", base_dir=tmp_path) == b'{"ok": true}' + + +def test_reads_absolute_source_through_symlinked_base(tmp_path: Path) -> None: + release = tmp_path / "releases" / "current" + release.mkdir(parents=True) + source = release / "artifact.json" + source.write_bytes(b"payload") + deployed = tmp_path / "deploy-current" + deployed.symlink_to(release, target_is_directory=True) + + assert read_bounded_artifact_bytes(deployed / source.name, base_dir=deployed) == b"payload" + + +def test_reads_absolute_source_through_lexically_normalized_symlinked_base(tmp_path: Path) -> None: + deploys = tmp_path / "deploys" + release = tmp_path / "releases" / "current" + deploys.mkdir() + release.mkdir(parents=True) + source = release / "artifact.json" + source.write_bytes(b"payload") + deployed = deploys / "current" + deployed.symlink_to(release, target_is_directory=True) + configured_base = deploys / ".." / "deploys" / "current" + + assert read_bounded_artifact_bytes(deployed / source.name, base_dir=configured_base) == b"payload" + + +def test_reads_absolute_source_under_canonical_base(tmp_path: Path) -> None: + release = tmp_path / "release" + release.mkdir() + source = release / "artifact.json" + source.write_bytes(b"payload") + deployed = tmp_path / "deploy-current" + deployed.symlink_to(release, target_is_directory=True) + + assert read_bounded_artifact_bytes(source, base_dir=deployed) == b"payload" + + +@pytest.mark.parametrize("source", ["../outside.json", "nested/../../outside.json"]) +def test_rejects_relative_escape(tmp_path: Path, source: str) -> None: + (tmp_path.parent / "outside.json").write_bytes(b"outside") + + with pytest.raises(SignalValidationError, match="beneath base"): + read_bounded_artifact_bytes(source, base_dir=tmp_path) + + +def test_rejects_absolute_source_outside_base(tmp_path: Path) -> None: + base = tmp_path / "base" + base.mkdir() + outside = tmp_path / "outside.json" + outside.write_bytes(b"outside") + + with pytest.raises(SignalValidationError, match="beneath base"): + read_bounded_artifact_bytes(outside, base_dir=base) + + +def test_rejects_leaf_symlink(tmp_path: Path) -> None: + target = tmp_path / "target.json" + target.write_bytes(b"payload") + (tmp_path / "artifact.json").symlink_to(target) + + with pytest.raises(SignalValidationError, match="readable regular file"): + read_bounded_artifact_bytes("artifact.json", base_dir=tmp_path) + + +def test_rejects_intermediate_symlink(tmp_path: Path) -> None: + real_dir = tmp_path / "real" + real_dir.mkdir() + (real_dir / "artifact.json").write_bytes(b"payload") + (tmp_path / "linked").symlink_to(real_dir, target_is_directory=True) + + with pytest.raises(SignalValidationError, match="readable regular file"): + read_bounded_artifact_bytes("linked/artifact.json", base_dir=tmp_path) + + +def test_rejects_fifo_without_blocking(tmp_path: Path) -> None: + fifo = tmp_path / "artifact.pipe" + os.mkfifo(fifo) + + with pytest.raises(SignalValidationError, match="regular file"): + read_bounded_artifact_bytes(fifo.name, base_dir=tmp_path) + + +def test_rejects_oversized_artifact(tmp_path: Path) -> None: + source = tmp_path / "artifact.json" + source.write_bytes(b"12345") + + with pytest.raises(SignalValidationError, match="size limit"): + read_bounded_artifact_bytes(source.name, base_dir=tmp_path, max_bytes=4) + + +def test_closes_all_opened_descriptors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + source = tmp_path / "nested" / "artifact.json" + source.parent.mkdir() + source.write_bytes(b"payload") + real_open = os.open + real_close = os.close + opened: list[int] = [] + closed: list[int] = [] + + def tracking_open(*args, **kwargs) -> int: + fd = real_open(*args, **kwargs) + opened.append(fd) + return fd + + def tracking_close(fd: int) -> None: + closed.append(fd) + real_close(fd) + + monkeypatch.setattr(os, "open", tracking_open) + monkeypatch.setattr(os, "close", tracking_close) + + assert read_bounded_artifact_bytes("nested/artifact.json", base_dir=tmp_path) == b"payload" + assert set(opened) <= set(closed)