Merge pull request hyperledger#2 from Param-S/attestationservice

New Block Attestation service

Author: yacovm

common/deliver/deliver.go

@@ -398,3 +398,21 @@ func (h *Handler) validateChannelHeader(ctx context.Context, chdr *cb.ChannelHea
 func noExpiration(_ []byte) time.Time {
 	return time.Time{}
 }
+
+func (h *Handler) HandleAttestation(ctx context.Context, srv *Server, env *cb.Envelope) error {
+	status, err := h.deliverBlocks(ctx, srv, env)
+	if err != nil {
+		return err
+	}
+
+	err = srv.SendStatusResponse(status)
+	if status != cb.Status_SUCCESS {
+		return err
+	}
+	if err != nil {
+		addr := util.ExtractRemoteAddress(ctx)
+		logger.Warningf("Error sending to %s: %s", addr, err)
+		return err
+	}
+	return nil
+}

common/deliver/deliver_test.go

@@ -651,6 +651,16 @@ var _ = Describe("Deliver", func() {
 			Expect(resp).To(Equal(cb.Status_SUCCESS))
 		})
 
+		It("HandleAttestation sends requested block", func() {
+			err := handler.HandleAttestation(context.Background(), server, envelope)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeResponseSender.SendBlockResponseCallCount()).To(Equal(1))
+			b, _, _, _ := fakeResponseSender.SendBlockResponseArgsForCall(0)
+			Expect(b).To(Equal(&cb.Block{
+				Header: &cb.BlockHeader{Number: 100},
+			}))
+		})
+
 		Context("when sending the success status fails", func() {
 			BeforeEach(func() {
 				fakeResponseSender.SendStatusResponseReturns(errors.New("send-success-fails"))
@@ -660,6 +670,11 @@ var _ = Describe("Deliver", func() {
 				err := handler.Handle(context.Background(), server)
 				Expect(err).To(MatchError("send-success-fails"))
 			})
+
+			It("HandleAttestation returns the error", func() {
+				err := handler.HandleAttestation(context.Background(), server, envelope)
+				Expect(err).To(MatchError("send-success-fails"))
+			})
 		})
 
 		Context("when receive fails", func() {
@@ -703,6 +718,15 @@ var _ = Describe("Deliver", func() {
 				resp := fakeResponseSender.SendStatusResponseArgsForCall(0)
 				Expect(resp).To(Equal(cb.Status_BAD_REQUEST))
 			})
+
+			It("sends a bad envelope to HandleAttestation", func() {
+				err := handler.HandleAttestation(context.Background(), server, envelope)
+				Expect(err).NotTo(HaveOccurred())
+
+				Expect(fakeResponseSender.SendStatusResponseCallCount()).To(Equal(1))
+				resp := fakeResponseSender.SendStatusResponseArgsForCall(0)
+				Expect(resp).To(Equal(cb.Status_BAD_REQUEST))
+			})
 		})
 
 		Context("when unmarshalling the channel header fails", func() {

orderer/common/server/attestationserver.go

@@ -0,0 +1,102 @@
+/*
+Copyright IBM Corp. 2017 All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package server
+
+import (
+	"runtime/debug"
+	"time"
+
+	cb "github.com/hyperledger/fabric-protos-go/common"
+	ab "github.com/hyperledger/fabric-protos-go/orderer"
+	"github.com/hyperledger/fabric/common/deliver"
+	"github.com/hyperledger/fabric/common/metrics"
+	"github.com/hyperledger/fabric/common/policies"
+	localconfig "github.com/hyperledger/fabric/orderer/common/localconfig"
+	"github.com/hyperledger/fabric/orderer/common/msgprocessor"
+	"github.com/hyperledger/fabric/orderer/common/multichannel"
+	"github.com/hyperledger/fabric/protoutil"
+	"github.com/pkg/errors"
+)
+
+type attestationserver struct {
+	dh    *deliver.Handler
+	debug *localconfig.Debug
+	*multichannel.Registrar
+}
+
+// NewServer creates an ab.AtomicBroadcastServer based on the broadcast target and ledger Reader
+func NewAttestationService(
+	r *multichannel.Registrar,
+	metricsProvider metrics.Provider,
+	debug *localconfig.Debug,
+	timeWindow time.Duration,
+	mutualTLS bool,
+	expirationCheckDisabled bool,
+) ab.BlockAttestationsServer {
+	s := &attestationserver{
+		dh:        deliver.NewHandler(deliverSupport{Registrar: r}, timeWindow, mutualTLS, deliver.NewMetrics(metricsProvider), expirationCheckDisabled),
+		debug:     debug,
+		Registrar: r,
+	}
+	return s
+}
+
+func (s *attestationserver) BlockAttestations(env *cb.Envelope, strm ab.BlockAttestations_BlockAttestationsServer) error {
+	logger.Debugf("Starting new handler for block attestation")
+	defer func() {
+		if r := recover(); r != nil {
+			logger.Criticalf("block attestation client triggered panic: %s\n%s", r, debug.Stack())
+		}
+		logger.Debugf("Closing attestation server stream")
+	}()
+
+	policyChecker := func(env *cb.Envelope, channelID string) error {
+		chain := s.GetChain(channelID)
+		if chain == nil {
+			return errors.Errorf("channel %s not found", channelID)
+		}
+		// In maintenance mode, we typically require the signature of /Channel/Orderer/Readers.
+		// This will block Deliver requests from peers (which normally satisfy /Channel/Readers).
+		sf := msgprocessor.NewSigFilter(policies.ChannelReaders, policies.ChannelOrdererReaders, chain)
+		return sf.Apply(env)
+	}
+	attestationServer := &deliver.Server{
+		PolicyChecker: deliver.PolicyCheckerFunc(policyChecker),
+		ResponseSender: &attestationSender{
+			BlockAttestations_BlockAttestationsServer: strm,
+		},
+	}
+	return s.dh.HandleAttestation(strm.Context(), attestationServer, env)
+}
+
+type attestationSender struct {
+	ab.BlockAttestations_BlockAttestationsServer
+}
+
+func (rs *attestationSender) SendStatusResponse(status cb.Status) error {
+	reply := &ab.BlockAttestationResponse{
+		Type: &ab.BlockAttestationResponse_Status{Status: status},
+	}
+	return rs.Send(reply)
+}
+
+func (rs *attestationSender) SendBlockResponse(
+	block *cb.Block,
+	channelID string,
+	chain deliver.Chain,
+	signedData *protoutil.SignedData,
+) error {
+	blockAttestation := ab.BlockAttestation{Header: block.Header, Metadata: block.Metadata}
+	response := &ab.BlockAttestationResponse{
+		Type: &ab.BlockAttestationResponse_BlockAttestation{BlockAttestation: &blockAttestation},
+	}
+	return rs.Send(response)
+}
+
+func (rs *attestationSender) DataType() string {
+	return "block_attestation"
+}