Repository: nats-io/nats-server
Pull request #3719 (2022-12-15 21:09:15)

Authorization Callouts
An implementation for a server-signed authorization request mechanism. The mechanism will be used in both server configuration and operator mode servers. This allows external NATS applications to securely authorize users with external AuthN.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core


ripienaar: I have a few comments / wishes to add here:

 * Can we extend the client connect event to include information that this was a user authorized by a callout and how long that call out took
 * Can we have events about callout fails which might include reasons from the auth callout side
 * Server should log that this is setup in static config mode, just a warning at start saying a callout is in use
 * In this pr the nil reply is all thats accepted, I think we should consider also accepting a json document with optional error details that could be logged or included in an event (first point) to asist with debugging

derekcollison: > I have a few comments / wishes to add here:
> 
> * Can we extend the client connect event to include information that this was a user authorized by a callout. Maybe include the issuer pub key.

This is visible in the account of the user and not the service, so might be considered leaking info.

> * Can we have events about callout fails which might include reasons from the auth callout side.

Phil asked about this as well. I think starting with server logs is right approach, and we can add in events. Not opposed.

> * Server should log that this is setup in static config mode, just a warning at start saying a callout is in use.

Can you give me an example of what would be helpful?

> * In this pr the nil reply is all thats accepted, I think we should consider also accepting a json document with optional error details that could be logged or included in an event (first point) to assist with debugging. JSON because we might want to extend later.

Possibly, we would need to go back into the jwt lib and add a response claim that would include the user claim such that we could trust the error response etc. So non-trivial.

> * Please include server tags in `server_id` so I can for example say some user is only allowed to connect in the eu

Good idea. I can add that, will also require update to jwt lib.

> * More logging

Examples?


ripienaar: > > I have a few comments / wishes to add here:
> > 
> > * Can we extend the client connect event to include information that this was a user authorized by a callout. Maybe include the issuer pub key.
> 
> This is visible in the account of the user and not the service, so might be considered leaking info.

I mean the ones that get sent to `$SYS.ACCOUNT.*.CONNECT` those are system only

> 
> > * Can we have events about callout fails which might include reasons from the auth callout side.
> 
> Phil asked about this as well. I think starting with server logs is right approach, and we can add in events. Not opposed.

fair

> 
> > * Server should log that this is setup in static config mode, just a warning at start saying a callout is in use.
> 
> Can you give me an example of what would be helpful?

If a server config significantly change the authentication parameters of behavior that's definitely something you want to communicate and warn about.  It stands out so there's no inadvertant mistakes or no surprises.

> > * In this pr the nil reply is all thats accepted, I think we should consider also accepting a json document with optional error details that could be logged or included in an event (first point) to assist with debugging. JSON because we might want to extend later.
> 
> Possibly, we would need to go back into the jwt lib and add a response claim that would include the user claim such that we could trust the error response etc. So non-trivial.
> 
> > * Please include server tags in `server_id` so I can for example say some user is only allowed to connect in the eu
> 
> Good idea. I can add that, will also require update to jwt lib.

cool

> 
> > * More logging
> 
> Examples?

At debug it would for sure be good to know you're calling out and when you got a response/or not/or timeout etc. It can be hard to debug this as its a bit of effect at a distance so debug info would help developers working on auth plugins.

derekcollison: Should also add in cluster to server portion if we are adding tags.

@ripienaar for the AuthroizationResponse, should it be code + description similar to how we evolved in JetStream?
ripienaar: Don’t think it needs code - since this is not for internal use it’s user supplied code so would be hard to standardise on coded errors. 

Yeah cluster would be good for sure 
derekcollison: So just error as a string an empty user_claim sounds good to you?

ripienaar: > So just error as a string an empty user_claim sounds good to you?

I thought maybe `{"error":"X"}` as a claim for now, we can also add additional fields later that way, where if we just do error string now it becomes a future pain to support both methods
derekcollison: I force pushed update that has the AuthorizationResponse now. 

Still working on the correct approach to auth error events offline with the team.

Patch:
diff --git a/README.md b/README.md
index 09f68f97..c91b6a39 100644
--- a/README.md
+++ b/README.md
@@ -37,8 +37,8 @@ If you are interested in contributing to NATS, read about our...
 [Fossa-Image]: https://app.fossa.io/api/projects/git%2Bgithub.com%2Fnats-io%2Fnats-server.svg?type=shield
 [Build-Status-Url]: https://travis-ci.com/github/nats-io/nats-server
 [Build-Status-Image]: https://travis-ci.com/nats-io/nats-server.svg?branch=main
-[Release-Url]: https://github.com/nats-io/nats-server/releases/tag/v2.9.10
-[Release-image]: https://img.shields.io/badge/release-v2.9.10-1eb0fc.svg
+[Release-Url]: https://github.com/nats-io/nats-server/releases/tag/v2.9.9
+[Release-image]: https://img.shields.io/badge/release-v2.9.9-1eb0fc.svg
 [Coverage-Url]: https://coveralls.io/r/nats-io/nats-server?branch=main
 [Coverage-image]: https://coveralls.io/repos/github/nats-io/nats-server/badge.svg?branch=main
 [ReportCard-Url]: https://goreportcard.com/report/nats-io/nats-server
diff --git a/docker/Dockerfile.nightly b/docker/Dockerfile.nightly
index 87f6d950..13fa7d54 100644
--- a/docker/Dockerfile.nightly
+++ b/docker/Dockerfile.nightly
@@ -9,7 +9,7 @@ RUN mkdir -p src/github.com/nats-io && \
     cd natscli/nats && \
     go build -ldflags "-w -X main.version=${VERSION}" -o /nats
 
-RUN go install github.com/nats-io/nsc/v2@latest
+RUN go install github.com/nats-io/nsc@latest
 
 FROM alpine:latest
 
diff --git a/go.mod b/go.mod
index 0bb92903..7d155cc2 100644
--- a/go.mod
+++ b/go.mod
@@ -5,13 +5,13 @@ go 1.19
 require (
 	github.com/klauspost/compress v1.15.11
 	github.com/minio/highwayhash v1.0.2
-	github.com/nats-io/jwt/v2 v2.3.0
+	github.com/nats-io/jwt/v2 v2.3.1-0.20221227170542-bdf40fa3627b
 	github.com/nats-io/nats.go v1.19.0
-	github.com/nats-io/nkeys v0.3.0
+	github.com/nats-io/nkeys v0.3.1-0.20221215194120-47c7408e7546
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.5.1
-	golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be
-	golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec
+	golang.org/x/crypto v0.3.0
+	golang.org/x/sys v0.2.0
 	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af
 )
 
diff --git a/go.sum b/go.sum
index 0198d625..53ebeebb 100644
--- a/go.sum
+++ b/go.sum
@@ -13,12 +13,13 @@ github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B
 github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
-github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
-github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
+github.com/nats-io/jwt/v2 v2.3.1-0.20221227170542-bdf40fa3627b h1:exHeHbghpBp1JvdYq7muaKFvJgLD93UDcmoIbFu/9PA=
+github.com/nats-io/jwt/v2 v2.3.1-0.20221227170542-bdf40fa3627b/go.mod h1:DYujvzCMZzUuqB3i1Pnpf1YtkuTwhdI84Aah9wRXkK0=
 github.com/nats-io/nats.go v1.19.0 h1:H6j8aBnTQFoVrTGB6Xjd903UMdE7jz6DS4YkmAqgZ9Q=
 github.com/nats-io/nats.go v1.19.0/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA=
-github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
+github.com/nats-io/nkeys v0.3.1-0.20221215194120-47c7408e7546 h1:7ZylVLLiDSFqxJTSibNsO2RjVSXj3QWnDc+zKara2HE=
+github.com/nats-io/nkeys v0.3.1-0.20221215194120-47c7408e7546/go.mod h1:JOEZlxMfMnmaLwr+mpmP+RGIYSxLNBFsZykCGaI2PvA=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -27,13 +28,13 @@ github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMT
 go.uber.org/automaxprocs v1.5.1 h1:e1YG66Lrk73dn4qhg8WFSvhF0JuFQF0ERIp4rpuV8Qk=
 go.uber.org/automaxprocs v1.5.1/go.mod h1:BF4eumQw0P9GtnuxxovUd06vwm1o18oMzFtK66vU6XU=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be h1:fmw3UbQh+nxngCAHrDCCztao/kbYFnWjoqop8dHx05A=
-golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.3.0 h1:a06MkbcxBrEFc0w0QIZWXrH/9cCX6KJyWbBOIwAn+7A=
+golang.org/x/crypto v0.3.0/go.mod h1:hebNnKkNXi2UzZN1eVRvBB7co0a+JxK6XbPiWVs/3J4=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec h1:BkDtF2Ih9xZ7le9ndzTA7KJow28VbQW3odyk/8drmuI=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.2.0 h1:ljd4t30dBnAvMZaQCevtY0xLLD0A+bRZXbgLMLU1F/A=
+golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20220922220347-f3bd1da661af h1:Yx9k8YCG3dvF87UAn2tu2HQLf2dt/eR1bXxpLMWeH+Y=
diff --git a/server/accounts.go b/server/accounts.go
index f39032ab..38f44de0 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -82,6 +82,7 @@ type Account struct {
 	expired      bool
 	incomplete   bool
 	signingKeys  map[string]jwt.Scope
+	extAuth      *jwt.ExternalAuthorization
 	srv          *Server // server this account is registered with (possibly nil)
 	lds          string  // loop detection subject for leaf nodes
 	siReply      []byte  // service reply prefix, will form wildcard subscription.
@@ -1928,6 +1929,13 @@ func (a *Account) subscribeInternal(subject string, cb msgHandler) (*subscriptio
 	return a.subscribeInternalEx(subject, cb, false)
 }
 
+// Unsubscribe from an internal account subscription.
+func (a *Account) unsubscribeInternal(sub *subscription) {
+	if ic := a.internalClient(); ic != nil {
+		ic.processUnsub(sub.sid)
+	}
+}
+
 // Creates internal subscription for service import responses.
 func (a *Account) subscribeServiceImportResponse(subject string) (*subscription, error) {
 	return a.subscribeInternalEx(subject, a.processServiceImportResponse, true)
@@ -2984,6 +2992,65 @@ func (a *Account) traceLabel() string {
 	return a.Name
 }
 
+// Check if an account has external auth set.
+// Operator/Account Resolver only.
+func (a *Account) hasExternalAuth() bool {
+	if a == nil {
+		return false
+	}
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return a.extAuth != nil
+}
+
+// Deterimine if this is an external auth user.
+func (a *Account) isExternalAuthUser(userID string) bool {
+	if a == nil {
+		return false
+	}
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	if a.extAuth != nil {
+		for _, u := range a.extAuth.AuthUsers {
+			if userID == u {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// Return the external authorization xkey if external authorization is enabled and the xkey is set.
+// Operator/Account Resolver only.
+func (a *Account) externalAuthXKey() string {
+	if a == nil {
+		return _EMPTY_
+	}
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	if a.extAuth != nil && a.extAuth.XKey != _EMPTY_ {
+		return a.extAuth.XKey
+	}
+	return _EMPTY_
+}
+
+// Check if an account switch for external authorization is allowed.
+func (a *Account) isAllowedAcount(acc string) bool {
+	if a == nil {
+		return false
+	}
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	if a.extAuth != nil {
+		for _, a := range a.extAuth.AllowedAccounts {
+			if a == acc {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // updateAccountClaimsWithRefresh will update an existing account with new claims.
 // If refreshImportingAccounts is true it will also update incomplete dependent accounts
 // This will replace any exports or imports previously defined.
@@ -3003,6 +3070,14 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 	a.nameTag = ac.Name
 	a.tags = ac.Tags
 
+	// Check for external authorization.
+	if ac.HasExternalAuthorization() {
+		a.extAuth = &jwt.ExternalAuthorization{}
+		a.extAuth.AuthUsers.Add(ac.Authorization.AuthUsers...)
+		a.extAuth.AllowedAccounts.Add(ac.Authorization.AllowedAccounts...)
+		a.extAuth.XKey = ac.Authorization.XKey
+	}
+
 	// Reset exports and imports here.
 
 	// Exports is creating a whole new map.
diff --git a/server/auth.go b/server/auth.go
index b6944c31..c7a440e2 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2022 The NATS Authors
+// Copyright 2012-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -262,7 +262,7 @@ func (s *Server) configureAuthorization() {
 	} else if opts.Nkeys != nil || opts.Users != nil {
 		s.nkeys, s.users = s.buildNkeysAndUsersFromOptions(opts.Nkeys, opts.Users)
 		s.info.AuthRequired = true
-	} else if opts.Username != "" || opts.Authorization != "" {
+	} else if opts.Username != _EMPTY_ || opts.Authorization != _EMPTY_ {
 		s.info.AuthRequired = true
 	} else {
 		s.users = nil
@@ -274,6 +274,27 @@ func (s *Server) configureAuthorization() {
 	s.wsConfigAuth(&opts.Websocket)
 	// And for mqtt config
 	s.mqttConfigAuth(&opts.MQTT)
+
+	// Check for server configured auth callouts.
+	if opts.AuthCallout != nil {
+		// Make sure we have a valid account and auth_users.
+		_, err := s.lookupAccount(opts.AuthCallout.Account)
+		if err != nil {
+			s.Errorf("Authorization callout account %q not valid", opts.AuthCallout.Account)
+		}
+		for _, u := range opts.AuthCallout.AuthUsers {
+			// Check for user in users and nkeys since this is server config.
+			var found bool
+			if len(s.users) > 0 {
+				_, found = s.users[u]
+			} else if len(s.nkeys) > 0 && !found {
+				_, found = s.nkeys[u]
+			}
+			if !found {
+				s.Errorf("Authorization callout user %q not valid: %v", u, err)
+			}
+		}
+	}
 }
 
 // Takes the given slices of NkeyUser and User options and build
@@ -547,7 +568,7 @@ func processUserPermissionsTemplate(lim jwt.UserPermissionLimits, ujwt *jwt.User
 	return lim, nil
 }
 
-func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) bool {
+func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (authorized bool) {
 	var (
 		nkey *NkeyUser
 		juc  *jwt.UserClaims
@@ -557,6 +578,70 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 		err  error
 		ao   bool // auth override
 	)
+
+	// Check if we have auth callouts enabled at the server level or in the bound account.
+	defer func() {
+		// Default reason
+		reason := AuthenticationViolation.String()
+		// No-op
+		if juc == nil && opts.AuthCallout == nil {
+			if !authorized {
+				s.sendAccountAuthErrorEvent(c, c.acc, reason)
+			}
+			return
+		}
+		// We have a juc defined here, check account.
+		if juc != nil && !acc.hasExternalAuth() {
+			if !authorized {
+				s.sendAccountAuthErrorEvent(c, c.acc, reason)
+			}
+			return
+		}
+
+		// We have auth callout set here.
+		var skip bool
+		// Check if we are on the list of auth_users.
+		userID := c.getRawAuthUser()
+		if juc != nil {
+			skip = acc.isExternalAuthUser(userID)
+		} else {
+			for _, u := range opts.AuthCallout.AuthUsers {
+				if userID == u {
+					skip = true
+					break
+				}
+			}
+		}
+
+		// If we are here we have an auth callout defined and we have failed auth so far
+		// so we will callout to our auth backend for processing.
+		if !skip {
+			authorized, reason = s.processClientOrLeafCallout(c, opts)
+		}
+		// Check if we are authorized and in the auth callout account, and if so add in deny publish permissions for the auth subject.
+		if authorized {
+			var authAccountName string
+			if juc == nil && opts.AuthCallout != nil {
+				authAccountName = opts.AuthCallout.Account
+			} else if juc != nil {
+				authAccountName = acc.Name
+			}
+			c.mu.Lock()
+			if c.acc != nil && c.acc.Name == authAccountName {
+				c.mergeDenyPermissions(pub, []string{AuthCalloutSubject})
+			}
+			c.mu.Unlock()
+		} else {
+			// If we are here we failed external authorization.
+			// Send an account scoped event. Server config mode acc will be nil,
+			// so lookup the auth callout assigned account, that is where this will be sent.
+			if acc == nil {
+				acc, _ = s.lookupAccount(opts.AuthCallout.Account)
+			}
+			s.sendAccountAuthErrorEvent(c, acc, reason)
+		}
+	}()
+
 	s.mu.Lock()
 	authRequired := s.info.AuthRequired
 	if !authRequired {
@@ -812,7 +897,7 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 			return false
 		}
 		if juc.BearerToken && acc.failBearer() {
-			c.Debugf("Account does not allow bearer token")
+			c.Debugf("Account does not allow bearer tokens")
 			return false
 		}
 		// skip validation of nonce when presented with a bearer token
@@ -1037,7 +1122,7 @@ func checkClientTLSCertSubject(c *client, fn tlsMapAuthFn) bool {
 	// https://github.com/golang/go/issues/12342
 	dn, err := ldap.FromRawCertSubject(cert.RawSubject)
 	if err == nil {
-		if match, ok := fn("", dn, false); ok {
+		if match, ok := fn(_EMPTY_, dn, false); ok {
 			c.Debugf("Using DistinguishedNameMatch for auth [%q]", match)
 			return true
 		}
@@ -1120,22 +1205,22 @@ func (s *Server) isRouterAuthorized(c *client) bool {
 
 	if opts.Cluster.TLSMap || opts.Cluster.TLSCheckKnownURLs {
 		return checkClientTLSCertSubject(c, func(user string, _ *ldap.DN, isDNSAltName bool) (string, bool) {
-			if user == "" {
-				return "", false
+			if user == _EMPTY_ {
+				return _EMPTY_, false
 			}
 			if opts.Cluster.TLSCheckKnownURLs && isDNSAltName {
 				if dnsAltNameMatches(dnsAltNameLabels(user), opts.Routes) {
-					return "", true
+					return _EMPTY_, true
 				}
 			}
 			if opts.Cluster.TLSMap && opts.Cluster.Username == user {
-				return "", true
+				return _EMPTY_, true
 			}
-			return "", false
+			return _EMPTY_, false
 		})
 	}
 
-	if opts.Cluster.Username == "" {
+	if opts.Cluster.Username == _EMPTY_ {
 		return true
 	}
 
@@ -1156,25 +1241,25 @@ func (s *Server) isGatewayAuthorized(c *client) bool {
 	// Check whether TLS map is enabled, otherwise use single user/pass.
 	if opts.Gateway.TLSMap || opts.Gateway.TLSCheckKnownURLs {
 		return checkClientTLSCertSubject(c, func(user string, _ *ldap.DN, isDNSAltName bool) (string, bool) {
-			if user == "" {
-				return "", false
+			if user == _EMPTY_ {
+				return _EMPTY_, false
 			}
 			if opts.Gateway.TLSCheckKnownURLs && isDNSAltName {
 				labels := dnsAltNameLabels(user)
 				for _, gw := range opts.Gateway.Gateways {
 					if gw != nil && dnsAltNameMatches(labels, gw.URLs) {
-						return "", true
+						return _EMPTY_, true
 					}
 				}
 			}
 			if opts.Gateway.TLSMap && opts.Gateway.Username == user {
-				return "", true
+				return _EMPTY_, true
 			}
-			return "", false
+			return _EMPTY_, false
 		})
 	}
 
-	if opts.Gateway.Username == "" {
+	if opts.Gateway.Username == _EMPTY_ {
 		return true
 	}
 
diff --git a/server/auth_callout.go b/server/auth_callout.go
new file mode 100644
index 00000000..73fab2da
--- /dev/null
+++ b/server/auth_callout.go
@@ -0,0 +1,359 @@
+// Copyright 2022-2023 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/pem"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nkeys"
+)
+
+const (
+	AuthCalloutSubject    = "$SYS.REQ.USER.AUTH"
+	AuthRequestSubject    = "nats-authorization-request"
+	AuthRequestXKeyHeader = "Nats-Server-Xkey"
+)
+
+// Process a callout on this client's behalf.
+func (s *Server) processClientOrLeafCallout(c *client, opts *Options) (authorized bool, errStr string) {
+	isOperatorMode := len(opts.TrustedKeys) > 0
+
+	var acc *Account
+	if !isOperatorMode && opts.AuthCallout != nil && opts.AuthCallout.Account != _EMPTY_ {
+		aname := opts.AuthCallout.Account
+		var err error
+		acc, err = s.LookupAccount(aname)
+		if err != nil {
+			errStr = fmt.Sprintf("No valid account %q for auth callout request: %v", aname, err)
+			s.Warnf(errStr)
+			return false, errStr
+		}
+	} else {
+		acc = c.acc
+	}
+
+	// Check if we have been requested to encrypt.
+	var xkp nkeys.KeyPair
+	var xkey string
+	var pubAccXKey string
+	if !isOperatorMode && opts.AuthCallout != nil && opts.AuthCallout.XKey != _EMPTY_ {
+		pubAccXKey = opts.AuthCallout.XKey
+	} else if isOperatorMode {
+		pubAccXKey = acc.externalAuthXKey()
+	}
+	// If set grab server's xkey keypair and public key.
+	if pubAccXKey != _EMPTY_ {
+		// These are only set on creation, so lock not needed.
+		xkp, xkey = s.xkp, s.info.XKey
+	}
+
+	// Create a keypair for the user. We will expect this public user to be in the signed response.
+	// This prevents replay attacks.
+	ukp, _ := nkeys.CreateUser()
+	pub, _ := ukp.PublicKey()
+
+	reply := s.newRespInbox()
+	respCh := make(chan string, 1)
+
+	processReply := func(_ *subscription, rc *client, racc *Account, subject, reply string, rmsg []byte) {
+		_, msg := rc.msgParts(rmsg)
+		// This signals not authorized.
+		// Since this is an account subscription will always have "\r\n".
+		if len(msg) <= LEN_CR_LF {
+			respCh <- fmt.Sprintf("Auth callout violation: %q on account %q", "no reason supplied", racc.Name)
+			return
+		}
+		// Strip trailing CRLF.
+		msg = msg[:len(msg)-LEN_CR_LF]
+
+		// If we sent an encrypted request the response could be encrypted as well.
+		if xkp != nil && len(msg) > len(jwtPrefix) && !bytes.HasPrefix(msg, []byte(jwtPrefix)) {
+			var err error
+			msg, err = xkp.Open(msg, pubAccXKey)
+			if err != nil {
+				respCh <- fmt.Sprintf("Error decrypting auth callout response on account %q: %v", racc.Name, err)
+				return
+			}
+		}
+
+		arc, err := jwt.DecodeAuthorizationResponseClaims(string(msg))
+		if err != nil {
+			respCh <- fmt.Sprintf("Error decoding auth callout response on account %q: %v", racc.Name, err)
+			return
+		}
+
+		// FIXME(dlc) - push error through here.
+		if arc.Error != nil || arc.User == nil {
+			if arc.Error != nil {
+				respCh <- fmt.Sprintf("Auth callout violation: %q on account %q", arc.Error.Description, racc.Name)
+			} else {
+				respCh <- fmt.Sprintf("Auth callout violation: no user returned  on account %q", racc.Name)
+			}
+			return
+		}
+
+		// Make sure correct issuer.
+		var issuer string
+		if opts.AuthCallout != nil {
+			issuer = opts.AuthCallout.Issuer
+		} else {
+			// Operator mode is who we send the request on unless switching accounts.
+			issuer = acc.Name
+		}
+		// By default issuer needs to match server config or the requesting account in operator mode.
+		if arc.Issuer != issuer {
+			if !isOperatorMode {
+				respCh <- fmt.Sprintf("Wrong issuer for auth callout response on account %q, expected %q got %q", racc.Name, issuer, arc.Issuer)
+				return
+			} else if !acc.isAllowedAcount(arc.Issuer) {
+				respCh <- fmt.Sprintf("Account %q not permitted as valid account option for auth callout for account %q",
+					arc.Issuer, racc.Name)
+				return
+			}
+		}
+
+		// Require the response to have pinned the audience to this server.
+		if arc.Audience != s.info.ID {
+			respCh <- fmt.Sprintf("Wrong server audience received for auth callout response on account %q, expected %q got %q",
+				racc.Name, s.info.ID, arc.Audience)
+			return
+		}
+
+		juc := arc.User
+		// Make sure that the user is what we requested.
+		if juc.Subject != pub {
+			respCh <- fmt.Sprintf("Expected authorized user of %q but got %q on account %q", pub, juc.Subject, racc.Name)
+			return
+		}
+
+		allowNow, validFor := validateTimes(juc)
+		if !allowNow {
+			c.Errorf("Outside connect times")
+			respCh <- fmt.Sprintf("Authorized user on account %q outside of valid connect times", racc.Name)
+			return
+		}
+		allowedConnTypes, err := convertAllowedConnectionTypes(juc.AllowedConnectionTypes)
+		if err != nil {
+			c.Debugf("%v", err)
+			if len(allowedConnTypes) == 0 {
+				respCh <- fmt.Sprintf("Authorized user on account %q using invalid connection type", racc.Name)
+				return
+			}
+		}
+		// Apply to this client.
+		targetAcc := acc
+		// Check if we are being asked to switch accounts.
+		if aname := juc.Audience; aname != _EMPTY_ {
+			targetAcc, err = s.LookupAccount(aname)
+			if err != nil {
+				respCh <- fmt.Sprintf("No valid account %q for auth callout response on account %q: %v", aname, racc.Name, err)
+				return
+			}
+			// In operator mode make sure this account matches the issuer.
+			if isOperatorMode && aname != arc.Issuer {
+				respCh <- fmt.Sprintf("Account %q does not match issuer %q", aname, juc.Issuer)
+				return
+			}
+		}
+
+		// Build internal user and bind to the targeted account.
+		nkuser := buildInternalNkeyUser(juc, allowedConnTypes, targetAcc)
+		if err := c.RegisterNkeyUser(nkuser); err != nil {
+			respCh <- fmt.Sprintf("Could not register auth callout user: %v", err)
+			return
+		}
+
+		// See if the response wants to override the username.
+		if juc.Name != _EMPTY_ {
+			c.mu.Lock()
+			c.opts.Username = juc.Name
+			// Clear any others.
+			c.opts.Nkey = _EMPTY_
+			c.pubKey = _EMPTY_
+			c.opts.Token = _EMPTY_
+			c.mu.Unlock()
+		}
+
+		// Check if we need to set an auth timer if the user jwt expires.
+		c.setExpiration(juc.Claims(), validFor)
+
+		respCh <- _EMPTY_
+	}
+
+	sub, err := acc.subscribeInternal(reply, processReply)
+	if err != nil {
+		errStr = fmt.Sprintf("Error setting up reply subscription for auth request: %v", err)
+		s.Warnf(errStr)
+		return false, errStr
+	}
+	defer acc.unsubscribeInternal(sub)
+
+	// Build our request claims.
+	claim := jwt.NewAuthorizationRequestClaims(AuthRequestSubject)
+	// Set expected public user nkey.
+	claim.UserNkey = pub
+
+	s.mu.RLock()
+	claim.Server = jwt.ServerID{
+		Name:    s.info.Name,
+		Host:    s.info.Host,
+		ID:      s.info.ID,
+		Version: s.info.Version,
+		Cluster: s.info.Cluster,
+	}
+	s.mu.RUnlock()
+
+	// Tags
+	claim.Server.Tags = s.getOpts().Tags
+
+	// Check if we have been requested to encrypt.
+	if xkp != nil {
+		claim.Server.XKey = xkey
+	}
+
+	authTimeout := secondsToDuration(s.getOpts().AuthTimeout)
+	claim.Expires = time.Now().Add(time.Duration(authTimeout)).UTC().Unix()
+	if opts.AuthCallout != nil {
+		claim.Audience = opts.AuthCallout.Issuer
+	} else {
+		claim.Audience = acc.Name
+	}
+
+	// Grab client info for the request.
+	c.mu.Lock()
+	c.fillClientInfo(&claim.ClientInformation)
+	c.fillConnectOpts(&claim.ConnectOptions)
+	// If we have a sig in the client opts, fill in nonce.
+	if claim.ConnectOptions.SignedNonce != _EMPTY_ {
+		claim.ClientInformation.Nonce = string(c.nonce)
+	}
+
+	// TLS
+	if c.flags.isSet(handshakeComplete) && c.nc != nil {
+		var ct jwt.ClientTLS
+		conn := c.nc.(*tls.Conn)
+		cs := conn.ConnectionState()
+		ct.Version = tlsVersion(cs.Version)
+		ct.Cipher = tlsCipher(cs.CipherSuite)
+		// Check verified chains.
+		for _, vs := range cs.VerifiedChains {
+			var certs []string
+			for _, c := range vs {
+				blk := &pem.Block{
+					Type:  "CERTIFICATE",
+					Bytes: c.Raw,
+				}
+				certs = append(certs, string(pem.EncodeToMemory(blk)))
+			}
+			ct.VerifiedChains = append(ct.VerifiedChains, certs)
+		}
+		// If we do not have verified chains put in peer certs.
+		if len(ct.VerifiedChains) == 0 {
+			for _, c := range cs.PeerCertificates {
+				blk := &pem.Block{
+					Type:  "CERTIFICATE",
+					Bytes: c.Raw,
+				}
+				ct.Certs = append(ct.Certs, string(pem.EncodeToMemory(blk)))
+			}
+		}
+		claim.TLS = &ct
+	}
+	c.mu.Unlock()
+
+	b, err := claim.Encode(s.kp)
+	if err != nil {
+		errStr = fmt.Sprintf("Error encoding auth request claim on account %q: %v", acc.Name, err)
+		s.Warnf(errStr)
+		return false, errStr
+	}
+	req := []byte(b)
+	var hdr map[string]string
+
+	// Check if we have been asked to encrypt.
+	if xkp != nil {
+		req, err = xkp.Seal([]byte(req), pubAccXKey)
+		if err != nil {
+			errStr = fmt.Sprintf("Error encrypting auth request claim on account %q: %v", acc.Name, err)
+			s.Warnf(errStr)
+			return false, errStr
+		}
+		hdr = map[string]string{AuthRequestXKeyHeader: xkey}
+	}
+
+	// Send out our request.
+	s.sendInternalAccountMsgWithReply(acc, AuthCalloutSubject, reply, hdr, req, false)
+
+	select {
+	case errStr = <-respCh:
+		if authorized = errStr == _EMPTY_; !authorized {
+			s.Warnf(errStr)
+		}
+	case <-time.After(authTimeout):
+		errStr = fmt.Sprintf("Authorization callout response not received in time on account %q", acc.Name)
+		s.Debugf(errStr)
+	}
+
+	return authorized, errStr
+}
+
+// Fill in client information for the request.
+// Lock should be held.
+func (c *client) fillClientInfo(ci *jwt.ClientInformation) {
+	if c == nil || (c.kind != CLIENT && c.kind != LEAF && c.kind != JETSTREAM && c.kind != ACCOUNT) {
+		return
+	}
+
+	// Do it this way to fail to compile if fields are added to jwt.ClientInformation.
+	*ci = jwt.ClientInformation{
+		Host:    c.host,
+		ID:      c.cid,
+		User:    c.getRawAuthUser(),
+		Name:    c.opts.Name,
+		Tags:    c.tags,
+		NameTag: c.nameTag,
+		Kind:    c.kindString(),
+		Type:    c.clientTypeString(),
+		MQTT:    c.getMQTTClientID(),
+	}
+}
+
+// Fill in client options.
+// Lock should be held.
+func (c *client) fillConnectOpts(opts *jwt.ConnectOptions) {
+	if c == nil || (c.kind != CLIENT && c.kind != LEAF && c.kind != JETSTREAM && c.kind != ACCOUNT) {
+		return
+	}
+
+	o := c.opts
+
+	// Do it this way to fail to compile if fields are added to jwt.ClientInformation.
+	*opts = jwt.ConnectOptions{
+		JWT:         o.JWT,
+		Nkey:        o.Nkey,
+		SignedNonce: o.Sig,
+		Token:       o.Token,
+		Username:    o.Username,
+		Password:    o.Password,
+		Name:        o.Name,
+		Lang:        o.Lang,
+		Version:     o.Version,
+		Protocol:    o.Protocol,
+	}
+}
diff --git a/server/client.go b/server/client.go
index cbfa3b6f..f00a3222 100644
--- a/server/client.go
+++ b/server/client.go
@@ -1987,14 +1987,14 @@ func (c *client) authViolation() {
 	var s *Server
 	var hasTrustedNkeys, hasNkeys, hasUsers bool
 	if s = c.srv; s != nil {
-		s.mu.Lock()
+		s.mu.RLock()
 		hasTrustedNkeys = s.trustedKeys != nil
 		hasNkeys = s.nkeys != nil
 		hasUsers = s.users != nil
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		defer s.sendAuthErrorEvent(c)
-
 	}
+
 	if hasTrustedNkeys {
 		c.Errorf("%v", ErrAuthentication)
 	} else if hasNkeys {
diff --git a/server/const.go b/server/const.go
index c1ed0900..de74f16f 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.9.11-beta"
+	VERSION = "2.10.0-beta.6"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index aceaf590..2240bf2f 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1122,7 +1122,10 @@ func (o *consumer) setLeader(isLeader bool) {
 }
 
 func (o *consumer) handleClusterConsumerInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
-	o.infoWithSnapAndReply(false, reply)
+	o.mu.RLock()
+	sysc := o.sysc
+	o.mu.RUnlock()
+	sysc.sendInternalMsg(reply, _EMPTY_, nil, o.info())
 }
 
 // Lock should be held.
@@ -2199,27 +2202,23 @@ func (o *consumer) info() *ConsumerInfo {
 }
 
 func (o *consumer) infoWithSnap(snap bool) *ConsumerInfo {
-	return o.infoWithSnapAndReply(snap, _EMPTY_)
-}
-
-func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
-	o.mu.Lock()
+	o.mu.RLock()
 	mset := o.mset
 	if mset == nil || mset.srv == nil {
-		o.mu.Unlock()
+		o.mu.RUnlock()
 		return nil
 	}
 	js := o.js
+	o.mu.RUnlock()
+
 	if js == nil {
-		o.mu.Unlock()
 		return nil
 	}
 
-	// Capture raftGroup.
-	var rg *raftGroup
-	if o.ca != nil {
-		rg = o.ca.Group
-	}
+	ci := js.clusterInfo(o.raftGroup())
+
+	o.mu.Lock()
+	defer o.mu.Unlock()
 
 	cfg := o.cfg
 	info := &ConsumerInfo{
@@ -2239,6 +2238,7 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		NumRedelivered: len(o.rdc),
 		NumPending:     o.streamNumPending(),
 		PushBound:      o.isPushMode() && o.active,
+		Cluster:        ci,
 	}
 	// Adjust active based on non-zero etc. Also make UTC here.
 	if !o.ldt.IsZero() {
@@ -2259,18 +2259,6 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 	if snap {
 		o.ici = info
 	}
-	sysc := o.sysc
-	o.mu.Unlock()
-
-	// Do cluster.
-	if rg != nil {
-		info.Cluster = js.clusterInfo(rg)
-	}
-
-	// If we have a reply subject send the response here.
-	if reply != _EMPTY_ && sysc != nil {
-		sysc.sendInternalMsg(reply, _EMPTY_, nil, info)
-	}
 
 	return info
 }
diff --git a/server/events.go b/server/events.go
index ac7e11aa..dce08086 100644
--- a/server/events.go
+++ b/server/events.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2022 The NATS Authors
+// Copyright 2018-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -47,20 +47,21 @@ const (
 	accPingReqSubj      = "$SYS.REQ.ACCOUNT.PING.%s" // atm. only used for STATZ and CONNZ import from system account
 	// kept for backward compatibility when using http resolver
 	// this overlaps with the names for events but you'd have to have the operator private key in order to succeed.
-	accUpdateEventSubjOld    = "$SYS.ACCOUNT.%s.CLAIMS.UPDATE"
-	accUpdateEventSubjNew    = "$SYS.REQ.ACCOUNT.%s.CLAIMS.UPDATE"
-	connsRespSubj            = "$SYS._INBOX_.%s"
-	accConnsEventSubjNew     = "$SYS.ACCOUNT.%s.SERVER.CONNS"
-	accConnsEventSubjOld     = "$SYS.SERVER.ACCOUNT.%s.CONNS" // kept for backward compatibility
-	shutdownEventSubj        = "$SYS.SERVER.%s.SHUTDOWN"
-	authErrorEventSubj       = "$SYS.SERVER.%s.CLIENT.AUTH.ERR"
-	serverStatsSubj          = "$SYS.SERVER.%s.STATSZ"
-	serverDirectReqSubj      = "$SYS.REQ.SERVER.%s.%s"
-	serverPingReqSubj        = "$SYS.REQ.SERVER.PING.%s"
-	serverStatsPingReqSubj   = "$SYS.REQ.SERVER.PING"             // use $SYS.REQ.SERVER.PING.STATSZ instead
-	leafNodeConnectEventSubj = "$SYS.ACCOUNT.%s.LEAFNODE.CONNECT" // for internal use only
-	remoteLatencyEventSubj   = "$SYS.LATENCY.M2.%s"
-	inboxRespSubj            = "$SYS._INBOX.%s.%s"
+	accUpdateEventSubjOld     = "$SYS.ACCOUNT.%s.CLAIMS.UPDATE"
+	accUpdateEventSubjNew     = "$SYS.REQ.ACCOUNT.%s.CLAIMS.UPDATE"
+	connsRespSubj             = "$SYS._INBOX_.%s"
+	accConnsEventSubjNew      = "$SYS.ACCOUNT.%s.SERVER.CONNS"
+	accConnsEventSubjOld      = "$SYS.SERVER.ACCOUNT.%s.CONNS" // kept for backward compatibility
+	shutdownEventSubj         = "$SYS.SERVER.%s.SHUTDOWN"
+	authErrorEventSubj        = "$SYS.SERVER.%s.CLIENT.AUTH.ERR"
+	authErrorAccountEventSubj = "$SYS.ACCOUNT.CLIENT.AUTH.ERR"
+	serverStatsSubj           = "$SYS.SERVER.%s.STATSZ"
+	serverDirectReqSubj       = "$SYS.REQ.SERVER.%s.%s"
+	serverPingReqSubj         = "$SYS.REQ.SERVER.PING.%s"
+	serverStatsPingReqSubj    = "$SYS.REQ.SERVER.PING"             // use $SYS.REQ.SERVER.PING.STATSZ instead
+	leafNodeConnectEventSubj  = "$SYS.ACCOUNT.%s.LEAFNODE.CONNECT" // for internal use only
+	remoteLatencyEventSubj    = "$SYS.LATENCY.M2.%s"
+	inboxRespSubj             = "$SYS._INBOX.%s.%s"
 
 	// Used to return information to a user on bound account and user permissions.
 	userDirectInfoSubj = "$SYS.REQ.USER.INFO"
@@ -193,7 +194,7 @@ type ClientInfo struct {
 	Start      *time.Time    `json:"start,omitempty"`
 	Host       string        `json:"host,omitempty"`
 	ID         uint64        `json:"id,omitempty"`
-	Account    string        `json:"acc"`
+	Account    string        `json:"acc,omitempty"`
 	Service    string        `json:"svc,omitempty"`
 	User       string        `json:"user,omitempty"`
 	Name       string        `json:"name,omitempty"`
@@ -211,6 +212,7 @@ type ClientInfo struct {
 	Kind       string        `json:"kind,omitempty"`
 	ClientType string        `json:"client_type,omitempty"`
 	MQTTClient string        `json:"client_id,omitempty"` // This is the MQTT client ID
+	Nonce      string        `json:"nonce,omitempty"`
 }
 
 // ServerStats hold various statistics that we will periodically send out.
@@ -510,6 +512,23 @@ func (s *Server) sendInternalAccountMsgWithReply(a *Account, subject, reply stri
 	return nil
 }
 
+// Send system style message to an account scope.
+func (s *Server) sendInternalAccountSysMsg(a *Account, subj string, si *ServerInfo, msg interface{}) {
+	s.mu.RLock()
+	if s.sys == nil || s.sys.sendq == nil || a == nil {
+		s.mu.RUnlock()
+		return
+	}
+	sendq := s.sys.sendq
+	s.mu.RUnlock()
+
+	a.mu.Lock()
+	c := a.internalClient()
+	a.mu.Unlock()
+
+	sendq.push(newPubMsg(c, subj, _EMPTY_, si, nil, msg, noCompression, false, false))
+}
+
 // This will queue up a message to be sent.
 // Lock should not be held.
 func (s *Server) sendInternalMsgLocked(subj, rply string, si *ServerInfo, msg interface{}) {
@@ -2015,6 +2034,7 @@ func (s *Server) accountDisconnectEvent(c *client, now time.Time, reason string)
 	s.sendInternalMsgLocked(subj, _EMPTY_, &m.Server, &m)
 }
 
+// This is the system level event sent to the system account for operators.
 func (s *Server) sendAuthErrorEvent(c *client) {
 	s.mu.Lock()
 	if !s.eventsEnabled() {
@@ -2069,6 +2089,61 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 	s.mu.Unlock()
 }
 
+// This is the account level event sent to the origin account for account owners.
+func (s *Server) sendAccountAuthErrorEvent(c *client, acc *Account, reason string) {
+	if acc == nil {
+		return
+	}
+	s.mu.Lock()
+	if !s.eventsEnabled() {
+		s.mu.Unlock()
+		return
+	}
+	eid := s.nextEventID()
+	s.mu.Unlock()
+
+	now := time.Now().UTC()
+	c.mu.Lock()
+	m := DisconnectEventMsg{
+		TypedEvent: TypedEvent{
+			Type: DisconnectEventMsgType,
+			ID:   eid,
+			Time: now,
+		},
+		Client: ClientInfo{
+			Start:      &c.start,
+			Stop:       &now,
+			Host:       c.host,
+			ID:         c.cid,
+			Account:    acc.Name,
+			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,
+			Kind:       c.kindString(),
+			ClientType: c.clientTypeString(),
+			MQTTClient: c.getMQTTClientID(),
+		},
+		Sent: DataStats{
+			Msgs:  c.inMsgs,
+			Bytes: c.inBytes,
+		},
+		Received: DataStats{
+			Msgs:  c.outMsgs,
+			Bytes: c.outBytes,
+		},
+		Reason: reason,
+	}
+	c.mu.Unlock()
+
+	s.sendInternalAccountSysMsg(acc, authErrorAccountEventSubj, &m.Server, &m)
+}
+
 // Internal message callback.
 // If the msg is needed past the callback it is required to be copied.
 // rmsg contains header and the message. use client.msgParts(rmsg) to split them apart
diff --git a/server/filestore.go b/server/filestore.go
index 6925b294..c62fc6e3 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -4911,11 +4911,6 @@ SKIP:
 		}
 		// Update blks slice.
 		fs.blks = copyMsgBlocks(fs.blks[deleted:])
-		if lb := len(fs.blks); lb == 0 {
-			fs.lmb = nil
-		} else {
-			fs.lmb = fs.blks[lb-1]
-		}
 	}
 
 	// Update top level accounting.
diff --git a/server/jetstream.go b/server/jetstream.go
index 4664bda1..c35a1ee4 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1887,11 +1887,7 @@ func (js *jetStream) limitsExceeded(storeType StorageType) bool {
 
 func tierName(cfg *StreamConfig) string {
 	// TODO (mh) this is where we could select based off a placement tag as well "qos:tier"
-	replicas := cfg.Replicas
-	if replicas == 0 {
-		replicas = 1
-	}
-	return fmt.Sprintf("R%d", replicas)
+	return fmt.Sprintf("R%d", cfg.Replicas)
 }
 
 func isSameTier(cfgA, cfgB *StreamConfig) bool {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 94e82be1..ff86678f 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -433,27 +433,6 @@ func (cc *jetStreamCluster) isConsumerCurrent(account, stream, consumer string)
 	return true
 }
 
-// subjectsOverlap checks all existing stream assignments for the account cross-cluster for subject overlap
-// Use only for clustered JetStream
-// Read lock should be held.
-func (jsc *jetStreamCluster) subjectsOverlap(acc string, subjects []string, osa *streamAssignment) bool {
-	asa := jsc.streams[acc]
-	for _, sa := range asa {
-		// can't overlap yourself, assume osa pre-checked for deep equal if passed
-		if osa != nil && sa == osa {
-			continue
-		}
-		for _, subj := range sa.Config.Subjects {
-			for _, tsubj := range subjects {
-				if SubjectsCollide(tsubj, subj) {
-					return true
-				}
-			}
-		}
-	}
-	return false
-}
-
 func (a *Account) getJetStreamFromAccount() (*Server, *jetStream, *jsAccount) {
 	a.mu.RLock()
 	jsa := a.js
@@ -5160,16 +5139,22 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 		return
 	}
 
-	var self *streamAssignment
-	if osa != nil && areEqual {
-		self = osa
-	}
-
 	// Check for subject collisions here.
-	if cc.subjectsOverlap(acc.Name, cfg.Subjects, self) {
-		resp.Error = NewJSStreamSubjectOverlapError()
-		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
-		return
+	asa := cc.streams[acc.Name]
+	for _, sa := range asa {
+		// If we found an osa and are here we are letting this through
+		if sa == osa && areEqual {
+			continue
+		}
+		for _, subj := range sa.Config.Subjects {
+			for _, tsubj := range cfg.Subjects {
+				if SubjectsCollide(tsubj, subj) {
+					resp.Error = NewJSStreamSubjectOverlapError()
+					s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+					return
+				}
+			}
+		}
 	}
 
 	apiErr = js.jsClusteredStreamLimitsCheck(acc, cfg)
@@ -5318,10 +5303,19 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	}
 
 	// Check for subject collisions here.
-	if cc.subjectsOverlap(acc.Name, cfg.Subjects, osa) {
-		resp.Error = NewJSStreamSubjectOverlapError()
-		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
-		return
+	for _, sa := range cc.streams[acc.Name] {
+		if sa == osa {
+			continue
+		}
+		for _, subj := range sa.Config.Subjects {
+			for _, tsubj := range newCfg.Subjects {
+				if SubjectsCollide(tsubj, subj) {
+					resp.Error = NewJSStreamSubjectOverlapError()
+					s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+					return
+				}
+			}
+		}
 	}
 
 	// Make copy so to not change original.
diff --git a/server/jwt.go b/server/jwt.go
index 4c97ed21..b9003265 100644
--- a/server/jwt.go
+++ b/server/jwt.go
@@ -14,6 +14,7 @@
 package server
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"os"
@@ -152,6 +153,12 @@ func validateTrustedOperators(o *Options) error {
 			o.resolverPinnedAccounts[o.SystemAccount] = struct{}{}
 		}
 	}
+
+	// If we have an auth callout defined make sure we are not in operator mode.
+	if o.AuthCallout != nil {
+		return errors.New("operators do not allow authorization callouts to be configured directly")
+	}
+
 	return nil
 }
 
diff --git a/server/monitor.go b/server/monitor.go
index 427b2581..5c5a5b7f 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -2631,8 +2631,6 @@ type JSzOptions struct {
 
 // HealthzOptions are options passed to Healthz
 type HealthzOptions struct {
-	// Deprecated: Use JSEnabledOnly instead
-	JSEnabled     bool `json:"js-enabled,omitempty"`
 	JSEnabledOnly bool `json:"js-enabled-only,omitempty"`
 	JSServerOnly  bool `json:"js-server-only,omitempty"`
 }
@@ -2987,8 +2985,7 @@ func (s *Server) HandleHealthz(w http.ResponseWriter, r *http.Request) {
 	}
 
 	hs := s.healthz(&HealthzOptions{
-		JSEnabled:     jsEnabled,
-		JSEnabledOnly: jsEnabledOnly,
+		JSEnabledOnly: jsEnabledOnly || jsEnabled,
 		JSServerOnly:  jsServerOnly,
 	})
 	if hs.Error != _EMPTY_ {
@@ -3034,7 +3031,7 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 		return health
 	}
 	// Only check if JS is enabled, skip meta and asset check.
-	if opts.JSEnabledOnly || opts.JSEnabled {
+	if opts.JSEnabledOnly {
 		return health
 	}
 
diff --git a/server/opts.go b/server/opts.go
index b69fc4ae..74c8da8c 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -202,6 +202,19 @@ type JSLimitOpts struct {
 	Duplicates      time.Duration
 }
 
+// And AuthCallout option used to map external AuthN to NATS based AuthZ.
+type AuthCallout struct {
+	// Must be a public account Nkey.
+	Issuer string
+	// Account to be used for sending requests.
+	Account string
+	// Users that will bypass auth_callout and be used for the auth service itself.
+	AuthUsers []string
+	// XKey is a public xkey for the authorization service.
+	// This will enable encryption for server requests and the authorization service responses.
+	XKey string
+}
+
 // Options block for nats-server.
 // NOTE: This structure is no longer used for monitoring endpoints
 // and json tags are deprecated and may be removed in the future.
@@ -233,6 +246,7 @@ type Options struct {
 	Username              string        `json:"-"`
 	Password              string        `json:"-"`
 	Authorization         string        `json:"-"`
+	AuthCallout           *AuthCallout  `json:"-"`
 	PingInterval          time.Duration `json:"ping_interval"`
 	MaxPingsOut           int           `json:"ping_max"`
 	HTTPHost              string        `json:"http_host"`
@@ -554,6 +568,8 @@ type authorization struct {
 	users              []*User
 	timeout            float64
 	defaultPermissions *Permissions
+	// Auth Callouts
+	callout *AuthCallout
 }
 
 // TLSConfigOpts holds the parsed tls config information,
@@ -810,6 +826,8 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 		o.Password = auth.pass
 		o.Authorization = auth.token
 		o.AuthTimeout = auth.timeout
+		o.AuthCallout = auth.callout
+
 		if (auth.user != _EMPTY_ || auth.pass != _EMPTY_) && auth.token != _EMPTY_ {
 			err := &configErr{tk, "Cannot have a user/pass and token"}
 			*errors = append(*errors, err)
@@ -1501,6 +1519,12 @@ func parseCluster(v interface{}, opts *Options, errors *[]error, warnings *[]err
 				*errors = append(*errors, err)
 				continue
 			}
+			if auth.callout != nil {
+				err := &configErr{tk, "Cluster authorization does not support callouts"}
+				*errors = append(*errors, err)
+				continue
+			}
+
 			opts.Cluster.Username = auth.user
 			opts.Cluster.Password = auth.pass
 			opts.Cluster.AuthTimeout = auth.timeout
@@ -1664,6 +1688,12 @@ func parseGateway(v interface{}, o *Options, errors *[]error, warnings *[]error)
 				*errors = append(*errors, err)
 				continue
 			}
+			if auth.callout != nil {
+				err := &configErr{tk, "Gateway authorization does not support callouts"}
+				*errors = append(*errors, err)
+				continue
+			}
+
 			o.Gateway.Username = auth.user
 			o.Gateway.Password = auth.pass
 			o.Gateway.AuthTimeout = auth.timeout
@@ -3453,6 +3483,13 @@ func parseAuthorization(v interface{}, opts *Options, errors *[]error, warnings
 				continue
 			}
 			auth.defaultPermissions = permissions
+		case "auth_callout", "auth_hook":
+			ac, err := parseAuthCallout(tk, errors, warnings)
+			if err != nil {
+				*errors = append(*errors, err)
+				continue
+			}
+			auth.callout = ac
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
@@ -3581,6 +3618,66 @@ func parseAllowedConnectionTypes(tk token, lt *token, mv interface{}, errors *[]
 	return m
 }
 
+// Helper function to parse auth callouts.
+func parseAuthCallout(mv interface{}, errors, warnings *[]error) (*AuthCallout, error) {
+	var (
+		tk token
+		lt token
+		ac = &AuthCallout{}
+	)
+	defer convertPanicToErrorList(&lt, errors)
+
+	tk, mv = unwrapValue(mv, &lt)
+	pm, ok := mv.(map[string]interface{})
+	if !ok {
+		return nil, &configErr{tk, fmt.Sprintf("Expected authorization callout to be a map/struct, got %+v", mv)}
+	}
+	for k, v := range pm {
+		tk, mv = unwrapValue(v, &lt)
+
+		switch strings.ToLower(k) {
+		case "issuer":
+			ac.Issuer = mv.(string)
+			if !nkeys.IsValidPublicAccountKey(ac.Issuer) {
+				return nil, &configErr{tk, fmt.Sprintf("Expected callout user to be a valid public account nkey, got %q", ac.Issuer)}
+			}
+		case "account", "acc":
+			ac.Account = mv.(string)
+		case "auth_users", "users":
+			aua, ok := mv.([]interface{})
+			if !ok {
+				return nil, &configErr{tk, fmt.Sprintf("Expected auth_users field to be an array, got %T", v)}
+			}
+			for _, uv := range aua {
+				_, uv = unwrapValue(uv, &lt)
+				ac.AuthUsers = append(ac.AuthUsers, uv.(string))
+			}
+		case "xkey", "key":
+			ac.XKey = mv.(string)
+			if !nkeys.IsValidPublicCurveKey(ac.XKey) {
+				return nil, &configErr{tk, fmt.Sprintf("Expected callout xkey to be a valid public xkey, got %q", ac.XKey)}
+			}
+		default:
+			if !tk.IsUsedVariable() {
+				err := &configErr{tk, fmt.Sprintf("Unknown field %q parsing authorization callout", k)}
+				*errors = append(*errors, err)
+			}
+		}
+	}
+	// Make sure we have all defined. All fields are required.
+	// If no account specified, selet $G.
+	if ac.Account == _EMPTY_ {
+		ac.Account = globalAccountName
+	}
+	if ac.Issuer == _EMPTY_ {
+		return nil, &configErr{tk, "Authorization callouts require an issuer to be specified"}
+	}
+	if len(ac.AuthUsers) == 0 {
+		return nil, &configErr{tk, "Authorization callouts require authorized users to be specified"}
+	}
+	return ac, nil
+}
+
 // Helper function to parse user/account permissions
 func parseUserPermissions(mv interface{}, errors, warnings *[]error) (*Permissions, error) {
 	var (
diff --git a/server/raft.go b/server/raft.go
index 7f0ad3fb..1bcfd745 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -951,6 +951,15 @@ func (n *raft) InstallSnapshot(data []byte) error {
 		return errCatchupsRunning
 	}
 
+	// We don't store snapshots for memory based WALs.
+	// This matches loadSnapshot logic now.
+	// But we should compact.
+	if _, ok := n.wal.(*memStore); ok {
+		n.wal.Compact(n.applied)
+		n.Unlock()
+		return nil
+	}
+
 	var state StreamState
 	n.wal.FastState(&state)
 
diff --git a/server/reload.go b/server/reload.go
index 7058417f..b25b43d6 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -949,6 +949,7 @@ func imposeOrder(value interface{}) error {
 		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList,
 		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher:
 		// explicitly skipped types
+	case *AuthCallout:
 	default:
 		// this will fail during unit tests
 		return fmt.Errorf("OnReload, sort or explicitly skip type: %s",
diff --git a/server/server.go b/server/server.go
index d39604ce..676de7ba 100644
--- a/server/server.go
+++ b/server/server.go
@@ -103,6 +103,8 @@ type Info struct {
 	// LeafNode Specific
 	LeafNodeURLs  []string `json:"leafnode_urls,omitempty"`  // LeafNode URLs that the server can reconnect to.
 	RemoteAccount string   `json:"remote_account,omitempty"` // Lets the other side know the remote account that they bind to.
+
+	XKey string `json:"xkey,omitempty"` // Public server's x25519 key.
 }
 
 // Server is our main struct.
@@ -114,6 +116,8 @@ type Server struct {
 	stats
 	mu                  sync.RWMutex
 	kp                  nkeys.KeyPair
+	xkp                 nkeys.KeyPair
+	xpub                string
 	info                Info
 	configFile          string
 	optsMu              sync.RWMutex
@@ -332,10 +336,14 @@ func NewServer(opts *Options) (*Server, error) {
 	tlsReq := opts.TLSConfig != nil
 	verify := (tlsReq && opts.TLSConfig.ClientAuth == tls.RequireAndVerifyClientCert)
 
-	// Created server's nkey identity.
+	// Create our server's nkey identity.
 	kp, _ := nkeys.CreateServer()
 	pub, _ := kp.PublicKey()
 
+	// Create an xkey for encrypting messages from this server.
+	xkp, _ := nkeys.CreateCurveKeys()
+	xpub, _ := xkp.PublicKey()
+
 	serverName := pub
 	if opts.ServerName != _EMPTY_ {
 		serverName = opts.ServerName
@@ -353,6 +361,7 @@ func NewServer(opts *Options) (*Server, error) {
 
 	info := Info{
 		ID:           pub,
+		XKey:         xpub,
 		Version:      VERSION,
 		Proto:        PROTO,
 		GitCommit:    gitCommit,
@@ -378,6 +387,8 @@ func NewServer(opts *Options) (*Server, error) {
 
 	s := &Server{
 		kp:                 kp,
+		xkp:                xkp,
+		xpub:               xpub,
 		configFile:         opts.ConfigFile,
 		info:               info,
 		opts:               opts,
diff --git a/server/stream.go b/server/stream.go
index 2431e0ba..669fd1fb 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -413,7 +413,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 
 	// Check for overlapping subjects with other streams.
 	// These are not allowed for now.
-	if jsa.subjectsOverlap(cfg.Subjects, nil) {
+	if jsa.subjectsOverlap(cfg.Subjects) {
 		jsa.mu.Unlock()
 		return nil, NewJSStreamSubjectOverlapError()
 	}
@@ -910,14 +910,10 @@ func (mset *stream) setCreatedTime(created time.Time) {
 	mset.mu.Unlock()
 }
 
-// subjectsOverlap to see if these subjects overlap with existing subjects.
-// Use only for non-clustered JetStream
-// RLock minimum should be held.
-func (jsa *jsAccount) subjectsOverlap(subjects []string, self *stream) bool {
+// Check to see if these subjects overlap with existing subjects.
+// Lock should be held.
+func (jsa *jsAccount) subjectsOverlap(subjects []string) bool {
 	for _, mset := range jsa.streams {
-		if self != nil && mset == self {
-			continue
-		}
 		for _, subj := range mset.cfg.Subjects {
 			for _, tsubj := range subjects {
 				if SubjectsCollide(tsubj, subj) {
@@ -1205,8 +1201,7 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 			return StreamConfig{}, NewJSMirrorWithSubjectsError()
 		}
 
-		// Check for literal duplication of subject interest in config
-		// and no overlap with any JS API subject space
+		// We can allow overlaps, but don't allow direct duplicates.
 		dset := make(map[string]struct{}, len(cfg.Subjects))
 		for _, subj := range cfg.Subjects {
 			if _, ok := dset[subj]; ok {
@@ -1439,11 +1434,6 @@ func (mset *stream) update(config *StreamConfig) error {
 
 // Update will allow certain configuration properties of an existing stream to be updated.
 func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool) error {
-	_, jsa, err := mset.acc.checkForJetStream()
-	if err != nil {
-		return err
-	}
-
 	mset.mu.RLock()
 	ocfg := mset.cfg
 	s := mset.srv
@@ -1454,13 +1444,6 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 		return NewJSStreamInvalidConfigError(err, Unless(err))
 	}
 
-	jsa.mu.RLock()
-	if jsa.subjectsOverlap(cfg.Subjects, mset) {
-		jsa.mu.RUnlock()
-		return NewJSStreamSubjectOverlapError()
-	}
-	jsa.mu.RUnlock()
-
 	mset.mu.Lock()
 	if mset.isLeader() {
 		// Now check for subject interest differences.


Test patch:
diff --git a/server/auth_callout_test.go b/server/auth_callout_test.go
new file mode 100644
index 00000000..1e3b476b
--- /dev/null
+++ b/server/auth_callout_test.go
@@ -0,0 +1,1000 @@
+// Copyright 2022-2023 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"bytes"
+	"crypto/x509"
+	"encoding/json"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
+)
+
+// Helper function to decode an auth request.
+func decodeAuthRequest(t *testing.T, ejwt []byte) (string, *jwt.ServerID, *jwt.ClientInformation, *jwt.ConnectOptions, *jwt.ClientTLS) {
+	t.Helper()
+	ac, err := jwt.DecodeAuthorizationRequestClaims(string(ejwt))
+	require_NoError(t, err)
+	return ac.UserNkey, &ac.Server, &ac.ClientInformation, &ac.ConnectOptions, ac.TLS
+}
+
+const (
+	authCalloutPub        = "UBO2MQV67TQTVIRV3XFTEZOACM4WLOCMCDMAWN5QVN5PI2N6JHTVDRON"
+	authCalloutSeed       = "SUAP277QP7U4JMFFPVZHLJYEQJ2UHOTYVEIZJYAWRJXQLP4FRSEHYZJJOU"
+	authCalloutIssuer     = "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+	authCalloutIssuerSeed = "SAANDLKMXL6CUS3CP52WIXBEDN6YJ545GDKC65U5JZPPV6WH6ESWUA6YAI"
+)
+
+// Will create a signed user jwt as an authorized user.
+func createAuthUser(t *testing.T, server, user, name, account string, akp nkeys.KeyPair, expires time.Duration, limits *jwt.UserPermissionLimits) string {
+	t.Helper()
+
+	if akp == nil {
+		var err error
+		akp, err = nkeys.FromSeed([]byte(authCalloutIssuerSeed))
+		require_NoError(t, err)
+	}
+
+	uclaim := jwt.NewUserClaims(user)
+	uclaim.Audience = account
+	if name != _EMPTY_ {
+		uclaim.Name = name
+	}
+	if expires > 0 {
+		uclaim.Expires = time.Now().Add(expires).Unix()
+	}
+	if limits != nil {
+		uclaim.UserPermissionLimits = *limits
+	}
+
+	vr := jwt.CreateValidationResults()
+	uclaim.Validate(vr)
+	require_Len(t, len(vr.Errors()), 0)
+
+	arc := jwt.NewAuthorizationResponseClaims(user)
+	arc.User = uclaim
+	arc.Audience = server
+
+	vr = jwt.CreateValidationResults()
+	arc.Validate(vr)
+	require_Len(t, len(vr.Errors()), 0)
+
+	rjwt, err := arc.Encode(akp)
+	require_NoError(t, err)
+
+	return rjwt
+}
+
+func TestAuthCalloutBasics(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: A
+		authorization {
+			timeout: 1s
+			users: [ { user: "auth", password: "pwd" } ]
+			auth_callout {
+				# Needs to be a public account nkey, will work for both server config and operator mode.
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				# users that will power the auth callout service.
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	callouts := uint32(0)
+
+	// This will not use callout since predefined as an auth_user.
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	// Make sure callout was not triggered.
+	require_True(t, atomic.LoadUint32(&callouts) == 0)
+
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		atomic.AddUint32(&callouts, 1)
+		user, si, ci, opts, _ := decodeAuthRequest(t, m.Data)
+		require_True(t, si.Name == "A")
+		require_True(t, ci.Host == "127.0.0.1")
+		// Allow dlc user.
+		if opts.Username == "dlc" && opts.Password == "zzz" {
+			var j jwt.UserPermissionLimits
+			j.Pub.Allow.Add("$SYS.>")
+			j.Payload = 1024
+			ujwt := createAuthUser(t, si.ID, user, _EMPTY_, globalAccountName, nil, 10*time.Minute, &j)
+			m.Respond([]byte(ujwt))
+		} else {
+			// Nil response signals no authentication.
+			m.Respond(nil)
+		}
+	})
+	require_NoError(t, err)
+
+	// This one should fail since bad password.
+	_, err = nats.Connect(s.ClientURL(), nats.UserInfo("dlc", "xxx"))
+	require_Error(t, err)
+
+	// This one will use callout since not defined in server config.
+	nc, err = nats.Connect(s.ClientURL(), nats.UserInfo("dlc", "zzz"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	resp, err := nc.Request(userDirectInfoSubj, nil, time.Second)
+	require_NoError(t, err)
+	response := ServerAPIResponse{Data: &UserInfo{}}
+	err = json.Unmarshal(resp.Data, &response)
+	require_NoError(t, err)
+
+	userInfo := response.Data.(*UserInfo)
+
+	dlc := &UserInfo{
+		UserID:  "dlc",
+		Account: globalAccountName,
+		Permissions: &Permissions{
+			Publish: &SubjectPermission{
+				Allow: []string{"$SYS.>"},
+				Deny:  []string{AuthCalloutSubject}, // Will be auto-added since in auth account.
+			},
+			Subscribe: &SubjectPermission{},
+		},
+	}
+	expires := userInfo.Expires
+	userInfo.Expires = 0
+	if !reflect.DeepEqual(dlc, userInfo) {
+		t.Fatalf("User info for %q did not match", "dlc")
+	}
+	if expires > 10*time.Minute || expires < (10*time.Minute-5*time.Second) {
+		t.Fatalf("Expected expires of ~%v, got %v", 10*time.Minute, expires)
+	}
+}
+
+func TestAuthCalloutMultiAccounts(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: ZZ
+		accounts {
+            AUTH { users [ {user: "auth", password: "pwd"} ] }
+			FOO {}
+			BAR {}
+			BAZ {}
+		}
+		authorization {
+			timeout: 1s
+			auth_callout {
+				# Needs to be a public account nkey, will work for both server config and operator mode.
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				account: AUTH
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// Auth callout user.
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+	// Should always make auth callouts queue subscribers or proper services.
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		user, si, ci, opts, _ := decodeAuthRequest(t, m.Data)
+		require_True(t, si.Name == "ZZ")
+		require_True(t, ci.Host == "127.0.0.1")
+		// Allow dlc user and map to the BAZ account.
+		if opts.Username == "dlc" && opts.Password == "zzz" {
+			ujwt := createAuthUser(t, si.ID, user, _EMPTY_, "BAZ", nil, 0, nil)
+			m.Respond([]byte(ujwt))
+		} else {
+			// Nil response signals no authentication.
+			m.Respond(nil)
+		}
+	})
+	require_NoError(t, err)
+
+	// This one will use callout since not defined in server config.
+	nc, err = nats.Connect(s.ClientURL(), nats.UserInfo("dlc", "zzz"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	resp, err := nc.Request(userDirectInfoSubj, nil, time.Second)
+	require_NoError(t, err)
+	response := ServerAPIResponse{Data: &UserInfo{}}
+	err = json.Unmarshal(resp.Data, &response)
+	require_NoError(t, err)
+	userInfo := response.Data.(*UserInfo)
+
+	require_True(t, userInfo.UserID == "dlc")
+	require_True(t, userInfo.Account == "BAZ")
+}
+
+func TestAuthCalloutClientTLSCerts(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "localhost:-1"
+		server_name: T
+
+		tls {
+			cert_file = "../test/configs/certs/tlsauth/server.pem"
+			key_file = "../test/configs/certs/tlsauth/server-key.pem"
+			ca_file = "../test/configs/certs/tlsauth/ca.pem"
+			verify = true
+		}
+
+		accounts {
+            AUTH { users [ {user: "auth", password: "pwd"} ] }
+			FOO {}
+		}
+		authorization {
+			timeout: 1s
+			auth_callout {
+				# Needs to be a public account nkey, will work for both server config and operator mode.
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				account: AUTH
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL(),
+		nats.UserInfo("auth", "pwd"),
+		nats.ClientCert("../test/configs/certs/tlsauth/client2.pem", "../test/configs/certs/tlsauth/client2-key.pem"),
+		nats.RootCAs("../test/configs/certs/tlsauth/ca.pem"),
+	)
+	require_NoError(t, err)
+	defer nc.Close()
+
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		user, si, ci, _, ctls := decodeAuthRequest(t, m.Data)
+		require_True(t, si.Name == "T")
+		require_True(t, ci.Host == "127.0.0.1")
+		require_True(t, ctls != nil)
+		// Zero since we are verified and will be under verified chains.
+		require_True(t, len(ctls.Certs) == 0)
+		require_True(t, len(ctls.VerifiedChains) == 1)
+		// Since we have a CA.
+		require_True(t, len(ctls.VerifiedChains[0]) == 2)
+		blk, _ := pem.Decode([]byte(ctls.VerifiedChains[0][0]))
+		cert, err := x509.ParseCertificate(blk.Bytes)
+		require_NoError(t, err)
+		if strings.HasPrefix(cert.Subject.String(), "CN=example.com") {
+			// Override blank name here, server will substitute.
+			ujwt := createAuthUser(t, si.ID, user, "dlc", "FOO", nil, 0, nil)
+			m.Respond([]byte(ujwt))
+		}
+	})
+	require_NoError(t, err)
+
+	// Will use client cert to determine user.
+	nc, err = nats.Connect(s.ClientURL(),
+		nats.ClientCert("../test/configs/certs/tlsauth/client2.pem", "../test/configs/certs/tlsauth/client2-key.pem"),
+		nats.RootCAs("../test/configs/certs/tlsauth/ca.pem"),
+	)
+	require_NoError(t, err)
+	defer nc.Close()
+
+	resp, err := nc.Request(userDirectInfoSubj, nil, time.Second)
+	require_NoError(t, err)
+	response := ServerAPIResponse{Data: &UserInfo{}}
+	err = json.Unmarshal(resp.Data, &response)
+	require_NoError(t, err)
+	userInfo := response.Data.(*UserInfo)
+
+	require_True(t, userInfo.UserID == "dlc")
+	require_True(t, userInfo.Account == "FOO")
+}
+
+func TestAuthCalloutVerifiedUserCalloutsWithSig(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: A
+		authorization {
+			timeout: 1s
+			users: [
+				{ user: "auth", password: "pwd" }
+ 				{ nkey: "UBO2MQV67TQTVIRV3XFTEZOACM4WLOCMCDMAWN5QVN5PI2N6JHTVDRON" }
+ 			]
+			auth_callout {
+				# Needs to be a public account nkey, will work for both server config and operator mode.
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				# users that will power the auth callout service.
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// This will not use callout since predefined as an auth_user.
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	callouts := uint32(0)
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		atomic.AddUint32(&callouts, 1)
+		user, si, ci, opts, _ := decodeAuthRequest(t, m.Data)
+		require_True(t, si.Name == "A")
+		require_True(t, ci.Host == "127.0.0.1")
+		require_True(t, opts.SignedNonce != _EMPTY_)
+		require_True(t, ci.Nonce != _EMPTY_)
+		ujwt := createAuthUser(t, si.ID, user, _EMPTY_, globalAccountName, nil, 0, nil)
+		m.Respond([]byte(ujwt))
+	})
+	require_NoError(t, err)
+
+	// This one will use callout since not part of auth_users.
+	// Even though we will internally verify this user the callout will still be called.
+	seedFile := createTempFile(t, _EMPTY_)
+	defer removeFile(t, seedFile.Name())
+	seedFile.WriteString(authCalloutSeed)
+	nkeyOpt, err := nats.NkeyOptionFromSeed(seedFile.Name())
+	require_NoError(t, err)
+
+	nc, err = nats.Connect(s.ClientURL(), nkeyOpt)
+	require_NoError(t, err)
+	defer nc.Close()
+
+	// Make sure that the callout was called.
+	if atomic.LoadUint32(&callouts) != 1 {
+		t.Fatalf("Expected callout to be called")
+	}
+
+	resp, err := nc.Request(userDirectInfoSubj, nil, time.Second)
+	require_NoError(t, err)
+	response := ServerAPIResponse{Data: &UserInfo{}}
+	err = json.Unmarshal(resp.Data, &response)
+	require_NoError(t, err)
+
+	userInfo := response.Data.(*UserInfo)
+
+	dlc := &UserInfo{
+		UserID:  "UBO2MQV67TQTVIRV3XFTEZOACM4WLOCMCDMAWN5QVN5PI2N6JHTVDRON",
+		Account: globalAccountName,
+		Permissions: &Permissions{
+			Publish: &SubjectPermission{
+				Deny: []string{AuthCalloutSubject}, // Will be auto-added since in auth account.
+			},
+			Subscribe: &SubjectPermission{},
+		},
+	}
+	if !reflect.DeepEqual(dlc, userInfo) {
+		t.Fatalf("User info for %q did not match", "dlc")
+	}
+}
+
+// For creating the authorized users in operator mode.
+func createAuthServiceUser(t *testing.T, accKp nkeys.KeyPair) (pub, creds string) {
+	t.Helper()
+	ukp, _ := nkeys.CreateUser()
+	seed, _ := ukp.Seed()
+	upub, _ := ukp.PublicKey()
+	uclaim := newJWTTestUserClaims()
+	uclaim.Subject = upub
+	vr := jwt.ValidationResults{}
+	uclaim.Validate(&vr)
+	require_Len(t, len(vr.Errors()), 0)
+	ujwt, err := uclaim.Encode(accKp)
+	require_NoError(t, err)
+	return upub, genCredsFile(t, ujwt, seed)
+}
+
+func createBasicAccountUser(t *testing.T, accKp nkeys.KeyPair) (creds string) {
+	t.Helper()
+	ukp, _ := nkeys.CreateUser()
+	seed, _ := ukp.Seed()
+	upub, _ := ukp.PublicKey()
+	uclaim := newJWTTestUserClaims()
+	uclaim.Subject = upub
+	// For these deny all permission
+	uclaim.Permissions.Pub.Deny.Add(">")
+	uclaim.Permissions.Sub.Deny.Add(">")
+	vr := jwt.ValidationResults{}
+	uclaim.Validate(&vr)
+	require_Len(t, len(vr.Errors()), 0)
+	ujwt, err := uclaim.Encode(accKp)
+	require_NoError(t, err)
+	return genCredsFile(t, ujwt, seed)
+}
+
+func TestAuthCalloutOperatorNoServerConfigCalloutAllowed(t *testing.T) {
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		operator: %s
+		resolver: MEM
+		authorization {
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				auth_users: [ auth ]
+			}
+		}
+    `, ojwt)))
+	defer removeFile(t, conf)
+
+	opts := LoadConfig(conf)
+	_, err := NewServer(opts)
+	require_Error(t, err, errors.New("operators do not allow authorization callouts to be configured directly"))
+}
+
+func TestAuthCalloutOperatorModeBasics(t *testing.T) {
+	_, spub := createKey(t)
+	sysClaim := jwt.NewAccountClaims(spub)
+	sysClaim.Name = "$SYS"
+	sysJwt, err := sysClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	// TEST account.
+	tkp, tpub := createKey(t)
+	accClaim := jwt.NewAccountClaims(tpub)
+	accClaim.Name = "TEST"
+	accJwt, err := accClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	// AUTH service account.
+	akp, err := nkeys.FromSeed([]byte(authCalloutIssuerSeed))
+	require_NoError(t, err)
+
+	apub, err := akp.PublicKey()
+	require_NoError(t, err)
+
+	// The authorized user for the service.
+	upub, creds := createAuthServiceUser(t, akp)
+	defer removeFile(t, creds)
+
+	authClaim := jwt.NewAccountClaims(apub)
+	authClaim.Name = "AUTH"
+	authClaim.EnableExternalAuthorization(upub)
+	authClaim.Authorization.AllowedAccounts.Add(tpub)
+	authJwt, err := authClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		operator: %s
+		system_account: %s
+		resolver: MEM
+		resolver_preload: {
+			%s: %s
+			%s: %s
+			%s: %s
+		}
+    `, ojwt, spub, apub, authJwt, tpub, accJwt, spub, sysJwt)))
+	defer removeFile(t, conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL(), nats.UserCredentials(creds))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	// Check that we have the deny permission autoset properly.
+	resp, err := nc.Request(userDirectInfoSubj, nil, time.Second)
+	require_NoError(t, err)
+	response := ServerAPIResponse{Data: &UserInfo{}}
+	err = json.Unmarshal(resp.Data, &response)
+	require_NoError(t, err)
+
+	userInfo := response.Data.(*UserInfo)
+	expected := &UserInfo{
+		UserID:  upub,
+		Account: apub,
+		Permissions: &Permissions{
+			Publish: &SubjectPermission{
+				Deny: []string{AuthCalloutSubject}, // Will be auto-added since in auth account.
+			},
+			Subscribe: &SubjectPermission{},
+		},
+	}
+	if !reflect.DeepEqual(expected, userInfo) {
+		t.Fatalf("User info did not match expected, expected auto-deny permissions on callout subject")
+	}
+
+	const secretToken = "--XX--"
+	const dummyToken = "--ZZ--"
+	dkp, notAllowAccountPub := createKey(t)
+
+	// Register authorization handlers.
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		user, si, _, opts, _ := decodeAuthRequest(t, m.Data)
+		if opts.Token == secretToken {
+			ujwt := createAuthUser(t, si.ID, user, "dlc", tpub, tkp, 0, nil)
+			m.Respond([]byte(ujwt))
+		} else if opts.Token == dummyToken {
+			ujwt := createAuthUser(t, si.ID, user, "dummy", notAllowAccountPub, dkp, 0, nil)
+			m.Respond([]byte(ujwt))
+		} else {
+			m.Respond(nil)
+		}
+	})
+	require_NoError(t, err)
+
+	// Bearer token etc..
+	// This is used by all users, and the customization will be in other connect args.
+	// This needs to also be bound to the authorization account.
+	creds = createBasicAccountUser(t, akp)
+	defer removeFile(t, creds)
+
+	// We require a token.
+	_, err = nats.Connect(s.ClientURL(), nats.UserCredentials(creds))
+	require_Error(t, err)
+
+	// Send correct token. This should switch us to the test account.
+	nc, err = nats.Connect(s.ClientURL(), nats.UserCredentials(creds), nats.Token(secretToken))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	resp, err = nc.Request(userDirectInfoSubj, nil, time.Second)
+	require_NoError(t, err)
+	response = ServerAPIResponse{Data: &UserInfo{}}
+	err = json.Unmarshal(resp.Data, &response)
+	require_NoError(t, err)
+
+	userInfo = response.Data.(*UserInfo)
+
+	// Make sure we switch accounts.
+	if userInfo.Account != tpub {
+		t.Fatalf("Expected to be switched to %q, but got %q", tpub, userInfo.Account)
+	}
+
+	// Now make sure that if the authorization service switches to an account that is not allowed, we reject.
+	_, err = nats.Connect(s.ClientURL(), nats.UserCredentials(creds), nats.Token(dummyToken))
+	require_Error(t, err)
+}
+
+const (
+	curveSeed   = "SXAAXMRAEP6JWWHNB6IKFL554IE6LZVT6EY5MBRICPILTLOPHAG73I3YX4"
+	curvePublic = "XAB3NANV3M6N7AHSQP2U5FRWKKUT7EG2ZXXABV4XVXYQRJGM4S2CZGHT"
+)
+
+func TestAuthCalloutServerConfigEncryption(t *testing.T) {
+	tmpl := `
+		listen: "127.0.0.1:-1"
+		server_name: A
+		authorization {
+			timeout: 1s
+			users: [ { user: "auth", password: "pwd" } ]
+			auth_callout {
+				# Needs to be a public account nkey, will work for both server config and operator mode.
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				# users that will power the auth callout service.
+				auth_users: [ auth ]
+				# This is a public xkey (x25519). The auth service has the private key.
+				xkey: "%s"
+			}
+		}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, curvePublic)))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// This will not use callout since predefined as an auth_user.
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	rkp, err := nkeys.FromCurveSeed([]byte(curveSeed))
+	require_NoError(t, err)
+
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		// This will be encrypted.
+		_, err := jwt.DecodeAuthorizationRequestClaims(string(m.Data))
+		require_Error(t, err)
+
+		xkey := m.Header.Get(AuthRequestXKeyHeader)
+		require_True(t, xkey != _EMPTY_)
+		decrypted, err := rkp.Open(m.Data, xkey)
+		require_NoError(t, err)
+		user, si, ci, opts, _ := decodeAuthRequest(t, decrypted)
+		// The header xkey must match the signed xkey in server info.
+		require_True(t, si.XKey == xkey)
+		require_True(t, si.Name == "A")
+		require_True(t, ci.Host == "127.0.0.1")
+		// Allow dlc user.
+		if opts.Username == "dlc" && opts.Password == "zzz" {
+			ujwt := createAuthUser(t, si.ID, user, _EMPTY_, globalAccountName, nil, 10*time.Minute, nil)
+			m.Respond([]byte(ujwt))
+		} else if opts.Username == "dlc" && opts.Password == "xxx" {
+			ujwt := createAuthUser(t, si.ID, user, _EMPTY_, globalAccountName, nil, 10*time.Minute, nil)
+			// Encrypt this response.
+			encryptedResponse, err := rkp.Seal([]byte(ujwt), si.XKey) // Server's public xkey.
+			require_NoError(t, err)
+			m.Respond(encryptedResponse)
+		} else {
+			// Nil response signals no authentication.
+			m.Respond(nil)
+		}
+	})
+	require_NoError(t, err)
+
+	nc, err = nats.Connect(s.ClientURL(), nats.UserInfo("dlc", "zzz"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	// Authorization services can optionally encrypt the reponses using the server's public xkey.
+	nc, err = nats.Connect(s.ClientURL(), nats.UserInfo("dlc", "xxx"))
+	require_NoError(t, err)
+	defer nc.Close()
+}
+
+func TestAuthCalloutOperatorModeEncryption(t *testing.T) {
+	_, spub := createKey(t)
+	sysClaim := jwt.NewAccountClaims(spub)
+	sysClaim.Name = "$SYS"
+	sysJwt, err := sysClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	// TEST account.
+	tkp, tpub := createKey(t)
+	accClaim := jwt.NewAccountClaims(tpub)
+	accClaim.Name = "TEST"
+	accJwt, err := accClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	// AUTH service account.
+	akp, err := nkeys.FromSeed([]byte(authCalloutIssuerSeed))
+	require_NoError(t, err)
+
+	apub, err := akp.PublicKey()
+	require_NoError(t, err)
+
+	// The authorized user for the service.
+	upub, creds := createAuthServiceUser(t, akp)
+	defer removeFile(t, creds)
+
+	authClaim := jwt.NewAccountClaims(apub)
+	authClaim.Name = "AUTH"
+	authClaim.EnableExternalAuthorization(upub)
+	authClaim.Authorization.AllowedAccounts.Add(tpub)
+	authClaim.Authorization.XKey = curvePublic
+
+	authJwt, err := authClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		operator: %s
+		system_account: %s
+		resolver: MEM
+		resolver_preload: {
+			%s: %s
+			%s: %s
+			%s: %s
+		}
+    `, ojwt, spub, apub, authJwt, tpub, accJwt, spub, sysJwt)))
+	defer removeFile(t, conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL(), nats.UserCredentials(creds))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	const tokenA = "--XX--"
+	const tokenB = "--ZZ--"
+
+	rkp, err := nkeys.FromCurveSeed([]byte(curveSeed))
+	require_NoError(t, err)
+
+	// Register authorization handlers.
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		// Make sure this is an encrypted request.
+		if bytes.HasPrefix(m.Data, []byte(jwtPrefix)) {
+			t.Fatalf("Request not encrypted")
+		}
+		xkey := m.Header.Get(AuthRequestXKeyHeader)
+		require_True(t, xkey != _EMPTY_)
+		decrypted, err := rkp.Open(m.Data, xkey)
+		require_NoError(t, err)
+		user, si, ci, opts, _ := decodeAuthRequest(t, decrypted)
+		// The header xkey must match the signed xkey in server info.
+		require_True(t, si.XKey == xkey)
+		require_True(t, ci.Host == "127.0.0.1")
+		if opts.Token == tokenA {
+			ujwt := createAuthUser(t, si.ID, user, "dlc", tpub, tkp, 0, nil)
+			m.Respond([]byte(ujwt))
+		} else if opts.Token == tokenB {
+			ujwt := createAuthUser(t, si.ID, user, "rip", tpub, tkp, 0, nil)
+			// Encrypt this response.
+			encryptedResponse, err := rkp.Seal([]byte(ujwt), si.XKey) // Server's public xkey.
+			require_NoError(t, err)
+			m.Respond(encryptedResponse)
+		} else {
+			m.Respond(nil)
+		}
+	})
+	require_NoError(t, err)
+
+	// Bearer token etc..
+	// This is used by all users, and the customization will be in other connect args.
+	// This needs to also be bound to the authorization account.
+	creds = createBasicAccountUser(t, akp)
+	defer removeFile(t, creds)
+
+	// This will receive an encrypted request to the auth service but send plaintext response.
+	nc, err = nats.Connect(s.ClientURL(), nats.UserCredentials(creds), nats.Token(tokenA))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	// This will receive an encrypted request to the auth service and send an encrypted response.
+	nc, err = nats.Connect(s.ClientURL(), nats.UserCredentials(creds), nats.Token(tokenB))
+	require_NoError(t, err)
+	defer nc.Close()
+}
+
+func TestAuthCalloutServerTags(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: A
+		server_tags: ["foo", "bar"]
+		authorization {
+			users: [ { user: "auth", password: "pwd" } ]
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// This will not use callout since predefined as an auth_user.
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	tch := make(chan jwt.TagList, 1)
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		user, si, _, _, _ := decodeAuthRequest(t, m.Data)
+		tch <- si.Tags
+		ujwt := createAuthUser(t, si.ID, user, _EMPTY_, globalAccountName, nil, 10*time.Minute, nil)
+		m.Respond([]byte(ujwt))
+	})
+	require_NoError(t, err)
+
+	_, err = nats.Connect(s.ClientURL())
+	require_NoError(t, err)
+
+	tags := <-tch
+	require_True(t, len(tags) == 2)
+	require_True(t, tags.Contains("foo"))
+	require_True(t, tags.Contains("bar"))
+}
+
+func TestAuthCalloutServerClusterAndVersion(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: A
+		authorization {
+			users: [ { user: "auth", password: "pwd" } ]
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				auth_users: [ auth ]
+			}
+		}
+		cluster { name: HUB }
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// This will not use callout since predefined as an auth_user.
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	ch := make(chan string, 2)
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		user, si, _, _, _ := decodeAuthRequest(t, m.Data)
+		ch <- si.Cluster
+		ch <- si.Version
+		ujwt := createAuthUser(t, si.ID, user, _EMPTY_, globalAccountName, nil, 10*time.Minute, nil)
+		m.Respond([]byte(ujwt))
+	})
+	require_NoError(t, err)
+
+	_, err = nats.Connect(s.ClientURL())
+	require_NoError(t, err)
+
+	cluster := <-ch
+	require_True(t, cluster == "HUB")
+
+	version := <-ch
+	require_True(t, len(version) > 0)
+	ok, err := versionAtLeastCheckError(version, 2, 10, 0)
+	require_NoError(t, err)
+	require_True(t, ok)
+}
+
+func createErrResponse(t *testing.T, user, server, errStr string, akp nkeys.KeyPair) string {
+	t.Helper()
+
+	if akp == nil {
+		var err error
+		akp, err = nkeys.FromSeed([]byte(authCalloutIssuerSeed))
+		require_NoError(t, err)
+	}
+
+	arc := jwt.NewAuthorizationResponseClaims(user)
+	arc.SetErrorDescription(errStr)
+	arc.Audience = server
+	vr := jwt.CreateValidationResults()
+	arc.Validate(vr)
+	require_Len(t, len(vr.Errors()), 0)
+
+	rjwt, err := arc.Encode(akp)
+	require_NoError(t, err)
+
+	return rjwt
+}
+
+func TestAuthCalloutErrorResponse(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: A
+		authorization {
+			users: [ { user: "auth", password: "pwd" } ]
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// This will not use callout since predefined as an auth_user.
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		user, si, _, _, _ := decodeAuthRequest(t, m.Data)
+		errResp := createErrResponse(t, user, si.ID, "BAD AUTH", nil)
+		m.Respond([]byte(errResp))
+	})
+	require_NoError(t, err)
+
+	_, err = nats.Connect(s.ClientURL(), nats.UserInfo("dlc", "zzz"))
+	require_Error(t, err)
+}
+
+func TestAuthCalloutAuthUserFailDoesNotInvokeCallout(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: A
+		authorization {
+			users: [ { user: "auth", password: "pwd" } ]
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	callouts := uint32(0)
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		atomic.AddUint32(&callouts, 1)
+		user, si, _, _, _ := decodeAuthRequest(t, m.Data)
+		errResp := createErrResponse(t, user, si.ID, "BAD AUTH", nil)
+		m.Respond([]byte(errResp))
+	})
+	require_NoError(t, err)
+
+	_, err = nats.Connect(s.ClientURL(), nats.UserInfo("auth", "zzz"))
+	require_Error(t, err)
+
+	if atomic.LoadUint32(&callouts) != 0 {
+		t.Fatalf("Expected callout to not be called")
+	}
+}
+
+func TestAuthCalloutAuthErrEvents(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: "127.0.0.1:-1"
+		server_name: A
+		accounts {
+            AUTH { users [ {user: "auth", password: "pwd"} ] }
+			FOO {}
+			BAR {}
+		}
+		authorization {
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				account: AUTH
+				auth_users: [ auth ]
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("auth", "pwd"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	// This is where the event fires, in this account.
+	sub, err := nc.SubscribeSync(authErrorAccountEventSubj)
+	require_NoError(t, err)
+
+	_, err = nc.Subscribe(AuthCalloutSubject, func(m *nats.Msg) {
+		user, si, _, opts, _ := decodeAuthRequest(t, m.Data)
+		// Allow dlc user and map to the BAZ account.
+		if opts.Username == "dlc" && opts.Password == "zzz" {
+			ujwt := createAuthUser(t, si.ID, user, _EMPTY_, "FOO", nil, 0, nil)
+			m.Respond([]byte(ujwt))
+		} else if opts.Username == "dlc" {
+			errResp := createErrResponse(t, user, si.ID, "WRONG PASSWORD", nil)
+			m.Respond([]byte(errResp))
+		} else {
+			errResp := createErrResponse(t, user, si.ID, "BAD CREDS", nil)
+			m.Respond([]byte(errResp))
+		}
+	})
+	require_NoError(t, err)
+
+	// This one will use callout since not defined in server config.
+	nc, err = nats.Connect(s.ClientURL(), nats.UserInfo("dlc", "zzz"))
+	require_NoError(t, err)
+	nc.Close()
+	checkSubsPending(t, sub, 0)
+
+	checkAuthErrEvent := func(user, pass, reason string) {
+		_, err = nats.Connect(s.ClientURL(), nats.UserInfo(user, pass))
+		require_Error(t, err)
+
+		m, err := sub.NextMsg(time.Second)
+		require_NoError(t, err)
+
+		var dm DisconnectEventMsg
+		err = json.Unmarshal(m.Data, &dm)
+		require_NoError(t, err)
+
+		if !strings.Contains(dm.Reason, reason) {
+			t.Fatalf("Expected %q reason, but got %q", reason, dm.Reason)
+		}
+
+	}
+
+	checkAuthErrEvent("dlc", "xxx", "WRONG PASSWORD")
+	checkAuthErrEvent("rip", "abc", "BAD CREDS")
+}
diff --git a/server/auth_test.go b/server/auth_test.go
index 145cab06..45f92d40 100644
--- a/server/auth_test.go
+++ b/server/auth_test.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2018 The NATS Authors
+// Copyright 2012-2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 0dd49c52..107779aa 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -4833,28 +4833,3 @@ func TestFileStoreBadFirstAndFailedExpireAfterRestart(t *testing.T) {
 	require_True(t, state.LastSeq == 9)
 	require_True(t, state.LastTime == ts)
 }
-
-func TestFileStoreCompactAllWithDanglingLMB(t *testing.T) {
-	storeDir := t.TempDir()
-
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	subj, msg := "foo", []byte("ZZ")
-	for i := 0; i < 100; i++ {
-		_, _, err := fs.StoreMsg(subj, nil, msg)
-		require_NoError(t, err)
-	}
-
-	fs.RemoveMsg(100)
-	purged, err := fs.Compact(100)
-	require_NoError(t, err)
-	require_True(t, purged == 99)
-
-	_, _, err = fs.StoreMsg(subj, nil, msg)
-	require_NoError(t, err)
-}
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 11be3931..c03bdee3 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -2286,86 +2286,3 @@ func TestJetStreamClusterAfterPeerRemoveZeroState(t *testing.T) {
 		return nil
 	})
 }
-
-func TestJetStreamClusterMemLeaderRestart(t *testing.T) {
-	// Test if R3 clustered mem store asset leader server restarted, that asset remains stable with final quorum
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-
-	ml := c.leader()
-	nc, jsc := jsClientConnect(t, ml)
-	defer nc.Close()
-
-	_, err := jsc.AddStream(&nats.StreamConfig{
-		Name:     "foo",
-		Storage:  nats.MemoryStorage,
-		Subjects: []string{"foo.*"},
-		Replicas: 3,
-	})
-	require_NoError(t, err)
-
-	// load up messages
-	toSend := 10000
-	for i := 1; i <= toSend; i++ {
-		msg := []byte("Hello World")
-		if _, err = jsc.Publish("foo.a", msg); err != nil {
-			t.Fatalf("unexpected publish error: %v", err)
-		}
-	}
-
-	osi, err := jsc.StreamInfo("foo")
-	require_NoError(t, err)
-	// make sure 10000 msgs
-	require_True(t, osi.State.Msgs == uint64(toSend))
-
-	// Shutdown the stream leader server
-	rs := c.serverByName(osi.Cluster.Leader)
-	rs.Shutdown()
-
-	// Make sure that we have a META leader (there can always be a re-election)
-	c.waitOnLeader()
-
-	// Should still have quorum and a new leader
-	checkFor(t, time.Second, 200*time.Millisecond, func() error {
-		osi, err = jsc.StreamInfo("foo")
-		if err != nil {
-			return fmt.Errorf("expected healthy stream asset, got %s", err.Error())
-		}
-		if osi.Cluster.Leader == "" {
-			return fmt.Errorf("expected healthy stream asset with new leader")
-		}
-		if osi.State.Msgs != uint64(toSend) {
-			return fmt.Errorf("expected healthy stream asset %d messages, got %d messages", toSend, osi.State.Msgs)
-		}
-		return nil
-	})
-
-	// Now restart the old leader peer (old stream state)
-	oldrs := rs
-	rs, _ = RunServerWithConfig(rs.getOpts().ConfigFile)
-	defer rs.Shutdown()
-
-	// Replaced old with new server
-	for i := 0; i < len(c.servers); i++ {
-		if c.servers[i] == oldrs {
-			c.servers[i] = rs
-		}
-	}
-
-	// Wait for cluster to be formed
-	checkClusterFormed(t, c.servers...)
-
-	// Make sure that we have a leader (there can always be a re-election)
-	c.waitOnLeader()
-
-	// Can we get stream info after return
-	osi, err = jsc.StreamInfo("foo")
-	if err != nil {
-		t.Fatalf("expected stream asset info return, got %s", err.Error())
-	}
-
-	// When asset leader came back did we re-form with quorum
-	if osi.Cluster.Leader == "" {
-		t.Fatalf("expected a current leader after old leader restarted")
-	}
-}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 78edcfb9..92fe7b16 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -19101,38 +19101,3 @@ func TestJetStreamServerCrashOnPullConsumerDeleteWithInactiveThresholdAfterAck(t
 	_, err = js.StreamInfo("TEST")
 	require_NoError(t, err)
 }
-
-func TestJetStreamStreamUpdateSubjectsOverlapOthers(t *testing.T) {
-	s := RunBasicJetStreamServer(t)
-	defer s.Shutdown()
-
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"TEST"},
-	})
-	require_NoError(t, err)
-
-	_, err = js.UpdateStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"TEST", "foo.a"},
-	})
-	require_NoError(t, err)
-
-	_, err = js.AddStream(&nats.StreamConfig{
-		Name:     "TEST2",
-		Subjects: []string{"TEST2"},
-	})
-	require_NoError(t, err)
-
-	// we expect an error updating stream TEST2 with subject that overlaps that used by TEST
-	// foo.a fails too, but foo.* also double-check for sophisticated overlap match
-	_, err = js.UpdateStream(&nats.StreamConfig{
-		Name:     "TEST2",
-		Subjects: []string{"TEST2", "foo.*"},
-	})
-	require_Error(t, err)
-	require_Contains(t, err.Error(), "overlap")
-}
diff --git a/server/jwt_test.go b/server/jwt_test.go
index 8d91c58b..5df09621 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -3141,7 +3141,7 @@ func TestJWTBearerWithBadIssuerToken(t *testing.T) {
 func TestJWTExpiredUserCredentialsRenewal(t *testing.T) {
 	createTmpFile := func(t *testing.T, content []byte) string {
 		t.Helper()
-		conf := createTempFile(t, "")
+		conf := createTempFile(t, _EMPTY_)
 		fName := conf.Name()
 		conf.Close()
 		if err := os.WriteFile(fName, content, 0666); err != nil {
diff --git a/server/reload_test.go b/server/reload_test.go
index 9f334db1..35712b61 100644
--- a/server/reload_test.go
+++ b/server/reload_test.go
@@ -1,4 +1,4 @@
-// Copyright 2017-2021 The NATS Authors
+// Copyright 2017-2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -66,7 +66,7 @@ func newOptionsFromContent(t *testing.T, content []byte) (*Options, string) {
 
 func createConfFile(t testing.TB, content []byte) string {
 	t.Helper()
-	conf := createTempFile(t, "")
+	conf := createTempFile(t, _EMPTY_)
 	fName := conf.Name()
 	conf.Close()
 	if err := os.WriteFile(fName, content, 0666); err != nil {