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..9b6e02b48 --- /dev/null +++ b/app/controlplane/internal/server/grpc_stream_test.go @@ -0,0 +1,109 @@ +// +// 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" + "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" + 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 before reaching a handler, while +// gRPC reflection (a development-only convenience) remains available. +func TestServerServesUnaryOnlyExceptReflection(t *testing.T) { + 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("registered streaming method is rejected before its handler runs", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + 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) { + 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()) + }) +}