Implement MAS v0.1 reference experiment (MAS-WP-0002 M0-M8, M9 first pass)
- mas_core: P2-maximal parser with required_profile, generator, statute schema + semantic lint + defaults-only inheritance, exact-host discovery with typed failures and same-origin redirects, pure policy resolver - mas_conformance: vendored spec fixtures (drift-tested) + property and statute cases - mas CLI: parse / explain / discover / lint-statute / test - mas-gateway: generic mail -> MAS event -> HTTP (MTA pipe + webhook) - examples: static great-service.example statute, reference ticket service - docs/experiment-review.md: first-pass findings; live MTA test still open Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 623586@bnt-lap001 Assistant-Session: 10b30928-5b3b-4c11-9934-5ecca07367d8
This commit is contained in:
parent
827977ea4e
commit
59815701c1
48 changed files with 4096 additions and 15 deletions
217
mas_cli/main.py
Normal file
217
mas_cli/main.py
Normal file
|
|
@ -0,0 +1,217 @@
|
|||
"""`mas` — inspection CLI for Mail Address Semantics v0.1."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from mas_core import MAS_VERSION, __version__
|
||||
from mas_core.discovery import Discovered, discover
|
||||
from mas_core.parser import MasSyntaxError, parse_address, parse_local_part, validate_profile
|
||||
from mas_core.policy import resolve
|
||||
from mas_core.statute import ERROR, SCHEMA, build_statute, lint
|
||||
|
||||
|
||||
def _out(args, data, text):
|
||||
print(json.dumps(data, indent=2) if args.json else text)
|
||||
|
||||
|
||||
def _parse_any(value):
|
||||
return parse_address(value) if "@" in value else parse_local_part(value)
|
||||
|
||||
|
||||
def _syntax_lines(d):
|
||||
lines = [f" target: {d['target']}", f" target syntax: {d['target_kind']}"]
|
||||
if d["target_kind"] == "resource":
|
||||
lines += [f" resource type: {d['resource_type']}", f" resource id: {d['resource_id']}"]
|
||||
if d["facet"]:
|
||||
lines.append(f" facet: {d['facet']} (base if stripped: {d['base']})")
|
||||
if d["interaction"]:
|
||||
lines.append(f" interaction: {d['interaction']}")
|
||||
lines.append(f" required profile: {d['required_profile']}")
|
||||
if not d["generator_allowed"]:
|
||||
lines.append(" note: valid to parse, but a v0.1 generator must not emit '+' with '!'")
|
||||
return lines
|
||||
|
||||
|
||||
def cmd_parse(args):
|
||||
try:
|
||||
p = _parse_any(args.address)
|
||||
except MasSyntaxError as e:
|
||||
_out(args, {"valid": False, "error": e.code, "message": str(e)},
|
||||
f"invalid MAS v0.1 syntax [{e.code}]: {e}")
|
||||
return 1
|
||||
d = p.to_dict()
|
||||
if args.profile:
|
||||
r = validate_profile(p, args.profile)
|
||||
d["profile_check"] = {"profile": args.profile.upper(), "ok": r.ok, "error": r.code}
|
||||
if not r:
|
||||
_out(args, {"valid": False, **d}, f"invalid under {args.profile.upper()} [{r.code}]: {r.message}")
|
||||
return 1
|
||||
_out(args, {"valid": True, **d}, "Syntax:\n" + "\n".join(_syntax_lines(d)))
|
||||
return 0
|
||||
|
||||
|
||||
def _discover(args, host):
|
||||
if args.statute:
|
||||
doc = json.loads(Path(args.statute).read_text())
|
||||
return Discovered(host, f"file:{args.statute}", build_statute(doc, host))
|
||||
return discover(host, scheme="http" if args.insecure_http else "https", port=args.port)
|
||||
|
||||
|
||||
def cmd_discover(args):
|
||||
r = _discover(args, args.host)
|
||||
if r.ok:
|
||||
st = r.statute
|
||||
text = [f"Statute for {r.host} ({r.url}): valid",
|
||||
f" profiles: P1, P2={'enabled' if st.p2 else 'disabled'}",
|
||||
f" resource types: {', '.join(k for k in st.resource_types if st.resource_type_enabled(k)) or '-'}",
|
||||
f" interaction modes: {', '.join(k for k in st.interaction_modes if st.interaction_mode_enabled(k)) or '-'}",
|
||||
f" equivalence: case={st.local_part_case}, dot_folding={st.dot_folding}",
|
||||
f" registry entries: {len(st.addresses)}"]
|
||||
if st.inherited_from:
|
||||
text.append(f" inherited defaults from: {', '.join(st.inherited_from)}")
|
||||
_out(args, r.to_dict(), "\n".join(text))
|
||||
return 0
|
||||
_out(args, r.to_dict(), f"No valid MAS statute for {r.host}\n failure: {r.kind.value}\n"
|
||||
f" url: {r.url}\n detail: {r.detail}")
|
||||
return 2
|
||||
|
||||
|
||||
def cmd_explain(args):
|
||||
try:
|
||||
addr = parse_address(args.address)
|
||||
except MasSyntaxError as e:
|
||||
_out(args, {"address": args.address, "syntax": {"valid": False, "error": e.code}},
|
||||
f"Address:\n {args.address}\n\nSyntax:\n invalid MAS v0.1 syntax [{e.code}]: {e}\n"
|
||||
" (the address may still be deliverable as ordinary email)")
|
||||
return 1
|
||||
registry = json.loads(Path(args.registry).read_text()) if args.registry else None
|
||||
disc = None if args.no_discover else _discover(args, addr.domain)
|
||||
statute = disc.statute if disc is not None and disc.ok else None
|
||||
sem = resolve(addr, statute, registry)
|
||||
syn = addr.to_dict()
|
||||
lines = ["Address:", f" {addr}", "", "Syntax:", *_syntax_lines(syn), "", "Discovery:"]
|
||||
if disc is None:
|
||||
lines.append(" skipped (--no-discover)")
|
||||
elif disc.ok:
|
||||
lines.append(f" valid statute for {disc.host} ({disc.url})")
|
||||
else:
|
||||
lines += [" no valid MAS statute found", f" failure: {disc.kind.value} — {disc.detail}"]
|
||||
lines += ["", "Semantics:", f" kind: {sem.kind}" + (f" (from {sem.kind_source})" if sem.kind_source != "none" else "")]
|
||||
if sem.registry_key:
|
||||
lines.append(f" registry entry: {sem.registry_key}")
|
||||
if sem.interaction and statute is not None and sem.ok:
|
||||
lines.append(f" interaction preference: {sem.interaction} (a request, not a command)")
|
||||
for f in sem.failures:
|
||||
lines.append(f" failure: {f.code} — {f.message}")
|
||||
if sem.unknowns:
|
||||
lines += ["", "Unknowns:"] + [f" - {u}" for u in sem.unknowns]
|
||||
data = {"address": str(addr), "syntax": syn,
|
||||
"discovery": None if disc is None else disc.to_dict(), "semantics": sem.to_dict()}
|
||||
_out(args, data, "\n".join(lines))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_lint(args):
|
||||
worst = 0
|
||||
reports = []
|
||||
for file in args.files:
|
||||
try:
|
||||
doc = json.loads(Path(file).read_text())
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
reports.append({"file": file, "findings": [{"layer": SCHEMA, "severity": ERROR,
|
||||
"code": "unreadable", "path": "", "message": str(e)}]})
|
||||
worst = 1
|
||||
continue
|
||||
findings = lint(doc, args.host)
|
||||
reports.append({"file": file, "findings": [f.__dict__ for f in findings]})
|
||||
if any(f.severity == ERROR for f in findings):
|
||||
worst = 1
|
||||
if args.json:
|
||||
print(json.dumps(reports, indent=2))
|
||||
return worst
|
||||
for rep in reports:
|
||||
fs = rep["findings"]
|
||||
errs = [f for f in fs if f["severity"] == ERROR]
|
||||
print(f"{rep['file']}: {'INVALID' if errs else 'OK'}"
|
||||
f" ({len(errs)} error(s), {len(fs) - len(errs)} warning(s))")
|
||||
for f in fs:
|
||||
label = "warning" if f["severity"] != ERROR else f"{f['layer']} error"
|
||||
print(f" {label} [{f['code']}] {f['path'] or '/'}: {f['message']}")
|
||||
return worst
|
||||
|
||||
|
||||
def cmd_test(args):
|
||||
from mas_conformance import statute_cases, statute_document
|
||||
from mas_conformance.runner import run_parser_corpus
|
||||
rep = run_parser_corpus()
|
||||
st_fail = []
|
||||
for c in statute_cases():
|
||||
fs = lint(statute_document(c["file"]), c["host"])
|
||||
errs = [f for f in fs if f.severity == ERROR]
|
||||
good = (not errs) if c["expect"] == "valid" else (
|
||||
errs and {f.layer for f in errs} == {c["expect"]}
|
||||
and set(c.get("codes", [])) <= {f.code for f in errs})
|
||||
if not good:
|
||||
st_fail.append((c["id"], c["file"], [str(f) for f in errs]))
|
||||
n_st = len(statute_cases())
|
||||
print(f"parser corpus: {rep.passed}/{rep.passed + len(rep.failures)} passed")
|
||||
for fid, lp, why in rep.failures:
|
||||
print(f" FAIL {fid} {lp!r}: {why}")
|
||||
print(f"statute corpus: {n_st - len(st_fail)}/{n_st} passed")
|
||||
for fid, file, why in st_fail:
|
||||
print(f" FAIL {fid} {file}: {why}")
|
||||
return 0 if rep.ok and not st_fail else 1
|
||||
|
||||
|
||||
def build_parser():
|
||||
ap = argparse.ArgumentParser(prog="mas", description=f"Mail Address Semantics v{MAS_VERSION} tools")
|
||||
ap.add_argument("--version", action="version", version=f"mas {__version__} (MAS {MAS_VERSION})")
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
def common(p):
|
||||
p.add_argument("--json", action="store_true", help="machine-readable output")
|
||||
return p
|
||||
|
||||
def disc_opts(p):
|
||||
p.add_argument("--statute", metavar="FILE", help="use a local statute file instead of discovery")
|
||||
p.add_argument("--insecure-http", action="store_true", help=argparse.SUPPRESS)
|
||||
p.add_argument("--port", type=int, help=argparse.SUPPRESS)
|
||||
|
||||
p = common(sub.add_parser("parse", help="show syntax only (no network)"))
|
||||
p.add_argument("address", help="local-part or full address")
|
||||
p.add_argument("--profile", choices=["p1", "p2", "P1", "P2"], help="check a parser conformance class")
|
||||
p.set_defaults(func=cmd_parse)
|
||||
|
||||
p = common(sub.add_parser("discover", help="fetch and validate https://HOST/.well-known/mas"))
|
||||
p.add_argument("host")
|
||||
disc_opts(p)
|
||||
p.set_defaults(func=cmd_discover)
|
||||
|
||||
p = common(sub.add_parser("explain", help="syntax, discovery and semantics, kept separate"))
|
||||
p.add_argument("address")
|
||||
disc_opts(p)
|
||||
p.add_argument("--no-discover", action="store_true", help="skip discovery (syntax-only semantics)")
|
||||
p.add_argument("--registry", metavar="FILE", help="local registry JSON (addresses/resource_types)")
|
||||
p.set_defaults(func=cmd_explain)
|
||||
|
||||
p = common(sub.add_parser("lint-statute", help="schema + semantic validation of statute files"))
|
||||
p.add_argument("files", nargs="+", metavar="FILE")
|
||||
p.add_argument("--host", help="host the statute is meant for (checks namespace)")
|
||||
p.set_defaults(func=cmd_lint)
|
||||
|
||||
p = sub.add_parser("test", help="run the bundled conformance corpus")
|
||||
p.set_defaults(func=cmd_test)
|
||||
return ap
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
Add table
Add a link
Reference in a new issue