diff --git a/enroll/cm.py b/enroll/cm.py index 47d00b6..e9b0208 100644 --- a/enroll/cm.py +++ b/enroll/cm.py @@ -751,32 +751,6 @@ 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 02553c7..8882d31 100644 --- a/enroll/diff.py +++ b/enroll/diff.py @@ -29,7 +29,6 @@ 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: @@ -652,26 +651,19 @@ 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 {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'))})" + 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')})" ) filt = report.get("filters", {}) or {} ex_paths = filt.get("exclude_paths", []) or [] if ex_paths: - lines.append(f"file exclude patterns: {', '.join(s(p) for p in ex_paths)}") + lines.append(f"file exclude patterns: {', '.join(str(p) for p in ex_paths)}") if filt.get("ignore_package_versions"): ignored = int( @@ -691,77 +683,73 @@ 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" + {s(p)}") + lines.append(f" + {p}") for p in pk.get("removed", []) or []: - lines.append(f" - {s(p)}") + lines.append(f" - {p}") for ch in pk.get("version_changed", []) or []: - lines.append( - f" ~ {s(ch.get('package'))}: {s(ch.get('old'))} -> {s(ch.get('new'))}" - ) + lines.append(f" ~ {ch.get('package')}: {ch.get('old')} -> {ch.get('new')}") sv = report.get("services", {}) lines.append("\nServices (enabled systemd units)") for u in sv.get("enabled_added", []) or []: - lines.append(f" + {s(u)}") + lines.append(f" + {u}") for u in sv.get("enabled_removed", []) or []: - lines.append(f" - {s(u)}") + lines.append(f" - {u}") for ch in sv.get("changed", []) or []: unit = ch.get("unit") - lines.append(f" * {s(unit)} changed") + lines.append(f" * {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(s(x) for x in a)}") + lines.append(f" packages +: {', '.join(a)}") if r: - lines.append(f" packages -: {', '.join(s(x) for x in r)}") + lines.append(f" packages -: {', '.join(r)}") else: - lines.append(f" {s(k)}: {s(v.get('old'))} -> {s(v.get('new'))}") + lines.append(f" {k}: {v.get('old')} -> {v.get('new')}") us = report.get("users", {}) lines.append("\nUsers") for u in us.get("added", []) or []: - lines.append(f" + {s(u)}") + lines.append(f" + {u}") for u in us.get("removed", []) or []: - lines.append(f" - {s(u)}") + lines.append(f" - {u}") for ch in us.get("changed", []) or []: name = ch.get("name") - lines.append(f" * {s(name)} changed") + lines.append(f" * {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(s(x) for x in a)}") + lines.append(f" groups +: {', '.join(a)}") if r: - lines.append(f" groups -: {', '.join(s(x) for x in r)}") + lines.append(f" groups -: {', '.join(r)}") else: - lines.append(f" {s(k)}: {s(v.get('old'))} -> {s(v.get('new'))}") + lines.append(f" {k}: {v.get('old')} -> {v.get('new')}") fl = report.get("files", {}) lines.append("\nFiles") for e in fl.get("added", []) or []: lines.append( - f" + {s(e.get('path'))} (role={s(e.get('role'))}, reason={s(e.get('reason'))})" + f" + {e.get('path')} (role={e.get('role')}, reason={e.get('reason')})" ) for e in fl.get("removed", []) or []: lines.append( - f" - {s(e.get('path'))} (role={s(e.get('role'))}, reason={s(e.get('reason'))})" + f" - {e.get('path')} (role={e.get('role')}, reason={e.get('reason')})" ) for ch in fl.get("changed", []) or []: p = ch.get("path") - lines.append(f" * {s(p)} changed") + lines.append(f" * {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: {s(v.get('old'))} -> {s(v.get('new'))}" - ) + lines.append(f" content: {v.get('old')} -> {v.get('new')}") else: - lines.append(f" {s(k)}: {s(v.get('old'))} -> {s(v.get('new'))}") + lines.append(f" {k}: {v.get('old')} -> {v.get('new')}") if not any( [ @@ -785,23 +773,14 @@ 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: `{m(report.get('generated_at'))}`\n") + out.append(f"Generated: `{report.get('generated_at')}`\n") out.append( - 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" + 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" ) filt = report.get("filters", {}) or {} @@ -809,7 +788,7 @@ def _report_markdown(report: Dict[str, Any]) -> str: if ex_paths: out.append( "- **File exclude patterns**: " - + ", ".join(f"`{m(p)}`" for p in ex_paths) + + ", ".join(f"`{p}`" for p in ex_paths) + "\n" ) @@ -826,10 +805,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" - `+ {m(p)}`\n") + out.append(f" - `+ {p}`\n") out.append(f"- Removed: {len(pk.get('removed', []) or [])}\n") for p in pk.get("removed", []) or []: - out.append(f" - `- {m(p)}`\n") + out.append(f" - `- {p}`\n") ignored_v = int(pk.get("version_changed_ignored_count") or 0) vc = len(pk.get("version_changed", []) or []) @@ -837,7 +816,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" - `~ {m(ch.get('package'))}`: `{m(ch.get('old'))}` → `{m(ch.get('new'))}`\n" + f" - `~ {ch.get('package')}`: `{ch.get('old')}` → `{ch.get('new')}`\n" ) sv = report.get("services", {}) @@ -845,64 +824,60 @@ 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" - `+ {m(u)}`\n") + out.append(f" - `+ {u}`\n") if sv.get("enabled_removed"): out.append("- Enabled removed\n") for u in sv.get("enabled_removed", []) or []: - out.append(f" - `- {m(u)}`\n") + out.append(f" - `- {u}`\n") if sv.get("changed"): out.append("- Changed\n") for ch in sv.get("changed", []) or []: unit = ch.get("unit") - out.append(f" - `{m(unit)}`\n") + out.append(f" - `{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('`'+m(x)+'`' for x in a)}\n" + f" - packages added: {', '.join('`'+x+'`' for x in a)}\n" ) if r: out.append( - f" - packages removed: {', '.join('`'+m(x)+'`' for x in r)}\n" + f" - packages removed: {', '.join('`'+x+'`' for x in r)}\n" ) else: - out.append( - f" - {m(k)}: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" - ) + out.append(f" - {k}: `{v.get('old')}` → `{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" - `+ {m(u)}`\n") + out.append(f" - `+ {u}`\n") if us.get("removed"): out.append("- Removed\n") for u in us.get("removed", []) or []: - out.append(f" - `- {m(u)}`\n") + out.append(f" - `- {u}`\n") if us.get("changed"): out.append("- Changed\n") for ch in us.get("changed", []) or []: name = ch.get("name") - out.append(f" - `{m(name)}`\n") + out.append(f" - `{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('`'+m(x)+'`' for x in a)}\n" + f" - groups added: {', '.join('`'+x+'`' for x in a)}\n" ) if r: out.append( - f" - groups removed: {', '.join('`'+m(x)+'`' for x in r)}\n" + f" - groups removed: {', '.join('`'+x+'`' for x in r)}\n" ) else: - out.append( - f" - {m(k)}: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" - ) + out.append(f" - {k}: `{v.get('old')}` → `{v.get('new')}`\n") fl = report.get("files", {}) out.append("## Files\n") @@ -910,31 +885,29 @@ def _report_markdown(report: Dict[str, Any]) -> str: out.append("- Added\n") for e in fl.get("added", []) or []: out.append( - f" - `+ {m(e.get('path'))}` (role={m(e.get('role'))}, reason={m(e.get('reason'))})\n" + f" - `+ {e.get('path')}` (role={e.get('role')}, reason={e.get('reason')})\n" ) if fl.get("removed"): out.append("- Removed\n") for e in fl.get("removed", []) or []: out.append( - f" - `- {m(e.get('path'))}` (role={m(e.get('role'))}, reason={m(e.get('reason'))})\n" + f" - `- {e.get('path')}` (role={e.get('role')}, reason={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" - `{m(p)}`\n") + out.append(f" - `{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: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" + f" - content: `{v.get('old')}` → `{v.get('new')}`\n" ) else: - out.append( - f" - {m(k)}: `{m(v.get('old'))}` → `{m(v.get('new'))}`\n" - ) + out.append(f" - {k}: `{v.get('old')}` → `{v.get('new')}`\n") if not any( [ diff --git a/enroll/explain.py b/enroll/explain.py index 23ddfda..17f1b4b 100644 --- a/enroll/explain.py +++ b/enroll/explain.py @@ -7,7 +7,6 @@ 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) @@ -528,25 +527,15 @@ def explain_state( if fmt == "json": return json.dumps(report, indent=2, sort_keys=True) - # 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 + # Text rendering out: List[str] = [] - out.append(f"Enroll explained: {s(harvest)}") + out.append(f"Enroll explained: {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: {s(hn)} (os: {s(os_family)}, pkg: {s(pkg_backend)})") - out.append(f"Enroll: {s(ver)}") + out.append(f"Host: {hn} (os: {os_family}, pkg: {pkg_backend})") + out.append(f"Enroll: {ver}") out.append("") out.append("Inventory") @@ -556,30 +545,30 @@ def explain_state( for ov in observed_via_summary: extra = "" if ov.get("top_refs"): - 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}") + extra = f" (e.g. {', '.join(ov['top_refs'])})" + out.append(f" - {ov['kind']}: {ov['count']} – {ov['why']}{extra}") out.append("") out.append("Roles collected") for rs in role_summaries: - out.append(f"- {s(rs['role'])}: {s(rs['summary'])}") + out.append(f"- {rs['role']}: {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(s(x) for x in inc[:max_examples])}{suffix}" + f" include_patterns: {', '.join(map(str, inc[:max_examples]))}{suffix}" ) if exc: suffix = "…" if len(exc) > max_examples else "" out.append( - f" exclude_patterns: {', '.join(s(x) for x in exc[:max_examples])}{suffix}" + f" exclude_patterns: {', '.join(map(str, exc[:max_examples]))}{suffix}" ) notes = rs.get("notes") or [] if notes: for n in notes[:max_examples]: - out.append(f" note: {s(n)}") + out.append(f" note: {n}") if len(notes) > max_examples: out.append( f" note: (+{len(notes) - max_examples} more. Use --format json to see them all)" @@ -590,8 +579,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(s(x) for x in exs)}" if exs else "" - out.append(f"- {s(r['reason'])} ({r['count']}): {s(r['why'])}.{ex_txt}") + ex_txt = f" Examples: {', '.join(exs)}" if exs else "" + out.append(f"- {r['reason']} ({r['count']}): {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)" @@ -603,15 +592,15 @@ def explain_state( out.append("") out.append("Why directories were included (managed_dirs.reason)") for r in managed_dir_reasons: - out.append(f"- {s(r['reason'])} ({r['count']}): {s(r['why'])}") + out.append(f"- {r['reason']} ({r['count']}): {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(s(x) for x in exs)}" if exs else "" - out.append(f"- {s(r['reason'])} ({r['count']}): {s(r['why'])}.{ex_txt}") + ex_txt = f" Examples: {', '.join(exs)}" if exs else "" + out.append(f"- {r['reason']} ({r['count']}): {r['why']}.{ex_txt}") else: out.append("- (no excluded paths)") diff --git a/enroll/ignore.py b/enroll/ignore.py index 563a9ee..d7c4198 100644 --- a/enroll/ignore.py +++ b/enroll/ignore.py @@ -129,55 +129,6 @@ HIGH_CONFIDENCE_SECRET_PATTERNS = [ # Credentials embedded in connection-string URIs, e.g. # postgres://user:pass@host, redis://:pass@host, amqp://u:p@host re.compile(rb"(?i)[a-z][a-z0-9+.-]*://[^/\s:@]*:[^/\s@]+@[^/\s]"), - # Whitespace-separated credential assignments for COMPOUND credential keys. - # - # The assignment pattern above only recognises a value when the key and value - # are joined by ':' or '='. Some real config styles (.netrc-like files, a - # number of daemon configs) instead separate a key from its value with one or - # more spaces/tabs, e.g. - # aws_secret_access_key wJalrXUtnFEMI/K7MDENG... - # client_secret hunter2value - # oauth_token ya29.aRealLookingToken - # The comment-aware soft tier below would not catch these because its keyword - # boundary (\b) is defeated by the leading underscore in a compound key - # (e.g. the 'secret' in 'client_secret' has a '_' -- a word char -- before it, - # so \bsecret\b does not match). This pattern closes that gap. - # - # It is deliberately restricted to *compound* keys -- a credential keyword that - # is joined to a prefix (case 1) or is itself a multi-word credential name - # (case 2). Standalone English words ('password', 'token', 'secret') are NOT - # matched here: they are already handled, comment-aware, by the soft tier, and - # matching them with a whitespace separator would false-positive on prose and - # on PAM 'password ...' stack directives. The trailing value test - # requires a value-like token (>=6 chars, or one containing a digit/secret-ish - # character) so a short config value such as 'yes'/'no' does not trip it. - re.compile( - rb"""(?ixm) - (?:^|[^A-Za-z0-9_.-]) - [\"']? - (?: - (?:[A-Za-z0-9]+[_.-])+ - (?:password|passwd|passphrase|pwd|pw|token|secret|credential|credentials) - | - (?:[A-Za-z0-9]+[_.-])* - (?: - auth[_.-]token|access[_.-]token|refresh[_.-]token| - client[_.-]secret|secret[_.-]key| - api[_.-]key|access[_.-]key|private[_.-]key| - aws[_.-]access[_.-]key[_.-]id|aws[_.-]secret[_.-]access[_.-]key| - azure[_.-]client[_.-]secret| - google[_.-]application[_.-]credentials|gcp[_.-]service[_.-]account| - service[_.-]account[_.-]key| - session[_.-]key - ) - ) - (?:[_.-][A-Za-z0-9]+)* - [\"']? - [ \t]+ - (?=\S) - (?:\S{6,}|\S*[0-9/+=._-]\S*) - """ - ), # HTTP(S) Authorization / Proxy-Authorization header values carrying a # bearer/basic/digest credential. re.compile( diff --git a/tests/test_diff_report_sanitize.py b/tests/test_diff_report_sanitize.py deleted file mode 100644 index 8f49dd6..0000000 --- a/tests/test_diff_report_sanitize.py +++ /dev/null @@ -1,114 +0,0 @@ -"""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 deleted file mode 100644 index 808cfc6..0000000 --- a/tests/test_explain_sanitize.py +++ /dev/null @@ -1,84 +0,0 @@ -"""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") diff --git a/tests/test_secret_detection.py b/tests/test_secret_detection.py index 3c98d64..607d059 100644 --- a/tests/test_secret_detection.py +++ b/tests/test_secret_detection.py @@ -221,65 +221,3 @@ def test_commented_out_credential_values_require_dangerous(): ), data # ...but --dangerous still collects it (operator's explicit choice). assert dangerous._content_deny_reason("/etc/app/app.conf", data) is None, data - - -def test_whitespace_separated_compound_credentials_denied(): - """Security regression (audit finding): a populated credential assignment - that uses whitespace (space/tab) instead of ':'/'=' as the separator must - still be refused in safe mode when the key is a *compound* credential name. - - The soft keyword tier cannot catch these because its ``\\b`` boundary is - defeated by the leading underscore in a compound key (e.g. the ``secret`` in - ``client_secret`` is preceded by ``_``, a word character), and the - assignment tier only recognised ``:``/``=`` separators. Some real config - styles (.netrc-like files, some daemon configs) use whitespace separators, - so a value such as ``aws_secret_access_key wJalr...`` would previously leak. - """ - safe = IgnorePolicy(dangerous=False) - dangerous = IgnorePolicy(dangerous=True) - leaking = [ - b"aws_secret_access_key wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY\n", - b"aws_access_key_id AKIAIOSFODNN7EXAMPLE\n", - b"client_secret hunter2deadbeef\n", - b"oauth_token ya29.A0ARrdaReALlOOkINgToKeN\n", - b"refresh_token 1//0longrefreshvalue\n", - b"my_token deadbeef1234\n", - b"app_password s3cr3tValue\n", - b"db_pw s3cr3tvalue\n", - b"private_key /etc/ssl/private/app.pem\n", - b"session_key 0xDEADBEEFCAFE\n", - b"\tapi_key AKIA12345678\n", - ] - for data in leaking: - assert ( - safe._content_deny_reason("/root/.netrc", data) == "sensitive_content" - ), data - # --dangerous remains the explicit opt-in to collect such material. - assert dangerous._content_deny_reason("/root/.netrc", data) is None, data - - -def test_whitespace_separated_non_credentials_still_allowed(): - """The whitespace-separated credential rule must not false-positive on - ordinary config or PAM directives. It is intentionally restricted to - *compound* credential keys with a value-like token, so standalone words and - common config lines are not affected by it. - - (Standalone ``password``/``token``/``secret`` words on a non-comment line - are still caught by the comment-aware soft tier; that is unchanged. The - cases here are ones that must remain collectable: PAM stack directives, SSH - key-file path directives, and plain settings.) - """ - pol = IgnorePolicy(dangerous=False) - allowed = [ - # SSH key-file path directives (key material is denied by path globs, - # not by content here). - b"HostKey /etc/ssh/ssh_host_ed25519_key\n", - b"host_key /etc/ssh/known\n", - b"public_key /home/u/.ssh/id_rsa.pub\n", - # Plain numeric / boolean settings. - b"max_connections 1024\n", - b"PermitRootLogin no\n", - b"PasswordAuthentication yes\n", - ] - for data in allowed: - assert pol._content_deny_reason("/etc/app/app.conf", data) is None, data