diff --git a/cmd/thv-operator/api/v1beta1/mcpserver_types.go b/cmd/thv-operator/api/v1beta1/mcpserver_types.go index 3a176cb476..e42d099c77 100644 --- a/cmd/thv-operator/api/v1beta1/mcpserver_types.go +++ b/cmd/thv-operator/api/v1beta1/mcpserver_types.go @@ -426,6 +426,15 @@ type MCPServerSpec struct { // Requires Redis session storage to be configured for distributed rate limiting. // +optional RateLimiting *ratelimittypes.RateLimitConfig `json:"rateLimiting,omitempty"` + + // ProxyReadTimeout bounds how long the proxy spends reading a full request + // (headers + body), mitigating slow-upload connection exhaustion. Applies to + // all transports. Defaults to 30s if not specified. Example: "1m". + // +kubebuilder:validation:Type=string + // +kubebuilder:validation:Format=duration + // +kubebuilder:validation:XValidation:rule="duration(self) >= duration('0s')",message="proxyReadTimeout must be non-negative" + // +optional + ProxyReadTimeout *metav1.Duration `json:"proxyReadTimeout,omitempty"` } // ResourceOverrides defines overrides for annotations and labels on created resources diff --git a/cmd/thv-operator/api/v1beta1/zz_generated.deepcopy.go b/cmd/thv-operator/api/v1beta1/zz_generated.deepcopy.go index c6362460aa..e022626570 100644 --- a/cmd/thv-operator/api/v1beta1/zz_generated.deepcopy.go +++ b/cmd/thv-operator/api/v1beta1/zz_generated.deepcopy.go @@ -1968,6 +1968,11 @@ func (in *MCPServerSpec) DeepCopyInto(out *MCPServerSpec) { *out = new(types.RateLimitConfig) (*in).DeepCopyInto(*out) } + if in.ProxyReadTimeout != nil { + in, out := &in.ProxyReadTimeout, &out.ProxyReadTimeout + *out = new(v1.Duration) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MCPServerSpec. diff --git a/cmd/thv-operator/controllers/mcpserver_runconfig.go b/cmd/thv-operator/controllers/mcpserver_runconfig.go index a98a57b73f..dcd6b26141 100644 --- a/cmd/thv-operator/controllers/mcpserver_runconfig.go +++ b/cmd/thv-operator/controllers/mcpserver_runconfig.go @@ -278,6 +278,11 @@ func (r *MCPServerReconciler) createRunConfigFromMCPServer(m *mcpv1beta1.MCPServ options = append(options, runner.WithRateLimitConfig(m.Namespace, m.Spec.RateLimiting)) } + // Add proxy HTTP server read timeout if specified + if m.Spec.ProxyReadTimeout != nil { + options = append(options, runner.WithProxyReadTimeout(m.Spec.ProxyReadTimeout.Duration)) + } + // Use the RunConfigBuilder for operator context with full builder pattern runConfig, err := runner.NewOperatorRunConfigBuilder( context.Background(), diff --git a/cmd/thv-operator/controllers/mcpserver_runconfig_test.go b/cmd/thv-operator/controllers/mcpserver_runconfig_test.go index 5be98c8f6d..f338f8052e 100644 --- a/cmd/thv-operator/controllers/mcpserver_runconfig_test.go +++ b/cmd/thv-operator/controllers/mcpserver_runconfig_test.go @@ -9,6 +9,7 @@ import ( "fmt" "reflect" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -61,6 +62,36 @@ func TestCreateRunConfigFromMCPServer(t *testing.T) { assert.Equal(t, 8080, config.Port) }, }, + { + name: "nil proxy read timeout leaves the RunConfig value empty", + mcpServer: v1beta1test.NewMCPServer("nil-timeout-server", "test-ns"), + //nolint:thelper // We want to see the error at the specific line + expected: func(t *testing.T, config *runner.RunConfig) { + assert.Empty(t, config.ProxyReadTimeout) + }, + }, + { + name: "zero proxy read timeout uses the proxy default", + mcpServer: v1beta1test.NewMCPServer("zero-timeout-server", "test-ns", + v1beta1test.Mutate(func(m *mcpv1beta1.MCPServer) { + m.Spec.ProxyReadTimeout = &metav1.Duration{} + })), + //nolint:thelper // We want to see the error at the specific line + expected: func(t *testing.T, config *runner.RunConfig) { + assert.Empty(t, config.ProxyReadTimeout) + }, + }, + { + name: "positive proxy read timeout is translated", + mcpServer: v1beta1test.NewMCPServer("positive-timeout-server", "test-ns", + v1beta1test.Mutate(func(m *mcpv1beta1.MCPServer) { + m.Spec.ProxyReadTimeout = &metav1.Duration{Duration: time.Minute} + })), + //nolint:thelper // We want to see the error at the specific line + expected: func(t *testing.T, config *runner.RunConfig) { + assert.Equal(t, "1m0s", config.ProxyReadTimeout) + }, + }, { name: "with environment variables", mcpServer: v1beta1test.NewMCPServer("env-server", "test-ns", diff --git a/cmd/thv-operator/test-integration/mcp-server/mcpserver_sessionstorage_cel_test.go b/cmd/thv-operator/test-integration/mcp-server/mcpserver_sessionstorage_cel_test.go index 508af923f1..5184466093 100644 --- a/cmd/thv-operator/test-integration/mcp-server/mcpserver_sessionstorage_cel_test.go +++ b/cmd/thv-operator/test-integration/mcp-server/mcpserver_sessionstorage_cel_test.go @@ -4,6 +4,8 @@ package controllers import ( + "time" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -117,4 +119,25 @@ var _ = Describe("CEL Validation for SessionStorageConfig on MCPServer", Expect(err).To(HaveOccurred()) }) }) + + Context("proxyReadTimeout field", func() { + DescribeTable("should accept non-negative values", + func(name string, timeout *metav1.Duration) { + server := newMinimalMCPServer(name, nil) + server.Spec.ProxyReadTimeout = timeout + err := k8sClient.Create(ctx, server) + Expect(err).NotTo(HaveOccurred()) + }, + Entry("when omitted", "mcp-proxy-read-timeout-omitted", nil), + Entry("when zero", "mcp-proxy-read-timeout-zero", &metav1.Duration{}), + Entry("when positive", "mcp-proxy-read-timeout-positive", &metav1.Duration{Duration: 45 * time.Second}), + ) + + It("should reject a negative value", func() { + server := newMinimalMCPServer("mcp-proxy-read-timeout-negative", nil) + server.Spec.ProxyReadTimeout = &metav1.Duration{Duration: -time.Second} + err := k8sClient.Create(ctx, server) + Expect(err).To(MatchError(ContainSubstring("proxyReadTimeout must be non-negative"))) + }) + }) }) diff --git a/cmd/thv/app/run_flags.go b/cmd/thv/app/run_flags.go index 31f703ef3c..25efeb20ad 100644 --- a/cmd/thv/app/run_flags.go +++ b/cmd/thv/app/run_flags.go @@ -124,6 +124,9 @@ type RunFlags struct { // MaxRequestBodySize is the maximum inbound request body size in bytes. Zero uses the default. MaxRequestBodySize int64 + // ProxyReadTimeout bounds reading a full request on the proxy. Zero uses the default. + ProxyReadTimeout time.Duration + // Network mode Network string @@ -315,6 +318,8 @@ func AddRunFlags(cmd *cobra.Command, config *RunFlags) { "Session inactivity timeout (e.g., 30m, 2h); zero uses the default (2h)") cmd.Flags().Int64Var(&config.MaxRequestBodySize, "max-request-body-size", 0, "Maximum inbound request body size in bytes; zero uses the default (8 MiB)") + cmd.Flags().DurationVar(&config.ProxyReadTimeout, "proxy-read-timeout", 0, + "Maximum time to read a full request on the proxy (e.g., 30s, 1m); zero uses the default (30s)") cmd.Flags().StringVar(&config.EndpointPrefix, "endpoint-prefix", "", "Path prefix to prepend to SSE endpoint URLs (e.g., /playwright)") cmd.Flags().StringVar(&config.Network, "network", "", @@ -741,6 +746,7 @@ func buildRunnerConfig( runner.WithStateless(runFlags.Stateless), runner.WithSessionTTL(runFlags.SessionTTL), runner.WithMaxRequestBodySize(runFlags.MaxRequestBodySize), + runner.WithProxyReadTimeout(runFlags.ProxyReadTimeout), runner.WithEndpointPrefix(runFlags.EndpointPrefix), runner.WithNetworkMode(runFlags.Network), runner.WithK8sPodPatch(runFlags.K8sPodPatch), diff --git a/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_mcpservers.yaml b/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_mcpservers.yaml index af6a2ef9c7..d996ca9190 100644 --- a/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_mcpservers.yaml +++ b/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_mcpservers.yaml @@ -397,6 +397,16 @@ spec: maximum: 65535 minimum: 1 type: integer + proxyReadTimeout: + description: |- + ProxyReadTimeout bounds how long the proxy spends reading a full request + (headers + body), mitigating slow-upload connection exhaustion. Applies to + all transports. Defaults to 30s if not specified. Example: "1m". + format: duration + type: string + x-kubernetes-validations: + - message: proxyReadTimeout must be non-negative + rule: duration(self) >= duration('0s') rateLimiting: description: |- RateLimiting defines rate limiting configuration for the MCP server. @@ -2331,6 +2341,16 @@ spec: maximum: 65535 minimum: 1 type: integer + proxyReadTimeout: + description: |- + ProxyReadTimeout bounds how long the proxy spends reading a full request + (headers + body), mitigating slow-upload connection exhaustion. Applies to + all transports. Defaults to 30s if not specified. Example: "1m". + format: duration + type: string + x-kubernetes-validations: + - message: proxyReadTimeout must be non-negative + rule: duration(self) >= duration('0s') rateLimiting: description: |- RateLimiting defines rate limiting configuration for the MCP server. diff --git a/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_mcpservers.yaml b/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_mcpservers.yaml index b729e93201..dc02baeabf 100644 --- a/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_mcpservers.yaml +++ b/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_mcpservers.yaml @@ -400,6 +400,16 @@ spec: maximum: 65535 minimum: 1 type: integer + proxyReadTimeout: + description: |- + ProxyReadTimeout bounds how long the proxy spends reading a full request + (headers + body), mitigating slow-upload connection exhaustion. Applies to + all transports. Defaults to 30s if not specified. Example: "1m". + format: duration + type: string + x-kubernetes-validations: + - message: proxyReadTimeout must be non-negative + rule: duration(self) >= duration('0s') rateLimiting: description: |- RateLimiting defines rate limiting configuration for the MCP server. @@ -2334,6 +2344,16 @@ spec: maximum: 65535 minimum: 1 type: integer + proxyReadTimeout: + description: |- + ProxyReadTimeout bounds how long the proxy spends reading a full request + (headers + body), mitigating slow-upload connection exhaustion. Applies to + all transports. Defaults to 30s if not specified. Example: "1m". + format: duration + type: string + x-kubernetes-validations: + - message: proxyReadTimeout must be non-negative + rule: duration(self) >= duration('0s') rateLimiting: description: |- RateLimiting defines rate limiting configuration for the MCP server. diff --git a/docs/arch/03-transport-architecture.md b/docs/arch/03-transport-architecture.md index 05e03b4909..e4e841de46 100644 --- a/docs/arch/03-transport-architecture.md +++ b/docs/arch/03-transport-architecture.md @@ -366,7 +366,9 @@ export TOOLHIVE_PROXY_REQUEST_TIMEOUT=5m thv run my-slow-server ``` -**Note:** This timeout only affects the streamable HTTP proxy used with stdio transport. The transparent proxy used by SSE and streamable-http transports (where the container runs its own HTTP server) does not impose a request timeout. +**Note:** This MCP response-correlation timeout only affects the streamable HTTP proxy used with stdio transport. +The transparent proxy used by SSE and streamable-http transports (where the container runs its own HTTP server) +does not impose an MCP response-correlation timeout, but it does enforce the HTTP request read timeout described below. ### Proxy Request Body Size Limit (All Proxy Transports) @@ -386,6 +388,19 @@ auth server's 64 KiB cap, and vMCP's 8 MiB cap are unaffected. **Implementation**: `pkg/bodylimit`, `pkg/runner/middleware.go` +### Proxy Request Read Timeout (All Transports) + +Every proxy HTTP server limits reading a complete inbound request, including its body, to 30 seconds by default. +This prevents a slow or stalled upload from holding a connection open indefinitely. +Operators can override the limit per workload with `thv run --proxy-read-timeout` or the MCPServer `spec.proxyReadTimeout` field. +RunConfig stores the same setting as `proxy_read_timeout`, using a Go duration string such as `45s` or `2m`. + +Omitting the setting or specifying zero retains the 30-second default; it never disables the timeout. +The read timeout does not limit response streaming, so long-lived SSE responses remain unaffected. + +This setting is distinct from `TOOLHIVE_PROXY_REQUEST_TIMEOUT` above: the read timeout bounds the client-to-proxy HTTP upload, +while the stdio proxy request timeout bounds how long an MCP request waits for its correlated server response. + ### Health Check Tuning Parameters **Implementation**: `pkg/transport/proxy/transparent/transparent_proxy.go` diff --git a/docs/arch/05-runconfig-and-permissions.md b/docs/arch/05-runconfig-and-permissions.md index ae81d92b01..d1b91b1060 100644 --- a/docs/arch/05-runconfig-and-permissions.md +++ b/docs/arch/05-runconfig-and-permissions.md @@ -125,7 +125,8 @@ thv run uvx://mcp-server \ "host": "127.0.0.1", "port": 8080, "proxy_mode": "streamable-http", - "max_request_body_size": 8388608 + "max_request_body_size": 8388608, + "proxy_read_timeout": "45s" } ``` @@ -148,6 +149,7 @@ thv run uvx://mcp-server \ - `target_host`: Container host (default: `127.0.0.1`) - `proxy_mode`: For stdio: `sse` or `streamable-http` - `max_request_body_size`: Maximum inbound MCP request body size in bytes; omitted or zero uses the 8 MiB default +- `proxy_read_timeout`: Maximum time to read a complete client request, as a Go duration string; omitted or zero uses the secure 30-second default **Implementation**: `pkg/runner/config.go` diff --git a/docs/cli/thv_run.md b/docs/cli/thv_run.md index 87fa8800f3..17b6c461a8 100644 --- a/docs/cli/thv_run.md +++ b/docs/cli/thv_run.md @@ -160,6 +160,7 @@ thv run [flags] SERVER_OR_IMAGE_OR_PROTOCOL [-- ARGS...] --print-resolved-overlays Debug: show resolved container paths for tmpfs overlays (default false) --proxy-mode string Proxy mode for stdio (streamable-http or sse (deprecated, will be removed)) (default "streamable-http") --proxy-port int Port for the HTTP proxy to listen on (host port) + --proxy-read-timeout duration Maximum time to read a full request on the proxy (e.g., 30s, 1m); zero uses the default (30s) -p, --publish stringArray Publish a container's port(s) to the host (format: hostPort:containerPort) --remote-auth Enable OAuth/OIDC authentication to remote MCP server (default false) --remote-auth-authorize-url string OAuth authorization endpoint URL (alternative to --remote-auth-issuer for non-OIDC OAuth) diff --git a/docs/operator/crd-api.md b/docs/operator/crd-api.md index 99a4ce5f0d..12b0c6e172 100644 --- a/docs/operator/crd-api.md +++ b/docs/operator/crd-api.md @@ -3358,6 +3358,7 @@ _Appears in:_ | `backendReplicas` _integer_ | BackendReplicas is the desired number of MCP server backend pod replicas.
This controls the backend Deployment (the MCP server container itself),
independent of the proxy runner controlled by Replicas.
When nil, the operator does not set Deployment.Spec.Replicas, leaving replica
management to an HPA or other external controller. | | Minimum: 0
Optional: \{\}
| | `sessionStorage` _[api.v1beta1.SessionStorageConfig](#apiv1beta1sessionstorageconfig)_ | SessionStorage configures session storage for stateful horizontal scaling.
When nil, no session storage is configured. | | Optional: \{\}
| | `rateLimiting` _[ratelimit.types.RateLimitConfig](#ratelimittypesratelimitconfig)_ | RateLimiting defines rate limiting configuration for the MCP server.
Requires Redis session storage to be configured for distributed rate limiting. | | Optional: \{\}
| +| `proxyReadTimeout` _[Duration](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#duration-v1-meta)_ | ProxyReadTimeout bounds how long the proxy spends reading a full request
(headers + body), mitigating slow-upload connection exhaustion. Applies to
all transports. Defaults to 30s if not specified. Example: "1m". | | Format: duration
Type: string
Optional: \{\}
| #### api.v1beta1.MCPServerStatus diff --git a/docs/server/docs.go b/docs/server/docs.go index 8789b61ed5..b78f0891a2 100644 --- a/docs/server/docs.go +++ b/docs/server/docs.go @@ -1797,6 +1797,11 @@ const docTemplate = `{ ], "type": "string" }, + "proxy_read_timeout": { + "description": "ProxyReadTimeout bounds reading the entire request (headers + body) on the\nproxy HTTP server, expressed as a Go duration string (e.g. \"30s\", \"1m\").\nEmpty uses the proxy default (30s). Negative durations and values that fail\ntime.ParseDuration are rejected at runtime. Applies to all HTTP transports.\nString (not time.Duration) keeps the wire format unit-explicit.", + "example": "30s", + "type": "string" + }, "publish": { "description": "Publish lists ports to publish to the host in format \"hostPort:containerPort\"", "items": { @@ -3327,6 +3332,11 @@ const docTemplate = `{ "description": "Port for the HTTP proxy to listen on", "type": "integer" }, + "proxy_read_timeout": { + "description": "Maximum time to read a complete MCP proxy request, expressed as a Go duration string.\nEmpty or zero uses the default timeout of 30 seconds.", + "example": "30s", + "type": "string" + }, "registry": { "description": "Registry is the optional registry name to resolve the server from (e.g. \"default\").", "type": "string" @@ -4239,6 +4249,11 @@ const docTemplate = `{ "description": "Port for the HTTP proxy to listen on", "type": "integer" }, + "proxy_read_timeout": { + "description": "Maximum time to read a complete MCP proxy request, expressed as a Go duration string.\nEmpty or zero uses the default timeout of 30 seconds.", + "example": "30s", + "type": "string" + }, "runtime_config": { "$ref": "#/components/schemas/templates.RuntimeConfig" }, diff --git a/docs/server/swagger.json b/docs/server/swagger.json index b51392f2cc..f46692dea2 100644 --- a/docs/server/swagger.json +++ b/docs/server/swagger.json @@ -1790,6 +1790,11 @@ ], "type": "string" }, + "proxy_read_timeout": { + "description": "ProxyReadTimeout bounds reading the entire request (headers + body) on the\nproxy HTTP server, expressed as a Go duration string (e.g. \"30s\", \"1m\").\nEmpty uses the proxy default (30s). Negative durations and values that fail\ntime.ParseDuration are rejected at runtime. Applies to all HTTP transports.\nString (not time.Duration) keeps the wire format unit-explicit.", + "example": "30s", + "type": "string" + }, "publish": { "description": "Publish lists ports to publish to the host in format \"hostPort:containerPort\"", "items": { @@ -3320,6 +3325,11 @@ "description": "Port for the HTTP proxy to listen on", "type": "integer" }, + "proxy_read_timeout": { + "description": "Maximum time to read a complete MCP proxy request, expressed as a Go duration string.\nEmpty or zero uses the default timeout of 30 seconds.", + "example": "30s", + "type": "string" + }, "registry": { "description": "Registry is the optional registry name to resolve the server from (e.g. \"default\").", "type": "string" @@ -4232,6 +4242,11 @@ "description": "Port for the HTTP proxy to listen on", "type": "integer" }, + "proxy_read_timeout": { + "description": "Maximum time to read a complete MCP proxy request, expressed as a Go duration string.\nEmpty or zero uses the default timeout of 30 seconds.", + "example": "30s", + "type": "string" + }, "runtime_config": { "$ref": "#/components/schemas/templates.RuntimeConfig" }, diff --git a/docs/server/swagger.yaml b/docs/server/swagger.yaml index 25b9670a3a..86a72b7d0a 100644 --- a/docs/server/swagger.yaml +++ b/docs/server/swagger.yaml @@ -1899,6 +1899,15 @@ components: - sse - streamable-http type: string + proxy_read_timeout: + description: |- + ProxyReadTimeout bounds reading the entire request (headers + body) on the + proxy HTTP server, expressed as a Go duration string (e.g. "30s", "1m"). + Empty uses the proxy default (30s). Negative durations and values that fail + time.ParseDuration are rejected at runtime. Applies to all HTTP transports. + String (not time.Duration) keeps the wire format unit-explicit. + example: 30s + type: string publish: description: Publish lists ports to publish to the host in format "hostPort:containerPort" items: @@ -3164,6 +3173,12 @@ components: proxy_port: description: Port for the HTTP proxy to listen on type: integer + proxy_read_timeout: + description: |- + Maximum time to read a complete MCP proxy request, expressed as a Go duration string. + Empty or zero uses the default timeout of 30 seconds. + example: 30s + type: string registry: description: Registry is the optional registry name to resolve the server from (e.g. "default"). @@ -3912,6 +3927,12 @@ components: proxy_port: description: Port for the HTTP proxy to listen on type: integer + proxy_read_timeout: + description: |- + Maximum time to read a complete MCP proxy request, expressed as a Go duration string. + Empty or zero uses the default timeout of 30 seconds. + example: 30s + type: string runtime_config: $ref: '#/components/schemas/templates.RuntimeConfig' secrets: diff --git a/pkg/api/v1/workload_service.go b/pkg/api/v1/workload_service.go index a44429d90f..a62ad3a524 100644 --- a/pkg/api/v1/workload_service.go +++ b/pkg/api/v1/workload_service.go @@ -386,6 +386,17 @@ func (s *WorkloadService) BuildFullRunConfig( } }() + proxyReadTimeout, parseErr := func() (time.Duration, error) { + if req.ProxyReadTimeout == "" { + return 0, nil + } + return time.ParseDuration(req.ProxyReadTimeout) + }() + if parseErr != nil { + return nil, fmt.Errorf("%w: invalid proxy_read_timeout %q: %w", + retriever.ErrInvalidRunConfig, req.ProxyReadTimeout, parseErr) + } + options := []runner.RunConfigBuilderOption{ runner.WithRuntime(s.containerRuntime), runner.WithCmdArgs(req.CmdArguments), @@ -409,6 +420,7 @@ func (s *WorkloadService) BuildFullRunConfig( runner.WithProxyMode(types.ProxyMode(req.ProxyMode)), runner.WithTransportAndPorts(req.Transport, req.ProxyPort, req.TargetPort), runner.WithMaxRequestBodySize(req.MaxRequestBodySize), + runner.WithProxyReadTimeout(proxyReadTimeout), runner.WithAuditEnabled(false, ""), runner.WithTokenValidatorConfig(apiOIDCConfig), runner.WithToolsFilter(req.ToolsFilter), diff --git a/pkg/api/v1/workload_service_test.go b/pkg/api/v1/workload_service_test.go index 14e86f0034..2476163590 100644 --- a/pkg/api/v1/workload_service_test.go +++ b/pkg/api/v1/workload_service_test.go @@ -448,6 +448,58 @@ func TestBuildFullRunConfig_MaxRequestBodySize(t *testing.T) { } } +func TestBuildFullRunConfig_ProxyReadTimeout(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + readTimeout string + wantTimeout string + wantErrorMsg string + }{ + {name: "empty preserves default semantics"}, + {name: "positive value is preserved", readTimeout: "45s", wantTimeout: "45s"}, + {name: "duration is normalized", readTimeout: "1m", wantTimeout: "1m0s"}, + {name: "invalid duration is rejected", readTimeout: "not-a-duration", wantErrorMsg: "proxy_read_timeout"}, + {name: "negative duration is rejected", readTimeout: "-1s", wantErrorMsg: "must be non-negative"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockGroupManager := groupsmocks.NewMockManager(ctrl) + mockGroupManager.EXPECT().Exists(gomock.Any(), "default").Return(true, nil) + + service := &WorkloadService{ + groupManager: mockGroupManager, + configProvider: config.NewDefaultProvider(), + } + req := &createRequest{ + Name: "testserver", + updateRequest: updateRequest{ + URL: "https://mcp.example.com/mcp", + ProxyReadTimeout: tt.readTimeout, + }, + } + + runConfig, err := service.BuildFullRunConfig(context.Background(), req, 0, nil) + if tt.wantErrorMsg != "" { + require.Error(t, err) + assert.Nil(t, runConfig) + assert.Contains(t, err.Error(), tt.wantErrorMsg) + return + } + + require.NoError(t, err) + assert.Equal(t, tt.wantTimeout, runConfig.ProxyReadTimeout) + }) + } +} + // TestBuildFullRunConfig_NoOtelConfigLeavesTelemetryNil verifies that when no // OpenTelemetry config is set, the RunConfig's TelemetryConfig remains nil — // the API path does not invent an endpoint. diff --git a/pkg/api/v1/workload_types.go b/pkg/api/v1/workload_types.go index 09adcbfcf1..02dab65425 100644 --- a/pkg/api/v1/workload_types.go +++ b/pkg/api/v1/workload_types.go @@ -60,6 +60,9 @@ type updateRequest struct { ProxyPort int `json:"proxy_port"` // Maximum inbound MCP proxy request body size in bytes. Zero uses the default limit of 8 MiB. MaxRequestBodySize int64 `json:"max_request_body_size,omitempty"` + // Maximum time to read a complete MCP proxy request, expressed as a Go duration string. + // Empty or zero uses the default timeout of 30 seconds. + ProxyReadTimeout string `json:"proxy_read_timeout,omitempty" example:"30s"` // Environment variables to set in the container EnvVars map[string]string `json:"env_vars"` // Secret parameters to inject @@ -373,6 +376,7 @@ func runConfigToCreateRequest(runConfig *runner.RunConfig) *createRequest { TargetPort: runConfig.TargetPort, ProxyPort: runConfig.Port, MaxRequestBodySize: runConfig.MaxRequestBodySize, + ProxyReadTimeout: runConfig.ProxyReadTimeout, EnvVars: runConfig.EnvVars, Secrets: secretParams, Volumes: runConfig.Volumes, diff --git a/pkg/api/v1/workloads_test.go b/pkg/api/v1/workloads_test.go index 588059fa25..1272b2e69f 100644 --- a/pkg/api/v1/workloads_test.go +++ b/pkg/api/v1/workloads_test.go @@ -739,20 +739,21 @@ func TestUpdateWorkload(t *testing.T) { } } -// TestUpdateWorkload_MaxRequestBodySizeRoundTrip verifies that the Workloads -// API includes the body limit in GET responses and preserves it when that +// TestUpdateWorkload_ProxyLimitsRoundTrip verifies that the Workloads API +// includes proxy request limits in GET responses and preserves them when that // response is submitted unchanged to the edit endpoint. // //nolint:paralleltest // SaveState/LoadState use process-wide XDG state settings; keep sequential. -func TestUpdateWorkload_MaxRequestBodySizeRoundTrip(t *testing.T) { +func TestUpdateWorkload_ProxyLimitsRoundTrip(t *testing.T) { t.Cleanup(xdg.Reload) t.Setenv("XDG_STATE_HOME", t.TempDir()) xdg.Reload() ctx := context.Background() const ( - workloadName = "body-limit-workload" - maxBytes = int64(16 << 20) + workloadName = "proxy-limits-workload" + maxBytes = int64(16 << 20) + proxyReadTimeout = "45s" ) persisted := runner.NewRunConfig() @@ -761,6 +762,7 @@ func TestUpdateWorkload_MaxRequestBodySizeRoundTrip(t *testing.T) { persisted.ContainerName = workloadName persisted.RemoteURL = "https://mcp.example.com/mcp" persisted.MaxRequestBodySize = maxBytes + persisted.ProxyReadTimeout = proxyReadTimeout require.NoError(t, persisted.SaveState(ctx)) ctrl := gomock.NewController(t) @@ -791,6 +793,7 @@ func TestUpdateWorkload_MaxRequestBodySizeRoundTrip(t *testing.T) { apierrors.ErrorHandler(routes.getWorkload).ServeHTTP(getRecorder, getReq) require.Equal(t, http.StatusOK, getRecorder.Code, getRecorder.Body.String()) assert.Contains(t, getRecorder.Body.String(), `"max_request_body_size":16777216`) + assert.Contains(t, getRecorder.Body.String(), `"proxy_read_timeout":"45s"`) mockWorkloadManager.EXPECT().GetWorkload(gomock.Any(), workloadName). Return(core.Workload{Name: workloadName}, nil) @@ -798,6 +801,7 @@ func TestUpdateWorkload_MaxRequestBodySizeRoundTrip(t *testing.T) { mockWorkloadManager.EXPECT().UpdateWorkload(gomock.Any(), workloadName, gomock.Any()). DoAndReturn(func(_ context.Context, _ string, runConfig *runner.RunConfig) (workloads.CompletionFunc, error) { assert.Equal(t, maxBytes, runConfig.MaxRequestBodySize) + assert.Equal(t, proxyReadTimeout, runConfig.ProxyReadTimeout) return nil, nil }) diff --git a/pkg/api/v1/workloads_types_test.go b/pkg/api/v1/workloads_types_test.go index 4669c17b13..5d0e4d73d1 100644 --- a/pkg/api/v1/workloads_types_test.go +++ b/pkg/api/v1/workloads_types_test.go @@ -96,6 +96,7 @@ func TestRunConfigToCreateRequest(t *testing.T) { CmdArgs: []string{"arg1", "arg2"}, TargetPort: 8080, MaxRequestBodySize: 16 << 20, + ProxyReadTimeout: "45s", EnvVars: map[string]string{"ENV1": "value1"}, Secrets: []string{"secret1,target=/path1", "secret2,target=/path2"}, Volumes: []string{"/host:/container"}, @@ -117,6 +118,7 @@ func TestRunConfigToCreateRequest(t *testing.T) { assert.Equal(t, 8080, result.TargetPort) assert.Equal(t, 3000, result.ProxyPort) assert.Equal(t, int64(16<<20), result.MaxRequestBodySize) + assert.Equal(t, "45s", result.ProxyReadTimeout) assert.Equal(t, map[string]string{"ENV1": "value1"}, result.EnvVars) require.Len(t, result.Secrets, 2) assert.Equal(t, "secret1", result.Secrets[0].Name) diff --git a/pkg/runner/config.go b/pkg/runner/config.go index d061092e75..c140b9fbfb 100644 --- a/pkg/runner/config.go +++ b/pkg/runner/config.go @@ -232,6 +232,13 @@ type RunConfig struct { // when the RunConfig is built or used at runtime. MaxRequestBodySize int64 `json:"max_request_body_size,omitempty" yaml:"max_request_body_size,omitempty"` + // ProxyReadTimeout bounds reading the entire request (headers + body) on the + // proxy HTTP server, expressed as a Go duration string (e.g. "30s", "1m"). + // Empty uses the proxy default (30s). Negative durations and values that fail + // time.ParseDuration are rejected at runtime. Applies to all HTTP transports. + // String (not time.Duration) keeps the wire format unit-explicit. + ProxyReadTimeout string `json:"proxy_read_timeout,omitempty" yaml:"proxy_read_timeout,omitempty" example:"30s"` + // ProxyMode is the effective HTTP protocol the proxy uses. // For stdio transports, this is the configured mode (sse or streamable-http). // For direct transports (sse/streamable-http), this matches the transport type. diff --git a/pkg/runner/config_builder.go b/pkg/runner/config_builder.go index 09b1c2eb83..4fa2920983 100644 --- a/pkg/runner/config_builder.go +++ b/pkg/runner/config_builder.go @@ -435,6 +435,26 @@ func WithMaxRequestBodySize(maxBytes int64) RunConfigBuilderOption { } } +// WithProxyReadTimeout sets http.Server.ReadTimeout on the proxy, bounding how +// long the server will spend reading a request (headers + body). Zero is valid +// and means "use the proxy default" (30s). Negative values return an error. +// +// The value is stored as a Go duration string on RunConfig so it survives a +// JSON/YAML round-trip; a time.Duration field would serialize as nanoseconds. +func WithProxyReadTimeout(d time.Duration) RunConfigBuilderOption { + return func(b *runConfigBuilder) error { + if d < 0 { + return fmt.Errorf("proxy-read-timeout must be non-negative, got %s", d) + } + if d == 0 { + b.config.ProxyReadTimeout = "" + return nil + } + b.config.ProxyReadTimeout = d.String() + return nil + } +} + // WithNetworkMode sets the network mode for the container. // The network mode will be applied to the permission profile after it is loaded. func WithNetworkMode(networkMode string) RunConfigBuilderOption { diff --git a/pkg/runner/config_builder_test.go b/pkg/runner/config_builder_test.go index 4e44c179c8..7914773ea3 100644 --- a/pkg/runner/config_builder_test.go +++ b/pkg/runner/config_builder_test.go @@ -1776,6 +1776,49 @@ func TestWithMaxRequestBodySize(t *testing.T) { } } +func TestWithProxyReadTimeout(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + value time.Duration + expectErr bool + expectedStr string + }{ + { + name: "zero is serialized as empty to use the proxy default", + value: 0, + expectedStr: "", + }, + { + name: "positive duration is stored as a Go duration string", + value: 45 * time.Second, + expectedStr: "45s", + }, + { + name: "negative duration returns an error", + value: -1 * time.Second, + expectErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + builder := &runConfigBuilder{config: NewRunConfig()} + err := WithProxyReadTimeout(tt.value)(builder) + + if tt.expectErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.expectedStr, builder.config.ProxyReadTimeout) + }) + } +} + // TestWithStrictProtocolValidation verifies the builder option sets // RunConfig.StrictProtocolValidation, mirroring WithTrustProxyHeaders's // plumbing (see cmd/thv/app/run_flags.go's --strict-protocol-validation flag). diff --git a/pkg/runner/runner.go b/pkg/runner/runner.go index d5bf30f663..7d4b08f1d2 100644 --- a/pkg/runner/runner.go +++ b/pkg/runner/runner.go @@ -193,6 +193,23 @@ func (c *RunConfig) GetPort() int { return c.Port } +// parseProxyTimeout parses an optional Go duration string from RunConfig. An +// empty string yields 0, which the proxy treats as "use the package default". +// Negative durations and unparsable values are rejected. +func parseProxyTimeout(name, value string) (time.Duration, error) { + if value == "" { + return 0, nil + } + d, err := time.ParseDuration(value) + if err != nil { + return 0, fmt.Errorf("invalid %s %q: %w", name, value, err) + } + if d < 0 { + return 0, fmt.Errorf("%s must be non-negative, got %s", name, d) + } + return d, nil +} + // Run runs the MCP server with the provided configuration // //nolint:gocyclo // This function is complex but manageable @@ -221,6 +238,13 @@ func (r *Runner) Run(ctx context.Context) error { } } + // Resolve the proxy HTTP server read timeout. Empty/zero means "use the proxy + // default"; the proxy option ignores non-positive values. + proxyReadTimeout, parseErr := parseProxyTimeout("proxy_read_timeout", r.Config.ProxyReadTimeout) + if parseErr != nil { + return parseErr + } + // Create transport with runtime transportConfig := types.Config{ Type: r.Config.Transport, @@ -234,6 +258,7 @@ func (r *Runner) Run(ctx context.Context) error { StrictProtocolValidation: r.Config.StrictProtocolValidation, EndpointPrefix: r.Config.EndpointPrefix, SessionTTL: effectiveSessionTTL, + ReadTimeout: proxyReadTimeout, } // Set proxy mode for stdio transport diff --git a/pkg/runner/runner_test.go b/pkg/runner/runner_test.go index 4ed5c1248c..54be820033 100644 --- a/pkg/runner/runner_test.go +++ b/pkg/runner/runner_test.go @@ -865,3 +865,34 @@ func TestRunner_GetUpstreamTokenReader(t *testing.T) { assert.Equal(t, svc, reader) }) } + +func TestParseProxyTimeout(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + value string + want time.Duration + expectErr bool + }{ + {name: "empty uses the proxy default", value: "", want: 0}, + {name: "valid duration", value: "45s", want: 45 * time.Second}, + {name: "explicit zero uses the proxy default", value: "0s", want: 0}, + {name: "negative duration is rejected", value: "-1s", expectErr: true}, + {name: "unparsable duration is rejected", value: "notaduration", expectErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := parseProxyTimeout("proxy_read_timeout", tt.value) + if tt.expectErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/pkg/transport/factory.go b/pkg/transport/factory.go index f54fb13a1f..adf9534cb6 100644 --- a/pkg/transport/factory.go +++ b/pkg/transport/factory.go @@ -6,6 +6,8 @@ package transport import ( + "fmt" + "github.com/stacklok/toolhive/pkg/transport/errors" "github.com/stacklok/toolhive/pkg/transport/types" ) @@ -43,6 +45,10 @@ func WithTargetURI(targetURI string) Option { // Create creates a transport based on the provided configuration func (*Factory) Create(config types.Config, opts ...Option) (types.Transport, error) { + if config.ReadTimeout < 0 { + return nil, fmt.Errorf("read timeout must be non-negative, got %s", config.ReadTimeout) + } + var tr types.Transport switch config.Type { @@ -57,6 +63,7 @@ func (*Factory) Create(config types.Config, opts ...Option) (types.Transport, er stdio.SetSessionStorage(config.SessionStorage) } stdio.SetSessionTTL(config.SessionTTL) + stdio.SetReadTimeout(config.ReadTimeout) if config.AuthInfoHandler != nil { stdio.SetAuthInfoHandler(config.AuthInfoHandler) } @@ -82,6 +89,7 @@ func (*Factory) Create(config types.Config, opts ...Option) (types.Transport, er ) httpTransport.sessionStorage = config.SessionStorage httpTransport.sessionTTL = config.SessionTTL + httpTransport.readTimeout = config.ReadTimeout tr = httpTransport case types.TransportTypeStreamableHTTP: httpTransport := NewHTTPTransport( @@ -101,6 +109,7 @@ func (*Factory) Create(config types.Config, opts ...Option) (types.Transport, er ) httpTransport.sessionStorage = config.SessionStorage httpTransport.sessionTTL = config.SessionTTL + httpTransport.readTimeout = config.ReadTimeout tr = httpTransport case types.TransportTypeInspector: // HTTP transport is not implemented yet diff --git a/pkg/transport/http.go b/pkg/transport/http.go index 7f9bb174b9..8dd3bc0482 100644 --- a/pkg/transport/http.go +++ b/pkg/transport/http.go @@ -86,6 +86,10 @@ type HTTPTransport struct { // underlying proxy. Zero uses the proxy's default. sessionTTL time.Duration + // readTimeout overrides http.Server.ReadTimeout on the underlying transparent + // proxy. Zero uses the proxy's default. + readTimeout time.Duration + // Transparent proxy proxy types.Proxy @@ -440,6 +444,9 @@ func (t *HTTPTransport) buildProxyOptions(remoteBasePath, remoteRawQuery string) if t.sessionTTL > 0 { opts = append(opts, transparent.WithSessionTTL(t.sessionTTL)) } + if t.readTimeout > 0 { + opts = append(opts, transparent.WithReadTimeout(t.readTimeout)) + } if t.sessionStorage != nil { opts = append(opts, transparent.WithSessionStorage(t.sessionStorage)) } diff --git a/pkg/transport/read_timeout_test.go b/pkg/transport/read_timeout_test.go new file mode 100644 index 0000000000..514dda061f --- /dev/null +++ b/pkg/transport/read_timeout_test.go @@ -0,0 +1,229 @@ +// SPDX-FileCopyrightText: Copyright 2026 Stacklok, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package transport + +import ( + "bytes" + "context" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/stacklok/toolhive/pkg/transport/proxy/httpsse" + "github.com/stacklok/toolhive/pkg/transport/proxy/streamable" + "github.com/stacklok/toolhive/pkg/transport/proxy/transparent" + "github.com/stacklok/toolhive/pkg/transport/types" +) + +const testProxyReadTimeout = 300 * time.Millisecond + +// plumbingSlowBody trickles a request body slowly enough that an applied test +// timeout expires well before the body is complete. Without the transport's +// WithReadTimeout plumbing, the request takes roughly three seconds instead. +type plumbingSlowBody struct { + total int + emitted int +} + +func (r *plumbingSlowBody) Read(p []byte) (int, error) { + if r.emitted >= r.total { + return 0, io.EOF + } + time.Sleep(200 * time.Millisecond) + if len(p) == 0 { + return 0, nil + } + r.emitted++ + p[0] = ' ' + return 1, nil +} + +// The stdio proxy constructors require a concrete port, so these cases run +// sequentially to minimize the close-and-rebind window around ephemeral ports. +func TestFactoryReadTimeoutChangesLiveProxyBehavior(t *testing.T) { //nolint:paralleltest + tests := []struct { + name string + startProxy func(t *testing.T) string + }{ + { + name: "stdio streamable HTTP proxy", + startProxy: startStdioStreamableProxyWithFactoryTimeout, + }, + { + name: "stdio SSE proxy", + startProxy: startStdioSSEProxyWithFactoryTimeout, + }, + { + name: "direct HTTP transparent proxy", + startProxy: startTransparentProxyWithFactoryTimeout, + }, + } + + // Keep the network cases sequential for the ephemeral-port handoff above. + for _, tt := range tests { //nolint:paralleltest + t.Run(tt.name, func(t *testing.T) { + assertSlowUploadTimesOut(t, tt.startProxy(t)) + }) + } +} + +func startStdioStreamableProxyWithFactoryTimeout(t *testing.T) string { + t.Helper() + + transport, err := NewFactory().Create(types.Config{ + Type: types.TransportTypeStdio, + ProxyMode: types.ProxyModeStreamableHTTP, + ReadTimeout: testProxyReadTimeout, + }) + require.NoError(t, err) + stdioTransport, ok := transport.(*StdioTransport) + require.True(t, ok, "factory should create a StdioTransport") + + port := reserveLoopbackPort(t) + proxy := streamable.NewHTTPProxy(LocalhostIPv4, port, nil, nil, stdioTransport.streamableProxyOptions()...) + require.NoError(t, proxy.Start(t.Context())) + t.Cleanup(func() { + stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = proxy.Stop(stopCtx) + }) + + addr := fmt.Sprintf("%s:%d", LocalhostIPv4, port) + waitForListener(t, addr) + return "http://" + addr + streamable.StreamableHTTPEndpoint +} + +func startStdioSSEProxyWithFactoryTimeout(t *testing.T) string { + t.Helper() + + transport, err := NewFactory().Create(types.Config{ + Type: types.TransportTypeStdio, + ProxyMode: types.ProxyModeSSE, + ReadTimeout: testProxyReadTimeout, + }) + require.NoError(t, err) + stdioTransport, ok := transport.(*StdioTransport) + require.True(t, ok, "factory should create a StdioTransport") + + port := reserveLoopbackPort(t) + proxy := httpsse.NewHTTPSSEProxy(LocalhostIPv4, port, false, nil, nil, stdioTransport.sseProxyOptions()...) + require.NoError(t, proxy.Start(t.Context())) + t.Cleanup(func() { + stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = proxy.Stop(stopCtx) + }) + + addr := fmt.Sprintf("%s:%d", LocalhostIPv4, port) + waitForListener(t, addr) + setupCtx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + t.Cleanup(cancel) + req, err := http.NewRequestWithContext(setupCtx, http.MethodGet, "http://"+addr+"/sse", nil) + require.NoError(t, err) + resp, err := http.DefaultClient.Do(req) //nolint:gosec // loopback test server + require.NoError(t, err) + t.Cleanup(func() { _ = resp.Body.Close() }) + require.Equal(t, http.StatusOK, resp.StatusCode) + + buf := make([]byte, 4096) + n, err := resp.Body.Read(buf) + require.NoError(t, err) + const marker = "session_id=" + start := bytes.Index(buf[:n], []byte(marker)) + require.NotEqual(t, -1, start, "SSE endpoint event should contain a session ID") + start += len(marker) + end := bytes.IndexByte(buf[start:n], '\n') + require.NotEqual(t, -1, end, "SSE endpoint event should terminate its data line") + sessionID := string(bytes.TrimSpace(buf[start : start+end])) + require.NotEmpty(t, sessionID) + + return fmt.Sprintf("http://%s/messages?session_id=%s", addr, sessionID) +} + +func startTransparentProxyWithFactoryTimeout(t *testing.T) string { + t.Helper() + + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(backend.Close) + + transport, err := NewFactory().Create(types.Config{ + Type: types.TransportTypeStreamableHTTP, + Host: LocalhostIPv4, + ReadTimeout: testProxyReadTimeout, + }) + require.NoError(t, err) + httpTransport, ok := transport.(*HTTPTransport) + require.True(t, ok, "factory should create an HTTPTransport") + httpTransport.SetRemoteURL(backend.URL + "/mcp") + require.NoError(t, httpTransport.Start(t.Context())) + t.Cleanup(func() { + stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = httpTransport.Stop(stopCtx) + }) + + proxy, ok := httpTransport.proxy.(*transparent.TransparentProxy) + require.True(t, ok, "HTTPTransport should start a transparent proxy") + addr := proxy.ListenerAddr() + require.NotEmpty(t, addr) + return "http://" + addr + "/mcp" +} + +func assertSlowUploadTimesOut(t *testing.T, url string) { + t.Helper() + + body := &plumbingSlowBody{total: 15} + req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, url, body) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + req.ContentLength = int64(body.total) + + client := &http.Client{Timeout: 5 * time.Second} + start := time.Now() + resp, requestErr := client.Do(req) + elapsed := time.Since(start) + if resp != nil { + t.Cleanup(func() { _ = resp.Body.Close() }) + } + + if requestErr == nil { + assert.NotEqual(t, http.StatusOK, resp.StatusCode, + "a slow upload should not reach a successful handler response") + } + assert.Less(t, elapsed, 2*time.Second, + "configured timeout should terminate the upload before its three-second body completes") +} + +func reserveLoopbackPort(t *testing.T) int { + t.Helper() + + listener, err := net.Listen("tcp", LocalhostIPv4+":0") + require.NoError(t, err) + port := listener.Addr().(*net.TCPAddr).Port + require.NoError(t, listener.Close()) + return port +} + +func waitForListener(t *testing.T, addr string) { + t.Helper() + + require.Eventually(t, func() bool { + conn, err := net.DialTimeout("tcp", addr, 50*time.Millisecond) + if err != nil { + return false + } + _ = conn.Close() + return true + }, 2*time.Second, 10*time.Millisecond, "proxy should begin listening") +} diff --git a/pkg/transport/stdio.go b/pkg/transport/stdio.go index 4a8c9b262c..06879c716f 100644 --- a/pkg/transport/stdio.go +++ b/pkg/transport/stdio.go @@ -65,6 +65,7 @@ type StdioTransport struct { trustProxyHeaders bool sessionStorage session.Storage sessionTTL time.Duration + readTimeout time.Duration authInfoHandler http.Handler prefixHandlers map[string]http.Handler @@ -165,6 +166,12 @@ func (t *StdioTransport) SetSessionTTL(ttl time.Duration) { t.sessionTTL = ttl } +// SetReadTimeout configures http.Server.ReadTimeout on the underlying proxy. +// Zero is valid and means "use the proxy's default". +func (t *StdioTransport) SetReadTimeout(d time.Duration) { + t.readTimeout = d +} + // SetAuthInfoHandler sets the RFC 9728 OAuth protected resource discovery handler. func (t *StdioTransport) SetAuthInfoHandler(h http.Handler) { t.authInfoHandler = h @@ -271,6 +278,9 @@ func (t *StdioTransport) streamableProxyOptions() []streamable.Option { if t.sessionTTL > 0 { opts = append(opts, streamable.WithSessionTTL(t.sessionTTL)) } + if t.readTimeout > 0 { + opts = append(opts, streamable.WithReadTimeout(t.readTimeout)) + } if t.sessionStorage != nil { opts = append(opts, streamable.WithSessionStorage(t.sessionStorage)) } @@ -288,6 +298,9 @@ func (t *StdioTransport) sseProxyOptions() []httpsse.Option { if t.sessionTTL > 0 { opts = append(opts, httpsse.WithSessionTTL(t.sessionTTL)) } + if t.readTimeout > 0 { + opts = append(opts, httpsse.WithReadTimeout(t.readTimeout)) + } if t.sessionStorage != nil { opts = append(opts, httpsse.WithSessionStorage(t.sessionStorage)) } diff --git a/pkg/transport/stdio_test.go b/pkg/transport/stdio_test.go index 4b5f135964..825e4a63ad 100644 --- a/pkg/transport/stdio_test.go +++ b/pkg/transport/stdio_test.go @@ -1195,3 +1195,74 @@ func TestFactory_Create_PreservesAuthFields(t *testing.T) { }) } } + +func TestFactory_Create_PreservesReadTimeout(t *testing.T) { + t.Parallel() + + const readTimeout = 45 * time.Second + tests := []struct { + name string + transportType types.TransportType + check func(t *testing.T, tr types.Transport) + }{ + { + name: "stdio", + transportType: types.TransportTypeStdio, + check: func(t *testing.T, tr types.Transport) { + t.Helper() + stdio, ok := tr.(*StdioTransport) + require.True(t, ok, "expected *StdioTransport") + assert.Equal(t, readTimeout, stdio.readTimeout) + }, + }, + { + name: "direct SSE", + transportType: types.TransportTypeSSE, + check: func(t *testing.T, tr types.Transport) { + t.Helper() + httpTransport, ok := tr.(*HTTPTransport) + require.True(t, ok, "expected *HTTPTransport") + assert.Equal(t, readTimeout, httpTransport.readTimeout) + }, + }, + { + name: "direct streamable HTTP", + transportType: types.TransportTypeStreamableHTTP, + check: func(t *testing.T, tr types.Transport) { + t.Helper() + httpTransport, ok := tr.(*HTTPTransport) + require.True(t, ok, "expected *HTTPTransport") + assert.Equal(t, readTimeout, httpTransport.readTimeout) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + factory := NewFactory() + tr, err := factory.Create(types.Config{ + Type: tt.transportType, + Host: "localhost", + ProxyPort: 8080, + ReadTimeout: readTimeout, + }) + require.NoError(t, err) + tt.check(t, tr) + }) + } +} + +func TestFactory_Create_RejectsNegativeReadTimeout(t *testing.T) { + t.Parallel() + + transport, err := NewFactory().Create(types.Config{ + Type: types.TransportTypeStdio, + ReadTimeout: -time.Second, + }) + + require.Error(t, err) + assert.Nil(t, transport) + assert.EqualError(t, err, "read timeout must be non-negative, got -1s") +} diff --git a/pkg/transport/types/transport.go b/pkg/transport/types/transport.go index f94254384a..d712150eef 100644 --- a/pkg/transport/types/transport.go +++ b/pkg/transport/types/transport.go @@ -288,6 +288,12 @@ type Config struct { // Sessions idle for longer than this duration are cleaned up by the session // manager's background worker. Zero uses session.DefaultSessionTTL. SessionTTL time.Duration + + // ReadTimeout bounds reading the entire request (headers + body) on the proxy + // http.Server. Zero uses the proxy package default; negative values are + // rejected. Applies to all HTTP transports; it never affects SSE responses, + // which stream on the response side. + ReadTimeout time.Duration } // ProxyMode represents the proxy mode for stdio transport. diff --git a/pkg/workloads/upgrade/applier.go b/pkg/workloads/upgrade/applier.go index a60d6d12ba..c4a3d96bdc 100644 --- a/pkg/workloads/upgrade/applier.go +++ b/pkg/workloads/upgrade/applier.go @@ -410,6 +410,7 @@ func preserveUserConfigFields(dst, old *runner.RunConfig) { // Proxy session / runtime / scheduling knobs. dst.SessionTTL = old.SessionTTL + dst.ProxyReadTimeout = old.ProxyReadTimeout dst.RuntimeConfig = old.RuntimeConfig dst.IgnoreConfig = old.IgnoreConfig dst.K8sPodTemplatePatch = old.K8sPodTemplatePatch diff --git a/pkg/workloads/upgrade/applier_test.go b/pkg/workloads/upgrade/applier_test.go index dec77ab6db..ae69a6bfd8 100644 --- a/pkg/workloads/upgrade/applier_test.go +++ b/pkg/workloads/upgrade/applier_test.go @@ -512,6 +512,7 @@ func fullyConfiguredOld(t *testing.T) *runner.RunConfig { EndpointPrefix: "/mcp", SessionTTL: "30m", MaxRequestBodySize: 16 << 20, + ProxyReadTimeout: "45s", Debug: true, ContainerLabels: map[string]string{"team": "platform"}, OIDCConfig: &auth.TokenValidatorConfig{Issuer: "https://issuer.example", Audience: "aud"}, @@ -576,6 +577,7 @@ func TestApplier_Apply_PreservesFullUserConfig(t *testing.T) { assert.Equal(t, old.EndpointPrefix, got.EndpointPrefix) assert.Equal(t, old.SessionTTL, got.SessionTTL) assert.Equal(t, old.MaxRequestBodySize, got.MaxRequestBodySize) + assert.Equal(t, old.ProxyReadTimeout, got.ProxyReadTimeout, "proxy read timeout preserved") assert.Equal(t, old.Debug, got.Debug) assert.Equal(t, "platform", got.ContainerLabels["team"], "user label preserved") assert.Equal(t, old.OIDCConfig, got.OIDCConfig, "OIDC config preserved") diff --git a/test/e2e/proxy_read_timeout_test.go b/test/e2e/proxy_read_timeout_test.go new file mode 100644 index 0000000000..813bd03f71 --- /dev/null +++ b/test/e2e/proxy_read_timeout_test.go @@ -0,0 +1,120 @@ +// SPDX-FileCopyrightText: Copyright 2026 Stacklok, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package e2e_test + +import ( + "context" + "io" + "net/http" + "strings" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/stacklok/toolhive/test/e2e" +) + +const ( + cliProxyReadTimeout = time.Second + slowUploadChunkDelay = 250 * time.Millisecond + slowUploadBodyBytes = 20 + slowUploadMaxDuration = 4 * time.Second +) + +// cliSlowUploadBody takes five seconds to produce its complete body. A proxy +// configured with cliProxyReadTimeout must terminate the request well before +// that, while an unwired flag leaves the default 30-second timeout in effect. +type cliSlowUploadBody struct { + emitted int +} + +func (b *cliSlowUploadBody) Read(p []byte) (int, error) { + if b.emitted >= slowUploadBodyBytes { + return 0, io.EOF + } + if len(p) == 0 { + return 0, nil + } + + time.Sleep(slowUploadChunkDelay) + b.emitted++ + p[0] = ' ' + return 1, nil +} + +var _ = Describe("Proxy read timeout CLI wiring", Label("proxy", "read-timeout", "e2e"), Serial, func() { + var ( + config *e2e.TestConfig + serverName string + mockServer *statelessMockMCPServer + proxyURL string + ) + + BeforeEach(func() { + config = e2e.NewTestConfig() + serverName = e2e.GenerateUniqueServerName("proxy-read-timeout") + + Expect(e2e.CheckTHVBinaryAvailable(config)).To(Succeed(), "thv binary should be available") + + var err error + mockServer, err = newStatelessMockMCPServer() + Expect(err).ToNot(HaveOccurred(), "should start the mock MCP server") + + By("starting a workload through thv run with --proxy-read-timeout") + e2e.NewTHVCommand(config, + "run", + "--name", serverName, + "--stateless", + "--proxy-read-timeout", cliProxyReadTimeout.String(), + mockServer.URL()+"/mcp", + ).ExpectSuccess() + + Expect(e2e.WaitForMCPServer(config, serverName, e2e.ServerReadyTimeout())).To(Succeed()) + + proxyURL, err = e2e.GetMCPServerURL(config, serverName) + Expect(err).ToNot(HaveOccurred(), "should discover the workload proxy URL") + if !strings.HasSuffix(proxyURL, "/mcp") { + proxyURL += "/mcp" + } + }) + + AfterEach(func() { + if config != nil && config.CleanupAfter && serverName != "" { + Expect(e2e.StopAndRemoveMCPServer(config, serverName)).To(Succeed()) + } + if mockServer != nil { + mockServer.Stop() + mockServer = nil + } + }) + + It("applies --proxy-read-timeout to a live proxy", func() { + body := &cliSlowUploadBody{} + + requestCtx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(requestCtx, http.MethodPost, proxyURL, body) + Expect(err).ToNot(HaveOccurred()) + req.Header.Set("Content-Type", "application/json") + req.ContentLength = slowUploadBodyBytes + + client := &http.Client{Timeout: 8 * time.Second} + started := time.Now() + resp, requestErr := client.Do(req) + elapsed := time.Since(started) + if resp != nil { + _, _ = io.Copy(io.Discard, resp.Body) + Expect(resp.Body.Close()).To(Succeed()) + } + + if requestErr == nil { + Expect(resp).ToNot(BeNil()) + Expect(resp.StatusCode).ToNot(Equal(http.StatusOK), + "a timed-out upload must not produce a successful MCP response") + } + Expect(elapsed).To(BeNumerically("<", slowUploadMaxDuration), + "the configured one-second timeout should terminate the five-second upload") + }) +}) diff --git a/test/e2e/thv-operator/acceptance_tests/proxy_read_timeout_test.go b/test/e2e/thv-operator/acceptance_tests/proxy_read_timeout_test.go new file mode 100644 index 0000000000..76056316d0 --- /dev/null +++ b/test/e2e/thv-operator/acceptance_tests/proxy_read_timeout_test.go @@ -0,0 +1,155 @@ +// SPDX-FileCopyrightText: Copyright 2026 Stacklok, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package acceptancetests + +import ( + "context" + "fmt" + "io" + "net/http" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + mcpv1beta1 "github.com/stacklok/toolhive/cmd/thv-operator/api/v1beta1" + "github.com/stacklok/toolhive/test/e2e/images" + "github.com/stacklok/toolhive/test/e2e/thv-operator/testutil" +) + +const ( + operatorProxyReadTimeout = time.Second + operatorSlowUploadDelay = 250 * time.Millisecond + operatorSlowUploadBytes = 20 + operatorSlowUploadMaxTime = 4 * time.Second +) + +// operatorSlowUploadBody takes five seconds to complete, so the request only +// ends before operatorSlowUploadMaxTime when the MCPServer setting reaches the +// live proxy runner. +type operatorSlowUploadBody struct { + emitted int +} + +func (b *operatorSlowUploadBody) Read(p []byte) (int, error) { + if b.emitted >= operatorSlowUploadBytes { + return 0, io.EOF + } + if len(p) == 0 { + return 0, nil + } + + time.Sleep(operatorSlowUploadDelay) + b.emitted++ + p[0] = ' ' + return 1, nil +} + +var _ = Describe("MCPServer proxy read timeout", Label("mcpserver", "read-timeout", "e2e"), Ordered, Serial, func() { + const ( + testNamespace = "proxy-read-timeout" + serverName = "proxy-read-timeout" + timeout = 3 * time.Minute + pollingInterval = time.Second + ) + + var nodePort int32 + + BeforeAll(func() { + By("creating an isolated namespace") + namespace := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: testNamespace}} + Expect(client.IgnoreAlreadyExists(k8sClient.Create(ctx, namespace))).To(Succeed()) + + By("creating an MCPServer with a short proxyReadTimeout") + server := &mcpv1beta1.MCPServer{ + ObjectMeta: metav1.ObjectMeta{ + Name: serverName, + Namespace: testNamespace, + }, + Spec: mcpv1beta1.MCPServerSpec{ + Image: images.YardstickServerImage, + Transport: "streamable-http", + ProxyPort: 8080, + MCPPort: 8080, + ProxyReadTimeout: &metav1.Duration{Duration: operatorProxyReadTimeout}, + Env: []mcpv1beta1.EnvVar{ + {Name: "TRANSPORT", Value: "streamable-http"}, + }, + }, + } + Expect(k8sClient.Create(ctx, server)).To(Succeed()) + + By("waiting for the MCPServer to become ready") + testutil.WaitForMCPServerRunning(ctx, k8sClient, serverName, testNamespace, timeout, pollingInterval) + + By("exposing the MCPServer proxy through an auto-assigned NodePort") + testutil.CreateNodePortService(ctx, k8sClient, serverName, testNamespace) + nodePort = testutil.GetNodePort( + ctx, + k8sClient, + serverName+"-nodeport", + testNamespace, + timeout, + pollingInterval, + ) + + By("waiting for the proxy endpoint to accept connections") + healthClient := &http.Client{Timeout: 5 * time.Second} + Eventually(func() error { + resp, err := healthClient.Get(fmt.Sprintf("http://localhost:%d/health", nodePort)) + if err != nil { + return err + } + defer func() { + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + }() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("health check returned %d", resp.StatusCode) + } + return nil + }, timeout, pollingInterval).Should(Succeed()) + }) + + AfterAll(func() { + By("deleting the isolated namespace") + _ = k8sClient.Delete(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: testNamespace}}) + }) + + It("terminates a slow upload using spec.proxyReadTimeout", func() { + body := &operatorSlowUploadBody{} + requestCtx, cancel := context.WithTimeout(ctx, 8*time.Second) + defer cancel() + + req, err := http.NewRequestWithContext( + requestCtx, + http.MethodPost, + fmt.Sprintf("http://localhost:%d/mcp", nodePort), + body, + ) + Expect(err).ToNot(HaveOccurred()) + req.Header.Set("Content-Type", "application/json") + req.ContentLength = operatorSlowUploadBytes + + httpClient := &http.Client{Timeout: 8 * time.Second} + started := time.Now() + resp, requestErr := httpClient.Do(req) + elapsed := time.Since(started) + if resp != nil { + _, _ = io.Copy(io.Discard, resp.Body) + Expect(resp.Body.Close()).To(Succeed()) + } + + if requestErr == nil { + Expect(resp).ToNot(BeNil()) + Expect(resp.StatusCode).ToNot(Equal(http.StatusOK), + "a timed-out upload must not produce a successful MCP response") + } + Expect(elapsed).To(BeNumerically("<", operatorSlowUploadMaxTime), + "the one-second MCPServer timeout should terminate the five-second upload") + }) +})