diff --git a/CHANGELOG.md b/CHANGELOG.md index e3e6942..c179b45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,6 @@ * 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 5fdb55a..2d22408 100644 --- a/README.md +++ b/README.md @@ -196,8 +196,6 @@ 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) @@ -287,15 +285,12 @@ 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 wish to opt in to collecting everything, use `--dangerous` mode, but be aware of what it means: +If you opt in to collecting everything: ### `--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**. -**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! +If you intend to keep harvests/manifests long-term (especially in git), 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 24c8593..974b9ca 100644 --- a/enroll/cli.py +++ b/enroll/cli.py @@ -22,7 +22,6 @@ 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, @@ -113,15 +112,6 @@ 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 []: @@ -145,75 +135,14 @@ 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. 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. + from PATH while harvesting and enforcing state. """ if entry == "": @@ -223,21 +152,16 @@ 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 - - final_reason = _root_trust_reason(p, final=True) - if final_reason: - return final_reason + if st.st_mode & stat.S_IWOTH: + return "directory is world-writable" + if st.st_mode & stat.S_IWGRP: + return "directory is group-writable" return None @@ -429,7 +353,7 @@ def _resolve_sops_out_file(out: Optional[str], *, hint: str) -> Path: def _tar_dir_to(path_dir: Path, tar_path: Path) -> None: - ensure_safe_output_parent(tar_path, label="harvest tar output") + tar_path.parent.mkdir(parents=True, exist_ok=True) 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=".") @@ -439,7 +363,7 @@ def _encrypt_harvest_dir_to_sops( bundle_dir: Path, out_file: Path, fps: list[str] ) -> Path: out_file = Path(out_file) - ensure_safe_output_parent(out_file, label="encrypted harvest output") + out_file.parent.mkdir(parents=True, exist_ok=True) # Create the tarball alongside the output file (keeps filesystem permissions/locality sane). fd, tmp_tgz = tempfile.mkstemp( @@ -502,8 +426,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 a path defined by the " - "ENROLL_CONFIG environment variable , ~/.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 " + "./enroll.ini, ./.enroll.ini, or ~/.config/enroll/enroll.ini (or $XDG_CONFIG_HOME/enroll/enroll.ini)." ), ) p.add_argument( @@ -963,11 +887,6 @@ 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. @@ -1102,7 +1021,9 @@ def main() -> None: out_path = getattr(args, "out", None) if out_path: - write_text_output_file(out_path, txt, label="validation report") + p = Path(out_path).expanduser() + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(txt, encoding="utf-8") else: sys.stdout.write(txt) @@ -1173,7 +1094,9 @@ def main() -> None: txt = format_report(report, fmt=str(getattr(args, "format", "text"))) out_path = getattr(args, "out", None) if out_path: - write_text_output_file(out_path, txt, label="diff report") + p = Path(out_path).expanduser() + 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") diff --git a/enroll/harvest_safety.py b/enroll/harvest_safety.py index b6e4bc4..75bed5a 100644 --- a/enroll/harvest_safety.py +++ b/enroll/harvest_safety.py @@ -2,7 +2,6 @@ from __future__ import annotations import os import stat -import tempfile from pathlib import Path @@ -10,13 +9,6 @@ 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) @@ -25,111 +17,8 @@ def _chmod_private(path: Path) -> None: pass -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. - """ +def _assert_no_existing_symlink_components(path: Path, *, label: str) -> None: + """Reject symlinks in existing parent components of an output path.""" parts = path.parts if not parts: @@ -141,96 +30,52 @@ def _assert_no_existing_symlink_components( 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: - os.chmod(tmp_name, mode) - except OSError: - pass - os.replace(tmp_name, out) - finally: - try: - os.unlink(tmp_name) - except FileNotFoundError: - pass - return out + 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 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) - return _mkdir_private_dir_tree(out, label=label, final_must_be_new=True) + 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 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. """ @@ -254,4 +99,6 @@ def ensure_private_empty_dir(path: str | Path, *, label: str = "output") -> Path _chmod_private(out) return out - return _mkdir_private_dir_tree(out, label=label, final_must_be_new=True) + out.mkdir(parents=True, exist_ok=False, mode=0o700) + _chmod_private(out) + return out diff --git a/enroll/manifest.py b/enroll/manifest.py index 3ebfefe..9c6806a 100644 --- a/enroll/manifest.py +++ b/enroll/manifest.py @@ -10,7 +10,6 @@ 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 ( @@ -92,7 +91,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) - ensure_safe_output_parent(tar_path, label="manifest tar output") + tar_path.parent.mkdir(parents=True, exist_ok=True) # Collect paths (dirs + files) paths: list[Path] = [src_dir] @@ -145,7 +144,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) - ensure_safe_output_parent(out_file, label="encrypted manifest output") + out_file.parent.mkdir(parents=True, exist_ok=True) fd, tmp_tgz = tempfile.mkstemp( prefix=".enroll-manifest-", diff --git a/enroll/manifest_safety.py b/enroll/manifest_safety.py index c7e64e1..41623bd 100644 --- a/enroll/manifest_safety.py +++ b/enroll/manifest_safety.py @@ -7,12 +7,6 @@ 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.""" @@ -111,13 +105,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 unsafe root output paths. + """Create a manifest output directory, refusing to overwrite anything. 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, - non-directory path, symlinked parent, or root-unsafe parent. + as --fqdn site mode, may allow an existing directory but never a symlink or + non-directory path. """ out = Path(out_dir).expanduser() @@ -126,12 +120,6 @@ 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( @@ -143,10 +131,12 @@ def prepare_manifest_output_dir( ) _assert_no_output_symlinks(out) return out + out.mkdir(parents=True, exist_ok=False, mode=0o700) try: - return prepare_new_private_dir(out, label="manifest output") - except OutputSafetyError as e: - raise ManifestOutputError(str(e)) from e + os.chmod(out, 0o700) + except OSError: + pass + return out def _assert_no_symlink_components(path: Path, *, root: Path) -> None: diff --git a/enroll/remote.py b/enroll/remote.py index ecb1c27..0a71d4c 100644 --- a/enroll/remote.py +++ b/enroll/remote.py @@ -578,14 +578,11 @@ 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)}") @@ -593,35 +590,10 @@ 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, @@ -663,10 +635,9 @@ 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(chown_target)}" + f"{shlex.quote(resolved_user)} {shlex.quote(rbundle)}" ) rc, out, err = _ssh_run_sudo( ssh, @@ -707,19 +678,7 @@ def _remote_harvest( _safe_extract_tar(tf, local_out_dir) finally: - # 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 + # Cleanup remote tmpdir even on failure. if rtmp: _ssh_run(ssh, f"rm -rf -- {shlex.quote(rtmp)}") try: diff --git a/enroll/sopsutil.py b/enroll/sopsutil.py index e8d31ac..de36d4f 100644 --- a/enroll/sopsutil.py +++ b/enroll/sopsutil.py @@ -7,8 +7,6 @@ import tempfile from pathlib import Path from typing import Iterable, List, Optional -from .harvest_safety import ensure_safe_output_parent - class SopsError(RuntimeError): pass @@ -48,7 +46,7 @@ def encrypt_file_binary( sops = require_sops_cmd() src_path = Path(src_path) dst_path = Path(dst_path) - ensure_safe_output_parent(dst_path, label="sops output") + dst_path.parent.mkdir(parents=True, exist_ok=True) res = subprocess.run( [ @@ -100,7 +98,7 @@ def decrypt_file_binary_to( sops = require_sops_cmd() src_path = Path(src_path) dst_path = Path(dst_path) - ensure_safe_output_parent(dst_path, label="sops output") + dst_path.parent.mkdir(parents=True, exist_ok=True) res = subprocess.run( [ diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py index b749597..7313b33 100644 --- a/tests/test_cli_helpers.py +++ b/tests/test_cli_helpers.py @@ -243,20 +243,3 @@ 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 94b5259..80f43e4 100644 --- a/tests/test_harvest_collectors.py +++ b/tests/test_harvest_collectors.py @@ -1,11 +1,8 @@ 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, ManagedFile, SysctlSnapshot +from enroll.harvest_types import FirewallRuntimeSnapshot, SysctlSnapshot from enroll.ignore import IgnorePolicy from enroll.pathfilter import PathFilter @@ -14,11 +11,11 @@ class _Backend: name = "dpkg" -def _context(tmp_path: Path, *, include=(), exclude=(), policy=None) -> HarvestContext: +def _context(tmp_path): return HarvestContext( - bundle_dir=str(tmp_path / "bundle"), - policy=policy or IgnorePolicy(), - path_filter=PathFilter(include=include, exclude=exclude), + bundle_dir=str(tmp_path), + policy=IgnorePolicy(), + path_filter=PathFilter(include=(), exclude=()), platform={}, backend=_Backend(), installed_pkgs={}, @@ -248,149 +245,3 @@ 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 deleted file mode 100644 index ca805dd..0000000 --- a/tests/test_harvest_collectors_package_manager.py +++ /dev/null @@ -1,84 +0,0 @@ -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 6c5b338..c6038a7 100644 --- a/tests/test_harvest_safety.py +++ b/tests/test_harvest_safety.py @@ -1,7 +1,6 @@ from __future__ import annotations import os -import stat from pathlib import Path import pytest @@ -14,8 +13,6 @@ 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): @@ -113,172 +110,3 @@ 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 deleted file mode 100644 index deb764d..0000000 --- a/tests/test_manifest_safety.py +++ /dev/null @@ -1,126 +0,0 @@ -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 deleted file mode 100644 index 76dd835..0000000 --- a/tests/test_package_hints.py +++ /dev/null @@ -1,70 +0,0 @@ -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 fe54d33..1c0bfd0 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -166,13 +166,6 @@ 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: @@ -189,8 +182,6 @@ 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()) @@ -232,11 +223,6 @@ 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) @@ -522,13 +508,6 @@ 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()) @@ -548,8 +527,6 @@ 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"")) @@ -586,10 +563,6 @@ 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"]