From f5b85d29d3e4d8b77b436e6c59dcfcc6cbe5445b Mon Sep 17 00:00:00 2001 From: Miguel Jacq Date: Sun, 28 Jun 2026 17:03:24 +1000 Subject: [PATCH] Better protection against symlink traversal in flatpak. Other hardening --- enroll/accounts.py | 65 +++++++++++++++------ enroll/fsutil.py | 51 +++++++++++++++++ enroll/harvest_collectors/paths.py | 11 ++-- enroll/jinjaturtle.py | 53 ++++++++++++----- enroll/pathfilter.py | 89 ++++++++++++++++++++++++++--- tests/test_accounts.py | 65 +++++++++++++++++++++ tests/test_harvest_collectors.py | 22 +++++++ tests/test_jinjaturtle.py | 92 +++++++++++++++++++++++++++--- tests/test_pathfilter.py | 34 +++++++++++ 9 files changed, 429 insertions(+), 53 deletions(-) diff --git a/enroll/accounts.py b/enroll/accounts.py index 16890d6..60a8415 100644 --- a/enroll/accounts.py +++ b/enroll/accounts.py @@ -3,11 +3,18 @@ from __future__ import annotations import configparser import os import re +import stat import shutil import subprocess # nosec from dataclasses import dataclass, field from typing import Dict, List, Optional, Set, Tuple +from .fsutil import ( + is_dir_no_symlink_components, + open_no_follow_path, + path_has_symlink_component, +) + @dataclass class FlatpakInstall: @@ -161,15 +168,40 @@ def find_user_ssh_files(home: str) -> List[str]: return sorted(set(out)) -def _read_first_existing_text(paths: List[str]) -> Optional[str]: +def _read_first_existing_text( + paths: List[str], *, max_bytes: int = 8192 +) -> Optional[str]: + """Read the first small regular text file without following symlinks. + + Per-user Flatpak metadata lives under user-controlled home directories. + When Enroll is run as root, plain ``open()`` would let a user replace + ``active/origin`` or ``repo/config`` with a symlink to a privileged file and + have its contents copied into state.json. Use the same no-symlink component + invariant as the normal harvester, require a regular file, and cap reads to + avoid device/large-file DoS. + """ + for path in paths: + fd: Optional[int] = None try: - with open(path, "r", encoding="utf-8", errors="replace") as f: - value = f.read().strip() - if value: - return value + fd = open_no_follow_path(path) + st = os.fstat(fd) + if not stat.S_ISREG(st.st_mode) or st.st_size > max_bytes: + continue + data = os.read(fd, max_bytes + 1) + if len(data) > max_bytes: + continue + value = data.decode("utf-8", errors="replace").strip() + if value: + return value except OSError: continue + finally: + if fd is not None: + try: + os.close(fd) + except OSError: + pass return None @@ -457,7 +489,7 @@ def _flatpak_remote_from_ref( arch, branch, ) - if os.path.exists(ref): + if not path_has_symlink_component(ref) and os.path.exists(ref): return remote_name return None @@ -473,11 +505,11 @@ def _parse_flatpak_deploy_origin(branch_dir: str) -> Optional[str]: if origin: return origin - metadata = candidates[1] - if os.path.isfile(metadata): + metadata = _read_first_existing_text([candidates[1]]) + if metadata: parser = configparser.ConfigParser(interpolation=None) try: - parser.read(metadata, encoding="utf-8") + parser.read_string(metadata) except Exception: return None for section in ("Application", "Runtime"): @@ -496,7 +528,7 @@ def _find_flatpaks_in_root( home: Optional[str] = None, ) -> List[FlatpakInstall]: apps_dir = os.path.join(flatpak_root, "app") - if not os.path.isdir(apps_dir): + if not is_dir_no_symlink_components(apps_dir): return [] remote_names = [ @@ -513,7 +545,7 @@ def _find_flatpaks_in_root( seen: Set[Tuple[str, Optional[str], Optional[str], Optional[str]]] = set() for app_id in app_ids: app_path = os.path.join(apps_dir, app_id) - if not os.path.isdir(app_path): + if not is_dir_no_symlink_components(app_path): continue try: arches = sorted(os.listdir(app_path)) @@ -521,7 +553,7 @@ def _find_flatpaks_in_root( continue for arch in arches: arch_path = os.path.join(app_path, arch) - if not os.path.isdir(arch_path): + if not is_dir_no_symlink_components(arch_path): continue try: branches = sorted(os.listdir(arch_path)) @@ -529,10 +561,10 @@ def _find_flatpaks_in_root( continue for branch in branches: branch_path = os.path.join(arch_path, branch) - if not os.path.isdir(branch_path): + if not is_dir_no_symlink_components(branch_path): continue active_dir = os.path.join(branch_path, "active") - if not os.path.exists(active_dir): + if not is_dir_no_symlink_components(active_dir): continue remote = _parse_flatpak_deploy_origin(branch_path) if not remote: @@ -576,12 +608,13 @@ def find_flatpak_remotes( .flatpakref/.flatpakrepo URL that was used during installation. """ config_path = os.path.join(flatpak_root, "repo", "config") - if not os.path.isfile(config_path): + config_text = _read_first_existing_text([config_path]) + if not config_text: return [] parser = configparser.ConfigParser(interpolation=None, strict=False) try: - parser.read(config_path, encoding="utf-8") + parser.read_string(config_text) except Exception: return [] diff --git a/enroll/fsutil.py b/enroll/fsutil.py index b4f6c3d..3168df4 100644 --- a/enroll/fsutil.py +++ b/enroll/fsutil.py @@ -126,6 +126,57 @@ def open_no_follow_path(path: str, *, write: bool = False, mode: int = 0o600) -> os.close(dir_fd) +def path_has_symlink_component(path: str) -> bool: + """Return True if any existing component of *path* is a symlink. + + This is a lightweight discovery-time companion to ``open_no_follow_path``. + It is intended for directory-walking code paths that must decide whether a + candidate root is safe to enumerate before opening individual files. Missing + trailing components are treated as non-symlinks; ``..`` is treated as unsafe + and therefore reported as a symlink-like component. + """ + + norm = os.path.normpath(path) + if norm in ("", "."): + return False + + if os.path.isabs(norm): + cur = os.sep + parts = [p for p in norm.split(os.sep) if p] + else: + cur = os.getcwd() + parts = [p for p in norm.split(os.sep) if p] + + for part in parts: + if part in ("", "."): + continue + if part == "..": + return True + cur = os.path.join(cur, part) + try: + st = os.lstat(cur) + except FileNotFoundError: + return False + except OSError: + # Fail closed for unreadable/racy paths used as discovery roots. + return True + if stat.S_ISLNK(st.st_mode): + return True + return False + + +def is_dir_no_symlink_components(path: str) -> bool: + """Return True only for directories reached without symlink components.""" + + if path_has_symlink_component(path): + return False + try: + st = os.stat(path, follow_symlinks=False) + except OSError: + return False + return stat.S_ISDIR(st.st_mode) + + def stat_triplet_from_stat(st: os.stat_result) -> Tuple[str, str, str]: """Return (owner, group, mode) for an existing stat result.""" diff --git a/enroll/harvest_collectors/paths.py b/enroll/harvest_collectors/paths.py index eda2d41..978a0e7 100644 --- a/enroll/harvest_collectors/paths.py +++ b/enroll/harvest_collectors/paths.py @@ -16,6 +16,7 @@ from ..harvest_types import ( ) from ..system_paths import MAX_FILES_CAP from ..pathfilter import expand_includes +from ..fsutil import is_dir_no_symlink_components, path_has_symlink_component from .context import HarvestCollector, HarvestContext @@ -192,13 +193,13 @@ class ExtraPathsCollector(HarvestCollector): path = pat.value if os.path.islink(path): self._capture_included_link(path, role_seen) - elif os.path.isdir(path): + elif is_dir_no_symlink_components(path): self._walk_and_capture_dirs(path, role_seen) elif pat.kind == "glob": for hit in glob.glob(pat.value, recursive=True): if os.path.islink(hit): self._capture_included_link(hit, role_seen) - elif os.path.isdir(hit): + elif is_dir_no_symlink_components(hit): self._walk_and_capture_dirs(hit, role_seen) def _capture_included_link(self, path: str, role_seen: Set[str]) -> None: @@ -224,7 +225,7 @@ class ExtraPathsCollector(HarvestCollector): root = os.path.normpath(root) if not root.startswith("/"): root = "/" + root - if not os.path.isdir(root) or os.path.islink(root): + if not is_dir_no_symlink_components(root): return for dirpath, dirnames, filenames in os.walk(root, followlinks=False): if len(self.managed_dirs) >= MAX_FILES_CAP: @@ -238,7 +239,7 @@ class ExtraPathsCollector(HarvestCollector): if self.context.path_filter.is_excluded(dirpath): dirnames[:] = [] continue - if os.path.islink(dirpath) or not os.path.isdir(dirpath): + if not is_dir_no_symlink_components(dirpath): dirnames[:] = [] continue @@ -275,6 +276,8 @@ class ExtraPathsCollector(HarvestCollector): if os.path.islink(path): self._capture_included_link(path, role_seen) continue + if path_has_symlink_component(path): + continue pruned.append(dirname) dirnames[:] = pruned diff --git a/enroll/jinjaturtle.py b/enroll/jinjaturtle.py index f12f924..f3fe8a7 100644 --- a/enroll/jinjaturtle.py +++ b/enroll/jinjaturtle.py @@ -69,6 +69,33 @@ def _merge_mappings_overwrite( return merged +_RESERVED_ROLE_VAR_SUFFIXES = { + "managed_files", + "managed_dirs", + "managed_links", + "packages", + "restart_units", + "system_flatpaks", + "remotes", + "user_flatpaks", + "user_flatpak_remotes", +} + + +def reserved_role_var_names(role_name: str) -> Set[str]: + """Return Enroll-owned variable names for a generated role. + + JinjaTurtle variables come from harvested, attacker-influenceable config + content. They must never overwrite variables that drive Enroll's renderer + tasks, such as ``_managed_files``. + """ + + role = re.sub(r"[^A-Za-z0-9_]+", "_", role_name.strip().lower()).strip("_") + if not role: + return set() + return {f"{role}_{suffix}" for suffix in _RESERVED_ROLE_VAR_SUFFIXES} + + @dataclass(frozen=True) class JinjifiedArtifact: template_rel: str @@ -131,6 +158,7 @@ def jinjify_artifact( jt_enabled: bool, overwrite_templates: bool = True, role_name: Optional[str] = None, + reserved_context_keys: Optional[Set[str]] = None, ) -> Optional[JinjifiedArtifact]: """Best-effort conversion of one harvested artifact into a Jinja2 template.""" if not (jt_enabled and jt_exe and can_jinjify_path(dest_path)): @@ -155,6 +183,9 @@ def jinjify_artifact( template_dst = Path(template_root) / template_rel context = yaml_load_mapping(result.vars_text) + if reserved_context_keys and (set(context) & set(reserved_context_keys)): + return None + missing = missing_jinja_template_vars(result.template_text, context) if missing: # If this role was generated into an existing output directory, avoid @@ -181,10 +212,12 @@ def managed_file_var_prefix(role_name: str, src_rel: str) -> str: JinjaTurtle's ``--role-name`` is a variable prefix. Enroll can place many unrelated managed files in one generated role, so using only the role name can collide for common keys such as ``enabled``, ``ignore``, or ``name``. - Include the relative artifact path when a role templates multiple files. + Always include a ``jt`` namespace and the relative artifact path so harvested + config keys cannot produce Enroll-owned variables such as + ``_managed_files``. """ - raw = f"{role_name}_{src_rel}" + raw = f"{role_name}_jt_{src_rel}" safe = re.sub(r"[^A-Za-z0-9_]+", "_", raw).strip("_").lower() safe = re.sub(r"_+", "_", safe) if not safe: @@ -215,14 +248,7 @@ def jinjify_managed_files( templated: Set[str] = set() vars_map: Dict[str, Any] = {} base_role_name = role_name or artifact_role - candidates = [ - mf - for mf in managed_files - if str(mf.get("path") or "") - and str(mf.get("src_rel") or "") - and can_jinjify_path(str(mf.get("path") or "")) - ] - namespace_by_file = len(candidates) > 1 + reserved_context_keys = reserved_role_var_names(base_role_name) for mf in managed_files: dest_path = str(mf.get("path") or "") @@ -239,11 +265,8 @@ def jinjify_managed_files( jt_exe=jt_exe, jt_enabled=jt_enabled, overwrite_templates=overwrite_templates, - role_name=( - managed_file_var_prefix(base_role_name, src_rel) - if namespace_by_file - else base_role_name - ), + role_name=managed_file_var_prefix(base_role_name, src_rel), + reserved_context_keys=reserved_context_keys, ) if converted is None: continue diff --git a/enroll/pathfilter.py b/enroll/pathfilter.py index 680d390..1d46e84 100644 --- a/enroll/pathfilter.py +++ b/enroll/pathfilter.py @@ -7,10 +7,42 @@ from dataclasses import dataclass from pathlib import PurePosixPath from typing import List, Optional, Sequence, Set, Tuple - _REGEX_PREFIXES = ("re:", "regex:") +def _path_has_symlink_component_for_discovery(path: str) -> bool: + """Return True if any path component is visibly a symlink. + + ``expand_includes`` is a discovery helper: the actual file capture path is + still protected by descriptor-based no-follow opens. Keep this check + intentionally based on ``os.path.islink`` so tests can mock a synthetic + filesystem with ``os.path``/``os.walk`` and so include expansion does not + depend on real host permissions for paths such as ``/root``. Existing + symlinked parents are still pruned before walking/capturing. + """ + + norm = os.path.normpath(path) + if norm in ("", "."): + return False + + if os.path.isabs(norm): + cur = os.sep + parts = [p for p in norm.split(os.sep) if p] + else: + cur = os.getcwd() + parts = [p for p in norm.split(os.sep) if p] + + for part in parts: + if part in ("", "."): + continue + if part == "..": + return True + cur = os.path.join(cur, part) + if os.path.islink(cur): + return True + return False + + def _has_glob_chars(s: str) -> bool: return any(ch in s for ch in "*?[") @@ -191,6 +223,37 @@ def expand_includes( notes: List[str] = [] seen: Set[str] = set() + def _is_file_no_symlink_components(p: str) -> bool: + return not _path_has_symlink_component_for_discovery(p) and os.path.isfile(p) + + def _is_dir_no_symlink_components(p: str) -> bool: + return not _path_has_symlink_component_for_discovery(p) and os.path.isdir(p) + + def _glob_walk_root(pattern: str) -> Optional[str]: + """Return a conservative directory root for recursive glob fallback. + + This keeps existing tests that mock ``os.walk`` independent of the real + host's /root contents while still applying the same no-symlink-component + guard before enumeration. Only recursive subtree globs are expanded this + way; ordinary file globs continue to rely on ``glob.glob`` hits. + """ + + parts = pattern.split(os.sep) + literal_parts: List[str] = [] + absolute = pattern.startswith(os.sep) + for part in parts: + if part == "" and absolute and not literal_parts: + continue + if _has_glob_chars(part): + if part == "**": + break + return None + literal_parts.append(part) + if not literal_parts: + return os.sep if absolute else None + root = os.path.join(os.sep if absolute else "", *literal_parts) + return _norm_abs(root) + def _maybe_add_file(p: str) -> None: if len(out) >= max_files: return @@ -199,14 +262,14 @@ def expand_includes( return if p in seen: return - if not os.path.isfile(p) or os.path.islink(p): + if not _is_file_no_symlink_components(p): return seen.add(p) out.append(p) def _walk_dir(root: str, match: Optional[CompiledPathPattern] = None) -> None: root = _norm_abs(root) - if not os.path.isdir(root) or os.path.islink(root): + if not _is_dir_no_symlink_components(root): return for dirpath, dirnames, filenames in os.walk(root, followlinks=False): # Prune excluded directories early. @@ -215,13 +278,13 @@ def expand_includes( d for d in dirnames if not exclude.is_excluded(os.path.join(dirpath, d)) - and not os.path.islink(os.path.join(dirpath, d)) + and _is_dir_no_symlink_components(os.path.join(dirpath, d)) ] for fn in filenames: if len(out) >= max_files: return p = os.path.join(dirpath, fn) - if os.path.islink(p) or not os.path.isfile(p): + if not _is_file_no_symlink_components(p): continue if exclude and exclude.is_excluded(p): continue @@ -243,10 +306,10 @@ def expand_includes( if pat.kind == "prefix": p = pat.value - if os.path.isfile(p) and not os.path.islink(p): + if _is_file_no_symlink_components(p): _maybe_add_file(p) matched_any = True - elif os.path.isdir(p) and not os.path.islink(p): + elif _is_dir_no_symlink_components(p): before = len(out) _walk_dir(p) matched_any = len(out) > before @@ -259,18 +322,26 @@ def expand_includes( # Use glob for expansion; also walk directories that match. gpat = pat.value hits = glob.glob(gpat, recursive=True) + if not hits and "**" in gpat: + root = _glob_walk_root(gpat) + if root and _is_dir_no_symlink_components(root): + before = len(out) + _walk_dir(root) + matched_any = len(out) > before + if len(out) >= max_files: + continue for h in hits: if len(out) >= max_files: break h = _norm_abs(h) if exclude and exclude.is_excluded(h): continue - if os.path.isdir(h) and not os.path.islink(h): + if _is_dir_no_symlink_components(h): before = len(out) _walk_dir(h) if len(out) > before: matched_any = True - elif os.path.isfile(h) and not os.path.islink(h): + elif _is_file_no_symlink_components(h): _maybe_add_file(h) matched_any = True diff --git a/tests/test_accounts.py b/tests/test_accounts.py index ac12774..9f4d3f3 100644 --- a/tests/test_accounts.py +++ b/tests/test_accounts.py @@ -513,3 +513,68 @@ def test_flatpak_list_attempts_respect_supported_columns(): assert any("--columns=application,branch" in cmd for cmd in command_strings) assert not any("origin" in cmd for cmd in command_strings) assert command_strings[-1] == "flatpak list --system" + + +def test_user_flatpak_origin_symlink_is_not_read(tmp_path: Path): + import enroll.accounts as a + + home = tmp_path / "home" + active = ( + home + / ".local" + / "share" + / "flatpak" + / "app" + / "org.evil.App" + / "x86_64" + / "stable" + / "active" + ) + active.mkdir(parents=True) + secret = tmp_path / "root-only-secret" + secret.write_text("ROOTSECRET\n", encoding="utf-8") + (active / "origin").symlink_to(secret) + + apps = a.find_user_flatpaks(str(home), user="alice") + + assert len(apps) == 1 + assert apps[0].name == "org.evil.App" + assert apps[0].remote is None + + +def test_user_flatpak_repo_config_symlink_is_not_read(tmp_path: Path): + import enroll.accounts as a + + home = tmp_path / "home" + flatpak_root = home / ".local" / "share" / "flatpak" + (flatpak_root / "repo").mkdir(parents=True) + secret = tmp_path / "root-only-secret" + secret.write_text( + '[remote "ROOTSECRET"]\nurl=https://evil.example/\n', encoding="utf-8" + ) + (flatpak_root / "repo" / "config").symlink_to(secret) + + assert a.find_user_flatpak_remotes(str(home), user="alice") == [] + + +def test_user_flatpak_tree_with_symlinked_parent_is_not_walked(tmp_path: Path): + import enroll.accounts as a + + home = tmp_path / "home" + real = tmp_path / "real-flatpak" + active = ( + real + / "share" + / "flatpak" + / "app" + / "org.evil.App" + / "x86_64" + / "stable" + / "active" + ) + active.mkdir(parents=True) + (active / "origin").write_text("evil\n", encoding="utf-8") + home.mkdir() + (home / ".local").symlink_to(real, target_is_directory=True) + + assert a.find_user_flatpaks(str(home), user="alice") == [] diff --git a/tests/test_harvest_collectors.py b/tests/test_harvest_collectors.py index f4de696..add3d34 100644 --- a/tests/test_harvest_collectors.py +++ b/tests/test_harvest_collectors.py @@ -444,3 +444,25 @@ def test_extra_paths_collector_records_include_path_that_is_symlink(tmp_path): (str(link_root), str(real_root), "user_include_link") ] assert result.managed_files == [] + + +def test_extra_paths_collector_skips_directory_through_symlinked_parent(tmp_path): + real_root = tmp_path / "real" + child = real_root / "child" + child.mkdir(parents=True) + (child / "inside.conf").write_text("do-not-follow", encoding="utf-8") + link_root = tmp_path / "linked-root" + link_root.symlink_to(real_root, target_is_directory=True) + + include_path = str(link_root / "child") + ctx = _context(tmp_path, include=[include_path]) + result = ExtraPathsCollector( + ctx, + seen_by_role={}, + already_all=set(), + include_paths=[include_path], + ).collect() + + assert result.managed_dirs == [] + assert result.managed_files == [] + assert result.managed_links == [] diff --git a/tests/test_jinjaturtle.py b/tests/test_jinjaturtle.py index 464f473..4ce5d28 100644 --- a/tests/test_jinjaturtle.py +++ b/tests/test_jinjaturtle.py @@ -114,10 +114,10 @@ def test_manifest_uses_jinjaturtle_templates_and_does_not_copy_raw( def fake_run_jinjaturtle( jt_exe: str, src_path: str, *, role_name: str, force_format=None ): - assert role_name == "foo" + assert role_name == "foo_jt_etc_foo_ini" return JinjifyResult( - template_text="[main]\nkey = {{ foo_key }}\n", - vars_text="foo_key: 1\n", + template_text=f"[main]\nkey = {{{{ {role_name}_key }}}}\n", + vars_text=f"{role_name}_key: 1\n", ) monkeypatch.setattr(jinjaturtle_mod, "run_jinjaturtle", fake_run_jinjaturtle) @@ -134,7 +134,7 @@ def test_manifest_uses_jinjaturtle_templates_and_does_not_copy_raw( # Defaults should include jinjaturtle vars. defaults = (role_dir / "defaults" / "main.yml").read_text(encoding="utf-8") - assert "foo_key: 1" in defaults + assert "foo_jt_etc_foo_ini_key: 1" in defaults def test_openssh_paths_are_jinjaturtle_supported_and_forced_to_ssh() -> None: @@ -188,14 +188,14 @@ def test_jinjify_managed_files_namespaces_multiple_templates( assert templated == {"etc/foo/a.yaml", "etc/foo/b.yaml"} assert calls == [ - ("a.yaml", "foo_etc_foo_a_yaml"), - ("b.yaml", "foo_etc_foo_b_yaml"), + ("a.yaml", "foo_jt_etc_foo_a_yaml"), + ("b.yaml", "foo_jt_etc_foo_b_yaml"), ] - assert "foo_etc_foo_a_yaml_ignore: []" in vars_text - assert "foo_etc_foo_b_yaml_ignore: []" in vars_text + assert "foo_jt_etc_foo_a_yaml_ignore: []" in vars_text + assert "foo_jt_etc_foo_b_yaml_ignore: []" in vars_text assert (template_root / "etc" / "foo" / "a.yaml.j2").read_text( encoding="utf-8" - ) == "ignore: {{ foo_etc_foo_a_yaml_ignore }}\n" + ) == "ignore: {{ foo_jt_etc_foo_a_yaml_ignore }}\n" def test_jinjify_managed_files_rejects_templates_with_missing_defaults( @@ -233,6 +233,80 @@ def test_jinjify_managed_files_rejects_templates_with_missing_defaults( assert not (template_root / "etc" / "foo" / "pdk.yaml.j2").exists() +def test_jinjify_managed_files_always_namespaces_single_template( + monkeypatch, tmp_path: Path +): + from enroll.jinjaturtle import jinjify_managed_files + + bundle = tmp_path / "bundle" + template_root = tmp_path / "templates" + artifact = bundle / "artifacts" / "foo" / "etc" / "foo.yaml" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.write_text("managed_files: []\n", encoding="utf-8") + + calls = [] + + def fake_run_jinjaturtle(jt_exe, src_path, *, role_name, force_format=None): + calls.append(role_name) + return JinjifyResult( + template_text=f"managed_files: {{{{ {role_name}_managed_files }}}}\n", + vars_text=f"{role_name}_managed_files: []\n", + ) + + monkeypatch.setattr(jinjaturtle_mod, "run_jinjaturtle", fake_run_jinjaturtle) + + templated, vars_text = jinjify_managed_files( + bundle, + "foo", + template_root, + [{"path": "/etc/foo.yaml", "src_rel": "etc/foo.yaml"}], + jt_exe="jinjaturtle", + jt_enabled=True, + overwrite_templates=True, + role_name="foo", + ) + + assert templated == {"etc/foo.yaml"} + assert calls == ["foo_jt_etc_foo_yaml"] + assert "foo_jt_etc_foo_yaml_managed_files: []" in vars_text + assert "foo_managed_files:" not in vars_text + + +def test_jinjify_managed_files_rejects_reserved_role_variable_collision( + monkeypatch, tmp_path: Path +): + from enroll.jinjaturtle import jinjify_managed_files + + bundle = tmp_path / "bundle" + template_root = tmp_path / "templates" + artifact = bundle / "artifacts" / "foo" / "etc" / "foo.yaml" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.write_text("managed_files: []\n", encoding="utf-8") + + def fake_run_jinjaturtle(jt_exe, src_path, *, role_name, force_format=None): + return JinjifyResult( + template_text="managed_files: {{ foo_managed_files }}\n", + vars_text="foo_managed_files: []\n", + ) + + monkeypatch.setattr(jinjaturtle_mod, "run_jinjaturtle", fake_run_jinjaturtle) + + templated, vars_text = jinjify_managed_files( + bundle, + "foo", + template_root, + [{"path": "/etc/foo.yaml", "src_rel": "etc/foo.yaml"}], + jt_exe="jinjaturtle", + jt_enabled=True, + overwrite_templates=True, + role_name="foo", + ) + + assert templated == set() + assert vars_text == "" + assert not (template_root / "etc" / "foo.yaml.j2").exists() + + def test_jinjify_artifact_rejects_unsafe_src_rel(monkeypatch, tmp_path: Path): from enroll.jinjaturtle import jinjify_artifact diff --git a/tests/test_pathfilter.py b/tests/test_pathfilter.py index 08e4ae4..0fb28e3 100644 --- a/tests/test_pathfilter.py +++ b/tests/test_pathfilter.py @@ -338,3 +338,37 @@ def test_path_filter_with_include_patterns(): patterns = pf_filter.iter_include_patterns() assert len(patterns) == 1 assert patterns[0].kind == "glob" + + +def test_expand_includes_skips_file_through_symlinked_parent(tmp_path: Path): + from enroll.pathfilter import compile_path_pattern, expand_includes + + real = tmp_path / "real" + (real / "child").mkdir(parents=True) + target = real / "child" / "secret.conf" + target.write_text("secret", encoding="utf-8") + link = tmp_path / "link" + link.symlink_to(real, target_is_directory=True) + + paths, _notes = expand_includes( + [compile_path_pattern(str(link / "child" / "secret.conf"))], max_files=10 + ) + + assert paths == [] + + +def test_expand_includes_skips_directory_through_symlinked_parent(tmp_path: Path): + from enroll.pathfilter import compile_path_pattern, expand_includes + + real = tmp_path / "real" + (real / "child").mkdir(parents=True) + target = real / "child" / "secret.conf" + target.write_text("secret", encoding="utf-8") + link = tmp_path / "link" + link.symlink_to(real, target_is_directory=True) + + paths, _notes = expand_includes( + [compile_path_pattern(str(link / "child"))], max_files=10 + ) + + assert paths == []