Sanitise the diff and explain markdown content like we do with README
This commit is contained in:
parent
a34813e788
commit
17f771cadd
GPG key ID:
03906B4110AAD3B8
5 changed files with 325 additions and 63 deletions
26
enroll/cm.py
26
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.
|
||||
|
||||
|
|
|
|||
123
enroll/diff.py
123
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(
|
||||
[
|
||||
|
|
|
|||
|
|
@ -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)")
|
||||
|
||||
|
|
|
|||
114
tests/test_diff_report_sanitize.py
Normal file
114
tests/test_diff_report_sanitize.py
Normal file
|
|
@ -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
|
||||
84
tests/test_explain_sanitize.py
Normal file
84
tests/test_explain_sanitize.py
Normal file
|
|
@ -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")
|
||||
Loading…
Add table
Add a link
Reference in a new issue