Compare commits
No commits in common. "5ffad1066596f5b1a67e831efca204b49987faf1" and "a34813e788a31ba9bec18ab28e1a546571e63df9" have entirely different histories.
5ffad10665
...
a34813e788
7 changed files with 63 additions and 436 deletions
26
enroll/cm.py
26
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.
|
||||
|
||||
|
|
|
|||
123
enroll/diff.py
123
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(
|
||||
[
|
||||
|
|
|
|||
|
|
@ -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)")
|
||||
|
||||
|
|
|
|||
|
|
@ -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 <verb> ...' 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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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")
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue