From 3f617bc937e83a0c79d8e336b21de0bb2852f4d7 Mon Sep 17 00:00:00 2001 From: Sanskarzz Date: Fri, 4 Sep 2026 19:56:43 +0530 Subject: [PATCH] feat make proxy request-body limit configurable Signed-off-by: Sanskarzz --- .../api/v1beta1/mcpserver_types.go | 9 +++ .../api/v1beta1/zz_generated.deepcopy.go | 5 ++ .../controllers/mcpserver_runconfig.go | 5 ++ .../controllers/mcpserver_runconfig_test.go | 31 ++++++++++ .../mcpserver_sessionstorage_cel_test.go | 23 ++++++++ cmd/thv/app/run_flags.go | 6 ++ .../toolhive.stacklok.dev_mcpservers.yaml | 20 +++++++ .../toolhive.stacklok.dev_mcpservers.yaml | 20 +++++++ docs/arch/03-transport-architecture.md | 13 +++++ docs/arch/05-runconfig-and-permissions.md | 4 +- docs/cli/thv_run.md | 1 + docs/operator/crd-api.md | 1 + docs/server/docs.go | 5 ++ docs/server/swagger.json | 5 ++ docs/server/swagger.yaml | 9 +++ pkg/runner/config.go | 7 +++ pkg/runner/config_builder.go | 20 +++++++ pkg/runner/config_builder_test.go | 43 ++++++++++++++ pkg/runner/runner.go | 25 ++++++++ pkg/runner/runner_test.go | 31 ++++++++++ pkg/transport/factory.go | 3 + pkg/transport/http.go | 7 +++ pkg/transport/stdio.go | 13 +++++ pkg/transport/stdio_test.go | 58 +++++++++++++++++++ pkg/transport/types/transport.go | 5 ++ 25 files changed, 368 insertions(+), 1 deletion(-) diff --git a/cmd/thv-operator/api/v1beta1/mcpserver_types.go b/cmd/thv-operator/api/v1beta1/mcpserver_types.go index 60a0981c07..b662e73e01 100644 --- a/cmd/thv-operator/api/v1beta1/mcpserver_types.go +++ b/cmd/thv-operator/api/v1beta1/mcpserver_types.go @@ -420,6 +420,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 de7541d3ec..97442b1626 100644 --- a/cmd/thv-operator/api/v1beta1/zz_generated.deepcopy.go +++ b/cmd/thv-operator/api/v1beta1/zz_generated.deepcopy.go @@ -1886,6 +1886,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 373fd0e0db..1781ed150c 100644 --- a/cmd/thv-operator/controllers/mcpserver_runconfig.go +++ b/cmd/thv-operator/controllers/mcpserver_runconfig.go @@ -277,6 +277,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 d3b1f9b942..2597f90daa 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 05bdbbefcc..60a5415160 100644 --- a/cmd/thv/app/run_flags.go +++ b/cmd/thv/app/run_flags.go @@ -121,6 +121,9 @@ type RunFlags struct { // SessionTTL is the session inactivity timeout. Zero uses the transport default. SessionTTL time.Duration + // ProxyReadTimeout bounds reading a full request on the proxy. Zero uses the default. + ProxyReadTimeout time.Duration + // Network mode Network string @@ -310,6 +313,8 @@ func AddRunFlags(cmd *cobra.Command, config *RunFlags) { "Use for MCP servers implementing streamable-HTTP stateless mode.") cmd.Flags().DurationVar(&config.SessionTTL, "session-ttl", 0, "Session inactivity timeout (e.g., 30m, 2h); zero uses the default (2h)") + 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", "", @@ -735,6 +740,7 @@ func buildRunnerConfig( runner.WithStrictProtocolValidation(runFlags.StrictProtocolValidation), runner.WithStateless(runFlags.Stateless), runner.WithSessionTTL(runFlags.SessionTTL), + 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 5063bc84e1..233517cb98 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 @@ -390,6 +390,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. @@ -2317,6 +2327,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 4e92f105f2..313df9b636 100644 --- a/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_mcpservers.yaml +++ b/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_mcpservers.yaml @@ -393,6 +393,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. @@ -2320,6 +2330,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 96d8a2ec75..81e5f94d20 100644 --- a/docs/arch/03-transport-architecture.md +++ b/docs/arch/03-transport-architecture.md @@ -368,6 +368,19 @@ 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. +### 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 0d3ab48578..f5157bab63 100644 --- a/docs/arch/05-runconfig-and-permissions.md +++ b/docs/arch/05-runconfig-and-permissions.md @@ -124,7 +124,8 @@ thv run uvx://mcp-server \ "transport": "stdio", "host": "127.0.0.1", "port": 8080, - "proxy_mode": "streamable-http" + "proxy_mode": "streamable-http", + "proxy_read_timeout": "45s" } ``` @@ -146,6 +147,7 @@ thv run uvx://mcp-server \ - `target_port`: Container port (SSE/Streamable only) - `target_host`: Container host (default: `127.0.0.1`) - `proxy_mode`: For stdio: `sse` or `streamable-http` +- `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 6402c98071..8894ecc2cb 100644 --- a/docs/cli/thv_run.md +++ b/docs/cli/thv_run.md @@ -159,6 +159,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 17be058246..17f46b19e7 100644 --- a/docs/operator/crd-api.md +++ b/docs/operator/crd-api.md @@ -3315,6 +3315,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 55fd152ecb..cfe9c2e2e3 100644 --- a/docs/server/docs.go +++ b/docs/server/docs.go @@ -1673,6 +1673,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": { diff --git a/docs/server/swagger.json b/docs/server/swagger.json index f5841802a5..31dd4512d3 100644 --- a/docs/server/swagger.json +++ b/docs/server/swagger.json @@ -1666,6 +1666,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": { diff --git a/docs/server/swagger.yaml b/docs/server/swagger.yaml index a44d2f8ab7..40ad38b4fc 100644 --- a/docs/server/swagger.yaml +++ b/docs/server/swagger.yaml @@ -1803,6 +1803,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: diff --git a/pkg/runner/config.go b/pkg/runner/config.go index 268c6692a8..0adf58e2fe 100644 --- a/pkg/runner/config.go +++ b/pkg/runner/config.go @@ -227,6 +227,13 @@ type RunConfig struct { // time.Duration field serializes as nanoseconds in JSON. SessionTTL string `json:"session_ttl,omitempty" yaml:"session_ttl,omitempty" example:"2h"` + // 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 d6367e9453..d4ecbbe98b 100644 --- a/pkg/runner/config_builder.go +++ b/pkg/runner/config_builder.go @@ -423,6 +423,26 @@ func WithSessionTTL(ttl time.Duration) 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 0087d97ffd..297baba34c 100644 --- a/pkg/runner/config_builder_test.go +++ b/pkg/runner/config_builder_test.go @@ -1542,6 +1542,49 @@ func TestWithSessionTTL(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 6750ef773c..a95210010e 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 @@ -215,6 +232,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, @@ -228,6 +252,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 be1780e17b..e74b52917c 100644 --- a/pkg/runner/runner_test.go +++ b/pkg/runner/runner_test.go @@ -823,3 +823,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..7ccfd0deb8 100644 --- a/pkg/transport/factory.go +++ b/pkg/transport/factory.go @@ -57,6 +57,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 +83,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 +103,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/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..76031b4a75 100644 --- a/pkg/transport/stdio_test.go +++ b/pkg/transport/stdio_test.go @@ -1195,3 +1195,61 @@ 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) + }) + } +} diff --git a/pkg/transport/types/transport.go b/pkg/transport/types/transport.go index f94254384a..8e7e9a7977 100644 --- a/pkg/transport/types/transport.go +++ b/pkg/transport/types/transport.go @@ -288,6 +288,11 @@ 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. 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.