Async Routing for MPC: A Procedural Implementation Guide

Asynchronous routing for Multi-Party Computation (MPC) decouples cryptographic round progression from network latency, letting privacy-preserving spatial analytics run reliably across distributed healthcare, financial, and GIS infrastructure where nodes are rarely online at the same instant. Instead of a blocking broadcast channel that stalls the entire protocol whenever one party drops a packet, an async router pushes encrypted coordinate shares, computation directives, and reconstruction triggers through a durable message queue so each participant can submit and consume work independently. This guide is part of the Secure Multi-Party Computation in Spatial Analytics reference set and assumes you already mask and split coordinates before they reach the wire.

Problem framing: routing MPC rounds across nodes that are never simultaneously online

The concrete scenario this page solves is a $t$-of-$n$ secure computation — say a proximity join between two hospital networks and a regional registry — where the three custodians sit behind different SLAs, NAT boundaries, and cross-border data-residency rules. A synchronous protocol requires all $n$ parties to complete a handshake inside one round-trip window; in practice cellular backhaul, maintenance windows, and regulator-mandated regional broker isolation make that window impossible to guarantee. The result is a protocol that deadlocks on the slowest link.

Async routing reframes each MPC round as a set of queue operations. A producer splits a coordinate into shares, frames and authenticates each share, and publishes it to a session-scoped exchange. Consumers on the compute nodes pull shares whenever they are online, verify integrity, buffer until the reconstruction threshold $t$ is met, and only then invoke the local MPC kernel. Cryptographic guarantees are preserved end to end — raw geometry never traverses the routing layer in plaintext — while throughput for spatial joins, proximity queries, and aggregation pipelines is no longer hostage to the worst node on the call.

Prerequisites

Before wiring the router, pin the following dependencies and assumptions. The routing layer sits downstream of masking and upstream of reconstruction, so its inputs must already be privacy-safe.

• Python 3.11+ with pika (AMQP 0-9-1 client), cryptography (HMAC, EC key generation), and the standard-library hmac, hashlib, and struct modules. Wire framing is portable binary, so no pickle.
• A durable broker with per-session ordering — RabbitMQ, Apache Kafka, or Redis Streams. The examples use RabbitMQ quorum queues; whichever broker you pick must enforce TLS 1.3 in transit and at-rest encryption for queued payloads.
• Coordinates already masked and split. Generalize and noise-inject raw geometry with the coordinate masking protocols first, then decompose each point into additive or Shamir shares per the secret sharing for coordinates guide. The router transports shares; it does not create them.
• A privacy-budget accounting decision. If downstream nodes run encrypted aggregation, their evaluation contexts must tolerate queue-induced latency — review homomorphic encryption basics to size noise budgets that survive multi-hop delays without decryption failure.
• An out-of-band channel for the session HMAC key. Producers and consumers share one symmetric key per session, distributed over the same secure channel that publishes session_params — never over the routing queue itself.

This pattern is the secure-computation analogue of the async execution patterns used in federated learning: same staleness-tolerant dispatch, but transporting cryptographic shares rather than model gradients.

Step-by-step procedure

Step 1: Cryptographic parameter initialization and queue provisioning

Provision a durable broker topology with strict ordering per MPC session ID. Generate session-specific parameters — threshold scheme, elliptic-curve group, nonce range — and bind every participating node to a routing channel tagged with a cryptographic session hash. Configure a dead-letter queue (DLQ) for malformed frames and set a message TTL aligned with your round-timeout budget so orphaned shares cannot accumulate.

import os
import uuid
from cryptography.hazmat.primitives.asymmetric import ec
from pika import BlockingConnection, ConnectionParameters, PlainCredentials, BasicProperties
from pika.exceptions import AMQPConnectionError


class MPCQueueProvisioner:
    def __init__(self, broker_url: str, session_id: str, hmac_key: bytes) -> None:
        self.session_id: str = session_id
        self.exchange: str = f"mpc.{session_id}.exchange"
        self.queue: str = f"mpc.{session_id}.shares"
        self.dlq: str = f"mpc.{session_id}.dlq"
        # Shared HMAC key so producers and consumers can verify integrity.
        # Distribute out-of-band over the same secure channel that publishes
        # session_params — never over the routing queue.
        self.hmac_key: bytes = hmac_key
        self.conn_params: ConnectionParameters = ConnectionParameters(
            host=broker_url,
            credentials=PlainCredentials(os.getenv("MQ_USER"), os.getenv("MQ_PASS")),
            ssl_options=None,  # TLS 1.3 enforced at the broker level
            heartbeat=600,
        )

    def initialize_session(self) -> dict:
        """Generate session crypto params and provision routing topology."""
        private_key = ec.generate_private_key(ec.SECP256R1())
        _public_key = private_key.public_key()

        session_params: dict = {
            "session_id": self.session_id,
            "curve": "secp256r1",
            "threshold": 2,        # t — shares required to reconstruct
            "total_shares": 3,     # n — total custodians
            "nonce_range": (0, 2 ** 128),
        }

        try:
            conn = BlockingConnection(self.conn_params)
            channel = conn.channel()
            channel.exchange_declare(exchange=self.exchange, exchange_type="direct", durable=True)
            channel.queue_declare(
                queue=self.queue, durable=True,
                arguments={"x-dead-letter-exchange": self.dlq, "x-message-ttl": 30_000},
            )
            channel.queue_declare(queue=self.dlq, durable=True)
            channel.queue_bind(queue=self.queue, exchange=self.exchange, routing_key="shares")
            conn.close()
        except AMQPConnectionError as exc:
            raise RuntimeError(f"Broker provisioning failed: {exc}") from exc

        return session_params


def _validate_session_params() -> None:
    """Runnable harness: verify threshold invariants without a live broker."""
    p = MPCQueueProvisioner.__new__(MPCQueueProvisioner)
    p.session_id = "sess-001"
    params = {"threshold": 2, "total_shares": 3, "nonce_range": (0, 2 ** 128)}
    assert 1 <= params["threshold"] <= params["total_shares"], "t must be in 1..n"
    assert params["nonce_range"][1] > params["nonce_range"][0], "nonce range non-empty"
    print("[Step 1] session parameter invariants hold")


if __name__ == "__main__":
    _validate_session_params()

The TTL of 30_000 ms encodes the round-timeout budget directly: any share older than the round it belongs to is dead-lettered rather than reconstructed late.

Step 2: Asynchronous dispatch and coordinate share publication

Serialize each share into a fixed-layout binary frame, attach an HMAC over the payload for integrity, and publish to the session exchange. A stable wire frame — [session_id(16)][share_index(4)][hmac(32)][payload] — keeps the 52-byte header portable across architectures and lets the consumer reject malformed input before any cryptographic work.

import hmac
import hashlib
import struct
import uuid
from typing import List


class MPCShareProducer:
    HEADER_LEN: int = 52  # 16 (sid) + 4 (idx) + 32 (hmac)

    def __init__(self, session_params: dict, provisioner: "MPCQueueProvisioner") -> None:
        self.session_id: str = session_params["session_id"]
        self.exchange: str = f"mpc.{self.session_id}.exchange"
        # Reuse the provisioner's shared HMAC key so consumers can verify.
        self.provisioner = provisioner
        self.hmac_key: bytes = provisioner.hmac_key

    def _compute_hmac(self, payload: bytes) -> bytes:
        return hmac.new(self.hmac_key, payload, hashlib.sha256).digest()

    def _pack_session_id(self) -> bytes:
        # Pad/truncate to a fixed 16-byte field so the wire frame is stable.
        return self.session_id.encode("utf-8")[:16].ljust(16, b"\x00")

    def build_frame(self, idx: int, share: bytes) -> bytes:
        # Little-endian, no alignment padding ("<") so the header is portable.
        return struct.pack(
            "<16sI32s", self._pack_session_id(), idx, self._compute_hmac(share)
        ) + share

    def publish_coordinate_shares(self, shares: List[bytes]) -> None:
        """Frame, authenticate, and dispatch every share to the async queue."""
        from pika import BlockingConnection, BasicProperties
        conn = BlockingConnection(self.provisioner.conn_params)
        channel = conn.channel()
        for idx, share in enumerate(shares):
            channel.basic_publish(
                exchange=self.exchange,
                routing_key="shares",
                body=self.build_frame(idx, share),
                properties=BasicProperties(
                    delivery_mode=2,  # persistent
                    content_type="application/octet-stream",
                    message_id=str(uuid.uuid4()),  # idempotency key
                ),
            )
        conn.close()


def _validate_framing() -> None:
    """Runnable harness: round-trip a frame without touching a broker."""
    prod = MPCShareProducer.__new__(MPCShareProducer)
    prod.session_id, prod.hmac_key = "sess-001", b"k" * 32
    frame = prod.build_frame(2, b"\x10\x20\x30")
    assert len(frame) == MPCShareProducer.HEADER_LEN + 3
    sid, idx, mac = struct.unpack("<16sI32s", frame[:52])
    assert sid.rstrip(b"\x00").decode() == "sess-001" and idx == 2
    assert hmac.compare_digest(mac, hmac.new(b"k" * 32, frame[52:], hashlib.sha256).digest())
    print("[Step 2] frame round-trips and HMAC verifies")


if __name__ == "__main__":
    _validate_framing()

The message_id doubles as an idempotency key: a retried publish carries the same UUID, so a deduplicating consumer counts the share once even if the broker delivers it twice.

Step 3: Consumer-side reconstruction and round management

Compute nodes pull shares asynchronously, validate the session ID and HMAC in constant time, and buffer payloads until the threshold $t$ is reached. Only then is the reconstructed value handed to the MPC kernel. Constant-time comparison matters here: a variable-time check leaks information about which frames are valid and can be turned into a timing oracle over coordinate density.

import hmac
import hashlib
import struct
from typing import Dict


class MPCShareConsumer:
    HEADER_LEN: int = 52

    def __init__(self, provisioner: "MPCQueueProvisioner", expected_shares: int) -> None:
        self.session_id: str = provisioner.session_id
        self.expected: int = expected_shares  # threshold t
        self.buffer: Dict[int, bytes] = {}
        # Reuse the same provisioner so the HMAC key matches the producer's.
        self.provisioner = provisioner

    def ingest(self, body: bytes) -> str:
        """Validate one frame. Returns 'ack', 'nack', or 'complete'."""
        if len(body) < self.HEADER_LEN:
            return "nack"
        session_bytes, idx, received_hmac = struct.unpack("<16sI32s", body[: self.HEADER_LEN])
        if session_bytes.rstrip(b"\x00").decode("utf-8") != self.session_id:
            return "nack"
        payload = body[self.HEADER_LEN:]
        expected_hmac = hmac.new(self.provisioner.hmac_key, payload, hashlib.sha256).digest()
        if not hmac.compare_digest(received_hmac, expected_hmac):  # constant-time
            return "nack"
        self.buffer[idx] = payload
        return "complete" if len(self.buffer) >= self.expected else "ack"

    def on_message(self, ch, method, properties, body: bytes) -> None:
        """AMQP callback: translate ingest() result into ack/nack/dispatch."""
        result = self.ingest(body)
        if result == "nack":
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)  # → DLQ
            return
        ch.basic_ack(delivery_tag=method.delivery_tag)
        if result == "complete":
            self._trigger_round_completion()

    def _trigger_round_completion(self) -> None:
        reconstructed = b"".join(self.buffer[i] for i in sorted(self.buffer))
        # Hand off to the MPC computation engine (MPyC, ABY, or a spatial kernel).
        print(f"[MPC Router] round {self.session_id} complete: {len(reconstructed)} bytes")


def _validate_consumer() -> None:
    """Runnable harness: good frame acks, tampered frame nacks, threshold fires."""
    prov = type("P", (), {"session_id": "sess-001", "hmac_key": b"k" * 32})()
    cons = MPCShareConsumer(prov, expected_shares=2)
    good = lambda i, p: struct.pack(
        "<16sI32s", b"sess-001".ljust(16, b"\x00"), i,
        hmac.new(b"k" * 32, p, hashlib.sha256).digest()) + p
    assert cons.ingest(good(0, b"\x01")) == "ack"
    assert cons.ingest(b"sess-001".ljust(16, b"\x00") + struct.pack("<I", 1) + b"\x00" * 32 + b"x") == "nack"
    assert cons.ingest(good(1, b"\x02")) == "complete"
    print("[Step 3] ack / nack / threshold transitions verified")


if __name__ == "__main__":
    _validate_consumer()

Step 4: Idempotency, audit, and compliance enforcement

Async delivery is at-least-once, so the producer’s message_id idempotency key and a monotonic share index together guarantee deterministic reconstruction under retries. For regulated deployments, log queue ingress/egress timestamps, the session hash, and the message ID — never the coordinate payload. Align the message TTL with the regulatory retention window the data falls under, and tie each control to a concrete parameter rather than a vague compliance note: the compliance framework mapping translates obligations such as GDPR Article 25 data-minimization into the grid-resolution and TTL constraints the router must enforce.

from typing import Dict, Set


class IdempotentRoundLog:
    """Deduplicate at-least-once delivery and emit payload-free audit records."""

    def __init__(self, session_hash: str) -> None:
        self.session_hash: str = session_hash
        self._seen: Set[str] = set()
        self.audit: list[dict] = []

    def record(self, message_id: str, share_index: int) -> bool:
        """Return True if this is the first time we have seen message_id."""
        first_seen = message_id not in self._seen
        self._seen.add(message_id)
        self.audit.append({
            "session": self.session_hash,    # hash only — no coordinates
            "message_id": message_id,
            "share_index": share_index,
            "duplicate": not first_seen,
        })
        return first_seen


def _validate_idempotency() -> None:
    log = IdempotentRoundLog("h7f3…")
    assert log.record("m-1", 0) is True
    assert log.record("m-1", 0) is False         # retry deduplicated
    assert all("lat" not in r and "lon" not in r for r in log.audit)  # no plaintext
    print("[Step 4] retries deduplicated and audit trail is coordinate-free")


if __name__ == "__main__":
    _validate_idempotency()

Threat model considerations

An adversary against the routing layer is assumed to observe queue traffic, inject frames, and replay captured messages, but not to break the symmetric HMAC or the underlying secret-sharing scheme. The capabilities that matter for this topic:

• Frame injection / queue poisoning — crafting malformed or forged shares to desynchronize a round or crash a consumer. Contained by strict binary framing plus constant-time HMAC verification, with rejects dead-lettered rather than requeued.
• Replay — re-publishing a captured valid share to force duplicate computation or skew an aggregate. Contained by per-message idempotency keys, session nonce ranges, and TTL expiry.
• Timing side channels — inferring coordinate density from how long validation or reconstruction takes. Contained by constant-time comparison and jittered consumer polling.
• Metadata correlation — linking session hashes, message volumes, or timing across rounds to re-identify a custodian. Contained by per-session channels and payload-free audit logs; the broader catalogue lives in threat mapping for GIS data.
• Residency violation — shares transiting a broker in another jurisdiction. Contained by region-locked broker deployments and zero-plaintext transit.

Validation and compliance checklist

1. Reconstruction fidelity — additive/Shamir reconstruction returns the original coordinate within floating-point tolerance ($\leq 10^{-9}$ degrees); fail the build otherwise.
2. Integrity rejection rate — 100% of frames with a corrupted HMAC are dead-lettered, with zero consumer crashes across a fuzzed corpus.
3. Idempotency — a share replayed $k$ times contributes exactly once to the reconstructed value; duplicate ratio in the audit log equals $k-1$.
4. Latency budget — 500+ concurrent shares under 300 ms injected jitter complete the round within the configured TTL (30 s) at the 99th percentile.
5. Audit cleanliness — broker and application logs contain only session hashes, message IDs, and routing metadata; a grep for lat/lon/decimal coordinates returns nothing.
6. Residency proof — every session’s broker resolves to an allowed region; cross-region publish attempts are rejected at provisioning.

Failure modes and remediation

• Round never reaches threshold (node dropout). Fewer than $t$ consumers come online before the TTL expires, so shares dead-letter and the round silently stalls. Remediate with a circuit breaker that trips after the TTL, tears the session down cleanly, and either re-provisions with a fresh nonce or falls back to a synchronous round among the live quorum.
• Budget exhaustion downstream. When reconstructed shares feed homomorphic aggregation, queue-induced delay can push ciphertexts past their noise budget and cause decryption failure. Detect by tracking remaining noise budget per session and refuse to enqueue new rounds once the budget margin is exceeded; size the initial budget per the homomorphic encryption guidance referenced above.
• CRS / precision mismatch. A producer that masks to a different grid resolution than its peers yields shares that reconstruct to a geometrically wrong point with a perfectly valid HMAC. Detect with a session-level CRS and grid-resolution assertion in session_params, validated by every consumer before it buffers.
• DLQ backpressure. A misbehaving producer floods the dead-letter queue and exhausts broker memory. Remediate with a bounded DLQ length, automated reconciliation that alerts on DLQ growth, and producer-side rejection of frames that fail local framing before publish.
• Broker partition. A network split orphans shares on the minority side. Use quorum queues so the majority partition retains durability, and have the minority side abort and resubmit once the partition heals.

Frequently asked questions

Does async routing weaken the cryptographic guarantees of the MPC protocol?

No. The router only transports shares that are already masked and secret-split; it never sees plaintext geometry and the HMAC authenticates integrity, not confidentiality. The privacy guarantee rests on the threshold scheme and masking layer, which are unchanged by moving from a synchronous channel to a queue.

How do I pick the message TTL?

Set it to the round-timeout budget — the maximum time you are willing to wait for $t$ of $n$ shares. A share older than its round must not be reconstructed, so the TTL is what converts a late share into a dead-lettered one. Match it to the regulatory retention window when that window is shorter.

What happens if more than total_shares − threshold nodes are offline?

The round cannot complete and shares dead-letter at TTL. This is the intended safe failure: reconstruction below the quorum is cryptographically prevented. Trip the circuit breaker, tear down the session, and re-provision rather than relaxing the threshold.

Can I run this over Kafka instead of RabbitMQ?

Yes. The framing and HMAC logic are broker-agnostic. Map the session exchange to a topic, use the session ID as the partition key to preserve per-session ordering, and replace the DLQ with a dead-letter topic. The idempotency key moves to the Kafka record key.

Related

• Coordinate Masking Protocols — the masking stage that must run before shares hit the router.
• Secret Sharing for Coordinates — how the shares the router transports are created.
• Homomorphic Encryption Basics — sizing noise budgets that survive queue latency.
• Async Execution Patterns — the federated-learning analogue of staleness-tolerant dispatch.
• Threat Mapping for GIS Data — the adversary catalogue behind this threat model.

Up one level: Secure Multi-Party Computation in Spatial Analytics.