Function TestMCPServerOAuth2PKCE

coderd/mcp_test.go:1128–1385

nolint:bodyclose

(t *testing.T)

// nolint:bodyclose
func TestMCPServerOAuth2PKCE(t *testing.T) {
	t.Parallel()

	t.Run("ConnectSetsPKCEParams", func(t *testing.T) {
		t.Parallel()

		ctx := testutil.Context(t, testutil.WaitLong)
		adminClient := newMCPClient(t)
		firstUser := coderdtest.CreateFirstUser(t, adminClient)
		memberClient, _ := coderdtest.CreateAnotherUser(t, adminClient, firstUser.OrganizationID)

		// Create an OAuth2 MCP server config.
		created, err := adminClient.CreateMCPServerConfig(ctx, codersdk.CreateMCPServerConfigRequest{
			DisplayName:    "PKCE Test",
			Slug:           "pkce-test",
			Transport:      "streamable_http",
			URL:            "https://mcp.example.com/pkce",
			AuthType:       "oauth2",
			OAuth2ClientID: "test-client",
			OAuth2AuthURL:  "https://auth.example.com/authorize",
			OAuth2TokenURL: "https://auth.example.com/token",
			Availability:   "default_on",
			Enabled:        true,
			ToolAllowList:  []string{},
			ToolDenyList:   []string{},
		})
		require.NoError(t, err)

		// Prevent the HTTP client from following redirects so we
		// can inspect the response headers and cookies directly.
		memberClient.HTTPClient.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
			return http.ErrUseLastResponse
		}

		connectURL, err := memberClient.URL.Parse(
			"/api/experimental/mcp/servers/" + created.ID.String() + "/oauth2/connect",
		)
		require.NoError(t, err)

		req, err := http.NewRequestWithContext(ctx, "GET", connectURL.String(), nil)
		require.NoError(t, err)
		req.AddCookie(&http.Cookie{
			Name:  codersdk.SessionTokenCookie,
			Value: memberClient.SessionToken(),
		})

		res, err := memberClient.HTTPClient.Do(req)
		require.NoError(t, err)
		defer res.Body.Close()

		require.Equal(t, http.StatusTemporaryRedirect, res.StatusCode)

		// The redirect URL must contain PKCE query parameters.
		location, err := res.Location()
		require.NoError(t, err)
		query := location.Query()
		require.Equal(t, "S256", query.Get("code_challenge_method"),
			"connect redirect must include code_challenge_method=S256")