coulomb/mail-address-semantics
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MAS-WP-0004: Postfix integration profile, corpus probe in pytest, Exim notes

Browse source 
Profile in integrations/postfix (install.sh + README), now used by the
live deployment. Probe runs every conformance-corpus local-part and the
relay negatives against it; tests/test_postfix_profile.py runs it where
Docker is available.

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
This commit is contained in:
tegwick 2026-10-19 09:25:03 +02:00
parent 082d4c39eb
commit d766d61e7f
12 changed files with 1029 additions and 98 deletions
Download patch file Download diff file Expand all files Collapse all files

51
experiments/postfix-probe/probe.py
View file

@ -4,6 +4,7 @@
python experiments/postfix-probe/probe.py --profile defaults
python experiments/postfix-probe/probe.py --profile defaults-local
python experiments/postfix-probe/probe.py --profile mas-safe-local
python experiments/postfix-probe/probe.py --profile mas-profile --corpus
Starts the mas-postfix-probe container, sends one message per address via
SMTP, then compares the envelope recipient Postfix handed to mas-gateway
@ -38,9 +39,33 @@ CASES = [
("quoted local-part", '"ticket=PR42"'),
("unknown resource", "ticket=DOES-NOT-EXIST"),
]
# Relay attempts: the MAS profile must refuse all of them (not an open relay).
RELAY_NEGATIVES = ["x@other.example", "x!y@other.example", "a@other.example@" + D,
"a@" + D + "@other.example", "x%y@other.example"]
def run(profile, port):
def unquoted(addr):
"""Strip RFC 5321 quoting. Postfix hands ${recipient} over in canonical
form: needless quotes dropped, quotes added around non-dot-atoms."""
local, _, dom = (addr or "").rpartition("@")
if len(local) > 1 and local[0] == local[-1] == '"':
local = local[1:-1]
return f"{local}@{dom}"
def corpus_cases():
"""Every distinct local-part of the shared conformance corpus."""
seen, out = set(), []
for c in json.loads((ROOT / "mas_conformance/addresses.json").read_text()):
if c["local_part"] not in seen:
seen.add(c["local_part"])
out.append((f"corpus {c['id']}", c["local_part"]))
return out
def run(profile, port, corpus=False):
name = f"mas-probe-{profile}"
subprocess.run(["docker", "rm", "-f", name], capture_output=True)
subprocess.run(["docker", "run", "-d", "--name", name, "-e", f"PROFILE={profile}",
@ -55,7 +80,7 @@ def run(profile, port):
except OSError:
time.sleep(0.5)
results, tid = [], None
for label, lp in CASES:
for label, lp in CASES + (corpus_cases() if corpus else []):
rcpt_lp = lp.replace("{ID}", tid or "MISSING")
rcpt = f"{rcpt_lp}@{D}"
msg = (f"From: probe@example.net\r\nTo: {rcpt}\r\nSubject: {label}\r\n"
@ -65,7 +90,7 @@ def run(profile, port):
with smtplib.SMTP("127.0.0.1", port, timeout=10) as s:
refused = s.sendmail("probe@example.net", [rcpt], msg)
smtp = "accepted" if not refused else f"refused {refused}"
except smtplib.SMTPException as e:
except (smtplib.SMTPException, UnicodeEncodeError) as e:
smtp = f"rejected: {e}"
got = wait_event(port, before) if smtp == "accepted" else None
arrived = got["event"]["recipient"] if got else None
@ -75,9 +100,19 @@ def run(profile, port):
tid = app.get("ticket_id")
results.append({"label": label, "sent": rcpt, "smtp": smtp, "arrived_as": arrived,
"unchanged": arrived == rcpt,
"equivalent": arrived is not None and unquoted(arrived) == unquoted(rcpt),
"mas_valid": mas.get("valid"), "kind": mas.get("kind"),
"app_status": got["app_status"] if got else None,
"app": (app or {}).get("status") if isinstance(app, dict) else app})
for rcpt in RELAY_NEGATIVES:
try:
with smtplib.SMTP("127.0.0.1", port, timeout=10) as s:
refused = s.sendmail("probe@example.net", [rcpt], f"Subject: relay\r\n\r\nx\r\n")
smtp = "accepted" if not refused else f"refused {refused}"
except smtplib.SMTPException as e:
smtp = f"rejected: {e}"
results.append({"label": "relay attempt", "sent": rcpt, "smtp": smtp, "arrived_as": None,
"unchanged": False, "relay_refused": smtp != "accepted"})
logs = subprocess.run(["docker", "logs", name], capture_output=True, text=True)
return results, logs.stdout + logs.stderr
finally:
@ -101,16 +136,18 @@ def wait_event(port, before, timeout=15):
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--profile", choices=["defaults", "defaults-local", "mas-safe-local", "local-domain", "local-domain-safe"], default="defaults")
ap.add_argument("--profile", choices=["mas-profile", "defaults", "defaults-local", "mas-safe-local",
"local-domain", "local-domain-safe"], default="mas-profile")
ap.add_argument("--port", type=int, default=2525)
ap.add_argument("--corpus", action="store_true", help="also send every conformance-corpus local-part")
ap.add_argument("--json", metavar="FILE")
ap.add_argument("--log", metavar="FILE")
a = ap.parse_args()
results, log = run(a.profile, a.port)
results, log = run(a.profile, a.port, a.corpus)
for r in results:
mark = "=" if r["unchanged"] else "!"
mark = ("R" if r["relay_refused"] else "X") if "relay_refused" in r else ("=" if r["unchanged"] else "~" if r["equivalent"] else "!")
print(f"{mark} {r['label']:<22} {r['sent']:<52} -> {r['arrived_as'] or r['smtp']}"
f" [{r['kind']}/{r['app']}]")
f" [{r.get('kind')}/{r.get('app')}]")
if a.json:
Path(a.json).write_text(json.dumps({"profile": a.profile, "results": results}, indent=2) + "\n")
if a.log: