From 1fdfb9acfde11b9d986205c2cf8f418158a34f91 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C3=ADas=20Insaurralde?= Date: Thu, 22 Oct 2026 18:29:43 -0300 Subject: [PATCH 1/2] fix(controlplane): reject streaming RPCs the server does not implement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The control plane's gRPC server implements only unary RPCs. gRPC reflection, a local-development convenience that is disabled in production, is the single streaming service it serves. Add a stream interceptor that returns Unimplemented for any other streaming method instead of dispatching it to a handler. Clients use unary RPCs and the HTTP endpoints, so there is no functional change for them. Signed-off-by: Matías Insaurralde --- app/controlplane/internal/server/grpc.go | 20 ++++++ .../internal/server/grpc_stream_test.go | 63 +++++++++++++++++++ 2 files changed, 83 insertions(+) create mode 100644 app/controlplane/internal/server/grpc_stream_test.go diff --git a/app/controlplane/internal/server/grpc.go b/app/controlplane/internal/server/grpc.go index 31425ed68..cc2dace49 100644 --- a/app/controlplane/internal/server/grpc.go +++ b/app/controlplane/internal/server/grpc.go @@ -21,6 +21,7 @@ import ( "fmt" "net/url" "regexp" + "strings" v1 "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1" conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1" @@ -51,6 +52,8 @@ import ( grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" grpcLib "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type Opts struct { @@ -189,6 +192,11 @@ func newGRPCServer(opts *Opts, extraOpts ...grpc.ServerOption) *grpc.Server { grpc.UnaryInterceptor( protovalidateMiddleware.UnaryServerInterceptor(opts.Validator), ), + // The control plane implements only unary RPCs. Reject any streaming + // method other than gRPC reflection so the server does not dispatch + // streaming methods it does not serve. Reflection is a local-development + // convenience (e.g. grpcurl) and is disabled in production below. + grpc.StreamInterceptor(rejectUnsupportedStreams), }, extraOpts...) // gRPC reflection advertises the already-public API schema and exists as a @@ -241,6 +249,18 @@ func newGRPCServer(opts *Opts, extraOpts ...grpc.ServerOption) *grpc.Server { return srv } +// rejectUnsupportedStreams blocks streaming RPCs that the control plane does not +// implement. The server exposes only unary RPCs; gRPC reflection is the single +// streaming service it serves (a local-development convenience, disabled in +// production). Any other streaming method returns Unimplemented instead of being +// dispatched to a handler. +func rejectUnsupportedStreams(srv interface{}, ss grpcLib.ServerStream, info *grpcLib.StreamServerInfo, handler grpcLib.StreamHandler) error { + if strings.HasPrefix(info.FullMethod, "/grpc.reflection.") { + return handler(srv, ss) + } + return status.Error(codes.Unimplemented, "streaming methods are not supported") +} + func craftMiddleware(opts *Opts) []middleware.Middleware { middlewares := []middleware.Middleware{ // Give each request its own Sentry hub before anything can report to Sentry diff --git a/app/controlplane/internal/server/grpc_stream_test.go b/app/controlplane/internal/server/grpc_stream_test.go new file mode 100644 index 000000000..f26d39a60 --- /dev/null +++ b/app/controlplane/internal/server/grpc_stream_test.go @@ -0,0 +1,63 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// 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 server + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + grpcLib "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/health/grpc_health_v1" + reflectionpb "google.golang.org/grpc/reflection/grpc_reflection_v1" + "google.golang.org/grpc/status" +) + +// The control plane serves only unary RPCs. Streaming methods it does not +// implement are rejected with Unimplemented, while gRPC reflection (a +// development-only convenience) remains available. +func TestServerServesUnaryOnlyExceptReflection(t *testing.T) { + addr, _, _ := startSinglePortApp(t, nil) + conn, err := grpcLib.NewClient(addr, grpcLib.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + defer conn.Close() + + t.Run("unsupported streaming method returns Unimplemented", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream, err := grpc_health_v1.NewHealthClient(conn).Watch(ctx, &grpc_health_v1.HealthCheckRequest{}) + require.NoError(t, err) + _, err = stream.Recv() + assert.Equal(t, codes.Unimplemented, status.Code(err)) + }) + + t.Run("reflection stream is still served in development", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream, err := reflectionpb.NewServerReflectionClient(conn).ServerReflectionInfo(ctx) + require.NoError(t, err) + require.NoError(t, stream.Send(&reflectionpb.ServerReflectionRequest{ + MessageRequest: &reflectionpb.ServerReflectionRequest_ListServices{ListServices: ""}, + })) + resp, err := stream.Recv() + require.NoError(t, err) + assert.NotEmpty(t, resp.GetListServicesResponse().GetService()) + }) +} From 2d58c344883bfb067d729031770f58bcd02a9800 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C3=ADas=20Insaurralde?= Date: Thu, 22 Oct 2026 18:51:37 -0300 Subject: [PATCH 2/2] test(controlplane): exercise the stream interceptor on a registered method MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Register a streaming service in the test and assert the server returns Unimplemented without invoking its handler, so the test verifies the interceptor itself rather than gRPC's handling of an unknown method. Keep the check that gRPC reflection remains served in development. Signed-off-by: Matías Insaurralde --- .../internal/server/grpc_stream_test.go | 64 ++++++++++++++++--- 1 file changed, 55 insertions(+), 9 deletions(-) diff --git a/app/controlplane/internal/server/grpc_stream_test.go b/app/controlplane/internal/server/grpc_stream_test.go index f26d39a60..9b6e02b48 100644 --- a/app/controlplane/internal/server/grpc_stream_test.go +++ b/app/controlplane/internal/server/grpc_stream_test.go @@ -17,35 +17,81 @@ package server import ( "context" + "sync/atomic" "testing" "time" + conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" grpcLib "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/health/grpc_health_v1" reflectionpb "google.golang.org/grpc/reflection/grpc_reflection_v1" "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/emptypb" ) // The control plane serves only unary RPCs. Streaming methods it does not -// implement are rejected with Unimplemented, while gRPC reflection (a -// development-only convenience) remains available. +// implement are rejected with Unimplemented before reaching a handler, while +// gRPC reflection (a development-only convenience) remains available. func TestServerServesUnaryOnlyExceptReflection(t *testing.T) { - addr, _, _ := startSinglePortApp(t, nil) - conn, err := grpcLib.NewClient(addr, grpcLib.WithTransportCredentials(insecure.NewCredentials())) + off := false + opts := testServerOpts(t, &conf.Server{ + Grpc: &conf.Server_GRPC{Addr: testAnyAddr}, + Http: &conf.Server_HTTP{Addr: testAnyAddr}, + LegacyGrpcListener: &off, + }) + gs, err := NewGRPCServers(opts) + require.NoError(t, err) + + // Register a streaming service on the real server so the test exercises the + // stream interceptor on a registered method (rather than gRPC's handling of an + // unknown method). The handler records whether it is ever reached. + var handlerCalled atomic.Bool + const probeMethod = "/chainloop.test.v1.StreamProbe/Open" + gs.Handler.RegisterService(&grpcLib.ServiceDesc{ + ServiceName: "chainloop.test.v1.StreamProbe", + HandlerType: (*interface{})(nil), + Streams: []grpcLib.StreamDesc{ + { + StreamName: "Open", + Handler: func(_ interface{}, stream grpcLib.ServerStream) error { + handlerCalled.Store(true) + return stream.RecvMsg(new(emptypb.Empty)) + }, + ServerStreams: true, + ClientStreams: true, + }, + }, + }, nil) + + api, err := NewHTTPServer(opts, gs) + require.NoError(t, err) + endpoint, err := api.Endpoint() + require.NoError(t, err) + go func() { _ = api.Start(context.Background()) }() + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = api.Stop(ctx) + }) + + conn, err := grpcLib.NewClient(endpoint.Host, grpcLib.WithTransportCredentials(insecure.NewCredentials())) require.NoError(t, err) defer conn.Close() - t.Run("unsupported streaming method returns Unimplemented", func(t *testing.T) { + t.Run("registered streaming method is rejected before its handler runs", func(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - stream, err := grpc_health_v1.NewHealthClient(conn).Watch(ctx, &grpc_health_v1.HealthCheckRequest{}) - require.NoError(t, err) - _, err = stream.Recv() + desc := &grpcLib.StreamDesc{StreamName: "Open", ServerStreams: true, ClientStreams: true} + stream, err := conn.NewStream(ctx, desc, probeMethod) + if err == nil { + _ = stream.SendMsg(&emptypb.Empty{}) + err = stream.RecvMsg(&emptypb.Empty{}) + } assert.Equal(t, codes.Unimplemented, status.Code(err)) + assert.False(t, handlerCalled.Load(), "the streaming handler must not be invoked") }) t.Run("reflection stream is still served in development", func(t *testing.T) {