diff --git a/CHANGELOG.md b/CHANGELOG.md index c179b45..e3e6942 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 2d22408..5fdb55a 100644 --- a/README.md +++ b/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. diff --git a/enroll/cli.py b/enroll/cli.py index 974b9ca..24c8593 100644 --- a/enroll/cli.py +++ b/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" - if st.st_mode & stat.S_IWGRP: - return "directory is group-writable" + + final_reason = _root_trust_reason(p, final=True) + if final_reason: + 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 " - "./enroll.ini, ./.enroll.ini, or ~/.config/enroll/enroll.ini (or $XDG_CONFIG_HOME/enroll/enroll.ini)." + "Path to an INI config file for default options. If omitted, enroll will look for a path defined by the " + "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() - p.parent.mkdir(parents=True, exist_ok=True) - p.write_text(txt, encoding="utf-8") + write_text_output_file(out_path, txt, label="validation report") 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() - p.parent.mkdir(parents=True, exist_ok=True) - p.write_text(txt, encoding="utf-8") + write_text_output_file(out_path, txt, label="diff report") else: print(txt, end="" if txt.endswith("\n") else "\n") diff --git a/enroll/harvest_safety.py b/enroll/harvest_safety.py index 75bed5a..b6e4bc4 100644 --- a/enroll/harvest_safety.py +++ b/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: - """Reject symlinks in existing parent components of an output path.""" +def _effective_uid() -> int | None: + 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() - 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}" - ) + os.chmod(tmp_name, mode) + except OSError: + pass + os.replace(tmp_name, out) + finally: + 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): - 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 + return _mkdir_private_dir_tree(out, label=label, final_must_be_new=True) 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) - _chmod_private(out) - return out + return _mkdir_private_dir_tree(out, label=label, final_must_be_new=True) diff --git a/enroll/manifest.py b/enroll/manifest.py index 9c6806a..3ebfefe 100644 --- a/enroll/manifest.py +++ b/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-", diff --git a/enroll/manifest_safety.py b/enroll/manifest_safety.py index 41623bd..c7e64e1 100644 --- a/enroll/manifest_safety.py +++ b/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 - non-directory path. + as --fqdn site mode, may allow an existing directory but never a symlink, + 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) - except OSError: - pass - return out + return prepare_new_private_dir(out, label="manifest output") + except OutputSafetyError as e: + raise ManifestOutputError(str(e)) from e def _assert_no_symlink_components(path: Path, *, root: Path) -> None: diff --git a/enroll/remote.py b/enroll/remote.py index 0a71d4c..ecb1c27 100644 --- a/enroll/remote.py +++ b/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: diff --git a/enroll/sopsutil.py b/enroll/sopsutil.py index de36d4f..e8d31ac 100644 --- a/enroll/sopsutil.py +++ b/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( [ diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py index 7313b33..b749597 100644 --- a/tests/test_cli_helpers.py +++ b/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) diff --git a/tests/test_harvest_collectors.py b/tests/test_harvest_collectors.py index 80f43e4..94b5259 100644 --- a/tests/test_harvest_collectors.py +++ b/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), - policy=IgnorePolicy(), - path_filter=PathFilter(include=(), exclude=()), + bundle_dir=str(tmp_path / "bundle"), + policy=policy or IgnorePolicy(), + 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", + ] diff --git a/tests/test_harvest_collectors_package_manager.py b/tests/test_harvest_collectors_package_manager.py new file mode 100644 index 0000000..ca805dd --- /dev/null +++ b/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 == [] diff --git a/tests/test_harvest_safety.py b/tests/test_harvest_safety.py index c6038a7..6c5b338 100644 --- a/tests/test_harvest_safety.py +++ b/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 diff --git a/tests/test_manifest_safety.py b/tests/test_manifest_safety.py new file mode 100644 index 0000000..deb764d --- /dev/null +++ b/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")) diff --git a/tests/test_package_hints.py b/tests/test_package_hints.py new file mode 100644 index 0000000..76dd835 --- /dev/null +++ b/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", + ] diff --git a/tests/test_remote.py b/tests/test_remote.py index 1c0bfd0..fe54d33 100644 --- a/tests/test_remote.py +++ b/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"]