Adversarial parity review of every ported action against its bash source confirmed 17 real behavioral drifts (of 22 claimed); all fixed here: - tiers/chown: defaults were compositions/apex/tierN (one level too deep); now the tier roots. shared/ base reset is non-recursive again. - compose: --dry-run was swallowed by argparse.REMAINDER; now stripped. - host.write_file: rendered files land 0644 (tempfile's 0600 made fresh systemd units/crowdsec configs owner-only). - systemd.deploy_units: top-level non-hidden files only. - ufw: missing docker/host config dir exits 66; missing /etc/ufw target aborts; empty rules dir warns; dotfile .rules excluded. - sync/packages: Telegram error-trap parity and failed notify exits 1. Same notify-failure exit for backup/run and sync/repository. repository regains the node-dir guard. - tiers/useradd: no more secret-bearing .env.bak per run. - tiers/link: EOF on confirm prompt declines and continues. - extract-traefik-certs: sudo mkdir inside the notify envelope. - generate-happ-subscriptions: realpath parent-exists guards + docker check. - configure/crowdsec: bouncer install goes through install_packages. - lint-docker-compose: malformed/unreadable anchor-IP inputs are lint errors. - core compose: APEX_RESTIC_REPOSITORY now ?error; RESTIC_COMPRESSION per-node via APEX_RESTIC_COMPRESSION (default auto). - sys.service_exists line-anchored; tpl docstring corrected.
39 lines
1.5 KiB
Python
39 lines
1.5 KiB
Python
# engine/lib/tpl.py
|
|
"""ctx.tpl — stdlib string.Template rendering (envsubst successor, SP2 decision #3).
|
|
|
|
safe_substitute replaces known $VAR/${VAR} and leaves unknown ones literal (deliberate
|
|
delta: envsubst emptied them — literal survivors surface typos instead of silently
|
|
blanking). Shell $(...) command-substitution survives untouched.
|
|
Loops/conditionals live in the python action, never in the template file.
|
|
"""
|
|
from __future__ import annotations
|
|
import os
|
|
from string import Template
|
|
from typing import Dict
|
|
|
|
|
|
class Tpl:
|
|
def __init__(self, configs_dir: str, log):
|
|
self.configs_dir = configs_dir
|
|
self.log = log
|
|
|
|
def _resolve(self, path: str) -> str:
|
|
return path if os.path.isabs(path) else os.path.join(self.configs_dir, path)
|
|
|
|
def render(self, path: str, vars: Dict[str, str]) -> str:
|
|
full = self._resolve(path)
|
|
with open(full) as f:
|
|
return Template(f.read()).safe_substitute(vars)
|
|
|
|
def render_dir(self, rel_dir: str, vars: Dict[str, str]) -> Dict[str, str]:
|
|
"""Render every file under configs/<rel_dir>. Returns {relpath: content}."""
|
|
base = self._resolve(rel_dir)
|
|
out: Dict[str, str] = {}
|
|
for dirpath, _dirs, files in os.walk(base):
|
|
for fn in sorted(files):
|
|
full = os.path.join(dirpath, fn)
|
|
rel = os.path.relpath(full, base).replace(os.sep, "/")
|
|
with open(full) as f:
|
|
out[rel] = Template(f.read()).safe_substitute(vars)
|
|
return out
|