coulomb-social/scripts/selected_mailbox_check.py

#!/usr/bin/env python3
"""Explicit selected test-mail retrieval; never a mailbox scanner or mail sender."""

from __future__ import annotations

import argparse
import base64
import hashlib
import imaplib
import ipaddress
import json
import os
import re
import shlex
import socket
import ssl
import subprocess
import sys
import uuid
from email import policy
from email.parser import BytesParser
from pathlib import Path

import mail_return_native as native
import mail_return_operation as operation

require = operation.require
HOST = "imap.ionos.de"
MAX_BYTES = 2 * 1024**2
FILES = (*native.FILES, "selected_mailbox_check.py")
TESTS = {
    "csoc-t02-delivery-20261102": "bernd.worsch@gmail.com",
    "csoc-t02-bounce-20261102": "csoc-t02-20261102@delivery-check.invalid",
}


def selected(raw, reference, recipient):
    message = BytesParser(policy=policy.default).parsebytes(raw)
    replies = " ".join(
        str(v) for key in ("In-Reply-To", "References") for v in message.get_all(key, [])
    )
    if reference in re.findall(r"<[^<>\s]+>", replies):
        return "reply"
    try:
        if operation.permanent_dsn(raw, recipient) == reference:
            return "permanent_dsn"
    except operation.Refused:
        pass
    return None


def retrieve(reference, recipient, username, password, client=imaplib.IMAP4_SSL):
    require(re.fullmatch(r"<[^<>\s\"\\]{1,250}>", reference), "invalid_message_id")
    mailbox = client(HOST, 993, ssl_context=ssl.create_default_context(), timeout=20)
    try:
        require(mailbox.login(username, password)[0] == "OK", "imap_login_failed")
        require(mailbox.select("INBOX", readonly=True)[0] == "OK", "readonly_inbox_failed")
        # TEXT searches this exact retained Message-ID, including DSN attachments.
        status, data = mailbox.uid("search", None, "TEXT", '"' + reference + '"')
        require(status == "OK" and len(data) == 1, "selected_search_failed")
        uids = data[0].split()
        require(
            len(uids) <= 5 and all(re.fullmatch(rb"[1-9][0-9]*", u) for u in uids),
            "selected_search_not_bounded",
        )
        result = []
        for uid in uids:
            status, size = mailbox.uid("fetch", uid, "(RFC822.SIZE)")
            sizes = re.findall(rb"RFC822.SIZE ([0-9]+)", b" ".join(size))
            require(
                status == "OK" and len(sizes) == 1 and 0 < int(sizes[0]) <= MAX_BYTES,
                "selected_message_too_large",
            )
            status, parts = mailbox.uid("fetch", uid, "(BODY.PEEK[])")
            bodies = [p[1] for p in parts if isinstance(p, tuple) and len(p) == 2]
            require(
                status == "OK" and len(bodies) == 1 and len(bodies[0]) == int(sizes[0]),
                "selected_fetch_unconfirmed",
            )
            raw = bodies[0]
            kind = selected(raw, reference, recipient)
            if kind:
                result.append(
                    {
                        "uid": uid.decode(),
                        "kind": kind,
                        "sha256": hashlib.sha256(raw).hexdigest(),
                        "raw_base64": base64.b64encode(raw).decode(),
                    }
                )
        return result
    finally:
        # Read-only EXAMINE + PEEK; never STORE, EXPUNGE, CLOSE, APPEND or SEND.
        try:
            mailbox.logout()
        except Exception:
            pass


POD = r"""
import hashlib,json,os,sqlite3,sys
from pathlib import Path
sys.dont_write_bytecode=True
r=json.load(sys.stdin)
assert __debug__ and os.getuid()==10001
from email_connect import transactional,callers
for module,key in ((transactional,'transactional'),(callers,'callers')):
 assert hashlib.sha256(Path(module.__file__).read_bytes()).hexdigest()==r[key]
assert os.environ.get('EMAIL_CONNECT_DATABASE_PATH')=='/data/email-connect.db'
assert os.environ.get('EMAIL_CONNECT_SMTP_USERNAME')=='platform@coulomb.social'
assert os.environ.get('EMAIL_CONNECT_SENDER')=='platform@coulomb.social'
assert os.environ.get('EMAIL_CONNECT_CALLERS_FILE')=='/etc/email-connect-callers/callers.json'
import types
for name in ('mail_return_operation','mail_return_native','selected_mailbox_check'):
 m=types.ModuleType(name);m.__file__='/reviewed/'+name+'.py';sys.modules[name]=m
 exec(compile(r['sources'][name+'.py'],'<reviewed-'+name+'>','exec'),m.__dict__)
core=sys.modules['mail_return_operation']
policy_path=Path('/etc/email-connect-callers/callers.json')
assert policy_path.stat().st_size<=64*1024
document=json.loads(policy_path.read_text())
policy_sha256=core.digest(document)
if r.get('policy_sha256') is not None:assert policy_sha256==r['policy_sha256']
caller_policy=core.selected_policy(document,'user-engine')
db=sqlite3.connect('file:/data/email-connect.db?mode=ro',uri=True)
try:
 rows=db.execute("SELECT v.username,v.email,v.caller_id,v.tenant,v.client_id FROM "
  "verification_attempts a JOIN verifications v ON a.request_id=v.request_id "
  "WHERE a.provider_ref=? AND a.state='accepted'",(r['reference'],)).fetchall()
 assert len(rows)==1
 username,recipient,caller,tenant,client=rows[0]
 assert r['tests'].get(username)==recipient
 assert (tenant,client)==('tenant:coulomb','coulomb-social')
 assert core.binding_allowed(caller_policy,tenant,client)
 assert caller=='user-engine' or (
  caller=='' and caller_policy.get('adopt_legacy_verifications') is True)
finally:db.close()
if r['action']=='binding':
 result={'accepted_test_binding_verified':True,'recipient':recipient,
  'caller_policy_sha256':policy_sha256,'legacy_adoption':caller==''}
else:
 result={'messages':m.retrieve(r['reference'],recipient,'platform@coulomb.social',
  os.environ['EMAIL_CONNECT_SMTP_PASSWORD'])}
print(json.dumps(result))
"""


def policy_for(name, addresses):
    require(
        addresses
        and len(addresses) <= 8
        and all(
            ipaddress.ip_address(a).version == 4 and ipaddress.ip_address(a).is_global
            for a in addresses
        ),
        "public_provider_addresses_required",
    )
    return {
        "apiVersion": "networking.k8s.io/v1",
        "kind": "NetworkPolicy",
        "metadata": {
            "name": name,
            "namespace": "email-connect",
            "annotations": {"coulomb.social/selected-mail-check": name},
        },
        "spec": {
            "podSelector": {"matchLabels": {"app.kubernetes.io/name": "email-connect"}},
            "policyTypes": ["Egress"],
            "egress": [
                {
                    "to": [{"ipBlock": {"cidr": a + "/32"}} for a in addresses],
                    "ports": [{"protocol": "TCP", "port": 993}],
                }
            ],
        },
    }


def run_node(request, command=native.native_command):
    action = request["action"]
    current = native.identity(command)
    if action == "prepare":
        addresses = sorted(
            {a[4][0] for a in socket.getaddrinfo(HOST, 993, socket.AF_INET, socket.SOCK_STREAM)}
        )
        name = "email-connect-selected-" + request["nonce"]
        pod_result = command(
            [
                "exec",
                "-i",
                "-n",
                "email-connect",
                current["pod_name"],
                "--",
                "python3",
                "-I",
                "-c",
                POD,
            ],
            operation.canonical(
                {
                    "action": "binding",
                    "reference": request["reference"],
                    "tests": TESTS,
                    "sources": request["sources"],
                    "transactional": native.TRANSACTIONAL,
                    "callers": native.CALLERS,
                }
            ),
        )
        require(native.identity(command) == current, "receiver_changed")
        return {
            "context": current,
            "reference": request["reference"],
            "policy": policy_for(name, addresses),
            "binding": pod_result,
        }
    packet = request["packet"]
    wanted = packet["policy"]
    name = wanted["metadata"]["name"]
    require(re.fullmatch(r"email-connect-selected-[a-f0-9]{16}", name), "invalid_policy_name")
    require(
        wanted
        == policy_for(
            name,
            [t["ipBlock"]["cidr"].removesuffix("/32") for t in wanted["spec"]["egress"][0]["to"]],
        ),
        "policy_scope_changed",
    )

    def existing():
        items = command(["get", "networkpolicies", "-n", "email-connect", "-o", "json"])["items"]
        found = [v for v in items if v["metadata"]["name"] == name]
        require(len(found) <= 1, "policy_ambiguous")
        return found[0] if found else None

    def cleanup():
        obj = existing()
        if obj is not None:
            require(
                obj["spec"] == wanted["spec"]
                and obj["metadata"].get("annotations") == wanted["metadata"]["annotations"],
                "policy_ownership_changed",
            )
            options = {
                "apiVersion": "v1",
                "kind": "DeleteOptions",
                "preconditions": {"uid": obj["metadata"]["uid"]},
            }
            command(
                [
                    "delete",
                    "--raw=/apis/networking.k8s.io/v1/namespaces/email-connect/"
                    "networkpolicies/" + name,
                    "-f",
                    "/dev/stdin",
                ],
                operation.canonical(options),
            )
        require(existing() is None, "policy_cleanup_unconfirmed")

    if action == "cleanup":
        require(
            all(
                current[k] == packet["context"][k]
                for k in ("cluster_uid", "namespace_uid", "deployment_uid", "image")
            ),
            "identity_changed",
        )
        cleanup()
        return {"temporary_policy_absent": True, "mailbox_accessed": False}
    require(action == "fetch" and current == packet["context"], "reviewed_receiver_changed")
    require(existing() is None, "prior_attempt_requires_cleanup")
    try:
        created = command(["create", "-f", "/dev/stdin", "-o", "json"], operation.canonical(wanted))
        require(created["spec"] == wanted["spec"], "policy_creation_unconfirmed")
        result = command(
            [
                "exec",
                "-i",
                "-n",
                "email-connect",
                current["pod_name"],
                "--",
                "python3",
                "-I",
                "-c",
                POD,
            ],
            operation.canonical(
                {
                    "action": "fetch",
                    "reference": packet["reference"],
                    "tests": TESTS,
                    "sources": request["sources"],
                    "policy_sha256": packet["binding"]["caller_policy_sha256"],
                    "transactional": native.TRANSACTIONAL,
                    "callers": native.CALLERS,
                }
            ),
        )
        require(native.identity(command) == current, "receiver_changed")
        return {**result, "temporary_policy_absent": True}
    finally:
        cleanup()


BOOTSTRAP = r"""
import hashlib,json,sys,tempfile
from pathlib import Path
sys.dont_write_bytecode=True
r=json.load(sys.stdin)
assert set(r['sources'])=={
 'mail_return_native.py','mail_return_operation.py','selected_mailbox_check.py'}
actual={n:hashlib.sha256(s.encode()).hexdigest() for n,s in r['sources'].items()}
assert actual==r['source_sha256']
with tempfile.TemporaryDirectory(prefix='selected-mail-check-') as d:
 for name,source in r['sources'].items():
  p=Path(d)/name;p.write_text(source);p.chmod(0o600)
 sys.path.insert(0,d)
 import selected_mailbox_check as check
 try:print(json.dumps(check.run_node(r)))
 except Exception:
  print(json.dumps({'status':'selected_check_unconfirmed','inspect_policy_before_retry':True}))
  sys.exit(1)
"""


def remote(request):
    sources = {name: (Path(__file__).parent / name).read_text() for name in FILES}
    hashes = {n: hashlib.sha256(s.encode()).hexdigest() for n, s in sources.items()}
    if "packet" in request:
        require(request["packet"]["source_sha256"] == hashes, "reviewed_source_changed")
    result = subprocess.run(
        [
            "ssh",
            "-o",
            "BatchMode=yes",
            "-o",
            "ConnectTimeout=10",
            "railiance01",
            "sudo -n python3 -I -c " + shlex.quote(BOOTSTRAP),
        ],
        input=operation.canonical({**request, "sources": sources, "source_sha256": hashes}),
        capture_output=True,
        timeout=180,
        check=False,
    )
    require(result.returncode == 0, "selected_check_unconfirmed_cleanup_before_retry")
    return {**json.loads(result.stdout), "source_sha256": hashes}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=("prepare", "fetch", "cleanup"))
    parser.add_argument("--packet", type=Path, required=True)
    parser.add_argument("--reference")
    parser.add_argument("--approve")
    args = parser.parse_args()
    operation.private_directory(args.packet.parent)
    if args.action == "prepare":
        require(
            args.reference and re.fullmatch(r"<[^<>\s\"\\]{1,250}>", args.reference),
            "valid_message_id_required",
        )
        result = remote(
            {"action": "prepare", "reference": args.reference, "nonce": uuid.uuid4().hex[:16]}
        )
        operation.durable(args.packet, operation.canonical(result))
        print(
            json.dumps(
                {
                    "status": "selected_check_prepared",
                    "digest": operation.digest(result),
                    "policy": result["policy"],
                    "mailbox_accessed": False,
                }
            )
        )
        return
    packet = operation.read_json(args.packet)
    require(args.approve == operation.digest(packet), "exact_packet_approval_required")
    if args.action == "fetch":
        # Persist an exclusive attempt before any network-policy or mailbox operation.
        operation.durable(
            args.packet.with_suffix(".intent.json"),
            operation.canonical({"digest": args.approve, "action": "fetch"}),
        )
    result = remote({"action": args.action, "packet": packet})
    metadata = []
    for item in result.pop("messages", []):
        raw = base64.b64decode(item.pop("raw_base64"), validate=True)
        require(
            len(raw) <= MAX_BYTES and hashlib.sha256(raw).hexdigest() == item["sha256"],
            "selected_raw_changed",
        )
        destination = args.packet.parent / (item["sha256"] + ".eml")
        if destination.exists():
            require(
                not destination.is_symlink()
                and destination.stat().st_uid == os.getuid()
                and not destination.stat().st_mode & 0o077
                and destination.read_bytes() == raw,
                "private_evidence_changed",
            )
        else:
            operation.durable(destination, raw)
        metadata.append(item)
    result["selected_messages"] = metadata
    operation.durable(
        args.packet.with_suffix("." + args.action + ".json"), operation.canonical(result)
    )
    print(json.dumps(result))


if __name__ == "__main__":
    try:
        main()
    except Exception as error:
        print(
            json.dumps(
                {
                    "status": "selected_check_refused_or_unconfirmed",
                    "error_type": type(error).__name__,
                    "retry_automatically": False,
                    "inspect_policy_before_retry": True,
                }
            )
        )
        sys.exit(1)