tegwick
e021fd979f
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
221 lines
10 KiB
Python
221 lines
10 KiB
Python
"""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 / "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 '<root>'}: {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")
|
|
for part, pid in parts:
|
|
if part.get("idempotent", node.get("idempotent")) == "conditional" and not (part.get("idempotent_note") or node.get("idempotent_note")):
|
|
m.errs.append(f"step {pid}: idempotent: conditional needs an idempotent_note saying when")
|
|
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))
|