diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 4cd9e037..b3bd0291 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -57,6 +57,7 @@ from strix.tools.reporting.tool import ( create_dependency_report, create_vulnerability_report, delete_vulnerability_report, + export_markdown_report, get_report, list_reports, update_vulnerability_report, @@ -599,6 +600,7 @@ _BASE_TOOLS: tuple[Tool, ...] = ( delete_vulnerability_report, list_reports, get_report, + export_markdown_report, list_requests, view_request, repeat_request, diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index 791bae79..24d18e41 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -11,11 +11,12 @@ import asyncio import json import logging import re -from pathlib import PurePosixPath +from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Any from agents import RunContextWrapper, function_tool +from strix.report.writer import atomic_write_text from strix.tools.nullish import clean_optional from strix.tools.proxy.tools import existing_request_ids @@ -171,6 +172,22 @@ _VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"}) _VALID_CONFIDENCE = frozenset({"high", "medium", "low"}) _MAX_HTTP_EXCHANGE_IDS = 10 _MAX_HTTP_EXCHANGE_ID_CHARS = 128 +_WORKSPACE_ROOT = "/workspace" +_MARKDOWN_REPORT_SUFFIX = ".md" +_MAX_EXPORTED_REPORT_BYTES = 1_048_576 +# Names the harness already writes beside the exported report. Overwriting them +# would replace a scan artifact with agent-authored markdown. +_RESERVED_REPORT_NAMES = frozenset( + { + "penetration_test_report.md", + "findings.sarif", + "vulnerabilities.json", + "vulnerabilities.csv", + "coverage.json", + "run.json", + "strix.log", + } +) def _validate_required_text(fields: dict[str, str]) -> list[str]: @@ -2513,6 +2530,245 @@ def _do_get_report(report_id: str, caller_agent_id: str | None = None) -> dict[s } +def _markdown_report_path(path: str) -> tuple[str | None, str | None, str | None]: + """Return ``(workspace_path, basename, error)`` for a sandbox markdown file. + + Only a markdown file inside the sandbox workspace can be copied. Absolute + paths elsewhere, traversal, and harness-owned artifact names are rejected + so an export cannot replace ``findings.sarif`` or the executive report, + and cannot read outside the workspace. + """ + raw = (path or "").strip() + prefix = f"{_WORKSPACE_ROOT}/" + if not raw.startswith(prefix): + return ( + None, + None, + "path must be an absolute markdown file under /workspace " + "(for example /workspace/report.md)", + ) + relative = raw[len(prefix) :].strip("/") + parts = relative.split("/") + if not relative or any(part in ("", ".", "..") for part in parts): + return None, None, "path must stay inside /workspace and must not contain '..'" + if any(ord(char) < 0x20 or ord(char) == 0x7F for char in relative): + return None, None, "path must not contain control characters" + name = parts[-1] + if not name.endswith(_MARKDOWN_REPORT_SUFFIX) or name == _MARKDOWN_REPORT_SUFFIX: + return None, None, "path must name a markdown file ending in .md" + if name in _RESERVED_REPORT_NAMES: + return ( + None, + None, + f"'{name}' is written by the scan harness; choose another markdown filename", + ) + return f"{_WORKSPACE_ROOT}/{relative}", name, None + + +def _sandbox_session(ctx: RunContextWrapper) -> Any: + inner = ctx.context if isinstance(ctx.context, dict) else {} + return inner.get("sandbox_session") + + +def _entry_kind(entry: Any) -> str: + kind = getattr(entry, "kind", None) + if kind is None: + return "file" + value = getattr(kind, "value", kind) + return str(value).lower() + + +def _entry_matches_path(entry: Any, path: str) -> bool: + entry_path = str(getattr(entry, "path", "") or "") + if not entry_path: + return False + wanted = PurePosixPath(path) + listed = PurePosixPath(entry_path) + return entry_path in {path, wanted.name} or listed.name == wanted.name + + +async def _inspect_workspace_file(session: Any, path: str) -> str | None: + """Reject symlinks and oversized files before the sandbox read follows them.""" + ls = getattr(session, "ls", None) + if not callable(ls): + return None + parent = PurePosixPath(path).parent.as_posix() + try: + entries = await ls(Path(parent)) + except Exception as exc: + logger.exception("export_markdown_report ls failed") + return f"Failed to inspect '{path}' in the sandbox: {exc!s}" + entry = next((item for item in entries or [] if _entry_matches_path(item, path)), None) + if entry is None: + return f"Markdown file not found: {path}" + return _reject_workspace_entry(path, entry) + + +def _reject_workspace_entry(path: str, entry: Any) -> str | None: + kind = _entry_kind(entry) + if kind == "symlink": + return f"'{path}' is a symlink; export a regular markdown file inside /workspace" + if kind != "file": + return f"'{path}' is not a regular markdown file" + size = getattr(entry, "size", None) + if isinstance(size, int) and size <= 0: + return f"Markdown file is empty: {path}" + if isinstance(size, int) and size > _MAX_EXPORTED_REPORT_BYTES: + return f"Markdown file exceeds {_MAX_EXPORTED_REPORT_BYTES} bytes ({size} bytes)" + return None + + +def _export_destination( + ctx: RunContextWrapper, path: str +) -> tuple[Any, Path, str, str] | dict[str, Any]: + workspace_path, name, path_error = _markdown_report_path(path) + if workspace_path is None or name is None: + return {"success": False, "error": path_error} + + from strix.report.state import get_global_report_state + + report_state = get_global_report_state() + if report_state is None: + return { + "success": False, + "error": "Report state unavailable - the host run directory is not ready", + } + session = _sandbox_session(ctx) + if session is None or not hasattr(session, "read"): + return { + "success": False, + "error": "Sandbox session unavailable - the markdown file cannot be copied", + } + return session, report_state.get_run_dir(), workspace_path, name + + +def _read_limited_bytes(stream: Any) -> tuple[bytes | None, str | None]: + """Copy at most one extra byte past the export limit, then close ``stream``.""" + limit = _MAX_EXPORTED_REPORT_BYTES + 1 + try: + chunks: list[bytes] = [] + remaining = limit + while remaining > 0: + chunk = stream.read(remaining) + if not chunk: + break + if isinstance(chunk, str): + chunk = chunk.encode("utf-8") + if not isinstance(chunk, bytes | bytearray): + return None, "sandbox read did not return file bytes" + chunks.append(bytes(chunk)) + remaining -= len(chunk) + return b"".join(chunks), None + finally: + close = getattr(stream, "close", None) + if callable(close): + close() + + +async def _read_workspace_markdown(session: Any, path: str) -> tuple[bytes | None, str | None]: + inspect_error = await _inspect_workspace_file(session, path) + if inspect_error is not None: + return None, inspect_error + try: + stream = await session.read(Path(path)) + except FileNotFoundError: + return None, f"Markdown file not found: {path}" + except Exception as exc: + logger.exception("export_markdown_report read failed") + return None, f"Failed to read '{path}' from the sandbox: {exc!s}" + data, read_error = _read_limited_bytes(stream) + if data is None: + return None, read_error + if len(data) > _MAX_EXPORTED_REPORT_BYTES: + return None, ( + f"Markdown file exceeds {_MAX_EXPORTED_REPORT_BYTES} bytes ({len(data)} bytes)" + ) + return data, None + + +def _markdown_text(path: str, payload: bytes) -> tuple[str | None, str | None]: + if not payload: + return None, f"Markdown file is empty: {path}" + if len(payload) > _MAX_EXPORTED_REPORT_BYTES: + return None, ( + f"Markdown file exceeds {_MAX_EXPORTED_REPORT_BYTES} bytes ({len(payload)} bytes)" + ) + if b"\x00" in payload: + return None, "path does not contain a markdown text file" + try: + return payload.decode("utf-8"), None + except UnicodeDecodeError: + return None, "Markdown file is not valid UTF-8 text" + + +async def _export_markdown_report(ctx: RunContextWrapper, path: str) -> dict[str, Any]: + """Copy one named markdown report from the sandbox into the host run dir.""" + ready = _export_destination(ctx, path) + if isinstance(ready, dict): + return ready + session, run_dir, workspace_path, name = ready + destination = run_dir / name + if destination.exists(): + return { + "success": False, + "error": ( + f"'{name}' already exists in the host run directory; " + "choose another markdown filename" + ), + } + + payload, read_error = await _read_workspace_markdown(session, workspace_path) + if payload is None: + return {"success": False, "error": read_error} + text, text_error = _markdown_text(workspace_path, payload) + if text is None: + return {"success": False, "error": text_error} + + try: + atomic_write_text(destination, text) + except OSError as exc: + logger.exception("export_markdown_report write failed") + return { + "success": False, + "error": f"Failed to write '{name}' into the host run directory: {exc!s}", + } + + logger.info("Exported sandbox markdown report to %s", destination) + return { + "success": True, + "message": ( + f"Copied {workspace_path} to the host run directory as {name}, " + "beside findings.sarif and penetration_test_report.md" + ), + "filename": name, + "bytes": len(payload), + } + + +@function_tool(timeout=60) +async def export_markdown_report(ctx: RunContextWrapper, path: str) -> str: + """Copy a markdown report from the sandbox workspace into the host run directory. + + The sandbox ``/workspace`` is not visible on the host. Use this after you + have written a named ``.md`` report there (for example + ``/workspace/report.md``) so the user can retrieve it beside + ``findings.sarif`` and ``penetration_test_report.md``. + + The file is copied under its own basename. It must be a non-empty UTF-8 + markdown file inside ``/workspace``. Harness-owned names such as + ``penetration_test_report.md`` and ``findings.sarif`` are rejected. + + Args: + path: Absolute sandbox path of the markdown report, such as + ``/workspace/report.md`` or ``/workspace/notes/final-report.md``. + """ + return json.dumps( + await _export_markdown_report(ctx, path), + ensure_ascii=False, + default=str, + ) + + @function_tool(timeout=30) async def list_reports( ctx: RunContextWrapper, diff --git a/tests/test_export_markdown_report.py b/tests/test_export_markdown_report.py new file mode 100644 index 00000000..e600caf1 --- /dev/null +++ b/tests/test_export_markdown_report.py @@ -0,0 +1,214 @@ +"""Copy a named sandbox markdown report into the host run directory.""" + +from __future__ import annotations + +import io +import json +from typing import TYPE_CHECKING, Any + +import pytest +from agents.sandbox.files import EntryKind, FileEntry +from agents.sandbox.types import Permissions +from agents.tool_context import ToolContext + +from strix.report.state import ReportState, set_global_report_state +from strix.tools.reporting.tool import _MAX_EXPORTED_REPORT_BYTES, export_markdown_report + + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + +def _file_entry(path: str, payload: bytes, *, kind: EntryKind = EntryKind.FILE) -> FileEntry: + return FileEntry( + path=path, + permissions=Permissions.from_mode(0o100644), + owner="root", + group="root", + size=len(payload), + kind=kind, + ) + + +class _SimpleEntry: + def __init__(self, path: str, *, kind: str = "file", size: int | None = None) -> None: + self.path = path + self.kind = kind + if size is not None: + self.size = size + + +class _Sandbox: + def __init__( + self, + files: dict[str, bytes], + *, + kinds: dict[str, EntryKind] | None = None, + listed: dict[str, object] | None = None, + ) -> None: + self.files = files + self.kinds = kinds or {} + self.listed = listed or {} + self.reads: list[str] = [] + self.bytes_read = 0 + + async def ls(self, path: Path) -> list[object]: + parent = path.as_posix().rstrip("/") or "/" + entries: list[object] = [] + for file_path, payload in self.files.items(): + if file_path.rsplit("/", 1)[0] == parent: + entries.append( + self.listed.get(file_path) + or _file_entry( + file_path, payload, kind=self.kinds.get(file_path, EntryKind.FILE) + ) + ) + return entries + + async def read(self, path: Path) -> io.BytesIO: + self.reads.append(path.as_posix()) + payload = self.files.get(path.as_posix()) + if payload is None: + raise FileNotFoundError(path.as_posix()) + + sandbox = self + + class _CountedBytes(io.BytesIO): + def read(self, size: int | None = -1) -> bytes: + chunk = super().read(-1 if size is None else size) + sandbox.bytes_read += len(chunk) + return chunk + + return _CountedBytes(payload) + + +@pytest.fixture +def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[ReportState]: + monkeypatch.chdir(tmp_path) + state = ReportState(run_name="test-run") + set_global_report_state(state) + yield state + set_global_report_state(None) + + +async def _export(context: dict[str, Any], path: str) -> dict[str, Any]: + ctx = ToolContext( + context=context, + tool_name="export_markdown_report", + tool_call_id="call-1", + tool_arguments="{}", + ) + raw = await export_markdown_report.on_invoke_tool(ctx, json.dumps({"path": path})) + return json.loads(raw) # type: ignore[no-any-return] + + +@pytest.mark.asyncio +async def test_copies_named_markdown_beside_existing_run_artifacts( + report_state: ReportState, +) -> None: + run_dir = report_state.get_run_dir() + (run_dir / "findings.sarif").write_text("{}", encoding="utf-8") + (run_dir / "penetration_test_report.md").write_text("# executive\n", encoding="utf-8") + sandbox = _Sandbox({"/workspace/notes/report.md": b"# Findings\n\nUse the patch.\n"}) + + result = await _export( + {"sandbox_session": sandbox}, + "/workspace/notes/report.md", + ) + + exported = run_dir / "report.md" + assert result["success"] is True + assert result["filename"] == "report.md" + assert sandbox.reads == ["/workspace/notes/report.md"] + assert exported.read_text(encoding="utf-8") == "# Findings\n\nUse the patch.\n" + assert (run_dir / "findings.sarif").read_text(encoding="utf-8") == "{}" + assert (run_dir / "penetration_test_report.md").read_text(encoding="utf-8") == "# executive\n" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "path", + [ + "/etc/report.md", + "/workspace/../report.md", + "/workspace/notes.txt", + "/workspace/penetration_test_report.md", + "report.md", + ], +) +async def test_rejects_paths_outside_a_named_workspace_markdown_file( + report_state: ReportState, + path: str, +) -> None: + sandbox = _Sandbox({"/workspace/report.md": b"# ok\n"}) + + result = await _export({"sandbox_session": sandbox}, path) + + assert result["success"] is False + assert sandbox.reads == [] + assert not (report_state.get_run_dir() / "report.md").exists() + assert not (report_state.get_run_dir() / "penetration_test_report.md").exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("payload", "kind", "match"), + [ + (b"", EntryKind.FILE, "empty"), + (b"# Findings\n\x00hidden\n", EntryKind.FILE, "markdown text file"), + (b"\xff\xfe# Findings\n", EntryKind.FILE, "UTF-8"), + (b"x" * (_MAX_EXPORTED_REPORT_BYTES + 1), EntryKind.FILE, "exceeds"), + (b"# Findings\n", EntryKind.SYMLINK, "symlink"), + ], +) +async def test_rejects_empty_invalid_oversized_and_symlink_reports( + report_state: ReportState, + payload: bytes, + kind: EntryKind, + match: str, +) -> None: + sandbox = _Sandbox({"/workspace/report.md": payload}, kinds={"/workspace/report.md": kind}) + + result = await _export({"sandbox_session": sandbox}, "/workspace/report.md") + + assert result["success"] is False + assert match in result["error"] + assert not (report_state.get_run_dir() / "report.md").exists() + if kind is EntryKind.SYMLINK or match in {"empty", "exceeds"}: + assert sandbox.reads == [] + if match == "exceeds": + assert sandbox.bytes_read == 0 + + +@pytest.mark.asyncio +async def test_size_limit_stops_the_sandbox_read_after_one_extra_byte( + report_state: ReportState, +) -> None: + oversized = b"x" * (_MAX_EXPORTED_REPORT_BYTES + 64) + sandbox = _Sandbox( + {"/workspace/report.md": oversized}, + listed={"/workspace/report.md": _SimpleEntry("/workspace/report.md")}, + ) + + result = await _export({"sandbox_session": sandbox}, "/workspace/report.md") + + assert result["success"] is False + assert "exceeds" in result["error"] + assert sandbox.bytes_read == _MAX_EXPORTED_REPORT_BYTES + 1 + assert not (report_state.get_run_dir() / "report.md").exists() + + +@pytest.mark.asyncio +async def test_same_named_exports_do_not_overwrite_an_existing_report( + report_state: ReportState, +) -> None: + run_dir = report_state.get_run_dir() + (run_dir / "report.md").write_text("# first\n", encoding="utf-8") + sandbox = _Sandbox({"/workspace/notes/report.md": b"# second\n"}) + + result = await _export({"sandbox_session": sandbox}, "/workspace/notes/report.md") + + assert result["success"] is False + assert "already exists" in result["error"] + assert (run_dir / "report.md").read_text(encoding="utf-8") == "# first\n"