Merge pull request #4412 from kobergj/AuthenticationForNats

Allow authentication for nats connections

changelog/unreleased/nats-authentication.md

@@ -0,0 +1,5 @@
+Enhancement: Allow authentication for nats connections
+
+Allows configuring username/password for nats connections
+
+https://github.com/cs3org/reva/pull/4412

go.sum

@@ -616,8 +616,6 @@ github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
 github.com/go-logr/logr v1.2.4/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
-github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20230807070816-bc05fb076ce7 h1:/RpJVLKmKT2OcEnKCPaS6n+zygNzYDzwoYgPQEgcEiQ=
-github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20230807070816-bc05fb076ce7/go.mod h1:lYuiEYKQTpbE2LA8HEcC8D6kQ29M7ILfEak3dzeucEg=
 github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20231215124540-f7f8d3274bf9 h1:YOIavj+ZgO9HzukpdXZCvQv+AahjW/fTVFVF4QFRabw=
 github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20231215124540-f7f8d3274bf9/go.mod h1:cL0O63th39fZ+M/aRJvajz7Qnmv+UTXugOq1k3qrYiQ=
 github.com/go-micro/plugins/v4/registry/consul v1.2.1 h1:3wctYMtstwQLCjoJ1HA6mKGGFF1hcdKDv5MzHakB1jE=

internal/grpc/interceptors/eventsmiddleware/events.go

@@ -37,6 +37,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/rgrpc"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
 	"github.com/cs3org/reva/v2/pkg/utils"
+	"github.com/mitchellh/mapstructure"
 )
 
 const (
@@ -223,17 +224,11 @@ func publisherFromConfig(m map[string]interface{}) (events.Publisher, error) {
 	default:
 		return nil, fmt.Errorf("stream type '%s' not supported", typ)
 	case "nats":
-		var tlsCert string
-		val, ok := m["tls-root-ca-cert"]
-		if ok {
-			tlsCert = val.(string)
+		var cfg stream.NatsConfig
+		if err := mapstructure.Decode(m, &cfg); err != nil {
+			return nil, err
 		}
-		return stream.NatsFromConfig(m["name"].(string), false, stream.NatsConfig{
-			Endpoint:             m["address"].(string),
-			Cluster:              m["clusterID"].(string),
-			EnableTLS:            m["enable-tls"].(bool),
-			TLSInsecure:          m["tls-insecure"].(bool),
-			TLSRootCACertificate: tlsCert,
-		})
+		name, _ := m["name"].(string)
+		return stream.NatsFromConfig(name, false, cfg)
 	}
 }

internal/grpc/services/storageprovider/storageprovider.go

@@ -76,6 +76,8 @@ type eventconfig struct {
 	TLSInsecure          bool   `mapstructure:"tls_insecure"  docs:"Whether to verify the server TLS certificates."`
 	TLSRootCACertificate string `mapstructure:"tls_root_ca_cert"  docs:"The root CA certificate used to validate the server's TLS certificate."`
 	EnableTLS            bool   `mapstructure:"nats_enable_tls" docs:"events tls switch"`
+	AuthUsername         string `mapstructure:"nats_username" docs:"event stream username"`
+	AuthPassword         string `mapstructure:"nats_password" docs:"event stream password"`
 }
 
 func (c *config) init() {

internal/http/services/dataprovider/dataprovider.go

@@ -48,6 +48,8 @@ type config struct {
 	NatsTLSInsecure    bool                              `mapstructure:"nats_tls_insecure"`
 	NatsRootCACertPath string                            `mapstructure:"nats_root_ca_cert_path"`
 	NatsEnableTLS      bool                              `mapstructure:"nats_enable_tls"`
+	NatsUsername       string                            `mapstructure:"nats_username"`
+	NatsPassword       string                            `mapstructure:"nats_password"`
 }
 
 func (c *config) init() {
@@ -86,6 +88,8 @@ func New(m map[string]interface{}, log *zerolog.Logger) (global.Service, error)
 			EnableTLS:            conf.NatsEnableTLS,
 			TLSInsecure:          conf.NatsTLSInsecure,
 			TLSRootCACertificate: conf.NatsRootCACertPath,
+			AuthUsername:         conf.NatsUsername,
+			AuthPassword:         conf.NatsPassword,
 		})
 		if err != nil {
 			return nil, err

pkg/events/stream/nats.go

@@ -17,11 +17,14 @@ import (
 
 // NatsConfig is the configuration needed for a NATS event stream
 type NatsConfig struct {
-	Endpoint             string // Endpoint of the nats server
-	Cluster              string // CluserID of the nats cluster
-	TLSInsecure          bool   // Whether to verify TLS certificates
-	TLSRootCACertificate string // The root CA certificate used to validate the TLS certificate
-	EnableTLS            bool   // Enable TLS
+	Endpoint             string `mapstructure:"address"`          // Endpoint of the nats server
+	Cluster              string `mapstructure:"clusterID"`        // CluserID of the nats cluster
+	TLSInsecure          bool   `mapstructure:"tls-insecure"`     // Whether to verify TLS certificates
+	TLSRootCACertificate string `mapstructure:"tls-root-ca-cert"` // The root CA certificate used to validate the TLS certificate
+	EnableTLS            bool   `mapstructure:"enable-tls"`       // Enable TLS
+	AuthUsername         string `mapstructure:"username"`         // Username for authentication
+	AuthPassword         string `mapstructure:"password"`         // Password for authentication
+
 }
 
 // NatsFromConfig returns a nats stream from the given config
@@ -55,14 +58,14 @@ func NatsFromConfig(connName string, disableDurability bool, cfg NatsConfig) (ev
 		natsjs.ClusterID(cfg.Cluster),
 		natsjs.SynchronousPublish(true),
 		natsjs.Name(connName),
+		natsjs.Authenticate(cfg.AuthUsername, cfg.AuthPassword),
 	}
 
 	if disableDurability {
 		opts = append(opts, natsjs.DisableDurableStreams())
 	}
 
 	return Nats(opts...)
-
 }
 
 // nats returns a nats streaming client

pkg/share/manager/jsoncs3/jsoncs3.go

@@ -133,6 +133,8 @@ type EventOptions struct {
 	TLSInsecure          bool   `mapstructure:"tlsinsecure"`
 	TLSRootCACertificate string `mapstructure:"tlsrootcacertificate"`
 	EnableTLS            bool   `mapstructure:"enabletls"`
+	AuthUsername         string `mapstructure:"authusername"`
+	AuthPassword         string `mapstructure:"authpassword"`
 }
 
 // Manager implements a share manager using a cs3 storage backend with local caching

pkg/storage/cache/cache.go

@@ -51,6 +51,8 @@ type Config struct {
 	TTL                time.Duration `mapstructure:"cache_ttl"`
 	Size               int           `mapstructure:"cache_size"`
 	DisablePersistence bool          `mapstructure:"cache_disable_persistence"`
+	AuthUsername       string        `mapstructure:"cache_auth_username"`
+	AuthPassword       string        `mapstructure:"cache_auth_password"`
 }
 
 // Cache handles key value operations on caches
@@ -240,5 +242,6 @@ func getStore(cfg Config) microstore.Store {
 		store.TTL(cfg.TTL),
 		store.Size(cfg.Size),
 		store.DisablePersistence(cfg.DisablePersistence),
+		store.Authentication(cfg.AuthUsername, cfg.AuthPassword),
 	)
 }

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -145,6 +145,7 @@ func NewDefault(m map[string]interface{}, bs tree.Blobstore, es events.Stream) (
 		microstore.Database(o.IDCache.Database),
 		microstore.Table(o.IDCache.Table),
 		store.DisablePersistence(o.IDCache.DisablePersistence),
+		store.Authentication(o.IDCache.AuthUsername, o.IDCache.AuthPassword),
 	))
 
 	permissionsSelector, err := pool.PermissionsSelector(o.PermissionsSVC, pool.WithTLSMode(o.PermTLSMode))

pkg/store/options.go

@@ -89,3 +89,17 @@ func DisablePersistence(val bool) store.Option {
 		o.Context = context.WithValue(o.Context, disablePersistanceContextKey{}, val)
 	}
 }
+
+type authenticationContextKey struct{}
+
+// Authentication configures the username and password to use for authentication.
+// Only supported by the `natsjskv` implementation.
+func Authentication(username, password string) store.Option {
+	return func(o *store.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+
+		o.Context = context.WithValue(o.Context, authenticationContextKey{}, []string{username, password})
+	}
+}

pkg/store/store.go

@@ -127,6 +127,10 @@ func Create(opts ...microstore.Option) microstore.Store {
 		// host, port, clusterid
 		natsOptions := nats.GetDefaultOptions()
 		natsOptions.Name = "TODO" // we can pass in the service name to allow identifying the client, but that requires adding a custom context option
+		if auth, ok := options.Context.Value(authenticationContextKey{}).([]string); ok && len(auth) == 2 {
+			natsOptions.User = auth[0]
+			natsOptions.Password = auth[1]
+		}
 		return natsjs.NewStore(
 			append(opts,
 				natsjs.NatsOptions(natsOptions), // always pass in properly initialized default nats options
@@ -141,6 +145,10 @@ func Create(opts ...microstore.Option) microstore.Store {
 
 		natsOptions := nats.GetDefaultOptions()
 		natsOptions.Name = "TODO" // we can pass in the service name to allow identifying the client, but that requires adding a custom context option
+		if auth, ok := options.Context.Value(authenticationContextKey{}).([]string); ok && len(auth) == 2 {
+			natsOptions.User = auth[0]
+			natsOptions.Password = auth[1]
+		}
 		return natsjskv.NewStore(
 			append(opts,
 				natsjs.NatsOptions(natsOptions), // always pass in properly initialized default nats options