Update compose and deployments

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

Makefile

@@ -197,7 +197,7 @@ 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 $(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
 
 clean:
 	rm -rf ${BUILD_DIR}
@@ -349,7 +349,23 @@ else
 	$(eval GRPC_MTLS :=)
 endif
 
-check_certs: check_mtls check_tls
+# The Rules Engine reaches FluxMQ on the mTLS local listener, and compose mounts
+# both sides of that connection. The certificates and the principal secret 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/fluxmq/secrets/re-current)","")
+	$(MAKE) -C docker/ssl 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))

README.md

@@ -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,88 @@ 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 to that listener |
+| `docker/fluxmq/secrets/re-current`                 | The principal secret, derived from `MG_RE_BROKER_SECRET` in `docker/.env`     |
+| `docker/re/secrets/trace.key`                      | HMAC key the Rules Engine signs its loop-detection traces with                |
+
+Both 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
+`MG_RE_BROKER_URL` at a host outside that set and the Rules Engine 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.*
+make -C docker/ssl fluxmq_service_certs
+```
+
+`make check_certs` skips this pair when it already exists, so a stale
+certificate has to be removed rather than merely re-run.
+
+Two of these are stateful, not merely derived. `re-current` must stay equal to
+`MG_RE_BROKER_SECRET`: change the variable and re-run `make -C docker/ssl
+fluxmq_service_secret`, or the Rules Engine fails to authenticate.
+`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

docker/.env

@@ -569,6 +569,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/docker-compose.yaml

@@ -255,6 +255,12 @@ 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 the
+      # local-principal secret it authenticates the Rules Engine against.
+      - ./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
       - magistrala-fluxmq-node1-volume:/tmp/fluxmq
 
   fluxmq-node2:
@@ -275,6 +281,12 @@ 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 the
+      # local-principal secret it authenticates the Rules Engine against.
+      - ./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
       - magistrala-fluxmq-node2-volume:/tmp/fluxmq
 
   fluxmq-node3:
@@ -295,6 +307,12 @@ 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 the
+      # local-principal secret it authenticates the Rules Engine against.
+      - ./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
       - magistrala-fluxmq-node3-volume:/tmp/fluxmq
 
   fluxmq-auth:
@@ -761,6 +779,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 +806,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/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
@@ -127,6 +141,29 @@ 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
+  # the secret file must hold the same value as MG_RE_BROKER_SECRET.
+  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 message stream the Rules Engine consumes.
+        subscribe:
+          - "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
@@ -124,6 +138,29 @@ 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
+  # the secret file must hold the same value as MG_RE_BROKER_SECRET.
+  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 message stream the Rules Engine consumes.
+        subscribe:
+          - "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
@@ -124,6 +138,29 @@ 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
+  # the secret file must hold the same value as MG_RE_BROKER_SECRET.
+  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 message stream the Rules Engine consumes.
+        subscribe:
+          - "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,14 @@ 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
 
 define GRPC_CERT_CONFIG
 [req]
@@ -65,6 +73,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 +131,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 re_trace_key
 
 # CA name and key is "ca".
 ca:
@@ -139,6 +198,91 @@ 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 the Rules Engine'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))
+
+# The SASL secret the Rules Engine presents alongside its certificate. FluxMQ
+# reads it from a file and the Rules Engine sends it in its broker URL, so both
+# sides must see the same value: .env is the single source and this target
+# derives the file from it. FluxMQ requires at least 32 bytes.
+fluxmq_service_secret:
+	@mkdir -p ../fluxmq/secrets
+	@secret="$${MG_RE_BROKER_SECRET}"; \
+	if [ -z "$$secret" ]; then \
+		secret=$$(grep -E "^MG_RE_BROKER_SECRET=" ../.env | cut -d= -f2-); \
+	fi; \
+	if [ -z "$$secret" ]; then \
+		echo "MG_RE_BROKER_SECRET is not set in docker/.env"; exit 1; \
+	fi; \
+	if [ $${#secret} -lt 32 ]; then \
+		echo "MG_RE_BROKER_SECRET must be at least 32 characters"; exit 1; \
+	fi; \
+	printf '%s' "$$secret" > ../fluxmq/secrets/re-current; \
+	chmod 600 ../fluxmq/secrets/re-current; \
+	echo "wrote docker/fluxmq/secrets/re-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))
 

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)