switch to emqx

nix/modules/default.nix

@@ -4,6 +4,6 @@
   gebos-supabase      = import ./gebos-supabase.nix;
   gebos-caddy         = import ./gebos-caddy.nix;
   gebos-postgres      = import ./gebos-postgres.nix;
-  gebos-hivemq        = import ./gebos-hivemq.nix;
+  gebos-emqx          = import ./gebos-emqx.nix;
   gebos-frontend      = import ./gebos-frontend.nix;
 }

nix/modules/gebos-caddy.nix

@@ -3,24 +3,8 @@
 let
   cfg = config.services.gebos.caddy;
 
-  tcpProxySites = lib.filterAttrs (_: s: s ? tcpProxy) cfg.sites;
-  httpSites = lib.filterAttrs (_: s: !(s ? tcpProxy)) cfg.sites;
-  needsLayer4 = tcpProxySites != { };
-
-  # External MQTTS port the senders connect to (mqtt_port = 8883, SSL).
-  tlsListenPort = 8883;
-
-  # Caddy built with the layer4 module (caddy-l4) — needed to TLS-terminate raw
-  # MQTT and proxy it to HiveMQ, which Caddy's HTTP-only core can't do. Only
-  # built when a host actually has a tcpProxy site; app-host's static/upstream
-  # sites run on stock Caddy so they don't pay for the custom build.
-  caddyWithL4 = pkgs.caddy.withPlugins {
-    plugins = [ "github.com/mholt/caddy-l4@v0.1.1" ];
-    hash = "sha256-O6GuC2q1mA/Fa0utb2Yg7ZE73iq13oVYhJI1IVyOvog=";
-  };
-
   # HTTP-style sites: static file server or reverse proxy.
-  httpSiteBlock = host: site:
+  siteBlock = host: site:
     if site ? staticRoot then ''
       ${host} {
         root * ${site.staticRoot}
@@ -35,40 +19,7 @@ let
         encode zstd gzip
       }
     ''
-    else throw "site `${host}` needs one of: staticRoot, upstream, tcpProxy";
-
-  # A tcpProxy site needs TWO Caddyfile pieces:
-  #   1. a layer4 listener (in the global options block) that terminates TLS on
-  #      :8883 and proxies cleartext to the upstream (HiveMQ on 127.0.0.1:1883);
-  #   2. a normal HTTPS site block for the host, purely so Caddy's automatic
-  #      HTTPS OBTAINS the ACME cert that the layer4 `tls` handler then serves by
-  #      SNI — caddy-l4's `tls` handler only terminates, it never provisions
-  #      certs. This is why ports 80/443 must stay open for the ACME challenge.
-  layer4Listener = _host: site: ''
-		:${toString tlsListenPort} {
-			route {
-				tls
-				proxy ${site.tcpProxy}
-			}
-		}
-'';
-
-  certSiteBlock = host: _site: ''
-    ${host} {
-    	respond "gebos MQTT ingest — connect MQTTS on :${toString tlsListenPort}" 200
-    }
-  '';
-
-  # caddy-l4 listeners, folded into one `layer4` directive that lives INSIDE the
-  # Caddyfile global options block. We must NOT emit our own `{ }` block here:
-  # the NixOS caddy module already generates the single global block (for
-  # `email`), and a second keyless block is invalid ("server block without any
-  # key is global configuration, and if used, it must be first"). So this goes to
-  # services.caddy.globalConfig, which the module places inside that one block.
-  layer4Global = lib.optionalString needsLayer4 ''
-    layer4 {
-    ${lib.concatStrings (lib.mapAttrsToList layer4Listener tcpProxySites)}	}
-  '';
+    else throw "site `${host}` needs one of: staticRoot, upstream";
 in
 {
   options.services.gebos.caddy = {
@@ -82,13 +33,8 @@ in
   config = lib.mkIf cfg.enable {
     services.caddy = {
       enable = true;
-      package = if needsLayer4 then caddyWithL4 else pkgs.caddy;
       email = "ops@gebos.online";  # TODO: confirm ACME contact
-      globalConfig = layer4Global;
-      extraConfig = lib.concatStringsSep "\n" (
-        lib.mapAttrsToList httpSiteBlock httpSites
-        ++ lib.mapAttrsToList certSiteBlock tcpProxySites
-      );
+      extraConfig = lib.concatStringsSep "\n" (lib.mapAttrsToList siteBlock cfg.sites);
     };
   };
 }

nix/modules/gebos-emqx.nix

@@ -0,0 +1,238 @@
+{ config, lib, pkgs, ... }:
+
+let
+  cfg = config.services.gebos.emqx;
+
+  stateDir = "/var/lib/gebos-emqx";
+
+  # UID/GID of the `emqx` user baked into the official image. Host-side files
+  # the container must read or write (certs, bootstrap users, data dir) are
+  # chowned to this numeric id. On these hosts uid 1000 is the `deploy` user,
+  # which already has passwordless sudo, so this grants it nothing new.
+  emqxUid = 1000;
+  emqxGid = 1000;
+
+  # Full emqx.conf, replacing the image default. EMQX reads HOCON natively;
+  # container paths below are bind mounts declared on the oci-container.
+  emqxConf = pkgs.writeText "emqx.conf" ''
+    node {
+      name = "emqx@127.0.0.1"
+      # Erlang distribution cookie. Not treated as a secret: the dist/epmd
+      # ports are never opened in the firewall and the only local users are
+      # root and deploy (wheel).
+      cookie = "gebos-emqx"
+      data_dir = "data"
+    }
+
+    cluster {
+      name = gebos
+      # Single node, and required to be `singleton` for durable sessions on
+      # the open-source builtin_local storage backend.
+      discovery_strategy = singleton
+    }
+
+    # Dashboard on loopback only (the container uses host networking) — reach
+    # it via `ssh -L 18083:127.0.0.1:18083`. First login is admin/public with a
+    # forced password change; the changed password persists in the data dir.
+    dashboard {
+      listeners.http.bind = "127.0.0.1:18083"
+    }
+
+    # Plain MQTT for the co-located ingester only.
+    listeners.tcp.default {
+      bind = "${cfg.bindAddress}:${toString cfg.port}"
+    }
+
+    # MQTTS for the senders. EMQX terminates TLS itself; certs are managed by
+    # security.acme on the host and installed into ${stateDir}/certs (mounted
+    # at /etc/emqx/certs). EMQX re-reads the PEM files from disk periodically
+    # (~120 s), so renewals need no restart — paths stay fixed, contents swap.
+    listeners.ssl.default {
+      bind = "0.0.0.0:${toString cfg.tlsPort}"
+      ssl_options {
+        certfile = "/etc/emqx/certs/fullchain.pem"
+        keyfile = "/etc/emqx/certs/privkey.pem"
+        versions = ["tlsv1.2", "tlsv1.3"]
+      }
+    }
+
+    # The schema defines websocket listeners on 8083/8084 by default — unused.
+    listeners.ws.default.enable = false
+    listeners.wss.default.enable = false
+
+    # Sessions and queued messages survive a broker restart (QoS > 0 with
+    # clean_start = false), mirroring HiveMQ's file persistence mode.
+    durable_sessions.enable = ${lib.boolToString cfg.persistentSessions}
+
+    # Username/password auth against the built-in database, seeded from the
+    # sops-rendered users.csv on first start (bootstrap only inserts users that
+    # do not exist yet — to rotate a password, delete the user in the dashboard
+    # or via `emqx ctl`, then restart to re-import).
+    authentication = [
+      {
+        mechanism = password_based
+        backend = built_in_database
+        user_id_type = username
+        password_hash_algorithm { name = plain, salt_position = disable }
+        bootstrap_file = "/etc/emqx/auth/users.csv"
+        bootstrap_type = plain
+      }
+    ]
+
+    # Topic-level authorization from the static acl.conf; anything not
+    # explicitly allowed is denied.
+    authorization {
+      no_match = deny
+      sources = [
+        {
+          type = file
+          enable = true
+          path = "/etc/emqx/auth/acl.conf"
+        }
+      ]
+    }
+  '';
+
+  # Topic policy (non-secret — usernames and topics only, no credentials).
+  # Devices publish under `acrios/<IMSI>/<metric>` (the sender firmware's
+  # mqtt_topic_base), so that is the namespace the ingester consumes and the
+  # device user is scoped to. `admin` is is_superuser in users.csv and bypasses
+  # this ACL entirely (break-glass).
+  #
+  # Option A (current): one shared device user (`bender`) for the whole fleet.
+  # Tenant isolation is enforced downstream by the ingester's IMSI→tenant
+  # registry lookup, NOT at the broker — any device could publish under any
+  # IMSI. TODO(Option B): per-device users with username == IMSI and an ACL
+  # rule scoped to `acrios/''${username}/#` for real anti-spoofing.
+  aclConf = pkgs.writeText "emqx-acl.conf" ''
+    %% ingester: consume the whole device tree.
+    {allow, {username, "ingester"}, subscribe, ["acrios/#"]}.
+    %% device fleet: publish telemetry + LWT/status, subscribe downlinks.
+    {allow, {username, "bender"}, all, ["acrios/#"]}.
+    {deny, all}.
+  '';
+in
+{
+  options.services.gebos.emqx = {
+    enable = lib.mkEnableOption "EMQX broker (official container image) with native TLS";
+
+    image = lib.mkOption {
+      type = lib.types.str;
+      default = "emqx/emqx:5.8.6";
+      description = "Official EMQX image reference to run.";
+    };
+
+    domain = lib.mkOption {
+      type = lib.types.str;
+      example = "ingest.gebos.online";
+      description = "Public hostname the senders connect to; ACME cert is issued for it.";
+    };
+
+    tlsPort = lib.mkOption {
+      type = lib.types.port;
+      default = 8883;
+      description = "MQTTS listener port (TLS terminated by EMQX itself).";
+    };
+
+    port = lib.mkOption {
+      type = lib.types.port;
+      default = 1883;
+      description = "Plain MQTT TCP listener port (what the ingester connects to).";
+    };
+
+    bindAddress = lib.mkOption {
+      type = lib.types.str;
+      default = "127.0.0.1";
+      description = ''
+        Address the plain MQTT listener binds to. Defaults to loopback:
+        external clients only ever reach the TLS listener.
+      '';
+    };
+
+    persistentSessions = lib.mkOption {
+      type = lib.types.bool;
+      default = true;
+      description = ''
+        Whether client sessions and queued messages survive a restart
+        (EMQX durable sessions) or are kept only in memory.
+      '';
+    };
+
+    acmeEmail = lib.mkOption {
+      type = lib.types.str;
+      default = "ops@gebos.online";
+      description = "ACME account contact for the broker certificate.";
+    };
+
+    bootstrapUsersFile = lib.mkOption {
+      type = lib.types.path;
+      default = config.services.gebos.secrets.emqxBootstrapUsersFile;
+      defaultText = lib.literalExpression "config.services.gebos.secrets.emqxBootstrapUsersFile";
+      description = ''
+        CSV (user_id,password,is_superuser) seeding EMQX's built-in auth
+        database. Defaults to the sops-rendered secret; copied into the state
+        dir on service start so the container user can read it.
+      '';
+    };
+  };
+
+  config = lib.mkIf cfg.enable {
+    virtualisation.oci-containers = {
+      backend = "docker";
+      containers.emqx = {
+        image = cfg.image;
+        # Host networking: EMQX binds host ports directly, so loopback-only
+        # listeners (plain MQTT, dashboard) really are loopback-only and the
+        # TLS listener sees real client source addresses.
+        extraOptions = [ "--network=host" ];
+        volumes = [
+          "${emqxConf}:/opt/emqx/etc/emqx.conf:ro"
+          "${aclConf}:/etc/emqx/auth/acl.conf:ro"
+          "${stateDir}/auth/users.csv:/etc/emqx/auth/users.csv:ro"
+          "${stateDir}/certs:/etc/emqx/certs:ro"
+          "${stateDir}/data:/opt/emqx/data"
+        ];
+      };
+    };
+
+    # The oci-containers unit is docker-emqx.service; extend it to stage the
+    # writable state the container mounts. The bootstrap users file is copied
+    # (not symlinked) out of the sops tmpfs so it can be owned by the
+    # container's emqx uid.
+    systemd.services.docker-emqx = {
+      # Don't start before the first cert issuance has been attempted. If ACME
+      # fails (e.g. Porkbun creds not provisioned yet), EMQX crash-loops on the
+      # missing certfile until the cert lands — Restart=always retries.
+      after = [ "acme-finished-${cfg.domain}.target" ];
+      wants = [ "acme-finished-${cfg.domain}.target" ];
+      preStart = ''
+        install -d -m 750 -o ${toString emqxUid} -g ${toString emqxGid} ${stateDir}/data
+        install -d -m 755 ${stateDir}/auth ${stateDir}/certs
+        install -m 400 -o ${toString emqxUid} -g ${toString emqxGid} \
+          ${cfg.bootstrapUsersFile} ${stateDir}/auth/users.csv
+      '';
+    };
+
+    # Cert issuance + renewal via HTTP-01: lego's built-in standalone server
+    # answers the challenge on :80 (nothing else listens there — Caddy is gone
+    # from this host), so no DNS API secrets are needed; the host just keeps
+    # port 80 open alongside 8883. postRun is the deploy hook: install the
+    # PEMs at the fixed paths EMQX watches, owned by the container user. EMQX
+    # hot-reloads them; no restart or `emqx ctl` needed. RSA keys because
+    # embedded sender TLS stacks often lack ECDSA.
+    security.acme = {
+      acceptTerms = true;
+      certs.${cfg.domain} = {
+        email = cfg.acmeEmail;
+        listenHTTP = ":80";
+        keyType = "rsa2048";
+        postRun = ''
+          install -D -m 644 -o ${toString emqxUid} -g ${toString emqxGid} \
+            fullchain.pem ${stateDir}/certs/fullchain.pem
+          install -D -m 600 -o ${toString emqxUid} -g ${toString emqxGid} \
+            key.pem ${stateDir}/certs/privkey.pem
+        '';
+      };
+    };
+  };
+}

nix/modules/gebos-hivemq.nix

@@ -1,276 +0,0 @@
-{ config, lib, pkgs, ... }:
-
-let
-  cfg = config.services.gebos.hivemq;
-
-  # HiveMQ CE — fetched from the upstream GitHub release and unpacked into the
-  # Nix store. The store tree is read-only; the runtime data/log/extension
-  # folders are redirected to the writable state dir (see ExecStart below).
-  hivemq-ce = pkgs.stdenvNoCC.mkDerivation rec {
-    pname = "hivemq-ce";
-    version = "2026.5";
-
-    src = pkgs.fetchurl {
-      url = "https://github.com/hivemq/hivemq-community-edition/releases/download/${version}/hivemq-ce-${version}.zip";
-      hash = "sha256-/QBeU17KC6/7CF4RsVEohdBs4AeG+WFwHHZEmBb33sU=";
-    };
-
-    nativeBuildInputs = [ pkgs.unzip ];
-    dontConfigure = true;
-    dontBuild = true;
-
-    installPhase = ''
-      runHook preInstall
-      mkdir -p "$out"
-      cp -r . "$out/"
-      runHook postInstall
-    '';
-
-    meta = with lib; {
-      description = "HiveMQ Community Edition — open-source MQTT 3.x / 5 broker";
-      homepage = "https://github.com/hivemq/hivemq-community-edition";
-      license = licenses.asl20;
-      platforms = platforms.linux;
-    };
-  };
-
-  # HiveMQ file-RBAC extension — adds username/password authentication and
-  # topic-level authorization, replacing the insecure hivemq-allow-all that
-  # ships in the broker package. The zip unpacks to a single top-level
-  # `hivemq-file-rbac-extension/` folder which we keep verbatim in $out.
-  hivemq-file-rbac = pkgs.stdenvNoCC.mkDerivation rec {
-    pname = "hivemq-file-rbac-extension";
-    version = "4.6.15";
-
-    src = pkgs.fetchurl {
-      url = "https://github.com/hivemq/hivemq-file-rbac-extension/releases/download/${version}/hivemq-file-rbac-extension-${version}.zip";
-      hash = "sha256-R6PM+qAvEKMHdX3FTR0CfR8yarJrc2WeFAEu4eXH3Ps=";
-    };
-
-    nativeBuildInputs = [ pkgs.unzip ];
-    dontConfigure = true;
-    dontBuild = true;
-
-    # stdenv unpacks the single top-level archive dir and cds into it, so $PWD
-    # already holds the extension's files (jar, conf/, hivemq-extension.xml).
-    installPhase = ''
-      runHook preInstall
-      mkdir -p "$out/hivemq-file-rbac-extension"
-      cp -r . "$out/hivemq-file-rbac-extension/"
-      runHook postInstall
-    '';
-
-    meta = with lib; {
-      description = "HiveMQ file-based role-based access control extension";
-      homepage = "https://github.com/hivemq/hivemq-file-rbac-extension";
-      license = licenses.asl20;
-      platforms = platforms.linux;
-    };
-  };
-
-  stateDir = "/var/lib/gebos-hivemq";
-
-  # Non-secret RBAC extension config (reload interval + password format). The
-  # users/roles live in the secret credentials.xml, symlinked in at runtime.
-  rbacConfigXml = pkgs.writeText "hivemq-file-rbac-config.xml" ''
-    <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
-    <extension-configuration>
-        <credentials-reload-interval>${toString cfg.rbac.reloadInterval}</credentials-reload-interval>
-        <password-type>${cfg.rbac.passwordType}</password-type>
-    </extension-configuration>
-  '';
-
-  # config.xml rendered from the module options. The local TCP listener is what
-  # the ingester (and, eventually, the Caddy TLS frontend) connects to.
-  configXml = pkgs.writeText "hivemq-config.xml" ''
-    <?xml version="1.0"?>
-    <hivemq xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-            xsi:noNamespaceSchemaLocation="config.xsd">
-        <listeners>
-            <tcp-listener>
-                <port>${toString cfg.port}</port>
-                <bind-address>${cfg.bindAddress}</bind-address>
-            </tcp-listener>
-        </listeners>
-        <persistence>
-            <mode>${if cfg.persistentSessions then "file" else "in-memory"}</mode>
-        </persistence>
-        <anonymous-usage-statistics>
-            <enabled>false</enabled>
-        </anonymous-usage-statistics>
-    </hivemq>
-  '';
-
-  # Read-only conf folder: our generated config.xml plus the upstream
-  # logback.xml / config.xsd so HiveMQ logs and validates as shipped.
-  confDir = pkgs.runCommand "hivemq-conf" { } ''
-    mkdir -p "$out"
-    cp ${configXml} "$out/config.xml"
-    cp ${hivemq-ce}/conf/logback.xml "$out/logback.xml"
-    cp ${hivemq-ce}/conf/config.xsd "$out/config.xsd"
-  '';
-
-  # run.sh's JVM flags, reproduced so we can invoke java directly and point
-  # HiveMQ at the writable state dir via -Dhivemq.*.folder.
-  javaOpts = lib.concatStringsSep " " [
-    "-Djava.net.preferIPv4Stack=true"
-    "--add-opens java.base/java.lang=ALL-UNNAMED"
-    "--add-opens java.base/java.nio=ALL-UNNAMED"
-    "--add-opens java.base/sun.nio.ch=ALL-UNNAMED"
-    "--add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED"
-    "--add-exports java.base/jdk.internal.misc=ALL-UNNAMED"
-    "-Djava.security.egd=file:/dev/./urandom"
-    "-Duser.language=en -Duser.region=US"
-    "-XX:+CrashOnOutOfMemoryError"
-    "-XX:+HeapDumpOnOutOfMemoryError"
-    "-XX:HeapDumpPath=${stateDir}/log/heap-dump.hprof"
-    "-XX:ErrorFile=${stateDir}/log/hs_err_pid%p.log"
-  ];
-in
-{
-  options.services.gebos.hivemq = {
-    enable = lib.mkEnableOption "HiveMQ CE single-node broker";
-
-    package = lib.mkOption {
-      type = lib.types.package;
-      default = hivemq-ce;
-      defaultText = lib.literalExpression "pkgs.hivemq-ce (vendored)";
-      description = "HiveMQ CE package to run.";
-    };
-
-    port = lib.mkOption {
-      type = lib.types.port;
-      default = 1883;
-      description = "Plain MQTT TCP listener port (what the ingester connects to).";
-    };
-
-    bindAddress = lib.mkOption {
-      type = lib.types.str;
-      default = "127.0.0.1";
-      description = ''
-        Address the MQTT listener binds to. Defaults to loopback: external
-        clients reach the broker through the Caddy TLS frontend, not directly.
-      '';
-    };
-
-    persistentSessions = lib.mkOption {
-      type = lib.types.bool;
-      default = true;
-      description = ''
-        Whether client sessions and queued messages survive a restart
-        (file persistence) or are kept only in memory.
-      '';
-    };
-
-    tlsPort = lib.mkOption {
-      type = lib.types.port;
-      default = 8883;
-      description = ''
-        Externally advertised MQTTS port. TLS is currently terminated by the
-        Caddy frontend (see gebos-caddy), not by HiveMQ directly, so this value
-        is informational until a keystore is wired into the broker.
-      '';
-    };
-
-    rbac = {
-      package = lib.mkOption {
-        type = lib.types.package;
-        default = hivemq-file-rbac;
-        defaultText = lib.literalExpression "pkgs.hivemq-file-rbac (vendored)";
-        description = "HiveMQ file-RBAC authn/authz extension package.";
-      };
-
-      passwordType = lib.mkOption {
-        type = lib.types.enum [ "HASHED" "PLAIN" ];
-        default = "PLAIN";
-        description = ''
-          Whether passwords in the credentials file are PBKDF2 HASHED or PLAIN
-          text. PLAIN is acceptable here because the credentials file is a
-          sops-encrypted secret at rest and rendered 0400/gebos-hivemq into a
-          tmpfs at runtime; switch to HASHED for defence-in-depth.
-        '';
-      };
-
-      reloadInterval = lib.mkOption {
-        type = lib.types.ints.positive;
-        default = 60;
-        description = "Seconds between credentials.xml reloads by the extension.";
-      };
-
-      credentialsFile = lib.mkOption {
-        type = lib.types.path;
-        default = config.services.gebos.secrets.hivemqCredentialsFile;
-        defaultText = lib.literalExpression "config.services.gebos.secrets.hivemqCredentialsFile";
-        description = ''
-          Path to the file-RBAC credentials.xml (users, roles, topic
-          permissions). Defaults to the sops-rendered secret; symlinked into
-          the extension's conf/ dir so it never lands in the writable state dir.
-        '';
-      };
-    };
-  };
-
-  config = lib.mkIf cfg.enable {
-    users.users.gebos-hivemq = {
-      isSystemUser = true;
-      group = "gebos-hivemq";
-      description = "Gebos HiveMQ CE service user";
-    };
-    users.groups.gebos-hivemq = { };
-
-    systemd.services.gebos-hivemq = {
-      description = "HiveMQ CE";
-      after = [ "network.target" ];
-      wantedBy = [ "multi-user.target" ];
-
-      # HiveMQ wants the extensions folder writable (it manages enable/disable
-      # markers there), so the extension lives under the state dir. We reconcile
-      # it on every start: (re)install the vendored file-RBAC extension when its
-      # store path changes, refresh its non-secret config, link the secret
-      # credentials from the sops tmpfs path, and remove the insecure allow-all
-      # extension that earlier deploys seeded from the broker package.
-      preStart = ''
-        set -eu
-        mkdir -p ${stateDir}/{data,log,extensions}
-
-        extDir=${stateDir}/extensions/hivemq-file-rbac-extension
-        marker=${stateDir}/extensions/.rbac-version
-
-        if [ "$(cat "$marker" 2>/dev/null || true)" != "${cfg.rbac.package}" ]; then
-          rm -rf "$extDir"
-          cp -r --no-preserve=mode ${cfg.rbac.package}/hivemq-file-rbac-extension "$extDir"
-          echo "${cfg.rbac.package}" > "$marker"
-        fi
-
-        cp --no-preserve=mode ${rbacConfigXml} "$extDir/conf/config.xml"
-        ln -sf ${cfg.rbac.credentialsFile} "$extDir/conf/credentials.xml"
-
-        rm -rf ${stateDir}/extensions/hivemq-allow-all-extension
-      '';
-
-      serviceConfig = {
-        Type = "simple";
-        User = "gebos-hivemq";
-        Group = "gebos-hivemq";
-        StateDirectory = "gebos-hivemq";
-        WorkingDirectory = stateDir;
-        ExecStart = ''
-          ${pkgs.jre_headless}/bin/java \
-            -Dhivemq.home=${cfg.package} \
-            -Dhivemq.config.folder=${confDir} \
-            -Dhivemq.log.folder=${stateDir}/log \
-            -Dhivemq.data.folder=${stateDir}/data \
-            -Dhivemq.extensions.folder=${stateDir}/extensions \
-            ${javaOpts} \
-            -jar ${cfg.package}/bin/hivemq.jar
-        '';
-        Restart = "on-failure";
-        RestartSec = "5s";
-        # HiveMQ exits 143 on SIGTERM shutdown — treat that as success.
-        SuccessExitStatus = "143";
-        KillSignal = "SIGTERM";
-        LimitNOFILE = 65536;
-      };
-    };
-  };
-}

nix/modules/gebos-secrets.nix

@@ -17,75 +17,30 @@ let
   ];
   ingesterKeys = [ "ingester_postgres_password" ];
   postgresAdminKeys = [ "postgres_admin_password" ];
-  hivemqKeys = [ "hivemq_ingester_password" "hivemq_admin_password" "hivemq_device_password" ];
+  emqxKeys = [
+    "emqx_ingester_password"
+    "emqx_admin_password"
+    "emqx_device_password"
+  ];
 
-  # HiveMQ file-RBAC credentials.xml. Only the passwords are secret; the role/
-  # topic policy is structural. Devices publish under `acrios/<IMSI>/<metric>`
-  # (set by the sender firmware's mqtt_topic_base), so that is the namespace the
-  # ingester consumes and the device role is scoped to.
+  # EMQX authn bootstrap CSV (bootstrap_type = plain). Seeds the broker's
+  # built-in auth database on first start. Broker users:
+  #   ingester — the Go ingester; SUBSCRIBE the whole device tree (see the ACL
+  #              in gebos-emqx.nix).
+  #   admin    — operational break-glass; is_superuser bypasses the ACL.
+  #   bender   — shared device user for the whole fleet (Option A). Tenant
+  #              isolation is enforced downstream by the ingester's IMSI→tenant
+  #              registry, NOT at the broker. TODO(Option B): per-device users
+  #              (username == IMSI) — see the ACL comment in gebos-emqx.nix.
   #
-  # Roles:
-  #   ingest    — the Go ingester; SUBSCRIBE the whole device tree.
-  #   device    — physical senders; full access under acrios/ (publish telemetry
-  #               + the LWT/status and any downlink topics the firmware uses).
-  #   superuser — operational break-glass.
-  #
-  # Option A (current): one shared `device` user (`bender`) for the whole fleet.
-  # Tenant isolation is enforced downstream by the ingester's IMSI→tenant
-  # registry lookup, NOT at the broker — any device could publish under any IMSI.
-  #
-  # TODO(Option B / per-device auth): give each sender its own RBAC user with
-  # username == IMSI and scope its role to `acrios/${{username}}/#`, so the
-  # broker enforces that a device can only publish under its own IMSI (real
-  # anti-spoofing + per-device revocation). Requires setting mqtt_user = <IMSI>
-  # on each device and a per-device password provisioning flow (generate → sops
-  # → push to device). file-RBAC hot-reloads, so added users need no restart.
-  hivemqCredentialsXml = ingesterPassword: adminPassword: devicePassword: ''
-    <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
-    <file-rbac>
-        <users>
-            <user>
-                <name>ingester</name>
-                <password>${ingesterPassword}</password>
-                <roles><id>ingest</id></roles>
-            </user>
-            <user>
-                <name>admin</name>
-                <password>${adminPassword}</password>
-                <roles><id>superuser</id></roles>
-            </user>
-            <user>
-                <name>bender</name>
-                <password>${devicePassword}</password>
-                <roles><id>device</id></roles>
-            </user>
-        </users>
-        <roles>
-            <role>
-                <id>ingest</id>
-                <permissions>
-                    <permission>
-                        <topic>acrios/#</topic>
-                        <activity>SUBSCRIBE</activity>
-                    </permission>
-                </permissions>
-            </role>
-            <role>
-                <id>device</id>
-                <permissions>
-                    <!-- No <activity> = PUBLISH and SUBSCRIBE; covers telemetry,
-                         the LWT/status topic, and any downlink under acrios/. -->
-                    <permission><topic>acrios/#</topic></permission>
-                </permissions>
-            </role>
-            <role>
-                <id>superuser</id>
-                <permissions>
-                    <permission><topic>#</topic></permission>
-                </permissions>
-            </role>
-        </roles>
-    </file-rbac>
+  # Plaintext passwords are acceptable here because the file is sops-encrypted
+  # at rest and rendered 0400 into a tmpfs at runtime; passwords must not
+  # contain commas (CSV).
+  emqxUsersCsv = ingesterPassword: adminPassword: devicePassword: ''
+    user_id,password,is_superuser
+    ingester,${ingesterPassword},false
+    admin,${adminPassword},true
+    bender,${devicePassword},false
   '';
 
   # Render an env file body from a list of (key, envVarName) pairs.
@@ -112,7 +67,7 @@ let
        lib.optionals cfg.supabase      supabaseKeys
     ++ lib.optionals cfg.ingester      ingesterKeys
     ++ lib.optionals cfg.postgresAdmin postgresAdminKeys
-    ++ lib.optionals cfg.hivemq        hivemqKeys;
+    ++ lib.optionals cfg.emqx          emqxKeys;
 in
 {
   options.services.gebos.secrets = {
@@ -126,24 +81,24 @@ in
     postgresAdmin = lib.mkEnableOption ''
       Postgres bootstrap admin password applied via a one-shot ALTER ROLE unit.'';
 
-    hivemq = lib.mkEnableOption ''
-      HiveMQ file-RBAC credentials.xml (ingester + admin MQTT users), rendered
-      0400/gebos-hivemq for the broker's file-RBAC extension.'';
+    emqx = lib.mkEnableOption ''
+      EMQX broker secrets: the authn bootstrap users.csv (ingester + admin +
+      device MQTT users).'';
 
-    hivemqCredentialsFile = lib.mkOption {
+    emqxBootstrapUsersFile = lib.mkOption {
       type = lib.types.path;
       readOnly = true;
       description = ''
-        Path of the rendered HiveMQ file-RBAC credentials.xml consumed by the
-        gebos-hivemq broker. Resolves to the sops-templated secret when sops
-        material is present and the hivemq toggle is on, else to a baked-in
+        Path of the rendered EMQX authn bootstrap users.csv consumed by the
+        gebos-emqx broker. Resolves to the sops-templated secret when sops
+        material is present and the emqx toggle is on, else to a baked-in
         dev-defaults file (so first-boot eval and local VMs work).
       '';
       default =
-        if hasSecrets && cfg.hivemq
-        then config.sops.templates."hivemq-credentials".path
-        else "${pkgs.writeText "hivemq-credentials-dev.xml"
-          (hivemqCredentialsXml "dev-ingester" "dev-admin" "dev-device")}";
+        if hasSecrets && cfg.emqx
+        then config.sops.templates."emqx-users.csv".path
+        else "${pkgs.writeText "emqx-users-dev.csv"
+          (emqxUsersCsv "dev-ingester" "dev-admin" "dev-device")}";
     };
 
     envFile = lib.mkOption {
@@ -182,16 +137,15 @@ in
         owner = "root";
       };
 
-      # Rendered as a standalone file (not env) — the file-RBAC extension reads
-      # credentials.xml directly. Owned by the broker user so preStart can link
-      # it into the extension's conf/ dir.
-      templates."hivemq-credentials" = lib.mkIf cfg.hivemq {
-        content = hivemqCredentialsXml
-          config.sops.placeholder.hivemq_ingester_password
-          config.sops.placeholder.hivemq_admin_password
-          config.sops.placeholder.hivemq_device_password;
+      # Rendered as a standalone file — the gebos-emqx preStart copies it into
+      # the broker state dir, chowned to the container's emqx uid.
+      templates."emqx-users.csv" = lib.mkIf cfg.emqx {
+        content = emqxUsersCsv
+          config.sops.placeholder.emqx_ingester_password
+          config.sops.placeholder.emqx_admin_password
+          config.sops.placeholder.emqx_device_password;
         mode = "0400";
-        owner = "gebos-hivemq";
+        owner = "root";
       };
     };
   };