diff --git a/Makefile b/Makefile index 536c96aaf..450d9bd79 100644 --- a/Makefile +++ b/Makefile @@ -369,6 +369,9 @@ 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 @@ -378,6 +381,9 @@ 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)","") @@ -403,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)" \ diff --git a/README.md b/README.md index 2a0c31849..f66cfc19d 100644 --- a/README.md +++ b/README.md @@ -306,17 +306,29 @@ This creates whatever is missing and leaves anything already present alone: | `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` and republishes under it, the writers only subscribe to `writers`. 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. +`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 @@ -348,6 +360,8 @@ changing one, re-run its 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. diff --git a/cmd/fluxmq/main.go b/cmd/fluxmq/main.go index 346d4826f..c2d55c7ab 100644 --- a/cmd/fluxmq/main.go +++ b/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 diff --git a/docker/.env b/docker/.env index 1eb94a617..a8a007956 100644 --- a/docker/.env +++ b/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 diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index 81f2a94e8..0f6561e4a 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -263,6 +263,7 @@ services: - ./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: @@ -291,6 +292,7 @@ services: - ./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: @@ -319,6 +321,7 @@ services: - ./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: @@ -352,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 diff --git a/docker/fluxmq/config_test.go b/docker/fluxmq/config_test.go new file mode 100644 index 000000000..1f6c703fd --- /dev/null +++ b/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) + } + } + } +} diff --git a/docker/fluxmq/node1.yaml b/docker/fluxmq/node1.yaml index 17446fb43..ff349ca4f 100644 --- a/docker/fluxmq/node1.yaml +++ b/docker/fluxmq/node1.yaml @@ -108,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: @@ -161,6 +166,15 @@ auth: # 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//#" binding is what captures it. + - routing_key_prefix: "$queue/writers/" + - routing_key_prefix: "$queue/alarms/" # The message stream the Rules Engine consumes. subscribe: - "m" @@ -182,6 +196,20 @@ auth: # 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..c.., 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" diff --git a/docker/fluxmq/node2.yaml b/docker/fluxmq/node2.yaml index a5274fb03..f9a57bb37 100644 --- a/docker/fluxmq/node2.yaml +++ b/docker/fluxmq/node2.yaml @@ -105,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: @@ -158,6 +163,15 @@ auth: # 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//#" binding is what captures it. + - routing_key_prefix: "$queue/writers/" + - routing_key_prefix: "$queue/alarms/" # The message stream the Rules Engine consumes. subscribe: - "m" @@ -179,6 +193,20 @@ auth: # 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..c.., 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" diff --git a/docker/fluxmq/node3.yaml b/docker/fluxmq/node3.yaml index 89f59bde0..eb01abc4c 100644 --- a/docker/fluxmq/node3.yaml +++ b/docker/fluxmq/node3.yaml @@ -105,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: @@ -158,6 +163,15 @@ auth: # 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//#" binding is what captures it. + - routing_key_prefix: "$queue/writers/" + - routing_key_prefix: "$queue/alarms/" # The message stream the Rules Engine consumes. subscribe: - "m" @@ -179,6 +193,20 @@ auth: # 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..c.., 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" diff --git a/docker/ssl/Makefile b/docker/ssl/Makefile index ccadc4b76..bcc7690cc 100644 --- a/docker/ssl/Makefile +++ b/docker/ssl/Makefile @@ -57,6 +57,9 @@ TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/timescale-wri 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] @@ -137,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 fluxmq_service_certs fluxmq_service_secret timescale_writer_fluxmq_service_secret postgres_writer_fluxmq_service_secret re_trace_key +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: @@ -258,6 +261,7 @@ fluxmq_service_certs: $(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. @@ -267,6 +271,9 @@ timescale_writer_fluxmq_client_cert: 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 @@ -299,6 +306,9 @@ timescale_writer_fluxmq_service_secret: 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. diff --git a/pkg/messaging/README.md b/pkg/messaging/README.md index 630703bfa..9429994cc 100644 --- a/pkg/messaging/README.md +++ b/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//#` 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/` 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 | diff --git a/pkg/messaging/writers/brokers_fluxmq.go b/pkg/messaging/writers/brokers_fluxmq.go index ddc4a9444..9ea249f64 100644 --- a/pkg/messaging/writers/brokers_fluxmq.go +++ b/pkg/messaging/writers/brokers_fluxmq.go @@ -58,8 +58,19 @@ func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...mes 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 } diff --git a/pkg/messaging/writers/brokers_nats.go b/pkg/messaging/writers/brokers_nats.go index c68d5ba21..97eef393c 100644 --- a/pkg/messaging/writers/brokers_nats.go +++ b/pkg/messaging/writers/brokers_nats.go @@ -55,8 +55,13 @@ func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...mes 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 }