diff --git a/strix/core/runner.py b/strix/core/runner.py index 8726f819..b4afdfaf 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -114,6 +114,7 @@ async def run_strix_scan( scan_id: str | None = None, image: str, local_sources: list[dict[str, Any]] | None = None, + extra_files: list[dict[str, Any]] | None = None, coordinator: AgentCoordinator | None = None, interactive: bool = False, max_turns: int = DEFAULT_MAX_TURNS, @@ -129,6 +130,9 @@ async def run_strix_scan( ``root_instructions_override`` adds root scan instructions to the rendered root prompt without replacing the system-verified scope block. + ``extra_files`` entries (``{"workspace_path", "content"}``) are placed into + the sandbox workspace at session bring-up; see + :func:`strix.runtime.session_manager.create_or_reuse`. ``extra_system_prompt_context`` is merged into the root agent's scan context before prompt rendering. Child agents keep the standard scan prompt and context. @@ -228,6 +232,7 @@ async def run_strix_scan( scan_id, image=image, local_sources=local_sources or [], + extra_files=extra_files, status_sink=status_sink, ) report("Waiting for the first model response") diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py index 4b61d735..28b62ce8 100644 --- a/strix/runtime/session_manager.py +++ b/strix/runtime/session_manager.py @@ -8,10 +8,11 @@ import sys from pathlib import Path from typing import TYPE_CHECKING, Any -from agents.sandbox.entries import BaseEntry, LocalDir +from agents.sandbox.entries import BaseEntry, File, LocalDir from agents.sandbox.manifest import Environment, Manifest from strix.config import load_settings +from strix.core.paths import run_dir_for, runtime_state_dir from strix.runtime.backends import backend_supports_bind_mounts, get_backend from strix.runtime.caido_bootstrap import bootstrap_caido @@ -73,6 +74,86 @@ def build_manifest_entries(local_sources: list[dict[str, Any]]) -> dict[str | Pa return entries +def _extra_file_rel_path(workspace_path: str) -> str | None: + """Validate an extra-file target path and return it relative to /workspace. + + Only absolute paths under the workspace root are accepted; anything else + (including ``..`` traversal segments) is rejected so callers cannot place + orchestrator-provided content outside the sandbox workspace. + """ + prefix = f"{_WORKSPACE_ROOT}/" + if not workspace_path.startswith(prefix): + return None + rel = workspace_path[len(prefix) :].strip("/") + if not rel or any(part in ("", ".", "..") for part in rel.split("/")): + return None + return rel + + +def _extra_file_content(extra_file: dict[str, Any]) -> bytes | None: + content = extra_file.get("content") + if isinstance(content, bytes | bytearray): + return bytes(content) + if isinstance(content, str): + return content.encode("utf-8") + return None + + +def build_extra_file_entries(extra_files: list[dict[str, Any]]) -> dict[str | Path, BaseEntry]: + """Map extra files to in-memory ``File`` manifest entries. + + Each item is ``{"workspace_path": "/workspace/", "content": bytes|str}``; + manifest backends materialize the entry at the requested path alongside the + ``LocalDir`` source uploads. Invalid items are skipped with a warning. + """ + entries: dict[str | Path, BaseEntry] = {} + for extra_file in extra_files: + rel = _extra_file_rel_path(str(extra_file.get("workspace_path") or "")) + content = _extra_file_content(extra_file) + if rel is None or content is None: + logger.warning( + "Skipping invalid extra file entry (workspace_path=%r)", + extra_file.get("workspace_path"), + ) + continue + entries[rel] = File(content=content) + return entries + + +def build_extra_file_bind_mounts( + extra_files: list[dict[str, Any]], + staging_dir: Path, +) -> list[dict[str, Any]]: + """Stage extra files on the host and map them to read-only bind mounts. + + Bind-mount backends bypass the manifest, so the content is written under + ``staging_dir`` (one numbered subdirectory per file to avoid basename + collisions) and mounted read-only at the same ``/workspace/`` path the + manifest path would use. Invalid items are skipped with a warning. + """ + mounts: list[dict[str, Any]] = [] + for index, extra_file in enumerate(extra_files): + rel = _extra_file_rel_path(str(extra_file.get("workspace_path") or "")) + content = _extra_file_content(extra_file) + if rel is None or content is None: + logger.warning( + "Skipping invalid extra file entry (workspace_path=%r)", + extra_file.get("workspace_path"), + ) + continue + host_file = staging_dir / str(index) / Path(rel).name + host_file.parent.mkdir(parents=True, exist_ok=True) + host_file.write_bytes(content) + mounts.append( + { + "source": str(host_file), + "target": f"{_WORKSPACE_ROOT}/{rel}", + "read_only": True, + } + ) + return mounts + + def _metadata_mounts(tree: Path, target: str) -> list[dict[str, Any]]: mounts: list[dict[str, Any]] = [] for name in _PROTECTED_METADATA_NAMES: @@ -111,12 +192,19 @@ async def create_or_reuse( *, image: str, local_sources: list[dict[str, Any]], + extra_files: list[dict[str, Any]] | None = None, status_sink: StatusSink | None = None, ) -> dict[str, Any]: """Return the existing session bundle for ``scan_id`` or create a new one. Each ``local_sources`` entry exposes its host ``source_path`` at ``/workspace/`` inside the container. + + Each ``extra_files`` entry (``{"workspace_path": "/workspace/", + "content": bytes | str}``) lands as a single file at its ``workspace_path`` + regardless of backend: an in-memory ``File`` manifest entry on manifest + backends, a read-only bind mount of a host-staged copy on bind-mount + backends. """ def report(phase: str) -> None: @@ -134,9 +222,14 @@ async def create_or_reuse( if backend_supports_bind_mounts(backend_name): bind_mounts = build_bind_mounts(local_sources) entries: dict[str | Path, BaseEntry] = {} + if extra_files: + staging_dir = runtime_state_dir(run_dir_for(scan_id)) / "extra_files" + bind_mounts.extend(build_extra_file_bind_mounts(extra_files, staging_dir)) else: bind_mounts = [] entries = build_manifest_entries(local_sources) + if extra_files: + entries.update(build_extra_file_entries(extra_files)) # Caido runs as an in-container sidecar; HTTP(S) traffic from any # process started via ``session.exec`` (the SDK's Shell tool, etc.) diff --git a/tests/test_session_entries.py b/tests/test_session_entries.py index 787422a1..da64202d 100644 --- a/tests/test_session_entries.py +++ b/tests/test_session_entries.py @@ -2,9 +2,10 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any +from pathlib import Path +from typing import Any -from agents.sandbox.entries import LocalDir +from agents.sandbox.entries import File, LocalDir from strix.runtime.backends import ( _BACKENDS, @@ -12,11 +13,12 @@ from strix.runtime.backends import ( backend_supports_bind_mounts, register_backend, ) -from strix.runtime.session_manager import build_bind_mounts, build_manifest_entries - - -if TYPE_CHECKING: - from pathlib import Path +from strix.runtime.session_manager import ( + build_bind_mounts, + build_extra_file_bind_mounts, + build_extra_file_entries, + build_manifest_entries, +) def _source(subdir: str, path: str, *, protect_metadata: bool = False) -> dict[str, Any]: @@ -163,6 +165,92 @@ def test_manifest_entries_skip_incomplete_sources() -> None: ) +def test_extra_file_becomes_in_memory_manifest_entry() -> None: + entries = build_extra_file_entries( + [{"workspace_path": "/workspace/.strix/dependency-issues.jsonl", "content": b"{}\n"}] + ) + + assert set(entries) == {".strix/dependency-issues.jsonl"} + entry = entries[".strix/dependency-issues.jsonl"] + assert isinstance(entry, File) + assert entry.content == b"{}\n" + + +def test_extra_file_str_content_is_encoded_utf8() -> None: + entries = build_extra_file_entries( + [{"workspace_path": "/workspace/.strix/note.txt", "content": "héllo"}] + ) + + entry = entries[".strix/note.txt"] + assert isinstance(entry, File) + assert entry.content == "héllo".encode() + + +def test_extra_file_invalid_paths_and_content_are_skipped() -> None: + assert ( + build_extra_file_entries( + [ + {"workspace_path": "/etc/passwd", "content": b"x"}, + {"workspace_path": "/workspace/../escape", "content": b"x"}, + {"workspace_path": "/workspace/a/../../escape", "content": b"x"}, + {"workspace_path": "/workspace/", "content": b"x"}, + {"workspace_path": "", "content": b"x"}, + {"workspace_path": "/workspace/ok.txt", "content": None}, + {"workspace_path": "/workspace/ok.txt"}, + ] + ) + == {} + ) + + +def test_extra_file_becomes_read_only_bind_mount_of_staged_copy(tmp_path: Path) -> None: + staging = tmp_path / "staging" + + mounts = build_extra_file_bind_mounts( + [{"workspace_path": "/workspace/.strix/dependency-issues.jsonl", "content": b"{}\n"}], + staging, + ) + + assert len(mounts) == 1 + mount = mounts[0] + assert mount["target"] == "/workspace/.strix/dependency-issues.jsonl" + assert mount["read_only"] is True + staged = Path(mount["source"]) + assert staged.read_bytes() == b"{}\n" + assert staged.is_relative_to(staging) + + +def test_extra_file_bind_mounts_and_entries_agree_on_the_sandbox_path(tmp_path: Path) -> None: + extra = [{"workspace_path": "/workspace/.strix/dependency-issues.jsonl", "content": b"{}\n"}] + + entries = build_extra_file_entries(extra) + mounts = build_extra_file_bind_mounts(extra, tmp_path) + + (rel,) = entries + assert mounts[0]["target"] == f"/workspace/{rel}" + + +def test_extra_file_bind_mounts_skip_invalid_entries(tmp_path: Path) -> None: + bad = [{"workspace_path": "/nope", "content": b"x"}] + assert build_extra_file_bind_mounts(bad, tmp_path) == [] + assert not tmp_path.exists() or list(tmp_path.iterdir()) == [] + + +def test_extra_file_bind_mounts_avoid_basename_collisions(tmp_path: Path) -> None: + mounts = build_extra_file_bind_mounts( + [ + {"workspace_path": "/workspace/a/data.txt", "content": b"a"}, + {"workspace_path": "/workspace/b/data.txt", "content": b"b"}, + ], + tmp_path, + ) + + assert [m["target"] for m in mounts] == ["/workspace/a/data.txt", "/workspace/b/data.txt"] + assert Path(mounts[0]["source"]).read_bytes() == b"a" + assert Path(mounts[1]["source"]).read_bytes() == b"b" + assert mounts[0]["source"] != mounts[1]["source"] + + def test_only_bind_mount_capable_backends_are_registered_as_such() -> None: assert backend_supports_bind_mounts("docker") assert not backend_supports_bind_mounts("e2b")