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
)

typeleafstruct {
// 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.
isSpokebool
// remoteCluster is when we are a hub but the spoke leafnode is part of a cluster.
remoteClusterstring
// remoteServer holds onto the remove server's name or ID.
remoteServerstring
// domain name of remote server
remoteDomainstring
// account name of remote server
remoteAccNamestring
// 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.
smapmap[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-.
tsubmap[*subscription]struct{}
tsubt*time.Timer
// Selected compression mode, which may be different from the server configured mode.
compressionstring
// This is for GW map replies.
gwSub*subscription
}

// Used for remote (solicited) leafnodes.
typeleafNodeCfgstruct {
 sync.RWMutex
*RemoteLeafOpts
urls []*url.URL
curURL*url.URL
tlsNamestring
usernamestring
passwordstring
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 {
returnc.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 {
returnc.kind==LEAF&&c.leaf.isSpoke
}

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

// This will spin up go routines to solicit the remote leaf node connections.
func (s*Server) solicitLeafNodeRemotes(remotes []*RemoteLeafOpts) {
sysAccName:=_EMPTY_
sAcc:=s.SystemAccount()
ifsAcc!=nil {
sysAccName=sAcc.Name
 }
addRemote:=func(r*RemoteLeafOpts, isSysAccRemotebool) *leafNodeCfg {
s.mu.Lock()
remote:=newLeafNodeCfg(r)
creds:=remote.Credentials
accName:=remote.LocalAccount
s.leafRemoteCfgs=append(s.leafRemoteCfgs, remote)
// Print notice if
ifisSysAccRemote {
iflen(remote.DenyExports) >0 {
s.Noticef("Remote for System Account uses restricted export permissions")
 }
iflen(remote.DenyImports) >0 {
s.Noticef("Remote for System Account uses restricted import permissions")
 }
 }
s.mu.Unlock()
ifcreds!=_EMPTY_ {
contents, err:=os.ReadFile(creds)
deferwipeSlice(contents)
iferr!=nil {
s.Errorf("Error reading LeafNode Remote Credentials file %q: %v", creds, err)
 } elseifitems:=credsRe.FindAllSubmatch(contents, -1); len(items) <2 {
s.Errorf("LeafNode Remote Credentials file %q malformed", creds)
 } elseif_, err:=nkeys.FromSeed(items[1][1]); err!=nil {
s.Errorf("LeafNode Remote Credentials file %q has malformed seed", creds)
 } elseifuc, err:=jwt.DecodeUserClaims(string(items[0][1])); err!=nil {
s.Errorf("LeafNode Remote Credentials file %q has malformed user jwt", creds)
 } elseifisSysAccRemote {
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)
 }
 }
 }
returnremote
 }
for_, r:=rangeremotes {
remote:=addRemote(r, r.LocalAccount==sysAccName)
s.startGoRoutine(func() { s.connectToRemoteLeafNode(remote, true) })
 }
}

func (s*Server) remoteLeafNodeStillValid(remote*leafNodeCfg) bool {
for_, ri:=ranges.getOpts().LeafNode.Remotes {
// FIXME(dlc) - What about auth changes?
ifreflect.DeepEqual(ri.URLs, remote.URLs) {
returntrue
 }
 }
returnfalse
}

// Ensure that leafnode is properly configured.
funcvalidateLeafNode(o*Options) error {
iferr:=validateLeafNodeAuthOptions(o); err!=nil {
returnerr
 }

// Users can bind to any local account, if its empty we will assume the $G account.
for_, r:=rangeo.LeafNode.Remotes {
ifr.LocalAccount==_EMPTY_ {
r.LocalAccount=globalAccountName
 }
 }

// In local config mode, check that leafnode configuration refers to accounts that exist.
iflen(o.TrustedOperators) ==0 {
accNames:=map[string]struct{}{}
for_, a:=rangeo.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
ifo.SystemAccount==_EMPTY_&&!o.NoSystemAccount {
accNames[DEFAULT_SYSTEM_ACCOUNT] =struct{}{}
 }
checkAccountExists:=func(accNamestring, cfgTypestring) error {
if_, ok:=accNames[accName]; !ok {
returnfmt.Errorf("cannot find local account %q specified in leafnode %s", accName, cfgType)
 }
returnnil
 }
iferr:=checkAccountExists(o.LeafNode.Account, "authorization"); err!=nil {
returnerr
 }
for_, lu:=rangeo.LeafNode.Users {
iflu.Account==nil { // means global account
continue
 }
iferr:=checkAccountExists(lu.Account.Name, "authorization"); err!=nil {
returnerr
 }
 }
for_, r:=rangeo.LeafNode.Remotes {
iferr:=checkAccountExists(r.LocalAccount, "remote"); err!=nil {
returnerr
 }
 }
 } else {
iflen(o.LeafNode.Users) !=0 {
returnfmt.Errorf("operator mode does not allow specifying users in leafnode config")
 }
for_, r:=rangeo.LeafNode.Remotes {
if!nkeys.IsValidPublicAccountKey(r.LocalAccount) {
returnfmt.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'")
 }
 }
ifo.LeafNode.Port!=0&&o.LeafNode.Account!=""&&!nkeys.IsValidPublicAccountKey(o.LeafNode.Account) {
returnfmt.Errorf("operator mode and non account nkeys are incompatible")
 }
 }

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

// If a remote has a websocket scheme, all need to have it.
for_, rcfg:=rangeo.LeafNode.Remotes {
iflen(rcfg.URLs) >=2 {
firstIsWS, ok:=isWSURL(rcfg.URLs[0]), true
fori:=1; i<len(rcfg.URLs); i++ {
u:=rcfg.URLs[i]
ifisWS:=isWSURL(u); isWS&&!firstIsWS||!isWS&&firstIsWS {
ok=false
break
 }
 }
if!ok {
returnfmt.Errorf("remote leaf node configuration cannot have a mix of websocket and non-websocket urls: %q", redactURLList(rcfg.URLs))
 }
 }
// Validate compression settings
ifrcfg.Compression.Mode!=_EMPTY_ {
iferr:=validateAndNormalizeCompressionOption(&rcfg.Compression, CompressionS2Auto); err!=nil {
returnerr
 }
 }
 }

ifo.LeafNode.Port==0 {
returnnil
 }

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

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

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

funccheckLeafMinVersionConfig(mvstring) error {
ifok, err:=versionAtLeastCheckError(mv, 2, 8, 0); !ok||err!=nil {
iferr!=nil {
returnfmt.Errorf("invalid leafnode's minimum version: %v", err)
 } else {
returnfmt.Errorf("the minimum version should be at least 2.8.0")
 }
 }
returnnil
}

// Used to validate user names in LeafNode configuration.
// - rejects mix of single and multiple users.
// - rejects duplicate user names.
funcvalidateLeafNodeAuthOptions(o*Options) error {
iflen(o.LeafNode.Users) ==0 {
returnnil
 }
ifo.LeafNode.Username!=_EMPTY_ {
returnfmt.Errorf("can not have a single user/pass and a users array")
 }
ifo.LeafNode.Nkey!=_EMPTY_ {
returnfmt.Errorf("can not have a single nkey and a users array")
 }
users:=map[string]struct{}{}
for_, u:=rangeo.LeafNode.Users {
if_, exists:=users[u.Username]; exists {
returnfmt.Errorf("duplicate user %q detected in leafnode authorization", u.Username)
 }
users[u.Username] =struct{}{}
 }
returnnil
}

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

s.mu.RLock()
defers.mu.RUnlock()

// Changes in the list of remote leaf nodes is not supported.
// However, make sure that we don't go over the arrays.
iflen(s.leafRemoteCfgs) <max {
max=len(s.leafRemoteCfgs)
 }
fori:=0; i<max; i++ {
ro:=opts.LeafNode.Remotes[i]
cfg:=s.leafRemoteCfgs[i]
ifro.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.
funcnewLeafNodeCfg(remote*RemoteLeafOpts) *leafNodeCfg {
cfg:=&leafNodeCfg{
RemoteLeafOpts: remote,
urls: make([]*url.URL, 0, len(remote.URLs)),
 }
iflen(remote.DenyExports) >0||len(remote.DenyImports) >0 {
perms:=&Permissions{}
iflen(remote.DenyExports) >0 {
perms.Publish=&SubjectPermission{Deny: remote.DenyExports}
 }
iflen(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, jint) {
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:=rangecfg.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
 }
 }
returncfg
}

// Will pick an URL from the list of available URLs.
func (cfg*leafNodeCfg) pickNextURL() *url.URL {
cfg.Lock()
defercfg.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.
ifcfg.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]
returncfg.curURL
}

// Returns the current URL
func (cfg*leafNodeCfg) getCurrentURL() *url.URL {
cfg.RLock()
defercfg.RUnlock()
returncfg.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()
returndelay
}

// 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
ifs.leafNodeOpts.dialTimeout==0 {
// Use same timeouts as routes for now.
s.leafNodeOpts.dialTimeout=DEFAULT_ROUTE_DIAL
 }
s.leafNodeOpts.resolver=opts.LeafNode.resolver
ifs.leafNodeOpts.resolver==nil {
s.leafNodeOpts.resolver=net.DefaultResolver
 }
}

constsharedSysAccDelay=250*time.Millisecond

func (s*Server) connectToRemoteLeafNode(remote*leafNodeCfg, firstConnectbool) {
defers.grWG.Done()

ifremote==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
varisSysAccbool
ifs.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.
iffirstConnect&&isSysAcc&&!s.standAloneMode() {
s.Debugf("Will delay first leafnode connect to shared system account due to clustering")
remote.setConnectDelay(sharedSysAccDelay)
 }

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

varconn net.Conn

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

attempts:=0

fors.isRunning() &&s.remoteLeafNodeStillValid(remote) {
rURL:=remote.pickNextURL()
url, err:=s.getRandomIP(resolver, rURL.Host, nil)
iferr==nil {
varipStrstring
ifurl!=rURL.Host {
ipStr=fmt.Sprintf(" (%s)", url)
 }
// Some test may want to disable remotes from connecting
ifs.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)
conn, err=natsDialTimeout("tcp", url, dialTimeout)
 }
 }
iferr!=nil {
jitter:=time.Duration(rand.Int63n(int64(reconnectDelay)))
delay:=reconnectDelay+jitter
attempts++
ifs.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.
ifremote.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
ifjetstreamMigrateDelay==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)
iferr!=nil {
s.Warnf("Error looking up account [%s] checking for JetStream clear observer state on a leafnode", accName)
return
 }

acc.jscmMu.Lock()
deferacc.jscmMu.Unlock()

// Walk all streams looking for any clustered stream, skip otherwise.
for_, mset:=rangeacc.streams() {
node:=mset.raftNode()
ifnode==nil {
// Not R>1
continue
 }
// Check consumers
for_, o:=rangemset.getConsumers() {
ifn:=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)
iferr!=nil {
s.Warnf("Error looking up account [%s] checking for JetStream migration on a leafnode", accName)
return
 }

acc.jscmMu.Lock()
deferacc.jscmMu.Unlock()

// Walk all streams looking for any clustered stream, skip otherwise.
// If we are the leader force stepdown.
for_, mset:=rangeacc.streams() {
node:=mset.raftNode()
ifnode==nil {
// Not R>1
continue
 }
// Collect any consumers
for_, o:=rangemset.getConsumers() {
ifn:=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()
defers.mu.RUnlock()
returns.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) {
ifcfg.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) {
ifcfg.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
ifport==-1 {
port=0
 }

ifs.isShuttingDown() {
return
 }

s.mu.Lock()
hp:=net.JoinHostPort(opts.LeafNode.Host, strconv.Itoa(port))
l, e:=natsListen("tcp", hp)
s.leafNodeListenerErr=e
ife!=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,
 }
// If we have selected a random port...
ifport==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
iferr:=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:=rangeopts.LeafNode.Remotes {
ifr.TLSConfig!=nil&&r.TLSConfig.InsecureSkipVerify {
warn=true
break
 }
 }
 }
ifwarn {
s.Warnf(leafnodeTLSInsecureWarning)
 }
gos.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.
varcredsRe=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(clusterNamestring, headersbool) 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(),
 }

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

 } elseifcreds:=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)
iferr!=nil {
c.Errorf("%v", err)
returnerr
 }
deferwipeSlice(contents)
items:=credsRe.FindAllSubmatch(contents, -1)
iflen(items) <2 {
c.Errorf("Credentials file malformed")
returnerr
 }
// 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])
iferr!=nil {
c.Errorf("Credentials file has malformed seed")
returnerr
 }
// Wipe our key on exit.
deferkp.Wipe()

sigraw, _:=kp.Sign(c.nonce)
sig:=base64.RawURLEncoding.EncodeToString(sigraw)
cinfo.JWT=bytesToString(tmp)
cinfo.Sig=sig
 } elseifnkey:=c.leaf.remote.Nkey; nkey!=_EMPTY_ {
kp, err:=nkeys.FromSeed([]byte(nkey))
iferr!=nil {
c.Errorf("Remote nkey has malformed seed")
returnerr
 }
// Wipe our key on exit.
deferkp.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.
ifuserInfo:=c.leaf.remote.curURL.User; userInfo!=nil {
// For backward compatibility, if only username is provided, set both
// Token and User, not just Token.
cinfo.User=userInfo.Username()
varokbool
cinfo.Pass, ok=userInfo.Password()
if!ok {
cinfo.Token=cinfo.User
 }
 } elseifc.leaf.remote.username!=_EMPTY_ {
cinfo.User=c.leaf.remote.username
cinfo.Pass=c.leaf.remote.password
 }
b, err:=json.Marshal(cinfo)
iferr!=nil {
c.Errorf("Error marshaling CONNECT to remote leafnode: %v\n", err)
returnerr
 }
// 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)))
returnnil
}

// 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.
iflen(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(urlStrstring) bool {
ifs.leafURLsMap.addUrl(urlStr) {
s.generateLeafNodeInfoJSON()
returntrue
 }
returnfalse
}

// 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(urlStrstring) bool {
// Don't need to do this if we are removing the route connection because
// we are shuting down...
ifs.isShuttingDown() {
returnfalse
 }
ifs.leafURLsMap.removeUrl(urlStr) {
s.generateLeafNodeInfoJSON()
returntrue
 }
returnfalse
}

// 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:=ranges.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.
ifmaxSubs==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{}

// 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.
ifremote!=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.
varsolicitedbool
varacc*Account
varremoteSuffixstring
ifremote!=nil {
// For now, if lookup fails, we will constantly try
// to recreate this LN connection.
lacc:=remote.LocalAccount
varerrerror