From 72eec40ef32433da214da36ca144379043cfa21f Mon Sep 17 00:00:00 2001 From: Vladimir Eremin Date: Sat, 4 Jul 2026 21:23:48 +0300 Subject: [PATCH] feat: port engine lib modules (log, sys, tpl, host) --- engine/lib/host.py | 140 +++++++++++++++++++++++++++++++++++++++++++++ engine/lib/log.py | 19 ++++++ engine/lib/sys.py | 94 ++++++++++++++++++++++++++++++ engine/lib/tpl.py | 37 ++++++++++++ tests/test_host.py | 39 +++++++++++++ tests/test_tpl.py | 33 +++++++++++ 6 files changed, 362 insertions(+) create mode 100644 engine/lib/host.py create mode 100644 engine/lib/log.py create mode 100644 engine/lib/sys.py create mode 100644 engine/lib/tpl.py create mode 100644 tests/test_host.py create mode 100644 tests/test_tpl.py diff --git a/engine/lib/host.py b/engine/lib/host.py new file mode 100644 index 0000000..1ac5d37 --- /dev/null +++ b/engine/lib/host.py @@ -0,0 +1,140 @@ +# engine/lib/host.py +"""ctx.host — idempotent host mutation: backup, write_file, inject_block, install_crontab. + +Ports root_template_crontab (install + rollback) and the ufw block-injection logic +(root_template.sh _root_template_inject_rules_to_ufw_file), generalized + self-healing. +""" +from __future__ import annotations +import os +import shutil +import subprocess +import tempfile +import time +from typing import List, Optional, Sequence, Tuple + + +def apply_block(text: str, content: str, start: str, end: str, + anchor: Optional[str] = None, + old_pairs: Sequence[Tuple[str, str]] = ()) -> Tuple[str, bool]: + """Pure transform: remove existing (start,end) and old_pairs blocks, insert a fresh + block after the first anchor line (or append if anchor is None). If an anchor is + given but not present, return (text, False) unchanged (matches bash skip behavior).""" + lines = text.splitlines() + if anchor is not None and not any(anchor in ln for ln in lines): + return text, False + + pairs = [(start, end), *old_pairs] + stripped_starts = {s: e for s, e in pairs} + out: List[str] = [] + skip_until: Optional[str] = None + for ln in lines: + if skip_until is None: + s = ln.strip() + if s in stripped_starts: + skip_until = stripped_starts[s] + continue + out.append(ln) + else: + if ln.strip() == skip_until: + skip_until = None + continue + + block = [start, *content.splitlines(), end] + if anchor is None: + out.extend(block) + return "\n".join(out) + "\n", True + + result: List[str] = [] + inserted = False + for ln in out: + result.append(ln) + if not inserted and anchor in ln: + result.extend(block) + inserted = True + return "\n".join(result) + "\n", True + + +class Host: + def __init__(self, log, sys_): + self.log = log + self.sys = sys_ + + # --- read/write with sudo awareness --- + def _read(self, path: str, sudo: bool = False) -> str: + if sudo: + cp = self.sys.run(["sudo", "cat", path], check=True, capture=True) + return cp.stdout or "" + with open(path) as f: + return f.read() + + def backup(self, path: str, sudo: bool = False) -> Optional[str]: + exists = (self.sys.ok(["sudo", "test", "-f", path]) if sudo + else os.path.isfile(path)) + if not exists: + return None + bkp = f"{path}.{int(time.time())}.bak" + if sudo: + self.sys.sudo(["cp", "-a", path, bkp]) + else: + shutil.copy2(path, bkp) + self.log.info(f"Backed up {path} -> {bkp}") + return bkp + + def write_file(self, path: str, content: str, backup: bool = True, + dry_run: bool = False, sudo: bool = False) -> None: + if dry_run: + self.log.info(f"[DRY-RUN] would write {len(content)} bytes to {path}") + return + if backup: + self.backup(path, sudo=sudo) + tmp = tempfile.NamedTemporaryFile("w", delete=False) + tmp.write(content) + tmp.close() + if sudo: + self.sys.sudo(["mkdir", "-p", os.path.dirname(path)]) + self.sys.sudo(["cp", tmp.name, path]) + os.unlink(tmp.name) + else: + os.makedirs(os.path.dirname(path), exist_ok=True) + shutil.move(tmp.name, path) + self.log.success(f"Wrote {path}") + + def inject_block(self, target: str, content: str, start: str, end: str, + anchor: Optional[str] = None, + old_pairs: Sequence[Tuple[str, str]] = (), + dry_run: bool = False, sudo: bool = True) -> bool: + exists = (self.sys.ok(["sudo", "test", "-f", target]) if sudo + else os.path.isfile(target)) + if not exists: + self.log.error(f"Target file does not exist: {target}") + return False + text = self._read(target, sudo=sudo) + new_text, injected = apply_block(text, content, start, end, anchor, old_pairs) + if not injected: + self.log.warn(f"Anchor {anchor!r} not found in {target}; skipping.") + return False + if dry_run: + self.log.info(f"[DRY-RUN] would inject block into {target}") + return True + self.write_file(target, new_text, backup=True, sudo=sudo) + self.log.success(f"Block injected into {target}") + return True + + def install_crontab(self, text: str, dry_run: bool = False) -> None: + """Port of root_template_crontab: backup, install via `crontab -`, rollback.""" + backup = subprocess.run(["crontab", "-l"], text=True, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) + prev = backup.stdout if backup.returncode == 0 else "" + if not text.strip(): + self.log.error("Rendered crontab is empty; aborting (previous kept).") + raise SystemExit(65) # EX_DATAERR + self.log.info("Rendered crontab preview:\n" + text) + if dry_run: + self.log.info("[DRY-RUN] would install the crontab above.") + return + res = subprocess.run(["crontab", "-"], input=text, text=True) + if res.returncode != 0: + self.log.error("crontab install failed; restoring previous.") + subprocess.run(["crontab", "-"], input=prev, text=True) + raise SystemExit(1) + self.log.success("Crontab installed.") diff --git a/engine/lib/log.py b/engine/lib/log.py new file mode 100644 index 0000000..9367446 --- /dev/null +++ b/engine/lib/log.py @@ -0,0 +1,19 @@ +# engine/lib/log.py +"""ctx.log — [node] [action] [level] message (port of root_log_*).""" +from __future__ import annotations +import sys + + +class Log: + def __init__(self, node: str = "global", action: str = "core"): + self.node = node + self.action = action + + def _emit(self, level: str, msg: str, stream=None) -> None: + print(f"[{self.node}] [{self.action}] [{level}] {msg}", + file=stream or sys.stdout, flush=True) + + def info(self, msg): self._emit("INFO", msg) + def warn(self, msg): self._emit("WARN", msg) + def error(self, msg): self._emit("ERROR", msg, sys.stderr) + def success(self, msg): self._emit("SUCCESS", msg) diff --git a/engine/lib/sys.py b/engine/lib/sys.py new file mode 100644 index 0000000..4d5d11f --- /dev/null +++ b/engine/lib/sys.py @@ -0,0 +1,94 @@ +# engine/lib/sys.py +"""ctx.sys — command execution + systemd service management (port of root_sys.sh).""" +from __future__ import annotations +import shutil +import subprocess +import time +from typing import List, Optional, Sequence + + +class Sys: + def __init__(self, log): + self.log = log + + # --- command execution --- + def run(self, cmd: Sequence[str], check: bool = True, capture: bool = False, + cwd: Optional[str] = None, env: Optional[dict] = None, + input: Optional[str] = None) -> subprocess.CompletedProcess: + return subprocess.run(list(cmd), check=check, cwd=cwd, env=env, text=True, + input=input, + stdout=subprocess.PIPE if capture else None, + stderr=subprocess.PIPE if capture else None) + + def sudo(self, cmd: Sequence[str], **kw) -> subprocess.CompletedProcess: + return self.run(["sudo", *cmd], **kw) + + def ok(self, cmd: Sequence[str], **kw) -> bool: + """Run, return True on exit 0, never raising.""" + return self.run(cmd, check=False, **kw).returncode == 0 + + # --- systemd (port of root_sys_*) --- + def service_exists(self, name: str) -> bool: + cp = self.run(["systemctl", "list-unit-files", "--type=service"], + check=False, capture=True) + return f"{name}.service" in (cp.stdout or "") + + def service_is_active(self, name: str) -> bool: + return self.ok(["systemctl", "is-active", "--quiet", name]) + + def service_is_enabled(self, name: str) -> bool: + return self.ok(["systemctl", "is-enabled", "--quiet", name]) + + def start_and_enable(self, name: str) -> bool: + if not self.service_exists(name): + self.log.warn(f"Service {name} does not exist on this system") + return False + if not self.service_is_active(name): + self.log.info(f"Starting {name} service...") + self.sudo(["systemctl", "start", name]) + if not self.service_is_enabled(name): + self.log.info(f"Enabling {name} service...") + self.sudo(["systemctl", "enable", name]) + return self.service_is_active(name) + + def ensure_running(self, name: str) -> None: + """Port of root_init_ensure_service_running: enable --now if not active.""" + if not self.service_is_active(name): + self.log.info(f"Starting {name} service...") + self.sudo(["systemctl", "enable", "--now", name]) + else: + self.log.info(f"{name} service already running.") + + def restart(self, name: str) -> bool: + self.log.info(f"Restarting {name} service...") + self.sudo(["systemctl", "restart", name]) + return self.service_is_active(name) + + def reload(self, name: str, dry_run: bool = False) -> None: + if dry_run: + self.log.info(f"[DRY-RUN] systemctl reload {name}") + return + self.log.info(f"Reloading {name} configuration...") + self.sudo(["systemctl", "reload", name]) + + def daemon_reload(self) -> None: + self.sudo(["systemctl", "daemon-reload"]) + + def wait_for(self, name: str, timeout: int = 30, interval: int = 2) -> bool: + elapsed = 0 + while elapsed < timeout: + if self.service_is_active(name): + return True + time.sleep(interval) + elapsed += interval + return False + + def install_packages(self, packages: List[str]) -> None: + """Port of root_init_install_packages (install if the binary is absent).""" + for pkg in packages: + if shutil.which(pkg): + self.log.info(f"{pkg} already installed.") + continue + self.log.info(f"Installing {pkg}...") + self.sudo(["apt-get", "update", "-y"]) + self.sudo(["apt-get", "install", "-y", pkg]) diff --git a/engine/lib/tpl.py b/engine/lib/tpl.py new file mode 100644 index 0000000..1e893d4 --- /dev/null +++ b/engine/lib/tpl.py @@ -0,0 +1,37 @@ +# engine/lib/tpl.py +"""ctx.tpl — stdlib string.Template rendering (envsubst successor, SP2 decision #3). + +safe_substitute matches envsubst's forgiving behavior: known $VAR/${VAR} replaced, +unknown left literal, and 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/. 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 diff --git a/tests/test_host.py b/tests/test_host.py new file mode 100644 index 0000000..c57e625 --- /dev/null +++ b/tests/test_host.py @@ -0,0 +1,39 @@ +# tests/test_host.py +import unittest +from engine.lib.host import apply_block + +class TestApplyBlock(unittest.TestCase): + def test_insert_after_anchor(self): + text = "line1\n# End required lines\nline2\n" + out, injected = apply_block(text, "ruleA\nruleB", + "# START APEX.ERMNVLDMR.COM RULES", + "# END APEX.ERMNVLDMR.COM RULES", + anchor="# End required lines") + self.assertTrue(injected) + self.assertIn("# End required lines\n# START APEX.ERMNVLDMR.COM RULES\nruleA\nruleB\n# END APEX.ERMNVLDMR.COM RULES", out) + + def test_self_heal_removes_old_and_new_blocks(self): + text = ("a\n# End required lines\n" + "# START ROOT.ERMNVLDMR.COM RULES\nold\n# END ROOT.ERMNVLDMR.COM RULES\n" + "# START APEX.ERMNVLDMR.COM RULES\nprev\n# END APEX.ERMNVLDMR.COM RULES\n" + "b\n") + out, injected = apply_block(text, "new", + "# START APEX.ERMNVLDMR.COM RULES", + "# END APEX.ERMNVLDMR.COM RULES", + anchor="# End required lines", + old_pairs=[("# START ROOT.ERMNVLDMR.COM RULES", + "# END ROOT.ERMNVLDMR.COM RULES")]) + self.assertTrue(injected) + self.assertNotIn("old", out) + self.assertNotIn("prev", out) + self.assertEqual(out.count("# START APEX.ERMNVLDMR.COM RULES"), 1) + self.assertIn("new", out) + + def test_anchor_missing_does_not_inject(self): + text = "a\nb\n" + out, injected = apply_block(text, "x", "S", "E", anchor="# End required lines") + self.assertFalse(injected) + self.assertEqual(out, text) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tpl.py b/tests/test_tpl.py new file mode 100644 index 0000000..bfa4876 --- /dev/null +++ b/tests/test_tpl.py @@ -0,0 +1,33 @@ +# tests/test_tpl.py +import os, tempfile, unittest +from engine.lib.tpl import Tpl +from engine.lib.log import Log + +class TestTpl(unittest.TestCase): + def setUp(self): + self.cfg = tempfile.mkdtemp() + self.tpl = Tpl(self.cfg, Log()) + + def _w(self, rel, body): + p = os.path.join(self.cfg, rel) + os.makedirs(os.path.dirname(p), exist_ok=True) + open(p, "w").write(body) + + def test_render_substitutes_known_vars(self): + self._w("cron/crontab", "root=${APEX_TIER1} node=$APEX_NODE\n") + out = self.tpl.render("cron/crontab", {"APEX_TIER1": "/mnt/t1", "APEX_NODE": "daedalus"}) + self.assertEqual(out, "root=/mnt/t1 node=daedalus\n") + + def test_render_leaves_shell_command_substitution(self): + self._w("c", "x $(cat /f) ${APEX_NODE}\n") + out = self.tpl.render("c", {"APEX_NODE": "d"}) + self.assertEqual(out, "x $(cat /f) d\n") # $(...) survives; ${APEX_NODE} replaced + + def test_render_dir_returns_relpath_map(self): + self._w("fail2ban/jail.local", "a=${X}\n") + self._w("fail2ban/sub/extra.conf", "b=${Y}\n") + out = self.tpl.render_dir("fail2ban", {"X": "1", "Y": "2"}) + self.assertEqual(out, {"jail.local": "a=1\n", "sub/extra.conf": "b=2\n"}) + +if __name__ == "__main__": + unittest.main()