mail-address-semantics/mas-v0.1-spec/reference/mas_parser.py

#!/usr/bin/env python3
"""Non-normative MAS v0.1 canonical ASCII reference parser."""

import argparse
import json
import re
from pathlib import Path

ATOM = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789#%&'*-/ ?_`{|}~".replace(" ", ""))
RTYPE = re.compile(r"[a-z0-9]+(?:[_-][a-z0-9]+)*")  # v0.1.1: lower-case, single separators
RESERVED = set("+!$^")

def _valid_dot_string(s, allowed):
    if not s or s.startswith(".") or s.endswith(".") or ".." in s:
        return False
    return all((c == "." or c in allowed) for c in s)

def parse(local_part, profile="p1"):
    if not isinstance(local_part, str) or not local_part:
        raise ValueError("empty local-part")
    try:
        local_part.encode("ascii")
    except UnicodeEncodeError:
        raise ValueError("non-ASCII outside canonical MAS v0.1")

    if '"' in local_part or "\\" in local_part or "@" in local_part:
        raise ValueError("quoted/escaped/full address outside canonical local-part grammar")
    if "$" in local_part or "^" in local_part:
        raise ValueError("reserved operator")

    if local_part.count("+") > 1:
        raise ValueError("multiple facet operators")
    if local_part.count("!") > 1:
        raise ValueError("multiple interaction operators")

    plus = local_part.find("+")
    bang = local_part.find("!")
    if plus >= 0 and bang >= 0 and bang < plus:
        raise ValueError("non-canonical operator order")
    if profile == "p1" and bang >= 0:
        raise ValueError("interaction not enabled in P1")
    if profile not in ("p1", "p2"):
        raise ValueError("unknown profile")

    interaction = None
    s = local_part
    if bang >= 0:
        interaction = s[bang+1:]
        if not interaction:
            raise ValueError("empty interaction")
        s = s[:bang]
        if not _valid_dot_string(interaction, ATOM | set("=")):
            raise ValueError("invalid interaction")

    facet = None
    plus = s.find("+")
    if plus >= 0:
        facet = s[plus+1:]
        if not facet:
            raise ValueError("empty facet")
        s = s[:plus]
        if not _valid_dot_string(facet, ATOM | set("=")):
            raise ValueError("invalid facet")

    target = s
    if not target:
        raise ValueError("empty target")

    out = {
        "target": target,
        "target_kind": "opaque",
        "facet": facet,
        "interaction": interaction,
    }

    if "=" in target:
        rtype, rid = target.split("=", 1)
        if not rtype:
            raise ValueError("empty resource type")
        if not rid:
            raise ValueError("empty resource id")
        if not RTYPE.fullmatch(rtype):
            raise ValueError("invalid resource type")
        if not _valid_dot_string(rid, ATOM | set("=")):
            raise ValueError("invalid resource id")
        out.update({
            "target_kind": "resource",
            "resource_type": rtype,
            "resource_id": rid,
        })
    else:
        if not _valid_dot_string(target, ATOM):
            raise ValueError("invalid opaque target")

    if len(local_part.encode("ascii")) > 64:
        raise ValueError("local-part exceeds 64 octets")

    return out

def self_test():
    here = Path(__file__).resolve().parents[1]
    fixtures = json.loads((here / "conformance" / "addresses.json").read_text())
    failures = []
    for f in fixtures:
        try:
            got = parse(f["local_part"], f["profile"])
            if not f["valid"]:
                failures.append((f["id"], "expected invalid", got))
            elif got != f["parsed"]:
                failures.append((f["id"], "parse mismatch", {"got":got, "expected":f["parsed"]}))
        except ValueError as e:
            if f["valid"]:
                failures.append((f["id"], "unexpected invalid", str(e)))
    if failures:
        print(json.dumps(failures, indent=2))
        raise SystemExit(1)
    print(f"PASS: {len(fixtures)} MAS v0.1 fixtures")

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("local_part", nargs="?")
    ap.add_argument("--profile", choices=["p1","p2"], default="p1")
    ap.add_argument("--self-test", action="store_true")
    args = ap.parse_args()
    if args.self_test:
        self_test()
    elif args.local_part:
        print(json.dumps(parse(args.local_part, args.profile), indent=2))
    else:
        ap.print_help()