"""Discover, load and validate runbook packs (spec runbook-pack/v0.1).""" from __future__ import annotations import json import re from dataclasses import dataclass, field from pathlib import Path import yaml from jsonschema import Draft202012Validator SCHEMA_PATH = Path(__file__).resolve().parent.parent / "schema" / "runbook-pack.schema.json" PLACEHOLDER = re.compile(r"\{\{\s*([a-z][a-z0-9_]*)\s*\}\}") SECRET_RE = re.compile( r"(hvs\.[A-Za-z0-9]{8,}|s\.[A-Za-z0-9]{24}|-----BEGIN [A-Z ]*PRIVATE KEY|otpauth://)") RISKY = {"changes-state", "attended", "destructive"} @dataclass class Pack: path: Path data: dict errors: list[str] = field(default_factory=list) @property def id(self) -> str: return str(self.data.get("id", self.path.parent.name)) @property def ok(self) -> bool: return not self.errors def discover(root: Path) -> list[Path]: """Find runbooks/*/pack.yaml under root (root itself may be a runbooks dir).""" root = Path(root) found = sorted(root.glob("runbooks/*/pack.yaml")) + sorted(root.glob("*/pack.yaml")) return list(dict.fromkeys(found)) def _schema_errors(data: dict) -> list[str]: schema = json.loads(SCHEMA_PATH.read_text()) v = Draft202012Validator(schema) return [f"{'/'.join(map(str, e.absolute_path)) or ''}: {e.message}" for e in sorted(v.iter_errors(data), key=lambda e: list(map(str, e.absolute_path)))] def _semantic_errors(data: dict) -> list[str]: errs: list[str] = [] params = {p["id"]: p for p in data.get("parameters", []) if isinstance(p, dict) and "id" in p} seen = [p["id"] for p in data.get("parameters", []) if isinstance(p, dict) and "id" in p] if len(seen) != len(set(seen)): errs.append("duplicate parameter id") for pid, p in params.items(): if p.get("type") == "enum" and not p.get("values"): errs.append(f"parameter {pid}: enum without values") d = p.get("default") if d is not None: if p.get("values") and d not in p["values"]: errs.append(f"parameter {pid}: default not in values") if p.get("pattern") and isinstance(d, str) and not re.fullmatch(p["pattern"], d): errs.append(f"parameter {pid}: default violates pattern") ids = [s["id"] for s in data.get("steps", []) if isinstance(s, dict) and "id" in s] if len(ids) != len(set(ids)): errs.append("duplicate step id") used: set[str] = set() for s in data.get("steps", []): if not isinstance(s, dict): continue texts = [s.get("command", ""), s.get("description", ""), (s.get("verify") or {}).get("command", "")] for t in texts: for name in PLACEHOLDER.findall(t or ""): used.add(name) if name not in params: errs.append(f"step {s.get('id')}: undefined placeholder {{{{{name}}}}}") if s.get("risk") in RISKY and not s.get("rollback"): errs.append(f"step {s.get('id')}: risk {s['risk']} requires rollback") v = s.get("verify") or {} if v.get("expect") in ("exit-0", "output-contains") and not v.get("command"): errs.append(f"step {s.get('id')}: verify expect {v['expect']} needs a command") if v.get("expect") == "output-contains" and not v.get("contains"): errs.append(f"step {s.get('id')}: verify output-contains needs 'contains'") for name in params: if name not in used and data.get("engine") == "native": errs.append(f"parameter {name}: declared but never used") if SECRET_RE.search(json.dumps(data)): errs.append("contains secret-looking marker") return errs def load(path: Path) -> Pack: path = Path(path) try: data = yaml.safe_load(path.read_text()) except yaml.YAMLError as e: return Pack(path, {}, [f"invalid YAML: {e}"]) if not isinstance(data, dict): return Pack(path, {}, ["manifest must be a mapping"]) errs = _schema_errors(data) if not errs: errs = _semantic_errors(data) return Pack(path, data, errs) def render(command: str, values: dict[str, str], pack: Pack) -> str: """Substitute parameters into a command template. Unknown or unset -> error.""" params = {p["id"]: p for p in pack.data.get("parameters", [])} def sub(m: re.Match) -> str: name = m.group(1) if name not in params: raise KeyError(f"undefined parameter {name}") val = values.get(name, params[name].get("default")) if val is None: raise KeyError(f"parameter {name} has no value") val = str(val) p = params[name] if p.get("values") and val not in p["values"]: raise ValueError(f"parameter {name}: {val!r} not in {p['values']}") if p.get("pattern") and not re.fullmatch(p["pattern"], val): raise ValueError(f"parameter {name}: {val!r} violates pattern") if re.search(r"[;&|`$<>\n\\'\"\s]", val) and p.get("type") != "path": raise ValueError(f"parameter {name}: unsafe characters") return val return PLACEHOLDER.sub(sub, command)