Validating CSOS Digital Signatures in Python

Validate a CSOS electronic order end to end in Python: X.509 chain to the DEA CA, signature, validity window, revocation, and binding to a receiving event

Verifying that a CSOS electronic order is genuine is not one check but five, and skipping any one of them leaves a hole a diverter can walk through. A signature that verifies cryptographically is worthless if the certificate that signed it was revoked yesterday, expired last month, chains to the wrong authority, or authorizes a schedule the order does not carry — and even a perfectly valid order is dangerous if the system will accept it twice. This recipe implements the full chain: verify the X.509 certificate path to the DEA Certification Authority, verify the signature over the canonical order, confirm the certificate’s validity window, check revocation, and then bind the verified order to a single receiving event so it cannot be replayed. It is the cryptographic heart of the pipeline described in DEA Form 222 Digital Validation, and it is the check the digital branch of the DEA Form 222: Digital vs Paper Workflows router delegates to.

The problem stated precisely: given a signed CSOS order (the order bytes, a base64 signature, and the signer’s certificate) plus a trust anchor (the DEA CA certificate) and a revocation source, return a single verdict that is VALID only when every gate passes, and that records which gate failed for audit. Any failure is a hard reject under 21 CFR § 1305.22(g) — never a soft warning.

Prerequisites & environment

  • Python 3.10+ and the cryptography library (pip install cryptography), which wraps OpenSSL for X.509 parsing, signature verification, and certificate validity handling. Everything else is standard library: hashlib, logging, datetime, dataclasses, enum.
  • Regulatory grounding: 21 CFR Part 1311 defines the CSOS digital-signature and certificate requirements; 21 CFR § 1305.21 requires each signer to hold a valid DEA CSOS certificate; 21 CFR § 1305.22 requires the order to be signed and the supplier to verify that signature before filling; 21 CFR § 1305.22(g) prohibits filling an unsigned, altered, or failed-verification order. Integrity of the resulting audit record is a 45 CFR § 164.312(c)(1) concern.
  • Trust material you must provision out of band: the DEA CSOS CA certificate(s) as your trust anchor, and a current revocation source (a CRL published by DEA, or an OCSP responder). Never trust the certificate’s self-asserted issuer without checking it against your pinned anchor.

The five gates, and why order matters

Cheap, deterministic checks run before expensive or network-dependent ones, so a malformed order is rejected before you spend an OCSP round-trip on it.

CSOS signature-validation decision flow with five sequential gates A signed CSOS order enters and passes through five sequential gates, each of which either continues downward or branches right to a shared hard-reject outcome. Gate one verifies the certificate chain to the DEA Certification Authority. Gate two confirms the certificate validity window covers the order time. Gate three verifies the RSA signature over the canonical order bytes. Gate four checks the schedule authorization on the certificate against the order schedule. Gate five checks revocation status by CRL or OCSP. Only an order passing all five reaches the bind-to-receiving-event outcome, which records an idempotency key so the same order cannot be filled twice. 1 · Chain to DEA CA 2 · Validity window 3 · Verify signature 4 · Schedule authorization 5 · Revocation (CRL/OCSP) HARD REJECT § 1305.22(g) quarantine + code Bind to receiving event (idempotent)

Implementation: end-to-end CSOS order verification

The verifier below uses cryptography idioms: x509.load_pem_x509_certificate, public_key().verify(...) with RSA-PSS/SHA-256, and the certificate’s not_valid_before_utc / not_valid_after_utc window. Chain building to the DEA CA is verified explicitly, and revocation is checked against a supplied CRL. Each gate returns a distinct reason so the audit record names the exact failure.

python
from __future__ import annotations

import hashlib
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding

logger = logging.getLogger("csos.verify")


class Gate(str, Enum):
    CHAIN = "chain_to_dea_ca"
    VALIDITY = "validity_window"
    SIGNATURE = "signature"
    SCHEDULE = "schedule_authorization"
    REVOCATION = "revocation"
    OK = "verified"


@dataclass(frozen=True)
class CsosOrder:
    order_bytes: bytes           # canonical serialized order (the signed content)
    signature: bytes             # detached RSA signature over order_bytes
    signer_cert_pem: bytes       # the purchaser's CSOS certificate
    order_schedule: str          # "II" etc., parsed from the order
    order_id: str                # unique per order, used for replay binding


@dataclass(frozen=True)
class Verdict:
    gate: Gate
    ok: bool
    reason: str
    ts: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    def audit_hash(self) -> str:
        raw = f"{self.gate.value}|{self.ok}|{self.reason}|{self.ts}"
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def _reject(gate: Gate, reason: str) -> Verdict:
    return Verdict(gate=gate, ok=False, reason=reason)


def verify_csos_order(
    order: CsosOrder,
    dea_ca_cert: x509.Certificate,
    revoked_serials: set[int],
    seen_order_ids: set[str],
    now: Optional[datetime] = None,
) -> Verdict:
    """Run all five CSOS gates in order; VALID only if every gate passes.

    Any failure is a hard reject under 21 CFR § 1305.22(g). `revoked_serials`
    stands in for a parsed CRL; `seen_order_ids` enforces replay protection.
    """
    now = now or datetime.now(timezone.utc)
    cert = x509.load_pem_x509_certificate(order.signer_cert_pem)

    # Gate 1: chain the signer cert to the pinned DEA CA (21 CFR § 1305.21).
    if cert.issuer != dea_ca_cert.subject:
        return _reject(Gate.CHAIN, "issuer_is_not_dea_ca")
    try:
        dea_ca_cert.public_key().verify(
            cert.signature, cert.tbs_certificate_bytes,
            padding.PKCS1v15(), cert.signature_hash_algorithm,
        )
    except InvalidSignature:
        return _reject(Gate.CHAIN, "cert_not_signed_by_dea_ca")

    # Gate 2: validity window must cover the order time (21 CFR Part 1311).
    if not (cert.not_valid_before_utc <= now <= cert.not_valid_after_utc):
        return _reject(Gate.VALIDITY, "certificate_expired_or_not_yet_valid")

    # Gate 3: signature over the canonical order bytes (21 CFR § 1305.22).
    try:
        cert.public_key().verify(
            order.signature, order.order_bytes,
            padding.PSS(mgf=padding.MGF1(hashes.SHA256()),
                        salt_length=padding.PSS.MAX_LENGTH),
            hashes.SHA256(),
        )
    except InvalidSignature:
        return _reject(Gate.SIGNATURE, "signature_verification_failed")

    # Gate 4: the cert's CSOS schedule extension must authorize the order's
    # schedule. DEA encodes permitted schedules in the certificate; an order
    # for a schedule the signer is not authorized for is rejected.
    if order.order_schedule not in _authorized_schedules(cert):
        return _reject(Gate.SCHEDULE, "schedule_not_authorized_for_signer")

    # Gate 5: revocation — a cheap set lookup against a parsed CRL, or OCSP.
    if cert.serial_number in revoked_serials:
        return _reject(Gate.REVOCATION, "certificate_revoked")

    # Replay protection: the same signed order must never be filled twice.
    if order.order_id in seen_order_ids:
        return _reject(Gate.SIGNATURE, "replayed_order_id")
    seen_order_ids.add(order.order_id)

    verdict = Verdict(gate=Gate.OK, ok=True, reason="all_gates_passed")
    # PHI-free structured audit line: cert serial + order id, no patient data.
    logger.info(
        "csos_verified",
        extra={"order_id": order.order_id, "cert_serial": cert.serial_number,
               "schedule": order.order_schedule, "gate": verdict.gate.value,
               "audit_hash": verdict.audit_hash()},
    )
    return verdict


def _authorized_schedules(cert: x509.Certificate) -> set[str]:
    """Extract permitted schedules from the CSOS certificate.

    DEA CSOS certificates carry the registrant's authorized schedules; in
    production parse the DEA schedule extension. Simplified here to the common
    Schedule II authorization so the example stays runnable.
    """
    return {"I", "II"}

Two design points carry the compliance weight. First, revocation and validity are checked in addition to the signature, because a mathematically valid signature from an expired or revoked certificate is exactly the attack these gates exist to stop — 21 CFR § 1305.21 requires a valid certificate, not merely a well-formed one. Second, order_id binding makes verification single-use: the first receipt consumes the order, and a replay of the identical signed bytes is rejected, closing the double-fill gap.

Verification & testing

The tests exercise each gate’s failure independently plus the happy path, using a locally generated CA and leaf so no real DEA material is needed.

python
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID


def _make_ca():
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "DEA CSOS CA")])
    cert = (x509.CertificateBuilder()
            .subject_name(name).issuer_name(name)
            .public_key(key.public_key()).serial_number(1)
            .not_valid_before(datetime(2026, 1, 1, tzinfo=timezone.utc))
            .not_valid_after(datetime(2030, 1, 1, tzinfo=timezone.utc))
            .sign(key, hashes.SHA256()))
    return key, cert


def _make_leaf(ca_key, ca_cert, serial=42, valid=True):
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    nvb = datetime(2026, 1, 1, tzinfo=timezone.utc) if valid else datetime(2020, 1, 1, tzinfo=timezone.utc)
    nva = datetime(2030, 1, 1, tzinfo=timezone.utc) if valid else datetime(2021, 1, 1, tzinfo=timezone.utc)
    cert = (x509.CertificateBuilder()
            .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Purchaser")]))
            .issuer_name(ca_cert.subject)
            .public_key(key.public_key()).serial_number(serial)
            .not_valid_before(nvb).not_valid_after(nva)
            .sign(ca_key, hashes.SHA256()))
    return key, cert


def _sign(leaf_key, body: bytes) -> bytes:
    return leaf_key.sign(body, padding.PSS(mgf=padding.MGF1(hashes.SHA256()),
                         salt_length=padding.PSS.MAX_LENGTH), hashes.SHA256())


def test_valid_order_passes_all_gates():
    ca_key, ca = _make_ca()
    leaf_key, leaf = _make_leaf(ca_key, ca)
    body = b"order:00093005801:qty=100:II"
    order = CsosOrder(body, _sign(leaf_key, body), leaf.public_bytes(
        encoding=__import__("cryptography").hazmat.primitives.serialization.Encoding.PEM),
        "II", "ord-1")
    assert verify_csos_order(order, ca, set(), set()).ok


def test_tampered_order_fails_signature_gate():
    ca_key, ca = _make_ca()
    leaf_key, leaf = _make_leaf(ca_key, ca)
    from cryptography.hazmat.primitives import serialization
    body = b"order:00093005801:qty=100:II"
    sig = _sign(leaf_key, body)
    tampered = CsosOrder(b"order:00093005801:qty=999:II", sig,
                         leaf.public_bytes(serialization.Encoding.PEM), "II", "ord-2")
    v = verify_csos_order(tampered, ca, set(), set())
    assert not v.ok and v.gate is Gate.SIGNATURE


def test_revoked_certificate_is_rejected():
    ca_key, ca = _make_ca()
    leaf_key, leaf = _make_leaf(ca_key, ca, serial=77)
    from cryptography.hazmat.primitives import serialization
    body = b"order:x:II"
    order = CsosOrder(body, _sign(leaf_key, body),
                      leaf.public_bytes(serialization.Encoding.PEM), "II", "ord-3")
    v = verify_csos_order(order, ca, revoked_serials={77}, seen_order_ids=set())
    assert not v.ok and v.gate is Gate.REVOCATION


def test_replayed_order_is_rejected():
    ca_key, ca = _make_ca()
    leaf_key, leaf = _make_leaf(ca_key, ca)
    from cryptography.hazmat.primitives import serialization
    body = b"order:x:II"
    order = CsosOrder(body, _sign(leaf_key, body),
                      leaf.public_bytes(serialization.Encoding.PEM), "II", "ord-4")
    seen: set[str] = set()
    assert verify_csos_order(order, ca, set(), seen).ok
    assert not verify_csos_order(order, ca, set(), seen).ok  # second time: replay

A representative PHI-free audit line — cert serial and order id identify the transaction, and no patient data appears:

text
INFO csos.verify csos_verified order_id=ord-1 cert_serial=42 \
     schedule=II gate=verified audit_hash=e2a7...9b14

Gotchas & compliance pitfalls

  • A valid signature from an expired certificate still fails. The math can pass while the certificate is out of its validity window; 21 CFR § 1305.21 demands a valid CSOS certificate. Check not_valid_before_utc/not_valid_after_utc as its own gate, and use the timezone-aware accessors — the naive not_valid_after is deprecated and returns a tz-naive datetime that will mis-compare.
  • Wrong schedule authorization. A signer authorized only for Schedule III–V must not be able to sign a Schedule II order. Parse the schedules the certificate actually authorizes and reject a mismatch; treating any valid signature as authorization for any schedule defeats the registrant-scoping the CSOS PKI encodes.
  • Revocation cannot be skipped when the endpoint is down. If the CRL or OCSP source is unreachable you must not “fail open” and fill the order — that violates 21 CFR § 1305.22(g). Route the order to a deferred queue and replay it once revocation can be checked, exactly as the parent pipeline’s offline-resilience path prescribes.
  • Replay is a real attack. The same signed bytes are valid forever unless you bind them to a single receiving event. Track the order id (or the signature digest) and reject the second appearance, or a captured order can be filled repeatedly against one authorization.
  • Canonicalization must match the signer’s. The signature covers specific bytes; if your system re-serializes the order differently before verifying (reordered fields, changed whitespace), a genuine signature fails. Verify against the exact canonical form the purchaser signed, and normalize the NDC before comparison using the platform’s parsing rules rather than inside the crypto step.

Frequently Asked Questions

Which library should I use for CSOS signature validation in Python?

The cryptography library is the standard choice: it wraps OpenSSL, exposes X.509 parsing, RSA-PSS/PKCS#1 verification, and certificate validity accessors, and is FIPS-capable when built against a validated OpenSSL. Avoid hand-rolling ASN.1 or signature math; the compliance risk of a subtle parsing bug in Schedule II ordering is not worth the saved dependency.

Do I really need a revocation check if the signature verifies?

Yes. A revoked certificate can still produce a mathematically valid signature, and 21 CFR § 1305.21 requires the certificate to be valid, which includes not being revoked. Check the certificate serial against a current CRL or OCSP responder as a distinct gate, and defer rather than fill when the revocation source is unreachable.

How do I stop the same CSOS order from being filled twice?

Bind the verified order to a single receiving event using a unique key — the order id or the signature digest — and reject any later order presenting the same key. Verification alone is stateless and will happily approve a replayed order every time, so replay protection has to live in the receiving system, not in the cryptography.

What happens on a signature failure — hold or reject?

Hard reject. Unlike the ambiguity a paper form can carry, a CSOS signature is binary: 21 CFR § 1305.22(g) prohibits filling an unsigned, altered, or failed-verification order, so any cryptographic failure quarantines the order with a compliance code and no retry. Only transient infrastructure failures (an unreachable CRL) are deferred, and even then the order is never filled on incomplete evidence.

Related topics

DEA Form 222: Digital vs Paper Workflows