bzzeth: initial support for bzz-eth protocol (ethersphere#1571)

* bzzeth: phase 0: handshake

Author: acud

bzzeth/bzzeth.go

@@ -0,0 +1,116 @@
+// Copyright 2019 The Swarm Authors
+// This file is part of the Swarm library.
+//
+// The Swarm library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The Swarm library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the Swarm library. If not, see <http://www.gnu.org/licenses/>.
+
+package bzzeth
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethersphere/swarm/log"
+	"github.com/ethersphere/swarm/p2p/protocols"
+)
+
+// BzzEth implements node.Service
+var _ node.Service = &BzzEth{}
+
+// BzzEth is a global module handling ethereum state on swarm
+type BzzEth struct {
+	peers *peers        // bzzeth peer pool
+	quit  chan struct{} // quit channel to close go routines
+}
+
+// New constructs the BzzEth node service
+func New() *BzzEth {
+	return &BzzEth{
+		peers: newPeers(),
+		quit:  make(chan struct{}),
+	}
+}
+
+// Run is the bzzeth protocol run function.
+// - creates a peer
+// - checks if it is a swarm node, put the protocol in idle mode
+// - performs handshake
+// - adds peer to the peerpool
+// - starts incoming message handler loop
+func (b *BzzEth) Run(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+	peer := protocols.NewPeer(p, rw, Spec)
+	bp := NewPeer(peer)
+
+	// perform handshake and register if peer serves headers
+	handshake, err := bp.Handshake(context.TODO(), Handshake{ServeHeaders: true}, nil)
+	if err != nil {
+		return err
+	}
+	bp.serveHeaders = handshake.(*Handshake).ServeHeaders
+	log.Debug("handshake", "hs", handshake, "peer", bp)
+
+	// This protocol is all about interaction between an Eth node and a Swarm Node.
+	// If another swarm node tries to connect then the protocol goes into idle
+	if isSwarmNodeFunc(bp) {
+		<-b.quit
+		return nil
+	}
+	b.peers.add(bp)
+	defer b.peers.remove(bp)
+
+	return peer.Run(b.handleMsg(bp))
+}
+
+// handleMsg is the message handler that delegates incoming messages
+// handlers are called asynchronously so handler calls do not block incoming msg processing
+func (b *BzzEth) handleMsg(p *Peer) func(context.Context, interface{}) error {
+	return func(ctx context.Context, msg interface{}) error {
+		p.logger.Debug("bzzeth.handleMsg")
+		switch msg.(type) {
+		default:
+		}
+		return nil
+	}
+}
+
+// Protocols returns the p2p protocol
+func (b *BzzEth) Protocols() []p2p.Protocol {
+	return []p2p.Protocol{
+		{
+			Name:    Spec.Name,
+			Version: Spec.Version,
+			Length:  Spec.Length(),
+			Run:     b.Run,
+		},
+	}
+}
+
+// APIs return APIs defined on the node service
+func (b *BzzEth) APIs() []rpc.API {
+	return nil
+}
+
+// Start starts the BzzEth node service
+func (b *BzzEth) Start(server *p2p.Server) error {
+	log.Info("bzzeth starting...")
+	return nil
+}
+
+// Stop stops the BzzEth node service
+func (b *BzzEth) Stop() error {
+	log.Info("bzzeth shutting down...")
+	close(b.quit)
+	return nil
+}

bzzeth/bzzeth_test.go

@@ -0,0 +1,157 @@
+// Copyright 2019 The Swarm Authors
+// This file is part of the Swarm library.
+//
+// The Swarm library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The Swarm library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the Swarm library. If not, see <http://www.gnu.org/licenses/>.
+
+package bzzeth
+
+import (
+	"errors"
+	"flag"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	p2ptest "github.com/ethersphere/swarm/p2p/testing"
+)
+
+var (
+	loglevel = flag.Int("loglevel", 0, "verbosity of logs")
+)
+
+func init() {
+	flag.Parse()
+
+	log.PrintOrigins(true)
+	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
+}
+
+func newBzzEthTester() (*p2ptest.ProtocolTester, *BzzEth, func(), error) {
+	b := New()
+
+	prvkey, err := crypto.GenerateKey()
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	protocolTester := p2ptest.NewProtocolTester(prvkey, 1, b.Run)
+	teardown := func() {
+		protocolTester.Stop()
+	}
+
+	return protocolTester, b, teardown, nil
+}
+
+func handshakeExchange(tester *p2ptest.ProtocolTester, peerID enode.ID, serveHeadersPeer, serveHeadersPivot bool) error {
+	return tester.TestExchanges(
+		p2ptest.Exchange{
+			Label: "Handshake",
+			Triggers: []p2ptest.Trigger{
+				{
+					Code: 0,
+					Msg: Handshake{
+						ServeHeaders: serveHeadersPeer,
+					},
+					Peer: peerID,
+				},
+			},
+			Expects: []p2ptest.Expect{
+				{
+					Code: 0,
+					Msg: Handshake{
+						ServeHeaders: serveHeadersPivot,
+					},
+					Peer: peerID,
+				},
+			},
+		})
+}
+
+// tests handshake between eth node and swarm node
+// on successful handshake the protocol does not go idle
+// peer added to the pool and serves headers is registered
+func TestBzzEthHandshake(t *testing.T) {
+	tester, b, teardown, err := newBzzEthTester()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer teardown()
+
+	node := tester.Nodes[0]
+	err = handshakeExchange(tester, node.ID(), true, true)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	// after successful handshake, expect peer added to peer pool
+	var p *Peer
+	for i := 0; i < 10; i++ {
+		p = b.peers.get(node.ID())
+		if p != nil {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	if p == nil {
+		t.Fatal("bzzeth peer not added")
+	}
+
+	if !p.serveHeaders {
+		t.Fatal("bzzeth peer serveHeaders not set")
+	}
+
+	close(b.quit)
+	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("?")})
+	if err == nil || err.Error() != "timed out waiting for peers to disconnect" {
+		t.Fatal(err)
+	}
+}
+
+// TestBzzBzzHandshake tests that a handshake between two Swarm nodes
+func TestBzzBzzHandshake(t *testing.T) {
+	tester, b, teardown, err := newBzzEthTester()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer teardown()
+
+	// redefine isSwarmNodeFunc to force recognise remote peer as swarm node
+	defer func(f func(*Peer) bool) {
+		isSwarmNodeFunc = f
+	}(isSwarmNodeFunc)
+	isSwarmNodeFunc = func(_ *Peer) bool { return true }
+
+	node := tester.Nodes[0]
+	err = handshakeExchange(tester, node.ID(), false, true)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	// after handshake expect protocol to hang, peer not added to pool
+	p := b.peers.get(node.ID())
+	if p != nil {
+		t.Fatal("bzzeth swarm peer incorrectly added")
+	}
+
+	// after closing the ptotocall, expect disconnect
+	close(b.quit)
+	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("protocol returned")})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+}