network: Add Capabilities to Kademlia database (ethersphere#1713)

network: Add capabilities to kademlia database

Author: nolash

cmd/swarm-snapshot/create.go

@@ -60,7 +60,7 @@ func createSnapshot(filename string, nodes int, services []string) (err error) {
 
 	sim := simulation.NewInProc(map[string]simulation.ServiceFunc{
 		"bzz": func(ctx *adapters.ServiceContext, bucket *sync.Map) (node.Service, func(), error) {
-			addr := network.NewAddr(ctx.Config.Node())
+			addr := network.NewBzzAddrFromEnode(ctx.Config.Node())
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			hp := network.NewHiveParams()
 			hp.KeepAliveInterval = time.Duration(200) * time.Millisecond

cmd/swarm/run_test.go

@@ -131,7 +131,10 @@ func newTestCluster(t *testing.T, size int) *testCluster {
 	}
 
 	// connect the nodes together
-	for _, node := range cluster.Nodes {
+	for i, node := range cluster.Nodes {
+		if i == 0 {
+			continue
+		}
 		if err := node.Client.Call(nil, "admin_addPeer", cluster.Nodes[0].Enode); err != nil {
 			t.Fatal(err)
 		}

network/capability/capability.go

@@ -185,3 +185,33 @@ func (c *Capabilities) DecodeRLP(s *rlp.Stream) error {
 
 	return nil
 }
+
+// Match returns true if all bits set in the argument is also set in the receiver
+func (c *Capability) Match(capCompare *Capability) bool {
+	if capCompare == nil || len(c.Cap) != len(capCompare.Cap) {
+		return false
+	}
+	// on the first occurrence of false where query has true we can fail
+	for i, flag := range capCompare.Cap {
+		if flag && !c.Cap[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// Match returns true if all bits set in all capability arguments are also set in the receiver's capabilities
+func (c *Capabilities) Match(capsCompare *Capabilities) bool {
+	for _, capCompare := range capsCompare.Caps {
+
+		// if queryied id doesn't exist in object we can nay right away
+		cap := c.Get(capCompare.Id)
+		if cap == nil {
+			return false
+		}
+		if !cap.Match(capCompare) {
+			return false
+		}
+	}
+	return true
+}

network/capability/capability_test.go

@@ -172,3 +172,59 @@ func TestCapabilitiesRLP(t *testing.T) {
 		t.Fatalf("cap 1 caps not correct, expected %v, got %v", cap2.Cap, cap2Restored.Cap)
 	}
 }
+
+// TestCapabilitiesQuery tests methods for quering capability states
+func TestCapabilitiesQuery(t *testing.T) {
+
+	// Initialize capability
+	caps := NewCapabilities()
+
+	// Register module. Should succeed
+	c1 := NewCapability(1, 3)
+	c1.Set(1)
+	err := caps.Add(c1)
+	if err != nil {
+		t.Fatalf("RegisterCapabilityModule fail: %v", err)
+	}
+
+	c2 := NewCapability(42, 9)
+	c2.Set(2)
+	c2.Set(8)
+	err = caps.Add(c2)
+	if err != nil {
+		t.Fatalf("RegisterCapabilityModule fail: %v", err)
+	}
+
+	capsCompare := NewCapabilities()
+	capCompare := NewCapability(42, 10)
+	capCompare.Set(2)
+	capCompare.Set(8)
+	capsCompare.Add(capCompare)
+	if caps.Match(capsCompare) {
+		t.Fatalf("Expected cCompare with mismatch length to fail; %s != %s", capsCompare, caps)
+	}
+	capsCompare = NewCapabilities()
+	capCompare = NewCapability(42, 9)
+	capCompare.Set(2)
+	capsCompare.Add(capCompare)
+	if !caps.Match(capsCompare) {
+		t.Fatalf("Expected %s to match %s", capsCompare, caps)
+	}
+
+	capCompare = NewCapability(1, 3)
+	capsCompare.Add(capCompare)
+	if !caps.Match(capsCompare) {
+		t.Fatalf("Expected %s to match %s", capsCompare, caps)
+	}
+
+	capCompare.Set(1)
+	if !caps.Match(capsCompare) {
+		t.Fatalf("Expected %s to match %s", capsCompare, caps)
+	}
+
+	capCompare.Set(2)
+	if caps.Match(capsCompare) {
+		t.Fatalf("Expected %s not to match %s", capsCompare, caps)
+	}
+
+}

network/discovery.go

@@ -21,6 +21,8 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethersphere/swarm/log"
 	"github.com/ethersphere/swarm/pot"
 )
 
@@ -95,6 +97,7 @@ func (d *Peer) NotifyPeer(a *BzzAddr, po uint8) {
 	resp := &peersMsg{
 		Peers: []*BzzAddr{a},
 	}
+	log.Warn("notifypeer", "notify", resp)
 	go d.Send(context.TODO(), resp)
 }
 
@@ -127,6 +130,27 @@ type peersMsg struct {
 	Peers []*BzzAddr
 }
 
+// DecodeRLP implements rlp.Decoder interface
+func (p *peersMsg) DecodeRLP(s *rlp.Stream) error {
+	_, err := s.List()
+	if err != nil {
+		return err
+	}
+	_, err = s.List()
+	if err != nil {
+		return err
+	}
+	for {
+		var addr BzzAddr
+		err = s.Decode(&addr)
+		if err != nil {
+			break
+		}
+		p.Peers = append(p.Peers, &addr)
+	}
+	return nil
+}
+
 // String pretty prints a peersMsg
 func (msg peersMsg) String() string {
 	return fmt.Sprintf("%T: %v", msg, msg.Peers)
@@ -140,7 +164,6 @@ func (d *Peer) handlePeersMsg(msg *peersMsg) error {
 	if len(msg.Peers) == 0 {
 		return nil
 	}
-
 	for _, a := range msg.Peers {
 		d.seen(a)
 		NotifyPeer(a, d.kad)

network/discovery_test.go

@@ -17,8 +17,8 @@
 package network
 
 import (
+	"bytes"
 	"crypto/ecdsa"
-	crand "crypto/rand"
 	"encoding/binary"
 	"fmt"
 	"math/rand"
@@ -48,7 +48,7 @@ func TestSubPeersMsg(t *testing.T) {
 	}
 
 	node := s.Nodes[0]
-	raddr := NewAddr(node)
+	raddr := NewBzzAddrFromEnode(node)
 	pp.Register(raddr)
 
 	// start the hive and wait for the connection
@@ -114,18 +114,15 @@ func testInitialPeersMsg(t *testing.T, peerPO, peerDepth int) {
 	connect := func(a pot.Address, po int) (addrs []*BzzAddr) {
 		n := rand.Intn(maxPeersPerPO)
 		for i := 0; i < n; i++ {
-			peer, err := newDiscPeer(pot.RandomAddressAt(a, po))
-			if err != nil {
-				t.Fatal(err)
-			}
+			peer := newDiscPeer(pot.RandomAddressAt(a, po))
 			hive.On(peer)
 			addrs = append(addrs, peer.BzzAddr)
 		}
 		return addrs
 	}
 	register := func(a pot.Address, po int) {
-		addr := pot.RandomAddressAt(a, po)
-		hive.Register(&BzzAddr{OAddr: addr[:]})
+		discPeer := newDiscPeer(a)
+		hive.Register(discPeer.BzzAddr)
 	}
 
 	// generate connected and just registered peers
@@ -238,20 +235,21 @@ func testSortPeers(peers []*BzzAddr) []*BzzAddr {
 // as we are not creating a real node via the protocol,
 // we need to create the discovery peer objects for the additional kademlia
 // nodes manually
-func newDiscPeer(addr pot.Address) (*Peer, error) {
-	pKey, err := ecdsa.GenerateKey(crypto.S256(), crand.Reader)
-	if err != nil {
-		return nil, err
-	}
+func newDiscPeer(addr pot.Address) *Peer {
+
+	// deterministically create enode id
+	// Input to the non-random input buffer is 2xaddress since it munches 256 bits
+	addrSeed := append(addr.Bytes(), addr.Bytes()...)
+	pKey, _ := ecdsa.GenerateKey(crypto.S256(), bytes.NewBuffer(addrSeed))
 	pubKey := pKey.PublicKey
 	nod := enode.NewV4(&pubKey, net.IPv4(127, 0, 0, 1), 0, 0)
-	bzzAddr := &BzzAddr{OAddr: addr[:], UAddr: []byte(nod.String())}
+	bzzAddr := NewBzzAddr(addr[:], []byte(nod.String()))
 	id := nod.ID()
 	p2pPeer := p2p.NewPeer(id, id.String(), nil)
 	return NewPeer(&BzzPeer{
 		Peer:    protocols.NewPeer(p2pPeer, &dummyMsgRW{}, DiscoverySpec),
 		BzzAddr: bzzAddr,
-	}, nil), nil
+	}, nil)
 }
 
 type dummyMsgRW struct{}

network/enr.go

@@ -48,35 +48,26 @@ func (b *ENRAddrEntry) DecodeRLP(s *rlp.Stream) error {
 	return nil
 }
 
-type ENRLightNodeEntry bool
-
-func (b ENRLightNodeEntry) ENRKey() string {
-	return "bzzlightnode"
-}
-
 type ENRBootNodeEntry bool
 
 func (b ENRBootNodeEntry) ENRKey() string {
 	return "bzzbootnode"
 }
 
 func getENRBzzPeer(p *p2p.Peer, rw p2p.MsgReadWriter, spec *protocols.Spec) *BzzPeer {
-	var lightnode ENRLightNodeEntry
 	var bootnode ENRBootNodeEntry
 
 	// retrieve the ENR Record data
 	record := p.Node().Record()
-	record.Load(&lightnode)
 	record.Load(&bootnode)
 
-	// get the address; separate function as long as we need swarm/network:NewAddr() to call it
+	// get the address; separate function as long as we need swarm/network:NewBzzAddrFromEnode() to call it
 	addr := getENRBzzAddr(p.Node())
 
 	// build the peer using the retrieved data
 	return &BzzPeer{
-		Peer:      protocols.NewPeer(p, rw, spec),
-		LightNode: bool(lightnode),
-		BzzAddr:   addr,
+		Peer:    protocols.NewPeer(p, rw, spec),
+		BzzAddr: addr,
 	}
 }
 
@@ -86,8 +77,5 @@ func getENRBzzAddr(nod *enode.Node) *BzzAddr {
 	record := nod.Record()
 	record.Load(&addr)
 
-	return &BzzAddr{
-		OAddr: addr.data,
-		UAddr: []byte(nod.String()),
-	}
+	return NewBzzAddr(addr.data, []byte(nod.String()))
 }

network/hive.go

@@ -209,7 +209,8 @@ func (h *Hive) PeerInfo(id enode.ID) interface{} {
 	if p == nil {
 		return nil
 	}
-	addr := NewAddr(p.Node())
+	// TODO this is bogus, the overlay address will not be correct
+	addr := NewBzzAddrFromEnode(p.Node())
 	return struct {
 		OAddr hexutil.Bytes
 		UAddr hexutil.Bytes

network/hive_test.go

@@ -55,7 +55,7 @@ func TestRegisterAndConnect(t *testing.T) {
 	}
 
 	node := s.Nodes[0]
-	raddr := NewAddr(node)
+	raddr := NewBzzAddrFromEnode(node)
 	pp.Register(raddr)
 
 	// start the hive
@@ -100,15 +100,15 @@ func TestRegisterAndConnect(t *testing.T) {
 	}
 }
 
-// TestHiveStatePersistance creates a protocol simulation with n peers for a node
+// TestHiveStatePersistence creates a protocol simulation with n peers for a node
 // After protocols complete, the node is shut down and the state is stored.
 // Another simulation is created, where 0 nodes are created, but where the stored state is passed
 // The test succeeds if all the peers from the stored state are known after the protocols of the
 // second simulation have completed
 //
 // Actual connectivity is not in scope for this test, as the peers loaded from state are not known to
 // the simulation; the test only verifies that the peers are known to the node
-func TestHiveStatePersistance(t *testing.T) {
+func TestHiveStatePersistence(t *testing.T) {
 	dir, err := ioutil.TempDir("", "hive_test_store")
 	if err != nil {
 		t.Fatal(err)
@@ -152,7 +152,7 @@ func TestHiveStatePersistance(t *testing.T) {
 	h1, cleanup1 := startHive(t, dir)
 	peers := make(map[string]bool)
 	for i := 0; i < peersCount; i++ {
-		raddr := RandomAddr()
+		raddr := RandomBzzAddr()
 		h1.Register(raddr)
 		peers[raddr.String()] = true
 	}