Repository: nats-io/nats-server
Pull request #2437 (2021-08-13 16:43:10)

Added in ability for normal accounts to access scoped connz info.
Added in client kind and sub type for clients.
Added in ability to filter connections based on matching subject interest.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 9fe4a81c..e20be9bc 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -816,8 +816,7 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 }
 
 // SubscriptionInterest returns true if this account has a matching subscription
-// for the given `subject`. Works only for literal subjects.
-// TODO: Add support for wildcards
+// for the given `subject`.
 func (a *Account) SubscriptionInterest(subject string) bool {
 	return a.Interest(subject) > 0
 }
@@ -1377,7 +1376,7 @@ func (a *Account) AddServiceImportWithClaim(destination *Account, from, to strin
 		return ErrMissingAccount
 	}
 	// Empty means use from.
-	if to == "" {
+	if to == _EMPTY_ {
 		to = from
 	}
 	if !IsValidSubject(from) || !IsValidSubject(to) {
@@ -2277,7 +2276,7 @@ func (a *Account) AddStreamImportWithClaim(account *Account, from, prefix string
 
 	// Check prefix if it exists and make sure its a literal.
 	// Append token separator if not already present.
-	if prefix != "" {
+	if prefix != _EMPTY_ {
 		// Make sure there are no wildcards here, this prefix needs to be a literal
 		// since it will be prepended to a publish subject.
 		if !subjectIsLiteral(prefix) {
@@ -2415,7 +2414,7 @@ func (a *Account) checkStreamImportAuthorizedNoLock(account *Account, subject st
 
 func (a *Account) checkAuth(ea *exportAuth, account *Account, imClaim *jwt.Import, tokens []string) bool {
 	// if ea is nil or ea.approved is nil, that denotes a public export
-	if ea == nil || (ea.approved == nil && !ea.tokenReq && ea.accountPos == 0) {
+	if ea == nil || (len(ea.approved) == 0 && !ea.tokenReq && ea.accountPos == 0) {
 		return true
 	}
 	// Check if the export is protected and enforces presence of importing account identity
@@ -2444,6 +2443,7 @@ func (a *Account) checkStreamExportApproved(account *Account, subject string, im
 		}
 		return a.checkAuth(&ea.exportAuth, account, imClaim, nil)
 	}
+
 	// ok if we are here we did not match directly so we need to test each one.
 	// The import subject arg has to take precedence, meaning the export
 	// has to be a true subset of the import claim. We already checked for
@@ -2464,7 +2464,7 @@ func (a *Account) checkServiceExportApproved(account *Account, subject string, i
 	// Check direct match of subject first
 	se, ok := a.exports.services[subject]
 	if ok {
-		// if ea is nil or eq.approved is nil, that denotes a public export
+		// if se is nil or eq.approved is nil, that denotes a public export
 		if se == nil {
 			return true
 		}
diff --git a/server/events.go b/server/events.go
index 2aabcc63..c80d9a60 100644
--- a/server/events.go
+++ b/server/events.go
@@ -57,6 +57,7 @@ const (
 	leafNodeConnectEventSubj = "$SYS.ACCOUNT.%s.LEAFNODE.CONNECT" // for internal use only
 	remoteLatencyEventSubj   = "$SYS.LATENCY.M2.%s"
 	inboxRespSubj            = "$SYS._INBOX.%s.%s"
+	accConnzReqSubj          = "$SYS.REQ.ACCOUNT.PING.CONNZ"
 
 	// FIXME(dlc) - Should account scope, even with wc for now, but later on
 	// we can then shard as needed.
@@ -464,9 +465,8 @@ func (s *Server) eventsRunning() bool {
 // a defined system account.
 func (s *Server) EventsEnabled() bool {
 	s.mu.Lock()
-	ee := s.eventsEnabled()
-	s.mu.Unlock()
-	return ee
+	defer s.mu.Unlock()
+	return s.eventsEnabled()
 }
 
 // eventsEnabled will report if events are enabled.
@@ -777,7 +777,7 @@ func (s *Server) initEventTracking() {
 	}
 	extractAccount := func(subject string) (string, error) {
 		if tk := strings.Split(subject, tsep); len(tk) != accReqTokens {
-			return "", fmt.Errorf("subject %q is malformed", subject)
+			return _EMPTY_, fmt.Errorf("subject %q is malformed", subject)
 		} else {
 			return tk[accReqAccIndex], nil
 		}
@@ -801,6 +801,14 @@ func (s *Server) initEventTracking() {
 				if acc, err := extractAccount(subject); err != nil {
 					return nil, err
 				} else {
+					if ci, _, _, _, err := c.srv.getRequestInfo(c, msg); err == nil && ci.Account != _EMPTY_ {
+						// Make sure the accounts match.
+						if ci.Account != acc {
+							// Do not leak too much here.
+							return nil, fmt.Errorf("bad request")
+						}
+						optz.ConnzOptions.isAccountReq = true
+					}
 					optz.ConnzOptions.Account = acc
 					return s.Connz(&optz.ConnzOptions)
 				}
@@ -863,14 +871,47 @@ func (s *Server) initEventTracking() {
 	}
 }
 
+// register existing accounts with any system exports.
+func (s *Server) registerSystemImportsForExisting() {
+	var accounts []*Account
+
+	s.mu.Lock()
+	if s.sys == nil {
+		s.mu.Unlock()
+		return
+	}
+	sacc := s.sys.account
+	s.accounts.Range(func(k, v interface{}) bool {
+		a := v.(*Account)
+		if a != sacc {
+			accounts = append(accounts, a)
+		}
+		return true
+	})
+	s.mu.Unlock()
+
+	for _, a := range accounts {
+		s.registerSystemImports(a)
+	}
+}
+
 // add all exports a system account will need
 func (s *Server) addSystemAccountExports(sacc *Account) {
 	if !s.EventsEnabled() {
 		return
 	}
+	accConnzSubj := fmt.Sprintf(accReqSubj, "*", "CONNZ")
+	if err := sacc.AddServiceExportWithResponse(accConnzSubj, Streamed, nil); err != nil {
+		s.Errorf("Error adding system service export for %q: %v", accConnzSubj, err)
+	}
+	// Register any accounts that existed prior.
+	s.registerSystemImportsForExisting()
+
+	// FIXME(dlc) - Old experiment, Remove?
 	if err := sacc.AddServiceExport(accSubsSubj, nil); err != nil {
 		s.Errorf("Error adding system service export for %q: %v", accSubsSubj, err)
 	}
+
 	if s.JetStreamEnabled() {
 		s.checkJetStreamExports()
 	}
@@ -1329,6 +1370,29 @@ func (s *Server) remoteConnsUpdate(sub *subscription, _ *client, _ *Account, sub
 	}
 }
 
+// This will import any system level exports.
+func (s *Server) registerSystemImports(a *Account) {
+	if a == nil || !s.eventsEnabled() {
+		return
+	}
+	sacc := s.SystemAccount()
+	if sacc == nil {
+		return
+	}
+	// FIXME(dlc) - make a shared list between sys exports etc.
+	connzSubj := fmt.Sprintf(serverPingReqSubj, "CONNZ")
+	mappedSubj := fmt.Sprintf(accReqSubj, a.Name, "CONNZ")
+
+	// Add in this to the account in 2 places.
+	// "$SYS.REQ.SERVER.PING.CONNZ" and "$SYS.REQ.ACCOUNT.CONNZ"
+	if err := a.AddServiceImport(sacc, connzSubj, mappedSubj); err != nil {
+		s.Errorf("Error setting up system service imports for account: %v", err)
+	}
+	if err := a.AddServiceImport(sacc, accConnzReqSubj, mappedSubj); err != nil {
+		s.Errorf("Error setting up system service imports for account: %v", err)
+	}
+}
+
 // Setup tracking for this account. This allows us to track global account activity.
 // Lock should be held on entry.
 func (s *Server) enableAccountTracking(a *Account) {
diff --git a/server/monitor.go b/server/monitor.go
index 9e27323b..dbe3cd78 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -86,6 +86,13 @@ type ConnzOptions struct {
 
 	// Filter by account.
 	Account string `json:"acc"`
+
+	// Filter by subject interest
+	FilterSubject string `json:"filter_subject"`
+
+	// Private indication that this request is from an account and not a system account.
+	// Used to not leak system level information to the account.
+	isAccountReq bool
 }
 
 // ConnState is for filtering states of connections. We will only have two, open and closed.
@@ -103,6 +110,8 @@ const (
 // ConnInfo has detailed information on a per connection basis.
 type ConnInfo struct {
 	Cid            uint64      `json:"cid"`
+	Kind           string      `json:"kind,omitempty"`
+	Type           string      `json:"type,omitempty"`
 	IP             string      `json:"ip"`
 	Port           int         `json:"port"`
 	Start          time.Time   `json:"start"`
@@ -170,11 +179,13 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 		state   = ConnOpen
 		user    string
 		acc     string
+		a       *Account
+		filter  string
 	)
 
 	if opts != nil {
 		// If no sort option given or sort is by uptime, then sort by cid
-		if opts.Sort == "" {
+		if opts.Sort == _EMPTY_ {
 			sortOpt = ByCid
 		} else {
 			sortOpt = opts.Sort
@@ -185,7 +196,7 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 
 		// Auth specifics.
 		auth = opts.Username
-		if !auth && (user != "" || acc != "") {
+		if !auth && (user != _EMPTY_ || acc != _EMPTY_) {
 			return nil, fmt.Errorf("filter by user or account only allowed with auth option")
 		}
 		user = opts.User
@@ -212,12 +223,18 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 		if sortOpt == ByReason && state != ConnClosed {
 			return nil, fmt.Errorf("sort by reason only valid on closed connections")
 		}
-
 		// If searching by CID
 		if opts.CID > 0 {
 			cid = opts.CID
 			limit = 1
 		}
+		// If filtering by subject.
+		if opts.FilterSubject != _EMPTY_ && opts.FilterSubject != fwcs {
+			if acc == _EMPTY_ {
+				return nil, fmt.Errorf("filter by subject only valid with account filtering")
+			}
+			filter = opts.FilterSubject
+		}
 	}
 
 	c := &Connz{
@@ -231,8 +248,31 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 	// Hold for closed clients if requested.
 	var closedClients []*closedClient
 
+	var clist map[uint64]*client
+
+	// If this is an account scoped request from a no $SYS account.
+	isAccReq := acc != _EMPTY_ && opts.isAccountReq
+
+	if acc != _EMPTY_ {
+		var err error
+		a, err = s.lookupAccount(acc)
+		if err != nil {
+			return c, nil
+		}
+		a.mu.RLock()
+		clist = make(map[uint64]*client, a.numLocalConnections())
+		for c := range a.clients {
+			clist[c.cid] = c
+		}
+		a.mu.RUnlock()
+	}
+
 	// Walk the open client list with server lock held.
 	s.mu.Lock()
+	// Default to all client unless filled in above.
+	if clist == nil {
+		clist = s.clients
+	}
 
 	// copy the server id for monitoring
 	c.ID = s.info.ID
@@ -241,14 +281,34 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 	// may be smaller if pagination is used.
 	switch state {
 	case ConnOpen:
-		c.Total = len(s.clients)
+		if isAccReq {
+			c.Total = a.NumLocalConnections()
+		} else {
+			c.Total = len(s.clients)
+		}
 	case ConnClosed:
-		c.Total = s.closed.len()
 		closedClients = s.closed.closedClients()
 		c.Total = len(closedClients)
 	case ConnAll:
+		if isAccReq {
+			c.Total = a.NumLocalConnections()
+		} else {
+			c.Total = len(s.clients)
+		}
 		closedClients = s.closed.closedClients()
-		c.Total = len(s.clients) + len(closedClients)
+		c.Total += len(closedClients)
+	}
+
+	// We may need to filter these connections.
+	if isAccReq && len(closedClients) > 0 {
+		var ccc []*closedClient
+		for _, cc := range closedClients {
+			if cc.acc == acc {
+				ccc = append(ccc, cc)
+			}
+		}
+		c.Total -= (len(closedClients) - len(ccc))
+		closedClients = ccc
 	}
 
 	totalClients := c.Total
@@ -294,13 +354,13 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 	} else {
 		// Gather all open clients.
 		if state == ConnOpen || state == ConnAll {
-			for _, client := range s.clients {
+			for _, client := range clist {
 				// If we have an account specified we need to filter.
-				if acc != "" && (client.acc == nil || client.acc.Name != acc) {
+				if acc != _EMPTY_ && (client.acc == nil || client.acc.Name != acc) {
 					continue
 				}
 				// Do user filtering second
-				if user != "" && client.opts.Username != user {
+				if user != _EMPTY_ && client.opts.Username != user {
 					continue
 				}
 				openClients = append(openClients, client)
@@ -309,6 +369,22 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 	}
 	s.mu.Unlock()
 
+	// Filter by subject now if needed. We do this outside of server lock.
+	if filter != _EMPTY_ {
+		var oc []*client
+		for _, c := range openClients {
+			c.mu.Lock()
+			for _, sub := range c.subs {
+				if SubjectsCollide(filter, string(sub.subject)) {
+					oc = append(oc, c)
+					break
+				}
+			}
+			c.mu.Unlock()
+			openClients = oc
+		}
+	}
+
 	// Just return with empty array if nothing here.
 	if len(openClients) == 0 && len(closedClients) == 0 {
 		c.Conns = ConnInfos{}
@@ -316,7 +392,6 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 	}
 
 	// Now whip through and generate ConnInfo entries
-
 	// Open Clients
 	i := 0
 	for _, client := range openClients {
@@ -354,11 +429,11 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 	}
 	for _, cc := range closedClients {
 		// If we have an account specified we need to filter.
-		if acc != "" && cc.acc != acc {
+		if acc != _EMPTY_ && cc.acc != acc {
 			continue
 		}
 		// Do user filtering second
-		if user != "" && cc.user != user {
+		if user != _EMPTY_ && cc.user != user {
 			continue
 		}
 
@@ -451,6 +526,8 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 // client should be locked.
 func (ci *ConnInfo) fill(client *client, nc net.Conn, now time.Time) {
 	ci.Cid = client.cid
+	ci.Kind = client.kindString()
+	ci.Type = client.clientTypeString()
 	ci.Start = client.start
 	ci.LastActivity = client.last
 	ci.Uptime = myUptime(now.Sub(client.start))
diff --git a/server/server.go b/server/server.go
index 5e993f87..cce12c7a 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1319,6 +1319,12 @@ func (s *Server) registerAccountNoLock(acc *Account) *Account {
 	s.accounts.Store(acc.Name, acc)
 	s.tmpAccounts.Delete(acc.Name)
 	s.enableAccountTracking(acc)
+
+	// Can not have server lock here.
+	s.mu.Unlock()
+	s.registerSystemImports(acc)
+	s.mu.Lock()
+
 	return nil
 }
 


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index abdd0a7b..f0b9df99 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1256,8 +1256,8 @@ func TestAccountReqMonitoring(t *testing.T) {
 	// query SUBSZ for account
 	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_subscriptions":1,`) {
-		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions":3,`) {
+		t.Fatalf("unexpected subs count (expected 3): %v", string(resp.Data))
 	}
 	// create a subscription
 	if sub, err := nc.Subscribe("foo", func(msg *nats.Msg) {}); err != nil {
@@ -1269,8 +1269,8 @@ func TestAccountReqMonitoring(t *testing.T) {
 	// query SUBSZ for account
 	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_subscriptions":2,`) {
-		t.Fatalf("unexpected subs count (expected 2): %v", string(resp.Data))
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions":4,`) {
+		t.Fatalf("unexpected subs count (expected 4): %v", string(resp.Data))
 	} else if !strings.Contains(string(resp.Data), `"subject":"foo"`) {
 		t.Fatalf("expected subscription foo: %v", string(resp.Data))
 	}
@@ -1362,15 +1362,15 @@ func TestAccountReqInfo(t *testing.T) {
 		t.Fatalf("Unmarshalling failed: %v", err)
 	} else if len(info.Exports) != 1 {
 		t.Fatalf("Unexpected value: %v", info.Exports)
-	} else if len(info.Imports) != 0 {
-		t.Fatalf("Unexpected value: %v", info.Imports)
+	} else if len(info.Imports) != 2 {
+		t.Fatalf("Unexpected value: %+v", info.Imports)
 	} else if info.Exports[0].Subject != "req.*" {
 		t.Fatalf("Unexpected value: %v", info.Exports)
 	} else if info.Exports[0].Type != jwt.Service {
 		t.Fatalf("Unexpected value: %v", info.Exports)
 	} else if info.Exports[0].ResponseType != jwt.ResponseTypeSingleton {
 		t.Fatalf("Unexpected value: %v", info.Exports)
-	} else if info.SubCnt != 0 {
+	} else if info.SubCnt != 2 {
 		t.Fatalf("Unexpected value: %v", info.SubCnt)
 	} else {
 		checkCommon(&info, &srv, pub1, ajwt1)
@@ -1383,16 +1383,26 @@ func TestAccountReqInfo(t *testing.T) {
 		t.Fatalf("Unmarshalling failed: %v", err)
 	} else if len(info.Exports) != 0 {
 		t.Fatalf("Unexpected value: %v", info.Exports)
-	} else if len(info.Imports) != 1 {
-		t.Fatalf("Unexpected value: %v", info.Imports)
-	} else if info.Imports[0].Subject != "req.1" {
-		t.Fatalf("Unexpected value: %v", info.Exports)
-	} else if info.Imports[0].Type != jwt.Service {
-		t.Fatalf("Unexpected value: %v", info.Exports)
-	} else if info.Imports[0].Account != pub1 {
-		t.Fatalf("Unexpected value: %v", info.Exports)
-	} else if info.SubCnt != 1 {
-		t.Fatalf("Unexpected value: %v", info.SubCnt)
+	} else if len(info.Imports) != 3 {
+		t.Fatalf("Unexpected value: %+v", info.Imports)
+	}
+	// Here we need to find our import
+	var si *ExtImport
+	for _, im := range info.Imports {
+		if im.Subject == "req.1" {
+			si = &im
+			break
+		}
+	}
+	if si == nil {
+		t.Fatalf("Could not find our import")
+	}
+	if si.Type != jwt.Service {
+		t.Fatalf("Unexpected value: %+v", si)
+	} else if si.Account != pub1 {
+		t.Fatalf("Unexpected value: %+v", si)
+	} else if info.SubCnt != 3 {
+		t.Fatalf("Unexpected value: %+v", si)
 	} else {
 		checkCommon(&info, &srv, pub2, ajwt2)
 	}
@@ -1449,7 +1459,7 @@ func TestAccountClaimsUpdatesWithServiceImports(t *testing.T) {
 	}
 	nc.Flush()
 
-	if startSubs != s.NumSubscriptions() {
+	if startSubs < s.NumSubscriptions() {
 		t.Fatalf("Subscriptions leaked: %d vs %d", startSubs, s.NumSubscriptions())
 	}
 }
@@ -1594,7 +1604,7 @@ func TestSystemAccountWithGateways(t *testing.T) {
 
 	// If this tests fails with wrong number after 10 seconds we may have
 	// added a new inititial subscription for the eventing system.
-	checkExpectedSubs(t, 37, sa)
+	checkExpectedSubs(t, 39, sa)
 
 	// Create a client on B and see if we receive the event
 	urlb := fmt.Sprintf("nats://%s:%d", ob.Host, ob.Port)
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index a953a7c4..78b7b6d4 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -8993,4 +8993,5 @@ func (c *cluster) stableTotalSubs() (total int) {
 		return fmt.Errorf("Still stabilizing")
 	})
 	return nsubs
+
 }
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index 72c325e3..d3121449 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -1285,8 +1285,8 @@ func TestLeafNodePermissions(t *testing.T) {
 
 	// Create a sub on ">" on LN1
 	subAll := natsSubSync(t, nc1, ">")
-	// this should be registered in LN2 (there is 1 sub for LN1 $LDS subject)
-	checkSubs(ln2.globalAccount(), 2)
+	// this should be registered in LN2 (there is 1 sub for LN1 $LDS subject) + SYS IMPORTS
+	checkSubs(ln2.globalAccount(), 6)
 
 	// Check deny export clause from messages published from LN2
 	for _, test := range []struct {
@@ -1312,8 +1312,8 @@ func TestLeafNodePermissions(t *testing.T) {
 	}
 
 	subAll.Unsubscribe()
-	// Goes down to 1 (the $LDS one)
-	checkSubs(ln2.globalAccount(), 1)
+	// Goes down by 1.
+	checkSubs(ln2.globalAccount(), 5)
 
 	// We used to make sure we would not do subscriptions however that
 	// was incorrect. We need to check publishes, not the subscriptions.
@@ -1338,7 +1338,7 @@ func TestLeafNodePermissions(t *testing.T) {
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			sub := natsSubSync(t, nc2, test.subSubject)
-			checkSubs(ln2.globalAccount(), 2)
+			checkSubs(ln2.globalAccount(), 6)
 
 			if !test.ok {
 				nc1.Publish(test.pubSubject, []byte("msg"))
@@ -1346,12 +1346,12 @@ func TestLeafNodePermissions(t *testing.T) {
 					t.Fatalf("Did not expect to get the message")
 				}
 			} else {
-				checkSubs(ln1.globalAccount(), 2)
+				checkSubs(ln1.globalAccount(), 6)
 				nc1.Publish(test.pubSubject, []byte("msg"))
 				natsNexMsg(t, sub, time.Second)
 			}
 			sub.Unsubscribe()
-			checkSubs(ln1.globalAccount(), 1)
+			checkSubs(ln1.globalAccount(), 5)
 
 		})
 	}
@@ -1472,8 +1472,8 @@ func TestLeafNodeExportPermissionsNotForSpecialSubs(t *testing.T) {
 	// The deny is totally restrictive, but make sure that we still accept the $LDS, $GR and _GR_ go from LN1.
 	checkFor(t, time.Second, 15*time.Millisecond, func() error {
 		// We should have registered the 3 subs from the accepting leafnode.
-		if n := ln2.globalAccount().TotalSubs(); n != 3 {
-			return fmt.Errorf("Expected %d subs, got %v", 3, n)
+		if n := ln2.globalAccount().TotalSubs(); n != 7 {
+			return fmt.Errorf("Expected %d subs, got %v", 7, n)
 		}
 		return nil
 	})
@@ -3269,8 +3269,8 @@ func TestLeafNodeRouteSubWithOrigin(t *testing.T) {
 	r1.Shutdown()
 	checkFor(t, time.Second, 15*time.Millisecond, func() error {
 		acc := l2.GlobalAccount()
-		if n := acc.TotalSubs(); n != 0 {
-			return fmt.Errorf("Account %q should have 0 sub, got %v", acc.GetName(), n)
+		if n := acc.TotalSubs(); n != 2 {
+			return fmt.Errorf("Account %q should have 2 subs, got %v", acc.GetName(), n)
 		}
 		return nil
 	})
@@ -3720,9 +3720,9 @@ func TestLeafNodeUniqueServerNameCrossJSDomain(t *testing.T) {
 		listen: -1
 		server_name: %s
 		jetstream {
-			max_mem_store: 256MB, 
-			max_file_store: 2GB, 
-			store_dir: "%s", 
+			max_mem_store: 256MB,
+			max_file_store: 2GB,
+			store_dir: "%s",
 			domain: hub
 		}
 		accounts {
@@ -3737,9 +3737,9 @@ func TestLeafNodeUniqueServerNameCrossJSDomain(t *testing.T) {
 		listen: -1
 		server_name: %s
 		jetstream {
-			max_mem_store: 256MB, 
-			max_file_store: 2GB, 
-			store_dir: "%s", 
+			max_mem_store: 256MB,
+			max_file_store: 2GB,
+			store_dir: "%s",
 			domain: %s
 		}
 		accounts {
diff --git a/server/monitor_test.go b/server/monitor_test.go
index b4de3d8d..cd163c01 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -3784,8 +3784,8 @@ func TestMonitorLeafz(t *testing.T) {
 				t.Fatalf("RTT not tracked?")
 			}
 			// LDS should be only one.
-			if ln.NumSubs != 1 || len(ln.Subs) != 1 {
-				t.Fatalf("Expected 1 sub, got %v (%v)", ln.NumSubs, ln.Subs)
+			if ln.NumSubs != 3 || len(ln.Subs) != 3 {
+				t.Fatalf("Expected 3 subs, got %v (%v)", ln.NumSubs, ln.Subs)
 			}
 		}
 	}
diff --git a/server/norace_test.go b/server/norace_test.go
index 32703dd6..0ee073c6 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -2550,6 +2550,134 @@ func TestNoRaceJetStreamStalledMirrorsAfterExpire(t *testing.T) {
 	})
 }
 
+// We will use JetStream helpers to create supercluster but this test is about exposing the ability to access
+// account scoped connz with subject interest filtering.
+func TestNoRaceAccountConnz(t *testing.T) {
+	// This has 4 different account, 3 general and system.
+	sc := createJetStreamSuperClusterWithTemplate(t, jsClusterAccountsTempl, 3, 3)
+	defer sc.shutdown()
+
+	// Create 20 connections on account one and two
+	num := 20
+	for i := 0; i < num; i++ {
+		nc, _ := jsClientConnect(t, sc.randomServer(), nats.UserInfo("one", "p"), nats.Name("one"))
+		defer nc.Close()
+
+		if i%2 == 0 {
+			nc.SubscribeSync("foo")
+		} else {
+			nc.SubscribeSync("bar")
+		}
+
+		nc, _ = jsClientConnect(t, sc.randomServer(), nats.UserInfo("two", "p"), nats.Name("two"))
+		nc.SubscribeSync("baz")
+		nc.SubscribeSync("foo.bar.*")
+		nc.SubscribeSync(fmt.Sprintf("id.%d", i+1))
+		defer nc.Close()
+	}
+
+	type czapi struct {
+		Server *ServerInfo
+		Data   *Connz
+		Error  *ApiError
+	}
+
+	parseConnz := func(buf []byte) *Connz {
+		t.Helper()
+		var cz czapi
+		if err := json.Unmarshal(buf, &cz); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if cz.Error != nil {
+			t.Fatalf("Unexpected error: %+v", cz.Error)
+		}
+		return cz.Data
+	}
+
+	doRequest := func(reqSubj, acc, filter string, expected int) {
+		t.Helper()
+		nc, _ := jsClientConnect(t, sc.randomServer(), nats.UserInfo(acc, "p"), nats.Name(acc))
+		defer nc.Close()
+
+		mch := make(chan *nats.Msg, 9)
+		sub, _ := nc.ChanSubscribe(nats.NewInbox(), mch)
+
+		var req []byte
+		if filter != _EMPTY_ {
+			req, _ = json.Marshal(&ConnzOptions{FilterSubject: filter})
+		}
+
+		if err := nc.PublishRequest(reqSubj, sub.Subject, req); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		// So we can igniore ourtselves.
+		cid, _ := nc.GetClientID()
+		sid := nc.ConnectedServerId()
+
+		wt := time.NewTimer(200 * time.Millisecond)
+		var conns []*ConnInfo
+	LOOP:
+		for {
+			select {
+			case m := <-mch:
+				if len(m.Data) == 0 {
+					t.Fatalf("No responders")
+				}
+				cr := parseConnz(m.Data)
+				// For account scoped, NumConns and Total should be the same (sans limits and offsets).
+				// It Total should not include other accounts since that would leak information about the system.
+				if filter == _EMPTY_ && cr.NumConns != cr.Total {
+					t.Fatalf("NumConns and Total should be same with account scoped connz, got %+v", cr)
+				}
+				for _, c := range cr.Conns {
+					if c.Name != acc {
+						t.Fatalf("Got wrong account: %q vs %q", acc, c.Account)
+					}
+					if !(c.Cid == cid && cr.ID == sid) {
+						conns = append(conns, c)
+					}
+				}
+				wt.Reset(200 * time.Millisecond)
+			case <-wt.C:
+				break LOOP
+			}
+		}
+		if len(conns) != expected {
+			t.Fatalf("Expected to see %d conns but got %d", expected, len(conns))
+		}
+	}
+
+	doSysRequest := func(acc string, expected int) {
+		t.Helper()
+		doRequest("$SYS.REQ.SERVER.PING.CONNZ", acc, _EMPTY_, expected)
+	}
+	doAccRequest := func(acc string, expected int) {
+		t.Helper()
+		doRequest("$SYS.REQ.ACCOUNT.PING.CONNZ", acc, _EMPTY_, expected)
+	}
+	doFiltered := func(acc, filter string, expected int) {
+		t.Helper()
+		doRequest("$SYS.REQ.SERVER.PING.CONNZ", acc, filter, expected)
+	}
+
+	doSysRequest("one", 20)
+	doAccRequest("one", 20)
+
+	doSysRequest("two", 20)
+	doAccRequest("two", 20)
+
+	// Now check filtering.
+	doFiltered("one", _EMPTY_, 20)
+	doFiltered("one", ">", 20)
+	doFiltered("one", "bar", 10)
+	doFiltered("two", "bar", 0)
+	doFiltered("two", "id.1", 1)
+	doFiltered("two", "id.*", 20)
+	doFiltered("two", "foo.bar.*", 20)
+	doFiltered("two", "foo.>", 20)
+}
+
 func TestNoRaceJetStreamClusterExtendedStreamPurge(t *testing.T) {
 	for _, st := range []StorageType{FileStorage, MemoryStorage} {
 		t.Run(st.String(), func(t *testing.T) {
diff --git a/server/routes_test.go b/server/routes_test.go
index be08a9bc..4c1771a2 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -1095,8 +1095,8 @@ func TestRouteNoCrashOnAddingSubToRoute(t *testing.T) {
 
 	// Make sure all subs are registered in s.
 	checkFor(t, time.Second, 15*time.Millisecond, func() error {
-		if s.globalAccount().TotalSubs() != int(numRoutes) {
-			return fmt.Errorf("Not all %v routed subs were registered", numRoutes)
+		if ts := s.globalAccount().TotalSubs() - 2; ts != int(numRoutes) {
+			return fmt.Errorf("Not all %d routed subs were registered: %d", numRoutes, ts)
 		}
 		return nil
 	})
diff --git a/test/gateway_test.go b/test/gateway_test.go
index efef6065..96948720 100644
--- a/test/gateway_test.go
+++ b/test/gateway_test.go
@@ -511,7 +511,7 @@ func TestGatewayNoAccUnsubAfterQSub(t *testing.T) {
 	// Simulate a client connecting to A and publishing a message
 	// so we get an A- from B since there is no interest.
 	gASend("RMSG $G foo 2\r\nok\r\n")
-	gAExpect(aunsubRe)
+	gAExpect(runsubRe)
 
 	// Now create client on B and create queue sub.
 	client := createClientConn(t, ob.Host, ob.Port)
diff --git a/test/leafnode_test.go b/test/leafnode_test.go
index 7207d131..03c088cf 100644
--- a/test/leafnode_test.go
+++ b/test/leafnode_test.go
@@ -440,7 +440,7 @@ func TestLeafNodeAndRoutes(t *testing.T) {
 	lc := createLeafConn(t, optsA.LeafNode.Host, optsA.LeafNode.Port)
 	defer lc.Close()
 
-	leafSend, leafExpect := setupLeaf(t, lc, 1)
+	leafSend, leafExpect := setupLeaf(t, lc, 3)
 	leafSend("PING\r\n")
 	leafExpect(pongRe)
 
@@ -834,7 +834,7 @@ func TestLeafNodeGatewaySendsSystemEvent(t *testing.T) {
 	defer lc.Close()
 
 	// This is for our global responses since we are setting up GWs above.
-	leafSend, leafExpect := setupLeaf(t, lc, 3)
+	leafSend, leafExpect := setupLeaf(t, lc, 5)
 	leafSend("PING\r\n")
 	leafExpect(pongRe)
 
@@ -878,14 +878,14 @@ func TestLeafNodeGatewayInterestPropagation(t *testing.T) {
 	buf := leafExpect(infoRe)
 	buf = infoRe.ReplaceAll(buf, []byte(nil))
 	foundFoo := false
-	for count := 0; count != 5; {
+	for count := 0; count != 7; {
 		// skip first time if we still have data (buf from above may already have some left)
 		if count != 0 || len(buf) == 0 {
 			buf = append(buf, leafExpect(anyRe)...)
 		}
 		count += len(lsubRe.FindAllSubmatch(buf, -1))
-		if count > 5 {
-			t.Fatalf("Expected %v matches, got %v (buf=%s)", 4, count, buf)
+		if count > 7 {
+			t.Fatalf("Expected %v matches, got %v (buf=%s)", 7, count, buf)
 		}
 		if strings.Contains(string(buf), "foo") {
 			foundFoo = true
@@ -937,7 +937,7 @@ func TestLeafNodeWithRouteAndGateway(t *testing.T) {
 	defer lc.Close()
 
 	// This is for our global responses since we are setting up GWs above.
-	leafSend, leafExpect := setupLeaf(t, lc, 3)
+	leafSend, leafExpect := setupLeaf(t, lc, 5)
 	leafSend("PING\r\n")
 	leafExpect(pongRe)
 
@@ -996,7 +996,7 @@ func TestLeafNodeWithGatewaysAndStaggeredStart(t *testing.T) {
 	lc := createLeafConn(t, opts.LeafNode.Host, opts.LeafNode.Port)
 	defer lc.Close()
 
-	leafSend, leafExpect := setupLeaf(t, lc, 3)
+	leafSend, leafExpect := setupLeaf(t, lc, 5)
 	leafSend("PING\r\n")
 	leafExpect(pongRe)
 
@@ -1036,7 +1036,7 @@ func TestLeafNodeWithGatewaysServerRestart(t *testing.T) {
 	lc := createLeafConn(t, opts.LeafNode.Host, opts.LeafNode.Port)
 	defer lc.Close()
 
-	leafSend, leafExpect := setupLeaf(t, lc, 3)
+	leafSend, leafExpect := setupLeaf(t, lc, 5)
 	leafSend("PING\r\n")
 	leafExpect(pongRe)
 
@@ -1070,7 +1070,7 @@ func TestLeafNodeWithGatewaysServerRestart(t *testing.T) {
 	lc = createLeafConn(t, opts.LeafNode.Host, opts.LeafNode.Port)
 	defer lc.Close()
 
-	_, leafExpect = setupLeaf(t, lc, 3)
+	_, leafExpect = setupLeaf(t, lc, 5)
 
 	// Now wait on GW solicit to fire
 	time.Sleep(500 * time.Millisecond)
@@ -1174,7 +1174,7 @@ func TestLeafNodeBasicAuth(t *testing.T) {
 	leafExpect(infoRe)
 	leafExpect(lsubRe)
 	leafSend("PING\r\n")
-	leafExpect(pongRe)
+	expectResult(t, lc, pongRe)
 
 	checkLeafNodeConnected(t, s)
 }
@@ -1576,7 +1576,7 @@ func TestLeafNodeMultipleAccounts(t *testing.T) {
 
 	// Wait for the subs to propagate. LDS + foo.test
 	checkFor(t, 2*time.Second, 10*time.Millisecond, func() error {
-		if subs := s.NumSubscriptions(); subs < 2 {
+		if subs := s.NumSubscriptions(); subs < 4 {
 			return fmt.Errorf("Number of subs is %d", subs)
 		}
 		return nil
@@ -1760,7 +1760,7 @@ func TestLeafNodeExportsImports(t *testing.T) {
 
 	// Wait for all subs to propagate.
 	checkFor(t, time.Second, 10*time.Millisecond, func() error {
-		if subs := s.NumSubscriptions(); subs < 3 {
+		if subs := s.NumSubscriptions(); subs < 5 {
 			return fmt.Errorf("Number of subs is %d", subs)
 		}
 		return nil
@@ -1924,8 +1924,8 @@ func TestLeafNodeExportImportComplexSetup(t *testing.T) {
 
 	// Wait for the sub to propagate to s2. LDS + subject above.
 	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-		if acc1.RoutedSubs() != 2 {
-			return fmt.Errorf("Still no routed subscription")
+		if acc1.RoutedSubs() != 4 {
+			return fmt.Errorf("Still no routed subscription: %d", acc1.RoutedSubs())
 		}
 		return nil
 	})
@@ -2510,7 +2510,7 @@ func TestLeafNodeSwitchGatewayToInterestModeOnly(t *testing.T) {
 	defer lc.Close()
 
 	// This is for our global responses since we are setting up GWs above.
-	leafSend, leafExpect := setupLeaf(t, lc, 3)
+	leafSend, leafExpect := setupLeaf(t, lc, 5)
 	leafSend("PING\r\n")
 	leafExpect(pongRe)
 }