Re-enable configtx handling

Before the orderer and peer components standardized on a proto message
format for shared communication, some proto messages were being
unmarshaled as the wrong type, resulting in messages which looked like
configuration transactions, but were not.  All these transactions were
then rejected because they did not apply correctly as configuration.

The fabric has since standardized on a shared set of protos, so it
should be safe to re-enable configuration transaction filtering.

This changeset re-enables the configuration filters.

Additionally, this changeset hooks the configuration transaction
execution (application) into the multichain manager writerinterceptor
path and adds tests for this path.

Change-Id: I5ba75861812d54e19fa6eeac98174215761f9ad2
Signed-off-by: Jason Yellick <jyellick@us.ibm.com>

orderer/multichain/chainsupport.go

@@ -20,12 +20,15 @@ import (
 	"github.com/hyperledger/fabric/orderer/common/blockcutter"
 	"github.com/hyperledger/fabric/orderer/common/broadcast"
 	"github.com/hyperledger/fabric/orderer/common/broadcastfilter"
+	"github.com/hyperledger/fabric/orderer/common/broadcastfilter/configfilter"
 	"github.com/hyperledger/fabric/orderer/common/configtx"
 	"github.com/hyperledger/fabric/orderer/common/deliver"
 	"github.com/hyperledger/fabric/orderer/common/policies"
 	"github.com/hyperledger/fabric/orderer/common/sharedconfig"
 	"github.com/hyperledger/fabric/orderer/rawledger"
 	cb "github.com/hyperledger/fabric/protos/common"
+
+	"github.com/golang/protobuf/proto"
 )
 
 // Consenter defines the backing ordering mechanism
@@ -112,7 +115,7 @@ func newChainSupport(configManager configtx.Manager, policyManager policies.Mana
 func createBroadcastRuleset(configManager configtx.Manager) *broadcastfilter.RuleSet {
 	return broadcastfilter.NewRuleSet([]broadcastfilter.Rule{
 		broadcastfilter.EmptyRejectRule,
-		// configfilter.New(configManager),
+		configfilter.New(configManager),
 		broadcastfilter.AcceptRule,
 	})
 }
@@ -153,17 +156,44 @@ func (cs *chainSupport) Enqueue(env *cb.Envelope) bool {
 	return cs.chain.Enqueue(env)
 }
 
+// writeInterceptor performs 'execution/processing' of blockContents before committing them to the normal passive ledger
+// This is intended to support reconfiguration transactions, and ultimately chain creation
 type writeInterceptor struct {
-	backing rawledger.Writer
+	configtxManager configtx.Manager
+	backing         rawledger.Writer
 }
 
-// TODO ultimately set write interception policy by config
-func newWriteInterceptor(configManager configtx.Manager, backing rawledger.Writer) *writeInterceptor {
+func newWriteInterceptor(configtxManager configtx.Manager, backing rawledger.Writer) *writeInterceptor {
 	return &writeInterceptor{
-		backing: backing,
+		backing:         backing,
+		configtxManager: configtxManager,
 	}
 }
 
 func (wi *writeInterceptor) Append(blockContents []*cb.Envelope, metadata [][]byte) *cb.Block {
+	// Note that in general any errors encountered in this path are fatal.
+	// The previous layers (broadcastfilters, blockcutter) should have scrubbed any invalid
+	// 'executable' transactions like config before committing via Append
+
+	if len(blockContents) == 1 {
+		payload := &cb.Payload{}
+		err := proto.Unmarshal(blockContents[0].Payload, payload)
+		if err != nil {
+			logger.Fatalf("Asked to write a malformed envelope to the chain: %s", err)
+		}
+
+		if payload.Header.ChainHeader.Type == int32(cb.HeaderType_CONFIGURATION_TRANSACTION) {
+			configEnvelope := &cb.ConfigurationEnvelope{}
+			err = proto.Unmarshal(payload.Data, configEnvelope)
+			if err != nil {
+				logger.Fatalf("Configuration envelope was malformed: %s", err)
+			}
+
+			err = wi.configtxManager.Apply(configEnvelope)
+			if err != nil {
+				logger.Fatalf("Error applying configuration transaction which was already validated: %s", err)
+			}
+		}
+	}
 	return wi.backing.Append(blockContents, metadata)
 }

orderer/multichain/chainsupport_mock_test.go

@@ -62,3 +62,28 @@ func (mch *mockChain) Start() {
 func (mch *mockChain) Halt() {
 	close(mch.queue)
 }
+
+type mockConfigtxManager struct {
+	config *cb.ConfigurationEnvelope
+}
+
+func (mcm *mockConfigtxManager) Apply(configtx *cb.ConfigurationEnvelope) error {
+	mcm.config = configtx
+	return nil
+}
+
+func (mcm *mockConfigtxManager) Validate(configtx *cb.ConfigurationEnvelope) error {
+	panic("Unimplemented")
+}
+
+func (mcm *mockConfigtxManager) ChainID() string {
+	panic("Unimplemented")
+}
+
+type mockLedgerWriter struct {
+}
+
+func (mlw *mockLedgerWriter) Append(blockContents []*cb.Envelope, metadata [][]byte) *cb.Block {
+	logger.Debugf("Committed block")
+	return nil
+}

orderer/multichain/chainsupport_test.go

@@ -0,0 +1,55 @@
+/*
+Copyright IBM Corp. 2016 All Rights Reserved.
+
+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 multichain
+
+import (
+	"testing"
+
+	cb "github.com/hyperledger/fabric/protos/common"
+)
+
+func TestCommitConfig(t *testing.T) {
+	mcm := &mockConfigtxManager{}
+	ctx1 := makeConfigTx("foo", 0)
+	wi := newWriteInterceptor(mcm, &mockLedgerWriter{})
+	wi.Append([]*cb.Envelope{ctx1}, nil)
+	if mcm.config == nil {
+		t.Fatalf("Should have applied configuration")
+	}
+}
+
+func TestIgnoreMultiConfig(t *testing.T) {
+	mcm := &mockConfigtxManager{}
+	ctx1 := makeConfigTx("foo", 0)
+	ctx2 := makeConfigTx("foo", 1)
+	wi := newWriteInterceptor(mcm, &mockLedgerWriter{})
+	wi.Append([]*cb.Envelope{ctx1, ctx2}, nil)
+	if mcm.config != nil {
+		t.Fatalf("Should not have applied configuration, we should only check batches with a single tx")
+	}
+}
+
+func TestIgnoreSingleNonConfig(t *testing.T) {
+	mcm := &mockConfigtxManager{}
+	ctx1 := makeNormalTx("foo", 0)
+	wi := newWriteInterceptor(mcm, &mockLedgerWriter{})
+	wi.Append([]*cb.Envelope{ctx1}, nil)
+	if mcm.config != nil {
+		t.Fatalf("Should not have applied configuration, it was a normal transaction")
+	}
+
+}