Gebos/nix/modules/gebos-emqx.nix

{ 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
        '';
      };
    };
  };
}