Repository: nats-io/nats-server
Pull request #1790 (2021-01-04 19:22:56)

[Added] support for wildcard services and import remapping by JWT.
Imports in JWT where extended to contain a new filed LocalSubject.
This Change pulls the new JWT library version in.
It was needed as prefix did not exist in the JWT library and the
original field could not be used. The field To has been deprecated.

When LocalSubject is set, service imports can be configured the same way
they are in config. Meaning, no reversal due to the type.

This change also ensures that wildcard references in transforms are only
set in To/LocalSubject. Before, for services, $1 would have to be set in Subject.
This simplifies tools/parsing as only one field will have to be sanitized.

Signed-off-by: Matthias Hanel <mh@synadia.com>




Patch:
diff --git a/go.mod b/go.mod
index 93995437..98bf972d 100644
--- a/go.mod
+++ b/go.mod
@@ -2,7 +2,7 @@ module github.com/nats-io/nats-server/v2
 
 require (
 	github.com/minio/highwayhash v1.0.0
-	github.com/nats-io/jwt/v2 v2.0.0-20201217175639-911e216824d5
+	github.com/nats-io/jwt/v2 v2.0.0-20210104170214-da06b7f0b569
 	github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0
 	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 ddb7c5fc..9f6c3707 100644
--- a/go.sum
+++ b/go.sum
@@ -17,10 +17,8 @@ github.com/nats-io/jwt v0.3.3-0.20200519195258-f2bf5ce574c7/go.mod h1:n3cvmLfBfn
 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-20201211164018-2e78446f4e6f h1:+F8Vkc9pDf2x6O+csstCXaubPu2njJkbXeTm5wfDzGI=
-github.com/nats-io/jwt/v2 v2.0.0-20201211164018-2e78446f4e6f/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
-github.com/nats-io/jwt/v2 v2.0.0-20201217175639-911e216824d5 h1:EsHCbyEzBfPN3Fk4NVCJEyNwCDAJIdUtr+61tddWIm8=
-github.com/nats-io/jwt/v2 v2.0.0-20201217175639-911e216824d5/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
+github.com/nats-io/jwt/v2 v2.0.0-20210104170214-da06b7f0b569 h1:s5y2uE1Q2ganLOqtiLsqeOu/oU7uxdgS0aGUUYPeaho=
+github.com/nats-io/jwt/v2 v2.0.0-20210104170214-da06b7f0b569/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/logger/syslog.go b/logger/syslog.go
index 4023da88..f211ccd0 100644
--- a/logger/syslog.go
+++ b/logger/syslog.go
@@ -102,9 +102,9 @@ func (l *SysLogger) Noticef(format string, v ...interface{}) {
 	l.writer.Notice(fmt.Sprintf(format, v...))
 }
 
-// Warnf logs a warning statement
+// Warnf logs a notice statement
 func (l *SysLogger) Warnf(format string, v ...interface{}) {
-	l.writer.Warning(fmt.Sprintf(format, v...))
+	l.writer.Notice(fmt.Sprintf(format, v...))
 }
 
 // Fatalf logs a fatal error
diff --git a/logger/syslog_windows.go b/logger/syslog_windows.go
index 77a098ce..1a89e0a2 100644
--- a/logger/syslog_windows.go
+++ b/logger/syslog_windows.go
@@ -80,7 +80,7 @@ func (l *SysLogger) Noticef(format string, v ...interface{}) {
 	l.writer.Info(1, formatMsg("NOTICE", format, v...))
 }
 
-// Warnf logs a warning statement
+// Noticef logs a notice statement
 func (l *SysLogger) Warnf(format string, v ...interface{}) {
 	l.writer.Info(1, formatMsg("WARN", format, v...))
 }
diff --git a/server/accounts.go b/server/accounts.go
index 23b2c69f..4f77b8f8 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1727,12 +1727,16 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		if to == from {
 			usePub = true
 		} else {
-			from, _ = transformUntokenize(from)
-			// Create a transform
-			if tr, err = newTransform(from, transformTokenize(to)); err != nil {
+			to, _ = transformUntokenize(to)
+			// Create a transform. Do so in reverse such that $ symbols only exist in to
+			if tr, err = newTransform(to, transformTokenize(from)); err != nil {
 				a.mu.Unlock()
 				return nil, fmt.Errorf("failed to create mapping transform for service import subject %q to %q: %v",
 					from, to, err)
+			} else {
+				// un-tokenize and reverse transform so we get the transform needed
+				from, _ = transformUntokenize(from)
+				tr = tr.reverse()
 			}
 		}
 	}
@@ -2250,6 +2254,7 @@ func (a *Account) AddMappedStreamImportWithClaim(account *Account, from, to stri
 				return fmt.Errorf("failed to create mapping transform for stream import subject %q to %q: %v",
 					from, to, err)
 			}
+			to, _ = transformUntokenize(to)
 		}
 	}
 
@@ -2813,7 +2818,7 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 	if a == nil {
 		return
 	}
-	s.Debugf("Updating account claims: %s", a.Name)
+	s.Debugf("Updating account claims: %s/%s", a.Name, ac.Name)
 	a.checkExpiration(ac.Claims())
 
 	a.mu.Lock()
@@ -2936,17 +2941,31 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 			incompleteImports = append(incompleteImports, i)
 			continue
 		}
+		from := string(i.Subject)
+		to := i.GetTo()
 		switch i.Type {
 		case jwt.Stream:
-			s.Debugf("Adding stream import %s:%q for %s:%q", acc.Name, i.Subject, a.Name, i.To)
-			if err := a.AddStreamImportWithClaim(acc, string(i.Subject), string(i.To), i); err != nil {
+			if i.LocalSubject != "" {
+				// set local subject implies to is empty
+				to = string(i.LocalSubject)
+				s.Debugf("Adding stream import %s:%q for %s:%q", acc.Name, from, a.Name, to)
+				err = a.AddMappedStreamImportWithClaim(acc, from, to, i)
+			} else {
+				s.Debugf("Adding stream import %s:%q for %s:%q", acc.Name, from, a.Name, to)
+				err = a.AddStreamImportWithClaim(acc, from, to, i)
+			}
+			if err != nil {
 				s.Debugf("Error adding stream import to account [%s]: %v", a.Name, err.Error())
 				incompleteImports = append(incompleteImports, i)
 			}
 		case jwt.Service:
 			// FIXME(dlc) - need to add in respThresh here eventually.
-			s.Debugf("Adding service import %s:%q for %s:%q", acc.Name, i.Subject, a.Name, i.To)
-			if err := a.AddServiceImportWithClaim(acc, string(i.Subject), string(i.To), i); err != nil {
+			if i.LocalSubject != _EMPTY_ {
+				from = string(i.LocalSubject)
+				to = string(i.Subject)
+			}
+			s.Debugf("Adding service import %s:%q for %s:%q", acc.Name, from, a.Name, to)
+			if err := a.AddServiceImportWithClaim(acc, from, to, i); err != nil {
 				s.Debugf("Error adding service import to account [%s]: %v", a.Name, err.Error())
 				incompleteImports = append(incompleteImports, i)
 			}
diff --git a/server/gateway.go b/server/gateway.go
index e41de23d..1629998c 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -679,7 +679,7 @@ func (s *Server) solicitGateway(cfg *gatewayCfg, firstConnect bool) {
 				s.gateway.Lock()
 				// We could have just accepted an inbound for this remote gateway.
 				// So if there is an inbound, let's try again to connect.
-				if s.gateway.hasInbound(cfg.Name) {
+				if len(s.gateway.in) > 0 {
 					s.gateway.Unlock()
 					continue
 				}
@@ -697,20 +697,6 @@ func (s *Server) solicitGateway(cfg *gatewayCfg, firstConnect bool) {
 	}
 }
 
-// Returns true if there is an inbound for the given `name`.
-// Lock held on entry.
-func (g *srvGateway) hasInbound(name string) bool {
-	for _, ig := range g.in {
-		ig.mu.Lock()
-		igname := ig.gw.name
-		ig.mu.Unlock()
-		if igname == name {
-			return true
-		}
-	}
-	return false
-}
-
 // Called when a gateway connection is either accepted or solicited.
 // If accepted, the gateway is marked as inbound.
 // If solicited, the gateway is marked as outbound.
diff --git a/server/leafnode.go b/server/leafnode.go
index 842e5a4c..a448c331 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -508,14 +508,13 @@ func (s *Server) startLeafNodeAcceptLoop() {
 var credsRe = regexp.MustCompile(`\s*(?:(?:[-]{3,}[^\n]*[-]{3,}\n)(.+)(?:\n\s*[-]{3,}[^\n]*[-]{3,}\n))`)
 
 // Lock should be held entering here.
-func (c *client) sendLeafConnect(clusterName string, tlsRequired, headers bool) error {
+func (c *client) sendLeafConnect(clusterName string, tlsRequired bool) error {
 	// We support basic user/pass and operator based user JWT with signatures.
 	cinfo := leafConnectInfo{
 		TLS:     tlsRequired,
 		Name:    c.srv.info.ID,
 		Hub:     c.leaf.remote.Hub,
 		Cluster: clusterName,
-		Headers: headers,
 	}
 
 	// Check for credentials first, that will take precedence..
@@ -692,7 +691,6 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 		s.generateNonce(nonce[:])
 	}
 	clusterName := s.info.Cluster
-	headers := s.supportsHeaders()
 	s.mu.Unlock()
 
 	// Grab lock
@@ -827,7 +825,7 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 			}
 		}
 
-		if err := c.sendLeafConnect(clusterName, tlsRequired, headers); err != nil {
+		if err := c.sendLeafConnect(clusterName, tlsRequired); err != nil {
 			c.mu.Unlock()
 			c.closeConnection(ProtocolViolation)
 			return nil
@@ -890,6 +888,7 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 
 		// Leaf nodes will always require a CONNECT to let us know
 		// when we are properly bound to an account.
+		// The connection may have been closed
 		c.setAuthTimer(secondsToDuration(opts.LeafNode.AuthTimeout))
 	}
 
@@ -1162,7 +1161,6 @@ type leafConnectInfo struct {
 	Name    string `json:"name,omitempty"`
 	Hub     bool   `json:"is_hub,omitempty"`
 	Cluster string `json:"cluster,omitempty"`
-	Headers bool   `json:"headers,omitempty"`
 
 	// Just used to detect wrong connection attempts.
 	Gateway string `json:"gateway,omitempty"`
@@ -1196,18 +1194,11 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 		return ErrWrongGateway
 	}
 
-	// Check if this server supports headers.
-	supportHeaders := c.srv.supportsHeaders()
-
 	c.mu.Lock()
 	// Leaf Nodes do not do echo or verbose or pedantic.
 	c.opts.Verbose = false
 	c.opts.Echo = false
 	c.opts.Pedantic = false
-	// This inbound connection will be marked as supporting headers if this server
-	// support headers and the remote has sent in the CONNECT protocol that it does
-	// support headers too.
-	c.headers = supportHeaders && proto.Headers
 
 	// If the other side has declared itself a hub, so we will take on the spoke role.
 	if proto.Hub {
diff --git a/server/stream.go b/server/stream.go
index f5c2b40f..fa9ab962 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -878,19 +878,12 @@ func getHdrVal(key string, hdr []byte) []byte {
 	if index < 0 {
 		return nil
 	}
-
 	var value []byte
-	hdrLen := len(hdr)
-	index += len(key) + 1
-	for hdr[index] == ' ' && index < hdrLen {
-		index++
-	}
-	for index < hdrLen {
-		if hdr[index] == '\r' && index < hdrLen-1 && hdr[index+1] == '\n' {
+	for i := index + len(key) + 2; i > 0 && i < len(hdr); i++ {
+		if hdr[i] == '\r' && i < len(hdr)-1 && hdr[i+1] == '\n' {
 			break
 		}
-		value = append(value, hdr[index])
-		index++
+		value = append(value, hdr[i])
 	}
 	return value
 }
@@ -955,7 +948,6 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 			}
 			return
 		}
-
 		// Expected stream.
 		if sname := getExpectedStream(hdr); sname != _EMPTY_ && sname != name {
 			mset.mu.Unlock()
diff --git a/vendor/github.com/nats-io/jwt/v2/genericlaims.go b/vendor/github.com/nats-io/jwt/v2/genericlaims.go
index 38a6c43d..6793c9ea 100644
--- a/vendor/github.com/nats-io/jwt/v2/genericlaims.go
+++ b/vendor/github.com/nats-io/jwt/v2/genericlaims.go
@@ -78,7 +78,10 @@ func DecodeGeneric(token string) (*GenericClaims, error) {
 			return nil, errors.New("claim failed V1 signature verification")
 		}
 		if tp := gc.GenericFields.Type; tp != "" {
-			gc.GenericClaims.Data["type"] = tp
+			// the conversion needs to be from a string because
+			// on custom types the type is not going to be one of
+			// the constants
+			gc.GenericClaims.Data["type"] = string(tp)
 		}
 		if tp := gc.GenericFields.Tags; len(tp) != 0 {
 			gc.GenericClaims.Data["tags"] = tp
diff --git a/vendor/github.com/nats-io/jwt/v2/imports.go b/vendor/github.com/nats-io/jwt/v2/imports.go
index 4acb0b13..8eb5396c 100644
--- a/vendor/github.com/nats-io/jwt/v2/imports.go
+++ b/vendor/github.com/nats-io/jwt/v2/imports.go
@@ -32,14 +32,21 @@ type Import struct {
 	Subject Subject `json:"subject,omitempty"`
 	Account string  `json:"account,omitempty"`
 	Token   string  `json:"token,omitempty"`
+	// Deprecated: use LocalSubject instead
 	// To field in an import is always from the perspective of the subscriber
 	// in the case of a stream it is the client of the stream (the importer),
 	// from the perspective of a service, it is the subscription waiting for
 	// requests (the exporter). If the field is empty, it will default to the
 	// value in the Subject field.
-	To    Subject    `json:"to,omitempty"`
-	Type  ExportType `json:"type,omitempty"`
-	Share bool       `json:"share,omitempty"`
+	To Subject `json:"to,omitempty"`
+	// Local subject used to subscribe (for streams) and publish (for services) to.
+	// This value only needs setting if you want to change the value of Subject.
+	// If the value of Subject ends in > then LocalSubject needs to end in > as well.
+	// LocalSubject can contain $<number> wildcard references where number references the nth wildcard in Subject.
+	// The sum of wildcard reference and * tokens needs to match the number of * token in Subject.
+	LocalSubject RenamingSubject `json:"local_subject,omitempty"`
+	Type         ExportType      `json:"type,omitempty"`
+	Share        bool            `json:"share,omitempty"`
 }
 
 // IsService returns true if the import is of type service
@@ -52,6 +59,11 @@ func (i *Import) IsStream() bool {
 	return i.Type == Stream
 }
 
+// Returns the value of To without triggering the deprecation warning for a read
+func (i *Import) GetTo() string {
+	return string(i.To)
+}
+
 // Validate checks if an import is valid for the wrapping account
 func (i *Import) Validate(actPubKey string, vr *ValidationResults) {
 	if i == nil {
@@ -67,6 +79,12 @@ func (i *Import) Validate(actPubKey string, vr *ValidationResults) {
 	}
 
 	i.Subject.Validate(vr)
+	if i.LocalSubject != "" {
+		i.LocalSubject.Validate(i.Subject, vr)
+		if i.To != "" {
+			vr.AddError("Local Subject replaces To")
+		}
+	}
 
 	if i.Share && !i.IsService() {
 		vr.AddError("sharing information (for latency tracking) is only valid for services: %q", i.Subject)
@@ -120,17 +138,29 @@ type Imports []*Import
 
 // Validate checks if an import is valid for the wrapping account
 func (i *Imports) Validate(acctPubKey string, vr *ValidationResults) {
-	toSet := make(map[Subject]bool, len(*i))
+	toSet := make(map[Subject]struct{}, len(*i))
 	for _, v := range *i {
 		if v == nil {
 			vr.AddError("null import is not allowed")
 			continue
 		}
 		if v.Type == Service {
-			if _, ok := toSet[v.To]; ok {
-				vr.AddError("Duplicate To subjects for %q", v.To)
+			sub := v.To
+			if sub == "" {
+				sub = v.LocalSubject.ToSubject()
+			}
+			if sub == "" {
+				sub = v.Subject
+			}
+			for k := range toSet {
+				if sub.IsContainedIn(k) || k.IsContainedIn(sub) {
+					vr.AddError("overlapping subject namespace for %q and %q", sub, k)
+				}
+			}
+			if _, ok := toSet[sub]; ok {
+				vr.AddError("overlapping subject namespace for %q", v.To)
 			}
-			toSet[v.To] = true
+			toSet[sub] = struct{}{}
 		}
 		v.Validate(acctPubKey, vr)
 	}
diff --git a/vendor/github.com/nats-io/jwt/v2/signingkeys.go b/vendor/github.com/nats-io/jwt/v2/signingkeys.go
index 715df926..44637ad9 100644
--- a/vendor/github.com/nats-io/jwt/v2/signingkeys.go
+++ b/vendor/github.com/nats-io/jwt/v2/signingkeys.go
@@ -165,6 +165,14 @@ func (sk SigningKeys) UnmarshalJSON(data []byte) error {
 	return nil
 }
 
+func (sk SigningKeys) Keys() []string {
+	var keys []string
+	for k := range sk {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
 // GetScope returns nil if the key is not associated
 func (sk SigningKeys) GetScope(k string) (Scope, bool) {
 	v, ok := sk[k]
diff --git a/vendor/github.com/nats-io/jwt/v2/types.go b/vendor/github.com/nats-io/jwt/v2/types.go
index 4a42ad7c..aee4bc34 100644
--- a/vendor/github.com/nats-io/jwt/v2/types.go
+++ b/vendor/github.com/nats-io/jwt/v2/types.go
@@ -21,6 +21,7 @@ import (
 	"net"
 	"net/url"
 	"reflect"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -101,6 +102,73 @@ func (t *ExportType) UnmarshalJSON(b []byte) error {
 	return fmt.Errorf("unknown export type %q", j)
 }
 
+type RenamingSubject Subject
+
+func (s RenamingSubject) Validate(from Subject, vr *ValidationResults) {
+	v := Subject(s)
+	v.Validate(vr)
+	if from == "" {
+		vr.AddError("subject cannot be empty")
+	}
+	if strings.Contains(string(s), " ") {
+		vr.AddError("subject %q cannot have spaces", v)
+	}
+	matchesSuffix := func(s Subject) bool {
+		return s == ">" || strings.HasSuffix(string(s), ".>")
+	}
+	if matchesSuffix(v) != matchesSuffix(from) {
+		vr.AddError("both, renaming subject and subject, need to end or not end in >")
+	}
+	fromCnt := from.countTokenWildcards()
+	refCnt := 0
+	for _, tk := range strings.Split(string(v), ".") {
+		if tk == "*" {
+			refCnt++
+		}
+		if len(tk) < 2 {
+			continue
+		}
+		if tk[0] == '$' {
+			if idx, err := strconv.Atoi(tk[1:]); err == nil {
+				if idx > fromCnt {
+					vr.AddError("Reference $%d in %q reference * in %q that do not exist", idx, s, from)
+				} else {
+					refCnt++
+				}
+			}
+		}
+	}
+	if refCnt != fromCnt {
+		vr.AddError("subject does not contain enough * or reference wildcards $[0-9]")
+	}
+}
+
+// Replaces reference tokens with *
+func (s RenamingSubject) ToSubject() Subject {
+	if !strings.Contains(string(s), "$") {
+		return Subject(s)
+	}
+	bldr := strings.Builder{}
+	tokens := strings.Split(string(s), ".")
+	for i, tk := range tokens {
+		convert := false
+		if len(tk) > 1 && tk[0] == '$' {
+			if _, err := strconv.Atoi(tk[1:]); err == nil {
+				convert = true
+			}
+		}
+		if convert {
+			bldr.WriteString("*")
+		} else {
+			bldr.WriteString(tk)
+		}
+		if i != len(tokens)-1 {
+			bldr.WriteString(".")
+		}
+	}
+	return Subject(bldr.String())
+}
+
 // Subject is a string that represents a NATS subject
 type Subject string
 
@@ -115,6 +183,21 @@ func (s Subject) Validate(vr *ValidationResults) {
 	}
 }
 
+func (s Subject) countTokenWildcards() int {
+	v := string(s)
+	if v == "*" {
+		return 1
+	}
+	cnt := strings.Count(v, ".*.")
+	if strings.HasSuffix(v, ".*") {
+		cnt++
+	}
+	if strings.HasPrefix(v, "*.") {
+		cnt++
+	}
+	return cnt
+}
+
 // HasWildCards is used to check if a subject contains a > or *
 func (s Subject) HasWildCards() bool {
 	v := string(s)
diff --git a/vendor/modules.txt b/vendor/modules.txt
index be96bcc2..97eab4be 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1,6 +1,6 @@
 # github.com/minio/highwayhash v1.0.0
 github.com/minio/highwayhash
-# github.com/nats-io/jwt/v2 v2.0.0-20201217175639-911e216824d5
+# github.com/nats-io/jwt/v2 v2.0.0-20210104170214-da06b7f0b569
 github.com/nats-io/jwt/v2
 # github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0
 github.com/nats-io/nats.go


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index d709152b..74c6dc94 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -2857,6 +2857,72 @@ func TestAccountImportsWithWildcardSupport(t *testing.T) {
 	}
 }
 
+// duplicates TestJWTAccountImportsWithWildcardSupport (jwt_test.go) in config
+func TestAccountImportsWithWildcardSupportStreamAndService(t *testing.T) {
+	cf := createConfFile(t, []byte(`
+    accounts {
+      foo {
+        users = [{user: derek, password: foo}]
+        exports = [
+          { service: "$request.*.$in.*.>" }
+          { stream: "$events.*.$in.*.>" }
+        ]
+      }
+      bar {
+        users = [{user: ivan, password: bar}]
+        imports = [
+          { service: {account: "foo", subject:"$request.*.$in.*.>"}, to:"my.request.$2.$1.>"}
+          { stream:  {account: "foo", subject:"$events.*.$in.*.>"}, to:"my.events.$2.$1.>"}
+        ]
+      }
+    }
+    `))
+	defer os.Remove(cf)
+
+	s, opts := RunServerWithConfig(cf)
+	defer s.Shutdown()
+
+	ncFoo := natsConnect(t, fmt.Sprintf("nats://derek:foo@%s:%d", opts.Host, opts.Port))
+	defer ncFoo.Close()
+
+	ncBar := natsConnect(t, fmt.Sprintf("nats://ivan:bar@%s:%d", opts.Host, opts.Port))
+	defer ncBar.Close()
+
+	// Create subscriber for the service endpoint in foo.
+	_, err := ncFoo.Subscribe("$request.>", func(m *nats.Msg) {
+		if m.Subject != "$request.2.$in.1.bar" {
+			t.Fatalf("Expected literal subject for request, got %q", m.Subject)
+		}
+		m.Respond([]byte("yes!"))
+	})
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+	ncFoo.Flush()
+
+	// Now test service import.
+	if resp, err := ncBar.Request("my.request.1.2.bar", []byte("yes?"), time.Second); err != nil {
+		t.Fatalf("Expected a response")
+	} else if string(resp.Data) != "yes!" {
+		t.Fatalf("Expected a response of %q, got %q", "yes!", resp.Data)
+	}
+	subBar, err := ncBar.SubscribeSync("my.events.>")
+	if err != nil {
+		t.Fatalf("Expected a response")
+	}
+	ncBar.Flush()
+
+	ncFoo.Publish("$events.1.$in.2.bar", nil)
+
+	m, err := subBar.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Expected a response")
+	}
+	if m.Subject != "my.events.2.1.bar" {
+		t.Fatalf("Expected literal subject for request, got %q", m.Subject)
+	}
+}
+
 func BenchmarkNewRouteReply(b *testing.B) {
 	opts := defaultServerOptions
 	s := New(&opts)
diff --git a/server/gateway_test.go b/server/gateway_test.go
index c634df28..fd0eeeb8 100644
--- a/server/gateway_test.go
+++ b/server/gateway_test.go
@@ -1290,81 +1290,6 @@ func TestGatewayImplicitReconnectRace(t *testing.T) {
 	waitForInboundGateways(t, sa2, 1, 2*time.Second)
 }
 
-type gwReconnAttemptLogger struct {
-	DummyLogger
-	errCh chan string
-}
-
-func (l *gwReconnAttemptLogger) Errorf(format string, v ...interface{}) {
-	msg := fmt.Sprintf(format, v...)
-	if strings.Contains(msg, `Error connecting to implicit gateway "A"`) {
-		select {
-		case l.errCh <- msg:
-		default:
-		}
-	}
-}
-
-func TestGatewayImplicitReconnectHonorConnectRetries(t *testing.T) {
-	ob := testDefaultOptionsForGateway("B")
-	ob.ReconnectErrorReports = 1
-	ob.Gateway.ConnectRetries = 2
-	sb := runGatewayServer(ob)
-	defer sb.Shutdown()
-
-	l := &gwReconnAttemptLogger{errCh: make(chan string, 3)}
-	sb.SetLogger(l, true, false)
-
-	oa := testGatewayOptionsFromToWithServers(t, "A", "B", sb)
-	sa := runGatewayServer(oa)
-	defer sa.Shutdown()
-
-	// Wait for the proper connections
-	waitForOutboundGateways(t, sa, 1, time.Second)
-	waitForOutboundGateways(t, sb, 1, time.Second)
-	waitForInboundGateways(t, sa, 1, time.Second)
-	waitForInboundGateways(t, sb, 1, time.Second)
-
-	// Now have C connect to B.
-	oc := testGatewayOptionsFromToWithServers(t, "C", "B", sb)
-	sc := runGatewayServer(oc)
-	defer sc.Shutdown()
-
-	// Wait for the proper connections
-	waitForOutboundGateways(t, sa, 2, time.Second)
-	waitForOutboundGateways(t, sb, 2, time.Second)
-	waitForOutboundGateways(t, sc, 2, time.Second)
-	waitForInboundGateways(t, sa, 2, time.Second)
-	waitForInboundGateways(t, sb, 2, time.Second)
-	waitForInboundGateways(t, sc, 2, time.Second)
-
-	// Shutdown sa now...
-	sa.Shutdown()
-
-	// B will try to reconnect to A 3 times (we stop after attempts > ConnectRetries)
-	timeout := time.NewTimer(time.Second)
-	for i := 0; i < 3; i++ {
-		select {
-		case <-l.errCh:
-			// OK
-		case <-timeout.C:
-			t.Fatal("Did not get debug trace about reconnect")
-		}
-	}
-	// If we get 1 more, we have an issue!
-	select {
-	case e := <-l.errCh:
-		t.Fatalf("Should not have attempted to reconnect: %q", e)
-	case <-time.After(250 * time.Millisecond):
-		// OK!
-	}
-
-	waitForOutboundGateways(t, sb, 1, 2*time.Second)
-	waitForInboundGateways(t, sb, 1, 2*time.Second)
-	waitForOutboundGateways(t, sc, 1, 2*time.Second)
-	waitForInboundGateways(t, sc, 1, 2*time.Second)
-}
-
 func TestGatewayURLsFromClusterSentInINFO(t *testing.T) {
 	o2 := testDefaultOptionsForGateway("B")
 	s2 := runGatewayServer(o2)
diff --git a/server/jwt_test.go b/server/jwt_test.go
index c4a1f663..2992279b 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -4549,36 +4549,39 @@ func TestJWTAccountOps(t *testing.T) {
 	}
 }
 
+func createKey(t *testing.T) (nkeys.KeyPair, string) {
+	t.Helper()
+	kp, _ := nkeys.CreateAccount()
+	syspub, _ := kp.PublicKey()
+	return kp, syspub
+}
+
+func encodeClaim(t *testing.T, claim *jwt.AccountClaims, pub string) string {
+	t.Helper()
+	theJWT, err := claim.Encode(oKp)
+	require_NoError(t, err)
+	return theJWT
+}
+
+func newUser(t *testing.T, accKp nkeys.KeyPair) string {
+	ukp, _ := nkeys.CreateUser()
+	seed, _ := ukp.Seed()
+	upub, _ := ukp.PublicKey()
+	uclaim := newJWTTestUserClaims()
+	uclaim.Subject = upub
+	ujwt, err := uclaim.Encode(accKp)
+	require_NoError(t, err)
+	return genCredsFile(t, ujwt, seed)
+}
+
 func TestJWTHeader(t *testing.T) {
-	createKey := func() (nkeys.KeyPair, string) {
-		t.Helper()
-		kp, _ := nkeys.CreateAccount()
-		syspub, _ := kp.PublicKey()
-		return kp, syspub
-	}
-	encode := func(claim *jwt.AccountClaims, pub string) string {
-		t.Helper()
-		theJWT, err := claim.Encode(oKp)
-		require_NoError(t, err)
-		return theJWT
-	}
-	newUser := func(accKp nkeys.KeyPair) string {
-		ukp, _ := nkeys.CreateUser()
-		seed, _ := ukp.Seed()
-		upub, _ := ukp.PublicKey()
-		uclaim := newJWTTestUserClaims()
-		uclaim.Subject = upub
-		ujwt, err := uclaim.Encode(accKp)
-		require_NoError(t, err)
-		return genCredsFile(t, ujwt, seed)
-	}
-	sysKp, syspub := createKey()
-	sysJwt := encode(jwt.NewAccountClaims(syspub), syspub)
-	sysCreds := newUser(sysKp)
+	sysKp, syspub := createKey(t)
+	sysJwt := encodeClaim(t, jwt.NewAccountClaims(syspub), syspub)
+	sysCreds := newUser(t, sysKp)
 	defer os.Remove(sysCreds)
 
 	test := func(share bool) {
-		aExpKp, aExpPub := createKey()
+		aExpKp, aExpPub := createKey(t)
 		aExpClaim := jwt.NewAccountClaims(aExpPub)
 		aExpClaim.Exports.Add(&jwt.Export{
 			Name:     "test",
@@ -4590,11 +4593,11 @@ func TestJWTHeader(t *testing.T) {
 				Results:  "res",
 			},
 		})
-		aExpJwt := encode(aExpClaim, aExpPub)
-		aExpCreds := newUser(aExpKp)
+		aExpJwt := encodeClaim(t, aExpClaim, aExpPub)
+		aExpCreds := newUser(t, aExpKp)
 		defer os.Remove(aExpCreds)
 
-		aImpKp, aImpPub := createKey()
+		aImpKp, aImpPub := createKey(t)
 		aImpClaim := jwt.NewAccountClaims(aImpPub)
 		aImpClaim.Imports.Add(&jwt.Import{
 			Name:    "test",
@@ -4603,8 +4606,8 @@ func TestJWTHeader(t *testing.T) {
 			Type:    jwt.Service,
 			Share:   share,
 		})
-		aImpJwt := encode(aImpClaim, aImpPub)
-		aImpCreds := newUser(aImpKp)
+		aImpJwt := encodeClaim(t, aImpClaim, aImpPub)
+		aImpCreds := newUser(t, aImpKp)
 		defer os.Remove(aImpCreds)
 
 		dirSrv := createDir(t, "srv")
@@ -4678,3 +4681,140 @@ func TestJWTHeader(t *testing.T) {
 	test(true)
 	test(false)
 }
+
+func TestJWTAccountImportsWithWildcardSupport(t *testing.T) {
+	test := func(aExpPub, aExpJwt, aExpCreds, aImpPub, aImpJwt, aImpCreds, exSubExpect, exPub, imReq, imSubExpect string) {
+		cf := createConfFile(t, []byte(fmt.Sprintf(`
+		port: -1
+		operator = %s
+		resolver = MEMORY
+		resolver_preload = {
+			%s : "%s"
+			%s : "%s"
+		}
+		`, ojwt, aExpPub, aExpJwt, aImpPub, aImpJwt)))
+		defer os.Remove(cf)
+
+		s, opts := RunServerWithConfig(cf)
+		defer s.Shutdown()
+
+		ncExp := natsConnect(t, fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port), nats.UserCredentials(aExpCreds))
+		defer ncExp.Close()
+
+		ncImp := natsConnect(t, fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port), nats.UserCredentials(aImpCreds))
+		defer ncImp.Close()
+
+		// Create subscriber for the service endpoint in foo.
+		_, err := ncExp.Subscribe(exSubExpect, func(m *nats.Msg) {
+			m.Respond([]byte("yes!"))
+		})
+		require_NoError(t, err)
+		ncExp.Flush()
+
+		// Now test service import.
+		if resp, err := ncImp.Request(imReq, []byte("yes?"), time.Second); err != nil {
+			t.Fatalf("Expected a response to request %s got: %v", imReq, err)
+		} else if string(resp.Data) != "yes!" {
+			t.Fatalf("Expected a response of %q, got %q", "yes!", resp.Data)
+		}
+		subBar, err := ncImp.SubscribeSync(imSubExpect)
+		require_NoError(t, err)
+		ncImp.Flush()
+
+		ncExp.Publish(exPub, []byte("event!"))
+
+		if m, err := subBar.NextMsg(time.Second); err != nil {
+			t.Fatalf("Expected a stream message got %v", err)
+		} else if string(m.Data) != "event!" {
+			t.Fatalf("Expected a response of %q, got %q", "event!", m.Data)
+		}
+	}
+	createExporter := func() (string, string, string) {
+		t.Helper()
+		aExpKp, aExpPub := createKey(t)
+		aExpClaim := jwt.NewAccountClaims(aExpPub)
+		aExpClaim.Name = "Export"
+		aExpClaim.Exports.Add(&jwt.Export{
+			Subject: "$request.*.$in.*.>",
+			Type:    jwt.Service,
+		}, &jwt.Export{
+			Subject: "$events.*.$in.*.>",
+			Type:    jwt.Stream,
+		})
+		aExpJwt := encodeClaim(t, aExpClaim, aExpPub)
+		aExpCreds := newUser(t, aExpKp)
+		return aExpPub, aExpJwt, aExpCreds
+	}
+	t.Run("To", func(t *testing.T) {
+		aExpPub, aExpJwt, aExpCreds := createExporter()
+		defer os.Remove(aExpCreds)
+		aImpKp, aImpPub := createKey(t)
+		aImpClaim := jwt.NewAccountClaims(aImpPub)
+		aImpClaim.Name = "Import"
+		aImpClaim.Imports.Add(&jwt.Import{
+			Subject: "my.request.*.*.>",
+			Type:    jwt.Service,
+			To:      "$request.*.$in.*.>", // services have local and remote switched between Subject and To
+			Account: aExpPub,
+		}, &jwt.Import{
+			Subject: "$events.*.$in.*.>",
+			Type:    jwt.Stream,
+			To:      "prefix",
+			Account: aExpPub,
+		})
+		aImpJwt := encodeClaim(t, aImpClaim, aImpPub)
+		aImpCreds := newUser(t, aImpKp)
+		defer os.Remove(aImpCreds)
+		test(aExpPub, aExpJwt, aExpCreds, aImpPub, aImpJwt, aImpCreds,
+			"$request.1.$in.2.bar", "$events.1.$in.2.bar",
+			"my.request.1.2.bar", "prefix.$events.1.$in.2.bar")
+	})
+	t.Run("LocalSubject-No-Reorder", func(t *testing.T) {
+		aExpPub, aExpJwt, aExpCreds := createExporter()
+		defer os.Remove(aExpCreds)
+		aImpKp, aImpPub := createKey(t)
+		aImpClaim := jwt.NewAccountClaims(aImpPub)
+		aImpClaim.Name = "Import"
+		aImpClaim.Imports.Add(&jwt.Import{
+			Subject:      "$request.*.$in.*.>",
+			Type:         jwt.Service,
+			LocalSubject: "my.request.*.*.>",
+			Account:      aExpPub,
+		}, &jwt.Import{
+			Subject:      "$events.*.$in.*.>",
+			Type:         jwt.Stream,
+			LocalSubject: "my.events.*.*.>",
+			Account:      aExpPub,
+		})
+		aImpJwt := encodeClaim(t, aImpClaim, aImpPub)
+		aImpCreds := newUser(t, aImpKp)
+		defer os.Remove(aImpCreds)
+		test(aExpPub, aExpJwt, aExpCreds, aImpPub, aImpJwt, aImpCreds,
+			"$request.1.$in.2.bar", "$events.1.$in.2.bar",
+			"my.request.1.2.bar", "my.events.1.2.bar")
+	})
+	t.Run("LocalSubject-Reorder", func(t *testing.T) {
+		aExpPub, aExpJwt, aExpCreds := createExporter()
+		defer os.Remove(aExpCreds)
+		aImpKp, aImpPub := createKey(t)
+		aImpClaim := jwt.NewAccountClaims(aImpPub)
+		aImpClaim.Name = "Import"
+		aImpClaim.Imports.Add(&jwt.Import{
+			Subject:      "$request.*.$in.*.>",
+			Type:         jwt.Service,
+			LocalSubject: "my.request.$2.$1.>",
+			Account:      aExpPub,
+		}, &jwt.Import{
+			Subject:      "$events.*.$in.*.>",
+			Type:         jwt.Stream,
+			LocalSubject: "my.events.$2.$1.>",
+			Account:      aExpPub,
+		})
+		aImpJwt := encodeClaim(t, aImpClaim, aImpPub)
+		aImpCreds := newUser(t, aImpKp)
+		defer os.Remove(aImpCreds)
+		test(aExpPub, aExpJwt, aExpCreds, aImpPub, aImpJwt, aImpCreds,
+			"$request.2.$in.1.bar", "$events.1.$in.2.bar",
+			"my.request.1.2.bar", "my.events.2.1.bar")
+	})
+}
diff --git a/test/leafnode_test.go b/test/leafnode_test.go
index 25d974ea..c4d91486 100644
--- a/test/leafnode_test.go
+++ b/test/leafnode_test.go
@@ -4240,61 +4240,3 @@ func TestLeafNodeStreamAndShadowSubs(t *testing.T) {
 		t.Fatalf("Did not receive message: %v", err)
 	}
 }
-
-func TestLeafnodeHeaders(t *testing.T) {
-	srv, opts := runLeafServer()
-	defer srv.Shutdown()
-	leaf, _ := runSolicitLeafServer(opts)
-	defer leaf.Shutdown()
-
-	snc, err := nats.Connect(srv.ClientURL())
-	if err != nil {
-		t.Fatalf(err.Error())
-	}
-	defer snc.Close()
-	ssub, err := snc.SubscribeSync("test")
-	if err != nil {
-		t.Fatalf("subscribe failed: %s", err)
-	}
-
-	lnc, err := nats.Connect(leaf.ClientURL())
-	if err != nil {
-		t.Fatalf(err.Error())
-	}
-	defer lnc.Close()
-	lsub, err := lnc.SubscribeSync("test")
-	if err != nil {
-		t.Fatalf("subscribe failed: %s", err)
-	}
-	lnc.Flush()
-
-	checkLeafNodeConnected(t, srv)
-	checkLeafNodeConnected(t, leaf)
-	checkSubInterest(t, srv, "$G", "test", time.Second)
-
-	msg := nats.NewMsg("test")
-	msg.Header.Add("Test", "Header")
-	if len(msg.Header) == 0 {
-		t.Fatalf("msg header is empty")
-	}
-	err = snc.PublishMsg(msg)
-	if err != nil {
-		t.Fatalf(err.Error())
-	}
-
-	smsg, err := ssub.NextMsg(time.Second)
-	if err != nil {
-		t.Fatalf("next failed: %s", err)
-	}
-	if len(smsg.Header) == 0 {
-		t.Fatalf("server msgs header is empty")
-	}
-
-	lmsg, err := lsub.NextMsg(time.Second)
-	if err != nil {
-		t.Fatalf("next failed: %s", err)
-	}
-	if len(lmsg.Header) == 0 {
-		t.Fatalf("leaf msg header is empty")
-	}
-}