feat(reporting): export sandbox markdown reports

This commit is contained in:
yusheng.wang 2026-09-29 16:33:00 +08:00
parent 0c702723aa
commit cb849e5f49
3 changed files with 279 additions and 1 deletions

View file

@ -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,

View file

@ -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,

View file

@ -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()