Update to latest go-fil-markets w/ enhanced Graphsync (#3978)

* feat(fetcher): update graphsync w/ extension security

Update graphsync, data-transfer, and go-fil-markets. Turn on default request rejection for
graphsync. Add extension for chainsync to identify chainsync requests.

* fix(deps): use master commit for deps update

* chore(lint): fix lint errors

go.mod

@@ -18,9 +18,9 @@ require (
 	github.com/filecoin-project/go-address v0.0.2-0.20200218010043-eb9bb40ed5be
 	github.com/filecoin-project/go-amt-ipld/v2 v2.0.1-0.20200131012142-05d80eeccc5e
 	github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03
-	github.com/filecoin-project/go-data-transfer v0.0.0-20191219005021-4accf56bd2ce
+	github.com/filecoin-project/go-data-transfer v0.0.0-20200408061858-82c58b423ca6
 	github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5
-	github.com/filecoin-project/go-fil-markets v0.0.0-20200326202058-16ce89e7bef4
+	github.com/filecoin-project/go-fil-markets v0.0.0-20200408062434-d92f329a6428
 	github.com/filecoin-project/go-leb128 v0.0.0-20190212224330-8d79a5489543
 	github.com/filecoin-project/go-paramfetch v0.0.2-0.20200218225740-47c639bab663
 	github.com/filecoin-project/go-statestore v0.1.0
@@ -34,17 +34,17 @@ require (
 	github.com/google/go-github v17.0.0+incompatible
 	github.com/google/go-querystring v1.0.0 // indirect
 	github.com/google/uuid v1.1.1
-	github.com/ipfs/go-bitswap v0.1.8
+	github.com/ipfs/go-bitswap v0.2.8
 	github.com/ipfs/go-block-format v0.0.2
-	github.com/ipfs/go-blockservice v0.1.3-0.20190908200855-f22eea50656c
+	github.com/ipfs/go-blockservice v0.1.3
 	github.com/ipfs/go-car v0.0.3-0.20200221191037-3762780fa84e
 	github.com/ipfs/go-cid v0.0.5
 	github.com/ipfs/go-datastore v0.4.4
 	github.com/ipfs/go-ds-badger2 v0.0.0-20200211201106-609c9d2a39c7
 	github.com/ipfs/go-fs-lock v0.0.1
-	github.com/ipfs/go-graphsync v0.0.4
+	github.com/ipfs/go-graphsync v0.0.6-0.20200408061628-e1a98fc64c42
 	github.com/ipfs/go-hamt-ipld v0.0.15-0.20200219230615-d4e402c05ab0
-	github.com/ipfs/go-ipfs-blockstore v0.1.3
+	github.com/ipfs/go-ipfs-blockstore v0.1.4
 	github.com/ipfs/go-ipfs-chunker v0.0.1
 	github.com/ipfs/go-ipfs-cmdkit v0.0.1
 	github.com/ipfs/go-ipfs-cmds v0.0.1
@@ -62,29 +62,26 @@ require (
 	github.com/ipfs/iptb v1.3.8-0.20190401234037-98ccf4228a73
 	github.com/ipld/go-ipld-prime v0.0.2-0.20191108012745-28a82f04c785
 	github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52
-	github.com/jbenet/goprocess v0.1.3
+	github.com/jbenet/goprocess v0.1.4
 	github.com/jstemmer/go-junit-report v0.9.1
 	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
-	github.com/libp2p/go-libp2p v0.5.2
+	github.com/libp2p/go-libp2p v0.6.1
 	github.com/libp2p/go-libp2p-autonat-svc v0.1.0
 	github.com/libp2p/go-libp2p-circuit v0.1.4
-	github.com/libp2p/go-libp2p-core v0.5.0
+	github.com/libp2p/go-libp2p-core v0.5.1
 	github.com/libp2p/go-libp2p-kad-dht v0.1.1
-	github.com/libp2p/go-libp2p-peerstore v0.1.4
+	github.com/libp2p/go-libp2p-peerstore v0.2.0
 	github.com/libp2p/go-libp2p-pubsub v0.2.6
 	github.com/libp2p/go-libp2p-swarm v0.2.2
-	github.com/libp2p/go-yamux v1.2.4 // indirect
 	github.com/mattn/go-isatty v0.0.10 // indirect
 	github.com/mattn/go-runewidth v0.0.8 // indirect
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/morikuni/aec v0.0.0-20170113033406-39771216ff4c // indirect
-	github.com/multiformats/go-multiaddr v0.2.0
+	github.com/multiformats/go-multiaddr v0.2.1
 	github.com/multiformats/go-multiaddr-dns v0.2.0
-	github.com/multiformats/go-multiaddr-net v0.1.2
+	github.com/multiformats/go-multiaddr-net v0.1.3
 	github.com/multiformats/go-multihash v0.0.13
-	github.com/onsi/ginkgo v1.10.3 // indirect
-	github.com/onsi/gomega v1.7.1 // indirect
 	github.com/opencontainers/go-digest v1.0.0-rc1 // indirect
 	github.com/opencontainers/image-spec v1.0.1 // indirect
 	github.com/otiai10/copy v1.0.2
@@ -97,7 +94,7 @@ require (
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/viper v1.5.0 // indirect
 	github.com/stretchr/testify v1.4.0
-	github.com/whyrusleeping/cbor-gen v0.0.0-20200321164527-9340289d0ca7
+	github.com/whyrusleeping/cbor-gen v0.0.0-20200402171437-3d27c146c105
 	github.com/whyrusleeping/go-logging v0.0.1
 	github.com/whyrusleeping/go-sysinfo v0.0.0-20190219211824-4a357d4b90b1
 	go.opencensus.io v0.22.3
@@ -113,5 +110,3 @@ require (
 )
 
 replace github.com/filecoin-project/filecoin-ffi => ./vendors/filecoin-ffi
-
-replace github.com/libp2p/go-libp2p-core => github.com/libp2p/go-libp2p-core v0.3.0

internal/app/go-filecoin/internal/submodule/network_submodule.go

@@ -10,7 +10,6 @@ import (
 	ds "github.com/ipfs/go-datastore"
 	graphsync "github.com/ipfs/go-graphsync"
 	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
-	"github.com/ipfs/go-graphsync/ipldbridge"
 	gsnet "github.com/ipfs/go-graphsync/network"
 	gsstoreutil "github.com/ipfs/go-graphsync/storeutil"
 	exchange "github.com/ipfs/go-ipfs-exchange-interface"
@@ -137,10 +136,9 @@ func NewNetworkSubmodule(ctx context.Context, config networkConfig, repo network
 
 	// set up graphsync
 	graphsyncNetwork := gsnet.NewFromLibp2pHost(peerHost)
-	bridge := ipldbridge.NewIPLDBridge()
 	loader := gsstoreutil.LoaderForBlockstore(blockstore.Blockstore)
 	storer := gsstoreutil.StorerForBlockstore(blockstore.Blockstore)
-	gsync := graphsyncimpl.New(ctx, graphsyncNetwork, bridge, loader, storer)
+	gsync := graphsyncimpl.New(ctx, graphsyncNetwork, loader, storer, graphsyncimpl.RejectAllRequestsByDefault())
 
 	// build network
 	network := net.New(peerHost, net.NewRouter(router), bandwidthTracker, net.NewPinger(peerHost, pingService))

internal/app/go-filecoin/internal/submodule/syncer_submodule.go

@@ -6,6 +6,8 @@ import (
 
 	fbig "github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-graphsync"
+	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/pkg/errors"
 
 	"github.com/filecoin-project/go-filecoin/internal/pkg/block"
@@ -75,6 +77,14 @@ func NewSyncerSubmodule(ctx context.Context, config syncerConfig, blockstore *Bl
 	nodeChainSelector := consensus.NewChainSelector(blockstore.CborStore, &stateViewer, config.GenesisCid())
 
 	// setup fecher
+	network.GraphExchange.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
+		_, has := requestData.Extension(fetcher.ChainsyncProtocolExtension)
+		if has {
+			// TODO: Don't just validate every request with the extension -- support only known selectors
+			// TODO: use separate block store for the chain (supported in GraphSync)
+			hookActions.ValidateRequest()
+		}
+	})
 	fetcher := fetcher.NewGraphSyncFetcher(ctx, network.GraphExchange, blockstore.Blockstore, blkValid, config.ChainClock(), discovery.PeerTracker)
 	faultCh := make(chan slashing.ConsensusFault)
 	faultDetector := slashing.NewConsensusFaultDetector(faultCh)

internal/pkg/chainsync/fetcher/graphsync_fetcher.go

@@ -52,6 +52,9 @@ const (
 	amtNodeValuesFieldIndex = 2
 )
 
+// ChainsyncProtocolExtension is the extension name to indicate graphsync requests are to sync the chain
+const ChainsyncProtocolExtension = graphsync.ExtensionName("chainsync")
+
 // interface conformance check
 var _ syncer.Fetcher = (*GraphSyncFetcher)(nil)
 
@@ -268,7 +271,7 @@ func (gsf *GraphSyncFetcher) fetchBlocks(ctx context.Context, selGen func() ipld
 	for _, c := range cids {
 		requestCtx, requestCancel := context.WithCancel(ctx)
 		defer requestCancel()
-		requestChan, errChan := gsf.exchange.Request(requestCtx, targetPeer, cidlink.Link{Cid: c}, selector)
+		requestChan, errChan := gsf.exchange.Request(requestCtx, targetPeer, cidlink.Link{Cid: c}, selector, graphsync.ExtensionData{Name: ChainsyncProtocolExtension})
 		wg.Add(1)
 		go func(requestChan <-chan graphsync.ResponseProgress, errChan <-chan error, cancelFunc func()) {
 			defer wg.Done()
@@ -353,7 +356,7 @@ func (gsf *GraphSyncFetcher) fetchBlocksRecursively(ctx context.Context, recSelG
 	defer requestCancel()
 	selector := recSelGen(recursionDepth)
 
-	requestChan, errChan := gsf.exchange.Request(requestCtx, targetPeer, cidlink.Link{Cid: baseCid}, selector)
+	requestChan, errChan := gsf.exchange.Request(requestCtx, targetPeer, cidlink.Link{Cid: baseCid}, selector, graphsync.ExtensionData{Name: ChainsyncProtocolExtension})
 	return gsf.consumeResponse(requestChan, errChan, requestCancel)
 }
 

internal/pkg/chainsync/fetcher/graphsync_fetcher_test.go

@@ -12,8 +12,8 @@ import (
 	"github.com/filecoin-project/go-address"
 	"github.com/ipfs/go-datastore"
 	dss "github.com/ipfs/go-datastore/sync"
-	graphsync "github.com/ipfs/go-graphsync/impl"
-	"github.com/ipfs/go-graphsync/ipldbridge"
+	"github.com/ipfs/go-graphsync"
+	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
 	gsnet "github.com/ipfs/go-graphsync/network"
 	gsstoreutil "github.com/ipfs/go-graphsync/storeutil"
 	bstore "github.com/ipfs/go-ipfs-blockstore"
@@ -803,8 +803,6 @@ func TestRealWorldGraphsyncFetchOnlyHeaders(t *testing.T) {
 
 	// setup a graphsync fetcher and a graphsync responder
 
-	bridge1 := ipldbridge.NewIPLDBridge()
-	bridge2 := ipldbridge.NewIPLDBridge()
 	bs := bstore.NewBlockstore(dss.MutexWrap(datastore.NewMapDatastore()))
 
 	bv := consensus.NewDefaultBlockValidator(chainClock)
@@ -814,7 +812,7 @@ func TestRealWorldGraphsyncFetchOnlyHeaders(t *testing.T) {
 	localLoader := gsstoreutil.LoaderForBlockstore(bs)
 	localStorer := gsstoreutil.StorerForBlockstore(bs)
 
-	localGraphsync := graphsync.New(ctx, gsnet1, bridge1, localLoader, localStorer)
+	localGraphsync := graphsyncimpl.New(ctx, gsnet1, localLoader, localStorer)
 
 	fetcher := fetcher.NewGraphSyncFetcher(ctx, localGraphsync, bs, bv, fc, pt)
 
@@ -826,7 +824,7 @@ func TestRealWorldGraphsyncFetchOnlyHeaders(t *testing.T) {
 		}
 		return bytes.NewBuffer(b.RawData()), nil
 	}
-	graphsync.New(ctx, gsnet2, bridge2, remoteLoader, nil)
+	graphsyncimpl.New(ctx, gsnet2, remoteLoader, nil)
 
 	tipsets, err := fetcher.FetchTipSetHeaders(ctx, final.Key(), host2.ID(), func(ts block.TipSet) (bool, error) {
 		if ts.Key().Equals(gen.Key()) {
@@ -897,8 +895,6 @@ func TestRealWorldGraphsyncFetchAcrossNetwork(t *testing.T) {
 
 	// setup a graphsync fetcher and a graphsync responder
 
-	bridge1 := ipldbridge.NewIPLDBridge()
-	bridge2 := ipldbridge.NewIPLDBridge()
 	bs := bstore.NewBlockstore(dss.MutexWrap(datastore.NewMapDatastore()))
 	bv := th.NewFakeBlockValidator()
 	fc := clock.NewFake(time.Now())
@@ -908,9 +904,8 @@ func TestRealWorldGraphsyncFetchAcrossNetwork(t *testing.T) {
 	localLoader := gsstoreutil.LoaderForBlockstore(bs)
 	localStorer := gsstoreutil.StorerForBlockstore(bs)
 
-	localGraphsync := graphsync.New(ctx, gsnet1, bridge1, localLoader, localStorer)
-
-	fetcher := fetcher.NewGraphSyncFetcher(ctx, localGraphsync, bs, bv, fc, pt)
+	localGraphsync := graphsyncimpl.New(ctx, gsnet1, localLoader, localStorer)
+	gsFetcher := fetcher.NewGraphSyncFetcher(ctx, localGraphsync, bs, bv, fc, pt)
 
 	remoteLoader := func(lnk ipld.Link, lnkCtx ipld.LinkContext) (io.Reader, error) {
 		cid := lnk.(cidlink.Link).Cid
@@ -920,9 +915,14 @@ func TestRealWorldGraphsyncFetchAcrossNetwork(t *testing.T) {
 		}
 		return bytes.NewBuffer(node.RawData()), nil
 	}
-	graphsync.New(ctx, gsnet2, bridge2, remoteLoader, nil)
-
-	tipsets, err := fetcher.FetchTipSets(ctx, final.Key(), host2.ID(), func(ts block.TipSet) (bool, error) {
+	otherGraphsync := graphsyncimpl.New(ctx, gsnet2, remoteLoader, nil, graphsyncimpl.RejectAllRequestsByDefault())
+	otherGraphsync.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
+		_, has := requestData.Extension(fetcher.ChainsyncProtocolExtension)
+		if has {
+			hookActions.ValidateRequest()
+		}
+	})
+	tipsets, err := gsFetcher.FetchTipSets(ctx, final.Key(), host2.ID(), func(ts block.TipSet) (bool, error) {
 		if ts.Key().Equals(gen.Key()) {
 			return true, nil
 		}

internal/pkg/chainsync/fetcher/graphsync_helpers_test.go

@@ -16,13 +16,13 @@ import (
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-graphsync"
-	"github.com/ipfs/go-graphsync/ipldbridge"
 	bstore "github.com/ipfs/go-ipfs-blockstore"
 	cbor "github.com/ipfs/go-ipld-cbor"
 	format "github.com/ipfs/go-ipld-format"
 	"github.com/ipld/go-ipld-prime"
 	ipldfree "github.com/ipld/go-ipld-prime/impl/free"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/ipld/go-ipld-prime/traversal"
 	"github.com/ipld/go-ipld-prime/traversal/selector"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/stretchr/testify/require"
@@ -250,16 +250,16 @@ func (mgs *mockableGraphsync) stubSingleResponseWithLoader(pid peer.ID, s select
 		return
 	}
 	visited := 0
-	visitor := func(tp ipldbridge.TraversalProgress, n ipld.Node, tr ipldbridge.TraversalReason) error {
+	visitor := func(tp traversal.Progress, n ipld.Node, tr traversal.VisitReason) error {
 		if hangup != noHangup && visited >= hangup {
 			return errHangup
 		}
 		visited++
 		responses = append(responses, graphsync.ResponseProgress{Node: n, Path: tp.Path, LastBlock: tp.LastBlock})
 		return nil
 	}
-	err = ipldbridge.TraversalProgress{
-		Cfg: &ipldbridge.TraversalConfig{
+	err = traversal.Progress{
+		Cfg: &traversal.Config{
 			Ctx:        mgs.ctx,
 			LinkLoader: linkLoader,
 		},

internal/pkg/protocol/storage/api.go

@@ -74,5 +74,5 @@ func (api *API) ProposeStorageDeal(
 
 // GetStorageDeal retrieves information about an in-progress deal
 func (api *API) GetStorageDeal(ctx context.Context, c cid.Cid) (storagemarket.ClientDeal, error) {
-	return api.storage.Client().GetInProgressDeal(ctx, c)
+	return api.storage.Client().GetLocalDeal(ctx, c)
 }