package gateway import ( "context" "errors" "fmt" "io" "log/slog" "net/http" "net/http/httptest" "strings" "sync" "sync/atomic" "testing" "github.com/gin-gonic/gin" agsserver "github.com/pingcap/agent-git-service/server" "github.com/stretchr/testify/require" repositoryroute "github.com/pingcap/agent-git-service/repositoryroutecontract" "github.com/tidbcloud/AgentTeam/internal/api" "github.com/tidbcloud/AgentTeam/internal/api/identity/lookup" "github.com/tidbcloud/AgentTeam/internal/store/identity" "github.com/tidbcloud/AgentTeam/internal/store/storeerr" "github.com/tidbcloud/AgentTeam/internal/store/workspace" "github.com/tidbcloud/AgentTeam/internal/workspace/router" ) func TestHandlerCentralExceptionOutranksAndStripsRouteHeader(t *testing.T) { t.Parallel() central := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/auth/session", request.URL.Path) require.Empty(t, request.Header.Get(api.HeaderRouteWorkspaceID)) _, _ = writer.Write([]byte("central-auth")) })) t.Cleanup(central.Close) handler := newTestHandler(t, central.URL, map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{}) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, "/api/v1/auth/session", nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_1") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, "central-auth", recorder.Body.String()) } func TestHandlerRoutesCentralWithoutWorkspaceHeader(t *testing.T) { t.Parallel() central := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces", request.URL.Path) require.Empty(t, request.Header.Get(api.HeaderRouteWorkspaceID)) _, _ = writer.Write([]byte("central")) })) t.Cleanup(central.Close) handler := newTestHandler(t, central.URL, map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{}) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/workspaces", nil)) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, "central", recorder.Body.String()) } func TestHandlerCentralOwnerIgnoresPlacementHeadersAndPreservesBusinessWorkspace(t *testing.T) { t.Parallel() central := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/billing/summary", request.URL.Path) require.Empty(t, request.Header.Get(api.HeaderRouteWorkspaceID)) require.Empty(t, request.Header.Get(router.HeaderResolvedWorkspaceGroupID)) require.Equal(t, "wsp_business", request.Header.Get(api.HeaderWorkspaceID)) writer.WriteHeader(http.StatusNoContent) })) t.Cleanup(central.Close) handler := newTestHandler(t, central.URL, map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{}) request := httptest.NewRequest(http.MethodGet, "/api/v1/billing/summary", nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_forged_route") request.Header.Set(api.HeaderWorkspaceID, "wsp_business") request.Header.Set(router.HeaderResolvedWorkspaceGroupID, "group_forged") recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusNoContent, recorder.Code) } func TestHandlerRoutesCentralV1ContractPaths(t *testing.T) { t.Parallel() paths := []string{ "/api/v1/me/behavior-settings", "/api/v1/me/model-profiles", "/api/v1/me/model-profiles:detectModels", "/api/v1/me/model-profiles:testConnection", "/api/avatars/avt_legacy", "/api/v1/users/usr_1/avatar", "/api/v1/workspaces", } for _, path := range paths { path := path t.Run(path, func(t *testing.T) { t.Parallel() central := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, path, request.URL.Path) require.Empty(t, request.Header.Get(api.HeaderRouteWorkspaceID)) _, _ = writer.Write([]byte("central")) })) t.Cleanup(central.Close) handler := newTestHandler(t, central.URL, map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{}) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, path, nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_1") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, "central", recorder.Body.String()) }) } } func TestHandlerRoutesMachineAuthResolutionToCentral(t *testing.T) { t.Parallel() central := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, http.MethodPost, request.Method) require.Equal(t, "/api/v1/machine-auth-resolutions", request.URL.Path) require.Empty(t, request.Header.Get(api.HeaderRouteWorkspaceID)) writer.WriteHeader(http.StatusNoContent) })) t.Cleanup(central.Close) handler := newTestHandler(t, central.URL, map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{}) request := httptest.NewRequest(http.MethodPost, "/api/v1/machine-auth-resolutions", nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_untrusted") recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusNoContent, recorder.Code) } func TestHandlerRoutesWorkspaceV1IMPathsByPathWorkspaceID(t *testing.T) { t.Parallel() paths := []string{ "/api/v1/workspaces/wsp_1/snapshot", "/api/v1/workspaces/wsp_1/events/stream", "/api/v1/workspaces/wsp_1/agents", "/api/v1/workspaces/wsp_1/agents:from-role", "/api/v1/workspaces/wsp_1/agents/agt_1/start", "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/workspace", "/api/v1/workspaces/wsp_1/onboarding/bootstrap", "/api/v1/workspaces/wsp_1/agent-collaborators", "/api/v1/workspaces/wsp_1/mention-candidates/teams", "/api/v1/workspaces/wsp_1/team-templates", "/api/v1/workspaces/wsp_1/team-templates/tpl_1/instantiate", "/api/v1/workspaces/wsp_1/teams", "/api/v1/workspaces/wsp_1/teams/team_1/members", "/api/v1/workspaces/wsp_1/tasks/workspace/browse", "/api/v1/workspaces/wsp_1/tasks/42", "/api/v1/workspaces/wsp_1/task-board/options", "/api/v1/workspaces/wsp_1/channels", "/api/v1/workspaces/wsp_1/channels/chn_1", "/api/v1/workspaces/wsp_1/channel-share-cards/msg_1", "/api/v1/workspaces/wsp_1/sidebar/channel-sections", "/api/v1/workspaces/wsp_1/threads", "/api/v1/workspaces/wsp_1/threads/chn_1", "/api/v1/workspaces/wsp_1/review/feedback", "/api/v1/workspaces/wsp_1/review-contexts/rctx_1/feedback", "/api/v1/workspaces/wsp_1/providers", "/api/v1/workspaces/wsp_1/providers/github/config", "/api/v1/workspaces/wsp_1/selectable-models", "/api/v1/workspaces/wsp_1/schedules", "/api/v1/workspaces/wsp_1/schedules:validate", "/api/v1/workspaces/wsp_1/schedules/sch_1:pause", "/api/v1/workspaces/wsp_1/triggers", "/api/v1/workspaces/wsp_1/triggers:validate", "/api/v1/workspaces/wsp_1/triggers/trg_1:resume", "/api/v1/workspaces/wsp_1/workflow/plans", "/api/v1/workspaces/wsp_1/workflow/plans:validate", "/api/v1/workspaces/wsp_1/workflow/plans/wpl_1/versions", "/api/v1/workspaces/wsp_1/workflow/bindings", "/api/v1/workspaces/wsp_1/workflow/bindings:validate", "/api/v1/workspaces/wsp_1/workflow/plans/wpl_1/invocations", "/api/v1/workspaces/wsp_1/workflow/invocations", "/api/v1/workspaces/wsp_1/workflow/invocations/wiv_1", "/api/v1/workspaces/wsp_1/workflow/runs", "/api/v1/workspaces/wsp_1/workflow/runs/wri_1/events", "/api/v1/workspaces/wsp_1/apps", "/api/v1/workspaces/wsp_1/app-installations", "/api/v1/workspaces/wsp_1/app-installations/ainst_1/runtime/profile", "/api/v1/workspaces/wsp_1/app-installations/ainst_1/service-event-subscriptions", "/api/v1/workspaces/wsp_1/service-account-create-requests", "/api/v1/workspaces/wsp_1/service-accounts/svc_1", "/api/v1/workspaces/wsp_1/avatar-assets/service-account/svc_1", "/api/v1/workspaces/wsp_1/daemon/connect", "/api/v1/workspaces/wsp_1/machines", "/api/v1/workspaces/wsp_1/machines/mac_1/workspace-directories", "/api/v1/workspaces/wsp_1/miniapp-pairing-codes", "/api/v1/workspaces/wsp_1/provider-watches/watch_1:pause", } for _, path := range paths { path := path t.Run(path, func(t *testing.T) { t.Parallel() workspaceUpstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "wsp_1", request.Header.Get(api.HeaderRouteWorkspaceID)) require.Equal(t, path, request.URL.Path) _, _ = writer.Write([]byte("workspace-im")) })) t.Cleanup(workspaceUpstream.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": workspaceUpstream.URL}, fakeResolver{target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}}) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, "workspace-im", recorder.Body.String()) }) } } func TestHandlerNormalizesLegacyRuntimeRoutes(t *testing.T) { for _, test := range []struct { name string method string path string expectedPath string }{ {name: "daemon", method: http.MethodGet, path: "/daemon/connect?key=machine-key", expectedPath: "/api/v1/workspaces/wsp_1/daemon/connect"}, {name: "app", method: http.MethodPost, path: "/internal/app/runtime/heartbeat", expectedPath: "/api/v1/workspaces/wsp_1/app-runtime/heartbeat"}, } { test := test t.Run(test.name, func(t *testing.T) { workspaceUpstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, test.expectedPath, request.URL.Path) require.Equal(t, "wsp_1", request.Header.Get(api.HeaderRouteWorkspaceID)) writer.WriteHeader(http.StatusNoContent) })) t.Cleanup(workspaceUpstream.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": workspaceUpstream.URL}, fakeResolver{ target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}, }) request := httptest.NewRequest(test.method, test.path, nil) request.Header.Set("Authorization", "Bearer runtime-token") recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusNoContent, recorder.Code) }) } } func TestHandlerUsesLegacyWorkspaceParameterAndRejectsConflicts(t *testing.T) { t.Parallel() workspaceUpstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/daemon/upgrade-info", request.URL.Path) require.Equal(t, "wsp_1", request.Header.Get(api.HeaderRouteWorkspaceID)) writer.WriteHeader(http.StatusNoContent) })) t.Cleanup(workspaceUpstream.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": workspaceUpstream.URL}, fakeResolver{ target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}, }) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/daemon/upgrade-info?workspaceId=wsp_1", nil)) require.Equal(t, http.StatusNoContent, recorder.Code) for _, path := range []string{ "/daemon/upgrade-info?workspaceId=wsp_1&workspaceId=wsp_2", "/daemon/upgrade-info?workspaceId=wsp_1&workspace_id=wsp_2", } { recorder = httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) require.Equal(t, http.StatusBadRequest, recorder.Code, path) require.Contains(t, recorder.Body.String(), "conflicting route workspace identity", path) } handler, err := NewHandler(fakeResolver{ target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}, }, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"workspace-default": workspaceUpstream.URL}, IdentityResolver: fakeWorkspaceIdentityResolver{workspaceID: "wsp_credential"}, }) require.NoError(t, err) recorder = httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/daemon/upgrade-info?workspaceId=wsp_1", nil)) require.Equal(t, http.StatusBadRequest, recorder.Code) require.Contains(t, recorder.Body.String(), "conflicting route workspace identity") } func TestHandlerRestoresLegacyWorkspaceMachineRoute(t *testing.T) { t.Parallel() workspaceUpstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/machines/mac_1/api-key-rotations", request.URL.Path) require.Equal(t, "wsp_1", request.Header.Get(api.HeaderRouteWorkspaceID)) writer.WriteHeader(http.StatusNoContent) })) t.Cleanup(workspaceUpstream.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": workspaceUpstream.URL}, fakeResolver{ target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}, }) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/api/workspaces/wsp_1/machines/mac_1/rotate-key", nil)) require.Equal(t, http.StatusNoContent, recorder.Code) } func TestHandlerRoutesWorkspaceByRouteHeaderOnly(t *testing.T) { t.Parallel() workspaceUpstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "wsp_1", request.Header.Get(api.HeaderRouteWorkspaceID)) require.Equal(t, "workspace-default", request.Header.Get("X-AgentTeam-Resolved-Workspace-Group-Id")) require.Equal(t, "/api/v1/workspaces/wsp_1/llm/healthz", request.URL.Path) _, _ = writer.Write([]byte("workspace")) })) t.Cleanup(workspaceUpstream.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": workspaceUpstream.URL}, fakeResolver{target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}}) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, "/api/llm/healthz", nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_1") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, "workspace", recorder.Body.String()) } func TestHandlerRoutesMeshXTrafficToWorkspaceOwner(t *testing.T) { t.Parallel() paths := []string{ "/api/v1/memory", "/api/v1/memory/agents/access:batch", "/api/v1/skills", "/api/v1/skills/agents/access:batch", "/api/v1/kb", "/api/v1/kb/agents/access:batch", "/api/v1/roles", "/api/v1/roles:from-agent", "/api/v1/roles/rol_1", "/api/v1/repository-owners", "/api/v1/repository-names/available", "/api/v1/repository-access-tokens", "/api/v1/repository-access-tokens/tok_1", "/api/v1/personal-access-tokens", "/api/v1/personal-access-tokens/tok_1", "/api/v1/automations/jobs", "/api/v1/mcp", "/api/v1/mcp/servers", "/api/v1/repositories", "/api/v1/repositories/owner/repo", } for _, path := range paths { path := path t.Run(path, func(t *testing.T) { t.Parallel() workspaceUpstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "wsp_1", request.Header.Get(api.HeaderRouteWorkspaceID)) require.Equal(t, "/api/v1/workspaces/wsp_1"+strings.TrimPrefix(path, "/api/v1"), request.URL.Path) _, _ = writer.Write([]byte("workspace")) })) t.Cleanup(workspaceUpstream.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": workspaceUpstream.URL}, fakeResolver{ target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}, }) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, path, nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_1") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusOK, recorder.Code, path) require.Equal(t, "workspace", recorder.Body.String(), path) }) } } func TestHandlerRejectsUnqualifiedAGSCompatibilityBeforeAnyUpstream(t *testing.T) { t.Parallel() var upstreamCalls atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { upstreamCalls.Add(1) })) t.Cleanup(upstream.Close) identity := &countingWorkspaceIdentityResolver{} placement := &countingPlacementResolver{} handler, err := NewHandler(placement, Config{ CentralUpstream: upstream.URL, WorkspaceUpstreams: map[string]string{ "workspace-default": upstream.URL, }, IdentityResolver: identity, }) require.NoError(t, err) for _, route := range []struct { method string path string }{ {method: http.MethodGet, path: "/api/v3"}, {method: http.MethodGet, path: "/api/v3/"}, {method: http.MethodGet, path: "/api/v3/openapi.json"}, {method: http.MethodGet, path: "/api/ext/v1"}, {method: http.MethodGet, path: "/api/ext/v1/"}, {method: http.MethodGet, path: "/api/ext/v1/openapi.json"}, {method: http.MethodGet, path: "/api/v3/meta"}, {method: http.MethodGet, path: "/api/v3/rate_limit"}, {method: http.MethodPost, path: "/login/device/code"}, {method: http.MethodPost, path: "/login/oauth/access_token"}, {method: http.MethodGet, path: "/login/oauth/authorize"}, {method: http.MethodGet, path: "/auth/connected/login"}, {method: http.MethodPost, path: "/api/ext/v1/agents"}, {method: http.MethodPost, path: "/api/ext/v1/analytics/events"}, {method: http.MethodOptions, path: "/api/v3/openapi.json"}, {method: http.MethodPost, path: "/api/v3/openapi.json"}, {method: http.MethodGet, path: "/api/v3/openapi.json/extra"}, {method: http.MethodGet, path: "/api/v3//"}, } { recorder := httptest.NewRecorder() request := httptest.NewRequest(route.method, route.path, nil) handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusNotFound, recorder.Code, "%s %s", route.method, route.path) } require.Zero(t, identity.calls.Load()) require.Zero(t, placement.calls.Load()) require.Zero(t, upstreamCalls.Load()) } func TestHandlerRejectsConflictingWorkspaceRouteHeaders(t *testing.T) { t.Parallel() handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:2"}, fakeResolver{target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}}) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodPost, "/api/llm/healthz", nil) request.Header.Set(api.HeaderWorkspaceID, "wsp_1") request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_2") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusBadRequest, recorder.Code) require.Contains(t, recorder.Body.String(), "conflicting route workspace identity") } func TestHandlerRoutesWorkspaceRESTContractPaths(t *testing.T) { t.Parallel() paths := []string{ "/api/v1/workspaces/wsp_1/threads/chn_1", "/api/v1/workspaces/wsp_1/threads/chn_1/membership", "/api/v1/workspaces/wsp_1/channels/chn_1/private-agent-sender-capability", } for _, path := range paths { path := path t.Run(path, func(t *testing.T) { t.Parallel() workspaceUpstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "wsp_1", request.Header.Get(api.HeaderRouteWorkspaceID)) require.Equal(t, path, request.URL.Path) _, _ = writer.Write([]byte("workspace")) })) t.Cleanup(workspaceUpstream.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": workspaceUpstream.URL}, fakeResolver{target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateActive}}) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, path, nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_1") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, "workspace", recorder.Body.String()) }) } } func TestHandlerFailClosesMissingPlacement(t *testing.T) { t.Parallel() handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{err: storeerr.NotFound("workspace placement for workspace wsp_missing not found")}) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, "/api/llm/healthz", nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_missing") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusNotFound, recorder.Code) require.Contains(t, recorder.Body.String(), "workspace placement for workspace wsp_missing not found") } func TestHandlerRejectsNonActivePlacement(t *testing.T) { t.Parallel() handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{target: workspace.WorkspaceRouteTarget{WorkspaceID: "wsp_1", WorkspaceGroupID: "workspace-default", PlacementState: workspace.PlacementStateMoving}}) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, "/api/llm/healthz", nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_1") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusConflict, recorder.Code) require.Contains(t, recorder.Body.String(), "cannot receive gateway traffic") } func TestHandlerRejectsHeaderlessRepositoryRawBeforeResolverAndOwner(t *testing.T) { t.Parallel() identity := &countingWorkspaceIdentityResolver{} placement := &countingPlacementResolver{} var ownerCalls atomic.Int32 globalOwner := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { ownerCalls.Add(1) })) t.Cleanup(globalOwner.Close) handler, err := NewHandler(placement, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"workspace-default": globalOwner.URL}, IdentityResolver: identity, }) require.NoError(t, err) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/owner/repo/raw/main/README.md", nil)) require.Equal(t, http.StatusBadRequest, recorder.Code) require.Zero(t, identity.calls.Load()) require.Zero(t, placement.calls.Load()) require.Zero(t, ownerCalls.Load()) } func TestDeferredAGSSplitPathsAreRejectedBeforeTheOwner(t *testing.T) { t.Parallel() var ownerRequests atomic.Int32 globalOwner := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { ownerRequests.Add(1) writer.WriteHeader(http.StatusNoContent) })) t.Cleanup(globalOwner.Close) handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": globalOwner.URL}, fakeResolver{}) for _, request := range []struct{ method, path string }{ {http.MethodGet, "/api/v3/repos/owner/repo/collaborators"}, {http.MethodPost, "/api/ext/v1/repos/owner/repo/transfer"}, {http.MethodPatch, "/api/v3/repos/owner/repo"}, {http.MethodGet, "/api/v3/repositories/42/environments"}, {http.MethodPost, "/api/v3/user/repos"}, {http.MethodPost, "/api/v3/search/code"}, {http.MethodPost, "/api/v3/search/labels"}, {http.MethodPost, "/api/ext/v1/user/tokens"}, {http.MethodPost, "/api/v3/orgs/acme/repos"}, {http.MethodPost, "/api/v3/repos/owner/repo/pages"}, {http.MethodPost, "/api/ext/v1/repos/owner/repo/summary"}, {http.MethodPost, "/api/v3/repos/owner/repo/assignees"}, {http.MethodPost, "/api/v3/repos/owner/repo/archive/refs/tags/skill-backup.zip"}, {http.MethodPost, "/api/ext/v1/repos/owner/repo/team-sharing/enable"}, {http.MethodPost, "/api/ext/v1/agents"}, } { require.Equal(t, routeUnknown, classify(request.method, request.path).family, request.path) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(request.method, request.path, nil)) require.Equal(t, http.StatusNotFound, recorder.Code, request.path) } require.Zero(t, ownerRequests.Load(), "deferred split routes must not reach the designated AGS owner") require.Equal(t, routeUnknown, classify(http.MethodGet, "/api/v3/user").family) require.Equal(t, routeUnknown, classify(http.MethodGet, "/api/ext/v1/user/tokens").family) } func TestSplitAGSActivationMatchesRealRegistration(t *testing.T) { t.Parallel() for _, test := range []struct{ method, path string }{ {http.MethodGet, "/api/v3/user/repos"}, {http.MethodGet, "/api/v3/search/code"}, {http.MethodGet, "/api/v3/search/labels"}, {http.MethodGet, "/api/v3/repos/acme/docs"}, {http.MethodGet, "/api/v3/repos/acme/docs/contents/README.md"}, {http.MethodGet, "/api/ext/v1/repos/acme/docs/summary"}, {http.MethodPost, "/api/v3/repos/acme/docs/git/refs"}, {http.MethodGet, "/acme/docs.git/info/refs?service=git-upload-pack"}, {http.MethodPost, "/acme/docs.git/git-upload-pack"}, {http.MethodPost, "/acme/docs.git/git-receive-pack"}, } { request := httptest.NewRequest(test.method, test.path, nil) registered, ok := agsserver.MatchRegisteredRoute(request) require.True(t, ok, test.path) authority, activated := repositoryroute.ActivatedAuthority(registered.Operation) require.True(t, activated, test.path) scope := repositoryroute.ScopeFromRegistration(registered.Operation, registered.Authentication, authority, registered.Repository, registered.Capability) require.Equal(t, repositoryroute.ScopeSourceAGSRegistered, scope.Source, test.path) require.True(t, repositoryroute.IsActivatedOperation(registered.Operation, authority), test.path) require.Equal(t, routeWorkspace, classifyRequest(request).family, test.path) } raw := httptest.NewRequest(http.MethodGet, "/acme/docs/raw/main/README.md", nil) require.Equal(t, repositoryroute.ScopeSourceAgentTeamSynthetic, repositoryroute.Classify(raw).Source) require.Equal(t, routeWorkspace, classifyRequest(raw).family) for _, test := range []struct{ method, path string }{ {http.MethodGet, "/api/v3/user"}, {http.MethodGet, "/api/ext/v1/user/tokens"}, {http.MethodGet, "/api/v3/openapi.json"}, {http.MethodGet, "/api/v3/repos/acme/docs/collaborators"}, {http.MethodPost, "/api/v3/repos/acme/docs/pages"}, {http.MethodPost, "/api/graphql"}, // The old family parser recognizes these prefixes, but the real AGS // registration tree has no matching operation for the descendants. {http.MethodGet, "/api/v3/repos/acme/docs/issues/1/not-registered"}, {http.MethodGet, "/acme/docs.git/not-registered"}, } { require.Equal(t, routeUnknown, classify(test.method, test.path).family, test.path) } } func TestHandlerRejectsHeaderlessRepositoryDataPlaneBeforeResolverAndOwner(t *testing.T) { t.Parallel() identity := &countingWorkspaceIdentityResolver{} placement := &countingPlacementResolver{} var ownerCalls atomic.Int32 globalOwner := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { ownerCalls.Add(1) })) t.Cleanup(globalOwner.Close) handler, err := NewHandler( placement, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"group-a": globalOwner.URL}, IdentityResolver: identity, }, ) require.NoError(t, err) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodGet, "/api/v3/repos/owner/repo/contents/skills/example/SKILL.md?ref=main", nil) request.Header.Set("Authorization", "Bearer sssssssssssssssssssssssssssssssssssssssssss") handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusBadRequest, recorder.Code) require.Zero(t, identity.calls.Load()) require.Zero(t, placement.calls.Load()) require.Zero(t, ownerCalls.Load()) } func TestHandlerRejectsUnsupportedRepositoryAuthoritiesBeforeResolverPlacementAndUpstream(t *testing.T) { t.Parallel() for _, test := range []struct { name string authorization []string wantStatus int }{ {name: "anonymous", wantStatus: http.StatusUnauthorized}, {name: "repository bearer", authorization: []string{"Bearer ags_rat_secret"}, wantStatus: http.StatusUnprocessableEntity}, {name: "repository token", authorization: []string{"Token ags_rat_secret"}, wantStatus: http.StatusUnprocessableEntity}, {name: "repository basic", authorization: []string{"Basic Z2l0OmFnc19yYXRfc2VjcmV0"}, wantStatus: http.StatusUnprocessableEntity}, {name: "multiple values", authorization: []string{"Bearer one", "Bearer two"}, wantStatus: http.StatusBadRequest}, {name: "comma ambiguity", authorization: []string{"Bearer one,Bearer two"}, wantStatus: http.StatusBadRequest}, {name: "extra whitespace", authorization: []string{"Bearer session"}, wantStatus: http.StatusBadRequest}, {name: "empty credential", authorization: []string{"Bearer "}, wantStatus: http.StatusBadRequest}, {name: "control character", authorization: []string{"Bearer bad\ttoken"}, wantStatus: http.StatusBadRequest}, {name: "unicode whitespace", authorization: []string{"Bearer \u00a0token"}, wantStatus: http.StatusBadRequest}, {name: "unicode credential", authorization: []string{"Bearer 令牌"}, wantStatus: http.StatusBadRequest}, {name: "oversized bearer", authorization: []string{"Bearer " + strings.Repeat("x", 16<<10)}, wantStatus: http.StatusBadRequest}, {name: "unicode token credential", authorization: []string{"Token 令牌"}, wantStatus: http.StatusBadRequest}, {name: "emoji token credential", authorization: []string{"Token 💩"}, wantStatus: http.StatusBadRequest}, {name: "unicode repository bearer", authorization: []string{"Bearer ags_rat_令牌"}, wantStatus: http.StatusBadRequest}, {name: "oversized token credential", authorization: []string{"Token " + strings.Repeat("x", 16<<10)}, wantStatus: http.StatusBadRequest}, } { t.Run(test.name, func(t *testing.T) { identity := &countingWorkspaceIdentityResolver{} placement := &countingPlacementResolver{} var upstreamCalls atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { upstreamCalls.Add(1) })) t.Cleanup(upstream.Close) handler, err := NewHandler(placement, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"group-a": upstream.URL}, IdentityResolver: identity, }) require.NoError(t, err) request := httptest.NewRequest(http.MethodGet, "/api/v3/repos/acme/repo/branches", nil) request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_a") for _, value := range test.authorization { request.Header.Add("Authorization", value) } recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) require.Equal(t, test.wantStatus, recorder.Code) require.Zero(t, identity.calls.Load()) require.Zero(t, placement.calls.Load()) require.Zero(t, upstreamCalls.Load()) }) } } func TestHandlerRejectsAmbiguousWorkspaceIdentityHeadersBeforeResolverPlacementAndUpstream(t *testing.T) { t.Parallel() for _, test := range []struct { name string routeValues []string legacyValues []string }{ {name: "duplicate same route", routeValues: []string{"wsp_a", "wsp_a"}}, {name: "duplicate different route", routeValues: []string{"wsp_a", "wsp_b"}}, {name: "comma joined route", routeValues: []string{"wsp_a,wsp_b"}}, {name: "route whitespace", routeValues: []string{" wsp_a"}}, {name: "duplicate same business", legacyValues: []string{"wsp_a", "wsp_a"}}, {name: "cross header conflict", routeValues: []string{"wsp_a"}, legacyValues: []string{"wsp_b"}}, } { t.Run(test.name, func(t *testing.T) { identity := &countingWorkspaceIdentityResolver{} placement := &countingPlacementResolver{} var upstreamCalls atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { upstreamCalls.Add(1) })) t.Cleanup(upstream.Close) handler, err := NewHandler(placement, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"group-a": upstream.URL}, IdentityResolver: identity, }) require.NoError(t, err) request := httptest.NewRequest(http.MethodGet, "/api/v3/repos/acme/repo/branches", nil) request.Header.Set("Authorization", "Bearer "+strings.Repeat("s", 43)) for index, value := range test.routeValues { name := api.HeaderRouteWorkspaceID if index%2 == 1 { name = strings.ToLower(name) } request.Header.Add(name, value) } for _, value := range test.legacyValues { request.Header.Add(api.HeaderWorkspaceID, value) } recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusBadRequest, recorder.Code) require.Zero(t, identity.calls.Load()) require.Zero(t, placement.calls.Load()) require.Zero(t, upstreamCalls.Load()) }) } } func TestHandlerRoutesRepositorySessionOperationsToClaimedWorkspaceGroupOnly(t *testing.T) { t.Parallel() type observation struct{ workspace, path string } observedA := make(chan observation, 2) observedB := make(chan observation, 2) newUpstream := func(observed chan<- observation) *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { observed <- observation{workspace: request.Header.Get(api.HeaderRouteWorkspaceID), path: request.URL.Path} writer.WriteHeader(http.StatusNoContent) })) } groupA, groupB := newUpstream(observedA), newUpstream(observedB) t.Cleanup(groupA.Close) t.Cleanup(groupB.Close) identity := &echoWorkspaceIdentityResolver{} placement := &countingPlacementResolver{targets: map[string]workspace.WorkspaceRouteTarget{ "wsp_a": {WorkspaceID: "wsp_a", WorkspaceGroupID: "group-a", PlacementState: workspace.PlacementStateActive}, "wsp_b": {WorkspaceID: "wsp_b", WorkspaceGroupID: "group-b", PlacementState: workspace.PlacementStateActive}, }} handler, err := NewHandler(placement, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"group-a": groupA.URL, "group-b": groupB.URL}, IdentityResolver: identity, }) require.NoError(t, err) paths := []string{"/api/v3/repos/acme/repo/branches", "/acme/repo/raw/main/README.md"} for _, workspaceID := range []string{"wsp_a", "wsp_b"} { for _, path := range paths { request := httptest.NewRequest(http.MethodGet, path, nil) request.Header.Set(api.HeaderRouteWorkspaceID, workspaceID) request.Header.Set("Authorization", "Bearer sssssssssssssssssssssssssssssssssssssssssss") recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusNoContent, recorder.Code, workspaceID+path) } } for _, expected := range []struct { observed chan observation workspace string }{{observedA, "wsp_a"}, {observedB, "wsp_b"}} { first, second := <-expected.observed, <-expected.observed require.Equal(t, expected.workspace, first.workspace) require.Equal(t, first.workspace, second.workspace) require.ElementsMatch(t, []string{ "/api/v1/workspaces/" + expected.workspace + "/ags" + paths[0], "/api/v1/workspaces/" + expected.workspace + "/ags" + paths[1], }, []string{first.path, second.path}) } require.Equal(t, int32(4), identity.calls.Load()) require.Equal(t, int32(4), placement.calls.Load()) } func TestHandlerRoutesRepositorySessionsThroughPrivateIdentityToTwoWorkspaceGroups(t *testing.T) { gin.SetMode(gin.ReleaseMode) sessions := &wireCountingSessions{} memberships := &wireCountingMemberships{allowed: map[string]string{"wsp_a": "wsp_a", "wsp_b": "wsp_b"}} privateRouter := gin.New() lookup.RegisterRoutes(privateRouter, lookup.Dependencies{ Resolver: lookup.NewAuthenticatedResolver(nil, sessions, memberships), InternalAdminToken: "private-token", }) privateServer := httptest.NewServer(privateRouter) t.Cleanup(privateServer.Close) identityResolver, err := NewHTTPWorkspaceIdentityResolver(privateServer.URL+"/internal/meshxruntime/workspace-route-identity", "private-token", privateServer.Client()) require.NoError(t, err) type observation struct{ workspace, path string } observedA := make(chan observation, 2) observedB := make(chan observation, 2) var defaultCalls atomic.Int32 newUpstream := func(observed chan<- observation) *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { observed <- observation{workspace: request.Header.Get(api.HeaderRouteWorkspaceID), path: request.URL.Path} writer.WriteHeader(http.StatusNoContent) })) } groupA, groupB := newUpstream(observedA), newUpstream(observedB) defaultGroup := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { defaultCalls.Add(1) })) t.Cleanup(groupA.Close) t.Cleanup(groupB.Close) t.Cleanup(defaultGroup.Close) placement := &countingPlacementResolver{targets: map[string]workspace.WorkspaceRouteTarget{ "wsp_a": {WorkspaceID: "wsp_a", WorkspaceGroupID: "group-a", PlacementState: workspace.PlacementStateActive}, "wsp_b": {WorkspaceID: "wsp_b", WorkspaceGroupID: "group-b", PlacementState: workspace.PlacementStateActive}, }} handler, err := NewHandler(placement, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"group-default": defaultGroup.URL, "group-a": groupA.URL, "group-b": groupB.URL}, IdentityResolver: identityResolver, }) require.NoError(t, err) paths := []string{"/api/v3/repos/acme/repo/branches", "/acme/repo/raw/main/README.md"} token := strings.Repeat("s", 43) for _, workspaceID := range []string{"wsp_a", "wsp_b"} { for _, path := range paths { request := httptest.NewRequest(http.MethodGet, path, nil) request.Header.Set(api.HeaderRouteWorkspaceID, workspaceID) request.Header.Set("Authorization", "Bearer "+token) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusNoContent, recorder.Code, workspaceID+path) } } for _, expected := range []struct { observed chan observation workspace string }{{observedA, "wsp_a"}, {observedB, "wsp_b"}} { first, second := <-expected.observed, <-expected.observed require.Equal(t, expected.workspace, first.workspace) require.Equal(t, expected.workspace, second.workspace) require.ElementsMatch(t, []string{ "/api/v1/workspaces/" + expected.workspace + "/ags" + paths[0], "/api/v1/workspaces/" + expected.workspace + "/ags" + paths[1], }, []string{first.path, second.path}) } require.Equal(t, int32(4), sessions.calls.Load()) require.Equal(t, int32(4), memberships.calls.Load()) require.ElementsMatch(t, []string{"wsp_a", "wsp_a", "wsp_b", "wsp_b"}, memberships.claims()) require.Equal(t, int32(4), placement.calls.Load()) require.Zero(t, defaultCalls.Load()) } func TestHandlerPrivateIdentityFailuresDoNotReachPlacementOrAnyGroup(t *testing.T) { gin.SetMode(gin.ReleaseMode) for _, test := range []struct { name string routeValues []string legacyValues []string authorization []string sessionError error allowed map[string]string wantStatus int wantIdentityCalls int32 wantSessionCalls int32 wantMembershipCalls int32 }{ {name: "duplicate route claim", routeValues: []string{"wsp_a", "wsp_a"}}, {name: "comma route claim", routeValues: []string{"wsp_a,wsp_b"}}, {name: "conflicting claims", routeValues: []string{"wsp_a"}, legacyValues: []string{"wsp_b"}}, {name: "unicode token credential", routeValues: []string{"wsp_a"}, authorization: []string{"Token 令牌"}}, {name: "emoji token credential", routeValues: []string{"wsp_a"}, authorization: []string{"Token 💩"}}, {name: "unicode repository bearer", routeValues: []string{"wsp_a"}, authorization: []string{"Bearer ags_rat_令牌"}}, {name: "oversized token credential", routeValues: []string{"wsp_a"}, authorization: []string{"Token " + strings.Repeat("x", 16<<10)}}, {name: "valid repository bearer stops before private identity", routeValues: []string{"wsp_a"}, authorization: []string{"Bearer ags_rat_secret"}, wantStatus: http.StatusUnprocessableEntity}, {name: "valid repository token stops before private identity", routeValues: []string{"wsp_a"}, authorization: []string{"Token existing-ags-token"}, wantStatus: http.StatusUnprocessableEntity}, {name: "valid repository basic stops before private identity", routeValues: []string{"wsp_a"}, authorization: []string{"Basic Z2l0OmFnc19yYXRfc2VjcmV0"}, wantStatus: http.StatusUnprocessableEntity}, {name: "session candidate rejected", routeValues: []string{"wsp_a"}, sessionError: storeerr.ErrUnauthorized, wantIdentityCalls: 1, wantSessionCalls: 1}, {name: "membership rejects forged claim", routeValues: []string{"wsp_forged"}, allowed: map[string]string{"wsp_a": "wsp_a"}, wantIdentityCalls: 1, wantSessionCalls: 1, wantMembershipCalls: 1}, {name: "membership returns conflicting workspace", routeValues: []string{"wsp_a"}, allowed: map[string]string{"wsp_a": "wsp_b"}, wantIdentityCalls: 1, wantSessionCalls: 1, wantMembershipCalls: 1}, } { t.Run(test.name, func(t *testing.T) { sessions := &wireCountingSessions{err: test.sessionError} memberships := &wireCountingMemberships{allowed: test.allowed} privateRouter := gin.New() lookup.RegisterRoutes(privateRouter, lookup.Dependencies{ Resolver: lookup.NewAuthenticatedResolver(nil, sessions, memberships), InternalAdminToken: "private-token", }) var identityCalls atomic.Int32 privateServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { identityCalls.Add(1) privateRouter.ServeHTTP(writer, request) })) t.Cleanup(privateServer.Close) identityResolver, err := NewHTTPWorkspaceIdentityResolver(privateServer.URL+"/internal/meshxruntime/workspace-route-identity", "private-token", privateServer.Client()) require.NoError(t, err) var upstreamCalls atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { upstreamCalls.Add(1) })) t.Cleanup(upstream.Close) placement := &countingPlacementResolver{} handler, err := NewHandler(placement, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"group-default": upstream.URL, "group-a": upstream.URL, "group-b": upstream.URL}, IdentityResolver: identityResolver, }) require.NoError(t, err) for _, path := range []string{"/api/v3/repos/acme/repo/branches", "/acme/repo/raw/main/README.md"} { request := httptest.NewRequest(http.MethodGet, path, nil) authorization := test.authorization if len(authorization) == 0 { authorization = []string{"Bearer " + strings.Repeat("s", 43)} } for _, value := range authorization { request.Header.Add("Authorization", value) } for _, value := range test.routeValues { request.Header.Add(api.HeaderRouteWorkspaceID, value) } for _, value := range test.legacyValues { request.Header.Add(api.HeaderWorkspaceID, value) } recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, request) wantStatus := test.wantStatus if wantStatus == 0 { wantStatus = http.StatusBadRequest } if test.sessionError != nil || test.wantMembershipCalls > 0 { wantStatus = http.StatusUnauthorized } require.Equal(t, wantStatus, recorder.Code, path) } require.Equal(t, test.wantIdentityCalls*2, identityCalls.Load()) require.Equal(t, test.wantSessionCalls*2, sessions.calls.Load()) require.Equal(t, test.wantMembershipCalls*2, memberships.calls.Load()) require.Zero(t, placement.calls.Load()) require.Zero(t, upstreamCalls.Load()) }) } } func TestHandlerRejectsUnsupportedGraphQLRoute(t *testing.T) { t.Parallel() handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:2"}, fakeResolver{}) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/api/graphql", strings.NewReader(`{"query":"{ viewer { login } }"}`))) require.Equal(t, http.StatusNotFound, recorder.Code) require.Contains(t, recorder.Body.String(), "unknown gateway route") } func TestHandlerRejectsRepositoryRawNonGet(t *testing.T) { t.Parallel() handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{}) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/owner/repo/raw/main/README.md", nil)) require.Equal(t, http.StatusNotFound, recorder.Code) } func TestHandlerRejectsInternalAdminAndUnknownRoutes(t *testing.T) { t.Parallel() handler := newTestHandler(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:1"}, fakeResolver{}) for _, path := range []string{"/internal/admin/runtime/db-pool", "/not-classified"} { recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) require.Equal(t, http.StatusNotFound, recorder.Code, path) } } func TestHandlerReadyzChecksAllConfiguredUpstreams(t *testing.T) { t.Parallel() var checkedHosts []string readinessClient := &http.Client{Transport: gatewayRoundTripFunc(func(request *http.Request) (*http.Response, error) { require.Equal(t, http.MethodGet, request.Method) require.Equal(t, "/readyz", request.URL.Path) checkedHosts = append(checkedHosts, request.URL.Host) return &http.Response{ StatusCode: http.StatusOK, Header: make(http.Header), Body: http.NoBody, Request: request, }, nil })} handler, err := NewHandler(fakeResolver{}, Config{ CentralUpstream: "http://central.internal", WorkspaceUpstreams: map[string]string{"workspace-default": "http://workspace.internal"}, IdentityResolver: fakeWorkspaceIdentityResolver{workspaceID: "wsp_1"}, ReadinessClient: readinessClient, }) require.NoError(t, err) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, "ready\n", recorder.Body.String()) require.Equal(t, []string{"central.internal", "workspace.internal"}, checkedHosts) } func TestHandlerReadyzFailsWhenAConfiguredUpstreamIsNotReady(t *testing.T) { t.Parallel() readinessClient := &http.Client{Transport: gatewayRoundTripFunc(func(request *http.Request) (*http.Response, error) { status := http.StatusOK if request.URL.Host == "workspace.internal" { status = http.StatusServiceUnavailable } return &http.Response{ StatusCode: status, Header: make(http.Header), Body: http.NoBody, Request: request, }, nil })} handler, err := NewHandler(fakeResolver{}, Config{ CentralUpstream: "http://central.internal", WorkspaceUpstreams: map[string]string{"workspace-default": "http://workspace.internal"}, IdentityResolver: fakeWorkspaceIdentityResolver{workspaceID: "wsp_1"}, ReadinessClient: readinessClient, }) require.NoError(t, err) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) require.Equal(t, http.StatusServiceUnavailable, recorder.Code) require.Contains(t, recorder.Body.String(), "workspace group workspace-default") } func TestNewHandlerValidatesConfig(t *testing.T) { t.Parallel() _, err := NewHandler(fakeResolver{}, Config{CentralUpstream: "", WorkspaceUpstreams: map[string]string{"workspace-default": "http://127.0.0.1:1"}}) require.EqualError(t, err, "gateway central upstream URL is required") _, err = NewHandler(fakeResolver{}, Config{CentralUpstream: "localhost", WorkspaceUpstreams: map[string]string{"workspace-default": "http://127.0.0.1:1"}}) require.EqualError(t, err, "gateway central upstream URL must include scheme and host") _, err = NewHandler(fakeResolver{}, Config{CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"workspace-default": "localhost"}}) require.EqualError(t, err, "gateway workspace group workspace-default upstream URL must include scheme and host") _, err = NewHandler(fakeResolver{}, Config{CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"workspace-default": "http://127.0.0.1:2"}}) require.EqualError(t, err, "gateway workspace identity resolver is required") } func readyzServer(t *testing.T, status int) *httptest.Server { t.Helper() server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/readyz", request.URL.Path) writer.WriteHeader(status) })) t.Cleanup(server.Close) return server } func newDiscardLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } func newTestHandler(t *testing.T, central string, upstreams map[string]string, resolver Resolver) *Handler { t.Helper() handler, err := NewHandler(resolver, Config{ CentralUpstream: central, WorkspaceUpstreams: upstreams, IdentityResolver: fakeWorkspaceIdentityResolver{workspaceID: "wsp_1"}, }) require.NoError(t, err) return handler } func firstWorkspaceGroup(upstreams map[string]string) string { for groupID := range upstreams { return groupID } return "" } type fakeResolver struct { target workspace.WorkspaceRouteTarget err error } type fakeWorkspaceIdentityResolver struct { workspaceID string err error } func (resolver fakeWorkspaceIdentityResolver) ResolveWorkspaceIdentity(context.Context, WorkspaceIdentityRequest) (RouteIdentity, error) { return RouteIdentity{WorkspaceID: resolver.workspaceID}, resolver.err } func (resolver fakeResolver) ResolveWorkspaceRouteTarget(_ context.Context, workspaceID string) (workspace.WorkspaceRouteTarget, error) { if resolver.err != nil { return workspace.WorkspaceRouteTarget{}, resolver.err } if resolver.target.WorkspaceID == "" { return workspace.WorkspaceRouteTarget{}, fmt.Errorf("unexpected workspace id %s", workspaceID) } return resolver.target, nil } // countingWorkspaceIdentityResolver tracks whether the gateway attempted to // resolve a workspace identity. Extension routes must never trigger resolution // because the Extension server re-authenticates workspace and agent itself // (spec §7). type countingWorkspaceIdentityResolver struct { calls atomic.Int32 workspaceID string err error } func (resolver *countingWorkspaceIdentityResolver) ResolveWorkspaceIdentity(context.Context, WorkspaceIdentityRequest) (RouteIdentity, error) { resolver.calls.Add(1) workspaceID := resolver.workspaceID if workspaceID == "" { workspaceID = "wsp_1" } return RouteIdentity{WorkspaceID: workspaceID}, resolver.err } type echoWorkspaceIdentityResolver struct{ calls atomic.Int32 } func (resolver *echoWorkspaceIdentityResolver) ResolveWorkspaceIdentity(_ context.Context, request WorkspaceIdentityRequest) (RouteIdentity, error) { resolver.calls.Add(1) return RouteIdentity{WorkspaceID: request.WorkspaceID}, nil } type countingPlacementResolver struct { calls atomic.Int32 targets map[string]workspace.WorkspaceRouteTarget } type wireCountingSessions struct { calls atomic.Int32 err error } func (sessions *wireCountingSessions) AuthenticateSession(context.Context, string) (identity.AuthenticatedUser, error) { sessions.calls.Add(1) if sessions.err != nil { return identity.AuthenticatedUser{}, sessions.err } return identity.AuthenticatedUser{User: identity.User{ID: "usr_wire"}}, nil } type wireCountingMemberships struct { calls atomic.Int32 claimsLock sync.Mutex seenClaims []string allowed map[string]string } func (memberships *wireCountingMemberships) GetWorkspaceForUser(_ context.Context, _ string, workspaceID string) (workspace.Summary, error) { memberships.calls.Add(1) memberships.claimsLock.Lock() memberships.seenClaims = append(memberships.seenClaims, workspaceID) memberships.claimsLock.Unlock() resolvedWorkspaceID, ok := memberships.allowed[workspaceID] if !ok { return workspace.Summary{}, storeerr.ErrUnauthorized } return workspace.Summary{WorkspaceID: resolvedWorkspaceID}, nil } func (memberships *wireCountingMemberships) claims() []string { memberships.claimsLock.Lock() defer memberships.claimsLock.Unlock() return append([]string(nil), memberships.seenClaims...) } func (resolver *countingPlacementResolver) ResolveWorkspaceRouteTarget(_ context.Context, workspaceID string) (workspace.WorkspaceRouteTarget, error) { resolver.calls.Add(1) target, ok := resolver.targets[workspaceID] if !ok { return workspace.WorkspaceRouteTarget{}, storeerr.NotFound("workspace placement not found") } return target, nil } // newTestHandlerWithExtension builds a gateway handler that also wires the // optional Extension upstream, plus a counting identity resolver so tests can // assert Extension routes skip workspace identity resolution. func newTestHandlerWithExtension(t *testing.T, central string, upstreams map[string]string, extension string) (*Handler, *countingWorkspaceIdentityResolver) { t.Helper() identity := &countingWorkspaceIdentityResolver{} handler, err := NewHandler(fakeResolver{}, Config{ CentralUpstream: central, WorkspaceUpstreams: upstreams, IdentityResolver: identity, ExtensionUpstream: extension, }) require.NoError(t, err) return handler, identity } // validExtensionRoutes is the closed set of Agent-visible Extension runtime // routes (spec §6.2) the gateway must classify into the Extension family. var validExtensionRoutes = []struct { method string path string }{ {http.MethodGet, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext"}, {http.MethodGet, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github"}, {http.MethodGet, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue"}, {http.MethodPost, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue"}, } // TestGateway_ExtensionRoutesForwardedOnlyToExtensionUpstream is the core // adversarial ownership assertion (spec §7): every valid Extension route is // classified into the Extension family and reaches ONLY the Extension upstream. // The central and workspace upstreams must never observe it, the path is // forwarded unchanged, and workspace identity is never resolved. func TestGateway_ExtensionRoutesForwardedOnlyToExtensionUpstream(t *testing.T) { t.Parallel() var centralHits, workspaceHits, extensionHits atomic.Int32 central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { centralHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(central.Close) workspace := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { workspaceHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(workspace.Close) extension := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { extensionHits.Add(1) // The Workspace header may arrive as a diagnostic hint but is never // authorization (spec §7); the Extension server re-authenticates. w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("ext-ok")) })) t.Cleanup(extension.Close) handler, identity := newTestHandlerWithExtension(t, central.URL, map[string]string{"workspace-default": workspace.URL}, extension.URL) for _, route := range validExtensionRoutes { recorder := httptest.NewRecorder() request := httptest.NewRequest(route.method, route.path, nil) // A caller-supplied workspace header is a routing hint, not authorization. request.Header.Set(api.HeaderRouteWorkspaceID, "wsp_1") handler.ServeHTTP(recorder, request) require.Equalf(t, http.StatusOK, recorder.Code, "%s %s", route.method, route.path) require.Equalf(t, "ext-ok", recorder.Body.String(), "%s %s body", route.method, route.path) } require.Equalf(t, int32(len(validExtensionRoutes)), extensionHits.Load(), "extension upstream must receive every valid extension route") require.Zero(t, centralHits.Load(), "extension routes must never reach the central upstream") require.Zero(t, workspaceHits.Load(), "extension routes must never reach the workspace upstream") require.Zero(t, identity.calls.Load(), "extension routes must not trigger workspace identity resolution") } // TestGateway_ExtensionRouteForwardedPathUnchanged asserts the gateway forwards // the original path to the Extension upstream without rewrite (spec §7). func TestGateway_ExtensionRouteForwardedPathUnchanged(t *testing.T) { t.Parallel() var seen []string extension := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { seen = append(seen, r.URL.Path) w.WriteHeader(http.StatusOK) })) t.Cleanup(extension.Close) handler, _ := newTestHandlerWithExtension(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:2"}, extension.URL) path := "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue" recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, path, strings.NewReader(`{}`))) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, []string{path}, seen, "gateway must forward the path unchanged to the extension upstream") } // TestGateway_ExtensionRouteInvalidMethodFailsClosed asserts that any HTTP // method outside the Extension closed set fails closed (404) and reaches NO // upstream. Extension routes must never be routed somewhere by method accident. func TestGateway_ExtensionRouteInvalidMethodFailsClosed(t *testing.T) { t.Parallel() var anyHit atomic.Int32 extension := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { anyHit.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(extension.Close) central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { anyHit.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(central.Close) workspace := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { anyHit.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(workspace.Close) handler, _ := newTestHandlerWithExtension(t, central.URL, map[string]string{"workspace-default": workspace.URL}, extension.URL) invalid := []struct { method string path string }{ {http.MethodPut, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext"}, {http.MethodDelete, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github"}, {http.MethodPatch, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue"}, {http.MethodHead, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext"}, {http.MethodOptions, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github"}, } for _, route := range invalid { recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(route.method, route.path, nil)) require.Equalf(t, http.StatusNotFound, recorder.Code, "%s %s must fail closed", route.method, route.path) } require.Zero(t, anyHit.Load(), "invalid methods on extension routes must reach no upstream") } // TestGateway_UnknownExtensionDescendantFailsClosed_NoWorkspaceFallback is the // critical no-fallthrough assertion (spec §7): unknown Extension descendants // fail closed and are NEVER handed to the workspace family. Without the // Extension classifier running before workspaceV1IMRoute, these paths (whose // 5th segment is "agents") would be stolen by the workspace family. func TestGateway_UnknownExtensionDescendantFailsClosed_NoWorkspaceFallback(t *testing.T) { t.Parallel() var workspaceHits, centralHits, extensionHits atomic.Int32 central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { centralHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(central.Close) workspace := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { workspaceHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(workspace.Close) extension := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { extensionHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(extension.Close) handler, _ := newTestHandlerWithExtension(t, central.URL, map[string]string{"workspace-default": workspace.URL}, extension.URL) unknown := []string{ // Unknown descendant beyond the action segment. "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue/bogus", // Unknown subresource shape under a domain. "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/foo", // Two segments after ext is never a valid shape. "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/foo/bar", // actions without a trailing action id. "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions", // Four segments after ext (e.g. a nested bogus resource). "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue/bogus/deeper", } for _, path := range unknown { recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) require.Equalf(t, http.StatusNotFound, recorder.Code, "unknown descendant %s must fail closed", path) } require.Zero(t, workspaceHits.Load(), "unknown extension descendants must NEVER fall through to workspace") require.Zero(t, centralHits.Load(), "unknown extension descendants must NEVER fall through to central") require.Zero(t, extensionHits.Load(), "unknown extension descendants must not reach the extension upstream") } // TestGateway_ExtensionUpstreamUnsetFailsClosed asserts that when no Extension // upstream is configured, recognized Extension routes fail closed at the gateway // (503) and are NEVER handed off to the central or workspace family (spec §7 // split-deployment model: "If unset, extension routes are not classified (or // fail)"). This is the conservative fail-closed choice. func TestGateway_ExtensionUpstreamUnsetFailsClosed(t *testing.T) { t.Parallel() var centralHits, workspaceHits atomic.Int32 central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { centralHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(central.Close) workspace := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { workspaceHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(workspace.Close) // No ExtensionUpstream configured. handler := newTestHandler(t, central.URL, map[string]string{"workspace-default": workspace.URL}, fakeResolver{}) for _, route := range validExtensionRoutes { recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(route.method, route.path, nil)) require.Equalf(t, http.StatusServiceUnavailable, recorder.Code, "%s %s must fail closed when extension upstream unset", route.method, route.path) require.Contains(t, recorder.Body.String(), "extension runtime upstream is not configured") } require.Zero(t, centralHits.Load(), "unset extension upstream must never spill extension traffic to central") require.Zero(t, workspaceHits.Load(), "unset extension upstream must never spill extension traffic to workspace") } // TestGateway_ReadinessIncludesExtensionUpstream asserts the gateway readiness // check includes the configured Extension upstream: it passes when extension is // ready and FAILS when the extension upstream is unreachable (spec §7). func TestGateway_ReadinessIncludesExtensionUpstream(t *testing.T) { t.Parallel() extensionReady := true readinessClient := &http.Client{Transport: gatewayRoundTripFunc(func(request *http.Request) (*http.Response, error) { require.Equal(t, "/readyz", request.URL.Path) if request.URL.Host == "extension.internal" && !extensionReady { return nil, errors.New("extension unavailable") } return &http.Response{ StatusCode: http.StatusOK, Header: make(http.Header), Body: http.NoBody, Request: request, }, nil })} handler, err := NewHandler(fakeResolver{}, Config{ CentralUpstream: "http://central.internal", WorkspaceUpstreams: map[string]string{"workspace-default": "http://workspace.internal"}, IdentityResolver: fakeWorkspaceIdentityResolver{workspaceID: "wsp_1"}, ExtensionUpstream: "http://extension.internal", ReadinessClient: readinessClient, }) require.NoError(t, err) recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) require.Equal(t, http.StatusOK, recorder.Code) // Take the extension upstream offline: readiness must now FAIL because the // Extension upstream is part of the readiness set. extensionReady = false recorder = httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) require.Equal(t, http.StatusServiceUnavailable, recorder.Code) require.Contains(t, recorder.Body.String(), "extension") } type gatewayRoundTripFunc func(*http.Request) (*http.Response, error) func (function gatewayRoundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { return function(request) } // TestGateway_NoExtensionUpstreamURLInErrorBodies is the diagnostic-redaction // assertion (spec §7: "never exposes its URL in errors"). When the Extension // upstream is unreachable, neither the proxy error body nor the readiness error // body may contain the internal upstream URL or host. The readiness message // names the target ("extension") but never the host; the proxy returns a fixed // message. func TestGateway_NoExtensionUpstreamURLInErrorBodies(t *testing.T) { t.Parallel() // An unreachable upstream whose host:port must never appear in any response. const unreachableURL = "http://127.0.0.1:9" transport := gatewayRoundTripFunc(func(request *http.Request) (*http.Response, error) { if request.URL.Scheme+"://"+request.URL.Host == unreachableURL { return nil, fmt.Errorf("dial %s: connection refused", request.URL.Host) } return &http.Response{ StatusCode: http.StatusOK, Header: make(http.Header), Body: http.NoBody, Request: request, }, nil }) handler, err := NewHandler(fakeResolver{}, Config{ CentralUpstream: "http://central.internal", WorkspaceUpstreams: map[string]string{"workspace-default": "http://workspace.internal"}, IdentityResolver: fakeWorkspaceIdentityResolver{workspaceID: "wsp_1"}, ExtensionUpstream: unreachableURL, ReadinessClient: &http.Client{Transport: transport}, }) require.NoError(t, err) // Exercise the proxy's error handler without depending on a real socket. handler.extensionProxy.Transport = transport // Valid extension route with an unreachable upstream: proxy ErrorHandler // returns a fixed message; the URL/host must not leak. recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext", nil)) require.Equal(t, http.StatusBadGateway, recorder.Code) require.NotContains(t, recorder.Body.String(), "127.0.0.1:9", "proxy error must not expose the upstream host") require.NotContains(t, recorder.Body.String(), unreachableURL, "proxy error must not expose the upstream URL") // Readiness with an unreachable extension upstream: the error names the // target but must not leak the URL or host. readyRecorder := httptest.NewRecorder() handler.ServeHTTP(readyRecorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) require.Equal(t, http.StatusServiceUnavailable, readyRecorder.Code) require.Contains(t, readyRecorder.Body.String(), "extension") require.NotContains(t, readyRecorder.Body.String(), "127.0.0.1:9", "readiness error must not expose the upstream host") require.NotContains(t, readyRecorder.Body.String(), unreachableURL, "readiness error must not expose the upstream URL") } // TestClassify_ExtensionRouteOrder pins the classification order (spec §7): // internal-only -> Extension -> central -> workspace -> unknown. Extension paths // are classified as routeExtension (valid) or routeUnknown (fail-closed) and // never as central/workspace, even though /api/v1/workspaces/* would otherwise // match isCentralRoute and workspaceV1IMRoute. func TestClassify_ExtensionRouteOrder(t *testing.T) { t.Parallel() valid := validExtensionRoutes for _, route := range valid { require.Equalf(t, routeExtension, classify(route.method, route.path).family, "%s %s", route.method, route.path) } // Invalid method on the extension prefix stays within the extension family // (fail-closed as routeUnknown), never central/workspace. require.Equal(t, routeUnknown, classify(http.MethodPut, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext").family) require.Equal(t, routeUnknown, classify(http.MethodDelete, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github").family) // Unknown descendant stays fail-closed (routeUnknown), never workspace. require.Equal(t, routeUnknown, classify(http.MethodGet, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue/bogus").family) // A non-extension workspace runtime path is still classified as workspace: // the extension classifier only owns the /runtime/ext prefix. require.Equal(t, routeWorkspace, classify(http.MethodPost, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/send").family) } // TestNewHandler_ExtensionUpstreamValidated asserts a malformed Extension // upstream URL is rejected at construction (it must include scheme and host). func TestNewHandler_ExtensionUpstreamValidated(t *testing.T) { t.Parallel() _, err := NewHandler(fakeResolver{}, Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"workspace-default": "http://127.0.0.1:2"}, IdentityResolver: fakeWorkspaceIdentityResolver{}, ExtensionUpstream: "agentteam-server-extension:3002", }) require.EqualError(t, err, "gateway extension upstream URL must include scheme and host") } // TestClassify_ExtensionRouteExactPathNoSlashTrimming pins the exact-path // contract (spec §7): the Extension namespace is claimed for slash aliases and // empty-id variants so they fail closed as routeUnknown, and never fall through // to central (which matches "/api/v1/workspaces/") or workspace (whose // workspaceV1IMRoute recognizes "/agents"). Only the resolved, exactly-parsed // path is classified as an Extension route. func TestClassify_ExtensionRouteExactPathNoSlashTrimming(t *testing.T) { t.Parallel() // Expected valid routes are still classified into the Extension family. for _, route := range validExtensionRoutes { require.Equalf(t, routeExtension, classify(route.method, route.path).family, "valid route %s %s", route.method, route.path) } // Adversarial slash-alias and empty-id variants: all owned by the Extension // namespace and fail closed as routeUnknown — NOT routeCentral and NOT // routeWorkspace (routeUnknown is a distinct family, so this also asserts // no spill to either). adversarial := []string{ "//api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext", // leading slash "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/", // trailing slash "/api/v1/workspaces//agents/agt_1/runtime/ext", // missing workspace id "/api/v1/workspaces/wsp_1/agents//runtime/ext", // missing agent id } for _, path := range adversarial { decision := classify(http.MethodGet, path) require.Equalf(t, routeUnknown, decision.family, "adversarial path %q must fail closed as routeUnknown", path) } } // TestGateway_ExtensionRouteMalformedPathsFailClosed_NoSpill is the end-to-end // no-spill assertion for the exact-path contract (spec §7): slash-alias and // empty-id variants of an Extension path fail closed (404) at the gateway and // reach NO upstream — not central (the spill risk for a missing workspace id), // not workspace (the spill risk for a missing agent id), and not extension. func TestGateway_ExtensionRouteMalformedPathsFailClosed_NoSpill(t *testing.T) { t.Parallel() var centralHits, workspaceHits, extensionHits atomic.Int32 central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { centralHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(central.Close) workspace := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { workspaceHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(workspace.Close) extension := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { extensionHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(extension.Close) handler, _ := newTestHandlerWithExtension(t, central.URL, map[string]string{"workspace-default": workspace.URL}, extension.URL) malformed := []string{ "//api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext", "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/", "/api/v1/workspaces//agents/agt_1/runtime/ext", "/api/v1/workspaces/wsp_1/agents//runtime/ext", } for _, path := range malformed { recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) require.Equalf(t, http.StatusNotFound, recorder.Code, "malformed extension path %q must fail closed", path) } require.Zero(t, centralHits.Load(), "malformed extension paths must NEVER spill to central") require.Zero(t, workspaceHits.Load(), "malformed extension paths must NEVER spill to workspace") require.Zero(t, extensionHits.Load(), "malformed extension paths must not reach the extension upstream") } // TestNewHandler_ExtensionUpstreamRequiresOriginOnlyURL pins the origin-only // contract (spec §7): the Extension upstream must be exactly http(s)://host[:port] // with no path, query, userinfo, or fragment. Any non-origin upstream is rejected // at config time so the reverse proxy cannot inject a prefix or query onto proxied // requests. func TestNewHandler_ExtensionUpstreamRequiresOriginOnlyURL(t *testing.T) { t.Parallel() base := Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"workspace-default": "http://127.0.0.1:2"}, IdentityResolver: fakeWorkspaceIdentityResolver{}, } for _, item := range []struct { name string upstream string errContains string }{ {name: "path forbidden", upstream: "http://agentteam-extension:3002/hidden-prefix", errContains: "path is forbidden"}, {name: "query forbidden", upstream: "http://agentteam-extension:3002?q=1", errContains: "query is forbidden"}, {name: "userinfo forbidden", upstream: "http://user:pass@agentteam-extension:3002", errContains: "userinfo is forbidden"}, {name: "fragment forbidden", upstream: "http://agentteam-extension:3002#secret", errContains: "fragment is forbidden"}, {name: "non-http scheme forbidden", upstream: "ftp://agentteam-extension:3002", errContains: "must use http or https scheme"}, } { item := item t.Run(item.name, func(t *testing.T) { t.Parallel() cfg := base cfg.ExtensionUpstream = item.upstream _, err := NewHandler(fakeResolver{}, cfg) require.Error(t, err) require.Contains(t, err.Error(), item.errContains) }) } // A resolved origin-only upstream is accepted. t.Run("origin-only accepted", func(t *testing.T) { t.Parallel() cfg := base cfg.ExtensionUpstream = "http://agentteam-extension:3002" _, err := NewHandler(fakeResolver{}, cfg) require.NoError(t, err) }) } // TestGateway_ExtensionRouteForwardsPathAndQueryUnchanged pins the path-unchanged // contract (spec §7): with an origin-only upstream, the gateway forwards the // original request path and query to the Extension upstream with no injected // prefix and no injected query parameter. func TestGateway_ExtensionRouteForwardsPathAndQueryUnchanged(t *testing.T) { t.Parallel() var seenPaths, seenQueries []string extension := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { seenPaths = append(seenPaths, r.URL.Path) seenQueries = append(seenQueries, r.URL.RawQuery) w.WriteHeader(http.StatusOK) })) t.Cleanup(extension.Close) handler, _ := newTestHandlerWithExtension(t, "http://127.0.0.1:1", map[string]string{"workspace-default": "http://127.0.0.1:2"}, extension.URL) const path = "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext/github/actions/create-issue" recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodPost, path+"?trace=abc&run=1", nil) handler.ServeHTTP(recorder, request) require.Equal(t, http.StatusOK, recorder.Code) require.Equal(t, []string{path}, seenPaths, "gateway must forward the path unchanged with no injected prefix") require.Equal(t, []string{"trace=abc&run=1"}, seenQueries, "gateway must forward the query unchanged with no injected query") } // interiorSlashNoiseAdversarialPaths are slash-noise variants that still // preserve the resolved Extension keyword spine api/v1/workspaces/{ws}/agents/ // {agent}/runtime/ext after collapseSlashes (spec §7, stage-4 round-3). Each // MUST be owned by the Extension namespace and fail closed as routeUnknown — // never central, never workspace. The slash noise lands where collapseSlashes // removes only an EMPTY segment (before "workspaces", between api and v1, or in // an empty id slot), so every load-bearing keyword — including "runtime" — is // still present and the path is genuinely an Extension runtime path. // // Slash noise that would collapse a literal keyword away is NOT Extension // namespace and is covered by TestClassify_ExtensionNamespaceNoOverclaim: e.g. // /workspaces/ws//runtime/ext (the "//" stands in for the missing agents slot) // and /workspaces/ws/agents/agent//ext (the "//" stands in for the missing // "runtime" keyword) collapse to workspace/central routes, not Extension. var interiorSlashNoiseAdversarialPaths = []string{ "/api/v1//workspaces/ws/agents/agent/runtime/ext", // slash before workspaces "/api/v1/workspaces//agents/agent/runtime/ext", // slash before agents (empty ws) "/api/v1/workspaces/ws/agents/agent/runtime//ext", // slash before ext (after runtime) "/api//v1/workspaces/ws/agents/agent/runtime/ext", // slash between api/v1 } // TestClassify_ExtensionRouteInteriorSlashNoiseOwned_FailsClosed pins the // systematic ownership normalization (spec §7): every interior-slash-noise // variant that previously spilled to routeCentral or routeWorkspace is now owned // by the Extension namespace and fails closed as routeUnknown. routeUnknown is a // distinct family from routeCentral and routeWorkspace, so this also asserts no // spill to either. func TestClassify_ExtensionRouteInteriorSlashNoiseOwned_FailsClosed(t *testing.T) { t.Parallel() for _, path := range interiorSlashNoiseAdversarialPaths { decision := classify(http.MethodGet, path) require.Equalf(t, routeUnknown, decision.family, "interior slash noise %q must be owned by Extension and fail closed as routeUnknown", path) } // Sanity: the resolved valid route is still classified as Extension. require.Equal(t, routeExtension, classify(http.MethodGet, "/api/v1/workspaces/wsp_1/agents/agt_1/runtime/ext").family) } // TestGateway_ExtensionRouteInteriorSlashNoiseFailClosed_NoSpill is the // end-to-end no-spill assertion for the systematic ownership normalization // (spec §7): every interior-slash-noise variant fails closed (404) at the gateway // and reaches NO upstream — not central (the spill risk for a slash before // workspaces/runtime, matched by the central "/api/v1/workspaces/" prefix), not // workspace (the spill risk for a slash before agents/ext, matched by // workspaceV1IMRoute's "/agents" suffix), and not extension. func TestGateway_ExtensionRouteInteriorSlashNoiseFailClosed_NoSpill(t *testing.T) { t.Parallel() var centralHits, workspaceHits, extensionHits atomic.Int32 central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { centralHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(central.Close) workspace := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { workspaceHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(workspace.Close) extension := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { extensionHits.Add(1) w.WriteHeader(http.StatusOK) })) t.Cleanup(extension.Close) handler, _ := newTestHandlerWithExtension(t, central.URL, map[string]string{"workspace-default": workspace.URL}, extension.URL) for _, path := range interiorSlashNoiseAdversarialPaths { recorder := httptest.NewRecorder() handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) require.Equalf(t, http.StatusNotFound, recorder.Code, "interior slash noise %q must fail closed as 404", path) } require.Zero(t, centralHits.Load(), "interior slash noise must NEVER spill to central") require.Zero(t, workspaceHits.Load(), "interior slash noise must NEVER spill to workspace") require.Zero(t, extensionHits.Load(), "interior slash noise must not reach the extension upstream") } // TestClassify_ExtensionNamespaceNoOverclaim pins the stage-4 round-3 ownership // narrowing (spec §7): ownsExtensionNamespace claims ONLY the collapsed keyword // spine api/v1/workspaces/{ws}/agents/{agent}/runtime/ext. Paths that lack the // "runtime" keyword before "ext" (workspace agents routes that merely end in // "ext") or lack the "agents" keyword before "runtime" are NOT Extension // namespace — they fall through to their rightful workspace or central family // instead of being stolen by Extension and fail-closed to routeUnknown. // // collapseSlashes can only remove EMPTY segments (slash noise), never a literal // keyword, so a double slash that stands in for a missing keyword slot collapses // to a genuinely different route: "/workspaces/ws//runtime/ext" has no agents // slot, and "/workspaces/ws/agents/agent//ext" has no runtime keyword. func TestClassify_ExtensionNamespaceNoOverclaim(t *testing.T) { t.Parallel() for _, item := range []struct { name string path string family routeFamily }{ {"agents route ending in ext is workspace", "/api/v1/workspaces/ws/agents/agent/ext", routeWorkspace}, {"agents root ending in ext is workspace", "/api/v1/workspaces/ws/agents/ext", routeWorkspace}, {"runtime foo ext (no agents) is central", "/api/v1/workspaces/ws/runtime/foo/ext", routeCentral}, {"slash before ext collapses runtime away is workspace", "/api/v1/workspaces/ws/agents/agent//ext", routeWorkspace}, {"slash before runtime collapses agents away is central", "/api/v1/workspaces/ws//runtime/ext", routeCentral}, } { decision := classify(http.MethodGet, item.path) require.NotEqualf(t, routeExtension, decision.family, "overclaim case %q (%s) must NEVER be classified as Extension", item.path, item.name) require.Equalf(t, item.family, decision.family, "overclaim case %q (%s) must fall through to its rightful family %v (got %v), not be stolen by Extension", item.path, item.name, item.family, decision.family) } // Sanity: the resolved Extension spine and its empty-id / descendant // variants are STILL owned by Extension (the narrowing did not over-reach). owned := []string{ "/api/v1/workspaces/ws/agents/agent/runtime/ext", "/api/v1/workspaces/ws/agents/agent/runtime/ext/extra", "/api/v1/workspaces//agents/agent/runtime/ext", "/api/v1/workspaces/ws/agents//runtime/ext", "/api/v1/workspaces/ws/agents/agent/runtime/ext/github/foo", } for _, path := range owned { decision := classify(http.MethodGet, path) require.Truef(t, decision.family == routeExtension || decision.family == routeUnknown, "valid-spine path %q must be owned by Extension (routeExtension or fail-closed routeUnknown), got %v", path, decision.family) } } // TestNewHandler_ExtensionUpstreamOriginAuthorityPrecision pins the origin-only // authority precision (spec §7): the Extension upstream must be exactly // http(s)://host[:port]. A bare query separator (ForceQuery, e.g. http://host?), // an empty hostname (http://:3002), and a malformed empty port (http://host:) are // all rejected, while the two equivalent root forms — no trailing slash // (http://host:3002) and a single root slash (http://host:3002/) — are both // accepted. A bracketed IPv6 literal (e.g. http://[::1]) is a valid host with no // port — its address colons are not a port separator — and an explicit port, // when present, must be a numeric value in [1, 65535] (so http://host:0 and // http://host:70000 are rejected while http://host:65535 is accepted). func TestNewHandler_ExtensionUpstreamOriginAuthorityPrecision(t *testing.T) { t.Parallel() base := Config{ CentralUpstream: "http://127.0.0.1:1", WorkspaceUpstreams: map[string]string{"workspace-default": "http://127.0.0.1:2"}, IdentityResolver: fakeWorkspaceIdentityResolver{}, } for _, item := range []struct { name string upstream string errContains string }{ {name: "force query rejected", upstream: "http://host:3002?", errContains: "query is forbidden"}, {name: "bare host force query rejected", upstream: "http://host?", errContains: "query is forbidden"}, {name: "empty hostname rejected", upstream: "http://:3002", errContains: "hostname is required"}, {name: "malformed empty port rejected", upstream: "http://host:", errContains: "port is malformed"}, {name: "port zero rejected", upstream: "http://host:0", errContains: "out of range"}, {name: "port over range rejected", upstream: "http://host:70000", errContains: "out of range"}, {name: "ipv6 port zero rejected", upstream: "http://[::1]:0", errContains: "out of range"}, {name: "ipv6 empty port rejected", upstream: "http://[::1]:", errContains: "port is malformed"}, } { item := item t.Run(item.name, func(t *testing.T) { t.Parallel() cfg := base cfg.ExtensionUpstream = item.upstream _, err := NewHandler(fakeResolver{}, cfg) require.Error(t, err) require.Contains(t, err.Error(), item.errContains) }) } for _, item := range []struct { name string upstream string }{ {name: "no trailing slash accepted", upstream: "http://host:3002"}, {name: "root slash accepted", upstream: "http://host:3002/"}, {name: "bare host accepted", upstream: "http://host"}, {name: "max port accepted", upstream: "http://host:65535"}, {name: "ipv6 with port accepted", upstream: "http://[::1]:3002"}, {name: "ipv6 no port accepted", upstream: "http://[::1]"}, } { item := item t.Run(item.name, func(t *testing.T) { t.Parallel() cfg := base cfg.ExtensionUpstream = item.upstream _, err := NewHandler(fakeResolver{}, cfg) require.NoError(t, err) }) } }