Merge branch 'master' into expiry

escrow/control_service.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/gogo/protobuf/sortkeys"
 	"github.com/singnet/snet-daemon/authutils"
 	"github.com/singnet/snet-daemon/blockchain"
 	log "github.com/sirupsen/logrus"
@@ -36,6 +37,10 @@ func (service *BlockChainDisabledProviderControlService) StartClaim(ctx context.
 	return &PaymentReply{}, nil
 }
 
+func (service *BlockChainDisabledProviderControlService) StartClaimForMultipleChannels(ctx context.Context, request *StartMultipleClaimRequest) (reply *PaymentsListReply, err error) {
+	return &PaymentsListReply{}, nil
+}
+
 func NewProviderControlService(channelService PaymentChannelService, serMetaData *blockchain.ServiceMetadata,
 	orgMetadata *blockchain.OrganizationMetaData) *ProviderControlService {
 	return &ProviderControlService{
@@ -69,6 +74,89 @@ func (service *ProviderControlService) GetListUnclaimed(ctx context.Context, req
 	return service.listChannels()
 }
 
+/*
+We need to have an ability to StartClaims for multiple channels too!
+The User makes a Daemon call to the get all the pending claims
+The user then needs to call the start claim for every channel that he intends to claim ( thus a new signature again ) for every invocation of StartClaim !
+PS You can make Multiple Claims in BlockChain in a single call multiChannelClaim
+You can now initiate multiple claims using this function StartClaimsOnMultipleChannels that takes a list of ChannelIds , What Daemon will do is invoke the StartClaim internally for every channel Id passed
+If an error is encountered , then we return back whatever was successful and stop with the channel ID that had an issue
+and return the error that was encountered.
+*/
+func (service *ProviderControlService) StartClaimForMultipleChannels(ctx context.Context, request *StartMultipleClaimRequest) (reply *PaymentsListReply, err error) {
+
+	if err := service.checkMpeAddress(request.GetMpeAddress()); err != nil {
+		return nil, err
+	}
+
+	if err := authutils.CompareWithLatestBlockNumber(big.NewInt(int64(request.CurrentBlock))); err != nil {
+		return nil, err
+	}
+
+	if err := service.verifySignerForStartClaimForMultipleChannels(request); err != nil {
+		return nil, err
+	}
+	err = service.removeClaimedPayments()
+	if err != nil {
+		log.Errorf("unable to remove payments from which are already claimed")
+		return nil, err
+	}
+	return service.startClaims(request)
+
+}
+
+func (service *ProviderControlService) startClaims(request *StartMultipleClaimRequest) (reply *PaymentsListReply, err error) {
+	reply = &PaymentsListReply{}
+	payments := make([]*PaymentReply, 0)
+
+	for _, channelId := range request.GetChannelIds() {
+		payment, err := service.beginClaimOnChannel(big.NewInt(int64(channelId)))
+		if err != nil {
+			// we stop here and return back what ever was successful as reply
+			return reply, err
+		}
+		payments = append(payments, payment)
+	}
+	reply.Payments = payments
+	return reply, nil
+}
+
+//("__StartClaimForMultipleChannels_, mpe_address,channel_id1,channel_id2,...,current_block_number)
+func (service *ProviderControlService) verifySignerForStartClaimForMultipleChannels(request *StartMultipleClaimRequest) error {
+	message := bytes.Join([][]byte{
+		[]byte("__StartClaimForMultipleChannels_"),
+		service.serviceMetaData.GetMpeAddress().Bytes(),
+		getBytesOfChannelIds(request),
+		abi.U256(big.NewInt(int64(request.CurrentBlock))),
+	}, nil)
+	return service.verifySigner(message, request.GetSignature())
+}
+
+func getBytesOfChannelIds(request *StartMultipleClaimRequest) []byte {
+	channelIds := make([]uint64, 0)
+	for _, channelId := range request.GetChannelIds() {
+		channelIds = append(channelIds, channelId)
+	}
+	//sort the channel Ids
+	sortkeys.Uint64s(channelIds)
+	channelIdInBytes := make([]byte, 0)
+
+	for index, channelId := range channelIds {
+		if index == 0 {
+			channelIdInBytes = bytes.Join([][]byte{
+				bigIntToBytes(big.NewInt(int64(channelId))),
+			}, nil)
+		} else {
+			channelIdInBytes = bytes.Join([][]byte{
+				channelIdInBytes,
+				bigIntToBytes(big.NewInt(int64(channelId))),
+			}, nil)
+		}
+
+	}
+	return channelIdInBytes
+}
+
 //Get the list of all claims that have been initiated but not completed yet.
 //Verify that mpe_address is correct
 //Verify that actual block_number is not very different (+-5 blocks) from the current_block_number from the signature
@@ -303,7 +391,6 @@ func (service *ProviderControlService) removeClaimedPayments() error {
 //Check if the mpe address passed matches to what is present in the metadata.
 func (service *ProviderControlService) checkMpeAddress(mpeAddress string) error {
 	passedAddress := common.HexToAddress(mpeAddress)
-
 	if !(service.mpeAddress == passedAddress) {
 		return fmt.Errorf("the mpeAddress: %s passed does not match to what has been registered", mpeAddress)
 	}

escrow/control_service.proto

@@ -15,8 +15,12 @@ service ProviderControlService {
     //get list of all payments in progress
     rpc GetListInProgress(GetPaymentsListRequest) returns (PaymentsListReply) {}
 
-    //initilize claim for specific channel
+    //initialize claim for specific channel
     rpc StartClaim(StartClaimRequest) returns (PaymentReply) {}
+
+    //initiate multiple claims at a time
+    rpc StartClaimForMultipleChannels(StartMultipleClaimRequest) returns (PaymentsListReply) {}
+
 }
 
 
@@ -58,3 +62,12 @@ message PaymentsListReply {
     repeated PaymentReply payments = 1;
 }
 
+message StartMultipleClaimRequest {
+    //address of MultiPartyEscrow contract
+    string mpe_address = 1;
+    repeated uint64 channel_ids    = 2;
+    uint64 current_block = 3;
+    //signature will be as follows ( sequence is an ascending order of channel_ids)
+    //("__StartClaimForMultipleChannels_, mpe_address,channel_id1,channel_id2,...,current_block_number)
+    bytes signature = 4;
+}

escrow/control_service_test.go

@@ -1,31 +1,192 @@
 package escrow
 
 import (
+	"bytes"
+	"crypto/ecdsa"
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/singnet/snet-daemon/authutils"
+	"github.com/singnet/snet-daemon/config"
+	"github.com/stretchr/testify/suite"
 	"math/big"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 
 	"github.com/singnet/snet-daemon/blockchain"
 )
 
-//todo, Work in progress
-func TestProviderControlService_GetListInProgress(t *testing.T) {
+type ControlServiceTestSuite struct {
+	suite.Suite
+	service         *ProviderControlService
+	senderAddress   common.Address
+	receiverAddress common.Address
+	receiverPvtKy   *ecdsa.PrivateKey
+	paymentStorage  *PaymentStorage
+	mpeAddress      common.Address
+	channelService  PaymentChannelService
+	serviceMetaData *blockchain.ServiceMetadata
+	orgMetaData     *blockchain.OrganizationMetaData
+	storage         *PaymentChannelStorage
+}
+
+func (suite *ControlServiceTestSuite) payment() *Payment {
+	payment := &Payment{
+		Amount:             big.NewInt(12300),
+		ChannelID:          big.NewInt(1),
+		ChannelNonce:       big.NewInt(0),
+		MpeContractAddress: suite.serviceMetaData.GetMpeAddress(),
+	}
+	SignTestPayment(payment, suite.receiverPvtKy)
+	return payment
+}
 
+func (suite *ControlServiceTestSuite) putChannel(channelId *big.Int) {
+	suite.storage.Put(&PaymentChannelKey{ID: channelId}, &PaymentChannelData{
+		ChannelID:        channelId,
+		Nonce:            big.NewInt(0),
+		Sender:           suite.senderAddress,
+		Recipient:        suite.receiverAddress,
+		GroupID:          suite.orgMetaData.GetGroupId(),
+		FullAmount:       big.NewInt(12345),
+		Expiration:       big.NewInt(100),
+		Signer:           suite.receiverAddress,
+		AuthorizedAmount: big.NewInt(10),
+		Signature:        blockchain.HexToBytes("0xa4d2ae6f3edd1f7fe77e4f6f78ba18d62e6093bcae01ef86d5de902d33662fa372011287ea2d8d8436d9db8a366f43480678df25453b484c67f80941ef2c05ef01"),
+	})
 }
 
-func TestProviderControlService_checkMpeAddress(t *testing.T) {
+func (suite *ControlServiceTestSuite) mpeChannel() *blockchain.MultiPartyEscrowChannel {
+	return &blockchain.MultiPartyEscrowChannel{
+		Sender:     suite.senderAddress,
+		Recipient:  suite.receiverAddress,
+		GroupId:    suite.orgMetaData.GetGroupId(),
+		Value:      big.NewInt(12345),
+		Nonce:      big.NewInt(0),
+		Expiration: big.NewInt(1000),
+		Signer:     suite.receiverAddress,
+	}
+}
+
+func (suite *ControlServiceTestSuite) SetupSuite() {
+	//
+	var errs error
+	suite.receiverPvtKy = GenerateTestPrivateKey()
+	println(errs)
+	suite.receiverAddress = crypto.PubkeyToAddress(suite.receiverPvtKy.PublicKey)
+	orgJson := strings.Replace(testJsonOrgGroupData, "0x671276c61943A35D5F230d076bDFd91B0c47bF09", suite.receiverAddress.Hex(), -1)
+	suite.orgMetaData, _ = blockchain.InitOrganizationMetaDataFromJson(orgJson)
+	suite.serviceMetaData, _ = blockchain.InitServiceMetaDataFromJson(testJsonData)
+	println("suite.orgMetaData.GetPaymentAddress().Hex() " + suite.orgMetaData.GetPaymentAddress().Hex())
+	println("suite.receiverAddress.Hex()" + suite.receiverAddress.Hex())
+
+	memoryStorage := NewMemStorage()
+	suite.storage = NewPaymentChannelStorage(memoryStorage)
+	suite.paymentStorage = NewPaymentStorage(memoryStorage)
+	suite.channelService = NewPaymentChannelService(
+		suite.storage,
+		suite.paymentStorage,
+		&BlockchainChannelReader{
+			readChannelFromBlockchain: func(channelID *big.Int) (*blockchain.MultiPartyEscrowChannel, bool, error) {
+				return suite.mpeChannel(), true, nil
+			},
+			recipientPaymentAddress: func() common.Address {
+				return suite.receiverAddress
+			},
+		},
+		NewEtcdLocker(memoryStorage),
+		&ChannelPaymentValidator{
+			currentBlock:               func() (*big.Int, error) { return big.NewInt(99), nil },
+			paymentExpirationThreshold: func() *big.Int { return big.NewInt(0) },
+		}, func() ([32]byte, error) {
+			return [32]byte{123}, nil
+		})
+
+	suite.service = NewProviderControlService(suite.channelService, suite.serviceMetaData, suite.orgMetaData)
+
+}
+
+func TestControlServiceTestSuite(t *testing.T) {
+	suite.Run(t, new(ControlServiceTestSuite))
+}
+
+func (suite *ControlServiceTestSuite) SignStartClaimForMultipleChannels(request *StartMultipleClaimRequest) {
+	message := bytes.Join([][]byte{
+		[]byte("__StartClaimForMultipleChannels_"),
+		suite.serviceMetaData.GetMpeAddress().Bytes(),
+		bigIntToBytes(big.NewInt(1)),
+		bigIntToBytes(big.NewInt(2)),
+		abi.U256(big.NewInt(int64(request.CurrentBlock))),
+	}, nil)
+	request.Signature = getSignature(message, suite.receiverPvtKy)
+}
+
+func (suite *ControlServiceTestSuite) SignListInProgress(request *GetPaymentsListRequest) {
+	message := bytes.Join([][]byte{
+		[]byte("__list_in_progress"),
+		suite.serviceMetaData.GetMpeAddress().Bytes(),
+		abi.U256(big.NewInt(int64(request.CurrentBlock))),
+	}, nil)
+	request.Signature = getSignature(message, suite.receiverPvtKy)
+}
+
+//Build Request
+//Validate Signature
+//Add 2 elements in ChannelData
+//After claims, check if 2 payments have come in
+func (suite *ControlServiceTestSuite) TestStartClaimForMultipleChannels() {
+	suite.putChannel(big.NewInt(1))
+	suite.putChannel(big.NewInt(2))
+	reply, err := suite.service.listChannels()
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), len(reply.Payments), 2)
+	ids := make([]uint64, 0)
+	//
+	ids = append(ids, 2)
+	ids = append(ids, 1)
+	config.Vip().Set(config.BlockChainNetworkSelected, "ropsten")
+	config.Validate()
+	blknum, _ := authutils.CurrentBlock()
+	startMultipleClaimRequest := &StartMultipleClaimRequest{
+		MpeAddress: suite.serviceMetaData.GetMpeAddress().Hex(), ChannelIds: ids, CurrentBlock: blknum.Uint64(),
+		Signature: nil}
+	suite.SignStartClaimForMultipleChannels(startMultipleClaimRequest)
+	replyMultipleClaims, err := suite.service.StartClaimForMultipleChannels(nil, startMultipleClaimRequest)
+	assert.Nil(suite.T(), err)
+	assert.True(suite.T(), bytesToBigInt(replyMultipleClaims.Payments[0].ChannelId).Int64() > 0)
+	assert.True(suite.T(), bytesToBigInt(replyMultipleClaims.Payments[1].ChannelId).Int64() > 0)
+	paymentsListRequest := &GetPaymentsListRequest{MpeAddress: suite.serviceMetaData.MpeAddress, CurrentBlock: blknum.Uint64()}
+	suite.SignListInProgress(paymentsListRequest)
+	replyListInProgress, err := suite.service.GetListInProgress(nil, paymentsListRequest)
+	assert.Nil(suite.T(), err)
+	assert.NotNil(suite.T(), replyListInProgress.Payments[0].Signature)
+	assert.NotNil(suite.T(), replyListInProgress.Payments[0].Signature)
+
+}
+
+func (suite *ControlServiceTestSuite) TestProviderControlService_checkMpeAddress() {
 	servicemetadata := blockchain.ServiceMetadata{}
 	servicemetadata.MpeAddress = "0xE8D09a6C296aCdd4c01b21f407ac93fdfC63E78C"
 	control_service := NewProviderControlService(nil, &servicemetadata, nil)
 	err := control_service.checkMpeAddress("0xe8D09a6C296aCdd4c01b21f407ac93fdfC63E78C")
-	assert.Nil(t, err)
+	assert.Nil(suite.T(), err)
 	err = control_service.checkMpeAddress("0xe9D09a6C296aCdd4c01b21f407ac93fdfC63E78C")
-	assert.Equal(t, err.Error(), "the mpeAddress: 0xe9D09a6C296aCdd4c01b21f407ac93fdfC63E78C passed does not match to what has been registered")
+	assert.Equal(suite.T(), err.Error(), "the mpeAddress: 0xe9D09a6C296aCdd4c01b21f407ac93fdfC63E78C passed does not match to what has been registered")
 }
 
-func TestBeginClaimOnChannel(t *testing.T) {
+func (suite *ControlServiceTestSuite) TestBeginClaimOnChannel() {
 	control_service := NewProviderControlService(&paymentChannelServiceMock{}, &blockchain.ServiceMetadata{MpeAddress: "0xe9D09a6C296aCdd4c01b21f407ac93fdfC63E78C"}, nil)
 	_, err := control_service.beginClaimOnChannel(big.NewInt(12345))
-	assert.Equal(t, err.Error(), "channel Id 12345 was not found on blockchain or storage")
+	assert.Equal(suite.T(), err.Error(), "channel Id 12345 was not found on blockchain or storage")
+}
+
+func (suite *ControlServiceTestSuite) TestVerifyInvalidSignature() {
+	blknum, _ := authutils.CurrentBlock()
+	unclaimedRequests := &GetPaymentsListRequest{MpeAddress: suite.serviceMetaData.MpeAddress, CurrentBlock: blknum.Uint64()}
+	suite.SignListInProgress(unclaimedRequests)
+	reply, err := suite.service.GetListUnclaimed(nil, unclaimedRequests)
+	assert.Nil(suite.T(), reply)
+	assert.Contains(suite.T(), err.Error(), "does not match to what has been expected / registered")
 }