opensource/magistrala
1
0
Fork 0
mirror of https://github.com/absmach/magistrala.git synced 2026-08-07 07:14:46 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

add back fluxmq-auth

Browse Source 
Signed-off-by: Arvindh <arvindh91@gmail.com>
This commit is contained in:
Arvindh
2026-06-25 21:50:18 +05:30
committed by dusan
parent 22ffcda201
commit 2dd5225288
16 changed files with 872 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+17 -4
View File
@@ -4,7 +4,7 @@
override MG_DOCKER_IMAGE_NAME_PREFIX := ghcr.io/absmach/magistrala
MG_DOCKER_VOLUME_NAME_PREFIX ?= magistrala
BUILD_DIR ?= build
SERVICES = atom-bootstrap notifications certs re postgres-writer postgres-reader timescale-writer timescale-reader alarms reports journal
SERVICES = atom-bootstrap notifications certs re postgres-writer postgres-reader timescale-writer timescale-reader alarms reports journal fluxmq
TEST_API_SERVICES = journal certs clients users channels groups domains
TEST_API = $(addprefix test_api_,$(TEST_API_SERVICES))
DOCKERS = $(addprefix docker_,$(SERVICES))
@@ -24,6 +24,7 @@ DOCKER_PROJECT ?= $(shell echo $(subst $(space),,$(USER_REPO)) | sed -E 's/[^a-z
DOCKER_COMPOSE_COMMANDS_SUPPORTED := up down config restart
DEFAULT_DOCKER_COMPOSE_COMMAND := up
ATOM_TOKENS_ENV ?= docker/.env.tokens
REQUIRED_ATOM_TOKEN_ENVS := MG_ATOM_TOKEN_FLUXMQ_AUTH MG_ATOM_TOKEN_FLUXMQ_NODE1 MG_ATOM_TOKEN_FLUXMQ_NODE2 MG_ATOM_TOKEN_FLUXMQ_NODE3 MG_ATOM_TOKEN_JOURNAL MG_ATOM_TOKEN_NOTIFICATIONS MG_ATOM_TOKEN_TIMESCALE_READER MG_ATOM_TOKEN_RE MG_ATOM_TOKEN_ALARMS MG_ATOM_TOKEN_REPORTS MG_ATOM_TOKEN_POSTGRES_READER
DOCKER_BASE_ENV_FILES := --env-file docker/.env
DOCKER_ENV_FILES = $(if $(filter down,$(DOCKER_COMPOSE_COMMAND)),$(DOCKER_BASE_ENV_FILES),$(DOCKER_BASE_ENV_FILES) --env-file $(ATOM_TOKENS_ENV))
DOCKER_PROVISION_ENV_FILES = $(DOCKER_BASE_ENV_FILES) $(if $(wildcard $(ATOM_TOKENS_ENV)),--env-file $(ATOM_TOKENS_ENV))
@@ -86,9 +87,21 @@ define make_docker_dev
endef
define require_atom_tokens_env
@if [ -z "$(filter down,$(DOCKER_COMPOSE_COMMAND))" ] && [ ! -f "$(ATOM_TOKENS_ENV)" ]; then \
echo "Missing $(ATOM_TOKENS_ENV). Run 'make provision_atom_tokens' before starting the Docker Compose stack."; \
exit 2; \
@if [ -z "$(filter down,$(DOCKER_COMPOSE_COMMAND))" ]; then \
if [ ! -f "$(ATOM_TOKENS_ENV)" ]; then \
echo "Missing $(ATOM_TOKENS_ENV). Run 'make provision_atom_tokens' before starting the Docker Compose stack."; \
exit 2; \
fi; \
missing=""; \
for env_name in $(REQUIRED_ATOM_TOKEN_ENVS); do \
if ! grep -q "^$${env_name}=" "$(ATOM_TOKENS_ENV)"; then \
missing="$${missing} $${env_name}"; \
fi; \
done; \
if [ -n "$${missing}" ]; then \
echo "Missing Atom service token(s) in $(ATOM_TOKENS_ENV):$${missing}. Run 'make provision_atom_tokens' before starting the Docker Compose stack."; \
exit 2; \
fi; \
fi
endef
cmd/fluxmq/main.go
+307
View File
@@ -0,0 +1,307 @@
// Copyright (c) Abstract Machines
// SPDX-License-Identifier: Apache-2.0
// Package main contains the FluxMQ auth bridge service entry point.
// This service implements the FluxMQ auth callout server using ConnectRPC,
// bridging authentication requests to Magistrala's Clients service and
// authorization requests to Magistrala's Channels service.
package main
import (
"context"
"errors"
"fmt"
"log"
"net/http"
"net/url"
"os"
"os/signal"
"syscall"
"connectrpc.com/connect"
"connectrpc.com/otelconnect"
"github.com/absmach/fluxmq/pkg/proto/auth/v1/authv1connect"
fluxmqgrpc "github.com/absmach/magistrala/fluxmq/api/grpc"
fluxmqhttp "github.com/absmach/magistrala/fluxmq/api/http"
"github.com/absmach/magistrala/internal/atom"
mglog "github.com/absmach/magistrala/logger"
atomauthn "github.com/absmach/magistrala/pkg/authn/atom"
jaegerclient "github.com/absmach/magistrala/pkg/jaeger"
"github.com/absmach/magistrala/pkg/messaging"
fluxmqbroker "github.com/absmach/magistrala/pkg/messaging/fluxmq"
"github.com/absmach/magistrala/pkg/server"
httpserver "github.com/absmach/magistrala/pkg/server/http"
"github.com/absmach/magistrala/pkg/uuid"
"github.com/caarlos0/env/v11"
"golang.org/x/net/http2"
"golang.org/x/net/http2/h2c"
"golang.org/x/sync/errgroup"
)
const (
svcName = "fluxmq-auth"
defSvcGRPCPort = "7016"
envPrefixCache = "MG_FLUXMQ_CACHE_"
envPrefixGRPC = "MG_FLUXMQ_GRPC_"
envPrefixHTTP = "MG_FLUXMQ_PUBLISH_HTTP_"
)
type config struct {
LogLevel string `env:"MG_FLUXMQ_LOG_LEVEL" envDefault:"info"`
BrokerURL string `env:"MG_MESSAGE_BROKER_URL" envDefault:"amqp://guest:guest@localhost:5682/"`
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:""`
}
type fanoutPublisher struct {
publishers []messaging.Publisher
}
func (fp fanoutPublisher) Publish(ctx context.Context, topic string, msg *messaging.Message) error {
for _, publisher := range fp.publishers {
if err := publisher.Publish(ctx, topic, msg); err != nil {
return err
}
}
return nil
}
func (fp fanoutPublisher) Close() error {
errs := make([]error, 0, len(fp.publishers))
for _, publisher := range fp.publishers {
errs = append(errs, publisher.Close())
}
return errors.Join(errs...)
}
type writerBridgeHandler struct {
ctx context.Context
publisher messaging.Publisher
}
func (h writerBridgeHandler) Handle(msg *messaging.Message) error {
return h.publisher.Publish(h.ctx, messaging.EncodeMessageTopic(msg), msg)
}
func (h writerBridgeHandler) Cancel() error {
return nil
}
func main() {
ctx, cancel := context.WithCancel(context.Background())
g, ctx := errgroup.WithContext(ctx)
cfg := config{}
if err := env.Parse(&cfg); err != nil {
log.Fatalf("failed to load %s configuration: %s", svcName, err)
}
logger, err := mglog.New(os.Stdout, cfg.LogLevel)
if err != nil {
log.Fatalf("failed to init logger: %s", err.Error())
}
var exitCode int
defer mglog.ExitWithError(&exitCode)
if cfg.InstanceID == "" {
if cfg.InstanceID, err = uuid.New().ID(); err != nil {
logger.Error(fmt.Sprintf("failed to generate instanceID: %s", err))
exitCode = 1
return
}
}
tp, err := jaegerclient.NewProvider(ctx, svcName, cfg.JaegerURL, cfg.InstanceID, cfg.TraceRatio)
if err != nil {
logger.Error(fmt.Sprintf("failed to init Jaeger: %s", err))
exitCode = 1
return
}
defer func() {
if err := tp.Shutdown(ctx); err != nil {
logger.Error(fmt.Sprintf("error shutting down tracer provider: %v", err))
}
}()
atomCfg := atom.LoadConfig()
if atomCfg.URL == "" {
logger.Error("ATOM_URL is required")
exitCode = 1
return
}
atomAuthz := atom.NewClient(atomCfg)
authn := atomauthn.NewAuthentication()
clientsClient := atom.NewClientsCompat(authn, atomAuthz)
domainsClient := atom.NewDomainsCompat(atomAuthz)
channelsClient := atom.NewChannelsCompat(atomAuthz)
logger.Info("FluxMQ authentication, authorization, and route resolution configured to use Atom")
// Topic parser with cache for route resolution.
cacheConfig := messaging.CacheConfig{}
if err := env.ParseWithOptions(&cacheConfig, env.Options{Prefix: envPrefixCache}); err != nil {
logger.Error(fmt.Sprintf("failed to load cache configuration: %s", err))
exitCode = 1
return
}
parser, err := messaging.NewTopicParser(cacheConfig, channelsClient, domainsClient)
if err != nil {
logger.Error(fmt.Sprintf("failed to create topic parser: %s", err))
exitCode = 1
return
}
// Start FluxMQ auth Connect/gRPC server over h2c.
grpcServerConfig := server.Config{Port: defSvcGRPCPort}
if err := env.ParseWithOptions(&grpcServerConfig, env.Options{Prefix: envPrefixGRPC}); err != nil {
logger.Error(fmt.Sprintf("failed to load gRPC server configuration: %s", err))
exitCode = 1
return
}
mux := http.NewServeMux()
otelInterceptor, err := otelconnect.NewInterceptor()
if err != nil {
logger.Error(fmt.Sprintf("failed to create OTel interceptor: %s", err))
exitCode = 1
return
}
path, handler := authv1connect.NewAuthServiceHandler(
fluxmqgrpc.NewServer(clientsClient, channelsClient, parser, atomAuthz),
connect.WithInterceptors(otelInterceptor),
)
mux.Handle(path, handler)
mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte(`{"status":"ok"}`)) //nolint:errcheck // HTTP response write; client disconnect is non-fatal.
})
address := fmt.Sprintf("%s:%s", grpcServerConfig.Host, grpcServerConfig.Port)
httpServer := &http.Server{
Addr: address,
Handler: h2c.NewHandler(mux, &http2.Server{}),
ReadTimeout: grpcServerConfig.ReadTimeout,
WriteTimeout: grpcServerConfig.WriteTimeout,
ReadHeaderTimeout: grpcServerConfig.ReadHeaderTimeout,
IdleTimeout: grpcServerConfig.IdleTimeout,
MaxHeaderBytes: grpcServerConfig.MaxHeaderBytes,
}
messagePublisher, err := fluxmqbroker.NewUndeclaredPublisher(
ctx,
cfg.BrokerURL,
fluxmqbroker.ConnectionName("fluxmq-ui-message-publish-proxy"),
)
if err != nil {
logger.Error(fmt.Sprintf("failed to create publish proxy message publisher: %s", err))
exitCode = 1
return
}
defer messagePublisher.Close()
writerPublisher, err := fluxmqbroker.NewUndeclaredPublisher(
ctx,
cfg.BrokerURL,
fluxmqbroker.Prefix("writers"),
fluxmqbroker.ConnectionName("fluxmq-ui-publish-proxy"),
)
if err != nil {
logger.Error(fmt.Sprintf("failed to create publish proxy writer publisher: %s", err))
exitCode = 1
return
}
defer writerPublisher.Close()
publisher := fanoutPublisher{publishers: []messaging.Publisher{messagePublisher, writerPublisher}}
writerBridge, err := fluxmqbroker.NewPubSub(
ctx,
cfg.BrokerURL,
logger,
fluxmqbroker.DirectTopicOnly(),
fluxmqbroker.ConnectionName("fluxmq-mqtt-writer-bridge"),
)
if err != nil {
logger.Error(fmt.Sprintf("failed to create MQTT writer bridge subscriber: %s", err))
exitCode = 1
return
}
defer writerBridge.Close()
if err := writerBridge.Subscribe(ctx, messaging.SubscriberConfig{
ID: cfg.InstanceID + "-mqtt-writer-bridge",
Topic: "m/#",
Handler: writerBridgeHandler{ctx: ctx, publisher: writerPublisher},
DeliveryPolicy: messaging.DeliverNewPolicy,
}); err != nil {
logger.Error(fmt.Sprintf("failed to subscribe MQTT writer bridge: %s", err))
exitCode = 1
return
}
logger.Info("FluxMQ MQTT writer bridge subscribed", "topic", "m/#")
httpServerConfig := server.Config{Port: "9026"}
if err := env.ParseWithOptions(&httpServerConfig, env.Options{Prefix: envPrefixHTTP}); err != nil {
logger.Error(fmt.Sprintf("failed to load publish proxy HTTP server configuration: %s", err))
exitCode = 1
return
}
hs := httpserver.NewServer(
ctx,
cancel,
"fluxmq-publish",
httpServerConfig,
fluxmqhttp.MakePublishHandler(authn, atomAuthz, publisher),
logger,
)
g.Go(func() error {
logger.Info(fmt.Sprintf("%s service h2c server listening at %s", svcName, address))
var err error
switch {
case grpcServerConfig.CertFile != "" || grpcServerConfig.KeyFile != "":
err = httpServer.ListenAndServeTLS(grpcServerConfig.CertFile, grpcServerConfig.KeyFile)
default:
err = httpServer.ListenAndServe()
}
if err != nil && err != http.ErrServerClosed {
cancel()
return err
}
return nil
})
g.Go(func() error {
return hs.Start()
})
g.Go(func() error {
<-ctx.Done()
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), server.StopWaitTime) //nolint:contextcheck
defer shutdownCancel()
if err := hs.Stop(); err != nil {
return fmt.Errorf("failed to shutdown publish proxy server: %w", err)
}
if err := httpServer.Shutdown(shutdownCtx); err != nil { //nolint:contextcheck
return fmt.Errorf("failed to shutdown %s server: %w", svcName, err)
}
logger.Info(fmt.Sprintf("%s service shutdown at %s", svcName, address))
return nil
})
g.Go(func() error {
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
select {
case sig := <-c:
cancel()
logger.Info(fmt.Sprintf("%s service shutdown by signal: %s", svcName, sig))
return nil
case <-ctx.Done():
return nil
}
})
if err := g.Wait(); err != nil {
logger.Error(fmt.Sprintf("%s service terminated: %s", svcName, err))
}
}
docker/.env
+13
View File
@@ -32,6 +32,8 @@ MG_FLUXMQ_API_PORT_3=9083
## Message Broker
MG_MESSAGE_BROKER_URL=amqp://guest:guest@nginx:${MG_NGINX_AMQP_PORT}/
MG_FLUXMQ_PUBLISH_HTTP_HOST=fluxmq-auth
MG_FLUXMQ_PUBLISH_HTTP_PORT=9026
## Redis
MG_REDIS_TCP_PORT=6379
@@ -382,6 +384,16 @@ MG_CHANNELS_GRPC_CLIENT_CERT=${GRPC_MTLS:+./ssl/certs/channels-grpc-client.crt}
MG_CHANNELS_GRPC_CLIENT_KEY=${GRPC_MTLS:+./ssl/certs/channels-grpc-client.key}
MG_CHANNELS_GRPC_CLIENT_CA_CERTS=${GRPC_MTLS:+./ssl/certs/ca.crt}
### FluxMQ Auth Bridge
MG_FLUXMQ_LOG_LEVEL=debug
MG_FLUXMQ_GRPC_HOST=fluxmq-auth
MG_FLUXMQ_GRPC_PORT=7016
MG_FLUXMQ_GRPC_URL=fluxmq-auth:7016
MG_FLUXMQ_INSTANCE_ID=
MG_FLUXMQ_CACHE_NUM_COUNTERS=200000
MG_FLUXMQ_CACHE_MAX_COST=1048576
MG_FLUXMQ_CACHE_BUFFER_ITEMS=64
### CoAP
MG_COAP_PORT=5683
@@ -695,6 +707,7 @@ MG_CHANNELS_URL=http://nginx:80/resources
MG_GROUPS_URL=http://nginx:80/groups
MG_BOOTSTRAP_URL=http://bootstrap:9013
MG_HTTP_ADAPTER_URL=http://nginx:80/http
MG_PUBLISH_PROXY_URL=http://nginx:80
MG_READER_URL=http://timescale-reader:9011
MG_JOURNAL_URL=http://journal:9021
docker/docker-compose.yaml
+41 -12
View File
@@ -339,10 +339,8 @@ services:
user: "0:0"
command: ["-config", "/etc/fluxmq/config.yaml"]
depends_on:
atom-bootstrap:
condition: service_completed_successfully
environment:
FLUXMQ_ATOM_SERVICE_TOKEN: ${MG_ATOM_TOKEN_FLUXMQ_NODE1}
fluxmq-auth:
condition: service_started
restart: on-failure
ports:
- ${MG_COAP_PORT}:5683/udp
@@ -362,10 +360,8 @@ services:
depends_on:
fluxmq-node1:
condition: service_started
atom-bootstrap:
condition: service_completed_successfully
environment:
FLUXMQ_ATOM_SERVICE_TOKEN: ${MG_ATOM_TOKEN_FLUXMQ_NODE2}
fluxmq-auth:
condition: service_started
restart: on-failure
ports:
- ${MG_FLUXMQ_API_PORT_2}:8082
@@ -384,10 +380,8 @@ services:
depends_on:
fluxmq-node1:
condition: service_started
atom-bootstrap:
condition: service_completed_successfully
environment:
FLUXMQ_ATOM_SERVICE_TOKEN: ${MG_ATOM_TOKEN_FLUXMQ_NODE3}
fluxmq-auth:
condition: service_started
restart: on-failure
ports:
- ${MG_FLUXMQ_API_PORT_3}:8082
@@ -398,6 +392,40 @@ services:
- ./fluxmq/node3.yaml:/etc/fluxmq/config.yaml:ro
- magistrala-fluxmq-node3-volume:/tmp/fluxmq
fluxmq-auth:
image: ghcr.io/absmach/magistrala/fluxmq:${MG_RELEASE_TAG}
container_name: magistrala-fluxmq-auth
depends_on:
atom-bootstrap:
condition: service_completed_successfully
restart: on-failure
environment:
MG_FLUXMQ_LOG_LEVEL: ${MG_FLUXMQ_LOG_LEVEL}
MG_FLUXMQ_GRPC_HOST: ${MG_FLUXMQ_GRPC_HOST}
MG_FLUXMQ_GRPC_PORT: ${MG_FLUXMQ_GRPC_PORT}
MG_FLUXMQ_INSTANCE_ID: ${MG_FLUXMQ_INSTANCE_ID}
MG_FLUXMQ_CACHE_NUM_COUNTERS: ${MG_FLUXMQ_CACHE_NUM_COUNTERS}
MG_FLUXMQ_CACHE_MAX_COST: ${MG_FLUXMQ_CACHE_MAX_COST}
MG_FLUXMQ_CACHE_BUFFER_ITEMS: ${MG_FLUXMQ_CACHE_BUFFER_ITEMS}
MG_MESSAGE_BROKER_URL: ${MG_MESSAGE_BROKER_URL}
MG_FLUXMQ_PUBLISH_HTTP_HOST: ${MG_FLUXMQ_PUBLISH_HTTP_HOST}
MG_FLUXMQ_PUBLISH_HTTP_PORT: ${MG_FLUXMQ_PUBLISH_HTTP_PORT}
ATOM_URL: ${ATOM_URL}
ATOM_SERVICE_TOKEN: ${MG_ATOM_TOKEN_FLUXMQ_AUTH}
ATOM_SERVICE_USERNAME: ${ATOM_SERVICE_USERNAME}
ATOM_SERVICE_SECRET: ${ATOM_SERVICE_SECRET}
ATOM_ADMIN_TOKEN: ${ATOM_ADMIN_TOKEN}
ATOM_ADMIN_USERNAME: ${ATOM_ADMIN_USERNAME}
ATOM_ADMIN_SECRET: ${ATOM_ADMIN_SECRET}
ATOM_JWKS_URL: ${ATOM_JWKS_URL}
ATOM_JWT_ISSUER: ${ATOM_JWT_ISSUER}
ATOM_JWT_AUDIENCE: ${ATOM_JWT_AUDIENCE}
ATOM_TIMEOUT: ${ATOM_TIMEOUT}
MG_JAEGER_URL: ${MG_JAEGER_URL}
MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
networks:
- magistrala-base-net
ui:
image: ghcr.io/absmach/magistrala/ui-mg:${MG_RELEASE_TAG}
container_name: magistrala-ui
@@ -417,6 +445,7 @@ services:
MG_GROUPS_URL: ${MG_GROUPS_URL}
MG_BOOTSTRAP_URL: ${MG_BOOTSTRAP_URL}
MG_HTTP_ADAPTER_URL: ${MG_HTTP_ADAPTER_URL}
MG_PUBLISH_PROXY_URL: ${MG_PUBLISH_PROXY_URL}
MG_READER_URL: ${MG_READER_URL}
MG_BACKEND_URL: ${MG_UI_BACKEND_URL}
MG_JOURNAL_URL: ${MG_JOURNAL_URL}
docker/fluxmq/node1.yaml
+2 -13
View File
@@ -127,7 +127,8 @@ queues:
max_length_bytes: 1073741824
auth:
provider: "atom"
url: "http://fluxmq-auth:7016"
transport: "grpc"
timeout: 15s
protocols:
mqtt: true
@@ -135,15 +136,3 @@ auth:
coap: true
amqp: true
amqp091: false
identity_cache_size: 50000
identity_cache_ttl: 1h
atom:
grpc_addr: "atom:8081"
insecure: true
service_token_env: "FLUXMQ_ATOM_SERVICE_TOKEN"
service_token_file: ""
topic_format: "magistrala"
authn_cache_ttl: 30s
alias_cache_ttl: 5m
decision_cache_ttl: 0s
unsupported_topic_policy: "deny"
docker/fluxmq/node2.yaml
+2 -13
View File
@@ -124,7 +124,8 @@ queues:
max_length_bytes: 1073741824
auth:
provider: "atom"
url: "http://fluxmq-auth:7016"
transport: "grpc"
timeout: 15s
protocols:
mqtt: true
@@ -132,15 +133,3 @@ auth:
coap: true
amqp: true
amqp091: false
identity_cache_size: 50000
identity_cache_ttl: 1h
atom:
grpc_addr: "atom:8081"
insecure: true
service_token_env: "FLUXMQ_ATOM_SERVICE_TOKEN"
service_token_file: ""
topic_format: "magistrala"
authn_cache_ttl: 30s
alias_cache_ttl: 5m
decision_cache_ttl: 0s
unsupported_topic_policy: "deny"
docker/fluxmq/node3.yaml
+2 -13
View File
@@ -124,7 +124,8 @@ queues:
max_length_bytes: 1073741824
auth:
provider: "atom"
url: "http://fluxmq-auth:7016"
transport: "grpc"
timeout: 15s
protocols:
mqtt: true
@@ -132,15 +133,3 @@ auth:
coap: true
amqp: true
amqp091: false
identity_cache_size: 50000
identity_cache_ttl: 1h
atom:
grpc_addr: "atom:8081"
insecure: true
service_token_env: "FLUXMQ_ATOM_SERVICE_TOKEN"
service_token_file: ""
topic_format: "magistrala"
authn_cache_ttl: 30s
alias_cache_ttl: 5m
decision_cache_ttl: 0s
unsupported_topic_policy: "deny"
docker/nginx/entrypoint.sh
+1
View File
@@ -22,6 +22,7 @@ envsubst '
${MG_RE_HTTP_PORT}
${MG_ALARMS_HTTP_PORT}
${MG_REPORTS_HTTP_PORT}
${MG_FLUXMQ_PUBLISH_HTTP_PORT}
${MG_NGINX_AMQP_PORT}' < /etc/nginx/nginx.conf.template > /etc/nginx/nginx.conf
exec nginx -g "daemon off;"
docker/nginx/nginx-key.conf
+8
View File
@@ -57,6 +57,7 @@ http {
set $rules_upstream "re:${MG_RE_HTTP_PORT}";
set $alarms_upstream "alarms:${MG_ALARMS_HTTP_PORT}";
set $reports_upstream "reports:${MG_REPORTS_HTTP_PORT}";
set $publish_upstream "fluxmq-auth:${MG_FLUXMQ_PUBLISH_HTTP_PORT}";
include snippets/ssl.conf;
@@ -145,6 +146,13 @@ http {
proxy_pass http://$atom_upstream;
}
# Proxy user-authenticated UI publishes to FluxMQ through Atom authz.
location ~ "^/([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/channels/([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/messages$" {
include snippets/proxy-headers.conf;
add_header Access-Control-Expose-Headers Location;
proxy_pass http://$publish_upstream;
}
# Proxy pass to FluxMQ HTTP API
location /http/ {
include snippets/proxy-headers.conf;
docker/nginx/nginx-x509.conf
+9
View File
@@ -64,6 +64,7 @@ http {
set $rules_upstream "re:${MG_RE_HTTP_PORT}";
set $alarms_upstream "alarms:${MG_ALARMS_HTTP_PORT}";
set $reports_upstream "reports:${MG_REPORTS_HTTP_PORT}";
set $publish_upstream "fluxmq-auth:${MG_FLUXMQ_PUBLISH_HTTP_PORT}";
ssl_verify_client optional;
include snippets/ssl.conf;
@@ -154,6 +155,14 @@ http {
proxy_pass http://$atom_upstream;
}
# Proxy user-authenticated UI publishes to FluxMQ through Atom authz.
location ~ "^/([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/channels/([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/messages$" {
include snippets/verify-ssl-client.conf;
include snippets/proxy-headers.conf;
add_header Access-Control-Expose-Headers Location;
proxy_pass http://$publish_upstream;
}
# Proxy pass to FluxMQ HTTP API
location /http/ {
include snippets/verify-ssl-client.conf;
fluxmq/api/grpc/doc.go
+7
View File
@@ -0,0 +1,7 @@
// Copyright (c) Abstract Machines
// SPDX-License-Identifier: Apache-2.0
// Package grpc contains the FluxMQ auth callout gRPC server implementation.
// It bridges FluxMQ broker authentication and authorization requests to
// Magistrala's Clients and Channels services.
package grpc
fluxmq/api/grpc/server.go
+197
View File
@@ -0,0 +1,197 @@
// Copyright (c) Abstract Machines
// SPDX-License-Identifier: Apache-2.0
package grpc
import (
"context"
"strings"
"connectrpc.com/connect"
authv1 "github.com/absmach/fluxmq/pkg/proto/auth/v1"
"github.com/absmach/fluxmq/pkg/proto/auth/v1/authv1connect"
grpcChannelsV1 "github.com/absmach/magistrala/api/grpc/channels/v1"
grpcClientsV1 "github.com/absmach/magistrala/api/grpc/clients/v1"
apiutil "github.com/absmach/magistrala/api/http/util"
"github.com/absmach/magistrala/internal/atom"
"github.com/absmach/magistrala/pkg/authn"
"github.com/absmach/magistrala/pkg/connections"
"github.com/absmach/magistrala/pkg/errors"
svcerr "github.com/absmach/magistrala/pkg/errors/service"
"github.com/absmach/magistrala/pkg/messaging"
"github.com/absmach/magistrala/pkg/policies"
)
var _ authv1connect.AuthServiceHandler = (*connectServer)(nil)
type connectServer struct {
authv1connect.UnimplementedAuthServiceHandler
clients grpcClientsV1.ClientsServiceClient
channels grpcChannelsV1.ChannelsServiceClient
atomAuth atom.Authorizer
parser messaging.TopicParser
}
// NewServer creates a FluxMQ AuthService Connect handler that bridges to
// Magistrala's Clients (authn) and Channels (authz) services.
func NewServer(
clients grpcClientsV1.ClientsServiceClient,
channels grpcChannelsV1.ChannelsServiceClient,
parser messaging.TopicParser,
atomAuth ...atom.Authorizer,
) authv1connect.AuthServiceHandler {
var authz atom.Authorizer
if len(atomAuth) > 0 {
authz = atomAuth[0]
}
return &connectServer{
clients: clients,
channels: channels,
atomAuth: authz,
parser: parser,
}
}
func (s *connectServer) Authenticate(ctx context.Context, req *connect.Request[authv1.AuthnReq]) (*connect.Response[authv1.AuthnRes], error) {
username := req.Msg.GetUsername()
password := req.Msg.GetPassword()
token := authn.AuthPack(authn.BasicAuth, username, password)
res, err := s.clients.Authenticate(ctx, &grpcClientsV1.AuthnReq{Token: token})
if err != nil {
if !shouldTryDomainAuth(req.Msg, username, password) {
return nil, encodeError(err)
}
token = authn.AuthPack(authn.DomainAuth, username, password)
res, err = s.clients.Authenticate(ctx, &grpcClientsV1.AuthnReq{Token: token})
if err != nil {
return nil, encodeError(err)
}
}
return connect.NewResponse(&authv1.AuthnRes{
Authenticated: res.GetAuthenticated(),
Id: res.GetId(),
}), nil
}
func (s *connectServer) Authorize(ctx context.Context, req *connect.Request[authv1.AuthzReq]) (*connect.Response[authv1.AuthzRes], error) {
connType := connections.ConnType(req.Msg.GetAction())
if err := connections.CheckConnType(connType); err != nil {
return nil, encodeError(err)
}
var domainID, channelID string
var topicType messaging.TopicType
var err error
switch connType {
case connections.Publish:
domainID, channelID, _, topicType, err = s.parser.ParsePublishTopic(ctx, req.Msg.GetTopic(), true)
case connections.Subscribe:
domainID, channelID, _, topicType, err = s.parser.ParseSubscribeTopic(ctx, req.Msg.GetTopic(), true)
}
if err != nil {
if shouldDenyAuthorize(err) {
return connect.NewResponse(&authv1.AuthzRes{Authorized: false}), nil
}
return nil, encodeError(err)
}
if topicType == messaging.HealthType {
return connect.NewResponse(&authv1.AuthzRes{Authorized: true}), nil
}
if s.atomAuth != nil {
action := "subscribe"
if connType == connections.Publish {
action = "publish"
}
res, err := s.atomAuth.CheckAuthz(ctx, atom.AuthzRequest{
SubjectID: req.Msg.GetExternalId(),
Action: action,
ResourceID: channelID,
ObjectKind: "resource",
ObjectID: channelID,
Context: map[string]any{
"domain_id": domainID,
"client_type": policies.ClientType,
"connection": connType.String(),
"topic_type": uint32(topicType),
},
})
if err != nil {
if shouldDenyAuthorize(err) {
return connect.NewResponse(&authv1.AuthzRes{Authorized: false}), nil
}
return nil, encodeError(err)
}
return connect.NewResponse(&authv1.AuthzRes{Authorized: res.Allowed}), nil
}
ar := &grpcChannelsV1.AuthzReq{
Type: uint32(connType),
ClientId: req.Msg.GetExternalId(),
ClientType: policies.ClientType,
ChannelId: channelID,
DomainId: domainID,
}
res, err := s.channels.Authorize(ctx, ar)
if err != nil {
if shouldDenyAuthorize(err) {
return connect.NewResponse(&authv1.AuthzRes{Authorized: false}), nil
}
return nil, encodeError(err)
}
return connect.NewResponse(&authv1.AuthzRes{
Authorized: res.GetAuthorized(),
}), nil
}
func shouldTryDomainAuth(msg *authv1.AuthnReq, username, password string) bool {
if username == "" || password == "" {
return false
}
return strings.HasPrefix(msg.GetClientId(), "http:")
}
func shouldDenyAuthorize(err error) bool {
if err == nil {
return false
}
switch {
case errors.Contains(err, svcerr.ErrAuthorization),
errors.Contains(err, svcerr.ErrNotFound),
errors.Contains(err, errors.ErrMalformedEntity),
errors.Contains(err, messaging.ErrMalformedTopic),
err == apiutil.ErrMissingID:
return true
}
// Backward compatibility for gRPC client layers that may return
// Internal with a payload containing "entity not found".
return strings.Contains(err.Error(), svcerr.ErrNotFound.Error())
}
func encodeError(err error) error {
switch {
case errors.Contains(err, nil):
return nil
case errors.Contains(err, errors.ErrMalformedEntity),
err == apiutil.ErrMissingID:
return connect.NewError(connect.CodeInvalidArgument, err)
case errors.Contains(err, svcerr.ErrAuthentication),
strings.Contains(err.Error(), "use of expired key"):
return connect.NewError(connect.CodeUnauthenticated, err)
case errors.Contains(err, svcerr.ErrAuthorization):
return connect.NewError(connect.CodePermissionDenied, err)
case errors.Contains(err, messaging.ErrMalformedTopic):
return connect.NewError(connect.CodeInvalidArgument, err)
default:
return connect.NewError(connect.CodeInternal, err)
}
}
fluxmq/api/http/publish.go
+258
View File
@@ -0,0 +1,258 @@
// Copyright (c) Abstract Machines
// SPDX-License-Identifier: Apache-2.0
// Package http exposes user-authenticated message publishing for the MG UI.
package http
import (
"context"
"encoding/json"
"fmt"
"net/http"
"strings"
"time"
"github.com/absmach/magistrala/internal/atom"
smqauthn "github.com/absmach/magistrala/pkg/authn"
"github.com/absmach/magistrala/pkg/messaging"
"github.com/go-chi/chi/v5"
)
const (
contentType = "application/json"
httpProto = "http"
)
type publishRequest struct {
ClientID string `json:"client_id"`
Subtopic string `json:"subtopic"`
Payload json.RawMessage `json:"payload"`
}
type publishResponse struct {
Status string `json:"status"`
}
type errorResponse struct {
Error string `json:"error"`
}
type publishHandler struct {
authn smqauthn.Authentication
atom *atom.Client
publisher messaging.Publisher
}
// MakePublishHandler returns an HTTP handler that authenticates the user with
// Atom, authorizes publish access in Atom, and writes directly to the message
// broker with the selected client as the publisher identity.
func MakePublishHandler(
authn smqauthn.Authentication,
atomClient *atom.Client,
publisher messaging.Publisher,
) http.Handler {
h := publishHandler{
authn: authn,
atom: atomClient,
publisher: publisher,
}
r := chi.NewRouter()
r.Post("/{domainID}/channels/{channelID}/messages", h.publish)
r.Get("/health", func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", contentType)
w.WriteHeader(http.StatusOK)
if err := json.NewEncoder(w).Encode(publishResponse{Status: "ok"}); err != nil {
return
}
})
return r
}
func (h publishHandler) publish(w http.ResponseWriter, r *http.Request) {
domainID := chi.URLParam(r, "domainID")
channelID := chi.URLParam(r, "channelID")
if domainID == "" || channelID == "" {
writeError(w, http.StatusBadRequest, "domainID and channelID are required")
return
}
token := bearerToken(r)
if token == "" {
writeError(w, http.StatusUnauthorized, "bearer token is required")
return
}
session, err := h.authn.Authenticate(r.Context(), token)
if err != nil {
writeError(w, http.StatusUnauthorized, "invalid bearer token")
return
}
var req publishRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid publish request")
return
}
payload, err := payloadBytes(req.Payload)
if err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
publisherID := session.UserID
if req.ClientID != "" {
if err := h.ensureClientPublisher(r.Context(), domainID, channelID, session.UserID, req.ClientID); err != nil {
writeError(w, http.StatusForbidden, err.Error())
return
}
publisherID = req.ClientID
}
if err := h.ensureUserPublish(r.Context(), domainID, channelID, session.UserID, req.ClientID); err != nil {
writeError(w, http.StatusForbidden, err.Error())
return
}
subtopic := cleanSubtopic(req.Subtopic)
topic := messaging.EncodeTopicSuffix(domainID, channelID, subtopic)
msg := &messaging.Message{
Domain: domainID,
Channel: channelID,
Subtopic: subtopic,
Publisher: publisherID,
ClientId: session.UserID,
Protocol: httpProto,
Payload: payload,
Created: time.Now().UnixNano(),
}
if err := h.publisher.Publish(r.Context(), topic, msg); err != nil {
writeError(w, http.StatusBadGateway, "failed to publish message")
return
}
w.Header().Set("Content-Type", contentType)
w.WriteHeader(http.StatusAccepted)
if err := json.NewEncoder(w).Encode(publishResponse{Status: "accepted"}); err != nil {
return
}
}
func (h publishHandler) ensureUserPublish(
ctx context.Context,
domainID string,
channelID string,
userID string,
clientID string,
) error {
res, err := h.atom.CheckAuthz(ctx, atom.AuthzRequest{
SubjectID: userID,
Action: "publish",
ResourceID: channelID,
ObjectKind: "resource",
ObjectID: channelID,
Context: map[string]any{
"domain_id": domainID,
"publisher_client_id": clientID,
},
})
if err != nil {
return err
}
if !res.Allowed {
return fmt.Errorf("user is not allowed to publish to channel")
}
return nil
}
func (h publishHandler) ensureClientPublisher(
ctx context.Context,
domainID string,
channelID string,
userID string,
clientID string,
) error {
client, err := h.atom.GetEntity(ctx, clientID)
if err != nil {
return fmt.Errorf("publisher client not found")
}
if client.Kind != "device" && attrString(client.Attributes, "magistrala_kind") != atom.KindClient {
return fmt.Errorf("publisher identity is not a client")
}
if client.TenantID == "" || client.TenantID != domainID {
return fmt.Errorf("publisher client belongs to a different domain")
}
userAccess, err := h.atom.CheckAuthz(ctx, atom.AuthzRequest{
SubjectID: userID,
Action: "read",
ResourceID: clientID,
ObjectKind: "entity",
ObjectID: clientID,
Context: map[string]any{
"domain_id": domainID,
},
})
if err != nil {
return err
}
if !userAccess.Allowed {
return fmt.Errorf("user is not allowed to use publisher client")
}
res, err := h.atom.CheckAuthz(ctx, atom.AuthzRequest{
SubjectID: clientID,
Action: "publish",
ResourceID: channelID,
ObjectKind: "resource",
ObjectID: channelID,
Context: map[string]any{
"domain_id": domainID,
},
})
if err != nil {
return err
}
if !res.Allowed {
return fmt.Errorf("publisher client is not connected for publish")
}
return nil
}
func bearerToken(r *http.Request) string {
token := r.Header.Get("Authorization")
return strings.TrimPrefix(token, "Bearer ")
}
func payloadBytes(raw json.RawMessage) ([]byte, error) {
if len(raw) == 0 {
return nil, fmt.Errorf("payload is required")
}
if raw[0] == '"' {
var value string
if err := json.Unmarshal(raw, &value); err != nil {
return nil, fmt.Errorf("payload must be a string or JSON value")
}
return []byte(value), nil
}
return raw, nil
}
func cleanSubtopic(subtopic string) string {
return strings.Trim(strings.ReplaceAll(subtopic, ".", "/"), "/")
}
func attrString(attrs atom.Attributes, key string) string {
value, ok := attrs[key]
if !ok || value == nil {
return ""
}
if str, ok := value.(string); ok {
return str
}
return fmt.Sprint(value)
}
func writeError(w http.ResponseWriter, status int, message string) {
w.Header().Set("Content-Type", contentType)
w.WriteHeader(status)
if err := json.NewEncoder(w).Encode(errorResponse{Error: message}); err != nil {
return
}
}
go.mod
+3 -1
View File
@@ -3,6 +3,8 @@ module github.com/absmach/magistrala
go 1.26.4
require (
connectrpc.com/connect v1.20.0
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.30.0
@@ -53,6 +55,7 @@ require (
go.opentelemetry.io/otel/sdk v1.44.0
go.opentelemetry.io/otel/trace v1.44.0
golang.org/x/crypto v0.53.0
golang.org/x/net v0.56.0
golang.org/x/sync v0.21.0
gonum.org/v1/gonum v0.17.0
google.golang.org/grpc v1.81.1
@@ -165,7 +168,6 @@ require (
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/net v0.56.0 // indirect
golang.org/x/sys v0.46.0 // indirect
golang.org/x/text v0.38.0 // indirect
golang.org/x/time v0.15.0 // indirect
go.sum
+4
View File
@@ -1,5 +1,9 @@
al.essio.dev/pkg/shellescape v1.5.1/go.mod h1:6sIqp7X2P6mThCQ7twERpZTuigpr6KbZWtls1U8I890=
cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ=
connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
connectrpc.com/otelconnect v0.9.0 h1:NggB3pzRC3pukQWaYbRHJulxuXvmCKCKkQ9hbrHAWoA=
connectrpc.com/otelconnect v0.9.0/go.mod h1:AEkVLjCPXra+ObGFCOClcJkNjS7zPaQSqvO0lCyjfZc=
dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8=
dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
internal/atom/service_tokens.go
+1
View File
@@ -36,6 +36,7 @@ type TokenProvisionResult struct {
func DefaultServiceTokenSpecs() []ServiceTokenSpec {
return []ServiceTokenSpec{
{Name: "fluxmq-auth", Env: "MG_ATOM_TOKEN_FLUXMQ_AUTH", Description: "Magistrala Docker Compose token for fluxmq-auth"},
{Name: "fluxmq-node1", Env: "MG_ATOM_TOKEN_FLUXMQ_NODE1", Description: "Magistrala Docker Compose token for fluxmq-node1"},
{Name: "fluxmq-node2", Env: "MG_ATOM_TOKEN_FLUXMQ_NODE2", Description: "Magistrala Docker Compose token for fluxmq-node2"},
{Name: "fluxmq-node3", Env: "MG_ATOM_TOKEN_FLUXMQ_NODE3", Description: "Magistrala Docker Compose token for fluxmq-node3"},