More hardening

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:
+        # High-confidence secret *material* (private keys, age secret keys,
-            match_path = normalize_for_match(path)
+        # populated credential assignments, credential URIs, Authorization
-            for g in self.allow_binary_globs or []:
+        # headers) is scanned against the raw bytes FIRST, before any
-                if fnmatch.fnmatch(match_path, g):
+        # binary/allowlist decision. This runs even for binary payloads on the
-                    # Binary is acceptable for explicitly-allowed paths.
+        # allow_binary_globs list: those globs exist to let genuinely-binary
-                    return None
+        # *public* keyring formats (e.g. APT/RPM GPG keyrings) through the
-            return "binary_like"
+        # "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):

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")
 

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
+    intentionally accumulate multiple nodes into one tree, so before merging we
-    the tree before merging to avoid writes being redirected outside *root*.
+    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:

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)

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
+    )