diff --git a/enroll/ignore.py b/enroll/ignore.py index 7919d1b..4cc6683 100644 --- a/enroll/ignore.py +++ b/enroll/ignore.py @@ -348,29 +348,42 @@ class IgnorePolicy: return None def _content_deny_reason(self, path: str, data: bytes) -> Optional[str]: - if b"\x00" in data: - match_path = normalize_for_match(path) - for g in self.allow_binary_globs or []: - if fnmatch.fnmatch(match_path, g): - # Binary is acceptable for explicitly-allowed paths. - return None - return "binary_like" - + # High-confidence secret *material* (private keys, age secret keys, + # populated credential assignments, credential URIs, Authorization + # headers) is scanned against the raw bytes FIRST, before any + # binary/allowlist decision. This runs even for binary payloads on the + # allow_binary_globs list: those globs exist to let genuinely-binary + # *public* keyring formats (e.g. APT/RPM GPG keyrings) through the + # "binary_like" denial, but they must NOT become a hole through which a + # keybox/keyring carrying *private* key material is captured unscanned + # in safe mode. The private-key markers are byte-oriented and match + # inside binary containers, so running them here closes that asymmetry + # while still allowing ordinary public keyrings. if not self.dangerous: - # High-confidence secret *material* (private keys, age secret keys) - # is scanned against the raw bytes and is NOT subject to comment - # stripping. A private key embedded in a file is sensitive regardless - # of comment framing, and this closes the bypass where opening a block - # comment (e.g. a leading "/*" line) hid key material from the - # line-oriented scanner. for pat in HIGH_CONFIDENCE_SECRET_PATTERNS: if pat.search(data): return "sensitive_content" + if b"\x00" in data: + match_path = normalize_for_match(path) + for g in self.allow_binary_globs or []: + if fnmatch.fnmatch(match_path, g): + # Binary is acceptable for explicitly-allowed paths. The + # high-confidence secret-material scan above has already run + # on the raw bytes, so an allowed binary keyring that + # actually embeds private-key material was refused before + # reaching here; only genuinely non-secret binary keyrings + # get this far. + return None + return "binary_like" + + if not self.dangerous: # Softer assignment/keyword/URI heuristics stay comment-aware so a # genuinely commented-out example does not make Enroll useless for # ordinary config files. iter_effective_lines() is hardened so an # unterminated/inline block comment cannot mask later real content. + # These line-oriented heuristics only make sense on text, so they + # run only after the NUL/binary check above has passed. for line in self.iter_effective_lines(data): for pat in SENSITIVE_CONTENT_PATTERNS: if pat.search(line): diff --git a/enroll/jinjaturtle.py b/enroll/jinjaturtle.py index 1f9248e..c6365ef 100644 --- a/enroll/jinjaturtle.py +++ b/enroll/jinjaturtle.py @@ -15,6 +15,44 @@ from .manifest_safety import ArtifactSafetyError, safe_artifact_file from .yamlutil import yaml_dump_mapping, yaml_load_mapping +# Bound the external JinjaTurtle subprocess. These are defence-in-depth limits +# for running a separate binary over harvested (possibly attacker-influenced) +# config: a hang or a runaway-size output must not stall or exhaust the manifest +# process. Both limits convert to an ordinary failure that jinjify_artifact() +# turns into a safe raw-file-copy fallback. The timeout can be overridden via +# ENROLL_JINJATURTLE_TIMEOUT (seconds) for unusually large legitimate configs. +_DEFAULT_JINJATURTLE_TIMEOUT_S = 30 +_MAX_JT_OUTPUT_BYTES = 16 * 1024 * 1024 + + +def _resolve_jt_timeout() -> float: + raw = os.environ.get("ENROLL_JINJATURTLE_TIMEOUT") + if raw: + try: + value = float(raw) + if value > 0: + return value + except ValueError: + pass + return float(_DEFAULT_JINJATURTLE_TIMEOUT_S) + + +JINJATURTLE_SUBPROCESS_TIMEOUT_S = _resolve_jt_timeout() + + +def _reject_oversized_jt_output(path: Path, label: str) -> None: + """Raise if a JinjaTurtle output file exceeds the ingest size cap.""" + try: + size = path.stat().st_size + except OSError as e: + raise RuntimeError(f"jinjaturtle {label} output is unreadable: {path}") from e + if size > _MAX_JT_OUTPUT_BYTES: + raise RuntimeError( + "jinjaturtle %s output is too large to ingest (%d bytes > %d)" + % (label, size, _MAX_JT_OUTPUT_BYTES) + ) + + SYSTEMD_SUFFIXES = { ".service", ".socket", @@ -479,13 +517,37 @@ def run_jinjaturtle( if force_format: cmd.extend(["-f", force_format]) - p = subprocess.run(cmd, text=True, capture_output=True) # nosec + # JinjaTurtle is an external binary run over harvested, potentially + # attacker-influenced config. Bound its runtime so a pathological or + # hostile input (e.g. a config that makes the converter loop or hang) + # cannot stall a manifest run indefinitely. A timeout is surfaced as an + # ordinary failure; jinjify_artifact() catches it and falls back to + # copying the raw file, which is the safe default. + try: + p = subprocess.run( + cmd, + text=True, + capture_output=True, + timeout=JINJATURTLE_SUBPROCESS_TIMEOUT_S, + ) # nosec + except subprocess.TimeoutExpired as e: + raise RuntimeError( + "jinjaturtle timed out after %ss for %s (role=%s)" + % (JINJATURTLE_SUBPROCESS_TIMEOUT_S, src_path, role_name) + ) from e if p.returncode != 0: raise RuntimeError( "jinjaturtle failed for %s (role=%s)\ncmd=%r\nstdout=%s\nstderr=%s" % (src_path, role_name, cmd, p.stdout, p.stderr) ) + # Cap how much generated output Enroll will ingest. A converter that + # emits an enormous template/vars file (accidentally or maliciously) + # should not be able to exhaust memory in the manifest process; refuse + # oversized output and fall back to a raw copy. + _reject_oversized_jt_output(defaults_out, "defaults") + _reject_oversized_jt_output(template_out, "template") + vars_text = defaults_out.read_text(encoding="utf-8").strip() template_text = template_out.read_text(encoding="utf-8") diff --git a/enroll/manifest_safety.py b/enroll/manifest_safety.py index 414adff..89e6a7c 100644 --- a/enroll/manifest_safety.py +++ b/enroll/manifest_safety.py @@ -12,6 +12,7 @@ from typing import Iterator, Optional, Tuple from .fsutil import open_no_follow_path from .harvest_safety import ( OutputSafetyError, + _effective_uid, ensure_safe_output_parent, prepare_new_private_dir, ) @@ -55,17 +56,66 @@ def validate_site_fqdn(value: str | None) -> str | None: return text +def _assert_root_safe_output_dir(path: Path, st: os.stat_result) -> None: + """Reject an existing output directory that is unsafe for a root-run merge. + + Only enforced when Enroll runs as root. Site/FQDN mode intentionally merges + generated files into an existing tree, and the individual writers re-open + files by path. A directory inside that tree that is owned by an unprivileged + user, or writable by group/other, lets that user pre-create or swap files + (including planting a symlink after this scan) that a later root-run write + would follow or clobber. A symlink-only scan cannot catch that race, so the + interior of a root-run output tree must additionally be root-owned and not + group/world-writable. + + The sticky-bit exception that ``harvest_safety`` allows for a shared *parent* + boundary such as ``/tmp`` is deliberately NOT honoured here: this is the + interior of Enroll's own output tree, not a shared staging root, so a + world-writable directory is never acceptable even if sticky. + """ + + if _effective_uid() != 0: + return + if st.st_uid != 0: + raise ManifestOutputError( + "manifest output tree contains a directory not owned by root; " + f"refusing root-run merge: {path}" + ) + if st.st_mode & (stat.S_IWGRP | stat.S_IWOTH): + raise ManifestOutputError( + "manifest output tree contains a group/other-writable directory; " + f"refusing root-run merge: {path}" + ) + + def _assert_no_output_symlinks(root: Path) -> None: - """Reject pre-existing symlinks in an output tree we are about to merge into. + """Reject unsafe pre-existing entries in an output tree we merge into. Non-site mode refuses existing output directories entirely. Site/FQDN modes - intentionally accumulate multiple nodes into one tree, so reject symlinks in - the tree before merging to avoid writes being redirected outside *root*. + intentionally accumulate multiple nodes into one tree, so before merging we + reject: + + * symlinks anywhere in the tree (a write could be redirected outside + *root*), and + * when running as root, any directory in the tree that is not root-owned + or is group/other-writable (an unprivileged owner could race the merge + by planting files/symlinks after this scan -- see + :func:`_assert_root_safe_output_dir`). + Version-control metadata can contain implementation-specific entries and is not part of Enroll's generated layout, so it is pruned from this check. """ skip_dirs = {".git", ".hg", ".svn"} + + # Check the root of the merge target itself, not only its descendants. + try: + root_st = root.lstat() + except FileNotFoundError: + root_st = None + if root_st is not None and not stat.S_ISLNK(root_st.st_mode): + _assert_root_safe_output_dir(root, root_st) + for dirpath, dirnames, filenames in os.walk(root, followlinks=False): dirpath_p = Path(dirpath) @@ -82,6 +132,7 @@ def _assert_no_output_symlinks(root: Path) -> None: raise ManifestOutputError( f"manifest output tree contains a symlink; refusing to merge: {p}" ) + _assert_root_safe_output_dir(p, st) for filename in filenames: if filename in skip_dirs: diff --git a/tests/test_manifest_safety.py b/tests/test_manifest_safety.py index a04c6a5..a7a0de4 100644 --- a/tests/test_manifest_safety.py +++ b/tests/test_manifest_safety.py @@ -242,3 +242,125 @@ def test_freeze_directory_bundle_fails_loudly_on_unreadable_subdir(tmp_path: Pat finally: # Restore perms so tmp_path cleanup can remove the tree. os.chmod(locked, 0o755) + + +class _FakeStat: + """Minimal stat_result stand-in so interior-dir tests can control uid/mode + independently of the CI runner's real uid and umask.""" + + def __init__(self, real_st, *, uid: int, mode_bits: int | None = None): + self.st_mode = real_st.st_mode if mode_bits is None else mode_bits + self.st_uid = uid + self.st_gid = getattr(real_st, "st_gid", 0) + + +def _patch_lstat(monkeypatch, ms, *, uid_for, mode_for=None): + """Patch Path.lstat so directories report a chosen uid / mode. + + ``uid_for(path) -> int`` and optional ``mode_for(path) -> int|None`` let a + test simulate a non-root-owned or writable interior directory without + needing real root privileges. + """ + real_lstat = ms.Path.lstat + + def fake_lstat(self): + st = real_lstat(self) + mode = None + if mode_for is not None: + mode = mode_for(self) + return _FakeStat(st, uid=uid_for(self), mode_bits=mode) + + monkeypatch.setattr(ms.Path, "lstat", fake_lstat) + + +def test_existing_site_tree_rejects_world_writable_interior_dir_as_root( + tmp_path: Path, monkeypatch +): + """Root-run site merge must refuse an attacker-writable interior directory. + + A symlink-only scan is insufficient: an unprivileged owner of a + group/other-writable directory inside the output tree can plant files or a + symlink after the scan and race the merge. Under root, such a directory is + rejected up front. + """ + import enroll.manifest_safety as ms + + out = tmp_path / "site" + out.mkdir() + (out / "roles").mkdir() + + interior = (out / "roles").resolve() + + monkeypatch.setattr(ms, "_effective_uid", lambda: 0) + # Everything root-owned; the interior "roles" dir is world-writable. + _patch_lstat( + monkeypatch, + ms, + uid_for=lambda p: 0, + mode_for=lambda p: (0o40777 if Path(p).resolve() == interior else 0o40700), + ) + with pytest.raises(ManifestOutputError, match="group/other-writable"): + ms._assert_no_output_symlinks(out) + + +def test_existing_site_tree_rejects_non_root_owned_interior_dir_as_root( + tmp_path: Path, monkeypatch +): + """Root-run site merge must refuse an interior directory owned by another + (unprivileged) user, even if it is not itself writable by group/other.""" + import enroll.manifest_safety as ms + + out = tmp_path / "site" + out.mkdir() + (out / "roles").mkdir() + + interior = (out / "roles").resolve() + + monkeypatch.setattr(ms, "_effective_uid", lambda: 0) + _patch_lstat( + monkeypatch, + ms, + uid_for=lambda p: (1000 if Path(p).resolve() == interior else 0), + mode_for=lambda p: 0o40755, + ) + with pytest.raises(ManifestOutputError, match="not owned by root"): + ms._assert_no_output_symlinks(out) + + +def test_existing_site_tree_allows_clean_root_owned_interior_as_root( + tmp_path: Path, monkeypatch +): + """A clean, root-owned, non-writable interior tree passes the merge check.""" + import enroll.manifest_safety as ms + + out = tmp_path / "site" + out.mkdir() + (out / "roles").mkdir() + (out / "roles" / "svc").mkdir() + + monkeypatch.setattr(ms, "_effective_uid", lambda: 0) + _patch_lstat( + monkeypatch, + ms, + uid_for=lambda p: 0, + mode_for=lambda p: 0o40755, + ) + # No exception: root-owned, not group/other-writable. + ms._assert_no_output_symlinks(out) + + +def test_existing_site_tree_interior_check_is_noop_for_non_root( + tmp_path: Path, monkeypatch +): + """Non-root runs keep the original symlink-only semantics (no ownership or + writability enforcement), so ordinary user workflows are unaffected.""" + import enroll.manifest_safety as ms + + out = tmp_path / "site" + out.mkdir() + (out / "roles").mkdir() + (out / "roles").chmod(0o777) + + monkeypatch.setattr(ms, "_effective_uid", lambda: 1000) + # No exception: interior writability is irrelevant when not running as root. + ms._assert_no_output_symlinks(out) diff --git a/tests/test_secret_detection.py b/tests/test_secret_detection.py index 8de887c..20a41bf 100644 --- a/tests/test_secret_detection.py +++ b/tests/test_secret_detection.py @@ -323,3 +323,51 @@ def test_whitespace_separated_non_credentials_still_allowed(): ] for data in allowed: assert pol._content_deny_reason("/etc/app/app.conf", data) is None, data + + +def test_binary_keyring_allowlist_still_scans_for_private_key_material(): + """A binary file on the keyring allow-list must still be refused in safe + mode if it embeds private-key material. + + ``DEFAULT_ALLOW_BINARY_GLOBS`` exists so genuinely-binary *public* keyrings + (APT/RPM GPG keyrings) survive the ``binary_like`` denial. It must not + become a hole through which a keybox/keyring carrying *private* key material + is captured unscanned. The high-confidence secret-material scan therefore + runs on the raw bytes before the binary allow-list decision. + """ + pol = IgnorePolicy(dangerous=False) + + private_binary = ( + b"\x99\x01\x0d\x04" + + b"\x00" * 20 + + b"-----BEGIN PGP PRIVATE KEY BLOCK-----\nsecret\n" + + b"-----END PGP PRIVATE KEY BLOCK-----" + + b"\x00" * 8 + ) + for path in ( + "/etc/apt/trusted.gpg", + "/etc/apt/trusted.gpg.d/mykey.gpg", + "/etc/apt/keyrings/mykey.gpg", + "/etc/pki/rpm-gpg/RPM-GPG-KEY-mine", + "/usr/share/keyrings/mykey.gpg", + ): + assert ( + pol._content_deny_reason(path, private_binary) == "sensitive_content" + ), path + + # A genuinely non-secret binary keyring at those same paths is still allowed + # (the allow-list still exempts it from the binary_like denial). + public_binary = b"\x99\x01\x0d\x04" + b"\x00" * 40 + b"public key packet data" + for path in ( + "/etc/apt/keyrings/pub.gpg", + "/usr/share/keyrings/pub.gpg", + "/etc/pki/rpm-gpg/RPM-GPG-KEY-pub", + ): + assert pol._content_deny_reason(path, public_binary) is None, path + + # --dangerous still collects everything. + dangerous = IgnorePolicy(dangerous=True) + assert ( + dangerous._content_deny_reason("/etc/apt/keyrings/mykey.gpg", private_binary) + is None + )