Merge pull request #204 from spacemeshos/p2p-port-on-hs

BUG #203

p2p/connectionpool/connectionpool.go

@@ -17,7 +17,7 @@ type dialResult struct {
 
 type networker interface {
 	Dial(address string, remotePublicKey crypto.PublicKey) (net.Connection, error) // Connect to a remote node. Can send when no error.
-	SubscribeOnNewRemoteConnections() chan net.Connection
+	SubscribeOnNewRemoteConnections() chan net.NewConnectionEvent
 	NetworkID() int8
 	ClosingConnections() chan net.Connection
 	Logger() *logging.Logger
@@ -37,13 +37,12 @@ type ConnectionPool struct {
 	dialWait    sync.WaitGroup
 	shutdown    bool
 
-	newRemoteConn chan net.Connection
+	newRemoteConn chan net.NewConnectionEvent
 	teardown      chan struct{}
 }
 
 // NewConnectionPool creates new ConnectionPool
 func NewConnectionPool(network networker, lPub crypto.PublicKey) *ConnectionPool {
-	connC := network.SubscribeOnNewRemoteConnections()
 	cPool := &ConnectionPool{
 		localPub:      lPub,
 		net:           network,
@@ -53,7 +52,7 @@ func NewConnectionPool(network networker, lPub crypto.PublicKey) *ConnectionPool
 		pendMutex:     sync.Mutex{},
 		dialWait:      sync.WaitGroup{},
 		shutdown:      false,
-		newRemoteConn: connC,
+		newRemoteConn: network.SubscribeOnNewRemoteConnections(),
 		teardown:      make(chan struct{}),
 	}
 	go cPool.beginEventProcessing()
@@ -205,8 +204,8 @@ func (cp *ConnectionPool) beginEventProcessing() {
 Loop:
 	for {
 		select {
-		case conn := <-cp.newRemoteConn:
-			cp.handleNewConnection(conn.RemotePublicKey(), conn, net.Remote)
+		case nce := <-cp.newRemoteConn:
+			cp.handleNewConnection(nce.Conn.RemotePublicKey(), nce.Conn, net.Remote)
 
 		case conn := <-cp.net.ClosingConnections():
 			cp.handleClosedConnection(conn)

p2p/connectionpool/connectionpool_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/spacemeshos/go-spacemesh/crypto"
 	"github.com/spacemeshos/go-spacemesh/p2p/net"
+	"github.com/spacemeshos/go-spacemesh/p2p/node"
 	"github.com/stretchr/testify/assert"
 	"math/rand"
 	"testing"
@@ -112,7 +113,7 @@ func TestRemoteConnectionWithNoConnection(t *testing.T) {
 	cPool := NewConnectionPool(n, generatePublicKey())
 	rConn := net.NewConnectionMock(remotePub)
 	rConn.SetSession(net.NewSessionMock([]byte("aaa")))
-	cPool.newRemoteConn <- rConn
+	cPool.newRemoteConn <- net.NewConnectionEvent{rConn, node.EmptyNode}
 	time.Sleep(50 * time.Millisecond)
 	conn, err := cPool.GetConnection(addr, remotePub)
 	assert.Equal(t, remotePub.String(), conn.RemotePublicKey().String())
@@ -133,7 +134,7 @@ func TestRemoteConnectionWithExistingConnection(t *testing.T) {
 	lConn, _ := cPool.GetConnection(addr, remotePub)
 	rConn := net.NewConnectionMock(remotePub)
 	rConn.SetSession(net.NewSessionMock([]byte("111")))
-	cPool.newRemoteConn <- rConn
+	cPool.newRemoteConn <- net.NewConnectionEvent{rConn, node.EmptyNode}
 	time.Sleep(20 * time.Millisecond)
 	assert.Equal(t, remotePub.String(), lConn.RemotePublicKey().String())
 	assert.Equal(t, int32(1), n.DialCount())
@@ -147,7 +148,7 @@ func TestRemoteConnectionWithExistingConnection(t *testing.T) {
 	lConn, _ = cPool.GetConnection(addr, remotePub)
 	rConn = net.NewConnectionMock(remotePub)
 	rConn.SetSession(net.NewSessionMock([]byte("110")))
-	cPool.newRemoteConn <- rConn
+	cPool.newRemoteConn <- net.NewConnectionEvent{rConn, node.EmptyNode}
 	time.Sleep(20 * time.Millisecond)
 	assert.Equal(t, remotePub.String(), lConn.RemotePublicKey().String())
 	assert.Equal(t, int32(2), n.DialCount())
@@ -267,7 +268,7 @@ func TestRandom(t *testing.T) {
 				sID := make([]byte, 4)
 				rand.Read(sID)
 				rConn.SetSession(net.NewSessionMock(sID))
-				cPool.newRemoteConn <- rConn
+				cPool.newRemoteConn <- net.NewConnectionEvent{rConn, node.EmptyNode}
 			}()
 		} else if r == 1 {
 			go func() {

p2p/dht/dht_test.go

@@ -62,11 +62,14 @@ func TestDHT_Update(t *testing.T) {
 
 	assert.True(t, size > config.DefaultConfig().SwarmConfig.RoutingTableBucketSize, "Routing table should be at least as big as bucket size")
 
-	looked, err := dht.Lookup(randnode.PublicKey().String())
+	lastnode := evenmorenodes[0]
 
-	assert.NoError(t, err, "error finding existing node")
+	looked, err := dht.Lookup(lastnode.PublicKey().String())
 
-	assert.True(t, looked.String() == randnode.String(), "didnt find the same node")
+	assert.NoError(t, err, "error finding existing node ")
+
+	assert.Equal(t, looked.String(), lastnode.String(), "didnt find the same node")
+	assert.Equal(t, looked.Address(), lastnode.Address(), "didnt find the same node")
 
 }
 
@@ -120,47 +123,41 @@ func TestDHT_Lookup2(t *testing.T) {
 
 }
 
+func simNodeWithDHT(t *testing.T, sc config.SwarmConfig, sim *simulator.Simulator) (*simulator.Node, DHT) {
+	ln, _ := node.GenerateTestNode(t)
+	n := sim.NewNodeFrom(ln.Node)
+	dht := New(ln, sc, n)
+	n.AttachDHT(dht)
+
+	return n, dht
+}
+
+func bootAndWait(t *testing.T, dht DHT, errchan chan error) {
+	err := dht.Bootstrap()
+	errchan <- err
+}
+
 func TestDHT_Bootstrap(t *testing.T) {
 	// Create a bootstrap node
-	ln, _ := node.GenerateTestNode(t)
-	cfg := config.DefaultConfig()
 	sim := simulator.New()
-	n1 := sim.NewNodeFrom(ln.Node)
-
-	dht := New(ln, cfg.SwarmConfig, n1)
+	bn, _ := simNodeWithDHT(t, config.DefaultConfig().SwarmConfig, sim)
 
 	// config for other nodes
 	cfg2 := config.DefaultConfig()
 	cfg2.SwarmConfig.RandomConnections = 2 // min numbers of peers to succeed in bootstrap
-	cfg2.SwarmConfig.BootstrapNodes = []string{node.StringFromNode(dht.local.Node)}
+	cfg2.SwarmConfig.BootstrapNodes = []string{node.StringFromNode(bn.Node)}
 
 	booted := make(chan error)
 
-	// Boot 3 more nodes
-	ln2, _ := node.GenerateTestNode(t)
-	n2 := sim.NewNodeFrom(ln2.Node)
-	dht2 := New(ln2, cfg2.SwarmConfig, n2)
-
-	ln3, _ := node.GenerateTestNode(t)
-	n3 := sim.NewNodeFrom(ln3.Node)
-	dht3 := New(ln3, cfg2.SwarmConfig, n3)
-
-	ln4, _ := node.GenerateTestNode(t)
-	n4 := sim.NewNodeFrom(ln4.Node)
-	dht4 := New(ln4, cfg2.SwarmConfig, n4)
-
-	go func() {
-		err2 := dht2.Bootstrap()
-		booted <- err2
-	}()
-	go func() {
-		err3 := dht3.Bootstrap()
-		booted <- err3
-	}()
-	go func() {
-		err4 := dht4.Bootstrap()
-		booted <- err4
-	}()
+	// boot 3 more dhts
+
+	_, dht2 := simNodeWithDHT(t, cfg2.SwarmConfig, sim)
+	_, dht3 := simNodeWithDHT(t, cfg2.SwarmConfig, sim)
+	_, dht4 := simNodeWithDHT(t, cfg2.SwarmConfig, sim)
+
+	go bootAndWait(t, dht2, booted)
+	go bootAndWait(t, dht3, booted)
+	go bootAndWait(t, dht4, booted)
 
 	// Collect errors
 	err := <-booted
@@ -181,29 +178,19 @@ func TestDHT_Bootstrap2(t *testing.T) {
 	sim := simulator.New()
 
 	// Create a bootstrap node
-	bs, _ := node.GenerateTestNode(t)
 	cfg := config.DefaultConfig()
-	bsn := sim.NewNodeFrom(bs.Node)
-
-	dht := New(bs, cfg.SwarmConfig, bsn)
+	bn, _ := simNodeWithDHT(t, cfg.SwarmConfig, sim)
 
 	// config for other nodes
 	cfg2 := config.DefaultConfig()
 	cfg2.SwarmConfig.RandomConnections = minToBoot // min numbers of peers to succeed in bootstrap
-	cfg2.SwarmConfig.BootstrapNodes = []string{node.StringFromNode(dht.local.Node)}
+	cfg2.SwarmConfig.BootstrapNodes = []string{node.StringFromNode(bn.Node)}
 
 	booted := make(chan error)
 
-	dhts := make([]*KadDHT, nodesNum)
-
 	for i := 0; i < nodesNum; i++ {
-		lln, _ := node.GenerateTestNode(t)
-		nn := sim.NewNodeFrom(lln.Node)
-
-		d := New(lln, cfg2.SwarmConfig, nn)
-
-		dhts[i] = d
-		go func(d *KadDHT) { err := d.Bootstrap(); booted <- err }(d)
+		_, d := simNodeWithDHT(t, cfg2.SwarmConfig, sim)
+		go bootAndWait(t, d, booted)
 	}
 
 	timer := time.NewTimer(timeout)

p2p/dht/findnode.go

@@ -154,8 +154,6 @@ func (p *findNodeProtocol) readLoop() {
 				return
 			}
 
-			p.rt.Update(msg.Sender())
-
 			if headers.Req {
 				p.handleIncomingRequest(msg.Sender().PublicKey(), headers.ReqID, headers.Payload)
 				return

p2p/dht/table.go

@@ -245,6 +245,12 @@ func (rt *routingTableImpl) update(p node.Node) {
 		return
 	}
 
+	if p.Address() == "" {
+		rt.log.Errorf("Updated non-existing peer without an address pubkey: %v", p.PublicKey().String())
+		return
+	}
+	// this is a new node.
+
 	// todo: consider connection metrics
 	if bucket.Len() >= rt.bucketsize { // bucket overflows
 		// TODO: if bucket is full ping oldest node and replace if it fails to answer

p2p/net/conn.go

@@ -152,12 +152,12 @@ func (c *FormattedConnection) Send(m []byte) error {
 // Close closes the connection (implements io.Closer). It is go safe.
 func (c *FormattedConnection) Close() {
 	c.closeOnce.Do(func() {
-		atomic.AddInt32(&c.closed, int32(1))
+		atomic.AddInt32(&c.closed, 1)
 		c.closeChan <- struct{}{}
 	})
 }
 
-// Reports whether the connection was closed. It is go safe.
+// Closed Reports whether the connection was closed. It is go safe.
 func (c *FormattedConnection) Closed() bool {
 	return atomic.LoadInt32(&c.closed) > 0
 }

p2p/net/handshake.go

@@ -32,7 +32,7 @@ const HandshakeResp = "/handshake/1.0/handshake-resp/"
 // Node that NetworkSession is not yet authenticated - this happens only when the handshake response is processed and authenticated
 // This is called by node1 (initiator)
 func GenerateHandshakeRequestData(localPublicKey crypto.PublicKey, localPrivateKey crypto.PrivateKey, remotePublicKey crypto.PublicKey,
-	networkID int8) (*pb.HandshakeData, NetworkSession, error) {
+	networkID int8, port uint16) (*pb.HandshakeData, NetworkSession, error) {
 
 	// we use the Elliptic Curve Encryption Scheme
 	// https://en.wikipedia.org/wiki/Integrated_Encryption_Scheme
@@ -74,6 +74,8 @@ func GenerateHandshakeRequestData(localPublicKey crypto.PublicKey, localPrivateK
 	data.Hmac = hm.Sum(nil)
 	data.Sign = ""
 
+	data.Port = uint32(port)
+
 	// sign corpus - marshall data without the signature to protobufs3 binary format
 	bin, err := proto.Marshal(data)
 	if err != nil {
@@ -105,7 +107,6 @@ func ProcessHandshakeRequest(networkID int8, lPub crypto.PublicKey, lPri crypto.
 	// check that received clientversion is valid client string
 	reqVersion := strings.Split(req.ClientVersion, "/")
 	if len(reqVersion) != 2 {
-		//node.Warning("Dropping incoming message - invalid client version")
 		return nil, nil, errors.New("invalid client version")
 	}
 
@@ -231,7 +232,7 @@ func ProcessHandshakeResponse(remotePub crypto.PublicKey, s NetworkSession, resp
 	expectedMAC := hm.Sum(nil)
 
 	if !hmac.Equal(resp.Hmac, expectedMAC) {
-		return errors.New("invalid hmac")
+		return fmt.Errorf("invalid hmac need %v got %v", resp.Hmac, expectedMAC)
 	}
 
 	// verify signature
@@ -253,9 +254,5 @@ func ProcessHandshakeResponse(remotePub crypto.PublicKey, s NetworkSession, resp
 		return errors.New("invalid signature")
 	}
 
-	// TODO does peer need to hold all sessions?
-	// update remote node session here
-	//r.UpdateSession(s.String(), s)
-
 	return nil
 }

p2p/net/handshake_test.go

@@ -6,6 +6,8 @@ import (
 	"github.com/spacemeshos/go-spacemesh/p2p/node"
 	"github.com/spacemeshos/go-spacemesh/p2p/pb"
 	"github.com/stretchr/testify/assert"
+	"net"
+	"strconv"
 	"testing"
 )
 
@@ -18,20 +20,27 @@ func TestGenerateHandshakeRequestData(t *testing.T) {
 	assert.NoError(t, err, "should be able to create localnode")
 	con := NewConnectionMock(remoteNode.PublicKey())
 	remoteNet, _ := NewNet(config.ConfigValues, remoteNode)
+
 	//outchan := remoteNet.SubscribeOnNewRemoteConnections()
-	_, _, er := GenerateHandshakeRequestData(localNode.PublicKey(), localNode.PrivateKey(), con.RemotePublicKey(), remoteNet.NetworkID())
+	_, _, er := GenerateHandshakeRequestData(localNode.PublicKey(), localNode.PrivateKey(), con.RemotePublicKey(), remoteNet.NetworkID(), getPort(t, remoteNode.Node))
 	assert.NoError(t, er, "Sanity failed")
 
 }
 
+func getPort(t *testing.T, remote node.Node) uint16 {
+	_, port, err := net.SplitHostPort(remote.Address())
+	assert.NoError(t, err)
+	portint, err := strconv.Atoi(port)
+	assert.NoError(t, err)
+	return uint16(portint)
+}
+
 func generateRequestData(t *testing.T) (*pb.HandshakeData, node.LocalNode, node.LocalNode, int8) {
 
 	localNode, _ := node.GenerateTestNode(t)
 	remoteNode, _ := node.GenerateTestNode(t)
 	netId := int8(1)
-	con := NewConnectionMock(remoteNode.PublicKey())
-	//outchan := remoteNet.SubscribeOnNewRemoteConnections()
-	out, _, err := GenerateHandshakeRequestData(localNode.PublicKey(), localNode.PrivateKey(), con.RemotePublicKey(), netId)
+	out, _, err := GenerateHandshakeRequestData(localNode.PublicKey(), localNode.PrivateKey(), remoteNode.PublicKey(), netId, getPort(t, remoteNode.Node))
 	assert.NoError(t, err, "Failed to generate request")
 	return out, *localNode, *remoteNode, netId
 }