Compare commits

..

No commits in common. "c7a6bfe97960d29d7634a72f24fe81438c32664b" and "1e61ae2ff94433d99aeae9bfe045d8a73ebce900" have entirely different histories.

15 changed files with 8 additions and 482 deletions

View file

@ -13,8 +13,6 @@
* Add support for detecting Flatpaks and Snaps. * Add support for detecting Flatpaks and Snaps.
* Stricter validation of harvests to ensure that they meet the schema and don't contain unsafe artifacts (e.g symlinks pointing outside the artifact tree) * Stricter validation of harvests to ensure that they meet the schema and don't contain unsafe artifacts (e.g symlinks pointing outside the artifact tree)
* Perform harvest validation before trying to manifest from it. * 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.
# 0.6.0 # 0.6.0

View file

@ -788,9 +788,6 @@ SOPS mode:
The renderers do not know about SOPS. The renderers do not know about SOPS.
Note: Manifest deliberately hooks into validate() to make sure the harvest meets the schema and
doesn't contain dangerous tamperings before turning it into config management code.
--- ---
## 12. The renderer-neutral `CMModule` model ## 12. The renderer-neutral `CMModule` model
@ -1383,14 +1380,11 @@ This is intended to answer “what did Enroll collect and why?”
4. every `managed_file.src_rel` points to an artifact file, 4. every `managed_file.src_rel` points to an artifact file,
5. firewall runtime generated artifacts exist, 5. firewall runtime generated artifacts exist,
6. there are no unreferenced artifact files, reported as warnings. 6. there are no unreferenced artifact files, reported as warnings.
7. there are no malicious or unsafe bits such as symlinks/hardlinks etc traversing out of the artifact tree
It returns a `ValidationResult` with `errors`, `warnings`, `ok()`, `to_dict()`, and `to_text()`. It returns a `ValidationResult` with `errors`, `warnings`, `ok()`, `to_dict()`, and `to_text()`.
The CLI supports local schema override with `--schema`, warning failure with `--fail-on-warnings`, JSON/text output, and `--out`. The CLI supports local schema override with `--schema`, warning failure with `--fail-on-warnings`, JSON/text output, and `--out`.
Note that manifest() hooks into validate() to make sure the harvest is safe before rendering it into config management code.
--- ---
## 19. Remote harvesting ## 19. Remote harvesting

View file

@ -106,9 +106,6 @@ Harvest state about a host and write a harvest bundle.
* If neither is provided, and Enroll detects an encrypted key in an interactive session, it will still fall back to prompting on-demand. * If neither is provided, and Enroll detects an encrypted key in an interactive session, it will still fall back to prompting on-demand.
* In non-interactive sessions, pass `--ask-key-passphrase` or `--ssh-key-passphrase-env ENV_VAR` when using encrypted private keys. * In non-interactive sessions, pass `--ask-key-passphrase` or `--ssh-key-passphrase-env ENV_VAR` when using encrypted private keys.
* Note: `--ask-key-passphrase` and `--ssh-key-passphrase-env` are mutually exclusive. * Note: `--ask-key-passphrase` and `--ssh-key-passphrase-env` are mutually exclusive.
- Root PATH safety:
- when run as root, Enroll warns and asks for confirmation if `PATH` contains `.`, an empty/relative entry, or a group/world-writable directory.
- use `--assume-safe-path` for trusted non-interactive automation where that `PATH` is intentional.
Examples (encrypted SSH key) Examples (encrypted SSH key)
@ -281,8 +278,6 @@ enroll validate ./harvest --fail-on-warnings
By default, `enroll` does **not** assume how you handle secrets in Ansible. It will attempt to avoid harvesting likely sensitive data (private keys, passwords, tokens, etc.). This can mean it skips some config files you may ultimately want to manage. By default, `enroll` does **not** assume how you handle secrets in Ansible. It will attempt to avoid harvesting likely sensitive data (private keys, passwords, tokens, etc.). This can mean it skips some config files you may ultimately want to manage.
Safe-mode content scanning is intentionally conservative. It treats common assignment-style credential keys as sensitive, including names such as `password`, `client_secret`, `secret_key`, `auth_token`, `api_key`, `aws_access_key_id`, `aws_secret_access_key`, `azure_client_secret`, `GOOGLE_APPLICATION_CREDENTIALS`, and service-account key names.
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. 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 opt in to collecting everything:

View file

@ -4,7 +4,6 @@ import argparse
import configparser import configparser
import json import json
import os import os
import stat
import sys import sys
import tarfile import tarfile
import tempfile import tempfile
@ -135,83 +134,6 @@ def _split_list_value(v: str) -> list[str]:
return [raw] if raw else [] return [raw] if raw else []
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
when Enroll is run as root because Enroll deliberately invokes host tools
from PATH while harvesting and enforcing state.
"""
if entry == "":
return "empty PATH entry resolves to the current directory"
if entry == ".":
return "'.' resolves to the current directory"
if not os.path.isabs(entry):
return "relative PATH entry resolves from the current directory"
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"
return None
def _unsafe_root_path_reasons(path_value: Optional[str] = None) -> list[str]:
"""Return unsafe PATH entries that should make root execution interactive."""
raw = os.environ.get("PATH", "") if path_value is None else str(path_value)
out: list[str] = []
for entry in raw.split(os.pathsep):
reason = _path_entry_is_unsafe(entry)
if reason:
label = entry if entry else "<empty>"
out.append(f"{label}: {reason}")
return out
def _is_effective_root() -> bool:
geteuid = getattr(os, "geteuid", None)
return bool(geteuid is not None and geteuid() == 0)
def _confirm_root_path_safety(*, force: bool = False) -> None:
"""Prompt before running as root with a PATH that trusts writable entries."""
if force or not _is_effective_root():
return
reasons = _unsafe_root_path_reasons()
if not reasons:
return
details = "\n".join(f" - {r}" for r in reasons)
msg = (
"warning: enroll is running as root and PATH contains entries that "
"could allow an untrusted binary to be executed:\n"
f"{details}\n"
)
if not sys.stdin.isatty():
raise SystemExit(
msg + "error: refusing to continue non-interactively. Re-run with "
"--assume-safe-path if you intentionally trust this PATH."
)
print(msg, file=sys.stderr, end="")
answer = input("Are you sure you want to continue? [y/N] ")
if answer.strip().lower() not in {"y", "yes"}:
raise SystemExit("aborted: unsafe root PATH was not confirmed")
def _section_to_argv( def _section_to_argv(
p: argparse.ArgumentParser, cfg: configparser.ConfigParser, section: str p: argparse.ArgumentParser, cfg: configparser.ConfigParser, section: str
) -> list[str]: ) -> list[str]:
@ -437,21 +359,6 @@ def _add_config_args(p: argparse.ArgumentParser) -> None:
) )
def _add_path_safety_args(
p: argparse.ArgumentParser, *, default: object = False
) -> None:
p.add_argument(
"--assume-safe-path",
action="store_true",
default=default,
help=(
"When running as root, continue without confirmation even if PATH "
"contains '.', an empty/relative entry, or a group/world-writable "
"directory. Intended for trusted non-interactive automation."
),
)
def _add_remote_args(p: argparse.ArgumentParser) -> None: def _add_remote_args(p: argparse.ArgumentParser) -> None:
p.add_argument( p.add_argument(
"--remote-host", "--remote-host",
@ -525,12 +432,10 @@ def main() -> None:
version=f"{get_enroll_version()}", version=f"{get_enroll_version()}",
) )
_add_config_args(ap) _add_config_args(ap)
_add_path_safety_args(ap)
sub = ap.add_subparsers(dest="cmd", required=True) sub = ap.add_subparsers(dest="cmd", required=True)
h = sub.add_parser("harvest", help="Harvest service/package/config state") h = sub.add_parser("harvest", help="Harvest service/package/config state")
_add_config_args(h) _add_config_args(h)
_add_path_safety_args(h, default=argparse.SUPPRESS)
_add_remote_args(h) _add_remote_args(h)
h.add_argument( h.add_argument(
"--out", "--out",
@ -583,7 +488,6 @@ def main() -> None:
"manifest", help="Render configuration-management code from a harvest" "manifest", help="Render configuration-management code from a harvest"
) )
_add_config_args(m) _add_config_args(m)
_add_path_safety_args(m, default=argparse.SUPPRESS)
m.add_argument( m.add_argument(
"--harvest", "--harvest",
required=True, required=True,
@ -618,7 +522,6 @@ def main() -> None:
help="Harvest state, then manifest configuration-management code, in one shot", help="Harvest state, then manifest configuration-management code, in one shot",
) )
_add_config_args(s) _add_config_args(s)
_add_path_safety_args(s, default=argparse.SUPPRESS)
_add_remote_args(s) _add_remote_args(s)
s.add_argument( s.add_argument(
"--harvest", "--harvest",
@ -679,7 +582,6 @@ def main() -> None:
d = sub.add_parser("diff", help="Compare two harvests and report differences") d = sub.add_parser("diff", help="Compare two harvests and report differences")
_add_config_args(d) _add_config_args(d)
_add_path_safety_args(d, default=argparse.SUPPRESS)
d.add_argument( d.add_argument(
"--old", "--old",
required=True, required=True,
@ -801,7 +703,6 @@ def main() -> None:
e = sub.add_parser("explain", help="Explain a harvest state.json") e = sub.add_parser("explain", help="Explain a harvest state.json")
_add_config_args(e) _add_config_args(e)
_add_path_safety_args(e, default=argparse.SUPPRESS)
e.add_argument( e.add_argument(
"harvest", "harvest",
help=( help=(
@ -830,7 +731,6 @@ def main() -> None:
"validate", help="Validate a harvest bundle (state.json + artifacts)" "validate", help="Validate a harvest bundle (state.json + artifacts)"
) )
_add_config_args(v) _add_config_args(v)
_add_path_safety_args(v, default=argparse.SUPPRESS)
v.add_argument( v.add_argument(
"harvest", "harvest",
help=( help=(
@ -887,8 +787,6 @@ def main() -> None:
) )
args = ap.parse_args(argv) args = ap.parse_args(argv)
_confirm_root_path_safety(force=bool(getattr(args, "assume_safe_path", False)))
# Preserve historical defaults for remote harvesting unless ssh_config lookup is enabled. # Preserve historical defaults for remote harvesting unless ssh_config lookup is enabled.
# This lets ssh_config values take effect when the user did not explicitly set # This lets ssh_config values take effect when the user did not explicitly set
# --remote-user / --remote-port. # --remote-user / --remote-port.

View file

@ -46,41 +46,9 @@ DEFAULT_ALLOW_BINARY_GLOBS = [
"/etc/pki/rpm-gpg/*", "/etc/pki/rpm-gpg/*",
] ]
# Conservative secret patterns for default/safe harvesting. These are
# intentionally biased towards false positives: operators can opt in with
# --dangerous or targeted include/exclude review when a file is genuinely
# needed.
#
# The assignment pattern catches INI/YAML/JSON/TOML-ish keys such as:
# password: hunter2
# "client_secret": "..."
# aws_secret_access_key = ...
# GOOGLE_APPLICATION_CREDENTIALS=/path/to/key.json
SENSITIVE_CONTENT_PATTERNS = [ SENSITIVE_CONTENT_PATTERNS = [
re.compile(rb"-----BEGIN (RSA |EC |OPENSSH |DSA |)PRIVATE KEY-----"), re.compile(rb"-----BEGIN (RSA |EC |OPENSSH |)PRIVATE KEY-----"),
re.compile( re.compile(rb"(?i)\bpassword\s*="),
rb"""(?ix)
(^|[^A-Za-z0-9])
[\"']?
(
[A-Za-z0-9_.-]*
(
password|passwd|passphrase|
token|auth[_-]?token|access[_-]?token|refresh[_-]?token|
secret|client[_-]?secret|secret[_-]?key|
api[_-]?key|access[_-]?key|private[_-]?key|
credential|credentials|
aws[_-]?access[_-]?key[_-]?id|aws[_-]?secret[_-]?access[_-]?key|
azure[_-]?client[_-]?secret|azure[_-]?tenant[_-]?id|azure[_-]?client[_-]?id|
google[_-]?application[_-]?credentials|gcp[_-]?service[_-]?account|
service[_-]?account[_-]?key
)
[A-Za-z0-9_.-]*
)
[\"']?
\s*[:=]
"""
),
re.compile(rb"(?i)\b(pass|passwd|token|secret|api[_-]?key)\b"), re.compile(rb"(?i)\b(pass|passwd|token|secret|api[_-]?key)\b"),
] ]

View file

@ -9,7 +9,6 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple from typing import Any, Dict, List, Optional, Set, Tuple
from .manifest_safety import ArtifactSafetyError, safe_artifact_file
from .yamlutil import yaml_dump_mapping, yaml_load_mapping from .yamlutil import yaml_dump_mapping, yaml_load_mapping
@ -158,9 +157,8 @@ def jinjify_artifact(
if not (jt_enabled and jt_exe and can_jinjify_path(dest_path)): if not (jt_enabled and jt_exe and can_jinjify_path(dest_path)):
return None return None
try: artifact_path = Path(bundle_dir) / "artifacts" / artifact_role / src_rel
artifact_path = safe_artifact_file(bundle_dir, artifact_role, src_rel) if not artifact_path.is_file():
except (ArtifactSafetyError, FileNotFoundError):
return None return None
try: try:

View file

@ -10,7 +10,6 @@ from typing import List, Optional
from .ansible import manifest_from_bundle_dir as manifest_ansible_from_bundle_dir 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 .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 .salt import manifest_from_bundle_dir as manifest_salt_from_bundle_dir
from .manifest_safety import validate_site_fqdn
from .remote import _safe_extract_tar from .remote import _safe_extract_tar
from .sopsutil import ( from .sopsutil import (
decrypt_file_binary_to, decrypt_file_binary_to,
@ -195,7 +194,6 @@ def manifest(
target = (target or "ansible").strip().lower() target = (target or "ansible").strip().lower()
if target not in {"ansible", "puppet", "salt"}: if target not in {"ansible", "puppet", "salt"}:
raise ValueError(f"unsupported manifest target: {target!r}") raise ValueError(f"unsupported manifest target: {target!r}")
fqdn = validate_site_fqdn(fqdn)
sops_mode = bool(sops_fingerprints) sops_mode = bool(sops_fingerprints)

View file

@ -1,7 +1,6 @@
from __future__ import annotations from __future__ import annotations
import os import os
import re
import shutil import shutil
import stat import stat
from pathlib import Path from pathlib import Path
@ -16,78 +15,6 @@ class ManifestOutputError(RuntimeError):
"""Raised when a manifest output path is unsafe to use.""" """Raised when a manifest output path is unsafe to use."""
_SITE_FQDN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,252}$")
def validate_site_fqdn(value: str | None) -> str | None:
"""Validate the optional site-mode host name/FQDN.
Renderers use this value in inventory data and, for Ansible, in output
paths. Keep it deliberately conservative so it cannot become a path
separator, absolute path, YAML/INI newline injection, or shell-ish text in
generated documentation/commands.
"""
if value is None:
return None
text = str(value).strip()
if not text:
return None
if any(ch in text for ch in ("/", "\\", "\x00", "\n", "\r")):
raise ManifestOutputError(
"--fqdn contains unsafe path or newline characters; use a simple "
"host/inventory name"
)
if text in {".", ".."} or not _SITE_FQDN_RE.fullmatch(text):
raise ManifestOutputError(
"--fqdn must start with a letter or digit and contain only "
"letters, digits, dot, underscore, or hyphen"
)
return text
def _assert_no_output_symlinks(root: Path) -> None:
"""Reject pre-existing symlinks in an output tree we are about to merge into.
Non-site mode refuses existing output directories entirely. Site/FQDN modes
intentionally accumulate multiple nodes into one tree, so reject symlinks in
the tree before merging to avoid writes being redirected outside *root*.
Version-control metadata can contain implementation-specific entries and is
not part of Enroll's generated layout, so it is pruned from this check.
"""
skip_dirs = {".git", ".hg", ".svn"}
for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
dirpath_p = Path(dirpath)
for dirname in list(dirnames):
if dirname in skip_dirs:
dirnames.remove(dirname)
continue
p = dirpath_p / dirname
try:
st = p.lstat()
except FileNotFoundError:
continue
if stat.S_ISLNK(st.st_mode):
raise ManifestOutputError(
f"manifest output tree contains a symlink; refusing to merge: {p}"
)
for filename in filenames:
if filename in skip_dirs:
continue
p = dirpath_p / filename
try:
st = p.lstat()
except FileNotFoundError:
continue
if stat.S_ISLNK(st.st_mode):
raise ManifestOutputError(
f"manifest output tree contains a symlink; refusing to merge: {p}"
)
def _safe_relative_path(value: str, *, field: str) -> Path: def _safe_relative_path(value: str, *, field: str) -> Path:
text = str(value or "").strip() text = str(value or "").strip()
if not text: if not text:
@ -129,7 +56,6 @@ def prepare_manifest_output_dir(
raise ManifestOutputError( raise ManifestOutputError(
f"manifest output path exists but is not a directory: {out}" f"manifest output path exists but is not a directory: {out}"
) )
_assert_no_output_symlinks(out)
return out return out
out.mkdir(parents=True, exist_ok=False) out.mkdir(parents=True, exist_ok=False)
return out return out

View file

@ -577,7 +577,7 @@ def _remote_harvest(
rtmp = out.strip() rtmp = out.strip()
# Be explicit: restrict the remote staging area to the current user. # Be explicit: restrict the remote staging area to the current user.
rc, out, err = _ssh_run(ssh, f"chmod 700 -- {shlex.quote(rtmp)}") rc, out, err = _ssh_run(ssh, f"chmod 700 {rtmp}")
if rc != 0: if rc != 0:
raise RuntimeError(f"Remote chmod failed: {err.strip()}") raise RuntimeError(f"Remote chmod failed: {err.strip()}")
@ -627,10 +627,7 @@ def _remote_harvest(
"Unable to determine remote username for chown. " "Unable to determine remote username for chown. "
"Pass --remote-user explicitly or use --no-sudo." "Pass --remote-user explicitly or use --no-sudo."
) )
chown_cmd = ( chown_cmd = f"chown -R {resolved_user} {rbundle}"
"chown -R -- "
f"{shlex.quote(resolved_user)} {shlex.quote(rbundle)}"
)
rc, out, err = _ssh_run_sudo( rc, out, err = _ssh_run_sudo(
ssh, ssh,
chown_cmd, chown_cmd,
@ -647,7 +644,7 @@ def _remote_harvest(
) )
# Stream a tarball back to the local machine (avoid creating a tar file on the remote). # Stream a tarball back to the local machine (avoid creating a tar file on the remote).
cmd = f"tar -cz -C {shlex.quote(rbundle)} ." cmd = f"tar -cz -C {rbundle} ."
_stdin, stdout, stderr = ssh.exec_command(cmd) # nosec _stdin, stdout, stderr = ssh.exec_command(cmd) # nosec
with open(local_tgz, "wb") as f: with open(local_tgz, "wb") as f:
while True: while True:
@ -672,7 +669,7 @@ def _remote_harvest(
finally: finally:
# Cleanup remote tmpdir even on failure. # Cleanup remote tmpdir even on failure.
if rtmp: if rtmp:
_ssh_run(ssh, f"rm -rf -- {shlex.quote(rtmp)}") _ssh_run(ssh, f"rm -rf {rtmp}")
try: try:
sftp.close() sftp.close()
ssh.close() ssh.close()

View file

@ -1,21 +1,7 @@
import sys import sys
from pathlib import Path from pathlib import Path
import pytest
# Ensure repository root is on sys.path so `import enroll` resolves to the local package. # Ensure repository root is on sys.path so `import enroll` resolves to the local package.
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path: if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT))
@pytest.fixture(autouse=True)
def _disable_cli_root_path_prompt_by_default(monkeypatch):
"""Keep CLI tests deterministic when the test runner itself runs as root.
Individual tests that cover the root PATH guard can override this monkeypatch.
"""
import enroll.cli as cli
monkeypatch.setattr(cli, "_is_effective_root", lambda: False)

View file

@ -73,64 +73,6 @@ def test_cli_manifest_subcommand_calls_manifest(monkeypatch, tmp_path):
assert called["target"] == "ansible" assert called["target"] == "ansible"
def test_cli_force_unsafe_path_before_subcommand_reaches_guard(monkeypatch, tmp_path):
seen = {}
def fake_confirm(*, force: bool = False) -> None:
seen["force"] = force
def fake_manifest(_harvest_dir: str, _out_dir: str, **_kwargs):
return None
monkeypatch.setattr(cli, "_confirm_root_path_safety", fake_confirm)
monkeypatch.setattr(cli, "manifest", fake_manifest)
monkeypatch.setattr(
sys,
"argv",
[
"enroll",
"--assume-safe-path",
"manifest",
"--harvest",
str(tmp_path / "bundle"),
"--out",
str(tmp_path / "ansible"),
],
)
cli.main()
assert seen["force"] is True
def test_cli_force_unsafe_path_after_subcommand_reaches_guard(monkeypatch, tmp_path):
seen = {}
def fake_confirm(*, force: bool = False) -> None:
seen["force"] = force
def fake_manifest(_harvest_dir: str, _out_dir: str, **_kwargs):
return None
monkeypatch.setattr(cli, "_confirm_root_path_safety", fake_confirm)
monkeypatch.setattr(cli, "manifest", fake_manifest)
monkeypatch.setattr(
sys,
"argv",
[
"enroll",
"manifest",
"--assume-safe-path",
"--harvest",
str(tmp_path / "bundle"),
"--out",
str(tmp_path / "ansible"),
],
)
cli.main()
assert seen["force"] is True
def test_cli_manifest_target_puppet_is_forwarded(monkeypatch, tmp_path): def test_cli_manifest_target_puppet_is_forwarded(monkeypatch, tmp_path):
called = {} called = {}

View file

@ -2,7 +2,6 @@ from __future__ import annotations
import argparse import argparse
import configparser import configparser
import os
import types import types
import textwrap import textwrap
from pathlib import Path from pathlib import Path
@ -176,70 +175,3 @@ def test_resolve_sops_out_file(tmp_path: Path, monkeypatch):
cli._resolve_sops_out_file(out=None, hint="bundle.tar.gz") cli._resolve_sops_out_file(out=None, hint="bundle.tar.gz")
== fake_cache.dir / "harvest.tar.gz.sops" == fake_cache.dir / "harvest.tar.gz.sops"
) )
def test_unsafe_root_path_reasons_flags_current_and_writable_dirs(tmp_path: Path):
from enroll.cli import _unsafe_root_path_reasons
group_writable = tmp_path / "group-writable"
world_writable = tmp_path / "world-writable"
safe = tmp_path / "safe"
group_writable.mkdir()
world_writable.mkdir()
safe.mkdir()
group_writable.chmod(0o775)
world_writable.chmod(0o777)
safe.chmod(0o755)
reasons = _unsafe_root_path_reasons(
os.pathsep.join(
[
"",
".",
"relative-bin",
str(group_writable),
str(world_writable),
str(safe),
]
)
)
text = "\n".join(reasons)
assert "<empty>: empty PATH entry" in text
assert "'.' resolves" in text
assert "relative-bin: relative PATH entry" in text
assert f"{group_writable}: directory is group-writable" in text
assert f"{world_writable}: directory is world-writable" in text
assert str(safe) not in text
def test_confirm_root_path_safety_refuses_noninteractive(monkeypatch):
from enroll import cli
monkeypatch.setattr(cli, "_is_effective_root", lambda: True)
monkeypatch.setattr(
cli,
"_unsafe_root_path_reasons",
lambda path_value=None: [".: '.' resolves to the current directory"],
)
monkeypatch.setattr(cli.sys.stdin, "isatty", lambda: False)
try:
cli._confirm_root_path_safety(force=False)
except SystemExit as e:
assert "--assume-safe-path" in str(e)
else: # pragma: no cover - defensive assertion path
raise AssertionError("expected SystemExit")
def test_confirm_root_path_safety_force_skips_prompt(monkeypatch):
from enroll import cli
monkeypatch.setattr(cli, "_is_effective_root", lambda: True)
monkeypatch.setattr(
cli,
"_unsafe_root_path_reasons",
lambda path_value=None: [".: '.' resolves to the current directory"],
)
cli._confirm_root_path_safety(force=True)

View file

@ -172,40 +172,6 @@ def test_deny_reason_private_key(tmp_path: Path):
assert reason == "sensitive_content" assert reason == "sensitive_content"
def test_deny_reason_sensitive_common_assignment_keys(tmp_path: Path):
pol = IgnorePolicy()
cases = {
"password_yaml": "password: hunter2\n",
"password_json": '{"password": "hunter2"}\n',
"db_password": "db_password: hunter2\n",
"client_secret": "client_secret: abc123\n",
"secret_key": "secret_key = abc123\n",
"auth_token": "auth_token: abc123\n",
"passphrase": "passphrase: abc123\n",
"credentials": "credentials = abc123\n",
}
for name, text in cases.items():
config = tmp_path / name
config.write_text(text, encoding="utf-8")
assert pol.deny_reason(str(config)) == "sensitive_content", name
def test_deny_reason_sensitive_common_cloud_assignment_keys(tmp_path: Path):
pol = IgnorePolicy()
cases = {
"aws_access_key_id": "aws_access_key_id = AKIAIOSFODNN7EXAMPLE\n",
"aws_secret_access_key": "aws_secret_access_key = wJalrXUtnFEMI/K7MDENG/bPxRfiCY\n",
"azure_client_secret": "azure_client_secret: abc123\n",
"google_application_credentials": "GOOGLE_APPLICATION_CREDENTIALS=/etc/app/key.json\n",
"gcp_service_account": "gcp_service_account: svc@example.iam.gserviceaccount.com\n",
"service_account_key": "service_account_key: abc123\n",
}
for name, text in cases.items():
config = tmp_path / name
config.write_text(text, encoding="utf-8")
assert pol.deny_reason(str(config)) == "sensitive_content", name
def test_deny_reason_too_large(tmp_path: Path): def test_deny_reason_too_large(tmp_path: Path):
pol = IgnorePolicy(max_file_bytes=100) pol = IgnorePolicy(max_file_bytes=100)
large = tmp_path / "large.txt" large = tmp_path / "large.txt"

View file

@ -231,34 +231,3 @@ def test_jinjify_managed_files_rejects_templates_with_missing_defaults(
assert templated == set() assert templated == set()
assert vars_text == "" assert vars_text == ""
assert not (template_root / "etc" / "foo" / "pdk.yaml.j2").exists() assert not (template_root / "etc" / "foo" / "pdk.yaml.j2").exists()
def test_jinjify_artifact_rejects_unsafe_src_rel(monkeypatch, tmp_path: Path):
from enroll.jinjaturtle import jinjify_artifact
bundle = tmp_path / "bundle"
template_root = tmp_path / "templates"
outside = tmp_path / "outside.yaml"
outside.write_text("key: value\n", encoding="utf-8")
called = False
def fake_run_jinjaturtle(*_args, **_kwargs):
nonlocal called
called = True
return JinjifyResult(template_text="key: {{ key }}\n", vars_text="key: value\n")
monkeypatch.setattr(jinjaturtle_mod, "run_jinjaturtle", fake_run_jinjaturtle)
result = jinjify_artifact(
bundle,
"foo",
"../outside.yaml",
"/etc/foo.yaml",
template_root,
jt_exe="jinjaturtle",
jt_enabled=True,
)
assert result is None
assert called is False

View file

@ -814,47 +814,6 @@ def test_manifest_fqdn_implies_no_common_roles(tmp_path: Path):
assert not (out / "roles" / "net").exists() assert not (out / "roles" / "net").exists()
def test_manifest_fqdn_rejects_unsafe_path_like_name(tmp_path: Path):
bundle = tmp_path / "bundle"
out = tmp_path / "ansible"
escape = tmp_path / "escape"
state = _minimal_package_state([])
_write_state(bundle, state)
with pytest.raises(Exception, match="--fqdn"):
manifest.manifest(str(bundle), str(out), fqdn=str(escape / "node"))
assert not (escape / "node.yml").exists()
assert not (escape / "node" / "users.yml").exists()
def test_manifest_fqdn_rejects_newline_inventory_injection(tmp_path: Path):
bundle = tmp_path / "bundle"
out = tmp_path / "ansible"
state = _minimal_package_state([])
_write_state(bundle, state)
with pytest.raises(Exception, match="--fqdn"):
manifest.manifest(str(bundle), str(out), fqdn="host1\nmalicious: true")
def test_manifest_fqdn_existing_output_rejects_symlink_component(tmp_path: Path):
bundle = tmp_path / "bundle"
out = tmp_path / "ansible"
escape = tmp_path / "escape"
state = _minimal_package_state([])
_write_state(bundle, state)
(out / "inventory").mkdir(parents=True)
escape.mkdir()
(out / "inventory" / "host_vars").symlink_to(escape, target_is_directory=True)
with pytest.raises(Exception, match="symlink"):
manifest.manifest(str(bundle), str(out), fqdn="host1.example.test")
assert not (escape / "host1.example.test" / "users.yml").exists()
def test_manifest_site_mode_creates_host_inventory_and_raw_files(tmp_path: Path): def test_manifest_site_mode_creates_host_inventory_and_raw_files(tmp_path: Path):
"""In --fqdn mode, host-specific state goes into inventory/host_vars.""" """In --fqdn mode, host-specific state goes into inventory/host_vars."""