From 04399b244f8080aae9ca972c57d5ac4e907bde63 Mon Sep 17 00:00:00 2001 From: Jakub Hrozek Date: Wed, 29 Jul 2026 14:17:16 +0200 Subject: [PATCH] Fix port-selection TOCTOU race in e2e tests networking.FindOrUsePort probes a port by binding then immediately releasing it, so anything else on the machine can steal the number before the real bind happens later. Under sharded CI this is lost often enough to flake (observed on PR #6088 as "OIDC mock server error: listen tcp :49278: bind: address already in use"). Two fixes depending on where the real bind happens: - In-process mock servers (OIDCMockServer, LLMGatewayMock) now open their own listener at construction time instead of accepting a pre-selected port, closing the window entirely. - Subprocess-launched proxies (thv serve, thv llm proxy start) get a shared RetryOnPortConflict helper that retries once on a detected bind conflict, generalizing the pattern already used for Docker workload port races in the vMCP dual-era test helpers. thv proxy needed a different approach: it silently substitutes a different port on conflict for a nonzero request (no error to retry on) and its real bind can lag port selection by a full OAuth exchange, so those tests instead pass --port 0 and read the real bound port back from the subprocess's own stdout. Production callers with the same root cause (Docker port bindings, in-process OAuth/proxy binders) are tracked separately in #6141. Generated with Claude Code (https://claude.com/claude-code) --- test/e2e/api_helpers.go | 126 +++++---- test/e2e/cli_llm_all_clients_test.go | 196 +++++++------- test/e2e/cli_llm_setup_test.go | 10 +- test/e2e/llm_gateway_mock.go | 37 ++- test/e2e/oidc_mock.go | 31 ++- test/e2e/proxy_oauth_test.go | 370 ++++++++++++++++----------- test/e2e/vmcp_infra_features_test.go | 4 +- 7 files changed, 462 insertions(+), 312 deletions(-) diff --git a/test/e2e/api_helpers.go b/test/e2e/api_helpers.go index 7612114f50..3ab58503fa 100644 --- a/test/e2e/api_helpers.go +++ b/test/e2e/api_helpers.go @@ -57,6 +57,31 @@ type Server struct { stdout *strings.Builder } +// RetryOnPortConflict runs attempt with port and returns its result. The +// port is only probed free at selection time (networking.FindOrUsePort); +// the real bind happens later, in a subprocess this test doesn't control, so +// another process can steal the port before that bind occurs. If attempt +// fails with an address-already-in-use signal, RetryOnPortConflict +// allocates a fresh port and retries attempt exactly once -- mirroring +// startEraBackendOnPort's handling of the same TOCTOU window for +// Docker-bound ports in test/e2e/vmcp_dual_era_helpers_test.go. It returns +// the port actually used (the original, or the fresh one on retry) along +// with attempt's result. +func RetryOnPortConflict(port int, attempt func(port int) error) (int, error) { + err := attempt(port) + if err == nil || !strings.Contains(err.Error(), "address already in use") { + return port, err + } + + newPort, ferr := networking.FindOrUsePort(0) + if ferr != nil { + GinkgoWriter.Printf("failed to find a fresh port after port %d lost a bind race: %v\n", port, ferr) + return port, err + } + GinkgoWriter.Printf("port %d lost a bind race, retrying once with fresh port %d: %v\n", port, newPort, err) + return newPort, attempt(newPort) +} + // NewServer creates and starts a new API server instance by running `thv serve` as a subprocess. func NewServer(config *ServerConfig) (*Server, error) { testConfig := NewTestConfig() @@ -87,54 +112,63 @@ func NewServer(config *ServerConfig) (*Server, error) { // would fail because no client config paths exist in the temp home dir. _ = os.WriteFile(filepath.Join(tempHome, ".claude.json"), []byte("{}"), 0600) - ctx, cancel := context.WithCancel(context.Background()) - - // Create string builders to capture output - var stdout, stderr strings.Builder - - // Create the command: thv serve --host 127.0.0.1 --port - //nolint:gosec // Intentional for e2e testing - cmd := exec.CommandContext( - ctx, - testConfig.THVBinary, - "serve", - "--host", - "127.0.0.1", - "--port", - strconv.Itoa(port), - ) - // Set environment variables including temporary config paths - cmd.Env = append([]string{ - "TOOLHIVE_DEV=true", - fmt.Sprintf("XDG_CONFIG_HOME=%s", tempXdgConfigHome), - fmt.Sprintf("HOME=%s", tempHome), - }, config.ExtraEnv...) - cmd.Stdout = &stdout - cmd.Stderr = &stderr - - // Start the server process - if err := cmd.Start(); err != nil { - cancel() - return nil, fmt.Errorf("failed to start thv serve: %w", err) - } + var server *Server + // The returned port is discarded here: the closure below assigns the + // real port onto server.port via the Server struct literal. + _, err = RetryOnPortConflict(port, func(p int) error { + ctx, cancel := context.WithCancel(context.Background()) + + // Create string builders to capture output + var stdout, stderr strings.Builder + + // Create the command: thv serve --host 127.0.0.1 --port + //nolint:gosec // Intentional for e2e testing + cmd := exec.CommandContext( + ctx, + testConfig.THVBinary, + "serve", + "--host", + "127.0.0.1", + "--port", + strconv.Itoa(p), + ) + // Set environment variables including temporary config paths + cmd.Env = append([]string{ + "TOOLHIVE_DEV=true", + fmt.Sprintf("XDG_CONFIG_HOME=%s", tempXdgConfigHome), + fmt.Sprintf("HOME=%s", tempHome), + }, config.ExtraEnv...) + cmd.Stdout = &stdout + cmd.Stderr = &stderr + + // Start the server process + if err := cmd.Start(); err != nil { + cancel() + return fmt.Errorf("failed to start thv serve: %w", err) + } - server := &Server{ - config: config, - baseURL: fmt.Sprintf("http://127.0.0.1:%d", port), - cmd: cmd, - ctx: ctx, - cancel: cancel, - httpClient: &http.Client{ - Timeout: config.RequestTimeout, - }, - port: port, - stdout: &stdout, - stderr: &stderr, - } + server = &Server{ + config: config, + baseURL: fmt.Sprintf("http://127.0.0.1:%d", p), + cmd: cmd, + ctx: ctx, + cancel: cancel, + httpClient: &http.Client{ + Timeout: config.RequestTimeout, + }, + port: p, + stdout: &stdout, + stderr: &stderr, + } - // Wait for server to be ready - if err := server.WaitForReady(); err != nil { - _ = server.Stop() + // Wait for server to be ready + if err := server.WaitForReady(); err != nil { + _ = server.Stop() + return err + } + return nil + }) + if err != nil { return nil, err } diff --git a/test/e2e/cli_llm_all_clients_test.go b/test/e2e/cli_llm_all_clients_test.go index 8a704af58d..faaae45342 100644 --- a/test/e2e/cli_llm_all_clients_test.go +++ b/test/e2e/cli_llm_all_clients_test.go @@ -254,12 +254,9 @@ var _ = Describe("thv llm — all-client matrix", Label("cli", "llm", "clients", binDir, err = e2e.CreateFakeBrowserDir(tempDir) Expect(err).ToNot(HaveOccurred()) - // Allocate a free port for the OIDC mock server. - oidcPort, err = networking.FindOrUsePort(0) - Expect(err).ToNot(HaveOccurred()) - - oidcServer, err = e2e.NewOIDCMockServer(oidcPort, clientID, clientSecret) + oidcServer, err = e2e.NewOIDCMockServer(0, clientID, clientSecret) Expect(err).ToNot(HaveOccurred()) + oidcPort = oidcServer.Port() oidcServer.EnableAutoComplete() Expect(oidcServer.Start()).To(Succeed()) @@ -541,39 +538,13 @@ var _ = Describe("thv llm — all-client matrix", Label("cli", "llm", "clients", proxyPort, portErr := networking.FindOrUsePort(0) Expect(portErr).ToNot(HaveOccurred()) - By(fmt.Sprintf("setting proxy port to %d", proxyPort)) - thvCmd("llm", "config", "set", "--proxy-port", fmt.Sprintf("%d", proxyPort)).ExpectSuccess() - - By("starting the proxy in a goroutine") - type proxyResult struct { - stdout, stderr string - err error - } - done := make(chan proxyResult, 1) - proxyCmd := thvCmd("llm", "proxy", "start") - go func() { - out, serr, rerr := proxyCmd.RunWithTimeout(15 * time.Second) - done <- proxyResult{out, serr, rerr} - }() - DeferCleanup(func() { - _ = proxyCmd.Interrupt() - select { - case <-done: - case <-time.After(5 * time.Second): - } + By(fmt.Sprintf("starting the proxy on port %d", proxyPort)) + _, portErr = e2e.RetryOnPortConflict(proxyPort, func(port int) error { + return startLLMProxy(thvCmd, func() *e2e.THVCommand { + return thvCmd("llm", "proxy", "start") + }, port, 15*time.Second) }) - - By(fmt.Sprintf("waiting for proxy to listen on port %d", proxyPort)) - proxyAddr := fmt.Sprintf("127.0.0.1:%d", proxyPort) - Eventually(func() error { - conn, err := net.DialTimeout("tcp", proxyAddr, 200*time.Millisecond) - if err != nil { - return err - } - _ = conn.Close() - return nil - }, 10*time.Second, 300*time.Millisecond).Should(Succeed(), - "proxy should be listening on %s", proxyAddr) + Expect(portErr).ToNot(HaveOccurred()) }) }) @@ -630,9 +601,7 @@ var _ = Describe("thv llm — all-client matrix", Label("cli", "llm", "clients", // Start a local HTTPS mock gateway so the proxy can forward requests // quickly rather than timing out on DNS resolution for a fake domain. - gwPort, portErr := networking.FindOrUsePort(0) - Expect(portErr).ToNot(HaveOccurred()) - gw, gwErr := e2e.NewLLMGatewayMock(gwPort) + gw, gwErr := e2e.NewLLMGatewayMock(0) Expect(gwErr).ToNot(HaveOccurred()) Expect(gw.Start()).To(Succeed()) defer func() { _ = gw.Stop() }() @@ -660,36 +629,17 @@ var _ = Describe("thv llm — all-client matrix", Label("cli", "llm", "clients", proxyPort, portErr2 := networking.FindOrUsePort(0) Expect(portErr2).ToNot(HaveOccurred()) - By(fmt.Sprintf("setting proxy port to %d and starting proxy", proxyPort)) - thvCmd("llm", "config", "set", "--proxy-port", fmt.Sprintf("%d", proxyPort)).ExpectSuccess() - - done := make(chan struct{}) - proxyCmd := thvCmd("llm", "proxy", "start").WithEnv( - "SSL_CERT_FILE="+gwCertFile, - rebindEnvKey+"="+rebindToken, - ) - go func() { - defer close(done) - _, _, _ = proxyCmd.RunWithTimeout(15 * time.Second) - }() - DeferCleanup(func() { - _ = proxyCmd.Interrupt() - select { - case <-done: - case <-time.After(5 * time.Second): - } + By(fmt.Sprintf("starting the proxy on port %d", proxyPort)) + proxyPort, portErr2 = e2e.RetryOnPortConflict(proxyPort, func(port int) error { + return startLLMProxy(thvCmd, func() *e2e.THVCommand { + return thvCmd("llm", "proxy", "start").WithEnv( + "SSL_CERT_FILE="+gwCertFile, + rebindEnvKey+"="+rebindToken, + ) + }, port, 15*time.Second) }) - + Expect(portErr2).ToNot(HaveOccurred()) proxyAddr := fmt.Sprintf("127.0.0.1:%d", proxyPort) - Eventually(func() error { - conn, dialErr := net.DialTimeout("tcp", proxyAddr, 200*time.Millisecond) - if dialErr != nil { - return dialErr - } - _ = conn.Close() - return nil - }, 10*time.Second, 300*time.Millisecond).Should(Succeed(), - "proxy should be listening on %s", proxyAddr) rebindClient := &http.Client{Timeout: 10 * time.Second} @@ -765,16 +715,15 @@ var _ = Describe("thv llm — all-client matrix", Label("cli", "llm", "clients", Skip(fmt.Sprintf("client %q not supported on %s", clientTC.name, runtime.GOOS)) } - // Allocate ports for the mock gateway and the proxy. - gatewayPort, portErr := networking.FindOrUsePort(0) - Expect(portErr).ToNot(HaveOccurred()) + // Allocate a port for the proxy. proxyPort, portErr := networking.FindOrUsePort(0) Expect(portErr).ToNot(HaveOccurred()) // Start the mock LLM gateway (HTTPS with self-signed cert). - By(fmt.Sprintf("[%s] starting mock LLM gateway on port %d", clientTC.name, gatewayPort)) - gateway, gwErr := e2e.NewLLMGatewayMock(gatewayPort) + gateway, gwErr := e2e.NewLLMGatewayMock(0) Expect(gwErr).ToNot(HaveOccurred()) + gatewayPort := gateway.Port() + By(fmt.Sprintf("[%s] starting mock LLM gateway on port %d", clientTC.name, gatewayPort)) Expect(gateway.Start()).To(Succeed()) defer func() { _ = gateway.Stop() }() @@ -822,34 +771,14 @@ var _ = Describe("thv llm — all-client matrix", Label("cli", "llm", "clients", } // Configure the proxy port and start it. - By(fmt.Sprintf("[%s] setting proxy port to %d", clientTC.name, proxyPort)) - thvCmd("llm", "config", "set", "--proxy-port", fmt.Sprintf("%d", proxyPort)).ExpectSuccess() - - By(fmt.Sprintf("[%s] starting the proxy", clientTC.name)) - done := make(chan struct{}) - proxyCmd := thvCmdWithToken("llm", "proxy", "start") - go func() { - defer close(done) - _, _, _ = proxyCmd.RunWithTimeout(20 * time.Second) - }() - DeferCleanup(func() { - _ = proxyCmd.Interrupt() - select { - case <-done: - case <-time.After(5 * time.Second): - } + By(fmt.Sprintf("[%s] starting the proxy on port %d", clientTC.name, proxyPort)) + proxyPort, portErr = e2e.RetryOnPortConflict(proxyPort, func(port int) error { + return startLLMProxy(thvCmd, func() *e2e.THVCommand { + return thvCmdWithToken("llm", "proxy", "start") + }, port, 20*time.Second) }) - + Expect(portErr).ToNot(HaveOccurred()) proxyAddr := fmt.Sprintf("127.0.0.1:%d", proxyPort) - Eventually(func() error { - conn, dialErr := net.DialTimeout("tcp", proxyAddr, 200*time.Millisecond) - if dialErr != nil { - return dialErr - } - _ = conn.Close() - return nil - }, 10*time.Second, 300*time.Millisecond).Should(Succeed(), - "proxy should be listening on %s", proxyAddr) // Send requests through the proxy and verify the gateway received them // with the correct Bearer token. Use a client with an explicit timeout @@ -1275,6 +1204,75 @@ var _ = Describe("thv llm — all-client matrix", Label("cli", "llm", "clients", // Helpers // ───────────────────────────────────────────────────────────────────────────── +// llmProxyResult captures a `thv llm proxy start` subprocess's outcome, so +// startLLMProxy can inspect it for an address-already-in-use signal if the +// subprocess exits before its listener comes up. +type llmProxyResult struct { + stdout, stderr string + err error +} + +// startLLMProxy sets the proxy port via `thv llm config set --proxy-port`, +// starts `thv llm proxy start` (built by buildProxyCmd, so each call site +// can inject its own env/token wiring) in the background, registers a +// DeferCleanup to interrupt it, and waits for it to accept TCP connections +// on 127.0.0.1:port. The port is only probed free at selection time +// (networking.FindOrUsePort); the real bind happens later in this +// subprocess, so another process can steal it in between. On that race, +// this returns an error containing "address already in use" so callers can +// retry via e2e.RetryOnPortConflict. +func startLLMProxy( + thvCmd func(args ...string) *e2e.THVCommand, + buildProxyCmd func() *e2e.THVCommand, + port int, + runTimeout time.Duration, +) error { + thvCmd("llm", "config", "set", "--proxy-port", fmt.Sprintf("%d", port)).ExpectSuccess() + + proxyCmd := buildProxyCmd() + done := make(chan llmProxyResult, 1) + go func() { + out, serr, rerr := proxyCmd.RunWithTimeout(runTimeout) + done <- llmProxyResult{out, serr, rerr} + }() + DeferCleanup(func() { + _ = proxyCmd.Interrupt() + select { + case <-done: + case <-time.After(5 * time.Second): + } + }) + + addr := fmt.Sprintf("127.0.0.1:%d", port) + deadline := time.After(10 * time.Second) + ticker := time.NewTicker(300 * time.Millisecond) + defer ticker.Stop() + for { + select { + case res := <-done: + return fmt.Errorf("proxy exited before listening on %s: err=%v stderr=%q", addr, res.err, res.stderr) + case <-deadline: + return fmt.Errorf("timed out waiting for proxy to listen on %s", addr) + case <-ticker.C: + conn, dialErr := net.DialTimeout("tcp", addr, 200*time.Millisecond) + if dialErr != nil { + continue + } + _ = conn.Close() + // A successful dial doesn't prove OUR subprocess is the listener -- if it + // lost the bind race, another process already listening on this exact + // port would also accept the connection. Give the subprocess's own exit + // (if any) a moment to land on `done` before trusting the dial. + select { + case res := <-done: + return fmt.Errorf("proxy exited before listening on %s: err=%v stderr=%q", addr, res.err, res.stderr) + case <-time.After(100 * time.Millisecond): + return nil + } + } + } +} + // createFakeBinary writes a minimal no-op shell script named `name` in dir. // This satisfies the LLMBinaryName check in DetectedLLMGatewayClients. func createFakeBinary(dir, name string) error { diff --git a/test/e2e/cli_llm_setup_test.go b/test/e2e/cli_llm_setup_test.go index 196ce7e6a3..56d6de13c6 100644 --- a/test/e2e/cli_llm_setup_test.go +++ b/test/e2e/cli_llm_setup_test.go @@ -18,7 +18,6 @@ import ( "gopkg.in/yaml.v3" "github.com/stacklok/toolhive/pkg/llm" - "github.com/stacklok/toolhive/pkg/networking" "github.com/stacklok/toolhive/test/e2e" ) @@ -160,15 +159,12 @@ var _ = Describe("thv llm setup / teardown", Label("cli", "llm", "setup", "e2e") By("Configuring environment secrets provider") thvCmd("secret", "provider", "environment").ExpectSuccess() - // Allocate a free port for the OIDC mock server. + // Create and start the mock OIDC server. var err error - oidcPort, err = networking.FindOrUsePort(0) + oidcServer, err = e2e.NewOIDCMockServer(0, clientID, clientSecret) Expect(err).ToNot(HaveOccurred()) - - // Create and start the mock OIDC server. + oidcPort = oidcServer.Port() By(fmt.Sprintf("Starting OIDC mock server on port %d", oidcPort)) - oidcServer, err = e2e.NewOIDCMockServer(oidcPort, clientID, clientSecret) - Expect(err).ToNot(HaveOccurred()) oidcServer.EnableAutoComplete() Expect(oidcServer.Start()).To(Succeed()) diff --git a/test/e2e/llm_gateway_mock.go b/test/e2e/llm_gateway_mock.go index f7fb27f208..a920103450 100644 --- a/test/e2e/llm_gateway_mock.go +++ b/test/e2e/llm_gateway_mock.go @@ -13,6 +13,7 @@ import ( "crypto/x509/pkix" "encoding/json" "encoding/pem" + "errors" "fmt" "math/big" "net" @@ -38,10 +39,11 @@ type GatewayRequest struct { // for plain HTTP. The HTTP variant is simpler for e2e tests where the thv // subprocess cannot be easily configured to trust a self-signed cert. type LLMGatewayMock struct { - server *http.Server - port int - useTLS bool - certPEM []byte // non-nil only when useTLS is true + server *http.Server + listener net.Listener + port int + useTLS bool + certPEM []byte // non-nil only when useTLS is true mu sync.Mutex requests []GatewayRequest @@ -50,18 +52,31 @@ type LLMGatewayMock struct { // NewLLMGatewayMock creates a mock LLM gateway that serves HTTPS with a // self-signed certificate. Use CertPEM / TLSClientConfig to build a trusting // HTTP client. Call Start to begin serving. +// +// The listener is bound before anything else, closing the window between +// port selection and bind that a separate "pick a port, then bind later" +// step would leave open. Pass port 0 to let the OS choose an ephemeral port; +// use Port() to read back the port actually bound. func NewLLMGatewayMock(port int) (*LLMGatewayMock, error) { + listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + return nil, fmt.Errorf("failed to listen: %w", err) + } + realPort := listener.Addr().(*net.TCPAddr).Port + certPEM, keyPEM, err := generateGatewayCert() if err != nil { + _ = listener.Close() return nil, err } cert, err := tls.X509KeyPair(certPEM, keyPEM) if err != nil { + _ = listener.Close() return nil, fmt.Errorf("loading TLS key pair: %w", err) } - m := &LLMGatewayMock{port: port, useTLS: true, certPEM: certPEM} + m := &LLMGatewayMock{listener: listener, port: realPort, useTLS: true, certPEM: certPEM} m.server = m.newServer() m.server.TLSConfig = &tls.Config{ Certificates: []tls.Certificate{cert}, @@ -71,6 +86,11 @@ func NewLLMGatewayMock(port int) (*LLMGatewayMock, error) { return m, nil } +// Port returns the port the mock gateway is bound to. +func (m *LLMGatewayMock) Port() int { + return m.port +} + func (m *LLMGatewayMock) newServer() *http.Server { mux := http.NewServeMux() mux.HandleFunc("/v1/models", m.handleModels) @@ -84,7 +104,6 @@ func (m *LLMGatewayMock) newServer() *http.Server { _, _ = w.Write([]byte("OK")) }) return &http.Server{ - Addr: fmt.Sprintf(":%d", m.port), Handler: mux, ReadHeaderTimeout: 10 * time.Second, } @@ -96,11 +115,11 @@ func (m *LLMGatewayMock) Start() error { go func() { var err error if m.useTLS { - err = m.server.ListenAndServeTLS("", "") + err = m.server.ServeTLS(m.listener, "", "") } else { - err = m.server.ListenAndServe() + err = m.server.Serve(m.listener) } - if err != nil && err != http.ErrServerClosed { + if err != nil && !errors.Is(err, http.ErrServerClosed) { errCh <- err } }() diff --git a/test/e2e/oidc_mock.go b/test/e2e/oidc_mock.go index af670d9f62..b7e3a1e4ad 100644 --- a/test/e2e/oidc_mock.go +++ b/test/e2e/oidc_mock.go @@ -11,6 +11,7 @@ import ( "encoding/json" "errors" "fmt" + "net" "net/http" "time" @@ -26,6 +27,7 @@ import ( // OIDCMockServer represents a lightweight OIDC server using Ory Fosite type OIDCMockServer struct { server *http.Server + listener net.Listener provider fosite.OAuth2Provider store *storage.MemoryStore port int @@ -71,14 +73,31 @@ func WithClientAudience(audiences ...string) OIDCMockOption { // NewOIDCMockServer creates a new OIDC mock server using Ory Fosite. // Use WithClientAudience to set client-level options and WithAccessTokenLifespan // for Fosite-level settings. Both option kinds may be mixed in a single call. +// +// The listener is bound before anything else, closing the window between +// port selection and bind that a separate "pick a port, then bind later" +// step would leave open. Pass port 0 to let the OS choose an ephemeral port; +// use Port() to read back the port actually bound. func NewOIDCMockServer(port int, clientID, clientSecret string, opts ...OIDCMockOption) (*OIDCMockServer, error) { - config := defaultFositeConfig(port) + listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + return nil, fmt.Errorf("failed to listen: %w", err) + } + realPort := listener.Addr().(*net.TCPAddr).Port + + config := defaultFositeConfig(realPort) for _, opt := range opts { if opt.fositeOpt != nil { opt.fositeOpt(config) } } - return newOIDCMockServer(port, clientID, clientSecret, config, opts...) + mockServer, err := newOIDCMockServer(realPort, clientID, clientSecret, config, opts...) + if err != nil { + _ = listener.Close() + return nil, err + } + mockServer.listener = listener + return mockServer, nil } // defaultFositeConfig returns the standard Fosite config for the mock server. @@ -174,7 +193,6 @@ func newOIDCMockServer( mockServer.setupRoutes(mux) mockServer.server = &http.Server{ - Addr: fmt.Sprintf(":%d", port), Handler: mux, ReadHeaderTimeout: 10 * time.Second, // Prevent Slowloris attacks } @@ -430,7 +448,7 @@ func (m *OIDCMockServer) handleJWKS(w http.ResponseWriter, _ *http.Request) { // Start starts the OIDC mock server func (m *OIDCMockServer) Start() error { go func() { - if err := m.server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + if err := m.server.Serve(m.listener); err != nil && !errors.Is(err, http.ErrServerClosed) { fmt.Printf("OIDC mock server error: %v\n", err) } }() @@ -440,6 +458,11 @@ func (m *OIDCMockServer) Start() error { return nil } +// Port returns the port the mock server is bound to. +func (m *OIDCMockServer) Port() int { + return m.port +} + // Stop stops the OIDC mock server func (m *OIDCMockServer) Stop() error { if m.server != nil { diff --git a/test/e2e/proxy_oauth_test.go b/test/e2e/proxy_oauth_test.go index 31d8bd793f..47bff7f8f3 100644 --- a/test/e2e/proxy_oauth_test.go +++ b/test/e2e/proxy_oauth_test.go @@ -15,12 +15,12 @@ import ( "regexp" "strconv" "strings" + "sync" "time" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - "github.com/stacklok/toolhive/pkg/networking" "github.com/stacklok/toolhive/test/e2e" ) @@ -54,27 +54,25 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" osvServerName = generateUniqueOIDCServerName("osv-oauth-target") proxyServerName = generateUniqueOIDCServerName("proxy-oauth-test") - // Find available ports for our mock servers using networking utilities - mockOIDCPort, err = networking.FindOrUsePort(0) - Expect(err).ToNot(HaveOccurred()) - - proxyPort, err = networking.FindOrUsePort(0) - Expect(err).ToNot(HaveOccurred()) - - mockOIDCBaseURL = fmt.Sprintf("http://localhost:%d", mockOIDCPort) + // proxyPort is discovered per-It from the `thv proxy` subprocess's own + // stdout once it binds -- see discoverProxyPort. It's started with + // port 0 rather than a pre-selected port to close the find-then-bind + // TOCTOU window. // Start mock OIDC server using Ory Fosite By("Starting mock OIDC server") specReport := CurrentSpecReport() if strings.Contains(specReport.FullText(), "Proxy OAuth Authentication E2E") { mockOIDCServer, err = e2e.NewOIDCMockServer( - mockOIDCPort, clientID, clientSecret, + 0, clientID, clientSecret, e2e.WithAccessTokenLifespan(2*time.Second), ) } else { - mockOIDCServer, err = e2e.NewOIDCMockServer(mockOIDCPort, clientID, clientSecret) + mockOIDCServer, err = e2e.NewOIDCMockServer(0, clientID, clientSecret) } Expect(err).ToNot(HaveOccurred()) + mockOIDCPort = mockOIDCServer.Port() + mockOIDCBaseURL = fmt.Sprintf("http://localhost:%d", mockOIDCPort) // Enable auto-complete for MCP tests mockOIDCServer.EnableAutoComplete() @@ -137,15 +135,20 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" base := fmt.Sprintf("%s://%s", parsedURL.Scheme, parsedURL.Host) By("Starting the proxy with OAuth configuration") - proxyCmd = startProxyWithOAuth( - config, - proxyServerName, - base, - proxyPort, - mockOIDCBaseURL, - clientID, - clientSecret, - ) + var out *syncBuffer + proxyCmd, out = startProxyWithOAuth(config, proxyServerName, base, 0, mockOIDCBaseURL, clientID, clientSecret) + // This test never drives the OAuth flow to completion (no + // WaitForAuthRequest/CompleteAuthRequest call), so the proxy may + // exit on --remote-auth-timeout before it ever binds a listener + // and prints the port line -- pre-existing behavior, unrelated to + // port selection. Discovery failure here is expected and + // harmless: the assertions below already tolerate the port being + // unset (proxyPort stays 0) or the proxy having exited. + if p, portErr := discoverProxyPort(out, 10*time.Second); portErr != nil { + GinkgoWriter.Printf("proxy port not discovered (OAuth flow may not have completed): %v\n", portErr) + } else { + proxyPort = p + } // Give the proxy some time to start and potentially complete OAuth flow time.Sleep(10 * time.Second) @@ -180,14 +183,11 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" base := fmt.Sprintf("%s://%s", parsedURL.Scheme, parsedURL.Host) By("Starting the proxy with OAuth auto-detection") - proxyCmd = startProxyWithOAuthDetection( - config, - proxyServerName, - base, - proxyPort, - clientID, - clientSecret, - ) + var out *syncBuffer + proxyCmd, out, proxyPort, err = startProxyAndDiscoverPort(func() (*exec.Cmd, *syncBuffer) { + return startProxyWithOAuthDetection(config, proxyServerName, base, 0, clientID, clientSecret) + }, 5*time.Second) + Expect(err).ToNot(HaveOccurred(), "proxy output: %s", out.String()) // Give the proxy time to start time.Sleep(5 * time.Second) @@ -212,15 +212,15 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" base := fmt.Sprintf("%s://%s", parsedURL.Scheme, parsedURL.Host) By("Starting the proxy with invalid OAuth credentials") - proxyCmd = startProxyWithOAuth( - config, - proxyServerName, - base, - proxyPort, - mockOIDCBaseURL, - "invalid-client", - "invalid-secret", - ) + var out *syncBuffer + proxyCmd, out = startProxyWithOAuth(config, proxyServerName, base, 0, mockOIDCBaseURL, "invalid-client", "invalid-secret") + // The proxy is expected to exit on the OAuth failure below before it + // ever binds a listener, so it never prints the port line -- this + // test doesn't use proxyPort, so a discovery failure is expected + // and harmless. + if _, portErr := discoverProxyPort(out, 2*time.Second); portErr != nil { + GinkgoWriter.Printf("proxy port not discovered (expected, proxy should exit on OAuth failure): %v\n", portErr) + } By("Verifying the proxy process exits due to OAuth failure") // The proxy should exit when OAuth fails due to invalid client credentials @@ -255,15 +255,16 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" base := fmt.Sprintf("%s://%s", parsedURL.Scheme, parsedURL.Host) By("Starting the proxy with missing OAuth issuer but remote-auth enabled") - proxyCmd = startProxyWithOAuth( - config, - proxyServerName, - base, - proxyPort, - "", // Empty issuer - clientID, - clientSecret, - ) + const emptyIssuer = "" + var out *syncBuffer + proxyCmd, out = startProxyWithOAuth(config, proxyServerName, base, 0, emptyIssuer, clientID, clientSecret) + // The proxy is expected to exit immediately below due to the + // missing issuer, before it ever binds a listener, so it never + // prints the port line -- this test doesn't use proxyPort, so a + // discovery failure is expected and harmless. + if _, portErr := discoverProxyPort(out, 2*time.Second); portErr != nil { + GinkgoWriter.Printf("proxy port not discovered (expected, proxy should exit due to missing issuer): %v\n", portErr) + } By("Verifying the proxy process exits due to missing issuer") // The proxy should exit immediately when --remote-auth is enabled but issuer is missing @@ -298,14 +299,11 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" base := fmt.Sprintf("%s://%s", parsedURL.Scheme, parsedURL.Host) By("Starting the proxy with auto-detection (no --remote-auth flag)") - proxyCmd = startProxyWithAutoDetection( - config, - proxyServerName, - base, - proxyPort, - clientID, - clientSecret, - ) + var out *syncBuffer + proxyCmd, out, proxyPort, err = startProxyAndDiscoverPort(func() (*exec.Cmd, *syncBuffer) { + return startProxyWithAutoDetection(config, proxyServerName, base, 0, clientID, clientSecret) + }, 5*time.Second) + Expect(err).ToNot(HaveOccurred(), "proxy output: %s", out.String()) // Give the proxy time to try auto-detection time.Sleep(5 * time.Second) @@ -330,50 +328,49 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" GinkgoWriter.Printf("Base URL for proxy: %s\n", baseURL) By("Starting the proxy with OAuth configuration and longer timeout") - var outputBuffer *bytes.Buffer - proxyCmd, outputBuffer = startProxyWithOAuthForMCP( - config, - proxyServerName, - baseURL, // Use base URL instead of full URL - proxyPort, - mockOIDCBaseURL, - clientID, - clientSecret, - ) - - By("Extracting OAuth URL from proxy output and completing the flow") - // Give the proxy a moment to start and display the OAuth URL - time.Sleep(5 * time.Second) - - // Extract OAuth URL from captured output - output := outputBuffer.String() - GinkgoWriter.Printf("Captured proxy output: %s\n", output) - - // Use regex to extract the OAuth URL - // Pattern: "Please open this URL in your browser: " - urlPattern := regexp.MustCompile(`Please open this URL in your browser: (https?://[^\s"]+)`) - matches := urlPattern.FindStringSubmatch(output) - - var authURL string - if len(matches) >= 2 { - authURL = matches[1] - GinkgoWriter.Printf("Extracted OAuth URL from buffer: %s\n", authURL) - } else { - // Fallback: construct the URL from what we know - // We can see the URL in the logs, so let's construct it - authURL = fmt.Sprintf("%s/auth?client_id=%s&response_type=code&scope=openid+profile+email", mockOIDCBaseURL, clientID) - GinkgoWriter.Printf("Using constructed OAuth URL: %s\n", authURL) - } - - // Complete the OAuth flow by visiting the URL with auto_complete parameter - err = completeOAuthFlow(authURL) - if err != nil { - GinkgoWriter.Printf("Failed to complete OAuth flow: %v\n", err) - Skip("Skipping MCP test due to OAuth flow completion failure") - } - - // Wait for proxy to complete OAuth and start - time.Sleep(5 * time.Second) + var outputBuffer *syncBuffer + // startProxyAndDiscoverPort retries the whole closure -- fresh subprocess, + // fresh port, fresh OAuth exchange -- if the previous attempt lost the + // real port-bind race (see the doc comment on discoverProxyPort). + proxyCmd, outputBuffer, proxyPort, err = startProxyAndDiscoverPort(func() (*exec.Cmd, *syncBuffer) { + // baseURL, not the full server URL -- the transparent proxy needs the + // base URL (see the comment above where it's derived). + cmd, out := startProxyWithOAuthForMCP(config, proxyServerName, baseURL, 0, mockOIDCBaseURL, clientID, clientSecret) + + By("Extracting OAuth URL from proxy output and completing the flow") + // Give the proxy a moment to start and display the OAuth URL + time.Sleep(5 * time.Second) + + // Extract OAuth URL from captured output + output := out.String() + GinkgoWriter.Printf("Captured proxy output: %s\n", output) + + // Use regex to extract the OAuth URL + // Pattern: "Please open this URL in your browser: " + urlPattern := regexp.MustCompile(`Please open this URL in your browser: (https?://[^\s"]+)`) + matches := urlPattern.FindStringSubmatch(output) + + var authURL string + if len(matches) >= 2 { + authURL = matches[1] + GinkgoWriter.Printf("Extracted OAuth URL from buffer: %s\n", authURL) + } else { + // Fallback: construct the URL from what we know + // We can see the URL in the logs, so let's construct it + authURL = fmt.Sprintf("%s/auth?client_id=%s&response_type=code&scope=openid+profile+email", mockOIDCBaseURL, clientID) + GinkgoWriter.Printf("Using constructed OAuth URL: %s\n", authURL) + } + + // Complete the OAuth flow by visiting the URL with auto_complete parameter + if flowErr := completeOAuthFlow(authURL); flowErr != nil { + GinkgoWriter.Printf("Failed to complete OAuth flow: %v\n", flowErr) + Skip("Skipping MCP test due to OAuth flow completion failure") + } + + return cmd, out + }, 15*time.Second) + By("Waiting for the proxy to complete OAuth and bind its listener") + Expect(err).ToNot(HaveOccurred(), "proxy output: %s", outputBuffer.String()) By("Testing MCP connection through proxy") proxyURL := fmt.Sprintf("http://localhost:%d/mcp", proxyPort) @@ -417,29 +414,27 @@ var _ = Describe("Proxy OAuth Authentication E2E", Label("proxy", "oauth", "e2e" GinkgoWriter.Printf("Base URL for proxy: %s\n", baseURL) By("Starting the proxy with OAuth-enabled MCP support") - var outputBuffer *bytes.Buffer - proxyCmd, outputBuffer = startProxyWithOAuthForMCP( - config, - proxyServerName, - baseURL, - proxyPort, - mockOIDCBaseURL, - clientID, - clientSecret, - ) - - By("Completing the initial OAuth flow") - Eventually(outputBuffer.String, 5*time.Second, 500*time.Millisecond). - Should(ContainSubstring("Please open this URL")) - - matches := regexp.MustCompile(`Please open this URL in your browser: (https?://[^\s"]+)`). - FindStringSubmatch(outputBuffer.String()) - Expect(matches).To(HaveLen(2)) - authURL := matches[1] - Expect(completeOAuthFlow(authURL)).To(Succeed()) - - By("Giving proxy time to finish OAuth exchange") - time.Sleep(2 * time.Second) + var outputBuffer *syncBuffer + // startProxyAndDiscoverPort retries the whole closure -- fresh subprocess, + // fresh port, fresh OAuth exchange -- if the previous attempt lost the + // real port-bind race (see the doc comment on discoverProxyPort). + proxyCmd, outputBuffer, proxyPort, err = startProxyAndDiscoverPort(func() (*exec.Cmd, *syncBuffer) { + cmd, out := startProxyWithOAuthForMCP(config, proxyServerName, baseURL, 0, mockOIDCBaseURL, clientID, clientSecret) + + By("Completing the initial OAuth flow") + Eventually(out.String, 5*time.Second, 500*time.Millisecond). + Should(ContainSubstring("Please open this URL")) + + matches := regexp.MustCompile(`Please open this URL in your browser: (https?://[^\s"]+)`). + FindStringSubmatch(out.String()) + Expect(matches).To(HaveLen(2)) + authURL := matches[1] + Expect(completeOAuthFlow(authURL)).To(Succeed()) + + return cmd, out + }, 10*time.Second) + By("Waiting for the proxy to finish the OAuth exchange and bind its listener") + Expect(err).ToNot(HaveOccurred(), "proxy output: %s", outputBuffer.String()) By("Waiting for access token to expire") time.Sleep(3 * time.Second) // longer than the 2s lifespan @@ -483,7 +478,108 @@ func checkServerHealth(healthUrl string) error { return fmt.Errorf("server not healthy, status: %d", resp.StatusCode) } -func startProxyWithOAuth(config *e2e.TestConfig, serverName, targetURL string, port int, issuer, clientID, clientSecret string) *exec.Cmd { +// syncBuffer is a bytes.Buffer guarded by a mutex, safe for concurrent +// writes (exec.Cmd's internal stdout/stderr copier goroutines) and reads +// (a polling goroutine like discoverProxyPort, or Eventually matchers) +// while the subprocess is still running. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (s *syncBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.Write(p) +} + +func (s *syncBuffer) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.String() +} + +// startLongRunningWithCapture is like e2e.StartLongRunningTHVCommand, but +// also captures stdout/stderr into a buffer. Callers inspect the buffer +// (e.g. via discoverProxyPort) rather than reaping the process via cmd.Wait +// -- cmd.Wait must never be called concurrently, and each test in this file +// that cares about the process's exit decides independently whether and +// when to call it. +func startLongRunningWithCapture(config *e2e.TestConfig, args ...string) (*exec.Cmd, *syncBuffer) { + cmd := exec.Command(config.THVBinary, args...) //nolint:gosec // Intentional for e2e testing + cmd.Env = os.Environ() + + out := &syncBuffer{} + multiWriter := io.MultiWriter(out, GinkgoWriter) + cmd.Stdout = multiWriter + cmd.Stderr = multiWriter + + Expect(cmd.Start()).To(Succeed()) + return cmd, out +} + +// proxyBoundPortPattern matches the port number `thv proxy` prints once its +// listener is actually bound, e.g. "... on port 54321 -> ..." (see +// cmd/thv/app/proxy.go's fmt.Printf after proxy.Start()). +var proxyBoundPortPattern = regexp.MustCompile(`on port (\d+)`) + +// discoverProxyPort polls out (the captured stdout/stderr of a `thv proxy` +// subprocess started with --port 0) for the line the subprocess prints once +// it has actually bound its listener, and returns the port it chose. +// +// Passing port 0 and reading the real port back afterwards -- rather than +// pre-selecting a port and handing it to the subprocess -- narrows the +// find-then-bind TOCTOU window but does not close it: FindOrUsePort(0) +// (cmd/thv/app/proxy.go) still does the classic probe-then-release, and the +// real bind happens later, inside transparent.NewTransparentProxy(...).Start(), +// after the OAuth exchange (handleOutgoingAuthentication) completes. Another +// process can still steal the port during that window; the window is just +// usually much shorter than the original pre-selected-port bug. See +// startProxyAndDiscoverPort for the retry that closes it at call sites that +// use it. +// +// It's expected (and safe to ignore) for this to time out on tests where the +// proxy process exits before ever binding, e.g. on an OAuth failure. +func discoverProxyPort(out *syncBuffer, timeout time.Duration) (int, error) { + deadline := time.Now().Add(timeout) + for { + if m := proxyBoundPortPattern.FindStringSubmatch(out.String()); m != nil { + return strconv.Atoi(m[1]) + } + if time.Now().After(deadline) { + return 0, fmt.Errorf("proxy port not found in output within %s", timeout) + } + time.Sleep(150 * time.Millisecond) + } +} + +// startProxyAndDiscoverPort starts a `thv proxy` subprocess via start, waits +// for it to report its real bound port via discoverProxyPort, and retries +// once (a fresh subprocess, drawing a fresh random port) if the failure was +// a lost port-bind race -- detected by the captured output containing +// "address already in use". A discovery failure for any OTHER reason (e.g. +// this call site's proxy is expected to exit before ever binding, such as +// on an OAuth-config failure) is returned as-is, uncorrected: the caller +// decides whether that's fatal or tolerable. +func startProxyAndDiscoverPort( + start func() (*exec.Cmd, *syncBuffer), + timeout time.Duration, +) (*exec.Cmd, *syncBuffer, int, error) { + cmd, out := start() + port, err := discoverProxyPort(out, timeout) + if err == nil { + return cmd, out, port, nil + } + if !strings.Contains(out.String(), "address already in use") { + return cmd, out, 0, err + } + GinkgoWriter.Printf("proxy lost a port-bind race, retrying once: %v\n", err) + cmd, out = start() + port, err = discoverProxyPort(out, timeout) + return cmd, out, port, err +} + +func startProxyWithOAuth(config *e2e.TestConfig, serverName, targetURL string, port int, issuer, clientID, clientSecret string) (*exec.Cmd, *syncBuffer) { args := []string{ "proxy", "--host", "localhost", @@ -513,10 +609,10 @@ func startProxyWithOAuth(config *e2e.TestConfig, serverName, targetURL string, p // Log the command for debugging GinkgoWriter.Printf("Starting proxy with args: %v\n", args) - return e2e.StartLongRunningTHVCommand(config, args...) + return startLongRunningWithCapture(config, args...) } -func startProxyWithOAuthDetection(config *e2e.TestConfig, serverName, targetURL string, port int, clientID, clientSecret string) *exec.Cmd { +func startProxyWithOAuthDetection(config *e2e.TestConfig, serverName, targetURL string, port int, clientID, clientSecret string) (*exec.Cmd, *syncBuffer) { args := []string{ "proxy", "--host", "localhost", @@ -528,10 +624,10 @@ func startProxyWithOAuthDetection(config *e2e.TestConfig, serverName, targetURL serverName, } - return e2e.StartLongRunningTHVCommand(config, args...) + return startLongRunningWithCapture(config, args...) } -func startProxyWithAutoDetection(config *e2e.TestConfig, serverName, targetURL string, port int, clientID, clientSecret string) *exec.Cmd { +func startProxyWithAutoDetection(config *e2e.TestConfig, serverName, targetURL string, port int, clientID, clientSecret string) (*exec.Cmd, *syncBuffer) { args := []string{ "proxy", "--host", "localhost", @@ -546,10 +642,10 @@ func startProxyWithAutoDetection(config *e2e.TestConfig, serverName, targetURL s // Log the command for debugging GinkgoWriter.Printf("Starting proxy with auto-detection args: %v\n", args) - return e2e.StartLongRunningTHVCommand(config, args...) + return startLongRunningWithCapture(config, args...) } -func startProxyWithOAuthForMCP(config *e2e.TestConfig, serverName, targetURL string, port int, issuer, clientID, clientSecret string) (*exec.Cmd, *bytes.Buffer) { +func startProxyWithOAuthForMCP(config *e2e.TestConfig, serverName, targetURL string, port int, issuer, clientID, clientSecret string) (*exec.Cmd, *syncBuffer) { args := []string{ "proxy", "--host", "localhost", @@ -567,23 +663,7 @@ func startProxyWithOAuthForMCP(config *e2e.TestConfig, serverName, targetURL str // Log the command for debugging GinkgoWriter.Printf("Starting proxy with OAuth for MCP args: %v\n", args) - // Create command - cmd := exec.Command(config.THVBinary, args...) - cmd.Env = os.Environ() - - // Create buffer to capture output (capture both stdout and stderr) - var outputBuffer bytes.Buffer - - // Use MultiWriter to write to both buffer and GinkgoWriter - multiWriter := io.MultiWriter(&outputBuffer, GinkgoWriter) - cmd.Stdout = multiWriter - cmd.Stderr = multiWriter // Capture stderr too since logger might write there - - // Start the command - err := cmd.Start() - Expect(err).ToNot(HaveOccurred()) - - return cmd, &outputBuffer + return startLongRunningWithCapture(config, args...) } // completeOAuthFlow programmatically completes the OAuth flow by visiting the authorization URL diff --git a/test/e2e/vmcp_infra_features_test.go b/test/e2e/vmcp_infra_features_test.go index 3e5c7415b1..8450402324 100644 --- a/test/e2e/vmcp_infra_features_test.go +++ b/test/e2e/vmcp_infra_features_test.go @@ -104,12 +104,12 @@ var _ = Describe("vMCP infra features", Label("vmcp", "e2e", "infra"), func() { BeforeEach(func() { fx.setup("vmcp-auth-oidc", "vmcp-auth-oidc-*") - oidcPort = allocateVMCPPort() var err error - oidcServer, err = e2e.NewOIDCMockServer(oidcPort, "test-client", "test-secret", + oidcServer, err = e2e.NewOIDCMockServer(0, "test-client", "test-secret", e2e.WithClientAudience("vmcp-e2e-test"), ) Expect(err).ToNot(HaveOccurred()) + oidcPort = oidcServer.Port() Expect(oidcServer.Start()).To(Succeed()) discoveryURL := fmt.Sprintf("http://localhost:%d/.well-known/openid-configuration", oidcPort) Eventually(func() error {