From 887b6287e9702bfdda136b1a6a6b06943babf45d Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Sun, 20 Sep 2026 19:41:09 +0200 Subject: [PATCH 1/2] tests: know a node's identity before it starts Integration tests learned a node's peer ID and listen address by asking its debug endpoints and decoding the JSON into structs kept in the test package, a copy of the node's unexported types that would drift with them. The information was never the node's to give: the test picks the API and libp2p ports, and the peer ID follows from the key in the node's store, which the test can generate with the same node.GetOrGenerateKey the binary uses. launchNode does that. It creates the key, closes the store again so the node can take the bbolt lock, appends the chosen --bind-addr and --listen, and hands back a backgroundNode whose apiAddr, peerID and p2pAddr are plain fields. waitForAPI polls /healthz, which the node only serves once Start succeeded, and fails with the log if the process exits first. The seventeen hand-rolled exec.Command blocks for nodes go, and so do the tests' own JSON mirrors of internal types. --- tests/integration/a2a_test.go | 13 +- tests/integration/agent_ingress_test.go | 12 +- tests/integration/agent_policy_test.go | 12 +- tests/integration/catalog_test.go | 107 +------- tests/integration/datapath_test.go | 36 +-- tests/integration/debug_endpoints_test.go | 18 +- tests/integration/identity_evidence_test.go | 12 +- tests/integration/local_policy_test.go | 97 +------ tests/integration/minimal_helpers_test.go | 18 -- tests/integration/node_helpers_test.go | 259 ++++++++++++++++++ tests/integration/openai_facade_test.go | 13 +- tests/integration/policy_permutations_test.go | 49 +--- tests/integration/revocation_test.go | 100 ++----- tests/integration/rotation_test.go | 2 +- tests/integration/sandbox_boundary_test.go | 12 +- tests/integration/service_discovery_test.go | 49 +--- 16 files changed, 356 insertions(+), 453 deletions(-) create mode 100644 tests/integration/node_helpers_test.go 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/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/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/minimal_helpers_test.go b/tests/integration/minimal_helpers_test.go index d2fe2dff..75cf7ec4 100644 --- a/tests/integration/minimal_helpers_test.go +++ b/tests/integration/minimal_helpers_test.go @@ -682,24 +682,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/node_helpers_test.go b/tests/integration/node_helpers_test.go new file mode 100644 index 00000000..c66c9622 --- /dev/null +++ b/tests/integration/node_helpers_test.go @@ -0,0 +1,259 @@ +// 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 ( + "fmt" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/google/sam/internal/node" + "github.com/libp2p/go-libp2p/core/peer" +) + +// 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): + } +} + +// 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/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/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) From edb1c1739f962e4caa6dfb488e88c9989285c0b2 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Sun, 20 Sep 2026 20:43:20 +0200 Subject: [PATCH 2/2] tests: watch the mesh, not the node's log The node lifecycle tests read the node's stdout for sentences: "Using stored identity.", "SAM Node Online", "Successfully reconnected to router via HTTP fallback". A reworded log line failed them while the behaviour stood, and a node that printed the line and died a moment later passed them. They also ran against mock routers that handed out a biscuit to anyone, so the enrollment they exercised was never judged. Every mesh now is a real control plane and a real router, and what the tests assert is what those two record: - /admin/status, decoded into storage.EnrolledNode and storage.RouterLease, says who enrolled which peer and which router each peer is connected to. Routers renew their lease every second in tests so the second is current. - The node's own store, opened after the node has stopped, says what it will do on its next start: its biscuit, its control plane, its router set. - /healthz answers only after Start succeeded, which needs a router to have authenticated the node; a node with no identity is known by the token-less MCP session it accepts. The mock OIDC provider gains the authorization code grant with PKCE so `sam-node join` signs in through it and the control plane verifies the result against the JWKS. The join, restart, mismatch, reset, no-identity and self-healing tests are rewritten on that, and the failover test watches the router leases on both control planes instead of the node's AutoRelay log. The mock router that spoke OIDC goes. Binaries are now built once per test process. Go caches compilation, but every go build -o links, and fifty tests linking three binaries each was a minute of nothing. The Help tests drop from ~2s to ~0.2s. --- tests/integration/cuj_test.go | 56 --- tests/integration/failover_test.go | 58 +-- tests/integration/fallback_test.go | 308 ++++---------- tests/integration/login_test.go | 489 +++++++++++----------- tests/integration/minimal_helpers_test.go | 332 ++++++++------- tests/integration/mock_oidc_test.go | 106 ++++- tests/integration/node_helpers_test.go | 54 +++ tests/integration/p2p_test.go | 40 -- tests/integration/proxy_test.go | 101 ----- 9 files changed, 672 insertions(+), 872 deletions(-) delete mode 100644 tests/integration/cuj_test.go delete mode 100644 tests/integration/p2p_test.go delete mode 100644 tests/integration/proxy_test.go 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/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/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 75cf7ec4..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() 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 index c66c9622..58ebd2dc 100644 --- a/tests/integration/node_helpers_test.go +++ b/tests/integration/node_helpers_test.go @@ -15,6 +15,7 @@ package integration_test import ( + "context" "fmt" "net/http" "os" @@ -26,6 +27,7 @@ import ( "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 @@ -240,6 +242,58 @@ func (n *backgroundNode) staysUp(t *testing.T, d time.Duration) { } } +// 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() 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/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) - } -}