Implement new bearer authentication scheme for SSO support

* Implement new bearer authentication scheme for SSO support

* Refactor expired auth error handling

* Wrap SSO token expiration error into user-facing error

Author: fbiville

neo4j/auth_tokens.go

@@ -28,6 +28,7 @@ const keyScheme = "scheme"
 const schemeNone = "none"
 const schemeBasic = "basic"
 const schemeKerberos = "kerberos"
+const schemeBearer = "bearer"
 const keyPrincipal = "principal"
 const keyCredentials = "credentials"
 const keyRealm = "realm"
@@ -67,6 +68,18 @@ func KerberosAuth(ticket string) AuthToken {
 	return token
 }
 
+// BearerAuth generates an authentication token with the provided base-64 value generated by a Single Sign-On provider
+func BearerAuth(token string) AuthToken {
+	result := AuthToken{
+		tokens: map[string]interface{}{
+			keyScheme: schemeBearer,
+			keyCredentials: token,
+		},
+	}
+
+	return result
+}
+
 // CustomAuth generates a custom authentication token with provided parameters
 func CustomAuth(scheme string, username string, password string, realm string, parameters map[string]interface{}) AuthToken {
 	tokens := map[string]interface{}{

neo4j/error.go

@@ -110,6 +110,17 @@ func IsTransactionExecutionLimit(err error) bool {
 	return is
 }
 
+// TokenExpiredError represent errors caused by the driver not being able to connect to Neo4j services,
+// or lost connections.
+type TokenExpiredError struct {
+	Code string
+	Message string
+}
+
+func (e *TokenExpiredError) Error() string {
+	return fmt.Sprintf("TokenExpiredError: %s (%s)", e.Code, e.Message)
+}
+
 func wrapError(err error) error {
 	if err == nil {
 		return nil
@@ -134,6 +145,10 @@ func wrapError(err error) error {
 			// most likely due to read timeout configuration hint being applied
 			return &ConnectivityError{inner: err}
 		}
+	case *db.Neo4jError:
+		if e.Code == "Neo.ClientError.Security.TokenExpired" {
+			return &TokenExpiredError{Code: e.Code, Message: e.Msg}
+		}
 	}
 	return err
 }

neo4j/internal/bolt/bolt4.go

@@ -157,10 +157,8 @@ func (b *bolt4) setError(err error, fatal bool) {
 		b.state = bolt4_failed
 	}
 
-	neo4jErr, _ := err.(*db.Neo4jError)
 	// Increase severity even if it was a previous error
-	// Treat expired auth as fatal so that pool is cleaned up of old connections
-	if fatal || (neo4jErr != nil && neo4jErr.Code == "Status.Security.AuthorizationExpired") {
+	if fatal {
 		b.state = bolt4_dead
 	}
 
@@ -171,7 +169,8 @@ func (b *bolt4) setError(err error, fatal bool) {
 	}
 
 	// Do not log big cypher statements as errors
-	if neo4jErr != nil && neo4jErr.Classification() == "ClientError" {
+	neo4jErr, casted := err.(*db.Neo4jError)
+	if casted && neo4jErr.Classification() == "ClientError" {
 		b.log.Debugf(log.Bolt4, b.logId, "%s", err)
 	} else {
 		b.log.Error(log.Bolt4, b.logId, err)
@@ -205,7 +204,7 @@ func (b *bolt4) receiveSuccess() *success {
 		}
 		return v
 	case *db.Neo4jError:
-		b.setError(v, false)
+		b.setError(v, isFatalError(v))
 		return nil
 	default:
 		// Unexpected message received
@@ -805,7 +804,7 @@ func (b *bolt4) receiveNext() (*db.Record, bool, *db.Summary) {
 		b.checkStreams()
 		return nil, false, sum
 	case *db.Neo4jError:
-		b.setError(x, false) // Will detach the stream
+		b.setError(x, isFatalError(x)) // Will detach the stream
 		return nil, false, nil
 	default:
 		// Unknown territory
@@ -983,3 +982,8 @@ func (b *bolt4) initializeReadTimeoutHint(hints map[string]interface{}) {
 	}
 	b.in.connReadTimeout = time.Duration(readTimeout) * time.Second
 }
+
+func isFatalError(err *db.Neo4jError) bool {
+	// Treat expired auth as fatal so that pool is cleaned up of old connections
+	return err != nil && err.Code == "Status.Security.AuthorizationExpired"
+}

neo4j/internal/bolt/bolt4_test.go

@@ -871,4 +871,29 @@ func TestBolt4(ot *testing.T) {
 		assertBoltState(t, bolt4_dead, bolt)
 		AssertError(t, err)
 	})
+
+	ot.Run("Immediately expired authentication token error triggers a connection failure", func(t *testing.T) {
+		bolt, cleanup := connectToServer(t, func(srv *bolt4server) {
+			srv.accept(4)
+			srv.sendFailureMsg("Neo.ClientError.Security.TokenExpired", "SSO token is... expired")
+		})
+		defer cleanup()
+
+		_, err := bolt.Run(db.Command{Cypher: "MATCH (n) RETURN n"}, db.TxConfig{Mode: db.ReadMode})
+		assertBoltState(t, bolt4_failed, bolt)
+		AssertError(t, err)
+	})
+
+	ot.Run("Expired authentication token error after run triggers a connection failure", func(t *testing.T) {
+		bolt, cleanup := connectToServer(t, func(srv *bolt4server) {
+			srv.accept(4)
+			srv.waitForRun()
+			srv.sendFailureMsg("Neo.ClientError.Security.TokenExpired", "SSO token is... expired")
+		})
+		defer cleanup()
+
+		_, err := bolt.Run(db.Command{Cypher: "MATCH (n) RETURN n"}, db.TxConfig{Mode: db.ReadMode})
+		assertBoltState(t, bolt4_failed, bolt)
+		AssertError(t, err)
+	})
 }

neo4j/internal/testutil/asserts.go

@@ -21,6 +21,7 @@
 package testutil
 
 import (
+	"fmt"
 	"io"
 	"reflect"
 	"strings"
@@ -81,10 +82,15 @@ func AssertNoError(t *testing.T, err error) {
 	}
 }
 
+func AssertErrorMessageContains(t *testing.T, err error, msg string, args ...interface{}) {
+	AssertError(t, err)
+	AssertStringContain(t, err.Error(), fmt.Sprintf(msg, args...))
+}
+
 func AssertError(t *testing.T, err error) {
 	t.Helper()
 	if err == nil {
-		t.Fatal("Expected an error but it wasn't")
+		t.Fatal("Expected an error but none occurred")
 	}
 }
 

neo4j/internal/testutil/connfake.go

@@ -48,8 +48,11 @@ type ConnFake struct {
 	Table          *db.RoutingTable
 	Err            error
 	Id             int
+	TxBeginErr     error
 	TxBeginHandle  db.TxHandle
+	RunErr         error
 	RunStream      db.StreamHandle
+	RunTxErr       error
 	RunTxStream    db.StreamHandle
 	Nexts          []Next
 	Bookm          string
@@ -112,7 +115,7 @@ func (c *ConnFake) GetRoutingTable(context map[string]string, bookmarks []string
 
 func (c *ConnFake) TxBegin(txConfig db.TxConfig) (db.TxHandle, error) {
 	c.RecordedTxs = append(c.RecordedTxs, RecordedTx{Origin: "TxBegin", Mode: txConfig.Mode, Bookmarks: txConfig.Bookmarks, Timeout: txConfig.Timeout, Meta: txConfig.Meta})
-	return c.TxBeginHandle, c.Err
+	return c.TxBeginHandle, c.TxBeginErr
 }
 
 func (c *ConnFake) TxRollback(tx db.TxHandle) error {
@@ -129,11 +132,11 @@ func (c *ConnFake) TxCommit(tx db.TxHandle) error {
 func (c *ConnFake) Run(runCommand db.Command, txConfig db.TxConfig) (db.StreamHandle, error) {
 
 	c.RecordedTxs = append(c.RecordedTxs, RecordedTx{Origin: "Run", Mode: txConfig.Mode, Bookmarks: txConfig.Bookmarks, Timeout: txConfig.Timeout, Meta: txConfig.Meta})
-	return c.RunStream, c.Err
+	return c.RunStream, c.RunErr
 }
 
 func (c *ConnFake) RunTx(tx db.TxHandle, runCommand db.Command) (db.StreamHandle, error) {
-	return c.RunTxStream, c.Err
+	return c.RunTxStream, c.RunTxErr
 }
 
 func (c *ConnFake) Keys(streamHandle db.StreamHandle) ([]string, error) {

neo4j/session.go

@@ -208,7 +208,7 @@ func (s *session) BeginTransaction(configurers ...func(*TransactionConfig)) (Tra
 	})
 	if err != nil {
 		s.pool.Return(conn)
-		return nil, err
+		return nil, wrapError(err)
 	}
 
 	// Create transaction wrapper