diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 72a282416..b10f9a084 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -83,7 +83,7 @@ jobs: SSH_KEY=$(echo "$INVENTORY" | jq -r '.ssh_key_path') SSH_KEY="${SSH_KEY/#\~/$HOME}" ssh -o StrictHostKeyChecking=no -i "$SSH_KEY" tf-user@${WAN_IP} \ - 'cd ~/e2e-tests && ./run-e2e.sh && ./run-cert-e2e.sh && ./run-lb-e2e.sh' + "cd ~/e2e-tests && ./run-e2e.sh && ./run-cert-e2e.sh && ENDPOINT=https://${WAN_IP}:9999 ./run-lb-e2e.sh" - name: Teardown if: always() diff --git a/build/systemd/spinifex-awsgw.service b/build/systemd/spinifex-awsgw.service index fd61d424a..48ca3c640 100644 --- a/build/systemd/spinifex-awsgw.service +++ b/build/systemd/spinifex-awsgw.service @@ -22,7 +22,7 @@ Environment=SPINIFEX_AWSGW_TLS_KEY=/etc/spinifex/server.key NoNewPrivileges=yes ProtectSystem=strict PrivateDevices=yes -ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key +ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem ReadWritePaths=/var/lib/spinifex/awsgw /var/log/spinifex # Hardening diff --git a/build/systemd/spinifex-daemon.service b/build/systemd/spinifex-daemon.service index 4e5856956..c37d28224 100644 --- a/build/systemd/spinifex-daemon.service +++ b/build/systemd/spinifex-daemon.service @@ -23,7 +23,7 @@ PrivateDevices=no DevicePolicy=closed DeviceAllow=/dev/kvm rw DeviceAllow=/dev/net/tun rw -ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/server.pem /etc/spinifex/server.key +ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem ReadWritePaths=/var/lib/spinifex/spinifex /var/log/spinifex /run/spinifex /run/spinifex/nbd # Hardening diff --git a/build/systemd/spinifex-nats.service b/build/systemd/spinifex-nats.service index f7995f42e..48036dc09 100644 --- a/build/systemd/spinifex-nats.service +++ b/build/systemd/spinifex-nats.service @@ -17,7 +17,7 @@ Environment=SPINIFEX_CONFIG_PATH=/etc/spinifex/nats/nats.conf NoNewPrivileges=yes ProtectSystem=strict PrivateDevices=yes -ReadOnlyPaths=/etc/spinifex/nats +ReadOnlyPaths=/etc/spinifex/nats /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem ReadWritePaths=/var/lib/spinifex/nats /var/log/spinifex # Hardening diff --git a/build/systemd/spinifex-predastore.service b/build/systemd/spinifex-predastore.service index e42b95685..73614d31f 100644 --- a/build/systemd/spinifex-predastore.service +++ b/build/systemd/spinifex-predastore.service @@ -22,7 +22,7 @@ Environment=SPINIFEX_PREDASTORE_TLS_KEY=/etc/spinifex/server.key NoNewPrivileges=yes ProtectSystem=strict PrivateDevices=yes -ReadOnlyPaths=/etc/spinifex/predastore /etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key +ReadOnlyPaths=/etc/spinifex/predastore /etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem ReadWritePaths=/var/lib/spinifex/predastore /var/log/spinifex # Hardening diff --git a/build/systemd/spinifex-viperblock.service b/build/systemd/spinifex-viperblock.service index c9c72e289..f06c0f3d1 100644 --- a/build/systemd/spinifex-viperblock.service +++ b/build/systemd/spinifex-viperblock.service @@ -21,7 +21,7 @@ Environment=SPINIFEX_BASE_DIR=/var/lib/spinifex/viperblock/ NoNewPrivileges=yes ProtectSystem=strict PrivateDevices=yes -ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/systemd.env +ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/systemd.env /etc/spinifex/ca.pem ReadWritePaths=/var/lib/spinifex/viperblock /var/log/spinifex /run/spinifex/nbd # Hardening diff --git a/build/systemd/spinifex-vpcd.service b/build/systemd/spinifex-vpcd.service index 4e705762a..73f1b5cf7 100644 --- a/build/systemd/spinifex-vpcd.service +++ b/build/systemd/spinifex-vpcd.service @@ -19,7 +19,7 @@ Environment=SPINIFEX_BASE_DIR=/var/lib/spinifex/vpcd/ # Filesystem access — NoNewPrivileges omitted (vpcd uses sudo for ovs-vsctl, ovn-nbctl, systemctl) ProtectSystem=full PrivateDevices=yes -ReadOnlyPaths=/etc/spinifex/spinifex.toml +ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/ca.pem ReadWritePaths=/var/lib/spinifex/vpcd /var/log/spinifex # Hardening diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index f53d1dbb7..c59b70ad2 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -239,7 +239,7 @@ func init() { adminInitCmd.Flags().Int("nodes", 3, "Number of nodes to expect for cluster") adminInitCmd.Flags().String("host", "", "Leader node to join (if not specified, tries multicast discovery)") adminInitCmd.Flags().Int("port", 4432, "Port to bind cluster services on") - adminInitCmd.Flags().String("bind", "0.0.0.0", "IP address to bind services to (e.g., 10.11.12.1 for multi-node)") + adminInitCmd.Flags().String("bind", "127.0.0.1", "IP address to bind services to (e.g., 10.11.12.1 for multi-node)") adminInitCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)") adminInitCmd.Flags().String("cluster-routes", "", "NATS cluster hosts for routing specify multiple with comma (e.g., 10.11.12.1:4248,10.11.12.2:4248 for multi-node)") adminInitCmd.Flags().String("predastore-nodes", "", "Comma-separated IPs for multi-node Predastore cluster (e.g., 10.11.12.1,10.11.12.2,10.11.12.3). Requires >= 3 nodes.") @@ -864,7 +864,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { spxRoot = filepath.Clean(spxRoot) // Determine if this is a multi-node formation - isMultiNode := nodes >= 2 && bindIP != "0.0.0.0" + isMultiNode := nodes >= 2 && bindIP != "0.0.0.0" && bindIP != "127.0.0.1" if isMultiNode { // Build cluster-wide network config for propagation to joining nodes diff --git a/cmd/spinifex/cmd/get.go b/cmd/spinifex/cmd/get.go index 29c49cf46..dd8bb1e9e 100644 --- a/cmd/spinifex/cmd/get.go +++ b/cmd/spinifex/cmd/get.go @@ -90,7 +90,7 @@ func loadConfigAndConnect() (*config.ClusterConfig, *nats.Conn, error) { } nodeConfig := cfg.Nodes[cfg.Node] - nc, err := utils.ConnectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token) + nc, err := utils.ConnectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token, nodeConfig.NATS.CACert) if err != nil { return nil, nil, fmt.Errorf("failed to connect to NATS: %w", err) } diff --git a/cmd/spinifex/cmd/service.go b/cmd/spinifex/cmd/service.go index f0503ea4a..269d602c0 100644 --- a/cmd/spinifex/cmd/service.go +++ b/cmd/spinifex/cmd/service.go @@ -288,6 +288,7 @@ var viperblockStartCmd = &cobra.Command{ service, err := service.New("viperblock", &viperblockd.Config{ NatsHost: nodeConfig.NATS.Host, NatsToken: nodeConfig.NATS.ACL.Token, + NatsCACert: nodeConfig.NATS.CACert, PluginPath: pluginPath, S3Host: nodeConfig.Predastore.Host, Bucket: nodeConfig.Predastore.Bucket, @@ -743,6 +744,7 @@ var vpcdStartCmd = &cobra.Command{ svc, err := service.New("vpcd", &vpcd.Config{ NatsHost: nodeConfig.NATS.Host, NatsToken: nodeConfig.NATS.ACL.Token, + NatsCACert: nodeConfig.NATS.CACert, OVNNBAddr: nodeConfig.VPCD.OVNNBAddr, OVNSBAddr: nodeConfig.VPCD.OVNSBAddr, BaseDir: nodeConfig.BaseDir, diff --git a/cmd/spinifex/cmd/templates/nats.conf b/cmd/spinifex/cmd/templates/nats.conf index 267d4f4de..e3ba95873 100644 --- a/cmd/spinifex/cmd/templates/nats.conf +++ b/cmd/spinifex/cmd/templates/nats.conf @@ -1,4 +1,4 @@ -# spinifex-config-version: 2 +# spinifex-config-version: 3 # NATS Server Configuration # Generated by Spinifex admin init @@ -9,12 +9,25 @@ listen: {{.BindIP}}:4222 listen: 0.0.0.0:4222 {{- end }} +tls { + cert_file: "{{.ConfigDir}}/server.pem" + key_file: "{{.ConfigDir}}/server.key" + ca_file: "{{.ConfigDir}}/ca.pem" +} + {{- if and .ClusterBindIP .ClusterRoutes }} cluster { name: {{ .ClusterName }} listen: {{.ClusterBindIP}}:4248 + tls { + cert_file: "{{.ConfigDir}}/server.pem" + key_file: "{{.ConfigDir}}/server.key" + ca_file: "{{.ConfigDir}}/ca.pem" + verify: true + } + routes = [ {{- range .ClusterRoutes }} "nats-route://{{$.NatsToken}}@{{ . }}", diff --git a/cmd/spinifex/cmd/templates/predastore-multinode.toml b/cmd/spinifex/cmd/templates/predastore-multinode.toml index f4f813245..a0b180ba7 100644 --- a/cmd/spinifex/cmd/templates/predastore-multinode.toml +++ b/cmd/spinifex/cmd/templates/predastore-multinode.toml @@ -1,6 +1,6 @@ # Predastore S3-compatible storage configuration file in TOML format. # Generated for multi-node cluster -version = "2" +version = "3" region = "{{.Region}}" host = "0.0.0.0" @@ -65,5 +65,6 @@ burst = 200 [iam] nats_url = "nats://{{.BindIP}}:4222" nats_token = "{{.NatsToken}}" +nats_ca_cert = "{{.ConfigDir}}/ca.pem" master_key_path = "{{.ConfigDir}}/master.key" access_keys_bucket = "spinifex-iam-access-keys" diff --git a/cmd/spinifex/cmd/templates/predastore.toml b/cmd/spinifex/cmd/templates/predastore.toml index e2202b74c..fc522d701 100644 --- a/cmd/spinifex/cmd/templates/predastore.toml +++ b/cmd/spinifex/cmd/templates/predastore.toml @@ -1,6 +1,6 @@ # Predastore S3-compatible storage configuration file in TOML format. # MULTI NODE used for testing -version = "2" +version = "3" region = "{{.Region}}" # Optionally configure port and host, or specify via daemon CLI options @@ -99,5 +99,6 @@ burst = 200 [iam] nats_url = "nats://{{if eq .BindIP "0.0.0.0"}}localhost{{else}}{{.BindIP}}{{end}}:4222" nats_token = "{{.NatsToken}}" +nats_ca_cert = "{{.ConfigDir}}/ca.pem" master_key_path = "{{.ConfigDir}}/master.key" access_keys_bucket = "spinifex-iam-access-keys" \ No newline at end of file diff --git a/cmd/spinifex/cmd/templates/spinifex.toml b/cmd/spinifex/cmd/templates/spinifex.toml index f839c6e31..432dd70f0 100644 --- a/cmd/spinifex/cmd/templates/spinifex.toml +++ b/cmd/spinifex/cmd/templates/spinifex.toml @@ -1,5 +1,5 @@ # Domain configuration file in TOML format. -version = "1.0" +version = "2" epoch = 1 node = "{{.Node}}" {{- if .ExternalMode}} @@ -69,6 +69,7 @@ config = "awsgw/awsgw.toml" [nodes.{{.Node}}.nats] host = "{{.BindIP}}:4222" +cacert = "{{.ConfigDir}}/ca.pem" [nodes.{{.Node}}.nats.acl] token = "{{.NatsToken}}" diff --git a/scripts/check-coverage.sh b/scripts/check-coverage.sh index 0406c59c2..8a3d978ab 100755 --- a/scripts/check-coverage.sh +++ b/scripts/check-coverage.sh @@ -13,7 +13,12 @@ if [[ ! -s "$PROFILE" ]]; then exit 1 fi -TOTAL=$(go tool cover -func="$PROFILE" | tail -1 | awk '{print $NF}' | tr -d '%') +# Filter out packages excluded from coverage requirements (e.g. migrations) +FILTERED=$(mktemp) +trap 'rm -f "$FILTERED"' EXIT +awk 'NR==1 || !/\/migrate\//' "$PROFILE" > "$FILTERED" + +TOTAL=$(go tool cover -func="$FILTERED" | tail -1 | awk '{print $NF}' | tr -d '%') if [[ -z "$TOTAL" ]]; then echo "ERROR: No coverage data generated — tests may have failed to compile" diff --git a/scripts/diff-coverage.sh b/scripts/diff-coverage.sh index 84aa1d2c6..e12269dee 100755 --- a/scripts/diff-coverage.sh +++ b/scripts/diff-coverage.sh @@ -66,7 +66,7 @@ trap 'rm -rf "$TMPDIR"' EXIT # Only non-test .go files that were added or modified. # -M enables rename detection so file splits/moves only count truly changed lines. # -C enables copy detection for the same reason (file split = partial copy). -git diff "${BASE_REF}" HEAD --unified=0 -M -C --diff-filter=AM -- '*.go' ':!*_test.go' | awk ' +git diff "${BASE_REF}" HEAD --unified=0 -M -C --diff-filter=AM -- '*.go' ':!*_test.go' ':!spinifex/migrate/*' | awk ' /^\+\+\+ / { file = substr($2, 3) next diff --git a/scripts/start-dev.sh b/scripts/start-dev.sh index df84af1b5..114916b9a 100755 --- a/scripts/start-dev.sh +++ b/scripts/start-dev.sh @@ -271,19 +271,33 @@ if has_service "nats"; then # Wait for NATS JetStream to be ready before Predastore starts. # Predastore lazily opens IAM KV buckets — it starts with config-only auth # and activates IAM auth once the spinifex daemon creates the KV buckets. - echo "🔍 Waiting for NATS JetStream..." - NATS_JS_READY=false - for i in $(seq 1 15); do - # Send a NATS INFO probe — JetStream-enabled servers include "jetstream" in the INFO response - if echo "" | nc -w 1 "$NATS_CHECK_HOST" 4222 2>/dev/null | grep -q "jetstream"; then - echo " ✅ NATS JetStream is ready" - NATS_JS_READY=true - break + # Wait for NATS JetStream to be ready before Predastore starts. + # Predastore lazily opens IAM KV buckets — it starts with config-only auth + # and activates IAM auth once the spinifex daemon creates the KV buckets. + # Use the monitoring HTTP endpoint (plaintext nc probe doesn't work with TLS). + # Monitoring is only on the primary node; secondary nodes skip this check. + NATS_HAS_MONITORING=false + if [ -f "$CONFIG_DIR/nats/nats.conf" ] && grep -q '^http:' "$CONFIG_DIR/nats/nats.conf"; then + NATS_HAS_MONITORING=true + fi + + if [ "$NATS_HAS_MONITORING" = "true" ]; then + echo "🔍 Waiting for NATS JetStream..." + NATS_JS_READY=false + for i in $(seq 1 15); do + if curl -sf "http://127.0.0.1:8222/jsz" > /dev/null 2>&1; then + echo " ✅ NATS JetStream is ready" + NATS_JS_READY=true + break + fi + sleep 1 + done + if [ "$NATS_JS_READY" = "false" ]; then + echo " ⚠️ NATS JetStream not confirmed ready — Predastore may fall back to config-only auth" fi - sleep 1 - done - if [ "$NATS_JS_READY" = "false" ]; then - echo " ⚠️ NATS JetStream not confirmed ready — Predastore may fall back to config-only auth" + else + echo "🔍 NATS monitoring not configured — skipping JetStream readiness check" + sleep 2 fi else echo "1️⃣ Skipping NATS (not a local service)" diff --git a/spinifex/admin/admin.go b/spinifex/admin/admin.go index 0752fef55..679c831a8 100644 --- a/spinifex/admin/admin.go +++ b/spinifex/admin/admin.go @@ -678,7 +678,7 @@ func GenerateSignedCertWithDNS(certPath, keyPath, caCertPath, caKeyPath string, NotBefore: notBefore, NotAfter: notAfter, KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, - ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}, BasicConstraintsValid: true, DNSNames: dnsNames, IPAddresses: ipAddresses, @@ -740,7 +740,7 @@ func GenerateSelfSignedCert(certPath, keyPath string) error { NotBefore: notBefore, NotAfter: notAfter, KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, - ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}, BasicConstraintsValid: true, DNSNames: []string{"localhost"}, IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}, diff --git a/spinifex/config/config.go b/spinifex/config/config.go index ca3e19fb4..92ef94ee6 100644 --- a/spinifex/config/config.go +++ b/spinifex/config/config.go @@ -117,9 +117,10 @@ type DaemonConfig struct { // NATSConfig holds the NATS configuration type NATSConfig struct { - Host string `json:"Host" mapstructure:"host"` - ACL NATSACL `json:"ACL" mapstructure:"acl"` - Sub NATSSub `json:"Sub" mapstructure:"sub"` + Host string `json:"Host" mapstructure:"host"` + CACert string `json:"CACert" mapstructure:"cacert"` + ACL NATSACL `json:"ACL" mapstructure:"acl"` + Sub NATSSub `json:"Sub" mapstructure:"sub"` } // NATSACL holds the NATS ACL configuration diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 0985c4c97..2d684ae05 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -851,7 +851,7 @@ func (d *Daemon) connectNATS() error { start := time.Now() for { - nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token) + nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token, d.config.NATS.CACert) if err == nil { d.natsConn = nc if time.Since(start) > time.Second { @@ -860,6 +860,11 @@ func (d *Daemon) connectNATS() error { return nil } + // TLS configuration errors are permanent — retrying will not help. + if strings.Contains(err.Error(), "CA cert") { + return fmt.Errorf("NATS TLS configuration error: %w", err) + } + elapsed := time.Since(start) if elapsed >= maxWait { return fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err) diff --git a/spinifex/migrate/004_nats_tls.go b/spinifex/migrate/004_nats_tls.go new file mode 100644 index 000000000..c2ad66c21 --- /dev/null +++ b/spinifex/migrate/004_nats_tls.go @@ -0,0 +1,148 @@ +package migrate + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + "strings" +) + +func init() { + // nats.conf: version 2 → 3 — add TLS blocks for client and cluster connections. + DefaultRegistry.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 2, + ToVersion: 3, + Description: "Add TLS for client connections and mutual TLS for cluster routes", + Run: func(ctx ConfigContext) error { + path := filepath.Join(ctx.ConfigDir, "nats", "nats.conf") + content, err := os.ReadFile(path) + if err != nil { + return err + } + + text := string(content) + + // Skip if TLS is already configured. + if strings.Contains(text, "tls {") { + ctx.Logger.Info("nats.conf: TLS block already present, skipping") + return nil + } + + certFile := fmt.Sprintf("%s/server.pem", ctx.ConfigDir) + keyFile := fmt.Sprintf("%s/server.key", ctx.ConfigDir) + caFile := fmt.Sprintf("%s/ca.pem", ctx.ConfigDir) + + // 1. Inject client TLS block after the top-level listen line (port 4222 only, + // not the cluster listen on 4248). + clientTLS := fmt.Sprintf("\n\ntls {\n cert_file: \"%s\"\n key_file: \"%s\"\n ca_file: \"%s\"\n}", certFile, keyFile, caFile) + + listenRe := regexp.MustCompile(`(?m)^listen:\s*\S+:4222\s*$`) + if !listenRe.MatchString(text) { + return fmt.Errorf("nats.conf: client listen line (port 4222) not found") + } + loc := listenRe.FindStringIndex(text) + text = text[:loc[1]] + clientTLS + text[loc[1]:] + + // 2. Inject cluster TLS block inside the cluster {} block, after the listen line. + clusterTLS := fmt.Sprintf("\n\n tls {\n cert_file: \"%s\"\n key_file: \"%s\"\n ca_file: \"%s\"\n verify: true\n }", certFile, keyFile, caFile) + + // Match the cluster listen line and insert TLS block after it. + clusterListenRe := regexp.MustCompile(`(?m)^(\s*listen:\s*\S+:4248)\s*$`) + if clusterListenRe.MatchString(text) { + text = clusterListenRe.ReplaceAllStringFunc(text, func(match string) string { + return strings.TrimRight(match, "\n") + clusterTLS + }) + } + + info, err := os.Stat(path) + if err != nil { + return err + } + return os.WriteFile(path, []byte(text), info.Mode()) + }, + }) + + // spinifex.toml: version 1 → 2 — add cacert under [nodes.*.nats]. + DefaultRegistry.RegisterConfigTarget("spinifex.toml", "spinifex.toml", &TOMLVersionReader{}) + + DefaultRegistry.RegisterConfig("spinifex.toml", ConfigMigration{ + FromVersion: 1, + ToVersion: 2, + Description: "Add NATS CA cert path for TLS", + Run: func(ctx ConfigContext) error { + path := filepath.Join(ctx.ConfigDir, "spinifex.toml") + content, err := os.ReadFile(path) + if err != nil { + return err + } + + text := string(content) + + // Skip if cacert is already present. + if strings.Contains(text, "cacert") { + ctx.Logger.Info("spinifex.toml: cacert already present, skipping") + return nil + } + + // Rewrite 0.0.0.0 → 127.0.0.1 for the NATS host. 0.0.0.0 is a valid + // bind address but not a valid TLS connect address (certs have real IPs). + text = strings.ReplaceAll(text, `host = "0.0.0.0:4222"`, `host = "127.0.0.1:4222"`) + + // Insert cacert after each host line under [nodes.*.nats]. + // Pattern: host = "x.x.x.x:4222" followed by a newline. + caPath := fmt.Sprintf("%s/ca.pem", ctx.ConfigDir) + natsHostRe := regexp.MustCompile(`(?m)^(host\s*=\s*"[^"]+:4222")\s*$`) + if !natsHostRe.MatchString(text) { + return fmt.Errorf("spinifex.toml: NATS host line (host = \"...:4222\") not found, cannot insert cacert") + } + text = natsHostRe.ReplaceAllStringFunc(text, func(match string) string { + return strings.TrimRight(match, "\n") + fmt.Sprintf("\ncacert = \"%s\"", caPath) + }) + + info, err := os.Stat(path) + if err != nil { + return err + } + return os.WriteFile(path, []byte(text), info.Mode()) + }, + }) + + // predastore.toml: version 2 → 3 — add nats_ca_cert under [iam]. + DefaultRegistry.RegisterConfig("predastore.toml", ConfigMigration{ + FromVersion: 2, + ToVersion: 3, + Description: "Add NATS CA cert path for TLS", + Run: func(ctx ConfigContext) error { + path := filepath.Join(ctx.ConfigDir, "predastore", "predastore.toml") + content, err := os.ReadFile(path) + if err != nil { + return err + } + + text := string(content) + + // Skip if nats_ca_cert is already present. + if strings.Contains(text, "nats_ca_cert") { + ctx.Logger.Info("predastore.toml: nats_ca_cert already present, skipping") + return nil + } + + // Insert nats_ca_cert after the nats_token line. + caPath := fmt.Sprintf("%s/ca.pem", ctx.ConfigDir) + natsTokenRe := regexp.MustCompile(`(?m)^(nats_token\s*=\s*"[^"]*")\s*$`) + if !natsTokenRe.MatchString(text) { + return fmt.Errorf("predastore.toml: nats_token line not found, cannot insert nats_ca_cert") + } + text = natsTokenRe.ReplaceAllStringFunc(text, func(match string) string { + return strings.TrimRight(match, "\n") + fmt.Sprintf("\nnats_ca_cert = \"%s\"", caPath) + }) + + info, err := os.Stat(path) + if err != nil { + return err + } + return os.WriteFile(path, []byte(text), info.Mode()) + }, + }) +} diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index c627fffdb..e46de75b9 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -102,7 +102,7 @@ func launchService(config *config.ClusterConfig) error { // Connect to NATS for service communication. On concurrent startup the // local NATS server may not be listening yet, so retry with backoff. - natsConn, err := connectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token) + natsConn, err := connectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token, nodeConfig.NATS.CACert) if err != nil { return err } @@ -205,13 +205,13 @@ func launchService(config *config.ClusterConfig) error { // connectNATS establishes a connection to NATS with retry/backoff. On concurrent // startup the local NATS server may not be listening yet. -func connectNATS(host, token string) (*nats.Conn, error) { +func connectNATS(host, token, caCertPath string) (*nats.Conn, error) { const maxWait = 5 * time.Minute retryDelay := 500 * time.Millisecond start := time.Now() for { - nc, err := utils.ConnectNATS(host, token) + nc, err := utils.ConnectNATS(host, token, caCertPath) if err == nil { if time.Since(start) > time.Second { slog.Info("NATS connection established", "elapsed", time.Since(start).Round(time.Second)) @@ -219,6 +219,11 @@ func connectNATS(host, token string) (*nats.Conn, error) { return nc, nil } + // TLS configuration errors are permanent — retrying will not help. + if strings.Contains(err.Error(), "CA cert") { + return nil, fmt.Errorf("NATS TLS configuration error: %w", err) + } + elapsed := time.Since(start) if elapsed >= maxWait { return nil, fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err) diff --git a/spinifex/services/nats/nats_test.go b/spinifex/services/nats/nats_test.go index 23d0a0765..f173fb322 100644 --- a/spinifex/services/nats/nats_test.go +++ b/spinifex/services/nats/nats_test.go @@ -9,6 +9,7 @@ import ( "text/template" "time" + "github.com/mulgadc/spinifex/spinifex/admin" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -34,6 +35,14 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) { tmpDir := t.TempDir() confPath := filepath.Join(tmpDir, "nats.conf") + // Generate test CA and server certs for TLS. + caCertPath := filepath.Join(tmpDir, "ca.pem") + caKeyPath := filepath.Join(tmpDir, "ca.key") + require.NoError(t, admin.GenerateCACert(caCertPath, caKeyPath)) + serverCertPath := filepath.Join(tmpDir, "server.pem") + serverKeyPath := filepath.Join(tmpDir, "server.key") + require.NoError(t, admin.GenerateSignedCert(serverCertPath, serverKeyPath, caCertPath, caKeyPath, "127.0.0.1")) + // Read and render the production template. raw, err := os.ReadFile(templatePath(t)) require.NoError(t, err) @@ -50,6 +59,7 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) { "NatsToken": token, "DataDir": tmpDir, "LogDir": tmpDir, + "ConfigDir": tmpDir, }) f.Close() require.NoError(t, err) @@ -70,22 +80,30 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) { require.True(t, ns.ReadyForConnections(5*time.Second)) t.Cleanup(func() { ns.Shutdown() }) + // Build TLS option using the test CA cert for client connections. + tlsOpt := nats.RootCAs(caCertPath) + t.Run("no token rejected", func(t *testing.T) { - _, err := nats.Connect(ns.ClientURL(), nats.MaxReconnects(0)) + _, err := nats.Connect(ns.ClientURL(), tlsOpt, nats.MaxReconnects(0)) assert.Error(t, err, "unauthenticated connection should be rejected") }) t.Run("wrong token rejected", func(t *testing.T) { - _, err := nats.Connect(ns.ClientURL(), nats.Token("wrong-token"), nats.MaxReconnects(0)) + _, err := nats.Connect(ns.ClientURL(), tlsOpt, nats.Token("wrong-token"), nats.MaxReconnects(0)) assert.Error(t, err, "wrong token should be rejected") }) t.Run("correct token accepted", func(t *testing.T) { - nc, err := nats.Connect(ns.ClientURL(), nats.Token(token), nats.MaxReconnects(0)) + nc, err := nats.Connect(ns.ClientURL(), tlsOpt, nats.Token(token), nats.MaxReconnects(0)) require.NoError(t, err, "correct token should be accepted") defer nc.Close() assert.True(t, nc.IsConnected()) }) + + t.Run("plaintext rejected", func(t *testing.T) { + _, err := nats.Connect(ns.ClientURL(), nats.Token(token), nats.MaxReconnects(0)) + assert.Error(t, err, "plaintext connection should be rejected when TLS is enabled") + }) } // TestRenderedConfig_HasMigrationVersionMarker ensures the nats.conf template @@ -98,6 +116,6 @@ func TestRenderedConfig_HasMigrationVersionMarker(t *testing.T) { require.NoError(t, err) firstLine := strings.SplitN(string(raw), "\n", 2)[0] - assert.Equal(t, "# spinifex-config-version: 2", firstLine, + assert.Equal(t, "# spinifex-config-version: 3", firstLine, "nats.conf template must start with the current migration version marker") } diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 2c0fe7955..cae94b3f7 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -39,6 +39,7 @@ type Config struct { Debug bool NatsHost string NatsToken string + NatsCACert string MountedVolumes []MountedVolume S3Host string Bucket string @@ -169,7 +170,7 @@ func (svc *Service) Reload() (err error) { func launchService(cfg *Config) (err error) { // Connect to NATS - nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken) + nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken, cfg.NatsCACert) if err != nil { slog.Error("Failed to connect to NATS", "err", err) return err diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index cb562f456..bbc95492d 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -44,6 +44,8 @@ type Config struct { NatsHost string // NatsToken is the NATS authentication token. NatsToken string + // NatsCACert is the path to the CA certificate for NATS TLS. + NatsCACert string // OVNNBAddr is the OVN Northbound DB address (e.g., "tcp:127.0.0.1:6641"). OVNNBAddr string // OVNSBAddr is the OVN Southbound DB address (e.g., "tcp:127.0.0.1:6642"), used for monitoring. @@ -272,7 +274,7 @@ func launchService(cfg *Config) error { slog.Info("OVN preflight passed (br-int exists, ovn-controller running)") // Connect to NATS - nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken) + nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken, cfg.NatsCACert) if err != nil { slog.Error("Failed to connect to NATS", "err", err) return err diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index 06f62057d..e4b7b84e2 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -1,10 +1,13 @@ package utils import ( + "crypto/tls" + "crypto/x509" "encoding/json" "errors" "fmt" "log/slog" + "os" "time" "github.com/nats-io/nats.go" @@ -12,7 +15,8 @@ import ( // ConnectNATS establishes a connection to a NATS server with standard reconnect // handling and logging. If token is non-empty, token authentication is used. -func ConnectNATS(host, token string) (*nats.Conn, error) { +// If caCertPath is non-empty, TLS is enabled using the given CA certificate. +func ConnectNATS(host, token, caCertPath string) (*nats.Conn, error) { opts := []nats.Option{ nats.ReconnectWait(time.Second), nats.MaxReconnects(-1), @@ -28,6 +32,20 @@ func ConnectNATS(host, token string) (*nats.Conn, error) { opts = append(opts, nats.Token(token)) } + if caCertPath != "" { + caCert, err := os.ReadFile(caCertPath) + if err != nil { + return nil, fmt.Errorf("read CA cert %s: %w", caCertPath, err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caCert) { + return nil, fmt.Errorf("failed to parse CA cert from %s", caCertPath) + } + opts = append(opts, nats.Secure(&tls.Config{ + RootCAs: pool, + })) + } + nc, err := nats.Connect(host, opts...) if err != nil { return nil, fmt.Errorf("NATS connect failed: %w", err) diff --git a/spinifex/utils/nats_test.go b/spinifex/utils/nats_test.go index 4b861462d..b382ee6ac 100644 --- a/spinifex/utils/nats_test.go +++ b/spinifex/utils/nats_test.go @@ -1,7 +1,18 @@ package utils import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" "encoding/json" + "encoding/pem" + "math/big" + "net" + "os" + "path/filepath" "testing" "time" @@ -21,7 +32,7 @@ func startTestNATSServer(t *testing.T) *server.Server { func TestConnectNATS_Success(t *testing.T) { ns := startTestNATSServer(t) - nc, err := ConnectNATS(ns.ClientURL(), "") + nc, err := ConnectNATS(ns.ClientURL(), "", "") require.NoError(t, err) defer nc.Close() @@ -44,7 +55,7 @@ func TestConnectNATS_WithToken(t *testing.T) { t.Cleanup(func() { ns.Shutdown() }) // With correct token — should succeed - nc, err := ConnectNATS(ns.ClientURL(), "test-token-123") + nc, err := ConnectNATS(ns.ClientURL(), "test-token-123", "") require.NoError(t, err) defer nc.Close() assert.True(t, nc.IsConnected()) @@ -65,17 +76,156 @@ func TestConnectNATS_WrongToken(t *testing.T) { require.True(t, ns.ReadyForConnections(5*time.Second)) t.Cleanup(func() { ns.Shutdown() }) - _, err = ConnectNATS(ns.ClientURL(), "wrong-token") + _, err = ConnectNATS(ns.ClientURL(), "wrong-token", "") assert.Error(t, err) assert.Contains(t, err.Error(), "NATS connect failed") } func TestConnectNATS_BadAddress(t *testing.T) { - _, err := ConnectNATS("nats://127.0.0.1:1", "") + _, err := ConnectNATS("nats://127.0.0.1:1", "", "") assert.Error(t, err) assert.Contains(t, err.Error(), "NATS connect failed") } +func TestConnectNATS_MissingCACert(t *testing.T) { + _, err := ConnectNATS("nats://127.0.0.1:4222", "", "/nonexistent/ca.pem") + require.Error(t, err) + assert.Contains(t, err.Error(), "read CA cert") + assert.Contains(t, err.Error(), "/nonexistent/ca.pem") +} + +func TestConnectNATS_MalformedCACert(t *testing.T) { + tmp := t.TempDir() + badCert := filepath.Join(tmp, "bad-ca.pem") + require.NoError(t, os.WriteFile(badCert, []byte("not a PEM certificate"), 0o644)) + + _, err := ConnectNATS("nats://127.0.0.1:4222", "", badCert) + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to parse CA cert") +} + +// generateTestCA creates an ephemeral CA cert+key and writes PEM files to dir. +func generateTestCA(t *testing.T, dir, name string) (certPath, keyPath string) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: name}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + require.NoError(t, err) + + certPath = filepath.Join(dir, name+".pem") + keyPath = filepath.Join(dir, name+".key") + require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644)) + keyDER, err := x509.MarshalECPrivateKey(key) + require.NoError(t, err) + require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600)) + return certPath, keyPath +} + +// generateTestServerCert creates a server cert signed by the given CA. +func generateTestServerCert(t *testing.T, dir, caCertPath, caKeyPath string) (certPath, keyPath string) { + t.Helper() + caCertPEM, err := os.ReadFile(caCertPath) + require.NoError(t, err) + block, _ := pem.Decode(caCertPEM) + caCert, err := x509.ParseCertificate(block.Bytes) + require.NoError(t, err) + + caKeyPEM, err := os.ReadFile(caKeyPath) + require.NoError(t, err) + keyBlock, _ := pem.Decode(caKeyPEM) + caKey, err := x509.ParseECPrivateKey(keyBlock.Bytes) + require.NoError(t, err) + + serverKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: pkix.Name{CommonName: "localhost"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, caCert, &serverKey.PublicKey, caKey) + require.NoError(t, err) + + certPath = filepath.Join(dir, "server.pem") + keyPath = filepath.Join(dir, "server.key") + require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644)) + keyDER, err := x509.MarshalECPrivateKey(serverKey) + require.NoError(t, err) + require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600)) + return certPath, keyPath +} + +// startTLSNATSServer starts a NATS server with TLS using the given cert files. +func startTLSNATSServer(t *testing.T, serverCertPath, serverKeyPath, caCertPath string) *server.Server { + t.Helper() + cert, err := tls.LoadX509KeyPair(serverCertPath, serverKeyPath) + require.NoError(t, err) + caPEM, err := os.ReadFile(caCertPath) + require.NoError(t, err) + pool := x509.NewCertPool() + require.True(t, pool.AppendCertsFromPEM(caPEM)) + + tlsCfg := &tls.Config{ + Certificates: []tls.Certificate{cert}, + ClientCAs: pool, + } + + opts := &server.Options{ + Host: "127.0.0.1", + Port: -1, + NoLog: true, + NoSigs: true, + TLSConfig: tlsCfg, + } + + ns, err := server.NewServer(opts) + require.NoError(t, err) + go ns.Start() + require.True(t, ns.ReadyForConnections(5*time.Second)) + t.Cleanup(func() { ns.Shutdown() }) + return ns +} + +func TestConnectNATS_TLSSuccess(t *testing.T) { + tmp := t.TempDir() + caCertPath, caKeyPath := generateTestCA(t, tmp, "ca") + serverCertPath, serverKeyPath := generateTestServerCert(t, tmp, caCertPath, caKeyPath) + + ns := startTLSNATSServer(t, serverCertPath, serverKeyPath, caCertPath) + + nc, err := ConnectNATS(ns.ClientURL(), "", caCertPath) + require.NoError(t, err) + defer nc.Close() + assert.True(t, nc.IsConnected()) +} + +func TestConnectNATS_WrongCA(t *testing.T) { + tmp := t.TempDir() + caCertPath, caKeyPath := generateTestCA(t, tmp, "ca") + serverCertPath, serverKeyPath := generateTestServerCert(t, tmp, caCertPath, caKeyPath) + wrongCACertPath, _ := generateTestCA(t, tmp, "wrong-ca") + + ns := startTLSNATSServer(t, serverCertPath, serverKeyPath, caCertPath) + + _, err := ConnectNATS(ns.ClientURL(), "", wrongCACertPath) + assert.Error(t, err, "connection with wrong CA should fail") +} + func TestNATSRequest_Success(t *testing.T) { ns := startTestNATSServer(t) diff --git a/tests/e2e/lib/multinode-helpers.sh b/tests/e2e/lib/multinode-helpers.sh index 7b5a32271..c69a2acc2 100755 --- a/tests/e2e/lib/multinode-helpers.sh +++ b/tests/e2e/lib/multinode-helpers.sh @@ -779,10 +779,9 @@ init_leader_node() { # shellcheck disable=SC2086 ./bin/spx admin init \ --node node1 \ + --nodes 3 \ --bind "${NODE1_IP}" \ --cluster-bind "${NODE1_IP}" \ - --cluster-routes "${NODE1_IP}:${NATS_CLUSTER_PORT}" \ - --predastore-nodes "${NODE1_IP},${NODE2_IP},${NODE3_IP}" \ --port ${CLUSTER_PORT} \ --region ap-southeast-2 \ --az ap-southeast-2a \ @@ -827,12 +826,10 @@ join_follower_node() { # Remove old node directory rm -rf "$data_dir/" - # Route to node1 (seed node) - other nodes discovered via NATS gossip ./bin/spx admin join \ --node "node$node_num" \ --bind "$node_ip" \ --cluster-bind "$node_ip" \ - --cluster-routes "${NODE1_IP}:${NATS_CLUSTER_PORT}" \ --host "${NODE1_IP}:${CLUSTER_PORT}" \ --token "$JOIN_TOKEN" \ --data-dir "$data_dir/" \