(v2.12?) [ADDED] Protocol: INFO in response to CONNECT contains account info. (#6936)

In response to a leafnode or client CONNECT protocol, the server
responds
with an INFO that contains the name of the account it is bound to and if
this account is the system account.
    
For client connections, since most of our supported clients send the
CONNECT and a PING and expect a PONG *first*, the server will send
the "Connect Info" protocol only after receiving the first PING and
sending back the first PONG.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

server/client.go

@@ -2507,36 +2507,29 @@ func (c *client) processPing() {
 	// If we are here, the CONNECT has been received so we know
 	// if this client supports async INFO or not.
 	var (
-		checkInfoChange bool
+		sendConnectInfo bool
 		srv             = c.srv
 	)
-	// For older clients, just flip the firstPongSent flag if not already
-	// set and we are done.
-	if c.opts.Protocol < ClientProtoInfo || srv == nil {
-		c.flags.setIfNotSet(firstPongSent)
-	} else {
-		// This is a client that supports async INFO protocols.
-		// If this is the first PING (so firstPongSent is not set yet),
-		// we will need to check if there was a change in cluster topology
-		// or we have a different max payload. We will send this first before
-		// pong since most clients do flush after connect call.
-		checkInfoChange = !c.flags.isSet(firstPongSent)
+	// For the first PING (so firstPongSet is false) and for clients
+	// that support async INFO protocols, we will send one with ConnectInfo=true,
+	// the name of the account the client is bound to, and if the
+	// account is the system account.
+	if !c.flags.isSet(firstPongSent) {
+		// Flip the flag.
+		c.flags.set(firstPongSent)
+		// Evaluate if we should send the INFO protocol.
+		sendConnectInfo = srv != nil && c.opts.Protocol >= ClientProtoInfo
 	}
 	c.mu.Unlock()
 
-	if checkInfoChange {
-		opts := srv.getOpts()
+	if sendConnectInfo {
 		srv.mu.Lock()
+		info := srv.copyInfo()
 		c.mu.Lock()
-		// Now that we are under both locks, we can flip the flag.
-		// This prevents sendAsyncInfoToClients() and code here to
-		// send a double INFO protocol.
-		c.flags.set(firstPongSent)
-		// If there was a cluster update since this client was created,
-		// send an updated INFO protocol now.
-		if srv.lastCURLsUpdate >= c.start.UnixNano() || c.mpay != int32(opts.MaxPayload) {
-			c.enqueueProto(c.generateClientInfoJSON(srv.copyInfo()))
-		}
+		info.RemoteAccount = c.acc.Name
+		info.IsSystemAccount = c.acc == srv.SystemAccount()
+		info.ConnectInfo = true
+		c.enqueueProto(c.generateClientInfoJSON(info))
 		c.mu.Unlock()
 		srv.mu.Unlock()
 	}

server/leafnode.go

@@ -1991,12 +1991,16 @@ func (c *client) remoteCluster() string {
 // its permission settings for local enforcement.
 func (s *Server) sendPermsAndAccountInfo(c *client) {
 	// Copy
+	s.mu.Lock()
 	info := s.copyLeafNodeInfo()
+	s.mu.Unlock()
 	c.mu.Lock()
 	info.CID = c.cid
 	info.Import = c.opts.Import
 	info.Export = c.opts.Export
 	info.RemoteAccount = c.acc.Name
+	// s.SystemAccount() uses an atomic operation and does not get the server lock, so this is safe.
+	info.IsSystemAccount = c.acc == s.SystemAccount()
 	info.ConnectInfo = true
 	c.enqueueProto(generateInfoJSON(info))
 	c.mu.Unlock()

server/server.go

@@ -129,6 +129,9 @@ type Info struct {
 	WSConnectURLs     []string `json:"ws_connect_urls,omitempty"` // Contains URLs a ws client can connect to.
 	LameDuckMode      bool     `json:"ldm,omitempty"`
 	Compression       string   `json:"compression,omitempty"`
+	ConnectInfo       bool     `json:"connect_info,omitempty"`   // When true this is the server INFO response to CONNECT
+	RemoteAccount     string   `json:"remote_account,omitempty"` // Lets the client or leafnode side know the remote account that they bind to.
+	IsSystemAccount   bool     `json:"acc_is_sys,omitempty"`     // Indicates if the account is a system account.
 
 	// Route Specific
 	Import        *SubjectPermission `json:"import,omitempty"`
@@ -136,7 +139,6 @@ type Info struct {
 	LNOC          bool               `json:"lnoc,omitempty"`
 	LNOCU         bool               `json:"lnocu,omitempty"`
 	InfoOnConnect bool               `json:"info_on_connect,omitempty"` // When true the server will respond to CONNECT with an INFO
-	ConnectInfo   bool               `json:"connect_info,omitempty"`    // When true this is the server INFO response to CONNECT
 	RoutePoolSize int                `json:"route_pool_size,omitempty"`
 	RoutePoolIdx  int                `json:"route_pool_idx,omitempty"`
 	RouteAccount  string             `json:"route_account,omitempty"`
@@ -153,8 +155,7 @@ type Info struct {
 	GatewayIOM        bool     `json:"gateway_iom,omitempty"`         // Indicate that all accounts will be switched to InterestOnly mode "right away"
 
 	// LeafNode Specific
-	LeafNodeURLs  []string `json:"leafnode_urls,omitempty"`  // LeafNode URLs that the server can reconnect to.
-	RemoteAccount string   `json:"remote_account,omitempty"` // Lets the other side know the remote account that they bind to.
+	LeafNodeURLs []string `json:"leafnode_urls,omitempty"` // LeafNode URLs that the server can reconnect to.
 
 	XKey string `json:"xkey,omitempty"` // Public server's x25519 key.
 }
@@ -259,8 +260,6 @@ type Server struct {
 	// Used internally for quick look-ups.
 	clientConnectURLsMap refCountedUrlSet
 
-	lastCURLsUpdate int64
-
 	// For Gateways
 	gatewayListener    net.Listener // Accept listener
 	gatewayListenerErr error
@@ -3490,8 +3489,6 @@ func (s *Server) updateServerINFOAndSendINFOToClients(curls, wsurls []string, ad
 		updateInfo(&s.info.WSConnectURLs, s.websocket.connectURLs, s.websocket.connectURLsMap)
 	}
 	if cliUpdated || wsUpdated {
-		// Update the time of this update
-		s.lastCURLsUpdate = time.Now().UnixNano()
 		// Send to all registered clients that support async INFO protocols.
 		s.sendAsyncInfoToClients(cliUpdated, wsUpdated)
 	}

server/server_test.go

@@ -1147,6 +1147,8 @@ func TestLameDuckModeInfo(t *testing.T) {
 
 	getInfo(false)
 	c.Write([]byte("CONNECT {\"protocol\":1,\"verbose\":false}\r\nPING\r\n"))
+	// Consume both the first PONG and INFO in response to the Connect.
+	client.ReadString('\n')
 	client.ReadString('\n')
 
 	optsB := testWSOptions()

test/client_auth_test.go

@@ -17,7 +17,9 @@ import (
 	"fmt"
 	"os"
 	"testing"
+	"time"
 
+	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 )
 
@@ -88,3 +90,69 @@ func TestTokenInConfig(t *testing.T) {
 		t.Fatal("Expected auth to be required for the server")
 	}
 }
+
+func TestClientConnectInfo(t *testing.T) {
+	for _, test := range []struct {
+		name     string
+		sys      string
+		hasSys   bool
+		protocol int
+	}{
+		{"async info support with explicit system account", "system_account: SYS", true, server.ClientProtoInfo},
+		{"async info support without explicit system account", "", false, server.ClientProtoInfo},
+		{"no async info support with explicit system account", "system_account: SYS", true, server.ClientProtoZero},
+		{"no async info support without explicit system account", "", false, server.ClientProtoZero},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			conf := createConfFile(t, []byte(fmt.Sprintf(`
+				port: -1
+				%s
+				accounts {
+					SYS: { users: [{ user: sys, password: pwd}] }
+					A:   { users: [{ user: a, password: pwd}] }
+					B:   { users: [{ user: b, password: pwd}] }
+				}
+			`, test.sys)))
+			hub, oHub := RunServerWithConfig(conf)
+			defer hub.Shutdown()
+
+			checkInfoOnConnect := func(user, acc string, isSys bool) {
+				t.Helper()
+				cc := createClientConn(t, oHub.Host, oHub.Port)
+				defer cc.Close()
+
+				checkInfoMsg(t, cc)
+				sendProto(t, cc, fmt.Sprintf("CONNECT {\"user\":%q,\"pass\":\"pwd\",\"pedantic\":false,\"verbose\":false,\"protocol\":%d}\r\nPING\r\n", user, test.protocol))
+				// Since the PONG and INFO may be receive as a single TCP read,
+				// make sure we consume the pong alone here.
+				pongBuf := make([]byte, len("PONG\r\n"))
+				cc.SetReadDeadline(time.Now().Add(2 * time.Second))
+				n, err := cc.Read(pongBuf)
+				cc.SetReadDeadline(time.Time{})
+				if n <= 0 && err != nil {
+					t.Fatalf("Error reading from conn: %v\n", err)
+				}
+				if !pongRe.Match(pongBuf) {
+					t.Fatalf("Response did not match expected: \n\tReceived:'%q'\n\tExpected:'%s'\n", pongBuf, pongRe)
+				}
+				if test.protocol < server.ClientProtoInfo {
+					expectNothing(t, cc)
+				} else {
+					info := checkInfoMsg(t, cc)
+					if !info.ConnectInfo {
+						t.Fatal("Expected ConnectInfo to be true")
+					}
+					if an := info.RemoteAccount; an != acc {
+						t.Fatalf("Expected account %q, got %q", acc, info.RemoteAccount)
+					}
+					if ais := info.IsSystemAccount; ais != isSys {
+						t.Fatalf("Expected IsSystemAccount to be %v, got %v", isSys, ais)
+					}
+				}
+			}
+			checkInfoOnConnect("a", "A", false)
+			checkInfoOnConnect("sys", "SYS", test.hasSys)
+			checkInfoOnConnect("b", "B", false)
+		})
+	}
+}

test/leafnode_test.go

@@ -4450,3 +4450,55 @@ func TestLeafNodeClusterNameWithSpacesRejected(t *testing.T) {
 	leafExpect(errRe)
 	expectDisconnect(t, lc)
 }
+
+func TestLeafNodeConnectInfo(t *testing.T) {
+	for _, test := range []struct {
+		name   string
+		sys    string
+		hasSys bool
+	}{
+		{"with explicit system account", "system_account: SYS", true},
+		{"without explicit system account", "", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			conf := createConfFile(t, []byte(fmt.Sprintf(`
+				port: -1
+				%s
+				accounts {
+					SYS: { users: [{ user: sys, password: pwd}] }
+					A:   { users: [{ user: a, password: pwd}] }
+					B:   { users: [{ user: b, password: pwd}] }
+				}
+				leafnodes {
+					port: -1
+				}
+			`, test.sys)))
+			hub, oHub := RunServerWithConfig(conf)
+			defer hub.Shutdown()
+
+			checkInfoOnConnect := func(user, acc string, isSys bool) {
+				t.Helper()
+				lc := createLeafConn(t, oHub.LeafNode.Host, oHub.LeafNode.Port)
+				defer lc.Close()
+
+				checkInfoMsg(t, lc)
+
+				sendProto(t, lc, fmt.Sprintf("CONNECT {\"user\":%q,\"pass\":\"pwd\"}\r\n", user))
+				info := checkInfoMsg(t, lc)
+				if !info.ConnectInfo {
+					t.Fatal("Expected ConnectInfo to be true")
+				}
+				if an := info.RemoteAccount; an != acc {
+					t.Fatalf("Expected account %q, got %q", acc, info.RemoteAccount)
+				}
+				if ais := info.IsSystemAccount; ais != isSys {
+					t.Fatalf("Expected IsSystemAccount to be %v, got %v", isSys, ais)
+				}
+				checkLeafNodeConnected(t, hub)
+			}
+			checkInfoOnConnect("a", "A", false)
+			checkInfoOnConnect("sys", "SYS", test.hasSys)
+			checkInfoOnConnect("b", "B", false)
+		})
+	}
+}

test/routes_test.go

@@ -586,11 +586,25 @@ func TestRouteSendAsyncINFOToClients(t *testing.T) {
 
 		newClientSend, newClientExpect := setupConnWithProto(t, newClient, clientProtoInfo)
 		newClientSend("PING\r\n")
-		newClientExpect(pongRe)
 
-		// Check that even a new client does not receive an async INFO at this point
-		// since there is no route created yet.
-		expectNothing(t, newClient)
+		// For new clients, the initial PING (after a CONNECT) will result in a
+		// PONG followed by an INFO, so make sure we receive them separately.
+		getPongAlone := func(c net.Conn) {
+			t.Helper()
+			pongBuf := make([]byte, len("PONG\r\n"))
+			c.SetReadDeadline(time.Now().Add(2 * time.Second))
+			n, err := c.Read(pongBuf)
+			c.SetReadDeadline(time.Time{})
+			if n <= 0 && err != nil {
+				t.Fatalf("Error reading from conn: %v\n", err)
+			}
+			if !pongRe.Match(pongBuf) {
+				t.Fatalf("Response did not match expected: \n\tReceived:'%q'\n\tExpected:'%s'\n", pongBuf, pongRe)
+			}
+		}
+		getPongAlone(newClient)
+		// The new client should receive an INFO with the ConnectInfo boolean set to true.
+		newClientExpect(infoRe)
 
 		routeID := "Server-B"
 
@@ -658,10 +672,16 @@ func TestRouteSendAsyncINFOToClients(t *testing.T) {
 			}
 		}
 
+		var checkConnectInfo bool
 		checkINFOReceived := func(client net.Conn, clientExpect expectFun, expectedURLs []string) {
 			if opts.Cluster.NoAdvertise {
-				expectNothing(t, client)
-				return
+				if !checkConnectInfo {
+					expectNothing(t, client)
+					return
+				}
+				// Since it is no advertise, but we still get an INFO, the URLs
+				// should be empty.
+				expectedURLs = nil
 			}
 			buf := clientExpect(infoRe)
 			info := server.Info{}
@@ -669,6 +689,17 @@ func TestRouteSendAsyncINFOToClients(t *testing.T) {
 				stackFatalf(t, "Could not unmarshal route info: %v", err)
 			}
 			checkClientConnectURLS(info.ClientConnectURLs, expectedURLs)
+			if checkConnectInfo {
+				if !info.ConnectInfo {
+					stackFatalf(t, "ConnectInfo should have been true")
+				}
+				if info.RemoteAccount != "$G" {
+					stackFatalf(t, "RemoteAccount should be %q, got %q", "$G", info.RemoteAccount)
+				}
+				if info.IsSystemAccount {
+					stackFatalf(t, "IsSystemAccount should have been false")
+				}
+			}
 		}
 
 		// Create a route
@@ -751,36 +782,20 @@ func TestRouteSendAsyncINFOToClients(t *testing.T) {
 
 		// Now send the first PING
 		clientNoPingSend("PING\r\n")
-		// Should receive PONG followed by INFO
-		// Receive PONG only first
-		pongBuf := make([]byte, len("PONG\r\n"))
-		clientNoPing.SetReadDeadline(time.Now().Add(2 * time.Second))
-		n, err := clientNoPing.Read(pongBuf)
-		clientNoPing.SetReadDeadline(time.Time{})
-		if n <= 0 && err != nil {
-			t.Fatalf("Error reading from conn: %v\n", err)
-		}
-		if !pongRe.Match(pongBuf) {
-			t.Fatalf("Response did not match expected: \n\tReceived:'%q'\n\tExpected:'%s'\n", pongBuf, pongRe)
-		}
+		getPongAlone(clientNoPing)
+		// We should receive an INFO but with ConnectInfo==true, etc..
+		checkConnectInfo = true
 		checkINFOReceived(clientNoPing, clientNoPingExpect, expectedURLs)
+		checkConnectInfo = false
 
 		// Have the client that did not send the connect do it now
 		clientNoConnectSend, clientNoConnectExpect := setupConnWithProto(t, clientNoConnect, clientProtoInfo)
 		// Send the PING
 		clientNoConnectSend("PING\r\n")
-		// Should receive PONG followed by INFO
-		// Receive PONG only first
-		clientNoConnect.SetReadDeadline(time.Now().Add(2 * time.Second))
-		n, err = clientNoConnect.Read(pongBuf)
-		clientNoConnect.SetReadDeadline(time.Time{})
-		if n <= 0 && err != nil {
-			t.Fatalf("Error reading from conn: %v\n", err)
-		}
-		if !pongRe.Match(pongBuf) {
-			t.Fatalf("Response did not match expected: \n\tReceived:'%q'\n\tExpected:'%s'\n", pongBuf, pongRe)
-		}
+		getPongAlone(clientNoConnect)
+		checkConnectInfo = true
 		checkINFOReceived(clientNoConnect, clientNoConnectExpect, expectedURLs)
+		checkConnectInfo = false
 
 		// Create a client connection and verify content of initial INFO contains array
 		// (but empty if no advertise option is set)
@@ -789,7 +804,7 @@ func TestRouteSendAsyncINFOToClients(t *testing.T) {
 		buf := expectResult(t, cli, infoRe)
 		js := infoRe.FindAllSubmatch(buf, 1)[0][1]
 		var sinfo server.Info
-		err = json.Unmarshal(js, &sinfo)
+		err := json.Unmarshal(js, &sinfo)
 		if err != nil {
 			t.Fatalf("Could not unmarshal INFO json: %v\n", err)
 		}

test/test.go

@@ -313,7 +313,7 @@ func setupConnWithAccount(t tLogger, s *server.Server, c net.Conn, account strin
 
 func setupConnWithUserPass(t tLogger, c net.Conn, username, password string) (sendFun, expectFun) {
 	checkInfoMsg(t, c)
-	cs := fmt.Sprintf("CONNECT {\"verbose\":%v,\"pedantic\":%v,\"tls_required\":%v,\"protocol\":1,\"user\":%q,\"pass\":%q}\r\n",
+	cs := fmt.Sprintf("CONNECT {\"verbose\":%v,\"pedantic\":%v,\"tls_required\":%v,\"user\":%q,\"pass\":%q}\r\n",
 		false, false, false, username, password)
 	sendProto(t, c, cs)
 	return sendCommand(t, c), expectLefMostCommand(t, c)