[FAB-2364] Create common orderer configupdate path

https://jira.hyperledger.org/browse/FAB-2364 The multichannel code has a special handling for config update messages, but this same config update mechanism needs to be used for channel reconfiguration. This changeset adds a new set of common configuration update processing code. It does not hook it into the broadcast path, or remove the duplicated functionality in the multichain package, so as to make this CR easier to review. Change-Id: Idc746c666a455bfba917ec88a1acbec0a2e1d33e Signed-off-by: Jason Yellick <jyellick@us.ibm.com>
hyperledger · Feb 19, 2017 · 1219131 · 1219131
1 parent 3c812af
commit 1219131
Show file tree

Hide file tree

Showing 4 changed files with 367 additions and 0 deletions.
diff --git a/common/configtx/util.go b/common/configtx/util.go
@@ -35,6 +35,15 @@ func UnmarshalConfig(data []byte) (*cb.Config, error) {
 	return config, nil
 }
 
+// UnmarshalConfig attempts to unmarshal bytes to a *cb.Config
+func UnmarshalConfigOrPanic(data []byte) *cb.Config {
+	result, err := UnmarshalConfig(data)
+	if err != nil {
+		panic(err)
+	}
+	return result
+}
+
 // UnmarshalConfigUpdate attempts to unmarshal bytes to a *cb.ConfigUpdate
 func UnmarshalConfigUpdate(data []byte) (*cb.ConfigUpdate, error) {
 	configUpdate := &cb.ConfigUpdate{}
@@ -45,6 +54,15 @@ func UnmarshalConfigUpdate(data []byte) (*cb.ConfigUpdate, error) {
 	return configUpdate, nil
 }
 
+// UnmarshalConfigUpdate attempts to unmarshal bytes to a *cb.ConfigUpdate or panics
+func UnmarshalConfigUpdateOrPanic(data []byte) *cb.ConfigUpdate {
+	result, err := UnmarshalConfigUpdate(data)
+	if err != nil {
+		panic(err)
+	}
+	return result
+}
+
 // UnmarshalConfigUpdateEnvelope attempts to unmarshal bytes to a *cb.ConfigUpdate
 func UnmarshalConfigUpdateEnvelope(data []byte) (*cb.ConfigUpdateEnvelope, error) {
 	configUpdateEnvelope := &cb.ConfigUpdateEnvelope{}
@@ -55,6 +73,15 @@ func UnmarshalConfigUpdateEnvelope(data []byte) (*cb.ConfigUpdateEnvelope, error
 	return configUpdateEnvelope, nil
 }
 
+// UnmarshalConfigUpdateEnvelope attempts to unmarshal bytes to a *cb.ConfigUpdateEnvelope or panics
+func UnmarshalConfigUpdateEnvelopeOrPanic(data []byte) *cb.ConfigUpdateEnvelope {
+	result, err := UnmarshalConfigUpdateEnvelope(data)
+	if err != nil {
+		panic(err)
+	}
+	return result
+}
+
 // UnmarshalConfigEnvelope attempts to unmarshal bytes to a *cb.ConfigEnvelope
 func UnmarshalConfigEnvelope(data []byte) (*cb.ConfigEnvelope, error) {
 	configEnv := &cb.ConfigEnvelope{}
@@ -65,6 +92,15 @@ func UnmarshalConfigEnvelope(data []byte) (*cb.ConfigEnvelope, error) {
 	return configEnv, nil
 }
 
+// UnmarshalConfigEnvelope attempts to unmarshal bytes to a *cb.ConfigEnvelope or panics
+func UnmarshalConfigEnvelopeOrPanic(data []byte) *cb.ConfigEnvelope {
+	result, err := UnmarshalConfigEnvelope(data)
+	if err != nil {
+		panic(err)
+	}
+	return result
+}
+
 // ConfigEnvelopeFromBlock extract the config envelope from a config block
 func ConfigEnvelopeFromBlock(block *cb.Block) (*cb.ConfigEnvelope, error) {
 	if block.Data == nil || len(block.Data.Data) != 1 {

diff --git a/orderer/configupdate/configupdate.go b/orderer/configupdate/configupdate.go
@@ -0,0 +1,150 @@
+/*
+Copyright IBM Corp. 2017 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.
+*/
+
+// configupdate is an implementation of the broadcast.Proccessor interface
+// It facilitates the preprocessing of CONFIG_UPDATE transactions which can
+// generate either new CONFIG transactions or new channel creation
+// ORDERER_TRANSACTION messages.
+package configupdate
+
+import (
+	"fmt"
+
+	"github.com/hyperledger/fabric/common/configtx"
+	"github.com/hyperledger/fabric/common/crypto"
+	cb "github.com/hyperledger/fabric/protos/common"
+	"github.com/hyperledger/fabric/protos/utils"
+
+	"github.com/op/go-logging"
+)
+
+var logger = logging.MustGetLogger("orderer/multichain")
+
+const (
+	// These should eventually be derived from the channel support once enabled
+	msgVersion = int32(0)
+	epoch      = 0
+)
+
+// SupportManager provides a way for the Handler to look up the Support for a chain
+type SupportManager interface {
+	// GetChain gets the chain support for a given ChannelId
+	GetChain(chainID string) (Support, bool)
+}
+
+// Support enumerates a subset of the full channel support function which is required for this package
+type Support interface {
+	// ProposeConfigUpdate applies a CONFIG_UPDATE to an existing config to produce a *cb.ConfigEnvelope
+	ProposeConfigUpdate(env *cb.Envelope) (*cb.ConfigEnvelope, error)
+}
+
+type Processor struct {
+	signer          crypto.LocalSigner
+	manager         SupportManager
+	systemChannelID string
+}
+
+func New(systemChannelID string, supportManager SupportManager, signer crypto.LocalSigner) *Processor {
+	return &Processor{
+		systemChannelID: systemChannelID,
+		manager:         supportManager,
+		signer:          signer,
+	}
+}
+
+func channelID(env *cb.Envelope) (string, error) {
+	envPayload, err := utils.UnmarshalPayload(env.Payload)
+	if err != nil {
+		return "", fmt.Errorf("Failing to process config update because of payload unmarshaling error: %s", err)
+	}
+
+	if envPayload.Header == nil || envPayload.Header.ChannelHeader == nil || envPayload.Header.ChannelHeader.ChannelId == "" {
+		return "", fmt.Errorf("Failing to process config update because no channel ID was set")
+	}
+
+	return envPayload.Header.ChannelHeader.ChannelId, nil
+}
+
+// Process takes in an envelope of type CONFIG_UPDATE and proceses it
+// to transform it either into to a new channel creation request, or
+// into a channel CONFIG transaction (or errors on failure)
+func (p *Processor) Process(envConfigUpdate *cb.Envelope) (*cb.Envelope, error) {
+	channelID, err := channelID(envConfigUpdate)
+	if err != nil {
+		return nil, err
+	}
+
+	support, ok := p.manager.GetChain(channelID)
+	if ok {
+		logger.Debugf("Processing channel reconfiguration request for channel %s", channelID)
+		return p.existingChannelConfig(envConfigUpdate, channelID, support)
+	}
+
+	logger.Debugf("Processing channel creation request for channel %s", channelID)
+	return p.newChannelConfig(channelID, envConfigUpdate)
+}
+
+func (p *Processor) existingChannelConfig(envConfigUpdate *cb.Envelope, channelID string, support Support) (*cb.Envelope, error) {
+	configEnvelope, err := support.ProposeConfigUpdate(envConfigUpdate)
+	if err != nil {
+		return nil, err
+	}
+
+	return utils.CreateSignedEnvelope(cb.HeaderType_CONFIG, channelID, p.signer, configEnvelope, msgVersion, epoch)
+}
+
+func createInitialConfig(envConfigUpdate *cb.Envelope) (*cb.ConfigEnvelope, error) {
+	// TODO, for now, this assumes the update contains the entire initial config
+	// in the future, a subset of config needs to be allowed
+
+	configUpdatePayload, err := utils.UnmarshalPayload(envConfigUpdate.Payload)
+	if err != nil {
+		return nil, fmt.Errorf("Failing initial channel config creation because of payload unmarshaling error: %s", err)
+	}
+
+	configUpdateEnv, err := configtx.UnmarshalConfigUpdateEnvelope(configUpdatePayload.Data)
+	if err != nil {
+		return nil, fmt.Errorf("Failing initial channel config creation because of config update envelope unmarshaling error: %s", err)
+	}
+
+	configUpdate, err := configtx.UnmarshalConfigUpdate(configUpdateEnv.ConfigUpdate)
+	if err != nil {
+		return nil, fmt.Errorf("Failing initial channel config creation because of config update unmarshaling error: %s", err)
+	}
+
+	return &cb.ConfigEnvelope{
+		Config: &cb.Config{
+			Header:  configUpdate.Header,
+			Channel: configUpdate.WriteSet,
+		},
+
+		LastUpdate: envConfigUpdate,
+	}, nil
+}
+
+func (p *Processor) newChannelConfig(channelID string, envConfigUpdate *cb.Envelope) (*cb.Envelope, error) {
+	initialConfig, err := createInitialConfig(envConfigUpdate)
+	if err != nil {
+		return nil, err
+	}
+
+	envConfig, err := utils.CreateSignedEnvelope(cb.HeaderType_CONFIG, channelID, p.signer, initialConfig, msgVersion, epoch)
+	if err != nil {
+		return nil, err
+	}
+
+	return utils.CreateSignedEnvelope(cb.HeaderType_ORDERER_TRANSACTION, p.systemChannelID, p.signer, envConfig, msgVersion, epoch)
+}
diff --git a/orderer/configupdate/configupdate_test.go b/orderer/configupdate/configupdate_test.go
@@ -0,0 +1,153 @@
+/*
+Copyright IBM Corp. 2017 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 configupdate
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/hyperledger/fabric/common/configtx"
+	mockcrypto "github.com/hyperledger/fabric/common/mocks/crypto"
+	cb "github.com/hyperledger/fabric/protos/common"
+	"github.com/hyperledger/fabric/protos/utils"
+
+	"github.com/op/go-logging"
+	"github.com/stretchr/testify/assert"
+)
+
+func init() {
+	logging.SetLevel(logging.DEBUG, "")
+}
+
+type mockSupport struct {
+	ProposeConfigUpdateVal *cb.ConfigEnvelope
+}
+
+func (ms *mockSupport) ProposeConfigUpdate(env *cb.Envelope) (*cb.ConfigEnvelope, error) {
+	var err error
+	if ms.ProposeConfigUpdateVal == nil {
+		err = fmt.Errorf("Nil result implies error in mock")
+	}
+	return ms.ProposeConfigUpdateVal, err
+}
+
+type mockSupportManager struct {
+	GetChainVal *mockSupport
+}
+
+func (msm *mockSupportManager) GetChain(chainID string) (Support, bool) {
+	return msm.GetChainVal, msm.GetChainVal != nil
+}
+
+func TestChannelID(t *testing.T) {
+	makeEnvelope := func(payload *cb.Payload) *cb.Envelope {
+		return &cb.Envelope{
+			Payload: utils.MarshalOrPanic(payload),
+		}
+	}
+
+	testChannelID := "foo"
+
+	result, err := channelID(makeEnvelope(&cb.Payload{
+		Header: &cb.Header{
+			ChannelHeader: &cb.ChannelHeader{
+				ChannelId: testChannelID,
+			},
+		},
+	}))
+	assert.NoError(t, err, "Channel ID was present")
+	assert.Equal(t, testChannelID, result, "Channel ID was present")
+
+	_, err = channelID(makeEnvelope(&cb.Payload{
+		Header: &cb.Header{
+			ChannelHeader: &cb.ChannelHeader{},
+		},
+	}))
+	assert.Error(t, err, "Channel ID was empty")
+
+	_, err = channelID(makeEnvelope(&cb.Payload{
+		Header: &cb.Header{},
+	}))
+	assert.Error(t, err, "ChannelHeader was missing")
+
+	_, err = channelID(makeEnvelope(&cb.Payload{}))
+	assert.Error(t, err, "Header was missing")
+
+	_, err = channelID(&cb.Envelope{})
+	assert.Error(t, err, "Payload was missing")
+}
+
+const systemChannelID = "system_channel"
+const testUpdateChannelID = "update_channel"
+
+func newTestInstance() (*mockSupportManager, *Processor) {
+	msm := &mockSupportManager{}
+	return msm, New(systemChannelID, msm, mockcrypto.FakeLocalSigner)
+}
+
+func testConfigUpdate() *cb.Envelope {
+	ch := &cb.ChannelHeader{
+		ChannelId: testUpdateChannelID,
+	}
+
+	return &cb.Envelope{
+		Payload: utils.MarshalOrPanic(&cb.Payload{
+			Header: &cb.Header{
+				ChannelHeader: ch,
+			},
+			Data: utils.MarshalOrPanic(&cb.ConfigUpdateEnvelope{
+				ConfigUpdate: utils.MarshalOrPanic(&cb.ConfigUpdate{
+					Header:   ch,
+					WriteSet: cb.NewConfigGroup(),
+				}),
+			}),
+		}),
+	}
+}
+
+func TestExistingChannel(t *testing.T) {
+	msm, p := newTestInstance()
+
+	testUpdate := testConfigUpdate()
+
+	dummyResult := &cb.ConfigEnvelope{LastUpdate: &cb.Envelope{Payload: []byte("DUMMY")}}
+
+	msm.GetChainVal = &mockSupport{ProposeConfigUpdateVal: dummyResult}
+	env, err := p.Process(testUpdate)
+	assert.NoError(t, err, "Valid config update")
+	_ = utils.UnmarshalPayloadOrPanic(env.Payload)
+	assert.Equal(t, dummyResult, configtx.UnmarshalConfigEnvelopeOrPanic(utils.UnmarshalPayloadOrPanic(env.Payload).Data), "Valid config update")
+
+	msm.GetChainVal = &mockSupport{}
+	_, err = p.Process(testUpdate)
+	assert.Error(t, err, "Invald ProposeUpdate result")
+}
+
+func TestNewChannel(t *testing.T) {
+	_, p := newTestInstance()
+
+	testUpdate := testConfigUpdate()
+
+	env, err := p.Process(testUpdate)
+	assert.NoError(t, err, "Valid config update")
+
+	resultChan, err := channelID(env)
+	assert.NoError(t, err, "Invalid envelope produced")
+
+	assert.Equal(t, systemChannelID, resultChan, "Wrapper TX should be bound for system channel")
+	assert.Equal(t, int32(cb.HeaderType_ORDERER_TRANSACTION), utils.UnmarshalPayloadOrPanic(env.Payload).Header.ChannelHeader.Type, "Wrong wrapper tx type")
+}