NOISSUE - Update FMQ version

Signed-off-by: dusan <borovcanindusan1@gmail.com>

Makefile

@@ -197,7 +197,14 @@ FILTERED_SERVICES = $(filter-out $(RUN_ADDON_ARGS), $(SERVICES))
 
 all: $(SERVICES)
 
-.PHONY: all $(SERVICES) dockers dockers_dev latest release provision_atom_tokens provision-atom-tokens migrate_atom run_latest run_latest_ci run_tls run_stable run_addons grpc_mtls_certs check_mtls check_certs test_api mocks
+.PHONY: all help $(SERVICES) dockers dockers_dev latest release provision_atom_tokens provision-atom-tokens migrate_atom run_latest run_latest_ci run_tls run_stable run_addons grpc_mtls_certs check_mtls check_certs check_fluxmq_service_certs check_re_trace_key test_api mocks
+
+help:
+	@printf 'Usage:\n  make <target> [VARIABLE=value ...]\n\nAvailable targets:\n'
+	@$(MAKE) -qpRr : 2>/dev/null | \
+		awk -F: '/^[[:alnum:]_][^$$#\/\t=]*:([^=]|$$)/ { split($$1, targets, /[[:space:]]+/); for (i in targets) if (targets[i] != "") print targets[i] }' | \
+		LC_ALL=C sort -u | \
+		awk '$$0 != "Makefile" { printf "  make %s\n", $$0 }'
 
 clean:
 	rm -rf ${BUILD_DIR}
@@ -349,7 +356,41 @@ else
 	$(eval GRPC_MTLS :=)
 endif
 
-check_certs: check_mtls check_tls
+# Internal services reach FluxMQ on the mTLS local listener, and Compose mounts
+# both sides of those connections. Certificates and principal secrets are
+# generated rather than committed, so make them before anything binds them.
+check_fluxmq_service_certs:
+ifeq ("$(wildcard docker/ssl/certs/re-fluxmq-client.crt)","")
+	$(MAKE) -C docker/ssl fluxmq_service_certs
+endif
+ifeq ("$(wildcard docker/ssl/certs/timescale-writer-fluxmq-client.crt)","")
+	$(MAKE) -C docker/ssl timescale_writer_fluxmq_client_cert
+endif
+ifeq ("$(wildcard docker/ssl/certs/postgres-writer-fluxmq-client.crt)","")
+	$(MAKE) -C docker/ssl postgres_writer_fluxmq_client_cert
+endif
+ifeq ("$(wildcard docker/ssl/certs/fluxmq-auth-fluxmq-client.crt)","")
+	$(MAKE) -C docker/ssl fluxmq_auth_fluxmq_client_cert
+endif
+ifeq ("$(wildcard docker/fluxmq/secrets/re-current)","")
+	$(MAKE) -C docker/ssl fluxmq_service_secret
+endif
+ifeq ("$(wildcard docker/fluxmq/secrets/timescale-writer-current)","")
+	$(MAKE) -C docker/ssl timescale_writer_fluxmq_service_secret
+endif
+ifeq ("$(wildcard docker/fluxmq/secrets/postgres-writer-current)","")
+	$(MAKE) -C docker/ssl postgres_writer_fluxmq_service_secret
+endif
+ifeq ("$(wildcard docker/fluxmq/secrets/fluxmq-auth-current)","")
+	$(MAKE) -C docker/ssl fluxmq_auth_fluxmq_service_secret
+endif
+
+check_re_trace_key:
+ifeq ("$(wildcard docker/re/secrets/trace.key)","")
+	$(MAKE) -C docker/ssl re_trace_key
+endif
+
+check_certs: check_mtls check_tls check_fluxmq_service_certs check_re_trace_key
 ifeq ($(GRPC_MTLS_CERT_FILES_EXISTS),0)
 ifeq ($(filter true,$(GRPC_MTLS) $(GRPC_TLS)),true)
 ifeq ($(filter $(DEFAULT_DOCKER_COMPOSE_COMMAND),$(DOCKER_COMPOSE_COMMAND)),$(DEFAULT_DOCKER_COMPOSE_COMMAND))
@@ -368,7 +409,7 @@ run_latest_ci: check_certs
 	$(SED_INPLACE) 's/^MG_RELEASE_TAG=.*/MG_RELEASE_TAG=latest/' docker/.env
 	$(DOCKER_PLATFORM) docker compose -f docker/docker-compose.yaml -f docker/docker-compose-ci.yaml $(DOCKER_ENV_FILES) -p $(DOCKER_PROJECT) $(DOCKER_COMPOSE_COMMAND) $(args)
 
-run_tls:
+run_tls: check_certs
 	@test -n "$(host)" || (echo "Usage: make run_tls host=example.com [email=admin@example.com] [letsencrypt=false] [staging=true] [force=true]" && exit 2)
 	@if [ "$(or $(letsencrypt),true)" != "false" ] && [ -z "$(email)" ]; then echo "Usage: make run_tls host=example.com email=admin@example.com [letsencrypt=false] [staging=true] [force=true]"; exit 2; fi
 	MG_PUBLIC_HOST="$(host)" \

README.md

@@ -116,14 +116,14 @@ Magistrala provides a complete set of building blocks for IoT systems — from d
 
 ### ⚙️ Processing & Automation
 
-- Rules engine for message processing and routing
-- Alarms and triggers for reacting to events
+- Rules engine for message processing and routing (Enterprise Edition)
+- Alarms and triggers for reacting to events (Enterprise Edition)
 - Scheduled actions for time-based workflows
 - Event-driven architecture as the foundation
 
 ### 📊 Observability
 
-- Audit logs for tracking system activity
+- Audit logs for tracking system activity (Enterprise Edition)
 - Metrics and tracing via Prometheus and OpenTelemetry
 - Built-in visibility into system behavior and data flows
 
@@ -160,13 +160,13 @@ Current Docker deployments use the Atom image configured by `ATOM_IMAGE` in `doc
 
 ### Core Entity Mapping
 
-| Magistrala concept | Atom concept | Meaning |
-|--------------------|--------------|---------|
-| Domain | Tenant | Isolation boundary for one organization, project, or environment |
-| User | Entity with kind `human` | A person who logs in and uses the UI/API |
-| Client | Entity with kind `device` | A device or application that sends/receives data |
-| Channel | Resource with kind `channel` | A messaging/data path that clients can publish or subscribe to |
-| Group | Group | A collection of users, clients, channels, or other grouped objects |
+| Magistrala concept | Atom concept                 | Meaning                                                            |
+| ------------------ | ---------------------------- | ------------------------------------------------------------------ |
+| Domain             | Tenant                       | Isolation boundary for one organization, project, or environment   |
+| User               | Entity with kind `human`     | A person who logs in and uses the UI/API                           |
+| Client             | Entity with kind `device`    | A device or application that sends/receives data                   |
+| Channel            | Resource with kind `channel` | A messaging/data path that clients can publish or subscribe to     |
+| Group              | Group                        | A collection of users, clients, channels, or other grouped objects |
 
 In simple terms:
 
@@ -182,12 +182,12 @@ MG Group   = Atom Group
 
 Atom access control has these basic parts:
 
-| Atom word | Simple meaning | Example |
-|-----------|----------------|---------|
-| Action | One permission verb | `read`, `write`, `delete`, `role.manage`, `policy.manage` |
-| Permission Block | Where actions apply | all channels in domain `d1` can `read`, `publish` |
-| Role | A bundle of permission blocks | `tenant-admin` bundles domain, role, and member access |
-| Role Assignment | Who gets a role | give `user1` the `tenant-admin` role |
+| Atom word        | Simple meaning                | Example                                                   |
+| ---------------- | ----------------------------- | --------------------------------------------------------- |
+| Action           | One permission verb           | `read`, `write`, `delete`, `role.manage`, `policy.manage` |
+| Permission Block | Where actions apply           | all channels in domain `d1` can `read`, `publish`         |
+| Role             | A bundle of permission blocks | `tenant-admin` bundles domain, role, and member access    |
+| Role Assignment  | Who gets a role               | give `user1` the `tenant-admin` role                      |
 
 Read an assignment like this:
 
@@ -221,15 +221,15 @@ MG UI shows actions such as:
 
 These are mapped to Atom actions:
 
-| MG action | Atom action |
-|-----------|-----------------|
-| view/read | `read` |
-| create/update/edit/connect | `write` |
-| delete/remove | `delete` |
-| manage roles | `role.manage` |
+| MG action                    | Atom action     |
+| ---------------------------- | --------------- |
+| view/read                    | `read`          |
+| create/update/edit/connect   | `write`         |
+| delete/remove                | `delete`        |
+| manage roles                 | `role.manage`   |
 | add/remove members or access | `policy.manage` |
-| channel publish | `publish` |
-| channel subscribe | `subscribe` |
+| channel publish              | `publish`       |
+| channel subscribe            | `subscribe`     |
 
 So when MG UI checks:
 
@@ -280,9 +280,122 @@ Then user2 can read only that channel, not the whole domain.
 ```bash
 git clone https://github.com/absmach/magistrala.git
 cd magistrala
+make provision_atom_tokens
 make run_latest
 ```
 
+A fresh clone carries no generated secrets. Two sets have to exist before the
+stack can start — certificates and keys the internal services authenticate
+with, and the Atom service tokens each service presents to Atom. `make
+run_latest` produces the first set itself but expects the second to be there
+already, which is why the token step comes first above.
+
+### Certificates, broker secret and trace key
+
+Generated by `make run_latest`, or on demand:
+
+```bash
+make check_certs
+```
+
+This creates whatever is missing and leaves anything already present alone:
+
+| Path                                                        | What it is                                                                |
+| ----------------------------------------------------------- | ------------------------------------------------------------------------- |
+| `docker/ssl/certs/fluxmq-service-server.{crt,key}`          | Server certificate for FluxMQ's mTLS service listener                     |
+| `docker/ssl/certs/re-fluxmq-client.{crt,key}`               | Client certificate whose URI SAN identifies the Rules Engine              |
+| `docker/ssl/certs/timescale-writer-fluxmq-client.{crt,key}` | Client certificate whose URI SAN identifies the Timescale writer          |
+| `docker/ssl/certs/postgres-writer-fluxmq-client.{crt,key}`  | Client certificate whose URI SAN identifies the Postgres writer           |
+| `docker/ssl/certs/fluxmq-auth-fluxmq-client.{crt,key}`      | Client certificate whose URI SAN identifies the publish proxy            |
+| `docker/fluxmq/secrets/re-current`                          | Rules Engine principal secret, from `MG_RE_BROKER_SECRET`                 |
+| `docker/fluxmq/secrets/timescale-writer-current`            | Timescale writer secret, from `MG_TIMESCALE_WRITER_BROKER_SECRET`         |
+| `docker/fluxmq/secrets/postgres-writer-current`             | Postgres writer secret, from `MG_POSTGRES_WRITER_BROKER_SECRET`           |
+| `docker/fluxmq/secrets/fluxmq-auth-current`                 | Publish proxy secret, from `MG_FLUXMQ_BROKER_SECRET`                     |
+| `docker/re/secrets/trace.key`                               | HMAC key the Rules Engine signs its loop-detection traces with            |
+
+Internal services reach the broker as *local principals* rather than as ordinary
+clients: each presents a client certificate whose URI SAN names it, plus a SASL
+secret, and the broker grants it only what it needs — the Rules Engine consumes
+`m`, republishes under it, and feeds the `writers` and `alarms` streams; the
+writers only subscribe to `writers`; the publish proxy that serves the UI's
+HTTP publish endpoint only publishes under `m.`. The principals are declared in
+`docker/fluxmq/node{1,2,3}.yaml`, and adding a service means adding an entry
+there alongside its certificate and secret.
+
+Being a local principal is also what preserves a message's origin. The broker
+stamps its own transport protocol and identity on anything published over a
+connection it does not trust, so a message relayed to the writers over the plain
+AMQP listener would be stored as `protocol: amqp` with the relaying service as
+its publisher. A `service`-role principal on the mTLS listener may state the
+origin instead, and the protocol the device actually published with survives to
+the database.
+
+The certificates are issued by the development CA committed at
+`docker/ssl/certs/ca.crt`, so no extra setup is needed for a local run. The
+generated material is gitignored.
+
+The server certificate is issued for `fluxmq` and `fluxmq-node{1,2,3}`, which
+covers both this Compose stack and a single-node deployment. Point any
+`MG_*_BROKER_URL` at a host outside that set and the service fails its TLS
+verification with `certificate is valid for ...`; add the name to
+`FLUXMQ_SERVICE_SERVER_CERT_CONFIG` in `docker/ssl/Makefile` and reissue:
+
+```bash
+rm -f docker/ssl/certs/fluxmq-service-server.* \
+  docker/ssl/certs/re-fluxmq-client.* \
+  docker/ssl/certs/timescale-writer-fluxmq-client.* \
+  docker/ssl/certs/postgres-writer-fluxmq-client.*
+make -C docker/ssl fluxmq_service_certs
+```
+
+`make check_certs` skips certificates that already exist, so stale certificates
+have to be removed rather than merely re-running the target.
+
+Each local-principal secret must stay equal to the corresponding value in
+`docker/.env`; a mismatch fails that service's broker authentication. After
+changing one, re-run its target:
+
+| Variable                            | Target                                    |
+| ----------------------------------- | ----------------------------------------- |
+| `MG_RE_BROKER_SECRET`               | `fluxmq_service_secret`                   |
+| `MG_TIMESCALE_WRITER_BROKER_SECRET` | `timescale_writer_fluxmq_service_secret`  |
+| `MG_POSTGRES_WRITER_BROKER_SECRET`  | `postgres_writer_fluxmq_service_secret`   |
+| `MG_FLUXMQ_BROKER_SECRET`           | `fluxmq_auth_fluxmq_service_secret`       |
+
+`trace.key` is created once and preserved on later runs — replacing it while
+messages are in flight would invalidate the rule traces they already carry, so
+delete it only deliberately. Every Rules Engine replica must read the same key.
+
+Start the stack through `make run_latest` rather than calling `docker compose
+up` directly. Compose creates a missing bind-mount source as an empty
+*directory*, so bringing up `re` or `fluxmq` before these files exist leaves the
+containers failing against a directory where they expect a key.
+
+### Atom service tokens
+
+Not generated automatically, because provisioning them starts Atom and runs a
+bootstrap job against it:
+
+```bash
+make provision_atom_tokens
+```
+
+This brings up Atom, runs `atom-bootstrap`, and writes the gitignored
+`docker/.env.tokens` with one service token per consumer —
+`MG_ATOM_TOKEN_FLUXMQ_AUTH`, `MG_ATOM_TOKEN_FLUXMQ_NODE{1,2,3}`,
+`MG_ATOM_TOKEN_RE`, `MG_ATOM_TOKEN_ALARMS`, `MG_ATOM_TOKEN_REPORTS`,
+`MG_ATOM_TOKEN_TIMESCALE_READER`, and `MG_ATOM_TOKEN_POSTGRES_READER`.
+
+`make run_latest` refuses to start when that file is absent or short of any of
+those variables, and names what is missing. To fold the step into the run:
+
+```bash
+make run_latest PROVISION_ATOM_TOKENS=true
+```
+
+Re-run `provision_atom_tokens` after anything that resets Atom's database; the
+old tokens do not survive it.
+
 ---
 
 ## Usage

cmd/fluxmq/main.go

@@ -51,6 +51,16 @@ type config struct {
 	JaegerURL  url.URL `env:"MG_JAEGER_URL"           envDefault:"http://localhost:4318/v1/traces"`
 	TraceRatio float64 `env:"MG_JAEGER_TRACE_RATIO"   envDefault:"1.0"`
 	InstanceID string  `env:"MG_FLUXMQ_INSTANCE_ID"   envDefault:""`
+	// The publish proxy states the origin of the message it relays: the
+	// protocol the user published with and the client it was published as.
+	// A broker stamps its own transport and identity on a publication from a
+	// connection it does not trust, which would store every message from this
+	// service as having arrived over AMQP. These point the publisher at the
+	// listener that accepts a relayed origin instead.
+	MsgBrokerURL     string `env:"MG_FLUXMQ_BROKER_URL"         envDefault:""`
+	BrokerClientCert string `env:"MG_FLUXMQ_BROKER_CLIENT_CERT" envDefault:""`
+	BrokerClientKey  string `env:"MG_FLUXMQ_BROKER_CLIENT_KEY"  envDefault:""`
+	BrokerCACerts    string `env:"MG_FLUXMQ_BROKER_CA_CERTS"    envDefault:""`
 }
 
 func main() {
@@ -154,11 +164,21 @@ func main() {
 		MaxHeaderBytes:    grpcServerConfig.MaxHeaderBytes,
 	}
 
-	messagePublisher, err := fluxmqbroker.NewUndeclaredPublisher(
-		ctx,
-		cfg.BrokerURL,
+	// The mTLS client identity is optional: brokers that expose an
+	// unauthenticated listener need none. A partial configuration still fails,
+	// since half an identity is not a usable one.
+	publisherOpts := []messaging.Option{
 		fluxmqbroker.ConnectionName("fluxmq-ui-message-publish-proxy"),
-	)
+	}
+	msgBrokerURL := cfg.BrokerURL
+	if cfg.MsgBrokerURL != "" {
+		msgBrokerURL = cfg.MsgBrokerURL
+	}
+	if cfg.BrokerClientCert != "" || cfg.BrokerClientKey != "" || cfg.BrokerCACerts != "" {
+		publisherOpts = append(publisherOpts, fluxmqbroker.InternalMetadata(cfg.BrokerClientCert, cfg.BrokerClientKey, cfg.BrokerCACerts))
+	}
+
+	messagePublisher, err := fluxmqbroker.NewUndeclaredPublisher(ctx, msgBrokerURL, publisherOpts...)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to create publish proxy message publisher: %s", err))
 		exitCode = 1

cmd/postgres-writer/main.go

@@ -20,8 +20,9 @@ import (
 	writerpg "github.com/absmach/magistrala/consumers/writers/postgres"
 	mglog "github.com/absmach/magistrala/logger"
 	jaegerclient "github.com/absmach/magistrala/pkg/jaeger"
+	"github.com/absmach/magistrala/pkg/messaging"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
-	brokers "github.com/absmach/magistrala/pkg/messaging/writers"
+	"github.com/absmach/magistrala/pkg/messaging/writers"
 	pgclient "github.com/absmach/magistrala/pkg/postgres"
 	"github.com/absmach/magistrala/pkg/prometheus"
 	"github.com/absmach/magistrala/pkg/server"
@@ -41,13 +42,16 @@ const (
 )
 
 type config struct {
-	LogLevel      string  `env:"MG_POSTGRES_WRITER_LOG_LEVEL"     envDefault:"info"`
-	ConfigPath    string  `env:"MG_POSTGRES_WRITER_CONFIG_PATH"   envDefault:"/config.toml"`
-	BrokerURL     string  `env:"MG_MESSAGE_BROKER_URL"            envDefault:"nats://localhost:4222"`
-	JaegerURL     url.URL `env:"MG_JAEGER_URL"                    envDefault:"http://localhost:4318/v1/traces"`
-	SendTelemetry bool    `env:"MG_SEND_TELEMETRY"                envDefault:"true"`
-	InstanceID    string  `env:"MG_POSTGRES_WRITER_INSTANCE_ID"   envDefault:""`
-	TraceRatio    float64 `env:"MG_JAEGER_TRACE_RATIO"            envDefault:"1.0"`
+	LogLevel         string  `env:"MG_POSTGRES_WRITER_LOG_LEVEL"          envDefault:"info"`
+	ConfigPath       string  `env:"MG_POSTGRES_WRITER_CONFIG_PATH"        envDefault:"/config.toml"`
+	BrokerURL        string  `env:"MG_MESSAGE_BROKER_URL"                 envDefault:"nats://localhost:4222"`
+	BrokerClientCert string  `env:"MG_POSTGRES_WRITER_BROKER_CLIENT_CERT" envDefault:""`
+	BrokerClientKey  string  `env:"MG_POSTGRES_WRITER_BROKER_CLIENT_KEY"  envDefault:""`
+	BrokerCACerts    string  `env:"MG_POSTGRES_WRITER_BROKER_CA_CERTS"    envDefault:""`
+	JaegerURL        url.URL `env:"MG_JAEGER_URL"                         envDefault:"http://localhost:4318/v1/traces"`
+	SendTelemetry    bool    `env:"MG_SEND_TELEMETRY"                     envDefault:"true"`
+	InstanceID       string  `env:"MG_POSTGRES_WRITER_INSTANCE_ID"        envDefault:""`
+	TraceRatio       float64 `env:"MG_JAEGER_TRACE_RATIO"                 envDefault:"1.0"`
 }
 
 func main() {
@@ -107,7 +111,16 @@ func main() {
 	}()
 	tracer := tp.Tracer(svcName)
 
-	pubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
+	// The mTLS client identity is optional: brokers that expose an unauthenticated
+	// listener need none. Passing the option with empty paths would fail the
+	// connection outright, so only ask for it once one of them is configured. A
+	// partial configuration still fails, since half an identity is not a usable one.
+	var brokerOpts []messaging.Option
+	if cfg.BrokerClientCert != "" || cfg.BrokerClientKey != "" || cfg.BrokerCACerts != "" {
+		brokerOpts = append(brokerOpts, writers.InternalMetadata(cfg.BrokerClientCert, cfg.BrokerClientKey, cfg.BrokerCACerts))
+	}
+
+	pubSub, err := writers.NewPubSub(ctx, cfg.BrokerURL, logger, brokerOpts...)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
 		exitCode = 1
@@ -119,7 +132,7 @@ func main() {
 	repo := newService(db, logger)
 	repo = consumertracing.NewBlocking(tracer, repo, httpServerConfig)
 
-	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, brokers.AllTopic, logger); err != nil {
+	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, writers.AllTopic, logger); err != nil {
 		logger.Error(fmt.Sprintf("failed to create Postgres writer: %s", err))
 		exitCode = 1
 		return

cmd/timescale-writer/main.go

@@ -20,8 +20,9 @@ import (
 	"github.com/absmach/magistrala/consumers/writers/timescale"
 	mglog "github.com/absmach/magistrala/logger"
 	jaegerclient "github.com/absmach/magistrala/pkg/jaeger"
+	"github.com/absmach/magistrala/pkg/messaging"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
-	brokers "github.com/absmach/magistrala/pkg/messaging/writers"
+	"github.com/absmach/magistrala/pkg/messaging/writers"
 	pgclient "github.com/absmach/magistrala/pkg/postgres"
 	"github.com/absmach/magistrala/pkg/prometheus"
 	"github.com/absmach/magistrala/pkg/server"
@@ -41,13 +42,16 @@ const (
 )
 
 type config struct {
-	LogLevel      string  `env:"MG_TIMESCALE_WRITER_LOG_LEVEL"     envDefault:"info"`
-	ConfigPath    string  `env:"MG_TIMESCALE_WRITER_CONFIG_PATH"   envDefault:"/config.toml"`
-	BrokerURL     string  `env:"MG_MESSAGE_BROKER_URL"            envDefault:"nats://localhost:4222"`
-	JaegerURL     url.URL `env:"MG_JAEGER_URL"                    envDefault:"http://localhost:4318/v1/traces"`
-	SendTelemetry bool    `env:"MG_SEND_TELEMETRY"                envDefault:"true"`
-	InstanceID    string  `env:"MG_TIMESCALE_WRITER_INSTANCE_ID"   envDefault:""`
-	TraceRatio    float64 `env:"MG_JAEGER_TRACE_RATIO"            envDefault:"1.0"`
+	LogLevel         string  `env:"MG_TIMESCALE_WRITER_LOG_LEVEL"          envDefault:"info"`
+	ConfigPath       string  `env:"MG_TIMESCALE_WRITER_CONFIG_PATH"        envDefault:"/config.toml"`
+	BrokerURL        string  `env:"MG_MESSAGE_BROKER_URL"                  envDefault:"nats://localhost:4222"`
+	BrokerClientCert string  `env:"MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT" envDefault:""`
+	BrokerClientKey  string  `env:"MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY"  envDefault:""`
+	BrokerCACerts    string  `env:"MG_TIMESCALE_WRITER_BROKER_CA_CERTS"    envDefault:""`
+	JaegerURL        url.URL `env:"MG_JAEGER_URL"                          envDefault:"http://localhost:4318/v1/traces"`
+	SendTelemetry    bool    `env:"MG_SEND_TELEMETRY"                      envDefault:"true"`
+	InstanceID       string  `env:"MG_TIMESCALE_WRITER_INSTANCE_ID"        envDefault:""`
+	TraceRatio       float64 `env:"MG_JAEGER_TRACE_RATIO"                  envDefault:"1.0"`
 }
 
 func main() {
@@ -112,7 +116,16 @@ func main() {
 	repo := newService(db, logger)
 	repo = consumertracing.NewBlocking(tracer, repo, httpServerConfig)
 
-	pubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
+	// The mTLS client identity is optional: brokers that expose an unauthenticated
+	// listener need none. Passing the option with empty paths would fail the
+	// connection outright, so only ask for it once one of them is configured. A
+	// partial configuration still fails, since half an identity is not a usable one.
+	var brokerOpts []messaging.Option
+	if cfg.BrokerClientCert != "" || cfg.BrokerClientKey != "" || cfg.BrokerCACerts != "" {
+		brokerOpts = append(brokerOpts, writers.InternalMetadata(cfg.BrokerClientCert, cfg.BrokerClientKey, cfg.BrokerCACerts))
+	}
+
+	pubSub, err := writers.NewPubSub(ctx, cfg.BrokerURL, logger, brokerOpts...)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
 		exitCode = 1
@@ -121,7 +134,7 @@ func main() {
 	defer pubSub.Close()
 	pubSub = brokerstracing.NewPubSub(httpServerConfig, tracer, pubSub)
 
-	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, brokers.AllTopic, logger); err != nil {
+	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, writers.AllTopic, logger); err != nil {
 		logger.Error(fmt.Sprintf("failed to create Timescale writer: %s", err))
 		exitCode = 1
 		return

docker/.env

@@ -388,6 +388,16 @@ MG_FLUXMQ_INSTANCE_ID=
 MG_FLUXMQ_CACHE_NUM_COUNTERS=200000
 MG_FLUXMQ_CACHE_MAX_COST=1048576
 MG_FLUXMQ_CACHE_BUFFER_ITEMS=64
+# The publish proxy relays a user's message, so it publishes on the mTLS service
+# listener rather than the plain one: only there may it state the origin
+# protocol and publisher instead of having the broker stamp its own. The
+# certificate identifies the local principal and its SASL secret must match
+# docker/fluxmq/secrets/fluxmq-auth-current.
+MG_FLUXMQ_BROKER_SECRET=fluxmq-auth-fluxmq-local-principal-secret-change-me
+MG_FLUXMQ_BROKER_URL=amqps://fluxmq-auth:${MG_FLUXMQ_BROKER_SECRET}@fluxmq-node1:5685/
+MG_FLUXMQ_BROKER_CLIENT_CERT=/etc/ssl/certs/fluxmq-auth-fluxmq-client.crt
+MG_FLUXMQ_BROKER_CLIENT_KEY=/etc/ssl/private/fluxmq-auth-fluxmq-client.key
+MG_FLUXMQ_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
 
 ### CoAP
 MG_COAP_PORT=5683
@@ -502,6 +512,14 @@ MG_POSTGRES_WRITER_HTTP_PORT=9007
 MG_POSTGRES_WRITER_HTTP_SERVER_CERT=
 MG_POSTGRES_WRITER_HTTP_SERVER_KEY=
 MG_POSTGRES_WRITER_INSTANCE_ID=
+# The writer is a passive consumer on FluxMQ's mTLS local listener. Its client
+# certificate identifies the local principal and its SASL secret must match
+# docker/fluxmq/secrets/postgres-writer-current.
+MG_POSTGRES_WRITER_BROKER_SECRET=postgres-writer-fluxmq-local-principal-secret-change-me
+MG_POSTGRES_WRITER_BROKER_URL=amqps://postgres-writer:${MG_POSTGRES_WRITER_BROKER_SECRET}@fluxmq-node1:5685/
+MG_POSTGRES_WRITER_BROKER_CLIENT_CERT=/etc/ssl/certs/postgres-writer-fluxmq-client.crt
+MG_POSTGRES_WRITER_BROKER_CLIENT_KEY=/etc/ssl/private/postgres-writer-fluxmq-client.key
+MG_POSTGRES_WRITER_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
 
 ### Postgres Reader
 MG_POSTGRES_READER_LOG_LEVEL=debug
@@ -529,6 +547,14 @@ MG_TIMESCALE_WRITER_HTTP_PORT=9012
 MG_TIMESCALE_WRITER_HTTP_SERVER_CERT=
 MG_TIMESCALE_WRITER_HTTP_SERVER_KEY=
 MG_TIMESCALE_WRITER_INSTANCE_ID=
+# The writer is a passive consumer on FluxMQ's mTLS local listener. Its client
+# certificate identifies the local principal and its SASL secret must match
+# docker/fluxmq/secrets/timescale-writer-current.
+MG_TIMESCALE_WRITER_BROKER_SECRET=timescale-writer-fluxmq-local-principal-secret-change-me
+MG_TIMESCALE_WRITER_BROKER_URL=amqps://timescale-writer:${MG_TIMESCALE_WRITER_BROKER_SECRET}@fluxmq-node1:5685/
+MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT=/etc/ssl/certs/timescale-writer-fluxmq-client.crt
+MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY=/etc/ssl/private/timescale-writer-fluxmq-client.key
+MG_TIMESCALE_WRITER_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
 
 ### Timescale Reader
 MG_TIMESCALE_READER_LOG_LEVEL=debug
@@ -569,6 +595,29 @@ MG_RE_DB_SSL_CERT=
 MG_RE_DB_SSL_KEY=
 MG_RE_DB_SSL_ROOT_CERT=
 MG_RE_INSTANCE_ID=
+MG_RE_MAX_RULE_HOPS=32
+MG_RE_TRACE_KEY=
+MG_RE_TRACE_KEY_FILE=/run/secrets/magistrala-re-trace-key
+# Rule loop detection needs the broker to carry the "_flux." message metadata
+# that holds the rule execution trace. On FluxMQ only the mTLS local listener
+# does, so the Rules Engine must reach the broker there. A plain FluxMQ
+# connection cannot provide loop protection and is rejected at startup.
+#
+# The certificates, broker secret, and persistent trace key come from the
+# corresponding targets in docker/ssl, all of which run as part of check_certs.
+# MG_RE_BROKER_SECRET is the SASL secret the Rules Engine presents alongside
+# its certificate; it is the single source for
+# docker/fluxmq/secrets/re-current and must be at least 32 characters.
+# MG_RE_TRACE_KEY_FILE names the mounted trace key. MG_RE_TRACE_KEY remains an
+# inline alternative; set exactly one, and use at least 32 bytes.
+#
+# The URL names one node directly because mTLS cannot pass through the nginx
+# AMQP proxy as configured.
+MG_RE_BROKER_SECRET=re-fluxmq-local-principal-secret-change-me
+MG_RE_BROKER_URL=amqps://rules-engine:${MG_RE_BROKER_SECRET}@fluxmq-node1:5685/
+MG_RE_BROKER_CLIENT_CERT=/etc/ssl/certs/re-fluxmq-client.crt
+MG_RE_BROKER_CLIENT_KEY=/etc/ssl/private/re-fluxmq-client.key
+MG_RE_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
 MG_RE_EMAIL_TEMPLATE=re.tmpl
 MG_RE_CALLOUT_URLS=""
 MG_RE_CALLOUT_METHOD="POST"

docker/addons/postgres-writer/docker-compose.yaml

@@ -35,7 +35,10 @@ services:
     image: ghcr.io/absmach/magistrala/postgres-writer:${MG_RELEASE_TAG}
     container_name: magistrala-postgres-writer
     depends_on:
-      - postgres
+      postgres:
+        condition: service_started
+      fluxmq-node1:
+        condition: service_healthy
     restart: on-failure
     environment:
       MG_POSTGRES_WRITER_LOG_LEVEL: ${MG_POSTGRES_WRITER_LOG_LEVEL}
@@ -53,7 +56,10 @@ services:
       MG_POSTGRES_SSL_CERT: ${MG_POSTGRES_SSL_CERT}
       MG_POSTGRES_SSL_KEY: ${MG_POSTGRES_SSL_KEY}
       MG_POSTGRES_SSL_ROOT_CERT: ${MG_POSTGRES_SSL_ROOT_CERT}
-      MG_MESSAGE_BROKER_URL: ${MG_MESSAGE_BROKER_URL}
+      MG_MESSAGE_BROKER_URL: ${MG_POSTGRES_WRITER_BROKER_URL}
+      MG_POSTGRES_WRITER_BROKER_CLIENT_CERT: ${MG_POSTGRES_WRITER_BROKER_CLIENT_CERT}
+      MG_POSTGRES_WRITER_BROKER_CLIENT_KEY: ${MG_POSTGRES_WRITER_BROKER_CLIENT_KEY}
+      MG_POSTGRES_WRITER_BROKER_CA_CERTS: ${MG_POSTGRES_WRITER_BROKER_CA_CERTS}
       MG_JAEGER_URL: ${MG_JAEGER_URL}
       MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
       MG_SEND_TELEMETRY: ${MG_SEND_TELEMETRY}
@@ -63,4 +69,11 @@ services:
     networks:
       - magistrala-base-net
     volumes:
-      - ./config.toml:/config.toml
+      # Relative paths in an addon file resolve against the project directory,
+      # which is the base compose file's directory rather than this one.
+      - ./addons/postgres-writer/config.toml:${MG_POSTGRES_WRITER_CONFIG_PATH}
+      # The URI SAN in this client certificate identifies the Postgres writer
+      # to the FluxMQ local listener; the broker ACL only permits `writers`.
+      - ./ssl/certs/postgres-writer-fluxmq-client.crt:${MG_POSTGRES_WRITER_BROKER_CLIENT_CERT}:ro
+      - ./ssl/certs/postgres-writer-fluxmq-client.key:${MG_POSTGRES_WRITER_BROKER_CLIENT_KEY}:ro
+      - ./ssl/certs/ca.crt:${MG_POSTGRES_WRITER_BROKER_CA_CERTS}:ro

docker/docker-compose.yaml

@@ -255,6 +255,15 @@ services:
         ipv4_address: 172.30.0.201
     volumes:
       - ./fluxmq/node1.yaml:/etc/fluxmq/config.yaml:ro
+      # Server certificate and CA for the mTLS service listener, plus each
+      # local principal's SASL secret.
+      - ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
+      - ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
+      - ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
+      - ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
+      - ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
+      - ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
+      - ./fluxmq/secrets/fluxmq-auth-current:/etc/fluxmq/secrets/fluxmq-auth-current:ro
       - magistrala-fluxmq-node1-volume:/tmp/fluxmq
 
   fluxmq-node2:
@@ -275,6 +284,15 @@ services:
         ipv4_address: 172.30.0.202
     volumes:
       - ./fluxmq/node2.yaml:/etc/fluxmq/config.yaml:ro
+      # Server certificate and CA for the mTLS service listener, plus each
+      # local principal's SASL secret.
+      - ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
+      - ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
+      - ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
+      - ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
+      - ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
+      - ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
+      - ./fluxmq/secrets/fluxmq-auth-current:/etc/fluxmq/secrets/fluxmq-auth-current:ro
       - magistrala-fluxmq-node2-volume:/tmp/fluxmq
 
   fluxmq-node3:
@@ -295,6 +313,15 @@ services:
         ipv4_address: 172.30.0.203
     volumes:
       - ./fluxmq/node3.yaml:/etc/fluxmq/config.yaml:ro
+      # Server certificate and CA for the mTLS service listener, plus each
+      # local principal's SASL secret.
+      - ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
+      - ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
+      - ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
+      - ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
+      - ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
+      - ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
+      - ./fluxmq/secrets/fluxmq-auth-current:/etc/fluxmq/secrets/fluxmq-auth-current:ro
       - magistrala-fluxmq-node3-volume:/tmp/fluxmq
 
   fluxmq-auth:
@@ -328,6 +355,17 @@ services:
       ATOM_TIMEOUT: ${ATOM_TIMEOUT}
       MG_JAEGER_URL: ${MG_JAEGER_URL}
       MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
+      MG_FLUXMQ_BROKER_URL: ${MG_FLUXMQ_BROKER_URL}
+      MG_FLUXMQ_BROKER_CLIENT_CERT: ${MG_FLUXMQ_BROKER_CLIENT_CERT}
+      MG_FLUXMQ_BROKER_CLIENT_KEY: ${MG_FLUXMQ_BROKER_CLIENT_KEY}
+      MG_FLUXMQ_BROKER_CA_CERTS: ${MG_FLUXMQ_BROKER_CA_CERTS}
+    volumes:
+      # The URI SAN in this client certificate identifies the publish proxy to
+      # the FluxMQ local listener; the broker ACL only permits publishing to the
+      # message topic namespace.
+      - ./ssl/certs/fluxmq-auth-fluxmq-client.crt:${MG_FLUXMQ_BROKER_CLIENT_CERT}:ro
+      - ./ssl/certs/fluxmq-auth-fluxmq-client.key:${MG_FLUXMQ_BROKER_CLIENT_KEY}:ro
+      - ./ssl/certs/ca.crt:${MG_FLUXMQ_BROKER_CA_CERTS}:ro
     networks:
       - magistrala-base-net
 
@@ -539,8 +577,12 @@ services:
     networks:
       - magistrala-base-net
 
+  # Creates the image bucket with the broker's own image rather than pulling a
+  # second one. An aws-cli here cost a 130 MB image for a single call, and it
+  # was the only untagged reference in this file, so what it resolved to
+  # depended on the day it was pulled.
   seaweedfs-init:
-    image: amazon/aws-cli
+    image: chrislusf/seaweedfs:4.40
     container_name: magistrala-seaweedfs-init
     entrypoint: /bin/sh
     depends_on:
@@ -548,26 +590,46 @@ services:
     command:
       - -c
       - |
-        echo "[INIT] Waiting 20s for SeaweedFS S3 to be ready...";
-        sleep 20;
-        OUT=$(aws --endpoint-url http://seaweedfs-s3:8333 s3api create-bucket --bucket $${BUCKET} 2>&1);
-        EXIT=$$?;
-        if [ $$EXIT -eq 0 ]; then
-          echo "[INIT] Bucket $${BUCKET} created successfully.";
-        elif echo "$$OUT" | grep -q 'BucketAlreadyOwnedByYou\|BucketAlreadyExists'; then
-          echo "[INIT] Bucket $${BUCKET} already exists, skipping.";
-        else
-          echo "[INIT] Failed to create bucket $${BUCKET}: $$OUT" >&2;
-          exit 1;
-        fi
+        set -eu
+        # weed shell blocks indefinitely against an unreachable master, so the
+        # wait is a port check and every weed call is bounded; left to itself it
+        # would hang this container rather than fail it.
+        #
+        # 60 ticks is ~2 minutes while the store is merely slow to boot, and
+        # ~3 minutes to give up when it never appears, since a failing lookup
+        # costs about a second on top of the sleep.
+        tick=0;
+        while ! nc -z seaweedfs-s3 9333 2>/dev/null; do
+          tick=$$((tick + 1));
+          if [ $$tick -ge 60 ]; then
+            echo "[INIT] SeaweedFS master did not come up" >&2;
+            exit 1;
+          fi;
+          sleep 2;
+        done;
+        printf 's3.bucket.create -name %s\n' "$${BUCKET}" > /tmp/create.cmd;
+        printf 's3.bucket.list\n' > /tmp/list.cmd;
+        try=0;
+        while [ $$try -lt 3 ]; do
+          try=$$((try + 1));
+          timeout 20 weed shell -master seaweedfs-s3:9333 < /tmp/create.cmd > /tmp/create.out 2>&1 || true;
+          timeout 20 weed shell -master seaweedfs-s3:9333 < /tmp/list.cmd > /tmp/list.out 2>/dev/null || true;
+          # s3.bucket.create reports success even for an existing bucket, and
+          # weed shell exits 0 whatever the command did, so the listing is the
+          # only trustworthy signal that the bucket is really there.
+          if awk -v b="$${BUCKET}" '{sub(/^[ \t]+/,""); split($$0,f,/[ \t]/); if (f[1]==b) found=1} END{exit !found}' /tmp/list.out; then
+            echo "[INIT] Bucket $${BUCKET} present.";
+            exit 0;
+          fi;
+          sleep 3;
+        done;
+        echo "[INIT] Bucket $${BUCKET} was not created after $$try attempts" >&2;
+        cat /tmp/create.out >&2;
+        exit 1
     networks:
       - magistrala-base-net
     environment:
       BUCKET: ${MG_BACKEND_OBJECT_STORAGE_BUCKET}
-      AWS_ACCESS_KEY_ID: ${MG_BACKEND_OBJECT_STORAGE_ACCESS_KEY}
-      AWS_SECRET_ACCESS_KEY: ${MG_BACKEND_OBJECT_STORAGE_SECRET_KEY}
-      AWS_DEFAULT_REGION: ${MG_BACKEND_OBJECT_STORAGE_REGION}
-      AWS_EC2_METADATA_DISABLED: "true"
 
   timescale:
     image: timescale/timescaledb:2.29.0-pg16-oss
@@ -583,13 +645,18 @@ services:
       - magistrala-base-net
     volumes:
       - magistrala-timescale-writer-volume:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
+      interval: 5s
+      timeout: 3s
+      retries: 60
 
   timescale-reader:
     image: ghcr.io/absmach/magistrala/timescale-reader:${MG_RELEASE_TAG}
     container_name: magistrala-timescale-reader
     depends_on:
       timescale:
-        condition: service_started
+        condition: service_healthy
       atom-bootstrap:
         condition: service_completed_successfully
     restart: on-failure
@@ -668,7 +735,10 @@ services:
     image: ghcr.io/absmach/magistrala/timescale-writer:${MG_RELEASE_TAG}
     container_name: magistrala-timescale-writer
     depends_on:
-      - timescale
+      timescale:
+        condition: service_healthy
+      fluxmq-node1:
+        condition: service_healthy
     restart: on-failure
     environment:
       MG_TIMESCALE_WRITER_LOG_LEVEL: ${MG_TIMESCALE_WRITER_LOG_LEVEL}
@@ -686,7 +756,10 @@ services:
       MG_TIMESCALE_SSL_CERT: ${MG_TIMESCALE_SSL_CERT}
       MG_TIMESCALE_SSL_KEY: ${MG_TIMESCALE_SSL_KEY}
       MG_TIMESCALE_SSL_ROOT_CERT: ${MG_TIMESCALE_SSL_ROOT_CERT}
-      MG_MESSAGE_BROKER_URL: ${MG_MESSAGE_BROKER_URL}
+      MG_MESSAGE_BROKER_URL: ${MG_TIMESCALE_WRITER_BROKER_URL}
+      MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT: ${MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT}
+      MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY: ${MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY}
+      MG_TIMESCALE_WRITER_BROKER_CA_CERTS: ${MG_TIMESCALE_WRITER_BROKER_CA_CERTS}
       MG_JAEGER_URL: ${MG_JAEGER_URL}
       MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
       MG_SEND_TELEMETRY: ${MG_SEND_TELEMETRY}
@@ -697,6 +770,11 @@ services:
       - magistrala-base-net
     volumes:
       - ./addons/timescale-writer/config.toml:${MG_TIMESCALE_WRITER_CONFIG_PATH}
+      # The URI SAN in this client certificate identifies the Timescale writer
+      # to the FluxMQ local listener; the broker ACL only permits `writers`.
+      - ./ssl/certs/timescale-writer-fluxmq-client.crt:${MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT}:ro
+      - ./ssl/certs/timescale-writer-fluxmq-client.key:${MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY}:ro
+      - ./ssl/certs/ca.crt:${MG_TIMESCALE_WRITER_BROKER_CA_CERTS}:ro
   re-db:
     image: docker.io/postgres:18.4-alpine3.24
     container_name: magistrala-re-db
@@ -761,6 +839,13 @@ services:
       MG_SEND_TELEMETRY: ${MG_SEND_TELEMETRY}
       MG_PERMISSIONS_FILE: ${MG_PERMISSIONS_FILE}
       MG_RE_INSTANCE_ID: ${MG_RE_INSTANCE_ID}
+      MG_RE_MAX_RULE_HOPS: ${MG_RE_MAX_RULE_HOPS}
+      MG_RE_TRACE_KEY: ${MG_RE_TRACE_KEY}
+      MG_RE_TRACE_KEY_FILE: ${MG_RE_TRACE_KEY_FILE}
+      MG_RE_BROKER_URL: ${MG_RE_BROKER_URL}
+      MG_RE_BROKER_CLIENT_CERT: ${MG_RE_BROKER_CLIENT_CERT}
+      MG_RE_BROKER_CLIENT_KEY: ${MG_RE_BROKER_CLIENT_KEY}
+      MG_RE_BROKER_CA_CERTS: ${MG_RE_BROKER_CA_CERTS}
       MG_EMAIL_HOST: ${MG_EMAIL_HOST}
       MG_EMAIL_PORT: ${MG_EMAIL_PORT}
       MG_EMAIL_USERNAME: ${MG_EMAIL_USERNAME}
@@ -781,6 +866,12 @@ services:
     volumes:
       - ./permission.yaml:${MG_PERMISSIONS_FILE}
       - ./templates/${MG_RE_EMAIL_TEMPLATE}:/email.tmpl
+      # Client certificate for the FluxMQ service listener. Its URI SAN is what
+      # the broker matches against the rules-engine local principal.
+      - ./ssl/certs/re-fluxmq-client.crt:${MG_RE_BROKER_CLIENT_CERT}:ro
+      - ./ssl/certs/re-fluxmq-client.key:${MG_RE_BROKER_CLIENT_KEY}:ro
+      - ./ssl/certs/ca.crt:${MG_RE_BROKER_CA_CERTS}:ro
+      - ./re/secrets/trace.key:/run/secrets/magistrala-re-trace-key:ro
 
   alarms-db:
     image: docker.io/postgres:18.4-alpine3.24

docker/fluxmq/.gitignore

@@ -0,0 +1,5 @@
+# Copyright (c) Abstract Machines
+# SPDX-License-Identifier: Apache-2.0
+
+# Derived from MG_RE_BROKER_SECRET by "make fluxmq_service_secret" in docker/ssl.
+secrets/

docker/fluxmq/config_test.go

@@ -0,0 +1,69 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package fluxmq_test
+
+import (
+	"os"
+	"slices"
+	"testing"
+
+	"github.com/absmach/fluxmq/topics"
+	"gopkg.in/yaml.v3"
+)
+
+type brokerConfig struct {
+	Queues []struct {
+		Name   string   `yaml:"name"`
+		Topics []string `yaml:"topics"`
+	} `yaml:"queues"`
+}
+
+func TestQueueBindingsDoNotOverlap(t *testing.T) {
+	testCases := []struct {
+		topic string
+		want  []string
+	}{
+		{topic: "$queue/mqtt/client", want: []string{"mqtt"}},
+		{topic: "$queue/events/domain/client", want: []string{"events"}},
+		{topic: "$queue/writers/domain/channel", want: []string{"writers"}},
+		{topic: "$queue/alarms/domain/channel", want: []string{"alarms"}},
+		// A queue addressed with no trailing path still has to land in its own
+		// stream: pkg/events/fluxmq addresses exactly "$queue/events" when the
+		// stream name resolves to an empty path, and a binding that stopped
+		// matching its own parent level would drop those publications with no
+		// error, since an unmatched topic capture is not a failure.
+		{topic: "$queue/events", want: []string{"events"}},
+		// Channel messages reach stream "m" through its own "m/#" binding, so
+		// nothing addresses the queue directly. Were that to change, the
+		// publication would match no queue at all rather than fall through to
+		// the reserved one.
+		{topic: "$queue/m/domain/channel", want: nil},
+	}
+
+	for _, configFile := range []string{"node1.yaml", "node2.yaml", "node3.yaml"} {
+		data, err := os.ReadFile(configFile)
+		if err != nil {
+			t.Fatalf("read %s: %v", configFile, err)
+		}
+		var cfg brokerConfig
+		if err := yaml.Unmarshal(data, &cfg); err != nil {
+			t.Fatalf("parse %s: %v", configFile, err)
+		}
+
+		for _, tc := range testCases {
+			var got []string
+			for _, queue := range cfg.Queues {
+				for _, pattern := range queue.Topics {
+					if topics.TopicMatch(pattern, tc.topic) {
+						got = append(got, queue.Name)
+						break
+					}
+				}
+			}
+			if !slices.Equal(got, tc.want) {
+				t.Errorf("%s: queues matching %q = %v, want %v", configFile, tc.topic, got, tc.want)
+			}
+		}
+	}
+}

docker/fluxmq/node1.yaml

@@ -37,6 +37,20 @@ server:
   amqp091:
     plain:
       addr: "0.0.0.0:5682"
+    # The local listener is how a first-party service reaches the broker. It
+    # requires mTLS: FluxMQ matches the URI SAN of the presented certificate
+    # against a principal under auth.local_principals, and only a connection
+    # admitted here may exchange "_flux." message metadata. The Rules Engine
+    # needs that metadata to carry its rule execution trace, so it connects
+    # here rather than on the plain listener.
+    local:
+      addr: "0.0.0.0:5685"
+      max_connections: 32
+      cert_file: "/etc/fluxmq/certs/fluxmq-service-server.crt"
+      key_file: "/etc/fluxmq/certs/fluxmq-service-server.key"
+      ca_file: "/etc/fluxmq/certs/ca.crt"
+      client_auth: "require"
+      min_version: "TLS1.2"
   health_addr: "0.0.0.0:8081"
   health_enabled: true
   shutdown_timeout: 30s
@@ -94,8 +108,13 @@ queue_manager:
 
 queues:
   - name: "mqtt"
+    # "mqtt" is this reserved queue's name, not an MQTT-only transport filter;
+    # every protocol adapter uses the same topic-based queue routing. A local-
+    # principal prefix grant is captured by every matching queue, so keep this
+    # binding inside the mqtt namespace rather than also persisting service
+    # streams such as writers and alarms here.
     topics:
-      - "$queue/#"
+      - "$queue/mqtt/#"
     reserved: true
   - name: "events"
     topics:
@@ -127,6 +146,70 @@ queues:
       max_length_bytes: 1073741824
 
 auth:
+  # Principals admitted on the service listener. The certificate_uri_san must
+  # match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
+  # each secret file must hold the value configured by its service.
+  local_principals:
+    - name: "rules-engine"
+      certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
+      # A service may consume and may relay the origin of messages it did not
+      # author, both of which the Rules Engine needs when it republishes a
+      # device message. The capability is the principal's, not the listener's.
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/re-current"
+      permissions:
+        # Rule outputs are republished to m.<domain>.c.<channel>.<subtopic>,
+        # whose tenant and channel identifiers cannot be enumerated here, so the
+        # grant is the topic namespace rather than a list of exact keys. Being a
+        # prefix it is also an ordinary topic publish, which the cluster
+        # forwards; an exact target would be durable on one node and could not
+        # run alongside cluster.enabled.
+        publish:
+          - routing_key_prefix: "m."
+          # The rule output streams. These grants have to live here: a
+          # publication from a connection the broker does not trust has the
+          # broker's own transport protocol and identity stamped on it, so a
+          # saved message would read as "amqp" rather than the protocol the
+          # device published with, and an alarm would name the engine's broker
+          # connection rather than the client. Same prefix reasoning as above --
+          # each queue's own "$queue/<name>/#" binding is what captures it.
+          - routing_key_prefix: "$queue/writers/"
+          - routing_key_prefix: "$queue/alarms/"
+        # The message stream the Rules Engine consumes.
+        subscribe:
+          - "m"
+    - name: "timescale-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "postgres-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "fluxmq-auth"
+      certificate_uri_san: "spiffe://absmach/magistrala/fluxmq-auth"
+      # The publish proxy relays messages a user published over HTTP. Stating
+      # their origin protocol and publisher is what the "service" role grants;
+      # on an untrusted connection the broker would stamp its own instead and
+      # every such message would be stored as having arrived over AMQP.
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/fluxmq-auth-current"
+      permissions:
+        # Messages are published to m.<domain>.c.<channel>.<subtopic>, whose
+        # tenant and channel identifiers cannot be enumerated here. It runs no
+        # consumer, so it is granted no subscribe.
+        publish:
+          - routing_key_prefix: "m."
   external:
     url: "http://fluxmq-auth:7016"
     transport: "grpc"

docker/fluxmq/node2.yaml

@@ -34,6 +34,20 @@ server:
   amqp091:
     plain:
       addr: "0.0.0.0:5682"
+    # The local listener is how a first-party service reaches the broker. It
+    # requires mTLS: FluxMQ matches the URI SAN of the presented certificate
+    # against a principal under auth.local_principals, and only a connection
+    # admitted here may exchange "_flux." message metadata. The Rules Engine
+    # needs that metadata to carry its rule execution trace, so it connects
+    # here rather than on the plain listener.
+    local:
+      addr: "0.0.0.0:5685"
+      max_connections: 32
+      cert_file: "/etc/fluxmq/certs/fluxmq-service-server.crt"
+      key_file: "/etc/fluxmq/certs/fluxmq-service-server.key"
+      ca_file: "/etc/fluxmq/certs/ca.crt"
+      client_auth: "require"
+      min_version: "TLS1.2"
   health_addr: "0.0.0.0:8081"
   health_enabled: true
   shutdown_timeout: 30s
@@ -91,8 +105,13 @@ queue_manager:
 
 queues:
   - name: "mqtt"
+    # "mqtt" is this reserved queue's name, not an MQTT-only transport filter;
+    # every protocol adapter uses the same topic-based queue routing. A local-
+    # principal prefix grant is captured by every matching queue, so keep this
+    # binding inside the mqtt namespace rather than also persisting service
+    # streams such as writers and alarms here.
     topics:
-      - "$queue/#"
+      - "$queue/mqtt/#"
     reserved: true
   - name: "events"
     topics:
@@ -124,6 +143,70 @@ queues:
       max_length_bytes: 1073741824
 
 auth:
+  # Principals admitted on the service listener. The certificate_uri_san must
+  # match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
+  # each secret file must hold the value configured by its service.
+  local_principals:
+    - name: "rules-engine"
+      certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
+      # A service may consume and may relay the origin of messages it did not
+      # author, both of which the Rules Engine needs when it republishes a
+      # device message. The capability is the principal's, not the listener's.
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/re-current"
+      permissions:
+        # Rule outputs are republished to m.<domain>.c.<channel>.<subtopic>,
+        # whose tenant and channel identifiers cannot be enumerated here, so the
+        # grant is the topic namespace rather than a list of exact keys. Being a
+        # prefix it is also an ordinary topic publish, which the cluster
+        # forwards; an exact target would be durable on one node and could not
+        # run alongside cluster.enabled.
+        publish:
+          - routing_key_prefix: "m."
+          # The rule output streams. These grants have to live here: a
+          # publication from a connection the broker does not trust has the
+          # broker's own transport protocol and identity stamped on it, so a
+          # saved message would read as "amqp" rather than the protocol the
+          # device published with, and an alarm would name the engine's broker
+          # connection rather than the client. Same prefix reasoning as above --
+          # each queue's own "$queue/<name>/#" binding is what captures it.
+          - routing_key_prefix: "$queue/writers/"
+          - routing_key_prefix: "$queue/alarms/"
+        # The message stream the Rules Engine consumes.
+        subscribe:
+          - "m"
+    - name: "timescale-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "postgres-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "fluxmq-auth"
+      certificate_uri_san: "spiffe://absmach/magistrala/fluxmq-auth"
+      # The publish proxy relays messages a user published over HTTP. Stating
+      # their origin protocol and publisher is what the "service" role grants;
+      # on an untrusted connection the broker would stamp its own instead and
+      # every such message would be stored as having arrived over AMQP.
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/fluxmq-auth-current"
+      permissions:
+        # Messages are published to m.<domain>.c.<channel>.<subtopic>, whose
+        # tenant and channel identifiers cannot be enumerated here. It runs no
+        # consumer, so it is granted no subscribe.
+        publish:
+          - routing_key_prefix: "m."
   external:
     url: "http://fluxmq-auth:7016"
     transport: "grpc"

docker/fluxmq/node3.yaml

@@ -34,6 +34,20 @@ server:
   amqp091:
     plain:
       addr: "0.0.0.0:5682"
+    # The local listener is how a first-party service reaches the broker. It
+    # requires mTLS: FluxMQ matches the URI SAN of the presented certificate
+    # against a principal under auth.local_principals, and only a connection
+    # admitted here may exchange "_flux." message metadata. The Rules Engine
+    # needs that metadata to carry its rule execution trace, so it connects
+    # here rather than on the plain listener.
+    local:
+      addr: "0.0.0.0:5685"
+      max_connections: 32
+      cert_file: "/etc/fluxmq/certs/fluxmq-service-server.crt"
+      key_file: "/etc/fluxmq/certs/fluxmq-service-server.key"
+      ca_file: "/etc/fluxmq/certs/ca.crt"
+      client_auth: "require"
+      min_version: "TLS1.2"
   health_addr: "0.0.0.0:8081"
   health_enabled: true
   shutdown_timeout: 30s
@@ -91,8 +105,13 @@ queue_manager:
 
 queues:
   - name: "mqtt"
+    # "mqtt" is this reserved queue's name, not an MQTT-only transport filter;
+    # every protocol adapter uses the same topic-based queue routing. A local-
+    # principal prefix grant is captured by every matching queue, so keep this
+    # binding inside the mqtt namespace rather than also persisting service
+    # streams such as writers and alarms here.
     topics:
-      - "$queue/#"
+      - "$queue/mqtt/#"
     reserved: true
   - name: "events"
     topics:
@@ -124,6 +143,70 @@ queues:
       max_length_bytes: 1073741824
 
 auth:
+  # Principals admitted on the service listener. The certificate_uri_san must
+  # match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
+  # each secret file must hold the value configured by its service.
+  local_principals:
+    - name: "rules-engine"
+      certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
+      # A service may consume and may relay the origin of messages it did not
+      # author, both of which the Rules Engine needs when it republishes a
+      # device message. The capability is the principal's, not the listener's.
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/re-current"
+      permissions:
+        # Rule outputs are republished to m.<domain>.c.<channel>.<subtopic>,
+        # whose tenant and channel identifiers cannot be enumerated here, so the
+        # grant is the topic namespace rather than a list of exact keys. Being a
+        # prefix it is also an ordinary topic publish, which the cluster
+        # forwards; an exact target would be durable on one node and could not
+        # run alongside cluster.enabled.
+        publish:
+          - routing_key_prefix: "m."
+          # The rule output streams. These grants have to live here: a
+          # publication from a connection the broker does not trust has the
+          # broker's own transport protocol and identity stamped on it, so a
+          # saved message would read as "amqp" rather than the protocol the
+          # device published with, and an alarm would name the engine's broker
+          # connection rather than the client. Same prefix reasoning as above --
+          # each queue's own "$queue/<name>/#" binding is what captures it.
+          - routing_key_prefix: "$queue/writers/"
+          - routing_key_prefix: "$queue/alarms/"
+        # The message stream the Rules Engine consumes.
+        subscribe:
+          - "m"
+    - name: "timescale-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "postgres-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "fluxmq-auth"
+      certificate_uri_san: "spiffe://absmach/magistrala/fluxmq-auth"
+      # The publish proxy relays messages a user published over HTTP. Stating
+      # their origin protocol and publisher is what the "service" role grants;
+      # on an untrusted connection the broker would stamp its own instead and
+      # every such message would be stored as having arrived over AMQP.
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/fluxmq-auth-current"
+      permissions:
+        # Messages are published to m.<domain>.c.<channel>.<subtopic>, whose
+        # tenant and channel identifiers cannot be enumerated here. It runs no
+        # consumer, so it is granted no subscribe.
+        publish:
+          - routing_key_prefix: "m."
   external:
     url: "http://fluxmq-auth:7016"
     transport: "grpc"

docker/re/.gitignore

@@ -0,0 +1,5 @@
+# Copyright (c) Abstract Machines
+# SPDX-License-Identifier: Apache-2.0
+
+# Persistent Rules Engine HMAC keys generated by docker/ssl/Makefile.
+secrets/

docker/ssl/.gitignore

@@ -1,10 +1,21 @@
 # Copyright (c) Abstract Machines
 # SPDX-License-Identifier: Apache-2.0
 
-*grpc-server*
-*grpc-client*
-*srl
-*conf
+# Everything under certs/ is generated by this directory's Makefile, except the
+# development CA and the default server certificate that ship with the
+# repository.
+#
+# This is a denylist by intent: listing generated certificates by name meant
+# each new one had to remember to add itself here, and forgetting silently left
+# a private key committable. Ignore first, then re-include what is deliberately
+# tracked.
+certs/*
+!certs/ca.crt
+!certs/ca.key
+!certs/magistrala-server.crt
+!certs/magistrala-server.key
+
+# Paths outside certs/.
 client.crt
 client.key
 certbot-www/

docker/ssl/Makefile

@@ -43,6 +43,23 @@ CHANNELS_GRPC_CLIENT_CRT_FILE_NAME=channels-grpc-client
 COAP_DTLS_SERVER_CONF_FILE_NAME=coap-server.conf
 COAP_DTLS_SERVER_CN=coap
 COAP_DTLS_SERVER_CRT_FILE_NAME=coap-server
+# FluxMQ service listener. The broker identifies a first-party service by the
+# URI SAN of the certificate it presents, matching it against a principal
+# declared in the broker's own configuration, so the SAN below must stay in
+# step with docker/fluxmq/node*.yaml.
+FLUXMQ_SERVICE_SERVER_CRT_FILE_NAME=fluxmq-service-server
+RE_FLUXMQ_CLIENT_CN=rules-engine
+RE_FLUXMQ_CLIENT_CRT_FILE_NAME=re-fluxmq-client
+RE_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/rules-engine
+TIMESCALE_WRITER_FLUXMQ_CLIENT_CN=timescale-writer
+TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME=timescale-writer-fluxmq-client
+TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/timescale-writer
+POSTGRES_WRITER_FLUXMQ_CLIENT_CN=postgres-writer
+POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME=postgres-writer-fluxmq-client
+POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/postgres-writer
+FLUXMQ_AUTH_FLUXMQ_CLIENT_CN=fluxmq-auth
+FLUXMQ_AUTH_FLUXMQ_CLIENT_CRT_FILE_NAME=fluxmq-auth-fluxmq-client
+FLUXMQ_AUTH_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/fluxmq-auth
 
 define GRPC_CERT_CONFIG
 [req]
@@ -65,6 +82,57 @@ subjectAltName = @alt_names
 DNS.1 = <<SERVICE_NAME>>
 endef
 
+define FLUXMQ_SERVICE_SERVER_CERT_CONFIG
+[req]
+req_extensions = v3_req
+distinguished_name = dn
+prompt = no
+
+[dn]
+CN = mg.svc
+C  = RS
+ST = RS
+L  = BELGRADE
+O  = SUPERMQ
+OU = SUPERMQ
+
+[v3_req]
+extendedKeyUsage = serverAuth
+subjectAltName = @alt_names
+
+# The Rules Engine verifies this certificate against the host in
+# MG_RE_BROKER_URL, so every name a deployment may dial the service listener by
+# has to appear here: the per-node names this Compose stack uses, and the plain
+# "fluxmq" of a single-node deployment.
+[alt_names]
+DNS.1 = fluxmq
+DNS.2 = fluxmq-node1
+DNS.3 = fluxmq-node2
+DNS.4 = fluxmq-node3
+endef
+
+define URI_SAN_CERT_CONFIG
+[req]
+req_extensions = v3_req
+distinguished_name = dn
+prompt = no
+
+[dn]
+CN = <<SERVICE_NAME>>
+C  = RS
+ST = RS
+L  = BELGRADE
+O  = SUPERMQ
+OU = SUPERMQ
+
+[v3_req]
+extendedKeyUsage = clientAuth
+subjectAltName = @alt_names
+
+[alt_names]
+URI.1 = <<URI_SAN>>
+endef
+
 define ANNOUNCE_BODY
 Version $(VERSION) of $(PACKAGE_NAME) has been released.
 
@@ -72,7 +140,7 @@ It can be downloaded from $(DOWNLOAD_URL).
 
 etc, etc.
 endef
-all: clean_certs ca server_cert auth_grpc_certs domains_grpc_certs groups_grpc_certs clients_grpc_certs channels_grpc_certs coap_dtls_certs
+all: clean_certs ca server_cert auth_grpc_certs domains_grpc_certs groups_grpc_certs clients_grpc_certs channels_grpc_certs coap_dtls_certs fluxmq_service_certs fluxmq_service_secret timescale_writer_fluxmq_service_secret postgres_writer_fluxmq_service_secret fluxmq_auth_fluxmq_service_secret re_trace_key
 
 # CA name and key is "ca".
 ca:
@@ -139,6 +207,120 @@ define gen_grpc_cert_pair
 	$(call gen_grpc_cert,$(3),$(4))
 endef
 
+# Usage: $(call gen_fluxmq_server_cert,cert_file_name)
+define gen_fluxmq_server_cert
+	$(file > $(CRT_LOCATION)/$(1).conf,$(FLUXMQ_SERVICE_SERVER_CERT_CONFIG))
+
+	openssl req -new -sha256 -newkey rsa:4096 -nodes \
+		-keyout $(CRT_LOCATION)/$(1).key \
+		-out $(CRT_LOCATION)/$(1).csr \
+		-config $(CRT_LOCATION)/$(1).conf \
+		-extensions v3_req
+
+	openssl x509 -req -sha256 \
+		-in $(CRT_LOCATION)/$(1).csr \
+		-CA $(CRT_LOCATION)/ca.crt \
+		-CAkey $(CRT_LOCATION)/ca.key \
+		-CAcreateserial \
+		-out $(CRT_LOCATION)/$(1).crt \
+		-days 365 \
+		-extfile $(CRT_LOCATION)/$(1).conf \
+		-extensions v3_req
+
+	rm -rf $(CRT_LOCATION)/$(1).csr $(CRT_LOCATION)/$(1).conf
+endef
+
+# Usage: $(call gen_uri_san_cert,cert_file_name,common_name,uri_san)
+define gen_uri_san_cert
+	$(file > $(CRT_LOCATION)/$(1).conf,$(subst <<URI_SAN>>,$(3),$(subst <<SERVICE_NAME>>,$(2),$(URI_SAN_CERT_CONFIG))))
+
+	openssl req -new -sha256 -newkey rsa:4096 -nodes \
+		-keyout $(CRT_LOCATION)/$(1).key \
+		-out $(CRT_LOCATION)/$(1).csr \
+		-config $(CRT_LOCATION)/$(1).conf \
+		-extensions v3_req
+
+	openssl x509 -req -sha256 \
+		-in $(CRT_LOCATION)/$(1).csr \
+		-CA $(CRT_LOCATION)/ca.crt \
+		-CAkey $(CRT_LOCATION)/ca.key \
+		-CAcreateserial \
+		-out $(CRT_LOCATION)/$(1).crt \
+		-days 365 \
+		-extfile $(CRT_LOCATION)/$(1).conf \
+		-extensions v3_req
+
+	rm -rf $(CRT_LOCATION)/$(1).csr $(CRT_LOCATION)/$(1).conf
+endef
+
+# Certificates for the FluxMQ service listener: a server certificate for the
+# broker and a client certificate carrying each local principal's URI SAN. The
+# listener requires mTLS, so both sides are issued by the same CA.
+fluxmq_service_certs:
+	$(call gen_fluxmq_server_cert,$(FLUXMQ_SERVICE_SERVER_CRT_FILE_NAME))
+	$(call gen_uri_san_cert,$(RE_FLUXMQ_CLIENT_CRT_FILE_NAME),$(RE_FLUXMQ_CLIENT_CN),$(RE_FLUXMQ_CLIENT_URI_SAN))
+	$(call gen_uri_san_cert,$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CN),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN))
+	$(call gen_uri_san_cert,$(POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(POSTGRES_WRITER_FLUXMQ_CLIENT_CN),$(POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN))
+	$(call gen_uri_san_cert,$(FLUXMQ_AUTH_FLUXMQ_CLIENT_CRT_FILE_NAME),$(FLUXMQ_AUTH_FLUXMQ_CLIENT_CN),$(FLUXMQ_AUTH_FLUXMQ_CLIENT_URI_SAN))
+
+# Generate a single principal's certificate when adding it to an existing
+# deployment, without rotating the broker certificate or the other principals.
+timescale_writer_fluxmq_client_cert:
+	$(call gen_uri_san_cert,$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CN),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN))
+
+postgres_writer_fluxmq_client_cert:
+	$(call gen_uri_san_cert,$(POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(POSTGRES_WRITER_FLUXMQ_CLIENT_CN),$(POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN))
+
+fluxmq_auth_fluxmq_client_cert:
+	$(call gen_uri_san_cert,$(FLUXMQ_AUTH_FLUXMQ_CLIENT_CRT_FILE_NAME),$(FLUXMQ_AUTH_FLUXMQ_CLIENT_CN),$(FLUXMQ_AUTH_FLUXMQ_CLIENT_URI_SAN))
+
+# The SASL secret a local principal presents alongside its certificate. FluxMQ
+# reads it from a file and the service sends it in its broker URL, so both sides
+# must see the same value: .env is the single source and these targets derive
+# the file from it. FluxMQ requires at least 32 bytes.
+#
+# Usage: $(call gen_fluxmq_service_secret,env_variable_name,secret_file_name)
+define gen_fluxmq_service_secret
+	@mkdir -p ../fluxmq/secrets
+	@secret="$${$(1)}"; \
+	if [ -z "$$secret" ]; then \
+		secret=$$(grep -E "^$(1)=" ../.env | cut -d= -f2-); \
+	fi; \
+	if [ -z "$$secret" ]; then \
+		echo "$(1) is not set in docker/.env"; exit 1; \
+	fi; \
+	if [ $${#secret} -lt 32 ]; then \
+		echo "$(1) must be at least 32 characters"; exit 1; \
+	fi; \
+	printf '%s' "$$secret" > ../fluxmq/secrets/$(2); \
+	chmod 600 ../fluxmq/secrets/$(2); \
+	echo "wrote docker/fluxmq/secrets/$(2)"
+endef
+
+fluxmq_service_secret:
+	$(call gen_fluxmq_service_secret,MG_RE_BROKER_SECRET,re-current)
+
+timescale_writer_fluxmq_service_secret:
+	$(call gen_fluxmq_service_secret,MG_TIMESCALE_WRITER_BROKER_SECRET,timescale-writer-current)
+
+postgres_writer_fluxmq_service_secret:
+	$(call gen_fluxmq_service_secret,MG_POSTGRES_WRITER_BROKER_SECRET,postgres-writer-current)
+
+fluxmq_auth_fluxmq_service_secret:
+	$(call gen_fluxmq_service_secret,MG_FLUXMQ_BROKER_SECRET,fluxmq-auth-current)
+
+# Persistent HMAC key shared by Rules Engine replicas. Do not replace an
+# existing key: doing so while old messages are in flight would invalidate
+# their loop traces. The containing directory is excluded from git.
+re_trace_key:
+	@mkdir -p ../re/secrets
+	@if [ ! -s ../re/secrets/trace.key ]; then \
+		umask 077; openssl rand -hex 32 > ../re/secrets/trace.key; \
+		echo "wrote docker/re/secrets/trace.key"; \
+	else \
+		echo "preserving docker/re/secrets/trace.key"; \
+	fi
+
 auth_grpc_certs:
 	$(call gen_grpc_cert_pair,$(AUTH_GRPC_SERVER_CRT_FILE_NAME),$(AUTH_GRPC_SERVER_CN),$(AUTH_GRPC_CLIENT_CRT_FILE_NAME),$(AUTH_GRPC_CLIENT_CN))
 

go.mod

@@ -7,7 +7,7 @@ require (
 	connectrpc.com/otelconnect v0.9.0
 	github.com/0x6flab/namegenerator v1.4.0
 	github.com/absmach/callhome v0.18.2
-	github.com/absmach/fluxmq v0.40.0
+	github.com/absmach/fluxmq v0.51.0
 	github.com/absmach/senml v1.0.8
 	github.com/caarlos0/env/v10 v10.0.0
 	github.com/caarlos0/env/v11 v11.4.1
@@ -17,7 +17,7 @@ require (
 	github.com/fiorix/go-smpp v0.0.0-20210403173735-2894b96e70ba
 	github.com/go-chi/chi/v5 v5.3.1
 	github.com/go-kit/kit v0.13.0
-	github.com/gofrs/uuid/v5 v5.4.0
+	github.com/gofrs/uuid/v5 v5.5.0
 	github.com/google/uuid v1.6.0
 	github.com/gookit/color v1.6.1
 	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
@@ -54,7 +54,7 @@ require (
 	golang.org/x/net v0.57.0
 	golang.org/x/sync v0.22.0
 	gonum.org/v1/gonum v0.17.0
-	google.golang.org/grpc v1.82.1
+	google.golang.org/grpc v1.83.0
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
 	gopkg.in/yaml.v3 v3.0.1
@@ -114,7 +114,7 @@ require (
 	github.com/lestrrat-go/iter v1.0.2 // indirect
 	github.com/lestrrat-go/option v1.0.1 // indirect
 	github.com/mattn/go-colorable v0.1.15 // indirect
-	github.com/mattn/go-isatty v0.0.22 // indirect
+	github.com/mattn/go-isatty v0.0.23 // indirect
 	github.com/mattn/go-sqlite3 v1.14.45 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/moby/api v1.54.2 // indirect
@@ -128,7 +128,7 @@ require (
 	github.com/opencontainers/image-spec v1.1.1 // indirect
 	github.com/opencontainers/runc v1.3.6 // indirect
 	github.com/pelletier/go-toml/v2 v2.4.1 // indirect
-	github.com/pion/dtls/v3 v3.1.4 // indirect
+	github.com/pion/dtls/v3 v3.1.5 // indirect
 	github.com/pion/logging v0.2.4 // indirect
 	github.com/pion/transport/v4 v4.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
@@ -156,11 +156,11 @@ require (
 	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/exp v0.0.0-20260611194520-c48552f49976 // indirect
+	golang.org/x/exp v0.0.0-20260718201538-764159d718ef // indirect
 	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260610212136-7ab31c22f7ad // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260610212136-7ab31c22f7ad // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260715232425-e75dac1f907d // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260715232425-e75dac1f907d // indirect
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
 )

go.sum

@@ -20,8 +20,8 @@ github.com/VividCortex/gohistogram v1.0.0 h1:6+hBz+qvs0JOrrNhhmR7lFxo5sINxBCGXrd
 github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g=
 github.com/absmach/callhome v0.18.2 h1:dmopRHm2qTheHN1hdUKRRYpKwRrj7X9d8AWCFrb+K6s=
 github.com/absmach/callhome v0.18.2/go.mod h1:LEXKhES9JJtj3tBgTZv7VPNjOi5ukJQB0mFic0QP60Q=
-github.com/absmach/fluxmq v0.40.0 h1:J7s6PHXliWRfwpQpt/umRNnrUegj9xYsVyK9BV8Azhk=
-github.com/absmach/fluxmq v0.40.0/go.mod h1:oVbq3VlkD0vPKc45gkcxXpD2tbweyF+CDw1x72AS+PA=
+github.com/absmach/fluxmq v0.51.0 h1:bts5xmgTpUFFmcya0YyaIgl3/Hb7/rzAlkrirLd5Dk0=
+github.com/absmach/fluxmq v0.51.0/go.mod h1:S/4pWXGHmGAwa8pZ4Doql9H1cr5jaSC1lFvs0JGmUVM=
 github.com/absmach/senml v1.0.8 h1:+opem/r4g6c6eA/JLyCIuksyEhj7eBdysY3pEmy1mqo=
 github.com/absmach/senml v1.0.8/go.mod h1:DRhzHLgvQoIUHroBgpFrSWso+bJZO9E96RlHAHy+VRI=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -114,8 +114,8 @@ github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPE
 github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
 github.com/goccy/go-json v0.10.6 h1:p8HrPJzOakx/mn/bQtjgNjdTcN+/S6FcG2CTtQOrHVU=
 github.com/goccy/go-json v0.10.6/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
-github.com/gofrs/uuid/v5 v5.4.0 h1:EfbpCTjqMuGyq5ZJwxqzn3Cbr2d0rUZU7v5ycAk/e/0=
-github.com/gofrs/uuid/v5 v5.4.0/go.mod h1:CDOjlDMVAtN56jqyRUZh58JT31Tiw7/oQyEXZV+9bD8=
+github.com/gofrs/uuid/v5 v5.5.0 h1:FkPv6jYQRbZtH3bD8yC7106u+CedTCLF8+t7CLHSZNo=
+github.com/gofrs/uuid/v5 v5.5.0/go.mod h1:bbAA98EoIlxyRHIVg6ektCSsZ5n8mSbwgEhvhMYlZgg=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
@@ -204,8 +204,8 @@ github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy
 github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
-github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4=
-github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4=
+github.com/mattn/go-isatty v0.0.23 h1:cYwCQTQf3HB6xUC+BtyCLZNr7IzbOmoZbmssVNzSyiQ=
+github.com/mattn/go-isatty v0.0.23/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
 github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/mattn/go-sqlite3 v1.14.45 h1:6KA/spDguL3KV8rnybG7ezSaE4SeMR3KC9VbUoAQaIk=
 github.com/mattn/go-sqlite3 v1.14.45/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ=
@@ -246,8 +246,8 @@ github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3v
 github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pelletier/go-toml/v2 v2.4.1 h1:j5OMOImsH+j2k7GJ5YO+RxfWwohNiH6t5zB/+h3bagc=
 github.com/pelletier/go-toml/v2 v2.4.1/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
-github.com/pion/dtls/v3 v3.1.4 h1:QhvtMflMfu9Kf0RcDC5BJBle4caPskByrKQR6uuYqpY=
-github.com/pion/dtls/v3 v3.1.4/go.mod h1:cr/qotLISUw/9C1m83ZPNZtj9WnXkYLpfCptPqbkInc=
+github.com/pion/dtls/v3 v3.1.5 h1:9xJtVsHwMYeSjPp5Hh1FTis4DchnQWtnOa5o+6ygqfc=
+github.com/pion/dtls/v3 v3.1.5/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU=
 github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8=
 github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so=
 github.com/pion/transport/v4 v4.0.2 h1:ifYlPqNwsy6aKQ9y8yzxXlHae5431ZrH2avkD/Rn6Tk=
@@ -362,8 +362,8 @@ go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
 golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
-golang.org/x/exp v0.0.0-20260611194520-c48552f49976 h1:X8Hz2ImujgbmetVuW+w2YkyZChE3cBpZi2P158rTG9M=
-golang.org/x/exp v0.0.0-20260611194520-c48552f49976/go.mod h1:vnf4pv9iKZXY58sQE1L86zmNWJ4159e1RkcWiLCkeEY=
+golang.org/x/exp v0.0.0-20260718201538-764159d718ef h1:LkZ48HFgy/TvhTI0bcWkjgFkgLyKUwcTbDjS0DUjw+A=
+golang.org/x/exp v0.0.0-20260718201538-764159d718ef/go.mod h1:EdfpwwqSu+0Li0mzskwHU6FWDV3t9Q+RZDo3QMUtL3Q=
 golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
 golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
 golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
@@ -383,12 +383,12 @@ golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
-google.golang.org/genproto/googleapis/api v0.0.0-20260610212136-7ab31c22f7ad h1:3iLyITS/sySRwbUKoC7ogfj2Yr1Cjs0pfaRKj5U5HEw=
-google.golang.org/genproto/googleapis/api v0.0.0-20260610212136-7ab31c22f7ad/go.mod h1:KdNqO+rCIWgFumrNBSEDlDNrkrQnpkax7Tv1WxNY8V4=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260610212136-7ab31c22f7ad h1:45WmJvIV6C2+O/jjLkPUH+F3aOj/1miDoU2DD0+NWbg=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260610212136-7ab31c22f7ad/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
-google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
-google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
+google.golang.org/genproto/googleapis/api v0.0.0-20260715232425-e75dac1f907d h1:QwnJwPte4XXAkhPu26LTDIahnsMSUV0kK8HkxbC+Pc4=
+google.golang.org/genproto/googleapis/api v0.0.0-20260715232425-e75dac1f907d/go.mod h1:WRrQ7/7N19PypuT0fxLOL5Lq0waoiRri4FbtHDEKrGE=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260715232425-e75dac1f907d h1:Jkpk39hlTZOIp3RbfvNX9R8Hv+Sw0X89nlU/xFOErsc=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260715232425-e75dac1f907d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
+google.golang.org/grpc v1.83.0 h1:JeNZEKJFbQxArAMl+hiytHauacDNqJUllNfmIMmpqnQ=
+google.golang.org/grpc v1.83.0/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc h1:2gGKlE2+asNV9m7xrywl36YYNnBG5ZQ0r/BOOxqPpmk=

pkg/messaging/README.md

@@ -24,6 +24,8 @@ Publish routing depends on the topic and the publisher prefix.
 
 The `$queue/` prefix lets any publisher force delivery into the durable stream queue regardless of its own prefix. This is used internally (e.g. by `writers`, `alarms`) to guarantee at-least-once delivery through the broker's stream.
 
+Addressing a queue is not the same as one existing. Each stream is captured by its own `$queue/<name>/#` binding in the broker configuration, and a publication matching no binding is dropped without an error — a failed or absent capture never fails the publish. A new `$queue/<name>` namespace therefore needs its queue declared in `docker/fluxmq/node{1,2,3}.yaml` before anything is published to it. The bindings are deliberately disjoint, so that a message lands in exactly one stream rather than also accumulating in the reserved `mqtt` queue; `docker/fluxmq/config_test.go` holds that invariant.
+
 ### Stream queues
 
 On startup, publishers and pubsub clients normally declare a durable stream queue named after their prefix. Stream subscribers use consumer groups, so each group receives every message exactly once. The default stream queue is named `m`. `InternalMetadata` instead requires that stream to be pre-provisioned by the broker and never attempts to create or modify it.
@@ -32,6 +34,12 @@ On startup, publishers and pubsub clients normally declare a durable stream queu
 
 `Subscribe` attaches to the durable stream queue via a consumer group filtered by topic. Optionally (when `DirectTopicIngress` is enabled), it also subscribes to the raw MQTT topic so that messages published directly by MQTT clients — bypassing the queue — are also received. A deployment using `InternalMetadata` must authorize the requested subscriptions explicitly; the Rules Engine local principal authorizes only pre-provisioned stream `m`.
 
+### Message origin
+
+A message carries the protocol it was published with (`mqtt`, `http`, `coap`, …) and the identity of its publisher. Both are broker-controlled: on a publication from an untrusted connection the broker overwrites them with the transport and identity of that connection, so a service that consumes a device message and republishes it — into the `writers` stream, for instance — turns every one of them into `protocol: amqp` published by that service.
+
+`InternalMetadata` is what avoids this. A connection authenticated as a `service`-role local principal on the mTLS listener may relay the origin protocol, publisher, `created` timestamp and metadata it received rather than having its own stamped on. Any service that republishes messages someone else authored has to use it, and its principal needs a `permissions.publish` entry for the destination.
+
 ### Options
 
 | Option                                 | Description                                                                                       |

pkg/messaging/fluxmq/pubsub.go

@@ -124,7 +124,13 @@ func (ps *pubsub) Subscribe(_ context.Context, cfg messaging.SubscriberConfig) e
 
 		sub.streamTopic = queueFilter(ps.prefix, cfg.Topic)
 	}
-	if ps.directTopicIngress {
+	// A preprovisioned connection is a local principal on the mTLS service
+	// listener, and that listener only serves queue addresses: FluxMQ resolves
+	// a bare topic filter to a pub/sub route and refuses it, because no
+	// subscribe ACL entry can name one. The direct subscription is also
+	// redundant there -- the broker-provisioned stream binds the same topic
+	// patterns, so a direct publish reaches the stream consumer anyway.
+	if ps.directTopicIngress && !ps.preprovisioned {
 		// Subscribe to regular MQTT topics so that messages published directly
 		// by MQTT clients (not through the stream queue) are also received.
 		sub.mqttTopic = topicFilter(ps.prefix, cfg.Topic)

pkg/messaging/writers/brokers_fluxmq.go

@@ -34,8 +34,23 @@ var cfg = jetstream.StreamConfig{
 	Storage:           jetstream.FileStorage,
 }
 
-func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.PubSub, error) {
-	pb, err := broker.NewPubSub(ctx, url, logger, broker.Prefix(prefix), broker.JSStreamConfig(cfg), broker.ConnectionName("writers-msg-pubsub"))
+// InternalMetadata returns an option for a trusted local-service connection
+// that carries internal metadata over mTLS and consumes the broker-provisioned
+// writers stream. It lives here rather than being taken from
+// pkg/messaging/brokers because that package selects its backend on a different
+// build tag: an untagged build would pair a FluxMQ option with a NATS PubSub.
+func InternalMetadata(certFile, keyFile, caFile string) messaging.Option {
+	return broker.InternalMetadata(certFile, keyFile, caFile)
+}
+
+func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...messaging.Option) (messaging.PubSub, error) {
+	brokerOpts := []messaging.Option{
+		broker.Prefix(prefix),
+		broker.JSStreamConfig(cfg),
+		broker.ConnectionName("writers-msg-pubsub"),
+	}
+	brokerOpts = append(brokerOpts, opts...)
+	pb, err := broker.NewPubSub(ctx, url, logger, brokerOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -43,8 +58,19 @@ func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.
 	return pb, nil
 }
 
-func NewPublisher(ctx context.Context, url string) (messaging.Publisher, error) {
-	pb, err := broker.NewPublisher(ctx, url, broker.Prefix(prefix), broker.JSStreamConfig(cfg), broker.ConnectionName("writers-msg-pub"))
+// NewPublisher creates the publisher that feeds the writers stream. Pass
+// InternalMetadata so it connects as a trusted local principal: the broker
+// stamps its own transport protocol and identity on a publication from an
+// untrusted connection, which would leave every stored message recorded as
+// having arrived over AMQP instead of the protocol its device spoke.
+func NewPublisher(ctx context.Context, url string, opts ...messaging.Option) (messaging.Publisher, error) {
+	brokerOpts := []messaging.Option{
+		broker.Prefix(prefix),
+		broker.JSStreamConfig(cfg),
+		broker.ConnectionName("writers-msg-pub"),
+	}
+	brokerOpts = append(brokerOpts, opts...)
+	pb, err := broker.NewPublisher(ctx, url, brokerOpts...)
 	if err != nil {
 		return nil, err
 	}

pkg/messaging/writers/brokers_nats.go

@@ -34,8 +34,20 @@ var cfg = jetstream.StreamConfig{
 	Storage:           jetstream.FileStorage,
 }
 
-func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.PubSub, error) {
-	pb, err := broker.NewPubSub(ctx, url, logger, broker.Prefix(prefix), broker.JSStreamConfig(cfg))
+// InternalMetadata is a no-op for the NATS backend. It exists for compile-time
+// compatibility with the FluxMQ variant; NATS carries metadata in the protobuf
+// message.
+func InternalMetadata(_, _, _ string) messaging.Option {
+	return func(_ any) error { return nil }
+}
+
+func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...messaging.Option) (messaging.PubSub, error) {
+	brokerOpts := []messaging.Option{
+		broker.Prefix(prefix),
+		broker.JSStreamConfig(cfg),
+	}
+	brokerOpts = append(brokerOpts, opts...)
+	pb, err := broker.NewPubSub(ctx, url, logger, brokerOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -43,8 +55,13 @@ func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.
 	return pb, nil
 }
 
-func NewPublisher(ctx context.Context, url string) (messaging.Publisher, error) {
-	pb, err := broker.NewPublisher(ctx, url, broker.Prefix(prefix), broker.JSStreamConfig(cfg))
+func NewPublisher(ctx context.Context, url string, opts ...messaging.Option) (messaging.Publisher, error) {
+	brokerOpts := []messaging.Option{
+		broker.Prefix(prefix),
+		broker.JSStreamConfig(cfg),
+	}
+	brokerOpts = append(brokerOpts, opts...)
+	pb, err := broker.NewPublisher(ctx, url, brokerOpts...)
 	if err != nil {
 		return nil, err
 	}