30189872764

Committed 24 Jul 2026 09:07AM UTC coverage: 76.546% (-0.9%) from 77.402%

Build # 30189872764

Build Type

push

github

Committed by

web-flow

Commit Message

[FIXED] MQTT: QoS2 message released on a resumed session lost its QoS and PI (#8414)

This came up while testing with
https://github.com/ConnectEverything/mqtt-test/pull/4

Two root causes. The delivery callbacks and the retain handling read the
publishing connection's c.mqtt.pp — the last PUBLISH parsed — which is
stale for a PUBREL- or will-initiated delivery, so a QoS2 message
released after a reconnect was delivered as QoS0 with no packet
identifier. And the staged copy in $MQTT_qos2in never persisted the
retain flag, so it could not be recovered on another connection at all.
Same-connection flows worked only while the scratch happened to still
hold the right packet. Because of the interplay, the two must be fixed
together.

The delivered message now becomes the client's current publish for the
duration of the broadcast (swapped into the readLoop-owned c.mqtt.pp,
restored by the same defer as the pubargs); wills build their own
mqttPublish and use the common path instead of writing into the scratch
by hand. The staged copy
persists the retain flag in a second, flags byte of the Nmqtt-Pub header
value, restored when the PUBREL rebuilds the publish; older servers read
only the first (QoS) byte, and messages without a flags byte read as not
retained.

Retained QoS2 messages now survive an in-flight reconnect: a PUBLISH
staged on one connection and released by the PUBREL on the next stores
the retained message, where before the retain bit died with the first
connection's parser state. This matches Spec [MQTT-4.3.3] Method A as
implemented by mosquitto, which likewise persists the retain flag with
the staged message and applies it at release, together with the
delivery.

Added regression tests: a PUBREL on a resumed session delivers at the
subscription's QoS exactly once, a retransmitted PUBREL is acknowledged
without redelivering, and the retain flag survives the staging round
trip.

Coverage Stats

72417 of 94606 relevant lines covered (76.55%)

491309.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.41

/server/leafnode.go

// Copyright 2019-2026 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 (
        "bufio"
        "bytes"
        "crypto/tls"
        "encoding/base64"
        "encoding/json"
        "fmt"
        "io"
        "math/rand"
        "net"
        "net/http"
        "net/url"
        "os"
        "path"
        "regexp"
        "runtime"
        "strconv"
        "strings"
        "sync"
        "sync/atomic"
        "time"

        "github.com/klauspost/compress/s2"
        "github.com/nats-io/jwt/v2"
        "github.com/nats-io/nkeys"
        "github.com/nats-io/nuid"
)

const (
        // Warning when user configures leafnode TLS insecure
        leafnodeTLSInsecureWarning = "TLS certificate chain and hostname of solicited leafnodes will not be verified. DO NOT USE IN PRODUCTION!"

        // When a loop is detected, delay the reconnect of solicited connection.
        leafNodeReconnectDelayAfterLoopDetected = 30 * time.Second

        // When a server receives a message causing a permission violation, the
        // connection is closed and it won't attempt to reconnect for that long.
        leafNodeReconnectAfterPermViolation = 30 * time.Second

        // When we have the same cluster name as the hub.
        leafNodeReconnectDelayAfterClusterNameSame = 30 * time.Second

        // Prefix for loop detection subject
        leafNodeLoopDetectionSubjectPrefix = "$LDS."

        // Path added to URL to indicate to WS server that the connection is a
        // LEAF connection as opposed to a CLIENT.
        leafNodeWSPath = "/leafnode"

        // When a soliciting leafnode is rejected because it does not meet the
        // configured minimum version, delay the next reconnect attempt by this long.
        leafNodeMinVersionReconnectDelay = 5 * time.Second
)

type leaf struct {
        // We have any auth stuff here for solicited connections.
        remote *leafNodeCfg
        // isSpoke tells us what role we are playing.
        // Used when we receive a connection but otherside tells us they are a hub.
        isSpoke bool
        // remoteCluster is when we are a hub but the spoke leafnode is part of a cluster.
        remoteCluster string
        // remoteServer holds onto the remote server's name or ID.
        remoteServer string
        // domain name of remote server
        remoteDomain string
        // account name of remote server
        remoteAccName string
        // Whether or not we want to propagate east-west interest from other LNs.
        isolated bool
        // Used to suppress sub and unsub interest. Same as routes but our audience
        // here is tied to this leaf node. This will hold all subscriptions except this
        // leaf nodes. This represents all the interest we want to send to the other side.
        smap map[string]int32
        // This map will contain all the subscriptions that have been added to the smap
        // during initLeafNodeSmapAndSendSubs. It is short lived and is there to avoid
        // race between processing of a sub where sub is added to account sublist but
        // updateSmap has not be called on that "thread", while in the LN readloop,
        // when processing CONNECT, initLeafNodeSmapAndSendSubs is invoked and add
        // this subscription to smap. When processing of the sub then calls updateSmap,
        // we would add it a second time in the smap causing later unsub to suppress the LS-.
        tsub  map[*subscription]struct{}
        tsubt *time.Timer
        // Selected compression mode, which may be different from the server configured mode.
        compression string
        // This is for GW map replies.
        gwSub *subscription
}

// Used for remote (solicited) leafnodes.
type leafNodeCfg struct {
        sync.RWMutex
        *RemoteLeafOpts
        urls           []*url.URL
        curURL         *url.URL
        tlsName        string
        username       string
        password       string
        perms          *Permissions
        connDelay      time.Duration // Delay before a connect, could be used while detecting loop condition, etc..
        jsMigrateTimer *time.Timer
        quitCh         chan struct{}
        removed        bool
        connInProgress bool
}

// Check to see if this is a solicited leafnode. We do special processing for solicited.
func (c *client) isSolicitedLeafNode() bool {
        return c.kind == LEAF && c.leaf != nil && c.leaf.remote != nil
}

// Returns true if this is a solicited leafnode and is not configured to be treated as a hub or a receiving
// connection leafnode where the otherside has declared itself to be the hub.
func (c *client) isSpokeLeafNode() bool {
        return c.kind == LEAF && c.leaf != nil && c.leaf.isSpoke
}

func (c *client) isHubLeafNode() bool {
        return c.kind == LEAF && c.leaf != nil && !c.leaf.isSpoke
}

func (c *client) isIsolatedLeafNode() bool {
        // TODO(nat): In future we may want to pass in and consider an isolation
        // group name here, which the hub and/or leaf could provide, so that we
        // can isolate away certain LNs but not others on an opt-in basis. For
        // now we will just isolate all LN interest until then.
        return c.kind == LEAF && c.leaf != nil && c.leaf.isolated
}

// This will spin up go routines to solicit the remote leaf node connections.
func (s *Server) solicitLeafNodeRemotes(remotes []*RemoteLeafOpts) {
        sysAccName := _EMPTY_
        sAcc := s.SystemAccount()
        if sAcc != nil {
                sysAccName = sAcc.Name
        }
        addRemote := func(r *RemoteLeafOpts, isSysAccRemote bool) *leafNodeCfg {
                s.mu.Lock()
                remote := newLeafNodeCfg(r)
                creds := remote.Credentials
                accName := remote.LocalAccount
                if s.leafRemoteCfgs == nil {
                        s.leafRemoteCfgs = make(map[*leafNodeCfg]struct{})
                }
                s.leafRemoteCfgs[remote] = struct{}{}
                // Print notice if
                if isSysAccRemote {
                        if len(remote.DenyExports) > 0 {
                                s.Noticef("Remote for System Account uses restricted export permissions")
                        }
                        if len(remote.DenyImports) > 0 {
                                s.Noticef("Remote for System Account uses restricted import permissions")
                        }
                }
                s.mu.Unlock()
                if creds != _EMPTY_ {
                        contents, err := os.ReadFile(creds)
                        defer wipeSlice(contents)
                        if err != nil {
                                s.Errorf("Error reading LeafNode Remote Credentials file %q: %v", creds, err)
                        } else if items := credsRe.FindAllSubmatch(contents, -1); len(items) < 2 {
                                s.Errorf("LeafNode Remote Credentials file %q malformed", creds)
                        } else if _, err := nkeys.FromSeed(items[1][1]); err != nil {
                                s.Errorf("LeafNode Remote Credentials file %q has malformed seed", creds)
                        } else if uc, err := jwt.DecodeUserClaims(string(items[0][1])); err != nil {
                                s.Errorf("LeafNode Remote Credentials file %q has malformed user jwt", creds)
                        } else if isSysAccRemote {
                                if !uc.Permissions.Pub.Empty() || !uc.Permissions.Sub.Empty() || uc.Permissions.Resp != nil {
                                        s.Noticef("LeafNode Remote for System Account uses credentials file %q with restricted permissions", creds)
                                }
                        } else {
                                if !uc.Permissions.Pub.Empty() || !uc.Permissions.Sub.Empty() || uc.Permissions.Resp != nil {
                                        s.Noticef("LeafNode Remote for Account %s uses credentials file %q with restricted permissions", accName, creds)
                                }
                        }
                }
                return remote
        }
        for _, r := range remotes {
                // We need to call this, even if the leaf is disabled. This is so that
                // the number of internal configuration matches the options' remote leaf
                // configuration required for configuration reload.
                remote := addRemote(r, r.LocalAccount == sysAccName)
                if !r.Disabled {
                        s.connectToRemoteLeafNodeAsynchronously(remote, true)
                }
        }
}

// Ensure that leafnode is properly configured.
func validateLeafNode(o *Options) error {
        if err := validateLeafNodeAuthOptions(o); err != nil {
                return err
        }

        if len(o.LeafNode.Remotes) > 0 {
                names := make(map[string]struct{})
                // Check for duplicate remotes, also, users can bind to any local account,
                // if its empty we will assume the $G account.
                for _, r := range o.LeafNode.Remotes {
                        if r.LocalAccount == _EMPTY_ {
                                r.LocalAccount = globalAccountName
                        }
                        rn := r.name()
                        if _, dup := names[rn]; dup {
                                return fmt.Errorf("duplicate remote %s", r.safeName())
                        }
                        names[rn] = struct{}{}
                }
        }

        // In local config mode, check that leafnode configuration refers to accounts that exist.
        if len(o.TrustedOperators) == 0 {
                accNames := map[string]struct{}{}
                for _, a := range o.Accounts {
                        accNames[a.Name] = struct{}{}
                }
                // global account is always created
                accNames[DEFAULT_GLOBAL_ACCOUNT] = struct{}{}
                // in the context of leaf nodes, empty account means global account
                accNames[_EMPTY_] = struct{}{}
                // system account either exists or, if not disabled, will be created
                if o.SystemAccount == _EMPTY_ && !o.NoSystemAccount {
                        accNames[DEFAULT_SYSTEM_ACCOUNT] = struct{}{}
                }
                checkAccountExists := func(accName string, cfgType string) error {
                        if _, ok := accNames[accName]; !ok {
                                return fmt.Errorf("cannot find local account %q specified in leafnode %s", accName, cfgType)
                        }
                        return nil
                }
                if err := checkAccountExists(o.LeafNode.Account, "authorization"); err != nil {
                        return err
                }
                for _, lu := range o.LeafNode.Users {
                        if lu.Account == nil { // means global account
                                continue
                        }
                        if err := checkAccountExists(lu.Account.Name, "authorization"); err != nil {
                                return err
                        }
                }
                for _, r := range o.LeafNode.Remotes {
                        if err := checkAccountExists(r.LocalAccount, "remote"); err != nil {
                                return err
                        }
                }
        } else {
                if len(o.LeafNode.Users) != 0 {
                        return fmt.Errorf("operator mode does not allow specifying users in leafnode config")
                }
                for _, r := range o.LeafNode.Remotes {
                        if !nkeys.IsValidPublicAccountKey(r.LocalAccount) {
                                return fmt.Errorf(
                                        "operator mode requires account nkeys in remotes. " +
                                                "Please add an `account` key to each remote in your `leafnodes` section, to assign it to an account. " +
                                                "Each account value should be a 56 character public key, starting with the letter 'A'")
                        }
                }
                if o.LeafNode.Port != 0 && o.LeafNode.Account != "" && !nkeys.IsValidPublicAccountKey(o.LeafNode.Account) {
                        return fmt.Errorf("operator mode and non account nkeys are incompatible")
                }
        }

        // Validate compression settings
        if o.LeafNode.Compression.Mode != _EMPTY_ {
                if err := validateAndNormalizeCompressionOption(&o.LeafNode.Compression, CompressionS2Auto); err != nil {
                        return err
                }
        }

        // If a remote has a websocket scheme, all need to have it.
        for _, rcfg := range o.LeafNode.Remotes {
                // Validate proxy configuration
                if _, err := validateLeafNodeProxyOptions(rcfg); err != nil {
                        return err
                }

                if len(rcfg.URLs) >= 2 {
                        firstIsWS, ok := isWSURL(rcfg.URLs[0]), true
                        for i := 1; i < len(rcfg.URLs); i++ {
                                u := rcfg.URLs[i]
                                if isWS := isWSURL(u); isWS && !firstIsWS || !isWS && firstIsWS {
                                        ok = false
                                        break
                                }
                        }
                        if !ok {
                                return fmt.Errorf("remote leaf node configuration cannot have a mix of websocket and non-websocket urls: %q", redactURLList(rcfg.URLs))
                        }
                }
                if !wsAllowedFIPS() {
                        for _, u := range rcfg.URLs {
                                if isWSURL(u) {
                                        return fmt.Errorf("remote leaf node URL %q cannot be used in FIPS-140 mode when built with this Go version, use Go 1.26 or later", redactURLString(u.String()))
                                }
                        }
                }
                // Validate compression settings
                if rcfg.Compression.Mode != _EMPTY_ {
                        if err := validateAndNormalizeCompressionOption(&rcfg.Compression, CompressionS2Auto); err != nil {
                                return err
                        }
                }
        }

        if o.LeafNode.Port == 0 {
                return nil
        }

        // If MinVersion is defined, check that it is valid.
        if mv := o.LeafNode.MinVersion; mv != _EMPTY_ {
                if err := checkLeafMinVersionConfig(mv); err != nil {
                        return err
                }
        }

        // The checks below will be done only when detecting that we are configured
        // with gateways. So if an option validation needs to be done regardless,
        // it MUST be done before this point!

        if o.Gateway.Name == _EMPTY_ && o.Gateway.Port == 0 {
                return nil
        }
        // If we are here we have both leaf nodes and gateways defined, make sure there
        // is a system account defined.
        if o.SystemAccount == _EMPTY_ {
                return fmt.Errorf("leaf nodes and gateways (both being defined) require a system account to also be configured")
        }
        if err := validatePinnedCerts(o.LeafNode.TLSPinnedCerts); err != nil {
                return fmt.Errorf("leafnode: %v", err)
        }
        return nil
}

func checkLeafMinVersionConfig(mv string) error {
        if ok, err := versionAtLeastCheckError(mv, 2, 8, 0); !ok || err != nil {
                if err != nil {
                        return fmt.Errorf("invalid leafnode's minimum version: %v", err)
                } else {
                        return fmt.Errorf("the minimum version should be at least 2.8.0")
                }
        }
        return nil
}

// Used to validate user names in LeafNode configuration.
// - rejects mix of single and multiple users.
// - rejects duplicate user names.
func validateLeafNodeAuthOptions(o *Options) error {
        if len(o.LeafNode.Users) == 0 {
                return nil
        }
        if o.LeafNode.Username != _EMPTY_ {
                return fmt.Errorf("can not have a single user/pass and a users array")
        }
        if o.LeafNode.Nkey != _EMPTY_ {
                return fmt.Errorf("can not have a single nkey and a users array")
        }
        users := map[string]struct{}{}
        for _, u := range o.LeafNode.Users {
                if _, exists := users[u.Username]; exists {
                        return fmt.Errorf("duplicate user %q detected in leafnode authorization", u.Username)
                }
                users[u.Username] = struct{}{}
        }
        return nil
}

func validateLeafNodeProxyOptions(remote *RemoteLeafOpts) ([]string, error) {
        var warnings []string

        if remote.Proxy.URL == _EMPTY_ {
                return warnings, nil
        }

        proxyURL, err := url.Parse(remote.Proxy.URL)
        if err != nil {
                return warnings, fmt.Errorf("invalid proxy URL: %v", err)
        }

        if proxyURL.Scheme != "http" && proxyURL.Scheme != "https" {
                return warnings, fmt.Errorf("proxy URL scheme must be http or https, got: %s", proxyURL.Scheme)
        }

        if proxyURL.Host == _EMPTY_ {
                return warnings, fmt.Errorf("proxy URL must specify a host")
        }

        if remote.Proxy.Timeout < 0 {
                return warnings, fmt.Errorf("proxy timeout must be >= 0")
        }

        if (remote.Proxy.Username == _EMPTY_) != (remote.Proxy.Password == _EMPTY_) {
                return warnings, fmt.Errorf("proxy username and password must both be specified or both be empty")
        }

        if len(remote.URLs) > 0 {
                hasWebSocketURL := false
                hasNonWebSocketURL := false

                for _, remoteURL := range remote.URLs {
                        if remoteURL.Scheme == wsSchemePrefix || remoteURL.Scheme == wsSchemePrefixTLS {
                                hasWebSocketURL = true
                                if (remoteURL.Scheme == wsSchemePrefixTLS) &&
                                        remote.TLSConfig == nil && !remote.TLS {
                                        return warnings, fmt.Errorf("proxy is configured but remote URL %s requires TLS and no TLS configuration is provided. When using proxy with TLS endpoints, ensure TLS is properly configured for the leafnode remote", remoteURL.String())
                                }
                        } else {
                                hasNonWebSocketURL = true
                        }
                }

                if !hasWebSocketURL {
                        warnings = append(warnings, "proxy configuration will be ignored: proxy settings only apply to WebSocket connections (ws:// or wss://), but all configured URLs use TCP connections (nats://)")
                } else if hasNonWebSocketURL {
                        warnings = append(warnings, "proxy configuration will only be used for WebSocket URLs: proxy settings do not apply to TCP connections (nats://)")
                }
        }

        return warnings, nil
}

// Wait for the configured reconnect interval before attempting to connect
// again to the remote leafnode.
func (s *Server) reConnectToRemoteLeafNode(remote *leafNodeCfg) {
        clearInProgress := true
        defer func() {
                s.grWG.Done()
                if clearInProgress {
                        remote.setConnectInProgress(false)
                }
        }()
        delay := s.getOpts().LeafNode.ReconnectInterval
        select {
        case <-time.After(delay):
        case <-remote.quitCh:
                return
        case <-s.quitCh:
                return
        }
        clearInProgress = !connectToRemoteLeafNode(s, remote, false)
}

// Creates a leafNodeCfg object that wraps the RemoteLeafOpts.
func newLeafNodeCfg(remote *RemoteLeafOpts) *leafNodeCfg {
        cfg := &leafNodeCfg{
                RemoteLeafOpts: remote,
                urls:           make([]*url.URL, 0, len(remote.URLs)),
                quitCh:         make(chan struct{}, 1),
        }
        if len(remote.DenyExports) > 0 || len(remote.DenyImports) > 0 {
                perms := &Permissions{}
                if len(remote.DenyExports) > 0 {
                        perms.Publish = &SubjectPermission{Deny: remote.DenyExports}
                }
                if len(remote.DenyImports) > 0 {
                        perms.Subscribe = &SubjectPermission{Deny: remote.DenyImports}
                }
                cfg.perms = perms
        }
        // Start with the one that is configured. We will add to this
        // array when receiving async leafnode INFOs.
        cfg.urls = append(cfg.urls, cfg.URLs...)
        // If allowed to randomize, do it on our copy of URLs
        if !remote.NoRandomize {
                rand.Shuffle(len(cfg.urls), func(i, j int) {
                        cfg.urls[i], cfg.urls[j] = cfg.urls[j], cfg.urls[i]
                })
        }
        // If we are TLS make sure we save off a proper servername if possible.
        // Do same for user/password since we may need them to connect to
        // a bare URL that we get from INFO protocol.
        for _, u := range cfg.urls {
                cfg.saveTLSHostname(u)
                cfg.saveUserPassword(u)
                // If the url(s) have the "wss://" scheme, and we don't have a TLS
                // config, mark that we should be using TLS anyway.
                if !cfg.TLS && isWSSURL(u) {
                        cfg.TLS = true
                }
        }
        return cfg
}

// Notifies the quit channel without blocking.
// No lock is needed to invoke this function.
func (cfg *leafNodeCfg) notifyQuitChannel() {
        select {
        case cfg.quitCh <- struct{}{}:
        default:
        }
}

// Sets the connect-in-progress status for this remote leaf configuration.
func (cfg *leafNodeCfg) setConnectInProgress(inProgress bool) {
        cfg.Lock()
        defer cfg.Unlock()
        // In both cases we want to drain the "quit" channel.
        select {
        case <-cfg.quitCh:
        default:
        }
        cfg.connInProgress = inProgress
}

// Returns `true` if this remote is in the middle of a connect, `false` otherwise.
func (cfg *leafNodeCfg) isConnectInProgress() bool {
        cfg.RLock()
        defer cfg.RUnlock()
        return cfg.connInProgress
}

// Mark that this remote is being removed from the configuration.
func (cfg *leafNodeCfg) markAsRemoved() {
        cfg.Lock()
        defer cfg.Unlock()
        // This function should be invoked only once, but protect.
        if cfg.removed {
                return
        }
        cfg.removed = true
        cfg.notifyQuitChannel()
}

// Returns false if it has been disabled or removed.
func (cfg *leafNodeCfg) stillValid() bool {
        cfg.RLock()
        defer cfg.RUnlock()
        return !cfg.Disabled && !cfg.removed
}

// Will pick an URL from the list of available URLs.
func (cfg *leafNodeCfg) pickNextURL() *url.URL {
        cfg.Lock()
        defer cfg.Unlock()
        // If the current URL is the first in the list and we have more than
        // one URL, then move that one to end of the list.
        if cfg.curURL != nil && len(cfg.urls) > 1 && urlsAreEqual(cfg.curURL, cfg.urls[0]) {
                first := cfg.urls[0]
                copy(cfg.urls, cfg.urls[1:])
                cfg.urls[len(cfg.urls)-1] = first
        }
        cfg.curURL = cfg.urls[0]
        return cfg.curURL
}

// Returns the current URL
func (cfg *leafNodeCfg) getCurrentURL() *url.URL {
        cfg.RLock()
        defer cfg.RUnlock()
        return cfg.curURL
}

// Returns how long the server should wait before attempting
// to solicit a remote leafnode connection.
func (cfg *leafNodeCfg) getConnectDelay() time.Duration {
        cfg.RLock()
        delay := cfg.connDelay
        cfg.RUnlock()
        return delay
}

// Sets the connect delay.
func (cfg *leafNodeCfg) setConnectDelay(delay time.Duration) {
        cfg.Lock()
        cfg.connDelay = delay
        cfg.Unlock()
}

// Ensure that non-exported options (used in tests) have
// been properly set.
func (s *Server) setLeafNodeNonExportedOptions() {
        opts := s.getOpts()
        s.leafNodeOpts.dialTimeout = opts.LeafNode.dialTimeout
        if s.leafNodeOpts.dialTimeout == 0 {
                // Use same timeouts as routes for now.
                s.leafNodeOpts.dialTimeout = DEFAULT_ROUTE_DIAL
        }
        s.leafNodeOpts.resolver = opts.LeafNode.resolver
        if s.leafNodeOpts.resolver == nil {
                s.leafNodeOpts.resolver = net.DefaultResolver
        }
}

const sharedSysAccDelay = 250 * time.Millisecond

// establishHTTPProxyTunnel establishes an HTTP CONNECT tunnel through a proxy server
func establishHTTPProxyTunnel(proxyURL, targetHost string, timeout time.Duration, username, password string) (net.Conn, error) {
        proxyAddr, err := url.Parse(proxyURL)
        if err != nil {
                // This should not happen since proxy URL is validated during configuration parsing
                return nil, fmt.Errorf("unexpected proxy URL parse error (URL was pre-validated): %v", err)
        }

        // Connect to the proxy server
        conn, err := natsDialTimeout("tcp", proxyAddr.Host, timeout)
        if err != nil {
                return nil, fmt.Errorf("failed to connect to proxy: %v", err)
        }

        // Set deadline for the entire proxy handshake
        if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil {
                conn.Close()
                return nil, fmt.Errorf("failed to set deadline: %v", err)
        }

        req := &http.Request{
                Method: http.MethodConnect,
                URL:    &url.URL{Opaque: targetHost}, // Opaque is required for CONNECT
                Host:   targetHost,
                Header: make(http.Header),
        }

        // Add proxy authentication if provided
        if username != "" && password != "" {
                req.Header.Set("Proxy-Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(username+":"+password)))
        }

        if err := req.Write(conn); err != nil {
                conn.Close()
                return nil, fmt.Errorf("failed to write CONNECT request: %v", err)
        }

        resp, err := http.ReadResponse(bufio.NewReader(conn), req)
        if err != nil {
                conn.Close()
                return nil, fmt.Errorf("failed to read proxy response: %v", err)
        }

        if resp.StatusCode != http.StatusOK {
                resp.Body.Close()
                conn.Close()
                return nil, fmt.Errorf("proxy CONNECT failed: %s", resp.Status)
        }

        // Close the response body
        resp.Body.Close()

        // Clear the deadline now that we've finished the proxy handshake
        if err := conn.SetDeadline(time.Time{}); err != nil {
                conn.Close()
                return nil, fmt.Errorf("failed to clear deadline: %v", err)
        }

        return conn, nil
}

// Connect to a remote leaf node asynchronously (that is, this function will do
// the connect in a go routine).
func (s *Server) connectToRemoteLeafNodeAsynchronously(remote *leafNodeCfg, firstConnect bool) {
        remote.setConnectInProgress(true)
        s.startGoRoutine(func() {
                defer s.grWG.Done()
                if !connectToRemoteLeafNode(s, remote, firstConnect) {
                        remote.setConnectInProgress(false)
                }
        })
}

// Connect to a remote leaf node. Should only be invoked from
// `s.connectToRemoteLeafNodeAsynchronously()` or `s.reConnectToRemoteLeafNode()`.
// Returns `true` if this function invoked `s.createLeafNode()`, false otherwise.
func connectToRemoteLeafNode(s *Server, remote *leafNodeCfg, firstConnect bool) bool {

        if remote == nil || len(remote.URLs) == 0 {
                s.Debugf("Empty remote leafnode definition, nothing to connect")
                return false
        }

        // If this remote is no longer valid by the time we return (e.g. it was
        // removed or disabled through a configuration reload), we will never
        // reconnect, so clear any JetStream observer state. Otherwise, the raft
        // nodes of this account's assets would remain observers.
        defer func() {
                if remote.stillValid() {
                        return
                }
                s.mu.RLock()
                shouldMigrate := remote.JetStreamClusterMigrate
                s.mu.RUnlock()
                if shouldMigrate {
                        s.clearObserverState(remote)
                }
        }()

        opts := s.getOpts()
        reconnectDelay := opts.LeafNode.ReconnectInterval
        s.mu.RLock()
        dialTimeout := s.leafNodeOpts.dialTimeout
        resolver := s.leafNodeOpts.resolver
        var isSysAcc bool
        if s.eventsEnabled() {
                isSysAcc = remote.LocalAccount == s.sys.account.Name
        }
        jetstreamMigrateDelay := remote.JetStreamClusterMigrateDelay
        s.mu.RUnlock()

        // If we are sharing a system account and we are not standalone delay to gather some info prior.
        if firstConnect && isSysAcc && !s.standAloneMode() {
                s.Debugf("Will delay first leafnode connect to shared system account due to clustering")
                remote.setConnectDelay(sharedSysAccDelay)
        }

        if connDelay := remote.getConnectDelay(); connDelay > 0 {
                select {
                case <-time.After(connDelay):
                case <-remote.quitCh:
                        return false
                case <-s.quitCh:
                        return false
                }
                remote.setConnectDelay(0)
        }

        var conn net.Conn

        const connErrFmt = "Error trying to connect as leafnode to remote server %q (attempt %v): %v"

        // Capture proxy configuration once before the loop with proper locking
        remote.RLock()
        proxyURL := remote.Proxy.URL
        proxyUsername := remote.Proxy.Username
        proxyPassword := remote.Proxy.Password
        proxyTimeout := remote.Proxy.Timeout
        remote.RUnlock()

        // Set default proxy timeout if not specified
        if proxyTimeout == 0 {
                proxyTimeout = dialTimeout
        }

        attempts := 0

        // In case the migrate timer was created but not canceled, do it when
        // this function exits. Note that the timer would not be created if
        // `jetstreamMigrateDelay == 0`.
        if jetstreamMigrateDelay > 0 {
                defer remote.cancelMigrateTimer()
        }

        reconnectTimer := time.NewTimer(reconnectDelay)
        reconnectTimer.Stop()
        defer stopAndClearTimer(&reconnectTimer)

        for s.isRunning() && remote.stillValid() {
                rURL := remote.pickNextURL()
                url, err := s.getRandomIP(resolver, rURL.Host, nil)
                if err == nil {
                        var ipStr string
                        if url != rURL.Host {
                                ipStr = fmt.Sprintf(" (%s)", url)
                        }
                        // Some test may want to disable remotes from connecting
                        if s.isLeafConnectDisabled() {
                                s.Debugf("Will not attempt to connect to remote server on %q%s, leafnodes currently disabled", rURL.Host, ipStr)
                                err = ErrLeafNodeDisabled
                        } else {
                                s.Debugf("Trying to connect as leafnode to remote server on %q%s", rURL.Host, ipStr)

                                // Check if proxy is configured
                                if proxyURL != _EMPTY_ {
                                        targetHost := rURL.Host
                                        // If URL doesn't include port, add the default port for the scheme
                                        if rURL.Port() == _EMPTY_ {
                                                defaultPort := "80"
                                                if rURL.Scheme == wsSchemePrefixTLS {
                                                        defaultPort = "443"
                                                }
                                                targetHost = net.JoinHostPort(rURL.Hostname(), defaultPort)
                                        }

                                        conn, err = establishHTTPProxyTunnel(proxyURL, targetHost, proxyTimeout, proxyUsername, proxyPassword)
                                } else {
                                        // Direct connection
                                        conn, err = natsDialTimeout("tcp", url, dialTimeout)
                                }
                        }
                }
                if err != nil {
                        jitter := time.Duration(rand.Int63n(int64(reconnectDelay)))
                        delay := reconnectDelay + jitter
                        attempts++
                        if s.shouldReportConnectErr(firstConnect, attempts) {
                                s.Errorf(connErrFmt, rURL.Host, attempts, err)
                        } else {
                                s.Debugf(connErrFmt, rURL.Host, attempts, err)
                        }
                        remote.Lock()
                        // if we are using a delay to start migrating assets, kick off a migrate timer.
                        if remote.jsMigrateTimer == nil && jetstreamMigrateDelay > 0 {
                                remote.jsMigrateTimer = time.AfterFunc(jetstreamMigrateDelay, func() {
                                        s.checkJetStreamMigrate(remote)
                                })
                        }
                        remote.Unlock()
                        reconnectTimer.Reset(delay)
                        select {
                        case <-s.quitCh:
                                return false
                        case <-remote.quitCh:
                                return false
                        case <-reconnectTimer.C:
                                // Check if we should migrate any JetStream assets immediately while this remote is down.
                                // This will be used if JetStreamClusterMigrateDelay was not set
                                if jetstreamMigrateDelay == 0 {
                                        s.checkJetStreamMigrate(remote)
                                }
                                continue
                        }
                }
                remote.cancelMigrateTimer()
                // We can check here, but really we will have to check again when the server
                // is about to add to the `s.leafs` map later in the process.
                if !remote.stillValid() {
                        conn.Close()
                        return false
                }

                // We have a connection here to a remote server.
                // Go ahead and create our leaf node and return.
                s.createLeafNode(conn, rURL, remote, nil)

                // Clear any observer states if we had them.
                s.clearObserverState(remote)

                return true
        }

        return false
}

func (cfg *leafNodeCfg) cancelMigrateTimer() {
        cfg.Lock()
        stopAndClearTimer(&cfg.jsMigrateTimer)
        cfg.Unlock()
}

// This will clear any observer state such that stream or consumer assets on this server can become leaders again.
func (s *Server) clearObserverState(remote *leafNodeCfg) {
        s.mu.RLock()
        accName := remote.LocalAccount
        s.mu.RUnlock()

        acc, err := s.LookupAccount(accName)
        if err != nil {
                s.Warnf("Error looking up account [%s] checking for JetStream clear observer state on a leafnode", accName)
                return
        }

        acc.jscmMu.Lock()
        defer acc.jscmMu.Unlock()

        // Walk all streams looking for any clustered stream, skip otherwise.
        for _, mset := range acc.streams() {
                node := mset.raftNode()
                if node == nil {
                        // Not R>1
                        continue
                }
                // Check consumers
                for _, o := range mset.getConsumers() {
                        if n := o.raftNode(); n != nil {
                                // Ensure we can become a leader again.
                                n.SetObserver(false)
                        }
                }
                // Ensure we can not become a leader again.
                node.SetObserver(false)
        }
}

// Check to see if we should migrate any assets from this account.
func (s *Server) checkJetStreamMigrate(remote *leafNodeCfg) {
        s.mu.RLock()
        accName, shouldMigrate := remote.LocalAccount, remote.JetStreamClusterMigrate
        s.mu.RUnlock()

        if !shouldMigrate {
                return
        }

        acc, err := s.LookupAccount(accName)
        if err != nil {
                s.Warnf("Error looking up account [%s] checking for JetStream migration on a leafnode", accName)
                return
        }

        acc.jscmMu.Lock()
        defer acc.jscmMu.Unlock()

        // Walk all streams looking for any clustered stream, skip otherwise.
        // If we are the leader force stepdown.
        for _, mset := range acc.streams() {
                node := mset.raftNode()
                if node == nil {
                        // Not R>1
                        continue
                }
                // Collect any consumers
                for _, o := range mset.getConsumers() {
                        if n := o.raftNode(); n != nil {
                                n.StepDown()
                                // Ensure we can not become a leader while in this state.
                                n.SetObserver(true)
                        }
                }
                // Stepdown if this stream was leader.
                node.StepDown()
                // Ensure we can not become a leader while in this state.
                node.SetObserver(true)
        }
}

// Helper for checking.
func (s *Server) isLeafConnectDisabled() bool {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.leafDisableConnect
}

// Save off the tlsName for when we use TLS and mix hostnames and IPs. IPs usually
// come from the server we connect to.
//
// We used to save the name only if there was a TLSConfig or scheme equal to "tls".
// However, this was causing failures for users that did not set the scheme (and
// their remote connections did not have a tls{} block).
// We now save the host name regardless in case the remote returns an INFO indicating
// that TLS is required.
//
// Lock held on entry.
func (cfg *leafNodeCfg) saveTLSHostname(u *url.URL) {
        if cfg.tlsName == _EMPTY_ && net.ParseIP(u.Hostname()) == nil {
                cfg.tlsName = u.Hostname()
        }
}

// Save off the username/password for when we connect using a bare URL
// that we get from the INFO protocol.
//
// Lock held on entry.
func (cfg *leafNodeCfg) saveUserPassword(u *url.URL) {
        if cfg.username == _EMPTY_ && u.User != nil {
                cfg.username = u.User.Username()
                cfg.password, _ = u.User.Password()
        }
}

// This starts the leafnode accept loop in a go routine, unless it
// is detected that the server has already been shutdown.
func (s *Server) startLeafNodeAcceptLoop() {
        // Snapshot server options.
        opts := s.getOpts()

        port := opts.LeafNode.Port
        if port == -1 {
                port = 0
        }

        if s.isShuttingDown() {
                return
        }

        s.mu.Lock()
        hp := net.JoinHostPort(opts.LeafNode.Host, strconv.Itoa(port))
        l, e := natsListen("tcp", hp)
        s.leafNodeListenerErr = e
        if e != nil {
                s.mu.Unlock()
                s.Fatalf("Error listening on leafnode port: %d - %v", opts.LeafNode.Port, e)
                return
        }

        s.Noticef("Listening for leafnode connections on %s",
                net.JoinHostPort(opts.LeafNode.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))

        tlsRequired := opts.LeafNode.TLSConfig != nil
        tlsVerify := tlsRequired && opts.LeafNode.TLSConfig.ClientAuth == tls.RequireAndVerifyClientCert
        // Do not set compression in this Info object, it would possibly cause
        // issues when sending asynchronous INFO to the remote.
        info := Info{
                ID:            s.info.ID,
                Name:          s.info.Name,
                Version:       s.info.Version,
                GitCommit:     gitCommit,
                GoVersion:     runtime.Version(),
                AuthRequired:  true,
                TLSRequired:   tlsRequired,
                TLSVerify:     tlsVerify,
                MaxPayload:    s.info.MaxPayload, // TODO(dlc) - Allow override?
                Headers:       s.supportsHeaders(),
                JetStream:     opts.JetStream,
                Domain:        opts.JetStreamDomain,
                Proto:         s.getServerProto(),
                InfoOnConnect: true,
                JSApiLevel:    JSApiLevel,
        }
        // If we have selected a random port...
        if port == 0 {
                // Write resolved port back to options.
                opts.LeafNode.Port = l.Addr().(*net.TCPAddr).Port
        }

        s.leafNodeInfo = info
        // Possibly override Host/Port and set IP based on Cluster.Advertise
        if err := s.setLeafNodeInfoHostPortAndIP(); err != nil {
                s.Fatalf("Error setting leafnode INFO with LeafNode.Advertise value of %s, err=%v", opts.LeafNode.Advertise, err)
                l.Close()
                s.mu.Unlock()
                return
        }
        s.leafURLsMap[s.leafNodeInfo.IP]++
        s.generateLeafNodeInfoJSON()

        // Setup state that can enable shutdown
        s.leafNodeListener = l

        // As of now, a server that does not have remotes configured would
        // never solicit a connection, so we should not have to warn if
        // InsecureSkipVerify is set in main LeafNodes config (since
        // this TLS setting matters only when soliciting a connection).
        // Still, warn if insecure is set in any of LeafNode block.
        // We need to check remotes, even if tls is not required on accept.
        warn := tlsRequired && opts.LeafNode.TLSConfig.InsecureSkipVerify
        if !warn {
                for _, r := range opts.LeafNode.Remotes {
                        if r.TLSConfig != nil && r.TLSConfig.InsecureSkipVerify {
                                warn = true
                                break
                        }
                }
        }
        if warn {
                s.Warnf(leafnodeTLSInsecureWarning)
        }
        go s.acceptConnections(l, "Leafnode", func(conn net.Conn) { s.createLeafNode(conn, nil, nil, nil) }, nil)
        s.mu.Unlock()
}

// RegEx to match a creds file with user JWT and Seed.
var credsRe = regexp.MustCompile(`\s*(?:(?:[-]{3,}.*[-]{3,}\r?\n)([\w\-.=]+)(?:\r?\n[-]{3,}.*[-]{3,}(\r?\n|\z)))`)

// clusterName is provided as argument to avoid lock ordering issues with the locked client c
// Lock should be held entering here.
func (c *client) sendLeafConnect(clusterName string, headers bool) error {
        // We support basic user/pass and operator based user JWT with signatures.
        cinfo := leafConnectInfo{
                Version:       VERSION,
                ID:            c.srv.info.ID,
                Domain:        c.srv.info.Domain,
                Name:          c.srv.info.Name,
                Hub:           c.leaf.remote.Hub,
                Cluster:       clusterName,
                Headers:       headers,
                JetStream:     c.acc.jetStreamConfigured(),
                DenyPub:       c.leaf.remote.DenyImports,
                Compression:   c.leaf.compression,
                RemoteAccount: c.acc.GetName(),
                Proto:         c.srv.getServerProto(),
                Isolate:       c.leaf.remote.RequestIsolation,
        }

        // If a signature callback is specified, this takes precedence over anything else.
        if cb := c.leaf.remote.SignatureCB; cb != nil {
                nonce := c.nonce
                c.mu.Unlock()
                jwt, sigraw, err := cb(nonce)
                c.mu.Lock()
                if err == nil && c.isClosed() {
                        err = ErrConnectionClosed
                }
                if err != nil {
                        c.Errorf("Error signing the nonce: %v", err)
                        return err
                }
                sig := base64.RawURLEncoding.EncodeToString(sigraw)
                cinfo.JWT, cinfo.Sig = jwt, sig

        } else if creds := c.leaf.remote.Credentials; creds != _EMPTY_ {
                // Check for credentials first, that will take precedence..
                c.Debugf("Authenticating with credentials file %q", c.leaf.remote.Credentials)
                contents, err := os.ReadFile(creds)
                if err != nil {
                        c.Errorf("%v", err)
                        return err
                }
                defer wipeSlice(contents)
                items := credsRe.FindAllSubmatch(contents, -1)
                if len(items) < 2 {
                        c.Errorf("Credentials file malformed")
                        return err
                }
                // First result should be the user JWT.
                // We copy here so that the file containing the seed will be wiped appropriately.
                raw := items[0][1]
                tmp := make([]byte, len(raw))
                copy(tmp, raw)
                // Seed is second item.
                kp, err := nkeys.FromSeed(items[1][1])
                if err != nil {
                        c.Errorf("Credentials file has malformed seed")
                        return err
                }
                // Wipe our key on exit.
                defer kp.Wipe()

                sigraw, _ := kp.Sign(c.nonce)
                sig := base64.RawURLEncoding.EncodeToString(sigraw)
                cinfo.JWT = bytesToString(tmp)
                cinfo.Sig = sig
        } else if nkey := c.leaf.remote.Nkey; nkey != _EMPTY_ {
                kp, err := nkeys.FromSeed([]byte(nkey))
                if err != nil {
                        c.Errorf("Remote nkey has malformed seed")
                        return err
                }
                // Wipe our key on exit.
                defer kp.Wipe()
                sigraw, _ := kp.Sign(c.nonce)
                sig := base64.RawURLEncoding.EncodeToString(sigraw)
                pkey, _ := kp.PublicKey()
                cinfo.Nkey = pkey
                cinfo.Sig = sig
        }
        // In addition, and this is to allow auth callout, set user/password or
        // token if applicable.
        if userInfo := c.leaf.remote.curURL.User; userInfo != nil {
                cinfo.User = userInfo.Username()
                var ok bool
                cinfo.Pass, ok = userInfo.Password()
                // For backward compatibility, if only username is provided, set both
                // Token and User, not just Token.
                if !ok {
                        cinfo.Token = cinfo.User
                }
        } else if c.leaf.remote.username != _EMPTY_ {
                cinfo.User = c.leaf.remote.username
                cinfo.Pass = c.leaf.remote.password
                // For backward compatibility, if only username is provided, set both
                // Token and User, not just Token.
                if cinfo.Pass == _EMPTY_ {
                        cinfo.Token = cinfo.User
                }
        }
        b, err := json.Marshal(cinfo)
        if err != nil {
                c.Errorf("Error marshaling CONNECT to remote leafnode: %v\n", err)
                return err
        }
        // Although this call is made before the writeLoop is created,
        // we don't really need to send in place. The protocol will be
        // sent out by the writeLoop.
        c.enqueueProto([]byte(fmt.Sprintf(ConProto, b)))
        return nil
}

// Makes a deep copy of the LeafNode Info structure.
// The server lock is held on entry.
func (s *Server) copyLeafNodeInfo() *Info {
        clone := s.leafNodeInfo
        // Copy the array of urls.
        if len(s.leafNodeInfo.LeafNodeURLs) > 0 {
                clone.LeafNodeURLs = append([]string(nil), s.leafNodeInfo.LeafNodeURLs...)
        }
        return &clone
}

// Adds a LeafNode URL that we get when a route connects to the Info structure.
// Regenerates the JSON byte array so that it can be sent to LeafNode connections.
// Returns a boolean indicating if the URL was added or not.
// Server lock is held on entry
func (s *Server) addLeafNodeURL(urlStr string) bool {
        if s.leafURLsMap.addUrl(urlStr) {
                s.generateLeafNodeInfoJSON()
                return true
        }
        return false
}

// Removes a LeafNode URL of the route that is disconnecting from the Info structure.
// Regenerates the JSON byte array so that it can be sent to LeafNode connections.
// Returns a boolean indicating if the URL was removed or not.
// Server lock is held on entry.
func (s *Server) removeLeafNodeURL(urlStr string) bool {
        // Don't need to do this if we are removing the route connection because
        // we are shuting down...
        if s.isShuttingDown() {
                return false
        }
        if s.leafURLsMap.removeUrl(urlStr) {
                s.generateLeafNodeInfoJSON()
                return true
        }
        return false
}

// Server lock is held on entry
func (s *Server) generateLeafNodeInfoJSON() {
        s.leafNodeInfo.Cluster = s.cachedClusterName()
        s.leafNodeInfo.LeafNodeURLs = s.leafURLsMap.getAsStringSlice()
        s.leafNodeInfo.WSConnectURLs = s.websocket.connectURLsMap.getAsStringSlice()
        s.leafNodeInfoJSON = generateInfoJSON(&s.leafNodeInfo)
}

// Sends an async INFO protocol so that the connected servers can update
// their list of LeafNode urls.
func (s *Server) sendAsyncLeafNodeInfo() {
        for _, c := range s.leafs {
                c.mu.Lock()
                c.enqueueProto(s.leafNodeInfoJSON)
                c.mu.Unlock()
        }
}

// Called when an inbound leafnode connection is accepted or we create one for a solicited leafnode.
func (s *Server) createLeafNode(conn net.Conn, rURL *url.URL, remote *leafNodeCfg, ws *websocket) *client {
        // Snapshot server options.
        opts := s.getOpts()

        maxPay := int32(opts.MaxPayload)
        maxSubs := int32(opts.MaxSubs)
        // For system, maxSubs of 0 means unlimited, so re-adjust here.
        if maxSubs == 0 {
                maxSubs = -1
        }
        now := time.Now().UTC()

        c := &client{srv: s, nc: conn, kind: LEAF, opts: defaultOpts, mpay: maxPay, msubs: maxSubs, start: now, last: now}
        // Do not update the smap here, we need to do it in initLeafNodeSmapAndSendSubs
        c.leaf = &leaf{}

        // If the leafnode subject interest should be isolated, flag it here.
        s.optsMu.RLock()
        if c.leaf.isolated = s.opts.LeafNode.IsolateLeafnodeInterest; !c.leaf.isolated && remote != nil {
                c.leaf.isolated = remote.LocalIsolation
        }
        s.optsMu.RUnlock()

        // For accepted LN connections, ws will be != nil if it was accepted
        // through the Websocket port.
        c.ws = ws

        // For remote, check if the scheme starts with "ws", if so, we will initiate
        // a remote Leaf Node connection as a websocket connection.
        if remote != nil && rURL != nil && isWSURL(rURL) {
                remote.RLock()
                c.ws = &websocket{compress: remote.Websocket.Compression, maskwrite: !remote.Websocket.NoMasking}
                remote.RUnlock()
        }

        // Determines if we are soliciting the connection or not.
        var solicited bool
        var acc *Account
        var remoteSuffix string
        if remote != nil {
                // For now, if lookup fails, we will constantly try
                // to recreate this LN connection.
                lacc := remote.LocalAccount
                var err error
                acc, err = s.LookupAccount(lacc)
                if err != nil {
                        // An account not existing is something that can happen with nats/http account resolver and the account
                        // has not yet been pushed, or the request failed for other reasons.
                        // remote needs to be set or retry won't happen
                        c.leaf.remote = remote
                        c.closeConnection(MissingAccount)
                        s.Errorf("Unable to lookup account %s for solicited leafnode connection: %v", lacc, err)
                        return nil
                }
                remoteSuffix = fmt.Sprintf(" for account: %s", acc.traceLabel())
        }

        c.mu.Lock()
        c.initClient()
        c.Noticef("Leafnode connection created%s %s", remoteSuffix, c.opts.Name)

        var (
                tlsFirst         bool
                tlsFirstFallback time.Duration
                infoTimeout      time.Duration
        )
        if remote != nil {
                solicited = true
                remote.Lock()
                c.leaf.remote = remote
                c.setPermissions(remote.perms)
                if !c.leaf.remote.Hub {
                        c.leaf.isSpoke = true
                }
                tlsFirst = remote.TLSHandshakeFirst
                infoTimeout = remote.FirstInfoTimeout
                remote.Unlock()
                c.acc = acc
        } else {
                c.flags.set(expectConnect)
                if ws != nil {
                        c.Debugf("Leafnode compression=%v", c.ws.compress)
                }
                tlsFirst = opts.LeafNode.TLSHandshakeFirst
                if f := opts.LeafNode.TLSHandshakeFirstFallback; f > 0 {
                        tlsFirstFallback = f
                }
        }
        c.mu.Unlock()

        var nonce [nonceLen]byte
        var info *Info

        // Grab this before the client lock below.
        if !solicited {
                // Grab server variables
                s.mu.Lock()
                info = s.copyLeafNodeInfo()
                // For tests that want to simulate old servers, do not set the compression
                // on the INFO protocol if configured with CompressionNotSupported.
                // Also suppress it if WebSocket compression is already in use, otherwise
                // an old soliciting peer would honor the advertised mode, switch to S2,
                // and then wait forever for a compressed INFO response from us.
                if cm := opts.LeafNode.Compression.Mode; cm != CompressionNotSupported && (ws == nil || !ws.compress) {
                        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()
        }

        // Grab lock
        c.mu.Lock()

        var preBuf []byte
        if solicited {
                // For websocket connection, we need to send an HTTP request,
                // and get the response before starting the readLoop to get
                // the INFO, etc..
                if c.isWebsocket() {
                        var err error
                        var closeReason ClosedState

                        preBuf, closeReason, err = c.leafNodeSolicitWSConnection(opts, rURL, remote)
                        if err != nil {
                                c.Errorf("Error soliciting websocket connection: %v", err)
                                c.mu.Unlock()
                                if closeReason != 0 {
                                        c.closeConnection(closeReason)
                                }
                                return nil
                        }
                } else {
                        // If configured to do TLS handshake first
                        if tlsFirst {
                                if _, err := c.leafClientHandshakeIfNeeded(remote, opts); err != nil {
                                        c.mu.Unlock()
                                        return nil
                                }
                        }
                        // We need to wait for the info, but not for too long.
                        c.nc.SetReadDeadline(time.Now().Add(infoTimeout))
                }

                // We will process the INFO from the readloop and finish by
                // sending the CONNECT and finish registration later.
        } else {
                // Send our info to the other side.
                // Remember the nonce we sent here for signatures, etc.
                c.nonce = make([]byte, nonceLen)
                copy(c.nonce, nonce[:])
                info.Nonce = bytesToString(c.nonce)
                info.CID = c.cid
                proto := generateInfoJSON(info)

                var pre []byte
                // We need first to check for "TLS First" fallback delay.
                if tlsFirstFallback > 0 {
                        // We wait and see if we are getting any data. Since we did not send
                        // the INFO protocol yet, only clients that use TLS first should be
                        // sending data (the TLS handshake). We don't really check the content:
                        // if it is a rogue agent and not an actual client performing the
                        // TLS handshake, the error will be detected when performing the
                        // handshake on our side.
                        pre = make([]byte, 4)
                        c.nc.SetReadDeadline(time.Now().Add(tlsFirstFallback))
                        n, _ := io.ReadFull(c.nc, pre[:])
                        c.nc.SetReadDeadline(time.Time{})
                        // If we get any data (regardless of possible timeout), we will proceed
                        // with the TLS handshake.
                        if n > 0 {
                                pre = pre[:n]
                        } else {
                                // We did not get anything so we will send the INFO protocol.
                                pre = nil
                                // Set the boolean to false for the rest of the function.
                                tlsFirst = false
                        }
                }

                if !tlsFirst {
                        // We have to send from this go routine because we may
                        // have to block for TLS handshake before we start our
                        // writeLoop go routine. The other side needs to receive
                        // this before it can initiate the TLS handshake..
                        c.sendProtoNow(proto)

                        // The above call could have marked the connection as closed (due to TCP error).
                        if c.isClosed() {
                                c.mu.Unlock()
                                c.closeConnection(WriteError)
                                return nil
                        }
                }

                // Check to see if we need to spin up TLS.
                if !c.isWebsocket() && info.TLSRequired {
                        // If we have a prebuffer create a multi-reader.
                        if len(pre) > 0 {
                                c.nc = &tlsMixConn{c.nc, bytes.NewBuffer(pre)}
                        }
                        // Perform server-side TLS handshake.
                        if err := c.doTLSServerHandshake(tlsHandshakeLeaf, opts.LeafNode.TLSConfig, opts.LeafNode.TLSTimeout, opts.LeafNode.TLSPinnedCerts); err != nil {
                                c.mu.Unlock()
                                return nil
                        }
                }

                // If the user wants the TLS handshake to occur first, now that it is
                // done, send the INFO protocol.
                if tlsFirst {
                        c.flags.set(didTLSFirst)
                        c.sendProtoNow(proto)
                        if c.isClosed() {
                                c.mu.Unlock()
                                c.closeConnection(WriteError)
                                return nil
                        }
                }

                // Leaf nodes will always require a CONNECT to let us know
                // when we are properly bound to an account.
                //
                // If compression is configured, we can't set the authTimer here because
                // it would cause the parser to fail any incoming protocol that is not a
                // CONNECT (and we need to exchange INFO protocols for compression
                // negotiation). So instead, use the ping timer until we are done with
                // negotiation and can set the auth timer.
                timeout := secondsToDuration(opts.LeafNode.AuthTimeout)
                if needsCompression(opts.LeafNode.Compression.Mode) {
                        c.ping.tmr = time.AfterFunc(timeout, func() {
                                c.authTimeout()
                        })
                } else {
                        c.setAuthTimer(timeout)
                }
        }

        // Keep track in case server is shutdown before we can successfully register.
        if !s.addToTempClients(c.cid, c) {
                c.mu.Unlock()
                c.setNoReconnect()
                c.closeConnection(ServerShutdown)
                return nil
        }

        // Spin up the read loop.
        s.startGoRoutine(func() { c.readLoop(preBuf) })

        // We will spin the write loop for solicited connections only
        // when processing the INFO and after switching to TLS if needed.
        if !solicited {
                s.startGoRoutine(func() { c.writeLoop() })
        }

        c.mu.Unlock()

        return c
}

// Will perform the client-side TLS handshake if needed. Assumes that this
// is called by the solicit side (remote will be non nil). Returns `true`
// if TLS is required, `false` otherwise.
// Lock held on entry.
func (c *client) leafClientHandshakeIfNeeded(remote *leafNodeCfg, opts *Options) (bool, error) {
        // Check if TLS is required and gather TLS config variables.
        tlsRequired, tlsConfig, tlsName, tlsTimeout := c.leafNodeGetTLSConfigForSolicit(remote)
        if !tlsRequired {
                return false, nil
        }

        // If TLS required, peform handshake.
        // Get the URL that was used to connect to the remote server.
        rURL := remote.getCurrentURL()

        // Perform the client-side TLS handshake.
        if resetTLSName, err := c.doTLSClientHandshake(tlsHandshakeLeaf, rURL, tlsConfig, tlsName, tlsTimeout, opts.LeafNode.TLSPinnedCerts); err != nil {
                // Check if we need to reset the remote's TLS name.
                if resetTLSName {
                        remote.Lock()
                        remote.tlsName = _EMPTY_
                        remote.Unlock()
                }
                return false, err
        }
        return true, nil
}

func (c *client) processLeafnodeInfo(info *Info) {
        c.mu.Lock()
        if c.leaf == nil || c.isClosed() {
                c.mu.Unlock()
                return
        }
        s := c.srv
        opts := s.getOpts()
        remote := c.leaf.remote
        didSolicit := remote != nil
        firstINFO := !c.flags.isSet(infoReceived)

        // In case of websocket, the TLS handshake has been already done.
        // So check only for non websocket connections and for configurations
        // where the TLS Handshake was not done first.
        if didSolicit && !c.flags.isSet(handshakeComplete) && !c.isWebsocket() && !remote.TLSHandshakeFirst {
                // If the server requires TLS, we need to set this in the remote
                // otherwise if there is no TLS configuration block for the remote,
                // the solicit side will not attempt to perform the TLS handshake.
                if firstINFO && info.TLSRequired {
                        // Check for TLS/proxy configuration mismatch
                        if remote.Proxy.URL != _EMPTY_ && !remote.TLS && remote.TLSConfig == nil {
                                c.mu.Unlock()
                                c.Errorf("TLS configuration mismatch: Hub requires TLS but leafnode remote is not configured for TLS. When using a proxy, ensure TLS leafnode configuration matches the Hub requirements.")
                                c.closeConnection(TLSHandshakeError)
                                return
                        }
                        remote.TLS = true
                }
                if _, err := c.leafClientHandshakeIfNeeded(remote, opts); err != nil {
                        c.mu.Unlock()
                        return
                }
        }

        // Check for compression, unless already done.
        if firstINFO && !c.flags.isSet(compressionNegotiated) {
                // A solicited leafnode connection must first receive a leafnode INFO.
                // Classify wrong-port connections before any leaf-specific negotiation.
                if didSolicit && (info.CID == 0 || info.LeafNodeURLs == nil) {
                        c.mu.Unlock()
                        c.Errorf(ErrConnectedToWrongPort.Error())
                        c.closeConnection(WrongPort)
                        return
                }

                // Prevent from getting back here.
                c.flags.set(compressionNegotiated)

                var co *CompressionOpts
                if !didSolicit {
                        co = &opts.LeafNode.Compression
                } else {
                        co = &remote.Compression
                }
                if needsCompression(co.Mode) {
                        // Release client lock since following function will need server lock.
                        c.mu.Unlock()
                        compress, err := s.negotiateLeafCompression(c, didSolicit, info.Compression, co)
                        if err != nil {
                                c.sendErrAndErr(err.Error())
                                c.closeConnection(ProtocolViolation)
                                return
                        }
                        if compress {
                                // Done for now, will get back another INFO protocol...
                                return
                        }
                        // No compression because one side does not want/can't, so proceed.
                        c.mu.Lock()
                        // Check that the connection did not close if the lock was released.
                        if c.isClosed() {
                                c.mu.Unlock()
                                return
                        }
                } else {
                        // Coming from an old server, the Compression field would be the empty
                        // string. For servers that are configured with CompressionNotSupported,
                        // this makes them behave as old servers.
                        if info.Compression == _EMPTY_ || co.Mode == CompressionNotSupported {
                                c.leaf.compression = CompressionNotSupported
                        } else {
                                c.leaf.compression = CompressionOff
                        }
                }
                // Accepting side does not normally process an INFO protocol during
                // initial connection handshake. So we keep it consistent by returning
                // if we are not soliciting.
                if !didSolicit {
                        // If we had created the ping timer instead of the auth timer, we will
                        // clear the ping timer and set the auth timer now that the compression
                        // negotiation is done.
                        if info.Compression != _EMPTY_ && c.ping.tmr != nil {
                                clearTimer(&c.ping.tmr)
                                c.setAuthTimer(secondsToDuration(opts.LeafNode.AuthTimeout))
                        }
                        c.mu.Unlock()
                        return
                }
                // Fall through and process the INFO protocol as usual.
        }

        // Note: For now, only the initial INFO has a nonce. We
        // will probably do auto key rotation at some point.
        if firstINFO {
                // Mark that the INFO protocol has been received.
                c.flags.set(infoReceived)
                // Prevent connecting to non leafnode port. Need to do this only for
                // the first INFO, not for async INFO updates...
                //
                // Content of INFO sent by the server when accepting a tcp connection.
                // -------------------------------------------------------------------
                // Listen Port Of | CID | ClientConnectURLs | LeafNodeURLs | Gateway |
                // -------------------------------------------------------------------
                //      CLIENT    |  X* |        X**        |              |         |
                //      ROUTE     |     |        X**        |      X***    |         |
                //     GATEWAY    |     |                   |              |    X    |
                //     LEAFNODE   |  X  |                   |       X      |         |
                // -------------------------------------------------------------------
                // *   Not on older servers.
                // **  Not if "no advertise" is enabled.
                // *** Not if leafnode's "no advertise" is enabled.
                //
                // Reject a cluster that contains spaces.
                if info.Cluster != _EMPTY_ && strings.Contains(info.Cluster, " ") {
                        c.mu.Unlock()
                        c.sendErrAndErr(ErrClusterNameHasSpaces.Error())
                        c.closeConnection(ProtocolViolation)
                        return
                }
                // For solicited outbound leaf connections, capture the remote's nonce.
                // For inbound leaf connections, keep using the server-issued nonce that
                // was sent in our initial INFO and must be signed in CONNECT.
                if didSolicit {
                        c.nonce = []byte(info.Nonce)
                }
                if info.TLSRequired && didSolicit {
                        remote.TLS = true
                }
                supportsHeaders := c.srv.supportsHeaders()
                c.headers = supportsHeaders && info.Headers

                // Remember the remote server.
                // Pre 2.2.0 servers are not sending their server name.
                // In that case, use info.ID, which, for those servers, matches
                // the content of the field `Name` in the leafnode CONNECT protocol.
                if info.Name == _EMPTY_ {
                        c.leaf.remoteServer = info.ID
                } else {
                        c.leaf.remoteServer = info.Name
                }
                c.leaf.remoteDomain = info.Domain
                c.leaf.remoteCluster = info.Cluster
                // We send the protocol version in the INFO protocol.
                // Keep track of it, so we know if this connection supports message
                // tracing for instance.
                c.opts.Protocol = info.Proto
        }

        // For both initial INFO and async INFO protocols, Possibly
        // update our list of remote leafnode URLs we can connect to,
        // unless we are instructed not to.
        if didSolicit && !remote.IgnoreDiscoveredServers &&
                (len(info.LeafNodeURLs) > 0 || len(info.WSConnectURLs) > 0) {
                // Consider the incoming array as the most up-to-date
                // representation of the remote cluster's list of URLs.
                c.updateLeafNodeURLs(info)
        }

        // Only solicited leafnode connections trust permission updates from INFO.
        if didSolicit && (info.Import != nil || info.Export != nil) {
                perms := &Permissions{
                        Publish:   info.Export,
                        Subscribe: info.Import,
                }
                // Check if we have local deny clauses that we need to merge.
                if remote := c.leaf.remote; remote != nil {
                        if len(remote.DenyExports) > 0 {
                                if perms.Publish == nil {
                                        perms.Publish = &SubjectPermission{}
                                }
                                perms.Publish.Deny = append(perms.Publish.Deny, remote.DenyExports...)
                        }
                        if len(remote.DenyImports) > 0 {
                                if perms.Subscribe == nil {
                                        perms.Subscribe = &SubjectPermission{}
                                }
                                perms.Subscribe.Deny = append(perms.Subscribe.Deny, remote.DenyImports...)
                        }
                }
                c.setPermissions(perms)
        }

        var resumeConnect bool

        // If this is a remote connection and this is the first INFO protocol,
        // then we need to finish the connect process by sending CONNECT, etc..
        if firstINFO && didSolicit {
                // Clear deadline that was set in createLeafNode while waiting for the INFO.
                c.nc.SetDeadline(time.Time{})
                resumeConnect = true
        } else if !firstINFO && didSolicit {
                c.leaf.remoteAccName = info.RemoteAccount
        }

        // Check if we have the remote account information and if so make sure it's stored.
        if info.RemoteAccount != _EMPTY_ {
                if c.acc == nil {
                        c.mu.Unlock()
                        c.sendErr("Authorization Violation")
                        c.closeConnection(ProtocolViolation)
                        return
                }
                s.leafRemoteAccounts.Store(c.acc.Name, info.RemoteAccount)
        }
        c.mu.Unlock()

        finishConnect := info.ConnectInfo
        if resumeConnect && s != nil {
                s.leafNodeResumeConnectProcess(c)
                if !info.InfoOnConnect {
                        finishConnect = true
                }
        }
        if finishConnect {
                s.leafNodeFinishConnectProcess(c)
        }

        // Check to see if we need to kick any internal source or mirror consumers.
        // This will be a no-op if JetStream not enabled for this server or if the bound account
        // does not have jetstream.
        s.checkInternalSyncConsumers(c.acc)
}

func (s *Server) negotiateLeafCompression(c *client, didSolicit bool, infoCompression string, co *CompressionOpts) (bool, error) {
        // If WebSocket compression is already negotiated on this connection then
        // we shouldn't layer S2 compression on top of it.
        c.mu.Lock()
        if c.ws != nil && c.ws.compress {
                c.leaf.compression = CompressionOff
                c.mu.Unlock()
                return false, nil
        }
        c.mu.Unlock()
        // Negotiate the appropriate compression mode (or no compression)
        cm, err := selectCompressionMode(co.Mode, infoCompression)
        if err != nil {
                return false, err
        }
        c.mu.Lock()
        // For "auto" mode, set the initial compression mode based on RTT
        if cm == CompressionS2Auto {
                if c.rttStart.IsZero() {
                        c.rtt = computeRTT(c.start)
                }
                cm = selectS2AutoModeBasedOnRTT(c.rtt, co.RTTThresholds)
        }
        // Keep track of the negotiated compression mode.
        c.leaf.compression = cm
        cid := c.cid
        var nonce string
        if !didSolicit {
                nonce = bytesToString(c.nonce)
        }
        c.mu.Unlock()

        if !needsCompression(cm) {
                return false, nil
        }

        // If we end-up doing compression...

        // Generate an INFO with the chosen compression mode.
        s.mu.Lock()
        info := s.copyLeafNodeInfo()
        info.Compression, info.CID, info.Nonce = compressionModeForInfoProtocol(co, cm), cid, nonce
        infoProto := generateInfoJSON(info)
        s.mu.Unlock()

        // If we solicited, then send this INFO protocol BEFORE switching
        // to compression writer. However, if we did not, we send it after.
        c.mu.Lock()
        if didSolicit {
                c.enqueueProto(infoProto)
                // Make sure it is completely flushed (the pending bytes goes to
                // 0) before proceeding.
                for c.out.pb > 0 && !c.isClosed() {
                        c.flushOutbound()
                }
        }
        // This is to notify the readLoop that it should switch to a
        // (de)compression reader.
        c.in.flags.set(switchToCompression)
        // Create the compress writer before queueing the INFO protocol for
        // a route that did not solicit. It will make sure that that proto
        // is sent with compression on.
        c.out.cw = s2.NewWriter(nil, s2WriterOptions(cm)...)
        if !didSolicit {
                c.enqueueProto(infoProto)
        }
        c.mu.Unlock()
        return true, nil
}

// When getting a leaf node INFO protocol, use the provided
// array of urls to update the list of possible endpoints.
func (c *client) updateLeafNodeURLs(info *Info) {
        cfg := c.leaf.remote
        cfg.Lock()
        defer cfg.Unlock()

        // We have ensured that if a remote has a WS scheme, then all are.
        // So check if first is WS, then add WS URLs, otherwise, add non WS ones.
        if len(cfg.URLs) > 0 && isWSURL(cfg.URLs[0]) {
                // It does not really matter if we use "ws://" or "wss://" here since
                // we will have already marked that the remote should use TLS anyway.
                // But use proper scheme for log statements, etc...
                proto := wsSchemePrefix
                if cfg.TLS {
                        proto = wsSchemePrefixTLS
                }
                c.doUpdateLNURLs(cfg, proto, info.WSConnectURLs)
                return
        }
        c.doUpdateLNURLs(cfg, "nats-leaf", info.LeafNodeURLs)
}

func (c *client) doUpdateLNURLs(cfg *leafNodeCfg, scheme string, URLs []string) {
        cfg.urls = make([]*url.URL, 0, 1+len(URLs))
        // Add the ones we receive in the protocol
        for _, surl := range URLs {
                url, err := url.Parse(fmt.Sprintf("%s://%s", scheme, surl))
                if err != nil {
                        // As per below, the URLs we receive should not have contained URL info, so this should be safe to log.
                        c.Errorf("Error parsing url %q: %v", surl, err)
                        continue
                }
                // Do not add if it's the same as what we already have configured.
                var dup bool
                for _, u := range cfg.URLs {
                        // URLs that we receive never have user info, but the
                        // ones that were configured may have. Simply compare
                        // host and port to decide if they are equal or not.
                        if url.Host == u.Host && url.Port() == u.Port() {
                                dup = true
                                break
                        }
                }
                if !dup {
                        cfg.urls = append(cfg.urls, url)
                        cfg.saveTLSHostname(url)
                }
        }
        // Add the configured one
        cfg.urls = append(cfg.urls, cfg.URLs...)
}

// Similar to setInfoHostPortAndGenerateJSON, but for leafNodeInfo.
func (s *Server) setLeafNodeInfoHostPortAndIP() error {
        opts := s.getOpts()
        if opts.LeafNode.Advertise != _EMPTY_ {
                advHost, advPort, err := parseHostPort(opts.LeafNode.Advertise, opts.LeafNode.Port)
                if err != nil {
                        return err
                }
                s.leafNodeInfo.Host = advHost
                s.leafNodeInfo.Port = advPort
        } else {
                s.leafNodeInfo.Host = opts.LeafNode.Host
                s.leafNodeInfo.Port = opts.LeafNode.Port
                // If the host is "0.0.0.0" or "::" we need to resolve to a public IP.
                // This will return at most 1 IP.
                hostIsIPAny, ips, err := s.getNonLocalIPsIfHostIsIPAny(s.leafNodeInfo.Host, false)
                if err != nil {
                        return err
                }
                if hostIsIPAny {
                        if len(ips) == 0 {
                                s.Errorf("Could not find any non-local IP for leafnode's listen specification %q",
                                        s.leafNodeInfo.Host)
                        } else {
                                // Take the first from the list...
                                s.leafNodeInfo.Host = ips[0]
                        }
                }
        }
        // Use just host:port for the IP
        s.leafNodeInfo.IP = net.JoinHostPort(s.leafNodeInfo.Host, strconv.Itoa(s.leafNodeInfo.Port))
        if opts.LeafNode.Advertise != _EMPTY_ {
                s.Noticef("Advertise address for leafnode is set to %s", s.leafNodeInfo.IP)
        }
        return nil
}

// Add the connection to the map of leaf nodes.
// If `checkForDup` is true (invoked when a leafnode is accepted), then we check
// if a connection already exists for the same server name and account.
// That can happen when the remote is attempting to reconnect while the accepting
// side did not detect the connection as broken yet.
// But it can also happen when there is a misconfiguration and the remote is
// creating two (or more) connections that bind to the same account on the accept
// side.
// When a duplicate is found, the new connection is accepted and the old is closed
// (this solves the stale connection situation). An error is returned to help the
// remote detect the misconfiguration when the duplicate is the result of that
// misconfiguration.
func (s *Server) addLeafNodeConnection(c *client, srvName, clusterName string, checkForDup bool) bool {
        var accName string
        c.mu.Lock()
        cid := c.cid
        acc := c.acc
        if acc != nil {
                accName = acc.Name
        }
        myRemoteDomain := c.leaf.remoteDomain
        mySrvName := c.leaf.remoteServer
        remoteAccName := c.leaf.remoteAccName
        myClustName := c.leaf.remoteCluster
        remote := c.leaf.remote
        solicited := remote != nil
        c.mu.Unlock()

        var old *client
        s.mu.Lock()
        // We check for empty because in some test we may send empty CONNECT{}
        if checkForDup && srvName != _EMPTY_ {
                for _, ol := range s.leafs {
                        ol.mu.Lock()
                        // We care here only about non solicited Leafnode. This function
                        // is more about replacing stale connections than detecting loops.
                        // We have code for the loop detection elsewhere, which also delays
                        // attempt to reconnect.
                        if !ol.isSolicitedLeafNode() && ol.leaf.remoteServer == srvName &&
                                ol.leaf.remoteCluster == clusterName && ol.acc.Name == accName &&
                                remoteAccName != _EMPTY_ && ol.leaf.remoteAccName == remoteAccName {
                                old = ol
                        }
                        ol.mu.Unlock()
                        if old != nil {
                                break
                        }
                }
        }
        // Now that we are under the server lock and before adding it to the map,
        // for a solicited leaf, we need to make sure that it has not been removed
        // from the config or disabled.
        if solicited {
                // If no longer valid, do not add to the server map. The connection
                // should have been marked so that it can't reconnect. When the caller
                // calls closeConnection(), cleanup (including clearing the connect-
                // in-progress flag) will occur at the appropriate time.
                if !remote.stillValid() {
                        // Prevent reconnect in case it was not yet done.
                        c.setNoReconnect()
                        s.mu.Unlock()
                        s.removeFromTempClients(cid)
                        return false
                }
                remote.setConnectInProgress(false)
        }
        // Store new connection in the map
        s.leafs[cid] = c
        s.mu.Unlock()
        s.removeFromTempClients(cid)

        // If applicable, evict the old one.
        if old != nil {
                old.sendErrAndErr(DuplicateRemoteLeafnodeConnection.String())
                old.closeConnection(DuplicateRemoteLeafnodeConnection)
                c.Warnf("Replacing connection from same server")
        }

        srvDecorated := func() string {
                if myClustName == _EMPTY_ {
                        return mySrvName
                }
                return fmt.Sprintf("%s/%s", mySrvName, myClustName)
        }

        opts := s.getOpts()
        sysAcc := s.SystemAccount()
        js := s.getJetStream()
        var meta *raft
        if js != nil {
                if mg := js.getMetaGroup(); mg != nil {
                        meta = mg.(*raft)
                }
        }
        blockMappingOutgoing := false
        // Deny (non domain) JetStream API traffic unless system account is shared
        // and domain names are identical and extending is not disabled

        // Check if backwards compatibility has been enabled and needs to be acted on
        forceSysAccDeny := false
        if len(opts.JsAccDefaultDomain) > 0 {
                if acc == sysAcc {
                        for _, d := range opts.JsAccDefaultDomain {
                                if d == _EMPTY_ {
                                        // Extending JetStream via leaf node is mutually exclusive with a domain mapping to the empty/default domain.
                                        // As soon as one mapping to "" is found, disable the ability to extend JS via a leaf node.
                                        c.Noticef("Not extending remote JetStream domain %q due to presence of empty default domain", myRemoteDomain)
                                        forceSysAccDeny = true
                                        break
                                }
                        }
                } else if domain, ok := opts.JsAccDefaultDomain[accName]; ok && domain == _EMPTY_ {
                        // for backwards compatibility with old setups that do not have a domain name set
                        c.Debugf("Skipping deny %q for account %q due to default domain", jsAllAPI, accName)
                        return true
                }
        }

        // If the server has JS disabled, it may still be part of a JetStream that could be extended.
        // This is either signaled by js being disabled and a domain set,
        // or in cases where no domain name exists, an extension hint is set.
        // However, this is only relevant in mixed setups.
        //
        // If the system account connects but default domains are present, JetStream can't be extended.
        if opts.JetStreamDomain != myRemoteDomain || (!opts.JetStream && (opts.JetStreamDomain == _EMPTY_ && opts.JetStreamExtHint != jsWillExtend)) ||
                sysAcc == nil || acc == nil || forceSysAccDeny {
                // If domain names mismatch always deny. This applies to system accounts as well as non system accounts.
                // Not having a system account, account or JetStream disabled is considered a mismatch as well.
                if acc != nil && acc == sysAcc {
                        c.Noticef("System account connected from %s", srvDecorated())
                        c.Noticef("JetStream not extended, domains differ")
                        c.mergeDenyPermissionsLocked(both, denyAllJs)
                        // When a remote with a system account is present in a server, unless otherwise disabled, the server will be
                        // started in observer mode. Now that it is clear that this not used, turn the observer mode off.
                        if solicited && meta != nil && meta.IsObserver() {
                                meta.setObserver(false, extNotExtended)
                                c.Debugf("Turning JetStream metadata controller Observer Mode off")
                                // Take note that the domain was not extended to avoid this state from startup.
                                writePeerState(c.srv.diskIOSemaphore(), js.config.StoreDir, meta.currentPeerState())
                                // Meta controller can't be leader yet.
                                // Yet it is possible that due to observer mode every server already stopped campaigning.
                                // Therefore this server needs to be kicked into campaigning gear explicitly.
                                meta.Campaign()
                        }
                } else {
                        c.Noticef("JetStream using domains: local %q, remote %q", opts.JetStreamDomain, myRemoteDomain)
                        c.mergeDenyPermissionsLocked(both, denyAllClientJs)
                }
                blockMappingOutgoing = true
        } else if acc == sysAcc {
                // system account and same domain
                s.sys.client.Noticef("Extending JetStream domain %q as System Account connected from server %s",
                        myRemoteDomain, srvDecorated())
                // In an extension use case, pin leadership to server remotes connect to.
                // Therefore, server with a remote that are not already in observer mode, need to be put into it.
                if solicited && meta != nil && !meta.IsObserver() {
                        c.Debugf("Turning JetStream metadata controller Observer Mode on - System Account Connected")
                        // Discard any local metagroup state accumulated before the SYS-account
                        // leaf came up (e.g. the wrong-hint case where this server bootstrapped
                        // its own metagroup). The parent's view is now authoritative; without
                        // this reset the two raft logs stay forked because the standalone log's
                        // commit prefix short-circuits the follower's AE handling.
                        meta.setObserver(true, extExtended)
                        meta.Reset()
                }
        } else {
                // This deny is needed in all cases (system account shared or not)
                // If the system account is shared, jsAllAPI traffic will go through the system account.
                // So in order to prevent duplicate delivery (from system and actual account) suppress it on the account.
                // If the system account is NOT shared, jsAllAPI traffic has no business
                c.Debugf("Adding deny %+v for account %q", denyAllClientJs, accName)
                c.mergeDenyPermissionsLocked(both, denyAllClientJs)
        }
        // If we have a specified JetStream domain we will want to add a mapping to
        // allow access cross domain for each non-system account.
        if opts.JetStreamDomain != _EMPTY_ && opts.JetStream && acc != nil && acc != sysAcc {
                for src, dest := range generateJSMappingTable(opts.JetStreamDomain) {
                        if err := acc.AddMapping(src, dest); err != nil {
                                c.Debugf("Error adding JetStream domain mapping: %s", err.Error())
                        } else {
                                c.Debugf("Adding JetStream Domain Mapping %q -> %s to account %q", src, dest, accName)
                        }
                }
                if blockMappingOutgoing {
                        src := fmt.Sprintf(jsDomainAPI, opts.JetStreamDomain)
                        // make sure that messages intended for this domain, do not leave the cluster via this leaf node connection
                        // This is a guard against a miss-config with two identical domain names and will only cover some forms
                        // of this issue, not all of them.
                        // This guards against a hub and a spoke having the same domain name.
                        // But not two spokes having the same one and the request coming from the hub.
                        c.mergeDenyPermissionsLocked(pub, []string{src})
                        c.Debugf("Adding deny %q for outgoing messages to account %q", src, accName)
                }
        }
        return true
}

func (s *Server) removeLeafNodeConnection(c *client) {
        s.mu.Lock()
        c.mu.Lock()
        cid := c.cid
        if c.leaf != nil {
                if c.leaf.tsubt != nil {
                        c.leaf.tsubt.Stop()
                        c.leaf.tsubt = nil
                }
                if c.leaf.gwSub != nil {
                        s.gwLeafSubs.Remove(c.leaf.gwSub)
                        // We need to set this to nil for GC to release the connection
                        c.leaf.gwSub = nil
                }
                if remote := c.leaf.remote; remote != nil {
                        // If "noReconnect" is true, then we won't attempt to reconnect, so
                        // we will clear the "connect-in-progress" flag. However, if we can
                        // reconnect, then we should set "connect-in-progress" to true while
                        // we are under the server/client lock. The go routine that performs
                        // the reconnect will be started later and there would be a gap with
                        // the wrong flag value otherwise.
                        remote.setConnectInProgress(!c.flags.isSet(noReconnect))
                }
        }
        proxyKey := c.proxyKey
        c.mu.Unlock()
        delete(s.leafs, cid)
        if proxyKey != _EMPTY_ {
                s.removeProxiedConn(proxyKey, cid)
        }
        s.mu.Unlock()
        s.removeFromTempClients(cid)
}

// Connect information for solicited leafnodes.
type leafConnectInfo struct {
        Version   string   `json:"version,omitempty"`
        Nkey      string   `json:"nkey,omitempty"`
        JWT       string   `json:"jwt,omitempty"`
        Sig       string   `json:"sig,omitempty"`
        User      string   `json:"user,omitempty"`
        Pass      string   `json:"pass,omitempty"`
        Token     string   `json:"auth_token,omitempty"`
        ID        string   `json:"server_id,omitempty"`
        Domain    string   `json:"domain,omitempty"`
        Name      string   `json:"name,omitempty"`
        Hub       bool     `json:"is_hub,omitempty"`
        Cluster   string   `json:"cluster,omitempty"`
        Headers   bool     `json:"headers,omitempty"`
        JetStream bool     `json:"jetstream,omitempty"`
        DenyPub   []string `json:"deny_pub,omitempty"`
        Isolate   bool     `json:"isolate,omitempty"`

        // There was an existing field called:
        // >> Comp bool `json:"compression,omitempty"`
        // that has never been used. With support for compression, we now need
        // a field that is a string. So we use a different json tag:
        Compression string `json:"compress_mode,omitempty"`

        // Just used to detect wrong connection attempts.
        Gateway string `json:"gateway,omitempty"`

        // Tells the accept side which account the remote is binding to.
        RemoteAccount string `json:"remote_account,omitempty"`

        // The accept side of a LEAF connection, unlike ROUTER and GATEWAY, receives
        // only the CONNECT protocol, and no INFO. So we need to send the protocol
        // version as part of the CONNECT. It will indicate if a connection supports
        // some features, such as message tracing.
        // We use `protocol` as the JSON tag, so this is automatically unmarshal'ed
        // in the low level process CONNECT.
        Proto int `json:"protocol,omitempty"`
}

// processLeafNodeConnect will process the inbound connect args.
// Once we are here we are bound to an account, so can send any interest that
// we would have to the other side.
func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) error {
        // Way to detect clients that incorrectly connect to the route listen
        // port. Client provided "lang" in the CONNECT protocol while LEAFNODEs don't.
        if lang != _EMPTY_ {
                c.sendErrAndErr(ErrClientConnectedToLeafNodePort.Error())
                c.closeConnection(WrongPort)
                return ErrClientConnectedToLeafNodePort
        }

        // Unmarshal as a leaf node connect protocol
        proto := &leafConnectInfo{}
        if err := json.Unmarshal(arg, proto); err != nil {
                return err
        }

        // Reject a cluster that contains spaces.
        if proto.Cluster != _EMPTY_ && strings.Contains(proto.Cluster, " ") {
                c.sendErrAndErr(ErrClusterNameHasSpaces.Error())
                c.closeConnection(ProtocolViolation)
                return ErrClusterNameHasSpaces
        }

        // Check for cluster name collisions.
        if cn := s.cachedClusterName(); cn != _EMPTY_ && proto.Cluster != _EMPTY_ && proto.Cluster == cn {
                c.sendErrAndErr(ErrLeafNodeHasSameClusterName.Error())
                c.closeConnection(ClusterNamesIdentical)
                return ErrLeafNodeHasSameClusterName
        }

        // Reject if this has Gateway which means that it would be from a gateway
        // connection that incorrectly connects to the leafnode port.
        if proto.Gateway != _EMPTY_ {
                errTxt := fmt.Sprintf("Rejecting connection from gateway %q on the leafnode port", proto.Gateway)
                c.Errorf(errTxt)
                c.sendErr(errTxt)
                c.closeConnection(WrongGateway)
                return ErrWrongGateway
        }

        if mv := s.getOpts().LeafNode.MinVersion; mv != _EMPTY_ {
                major, minor, update, _ := versionComponents(mv)
                if !versionAtLeast(proto.Version, major, minor, update) {
                        // Send back an INFO so recent remote servers process the rejection
                        // cleanly, then close immediately. The soliciting side applies the
                        // reconnect delay when it processes the error.
                        s.sendPermsAndAccountInfo(c)
                        c.sendErrAndErr(fmt.Sprintf("%s %q", ErrLeafNodeMinVersionRejected, mv))
                        c.closeConnection(MinimumVersionRequired)
                        return ErrMinimumVersionRequired
                }
        }

        // Check if this server supports headers.
        supportHeaders := c.srv.supportsHeaders()

        c.mu.Lock()
        // Leaf Nodes do not do echo or verbose or pedantic.
        c.opts.Verbose = false
        c.opts.Echo = false
        c.opts.Pedantic = false
        // This inbound connection will be marked as supporting headers if this server
        // support headers and the remote has sent in the CONNECT protocol that it does
        // support headers too.
        c.headers = supportHeaders && proto.Headers
        // If the compression level is still not set, set it based on what has been
        // given to us in the CONNECT protocol.
        if c.leaf.compression == _EMPTY_ {
                // But if proto.Compression is _EMPTY_, set it to CompressionNotSupported
                if proto.Compression == _EMPTY_ {
                        c.leaf.compression = CompressionNotSupported
                } else {
                        c.leaf.compression = proto.Compression
                }
        }

        // Remember the remote server.
        c.leaf.remoteServer = proto.Name
        // Remember the remote account name
        c.leaf.remoteAccName = proto.RemoteAccount
        // Remember if the leafnode requested isolation.
        c.leaf.isolated = c.leaf.isolated || proto.Isolate

        // If the other side has declared itself a hub, so we will take on the spoke role.
        if proto.Hub {
                c.leaf.isSpoke = true
        }

        // The soliciting side is part of a cluster.
        if proto.Cluster != _EMPTY_ {
                c.leaf.remoteCluster = proto.Cluster
        }

        c.leaf.remoteDomain = proto.Domain

        // When a leaf solicits a connection to a hub, the perms that it will use on the soliciting leafnode's
        // behalf are correct for them, but inside the hub need to be reversed since data is flowing in the opposite direction.
        if !c.isSolicitedLeafNode() && c.perms != nil {
                sp, pp := c.perms.sub, c.perms.pub
                c.perms.sub, c.perms.pub = pp, sp
                if c.opts.Import != nil {
                        c.darray = c.opts.Import.Deny
                } else {
                        c.darray = nil
                }
        }

        // Set the Ping timer
        c.setFirstPingTimer()

        // If we received pub deny permissions from the other end, merge with existing ones.
        c.mergeDenyPermissions(pub, proto.DenyPub)

        acc := c.acc
        c.mu.Unlock()

        // If the account is not set (e.g. connection was closed due to auth
        // timeout while still being processed), bail out to avoid a panic.
        if acc == nil {
                c.closeConnection(MissingAccount)
                return ErrMissingAccount
        }

        // Register the cluster, even if empty, as long as we are acting as a hub.
        if !proto.Hub {
                acc.registerLeafNodeCluster(proto.Cluster)
        }

        // Add in the leafnode here since we passed through auth at this point.
        s.addLeafNodeConnection(c, proto.Name, proto.Cluster, true)

        // If we have permissions bound to this leafnode we need to send then back to the
        // origin server for local enforcement.
        s.sendPermsAndAccountInfo(c)

        // Create and initialize the smap since we know our bound account now.
        // This will send all registered subs too.
        s.initLeafNodeSmapAndSendSubs(c)

        // Announce the account connect event for a leaf node.
        // This will be a no-op as needed.
        s.sendLeafNodeConnect(c.acc)

        // Check to see if we need to kick any internal source or mirror consumers.
        // This will be a no-op if JetStream not enabled for this server or if the bound account
        // does not have jetstream.
        s.checkInternalSyncConsumers(acc)

        return nil
}

// checkInternalSyncConsumers
func (s *Server) checkInternalSyncConsumers(acc *Account) {
        // Grab our js
        js := s.getJetStream()

        // Only applicable if we have JS and the leafnode has JS as well.
        // We check for remote JS outside.
        if !js.isEnabled() || acc == nil {
                return
        }

        // We will check all streams in our local account. They must be a leader and
        // be sourcing or mirroring. We will check the external config on the stream itself
        // if this is cross domain, or if the remote domain is empty, meaning we might be
        // extending the system across this leafnode connection and hence we would be extending
        // our own domain.
        jsa := js.lookupAccount(acc)
        if jsa == nil {
                return
        }

        var streams []*stream
        jsa.mu.RLock()
        for _, mset := range jsa.streams {
                mset.cfgMu.RLock()
                // We need to have a mirror or source defined.
                // We do not want to force another lock here to look for leader status,
                // so collect and after we release jsa will make sure.
                if mset.cfg.Mirror != nil || len(mset.cfg.Sources) > 0 {
                        streams = append(streams, mset)
                }
                mset.cfgMu.RUnlock()
        }
        jsa.mu.RUnlock()

        // Now loop through all candidates and check if we are the leader and have NOT
        // created the sync up consumer.
        for _, mset := range streams {
                mset.retryDisconnectedSyncConsumers()
        }
}

// Returns the remote cluster name. This is set only once so does not require a lock.
func (c *client) remoteCluster() string {
        if c.leaf == nil {
                return _EMPTY_
        }
        return c.leaf.remoteCluster
}

// Sends back an info block to the soliciting leafnode to let it know about
// its permission settings for local enforcement.
func (s *Server) sendPermsAndAccountInfo(c *client) {
        // Copy
        s.mu.Lock()
        info := s.copyLeafNodeInfo()
        s.mu.Unlock()
        c.mu.Lock()
        info.CID = c.cid
        info.Import = c.opts.Import
        info.Export = c.opts.Export
        info.RemoteAccount = c.acc.Name
        // s.SystemAccount() uses an atomic operation and does not get the server lock, so this is safe.
        info.IsSystemAccount = c.acc == s.SystemAccount()
        info.ConnectInfo = true
        c.enqueueProto(generateInfoJSON(info))
        c.mu.Unlock()
}

// Snapshot the current subscriptions from the sublist into our smap which
// we will keep updated from now on.
// Also send the registered subscriptions.
func (s *Server) initLeafNodeSmapAndSendSubs(c *client) {
        acc := c.acc
        if acc == nil {
                c.Debugf("Leafnode does not have an account bound")
                return
        }
        // Collect all account subs here.
        _subs := [1024]*subscription{}
        subs := _subs[:0]
        ims := []string{}

        // Hold the client lock otherwise there can be a race and miss some subs.
        c.mu.Lock()
        defer c.mu.Unlock()

        acc.mu.RLock()
        accName := acc.Name
        accNTag := acc.nameTag

        // To make printing look better when no friendly name present.
        if accNTag != _EMPTY_ {
                accNTag = "/" + accNTag
        }

        // If we are solicited we only send interest for local clients.
        if c.isSpokeLeafNode() {
                acc.sl.localSubs(&subs, true)
        } else {
                acc.sl.All(&subs)
        }

        // Check if we have an existing service import reply.
        siReply := copyBytes(acc.siReply)

        // Since leaf nodes only send on interest, if the bound
        // account has import services we need to send those over.
        for isubj := range acc.imports.services {
                if c.isSpokeLeafNode() && !c.canSubscribe(isubj) {
                        c.Debugf("Not permitted to import service %q on behalf of %s%s", isubj, accName, accNTag)
                        continue
                }
                ims = append(ims, isubj)
        }
        // Likewise for mappings.
        for _, m := range acc.mappings {
                if c.isSpokeLeafNode() && !c.canSubscribe(m.src) {
                        c.Debugf("Not permitted to import mapping %q on behalf of %s%s", m.src, accName, accNTag)
                        continue
                }
                ims = append(ims, m.src)
        }

        // Create a unique subject that will be used for loop detection.
        lds := acc.lds
        acc.mu.RUnlock()

        // Check if we have to create the LDS.
        if lds == _EMPTY_ {
                lds = leafNodeLoopDetectionSubjectPrefix + nuid.Next()
                acc.mu.Lock()
                acc.lds = lds
                acc.mu.Unlock()
        }

        // Now check for gateway interest. Leafnodes will put this into
        // the proper mode to propagate, but they are not held in the account.
        gwsa := [16]*client{}
        gws := gwsa[:0]
        s.getOutboundGatewayConnections(&gws)
        for _, cgw := range gws {
                cgw.mu.Lock()
                gw := cgw.gw
                cgw.mu.Unlock()
                if gw != nil {
                        if ei, _ := gw.outsim.Load(accName); ei != nil {
                                if e := ei.(*outsie); e != nil && e.sl != nil {
                                        e.sl.All(&subs)
                                }
                        }
                }
        }

        applyGlobalRouting := s.gateway.enabled
        if c.isSpokeLeafNode() {
                // Add a fake subscription for this solicited leafnode connection
                // so that we can send back directly for mapped GW replies.
                // We need to keep track of this subscription so it can be removed
                // when the connection is closed so that the GC can release it.
                c.leaf.gwSub = &subscription{client: c, subject: []byte(gwReplyPrefix + ">")}
                c.srv.gwLeafSubs.Insert(c.leaf.gwSub)
        }

        // Now walk the results and add them to our smap
        rc := c.leaf.remoteCluster
        c.leaf.smap = make(map[string]int32)
        for _, sub := range subs {
                // Check perms regardless of role.
                if c.perms != nil && !c.canSubscribe(string(sub.subject)) {
                        c.Debugf("Not permitted to subscribe to %q on behalf of %s%s", sub.subject, accName, accNTag)
                        continue
                }
                // Don't advertise interest from leafnodes to other isolated leafnodes.
                if sub.client.kind == LEAF && c.isIsolatedLeafNode() {
                        continue
                }
                // We ignore ourselves here.
                // Also don't add the subscription if it has a origin cluster and the
                // cluster name matches the one of the client we are sending to.
                if c != sub.client && (sub.origin == nil || (bytesToString(sub.origin) != rc)) {
                        count := int32(1)
                        if len(sub.queue) > 0 && sub.qw > 0 {
                                count = sub.qw
                        }
                        c.leaf.smap[keyFromSub(sub)] += count
                        if c.leaf.tsub == nil {
                                c.leaf.tsub = make(map[*subscription]struct{})
                        }
                        c.leaf.tsub[sub] = struct{}{}
                }
        }
        // FIXME(dlc) - We need to update appropriately on an account claims update.
        for _, isubj := range ims {
                c.leaf.smap[isubj]++
        }
        // If we have gateways enabled we need to make sure the other side sends us responses
        // that have been augmented from the original subscription.
        // TODO(dlc) - Should we lock this down more?
        if applyGlobalRouting {
                c.leaf.smap[oldGWReplyPrefix+"*.>"]++
                c.leaf.smap[gwReplyPrefix+">"]++
        }
        // Detect loops by subscribing to a specific subject and checking
        // if this sub is coming back to us.
        c.leaf.smap[lds]++

        // Check if we need to add an existing siReply to our map.
        // This will be a prefix so add on the wildcard.
        if siReply != nil {
                wcsub := append(siReply, '>')
                c.leaf.smap[string(wcsub)]++
        }
        // Queue all protocols. There is no max pending limit for LN connection,
        // so we don't need chunking. The writes will happen from the writeLoop.
        var b bytes.Buffer
        for key, n := range c.leaf.smap {
                c.writeLeafSub(&b, key, n)
        }
        if b.Len() > 0 {
                c.enqueueProto(b.Bytes())
        }
        if c.leaf.tsub != nil {
                // Clear the tsub map after 5 seconds.
                c.leaf.tsubt = time.AfterFunc(5*time.Second, func() {
                        c.mu.Lock()
                        if c.leaf != nil {
                                c.leaf.tsub = nil
                                c.leaf.tsubt = nil
                        }
                        c.mu.Unlock()
                })
        }
}

// updateInterestForAccountOnGateway called from gateway code when processing RS+ and RS-.
func (s *Server) updateInterestForAccountOnGateway(accName string, sub *subscription, delta int32) {
        // Since we're in the gateway's readLoop, and we would otherwise block, don't allow fetching.
        acc, err := s.lookupOrFetchAccount(accName, false)
        if acc == nil || err != nil {
                s.Debugf("No or bad account for %q, failed to update interest from gateway", accName)
                return
        }
        acc.updateLeafNodes(sub, delta)
}

// updateLeafNodesEx will make sure to update the account smap for the subscription.
// Will also forward to all leaf nodes as needed.
// If `hubOnly` is true, then will update only leaf nodes that connect to this server
// (that is, for which this server acts as a hub to them).
func (acc *Account) updateLeafNodesEx(sub *subscription, delta int32, hubOnly bool) {
        if acc == nil || sub == nil {
                return
        }

        // We will do checks for no leafnodes and same cluster here inline and under the
        // general account read lock.
        // If we feel we need to update the leafnodes we will do that out of line to avoid
        // blocking routes or GWs.

        acc.mu.RLock()
        // First check if we even have leafnodes here.
        if acc.nleafs == 0 {
                acc.mu.RUnlock()
                return
        }

        // Is this a loop detection subject.
        isLDS := bytes.HasPrefix(sub.subject, []byte(leafNodeLoopDetectionSubjectPrefix))

        // Capture the cluster even if its empty.
        var cluster string
        if sub.origin != nil {
                cluster = bytesToString(sub.origin)
        }

        // If we have an isolated cluster we can return early, as long as it is not a loop detection subject.
        // Empty clusters will return false for the check.
        if !isLDS && acc.isLeafNodeClusterIsolated(cluster) {
                acc.mu.RUnlock()
                return
        }

        // We can release the general account lock.
        acc.mu.RUnlock()

        // We can hold the list lock here to avoid having to copy a large slice.
        acc.lmu.RLock()
        defer acc.lmu.RUnlock()

        // Do this once.
        subject := string(sub.subject)

        // Walk the connected leafnodes from a random starting point to avoid
        // concurrent callers all contending over leafs in the same order.
        nleafs := len(acc.lleafs)
        start := 0
        if nleafs > 1 {
                start = rand.Intn(nleafs)
        }
        for i := 0; i < nleafs; i++ {
                ln := acc.lleafs[(start+i)%nleafs]
                if ln == sub.client {
                        continue
                }
                ln.mu.RLock()
                // Don't advertise interest from leafnodes to other isolated leafnodes.
                if sub.client.kind == LEAF && ln.isIsolatedLeafNode() {
                        ln.mu.RUnlock()
                        continue
                }
                // If `hubOnly` is true, it means that we want to update only leafnodes
                // that connect to this server (so isHubLeafNode() would return `true`).
                if hubOnly && !ln.isHubLeafNode() {
                        ln.mu.RUnlock()
                        continue
                }
                // Check to make sure this sub does not have an origin cluster that matches the leafnode.
                // If skipped, make sure that we still let go the "$LDS." subscription that allows
                // the detection of loops as long as different cluster.
                clusterDifferent := cluster != ln.remoteCluster()
                update := (isLDS && clusterDifferent) ||
                        ((cluster == _EMPTY_ || clusterDifferent) && (delta <= 0 || ln.canSubscribeInternal(subject)))
                ln.mu.RUnlock()
                if update {
                        ln.mu.Lock()
                        // The leaf role, isolation mode, and remote cluster are stable
                        // for the connection. Recheck canSubscribe here since permissions
                        // can change, and to initializes mperms for wildcard subscriptions
                        // that collide with deny rules.
                        if isLDS || delta <= 0 || ln.canSubscribe(subject) {
                                ln.updateSmap(sub, delta, isLDS)
                        }
                        ln.mu.Unlock()
                }
        }
}

// updateLeafNodes will make sure to update the account smap for the subscription.
// Will also forward to all leaf nodes as needed.
func (acc *Account) updateLeafNodes(sub *subscription, delta int32) {
        acc.updateLeafNodesEx(sub, delta, false)
}

// This will make an update to our internal smap and determine if we should send out
// an interest update to the remote side.
// Lock should be held.
func (c *client) updateSmap(sub *subscription, delta int32, isLDS bool) {
        if c.leaf.smap == nil {
                return
        }

        // If we are solicited make sure this is a local client or a non-solicited leaf node
        skind := sub.client.kind
        updateClient := skind == CLIENT || skind == SYSTEM || skind == JETSTREAM || skind == ACCOUNT
        if !isLDS && c.isSpokeLeafNode() && !(updateClient || (skind == LEAF && !sub.client.isSpokeLeafNode())) {
                return
        }

        // For additions, check if that sub has just been processed during initLeafNodeSmapAndSendSubs
        if delta > 0 && c.leaf.tsub != nil {
                if _, present := c.leaf.tsub[sub]; present {
                        delete(c.leaf.tsub, sub)
                        if len(c.leaf.tsub) == 0 {
                                c.leaf.tsub = nil
                                c.leaf.tsubt.Stop()
                                c.leaf.tsubt = nil
                        }
                        return
                }
        }

        key := keyFromSub(sub)
        n, ok := c.leaf.smap[key]
        if delta < 0 && !ok {
                return
        }

        // We will update if its a queue, if count is zero (or negative), or we were 0 and are N > 0.
        update := sub.queue != nil || (n <= 0 && n+delta > 0) || (n > 0 && n+delta <= 0)
        n += delta
        if n > 0 {
                c.leaf.smap[key] = n
        } else {
                delete(c.leaf.smap, key)
        }
        if update {
                c.sendLeafNodeSubUpdate(key, n)
        }
}

// Used to force add subjects to the subject map.
func (c *client) forceAddToSmap(subj string) {
        c.mu.Lock()
        defer c.mu.Unlock()

        if c.leaf.smap == nil {
                return
        }
        n := c.leaf.smap[subj]
        if n != 0 {
                return
        }
        // Place into the map since it was not there.
        c.leaf.smap[subj] = 1
        c.sendLeafNodeSubUpdate(subj, 1)
}

// Used to force remove a subject from the subject map.
func (c *client) forceRemoveFromSmap(subj string) {
        c.mu.Lock()
        defer c.mu.Unlock()

        if c.leaf.smap == nil {
                return
        }
        n := c.leaf.smap[subj]
        if n == 0 {
                return
        }
        n--
        if n == 0 {
                // Remove is now zero
                delete(c.leaf.smap, subj)
                c.sendLeafNodeSubUpdate(subj, 0)
        } else {
                c.leaf.smap[subj] = n
        }
}

// Send the subscription interest change to the other side.
// Lock should be held.
func (c *client) sendLeafNodeSubUpdate(key string, n int32) {
        // If we are a spoke, we need to check if we are allowed to send this subscription over to the hub.
        if c.isSpokeLeafNode() {
                checkPerms := true
                if len(key) > 0 && (key[0] == '$' || key[0] == '_') {
                        if strings.HasPrefix(key, leafNodeLoopDetectionSubjectPrefix) ||
                                strings.HasPrefix(key, oldGWReplyPrefix) ||
                                strings.HasPrefix(key, gwReplyPrefix) {
                                checkPerms = false
                        }
                }
                if checkPerms {
                        var subject string
                        if sep := strings.IndexByte(key, ' '); sep != -1 {
                                subject = key[:sep]
                        } else {
                                subject = key
                        }
                        if !c.canSubscribe(subject) {
                                return
                        }
                }
        }
        // If we are here we can send over to the other side.
        _b := [64]byte{}
        b := bytes.NewBuffer(_b[:0])
        c.writeLeafSub(b, key, n)
        c.enqueueProto(b.Bytes())
}

// Helper function to build the key.
func keyFromSub(sub *subscription) string {
        var sb strings.Builder
        sb.Grow(len(sub.subject) + len(sub.queue) + 1)
        sb.Write(sub.subject)
        if sub.queue != nil {
                // Just make the key subject spc group, e.g. 'foo bar'
                sb.WriteByte(' ')
                sb.Write(sub.queue)
        }
        return sb.String()
}

const (
        keyRoutedSub         = "R"
        keyRoutedSubByte     = 'R'
        keyRoutedLeafSub     = "L"
        keyRoutedLeafSubByte = 'L'
)

// Helper function to build the key that prevents collisions between normal
// routed subscriptions and routed subscriptions on behalf of a leafnode.
// Keys will look like this:
// "R foo"          -> plain routed sub on "foo"
// "R foo bar"      -> queue routed sub on "foo", queue "bar"
// "L foo bar"      -> plain routed leaf sub on "foo", leaf "bar"
// "L foo bar baz"  -> queue routed sub on "foo", queue "bar", leaf "baz"
func keyFromSubWithOrigin(sub *subscription) string {
        var sb strings.Builder
        sb.Grow(2 + len(sub.origin) + 1 + len(sub.subject) + 1 + len(sub.queue))
        leaf := len(sub.origin) > 0
        if leaf {
                sb.WriteByte(keyRoutedLeafSubByte)
        } else {
                sb.WriteByte(keyRoutedSubByte)
        }
        sb.WriteByte(' ')
        sb.Write(sub.subject)
        if sub.queue != nil {
                sb.WriteByte(' ')
                sb.Write(sub.queue)
        }
        if leaf {
                sb.WriteByte(' ')
                sb.Write(sub.origin)
        }
        return sb.String()
}

// Lock should be held.
func (c *client) writeLeafSub(w *bytes.Buffer, key string, n int32) {
        if key == _EMPTY_ {
                return
        }
        if n > 0 {
                w.WriteString("LS+ " + key)
                // Check for queue semantics, if found write n.
                if strings.Contains(key, " ") {
                        w.WriteString(" ")
                        var b [12]byte
                        var i = len(b)
                        for l := n; l > 0; l /= 10 {
                                i--
                                b[i] = digits[l%10]
                        }
                        w.Write(b[i:])
                        if c.trace {
                                arg := fmt.Sprintf("%s %d", key, n)
                                c.traceOutOp("LS+", []byte(arg))
                        }
                } else if c.trace {
                        c.traceOutOp("LS+", []byte(key))
                }
        } else {
                w.WriteString("LS- " + key)
                if c.trace {
                        c.traceOutOp("LS-", []byte(key))
                }
        }
        w.WriteString(CR_LF)
}

// processLeafSub will process an inbound sub request for the remote leaf node.
func (c *client) processLeafSub(argo []byte) (err error) {
        // Indicate activity.
        c.in.subs++

        srv := c.srv
        if srv == nil {
                return nil
        }

        // Copy so we do not reference a potentially large buffer
        arg := make([]byte, len(argo))
        copy(arg, argo)

        args := splitArg(arg)
        sub := &subscription{client: c}

        delta := int32(1)
        switch len(args) {
        case 1:
                sub.queue = nil
        case 3:
                sub.queue = args[1]
                sub.qw = int32(parseSize(args[2]))
                // TODO: (ik) We should have a non empty queue name and a queue
                // weight >= 1. For 2.11, we may want to return an error if that
                // is not the case, but for now just overwrite `delta` if queue
                // weight is greater than 1 (it is possible after a reconnect/
                // server restart to receive a queue weight > 1 for a new sub).
                if sub.qw > 1 {
                        delta = sub.qw
                }
        default:
                return fmt.Errorf("processLeafSub Parse Error: '%s'", arg)
        }
        sub.subject = args[0]

        c.mu.Lock()
        if c.isClosed() {
                c.mu.Unlock()
                return nil
        }

        acc := c.acc
        // Guard against LS+ arriving before CONNECT has been processed, which
        // can happen when compression is enabled.
        if acc == nil {
                c.mu.Unlock()
                c.sendErr("Authorization Violation")
                c.closeConnection(ProtocolViolation)
                return nil
        }
        // Check if we have a loop.
        ldsPrefix := bytes.HasPrefix(sub.subject, []byte(leafNodeLoopDetectionSubjectPrefix))

        if ldsPrefix && bytesToString(sub.subject) == acc.getLDSubject() {
                c.mu.Unlock()
                c.handleLeafNodeLoop(true)
                return nil
        }

        // Check permissions if applicable. (but exclude the $LDS, $GR and _GR_)
        checkPerms := true
        if sub.subject[0] == '$' || sub.subject[0] == '_' {
                if ldsPrefix ||
                        bytes.HasPrefix(sub.subject, []byte(oldGWReplyPrefix)) ||
                        bytes.HasPrefix(sub.subject, []byte(gwReplyPrefix)) {
                        checkPerms = false
                }
        }

        // If we are a hub check that we can publish to this subject.
        if checkPerms {
                subj := string(sub.subject)
                if subjectIsLiteral(subj) && !c.pubAllowedFullCheck(subj, true, true) {
                        c.mu.Unlock()
                        c.leafSubPermViolation(sub.subject)
                        c.Debugf(fmt.Sprintf("Permissions Violation for Subscription to %q", sub.subject))
                        return nil
                }
        }

        // Check if we have a maximum on the number of subscriptions.
        if c.subsAtLimit() {
                c.mu.Unlock()
                c.maxSubsExceeded()
                return nil
        }

        // If we have an origin cluster associated mark that in the sub.
        if rc := c.remoteCluster(); rc != _EMPTY_ {
                sub.origin = []byte(rc)
        }

        // Like Routes, we store local subs by account and subject and optionally queue name.
        // If we have a queue it will have a trailing weight which we do not want.
        if sub.queue != nil {
                sub.sid = arg[:len(arg)-len(args[2])-1]
        } else {
                sub.sid = arg
        }
        key := bytesToString(sub.sid)
        osub := c.subs[key]
        if osub == nil {
                c.subs[key] = sub
                // Now place into the account sl.
                if err := acc.sl.Insert(sub); err != nil {
                        delete(c.subs, key)
                        c.mu.Unlock()
                        c.Errorf("Could not insert subscription: %v", err)
                        c.sendErr("Invalid Subscription")
                        return nil
                }
        } else if sub.queue != nil {
                // For a queue we need to update the weight.
                delta = sub.qw - atomic.LoadInt32(&osub.qw)
                atomic.StoreInt32(&osub.qw, sub.qw)
                acc.sl.UpdateRemoteQSub(osub)
        }
        spoke := c.isSpokeLeafNode()
        c.mu.Unlock()

        // Only add in shadow subs if a new sub or qsub.
        if osub == nil {
                if err := c.addShadowSubscriptions(acc, sub); err != nil {
                        c.Errorf(err.Error())
                }
        }

        // If we are not solicited, treat leaf node subscriptions similar to a
        // client subscription, meaning we forward them to routes, gateways and
        // other leaf nodes as needed.
        if !spoke {
                // If we are routing add to the route map for the associated account.
                srv.updateRouteSubscriptionMap(acc, sub, delta)
                if srv.gateway.enabled {
                        srv.gatewayUpdateSubInterest(acc.Name, sub, delta)
                }
        }
        // Now check on leafnode updates for other leaf nodes. We understand solicited
        // and non-solicited state in this call so we will do the right thing.
        acc.updateLeafNodes(sub, delta)

        return nil
}

// If the leafnode is a solicited, set the connect delay based on default
// or private option (for tests). Sends the error to the other side, log and
// close the connection.
func (c *client) handleLeafNodeLoop(sendErr bool) {
        accName, delay := c.setLeafConnectDelayIfSoliciting(leafNodeReconnectDelayAfterLoopDetected)
        errTxt := fmt.Sprintf("Loop detected for leafnode account=%q. Delaying attempt to reconnect for %v", accName, delay)
        if sendErr {
                c.sendErr(errTxt)
        }

        c.Errorf(errTxt)
        // If we are here with "sendErr" false, it means that this is the server
        // that received the error. The other side will have closed the connection,
        // but does not hurt to close here too.
        c.closeConnection(ProtocolViolation)
}

// processLeafUnsub will process an inbound unsub request for the remote leaf node.
func (c *client) processLeafUnsub(arg []byte) error {
        // Indicate any activity, so pub and sub or unsubs.
        c.in.subs++

        srv := c.srv

        c.mu.Lock()
        if c.isClosed() {
                c.mu.Unlock()
                return nil
        }

        acc := c.acc
        // Guard against LS- arriving before CONNECT has been processed.
        if acc == nil {
                c.mu.Unlock()
                c.sendErr("Authorization Violation")
                c.closeConnection(ProtocolViolation)
                return nil
        }

        spoke := c.isSpokeLeafNode()
        // We store local subs by account and subject and optionally queue name.
        // LS- will have the arg exactly as the key.
        sub, ok := c.subs[string(arg)]
        if !ok {
                // If not found, don't try to update routes/gws/leaf nodes.
                c.mu.Unlock()
                return nil
        }
        delta := int32(1)
        if len(sub.queue) > 0 {
                delta = sub.qw
        }
        c.mu.Unlock()

        c.unsubscribe(acc, sub, true, true)
        if !spoke {
                // If we are routing subtract from the route map for the associated account.
                srv.updateRouteSubscriptionMap(acc, sub, -delta)
                // Gateways
                if srv.gateway.enabled {
                        srv.gatewayUpdateSubInterest(acc.Name, sub, -delta)
                }
        }
        // Now check on leafnode updates for other leaf nodes.
        acc.updateLeafNodes(sub, -delta)
        return nil
}

func (c *client) processLeafHeaderMsgArgs(arg []byte) error {
        // Unroll splitArgs to avoid runtime/heap issues
        args := c.argsa[:0]
        start := -1
        for i, b := range arg {
                switch b {
                case ' ', '\t', '\r', '\n':
                        if start >= 0 {
                                args = append(args, arg[start:i])
                                start = -1
                        }
                default:
                        if start < 0 {
                                start = i
                        }
                }
        }
        if start >= 0 {
                args = append(args, arg[start:])
        }

        c.pa.arg = arg
        switch len(args) {
        case 0, 1, 2:
                return fmt.Errorf("processLeafHeaderMsgArgs Parse Error: '%s'", args)
        case 3:
                c.pa.reply = nil
                c.pa.queues = nil
                c.pa.hdb = args[1]
                c.pa.hdr = parseSize(args[1])
                c.pa.szb = args[2]
                c.pa.size = parseSize(args[2])
        case 4:
                c.pa.reply = args[1]
                c.pa.queues = nil
                c.pa.hdb = args[2]
                c.pa.hdr = parseSize(args[2])
                c.pa.szb = args[3]
                c.pa.size = parseSize(args[3])
        default:
                // args[1] is our reply indicator. Should be + or | normally.
                if len(args[1]) != 1 {
                        return fmt.Errorf("processLeafHeaderMsgArgs Bad or Missing Reply Indicator: '%s'", args[1])
                }
                switch args[1][0] {
                case '+':
                        c.pa.reply = args[2]
                case '|':
                        c.pa.reply = nil
                default:
                        return fmt.Errorf("processLeafHeaderMsgArgs Bad or Missing Reply Indicator: '%s'", args[1])
                }
                // Grab header size.
                c.pa.hdb = args[len(args)-2]
                c.pa.hdr = parseSize(c.pa.hdb)

                // Grab size.
                c.pa.szb = args[len(args)-1]
                c.pa.size = parseSize(c.pa.szb)

                // Grab queue names.
                if c.pa.reply != nil {
                        c.pa.queues = args[3 : len(args)-2]
                } else {
                        c.pa.queues = args[2 : len(args)-2]
                }
        }
        if c.pa.hdr < 0 {
                return fmt.Errorf("processLeafHeaderMsgArgs Bad or Missing Header Size: '%s'", arg)
        }
        if c.pa.size < 0 {
                return fmt.Errorf("processLeafHeaderMsgArgs Bad or Missing Size: '%s'", args)
        }
        if c.pa.hdr > c.pa.size {
                return fmt.Errorf("processLeafHeaderMsgArgs Header Size larger then TotalSize: '%s'", arg)
        }
        maxPayload := atomic.LoadInt32(&c.mpay)
        if maxPayload != jwt.NoLimit && int64(c.pa.size) > int64(maxPayload) {
                c.maxPayloadViolation(c.pa.size, maxPayload)
                return ErrMaxPayload
        }

        // Common ones processed after check for arg length
        c.pa.subject = args[0]

        return nil
}

func (c *client) processLeafMsgArgs(arg []byte) error {
        // Unroll splitArgs to avoid runtime/heap issues
        args := c.argsa[:0]
        start := -1
        for i, b := range arg {
                switch b {
                case ' ', '\t', '\r', '\n':
                        if start >= 0 {
                                args = append(args, arg[start:i])
                                start = -1
                        }
                default:
                        if start < 0 {
                                start = i
                        }
                }
        }
        if start >= 0 {
                args = append(args, arg[start:])
        }

        c.pa.arg = arg
        switch len(args) {
        case 0, 1:
                return fmt.Errorf("processLeafMsgArgs Parse Error: '%s'", args)
        case 2:
                c.pa.reply = nil
                c.pa.queues = nil
                c.pa.szb = args[1]
                c.pa.size = parseSize(args[1])
        case 3:
                c.pa.reply = args[1]
                c.pa.queues = nil
                c.pa.szb = args[2]
                c.pa.size = parseSize(args[2])
        default:
                // args[1] is our reply indicator. Should be + or | normally.
                if len(args[1]) != 1 {
                        return fmt.Errorf("processLeafMsgArgs Bad or Missing Reply Indicator: '%s'", args[1])
                }
                switch args[1][0] {
                case '+':
                        c.pa.reply = args[2]
                case '|':
                        c.pa.reply = nil
                default:
                        return fmt.Errorf("processLeafMsgArgs Bad or Missing Reply Indicator: '%s'", args[1])
                }
                // Grab size.
                c.pa.szb = args[len(args)-1]
                c.pa.size = parseSize(c.pa.szb)

                // Grab queue names.
                if c.pa.reply != nil {
                        c.pa.queues = args[3 : len(args)-1]
                } else {
                        c.pa.queues = args[2 : len(args)-1]
                }
        }
        if c.pa.size < 0 {
                return fmt.Errorf("processLeafMsgArgs Bad or Missing Size: '%s'", args)
        }
        maxPayload := atomic.LoadInt32(&c.mpay)
        if maxPayload != jwt.NoLimit && int64(c.pa.size) > int64(maxPayload) {
                c.maxPayloadViolation(c.pa.size, maxPayload)
                return ErrMaxPayload
        }

        // Common ones processed after check for arg length
        c.pa.subject = args[0]

        return nil
}

// processInboundLeafMsg is called to process an inbound msg from a leaf node.
func (c *client) processInboundLeafMsg(msg []byte) {
        // Update statistics
        // The msg includes the CR_LF, so pull back out for accounting.
        c.in.msgs++
        c.in.bytes += int32(len(msg) - LEN_CR_LF)

        srv, acc, subject := c.srv, c.acc, string(c.pa.subject)

        // Mostly under testing scenarios.
        if srv == nil || acc == nil {
                return
        }

        // Check that leaf messages respect the subject permissions.
        if c.perms != nil && !c.leafMsgAllowed() {
                c.leafPubPermViolation(c.pa.subject)
                return
        }

        // Match the subscriptions. We will use our own L1 map if
        // it's still valid, avoiding contention on the shared sublist.
        var r *SublistResult
        var ok bool

        genid := atomic.LoadUint64(&c.acc.sl.genid)
        if genid == c.in.genid && c.in.results != nil {
                r, ok = c.in.results[subject]
        } else {
                // Reset our L1 completely.
                c.in.results = make(map[string]*SublistResult)
                c.in.genid = genid
        }

        // Go back to the sublist data structure.
        if !ok {
                r = c.acc.sl.Match(subject)
                // Prune the results cache. Keeps us from unbounded growth. Random delete.
                if len(c.in.results) >= maxResultCacheSize {
                        n := 0
                        for subj := range c.in.results {
                                delete(c.in.results, subj)
                                if n++; n > pruneSize {
                                        break
                                }
                        }
                }
                // Then add the new cache entry.
                c.in.results[subject] = r
        }

        // Collect queue names if needed.
        var qnames [][]byte

        // Check for no interest, short circuit if so.
        // This is the fanout scale.
        if len(r.psubs)+len(r.qsubs) > 0 {
                flag := pmrNoFlag
                // If we have queue subs in this cluster, then if we run in gateway
                // mode and the remote gateways have queue subs, then we need to
                // collect the queue groups this message was sent to so that we
                // exclude them when sending to gateways.
                if len(r.qsubs) > 0 && c.srv.gateway.enabled &&
                        atomic.LoadInt64(&c.srv.gateway.totalQSubs) > 0 {
                        flag |= pmrCollectQueueNames
                }
                // If this is a mapped subject that means the mapped interest
                // is what got us here, but this might not have a queue designation
                // If that is the case, make sure we ignore to process local queue subscribers.
                if len(c.pa.mapped) > 0 && len(c.pa.queues) == 0 {
                        flag |= pmrIgnoreEmptyQueueFilter
                }
                _, qnames = c.processMsgResults(acc, r, msg, nil, c.pa.subject, c.pa.reply, flag)
        }

        // Now deal with gateways
        if c.srv.gateway.enabled {
                c.sendMsgToGateways(acc, msg, c.pa.subject, c.pa.reply, qnames, true)
        }
}

// Checks whether the inbound leaf message is allowed by the
// connection's permissions. On the hub side this enforces what
// the remote leaf may publish. On the spoke side this enforces
// import restrictions such as deny_imports.
func (c *client) leafMsgAllowed() bool {
        wireSubject := c.pa.subject
        if len(c.pa.mapped) > 0 {
                // Mappings rewrite c.pa.subject to the internal
                // destination. For leaf ACLs, need to check
                // the original wire subject from the remote side.
                wireSubject = c.pa.mapped
        }
        // Strip any gateway routing prefix for the permission check.
        subjectToCheck, isGW := getGWRoutedSubjectOrSelf(wireSubject)

        // Service-import replies (_R_), JS ack subjects ($JS.ACK.)
        // are internal routing subjects forwarded via LS+ without
        // permission checks.
        if isServiceReply(subjectToCheck) || isJSAckSubject(subjectToCheck) {
                return true
        }

        c.mu.RLock()
        if c.isSpokeLeafNode() {
                // Gateway routed replies are forwarded without
                // permission checks.
                if isGW || c.leafReceiveAllowed(subjectToCheck) {
                        c.mu.RUnlock()
                        return true
                }
        } else if c.leafSendAllowed(subjectToCheck) {
                c.mu.RUnlock()
                return true
        }

        // If allow_responses is not configured, or there is no tracked reply for
        // this subject, the answer is "denied" and we can return it while still
        // holding only the read lock.
        replySubject := bytesToString(wireSubject)
        if c.perms == nil || c.perms.resp == nil || c.replies[replySubject] == nil {
                c.mu.RUnlock()
                return false
        }
        c.mu.RUnlock()

        // Check tracked reply permissions (allow_responses).
        // Use the pre-strip subject since deliverMsg tracks
        // replies under the original form, which includes
        // the GW routing prefix for routed requests.
        c.mu.Lock()
        defer c.mu.Unlock()
        return c.responseAllowed(replySubject)
}

// Returns true if the leaf side ACLs allow importing this subject,
// based on the permissions received over INFO and any local deny_imports.
// At least a read lock must be held.
func (c *client) leafReceiveAllowed(subject []byte) bool {
        return c.canSubscribeInternal(bytesToString(subject))
}

// Returns true if the hub side ACLs allow the remote leaf to send
// this subject.
// At least a read lock must be held.
func (c *client) leafSendAllowed(bsubject []byte) bool {
        // Use the original export ACL captured for this accepted leaf.
        // The live perms also contain additional JetStream denies used by
        // the normal forwarding path, and applying them here would reject
        // legitimate inbound JS API requests.
        subject := bytesToString(bsubject)
        perms := c.opts.Export
        if perms == nil || (perms.Allow == nil && perms.Deny == nil) {
                return true
        }

        allowed := true
        if perms.Allow != nil && !strings.HasPrefix(subject, mqttPrefix) {
                allowed = false
                for _, allowSubj := range perms.Allow {
                        if matchLiteral(subject, allowSubj) {
                                allowed = true
                                break
                        }
                }
        }

        if allowed && len(perms.Deny) > 0 {
                for _, denySubj := range perms.Deny {
                        if matchLiteral(subject, denySubj) {
                                allowed = false
                                break
                        }
                }
        }
        return allowed
}

// Handles a subscription permission violation.
// See leafPermViolation() for details.
func (c *client) leafSubPermViolation(subj []byte) {
        c.leafPermViolation(false, subj)
}

// Handles a publish permission violation.
// See leafPermViolation() for details.
func (c *client) leafPubPermViolation(subj []byte) {
        c.leafPermViolation(true, subj)
}

// Common function to process publish or subscribe leafnode permission violation.
// Sends the permission violation error to the remote, logs it and closes the connection.
// If this is from a server soliciting, the reconnection will be delayed.
func (c *client) leafPermViolation(pub bool, subj []byte) {
        if c.isSpokeLeafNode() {
                // For spokes these are no-ops since the hub server told us our permissions.
                // We just need to not send these over to the other side since we will get cutoff.
                return
        }
        // FIXME(dlc) ?
        c.setLeafConnectDelayIfSoliciting(leafNodeReconnectAfterPermViolation)
        var action string
        if pub {
                c.sendErr(fmt.Sprintf("Permissions Violation for Publish to %q", subj))
                action = "Publish"
        } else {
                c.sendErr(fmt.Sprintf("Permissions Violation for Subscription to %q", subj))
                action = "Subscription"
        }
        c.Errorf("%s Violation on %q - Check other side configuration", action, subj)
        // TODO: add a new close reason that is more appropriate?
        c.closeConnection(ProtocolViolation)
}

// Invoked from generic processErr() for LEAF connections.
func (c *client) leafProcessErr(errStr string) {
        // Check if we got a cluster name collision.
        if strings.Contains(errStr, ErrLeafNodeHasSameClusterName.Error()) {
                _, delay := c.setLeafConnectDelayIfSoliciting(leafNodeReconnectDelayAfterClusterNameSame)
                c.Errorf("Leafnode connection dropped with same cluster name error. Delaying attempt to reconnect for %v", delay)
                return
        }
        if strings.Contains(errStr, ErrLeafNodeMinVersionRejected.Error()) {
                _, delay := c.setLeafConnectDelayIfSoliciting(leafNodeMinVersionReconnectDelay)
                c.Errorf("Leafnode connection dropped due to minimum version requirement. Delaying attempt to reconnect for %v", delay)
                return
        }

        // We will look for Loop detected error coming from the other side.
        // If we solicit, set the connect delay.
        if !strings.Contains(errStr, "Loop detected") {
                return
        }
        c.handleLeafNodeLoop(false)
}

// If this leaf connection solicits, sets the connect delay to the given value,
// or the one from the server option's LeafNode.connDelay if one is set (for tests).
// Returns the connection's account name and delay.
func (c *client) setLeafConnectDelayIfSoliciting(delay time.Duration) (string, time.Duration) {
        c.mu.Lock()
        if c.isSolicitedLeafNode() {
                if s := c.srv; s != nil {
                        if srvdelay := s.getOpts().LeafNode.connDelay; srvdelay != 0 {
                                delay = srvdelay
                        }
                }
                c.leaf.remote.setConnectDelay(delay)
        }
        var accName string
        if c.acc != nil {
                accName = c.acc.Name
        }
        c.mu.Unlock()
        return accName, delay
}

// For the given remote Leafnode configuration, this function returns
// if TLS is required, and if so, will return a clone of the TLS Config
// (since some fields will be changed during handshake), the TLS server
// name that is remembered, and the TLS timeout.
func (c *client) leafNodeGetTLSConfigForSolicit(remote *leafNodeCfg) (bool, *tls.Config, string, float64) {
        var (
                tlsConfig  *tls.Config
                tlsName    string
                tlsTimeout float64
        )

        remote.RLock()
        defer remote.RUnlock()

        tlsRequired := remote.TLS || remote.TLSConfig != nil
        if tlsRequired {
                if remote.TLSConfig != nil {
                        tlsConfig = remote.TLSConfig.Clone()
                } else {
                        tlsConfig = &tls.Config{MinVersion: tls.VersionTLS12}
                }
                tlsName = remote.tlsName
                tlsTimeout = remote.TLSTimeout
                if tlsTimeout == 0 {
                        tlsTimeout = float64(TLS_TIMEOUT / time.Second)
                }
        }

        return tlsRequired, tlsConfig, tlsName, tlsTimeout
}

// Initiates the LeafNode Websocket connection by:
// - doing the TLS handshake if needed
// - sending the HTTP request
// - waiting for the HTTP response
//
// Since some bufio reader is used to consume the HTTP response, this function
// returns the slice of buffered bytes (if any) so that the readLoop that will
// be started after that consume those first before reading from the socket.
// The boolean
//
// Lock held on entry.
func (c *client) leafNodeSolicitWSConnection(opts *Options, rURL *url.URL, remote *leafNodeCfg) ([]byte, ClosedState, error) {
        remote.RLock()
        compress := remote.Websocket.Compression
        // By default the server will mask outbound frames, but it can be disabled with this option.
        noMasking := remote.Websocket.NoMasking
        infoTimeout := remote.FirstInfoTimeout
        remote.RUnlock()
        // Will do the client-side TLS handshake if needed.
        tlsRequired, err := c.leafClientHandshakeIfNeeded(remote, opts)
        if err != nil {
                // 0 will indicate that the connection was already closed
                return nil, 0, err
        }

        // For http request, we need the passed URL to contain either http or https scheme.
        scheme := "http"
        if tlsRequired {
                scheme = "https"
        }
        // We will use the `/leafnode` path to tell the accepting WS server that it should
        // create a LEAF connection, not a CLIENT.
        // In case we use the user's URL path in the future, make sure we append the user's
        // path to our `/leafnode` path.
        lpath := leafNodeWSPath
        if curPath := rURL.EscapedPath(); curPath != _EMPTY_ {
                if curPath[0] == '/' {
                        curPath = curPath[1:]
                }
                lpath = path.Join(curPath, lpath)
        } else {
                lpath = lpath[1:]
        }
        ustr := fmt.Sprintf("%s://%s/%s", scheme, rURL.Host, lpath)
        u, _ := url.Parse(ustr)
        req := &http.Request{
                Method:     "GET",
                URL:        u,
                Proto:      "HTTP/1.1",
                ProtoMajor: 1,
                ProtoMinor: 1,
                Header:     make(http.Header),
                Host:       u.Host,
        }
        wsKey, err := wsMakeChallengeKey()
        if err != nil {
                return nil, WriteError, err
        }

        req.Header["Upgrade"] = []string{"websocket"}
        req.Header["Connection"] = []string{"Upgrade"}
        req.Header["Sec-WebSocket-Key"] = []string{wsKey}
        req.Header["Sec-WebSocket-Version"] = []string{"13"}
        if compress {
                req.Header.Add("Sec-WebSocket-Extensions", wsPMCReqHeaderValue)
        }
        if noMasking {
                req.Header.Add(wsNoMaskingHeader, wsNoMaskingValue)
        }
        c.nc.SetDeadline(time.Now().Add(infoTimeout))
        if err := req.Write(c.nc); err != nil {
                return nil, WriteError, err
        }

        var resp *http.Response

        br := bufio.NewReaderSize(c.nc, MAX_CONTROL_LINE_SIZE)
        resp, err = http.ReadResponse(br, req)
        if err == nil &&
                (resp.StatusCode != 101 ||
                        !strings.EqualFold(resp.Header.Get("Upgrade"), "websocket") ||
                        !strings.EqualFold(resp.Header.Get("Connection"), "upgrade") ||
                        resp.Header.Get("Sec-Websocket-Accept") != wsAcceptKey(wsKey)) {

                err = fmt.Errorf("invalid websocket connection")
        }
        // Check compression extension...
        if err == nil && c.ws.compress {
                // Check that not only permessage-deflate extension is present, but that
                // we also have server and client no context take over.
                srvCompress, noCtxTakeover := wsPMCExtensionSupport(resp.Header, false)

                // If server does not support compression, then simply disable it in our side.
                if !srvCompress {
                        c.ws.compress = false
                } else if !noCtxTakeover {
                        err = fmt.Errorf("compression negotiation error")
                }
        }
        // Same for no masking...
        if err == nil && noMasking {
                // Check if server accepts no masking
                if resp.Header.Get(wsNoMaskingHeader) != wsNoMaskingValue {
                        // Nope, need to mask our writes as any client would do.
                        c.ws.maskwrite = true
                }
        }
        if resp != nil {
                resp.Body.Close()
        }
        if err != nil {
                return nil, ReadError, err
        }
        c.Debugf("Leafnode compression=%v masking=%v", c.ws.compress, c.ws.maskwrite)

        var preBuf []byte
        // We have to slurp whatever is in the bufio reader and pass that to the readloop.
        if n := br.Buffered(); n != 0 {
                preBuf, _ = br.Peek(n)
        }
        return preBuf, 0, nil
}

const connectProcessTimeout = 2 * time.Second

// This is invoked for remote LEAF remote connections after processing the INFO
// protocol.
func (s *Server) leafNodeResumeConnectProcess(c *client) {
        clusterName := s.ClusterName()

        c.mu.Lock()
        if c.isClosed() {
                c.mu.Unlock()
                return
        }
        if err := c.sendLeafConnect(clusterName, c.headers); err != nil {
                c.mu.Unlock()
                c.closeConnection(WriteError)
                return
        }

        // Spin up the write loop.
        s.startGoRoutine(func() { c.writeLoop() })

        // timeout leafNodeFinishConnectProcess
        c.ping.tmr = time.AfterFunc(connectProcessTimeout, func() {
                c.mu.Lock()
                // check if leafNodeFinishConnectProcess was called and prevent later leafNodeFinishConnectProcess
                if !c.flags.setIfNotSet(connectProcessFinished) {
                        c.mu.Unlock()
                        return
                }
                clearTimer(&c.ping.tmr)
                closed := c.isClosed()
                c.mu.Unlock()
                if !closed {
                        c.sendErrAndDebug("Stale Leaf Node Connection - Closing")
                        c.closeConnection(StaleConnection)
                }
        })
        c.mu.Unlock()
        c.Debugf("Remote leafnode connect msg sent")
}

// This is invoked for remote LEAF connections after processing the INFO
// protocol and leafNodeResumeConnectProcess.
// This will send LS+ the CONNECT protocol and register the leaf node.
func (s *Server) leafNodeFinishConnectProcess(c *client) {
        c.mu.Lock()
        if !c.flags.setIfNotSet(connectProcessFinished) {
                c.mu.Unlock()
                return
        }
        if c.isClosed() {
                c.mu.Unlock()
                s.removeLeafNodeConnection(c)
                return
        }
        remote := c.leaf.remote
        if remote == nil || c.acc == nil {
                c.mu.Unlock()
                c.sendErr("Authorization Violation")
                c.closeConnection(ProtocolViolation)
                return
        }
        // Check if we will need to send the system connect event.
        remote.RLock()
        sendSysConnectEvent := remote.Hub
        remote.RUnlock()

        // Capture account before releasing lock
        acc := c.acc
        // cancel connectProcessTimeout
        clearTimer(&c.ping.tmr)
        c.mu.Unlock()

        // Make sure we register with the account here.
        if err := c.registerWithAccount(acc); err != nil {
                if err == ErrTooManyAccountConnections {
                        c.maxAccountConnExceeded()
                        return
                } else if err == ErrLeafNodeLoop {
                        c.handleLeafNodeLoop(true)
                        return
                }
                c.Errorf("Registering leaf with account %s resulted in error: %v", acc.Name, err)
                c.closeConnection(ProtocolViolation)
                return
        }
        if !s.addLeafNodeConnection(c, _EMPTY_, _EMPTY_, false) {
                // Was not added, could be because the remote configuration has been removed.
                c.closeConnection(ClientClosed)
                return
        }
        s.initLeafNodeSmapAndSendSubs(c)
        if sendSysConnectEvent {
                s.sendLeafNodeConnect(acc)
        }
        s.accountConnectEvent(c)

        // The above functions are not running under the client lock, so it is
        // possible that between the time we have started the read/write loops
        // and now, that the connection was closed. This would leave the closed
        // LN connection possibly registered with the account and/or the server's
        // leafs map. So check if connection is closed, and if so, manually cleanup.
        c.mu.Lock()
        closed := c.isClosed()
        if !closed {
                c.setFirstPingTimer()
        }
        c.mu.Unlock()
        if closed {
                s.removeLeafNodeConnection(c)
                if prev := acc.removeClient(c); prev == 1 {
                        s.decActiveAccounts()
                }
        }
}

nats-io / nats-server / 30189872764

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous