forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/network: port to p2p/enode

Browse source 
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.
This commit is contained in:
Felix Lange 2018-09-11 18:36:35 +02:00
parent 6b32109e5e
commit 86d67eda62
9 changed files with 141 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
swarm/network/discovery.go
View file

@ -87,7 +87,7 @@ func NotifyPeer(p *BzzAddr, k *Kademlia) {
// unless already notified during the connection session // unless already notified during the connection session
func (d *Peer) NotifyPeer(a *BzzAddr, po uint8) { func (d *Peer) NotifyPeer(a *BzzAddr, po uint8) {
// immediately return // 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 return
} }
resp := &peersMsg{ resp := &peersMsg{
@ -161,7 +161,7 @@ func (d *Peer) handleSubPeersMsg(msg *subPeersMsg) error {
d.setDepth(msg.Depth) d.setDepth(msg.Depth)
var peers []*BzzAddr var peers []*BzzAddr
d.kad.EachConn(d.Over(), 255, func(p *Peer, po int, isproxbin bool) bool { 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 return false
} }
if !d.seen(p.BzzAddr) { if !d.seen(p.BzzAddr) {

6
swarm/network/discovery_test.go
View file

@ -31,8 +31,8 @@ func TestDiscovery(t *testing.T) {
params := NewHiveParams() params := NewHiveParams()
s, pp := newHiveTester(t, params, 1, nil) s, pp := newHiveTester(t, params, 1, nil)
id := s.IDs[0] node := s.Nodes[0]
raddr := NewAddrFromNodeID(id) raddr := NewAddr(node)
pp.Register(raddr) pp.Register(raddr)
// start the hive and wait for the connection // start the hive and wait for the connection
@ -46,7 +46,7 @@ func TestDiscovery(t *testing.T) {
{ {
Code: 1, Code: 1,
Msg: &subPeersMsg{Depth: 0}, Msg: &subPeersMsg{Depth: 0},
Peer: id, Peer: node.ID(),
}, },
}, },
}) })

38
swarm/network/fetcher.go
View file

@ -22,7 +22,7 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/log" "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" "github.com/ethereum/go-ethereum/swarm/storage"
) )
@ -32,7 +32,7 @@ var searchTimeout = 1 * time.Second
// Also used in stream delivery. // Also used in stream delivery.
var RequestTimeout = 10 * time.Second 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 // 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: // 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. // Fetcher self destroys itself after it is completed.
// TODO: cancel all forward requests after termination // TODO: cancel all forward requests after termination
type Fetcher struct { type Fetcher struct {
protoRequestFunc RequestFunc // request function fetcher calls to issue retrieve request for a chunk protoRequestFunc RequestFunc // request function fetcher calls to issue retrieve request for a chunk
addr storage.Address // the address of the chunk to be fetched addr storage.Address // the address of the chunk to be fetched
offerC chan *discover.NodeID // channel of sources (peer node id strings) offerC chan *enode.ID // channel of sources (peer node id strings)
requestC chan struct{} requestC chan struct{}
skipCheck bool skipCheck bool
} }
type Request struct { type Request struct {
Addr storage.Address // chunk address Addr storage.Address // chunk address
Source *discover.NodeID // nodeID of peer to request from (can be nil) Source *enode.ID // nodeID of peer to request from (can be nil)
SkipCheck bool // whether to offer the chunk first or deliver directly 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) 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 // 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{ return &Fetcher{
addr: addr, addr: addr,
protoRequestFunc: rf, protoRequestFunc: rf,
offerC: make(chan *discover.NodeID), offerC: make(chan *enode.ID),
requestC: make(chan struct{}), requestC: make(chan struct{}),
skipCheck: skipCheck, 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. // 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 // First we need to have this select to make sure that we return if context is done
select { select {
case <-ctx.Done(): 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 // it keeps the Fetcher alive within the lifecycle of the passed context
func (f *Fetcher) run(ctx context.Context, peers *sync.Map) { func (f *Fetcher) run(ctx context.Context, peers *sync.Map) {
var ( var (
doRequest bool // determines if retrieval is initiated in the current iteration doRequest bool // determines if retrieval is initiated in the current iteration
wait *time.Timer // timer for search timeout wait *time.Timer // timer for search timeout
waitC <-chan time.Time // timer channel waitC <-chan time.Time // timer channel
sources []*discover.NodeID // known sources, ie. peers that offered the chunk sources []*enode.ID // known sources, ie. peers that offered the chunk
requested bool // true if the chunk was actually requested 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 // loop that keeps the fetching process alive
// after every request a timer is set. If this goes off we request again from another peer // 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 added to the set of peers to skip
// * the peer's address is removed from prospective sources, and // * 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) // * 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 i int
var sourceID *discover.NodeID var sourceID *enode.ID
var quit chan struct{} var quit chan struct{}
req := &Request{ req := &Request{

18
swarm/network/fetcher_test.go
View file

@ -22,11 +22,11 @@ import (
"testing" "testing"
"time" "time"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/enode"
) )
var requestedPeerID = discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") var requestedPeerID = enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8")
var sourcePeerID = discover.MustHexID("2dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") var sourcePeerID = enode.HexID("99d8594b52298567d2ca3f4c441a5ba0140ee9245e26460d01102a52773c73b9")
// mockRequester pushes every request to the requestC channel when its doRequest function is called // mockRequester pushes every request to the requestC channel when its doRequest function is called
type mockRequester struct { 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) waitTime := time.Duration(0)
if m.ctr < len(m.waitTimes) { if m.ctr < len(m.waitTimes) {
waitTime = m.waitTimes[m.ctr] waitTime = m.waitTimes[m.ctr]
@ -389,9 +389,9 @@ func TestFetcherRequestQuitRetriesRequest(t *testing.T) {
// and not skip unknown one. // and not skip unknown one.
func TestRequestSkipPeer(t *testing.T) { func TestRequestSkipPeer(t *testing.T) {
addr := make([]byte, 32) addr := make([]byte, 32)
peers := []discover.NodeID{ peers := []enode.ID{
discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"), enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8"),
discover.MustHexID("2dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"), enode.HexID("99d8594b52298567d2ca3f4c441a5ba0140ee9245e26460d01102a52773c73b9"),
} }
peersToSkip := new(sync.Map) peersToSkip := new(sync.Map)
@ -411,7 +411,7 @@ func TestRequestSkipPeer(t *testing.T) {
// after RequestTimeout has passed. // after RequestTimeout has passed.
func TestRequestSkipPeerExpired(t *testing.T) { func TestRequestSkipPeerExpired(t *testing.T) {
addr := make([]byte, 32) addr := make([]byte, 32)
peer := discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") peer := enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8")
// set RequestTimeout to a low value and reset it after the test // set RequestTimeout to a low value and reset it after the test
defer func(t time.Duration) { RequestTimeout = t }(RequestTimeout) 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. // by value to peersToSkip map is not time.Duration.
func TestRequestSkipPeerPermanent(t *testing.T) { func TestRequestSkipPeerPermanent(t *testing.T) {
addr := make([]byte, 32) addr := make([]byte, 32)
peer := discover.MustHexID("1dd9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439") peer := enode.HexID("3431c3939e1ee2a6345e976a8234f9870152d64879f30bc272a074f6859e75e8")
// set RequestTimeout to a low value and reset it after the test // set RequestTimeout to a low value and reset it after the test
defer func(t time.Duration) { RequestTimeout = t }(RequestTimeout) defer func(t time.Duration) { RequestTimeout = t }(RequestTimeout)

40
swarm/network/hive.go
View file

@ -23,7 +23,7 @@ import (
"github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/p2p" "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/log"
"github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/state"
) )
@ -56,12 +56,13 @@ func NewHiveParams() *HiveParams {
// Hive manages network connections of the swarm node // Hive manages network connections of the swarm node
type Hive struct { type Hive struct {
*HiveParams // settings *HiveParams // settings
*Kademlia // the overlay connectiviy driver *Kademlia // the overlay connectiviy driver
Store state.Store // storage interface to save peers across sessions Store state.Store // storage interface to save peers across sessions
addPeer func(*discover.Node) // server callback to connect to a peer addPeer func(*enode.Node) // server callback to connect to a peer
// bookkeeping // bookkeeping
lock sync.Mutex lock sync.Mutex
peers map[enode.ID]*BzzPeer
ticker *time.Ticker ticker *time.Ticker
} }
@ -74,6 +75,7 @@ func NewHive(params *HiveParams, kad *Kademlia, store state.Store) *Hive {
HiveParams: params, HiveParams: params,
Kademlia: kad, Kademlia: kad,
Store: store, 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])) 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 { 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)) log.Warn(fmt.Sprintf("%08x unable to connect to bee %08x: invalid node URL: %v", h.BaseAddr()[:4], addr.Address()[:4], err))
continue continue
@ -149,6 +151,9 @@ func (h *Hive) connect() {
// Run protocol run function // Run protocol run function
func (h *Hive) Run(p *BzzPeer) error { func (h *Hive) Run(p *BzzPeer) error {
h.trackPeer(p)
defer h.untrackPeer(p)
dp := NewPeer(p, h.Kademlia) dp := NewPeer(p, h.Kademlia)
depth, changed := h.On(dp) depth, changed := h.On(dp)
// if we want discovery, advertise change of depth // if we want discovery, advertise change of depth
@ -166,6 +171,18 @@ func (h *Hive) Run(p *BzzPeer) error {
return dp.Run(dp.HandleMsg) 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 // NodeInfo function is used by the p2p.server RPC interface to display
// protocol specific node information // protocol specific node information
func (h *Hive) NodeInfo() interface{} { 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 // PeerInfo function is used by the p2p.server RPC interface to display
// protocol specific information any connected peer referred to by their NodeID // protocol specific information any connected peer referred to by their NodeID
func (h *Hive) PeerInfo(id discover.NodeID) interface{} { func (h *Hive) PeerInfo(id enode.ID) interface{} {
addr := NewAddrFromNodeID(id) h.lock.Lock()
p := h.peers[id]
h.lock.Unlock()
if p == nil {
return nil
}
addr := NewAddr(p.Node())
return struct { return struct {
OAddr hexutil.Bytes OAddr hexutil.Bytes
UAddr hexutil.Bytes UAddr hexutil.Bytes

17
swarm/network/hive_test.go
View file

@ -39,8 +39,8 @@ func TestRegisterAndConnect(t *testing.T) {
params := NewHiveParams() params := NewHiveParams()
s, pp := newHiveTester(t, params, 1, nil) s, pp := newHiveTester(t, params, 1, nil)
id := s.IDs[0] node := s.Nodes[0]
raddr := NewAddrFromNodeID(id) raddr := NewAddr(node)
pp.Register(raddr) pp.Register(raddr)
// start the hive and wait for the connection // start the hive and wait for the connection
@ -51,7 +51,7 @@ func TestRegisterAndConnect(t *testing.T) {
defer pp.Stop() defer pp.Stop()
// retrieve and broadcast // retrieve and broadcast
err = s.TestDisconnected(&p2ptest.Disconnect{ err = s.TestDisconnected(&p2ptest.Disconnect{
Peer: s.IDs[0], Peer: s.Nodes[0].ID(),
Error: nil, Error: nil,
}) })
@ -75,8 +75,8 @@ func TestHiveStatePersistance(t *testing.T) {
s, pp := newHiveTester(t, params, 5, store) s, pp := newHiveTester(t, params, 5, store)
peers := make(map[string]bool) peers := make(map[string]bool)
for _, id := range s.IDs { for _, node := range s.Nodes {
raddr := NewAddrFromNodeID(id) raddr := NewAddr(node)
pp.Register(raddr) pp.Register(raddr)
peers[raddr.String()] = true peers[raddr.String()] = true
} }
@ -102,7 +102,10 @@ func TestHiveStatePersistance(t *testing.T) {
i++ i++
return true return true
}) })
if len(peers) != 0 || i != 5 { if i != 5 {
t.Fatalf("invalid peers loaded") 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)
} }
} }

23
swarm/network/networkid_test.go
View file

@ -29,7 +29,7 @@ import (
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p" "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"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
@ -38,8 +38,8 @@ import (
var ( var (
currentNetworkID int currentNetworkID int
cnt int cnt int
nodeMap map[int][]discover.NodeID nodeMap map[int][]enode.ID
kademlias map[discover.NodeID]*Kademlia kademlias map[enode.ID]*Kademlia
) )
const ( const (
@ -70,7 +70,7 @@ func TestNetworkID(t *testing.T) {
//arbitrarily set the number of nodes. It could be any number //arbitrarily set the number of nodes. It could be any number
numNodes := 24 numNodes := 24
//the nodeMap maps all nodes (slice value) with the same network ID (key) //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 //set up the network and connect nodes
net, err := setupNetwork(numNodes) net, err := setupNetwork(numNodes)
if err != nil { if err != nil {
@ -95,7 +95,7 @@ func TestNetworkID(t *testing.T) {
kademlias[node].EachAddr(nil, 0, func(addr *BzzAddr, _ int, _ bool) bool { kademlias[node].EachAddr(nil, 0, func(addr *BzzAddr, _ int, _ bool) bool {
found := false found := false
for _, nd := range netIDGroup { for _, nd := range netIDGroup {
p := ToOverlayAddr(nd.Bytes()) p := nd.Bytes()
if bytes.Equal(p, addr.Address()) { if bytes.Equal(p, addr.Address()) {
found = true found = true
} }
@ -183,12 +183,11 @@ func setupNetwork(numnodes int) (net *simulations.Network, err error) {
} }
func newServices() adapters.Services { func newServices() adapters.Services {
kademlias = make(map[discover.NodeID]*Kademlia) kademlias = make(map[enode.ID]*Kademlia)
kademlia := func(id discover.NodeID) *Kademlia { kademlia := func(id enode.ID) *Kademlia {
if k, ok := kademlias[id]; ok { if k, ok := kademlias[id]; ok {
return k return k
} }
addr := NewAddrFromNodeID(id)
params := NewKadParams() params := NewKadParams()
params.MinProxBinSize = 2 params.MinProxBinSize = 2
params.MaxBinSize = 3 params.MaxBinSize = 3
@ -196,19 +195,19 @@ func newServices() adapters.Services {
params.MaxRetries = 1000 params.MaxRetries = 1000
params.RetryExponent = 2 params.RetryExponent = 2
params.RetryInterval = 1000000 params.RetryInterval = 1000000
kademlias[id] = NewKademlia(addr.Over(), params) kademlias[id] = NewKademlia(id[:], params)
return kademlias[id] return kademlias[id]
} }
return adapters.Services{ return adapters.Services{
"bzz": func(ctx *adapters.ServiceContext) (node.Service, error) { "bzz": func(ctx *adapters.ServiceContext) (node.Service, error) {
addr := NewAddrFromNodeID(ctx.Config.ID) addr := NewAddr(ctx.Config.Node())
hp := NewHiveParams() hp := NewHiveParams()
hp.Discovery = false hp.Discovery = false
cnt++ cnt++
//assign the network ID //assign the network ID
currentNetworkID = cnt % NumberOfNets currentNetworkID = cnt % NumberOfNets
if ok := nodeMap[currentNetworkID]; ok == nil { 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 //add this node to the group sharing the same network ID
nodeMap[currentNetworkID] = append(nodeMap[currentNetworkID], ctx.Config.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) events := make(chan *p2p.PeerEvent)
sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents")
if err != nil { if err != nil {

85
swarm/network/protocol.go
View file

@ -26,7 +26,7 @@ import (
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p" "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/p2p/protocols"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/log"
@ -78,7 +78,7 @@ type Bzz struct {
LightNode bool LightNode bool
localAddr *BzzAddr localAddr *BzzAddr
mtx sync.Mutex mtx sync.Mutex
handshakes map[discover.NodeID]*HandshakeMsg handshakes map[enode.ID]*HandshakeMsg
streamerSpec *protocols.Spec streamerSpec *protocols.Spec
streamerRun func(*BzzPeer) error streamerRun func(*BzzPeer) error
} }
@ -94,7 +94,7 @@ func NewBzz(config *BzzConfig, kad *Kademlia, store state.Store, streamerSpec *p
NetworkID: config.NetworkID, NetworkID: config.NetworkID,
LightNode: config.LightNode, LightNode: config.LightNode,
localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr}, localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
handshakes: make(map[discover.NodeID]*HandshakeMsg), handshakes: make(map[enode.ID]*HandshakeMsg),
streamerRun: streamerRun, streamerRun: streamerRun,
streamerSpec: streamerSpec, 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 // the handshake has succeeded so construct the BzzPeer and run the protocol
peer := &BzzPeer{ peer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, spec), Peer: protocols.NewPeer(p, rw, spec),
localAddr: b.localAddr,
BzzAddr: handshake.peerAddr, BzzAddr: handshake.peerAddr,
lastActive: time.Now(), lastActive: time.Now(),
LightNode: handshake.LightNode, 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 { func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error {
handshake, _ := b.GetHandshake(p.ID()) handshake, _ := b.GetHandshake(p.ID())
if !<-handshake.init { 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) close(handshake.init)
defer b.removeHandshake(p.ID()) defer b.removeHandshake(p.ID())
peer := protocols.NewPeer(p, rw, BzzSpec) peer := protocols.NewPeer(p, rw, BzzSpec)
err := b.performHandshake(peer, handshake) err := b.performHandshake(peer, handshake)
if err != nil { 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 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 // implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer
type BzzPeer struct { type BzzPeer struct {
*protocols.Peer // represents the connection for online peers *protocols.Peer // represents the connection for online peers
localAddr *BzzAddr // local Peers address
*BzzAddr // remote address -> implements Addr interface = protocols.Peer *BzzAddr // remote address -> implements Addr interface = protocols.Peer
lastActive time.Time // time is updated whenever mutexes are releasing lastActive time.Time // time is updated whenever mutexes are releasing
LightNode bool LightNode bool
} }
func NewBzzPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer { func NewBzzPeer(p *protocols.Peer) *BzzPeer {
return &BzzPeer{ return &BzzPeer{Peer: p, BzzAddr: NewAddr(p.Node())}
Peer: p,
localAddr: addr,
BzzAddr: NewAddrFromNodeID(p.ID()),
}
} }
// LastActive returns the time the peer was last active // LastActive returns the time the peer was last active
@ -261,6 +255,14 @@ func (p *BzzPeer) LastActive() time.Time {
return p.lastActive 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 Handshake
@ -301,14 +303,14 @@ func (b *Bzz) checkHandshake(hs interface{}) error {
// removeHandshake removes handshake for peer with peerID // removeHandshake removes handshake for peer with peerID
// from the bzz handshake store // from the bzz handshake store
func (b *Bzz) removeHandshake(peerID discover.NodeID) { func (b *Bzz) removeHandshake(peerID enode.ID) {
b.mtx.Lock() b.mtx.Lock()
defer b.mtx.Unlock() defer b.mtx.Unlock()
delete(b.handshakes, peerID) delete(b.handshakes, peerID)
} }
// GetHandshake returns the bzz handhake that the remote peer with peerID sent // 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() b.mtx.Lock()
defer b.mtx.Unlock() defer b.mtx.Unlock()
handshake, found := b.handshakes[peerID] handshake, found := b.handshakes[peerID]
@ -336,24 +338,28 @@ type BzzAddr struct {
UAddr []byte 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 { func (a *BzzAddr) Address() []byte {
return a.OAddr return a.OAddr
} }
// Over returns the overlay address // Over returns the overlay address.
func (a *BzzAddr) Over() []byte { func (a *BzzAddr) Over() []byte {
return a.OAddr return a.OAddr
} }
// Under returns the underlay address // Under returns the underlay address.
func (a *BzzAddr) Under() []byte { func (a *BzzAddr) Under() []byte {
return a.UAddr return a.UAddr
} }
// ID returns the nodeID from the underlay enode address // ID returns the node identifier in the underlay.
func (a *BzzAddr) ID() discover.NodeID { func (a *BzzAddr) ID() enode.ID {
return discover.MustParseNode(string(a.UAddr)).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 // Update updates the underlay address of a peer record
@ -372,38 +378,11 @@ func RandomAddr() *BzzAddr {
if err != nil { if err != nil {
panic("unable to generate key") panic("unable to generate key")
} }
pubkey := crypto.FromECDSAPub(&key.PublicKey) node := enode.NewV4(&key.PublicKey, net.IP{127, 0, 0, 1}, 30303, 30303)
var id discover.NodeID return NewAddr(node)
copy(id[:], pubkey[1:])
return NewAddrFromNodeID(id)
} }
// NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID // NewAddr constucts a BzzAddr from a node record.
func NewNodeIDFromAddr(addr *BzzAddr) discover.NodeID { func NewAddr(node *enode.Node) *BzzAddr {
log.Info(fmt.Sprintf("uaddr=%s", string(addr.UAddr))) return &BzzAddr{OAddr: node.ID().Bytes(), UAddr: []byte(node.String())}
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)
} }

50
swarm/network/protocol_test.go
View file

@ -25,7 +25,7 @@ import (
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p" "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/p2p/protocols"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing" p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
) )
@ -71,7 +71,7 @@ func (t *testStore) Save(key string, v []byte) error {
return nil 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{ 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 { protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return srv(&BzzPeer{ return srv(&BzzPeer{Peer: protocols.NewPeer(p, rw, spec), BzzAddr: NewAddr(p.Node())})
Peer: protocols.NewPeer(p, rw, spec),
localAddr: addr,
BzzAddr: NewAddrFromNodeID(p.ID()),
})
} }
s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, protocol) s := p2ptest.NewProtocolTester(t, addr.ID(), n, protocol)
for _, id := range s.IDs { for _, node := range s.Nodes {
cs[id.String()] = make(chan bool) cs[node.ID().String()] = make(chan bool)
} }
return &bzzTester{ 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 { func newBzzHandshakeTester(t *testing.T, n int, addr *BzzAddr, lightNode bool) *bzzTester {
bzz := newBzz(addr, lightNode) bzz := newBzz(addr, lightNode)
pt := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, bzz.runBzz) pt := p2ptest.NewProtocolTester(t, addr.ID(), n, bzz.runBzz)
return &bzzTester{ return &bzzTester{
addr: addr, addr: addr,
@ -161,14 +157,14 @@ func newBzzHandshakeTester(t *testing.T, n int, addr *BzzAddr, lightNode bool) *
// should test handshakes in one exchange? parallelisation // should test handshakes in one exchange? parallelisation
func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptest.Disconnect) error { func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptest.Disconnect) error {
var peers []discover.NodeID var peers []enode.ID
id := NewNodeIDFromAddr(rhs.Addr) id := rhs.Addr.ID()
if len(disconnects) > 0 { if len(disconnects) > 0 {
for _, d := range disconnects { for _, d := range disconnects {
peers = append(peers, d.Peer) peers = append(peers, d.Peer)
} }
} else { } else {
peers = []discover.NodeID{id} peers = []enode.ID{id}
} }
if err := s.TestExchanges(HandshakeMsgExchange(lhs, rhs, id)...); err != nil { 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 // If we don't expect disconnect, ensure peers remain connected
err := s.TestDisconnected(&p2ptest.Disconnect{ err := s.TestDisconnected(&p2ptest.Disconnect{
Peer: s.IDs[0], Peer: s.Nodes[0].ID(),
Error: nil, Error: nil,
}) })
@ -209,12 +205,12 @@ func TestBzzHandshakeNetworkIDMismatch(t *testing.T) {
lightNode := false lightNode := false
addr := RandomAddr() addr := RandomAddr()
s := newBzzHandshakeTester(t, 1, addr, lightNode) s := newBzzHandshakeTester(t, 1, addr, lightNode)
id := s.IDs[0] node := s.Nodes[0]
err := s.testHandshake( err := s.testHandshake(
correctBzzHandshake(addr, lightNode), correctBzzHandshake(addr, lightNode),
&HandshakeMsg{Version: TestProtocolVersion, NetworkID: 321, Addr: NewAddrFromNodeID(id)}, &HandshakeMsg{Version: TestProtocolVersion, NetworkID: 321, Addr: NewAddr(node)},
&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): network id mismatch 321 (!= 3)")}, &p2ptest.Disconnect{Peer: node.ID(), Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): network id mismatch 321 (!= 3)")},
) )
if err != nil { if err != nil {
@ -226,12 +222,12 @@ func TestBzzHandshakeVersionMismatch(t *testing.T) {
lightNode := false lightNode := false
addr := RandomAddr() addr := RandomAddr()
s := newBzzHandshakeTester(t, 1, addr, lightNode) s := newBzzHandshakeTester(t, 1, addr, lightNode)
id := s.IDs[0] node := s.Nodes[0]
err := s.testHandshake( err := s.testHandshake(
correctBzzHandshake(addr, lightNode), correctBzzHandshake(addr, lightNode),
&HandshakeMsg{Version: 0, NetworkID: TestProtocolNetworkID, Addr: NewAddrFromNodeID(id)}, &HandshakeMsg{Version: 0, NetworkID: TestProtocolNetworkID, Addr: NewAddr(node)},
&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): version mismatch 0 (!= %d)", TestProtocolVersion)}, &p2ptest.Disconnect{Peer: node.ID(), Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): version mismatch 0 (!= %d)", TestProtocolVersion)},
) )
if err != nil { if err != nil {
@ -243,11 +239,11 @@ func TestBzzHandshakeSuccess(t *testing.T) {
lightNode := false lightNode := false
addr := RandomAddr() addr := RandomAddr()
s := newBzzHandshakeTester(t, 1, addr, lightNode) s := newBzzHandshakeTester(t, 1, addr, lightNode)
id := s.IDs[0] node := s.Nodes[0]
err := s.testHandshake( err := s.testHandshake(
correctBzzHandshake(addr, lightNode), correctBzzHandshake(addr, lightNode),
&HandshakeMsg{Version: TestProtocolVersion, NetworkID: TestProtocolNetworkID, Addr: NewAddrFromNodeID(id)}, &HandshakeMsg{Version: TestProtocolVersion, NetworkID: TestProtocolNetworkID, Addr: NewAddr(node)},
) )
if err != nil { if err != nil {
@ -268,8 +264,8 @@ func TestBzzHandshakeLightNode(t *testing.T) {
t.Run(test.name, func(t *testing.T) { t.Run(test.name, func(t *testing.T) {
randomAddr := RandomAddr() randomAddr := RandomAddr()
pt := newBzzHandshakeTester(t, 1, randomAddr, false) pt := newBzzHandshakeTester(t, 1, randomAddr, false)
id := pt.IDs[0] node := pt.Nodes[0]
addr := NewAddrFromNodeID(id) addr := NewAddr(node)
err := pt.testHandshake( err := pt.testHandshake(
correctBzzHandshake(randomAddr, false), correctBzzHandshake(randomAddr, false),
@ -280,8 +276,8 @@ func TestBzzHandshakeLightNode(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
if 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[id].LightNode, test.lightNode) t.Fatalf("peer LightNode flag is %v, should be %v", pt.bzz.handshakes[node.ID()].LightNode, test.lightNode)
} }
}) })
} }