Repository: nats-io/nats-server
Pull request #2233 (2021-05-20 05:51:56)

[added] pinned_cert option to tls block hex(sha256(spki))
Signed-off-by: Matthias Hanel <mh@synadia.com>

To create the hex(sha256(spki)) identifier:
openssl x509 -noout -pubkey -in certfile.pem | openssl pkey -pubin -outform DER | openssl dgst -sha256


Patch:
diff --git a/server/auth.go b/server/auth.go
index 1d7b45e4..f9bdb4ae 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -14,10 +14,12 @@
 package server
 
 import (
+	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509/pkix"
 	"encoding/asn1"
 	"encoding/base64"
+	"encoding/hex"
 	"fmt"
 	"net"
 	"net/url"
@@ -185,6 +187,19 @@ func (s *Server) checkAuthforWarnings() {
 		// Warning about using plaintext passwords.
 		s.Warnf("Plaintext passwords detected, use nkeys or bcrypt")
 	}
+	// only warn about client connections others use bidirectional TLS
+	if len(s.opts.TLSPinnedCerts) > 0 && !(s.opts.TLSVerify || s.opts.TLSMap) {
+		s.Warnf("Pinned Certs specified but no verify or verify_and_map that would require presenting a client cert")
+	}
+	if len(s.opts.Websocket.TLSPinnedCerts) > 0 && !(s.opts.Websocket.TLSMap) {
+		s.Warnf("Websocket Pinned Certs specified but no verify_and_map that would require presenting a client cert")
+	}
+	if len(s.opts.MQTT.TLSPinnedCerts) > 0 && !(s.opts.MQTT.TLSMap) {
+		s.Warnf("MQTT Pinned Certs specified but no verify_and_map that would require presenting a client cert")
+	}
+	if len(s.opts.LeafNode.TLSPinnedCerts) > 0 && !(s.opts.LeafNode.TLSMap) {
+		s.Warnf("Leaf Pinned Certs specified but verify_and_map that would require presenting a client cert")
+	}
 }
 
 // If Users or Nkeys options have definitions without an account defined,
@@ -344,6 +359,25 @@ func (s *Server) isClientAuthorized(c *client) bool {
 	return true
 }
 
+// returns false if the client needs to be disconnected
+func (s *Server) matchesPinnedCert(c *client, tlsPinnedCerts PinnedCertSet) bool {
+	if tlsPinnedCerts == nil {
+		return true
+	}
+	tlsState := c.GetTLSConnectionState()
+	if tlsState == nil || len(tlsState.PeerCertificates) == 0 || tlsState.PeerCertificates[0] == nil {
+		c.Debugf("Failed pinned cert test as client did not provide a certificate")
+		return false
+	}
+	sha := sha256.Sum256(tlsState.PeerCertificates[0].RawSubjectPublicKeyInfo)
+	keyId := hex.EncodeToString(sha[:])
+	if _, ok := tlsPinnedCerts[keyId]; !ok {
+		c.Debugf("Failed pinned cert test for key id: %s", keyId)
+		return false
+	}
+	return true
+}
+
 func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) bool {
 	var (
 		nkey *NkeyUser
@@ -366,11 +400,6 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 			authRequired = s.websocket.authOverride
 		}
 	}
-	if !authRequired {
-		// TODO(dlc) - If they send us credentials should we fail?
-		s.mu.Unlock()
-		return true
-	}
 	var (
 		username   string
 		password   string
@@ -378,12 +407,14 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 		noAuthUser string
 	)
 	tlsMap := opts.TLSMap
+	tlsPinnedCerts := opts.TLSPinnedCerts
 	if c.kind == CLIENT {
 		switch c.clientType() {
 		case MQTT:
 			mo := &opts.MQTT
 			// Always override TLSMap.
 			tlsMap = mo.TLSMap
+			tlsPinnedCerts = mo.TLSPinnedCerts
 			// The rest depends on if there was any auth override in
 			// the mqtt's config.
 			if s.mqtt.authOverride {
@@ -397,6 +428,7 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 			wo := &opts.Websocket
 			// Always override TLSMap.
 			tlsMap = wo.TLSMap
+			tlsPinnedCerts = wo.TLSPinnedCerts
 			// The rest depends on if there was any auth override in
 			// the websocket's config.
 			if s.websocket.authOverride {
@@ -409,7 +441,18 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 		}
 	} else {
 		tlsMap = opts.LeafNode.TLSMap
+		tlsPinnedCerts = opts.LeafNode.TLSPinnedCerts
 	}
+	if !s.matchesPinnedCert(c, tlsPinnedCerts) {
+		s.mu.Unlock()
+		return false
+	}
+	if !authRequired {
+		// TODO(dlc) - If they send us credentials should we fail?
+		s.mu.Unlock()
+		return true
+	}
+
 	if !ao {
 		noAuthUser = opts.NoAuthUser
 		username = opts.Username
@@ -872,6 +915,10 @@ func (s *Server) isRouterAuthorized(c *client) bool {
 		return s.opts.CustomRouterAuthentication.Check(c)
 	}
 
+	if !s.matchesPinnedCert(c, opts.Cluster.TLSPinnedCerts) {
+		return false
+	}
+
 	if opts.Cluster.TLSMap || opts.Cluster.TLSCheckKnownURLs {
 		return checkClientTLSCertSubject(c, func(user string, _ *ldap.DN, isDNSAltName bool) (string, bool) {
 			if user == "" {
@@ -907,6 +954,10 @@ func (s *Server) isGatewayAuthorized(c *client) bool {
 	// Snapshot server options.
 	opts := s.getOpts()
 
+	if !s.matchesPinnedCert(c, opts.Gateway.TLSPinnedCerts) {
+		return false
+	}
+
 	// Check whether TLS map is enabled, otherwise use single user/pass.
 	if opts.Gateway.TLSMap || opts.Gateway.TLSCheckKnownURLs {
 		return checkClientTLSCertSubject(c, func(user string, _ *ldap.DN, isDNSAltName bool) (string, bool) {
@@ -969,6 +1020,10 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 		return s.registerLeafWithAccount(c, account)
 	}
 
+	if !s.matchesPinnedCert(c, opts.LeafNode.TLSPinnedCerts) {
+		return false
+	}
+
 	// If leafnodes config has an authorization{} stanza, this takes precedence.
 	// The user in CONNECT must match. We will bind to the account associated
 	// with that user (from the leafnode's authorization{} config).
@@ -1050,6 +1105,9 @@ func comparePasswords(serverPassword, clientPassword string) bool {
 }
 
 func validateAuth(o *Options) error {
+	if err := validatePinnedCerts(o.TLSPinnedCerts); err != nil {
+		return err
+	}
 	for _, u := range o.Users {
 		if err := validateAllowedConnectionTypes(u.AllowedConnectionTypes); err != nil {
 			return err
diff --git a/server/dirstore.go b/server/dirstore.go
index 720657b3..b1ad4e86 100644
--- a/server/dirstore.go
+++ b/server/dirstore.go
@@ -91,7 +91,7 @@ func newDir(dirPath string, create bool) (string, error) {
 		if !create {
 			return "", err
 		}
-		if err = os.MkdirAll(dirPath, 0755); err != nil {
+		if err = os.MkdirAll(dirPath, defaultDirPerms); err != nil {
 			return "", err
 		}
 		if fullPath, err = validateDirPath(dirPath); err != nil {
@@ -424,7 +424,7 @@ func (store *DirJWTStore) write(path string, publicKey string, theJWT string) (b
 			}
 		}
 	}
-	if err := ioutil.WriteFile(path, []byte(theJWT), 0644); err != nil {
+	if err := ioutil.WriteFile(path, []byte(theJWT), defaultFilePerms); err != nil {
 		return false, err
 	} else if store.expiration != nil {
 		store.expiration.track(publicKey, newHash, theJWT)
@@ -473,7 +473,7 @@ func (store *DirJWTStore) save(publicKey string, theJWT string) error {
 	}
 	dirPath := filepath.Dir(path)
 	if _, err := validateDirPath(dirPath); err != nil {
-		if err := os.MkdirAll(dirPath, 0755); err != nil {
+		if err := os.MkdirAll(dirPath, defaultDirPerms); err != nil {
 			store.Unlock()
 			return err
 		}
@@ -499,7 +499,7 @@ func (store *DirJWTStore) saveIfNewer(publicKey string, theJWT string) error {
 	}
 	dirPath := filepath.Dir(path)
 	if _, err := validateDirPath(dirPath); err != nil {
-		if err := os.MkdirAll(dirPath, 0755); err != nil {
+		if err := os.MkdirAll(dirPath, defaultDirPerms); err != nil {
 			return err
 		}
 	}
diff --git a/server/disk_avail.go b/server/disk_avail.go
index 8abe3e49..993f5188 100644
--- a/server/disk_avail.go
+++ b/server/disk_avail.go
@@ -23,7 +23,7 @@ import (
 func diskAvailable(storeDir string) int64 {
 	var ba int64
 	if _, err := os.Stat(storeDir); os.IsNotExist(err) {
-		os.MkdirAll(storeDir, 0755)
+		os.MkdirAll(storeDir, defaultDirPerms)
 	}
 	var fs syscall.Statfs_t
 	if err := syscall.Statfs(storeDir, &fs); err == nil {
diff --git a/server/filestore.go b/server/filestore.go
index eec64aad..4036e5bf 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -64,6 +64,12 @@ type FileConsumerInfo struct {
 	ConsumerConfig
 }
 
+// Default file and directory permissions.
+const (
+	defaultDirPerms  = os.FileMode(0750)
+	defaultFilePerms = os.FileMode(0640)
+)
+
 type fileStore struct {
 	mu      sync.RWMutex
 	state   StreamState
@@ -225,7 +231,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 
 	// Check the directory
 	if stat, err := os.Stat(fcfg.StoreDir); os.IsNotExist(err) {
-		if err := os.MkdirAll(fcfg.StoreDir, 0755); err != nil {
+		if err := os.MkdirAll(fcfg.StoreDir, defaultDirPerms); err != nil {
 			return nil, fmt.Errorf("could not create storage directory - %v", err)
 		}
 	} else if stat == nil || !stat.IsDir() {
@@ -250,10 +256,10 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	// Check if this is a new setup.
 	mdir := path.Join(fcfg.StoreDir, msgDir)
 	odir := path.Join(fcfg.StoreDir, consumerDir)
-	if err := os.MkdirAll(mdir, 0755); err != nil {
+	if err := os.MkdirAll(mdir, defaultDirPerms); err != nil {
 		return nil, fmt.Errorf("could not create message storage directory - %v", err)
 	}
-	if err := os.MkdirAll(odir, 0755); err != nil {
+	if err := os.MkdirAll(odir, defaultDirPerms); err != nil {
 		return nil, fmt.Errorf("could not create consumer storage directory - %v", err)
 	}
 
@@ -372,14 +378,14 @@ func (fs *fileStore) writeStreamMeta() error {
 	if err != nil {
 		return err
 	}
-	if err := ioutil.WriteFile(meta, b, 0644); err != nil {
+	if err := ioutil.WriteFile(meta, b, defaultFilePerms); err != nil {
 		return err
 	}
 	fs.hh.Reset()
 	fs.hh.Write(b)
 	checksum := hex.EncodeToString(fs.hh.Sum(nil))
 	sum := path.Join(fs.fcfg.StoreDir, JetStreamMetaFileSum)
-	if err := ioutil.WriteFile(sum, []byte(checksum), 0644); err != nil {
+	if err := ioutil.WriteFile(sum, []byte(checksum), defaultFilePerms); err != nil {
 		return err
 	}
 	return nil
@@ -513,7 +519,7 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 		if mb.mfd != nil {
 			fd = mb.mfd
 		} else {
-			fd, err = os.OpenFile(mb.mfn, os.O_RDWR, 0644)
+			fd, err = os.OpenFile(mb.mfn, os.O_RDWR, defaultFilePerms)
 			if err != nil {
 				defer fd.Close()
 			}
@@ -911,7 +917,7 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 
 	mdir := path.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = path.Join(mdir, fmt.Sprintf(blkScan, mb.index))
-	mfd, err := os.OpenFile(mb.mfn, os.O_CREATE|os.O_RDWR, 0644)
+	mfd, err := os.OpenFile(mb.mfn, os.O_CREATE|os.O_RDWR, defaultFilePerms)
 	if err != nil {
 		mb.dirtyCloseWithRemove(true)
 		return nil, fmt.Errorf("Error creating msg block file [%q]: %v", mb.mfn, err)
@@ -919,7 +925,7 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	mb.mfd = mfd
 
 	mb.ifn = path.Join(mdir, fmt.Sprintf(indexScan, mb.index))
-	ifd, err := os.OpenFile(mb.ifn, os.O_CREATE|os.O_RDWR, 0644)
+	ifd, err := os.OpenFile(mb.ifn, os.O_CREATE|os.O_RDWR, defaultFilePerms)
 	if err != nil {
 		mb.dirtyCloseWithRemove(true)
 		return nil, fmt.Errorf("Error creating msg index file [%q]: %v", mb.mfn, err)
@@ -959,7 +965,7 @@ func (fs *fileStore) enableLastMsgBlockForWriting() error {
 	if mb.mfd != nil {
 		return nil
 	}
-	mfd, err := os.OpenFile(mb.mfn, os.O_CREATE|os.O_RDWR, 0644)
+	mfd, err := os.OpenFile(mb.mfn, os.O_CREATE|os.O_RDWR, defaultFilePerms)
 	if err != nil {
 		return fmt.Errorf("error opening msg block file [%q]: %v", mb.mfn, err)
 	}
@@ -1520,7 +1526,7 @@ func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int) error {
 
 	// Disk
 	if mb.cache.off+mb.cache.wp > ri {
-		mfd, err := os.OpenFile(mb.mfn, os.O_RDWR, 0644)
+		mfd, err := os.OpenFile(mb.mfn, os.O_RDWR, defaultFilePerms)
 		if err != nil {
 			return err
 		}
@@ -2740,7 +2746,7 @@ func (mb *msgBlock) writeIndexInfo() error {
 	}
 	var err error
 	if mb.ifd == nil {
-		ifd, err := os.OpenFile(mb.ifn, os.O_CREATE|os.O_RDWR, 0644)
+		ifd, err := os.OpenFile(mb.ifn, os.O_CREATE|os.O_RDWR, defaultFilePerms)
 		if err != nil {
 			return err
 		}
@@ -2939,7 +2945,7 @@ func (fs *fileStore) purge(fseq uint64) (uint64, error) {
 	os.Rename(mdir, pdir)
 	go os.RemoveAll(pdir)
 	// Create new one.
-	os.MkdirAll(mdir, 0755)
+	os.MkdirAll(mdir, defaultDirPerms)
 
 	// Make sure we have a lmb to write to.
 	if _, err := fs.newMsgBlockForWrite(); err != nil {
@@ -3521,7 +3527,7 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 		return nil, fmt.Errorf("bad consumer config")
 	}
 	odir := path.Join(fs.fcfg.StoreDir, consumerDir, name)
-	if err := os.MkdirAll(odir, 0755); err != nil {
+	if err := os.MkdirAll(odir, defaultDirPerms); err != nil {
 		return nil, fmt.Errorf("could not create consumer directory - %v", err)
 	}
 	csi := &FileConsumerInfo{ConsumerConfig: *cfg}
@@ -3886,14 +3892,14 @@ func (cfs *consumerFileStore) writeConsumerMeta() error {
 	if err != nil {
 		return err
 	}
-	if err := ioutil.WriteFile(meta, b, 0644); err != nil {
+	if err := ioutil.WriteFile(meta, b, defaultFilePerms); err != nil {
 		return err
 	}
 	cfs.hh.Reset()
 	cfs.hh.Write(b)
 	checksum := hex.EncodeToString(cfs.hh.Sum(nil))
 	sum := path.Join(cfs.odir, JetStreamMetaFileSum)
-	if err := ioutil.WriteFile(sum, []byte(checksum), 0644); err != nil {
+	if err := ioutil.WriteFile(sum, []byte(checksum), defaultFilePerms); err != nil {
 		return err
 	}
 	return nil
@@ -3912,7 +3918,7 @@ func (o *consumerFileStore) syncStateFile() {
 // Lock should be held.
 func (o *consumerFileStore) ensureStateFileOpen() error {
 	if o.ifd == nil {
-		ifd, err := os.OpenFile(o.ifn, os.O_CREATE|os.O_RDWR, 0644)
+		ifd, err := os.OpenFile(o.ifn, os.O_CREATE|os.O_RDWR, defaultFilePerms)
 		if err != nil {
 			return err
 		}
@@ -4206,7 +4212,7 @@ func newTemplateFileStore(storeDir string) *templateFileStore {
 
 func (ts *templateFileStore) Store(t *streamTemplate) error {
 	dir := path.Join(ts.dir, t.Name)
-	if err := os.MkdirAll(dir, 0755); err != nil {
+	if err := os.MkdirAll(dir, defaultDirPerms); err != nil {
 		return fmt.Errorf("could not create templates storage directory for %q- %v", t.Name, err)
 	}
 	meta := path.Join(dir, JetStreamMetaFile)
@@ -4219,7 +4225,7 @@ func (ts *templateFileStore) Store(t *streamTemplate) error {
 	if err != nil {
 		return err
 	}
-	if err := ioutil.WriteFile(meta, b, 0644); err != nil {
+	if err := ioutil.WriteFile(meta, b, defaultFilePerms); err != nil {
 		return err
 	}
 	// FIXME(dlc) - Do checksum
@@ -4227,7 +4233,7 @@ func (ts *templateFileStore) Store(t *streamTemplate) error {
 	ts.hh.Write(b)
 	checksum := hex.EncodeToString(ts.hh.Sum(nil))
 	sum := path.Join(dir, JetStreamMetaFileSum)
-	if err := ioutil.WriteFile(sum, []byte(checksum), 0644); err != nil {
+	if err := ioutil.WriteFile(sum, []byte(checksum), defaultFilePerms); err != nil {
 		return err
 	}
 	return nil
diff --git a/server/gateway.go b/server/gateway.go
index d9ed2889..b5662267 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -296,6 +296,9 @@ func validateGatewayOptions(o *Options) error {
 			return fmt.Errorf("gateway %q has no URL", g.Name)
 		}
 	}
+	if err := validatePinnedCerts(o.Gateway.TLSPinnedCerts); err != nil {
+		return fmt.Errorf("gateway %q: %v", o.Gateway.Name, err)
+	}
 	return nil
 }
 
diff --git a/server/jetstream.go b/server/jetstream.go
index 1963d777..ac3985bf 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -216,7 +216,7 @@ func (s *Server) checkStoreDir(cfg *JetStreamConfig) error {
 			// like streams and consumers.
 			if ok {
 				if !haveJetstreamDir {
-					err := os.Mkdir(filepath.Join(filepath.Dir(cfg.StoreDir), JetStreamStoreDir), 0755)
+					err := os.Mkdir(filepath.Join(filepath.Dir(cfg.StoreDir), JetStreamStoreDir), defaultDirPerms)
 					if err != nil {
 						return err
 					}
@@ -243,7 +243,7 @@ func (s *Server) enableJetStream(cfg JetStreamConfig) error {
 
 	// FIXME(dlc) - Allow memory only operation?
 	if stat, err := os.Stat(cfg.StoreDir); os.IsNotExist(err) {
-		if err := os.MkdirAll(cfg.StoreDir, 0755); err != nil {
+		if err := os.MkdirAll(cfg.StoreDir, defaultDirPerms); err != nil {
 			return fmt.Errorf("could not create storage directory - %v", err)
 		}
 	} else {
@@ -861,9 +861,12 @@ func (a *Account) EnableJetStream(limits *JetStreamAccountLimits) error {
 		}
 	}
 
+	// Clean up any old snapshots that were orphaned while staging.
+	os.RemoveAll(path.Join(js.config.StoreDir, snapStagingDir))
+
 	sdir := path.Join(jsa.storeDir, streamsDir)
 	if _, err := os.Stat(sdir); os.IsNotExist(err) {
-		if err := os.MkdirAll(sdir, 0755); err != nil {
+		if err := os.MkdirAll(sdir, defaultDirPerms); err != nil {
 			return fmt.Errorf("could not create storage streams directory - %v", err)
 		}
 	}
@@ -1615,6 +1618,8 @@ const (
 	JetStreamMaxStoreDefault = 1024 * 1024 * 1024 * 1024
 	// JetStreamMaxMemDefault is only used when we can't determine system memory. 256MB
 	JetStreamMaxMemDefault = 1024 * 1024 * 256
+	// snapshot staging for restores.
+	snapStagingDir = ".snap-staging"
 )
 
 // Dynamically create a config with a tmp based directory (repeatable) and 75% of system memory.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 14654927..a4488910 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -20,6 +20,7 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"os"
+	"path"
 	"sort"
 	"strconv"
 	"strings"
@@ -603,6 +604,7 @@ var (
 	jsClusterNotAvailErr   = &ApiError{Code: 503, Description: "JetStream system temporarily unavailable"}
 	jsClusterRequiredErr   = &ApiError{Code: 503, Description: "JetStream clustering support required"}
 	jsPeerNotMemberErr     = &ApiError{Code: 400, Description: "peer not a member"}
+	jsPeerRemapErr         = &ApiError{Code: 503, Description: "peer remap failed"}
 	jsClusterIncompleteErr = &ApiError{Code: 503, Description: "incomplete results"}
 	jsClusterTagsErr       = &ApiError{Code: 400, Description: "tags placement not supported for operation"}
 	jsClusterNoPeersErr    = &ApiError{Code: 400, Description: "no suitable peers for placement"}
@@ -2014,9 +2016,11 @@ func (s *Server) jsStreamRemovePeerRequest(sub *subscription, c *client, subject
 	}
 
 	// If we are here we have a valid peer member set for removal.
-	js.mu.Lock()
-	js.removePeerFromStream(sa, nodeName)
-	js.mu.Unlock()
+	if !js.removePeerFromStream(sa, nodeName) {
+		resp.Error = jsPeerRemapErr
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	resp.Success = true
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
@@ -2629,9 +2633,16 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 
 	var resp = JSApiStreamRestoreResponse{ApiResponse: ApiResponse{Type: JSApiStreamRestoreResponseType}}
 
-	// FIXME(dlc) - Need to close these up if we fail for some reason.
-	// TODO(dlc) - Might need to make configurable or stream direct to storage dir.
-	tfile, err := ioutil.TempFile("", "jetstream-restore-")
+	snapDir := path.Join(js.config.StoreDir, snapStagingDir)
+	if _, err := os.Stat(snapDir); os.IsNotExist(err) {
+		if err := os.MkdirAll(snapDir, defaultDirPerms); err != nil {
+			resp.Error = &ApiError{Code: 503, Description: "JetStream unable to create temp storage for restore"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return nil
+		}
+	}
+
+	tfile, err := ioutil.TempFile(snapDir, "js-restore-")
 	if err != nil {
 		resp.Error = &ApiError{Code: 500, Description: "JetStream unable to open temp storage for restore"}
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -2668,7 +2679,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 
 	var total int
 
-	// FIXM(dlc) - Probably take out of network path eventually do to disk I/O?
+	// FIXM(dlc) - Probably take out of network path eventually due to disk I/O?
 	processChunk := func(sub *subscription, c *client, subject, reply string, msg []byte) {
 		// We require reply subjects to communicate back failures, flow etc. If they do not have one log and cancel.
 		if reply == _EMPTY_ {
@@ -2723,6 +2734,8 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 
 	sub, err := acc.subscribeInternal(restoreSubj, processChunk)
 	if err != nil {
+		tfile.Close()
+		os.Remove(tfile.Name())
 		resp.Error = &ApiError{Code: 500, Description: "JetStream unable to subscribe to restore snapshot " + restoreSubj + ": " + err.Error()}
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return nil
@@ -2730,7 +2743,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 
 	// Mark the subject so the end user knows where to send the snapshot chunks.
 	resp.DeliverSubject = restoreSubj
-	s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
+	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 
 	doneCh := make(chan error, 1)
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index ff4ca49f..973f65d9 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -305,7 +305,7 @@ func (cc *jetStreamCluster) isLeader() bool {
 		// Non-clustered mode
 		return true
 	}
-	return cc.meta.Leader()
+	return cc.meta != nil && cc.meta.Leader()
 }
 
 // isCurrent will determine if this node is a leader or an up to date follower.
@@ -1044,14 +1044,16 @@ func (js *jetStream) processRemovePeer(peer string) {
 }
 
 // Assumes all checks have already been done.
-// Lock should be held.
-func (js *jetStream) removePeerFromStream(sa *streamAssignment, peer string) {
-	s, cc := js.srv, js.cluster
+func (js *jetStream) removePeerFromStream(sa *streamAssignment, peer string) bool {
+	js.mu.Lock()
+	defer js.mu.Unlock()
 
-	csa := sa.copyGroup()
+	s, cc, csa := js.srv, js.cluster, sa.copyGroup()
 	if !cc.remapStreamAssignment(csa, peer) {
 		s.Warnf("JetStream cluster could not remap stream '%s > %s'", sa.Client.serviceAccount(), sa.Config.Name)
+		return false
 	}
+
 	// Send our proposal for this csa. Also use same group definition for all the consumers as well.
 	cc.meta.Propose(encodeAddStreamAssignment(csa))
 	rg := csa.Group
@@ -1060,6 +1062,7 @@ func (js *jetStream) removePeerFromStream(sa *streamAssignment, peer string) {
 		cca.Group.Peers = rg.Peers
 		cc.meta.Propose(encodeAddConsumerAssignment(&cca))
 	}
+	return true
 }
 
 // Check if we have peer related entries.
@@ -1681,28 +1684,14 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 			}
 		} else if e.Type == EntryRemovePeer {
 			js.mu.RLock()
-			ourID := js.cluster.meta.ID()
+			var ourID string
+			if js.cluster != nil && js.cluster.meta != nil {
+				ourID = js.cluster.meta.ID()
+			}
 			js.mu.RUnlock()
+			// We only need to do processing if this is us.
 			if peer := string(e.Data); peer == ourID {
-				shouldDelete := true
-				if sa := mset.streamAssignment(); sa != nil {
-					js.mu.Lock()
-					// Make sure we are not part of this assignment. If we are
-					// we need to ignore this remove.
-					if sa.Group.isMember(ourID) {
-						shouldDelete = false
-					} else {
-						if node := sa.Group.node; node != nil {
-							node.ProposeRemovePeer(ourID)
-						}
-						sa.Group.node = nil
-						sa.err = nil
-					}
-					js.mu.Unlock()
-				}
-				if shouldDelete {
-					mset.stop(true, false)
-				}
+				mset.stop(true, false)
 			}
 			return nil
 		}
@@ -1959,23 +1948,14 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) bool {
 	if isMember {
 		js.processClusterCreateStream(acc, sa)
 	} else {
-		// Clear our raft node here.
-		// TODO(dlc) - This might be better if done by leader, not the one who is being removed
-		// since we are most likely offline.
+		// Check if we have a raft node running, meaning we are no longer part of the group but were.
 		js.mu.Lock()
 		if node := sa.Group.node; node != nil {
 			node.ProposeRemovePeer(ourID)
-			didRemove = true
 		}
 		sa.Group.node = nil
 		sa.err = nil
 		js.mu.Unlock()
-
-		if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
-			// We have one here even though we are not a member. This can happen on re-assignment.
-			s.Debugf("JetStream removing stream '%s > %s' from this server, reassigned", sa.Client.serviceAccount(), sa.Config.Name)
-			mset.stop(true, false)
-		}
 	}
 
 	return didRemove
@@ -2392,8 +2372,7 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		return
 	}
 
-	acc, err := s.LookupAccount(accName)
-	if err != nil {
+	if _, err := s.LookupAccount(accName); err != nil {
 		ll := fmt.Sprintf("Account [%s] lookup for consumer create failed: %v", accName, err)
 		if isMember {
 			// If we can not lookup the account and we are a member, send this result back to the metacontroller leader.
@@ -2443,9 +2422,7 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 	if isMember {
 		js.processClusterCreateConsumer(ca, state)
 	} else {
-		// Clear our raft node here.
-		// TODO(dlc) - This might be better if done by leader, not the one who is being removed
-		// since we are most likely offline.
+		// Check if we have a raft node running, meaning we are no longer part of the group but were.
 		js.mu.Lock()
 		if node := ca.Group.node; node != nil {
 			node.ProposeRemovePeer(ourID)
@@ -2453,16 +2430,6 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		ca.Group.node = nil
 		ca.err = nil
 		js.mu.Unlock()
-
-		// We are not a member, if we have this consumer on this
-		// server remove it.
-		if mset, _ := acc.lookupStream(stream); mset != nil {
-			if o := mset.lookupConsumer(consumer); o != nil {
-				s.Debugf("JetStream removing consumer '%s > %s > %s' from this server, re-assigned",
-					ca.Client.serviceAccount(), stream, consumer)
-				o.stopWithFlags(true, false, false)
-			}
-		}
 	}
 }
 
@@ -2835,23 +2802,7 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 			}
 			js.mu.RUnlock()
 			if peer := string(e.Data); peer == ourID {
-				shouldDelete := true
-				if ca := o.consumerAssignment(); ca != nil {
-					js.mu.Lock()
-					if ca.Group.isMember(ourID) {
-						shouldDelete = false
-					} else {
-						if node := ca.Group.node; node != nil {
-							node.ProposeRemovePeer(ourID)
-						}
-						ca.Group.node = nil
-						ca.err = nil
-					}
-					js.mu.Unlock()
-				}
-				if shouldDelete {
-					o.stopWithFlags(true, false, false)
-				}
+				o.stopWithFlags(true, false, false)
 			}
 			return nil
 		} else {
diff --git a/server/leafnode.go b/server/leafnode.go
index 6404381f..be88167c 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -308,6 +308,9 @@ func validateLeafNode(o *Options) error {
 	if o.SystemAccount == "" {
 		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
 }
 
diff --git a/server/mqtt.go b/server/mqtt.go
index 5ed0a72e..eb147b62 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -554,6 +554,9 @@ func validateMQTTOptions(o *Options) error {
 		o.LeafNode.Port == 0 && len(o.LeafNode.Remotes) == 0 {
 		return errMQTTStandaloneNeedsJetStream
 	}
+	if err := validatePinnedCerts(mo.TLSPinnedCerts); err != nil {
+		return fmt.Errorf("mqtt: %v", err)
+	}
 	return nil
 }
 
diff --git a/server/opts.go b/server/opts.go
index 639d0c74..dedb8be4 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -53,6 +53,9 @@ func NoErrOnUnknownFields(noError bool) {
 	atomic.StoreInt32(&allowUnknownTopLevelField, val)
 }
 
+// Set of lower case hex-encoded sha256 of DER encoded SubjectPublicKeyInfo
+type PinnedCertSet map[string]struct{}
+
 // ClusterOpts are options for clusters.
 // NOTE: This structure is no longer used for monitoring endpoints
 // and json tags are deprecated and may be removed in the future.
@@ -68,6 +71,7 @@ type ClusterOpts struct {
 	TLSConfig         *tls.Config       `json:"-"`
 	TLSMap            bool              `json:"-"`
 	TLSCheckKnownURLs bool              `json:"-"`
+	TLSPinnedCerts    PinnedCertSet     `json:"-"`
 	ListenStr         string            `json:"-"`
 	Advertise         string            `json:"-"`
 	NoAdvertise       bool              `json:"-"`
@@ -91,6 +95,7 @@ type GatewayOpts struct {
 	TLSTimeout        float64              `json:"tls_timeout,omitempty"`
 	TLSMap            bool                 `json:"-"`
 	TLSCheckKnownURLs bool                 `json:"-"`
+	TLSPinnedCerts    PinnedCertSet        `json:"-"`
 	Advertise         string               `json:"advertise,omitempty"`
 	ConnectRetries    int                  `json:"connect_retries,omitempty"`
 	Gateways          []*RemoteGatewayOpts `json:"gateways,omitempty"`
@@ -123,6 +128,7 @@ type LeafNodeOpts struct {
 	TLSConfig         *tls.Config   `json:"-"`
 	TLSTimeout        float64       `json:"tls_timeout,omitempty"`
 	TLSMap            bool          `json:"-"`
+	TLSPinnedCerts    PinnedCertSet `json:"-"`
 	Advertise         string        `json:"-"`
 	NoAdvertise       bool          `json:"-"`
 	ReconnectInterval time.Duration `json:"-"`
@@ -226,6 +232,7 @@ type Options struct {
 	TLSKey                string        `json:"-"`
 	TLSCaCert             string        `json:"-"`
 	TLSConfig             *tls.Config   `json:"-"`
+	TLSPinnedCerts        PinnedCertSet `json:"-"`
 	AllowNonTLS           bool          `json:"-"`
 	WriteDeadline         time.Duration `json:"-"`
 	MaxClosedClients      int           `json:"-"`
@@ -316,6 +323,9 @@ type WebsocketOpts struct {
 	// If true, map certificate values for authentication purposes.
 	TLSMap bool
 
+	// When present, accepted client certificates (verify/verify_and_map) must be in this list
+	TLSPinnedCerts PinnedCertSet
+
 	// If true, the Origin header must match the request's host.
 	SameOrigin bool
 
@@ -361,6 +371,8 @@ type MQTTOpts struct {
 	TLSMap bool
 	// Timeout for the TLS handshake
 	TLSTimeout float64
+	// Set of allowable certificates
+	TLSPinnedCerts PinnedCertSet
 
 	// AckWait is the amount of time after which a QoS 1 message sent to
 	// a client is redelivered as a DUPLICATE if the server has not
@@ -470,6 +482,7 @@ type TLSConfigOpts struct {
 	Timeout           float64
 	Ciphers           []uint16
 	CurvePreferences  []tls.CurveID
+	PinnedCerts       PinnedCertSet
 }
 
 var tlsUsage = `
@@ -827,7 +840,7 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 		}
 		o.TLSTimeout = tc.Timeout
 		o.TLSMap = tc.Map
-
+		o.TLSPinnedCerts = tc.PinnedCerts
 	case "allow_non_tls":
 		o.AllowNonTLS = v.(bool)
 	case "write_deadline":
@@ -1272,6 +1285,7 @@ func parseCluster(v interface{}, opts *Options, errors *[]error, warnings *[]err
 			opts.Cluster.TLSConfig = config
 			opts.Cluster.TLSTimeout = tlsopts.Timeout
 			opts.Cluster.TLSMap = tlsopts.Map
+			opts.Cluster.TLSPinnedCerts = tlsopts.PinnedCerts
 			opts.Cluster.TLSCheckKnownURLs = tlsopts.TLSCheckKnownURLs
 		case "cluster_advertise", "advertise":
 			opts.Cluster.Advertise = mv.(string)
@@ -1389,6 +1403,7 @@ func parseGateway(v interface{}, o *Options, errors *[]error, warnings *[]error)
 			o.Gateway.TLSTimeout = tlsopts.Timeout
 			o.Gateway.TLSMap = tlsopts.Map
 			o.Gateway.TLSCheckKnownURLs = tlsopts.TLSCheckKnownURLs
+			o.Gateway.TLSPinnedCerts = tlsopts.PinnedCerts
 		case "advertise":
 			o.Gateway.Advertise = mv.(string)
 		case "connect_retries":
@@ -1612,6 +1627,7 @@ func parseLeafNodes(v interface{}, opts *Options, errors *[]error, warnings *[]e
 			}
 			opts.LeafNode.TLSTimeout = tc.Timeout
 			opts.LeafNode.TLSMap = tc.Map
+			opts.LeafNode.TLSPinnedCerts = tc.PinnedCerts
 		case "leafnode_advertise", "advertise":
 			opts.LeafNode.Advertise = mv.(string)
 		case "no_advertise":
@@ -3444,6 +3460,24 @@ func parseTLS(v interface{}, isClientCtx bool) (t *TLSConfigOpts, retErr error)
 				at = mv
 			}
 			tc.Timeout = at
+		case "pinned_certs":
+			ra, ok := mv.([]interface{})
+			if !ok {
+				return nil, &configErr{tk, "error parsing tls config, expected 'pinned_certs' to be a list of hex-encoded sha256 of DER encoded SubjectPublicKeyInfo"}
+			}
+			if len(ra) != 0 {
+				wl := PinnedCertSet{}
+				re := regexp.MustCompile("^[A-Fa-f0-9]{64}$")
+				for _, r := range ra {
+					tk, r := unwrapValue(r, &lt)
+					entry := strings.ToLower(r.(string))
+					if !re.MatchString(entry) {
+						return nil, &configErr{tk, fmt.Sprintf("error parsing tls config, 'pinned_certs' key %s does not look like hex-encoded sha256 of DER encoded SubjectPublicKeyInfo", entry)}
+					}
+					wl[entry] = struct{}{}
+				}
+				tc.PinnedCerts = wl
+			}
 		default:
 			return nil, &configErr{tk, fmt.Sprintf("error parsing tls config, unknown field [%q]", mk)}
 		}
@@ -3573,6 +3607,7 @@ func parseWebsocket(v interface{}, o *Options, errors *[]error, warnings *[]erro
 				continue
 			}
 			o.Websocket.TLSMap = tc.Map
+			o.Websocket.TLSPinnedCerts = tc.PinnedCerts
 		case "same_origin":
 			o.Websocket.SameOrigin = mv.(bool)
 		case "allowed_origins", "allowed_origin", "allow_origins", "allow_origin", "origins", "origin":
@@ -3662,6 +3697,7 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er
 			}
 			o.MQTT.TLSTimeout = tc.Timeout
 			o.MQTT.TLSMap = tc.Map
+			o.MQTT.TLSPinnedCerts = tc.PinnedCerts
 		case "authorization", "authentication":
 			auth := parseSimpleAuth(tk, errors, warnings)
 			o.MQTT.Username = auth.user
diff --git a/server/raft.go b/server/raft.go
index bf25d104..5c1d4755 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -673,7 +673,7 @@ func (n *raft) ProposeAddPeer(peer string) error {
 func (n *raft) ProposeRemovePeer(peer string) error {
 	n.RLock()
 	propc, subj := n.propc, n.rpsubj
-	isLeader, isUs := n.state == Leader, n.id == peer
+	isLeader := n.state == Leader
 	werr := n.werr
 	n.RUnlock()
 
@@ -685,9 +685,6 @@ func (n *raft) ProposeRemovePeer(peer string) error {
 	if isLeader {
 		select {
 		case propc <- &Entry{EntryRemovePeer, []byte(peer)}:
-			if isUs {
-				n.attemptStepDown(noLeader)
-			}
 		default:
 			return errProposalFailed
 		}
@@ -1111,6 +1108,19 @@ func (n *raft) GroupLeader() string {
 	return n.leader
 }
 
+// Guess the best next leader. Stepdown will check more thoroughly.
+func (n *raft) selectNextLeader() string {
+	nextLeader, hli := noLeader, uint64(0)
+	for peer, ps := range n.peers {
+		if peer == n.id || ps.li <= hli {
+			continue
+		}
+		hli = ps.li
+		nextLeader = peer
+	}
+	return nextLeader
+}
+
 // StepDown will have a leader stepdown and optionally do a leader transfer.
 func (n *raft) StepDown(preferred ...string) error {
 	n.Lock()
@@ -2120,6 +2130,12 @@ func (n *raft) applyCommit(index uint64) error {
 				}
 			}
 
+			// If this is us and we are the leader we should attempt to stepdown.
+			if peer == n.id && n.state == Leader {
+				n.attemptStepDown(n.selectNextLeader())
+			}
+
+			// Write out our new state.
 			n.writePeerState(&peerState{n.peerNames(), n.csz})
 			// We pass these up as well.
 			committed = append(committed, e)
diff --git a/server/reload.go b/server/reload.go
index 189dfa03..424cb4f1 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -215,6 +215,17 @@ func (t *tlsTimeoutOption) Apply(server *Server) {
 	server.Noticef("Reloaded: tls timeout = %v", t.newValue)
 }
 
+// tlsPinnedCertOption implements the option interface for the tls `pinned_certs` setting.
+type tlsPinnedCertOption struct {
+	noopOption
+	newValue PinnedCertSet
+}
+
+// Apply is a no-op because the pinned certs will be reloaded after options are  applied.
+func (t *tlsPinnedCertOption) Apply(server *Server) {
+	server.Noticef("Reloaded: %d pinned_certs", len(t.newValue))
+}
+
 // authOption is a base struct that provides default option behaviors.
 type authOption struct {
 	noopOption
@@ -790,7 +801,7 @@ func imposeOrder(value interface{}) error {
 		})
 	case WebsocketOpts:
 		sort.Strings(value.AllowedOrigins)
-	case string, bool, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config,
+	case string, bool, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
 		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList:
 		// explicitly skipped types
 	default:
@@ -864,6 +875,8 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			diffOpts = append(diffOpts, &tlsOption{newValue: newValue.(*tls.Config)})
 		case "tlstimeout":
 			diffOpts = append(diffOpts, &tlsTimeoutOption{newValue: newValue.(float64)})
+		case "tlspinnedcerts":
+			diffOpts = append(diffOpts, &tlsPinnedCertOption{newValue: newValue.(PinnedCertSet)})
 		case "username":
 			diffOpts = append(diffOpts, &usernameOption{})
 		case "password":
diff --git a/server/server.go b/server/server.go
index 26dd28d5..6fde16c0 100644
--- a/server/server.go
+++ b/server/server.go
@@ -26,6 +26,7 @@ import (
 	"math/rand"
 	"net"
 	"net/http"
+	"regexp"
 
 	// Allow dynamic profiling.
 	_ "net/http/pprof"
@@ -343,6 +344,7 @@ func NewServer(opts *Options) (*Server, error) {
 		httpBasePath: httpBasePath,
 		eventIds:     nuid.New(),
 		routesToSelf: make(map[string]struct{}),
+		httpReqStats: make(map[string]uint64), // Used to track HTTP requests
 	}
 
 	// Trusted root operator keys.
@@ -548,7 +550,10 @@ func (s *Server) ClientURL() string {
 	return fmt.Sprintf("%s%s:%d", scheme, opts.Host, opts.Port)
 }
 
-func validateClusterName(o *Options) error {
+func validateCluster(o *Options) error {
+	if err := validatePinnedCerts(o.Cluster.TLSPinnedCerts); err != nil {
+		return fmt.Errorf("cluster: %v", err)
+	}
 	// Check that cluster name if defined matches any gateway name.
 	if o.Gateway.Name != "" && o.Gateway.Name != o.Cluster.Name {
 		if o.Cluster.Name != "" {
@@ -560,6 +565,17 @@ func validateClusterName(o *Options) error {
 	return nil
 }
 
+func validatePinnedCerts(pinned PinnedCertSet) error {
+	re := regexp.MustCompile("^[a-f0-9]{64}$")
+	for certId := range pinned {
+		entry := strings.ToLower(certId)
+		if !re.MatchString(entry) {
+			return fmt.Errorf("error parsing 'pinned_certs' key %s does not look like lower case hex-encoded sha256 of DER encoded SubjectPublicKeyInfo", entry)
+		}
+	}
+	return nil
+}
+
 func validateOptions(o *Options) error {
 	if o.LameDuckDuration > 0 && o.LameDuckGracePeriod >= o.LameDuckDuration {
 		return fmt.Errorf("lame duck grace period (%v) should be strictly lower than lame duck duration (%v)",
@@ -584,7 +600,7 @@ func validateOptions(o *Options) error {
 		return err
 	}
 	// Check that cluster name if defined matches any gateway name.
-	if err := validateClusterName(o); err != nil {
+	if err := validateCluster(o); err != nil {
 		return err
 	}
 	if err := validateMQTTOptions(o); err != nil {
@@ -2081,16 +2097,6 @@ func (s *Server) startMonitoring(secure bool) error {
 	// Snapshot server options.
 	opts := s.getOpts()
 
-	// Used to track HTTP requests
-	s.httpReqStats = map[string]uint64{
-		RootPath:     0,
-		VarzPath:     0,
-		ConnzPath:    0,
-		RoutezPath:   0,
-		GatewayzPath: 0,
-		SubszPath:    0,
-	}
-
 	var (
 		hp           string
 		err          error
diff --git a/server/stream.go b/server/stream.go
index 1e11d093..8d74a163 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1499,7 +1499,7 @@ func (mset *stream) setupMirrorConsumer() error {
 	respCh := make(chan *JSApiConsumerCreateResponse, 1)
 	reply := infoReplySubject()
 	crSub, _ := mset.subscribeInternal(reply, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
-		mset.unsubscribe(sub)
+		mset.unsubscribeUnlocked(sub)
 		_, msg := c.msgParts(rmsg)
 
 		var ccr JSApiConsumerCreateResponse
@@ -1569,7 +1569,7 @@ func (mset *stream) setupMirrorConsumer() error {
 			}
 			mset.setMirrorErr(ccr.Error)
 		case <-time.After(10 * time.Second):
-			mset.unsubscribe(crSub)
+			mset.unsubscribeUnlocked(crSub)
 			return
 		}
 	}()
@@ -3240,7 +3240,7 @@ func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error
 
 	sd := path.Join(jsa.storeDir, snapsDir)
 	if _, err := os.Stat(sd); os.IsNotExist(err) {
-		if err := os.MkdirAll(sd, 0755); err != nil {
+		if err := os.MkdirAll(sd, defaultDirPerms); err != nil {
 			return nil, fmt.Errorf("could not create snapshots directory - %v", err)
 		}
 	}
@@ -3249,7 +3249,7 @@ func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error
 		return nil, err
 	}
 	if _, err := os.Stat(sdir); os.IsNotExist(err) {
-		if err := os.MkdirAll(sdir, 0755); err != nil {
+		if err := os.MkdirAll(sdir, defaultDirPerms); err != nil {
 			return nil, fmt.Errorf("could not create snapshots directory - %v", err)
 		}
 	}
@@ -3266,7 +3266,7 @@ func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error
 		}
 		fpath := path.Join(sdir, filepath.Clean(hdr.Name))
 		pdir := filepath.Dir(fpath)
-		os.MkdirAll(pdir, 0750)
+		os.MkdirAll(pdir, defaultDirPerms)
 		fd, err := os.OpenFile(fpath, os.O_CREATE|os.O_RDWR, 0600)
 		if err != nil {
 			return nil, err
diff --git a/server/websocket.go b/server/websocket.go
index 940ac627..b699631d 100644
--- a/server/websocket.go
+++ b/server/websocket.go
@@ -886,6 +886,9 @@ func validateWebsocketOptions(o *Options) error {
 			return fmt.Errorf("trusted operators or trusted keys configuration is required for JWT authentication via cookie %q", wo.JWTCookie)
 		}
 	}
+	if err := validatePinnedCerts(wo.TLSPinnedCerts); err != nil {
+		return fmt.Errorf("websocket: %v", err)
+	}
 	return nil
 }
 


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index f83382b6..f5b94cf2 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -3769,13 +3769,7 @@ func TestJetStreamClusterRemovePeer(t *testing.T) {
 		t.Fatalf("Unexpected error: %+v", rpResp.Error)
 	}
 
-	// Grab shorter timeout jetstream context.
-	js, err = nc.JetStream(nats.MaxWait(100 * time.Millisecond))
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
 		si, err := js.StreamInfo("TEST")
 		if err != nil {
 			return fmt.Errorf("Could not fetch stream info: %v", err)
@@ -6385,8 +6379,9 @@ func TestJetStreamClusterDomainsAndSameNameSources(t *testing.T) {
 	if ssi.Sources[0].External == nil {
 		t.Fatalf("Expected a non-nil external designation")
 	}
-	if ssi.Sources[0].External.ApiPrefix != "$JS.SPOKE-1.API" {
-		t.Fatalf("Expected external api of %q, got %q", "$JS.SPOKE-1.API", ssi.Sources[0].External.ApiPrefix)
+	pre := ssi.Sources[0].External.ApiPrefix
+	if pre != "$JS.SPOKE-1.API" && pre != "$JS.SPOKE-2.API" {
+		t.Fatalf("Expected external api of %q, got %q", "$JS.SPOKE-[1|2].API", ssi.Sources[0].External.ApiPrefix)
 	}
 
 	// Also create a mirror.
@@ -6768,7 +6763,7 @@ func TestJetStreamClusterCrossAccountInterop(t *testing.T) {
 	template := `
 	listen: 127.0.0.1:-1
 	server_name: %s
-	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: "%s"}
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, domain: HUB, store_dir: "%s"}
 
 	cluster {
 		name: %s
@@ -6782,6 +6777,8 @@ func TestJetStreamClusterCrossAccountInterop(t *testing.T) {
 			users = [ { user: "rip", pass: "pass" } ]
 			exports [
 				{ service: "$JS.API.CONSUMER.INFO.>" }
+				{ service: "$JS.HUB.API.CONSUMER.>", response: stream }
+				{ stream: "M.SYNC.>" } # For the mirror
 			]
 		}
 		IA {
@@ -6789,13 +6786,15 @@ func TestJetStreamClusterCrossAccountInterop(t *testing.T) {
 			users = [ { user: "dlc", pass: "pass" } ]
 			imports [
 				{ service: { account: JS, subject: "$JS.API.CONSUMER.INFO.TEST.DLC"}, to: "FROM.DLC" }
+				{ service: { account: JS, subject: "$JS.HUB.API.CONSUMER.>"}, to: "js.xacc.API.CONSUMER.>" }
+				{ stream: { account: JS, subject: "M.SYNC.>"} }
 			]
 		}
 		$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 	}
 	`
 
-	c := createJetStreamClusterWithTemplate(t, template, "C22", 3)
+	c := createJetStreamClusterWithTemplate(t, template, "HUB", 3)
 	defer c.shutdown()
 
 	// Create the stream and the consumer under the JS/rip user.
@@ -6811,13 +6810,57 @@ func TestJetStreamClusterCrossAccountInterop(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
+	// Also create a stream via the domain qualified API.
+	js, err = nc.JetStream(nats.APIPrefix("$JS.HUB.API"))
+	if err != nil {
+		t.Fatalf("Unexpected error getting JetStream context: %v", err)
+	}
+	if _, err := js.AddStream(&nats.StreamConfig{Name: "ORDERS", Replicas: 2}); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
 	// Now we want to access the consumer info from IA/dlc.
-	nc, _ = jsClientConnect(t, c.randomServer(), nats.UserInfo("dlc", "pass"))
-	defer nc.Close()
+	nc2, js2 := jsClientConnect(t, c.randomServer(), nats.UserInfo("dlc", "pass"))
+	defer nc2.Close()
 
-	if _, err := nc.Request("FROM.DLC", nil, time.Second); err != nil {
+	if _, err := nc2.Request("FROM.DLC", nil, time.Second); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+
+	// Make sure domain mappings etc work across accounts.
+	// Setup a mirror.
+	_, err = js2.AddStream(&nats.StreamConfig{
+		Name: "MIRROR",
+		Mirror: &nats.StreamSource{
+			Name: "ORDERS",
+			External: &nats.ExternalStream{
+				APIPrefix:     "js.xacc.API",
+				DeliverPrefix: "M.SYNC",
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Send 10 messages..
+	msg, toSend := []byte("Hello mapped domains"), 10
+	for i := 0; i < toSend; i++ {
+		if _, err = js.Publish("ORDERS", msg); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("MIRROR")
+		if err != nil {
+			return fmt.Errorf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 10 {
+			return fmt.Errorf("Expected 10 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
 }
 
 // Support functions
diff --git a/server/opts_test.go b/server/opts_test.go
index 3714c21f..09cbd0c6 100644
--- a/server/opts_test.go
+++ b/server/opts_test.go
@@ -890,6 +890,94 @@ func TestNkeyUsersConfig(t *testing.T) {
 	}
 }
 
+// Test pinned certificates
+func TestTlsPinnedCertificates(t *testing.T) {
+	confFileName := createConfFile(t, []byte(`
+	tls {
+		cert_file: "./configs/certs/server.pem"
+		key_file: "./configs/certs/key.pem"
+		# Require a client certificate and map user id from certificate
+		verify: true
+		pinned_certs: ["7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069", 
+			"a8f407340dcc719864214b85ed96f98d16cbffa8f509d9fa4ca237b7bb3f9c32"]
+	}
+	cluster {
+		port -1
+		name cluster-hub
+		tls {
+			cert_file: "./configs/certs/server.pem"
+			key_file: "./configs/certs/key.pem"
+			# Require a client certificate and map user id from certificate
+			verify: true
+			pinned_certs: ["7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069", 
+				"a8f407340dcc719864214b85ed96f98d16cbffa8f509d9fa4ca237b7bb3f9c32"]
+		}
+	}
+	leafnodes {
+		port -1
+		tls {
+			cert_file: "./configs/certs/server.pem"
+			key_file: "./configs/certs/key.pem"
+			# Require a client certificate and map user id from certificate
+			verify: true
+			pinned_certs: ["7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069", 
+				"a8f407340dcc719864214b85ed96f98d16cbffa8f509d9fa4ca237b7bb3f9c32"]
+		}
+	}
+	gateway {
+		name: "A"
+		port -1
+		tls {
+			cert_file: "./configs/certs/server.pem"
+			key_file: "./configs/certs/key.pem"
+			# Require a client certificate and map user id from certificate
+			verify: true
+			pinned_certs: ["7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069", 
+				"a8f407340dcc719864214b85ed96f98d16cbffa8f509d9fa4ca237b7bb3f9c32"]
+		}
+	}
+	websocket {
+		port -1
+		tls {
+			cert_file: "./configs/certs/server.pem"
+			key_file: "./configs/certs/key.pem"
+			# Require a client certificate and map user id from certificate
+			verify: true
+			pinned_certs: ["7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069", 
+				"a8f407340dcc719864214b85ed96f98d16cbffa8f509d9fa4ca237b7bb3f9c32"]
+		}
+	}
+	mqtt {
+		port -1
+		tls {
+			cert_file: "./configs/certs/server.pem"
+			key_file: "./configs/certs/key.pem"
+			# Require a client certificate and map user id from certificate
+			verify: true
+			pinned_certs: ["7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069", 
+				"a8f407340dcc719864214b85ed96f98d16cbffa8f509d9fa4ca237b7bb3f9c32"]
+		}
+	}`))
+	defer removeFile(t, confFileName)
+	opts, err := ProcessConfigFile(confFileName)
+	if err != nil {
+		t.Fatalf("Received an error reading config file: %v", err)
+	}
+	check := func(set PinnedCertSet) {
+		t.Helper()
+		if l := len(set); l != 2 {
+			t.Fatalf("Expected 2 pinned certificates, got got %d", l)
+		}
+	}
+
+	check(opts.TLSPinnedCerts)
+	check(opts.LeafNode.TLSPinnedCerts)
+	check(opts.Cluster.TLSPinnedCerts)
+	check(opts.MQTT.TLSPinnedCerts)
+	check(opts.Gateway.TLSPinnedCerts)
+	check(opts.Websocket.TLSPinnedCerts)
+}
+
 func TestNkeyUsersDefaultPermissionsConfig(t *testing.T) {
 	confFileName := createConfFile(t, []byte(`
 	authorization {
diff --git a/test/tls_test.go b/test/tls_test.go
index 87c77dbf..dce01490 100644
--- a/test/tls_test.go
+++ b/test/tls_test.go
@@ -1935,3 +1935,48 @@ func TestTLSClientSVIDAuth(t *testing.T) {
 		})
 	}
 }
+
+func TestTLSPinnedCerts(t *testing.T) {
+	tmpl := `
+	host: localhost
+	port: -1
+	tls {
+		ca_file: "configs/certs/ca.pem"
+		cert_file: "configs/certs/server-cert.pem"
+		key_file: "configs/certs/server-key.pem"
+		# Require a client certificate and map user id from certificate
+		verify: true
+		pinned_certs: ["%s"]
+	}`
+
+	confFileName := createConfFile(t, []byte(fmt.Sprintf(tmpl, "aaaaaaaa09fde09451411ba3b42c0f74727d61a974c69fd3cf5257f39c75f0e9")))
+	defer removeFile(t, confFileName)
+	srv, o := RunServerWithConfig(confFileName)
+	defer srv.Shutdown()
+
+	if len(o.TLSPinnedCerts) != 1 {
+		t.Fatal("expected one pinned cert")
+	}
+
+	opts := []nats.Option{
+		nats.RootCAs("configs/certs/ca.pem"),
+		nats.ClientCert("./configs/certs/client-cert.pem", "./configs/certs/client-key.pem"),
+	}
+
+	nc, err := nats.Connect(srv.ClientURL(), opts...)
+	if err == nil {
+		nc.Close()
+		t.Fatalf("Expected error trying to connect without a certificate in pinned_certs")
+	}
+
+	ioutil.WriteFile(confFileName, []byte(fmt.Sprintf(tmpl, "bf6f821f09fde09451411ba3b42c0f74727d61a974c69fd3cf5257f39c75f0e9")), 0660)
+	if err := srv.Reload(); err != nil {
+		t.Fatalf("on Reload got %v", err)
+	}
+	// reload pinned to the certs used
+	nc, err = nats.Connect(srv.ClientURL(), opts...)
+	if err != nil {
+		t.Fatalf("Expected no error, got: %v", err)
+	}
+	nc.Close()
+}