From 1300d5a615f80d9acff3c106bc03b4b730344097 Mon Sep 17 00:00:00 2001 From: yoni Date: Fri, 14 Aug 2026 19:19:54 +0000 Subject: [PATCH] reject extra-file paths that collide with a local source tree --- strix/runtime/session_manager.py | 61 +++++++++++++++++++++++++++++--- tests/test_session_entries.py | 37 +++++++++++++++++++ 2 files changed, 93 insertions(+), 5 deletions(-) diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py index 28b62ce8..6ab0d80b 100644 --- a/strix/runtime/session_manager.py +++ b/strix/runtime/session_manager.py @@ -90,6 +90,33 @@ def _extra_file_rel_path(workspace_path: str) -> str | None: return rel +def _source_root_rels(local_sources: list[dict[str, Any]] | None) -> list[str]: + """Workspace-relative roots the local sources occupy (e.g. ``["repo"]``).""" + if not local_sources: + return [] + return [ + str(src.get("workspace_subdir") or "").strip("/") + for src in local_sources + if src.get("workspace_subdir") and src.get("source_path") + ] + + +def _collides_with_source_root(rel: str, source_roots: list[str]) -> bool: + """True when an extra-file path would land on or inside a source tree. + + An exact match would replace the whole source tree with one file (a + manifest ``entries`` key collision); a path nested under a source root + would race the source upload; a path that is an ancestor of a source root + would shadow the directory the source materializes into. + """ + for root in source_roots: + if not root: + continue + if rel == root or rel.startswith(f"{root}/") or root.startswith(f"{rel}/"): + return True + return False + + def _extra_file_content(extra_file: dict[str, Any]) -> bytes | None: content = extra_file.get("content") if isinstance(content, bytes | bytearray): @@ -99,13 +126,19 @@ def _extra_file_content(extra_file: dict[str, Any]) -> bytes | None: return None -def build_extra_file_entries(extra_files: list[dict[str, Any]]) -> dict[str | Path, BaseEntry]: +def build_extra_file_entries( + extra_files: list[dict[str, Any]], + local_sources: list[dict[str, Any]] | None = None, +) -> 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. + ``LocalDir`` source uploads. Invalid items — including paths that collide + with a ``local_sources`` tree, which would otherwise replace its manifest + entry — are skipped with a warning. """ + source_roots = _source_root_rels(local_sources) entries: dict[str | Path, BaseEntry] = {} for extra_file in extra_files: rel = _extra_file_rel_path(str(extra_file.get("workspace_path") or "")) @@ -116,6 +149,12 @@ def build_extra_file_entries(extra_files: list[dict[str, Any]]) -> dict[str | Pa extra_file.get("workspace_path"), ) continue + if _collides_with_source_root(rel, source_roots): + logger.warning( + "Skipping extra file colliding with a local source tree (workspace_path=%r)", + extra_file.get("workspace_path"), + ) + continue entries[rel] = File(content=content) return entries @@ -123,14 +162,18 @@ def build_extra_file_entries(extra_files: list[dict[str, Any]]) -> dict[str | Pa def build_extra_file_bind_mounts( extra_files: list[dict[str, Any]], staging_dir: Path, + local_sources: list[dict[str, Any]] | None = None, ) -> 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. + manifest path would use. Invalid items — including paths that collide with + a ``local_sources`` tree, which would duplicate or shadow its mount target + — are skipped with a warning. """ + source_roots = _source_root_rels(local_sources) 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 "")) @@ -141,6 +184,12 @@ def build_extra_file_bind_mounts( extra_file.get("workspace_path"), ) continue + if _collides_with_source_root(rel, source_roots): + logger.warning( + "Skipping extra file colliding with a local source tree (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) @@ -224,12 +273,14 @@ async def create_or_reuse( 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)) + bind_mounts.extend( + build_extra_file_bind_mounts(extra_files, staging_dir, local_sources) + ) else: bind_mounts = [] entries = build_manifest_entries(local_sources) if extra_files: - entries.update(build_extra_file_entries(extra_files)) + entries.update(build_extra_file_entries(extra_files, local_sources)) # 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 da64202d..16b59493 100644 --- a/tests/test_session_entries.py +++ b/tests/test_session_entries.py @@ -203,6 +203,43 @@ def test_extra_file_invalid_paths_and_content_are_skipped() -> None: ) +def test_extra_file_colliding_with_a_source_tree_is_skipped(tmp_path: Path) -> None: + sources = [_source("repo", str(tmp_path))] + colliding = [ + {"workspace_path": "/workspace/repo", "content": b"x"}, # exact: would drop the tree + {"workspace_path": "/workspace/repo/inside.txt", "content": b"x"}, # nested inside it + {"workspace_path": "/workspace/repo/deep/inside.txt", "content": b"x"}, + ] + + assert build_extra_file_entries(colliding, sources) == {} + assert build_extra_file_bind_mounts(colliding, tmp_path / "staging", sources) == [] + + +def test_extra_file_shadowing_a_nested_source_root_is_skipped(tmp_path: Path) -> None: + sources = [_source("nested/repo", str(tmp_path))] + shadowing = [{"workspace_path": "/workspace/nested", "content": b"x"}] + + assert build_extra_file_entries(shadowing, sources) == {} + assert build_extra_file_bind_mounts(shadowing, tmp_path / "staging", sources) == [] + + +def test_extra_file_beside_a_source_tree_is_kept(tmp_path: Path) -> None: + sources = [_source("repo", str(tmp_path))] + beside = [ + {"workspace_path": "/workspace/.strix/dependency-issues.jsonl", "content": b"{}\n"}, + {"workspace_path": "/workspace/repo-notes.txt", "content": b"x"}, # sibling, no prefix + ] + + entries = build_extra_file_entries(beside, sources) + mounts = build_extra_file_bind_mounts(beside, tmp_path / "staging", sources) + + assert set(entries) == {".strix/dependency-issues.jsonl", "repo-notes.txt"} + assert [m["target"] for m in mounts] == [ + "/workspace/.strix/dependency-issues.jsonl", + "/workspace/repo-notes.txt", + ] + + def test_extra_file_becomes_read_only_bind_mount_of_staged_copy(tmp_path: Path) -> None: staging = tmp_path / "staging"