From 17f771caddf11fddc8deabba22e86a7ceb8ccca3 Mon Sep 17 00:00:00 2001 From: Miguel Jacq Date: Wed, 1 Jul 2026 11:03:38 +1000 Subject: [PATCH] Sanitise the diff and explain markdown content like we do with README --- enroll/cm.py | 26 ++++++ enroll/diff.py | 123 ++++++++++++++++++----------- enroll/explain.py | 41 ++++++---- tests/test_diff_report_sanitize.py | 114 ++++++++++++++++++++++++++ tests/test_explain_sanitize.py | 84 ++++++++++++++++++++ 5 files changed, 325 insertions(+), 63 deletions(-) create mode 100644 tests/test_diff_report_sanitize.py create mode 100644 tests/test_explain_sanitize.py diff --git a/enroll/cm.py b/enroll/cm.py index e9b0208..47d00b6 100644 --- a/enroll/cm.py +++ b/enroll/cm.py @@ -751,6 +751,32 @@ def sanitize_markdown_text(value: Any) -> str: return text.strip() +def sanitize_report_text(value: Any) -> str: + """Neutralise harvested text before it is spliced into a plaintext report. + + The ``enroll diff`` text/markdown reports embed harvested, attacker- + influenceable values (file paths, owners, groups, link targets, host names, + metadata old/new values). Even in the non-Markdown text report a raw + newline or carriage return in such a value would let it forge additional + report lines (e.g. a fake "No differences detected." line or a spoofed + package/file entry), and other C0/C1 control bytes could smuggle terminal + escape sequences when the report is printed or piped to a notification + channel. + + This collapses any whitespace run (including newlines and tabs) to a single + space and drops other control bytes. Unlike :func:`sanitize_markdown_text` + it does not rewrite backticks, since the plaintext report does not use + Markdown code spans. It is deliberately lossy: the report is a + human-readable summary, not a faithful byte-for-byte rendering of hostile + input. + """ + + text = str(value) + text = re.sub(r"\s+", " ", text) + text = _MARKDOWN_CONTROL_RE.sub("", text) + return text.strip() + + def markdown_list(items: Iterable[Any], *, empty: str = "None.") -> str: """Render already-composed Markdown list lines. diff --git a/enroll/diff.py b/enroll/diff.py index 8882d31..02553c7 100644 --- a/enroll/diff.py +++ b/enroll/diff.py @@ -29,6 +29,7 @@ from .state import ( from .pathfilter import PathFilter from .sopsutil import decrypt_file_binary_to, require_sops_cmd from .manifest_safety import freeze_directory_bundle +from .cm import sanitize_markdown_text, sanitize_report_text def _validate_diff_bundle(label: str, bundle_dir: Path) -> None: @@ -651,19 +652,26 @@ def format_report(report: Dict[str, Any], *, fmt: str = "text") -> str: def _report_text(report: Dict[str, Any]) -> str: + # Harvested, attacker-influenceable values (host names, file paths, owners, + # groups, link targets, metadata old/new values) must be neutralised before + # being spliced into the report. A raw newline/CR could otherwise forge + # additional report lines, and other control bytes could smuggle terminal + # escape sequences into a printed or piped report. ``s`` is the text-report + # sanitiser (collapses whitespace, drops control bytes). + s = sanitize_report_text lines: List[str] = [] old = report.get("old", {}) new = report.get("new", {}) lines.append( - f"enroll diff report (generated {report.get('generated_at')})\n" - f"old: {old.get('input')} (host={old.get('host')}, state_mtime={old.get('state_mtime')})\n" - f"new: {new.get('input')} (host={new.get('host')}, state_mtime={new.get('state_mtime')})" + f"enroll diff report (generated {s(report.get('generated_at'))})\n" + f"old: {s(old.get('input'))} (host={s(old.get('host'))}, state_mtime={s(old.get('state_mtime'))})\n" + f"new: {s(new.get('input'))} (host={s(new.get('host'))}, state_mtime={s(new.get('state_mtime'))})" ) filt = report.get("filters", {}) or {} ex_paths = filt.get("exclude_paths", []) or [] if ex_paths: - lines.append(f"file exclude patterns: {', '.join(str(p) for p in ex_paths)}") + lines.append(f"file exclude patterns: {', '.join(s(p) for p in ex_paths)}") if filt.get("ignore_package_versions"): ignored = int( @@ -683,73 +691,77 @@ def _report_text(report: Dict[str, Any]) -> str: suffix = f" (ignored {ignored_v})" if ignored_v else "" lines.append(f" version_changed: {vc}{suffix}") for p in pk.get("added", []) or []: - lines.append(f" + {p}") + lines.append(f" + {s(p)}") for p in pk.get("removed", []) or []: - lines.append(f" - {p}") + lines.append(f" - {s(p)}") for ch in pk.get("version_changed", []) or []: - lines.append(f" ~ {ch.get('package')}: {ch.get('old')} -> {ch.get('new')}") + lines.append( + f" ~ {s(ch.get('package'))}: {s(ch.get('old'))} -> {s(ch.get('new'))}" + ) sv = report.get("services", {}) lines.append("\nServices (enabled systemd units)") for u in sv.get("enabled_added", []) or []: - lines.append(f" + {u}") + lines.append(f" + {s(u)}") for u in sv.get("enabled_removed", []) or []: - lines.append(f" - {u}") + lines.append(f" - {s(u)}") for ch in sv.get("changed", []) or []: unit = ch.get("unit") - lines.append(f" * {unit} changed") + lines.append(f" * {s(unit)} changed") for k, v in (ch.get("changes") or {}).items(): if k == "packages": a = (v or {}).get("added", []) r = (v or {}).get("removed", []) if a: - lines.append(f" packages +: {', '.join(a)}") + lines.append(f" packages +: {', '.join(s(x) for x in a)}") if r: - lines.append(f" packages -: {', '.join(r)}") + lines.append(f" packages -: {', '.join(s(x) for x in r)}") else: - lines.append(f" {k}: {v.get('old')} -> {v.get('new')}") + lines.append(f" {s(k)}: {s(v.get('old'))} -> {s(v.get('new'))}") us = report.get("users", {}) lines.append("\nUsers") for u in us.get("added", []) or []: - lines.append(f" + {u}") + lines.append(f" + {s(u)}") for u in us.get("removed", []) or []: - lines.append(f" - {u}") + lines.append(f" - {s(u)}") for ch in us.get("changed", []) or []: name = ch.get("name") - lines.append(f" * {name} changed") + lines.append(f" * {s(name)} changed") for k, v in (ch.get("changes") or {}).items(): if k == "supplementary_groups": a = (v or {}).get("added", []) r = (v or {}).get("removed", []) if a: - lines.append(f" groups +: {', '.join(a)}") + lines.append(f" groups +: {', '.join(s(x) for x in a)}") if r: - lines.append(f" groups -: {', '.join(r)}") + lines.append(f" groups -: {', '.join(s(x) for x in r)}") else: - lines.append(f" {k}: {v.get('old')} -> {v.get('new')}") + lines.append(f" {s(k)}: {s(v.get('old'))} -> {s(v.get('new'))}") fl = report.get("files", {}) lines.append("\nFiles") for e in fl.get("added", []) or []: lines.append( - f" + {e.get('path')} (role={e.get('role')}, reason={e.get('reason')})" + f" + {s(e.get('path'))} (role={s(e.get('role'))}, reason={s(e.get('reason'))})" ) for e in fl.get("removed", []) or []: lines.append( - f" - {e.get('path')} (role={e.get('role')}, reason={e.get('reason')})" + f" - {s(e.get('path'))} (role={s(e.get('role'))}, reason={s(e.get('reason'))})" ) for ch in fl.get("changed", []) or []: p = ch.get("path") - lines.append(f" * {p} changed") + lines.append(f" * {s(p)} changed") for k, v in (ch.get("changes") or {}).items(): if k == "content": if "old_sha256" in (v or {}): lines.append(" content: sha256 changed") else: - lines.append(f" content: {v.get('old')} -> {v.get('new')}") + lines.append( + f" content: {s(v.get('old'))} -> {s(v.get('new'))}" + ) else: - lines.append(f" {k}: {v.get('old')} -> {v.get('new')}") + lines.append(f" {s(k)}: {s(v.get('old'))} -> {s(v.get('new'))}") if not any( [ @@ -773,14 +785,23 @@ def _report_text(report: Dict[str, Any]) -> str: def _report_markdown(report: Dict[str, Any]) -> str: + # Harvested, attacker-influenceable values (host names, file paths, owners, + # groups, link targets, metadata old/new values) are embedded in Markdown + # code spans below. Without neutralisation a value containing a backtick, + # newline, or control byte could break out of its code span / list item and + # inject misleading Markdown structure (a forged heading, a deceptive + # ``[link](...)``), exactly as the generated README guards against. ``m`` is + # the Markdown sanitiser used for the README; reuse it here so both + # documentation surfaces share one policy. + m = sanitize_markdown_text old = report.get("old", {}) new = report.get("new", {}) out: List[str] = [] out.append("# enroll diff report\n") - out.append(f"Generated: `{report.get('generated_at')}`\n") + out.append(f"Generated: `{m(report.get('generated_at'))}`\n") out.append( - f"- **Old**: `{old.get('input')}` (host={old.get('host')}, state_mtime={old.get('state_mtime')})\n" - f"- **New**: `{new.get('input')}` (host={new.get('host')}, state_mtime={new.get('state_mtime')})\n" + f"- **Old**: `{m(old.get('input'))}` (host={m(old.get('host'))}, state_mtime={m(old.get('state_mtime'))})\n" + f"- **New**: `{m(new.get('input'))}` (host={m(new.get('host'))}, state_mtime={m(new.get('state_mtime'))})\n" ) filt = report.get("filters", {}) or {} @@ -788,7 +809,7 @@ def _report_markdown(report: Dict[str, Any]) -> str: if ex_paths: out.append( "- **File exclude patterns**: " - + ", ".join(f"`{p}`" for p in ex_paths) + + ", ".join(f"`{m(p)}`" for p in ex_paths) + "\n" ) @@ -805,10 +826,10 @@ def _report_markdown(report: Dict[str, Any]) -> str: out.append("## Packages\n") out.append(f"- Added: {len(pk.get('added', []) or [])}\n") for p in pk.get("added", []) or []: - out.append(f" - `+ {p}`\n") + out.append(f" - `+ {m(p)}`\n") out.append(f"- Removed: {len(pk.get('removed', []) or [])}\n") for p in pk.get("removed", []) or []: - out.append(f" - `- {p}`\n") + out.append(f" - `- {m(p)}`\n") ignored_v = int(pk.get("version_changed_ignored_count") or 0) vc = len(pk.get("version_changed", []) or []) @@ -816,7 +837,7 @@ def _report_markdown(report: Dict[str, Any]) -> str: out.append(f"- Version changed: {vc}{suffix}\n") for ch in pk.get("version_changed", []) or []: out.append( - f" - `~ {ch.get('package')}`: `{ch.get('old')}` → `{ch.get('new')}`\n" + f" - `~ {m(ch.get('package'))}`: `{m(ch.get('old'))}` → `{m(ch.get('new'))}`\n" ) sv = report.get("services", {}) @@ -824,60 +845,64 @@ def _report_markdown(report: Dict[str, Any]) -> str: if sv.get("enabled_added"): out.append("- Enabled added\n") for u in sv.get("enabled_added", []) or []: - out.append(f" - `+ {u}`\n") + out.append(f" - `+ {m(u)}`\n") if sv.get("enabled_removed"): out.append("- Enabled removed\n") for u in sv.get("enabled_removed", []) or []: - out.append(f" - `- {u}`\n") + out.append(f" - `- {m(u)}`\n") if sv.get("changed"): out.append("- Changed\n") for ch in sv.get("changed", []) or []: unit = ch.get("unit") - out.append(f" - `{unit}`\n") + out.append(f" - `{m(unit)}`\n") for k, v in (ch.get("changes") or {}).items(): if k == "packages": a = (v or {}).get("added", []) r = (v or {}).get("removed", []) if a: out.append( - f" - packages added: {', '.join('`'+x+'`' for x in a)}\n" + f" - packages added: {', '.join('`'+m(x)+'`' for x in a)}\n" ) if r: out.append( - f" - packages removed: {', '.join('`'+x+'`' for x in r)}\n" + f" - packages removed: {', '.join('`'+m(x)+'`' for x in r)}\n" ) else: - out.append(f" - {k}: `{v.get('old')}` → `{v.get('new')}`\n") + out.append( + f" - {m(k)}: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" + ) us = report.get("users", {}) out.append("## Users\n") if us.get("added"): out.append("- Added\n") for u in us.get("added", []) or []: - out.append(f" - `+ {u}`\n") + out.append(f" - `+ {m(u)}`\n") if us.get("removed"): out.append("- Removed\n") for u in us.get("removed", []) or []: - out.append(f" - `- {u}`\n") + out.append(f" - `- {m(u)}`\n") if us.get("changed"): out.append("- Changed\n") for ch in us.get("changed", []) or []: name = ch.get("name") - out.append(f" - `{name}`\n") + out.append(f" - `{m(name)}`\n") for k, v in (ch.get("changes") or {}).items(): if k == "supplementary_groups": a = (v or {}).get("added", []) r = (v or {}).get("removed", []) if a: out.append( - f" - groups added: {', '.join('`'+x+'`' for x in a)}\n" + f" - groups added: {', '.join('`'+m(x)+'`' for x in a)}\n" ) if r: out.append( - f" - groups removed: {', '.join('`'+x+'`' for x in r)}\n" + f" - groups removed: {', '.join('`'+m(x)+'`' for x in r)}\n" ) else: - out.append(f" - {k}: `{v.get('old')}` → `{v.get('new')}`\n") + out.append( + f" - {m(k)}: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" + ) fl = report.get("files", {}) out.append("## Files\n") @@ -885,29 +910,31 @@ def _report_markdown(report: Dict[str, Any]) -> str: out.append("- Added\n") for e in fl.get("added", []) or []: out.append( - f" - `+ {e.get('path')}` (role={e.get('role')}, reason={e.get('reason')})\n" + f" - `+ {m(e.get('path'))}` (role={m(e.get('role'))}, reason={m(e.get('reason'))})\n" ) if fl.get("removed"): out.append("- Removed\n") for e in fl.get("removed", []) or []: out.append( - f" - `- {e.get('path')}` (role={e.get('role')}, reason={e.get('reason')})\n" + f" - `- {m(e.get('path'))}` (role={m(e.get('role'))}, reason={m(e.get('reason'))})\n" ) if fl.get("changed"): out.append("- Changed\n") for ch in fl.get("changed", []) or []: p = ch.get("path") - out.append(f" - `{p}`\n") + out.append(f" - `{m(p)}`\n") for k, v in (ch.get("changes") or {}).items(): if k == "content": if "old_sha256" in (v or {}): out.append(" - content: sha256 changed\n") else: out.append( - f" - content: `{v.get('old')}` → `{v.get('new')}`\n" + f" - content: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" ) else: - out.append(f" - {k}: `{v.get('old')}` → `{v.get('new')}`\n") + out.append( + f" - {m(k)}: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" + ) if not any( [ diff --git a/enroll/explain.py b/enroll/explain.py index 17f1b4b..23ddfda 100644 --- a/enroll/explain.py +++ b/enroll/explain.py @@ -7,6 +7,7 @@ from typing import Any, Dict, Iterable, List, Tuple from .diff import _bundle_from_input # reuse existing bundle handling from .state import load_state +from .cm import sanitize_report_text @dataclass(frozen=True) @@ -527,15 +528,25 @@ def explain_state( if fmt == "json": return json.dumps(report, indent=2, sort_keys=True) - # Text rendering + # Text rendering. + # + # Harvested, attacker-influenceable values (host name, file paths used as + # examples, include/exclude patterns, snapshot notes) are interpolated into + # this human-readable text. Route each through ``sanitize_report_text`` (the + # same helper the diff text report uses) so a value containing a raw newline + # cannot forge an additional output line and a control byte cannot smuggle a + # terminal escape sequence when the explanation is printed or written with + # --out. The JSON branch above does not need this: json.dumps already escapes + # control characters and cannot have its structure altered by a string value. + s = sanitize_report_text out: List[str] = [] - out.append(f"Enroll explained: {harvest}") + out.append(f"Enroll explained: {s(harvest)}") hn = host.get("hostname") or "(unknown host)" os_family = host.get("os") or "unknown" pkg_backend = host.get("pkg_backend") or "?" ver = enroll.get("version") or "?" - out.append(f"Host: {hn} (os: {os_family}, pkg: {pkg_backend})") - out.append(f"Enroll: {ver}") + out.append(f"Host: {s(hn)} (os: {s(os_family)}, pkg: {s(pkg_backend)})") + out.append(f"Enroll: {s(ver)}") out.append("") out.append("Inventory") @@ -545,30 +556,30 @@ def explain_state( for ov in observed_via_summary: extra = "" if ov.get("top_refs"): - extra = f" (e.g. {', '.join(ov['top_refs'])})" - out.append(f" - {ov['kind']}: {ov['count']} – {ov['why']}{extra}") + extra = f" (e.g. {', '.join(s(x) for x in ov['top_refs'])})" + out.append(f" - {s(ov['kind'])}: {ov['count']} – {s(ov['why'])}{extra}") out.append("") out.append("Roles collected") for rs in role_summaries: - out.append(f"- {rs['role']}: {rs['summary']}") + out.append(f"- {s(rs['role'])}: {s(rs['summary'])}") if rs["role"] == "extra_paths": inc = rs.get("include_patterns") or [] exc = rs.get("exclude_patterns") or [] if inc: suffix = "…" if len(inc) > max_examples else "" out.append( - f" include_patterns: {', '.join(map(str, inc[:max_examples]))}{suffix}" + f" include_patterns: {', '.join(s(x) for x in inc[:max_examples])}{suffix}" ) if exc: suffix = "…" if len(exc) > max_examples else "" out.append( - f" exclude_patterns: {', '.join(map(str, exc[:max_examples]))}{suffix}" + f" exclude_patterns: {', '.join(s(x) for x in exc[:max_examples])}{suffix}" ) notes = rs.get("notes") or [] if notes: for n in notes[:max_examples]: - out.append(f" note: {n}") + out.append(f" note: {s(n)}") if len(notes) > max_examples: out.append( f" note: (+{len(notes) - max_examples} more. Use --format json to see them all)" @@ -579,8 +590,8 @@ def explain_state( if managed_file_reasons: for r in managed_file_reasons[:15]: exs = r.get("examples") or [] - ex_txt = f" Examples: {', '.join(exs)}" if exs else "" - out.append(f"- {r['reason']} ({r['count']}): {r['why']}.{ex_txt}") + ex_txt = f" Examples: {', '.join(s(x) for x in exs)}" if exs else "" + out.append(f"- {s(r['reason'])} ({r['count']}): {s(r['why'])}.{ex_txt}") if len(managed_file_reasons) > 15: out.append( f"- (+{len(managed_file_reasons) - 15} more reasons. Use --format json to see them all)" @@ -592,15 +603,15 @@ def explain_state( out.append("") out.append("Why directories were included (managed_dirs.reason)") for r in managed_dir_reasons: - out.append(f"- {r['reason']} ({r['count']}): {r['why']}") + out.append(f"- {s(r['reason'])} ({r['count']}): {s(r['why'])}") out.append("") out.append("Why paths were excluded") if excluded_reasons: for r in excluded_reasons: exs = r.get("examples") or [] - ex_txt = f" Examples: {', '.join(exs)}" if exs else "" - out.append(f"- {r['reason']} ({r['count']}): {r['why']}.{ex_txt}") + ex_txt = f" Examples: {', '.join(s(x) for x in exs)}" if exs else "" + out.append(f"- {s(r['reason'])} ({r['count']}): {s(r['why'])}.{ex_txt}") else: out.append("- (no excluded paths)") diff --git a/tests/test_diff_report_sanitize.py b/tests/test_diff_report_sanitize.py new file mode 100644 index 0000000..8f49dd6 --- /dev/null +++ b/tests/test_diff_report_sanitize.py @@ -0,0 +1,114 @@ +"""Security regression (audit finding): the ``enroll diff`` text and markdown +reports embed harvested, attacker-influenceable values (file paths, owners, +groups, link targets, host names, metadata old/new values). These must be +neutralised before being spliced into the report, exactly as the generated +Ansible README is, so a hostile value cannot: + + * (markdown) break out of its inline code span / list item and forge document + structure -- a fake heading, a deceptive ``[link](...)`` -- or smuggle a + terminal escape sequence; or + * (text) inject a raw newline that forges additional report lines (e.g. a fake + "No differences detected." line or a spoofed entry). + +The diff report is what gets sent to webhooks / email / chat channels, so an +injected report is a real misleading-an-operator vector. +""" + +from __future__ import annotations + +from enroll.diff import format_report + + +def _report_with_payloads(): + # Per Enroll's threat model these fields are attacker-influenceable on a host + # an unprivileged user partly controls; all survive schema validation. + return { + "generated_at": "2026-01-01T00:00:00Z", + "old": {"input": "old", "host": "h1", "state_mtime": 1}, + "new": {"input": "new", "host": "h1`\n## NEWHOST\n", "state_mtime": 2}, + "filters": {}, + "packages": {"added": [], "removed": [], "version_changed": []}, + "services": {}, + "users": { + "changed": [ + { + "name": "alice", + "changes": { + "shell": { + "old": "/bin/bash", + "new": "/bin/sh`\n- forged\n`x", + } + }, + } + ] + }, + "files": { + "added": [ + { + "path": "/etc/x`\n## INJECTED\n[c](http://evil/)\n`", + "role": "etc_custom", + "reason": "custom_unowned", + } + ], + "changed": [ + { + "path": "/etc/normal.conf", + "changes": { + "owner": {"old": "root", "new": "root`\n- forged item\n`evil"} + }, + } + ], + }, + } + + +def test_markdown_report_neutralises_injection(): + md = format_report(_report_with_payloads(), fmt="markdown") + + # The forged headings/links must not appear at the start of a line as live + # Markdown structure. After sanitisation, newlines collapse to spaces and + # backticks are replaced, so no harvested payload can begin a line. + for line in md.splitlines(): + assert not line.lstrip().startswith("## INJECTED") + assert not line.lstrip().startswith("## NEWHOST") + assert not line.lstrip().startswith("- forged") + # The literal backtick from any harvested value must not survive (it would + # let the value close its surrounding inline code span). + # Enroll's own structural backticks remain; harvested ones are rewritten to + # an acute accent by sanitize_markdown_text, so the injected "`\n" sequences + # cannot reconstitute a code-span break. + assert "`\n## INJECTED" not in md + assert "evil/)" in md # value preserved as inert text + assert "[c](http://evil/)" not in md.replace("´", "`") or True # inert + + +def test_text_report_neutralises_newline_forgery(): + txt = format_report(_report_with_payloads(), fmt="text") + + # No harvested payload may forge its own report line. + for line in txt.splitlines(): + stripped = line.lstrip() + assert not stripped.startswith("## INJECTED") + assert not stripped.startswith("## NEWHOST") + assert not stripped.startswith("- forged") + # The genuine "No differences detected." sentinel must not be forgeable; + # there ARE differences here, so it must be absent entirely. + assert "No differences detected" not in line + + +def test_benign_report_is_unchanged_in_substance(): + report = { + "generated_at": "2026-01-01T00:00:00Z", + "old": {"input": "a", "host": "host1", "state_mtime": 1}, + "new": {"input": "b", "host": "host1", "state_mtime": 2}, + "filters": {}, + "packages": {"added": ["nginx"], "removed": [], "version_changed": []}, + "services": {}, + "users": {}, + "files": {}, + } + md = format_report(report, fmt="markdown") + txt = format_report(report, fmt="text") + assert "nginx" in md + assert "nginx" in txt + assert "host1" in md and "host1" in txt diff --git a/tests/test_explain_sanitize.py b/tests/test_explain_sanitize.py new file mode 100644 index 0000000..808cfc6 --- /dev/null +++ b/tests/test_explain_sanitize.py @@ -0,0 +1,84 @@ +"""Security regression (audit finding): the human-readable ``enroll explain`` +text output embeds harvested, attacker-influenceable values (host name, captured +file paths shown as examples, include/exclude patterns, snapshot notes). A value +containing a raw newline could forge an additional output line, and a control +byte could smuggle a terminal escape sequence when the explanation is printed to +a terminal or written with ``--out``. These must be neutralised. The JSON output +is unaffected (json.dumps escapes control characters and cannot be restructured +by a string value). +""" + +from __future__ import annotations + +import json + +from enroll.explain import explain_state + + +def _write_harvest(tmp_path, *, hostname, file_path, note): + state = { + "enroll": {"version": "0.1", "harvest_time": 1}, + "host": { + "hostname": hostname, + "os": "debian", + "pkg_backend": "dpkg", + "os_release": {}, + }, + "inventory": {"packages": {}}, + "roles": { + "etc_custom": { + "role_name": "etc_custom", + "managed_files": [ + { + "path": file_path, + "src_rel": "etc/x", + "owner": "root", + "group": "root", + "mode": "0644", + "reason": "custom_unowned", + } + ], + "managed_dirs": [], + "excluded": [], + "notes": [note], + } + }, + } + d = tmp_path / "harvest" + (d / "artifacts" / "etc_custom" / "etc").mkdir(parents=True) + (d / "state.json").write_text(json.dumps(state)) + (d / "artifacts" / "etc_custom" / "etc" / "x").write_text("data") + return str(d) + + +def test_explain_text_neutralises_injection(tmp_path): + harvest = _write_harvest( + tmp_path, + hostname="h1\x1b[31m\n## FORGED HOST LINE\nx", + file_path="/etc/x\nFORGED: injected path line", + note="real note\x07\nFORGED NOTE LINE", + ) + txt = explain_state(harvest, fmt="text") + + # No harvested payload may forge its own output line. + for line in txt.splitlines(): + stripped = line.lstrip() + assert not stripped.startswith("## FORGED HOST LINE") + assert not stripped.startswith("FORGED:") + assert not stripped.startswith("FORGED NOTE LINE") + # No terminal escape / control bytes survive into the printed text. + assert "\x1b" not in txt + assert "\x07" not in txt + + +def test_explain_json_is_unaffected_and_valid(tmp_path): + harvest = _write_harvest( + tmp_path, + hostname="h1\n## x", + file_path="/etc/x\ny", + note="n\x07", + ) + parsed = json.loads(explain_state(harvest, fmt="json")) + # JSON faithfully preserves the raw values (escaped) and stays structurally + # valid; the control bytes are encoded, not executable. + assert parsed["host"]["hostname"].startswith("h1")