(2.12) [ADDED] Trusted proxies support

See [ADR](https://github.com/nats-io/nats-architecture-and-design/pull/361)

The server will now log:
```
[11484] 2025/08/07 14:26:12.937393 [ERR] 127.0.0.1:54959 - cid:23 - proxy connection required - User "user"
```

And here is the output of subs on different system subjects:
```
$ nats sub -s nats://sys:pass@127.0.0.1:4222 '$SYS.SERVER.*.CLIENT.AUTH.ERR'
14:26:06 Subscribing on $SYS.SERVER.*.CLIENT.AUTH.ERR
[#1] Received on "$SYS.SERVER.NDBBGS7SKCSV7C2V64FQLG37EZ3IJT3LGT2G57KFKP6CMHAE2VTU4H2O.CLIENT.AUTH.ERR"
{"type":"io.nats.server.advisory.v1.client_disconnect",...,"reason":"Proxy Required"}
```

```
$ nats sub -s nats://sys:pass@127.0.0.1:4222 '$SYS.ACCOUNT.$G.DISCONNECT'
14:26:10 Subscribing on $SYS.ACCOUNT.$G.DISCONNECT
[#1] Received on "$SYS.ACCOUNT.$G.DISCONNECT"
{"type":"io.nats.server.advisory.v1.client_disconnect",...,"reason":"Proxy Required"}
```

And Connz:
```
nats req -s nats://sys:pass@127.0.0.1:4222 '$SYS.REQ.SERVER.NDBBGS7SKCSV7C2V64FQLG37EZ3IJT3LGT2G57KFKP6CMHAE2VTU4H2O.CONNZ' '{"state":1,"cid":23}'
14:34:16 Sending request on "$SYS.REQ.SERVER.NDBBGS7SKCSV7C2V64FQLG37EZ3IJT3LGT2G57KFKP6CMHAE2VTU4H2O.CONNZ"
14:34:16 Received with rtt 229.43µs
{"server":{"name":"NDBBGS7SKCSV7C2V64FQLG37EZ3IJT3LGT2G57KFKP6CMHAE2VTU4H2O",...,"connections":[{"cid":23,"kind":"Client",...,"reason":"Proxy Required"..."}]}}
```

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/google/go-tpm v0.9.5
 	github.com/klauspost/compress v1.18.0
 	github.com/minio/highwayhash v1.0.3
-	github.com/nats-io/jwt/v2 v2.7.4
+	github.com/nats-io/jwt/v2 v2.8.0
 	github.com/nats-io/nats.go v1.44.0
 	github.com/nats-io/nkeys v0.4.11
 	github.com/nats-io/nuid v1.0.1

go.sum

@@ -8,8 +8,8 @@ github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zt
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q=
 github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
-github.com/nats-io/jwt/v2 v2.7.4 h1:jXFuDDxs/GQjGDZGhNgH4tXzSUK6WQi2rsj4xmsNOtI=
-github.com/nats-io/jwt/v2 v2.7.4/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
+github.com/nats-io/jwt/v2 v2.8.0 h1:K7uzyz50+yGZDO5o772eRE7atlcSEENpL7P+b74JV1g=
+github.com/nats-io/jwt/v2 v2.8.0/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
 github.com/nats-io/nats.go v1.44.0 h1:ECKVrDLdh/kDPV1g0gAQ+2+m2KprqZK5O/eJAyAnH2M=
 github.com/nats-io/nats.go v1.44.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
 github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=

server/auth.go

@@ -25,6 +25,7 @@ import (
 	"net"
 	"net/url"
 	"regexp"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -65,6 +66,7 @@ type NkeyUser struct {
 	Account                *Account            `json:"account,omitempty"`
 	SigningKey             string              `json:"signing_key,omitempty"`
 	AllowedConnectionTypes map[string]struct{} `json:"connection_types,omitempty"`
+	ProxyRequired          bool                `json:"proxy_required,omitempty"`
 }
 
 // User is for multiple accounts/users.
@@ -75,6 +77,7 @@ type User struct {
 	Account                *Account            `json:"account,omitempty"`
 	ConnectionDeadline     time.Time           `json:"connection_deadline,omitempty"`
 	AllowedConnectionTypes map[string]struct{} `json:"connection_types,omitempty"`
+	ProxyRequired          bool                `json:"proxy_required,omitempty"`
 }
 
 // clone performs a deep copy of the User struct, returning a new clone with
@@ -593,10 +596,30 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 		ao   bool // auth override
 	)
 
+	// Little helper that will log the error as a debug statement, set the auth error in
+	// the connection and return false to indicate authentication failure.
+	setProxyAuthError := func(err error) bool {
+		c.Debugf(err.Error())
+		c.setAuthError(err)
+		return false
+	}
+
+	// Indicate if this connection came from a trusted proxy. Note that if
+	// trustedProxy could be false even if the connection is proxied, but it
+	// means that there was no trusted proxy configured.
+	trustedProxy, ok := s.proxyCheck(c, opts)
+	if trustedProxy && !ok {
+		return setProxyAuthError(ErrAuthProxyNotTrusted)
+	}
+
+	var proxyRequired bool
 	// Check if we have auth callouts enabled at the server level or in the bound account.
 	defer func() {
-		// Default reason
-		reason := AuthenticationViolation.String()
+		authErr := c.getAuthError()
+		if authErr == nil {
+			authErr = ErrAuthentication
+		}
+		reason := getAuthErrClosedState(authErr).String()
 		// No-op
 		if juc == nil && opts.AuthCallout == nil {
 			if !authorized {
@@ -656,7 +679,7 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 		// 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)
+			authorized, reason = s.processClientOrLeafCallout(c, opts, proxyRequired, trustedProxy)
 		}
 		// 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 {
@@ -774,6 +797,10 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 			c.Debugf("User JWT not valid: %v", err)
 			return false
 		}
+		if proxyRequired = juc.ProxyRequired; proxyRequired && !trustedProxy {
+			s.mu.Unlock()
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		vr := jwt.CreateValidationResults()
 		juc.Validate(vr)
 		if vr.IsBlocking(true) {
@@ -1050,6 +1077,9 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 	}
 
 	if nkey != nil {
+		if proxyRequired = nkey.ProxyRequired; proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		// If we did not match noAuthUser check signature which is required.
 		if nkey.Nkey != noAuthUser {
 			if c.opts.Sig == _EMPTY_ {
@@ -1081,6 +1111,9 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 		return true
 	}
 	if user != nil {
+		if proxyRequired = user.ProxyRequired; proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		ok = comparePasswords(user.Password, c.opts.Password)
 		// If we are authorized, register the user which will properly setup any permissions
 		// for pub/sub authorizations.
@@ -1091,6 +1124,9 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 	}
 
 	if c.kind == CLIENT {
+		if proxyRequired = opts.ProxyRequired; proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		if token != _EMPTY_ {
 			return comparePasswords(token, c.opts.Token)
 		} else if username != _EMPTY_ {
@@ -1103,6 +1139,61 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 	return false
 }
 
+// If there are configured trusted proxies and this connection comes
+// from a proxy whose signature can be verified by one of the known
+// trusted key, this function will return `true, true`. If the signature
+// cannot be verified by any, it will return `true, false`.
+// If the connectio is not proxied, or there are no configured trusted
+// proxies, then this function returns `false, false`.
+//
+// Server lock MUST NOT be held on entry since this function will grab
+// the read lock to extract the list of proxy trusted keys. The signature
+// verification process will be done outside of the lock.
+func (s *Server) proxyCheck(c *client, opts *Options) (bool, bool) {
+	// If there is no signature or no configured trusted proxy, return false.
+	psig := c.opts.ProxySig
+	if psig == _EMPTY_ || opts.Proxies == nil || len(opts.Proxies.Trusted) == 0 {
+		return false, false
+	}
+	// Decode the signature.
+	sig, err := base64.RawURLEncoding.DecodeString(psig)
+	if err != nil {
+		c.Debugf("Proxy signature not valid base64")
+		return true, false
+	}
+	// Go through the trusted keys and verify the signature.
+	s.mu.RLock()
+	keys := slices.Clone(s.proxiesKeyPairs)
+	s.mu.RUnlock()
+	for _, kp := range keys {
+		// We stop at the first that is valid.
+		if err := kp.Verify(c.nonce, sig); err == nil {
+			pub, _ := kp.PublicKey()
+			// Track which proxy public key is used by this connection.
+			c.mu.Lock()
+			c.proxyKey = pub
+			cid := c.cid
+			c.mu.Unlock()
+			// Track this proxied connection so that it can be closed
+			// if the trusted key is removed on configuration reload.
+			s.mu.Lock()
+			if s.proxiedConns == nil {
+				s.proxiedConns = make(map[string]map[uint64]*client)
+			}
+			clients := s.proxiedConns[pub]
+			if clients == nil {
+				clients = make(map[uint64]*client)
+			}
+			clients[cid] = c
+			s.proxiedConns[pub] = clients
+			s.mu.Unlock()
+			return true, true
+		}
+	}
+	// We could not verify the signature, so indicate failure.
+	return true, false
+}
+
 func getTLSAuthDCs(rdns *pkix.RDNSequence) string {
 	dcOID := asn1.ObjectIdentifier{0, 9, 2342, 19200300, 100, 1, 25}
 	dcs := []string{}
@@ -1351,7 +1442,25 @@ func (s *Server) registerLeafWithAccount(c *client, account string) bool {
 func (s *Server) isLeafNodeAuthorized(c *client) bool {
 	opts := s.getOpts()
 
-	isAuthorized := func(username, password, account string) bool {
+	setProxyAuthError := func(err error) bool {
+		c.Debugf(err.Error())
+		c.setAuthError(err)
+		return false
+	}
+
+	isAuthorized := func(username, password, account string, proxyRequired bool) bool {
+		trustedProxy, ok := s.proxyCheck(c, opts)
+		if trustedProxy && !ok {
+			return setProxyAuthError(ErrAuthProxyNotTrusted)
+		}
+		// A given user may not be required, but if the boolean is set at the
+		// authorization top-level, then override.
+		if !proxyRequired && opts.LeafNode.ProxyRequired {
+			proxyRequired = true
+		}
+		if proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		if username != c.opts.Username {
 			return false
 		}
@@ -1365,8 +1474,16 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 	// The user in CONNECT must match. We will bind to the account associated
 	// with that user (from the leafnode's authorization{} config).
 	if opts.LeafNode.Username != _EMPTY_ {
-		return isAuthorized(opts.LeafNode.Username, opts.LeafNode.Password, opts.LeafNode.Account)
+		return isAuthorized(opts.LeafNode.Username, opts.LeafNode.Password, opts.LeafNode.Account,
+			opts.LeafNode.ProxyRequired)
 	} else if opts.LeafNode.Nkey != _EMPTY_ {
+		trustedProxy, ok := s.proxyCheck(c, opts)
+		if trustedProxy && !ok {
+			return false
+		}
+		if opts.LeafNode.ProxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		if c.opts.Nkey != opts.LeafNode.Nkey {
 			return false
 		}
@@ -1420,7 +1537,7 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 			}
 			// This will authorize since are using an existing user,
 			// but it will also register with proper account.
-			return isAuthorized(user.Username, user.Password, accName)
+			return isAuthorized(user.Username, user.Password, accName, user.ProxyRequired)
 		}
 
 		// This is expected to be a very small array.
@@ -1430,7 +1547,7 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 				if u.Account != nil {
 					accName = u.Account.Name
 				}
-				return isAuthorized(u.Username, u.Password, accName)
+				return isAuthorized(u.Username, u.Password, accName, u.ProxyRequired)
 			}
 		}
 		return false
@@ -1535,3 +1652,45 @@ func validateNoAuthUser(o *Options, noAuthUser string) error {
 		`no_auth_user: "%s" not present as user or nkey in authorization block or account configuration`,
 		noAuthUser)
 }
+
+func validateProxies(o *Options) error {
+	if o.Proxies == nil {
+		return nil
+	}
+	for _, p := range o.Proxies.Trusted {
+		if !nkeys.IsValidPublicKey(p.Key) {
+			return fmt.Errorf("proxy trusted key %q is invalid", p.Key)
+		}
+	}
+	return nil
+}
+
+// Create a list of nkeys.KeyPair corresponding to the public keys
+// of the Proxies.TrustedKeys list.
+// Server lock must be held on entry.
+func (s *Server) processProxiesTrustedKeys() {
+	// We could be here on reload.
+	if s.proxiesKeyPairs != nil {
+		s.proxiesKeyPairs = s.proxiesKeyPairs[:0]
+	}
+	if opts := s.getOpts(); opts.Proxies == nil {
+		return
+	}
+	for _, p := range s.getOpts().Proxies.Trusted {
+		// Can't fail since we have already checked that it was a valid key.
+		kp, _ := nkeys.FromPublicKey(p.Key)
+		s.proxiesKeyPairs = append(s.proxiesKeyPairs, kp)
+	}
+}
+
+// Returns the connection's `ClosedState` for the given authenication error.
+func getAuthErrClosedState(authErr error) ClosedState {
+	switch authErr {
+	case ErrAuthProxyNotTrusted:
+		return ProxyNotTrusted
+	case ErrAuthProxyRequired:
+		return ProxyRequired
+	default:
+		return AuthenticationViolation
+	}
+}

server/auth_callout.go

@@ -33,7 +33,7 @@ const (
 )
 
 // Process a callout on this client's behalf.
-func (s *Server) processClientOrLeafCallout(c *client, opts *Options) (authorized bool, errStr string) {
+func (s *Server) processClientOrLeafCallout(c *client, opts *Options, proxyRequired, trustedProxy bool) (authorized bool, errStr string) {
 	isOperatorMode := len(opts.TrustedKeys) > 0
 
 	// this is the account the user connected in, or the one running the callout
@@ -245,6 +245,16 @@ func (s *Server) processClientOrLeafCallout(c *client, opts *Options) (authorize
 			respCh <- titleCase(err.Error())
 			return
 		}
+		// If the caller had established that the user should go through a proxy,
+		// or if the `arc` JWT requires it, and we don't have a trusted proxy,
+		// reject the connection.
+		if (proxyRequired || arc.ProxyRequired) && !trustedProxy {
+			err = ErrAuthProxyRequired
+			c.setAuthError(err)
+			c.authViolation()
+			respCh <- titleCase(err.Error())
+			return
+		}
 		vr := jwt.CreateValidationResults()
 		arc.Validate(vr)
 		if len(vr.Issues) > 0 {

server/auth_callout_test.go

@@ -241,6 +241,11 @@ func TestAuthCalloutBasics(t *testing.T) {
 			}
 			ujwt := createAuthUser(t, user, _EMPTY_, globalAccountName, "", nil, 10*time.Minute, &j)
 			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
+		} else if opts.Username == "proxy" {
+			var j jwt.UserPermissionLimits
+			j.ProxyRequired = true
+			ujwt := createAuthUser(t, user, _EMPTY_, globalAccountName, "", nil, 10*time.Minute, &j)
+			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
 		} else {
 			// Nil response signals no authentication.
 			m.Respond(nil)
@@ -252,6 +257,9 @@ func TestAuthCalloutBasics(t *testing.T) {
 	// This one should fail since bad password.
 	at.RequireConnectError(nats.UserInfo("dlc", "xxx"))
 
+	// This one should fail because it will require to be proxied and it is not.
+	at.RequireConnectError(nats.UserInfo("proxy", "xxx"))
+
 	// This one will use callout since not defined in server config.
 	nc := at.Connect(nats.UserInfo("dlc", "zzz"))
 	defer nc.Close()
@@ -1355,8 +1363,13 @@ func TestAuthCalloutAuthErrEvents(t *testing.T) {
 			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
 		} else if opts.Username == "dlc" {
 			m.Respond(serviceResponse(t, user, si.ID, "", "WRONG PASSWORD", 0))
-		} else {
+		} else if opts.Username == "rip" {
 			m.Respond(serviceResponse(t, user, si.ID, "", "BAD CREDS", 0))
+		} else if opts.Username == "proxy" {
+			var j jwt.UserPermissionLimits
+			j.ProxyRequired = true
+			ujwt := createAuthUser(t, user, _EMPTY_, "FOO", "", nil, 0, &j)
+			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
 		}
 	}
 
@@ -1382,13 +1395,19 @@ func TestAuthCalloutAuthErrEvents(t *testing.T) {
 		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)
+		// Convert both reasons to lower case to do the comparison.
+		dmr := strings.ToLower(dm.Reason)
+		r := strings.ToLower(reason)
+
+		if !strings.Contains(dmr, r) {
+			t.Fatalf("Expected %q reason, but got %q", r, dmr)
 		}
 	}
 
 	checkAuthErrEvent("dlc", "xxx", "WRONG PASSWORD")
 	checkAuthErrEvent("rip", "abc", "BAD CREDS")
+	// The auth callout uses as the reason the error string, not a closed state.
+	checkAuthErrEvent("proxy", "proxy", ErrAuthProxyRequired.Error())
 }
 
 func TestAuthCalloutConnectEvents(t *testing.T) {
@@ -2402,3 +2421,74 @@ func TestAuthCalloutLeafNodeAndConfigMode(t *testing.T) {
 	}
 
 }
+
+func TestAuthCalloutProxyRequiredInUserNotInAuthJWT(t *testing.T) {
+	conf := `
+		listen: "127.0.0.1:-1"
+		server_name: A
+		accounts {
+			AUTH: {
+				users: [ { user: auth, password: auth } ]
+			}
+			APP: {
+				users: [
+					{ user: user, password: pwd }
+					{ user: proxy, password: pwd, proxy_required: true }
+				]
+			}
+			SYS: {}
+		}
+		system_account: SYS
+
+		authorization {
+			timeout: 1s
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				auth_users: [ auth ]
+				account: AUTH
+			}
+		}
+	`
+	var invoked atomic.Int32
+	handler := func(m *nats.Msg) {
+		invoked.Add(1)
+		user, si, ci, opts, _ := decodeAuthRequest(t, m.Data)
+		require_True(t, si.Name == "A")
+		require_True(t, ci.Host == "127.0.0.1")
+		if opts.Username == "proxy" {
+			// If we don't set a ProxyRequired property explicitly here, but the
+			// user has it, so it should still be rejected.
+			ujwt := createAuthUser(t, user, _EMPTY_, "APP", _EMPTY_, nil, 10*time.Minute, nil)
+			m.Respond(serviceResponse(t, user, si.ID, ujwt, _EMPTY_, 0))
+		} else {
+			m.Respond(nil)
+		}
+	}
+	at := NewAuthTest(t, conf, handler, nats.UserInfo("auth", "auth"))
+	defer at.Cleanup()
+
+	sub, err := at.authClient.SubscribeSync(authErrorAccountEventSubj)
+	require_NoError(t, err)
+
+	// It should fail, even if the auth callout does not require proxy in its JWT.
+	// In other words, we reject if not proxied and the user config or the auth JWT
+	// requires proxy connection.
+	at.RequireConnectError(nats.UserInfo("proxy", "pwd"))
+
+	m, err := sub.NextMsg(time.Second)
+	require_NoError(t, err)
+
+	var dm DisconnectEventMsg
+	err = json.Unmarshal(m.Data, &dm)
+	require_NoError(t, err)
+
+	// Convert both reasons to lower case to do the comparison.
+	dmr := strings.ToLower(dm.Reason)
+	r := strings.ToLower(ErrAuthProxyRequired.Error())
+	if !strings.Contains(dmr, r) {
+		t.Fatalf("Expected %q reason, but got %q", r, dmr)
+	}
+
+	// Auth callout should have been invoked once.
+	require_Equal(t, 1, int(invoked.Load()))
+}

server/auth_test.go

@@ -16,6 +16,7 @@ package server
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"net/url"
@@ -27,6 +28,7 @@ import (
 
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
 )
 
 func TestUserCloneNilPermissions(t *testing.T) {
@@ -388,3 +390,373 @@ func TestNoAuthUserNoConnectProto(t *testing.T) {
 	time.Sleep(1200 * time.Millisecond)
 	checkClientsCount(t, s, 0)
 }
+
+type captureProxyRequiredLogger struct {
+	DummyLogger
+	ch chan string
+}
+
+func (l *captureProxyRequiredLogger) Debugf(format string, args ...any) {
+	msg := fmt.Sprintf(format, args...)
+	if strings.Contains(msg, ErrAuthProxyRequired.Error()) {
+		select {
+		case l.ch <- msg:
+		default:
+		}
+	}
+}
+
+func TestAuthProxyRequired(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		port: -1
+		authorization {
+			user: user
+			password: pwd
+			proxy_required: true
+		}
+	`))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	l := &captureProxyRequiredLogger{ch: make(chan string, 1)}
+	s.SetLogger(l, true, false)
+
+	_, err := nats.Connect(s.ClientURL(), nats.UserInfo("user", "pwd"))
+	require_True(t, errors.Is(err, nats.ErrAuthorization))
+
+	checkLog := func() {
+		t.Helper()
+		select {
+		case <-l.ch:
+			return
+		case <-time.After(time.Second):
+			t.Fatal("Did not get log statement")
+		}
+	}
+	checkLog()
+
+	drainLog := func() {
+		for {
+			select {
+			case <-l.ch:
+			default:
+				return
+			}
+		}
+	}
+	s.Shutdown()
+	drainLog()
+
+	nkUsr1, err := nkeys.CreateUser()
+	require_NoError(t, err)
+	nkPub1, err := nkUsr1.PublicKey()
+	require_NoError(t, err)
+
+	nkUsr2, err := nkeys.CreateUser()
+	require_NoError(t, err)
+	nkPub2, err := nkUsr2.PublicKey()
+	require_NoError(t, err)
+
+	conf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		authorization {
+			users: [
+				{user: user1, password: pwd1}
+				{user: user2, password: pwd2, proxy_required: true}
+				{user: user3, password: pwd3, proxy_required: false}
+				{nkey: "%s", proxy_required: true}
+				{nkey: "%s", proxy_required: false}
+			]
+		}
+	`, nkPub1, nkPub2))
+	s, _ = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	checkClients := func() {
+		t.Helper()
+		// Should connect ok.
+		nc := natsConnect(t, s.ClientURL(), nats.UserInfo("user1", "pwd1"))
+		nc.Close()
+
+		// Should not, since it requires going through proxy.
+		_, err = nats.Connect(s.ClientURL(), nats.UserInfo("user2", "pwd2"))
+		require_True(t, errors.Is(err, nats.ErrAuthorization))
+		checkLog()
+
+		// Should connect ok.
+		nc = natsConnect(t, s.ClientURL(), nats.UserInfo("user3", "pwd3"))
+		nc.Close()
+
+		// Should not, since it requires going through proxy.
+		_, err = nats.Connect(s.ClientURL(), nats.Nkey(nkPub1, func(nonce []byte) ([]byte, error) {
+			return nkUsr1.Sign(nonce)
+		}))
+		require_True(t, errors.Is(err, nats.ErrAuthorization))
+		checkLog()
+
+		// Should connect ok.
+		nc = natsConnect(t, s.ClientURL(), nats.Nkey(nkPub2, func(nonce []byte) ([]byte, error) {
+			return nkUsr2.Sign(nonce)
+		}))
+		nc.Close()
+	}
+	checkClients()
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		accounts {
+			A {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+					{nkey: "%s", proxy_required: true}
+					{nkey: "%s", proxy_required: false}
+				]
+			}
+			SYS { users: [ {user:sys, password: pwd} ] }
+		}
+		system_account: SYS
+	`, nkPub1, nkPub2))
+	s, _ = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	snc := natsConnect(t, s.ClientURL(), nats.UserInfo("sys", "pwd"))
+	defer snc.Close()
+	sub := natsSubSync(t, snc, "$SYS.SERVER.*.CLIENT.AUTH.ERR")
+	sub2 := natsSubSync(t, snc, "$SYS.ACCOUNT.A.DISCONNECT")
+	natsFlush(t, snc)
+
+	checkClients()
+
+	// We should get 2 authentication error messages, saying that
+	// the reason is "ProxyRequired".
+	for range 2 {
+		var de DisconnectEventMsg
+		msg := natsNexMsg(t, sub, time.Second)
+		err = json.Unmarshal(msg.Data, &de)
+		require_NoError(t, err)
+		require_Equal(t, de.Reason, ProxyRequired.String())
+	}
+
+	// We should get 3 disconnect events since only 3 have connected
+	// ok and then just closed.
+	for range 3 {
+		var de DisconnectEventMsg
+		msg := natsNexMsg(t, sub2, time.Second)
+		err = json.Unmarshal(msg.Data, &de)
+		require_NoError(t, err)
+		require_Equal(t, de.Reason, ClientClosed.String())
+	}
+	// Make sure there is no more.
+	if msg, err := sub2.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Should have received only 3 disconnect messages, got another: %s", msg.Data)
+	}
+
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes: {
+			port: -1
+			authorization {
+				user: user
+				password: pwd
+				proxy_required: true
+			}
+		}
+	`))
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	lconf := createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [ {url: "nats://user:pwd@127.0.0.1:%d"} ]
+		}
+	`, o.LeafNode.Port))
+	leaf, _ := RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes: {
+			port: -1
+			reconnect_interval: "50ms"
+			authorization {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+				]
+			}
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	lconf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [ {url: "nats://user2:pwd@127.0.0.1:%d"} ]
+		}
+	`, o.LeafNode.Port))
+	leaf, _ = RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes: {
+			port: -1
+			authorization {
+				nkey: %s
+				proxy_required: true
+			}
+		}
+	`, nkPub1))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	nkSeed1, err := nkUsr1.Seed()
+	require_NoError(t, err)
+
+	lconf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [
+				{
+					url: "nats://127.0.0.1:%d"
+					nkey: "%s"
+				}
+			]
+		}
+	`, o.LeafNode.Port, nkSeed1))
+	leaf, _ = RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+	s.Shutdown()
+	drainLog()
+
+	// Check with operator mode.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		server_name: OP
+		operator = "../test/configs/nkeys/op.jwt"
+		resolver = MEMORY
+		listen: "127.0.0.1:-1"
+		leafnodes {
+			listen: "127.0.0.1:-1"
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	_, akp := createAccount(s)
+	kp, err := nkeys.CreateUser()
+	require_NoError(t, err)
+	pub, err := kp.PublicKey()
+	require_NoError(t, err)
+	nuc := jwt.NewUserClaims(pub)
+	nuc.ProxyRequired = true
+	ujwt, err := nuc.Encode(akp)
+	require_NoError(t, err)
+
+	lopts := &DefaultTestOptions
+	u, err := url.Parse(fmt.Sprintf("nats://%s:%d", o.LeafNode.Host, o.LeafNode.Port))
+	require_NoError(t, err)
+	remote := &RemoteLeafOpts{URLs: []*url.URL{u}}
+	remote.SignatureCB = func(nonce []byte) (string, []byte, error) {
+		sig, err := kp.Sign(nonce)
+		return ujwt, sig, err
+	}
+	lopts.LeafNode.Remotes = []*RemoteLeafOpts{remote}
+	lopts.LeafNode.ReconnectInterval = 100 * time.Millisecond
+	leaf = RunServer(lopts)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+
+	// Try with an user.
+	drainLog()
+
+	_, err = nats.Connect(s.ClientURL(), createUserCredsEx(t, nuc, akp))
+	require_True(t, errors.Is(err, nats.ErrAuthorization))
+	checkLog()
+
+	drainLog()
+
+	// Try with creds file.
+	kp, err = nkeys.CreateUser()
+	require_NoError(t, err)
+	useed, err := kp.Seed()
+	require_NoError(t, err)
+	pub, err = kp.PublicKey()
+	require_NoError(t, err)
+	nuc = jwt.NewUserClaims(pub)
+	nuc.ProxyRequired = true
+	ujwt, err = nuc.Encode(akp)
+	require_NoError(t, err)
+	credsFile := genCredsFile(t, ujwt, useed)
+
+	lconf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [
+				{
+					url: "nats://127.0.0.1:%d"
+					credentials: '%s'
+				}
+			]
+		}
+	`, o.LeafNode.Port, credsFile))
+	leaf, _ = RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+
+	s.Shutdown()
+	drainLog()
+}

server/client.go

@@ -223,6 +223,8 @@ const (
 	MinimumVersionRequired
 	ClusterNamesIdentical
 	Kicked
+	ProxyNotTrusted
+	ProxyRequired
 )
 
 // Some flags passed to processMsgResults
@@ -271,6 +273,7 @@ type client struct {
 	msgb       [msgScratchSize]byte
 	last       time.Time
 	lastIn     time.Time
+	proxyKey   string
 
 	repliesSincePrune uint16
 	lastReplyPrune    time.Time
@@ -299,6 +302,13 @@ type client struct {
 	nameTag string
 
 	tlsTo *time.Timer
+
+	// Authentication error override. This is used because the authentication
+	// stack is simply returning a boolean, and the only authentication error
+	// reported is the generic `ErrAuthentication`. In the authentication code,
+	// if we want to report a different error, we can now set this field
+	// and `authViolation()` will use that one.
+	authErr error
 }
 
 type rrTracking struct {
@@ -653,6 +663,9 @@ type ClientOpts struct {
 
 	// Leafnodes
 	RemoteAccount string `json:"remote_account,omitempty"`
+
+	// Proxy would include its own nonce signature.
+	ProxySig string `json:"proxy_sig,omitempty"`
 }
 
 var defaultOpts = ClientOpts{Verbose: true, Pedantic: true, Echo: true}
@@ -2277,6 +2290,12 @@ func (c *client) accountAuthExpired() {
 }
 
 func (c *client) authViolation() {
+	authErr := c.getAuthError()
+	if authErr == nil {
+		authErr = ErrAuthentication
+	}
+	reason := getAuthErrClosedState(authErr)
+
 	var s *Server
 	var hasTrustedNkeys, hasNkeys, hasUsers bool
 	if s = c.srv; s != nil {
@@ -2285,30 +2304,31 @@ func (c *client) authViolation() {
 		hasNkeys = s.nkeys != nil
 		hasUsers = s.users != nil
 		s.mu.RUnlock()
-		defer s.sendAuthErrorEvent(c)
+		defer s.sendAuthErrorEvent(c, reason.String())
 	}
 
 	if hasTrustedNkeys {
-		c.Errorf("%v", ErrAuthentication)
+		c.Errorf("%v", authErr)
 	} else if hasNkeys {
 		c.Errorf("%s - Nkey %q",
-			ErrAuthentication.Error(),
+			authErr.Error(),
 			c.opts.Nkey)
 	} else if hasUsers {
 		c.Errorf("%s - User %q",
-			ErrAuthentication.Error(),
+			authErr.Error(),
 			c.opts.Username)
 	} else {
 		if c.srv != nil {
-			c.Errorf(ErrAuthentication.Error())
+			c.Errorf(authErr.Error())
 		}
 	}
 	if c.isMqtt() {
 		c.mqttEnqueueConnAck(mqttConnAckRCNotAuthorized, false)
 	} else {
+		// Send this to client, regardless of the authErr override.
 		c.sendErr("Authorization Violation")
 	}
-	c.closeConnection(AuthenticationViolation)
+	c.closeConnection(reason)
 }
 
 func (c *client) maxAccountConnExceeded() {
@@ -6398,3 +6418,18 @@ func (c *client) setFirstPingTimer() {
 	}
 	c.ping.tmr = time.AfterFunc(d, c.processPingTimer)
 }
+
+// Sets this error as the authentication error. To be used in authViolation()
+// to report an error different of `ErrAuthentication`.
+func (c *client) setAuthError(err error) {
+	c.mu.Lock()
+	c.authErr = err
+	c.mu.Unlock()
+}
+
+// Returns the authentication error set in the connection, possibly nil.
+func (c *client) getAuthError() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.authErr
+}

server/config_check_test.go

@@ -2153,6 +2153,251 @@ func TestConfigCheck(t *testing.T) {
 			errorLine: 3,
 			errorPos:  5,
 		},
+		{
+			name: "Proxies wrong type",
+			config: `
+				port: -1
+				proxies: 123
+			`,
+			err:       errors.New("expected proxies to be a map/struct, got int64"),
+			errorLine: 3,
+			errorPos:  5,
+		},
+		{
+			name: "Proxies unknown field",
+			config: `
+				port: -1
+				proxies: {
+					field1: 123
+				}
+			`,
+			err:       errors.New("unknown field"),
+			errorLine: 4,
+			errorPos:  6,
+		},
+		{
+			name: "Proxies trusted wrong type",
+			config: `
+				port: -1
+				proxies: {
+					trusted: 123
+				}
+			`,
+			err:       errors.New("expected proxies' trusted field to be an array, got int64"),
+			errorLine: 4,
+			errorPos:  6,
+		},
+		{
+			name: "Proxies trusted key wrong type",
+			config: `
+				port: -1
+				proxies: {
+					trusted: [
+						"abc",
+						"def"
+					]
+				}
+			`,
+			err:       errors.New("expected proxies' trusted entry to be a map/struct, got string"),
+			errorLine: 5,
+			errorPos:  8,
+		},
+		{
+			name: "Proxies trusted unknown field",
+			config: `
+				port: -1
+				proxies: {
+					trusted: [
+						{
+							key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+							field1: 123
+						}
+					]
+				}
+			`,
+			err:       errors.New("unknown field"),
+			errorLine: 7,
+			errorPos:  8,
+		},
+		{
+			name: "Proxies trusted key invalid",
+			config: `
+				port: -1
+				proxies: {
+					trusted: [
+						{key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"}
+						{key: "bad1"}
+						{key: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI"}
+						{key: "bad2"}
+					]
+				}
+			`,
+			err:       errors.New("invalid proxy key"),
+			errorLine: 6,
+			errorPos:  8,
+		},
+		{
+			name: "Auth user require proxied wrong type",
+			config: `
+				port: -1
+				authorization {
+					user: user
+					password: pwd
+					proxy_required: 123
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 6,
+			errorPos:  6,
+		},
+		{
+			name: "Auth users require proxied wrong type",
+			config: `
+				port: -1
+				authorization {
+					users: [
+						{
+							user: user
+							password: pwd
+							proxy_required: 123
+						}
+					]
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 8,
+			errorPos:  8,
+		},
+		{
+			name: "Auth nkey users require proxied wrong type",
+			config: `
+				port: -1
+				authorization {
+					users: [
+						{
+							nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+							proxy_required: 123
+						}
+					]
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 7,
+			errorPos:  8,
+		},
+		{
+			name: "Account users require proxied wrong type",
+			config: `
+				port: -1
+				accounts {
+					A: {
+						users: [
+							{
+								user: user
+								password: pwd
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 9,
+			errorPos:  9,
+		},
+		{
+			name: "Account nkey users require proxied wrong type",
+			config: `
+				port: -1
+				accounts {
+					A: {
+						users: [
+							{
+								nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 8,
+			errorPos:  9,
+		},
+		{
+			name: "Leafnode user require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						user: user
+						password: pwd
+						proxy_required: 123
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 8,
+			errorPos:  7,
+		},
+		{
+			name: "Leafnode neky user require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+						proxy_required: 123
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 7,
+			errorPos:  7,
+		},
+		{
+			name: "Leafnode users require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						users: [
+							{
+								user: user
+								password: pwd
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 10,
+			errorPos:  9,
+		},
+		{
+			name: "Leafnode nkey users require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						users: [
+							{
+								nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 9,
+			errorPos:  9,
+		},
 	}
 
 	checkConfig := func(config string) error {

server/errors.go

@@ -31,6 +31,14 @@ var (
 	// ErrAuthExpired represents an expired authorization due to timeout.
 	ErrAuthExpired = errors.New("authentication expired")
 
+	// ErrAuthProxyNotTrusted represents an error condition on failed authentication
+	// due to a connection from a proxy not in the list of trusted proxies.
+	ErrAuthProxyNotTrusted = errors.New("proxy is not trusted")
+
+	// ErrAuthProxyRequired represents an error condition on failed authentication
+	// due to a connection not coming from a proxy.
+	ErrAuthProxyRequired = errors.New("proxy connection required")
+
 	// ErrMaxPayload represents an error condition when the payload is too big.
 	ErrMaxPayload = errors.New("maximum payload exceeded")
 

server/events.go

@@ -2610,7 +2610,7 @@ func (s *Server) accountDisconnectEvent(c *client, now time.Time, reason string)
 }
 
 // This is the system level event sent to the system account for operators.
-func (s *Server) sendAuthErrorEvent(c *client) {
+func (s *Server) sendAuthErrorEvent(c *client, reason string) {
 	s.mu.Lock()
 	if !s.eventsEnabled() {
 		s.mu.Unlock()
@@ -2658,7 +2658,7 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 				Bytes: c.outBytes,
 			},
 		},
-		Reason: AuthenticationViolation.String(),
+		Reason: reason,
 	}
 	c.mu.Unlock()
 

server/leafnode.go

@@ -756,6 +756,7 @@ func (s *Server) startLeafNodeAcceptLoop() {
 		Domain:        opts.JetStreamDomain,
 		Proto:         s.getServerProto(),
 		InfoOnConnect: true,
+		JSApiLevel:    JSApiLevel,
 	}
 	// If we have selected a random port...
 	if port == 0 {
@@ -1065,6 +1066,8 @@ func (s *Server) createLeafNode(conn net.Conn, rURL *url.URL, remote *leafNodeCf
 		if cm := opts.LeafNode.Compression.Mode; cm != CompressionNotSupported {
 			info.Compression = cm
 		}
+		// We always send a nonce for LEAF connections. Do not change that without
+		// taking into account presence of proxy trusted keys.
 		s.generateNonce(nonce[:])
 		s.mu.Unlock()
 	}
@@ -1804,9 +1807,13 @@ func (s *Server) removeLeafNodeConnection(c *client) {
 			c.leaf.gwSub = nil
 		}
 	}
+	proxyKey := c.proxyKey
 	c.mu.Unlock()
 	s.mu.Lock()
 	delete(s.leafs, cid)
+	if proxyKey != _EMPTY_ {
+		s.removeProxiedConn(proxyKey, cid)
+	}
 	s.mu.Unlock()
 	s.removeFromTempClients(cid)
 }

server/monitor.go

@@ -146,11 +146,17 @@ type ConnInfo struct {
 	NameTag        string         `json:"name_tag,omitempty"`
 	Tags           jwt.TagList    `json:"tags,omitempty"`
 	MQTTClient     string         `json:"mqtt_client,omitempty"` // This is the MQTT client id
+	Proxy          *ProxyInfo     `json:"proxy,omitempty"`
 
 	// Internal
 	rtt int64 // For fast sorting
 }
 
+// ProxyInfo represents the information about this proxied connection.
+type ProxyInfo struct {
+	Key string `json:"key"`
+}
+
 // TLSPeerCert contains basic information about a TLS peer certificate
 type TLSPeerCert struct {
 	Subject          string `json:"subject,omitempty"`
@@ -571,6 +577,7 @@ func (ci *ConnInfo) fill(client *client, nc net.Conn, now time.Time, auth bool)
 	// we need to use atomic here.
 	ci.InMsgs = atomic.LoadInt64(&client.inMsgs)
 	ci.InBytes = atomic.LoadInt64(&client.inBytes)
+	ci.Proxy = createProxyInfo(client)
 
 	// If the connection is gone, too bad, we won't set TLSVersion and TLSCipher.
 	// Exclude clients that are still doing handshake so we don't block in
@@ -593,6 +600,17 @@ func (ci *ConnInfo) fill(client *client, nc net.Conn, now time.Time, auth bool)
 	}
 }
 
+// If this client came from a trusted proxy, this will return a ProxyInfo
+// to be used in ConnInfo or LeafInfo.
+//
+// Client lock must be held on entry.
+func createProxyInfo(c *client) *ProxyInfo {
+	if c.proxyKey == _EMPTY_ {
+		return nil
+	}
+	return &ProxyInfo{Key: c.proxyKey}
+}
+
 func makePeerCerts(pc []*x509.Certificate) []*TLSPeerCert {
 	res := make([]*TLSPeerCert, len(pc))
 	for i, c := range pc {
@@ -1254,6 +1272,7 @@ type Varz struct {
 	PinnedAccountFail     uint64                 `json:"pinned_account_fails,omitempty"`    // PinnedAccountFail is how often user logon fails due to the issuer account not being pinned.
 	OCSPResponseCache     *OCSPResponseCacheVarz `json:"ocsp_peer_cache,omitempty"`         // OCSPResponseCache is the state of the OCSP cache // OCSPResponseCache holds information about
 	SlowConsumersStats    *SlowConsumersStats    `json:"slow_consumer_stats"`               // SlowConsumersStats is statistics about all detected Slow Consumer
+	Proxies               *ProxiesOptsVarz       `json:"proxies,omitempty"`
 }
 
 // JetStreamVarz contains basic runtime information about jetstream
@@ -1370,6 +1389,16 @@ type OCSPResponseCacheVarz struct {
 	Unknowns  int64  `json:"cached_unknown_responses,omitempty"` // Unknowns  is how many of the stored cache entries are unknown responses
 }
 
+// ProxiesOptsVarz contains monitoring proxies information
+type ProxiesOptsVarz struct {
+	Trusted []*ProxyOptsVarz `json:"trusted,omitempty"`
+}
+
+// ProxyOptsVarz contains monitoring proxy information
+type ProxyOptsVarz struct {
+	Key string `json:"key"`
+}
+
 // VarzOptions are the options passed to Varz().
 // Currently, there are no options defined.
 type VarzOptions struct{}
@@ -1720,6 +1749,19 @@ func (s *Server) updateVarzConfigReloadableFields(v *Varz) {
 	v.Websocket.TLSPinnedCerts = getPinnedCertsAsSlice(opts.Websocket.TLSPinnedCerts)
 
 	v.TLSOCSPPeerVerify = s.ocspPeerVerify && v.TLSRequired && s.opts.tlsConfigOpts != nil && s.opts.tlsConfigOpts.OCSPPeerConfig != nil && s.opts.tlsConfigOpts.OCSPPeerConfig.Verify
+
+	if opts.Proxies != nil {
+		if v.Proxies == nil {
+			v.Proxies = &ProxiesOptsVarz{}
+		}
+		trusted := make([]*ProxyOptsVarz, 0, len(opts.Proxies.Trusted))
+		for _, t := range opts.Proxies.Trusted {
+			trusted = append(trusted, &ProxyOptsVarz{Key: t.Key})
+		}
+		v.Proxies.Trusted = trusted
+	} else {
+		v.Proxies = nil
+	}
 }
 
 func getPinnedCertsAsSlice(certs PinnedCertSet) []string {
@@ -2254,20 +2296,21 @@ type LeafzOptions struct {
 
 // LeafInfo has detailed information on each remote leafnode connection.
 type LeafInfo struct {
-	ID          uint64   `json:"id"`
-	Name        string   `json:"name"`
-	IsSpoke     bool     `json:"is_spoke"`
-	Account     string   `json:"account"`
-	IP          string   `json:"ip"`
-	Port        int      `json:"port"`
-	RTT         string   `json:"rtt,omitempty"`
-	InMsgs      int64    `json:"in_msgs"`
-	OutMsgs     int64    `json:"out_msgs"`
-	InBytes     int64    `json:"in_bytes"`
-	OutBytes    int64    `json:"out_bytes"`
-	NumSubs     uint32   `json:"subscriptions"`
-	Subs        []string `json:"subscriptions_list,omitempty"`
-	Compression string   `json:"compression,omitempty"`
+	ID          uint64     `json:"id"`
+	Name        string     `json:"name"`
+	IsSpoke     bool       `json:"is_spoke"`
+	Account     string     `json:"account"`
+	IP          string     `json:"ip"`
+	Port        int        `json:"port"`
+	RTT         string     `json:"rtt,omitempty"`
+	InMsgs      int64      `json:"in_msgs"`
+	OutMsgs     int64      `json:"out_msgs"`
+	InBytes     int64      `json:"in_bytes"`
+	OutBytes    int64      `json:"out_bytes"`
+	NumSubs     uint32     `json:"subscriptions"`
+	Subs        []string   `json:"subscriptions_list,omitempty"`
+	Compression string     `json:"compression,omitempty"`
+	Proxy       *ProxyInfo `json:"proxy,omitempty"`
 }
 
 // Leafz returns a Leafz structure containing information about leafnodes.
@@ -2310,6 +2353,7 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 				OutBytes:    ln.outBytes,
 				NumSubs:     uint32(len(ln.subs)),
 				Compression: ln.leaf.compression,
+				Proxy:       createProxyInfo(ln),
 			}
 			if opts != nil && opts.Subscriptions {
 				lni.Subs = make([]string, 0, len(ln.subs))
@@ -2522,6 +2566,10 @@ func (reason ClosedState) String() string {
 		return "Cluster Names Identical"
 	case Kicked:
 		return "Kicked"
+	case ProxyNotTrusted:
+		return "Proxy Not Trusted"
+	case ProxyRequired:
+		return "Proxy Required"
 	}
 
 	return "Unknown State"

server/nkey.go

@@ -34,7 +34,7 @@ func (s *Server) NonceRequired() bool {
 // nonceRequired tells us if we should send a nonce.
 // Lock should be held on entry.
 func (s *Server) nonceRequired() bool {
-	return s.getOpts().AlwaysEnableNonce || len(s.nkeys) > 0 || s.trustedKeys != nil
+	return s.getOpts().AlwaysEnableNonce || len(s.nkeys) > 0 || s.trustedKeys != nil || len(s.proxiesKeyPairs) > 0
 }
 
 // Generate a nonce for INFO challenge.

server/opts.go

@@ -151,6 +151,7 @@ type LeafNodeOpts struct {
 	Port           int           `json:"port,omitempty"`
 	Username       string        `json:"-"`
 	Password       string        `json:"-"`
+	ProxyRequired  bool          `json:"-"`
 	Nkey           string        `json:"-"`
 	Account        string        `json:"-"`
 	Users          []*User       `json:"-"`
@@ -324,6 +325,7 @@ type Options struct {
 	NoSystemAccount            bool          `json:"-"`
 	Username                   string        `json:"-"`
 	Password                   string        `json:"-"`
+	ProxyRequired              bool          `json:"-"`
 	Authorization              string        `json:"-"`
 	AuthCallout                *AuthCallout  `json:"-"`
 	PingInterval               time.Duration `json:"ping_interval"`
@@ -445,6 +447,9 @@ type Options struct {
 	OCSPConfig    *OCSPConfig
 	tlsConfigOpts *TLSConfigOpts
 
+	// Proxies configuration.
+	Proxies *ProxiesConfig
+
 	// private fields, used to know if bool options are explicitly
 	// defined in config and/or command line params.
 	inConfig  map[string]bool
@@ -729,6 +734,8 @@ type authorization struct {
 	token string
 	nkey  string
 	acc   string
+	// If connection must come through proxy
+	proxyRequired bool
 	// Multiple Nkeys/Users
 	nkeys              []*NkeyUser
 	users              []*User
@@ -781,6 +788,17 @@ type OCSPConfig struct {
 	OverrideURLs []string
 }
 
+// ProxiesConfig represents the options of Proxies.
+type ProxiesConfig struct {
+	Trusted []*ProxyConfig
+}
+
+// ProxyConfig represents the options of Proxy.
+type ProxyConfig struct {
+	// Public key.
+	Key string
+}
+
 var tlsUsage = `
 TLS configuration is specified in the tls section of a configuration file:
 
@@ -1063,6 +1081,7 @@ func (o *Options) processConfigFileLine(k string, v any, errors *[]error, warnin
 		o.authBlockDefined = true
 		o.Username = auth.user
 		o.Password = auth.pass
+		o.ProxyRequired = auth.proxyRequired
 		o.Authorization = auth.token
 		o.AuthTimeout = auth.timeout
 		o.AuthCallout = auth.callout
@@ -1727,6 +1746,13 @@ func (o *Options) processConfigFileLine(k string, v any, errors *[]error, warnin
 		o.NoFastProducerStall = v.(bool)
 	case "max_closed_clients":
 		o.MaxClosedClients = int(v.(int64))
+	case "proxies":
+		proxies, err := parseProxies(tk, errors)
+		if err != nil {
+			*errors = append(*errors, err)
+			return
+		}
+		o.Proxies = proxies
 	default:
 		if au := atomic.LoadInt32(&allowUnknownTopLevelField); au == 0 && !tk.IsUsedVariable() {
 			err := &unknownConfigFieldErr{
@@ -2600,6 +2626,7 @@ func parseLeafNodes(v any, opts *Options, errors *[]error, warnings *[]error) er
 			}
 			opts.LeafNode.Username = auth.user
 			opts.LeafNode.Password = auth.pass
+			opts.LeafNode.ProxyRequired = auth.proxyRequired
 			opts.LeafNode.AuthTimeout = auth.timeout
 			opts.LeafNode.Account = auth.acc
 			opts.LeafNode.Users = auth.users
@@ -2726,6 +2753,8 @@ func parseLeafAuthorization(v any, errors, warnings *[]error) (*authorization, e
 			auth.users = users
 		case "account":
 			auth.acc = mv.(string)
+		case "proxy_required":
+			auth.proxyRequired = mv.(bool)
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
@@ -2784,6 +2813,8 @@ func parseLeafUsers(mv any, errors *[]error) ([]*User, error) {
 				// we need to create internal objects to store u/p and account
 				// name and have a server structure to hold that.
 				user.Account = NewAccount(v.(string))
+			case "proxy_required":
+				user.ProxyRequired = v.(bool)
 			default:
 				if !tk.IsUsedVariable() {
 					err := &unknownConfigFieldErr{
@@ -4273,6 +4304,8 @@ func parseAuthorization(v any, errors, warnings *[]error) (*authorization, error
 				continue
 			}
 			auth.callout = ac
+		case "proxy_required":
+			auth.proxyRequired = mv.(bool)
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
@@ -4345,6 +4378,9 @@ func parseUsers(mv any, errors *[]error) ([]*NkeyUser, []*User, error) {
 				cts := parseAllowedConnectionTypes(tk, &lt, v, errors)
 				nkey.AllowedConnectionTypes = cts
 				user.AllowedConnectionTypes = cts
+			case "proxy_required":
+				nkey.ProxyRequired = v.(bool)
+				user.ProxyRequired = v.(bool)
 			default:
 				if !tk.IsUsedVariable() {
 					err := &unknownConfigFieldErr{
@@ -5358,6 +5394,94 @@ func parseMQTT(v any, o *Options, errors *[]error, warnings *[]error) error {
 	return nil
 }
 
+func parseProxies(mv any, errors *[]error) (*ProxiesConfig, error) {
+	var (
+		tk      token
+		lt      token
+		proxies = &ProxiesConfig{}
+	)
+	defer convertPanicToErrorList(&lt, errors)
+
+	tk, mv = unwrapValue(mv, &lt)
+	pm, ok := mv.(map[string]any)
+	if !ok {
+		return nil, &configErr{tk, fmt.Sprintf("expected proxies to be a map/struct, got %T", mv)}
+	}
+	for mk, mv := range pm {
+		tk, _ = unwrapValue(mv, &lt)
+
+		switch strings.ToLower(mk) {
+		case "trusted":
+			trusted, err := parseProxiesTrusted(tk, errors)
+			if err != nil {
+				*errors = append(*errors, err)
+				continue
+			}
+			proxies.Trusted = trusted
+		default:
+			if !tk.IsUsedVariable() {
+				err := &unknownConfigFieldErr{
+					field: mk,
+					configErr: configErr{
+						token: tk,
+					},
+				}
+				*errors = append(*errors, err)
+			}
+		}
+	}
+	return proxies, nil
+}
+
+func parseProxiesTrusted(mv any, errors *[]error) ([]*ProxyConfig, error) {
+	var (
+		tk      token
+		lt      token
+		trusted []*ProxyConfig
+	)
+	defer convertPanicToErrorList(&lt, errors)
+
+	tk, mv = unwrapValue(mv, &lt)
+	ta, ok := mv.([]any)
+	if !ok {
+		return nil, &configErr{tk, fmt.Sprintf("expected proxies' trusted field to be an array, got %T", mv)}
+	}
+	for _, t := range ta {
+		tk, t = unwrapValue(t, &lt)
+		// Check its a map/struct
+		tm, ok := t.(map[string]any)
+		if !ok {
+			err := &configErr{tk, fmt.Sprintf("expected proxies' trusted entry to be a map/struct, got %T", t)}
+			*errors = append(*errors, err)
+			continue
+		}
+		proxy := &ProxyConfig{}
+		for k, v := range tm {
+			tk, v = unwrapValue(v, &lt)
+			switch strings.ToLower(k) {
+			case "key", "public_key":
+				proxy.Key = v.(string)
+				if !nkeys.IsValidPublicKey(proxy.Key) {
+					*errors = append(*errors, &configErr{tk, fmt.Sprintf("invalid proxy key %q", proxy.Key)})
+				}
+			default:
+				if !tk.IsUsedVariable() {
+					err := &unknownConfigFieldErr{
+						field: k,
+						configErr: configErr{
+							token: tk,
+						},
+					}
+					*errors = append(*errors, err)
+					continue
+				}
+			}
+		}
+		trusted = append(trusted, proxy)
+	}
+	return trusted, nil
+}
+
 // GenTLSConfig loads TLS related configuration parameters.
 func GenTLSConfig(tc *TLSConfigOpts) (*tls.Config, error) {
 	// Create the tls.Config from our options before including the certs.

server/opts_test.go

@@ -3753,3 +3753,171 @@ func parseConfigTolerantly(t *testing.T, data string) (*Options, error) {
 
 	return o, nil
 }
+
+func TestOptionsProxyTrustedKeys(t *testing.T) {
+	o := DefaultOptions()
+	o.Proxies = &ProxiesConfig{
+		Trusted: []*ProxyConfig{
+			{Key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"},
+			{Key: "bad1"},
+			{Key: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI"},
+			{Key: "bad2"},
+		},
+	}
+	err := validateOptions(o)
+	require_Error(t, err)
+	require_Equal(t, "proxy trusted key \"bad1\" is invalid", err.Error())
+
+	o.Proxies = &ProxiesConfig{
+		Trusted: []*ProxyConfig{
+			{Key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"},
+			{Key: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI"},
+		},
+	}
+	s := RunServer(o)
+	defer s.Shutdown()
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	opts := s.getOpts()
+	for i, kp := range s.proxiesKeyPairs {
+		pub, err := kp.PublicKey()
+		require_NoError(t, err)
+		require_Equal(t, opts.Proxies.Trusted[i].Key, pub)
+	}
+}
+
+func TestOptionsProxyRequired(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		port: -1
+		authorization {
+			user: user
+			password: pwd
+			proxy_required: true
+		}
+	`))
+	o, err := ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Equal(t, o.Username, "user")
+	require_Equal(t, o.Password, "pwd")
+	require_True(t, o.ProxyRequired)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		authorization {
+			users: [
+				{user: user1, password: pwd1}
+				{user: user2, password: pwd2, proxy_required: true}
+				{user: user3, password: pwd3, proxy_required: false}
+				{nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3", proxy_required: true}
+				{nkey: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI", proxy_required: false}
+			]
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+
+	checkUsersAndNkeys := func(users []*User, hasNkeys bool, nkeys []*NkeyUser) {
+		t.Helper()
+		var found bool
+
+		require_Len(t, len(users), 3)
+		for _, u := range users {
+			switch u.Username {
+			case "user1", "user3":
+				require_False(t, u.ProxyRequired)
+			case "user2":
+				require_True(t, u.ProxyRequired)
+				found = true
+			}
+		}
+		require_True(t, found)
+
+		if !hasNkeys {
+			return
+		}
+
+		found = false
+		require_Len(t, len(nkeys), 2)
+		for _, u := range nkeys {
+			switch u.Nkey {
+			case "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3":
+				require_True(t, u.ProxyRequired)
+				found = true
+			case "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI":
+				require_False(t, u.ProxyRequired)
+			}
+		}
+		require_True(t, found)
+	}
+	checkUsersAndNkeys(o.Users, true, o.Nkeys)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		accounts {
+			A {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+					{nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3", proxy_required: true}
+					{nkey: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI", proxy_required: false}
+				]
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Len(t, len(o.Accounts), 1)
+	require_Equal(t, o.Accounts[0].Name, "A")
+	checkUsersAndNkeys(o.Users, true, o.Nkeys)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				user: user
+				password: pwd
+				proxy_required: true
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Equal(t, o.LeafNode.Username, "user")
+	require_Equal(t, o.LeafNode.Password, "pwd")
+	require_True(t, o.LeafNode.ProxyRequired)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+				proxy_required: true
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Equal(t, o.LeafNode.Nkey, "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3")
+	require_True(t, o.LeafNode.ProxyRequired)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes: {
+			port: -1
+			authorization {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+				]
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	checkUsersAndNkeys(o.LeafNode.Users, false, nil)
+}

server/reload.go

@@ -1052,6 +1052,38 @@ func (s *Server) recheckPinnedCerts(curOpts *Options, newOpts *Options) {
 	}
 }
 
+type proxiesReload struct {
+	noopOption
+	add []string
+	del []string
+}
+
+func (p *proxiesReload) Apply(s *Server) {
+	var clients []*client
+	s.mu.Lock()
+	for _, k := range p.del {
+		cc := s.proxiedConns[k]
+		delete(s.proxiedConns, k)
+		if len(cc) > 0 {
+			for _, c := range cc {
+				clients = append(clients, c)
+			}
+		}
+	}
+	s.processProxiesTrustedKeys()
+	s.mu.Unlock()
+	if len(p.del) > 0 {
+		for _, c := range clients {
+			c.setAuthError(ErrAuthProxyNotTrusted)
+			c.authViolation()
+		}
+		s.Noticef("Reloaded: proxies trusted keys %q were removed", p.add)
+	}
+	if len(p.add) > 0 {
+		s.Noticef("Reloaded: proxies trusted keys %q were added", p.add)
+	}
+}
+
 // Reload reads the current configuration file and calls out to ReloadOptions
 // to apply the changes. This returns an error if the server was not started
 // with a config file or an option which doesn't support hot-swapping was changed.
@@ -1227,7 +1259,7 @@ func imposeOrder(value any) error {
 		slices.Sort(value.AllowedOrigins)
 	case string, bool, uint8, uint16, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
 		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList,
-		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher, *OCSPResponseCacheConfig:
+		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher, *OCSPResponseCacheConfig, *ProxiesConfig:
 		// explicitly skipped types
 	case *AuthCallout:
 	case JSTpmOpts:
@@ -1722,6 +1754,12 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			continue
 		case "nofastproducerstall":
 			diffOpts = append(diffOpts, &noFastProdStallReload{noStall: newValue.(bool)})
+		case "proxies":
+			new := newValue.(*ProxiesConfig)
+			old := oldValue.(*ProxiesConfig)
+			if add, del := diffProxiesTrustedKeys(old.Trusted, new.Trusted); len(add) > 0 || len(del) > 0 {
+				diffOpts = append(diffOpts, &proxiesReload{add: add, del: del})
+			}
 		default:
 			// TODO(ik): Implement String() on those options to have a nice print.
 			// %v is difficult to figure what's what, %+v print private fields and
@@ -2589,3 +2627,34 @@ addLoop:
 
 	return add, remove
 }
+
+func diffProxiesTrustedKeys(old, new []*ProxyConfig) ([]string, []string) {
+	var add []string
+	var del []string
+	// Both "old" and "new" lists should be small...
+	for _, op := range old {
+		var found bool
+		for _, np := range new {
+			if op.Key == np.Key {
+				found = true
+				break
+			}
+		}
+		if !found {
+			del = append(del, op.Key)
+		}
+	}
+	for _, np := range new {
+		var found bool
+		for _, op := range old {
+			if np.Key == op.Key {
+				found = true
+				break
+			}
+		}
+		if !found {
+			add = append(add, np.Key)
+		}
+	}
+	return add, del
+}

server/server.go

@@ -133,6 +133,7 @@ type Info struct {
 	ConnectInfo       bool     `json:"connect_info,omitempty"`   // When true this is the server INFO response to CONNECT
 	RemoteAccount     string   `json:"remote_account,omitempty"` // Lets the client or leafnode side know the remote account that they bind to.
 	IsSystemAccount   bool     `json:"acc_is_sys,omitempty"`     // Indicates if the account is a system account.
+	JSApiLevel        int      `json:"api_lvl,omitempty"`
 
 	// Route Specific
 	Import        *SubjectPermission `json:"import,omitempty"`
@@ -371,6 +372,11 @@ type Server struct {
 	// Controls whether or not the account NRG capability is set in statsz.
 	// Currently used by unit tests to simulate nodes not supporting account NRG.
 	accountNRGAllowed atomic.Bool
+
+	// List of proxies trusted keys in `KeyPair` form so we can do signature
+	// verification when processing incoming proxy connections.
+	proxiesKeyPairs []nkeys.KeyPair
+	proxiedConns    map[string]map[uint64]*client
 }
 
 // For tracking JS nodes.
@@ -732,6 +738,7 @@ func NewServer(opts *Options) (*Server, error) {
 		Headers:      !opts.NoHeaderSupport,
 		Cluster:      opts.Cluster.Name,
 		Domain:       opts.JetStreamDomain,
+		JSApiLevel:   JSApiLevel,
 	}
 
 	if tlsReq && !info.TLSRequired {
@@ -800,6 +807,11 @@ func NewServer(opts *Options) (*Server, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	// If there are proxies trusted public keys in the configuration
+	// this will fill create the corresponding list of nkeys.KeyPair
+	// that we can use for signature verification.
+	s.processProxiesTrustedKeys()
+
 	// Place ourselves in the JetStream nodeInfo if needed.
 	if opts.JetStream {
 		ourNode := getHash(serverName)
@@ -1125,6 +1137,10 @@ func validateOptions(o *Options) error {
 	if err := validateAuth(o); err != nil {
 		return err
 	}
+	// Check that proxies is properly configured.
+	if err := validateProxies(o); err != nil {
+		return err
+	}
 	// Check that gateway is properly configured. Returns no error
 	// if there is no gateway defined.
 	if err := validateGatewayOptions(o); err != nil {
@@ -3589,6 +3605,7 @@ func (s *Server) removeClient(c *client) {
 		if c.kind == CLIENT && c.opts.Protocol >= ClientProtoInfo {
 			updateProtoInfoCount = true
 		}
+		proxyKey := c.proxyKey
 		c.mu.Unlock()
 
 		s.mu.Lock()
@@ -3596,6 +3613,9 @@ func (s *Server) removeClient(c *client) {
 		if updateProtoInfoCount {
 			s.cproto--
 		}
+		if proxyKey != _EMPTY_ {
+			s.removeProxiedConn(proxyKey, cid)
+		}
 		s.mu.Unlock()
 	case ROUTER:
 		s.removeRoute(c)
@@ -3606,6 +3626,18 @@ func (s *Server) removeClient(c *client) {
 	}
 }
 
+// Remove the connection with id `cid` from the map of connections
+// under the public key `key` of the trusted proxies.
+//
+// Server lock must be held on entry.
+func (s *Server) removeProxiedConn(key string, cid uint64) {
+	conns := s.proxiedConns[key]
+	delete(conns, cid)
+	if len(conns) == 0 {
+		delete(s.proxiedConns, key)
+	}
+}
+
 func (s *Server) removeFromTempClients(cid uint64) {
 	s.grMu.Lock()
 	delete(s.grTmpClients, cid)

test/client_auth_test.go

@@ -14,13 +14,16 @@
 package test
 
 import (
+	"encoding/base64"
 	"fmt"
+	"net"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
 )
 
 func TestMultipleUserAuth(t *testing.T) {
@@ -156,3 +159,357 @@ func TestClientConnectInfo(t *testing.T) {
 		})
 	}
 }
+
+func TestProxyKeyVerification(t *testing.T) {
+	u1, _ := nkeys.CreateUser()
+	u1Pub, _ := u1.PublicKey()
+
+	u2, _ := nkeys.CreateUser()
+	u2Pub, _ := u2.PublicKey()
+
+	u3, _ := nkeys.CreateUser()
+	u3Pub, _ := u3.PublicKey()
+
+	tmpl := `
+		listen: "127.0.0.1:-1"
+		proxies {
+			trusted [
+				{key: %q}
+				{key: %q}
+			]
+		}
+		leafnodes {
+			listen: "127.0.0.1:-1"
+		}
+	`
+	conf := createConfFile(t, fmt.Appendf(nil, tmpl, u1Pub, u2Pub))
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	varz, err := s.Varz(nil)
+	if err != nil {
+		t.Fatalf("Error getting varz: %v", err)
+	}
+	vproxies := varz.Proxies
+	if vproxies == nil {
+		t.Fatal("Expected proxies to be not nil")
+	}
+	trusted := vproxies.Trusted
+	if len(trusted) != 2 {
+		t.Fatalf("Expected 2 trusted proxies, got %+v", trusted)
+	}
+	if k := trusted[0].Key; k != u1Pub {
+		t.Fatalf("Expected key to be %q, got %q", u1Pub, k)
+	}
+	if k := trusted[1].Key; k != u2Pub {
+		t.Fatalf("Expected key to be %q, got %q", u2Pub, k)
+	}
+
+	var currentCID uint64
+	connect := func(t *testing.T, kp nkeys.KeyPair, bsig bool, port int, isClient bool, user string) (expectFun, net.Conn) {
+		c := createClientConn(t, "127.0.0.1", port)
+		expect := expectCommand(t, c)
+		info := checkInfoMsg(t, c)
+		currentCID = info.CID
+		if info.Nonce == "" {
+			c.Close()
+			t.Fatalf("Expected nonce from INFO, got %+v", info)
+		}
+		if info.JSApiLevel != server.JSApiLevel {
+			c.Close()
+			t.Fatalf("Expected JSApiLevel to be set to %v, got %v", server.JSApiLevel, info.JSApiLevel)
+		}
+		sig, err := kp.Sign([]byte(info.Nonce))
+		if err != nil {
+			c.Close()
+			t.Fatalf("Error signing nonce %q: %v", info.Nonce, err)
+		}
+		encodedSig := make([]byte, base64.RawURLEncoding.EncodedLen(len(sig)))
+		base64.RawURLEncoding.Encode(encodedSig, sig)
+		if bsig {
+			encodedSig[0] = '*'
+		}
+		sendProto(t, c,
+			fmt.Sprintf("CONNECT {\"pedantic\":false,\"verbose\":false,\"user\":%q,\"pass\":\"pwd\",\"proxy_sig\":\"%s\"}\r\n",
+				user, string(encodedSig)))
+		if isClient {
+			sendProto(t, c, "PING\r\n")
+		}
+		return expect, c
+	}
+
+	checkClosedState := func(t *testing.T, cid uint64, reason server.ClosedState) {
+		// Closed connection getting in the closed state is done in a go routine
+		// so make sure we wait for it.
+		checkFor(t, time.Second, 100*time.Millisecond, func() error {
+			connz, err := s.Connz(&server.ConnzOptions{
+				CID:   cid,
+				State: server.ConnClosed,
+			})
+			if err != nil {
+				return fmt.Errorf("Error getting connz for closed connection %v: %v", cid, err)
+			}
+			if len(connz.Conns) != 1 {
+				return fmt.Errorf("Expected 1 connection, got %v", connz.Conns)
+			}
+			conn := connz.Conns[0]
+			if conn.Reason != reason.String() {
+				return fmt.Errorf("Expected reason %q, got %q", reason.String(), conn.Reason)
+			}
+			return nil
+		})
+	}
+
+	for _, test := range []struct {
+		name string
+		kp   nkeys.KeyPair
+		bsig bool
+		ok   bool
+	}{
+		{"key present first", u1, false, true},
+		{"key present last", u2, false, true},
+		{"key not present", u3, false, false},
+		{"bad signature", u2, true, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			for _, subTest := range []struct {
+				name     string
+				port     int
+				isClient bool
+			}{
+				{"client", o.Port, true},
+				{"leafnodes", o.LeafNode.Port, false},
+			} {
+				t.Run(subTest.name, func(t *testing.T) {
+					expect, c := connect(t, test.kp, test.bsig, subTest.port, subTest.isClient, "user")
+					defer c.Close()
+					if test.ok {
+						var p *server.ProxyInfo
+						if subTest.isClient {
+							expect(pongRe)
+
+							connz, err := s.Connz(nil)
+							if err != nil {
+								t.Fatalf("Error getting connz: %v", err)
+							}
+							if n := len(connz.Conns); n != 1 {
+								t.Fatalf("Expected 1 ConnInfo, got %v", n)
+							}
+							ci := connz.Conns[0]
+							p = ci.Proxy
+						} else {
+							expect(infoRe)
+
+							leafz, err := s.Leafz(nil)
+							if err != nil {
+								t.Fatalf("Error getting leafz: %v", err)
+							}
+							if n := len(leafz.Leafs); n != 1 {
+								t.Fatalf("Expected 1 LeafInfo, got %v", n)
+							}
+							li := leafz.Leafs[0]
+							if li.Proxy == nil {
+								t.Fatalf("Required Proxy to be present, was not: %+v", li)
+							}
+							p = li.Proxy
+						}
+						if p == nil {
+							t.Fatalf("Required Proxy to be present, was not")
+						}
+						pub, _ := test.kp.PublicKey()
+						if p.Key != pub {
+							t.Fatalf("Expected public key to be %q, got %q", pub, p.Key)
+						}
+					} else {
+						expect(errRe)
+						checkClosedState(t, currentCID, server.ProxyNotTrusted)
+					}
+				})
+			}
+		})
+	}
+
+	// Create a connection using u1Pub and we will config reload and remove
+	// that key from the list. The client should get disconnected.
+	expectc, c := connect(t, u1, false, o.Port, true, "user")
+	defer c.Close()
+	expectc(pongRe)
+	// Capture this connection CID before the next connect.
+	cid1 := currentCID
+
+	expectl, l := connect(t, u1, false, o.LeafNode.Port, false, "user")
+	defer l.Close()
+	expectl(infoRe)
+	cid2 := currentCID
+	checkLeafNodeConnected(t, s)
+
+	os.WriteFile(conf, fmt.Appendf(nil, tmpl, u3Pub, u2Pub), 0660)
+	if err := s.Reload(); err != nil {
+		t.Fatalf("Reload failed: %v", err)
+	}
+	// Connections should get disconnected.
+	// We need to consume what is sent by the server, but for leaf we may
+	// get some LS+, etc... so just consumer until we get the io.EOF
+	readAll := func(c net.Conn) {
+		for {
+			var buf [1024]byte
+			if _, err := c.Read(buf[:]); err != nil {
+				break
+			}
+		}
+	}
+	readAll(c)
+	expectDisconnect(t, c)
+	readAll(l)
+	expectDisconnect(t, l)
+	c.Close()
+	l.Close()
+
+	// Now check that the connection with CID `cid` was closed because the
+	// proxy is (no longer) trusted.
+	checkClosedState(t, cid1, server.ProxyNotTrusted)
+	checkClosedState(t, cid2, server.ProxyNotTrusted)
+
+	varz, err = s.Varz(nil)
+	if err != nil {
+		t.Fatalf("Error getting varz: %v", err)
+	}
+	vproxies = varz.Proxies
+	if vproxies == nil {
+		t.Fatal("Expected proxies to be not nil")
+	}
+	trusted = vproxies.Trusted
+	if len(trusted) != 2 {
+		t.Fatalf("Expected 2 trusted proxies, got %+v", trusted)
+	}
+	if k := trusted[0].Key; k != u3Pub {
+		t.Fatalf("Expected key to be %q, got %q", u3Pub, k)
+	}
+	if k := trusted[1].Key; k != u2Pub {
+		t.Fatalf("Expected key to be %q, got %q", u2Pub, k)
+	}
+
+	// And we should be able to connect using u3 key pair.
+	expectc, c = connect(t, u3, false, o.Port, true, "user")
+	expectc(pongRe)
+	c.Close()
+	expectl, l = connect(t, u3, false, o.LeafNode.Port, false, "user")
+	expectl(infoRe)
+	l.Close()
+
+	// But now using u1 key pair should fail.
+	expectc, c = connect(t, u1, false, o.Port, true, "user")
+	expectc(errRe)
+	c.Close()
+	expectl, l = connect(t, u1, false, o.LeafNode.Port, false, "user")
+	expectl(errRe)
+	l.Close()
+
+	s.Shutdown()
+
+	// Now check that if no trusted proxy are provided, a proxy connection is
+	// accepted. Use leafnodes since we always send NONCE for this type of connection.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	for _, test := range []struct {
+		name string
+		bsig bool
+	}{
+		{"no trusted proxy and correct signature", false},
+		{"no trusted proxy and invalid signature", true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			expect, c := connect(t, u1, test.bsig, o.LeafNode.Port, false, "user")
+			defer c.Close()
+			// Does not matter if bad signature, since there is no trusted keys
+			// in the server, we accept without signature check.
+			expect(infoRe)
+		})
+	}
+
+	s.Shutdown()
+
+	// Finally, check that if there are no configured trusted proxies
+	// and a user requires a proxy connection, then the connection will fail.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				users [
+					{user: u1, password: pwd}
+					{user: u2, password: pwd, proxy_required: true}
+				]
+			}
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	for _, test := range []struct {
+		name string
+		user string
+		ok   bool
+	}{
+		{"no trusted proxy configured no proxy required", "u1", true},
+		{"no trusted proxy configured and proxy required", "u2", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			// We can use any key pair here, it is not important.
+			expect, c := connect(t, u1, false, o.LeafNode.Port, false, test.user)
+			defer c.Close()
+			if test.ok {
+				expect(infoRe)
+			} else {
+				expect(errRe)
+				// Check that the reason is "proxy required"
+				checkClosedState(t, currentCID, server.ProxyRequired)
+			}
+		})
+	}
+
+	// Same test, but the "proxy_required" is at the authorization{} to-level,
+	// so any user should fail.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				users [
+					{user: u1, password: pwd}
+					{user: u2, password: pwd}
+				]
+				proxy_required: true
+			}
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	t.Run("all users required to use trusted proxy", func(t *testing.T) {
+		for _, user := range []string{"u1", "u2"} {
+			// We can use any key pair here, it is not important.
+			expect, c := connect(t, u1, false, o.LeafNode.Port, false, user)
+			defer c.Close()
+			expect(errRe)
+			// Check that the reason is "proxy required"
+			checkClosedState(t, currentCID, server.ProxyRequired)
+		}
+	})
+
+	varz, err = s.Varz(nil)
+	if err != nil {
+		t.Fatalf("Error getting varz: %v", err)
+	}
+	vproxies = varz.Proxies
+	if vproxies != nil {
+		t.Fatalf("Expected proxies to be nil, got %+v", vproxies)
+	}
+}