"""`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.generator import DESTINATION_PROFILES, CannotExpress, for_destination, generate_local_part from mas_core.model import ResourceTarget 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.stand_ins: lines.append(f" read as: {sem.to_dict()['canonical_local_part']} " f"(v0.2 stand-in: {', '.join(sem.stand_ins)})") if sem.registry_key: lines.append(f" registry entry: {sem.registry_key}") if sem.intake_for: lines.append(f" intake for resource type: {sem.intake_for}") 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 MAX_OCTETS = 64 TIGHT_OCTETS = 16 # fewer left than this: warn, stand-ins and forwarders add octets def _forwarder_needs(statute, local_part): """The forwarder-safe form (spec v0.2 section 12.1): the stand-ins the statute declares, then lower case.""" p = parse_local_part(local_part) needs = [] if statute.alt_resource and isinstance(p.target, ResourceTarget): needs.append("resource-alternate") if statute.alt_interaction and p.interaction is not None: needs.append("interaction-alternate") return tuple(needs) + ("lowercase",) def cmd_compose(args): try: local = generate_local_part(args.target, facet=args.facet, interaction=args.interaction) except MasSyntaxError as e: hint = (" a facet and an interaction are exclusive: pick one, or address the" " resource and state the mode in the message\n" if e.code == "generator-facet-and-interaction" else "") _out(args, {"ok": False, "error": e.code, "message": str(e)}, f"cannot compose [{e.code}]: {e}\n{hint}".rstrip()) return 1 disc = None if args.no_discover else _discover(args, args.domain) statute = disc.statute if disc is not None and disc.ok else None warnings = [] if disc is not None and not disc.ok: warnings.append(f"no valid MAS statute for {args.domain} ({disc.kind.value}); " "the receiver may read this as ordinary email") if statute is not None: sem = resolve(parse_local_part(local), statute) warnings += [f"{f.code}: {f.message}" for f in sem.failures] octets = len(local.encode()) if MAX_OCTETS - octets < TIGHT_OCTETS: warnings.append(f"{octets}/{MAX_OCTETS} octets: little room left for forwarders") forms = {} if statute is not None: wanted = {name: DESTINATION_PROFILES[name] for name in args.destination} if statute.alt_resource or statute.alt_interaction: wanted = {"forwarder-safe": _forwarder_needs(statute, local), **wanted} for name, needs in wanted.items(): try: r = for_destination(local, statute, needs) forms[name] = {"local_part": r.local_part, "address": f"{r.local_part}@{args.domain}", "applied": list(r.applied), "warnings": list(r.warnings)} except CannotExpress as e: forms[name] = {"local_part": None, "error": e.code, "message": str(e)} elif args.destination: warnings.append("destination forms need the statute; none was used") lines = [f"{local}@{args.domain}", f" octets: {octets}/{MAX_OCTETS} (local-part)"] for name, f in forms.items(): if f["local_part"] is None: lines.append(f" {name}: cannot express [{f['error']}]: {f['message']}") else: how = f" ({', '.join(f['applied'])})" if f["applied"] else " (unchanged)" lines.append(f" {name}: {f['address']}{how}") lines += [f" warning: {w}" for w in warnings] _out(args, {"ok": True, "local_part": local, "address": f"{local}@{args.domain}", "octets": octets, "forms": forms, "warnings": warnings}, "\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("compose", help="write the canonical address for a target, and its " "forwarder-safe form")) p.add_argument("domain", help="receiving domain (its statute is discovered)") p.add_argument("target", help="resource TYPE=ID, or a registry name such as 'ticket'") p.add_argument("--facet", help="facet (the part after '+')") p.add_argument("--interaction", metavar="MODE", help="requested interaction mode (P2 '!MODE')") p.add_argument("--for", dest="destination", action="append", default=[], choices=sorted(set(DESTINATION_PROFILES) - {"canonical"}), help="also write the form for this destination (repeatable)") disc_opts(p) p.add_argument("--no-discover", action="store_true", help="skip discovery (canonical form only)") p.set_defaults(func=cmd_compose) 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())