[FAB-1576] Enable deliver request policy filtering

https://jira.hyperledger.org/browse/FAB-1576 This changeset completes the enablement of deliver request filtering based on the orderer egress policy. Change-Id: I62b3dbcef2f733d2520e5bf39ef4c11fcb6bfcb5 Signed-off-by: Jason Yellick <jyellick@us.ibm.com>
hyperledger · Jan 11, 2017 · 16e170c · 16e170c
1 parent 3737e31
commit 16e170c
Show file tree

Hide file tree

Showing 5 changed files with 87 additions and 17 deletions.
diff --git a/orderer/common/deliver/deliver.go b/orderer/common/deliver/deliver.go
@@ -20,6 +20,9 @@ import (
 	"fmt"
 
 	"github.com/hyperledger/fabric/common/policies"
+	"github.com/hyperledger/fabric/orderer/common/filter"
+	"github.com/hyperledger/fabric/orderer/common/sharedconfig"
+	"github.com/hyperledger/fabric/orderer/common/sigfilter"
 	"github.com/hyperledger/fabric/orderer/rawledger"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
@@ -47,6 +50,9 @@ type Support interface {
 
 	// Reader returns the chain Reader for the chain
 	Reader() rawledger.Reader
+
+	// SharedConfig returns the shared config manager for this chain
+	SharedConfig() sharedconfig.Manager
 }
 
 type deliverServer struct {
@@ -86,7 +92,11 @@ func (ds *deliverServer) Handle(srv ab.AtomicBroadcast_DeliverServer) error {
 			return sendStatusReply(srv, cb.Status_NOT_FOUND)
 		}
 
-		// XXX add deliver authorization checking
+		sf := sigfilter.New(chain.SharedConfig().EgressPolicy, chain.PolicyManager())
+		result, _ := sf.Apply(envelope)
+		if result != filter.Forward {
+			return sendStatusReply(srv, cb.Status_FORBIDDEN)
+		}
 
 		seekInfo := &ab.SeekInfo{}
 		if err = proto.Unmarshal(payload.Data, seekInfo); err != nil {

diff --git a/orderer/common/deliver/deliver_test.go b/orderer/common/deliver/deliver_test.go
@@ -23,7 +23,10 @@ import (
 
 	"github.com/hyperledger/fabric/common/policies"
 	"github.com/hyperledger/fabric/orderer/common/bootstrap/provisional"
+	"github.com/hyperledger/fabric/orderer/common/sharedconfig"
 	"github.com/hyperledger/fabric/orderer/localconfig"
+	mockpolicies "github.com/hyperledger/fabric/orderer/mocks/policies"
+	mocksharedconfig "github.com/hyperledger/fabric/orderer/mocks/sharedconfig"
 	"github.com/hyperledger/fabric/orderer/rawledger"
 	"github.com/hyperledger/fabric/orderer/rawledger/ramledger"
 	cb "github.com/hyperledger/fabric/protos/common"
@@ -81,11 +84,13 @@ func (mm *mockSupportManager) GetChain(chainID string) (Support, bool) {
 }
 
 type mockSupport struct {
-	ledger rawledger.ReadWriter
+	ledger        rawledger.ReadWriter
+	sharedConfig  *mocksharedconfig.Manager
+	policyManager *mockpolicies.Manager
 }
 
 func (mcs *mockSupport) PolicyManager() policies.Manager {
-	panic("Unimplemented")
+	return mcs.policyManager
 }
 
 func (mcs *mockSupport) Reader() rawledger.Reader {
@@ -99,13 +104,19 @@ func NewRAMLedger() rawledger.ReadWriter {
 	return rl
 }
 
+func (mcs *mockSupport) SharedConfig() sharedconfig.Manager {
+	return mcs.sharedConfig
+}
+
 func newMockMultichainManager() *mockSupportManager {
 	rl := NewRAMLedger()
 	mm := &mockSupportManager{
 		chains: make(map[string]*mockSupport),
 	}
-	mm.chains[string(systemChainID)] = &mockSupport{
-		ledger: rl,
+	mm.chains[systemChainID] = &mockSupport{
+		ledger:        rl,
+		sharedConfig:  &mocksharedconfig.Manager{},
+		policyManager: &mockpolicies.Manager{Policy: &mockpolicies.Policy{}},
 	}
 	return mm
 }
@@ -134,7 +145,7 @@ func makeSeek(chainID string, seekInfo *ab.SeekInfo) *cb.Envelope {
 func TestOldestSeek(t *testing.T) {
 	mm := newMockMultichainManager()
 	for i := 1; i < ledgerSize; i++ {
-		ledger := mm.chains[string(systemChainID)].ledger
+		ledger := mm.chains[systemChainID].ledger
 		ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", i))}}))
 	}
 
@@ -173,7 +184,7 @@ func TestOldestSeek(t *testing.T) {
 func TestNewestSeek(t *testing.T) {
 	mm := newMockMultichainManager()
 	for i := 1; i < ledgerSize; i++ {
-		ledger := mm.chains[string(systemChainID)].ledger
+		ledger := mm.chains[systemChainID].ledger
 		ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", i))}}))
 	}
 
@@ -205,7 +216,7 @@ func TestNewestSeek(t *testing.T) {
 func TestSpecificSeek(t *testing.T) {
 	mm := newMockMultichainManager()
 	for i := 1; i < ledgerSize; i++ {
-		ledger := mm.chains[string(systemChainID)].ledger
+		ledger := mm.chains[systemChainID].ledger
 		ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", i))}}))
 	}
 
@@ -240,10 +251,36 @@ func TestSpecificSeek(t *testing.T) {
 	}
 }
 
+func TestUnauthorizedSeek(t *testing.T) {
+	mm := newMockMultichainManager()
+	for i := 1; i < ledgerSize; i++ {
+		ledger := mm.chains[systemChainID].ledger
+		ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", i))}}))
+	}
+	mm.chains[systemChainID].policyManager.Policy.Err = fmt.Errorf("Fail to evaluate policy")
+
+	m := newMockD()
+	defer close(m.recvChan)
+	ds := NewHandlerImpl(mm)
+
+	go ds.Handle(m)
+
+	m.recvChan <- makeSeek(systemChainID, &ab.SeekInfo{Start: seekSpecified(uint64(0)), Stop: seekSpecified(uint64(0)), Behavior: ab.SeekInfo_BLOCK_UNTIL_READY})
+
+	select {
+	case deliverReply := <-m.sendChan:
+		if deliverReply.GetStatus() != cb.Status_FORBIDDEN {
+			t.Fatalf("Received wrong error on the reply channel")
+		}
+	case <-time.After(time.Second):
+		t.Fatalf("Timed out waiting to get all blocks")
+	}
+}
+
 func TestBadSeek(t *testing.T) {
 	mm := newMockMultichainManager()
 	for i := 1; i < ledgerSize; i++ {
-		ledger := mm.chains[string(systemChainID)].ledger
+		ledger := mm.chains[systemChainID].ledger
 		ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", i))}}))
 	}
 
@@ -268,7 +305,7 @@ func TestBadSeek(t *testing.T) {
 func TestFailFastSeek(t *testing.T) {
 	mm := newMockMultichainManager()
 	for i := 1; i < ledgerSize; i++ {
-		ledger := mm.chains[string(systemChainID)].ledger
+		ledger := mm.chains[systemChainID].ledger
 		ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", i))}}))
 	}
 
@@ -302,7 +339,7 @@ func TestFailFastSeek(t *testing.T) {
 func TestBlockingSeek(t *testing.T) {
 	mm := newMockMultichainManager()
 	for i := 1; i < ledgerSize; i++ {
-		ledger := mm.chains[string(systemChainID)].ledger
+		ledger := mm.chains[systemChainID].ledger
 		ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", i))}}))
 	}
 
@@ -329,7 +366,7 @@ func TestBlockingSeek(t *testing.T) {
 	case <-time.After(50 * time.Millisecond):
 	}
 
-	ledger := mm.chains[string(systemChainID)].ledger
+	ledger := mm.chains[systemChainID].ledger
 	ledger.Append(rawledger.CreateNextBlock(ledger, []*cb.Envelope{&cb.Envelope{Payload: []byte(fmt.Sprintf("%d", ledgerSize+1))}}))
 
 	select {

diff --git a/orderer/common/sigfilter/sigfilter.go b/orderer/common/sigfilter/sigfilter.go
@@ -33,6 +33,9 @@ type sigFilter struct {
 
 // New creates a new signature filter, at every evaluation, the policySource is called
 // just before evaluation to get the policy name to use when evaluating the filter
+// In general, both the policy name and the policy itself are mutable, this is why
+// not only the policy is retrieved at each invocation, but also the name of which
+// policy to retrieve
 func New(policySource func() string, policyManager policies.Manager) filter.Rule {
 	return &sigFilter{
 		policySource:  policySource,

diff --git a/orderer/multichain/chainsupport.go b/orderer/multichain/chainsupport.go
@@ -22,7 +22,6 @@ import (
 	"github.com/hyperledger/fabric/common/util"
 	"github.com/hyperledger/fabric/orderer/common/blockcutter"
 	"github.com/hyperledger/fabric/orderer/common/broadcast"
-	"github.com/hyperledger/fabric/orderer/common/deliver"
 	"github.com/hyperledger/fabric/orderer/common/filter"
 	"github.com/hyperledger/fabric/orderer/common/sharedconfig"
 	"github.com/hyperledger/fabric/orderer/common/sigfilter"
@@ -70,8 +69,16 @@ type ConsenterSupport interface {
 
 // ChainSupport provides a wrapper for the resources backing a chain
 type ChainSupport interface {
+	// This interface is actually the union with the deliver.Support but because of a golang
+	// limitation https://github.com/golang/go/issues/6977 the methods must be explicitly declared
+
+	// PolicyManager returns the current policy manager as specified by the chain configuration
+	PolicyManager() policies.Manager
+
+	// Reader returns the chain Reader for the chain
+	Reader() rawledger.Reader
+
 	broadcast.Support
-	deliver.Support
 	ConsenterSupport
 
 	// ConfigTxManager returns the corresponding configtx.Manager for this chain

diff --git a/orderer/sbft/backend/backendab.go b/orderer/sbft/backend/backendab.go
@@ -19,6 +19,9 @@ package backend
 import (
 	"github.com/hyperledger/fabric/common/policies"
 	"github.com/hyperledger/fabric/orderer/common/deliver"
+	"github.com/hyperledger/fabric/orderer/common/sharedconfig"
+	mockpolicies "github.com/hyperledger/fabric/orderer/mocks/policies"
+	mocksharedconfig "github.com/hyperledger/fabric/orderer/mocks/sharedconfig"
 	"github.com/hyperledger/fabric/orderer/rawledger"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
@@ -39,11 +42,17 @@ func (xxx *xxxSupportManager) GetChain(id string) (deliver.Support, bool) {
 }
 
 type xxxSupport struct {
-	reader rawledger.Reader
+	reader        rawledger.Reader
+	sharedConfig  sharedconfig.Manager
+	policyManager policies.Manager
 }
 
 func (xxx *xxxSupport) PolicyManager() policies.Manager {
-	panic("Unimplemented")
+	return xxx.policyManager
+}
+
+func (xxx *xxxSupport) SharedConfig() sharedconfig.Manager {
+	return xxx.sharedConfig
 }
 
 func (xxx *xxxSupport) Reader() rawledger.Reader {
@@ -77,7 +86,11 @@ func NewBackendAB(backend *Backend) *BackendAB {
 
 	manager := &xxxSupportManager{
 		chainID: payload.Header.ChainHeader.ChainID,
-		support: &xxxSupport{reader: backend.ledger},
+		support: &xxxSupport{
+			reader:        backend.ledger,
+			sharedConfig:  &mocksharedconfig.Manager{},
+			policyManager: &mockpolicies.Manager{Policy: &mockpolicies.Policy{}},
+		},
 	}
 	// XXX End hackiness