tegwick
71c17fb5f9
impl/js/src/email.mjs (written from CONTRACT.md + FLOWS.md): both roles, case store, handoffs, timeouts, one-cap outbox with holding, authorization. Passes all 22 flows from the bundle; with the JS master all 12 slave flows (no Python contract party). Writing it found six contract gaps, folded into the draft as MCP-1..6 (acknowledgement, details convention, resume names the outcome, opaque case key, error table, transfer reporting); flows now check outcome tokens, not reference prose. A mutation check found a missing flow: a-008 (per-type master check). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 5787@bnt-lap001 Assistant-Session: 5db26b82-138b-40cc-9856-4d5eb0f752fd
320 lines
13 KiB
Python
320 lines
13 KiB
Python
"""Run the mode-contract conformance bundle against this email side.
|
|
|
|
Message fixtures and the transition table are checked against the schema
|
|
and `contract.allowed`; flows run through `harness.Harness` with the real
|
|
gateway, the callback stub and the stub outside master. Used by `mas test`
|
|
and pytest.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
from collections import Counter
|
|
from dataclasses import dataclass, field
|
|
from importlib import resources
|
|
from typing import Optional
|
|
|
|
from . import _load, message_errors, message_fixtures, transitions
|
|
from .contract import allowed
|
|
from .harness import Harness
|
|
from .signing import AuthError, sign, signature, verify
|
|
|
|
@dataclass
|
|
class Report:
|
|
passed: int = 0
|
|
failures: list = field(default_factory=list)
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return not self.failures
|
|
|
|
|
|
def check_messages(report: Report):
|
|
fx = message_fixtures()
|
|
for c in fx["valid"]:
|
|
errs = message_errors(c["message"])
|
|
(report.failures.append((c["case_id"], f"should be valid: {errs}")) if errs
|
|
else _ok(report))
|
|
for c in fx["invalid"]:
|
|
(report.failures.append((c["case_id"], f"should be invalid ({c['why']})"))
|
|
if not message_errors(c["message"]) else _ok(report))
|
|
t = transitions()
|
|
statuses = set(t["terminal"]) | set(t["transitions"]) - {"(new)"}
|
|
for old in [None, *sorted(statuses)]:
|
|
for new in sorted(statuses):
|
|
want = new in t["transitions"].get(old or "(new)", [])
|
|
if allowed(old, new) != want:
|
|
report.failures.append((f"transition {old}->{new}", f"expected allowed={want}"))
|
|
else:
|
|
_ok(report)
|
|
|
|
|
|
def _ok(report):
|
|
report.passed += 1
|
|
|
|
|
|
def check_auth(report: Report):
|
|
"""Signing vectors and verification cases (CONTRACT.md section 7)."""
|
|
fx = _load("conformance/auth.json")
|
|
for v in fx["vectors"]:
|
|
got = signature(bytes.fromhex(v["key_hex"]), v["timestamp"], v["body"].encode())
|
|
(report.failures.append((v["id"], f"signature {got}")) if got != v["signature"] else _ok(report))
|
|
for c in fx["verify"]:
|
|
keys = {p: bytes.fromhex(k) for p, k in c["keys"].items()}
|
|
try:
|
|
got = verify(c["headers"], c["body"].encode(), keys, now=c["now"])
|
|
except AuthError as e:
|
|
got = "error:" + e.code
|
|
want = c["peer"] if "peer" in c else "error:" + c["error"]
|
|
(report.failures.append((c["id"], f"got {got}, expected {want}")) if got != want else _ok(report))
|
|
|
|
|
|
def _inject(h, spec, ctx):
|
|
"""Send a crafted request straight to an email endpoint (authorization flows)."""
|
|
if spec["endpoint"] == "status":
|
|
reqs = [m for m in h.callback.received if m["type"] == "handoff.request"]
|
|
url = reqs[-1]["status_url"]
|
|
if spec.get("token") == "bogus":
|
|
url = url.rsplit("/", 1)[0] + "/not-a-token"
|
|
else:
|
|
url = h.email.base_url + {"master": "/master", "events": "/mas-events"}[spec["endpoint"]]
|
|
msg = {"mas_modes_version": "0.1", "id": uuid.uuid4().hex, **_fill(spec["message"], ctx)}
|
|
if msg.get("case") == "current": # the case (and handoff) the flow is working on
|
|
if spec["endpoint"] == "status":
|
|
msg["case"], msg["handoff_id"] = reqs[-1]["case"], reqs[-1]["handoff_id"]
|
|
else:
|
|
msg["case"] = {"address": ctx["case_address"], "key": ctx["case_key"]}
|
|
body = json.dumps(msg).encode()
|
|
headers = {"Content-Type": "application/json"}
|
|
mode = spec.get("sign", "valid")
|
|
if mode != "none":
|
|
peer = spec["as"]
|
|
key = h.keys.get(peer, b"k" * 32) if mode != "wrong-key" else b"x" * 32
|
|
ts = int(time.time()) - (400 if mode == "stale" else 0)
|
|
headers.update(sign(body, peer, key, ts))
|
|
req = urllib.request.Request(url, data=body, method="POST", headers=headers)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return r.status, json.loads(r.read().decode() or "null")
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read().decode() or "null")
|
|
|
|
|
|
STEPS = {"mail", "mode", "advance", "master", "inject"}
|
|
EXPECT = {"event_accepted", "mode_statuses", "mode_received", "mode_received_last",
|
|
"master_received", "relayed", "mails_sent", "last_mail_contains", "response_ok",
|
|
"response_reason", "response_held", "case_address", "http_status", "error"}
|
|
|
|
|
|
def undefined_keys(flow: dict) -> list[str]:
|
|
"""Steps or expectations not defined in conformance/FLOWS.md."""
|
|
bad = []
|
|
for n, step in enumerate(flow["steps"], 1):
|
|
bad += [f"step {n}: {k}" for k in set(step) - STEPS - {"expect"}]
|
|
bad += [f"step {n}: expect {k}" for k in set(step.get("expect", {})) - EXPECT]
|
|
return bad
|
|
|
|
|
|
def _fill(value, ctx):
|
|
if isinstance(value, str):
|
|
for k, v in ctx.items():
|
|
if isinstance(v, str):
|
|
value = value.replace("{" + k + "}", v)
|
|
return value
|
|
if isinstance(value, dict):
|
|
return {k: _fill(v, ctx) for k, v in value.items()}
|
|
return value
|
|
|
|
|
|
def _roles(setup: dict) -> dict:
|
|
"""A flow setup (peer ids only) as harness role configs (with stub URLs)."""
|
|
def url_for(v):
|
|
if isinstance(v, str) and v.startswith("<mode:"):
|
|
return "{callback}"
|
|
if isinstance(v, dict):
|
|
return {k: url_for(x) for k, x in v.items()}
|
|
return v
|
|
roles = {}
|
|
for rtype, cfg in setup["roles"].items():
|
|
cfg = url_for(dict(cfg))
|
|
if cfg["role"] == "master":
|
|
cfg["modes"] = {m: {"url": "{callback}", **e} for m, e in cfg.get("modes", {}).items()}
|
|
cfg["fields"] = tuple(cfg.get("fields", ("phone", "time")))
|
|
else:
|
|
cfg["master_url"] = "{master}"
|
|
roles[rtype] = cfg
|
|
return roles
|
|
|
|
|
|
def _capture(ctx, items):
|
|
for m in items:
|
|
if not isinstance(m, dict):
|
|
continue
|
|
if isinstance(m.get("case"), dict):
|
|
ctx["case"] = m["case"]
|
|
ctx["case_id"] = m["case"]["address"].split("=", 1)[-1].rsplit("@", 1)[0]
|
|
if m.get("type") in ("case.message", None) and m.get("message_ref"):
|
|
ctx["message_ref"] = m["message_ref"]
|
|
if m.get("type") == "case.intake":
|
|
ctx["intake_ref"] = m["intake_ref"]
|
|
if m.get("type") in ("handoff.request", "handoff.requested"):
|
|
ctx["handoff_id"] = m["handoff_id"]
|
|
|
|
|
|
def _inject(h, spec, ctx):
|
|
"""Send a crafted request straight to an email endpoint (FLOWS.md, inject)."""
|
|
if spec["endpoint"] == "status":
|
|
reqs = [m for m in h.callback.received if m["type"] == "handoff.request"]
|
|
url = reqs[-1]["status_url"]
|
|
if spec.get("token") == "bogus":
|
|
url = url.rsplit("/", 1)[0] + "/not-a-token"
|
|
else:
|
|
url = h.email.base_url + {"master": "/master", "events": "/mas-events"}[spec["endpoint"]]
|
|
msg = {"mas_modes_version": "0.1", "id": uuid.uuid4().hex, **_fill(spec["message"], ctx)}
|
|
if msg.get("case") == "current":
|
|
msg["case"] = ctx["case"]
|
|
body = json.dumps(msg).encode()
|
|
headers = {"Content-Type": "application/json"}
|
|
mode = spec.get("sign", "valid")
|
|
if mode != "none":
|
|
peer = spec["as"]
|
|
key = h.keys.get(peer, b"k" * 32) if mode != "wrong-key" else b"x" * 32
|
|
ts = int(time.time()) - (400 if mode == "stale" else 0)
|
|
headers.update(sign(body, peer, key, ts))
|
|
req = urllib.request.Request(url, data=body, method="POST", headers=headers)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return r.status, json.loads(r.read().decode() or "null")
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read().decode() or "null")
|
|
|
|
|
|
def _master(h, cmd, ctx):
|
|
c = cmd["command"]
|
|
if c == "assign":
|
|
return h.master.assign(cmd.get("intake_ref", ctx.get("intake_ref")), cmd["resource_id"])
|
|
if c == "send":
|
|
return h.master.send(ctx["case"], cmd["in_reply_to"], cmd["body"], cmd.get("subject"))
|
|
if c == "cancel":
|
|
return h.master.cancel(ctx["case"], cmd["handoff_id"])
|
|
if c == "policy":
|
|
return h.master.policy(ctx["case"], cmd.get("autonomous_send"), cmd.get("transfers"))
|
|
raise KeyError(f"unknown master command {c!r}")
|
|
|
|
|
|
def run_flow(flow: dict, doc: Optional[dict] = None, master_factory=None, email_factory=None) -> list[str]:
|
|
"""Run one flow from the bundle (FLOWS.md). Returns the problems found.
|
|
master_factory swaps in another outside master (e.g. the JS one)."""
|
|
doc = doc or _load("conformance/flows.json")
|
|
bad = undefined_keys(flow)
|
|
if bad:
|
|
return [f"undefined key(s): {bad}"]
|
|
keys = {p: bytes.fromhex(k) for p, k in doc["peers"].items()}
|
|
with resources.as_file(resources.files(__package__).joinpath("conformance", doc["statute"])) as st:
|
|
h = Harness(st, _roles(doc["setups"][flow["setup"]]), scenario=flow["scenario"],
|
|
auto_assign=False, max_sent=flow.get("max_sent"), keys=keys,
|
|
master_factory=master_factory, email_factory=email_factory)
|
|
ctx: dict = {}
|
|
problems = []
|
|
try:
|
|
for n, step in enumerate(flow["steps"], 1):
|
|
try:
|
|
_step(h, n, step, ctx, problems)
|
|
except Exception as e: # a broken peer must fail the flow, not the run
|
|
problems.append(f"step {n} raised {type(e).__name__}: {e}")
|
|
break
|
|
finally:
|
|
h.close()
|
|
return problems
|
|
|
|
|
|
def _step(h, n, step, ctx, problems):
|
|
"""One flow step (FLOWS.md), then its expectations."""
|
|
seen = (len(h.callback.received), len(h.master.received))
|
|
event, response = None, None
|
|
if "mail" in step:
|
|
m = _fill(step["mail"], ctx)
|
|
event = h.mail(m["to"], m.get("text", ""), subject=m.get("subject", "Help needed"),
|
|
dmarc=m.get("dmarc", "pass"))
|
|
elif "mode" in step:
|
|
h.callback.run()
|
|
elif "advance" in step:
|
|
h.advance(step["advance"])
|
|
elif "master" in step:
|
|
response = _master(h, _fill(step["master"], ctx), ctx)
|
|
elif "inject" in step:
|
|
response = _inject(h, step["inject"], ctx)
|
|
_capture(ctx, h.callback.received[seen[0]:] + h.master.received[seen[1]:])
|
|
if response is not None:
|
|
_capture(ctx, [response[1]])
|
|
for key, want in step.get("expect", {}).items():
|
|
got = _observe(h, key, event, response)
|
|
if key == "master_received":
|
|
got = {t: got.get(t, 0) for t in want}
|
|
elif key == "last_mail_contains":
|
|
got, want = (want in (got or "")), True
|
|
if got != want:
|
|
problems.append(f"step {n} {key}: got {got!r}, expected {want!r}")
|
|
|
|
|
|
def _observe(h, key, event, response):
|
|
body = (response[1] if response else None) or {}
|
|
if key == "event_accepted":
|
|
return event is not None and event.status is not None and 200 <= event.status < 300
|
|
if key == "mode_statuses":
|
|
return [m["status"] for hd in h.callback.handoffs.values() for m in hd.sent]
|
|
if key == "mode_received":
|
|
return len(h.callback.received)
|
|
if key == "mode_received_last":
|
|
return h.callback.received[-1]["type"] if h.callback.received else None
|
|
if key == "master_received":
|
|
return dict(Counter(m["type"] for m in h.master.received))
|
|
if key == "relayed":
|
|
return [m["status"] for m in h.master.of_type("handoff.status")]
|
|
if key == "mails_sent":
|
|
return len(h.outbox())
|
|
if key == "last_mail_contains":
|
|
mails = h.outbox()
|
|
return mails[-1]["body"] if mails else None
|
|
if key == "response_ok":
|
|
return body.get("ok")
|
|
if key == "response_reason":
|
|
return body.get("reason")
|
|
if key == "response_held":
|
|
return body.get("held")
|
|
if key == "case_address":
|
|
return (body.get("case") or {}).get("address")
|
|
if key == "http_status":
|
|
return response[0] if response else None
|
|
if key == "error":
|
|
return body.get("error")
|
|
raise KeyError(f"unknown expectation {key!r}")
|
|
|
|
|
|
def run_mode_corpus(master_factory=None, roles=("master", "slave"), email_factory=None) -> Report:
|
|
"""All fixtures and flows. With master_factory, only the flows of the given
|
|
setup roles run, against that outside master."""
|
|
report = Report()
|
|
if master_factory is None and email_factory is None:
|
|
check_messages(report)
|
|
check_auth(report)
|
|
doc = _load("conformance/flows.json")
|
|
for flow in doc["flows"]:
|
|
setup_roles = {c["role"] for c in doc["setups"][flow["setup"]]["roles"].values()}
|
|
if not setup_roles & set(roles):
|
|
continue
|
|
problems = run_flow(flow, doc, master_factory, email_factory)
|
|
if problems:
|
|
report.failures.append((flow["id"], "; ".join(problems)))
|
|
else:
|
|
report.passed += 1
|
|
return report
|
|
|
|
|
|
if __name__ == "__main__":
|
|
r = run_mode_corpus()
|
|
print(json.dumps({"passed": r.passed, "failures": r.failures}, indent=2))
|