diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 063a5dd4..e3d9c248 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, @@ -593,6 +594,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 45f4f658..c2a6abf7 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]: @@ -2482,6 +2499,166 @@ 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 _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 + + +async def _read_workspace_markdown(session: Any, path: str) -> tuple[bytes | None, str | None]: + 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}" + try: + data = stream.read() + finally: + close = getattr(stream, "close", None) + if callable(close): + close() + if isinstance(data, str): + return data.encode("utf-8"), None + if isinstance(data, bytes | bytearray): + return bytes(data), None + return None, "sandbox read did not return file bytes" + + +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 + + 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} + + destination = run_dir / name + 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..ae16cf89 --- /dev/null +++ b/tests/test_export_markdown_report.py @@ -0,0 +1,99 @@ +"""Copy a named sandbox markdown report into the host run directory.""" + +from __future__ import annotations + +import io +import json +from collections.abc import Iterator +from typing import TYPE_CHECKING, Any + +import pytest +from agents.tool_context import ToolContext + +from strix.report.state import ReportState, set_global_report_state +from strix.tools.reporting.tool import export_markdown_report + + +if TYPE_CHECKING: + from pathlib import Path + + +class _Sandbox: + def __init__(self, files: dict[str, bytes]) -> None: + self.files = files + self.reads: list[str] = [] + + 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()) + return io.BytesIO(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()