"""Selected, operator-reviewed DSN correlation and auditable global suppression. No network, mailbox scan, message submission, schema change or ledger pruning. This module is invoked by an authorized operator, never an untrusted mail worker. """ from __future__ import annotations import argparse import fcntl import fnmatch import hashlib import json import os import re import sqlite3 import sys from contextlib import closing, contextmanager from datetime import UTC, datetime, timedelta from email import policy from email.parser import BytesParser from pathlib import Path SCHEMA = "coulomb.selected-return-operation.v1" MAX_MESSAGE = 2 * 1024**2 class Refused(RuntimeError): pass def require(condition, code): if not condition: raise Refused(code) def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() def digest(value): return hashlib.sha256(canonical(value)).hexdigest() def timestamp(now): require(now.tzinfo is not None, "timezone_required") return now.astimezone(UTC).isoformat() def permanent_dsn(raw, recipient): """Only a structured RFC 3464 failure is eligible for manual suppression. MIME structure and correlation do not authenticate the sender. The operator must separately attest the selected return's provenance; no automatic action. """ require(0 < len(raw) <= MAX_MESSAGE, "message_size_outside_contract") message = BytesParser(policy=policy.default).parsebytes(raw) require( not message.defects and message.get_content_type() == "multipart/report" and message.get_param("report-type") == "delivery-status", "structured_dsn_required", ) references, failures = set(), [] reports = [ part for part in message.walk() if part.get_content_type() == "message/delivery-status" ] require(len(reports) == 1, "one_delivery_report_required") blocks = reports[0].get_payload() require(isinstance(blocks, list) and len(blocks) >= 2, "recipient_report_required") for block in blocks: require(not block.defects, "defective_report") for reference in block.get_all("Original-Message-ID", []): references.add(str(reference).strip()) if block.get("Final-Recipient") is None: continue require( all(len(block.get_all(k, [])) == 1 for k in ("Final-Recipient", "Action", "Status")), "ambiguous_report", ) kind, separator, address = str(block["Final-Recipient"]).partition(";") require(separator and kind.strip().lower() == "rfc822", "recipient_type_unsupported") failures.append( ( address.strip().lower(), str(block["Action"]).strip().lower(), str(block["Status"]).strip(), ) ) # One selected recipient prevents applying a whole multi-recipient report. require(len(failures) == 1, "single_recipient_report_required") address, action, status = failures[0] require( address == recipient.lower() and action == "failed" and re.fullmatch(r"5\.\d{1,3}\.\d{1,3}", status), "permanent_recipient_failure_required", ) for part in message.walk(): if part.get_content_type() in {"message/rfc822", "text/rfc822-headers"}: attached = part.get_payload() if isinstance(attached, list): for original in attached: references.update(str(v).strip() for v in original.get_all("Message-ID", [])) elif isinstance(attached, str): original = BytesParser(policy=policy.default).parsebytes(attached.encode()) references.update(str(v).strip() for v in original.get_all("Message-ID", [])) require(len(references) == 1, "unambiguous_original_message_id_required") reference = references.pop() require(re.fullmatch(r"<[^<>\s]{1,250}>", reference), "original_message_id_invalid") return reference def selected_policy(document, caller_id): require(document.get("schema_version") == 1, "policy_schema_changed") callers = [p for p in document["callers"] if p.get("caller_id") == caller_id] require(len(callers) == 1 and caller_id == "user-engine", "reviewed_caller_required") return callers[0] def binding_allowed(caller, tenant, client): require( all( isinstance(v, str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", v) for v in (tenant, client) ), "invalid_binding", ) return [tenant, client] in caller["bindings"] or any( fnmatch.fnmatchcase(tenant, t) and fnmatch.fnmatchcase(client, c) for t, c in caller.get("binding_patterns", []) ) def correlate(db, reference, recipient, caller_id, document, invitation_payload=None): caller = selected_policy(document, caller_id) # The stored SMTP Message-ID, not the caller-facing delivery reference, is # authoritative. Reject duplicate references across either send mechanism. verification = db.execute( "SELECT v.request_id,v.email,v.tenant,v.client_id,v.caller_id FROM " "verification_attempts a JOIN verifications v ON a.request_id=v.request_id " "WHERE a.provider_ref=? AND a.state='accepted'", (reference,), ).fetchall() invitation = db.execute( "SELECT d.event_id,a.fingerprint FROM deliveries d JOIN send_attempts a " "ON a.event_id=d.event_id WHERE d.provider_ref=?", (reference,), ).fetchall() require(len(verification) + len(invitation) == 1, "one_accepted_outbound_required") if verification: request_id, email, tenant, client, recorded_caller = verification[0] require( email.lower() == recipient.lower() and ( recorded_caller == caller_id or (recorded_caller == "" and caller.get("adopt_legacy_verifications") is True) ), "recipient_or_caller_mismatch", ) require(binding_allowed(caller, tenant, client), "caller_binding_denied") return { "kind": "verification", "id": request_id, "caller_id": caller_id, "tenant": tenant, "client_id": client, "recipient": email.lower(), "provider_ref": reference, } # Invitation rows retain a fingerprint, not plaintext recipient/binding. # Require the original caller payload and reproduce the admitted mapping. require(isinstance(invitation_payload, dict), "original_invitation_payload_required") payload = dict(invitation_payload) tenant, client = payload.get("tenant"), payload.get("client_id") if (tenant, client) == (None, None) and caller.get("legacy_invitation_binding"): tenant, client = caller["legacy_invitation_binding"] require(binding_allowed(caller, tenant, client), "caller_binding_denied") require( payload.get("source") == caller["source"] and payload.get("type") in caller["templates"] and payload.get("data", {}).get("primary_email", "").lower() == recipient.lower(), "invitation_scope_mismatch", ) event = payload.get("id") require( isinstance(event, str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", event) and not event.startswith("ec2_"), "invalid_event_id", ) key = ( event if (caller.get("adopt_legacy_invitations") or caller.get("legacy_invitation_binding")) else ("ec2_" + hashlib.sha256(json.dumps([caller_id, event]).encode()).hexdigest()) ) payload.update(id=key, source="user-engine") candidates = [digest(payload)] if caller.get("adopt_legacy_invitations"): legacy = dict(payload) legacy.pop("client_id", None) candidates.append(digest(legacy)) require( invitation[0][0] == key and invitation[0][1] in candidates, "invitation_fingerprint_mismatch", ) return { "kind": "invitation", "id": key, "caller_id": caller_id, "tenant": tenant, "client_id": client, "recipient": recipient.lower(), "provider_ref": reference, } def connect(path, *, writable=False): path = Path(path).absolute() require( path.is_file() and not any(p.is_symlink() for p in (path, *path.parents)), "real_existing_ledger_required", ) db = sqlite3.connect( path.as_uri() + ("?mode=rw" if writable else "?mode=ro"), uri=True, timeout=10 ) db.execute("PRAGMA trusted_schema=OFF") if not writable: db.execute("PRAGMA query_only=ON") return db def suppression(db, recipient): row = db.execute( "SELECT email,reason,created_at FROM suppressions WHERE email=?", (recipient,) ).fetchone() return list(row) if row else None def prepare(path, raw, recipient, document, trust_note, *, invitation_payload=None, now=None): now = now or datetime.now(UTC) require( isinstance(trust_note, str) and 10 <= len(trust_note.strip()) <= 500, "explicit_operator_provenance_review_required", ) reference = permanent_dsn(raw, recipient) with closing(connect(path)) as db: db.execute("BEGIN") bound = correlate(db, reference, recipient, "user-engine", document, invitation_payload) before = suppression(db, bound["recipient"]) require(before is None, "existing_suppression_requires_review") return { "schema": SCHEMA, "action": "suppress", "created_at": timestamp(now), "binding": bound, "policy_sha256": digest(document), "evidence_sha256": hashlib.sha256(raw).hexdigest(), "operator_review": trust_note, "invitation_payload": invitation_payload, "before": before, "after": [bound["recipient"], "hard_bounce", timestamp(now)], "scope": "recipient-global across callers and tenants", } def private_directory(path): path = Path(path) require( path.is_absolute() and path.is_dir() and not any(p.is_symlink() for p in (path, *path.parents)), "private_directory_required", ) info = path.stat() require(info.st_uid == os.getuid() and not info.st_mode & 0o077, "private_directory_required") def sync_directory(path): fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) def durable(path, raw): fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, "wb") as stream: stream.write(raw) stream.flush() os.fsync(stream.fileno()) sync_directory(path.parent) def read_json(path): info = path.lstat() require( not path.is_symlink() and path.is_file() and info.st_uid == os.getuid() and not info.st_mode & 0o077, "private_regular_receipt_required", ) require(info.st_size <= MAX_MESSAGE, "receipt_size_outside_contract") return json.loads(path.read_text()) def completed_operation(root, value, status): require(re.fullmatch(r"[a-f0-9]{64}", value), "operation_id_invalid") directory = Path(root) / value private_directory(directory) plan = read_json(directory / "plan.json") require(digest(plan) == value, "retained_plan_changed") completed = read_json(directory / "completed.json") require( completed.get("plan_sha256") == value and completed.get("status") == status, "completion_receipt_changed", ) return plan def prepare_release(path, audit_root, source, review, *, now=None): require( isinstance(review, str) and 10 <= len(review.strip()) <= 500, "explicit_release_review_required", ) private_directory(audit_root) original = completed_operation(audit_root, source, "suppression_applied") require( original.get("schema") == SCHEMA and original.get("action") == "suppress", "original_suppression_required", ) with closing(connect(path)) as db: require( suppression(db, original["binding"]["recipient"]) == original["after"], "suppression_changed", ) return { "schema": SCHEMA, "action": "release", "created_at": timestamp(now or datetime.now(UTC)), "source_plan_sha256": source, "binding": original["binding"], "operator_review": review, "policy_sha256": original["policy_sha256"], "before": original["after"], "after": None, "scope": original["scope"], } @contextmanager def exclusive(directory): private_directory(directory) fd = os.open(directory / "operation.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) try: try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise Refused("operation_still_running") from None yield finally: os.close(fd) def apply(path, plan, raw, document, expected, audit_root): require( plan.get("schema") == SCHEMA and plan.get("action") in {"suppress", "release"} and expected == digest(plan), "reviewed_plan_required", ) private_directory(audit_root) if plan["action"] == "suppress": require(digest(document) == plan["policy_sha256"], "caller_policy_changed") require( hashlib.sha256(raw).hexdigest() == plan["evidence_sha256"], "selected_evidence_changed" ) require( permanent_dsn(raw, plan["binding"]["recipient"]) == plan["binding"]["provider_ref"], "dsn_correlation_changed", ) else: original = completed_operation( audit_root, plan["source_plan_sha256"], "suppression_applied" ) require( plan["binding"] == original["binding"] and plan["before"] == original["after"] and plan["after"] is None, "release_delta_invalid", ) directory = Path(audit_root) / expected directory.mkdir(mode=0o700) # A prior intent must be inspected, never replayed. sync_directory(audit_root) with exclusive(directory): return commit(path, plan, raw, document, expected, directory) def commit(path, plan, raw, document, expected, directory): durable(directory / "plan.json", canonical(plan) + b"\n") if plan["action"] == "suppress": durable(directory / "message.eml", raw) with closing(connect(path, writable=True)) as db: try: db.execute("BEGIN IMMEDIATE") recipient = plan["binding"]["recipient"] if plan["action"] == "suppress": bound = correlate( db, plan["binding"]["provider_ref"], recipient, plan["binding"]["caller_id"], document, plan["invitation_payload"], ) require(bound == plan["binding"], "accepted_outbound_changed") require( plan["before"] is None and plan["after"] == [recipient, "hard_bounce", plan["created_at"]], "suppression_delta_invalid", ) require(suppression(db, recipient) == plan["before"], "suppression_changed") durable( directory / "intent.json", canonical( { "plan_sha256": expected, "status": "commit_unconfirmed", "started_at": timestamp(datetime.now(UTC)), } ), ) if plan["action"] == "suppress": db.execute( "INSERT INTO suppressions (email,reason,created_at) VALUES (?,?,?)", plan["after"], ) else: db.execute("DELETE FROM suppressions WHERE email=?", (recipient,)) db.commit() except BaseException: db.rollback() raise status = "suppression_applied" if plan["action"] == "suppress" else "suppression_released" durable( directory / "completed.json", canonical( { "plan_sha256": expected, "status": status, "completed_at": timestamp(datetime.now(UTC)), } ), ) return {"status": status, "plan_sha256": expected, "scope": plan["scope"], "mail_sent": False} def inspect(path, directory): private_directory(directory) plan = read_json(Path(directory) / "plan.json") require(digest(plan) == Path(directory).name, "retained_plan_changed") with closing(connect(path)) as db: current = suppression(db, plan["binding"]["recipient"]) state = ( "planned_state_present" if current == plan["after"] else "original_state_present" if current == plan["before"] else "later_state_requires_review" ) return { "plan_sha256": digest(plan), "current_state": state, "commit_receipt_present": (Path(directory) / "completed.json").is_file(), "retry_automatically": False, } def reconcile(path, directory, expected): """Verify a stopped attempt's actual state; never repeat its SQL mutation.""" directory = Path(directory) require(directory.name == expected, "reviewed_plan_required") with exclusive(directory), closing(connect(path, writable=True)) as db: plan = read_json(directory / "plan.json") require(digest(plan) == expected, "retained_plan_changed") require(not (directory / "completed.json").exists(), "already_completed") db.execute("BEGIN IMMEDIATE") try: current = suppression(db, plan["binding"]["recipient"]) if current == plan["after"]: intent = read_json(directory / "intent.json") require(intent.get("plan_sha256") == expected, "commit_intent_required") status = ( "suppression_applied" if plan["action"] == "suppress" else "suppression_released" ) elif current == plan["before"]: status = "operation_not_applied" else: raise Refused("later_state_requires_review") completed = { "plan_sha256": expected, "status": status, "completed_at": timestamp(datetime.now(UTC)), "reconciled_state_verified": True, "sql_mutated": False, } durable(directory / "completed.json", canonical(completed)) return completed finally: db.rollback() def expire_raw(audit_root, *, now, days=30, execute=True): """Expire only copied raw evidence; never SQL, decision metadata or source mail. Active suppressions retain their binding/review records. Their release is a separate explicit operator decision. Ambiguous/incomplete operations stay held. """ require(days == 30, "reviewed_retention_required") timestamp(now) private_directory(audit_root) cutoff = now - timedelta(days=days) expired = [] for directory in sorted(Path(audit_root).iterdir()): private_directory(directory) require(re.fullmatch(r"[a-f0-9]{64}", directory.name), "unknown_audit_entry") if not (directory / "plan.json").exists(): # A stopped attempt can die after mkdir and before its first plan. # Hold that directory; it establishes no authority to remove files. continue plan = read_json(directory / "plan.json") require(digest(plan) == directory.name, "retained_plan_changed") raw = directory / "message.eml" if not (directory / "completed.json").is_file() or not raw.exists(): continue completed = read_json(directory / "completed.json") require( completed.get("plan_sha256") == directory.name and completed.get("status") in {"suppression_applied", "operation_not_applied"}, "completion_receipt_changed", ) if datetime.fromisoformat(completed["completed_at"]) > cutoff: continue require(not raw.is_symlink() and raw.is_file(), "real_evidence_copy_required") require( hashlib.sha256(raw.read_bytes()).hexdigest() == plan["evidence_sha256"], "retained_evidence_changed", ) if not execute: expired.append(directory.name) continue # Durable intent precedes unlink; interruption is reconciled from that # intent and file presence, without touching the original mailbox/file. intent = directory / "expiry.json" if not intent.exists(): durable( intent, canonical( { "raw_sha256": plan["evidence_sha256"], "expired_at": timestamp(now), "retention_days": days, } ), ) else: require( read_json(intent)["raw_sha256"] == plan["evidence_sha256"], "expiry_intent_changed" ) raw.unlink() fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) expired.append(directory.name) return { "expired_raw_copies": expired, "ledger_modified": False, "decision_metadata_retained": True, "source_mail_modified": False, } def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "action", choices=("prepare", "release", "apply", "inspect", "reconcile", "expire-raw") ) parser.add_argument("--ledger", type=Path) parser.add_argument("--caller-policy", type=Path) parser.add_argument("--message", type=Path) parser.add_argument("--recipient") parser.add_argument("--review") parser.add_argument("--invitation-payload", type=Path) parser.add_argument("--packet", type=Path) parser.add_argument("--audit-root", type=Path) parser.add_argument("--operation") parser.add_argument("--expect-digest") parser.add_argument("--execute", action="store_true", help="Execute the reviewed raw expiry") parser.add_argument("--as-of", help="Use the timestamp returned by an expiry preview") args = parser.parse_args(argv) require(__debug__, "optimized_python_refused") if args.action == "expire-raw": require(args.audit_root is not None, "audit_root_required") now = datetime.fromisoformat(args.as_of) if args.as_of else datetime.now(UTC) require( now.tzinfo is not None and now <= datetime.now(UTC), "observed_expiry_time_required" ) preview = expire_raw(args.audit_root, now=now, execute=False) expiry_digest = digest({"as_of": timestamp(now), "copies": preview["expired_raw_copies"]}) if args.execute: require(args.expect_digest == expiry_digest, "reviewed_expiry_required") preview = expire_raw(args.audit_root, now=now) print( json.dumps( { **preview, "as_of": timestamp(now), "expiry_digest": expiry_digest, "executed": args.execute, } ) ) return 0 require(args.ledger is not None, "ledger_required") if args.action in {"prepare", "release"}: require(args.packet is not None, "private_new_packet_required") private_directory(args.packet.parent) raw = None if args.action == "prepare": require( args.message is not None and args.caller_policy is not None and args.recipient is not None, "selected_message_recipient_and_policy_required", ) require( args.message.is_file() and not args.message.is_symlink() and args.message.stat().st_size <= MAX_MESSAGE, "bounded_selected_file_required", ) raw = args.message.read_bytes() document = json.loads(args.caller_policy.read_text()) invitation = ( json.loads(args.invitation_payload.read_text()) if (args.invitation_payload is not None) else None ) plan = prepare( args.ledger, raw, args.recipient, document, args.review, invitation_payload=invitation, ) else: require( args.audit_root is not None and args.operation is not None, "source_suppression_operation_required", ) plan = prepare_release(args.ledger, args.audit_root, args.operation, args.review) packet = { "plan": plan, "plan_sha256": digest(plan), "source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), } durable(args.packet, canonical(packet) + b"\n") print( json.dumps( { "status": "return_operation_prepared", "plan_sha256": digest(plan), "scope": plan["scope"], "ledger_modified": False, } ) ) return 0 require(args.audit_root is not None, "audit_root_required") private_directory(args.audit_root) if args.action in {"inspect", "reconcile"}: require( args.operation is not None and re.fullmatch(r"[a-f0-9]{64}", args.operation), "operation_id_invalid", ) directory = args.audit_root / args.operation if args.action == "inspect": result = inspect(args.ledger, directory) else: require(args.expect_digest == args.operation, "reviewed_plan_required") result = reconcile(args.ledger, directory, args.expect_digest) else: require(args.packet is not None, "private_packet_required") private_directory(args.packet.parent) packet = read_json(args.packet) require( packet["source_sha256"] == hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "reviewed_source_changed", ) require(packet["plan_sha256"] == args.expect_digest, "reviewed_plan_required") plan = packet["plan"] raw = None if plan["action"] == "suppress": require( args.message is not None and args.message.is_file() and not args.message.is_symlink() and args.message.stat().st_size <= MAX_MESSAGE, "bounded_selected_file_required", ) raw = args.message.read_bytes() require( plan["action"] == "release" or args.caller_policy is not None, "caller_policy_required" ) document = json.loads(args.caller_policy.read_text()) if args.caller_policy else None result = apply(args.ledger, plan, raw, document, args.expect_digest, args.audit_root) print(json.dumps(result)) return 0 if __name__ == "__main__": try: sys.exit(main()) except Exception as error: # Do not echo MIME, recipient values, payloads or SQL errors into logs. reason = str(error) if isinstance(error, Refused) else type(error).__name__ print( json.dumps( { "status": "return_operation_requires_inspection", "reason": reason, "retry_automatically": False, } ) ) sys.exit(1)