mig5/enroll
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Activity Actions

Remove puppet and salt

Browse source
This commit is contained in:
Miguel Jacq 2026-06-25 16:54:23 +10:00
parent 88e9dba39a
commit e9d7d74445
Signed by: mig5
GPG key ID: 03906B4110AAD3B8
24 changed files with 256 additions and 7714 deletions
Download patch file Download diff file Expand all files Collapse all files

2
enroll/ansible.py
View file

@ -724,7 +724,7 @@ def _write_ansible_role_vars(
*,
site_defaults: Optional[Dict[str, Any]] = None,
) -> None:
"""Write role variables using the same mode split as Puppet Hiera/Salt Pillar."""
"""Write role variables using the single-site/site-mode split."""
if ctx.site_mode:
_write_role_defaults(role_dir, site_defaults or {})

25
enroll/cli.py
View file

@ -458,15 +458,9 @@ def _encrypt_harvest_dir_to_sops(
def _add_common_manifest_args(p: argparse.ArgumentParser) -> None:
p.add_argument(
"--target",
choices=["ansible", "puppet", "salt"],
default="ansible",
help="Manifest target to generate (default: ansible).",
)
p.add_argument(
"--fqdn",
help="Host FQDN/name for site-mode output (creates target-specific host inventory/data such as Ansible host_vars, Puppet Hiera, or Salt pillar).",
help="Host FQDN/name for site-mode output (creates Ansible host_vars for that host).",
)
p.add_argument(
"--no-common-roles",
@ -808,15 +802,6 @@ def main() -> None:
"Enroll does not attempt to downgrade packages; if the only drift is package version upgrades (or newly installed packages), enforcement is skipped."
),
)
d.add_argument(
"--target",
choices=["ansible", "puppet", "salt"],
default="ansible",
help=(
"Configuration-management target to use with --enforce (default: ansible). "
"Requires ansible-playbook, puppet, or salt-call on PATH as appropriate."
),
)
d.add_argument(
"--out",
help="Write the report to this file instead of stdout.",
@ -1119,7 +1104,6 @@ def main() -> None:
jinjaturtle=_jt_mode(args),
sops_fingerprints=getattr(args, "sops", None),
no_common_roles=bool(getattr(args, "no_common_roles", False)),
target=getattr(args, "target", "ansible"),
)
if getattr(args, "sops", None) and out_enc:
print(str(out_enc))
@ -1135,7 +1119,7 @@ def main() -> None:
)
# Optional enforcement: if drift is detected, attempt to restore the
# system to the *old* (baseline) state using the selected target.
# system to the *old* (baseline) state using ansible.
if bool(getattr(args, "enforce", False)):
if has_changes:
if not has_enforceable_drift(report):
@ -1153,7 +1137,6 @@ def main() -> None:
args.old,
sops_mode=bool(getattr(args, "sops", False)),
report=report,
target=getattr(args, "target", "ansible"),
)
except Exception as e:
raise SystemExit(
@ -1258,7 +1241,6 @@ def main() -> None:
jinjaturtle=_jt_mode(args),
sops_fingerprints=list(sops_fps),
no_common_roles=bool(getattr(args, "no_common_roles", False)),
target=getattr(args, "target", "ansible"),
)
if not args.harvest:
print(str(out_file))
@ -1291,7 +1273,6 @@ def main() -> None:
fqdn=args.fqdn,
jinjaturtle=_jt_mode(args),
no_common_roles=bool(getattr(args, "no_common_roles", False)),
target=getattr(args, "target", "ansible"),
)
# For usability (when --harvest wasn't provided), print the harvest path.
if not args.harvest:
@ -1324,7 +1305,6 @@ def main() -> None:
jinjaturtle=_jt_mode(args),
sops_fingerprints=list(sops_fps),
no_common_roles=bool(getattr(args, "no_common_roles", False)),
target=getattr(args, "target", "ansible"),
)
if not args.harvest:
print(str(out_file))
@ -1345,7 +1325,6 @@ def main() -> None:
fqdn=args.fqdn,
jinjaturtle=_jt_mode(args),
no_common_roles=bool(getattr(args, "no_common_roles", False)),
target=getattr(args, "target", "ansible"),
)
except RemoteSudoPasswordRequired:
raise SystemExit(

15
enroll/cm.py
View file

@ -22,9 +22,9 @@ from .state import load_state, state_path, write_state
class CMModule:
"""Renderer-neutral configuration-management resource group.
A CMModule is intentionally small: it captures the resources that a target
renderer can turn into Ansible tasks, Puppet resources, Salt states, etc.
The renderer may still decide how to name/include/order the group.
A CMModule is intentionally small: it captures the resources that the
renderer turns into Ansible tasks. The renderer may still decide how to
name/include/order the group.
"""
role_name: str
@ -806,12 +806,11 @@ def _drop_duplicate_mapping_items(
def resolve_catalog_conflicts(modules: Iterable[CMModule]) -> None:
"""Resolve global catalog conflicts before renderer output.
"""Resolve global catalog conflicts in the shared model.
Puppet and Salt compile a single resource catalog. Ansible can tolerate the
same package, service, or parent directory appearing in more than one role;
catalog targets cannot. Resolve those conflicts in the shared model rather
than deleting renderer output after the fact.
Deduplicates the same package, service, or parent directory appearing in
more than one role. The Ansible renderer tolerates such duplicates, but this
helper remains available for any catalog-style consumer of the shared model.
"""
ordered = list(modules)

6
enroll/debian.py
View file

@ -228,6 +228,10 @@ def read_pkg_md5sums(pkg: str) -> Dict[str, str]:
line = line.strip()
if not line:
continue
md5, rel = line.split(None, 1)
parts = line.split(None, 1)
if len(parts) != 2:
# Skip malformed/truncated lines instead of aborting the harvest.
continue
md5, rel = parts
m[rel.strip()] = md5.strip()
return m

175
enroll/diff.py
View file

@ -682,40 +682,7 @@ def _role_tag(role: str) -> str:
return f"role_{safe}"
def _normalise_enforcement_target(target: str) -> str:
t = str(target or "ansible").strip().lower()
if t not in {"ansible", "puppet", "salt"}:
raise ValueError(f"unsupported enforcement target: {target!r}")
return t
def _enforcement_tool(target: str) -> Tuple[str, str]:
"""Return (binary-name, human-label) for a local enforcement target."""
if target == "puppet":
return "puppet", "puppet apply"
if target == "salt":
return "salt-call", "salt-call"
return "ansible-playbook", "ansible-playbook"
def _require_enforcement_tool(target: str) -> Tuple[str, str]:
binary, label = _enforcement_tool(target)
exe = shutil.which(binary)
if not exe:
install_hint = {
"ansible": "Ansible",
"puppet": "Puppet",
"salt": "Salt",
}.get(target, target)
raise RuntimeError(
f"{binary} not found on PATH "
f"(cannot enforce with target {target}; install {install_hint})"
)
return exe, label
def _enforcement_command(
target: str,
exe: str,
manifest_dir: Path,
*,
@ -724,69 +691,27 @@ def _enforcement_command(
"""Return the local apply command and environment for a rendered manifest."""
env = dict(os.environ)
if target == "ansible":
playbook = manifest_dir / "playbook.yml"
if not playbook.exists():
raise RuntimeError(
f"manifest did not produce expected playbook.yml at {playbook}"
)
playbook = manifest_dir / "playbook.yml"
if not playbook.exists():
raise RuntimeError(
f"manifest did not produce expected playbook.yml at {playbook}"
)
cfg = manifest_dir / "ansible.cfg"
if cfg.exists():
env["ANSIBLE_CONFIG"] = str(cfg)
cfg = manifest_dir / "ansible.cfg"
if cfg.exists():
env["ANSIBLE_CONFIG"] = str(cfg)
cmd = [
exe,
"-i",
"localhost,",
"-c",
"local",
str(playbook),
]
if tags:
cmd.extend(["--tags", ",".join(tags)])
return cmd, env
if target == "puppet":
site_pp = manifest_dir / "manifests" / "site.pp"
if not site_pp.exists():
raise RuntimeError(
f"manifest did not produce expected Puppet site.pp at {site_pp}"
)
cmd = [
exe,
"apply",
"--modulepath",
str(manifest_dir / "modules"),
]
hiera_config = manifest_dir / "hiera.yaml"
if hiera_config.exists():
cmd.extend(["--hiera_config", str(hiera_config)])
cmd.append(str(site_pp))
return cmd, env
if target == "salt":
states_dir = manifest_dir / "states"
top_sls = states_dir / "top.sls"
if not top_sls.exists():
raise RuntimeError(
f"manifest did not produce expected Salt top.sls at {top_sls}"
)
cmd = [
exe,
"--local",
"--file-root",
str(states_dir),
]
pillar_dir = manifest_dir / "pillar"
if pillar_dir.exists():
cmd.extend(["--pillar-root", str(pillar_dir)])
cmd.extend(["state.apply"])
return cmd, env
raise ValueError(f"unsupported enforcement target: {target!r}")
cmd = [
exe,
"-i",
"localhost,",
"-c",
"local",
str(playbook),
]
if tags:
cmd.extend(["--tags", ",".join(tags)])
return cmd, env
def _enforcement_plan(
@ -898,22 +823,18 @@ def enforce_old_harvest(
*,
sops_mode: bool = False,
report: Optional[Dict[str, Any]] = None,
target: str = "ansible",
) -> Dict[str, Any]:
"""Enforce the *old* (baseline) harvest state on the current machine.
This renders a temporary manifest from the old harvest using the requested
target, then runs the target's local apply command:
- ansible: ansible-playbook -i localhost, -c local playbook.yml
- puppet: puppet apply --modulepath ./modules manifests/site.pp
- salt: salt-call --local --file-root ./states state.apply
This renders a temporary Ansible manifest from the old harvest, then runs
the local apply command:
- ansible-playbook -i localhost, -c local playbook.yml
Returns a dict suitable for attaching to the diff report under
report['enforcement'].
"""
target = _normalise_enforcement_target(target)
tool_exe, tool_label = _require_enforcement_tool(target)
tool_exe = "ansible-playbook"
# Import lazily to avoid heavy import cost and potential CLI cycles.
from .manifest import manifest
@ -933,12 +854,10 @@ def enforce_old_harvest(
if report is not None:
plan = _enforcement_plan(report, old_state, old_b.dir)
roles = list(plan.get("roles") or [])
# Only Ansible has generated per-role tags that can safely narrow
# the apply scope. Puppet and Salt enforcement deliberately run the
# full generated local manifest/catalog for now.
if target == "ansible":
t = list(plan.get("tags") or [])
tags = t if t else None
# Ansible has generated per-role tags that can safely narrow the
# apply scope.
t = list(plan.get("tags") or [])
tags = t if t else None
with tempfile.TemporaryDirectory(prefix="enroll-enforce-") as td:
td_path = Path(td)
@ -951,11 +870,10 @@ def enforce_old_harvest(
# 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)
manifest(str(old_b.dir), str(manifest_dir))
# 2) Apply it locally.
cmd, env = _enforcement_command(
target,
tool_exe,
manifest_dir,
tags=tags,
@ -967,12 +885,12 @@ def enforce_old_harvest(
if _progress_enabled():
if tags:
sys.stderr.write(
f"Enforce: running {tool_label} (tags: {','.join(tags)})\n",
f"Enforce: running {tool_exe} tags: {','.join(tags)})\n",
)
else:
sys.stderr.write(f"Enforce: running {tool_label}\n")
sys.stderr.write(f"Enforce: running {tool_exe}\n")
sys.stderr.flush()
spinner = _Spinner(f" {tool_label}")
spinner = _Spinner(f" {tool_exe}")
spinner.start()
try:
@ -990,7 +908,7 @@ def enforce_old_harvest(
rc = p.returncode if p is not None else None
spinner.stop(
final_line=(
f"Enforce: {tool_label} finished in {elapsed:0.1f}s"
f"Enforce: {tool_exe} finished in {elapsed:0.1f}s"
+ (f" (rc={rc})" if rc is not None else "")
),
)
@ -999,22 +917,15 @@ def enforce_old_harvest(
info: Dict[str, Any] = {
"status": "applied" if p.returncode == 0 else "failed",
"target": target,
"tool": tool_label,
"executable": tool_exe,
"started_at": started_at,
"finished_at": finished_at,
"command": cmd,
"returncode": int(p.returncode),
}
# Keep the original Ansible-specific field for compatibility with
# existing consumers of the JSON report.
if target == "ansible":
info["ansible_playbook"] = tool_exe
elif target == "puppet":
info["puppet"] = tool_exe
elif target == "salt":
info["salt_call"] = tool_exe
# Keep the Ansible-specific field for compatibility with existing
# consumers of the JSON report.
info["ansible_playbook"] = tool_exe
info["roles"] = roles
info["tags"] = list(tags or [])
@ -1024,7 +935,7 @@ def enforce_old_harvest(
if p.returncode != 0:
err = (p.stderr or p.stdout or "").strip()
raise RuntimeError(
f"{tool_label} failed"
f"{tool_exe} failed"
+ (f" (rc={p.returncode})" if p.returncode is not None else "")
+ (f": {err}" if err else "")
)
@ -1069,9 +980,6 @@ def _report_text(report: Dict[str, Any]) -> str:
if enf:
lines.append("\nEnforcement")
status = str(enf.get("status") or "").strip().lower()
tool = str(enf.get("tool") or "ansible-playbook")
target = str(enf.get("target") or "ansible")
via = f"{tool} ({target})" if target and target not in tool else tool
if status == "applied":
extra = ""
tags = enf.get("tags") or []
@ -1081,7 +989,7 @@ def _report_text(report: Dict[str, Any]) -> str:
elif scope:
extra = f" ({scope})"
lines.append(
f" applied old harvest via {via} (rc={enf.get('returncode')})"
f" applied old harvest (rc={enf.get('returncode')})"
+ extra
+ (
f" (finished {enf.get('finished_at')})"
@ -1091,7 +999,7 @@ def _report_text(report: Dict[str, Any]) -> str:
)
elif status == "failed":
lines.append(
f" attempted enforcement but {via} failed (rc={enf.get('returncode')})"
f" attempted enforcement but failed (rc={enf.get('returncode')})"
)
elif status == "skipped":
r = enf.get("reason")
@ -1231,9 +1139,6 @@ def _report_markdown(report: Dict[str, Any]) -> str:
if enf:
out.append("\n## Enforcement\n")
status = str(enf.get("status") or "").strip().lower()
tool = str(enf.get("tool") or "ansible-playbook")
target = str(enf.get("target") or "ansible")
via = f"{tool} ({target})" if target and target not in tool else tool
if status == "applied":
extra = ""
tags = enf.get("tags") or []
@ -1243,7 +1148,7 @@ def _report_markdown(report: Dict[str, Any]) -> str:
elif scope:
extra = f" ({scope})"
out.append(
f"- ✅ Applied old harvest via {via}"
"- ✅ Applied old harvest"
+ extra
+ (
f" (rc={enf.get('returncode')})"
@ -1259,7 +1164,7 @@ def _report_markdown(report: Dict[str, Any]) -> str:
)
elif status == "failed":
out.append(
f"- ⚠️ Attempted enforcement but {via} failed"
"- ⚠️ Attempted enforcement but failed"
+ (
f" (rc={enf.get('returncode')})"
if enf.get("returncode") is not None

18
enroll/ignore.py
View file

@ -27,6 +27,9 @@ DEFAULT_DENY_GLOBS = [
"/etc/gshadow",
"/etc/*shadow",
"/etc/letsencrypt/*",
"/etc/ppp/chap-secrets",
"/etc/ppp/pap-secrets",
"/etc/ppp/*-secrets",
"/usr/local/etc/ssl/private/*",
"/usr/local/etc/ssh/ssh_host_*",
"/usr/local/etc/*shadow",
@ -91,7 +94,20 @@ SENSITIVE_CONTENT_PATTERNS = [
\s*[:=]
"""
),
re.compile(rb"(?i)\b(pass|passwd|token|secret|api[_-]?key)\b"),
re.compile(
rb"(?i)\b(pass|passwd|password|passphrase|token|secret|"
rb"credentials?|api[_-]?key)\b"
),
# Credentials embedded in connection-string URIs, e.g.
# postgres://user:pass@host, redis://:pass@host, amqp://u:p@host
# The keyword regex above keys on assignment-style names and misses these.
re.compile(rb"(?i)[a-z][a-z0-9+.-]*://[^/\s:@]*:[^/\s@]+@[^/\s]"),
# HTTP(S) Authorization / Proxy-Authorization header values carrying a
# bearer/basic/digest credential.
re.compile(
rb"(?im)^\s*(?:proxy-)?authorization\s*:\s*"
rb"(?:bearer|basic|token|digest)\s+\S"
),
]
COMMENT_PREFIXES = (b"#", b";", b"//")

69
enroll/jinjaturtle.py
View file

@ -82,7 +82,6 @@ _JINJA_FOR_RE = re.compile(
r"{%\s*for\s+([A-Za-z_][A-Za-z0-9_]*)\s+in\s+([A-Za-z_][A-Za-z0-9_]*)\b"
)
_JINJA_SPECIAL_VARS = {"loop", "true", "false", "none", "True", "False", "None"}
_ERB_INSTANCE_VAR_RE = re.compile(r"<%=?[^%]*@([A-Za-z_][A-Za-z0-9_]*)", re.S)
def _find_undeclared_jinja_vars(template_text: str) -> Set[str]:
@ -121,21 +120,6 @@ def missing_jinja_template_vars(
return {name for name in referenced if name not in context}
def missing_erb_template_vars(template_text: str, context: Dict[str, Any]) -> Set[str]:
"""Return ERB ``@param`` references absent from Puppet Hiera/class data."""
local_names: Set[str] = set()
for key in context:
text = str(key)
if "::" in text:
local_names.add(text.split("::", 1)[1])
else:
local_names.add(text)
referenced = set(_ERB_INSTANCE_VAR_RE.findall(template_text))
return {name for name in referenced if name not in local_names}
def jinjify_artifact(
bundle_dir: str | Path,
artifact_role: str,
@ -147,14 +131,8 @@ def jinjify_artifact(
jt_enabled: bool,
overwrite_templates: bool = True,
role_name: Optional[str] = None,
template_engine: str = "jinja2",
puppet_class: Optional[str] = None,
) -> Optional[JinjifiedArtifact]:
"""Best-effort conversion of one harvested artifact into a template.
Ansible/Salt use Jinja2 output. Puppet uses ERB output with Puppet Hiera
keys when a new enough JinjaTurtle is available.
"""
"""Best-effort conversion of one harvested artifact into a Jinja2 template."""
if not (jt_enabled and jt_exe and can_jinjify_path(dest_path)):
return None
@ -164,31 +142,20 @@ def jinjify_artifact(
return None
try:
run_kwargs: Dict[str, Any] = {
"role_name": role_name or artifact_role,
"force_format": infer_other_formats(dest_path),
}
# Keep the historical call shape for Ansible/Salt and for tests that
# monkeypatch run_jinjaturtle with the old signature. Puppet/ERB is
# the only path that needs the newer JinjaTurtle CLI switches.
if template_engine != "jinja2":
run_kwargs["template_engine"] = template_engine
if puppet_class:
run_kwargs["puppet_class"] = puppet_class
result = run_jinjaturtle(jt_exe, str(artifact_path), **run_kwargs)
result = run_jinjaturtle(
jt_exe,
str(artifact_path),
role_name=role_name or artifact_role,
force_format=infer_other_formats(dest_path),
)
except Exception:
return None # nosec - best-effort template generation
ext = "erb" if template_engine == "erb" else "j2"
template_rel = Path(src_rel).as_posix() + f".{ext}"
template_rel = Path(src_rel).as_posix() + ".j2"
template_dst = Path(template_root) / template_rel
context = yaml_load_mapping(result.vars_text)
missing = (
missing_erb_template_vars(result.template_text, context)
if template_engine == "erb"
else missing_jinja_template_vars(result.template_text, context)
)
missing = missing_jinja_template_vars(result.template_text, context)
if missing:
# If this role was generated into an existing output directory, avoid
# leaving an obsolete template behind after falling back to a raw copy.
@ -243,10 +210,7 @@ def jinjify_managed_files(
) -> Tuple[Set[str], str]:
"""Jinjify a list of managed files and return Ansible-style vars text.
The return shape intentionally matches the historical Ansible helper:
``(templated_src_rels, combined_vars_text)``. Salt uses
:func:`jinjify_artifact` directly because it stores variables as a context
map per managed file.
The return shape is ``(templated_src_rels, combined_vars_text)``.
"""
templated: Set[str] = set()
vars_map: Dict[str, Any] = {}
@ -340,8 +304,6 @@ def run_jinjaturtle(
*,
role_name: str,
force_format: Optional[str] = None,
template_engine: str = "jinja2",
puppet_class: Optional[str] = None,
) -> JinjifyResult:
"""
Run jinjaturtle against src_path and return (template, defaults-yaml).
@ -349,9 +311,6 @@ def run_jinjaturtle(
jinjaturtle CLI:
jinjaturtle <config> -r <role> [-f <format>] [-d <defaults-output>] [-t <template-output>]
Newer JinjaTurtle versions also support ``--template-engine erb`` and
``--puppet-class`` for Puppet/Hiera output.
"""
src = Path(src_path)
if not src.is_file():
@ -360,9 +319,7 @@ def run_jinjaturtle(
with tempfile.TemporaryDirectory(prefix="enroll-jt-") as td:
td_path = Path(td)
defaults_out = td_path / "defaults.yml"
template_out = td_path / (
"template.erb" if template_engine == "erb" else "template.j2"
)
template_out = td_path / "template.j2"
cmd = [
jt_exe,
@ -376,10 +333,6 @@ def run_jinjaturtle(
]
if force_format:
cmd.extend(["-f", force_format])
if template_engine != "jinja2":
cmd.extend(["--template-engine", template_engine])
if puppet_class:
cmd.extend(["--puppet-class", puppet_class])
p = subprocess.run(cmd, text=True, capture_output=True) # nosec
if p.returncode != 0:

68
enroll/manifest.py
View file

@ -8,8 +8,6 @@ from pathlib import Path
from typing import List, Optional
from .ansible import manifest_from_bundle_dir as manifest_ansible_from_bundle_dir
from .puppet import manifest_from_bundle_dir as manifest_puppet_from_bundle_dir
from .salt import manifest_from_bundle_dir as manifest_salt_from_bundle_dir
from .harvest_safety import ensure_safe_output_parent
from .manifest_safety import validate_site_fqdn
from .remote import _safe_extract_tar
@ -175,7 +173,6 @@ def manifest(
jinjaturtle: str = "auto", # auto|on|off
sops_fingerprints: Optional[List[str]] = None,
no_common_roles: bool = False,
target: str = "ansible",
) -> Optional[str]:
"""Render a configuration-management manifest from a harvest.
@ -193,9 +190,6 @@ def manifest(
- In SOPS mode: the path to the encrypted manifest bundle (.sops)
- In plain mode: None
"""
target = (target or "ansible").strip().lower()
if target not in {"ansible", "puppet", "salt"}:
raise ValueError(f"unsupported manifest target: {target!r}")
fqdn = validate_site_fqdn(fqdn)
sops_mode = bool(sops_fingerprints)
@ -216,30 +210,13 @@ def manifest(
)
if not sops_mode:
if target == "puppet":
manifest_puppet_from_bundle_dir(
resolved_bundle_dir,
out,
fqdn=fqdn,
no_common_roles=no_common_roles,
jinjaturtle=jinjaturtle,
)
elif target == "salt":
manifest_salt_from_bundle_dir(
resolved_bundle_dir,
out,
fqdn=fqdn,
no_common_roles=no_common_roles,
jinjaturtle=jinjaturtle,
)
else:
manifest_ansible_from_bundle_dir(
resolved_bundle_dir,
out,
fqdn=fqdn,
jinjaturtle=jinjaturtle,
no_common_roles=no_common_roles,
)
manifest_ansible_from_bundle_dir(
resolved_bundle_dir,
out,
fqdn=fqdn,
jinjaturtle=jinjaturtle,
no_common_roles=no_common_roles,
)
return None
# SOPS mode: generate into a secure temp dir, then tar+encrypt into a single file.
@ -248,30 +225,13 @@ def manifest(
td_out = tempfile.TemporaryDirectory(prefix="enroll-manifest-")
tmp_out = Path(td_out.name) / "out"
if target == "puppet":
manifest_puppet_from_bundle_dir(
resolved_bundle_dir,
str(tmp_out),
fqdn=fqdn,
no_common_roles=no_common_roles,
jinjaturtle=jinjaturtle,
)
elif target == "salt":
manifest_salt_from_bundle_dir(
resolved_bundle_dir,
str(tmp_out),
fqdn=fqdn,
no_common_roles=no_common_roles,
jinjaturtle=jinjaturtle,
)
else:
manifest_ansible_from_bundle_dir(
resolved_bundle_dir,
str(tmp_out),
fqdn=fqdn,
jinjaturtle=jinjaturtle,
no_common_roles=no_common_roles,
)
manifest_ansible_from_bundle_dir(
resolved_bundle_dir,
str(tmp_out),
fqdn=fqdn,
jinjaturtle=jinjaturtle,
no_common_roles=no_common_roles,
)
enc = _encrypt_manifest_out_dir_to_sops(
tmp_out, out_file, list(sops_fingerprints or [])

1840
enroll/puppet.py
View file

File diff suppressed because it is too large Load diff

182
enroll/render_safety.py
View file

@ -1,7 +1,5 @@
from __future__ import annotations
import json
import re
from collections.abc import Mapping, Set as AbstractSet
from typing import Any
@ -50,183 +48,3 @@ def ansible_unsafe_data(value: Any) -> Any:
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

1759
enroll/salt.py
View file

File diff suppressed because it is too large Load diff