Update tests

CHANGELOG.md

@@ -2,6 +2,7 @@
 
  * BREAKING CHANGE: Group all package and systemd-unit roles into Debian Section/RPM Group roles by default, including managed config files and unit state. This mode is not used if `--fqdn` or `--no-common-roles` is set, in which case, the traditional behaviour of preserving one role per package/unit is used instead.
  * BREAKING CHANGE: Only capture user-specific .bashrc style files when using `--dangerous` mode, in case they contain sensitive env vars.
+ * BREAKING CHANGE: Don't allow reading `.enroll.ini` in the CWD. Use only the ENROLL_CONFIG env var, an explicit `--config` path or else the XDG default location (or `~/.config/enroll/enroll.ini` if `XDG_CONFIG_HOME` is not set).
  * Detect active sysctl parameters and write them to a `/etc/sysctl.d/99-enroll.conf` file
  * Use `no_log` on systemd unit interrogations to suppress potential sensitive output when applying Ansible
  * Support manifesting Puppet code, as well as Ansible!
@@ -10,6 +11,8 @@
  * A lot of under-the-bonnet refactoring to make it easier to extend to cover other config managers (that don't suck) in future.
  * Support for detecting Docker and Podman images and enforcing their presence (by SHA256 hash).
  * 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)
+ * Perform harvest validation before trying to manifest from it.
 
 # 0.6.0
 

README.md

@@ -656,8 +656,8 @@ Enroll supports reading an ini-style file of all the arguments for each subcomma
 ### Location of the config file
 
 The path the config file can be specified with `-c` or `--config` on the command-line. Otherwise,
-Enroll will look for `./enroll.ini`, `./.enroll.ini` (in the current working directory),
+Enroll will look for the `ENROLL_CONFIG` environment variable, `$XDG_CONFIG_HOME/enroll/enroll.ini`,
-`~/.config/enroll/enroll.ini` (or `$XDG_CONFIG_HOME/enroll/enroll.ini`).
+or `~/.config/enroll/enroll.ini`.
 
 You may also pass `--no-config` if you deliberately want to ignore the config file even if it existed.
 

enroll/ansible.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import os
 import re
-import shutil
 import stat
 import tempfile
 from dataclasses import dataclass
@@ -14,6 +13,11 @@ from .jinjaturtle import (
     jinjify_managed_files as _jinjify_managed_files,
     resolve_jinjaturtle_mode,
 )
+from .manifest_safety import (
+    copy_safe_artifact_file,
+    iter_safe_artifact_files,
+    prepare_manifest_output_dir,
+)
 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
@@ -408,7 +412,8 @@ def _prepare_ansible_context(
     site_mode = fqdn is not None and fqdn != ""
     jt_exe, jt_enabled = resolve_jinjaturtle_mode(jinjaturtle)
 
-    os.makedirs(out_dir, exist_ok=True)
+    out = prepare_manifest_output_dir(out_dir, allow_existing=site_mode)
+    out_dir = str(out)
     roles_root = os.path.join(out_dir, "roles")
     os.makedirs(roles_root, exist_ok=True)
 
@@ -445,7 +450,7 @@ def _copy2_replace(src: str, dst: str) -> None:
     fd, tmp = tempfile.mkstemp(prefix=".enroll-tmp-", dir=dst_dir)
     os.close(fd)
     try:
-        shutil.copy2(src, tmp)
+        copy_safe_artifact_file(src, tmp)
 
         # Ensure the working tree stays mergeable: make the file user-writable.
         st = os.stat(tmp, follow_symlinks=False)
@@ -475,29 +480,23 @@ def _copy_artifacts(
     In --fqdn site mode, this is usually:
       inventory/host_vars/<fqdn>/<role>/.files
     """
-    artifacts_dir = os.path.join(bundle_dir, "artifacts", role)
+    for src, rel in iter_safe_artifact_files(bundle_dir, role):
-    if not os.path.isdir(artifacts_dir):
+        dst = os.path.join(dst_files_dir, rel)
-        return
-    for root, _, files in os.walk(artifacts_dir):
-        for fn in files:
-            src = os.path.join(root, fn)
-            rel = os.path.relpath(src, artifacts_dir)
-            dst = os.path.join(dst_files_dir, rel)
 
-            # If a file was successfully templatised by JinjaTurtle, do NOT
+        # If a file was successfully templatised by JinjaTurtle, do NOT
-            # also materialise the raw copy in the destination files dir.
+        # also materialise the raw copy in the destination files dir.
-            if exclude_rels and rel in exclude_rels:
+        if exclude_rels and rel in exclude_rels:
-                try:
+            try:
-                    if os.path.isfile(dst):
+                if os.path.isfile(dst):
-                        os.remove(dst)
+                    os.remove(dst)
-                except Exception:
+            except Exception:
-                    pass  # nosec
+                pass  # nosec
-                continue
+            continue
 
-            if preserve_existing and os.path.exists(dst):
+        if preserve_existing and os.path.exists(dst):
-                continue
+            continue
-            os.makedirs(os.path.dirname(dst), exist_ok=True)
+        os.makedirs(os.path.dirname(dst), exist_ok=True)
-            _copy2_replace(src, dst)
+        _copy2_replace(str(src), dst)
 
 
 def _write_role_scaffold(role_dir: str) -> None:

enroll/cli.py

@@ -39,8 +39,10 @@ def _discover_config_path(argv: list[str]) -> Optional[Path]:
       1) --no-config disables loading.
       2) --config PATH (or -c PATH)
       3) $ENROLL_CONFIG
-      4) ./enroll.ini, ./.enroll.ini
+      4) $XDG_CONFIG_HOME/enroll/enroll.ini (or ~/.config/enroll/enroll.ini)
-      5) $XDG_CONFIG_HOME/enroll/enroll.ini (or ~/.config/enroll/enroll.ini)
+
+    Current-directory config files are deliberately not auto-loaded; use
+    --config ./enroll.ini if that behaviour is desired.
 
     The config file is optional; if no file is found, returns None.
     """
@@ -66,12 +68,6 @@ def _discover_config_path(argv: list[str]) -> Optional[Path]:
     if envp:
         return Path(envp).expanduser()
 
-    cwd = Path.cwd()
-    for name in ("enroll.ini", ".enroll.ini"):
-        cp = cwd / name
-        if cp.exists() and cp.is_file():
-            return cp
-
     xdg = os.environ.get("XDG_CONFIG_HOME")
     if xdg:
         base = Path(xdg).expanduser()

enroll/diff.py

@@ -923,14 +923,17 @@ def enforce_old_harvest(
             except OSError:
                 pass
 
-            # 1) Generate a manifest in a temp directory.
+            # 1) Generate a manifest in a temp directory.  The renderer now
-            manifest(str(old_b.dir), str(td_path), target=target)
+            # refuses to write into an existing destination, so use a fresh
+            # child path under the secure temporary directory.
+            manifest_dir = td_path / "manifest"
+            manifest(str(old_b.dir), str(manifest_dir), target=target)
 
             # 2) Apply it locally.
             cmd, env = _enforcement_command(
                 target,
                 tool_exe,
-                td_path,
+                manifest_dir,
                 tags=tags,
             )
 
@@ -1454,8 +1457,14 @@ def send_email(
         try:
             s.starttls()
             s.ehlo()
-        except Exception:
+        except Exception as e:
-            # STARTTLS is optional; ignore if unsupported.
+            if smtp_user or smtp_password:
+                raise RuntimeError(
+                    "email: SMTP STARTTLS failed; refusing to send credentials "
+                    "without TLS"
+                ) from e
+            # Without credentials, keep STARTTLS opportunistic so localhost or
+            # unauthenticated relay setups continue to work.
             pass  # nosec
         if smtp_user:
             s.login(smtp_user, smtp_password or "")

enroll/manifest.py

@@ -16,6 +16,7 @@ from .sopsutil import (
     encrypt_file_binary,
     require_sops_cmd,
 )
+from .validate import validate_harvest
 
 
 def _prepare_bundle_dir(
@@ -203,6 +204,14 @@ def manifest(
 
     td_out: Optional[tempfile.TemporaryDirectory] = None
     try:
+        validation = validate_harvest(resolved_bundle_dir)
+        if not validation.ok:
+            raise RuntimeError(
+                "harvest state does not match this Enroll version's schema; "
+                "please re-harvest the host with this version of Enroll.\n"
+                + validation.to_text().strip()
+            )
+
         if not sops_mode:
             if target == "puppet":
                 manifest_puppet_from_bundle_dir(
@@ -235,11 +244,6 @@ def manifest(
 
         td_out = tempfile.TemporaryDirectory(prefix="enroll-manifest-")
         tmp_out = Path(td_out.name) / "out"
-        tmp_out.mkdir(parents=True, exist_ok=True)
-        try:
-            os.chmod(tmp_out, 0o700)
-        except OSError:
-            pass
 
         if target == "puppet":
             manifest_puppet_from_bundle_dir(

enroll/manifest_safety.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import os
+import shutil
+import stat
+from pathlib import Path
+from typing import Iterator, Tuple
+
+
+class ArtifactSafetyError(RuntimeError):
+    """Raised when a harvest artifact path is unsafe to consume."""
+
+
+class ManifestOutputError(RuntimeError):
+    """Raised when a manifest output path is unsafe to use."""
+
+
+def _safe_relative_path(value: str, *, field: str) -> Path:
+    text = str(value or "").strip()
+    if not text:
+        raise ArtifactSafetyError(f"empty {field}")
+    if "\x00" in text:
+        raise ArtifactSafetyError(f"{field} contains NUL byte: {text!r}")
+    p = Path(text)
+    if p.is_absolute():
+        raise ArtifactSafetyError(f"{field} must be relative: {text!r}")
+    if any(part in {"", ".", ".."} for part in p.parts):
+        raise ArtifactSafetyError(f"{field} contains unsafe path component: {text!r}")
+    return p
+
+
+def prepare_manifest_output_dir(
+    out_dir: str | Path, *, allow_existing: bool = False
+) -> Path:
+    """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
+    with it by default; callers that intentionally support accumulation, such
+    as --fqdn site mode, may allow an existing directory but never a symlink or
+    non-directory path.
+    """
+
+    out = Path(out_dir).expanduser()
+    if os.path.lexists(out):
+        if not allow_existing:
+            raise ManifestOutputError(
+                "manifest output path already exists; refusing to overwrite: " f"{out}"
+            )
+        st = out.lstat()
+        if stat.S_ISLNK(st.st_mode):
+            raise ManifestOutputError(
+                f"manifest output path is a symlink; refusing to use: {out}"
+            )
+        if not out.is_dir():
+            raise ManifestOutputError(
+                f"manifest output path exists but is not a directory: {out}"
+            )
+        return out
+    out.mkdir(parents=True, exist_ok=False)
+    return out
+
+
+def _assert_no_symlink_components(path: Path, *, root: Path) -> None:
+    """Reject symlinks in any existing path component between root and path."""
+
+    try:
+        rel = path.relative_to(root)
+    except ValueError as e:
+        raise ArtifactSafetyError(f"artifact path escapes artifact root: {path}") from e
+
+    cur = root
+    for part in rel.parts:
+        cur = cur / part
+        try:
+            st = cur.lstat()
+        except FileNotFoundError:
+            # Missing components are handled by the final caller where relevant.
+            return
+        if stat.S_ISLNK(st.st_mode):
+            raise ArtifactSafetyError(f"artifact path contains symlink: {cur}")
+
+
+def safe_artifact_file(bundle_dir: str | Path, role: str, src_rel: str) -> Path:
+    """Return a harvested artifact file path only if it is safe to copy.
+
+    The path must remain under artifacts/<role>, contain no absolute or '..'
+    components, contain no symlinks in any path component, and refer to a
+    regular, non-hardlinked file.  This deliberately mirrors the tar extraction
+    hardening used for remote/SOPS/plain tarball bundles, but applies it to
+    directory bundles too.
+    """
+
+    role_path = _safe_relative_path(role, field="artifact role")
+    src_path = _safe_relative_path(src_rel, field="artifact src_rel")
+
+    artifacts_root = Path(bundle_dir).expanduser() / "artifacts"
+    root = artifacts_root / role_path
+    candidate = root / src_path
+
+    if artifacts_root.exists():
+        st = artifacts_root.lstat()
+        if stat.S_ISLNK(st.st_mode):
+            raise ArtifactSafetyError(
+                f"artifacts directory is a symlink: {artifacts_root}"
+            )
+
+    if root.exists():
+        _assert_no_symlink_components(root, root=artifacts_root)
+
+    _assert_no_symlink_components(candidate, root=artifacts_root)
+
+    try:
+        st = candidate.lstat()
+    except FileNotFoundError:
+        raise
+
+    if stat.S_ISLNK(st.st_mode):
+        raise ArtifactSafetyError(f"artifact is a symlink: {candidate}")
+    if not stat.S_ISREG(st.st_mode):
+        raise ArtifactSafetyError(f"artifact is not a regular file: {candidate}")
+    if st.st_nlink > 1:
+        raise ArtifactSafetyError(f"artifact is hardlinked: {candidate}")
+
+    resolved_root = artifacts_root.resolve(strict=True)
+    resolved_candidate = candidate.resolve(strict=True)
+    try:
+        resolved_candidate.relative_to(resolved_root)
+    except ValueError as e:
+        raise ArtifactSafetyError(
+            f"artifact path escapes artifact root: {candidate}"
+        ) from e
+
+    return candidate
+
+
+def iter_safe_artifact_files(
+    bundle_dir: str | Path, role: str
+) -> Iterator[Tuple[Path, str]]:
+    """Yield safe artifact files for a role as (path, src_rel)."""
+
+    role_path = _safe_relative_path(role, field="artifact role")
+    artifacts_dir = Path(bundle_dir).expanduser() / "artifacts" / role_path
+    if not artifacts_dir.exists():
+        return
+    if not artifacts_dir.is_dir():
+        raise ArtifactSafetyError(
+            f"artifact role path is not a directory: {artifacts_dir}"
+        )
+
+    for root, dirs, files in os.walk(artifacts_dir, followlinks=False):
+        root_p = Path(root)
+        for dirname in list(dirs):
+            p = root_p / dirname
+            try:
+                st = p.lstat()
+            except FileNotFoundError:
+                continue
+            if stat.S_ISLNK(st.st_mode):
+                raise ArtifactSafetyError(f"artifact directory is a symlink: {p}")
+        for filename in files:
+            p = root_p / filename
+            rel = p.relative_to(artifacts_dir).as_posix()
+            yield safe_artifact_file(bundle_dir, role, rel), rel
+
+
+def copy_safe_artifact_file(src: str | Path, dst: str | Path) -> None:
+    """Copy an already validated artifact file without following symlinks."""
+
+    shutil.copy2(src, dst, follow_symlinks=False)

enroll/puppet.py

@@ -4,7 +4,6 @@ import hashlib
 import json
 import re
 import shlex
-import shutil
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
 
@@ -16,6 +15,11 @@ from .cm import (
     role_order_key,
     markdown_list,
 )
+from .manifest_safety import (
+    copy_safe_artifact_file,
+    prepare_manifest_output_dir,
+    safe_artifact_file,
+)
 from .state import inventory_packages_from_state, roles_from_state
 from .jinjaturtle import (
     can_jinjify_path,
@@ -628,13 +632,14 @@ def _copy_artifact(
 ) -> Optional[str]:
     if not role or not src_rel:
         return None
-    src = Path(bundle_dir) / "artifacts" / role / src_rel
+    try:
-    if not src.is_file():
+        src = safe_artifact_file(bundle_dir, role, src_rel)
+    except FileNotFoundError:
         return None
     module_rel = Path(dst_prefix or "") / src_rel
     dst = dst_files_dir / module_rel
     dst.parent.mkdir(parents=True, exist_ok=True)
-    shutil.copy2(src, dst)
+    copy_safe_artifact_file(src, dst)
     return module_rel.as_posix()
 
 
@@ -1712,10 +1717,8 @@ class PuppetManifestRenderer:
         no_common_roles = self.no_common_roles
 
         state = PuppetRole.load_state(bundle_dir)
-        out = Path(out_dir)
         hiera_mode = bool(fqdn)
-        if out.exists() and not hiera_mode:
+        out = prepare_manifest_output_dir(out_dir, allow_existing=hiera_mode)
-            shutil.rmtree(out)
         manifests_dir = out / "manifests"
         modules_dir = out / "modules"
         manifests_dir.mkdir(parents=True, exist_ok=True)

enroll/salt.py

@@ -17,6 +17,11 @@ from .cm import (
     markdown_list,
 )
 from .jinjaturtle import jinjify_artifact, resolve_jinjaturtle_mode
+from .manifest_safety import (
+    copy_safe_artifact_file,
+    prepare_manifest_output_dir,
+    safe_artifact_file,
+)
 from .state import inventory_packages_from_state, roles_from_state
 from .yamlutil import yaml_dump_mapping, yaml_load_mapping_file
 
@@ -610,13 +615,14 @@ def _copy_artifact(
 ) -> Optional[str]:
     if not role or not src_rel:
         return None
-    src = Path(bundle_dir) / "artifacts" / role / src_rel
+    try:
-    if not src.is_file():
+        src = safe_artifact_file(bundle_dir, role, src_rel)
+    except FileNotFoundError:
         return None
     role_rel = Path(dst_prefix or "") / src_rel
     dst = dst_files_dir / role_rel
     dst.parent.mkdir(parents=True, exist_ok=True)
-    shutil.copy2(src, dst)
+    copy_safe_artifact_file(src, dst)
     return role_rel.as_posix()
 
 
@@ -1687,13 +1693,11 @@ class SaltManifestRenderer:
 
     def render(self) -> None:
         state = SaltRole.load_state(self.bundle_dir)
-        out = Path(self.out_dir)
+        fqdn_mode = bool(self.fqdn)
+        out = prepare_manifest_output_dir(self.out_dir, allow_existing=fqdn_mode)
         states_dir = out / "states"
         pillar_dir = out / "pillar"
-        fqdn_mode = bool(self.fqdn)
 
-        if out.exists() and not fqdn_mode:
-            shutil.rmtree(out)
         states_dir.mkdir(parents=True, exist_ok=True)
         if fqdn_mode:
             pillar_dir.mkdir(parents=True, exist_ok=True)

enroll/validate.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import json
+import os
+import stat
 import urllib.request
 from dataclasses import dataclass
 from pathlib import Path
@@ -9,6 +11,7 @@ from typing import Any, Dict, List, Optional, Set, Tuple
 import jsonschema
 
 from .diff import BundleRef, _bundle_from_input
+from .manifest_safety import ArtifactSafetyError, safe_artifact_file
 from .state import load_state
 
 
@@ -171,7 +174,7 @@ def validate_harvest(
             except Exception as e:  # noqa: BLE001
                 errors.append(f"failed to load/validate schema: {e!r}")
 
-        # Artifact existence checks
+        # Artifact existence and safety checks.
         artifacts_dir = bundle.dir / "artifacts"
         referenced: Set[Tuple[str, str]] = set()
         for role_name, mf in _iter_managed_files(state):
@@ -188,15 +191,15 @@ def validate_harvest(
                 continue
 
             referenced.add((role_name, src_rel))
-            p = artifacts_dir / role_name / src_rel
+            try:
-            if not p.exists():
+                safe_artifact_file(bundle.dir, role_name, src_rel)
+            except FileNotFoundError:
                 errors.append(
                     f"missing artifact for role {role_name}: artifacts/{role_name}/{src_rel}"
                 )
-                continue
+            except ArtifactSafetyError as e:
-            if not p.is_file():
                 errors.append(
-                    f"artifact is not a file for role {role_name}: artifacts/{role_name}/{src_rel}"
+                    f"unsafe artifact for role {role_name}: artifacts/{role_name}/{src_rel}: {e}"
                 )
 
         # Runtime firewall snapshots are generated artifacts rather than managed files.
@@ -211,43 +214,83 @@ def validate_harvest(
                         f"firewall_runtime {key} has suspicious src_rel: {src_rel!r}"
                     )
                     continue
-                referenced.add(
+                role_name = str(fw.get("role_name") or "firewall_runtime")
-                    (str(fw.get("role_name") or "firewall_runtime"), src_rel)
+                referenced.add((role_name, src_rel))
-                )
+                try:
-                p = (
+                    safe_artifact_file(bundle.dir, role_name, src_rel)
-                    artifacts_dir
+                except FileNotFoundError:
-                    / str(fw.get("role_name") or "firewall_runtime")
-                    / src_rel
-                )
-                if not p.exists():
                     errors.append(
                         "missing firewall runtime artifact: "
-                        f"artifacts/{fw.get('role_name') or 'firewall_runtime'}/{src_rel}"
+                        f"artifacts/{role_name}/{src_rel}"
                     )
-                elif not p.is_file():
+                except ArtifactSafetyError as e:
                     errors.append(
-                        "firewall runtime artifact is not a file: "
+                        "unsafe firewall runtime artifact: "
-                        f"artifacts/{fw.get('role_name') or 'firewall_runtime'}/{src_rel}"
+                        f"artifacts/{role_name}/{src_rel}: {e}"
                     )
 
-        # Warn if there are extra files in artifacts not referenced.
+        # Validate the whole artifact tree too, so unreferenced symlinks,
+        # hardlinks, special files, and path-shaping tricks do not survive
+        # validation simply because no managed_file currently references them.
         if artifacts_dir.exists() and artifacts_dir.is_dir():
-            for fp in artifacts_dir.rglob("*"):
+            for root, dirs, files in os.walk(artifacts_dir, followlinks=False):
-                if not fp.is_file():
+                root_p = Path(root)
-                    continue
+                for name in list(dirs):
-                try:
+                    fp = root_p / name
-                    rel = fp.relative_to(artifacts_dir)
+                    try:
-                except ValueError:
+                        st = fp.lstat()
-                    continue
+                    except FileNotFoundError:
-                parts = rel.parts
+                        continue
-                if len(parts) < 2:
+                    if stat.S_ISLNK(st.st_mode):
-                    continue
+                        errors.append(f"artifact directory is a symlink: {fp}")
-                role_name = parts[0]
+                    elif not stat.S_ISDIR(st.st_mode):
-                src_rel = "/".join(parts[1:])
+                        errors.append(f"artifact directory is not a directory: {fp}")
-                if (role_name, src_rel) not in referenced:
+
-                    warnings.append(
+                for name in files:
-                        f"unreferenced artifact present: artifacts/{role_name}/{src_rel}"
+                    fp = root_p / name
-                    )
+                    try:
+                        st = fp.lstat()
+                    except FileNotFoundError:
+                        continue
+                    try:
+                        rel = fp.relative_to(artifacts_dir)
+                    except ValueError:
+                        errors.append(f"artifact escapes artifact root: {fp}")
+                        continue
+                    parts = rel.parts
+                    if len(parts) < 2:
+                        errors.append(f"artifact is not under a role directory: {fp}")
+                        continue
+                    role_name = parts[0]
+                    src_rel = "/".join(parts[1:])
+
+                    if stat.S_ISLNK(st.st_mode):
+                        errors.append(
+                            f"artifact is a symlink: artifacts/{role_name}/{src_rel}"
+                        )
+                        continue
+                    if not stat.S_ISREG(st.st_mode):
+                        errors.append(
+                            f"artifact is not a regular file: artifacts/{role_name}/{src_rel}"
+                        )
+                        continue
+                    if st.st_nlink > 1:
+                        errors.append(
+                            f"artifact is hardlinked: artifacts/{role_name}/{src_rel}"
+                        )
+                        continue
+                    try:
+                        safe_artifact_file(bundle.dir, role_name, src_rel)
+                    except (FileNotFoundError, ArtifactSafetyError) as e:
+                        errors.append(
+                            f"unsafe artifact: artifacts/{role_name}/{src_rel}: {e}"
+                        )
+                        continue
+
+                    if (role_name, src_rel) not in referenced:
+                        warnings.append(
+                            f"unreferenced artifact present: artifacts/{role_name}/{src_rel}"
+                        )
 
         return ValidationResult(errors=errors, warnings=warnings)
     finally:

tests/state_helpers.py

@@ -0,0 +1,234 @@
+from __future__ import annotations
+
+import copy
+import json
+from pathlib import Path
+from typing import Any
+
+_VALID_REASON_FALLBACKS = {
+    "dangerous_user_dotfile": "user_shell_rc",
+    "possible_secret": "sensitive_content",
+}
+
+_COMMON_ROLES = {
+    "users",
+    "apt_config",
+    "dnf_config",
+    "etc_custom",
+    "usr_local_custom",
+    "extra_paths",
+}
+
+
+def _common_role(name: str) -> dict[str, Any]:
+    out: dict[str, Any] = {
+        "role_name": name,
+        "managed_dirs": [],
+        "managed_files": [],
+        "excluded": [],
+        "notes": [],
+    }
+    if name == "users":
+        out["users"] = []
+    if name == "extra_paths":
+        out["include_patterns"] = []
+        out["exclude_patterns"] = []
+        out["managed_links"] = []
+    return out
+
+
+def _normalise_managed_file(mf: dict[str, Any]) -> None:
+    reason = mf.get("reason")
+    if reason in _VALID_REASON_FALLBACKS:
+        mf["reason"] = _VALID_REASON_FALLBACKS[reason]
+    mf.setdefault("owner", "root")
+    mf.setdefault("group", "root")
+    mf.setdefault("mode", "0644")
+    mf.setdefault("reason", "modified_conffile")
+
+
+def _normalise_managed_dir(md: dict[str, Any]) -> None:
+    md.setdefault("owner", "root")
+    md.setdefault("group", "root")
+    md.setdefault("mode", "0755")
+    if md.get("reason") in {None, "parent_dir"}:
+        md["reason"] = "parent_of_managed_file"
+
+
+def _normalise_managed_link(ml: dict[str, Any]) -> None:
+    ml.setdefault("reason", "enabled_symlink")
+
+
+def _normalise_common_role(role: dict[str, Any], name: str) -> None:
+    role.setdefault("role_name", name)
+    role.setdefault("managed_dirs", [])
+    role.setdefault("managed_files", [])
+    role.setdefault("excluded", [])
+    role.setdefault("notes", [])
+    for mf in role.get("managed_files") or []:
+        if isinstance(mf, dict):
+            _normalise_managed_file(mf)
+    for md in role.get("managed_dirs") or []:
+        if isinstance(md, dict):
+            _normalise_managed_dir(md)
+    for ml in role.get("managed_links") or []:
+        if isinstance(ml, dict):
+            _normalise_managed_link(ml)
+    for ex in role.get("excluded") or []:
+        if isinstance(ex, dict) and ex.get("reason") in _VALID_REASON_FALLBACKS:
+            ex["reason"] = _VALID_REASON_FALLBACKS[ex["reason"]]
+
+
+def make_schema_valid_state(state: dict[str, Any]) -> dict[str, Any]:
+    """Return a current-schema harvest state from a compact renderer fixture.
+
+    Many renderer tests intentionally build only the fields needed by the
+    renderer under test. Manifest now validates strictly before rendering, so
+    those fixtures need current-schema boilerplate too.
+    """
+
+    st = copy.deepcopy(state)
+    st.pop("schema_version", None)
+
+    enroll = st.setdefault("enroll", {})
+    enroll.setdefault("version", "0.0.test")
+    enroll.setdefault("harvest_time", 0)
+
+    host = st.setdefault("host", {})
+    host.setdefault("hostname", "testhost")
+    host.setdefault("os", "unknown")
+    host.setdefault("pkg_backend", "dpkg")
+    host.setdefault("os_release", {})
+
+    inv = st.setdefault("inventory", {})
+    inv.setdefault("packages", {})
+    for pkg in (inv.get("packages") or {}).values():
+        if not isinstance(pkg, dict):
+            continue
+        pkg.setdefault("version", None)
+        pkg.setdefault("arches", [])
+        installations = pkg.setdefault("installations", [])
+        for inst in installations:
+            if isinstance(inst, dict):
+                inst.setdefault("version", str(pkg.get("version") or "1.0"))
+                inst.setdefault("arch", "amd64")
+        observed = pkg.setdefault("observed_via", [])
+        for ov in observed:
+            if isinstance(ov, dict) and ov.get("kind") not in {
+                "user_installed",
+                "systemd_unit",
+                "package_role",
+                "firewall_runtime",
+            }:
+                ov["kind"] = "package_role"
+                ov.setdefault("ref", "package")
+        pkg.setdefault("roles", [])
+
+    roles = st.setdefault("roles", {})
+    for name in _COMMON_ROLES:
+        cur = roles.get(name)
+        if not isinstance(cur, dict):
+            roles[name] = _common_role(name)
+        else:
+            _normalise_common_role(cur, name)
+
+    roles.setdefault("services", [])
+    roles.setdefault("packages", [])
+
+    users = roles.get("users") or {}
+    users.setdefault("users", [])
+    for user in users.get("users") or []:
+        if not isinstance(user, dict):
+            continue
+        user.setdefault("uid", 0)
+        user.setdefault("gid", user.get("uid", 0))
+        user.setdefault("gecos", "")
+        user.setdefault("home", f"/home/{user.get('name', 'user')}")
+        user.setdefault("shell", "/bin/sh")
+        user.setdefault("primary_group", user.get("name", "users"))
+        user.setdefault("supplementary_groups", [])
+
+    extra = roles.get("extra_paths") or {}
+    extra.setdefault("include_patterns", [])
+    extra.setdefault("exclude_patterns", [])
+    extra.setdefault("managed_links", [])
+
+    for svc in roles.get("services") or []:
+        if not isinstance(svc, dict):
+            continue
+        _normalise_common_role(svc, str(svc.get("role_name") or "service_role"))
+        svc.setdefault("unit", "example.service")
+        svc.setdefault("packages", [])
+        svc.setdefault("active_state", None)
+        svc.setdefault("sub_state", None)
+        svc.setdefault("unit_file_state", None)
+        svc.setdefault("condition_result", None)
+
+    for pkg in roles.get("packages") or []:
+        if not isinstance(pkg, dict):
+            continue
+        _normalise_common_role(
+            pkg, str(pkg.get("role_name") or pkg.get("package") or "package_role")
+        )
+        pkg.setdefault("package", str(pkg.get("role_name") or "package"))
+
+    if isinstance(roles.get("sysctl"), dict):
+        sysctl = roles["sysctl"]
+        sysctl.setdefault("role_name", "sysctl")
+        sysctl.setdefault("managed_files", [])
+        sysctl.setdefault("parameters", {})
+        sysctl.setdefault("notes", [])
+        sysctl.pop("managed_dirs", None)
+        sysctl.pop("managed_links", None)
+        for mf in sysctl.get("managed_files") or []:
+            if isinstance(mf, dict):
+                _normalise_managed_file(mf)
+
+    if isinstance(roles.get("firewall_runtime"), dict):
+        fw = roles["firewall_runtime"]
+        fw.setdefault("role_name", "firewall_runtime")
+        fw.setdefault("packages", [])
+        fw.setdefault("ipset_save", None)
+        fw.setdefault("ipset_sets", [])
+        fw.setdefault("iptables_v4_save", None)
+        fw.setdefault("iptables_v6_save", None)
+        fw.setdefault("notes", [])
+
+    if isinstance(roles.get("flatpak"), dict):
+        roles["flatpak"].setdefault("role_name", "flatpak")
+    if isinstance(roles.get("snap"), dict):
+        roles["snap"].setdefault("role_name", "snap")
+    if isinstance(roles.get("container_images"), dict):
+        ci = roles["container_images"]
+        ci.setdefault("role_name", "container_images")
+        ci.setdefault("images", [])
+        ci.setdefault("notes", [])
+        for img in ci.get("images") or []:
+            if not isinstance(img, dict):
+                continue
+            img.setdefault("engine", "docker")
+            img.setdefault("scope", "system")
+            img.setdefault("user", None)
+            img.setdefault("home", None)
+            img.setdefault("image_id", None)
+            img.setdefault("repo_tags", [])
+            img.setdefault("repo_digests", [])
+            img.setdefault("pull_ref", None)
+            img.setdefault("tag_aliases", [])
+            img.setdefault("os", None)
+            img.setdefault("architecture", None)
+            img.setdefault("variant", None)
+            img.setdefault("platform", None)
+            img.setdefault("size", None)
+            img.setdefault("created", None)
+            img.setdefault("source", "test")
+            img.setdefault("notes", [])
+
+    return st
+
+
+def write_schema_state(bundle: Path, state: dict[str, Any]) -> None:
+    bundle.mkdir(parents=True, exist_ok=True)
+    (bundle / "state.json").write_text(
+        json.dumps(make_schema_valid_state(state), indent=2), encoding="utf-8"
+    )

tests/test_cli_config_and_sops.py

@@ -23,10 +23,10 @@ def test_discover_config_path_precedence(monkeypatch, tmp_path: Path):
     assert _discover_config_path(["harvest"]) == cfg
 
 
-def test_discover_config_path_finds_local_and_xdg(monkeypatch, tmp_path: Path):
+def test_discover_config_path_ignores_local_and_finds_xdg(monkeypatch, tmp_path: Path):
     from enroll.cli import _discover_config_path
 
-    # local file in cwd
+    # local files in cwd are deliberately ignored unless passed via --config
     cwd = tmp_path / "cwd"
     cwd.mkdir()
     local = cwd / "enroll.ini"
@@ -35,7 +35,8 @@ def test_discover_config_path_finds_local_and_xdg(monkeypatch, tmp_path: Path):
     monkeypatch.chdir(cwd)
     monkeypatch.delenv("ENROLL_CONFIG", raising=False)
     monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
-    assert _discover_config_path(["harvest"]) == local
+    assert _discover_config_path(["harvest"]) is None
+    assert _discover_config_path(["--config", str(local), "harvest"]) == local
 
     # xdg config fallback
     monkeypatch.chdir(tmp_path)

tests/test_cli_helpers.py

@@ -8,7 +8,7 @@ from pathlib import Path
 
 
 def test_discover_config_path_precedence(tmp_path: Path, monkeypatch):
-    """_discover_config_path: --config > ENROLL_CONFIG > ./enroll.ini > XDG."""
+    """_discover_config_path: --config > ENROLL_CONFIG > XDG."""
     from enroll.cli import _discover_config_path
 
     cfg1 = tmp_path / "one.ini"
@@ -27,14 +27,14 @@ def test_discover_config_path_precedence(tmp_path: Path, monkeypatch):
     monkeypatch.setenv("ENROLL_CONFIG", str(cfg2))
     assert _discover_config_path([]) == cfg2
 
-    # Local ./enroll.ini fallback.
+    # Local ./enroll.ini is ignored unless passed explicitly.
     monkeypatch.delenv("ENROLL_CONFIG", raising=False)
     local = tmp_path / "enroll.ini"
     local.write_text("[enroll]\n", encoding="utf-8")
-    assert _discover_config_path([]) == local
+    assert _discover_config_path([]) is None
+    assert _discover_config_path(["--config", str(local)]) == local
 
     # XDG fallback.
-    local.unlink()
     xdg = tmp_path / "xdg"
     cfg3 = xdg / "enroll" / "enroll.ini"
     cfg3.parent.mkdir(parents=True)

tests/test_diff_ignore_versions_exclude_enforce.py

@@ -244,6 +244,7 @@ def test_enforce_old_harvest_runs_ansible_with_tags_from_file_drift(
     # Stub manifest generation to only create playbook.yml (fast, no real roles needed).
     def fake_manifest(_harvest_dir: str, out_dir: str, **_kwargs):
         out = Path(out_dir)
+        out.mkdir(parents=True, exist_ok=False)
         (out / "playbook.yml").write_text(
             "---\n- hosts: all\n  gather_facts: false\n  roles: []\n",
             encoding="utf-8",
@@ -363,7 +364,10 @@ def test_enforce_old_harvest_runs_puppet_target(monkeypatch, tmp_path: Path):
     argv = calls.get("argv")
     assert argv and argv[:2] == ["/usr/bin/puppet", "apply"]
     assert "--modulepath" in argv
-    assert str(Path(calls["cwd"]) / "manifests" / "site.pp") in argv
+    assert any(
+        str(Path(calls["cwd"]) / "manifest" / "manifests" / "site.pp") == str(a)
+        for a in argv
+    )
 
 
 def test_enforce_old_harvest_runs_salt_target(monkeypatch, tmp_path: Path):
@@ -418,7 +422,7 @@ def test_enforce_old_harvest_runs_salt_target(monkeypatch, tmp_path: Path):
     assert "--local" in argv
     assert "--file-root" in argv
     assert "state.apply" in argv
-    assert str(Path(calls["cwd"]) / "states") in argv
+    assert str(Path(calls["cwd"]) / "manifest" / "states") in argv
 
 
 def test_cli_diff_enforce_forwards_target(monkeypatch):

tests/test_diff_notifications.py

@@ -81,3 +81,42 @@ def test_send_email_raises_when_no_delivery_method(monkeypatch):
             from_addr="a@example.com",
             to_addrs=["b@example.com"],
         )
+
+
+def test_send_email_refuses_smtp_auth_without_starttls(monkeypatch):
+    from enroll.diff import send_email
+
+    class FakeSMTP:
+        def __init__(self, *_args, **_kwargs):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def ehlo(self):
+            pass
+
+        def starttls(self):
+            raise RuntimeError("no starttls")
+
+        def login(self, *_args):
+            raise AssertionError("login should not be called without TLS")
+
+        def send_message(self, *_args):
+            raise AssertionError("message should not be sent without TLS")
+
+    monkeypatch.setattr("smtplib.SMTP", FakeSMTP)
+
+    with pytest.raises(RuntimeError, match="STARTTLS failed"):
+        send_email(
+            subject="Subj",
+            body="Body",
+            from_addr="a@example.com",
+            to_addrs=["b@example.com"],
+            smtp="smtp.example.com:587",
+            smtp_user="user",
+            smtp_password="secret",
+        )

tests/test_jinjaturtle.py

@@ -1,6 +1,7 @@
-import json
 from pathlib import Path
 
+from tests.state_helpers import write_schema_state
+
 import enroll.manifest as manifest_mod
 import enroll.jinjaturtle as jinjaturtle_mod
 from enroll.jinjaturtle import JinjifyResult
@@ -103,7 +104,7 @@ def test_manifest_uses_jinjaturtle_templates_and_does_not_copy_raw(
     }
 
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     # Pretend jinjaturtle exists.
     monkeypatch.setattr(

tests/test_manifest.py

@@ -1,4 +1,3 @@
-import json
 from pathlib import Path
 
 import os
@@ -7,6 +6,7 @@ import tarfile
 import pytest
 
 import enroll.manifest as manifest
+from tests.state_helpers import write_schema_state
 import enroll.jinjaturtle as jinjaturtle_mod
 from enroll import ansible as ansible_layout
 from enroll import ansible as ansible_tasks
@@ -84,8 +84,7 @@ def _minimal_package_state(packages):
 
 
 def _write_state(bundle: Path, state: dict) -> None:
-    bundle.mkdir(parents=True, exist_ok=True)
+    write_schema_state(bundle, state)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
 
 
 def test_manifest_writes_roles_and_playbook_with_clean_when(tmp_path: Path):
@@ -230,7 +229,7 @@ def test_manifest_writes_roles_and_playbook_with_clean_when(tmp_path: Path):
     }
 
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     # Create artifact for etc_custom file so copy works
     (bundle / "artifacts" / "etc_custom" / "etc" / "default").mkdir(
@@ -936,7 +935,7 @@ def test_manifest_site_mode_creates_host_inventory_and_raw_files(tmp_path: Path)
     }
 
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     # Artifacts for usr_local_custom file so copy works.
     (bundle / "artifacts" / "usr_local_custom" / "usr" / "local" / "etc").mkdir(
@@ -1087,7 +1086,7 @@ def test_manifest_includes_dnf_config_role_when_present(tmp_path: Path):
     }
 
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -1224,7 +1223,7 @@ def test_manifest_orders_cron_and_logrotate_at_playbook_tail(tmp_path: Path):
     )
 
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -1363,9 +1362,7 @@ def test_manifest_applies_jinjaturtle_to_jinjifyable_managed_file(
             },
         },
     }
-    (bundle / "state.json").write_text(
+    write_schema_state(bundle, state)
-        __import__("json").dumps(state), encoding="utf-8"
-    )
 
     monkeypatch.setattr(jinjaturtle_mod, "find_jinjaturtle_cmd", lambda: "jinjaturtle")
 
@@ -1463,7 +1460,7 @@ def test_manifest_writes_firewall_runtime_role(tmp_path: Path):
             },
         },
     }
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -1768,7 +1765,7 @@ def test_manifest_renders_flatpak_and_snap_details(tmp_path: Path):
         },
     }
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -1844,7 +1841,7 @@ def test_users_role_without_portable_apps_omits_community_general_tasks(tmp_path
         },
     }
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -1922,7 +1919,7 @@ def test_users_role_only_creates_ssh_dir_when_managed_ssh_files_exist(tmp_path):
         },
     }
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -1972,7 +1969,7 @@ def test_manifest_emits_flatpak_role_even_when_no_flatpaks(tmp_path):
         }
     }
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -2039,7 +2036,7 @@ def test_manifest_avoids_package_role_collision_with_flatpak_singleton(tmp_path)
         }
     }
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out), no_common_roles=True)
 
@@ -2135,7 +2132,7 @@ def test_manifest_writes_sysctl_role(tmp_path: Path):
             },
         },
     }
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -2237,7 +2234,7 @@ def test_manifest_renders_container_image_role_for_ansible(tmp_path: Path):
         }
     }
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 
@@ -2314,7 +2311,7 @@ def test_manifest_writes_container_images_to_hostvars_in_fqdn_mode(tmp_path: Pat
         }
     }
     bundle.mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out), fqdn="host.example.test")
 
@@ -2329,3 +2326,14 @@ def test_manifest_writes_container_images_to_hostvars_in_fqdn_mode(tmp_path: Pat
     assert "container_images: []" in defaults
     assert digest in hostvars
     assert "role: container_images" in playbook
+
+
+def test_manifest_non_fqdn_refuses_existing_output(tmp_path: Path):
+    bundle = tmp_path / "bundle"
+    out = tmp_path / "ansible"
+    bundle.mkdir(parents=True)
+    out.mkdir()
+    write_schema_state(bundle, _minimal_package_state([]))
+
+    with pytest.raises(RuntimeError, match="already exists"):
+        manifest.manifest(str(bundle), str(out), no_common_roles=True)

tests/test_manifest_puppet.py

@@ -5,6 +5,8 @@ from pathlib import Path
 
 import yaml
 
+from tests.state_helpers import write_schema_state
+
 from enroll import manifest
 from enroll.puppet import (
     PuppetRole,
@@ -15,8 +17,7 @@ from enroll.puppet import (
 
 
 def _write_state(bundle: Path, state: dict) -> None:
-    bundle.mkdir(parents=True, exist_ok=True)
+    write_schema_state(bundle, state)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
 
 
 def test_manifest_puppet_writes_control_repo_style_output(tmp_path: Path):

tests/test_manifest_salt.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
-import json
 from collections import OrderedDict
 from pathlib import Path
 
 import yaml
 
+from tests.state_helpers import write_schema_state
+
 from enroll import manifest
 from enroll.salt import (
     SaltRole,
@@ -18,8 +19,7 @@ from enroll.salt import (
 
 
 def _write_state(bundle: Path, state: dict) -> None:
-    bundle.mkdir(parents=True, exist_ok=True)
+    write_schema_state(bundle, state)
-    (bundle / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
 
 
 def _sample_state() -> dict:

tests/test_manifest_symlinks.py

@@ -1,6 +1,7 @@
-import json
 from pathlib import Path
 
+from tests.state_helpers import write_schema_state
+
 import enroll.manifest as manifest
 
 
@@ -92,7 +93,7 @@ def test_manifest_emits_symlink_tasks_and_vars(tmp_path: Path):
 
     bundle.mkdir(parents=True, exist_ok=True)
     (bundle / "artifacts").mkdir(parents=True, exist_ok=True)
-    (bundle / "state.json").write_text(json.dumps(state), encoding="utf-8")
+    write_schema_state(bundle, state)
 
     manifest.manifest(str(bundle), str(out))
 

tests/test_validate.py

@@ -411,3 +411,45 @@ def test_validate_harvest_no_schema_option(tmp_path: Path):
     result = validate_harvest(str(bundle_dir), no_schema=True)
     assert result.ok is False
     assert any("failed to parse" in e for e in result.errors)
+
+
+def test_validate_harvest_rejects_artifact_symlink(tmp_path: Path):
+    bundle_dir = tmp_path / "bundle"
+    artifact = bundle_dir / "artifacts" / "users" / "etc" / "shadow"
+    artifact.parent.mkdir(parents=True)
+    artifact.symlink_to("/etc/shadow")
+    (bundle_dir / "state.json").write_text(
+        json.dumps(
+            {
+                "roles": {
+                    "users": {
+                        "managed_files": [
+                            {"path": "/etc/shadow", "src_rel": "etc/shadow"}
+                        ]
+                    }
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    result = validate_harvest(str(bundle_dir), no_schema=True)
+
+    assert result.ok is False
+    assert any("symlink" in e for e in result.errors)
+
+
+def test_validate_harvest_rejects_unreferenced_artifact_symlink(tmp_path: Path):
+    bundle_dir = tmp_path / "bundle"
+    artifact = bundle_dir / "artifacts" / "users" / "etc" / "shadow"
+    artifact.parent.mkdir(parents=True)
+    artifact.symlink_to("/etc/shadow")
+    (bundle_dir / "state.json").write_text(
+        json.dumps({"roles": {"users": {"managed_files": []}}}),
+        encoding="utf-8",
+    )
+
+    result = validate_harvest(str(bundle_dir), no_schema=True)
+
+    assert result.ok is False
+    assert any("symlink" in e for e in result.errors)