Repository: nats-io/nats-server
Pull request #1830 (2021-01-21 00:03:24)

[added] jwt/issuerkey/nametag/tags to monitoring and event endpoints
Also added a trace on jwt authentication

Signed-off-by: Matthias Hanel <mh@synadia.com>
```
[53472] [TRC] 127.0.0.1:51931 - cid:3 - "v1.11.0:go" - Authenticated JWT: Client "UAW2WHOE5AKJHEWVHJJMDWSVIN4BHU2HGNFKS4ZL2JOZ5RN774CP3ZVF" (claim-name: "a name", claim-tags: ["tags" "other"]) signed with "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" by Account "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" (claim-name: "", claim-tags: []) signed with "OCAT33MTVU2VUOIMGNGUNXJ66AH2RLSDAF3MUBCYAY5QMIL65NQM6XQG"
[53472] [TRC] 127.0.0.1:51932 - cid:4 - "v1.11.0:go" - Authenticated JWT: Client "UAW2WHOE5AKJHEWVHJJMDWSVIN4BHU2HGNFKS4ZL2JOZ5RN774CP3ZVF" (claim-name: "a name", claim-tags: ["tags" "other"]) signed with "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" by Account "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" (claim-name: "some account", claim-tags: ["foo" "bar"]) signed with "OCAT33MTVU2VUOIMGNGUNXJ66AH2RLSDAF3MUBCYAY5QMIL65NQM6XQG"
[53472] [TRC] 127.0.0.1:51933 - cid:5 - "v1.11.0:go:conn name" - Authenticated JWT: Client "UAW2WHOE5AKJHEWVHJJMDWSVIN4BHU2HGNFKS4ZL2JOZ5RN774CP3ZVF" (claim-name: "a name", claim-tags: ["tags" "other"]) signed with "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" by Account "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" (claim-name: "some account", claim-tags: ["foo" "bar"]) signed with "OCAT33MTVU2VUOIMGNGUNXJ66AH2RLSDAF3MUBCYAY5QMIL65NQM6XQG"
[53472] [TRC] 127.0.0.1:51934 - cid:6 - "v1.11.0:go" - Authenticated JWT: Client "UABC2XJEOWPA4DDDYKIDVS24SOLYTCILDFCHAGDG3JNS3RR6O77ZAJK3" (claim-name: "a name", claim-tags: ["tags" "other"]) signed with "AD7EMU4HIRDKAUIY5KURVIY4BEECIBJY4E6FQEL7OBVOGXZHYPXHLTAF" by Account "AD7EMU4HIRDKAUIY5KURVIY4BEECIBJY4E6FQEL7OBVOGXZHYPXHLTAF" (claim-name: "some account", claim-tags: ["foo" "bar"]) signed with "OCAT33MTVU2VUOIMGNGUNXJ66AH2RLSDAF3MUBCYAY5QMIL65NQM6XQG"
[53472] [TRC] 127.0.0.1:51935 - cid:7 - "v1.11.0:go" - Authenticated JWT: Client "UAW2WHOE5AKJHEWVHJJMDWSVIN4BHU2HGNFKS4ZL2JOZ5RN774CP3ZVF" (claim-name: "a name", claim-tags: ["tags" "other"]) signed with "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" by Account "ACMAPDNT4LQZU6GYD22JVXLUZJT6AK4ONHKIB4TSJEAUHAEOONFLY22R" (claim-name: "some account", claim-tags: ["foo" "bar"]) signed with "OCAT33MTVU2VUOIMGNGUNXJ66AH2RLSDAF3MUBCYAY5QMIL65NQM6XQG"
[53472] [TRC] 127.0.0.1:51937 - cid:9 - "v1.11.0:go" - Authenticated JWT: Client "UABC2XJEOWPA4DDDYKIDVS24SOLYTCILDFCHAGDG3JNS3RR6O77ZAJK3" (claim-name: "a name", claim-tags: ["tags" "other"]) signed with "AD7EMU4HIRDKAUIY5KURVIY4BEECIBJY4E6FQEL7OBVOGXZHYPXHLTAF" by Account "AD7EMU4HIRDKAUIY5KURVIY4BEECIBJY4E6FQEL7OBVOGXZHYPXHLTAF" (claim-name: "some account", claim-tags: ["foo" "bar"]) signed with "OCAT33MTVU2VUOIMGNGUNXJ66AH2RLSDAF3MUBCYAY5QMIL65NQM6XQG"
```

matthiashanel: > I think setting fields in the client during auth should be protected.
> Also, I wonder if this needs separate configuration to be activated? Not sure if in the use case this came up from, user would be willing to run under `-V` (which generates a lot of output) in order to get this trace. Also wonder if there is anything sensitive in this trace, which would make the opt-in option even more necessary.

I spoke to them and for now I'll keep it at debug, hoping that the system eventing stuff is good enough for them.
If it's not, they'll let me know. then we can talk about possibly adding a switch for this trace.

Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 880bb1fb..388580b9 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -82,6 +82,8 @@ type Account struct {
 	eventIds     *nuid.NUID
 	eventIdsMu   sync.Mutex
 	defaultPerms *Permissions
+	tags         jwt.TagList
+	nameTag      string
 }
 
 // Account based limits.
@@ -2813,6 +2815,10 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 	// Clone to update, only select certain fields.
 	old := &Account{Name: a.Name, exports: a.exports, limits: a.limits, signingKeys: a.signingKeys}
 
+	// overwrite claim meta data
+	a.nameTag = ac.Name
+	a.tags = ac.Tags
+
 	// Reset exports and imports here.
 
 	// Exports is creating a whole new map.
diff --git a/server/auth.go b/server/auth.go
index 52529bdf..6265dfcd 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -627,13 +627,23 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 			return false
 		}
 		// Hold onto the user's public key.
+		c.mu.Lock()
 		c.pubKey = juc.Subject
+		c.tags = juc.Tags
+		c.nameTag = juc.Name
+		c.mu.Unlock()
 
 		// Generate an event if we have a system account.
 		s.accountConnectEvent(c)
 
 		// Check if we need to set an auth timer if the user jwt expires.
 		c.setExpiration(juc.Claims(), validFor)
+
+		acc.mu.RLock()
+		c.Debugf("Authenticated JWT: %s %q (claim-name: %q, claim-tags: %q) "+
+			"signed with %q by Account %q (claim-name: %q, claim-tags: %q) signed with %q",
+			c.typeString(), juc.Subject, juc.Name, juc.Tags, juc.Issuer, issuer, acc.nameTag, acc.tags, acc.Issuer)
+		acc.mu.RUnlock()
 		return true
 	}
 
diff --git a/server/client.go b/server/client.go
index dac725e4..2c09f55b 100644
--- a/server/client.go
+++ b/server/client.go
@@ -263,6 +263,9 @@ type client struct {
 
 	trace bool
 	echo  bool
+
+	tags    jwt.TagList
+	nameTag string
 }
 
 type rrTracking struct {
@@ -4662,6 +4665,10 @@ func (c *client) getClientInfo(detailed bool) *ClientInfo {
 		ci.Lang = c.opts.Lang
 		ci.Version = c.opts.Version
 		ci.Server = sn
+		ci.Jwt = c.opts.JWT
+		ci.IssuerKey = issuerForClient(c)
+		ci.NameTag = c.nameTag
+		ci.Tags = c.tags
 	}
 	c.mu.Unlock()
 	return &ci
diff --git a/server/events.go b/server/events.go
index 2a76192a..a0fb1e4d 100644
--- a/server/events.go
+++ b/server/events.go
@@ -163,17 +163,21 @@ type ServerInfo struct {
 
 // ClientInfo is detailed information about the client forming a connection.
 type ClientInfo struct {
-	Start   *time.Time    `json:"start,omitempty"`
-	Host    string        `json:"host,omitempty"`
-	ID      uint64        `json:"id,omitempty"`
-	Account string        `json:"acc"`
-	User    string        `json:"user,omitempty"`
-	Name    string        `json:"name,omitempty"`
-	Lang    string        `json:"lang,omitempty"`
-	Version string        `json:"ver,omitempty"`
-	RTT     time.Duration `json:"rtt,omitempty"`
-	Server  string        `json:"server,omitempty"`
-	Stop    *time.Time    `json:"stop,omitempty"`
+	Start     *time.Time    `json:"start,omitempty"`
+	Host      string        `json:"host,omitempty"`
+	ID        uint64        `json:"id,omitempty"`
+	Account   string        `json:"acc"`
+	User      string        `json:"user,omitempty"`
+	Name      string        `json:"name,omitempty"`
+	Lang      string        `json:"lang,omitempty"`
+	Version   string        `json:"ver,omitempty"`
+	RTT       time.Duration `json:"rtt,omitempty"`
+	Server    string        `json:"server,omitempty"`
+	Stop      *time.Time    `json:"stop,omitempty"`
+	Jwt       string        `json:"jwt,omitempty"`
+	IssuerKey string        `json:"issuer_key,omitempty"`
+	NameTag   string        `json:"name_tag,omitempty"`
+	Tags      jwt.TagList   `json:"tags,omitempty"`
 }
 
 // ServerStats hold various statistics that we will periodically send out.
@@ -957,9 +961,10 @@ func (s *Server) leafNodeConnected(sub *subscription, _ *client, subject, reply
 
 // Common filter options for system requests STATSZ VARZ SUBSZ CONNZ ROUTEZ GATEWAYZ LEAFZ
 type EventFilterOptions struct {
-	Name    string `json:"server_name,omitempty"` // filter by server name
-	Cluster string `json:"cluster,omitempty"`     // filter by cluster name
-	Host    string `json:"host,omitempty"`        // filter by host name
+	Name    string   `json:"server_name,omitempty"` // filter by server name
+	Cluster string   `json:"cluster,omitempty"`     // filter by cluster name
+	Host    string   `json:"host,omitempty"`        // filter by host name
+	Tags    []string `json:"tags,omitempty"`        // filter by tags (must match all tags)
 }
 
 // StatszEventOptions are options passed to Statsz
@@ -1033,6 +1038,14 @@ func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
 			return true
 		}
 	}
+	if len(fOpts.Tags) > 0 {
+		opts := s.getOpts()
+		for _, t := range fOpts.Tags {
+			if !opts.Tags.Contains(t) {
+				return true
+			}
+		}
+	}
 	return false
 }
 
@@ -1268,14 +1281,18 @@ func (s *Server) accountConnectEvent(c *client) {
 			Time: time.Now().UTC(),
 		},
 		Client: ClientInfo{
-			Start:   &c.start,
-			Host:    c.host,
-			ID:      c.cid,
-			Account: accForClient(c),
-			User:    c.getRawAuthUser(),
-			Name:    c.opts.Name,
-			Lang:    c.opts.Lang,
-			Version: c.opts.Version,
+			Start:     &c.start,
+			Host:      c.host,
+			ID:        c.cid,
+			Account:   accForClient(c),
+			User:      c.getRawAuthUser(),
+			Name:      c.opts.Name,
+			Lang:      c.opts.Lang,
+			Version:   c.opts.Version,
+			Jwt:       c.opts.JWT,
+			IssuerKey: issuerForClient(c),
+			Tags:      c.tags,
+			NameTag:   c.nameTag,
 		},
 	}
 	c.mu.Unlock()
@@ -1311,16 +1328,20 @@ func (s *Server) accountDisconnectEvent(c *client, now time.Time, reason string)
 			Time: now.UTC(),
 		},
 		Client: ClientInfo{
-			Start:   &c.start,
-			Stop:    &now,
-			Host:    c.host,
-			ID:      c.cid,
-			Account: accForClient(c),
-			User:    c.getRawAuthUser(),
-			Name:    c.opts.Name,
-			Lang:    c.opts.Lang,
-			Version: c.opts.Version,
-			RTT:     c.getRTT(),
+			Start:     &c.start,
+			Stop:      &now,
+			Host:      c.host,
+			ID:        c.cid,
+			Account:   accForClient(c),
+			User:      c.getRawAuthUser(),
+			Name:      c.opts.Name,
+			Lang:      c.opts.Lang,
+			Version:   c.opts.Version,
+			RTT:       c.getRTT(),
+			Jwt:       c.opts.JWT,
+			IssuerKey: issuerForClient(c),
+			Tags:      c.tags,
+			NameTag:   c.nameTag,
 		},
 		Sent: DataStats{
 			Msgs:  atomic.LoadInt64(&c.inMsgs),
@@ -1356,16 +1377,20 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 			Time: now.UTC(),
 		},
 		Client: ClientInfo{
-			Start:   &c.start,
-			Stop:    &now,
-			Host:    c.host,
-			ID:      c.cid,
-			Account: accForClient(c),
-			User:    c.getRawAuthUser(),
-			Name:    c.opts.Name,
-			Lang:    c.opts.Lang,
-			Version: c.opts.Version,
-			RTT:     c.getRTT(),
+			Start:     &c.start,
+			Stop:      &now,
+			Host:      c.host,
+			ID:        c.cid,
+			Account:   accForClient(c),
+			User:      c.getRawAuthUser(),
+			Name:      c.opts.Name,
+			Lang:      c.opts.Lang,
+			Version:   c.opts.Version,
+			RTT:       c.getRTT(),
+			Jwt:       c.opts.JWT,
+			IssuerKey: issuerForClient(c),
+			Tags:      c.tags,
+			NameTag:   c.nameTag,
 		},
 		Sent: DataStats{
 			Msgs:  c.inMsgs,
@@ -1751,6 +1776,18 @@ func accForClient(c *client) string {
 	return "N/A"
 }
 
+// Helper to grab issuer for a client.
+func issuerForClient(c *client) (issuerKey string) {
+	if c == nil || c.user == nil {
+		return
+	}
+	issuerKey = c.user.SigningKey
+	if issuerKey == "" && c.user.Account != nil {
+		issuerKey = c.user.Account.Name
+	}
+	return
+}
+
 // Helper to clear timers.
 func clearTimer(tp **time.Timer) {
 	if t := *tp; t != nil {
diff --git a/server/monitor.go b/server/monitor.go
index 22b12fd3..6319f6fd 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -127,6 +127,10 @@ type ConnInfo struct {
 	Account        string      `json:"account,omitempty"`
 	Subs           []string    `json:"subscriptions_list,omitempty"`
 	SubsDetail     []SubDetail `json:"subscriptions_list_detail,omitempty"`
+	JWT            string      `json:"jwt,omitempty"`
+	IssuerKey      string      `json:"issuer_key,omitempty"`
+	NameTag        string      `json:"name_tag,omitempty"`
+	Tags           jwt.TagList `json:"tags,omitempty"`
 }
 
 // DefaultConnListSize is the default size of the connection list.
@@ -334,6 +338,10 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 			if client.acc != nil && (client.acc.Name != globalAccountName) {
 				ci.Account = client.acc.Name
 			}
+			ci.JWT = client.opts.JWT
+			ci.IssuerKey = issuerForClient(client)
+			ci.Tags = client.tags
+			ci.NameTag = client.nameTag
 		}
 		client.mu.Unlock()
 		pconns[i] = ci
@@ -1036,6 +1044,7 @@ type Varz struct {
 	Subscriptions     uint32            `json:"subscriptions"`
 	HTTPReqStats      map[string]uint64 `json:"http_req_stats"`
 	ConfigLoadTime    time.Time         `json:"config_load_time"`
+	Tags              jwt.TagList       `json:"tags,omitempty"`
 }
 
 // JetStreamVarz contains basic runtime information about jetstream
@@ -1228,6 +1237,7 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 		MaxSubs:  opts.MaxSubs,
 		Cores:    numCores,
 		MaxProcs: maxProcs,
+		Tags:     opts.Tags,
 	}
 	if len(opts.Routes) > 0 {
 		varz.Cluster.URLs = urlsToStrings(opts.Routes)
@@ -1974,6 +1984,9 @@ type AccountInfo struct {
 	Exports     []ExtExport          `json:"exports,omitempty"`
 	Imports     []ExtImport          `json:"imports,omitempty"`
 	Jwt         string               `json:"jwt,omitempty"`
+	IssuerKey   string               `json:"issuer_key,omitempty"`
+	NameTag     string               `json:"name_tag,omitempty"`
+	Tags        jwt.TagList          `json:"tags,omitempty"`
 	Claim       *jwt.AccountClaims   `json:"decoded_jwt,omitempty"`
 	Vr          []ExtVrIssues        `json:"validation_result_jwt,omitempty"`
 	RevokedUser map[string]time.Time `json:"revoked_user,omitempty"`
@@ -2172,6 +2185,9 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 		exports,
 		imports,
 		a.claimJWT,
+		a.Issuer,
+		a.nameTag,
+		a.tags,
 		claim,
 		vrIssues,
 		collectRevocations(a.usersRevoked),
diff --git a/server/opts.go b/server/opts.go
index 915e01d3..9330a21a 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -247,6 +247,10 @@ type Options struct {
 	// that this applies to reconnect events.
 	ReconnectErrorReports int
 
+	// Tags describing the server. They will be included in varz
+	// and used as a filter criteria for some system requests
+	Tags jwt.TagList `json:"-"`
+
 	// private fields, used to know if bool options are explicitly
 	// defined in config and/or command line params.
 	inConfig  map[string]bool
@@ -1060,6 +1064,34 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 			*errors = append(*errors, err)
 			return
 		}
+	case "server_tags":
+		var err error
+		switch v := v.(type) {
+		case string:
+			o.Tags.Add(v)
+		case []string:
+			o.Tags.Add(v...)
+		case []interface{}:
+			for _, t := range v {
+				if t, ok := t.(token); ok {
+					if t, ok := t.Value().(string); ok {
+						o.Tags.Add(t)
+						continue
+					} else {
+						err = &configErr{tk, fmt.Sprintf("error parsing tags: unsupported type %T where string is expected", t)}
+					}
+				} else {
+					err = &configErr{tk, fmt.Sprintf("error parsing tags: unsupported type %T", t)}
+				}
+				break
+			}
+		default:
+			err = &configErr{tk, fmt.Sprintf("error parsing tags: unsupported type %T", v)}
+		}
+		if err != nil {
+			*errors = append(*errors, err)
+			return
+		}
 	default:
 		if au := atomic.LoadInt32(&allowUnknownTopLevelField); au == 0 && !tk.IsUsedVariable() {
 			err := &unknownConfigFieldErr{
diff --git a/server/reload.go b/server/reload.go
index a5339920..741452a5 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -788,8 +788,8 @@ func imposeOrder(value interface{}) error {
 		})
 	case WebsocketOpts:
 		sort.Strings(value.AllowedOrigins)
-	case string, bool, int, int32, int64, time.Duration, float64, nil,
-		LeafNodeOpts, ClusterOpts, *tls.Config, *URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts:
+	case string, bool, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config,
+		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList:
 		// explicitly skipped types
 	default:
 		// this will fail during unit tests
diff --git a/server/server.go b/server/server.go
index d1307f43..a1b15b37 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2343,6 +2343,10 @@ func (s *Server) saveClosedClient(c *client, nc net.Conn, reason ClosedState) {
 	if c.acc != nil && c.acc.Name != globalAccountName {
 		cc.acc = c.acc.Name
 	}
+	cc.JWT = c.opts.JWT
+	cc.IssuerKey = issuerForClient(c)
+	cc.Tags = c.tags
+	cc.NameTag = c.nameTag
 	c.mu.Unlock()
 
 	// Place in the ring buffer


Test patch:
diff --git a/server/dirstore_test.go b/server/dirstore_test.go
index e47dfef8..8d0d43c0 100644
--- a/server/dirstore_test.go
+++ b/server/dirstore_test.go
@@ -54,6 +54,15 @@ func require_NoError(t *testing.T, err error) {
 	}
 }
 
+func require_Contains(t *testing.T, s string, subStrs ...string) {
+	t.Helper()
+	for _, subStr := range subStrs {
+		if !strings.Contains(s, subStr) {
+			t.Fatalf("require %q to be contained in %q", subStr, s)
+		}
+	}
+}
+
 func require_Error(t *testing.T, err error) {
 	t.Helper()
 	if err == nil {
diff --git a/server/events_test.go b/server/events_test.go
index 7d5bcca2..0b9874ff 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -2178,3 +2178,90 @@ func TestServerEventsReceivedByQSubs(t *testing.T) {
 		t.Fatalf("Expected auth error, got %q", dem.Reason)
 	}
 }
+
+func TestServerEventsFilteredByTag(t *testing.T) {
+	confA := createConfFile(t, []byte(`
+		listen: -1
+		server_name: srv-A
+		server_tags: ["foo", "bar"]
+		cluster {
+			name: clust
+			listen: -1
+			no_advertise: true
+		}
+		system_account: SYS
+		accounts: {
+			SYS: {
+				users: [
+					{user: b, password: b}
+				]
+			}
+		}
+		no_auth_user: b
+    `))
+	defer os.Remove(confA)
+	sA, _ := RunServerWithConfig(confA)
+	defer sA.Shutdown()
+	confB := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: -1
+		server_name: srv-B
+		server_tags: ["bar", "baz"]
+		cluster {
+			name: clust
+			listen: -1
+			no_advertise: true
+			routes [
+				nats-route://localhost:%d
+			]
+		}
+		system_account: SYS
+		accounts: {
+			SYS: {
+				users: [
+					{user: b, password: b}
+				]
+			}
+		}
+		no_auth_user: b
+    `, sA.opts.Cluster.Port)))
+	defer os.Remove(confB)
+	sB, _ := RunServerWithConfig(confB)
+	defer sB.Shutdown()
+	checkClusterFormed(t, sA, sB)
+	nc := natsConnect(t, sA.ClientURL())
+	defer nc.Close()
+
+	ib := nats.NewInbox()
+	req := func(tags ...string) {
+		t.Helper()
+		r, err := json.Marshal(VarzEventOptions{EventFilterOptions: EventFilterOptions{Tags: tags}})
+		require_NoError(t, err)
+		err = nc.PublishRequest(fmt.Sprintf(serverPingReqSubj, "VARZ"), ib, r)
+		require_NoError(t, err)
+	}
+
+	msgs := make(chan *nats.Msg, 10)
+	defer close(msgs)
+	_, err := nc.ChanSubscribe(ib, msgs)
+	require_NoError(t, err)
+	req("none")
+	select {
+	case <-msgs:
+		t.Fatalf("no message expected")
+	case <-time.After(200 * time.Millisecond):
+	}
+	req("foo")
+	m := <-msgs
+	require_Contains(t, string(m.Data), "srv-A", "foo", "bar")
+	req("foo", "bar")
+	m = <-msgs
+	require_Contains(t, string(m.Data), "srv-A", "foo", "bar")
+	req("baz")
+	m = <-msgs
+	require_Contains(t, string(m.Data), "srv-B", "bar", "baz")
+	req("bar")
+	m1 := <-msgs
+	m2 := <-msgs
+	require_Contains(t, string(m1.Data)+string(m2.Data), "srv-A", "srv-B", "foo", "bar", "baz")
+	require_Len(t, len(msgs), 0)
+}
diff --git a/server/jwt_test.go b/server/jwt_test.go
index cc0bfce3..739fbcc0 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -4431,6 +4431,11 @@ func TestJWTUserRevocation(t *testing.T) {
 	defer srv.Shutdown()
 	updateJwt(t, srv.ClientURL(), sysCreds, sysjwt, 1) // update system account jwt
 	updateJwt(t, srv.ClientURL(), sysCreds, ajwt1, 1)  // set account jwt without revocation
+	ncSys := natsConnect(t, srv.ClientURL(), nats.UserCredentials(sysCreds), nats.Name("conn name"))
+	defer ncSys.Close()
+	ncChan := make(chan *nats.Msg, 10)
+	defer close(ncChan)
+	ncSys.ChanSubscribe(fmt.Sprintf(disconnectEventSubj, apub), ncChan) // observe disconnect message
 	// use credentials that will be revoked ans assure that the connection will be disconnected
 	nc := natsConnect(t, srv.ClientURL(), nats.UserCredentials(aCreds1),
 		nats.DisconnectErrHandler(func(conn *nats.Conn, err error) {
@@ -4440,7 +4445,7 @@ func TestJWTUserRevocation(t *testing.T) {
 		}))
 	defer nc.Close()
 	// update account jwt to contain revocation
-	if passCnt := updateJwt(t, srv.ClientURL(), sysCreds, ajwt2, 1); passCnt != 1 {
+	if updateJwt(t, srv.ClientURL(), sysCreds, ajwt2, 1) != 1 {
 		t.Fatalf("Expected jwt update to pass")
 	}
 	// assure that nc got disconnected due to the revocation
@@ -4449,6 +4454,9 @@ func TestJWTUserRevocation(t *testing.T) {
 	case <-time.After(time.Second):
 		t.Fatalf("Expected connection to have failed")
 	}
+	m := <-ncChan
+	require_Len(t, strings.Count(string(m.Data), apub), 2)
+	require_True(t, strings.Contains(string(m.Data), `"jwt": "eyJ0`))
 	// try again with old credentials. Expected to fail
 	if nc1, err := nats.Connect(srv.ClientURL(), nats.UserCredentials(aCreds1)); err == nil {
 		nc1.Close()
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 51dc8f58..3955b7b9 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -51,6 +51,7 @@ func DefaultMonitorOptions() *Options {
 		ServerName:   "monitor_server",
 		NoLog:        true,
 		NoSigs:       true,
+		Tags:         []string{"tag"},
 	}
 }
 
@@ -244,6 +245,9 @@ func TestHandleVarz(t *testing.T) {
 		if v.Name != "monitor_server" {
 			t.Fatal("Expected ServerName to be 'monitor_server'")
 		}
+		if !v.Tags.Contains("tag") {
+			t.Fatal("Expected tags to be 'tag'")
+		}
 	}
 
 	// Test JSONP