feat: use mgate to proxy coap with dtls

Signed-off-by: Felix Gateru <[email protected]>

Author: felixgateru

cmd/coap/main.go

@@ -8,14 +8,19 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"log/slog"
 	"net/url"
 	"os"
 
 	chclient "github.com/absmach/callhome/pkg/client"
+	"github.com/absmach/mgate"
+	mgatecoap "github.com/absmach/mgate/pkg/coap"
+	"github.com/absmach/mgate/pkg/session"
+	mgtls "github.com/absmach/mgate/pkg/tls"
 	"github.com/absmach/supermq"
 	"github.com/absmach/supermq/coap"
 	httpapi "github.com/absmach/supermq/coap/api"
-	"github.com/absmach/supermq/coap/tracing"
+	"github.com/absmach/supermq/coap/middleware"
 	smqlog "github.com/absmach/supermq/logger"
 	domainsAuthz "github.com/absmach/supermq/pkg/domains/grpcclient"
 	"github.com/absmach/supermq/pkg/grpcclient"
@@ -30,19 +35,24 @@ import (
 	httpserver "github.com/absmach/supermq/pkg/server/http"
 	"github.com/absmach/supermq/pkg/uuid"
 	"github.com/caarlos0/env/v11"
+	"github.com/pion/dtls/v3"
 	"golang.org/x/sync/errgroup"
 )
 
 const (
-	svcName           = "coap_adapter"
-	envPrefix         = "SMQ_COAP_ADAPTER_"
-	envPrefixHTTP     = "SMQ_COAP_ADAPTER_HTTP_"
-	envPrefixCache    = "SMQ_COAP_CACHE_"
-	envPrefixClients  = "SMQ_CLIENTS_GRPC_"
-	envPrefixChannels = "SMQ_CHANNELS_GRPC_"
-	envPrefixDomains  = "SMQ_DOMAINS_GRPC_"
-	defSvcHTTPPort    = "5683"
-	defSvcCoAPPort    = "5683"
+	svcName            = "coap_adapter"
+	envPrefix          = "SMQ_COAP_ADAPTER_"
+	envPrefixHTTP      = "SMQ_COAP_ADAPTER_HTTP_"
+	envPrefixCache     = "SMQ_COAP_CACHE_"
+	envPrefixClients   = "SMQ_CLIENTS_GRPC_"
+	envPrefixChannels  = "SMQ_CHANNELS_GRPC_"
+	envPrefixDomains   = "SMQ_DOMAINS_GRPC_"
+	defSvcHTTPPort     = "5683"
+	defSvcCoAPPort     = "5683"
+	targetProtocol     = "coap"
+	targetCoapHost     = "localhost"
+	targetCoapPort     = "5683"
+	targetCoapDtlsPort = "5684"
 )
 
 type config struct {
@@ -94,6 +104,13 @@ func main() {
 		return
 	}
 
+	dtlsCfg, err := mgtls.NewConfig(env.Options{Prefix: envPrefix})
+	if err != nil {
+		logger.Error(fmt.Sprintf("failed to load %s DTLS configuration : %s", svcName, err))
+		exitCode = 1
+		return
+	}
+
 	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))
@@ -181,12 +198,12 @@ func main() {
 
 	svc := coap.New(clientsClient, channelsClient, nps)
 
-	svc = tracing.New(tracer, svc)
+	svc = middleware.TracingMiddleware(tracer, svc)
 
-	svc = httpapi.LoggingMiddleware(svc, logger)
+	svc = middleware.LoggingMiddleware(svc, logger)
 
 	counter, latency := prometheus.MakeMetrics(svcName, "api")
-	svc = httpapi.MetricsMiddleware(svc, counter, latency)
+	svc = middleware.MetricsMiddleware(svc, counter, latency)
 
 	hs := httpserver.NewServer(ctx, cancel, svcName, httpServerConfig, httpapi.MakeHandler(cfg.InstanceID), logger)
 
@@ -207,7 +224,8 @@ func main() {
 		return hs.Start()
 	})
 	g.Go(func() error {
-		return cs.Start()
+		handler := coap.NewHandler(nps, logger, clientsClient, channelsClient, parser)
+		return proxyCoAP(ctx, coapServerConfig, dtlsCfg, handler, logger)
 	})
 	g.Go(func() error {
 		return server.StopSignalHandler(ctx, cancel, logger, svcName, hs, cs)
@@ -217,3 +235,43 @@ func main() {
 		logger.Error(fmt.Sprintf("CoAP adapter service terminated: %s", err))
 	}
 }
+
+func proxyCoAP(ctx context.Context, cfg server.Config, dtlsCfg mgtls.Config, handler session.Handler, logger *slog.Logger) error {
+	var err error
+	config := mgate.Config{
+		Host:           cfg.Host,
+		Port:           cfg.Port,
+		TargetProtocol: targetProtocol,
+		TargetHost:     targetCoapHost,
+		TargetPort:     targetCoapPort,
+	}
+
+	mg := mgatecoap.NewProxy(config, handler, logger)
+
+	errCh := make(chan error)
+
+	logger.Info(fmt.Sprintf("Starting COAP without DTLS proxy on port %s", cfg.Port))
+	go func() {
+		errCh <- mg.Listen(ctx)
+	}()
+	config.DTLSConfig, err = mgtls.LoadTLSConfig(&dtlsCfg, &dtls.Config{})
+	if err != nil {
+		return err
+	}
+
+	if config.DTLSConfig != nil {
+		config.Port = targetCoapDtlsPort
+		mgDtls := mgatecoap.NewProxy(config, handler, logger)
+		logger.Info(fmt.Sprintf("Starting COAP with DTLS proxy on port %s", cfg.Port))
+		go func() {
+			errCh <- mgDtls.Listen(ctx)
+		}()
+	}
+	select {
+	case <-ctx.Done():
+		logger.Info(fmt.Sprintf("proxy COAP shutdown at %s:%s", config.Host, config.Port))
+		return nil
+	case err := <-errCh:
+		return err
+	}
+}

coap/adapter.go

@@ -23,15 +23,11 @@ var errFailedToDisconnectClient = errors.New("failed to disconnect client")
 
 // Service specifies CoAP service API.
 type Service interface {
-	// Publish publishes message to specified channel.
-	// Key is used to authorize publisher.
-	Publish(ctx context.Context, key string, msg *messaging.Message) error
-
 	// Subscribes to channel with specified id, domainID, subtopic and adds subscription to
 	// service map of subscriptions under given ID.
 	Subscribe(ctx context.Context, key, domainID, chanID, subtopic string, c Client) error
 
-	// Unsubscribe method is used to stop observing resource.
+	// Unsubscribe methdod is used to stop observing resource.
 	Unsubscribe(ctx context.Context, key, domainID, chanID, subptopic, token string) error
 
 	// DisconnectHandler method is used to disconnected the client
@@ -58,36 +54,6 @@ func New(clients grpcClientsV1.ClientsServiceClient, channels grpcChannelsV1.Cha
 	return as
 }
 
-func (svc *adapterService) Publish(ctx context.Context, key string, msg *messaging.Message) error {
-	authnRes, err := svc.clients.Authenticate(ctx, &grpcClientsV1.AuthnReq{
-		Token: authn.AuthPack(authn.DomainAuth, msg.GetDomain(), key),
-	})
-	if err != nil {
-		return errors.Wrap(svcerr.ErrAuthentication, err)
-	}
-	if !authnRes.Authenticated {
-		return svcerr.ErrAuthentication
-	}
-
-	authzRes, err := svc.channels.Authorize(ctx, &grpcChannelsV1.AuthzReq{
-		DomainId:   msg.GetDomain(),
-		ClientId:   authnRes.GetId(),
-		ClientType: policies.ClientType,
-		Type:       uint32(connections.Publish),
-		ChannelId:  msg.GetChannel(),
-	})
-	if err != nil {
-		return errors.Wrap(svcerr.ErrAuthorization, err)
-	}
-	if !authzRes.Authorized {
-		return svcerr.ErrAuthorization
-	}
-
-	msg.Publisher = authnRes.GetId()
-
-	return svc.pubsub.Publish(ctx, messaging.EncodeMessageTopic(msg), msg)
-}
-
 func (svc *adapterService) Subscribe(ctx context.Context, key, domainID, chanID, subtopic string, c Client) error {
 	authnRes, err := svc.clients.Authenticate(ctx, &grpcClientsV1.AuthnReq{
 		Token: authn.AuthPack(authn.DomainAuth, domainID, key),
@@ -100,19 +66,6 @@ func (svc *adapterService) Subscribe(ctx context.Context, key, domainID, chanID,
 	}
 
 	clientID := authnRes.GetId()
-	authzRes, err := svc.channels.Authorize(ctx, &grpcChannelsV1.AuthzReq{
-		DomainId:   domainID,
-		ClientId:   clientID,
-		ClientType: policies.ClientType,
-		Type:       uint32(connections.Subscribe),
-		ChannelId:  chanID,
-	})
-	if err != nil {
-		return errors.Wrap(svcerr.ErrAuthorization, err)
-	}
-	if !authzRes.Authorized {
-		return svcerr.ErrAuthorization
-	}
 
 	subject := messaging.EncodeTopic(domainID, chanID, subtopic)
 	authzc := newAuthzClient(clientID, domainID, chanID, subtopic, svc.channels, c)

coap/api/transport.go

@@ -104,7 +104,7 @@ func (h *CoAPHandler) ServeCOAP(w mux.ResponseWriter, m *mux.Message) {
 		err = h.handleGet(m, w, msg, key)
 	case codes.POST:
 		resp.SetCode(codes.Created)
-		err = h.service.Publish(m.Context(), key, msg)
+		err = nil
 	default:
 		err = errMethodNotAllowed
 	}