minicheck-0.4.0-py3-none-any.whl
protocol_bench-1.1.0-py3-none-any.whl
from __future__ import annotations
import json
import gradio as gr
from protocol_bench import BASELINES, load_tasks, score, validate_trace
from minicheck import RenderTooLarge, SpecError, check_liveness, check_safety, protocol_from_spec, to_mermaid
TASKS = {t.id: t for t in load_tasks()}
EXAMPLE_BROKEN = {
"name": "mutex",
"fields": ["a", "b", "lock"],
"initial": {"a": 0, "b": 0, "lock": 0},
"transitions": [
{"label": "a_enter", "when": {"a": 0}, "set": {"a": 1, "lock": 1}},
{"label": "b_enter", "when": {"b": 0}, "set": {"b": 1, "lock": 1}},
{"label": "a_exit", "when": {"a": 1}, "set": {"a": 0, "lock": 0}},
{"label": "b_exit", "when": {"b": 1}, "set": {"b": 0, "lock": 0}},
],
"invariants": {"mutual_exclusion": {"forbid": {"a": 1, "b": 1}}},
}
EXAMPLE_FIXED = {
**EXAMPLE_BROKEN,
"transitions": [
{"label": "a_enter", "when": {"a": 0, "lock": 0}, "set": {"a": 1, "lock": 1}},
{"label": "b_enter", "when": {"b": 0, "lock": 0}, "set": {"b": 1, "lock": 1}},
{"label": "a_exit", "when": {"a": 1}, "set": {"a": 0, "lock": 0}},
{"label": "b_exit", "when": {"b": 1}, "set": {"b": 0, "lock": 0}},
],
}
EXAMPLE_RETRY = {
"name": "retry_loop",
"fields": ["tries", "done"],
"initial": {"tries": 0, "done": 0},
"transitions": [
{"label": "attempt", "when": {"done": 0}, "set": {"tries": {"incr": 1}}},
{"label": "succeed", "when": {"done": 0}, "set": {"done": 1}},
],
"invariants": {"at_most_3_retries": {"forbid": {"tries": 4}}},
"goal": {"require": {"done": 1}},
}
EXAMPLES = {
"Mutual exclusion — BROKEN (no lock guard)": EXAMPLE_BROKEN,
"Mutual exclusion — FIXED": EXAMPLE_FIXED,
"Retry loop — unbounded retries": EXAMPLE_RETRY,
}
def _fmt_state(state: dict) -> str:
return ", ".join(f"{k}={v}" for k, v in state.items())
def check_spec(spec_text: str) -> str:
"""Check a user-supplied declarative spec. Returns markdown."""
try:
spec = json.loads(spec_text)
except json.JSONDecodeError as e:
return f"### ⚠️ Not valid JSON\n\n```\n{e}\n```"
try:
model = protocol_from_spec(spec)
except SpecError as e:
return f"### ⚠️ Spec rejected\n\n```\n{e}\n```\n\nSee **Spec format** below."
if not model.invariants:
return "### ⚠️ No invariants declared\n\nAdd an `invariants` block to check something."
res = check_safety(model)
out = [f"**Explored {res['reachable_states']} reachable states.**\n"]
if not res["exhaustive"]:
# Say this before any verdict, because it changes how every one of them reads.
out.append(
"> ⚠️ **The search did not finish**, so it did not cover the whole state space. "
"Anything not refuted below is **undetermined**, not proved.\n"
)
out.append(f"> _{res.get('incomplete_reason', 'the sweep stopped early')}_\n")
for name, r in res["properties"].items():
if r["holds"] is True:
out.append(f"### ✅ `{name}` HOLDS\n")
out.append("No reachable state violates it, and every reachable state was checked.\n")
elif r["holds"] is None:
# Three-valued: this is NOT a pass, and must not be rendered as one.
out.append(f"### ❓ `{name}` UNDETERMINED\n")
out.append(
"No violation was found, but the search did not complete — so this is not a proof. "
"Bound the growing field with a `when` guard, then try again.\n"
)
else:
cex = r["counterexample"]
out.append(f"### ❌ `{name}` VIOLATED in {len(cex) - 1} steps\n")
out.append("| # | action | state |")
out.append("|---|---|---|")
for i, step in enumerate(cex):
label = step["label"] or "_initial_"
out.append(f"| {i} | `{label}` | `{_fmt_state(step['state'])}` |")
out.append("")
# A picture of the counterexample, when there is one to draw.
cex = next((r["counterexample"] for r in res["properties"].values() if r["holds"] is False), None)
if cex is not None:
try:
out.append("\n**Counterexample diagram** (Mermaid — paste into GitHub to render):\n")
out.append("```mermaid\n" + to_mermaid(model, cex, verdict="REFUTED") + "\n```\n")
except RenderTooLarge:
out.append("\n_The graph is too large to draw legibly, so no diagram is shown._\n")
if model.goal is not None:
live = check_liveness(model)
if live["holds"] is True:
out.append("### ✅ Liveness: every reachable state can still reach the goal\n")
elif live["holds"] is None:
out.append("### ❓ Liveness: undetermined\n")
else:
out.append("### ❌ Liveness: a state exists from which the goal is unreachable\n")
if live.get("counterexample"):
trap = live["counterexample"][-1]["state"]
out.append(f"Trap state: `{_fmt_state(trap)}`\n")
return "\n".join(out)
def load_example(name: str) -> str:
return json.dumps(EXAMPLES.get(name, EXAMPLE_BROKEN), indent=2)
def list_benchmark_tasks() -> list[list]:
"""Rows for the benchmark table."""
return [[t.id, t.standards_body, t.property, t.label, "yes" if t.fixed_builder else "—"] for t in TASKS.values()]
def inspect_task(task_id: str) -> tuple[str, str]:
"""Render a benchmark task: its state machine, and the ground-truth verdict."""
task = TASKS.get(task_id)
if task is None:
return "Unknown task.", ""
model = task.build()
lines = [
f"**{task.id}** — {task.standards_body}",
f"*{task.spec_clause}*",
"",
f"Property: `{task.property}`",
f"State fields: `{', '.join(model.fields)}`",
f"Initial: `{_fmt_state(dict(zip(model.fields, model.initial)))}`",
"",
"**Transitions**",
"",
"| from | action | to |",
"|---|---|---|",
]
from minicheck._core import _reachable
_, order = _reachable(model)
for s in order:
for label, ns in model.transitions(s):
lines.append(
f"| `{_fmt_state(dict(zip(model.fields, s)))}` | `{label}` "
f"| `{_fmt_state(dict(zip(model.fields, ns)))}` |"
)
res = check_safety(model)["properties"][task.property]
verdict = [f"### Ground truth: `{task.label}`\n"]
if res["holds"] is True:
verdict.append("The property **holds** over every reachable state.\n")
elif res["holds"] is None:
verdict.append("The property is **undetermined** — the search did not cover the whole space.\n")
else:
cex = res["counterexample"]
verdict.append(f"The property is **violated** in {len(cex) - 1} steps:\n")
verdict.append("| # | action | state |")
verdict.append("|---|---|---|")
for i, step in enumerate(cex):
verdict.append(f"| {i} | `{step['label'] or '_initial_'}` | `{_fmt_state(step['state'])}` |")
verdict.append("")
if task.citation:
verdict.append(f"**Citation:** {task.citation}")
if task.fixed_builder:
fixed = check_safety(task.build_fixed())["properties"][task.property]
verdict.append(
f"\n**Repaired twin:** property "
f"{'holds' if fixed['holds'] is True else ('is undetermined' if fixed['holds'] is None else 'still fails')} — "
"the counterexample is removable, so it is not a modelling artefact."
)
return "\n".join(lines), "\n".join(verdict)
def score_submission(text: str) -> str:
"""Score a pasted submission JSON, replaying every claimed counterexample."""
if not (text or "").strip():
return "Paste a submission, or press one of the baseline buttons."
try:
submission = json.loads(text)
except json.JSONDecodeError as e:
return f"### ⚠️ Not valid JSON\n\n```\n{e}\n```"
if not isinstance(submission, dict):
return "### ⚠️ Expected an object mapping task id to a prediction."
res = score(submission)
rows = [
"| metric | value |",
"|---|---|",
f"| **balanced accuracy** | **{res['balanced_accuracy']:.3f}** |",
f"| accuracy | {res['accuracy']:.3f} |",
f"| _trivial always-safe accuracy_ | _{res['trivial_always_safe_accuracy']:.3f}_ |",
f"| detections claimed | {res['detections_claimed']} |",
f"| **counterexamples that replayed** | **{res['valid_counterexamples']}** |",
f"| TP / FP / FN / TN | {res['true_positives']} / {res['false_positives']} "
f"/ {res['false_negatives']} / {res['true_negatives']} |",
"",
]
bad = [r for r in res["per_task"] if r["trace"].get("supplied") and not r["trace"]["valid"]]
if bad:
rows.append("**Traces that did not replay**\n")
rows.append("| task | reason |")
rows.append("|---|---|")
for r in bad:
rows.append(f"| `{r['id']}` | {r['trace']['reason']} |")
return "\n".join(rows)
def run_baseline(name: str) -> tuple[str, str]:
"""Run a built-in baseline; return (submission json, scorecard)."""
submission = BASELINES[name]()
text = json.dumps(submission, indent=2, default=str)
return text, score_submission(text)
def validate_single_trace(task_id: str, trace_text: str) -> str:
"""Replay one claimed counterexample against its model."""
task = TASKS.get(task_id)
if task is None:
return "Unknown task."
if not (trace_text or "").strip():
return 'Paste a trace: a JSON list of `{"state": {...}}` entries.'
try:
trace = json.loads(trace_text)
except json.JSONDecodeError as e:
return f"### ⚠️ Not valid JSON\n\n```\n{e}\n```"
v = validate_trace(task, trace)
if v["valid"]:
return (
f"### ✅ Replayed against the model\n\n{v['length']} steps. It starts at the initial "
"state, every step is a real transition, and the final state violates the property."
)
return f"### ❌ Did not replay\n\n**Reason:** {v['reason']}"
INTRO = """
# Protocol-Bench & minicheck
**Check a state machine for a safety violation and get the shortest sequence of steps that breaks
it.** Or score a submission against 15 published IEEE 802.11 / 3GPP procedures — where a claimed
detection has to *replay* before it counts.
This runs a real model checker **entirely in your browser** (Pyodide). Nothing is pre-computed and
nothing is sent to a server.
"""
SPEC_FORMAT = """
### Spec format
- `when` — a conjunction of `field == value` tests. Omit for an always-enabled transition.
- `set` — assign a literal, or `{"incr": n}` / `{"decr": n}` for integers.
- `invariants` — `{"forbid": {...}}` fails when **every** listed field matches;
`{"require": {...}}` fails unless they do.
- `goal` — optional, same shape, used for the liveness check.
Specs are **data, never code**: nothing you paste is executed.
"""
with gr.Blocks(title="Protocol-Bench & minicheck") as demo:
gr.Markdown(INTRO)
with gr.Tab("Check a state machine"):
example_pick = gr.Dropdown(choices=list(EXAMPLES), value=list(EXAMPLES)[0], label="Load an example")
spec_box = gr.Code(value=load_example(list(EXAMPLES)[0]), language="json", label="Spec", lines=18)
check_btn = gr.Button("Check", variant="primary")
spec_out = gr.Markdown()
gr.Markdown(SPEC_FORMAT)
example_pick.change(load_example, example_pick, spec_box)
check_btn.click(check_spec, spec_box, spec_out)
with gr.Tab("Browse the benchmark"):
gr.Markdown("15 published procedures. 13 hold; 2 do not.")
gr.Dataframe(
value=list_benchmark_tasks(),
headers=["task", "body", "property", "label", "repaired twin"],
interactive=False,
wrap=True,
)
task_pick = gr.Dropdown(choices=sorted(TASKS), value="ieee_4way_handshake_krack", label="Inspect a task")
task_model = gr.Markdown()
task_verdict = gr.Markdown()
task_pick.change(inspect_task, task_pick, [task_model, task_verdict])
with gr.Tab("Score a submission"):
gr.Markdown(
"Every trace you supply is **replayed against the model**. Try the baselines: "
"`always-safe` scores 0.867 plain accuracy and 0.500 balanced; `always-violated` "
"claims 15 detections and validates **zero**."
)
with gr.Row():
b_bfs = gr.Button("baseline: bfs")
b_safe = gr.Button("baseline: always-safe")
b_viol = gr.Button("baseline: always-violated")
sub_box = gr.Code(language="json", label="Submission", lines=14)
score_btn = gr.Button("Score", variant="primary")
sub_out = gr.Markdown()
b_bfs.click(lambda: run_baseline("bfs"), None, [sub_box, sub_out])
b_safe.click(lambda: run_baseline("always-safe"), None, [sub_box, sub_out])
b_viol.click(lambda: run_baseline("always-violated"), None, [sub_box, sub_out])
score_btn.click(score_submission, sub_box, sub_out)
with gr.Tab("Replay one trace"):
trace_task = gr.Dropdown(choices=sorted(TASKS), value="ieee_4way_handshake_krack", label="Task")
trace_box = gr.Code(
language="json",
label="Trace",
lines=10,
value='[{"state": {"ptk_installed": false, "tx_nonce": 0, "nonce_reused": false}}]',
)
trace_btn = gr.Button("Replay", variant="primary")
trace_out = gr.Markdown()
trace_btn.click(validate_single_trace, [trace_task, trace_box], trace_out)
gr.Markdown(
"---\n"
"**A verdict you cannot check is not a verdict** — and *undetermined is not a pass*. "
"This demo is one piece of a larger set.\n\n"
"[Documentation](https://nickharris808.github.io/verification-docs/) · "
"[minicheck](https://github.com/nickharris808/minicheck) · "
"[protocol-bench](https://github.com/nickharris808/protocol-bench) · "
"[specforge](https://github.com/nickharris808/specforge) · "
"[the leaderboard](https://huggingface.co/spaces/nickh007/specforge-leaderboard)"
)
demo.launch()