"""Validation for runbook-pack/v0.2: JSON Schema plus semantic checks.""" from __future__ import annotations import json import re from pathlib import Path from jsonschema import Draft202012Validator from .pack import PLACEHOLDER, SECRET_RE SCHEMA_PATH = Path(__file__).resolve().parent.parent / "schema" / "runbook-pack-v0.2.schema.json" MAX_DEPTH = 3 def schema_errors(data: dict) -> list[str]: v = Draft202012Validator(json.loads(SCHEMA_PATH.read_text())) 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)))] class _Model: """Flattened view of a v0.2 pack used by the semantic checks.""" def __init__(self, data: dict): self.data = data self.errs: list[str] = [] self.flow: dict[str, dict] = {} # stages, steps, gates share one id space self.kind: dict[str, str] = {} self.parent: dict[str, str | None] = {} self.children: dict[str, list[str]] = {} self.alts: dict[str, str] = {} # alternative id -> parent step id self.depth_ok = True def add(self, id_: str, kind: str, node: dict, parent: str | None) -> None: if id_ in self.flow or id_ in self.alts: self.errs.append(f"duplicate id: {id_}") return # keep the first definition; never overwrite the tree self.flow[id_] = node self.kind[id_] = kind self.parent[id_] = parent if parent: self.children.setdefault(parent, []).append(id_) def walk(self, stages: list[dict], parent: str | None, depth: int) -> None: for st in stages: if depth > MAX_DEPTH: self.errs.append(f"stage {st['id']}: nesting deeper than {MAX_DEPTH} levels") self.add(st["id"], "stage", st, parent) for s in st.get("steps", []): self.add(s["id"], "step", s, st["id"]) for a in s.get("one_of", []): if a["id"] in self.flow or a["id"] in self.alts: self.errs.append(f"duplicate id: {a['id']}") self.alts[a["id"]] = s["id"] for g in st.get("gates", []): self.add(g["id"], "gate", g, st["id"]) self.walk(st.get("stages", []), st["id"], depth + 1) def resolve(self, ref: str) -> str | None: """requires reference -> flow id, or None if unresolved.""" if ref.startswith("gate:"): g = ref[5:] return g if self.kind.get(g) == "gate" else None return ref if self.kind.get(ref) in ("stage", "step") else None def _check_refs(m: _Model, data: dict) -> None: actors = {a["id"] for a in data.get("actors", [])} dps = {d["id"]: d for d in data.get("datapoints", [])} arts = {a["id"] for a in data.get("artefacts", [])} steps = {i for i, k in m.kind.items() if k == "step"} | set(m.alts) gates = {i for i, k in m.kind.items() if k == "gate"} stages = {i for i, k in m.kind.items() if k == "stage"} for label, ids in (("actor", [a["id"] for a in data.get("actors", [])]), ("datapoint", list(dps)), ("artefact", [a["id"] for a in data.get("artefacts", [])])): if len(ids) != len(set(ids)): m.errs.append(f"duplicate {label} id") def check_ck(ck: dict, where: str, uses: set[str]) -> None: exp = ck.get("expect", "manual") if exp in ("exit-0", "output-contains") and not ck.get("command"): m.errs.append(f"{where}: verify expect {exp} needs a command") if exp == "output-contains" and not ck.get("contains"): m.errs.append(f"{where}: verify output-contains needs 'contains'") for d in ck.get("datapoints_ok", []): if d not in dps: m.errs.append(f"{where}: datapoints_ok names unknown datapoint {d}") for t in (ck.get("command", ""), ck.get("contains", "")): for name in PLACEHOLDER.findall(t or ""): if name not in dps: m.errs.append(f"{where}: undefined placeholder {{{{{name}}}}}") elif uses and name not in uses: m.errs.append(f"{where}: placeholder {name} not listed in uses") def check_text(text: str, where: str, uses: set[str]) -> None: for name in PLACEHOLDER.findall(text or ""): if name not in dps: m.errs.append(f"{where}: undefined placeholder {{{{{name}}}}}") elif name not in uses: m.errs.append(f"{where}: placeholder {name} not listed in uses") for id_, node in m.flow.items(): kind = m.kind[id_] if kind == "step": if node["actor"] not in actors: m.errs.append(f"step {id_}: unknown actor {node['actor']}") uses = set(node.get("uses", [])) for u in uses: if u not in dps: m.errs.append(f"step {id_}: uses unknown datapoint {u}") for p in node.get("produces", []): if p not in dps and p not in arts: m.errs.append(f"step {id_}: produces unknown datapoint or artefact {p}") parts = [(node, id_)] + [(a, a["id"]) for a in node.get("one_of", [])] for part, pid in parts: check_text(part.get("command", ""), f"step {pid}", uses) check_text(part.get("rollback", ""), f"step {pid}", uses) if "verify" in part: check_ck(part["verify"], f"step {pid}", uses) if part.get("reversible", node.get("reversible", True)) is False: if not (node.get("confirm_phrase") or data.get("confirm_phrase")): m.errs.append(f"step {pid}: irreversible step needs a confirm_phrase (step or pack)") for k, vals in (part.get("recommended_when") or {}).items(): if k not in dps: m.errs.append(f"step {pid}: recommended_when names unknown datapoint {k}") elif dps[k].get("values"): for v in vals: if v not in dps[k]["values"]: m.errs.append(f"step {pid}: recommended_when {k}={v} not in values") if node.get("one_of") and node.get("command"): m.errs.append(f"step {id_}: a step with one_of must not also have its own command") elif kind == "gate": by = (node.get("waivable") or {}).get("by") if by and by not in actors: m.errs.append(f"gate {id_}: waivable.by names unknown actor {by}") for ck in node["checks"]: check_ck(ck, f"gate {id_}", set()) for d in dps.values(): if d["type"] == "enum" and not d.get("values"): m.errs.append(f"datapoint {d['id']}: enum without values") dv = d.get("default") if dv is not None: if d.get("values") and dv not in d["values"]: m.errs.append(f"datapoint {d['id']}: default not in values") if d.get("pattern") and isinstance(dv, str) and not re.fullmatch(d["pattern"], dv): m.errs.append(f"datapoint {d['id']}: default violates pattern") if d.get("ask_in") and d["ask_in"] not in stages: m.errs.append(f"datapoint {d['id']}: ask_in names unknown stage {d['ask_in']}") if d.get("produced_by") and d["produced_by"] not in steps: m.errs.append(f"datapoint {d['id']}: produced_by names unknown step {d['produced_by']}") if d.get("confirmed_by") and d["confirmed_by"] not in steps | gates: m.errs.append(f"datapoint {d['id']}: confirmed_by names unknown step or gate") for g_id in gates: # a gate that requires a datapoint to be ok cannot be what confirms it for ck in m.flow[g_id]["checks"]: for d in ck.get("datapoints_ok", []): if dps.get(d, {}).get("confirmed_by") == g_id: m.errs.append(f"gate {g_id}: checks datapoint {d} that it is itself confirmed_by (circular)") for a in data.get("artefacts", []): if a["location"]["datapoint"] not in dps: m.errs.append(f"artefact {a['id']}: location names unknown datapoint") if a["produced_by"] not in steps: m.errs.append(f"artefact {a['id']}: produced_by names unknown step") if a.get("lives_until") and a["lives_until"] not in steps: m.errs.append(f"artefact {a['id']}: lives_until names unknown step") def _check_graph(m: _Model) -> None: deps: dict[str, set[str]] = {i: set() for i in m.flow} for id_, node in m.flow.items(): for ref in node.get("requires", []): r = m.resolve(ref) if r is None: m.errs.append(f"{m.kind[id_]} {id_}: requires unknown item {ref}") else: deps[id_].add(r) # a node also waits for whatever its enclosing stages wait for p, seen = m.parent[id_], set() while p and p not in seen: seen.add(p) for ref in m.flow[p].get("requires", []): r = m.resolve(ref) if r: deps[id_].add(r) p = m.parent[p] # a stage is finished when its contents are for c in m.children.get(id_, []): deps[id_].add(c) state: dict[str, int] = {} def visit(n: str, path: list[str]) -> None: if state.get(n) == 2: return if state.get(n) == 1: m.errs.append("requires cycle: " + " -> ".join(path[path.index(n):] + [n])) return state[n] = 1 for d in sorted(deps[n]): visit(d, path + [n]) state[n] = 2 for n in sorted(deps): visit(n, []) def validate(data: dict) -> list[str]: errs = schema_errors(data) if errs: return errs if data["engine"] == "legacy-command": return [] m = _Model(data) m.walk(data["stages"], None, 1) _check_refs(m, data) _check_graph(m) if SECRET_RE.search(json.dumps(data)): m.errs.append("contains secret-looking marker") return sorted(set(m.errs))