How to map 21 CFR Part 11 requirements to MQTT payloads

import hashlib
import hmac
import json
import logging
import os
from datetime import datetime, timezone
from typing import Any, Dict, Optional

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

# Verifier-side convention: schema_hash + signature_manifest are excluded from
# the canonical form used to compute the HMAC, because they are added AFTER
# the payload is built. Verifiers must follow the same convention.
_SIGNATURE_EXCLUDED_KEYS = {"schema_hash", "signature_manifest"}


def _canonicalize(payload: Dict[str, Any], exclude: set[str] = frozenset()) -> bytes:
    """Deterministic JSON serialization for hashing/signing."""
    filtered = {k: v for k, v in payload.items() if k not in exclude}
    return json.dumps(filtered, sort_keys=True, separators=(",", ":")).encode("utf-8")


class Part11PayloadBuilder:
    SCHEMA_VERSION = "2.1.0"

    def __init__(self, device_id: str, signing_secret: bytes):
        if len(signing_secret) < 32:
            raise ValueError("signing_secret must be at least 32 bytes")
        self.device_id = device_id
        self.signing_secret = signing_secret

    @staticmethod
    def _utc_iso() -> str:
        return datetime.now(timezone.utc).isoformat(timespec="seconds")

    def build(
        self,
        sensor_value: float,
        unit: str,
        operator_id: str,
        action: str,
        previous_value: Optional[float] = None,
        reason_for_change: str = "routine_telemetry",
    ) -> Dict[str, Any]:
        ts = self._utc_iso()
        payload: Dict[str, Any] = {
            "payload_version": self.SCHEMA_VERSION,
            "device_id": self.device_id,
            "timestamp_utc": ts,
            "sensor_reading": {"value": sensor_value, "unit": unit},
            "audit_trail": [{
                "action": action,
                "operator_id": operator_id,
                "timestamp_utc": ts,
                "previous_value": previous_value,
                "new_value": sensor_value,
                "reason_for_change": reason_for_change,
            }],
            "access_control": {
                "operator_role": "cold_chain_monitor",
                "device_cert_fingerprint": "sha256:pending_attestation",
                "access_level": "read_write",
            },
            "transmission_params": {
                "broker_endpoint": "mqtt.pharma-edge.internal",
                "tls_version": "1.3",
                "qos_level": 1,
                "retain_flag": False,
            },
        }

        # Both hashes use the SAME canonicalization with the same exclusion
        # set, so a verifier can reconstruct them deterministically.
        canonical = _canonicalize(payload, exclude=_SIGNATURE_EXCLUDED_KEYS)
        payload["schema_hash"] = hashlib.sha256(canonical).hexdigest()
        payload["signature_manifest"] = {
            "signer_name": self.device_id,
            "signer_role": "automated_edge_agent",
            "signature_timestamp": ts,
            "cryptographic_hash": hmac.new(
                self.signing_secret, canonical, hashlib.sha256
            ).hexdigest(),
        }
        return payload

    @staticmethod
    def verify(payload: Dict[str, Any], signing_secret: bytes) -> bool:
        """Recompute the canonical form and HMAC, returning True iff they match."""
        try:
            expected = hmac.new(
                signing_secret,
                _canonicalize(payload, exclude=_SIGNATURE_EXCLUDED_KEYS),
                hashlib.sha256,
            ).hexdigest()
            actual = payload["signature_manifest"]["cryptographic_hash"]
        except (KeyError, TypeError):
            return False
        return hmac.compare_digest(expected, actual)

    @staticmethod
    def validate(payload: Dict[str, Any]) -> bool:
        required_keys = {
            "payload_version", "device_id", "timestamp_utc", "sensor_reading",
            "audit_trail", "access_control", "transmission_params",
            "schema_hash", "signature_manifest",
        }
        if not required_keys.issubset(payload.keys()):
            return False
        if not isinstance(payload["audit_trail"], list) or not payload["audit_trail"]:
            return False
        return True


class CompliantMQTTPublisher:
    def __init__(self, broker: str, port: int, client_id: str, ca_cert: Optional[str] = None):
        self.client = mqtt.Client(
            CallbackAPIVersion.VERSION2,
            client_id=client_id,
            protocol=mqtt.MQTTv5,
        )
        if ca_cert is not None:
            self.client.tls_set(ca_certs=ca_cert)
        else:
            # Use the system CA bundle and require TLS even without a custom CA.
            self.client.tls_set()
        self.client.connect(broker, port, keepalive=60)

    def publish(self, topic: str, payload: Dict[str, Any], qos: int = 1) -> None:
        if not Part11PayloadBuilder.validate(payload):
            raise ValueError("Payload failed Part 11 schema validation. Transmission aborted.")
        retain = bool(payload.get("transmission_params", {}).get("retain_flag", False))
        info = self.client.publish(
            topic,
            json.dumps(payload, separators=(",", ":")).encode("utf-8"),
            qos=qos,
            retain=retain,
        )
        info.wait_for_publish(timeout=5.0)
        logging.info("Compliant payload published to %s (MID: %s)", topic, info.mid)


# Usage Example. The signing secret MUST come from a KMS/HSM or secrets manager
# — never inline. The env-var fallback below is for local development only.
if __name__ == "__main__":
    signing_secret = os.environb.get(b"PHARMA_HMAC_SECRET")
    if not signing_secret:
        raise SystemExit("PHARMA_HMAC_SECRET environment variable is required")
    builder = Part11PayloadBuilder("EDGE-TH-0042", signing_secret)
    compliant_payload = builder.build(
        sensor_value=2.4, unit="°C", operator_id="SYS_AUTO",
        action="temperature_read", previous_value=2.3,
    )
    publisher = CompliantMQTTPublisher("mqtt.pharma-edge.internal", 8883, "EDGE-TH-0042")
    publisher.publish("pharma/coldchain/warehouse-a/zone-3/telemetry", compliant_payload)