mqtt-broker-stack/backend/e2e_mtls.py

"""E2E: client-certificate mTLS publish chain (spec sections 20, 24).

Phase deploy: CA -> MQTT client -> role with publish ACL -> client cert
(CN=username) -> server cert -> mTLS listener -> deploy. State is saved
to /tmp/e2e_state.json; the caller then RESTARTS mosquitto (a new listener
needs a restart, not a reload — spec section 28) and runs phase verify.

Phase verify: connects to mosquitto:8884 with the client certificate,
publishes to an allowed topic (expect OK) and a forbidden topic (expect
deny), then cleans everything up.

Usage: python /tmp/e2e.py [--phase deploy|verify|all]
"""
import http.cookiejar
import json
import os
import ssl
import sys
import tempfile
import time
import urllib.request

BASE = "http://127.0.0.1:8000/api/v1"
STATE = "/tmp/e2e_state.json"

phase = "all"
if "--phase" in sys.argv:
    phase = sys.argv[sys.argv.index("--phase") + 1]

cj = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))


def call(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(
        f"{BASE}{path}",
        data=data,
        headers={"Content-Type": "application/json"},
        method=method,
    )
    resp = opener.open(req)
    return json.load(resp) if resp.status != 204 else None


call("POST", "/auth/login", {
    "username": os.environ["ADMIN_INITIAL_USERNAME"],
    "password": os.environ["ADMIN_INITIAL_PASSWORD"],
})

state = {}
if phase == "verify":
    with open(STATE) as f:
        state = json.load(f)

# 1. CA
if phase in ("deploy", "all"):
    ca = call("POST", "/pki/ca", {
        "common_name": "E2E mTLS CA", "validity_days": 3650,
        "key_algorithm": "ecdsa-p256",
    })
    print("CA:", ca["id"][:8])

    # 2. MQTT client + role with a publish ACL (idempotent: clear any
    # residue from a previous aborted run first).
    for path in (
        "/mqtt/clients/e2e-device",
        "/mqtt/roles/e2e-publisher",
    ):
        try:
            call("DELETE", path)
        except urllib.error.HTTPError:
            pass
    call("POST", "/mqtt/clients", {"username": "e2e-device"})
    call("POST", "/mqtt/roles", {
        "rolename": "e2e-publisher", "acls": [{
            "acltype": "publishClientSend", "topic": "e2e/%u/data/#",
            "allow": True, "priority": 1,
        }],
    })
    call("POST", "/mqtt/clients/e2e-device/roles",
         {"rolename": "e2e-publisher", "priority": 1})
    print("client + role + ACL ready")

    # 3. Client certificate (CN = username)
    bundle = call("POST", "/pki/certificates/client", {
        "username": "e2e-device", "ca_id": ca["id"], "validity_days": 365,
    })
    print("client cert issued:", bundle["certificate_id"][:8])

    # 4. Server certificate for the listener
    server_cert = call("POST", "/pki/certificates", {
        "ca_id": ca["id"], "common_name": "mosquitto", "validity_days": 365,
        "dns_sans": ["mosquitto", "localhost"],
    })
    print("server cert issued:", server_cert["id"][:8])

    # 5. mTLS listener on 8884 + deploy
    listener = call("POST", "/broker/listeners", {
        "name": "E2E mTLS", "port": 8884, "protocol": "mqtt",
        "tls_enabled": True, "tls_min_version": "tlsv1.2",
        "require_certificate": True, "use_identity_as_username": True,
        "cert_id": server_cert["id"],
    })
    deploy = call("POST", "/broker/config/deploy", {"comment": "E2E mTLS test"})
    print("deployed revision:", deploy.get("revision"),
          "broker_healthy:", deploy.get("broker_healthy"))

    state = {
        "ca": ca["id"],
        "bundle": bundle,
        "server_cert": server_cert["id"],
        "listener": listener["id"],
    }
    with open(STATE, "w") as f:
        json.dump(state, f)

    if phase == "deploy":
        print("state saved — restart mosquitto, then run --phase verify")
        sys.exit(0)

# A NEW listener cannot be added by config reload (spec section 28) —
# the broker must be restarted for it to start listening. The restart is
# performed externally (docker compose restart mosquitto) between the
# deploy phase and the connect phase; this script is invoked with
# --phase deploy|verify to allow that.
import sys
import time

phase = "all"
if len(sys.argv) > 1 and sys.argv[1] == "--phase":
    phase = sys.argv[2]

# 6. Connect with the client certificate and publish
if phase in ("verify", "all"):
    import paho.mqtt.client as mqtt
    from paho.mqtt.enums import CallbackAPIVersion

    bundle = state["bundle"]
    tmp = tempfile.mkdtemp()
    cert_file = os.path.join(tmp, "dev.crt")
    key_file = os.path.join(tmp, "dev.key")
    ca_file = os.path.join(tmp, "ca.crt")
    open(cert_file, "w").write(bundle["certificate_pem"])
    open(key_file, "w").write(bundle["private_key_pem"])
    open(ca_file, "w").write(bundle["ca_chain_pem"])

    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.load_verify_locations(ca_file)
    ctx.load_cert_chain(cert_file, key_file)
    ctx.check_hostname = False   # CN is "mosquitto"; connect by DNS name
    ctx.verify_mode = ssl.CERT_REQUIRED

    results = {}
    client = mqtt.Client(CallbackAPIVersion.VERSION2, client_id="e2e-test")

    def on_connect(cl, u, f, rc, props):
        results["connack"] = rc

    client.on_connect = on_connect
    client.tls_set_context(ctx)
    # The backend container shares backend-net with mosquitto; Docker
    # DNS resolves the name (127.0.0.1 here is the BACKEND, not broker).
    client.connect("mosquitto", 8884, keepalive=30)
    client.loop_start()
    time.sleep(2)

    if results.get("connack") == 0:
        print("CONNECT: OK — certificate accepted, CN mapped to identity")
        allowed = client.publish("e2e/e2e-device/data/1", b"hello", qos=1)
        allowed.wait_for_publish(timeout=5)
        print("PUBLISH allowed topic rc:", allowed.rc, "(0 = OK)")
        denied = client.publish("forbidden/secret", b"x", qos=1)
        denied.wait_for_publish(timeout=5)
        print("PUBLISH forbidden topic rc:", denied.rc,
              "(non-zero = ACL deny enforced)")
    else:
        print("CONNECT FAILED rc:", results.get("connack"))
    client.loop_stop()

# 7. Cleanup
if phase in ("verify", "all"):
    def try_call(method, path, body=None):
        try:
            call(method, path, body)
        except urllib.error.HTTPError:
            pass

    try_call("DELETE", f"/broker/listeners/{state['listener']}")
    try_call("POST", "/broker/config/deploy", {"comment": "E2E cleanup"})
    try_call("DELETE", f"/pki/certificates/{state['bundle']['certificate_id']}")
    try_call("DELETE", f"/pki/certificates/{state['server_cert']}")
    try_call("DELETE", f"/pki/certificates/{state['ca']}")
    try_call("DELETE", "/mqtt/clients/e2e-device")
    try_call("DELETE", "/mqtt/roles/e2e-publisher")
    print("cleanup done")