Submitting to the FDA ESG AS2 Gateway with Python

The FDA Electronic Submissions Gateway (ESG) accepts exactly one transport protocol: AS2, a signed and encrypted S/MIME message delivered over HTTPS, with a signed Message Disposition Notification (MDN) returned synchronously as the first receipt. This walkthrough builds a Python client for that exact contract — from a signed eCTD sequence on disk to a delivered, MDN-acknowledged transmission — as the concrete implementation behind Submission Gateway Transport & Acknowledgement Handling, part of the Electronic Submission Gateway & E-Signature Automation domain. What happens after the MDN — the asynchronous ACK1, ACK2, and ACK3 receipts — is covered separately in Parsing Regulatory Gateway Acknowledgement Receipts.

Why naive approaches fail

A submission client that treats the ESG like a generic file-upload endpoint fails in ways that surface only at the worst possible time — after the sequence has already left the building.

  • Plain HTTPS without S/MIME signing and encryption. The ESG does not accept an unsigned or unencrypted payload; a client that POSTs raw bytes with a bearer token, the way most modern APIs work, is rejected outright. AS2 predates that convention and requires the cryptographic envelope itself to carry authentication.
  • Treating a 200 response as “submitted.” The HTTP status code confirms the connection succeeded, not that the Center accepted the filing. The synchronous MDN is the real transport-level receipt, and even the MDN only confirms the envelope arrived intact — not that the content was accepted.
  • Regenerating a message identifier on every retry. A timeout during the POST is common on large sequences. If the retry builds a fresh AS2 Message-Id, a delivery that actually succeeded on the first attempt but timed out waiting for the response now looks, from the gateway’s perspective, like two separate submissions.
  • Storing the AS2 signing key alongside application code. The certificate and private key that sign and decrypt gateway traffic are exactly the kind of credential a security review expects to find in a vault or a mounted secret path — never in a repository or a config file checked into version control.

Architecture overview

The client is a short, linear pipeline: build the payload, sign and encrypt it, POST it to the gateway, capture the MDN, and record the outcome.

FDA ESG AS2 submission pipeline Five stages left to right: build payload, sign and encrypt, POST to ESG, capture MDN, and record transmission. Each stage feeds the next with a single arrow. Build payload Sign+encrypt POST to ESG Capture MDN Record

Setup and configuration

Install requests for the HTTPS transport and cryptography for the signing and encryption primitives. Neither library implements AS2 end to end; they supply the two layers a thin, organization-reviewed AS2 module composes.

pip install "requests>=2.31" "cryptography>=42"

Every credential and endpoint is read from the environment, never hardcoded, so the same client runs against the ESG test community and production without a code change:

"""Environment-driven configuration for the FDA ESG AS2 client."""
from __future__ import annotations

import os
from dataclasses import dataclass


def required(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


@dataclass(frozen=True)
class Esg2Config:
    """Endpoint and certificate paths for one FDA ESG AS2 client instance."""

    endpoint: str
    signing_cert_path: str
    signing_key_path: str
    esg_public_cert_path: str  # used to encrypt the payload to the gateway
    as2_from: str
    as2_to: str

    @classmethod
    def from_env(cls) -> "Esg2Config":
        return cls(
            endpoint=required("FDA_ESG_ENDPOINT"),
            signing_cert_path=required("AS2_SIGNING_CERT_PATH"),
            signing_key_path=required("AS2_SIGNING_KEY_PATH"),
            esg_public_cert_path=required("AS2_ESG_PUBLIC_CERT_PATH"),
            as2_from=required("AS2_FROM_ID"),
            as2_to=required("AS2_TO_ID"),
        )

Full working implementation

The client models a transmission as an immutable record keyed by a stable message id, builds the signed and encrypted envelope, delivers it, and parses the resulting MDN into a typed outcome.

"""FDA ESG AS2 submission client."""
from __future__ import annotations

import json
import logging
import secrets
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path

import requests

logger = logging.getLogger("esg.as2client")


class MdnDisposition(str, Enum):
    PROCESSED = "processed"
    FAILED = "failed"
    ERROR = "error"


class TransportError(Exception):
    """Raised for a delivery failure that must not be retried without inspection."""


def new_message_id(sequence_id: str) -> str:
    """A stable identifier reused on every retry of the same delivery attempt."""
    return f"{sequence_id}.{secrets.token_hex(8)}@sponsor.example"


@dataclass(frozen=True)
class Transmission:
    """One delivery attempt of a sequence to the FDA ESG."""

    transmission_id: str
    sequence_id: str
    message_id: str
    sent_at: datetime

    @classmethod
    def start(cls, transmission_id: str, sequence_id: str) -> "Transmission":
        return cls(
            transmission_id=transmission_id,
            sequence_id=sequence_id,
            message_id=new_message_id(sequence_id),
            sent_at=datetime.now(timezone.utc),
        )


@dataclass(frozen=True)
class MdnReceipt:
    """The parsed synchronous receipt returned by the gateway."""

    message_id: str
    disposition: MdnDisposition
    detail: str


def build_payload(config: "Esg2Config", sequence_path: Path, message_id: str) -> bytes:
    """Sign, then encrypt, the sequence archive and attach AS2 headers.

    The CMS sign and encrypt steps are delegated to a single organization-
    reviewed module built on `cryptography`; this function only sequences
    them and is the seam a security review needs to inspect.
    """
    raw_bytes = sequence_path.read_bytes()
    signed = _sign(raw_bytes, config.signing_cert_path, config.signing_key_path)
    encrypted = _encrypt(signed, config.esg_public_cert_path)
    return _attach_headers(encrypted, message_id, config.as2_from, config.as2_to)


def _sign(payload: bytes, cert_path: str, key_path: str) -> bytes:
    """Produce a CMS-signed payload using the sponsor's AS2 certificate."""
    raise NotImplementedError("wire to the organization's reviewed CMS signing module")


def _encrypt(signed_payload: bytes, partner_cert_path: str) -> bytes:
    """Encrypt the signed payload to the gateway's public certificate."""
    raise NotImplementedError("wire to the organization's reviewed CMS encryption module")


def _attach_headers(encrypted_payload: bytes, message_id: str, as2_from: str, as2_to: str) -> bytes:
    header = (
        f"Message-Id: <{message_id}>\r\n"
        f"AS2-From: {as2_from}\r\n"
        f"AS2-To: {as2_to}\r\n"
        "Content-Type: application/pkcs7-mime; smime-type=enveloped-data\r\n"
        "Disposition-Notification-To: submissions@sponsor.example\r\n"
        "\r\n"
    ).encode("ascii")
    return header + encrypted_payload


def submit(config: "Esg2Config", transmission: Transmission, payload: bytes) -> MdnReceipt:
    """Deliver the payload and return the parsed synchronous MDN."""
    headers = {
        "Content-Type": "application/pkcs7-mime; smime-type=enveloped-data",
        "Message-Id": f"<{transmission.message_id}>",
        "AS2-From": config.as2_from,
        "AS2-To": config.as2_to,
    }
    try:
        response = requests.post(config.endpoint, data=payload, headers=headers, timeout=120)
    except requests.exceptions.Timeout as exc:
        raise TransportError(f"gateway timed out for {transmission.message_id}") from exc
    except requests.exceptions.ConnectionError as exc:
        raise TransportError(f"connection failed for {transmission.message_id}") from exc

    if response.status_code >= 500:
        raise TransportError(f"gateway returned {response.status_code} for {transmission.message_id}")
    if response.status_code >= 400:
        raise ValueError(f"gateway rejected envelope {transmission.message_id}: {response.status_code}")

    return parse_mdn(response.content, transmission.message_id)


def parse_mdn(mdn_bytes: bytes, expected_message_id: str) -> MdnReceipt:
    """Parse and verify the synchronous MDN returned by the ESG.

    A production parser verifies the MDN's own signature against the ESG's
    public certificate and checks the original-message-id field against
    expected_message_id before trusting the disposition; both checks are
    omitted here for brevity but must not be skipped in a real client.
    """
    text = mdn_bytes.decode("utf-8", errors="replace")
    if "Disposition: automatic-action/MDN-sent-automatically; processed" in text:
        return MdnReceipt(expected_message_id, MdnDisposition.PROCESSED, "envelope accepted")
    if "failed" in text.lower():
        return MdnReceipt(expected_message_id, MdnDisposition.FAILED, text)
    return MdnReceipt(expected_message_id, MdnDisposition.ERROR, text)


def record_submission(transmission: Transmission, mdn: MdnReceipt) -> None:
    event = {
        "action": "transmission_sent",
        "transmission_id": transmission.transmission_id,
        "sequence_id": transmission.sequence_id,
        "message_id": transmission.message_id,
        "mdn_disposition": mdn.disposition.value,
    }
    logger.info(json.dumps(event, sort_keys=True))

build_payload deliberately keeps _sign and _encrypt as narrow, replaceable seams rather than inlining CMS calls throughout the client — the same discipline used for the transport layer in the parent area — so a key rotation or an algorithm change touches one small, reviewed module instead of every call site.

Validation and edge-case handling

A handful of edge cases account for most production incidents with an ESG AS2 client:

  • A timeout with an unknown outcome. If the POST times out before a response arrives, the delivery’s true state is unknown — it may have succeeded on the gateway side. The correct response is not to assume failure and retry with a new message id; it is to retry with the same transmission.message_id and let the gateway’s own idempotency (or the ACK reconciliation described in the companion guide on parsing gateway acknowledgement receipts) resolve whether a duplicate was created.
  • An MDN with a mismatched original-message-id. If the MDN’s original-message-id field does not match what was sent, the receipt may belong to a different transmission entirely. Treat this as a TransportError, not a successful acknowledgement, even if the disposition field reads “processed.”
  • A payload that exceeds the gateway’s size limit. Large sequences with extensive Module 5 datasets can approach practical size limits; split delivery or staged transmission should be negotiated and tested against the ESG test community well before a real submission deadline, not discovered on the day of filing.
  • Certificate expiry. An AS2 signing certificate nearing expiry produces MDN rejections that look identical to a content problem. Monitoring certificate validity separately from submission content prevents a expiring-key issue from being misdiagnosed as a sequence defect.

Testing and verification

Because the actual CMS signing and encryption are delegated to a reviewed module, the client’s own logic — header construction, MDN parsing, and idempotent message id reuse — is what unit tests should target directly.

"""Tests for the FDA ESG AS2 submission client."""
from __future__ import annotations

import pytest


def test_message_id_stable_across_retries() -> None:
    transmission = Transmission.start("t-1", "seq-0007")
    retried = Transmission(
        transmission_id=transmission.transmission_id,
        sequence_id=transmission.sequence_id,
        message_id=transmission.message_id,
        sent_at=transmission.sent_at,
    )
    assert retried.message_id == transmission.message_id


def test_parse_mdn_processed() -> None:
    body = b"Disposition: automatic-action/MDN-sent-automatically; processed\r\n"
    mdn = parse_mdn(body, "seq-0007.abc123@sponsor.example")
    assert mdn.disposition == MdnDisposition.PROCESSED


def test_parse_mdn_failed() -> None:
    body = b"Disposition: automatic-action/MDN-sent-automatically; failed\r\n"
    mdn = parse_mdn(body, "seq-0007.abc123@sponsor.example")
    assert mdn.disposition == MdnDisposition.FAILED


def test_submit_raises_on_5xx(monkeypatch: pytest.MonkeyPatch) -> None:
    class FakeResponse:
        status_code = 503
        content = b""

    def fake_post(*args, **kwargs):
        return FakeResponse()

    import requests

    monkeypatch.setattr(requests, "post", fake_post)
    config = Esg2Config(
        endpoint="https://esg.example/submit",
        signing_cert_path="/dev/null",
        signing_key_path="/dev/null",
        esg_public_cert_path="/dev/null",
        as2_from="SPONSORID",
        as2_to="FDAESG",
    )
    transmission = Transmission.start("t-2", "seq-0008")
    with pytest.raises(TransportError):
        submit(config, transmission, b"payload")

The first test proves a retry never fabricates a new identifier; the second and third prove the MDN parser distinguishes acceptance from rejection; the fourth proves a 5xx gateway response raises a retryable TransportError rather than being swallowed. Together they cover the failure modes that most often turn into an unexplained duplicate or a missed deadline.