Repository: nats-io/nats-server
Pull request #1897 (2021-02-08 16:57:50)

[added] support for jwt based account mappings
Signed-off-by: Matthias Hanel <mh@synadia.com>

added code in accounts.go deals with:
assembling a remove list
adding/editing mappings
removing mappings


Patch:
diff --git a/go.mod b/go.mod
index b5c73c70..eb08e6b8 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.15
 require (
 	github.com/klauspost/compress v1.11.7
 	github.com/minio/highwayhash v1.0.0
-	github.com/nats-io/jwt/v2 v2.0.0-20210125223648-1c24d462becc
+	github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813
 	github.com/nats-io/nats.go v1.10.1-0.20210127212649-5b4924938a9a
 	github.com/nats-io/nkeys v0.2.0
 	github.com/nats-io/nuid v1.0.1
diff --git a/go.sum b/go.sum
index 39ea18ff..d2a5e843 100644
--- a/go.sum
+++ b/go.sum
@@ -19,8 +19,8 @@ github.com/nats-io/jwt v1.1.0 h1:+vOlgtM0ZsF46GbmUoadq0/2rChNS45gtxHEa3H1gqM=
 github.com/nats-io/jwt v1.1.0/go.mod h1:n3cvmLfBfnpV4JJRN7lRYCyZnw48ksGsbThGXEk4w9M=
 github.com/nats-io/jwt/v2 v2.0.0-20200916203241-1f8ce17dff02/go.mod h1:vs+ZEjP+XKy8szkBmQwCB7RjYdIlMaPsFPs4VdS4bTQ=
 github.com/nats-io/jwt/v2 v2.0.0-20201015190852-e11ce317263c/go.mod h1:vs+ZEjP+XKy8szkBmQwCB7RjYdIlMaPsFPs4VdS4bTQ=
-github.com/nats-io/jwt/v2 v2.0.0-20210125223648-1c24d462becc h1:pu+s4XC+bYnI0iD2vDtOl83zjCYUau/q6c83pEvsGZc=
-github.com/nats-io/jwt/v2 v2.0.0-20210125223648-1c24d462becc/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
+github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813 h1:km4lLzT86NyJRhO++VqfP/vn5cbfm+E05i2bGdqDbrY=
+github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200524125952-51ebd92a9093/go.mod h1:rQnBf2Rv4P9adtAs/Ti6LfFmVtFG6HLhl/H7cVshcJU=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200601203034-f8d6dd992b71/go.mod h1:Nan/1L5Sa1JRW+Thm4HNYcIDcVRFc5zK9OpSZeI2kk4=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad/go.mod h1:TkHpUIDETmTI7mrHN40D1pzxfzHZuGmtMbtb83TGVQw=
diff --git a/server/accounts.go b/server/accounts.go
index 35bba947..24b53795 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -506,9 +506,9 @@ func (a *Account) TotalSubs() int {
 
 // MapDest is for mapping published subjects for clients.
 type MapDest struct {
-	Subject    string `json:"subject"`
-	Weight     uint8  `json:"weight"`
-	OptCluster string `json:"cluster,omitempty"`
+	Subject string `json:"subject"`
+	Weight  uint8  `json:"weight"`
+	Cluster string `json:"cluster,omitempty"`
 }
 
 func NewMapDest(subject string, weight uint8) *MapDest {
@@ -573,16 +573,16 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 		if err != nil {
 			return err
 		}
-		if d.OptCluster == "" {
+		if d.Cluster == "" {
 			m.dests = append(m.dests, &destination{tr, d.Weight})
 		} else {
 			// We have a cluster scoped filter.
 			if m.cdests == nil {
 				m.cdests = make(map[string][]*destination)
 			}
-			ad := m.cdests[d.OptCluster]
+			ad := m.cdests[d.Cluster]
 			ad = append(ad, &destination{tr, d.Weight})
-			m.cdests[d.OptCluster] = ad
+			m.cdests[d.Cluster] = ad
 		}
 	}
 
@@ -2871,8 +2871,32 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 			alteredScope[k] = struct{}{}
 		}
 	}
+	// collect mappings that need to be removed
+	removeList := []string{}
+	for _, m := range a.mappings {
+		if _, ok := ac.Mappings[jwt.Subject(m.src)]; !ok {
+			removeList = append(removeList, m.src)
+		}
+	}
 	a.mu.Unlock()
 
+	for sub, wm := range ac.Mappings {
+		mappings := make([]*MapDest, len(wm))
+		for i, m := range wm {
+			mappings[i] = &MapDest{
+				Subject: string(m.Subject),
+				Weight:  m.GetWeight(),
+				Cluster: m.Cluster,
+			}
+		}
+		// This will overwrite existing entries
+		a.AddWeightedMappings(string(sub), mappings...)
+	}
+	// remove mappings
+	for _, rmMapping := range removeList {
+		a.RemoveMapping(rmMapping)
+	}
+
 	gatherClients := func() []*client {
 		a.mu.RLock()
 		clients := make([]*client, 0, len(a.clients))
diff --git a/server/client.go b/server/client.go
index 527c9017..d328f69b 100644
--- a/server/client.go
+++ b/server/client.go
@@ -2503,21 +2503,19 @@ func (c *client) addShadowSub(sub *subscription, ime *ime) (*subscription, error
 	nsub := *sub // copy
 	nsub.im = im
 
-	// Check if we need to change shadow subscription's subject.
-	if !im.usePub {
-		if ime.dyn {
-			if im.rtr == nil {
-				im.rtr = im.tr.reverse()
-			}
-			subj, err := im.rtr.transformSubject(string(nsub.subject))
-			if err != nil {
-				return nil, err
-			}
-			nsub.subject = []byte(subj)
-		} else {
-			nsub.subject = []byte(im.from)
+	if !im.usePub && ime.dyn {
+		if im.rtr == nil {
+			im.rtr = im.tr.reverse()
+		}
+		subj, err := im.rtr.transformSubject(string(nsub.subject))
+		if err != nil {
+			return nil, err
 		}
+		nsub.subject = []byte(subj)
+	} else if !im.usePub || !ime.dyn {
+		nsub.subject = []byte(im.from)
 	}
+	// Else use original subject
 	c.Debugf("Creating import subscription on %q from account %q", nsub.subject, im.acc.Name)
 
 	if err := im.acc.sl.Insert(&nsub); err != nil {
diff --git a/server/monitor.go b/server/monitor.go
index 91ce66e1..540dea6e 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1047,6 +1047,7 @@ type Varz struct {
 	Tags                  jwt.TagList           `json:"tags,omitempty"`
 	TrustedOperatorsJwt   []string              `json:"trusted_operators_jwt,omitempty"`
 	TrustedOperatorsClaim []*jwt.OperatorClaims `json:"trusted_operators_claim,omitempty"`
+	SystemAccount         string                `json:"system_account,omitempty"`
 }
 
 // JetStreamVarz contains basic runtime information about jetstream
@@ -1334,6 +1335,9 @@ func (s *Server) updateVarzConfigReloadableFields(v *Varz) {
 		v.Cluster.URLs = urlsToStrings(opts.Routes)
 		s.varzUpdateRouteURLs = false
 	}
+	if s.sys != nil && s.sys.account != nil {
+		v.SystemAccount = s.sys.account.GetName()
+	}
 }
 
 // Updates the runtime Varz fields, that is, fields that change during
@@ -1989,6 +1993,7 @@ type ExtMap map[string][]*MapDest
 type AccountInfo struct {
 	AccountName string               `json:"account_name"`
 	LastUpdate  time.Time            `json:"update_time,omitempty"`
+	IsSystem    bool                 `json:"is_system,omitempty"`
 	Expired     bool                 `json:"expired"`
 	Complete    bool                 `json:"complete"`
 	JetStream   bool                 `json:"jetstream_enabled"`
@@ -2011,10 +2016,11 @@ type AccountInfo struct {
 }
 
 type Accountz struct {
-	ID       string       `json:"server_id"`
-	Now      time.Time    `json:"now"`
-	Accounts []string     `json:"accounts,omitempty"`
-	Account  *AccountInfo `json:"account_detail,omitempty"`
+	ID            string       `json:"server_id"`
+	Now           time.Time    `json:"now"`
+	SystemAccount string       `json:"system_account,omitempty"`
+	Accounts      []string     `json:"accounts,omitempty"`
+	Account       *AccountInfo `json:"account_detail,omitempty"`
 }
 
 // HandleAccountz process HTTP requests for account information.
@@ -2039,6 +2045,9 @@ func (s *Server) Accountz(optz *AccountzOptions) (*Accountz, error) {
 		ID:  s.ID(),
 		Now: time.Now(),
 	}
+	if sacc := s.SystemAccount(); sacc != nil {
+		a.SystemAccount = sacc.GetName()
+	}
 	if optz.Account == "" {
 		a.Accounts = []string{}
 		s.accounts.Range(func(key, value interface{}) bool {
@@ -2086,6 +2095,7 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 	} else {
 		a = v.(*Account)
 	}
+	isSys := a == s.SystemAccount()
 	a.mu.RLock()
 	defer a.mu.RUnlock()
 	var vrIssues []ExtVrIssues
@@ -2190,6 +2200,7 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 	return &AccountInfo{
 		accName,
 		a.updated,
+		isSys,
 		a.expired,
 		!a.incomplete,
 		a.js != nil,
@@ -2212,15 +2223,16 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 	}, nil
 }
 
-// LeafzOptions are options passed to Jsz
+// JSzOptions are options passed to Jsz
 type JSzOptions struct {
-	Account  string `json:"account,omitempty"`
-	Accounts bool   `json:"accounts,omitempty"`
-	Streams  bool   `json:"streams,omitempty"`
-	Consumer bool   `json:"consumer,omitempty"`
-	Config   bool   `json:"config,omitempty"`
-	Offset   int    `json:"offset,omitempty"`
-	Limit    int    `json:"limit,omitempty"`
+	Account    string `json:"account,omitempty"`
+	Accounts   bool   `json:"accounts,omitempty"`
+	Streams    bool   `json:"streams,omitempty"`
+	Consumer   bool   `json:"consumer,omitempty"`
+	Config     bool   `json:"config,omitempty"`
+	LeaderOnly bool   `json:"leader_only,omitempty"`
+	Offset     int    `json:"offset,omitempty"`
+	Limit      int    `json:"limit,omitempty"`
 }
 
 type StreamDetail struct {
@@ -2337,6 +2349,25 @@ func (s *Server) Jsz(opts *JSzOptions) (*JSInfo, error) {
 	if opts.Streams {
 		opts.Accounts = true
 	}
+
+	// Check if we want a response from the leader only.
+	if opts.LeaderOnly {
+		js, cc := s.getJetStreamCluster()
+		if js == nil {
+			// Ignore
+			return nil, ErrJetStreamNotEnabled
+		}
+		// So if we have JS but no clustering, we are the leader so allow.
+		if cc != nil {
+			js.mu.RLock()
+			isLeader := cc.isLeader()
+			js.mu.RUnlock()
+			if !isLeader {
+				return nil, errNotLeader
+			}
+		}
+	}
+
 	// helper to get cluster info from node via dummy group
 	toClusterInfo := func(node RaftNode) *ClusterInfo {
 		peers := node.Peers()
@@ -2449,12 +2480,18 @@ func (s *Server) HandleJsz(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		return
 	}
+	leader, err := decodeBool(w, r, "leader-only")
+	if err != nil {
+		return
+	}
+
 	l, err := s.Jsz(&JSzOptions{
 		r.URL.Query().Get("acc"),
 		accounts,
 		streams,
 		consumers,
 		config,
+		leader,
 		offset,
 		limit})
 	if err != nil {
diff --git a/server/opts.go b/server/opts.go
index 0f288ca5..2915ea05 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -2018,7 +2018,7 @@ func parseAccountMapDest(v interface{}, tk token, errors *[]error, warnings *[]e
 				return nil, err
 			}
 		case "cluster":
-			mdest.OptCluster = dmv.(string)
+			mdest.Cluster = dmv.(string)
 		default:
 			err := &configErr{tk, fmt.Sprintf("Unknown field %q for mapping destination", k)}
 			*errors = append(*errors, err)
diff --git a/vendor/github.com/nats-io/jwt/v2/account_claims.go b/vendor/github.com/nats-io/jwt/v2/account_claims.go
index 67169483..04065b6a 100644
--- a/vendor/github.com/nats-io/jwt/v2/account_claims.go
+++ b/vendor/github.com/nats-io/jwt/v2/account_claims.go
@@ -84,6 +84,44 @@ func (o *OperatorLimits) Validate(_ *ValidationResults) {
 	// negative values mean unlimited, so all numbers are valid
 }
 
+// Mapping for publishes
+type WeightedMapping struct {
+	Subject Subject `json:"subject"`
+	Weight  uint8   `json:"weight,omitempty"`
+	Cluster string  `json:"cluster,omitempty"`
+}
+
+func (m *WeightedMapping) GetWeight() uint8 {
+	if m.Weight == 0 {
+		return 100
+	}
+	return m.Weight
+}
+
+type Mapping map[Subject][]WeightedMapping
+
+func (m *Mapping) Validate(vr *ValidationResults) {
+	for ubFrom, wm := range (map[Subject][]WeightedMapping)(*m) {
+		ubFrom.Validate(vr)
+		total := uint8(0)
+		for _, wm := range wm {
+			wm.Subject.Validate(vr)
+			if wm.Subject.HasWildCards() {
+				vr.AddError("Subject %q in weighted mapping %q is not allowed to contains wildcard",
+					string(wm.Subject), ubFrom)
+			}
+			total += wm.GetWeight()
+		}
+		if total > 100 {
+			vr.AddError("Mapping %q exceeds 100%% among all of it's weighted to mappings", ubFrom)
+		}
+	}
+}
+
+func (a *Account) AddMapping(sub Subject, to ...WeightedMapping) {
+	a.Mappings[sub] = to
+}
+
 // Account holds account specific claims data
 type Account struct {
 	Imports            Imports        `json:"imports,omitempty"`
@@ -92,6 +130,7 @@ type Account struct {
 	SigningKeys        SigningKeys    `json:"signing_keys,omitempty"`
 	Revocations        RevocationList `json:"revocations,omitempty"`
 	DefaultPermissions Permissions    `json:"default_permissions,omitempty"`
+	Mappings           Mapping        `json:"mappings,omitempty"`
 	Info
 	GenericFields
 }
@@ -102,6 +141,7 @@ func (a *Account) Validate(acct *AccountClaims, vr *ValidationResults) {
 	a.Exports.Validate(vr)
 	a.Limits.Validate(vr)
 	a.DefaultPermissions.Validate(vr)
+	a.Mappings.Validate(vr)
 
 	if !a.Limits.IsEmpty() && a.Limits.Imports >= 0 && int64(len(a.Imports)) > a.Limits.Imports {
 		vr.AddError("the account contains more imports than allowed by the operator")
@@ -150,6 +190,7 @@ func NewAccountClaims(subject string) *AccountClaims {
 		AccountLimits{NoLimit, NoLimit, true, NoLimit, NoLimit},
 		JetStreamLimits{0, 0, 0, 0}}
 	c.Subject = subject
+	c.Mappings = Mapping{}
 	return c
 }
 
diff --git a/vendor/github.com/nats-io/jwt/v2/exports.go b/vendor/github.com/nats-io/jwt/v2/exports.go
index 23065fbc..852ffae3 100644
--- a/vendor/github.com/nats-io/jwt/v2/exports.go
+++ b/vendor/github.com/nats-io/jwt/v2/exports.go
@@ -118,6 +118,7 @@ type Export struct {
 	ResponseThreshold    time.Duration   `json:"response_threshold,omitempty"`
 	Latency              *ServiceLatency `json:"service_latency,omitempty"`
 	AccountTokenPosition uint            `json:"account_token_position,omitempty"`
+	Advertise            bool            `json:"advertise,omitempty"`
 	Info
 }
 
diff --git a/vendor/github.com/nats-io/jwt/v2/imports.go b/vendor/github.com/nats-io/jwt/v2/imports.go
index 8eb5396c..705c630f 100644
--- a/vendor/github.com/nats-io/jwt/v2/imports.go
+++ b/vendor/github.com/nats-io/jwt/v2/imports.go
@@ -15,13 +15,6 @@
 
 package jwt
 
-import (
-	"io/ioutil"
-	"net/http"
-	"net/url"
-	"time"
-)
-
 // Import describes a mapping from another account into this one
 type Import struct {
 	Name string `json:"name,omitempty"`
@@ -92,32 +85,10 @@ func (i *Import) Validate(actPubKey string, vr *ValidationResults) {
 	var act *ActivationClaims
 
 	if i.Token != "" {
-		// Check to see if its an embedded JWT or a URL.
-		if u, err := url.Parse(i.Token); err == nil && u.Scheme != "" {
-			c := &http.Client{Timeout: 5 * time.Second}
-			resp, err := c.Get(u.String())
-			if err != nil {
-				vr.AddWarning("import %s contains an unreachable token URL %q", i.Subject, i.Token)
-			}
-
-			if resp != nil {
-				defer resp.Body.Close()
-				body, err := ioutil.ReadAll(resp.Body)
-				if err != nil {
-					vr.AddWarning("import %s contains an unreadable token URL %q", i.Subject, i.Token)
-				} else {
-					act, err = DecodeActivationClaims(string(body))
-					if err != nil {
-						vr.AddWarning("import %s contains a URL %q with an invalid activation token", i.Subject, i.Token)
-					}
-				}
-			}
-		} else {
-			var err error
-			act, err = DecodeActivationClaims(i.Token)
-			if err != nil {
-				vr.AddWarning("import %q contains an invalid activation token", i.Subject)
-			}
+		var err error
+		act, err = DecodeActivationClaims(i.Token)
+		if err != nil {
+			vr.AddWarning("import %q contains an invalid activation token", i.Subject)
 		}
 	}
 
diff --git a/vendor/github.com/nats-io/jwt/v2/operator_claims.go b/vendor/github.com/nats-io/jwt/v2/operator_claims.go
index 61d474e5..f806002d 100644
--- a/vendor/github.com/nats-io/jwt/v2/operator_claims.go
+++ b/vendor/github.com/nats-io/jwt/v2/operator_claims.go
@@ -166,6 +166,7 @@ func NewOperatorClaims(subject string) *OperatorClaims {
 	}
 	c := &OperatorClaims{}
 	c.Subject = subject
+	c.Issuer = subject
 	return c
 }
 
@@ -176,7 +177,10 @@ func (oc *OperatorClaims) DidSign(op Claims) bool {
 	}
 	issuer := op.Claims().Issuer
 	if issuer == oc.Subject {
-		return !oc.StrictSigningKeyUsage
+		if !oc.StrictSigningKeyUsage {
+			return true
+		}
+		return op.Claims().Subject == oc.Subject
 	}
 	return oc.SigningKeys.Contains(issuer)
 }
diff --git a/vendor/github.com/nats-io/jwt/v2/types.go b/vendor/github.com/nats-io/jwt/v2/types.go
index 3aa4f3a5..b538b62a 100644
--- a/vendor/github.com/nats-io/jwt/v2/types.go
+++ b/vendor/github.com/nats-io/jwt/v2/types.go
@@ -26,7 +26,7 @@ import (
 	"time"
 )
 
-const MaxInfoLength = 255
+const MaxInfoLength = 8 * 1024
 
 type Info struct {
 	Description string `json:"description,omitempty"`
diff --git a/vendor/modules.txt b/vendor/modules.txt
index af89062a..bd2271e8 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -4,7 +4,7 @@ github.com/klauspost/compress/s2
 # github.com/minio/highwayhash v1.0.0
 ## explicit
 github.com/minio/highwayhash
-# github.com/nats-io/jwt/v2 v2.0.0-20210125223648-1c24d462becc
+# github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813
 ## explicit
 github.com/nats-io/jwt/v2
 # github.com/nats-io/nats.go v1.10.1-0.20210127212649-5b4924938a9a


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index afda8a83..f84326bc 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nkeys"
 )
@@ -106,6 +107,188 @@ func TestAccountIsolation(t *testing.T) {
 	}
 }
 
+func TestAccountIsolationExportImport(t *testing.T) {
+	checkIsolation := func(t *testing.T, pubSubj string, ncExp, ncImp *nats.Conn) {
+		// We keep track of 2 subjects.
+		// One subject (pubSubj) is based off the stream import.
+		// The other subject "fizz" is not imported and should be isolated.
+
+		gotSubjs := map[string]int{
+			pubSubj: 0,
+			"fizz":  0,
+		}
+		sub, err := ncImp.Subscribe(">", func(m *nats.Msg) {
+			gotSubjs[m.Subject] += 1
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		time.Sleep(1 * time.Second)
+		if err := ncExp.Publish(pubSubj, []byte(fmt.Sprintf("ncExp pub %s", pubSubj))); err != nil {
+			t.Fatal(err)
+		}
+		if err := ncImp.Publish(pubSubj, []byte(fmt.Sprintf("ncImp pub %s", pubSubj))); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := ncExp.Publish("fizz", []byte("ncExp pub fizz")); err != nil {
+			t.Fatal(err)
+		}
+		if err := ncImp.Publish("fizz", []byte("ncImp pub fizz")); err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(1 * time.Second)
+		if err := sub.Unsubscribe(); err != nil {
+			t.Fatal(err)
+		}
+
+		wantSubjs := map[string]int{
+			// Subscriber ncImp should receive publishes from ncExp and ncImp.
+			pubSubj: 2,
+			// Subscriber ncImp should only receive the publish from ncImp.
+			"fizz": 1,
+		}
+		if got, want := len(gotSubjs), len(wantSubjs); got != want {
+			t.Fatalf("unexpected subjs len, got=%d; want=%d", got, want)
+		}
+
+		for key, gotCnt := range gotSubjs {
+			if wantCnt := wantSubjs[key]; gotCnt != wantCnt {
+				t.Errorf("unexpected receive count for subject %q, got=%d, want=%d", key, gotCnt, wantCnt)
+			}
+		}
+	}
+
+	cases := []struct {
+		name     string
+		exp, imp string
+		pubSubj  string
+	}{
+		{
+			name: "export literal, import literal",
+			exp:  "foo", imp: "foo",
+			pubSubj: "foo",
+		},
+		{
+			name: "export full wildcard, import literal",
+			exp:  "foo.>", imp: "foo.bar",
+			pubSubj: "foo.bar",
+		},
+		{
+			name: "export full wildcard, import sublevel full wildcard",
+			exp:  "foo.>", imp: "foo.bar.>",
+			pubSubj: "foo.bar.whizz",
+		},
+		{
+			name: "export full wildcard, import full wildcard",
+			exp:  "foo.>", imp: "foo.>",
+			pubSubj: "foo.bar",
+		},
+		{
+			name: "export partial wildcard, import partial wildcard",
+			exp:  "foo.*", imp: "foo.*",
+			pubSubj: "foo.bar",
+		},
+		{
+			name: "export mid partial wildcard, import mid partial wildcard",
+			exp:  "foo.*.bar", imp: "foo.*.bar",
+			pubSubj: "foo.whizz.bar",
+		},
+	}
+	for _, c := range cases {
+		t.Run(fmt.Sprintf("%s jwt", c.name), func(t *testing.T) {
+			// Setup NATS server.
+			s := opTrustBasicSetup()
+			defer s.Shutdown()
+			go s.Start()
+			if !s.ReadyForConnections(5 * time.Second) {
+				t.Fatal("failed to be ready for connections")
+			}
+			buildMemAccResolver(s)
+
+			// Setup exporter account.
+			accExpPair, accExpPub := createKey(t)
+			accExpClaims := jwt.NewAccountClaims(accExpPub)
+			if c.exp != "" {
+				accExpClaims.Limits.WildcardExports = true
+				accExpClaims.Exports.Add(&jwt.Export{
+					Name:    fmt.Sprintf("%s-stream-export", c.exp),
+					Subject: jwt.Subject(c.exp),
+					Type:    jwt.Stream,
+				})
+			}
+			accExpJWT, err := accExpClaims.Encode(oKp)
+			require_NoError(t, err)
+			addAccountToMemResolver(s, accExpPub, accExpJWT)
+
+			// Setup importer account.
+			accImpPair, accImpPub := createKey(t)
+			accImpClaims := jwt.NewAccountClaims(accImpPub)
+			if c.imp != "" {
+				accImpClaims.Imports.Add(&jwt.Import{
+					Name:    fmt.Sprintf("%s-stream-import", c.imp),
+					Subject: jwt.Subject(c.imp),
+					Account: accExpPub,
+					Type:    jwt.Stream,
+				})
+			}
+			accImpJWT, err := accImpClaims.Encode(oKp)
+			require_NoError(t, err)
+			addAccountToMemResolver(s, accImpPub, accImpJWT)
+
+			// Connect with different accounts.
+			ncExp := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, accExpPair),
+				nats.Name(fmt.Sprintf("nc-exporter-%s", c.exp)))
+			ncImp := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, accImpPair),
+				nats.Name(fmt.Sprintf("nc-importer-%s", c.imp)))
+			defer ncExp.Close()
+			defer ncImp.Close()
+
+			checkIsolation(t, c.pubSubj, ncExp, ncImp)
+			if t.Failed() {
+				t.Logf("exported=%q; imported=%q", c.exp, c.imp)
+			}
+		})
+
+		t.Run(fmt.Sprintf("%s conf", c.name), func(t *testing.T) {
+			// Setup NATS server.
+			cf := createConfFile(t, []byte(fmt.Sprintf(`
+				port: -1
+
+				accounts: {
+					accExp: {
+						users: [{user: accExp, password: accExp}]
+						exports: [{stream: %q}]
+					}
+					accImp: {
+						users: [{user: accImp, password: accImp}]
+						imports: [{stream: {account: accExp, subject: %q}}]
+					}
+				}
+			`,
+				c.exp, c.imp,
+			)))
+			defer os.Remove(cf)
+			s, _ := RunServerWithConfig(cf)
+			defer s.Shutdown()
+
+			// Connect with different accounts.
+			ncExp := natsConnect(t, s.ClientURL(), nats.UserInfo("accExp", "accExp"),
+				nats.Name(fmt.Sprintf("nc-exporter-%s", c.exp)))
+			ncImp := natsConnect(t, s.ClientURL(), nats.UserInfo("accImp", "accImp"),
+				nats.Name(fmt.Sprintf("nc-importer-%s", c.imp)))
+			defer ncExp.Close()
+			defer ncImp.Close()
+
+			checkIsolation(t, c.pubSubj, ncExp, ncImp)
+			if t.Failed() {
+				t.Logf("exported=%q; imported=%q", c.exp, c.imp)
+			}
+		})
+	}
+}
+
 func TestAccountFromOptions(t *testing.T) {
 	opts := defaultServerOptions
 	opts.Accounts = []*Account{NewAccount("foo"), NewAccount("bar")}
diff --git a/server/events_test.go b/server/events_test.go
index 8c6f2d2e..1bd0a4f1 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1933,7 +1933,7 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 		respField []string
 	}{
 		{"VARZ", nil, &Varz{},
-			[]string{"now", "cpu"}},
+			[]string{"now", "cpu", "system_account"}},
 		{"SUBSZ", nil, &Subsz{},
 			[]string{"num_subscriptions", "num_cache"}},
 		{"CONNZ", nil, &Connz{},
diff --git a/server/jwt_test.go b/server/jwt_test.go
index a78e7313..3bf0f5ac 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -5429,3 +5429,66 @@ func TestJWTAccountProtectedImport(t *testing.T) {
 		require_True(t, len(msgChan) == 0)
 	})
 }
+
+func TestJWTMappings(t *testing.T) {
+	sysKp, syspub := createKey(t)
+	sysJwt := encodeClaim(t, jwt.NewAccountClaims(syspub), syspub)
+	sysCreds := newUser(t, sysKp)
+	defer os.Remove(sysCreds)
+
+	// create two jwt, one with and one without mapping
+	aKp, aPub := createKey(t)
+	aClaim := jwt.NewAccountClaims(aPub)
+	aJwtNoM := encodeClaim(t, aClaim, aPub)
+	aClaim.AddMapping("foo1", jwt.WeightedMapping{Subject: "bar1"})
+	aJwtMap1 := encodeClaim(t, aClaim, aPub)
+
+	aClaim.Mappings = map[jwt.Subject][]jwt.WeightedMapping{}
+	aClaim.AddMapping("foo2", jwt.WeightedMapping{Subject: "bar2"})
+	aJwtMap2 := encodeClaim(t, aClaim, aPub)
+
+	dirSrv := createDir(t, "srv")
+	defer os.RemoveAll(dirSrv)
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: -1
+		operator: %s
+		system_account: %s
+		resolver: {
+			type: full
+			dir: %s
+		}
+    `, ojwt, syspub, dirSrv)))
+	defer os.Remove(conf)
+	srv, _ := RunServerWithConfig(conf)
+	defer srv.Shutdown()
+	updateJwt(t, srv.ClientURL(), sysCreds, sysJwt, 1) // update system account jwt
+
+	test := func(pub, sub string, fail bool) {
+		t.Helper()
+		nc := natsConnect(t, srv.ClientURL(), createUserCreds(t, srv, aKp))
+		defer nc.Close()
+		s, err := nc.SubscribeSync(sub)
+		require_NoError(t, err)
+		nc.Flush()
+		err = nc.Publish(pub, nil)
+		require_NoError(t, err)
+		_, err = s.NextMsg(500 * time.Millisecond)
+		switch {
+		case fail && err == nil:
+			t.Fatal("expected error, got none")
+		case !fail && err != nil:
+			t.Fatalf("expected no error, got %v", err)
+		}
+	}
+
+	// turn mappings on
+	require_Len(t, 1, updateJwt(t, srv.ClientURL(), sysCreds, aJwtMap1, 1))
+	test("foo1", "bar1", false)
+	// alter mappings
+	require_Len(t, 1, updateJwt(t, srv.ClientURL(), sysCreds, aJwtMap2, 1))
+	test("foo1", "bar1", true)
+	test("foo2", "bar2", false)
+	// turn mappings off
+	require_Len(t, 1, updateJwt(t, srv.ClientURL(), sysCreds, aJwtNoM, 1))
+	test("foo2", "bar2", true)
+}
diff --git a/server/monitor_test.go b/server/monitor_test.go
index d2c2932f..b32f008c 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -3783,6 +3783,8 @@ func TestMonitorAccountz(t *testing.T) {
 		t.Fatalf("Body missing value. Contains: %s", body)
 	} else if !strings.Contains(body, `"accounts": [`) {
 		t.Fatalf("Body missing value. Contains: %s", body)
+	} else if !strings.Contains(body, `"system_account": "$SYS"`) {
+		t.Fatalf("Body missing value. Contains: %s", body)
 	}
 	body = string(readBody(t, fmt.Sprintf("http://127.0.0.1:%d/accountz?acc=$SYS", s.MonitorAddr().Port)))
 	if !strings.Contains(body, `"account_detail": {`) {
@@ -3791,6 +3793,10 @@ func TestMonitorAccountz(t *testing.T) {
 		t.Fatalf("Body missing value. Contains: %s", body)
 	} else if !strings.Contains(body, `"subscriptions": 36,`) {
 		t.Fatalf("Body missing value. Contains: %s", body)
+	} else if !strings.Contains(body, `"is_system": true,`) {
+		t.Fatalf("Body missing value. Contains: %s", body)
+	} else if !strings.Contains(body, `"system_account": "$SYS"`) {
+		t.Fatalf("Body missing value. Contains: %s", body)
 	}
 }