runbook-tutorials/rtut/pack.py

"""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 '<root>'}: {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)