diff --git a/enroll/ansible.py b/enroll/ansible.py index 2eaec0a..e0fcd0c 100644 --- a/enroll/ansible.py +++ b/enroll/ansible.py @@ -18,7 +18,6 @@ from .manifest_safety import ( iter_safe_artifact_files, prepare_manifest_output_dir, ) -from .render_safety import ansible_unsafe_data from .role_names import avoid_reserved_role_name from .state import inventory_packages_from_state, roles_from_state from .yamlutil import yaml_dump_mapping, yaml_load_mapping @@ -689,7 +688,7 @@ def _write_hostvars(site_root: str, fqdn: str, role: str, data: Dict[str, Any]) except Exception: existing_map = {} - merged = _merge_mappings_overwrite(existing_map, ansible_unsafe_data(data)) + merged = _merge_mappings_overwrite(existing_map, data) out = "---\n" + yaml_dump_mapping(merged, sort_keys=True) with open(path, "w", encoding="utf-8") as f: @@ -700,7 +699,7 @@ def _write_role_defaults(role_dir: str, mapping: Dict[str, Any]) -> None: """Overwrite role defaults/main.yml with the provided mapping.""" defaults_path = os.path.join(role_dir, "defaults", "main.yml") os.makedirs(os.path.dirname(defaults_path), exist_ok=True) - out = "---\n" + yaml_dump_mapping(ansible_unsafe_data(mapping), sort_keys=True) + out = "---\n" + yaml_dump_mapping(mapping, sort_keys=True) with open(defaults_path, "w", encoding="utf-8") as f: f.write(out) diff --git a/enroll/puppet.py b/enroll/puppet.py index baf7596..e034cea 100644 --- a/enroll/puppet.py +++ b/enroll/puppet.py @@ -20,7 +20,6 @@ from .manifest_safety import ( prepare_manifest_output_dir, safe_artifact_file, ) -from .render_safety import puppet_hiera_safe_data from .state import inventory_packages_from_state, roles_from_state from .jinjaturtle import ( can_jinjify_path, @@ -1587,9 +1586,7 @@ def _render_hiera_yaml() -> str: def _write_yaml(path: Path, data: Dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( - yaml.safe_dump( - puppet_hiera_safe_data(data), sort_keys=True, explicit_start=True - ), + yaml.safe_dump(data, sort_keys=True, explicit_start=True), encoding="utf-8", ) diff --git a/enroll/render_safety.py b/enroll/render_safety.py deleted file mode 100644 index e8fc54a..0000000 --- a/enroll/render_safety.py +++ /dev/null @@ -1,232 +0,0 @@ -from __future__ import annotations - -import json -import re -from collections.abc import Mapping, Set as AbstractSet -from typing import Any - - -ANSIBLE_JINJA_STARTS = ("{{", "{%", "{#") - - -class AnsibleUnsafeText(str): - """String subclass dumped as Ansible's ``!unsafe`` YAML scalar. - - Ansible templating can recursively evaluate Jinja delimiters that arrive - through variables/defaults. Harvested data is not authored playbook code; - values containing Jinja starts must be tagged as unsafe data before they are - written to Ansible variable files. - """ - - -def is_ansible_template_like(value: str) -> bool: - """Return true if *value* contains a Jinja start delimiter.""" - - return any(marker in value for marker in ANSIBLE_JINJA_STARTS) - - -def ansible_unsafe_data(value: Any) -> Any: - """Recursively mark template-looking harvested strings as Ansible data. - - Keep ordinary strings untouched so generated output remains readable and so - existing tests/tools that use ``yaml.safe_load`` continue to work for normal - data. Mapping keys are also strings in Ansible data structures, so protect - keys as well as values. - """ - - if isinstance(value, AnsibleUnsafeText): - return value - if isinstance(value, str): - return AnsibleUnsafeText(value) if is_ansible_template_like(value) else value - if isinstance(value, Mapping): - return { - ansible_unsafe_data(str(key)): ansible_unsafe_data(inner) - for key, inner in value.items() - } - if isinstance(value, list): - return [ansible_unsafe_data(item) for item in value] - if isinstance(value, tuple): - return [ansible_unsafe_data(item) for item in value] - if isinstance(value, AbstractSet): - return sorted(ansible_unsafe_data(item) for item in value) - return value - - -def escape_puppet_hiera_interpolation(value: str) -> str: - """Preserve literal ``%{`` text in Puppet Hiera data sources. - - Hiera treats ``%{...}`` in data values as interpolation. Enroll's Hiera - data is generated from harvested values, not authored Hiera expressions, so - any literal interpolation opener is escaped with Hiera's documented - ``literal('%')`` helper. - """ - - return str(value).replace("%{", "%{literal('%')}{") - - -def puppet_hiera_safe_data(value: Any) -> Any: - """Recursively escape Hiera interpolation openers in harvested data.""" - - if isinstance(value, Mapping): - return { - escape_puppet_hiera_interpolation(str(key)): puppet_hiera_safe_data(inner) - for key, inner in value.items() - } - if isinstance(value, list): - return [puppet_hiera_safe_data(item) for item in value] - if isinstance(value, tuple): - return [puppet_hiera_safe_data(item) for item in value] - if isinstance(value, AbstractSet): - return sorted(puppet_hiera_safe_data(item) for item in value) - if isinstance(value, str): - return escape_puppet_hiera_interpolation(value) - return value - - -def _plain_json_data(value: Any) -> Any: - if isinstance(value, Mapping): - return {str(key): _plain_json_data(inner) for key, inner in value.items()} - if isinstance(value, list): - return [_plain_json_data(item) for item in value] - if isinstance(value, tuple): - return [_plain_json_data(item) for item in value] - if isinstance(value, AbstractSet): - return sorted(_plain_json_data(item) for item in value) - return value - - -def _escape_braces_inside_json_strings(text: str) -> str: - """Replace literal braces only while scanning JSON string tokens.""" - - out: list[str] = [] - in_string = False - escaped = False - for ch in text: - if not in_string: - out.append(ch) - if ch == '"': - in_string = True - continue - - if escaped: - out.append(ch) - escaped = False - elif ch == "\\": - out.append(ch) - escaped = True - elif ch == '"': - out.append(ch) - in_string = False - elif ch == "{": - out.append("\\u007b") - elif ch == "}": - out.append("\\u007d") - else: - out.append(ch) - return "".join(out) - - -def salt_sls_json_quote(value: Any) -> str: - """Return a double-quoted YAML/JSON scalar safe for Salt's Jinja pass. - - Salt state and pillar SLS files normally use the ``jinja|yaml`` renderer - pipeline. YAML/JSON quoting alone does not stop ``{{ ... }}``, ``{% ... %}`` - or ``{# ... #}`` inside harvested values from being evaluated before YAML is - parsed. JSON/YAML double-quoted scalars decode ``\u007b`` and ``\u007d`` - after Jinja has run, so encode braces inside string tokens as Unicode escapes. - """ - - dumped = json.dumps(str(value), ensure_ascii=False) - return _escape_braces_inside_json_strings(dumped) - - -_PLAIN_YAML_KEY_RE = re.compile(r"^[A-Za-z0-9_./:-]+$") - - -def _salt_yaml_key(value: Any) -> str: - text = str(value) - if text and _PLAIN_YAML_KEY_RE.match(text) and not text.startswith(("-", "?", ":")): - return text - return salt_sls_json_quote(text) - - -def _salt_yaml_scalar(value: Any) -> str: - if value is None: - return "null" - if isinstance(value, bool): - return "true" if value else "false" - if isinstance(value, int) and not isinstance(value, bool): - return str(value) - if isinstance(value, float): - return json.dumps(value, allow_nan=False) - return salt_sls_json_quote(value) - - -def _salt_yaml_lines( - value: Any, indent: int = 0, *, sort_keys: bool = True -) -> list[str]: - prefix = " " * indent - if isinstance(value, Mapping): - if not value: - return [prefix + "{}"] - keys = sorted(value, key=lambda item: str(item)) if sort_keys else list(value) - lines: list[str] = [] - for key in keys: - inner = value[key] - key_text = _salt_yaml_key(key) - if isinstance(inner, Mapping): - if not inner: - lines.append(f"{prefix}{key_text}: {{}}") - else: - lines.append(f"{prefix}{key_text}:") - lines.extend( - _salt_yaml_lines(inner, indent + 2, sort_keys=sort_keys) - ) - elif isinstance(inner, (list, tuple, set)): - seq = list(inner) if not isinstance(inner, set) else sorted(inner) - if not seq: - lines.append(f"{prefix}{key_text}: []") - else: - lines.append(f"{prefix}{key_text}:") - lines.extend(_salt_yaml_lines(seq, indent + 2, sort_keys=sort_keys)) - else: - lines.append(f"{prefix}{key_text}: {_salt_yaml_scalar(inner)}") - return lines - - if isinstance(value, (list, tuple, set)): - seq = list(value) if not isinstance(value, set) else sorted(value) - if not seq: - return [prefix + "[]"] - lines = [] - for item in seq: - if isinstance(item, Mapping): - if not item: - lines.append(prefix + "- {}") - else: - lines.append(prefix + "-") - lines.extend( - _salt_yaml_lines(item, indent + 2, sort_keys=sort_keys) - ) - elif isinstance(item, (list, tuple, set)): - lines.append(prefix + "-") - lines.extend(_salt_yaml_lines(item, indent + 2, sort_keys=sort_keys)) - else: - lines.append(f"{prefix}- {_salt_yaml_scalar(item)}") - return lines - - return [prefix + _salt_yaml_scalar(value)] - - -def salt_sls_yaml_dump( - value: Any, - *, - sort_keys: bool = True, - explicit_start: bool = False, -) -> str: - """Dump block YAML whose string braces cannot form Salt Jinja delimiters.""" - - lines = _salt_yaml_lines(_plain_json_data(value), sort_keys=sort_keys) - rendered = "\n".join(lines).rstrip() + "\n" - if explicit_start: - rendered = "---\n" + rendered - return rendered diff --git a/enroll/salt.py b/enroll/salt.py index 2a9fc69..27ec915 100644 --- a/enroll/salt.py +++ b/enroll/salt.py @@ -8,6 +8,7 @@ import shutil from pathlib import Path from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple +import yaml from .cm import ( CMModule, @@ -21,9 +22,8 @@ from .manifest_safety import ( prepare_manifest_output_dir, safe_artifact_file, ) -from .render_safety import salt_sls_json_quote, salt_sls_yaml_dump from .state import inventory_packages_from_state, roles_from_state -from .yamlutil import yaml_load_mapping_file +from .yamlutil import yaml_dump_mapping, yaml_load_mapping_file class SaltRole(CMModule): @@ -381,7 +381,7 @@ def _active_service_state_ids_by_unit( def _yaml_quote(value: Any) -> str: - return salt_sls_json_quote(value) + return json.dumps(str(value), ensure_ascii=False) def _yaml_bool(value: Any) -> str: @@ -870,7 +870,9 @@ def _collect_salt_roles( def _append_yaml_value(lines: List[str], key: str, value: Any, *, indent: int) -> None: prefix = " " * indent if isinstance(value, dict): - dumped = salt_sls_yaml_dump(_plain_salt_data(value), sort_keys=True).rstrip() + dumped = yaml.safe_dump( + _plain_salt_data(value), sort_keys=True, default_flow_style=False + ).rstrip() if not dumped: lines.append(f"{prefix}- {key}: {{}}") return @@ -1499,7 +1501,7 @@ def _render_pillar_role(srole: SaltRole) -> str: def _write_yaml(path: Path, data: Dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( - salt_sls_yaml_dump(data, sort_keys=True, explicit_start=True), + yaml_dump_mapping(data, sort_keys=True, explicit_start=True), encoding="utf-8", ) diff --git a/enroll/state.py b/enroll/state.py index 633a22f..8d469b6 100644 --- a/enroll/state.py +++ b/enroll/state.py @@ -2,25 +2,13 @@ from __future__ import annotations import json import os -import stat import tempfile from pathlib import Path -from typing import Any, Dict, Mapping, TextIO, Union - -from .fsutil import open_no_follow_path +from typing import Any, Dict, Mapping, Union BundlePath = Union[str, Path] State = Dict[str, Any] -# state.json should contain structured metadata, not harvested file content. Keep -# this generous so large package inventories still work while rejecting obvious -# accidental/malicious memory-exhaustion inputs. -MAX_STATE_JSON_BYTES = 16 * 1024 * 1024 - - -class StateSafetyError(RuntimeError): - """Raised when a harvest bundle's state.json is unsafe to parse.""" - def state_path(bundle_dir: BundlePath) -> Path: """Return the canonical state.json path for a harvest bundle.""" @@ -28,67 +16,10 @@ def state_path(bundle_dir: BundlePath) -> Path: return Path(bundle_dir) / "state.json" -def _check_state_stat(path: Path, st: os.stat_result, *, max_bytes: int) -> None: - if stat.S_ISLNK(st.st_mode): - raise StateSafetyError(f"state.json is a symlink; refusing to read: {path}") - if not stat.S_ISREG(st.st_mode): - raise StateSafetyError(f"state.json is not a regular file: {path}") - if st.st_nlink > 1: - raise StateSafetyError(f"state.json is hardlinked; refusing to read: {path}") - if st.st_size > max_bytes: - raise StateSafetyError( - f"state.json is too large to parse safely " - f"({st.st_size} bytes > {max_bytes} bytes): {path}" - ) - - -def open_state_file(bundle_dir: BundlePath, *, max_bytes: int | None = None) -> TextIO: - """Open state.json only after verifying it is safe to parse. - - Direct directory bundles are more mutable than SOPS/tar/remote bundles, so do - not follow a symlinked state.json and do not parse special files, hardlinks, or - unexpectedly huge inputs. The final open also uses no-follow semantics and the - inode is compared with the pre-open lstat result to catch swaps between the - check and open. - """ - - if max_bytes is None: - max_bytes = MAX_STATE_JSON_BYTES - - path = state_path(bundle_dir) - try: - pre = path.lstat() - except FileNotFoundError: - raise FileNotFoundError(f"missing state.json: {path}") - - _check_state_stat(path, pre, max_bytes=max_bytes) - - fd = -1 - try: - fd = open_no_follow_path(str(path), write=False) - opened = os.fstat(fd) - if (opened.st_dev, opened.st_ino) != (pre.st_dev, pre.st_ino): - raise StateSafetyError( - f"state.json changed while it was being opened; refusing to read: {path}" - ) - _check_state_stat(path, opened, max_bytes=max_bytes) - f = os.fdopen(fd, "r", encoding="utf-8") - fd = -1 - return f - except OSError as e: - raise StateSafetyError(f"unable to safely open state.json: {path}: {e}") from e - finally: - if fd >= 0: - try: - os.close(fd) - except OSError: - pass - - def load_state(bundle_dir: BundlePath) -> State: """Load state.json from a harvest bundle directory.""" - with open_state_file(bundle_dir) as f: + with open(state_path(bundle_dir), "r", encoding="utf-8") as f: return json.load(f) diff --git a/enroll/yamlutil.py b/enroll/yamlutil.py index b3bf10d..00e8496 100644 --- a/enroll/yamlutil.py +++ b/enroll/yamlutil.py @@ -5,21 +5,6 @@ from typing import Any, Dict, Mapping import yaml -from .render_safety import AnsibleUnsafeText - - -class IndentedSafeLoader(yaml.SafeLoader): # type: ignore[misc] - """PyYAML loader that understands Ansible's ``!unsafe`` tag.""" - - -def _construct_ansible_unsafe( - loader: yaml.Loader, node: yaml.Node -) -> AnsibleUnsafeText: - return AnsibleUnsafeText(loader.construct_scalar(node)) - - -IndentedSafeLoader.add_constructor("!unsafe", _construct_ansible_unsafe) - class IndentedSafeDumper(yaml.SafeDumper): # type: ignore[misc] """PyYAML dumper that indents sequences under mapping keys.""" @@ -32,17 +17,10 @@ class IndentedSafeDumper(yaml.SafeDumper): # type: ignore[misc] def yaml_load_mapping(text: str) -> Dict[str, Any]: - """Load YAML text and return a mapping, or an empty mapping on failure. - - Enroll may re-read Ansible host_vars that contain ``!unsafe`` scalars - written during the same manifest operation, so the loader accepts that tag - while remaining otherwise based on PyYAML's SafeLoader. - """ + """Load YAML text and return a mapping, or an empty mapping on failure.""" try: - obj = yaml.load( - text, Loader=IndentedSafeLoader - ) # nosec B506 - subclasses yaml.SafeLoader; only adds !unsafe scalar support. + obj = yaml.safe_load(text) except Exception: return {} return obj if isinstance(obj, dict) else {} @@ -56,15 +34,6 @@ def yaml_load_mapping_file(path: Path) -> Dict[str, Any]: return yaml_load_mapping(path.read_text(encoding="utf-8")) -def _represent_ansible_unsafe( - dumper: yaml.Dumper, data: AnsibleUnsafeText -) -> yaml.Node: - return dumper.represent_scalar("!unsafe", str(data)) - - -IndentedSafeDumper.add_representer(AnsibleUnsafeText, _represent_ansible_unsafe) - - def yaml_dump_mapping( obj: Mapping[str, Any], *, diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 27721ea..94148ad 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -2378,15 +2378,3 @@ def test_manifest_non_fqdn_refuses_existing_output(tmp_path: Path): with pytest.raises(RuntimeError, match="already exists"): manifest.manifest(str(bundle), str(out), no_common_roles=True) - - -def test_yaml_dump_mapping_emits_ansible_unsafe_tag_for_marked_values(): - from enroll.render_safety import ansible_unsafe_data - - data = ansible_unsafe_data({"value": "{{ lookup('pipe','id') }}"}) - dumped = yaml_helpers.yaml_dump_mapping(data) - - assert "value: !unsafe" in dumped - assert "{{ lookup(''pipe'',''id'') }}" in dumped - loaded = yaml_helpers.yaml_load_mapping(dumped) - assert loaded["value"] == "{{ lookup('pipe','id') }}" diff --git a/tests/test_manifest_ansible.py b/tests/test_manifest_ansible.py index 48a2721..88e4995 100644 --- a/tests/test_manifest_ansible.py +++ b/tests/test_manifest_ansible.py @@ -89,74 +89,3 @@ def test_ansible_role_normalises_package_snapshot(): assert role.files["/etc/curlrc"]["dest"] == "/etc/curlrc" assert role.services == {} assert role.origin_lines == ["package `curl` from role `curl`"] - - -from pathlib import Path - -from state_helpers import write_schema_state - -from enroll import manifest, yamlutil as yaml_helpers - - -def _ansible_jinja_payload_state(payload: str) -> dict: - return { - "schema_version": 3, - "host": {"hostname": "test", "os": "debian", "pkg_backend": "dpkg"}, - "inventory": {"packages": {}}, - "roles": { - "users": { - "role_name": "users", - "users": [ - { - "name": "alice", - "uid": 1000, - "gid": 1000, - "gecos": payload, - "home": "/home/alice", - "shell": "/bin/bash", - "primary_group": "alice", - "supplementary_groups": [], - } - ], - "managed_dirs": [], - "managed_files": [], - "managed_links": [], - "excluded": [], - "notes": [], - }, - "services": [], - "packages": [], - }, - } - - -def test_ansible_static_marks_harvested_jinja_values_unsafe(tmp_path: Path): - bundle = tmp_path / "bundle" - out = tmp_path / "out" - payload = "{{ lookup('pipe','touch /tmp/PWNED_BY_ENROLL_ANSIBLE') }}" - write_schema_state(bundle, _ansible_jinja_payload_state(payload)) - - manifest.manifest(str(bundle), str(out), target="ansible") - - defaults = out / "roles" / "users" / "defaults" / "main.yml" - text = defaults.read_text(encoding="utf-8") - assert "gecos: !unsafe" in text - assert "lookup(''pipe'',''touch /tmp/PWNED_BY_ENROLL_ANSIBLE'')" in text - loaded = yaml_helpers.yaml_load_mapping(text) - assert loaded["users_users"][0]["gecos"] == payload - - -def test_ansible_fqdn_marks_harvested_jinja_values_unsafe(tmp_path: Path): - bundle = tmp_path / "bundle" - out = tmp_path / "out" - payload = "{{ lookup('pipe','touch /tmp/PWNED_BY_ENROLL_ANSIBLE') }}" - write_schema_state(bundle, _ansible_jinja_payload_state(payload)) - - manifest.manifest(str(bundle), str(out), target="ansible", fqdn="host.example.test") - - hostvars = out / "inventory" / "host_vars" / "host.example.test" / "users.yml" - text = hostvars.read_text(encoding="utf-8") - assert "gecos: !unsafe" in text - assert "lookup(''pipe'',''touch /tmp/PWNED_BY_ENROLL_ANSIBLE'')" in text - loaded = yaml_helpers.yaml_load_mapping(text) - assert loaded["users_users"][0]["gecos"] == payload diff --git a/tests/test_manifest_puppet.py b/tests/test_manifest_puppet.py index 3aac35f..c54f6c7 100644 --- a/tests/test_manifest_puppet.py +++ b/tests/test_manifest_puppet.py @@ -1408,72 +1408,3 @@ def test_manifest_puppet_user_gecos_with_newline_is_single_line(tmp_path: Path): assert 'comment => "Real Name\\ntouch /tmp/pwned"' in init_pp # And there must be no line that is just the injected command. assert "\ntouch /tmp/pwned\n" not in init_pp - - -def _puppet_hiera_payload_state(payload: str) -> dict: - return { - "schema_version": 3, - "host": {"hostname": "test", "os": "debian", "pkg_backend": "dpkg"}, - "inventory": {"packages": {}}, - "roles": { - "users": { - "role_name": "users", - "users": [ - { - "name": "alice", - "uid": 1000, - "gid": 1000, - "gecos": payload, - "home": "/home/alice", - "shell": "/bin/bash", - "primary_group": "alice", - "supplementary_groups": [], - } - ], - "managed_dirs": [], - "managed_files": [], - "managed_links": [], - "excluded": [], - "notes": [], - }, - "services": [], - "packages": [], - }, - } - - -def test_manifest_puppet_static_quotes_template_like_harvested_values( - tmp_path: Path, -): - bundle = tmp_path / "bundle" - out = tmp_path / "puppet" - payload = "%{lookup('enroll::classes')}" - _write_state(bundle, _puppet_hiera_payload_state(payload)) - - manifest.manifest(str(bundle), str(out), target="puppet") - - init_pp = (out / "modules" / "users" / "manifests" / "init.pp").read_text( - encoding="utf-8" - ) - assert "comment => '%{lookup(\\'enroll::classes\\')}'" in init_pp - - -def test_manifest_puppet_hiera_escapes_harvested_interpolation_tokens( - tmp_path: Path, -): - bundle = tmp_path / "bundle" - out = tmp_path / "puppet" - payload = "%{lookup('enroll::classes')}" - _write_state(bundle, _puppet_hiera_payload_state(payload)) - - manifest.manifest(str(bundle), str(out), target="puppet", fqdn="node.example") - - node_yaml = out / "data" / "nodes" / "node.example.yaml" - text = node_yaml.read_text(encoding="utf-8") - assert payload not in text - assert "%{literal(''%'')}{lookup(''enroll::classes'')}" in text - data = yaml.safe_load(text) - assert ( - data["users::users"]["alice"]["comment"] - == "%{literal('%')}{lookup('enroll::classes')}" - ) diff --git a/tests/test_manifest_salt.py b/tests/test_manifest_salt.py index aa3e5ab..3c8aec4 100644 --- a/tests/test_manifest_salt.py +++ b/tests/test_manifest_salt.py @@ -964,82 +964,3 @@ def test_salt_names_are_sanitised_for_target_reserved_words() -> None: assert _salt_name("123") == "role_123" assert _salt_name("top") == "role_top" assert _salt_name("web-app") == "web_app" - - -def test_manifest_salt_static_escapes_harvested_jinja_delimiters(tmp_path: Path): - bundle = tmp_path / "bundle" - out = tmp_path / "salt" - state = _sample_state() - payload = "{{ salt['cmd.run']('touch /tmp/PWNED_BY_ENROLL_SALT') }}" - state["roles"]["users"]["users"][0]["gecos"] = payload - _write_sample_artifacts(bundle) - _write_state(bundle, state) - - manifest.manifest(str(bundle), str(out), target="salt") - - users_sls = (out / "states" / "roles" / "users" / "init.sls").read_text( - encoding="utf-8" - ) - assert payload not in users_sls - assert "\\u007b\\u007b salt['cmd.run']" in users_sls - - calls = [] - - class FakeCmd: - def run(self, command): - calls.append(command) - return "EXECUTED" - - from jinja2 import Template - - rendered = Template(users_sls).render(salt={"cmd.run": FakeCmd().run}) - rendered_data = yaml.safe_load(rendered) - assert calls == [] - user_state = next( - state - for state in rendered_data.values() - if isinstance(state, dict) and "user.present" in state - ) - attrs = user_state["user.present"] - fullname = next(item["fullname"] for item in attrs if "fullname" in item) - assert fullname == payload - - -def test_manifest_salt_fqdn_escapes_harvested_jinja_delimiters_in_pillar( - tmp_path: Path, -): - bundle = tmp_path / "bundle" - out = tmp_path / "salt" - state = _sample_state() - payload = "{{ salt['cmd.run']('touch /tmp/PWNED_BY_ENROLL_SALT') }}" - state["roles"]["users"]["users"][0]["gecos"] = payload - _write_sample_artifacts(bundle) - _write_state(bundle, state) - - manifest.manifest(str(bundle), str(out), target="salt", fqdn="node.example") - - pillar_top = yaml.safe_load( - (out / "pillar" / "top.sls").read_text(encoding="utf-8") - ) - node_sls = pillar_top["base"]["node.example"][0] - pillar_path = out / "pillar" / Path(*node_sls.split(".")) - text = pillar_path.with_suffix(".sls").read_text(encoding="utf-8") - assert payload not in text - assert "\\u007b\\u007b salt['cmd.run']" in text - - calls = [] - - class FakeCmd: - def run(self, command): - calls.append(command) - return "EXECUTED" - - from jinja2 import Template - - rendered = Template(text).render(salt={"cmd.run": FakeCmd().run}) - rendered_data = yaml.safe_load(rendered) - assert calls == [] - assert ( - rendered_data["enroll"]["roles"]["users"]["users"]["alice"]["fullname"] - == payload - ) diff --git a/tests/test_state_safety.py b/tests/test_state_safety.py deleted file mode 100644 index 5af9ff1..0000000 --- a/tests/test_state_safety.py +++ /dev/null @@ -1,49 +0,0 @@ -from __future__ import annotations - -import json -import os -from pathlib import Path - -import pytest - -from enroll.state import StateSafetyError, load_state, open_state_file - - -def test_load_state_reads_regular_state_json(tmp_path: Path): - (tmp_path / "state.json").write_text( - json.dumps({"host": {"hostname": "test-host"}}), encoding="utf-8" - ) - - assert load_state(tmp_path)["host"]["hostname"] == "test-host" - - -def test_load_state_rejects_state_json_symlink(tmp_path: Path): - target = tmp_path / "target.json" - target.write_text("{}", encoding="utf-8") - (tmp_path / "state.json").symlink_to(target) - - with pytest.raises(StateSafetyError, match="state.json is a symlink"): - load_state(tmp_path) - - -def test_load_state_rejects_non_regular_state_json(tmp_path: Path): - (tmp_path / "state.json").mkdir() - - with pytest.raises(StateSafetyError, match="state.json is not a regular file"): - load_state(tmp_path) - - -def test_load_state_rejects_hardlinked_state_json(tmp_path: Path): - state_file = tmp_path / "state.json" - state_file.write_text("{}", encoding="utf-8") - os.link(state_file, tmp_path / "state-copy.json") - - with pytest.raises(StateSafetyError, match="state.json is hardlinked"): - load_state(tmp_path) - - -def test_open_state_file_rejects_oversized_state_json(tmp_path: Path): - (tmp_path / "state.json").write_text("{}", encoding="utf-8") - - with pytest.raises(StateSafetyError, match="state.json is too large"): - open_state_file(tmp_path, max_bytes=1)