|
|
@@ -0,0 +1,323 @@
|
|
|
+# M4 Verification — MQTT ingest
|
|
|
+
|
|
|
+This is the live smoke test for the M4 milestone
|
|
|
+(**MQTT ingest**, SPEC §16 + §18 + §22). It exercises the
|
|
|
+new path end-to-end:
|
|
|
+
|
|
|
+1. **EMQX** is up with **per-source ACLs** (default deny on
|
|
|
+ `#`). The bootstrap file
|
|
|
+ `deploy/emqx/auth-built-in-db-bootstrap.csv` declares the
|
|
|
+ sources; `deploy/emqx/acl.conf` restricts each user to
|
|
|
+ their own `ba/<co>/<src>/incoming` topic.
|
|
|
+2. **`internal/mqttclient`** — shared paho.MQTT wrapper used
|
|
|
+ by both the ingestd subscriber and the new loadgen
|
|
|
+ publisher.
|
|
|
+3. **`cmd/ingestd/mqtt.go`** — MQTT subscriber that runs
|
|
|
+ every message through the same `processDeps.ProcessAlert`
|
|
|
+ pipeline as the HTTP POST handler (parse → validate →
|
|
|
+ HMAC verify → rate-limit → dedupe → publish to NATS).
|
|
|
+4. **`loadgen/cmd/mqtt/`** — `loadgen-mqtt` publisher. Same
|
|
|
+ data shape as `loadgen-http` (severity mix, dedupe ratio,
|
|
|
+ burst mode).
|
|
|
+5. **Topic namespace** — `ba/<company_id>/<source_id>/incoming`.
|
|
|
+ Sources publish here; `ingestd` subscribes to the wildcard
|
|
|
+ `ba/+/+/incoming`. The `+` is the EMQX single-level
|
|
|
+ wildcard.
|
|
|
+6. **Auth model** — twofold:
|
|
|
+ - **MQTT username/password** for the broker-level gate.
|
|
|
+ Username = `<source_id>-<company_id>`, password = HMAC
|
|
|
+ secret (so the same secret serves both as MQTT auth and
|
|
|
+ per-message HMAC). The `acl.conf` then gates which topic
|
|
|
+ a given user can publish to.
|
|
|
+ - **Per-message X-BA-Signature** in the JSON envelope's
|
|
|
+ `auth` field. Same `verifyHMAC()` is reused unchanged.
|
|
|
+
|
|
|
+## Prerequisites
|
|
|
+
|
|
|
+- Stack is up: `docker compose up -d`
|
|
|
+- All M1–M3 migrations applied: `docker compose up seed`
|
|
|
+- `BA_INGESTD_SOURCES` includes `acme-001:prom-prod:s3cret-acme`
|
|
|
+ (and optionally `globex-002:grafana:s3cret-globex` for the
|
|
|
+ ACL-violation test in step 5)
|
|
|
+- The M4 ingestd env vars are set:
|
|
|
+ ```
|
|
|
+ BA_INGESTD_MQTT_BROKER=tcp://emqx:1883
|
|
|
+ BA_INGESTD_MQTT_USERNAME=ingestd
|
|
|
+ BA_INGESTD_MQTT_PASSWORD=ingestd-broker-only
|
|
|
+ BA_INGESTD_MQTT_SUBSCRIBE=ba/+/+/incoming
|
|
|
+ ```
|
|
|
+- The EMQX bootstrap files are mounted (in compose by
|
|
|
+ default): `deploy/emqx/{acl.conf,auth-built-in-db-bootstrap.csv}`.
|
|
|
+- `loadgen-mqtt` is built:
|
|
|
+ ```bash
|
|
|
+ cd loadgen && go build -o /tmp/loadgen-mqtt ./cmd/mqtt
|
|
|
+ ```
|
|
|
+
|
|
|
+## Automated smoke
|
|
|
+
|
|
|
+```bash
|
|
|
+bash scripts/m4_smoke.sh
|
|
|
+```
|
|
|
+
|
|
|
+The script:
|
|
|
+1. Builds `loadgen-mqtt` and the 3 failure-path test
|
|
|
+ binaries (m4_badsig, m4_acl_violation, m4_badjson) if
|
|
|
+ not already present.
|
|
|
+2. Walks the 5 step-by-step scenarios below, asserting the
|
|
|
+ expected delivery counts and metric counter deltas.
|
|
|
+3. Exits 0 on success, $fail_count on failure.
|
|
|
+
|
|
|
+## Step-by-step (also documented in `M4_SMOKE_LOG.md`)
|
|
|
+
|
|
|
+### Step 1 — confirm M4 wiring is up
|
|
|
+
|
|
|
+```bash
|
|
|
+# ingestd's MQTT subscriber is connected
|
|
|
+docker logs broad-announce-ingestd-1 2>&1 | tail -3
|
|
|
+# should show: "mqtt connected" + "mqtt subscribed" + "ba/+/+/incoming"
|
|
|
+
|
|
|
+# EMQX dashboard is reachable
|
|
|
+curl -sS http://localhost:18083/
|
|
|
+# EMQX 5.x landing page
|
|
|
+
|
|
|
+# emqx is healthy
|
|
|
+docker inspect broad-announce-emqx-1 --format '{{.State.Health.Status}}'
|
|
|
+# "healthy"
|
|
|
+```
|
|
|
+
|
|
|
+### Step 2 — happy path: 1 alert via MQTT
|
|
|
+
|
|
|
+```bash
|
|
|
+/tmp/loadgen-mqtt \
|
|
|
+ --broker tcp://localhost:1883 \
|
|
|
+ --api-key acme-001:prom-prod:s3cret-acme \
|
|
|
+ --count 1 --rate 5
|
|
|
+```
|
|
|
+
|
|
|
+Wait 4s, then check `deliveries`:
|
|
|
+
|
|
|
+```bash
|
|
|
+docker exec -i broad-announce-postgres-1 psql -U ba -d ba -c "
|
|
|
+SELECT individual_id, channel, status FROM deliveries ORDER BY id;"
|
|
|
+```
|
|
|
+
|
|
|
+**Expected**: 2 rows — Alice fcm (sent) + Alice telegram
|
|
|
+(sent). Bob has no `telegram_chat_id` linked at this point
|
|
|
+(still in M2 default state for the smoke).
|
|
|
+
|
|
|
+Ingestd's MQTT counters tick:
|
|
|
+
|
|
|
+```bash
|
|
|
+curl -sS http://localhost:8800/metrics | grep ba_ingestd_mqtt
|
|
|
+# ba_ingestd_mqtt_messages_total{result="received",service="ingestd"} <N+1>
|
|
|
+# ba_ingestd_mqtt_messages_total{result="accepted",service="ingestd"} <M+1>
|
|
|
+```
|
|
|
+
|
|
|
+### Step 3 — 5 alerts, mixed severity, 30% dedupe
|
|
|
+
|
|
|
+```bash
|
|
|
+/tmp/loadgen-mqtt \
|
|
|
+ --broker tcp://localhost:1883 \
|
|
|
+ --api-key acme-001:prom-prod:s3cret-acme \
|
|
|
+ --count 5 --rate 5 --mode normal
|
|
|
+```
|
|
|
+
|
|
|
+**Expected**: 5 distinct `alert_id`s, ≥ 10 deliveries
|
|
|
+(5 alerts × 2 channels for Alice; +4 per alert with
|
|
|
+severity ≥ critical for Bob). At least 1 dedupe (30% of
|
|
|
+alerts share a `dedupe_key`).
|
|
|
+
|
|
|
+Ingestd counter `mqtt_messages{result="deduped"}` advances
|
|
|
+by ≥ 1.
|
|
|
+
|
|
|
+### Step 4 — bad signature
|
|
|
+
|
|
|
+Build the test binary (script does it automatically):
|
|
|
+
|
|
|
+```bash
|
|
|
+cat > /tmp/m4_smoke/badsig.go <<'GO'
|
|
|
+package main
|
|
|
+import (
|
|
|
+ "encoding/json"
|
|
|
+ "fmt"
|
|
|
+ mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
+ "os"
|
|
|
+ "time"
|
|
|
+)
|
|
|
+func main() {
|
|
|
+ opts := mqtt.NewClientOptions().
|
|
|
+ AddBroker("tcp://localhost:1883").
|
|
|
+ SetClientID("m4-badsig").
|
|
|
+ SetUsername("prom-prod-acme-001").
|
|
|
+ SetPassword("s3cret-acme").
|
|
|
+ SetCleanSession(true)
|
|
|
+ c := mqtt.NewClient(opts)
|
|
|
+ if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
|
|
|
+ fmt.Println("connect err:", tok.Error()); os.Exit(0)
|
|
|
+ }
|
|
|
+ body, _ := json.Marshal(map[string]any{
|
|
|
+ "alert": json.RawMessage(`{"company_id":"acme-001","source_id":"prom-prod","severity":"info","title":"bad sig"}`),
|
|
|
+ "auth": "t=1700000000,v1=deadbeef",
|
|
|
+ })
|
|
|
+ tok := c.Publish("ba/acme-001/prom-prod/incoming", 1, false, body)
|
|
|
+ if !tok.WaitTimeout(5*time.Second) { os.Exit(0) }
|
|
|
+ fmt.Println("bad-sig published; err:", tok.Error())
|
|
|
+ os.Exit(0)
|
|
|
+}
|
|
|
+GO
|
|
|
+(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badsig /tmp/m4_smoke/badsig.go)
|
|
|
+/tmp/m4_smoke/m4_badsig
|
|
|
+```
|
|
|
+
|
|
|
+**Expected**:
|
|
|
+- 0 deliveries (HMAC failed, alert dropped)
|
|
|
+- ingestd log: `WARN msg="mqtt alert rejected" reason=bad_signature`
|
|
|
+- `ba_ingestd_mqtt_messages_total{result="bad_signature"}` ticks
|
|
|
+
|
|
|
+### Step 5 — ACL violation
|
|
|
+
|
|
|
+```bash
|
|
|
+cat > /tmp/m4_smoke/acl_violation.go <<'GO'
|
|
|
+package main
|
|
|
+import (
|
|
|
+ "fmt"
|
|
|
+ mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
+ "os"
|
|
|
+ "time"
|
|
|
+)
|
|
|
+func main() {
|
|
|
+ opts := mqtt.NewClientOptions().
|
|
|
+ AddBroker("tcp://localhost:1883").
|
|
|
+ SetClientID("m4-acl-violation").
|
|
|
+ SetUsername("prom-prod-acme-001"). // <-- authed as acme source
|
|
|
+ SetPassword("s3cret-acme").
|
|
|
+ SetCleanSession(true)
|
|
|
+ c := mqtt.NewClient(opts)
|
|
|
+ if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
|
|
|
+ fmt.Println("connect err:", tok.Error()); os.Exit(0)
|
|
|
+ }
|
|
|
+ // Try to publish to GLOBEX's topic — should be denied by ACL
|
|
|
+ tok := c.Publish("ba/globex-002/grafana/incoming", 1, false, []byte("hack"))
|
|
|
+ if !tok.WaitTimeout(5*time.Second) { os.Exit(0) }
|
|
|
+ fmt.Println("published to wrong topic; err:", tok.Error())
|
|
|
+ os.Exit(0)
|
|
|
+}
|
|
|
+GO
|
|
|
+(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_acl_violation /tmp/m4_smoke/acl_violation.go)
|
|
|
+/tmp/m4_smoke/m4_acl_violation
|
|
|
+```
|
|
|
+
|
|
|
+**Expected**:
|
|
|
+- 0 deliveries (broker dropped the publish before the
|
|
|
+ subscriber saw it)
|
|
|
+- EMQX log: `tag: AUTHZ, msg: cannot_publish_to_topic_due_to_not_authorized, ... topic: ba/globex-002/grafana/incoming`
|
|
|
+- `ba_ingestd_mqtt_messages_total{result="received"}` is
|
|
|
+ **unchanged** (broker dropped before subscriber)
|
|
|
+
|
|
|
+### Step 6 — non-JSON body
|
|
|
+
|
|
|
+```bash
|
|
|
+cat > /tmp/m4_smoke/badjson.go <<'GO'
|
|
|
+package main
|
|
|
+import (
|
|
|
+ "fmt"
|
|
|
+ mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
+ "os"
|
|
|
+ "time"
|
|
|
+)
|
|
|
+func main() {
|
|
|
+ opts := mqtt.NewClientOptions().
|
|
|
+ AddBroker("tcp://localhost:1883").
|
|
|
+ SetClientID("m4-badjson").
|
|
|
+ SetUsername("prom-prod-acme-001").
|
|
|
+ SetPassword("s3cret-acme").
|
|
|
+ SetCleanSession(true)
|
|
|
+ c := mqtt.NewClient(opts)
|
|
|
+ if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
|
|
|
+ fmt.Println("connect err:", tok.Error()); os.Exit(0)
|
|
|
+ }
|
|
|
+ tok := c.Publish("ba/acme-001/prom-prod/incoming", 1, false, []byte("not json"))
|
|
|
+ if !tok.WaitTimeout(5*time.Second) { os.Exit(0) }
|
|
|
+ fmt.Println("bad-json published; err:", tok.Error())
|
|
|
+ os.Exit(0)
|
|
|
+}
|
|
|
+GO
|
|
|
+(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badjson /tmp/m4_smoke/badjson.go)
|
|
|
+/tmp/m4_smoke/m4_badjson
|
|
|
+```
|
|
|
+
|
|
|
+**Expected**:
|
|
|
+- 0 deliveries (alert.Alert.Unmarshal failed)
|
|
|
+- ingestd log: `WARN msg="mqtt alert rejected" reason=invalid_json`
|
|
|
+- `ba_ingestd_mqtt_messages_total{result="invalid_json"}` ticks
|
|
|
+
|
|
|
+## What this proves about M4
|
|
|
+
|
|
|
+1. **Per-source ACLs work.** The `acl.conf` file uses
|
|
|
+ Erlang-term rules keyed on the username. A user can
|
|
|
+ publish only to its own `ba/<co>/<src>/incoming`. Step 5
|
|
|
+ proves a different source can't publish to another
|
|
|
+ source's topic — the broker drops the publish before any
|
|
|
+ subscriber sees it.
|
|
|
+2. **The same pipeline serves HTTP and MQTT.** Steps 2 + 3
|
|
|
+ show the same `ProcessAlert` path that ran M0–M3's
|
|
|
+ HTTP-only smoke now runs the MQTT path with no
|
|
|
+ code changes to the rest of the stack (rate limits,
|
|
|
+ dedupe, NATS publish, alert.ID, alert.DedupeCount all
|
|
|
+ behave identically).
|
|
|
+3. **Per-message HMAC is enforced.** Step 4 proves that
|
|
|
+ even an authenticated source can't bypass per-message
|
|
|
+ auth by sending an arbitrary body.
|
|
|
+4. **Topic parsing is strict.** `parseIncomingTopic` only
|
|
|
+ accepts `ba/<co>/<src>/incoming` (4 segments, last =
|
|
|
+ `incoming`). Anything else lands in
|
|
|
+ `mqtt_messages{result="bad_topic"}` and is dropped. The
|
|
|
+ broker's ACL is the first line of defense; this is the
|
|
|
+ second.
|
|
|
+5. **Dedupe window is per-source, not per-transport.** The
|
|
|
+ same `dedupe.Deduper` (Redis-backed) is reused for HTTP
|
|
|
+ and MQTT. Step 3's 30% dedupe ratio produces the
|
|
|
+ expected dedupe count in the deliveries table.
|
|
|
+6. **Metrics are per-message.** `ba_ingestd_mqtt_messages_total{result="..."}`
|
|
|
+ exposes received/accepted/deduped/bad_signature/bad_topic/
|
|
|
+ invalid_json/... as separate labels. M9 promotes these
|
|
|
+ to the alerts-received total.
|
|
|
+
|
|
|
+## Known quirks (and why the smoke script does what it does)
|
|
|
+
|
|
|
+1. **faketgmd's `getUpdates` is a fake** (carried over from
|
|
|
+ M3). The smoke script still calls
|
|
|
+ `docker compose restart telegramd` between steps so the
|
|
|
+ bot's in-process long-poll offset aligns with the reset
|
|
|
+ state. Same quirk; same workaround.
|
|
|
+2. **EMQX's `deny_action = disconnect`** is set, but in
|
|
|
+ practice the paho client may reconnect fast enough that
|
|
|
+ the test code sees "still connected" after a denied
|
|
|
+ publish. The end-state is the same: the publish was
|
|
|
+ dropped, no delivery happened, and `received` counter
|
|
|
+ didn't tick.
|
|
|
+3. **The 3 failure-path test binaries** (`m4_badsig`,
|
|
|
+ `m4_acl_violation`, `m4_badjson`) are tiny, statically
|
|
|
+ linked, and live in `/tmp/m4_smoke/`. The smoke script
|
|
|
+ auto-builds them on first run. They are not committed
|
|
|
+ to the repo.
|
|
|
+4. **EMQX env-var config beats `emqx.conf`.** Earlier
|
|
|
+ versions of the compose tried to volume-mount
|
|
|
+ `emqx.conf`; EMQX 5.x rejects partial overrides there
|
|
|
+ with `node.cookie required_field` (because `emqx.conf`
|
|
|
+ is a *full* config, not a partial). The current
|
|
|
+ compose uses env vars
|
|
|
+ (`EMQX_AUTHENTICATION__1__BACKEND`,
|
|
|
+ `EMQX_AUTHORIZATION__NO_MATCH`, etc.) and the volume
|
|
|
+ mount is just `acl.conf` + the bootstrap CSV.
|
|
|
+
|
|
|
+## What stays out of M4 (and not supposed to be)
|
|
|
+
|
|
|
+- Per-IP concurrency cap (M5 with WS, SPEC §22 layer 2)
|
|
|
+- Circuit breaker + quarantine (M9)
|
|
|
+- TLS to EMQX (M11, security milestone)
|
|
|
+- Persistent sessions (M11)
|
|
|
+- Per-company bot token resolution for sources that share
|
|
|
+ a single EMQX user across companies (M3+ generalization)
|
|
|
+- HTTP-style signature in a header (MQTT has no headers;
|
|
|
+ the `auth` field in the JSON envelope is the equivalent)
|
netbot