CVE-2022-31121: Improper Input Validation (CWE-20)
value: Hyperledger Fabric is a permissioned distributed ledger framework. In affected versions if a consensus client sends a malformed consensus request to an orderer it may crash the orderer node. A fix has been added in commit 0f1835949 which checks for missing consensus messages and returns an error to the consensus client should the message be missing. Users are advised to upgrade to versions 2.2.7 or v2.4.5. There are no known workarounds for this issue.
CWE description: The product receives input or data, but it does
        not validate or incorrectly validates that the input has the
        properties that are required to process the data safely and
        correctly.
CVSS v3 base score: 7.5
Published: 2022-07-07T18:15Z
Repository: https://github.com/hyperledger/fabric
Language: Go
Files changed: orderer/common/cluster/comm_test.go, orderer/common/cluster/service.go
Commit date: 2022-06-21 08:22:54 -0400
Commit message: Check if inner consensus message is missing

Change-Id: I06b466e6ff6d43f2b9804dd21185241716356050
Signed-off-by: Yacov Manevich <yacovm@il.ibm.com>
(cherry picked from commit 6e5e693a2d12aabee4bcf103ab7de3df4ccc49f9)

Diff: diff --git a/orderer/common/cluster/comm_test.go b/orderer/common/cluster/comm_test.go
index fff537994..3b9b5f0e1 100644
--- a/orderer/common/cluster/comm_test.go
+++ b/orderer/common/cluster/comm_test.go
@@ -118,10 +118,15 @@ func (*mockChannelExtractor) TargetChannel(msg proto.Message) string {
 	default:
 		return ""
 	}
 }
 
+type clusterServer interface {
+	// Step passes an implementation-specific message to another cluster member.
+	Step(server orderer.Cluster_StepServer) error
+}
+
 type clusterNode struct {
 	lock         sync.Mutex
 	frozen       bool
 	freezeCond   sync.Cond
 	dialer       *cluster.PredicateDialer
@@ -130,10 +135,11 @@ type clusterNode struct {
 	srv          *comm_utils.GRPCServer
 	bindAddress  string
 	clientConfig comm_utils.ClientConfig
 	serverConfig comm_utils.ServerConfig
 	c            *cluster.Comm
+	dispatcher   clusterServer
 }
 
 func (cn *clusterNode) Step(stream orderer.Cluster_StepServer) error {
 	cn.waitIfFrozen()
 	req, err := stream.Recv()
@@ -177,11 +183,11 @@ func (cn *clusterNode) resurrect() {
 	gRPCServer, err := comm_utils.NewGRPCServer(cn.bindAddress, cn.serverConfig)
 	if err != nil {
 		panic(fmt.Errorf("failed starting gRPC server: %v", err))
 	}
 	cn.srv = gRPCServer
-	orderer.RegisterClusterServer(gRPCServer.Server(), cn)
+	orderer.RegisterClusterServer(gRPCServer.Server(), cn.dispatcher)
 	go cn.srv.Start()
 }
 
 func (cn *clusterNode) stop() {
 	cn.srv.Stop()
@@ -255,10 +261,14 @@ func newTestNodeWithMetrics(t *testing.T, metrics cluster.MetricsProvider, tlsCo
 			ClientTLSCert: clientKeyPair.TLSCert.Raw,
 		},
 		srv: gRPCServer,
 	}
 
+	if tstSrv.dispatcher == nil {
+		tstSrv.dispatcher = tstSrv
+	}
+
 	tstSrv.freezeCond.L = &tstSrv.lock
 
 	compareCert := cluster.CachePublicKeyComparisons(func(a, b []byte) bool {
 		return crypto.CertificatesWithSamePublicKey(a, b) == nil
 	})
@@ -273,11 +283,11 @@ func newTestNodeWithMetrics(t *testing.T, metrics cluster.MetricsProvider, tlsCo
 		Connections:             cluster.NewConnectionStore(dialer, tlsConnGauge),
 		Metrics:                 cluster.NewMetrics(metrics),
 		CompareCertificate:      compareCert,
 	}
 
-	orderer.RegisterClusterServer(gRPCServer.Server(), tstSrv)
+	orderer.RegisterClusterServer(gRPCServer.Server(), tstSrv.dispatcher)
 	go gRPCServer.Start()
 	return tstSrv
 }
 
 func newTestNode(t *testing.T) *clusterNode {
@@ -479,10 +489,56 @@ func TestBlockingSend(t *testing.T) {
 			}
 		})
 	}
 }
 
+func TestEmptyRequest(t *testing.T) {
+	// Scenario: Ensures empty messages are discarded and an error is returned
+	// back to the sender.
+
+	node1 := newTestNode(t)
+	node2 := newTestNode(t)
+
+	node2.srv.Stop()
+	svc := &cluster.Service{
+		StepLogger: flogging.MustGetLogger("test"),
+		Logger:     flogging.MustGetLogger("test"),
+		StreamCountReporter: &cluster.StreamCountReporter{
+			Metrics: cluster.NewMetrics(&disabled.Provider{}),
+		},
+		Dispatcher: node2.c,
+	}
+	node2.dispatcher = svc
+
+	// Sleep to let the gRPC service be closed
+	time.Sleep(time.Second)
+
+	// Resurrect the node with the new dispatcher
+	node2.resurrect()
+
+	defer node1.stop()
+	defer node2.stop()
+
+	config := []cluster.RemoteNode{node1.nodeInfo, node2.nodeInfo}
+	node1.c.Configure(testChannel, config)
+	node2.c.Configure(testChannel, config)
+
+	assertBiDiCommunication(t, node1, node2, testReq)
+
+	rm, err := node1.c.Remote(testChannel, node2.nodeInfo.ID)
+	require.NoError(t, err)
+
+	stream, err := rm.NewStream(time.Second * 10)
+	require.NoError(t, err)
+
+	err = stream.Send(&orderer.StepRequest{})
+	require.NoError(t, err)
+
+	_, err = stream.Recv()
+	require.Error(t, err, "message is neither a Submit nor a Consensus request")
+}
+
 func TestBasic(t *testing.T) {
 	// Scenario: Basic test that spawns 2 nodes and sends each other
 	// messages that are expected to be echoed back
 
 	node1 := newTestNode(t)
diff --git a/orderer/common/cluster/service.go b/orderer/common/cluster/service.go
index 214b9525d..cc8302028 100644
--- a/orderer/common/cluster/service.go
+++ b/orderer/common/cluster/service.go
@@ -12,10 +12,11 @@ import (
 	"time"
 
 	"github.com/hyperledger/fabric-protos-go/orderer"
 	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/hyperledger/fabric/common/util"
+	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 )
 
 //go:generate mockery -dir . -name Dispatcher -case underscore -output ./mocks/
@@ -88,14 +89,15 @@ func (s *Service) handleMessage(stream StepStream, addr string, exp *certificate
 
 	if submitReq := request.GetSubmitRequest(); submitReq != nil {
 		nodeName := commonNameFromContext(stream.Context())
 		s.Logger.Debugf("Received message from %s(%s): %v", nodeName, addr, requestAsString(request))
 		return s.handleSubmit(submitReq, stream, addr)
+	} else if consensusReq := request.GetConsensusRequest(); consensusReq != nil {
+		return s.Dispatcher.DispatchConsensus(stream.Context(), request.GetConsensusRequest())
 	}
 
-	// Else, it's a consensus message.
-	return s.Dispatcher.DispatchConsensus(stream.Context(), request.GetConsensusRequest())
+	return errors.Errorf("message is neither a Submit nor a Consensus request")
 }
 
 func (s *Service) handleSubmit(request *orderer.SubmitRequest, stream StepStream, addr string) error {
 	err := s.Dispatcher.DispatchSubmit(stream.Context(), request)
 	if err != nil {

Vulnerable code: 	orderer.RegisterClusterServer(gRPCServer.Server(), cn)
	orderer.RegisterClusterServer(gRPCServer.Server(), tstSrv)
	// Else, it's a consensus message.
	return s.Dispatcher.DispatchConsensus(stream.Context(), request.GetConsensusRequest())
Fixed code: type clusterServer interface {
	// Step passes an implementation-specific message to another cluster member.
	Step(server orderer.Cluster_StepServer) error
}

	dispatcher   clusterServer
	orderer.RegisterClusterServer(gRPCServer.Server(), cn.dispatcher)
	if tstSrv.dispatcher == nil {
		tstSrv.dispatcher = tstSrv
	}

	orderer.RegisterClusterServer(gRPCServer.Server(), tstSrv.dispatcher)
func TestEmptyRequest(t *testing.T) {
	// Scenario: Ensures empty messages are discarded and an error is returned
	// back to the sender.

	node1 := newTestNode(t)
	node2 := newTestNode(t)

	node2.srv.Stop()
	svc := &cluster.Service{
		StepLogger: flogging.MustGetLogger("test"),
		Logger:     flogging.MustGetLogger("test"),
		StreamCountReporter: &cluster.StreamCountReporter{
			Metrics: cluster.NewMetrics(&disabled.Provider{}),
		},
		Dispatcher: node2.c,
	}
	node2.dispatcher = svc

	// Sleep to let the gRPC service be closed
	time.Sleep(time.Second)

	// Resurrect the node with the new dispatcher
	node2.resurrect()

	defer node1.stop()
	defer node2.stop()

	config := []cluster.RemoteNode{node1.nodeInfo, node2.nodeInfo}
	node1.c.Configure(testChannel, config)
	node2.c.Configure(testChannel, config)

	assertBiDiCommunication(t, node1, node2, testReq)

	rm, err := node1.c.Remote(testChannel, node2.nodeInfo.ID)
	require.NoError(t, err)

	stream, err := rm.NewStream(time.Second * 10)
	require.NoError(t, err)

	err = stream.Send(&orderer.StepRequest{})
	require.NoError(t, err)

	_, err = stream.Recv()
	require.Error(t, err, "message is neither a Submit nor a Consensus request")
}

	"github.com/pkg/errors"
	} else if consensusReq := request.GetConsensusRequest(); consensusReq != nil {
		return s.Dispatcher.DispatchConsensus(stream.Context(), request.GetConsensusRequest())
	return errors.Errorf("message is neither a Submit nor a Consensus request")