fix(batch): prevent Markdown report injection

contrib/batch_scan/reports.py

@@ -23,11 +23,14 @@
 from __future__ import annotations
 
 import json
+import re
 from collections import defaultdict
 from datetime import UTC, datetime
+from html import escape as escape_html
 from io import StringIO
 
 from skillspector import __version__ as _skillspector_version
+from skillspector.nodes.report import _clean_text
 
 
 def sorted_results(results: list[dict[str, object]]) -> list[dict[str, object]]:
@@ -373,6 +376,29 @@ def _format_json(results: list[dict[str, object]]) -> str:
 # ═══════════════════════════════════════════════════════════════════
 
 
+def _markdown_plain_text(value: object) -> str:
+    """Remove display controls and keep a Markdown field on one line."""
+    text = _clean_text(str(value)) or ""
+    text = "".join(c for c in text if c.isprintable() or c.isspace())
+    return " ".join(text.splitlines())
+
+
+def _markdown_text(value: object) -> str:
+    """Keep scan-derived prose within its own Markdown line."""
+    text = escape_html(_markdown_plain_text(value), quote=False)
+    return re.sub(r"([\\`*_{}\[\]()#+.!|>~-])", r"\\\1", text)
+
+
+def _markdown_code(value: object, *, table_cell: bool = False) -> str:
+    """Quote literal text without allowing code-span or table-cell breakouts."""
+    text = _markdown_plain_text(value)
+    if table_cell:
+        text = text.replace("|", "\\|")
+    delimiter = "`" * (max((len(run) for run in re.findall(r"`+", text)), default=0) + 1)
+    padding = " " if text.startswith(("`", " ")) or text.endswith(("`", " ")) else ""
+    return f"{delimiter}{padding}{text}{padding}{delimiter}"
+
+
 def _format_markdown(results: list[dict[str, object]]) -> str:
     lines: list[str] = []
     total = len(results)
@@ -438,9 +464,12 @@ def _format_markdown(results: list[dict[str, object]]) -> str:
         lang = skill.get("language", "en")
 
         if r.get("error"):
-            lines.append(f"| `{name}` | ERR | ERROR | — | {lang} |")
+            lines.append(f"| {_markdown_code(name, table_cell=True)} | ERR | ERROR | — | {_markdown_text(lang)} |")
         else:
-            lines.append(f"| `{name}` | {score}/100 | {sev} | {issues} | {lang} |")
+            lines.append(
+                f"| {_markdown_code(name, table_cell=True)} | {score}/100 | {_markdown_text(sev)} | "
+                f"{issues} | {_markdown_text(lang)} |"
+            )
     lines.append("")
 
     # ── Issue details for HIGH / CRITICAL ────────────────────────
@@ -458,8 +487,8 @@ def _format_markdown(results: list[dict[str, object]]) -> str:
             risk = r.get("risk_assessment", {})
             name = skill.get("name", "?")
             lines.append(
-                f"### {name} — {risk.get('score', 0)}/100 "
-                f"{risk.get('severity', 'HIGH')}\n"
+                f"### {_markdown_text(name)} — {risk.get('score', 0)}/100 "
+                f"{_markdown_text(risk.get('severity', 'HIGH'))}\n"
             )
             for issue in r.get("issues", []):
                 sev = str(issue.get("severity", "LOW")).upper()
@@ -469,26 +498,27 @@ def _format_markdown(results: list[dict[str, object]]) -> str:
                 loc_file = loc.get("file", "") if isinstance(loc, dict) else ""
                 rule_id = issue.get("id", "?")
                 explanation = issue.get("explanation", issue.get("message", ""))
-                lines.append(f"- **{emoji} {rule_id}**: {explanation}")
+                lines.append(f"- **{emoji} {_markdown_text(rule_id)}**: {_markdown_text(explanation)}")
                 if loc_file:
-                    lines.append(f"  - Location: `{loc_file}:{loc_start}`")
+                    lines.append(f"  - Location: {_markdown_code(f'{loc_file}:{loc_start}')}")
                 conf = issue.get("confidence", 0)
                 lines.append(f"  - Confidence: {float(conf):.0%}")
                 rem = issue.get("remediation")
                 if rem:
-                    lines.append(f"  - Remediation: {rem}")
+                    lines.append(f"  - Remediation: {_markdown_text(rem)}")
                 lines.append("")
         lines.append("")
 
     exception_groups = _exception_groups(results)
     if exception_groups:
         lines.append("## Ledger Exceptions\n")
         for skill_name, exceptions in exception_groups:
-            lines.append(f"### {skill_name}\n")
+            lines.append(f"### {_markdown_text(skill_name)}\n")
             for exception in exceptions:
                 lines.append(
-                    f"- **{exception.get('reason_code', 'unknown')}** "
-                    f"`{exception.get('path', '')}`: {exception.get('message', '')}"
+                    f"- **{_markdown_text(exception.get('reason_code', 'unknown'))}** "
+                    f"{_markdown_code(exception.get('path', ''))}: "
+                    f"{_markdown_text(exception.get('message', ''))}"
                 )
             lines.append("")
 

contrib/batch_scan/tests/test_inspection_reporting.py

@@ -8,6 +8,8 @@
 import json
 from pathlib import Path
 
+from markdown_it import MarkdownIt
+
 from contrib.batch_scan.reports import _format_json, _format_markdown, _format_terminal
 from contrib.batch_scan.runner import entry_from_result
 
@@ -61,6 +63,9 @@ def test_batch_formats_preserve_every_child_ledger_exception() -> None:
     assert [item["reason_code"] for item in exceptions] == ["read_error", "syntax_error"]
     assert payload["skills"][0]["issues"][0]["finding_id"] == "finding-batch-1"
 
-    for rendered in (_format_terminal([entry]), _format_markdown([entry])):
+    for rendered in (
+        _format_terminal([entry]),
+        MarkdownIt().enable("table").render(_format_markdown([entry])),
+    ):
         assert "read_error" in rendered
         assert "syntax_error" in rendered

tests/test_batch_scan_reports.py

@@ -6,8 +6,12 @@
 from __future__ import annotations
 
 import json
+from copy import deepcopy
 
-from contrib.batch_scan.reports import _format_json
+import pytest
+from markdown_it import MarkdownIt
+
+from contrib.batch_scan.reports import _format_json, _format_markdown
 
 
 def test_json_marks_error_entries_as_unsuccessful() -> None:
@@ -23,3 +27,101 @@ def test_json_marks_error_entries_as_unsuccessful() -> None:
 
     assert payload["skills"][0]["error"] == "scan crashed"
     assert payload["skills"][0]["execution_successful"] is False
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        "safe` | 0/100 | LOW | 0 | en |\n<!--",
+        "\r\n## Issues (0)\rNo security issues detected.\n<!--",
+        "` `` ``` <script>alert(1)</script> [safe](https://example.invalid)",
+        "\\| **safe** &lt;!--",
+        " leading and trailing spaces ",
+        "~~hidden~~",
+        "\x1b[2J\x00\u202eLOW\u202c\x9b\u2066safe\u2069",
+    ],
+)
+@pytest.mark.parametrize(
+    "field",
+    [
+        "name",
+        "language",
+        "id",
+        "message",
+        "explanation",
+        "remediation",
+        "file",
+        "reason_code",
+        "path",
+        "ledger_message",
+    ],
+)
+def test_batch_markdown_treats_scan_content_as_literal_text(payload: str, field: str) -> None:
+    entry = {
+        "skill": {"name": "sample", "language": "zh"},
+        "risk_assessment": {"score": 90, "severity": "CRITICAL"},
+        "issues": [
+            {
+                "id": "P1",
+                "message": "finding",
+                "remediation": "review",
+                "location": {"file": "SKILL.md", "start_line": 1},
+            }
+        ],
+        "analysis_completeness": {
+            "ledger_exceptions": [
+                {"reason_code": "partial", "path": "run.py", "message": "inspect"}
+            ]
+        },
+    }
+    parser = MarkdownIt("commonmark", {"html": True}).enable(["table", "strikethrough"])
+    original_blocks = [token.type for token in parser.parse(_format_markdown([entry]))]
+    if field in {"name", "language"}:
+        entry["skill"][field] = payload
+    elif field == "file":
+        entry["issues"][0]["location"][field] = payload
+    elif field in {"reason_code", "path", "ledger_message"}:
+        entry["analysis_completeness"]["ledger_exceptions"][0][
+            "message" if field == "ledger_message" else field
+        ] = payload
+    else:
+        entry["issues"][0][field] = payload
+    original = deepcopy(entry)
+
+    report = _format_markdown([entry])
+    tokens = parser.parse(report)
+
+    assert all(character.isprintable() or character in "\n\t" for character in report)
+    assert [token.type for token in tokens] == original_blocks
+    for token in tokens:
+        assert token.type not in {"html_block", "fence", "code_block"}
+        assert not any(
+            child.type in {"html_inline", "link_open", "image", "s_open"}
+            for child in token.children or []
+        )
+    assert entry == original
+    assert json.loads(_format_json([entry]))["skills"][0]["skill"]["name"] == entry["skill"]["name"]
+
+
+@pytest.mark.parametrize("value", ["a|b", r"a\|b", "`a`", "a``b`", "<script> & value"])
+@pytest.mark.parametrize("table_cell", [False, True])
+def test_batch_markdown_code_preserves_literal_values(value: str, table_cell: bool) -> None:
+    from contrib.batch_scan.reports import _markdown_code
+
+    source = _markdown_code(value, table_cell=table_cell)
+    if table_cell:
+        source = f"| Path |\n|---|\n| {source} |"
+    tokens = MarkdownIt("commonmark").enable("table").parse(source)
+    code = [
+        child.content
+        for token in tokens
+        for child in token.children or []
+        if child.type == "code_inline"
+    ]
+    assert code == [value]
+
+
+def test_batch_markdown_strips_complete_ansi_sequences() -> None:
+    from contrib.batch_scan.reports import _markdown_plain_text
+
+    assert _markdown_plain_text("a\x1b[2Jb\x1b[31mc\x1b[0m") == "abc"