[FAB-1568] Add orderer shared config ingresspolicy

https://jira.hyperledger.org/browse/FAB-1568

This is the third a series meant to enable broadcast filtering based on
policy.

This changeset adds an IngressPolicy orderer configuration item, which
references a policy by name.  It is not currently used, but will
eventually be used by the signature filter.

Change-Id: I52598fa395d95112accfd8baa69c9be3d9cd6bdd
Signed-off-by: Jason Yellick <[email protected]>

Author: Jason Yellick

orderer/common/sharedconfig/sharedconfig.go

@@ -45,6 +45,9 @@ const (
 
 	// KafkaBrokersKey is the cb.ConfigurationItem type key name for the KafkaBrokers message
 	KafkaBrokersKey = "KafkaBrokers"
+
+	// IngressPolicyKey is the cb.ConfigurationItem type key name for the IngressPolicy message
+	IngressPolicyKey = "IngressPolicy"
 )
 
 var logger = logging.MustGetLogger("orderer/common/sharedconfig")
@@ -71,6 +74,9 @@ type Manager interface {
 	// Kafka brokers, i.e. this is not necessarily the entire set of Kafka brokers
 	// used for ordering
 	KafkaBrokers() []string
+
+	// IngressPolicy returns the name of the policy to validate incoming broadcast messages against
+	IngressPolicy() string
 }
 
 type ordererConfig struct {
@@ -79,6 +85,7 @@ type ordererConfig struct {
 	batchTimeout  time.Duration
 	chainCreators []string
 	kafkaBrokers  []string
+	ingressPolicy string
 }
 
 // ManagerImpl is an implementation of Manager and configtx.ConfigHandler
@@ -123,6 +130,11 @@ func (pm *ManagerImpl) KafkaBrokers() []string {
 	return pm.config.kafkaBrokers
 }
 
+// IngressPolicy returns the name of the policy to validate incoming broadcast messages against
+func (pm *ManagerImpl) IngressPolicy() string {
+	return pm.config.ingressPolicy
+}
+
 // BeginConfig is used to start a new configuration proposal
 func (pm *ManagerImpl) BeginConfig() {
 	if pm.pendingConfig != nil {
@@ -194,6 +206,12 @@ func (pm *ManagerImpl) ProposeConfig(configItem *cb.ConfigurationItem) error {
 			return fmt.Errorf("Unmarshaling error for ChainCreator: %s", err)
 		}
 		pm.pendingConfig.chainCreators = chainCreators.Policies
+	case IngressPolicyKey:
+		ingressPolicy := &ab.IngressPolicy{}
+		if err := proto.Unmarshal(configItem.Value, ingressPolicy); err != nil {
+			return fmt.Errorf("Unmarshaling error for IngressPolicy: %s", err)
+		}
+		pm.pendingConfig.ingressPolicy = ingressPolicy.Name
 	case KafkaBrokersKey:
 		kafkaBrokers := &ab.KafkaBrokers{}
 		if err := proto.Unmarshal(configItem.Value, kafkaBrokers); err != nil {

orderer/common/sharedconfig/sharedconfig_test.go

@@ -298,3 +298,44 @@ func TestKafkaBrokers(t *testing.T) {
 		return
 	}
 }
+
+func TestIngressPolicy(t *testing.T) {
+	endPolicy := "foo"
+	invalidMessage :=
+		&cb.ConfigurationItem{
+			Type:  cb.ConfigurationItem_Orderer,
+			Key:   IngressPolicyKey,
+			Value: []byte("Garbage Data"),
+		}
+	validMessage := &cb.ConfigurationItem{
+		Type:  cb.ConfigurationItem_Orderer,
+		Key:   IngressPolicyKey,
+		Value: utils.MarshalOrPanic(&ab.IngressPolicy{Name: endPolicy}),
+	}
+	m := NewManagerImpl()
+	m.BeginConfig()
+
+	err := m.ProposeConfig(validMessage)
+	if err != nil {
+		t.Fatalf("Error applying valid config: %s", err)
+	}
+
+	m.CommitConfig()
+	m.BeginConfig()
+
+	err = m.ProposeConfig(invalidMessage)
+	if err == nil {
+		t.Fatalf("Should have failed on invalid message")
+	}
+
+	err = m.ProposeConfig(validMessage)
+	if err != nil {
+		t.Fatalf("Error re-applying valid config: %s", err)
+	}
+
+	m.CommitConfig()
+
+	if nowPolicy := m.IngressPolicy(); nowPolicy != endPolicy {
+		t.Fatalf("IngressPolicy should have ended as %s but was %s", endPolicy, nowPolicy)
+	}
+}

orderer/mocks/sharedconfig/sharedconfig.go

@@ -31,6 +31,8 @@ type Manager struct {
 	ChainCreatorsVal []string
 	// KafkaBrokersVal is returned as the result of KafkaBrokers()
 	KafkaBrokersVal []string
+	// IngressPolicyVal is returned as the result of IngressPolicy()
+	IngressPolicyVal string
 }
 
 // ConsensusType returns the ConsensusTypeVal
@@ -57,3 +59,8 @@ func (scm *Manager) ChainCreators() []string {
 func (scm *Manager) KafkaBrokers() []string {
 	return scm.KafkaBrokersVal
 }
+
+// IngressPolicy returns the IngressPolicyVal
+func (scm *Manager) IngressPolicy() string {
+	return scm.IngressPolicyVal
+}

orderer/multichain/systemchain_test.go

@@ -19,13 +19,13 @@ package multichain
 import (
 	"reflect"
 	"testing"
-	"time"
 
 	"github.com/hyperledger/fabric/common/policies"
 	coreutil "github.com/hyperledger/fabric/core/util"
 	"github.com/hyperledger/fabric/orderer/common/bootstrap/provisional"
 	"github.com/hyperledger/fabric/orderer/common/filter"
 	"github.com/hyperledger/fabric/orderer/common/sharedconfig"
+	mocksharedconfig "github.com/hyperledger/fabric/orderer/mocks/sharedconfig"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
 	"github.com/hyperledger/fabric/protos/utils"
@@ -47,41 +47,17 @@ func (mpm *mockPolicyManager) GetPolicy(id string) (policies.Policy, bool) {
 	return mpm.mp, mpm.mp != nil
 }
 
-type mockSharedConfig struct {
-	chainCreators []string
-}
-
-func (msc *mockSharedConfig) ConsensusType() string {
-	panic("Unimplemented")
-}
-
-func (msc *mockSharedConfig) BatchSize() *ab.BatchSize {
-	panic("Unimplemented")
-}
-
-func (msc *mockSharedConfig) BatchTimeout() time.Duration {
-	panic("Unimplemented")
-}
-
-func (msc *mockSharedConfig) ChainCreators() []string {
-	return msc.chainCreators
-}
-
-func (msc *mockSharedConfig) KafkaBrokers() []string {
-	panic("Unimplemented")
-}
-
 type mockSupport struct {
 	mpm     *mockPolicyManager
-	msc     *mockSharedConfig
+	msc     *mocksharedconfig.Manager
 	chainID string
 	queue   []*cb.Envelope
 }
 
 func newMockSupport(chainID string) *mockSupport {
 	return &mockSupport{
 		mpm:     &mockPolicyManager{},
-		msc:     &mockSharedConfig{},
+		msc:     &mocksharedconfig.Manager{},
 		chainID: chainID,
 	}
 }
@@ -129,7 +105,7 @@ func TestGoodProposal(t *testing.T) {
 	newChainID := "NewChainID"
 
 	mcc := newMockChainCreator()
-	mcc.ms.msc.chainCreators = []string{provisional.AcceptAllPolicyKey}
+	mcc.ms.msc.ChainCreatorsVal = []string{provisional.AcceptAllPolicyKey}
 	mcc.ms.mpm.mp = &mockPolicy{}
 
 	chainCreateTx := &cb.ConfigurationItem{
@@ -214,7 +190,7 @@ func TestProposalWithMissingPolicy(t *testing.T) {
 	newChainID := "NewChainID"
 
 	mcc := newMockChainCreator()
-	mcc.ms.msc.chainCreators = []string{provisional.AcceptAllPolicyKey}
+	mcc.ms.msc.ChainCreatorsVal = []string{provisional.AcceptAllPolicyKey}
 
 	chainCreateTx := &cb.ConfigurationItem{
 		Key:  utils.CreationPolicyKey,
@@ -238,7 +214,7 @@ func TestProposalWithBadDigest(t *testing.T) {
 
 	mcc := newMockChainCreator()
 	mcc.ms.mpm.mp = &mockPolicy{}
-	mcc.ms.msc.chainCreators = []string{provisional.AcceptAllPolicyKey}
+	mcc.ms.msc.ChainCreatorsVal = []string{provisional.AcceptAllPolicyKey}
 
 	chainCreateTx := &cb.ConfigurationItem{
 		Key:  utils.CreationPolicyKey,

protos/orderer/ab.pb.go

@@ -61,6 +61,11 @@ message CreationPolicy {
     bytes digest = 2;
 }
 
+// IngressPolicy is the name of the policy which incoming Broadcast messages are filtered against
+message IngressPolicy {
+    string name = 1;
+}
+
 message ChainCreators {
     // A list of policies, any of which may be specified as the chain creation
     // policy in a chain creation request