Add warning about --dangerous mode if sops is not in use

CHANGELOG.md

@@ -15,6 +15,7 @@
  * Perform harvest validation before trying to manifest from it.
  * Stricter validation on FQDN name in multisite mode.
  * Strict check of `$PATH` when running harvest as root, in case it could lead to execution of unsafe binaries during harvest. Override with `--assume-safe-path` for non-interactive or CI purposes.
+ * Stricter validation of the destination dirs that harvest or manifest write to, to prevent writing to a different user-controlled area. Stricter permissions on the output dirs too.
 
 # 0.6.0
 

README.md

@@ -196,6 +196,8 @@ Enforcement is intentionally “safe”:
 
 If the config manager tool is not on `PATH`, Enroll returns an error and does not enforce.
 
+**IMPORTANT**: Only enforce harvest bundles that you trust. Validation checks bundle structure and artifact safety; it does not prove that the described system state is safe to apply, e.g. hasn't been tampered with by another user with sufficient permission to do so!
+
 
 **Output formats**
 - `--format json` (default for webhooks)
@@ -285,12 +287,15 @@ Safe-mode content scanning is intentionally conservative. It treats common assig
 
 Automatic harvesting of per-user shell dotfiles is also disabled by default, even when those files differ from `/etc/skel`, because `.bashrc`, `.profile`, `.bash_aliases`, and similar files commonly contain exported tokens, credentials, or aliases/functions with embedded secrets. Use `--dangerous` for automatic shell-dotfile capture, or use targeted `--include-path` patterns for narrower safe-mode review.
 
-If you opt in to collecting everything:
+If you wish to opt in to collecting everything, use `--dangerous` mode, but be aware of what it means:
 
 ### `--dangerous`
-**WARNING:** disables “likely secret” safety checks. This can copy private keys, TLS key material, API tokens, database passwords, and other credentials into the harvest output **in plaintext**.
 
-If you intend to keep harvests/manifests long-term (especially in git), strongly consider encrypting them at rest.
+**IMPORTANT:** 'dangerous' mode is exactly that: it disables “likely secret” safety checks when harvesting system data.
+
+This means it can copy private keys, TLS key material, API tokens, database passwords, and other credentials into the harvest output **in plaintext**, including paths that would normally be considered very secret.
+
+If you intend to keep harvests/manifests long-term on disk away from the host or its usual protected paths, strongly consider encrypting them at rest!
 
 ### Encrypt bundles at rest with `--sops`
 `--sops` encrypts the harvest and/or manifest outputs into a single `.tar.gz.sops` file (GPG). This is for **storage-at-rest**, not for direct “Ansible SOPS inventory” workflows.

enroll/cli.py

@@ -22,6 +22,7 @@ from .diff import (
 )
 from .explain import explain_state
 from .harvest import harvest
+from .harvest_safety import ensure_safe_output_parent, write_text_output_file
 from .manifest import manifest
 from .remote import (
     remote_harvest,
@@ -112,6 +113,15 @@ def _action_lookup(p: argparse.ArgumentParser) -> dict[str, argparse.Action]:
     return m
 
 
+def _warn_dangerous_harvest(*, sops_enabled: bool) -> None:
+    if not sops_enabled:
+        print(
+            "warning: --dangerous is enabled. The harvest may contain sensitive "
+            "files, credentials, private keys, tokens, or application secrets. "
+            "Consider using --sops to encrypt the harvest at rest."
+        )
+
+
 def _choose_flag(a: argparse.Action) -> Optional[str]:
     # Prefer a long flag if available (e.g. --dangerous over -d)
     for s in getattr(a, "option_strings", []) or []:
@@ -135,14 +145,75 @@ def _split_list_value(v: str) -> list[str]:
     return [raw] if raw else []
 
 
+def _root_trust_reason(path: Path, *, final: bool) -> Optional[str]:
+    """Return why a PATH directory/ancestor is unsafe for root execution."""
+
+    running_as_root = _is_effective_root()
+    if not final and not running_as_root:
+        return None
+    try:
+        st = os.stat(path)
+    except OSError:
+        return None
+    if not stat.S_ISDIR(st.st_mode):
+        return None
+
+    subject = "directory" if final else "parent directory"
+    if running_as_root and st.st_uid != 0:
+        return f"{subject} is not owned by root"
+
+    writable_by_group = bool(st.st_mode & stat.S_IWGRP)
+    writable_by_other = bool(st.st_mode & stat.S_IWOTH)
+    sticky = bool(st.st_mode & stat.S_ISVTX)
+
+    # A sticky shared ancestor such as /tmp may contain a root-owned PATH
+    # directory safely enough for this check, but the PATH entry itself must
+    # never be writable by group/other because that permits command planting.
+    if final or not sticky:
+        if writable_by_other:
+            return f"{subject} is world-writable"
+        if writable_by_group:
+            return f"{subject} is group-writable"
+    return None
+
+
+def _root_parent_trust_reason(path: Path) -> Optional[str]:
+    """Check original and resolved PATH ancestors for root trust."""
+
+    if not _is_effective_root():
+        return None
+
+    candidates: list[Path] = []
+    candidates.extend(reversed(path.parents))
+    try:
+        resolved = path.resolve(strict=True)
+    except OSError:
+        resolved = None
+    if resolved is not None and resolved != path:
+        candidates.extend(reversed(resolved.parents))
+
+    seen: set[str] = set()
+    for parent in candidates:
+        key = str(parent)
+        if key in seen:
+            continue
+        seen.add(key)
+        reason = _root_trust_reason(parent, final=False)
+        if reason:
+            return f"{reason}: {parent}"
+    return None
+
+
 def _path_entry_is_unsafe(entry: str) -> Optional[str]:
     """Return a human-readable reason if a PATH entry is unsafe for root.
 
     Empty PATH entries and relative entries resolve via the current working
     directory, which is equivalent to trusting whatever directory the operator
-    happens to be in.  Existing group/world-writable directories are also risky
+    happens to be in. Existing group/world-writable directories are also risky
     when Enroll is run as root because Enroll deliberately invokes host tools
-    from PATH while harvesting and enforcing state.
+    from PATH while harvesting and enforcing state. When running as root, an
+    existing PATH directory must also be root-owned; a non-root-owned 0755
+    directory is still attacker-controlled by its owner.
     """
 
     if entry == "":
@@ -152,16 +223,21 @@ def _path_entry_is_unsafe(entry: str) -> Optional[str]:
     if not os.path.isabs(entry):
         return "relative PATH entry resolves from the current directory"
 
+    p = Path(entry)
+    parent_reason = _root_parent_trust_reason(p)
+    if parent_reason:
+        return parent_reason
+
     try:
         st = os.stat(entry)
     except OSError:
         return None
     if not stat.S_ISDIR(st.st_mode):
         return None
-    if st.st_mode & stat.S_IWOTH:
+
-        return "directory is world-writable"
+    final_reason = _root_trust_reason(p, final=True)
-    if st.st_mode & stat.S_IWGRP:
+    if final_reason:
-        return "directory is group-writable"
+        return final_reason
     return None
 
 
@@ -353,7 +429,7 @@ def _resolve_sops_out_file(out: Optional[str], *, hint: str) -> Path:
 
 
 def _tar_dir_to(path_dir: Path, tar_path: Path) -> None:
-    tar_path.parent.mkdir(parents=True, exist_ok=True)
+    ensure_safe_output_parent(tar_path, label="harvest tar output")
     with tarfile.open(tar_path, mode="w:gz") as tf:
         # Keep a stable on-disk layout when extracted: state.json + artifacts/
         tf.add(str(path_dir), arcname=".")
@@ -363,7 +439,7 @@ def _encrypt_harvest_dir_to_sops(
     bundle_dir: Path, out_file: Path, fps: list[str]
 ) -> Path:
     out_file = Path(out_file)
-    out_file.parent.mkdir(parents=True, exist_ok=True)
+    ensure_safe_output_parent(out_file, label="encrypted harvest output")
 
     # Create the tarball alongside the output file (keeps filesystem permissions/locality sane).
     fd, tmp_tgz = tempfile.mkstemp(
@@ -426,8 +502,8 @@ def _add_config_args(p: argparse.ArgumentParser) -> None:
         "-c",
         "--config",
         help=(
-            "Path to an INI config file for default options. If omitted, enroll will look for "
+            "Path to an INI config file for default options. If omitted, enroll will look for a path defined by the "
-            "./enroll.ini, ./.enroll.ini, or ~/.config/enroll/enroll.ini (or $XDG_CONFIG_HOME/enroll/enroll.ini)."
+            "ENROLL_CONFIG environment variable , ~/.config/enroll/enroll.ini (or $XDG_CONFIG_HOME/enroll/enroll.ini)."
         ),
     )
     p.add_argument(
@@ -887,6 +963,11 @@ def main() -> None:
     )
     args = ap.parse_args(argv)
 
+    if args.cmd in {"harvest", "single-shot"} and bool(
+        getattr(args, "dangerous", False)
+    ):
+        _warn_dangerous_harvest(sops_enabled=bool(getattr(args, "sops", None)))
+
     _confirm_root_path_safety(force=bool(getattr(args, "assume_safe_path", False)))
 
     # Preserve historical defaults for remote harvesting unless ssh_config lookup is enabled.
@@ -1021,9 +1102,7 @@ def main() -> None:
 
             out_path = getattr(args, "out", None)
             if out_path:
-                p = Path(out_path).expanduser()
+                write_text_output_file(out_path, txt, label="validation report")
-                p.parent.mkdir(parents=True, exist_ok=True)
-                p.write_text(txt, encoding="utf-8")
             else:
                 sys.stdout.write(txt)
 
@@ -1094,9 +1173,7 @@ def main() -> None:
             txt = format_report(report, fmt=str(getattr(args, "format", "text")))
             out_path = getattr(args, "out", None)
             if out_path:
-                p = Path(out_path).expanduser()
+                write_text_output_file(out_path, txt, label="diff report")
-                p.parent.mkdir(parents=True, exist_ok=True)
-                p.write_text(txt, encoding="utf-8")
             else:
                 print(txt, end="" if txt.endswith("\n") else "\n")
 

enroll/harvest_safety.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 import stat
+import tempfile
 from pathlib import Path
 
 
@@ -9,6 +10,13 @@ class OutputSafetyError(RuntimeError):
     """Raised when an output path is unsafe for root-run plaintext output."""
 
 
+# Keep a reference to the real euid getter so tests that monkeypatch
+# enroll.harvest.os.geteuid do not accidentally make output-safety code
+# believe a non-root test process is running as root. Tests that need to
+# exercise root behavior can still monkeypatch _effective_uid directly.
+_OS_GETEUID = getattr(os, "geteuid", None)
+
+
 def _chmod_private(path: Path) -> None:
     try:
         os.chmod(path, 0o700)
@@ -17,8 +25,111 @@ def _chmod_private(path: Path) -> None:
         pass
 
 
-def _assert_no_existing_symlink_components(path: Path, *, label: str) -> None:
+def _effective_uid() -> int | None:
-    """Reject symlinks in existing parent components of an output path."""
+    if _OS_GETEUID is None:
+        return None
+    try:
+        return int(_OS_GETEUID())
+    except OSError:
+        return None
+
+
+def _assert_trusted_root_parent(path: Path, st: os.stat_result, *, label: str) -> None:
+    """Reject parent directories that are unsafe when Enroll runs as root.
+
+    Enroll deliberately invokes host tools and writes host configuration state,
+    so root-run output should not pass through parent directories controlled by
+    an unprivileged user. Root-owned sticky shared directories such as /tmp are
+    allowed as a boundary, but any existing child below them must still be
+    root-owned and non-writable by group/other.
+    """
+
+    if _effective_uid() != 0:
+        return
+    if not stat.S_ISDIR(st.st_mode):
+        raise OutputSafetyError(f"{label} parent is not a directory: {path}")
+    if st.st_uid != 0:
+        raise OutputSafetyError(
+            f"{label} parent is not owned by root; refusing root-run output: {path}"
+        )
+    writable_by_group_or_other = st.st_mode & (stat.S_IWGRP | stat.S_IWOTH)
+    sticky = st.st_mode & stat.S_ISVTX
+    if writable_by_group_or_other and not sticky:
+        raise OutputSafetyError(
+            f"{label} parent is writable by group/other; refusing root-run output: {path}"
+        )
+
+
+def _assert_existing_output_dir_component(path: Path, *, label: str) -> None:
+    try:
+        st = path.lstat()
+    except OSError as e:
+        raise OutputSafetyError(f"unable to inspect {label} parent: {path}") from e
+    if stat.S_ISLNK(st.st_mode):
+        raise OutputSafetyError(
+            f"{label} parent path contains a symlink; refusing: {path}"
+        )
+    _assert_trusted_root_parent(path, st, label=label)
+
+
+def _mkdir_private_dir_tree(
+    path: Path, *, label: str, final_must_be_new: bool = False
+) -> Path:
+    """Create a directory tree one component at a time with safety checks.
+
+    pathlib.mkdir(parents=True) can traverse a symlink inserted after a parent
+    pre-check and create deeper components in the symlink target. Walking one
+    component at a time avoids that class of race for root-run output paths.
+    """
+
+    out = Path(path).expanduser()
+    parts = out.parts
+    if not parts:
+        return out
+
+    if out.is_absolute():
+        cur = Path(parts[0])
+        rest = parts[1:]
+        _assert_existing_output_dir_component(cur, label=label)
+    else:
+        cur = Path.cwd()
+        rest = parts
+        _assert_existing_output_dir_component(cur, label=label)
+
+    for idx, part in enumerate(rest):
+        cur = cur / part
+        is_final = idx == len(rest) - 1
+        if os.path.lexists(cur):
+            if is_final and final_must_be_new:
+                raise OutputSafetyError(
+                    f"{label} path already exists; refusing to overwrite or merge: {cur}"
+                )
+            _assert_existing_output_dir_component(cur, label=label)
+            continue
+        try:
+            os.mkdir(cur, 0o700)
+        except FileExistsError:
+            if is_final and final_must_be_new:
+                raise OutputSafetyError(
+                    f"{label} path already exists; refusing to overwrite or merge: {cur}"
+                )
+            _assert_existing_output_dir_component(cur, label=label)
+            continue
+        _chmod_private(cur)
+        _assert_existing_output_dir_component(cur, label=label)
+
+    return out
+
+
+def _assert_no_existing_symlink_components(
+    path: Path, *, label: str, require_trusted_root_parents: bool = True
+) -> None:
+    """Reject unsafe existing parent components of an output path.
+
+    This catches symlink parents for all users. When running as root, it also
+    rejects existing parents controlled by an unprivileged user so an attacker
+    cannot redirect root output by racing or replacing a parent directory.
+    """
 
     parts = path.parts
     if not parts:
@@ -30,52 +141,96 @@ def _assert_no_existing_symlink_components(path: Path, *, label: str) -> None:
     else:
         cur = Path.cwd()
         rest = parts[:-1]
+        if require_trusted_root_parents:
+            _assert_existing_output_dir_component(cur, label=label)
 
     for part in rest:
         cur = cur / part
         if not os.path.lexists(cur):
             return
+        if require_trusted_root_parents:
+            _assert_existing_output_dir_component(cur, label=label)
+        else:
+            try:
+                st = cur.lstat()
+            except OSError as e:
+                raise OutputSafetyError(
+                    f"unable to inspect {label} parent: {cur}"
+                ) from e
+            if stat.S_ISLNK(st.st_mode):
+                raise OutputSafetyError(
+                    f"{label} parent path contains a symlink; refusing: {cur}"
+                )
+
+
+def ensure_safe_output_parent(path: str | Path, *, label: str = "output") -> Path:
+    """Create and validate the parent directory for a root-run output file.
+
+    The parent is checked with the same symlink/root-trust rules as plaintext
+    bundle directories. This is for output *files* such as reports and SOPS
+    bundles, where replacing an existing regular file is acceptable but
+    following attacker-controlled parent paths is not.
+    """
+
+    out = Path(path).expanduser()
+    parent = out.parent if out.parent != Path("") else Path(".")
+    sentinel = parent / ".enroll-output-parent-check"
+    _assert_no_existing_symlink_components(sentinel, label=label)
+    _mkdir_private_dir_tree(parent, label=label, final_must_be_new=False)
+    _assert_no_existing_symlink_components(sentinel, label=label)
+    return parent
+
+
+def write_text_output_file(
+    path: str | Path,
+    text: str,
+    *,
+    label: str = "output file",
+    mode: int = 0o600,
+) -> Path:
+    """Safely write a user-facing output text file.
+
+    The write is staged in the destination directory and atomically renamed into
+    place. A final-path symlink is replaced rather than followed, while parent
+    symlinks or root-unsafe parents are refused by ensure_safe_output_parent().
+    """
+
+    out = Path(path).expanduser()
+    parent = ensure_safe_output_parent(out, label=label)
+    fd, tmp_name = tempfile.mkstemp(prefix=".enroll-output-", dir=str(parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(text)
         try:
-            st = cur.lstat()
+            os.chmod(tmp_name, mode)
-        except OSError as e:
+        except OSError:
-            raise OutputSafetyError(f"unable to inspect {label} parent: {cur}") from e
+            pass
-        if stat.S_ISLNK(st.st_mode):
+        os.replace(tmp_name, out)
-            raise OutputSafetyError(
+    finally:
-                f"{label} parent path contains a symlink; refusing: {cur}"
+        try:
-            )
+            os.unlink(tmp_name)
+        except FileNotFoundError:
+            pass
+    return out
 
 
 def prepare_new_private_dir(path: str | Path, *, label: str = "output") -> Path:
     """Create a brand-new private output directory.
 
-    Refuse existing paths, including symlinks.  This prevents root-run harvests
+    Refuse existing paths, including symlinks. This prevents root-run harvests
     from writing into attacker-precreated directories in shared locations such
     as /tmp, and keeps plaintext bundles private by default.
     """
 
     out = Path(path).expanduser()
     _assert_no_existing_symlink_components(out, label=label)
-    if os.path.lexists(out):
+    return _mkdir_private_dir_tree(out, label=label, final_must_be_new=True)
-        raise OutputSafetyError(
-            f"{label} path already exists; refusing to overwrite or merge: {out}"
-        )
-
-    out.mkdir(parents=True, exist_ok=False, mode=0o700)
-    _chmod_private(out)
-
-    try:
-        st = out.lstat()
-    except OSError as e:
-        raise OutputSafetyError(f"unable to inspect {label} path: {out}") from e
-    if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode):
-        raise OutputSafetyError(f"{label} path is not a real directory: {out}")
-    return out
 
 
 def ensure_private_empty_dir(path: str | Path, *, label: str = "output") -> Path:
     """Create or validate a private empty directory.
 
-    This is for internally-generated random cache/temp directories.  User-facing
+    This is for internally-generated random cache/temp directories. User-facing
     --out paths should normally use prepare_new_private_dir() instead.
     """
 
@@ -99,6 +254,4 @@ def ensure_private_empty_dir(path: str | Path, *, label: str = "output") -> Path
         _chmod_private(out)
         return out
 
-    out.mkdir(parents=True, exist_ok=False, mode=0o700)
+    return _mkdir_private_dir_tree(out, label=label, final_must_be_new=True)
-    _chmod_private(out)
-    return out

enroll/manifest.py

@@ -10,6 +10,7 @@ from typing import List, Optional
 from .ansible import manifest_from_bundle_dir as manifest_ansible_from_bundle_dir
 from .puppet import manifest_from_bundle_dir as manifest_puppet_from_bundle_dir
 from .salt import manifest_from_bundle_dir as manifest_salt_from_bundle_dir
+from .harvest_safety import ensure_safe_output_parent
 from .manifest_safety import validate_site_fqdn
 from .remote import _safe_extract_tar
 from .sopsutil import (
@@ -91,7 +92,7 @@ def _tar_dir_to_with_progress(
     """Create a tar.gz of src_dir at tar_path, with a simple per-entry progress display."""
     src_dir = Path(src_dir)
     tar_path = Path(tar_path)
-    tar_path.parent.mkdir(parents=True, exist_ok=True)
+    ensure_safe_output_parent(tar_path, label="manifest tar output")
 
     # Collect paths (dirs + files)
     paths: list[Path] = [src_dir]
@@ -144,7 +145,7 @@ def _encrypt_manifest_out_dir_to_sops(
     """Tar+encrypt the generated manifest output directory into a single .sops file."""
     require_sops_cmd()
     out_file = Path(out_file)
-    out_file.parent.mkdir(parents=True, exist_ok=True)
+    ensure_safe_output_parent(out_file, label="encrypted manifest output")
 
     fd, tmp_tgz = tempfile.mkstemp(
         prefix=".enroll-manifest-",

enroll/manifest_safety.py

@@ -7,6 +7,12 @@ import stat
 from pathlib import Path
 from typing import Iterator, Tuple
 
+from .harvest_safety import (
+    OutputSafetyError,
+    ensure_safe_output_parent,
+    prepare_new_private_dir,
+)
+
 
 class ArtifactSafetyError(RuntimeError):
     """Raised when a harvest artifact path is unsafe to consume."""
@@ -105,13 +111,13 @@ def _safe_relative_path(value: str, *, field: str) -> Path:
 def prepare_manifest_output_dir(
     out_dir: str | Path, *, allow_existing: bool = False
 ) -> Path:
-    """Create a manifest output directory, refusing to overwrite anything.
+    """Create a manifest output directory, refusing unsafe root output paths.
 
     Rendering a manifest may be run by root and may target configuration-
-    management trees.  Refuse an existing path rather than deleting or merging
+    management trees. Refuse an existing path rather than deleting or merging
     with it by default; callers that intentionally support accumulation, such
-    as --fqdn site mode, may allow an existing directory but never a symlink or
+    as --fqdn site mode, may allow an existing directory but never a symlink,
-    non-directory path.
+    non-directory path, symlinked parent, or root-unsafe parent.
     """
 
     out = Path(out_dir).expanduser()
@@ -120,6 +126,12 @@ def prepare_manifest_output_dir(
             raise ManifestOutputError(
                 "manifest output path already exists; refusing to overwrite: " f"{out}"
             )
+        try:
+            ensure_safe_output_parent(
+                out / ".enroll-manifest-output-check", label="manifest output"
+            )
+        except OutputSafetyError as e:
+            raise ManifestOutputError(str(e)) from e
         st = out.lstat()
         if stat.S_ISLNK(st.st_mode):
             raise ManifestOutputError(
@@ -131,12 +143,10 @@ def prepare_manifest_output_dir(
             )
         _assert_no_output_symlinks(out)
         return out
-    out.mkdir(parents=True, exist_ok=False, mode=0o700)
     try:
-        os.chmod(out, 0o700)
+        return prepare_new_private_dir(out, label="manifest output")
-    except OSError:
+    except OutputSafetyError as e:
-        pass
+        raise ManifestOutputError(str(e)) from e
-    return out
 
 
 def _assert_no_symlink_components(path: Path, *, root: Path) -> None:

enroll/remote.py

@@ -578,11 +578,14 @@ def _remote_harvest(
 
         sftp = ssh.open_sftp()
         rtmp: Optional[str] = None
+        remote_root_tmp: Optional[str] = None
         try:
             rc, out, err = _ssh_run(ssh, "mktemp -d")
             if rc != 0:
                 raise RuntimeError(f"Remote mktemp failed: {err.strip()}")
             rtmp = out.strip()
+            if not rtmp:
+                raise RuntimeError("Remote mktemp returned an empty path")
 
             # Be explicit: restrict the remote staging area to the current user.
             rc, out, err = _ssh_run(ssh, f"chmod 700 -- {shlex.quote(rtmp)}")
@@ -590,10 +593,35 @@ def _remote_harvest(
                 raise RuntimeError(f"Remote chmod failed: {err.strip()}")
 
             rapp = f"{rtmp}/enroll.pyz"
-            rbundle = f"{rtmp}/bundle"
-
             sftp.put(str(pyz), rapp)
 
+            if not no_sudo:
+                # The remote zipapp is staged as the SSH user, but the harvest
+                # itself runs as root.  Root must not write its bundle under the
+                # SSH user's mktemp directory: the root-output safety checks
+                # deliberately reject user-owned parents to avoid symlink/race
+                # issues.  Create a separate sudo-owned tempdir for the bundle.
+                rc, out, err = _ssh_run_sudo(
+                    ssh, "mktemp -d", sudo_password=sudo_password, get_pty=True
+                )
+                if rc != 0:
+                    raise RuntimeError(f"Remote sudo mktemp failed: {err.strip()}")
+                remote_root_tmp = out.strip()
+                if not remote_root_tmp:
+                    raise RuntimeError("Remote sudo mktemp returned an empty path")
+
+                rc, out, err = _ssh_run_sudo(
+                    ssh,
+                    f"chmod 700 -- {shlex.quote(remote_root_tmp)}",
+                    sudo_password=sudo_password,
+                    get_pty=True,
+                )
+                if rc != 0:
+                    raise RuntimeError(f"Remote sudo chmod failed: {err.strip()}")
+                rbundle = f"{remote_root_tmp}/bundle"
+            else:
+                rbundle = f"{rtmp}/bundle"
+
             # Run remote harvest.
             argv: list[str] = [
                 remote_python,
@@ -635,9 +663,10 @@ def _remote_harvest(
                         "Unable to determine remote username for chown. "
                         "Pass --remote-user explicitly or use --no-sudo."
                     )
+                chown_target = remote_root_tmp or rbundle
                 chown_cmd = (
                     "chown -R -- "
-                    f"{shlex.quote(resolved_user)} {shlex.quote(rbundle)}"
+                    f"{shlex.quote(resolved_user)} {shlex.quote(chown_target)}"
                 )
                 rc, out, err = _ssh_run_sudo(
                     ssh,
@@ -678,7 +707,19 @@ def _remote_harvest(
                 _safe_extract_tar(tf, local_out_dir)
 
         finally:
-            # Cleanup remote tmpdir even on failure.
+            # Cleanup remote tmpdirs even on failure.  The sudo-owned harvest
+            # tempdir may still be root-owned if harvest/chown failed, so remove
+            # it via sudo and avoid masking the original error if cleanup fails.
+            if remote_root_tmp:
+                try:
+                    _ssh_run_sudo(
+                        ssh,
+                        f"rm -rf -- {shlex.quote(remote_root_tmp)}",
+                        sudo_password=sudo_password,
+                        get_pty=True,
+                    )
+                except Exception:
+                    pass  # nosec - best-effort remote cleanup
             if rtmp:
                 _ssh_run(ssh, f"rm -rf -- {shlex.quote(rtmp)}")
             try:

enroll/sopsutil.py

@@ -7,6 +7,8 @@ import tempfile
 from pathlib import Path
 from typing import Iterable, List, Optional
 
+from .harvest_safety import ensure_safe_output_parent
+
 
 class SopsError(RuntimeError):
     pass
@@ -46,7 +48,7 @@ def encrypt_file_binary(
     sops = require_sops_cmd()
     src_path = Path(src_path)
     dst_path = Path(dst_path)
-    dst_path.parent.mkdir(parents=True, exist_ok=True)
+    ensure_safe_output_parent(dst_path, label="sops output")
 
     res = subprocess.run(
         [
@@ -98,7 +100,7 @@ def decrypt_file_binary_to(
     sops = require_sops_cmd()
     src_path = Path(src_path)
     dst_path = Path(dst_path)
-    dst_path.parent.mkdir(parents=True, exist_ok=True)
+    ensure_safe_output_parent(dst_path, label="sops output")
 
     res = subprocess.run(
         [

tests/test_cli_helpers.py

@@ -243,3 +243,20 @@ def test_confirm_root_path_safety_force_skips_prompt(monkeypatch):
     )
 
     cli._confirm_root_path_safety(force=True)
+
+
+def test_unsafe_root_path_reasons_flags_non_root_owned_dir(tmp_path: Path, monkeypatch):
+    from enroll import cli
+
+    non_root_owned = tmp_path / "user-bin"
+    non_root_owned.mkdir()
+    if hasattr(os, "geteuid") and os.geteuid() == 0:
+        try:
+            os.chown(non_root_owned, 65534, -1)
+        except OSError:
+            pass
+
+    monkeypatch.setattr(cli, "_is_effective_root", lambda: True)
+    reasons = cli._unsafe_root_path_reasons(str(non_root_owned))
+
+    assert any("not owned by root" in reason for reason in reasons)

tests/test_harvest_collectors.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 from enroll.harvest_collectors.context import HarvestContext
+from enroll.harvest_collectors.paths import ExtraPathsCollector, UsrLocalCustomCollector
 from enroll.harvest_collectors.runtime import RuntimeStateCollector
-from enroll.harvest_types import FirewallRuntimeSnapshot, SysctlSnapshot
+from enroll.harvest_types import FirewallRuntimeSnapshot, ManagedFile, SysctlSnapshot
 from enroll.ignore import IgnorePolicy
 from enroll.pathfilter import PathFilter
 
@@ -11,11 +14,11 @@ class _Backend:
     name = "dpkg"
 
 
-def _context(tmp_path):
+def _context(tmp_path: Path, *, include=(), exclude=(), policy=None) -> HarvestContext:
     return HarvestContext(
-        bundle_dir=str(tmp_path),
+        bundle_dir=str(tmp_path / "bundle"),
-        policy=IgnorePolicy(),
+        policy=policy or IgnorePolicy(),
-        path_filter=PathFilter(include=(), exclude=()),
+        path_filter=PathFilter(include=include, exclude=exclude),
         platform={},
         backend=_Backend(),
         installed_pkgs={},
@@ -245,3 +248,149 @@ def test_container_images_collector_notes_unexpected_inspect_shape(
 
     assert result.images == []
     assert "Unexpected docker image inspect JSON shape" in result.notes[0]
+
+
+def test_extra_paths_collector_records_dirs_files_notes_and_excludes(
+    monkeypatch, tmp_path
+):
+    from enroll.harvest_collectors import paths
+
+    root = tmp_path / "include"
+    sub = root / "sub"
+    skip = root / "skip"
+    sub.mkdir(parents=True)
+    skip.mkdir()
+    keep_file = sub / "keep.conf"
+    keep_file.write_text("ok", encoding="utf-8")
+    skip_file = skip / "skip.conf"
+    skip_file.write_text("no", encoding="utf-8")
+
+    class Policy(IgnorePolicy):
+        def deny_reason_dir(self, path: str):
+            return "denied_dir" if path == str(sub) else None
+
+    def fake_stat_triplet(path: str):
+        return ("root", "root", "0755")
+
+    def fake_capture_file(**kwargs):
+        kwargs["managed_out"].append(
+            ManagedFile(
+                path=kwargs["abs_path"],
+                src_rel=kwargs["abs_path"].lstrip("/"),
+                owner="root",
+                group="root",
+                mode="0644",
+                reason=kwargs["reason"],
+            )
+        )
+        return True
+
+    monkeypatch.setattr(paths.h, "stat_triplet", fake_stat_triplet)
+    monkeypatch.setattr(paths, "capture_file", lambda *a, **kw: fake_capture_file(**kw))
+
+    ctx = _context(
+        tmp_path,
+        include=[str(root)],
+        exclude=[str(skip)],
+        policy=Policy(),
+    )
+    result = ExtraPathsCollector(
+        ctx,
+        seen_by_role={},
+        already_all=set(),
+        include_paths=[str(root)],
+        exclude_paths=[str(skip)],
+    ).collect()
+
+    managed_dirs = {d.path for d in result.managed_dirs}
+    assert str(root) in managed_dirs
+    assert str(sub) not in managed_dirs  # denied by policy
+    assert str(skip) not in managed_dirs  # pruned by exclude filter
+    assert [m.path for m in result.managed_files] == [str(keep_file)]
+    assert "User include patterns:" in result.notes
+    assert f"- {root}" in result.notes
+    assert f"- {skip}" in result.notes
+
+
+def test_extra_paths_collector_skips_already_captured_files(monkeypatch, tmp_path):
+    from enroll.harvest_collectors import paths
+
+    root = tmp_path / "include"
+    root.mkdir()
+    file_path = root / "keep.conf"
+    file_path.write_text("ok", encoding="utf-8")
+    calls: list[str] = []
+
+    monkeypatch.setattr(paths.h, "stat_triplet", lambda p: ("root", "root", "0755"))
+    monkeypatch.setattr(
+        paths, "capture_file", lambda *a, **kw: calls.append(kw["abs_path"]) or True
+    )
+
+    ctx = _context(tmp_path, include=[str(root)])
+    result = ExtraPathsCollector(
+        ctx,
+        seen_by_role={},
+        already_all={str(file_path)},
+        include_paths=[str(root)],
+    ).collect()
+
+    assert result.managed_files == []
+    assert calls == []
+
+
+def test_usr_local_custom_collector_scans_executable_bin_and_notes_cap(
+    monkeypatch, tmp_path
+):
+    from enroll.harvest_collectors import paths
+
+    captured: list[str] = []
+
+    def fake_isdir(path: str) -> bool:
+        return path in {"/usr/local/etc", "/usr/local/bin"}
+
+    def fake_walk(root: str):
+        if root == "/usr/local/etc":
+            yield root, [], ["app.conf"]
+        elif root == "/usr/local/bin":
+            yield root, [], ["tool", "not-exec"]
+
+    def fake_isfile(path: str) -> bool:
+        return path in {
+            "/usr/local/etc/app.conf",
+            "/usr/local/bin/tool",
+            "/usr/local/bin/not-exec",
+        }
+
+    def fake_stat_triplet(path: str):
+        mode = "0755" if path == "/usr/local/bin/tool" else "0644"
+        return ("root", "root", mode)
+
+    def fake_capture_file(**kwargs):
+        captured.append(kwargs["abs_path"])
+        kwargs["managed_out"].append(
+            ManagedFile(
+                path=kwargs["abs_path"],
+                src_rel=kwargs["abs_path"].lstrip("/"),
+                owner="root",
+                group="root",
+                mode="0644",
+                reason=kwargs["reason"],
+            )
+        )
+        return True
+
+    monkeypatch.setattr(paths.os.path, "isdir", fake_isdir)
+    monkeypatch.setattr(paths.os, "walk", fake_walk)
+    monkeypatch.setattr(paths.os.path, "isfile", fake_isfile)
+    monkeypatch.setattr(paths.os.path, "islink", lambda p: False)
+    monkeypatch.setattr(paths.h, "stat_triplet", fake_stat_triplet)
+    monkeypatch.setattr(paths, "capture_file", lambda *a, **kw: fake_capture_file(**kw))
+
+    ctx = _context(tmp_path)
+    result = UsrLocalCustomCollector(ctx, seen_by_role={}, already_all=set()).collect()
+
+    assert captured == ["/usr/local/etc/app.conf", "/usr/local/bin/tool"]
+    assert [m.reason for m in result.managed_files] == [
+        "usr_local_etc_custom",
+        "usr_local_bin_script",
+    ]

tests/test_harvest_collectors_package_manager.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from enroll.harvest_collectors.context import HarvestContext
+from enroll.harvest_collectors.package_manager import PackageManagerConfigCollector
+from enroll.harvest_types import ManagedFile
+from enroll.ignore import IgnorePolicy
+from enroll.pathfilter import PathFilter
+
+
+class _Backend:
+    def __init__(self, name: str):
+        self.name = name
+
+
+def _context(tmp_path: Path, backend_name: str) -> HarvestContext:
+    return HarvestContext(
+        bundle_dir=str(tmp_path / "bundle"),
+        policy=IgnorePolicy(),
+        path_filter=PathFilter(include=(), exclude=()),
+        platform={},
+        backend=_Backend(backend_name),
+        installed_pkgs={},
+        installed_names=set(),
+        owned_etc=set(),
+        etc_owner_map={},
+        topdir_to_pkgs={},
+        pkg_to_etc_paths={},
+        captured_global=set(),
+    )
+
+
+def _fake_capture(**kwargs):
+    kwargs["managed_out"].append(
+        ManagedFile(
+            path=kwargs["abs_path"],
+            src_rel=kwargs["abs_path"].lstrip("/"),
+            owner="root",
+            group="root",
+            mode="0644",
+            reason=kwargs["reason"],
+        )
+    )
+    return True
+
+
+def test_package_manager_config_collector_captures_apt_branch(monkeypatch, tmp_path):
+    from enroll.harvest_collectors import package_manager as pm
+
+    monkeypatch.setattr(
+        pm, "iter_apt_capture_paths", lambda: [("/etc/apt/a.conf", "apt")]
+    )
+    monkeypatch.setattr(pm, "capture_file", lambda *a, **kw: _fake_capture(**kw))
+
+    result = PackageManagerConfigCollector(_context(tmp_path, "dpkg"), {}).collect()
+
+    assert [m.path for m in result.apt_config_snapshot.managed_files] == [
+        "/etc/apt/a.conf"
+    ]
+    assert result.dnf_config_snapshot.managed_files == []
+
+
+def test_package_manager_config_collector_captures_dnf_branch(monkeypatch, tmp_path):
+    from enroll.harvest_collectors import package_manager as pm
+
+    monkeypatch.setattr(
+        pm, "iter_dnf_capture_paths", lambda: [("/etc/dnf/d.conf", "dnf")]
+    )
+    monkeypatch.setattr(pm, "capture_file", lambda *a, **kw: _fake_capture(**kw))
+
+    result = PackageManagerConfigCollector(_context(tmp_path, "rpm"), {}).collect()
+
+    assert result.apt_config_snapshot.managed_files == []
+    assert [m.path for m in result.dnf_config_snapshot.managed_files] == [
+        "/etc/dnf/d.conf"
+    ]
+
+
+def test_package_manager_config_collector_unknown_backend_returns_empty(tmp_path):
+    result = PackageManagerConfigCollector(_context(tmp_path, "apk"), {}).collect()
+
+    assert result.apt_config_snapshot.managed_files == []
+    assert result.dnf_config_snapshot.managed_files == []

tests/test_harvest_safety.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import stat
 from pathlib import Path
 
 import pytest
@@ -13,6 +14,8 @@ from enroll.manifest_safety import prepare_manifest_output_dir
 from enroll.harvest_safety import OutputSafetyError, prepare_new_private_dir
 from enroll.pathfilter import PathFilter
 
+import enroll.harvest_safety as hs
+
 
 class _RacePolicy(IgnorePolicy):
     def inspect_file(self, path: str):
@@ -110,3 +113,172 @@ def test_prepare_new_private_dir_rejects_symlink_parent(tmp_path: Path):
 
     with pytest.raises(OutputSafetyError, match="parent path contains a symlink"):
         prepare_new_private_dir(link / "bundle", label="harvest output")
+
+
+def test_manifest_output_dir_rejects_symlink_parent(tmp_path: Path):
+    from enroll.manifest_safety import ManifestOutputError
+
+    real = tmp_path / "real"
+    real.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(real, target_is_directory=True)
+
+    with pytest.raises(ManifestOutputError, match="parent path contains a symlink"):
+        prepare_manifest_output_dir(link / "manifest")
+
+
+def test_prepare_new_private_dir_rejects_untrusted_root_parent(
+    tmp_path: Path, monkeypatch
+):
+    import enroll.harvest_safety as hs
+
+    untrusted = tmp_path / "untrusted"
+    untrusted.mkdir()
+    if hasattr(os, "geteuid") and os.geteuid() == 0:
+        try:
+            os.chown(untrusted, 65534, -1)
+        except OSError:
+            pass
+
+    monkeypatch.setattr(hs, "_effective_uid", lambda: 0)
+    with pytest.raises(OutputSafetyError, match="not owned by root"):
+        prepare_new_private_dir(untrusted / "bundle", label="harvest output")
+
+
+def test_prepare_new_private_dir_uses_real_euid_despite_os_geteuid_monkeypatch(
+    tmp_path: Path, monkeypatch
+):
+    import enroll.harvest_safety as hs
+
+    monkeypatch.setattr(hs.os, "geteuid", lambda: 0)
+    out = prepare_new_private_dir(tmp_path / "bundle", label="harvest output")
+
+    assert out.is_dir()
+    assert (out.stat().st_mode & 0o777) == 0o700
+
+
+def test_write_text_output_file_replaces_final_symlink_not_target(tmp_path: Path):
+    from enroll.harvest_safety import write_text_output_file
+
+    target = tmp_path / "target.txt"
+    target.write_text("old\n", encoding="utf-8")
+    link = tmp_path / "report.txt"
+    link.symlink_to(target)
+
+    write_text_output_file(link, "new\n", label="test report")
+
+    assert not link.is_symlink()
+    assert link.read_text(encoding="utf-8") == "new\n"
+    assert target.read_text(encoding="utf-8") == "old\n"
+
+
+def test_safe_output_parent_does_not_descend_into_raced_symlink(
+    tmp_path: Path, monkeypatch
+):
+    import enroll.harvest_safety as hs
+
+    target = tmp_path / "target"
+    target.mkdir()
+    link = tmp_path / "link"
+    real_mkdir = os.mkdir
+
+    def racing_mkdir(path, mode=0o777, *, dir_fd=None):
+        if Path(path) == link and not link.exists():
+            link.symlink_to(target, target_is_directory=True)
+        if dir_fd is not None:
+            return real_mkdir(path, mode, dir_fd=dir_fd)
+        return real_mkdir(path, mode)
+
+    monkeypatch.setattr(hs.os, "mkdir", racing_mkdir)
+
+    with pytest.raises(OutputSafetyError, match="parent path contains a symlink"):
+        hs.ensure_safe_output_parent(link / "subdir" / "report.txt", label="report")
+
+    assert not (target / "subdir").exists()
+
+
+def _stat_result(mode: int, *, uid: int = 0) -> os.stat_result:
+    return os.stat_result((mode, 1, 1, 1, uid, 0, 0, 0, 0, 0))
+
+
+def test_effective_uid_handles_missing_geteuid(monkeypatch):
+    monkeypatch.setattr(hs, "_OS_GETEUID", None)
+    assert hs._effective_uid() is None
+
+
+def test_effective_uid_handles_geteuid_error(monkeypatch):
+    def boom():
+        raise OSError("no euid")
+
+    monkeypatch.setattr(hs, "_OS_GETEUID", boom)
+    assert hs._effective_uid() is None
+
+
+def test_trusted_root_parent_skips_checks_when_not_root(monkeypatch):
+    monkeypatch.setattr(hs, "_effective_uid", lambda: 1000)
+    hs._assert_trusted_root_parent(
+        Path("not-a-dir"), _stat_result(stat.S_IFREG | 0o644, uid=1234), label="x"
+    )
+
+
+def test_trusted_root_parent_rejects_non_directory(monkeypatch):
+    monkeypatch.setattr(hs, "_effective_uid", lambda: 0)
+    with pytest.raises(OutputSafetyError, match="parent is not a directory"):
+        hs._assert_trusted_root_parent(
+            Path("file"), _stat_result(stat.S_IFREG | 0o644), label="x"
+        )
+
+
+def test_trusted_root_parent_rejects_group_or_world_writable(monkeypatch):
+    monkeypatch.setattr(hs, "_effective_uid", lambda: 0)
+    with pytest.raises(OutputSafetyError, match="writable by group/other"):
+        hs._assert_trusted_root_parent(
+            Path("open-dir"), _stat_result(stat.S_IFDIR | 0o777), label="x"
+        )
+
+
+def test_trusted_root_parent_allows_root_owned_sticky_shared_dir(monkeypatch):
+    monkeypatch.setattr(hs, "_effective_uid", lambda: 0)
+    hs._assert_trusted_root_parent(
+        Path("tmp"), _stat_result(stat.S_IFDIR | stat.S_ISVTX | 0o777), label="x"
+    )
+
+
+def test_assert_no_existing_symlink_components_without_root_trust_still_rejects_symlink(
+    tmp_path: Path,
+):
+    real = tmp_path / "real"
+    real.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(real, target_is_directory=True)
+
+    with pytest.raises(OutputSafetyError, match="parent path contains a symlink"):
+        hs._assert_no_existing_symlink_components(
+            link / "leaf", label="x", require_trusted_root_parents=False
+        )
+
+
+def test_ensure_private_empty_dir_rejects_bad_existing_paths(tmp_path: Path):
+    file_path = tmp_path / "file"
+    file_path.write_text("x", encoding="utf-8")
+    with pytest.raises(OutputSafetyError, match="not a directory"):
+        hs.ensure_private_empty_dir(file_path, label="cache")
+
+    nonempty = tmp_path / "nonempty"
+    nonempty.mkdir()
+    (nonempty / "child").write_text("x", encoding="utf-8")
+    with pytest.raises(OutputSafetyError, match="not empty"):
+        hs.ensure_private_empty_dir(nonempty, label="cache")
+
+    real = tmp_path / "real"
+    real.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(real, target_is_directory=True)
+    with pytest.raises(OutputSafetyError, match="symlink"):
+        hs.ensure_private_empty_dir(link, label="cache")
+
+
+def test_ensure_private_empty_dir_creates_private_dir(tmp_path: Path):
+    out = hs.ensure_private_empty_dir(tmp_path / "new-cache", label="cache")
+    assert out.is_dir()
+    assert (out.stat().st_mode & 0o777) == 0o700

tests/test_manifest_safety.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from enroll.manifest_safety import (
+    ArtifactSafetyError,
+    ManifestOutputError,
+    copy_safe_artifact_file,
+    iter_safe_artifact_files,
+    prepare_manifest_output_dir,
+    safe_artifact_file,
+    validate_site_fqdn,
+)
+
+
+def test_validate_site_fqdn_accepts_and_normalises_simple_values():
+    assert validate_site_fqdn(None) is None
+    assert validate_site_fqdn("  ") is None
+    assert validate_site_fqdn("host_1.example") == "host_1.example"
+
+
+@pytest.mark.parametrize(
+    "value", ["../host", "host/name", "host\\name", "host\nname", "-bad", ".", ".."]
+)
+def test_validate_site_fqdn_rejects_path_or_inventory_injection(value: str):
+    with pytest.raises(ManifestOutputError):
+        validate_site_fqdn(value)
+
+
+def test_prepare_manifest_output_dir_allows_existing_clean_tree_in_site_mode(
+    tmp_path: Path,
+):
+    out = tmp_path / "site"
+    out.mkdir()
+    (out / ".git").mkdir()
+    (out / ".git" / "ignored-link").symlink_to(tmp_path, target_is_directory=True)
+
+    assert prepare_manifest_output_dir(out, allow_existing=True) == out
+
+
+def test_prepare_manifest_output_dir_rejects_existing_tree_symlink(tmp_path: Path):
+    out = tmp_path / "site"
+    out.mkdir()
+    (out / "bad-link").symlink_to(tmp_path, target_is_directory=True)
+
+    with pytest.raises(ManifestOutputError, match="contains a symlink"):
+        prepare_manifest_output_dir(out, allow_existing=True)
+
+
+def test_safe_artifact_file_accepts_regular_file_and_copy(tmp_path: Path):
+    bundle = tmp_path / "bundle"
+    artifact = bundle / "artifacts" / "role" / "etc" / "app.conf"
+    artifact.parent.mkdir(parents=True)
+    artifact.write_text("managed=true\n", encoding="utf-8")
+
+    assert safe_artifact_file(bundle, "role", "etc/app.conf") == artifact
+
+    dst = tmp_path / "copy.conf"
+    copy_safe_artifact_file(artifact, dst)
+    assert dst.read_text(encoding="utf-8") == "managed=true\n"
+
+
+def test_safe_artifact_file_rejects_unsafe_role_and_src(tmp_path: Path):
+    bundle = tmp_path / "bundle"
+    with pytest.raises(ArtifactSafetyError, match="must be relative"):
+        safe_artifact_file(bundle, "/role", "file")
+    with pytest.raises(ArtifactSafetyError, match="unsafe path component"):
+        safe_artifact_file(bundle, "role", "../file")
+    with pytest.raises(ArtifactSafetyError, match="NUL"):
+        safe_artifact_file(bundle, "role", "bad\x00file")
+
+
+def test_safe_artifact_file_rejects_artifacts_symlink(tmp_path: Path):
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    (bundle / "artifacts").symlink_to(tmp_path, target_is_directory=True)
+
+    with pytest.raises(ArtifactSafetyError, match="artifacts directory is a symlink"):
+        safe_artifact_file(bundle, "role", "file")
+
+
+def test_safe_artifact_file_rejects_bad_artifact_kinds(tmp_path: Path):
+    bundle = tmp_path / "bundle"
+    role_dir = bundle / "artifacts" / "role"
+    role_dir.mkdir(parents=True)
+
+    target = role_dir / "target"
+    target.write_text("x", encoding="utf-8")
+    (role_dir / "link").symlink_to(target)
+    with pytest.raises(ArtifactSafetyError, match="symlink"):
+        safe_artifact_file(bundle, "role", "link")
+
+    (role_dir / "dir-artifact").mkdir()
+    with pytest.raises(ArtifactSafetyError, match="not a regular file"):
+        safe_artifact_file(bundle, "role", "dir-artifact")
+
+    hardlink = role_dir / "hardlink"
+    os.link(target, hardlink)
+    with pytest.raises(ArtifactSafetyError, match="hardlinked"):
+        safe_artifact_file(bundle, "role", "target")
+
+
+def test_iter_safe_artifact_files_handles_missing_and_bad_role_dirs(tmp_path: Path):
+    bundle = tmp_path / "bundle"
+    assert list(iter_safe_artifact_files(bundle, "missing")) == []
+
+    role_file = bundle / "artifacts" / "role"
+    role_file.parent.mkdir(parents=True)
+    role_file.write_text("not a dir", encoding="utf-8")
+    with pytest.raises(ArtifactSafetyError, match="not a directory"):
+        list(iter_safe_artifact_files(bundle, "role"))
+
+
+def test_iter_safe_artifact_files_rejects_symlink_subdir(tmp_path: Path):
+    bundle = tmp_path / "bundle"
+    role_dir = bundle / "artifacts" / "role"
+    role_dir.mkdir(parents=True)
+    real = tmp_path / "real"
+    real.mkdir()
+    (role_dir / "linkdir").symlink_to(real, target_is_directory=True)
+
+    with pytest.raises(ArtifactSafetyError, match="directory is a symlink"):
+        list(iter_safe_artifact_files(bundle, "role"))

tests/test_package_hints.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from enroll.package_hints import (
+    add_pkgs_from_etc_topdirs,
+    hint_names,
+    maybe_add_specific_paths,
+    package_section_from_installations,
+    role_id,
+    role_name_from_pkg,
+    role_name_from_unit,
+    safe_name,
+)
+
+
+class _Backend:
+    def __init__(self, *, fail: bool = False):
+        self.fail = fail
+
+    def specific_paths_for_hints(self, hints):
+        if self.fail:
+            raise RuntimeError("backend unavailable")
+        return [f"/backend/{h}" for h in sorted(hints)]
+
+
+def test_role_name_helpers_sanitise_reserved_and_odd_names():
+    assert safe_name("pkg.name+with-dash") == "pkg_name_with_dash"
+    assert role_id("123 Camel-Case!!") == "r_123_camel_case"
+    assert role_name_from_unit("class.service") == "class"
+    assert role_name_from_pkg("flatpak") == "package_flatpak"
+
+
+def test_package_section_from_installations_filters_empty_and_unspecified():
+    assert package_section_from_installations([]) is None
+    assert (
+        package_section_from_installations([{"section": "none"}, {"group": ""}]) is None
+    )
+    assert (
+        package_section_from_installations([{"section": "z-utils"}, {"group": "admin"}])
+        == "admin"
+    )
+
+
+def test_hint_names_expands_templates_packages_and_dot_prefixes():
+    assert hint_names("postgresql@14-main.service", {"postgresql.14"}) == {
+        "postgresql@14-main",
+        "postgresql",
+        "postgresql.14",
+    }
+
+
+def test_add_pkgs_from_etc_topdirs_skips_shared_dirs():
+    pkgs: set[str] = set()
+    add_pkgs_from_etc_topdirs(
+        {"ssh", "nginx"},
+        {"ssh": {"openssh-server"}, "nginx": {"nginx"}, "nginx.d": {"nginx-extra"}},
+        pkgs,
+    )
+    assert pkgs == {"nginx", "nginx-extra"}
+
+
+def test_maybe_add_specific_paths_uses_backend_and_fallback():
+    assert maybe_add_specific_paths({"a", "b"}, _Backend()) == [
+        "/backend/a",
+        "/backend/b",
+    ]
+    assert maybe_add_specific_paths({"svc"}, _Backend(fail=True)) == [
+        "/etc/default/svc",
+        "/etc/init.d/svc",
+        "/etc/sysctl.d/svc.conf",
+    ]

tests/test_remote.py

@@ -166,6 +166,13 @@ def test_remote_harvest_happy_path(tmp_path: Path, monkeypatch):
                 return (None, _Stdout(b"alice\n"), _Stderr())
             if cmd == "mktemp -d":
                 return (None, _Stdout(b"/tmp/enroll-remote-123\n"), _Stderr())
+            if cmd.startswith("sudo -n") and " mktemp -d" in cmd:
+                return (None, _Stdout(b"/tmp/enroll-root-123\n"), _Stderr())
+            if (
+                cmd.startswith("sudo -n")
+                and " chmod 700 -- /tmp/enroll-root-123" in cmd
+            ):
+                return (None, _Stdout(b""), _Stderr())
             if cmd.startswith("chmod 700"):
                 return (None, _Stdout(b""), _Stderr())
             if cmd.startswith("sudo -n") and " harvest " in cmd:
@@ -182,6 +189,8 @@ def test_remote_harvest_happy_path(tmp_path: Path, monkeypatch):
                     msg = b"sudo: sorry, you must have a tty to run sudo\n"
                     return (None, _Stdout(b"", rc=1, err=msg), _Stderr(msg))
                 return (None, _Stdout(b"", rc=0), _Stderr(b""))
+            if cmd.startswith("sudo -n") and " rm -rf -- /tmp/enroll-root-123" in cmd:
+                return (None, _Stdout(b""), _Stderr())
             if cmd.startswith("rm -rf"):
                 return (None, _Stdout(b""), _Stderr())
 
@@ -223,6 +232,11 @@ def test_remote_harvest_happy_path(tmp_path: Path, monkeypatch):
     assert "--dangerous" in joined
     assert "--include-path" in joined
     assert "--exclude-path" in joined
+    assert "sudo -n -p '' -- mktemp -d" in joined
+    assert "--out /tmp/enroll-root-123/bundle" in joined
+    assert "--out /tmp/enroll-remote-123/bundle" not in joined
+    assert "chown -R -- alice /tmp/enroll-root-123" in joined
+    assert "tar -cz -C /tmp/enroll-root-123/bundle ." in joined
 
     # Ensure we fall back to PTY only when sudo reports it is required.
     assert any(c == "id -un" and pty is False for c, pty in calls)
@@ -508,6 +522,13 @@ def test_remote_harvest_sudo_password_retry_uses_sudo_s_and_writes_password(
 
             if cmd == "mktemp -d":
                 return (_Stdin(cmd), _Stdout(b"/tmp/enroll-remote-789\n"), _Stderr())
+            if cmd.startswith("sudo -n") and " mktemp -d" in cmd:
+                return (_Stdin(cmd), _Stdout(b"/tmp/enroll-root-789\n"), _Stderr())
+            if (
+                cmd.startswith("sudo -n")
+                and " chmod 700 -- /tmp/enroll-root-789" in cmd
+            ):
+                return (_Stdin(cmd), _Stdout(b""), _Stderr())
             if cmd.startswith("chmod 700"):
                 return (_Stdin(cmd), _Stdout(b""), _Stderr())
 
@@ -527,6 +548,8 @@ def test_remote_harvest_sudo_password_retry_uses_sudo_s_and_writes_password(
             if cmd.startswith("sudo -n") and " chown -R" in cmd:
                 return (_Stdin(cmd), _Stdout(b"", rc=0), _Stderr(b""))
 
+            if cmd.startswith("sudo -n") and " rm -rf -- /tmp/enroll-root-789" in cmd:
+                return (_Stdin(cmd), _Stdout(b"", rc=0), _Stderr(b""))
             if cmd.startswith("rm -rf"):
                 return (_Stdin(cmd), _Stdout(b"", rc=0), _Stderr(b""))
 
@@ -563,6 +586,10 @@ def test_remote_harvest_sudo_password_retry_uses_sudo_s_and_writes_password(
     sudo_s = [c for c, _pty in calls if c.startswith("sudo -S") and " harvest " in c]
     assert len(sudo_n) == 1
     assert len(sudo_s) == 1
+    joined = "\n".join([c for c, _pty in calls])
+    assert "sudo -n -p '' -- mktemp -d" in joined
+    assert "--out /tmp/enroll-root-789/bundle" in joined
+    assert "--out /tmp/enroll-remote-789/bundle" not in joined
 
     # Ensure the password was written to stdin for the -S invocation.
     assert stdin_by_cmd.get(sudo_s[0]) == ["s3cr3t\n"]