coulomb-social/scripts/platform_mail_probe.py

#!/usr/bin/env python3
"""Prepare or explicitly send the two fixed T02 public registration requests."""

import argparse
import hashlib
import json
import secrets
import sys
import urllib.error
import urllib.request
from pathlib import Path

import mail_return_operation as operation
from selected_mailbox_check import TESTS

ENDPOINT = "https://users.coulomb.social/api/v1/public/registrations"


def prepare():
    return {
        "endpoint": ENDPOINT,
        "source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "requests": [
            {
                "idempotency_key": "csoc-t02-" + secrets.token_hex(24),
                "body": {
                    "username": username,
                    "email": email,
                    "display_name": display,
                    "client_id": "coulomb-social",
                    "tenant": "tenant:coulomb",
                },
            }
            for (username, email), display in zip(
                TESTS.items(), ("T02 delivery check", "T02 return check"), strict=True
            )
        ],
    }


def send(packet, packet_path, approve, opener=urllib.request.urlopen):
    operation.require(operation.digest(packet) == approve, "exact_probe_approval_required")
    expected = prepare()
    operation.require(
        packet["endpoint"] == ENDPOINT
        and packet["source_sha256"] == expected["source_sha256"]
        and len(packet["requests"]) == 2,
        "probe_source_or_scope_changed",
    )
    for actual, fixed in zip(packet["requests"], expected["requests"], strict=True):
        operation.require(
            actual["body"] == fixed["body"] and len(actual["idempotency_key"]) == 57,
            "probe_scope_changed",
        )
    # A lost response leaves an explicit intent; inspect the authoritative ledger.
    operation.durable(
        packet_path.with_suffix(".intent.json"), operation.canonical({"digest": approve})
    )
    results = []
    for index, item in enumerate(packet["requests"]):
        request = urllib.request.Request(
            ENDPOINT,
            data=operation.canonical(item["body"]),
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Idempotency-Key": item["idempotency_key"],
            },
        )
        try:
            with opener(request, timeout=45) as response:
                body = response.read(4096)
                operation.require(
                    response.status == 202
                    and json.loads(body) == {"status": "verification_requested"},
                    "probe_outcome_unconfirmed",
                )
                result = {
                    "request_index": index,
                    "status": "verification_requested",
                    "http_status": 202,
                }
        except Exception:
            result = {"request_index": index, "status": "unconfirmed_inspect_before_retry"}
        operation.durable(
            packet_path.with_suffix(".request-" + str(index) + ".json"), operation.canonical(result)
        )
        results.append(result)
        if result["status"] != "verification_requested":
            break
    return {"requests": results, "identity_provisioned": False, "retry_automatically": False}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=("prepare", "send"))
    parser.add_argument("--packet", required=True, type=Path)
    parser.add_argument("--approve")
    args = parser.parse_args()
    operation.private_directory(args.packet.parent)
    if args.action == "prepare":
        packet = prepare()
        operation.durable(args.packet, operation.canonical(packet))
        print(
            json.dumps(
                {
                    "status": "probe_prepared",
                    "digest": operation.digest(packet),
                    "requests": [r["body"] for r in packet["requests"]],
                    "mail_sent": False,
                }
            )
        )
    else:
        print(json.dumps(send(operation.read_json(args.packet), args.packet, args.approve)))


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