Cep 50

auth.go

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package gocql
+
+import (
+	"fmt"
+	"sync"
+)
+
+// AuthRegistry is the interface for a registry of authenticators to support authentication negotiation between the client and the server (CEP-50).
+type AuthRegistry interface {
+	// Returns the list of all registered authenticators.
+	Authenticators() []NegotiableAuthenticator
+
+	// Registers a new authenticator.
+	//
+	// authenticator is the NegotiableAuthenticator to register.
+	Register(authenticator NegotiableAuthenticator)
+
+	// Returns the authenticator for the given class name.
+	// If the authenticator is not found, returns false.
+	AuthenticatorFor(className string) (NegotiableAuthenticator, bool)
+}
+
+// The default implementation of the AuthRegistry interface.
+type defaultAuthRegistry struct {
+	mu sync.RWMutex
+	// Map of Java class name to authenticator.
+	byClassName map[string]NegotiableAuthenticator
+	// List of all registered authenticators. Used for iteration.
+	all []NegotiableAuthenticator
+}
+
+// Creates a new default implementation of the AuthRegistry interface.
+func NewDefaultAuthRegistry() AuthRegistry {
+	return &defaultAuthRegistry{
+		byClassName: make(map[string]NegotiableAuthenticator),
+		all:         make([]NegotiableAuthenticator, 0),
+	}
+}
+
+func (r *defaultAuthRegistry) Register(authenticator NegotiableAuthenticator) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.mustRegisterLocked(authenticator)
+}
+
+// Registers the authenticator with the given class name.
+// Panics if the authenticator is already registered.
+// Must be called with the lock held.
+func (r *defaultAuthRegistry) mustRegisterLocked(authenticator NegotiableAuthenticator) {
+	className := authenticator.ClassName()
+	if _, ok := r.byClassName[className]; ok {
+		panic(fmt.Sprintf("gocql: authenticator %s already registered", className))
+	}
+	r.byClassName[className] = authenticator
+	r.all = append(r.all, authenticator)
+}
+
+// Implements the AuthRegistry interface. Returns a copy of the list of all registered authenticators.
+func (r *defaultAuthRegistry) Authenticators() []NegotiableAuthenticator {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return append([]NegotiableAuthenticator(nil), r.all...)
+}
+
+// Returns the authenticator for the given class name.
+// If the authenticator is not found, returns false.
+func (r *defaultAuthRegistry) AuthenticatorFor(className string) (NegotiableAuthenticator, bool) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	authenticator, ok := r.byClassName[className]
+	return authenticator, ok
+}

cluster.go

@@ -165,6 +165,10 @@ type ClusterConfig struct {
 	// Default: nil
 	AuthProvider func(h *HostInfo) (Authenticator, error)
 
+	// AuthRegistry is the registry of authenticators to support authentication negotiation between the client and the server (CEP-50).
+	// Default: nil
+	AuthRegistry AuthRegistry
+
 	// Default retry policy to use for queries.
 	// Default: no retries.
 	RetryPolicy RetryPolicy

conn.go

@@ -74,6 +74,23 @@ type Authenticator interface {
 	Success(data []byte) error
 }
 
+// NegotiableAuthenticator is an authenticator that can be negotiated with the server.
+// It is an optional interface that can be implemented by the authenticator to allow for negotiation with the server.
+// If the authenticator implements this interface, it will be used to negotiate the authenticator with the server.
+type NegotiableAuthenticator interface {
+	Authenticator
+
+	// Java class name of the authenticator. Used to match the authenticator with the one chosen by the server for authentication.
+	//
+	// Example: "org.apache.cassandra.auth.PasswordAuthenticator"
+	ClassName() string
+
+	// Authentication mode of the authenticator. Used as a payload in STARTUP frame.
+	//
+	// Example: "Unauthenticated", "Password", "MutualTLS"
+	AuthenticationMode() string
+}
+
 // PasswordAuthenticator specifies credentials to be used when authenticating.
 // It can be configured with an "allow list" of authenticator class names to avoid
 // attempting to authenticate with Cassandra if it doesn't provide an expected authenticator.
@@ -102,6 +119,14 @@ func (p PasswordAuthenticator) Success(data []byte) error {
 	return nil
 }
 
+func (p PasswordAuthenticator) ClassName() string {
+	return "org.apache.cassandra.auth.PasswordAuthenticator"
+}
+
+func (p PasswordAuthenticator) AuthenticationMode() string {
+	return "Password"
+}
+
 // SslOptions configures TLS use.
 //
 // Warning: Due to historical reasons, the SslOptions is insecure by default, so you need to set EnableHostVerification
@@ -482,6 +507,19 @@ func (s *startupCoordinator) startup(ctx context.Context, supported map[string][
 		}
 	}
 
+	// If the server supports authentication negotiation and the client has a registry of authenticators, then we should negotiate authentication.
+	_, supportsNegotiation := supported["AUTHENTICATORS"]
+	shouldNegotiateAuthentication := supportsNegotiation && s.conn.session.cfg.AuthRegistry != nil
+	if shouldNegotiateAuthentication {
+		authenticators := s.conn.session.cfg.AuthRegistry.Authenticators()
+		authClasses := make([]string, len(authenticators))
+		for i, authenticator := range authenticators {
+			authClasses[i] = authenticator.AuthenticationMode()
+		}
+		// The value should be a comma-separated list of Java class names according to the CEP-50 specification.
+		m["AUTHENTICATORS"] = strings.Join(authClasses, ",")
+	}
+
 	frame, err := s.write(ctx, &writeStartupFrame{opts: m}, startupCompleted)
 	if err != nil {
 		return err
@@ -497,14 +535,23 @@ func (s *startupCoordinator) startup(ctx context.Context, supported map[string][
 	case *authenticateFrame:
 		// Startup is successfully completed, so we could use Native Protocol 5
 		startupCompleted.Store(true)
-		return s.authenticateHandshake(ctx, v, startupCompleted)
+		return s.authenticateHandshake(ctx, v, startupCompleted, shouldNegotiateAuthentication)
 	default:
 		return NewErrProtocol("Unknown type of response to startup frame: %s", v)
 	}
 }
 
-func (s *startupCoordinator) authenticateHandshake(ctx context.Context, authFrame *authenticateFrame, startupCompleted *atomic.Bool) error {
-	if s.conn.auth == nil {
+func (s *startupCoordinator) authenticateHandshake(ctx context.Context, authFrame *authenticateFrame, startupCompleted *atomic.Bool, shouldNegotiateAuthentication bool) error {
+	if shouldNegotiateAuthentication {
+		auth, ok := s.conn.session.cfg.AuthRegistry.AuthenticatorFor(authFrame.class)
+		// It should never happen, but we should handle it gracefully.
+		if !ok {
+			return fmt.Errorf("the server requested an unknown authenticator during authentication negotiation: %q", authFrame.class)
+		}
+		// Set the authenticator to use for the authentication handshake.
+		s.conn.auth = auth
+		s.conn.logger.Debug("Authenticator selected for authentication negotiation", NewLogFieldString("authenticator", auth.ClassName()))
+	} else if s.conn.auth == nil {
 		return fmt.Errorf("authentication required (using %q)", authFrame.class)
 	}
 

integration_test.go

@@ -38,6 +38,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	inf "gopkg.in/inf.v0"
 )
 
@@ -68,6 +69,44 @@ func TestAuthentication(t *testing.T) {
 	session.Close()
 }
 
+// Implements the NegotiableAuthenticator interface.
+// TODO: Should it be part of the gocql package?
+type AllowAllAuthenticator struct{}
+
+func (p AllowAllAuthenticator) Challenge(req []byte) ([]byte, Authenticator, error) {
+	return nil, nil, nil
+}
+
+func (p AllowAllAuthenticator) Success(data []byte) error {
+	return nil
+}
+
+func (p AllowAllAuthenticator) AuthenticationMode() string {
+	return "Unauthenticated"
+}
+
+func (p AllowAllAuthenticator) ClassName() string {
+	return "org.apache.cassandra.auth.AllowAllAuthenticator"
+}
+
+func TestAuthenticationNegotiation(t *testing.T) {
+	if !*flagRunAuthTest {
+		t.Skip("Authentication is not configured in the target cluster")
+	}
+
+	cluster := createCluster()
+	cluster.AuthRegistry = NewDefaultAuthRegistry()
+	cluster.AuthRegistry.Register(PasswordAuthenticator{
+		Username: "cassandra",
+		Password: "cassandra",
+	})
+	cluster.AuthRegistry.Register(AllowAllAuthenticator{})
+
+	session, err := cluster.CreateSession()
+	require.NoError(t, err)
+	session.Close()
+}
+
 func TestGetHosts(t *testing.T) {
 	clusterHosts := getClusterHosts()
 	cluster := createCluster()

session.go

@@ -141,6 +141,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 	}
 
 	// Check that either Authenticator is set or AuthProvider, not both
+	// TODO: add AuthRegistry check
 	if cfg.Authenticator != nil && cfg.AuthProvider != nil {
 		return nil, errors.New("Can't use both Authenticator and AuthProvider in cluster config.")
 	}