Add Flag for Custom Authenticators in Cassandra Storage (#5628)

<!--
!! Please DELETE this comment before posting.
We appreciate your contribution to the Jaeger project! 👋🎉
-->

## Which problem is this PR solving?
- #5627

## Description of the changes
- added defaultApprovedAuthenticators

## How was this change tested?
- 

## Checklist
- [ ] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [ ] I have signed all commits
- [ ] I have added unit tests for the new functionality
- [ ] I have run lint and test steps successfully
  - for `jaeger`: `make lint test`
  - for `jaeger-ui`: `yarn lint` and `yarn test`

---------

Signed-off-by: mehul gautam <mehulsharma4786@gmail.com>

pkg/cassandra/config/config.go

@@ -56,8 +56,9 @@ type Authenticator struct {
 
 // BasicAuthenticator holds the username and password for a password authenticator for a Cassandra cluster
 type BasicAuthenticator struct {
-	Username string `yaml:"username" mapstructure:"username"`
-	Password string `yaml:"password" mapstructure:"password" json:"-"`
+	Username              string   `yaml:"username" mapstructure:"username"`
+	Password              string   `yaml:"password" mapstructure:"password" json:"-"`
+	AllowedAuthenticators []string `yaml:"allowed_authenticators" mapstructure:"allowed_authenticators"`
 }
 
 // ApplyDefaults copies settings from source unless its own value is non-zero.
@@ -143,8 +144,9 @@ func (c *Configuration) NewCluster(logger *zap.Logger) (*gocql.ClusterConfig, er
 
 	if c.Authenticator.Basic.Username != "" && c.Authenticator.Basic.Password != "" {
 		cluster.Authenticator = gocql.PasswordAuthenticator{
-			Username: c.Authenticator.Basic.Username,
-			Password: c.Authenticator.Basic.Password,
+			Username:              c.Authenticator.Basic.Username,
+			Password:              c.Authenticator.Basic.Password,
+			AllowedAuthenticators: c.Authenticator.Basic.AllowedAuthenticators,
 		}
 	}
 	tlsCfg, err := c.TLS.Config(logger)

plugin/storage/cassandra/options.go

@@ -45,7 +45,7 @@ const (
 	suffixSocketKeepAlive    = ".socket-keep-alive"
 	suffixUsername           = ".username"
 	suffixPassword           = ".password"
-
+	suffixAuth               = ".basic.allowed-authenticators"
 	// common storage settings
 	suffixSpanStoreWriteCacheTTL = ".span-store-write-cache-ttl"
 	suffixIndexTagsBlacklist     = ".index.tag-blacklist"
@@ -214,6 +214,13 @@ func addFlags(flagSet *flag.FlagSet, nsConfig namespaceConfig) {
 		nsConfig.namespace+suffixPassword,
 		nsConfig.Authenticator.Basic.Password,
 		"Password for password authentication for Cassandra")
+	flagSet.String(
+		nsConfig.namespace+suffixAuth,
+		"",
+		"The comma-separated list of allowed password authenticators for Cassandra."+
+			"If none are specified, there is a default 'approved' list that is used "+
+			"(https://github.com/gocql/gocql/blob/34fdeebefcbf183ed7f916f931aa0586fdaa1b40/conn.go#L27). "+
+			"If a non-empty list is provided, only specified authenticators are allowed.")
 }
 
 // InitFromViper initializes Options with properties from viper
@@ -256,6 +263,8 @@ func (cfg *namespaceConfig) initFromViper(v *viper.Viper) {
 	cfg.SocketKeepAlive = v.GetDuration(cfg.namespace + suffixSocketKeepAlive)
 	cfg.Authenticator.Basic.Username = v.GetString(cfg.namespace + suffixUsername)
 	cfg.Authenticator.Basic.Password = v.GetString(cfg.namespace + suffixPassword)
+	authentication := stripWhiteSpace(v.GetString(cfg.namespace + suffixAuth))
+	cfg.Authenticator.Basic.AllowedAuthenticators = strings.Split(authentication, ",")
 	cfg.DisableCompression = v.GetBool(cfg.namespace + suffixDisableCompression)
 	var err error
 	cfg.TLS, err = tlsFlagsConfig.InitFromViper(v)

plugin/storage/cassandra/options_test.go

@@ -64,17 +64,24 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--cas.index.tag-whitelist=flerg, flarg,florg ",
 		"--cas.index.tags=true",
 		"--cas.index.process-tags=false",
+		"--cas.basic.allowed-authenticators=org.apache.cassandra.auth.PasswordAuthenticator,com.datastax.bdp.cassandra.auth.DseAuthenticator",
+		"--cas.username=username",
+		"--cas.password=password",
 		// enable aux with a couple overrides
 		"--cas-aux.enabled=true",
 		"--cas-aux.keyspace=jaeger-archive",
 		"--cas-aux.servers=3.3.3.3, 4.4.4.4",
+		"--cas-aux.username=username",
+		"--cas-aux.password=password",
+		"--cas-aux.basic.allowed-authenticators=org.apache.cassandra.auth.PasswordAuthenticator,com.ericsson.bss.cassandra.ecaudit.auth.AuditAuthenticator",
 	})
 	opts.InitFromViper(v)
 
 	primary := opts.GetPrimary()
 	assert.Equal(t, "jaeger", primary.Keyspace)
 	assert.Equal(t, "mojave", primary.LocalDC)
 	assert.Equal(t, []string{"1.1.1.1", "2.2.2.2"}, primary.Servers)
+	assert.Equal(t, []string{"org.apache.cassandra.auth.PasswordAuthenticator", "com.datastax.bdp.cassandra.auth.DseAuthenticator"}, primary.Authenticator.Basic.AllowedAuthenticators)
 	assert.Equal(t, "ONE", primary.Consistency)
 	assert.Equal(t, []string{"blerg", "blarg", "blorg"}, opts.TagIndexBlacklist())
 	assert.Equal(t, []string{"flerg", "flarg", "florg"}, opts.TagIndexWhitelist())
@@ -86,6 +93,7 @@ func TestOptionsWithFlags(t *testing.T) {
 	require.NotNil(t, aux)
 	assert.Equal(t, "jaeger-archive", aux.Keyspace)
 	assert.Equal(t, []string{"3.3.3.3", "4.4.4.4"}, aux.Servers)
+	assert.Equal(t, []string{"org.apache.cassandra.auth.PasswordAuthenticator", "com.ericsson.bss.cassandra.ecaudit.auth.AuditAuthenticator"}, aux.Authenticator.Basic.AllowedAuthenticators)
 	assert.Equal(t, 42, aux.ConnectionsPerHost)
 	assert.Equal(t, 42, aux.MaxRetryAttempts)
 	assert.Equal(t, 42*time.Second, aux.Timeout)

plugin/storage/integration/cassandra_test.go

@@ -62,6 +62,9 @@ func (*CassandraStorageIntegration) initializeCassandraFactory(t *testing.T, fla
 
 func (s *CassandraStorageIntegration) initializeCassandra(t *testing.T) {
 	f := s.initializeCassandraFactory(t, []string{
+		"--cassandra.basic.allowed-authenticators=",
+		"--cassandra.password=password",
+		"--cassandra.username=username",
 		"--cassandra.keyspace=jaeger_v1_dc1",
 		"--cassandra-archive.keyspace=jaeger_v1_dc1_archive",
 		"--cassandra-archive.enabled=true",