Set finite read limits for RPCs
diff --git a/private/buf/bufcli/connectclient_config.go b/private/buf/bufcli/connectclient_config.go index 628346e..e75c210 100644 --- a/private/buf/bufcli/connectclient_config.go +++ b/private/buf/bufcli/connectclient_config.go
@@ -26,6 +26,16 @@ "github.com/bufbuild/buf/private/pkg/transport/http/httpclient" ) +// defaultReadMaxBytes is the maximum size of a single message that a registry +// client will read. +// +// Connect defaults to allowing any message size, and is expected to adopt a +// 4 MiB default. That is well below what the BSR already returns for module, +// plugin, and policy downloads, so we set an explicit bound instead of +// inheriting one. Individual clients that need more can override this by +// passing connect.WithReadMaxBytes to connectclient.Make. +const defaultReadMaxBytes = 128 << 20 // 128 MiB + // NewConnectClientConfig creates a new connect.ClientConfig which uses a token reader to look // up the token in the container or in netrc based on the address of each individual client. // It is then set in the header of all outgoing requests from clients created using this config. @@ -85,6 +95,9 @@ otelconnectInterceptor, }, ), + connectclient.WithClientOptions( + connect.WithReadMaxBytes(defaultReadMaxBytes), + ), } return connectclient.NewConfig(client, append(options, opts...)...), nil }
diff --git a/private/buf/bufcurl/invoker.go b/private/buf/bufcurl/invoker.go index 46a8581..2b18870 100644 --- a/private/buf/bufcurl/invoker.go +++ b/private/buf/bufcurl/invoker.go
@@ -24,6 +24,7 @@ "maps" "net/http" "net/http/httptest" + "slices" "sync" "buf.build/go/app" @@ -86,7 +87,13 @@ // extensions that appear in the input or output. Other parameters are used // to create a Connect client, for issuing the RPC. func NewInvoker(container appext.Container, verbosePrinter verbose.Printer, md protoreflect.MethodDescriptor, res protoencoding.Resolver, emitDefaults bool, httpClient connect.HTTPClient, opts []connect.ClientOption, url string, out io.Writer) Invoker { - opts = append(opts, connect.WithCodec(protoCodec{})) + // buf curl invokes whatever RPC the user names, so there is no response size + // we can assume is illegitimate. Connect is expected to adopt a default + // per-message read limit, so opt out explicitly to preserve behavior. + // + // Clone rather than append in place: callers share this slice with other + // clients. + opts = append(slices.Clone(opts), connect.WithReadMaxBytes(0), connect.WithCodec(protoCodec{})) // TODO: could also provide custom compressor implementations that could give us // optics into when request and response messages are compressed (which could be // useful to include in verbose output).
diff --git a/private/buf/bufcurl/reflection_resolver.go b/private/buf/bufcurl/reflection_resolver.go index 17adc32..c0200dc 100644 --- a/private/buf/bufcurl/reflection_resolver.go +++ b/private/buf/bufcurl/reflection_resolver.go
@@ -20,6 +20,7 @@ "fmt" "maps" "net/http" + "slices" "strconv" "strings" "sync" @@ -104,6 +105,14 @@ printer verbose.Printer, ) (r Resolver, closeResolver func()) { baseURL = strings.TrimSuffix(baseURL, "/") + // Reflection responses carry FileDescriptorProtos, which are unbounded in + // the size of the schema being served. Connect is expected to adopt a + // default per-message read limit, so opt out explicitly to preserve + // behavior. + // + // Clone rather than append in place: callers share this slice with other + // clients. + opts = append(slices.Clone(opts), connect.WithReadMaxBytes(0)) var v1Client, v1alphaClient *reflectClient if reflectProtocol != ReflectProtocolGRPCV1 { v1alphaClient = connect.NewClient[reflectionv1.ServerReflectionRequest, reflectionv1.ServerReflectionResponse](httpClient, baseURL+"/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo", opts...)
diff --git a/private/buf/bufgen/generator.go b/private/buf/bufgen/generator.go index 283f85d..6d2ddd2 100644 --- a/private/buf/bufgen/generator.go +++ b/private/buf/bufgen/generator.go
@@ -42,6 +42,14 @@ "google.golang.org/protobuf/types/pluginpb" ) +// codeGenerationReadMaxBytes is the maximum size of a single GenerateCode +// response that remote generation will read. +// +// Remote generation returns a whole generated SDK in one message, so it is by +// far the largest response the CLI reads - observed peaks are in the hundreds +// of megabytes. It needs a much higher bound than other registry clients. +const codeGenerationReadMaxBytes = 1 << 30 // 1 GiB + type generator struct { logger *slog.Logger storageosProvider storageos.Provider @@ -360,7 +368,12 @@ } requests[i] = request } - codeGenerationService := connectclient.Make(g.clientConfig, remote, registryv1alpha1connect.NewCodeGenerationServiceClient) + codeGenerationService := connectclient.Make( + g.clientConfig, + remote, + registryv1alpha1connect.NewCodeGenerationServiceClient, + connect.WithReadMaxBytes(codeGenerationReadMaxBytes), + ) protoImage, err := bufimage.ImageToProtoImage(image) if err != nil { return nil, err
diff --git a/private/buf/bufstudioagent/plain_post_handler.go b/private/buf/bufstudioagent/plain_post_handler.go index 3e2f30b..5980998 100644 --- a/private/buf/bufstudioagent/plain_post_handler.go +++ b/private/buf/bufstudioagent/plain_post_handler.go
@@ -156,6 +156,11 @@ http.Error(w, err.Error(), http.StatusBadRequest) return } + // The agent proxies requests to whatever target the caller names, so there + // is no response size it can assume is illegitimate. Connect is expected to + // adopt a default per-message read limit, so opt out explicitly to preserve + // behavior. + clientOptions = append(clientOptions, connect.WithReadMaxBytes(0)) client := connect.NewClient[bytes.Buffer, bytes.Buffer]( httpClient, targetURL.String(),
diff --git a/private/pkg/connectclient/connectclient.go b/private/pkg/connectclient/connectclient.go index b89eeac..55666d2 100644 --- a/private/pkg/connectclient/connectclient.go +++ b/private/pkg/connectclient/connectclient.go
@@ -28,6 +28,7 @@ addressMapper func(string) string interceptors []connect.Interceptor authInterceptorProvider func(string) connect.UnaryInterceptorFunc + clientOptions []connect.ClientOption } // NewConfig creates a new client configuration with the given HTTP client @@ -67,11 +68,24 @@ } } +// WithClientOptions adds the given options to every client returned from this config. +// +// Options passed to Make are applied after these, so an individual client can +// override a default set here. +func WithClientOptions(clientOptions ...connect.ClientOption) ConfigOption { + return func(cfg *Config) { + cfg.clientOptions = clientOptions + } +} + // StubFactory is the type of a generated factory function, for creating Connect client stubs. type StubFactory[T any] func(connect.HTTPClient, string, ...connect.ClientOption) T // Make uses the given generated factory function to create a new connect client. -func Make[T any](cfg *Config, address string, factory StubFactory[T]) T { +// +// The given options are applied after any set with WithClientOptions, so they +// take precedence over the config's defaults. +func Make[T any](cfg *Config, address string, factory StubFactory[T], options ...connect.ClientOption) T { interceptors := slices.Clone(cfg.interceptors) if cfg.authInterceptorProvider != nil { interceptor := cfg.authInterceptorProvider(address) @@ -80,5 +94,9 @@ if cfg.addressMapper != nil { address = cfg.addressMapper(address) } - return factory(cfg.httpClient, address, connect.WithInterceptors(interceptors...)) + clientOptions := make([]connect.ClientOption, 0, len(cfg.clientOptions)+len(options)+1) + clientOptions = append(clientOptions, connect.WithInterceptors(interceptors...)) + clientOptions = append(clientOptions, cfg.clientOptions...) + clientOptions = append(clientOptions, options...) + return factory(cfg.httpClient, address, clientOptions...) }
diff --git a/private/pkg/connectclient/connectclient_test.go b/private/pkg/connectclient/connectclient_test.go new file mode 100644 index 0000000..d816351 --- /dev/null +++ b/private/pkg/connectclient/connectclient_test.go
@@ -0,0 +1,101 @@ +// Copyright 2020-2026 Buf Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package connectclient + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "testing" + + "connectrpc.com/connect" + reflectionv1 "github.com/bufbuild/buf/private/gen/proto/go/grpc/reflection/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testResponseSize = 64 * 1024 + +// TestMakeReadMaxBytes verifies that a read limit set on the Config applies to +// every client it builds, and that an option passed to Make overrides it. +// +// The CLI relies on both: a default bound is set once on the Config, and remote +// generation raises it for its own client. +func TestMakeReadMaxBytes(t *testing.T) { + t.Parallel() + + server := newReflectionServer(t) + factory := func(httpClient connect.HTTPClient, address string, options ...connect.ClientOption) *connect.Client[reflectionv1.ServerReflectionRequest, reflectionv1.ServerReflectionResponse] { + return connect.NewClient[reflectionv1.ServerReflectionRequest, reflectionv1.ServerReflectionResponse]( + httpClient, + address+"/grpc.reflection.v1.ServerReflection/ServerReflectionInfo", + options..., + ) + } + + t.Run("ConfigLimitApplies", func(t *testing.T) { + t.Parallel() + config := NewConfig(server.Client(), WithClientOptions(connect.WithReadMaxBytes(testResponseSize/2))) + _, err := call(t, Make(config, server.URL, factory)) + require.Error(t, err) + assert.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err)) + }) + + t.Run("MakeOptionOverridesConfigLimit", func(t *testing.T) { + t.Parallel() + config := NewConfig(server.Client(), WithClientOptions(connect.WithReadMaxBytes(testResponseSize/2))) + client := Make(config, server.URL, factory, connect.WithReadMaxBytes(testResponseSize*2)) + _, err := call(t, client) + require.NoError(t, err) + }) + + t.Run("NoLimitByDefault", func(t *testing.T) { + t.Parallel() + config := NewConfig(server.Client()) + _, err := call(t, Make(config, server.URL, factory)) + require.NoError(t, err) + }) +} + +func call( + t *testing.T, + client *connect.Client[reflectionv1.ServerReflectionRequest, reflectionv1.ServerReflectionResponse], +) (*connect.Response[reflectionv1.ServerReflectionResponse], error) { + t.Helper() + return client.CallUnary(t.Context(), connect.NewRequest(&reflectionv1.ServerReflectionRequest{})) +} + +// newReflectionServer returns a server whose single unary method always replies +// with a response of at least testResponseSize bytes. +func newReflectionServer(t *testing.T) *httptest.Server { + t.Helper() + handler := connect.NewUnaryHandler( + "/grpc.reflection.v1.ServerReflection/ServerReflectionInfo", + func( + _ context.Context, + _ *connect.Request[reflectionv1.ServerReflectionRequest], + ) (*connect.Response[reflectionv1.ServerReflectionResponse], error) { + return connect.NewResponse(reflectionv1.ServerReflectionResponse_builder{ + ValidHost: string(bytes.Repeat([]byte("a"), testResponseSize)), + }.Build()), nil + }, + ) + mux := http.NewServeMux() + mux.Handle("/grpc.reflection.v1.ServerReflection/ServerReflectionInfo", handler) + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + return server +}