Timmy Jr.
a3330a4642
Addresses the four release-critical blockers from Timmy's review of PR #42:
1. Guard enum/allowlist membership against unhashable JSON (lists/dicts) in
form.state, receipt.signed_by, and validate_transition target.
2. Require audience_choice.source_issue to be a positive non-bool integer.
3. Make power/scar/cost deltas structured (before -> after), not prose.
4. Remove tracked .pyc bytecode; prove clean checkout stays clean after tests.
Adds 21 new negative tests across HostileJsonFailClosedTests,
SourceIssueValidationTests, and StructuredDeltaTests.
Based on PR #42 head (0b74302). Vincent retains full authorship of the
canon validator, lifecycle model, and r2 review response; this branch
contains only the integration fixes needed to make the stated security
contract true at hostile-input boundaries.
Refs: #19, PR #42
351 lines
13 KiB
Python
351 lines
13 KiB
Python
"""
|
|
Dependency-free, fail-closed validator for Cannon Chain canon lifecycle.
|
|
|
|
Revision 2 (schema_version 2). Addresses the formal changes-requested review
|
|
on PR #27:
|
|
1. schema_version is bumped to 2 and *enforced* (not just present).
|
|
2. audience_choice requires the chosen form to be the SOLE survivor and
|
|
every other form to be an archive_ghost (no lingering candidates).
|
|
3. Receipts require a non-empty action and exact power/scar/cost deltas.
|
|
4. Provenance is a structured object: source_issue (int) plus a PR or
|
|
commit binding. A free-text string is rejected.
|
|
5. Signer check is a true configured allowlist (a set), not a namespace
|
|
regex. `human:anyone` is rejected unless explicitly configured.
|
|
6. audience_choice source fields (stolen_trait, source_episode,
|
|
source_issue) must be non-empty.
|
|
7. Malformed containers/items (None, wrong types, non-dict receipts)
|
|
return validation errors and NEVER raise.
|
|
|
|
Usage:
|
|
from canon_validator import validate_canon
|
|
result = validate_canon(data)
|
|
assert result["valid"], result["errors"]
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
CANON_PATH = Path(__file__).parent / "canon.json"
|
|
|
|
SCHEMA_VERSION = 2
|
|
VALID_STATES = {"candidate", "survivor", "archive_ghost"}
|
|
VALID_TRANSITIONS = {
|
|
"candidate": {"survivor", "archive_ghost"},
|
|
"survivor": set(), # terminal
|
|
"archive_ghost": set(), # terminal
|
|
}
|
|
|
|
# True configured allowlist — explicit identities only, no namespace wildcards.
|
|
DEFAULT_SIGNERS = frozenset({
|
|
"agent:vincent",
|
|
"agent:timmy",
|
|
"agent:nightprowl",
|
|
"human:grepples",
|
|
})
|
|
|
|
_COMMIT_RE = re.compile(r"^[0-9a-f]{7,40}$")
|
|
|
|
|
|
def _is_nonempty_str(v) -> bool:
|
|
return isinstance(v, str) and v.strip() != ""
|
|
|
|
|
|
def _is_structured_delta(v) -> bool:
|
|
"""A receipt delta must bind a before/after change, not just prose."""
|
|
if not isinstance(v, str):
|
|
return False
|
|
if "->" not in v:
|
|
return False
|
|
parts = v.split("->", 1)
|
|
return parts[0].strip() != "" and parts[1].strip() != ""
|
|
|
|
|
|
def _validate_provenance(prov, prefix, errors) -> None:
|
|
if not isinstance(prov, dict):
|
|
errors.append(f"{prefix}: provenance must be an object, got {type(prov).__name__}")
|
|
return
|
|
si = prov.get("source_issue")
|
|
if not isinstance(si, int) or isinstance(si, bool) or si <= 0:
|
|
errors.append(f"{prefix}: provenance.source_issue must be a positive integer")
|
|
has_pull = isinstance(prov.get("pull"), int) and not isinstance(prov.get("pull"), bool) and prov["pull"] > 0
|
|
commit = prov.get("commit")
|
|
has_commit = isinstance(commit, str) and bool(_COMMIT_RE.match(commit))
|
|
if not has_pull and not has_commit:
|
|
errors.append(
|
|
f"{prefix}: provenance must bind a pull (int) or commit (7-40 hex)"
|
|
)
|
|
|
|
|
|
def _validate_receipt(receipt, idx, form_ids, signers, errors) -> None:
|
|
prefix = f"receipts[{idx}]"
|
|
if not isinstance(receipt, dict):
|
|
errors.append(f"{prefix}: receipt must be an object, got {type(receipt).__name__}")
|
|
return
|
|
|
|
# Exact required string fields (non-empty).
|
|
for field in ("episode_id", "form_id", "action"):
|
|
if field not in receipt:
|
|
errors.append(f"{prefix}: missing '{field}'")
|
|
elif not _is_nonempty_str(receipt[field]):
|
|
errors.append(f"{prefix}: '{field}' must be a non-empty string")
|
|
|
|
# Structured power/scar/cost deltas: must bind before/after, not just prose.
|
|
for field in ("power_delta", "scar_delta", "cost_delta"):
|
|
if field not in receipt:
|
|
errors.append(f"{prefix}: missing '{field}'")
|
|
elif not _is_structured_delta(receipt[field]):
|
|
errors.append(f"{prefix}: '{field}' must bind before/after (e.g. '0 -> 1'), got {receipt[field]!r}")
|
|
|
|
# form_id must reference a known form.
|
|
fid = receipt.get("form_id")
|
|
if isinstance(fid, str) and fid and fid not in form_ids:
|
|
errors.append(f"{prefix}: form_id '{fid}' not found in forms")
|
|
|
|
# Structured provenance.
|
|
if "provenance" not in receipt:
|
|
errors.append(f"{prefix}: missing 'provenance'")
|
|
else:
|
|
_validate_provenance(receipt["provenance"], prefix, errors)
|
|
|
|
# Configured signer allowlist.
|
|
if "signed_by" not in receipt:
|
|
errors.append(f"{prefix}: missing 'signed_by'")
|
|
else:
|
|
signer = receipt.get("signed_by")
|
|
if not isinstance(signer, str):
|
|
errors.append(f"{prefix}: 'signed_by' must be a non-empty string")
|
|
elif signer not in signers:
|
|
errors.append(f"{prefix}: signed_by '{signer}' is not in the configured allowlist")
|
|
|
|
|
|
def _validate_audience_choice(choice, form_states, form_ids, errors) -> None:
|
|
if not isinstance(choice, dict):
|
|
errors.append(
|
|
f"audience_choice must be an object, got {type(choice).__name__}"
|
|
)
|
|
return
|
|
|
|
for field in ("survivor_id", "stolen_trait", "source_episode", "source_issue", "committed_at"):
|
|
if field not in choice:
|
|
errors.append(f"audience_choice: missing '{field}'")
|
|
|
|
sid = choice.get("survivor_id")
|
|
if "survivor_id" in choice:
|
|
if not _is_nonempty_str(sid):
|
|
errors.append("audience_choice: survivor_id must be a non-empty string")
|
|
elif sid not in form_ids:
|
|
errors.append(f"audience_choice: survivor_id '{sid}' not found in forms")
|
|
|
|
# Non-empty source fields.
|
|
for field in ("stolen_trait", "source_episode"):
|
|
if field in choice and not _is_nonempty_str(choice[field]):
|
|
errors.append(f"audience_choice: '{field}' must be a non-empty string")
|
|
|
|
# source_issue must be a positive non-bool integer.
|
|
si = choice.get("source_issue")
|
|
if "source_issue" in choice:
|
|
if not isinstance(si, int) or isinstance(si, bool) or si <= 0:
|
|
errors.append("audience_choice: source_issue must be a positive integer")
|
|
|
|
# committed_at must be a valid ISO timestamp.
|
|
if "committed_at" in choice:
|
|
ts = choice["committed_at"]
|
|
if not _is_nonempty_str(ts):
|
|
errors.append("audience_choice: committed_at must be a non-empty ISO timestamp")
|
|
else:
|
|
try:
|
|
from datetime import datetime
|
|
datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
|
except (ValueError, TypeError):
|
|
errors.append(f"audience_choice: committed_at '{ts}' is not a valid ISO timestamp")
|
|
|
|
# Commitment invariants: the chosen form is the sole survivor and every
|
|
# other form is an archive_ghost.
|
|
survivors = [fid for fid, st in form_states.items() if st == "survivor"]
|
|
if len(survivors) != 1:
|
|
errors.append(
|
|
f"audience_choice is set but there are {len(survivors)} survivors (expected exactly 1)"
|
|
)
|
|
elif survivors[0] != sid:
|
|
errors.append(
|
|
f"audience_choice survivor_id '{sid}' does not match the sole survivor '{survivors[0]}'"
|
|
)
|
|
|
|
for fid, st in form_states.items():
|
|
if fid != sid and st != "archive_ghost":
|
|
errors.append(
|
|
f"audience_choice is set but form '{fid}' is '{st}', expected 'archive_ghost'"
|
|
)
|
|
|
|
|
|
def validate_canon(data, signers=None) -> dict:
|
|
"""
|
|
Validate canon.json data against the revision-2 lifecycle contract.
|
|
|
|
Args:
|
|
data: parsed canon.json dict.
|
|
signers: optional iterable of allowed signer identities. Defaults to
|
|
DEFAULT_SIGNERS.
|
|
|
|
Returns:
|
|
{"valid": bool, "errors": [str], "warnings": [str]}
|
|
|
|
Never raises on malformed input; all problems are returned as errors.
|
|
"""
|
|
errors = []
|
|
warnings = []
|
|
if signers is None:
|
|
signers = DEFAULT_SIGNERS
|
|
try:
|
|
signers = frozenset(signers)
|
|
except TypeError:
|
|
errors.append("signers must be an iterable of strings")
|
|
signers = frozenset()
|
|
|
|
if not isinstance(data, dict):
|
|
return {
|
|
"valid": False,
|
|
"errors": [f"canon must be an object, got {type(data).__name__}"],
|
|
"warnings": warnings,
|
|
}
|
|
|
|
# --- schema_version (enforced, not just present) ---
|
|
if "schema_version" not in data:
|
|
errors.append("missing schema_version")
|
|
elif data["schema_version"] != SCHEMA_VERSION:
|
|
errors.append(
|
|
f"schema_version {data['schema_version']!r} unsupported, expected {SCHEMA_VERSION}"
|
|
)
|
|
|
|
# --- forms ---
|
|
forms = data.get("forms")
|
|
if not isinstance(forms, list) or len(forms) == 0:
|
|
errors.append("forms must be a non-empty array")
|
|
forms = []
|
|
|
|
form_ids = set()
|
|
form_states = {}
|
|
survivor_count = 0
|
|
|
|
for i, form in enumerate(forms):
|
|
prefix = f"forms[{i}]"
|
|
if not isinstance(form, dict):
|
|
errors.append(f"{prefix}: form must be an object, got {type(form).__name__}")
|
|
continue
|
|
|
|
fid = form.get("id")
|
|
if not _is_nonempty_str(fid):
|
|
errors.append(f"{prefix}: missing or empty id")
|
|
else:
|
|
if fid in form_ids:
|
|
errors.append(f"{prefix}: duplicate form id '{fid}'")
|
|
form_ids.add(fid)
|
|
|
|
for field in ("power", "scar", "cost"):
|
|
if field not in form or not _is_nonempty_str(form[field]):
|
|
errors.append(f"{prefix}: missing or empty '{field}'")
|
|
|
|
palette = form.get("palette")
|
|
if not isinstance(palette, list):
|
|
errors.append(f"{prefix}: missing or invalid palette")
|
|
elif len(palette) < 3:
|
|
errors.append(f"{prefix}: palette must have at least 3 colors, got {len(palette)}")
|
|
|
|
state = form.get("state")
|
|
if state is None:
|
|
errors.append(f"{prefix}: missing lifecycle state")
|
|
elif not isinstance(state, str) or state not in VALID_STATES:
|
|
errors.append(f"{prefix}: invalid state '{state}', expected one of {sorted(VALID_STATES)}")
|
|
else:
|
|
form_states[fid] = state
|
|
if state == "survivor":
|
|
survivor_count += 1
|
|
|
|
if survivor_count > 1:
|
|
errors.append(f"too many survivors: {survivor_count} (expected 0 or 1)")
|
|
|
|
# --- episodes ---
|
|
episodes = data.get("episodes")
|
|
if isinstance(episodes, list):
|
|
ready_count = 0
|
|
found_ready = False
|
|
for ep in episodes:
|
|
if not isinstance(ep, dict):
|
|
errors.append(f"episodes: item must be an object, got {type(ep).__name__}")
|
|
continue
|
|
st = ep.get("state")
|
|
if st == "ready":
|
|
ready_count += 1
|
|
if found_ready:
|
|
errors.append(f"episodes: ready episode '{ep.get('id')}' after another ready episode")
|
|
found_ready = True
|
|
elif st != "locked":
|
|
errors.append(f"episodes: '{ep.get('id')}' has invalid state '{st}'")
|
|
if ready_count > 1:
|
|
errors.append(f"episodes: too many ready episodes: {ready_count} (expected at most 1)")
|
|
|
|
# --- audience choice ---
|
|
choice = data.get("audience_choice")
|
|
if choice is not None:
|
|
_validate_audience_choice(choice, form_states, form_ids, errors)
|
|
|
|
# --- receipts ---
|
|
receipts = data.get("receipts", [])
|
|
if not isinstance(receipts, list):
|
|
errors.append(f"receipts must be an array, got {type(receipts).__name__}")
|
|
receipts = []
|
|
for i, receipt in enumerate(receipts):
|
|
_validate_receipt(receipt, i, form_ids, signers, errors)
|
|
|
|
return {
|
|
"valid": len(errors) == 0,
|
|
"errors": errors,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def validate_transition(form, new_state) -> dict:
|
|
"""
|
|
Validate a state transition for a single form.
|
|
|
|
Returns {"valid": bool, "errors": [str]}. Never raises.
|
|
"""
|
|
errors = []
|
|
if not isinstance(form, dict):
|
|
return {"valid": False, "errors": [f"form must be an object, got {type(form).__name__}"]}
|
|
|
|
current = form.get("state")
|
|
if not isinstance(current, str) or current not in VALID_TRANSITIONS:
|
|
return {"valid": False, "errors": [f"invalid current state {current!r}"]}
|
|
if not isinstance(new_state, str) or new_state not in VALID_STATES:
|
|
return {"valid": False, "errors": [f"invalid target state {new_state!r}"]}
|
|
if new_state not in VALID_TRANSITIONS[current]:
|
|
return {"valid": False, "errors": [f"illegal transition: '{current}' -> '{new_state}'"]}
|
|
return {"valid": True, "errors": []}
|
|
|
|
|
|
def load_canon(path=None) -> dict:
|
|
target = Path(path) if path else CANON_PATH
|
|
return json.loads(target.read_text())
|
|
|
|
|
|
def check_canon(path=None, signers=None) -> dict:
|
|
data = load_canon(path)
|
|
return validate_canon(data, signers=signers)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
result = check_canon()
|
|
if result["valid"]:
|
|
print(f"canon.json: VALID (schema_version {SCHEMA_VERSION})")
|
|
for w in result["warnings"]:
|
|
print(f" WARNING: {w}")
|
|
else:
|
|
print("canon.json: INVALID")
|
|
for e in result["errors"]:
|
|
print(f" ERROR: {e}")
|
|
sys.exit(1)
|