Repository: nats-io/nats-server
Pull request #1481 (2020-06-16 19:39:44)

Add filtering by name and cluster to PING events
Issue reset after cluster name changed. 
Add filtering by name and cluster to system events.

empty cluster name
```
 nats -s localhost:4141 --creds=$NATS_ADMIN_CREDS  req "\$SYS.REQ.SERVER.PING.VARZ" "{\"cluster\":\"\",\"subscriptions\":true}"
15:30:43 Sending request on [$SYS.REQ.SERVER.PING.VARZ]
15:30:43 Received on [_INBOX.SMi3XB4JfXyrhLMLK60lt2]: '{
  "data": {
    "server_id": "NBZEZ5V6J7N6MXVZPH7VOHFGW46NUMSJYXEDKZSBA5Y3USF4Z5J7SX74",
    "server_name": "NBZEZ5V6J7N6MXVZPH7VOHFGW46NUMSJYXEDKZSBA5Y3USF4Z5J7SX74",
    "version": "2.2.0-beta.15",
    "proto": 1,
    "go": "go1.14.1",
    "host": "localhost",
    "port": 4141,
    "auth_required": true,
    "max_connections": 65536,
    "ping_interval": 120000000000,
    "ping_max": 2,
    "http_host": "localhost",
    "http_port": 0,
    "http_base_path": "",
    "https_port": 0,
    "auth_timeout": 1,
    "max_control_line": 4096,
    "max_payload": 1048576,
    "max_pending": 67108864,
    "cluster": {
      "name": "testcluster"
    },
    "gateway": {},
    "leaf": {},
    "jetstream": {},
    "tls_timeout": 0.5,
    "write_deadline": 2000000000,
    "start": "2020-06-16T15:30:41.518627-04:00",
    "now": "2020-06-16T15:30:43.46485-04:00",
    "uptime": "1s",
    "mem": 8835072,
    "cores": 16,
    "gomaxprocs": 16,
    "cpu": 0,
    "connections": 1,
    "total_connections": 1,
    "routes": 0,
    "remotes": 0,
    "leafnodes": 0,
    "in_msgs": 0,
    "out_msgs": 0,
    "in_bytes": 0,
    "out_bytes": 0,
    "slow_consumers": 0,
    "subscriptions": 0,
    "http_req_stats": {},
    "config_load_time": "2020-06-16T15:30:41.518627-04:00"
  },
  "server": {
    "name": "NBZEZ5V6J7N6MXVZPH7VOHFGW46NUMSJYXEDKZSBA5Y3USF4Z5J7SX74",
    "host": "localhost",
    "id": "NBZEZ5V6J7N6MXVZPH7VOHFGW46NUMSJYXEDKZSBA5Y3USF4Z5J7SX74",
    "cluster": "testcluster",
    "ver": "2.2.0-beta.15",
    "seq": 8,
    "jetstream": false,
    "time": "2020-06-16T15:30:43.464889-04:00"
  }
}'
```
matching cluster name
```
nats -s localhost:4141 --creds=$NATS_ADMIN_CREDS  req "\$SYS.REQ.SERVER.PING.VARZ" "{\"cluster\":\"test\",\"subscriptions\":true}"
15:34:39 Sending request on [$SYS.REQ.SERVER.PING.VARZ]
15:34:39 Received on [_INBOX.iTNqmEGarJYdshuXRGIgwQ]: '{
  "data": {
    "server_id": "NDAWGFQUNDC3AB5RANCCNL5NK6SPHUDANUQAQLIPMGLDDEJJOQI7R7WR",
    "server_name": "NDAWGFQUNDC3AB5RANCCNL5NK6SPHUDANUQAQLIPMGLDDEJJOQI7R7WR",
    "version": "2.2.0-beta.15",
    "proto": 1,
    "go": "go1.14.1",
    "host": "localhost",
    "port": 4141,
    "auth_required": true,
    "max_connections": 65536,
    "ping_interval": 120000000000,
    "ping_max": 2,
    "http_host": "localhost",
    "http_port": 0,
    "http_base_path": "",
    "https_port": 0,
    "auth_timeout": 1,
    "max_control_line": 4096,
    "max_payload": 1048576,
    "max_pending": 67108864,
    "cluster": {
      "name": "testcluster"
    },
    "gateway": {},
    "leaf": {},
    "jetstream": {},
    "tls_timeout": 0.5,
    "write_deadline": 2000000000,
    "start": "2020-06-16T15:34:37.454438-04:00",
    "now": "2020-06-16T15:34:39.372401-04:00",
    "uptime": "1s",
    "mem": 9154560,
    "cores": 16,
    "gomaxprocs": 16,
    "cpu": 0,
    "connections": 1,
    "total_connections": 1,
    "routes": 0,
    "remotes": 0,
    "leafnodes": 0,
    "in_msgs": 0,
    "out_msgs": 0,
    "in_bytes": 0,
    "out_bytes": 0,
    "slow_consumers": 0,
    "subscriptions": 0,
    "http_req_stats": {},
    "config_load_time": "2020-06-16T15:34:37.454438-04:00"
  },
  "server": {
    "name": "NDAWGFQUNDC3AB5RANCCNL5NK6SPHUDANUQAQLIPMGLDDEJJOQI7R7WR",
    "host": "localhost",
    "id": "NDAWGFQUNDC3AB5RANCCNL5NK6SPHUDANUQAQLIPMGLDDEJJOQI7R7WR",
    "cluster": "testcluster",
    "ver": "2.2.0-beta.15",
    "seq": 8,
    "jetstream": false,
    "time": "2020-06-16T15:34:39.372457-04:00"
  }
}'
```

unused cluster name
```
 nats -s localhost:4141 --creds=$NATS_ADMIN_CREDS  req "\$SYS.REQ.SERVER.PING.VARZ" "{\"cluster\":\"sdf\",\"subscriptions\":true}"
15:30:48 Sending request on [$SYS.REQ.SERVER.PING.VARZ]
nats: error: nats: timeout, try --help
```
Signed-off-by: Matthias Hanel <mh@synadia.com>




Patch:
diff --git a/server/auth.go b/server/auth.go
index ad9de95b..9a568453 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -808,7 +808,7 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 		return false
 	}
 
-	// We are here if we accept leafnode connections without any credentials.
+	// We are here if we accept leafnode connections without any credential.
 
 	// Still, if the CONNECT has some user info, we will bind to the
 	// user's account or to the specified default account (if provided)
diff --git a/server/client.go b/server/client.go
index 2e625390..6d1728a1 100644
--- a/server/client.go
+++ b/server/client.go
@@ -440,7 +440,7 @@ type clientOpts struct {
 	Headers      bool   `json:"headers,omitempty"`
 	NoResponders bool   `json:"no_responders,omitempty"`
 
-	// Routes and Leafnodes only
+	// Routes only
 	Import *SubjectPermission `json:"import,omitempty"`
 	Export *SubjectPermission `json:"export,omitempty"`
 }
@@ -748,15 +748,6 @@ func (c *client) setPermissions(perms *Permissions) {
 			c.perms.sub.deny.Insert(sub)
 		}
 	}
-
-	// If we are a leafnode and we are the hub copy the extracted perms
-	// to resend back to soliciting server. These are reversed from the
-	// way routes interpret them since this is how the soliciting server
-	// will receive these back in an update INFO.
-	if c.isHubLeafNode() {
-		c.opts.Import = perms.Subscribe
-		c.opts.Export = perms.Publish
-	}
 }
 
 // Check to see if we have an expiration for the user JWT via base claims.
@@ -2707,14 +2698,6 @@ func (c *client) deliverMsg(sub *subscription, subject, mh, msg []byte, gwrply b
 		return false
 	}
 
-	// Check if we are a leafnode and have perms to check.
-	if client.kind == LEAF && client.perms != nil {
-		if !client.pubAllowed(string(subject)) {
-			client.mu.Unlock()
-			return false
-		}
-	}
-
 	srv := client.srv
 
 	sub.nm++
@@ -3820,7 +3803,7 @@ func (c *client) typeString() string {
 	case GATEWAY:
 		return "Gateway"
 	case LEAF:
-		return "Leafnode"
+		return "LeafNode"
 	case JETSTREAM:
 		return "JetStream"
 	case ACCOUNT:
diff --git a/server/const.go b/server/const.go
index 4daf250e..f590a3cc 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.16"
+	VERSION = "2.2.0-beta.15"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/events.go b/server/events.go
index dbb94cfb..5c19e507 100644
--- a/server/events.go
+++ b/server/events.go
@@ -242,7 +242,7 @@ RESET:
 	servername := s.info.Name
 	seqp := &s.sys.seq
 	js := s.js != nil
-	var cluster string
+	cluster := s.info.Cluster
 	if s.gateway.enabled {
 		cluster = s.getGatewayName()
 	}
@@ -832,14 +832,61 @@ func (s *Server) leafNodeConnected(sub *subscription, _ *client, subject, reply
 	}
 }
 
+// Common filter options for system requests STATSZ VARZ SUBSZ CONNZ ROUTEZ GATEWAYZ LEAFZ
+type ZFilterOptions struct {
+	Name    string `json:"name"`
+	Cluster string `json:"cluster"`
+	Host    string `json:"host"`
+}
+
+// returns true if the request does NOT apply to this server and can be ignored.
+// DO NOT hold the server lock when
+func (s *Server) filterRequest(msg []byte) (bool, error) {
+	if len(msg) == 0 {
+		return false, nil
+	}
+	var fOpts ZFilterOptions
+	err := json.Unmarshal(msg, &fOpts)
+	if err != nil {
+		return false, err
+	}
+	if fOpts.Name != "" && !strings.Contains(s.info.Name, fOpts.Name) {
+		return true, nil
+	}
+	if fOpts.Host != "" && !strings.Contains(s.info.Host, fOpts.Host) {
+		return true, nil
+	}
+	if fOpts.Cluster != "" {
+		s.mu.Lock()
+		cluster := s.info.Cluster
+		s.mu.Unlock()
+		if !strings.Contains(cluster, fOpts.Cluster) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 // statszReq is a request for us to respond with current statz.
 func (s *Server) statszReq(sub *subscription, _ *client, subject, reply string, msg []byte) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if !s.eventsEnabled() || reply == _EMPTY_ {
+	if !s.EventsEnabled() || reply == _EMPTY_ {
 		return
 	}
+	if ignore, err := s.filterRequest(msg); err != nil {
+		server := &ServerInfo{}
+		response := map[string]interface{}{"server": server}
+		response["error"] = map[string]interface{}{
+			"code":        http.StatusBadRequest,
+			"description": err.Error(),
+		}
+		s.sendInternalMsgLocked(reply, _EMPTY_, server, response)
+		return
+	} else if ignore {
+		return
+	}
+	s.mu.Lock()
 	s.sendStatsz(reply)
+	s.mu.Unlock()
 }
 
 func (s *Server) zReq(reply string, msg []byte, optz interface{}, respf func() (interface{}, error)) {
@@ -851,7 +898,12 @@ func (s *Server) zReq(reply string, msg []byte, optz interface{}, respf func() (
 	var err error
 	status := 0
 	if len(msg) != 0 {
-		err = json.Unmarshal(msg, optz)
+		filter := false
+		if filter, err = s.filterRequest(msg); filter {
+			return
+		} else if err == nil {
+			err = json.Unmarshal(msg, optz)
+		}
 		status = http.StatusBadRequest // status is only included on error, so record how far execution got
 	}
 	if err == nil {
diff --git a/server/leafnode.go b/server/leafnode.go
index dedd78a8..8ccbeacf 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -177,10 +177,10 @@ func newLeafNodeCfg(remote *RemoteLeafOpts) *leafNodeCfg {
 	if len(remote.DenyExports) > 0 || len(remote.DenyImports) > 0 {
 		perms := &Permissions{}
 		if len(remote.DenyExports) > 0 {
-			perms.Publish = &SubjectPermission{Deny: remote.DenyExports}
+			perms.Subscribe = &SubjectPermission{Deny: remote.DenyExports}
 		}
 		if len(remote.DenyImports) > 0 {
-			perms.Subscribe = &SubjectPermission{Deny: remote.DenyImports}
+			perms.Publish = &SubjectPermission{Deny: remote.DenyImports}
 		}
 		cfg.perms = perms
 	}
@@ -907,24 +907,6 @@ func (c *client) processLeafnodeInfo(info *Info) error {
 		c.updateLeafNodeURLs(info)
 	}
 
-	// Check to see if we have permissions updates here.
-	if info.Import != nil || info.Export != nil {
-		perms := &Permissions{
-			Publish:   info.Export,
-			Subscribe: info.Import,
-		}
-		// Check if we have local deny clauses that we need to merge.
-		if remote := c.leaf.remote; remote != nil {
-			if len(remote.DenyExports) > 0 {
-				perms.Publish.Deny = append(perms.Publish.Deny, remote.DenyExports...)
-			}
-			if len(remote.DenyImports) > 0 {
-				perms.Subscribe.Deny = append(perms.Subscribe.Deny, remote.DenyImports...)
-			}
-		}
-		c.setPermissions(perms)
-	}
-
 	return nil
 }
 
@@ -1075,10 +1057,6 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 		c.leaf.isSpoke = true
 	}
 
-	// If we have permissions bound to this leafnode we need to send then back to the
-	// origin server for local enforcement.
-	s.sendPermsInfo(c)
-
 	// Create and initialize the smap since we know our bound account now.
 	// This will send all registered subs too.
 	s.initLeafNodeSmapAndSendSubs(c)
@@ -1093,24 +1071,6 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 	return nil
 }
 
-// Sends back an info block to the soliciting leafnode to let it know about
-// its permission settings for local enforcement.
-func (s *Server) sendPermsInfo(c *client) {
-	if c.perms == nil {
-		return
-	}
-	// Copy
-	info := s.copyLeafNodeInfo()
-	c.mu.Lock()
-	info.CID = c.cid
-	info.Import = c.opts.Import
-	info.Export = c.opts.Export
-	b, _ := json.Marshal(info)
-	pcs := [][]byte{[]byte("INFO"), b, []byte(CR_LF)}
-	c.enqueueProto(bytes.Join(pcs, []byte(" ")))
-	c.mu.Unlock()
-}
-
 // Snapshot the current subscriptions from the sublist into our smap which
 // we will keep updated from now on.
 // Also send the registered subscriptions.
@@ -1317,21 +1277,6 @@ func (c *client) updateSmap(sub *subscription, delta int32) {
 // Send the subscription interest change to the other side.
 // Lock should be held.
 func (c *client) sendLeafNodeSubUpdate(key string, n int32) {
-	// If we are a spoke, we need to check if we are allowed to send this subscription over to the hub.
-	if c.isSpokeLeafNode() {
-		checkPerms := true
-		if len(key) > 0 && (key[0] == '$' || key[0] == '_') {
-			if strings.HasPrefix(key, leafNodeLoopDetectionSubjectPrefix) ||
-				strings.HasPrefix(key, oldGWReplyPrefix) ||
-				strings.HasPrefix(key, gwReplyPrefix) {
-				checkPerms = false
-			}
-		}
-		if checkPerms && !c.canSubscribe(key) {
-			return
-		}
-	}
-	// If we are here we can send over to the other side.
 	_b := [64]byte{}
 	b := bytes.NewBuffer(_b[:0])
 	c.writeLeafSub(b, key, n)
@@ -1440,7 +1385,7 @@ func (c *client) processLeafSub(argo []byte) (err error) {
 			checkPerms = false
 		}
 	}
-	if checkPerms && c.isHubLeafNode() && !c.canSubscribe(string(sub.subject)) {
+	if checkPerms && !c.canExport(string(sub.subject)) {
 		c.mu.Unlock()
 		c.leafSubPermViolation(sub.subject)
 		return nil
@@ -1717,7 +1662,7 @@ func (c *client) processInboundLeafMsg(msg []byte) {
 	c.in.bytes += int32(len(msg) - LEN_CR_LF)
 
 	// Check pub permissions
-	if c.perms != nil && (c.perms.pub.allow != nil || c.perms.pub.deny != nil) && c.isHubLeafNode() && !c.pubAllowed(string(c.pa.subject)) {
+	if c.perms != nil && (c.perms.pub.allow != nil || c.perms.pub.deny != nil) && !c.pubAllowed(string(c.pa.subject)) {
 		c.leafPubPermViolation(c.pa.subject)
 		return
 	}
@@ -1800,12 +1745,6 @@ func (c *client) leafSubPermViolation(subj []byte) {
 // Sends the permission violation error to the remote, logs it and closes the connection.
 // If this is from a server soliciting, the reconnection will be delayed.
 func (c *client) leafPermViolation(pub bool, subj []byte) {
-	if c.isSpokeLeafNode() {
-		// For spokes these are no-ops since the hub server told us our permissions.
-		// We just need to not send these over to the other side since we will get cutoff.
-		return
-	}
-	// FIXME(dlc) ?
 	c.setLeafConnectDelayIfSoliciting(leafNodeReconnectAfterPermViolation)
 	var action string
 	if pub {
diff --git a/server/monitor.go b/server/monitor.go
index 81b4c2e1..c4345ab5 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1047,6 +1047,7 @@ type JetStreamVarz struct {
 
 // ClusterOptsVarz contains monitoring cluster information
 type ClusterOptsVarz struct {
+	Name        string   `json:"name,omitempty"`
 	Host        string   `json:"addr,omitempty"`
 	Port        int      `json:"cluster_port,omitempty"`
 	AuthTimeout float64  `json:"auth_timeout,omitempty"`
@@ -1197,6 +1198,7 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 		HTTPBasePath: opts.HTTPBasePath,
 		HTTPSPort:    opts.HTTPSPort,
 		Cluster: ClusterOptsVarz{
+			Name:        info.Cluster,
 			Host:        c.Host,
 			Port:        c.Port,
 			AuthTimeout: c.AuthTimeout,
diff --git a/server/server.go b/server/server.go
index c7ff96b8..e22514e0 100644
--- a/server/server.go
+++ b/server/server.go
@@ -427,9 +427,17 @@ func (s *Server) ClusterName() string {
 // setClusterName will update the cluster name for this server.
 func (s *Server) setClusterName(name string) {
 	s.mu.Lock()
+	var resetCh chan struct{}
+	if s.sys != nil && s.info.Cluster != name {
+		// can't hold the lock as go routine reading it may be waiting for lock as well
+		resetCh = s.sys.resetCh
+	}
 	s.info.Cluster = name
 	s.routeInfo.Cluster = name
 	s.mu.Unlock()
+	if resetCh != nil {
+		resetCh <- struct{}{}
+	}
 	s.Noticef("Cluster name updated to %s", name)
 }
 


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 774b2f1b..0c72474a 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -100,7 +100,7 @@ func runTrustedCluster(t *testing.T) (*Server, *Options, *Server, *Options, nkey
 	optsA.TrustedKeys = []string{pub}
 	optsA.AccountResolver = mr
 	optsA.SystemAccount = apub
-	optsA.ServerName = "A"
+	optsA.ServerName = "A_SRV"
 	// Add in dummy gateway
 	optsA.Gateway.Name = "TEST CLUSTER 22"
 	optsA.Gateway.Host = "127.0.0.1"
@@ -110,7 +110,7 @@ func runTrustedCluster(t *testing.T) (*Server, *Options, *Server, *Options, nkey
 	sa := RunServer(optsA)
 
 	optsB := nextServerOpts(optsA)
-	optsB.ServerName = "B"
+	optsB.ServerName = "B_SRV"
 	optsB.Routes = RoutesFromStr(fmt.Sprintf("nats://%s:%d", optsA.Cluster.Host, optsA.Cluster.Port))
 	sb := RunServer(optsB)
 
@@ -1452,7 +1452,7 @@ func TestServerEventsStatsZ(t *testing.T) {
 	if lr := len(m3.Stats.Routes); lr != 1 {
 		t.Fatalf("Expected a route, but got %d", lr)
 	}
-	if sr := m3.Stats.Routes[0]; sr.Name != "B" {
+	if sr := m3.Stats.Routes[0]; sr.Name != "B_SRV" {
 		t.Fatalf("Expected server A's route to B to have Name set to %q, got %q", "B", sr.Name)
 	}
 
@@ -1470,8 +1470,8 @@ func TestServerEventsStatsZ(t *testing.T) {
 	if lr := len(m.Stats.Routes); lr != 1 {
 		t.Fatalf("Expected a route, but got %d", lr)
 	}
-	if sr := m.Stats.Routes[0]; sr.Name != "A" {
-		t.Fatalf("Expected server B's route to A to have Name set to %q, got %q", "A", sr.Name)
+	if sr := m.Stats.Routes[0]; sr.Name != "A_SRV" {
+		t.Fatalf("Expected server B's route to A to have Name set to %q, got %q", "A_SRV", sr.Name)
 	}
 }
 
@@ -1487,28 +1487,94 @@ func TestServerEventsPingStatsZ(t *testing.T) {
 	}
 	defer nc.Close()
 
-	reply := nc.NewRespInbox()
-	sub, _ := nc.SubscribeSync(reply)
+	requestTbl := []string{
+		`{"cluster":"TEST"}`,
+		`{"cluster":"CLUSTER"}`,
+		`{"name":"SRV"}`,
+		`{"name":"_"}`,
+		fmt.Sprintf(`{"host":"%s"}`, optsB.Host),
+		fmt.Sprintf(`{"host":"%s", "cluster":"CLUSTER", "name":"SRV"}`, optsB.Host),
+	}
 
-	nc.PublishRequest(serverStatsPingReqSubj, reply, nil)
+	for _, msg := range requestTbl {
+		t.Run(msg, func(t *testing.T) {
+			reply := nc.NewRespInbox()
+			sub, _ := nc.SubscribeSync(reply)
 
-	// Make sure its a statsz
-	m := ServerStatsMsg{}
+			nc.PublishRequest(serverStatsPingReqSubj, reply, []byte(msg))
 
-	// Receive both manually.
-	msg, err := sub.NextMsg(time.Second)
+			// Make sure its a statsz
+			m := ServerStatsMsg{}
+
+			// Receive both manually.
+			msg, err := sub.NextMsg(time.Second)
+			if err != nil {
+				t.Fatalf("Error receiving msg: %v", err)
+			}
+			if err := json.Unmarshal(msg.Data, &m); err != nil {
+				t.Fatalf("Error unmarshalling the statz json: %v", err)
+			}
+			msg, err = sub.NextMsg(time.Second)
+			if err != nil {
+				t.Fatalf("Error receiving msg: %v", err)
+			}
+			if err := json.Unmarshal(msg.Data, &m); err != nil {
+				t.Fatalf("Error unmarshalling the statz json: %v", err)
+			}
+		})
+	}
+}
+
+func TestServerEventsPingStatsZFilter(t *testing.T) {
+	sa, _, sb, optsB, akp := runTrustedCluster(t)
+	defer sa.Shutdown()
+	defer sb.Shutdown()
+
+	url := fmt.Sprintf("nats://%s:%d", optsB.Host, optsB.Port)
+	nc, err := nats.Connect(url, createUserCreds(t, sb, akp))
 	if err != nil {
-		t.Fatalf("Error receiving msg: %v", err)
+		t.Fatalf("Error on connect: %v", err)
 	}
-	if err := json.Unmarshal(msg.Data, &m); err != nil {
-		t.Fatalf("Error unmarshalling the statz json: %v", err)
+	defer nc.Close()
+
+	requestTbl := []string{
+		`{"cluster":"DOESNOTEXIST"}`,
+		`{"host":"DOESNOTEXIST"}`,
+		`{"name":"DOESNOTEXIST"}`,
 	}
-	msg, err = sub.NextMsg(time.Second)
+	for _, msg := range requestTbl {
+		t.Run(msg, func(t *testing.T) {
+			// Receive both manually.
+			if _, err := nc.Request(serverStatsPingReqSubj, []byte(msg), time.Second/4); err != nats.ErrTimeout {
+				t.Fatalf("Error, expected timeout: %v", err)
+			}
+		})
+	}
+}
+
+func TestServerEventsPingStatsZFailFilter(t *testing.T) {
+	sa, _, sb, optsB, akp := runTrustedCluster(t)
+	defer sa.Shutdown()
+	defer sb.Shutdown()
+
+	url := fmt.Sprintf("nats://%s:%d", optsB.Host, optsB.Port)
+	nc, err := nats.Connect(url, createUserCreds(t, sb, akp))
 	if err != nil {
-		t.Fatalf("Error receiving msg: %v", err)
+		t.Fatalf("Error on connect: %v", err)
 	}
-	if err := json.Unmarshal(msg.Data, &m); err != nil {
-		t.Fatalf("Error unmarshalling the statz json: %v", err)
+	defer nc.Close()
+
+	// Receive both manually.
+	if msg, err := nc.Request(serverStatsPingReqSubj, []byte(`{MALFORMEDJSON`), time.Second/4); err != nil {
+		t.Fatalf("Error: %v", err)
+	} else {
+		resp := make(map[string]map[string]interface{})
+		if err := json.Unmarshal(msg.Data, &resp); err != nil {
+			t.Fatalf("Error unmarshalling the response json: %v", err)
+		}
+		if resp["error"]["code"].(float64) != http.StatusBadRequest {
+			t.Fatal("bad error code")
+		}
 	}
 }
 
@@ -1566,6 +1632,17 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 			[]string{"now", "outbound_gateways", "inbound_gateways"}},
 		{"LEAFZ", &LeafzOptions{Subscriptions: true}, &Leafz{},
 			[]string{"now", "leafs"}},
+
+		{"ROUTEZ", json.RawMessage(`{"cluster":""}`), &Routez{},
+			[]string{"now", "routes"}},
+		{"ROUTEZ", json.RawMessage(`{"name":""}`), &Routez{},
+			[]string{"now", "routes"}},
+		{"ROUTEZ", json.RawMessage(`{"cluster":"TEST CLUSTER 22"}`), &Routez{},
+			[]string{"now", "routes"}},
+		{"ROUTEZ", json.RawMessage(`{"cluster":"CLUSTER"}`), &Routez{},
+			[]string{"now", "routes"}},
+		{"ROUTEZ", json.RawMessage(`{"cluster":"TEST CLUSTER 22", "subscriptions":true}`), &Routez{},
+			[]string{"now", "routes"}},
 	}
 
 	for i, test := range tests {
@@ -1595,10 +1672,10 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 			}
 
 			serverName := ""
-			if response1["server"]["name"] == "A" {
-				serverName = "B"
-			} else if response1["server"]["name"] == "B" {
-				serverName = "A"
+			if response1["server"]["name"] == "A_SRV" {
+				serverName = "B_SRV"
+			} else if response1["server"]["name"] == "B_SRV" {
+				serverName = "A_SRV"
 			} else {
 				t.Fatalf("Error finding server in %s", string(msg.Data))
 			}
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index 96bf157a..a2ad76fb 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -1186,11 +1186,46 @@ func TestLeafNodePermissions(t *testing.T) {
 	// Goes down to 1 (the $LDS one)
 	checkSubs(ln2.globalAccount(), 1)
 
-	// We used to make sure we would not do subscriptions however that
-	// was incorrect. We need to check publishes, not the subscriptions.
-	// For instance if we can publish across a leafnode to foo, and the
-	// other side has a subsxcription for '*' the message should cross
-	// the leafnode. The old way would not allow this.
+	// Check that deny export clause prevents subs from LN1 on matching
+	// subjects in the deny list to be registered on LN2. Furthermore,
+	// LN1 will receive a permission violation and the connection will
+	// be closed.
+	for _, test := range []struct {
+		name    string
+		subject string
+		ok      bool
+	}{
+		{"reject remote sub on export.*", "export.*", false},
+		{"reject remote sub on export", "export", false},
+		{"accepts remote sub on foo", "foo", true},
+		{"accepts remote sub on export.this.one.ok", "export.this.one.ok", true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			sub := natsSubSync(t, nc1, test.subject)
+			if !test.ok {
+				select {
+				case e := <-errLog.errCh:
+					if !strings.Contains(e, "Permissions Violation for Subscription to") {
+						t.Fatalf("Unexpected error: %q", e)
+					}
+				case <-time.After(time.Second):
+					t.Fatalf("Should have got an error")
+				}
+				sub.Unsubscribe()
+				// Should have disconnected
+				checkLeafNodeConnectedCount(t, ln2, 0)
+				checkSubs(ln2.globalAccount(), 0)
+				// But will reconnect after the connect delay
+				checkLeafNodeConnectedCount(t, ln2, 1)
+			} else {
+				checkSubs(ln2.globalAccount(), 2)
+				nc2.Publish(test.subject, []byte("msg"))
+				natsNexMsg(t, sub, time.Second)
+				sub.Unsubscribe()
+				checkSubs(ln2.globalAccount(), 1)
+			}
+		})
+	}
 
 	// Now check deny import clause.
 	// As of now, we don't suppress forwarding of subscriptions on LN2 that
@@ -1209,21 +1244,28 @@ func TestLeafNodePermissions(t *testing.T) {
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			sub := natsSubSync(t, nc2, test.subSubject)
-			checkSubs(ln2.globalAccount(), 2)
-
+			checkSubs(ln1.globalAccount(), 2)
+			nc1.Publish(test.pubSubject, []byte("msg"))
 			if !test.ok {
-				nc1.Publish(test.pubSubject, []byte("msg"))
-				if _, err := sub.NextMsg(50 * time.Millisecond); err == nil {
-					t.Fatalf("Did not expect to get the message")
+				select {
+				case e := <-errLog.errCh:
+					if !strings.Contains(e, "Permissions Violation for Publish to") {
+						t.Fatalf("Unexpected error: %q", e)
+					}
+				case <-time.After(time.Second):
+					t.Fatalf("Should have got an error")
 				}
+				sub.Unsubscribe()
+				// Should have disconnected
+				checkLeafNodeConnectedCount(t, ln2, 0)
+				checkSubs(ln1.globalAccount(), 0)
+				// But will reconnect after the connect delay
+				checkLeafNodeConnectedCount(t, ln2, 1)
 			} else {
-				checkSubs(ln1.globalAccount(), 2)
-				nc1.Publish(test.pubSubject, []byte("msg"))
 				natsNexMsg(t, sub, time.Second)
+				sub.Unsubscribe()
+				checkSubs(ln1.globalAccount(), 1)
 			}
-			sub.Unsubscribe()
-			checkSubs(ln1.globalAccount(), 1)
-
 		})
 	}
 }
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 6d0cc10b..2a5555c0 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -2415,6 +2415,7 @@ func TestMonitorCluster(t *testing.T) {
 	defer s.Shutdown()
 
 	expected := ClusterOptsVarz{
+		"A",
 		opts.Cluster.Host,
 		opts.Cluster.Port,
 		opts.Cluster.AuthTimeout,
@@ -2434,11 +2435,12 @@ func TestMonitorCluster(t *testing.T) {
 
 		// Having this here to make sure that if fields are added in ClusterOptsVarz,
 		// we make sure to update this test (compiler will report an error if we don't)
-		_ = ClusterOptsVarz{"", 0, 0, nil}
+		_ = ClusterOptsVarz{"", "", 0, 0, nil}
 
 		// Alter the fields to make sure that we have a proper deep copy
 		// of what may be stored in the server. Anything we change here
 		// should not affect the next returned value.
+		v.Cluster.Name = "wrong"
 		v.Cluster.Host = "wrong"
 		v.Cluster.Port = 0
 		v.Cluster.AuthTimeout = 0
diff --git a/test/cluster_test.go b/test/cluster_test.go
index abc43bf4..1a0dd7b8 100644
--- a/test/cluster_test.go
+++ b/test/cluster_test.go
@@ -89,22 +89,6 @@ func checkSubInterest(t *testing.T, s *server.Server, accName, subject string, t
 	})
 }
 
-func checkNoSubInterest(t *testing.T, s *server.Server, accName, subject string, timeout time.Duration) {
-	t.Helper()
-	acc, err := s.LookupAccount(accName)
-	if err != nil {
-		t.Fatalf("error looking up account %q: %v", accName, err)
-	}
-
-	start := time.Now()
-	for time.Now().Before(start.Add(timeout)) {
-		if acc.SubscriptionInterest(subject) {
-			t.Fatalf("Did not expect interest for %q", subject)
-		}
-		time.Sleep(5 * time.Millisecond)
-	}
-}
-
 func runThreeServers(t *testing.T) (srvA, srvB, srvC *server.Server, optsA, optsB, optsC *server.Options) {
 	srvA, optsA = RunServerWithConfig("./configs/srv_a.conf")
 	srvB, optsB = RunServerWithConfig("./configs/srv_b.conf")
diff --git a/test/leafnode_test.go b/test/leafnode_test.go
index 300f653c..b19fb010 100644
--- a/test/leafnode_test.go
+++ b/test/leafnode_test.go
@@ -1364,101 +1364,6 @@ func TestLeafNodeOperatorModel(t *testing.T) {
 	checkLeafNodeConnected(t, s)
 }
 
-func TestLeafNodeUserPermsForConnection(t *testing.T) {
-	s, opts, conf := runLeafNodeOperatorServer(t)
-	defer os.Remove(conf)
-	defer s.Shutdown()
-
-	// Setup account and a user that will be used by the remote leaf node server.
-	// createAccount automatically registers with resolver etc..
-	acc, akp := createAccount(t, s)
-	kp, _ := nkeys.CreateUser()
-	pub, _ := kp.PublicKey()
-	nuc := jwt.NewUserClaims(pub)
-	nuc.Permissions.Pub.Allow.Add("foo.>")
-	nuc.Permissions.Pub.Allow.Add("baz.>")
-	nuc.Permissions.Sub.Allow.Add("foo.>")
-	ujwt, err := nuc.Encode(akp)
-	if err != nil {
-		t.Fatalf("Error generating user JWT: %v", err)
-	}
-	seed, _ := kp.Seed()
-	mycreds := genCredsFile(t, ujwt, seed)
-	defer os.Remove(mycreds)
-
-	content := `
-		port: -1
-		leafnodes {
-			remotes = [
-				{
-					url: nats-leaf://127.0.0.1:%d
-					credentials: '%s'
-					deny_import: "foo.33"
-					deny_export: "foo.33"
-				}
-			]
-		}
-		`
-	config := fmt.Sprintf(content, opts.LeafNode.Port, mycreds)
-	lnconf := createConfFile(t, []byte(config))
-	defer os.Remove(lnconf)
-	sl, _ := RunServerWithConfig(lnconf)
-	defer sl.Shutdown()
-
-	checkLeafNodeConnected(t, s)
-
-	// Create credentials for a normal unrestricted user that we will connect to the op server.
-	url := fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port)
-	nc, err := nats.Connect(url, createUserCreds(t, s, akp))
-	if err != nil {
-		t.Fatalf("Error on connect: %v", err)
-	}
-	defer nc.Close()
-
-	// Create a user on the leafnode server that solicited.
-	nc2, err := nats.Connect(sl.ClientURL())
-	if err != nil {
-		t.Fatalf("Error on connect: %v", err)
-	}
-	defer nc2.Close()
-
-	// Make sure subscriptions properly do or do not make it to the hub.
-	// Note that all hub subscriptions will make it to the leafnode.
-	nc2.SubscribeSync("bar")
-	checkNoSubInterest(t, s, acc.GetName(), "bar", 20*time.Millisecond)
-	// This one should.
-	nc2.SubscribeSync("foo.22")
-	checkSubInterest(t, s, acc.GetName(), "foo.22", 20*time.Millisecond)
-
-	// Capture everything.
-	sub, _ := nc.SubscribeSync(">")
-	nc.Flush()
-
-	// Now check local pubs are not forwarded.
-	nc2.Publish("baz.22", nil)
-	m, err := sub.NextMsg(1 * time.Second)
-	if err != nil || m.Subject != "baz.22" {
-		t.Fatalf("Expected to received this message")
-	}
-	nc2.Publish("bar.22", nil)
-	if _, err := sub.NextMsg(100 * time.Millisecond); err == nil {
-		t.Fatalf("Did not expect to receive this message")
-	}
-
-	// Check local overrides work.
-	nc2.Publish("foo.33", nil)
-	if _, err := sub.NextMsg(100 * time.Millisecond); err == nil {
-		t.Fatalf("Did not expect to receive this message")
-	}
-
-	// This would trigger the sub interest below.
-	sub.Unsubscribe()
-	nc.Flush()
-
-	nc2.SubscribeSync("foo.33")
-	checkNoSubInterest(t, s, acc.GetName(), "foo.33", 20*time.Millisecond)
-}
-
 func TestLeafNodeMultipleAccounts(t *testing.T) {
 	// So we will create a main server with two accounts. The remote server, acting as a leaf node, will simply have
 	// the $G global account and no auth. Make sure things work properly here.