[FAB-7639] Block expired x509 identities in events

This change set blocks event registration in expired x509 identities,
and also makes the events producer not send events to x509 clients
which their identity has expired during the session.

Change-Id: I1b32c21ced5a7b531a427fb14ee331092a8eae3f
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

events/producer/events.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/hyperledger/fabric/common/util"
 	pb "github.com/hyperledger/fabric/protos/peer"
 )
 
@@ -217,7 +218,6 @@ func (ep *eventProcessor) start() {
 	for {
 		//wait for event
 		e := <-ep.eventChannel
-
 		var hl handlerList
 		eType := getMessageType(e)
 		ep.Lock()
@@ -229,7 +229,15 @@ func (ep *eventProcessor) start() {
 		//lock the handler map lock
 		ep.Unlock()
 
+		now := time.Now()
 		hl.foreach(e, func(h *handler) {
+			if hasSessionExpired(now, h.sessionEndTime) {
+				addr := util.ExtractRemoteAddress(h.ChatStream.Context())
+				logger.Warning("Client's", addr, " identity has expired")
+				// We have to call Stop asynchronously because hl.foreach() holds a lock on hl
+				go h.Stop()
+				return
+			}
 			if e.Event != nil {
 				h.SendMessage(e)
 			}
@@ -238,6 +246,10 @@ func (ep *eventProcessor) start() {
 	}
 }
 
+func hasSessionExpired(now, sessionEndTime time.Time) bool {
+	return !sessionEndTime.IsZero() && now.After(sessionEndTime)
+}
+
 //initialize and start
 func initializeEvents(config *EventsServerConfig) {
 	if gEventProcessor != nil {

events/producer/handler.go

@@ -24,13 +24,15 @@ import (
 
 	"github.com/golang/protobuf/proto"
 
+	"github.com/hyperledger/fabric/common/crypto"
 	"github.com/hyperledger/fabric/msp/mgmt"
 	pb "github.com/hyperledger/fabric/protos/peer"
 )
 
 type handler struct {
 	ChatStream       pb.Events_ChatServer
 	interestedEvents map[string]*pb.Interest
+	sessionEndTime   time.Time
 }
 
 func newEventHandler(stream pb.Events_ChatServer) *handler {
@@ -162,6 +164,12 @@ func (d *handler) validateEventMessage(signedEvt *pb.SignedEvent) (*pb.Event, er
 		return nil, fmt.Errorf("error unmarshaling the event bytes in the SignedEvent: %s", err)
 	}
 
+	expirationTime := crypto.ExpiresAt(evt.Creator)
+	if !expirationTime.IsZero() && time.Now().After(expirationTime) {
+		return nil, fmt.Errorf("identity expired")
+	}
+	d.sessionEndTime = expirationTime
+
 	if evt.GetTimestamp() != nil {
 		evtTime := time.Unix(evt.GetTimestamp().Seconds, int64(evt.GetTimestamp().Nanos)).UTC()
 		peerTime := time.Now()

events/producer/producer_test.go

@@ -35,6 +35,9 @@ import (
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/grpclog"
 
+	"io/ioutil"
+	"path/filepath"
+
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/hyperledger/fabric/common/ledger/testutil"
@@ -47,6 +50,8 @@ import (
 	"github.com/hyperledger/fabric/msp"
 	"github.com/hyperledger/fabric/msp/mgmt"
 	"github.com/hyperledger/fabric/msp/mgmt/testtools"
+	"github.com/hyperledger/fabric/protos/common"
+	msp2 "github.com/hyperledger/fabric/protos/msp"
 	pb "github.com/hyperledger/fabric/protos/peer"
 	"github.com/hyperledger/fabric/protos/utils"
 	"github.com/spf13/viper"
@@ -110,7 +115,7 @@ func (a *Adapter) Disconnected(err error) {
 	}
 }
 
-func createRegisterEvent(timestamp *timestamp.Timestamp, cert *x509.Certificate) (*pb.Event, error) {
+func createRegisterEvent(timestamp *timestamp.Timestamp, tlsCert *x509.Certificate) (*pb.Event, error) {
 	events := make([]*pb.Interest, 2)
 	events[0] = &pb.Interest{
 		EventType: pb.EventType_BLOCK,
@@ -129,8 +134,8 @@ func createRegisterEvent(timestamp *timestamp.Timestamp, cert *x509.Certificate)
 		Creator:   signerSerialized,
 		Timestamp: timestamp,
 	}
-	if cert != nil {
-		evt.TlsCertHash = util.ComputeSHA256(cert.Raw)
+	if tlsCert != nil {
+		evt.TlsCertHash = util.ComputeSHA256(tlsCert.Raw)
 	}
 	return evt, nil
 }
@@ -157,11 +162,24 @@ func corrupt(bytes []byte) {
 	bytes[r.Int31n(int32(len(bytes)))]--
 }
 
+func createExpiredIdentity(t *testing.T) []byte {
+	certBytes, err := ioutil.ReadFile(filepath.Join("testdata", "expiredCert.pem"))
+	assert.NoError(t, err)
+	sId := &msp2.SerializedIdentity{
+		IdBytes: certBytes,
+	}
+	serializedIdentity, err := proto.Marshal(sId)
+	assert.NoError(t, err)
+	return serializedIdentity
+}
+
 func TestSignedEvent(t *testing.T) {
 	recvChan := make(chan *streamEvent)
 	sendChan := make(chan *pb.Event)
 	stream := &mockEventStream{recvChan: recvChan, sendChan: sendChan}
 	mockHandler := &handler{ChatStream: stream}
+	backupSerializedIdentity := signerSerialized
+	signerSerialized = createExpiredIdentity(t)
 	// get a test event
 	evt, err := createRegisterEvent(nil, nil)
 	if err != nil {
@@ -176,6 +194,29 @@ func TestSignedEvent(t *testing.T) {
 		return
 	}
 
+	// validate it. Expected to fail because the identity expired
+	_, err = mockHandler.validateEventMessage(sEvt)
+	assert.Equal(t, err.Error(), "identity expired")
+	if err == nil {
+		t.Fatalf("validateEventMessage succeeded but should have failed")
+		return
+	}
+
+	// Restore the original legit serialized identity
+	signerSerialized = backupSerializedIdentity
+	evt, err = createRegisterEvent(nil, nil)
+	if err != nil {
+		t.Fatalf("createEvent failed, err %s", err)
+		return
+	}
+
+	// sign it
+	sEvt, err = utils.GetSignedEvent(evt, signer)
+	if err != nil {
+		t.Fatalf("GetSignedEvent failed, err %s", err)
+		return
+	}
+
 	// validate it. Expected to succeed
 	_, err = mockHandler.validateEventMessage(sEvt)
 	if err != nil {
@@ -440,6 +481,67 @@ func TestRegister_MutualTLS(t *testing.T) {
 	}
 }
 
+func TestRegister_ExpiredIdentity(t *testing.T) {
+	m := newMockEventhub()
+	defer close(m.recvChan)
+
+	go ehServer.Chat(m)
+
+	publishBlock := func() {
+		gEventProcessor.eventChannel <- &pb.Event{
+			Event: &pb.Event_Block{
+				Block: &common.Block{
+					Header: &common.BlockHeader{
+						Number: 100,
+					},
+				},
+			},
+		}
+	}
+
+	expireSessions := func() {
+		gEventProcessor.RLock()
+		handlerList := gEventProcessor.eventConsumers[pb.EventType_BLOCK].(*genericHandlerList)
+		handlerList.RLock()
+		for k := range handlerList.handlers {
+			// Artificially move the session end time a minute into the past
+			k.sessionEndTime = time.Now().Add(-1 * time.Minute)
+		}
+		handlerList.RUnlock()
+		gEventProcessor.RUnlock()
+	}
+
+	sEvt, err := createSignedRegisterEvent(util.CreateUtcTimestamp(), nil)
+	assert.NoError(t, err)
+	m.recvChan <- &streamEvent{event: sEvt}
+
+	// Wait for register Ack
+	select {
+	case <-m.sendChan:
+	case <-time.After(time.Millisecond * 500):
+		assert.Fail(t, "Didn't receive back a register ack on time")
+	}
+
+	// Publish a block and make sure we receive it
+	publishBlock()
+	select {
+	case resp := <-m.sendChan:
+		assert.Equal(t, uint64(100), resp.GetBlock().Header.Number)
+	case <-time.After(time.Millisecond * 500):
+		assert.Fail(t, "Didn't receive the block on time, but should have")
+	}
+
+	// Expire the sessions, and publish a block again
+	expireSessions()
+	publishBlock()
+	// Make sure we don't receive it
+	select {
+	case resp := <-m.sendChan:
+		t.Fatalf("Received a block (%v) but wasn't supposed to", resp.GetBlock())
+	case <-time.After(time.Millisecond * 500):
+	}
+}
+
 func resetEventProcessor(useMutualTLS bool) {
 	extract := func(msg proto.Message) []byte {
 		evt, isEvent := msg.(*pb.Event)

events/producer/testdata/expiredCert.pem

@@ -0,0 +1,14 @@
+-----BEGIN CERTIFICATE-----
+MIICGTCCAcCgAwIBAgIRAMjlb3tiiXUty7eFbtauwM0wCgYIKoZIzj0EAwIwczEL
+MAkGA1UEBhMCVVMxEzARBgNVBAgTCkNhbGlmb3JuaWExFjAUBgNVBAcTDVNhbiBG
+cmFuY2lzY28xGTAXBgNVBAoTEG9yZzEuZXhhbXBsZS5jb20xHDAaBgNVBAMTE2Nh
+Lm9yZzEuZXhhbXBsZS5jb20wHhcNMTgwMTA3MTIxMDM2WhcNMTgwMTA3MTIwNTM2
+WjBbMQswCQYDVQQGEwJVUzETMBEGA1UECBMKQ2FsaWZvcm5pYTEWMBQGA1UEBxMN
+U2FuIEZyYW5jaXNjbzEfMB0GA1UEAwwWVXNlcjFAb3JnMS5leGFtcGxlLmNvbTBZ
+MBMGByqGSM49AgEGCCqGSM49AwEHA0IABGzOvcOtZ+WvatGfkU/iQPfbU9AlvD+O
+bKuyUD35v71WvQf1sasUPKp0gBt4SaGLe0bOnvFTGKvTr0TgVu6T85ejTTBLMA4G
+A1UdDwEB/wQEAwIHgDAMBgNVHRMBAf8EAjAAMCsGA1UdIwQkMCKAIAvawoCOaQSY
+mSfQpUm8eezH91vKkfQM2Ui0RVffql3bMAoGCCqGSM49BAMCA0cAMEQCIEbzjljz
+s0QcGPZZGhPkbaSK1MurBzfmE4izQ6roccLLAiBGZo5i59k+gNTsVCsS4s2HBF6U
+rj/t6mYKdOVxhPXFbw==
+-----END CERTIFICATE-----