feat: Add injection of Secure MessageBus creds for eKuiper connections

close #3767

Signed-off-by: Leonard Goodell <[email protected]>

cmd/security-secretstore-setup/res/configuration.toml

@@ -86,3 +86,4 @@ ConfigJWTDuration = "1h"
 [SecureMessageBus]
 Type = "none" # blank or none if MessageBus not secured, "redis" if secured. "mqtt" is TBD
 KuiperConfigPath = "/tmp/kuiper/edgex.yaml"
+KuiperConnectionsPath = "/tmp/kuiper-connections/connection.yaml"

go.sum

@@ -55,6 +55,7 @@ github.com/edgexfoundry/go-mod-bootstrap/v2 v2.0.1-dev.14/go.mod h1:84xs+nDgmAu8
 github.com/edgexfoundry/go-mod-configuration/v2 v2.0.1-dev.5 h1:icE1aVlX7I3SJ0qPqZJchCr2JLe2TMRZlUMIM2qoivo=
 github.com/edgexfoundry/go-mod-configuration/v2 v2.0.1-dev.5/go.mod h1:MvHit0MxBXN4bC8LL0NZRsw72ByRE1XwtVLQP9C+2vg=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.0/go.mod h1:pfXURRetgIto0GR0sCjDrfa71hqJ1wxmQWi/mOzWfWU=
+github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.1-dev.23/go.mod h1:I6UhBPCREubcU0ouIGBdZlNG5Xx4NijUVN5rvEtD03k=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.1-dev.25 h1:AFQD5sbxpAfwESF/SXApyq7piSDgoioWLL5D3GY8qvw=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.1-dev.25/go.mod h1:I6UhBPCREubcU0ouIGBdZlNG5Xx4NijUVN5rvEtD03k=
 github.com/edgexfoundry/go-mod-messaging/v2 v2.0.1 h1:8nT3CiPLIft5RmR+vbmXBW9Kbz7TqPZ6C8QuQ6TTn6w=

internal/security/secretstore/config/config.go

@@ -37,6 +37,7 @@ type Database struct {
 type SecureMessageBusInfo struct {
 	Type             string
 	KuiperConfigPath string
+	KuiperConnectionsPath string
 }
 
 type SecretStoreInfo struct {

internal/security/secretstore/secure-messagebus.go

@@ -20,12 +20,13 @@ import (
 	"os"
 	"text/template"
 
-	"github.com/edgexfoundry/edgex-go/internal/security/secretstore/config"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/clients/logger"
+
+	"github.com/edgexfoundry/edgex-go/internal/security/secretstore/config"
 )
 
 const (
-	kuiperConfigTemplate = `
+	eKuiperEdgeXSourceTemplate = `
 application_conf:
   port: 5571
   protocol: tcp
@@ -50,6 +51,18 @@ mqtt_conf:
   topic: events
   type: mqtt
 `
+
+eKuiperConnectionsTemplate = `
+edgex:
+  redisMsgBus: #connection key
+    protocol: redis
+    server: localhost
+    port: 6379
+    type: redis
+    optional:
+      Username: {{.User}}
+      Password: {{.Password}}
+`
 	// Can't use constants from go-mod-messaging since that will create ZMQ dependency, which we do not want!
 	redisSecureMessageBusType = "redis"
 	mqttSecureMessageBusType  = "mqtt"
@@ -59,36 +72,43 @@ mqtt_conf:
 
 func ConfigureSecureMessageBus(secureMessageBus config.SecureMessageBusInfo, redis5Pair UserPasswordPair, lc logger.LoggingClient) error {
 	switch secureMessageBus.Type {
-	// Currently only support Secure MessageBus when using the Redis implementation
+	// Currently, only support Secure MessageBus when using the Redis implementation.
 	case redisSecureMessageBusType:
-		err := configureKuiperForSecureMessageBus(redis5Pair, secureMessageBus.KuiperConfigPath, lc)
+		// eKuiper now has two configuration files (EdgeX Sources and Connections)
+
+		err := configureKuiperForSecureMessageBus(redis5Pair, "EdgeX Source", eKuiperEdgeXSourceTemplate, secureMessageBus.KuiperConfigPath, lc)
+		if err != nil {
+			return err
+		}
+
+		err = configureKuiperForSecureMessageBus(redis5Pair, "Connections", eKuiperConnectionsTemplate, secureMessageBus.KuiperConnectionsPath, lc)
 		if err != nil {
 			return err
 		}
 
 	// TODO: Add support for secure MQTT MessageBus
 	case mqttSecureMessageBusType:
-		return fmt.Errorf("Secure MQTT MessageBus not yet supported")
+		return fmt.Errorf("secure MQTT MessageBus not yet supported")
 
 	case noneSecureMessageBusType, blankSecureMessageBusType:
 		return nil
 
 	default:
-		return fmt.Errorf("Invalid Secure MessageBus Type of '%s'", secureMessageBus.Type)
+		return fmt.Errorf("invalid Secure MessageBus Type of '%s'", secureMessageBus.Type)
 	}
 
 	return nil
 }
 
-func configureKuiperForSecureMessageBus(credentials UserPasswordPair, configPath string, lc logger.LoggingClient) error {
-	tmpl, err := template.New("kuiper").Parse(kuiperConfigTemplate)
+func configureKuiperForSecureMessageBus(credentials UserPasswordPair, fileType string, fileTemplate string, path string, lc logger.LoggingClient) error  {
+	tmpl, err := template.New("eKuiper").Parse(fileTemplate)
 	if err != nil {
-		return fmt.Errorf("failed to parse Kuiper Edgex config template: %w", err)
+		return fmt.Errorf("failed to parse eKuiper %s template: %w", fileType, err)
 	}
 
-	file, err := os.OpenFile(configPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
+	file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
 	if err != nil {
-		return fmt.Errorf("failed to open/create Kuiper Edgex config file %s: %w", configPath, err)
+		return fmt.Errorf("failed to open/create eKuiper %s file %s: %w", fileType, path, err)
 	}
 
 	defer func() {
@@ -97,10 +117,10 @@ func configureKuiperForSecureMessageBus(credentials UserPasswordPair, configPath
 
 	err = tmpl.Execute(file, credentials)
 	if err != nil {
-		return fmt.Errorf("failed to write Kuiper Edgex config file %s: %w", configPath, err)
+		return fmt.Errorf("failed to write eKuiper  %s file %s: %w", fileType, path, err)
 	}
 
-	lc.Infof("Wrote Kuiper config at %s with secure MessageBus credentials", configPath)
+	lc.Infof("Wrote eKuiper %s at %s with Secure MessageBus credentials", fileType, path)
 
 	return nil
 }

internal/security/secretstore/secure-messagebus_test.go

@@ -20,15 +20,17 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/edgexfoundry/edgex-go/internal/security/secretstore/config"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/clients/logger"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/edgexfoundry/edgex-go/internal/security/secretstore/config"
 )
 
 func TestConfigureSecureMessageBus(t *testing.T) {
 	secureMessageBus := config.SecureMessageBusInfo{
 		KuiperConfigPath: "./testdata/edgex.yaml",
+		KuiperConnectionsPath: "./testdata/connection.yaml",
 	}
 
 	validExpected := UserPasswordPair{
@@ -52,8 +54,11 @@ func TestConfigureSecureMessageBus(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.Name, func(t *testing.T) {
 			_ = os.Remove(secureMessageBus.KuiperConfigPath)
+			_ = os.Remove(secureMessageBus.KuiperConnectionsPath)
+
 			defer func() {
 				_ = os.Remove(secureMessageBus.KuiperConfigPath)
+				_ = os.Remove(secureMessageBus.KuiperConnectionsPath)
 			}()
 
 			secureMessageBus.Type = test.Type
@@ -66,20 +71,28 @@ func TestConfigureSecureMessageBus(t *testing.T) {
 			require.NoError(t, err)
 
 			if test.Expected == nil {
-				// Config file should not have been written
+				// Source Config file should not have been written
 				_, err = os.Stat(secureMessageBus.KuiperConfigPath)
 				require.True(t, os.IsNotExist(err))
+
+				// Connections file should not have been written
+				_, err = os.Stat(secureMessageBus.KuiperConnectionsPath)
+				require.True(t, os.IsNotExist(err))
+
 				return
 			}
 
-			// Config file should have been written
+			// Source Config file should have been written
 			contents, err := os.ReadFile(secureMessageBus.KuiperConfigPath)
 			require.NoError(t, err)
 			assert.True(t, strings.Contains(string(contents), test.Expected.User))
 			assert.True(t, strings.Contains(string(contents), test.Expected.Password))
-			err = os.Remove(secureMessageBus.KuiperConfigPath)
-			require.NoError(t, err)
 
+			// Connections file should have been written
+			contents, err = os.ReadFile(secureMessageBus.KuiperConnectionsPath)
+			require.NoError(t, err)
+			assert.True(t, strings.Contains(string(contents), test.Expected.User))
+			assert.True(t, strings.Contains(string(contents), test.Expected.Password))
 		})
 	}
 }

snap/snapcraft.yaml

@@ -208,6 +208,7 @@ apps:
       # enable secure message bus setup for kuiper
       SECUREMESSAGEBUS_TYPE: "redis"
       SECUREMESSAGEBUS_KUIPERCONFIGPATH: "$SNAP_DATA/kuiper/etc/sources/edgex.yaml"
+      SECUREMESSAGEBUS_KUIPERCONNECTIONSPATH: "$SNAP_DATA/kuiper/etc/connections/connection.yaml"
 
     start-timeout: 15m
     plugs: [network]