From 86d67eda62cbc51b7b55a72b638ee010ed865ba0 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Tue, 11 Sep 2018 18:36:35 +0200 Subject: [PATCH] swarm/network: port to p2p/enode Swarm has its own node address representation, BzzAddr, containing both an overlay address (the hash of a secp256k1 public key) and an underlay address (enode:// URL). There are no changes to the BzzAddr format in this commit, but certain operations such as creating a BzzAddr from a node ID are now impossible because node IDs aren't public keys anymore. Most swarm-related changes in the series remove uses of NewAddrFromNodeID, replacing it with NewAddr which takes a complete node as argument. ToOverlayAddr is removed because we can just use the node ID directly. --- swarm/network/discovery.go | 4 +- swarm/network/discovery_test.go | 6 +-- swarm/network/fetcher.go | 38 +++++++-------- swarm/network/fetcher_test.go | 18 +++---- swarm/network/hive.go | 40 ++++++++++++---- swarm/network/hive_test.go | 17 ++++--- swarm/network/networkid_test.go | 23 +++++---- swarm/network/protocol.go | 85 +++++++++++++-------------------- swarm/network/protocol_test.go | 50 +++++++++---------- 9 files changed, 141 insertions(+), 140 deletions(-) diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index 3019594809..21703e70f1 100644 --- a/swarm/network/discovery.go +++ b/swarm/network/discovery.go @@ -87,7 +87,7 @@ func NotifyPeer(p *BzzAddr, k *Kademlia) { // unless already notified during the connection session func (d *Peer) NotifyPeer(a *BzzAddr, po uint8) { // immediately return - if (po < d.getDepth() && pot.ProxCmp(d.localAddr, d, a) != 1) || d.seen(a) { + if (po < d.getDepth() && pot.ProxCmp(d.kad.BaseAddr(), d, a) != 1) || d.seen(a) { return } resp := &peersMsg{ @@ -161,7 +161,7 @@ func (d *Peer) handleSubPeersMsg(msg *subPeersMsg) error { d.setDepth(msg.Depth) var peers []*BzzAddr d.kad.EachConn(d.Over(), 255, func(p *Peer, po int, isproxbin bool) bool { - if pob, _ := pof(d, d.localAddr, 0); pob > po { + if pob, _ := pof(d, d.kad.BaseAddr(), 0); pob > po { return false } if !d.seen(p.BzzAddr) { diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index 494bc81969..dd3299c0f1 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -31,8 +31,8 @@ func TestDiscovery(t *testing.T) { params := NewHiveParams() s, pp := newHiveTester(t, params, 1, nil) - id := s.IDs[0] - raddr := NewAddrFromNodeID(id) + node := s.Nodes[0] + raddr := NewAddr(node) pp.Register(raddr) // start the hive and wait for the connection @@ -46,7 +46,7 @@ func TestDiscovery(t *testing.T) { { Code: 1, Msg: &subPeersMsg{Depth: 0}, - Peer: id, + Peer: node.ID(), }, }, }) diff --git a/swarm/network/fetcher.go b/swarm/network/fetcher.go index 413b40cb5b..5b4b61c7e2 100644 --- a/swarm/network/fetcher.go +++ b/swarm/network/fetcher.go @@ -22,7 +22,7 @@ import ( "time" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -32,7 +32,7 @@ var searchTimeout = 1 * time.Second // Also used in stream delivery. var RequestTimeout = 10 * time.Second -type RequestFunc func(context.Context, *Request) (*discover.NodeID, chan struct{}, error) +type RequestFunc func(context.Context, *Request) (*enode.ID, chan struct{}, error) // Fetcher is created when a chunk is not found locally. It starts a request handler loop once and // keeps it alive until all active requests are completed. This can happen: @@ -41,18 +41,18 @@ type RequestFunc func(context.Context, *Request) (*discover.NodeID, chan struct{ // Fetcher self destroys itself after it is completed. // TODO: cancel all forward requests after termination type Fetcher struct { - protoRequestFunc RequestFunc // request function fetcher calls to issue retrieve request for a chunk - addr storage.Address // the address of the chunk to be fetched - offerC chan *discover.NodeID // channel of sources (peer node id strings) + protoRequestFunc RequestFunc // request function fetcher calls to issue retrieve request for a chunk + addr storage.Address // the address of the chunk to be fetched + offerC chan *enode.ID // channel of sources (peer node id strings) requestC chan struct{} skipCheck bool } type Request struct { - Addr storage.Address // chunk address - Source *discover.NodeID // nodeID of peer to request from (can be nil) - SkipCheck bool // whether to offer the chunk first or deliver directly - peersToSkip *sync.Map // peers not to request chunk from (only makes sense if source is nil) + Addr storage.Address // chunk address + Source *enode.ID // nodeID of peer to request from (can be nil) + SkipCheck bool // whether to offer the chunk first or deliver directly + peersToSkip *sync.Map // peers not to request chunk from (only makes sense if source is nil) } // NewRequest returns a new instance of Request based on chunk address skip check and @@ -112,14 +112,14 @@ func NewFetcher(addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher { return &Fetcher{ addr: addr, protoRequestFunc: rf, - offerC: make(chan *discover.NodeID), + offerC: make(chan *enode.ID), requestC: make(chan struct{}), skipCheck: skipCheck, } } // Offer is called when an upstream peer offers the chunk via syncing as part of `OfferedHashesMsg` and the node does not have the chunk locally. -func (f *Fetcher) Offer(ctx context.Context, source *discover.NodeID) { +func (f *Fetcher) Offer(ctx context.Context, source *enode.ID) { // First we need to have this select to make sure that we return if context is done select { case <-ctx.Done(): @@ -156,13 +156,13 @@ func (f *Fetcher) Request(ctx context.Context) { // it keeps the Fetcher alive within the lifecycle of the passed context func (f *Fetcher) run(ctx context.Context, peers *sync.Map) { var ( - doRequest bool // determines if retrieval is initiated in the current iteration - wait *time.Timer // timer for search timeout - waitC <-chan time.Time // timer channel - sources []*discover.NodeID // known sources, ie. peers that offered the chunk - requested bool // true if the chunk was actually requested + doRequest bool // determines if retrieval is initiated in the current iteration + wait *time.Timer // timer for search timeout + waitC <-chan time.Time // timer channel + sources []*enode.ID // known sources, ie. peers that offered the chunk + requested bool // true if the chunk was actually requested ) - gone := make(chan *discover.NodeID) // channel to signal that a peer we requested from disconnected + gone := make(chan *enode.ID) // channel to signal that a peer we requested from disconnected // loop that keeps the fetching process alive // after every request a timer is set. If this goes off we request again from another peer @@ -251,9 +251,9 @@ func (f *Fetcher) run(ctx context.Context, peers *sync.Map) { // * the peer's address is added to the set of peers to skip // * the peer's address is removed from prospective sources, and // * a go routine is started that reports on the gone channel if the peer is disconnected (or terminated their streamer) -func (f *Fetcher) doRequest(ctx context.Context, gone chan *discover.NodeID, peersToSkip *sync.Map, sources []*discover.NodeID) ([]*discover.NodeID, error) { +func (f *Fetcher) doRequest(ctx context.Context, gone chan *enode.ID, peersToSkip *sync.Map, sources []*enode.ID) ([]*enode.ID, error) { var i int - var sourceID *discover.NodeID + var sourceID *enode.ID var quit chan struct{} req := &Request{ diff --git a/swarm/network/fetcher_test.go b/swarm/network/fetcher_test.go index 21b81d652c..b2316b0976 100644 --- a/swarm/network/fetcher_test.go +++ b/swarm/network/fetcher_test.go @@ -22,11 +22,11 @@ import ( "testing" "time" - "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/enode" ) -var requestedPeerID = discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") -var sourcePeerID = discover.MustHexID("2dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") +var requestedPeerID = enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8") +var sourcePeerID = enode.HexID("99d8594b52298567d2ca3f4c441a5ba0140ee9245e26460d01102a52773c73b9") // mockRequester pushes every request to the requestC channel when its doRequest function is called type mockRequester struct { @@ -45,7 +45,7 @@ func newMockRequester(waitTimes ...time.Duration) *mockRequester { } } -func (m *mockRequester) doRequest(ctx context.Context, request *Request) (*discover.NodeID, chan struct{}, error) { +func (m *mockRequester) doRequest(ctx context.Context, request *Request) (*enode.ID, chan struct{}, error) { waitTime := time.Duration(0) if m.ctr < len(m.waitTimes) { waitTime = m.waitTimes[m.ctr] @@ -389,9 +389,9 @@ func TestFetcherRequestQuitRetriesRequest(t *testing.T) { // and not skip unknown one. func TestRequestSkipPeer(t *testing.T) { addr := make([]byte, 32) - peers := []discover.NodeID{ - discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"), - discover.MustHexID("2dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"), + peers := []enode.ID{ + enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8"), + enode.HexID("99d8594b52298567d2ca3f4c441a5ba0140ee9245e26460d01102a52773c73b9"), } peersToSkip := new(sync.Map) @@ -411,7 +411,7 @@ func TestRequestSkipPeer(t *testing.T) { // after RequestTimeout has passed. func TestRequestSkipPeerExpired(t *testing.T) { addr := make([]byte, 32) - peer := discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") + peer := enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8") // set RequestTimeout to a low value and reset it after the test defer func(t time.Duration) { RequestTimeout = t }(RequestTimeout) @@ -437,7 +437,7 @@ func TestRequestSkipPeerExpired(t *testing.T) { // by value to peersToSkip map is not time.Duration. func TestRequestSkipPeerPermanent(t *testing.T) { addr := make([]byte, 32) - peer := discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") + peer := enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8") // set RequestTimeout to a low value and reset it after the test defer func(t time.Duration) { RequestTimeout = t }(RequestTimeout) diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 425c1d5a1e..1aa1ae42a2 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -23,7 +23,7 @@ import ( "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/state" ) @@ -56,12 +56,13 @@ func NewHiveParams() *HiveParams { // Hive manages network connections of the swarm node type Hive struct { - *HiveParams // settings - *Kademlia // the overlay connectiviy driver - Store state.Store // storage interface to save peers across sessions - addPeer func(*discover.Node) // server callback to connect to a peer + *HiveParams // settings + *Kademlia // the overlay connectiviy driver + Store state.Store // storage interface to save peers across sessions + addPeer func(*enode.Node) // server callback to connect to a peer // bookkeeping lock sync.Mutex + peers map[enode.ID]*BzzPeer ticker *time.Ticker } @@ -74,6 +75,7 @@ func NewHive(params *HiveParams, kad *Kademlia, store state.Store) *Hive { HiveParams: params, Kademlia: kad, Store: store, + peers: make(map[enode.ID]*BzzPeer), } } @@ -137,7 +139,7 @@ func (h *Hive) connect() { } log.Trace(fmt.Sprintf("%08x hive connect() suggested %08x", h.BaseAddr()[:4], addr.Address()[:4])) - under, err := discover.ParseNode(string(addr.Under())) + under, err := enode.ParseV4(string(addr.Under())) if err != nil { log.Warn(fmt.Sprintf("%08x unable to connect to bee %08x: invalid node URL: %v", h.BaseAddr()[:4], addr.Address()[:4], err)) continue @@ -149,6 +151,9 @@ func (h *Hive) connect() { // Run protocol run function func (h *Hive) Run(p *BzzPeer) error { + h.trackPeer(p) + defer h.untrackPeer(p) + dp := NewPeer(p, h.Kademlia) depth, changed := h.On(dp) // if we want discovery, advertise change of depth @@ -166,6 +171,18 @@ func (h *Hive) Run(p *BzzPeer) error { return dp.Run(dp.HandleMsg) } +func (h *Hive) trackPeer(p *BzzPeer) { + h.lock.Lock() + h.peers[p.ID()] = p + h.lock.Unlock() +} + +func (h *Hive) untrackPeer(p *BzzPeer) { + h.lock.Lock() + delete(h.peers, p.ID()) + h.lock.Unlock() +} + // NodeInfo function is used by the p2p.server RPC interface to display // protocol specific node information func (h *Hive) NodeInfo() interface{} { @@ -174,8 +191,15 @@ func (h *Hive) NodeInfo() interface{} { // PeerInfo function is used by the p2p.server RPC interface to display // protocol specific information any connected peer referred to by their NodeID -func (h *Hive) PeerInfo(id discover.NodeID) interface{} { - addr := NewAddrFromNodeID(id) +func (h *Hive) PeerInfo(id enode.ID) interface{} { + h.lock.Lock() + p := h.peers[id] + h.lock.Unlock() + + if p == nil { + return nil + } + addr := NewAddr(p.Node()) return struct { OAddr hexutil.Bytes UAddr hexutil.Bytes diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index 7ea000c1ae..059c3dc96d 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -39,8 +39,8 @@ func TestRegisterAndConnect(t *testing.T) { params := NewHiveParams() s, pp := newHiveTester(t, params, 1, nil) - id := s.IDs[0] - raddr := NewAddrFromNodeID(id) + node := s.Nodes[0] + raddr := NewAddr(node) pp.Register(raddr) // start the hive and wait for the connection @@ -51,7 +51,7 @@ func TestRegisterAndConnect(t *testing.T) { defer pp.Stop() // retrieve and broadcast err = s.TestDisconnected(&p2ptest.Disconnect{ - Peer: s.IDs[0], + Peer: s.Nodes[0].ID(), Error: nil, }) @@ -75,8 +75,8 @@ func TestHiveStatePersistance(t *testing.T) { s, pp := newHiveTester(t, params, 5, store) peers := make(map[string]bool) - for _, id := range s.IDs { - raddr := NewAddrFromNodeID(id) + for _, node := range s.Nodes { + raddr := NewAddr(node) pp.Register(raddr) peers[raddr.String()] = true } @@ -102,7 +102,10 @@ func TestHiveStatePersistance(t *testing.T) { i++ return true }) - if len(peers) != 0 || i != 5 { - t.Fatalf("invalid peers loaded") + if i != 5 { + t.Errorf("invalid number of entries: got %v, want %v", i, 5) + } + if len(peers) != 0 { + t.Fatalf("%d peers left over: %v", len(peers), peers) } } diff --git a/swarm/network/networkid_test.go b/swarm/network/networkid_test.go index 91a1f6d7bf..d1d359de68 100644 --- a/swarm/network/networkid_test.go +++ b/swarm/network/networkid_test.go @@ -29,7 +29,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" @@ -38,8 +38,8 @@ import ( var ( currentNetworkID int cnt int - nodeMap map[int][]discover.NodeID - kademlias map[discover.NodeID]*Kademlia + nodeMap map[int][]enode.ID + kademlias map[enode.ID]*Kademlia ) const ( @@ -70,7 +70,7 @@ func TestNetworkID(t *testing.T) { //arbitrarily set the number of nodes. It could be any number numNodes := 24 //the nodeMap maps all nodes (slice value) with the same network ID (key) - nodeMap = make(map[int][]discover.NodeID) + nodeMap = make(map[int][]enode.ID) //set up the network and connect nodes net, err := setupNetwork(numNodes) if err != nil { @@ -95,7 +95,7 @@ func TestNetworkID(t *testing.T) { kademlias[node].EachAddr(nil, 0, func(addr *BzzAddr, _ int, _ bool) bool { found := false for _, nd := range netIDGroup { - p := ToOverlayAddr(nd.Bytes()) + p := nd.Bytes() if bytes.Equal(p, addr.Address()) { found = true } @@ -183,12 +183,11 @@ func setupNetwork(numnodes int) (net *simulations.Network, err error) { } func newServices() adapters.Services { - kademlias = make(map[discover.NodeID]*Kademlia) - kademlia := func(id discover.NodeID) *Kademlia { + kademlias = make(map[enode.ID]*Kademlia) + kademlia := func(id enode.ID) *Kademlia { if k, ok := kademlias[id]; ok { return k } - addr := NewAddrFromNodeID(id) params := NewKadParams() params.MinProxBinSize = 2 params.MaxBinSize = 3 @@ -196,19 +195,19 @@ func newServices() adapters.Services { params.MaxRetries = 1000 params.RetryExponent = 2 params.RetryInterval = 1000000 - kademlias[id] = NewKademlia(addr.Over(), params) + kademlias[id] = NewKademlia(id[:], params) return kademlias[id] } return adapters.Services{ "bzz": func(ctx *adapters.ServiceContext) (node.Service, error) { - addr := NewAddrFromNodeID(ctx.Config.ID) + addr := NewAddr(ctx.Config.Node()) hp := NewHiveParams() hp.Discovery = false cnt++ //assign the network ID currentNetworkID = cnt % NumberOfNets if ok := nodeMap[currentNetworkID]; ok == nil { - nodeMap[currentNetworkID] = make([]discover.NodeID, 0) + nodeMap[currentNetworkID] = make([]enode.ID, 0) } //add this node to the group sharing the same network ID nodeMap[currentNetworkID] = append(nodeMap[currentNetworkID], ctx.Config.ID) @@ -224,7 +223,7 @@ func newServices() adapters.Services { } } -func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) { +func watchSubscriptionEvents(ctx context.Context, id enode.ID, client *rpc.Client, errc chan error, quitC chan struct{}) { events := make(chan *p2p.PeerEvent) sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index d509d157bb..66ae94a88b 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -26,7 +26,7 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/log" @@ -78,7 +78,7 @@ type Bzz struct { LightNode bool localAddr *BzzAddr mtx sync.Mutex - handshakes map[discover.NodeID]*HandshakeMsg + handshakes map[enode.ID]*HandshakeMsg streamerSpec *protocols.Spec streamerRun func(*BzzPeer) error } @@ -94,7 +94,7 @@ func NewBzz(config *BzzConfig, kad *Kademlia, store state.Store, streamerSpec *p NetworkID: config.NetworkID, LightNode: config.LightNode, localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr}, - handshakes: make(map[discover.NodeID]*HandshakeMsg), + handshakes: make(map[enode.ID]*HandshakeMsg), streamerRun: streamerRun, streamerSpec: streamerSpec, } @@ -183,7 +183,6 @@ func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*BzzPeer) error) func(* // the handshake has succeeded so construct the BzzPeer and run the protocol peer := &BzzPeer{ Peer: protocols.NewPeer(p, rw, spec), - localAddr: b.localAddr, BzzAddr: handshake.peerAddr, lastActive: time.Now(), LightNode: handshake.LightNode, @@ -218,14 +217,14 @@ func (b *Bzz) performHandshake(p *protocols.Peer, handshake *HandshakeMsg) error func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error { handshake, _ := b.GetHandshake(p.ID()) if !<-handshake.init { - return fmt.Errorf("%08x: bzz already started on peer %08x", b.localAddr.Over()[:4], ToOverlayAddr(p.ID().Bytes())[:4]) + return fmt.Errorf("%08x: bzz already started on peer %08x", b.localAddr.Over()[:4], p.ID().Bytes()[:4]) } close(handshake.init) defer b.removeHandshake(p.ID()) peer := protocols.NewPeer(p, rw, BzzSpec) err := b.performHandshake(peer, handshake) if err != nil { - log.Warn(fmt.Sprintf("%08x: handshake failed with remote peer %08x: %v", b.localAddr.Over()[:4], ToOverlayAddr(p.ID().Bytes())[:4], err)) + log.Warn(fmt.Sprintf("%08x: handshake failed with remote peer %08x: %v", b.localAddr.Over()[:4], p.ID().Bytes()[:4], err)) return err } @@ -242,18 +241,13 @@ func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error { // implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer type BzzPeer struct { *protocols.Peer // represents the connection for online peers - localAddr *BzzAddr // local Peers address *BzzAddr // remote address -> implements Addr interface = protocols.Peer lastActive time.Time // time is updated whenever mutexes are releasing LightNode bool } -func NewBzzPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer { - return &BzzPeer{ - Peer: p, - localAddr: addr, - BzzAddr: NewAddrFromNodeID(p.ID()), - } +func NewBzzPeer(p *protocols.Peer) *BzzPeer { + return &BzzPeer{Peer: p, BzzAddr: NewAddr(p.Node())} } // LastActive returns the time the peer was last active @@ -261,6 +255,14 @@ func (p *BzzPeer) LastActive() time.Time { return p.lastActive } +// ID returns the peer's underlay node identifier. +func (p *BzzPeer) ID() enode.ID { + // This is here to resolve a method tie: both protocols.Peer and BzzAddr are embedded + // into the struct and provide ID(). The protocols.Peer version is faster, ensure it + // gets used. + return p.Peer.ID() +} + /* Handshake @@ -301,14 +303,14 @@ func (b *Bzz) checkHandshake(hs interface{}) error { // removeHandshake removes handshake for peer with peerID // from the bzz handshake store -func (b *Bzz) removeHandshake(peerID discover.NodeID) { +func (b *Bzz) removeHandshake(peerID enode.ID) { b.mtx.Lock() defer b.mtx.Unlock() delete(b.handshakes, peerID) } // GetHandshake returns the bzz handhake that the remote peer with peerID sent -func (b *Bzz) GetHandshake(peerID discover.NodeID) (*HandshakeMsg, bool) { +func (b *Bzz) GetHandshake(peerID enode.ID) (*HandshakeMsg, bool) { b.mtx.Lock() defer b.mtx.Unlock() handshake, found := b.handshakes[peerID] @@ -336,24 +338,28 @@ type BzzAddr struct { UAddr []byte } -// Address implements OverlayPeer interface to be used in Overlay +// Address implements OverlayPeer interface to be used in Overlay. func (a *BzzAddr) Address() []byte { return a.OAddr } -// Over returns the overlay address +// Over returns the overlay address. func (a *BzzAddr) Over() []byte { return a.OAddr } -// Under returns the underlay address +// Under returns the underlay address. func (a *BzzAddr) Under() []byte { return a.UAddr } -// ID returns the nodeID from the underlay enode address -func (a *BzzAddr) ID() discover.NodeID { - return discover.MustParseNode(string(a.UAddr)).ID +// ID returns the node identifier in the underlay. +func (a *BzzAddr) ID() enode.ID { + n, err := enode.ParseV4(string(a.UAddr)) + if err != nil { + return enode.ID{} + } + return n.ID() } // Update updates the underlay address of a peer record @@ -372,38 +378,11 @@ func RandomAddr() *BzzAddr { if err != nil { panic("unable to generate key") } - pubkey := crypto.FromECDSAPub(&key.PublicKey) - var id discover.NodeID - copy(id[:], pubkey[1:]) - return NewAddrFromNodeID(id) + node := enode.NewV4(&key.PublicKey, net.IP{127, 0, 0, 1}, 30303, 30303) + return NewAddr(node) } -// NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID -func NewNodeIDFromAddr(addr *BzzAddr) discover.NodeID { - log.Info(fmt.Sprintf("uaddr=%s", string(addr.UAddr))) - node := discover.MustParseNode(string(addr.UAddr)) - return node.ID -} - -// NewAddrFromNodeID constucts a BzzAddr from a discover.NodeID -// the overlay address is derived as the hash of the nodeID -func NewAddrFromNodeID(id discover.NodeID) *BzzAddr { - return &BzzAddr{ - OAddr: ToOverlayAddr(id.Bytes()), - UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()), - } -} - -// NewAddrFromNodeIDAndPort constucts a BzzAddr from a discover.NodeID and port uint16 -// the overlay address is derived as the hash of the nodeID -func NewAddrFromNodeIDAndPort(id discover.NodeID, host net.IP, port uint16) *BzzAddr { - return &BzzAddr{ - OAddr: ToOverlayAddr(id.Bytes()), - UAddr: []byte(discover.NewNode(id, host, port, port).String()), - } -} - -// ToOverlayAddr creates an overlayaddress from a byte slice -func ToOverlayAddr(id []byte) []byte { - return crypto.Keccak256(id) +// NewAddr constucts a BzzAddr from a node record. +func NewAddr(node *enode.Node) *BzzAddr { + return &BzzAddr{OAddr: node.ID().Bytes(), UAddr: []byte(node.String())} } diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index 0fc8583711..4b83c7a278 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -25,7 +25,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/protocols" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" ) @@ -71,7 +71,7 @@ func (t *testStore) Save(key string, v []byte) error { return nil } -func HandshakeMsgExchange(lhs, rhs *HandshakeMsg, id discover.NodeID) []p2ptest.Exchange { +func HandshakeMsgExchange(lhs, rhs *HandshakeMsg, id enode.ID) []p2ptest.Exchange { return []p2ptest.Exchange{ { @@ -108,17 +108,13 @@ func newBzzBaseTester(t *testing.T, n int, addr *BzzAddr, spec *protocols.Spec, } protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - return srv(&BzzPeer{ - Peer: protocols.NewPeer(p, rw, spec), - localAddr: addr, - BzzAddr: NewAddrFromNodeID(p.ID()), - }) + return srv(&BzzPeer{Peer: protocols.NewPeer(p, rw, spec), BzzAddr: NewAddr(p.Node())}) } - s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, protocol) + s := p2ptest.NewProtocolTester(t, addr.ID(), n, protocol) - for _, id := range s.IDs { - cs[id.String()] = make(chan bool) + for _, node := range s.Nodes { + cs[node.ID().String()] = make(chan bool) } return &bzzTester{ @@ -150,7 +146,7 @@ func newBzz(addr *BzzAddr, lightNode bool) *Bzz { func newBzzHandshakeTester(t *testing.T, n int, addr *BzzAddr, lightNode bool) *bzzTester { bzz := newBzz(addr, lightNode) - pt := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, bzz.runBzz) + pt := p2ptest.NewProtocolTester(t, addr.ID(), n, bzz.runBzz) return &bzzTester{ addr: addr, @@ -161,14 +157,14 @@ func newBzzHandshakeTester(t *testing.T, n int, addr *BzzAddr, lightNode bool) * // should test handshakes in one exchange? parallelisation func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptest.Disconnect) error { - var peers []discover.NodeID - id := NewNodeIDFromAddr(rhs.Addr) + var peers []enode.ID + id := rhs.Addr.ID() if len(disconnects) > 0 { for _, d := range disconnects { peers = append(peers, d.Peer) } } else { - peers = []discover.NodeID{id} + peers = []enode.ID{id} } if err := s.TestExchanges(HandshakeMsgExchange(lhs, rhs, id)...); err != nil { @@ -181,7 +177,7 @@ func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptes // If we don't expect disconnect, ensure peers remain connected err := s.TestDisconnected(&p2ptest.Disconnect{ - Peer: s.IDs[0], + Peer: s.Nodes[0].ID(), Error: nil, }) @@ -209,12 +205,12 @@ func TestBzzHandshakeNetworkIDMismatch(t *testing.T) { lightNode := false addr := RandomAddr() s := newBzzHandshakeTester(t, 1, addr, lightNode) - id := s.IDs[0] + node := s.Nodes[0] err := s.testHandshake( correctBzzHandshake(addr, lightNode), - &HandshakeMsg{Version: TestProtocolVersion, NetworkID: 321, Addr: NewAddrFromNodeID(id)}, - &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): network id mismatch 321 (!= 3)")}, + &HandshakeMsg{Version: TestProtocolVersion, NetworkID: 321, Addr: NewAddr(node)}, + &p2ptest.Disconnect{Peer: node.ID(), Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): network id mismatch 321 (!= 3)")}, ) if err != nil { @@ -226,12 +222,12 @@ func TestBzzHandshakeVersionMismatch(t *testing.T) { lightNode := false addr := RandomAddr() s := newBzzHandshakeTester(t, 1, addr, lightNode) - id := s.IDs[0] + node := s.Nodes[0] err := s.testHandshake( correctBzzHandshake(addr, lightNode), - &HandshakeMsg{Version: 0, NetworkID: TestProtocolNetworkID, Addr: NewAddrFromNodeID(id)}, - &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): version mismatch 0 (!= %d)", TestProtocolVersion)}, + &HandshakeMsg{Version: 0, NetworkID: TestProtocolNetworkID, Addr: NewAddr(node)}, + &p2ptest.Disconnect{Peer: node.ID(), Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): version mismatch 0 (!= %d)", TestProtocolVersion)}, ) if err != nil { @@ -243,11 +239,11 @@ func TestBzzHandshakeSuccess(t *testing.T) { lightNode := false addr := RandomAddr() s := newBzzHandshakeTester(t, 1, addr, lightNode) - id := s.IDs[0] + node := s.Nodes[0] err := s.testHandshake( correctBzzHandshake(addr, lightNode), - &HandshakeMsg{Version: TestProtocolVersion, NetworkID: TestProtocolNetworkID, Addr: NewAddrFromNodeID(id)}, + &HandshakeMsg{Version: TestProtocolVersion, NetworkID: TestProtocolNetworkID, Addr: NewAddr(node)}, ) if err != nil { @@ -268,8 +264,8 @@ func TestBzzHandshakeLightNode(t *testing.T) { t.Run(test.name, func(t *testing.T) { randomAddr := RandomAddr() pt := newBzzHandshakeTester(t, 1, randomAddr, false) - id := pt.IDs[0] - addr := NewAddrFromNodeID(id) + node := pt.Nodes[0] + addr := NewAddr(node) err := pt.testHandshake( correctBzzHandshake(randomAddr, false), @@ -280,8 +276,8 @@ func TestBzzHandshakeLightNode(t *testing.T) { t.Fatal(err) } - if pt.bzz.handshakes[id].LightNode != test.lightNode { - t.Fatalf("peer LightNode flag is %v, should be %v", pt.bzz.handshakes[id].LightNode, test.lightNode) + if pt.bzz.handshakes[node.ID()].LightNode != test.lightNode { + t.Fatalf("peer LightNode flag is %v, should be %v", pt.bzz.handshakes[node.ID()].LightNode, test.lightNode) } }) }