leafnode.go

// Copyright 2019-2025 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"
	"reflect"
	"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"

	// This is the time the server will wait, when receiving a CONNECT,
	// before closing the connection if the required minimum version is not met.
	leafNodeWaitBeforeClose = 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
}

// 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.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.isSpoke
}

func (c *client) isHubLeafNode() bool {
	return c.kind == LEAF && !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.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
		s.leafRemoteCfgs = append(s.leafRemoteCfgs, remote)
		// 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.startGoRoutine(func() { s.connectToRemoteLeafNode(remote, true) })
		}
	}
}

func (s *Server) remoteLeafNodeStillValid(remote *leafNodeCfg) bool {
	if remote.Disabled {
		return false
	}
	for _, ri := range s.getOpts().LeafNode.Remotes {
		// FIXME(dlc) - What about auth changes?
		if reflect.DeepEqual(ri.URLs, remote.URLs) {
			return true
		}
	}
	return false
}

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

	// 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
		}
	}

	// 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))
			}
		}
		// 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
}

// Update remote LeafNode TLS configurations after a config reload.
func (s *Server) updateRemoteLeafNodesTLSConfig(opts *Options) {
	max := len(opts.LeafNode.Remotes)
	if max == 0 {
		return
	}

	s.mu.RLock()
	defer s.mu.RUnlock()

	// Changes in the list of remote leaf nodes is not supported.
	// However, make sure that we don't go over the arrays.
	if len(s.leafRemoteCfgs) < max {
		max = len(s.leafRemoteCfgs)
	}
	for i := 0; i < max; i++ {
		ro := opts.LeafNode.Remotes[i]
		cfg := s.leafRemoteCfgs[i]
		if ro.TLSConfig != nil {
			cfg.Lock()
			cfg.TLSConfig = ro.TLSConfig.Clone()
			cfg.TLSHandshakeFirst = ro.TLSHandshakeFirst
			cfg.Unlock()
		}
	}
}

func (s *Server) reConnectToRemoteLeafNode(remote *leafNodeCfg) {
	delay := s.getOpts().LeafNode.ReconnectInterval
	select {
	case <-time.After(delay):
	case <-s.quitCh:
		s.grWG.Done()
		return
	}
	s.connectToRemoteLeafNode(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)),
	}
	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
}

// 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
}

func (s *Server) connectToRemoteLeafNode(remote *leafNodeCfg, firstConnect bool) {
	defer s.grWG.Done()

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

	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 <-s.quitCh:
			return
		}
		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

	for s.isRunning() && s.remoteLeafNodeStillValid(remote) {
		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 first, then check if URL supports it
				if proxyURL != _EMPTY_ && isWSURL(rURL) {
					// Use proxy for WebSocket connections - use original hostname, resolved IP for connection
					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()
			select {
			case <-s.quitCh:
				remote.cancelMigrateTimer()
				return
			case <-time.After(delay):
				// 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()
		if !s.remoteLeafNodeStillValid(remote) {
			conn.Close()
			return
		}

		// 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
	}
}

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.
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.
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.
		if cm := opts.LeafNode.Compression.Mode; cm != CompressionNotSupported {
			info.Compression = cm
		}
		// We always send a nonce for LEAF connections. Do not change that without
		// taking into account presence of proxy trusted keys.
		s.generateNonce(nonce[:])
		s.mu.Unlock()
	}

	// 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) {
		// 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.
		//
		// As seen from above, a solicited LeafNode connection should receive
		// from the remote server an INFO with CID and LeafNodeURLs. Anything
		// else should be considered an attempt to connect to a wrong port.
		if didSolicit && (info.CID == 0 || info.LeafNodeURLs == nil) {
			c.mu.Unlock()
			c.Errorf(ErrConnectedToWrongPort.Error())
			c.closeConnection(WrongPort)
			return
		}
		// 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
		}
		// Capture a nonce here.
		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.
	if didSolicit && (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)
	}

	// Check to see if we have permissions updates here.
	if 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_ {
		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) {
	// 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) {
	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
	solicited := c.leaf.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
			}
		}
	}
	// 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
		}
	}

	// 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(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() {
			meta.setObserver(true, extExtended)
			c.Debugf("Turning JetStream metadata controller Observer Mode on - System Account Connected")
			// Take note that the domain was not extended to avoid this state next startup.
			writePeerState(js.config.StoreDir, meta.currentPeerState())
			// If this server is the leader already, step down so a new leader can be elected (that is not an observer)
			meta.StepDown()
		}
	} 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)
		}
	}
}

func (s *Server) removeLeafNodeConnection(c *client) {
	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
		}
	}
	proxyKey := c.proxyKey
	c.mu.Unlock()
	s.mu.Lock()
	delete(s.leafs, cid)
	if proxyKey != _EMPTY_ {
		s.removeProxiedConn(proxyKey, cid)
	}
	s.mu.Unlock()
	s.removeFromTempClients(cid)
}

// 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) {
			// We are going to send back an INFO because otherwise recent
			// versions of the remote server would simply break the connection
			// after 2 seconds if not receiving it. Instead, we want the
			// other side to just "stall" until we finish waiting for the holding
			// period and close the connection below.
			s.sendPermsAndAccountInfo(c)
			c.sendErrAndErr(fmt.Sprintf("connection rejected since minimum version required is %q", mv))
			select {
			case <-c.srv.quitCh:
			case <-time.After(leafNodeWaitBeforeClose):
			}
			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()

	// 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.
	for _, ln := range acc.lleafs {
		if ln == sub.client {
			continue
		}
		ln.mu.Lock()
		// Don't advertise interest from leafnodes to other isolated leafnodes.
		if sub.client.kind == LEAF && ln.isIsolatedLeafNode() {
			ln.mu.Unlock()
			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.Unlock()
			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()
		if (isLDS && clusterDifferent) || ((cluster == _EMPTY_ || clusterDifferent) && (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
	// 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++

	acc := c.acc
	srv := c.srv

	c.mu.Lock()
	if c.isClosed() {
		c.mu.Unlock()
		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
	a := [MAX_MSG_ARGS][]byte{}
	args := a[: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)
	}

	// 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
	a := [MAX_MSG_ARGS][]byte{}
	args := a[: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)
	}

	// 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
	}

	// 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)
	}
}

// Handles a subscription permission violation.
// See leafPermViolation() for details.
func (c *client) leafSubPermViolation(subj []byte) {
	c.leafPermViolation(false, 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
	}

	// 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)
	}
	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
	// 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
	}
	s.addLeafNodeConnection(c, _EMPTY_, _EMPTY_, false)
	s.initLeafNodeSmapAndSendSubs(c)
	if sendSysConnectEvent {
		s.sendLeafNodeConnect(acc)
	}

	// The above functions are not atomically under the client
	// lock doing those operations. It is possible - since we
	// have started the read/write loops - that the connection
	// is closed before or in between. This would leave the
	// closed LN connection possible 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()
		}
	}
}