diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index 3251198a7..872455236 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -23,8 +23,10 @@ import ( _ "embed" "encoding/base64" "encoding/json" + "errors" "fmt" "io" + "log/slog" "maps" "net" "net/http" @@ -242,6 +244,7 @@ func init() { 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.") adminInitCmd.Flags().String("formation-timeout", "10m", "Timeout for cluster formation (e.g., 5m, 30s)") + adminInitCmd.Flags().String("token-ttl", "30m", "Join token validity duration (e.g. 30m, 1h, 2h)") adminInitCmd.Flags().String("cluster-name", "spinifex", "NATS cluster name") adminInitCmd.Flags().Bool("no-telemetry", false, "Disable telemetry metrics sent during init (default: enabled)") adminInitCmd.Flags().StringSlice("services", nil, "Services this node runs (default: all). Valid: nats,predastore,viperblock,daemon,awsgw,ui") @@ -266,10 +269,12 @@ func init() { adminJoinCmd.Flags().String("bind", "0.0.0.0", "IP address to bind services to (e.g., 10.11.12.2 for multi-node on single host)") adminJoinCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)") adminJoinCmd.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)") + adminJoinCmd.Flags().String("token", "", "Join token from the init node (required)") adminJoinCmd.Flags().StringSlice("services", nil, "Services this node runs (default: all)") adminJoinCmd.Flags().Bool("no-telemetry", false, "Disable telemetry metrics sent during join (default: enabled)") adminJoinCmd.MarkFlagRequired("node") adminJoinCmd.MarkFlagRequired("host") + adminJoinCmd.MarkFlagRequired("token") imagesImportCmd.Flags().String("tmp-dir", os.TempDir(), "Temporary directory for image import processing") @@ -600,6 +605,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { } predastoreNodesStr, _ := cmd.Flags().GetString("predastore-nodes") formationTimeoutStr, _ := cmd.Flags().GetString("formation-timeout") + tokenTTLStr, _ := cmd.Flags().GetString("token-ttl") clusterName, _ := cmd.Flags().GetString("cluster-name") services, _ := cmd.Flags().GetStringSlice("services") @@ -889,7 +895,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { runAdminInitMultiNode(cmd, accessKey, secretKey, accountID, natsToken, clusterName, configDir, spxRoot, certPath, region, az, node, bindIP, clusterBind, - port, nodes, formationTimeoutStr, services, networkConfig) + port, nodes, formationTimeoutStr, tokenTTLStr, services, networkConfig) return } @@ -1073,13 +1079,29 @@ func runAdminInit(cmd *cobra.Command, args []string) { // then generates configs with complete cluster topology. func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, natsToken, clusterName, configDir, spxRoot, certPath, region, az, node, bindIP, clusterBind string, - port, expectedNodes int, formationTimeoutStr string, services []string, networkConfig *formation.NetworkConfig) { + port, expectedNodes int, formationTimeoutStr, tokenTTLStr string, services []string, networkConfig *formation.NetworkConfig) { formationTimeout, err := time.ParseDuration(formationTimeoutStr) if err != nil { fmt.Fprintf(os.Stderr, "❌ Error: Invalid --formation-timeout: %v\n", err) os.Exit(1) } + tokenTTL, err := time.ParseDuration(tokenTTLStr) + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error: Invalid --token-ttl: %v\n", err) + os.Exit(1) + } + if tokenTTL < formationTimeout+1*time.Minute { + fmt.Fprintf(os.Stderr, "❌ Error: --token-ttl (%s) must be >= --formation-timeout + 1m (%s)\n", tokenTTL, formationTimeout+1*time.Minute) + os.Exit(1) + } + + joinToken, err := formation.GenerateJoinToken() + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error generating join token: %v\n", err) + os.Exit(1) + } + // Generate IAM master key for the cluster masterKey, err := handlers_iam.GenerateMasterKey() if err != nil { @@ -1118,7 +1140,7 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, AdminSecretKey: bootstrapResult.AdminSecretKey, } - fs := formation.NewFormationServer(expectedNodes, creds, string(caCertData), string(caKeyData), networkConfig) + fs := formation.NewFormationServer(expectedNodes, creds, string(caCertData), string(caKeyData), networkConfig, joinToken, tokenTTL) // Include master key in formation server for distribution to joining nodes fs.SetMasterKey(base64.StdEncoding.EncodeToString(masterKey)) @@ -1137,6 +1159,13 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, os.Exit(1) } + // Write join token to file for automated workflows + tokenPath := filepath.Join(configDir, "join-token") + if err := os.WriteFile(tokenPath, []byte(joinToken), 0600); err != nil { + fmt.Fprintf(os.Stderr, "❌ Error writing join token: %v\n", err) + os.Exit(1) + } + // Start formation server formationAddr := fmt.Sprintf("%s:%d", bindIP, port) if err := fs.Start(formationAddr); err != nil { @@ -1146,12 +1175,15 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, fmt.Printf("\n📡 Formation server started on %s\n", formationAddr) fmt.Printf(" Waiting for %d more node(s) to join...\n", expectedNodes-1) - fmt.Printf(" Other nodes should run: spx admin join --host %s --node --bind \n\n", formationAddr) + fmt.Printf(" Token expires in %s\n\n", tokenTTL) + fmt.Printf(" Other nodes should run:\n") + fmt.Printf(" spx admin join --host %s --token %s --node --bind \n\n", formationAddr, joinToken) // Wait for all nodes to register if err := fs.WaitForCompletion(formationTimeout); err != nil { fmt.Fprintf(os.Stderr, "❌ %v\n", err) fs.Shutdown(context.Background()) + os.Remove(tokenPath) os.Exit(1) } @@ -1295,6 +1327,9 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, // Shutdown formation server fs.Shutdown(context.Background()) + if err := os.Remove(tokenPath); err != nil && !errors.Is(err, os.ErrNotExist) { + slog.Warn("Failed to remove join token file", "path", tokenPath, "error", err) + } // Print cluster summary fmt.Println("\n🎉 Cluster formation complete!") @@ -1319,6 +1354,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) { node, _ := cmd.Flags().GetString("node") leaderHost, _ := cmd.Flags().GetString("host") + joinToken, _ := cmd.Flags().GetString("token") region, _ := cmd.Flags().GetString("region") az, _ := cmd.Flags().GetString("az") dataDir, _ := cmd.Flags().GetString("data-dir") @@ -1401,7 +1437,15 @@ func runAdminJoin(cmd *cobra.Command, args []string) { } joinURL := fmt.Sprintf("https://%s/formation/join", leaderHost) - resp, err := client.Post(joinURL, "application/json", bytes.NewBuffer(reqBody)) + req, err := http.NewRequest(http.MethodPost, joinURL, bytes.NewBuffer(reqBody)) + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error creating join request: %v\n", err) + os.Exit(1) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+joinToken) + + resp, err := client.Do(req) if err != nil { fmt.Fprintf(os.Stderr, "❌ Error connecting to formation server: %v\n", err) fmt.Fprintf(os.Stderr, "Make sure the leader node has run 'spx admin init' and is accessible at %s\n", leaderHost) @@ -1433,7 +1477,14 @@ func runAdminJoin(cmd *cobra.Command, args []string) { var statusResp formation.StatusResponse for { - sResp, err := client.Get(statusURL) + statusReq, err := http.NewRequest(http.MethodGet, statusURL, nil) + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error creating status request: %v\n", err) + os.Exit(1) + } + statusReq.Header.Set("Authorization", "Bearer "+joinToken) + + sResp, err := client.Do(statusReq) if err != nil { fmt.Fprintf(os.Stderr, "❌ Error polling formation status: %v\n", err) os.Exit(1) @@ -1446,8 +1497,17 @@ func runAdminJoin(cmd *cobra.Command, args []string) { os.Exit(1) } + if sResp.StatusCode == http.StatusUnauthorized { + fmt.Fprintf(os.Stderr, "❌ Error: join token rejected by formation server (expired or invalid)\n") + os.Exit(1) + } + if sResp.StatusCode != http.StatusOK { + fmt.Fprintf(os.Stderr, "❌ Error: unexpected status %d from formation server\n", sResp.StatusCode) + os.Exit(1) + } + if err := json.Unmarshal(sBody, &statusResp); err != nil { - fmt.Fprintf(os.Stderr, "Error parsing status response: %v\n", err) + fmt.Fprintf(os.Stderr, "❌ Error parsing status response: %v\n", err) os.Exit(1) } diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index f3850d005..90410eeb0 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -51,8 +51,8 @@ Operational commands for inspecting cluster state. These fan out NATS requests t | Command | Flags | Prerequisites | Basic Logic | Test Cases | Status | |---------|-------|---------------|-------------|------------|--------| -| `spx admin init` | `--nodes`, `--node`, `--bind`, `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--services`, `--formation-timeout`, `--force` | None (first-time setup) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` | 1. Init creates all config files
2. Root credentials printed once
3. master.key is 32 bytes, mode 0600
4. bootstrap.json consumed on first start
5. `--force` re-initializes existing config
6. AWS CLI profile `spx` auto-configured | **DONE** | -| `spx admin join` | `--host` (required), `--node` (required), `--bind`, `--port`, `--region`, `--az`, `--cluster-bind`, `--cluster-routes`, `--data-dir`, `--services` | Leader node must be running | Connects to leader node → retrieves cluster configuration → configures local node to join cluster and participate in distributed operations | 1. Join existing cluster
2. Missing host (error)
3. Missing node name (error) | **DONE** | +| `spx admin init` | `--nodes`, `--node`, `--bind`, `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--services`, `--formation-timeout`, `--token-ttl`, `--force` | None (first-time setup) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → generates join token (written to `join-token` file, displayed in join command) → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` | 1. Init creates all config files
2. Root credentials printed once
3. master.key is 32 bytes, mode 0600
4. bootstrap.json consumed on first start
5. `--force` re-initializes existing config
6. AWS CLI profile `spx` auto-configured
7. Multi-node init generates join token and writes to `/join-token` | **DONE** | +| `spx admin join` | `--host` (required), `--node` (required), `--token` (required), `--bind`, `--port`, `--region`, `--az`, `--cluster-bind`, `--cluster-routes`, `--data-dir`, `--services` | Leader node must be running | Connects to leader node with join token (Authorization: Bearer header) → retrieves cluster configuration → configures local node to join cluster and participate in distributed operations | 1. Join existing cluster
2. Missing host (error)
3. Missing node name (error)
4. Missing token (error)
5. Invalid token (401)
6. Expired token (401) | **DONE** | ### Version diff --git a/docs/install-multi-node/README.md b/docs/install-multi-node/README.md index 7075dd268..1417d3324 100644 --- a/docs/install-multi-node/README.md +++ b/docs/install-multi-node/README.md @@ -106,12 +106,24 @@ sudo spx admin init \ Save the admin credentials — they will not be shown again. +The init output displays the join command including the token: + +``` +📡 Formation server started on 10.0.0.1:4432 + Waiting for 2 more node(s) to join... + Token expires in 30m0s + + Other nodes should run: + spx admin join --host 10.0.0.1:4432 --token spx_join_a8Bf3x9Kz2mN --node --bind +``` + **Server 2 — Join** (while init is running): ```bash sudo spx admin join \ --node node2 --bind $SPINIFEX_NODE2 --cluster-bind $SPINIFEX_NODE2 \ - --host $SPINIFEX_NODE1:4432 --region $AWS_REGION --az $AWS_AZ + --host $SPINIFEX_NODE1:4432 --token \ + --region $AWS_REGION --az $AWS_AZ ``` **Server 3 — Join** (while init is running): @@ -119,9 +131,13 @@ sudo spx admin join \ ```bash sudo spx admin join \ --node node3 --bind $SPINIFEX_NODE3 --cluster-bind $SPINIFEX_NODE3 \ - --host $SPINIFEX_NODE1:4432 --region $AWS_REGION --az $AWS_AZ + --host $SPINIFEX_NODE1:4432 --token \ + --region $AWS_REGION --az $AWS_AZ ``` +> **Note:** The join token expires 30 minutes after init by default. For larger deployments +> with slower provisioning, use `--token-ttl 2h` on the init command. + ## Step 5. Start Services On **all servers**: diff --git a/scripts/create-multi-node.sh b/scripts/create-multi-node.sh index ab0a45b45..c1ae2ca3a 100755 --- a/scripts/create-multi-node.sh +++ b/scripts/create-multi-node.sh @@ -39,7 +39,26 @@ echo "Starting init node (127.0.0.1) and joining nodes (127.0.0.2, 127.0.0.3)... INIT_PID=$! # Wait for formation server to start -sleep 2 +echo "Waiting for formation server..." +for i in $(seq 1 30); do + if curl -sk "https://127.0.0.1:4432/formation/health" > /dev/null 2>&1; then + echo "Formation server ready" + break + fi + if [ "$i" -eq 30 ]; then + echo "ERROR: Formation server failed to start" + kill $INIT_PID 2>/dev/null || true + exit 1 + fi + sleep 1 +done + +# Read join token +JOIN_TOKEN=$(cat ~/node1/config/join-token) +if [ -z "$JOIN_TOKEN" ]; then + echo "ERROR: Join token file is empty" + exit 1 +fi # Join nodes concurrently ./bin/spx admin join \ @@ -47,6 +66,7 @@ sleep 2 --bind 127.0.0.2 \ --cluster-bind 127.0.0.2 \ --host 127.0.0.1:4432 \ + --token "$JOIN_TOKEN" \ --data-dir ~/node2/ \ --config-dir ~/node2/config/ \ --region ap-southeast-2 \ @@ -58,6 +78,7 @@ JOIN2_PID=$! --bind 127.0.0.3 \ --cluster-bind 127.0.0.3 \ --host 127.0.0.1:4432 \ + --token "$JOIN_TOKEN" \ --data-dir ~/node3/ \ --config-dir ~/node3/config/ \ --region ap-southeast-2 \ diff --git a/scripts/iac/configure-cluster.sh b/scripts/iac/configure-cluster.sh index 095e25451..48ee7d263 100755 --- a/scripts/iac/configure-cluster.sh +++ b/scripts/iac/configure-cluster.sh @@ -121,6 +121,14 @@ while ! curl -sk --connect-timeout 3 "https://$NODE1_IP:4432/formation/health" > done echo " Formation server ready" +# Read join token from init node +JOIN_TOKEN=$(remote "$NODE1_IP" "cat ~/spinifex/config/join-token") +if [ -z "$JOIN_TOKEN" ]; then + echo "Error: Failed to read join token from $NODE1_IP" + exit 1 +fi +echo " Join token loaded from node1" + # Nodes 2..N: join if [ "$NODE_COUNT" -gt 1 ]; then JOIN_PIDS=() @@ -134,6 +142,7 @@ if [ "$NODE_COUNT" -gt 1 ]; then --cluster-bind $ip \ --cluster-routes $NODE1_IP:4248 \ --host $NODE1_IP:4432 \ + --token $JOIN_TOKEN \ --data-dir ~/spinifex/ \ --config-dir ~/spinifex/config/ \ --region $REGION \ diff --git a/scripts/start-dev.sh b/scripts/start-dev.sh index 46a75b9d8..df84af1b5 100755 --- a/scripts/start-dev.sh +++ b/scripts/start-dev.sh @@ -329,7 +329,11 @@ if has_service "predastore"; then PREDASTORE_CMD="./bin/spx service predastore start" start_service "predastore" "$PREDASTORE_CMD" set_oom_score "predastore" "-500" - check_service "Predastore" "$SPINIFEX_PREDASTORE_HOST" "$SPINIFEX_PREDASTORE_PORT" + if is_multinode; then + echo " ⏭️ Skipping Predastore connectivity check (multi-node: needs quorum from peer nodes)" + else + check_service "Predastore" "$SPINIFEX_PREDASTORE_HOST" "$SPINIFEX_PREDASTORE_PORT" + fi else echo "2️⃣ Skipping Predastore (not a local service)" fi diff --git a/spinifex/formation/formation.go b/spinifex/formation/formation.go index 124820a14..e0a29acc8 100644 --- a/spinifex/formation/formation.go +++ b/spinifex/formation/formation.go @@ -2,7 +2,10 @@ package formation import ( "context" + crypto_rand "crypto/rand" + "crypto/subtle" "crypto/tls" + "encoding/base64" "encoding/json" "fmt" "log/slog" @@ -98,12 +101,25 @@ type FormationServer struct { caKey string masterKey string networkConfig *NetworkConfig + joinToken string + tokenExpiry time.Time done chan struct{} server *http.Server } +// GenerateJoinToken returns a token of the form "spx_join_<16 base64url chars>". +// Uses crypto/rand for 12 random bytes (96 bits entropy), base64 URL-safe encoded. +func GenerateJoinToken() (string, error) { + b := make([]byte, 12) + if _, err := crypto_rand.Read(b); err != nil { + return "", fmt.Errorf("generate join token: %w", err) + } + return "spx_join_" + base64.RawURLEncoding.EncodeToString(b), nil +} + // NewFormationServer creates a new formation server expecting the given number of nodes. -func NewFormationServer(expected int, creds *SharedCredentials, caCert, caKey string, networkConfig *NetworkConfig) *FormationServer { +func NewFormationServer(expected int, creds *SharedCredentials, caCert, caKey string, + networkConfig *NetworkConfig, joinToken string, tokenTTL time.Duration) *FormationServer { return &FormationServer{ expected: expected, nodes: make(map[string]NodeInfo), @@ -111,6 +127,8 @@ func NewFormationServer(expected int, creds *SharedCredentials, caCert, caKey st caCert: caCert, caKey: caKey, networkConfig: networkConfig, + joinToken: joinToken, + tokenExpiry: time.Now().Add(tokenTTL), done: make(chan struct{}), } } @@ -238,7 +256,44 @@ func (fs *FormationServer) Shutdown(ctx context.Context) error { return fs.server.Shutdown(ctx) } +// validateToken checks the Authorization header against the join token. +// Returns an error if the token is missing, invalid, or expired. +func (fs *FormationServer) validateToken(r *http.Request) error { + fs.mu.RLock() + defer fs.mu.RUnlock() + + if time.Now().After(fs.tokenExpiry) { + return fmt.Errorf("join token has expired") + } + + auth := r.Header.Get("Authorization") + if auth == "" { + return fmt.Errorf("missing Authorization header") + } + + const prefix = "Bearer " + if len(auth) < len(prefix) || auth[:len(prefix)] != prefix { + return fmt.Errorf("invalid Authorization header format") + } + token := auth[len(prefix):] + + if subtle.ConstantTimeCompare([]byte(token), []byte(fs.joinToken)) != 1 { + return fmt.Errorf("invalid join token") + } + + return nil +} + func (fs *FormationServer) handleJoin(w http.ResponseWriter, r *http.Request) { + if err := fs.validateToken(r); err != nil { + slog.Warn("Formation join rejected", "error", err, "remote_addr", r.RemoteAddr) + writeJSON(w, http.StatusUnauthorized, JoinResponse{ + Success: false, + Message: "unauthorized: " + err.Error(), + }) + return + } + r.Body = http.MaxBytesReader(w, r.Body, 1*1024*1024) // 1 MB limit var req JoinRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -270,6 +325,12 @@ func (fs *FormationServer) handleJoin(w http.ResponseWriter, r *http.Request) { } func (fs *FormationServer) handleStatus(w http.ResponseWriter, r *http.Request) { + if err := fs.validateToken(r); err != nil { + slog.Warn("Formation status rejected", "error", err, "remote_addr", r.RemoteAddr) + http.Error(w, "unauthorized: "+err.Error(), http.StatusUnauthorized) + return + } + fs.mu.RLock() defer fs.mu.RUnlock() diff --git a/spinifex/formation/formation_test.go b/spinifex/formation/formation_test.go index 5357f4107..480d1b9f9 100644 --- a/spinifex/formation/formation_test.go +++ b/spinifex/formation/formation_test.go @@ -12,6 +12,9 @@ import ( "github.com/stretchr/testify/require" ) +const testToken = "spx_join_dGVzdHRva2VuMTIz" +const testTokenTTL = 1 * time.Hour + func testCreds() *SharedCredentials { return &SharedCredentials{ AccessKey: "AKIATEST1234567890AB", @@ -33,9 +36,53 @@ func testNode(name, ip string) NodeInfo { } } +func testServer(t *testing.T, fs *FormationServer) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("POST /formation/join", fs.handleJoin) + mux.HandleFunc("GET /formation/status", fs.handleStatus) + mux.HandleFunc("GET /formation/health", fs.handleHealth) + return httptest.NewServer(mux) +} + +func authPost(t *testing.T, url, token string, body []byte) *http.Response { + t.Helper() + req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + return resp +} + +func authGet(t *testing.T, url, token string) *http.Response { + t.Helper() + req, err := http.NewRequest(http.MethodGet, url, nil) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer "+token) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + return resp +} + +func TestGenerateJoinToken(t *testing.T) { + t.Parallel() + token1, err := GenerateJoinToken() + require.NoError(t, err) + + assert.True(t, len(token1) > len("spx_join_")) + assert.Equal(t, "spx_join_", token1[:9]) + assert.Len(t, token1[9:], 16) + + token2, err := GenerateJoinToken() + require.NoError(t, err) + assert.NotEqual(t, token1, token2) +} + func TestNewFormationServer(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "ca-cert-pem", "ca-key-pem", nil) + fs := NewFormationServer(3, testCreds(), "ca-cert-pem", "ca-key-pem", nil, testToken, testTokenTTL) assert.Equal(t, 3, fs.expected) assert.Empty(t, fs.nodes) @@ -44,7 +91,7 @@ func TestNewFormationServer(t *testing.T) { func TestRegisterNode(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "", "", nil) + fs := NewFormationServer(3, testCreds(), "", "", nil, testToken, testTokenTTL) err := fs.RegisterNode(testNode("node1", "10.0.0.1")) require.NoError(t, err) @@ -56,7 +103,7 @@ func TestRegisterNode(t *testing.T) { func TestRegisterDuplicateName(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "", "", nil) + fs := NewFormationServer(3, testCreds(), "", "", nil, testToken, testTokenTTL) require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) err := fs.RegisterNode(testNode("node1", "10.0.0.2")) @@ -66,7 +113,7 @@ func TestRegisterDuplicateName(t *testing.T) { func TestRegisterDuplicateIP(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "", "", nil) + fs := NewFormationServer(3, testCreds(), "", "", nil, testToken, testTokenTTL) require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) err := fs.RegisterNode(testNode("node2", "10.0.0.1")) @@ -76,7 +123,7 @@ func TestRegisterDuplicateIP(t *testing.T) { func TestIsComplete(t *testing.T) { t.Parallel() - fs := NewFormationServer(2, testCreds(), "", "", nil) + fs := NewFormationServer(2, testCreds(), "", "", nil, testToken, testTokenTTL) assert.False(t, fs.IsComplete()) require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) @@ -87,7 +134,7 @@ func TestIsComplete(t *testing.T) { func TestWaitForCompletion(t *testing.T) { t.Parallel() - fs := NewFormationServer(2, testCreds(), "", "", nil) + fs := NewFormationServer(2, testCreds(), "", "", nil, testToken, testTokenTTL) go func() { time.Sleep(50 * time.Millisecond) @@ -102,7 +149,7 @@ func TestWaitForCompletion(t *testing.T) { func TestWaitForCompletionTimeout(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "", "", nil) + fs := NewFormationServer(3, testCreds(), "", "", nil, testToken, testTokenTTL) require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) @@ -111,18 +158,62 @@ func TestWaitForCompletionTimeout(t *testing.T) { assert.Contains(t, err.Error(), "timed out") } -func testServer(t *testing.T, fs *FormationServer) *httptest.Server { - t.Helper() - mux := http.NewServeMux() - mux.HandleFunc("POST /formation/join", fs.handleJoin) - mux.HandleFunc("GET /formation/status", fs.handleStatus) - mux.HandleFunc("GET /formation/health", fs.handleHealth) - return httptest.NewServer(mux) +func TestValidateToken_Valid(t *testing.T) { + t.Parallel() + fs := NewFormationServer(2, testCreds(), "", "", nil, testToken, testTokenTTL) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Header.Set("Authorization", "Bearer "+testToken) + assert.NoError(t, fs.validateToken(req)) } -func TestJoinEndpoint(t *testing.T) { +func TestValidateToken_Missing(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "", "", nil) + fs := NewFormationServer(2, testCreds(), "", "", nil, testToken, testTokenTTL) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + err := fs.validateToken(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "missing Authorization header") +} + +func TestValidateToken_Invalid(t *testing.T) { + t.Parallel() + fs := NewFormationServer(2, testCreds(), "", "", nil, testToken, testTokenTTL) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Header.Set("Authorization", "Bearer wrong-token") + err := fs.validateToken(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "invalid join token") +} + +func TestValidateToken_Expired(t *testing.T) { + t.Parallel() + fs := NewFormationServer(2, testCreds(), "", "", nil, testToken, 1*time.Millisecond) + time.Sleep(5 * time.Millisecond) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Header.Set("Authorization", "Bearer "+testToken) + err := fs.validateToken(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "expired") +} + +func TestValidateToken_BadFormat(t *testing.T) { + t.Parallel() + fs := NewFormationServer(2, testCreds(), "", "", nil, testToken, testTokenTTL) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Header.Set("Authorization", "Basic "+testToken) + err := fs.validateToken(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "invalid Authorization header format") +} + +func TestJoinEndpoint_Unauthorized(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "", nil, testToken, testTokenTTL) ts := testServer(t, fs) defer ts.Close() @@ -131,6 +222,19 @@ func TestJoinEndpoint(t *testing.T) { require.NoError(t, err) defer resp.Body.Close() + assert.Equal(t, http.StatusUnauthorized, resp.StatusCode) +} + +func TestJoinEndpoint_WithToken(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "", nil, testToken, testTokenTTL) + ts := testServer(t, fs) + defer ts.Close() + + body, _ := json.Marshal(JoinRequest{NodeInfo: testNode("node1", "10.0.0.1")}) + resp := authPost(t, ts.URL+"/formation/join", testToken, body) + defer resp.Body.Close() + assert.Equal(t, http.StatusOK, resp.StatusCode) var jr JoinResponse @@ -142,33 +246,43 @@ func TestJoinEndpoint(t *testing.T) { func TestJoinEndpointDuplicate(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "", "", nil) + fs := NewFormationServer(3, testCreds(), "", "", nil, testToken, testTokenTTL) ts := testServer(t, fs) defer ts.Close() body, _ := json.Marshal(JoinRequest{NodeInfo: testNode("node1", "10.0.0.1")}) - resp1, err := http.Post(ts.URL+"/formation/join", "application/json", bytes.NewReader(body)) - require.NoError(t, err) + resp1 := authPost(t, ts.URL+"/formation/join", testToken, body) resp1.Body.Close() assert.Equal(t, http.StatusOK, resp1.StatusCode) - resp2, err := http.Post(ts.URL+"/formation/join", "application/json", bytes.NewReader(body)) - require.NoError(t, err) + resp2 := authPost(t, ts.URL+"/formation/join", testToken, body) defer resp2.Body.Close() assert.Equal(t, http.StatusConflict, resp2.StatusCode) } +func TestStatusEndpoint_Unauthorized(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "ca-cert", "ca-key", nil, testToken, testTokenTTL) + ts := testServer(t, fs) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/formation/status") + require.NoError(t, err) + defer resp.Body.Close() + + assert.Equal(t, http.StatusUnauthorized, resp.StatusCode) +} + func TestStatusEndpointIncomplete(t *testing.T) { t.Parallel() - fs := NewFormationServer(3, testCreds(), "ca-cert", "ca-key", nil) + fs := NewFormationServer(3, testCreds(), "ca-cert", "ca-key", nil, testToken, testTokenTTL) ts := testServer(t, fs) defer ts.Close() require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) - resp, err := http.Get(ts.URL + "/formation/status") - require.NoError(t, err) + resp := authGet(t, ts.URL+"/formation/status", testToken) defer resp.Body.Close() var sr StatusResponse @@ -184,15 +298,14 @@ func TestStatusEndpointIncomplete(t *testing.T) { func TestStatusEndpointComplete(t *testing.T) { t.Parallel() creds := testCreds() - fs := NewFormationServer(2, creds, "ca-cert-data", "ca-key-data", nil) + fs := NewFormationServer(2, creds, "ca-cert-data", "ca-key-data", nil, testToken, testTokenTTL) ts := testServer(t, fs) defer ts.Close() require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) require.NoError(t, fs.RegisterNode(testNode("node2", "10.0.0.2"))) - resp, err := http.Get(ts.URL + "/formation/status") - require.NoError(t, err) + resp := authGet(t, ts.URL+"/formation/status", testToken) defer resp.Body.Close() var sr StatusResponse @@ -209,14 +322,13 @@ func TestStatusEndpointComplete(t *testing.T) { func TestStatusEndpointMasterKey(t *testing.T) { t.Parallel() - fs := NewFormationServer(2, testCreds(), "ca-cert", "ca-key", nil) + fs := NewFormationServer(2, testCreds(), "ca-cert", "ca-key", nil, testToken, testTokenTTL) fs.SetMasterKey("dGVzdC1tYXN0ZXIta2V5LWJhc2U2NC1lbmNvZGVk") ts := testServer(t, fs) defer ts.Close() // Before completion: master key should not be exposed - resp, err := http.Get(ts.URL + "/formation/status") - require.NoError(t, err) + resp := authGet(t, ts.URL+"/formation/status", testToken) var sr StatusResponse require.NoError(t, json.NewDecoder(resp.Body).Decode(&sr)) resp.Body.Close() @@ -227,17 +339,16 @@ func TestStatusEndpointMasterKey(t *testing.T) { require.NoError(t, fs.RegisterNode(testNode("node2", "10.0.0.2"))) // After completion: master key should be present - resp, err = http.Get(ts.URL + "/formation/status") - require.NoError(t, err) + resp = authGet(t, ts.URL+"/formation/status", testToken) defer resp.Body.Close() require.NoError(t, json.NewDecoder(resp.Body).Decode(&sr)) assert.True(t, sr.Complete) assert.Equal(t, "dGVzdC1tYXN0ZXIta2V5LWJhc2U2NC1lbmNvZGVk", sr.MasterKey) } -func TestHealthEndpoint(t *testing.T) { +func TestHealthEndpoint_NoAuth(t *testing.T) { t.Parallel() - fs := NewFormationServer(1, testCreds(), "", "", nil) + fs := NewFormationServer(1, testCreds(), "", "", nil, testToken, testTokenTTL) ts := testServer(t, fs) defer ts.Close() @@ -289,22 +400,21 @@ func TestBuildPredastoreNodes(t *testing.T) { assert.Equal(t, "10.0.0.3", pnodes[2].Host) } -func TestFullFormationFlow(t *testing.T) { +func TestFullFormationFlow_WithToken(t *testing.T) { t.Parallel() creds := testCreds() - fs := NewFormationServer(3, creds, "-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----", "-----BEGIN PRIVATE KEY-----\ntest\n-----END PRIVATE KEY-----", nil) + fs := NewFormationServer(3, creds, "-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----", "-----BEGIN PRIVATE KEY-----\ntest\n-----END PRIVATE KEY-----", nil, testToken, testTokenTTL) ts := testServer(t, fs) defer ts.Close() - // 3 nodes join sequentially + // 3 nodes join sequentially with auth for i, n := range []struct{ name, ip string }{ {"node1", "10.0.0.1"}, {"node2", "10.0.0.2"}, {"node3", "10.0.0.3"}, } { body, _ := json.Marshal(JoinRequest{NodeInfo: testNode(n.name, n.ip)}) - resp, err := http.Post(ts.URL+"/formation/join", "application/json", bytes.NewReader(body)) - require.NoError(t, err) + resp := authPost(t, ts.URL+"/formation/join", testToken, body) var jr JoinResponse require.NoError(t, json.NewDecoder(resp.Body).Decode(&jr)) @@ -314,9 +424,8 @@ func TestFullFormationFlow(t *testing.T) { assert.Equal(t, i+1, jr.Joined) } - // All nodes poll status — should be complete - resp, err := http.Get(ts.URL + "/formation/status") - require.NoError(t, err) + // All nodes poll status with auth — should be complete + resp := authGet(t, ts.URL+"/formation/status", testToken) defer resp.Body.Close() var sr StatusResponse diff --git a/tests/e2e/TEST_COVERAGE.md b/tests/e2e/TEST_COVERAGE.md index 9878be84a..506cfa365 100644 --- a/tests/e2e/TEST_COVERAGE.md +++ b/tests/e2e/TEST_COVERAGE.md @@ -306,11 +306,11 @@ Tests that EC2 resources are properly isolated between tenant accounts (Alpha, B - Simulated network IPs (no ramdisk — start-dev.sh uses disk-backed WAL/VB in CI) ### Phase 2: Cluster Initialization -- `spx admin init` (leader node1) +- `spx admin init` (leader node1) — generates join token, writes to `~/node1/config/join-token` - CA certificate trust -- Start node1 services -- `spx admin join` (node2, node3) -- Start node2 + node3 services +- Read join token from node1 config +- `spx admin join` with `--token` (node2, node3 — concurrent) +- Start all 3 node services concurrently ### Phase 3: Cluster Health Verification - Verify NATS cluster (3 nodes) @@ -562,7 +562,7 @@ Runs on a real 3-node libvirt cluster provisioned by OpenTofu (`scripts/tofu-clu 4. `git fetch` + checkout test branch on all nodes (all 3 repos) 5. `make build` on all nodes (parallel) 6. `setup-ovn.sh` — primary gets `--management`, secondaries connect to primary's OVN central -7. `spx admin init` on primary, `spx admin join` on secondaries (formation server handshake) +7. `spx admin init` on primary (generates join token), read token via SSH, `spx admin join --token` on secondaries 8. `start-dev.sh` on all nodes 9. Wait for `/health` (awsgw=ok) on all nodes 10. Install Spinifex CA certificate on all nodes diff --git a/tests/e2e/lib/multinode-helpers.sh b/tests/e2e/lib/multinode-helpers.sh index 98ac014c0..4613fbc6b 100755 --- a/tests/e2e/lib/multinode-helpers.sh +++ b/tests/e2e/lib/multinode-helpers.sh @@ -796,6 +796,16 @@ init_leader_node() { for i in $(seq 1 60); do if curl -sk "https://${NODE1_IP}:${CLUSTER_PORT}/formation/health" > /dev/null 2>&1; then echo " Formation server is ready (PID: $LEADER_INIT_PID)" + + # Read join token written by admin init + JOIN_TOKEN=$(cat "$HOME/node1/config/join-token") + if [ -z "$JOIN_TOKEN" ]; then + echo " ERROR: Join token file is empty" + return 1 + fi + echo " Join token loaded" + export JOIN_TOKEN + return 0 fi sleep 1 @@ -824,6 +834,7 @@ join_follower_node() { --cluster-bind "$node_ip" \ --cluster-routes "${NODE1_IP}:${NATS_CLUSTER_PORT}" \ --host "${NODE1_IP}:${CLUSTER_PORT}" \ + --token "$JOIN_TOKEN" \ --data-dir "$data_dir/" \ --config-dir "$data_dir/config/" \ --region ap-southeast-2 \