Add support for TLS configuration for NATS

cfg/config.go

@@ -68,6 +68,9 @@ type NATSConfiguration struct {
 	SeedFile         string   `toml:"seed_file"`
 	CredsUser        string   `toml:"user_name"`
 	CredsPassword    string   `toml:"user_password"`
+	CAFile           string   `toml:"ca_file"`
+	CertFile         string   `toml:"cert_file"`
+	KeyFile          string   `toml:"key_file"`
 	BindAddress      string   `toml:"bind_address"`
 }
 

logstream/replicator.go

@@ -280,6 +280,24 @@ func (r *Replicator) ForceSaveSnapshot() {
 	r.lastSnapshot = time.Now()
 }
 
+func (r *Replicator) ReloadCertificates() error {
+	if cfg.Config.NATS.CAFile != "" {
+		err := nats.RootCAs(cfg.Config.NATS.CAFile)(&r.client.Opts)
+		if err != nil {
+			return err
+		}
+	}
+
+	if cfg.Config.NATS.CertFile != "" && cfg.Config.NATS.KeyFile != "" {
+		err := nats.ClientCert(cfg.Config.NATS.CertFile, cfg.Config.NATS.KeyFile)(&r.client.Opts)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (r *Replicator) invokeListener(callback func(payload []byte) error, msg *nats.Msg) error {
 	var err error
 	payload := msg.Data

stream/nats.go

@@ -19,7 +19,13 @@ func Connect() (*nats.Conn, error) {
 		return nil, err
 	}
 
+	tls, err := getNatsTLSFromConfig()
+	if err != nil {
+		return nil, err
+	}
+
 	opts = append(opts, creds...)
+	opts = append(opts, tls...)
 	if len(cfg.Config.NATS.URLs) == 0 {
 		embedded, err := startEmbeddedServer(cfg.Config.NodeName())
 		if err != nil {
@@ -54,3 +60,19 @@ func getNatsAuthFromConfig() ([]nats.Option, error) {
 
 	return opts, nil
 }
+
+func getNatsTLSFromConfig() ([]nats.Option, error) {
+	opts := make([]nats.Option, 0)
+
+	if cfg.Config.NATS.CAFile != "" {
+		opt := nats.RootCAs(cfg.Config.NATS.CAFile)
+		opts = append(opts, opt)
+	}
+
+	if cfg.Config.NATS.CertFile != "" && cfg.Config.NATS.KeyFile != "" {
+		opt := nats.ClientCert(cfg.Config.NATS.CertFile, cfg.Config.NATS.KeyFile)
+		opts = append(opts, opt)
+	}
+
+	return opts, nil
+}