MG-2126 - Update messaging with mProxy TLS

cmd/auth/main.go

@@ -35,7 +35,7 @@ import (
 	v1 "github.com/authzed/authzed-go/proto/authzed/api/v1"
 	"github.com/authzed/authzed-go/v1"
 	"github.com/authzed/grpcutil"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"

cmd/bootstrap/main.go

@@ -32,7 +32,7 @@ import (
 	"github.com/absmach/magistrala/pkg/events/store"
 	mgsdk "github.com/absmach/magistrala/pkg/sdk/go"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"

cmd/cassandra-reader/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/absmach/magistrala/readers"
 	"github.com/absmach/magistrala/readers/api"
 	"github.com/absmach/magistrala/readers/cassandra"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/gocql/gocql"
 	"golang.org/x/sync/errgroup"
 )

cmd/cassandra-writer/main.go

@@ -27,7 +27,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/gocql/gocql"
 	"golang.org/x/sync/errgroup"
 )

cmd/certs/main.go

@@ -29,7 +29,7 @@ import (
 	"github.com/absmach/magistrala/pkg/auth"
 	mgsdk "github.com/absmach/magistrala/pkg/sdk/go"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"

cmd/coap/main.go

@@ -26,7 +26,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"golang.org/x/sync/errgroup"
 )
 

cmd/http/main.go

@@ -6,7 +6,6 @@ package main
 
 import (
 	"context"
-	"crypto/tls"
 	"fmt"
 	"log"
 	"log/slog"
@@ -32,7 +31,7 @@ import (
 	"github.com/absmach/mproxy"
 	mproxyhttp "github.com/absmach/mproxy/pkg/http"
 	"github.com/absmach/mproxy/pkg/session"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
 )
@@ -163,43 +162,35 @@ func newService(pub messaging.Publisher, tc magistrala.AuthzServiceClient, logge
 }
 
 func proxyHTTP(ctx context.Context, cfg server.Config, logger *slog.Logger, sessionHandler session.Handler) error {
-	config := mproxy.Config{
-		Address:    fmt.Sprintf("%s:%s", "", cfg.Port),
-		Target:     fmt.Sprintf("%s:%s", targetHTTPHost, targetHTTPPort),
-		PathPrefix: "/",
-	}
-	if cfg.CertFile != "" || cfg.KeyFile != "" {
-		tlsCert, err := tls.LoadX509KeyPair(cfg.CertFile, cfg.KeyFile)
-		if err != nil {
-			return err
-		}
-		config.TLSConfig = &tls.Config{
-			Certificates: []tls.Certificate{tlsCert},
-		}
+	httpConfig, err := mproxy.NewConfig(env.Options{Prefix: envPrefix})
+	if err != nil {
+		return err
+	}
+	if httpConfig.Address == "" {
+		httpConfig.Address = fmt.Sprintf("%s:%s", "", cfg.Port)
 	}
-	mp, err := mproxyhttp.NewProxy(config, sessionHandler, logger)
+	if httpConfig.Target == "" {
+		httpConfig.Target = fmt.Sprintf("%s:%s", targetHTTPHost, targetHTTPPort)
+	}
+
+	mp, err := mproxyhttp.NewProxy(httpConfig, sessionHandler, logger)
 	if err != nil {
 		return err
 	}
 	http.HandleFunc("/", mp.ServeHTTP)
 
 	errCh := make(chan error)
-	switch {
-	case cfg.CertFile != "" || cfg.KeyFile != "":
-		go func() {
-			errCh <- mp.Listen(ctx)
-		}()
+	go func() {
+		errCh <- mp.Listen(ctx)
+	}()
+	if httpConfig.TLSConfig != nil {
 		logger.Info(fmt.Sprintf("%s service https server listening at %s:%s with TLS cert %s and key %s", svcName, cfg.Host, cfg.Port, cfg.CertFile, cfg.KeyFile))
-	default:
-		go func() {
-			errCh <- mp.Listen(ctx)
-		}()
+	} else {
 		logger.Info(fmt.Sprintf("%s service http server listening at %s:%s without TLS", svcName, cfg.Host, cfg.Port))
 	}
-
 	select {
 	case <-ctx.Done():
-		logger.Info(fmt.Sprintf("proxy HTTP shutdown at %s", config.Target))
+		logger.Info(fmt.Sprintf("proxy HTTP shutdown at %s", httpConfig.Target))
 		return nil
 	case err := <-errCh:
 		return err

cmd/influxdb-reader/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/absmach/magistrala/readers"
 	"github.com/absmach/magistrala/readers/api"
 	"github.com/absmach/magistrala/readers/influxdb"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	"golang.org/x/sync/errgroup"
 )

cmd/influxdb-writer/main.go

@@ -26,7 +26,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"golang.org/x/sync/errgroup"
 )
 

cmd/invitations/main.go

@@ -28,7 +28,7 @@ import (
 	"github.com/absmach/magistrala/pkg/auth"
 	mgsdk "github.com/absmach/magistrala/pkg/sdk/go"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"

cmd/lora/main.go

@@ -31,7 +31,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	mqttpaho "github.com/eclipse/paho.mqtt.golang"
 	"github.com/go-redis/redis/v8"
 	"golang.org/x/sync/errgroup"

cmd/mongodb-reader/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/absmach/magistrala/readers"
 	"github.com/absmach/magistrala/readers/api"
 	"github.com/absmach/magistrala/readers/mongodb"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"go.mongodb.org/mongo-driver/mongo"
 	"golang.org/x/sync/errgroup"
 )

cmd/mongodb-writer/main.go

@@ -27,7 +27,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"go.mongodb.org/mongo-driver/mongo"
 	"golang.org/x/sync/errgroup"
 )

cmd/mqtt/main.go

@@ -36,14 +36,16 @@ import (
 	mproxymqtt "github.com/absmach/mproxy/pkg/mqtt"
 	"github.com/absmach/mproxy/pkg/mqtt/websocket"
 	"github.com/absmach/mproxy/pkg/session"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/cenkalti/backoff/v4"
 	"golang.org/x/sync/errgroup"
 )
 
 const (
 	svcName        = "mqtt"
 	envPrefixAuthz = "MG_THINGS_AUTH_GRPC_"
+	envPrefixMQTT  = "MG_MQTT_ADAPTER_"
+	envPrefixWS    = "MG_MQTT_ADAPTER_WS_"
 )
 
 type config struct {
@@ -57,7 +59,6 @@ type config struct {
 	HTTPPort              string        `env:"MG_MQTT_ADAPTER_WS_PORT"                      envDefault:"8080"`
 	HTTPTargetHost        string        `env:"MG_MQTT_ADAPTER_WS_TARGET_HOST"               envDefault:"localhost"`
 	HTTPTargetPort        string        `env:"MG_MQTT_ADAPTER_WS_TARGET_PORT"               envDefault:"8080"`
-	HTTPTargetPath        string        `env:"MG_MQTT_ADAPTER_WS_TARGET_PATH"               envDefault:"/mqtt"`
 	Instance              string        `env:"MG_MQTT_ADAPTER_INSTANCE"                     envDefault:""`
 	JaegerURL             url.URL       `env:"MG_JAEGER_URL"                                envDefault:"http://localhost:14268/api/traces"`
 	BrokerURL             string        `env:"MG_MESSAGE_BROKER_URL"                        envDefault:"nats://localhost:4222"`
@@ -75,7 +76,6 @@ func main() {
 	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())
@@ -188,16 +188,16 @@ func main() {
 		chc := chclient.New(svcName, magistrala.Version, logger, cancel)
 		go chc.CallHome(ctx)
 	}
-
 	var interceptor session.Interceptor
+
 	logger.Info(fmt.Sprintf("Starting MQTT proxy on port %s", cfg.MQTTPort))
 	g.Go(func() error {
-		return proxyMQTT(ctx, cfg, logger, h, interceptor)
+		return proxyMQTT(ctx, logger, h, interceptor)
 	})
 
 	logger.Info(fmt.Sprintf("Starting MQTT over WS  proxy on port %s", cfg.HTTPPort))
 	g.Go(func() error {
-		return proxyWS(ctx, cfg, logger, h, interceptor)
+		return proxyWS(ctx, logger, h, interceptor)
 	})
 
 	g.Go(func() error {
@@ -209,36 +209,36 @@ func main() {
 	}
 }
 
-func proxyMQTT(ctx context.Context, cfg config, logger *slog.Logger, sessionHandler session.Handler, interceptor session.Interceptor) error {
-	config := mproxy.Config{
-		Address: fmt.Sprintf(":%s", cfg.MQTTPort),
-		Target:  fmt.Sprintf("%s:%s", cfg.MQTTTargetHost, cfg.MQTTTargetPort),
+func proxyMQTT(ctx context.Context, logger *slog.Logger, sessionHandler session.Handler, interceptor session.Interceptor) error {
+	mqttConfig, err := mproxy.NewConfig(env.Options{Prefix: envPrefixMQTT})
+	if err != nil {
+		return (err)
 	}
-	mproxy := mproxymqtt.New(config, sessionHandler, interceptor, logger)
+
+	mqttProxy := mproxymqtt.New(mqttConfig, sessionHandler, interceptor, logger)
 
 	errCh := make(chan error)
 	go func() {
-		errCh <- mproxy.Listen(ctx)
+		errCh <- mqttProxy.Listen(ctx)
 	}()
 
 	select {
 	case <-ctx.Done():
-		logger.Info(fmt.Sprintf("proxy MQTT shutdown at %s", config.Target))
+		logger.Info(fmt.Sprintf("proxy MQTT shutdown at %s", mqttConfig.Target))
 		return nil
 	case err := <-errCh:
 		return err
 	}
 }
 
-func proxyWS(ctx context.Context, cfg config, logger *slog.Logger, sessionHandler session.Handler, interceptor session.Interceptor) error {
-	config := mproxy.Config{
-		Address:    fmt.Sprintf("%s:%s", "", cfg.HTTPPort),
-		Target:     fmt.Sprintf("%s:%s", cfg.HTTPTargetHost, cfg.HTTPTargetPort),
-		PathPrefix: "/mqtt",
+func proxyWS(ctx context.Context, logger *slog.Logger, sessionHandler session.Handler, interceptor session.Interceptor) error {
+	wsConfig, err := mproxy.NewConfig(env.Options{Prefix: envPrefixWS})
+	if err != nil {
+		return (err)
 	}
 
-	wp := websocket.New(config, sessionHandler, interceptor, logger)
-	http.HandleFunc("/mqtt", wp.ServeHTTP)
+	wp := websocket.New(wsConfig, sessionHandler, interceptor, logger)
+	http.Handle("/mqtt", wp)
 
 	errCh := make(chan error)
 
@@ -248,7 +248,7 @@ func proxyWS(ctx context.Context, cfg config, logger *slog.Logger, sessionHandle
 
 	select {
 	case <-ctx.Done():
-		logger.Info(fmt.Sprintf("proxy MQTT WS shutdown at %s", config.Target))
+		logger.Info(fmt.Sprintf("proxy MQTT WS shutdown at %s", wsConfig.Target))
 		return nil
 	case err := <-errCh:
 		return err

cmd/opcua/main.go

@@ -30,7 +30,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/go-redis/redis/v8"
 	"golang.org/x/sync/errgroup"
 )

cmd/postgres-reader/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/absmach/magistrala/readers"
 	"github.com/absmach/magistrala/readers/api"
 	"github.com/absmach/magistrala/readers/postgres"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"golang.org/x/sync/errgroup"
 )

cmd/postgres-writer/main.go

@@ -27,7 +27,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"golang.org/x/sync/errgroup"
 )

cmd/provision/main.go

@@ -24,7 +24,7 @@ import (
 	"github.com/absmach/magistrala/pkg/uuid"
 	"github.com/absmach/magistrala/provision"
 	"github.com/absmach/magistrala/provision/api"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"golang.org/x/sync/errgroup"
 )
 

cmd/smpp-notifier/main.go

@@ -31,7 +31,7 @@ import (
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/ulid"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"

cmd/smtp-notifier/main.go

@@ -32,7 +32,7 @@ import (
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/ulid"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"

cmd/things/main.go

@@ -41,7 +41,7 @@ import (
 	thingspg "github.com/absmach/magistrala/things/postgres"
 	localusers "github.com/absmach/magistrala/things/standalone"
 	ctracing "github.com/absmach/magistrala/things/tracing"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/go-chi/chi/v5"
 	"github.com/go-redis/redis/v8"
 	"github.com/jmoiron/sqlx"

cmd/timescale-reader/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/absmach/magistrala/readers"
 	"github.com/absmach/magistrala/readers/api"
 	"github.com/absmach/magistrala/readers/timescale"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"golang.org/x/sync/errgroup"
 )

cmd/timescale-writer/main.go

@@ -27,7 +27,7 @@ import (
 	"github.com/absmach/magistrala/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
 	"github.com/absmach/magistrala/pkg/uuid"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/jmoiron/sqlx"
 	"golang.org/x/sync/errgroup"
 )

cmd/twins/main.go

@@ -33,7 +33,7 @@ import (
 	"github.com/absmach/magistrala/twins/events"
 	twmongodb "github.com/absmach/magistrala/twins/mongodb"
 	"github.com/absmach/magistrala/twins/tracing"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/go-redis/redis/v8"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.opentelemetry.io/otel/trace"

cmd/users/main.go

@@ -44,7 +44,7 @@ import (
 	"github.com/absmach/magistrala/users/hasher"
 	clientspg "github.com/absmach/magistrala/users/postgres"
 	ctracing "github.com/absmach/magistrala/users/tracing"
-	"github.com/caarlos0/env/v10"
+	"github.com/caarlos0/env/v11"
 	"github.com/go-chi/chi/v5"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"