Merge branch '4.4' into optional-realm-in-testkit-auth-token

Author: robsdedude

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"
@@ -70,6 +71,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

@@ -74,25 +74,26 @@ func (i *internalTx4) toMeta() map[string]interface{} {
 }
 
 type bolt4 struct {
-	state                 int
-	txId                  db.TxHandle
-	streams               openstreams
-	conn                  net.Conn
-	serverName            string
-	out                   outgoing
-	in                    incoming
-	connId                string
-	logId                 string
-	serverVersion         string
-	tfirst                int64       // Time that server started streaming
-	pendingTx             internalTx4 // Stashed away when tx started explicitly
-	hasPendingTx          bool
-	bookmark              string // Last bookmark
-	birthDate             time.Time
-	log                   log.Logger
-	databaseName          string
-	err                   error // Last fatal error
-	minor                 int
+	state         int
+	txId          db.TxHandle
+	streams       openstreams
+	conn          net.Conn
+	serverName    string
+	out           outgoing
+	in            incoming
+	connId        string
+	logId         string
+	serverVersion string
+	tfirst        int64       // Time that server started streaming
+	pendingTx     internalTx4 // Stashed away when tx started explicitly
+	hasPendingTx  bool
+	bookmark      string // Last bookmark
+	birthDate     time.Time
+	log           log.Logger
+	databaseName  string
+	err           error // Last fatal error
+	minor         int
+	lastQid       int64 // Last seen qid
 }
 
 func NewBolt4(serverName string, conn net.Conn, log log.Logger, boltLog log.BoltLogger) *bolt4 {
@@ -156,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
 	}
 
@@ -170,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)
@@ -199,9 +199,12 @@ func (b *bolt4) receiveSuccess() *success {
 
 	switch v := msg.(type) {
 	case *success:
+		if v.qid > -1 {
+			b.lastQid = v.qid
+		}
 		return v
 	case *db.Neo4jError:
-		b.setError(v, false)
+		b.setError(v, isFatalError(v))
 		return nil
 	default:
 		// Unexpected message received
@@ -433,7 +436,7 @@ func (b *bolt4) discardStream() {
 			// already sent a discard.
 			discarded = true
 			stream.fetchSize = -1
-			if b.state == bolt4_streamingtx {
+			if b.state == bolt4_streamingtx && stream.qid != b.lastQid {
 				b.out.appendDiscardNQid(stream.fetchSize, stream.qid)
 			} else {
 				b.out.appendDiscardN(stream.fetchSize)
@@ -464,7 +467,12 @@ func (b *bolt4) sendPullN() {
 		b.out.appendPullN(b.streams.curr.fetchSize)
 		b.out.send(b.conn)
 	} else if b.state == bolt4_streamingtx {
-		b.out.appendPullNQid(b.streams.curr.fetchSize, b.streams.curr.qid)
+		fetchSize := b.streams.curr.fetchSize
+		if b.streams.curr.qid == b.lastQid {
+			b.out.appendPullN(fetchSize)
+		} else {
+			b.out.appendPullNQid(fetchSize, b.streams.curr.qid)
+		}
 		b.out.send(b.conn)
 	}
 }
@@ -796,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
@@ -974,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

@@ -333,8 +333,8 @@ func TestBolt4(ot *testing.T) {
 			srv.send(msgRecord, []interface{}{"v1"})
 			// ... and the batch summary
 			srv.send(msgSuccess, map[string]interface{}{"has_more": true})
-			// Wait for the discard message
-			srv.waitForDiscardNAndQid(-1, int(qid))
+			// Wait for the discard message (no need for qid since the last executed query is discarded)
+			srv.waitForDiscardN(-1)
 			// Respond to discard with has more to indicate that there are more records
 			srv.send(msgSuccess, map[string]interface{}{"has_more": true})
 			// Wait for the commit
@@ -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/bolt/bolt_logging.go

@@ -53,10 +53,11 @@ type loggedSuccess struct {
 	TLast        string   `json:"t_last,omitempty"`
 	HasMore      bool     `json:"has_more,omitempy"`
 	Db           string   `json:"db,omitempty"`
+	Qid          int64    `json:"qid,omitempty"`
 }
 
 func (s loggableSuccess) String() string {
-	return serializeTrace(loggedSuccess{
+	success := loggedSuccess{
 		Server:       s.server,
 		ConnectionId: s.connectionId,
 		Fields:       s.fields,
@@ -65,7 +66,12 @@ func (s loggableSuccess) String() string {
 		TLast:        formatOmittingZero(s.tlast),
 		HasMore:      s.hasMore,
 		Db:           s.db,
-	})
+	}
+	if s.qid > -1 {
+		success.Qid = s.qid
+	}
+	return serializeTrace(success)
+
 }
 
 func formatOmittingZero(i int64) string {

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/resultsummary.go

@@ -44,8 +44,10 @@ const (
 type ResultSummary interface {
 	// Server returns basic information about the server where the statement is carried out.
 	Server() ServerInfo
-	// Statement returns statement that has been executed.
+	// Deprecated: since 4.4, will be removed in 5.0. Use Query instead
 	Statement() Statement
+	// Query returns the query that has been executed.
+	Query() Query
 	// StatementType returns type of statement that has been executed.
 	StatementType() StatementType
 	// Counters returns statistics counts for the statement.
@@ -96,10 +98,16 @@ type Counters interface {
 }
 
 type Statement interface {
+	Query
+}
+
+type Query interface {
 	// Text returns the statement's text.
 	Text() string
-	// Params returns the statement's parameters.
+	// Deprecated: since 4.4, will be removed in 5.0. Use Parameters instead
 	Params() map[string]interface{}
+	// Parameters returns the statement's parameters.
+	Parameters() map[string]interface{}
 }
 
 // ServerInfo contains basic information of the server.
@@ -219,6 +227,10 @@ func (s *resultSummary) Statement() Statement {
 	return s
 }
 
+func (s *resultSummary) Query() Query {
+	return s
+}
+
 func (s *resultSummary) StatementType() StatementType {
 	return StatementType(s.sum.StmntType)
 }
@@ -228,6 +240,10 @@ func (s *resultSummary) Text() string {
 }
 
 func (s *resultSummary) Params() map[string]interface{} {
+	return s.Parameters()
+}
+
+func (s *resultSummary) Parameters() map[string]interface{} {
 	return s.params
 }
 

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
@@ -458,9 +458,7 @@ func (s *session) Close() error {
 	var err error
 
 	if s.txExplicit != nil {
-		s.txExplicit.Close()
-		err = &UsageError{Message: "Closing session with a pending transaction"}
-		s.log.Warnf(log.Session, s.logId, err.Error())
+		err = s.txExplicit.Close()
 	}
 
 	if s.txAuto != nil {