diff --git a/apps/api/src/routes/nodes.ts b/apps/api/src/routes/nodes.ts index 2fa86c45e7..6aafe2775f 100644 --- a/apps/api/src/routes/nodes.ts +++ b/apps/api/src/routes/nodes.ts @@ -591,9 +591,26 @@ nodesRoutes.get('/:id/logs/stream', async (c) => { } } - // Proxy the WebSocket upgrade to the VM agent - const headers = new Headers(c.req.raw.headers); - headers.delete('x-sam-node-id'); + // Proxy only WebSocket handshake headers to the VM agent. Browser/control-plane + // credentials such as Cookie or Authorization must not be forwarded because + // VM-agent diagnostic auth gives Authorization precedence over the query token. + const clientHeaders = c.req.raw.headers; + const headers = new Headers(); + for (const name of [ + 'Upgrade', + 'Connection', + 'Sec-WebSocket-Key', + 'Sec-WebSocket-Version', + 'Sec-WebSocket-Protocol', + 'Sec-WebSocket-Extensions', + 'Origin', + ]) { + const value = clientHeaders.get(name); + if (value) { + headers.set(name, value); + } + } + headers.set('Authorization', `Bearer ${token}`); headers.set('X-SAM-Node-Id', nodeId); return fetchNodeAgent( diff --git a/apps/api/tests/unit/routes/node-observability-logs.test.ts b/apps/api/tests/unit/routes/node-observability-logs.test.ts index 54ce1e2e41..83732d4651 100644 --- a/apps/api/tests/unit/routes/node-observability-logs.test.ts +++ b/apps/api/tests/unit/routes/node-observability-logs.test.ts @@ -6,6 +6,9 @@ import type { Env } from '../../../src/env'; const mockRequireNodeOwnership = vi.fn(); const mockGetNodeLogsFromNode = vi.fn(); const mockListNodeContainersFromNode = vi.fn(); +const mockFetchNodeAgent = vi.fn(); +const mockGetNodeAgentRequestTimeoutMs = vi.fn(); +const mockSignNodeManagementToken = vi.fn(); vi.mock('../../../src/middleware/auth', () => ({ requireAuth: () => vi.fn((_c: any, next: any) => next()), @@ -18,6 +21,8 @@ vi.mock('../../../src/middleware/node-auth', () => ({ })); vi.mock('../../../src/services/node-agent', () => ({ + fetchNodeAgent: (...args: unknown[]) => mockFetchNodeAgent(...args), + getNodeAgentRequestTimeoutMs: (...args: unknown[]) => mockGetNodeAgentRequestTimeoutMs(...args), getNodeLogsFromNode: (...args: unknown[]) => mockGetNodeLogsFromNode(...args), listNodeContainersFromNode: (...args: unknown[]) => mockListNodeContainersFromNode(...args), getNodeSystemInfoFromNode: vi.fn(), @@ -40,7 +45,7 @@ vi.mock('../../../src/services/nodes', () => ({ })); vi.mock('../../../src/services/jwt', () => ({ - signNodeManagementToken: vi.fn(), + signNodeManagementToken: (...args: unknown[]) => mockSignNodeManagementToken(...args), })); vi.mock('../../../src/services/limits', () => ({ @@ -67,6 +72,12 @@ describe('node observability log routes', () => { beforeEach(() => { vi.clearAllMocks(); mockRequireNodeOwnership.mockResolvedValue({ id: 'node-1', status: 'running', userId: 'user-1' }); + mockGetNodeAgentRequestTimeoutMs.mockReturnValue(30_000); + mockSignNodeManagementToken.mockResolvedValue({ + token: 'node-management-token', + expiresAt: '2026-08-23T05:00:00.000Z', + }); + mockFetchNodeAgent.mockResolvedValue(new Response('proxied', { status: 200 })); }); it('returns docker container entries from the node agent proxy', async () => { @@ -105,4 +116,46 @@ describe('node observability log routes', () => { expect(body.containers).toHaveLength(1); expect(body.containers[0].name).toBe('web-1'); }); + + it('proxies log stream with node-management auth and strips client auth material', async () => { + const response = await createApp().request( + '/api/nodes/node-1/logs/stream?source=docker&level=debug&token=client-supplied-token', + { + headers: { + Authorization: 'Bearer user-api-token', + Cookie: 'better-auth.session_token=user-session', + Upgrade: 'websocket', + Connection: 'Upgrade', + 'Sec-WebSocket-Key': 'websocket-upgrade-key-placeholder', + 'Sec-WebSocket-Version': '13', + 'Sec-WebSocket-Protocol': 'sam.logs', + 'Sec-WebSocket-Extensions': 'permessage-deflate', + Origin: 'https://app.example.com', + }, + }, + { + BASE_DOMAIN: 'example.com', + VM_AGENT_PROTOCOL: 'https', + VM_AGENT_PORT: '8443', + } as Env, + ); + + expect(response.status).toBe(200); + expect(mockSignNodeManagementToken).toHaveBeenCalledWith('user-1', 'node-1', null, expect.anything()); + expect(mockFetchNodeAgent).toHaveBeenCalledTimes(1); + + const [, , vmUrl, init] = mockFetchNodeAgent.mock.calls[0]; + const parsedVmUrl = new URL(vmUrl as string); + expect(parsedVmUrl.pathname).toBe('/logs/stream'); + expect(parsedVmUrl.searchParams.get('token')).toBe('node-management-token'); + expect(parsedVmUrl.searchParams.get('source')).toBe('docker'); + expect(parsedVmUrl.searchParams.get('level')).toBe('debug'); + + const forwardedHeaders = (init as { headers: Headers }).headers; + expect(forwardedHeaders.get('Authorization')).toBe('Bearer node-management-token'); + expect(forwardedHeaders.get('Cookie')).toBeNull(); + expect(forwardedHeaders.get('X-SAM-Node-Id')).toBe('node-1'); + expect(forwardedHeaders.get('Upgrade')).toBe('websocket'); + expect(forwardedHeaders.get('Sec-WebSocket-Protocol')).toBe('sam.logs'); + }); }); diff --git a/apps/www/src/content/docs/docs/reference/vm-agent.md b/apps/www/src/content/docs/docs/reference/vm-agent.md index 2ce3caae9d..96051f867c 100644 --- a/apps/www/src/content/docs/docs/reference/vm-agent.md +++ b/apps/www/src/content/docs/docs/reference/vm-agent.md @@ -198,6 +198,8 @@ GET /containers The `/debug-package` endpoint bundles cloud-init logs, journald, Docker logs, system info, events/metrics databases, provisioning timings, and network config into a single downloadable archive — the fastest way to diagnose a node without SSH. +Node-wide diagnostics require a node-scoped management token issued by the control plane. Workspace browser sessions and workspace-scoped management tokens are not accepted for these routes because a single node can host multiple workspaces. User-facing node observability should go through the control-plane `/api/nodes/{nodeId}/...` proxy routes, which verify node ownership and sign the node-scoped token for the VM Agent. + ## Subsystems ### PTY Manager @@ -229,11 +231,12 @@ Responses are serialized via `orderedPipe` to prevent token reordering from conc ### JWT Validator -Validates workspace JWTs using the API's JWKS endpoint: +Validates workspace and node-management JWTs using the API's JWKS endpoint: - Fetches public keys from `/.well-known/jwks.json` - Caches keys with periodic refresh -- Extracts workspace ID and user ID from claims +- Enforces workspace claims on workspace-scoped routes +- Enforces node-scoped management tokens on node-wide diagnostic routes ## Configuration diff --git a/packages/vm-agent/internal/server/events.go b/packages/vm-agent/internal/server/events.go index 60601b717d..83a0aedf76 100644 --- a/packages/vm-agent/internal/server/events.go +++ b/packages/vm-agent/internal/server/events.go @@ -10,8 +10,6 @@ import ( ) func (s *Server) handleListNodeEvents(w http.ResponseWriter, r *http.Request) { - // Accept browser-facing auth: workspace request auth (any workspace on this node - // proves node ownership) or management token via Authorization header / ?token= query param. if !s.requireNodeEventAuth(w, r) { return } @@ -64,48 +62,60 @@ func (s *Server) handleListWorkspaceEvents(w http.ResponseWriter, r *http.Reques }) } -// requireNodeEventAuth authenticates node-level event requests. -// Accepts: -// 1. Node management token via Authorization header (control-plane proxy) -// 2. Node management token via ?token= query parameter (browser direct call) -// 3. Any valid workspace session cookie for a workspace on this node (browser) +// requireNodeEventAuth authenticates node-wide diagnostic requests. +// +// These routes expose node-wide observability state and raw diagnostic artifacts +// for every workspace on a node. They must therefore require a node-scoped +// management token minted by the control plane/operator. Workspace browser +// cookies and workspace-scoped management tokens are intentionally rejected. func (s *Server) requireNodeEventAuth(w http.ResponseWriter, r *http.Request) bool { - // Try management token from Authorization header first (existing pattern). + // Authorization takes precedence. A malformed/replayed bearer token must not + // silently fall through to any weaker credential on the same request. authHeader := strings.TrimSpace(r.Header.Get("Authorization")) - if authHeader != "" && strings.HasPrefix(authHeader, "Bearer ") { + if authHeader != "" { + if !strings.HasPrefix(authHeader, "Bearer ") { + writeError(w, http.StatusUnauthorized, "invalid Authorization header") + return false + } token := strings.TrimSpace(strings.TrimPrefix(authHeader, "Bearer ")) - if token != "" { - claims, err := s.jwtValidator.ValidateNodeManagementToken(token, "") - if err == nil { - routedNode := s.routedNodeID(r) - if routedNode == "" || routedNode == s.config.NodeID { - _ = claims - return true - } - } + if token == "" { + writeError(w, http.StatusUnauthorized, "missing bearer token") + return false } + return s.requireNodeScopedManagementToken(w, r, token) } - // Try management token from ?token= query parameter (browser direct call). + // WebSocket control-plane proxying uses ?token= because browsers cannot set + // custom Authorization headers during a WebSocket upgrade. queryToken := strings.TrimSpace(r.URL.Query().Get("token")) if queryToken != "" { - claims, err := s.jwtValidator.ValidateNodeManagementToken(queryToken, "") - if err == nil { - _ = claims - return true - } - } - - // Try workspace session cookie — any valid workspace session for this node proves access. - session := s.sessionManager.GetSessionFromRequest(r) - if session != nil && session.Claims != nil && session.Claims.Workspace != "" { - return true + return s.requireNodeScopedManagementToken(w, r, queryToken) } writeError(w, http.StatusUnauthorized, "authentication required") return false } +func (s *Server) requireNodeScopedManagementToken(w http.ResponseWriter, r *http.Request, token string) bool { + claims, err := s.jwtValidator.ValidateNodeManagementToken(token, "") + if err != nil { + writeError(w, http.StatusUnauthorized, "invalid management token") + return false + } + if claims.Workspace != "" { + writeError(w, http.StatusForbidden, "node-scoped management token required") + return false + } + + routedNode := s.routedNodeID(r) + if routedNode != "" && routedNode != s.config.NodeID { + writeError(w, http.StatusForbidden, "node route mismatch") + return false + } + + return true +} + func parseEventLimit(raw string) int { if raw == "" { return 100 diff --git a/packages/vm-agent/internal/server/events_test.go b/packages/vm-agent/internal/server/events_test.go index e1e6e6931a..a6be3ae33d 100644 --- a/packages/vm-agent/internal/server/events_test.go +++ b/packages/vm-agent/internal/server/events_test.go @@ -1,55 +1,613 @@ package server import ( - "os" - "path/filepath" + "archive/tar" + "bytes" + "compress/gzip" + "context" + "crypto/rand" + "crypto/rsa" + "encoding/base64" + "encoding/json" + "io" + "math/big" + "net/http" + "net/http/httptest" + "net/url" + "os/exec" "strings" "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/gorilla/websocket" + "github.com/workspace/vm-agent/internal/auth" + "github.com/workspace/vm-agent/internal/config" + "github.com/workspace/vm-agent/internal/eventstore" + "github.com/workspace/vm-agent/internal/logreader" + "github.com/workspace/vm-agent/internal/resourcemon" +) + +const ( + diagnosticAuthNodeID = "node-diagnostics-1" + workspaceAID = "ws-alpha" + workspaceBID = "ws-bravo" + workspaceBCanary = "CANARY_WS_BRAVO_NODE_DIAGNOSTIC_MARKER_12345" + workspaceACleanSignal = "workspace-alpha-visible-event" ) -func TestEventsSourceContract(t *testing.T) { - path := filepath.Join("events.go") - contentBytes, err := os.ReadFile(path) +type diagnosticAuthFixture struct { + server *Server + nodeToken string + workspaceMgmtA string + signingKey *rsa.PrivateKey + workspaceCookieA *http.Cookie + stop func() +} + +func newDiagnosticAuthFixture(t *testing.T) diagnosticAuthFixture { + t.Helper() + originalDebugPackageTimeout := debugPackageTimeout + originalDebugPackageLogLimit := debugPackageLogLimit + debugPackageTimeout = 2 * time.Second + debugPackageLogLimit = 10 + t.Cleanup(func() { + debugPackageTimeout = originalDebugPackageTimeout + debugPackageLogLimit = originalDebugPackageLogLimit + }) + + validator, signingKey := newServerJWTValidator(t, diagnosticAuthNodeID) + sessionManager := auth.NewSessionManagerWithConfig(auth.SessionManagerConfig{ + CookieName: "vm_session", + Secure: false, + TTL: time.Hour, + CleanupInterval: time.Hour, + MaxSessions: 10, + }) + + sessionA, err := sessionManager.CreateSession(&auth.Claims{ + RegisteredClaims: jwt.RegisteredClaims{Subject: "user-alpha"}, + Workspace: workspaceAID, + }) + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + + eventStore, err := eventstore.New(t.TempDir() + "/events.db") if err != nil { - t.Fatalf("read %s: %v", path, err) + t.Fatalf("eventstore.New: %v", err) } - content := string(contentBytes) + eventStore.Append(eventstore.EventRecord{ + ID: "ws-a-event", + NodeID: diagnosticAuthNodeID, + WorkspaceID: workspaceAID, + Level: "info", + Type: "workspace.alpha", + Message: workspaceACleanSignal, + Detail: map[string]interface{}{"workspace": workspaceAID}, + CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), + }) + eventStore.Append(eventstore.EventRecord{ + ID: "ws-b-event", + NodeID: diagnosticAuthNodeID, + WorkspaceID: workspaceBID, + Level: "error", + Type: "workspace.bravo.secret", + Message: "workspace B secret event " + workspaceBCanary, + Detail: map[string]interface{}{"secret": workspaceBCanary}, + CreatedAt: time.Now().UTC().Add(time.Second).Format(time.RFC3339Nano), + }) - for _, needle := range []string{ - "handleListNodeEvents", - "handleListWorkspaceEvents", - "nextCursor", - "parseEventLimit", - "requireNodeEventAuth", - } { - if !strings.Contains(content, needle) { - t.Fatalf("expected %q in %s", needle, path) + resourceMonitor, err := resourcemon.New(t.TempDir()+"/metrics.db", time.Hour) + if err != nil { + t.Fatalf("resourcemon.New: %v", err) + } + + fakeLogs := logreader.NewReaderWithExecutor(func(ctx context.Context, name string, args ...string) *exec.Cmd { + if name == "docker" && len(args) > 0 && args[0] == "ps" { + return exec.CommandContext(ctx, "printf", "%s", `{"ID":"bravo12345678","Names":"ws-bravo-devcontainer","Image":"repo:latest","State":"running","Status":"Up 1 minute"}`+"\n") + } + if name == "docker" && len(args) > 0 && args[0] == "logs" { + line := "2026-08-23T12:00:00Z workspace B log leaked " + workspaceBCanary + "\n" + return exec.CommandContext(ctx, "printf", "%s", line) } + return exec.CommandContext(ctx, "printf", "") + }) + + srv := &Server{ + config: &config.Config{ + NodeID: diagnosticAuthNodeID, + WSReadBufferSize: 1024, + WSWriteBufferSize: 1024, + LogStreamPingInterval: time.Hour, + LogStreamPongTimeout: time.Hour, + LogStreamPingWriteTimeout: time.Second, + }, + jwtValidator: validator, + sessionManager: sessionManager, + nodeEvents: []EventRecord{{ID: "node-event", NodeID: diagnosticAuthNodeID, WorkspaceID: workspaceBID, Level: "error", Type: "node.secret", Message: "node event includes " + workspaceBCanary, Detail: map[string]interface{}{"secret": workspaceBCanary}, CreatedAt: time.Now().UTC().Format(time.RFC3339Nano)}}, + workspaceEvents: map[string][]EventRecord{workspaceAID: {{ID: "workspace-a", WorkspaceID: workspaceAID, Level: "info", Type: "workspace.alpha", Message: workspaceACleanSignal, CreatedAt: time.Now().UTC().Format(time.RFC3339Nano)}}, workspaceBID: {{ID: "workspace-b", WorkspaceID: workspaceBID, Level: "error", Type: "workspace.bravo", Message: workspaceBCanary, CreatedAt: time.Now().UTC().Format(time.RFC3339Nano)}}}, + eventStore: eventStore, + resourceMonitor: resourceMonitor, + logReader: fakeLogs, + } + + return diagnosticAuthFixture{ + server: srv, + nodeToken: signServerTestToken(t, signingKey, diagnosticAuthNodeID, ""), + workspaceMgmtA: signServerTestToken(t, signingKey, diagnosticAuthNodeID, workspaceAID), + signingKey: signingKey, + workspaceCookieA: &http.Cookie{Name: "vm_session", Value: sessionA.ID}, + stop: func() { + _ = eventStore.Close() + _ = resourceMonitor.Close() + sessionManager.Stop() + validator.Close() + }, } } -func TestEventsHandlersAcceptBrowserAuth(t *testing.T) { - path := filepath.Join("events.go") - contentBytes, err := os.ReadFile(path) +func TestNodeWideDiagnosticsRejectWorkspaceAuthAcrossAffectedEndpoints(t *testing.T) { + fx := newDiagnosticAuthFixture(t) + defer fx.stop() + + tests := []struct { + name string + path string + handler http.HandlerFunc + }{ + {name: "events", path: "/events", handler: fx.server.handleListNodeEvents}, + {name: "events export", path: "/events/export", handler: fx.server.handleExportEvents}, + {name: "metrics export", path: "/metrics/export", handler: fx.server.handleExportMetrics}, + {name: "logs", path: "/logs?source=docker&level=debug", handler: fx.server.handleLogs}, + {name: "containers", path: "/containers", handler: fx.server.handleContainers}, + {name: "debug package", path: "/debug-package", handler: fx.server.handleDebugPackage}, + {name: "system info", path: "/system-info", handler: fx.server.handleSystemInfo}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, tt.path, nil) + req.AddCookie(fx.workspaceCookieA) + rec := httptest.NewRecorder() + + tt.handler.ServeHTTP(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusUnauthorized, rec.Body.String()) + } + assertBodyDoesNotContainCanary(t, rec.Body.Bytes()) + if strings.Contains(rec.Header().Get("Content-Type"), "gzip") { + t.Fatalf("unauthorized %s response must not start a debug archive", tt.name) + } + if tt.name == "debug package" { + assertNotDebugArchive(t, rec) + } + }) + } +} + +func TestNodeWideLogStreamRejectsWorkspaceAuthBeforeUpgrade(t *testing.T) { + fx := newDiagnosticAuthFixture(t) + defer fx.stop() + + ts := httptest.NewServer(http.HandlerFunc(fx.server.handleLogStream)) + defer ts.Close() + + wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/logs/stream?source=docker&level=debug" + headers := http.Header{} + headers.Add("Cookie", fx.workspaceCookieA.String()) + + conn, resp, err := websocket.DefaultDialer.Dial(wsURL, headers) + if conn != nil { + _ = conn.Close() + } + if err == nil { + t.Fatal("expected WebSocket upgrade to fail under workspace cookie auth") + } + if resp == nil { + t.Fatalf("expected HTTP response for failed upgrade: %v", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("status = %d, want %d; body=%s", resp.StatusCode, http.StatusUnauthorized, string(body)) + } + assertBodyDoesNotContainCanary(t, body) +} + +func TestNodeWideLogStreamAcceptsNodeScopedManagementToken(t *testing.T) { + fx := newDiagnosticAuthFixture(t) + defer fx.stop() + + ts := httptest.NewServer(http.HandlerFunc(fx.server.handleLogStream)) + defer ts.Close() + + wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/logs/stream?source=docker&level=debug&token=" + url.QueryEscape(fx.nodeToken) + headers := http.Header{} + headers.Set("X-SAM-Node-Id", diagnosticAuthNodeID) + + conn, resp, err := websocket.DefaultDialer.Dial(wsURL, headers) if err != nil { - t.Fatalf("read %s: %v", path, err) + if resp != nil && resp.Body != nil { + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + t.Fatalf("expected WebSocket upgrade to succeed, status=%d body=%s err=%v", resp.StatusCode, string(body), err) + } + t.Fatalf("expected WebSocket upgrade to succeed: %v", err) + } + defer conn.Close() + + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("read first log stream message: %v", err) + } + if !bytes.Contains(msg, []byte(workspaceBCanary)) { + t.Fatalf("management-token log stream did not include expected node observability data: %s", string(msg)) + } +} + +func TestNodeWideDiagnosticsAcceptNodeScopedManagementToken(t *testing.T) { + fx := newDiagnosticAuthFixture(t) + defer fx.stop() + + tests := []nodeWideManagementTokenCase{ + {name: "events", path: "/events", handler: fx.server.handleListNodeEvents, wantCT: "application/json", wantBody: workspaceBCanary}, + {name: "events export", path: "/events/export", handler: fx.server.handleExportEvents, wantCT: "application/x-sqlite3", wantBody: workspaceBCanary}, + {name: "metrics export", path: "/metrics/export", handler: fx.server.handleExportMetrics, wantCT: "application/x-sqlite3", readLimit: 4096}, + {name: "logs", path: "/logs?source=docker&level=debug", handler: fx.server.handleLogs, wantCT: "application/json", wantBody: workspaceBCanary}, + {name: "containers", path: "/containers", handler: fx.server.handleContainers, wantCT: "application/json", wantBody: "ws-bravo-devcontainer"}, + {name: "debug package via query token", path: "/debug-package", handler: fx.server.handleDebugPackage, useQuery: true, wantCT: "application/gzip", wantBody: workspaceBCanary}, } - content := string(contentBytes) - // Workspace events should accept browser workspace auth (same pattern as tabs). - // Uses checkWorkspaceRequestAuth (non-writing variant) to avoid double-write - // when falling through to management auth. - if !strings.Contains(content, "checkWorkspaceRequestAuth") { - t.Fatal("workspace events handler must accept workspace request auth for browser direct access") + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assertNodeWideManagementTokenAccepted(t, fx, tt) + }) } +} + +func TestWorkspaceScopedEventsRemainScopedAndRedactedFromOtherWorkspaceCanaries(t *testing.T) { + fx := newDiagnosticAuthFixture(t) + defer fx.stop() + + req := httptest.NewRequest(http.MethodGet, "/workspaces/"+workspaceAID+"/events", nil) + req.SetPathValue("workspaceId", workspaceAID) + req.Header.Set("X-SAM-Workspace-Id", workspaceAID) + req.AddCookie(fx.workspaceCookieA) + rec := httptest.NewRecorder() - // Node events should accept management token via query param - if !strings.Contains(content, "r.URL.Query().Get(\"token\")") { - t.Fatal("node events handler must accept token via query parameter for browser direct access") + fx.server.handleListWorkspaceEvents(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + body := rec.Body.String() + if !strings.Contains(body, workspaceACleanSignal) { + t.Fatalf("workspace-scoped response missing workspace A event: %s", body) } + if strings.Contains(body, workspaceBCanary) || strings.Contains(body, workspaceBID) { + t.Fatalf("workspace-scoped response leaked workspace B data: %s", body) + } +} - // Node events should accept workspace session cookie - if !strings.Contains(content, "sessionManager.GetSessionFromRequest") { - t.Fatal("node events handler must accept workspace session cookie for browser direct access") +func TestNodeWideDiagnosticsRejectWorkspaceScopedAndInvalidTokensWithoutCookieFallback(t *testing.T) { + fx := newDiagnosticAuthFixture(t) + defer fx.stop() + + t.Run("workspace scoped management token cannot be replayed against node-wide route", func(t *testing.T) { + assertNodeEventsDenied(t, fx, "Bearer "+fx.workspaceMgmtA, http.StatusForbidden) + }) + + t.Run("invalid bearer is not bypassed by valid query token or workspace cookie", func(t *testing.T) { + assertInvalidBearerBlocksEventFallback(t, fx) + }) + + t.Run("invalid query token is not bypassed by workspace cookie for debug archive", func(t *testing.T) { + assertDebugPackageQueryTokenDenied(t, fx, "not-a-valid-jwt") + }) + + t.Run("expired query token is not bypassed by workspace cookie for debug archive", func(t *testing.T) { + expired := signServerTestTokenWithExpiry(t, fx.signingKey, diagnosticAuthNodeID, "", time.Now().Add(-time.Minute)) + assertDebugPackageQueryTokenDenied(t, fx, expired) + }) + + t.Run("invalid query token is not bypassed by workspace cookie for websocket", func(t *testing.T) { + assertLogStreamQueryTokenDenied(t, fx, "not-a-valid-jwt") + }) + + t.Run("expired query token is not bypassed by workspace cookie for websocket", func(t *testing.T) { + expired := signServerTestTokenWithExpiry(t, fx.signingKey, diagnosticAuthNodeID, "", time.Now().Add(-time.Minute)) + assertLogStreamQueryTokenDenied(t, fx, expired) + }) + + t.Run("valid bearer takes precedence over invalid query token", func(t *testing.T) { + assertValidBearerOverridesInvalidQuery(t, fx) + }) + + t.Run("expired node token is rejected", func(t *testing.T) { + expired := signServerTestTokenWithExpiry(t, fx.signingKey, diagnosticAuthNodeID, "", time.Now().Add(-time.Minute)) + assertExpiredNodeBearerDenied(t, fx, expired) + }) +} + +type nodeWideManagementTokenCase struct { + name string + path string + handler http.HandlerFunc + useQuery bool + wantCT string + wantBody string + readLimit int64 +} + +func assertNodeWideManagementTokenAccepted(t *testing.T, fx diagnosticAuthFixture, tt nodeWideManagementTokenCase) { + t.Helper() + rec := serveNodeWideManagementTokenRequest(t, fx, tt) + assertStatus(t, rec.Code, http.StatusOK, rec.Body.String()) + if got := rec.Header().Get("Content-Type"); !strings.Contains(got, tt.wantCT) { + t.Fatalf("Content-Type = %q, want containing %q", got, tt.wantCT) + } + assertExpectedManagementBody(t, rec.Body.Bytes(), tt) +} + +func serveNodeWideManagementTokenRequest(t *testing.T, fx diagnosticAuthFixture, tt nodeWideManagementTokenCase) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodGet, pathWithOptionalToken(t, tt.path, tt.useQuery, fx.nodeToken), nil) + if !tt.useQuery { + req.Header.Set("Authorization", "Bearer "+fx.nodeToken) + } + req.Header.Set("X-SAM-Node-Id", diagnosticAuthNodeID) + rec := httptest.NewRecorder() + tt.handler.ServeHTTP(rec, req) + return rec +} + +func pathWithOptionalToken(t *testing.T, path string, useQuery bool, token string) string { + t.Helper() + if !useQuery { + return path + } + parsed, err := url.Parse(path) + if err != nil { + t.Fatalf("parse test path: %v", err) + } + q := parsed.Query() + q.Set("token", token) + parsed.RawQuery = q.Encode() + return parsed.String() +} + +func assertExpectedManagementBody(t *testing.T, responseBody []byte, tt nodeWideManagementTokenCase) { + t.Helper() + if tt.wantBody == "" { + return + } + body := limitedBody(responseBody, tt.readLimit) + if strings.Contains(tt.wantCT, "gzip") { + body = mustReadTarGz(t, responseBody) + } + if !bytes.Contains(body, []byte(tt.wantBody)) { + t.Fatalf("management-token response did not include expected observability data %q", tt.wantBody) + } +} + +func limitedBody(body []byte, limit int64) []byte { + if limit <= 0 || int64(len(body)) <= limit { + return body + } + return body[:limit] +} + +func assertNodeEventsDenied(t *testing.T, fx diagnosticAuthFixture, authorization string, wantStatus int) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/events", nil) + req.Header.Set("Authorization", authorization) + rec := httptest.NewRecorder() + fx.server.handleListNodeEvents(rec, req) + assertStatus(t, rec.Code, wantStatus, rec.Body.String()) + assertBodyDoesNotContainCanary(t, rec.Body.Bytes()) +} + +func assertInvalidBearerBlocksEventFallback(t *testing.T, fx diagnosticAuthFixture) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/events?token="+url.QueryEscape(fx.nodeToken), nil) + req.Header.Set("Authorization", "Bearer not-a-valid-jwt") + req.AddCookie(fx.workspaceCookieA) + rec := httptest.NewRecorder() + fx.server.handleListNodeEvents(rec, req) + assertStatus(t, rec.Code, http.StatusUnauthorized, rec.Body.String()) + assertBodyDoesNotContainCanary(t, rec.Body.Bytes()) +} + +func assertDebugPackageQueryTokenDenied(t *testing.T, fx diagnosticAuthFixture, token string) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/debug-package?token="+url.QueryEscape(token), nil) + req.AddCookie(fx.workspaceCookieA) + rec := httptest.NewRecorder() + fx.server.handleDebugPackage(rec, req) + assertStatus(t, rec.Code, http.StatusUnauthorized, rec.Body.String()) + assertBodyDoesNotContainCanary(t, rec.Body.Bytes()) + assertNotDebugArchive(t, rec) +} + +func assertLogStreamQueryTokenDenied(t *testing.T, fx diagnosticAuthFixture, token string) { + t.Helper() + ts := httptest.NewServer(http.HandlerFunc(fx.server.handleLogStream)) + defer ts.Close() + + body, status := dialRejectedLogStream(t, ts.URL, token, fx.workspaceCookieA.String()) + assertStatus(t, status, http.StatusUnauthorized, string(body)) + assertBodyDoesNotContainCanary(t, body) +} + +func dialRejectedLogStream(t *testing.T, serverURL string, token string, cookie string) ([]byte, int) { + t.Helper() + wsURL := "ws" + strings.TrimPrefix(serverURL, "http") + "/logs/stream?source=docker&level=debug&token=" + url.QueryEscape(token) + headers := http.Header{} + headers.Add("Cookie", cookie) + + conn, resp, err := websocket.DefaultDialer.Dial(wsURL, headers) + if conn != nil { + _ = conn.Close() + } + if err == nil { + t.Fatal("expected WebSocket upgrade to fail") + } + if resp == nil { + t.Fatalf("expected HTTP response for failed upgrade: %v", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return body, resp.StatusCode +} + +func assertValidBearerOverridesInvalidQuery(t *testing.T, fx diagnosticAuthFixture) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/logs?source=docker&level=debug&token=not-a-valid-jwt", nil) + req.Header.Set("Authorization", "Bearer "+fx.nodeToken) + req.AddCookie(fx.workspaceCookieA) + rec := httptest.NewRecorder() + fx.server.handleLogs(rec, req) + assertStatus(t, rec.Code, http.StatusOK, rec.Body.String()) + if !bytes.Contains(rec.Body.Bytes(), []byte(workspaceBCanary)) { + t.Fatalf("valid bearer response did not include expected node observability data: %s", rec.Body.String()) + } +} + +func assertExpiredNodeBearerDenied(t *testing.T, fx diagnosticAuthFixture, token string) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/debug-package", nil) + req.Header.Set("Authorization", "Bearer "+token) + req.AddCookie(fx.workspaceCookieA) + rec := httptest.NewRecorder() + fx.server.handleDebugPackage(rec, req) + assertStatus(t, rec.Code, http.StatusUnauthorized, rec.Body.String()) + assertBodyDoesNotContainCanary(t, rec.Body.Bytes()) + assertNotDebugArchive(t, rec) +} + +func assertStatus(t *testing.T, got int, want int, body string) { + t.Helper() + if got != want { + t.Fatalf("status = %d, want %d; body=%s", got, want, body) + } +} + +func assertBodyDoesNotContainCanary(t *testing.T, body []byte) { + t.Helper() + if bytes.Contains(body, []byte(workspaceBCanary)) || bytes.Contains(body, []byte(workspaceBID)) { + t.Fatalf("response leaked workspace B canary/body=%q", string(body)) + } +} + +func assertNotDebugArchive(t *testing.T, rec *httptest.ResponseRecorder) { + t.Helper() + if strings.Contains(rec.Header().Get("Content-Type"), "gzip") { + t.Fatal("denied debug-package response must not use gzip content type") + } + if rec.Header().Get("Content-Disposition") != "" { + t.Fatalf("denied debug-package response must not use attachment disposition: %q", rec.Header().Get("Content-Disposition")) + } + body := rec.Body.Bytes() + if len(body) >= 2 && body[0] == 0x1f && body[1] == 0x8b { + t.Fatal("denied debug-package response must not contain gzip magic bytes") + } +} + +func mustReadTarGz(t *testing.T, data []byte) []byte { + t.Helper() + gr, err := gzip.NewReader(bytes.NewReader(data)) + if err != nil { + t.Fatalf("gzip.NewReader: %v", err) + } + defer gr.Close() + + tr := tar.NewReader(gr) + var out bytes.Buffer + for { + hdr, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("tar.Next: %v", err) + } + out.WriteString(hdr.Name) + out.WriteByte('\n') + if _, err := io.Copy(&out, tr); err != nil { + t.Fatalf("read tar member %s: %v", hdr.Name, err) + } + out.WriteByte('\n') + } + return out.Bytes() +} + +func newServerJWTValidator(t *testing.T, nodeID string) (*auth.JWTValidator, *rsa.PrivateKey) { + t.Helper() + privateKey, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatalf("generate RSA key: %v", err) + } + pubKey := privateKey.Public().(*rsa.PublicKey) + jwksJSON := buildServerTestJWKSJSON(pubKey) + + jwksServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(jwksJSON) + })) + t.Cleanup(jwksServer.Close) + + validator, err := auth.NewJWTValidator(jwksServer.URL, nodeID, "test-issuer", "workspace-terminal") + if err != nil { + t.Fatalf("auth.NewJWTValidator: %v", err) + } + return validator, privateKey +} + +func buildServerTestJWKSJSON(pub *rsa.PublicKey) []byte { + n := base64.RawURLEncoding.EncodeToString(pub.N.Bytes()) + e := base64.RawURLEncoding.EncodeToString(big.NewInt(int64(pub.E)).Bytes()) + data, _ := json.Marshal(map[string]interface{}{ + "keys": []map[string]interface{}{{ + "kty": "RSA", + "alg": "RS256", + "use": "sig", + "kid": "diagnostic-test-key", + "n": n, + "e": e, + }}, + }) + return data +} + +func signServerTestToken(t *testing.T, key *rsa.PrivateKey, nodeID, workspaceID string) string { + t.Helper() + return signServerTestTokenWithExpiry(t, key, nodeID, workspaceID, time.Now().Add(time.Hour)) +} + +func signServerTestTokenWithExpiry(t *testing.T, key *rsa.PrivateKey, nodeID, workspaceID string, expiresAt time.Time) string { + t.Helper() + claims := auth.Claims{ + RegisteredClaims: jwt.RegisteredClaims{ + Issuer: "test-issuer", + Subject: "user-alpha", + Audience: jwt.ClaimStrings{"node-management"}, + ExpiresAt: jwt.NewNumericDate(expiresAt), + IssuedAt: jwt.NewNumericDate(time.Now()), + }, + Node: nodeID, + Type: "node-management", + } + if workspaceID != "" { + claims.Workspace = workspaceID + } + token := jwt.NewWithClaims(jwt.SigningMethodRS256, claims) + token.Header["kid"] = "diagnostic-test-key" + signed, err := token.SignedString(key) + if err != nil { + t.Fatalf("sign token: %v", err) } + return signed } diff --git a/specs/014-multi-workspace-nodes/contracts/node-agent-api.md b/specs/014-multi-workspace-nodes/contracts/node-agent-api.md index 8be5444d6d..31461a0300 100644 --- a/specs/014-multi-workspace-nodes/contracts/node-agent-api.md +++ b/specs/014-multi-workspace-nodes/contracts/node-agent-api.md @@ -11,6 +11,7 @@ - `X-SAM-Node-Id` is required for all Node Agent management/observability requests. - `X-SAM-Workspace-Id` is required for workspace-scoped routes. - Header trust rule: Node Agent only trusts `X-SAM-*` headers from Control Plane ingress; direct client traffic is not an authoritative source. +- Node-wide diagnostic routes require node-scoped management JWTs with no workspace claim. Workspace browser sessions and workspace-scoped management JWTs cannot authorize node-wide raw diagnostics on multi-workspace nodes. ## Node-Level Endpoints @@ -46,6 +47,7 @@ - `GET /workspaces/{workspaceId}/events` - Returns recent workspace-scoped events/log entries. - Query: `limit`, `cursor`. +- May also be reached with a valid workspace session for the matching workspace; responses remain scoped to that workspace. ## Agent Session Endpoints diff --git a/specs/020-node-observability/contracts/vm-agent-logs.md b/specs/020-node-observability/contracts/vm-agent-logs.md index 41b22a7e74..6c10ff2dbb 100644 --- a/specs/020-node-observability/contracts/vm-agent-logs.md +++ b/specs/020-node-observability/contracts/vm-agent-logs.md @@ -5,10 +5,12 @@ ## Authentication -All log endpoints use the same authentication as existing node-level endpoints (`requireNodeEventAuth`): -1. Node management JWT via `Authorization: Bearer ` header (control plane proxy) -2. Node management JWT via `?token=` query parameter (browser direct) -3. Any valid workspace session cookie for a workspace on this node (browser) +Log endpoints are node-wide diagnostics and require a node-scoped management JWT: + +1. `Authorization: Bearer ` header for control-plane proxy requests. +2. `?token=` query parameter for WebSocket upgrade requests that cannot set custom headers. + +Workspace browser session cookies and workspace-scoped management tokens are rejected. Browser/user-facing log access goes through the control-plane `/api/nodes/:nodeId/...` proxy routes, which authenticate the user, verify node ownership, and mint a node-scoped management token for the VM Agent. --- @@ -80,7 +82,7 @@ Real-time log streaming via WebSocket. Delivers new log entries as they are gene ### Connection -WebSocket upgrade at `GET /logs/stream`. Authentication via `?token=` query parameter. +WebSocket upgrade at `GET /logs/stream`. Authentication via `?token=` query parameter. ### Query Parameters (set at connection time) @@ -89,7 +91,7 @@ WebSocket upgrade at `GET /logs/stream`. Authentication via `?token=