mesh-lab/test/integration/assigned-catalogue-mqtt.test.ts

/**
 * The mesh assigns mosquitto, and a run-once step seeds the Dynamic Security store BEFORE the broker
 * starts — the end-to-end proof of the run-once primitive (novox/hq ADR 0052).
 *
 * mosquitto is the case ADR 0052 was written for. Its dynsec plugin loads at broker start and refuses
 * to bring the broker up unless `dynamic-security.json` already holds an admin client — a file that is
 * not state a reconcile loop can carry, but a step that must run once, after the data directory
 * exists and before the broker container starts. The manifest declares that step as a `run-once: true`
 * init container placed before the `server` (broker) container: the module's own runtime image,
 * carrying mosquitto's `bootstrap/` entrypoint and `mosquitto_ctrl`, run to completion under the
 * module's own account (ADR 0047). The host runs it, requires exit 0, and — because a failed step
 * gates the apply — only then reaches the broker.
 *
 * So the proof is layered, and the centre of it is that the broker came up at all:
 *   - the run-once step ran to completion and was not left running (a step, not a service);
 *   - the seed file is on disk, owned by the broker's uid so the broker can read it AND persist to it
 *     (the ownership question ADR 0052 left for the lab to settle — fixed in mosquitto's bootstrap);
 *   - the broker is up and STABLE — which it cannot be against an unseeded store, so a broker that
 *     stays up is itself the proof the seed ran at the right phase;
 *   - the node reached `current`, which the apply reaches only past a completed run-once step.
 * And, as a consumer-facing proof on top: the seeded admin authenticates over MQTT (the plugin loaded
 * the seed), and mosquitto's provisioner — running in the assigned runtime — creates a scoped client
 * for a contribution the mesh delivered, which then authenticates with the password the mesh minted.
 *
 *   MESH_LAB_HOST_BINARY=.../mesh-host   MESH_LAB_BUNDLE=.../examples/substrate-first-node.lock
 *   scripts/build-module-runtime.sh mosquitto builds mesh-runtime-mosquitto:development (carrying
 *   mosquitto_ctrl and the compiled bootstrap entrypoint) into the local daemon, which
 *   scenarios/catalogue-mqtt.yml stocks. eclipse-mosquitto:2 must be in the local daemon to be
 *   stocked; the host pulls both from the scenario's own registry by digest.
 */

import { test, before, after } from "node:test";
import assert from "node:assert/strict";
import { existsSync, readFileSync } from "node:fs";
import { loadScenario } from "../../src/declaration/parse.ts";
import { raise } from "../../src/lifecycle/raise.ts";
import { destroy, exec } from "../../src/lifecycle/operate.ts";
import { hostBinaryPath, HOST_PATH } from "../../src/lifecycle/place.ts";
import { labIsUsable, destroyAll } from "./harness.ts";

const capability = await labIsUsable();
const binary = hostBinaryPath();
const bundle = process.env["MESH_LAB_BUNDLE"] ?? "";

const skip = !capability.usable
  ? `lab not usable: ${capability.why}`
  : !binary || !existsSync(binary)
    ? "MESH_LAB_HOST_BINARY is not set to a built mesh-host"
    : !bundle || !existsSync(bundle)
      ? "MESH_LAB_BUNDLE is not set to a substrate bundle (mesh-host examples/)"
      : false;

const SCENARIO = "catalogue-mqtt";
const MACHINE = "anchor";

let instanceId = "";
let stocked: string[] = [];

function quote(s: string): string {
  return `'${s.replaceAll("'", `'\\''`)}'`;
}

async function on(command: string, timeoutMs?: number): Promise<{ out: string; ok: boolean }> {
  const { stdout } = await exec(instanceId, MACHINE, [
    "sh", "-c", `exec 2>&1\n${command}\necho "__exit=$?"`,
  ], timeoutMs);
  const marker = stdout.lastIndexOf("__exit=");
  if (marker < 0) return { out: stdout, ok: false };
  return { out: stdout.slice(0, marker), ok: stdout.slice(marker + 7).trim() === "0" };
}

async function must(command: string, timeoutMs?: number): Promise<string> {
  const { out, ok } = await on(command, timeoutMs);
  if (!ok) throw new Error(`${MACHINE}: ${command}\n${out}`);
  return out;
}

/** The control plane, a container on the node. */
async function mesh(command: string, timeoutMs?: number): Promise<string> {
  return must(`docker exec mesh-control /mesh-control ${command}`, timeoutMs);
}

/** The pinned reference for one of the scenario's images, by repository. */
function pinned(repository: string): string {
  const found = stocked.find((r) => r.slice(r.indexOf("/") + 1, r.indexOf("@")) === repository);
  assert.ok(found, `the scenario stocks no ${repository}; it serves ${stocked.join(", ")}`);
  return found;
}

/** The substrate bundle, its image references pointed at this scenario's own registry. */
function bundleFor(images: string[]): string {
  let text = readFileSync(bundle, "utf8");
  for (const ref of images) {
    const repository = ref.slice(ref.indexOf("/") + 1, ref.indexOf("@"));
    const escaped = repository.replaceAll("/", "\\/").replaceAll(".", "\\.");
    text = text.replaceAll(new RegExp(`[A-Za-z0-9_.:-]+\\/${escaped}@sha256:[0-9a-f]+`, "g"), ref);
  }
  return text;
}

function tokenFrom(said: string): string {
  const found = said.split("\n").map((l) => l.trim()).find((l) => l.length > 100 && !l.includes(" "));
  assert.ok(found, `no token in:\n${said}`);
  return found;
}

async function settled(withinMs = 600_000): Promise<void> {
  const until = Date.now() + withinMs;
  let last = "";
  while (Date.now() < until) {
    const asked = await on(`docker exec mesh-control /mesh-control status --json`);
    if (asked.ok) {
      try {
        const state = JSON.parse(asked.out) as {
          wrong: { node: string; outcome: string }[];
          waiting: { node: string }[];
          reported: { node: string; outcome: string; current: boolean }[];
        };
        const bad = state.wrong.find((w) => w.node === MACHINE);
        if (bad) throw new Error(`${MACHINE} did not apply what it was sent: ${bad.outcome}\n${asked.out}`);
        const word = state.reported.find((r) => r.node === MACHINE);
        if (!state.waiting.some((w) => w.node === MACHINE) && word?.outcome === "applied" && word.current) return;
        last = asked.out;
      } catch (err) {
        if (err instanceof Error && err.message.includes("did not apply")) throw err;
        last = asked.out;
      }
    }
    await new Promise((r) => setTimeout(r, 5000));
  }
  throw new Error(`${MACHINE} never caught up within ${Math.round(withinMs / 1000)}s. Last:\n${last}`);
}

before(async () => {
  if (skip) return;

  const raised = await raise(loadScenario(`scenarios/${SCENARIO}.yml`), {
    onProgress: (m) => console.log(`raise: ${m}`),
  });
  instanceId = raised.instanceId;
  stocked = raised.images;

  // Raise the substrate — store, broker, control — from the bundle.
  await must(`cat > /tmp/substrate.lock <<'MESHBUNDLE'\n${bundleFor(raised.images)}\nMESHBUNDLE`);
  await must(`${HOST_PATH} apply /tmp/substrate.lock`, 600_000);
  const up = await must(`docker ps --format '{{.Names}}'`);
  for (const c of ["mesh-store", "mesh-broker", "mesh-control"]) {
    assert.match(up, new RegExp(c), `the substrate did not raise ${c}:\n${up}`);
  }

  // The node joins its own mesh and starts the host so it applies what it is pushed.
  await mesh(`node add ${MACHINE}`);
  const token = tokenFrom(await mesh(`token issue --node ${MACHINE}`));
  await must(`${HOST_PATH} enrol --token ${quote(token)}`);
  await must(`nohup ${HOST_PATH} run > /var/log/mesh-host.log 2>&1 & sleep 3`);
}, { timeout: 1_800_000 });

after(async () => {
  if (instanceId) await destroy(instanceId);
  await destroyAll(`${SCENARIO}-`);
}, { timeout: 600_000 });

test("the mesh assigns mosquitto: a run-once step seeds dynsec before the broker, and the broker comes up", {
  skip, timeout: 1_500_000,
}, async () => {
  // mosquitto's dynsec config: the plugin refuses to start unless dynamic-security.json holds an
  // admin client, so the store MUST be seeded first. The `run-once` bootstrap container is declared
  // BEFORE `server` (the broker) and reuses the module's runtime image; the host runs it to
  // completion, then starts the broker. The runtime `server`/`runtime` shape mirrors the committed
  // manifest, with images pinned to what this scenario serves by digest.
  const mosquittoConf =
    "persistence true\n" +
    "persistence_location /mosquitto/data\n\n" +
    "log_dest stdout\n" +
    "log_type warning\n" +
    "log_type error\n" +
    "log_type notice\n\n" +
    "# Every client authenticates; identities and their per-topic ACLs are managed\n" +
    "# at runtime by the dynamic security plugin, whose store the plugin itself owns.\n" +
    "allow_anonymous false\n" +
    "plugin /usr/lib/mosquitto_dynamic_security.so\n" +
    "plugin_opt_config_file /mosquitto/data/dynamic-security.json\n\n" +
    "# MQTT listener\n" +
    "listener 1883\n\n" +
    "# MQTT-over-WebSockets listener\n" +
    "listener 8081\n" +
    "protocol websockets\n";

  const manifest = JSON.stringify({
    module: "mosquitto",
    version: "1",
    provides: [{ name: "mqtt-topic", scope: "mesh" }],
    serves: { "mqtt-topic": {} },
    emits: ["module.mosquitto.topic.provisioned", "module.mosquitto.topic.deprovisioned"],
    // The events entrypoint subscribes to its own lifecycle events (an audit log), so it consumes
    // them too — declared, or the substrate never makes the queue the runtime binds (ADR 0046).
    consumes: ["module.mosquitto.topic.provisioned", "module.mosquitto.topic.deprovisioned"],
    receives: { "mqtt-topic": "/var/lib/mosquitto-module/grants/mesh.json" },
    grants: { "mqtt-topic": "/var/lib/mosquitto-module/grants" },
    "own-secrets": {
      admin: "/var/lib/mosquitto-module/admin.secret",
      broker: "/var/lib/mesh/mosquitto/broker",
    },
    resources: [
      { id: "mesh-state", type: "directory", path: "/var/lib/mesh/mosquitto", mode: "0700" },
      { id: "state", type: "directory", path: "/var/lib/mosquitto-module", mode: "0700" },
      { id: "grants-dir", type: "directory", path: "/var/lib/mosquitto-module/grants", mode: "0700" },
      // The broker runs as uid 1883, so the shared data directory it seeds into and persists to is
      // its own.
      { id: "data", type: "directory", path: "/services/mosquitto/data", mode: "0700", owner: "1883:1883" },
      {
        id: "server-conf", type: "file", path: "/var/lib/mosquitto-module/mosquitto.conf",
        mode: "0600", owner: "1883:1883", content: mosquittoConf,
      },
      { id: "net", type: "network", name: "mosquitto" },
      // THE run-once step: seed dynsec offline, once, before the broker. It reuses the runtime image
      // (`mesh-tools run <bootstrap>` imports mosquitto's bootstrap entrypoint, which writes the
      // admin client into dynamic-security.json and chowns it to the broker's uid, then exits). It is
      // declared BEFORE `server`; the host runs it to completion and requires exit 0 before starting
      // the broker.
      {
        id: "bootstrap", type: "container", name: "mosquitto-bootstrap",
        image: pinned("mesh-runtime-mosquitto"), "run-once": true,
        volumes: [
          "/services/mosquitto/data:/mosquitto/data",
          "/var/lib/mosquitto-module/admin.secret:/run/secrets/admin:ro",
        ],
        env: {
          MESH_PROVISION_MQTT: "mosquitto:1883",
          MESH_PROVISION_ADMIN_USER: "mesh-admin",
          MESH_PROVISION_PASSWORD_FILE: "/run/secrets/admin",
          MESH_DYNSEC_FILE: "/mosquitto/data/dynamic-security.json",
        },
        args: ["run", "/app/modules/mosquitto/dist/bootstrap/index.js"],
      },
      {
        id: "server", type: "container", name: "mosquitto", image: pinned("eclipse-mosquitto"),
        network: "mosquitto", ports: ["1883", "8081"],
        volumes: [
          "/services/mosquitto/data:/mosquitto/data",
          "/var/lib/mosquitto-module/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro",
        ],
      },
      {
        id: "runtime", type: "container", name: "mesh-mosquitto",
        image: pinned("mesh-runtime-mosquitto"), network: "mosquitto",
        volumes: [
          "/var/lib/mesh/mosquitto/broker:/run/secrets/broker:ro",
          "/var/lib/mosquitto-module/grants:/var/lib/mosquitto-module/grants:ro",
          "/var/lib/mosquitto-module/admin.secret:/run/secrets/admin:ro",
        ],
        env: {
          MESH_BROKER_FILE: "/run/secrets/broker",
          MESH_RECEIVES: "/var/lib/mosquitto-module/grants/mesh.json",
          MESH_PROVISION_MQTT: "mosquitto:1883",
          MESH_PROVISION_ADMIN_USER: "mesh-admin",
          MESH_PROVISION_PASSWORD_FILE: "/run/secrets/admin",
        },
      },
    ],
  });

  await must(`printf %s ${quote(manifest)} > /tmp/mosquitto.json && docker cp /tmp/mosquitto.json mesh-control:/mosquitto.json`);
  await mesh("module add /mosquitto.json");
  const issued = await mesh(`module issue mosquitto --node ${MACHINE}`);
  assert.match(issued, /scoped to what it emits and consumes/, issued);
  await mesh(`assign ${MACHINE} mosquitto`);

  // ONE push, ONE convergence — and reaching `current` requires the run-once step to have completed,
  // because the apply is gated on it and stops if it fails.
  await mesh(`push ${MACHINE}`);
  await settled();

  // --- the run-once step ran to completion, and is NOT left running -------------------------------
  // A step is run in the foreground and removed once it exits 0; a service is left running. So the
  // bootstrap container must be gone from `docker ps -a`, not lingering.
  const allContainers = await must(`docker ps -a --format '{{.Names}}'`);
  assert.doesNotMatch(allContainers, /(^|\n)mosquitto-bootstrap(\n|$)/,
    `the run-once bootstrap was left as a container instead of run to completion:\n${allContainers}`);

  // --- the seed is on disk, owned by the broker's uid so the broker can read AND persist it --------
  const seedPath = "/services/mosquitto/data/dynamic-security.json";
  const seedExists = await on(`test -f ${seedPath}`);
  assert.ok(seedExists.ok,
    `the run-once step did not seed ${seedPath}:\n${(await on(`tail -40 /var/log/mesh-host.log`)).out}`);
  const owner = (await must(`stat -c '%u:%g' ${seedPath}`)).trim();
  assert.equal(owner, "1883:1883",
    `the seeded store is not owned by the broker's uid, so the broker cannot read/persist it: ${owner}`);
  const seed = await must(`cat ${seedPath}`);
  assert.match(seed, /mesh-admin/, `the seed does not hold the dynsec admin client:\n${seed}`);

  // --- the broker is up and STABLE — the proof the seed was loaded --------------------------------
  // An unseeded store crash-loops the broker: dynsec aborts start. So a broker that is up now, and
  // still up after a moment with no restart, is the end-to-end proof the store was seeded before it
  // started.
  const running = await must(`docker ps --format '{{.Names}}'`);
  for (const name of ["mosquitto", "mesh-mosquitto"]) {
    assert.match(running, new RegExp(`(^|\\n)${name}(\\n|$)`),
      `${name} is not running after the push:\n${running}\n---host log---\n${(await on(`tail -40 /var/log/mesh-host.log`)).out}`);
  }
  await new Promise((r) => setTimeout(r, 8000));
  const brokerState = (await must(`docker inspect -f '{{.State.Running}} {{.RestartCount}}' mosquitto`)).trim();
  assert.equal(brokerState, "true 0",
    `the broker is not up and stable — it crash-loops against an unseeded store:\n${brokerState}\n` +
      `${(await on(`docker logs mosquitto 2>&1 | tail -30`)).out}`);
  const brokerLog = (await on(`docker logs mosquitto 2>&1`)).out;
  assert.doesNotMatch(brokerLog, /Error found at|Unable to load|unable to load Dynamic security/i,
    `the broker reported a dynsec load error against the seeded store:\n${brokerLog}`);

  // --- the scoped broker account exists ------------------------------------------------------------
  const users = await must(`docker exec mesh-broker lavinmqctl list_users 2>&1`);
  assert.match(users, /anchor-mosquitto/, `the scoped account is not on the broker:\n${users}`);

  // --- the seeded admin authenticates over MQTT: the plugin loaded the seed ------------------------
  const adminPw = (await must(`cat /var/lib/mosquitto-module/admin.secret`)).trim();
  let adminClients = { out: "", ok: false };
  const untilAdmin = Date.now() + 60_000;
  while (Date.now() < untilAdmin) {
    adminClients = await on(
      `docker exec mosquitto mosquitto_ctrl -h 127.0.0.1 -p 1883 -u mesh-admin -P ${quote(adminPw)} dynsec listClients 2>&1`,
    );
    if (adminClients.ok && /mesh-admin/.test(adminClients.out)) break;
    await new Promise((r) => setTimeout(r, 3000));
  }
  assert.match(adminClients.out, /mesh-admin/,
    `the seeded admin could not authenticate / dynsec did not load the seed:\n${adminClients.out}`);

  // --- the provider path: a consumer the mesh delivered gets a scoped client that authenticates ----
  // mosquitto's provisioner reconciles the contributions file the mesh writes at MESH_RECEIVES: for
  // each consumer it reads the login the mesh derived and the mesh-minted password the host unsealed,
  // and creates exactly that dynsec client, scoped to its own topic subtree (ADR 0048). A
  // hand-written contributions file + secret stand in for the control plane's write; their SHAPE is
  // what mesh-control produces. The proof is authentication as the consumer with the mesh's password
  // — a provisioner that invented its own would refuse the connection.
  const consumerPw = "mesh-minted-mqtt-7b2e1a";
  await must(`printf %s ${quote(consumerPw)} > /var/lib/mosquitto-module/grants/app.secret`);
  const contributions = JSON.stringify({
    contributions: 1,
    requirement: "mqtt-topic",
    generated: "by the mesh — do not edit",
    given: [
      { as: "app-one", node: "app-node", at: "192.0.2.20:1883", secret: "/var/lib/mosquitto-module/grants/app.secret", values: {} },
    ],
  });
  await must(`printf %s ${quote(contributions)} > /var/lib/mosquitto-module/grants/mesh.json`);

  let created = { out: "", ok: false };
  const untilCreated = Date.now() + 90_000;
  while (Date.now() < untilCreated) {
    created = await on(
      `docker exec mosquitto mosquitto_ctrl -h 127.0.0.1 -p 1883 -u mesh-admin -P ${quote(adminPw)} dynsec listClients 2>&1`,
    );
    if (created.ok && /app-one/.test(created.out)) break;
    await new Promise((r) => setTimeout(r, 3000));
  }
  assert.match(created.out, /app-one/,
    `mosquitto's provisioner never created the consumer's scoped client from the delivered contribution:\n` +
      `${(await on(`docker logs mesh-mosquitto 2>&1 | tail -30`)).out}\n---\n${created.out}`);

  // Authenticate as that consumer with the password the MESH minted, and publish to its own subtree.
  // A successful connect+publish (exit 0) means the provisioner created the client with exactly that
  // password; a wrong password would refuse the connection.
  const published = await on(
    `docker exec mosquitto mosquitto_pub -h 127.0.0.1 -p 1883 -u app-one -P ${quote(consumerPw)} -q 1 -t app-one/hello -m hi 2>&1`,
  );
  assert.ok(published.ok,
    `the consumer could not authenticate as app-one with the mesh-minted password and publish to its subtree:\n${published.out}`);
});