diff --git a/tests/integration/a2a_test.go b/tests/integration/a2a_test.go index 04963c90..cca8933d 100644 --- a/tests/integration/a2a_test.go +++ b/tests/integration/a2a_test.go @@ -19,7 +19,6 @@ import ( "iter" "net/http" "net/http/httptest" - "path/filepath" "strings" "sync/atomic" "testing" @@ -103,25 +102,23 @@ func TestA2ACUJ(t *testing.T) { defer agent.Close() t.Log("Starting Node A (provider, region=eu)...") - _ = startBackgroundNode(t, nodeBin, hubAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, hubAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", "--config", writeNodeConfig(t, homeA, map[string]string{"region": "eu"}, svcDecl{Type: "a2a", Name: "echo-agent", TargetURL: agent.URL}), ) t.Log("Starting Node B (consumer)...") - _ = startBackgroundNode(t, nodeBin, hubAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, hubAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", ) - apiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - apiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, apiAddrA) - waitForAPI(t, apiAddrB) + apiAddrA := nodeA.waitForAPI(t) + apiAddrB := nodeB.waitForAPI(t) - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr connectPeer(t, apiAddrB, addrA) waitForDHTPeers(t, apiAddrA) diff --git a/tests/integration/agent_ingress_test.go b/tests/integration/agent_ingress_test.go index 26422587..e7f408b5 100644 --- a/tests/integration/agent_ingress_test.go +++ b/tests/integration/agent_ingress_test.go @@ -104,13 +104,13 @@ func TestAgentIngressCUJ(t *testing.T) { t.Fatalf("writing node config: %v", err) } - _ = startBackgroundNode(t, nodeBin, hubAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, hubAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", ) // Node B hosts the agent, and is the one that advertises its service. - _ = startBackgroundNode(t, nodeBin, hubAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, hubAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", @@ -118,12 +118,10 @@ func TestAgentIngressCUJ(t *testing.T) { "--config", cfgPath, ) - apiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - apiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, apiAddrA) - waitForAPI(t, apiAddrB) + apiAddrA := nodeA.waitForAPI(t) + apiAddrB := nodeB.waitForAPI(t) - addrB := waitForPeerInfoInLog(t, filepath.Join(homeB, "node.log")) + addrB := nodeB.p2pAddr peerB := extractPeerID(addrB) connectPeer(t, apiAddrA, addrB) waitForDHTPeers(t, apiAddrB) diff --git a/tests/integration/agent_policy_test.go b/tests/integration/agent_policy_test.go index 087486c4..30911555 100644 --- a/tests/integration/agent_policy_test.go +++ b/tests/integration/agent_policy_test.go @@ -83,25 +83,23 @@ services: t.Cleanup(func() { _ = os.RemoveAll(sockDir) }) nodeSocket := filepath.Join(sockDir, "node.sock") - _ = startBackgroundNode(t, nodeBin, hubAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, hubAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", "--config", configA, ) - _ = startBackgroundNode(t, nodeBin, hubAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, hubAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", "--socket-path", nodeSocket, ) - apiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - apiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, apiAddrA) - waitForAPI(t, apiAddrB) + apiAddrA := nodeA.waitForAPI(t) + apiAddrB := nodeB.waitForAPI(t) - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr connectPeer(t, apiAddrB, addrA) waitForDHTPeers(t, apiAddrA) diff --git a/tests/integration/catalog_test.go b/tests/integration/catalog_test.go index de8e2110..7a656514 100644 --- a/tests/integration/catalog_test.go +++ b/tests/integration/catalog_test.go @@ -20,7 +20,6 @@ import ( "fmt" "net/http" "os" - "os/exec" "path/filepath" "sort" "strings" @@ -87,62 +86,6 @@ func writeNodeConfig(t *testing.T, dir string, labels map[string]string, service return p } -func startBackgroundNode(t *testing.T, nodeBin string, routerAddr string, homeDir string, args ...string) *exec.Cmd { - t.Helper() - env := append(os.Environ(), - "HOME="+homeDir, - "XDG_CONFIG_HOME="+filepath.Join(homeDir, ".config"), - "SAM_API_TOKEN=test-token", // per-test overrides use --api-token-path, which wins - ) - allArgs := append([]string{"run", "--control-plane", routerAddr, "--jwt", "test-jwt", "--bind-addr", "127.0.0.1:0", "--allow-loopback"}, args...) - cmd := exec.Command(nodeBin, allArgs...) - cmd.Env = env - - logFile, err := os.Create(filepath.Join(homeDir, "node.log")) - if err != nil { - t.Fatalf("failed to create log file: %v", err) - } - cmd.Stdout = logFile - cmd.Stderr = logFile - - if err := cmd.Start(); err != nil { - t.Fatalf("failed to start background node: %v", err) - } - - t.Cleanup(func() { - if err := cmd.Process.Kill(); err != nil { - t.Logf("warning: failed to kill background node: %v", err) - } - if err := logFile.Close(); err != nil { - t.Logf("warning: failed to close log file: %v", err) - } - }) - - return cmd -} - -func waitForMCPAddr(t *testing.T, logPath string) string { - t.Helper() - // Generous under CI load; polling returns as soon as the line appears. - deadline := time.Now().Add(10 * time.Second) - for time.Now().Before(deadline) { - data, _ := os.ReadFile(logPath) - lines := strings.Split(string(data), "\n") - for _, line := range lines { - if strings.Contains(line, "Starting MCP server on TCP address ") { - parts := strings.Split(line, "Starting MCP server on TCP address ") - if len(parts) > 1 { - return strings.TrimSpace(parts[1]) - } - } - } - time.Sleep(100 * time.Millisecond) - } - data, _ := os.ReadFile(logPath) - t.Fatalf("timeout waiting for MCP addr in log: %s\n--- log contents ---\n%s", logPath, string(data)) - return "" -} - func callMCP(t *testing.T, mcpAddr string, toolName string, params map[string]any) string { t.Helper() ctx := context.Background() @@ -201,36 +144,6 @@ func (a *authRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) return a.rt.RoundTrip(clone) } -func waitForPeerInfoInLog(t *testing.T, logPath string) string { - t.Helper() - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - data, _ := os.ReadFile(logPath) - lines := strings.Split(string(data), "\n") - var peerID string - var tcpAddr string - for _, line := range lines { - if strings.HasPrefix(line, "PeerID: ") { - peerID = strings.TrimPrefix(line, "PeerID: ") - } - if strings.Contains(line, "Listening on: ") { - parts := strings.Split(line, " ") - for _, p := range parts { - if strings.Contains(p, "/tcp/") { - tcpAddr = strings.Trim(p, "[]") - } - } - } - } - if peerID != "" && tcpAddr != "" { - return tcpAddr + "/p2p/" + peerID - } - time.Sleep(100 * time.Millisecond) - } - t.Fatalf("timeout waiting for peer info in log: %s", logPath) - return "" -} - func TestCatalogRoutingAndFailover(t *testing.T) { nodeBin := buildBinary(t, "./cmd/sam-node") _, routerAddr := startMockRouter(t) @@ -241,25 +154,25 @@ func TestCatalogRoutingAndFailover(t *testing.T) { // Start Node A (Client) t.Log("Starting Node A...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms") + nodeA := startBackgroundNode(t, nodeBin, routerAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms") t.Log("Node A started.") - // Wait for Node A to start and get its MCP address - mcpAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) + mcpAddrA := nodeA.waitForAPI(t) // Start Node B (Provider 1) t.Log("Starting Node B...") - cmdB := startBackgroundNode(t, nodeBin, routerAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms") + nodeB := startBackgroundNode(t, nodeBin, routerAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms") t.Log("Node B started.") // Start Node C (Provider 2) t.Log("Starting Node C...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeC, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms") + nodeC := startBackgroundNode(t, nodeBin, routerAddr, homeC, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms") t.Log("Node C started.") - // Wait for Node B and C to start and get their addresses - addrB := waitForPeerInfoInLog(t, filepath.Join(homeB, "node.log")) - addrC := waitForPeerInfoInLog(t, filepath.Join(homeC, "node.log")) + nodeB.waitForAPI(t) + nodeC.waitForAPI(t) + addrB := nodeB.p2pAddr + addrC := nodeC.p2pAddr // Force Node A to connect to Node B and Node C connectPeer(t, mcpAddrA, addrB) @@ -329,9 +242,7 @@ func TestCatalogRoutingAndFailover(t *testing.T) { t.Logf("First call response: %s", respData) // Now kill Node B and assert failover to Node C - if err := cmdB.Process.Kill(); err != nil { - t.Fatalf("failed to kill Node B: %v", err) - } + nodeB.kill() // Wait a bit for catalog update or failover to happen on next call time.Sleep(500 * time.Millisecond) diff --git a/tests/integration/cuj_test.go b/tests/integration/cuj_test.go deleted file mode 100644 index e27609b2..00000000 --- a/tests/integration/cuj_test.go +++ /dev/null @@ -1,56 +0,0 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package integration_test - -import ( - "context" - "os" - "path/filepath" - "strings" - "testing" - "time" -) - -func TestSamNodeRunWithManualTokenStarts(t *testing.T) { - nodeBin := buildBinary(t, "./cmd/sam-node") - _, routerAddr := startMockRouter(t) - tmpHome := t.TempDir() - env := append(os.Environ(), - "HOME="+tmpHome, - "XDG_CONFIG_HOME="+filepath.Join(tmpHome, ".config"), - ) - - stdout, stderr, err := runCommand( - t, - repoRoot(t), - 3*time.Second, - env, - "", - nodeBin, - "run", "--control-plane", routerAddr, - "--jwt", "test-jwt", - "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", - "--api-token-path", tokenPath(t, "dummy-token"), - ) - if err != context.DeadlineExceeded { - t.Fatalf("expected run command to keep running until timeout, got: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) - } - out := stdout + stderr - if !strings.Contains(out, "SAM Node Online") { - t.Fatalf("node did not reach online state:\n%s", out) - } -} diff --git a/tests/integration/datapath_test.go b/tests/integration/datapath_test.go index ab29075b..71b51400 100644 --- a/tests/integration/datapath_test.go +++ b/tests/integration/datapath_test.go @@ -48,34 +48,27 @@ func TestIntegrationStdioDatapath(t *testing.T) { // Start Node A t.Log("Starting Node A...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, routerAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), "--config", cfgA, ) // Start Node B t.Log("Starting Node B...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, routerAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), ) - // Resolve actual addresses from logs - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) + nodeA.waitForAPI(t) + actualApiAddrB := nodeB.waitForAPI(t) - // Wait for nodes to start sidecar API - waitForAPI(t, actualApiAddrA) - waitForAPI(t, actualApiAddrB) - - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr peerIDA := getPeerIDFromAddr(addrA) // Connect Node B to Node A @@ -177,36 +170,29 @@ func TestIntegrationHTTPDatapath(t *testing.T) { // Start Node A t.Log("Starting Node A...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, routerAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), "--config", cfgA, ) // Start Node B t.Log("Starting Node B...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, routerAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), ) - // Resolve actual addresses from logs - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - - // Wait for nodes to start sidecar API - waitForAPI(t, actualApiAddrA) - waitForAPI(t, actualApiAddrB) + nodeA.waitForAPI(t) + actualApiAddrB := nodeB.waitForAPI(t) - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr peerIDA := getPeerIDFromAddr(addrA) - addrB := waitForPeerInfoInLog(t, filepath.Join(homeB, "node.log")) + addrB := nodeB.p2pAddr peerIDB := getPeerIDFromAddr(addrB) // Connect Node B to Node A diff --git a/tests/integration/debug_endpoints_test.go b/tests/integration/debug_endpoints_test.go index bf80a9a0..40d7587c 100644 --- a/tests/integration/debug_endpoints_test.go +++ b/tests/integration/debug_endpoints_test.go @@ -113,34 +113,26 @@ roles: [] // Start Node A t.Log("Starting Node A...") - _ = startBackgroundNode(t, nodeBin, controlPlaneURL, homeA, + nodeA := startBackgroundNode(t, nodeBin, controlPlaneURL, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), "--jwt", nodeJWT, ) // Start Node B t.Log("Starting Node B...") - _ = startBackgroundNode(t, nodeBin, controlPlaneURL, homeB, + nodeB := startBackgroundNode(t, nodeBin, controlPlaneURL, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), "--jwt", nodeJWT, ) - // Resolve actual local API address from log - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) + actualApiAddrA := nodeA.waitForAPI(t) + actualApiAddrB := nodeB.waitForAPI(t) - // Wait for nodes to start sidecar API - waitForAPI(t, actualApiAddrA) - waitForAPI(t, actualApiAddrB) - - // Resolve Peer addresses - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr // Connect Node B to Node A directly; exercises POST /debug/connect-peer connectPeer(t, actualApiAddrB, addrA) diff --git a/tests/integration/failover_test.go b/tests/integration/failover_test.go index 6b05ea6d..a91c4460 100644 --- a/tests/integration/failover_test.go +++ b/tests/integration/failover_test.go @@ -25,7 +25,6 @@ import ( "os" "os/exec" "path/filepath" - "strings" "sync" "testing" "time" @@ -150,6 +149,7 @@ roles: [] "--keys-path", filepath.Join(tmpDir, "router_keysA.db"), "--allow-loopback", "--oidc-token", routerJWT, + "--lease-renew-interval", "1s", ) if err := cmdRouterA.Start(); err != nil { t.Fatalf("failed to start Router A: %v", err) @@ -182,6 +182,7 @@ roles: [] "--keys-path", filepath.Join(tmpDir, "router_keysB.db"), "--allow-loopback", "--oidc-token", routerJWT, + "--lease-renew-interval", "1s", ) if err := cmdRouterB.Start(); err != nil { t.Fatalf("failed to start Router B: %v", err) @@ -202,10 +203,8 @@ roles: [] "HOME="+nodeHome, "XDG_CONFIG_HOME="+filepath.Join(nodeHome, ".config"), ) - cmdNode := exec.Command(nodeBin, "run", "--control-plane", lb.URL, - "--listen", "/ip4/127.0.0.1/tcp/0", + samNode := launchNode(t, nodeBin, env, nodeHome, "run", "--control-plane", lb.URL, "--jwt-path", jwtPath, - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, "dummy-token"), "--allow-loopback", "--monitor-bootstrap", "1s", @@ -214,37 +213,11 @@ roles: [] "--autorelay-backoff", "1s", "--autorelay-boot-delay", "0s", ) - cmdNode.Dir = repoRoot(t) - cmdNode.Env = env - var stdoutNode, stderrNode safeBuffer - cmdNode.Stdout = &stdoutNode - cmdNode.Stderr = &stderrNode + samNode.waitForAPI(t) + nodePeerID := samNode.peerID.String() - if err := cmdNode.Start(); err != nil { - t.Fatal(err) - } - defer func() { _ = cmdNode.Process.Kill(); _ = cmdNode.Wait() }() - - // Wait for Node to get a reservation on Router A - var nodePeerID string - var out string - for i := 0; i < 100; i++ { - out = stdoutNode.String() + stderrNode.String() - if strings.Contains(out, "PeerID:") { - idx := strings.Index(out, "PeerID:") - parts := strings.Split(strings.TrimSpace(out[idx+len("PeerID:"):]), "\n") - if len(parts) > 0 { - nodePeerID = strings.TrimSpace(parts[0]) - } - } - if nodePeerID != "" && strings.Contains(out, "Yielding static relays to AutoRelay") { - break - } - time.Sleep(100 * time.Millisecond) - } - if nodePeerID == "" { - t.Fatalf("Node failed to get PeerID in time.\nOutput:\n%s", out) - } + // The node is on the mesh once Router A reports it connected to CP A. + waitForPeerOnRouter(t, httpPortCP_A, testAdminToken, nodePeerID, 15*time.Second) t.Logf("Node started. Node PeerID: %s", nodePeerID) // Now FAILOVER: Switch LB to CP B and KILL CP A and Router A @@ -255,18 +228,9 @@ roles: [] _ = cmdCP_A.Process.Kill() _ = cmdRouterA.Process.Kill() - // Wait for Node's AutoRelay to get updated - for i := 0; i < 150; i++ { - out = stdoutNode.String() + stderrNode.String() - if strings.Contains(out, "Successfully reconnected to router via HTTP fallback") { - break - } - time.Sleep(100 * time.Millisecond) - } - if !strings.Contains(out, "Successfully reconnected to router via HTTP fallback") { - t.Fatalf("Node failed to detect failover and reconnect.\nOutput:\n%s", out) - } - t.Log("Node successfully reconnected to router B!") + // The node has failed over once Router B reports it connected to CP B. + waitForPeerOnRouter(t, httpPortCP_B, testAdminToken, nodePeerID, 20*time.Second) + t.Log("Node reconnected to router B") // Final verification: Ensure we can actually reach Node B via the Router B relay relayAddrStr := fmt.Sprintf("/ip4/127.0.0.1/tcp/%d/p2p/%s/p2p-circuit/p2p/%s", routerPortB, peerIDB, nodePeerID) @@ -311,7 +275,7 @@ roles: [] } if connectErr != nil { - t.Fatalf("Failed to connect to Node B via router B relay: %v\nOutput: %s", connectErr, stdoutNode.String()+stderrNode.String()) + t.Fatalf("Failed to connect to Node B via router B relay: %v\n--- node.log ---\n%s", connectErr, samNode.log()) } t.Log("Successfully connected to Node B via router B relay circuit!") diff --git a/tests/integration/fallback_test.go b/tests/integration/fallback_test.go index f8ffb9cf..704e49c7 100644 --- a/tests/integration/fallback_test.go +++ b/tests/integration/fallback_test.go @@ -16,31 +16,15 @@ package integration_test import ( "bytes" - "crypto/ed25519" - "crypto/rand" - "encoding/json" "fmt" - "io" - - "os" - "os/exec" "path/filepath" - "strings" + "slices" "sync" - "sync/atomic" "testing" "time" - "net/http" - "net/http/httptest" - "github.com/biscuit-auth/biscuit-go/v2/parser" - "github.com/google/sam/api" - "github.com/libp2p/go-libp2p" - "github.com/libp2p/go-libp2p/core/host" - "github.com/libp2p/go-libp2p/core/network" - "github.com/libp2p/go-msgio" - "google.golang.org/protobuf/proto" + "github.com/google/sam/internal/node" ) // init forces the biscuit-go parser to build its underlying participle @@ -68,230 +52,92 @@ func (s *safeBuffer) String() string { } // TestSelfHealingHTTPFallback: the routers a node stored can be gone by its -// next start (a redeploy moves every router's address); the node must ask the -// control plane for the current ones and reach one. The assertions are the two -// ends of that path: the control plane saw the request, and the router that -// only exists since the address change completed an auth handshake with the -// node. Neither depends on what the node logs. +// next start (a redeploy moves every router's address); the node must ask +// the control plane for the current ones, reach one, and remember it. func TestSelfHealingHTTPFallback(t *testing.T) { nodeBin := buildBinary(t, "./cmd/sam-node") - - var mu sync.Mutex - var currentP2PAddr string - var infoRequests atomic.Int32 - - pub, priv, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - t.Fatalf("Failed to generate control plane key: %v", err) + tmpDir := t.TempDir() + oidcURL, mintToken := startCustomMockOIDC(t) + + // Leases are short so the control plane forgets the old router soon + // after it is gone, the way a redeploy replaces the router set. + meshDir := filepath.Join(tmpDir, "mesh") + cpPort, _ := startControlPlane(t, meshDir, oidcURL, meshPolicyFile(t, meshDir), "--lease-duration", "2s") + cpURL := fmt.Sprintf("http://127.0.0.1:%d", cpPort) + oldRouter, stopOldRouter := startRouter(t, meshDir, cpPort, mintToken, "old-router") + + home := filepath.Join(tmpDir, "home") + env, dataDir := nodeHome(home) + token := tokenPath(t, "node-token") + + // Enroll while the old router is the only one, so it is the one stored. + first := launchNode(t, nodeBin, env, home, "run", "--control-plane", cpURL, + "--jwt", mintToken(map[string]interface{}{"sub": jwtUser}), "--allow-loopback", "--api-token-path", token) + first.waitForAPI(t) + waitForPeerOnRouter(t, cpPort, testAdminToken, first.peerID.String(), 10*time.Second) + first.kill() + if addrs := storedRouters(t, dataDir); !slices.Contains(addrs, oldRouter) { + t.Fatalf("stored routers %v do not include the router the node enrolled through, %s", addrs, oldRouter) } - // createNewHost is a router as the node sees it: the auth handshake, and a - // channel closed once a node has completed it. - createNewHost := func() (host.Host, <-chan struct{}) { - newH, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) - if err != nil { - t.Fatal(err) - } - authenticated := make(chan struct{}) - var once sync.Once - newH.SetStreamHandler(api.AuthProtocolID, func(s network.Stream) { - defer func() { _ = s.Close() }() - reader := msgio.NewVarintReaderSize(s, 1024*64) - msg, err := reader.ReadMsg() - if err != nil { - return - } - defer reader.ReleaseMsg(msg) - - writer := msgio.NewVarintWriter(s) - resp := &api.AuthResponse{ - Success: true, - Biscuit: createMockBiscuitToken(t, newH.ID().String(), priv, api.RoleRouter, nil), - } - respBytes, err := proto.Marshal(resp) - if err != nil { - t.Errorf("marshal auth response: %v", err) - return - } - if err := writer.WriteMsg(respBytes); err != nil { - t.Errorf("write auth response: %v", err) - return - } - once.Do(func() { close(authenticated) }) - }) - return newH, authenticated + // The redeploy: the stored router is gone, a new one is up, and only the + // control plane knows about it. + stopOldRouter() + newRouter, _ := startRouter(t, meshDir, cpPort, mintToken, "new-router") + waitForRouterSet(t, cpPort, []string{newRouter}, 10*time.Second) + + // The node comes back on its stored config and no --control-plane: what + // it stored is a dead address, so reaching the new router at all means + // it asked the control plane. /healthz answers only once Start succeeded, + // and Start fails unless a router authenticated the node. + again := launchNode(t, nodeBin, env, home, "run", "--allow-loopback", "--api-token-path", token) + again.waitForAPI(t) + lease := waitForPeerOnRouter(t, cpPort, testAdminToken, again.peerID.String(), 10*time.Second) + if !slices.Contains(lease.Addresses, newRouter) { + t.Fatalf("the node is connected to %v, not to the new router %s", lease.Addresses, newRouter) } - h, _ := createNewHost() - defer func() { _ = h.Close() }() - - mu.Lock() - currentP2PAddr = h.Addrs()[0].String() + "/p2p/" + h.ID().String() - mu.Unlock() - - mux := http.NewServeMux() - - // Mock OIDC server for device flow - mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "issuer": "http://" + r.Host, - "token_endpoint": "http://" + r.Host + "/token", - "authorization_endpoint": "http://" + r.Host + "/auth", - }) - }) - mux.HandleFunc("/device/code", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "device_code": "dev_code_123", - "user_code": "ABCD-1234", - "verification_uri": "http://example.com/verify", - "verification_uri_complete": "http://example.com/verify?code=ABCD-1234", - "expires_in": 60, - "interval": 1, - }) - }) - mux.HandleFunc("/token", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]string{ - "access_token": "test-jwt-token", - "id_token": "test-jwt-token", - }) - }) - - mux.HandleFunc("/register", func(w http.ResponseWriter, r *http.Request) { - mu.Lock() - addr := currentP2PAddr - mu.Unlock() + // Healed for good: the next start does not need the control plane to + // find a router either. + again.kill() + addrs := storedRouters(t, dataDir) + if !slices.Contains(addrs, newRouter) || slices.Contains(addrs, oldRouter) { + t.Fatalf("stored routers %v: want %s and not %s", addrs, newRouter, oldRouter) + } +} - body, err := io.ReadAll(r.Body) - if err != nil { - http.Error(w, "Failed to read body", http.StatusBadRequest) - return - } - var enrollReq api.EnrollRequest - if err := proto.Unmarshal(body, &enrollReq); err != nil { - http.Error(w, "Invalid request", http.StatusBadRequest) - return - } +// storedRouters is the router set the node would dial on its next start. +func storedRouters(t *testing.T, dataDir string) []string { + t.Helper() + store, err := node.NewStore(dataDir) + if err != nil { + t.Fatalf("open node store: %v", err) + } + defer func() { _ = store.Close() }() + _, addrs, err := store.LoadMeshConfig() + if err != nil { + t.Fatalf("load mesh config: %v", err) + } + return addrs +} - resp := &api.EnrollResponse{ - BiscuitToken: createMockBiscuitToken(t, enrollReq.PeerId, priv, api.RoleNode, nil), - ControlPlanePublicKey: pub, - RouterAddresses: []string{addr}, - } - data, err := proto.Marshal(resp) - if err != nil { - t.Errorf("marshal /register: %v", err) - http.Error(w, err.Error(), http.StatusInternalServerError) - return +// waitForRouterSet waits until the control plane's active routers listen on +// exactly want. +func waitForRouterSet(t *testing.T, cpPort int, want []string, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for { + var got []string + for _, lease := range fetchAdminStatus(t, cpPort, testAdminToken).ActiveRouters { + got = append(got, lease.Addresses...) } - w.Header().Set("Content-Type", "application/x-protobuf") - if _, err := w.Write(data); err != nil { - t.Errorf("write /register: %v", err) - } - }) - mux.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) { - infoRequests.Add(1) - mu.Lock() - addr := currentP2PAddr - mu.Unlock() - resp := &api.ControlPlaneInfoResponse{ - OidcIssuer: "http://" + r.Host, - ClientId: "sam-mesh-audience", - Audience: "sam-mesh-audience", - RouterAddresses: []string{addr}, - } - data, err := proto.Marshal(resp) - if err != nil { - t.Errorf("marshal /info: %v", err) - http.Error(w, err.Error(), http.StatusInternalServerError) + slices.Sort(got) + if slices.Equal(got, want) { return } - w.Header().Set("Content-Type", "application/x-protobuf") - if _, err := w.Write(data); err != nil { - t.Errorf("write /info: %v", err) + if time.Now().After(deadline) { + t.Fatalf("control plane :%d routers = %v, want %v", cpPort, got, want) } - }) - - httpServer := httptest.NewServer(mux) - defer httpServer.Close() - - tmpHome := t.TempDir() - env := append(os.Environ(), - "HOME="+tmpHome, - "XDG_CONFIG_HOME="+filepath.Join(tmpHome, ".config"), - "BROWSER=echo", - ) - - // Step 1: Enroll via Join (using mock OIDC) - joinStdout, joinStderr, err := runCommandWithCallback( - t, - repoRoot(t), - 5*time.Second, - env, - "", - nodeBin, - "join", - httpServer.URL, - ) - if err != nil { - t.Fatalf("Join failed: %v\nstdout:\n%s\nstderr:\n%s", err, joinStdout, joinStderr) - } - - out := joinStdout + joinStderr - if !strings.Contains(out, "Successfully joined the Sovereign Agent Mesh!") { - t.Fatalf("Join did not succeed:\n%s", out) - } - - // Step 2: Simulate router changing its P2P port (HTTP URL stays the same). - // The stored address now points at nothing; only /info knows the new one. - _ = h.Close() - newRouter, authenticated := createNewHost() - defer func() { _ = newRouter.Close() }() - - mu.Lock() - currentP2PAddr = newRouter.Addrs()[0].String() + "/p2p/" + newRouter.ID().String() - mu.Unlock() - infoRequests.Store(0) - - // Step 3: Start sam-node run, with everything it needs to stay up: a - // node that reaches the router and then exits has not healed. - runCmd := exec.Command(nodeBin, "run", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", getFreePort(t)), - "--api-token-path", tokenPath(t, "fallback-test-token"), - ) - runCmd.Env = env - var output safeBuffer - runCmd.Stdout = &output - runCmd.Stderr = &output - - if err := runCmd.Start(); err != nil { - t.Fatal(err) - } - exited := make(chan error, 1) - go func() { exited <- runCmd.Wait() }() - defer func() { - _ = runCmd.Process.Kill() - <-exited - }() - - select { - case <-authenticated: - case err := <-exited: - t.Fatalf("sam-node run exited (%v) before authenticating with the moved router.\nOutput:\n%s", err, output.String()) - case <-time.After(5 * time.Second): - t.Fatalf("sam-node run never authenticated with the moved router.\nOutput:\n%s", output.String()) - } - if infoRequests.Load() == 0 { - t.Fatal("the node reached the moved router without asking the control plane for its address") - } - - // Reaching the router is not the end: the node must stay up on it. - select { - case err := <-exited: - t.Fatalf("sam-node run exited (%v) right after authenticating.\nOutput:\n%s", err, output.String()) - case <-time.After(500 * time.Millisecond): + time.Sleep(200 * time.Millisecond) } } diff --git a/tests/integration/identity_evidence_test.go b/tests/integration/identity_evidence_test.go index 1b58b187..af0d22db 100644 --- a/tests/integration/identity_evidence_test.go +++ b/tests/integration/identity_evidence_test.go @@ -56,24 +56,22 @@ func TestIdentityEvidenceOperatorFlow(t *testing.T) { t.Cleanup(func() { _ = os.RemoveAll(socketDir) }) ownerSocket := filepath.Join(socketDir, "owner.sock") - _ = startBackgroundNode(t, nodeBin, controlPlaneURL, ownerHome, + owner := startBackgroundNode(t, nodeBin, controlPlaneURL, ownerHome, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", "--socket-path", ownerSocket, ) - _ = startBackgroundNode(t, nodeBin, controlPlaneURL, providerHome, + provider := startBackgroundNode(t, nodeBin, controlPlaneURL, providerHome, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", ) - ownerAPI := waitForMCPAddr(t, filepath.Join(ownerHome, "node.log")) - providerAPI := waitForMCPAddr(t, filepath.Join(providerHome, "node.log")) - waitForAPI(t, ownerAPI) - waitForAPI(t, providerAPI) + ownerAPI := owner.waitForAPI(t) + provider.waitForAPI(t) - providerAddr := waitForPeerInfoInLog(t, filepath.Join(providerHome, "node.log")) + providerAddr := provider.p2pAddr providerPeerID := getPeerIDFromAddr(providerAddr) if providerPeerID == "" { t.Fatalf("provider address %q has no PeerID", providerAddr) diff --git a/tests/integration/local_policy_test.go b/tests/integration/local_policy_test.go index 41fb680f..21cdd1c6 100644 --- a/tests/integration/local_policy_test.go +++ b/tests/integration/local_policy_test.go @@ -17,7 +17,6 @@ package integration_test import ( "fmt" "os" - "os/exec" "path/filepath" "strings" "testing" @@ -69,40 +68,21 @@ attenuation: homeB := filepath.Join(tmpDir, "nodeB") apiTokenB := "tokenB" - apiPortB := getFreePort(t) - cmdB := exec.Command(nodeBin, "run", + nodeB := launchNode(t, nodeBin, os.Environ(), homeB, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", httpPortCP), "--data-dir", homeB, - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", apiPortB), "--api-token-path", tokenPath(t, apiTokenB), "--jwt", mintToken(map[string]interface{}{ "sub": "nodeB-user", "roles": []string{api.RoleNode}, }), - "--listen", "/ip4/127.0.0.1/tcp/0", "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--allow-loopback", "--config", nodeBPolicyFile, ) - if err := os.MkdirAll(homeB, 0755); err != nil { - t.Fatal(err) - } - logFileB, err := os.Create(filepath.Join(homeB, "node.log")) - if err != nil { - t.Fatal(err) - } - defer func() { _ = logFileB.Close() }() - cmdB.Stdout = logFileB - cmdB.Stderr = logFileB - if err := cmdB.Start(); err != nil { - t.Fatalf("Failed to start Node B: %v", err) - } - defer func() { _ = cmdB.Process.Kill(); _ = cmdB.Wait() }() - - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, actualApiAddrB) - addrB := waitForPeerInfoInLog(t, filepath.Join(homeB, "node.log")) + nodeB.waitForAPI(t) + addrB := nodeB.p2pAddr parts := strings.Split(addrB, "/p2p/") if len(parts) != 2 { @@ -113,38 +93,19 @@ attenuation: // Node A Config (unprivileged user) homeA := filepath.Join(tmpDir, "nodeA") apiTokenA := "tokenA" - apiPortA := getFreePort(t) - cmdA := exec.Command(nodeBin, "run", + nodeA := launchNode(t, nodeBin, os.Environ(), homeA, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", httpPortCP), "--data-dir", homeA, - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", apiPortA), "--api-token-path", tokenPath(t, apiTokenA), "--jwt", mintToken(map[string]interface{}{ "sub": "unprivileged-user", "roles": []string{api.RoleNode}, }), - "--listen", "/ip4/127.0.0.1/tcp/0", "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--allow-loopback", ) - if err := os.MkdirAll(homeA, 0755); err != nil { - t.Fatal(err) - } - logFileA, err := os.Create(filepath.Join(homeA, "node.log")) - if err != nil { - t.Fatal(err) - } - defer func() { _ = logFileA.Close() }() - cmdA.Stdout = logFileA - cmdA.Stderr = logFileA - if err := cmdA.Start(); err != nil { - t.Fatalf("Failed to start Node A: %v", err) - } - defer func() { _ = cmdA.Process.Kill(); _ = cmdA.Wait() }() - - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - waitForAPI(t, actualApiAddrA) + actualApiAddrA := nodeA.waitForAPI(t) // Make request from Node A to Node B // Even though Node A has no control plane permissions, Node B's local policy "allow if true;" should permit it. @@ -215,40 +176,21 @@ attenuation: homeB := filepath.Join(tmpDir, "nodeB") apiTokenB := "tokenB" - apiPortB := getFreePort(t) - cmdB := exec.Command(nodeBin, "run", + nodeB := launchNode(t, nodeBin, os.Environ(), homeB, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", httpPortCP), "--data-dir", homeB, - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", apiPortB), "--api-token-path", tokenPath(t, apiTokenB), "--jwt", mintToken(map[string]interface{}{ "sub": "nodeB-user", "roles": []string{api.RoleNode}, }), - "--listen", "/ip4/127.0.0.1/tcp/0", "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--allow-loopback", "--config", nodeBPolicyFile, ) - if err := os.MkdirAll(homeB, 0755); err != nil { - t.Fatal(err) - } - logFileB, err := os.Create(filepath.Join(homeB, "node.log")) - if err != nil { - t.Fatal(err) - } - defer func() { _ = logFileB.Close() }() - cmdB.Stdout = logFileB - cmdB.Stderr = logFileB - if err := cmdB.Start(); err != nil { - t.Fatalf("Failed to start Node B: %v", err) - } - defer func() { _ = cmdB.Process.Kill(); _ = cmdB.Wait() }() - - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, actualApiAddrB) - addrB := waitForPeerInfoInLog(t, filepath.Join(homeB, "node.log")) + nodeB.waitForAPI(t) + addrB := nodeB.p2pAddr parts := strings.Split(addrB, "/p2p/") if len(parts) != 2 { @@ -259,38 +201,19 @@ attenuation: // Node A Config (Client) homeA := filepath.Join(tmpDir, "nodeA") apiTokenA := "tokenA" - apiPortA := getFreePort(t) - cmdA := exec.Command(nodeBin, "run", + nodeA := launchNode(t, nodeBin, os.Environ(), homeA, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", httpPortCP), "--data-dir", homeA, - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", apiPortA), "--api-token-path", tokenPath(t, apiTokenA), "--jwt", mintToken(map[string]interface{}{ "sub": "client-user", "roles": []string{api.RoleNode}, }), - "--listen", "/ip4/127.0.0.1/tcp/0", "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--allow-loopback", ) - if err := os.MkdirAll(homeA, 0755); err != nil { - t.Fatal(err) - } - logFileA, err := os.Create(filepath.Join(homeA, "node.log")) - if err != nil { - t.Fatal(err) - } - defer func() { _ = logFileA.Close() }() - cmdA.Stdout = logFileA - cmdA.Stderr = logFileA - if err := cmdA.Start(); err != nil { - t.Fatalf("Failed to start Node A: %v", err) - } - defer func() { _ = cmdA.Process.Kill(); _ = cmdA.Wait() }() - - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - waitForAPI(t, actualApiAddrA) + actualApiAddrA := nodeA.waitForAPI(t) // Node A attempts to call Node B. // Node A's token allows calling "*" but restricts the target to "group:admin-only". diff --git a/tests/integration/login_test.go b/tests/integration/login_test.go index de1da1d4..28b7c5b1 100644 --- a/tests/integration/login_test.go +++ b/tests/integration/login_test.go @@ -15,109 +15,142 @@ package integration_test import ( - "context" + "bytes" "encoding/json" - "net/http" - "net/http/httptest" + "fmt" "os" "path/filepath" "strings" "testing" "time" + "github.com/google/sam/api" "github.com/google/sam/internal/node" ) -func TestSamNodeJoin(t *testing.T) { - nodeBin := buildBinary(t, "./cmd/sam-node") - tmpHome := t.TempDir() - env := append(os.Environ(), - "HOME="+tmpHome, - "XDG_CONFIG_HOME="+filepath.Join(tmpHome, ".config"), - "BROWSER=echo", - ) +// These tests cover how a node comes to hold, keep and lose its mesh +// identity. They watch the two places that identity lives: the control +// plane's enrollment records and the node's own store. What the node prints +// along the way is not asserted on. - // Mock OIDC server for device flow - mux := http.NewServeMux() - mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(map[string]interface{}{ - "issuer": "http://" + r.Host, - "token_endpoint": "http://" + r.Host + "/token", - "authorization_endpoint": "http://" + r.Host + "/auth", - }); err != nil { - t.Errorf("Failed to encode response: %v", err) - } - }) - mux.HandleFunc("/device/code", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(map[string]interface{}{ - "device_code": "dev_code_123", - "user_code": "ABCD-1234", - "verification_uri": "http://example.com/verify", - "verification_uri_complete": "http://example.com/verify?code=ABCD-1234", - "expires_in": 60, - "interval": 1, - }); err != nil { - t.Errorf("Failed to encode response: %v", err) - } - }) - mux.HandleFunc("/token", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(map[string]string{ - "access_token": "test-jwt-token", - "id_token": "test-jwt-token", - }); err != nil { - t.Errorf("Failed to encode response: %v", err) - } - }) - oidcServer := httptest.NewServer(mux) - defer oidcServer.Close() - - // Start mock libp2p router that knows about our mock OIDC server - _, routerAddr := startMockRouterWithOIDC(t, oidcServer.URL) - - stdout, stderr, err := runCommandWithCallback( - t, - repoRoot(t), - 5*time.Second, - env, - "", // No stdin needed - nodeBin, - "join", - routerAddr, - ) +const testAdminToken = "test-admin-token" + +// jwtUser is the subject of tokens handed to a node directly with --jwt. +const jwtUser = "mock-user" + +// meshPolicyFile writes the policy these tests run under: the mock +// provider's browser user and jwtUser are seated as nodes. +func meshPolicyFile(t *testing.T, dir string) string { + t.Helper() + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + policy := fmt.Sprintf("bindings:\n - role: %s\n members: [%q, %q]\nroles: []\n", + api.RoleNode, "user:"+mockOIDCUser, "user:"+jwtUser) + policyFile := filepath.Join(dir, "policies.yaml") + if err := os.WriteFile(policyFile, []byte(policy), 0o644); err != nil { + t.Fatal(err) + } + return policyFile +} + +// startMesh starts a control plane and a router under dir, with the policy +// from meshPolicyFile. It returns the control plane's port and URL. +func startMesh(t *testing.T, dir, oidcURL string, mintToken func(map[string]interface{}) string) (int, string) { + t.Helper() + cpPort, cleanup := startControlPlaneAndRouter(t, dir, oidcURL, mintToken, meshPolicyFile(t, dir)) + t.Cleanup(cleanup) + return cpPort, fmt.Sprintf("http://127.0.0.1:%d", cpPort) +} + +// nodeHome is the environment of a node whose files live under home, and +// the store that environment resolves to. +func nodeHome(home string) ([]string, string) { + env := append(os.Environ(), "HOME="+home, "XDG_CONFIG_HOME="+filepath.Join(home, ".config")) + return env, nodeDataDir(env, nil) +} + +// storedMembership is what the node keeps of its enrollment: the biscuit +// and the control plane that issued it, both empty when there is none. The +// store is the node's while it runs, so this is for a node that has stopped. +func storedMembership(t *testing.T, dataDir string) ([]byte, string) { + t.Helper() + store, err := node.NewStore(dataDir) if err != nil { - t.Fatalf("join command failed: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) + t.Fatalf("open node store: %v", err) } + defer func() { _ = store.Close() }() + identity, _ := store.LoadIdentity() + cpURL, err := store.LoadControlPlaneURL() + if err != nil { + t.Fatalf("load control plane URL: %v", err) + } + return identity, cpURL +} - out := stdout + stderr - if !strings.Contains(out, "Successfully joined the Sovereign Agent Mesh!") { - t.Fatalf("join did not succeed:\n%s", out) +// join runs `sam-node join` for the node living under env, with the mock +// provider playing the user who signs in. +func join(t *testing.T, nodeBin string, env []string, cpURL string, extra ...string) { + t.Helper() + args := append([]string{"join"}, extra...) + args = append(args, cpURL) + stdout, stderr, err := runCommandWithCallback(t, repoRoot(t), 15*time.Second, env, "", nodeBin, args...) + if err != nil { + t.Fatalf("sam-node %s: %v\nstdout:\n%s\nstderr:\n%s", strings.Join(args, " "), err, stdout, stderr) } +} - // Verify that the identity is stored and node can run - stdout, stderr, err = runCommand( - t, - repoRoot(t), - 3*time.Second, - env, - "", - nodeBin, - "run", "--control-plane", routerAddr, - "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", - "--api-token-path", tokenPath(t, "dummy-token"), - ) - if err != context.DeadlineExceeded { - t.Fatalf("expected run command to keep running, got: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) +// enrolledAs fails unless the control plane holds an enrollment for peerID +// made by subject, and returns it. +func enrolledAs(t *testing.T, cpPort int, peerID string, subject string) []byte { + t.Helper() + record := fetchAdminStatus(t, cpPort, testAdminToken).enrolledNode(peerID) + if record == nil { + t.Fatalf("control plane :%d has no enrollment for %s", cpPort, peerID) + } + if record.Role != api.RoleNode { + t.Fatalf("enrollment role = %q, want %q", record.Role, api.RoleNode) } + var claims struct { + Sub string `json:"sub"` + } + if err := json.Unmarshal([]byte(record.ClaimsJSON), &claims); err != nil { + t.Fatalf("enrollment claims %q: %v", record.ClaimsJSON, err) + } + if claims.Sub != subject { + t.Fatalf("enrolled by %q, want %q", claims.Sub, subject) + } + return record.Biscuit +} + +func TestSamNodeJoin(t *testing.T) { + nodeBin := buildBinary(t, "./cmd/sam-node") + tmpDir := t.TempDir() + oidcURL, mintToken := startCustomMockOIDC(t) + cpPort, cpURL := startMesh(t, filepath.Join(tmpDir, "mesh"), oidcURL, mintToken) + + home := filepath.Join(tmpDir, "home") + env, dataDir := nodeHome(home) + peerID := ensureNodeKey(t, dataDir).String() - out = stdout + stderr - if !strings.Contains(out, "Using stored identity.") { - t.Fatalf("node did not use stored identity:\n%s", out) + join(t, nodeBin, env, cpURL) + + // The control plane enrolled this node for the user who signed in, and + // the node holds the identity it was issued for that mesh. + issued := enrolledAs(t, cpPort, peerID, mockOIDCUser) + identity, storedURL := storedMembership(t, dataDir) + if !bytes.Equal(identity, issued) { + t.Fatal("the node's stored identity is not the biscuit the control plane issued") + } + if storedURL != cpURL { + t.Fatalf("stored control plane = %q, want %q", storedURL, cpURL) } + + // From here on the node needs neither a login nor a --control-plane: it + // comes up on what it stored and the router admits it. + n := launchNode(t, nodeBin, env, home, "run", "--allow-loopback", "--api-token-path", tokenPath(t, "node-token")) + n.waitForAPI(t) + waitForPeerOnRouter(t, cpPort, testAdminToken, peerID, 10*time.Second) } // TestSamNodeJoinOfflineAccessSavesRefreshToken guards against a real bug: @@ -126,199 +159,161 @@ func TestSamNodeJoin(t *testing.T) { // breaking --offline-access for both "join" and "run --join". func TestSamNodeJoinOfflineAccessSavesRefreshToken(t *testing.T) { nodeBin := buildBinary(t, "./cmd/sam-node") - tmpHome := t.TempDir() - env := append(os.Environ(), - "HOME="+tmpHome, - "XDG_CONFIG_HOME="+filepath.Join(tmpHome, ".config"), - "BROWSER=echo", - ) - - mux := http.NewServeMux() - mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(map[string]interface{}{ - "issuer": "http://" + r.Host, - "token_endpoint": "http://" + r.Host + "/token", - "authorization_endpoint": "http://" + r.Host + "/auth", - }); err != nil { - t.Errorf("Failed to encode response: %v", err) - } - }) - mux.HandleFunc("/device/code", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(map[string]interface{}{ - "device_code": "dev_code_123", - "user_code": "ABCD-1234", - "verification_uri": "http://example.com/verify", - "verification_uri_complete": "http://example.com/verify?code=ABCD-1234", - "expires_in": 60, - "interval": 1, - }); err != nil { - t.Errorf("Failed to encode response: %v", err) - } - }) - mux.HandleFunc("/token", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(map[string]string{ - "access_token": "test-jwt-token", - "id_token": "test-jwt-token", - "refresh_token": "test-refresh-token", - }); err != nil { - t.Errorf("Failed to encode response: %v", err) - } - }) - oidcServer := httptest.NewServer(mux) - defer oidcServer.Close() + tmpDir := t.TempDir() + oidcURL, mintToken := startCustomMockOIDC(t) + cpPort, cpURL := startMesh(t, filepath.Join(tmpDir, "mesh"), oidcURL, mintToken) - _, routerAddr := startMockRouterWithOIDC(t, oidcServer.URL) + home := filepath.Join(tmpDir, "home") + env, dataDir := nodeHome(home) + peerID := ensureNodeKey(t, dataDir).String() - stdout, stderr, err := runCommandWithCallback( - t, repoRoot(t), 5*time.Second, env, "", - nodeBin, "join", "--offline-access", routerAddr, - ) - if err != nil { - t.Fatalf("join command failed: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) - } - if !strings.Contains(stdout+stderr, "Successfully joined the Sovereign Agent Mesh!") { - t.Fatalf("join did not succeed:\n%s", stdout+stderr) - } + join(t, nodeBin, env, cpURL, "--offline-access") + enrolledAs(t, cpPort, peerID, mockOIDCUser) - store, err := node.NewStore(filepath.Join(tmpHome, ".config", "sam-mesh")) + store, err := node.NewStore(dataDir) if err != nil { - t.Fatalf("failed to open store: %v", err) + t.Fatalf("open node store: %v", err) } defer func() { _ = store.Close() }() - - refreshToken, err := store.LoadRefreshToken() + if _, err := store.LoadRefreshToken(); err != nil { + t.Fatalf("no refresh token saved after join --offline-access: %v", err) + } + issuer, clientID, _, err := store.LoadOIDCConfig() if err != nil { - t.Fatalf("expected a refresh token to be saved after --offline-access join, got error: %v", err) + t.Fatal(err) } - if refreshToken != "test-refresh-token" { - t.Errorf("expected saved refresh token %q, got %q", "test-refresh-token", refreshToken) + if issuer != oidcURL || clientID == "" { + t.Fatalf("stored OIDC config (%q, %q) is not what the control plane advertised (%q)", issuer, clientID, oidcURL) } } -// TestSamNodeRunJoinNonInteractive covers the safe fallback: since the test -// harness gives the child process no TTY (stdin defaults to /dev/null), -// "run --join" must not attempt (and block on) an interactive OIDC login; it -// should fall back to the same unauthenticated MCP sidecar as plain "run". -func TestSamNodeRunJoinNonInteractive(t *testing.T) { +// TestSamNodeRunWithoutIdentityServesEnrollment covers a node that has never +// enrolled: it must not fail, nor block on a login it cannot complete +// without a terminal, but serve the enrollment-only MCP surface until +// someone enrolls it. +func TestSamNodeRunWithoutIdentityServesEnrollment(t *testing.T) { nodeBin := buildBinary(t, "./cmd/sam-node") - tmpHome := t.TempDir() - env := []string{ - "HOME=" + tmpHome, - "XDG_CONFIG_HOME=" + filepath.Join(tmpHome, ".config"), + for _, tc := range []struct { + name string + args []string + }{ + {"plain run", nil}, + // --join without a TTY has nobody to log in; the node falls back + // rather than hanging on a prompt. The control plane is never + // contacted, so any address will do. + {"run --join without a TTY", []string{"--join", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", getFreePort(t))}}, + } { + t.Run(tc.name, func(t *testing.T) { + home := t.TempDir() + env, dataDir := nodeHome(home) + n := launchNode(t, nodeBin, env, home, append([]string{"run"}, tc.args...)...) + n.waitForEnrollmentSidecar(t) + n.staysUp(t, time.Second) + n.kill() + if identity, _ := storedMembership(t, dataDir); len(identity) != 0 { + t.Fatal("a node that never enrolled holds an identity") + } + }) } +} + +// TestSamNodeRunWithStoredIdentity covers the everyday restart: a node that +// enrolled once comes back with no token and no --control-plane, on what it +// stored, and the router admits it again. +func TestSamNodeRunWithStoredIdentity(t *testing.T) { + nodeBin := buildBinary(t, "./cmd/sam-node") + tmpDir := t.TempDir() + oidcURL, mintToken := startCustomMockOIDC(t) + cpPort, cpURL := startMesh(t, filepath.Join(tmpDir, "mesh"), oidcURL, mintToken) + + home := filepath.Join(tmpDir, "home") + env, dataDir := nodeHome(home) + token := tokenPath(t, "node-token") - _, routerAddr := startMockRouter(t) - - stdout, stderr, err := runCommand( - t, - repoRoot(t), - 3*time.Second, - env, - "", // no stdin: child sees no TTY, same as a container without -it - nodeBin, - "run", "--join", "--control-plane", routerAddr, - "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", + first := launchNode(t, nodeBin, env, home, "run", + "--control-plane", cpURL, + "--jwt", mintToken(map[string]interface{}{"sub": jwtUser}), + "--allow-loopback", + "--api-token-path", token, ) - if err != context.DeadlineExceeded { - t.Fatalf("expected run --join to keep running via the unauthenticated sidecar, got: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) - } + first.waitForAPI(t) + issued := enrolledAs(t, cpPort, first.peerID.String(), jwtUser) + first.kill() - out := stdout + stderr - if !strings.Contains(out, "no interactive terminal available") { - t.Fatalf("expected --join to warn about falling back without a TTY:\n%s", out) + // /healthz answers only once Start succeeded, and Start fails unless a + // router authenticated the node: coming up is the proof. + again := launchNode(t, nodeBin, env, home, "run", "--allow-loopback", "--api-token-path", token) + again.waitForAPI(t) + if again.peerID != first.peerID { + t.Fatalf("peer ID changed across restarts: %s then %s", first.peerID, again.peerID) } - if !strings.Contains(out, "Starting unauthenticated sidecar for enrollment over MCP") { - t.Fatalf("expected the unauthenticated sidecar to start:\n%s", out) + again.kill() + identity, _ := storedMembership(t, dataDir) + if !bytes.Equal(identity, issued) { + t.Fatal("restart replaced the stored identity instead of using it") } } -// TestSamNodeRunControlPlaneMismatch covers a node enrolled with one control -// plane being pointed at a different one: it must fail loudly (never -// silently keep using the originally-enrolled mesh), with or without -// --join, and "reset" must clear the way for a fresh enrollment elsewhere. +// TestSamNodeRunControlPlaneMismatch covers a node enrolled with one mesh +// being pointed at another: it must refuse, with or without --join, keeping +// its identity for the first mesh and leaving no trace on the second, and +// "reset" must clear the way for a fresh enrollment there. func TestSamNodeRunControlPlaneMismatch(t *testing.T) { nodeBin := buildBinary(t, "./cmd/sam-node") - tmpHome := t.TempDir() - env := []string{ - "HOME=" + tmpHome, - "XDG_CONFIG_HOME=" + filepath.Join(tmpHome, ".config"), - } + tmpDir := t.TempDir() + oidcURL, mintToken := startCustomMockOIDC(t) + cpPortA, cpURLA := startMesh(t, filepath.Join(tmpDir, "meshA"), oidcURL, mintToken) + cpPortB, cpURLB := startMesh(t, filepath.Join(tmpDir, "meshB"), oidcURL, mintToken) - _, routerA := startMockRouter(t) - _, routerB := startMockRouter(t) - - // Enroll against control plane A. - stdout, stderr, err := runCommand( - t, repoRoot(t), 3*time.Second, env, "", - nodeBin, "run", "--control-plane", routerA, "--jwt", "test-jwt", - "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", - "--api-token-path", tokenPath(t, "dummy-token"), - ) - if err != context.DeadlineExceeded { - t.Fatalf("expected initial enroll+run against A to keep running, got: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) - } + home := filepath.Join(tmpDir, "home") + env, dataDir := nodeHome(home) + token := tokenPath(t, "node-token") + jwt := mintToken(map[string]interface{}{"sub": jwtUser}) - // Pointing the same store at B, without --join, must fail rather than - // silently keep talking to A. - stdout, stderr, err = runCommand( - t, repoRoot(t), 5*time.Second, env, "", - nodeBin, "run", "--control-plane", routerB, - "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", - "--api-token-path", tokenPath(t, "dummy-token"), - ) - if err == nil || err == context.DeadlineExceeded { - t.Fatalf("expected run against a mismatched control plane to fail fast, got: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) - } - out := stdout + stderr - if !strings.Contains(out, "does not match the mesh") { - t.Fatalf("expected a control-plane mismatch error, got:\n%s", out) - } + // Enroll with mesh A. + onA := launchNode(t, nodeBin, env, home, "run", "--control-plane", cpURLA, "--jwt", jwt, "--allow-loopback", "--api-token-path", token) + onA.waitForAPI(t) + peerID := onA.peerID.String() + issuedByA := enrolledAs(t, cpPortA, peerID, jwtUser) + onA.kill() - // Same mismatch with --join, but no TTY to confirm the switch: must also - // fail fast rather than silently ignore --join or block indefinitely. - stdout, stderr, err = runCommand( - t, repoRoot(t), 5*time.Second, env, "", - nodeBin, "run", "--join", "--control-plane", routerB, - "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", - "--api-token-path", tokenPath(t, "dummy-token"), - ) - if err == nil || err == context.DeadlineExceeded { - t.Fatalf("expected run --join against a mismatched control plane to fail fast without a TTY, got: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) - } - out = stdout + stderr - if !strings.Contains(out, "does not match the mesh") { - t.Fatalf("expected a control-plane mismatch error, got:\n%s", out) + // Pointed at B, with or without --join (no TTY to confirm a switch), + // the node refuses rather than reusing A's identity or enrolling anew. + for _, args := range [][]string{ + {"run", "--control-plane", cpURLB}, + {"run", "--join", "--control-plane", cpURLB}, + } { + n := launchNode(t, nodeBin, env, home, append(args, "--allow-loopback", "--api-token-path", token)...) + if err := n.exitsWithin(t, 10*time.Second); err == nil { + t.Fatalf("sam-node %s exited cleanly against a mesh it is not enrolled with", strings.Join(args, " ")) + } + identity, storedURL := storedMembership(t, dataDir) + if !bytes.Equal(identity, issuedByA) || storedURL != cpURLA { + t.Fatalf("sam-node %s changed the stored membership (control plane now %q)", strings.Join(args, " "), storedURL) + } + if fetchAdminStatus(t, cpPortB, testAdminToken).enrolledNode(peerID) != nil { + t.Fatalf("sam-node %s enrolled with mesh B", strings.Join(args, " ")) + } } - // "reset" clears the way for a fresh enrollment against B. - stdout, stderr, err = runCommand(t, repoRoot(t), 3*time.Second, env, "", nodeBin, "reset") + // "reset" forgets the membership and nothing else: the key, hence the + // peer ID, stays. + stdout, stderr, err := runCommand(t, repoRoot(t), 5*time.Second, env, "", nodeBin, "reset") if err != nil { - t.Fatalf("reset failed: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) + t.Fatalf("sam-node reset: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) } - if !strings.Contains(stdout+stderr, "Cleared stored mesh identity") { - t.Fatalf("expected reset confirmation, got:\n%s", stdout+stderr) + if identity, storedURL := storedMembership(t, dataDir); len(identity) != 0 || storedURL != "" { + t.Fatalf("reset left a membership behind (control plane %q)", storedURL) } - stdout, stderr, err = runCommand( - t, repoRoot(t), 3*time.Second, env, "", - nodeBin, "run", "--control-plane", routerB, "--jwt", "test-jwt", - "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", - "--api-token-path", tokenPath(t, "dummy-token"), - ) - if err != context.DeadlineExceeded { - t.Fatalf("expected re-enroll against B after reset to keep running, got: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr) + onB := launchNode(t, nodeBin, env, home, "run", "--control-plane", cpURLB, "--jwt", jwt, "--allow-loopback", "--api-token-path", token) + onB.waitForAPI(t) + if onB.peerID.String() != peerID { + t.Fatalf("reset changed the peer ID: %s then %s", peerID, onB.peerID) + } + enrolledAs(t, cpPortB, peerID, jwtUser) + onB.kill() + if _, storedURL := storedMembership(t, dataDir); storedURL != cpURLB { + t.Fatalf("stored control plane = %q, want %q", storedURL, cpURLB) } } diff --git a/tests/integration/minimal_helpers_test.go b/tests/integration/minimal_helpers_test.go index d2fe2dff..5ed1a560 100644 --- a/tests/integration/minimal_helpers_test.go +++ b/tests/integration/minimal_helpers_test.go @@ -25,6 +25,7 @@ import ( "path/filepath" "runtime" "strings" + "sync" "testing" "time" @@ -36,10 +37,10 @@ import ( "net" "net/http" "net/http/httptest" - "net/url" "github.com/biscuit-auth/biscuit-go/v2" "github.com/google/sam/api" + "github.com/google/sam/internal/storage" "github.com/libp2p/go-libp2p" dht "github.com/libp2p/go-libp2p-kad-dht" "github.com/libp2p/go-libp2p/core/network" @@ -57,16 +58,50 @@ func repoRoot(t *testing.T) string { return filepath.Clean(filepath.Join(filepath.Dir(thisFile), "..", "..")) } +func TestMain(m *testing.M) { + code := m.Run() + builtBinaries.Range(func(_, entry any) bool { + if b := entry.(*binaryBuild); b.path != "" { + _ = os.RemoveAll(filepath.Dir(b.path)) + } + return true + }) + os.Exit(code) +} + +// builtBinaries holds one build per package for the whole test process: +// Go caches compilation, but every `go build -o` links again, and fifty +// tests each linking three binaries is a minute of nothing. +var builtBinaries sync.Map // pkgPath -> *binaryBuild + +type binaryBuild struct { + once sync.Once + path string + err error +} + func buildBinary(t *testing.T, pkgPath string) string { t.Helper() - root := repoRoot(t) - out := filepath.Join(t.TempDir(), filepath.Base(pkgPath)) - cmd := exec.Command("go", "build", "-o", out, ".") - cmd.Dir = filepath.Join(root, pkgPath) - if output, err := cmd.CombinedOutput(); err != nil { - t.Fatalf("building %s failed: %v\n%s", pkgPath, err, string(output)) - } - return out + entry, _ := builtBinaries.LoadOrStore(pkgPath, &binaryBuild{}) + b := entry.(*binaryBuild) + b.once.Do(func() { + root := repoRoot(t) + dir, err := os.MkdirTemp("", "sam-integration-bin-") + if err != nil { + b.err = err + return + } + b.path = filepath.Join(dir, filepath.Base(pkgPath)) + cmd := exec.Command("go", "build", "-o", b.path, ".") + cmd.Dir = filepath.Join(root, pkgPath) + if output, err := cmd.CombinedOutput(); err != nil { + b.err = fmt.Errorf("%v\n%s", err, output) + } + }) + if b.err != nil { + t.Fatalf("building %s failed: %v", pkgPath, b.err) + } + return b.path } func runCommand( @@ -133,23 +168,18 @@ func (iw *interceptorWriter) Write(p []byte) (n int, err error) { if strings.Contains(s, "redirect_uri=") && strings.Contains(s, "state=") { iw.handled = true go func() { - // Parse URL from the buffer to extract state and redirect_uri - // The URL is printed like: " http://...redirect_uri=...&state=..." - lines := strings.Split(s, "\n") - for _, line := range lines { - if strings.Contains(line, "redirect_uri=") { - parts := strings.Split(strings.TrimSpace(line), " ") - for _, p := range parts { - if strings.HasPrefix(p, "http") { - u, err := url.Parse(p) - if err == nil { - redirectURI := u.Query().Get("redirect_uri") - state := u.Query().Get("state") - if redirectURI != "" && state != "" { - time.Sleep(100 * time.Millisecond) - _, _ = http.Get(redirectURI + "?code=dev_code_123&state=" + state) - } - } + // Play the user: open the authorization URL the CLI printed. The + // provider signs the user in and redirects to the CLI's loopback + // callback, which the client follows. + for _, line := range strings.Split(s, "\n") { + if !strings.Contains(line, "redirect_uri=") { + continue + } + for _, p := range strings.Split(strings.TrimSpace(line), " ") { + if strings.HasPrefix(p, "http") { + time.Sleep(100 * time.Millisecond) + if resp, err := http.Get(p); err == nil { + _ = resp.Body.Close() } } } @@ -296,106 +326,6 @@ func startMockRouterWithControlPlaneKey(t *testing.T) (peer.ID, string, ed25519. return h.ID(), httpServer.URL, pub } -func startMockRouterWithOIDC(t *testing.T, oidcIssuerURL string) (peer.ID, string) { - t.Helper() - - pub, priv, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - t.Fatalf("Failed to generate control plane key: %v", err) - } - - h, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) - if err != nil { - t.Fatalf("failed to create mock libp2p host: %v", err) - } - - h.SetStreamHandler(api.AuthProtocolID, func(s network.Stream) { - defer func() { _ = s.Close() }() - reader := msgio.NewVarintReaderSize(s, 1024*64) - msg, err := reader.ReadMsg() - if err != nil { - return - } - defer reader.ReleaseMsg(msg) - - writer := msgio.NewVarintWriter(s) - resp := &api.AuthResponse{ - Success: true, - Biscuit: createMockBiscuitToken(t, h.ID().String(), priv, api.RoleRouter, nil), - } - respBytes, _ := proto.Marshal(resp) - _ = writer.WriteMsg(respBytes) - }) - - kdht, err := dht.New(h, dht.Mode(dht.ModeServer), dht.ProtocolPrefix("/sam")) - if err != nil { - t.Fatalf("failed to create DHT on mock router: %v", err) - } - - // Start HTTP server for enrollment and info - - mux := http.NewServeMux() - mux.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - resp := &api.ControlPlaneInfoResponse{ - OidcIssuer: oidcIssuerURL, - ClientId: "sam-mesh-audience", - Audience: "sam-mesh-audience", - } - data, err := proto.Marshal(resp) - if err != nil { - http.Error(w, "Failed to marshal response", http.StatusInternalServerError) - return - } - w.Header().Set("Content-Type", "application/x-protobuf") - w.WriteHeader(http.StatusOK) - _, _ = w.Write(data) - }) - mux.HandleFunc("/register", func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - body, err := io.ReadAll(r.Body) - if err != nil { - http.Error(w, "Failed to read body", http.StatusBadRequest) - return - } - var req api.EnrollRequest - if err := proto.Unmarshal(body, &req); err != nil { - http.Error(w, "Invalid request format", http.StatusBadRequest) - return - } - - resp := &api.EnrollResponse{ - BiscuitToken: createMockBiscuitToken(t, req.PeerId, priv, api.RoleNode, req.Labels), - ControlPlanePublicKey: pub, - RouterAddresses: []string{h.Addrs()[0].String() + "/p2p/" + h.ID().String()}, - } - data, err := proto.Marshal(resp) - if err != nil { - http.Error(w, "Failed to marshal response", http.StatusInternalServerError) - return - } - w.Header().Set("Content-Type", "application/x-protobuf") - w.WriteHeader(http.StatusOK) - _, _ = w.Write(data) - }) - - httpServer := httptest.NewServer(mux) - - t.Cleanup(func() { - httpServer.Close() - _ = kdht.Close() - _ = h.Close() - }) - - return h.ID(), httpServer.URL -} - func createMockBiscuitToken(t *testing.T, peerID string, priv ed25519.PrivateKey, role string, labels map[string]string) []byte { builder := biscuit.NewBuilder(priv) err := builder.AddAuthorityFact(biscuit.Fact{Predicate: biscuit.Predicate{Name: "target_unrestricted"}}) @@ -472,11 +402,27 @@ func createMockBiscuitToken(t *testing.T, peerID string, priv ed25519.PrivateKey } func startControlPlaneAndRouter(t *testing.T, tmpDir string, oidcURL string, mintToken func(map[string]interface{}) string, policyFile string) (int, func()) { - cpBin := buildBinary(t, "./cmd/sam-control-plane") - routerBin := buildBinary(t, "./cmd/sam-router") + cpPort, stopCP := startControlPlane(t, tmpDir, oidcURL, policyFile) + _, stopRouter := startRouter(t, tmpDir, cpPort, mintToken, "router") + + // Wait for router lease to be active and registered + fetchPeerID(t, cpPort) + + cleanup := func() { + stopRouter() + stopCP() + } + + return cpPort, cleanup +} +// startControlPlane starts a sam-control-plane under tmpDir, trusting oidcURL +// and holding the policy in policyFile with the router role added, plus any +// extra flags. It returns the port and a stop function. +func startControlPlane(t *testing.T, tmpDir string, oidcURL string, policyFile string, extra ...string) (int, func()) { + t.Helper() + cpBin := buildBinary(t, "./cmd/sam-control-plane") cpPort := getFreePort(t) - routerPort := getFreePort(t) // Automatically adjust the policy file to grant the "router" role to group "routers" originalPolicy, err := os.ReadFile(policyFile) @@ -484,29 +430,40 @@ func startControlPlaneAndRouter(t *testing.T, tmpDir string, oidcURL string, min writePolicyWithRouter(t, policyFile, string(originalPolicy)) } - // 1. Start Control Plane - cpCmd := exec.Command(cpBin, + cpCmd := exec.Command(cpBin, append([]string{ "--bind-address", fmt.Sprintf("127.0.0.1:%d", cpPort), "--db-dsn", filepath.Join(tmpDir, "cp-keys.db"), "--issuer", oidcURL, "--insecure-skip-tls-verify", "--admin-token-path", tokenPath(t, "test-admin-token"), - ) + }, extra...)...) cpCmd.Stdout = os.Stdout cpCmd.Stderr = os.Stderr if err := cpCmd.Start(); err != nil { t.Fatalf("failed to start control plane: %v", err) } + stop := func() { + _ = cpCmd.Process.Kill() + _ = cpCmd.Wait() + } + t.Cleanup(stop) - // Wait for CP to be up waitForControlPlane(t, cpPort) - - // Inject policy into CP database via API injectPolicyYAML(t, cpPort, "test-admin-token", policyFile) + return cpPort, stop +} + +// startRouter starts a sam-router named name against the control plane on +// cpPort, renewing its lease every second so the control plane's view of the +// router's peers is current. It returns the router's p2p address and a stop +// function. +func startRouter(t *testing.T, tmpDir string, cpPort int, mintToken func(map[string]interface{}) string, name string) (string, func()) { + t.Helper() + routerBin := buildBinary(t, "./cmd/sam-router") + routerPort := getFreePort(t) - // 2. Start Router routerJWT := mintToken(map[string]interface{}{ - "sub": "router-integration-1", + "sub": name, "groups": []string{"routers"}, "roles": []string{api.RoleRouter}, }) @@ -514,29 +471,40 @@ func startControlPlaneAndRouter(t *testing.T, tmpDir string, oidcURL string, min routerCmd := exec.Command(routerBin, "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", cpPort), "--listen", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", routerPort), - "--keys-path", filepath.Join(tmpDir, "router-keys.db"), + "--keys-path", filepath.Join(tmpDir, name+"-keys.db"), "--allow-loopback", "--oidc-token", routerJWT, + // Each renewal carries the router's connected peers, which is how + // tests see a node reach the mesh through the control plane. + "--lease-renew-interval", "1s", ) routerCmd.Stdout = os.Stdout routerCmd.Stderr = os.Stderr if err := routerCmd.Start(); err != nil { - _ = cpCmd.Process.Kill() - _ = cpCmd.Wait() t.Fatalf("failed to start router: %v", err) } - - // Wait for router lease to be active and registered - fetchPeerID(t, cpPort) - - cleanup := func() { + stop := func() { _ = routerCmd.Process.Kill() _ = routerCmd.Wait() - _ = cpCmd.Process.Kill() - _ = cpCmd.Wait() } + t.Cleanup(stop) - return cpPort, cleanup + // The router's peer ID is its own; the control plane learns it from the + // first lease, which is also when the router is ready for nodes. + deadline := time.Now().Add(10 * time.Second) + for { + for _, lease := range fetchAdminStatus(t, cpPort, "test-admin-token").ActiveRouters { + for _, addr := range lease.Addresses { + if strings.HasPrefix(addr, fmt.Sprintf("/ip4/127.0.0.1/tcp/%d/", routerPort)) { + return addr, stop + } + } + } + if time.Now().After(deadline) { + t.Fatalf("router %s never leased with control plane :%d", name, cpPort) + } + time.Sleep(100 * time.Millisecond) + } } func getFreePort(t *testing.T) int { @@ -550,6 +518,74 @@ func getFreePort(t *testing.T) int { return port } +// adminStatus is the control plane's view of the mesh, in the types it +// serializes on /admin/status. +type adminStatus struct { + EnrolledNodes []storage.EnrolledNode `json:"enrolled_nodes"` + ActiveRouters []storage.RouterLease `json:"active_routers"` +} + +func fetchAdminStatus(t *testing.T, cpPort int, adminToken string) adminStatus { + t.Helper() + req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/admin/status", cpPort), nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer "+adminToken) + client := &http.Client{Timeout: 5 * time.Second} + resp, err := client.Do(req) + if err != nil { + t.Fatalf("GET /admin/status: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET /admin/status: %s", resp.Status) + } + var status adminStatus + if err := json.NewDecoder(resp.Body).Decode(&status); err != nil { + t.Fatalf("decode /admin/status: %v", err) + } + return status +} + +// enrolledNode is the control plane's record of peerID, or nil. +func (s adminStatus) enrolledNode(peerID string) *storage.EnrolledNode { + for i := range s.EnrolledNodes { + if s.EnrolledNodes[i].PeerID == peerID { + return &s.EnrolledNodes[i] + } + } + return nil +} + +// routerWith is the lease of a router that reports peerID connected, or nil. +func (s adminStatus) routerWith(peerID string) *storage.RouterLease { + for i := range s.ActiveRouters { + for _, p := range s.ActiveRouters[i].ConnectedPeers { + if p == peerID { + return &s.ActiveRouters[i] + } + } + } + return nil +} + +// waitForPeerOnRouter returns the lease of the router peerID is connected +// to, as the control plane learns it from the router's lease renewals. +func waitForPeerOnRouter(t *testing.T, cpPort int, adminToken string, peerID string, timeout time.Duration) *storage.RouterLease { + t.Helper() + deadline := time.Now().Add(timeout) + for { + if lease := fetchAdminStatus(t, cpPort, adminToken).routerWith(peerID); lease != nil { + return lease + } + if time.Now().After(deadline) { + t.Fatalf("no router on control plane :%d reported %s connected within %v", cpPort, peerID, timeout) + } + time.Sleep(200 * time.Millisecond) + } +} + func fetchPeerID(t *testing.T, port int) string { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -682,24 +718,6 @@ func writePolicyWithRouter(t *testing.T, path string, yamlContent string) { } } -func waitForNodeOnline(t *testing.T, logPath string) { - t.Helper() - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - - for { - data, err := os.ReadFile(logPath) - if err == nil && strings.Contains(string(data), "SAM Node Online") { - return - } - select { - case <-ctx.Done(): - t.Fatalf("timed out waiting for node to go online") - case <-time.After(100 * time.Millisecond): - } - } -} - // injectPolicyYAML posts a policy fixture through the same conversion the console // performs: YAML for readability, protojson on the wire. Going through JSON keeps // this helper free of any field list, so a new PolicyRole field needs no change diff --git a/tests/integration/mock_oidc_test.go b/tests/integration/mock_oidc_test.go index 708e20a9..54f12c32 100644 --- a/tests/integration/mock_oidc_test.go +++ b/tests/integration/mock_oidc_test.go @@ -17,17 +17,31 @@ package integration_test import ( "crypto/rand" "crypto/rsa" + "crypto/sha256" "encoding/base64" + "encoding/hex" "encoding/json" "math/big" "net/http" "net/http/httptest" + "net/url" + "strings" + "sync" "testing" "time" "github.com/golang-jwt/jwt/v5" ) +// mockOIDCUser is the subject the mock provider signs in as whoever +// completes its browser authorization flow. +const mockOIDCUser = "browser-user" + +// startCustomMockOIDC is the identity provider the mesh trusts in these +// tests: discovery, a JWKS, and the authorization code grant with PKCE that +// an interactive join drives. The tokens it issues are real RS256 JWTs a +// control plane verifies against the JWKS. mintToken issues one directly, +// for components that are configured with a token instead of logging in. func startCustomMockOIDC(t *testing.T) (string, func(claims map[string]interface{}) string) { privKey, err := rsa.GenerateKey(rand.Reader, 2048) if err != nil { @@ -42,8 +56,10 @@ func startCustomMockOIDC(t *testing.T) (string, func(claims map[string]interface mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "issuer": issuer, - "jwks_uri": issuer + "/keys", + "issuer": issuer, + "jwks_uri": issuer + "/keys", + "authorization_endpoint": issuer + "/auth", + "token_endpoint": issuer + "/token", }) }) @@ -81,5 +97,91 @@ func startCustomMockOIDC(t *testing.T) (string, func(claims map[string]interface return jwtStr } + // Authorization codes handed out by /auth, redeemed once at /token. + type authCode struct { + challenge string + redirectURI string + offline bool + } + var mu sync.Mutex + codes := map[string]authCode{} + newCode := func() string { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + t.Fatalf("random code: %v", err) + } + return hex.EncodeToString(b) + } + + // The user opens this URL; the provider signs them in as mockOIDCUser + // and sends them back to the client with a code bound to its PKCE + // challenge. + mux.HandleFunc("/auth", func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + if q.Get("response_type") != "code" || q.Get("client_id") == "" || q.Get("redirect_uri") == "" || q.Get("code_challenge_method") != "S256" { + http.Error(w, "unsupported authorization request", http.StatusBadRequest) + return + } + code := newCode() + mu.Lock() + codes[code] = authCode{ + challenge: q.Get("code_challenge"), + redirectURI: q.Get("redirect_uri"), + offline: strings.Contains(q.Get("scope"), "offline_access"), + } + mu.Unlock() + back, err := url.Parse(q.Get("redirect_uri")) + if err != nil { + http.Error(w, "bad redirect_uri", http.StatusBadRequest) + return + } + bq := back.Query() + bq.Set("code", code) + bq.Set("state", q.Get("state")) + back.RawQuery = bq.Encode() + http.Redirect(w, r, back.String(), http.StatusFound) + }) + + oauthError := func(w http.ResponseWriter, code string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]string{"error": code}) + } + mux.HandleFunc("/token", func(w http.ResponseWriter, r *http.Request) { + if err := r.ParseForm(); err != nil { + oauthError(w, "invalid_request") + return + } + if r.PostForm.Get("grant_type") != "authorization_code" { + oauthError(w, "unsupported_grant_type") + return + } + mu.Lock() + granted, ok := codes[r.PostForm.Get("code")] + delete(codes, r.PostForm.Get("code")) + mu.Unlock() + verifier := sha256.Sum256([]byte(r.PostForm.Get("code_verifier"))) + if !ok || granted.redirectURI != r.PostForm.Get("redirect_uri") || + base64.RawURLEncoding.EncodeToString(verifier[:]) != granted.challenge { + oauthError(w, "invalid_grant") + return + } + idToken := mintToken(map[string]interface{}{ + "sub": mockOIDCUser, + "email": mockOIDCUser + "@example.com", + }) + resp := map[string]interface{}{ + "access_token": idToken, + "id_token": idToken, + "token_type": "Bearer", + "expires_in": 3600, + } + if granted.offline { + resp["refresh_token"] = newCode() + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(resp) + }) + return issuer, mintToken } diff --git a/tests/integration/node_helpers_test.go b/tests/integration/node_helpers_test.go new file mode 100644 index 00000000..58ebd2dc --- /dev/null +++ b/tests/integration/node_helpers_test.go @@ -0,0 +1,313 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package integration_test + +import ( + "context" + "fmt" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/google/sam/internal/node" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// A node under test is observed through what it does, never through what it +// prints: the log is kept for the failure message only. Its identity and +// addresses are known before it starts, because the test chooses the ports +// and the node's key lives in its store, so nothing has to be read back out. + +// backgroundNode is a `sam-node run` started by a test. +type backgroundNode struct { + cmd *exec.Cmd + apiAddr string // host:port the sidecar API listens on + token string // API token, "" for a node running the enrollment sidecar + dataDir string // the node's store + peerID peer.ID // derived from the key in dataDir before the node started + p2pAddr string // a TCP address the node listens on, with /p2p/ + logPath string + exited chan error +} + +// launchNode starts nodeBin with args plus a --bind-addr and a --listen the +// test chose. Both are appended so they win over (bind) or add to (listen) +// what args carry. The node's key is generated in its store first, so its +// peer ID is known up front. Output goes to logDir/node.log. +func launchNode(t *testing.T, nodeBin string, env []string, logDir string, args ...string) *backgroundNode { + t.Helper() + if err := os.MkdirAll(logDir, 0o755); err != nil { + t.Fatalf("create log dir: %v", err) + } + logPath := filepath.Join(logDir, "node.log") + logFile, err := os.Create(logPath) + if err != nil { + t.Fatalf("create node log: %v", err) + } + + dataDir := nodeDataDir(env, args) + peerID := ensureNodeKey(t, dataDir) + apiAddr := fmt.Sprintf("127.0.0.1:%d", getFreePort(t)) + p2pPort := getFreePort(t) + + fullArgs := append(append([]string{}, args...), + "--bind-addr", apiAddr, + "--listen", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", p2pPort), + ) + cmd := exec.Command(nodeBin, fullArgs...) + cmd.Dir = repoRoot(t) + cmd.Env = env + cmd.Stdout = logFile + cmd.Stderr = logFile + if err := cmd.Start(); err != nil { + _ = logFile.Close() + t.Fatalf("start sam-node: %v", err) + } + n := &backgroundNode{ + cmd: cmd, + apiAddr: apiAddr, + token: nodeToken(t, env, args), + dataDir: dataDir, + peerID: peerID, + p2pAddr: fmt.Sprintf("/ip4/127.0.0.1/tcp/%d/p2p/%s", p2pPort, peerID), + logPath: logPath, + exited: make(chan error, 1), + } + go func() { + n.exited <- cmd.Wait() + _ = logFile.Close() + }() + t.Cleanup(n.kill) + return n +} + +// nodeDataDir resolves the store the node will open, the way the binary +// does: --data-dir, else the user config dir under XDG_CONFIG_HOME or HOME. +func nodeDataDir(env []string, args []string) string { + for i, a := range args { + if a == "--data-dir" && i+1 < len(args) { + return args[i+1] + } + } + var home, xdg string + for _, e := range env { + if v, ok := strings.CutPrefix(e, "XDG_CONFIG_HOME="); ok { + xdg = v + } + if v, ok := strings.CutPrefix(e, "HOME="); ok { + home = v + } + } + if xdg != "" { + return filepath.Join(xdg, "sam-mesh") + } + return filepath.Join(home, ".config", "sam-mesh") +} + +// ensureNodeKey creates the node's key in dataDir if there is none, exactly +// as the node itself would, and returns the peer ID it implies. The store is +// closed again before the node starts: bbolt holds an exclusive lock. +func ensureNodeKey(t *testing.T, dataDir string) peer.ID { + t.Helper() + store, err := node.NewStore(dataDir) + if err != nil { + t.Fatalf("open node store %s: %v", dataDir, err) + } + priv := node.GetOrGenerateKey(store) + if err := store.Close(); err != nil { + t.Fatalf("close node store: %v", err) + } + id, err := peer.IDFromPrivateKey(priv) + if err != nil { + t.Fatalf("peer ID from node key: %v", err) + } + return id +} + +func nodeToken(t *testing.T, env []string, args []string) string { + t.Helper() + for i, a := range args { + if a == "--api-token-path" && i+1 < len(args) { + data, err := os.ReadFile(args[i+1]) + if err != nil { + t.Fatalf("read api token: %v", err) + } + return strings.TrimSpace(string(data)) + } + } + for _, e := range env { + if v, ok := strings.CutPrefix(e, "SAM_API_TOKEN="); ok { + return v + } + } + return "" +} + +// startBackgroundNode is launchNode for a node that enrolls with --jwt +// against routerAddr, with homeDir as its home and the shared test token. +func startBackgroundNode(t *testing.T, nodeBin string, routerAddr string, homeDir string, args ...string) *backgroundNode { + t.Helper() + env := append(os.Environ(), + "HOME="+homeDir, + "XDG_CONFIG_HOME="+filepath.Join(homeDir, ".config"), + "SAM_API_TOKEN=test-token", // per-test overrides use --api-token-path, which wins + ) + allArgs := append([]string{"run", "--control-plane", routerAddr, "--jwt", "test-jwt", "--allow-loopback"}, args...) + return launchNode(t, nodeBin, env, homeDir, allArgs...) +} + +func (n *backgroundNode) kill() { + if n.cmd.Process != nil { + _ = n.cmd.Process.Kill() + } + <-n.exited + // Re-arm so a second kill (test cleanup after an explicit one) does not block. + n.exited <- nil +} + +// log is the node's output, for failure messages only. +func (n *backgroundNode) log() string { + data, err := os.ReadFile(n.logPath) + if err != nil { + return fmt.Sprintf("(no log: %v)", err) + } + return string(data) +} + +// waitForAPI returns the API address once /healthz answers, which the node +// only does after Start succeeded (enrolled, router authenticated), and fails +// if the node exits first or does not come up in time. +func (n *backgroundNode) waitForAPI(t *testing.T) string { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + client := &http.Client{Timeout: time.Second} + for time.Now().Before(deadline) { + select { + case err := <-n.exited: + n.exited <- err + t.Fatalf("sam-node exited (%v) before serving its API.\n--- node.log ---\n%s", err, n.log()) + default: + } + resp, err := client.Get("http://" + n.apiAddr + "/healthz") + if err == nil { + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return n.apiAddr + } + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("sam-node did not serve its API at %s in time.\n--- node.log ---\n%s", n.apiAddr, n.log()) + return "" +} + +// exitsWithin fails unless the node exits within d; it returns the exit error. +func (n *backgroundNode) exitsWithin(t *testing.T, d time.Duration) error { + t.Helper() + select { + case err := <-n.exited: + n.exited <- err + return err + case <-time.After(d): + t.Fatalf("sam-node still running after %s.\n--- node.log ---\n%s", d, n.log()) + return nil + } +} + +// staysUp fails if the node exits within d. +func (n *backgroundNode) staysUp(t *testing.T, d time.Duration) { + t.Helper() + select { + case err := <-n.exited: + n.exited <- err + t.Fatalf("sam-node exited (%v) within %s.\n--- node.log ---\n%s", err, d, n.log()) + case <-time.After(d): + } +} + +// waitForEnrollmentSidecar returns once the node serves the enrollment-only +// MCP surface: a session opened without a token that offers +// get_login_instructions. A node holding an identity refuses token-less +// sessions, so this also tells which of the two sidecars came up. +func (n *backgroundNode) waitForEnrollmentSidecar(t *testing.T) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + var last error + for time.Now().Before(deadline) { + select { + case err := <-n.exited: + n.exited <- err + t.Fatalf("sam-node exited (%v) before serving its enrollment sidecar.\n--- node.log ---\n%s", err, n.log()) + default: + } + tools, err := listMCPTools(n.apiAddr) + if err == nil { + for _, name := range tools { + if name == "get_login_instructions" { + return + } + } + err = fmt.Errorf("tools %v do not include get_login_instructions", tools) + } + last = err + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("sam-node at %s did not serve the enrollment sidecar in time: %v\n--- node.log ---\n%s", n.apiAddr, last, n.log()) +} + +// listMCPTools opens a token-less MCP session against apiAddr and returns the +// names of the tools it offers. +func listMCPTools(apiAddr string) ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + client := mcp.NewClient(&mcp.Implementation{Name: "test-client", Version: "0.1.0"}, nil) + session, err := client.Connect(ctx, &mcp.StreamableClientTransport{Endpoint: "http://" + apiAddr + "/mcp"}, nil) + if err != nil { + return nil, err + } + defer func() { _ = session.Close() }() + res, err := session.ListTools(ctx, nil) + if err != nil { + return nil, err + } + names := make([]string, 0, len(res.Tools)) + for _, tool := range res.Tools { + names = append(names, tool.Name) + } + return names, nil +} + +// waitForAPI polls addr's /healthz for a node the test started itself. +func waitForAPI(t *testing.T, addr string) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + client := &http.Client{Timeout: time.Second} + for time.Now().Before(deadline) { + resp, err := client.Get("http://" + addr + "/healthz") + if err == nil { + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return + } + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("timeout waiting for API at %s", addr) +} diff --git a/tests/integration/openai_facade_test.go b/tests/integration/openai_facade_test.go index 729519dd..7db9438f 100644 --- a/tests/integration/openai_facade_test.go +++ b/tests/integration/openai_facade_test.go @@ -19,7 +19,6 @@ import ( "io" "net/http" "net/http/httptest" - "path/filepath" "strings" "sync/atomic" "testing" @@ -75,7 +74,7 @@ func TestOpenAIFacadeCUJ(t *testing.T) { defer backend.Close() t.Log("Starting Node A (provider)...") - _ = startBackgroundNode(t, nodeBin, hubAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, hubAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", @@ -83,18 +82,16 @@ func TestOpenAIFacadeCUJ(t *testing.T) { "--config", writeNodeConfig(t, homeA, map[string]string{"region": "eu"}, svcDecl{Type: "inference", Name: "test-llm", TargetURL: backend.URL}), ) t.Log("Starting Node B (consumer)...") - _ = startBackgroundNode(t, nodeBin, hubAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, hubAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", ) - apiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - apiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, apiAddrA) - waitForAPI(t, apiAddrB) + apiAddrA := nodeA.waitForAPI(t) + apiAddrB := nodeB.waitForAPI(t) - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr connectPeer(t, apiAddrB, addrA) waitForDHTPeers(t, apiAddrA) diff --git a/tests/integration/p2p_test.go b/tests/integration/p2p_test.go deleted file mode 100644 index ac8a89a8..00000000 --- a/tests/integration/p2p_test.go +++ /dev/null @@ -1,40 +0,0 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package integration_test - -import ( - "os" - "path/filepath" - "strings" - "testing" - "time" -) - -func TestSamNodeRunWithoutIdentity(t *testing.T) { - nodeBin := buildBinary(t, "./cmd/sam-node") - tmpHome := t.TempDir() - env := []string{ - "HOME=" + tmpHome, - "XDG_CONFIG_HOME=" + filepath.Join(tmpHome, ".config"), - } - stdout, stderr, err := runCommand(t, repoRoot(t), 10*time.Second, append(os.Environ(), env...), "", nodeBin, "run") - if err == nil { - t.Fatalf("expected sam-node run without identity to fail, but it succeeded\nstdout:\n%s\nstderr:\n%s", stdout, stderr) - } - out := stdout + stderr - if !strings.Contains(out, "No identity found. Starting unauthenticated sidecar for enrollment over MCP") { - t.Fatalf("expected missing identity message, got:\n%s", out) - } -} diff --git a/tests/integration/policy_permutations_test.go b/tests/integration/policy_permutations_test.go index 6ae838df..b7caeeeb 100644 --- a/tests/integration/policy_permutations_test.go +++ b/tests/integration/policy_permutations_test.go @@ -19,7 +19,6 @@ import ( "fmt" "net/http" "os" - "os/exec" "path/filepath" "strings" "testing" @@ -169,12 +168,10 @@ services: homeB := filepath.Join(tmpDir, "nodeB") apiTokenB := "tokenB" - apiPortB := getFreePort(t) - cmdB := exec.Command(nodeBin, "run", + nodeB := launchNode(t, nodeBin, os.Environ(), homeB, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", httpPortCP), "--data-dir", homeB, - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", apiPortB), "--api-token-path", tokenPath(t, apiTokenB), "--jwt", mintToken(map[string]interface{}{ "sub": "bob-subject", @@ -182,29 +179,12 @@ services: "email": "nodeB@example.com", "groups": []string{"compute", "backend"}, }), - "--listen", "/ip4/127.0.0.1/tcp/0", "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--allow-loopback", "--config", nodeBPolicyFile, ) - if err := os.MkdirAll(homeB, 0755); err != nil { - t.Fatal(err) - } - logFileB, err := os.Create(filepath.Join(homeB, "node.log")) - if err != nil { - t.Fatal(err) - } - defer func() { _ = logFileB.Close() }() - cmdB.Stdout = logFileB - cmdB.Stderr = logFileB - if err := cmdB.Start(); err != nil { - t.Fatalf("Failed to start Node B: %v", err) - } - defer func() { _ = cmdB.Process.Kill(); _ = cmdB.Wait() }() - - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, actualApiAddrB) - addrB := waitForPeerInfoInLog(t, filepath.Join(homeB, "node.log")) + nodeB.waitForAPI(t) + addrB := nodeB.p2pAddr // 3. Test Permutations tests := []struct { @@ -264,39 +244,20 @@ services: t.Run(tt.name, func(t *testing.T) { homeA := filepath.Join(tmpDir, fmt.Sprintf("nodeA_%d", i)) apiTokenA := "tokenA" - apiPortA := getFreePort(t) // The seat comes from the sam:system:authenticated binding above, // never from a roles claim naming sam:role:node. jwtA := mintToken(tt.jwtClaims) - cmdA := exec.Command(nodeBin, "run", + nodeA := launchNode(t, nodeBin, os.Environ(), homeA, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", httpPortCP), "--data-dir", homeA, - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", apiPortA), "--api-token-path", tokenPath(t, apiTokenA), "--jwt", jwtA, - "--listen", "/ip4/127.0.0.1/tcp/0", "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--allow-loopback", ) - if err := os.MkdirAll(homeA, 0755); err != nil { - t.Fatal(err) - } - logFileA, err := os.Create(filepath.Join(homeA, "node.log")) - if err != nil { - t.Fatal(err) - } - defer func() { _ = logFileA.Close() }() - cmdA.Stdout = logFileA - cmdA.Stderr = logFileA - if err := cmdA.Start(); err != nil { - t.Fatalf("Failed to start Node A: %v", err) - } - defer func() { _ = cmdA.Process.Kill(); _ = cmdA.Wait() }() - - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - waitForAPI(t, actualApiAddrA) + actualApiAddrA := nodeA.waitForAPI(t) parts := strings.Split(addrB, "/p2p/") if len(parts) != 2 { diff --git a/tests/integration/proxy_test.go b/tests/integration/proxy_test.go deleted file mode 100644 index 2ccf1d10..00000000 --- a/tests/integration/proxy_test.go +++ /dev/null @@ -1,101 +0,0 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package integration_test - -import ( - "context" - "crypto/ed25519" - "crypto/rand" - "encoding/json" - "os" - "path/filepath" - "strings" - "testing" - "time" - - "go.etcd.io/bbolt" -) - -func TestSamNodeRunWithStoredIdentity(t *testing.T) { - nodeBin := buildBinary(t, "./cmd/sam-node") - tmpHome := t.TempDir() - - // Pre-populate store - configDir := filepath.Join(tmpHome, ".config", "sam-mesh") - err := os.MkdirAll(configDir, 0700) - if err != nil { - t.Fatalf("failed to create config dir: %v", err) - } - - dbPath := filepath.Join(configDir, "agent.db") - db, err := bbolt.Open(dbPath, 0600, nil) - if err != nil { - t.Fatalf("failed to open db: %v", err) - } - err = db.Update(func(tx *bbolt.Tx) error { - b, err := tx.CreateBucketIfNotExists([]byte("identity")) - if err != nil { - return err - } - pub, priv, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - return err - } - biscuitBytes := createMockBiscuitToken(t, "12D3KooWRHDt3Ajd1t7YikBXyK2Uw1wrJwEX88XbwJhHh1XtBNkx", priv, "sam:role:node", nil) - if err := b.Put([]byte("identity_biscuit"), biscuitBytes); err != nil { - return err - } - if err := b.Put([]byte("control_plane_public_key"), pub); err != nil { - return err - } - addrsData, _ := json.Marshal([]string{"/ip4/127.0.0.1/tcp/4002/p2p/Qm..."}) - if err := b.Put([]byte("router_addresses"), addrsData); err != nil { - return err - } - return nil - }) - if err != nil { - t.Fatalf("failed to update db: %v", err) - } - _ = db.Close() - - env := append(os.Environ(), "HOME="+tmpHome) - - runOut, runErrOut, err := runCommand( - t, - repoRoot(t), - 3*time.Second, - env, - "", - nodeBin, - "run", "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", - "--bind-addr", "127.0.0.1:0", - "--api-token-path", tokenPath(t, "dummy-token"), - "--data-dir", configDir, - ) - if err != context.DeadlineExceeded { - t.Fatalf("expected run command to keep running until timeout, got: %v\nstdout:\n%s\nstderr:\n%s", err, runOut, runErrOut) - } - t.Logf("stdout:\n%s\nstderr:\n%s", runOut, runErrOut) - - out := runOut + runErrOut - if !strings.Contains(out, "Using stored identity.") { - t.Fatalf("run command did not use stored identity:\n%s", out) - } - if !strings.Contains(out, "SAM Node Online") { - t.Fatalf("node did not reach online state:\n%s", out) - } -} diff --git a/tests/integration/revocation_test.go b/tests/integration/revocation_test.go index 14d015cb..3767843b 100644 --- a/tests/integration/revocation_test.go +++ b/tests/integration/revocation_test.go @@ -21,7 +21,6 @@ import ( "os" "os/exec" "path/filepath" - "regexp" "strings" "testing" "time" @@ -113,97 +112,45 @@ roles: fetchPeerID(t, cpPort) // 4. Start Node 1 - node1ApiPort := getFreePort(t) node1Home := filepath.Join(tmpDir, "node1_home") - _ = os.MkdirAll(node1Home, 0755) - - logDir := filepath.Join(repoRoot(t), "tests/integration/logs") - _ = os.MkdirAll(logDir, 0755) - - node1LogPath := filepath.Join(logDir, "node1.log") - node1LogFile, _ := os.Create(node1LogPath) - defer func() { _ = node1LogFile.Close() }() - - node1Cmd := exec.Command(nodeBin, "run", + node1 := launchNode(t, nodeBin, + append(os.Environ(), "HOME="+node1Home, "XDG_CONFIG_HOME="+filepath.Join(node1Home, ".config")), + node1Home, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", cpPort), "--jwt", mintToken(map[string]interface{}{"sub": "mock-user", "roles": []string{api.RoleNode}}), "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", "--allow-loopback", - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", node1ApiPort), "--api-token-path", tokenPath(t, "node1-token"), "--log-level", "debug", ) - node1Cmd.Env = append(os.Environ(), "HOME="+node1Home, "XDG_CONFIG_HOME="+filepath.Join(node1Home, ".config")) - node1Cmd.Stdout = node1LogFile - node1Cmd.Stderr = node1LogFile - if err := node1Cmd.Start(); err != nil { - t.Fatalf("failed to start node 1: %v", err) - } - defer func() { - _ = node1Cmd.Process.Kill() - _ = node1Cmd.Wait() - }() // 5. Start Node 2 - node2ApiPort := getFreePort(t) node2Home := filepath.Join(tmpDir, "node2_home") - _ = os.MkdirAll(node2Home, 0755) - - node2LogPath := filepath.Join(logDir, "node2.log") - node2LogFile, _ := os.Create(node2LogPath) - defer func() { _ = node2LogFile.Close() }() - - node2Cmd := exec.Command(nodeBin, "run", + node2 := launchNode(t, nodeBin, + append(os.Environ(), "HOME="+node2Home, "XDG_CONFIG_HOME="+filepath.Join(node2Home, ".config")), + node2Home, "run", "--control-plane", fmt.Sprintf("http://127.0.0.1:%d", cpPort), "--jwt", mintToken(map[string]interface{}{"sub": "mock-user", "roles": []string{api.RoleNode}}), "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", - "--listen", "/ip4/127.0.0.1/tcp/0", "--allow-loopback", - "--bind-addr", fmt.Sprintf("127.0.0.1:%d", node2ApiPort), "--api-token-path", tokenPath(t, "node2-token"), "--log-level", "debug", ) - node2Cmd.Env = append(os.Environ(), "HOME="+node2Home, "XDG_CONFIG_HOME="+filepath.Join(node2Home, ".config")) - node2Cmd.Stdout = node2LogFile - node2Cmd.Stderr = node2LogFile - if err := node2Cmd.Start(); err != nil { - t.Fatalf("failed to start node 2: %v", err) - } - defer func() { - _ = node2Cmd.Process.Kill() - _ = node2Cmd.Wait() - }() - // Wait for nodes to go online actively - waitForNodeOnline(t, node1LogPath) - waitForNodeOnline(t, node2LogPath) + node1API := node1.waitForAPI(t) + node2.waitForAPI(t) - // Extract Node 2's PeerID and Address from its log - node2LogData, _ := os.ReadFile(node2LogPath) - rePeerID := regexp.MustCompile(`PeerID: (12D3Koo[a-zA-Z0-9]+)`) - matches := rePeerID.FindStringSubmatch(string(node2LogData)) - if len(matches) < 2 { - t.Fatalf("failed to find Node 2 peer ID in logs:\n%s", string(node2LogData)) - } - node2PeerID := matches[1] - - reAddr := regexp.MustCompile(`Listening on: \[(/ip4/127.0.0.1/tcp/\d+)`) - matchesAddr := reAddr.FindStringSubmatch(string(node2LogData)) - if len(matchesAddr) < 2 { - t.Fatalf("failed to find Node 2 listening TCP address in logs:\n%s", string(node2LogData)) - } - node2TCPAddr := matchesAddr[1] + node2AddrStr := node2.p2pAddr + node2PeerID := node2.peerID.String() // 6. Request Node 1 to connect to Node 2 - connectPeerWithToken(t, fmt.Sprintf("127.0.0.1:%d", node1ApiPort), "node1-token", - fmt.Sprintf("%s/p2p/%s", node2TCPAddr, node2PeerID)) + connectPeerWithToken(t, node1API, "node1-token", node2AddrStr) // Verify Node 1 is connected to Node 2 time.Sleep(1 * time.Second) stdout, stderr, err := runCommand(t, repoRoot(t), 5*time.Second, nil, "", clientBin, - "-url", fmt.Sprintf("http://127.0.0.1:%d/mcp", node1ApiPort), + "-url", "http://"+node1API+"/mcp", "-token", "node1-token", "-tool", "get_mesh_info", "-args", "{}", @@ -259,26 +206,11 @@ roles: } event.Signature = ed25519.Sign(cpPrivKey, eventData) - // Extract Node 1's PeerID and Address from its log - node1LogData, _ := os.ReadFile(node1LogPath) - matches1 := rePeerID.FindStringSubmatch(string(node1LogData)) - if len(matches1) < 2 { - t.Fatalf("failed to find Node 1 peer ID in logs:\n%s", string(node1LogData)) - } - node1PeerID := matches1[1] - - matchesAddr1 := reAddr.FindStringSubmatch(string(node1LogData)) - if len(matchesAddr1) < 2 { - t.Fatalf("failed to find Node 1 listening TCP address in logs:\n%s", string(node1LogData)) - } - node1TCPAddr := matchesAddr1[1] - // Publish Gossip event directly to Node 1 - node1AddrStr := fmt.Sprintf("%s/p2p/%s", node1TCPAddr, node1PeerID) - publishGossipEvent(t, node1AddrStr, event) + publishGossipEvent(t, node1.p2pAddr, event) // 8. Wait for revocation event to propagate and Node 1 to disconnect Node 2 actively - waitForNodeDisconnection(t, clientBin, node1ApiPort, node2PeerID) + waitForNodeDisconnection(t, clientBin, node1API, node2PeerID) } func publishGossipEvent(t *testing.T, routerAddrStr string, event *api.MeshEvent) { @@ -350,7 +282,7 @@ func publishGossipEvent(t *testing.T, routerAddrStr string, event *api.MeshEvent } } -func waitForNodeDisconnection(t *testing.T, clientBin string, node1ApiPort int, node2PeerID string) { +func waitForNodeDisconnection(t *testing.T, clientBin string, node1API string, node2PeerID string) { t.Helper() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -358,7 +290,7 @@ func waitForNodeDisconnection(t *testing.T, clientBin string, node1ApiPort int, for { stdout, _, err := runCommand(t, repoRoot(t), 2*time.Second, nil, "", clientBin, - "-url", fmt.Sprintf("http://127.0.0.1:%d/mcp", node1ApiPort), + "-url", "http://"+node1API+"/mcp", "-token", "node1-token", "-tool", "get_mesh_info", "-args", "{}", diff --git a/tests/integration/rotation_test.go b/tests/integration/rotation_test.go index a188900d..6681c600 100644 --- a/tests/integration/rotation_test.go +++ b/tests/integration/rotation_test.go @@ -148,7 +148,7 @@ roles: }() // Wait for Node to be online actively - waitForNodeOnline(t, nodeLogPath) + waitForAPI(t, fmt.Sprintf("127.0.0.1:%d", nodeApiPort)) // Get initial keys initialKeys := fetchPublicKeys(t, cpPort) diff --git a/tests/integration/sandbox_boundary_test.go b/tests/integration/sandbox_boundary_test.go index 782ea842..9f385c53 100644 --- a/tests/integration/sandbox_boundary_test.go +++ b/tests/integration/sandbox_boundary_test.go @@ -85,7 +85,7 @@ func TestSandboxBoundaryCUJ(t *testing.T) { defer external.Close() t.Log("Starting node A (provider) and node B (the gateway's node)...") - _ = startBackgroundNode(t, nodeBin, hubAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, hubAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", @@ -93,19 +93,17 @@ func TestSandboxBoundaryCUJ(t *testing.T) { svcDecl{Type: "inference", Name: "test-llm", TargetURL: inference.URL}, svcDecl{Type: "mcp", Name: "calc", TargetURL: tools.URL}), ) - _ = startBackgroundNode(t, nodeBin, hubAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, hubAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", "--socket-path", nodeSocket, ) - apiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - apiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - waitForAPI(t, apiAddrA) - waitForAPI(t, apiAddrB) + apiAddrA := nodeA.waitForAPI(t) + apiAddrB := nodeB.waitForAPI(t) - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr peerA := extractPeerID(addrA) connectPeer(t, apiAddrB, addrA) waitForDHTPeers(t, apiAddrA) diff --git a/tests/integration/service_discovery_test.go b/tests/integration/service_discovery_test.go index 796120a1..3f53d210 100644 --- a/tests/integration/service_discovery_test.go +++ b/tests/integration/service_discovery_test.go @@ -20,7 +20,6 @@ import ( "io" "net/http" "net/http/httptest" - "path/filepath" "strings" "testing" "time" @@ -48,34 +47,27 @@ func TestServiceDiscovery(t *testing.T) { // Start Node A t.Log("Starting Node A...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, routerAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), "--config", writeNodeConfig(t, homeA, nil, svcDecl{Type: "mcp", Name: serviceName, TargetURL: mockServer.URL}), ) // Start Node B t.Log("Starting Node B...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, routerAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), ) - // Resolve actual addresses from logs - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) + actualApiAddrA := nodeA.waitForAPI(t) + actualApiAddrB := nodeB.waitForAPI(t) - // Wait for nodes to start sidecar API - waitForAPI(t, actualApiAddrA) - waitForAPI(t, actualApiAddrB) - - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr // Connect Node B to Node A (to ensure they are in same network) // We use the multiplexed HTTP address for MCP calls too! @@ -149,34 +141,27 @@ func TestServiceDiscoveryStreaming(t *testing.T) { // Start Node A t.Log("Starting Node A...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeA, + nodeA := startBackgroundNode(t, nodeBin, routerAddr, homeA, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), "--config", writeNodeConfig(t, homeA, nil, svcDecl{Type: "mcp", Name: serviceName, TargetURL: mockServer.URL}), ) // Start Node B t.Log("Starting Node B...") - _ = startBackgroundNode(t, nodeBin, routerAddr, homeB, + nodeB := startBackgroundNode(t, nodeBin, routerAddr, homeB, "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", "--listen", "/ip4/127.0.0.1/tcp/0", "--discovery-interval", "100ms", - "--bind-addr", "127.0.0.1:0", "--api-token-path", tokenPath(t, apiToken), ) - // Resolve actual addresses from logs - actualApiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) - actualApiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) - - // Wait for nodes to start sidecar API - waitForAPI(t, actualApiAddrA) - waitForAPI(t, actualApiAddrB) + actualApiAddrA := nodeA.waitForAPI(t) + actualApiAddrB := nodeB.waitForAPI(t) - addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + addrA := nodeA.p2pAddr // Connect Node B to Node A connectPeer(t, actualApiAddrB, addrA) @@ -275,20 +260,6 @@ func TestServiceDiscoveryStreaming(t *testing.T) { } } -func waitForAPI(t *testing.T, addr string) { - t.Helper() - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - resp, err := http.Get("http://" + addr + "/healthz") - if err == nil && resp.StatusCode == http.StatusOK { - _ = resp.Body.Close() - return - } - time.Sleep(100 * time.Millisecond) - } - t.Fatalf("timeout waiting for API at %s", addr) -} - func discoverService(t *testing.T, apiAddr, token, serviceName string) []peer.AddrInfo { t.Helper() req, _ := http.NewRequest("GET", "http://"+apiAddr+"/sam/service/discover?type=mcp&name="+serviceName, nil)