diff --git a/cmd/mist/ui_lib.go b/cmd/mist/ui_lib.go index f3559183e6..01c07dc10a 100644 --- a/cmd/mist/ui_lib.go +++ b/cmd/mist/ui_lib.go @@ -199,7 +199,7 @@ func (ui *UiLib) Connect(button qml.Object) { } func (ui *UiLib) ConnectToPeer(addr string) { - if err := ui.eth.SuggestPeer(addr); err != nil { + if err := ui.eth.SuggestPeer(addr, []byte{}); err != nil { guilogger.Infoln(err) } } diff --git a/eth/backend.go b/eth/backend.go index ad04863092..178ff63521 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -2,7 +2,6 @@ package eth import ( "fmt" - "net" "sync" "github.com/ethereum/go-ethereum/core" @@ -21,6 +20,8 @@ const ( seedNodeAddress = "poc-8.ethdev.com:30303" ) +var seednodeId []byte = nil + type Config struct { Name string Version string @@ -244,7 +245,7 @@ func (s *Ethereum) Start(seed bool) error { // TODO: read peers here if seed { logger.Infof("Connect to seed node %v", seedNodeAddress) - if err := s.SuggestPeer(seedNodeAddress); err != nil { + if err := s.SuggestPeer(seedNodeAddress, seednodeId); err != nil { return err } } @@ -253,14 +254,8 @@ func (s *Ethereum) Start(seed bool) error { return nil } -func (self *Ethereum) SuggestPeer(addr string) error { - netaddr, err := net.ResolveTCPAddr("tcp", addr) - if err != nil { - logger.Errorf("couldn't resolve %s:", addr, err) - return err - } - - self.net.SuggestPeer(netaddr.IP, netaddr.Port, nil) +func (self *Ethereum) SuggestPeer(addr string, pubkey []byte) error { + self.net.SuggestPeer(addr, pubkey) return nil } diff --git a/eth/peer_util.go b/eth/peer_util.go deleted file mode 100644 index 6cf80cde29..0000000000 --- a/eth/peer_util.go +++ /dev/null @@ -1,23 +0,0 @@ -package eth - -import ( - "encoding/json" - - "github.com/ethereum/go-ethereum/ethutil" -) - -func WritePeers(path string, addresses []string) { - if len(addresses) > 0 { - data, _ := json.MarshalIndent(addresses, "", " ") - ethutil.WriteFile(path, data) - } -} - -func ReadPeers(path string) (ips []string, err error) { - var data string - data, err = ethutil.ReadAllFile(path) - if err != nil { - json.Unmarshal([]byte(data), &ips) - } - return -} diff --git a/javascript/javascript_runtime.go b/javascript/javascript_runtime.go index adc9022b74..c5f5b83d07 100644 --- a/javascript/javascript_runtime.go +++ b/javascript/javascript_runtime.go @@ -202,7 +202,7 @@ func (self *JSRE) addPeer(call otto.FunctionCall) otto.Value { if err != nil { return otto.FalseValue() } - self.ethereum.SuggestPeer(host) + self.ethereum.SuggestPeer(host, nil) return otto.TrueValue() } diff --git a/p2p/cademlia.go b/p2p/cademlia.go new file mode 100644 index 0000000000..e356a3f133 --- /dev/null +++ b/p2p/cademlia.go @@ -0,0 +1,287 @@ +package p2p + +import ( + "fmt" + "math" + "sync" + "time" + + ethlogger "github.com/ethereum/go-ethereum/logger" +) + +var cadlogger = ethlogger.NewLogger("CAD") + +const ( + hashBytes = 20 + rowLength = 10 + maxProx = 20 +) + +var maxAge = 180 * time.Nanosecond +var purgeInterval = 300 * time.Second + +type Cademlia struct { + Hash []byte + HashBytes int + RowLength int + + MaxProx int + MaxProxBinSize int + + MaxAge time.Duration + PurgeInterval time.Duration + + proxLimit int + proxSize int + + rows []*row + + lock sync.RWMutex + quitC chan bool +} + +// public constructor with compulsory arguments +// hash is a byte slice of length equal to self.HashBytes +func NewCademlia(hash []byte) *Cademlia { + return &Cademlia{ + Hash: hash, // compulsory fields without default + } +} + +// Start brings up a pool of peers potentially from an offline persisted source +// and sets default values for optional parameters +func (self *Cademlia) Start() error { + self.lock.Lock() + defer self.lock.Unlock() + if self.quitC != nil { + return nil + } + // these + self.Hash can and should be checked against the + // saved file/db + if self.HashBytes == 0 { + self.HashBytes = hashBytes + } + if self.MaxProx == 0 { + self.MaxProx = maxProx + } + if self.RowLength == 0 { + self.RowLength = rowLength + } + // runtime parameters + if self.MaxProxBinSize == 0 { + self.MaxProxBinSize = self.RowLength + } + if self.MaxAge == time.Duration(0) { + self.MaxAge = maxAge + } + if self.PurgeInterval == time.Duration(0) { + self.PurgeInterval = purgeInterval + } + self.rows = make([]*row, self.MaxProx) + for i, _ := range self.rows { + self.rows[i] = &row{} // will initialise row{int(0),[]*entry(nil),sync.Mutex} + } + self.quitC = make(chan bool) + go self.purgeLoop() + return nil +} + +// Stop saves the routing table into a persistant form +func (self *Cademlia) Stop() { + self.lock.Lock() + defer self.lock.Unlock() + if self.quitC == nil { + return + } + close(self.quitC) + self.quitC = nil +} + +// AddPeer is the entry point where new peers are suggested for addition to the peer pool +// peers conform to the peerrInfo interface +// AddPeer(peer) returns an error if it deems the peer unworthy +func (self *Cademlia) AddPeer(peer peerInfo) (err error) { + self.lock.Lock() + defer self.lock.Unlock() + index := self.ProximityBin(peer.Hash()) + row := self.rows[index] + added := row.insert(&entry{peer: peer}) + if added { + if index >= self.proxLimit { + go self.adjustProx(index, 1) + } + cadlogger.Infof("accept peer %x...", peer.Hash()[:8]) + } else { + err = fmt.Errorf("no worse peer found") + cadlogger.Infof("reject peer %x..: %v", peer.Hash()[:8], err) + } + return +} + +// adjust Prox (proxLimit and proxSize after an insertion of add entries into row r) +func (self *Cademlia) adjustProx(r int, add int) { + self.lock.Lock() + defer self.lock.Unlock() + if r >= self.proxLimit && + self.proxSize+add > self.MaxProxBinSize && + self.rows[r].len() > 0 { + self.proxLimit++ + } else { + self.proxSize += add + } +} + +// updates Prox (proxLimit and proxSize after purging entries) +func (self *Cademlia) updateProx() { + self.lock.Lock() + defer self.lock.Unlock() + var sum int + for i := self.MaxProx - 1; i >= 0; i-- { + l := self.rows[i].len() + sum += l + if sum <= self.MaxProxBinSize || l == 0 { + self.proxSize = sum + } + } +} + +// GetPeers(target) returns the list of peers belonging to the same proximity bin as the target. The most proximate bin will be the union of the bins between proxLimit and MaxProx. proxLimit is dynamically adjusted so that 1) there is no empty rows in bin < proxLimit and 2) the sum of all items are the maximum possible but lower than MaxProxBinSize +func (self *Cademlia) GetPeers(target []byte) (peers []peerInfo) { + self.lock.RLock() + defer self.lock.RUnlock() + index := self.ProximityBin(target) + var entries []*entry + if index >= self.proxLimit { + for i := self.proxLimit; i < self.MaxProx; i++ { + entries = append(entries, self.rows[i].row...) + } + } else { + entries = self.rows[index].row + } + for _, entry := range entries { + peers = append(peers, entry.peer) + } + return +} + +// entry wrapper type for peer object adding potentially persisted metadata for offline permanent record +type entry struct { + peer peerInfo + // metadata +} + +// in situ mutable row +type row struct { + length int + row []*entry + lock sync.RWMutex +} + +func (self *row) len() int { + self.lock.RLock() + defer self.lock.RUnlock() + return self.length +} + +// insert adds a peer to a row either by appending to existing items if row length does not exceed RowLength, or by replacing the worst entry in the row +func (self *row) insert(entry *entry) (added bool) { + self.lock.Lock() + defer self.lock.Unlock() + if len(self.row) >= self.length { // >= allows us to add peers beyond the Rowlength limitation + worst := self.worst() + self.row[worst].peer.Disconnect(DiscSubprotocolError) + self.row[worst] = entry + } else { + self.row = append(self.row, entry) + added = true + self.length++ + } + return +} + +// worst expunges the single worst entry in a row, where worst entry is with a peer that has not been active the longests +func (self *row) worst() (index int) { + var oldest time.Time + for i, entry := range self.row { + if (oldest == time.Time{}) || entry.peer.LastActive().Before(oldest) { + oldest = entry.peer.LastActive() + index = i + } + } + return +} + +// expunges entries from a row that were last active more that MaxAge ago +// calls Disconnect on entry.peer +func (self *row) purge(recently time.Time) { + self.lock.Lock() + var newRow []*entry + for _, entry := range self.row { + if !entry.peer.LastActive().Before(recently) { + newRow = append(newRow, entry) + } else { + entry.peer.Disconnect(DiscSubprotocolError) + } + } + self.row = newRow + self.length = len(newRow) + self.lock.Unlock() +} + +func Hash(key []byte) []byte { + return key +} + +func (self *Cademlia) purgeLoop() { + ticker := time.Tick(self.PurgeInterval) + for { + select { + case <-self.quitC: + return + case <-ticker: + self.lock.Lock() + for _, r := range self.rows { + r.purge(time.Now().Add(-self.MaxAge)) + } + self.updateProx() + self.lock.Unlock() + } + } +} + +/* +Taking the proximity value relative to a fix point x classifies the points in the space (n byte long byte sequences) into bins the items in which are each at most half as distant from x as items in the previous bin. Given a sample of uniformly distrbuted items (a hash function over arbitrary sequence) the proximity scale maps onto series of subsets with cardinalities on a negative exponential scale. + +It also has the property that any two item belonging to the same bin are at most half as distant from each other as they are from x. + +If we think of random sample of items in the bins as connections in a network of interconnected nodes than relative proximity can serve as the basis for local decisions for graph traversal where the task is to find a route between two points. Since in every step of forwarding, the finite distance halves, there is a guaranteed constant maximum limit on the number of hops needed to reach one node from the other. +*/ + +func (self *Cademlia) ProximityBin(other []byte) (ret int) { + return int(math.Min(float64(self.MaxProx), float64(self.Proximity(self.Hash, other)))) +} + +/* +The distance metric MSB(x, y) of two equal length bytesequences x an y is the value of the +binary integer cast of the xor-ed bytesequence most significant bit first. +Proximity(x, y) counts the common zeros in the front of this distance measure. +*/ + +func (self *Cademlia) Proximity(one, other []byte) (ret int) { + xor := Xor(one, other) + for i := 0; i < self.HashBytes; i++ { + for j := 0; j < 8; j++ { + if (xor[i]>>uint8(7-j))&0x1 != 0 { + return i*8 + j + } + } + } + return self.HashBytes*8 - 1 +} + +func Xor(one, other []byte) (xor []byte) { + for i := 0; i < len(one); i++ { + xor[i] = one[i] ^ other[i] + } + return +} diff --git a/p2p/peer.go b/p2p/peer.go index 2380a3285b..87aaa10afe 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -45,8 +45,8 @@ func (d peerAddr) String() string { return fmt.Sprintf("%v:%d", d.IP, d.Port) } -func (d *peerAddr) RlpData() interface{} { - return []interface{}{string(d.IP), d.Port, d.Pubkey} +func (d peerAddr) RlpData() interface{} { + return []interface{}{d.IP, d.Port, d.Pubkey} } // Peer represents a remote peer. @@ -55,7 +55,10 @@ type Peer struct { // Use them to display messages related to the peer. *logger.Logger - infolock sync.Mutex + lastActive time.Time // updated persisted + lastActiveC chan time.Time // for constant querying + + infolock sync.RWMutex identity ClientIdentity caps []Cap listenAddr *peerAddr // what remote peer is listening on @@ -85,51 +88,92 @@ type Peer struct { // These fields are kept so base protocol can access them. // TODO: this should be one or more interfaces - ourID ClientIdentity // client id of the Server - ourListenAddr *peerAddr // listen addr of Server, nil if not listening - newPeerAddr chan<- *peerAddr // tell server about received peers - otherPeers func() []*Peer // should return the list of all peers - pubkeyHook func(*peerAddr) error // called at end of handshake to validate pubkey + hash []byte // hash of pubkey used as address + ourID ClientIdentity // client id of the Server + ourListenAddr *peerAddr // listen addr of Server, nil if not listening + addPeer func(*peerAddr) error // tell server about received peers + getPeers func(...[]byte) []*peerAddr // should return the list of all peers + verifyPeerHook func(*Peer) error // called at end of handshake to validate peer } // NewPeer returns a peer for testing purposes. func NewPeer(id ClientIdentity, caps []Cap) *Peer { conn, _ := net.Pipe() - peer := newPeer(conn, nil, nil) + peer := &Peer{} + peer.init(conn) peer.setHandshakeInfo(id, nil, caps) close(peer.closed) return peer } -func newServerPeer(server *Server, conn net.Conn, dialAddr *peerAddr) *Peer { - p := newPeer(conn, server.Protocols, dialAddr) - p.ourID = server.Identity - p.newPeerAddr = server.peerConnect - p.otherPeers = server.Peers - p.pubkeyHook = server.verifyPeer - p.runBaseProtocol = true - - // laddr can be updated concurrently by NAT traversal. - // newServerPeer must be called with the server lock held. - if server.laddr != nil { - p.ourListenAddr = newPeerAddr(server.laddr, server.Identity.Pubkey()) - } - return p +func (self *Peer) init(conn net.Conn) { + self.conn = conn + self.Logger = logger.NewLogger("P2P " + conn.RemoteAddr().String()) + self.bufconn = bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)) + self.running = make(map[string]*proto) + self.disc = make(chan DiscReason) + self.protoErr = make(chan error) + self.closed = make(chan struct{}) } -func newPeer(conn net.Conn, protocols []Protocol, dialAddr *peerAddr) *Peer { - p := &Peer{ - Logger: logger.NewLogger("P2P " + conn.RemoteAddr().String()), - conn: conn, - dialAddr: dialAddr, - bufconn: bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)), - protocols: protocols, - running: make(map[string]*proto), - disc: make(chan DiscReason), - protoErr: make(chan error), - closed: make(chan struct{}), +// ActiveAddresses returns addresses of all connected peers. +// Actually if the peer selector keeps historical info , then once active peers +// will be included too. +func ActiveAddresses(peers ...peerInfo) (addrs []*peerAddr) { + for _, peer := range peers { + if peer != nil { + addr := peer.Addr() + // filter out peers that are not listening or + // have not completed the handshake. + // the peer selector can track previously sent peers and exclude them as well. + if addr == nil { + continue + } + addrs = append(addrs, addr) + } } - return p + return addrs +} + +// implements the peerInfo interface +func (self *Peer) Addr() *peerAddr { + self.infolock.RLock() + defer self.infolock.RUnlock() + return self.listenAddr +} + +func (self *Peer) Hash() []byte { + if self.hash == nil { + self.hash = Hash(self.Pubkey()) + } + return self.hash +} + +func (self *Peer) Pubkey() (pubkey []byte) { + self.infolock.Lock() + defer self.infolock.Unlock() + switch { + case self.identity != nil: + pubkey = self.identity.Pubkey() + case self.dialAddr != nil: + pubkey = self.dialAddr.Pubkey + case self.listenAddr != nil: + pubkey = self.listenAddr.Pubkey + } + return +} + +func (self *Peer) LastActive() (lastActive time.Time) { + var ok bool + select { + case lastActive, ok = <-self.lastActiveC: + if ok { + self.lastActive = lastActive + } + default: + lastActive = self.lastActive + } + return } // Identity returns the client identity of the remote peer. The diff --git a/p2p/peer_error.go b/p2p/peer_error.go index 0eb7ec838d..73587ddca4 100644 --- a/p2p/peer_error.go +++ b/p2p/peer_error.go @@ -14,7 +14,11 @@ const ( errP2PVersionMismatch errPubkeyMissing errPubkeyInvalid - errPubkeyForbidden + errPubkeyMismatch + errBlacklistedPeer + errSelfConnection + errConnectedPeer + errRejectedPeer errProtocolBreach errPingTimeout errInvalidNetworkId @@ -31,7 +35,11 @@ var errorToString = map[int]string{ errP2PVersionMismatch: "P2P Version Mismatch", errPubkeyMissing: "Public key missing", errPubkeyInvalid: "Public key invalid", - errPubkeyForbidden: "Public key forbidden", + errPubkeyMismatch: "Public key mismatch", + errBlacklistedPeer: "Blacklisted peer", + errSelfConnection: "Self connection", + errConnectedPeer: "Connected peer", + errRejectedPeer: "Rejected peer", errProtocolBreach: "Protocol Breach", errPingTimeout: "Ping timeout", errInvalidNetworkId: "Invalid network id", @@ -117,9 +125,9 @@ func discReasonForError(err error) DiscReason { switch peerError.Code { case errP2PVersionMismatch: return DiscIncompatibleVersion - case errPubkeyMissing, errPubkeyInvalid: + case errPubkeyMissing, errPubkeyMismatch, errPubkeyInvalid: return DiscInvalidIdentity - case errPubkeyForbidden: + case errBlacklistedPeer, errSelfConnection, errConnectedPeer, errRejectedPeer: return DiscUselessPeer case errInvalidMsgCode, errMagicTokenMismatch, errProtocolBreach: return DiscProtocolError diff --git a/p2p/peer_selector.go b/p2p/peer_selector.go new file mode 100644 index 0000000000..2c1b2260f8 --- /dev/null +++ b/p2p/peer_selector.go @@ -0,0 +1,79 @@ +package p2p + +import ( + "encoding/json" + "path" + "time" + + "github.com/ethereum/go-ethereum/ethutil" +) + +type peerInfo interface { + Addr() *peerAddr + Hash() []byte + LastActive() time.Time + Disconnect(DiscReason) +} + +type peerSelector interface { + AddPeer(peer peerInfo) error + GetPeers(target ...[]byte) []peerInfo + Start() error + Stop() error +} + +type BaseSelector struct { + DirPath string + getPeers func() []peerInfo + peers []peerInfo +} + +func (self *BaseSelector) AddPeer(peer peerInfo) error { + return nil +} + +func (self *BaseSelector) GetPeers(target ...[]byte) []peerInfo { + return self.getPeers() +} + +func (self *BaseSelector) Start() error { + if len(self.DirPath) > 0 { + path := path.Join(self.DirPath, "peers.json") + peers, err := ReadPeers(path) + if err != nil { + return err + } + self.peers = peers + } + return nil +} + +func (self *BaseSelector) Stop() error { + if len(self.DirPath) > 0 { + path := path.Join(self.DirPath, "peers.json") + if err := WritePeers(path, self.peers); err != nil { + return err + } + } + return nil +} + +func WritePeers(path string, addresses []peerInfo) error { + if len(addresses) > 0 { + data, err := json.MarshalIndent(addresses, "", " ") + if err == nil { + ethutil.WriteFile(path, data) + } + return err + } + return nil +} + +func ReadPeers(path string) (peers []peerInfo, err error) { + var data string + data, err = ethutil.ReadAllFile(path) + if err == nil { + json.Unmarshal([]byte(data), &peers) + } + return +} diff --git a/p2p/peer_test.go b/p2p/peer_test.go index 4ee88f112b..c8ccf25020 100644 --- a/p2p/peer_test.go +++ b/p2p/peer_test.go @@ -30,9 +30,11 @@ var discard = Protocol{ func testPeer(protos []Protocol) (net.Conn, *Peer, <-chan error) { conn1, conn2 := net.Pipe() - peer := newPeer(conn1, protos, nil) + peer := &Peer{} + peer.init(conn1) + peer.protocols = protos peer.ourID = &peerId{} - peer.pubkeyHook = func(*peerAddr) error { return nil } + peer.verifyPeerHook = func(*Peer) error { return nil } errc := make(chan error, 1) go func() { _, err := peer.loop() diff --git a/p2p/protocol.go b/p2p/protocol.go index 1d121a8855..908cdc8161 100644 --- a/p2p/protocol.go +++ b/p2p/protocol.go @@ -114,6 +114,7 @@ func (bp *baseProtocol) loop(quit <-chan error) error { ping := time.NewTimer(pingTimeout) activity := bp.peer.activity.Subscribe(time.Time{}) lastActive := time.Time{} + lastActiveC := bp.peer.lastActiveC defer ping.Stop() defer activity.Unsubscribe() @@ -125,6 +126,7 @@ func (bp *baseProtocol) loop(quit <-chan error) error { select { case err = <-quit: return err + case lastActiveC <- lastActive: case <-getPeersTick.C: err = EncodeMsg(bp.rw, getPeersMsg) case event := <-activity.Chan(): @@ -169,15 +171,35 @@ func (bp *baseProtocol) handle(rw MsgReadWriter) error { case pongMsg: case getPeersMsg: - peers := bp.peerList() + var target [][]byte + if err := msg.Decode(&target); err != nil { + return newPeerError(errInvalidMsg, "%v", err) + } + + peers := bp.peer.getPeers(target...) + if len(target) == 0 { + // then add ourselves to the list + ourAddr := bp.peer.ourListenAddr + if ourAddr != nil && !ourAddr.IP.IsLoopback() && !ourAddr.IP.IsUnspecified() { + peers = append(peers, ourAddr) + } + } + addrs := make([]interface{}, 0, len(peers)) + // encode and filter out requesting peer + for _, addr := range peers { + if addr != bp.peer.Addr() { + addrs = append(addrs, addr) + } + } + // this is dangerous. the spec says that we should _delay_ // sending the response if no new information is available. // this means that would need to send a response later when // new peers become available. // // TODO: add event mechanism to notify baseProtocol for new peers - if len(peers) > 0 { - return EncodeMsg(bp.rw, peersMsg, peers...) + if len(addrs) > 0 { + return EncodeMsg(bp.rw, peersMsg, addrs...) } case peersMsg: @@ -187,7 +209,7 @@ func (bp *baseProtocol) handle(rw MsgReadWriter) error { } for _, addr := range peers { bp.peer.Debugf("received peer suggestion: %v", addr) - bp.peer.newPeerAddr <- addr + bp.peer.addPeer(addr) } default: @@ -227,13 +249,9 @@ func (bp *baseProtocol) readHandshake() error { // verify that the peer we wanted to connect to // actually holds the target public key. if da.Pubkey != nil && !bytes.Equal(da.Pubkey, hs.NodeID) { - return newPeerError(errPubkeyForbidden, "dial address pubkey mismatch") + return newPeerError(errPubkeyMismatch, "dial address pubkey mismatch: %x vs %x", da.Pubkey, hs.NodeID) } } - pa := newPeerAddr(bp.peer.conn.RemoteAddr(), hs.NodeID) - if err := bp.peer.pubkeyHook(pa); err != nil { - return newPeerError(errPubkeyForbidden, "%v", err) - } // TODO: remove Caps with empty name var addr *peerAddr if hs.ListenPort != 0 { @@ -241,6 +259,9 @@ func (bp *baseProtocol) readHandshake() error { addr.Port = hs.ListenPort } bp.peer.setHandshakeInfo(&hs, addr, hs.Caps) + if err := bp.peer.verifyPeerHook(bp.peer); err != nil { + return err + } bp.peer.startSubprotocols(hs.Caps) return nil } @@ -264,25 +285,3 @@ func (bp *baseProtocol) handshakeMsg() Msg { bp.peer.ourID.Pubkey()[1:], ) } - -func (bp *baseProtocol) peerList() []interface{} { - peers := bp.peer.otherPeers() - ds := make([]interface{}, 0, len(peers)) - for _, p := range peers { - p.infolock.Lock() - addr := p.listenAddr - p.infolock.Unlock() - // filter out this peer and peers that are not listening or - // have not completed the handshake. - // TODO: track previously sent peers and exclude them as well. - if p == bp.peer || addr == nil { - continue - } - ds = append(ds, addr) - } - ourAddr := bp.peer.ourListenAddr - if ourAddr != nil && !ourAddr.IP.IsLoopback() && !ourAddr.IP.IsUnspecified() { - ds = append(ds, ourAddr) - } - return ds -} diff --git a/p2p/protocol_test.go b/p2p/protocol_test.go index b1d10ac536..6a71e30add 100644 --- a/p2p/protocol_test.go +++ b/p2p/protocol_test.go @@ -29,10 +29,10 @@ func (self *peerId) Pubkey() (pubkey []byte) { func newTestPeer() (peer *Peer) { peer = NewPeer(&peerId{}, []Cap{}) - peer.pubkeyHook = func(*peerAddr) error { return nil } + peer.verifyPeerHook = func(*Peer) error { return nil } peer.ourID = &peerId{} peer.listenAddr = &peerAddr{} - peer.otherPeers = func() []*Peer { return nil } + peer.getPeers = func(...[]byte) []*peerAddr { return nil } return } @@ -79,10 +79,10 @@ func TestBaseProtocolPeers(t *testing.T) { // run first peer (in background) peer1 := newTestPeer() peer1.ourListenAddr = listenAddr - peer1.otherPeers = func() []*Peer { - pl := make([]*Peer, len(peerList)) + peer1.getPeers = func(...[]byte) []*peerAddr { + pl := make([]*peerAddr, len(peerList)) for i, addr := range peerList { - pl[i] = &Peer{listenAddr: addr} + pl[i] = addr } return pl } @@ -94,7 +94,10 @@ func TestBaseProtocolPeers(t *testing.T) { // run second peer peer2 := newTestPeer() - peer2.newPeerAddr = addrChan // feed peer suggestions into matcher + peer2.addPeer = func(addr *peerAddr) error { + addrChan <- addr // feed peer suggestions into matcher + return nil + } if err := runBaseProtocol(peer2, rw2); err != ErrPipeClosed { t.Errorf("peer2 terminated with unexpected error: %v", err) } @@ -107,7 +110,7 @@ func TestBaseProtocolPeers(t *testing.T) { func TestBaseProtocolDisconnect(t *testing.T) { peer := NewPeer(&peerId{}, nil) peer.ourID = &peerId{} - peer.pubkeyHook = func(*peerAddr) error { return nil } + peer.verifyPeerHook = func(*Peer) error { return nil } rw1, rw2 := MsgPipe() done := make(chan struct{}) diff --git a/p2p/server.go b/p2p/server.go index cfff442f71..819d38607e 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -60,11 +60,16 @@ type Server struct { Dialer *net.Dialer // If NoDial is true, the server will not dial any peers. + // this maybe used in test environments where we want to prevent a node from + // connecting (and synchronising) with other nodes NoDial bool + // peer selector + PeerSelector peerSelector + // Hook for testing. This is useful because we can inhibit // the whole protocol stack. - newPeerFunc peerFunc + // newPeerFunc peerFunc lock sync.RWMutex running bool @@ -74,9 +79,10 @@ type Server struct { peerSlots chan int peerCount int + connectFunc func(*Peer, net.Conn) + quit chan struct{} wg sync.WaitGroup - peerConnect chan *peerAddr peerDisconnect chan *Peer } @@ -90,9 +96,7 @@ type NAT interface { String() string } -type peerFunc func(srv *Server, c net.Conn, dialAddr *peerAddr) *Peer - -// Peers returns all connected peers. +// Peers returns all currently connected peers. func (srv *Server) Peers() (peers []*Peer) { srv.lock.RLock() defer srv.lock.RUnlock() @@ -104,6 +108,21 @@ func (srv *Server) Peers() (peers []*Peer) { return } +// GetPeers returns addresses near target if given or supported by the client +// or falls back to all actively connected peers +func (srv *Server) GetPeers(target ...[]byte) []*peerAddr { + if len(target) == 1 { // delegate to selector + return ActiveAddresses(srv.PeerSelector.GetPeers(target[0])...) + } else { + // in fact it is not clear why the selector would not want to reply to this case as well + var peers []peerInfo + for _, peer := range srv.Peers() { + peers = append(peers, peer) + } + return ActiveAddresses(peers...) + } +} + // PeerCount returns the number of connected peers. func (srv *Server) PeerCount() int { srv.lock.RLock() @@ -111,13 +130,75 @@ func (srv *Server) PeerCount() int { return srv.peerCount } -// SuggestPeer injects an address into the outbound address pool. -func (srv *Server) SuggestPeer(ip net.IP, port int, nodeID []byte) { - addr := &peerAddr{ip, uint64(port), nodeID} +// SuggestPeer is a convenient method that does dns resolution +// and passes on the request to AddPeer +func (srv *Server) SuggestPeer(addr string, pubkey []byte) error { + netaddr, err := net.ResolveTCPAddr("tcp", addr) + if err != nil { + srvlog.Errorf("couldn't resolve %s:", addr, err) + return err + } + peerAddr := &peerAddr{netaddr.IP, uint64(netaddr.Port), pubkey} + return srv.AddPeer(peerAddr) +} + +// AddPeer takes a peerAddr address as argument. +// If not found among connected peers turns to the peerSelector +// to decide if it is a worthwhile connection +func (srv *Server) AddPeer(addr *peerAddr) (err error) { + if srv.NoDial { + return fmt.Errorf("no dial out") + } + // need to look up nodeID first + peer := &Peer{ + dialAddr: addr, + lastActiveC: make(chan time.Time), + lastActive: time.Now().Add(-24 * time.Hour), + } + if err = srv.PeerSelector.AddPeer(peer); err == nil { + srvlog.Infof("peer %v accepted by peer selection", addr) + err = srv.dialPeer(peer) + } else { + srvlog.Infof("peer %v rejected by peer selection", addr) + } + return +} + +func (srv *Server) dialPeer(peer *Peer) (err error) { + timeout := time.After(5 * time.Second) select { - case srv.peerConnect <- addr: - default: // don't block - srvlog.Warnf("peer suggestion %v ignored", addr) + case <-timeout: + err = fmt.Errorf("Too many connections. No slot available") + case slot := <-srv.peerSlots: // there is a slot available + srvlog.Infof("Dialing %v (slot %d)\n", peer.dialAddr, slot) + conn, dialErr := srv.Dialer.Dial(peer.dialAddr.Network(), peer.dialAddr.String()) + if dialErr != nil { + err = fmt.Errorf("Dial error: %v", dialErr) + srvlog.Errorln(err) + srv.peerSlots <- slot + return + } + srvlog.Infof("Connected to %v (slot %d)\n", peer.dialAddr, slot) + peer.slot = slot + srv.connectFunc(peer, conn) + go srv.addPeer(peer) + } + return + +} + +func (srv *Server) connect(p *Peer, conn net.Conn) { + p.init(conn) + p.ourID = srv.Identity + p.addPeer = srv.AddPeer + p.getPeers = srv.GetPeers + p.verifyPeerHook = srv.verifyPeer + p.runBaseProtocol = true + p.protocols = srv.Protocols + + // laddr can be updated concurrently by NAT traversal. + if srv.laddr != nil { + p.ourListenAddr = newPeerAddr(srv.laddr, srv.Identity.Pubkey()) } } @@ -162,11 +243,10 @@ func (srv *Server) Start() (err error) { srv.quit = make(chan struct{}) srv.peers = make([]*Peer, srv.MaxPeers) srv.peerSlots = make(chan int, srv.MaxPeers) - srv.peerConnect = make(chan *peerAddr, outboundAddressPoolSize) srv.peerDisconnect = make(chan *Peer) - if srv.newPeerFunc == nil { - srv.newPeerFunc = newServerPeer - } + // if srv.newPeerFunc == nil { + // srv.newPeerFunc = newServerPeer + // } if srv.Blacklist == nil { srv.Blacklist = NewBlacklist() } @@ -179,12 +259,13 @@ func (srv *Server) Start() (err error) { return err } } - if !srv.NoDial { - srv.wg.Add(1) - go srv.dialLoop() + + if srv.PeerSelector == nil { + srv.PeerSelector = &BaseSelector{} } - if srv.NoDial && srv.ListenAddr == "" { - srvlog.Warnln("I will be kind-of useless, neither dialing nor listening.") + + if srv.connectFunc == nil { + srv.connectFunc = srv.connect } // make all slots available @@ -266,8 +347,10 @@ func (srv *Server) listenLoop() { srv.peerSlots <- slot return } - srvlog.Debugf("Accepted conn %v (slot %d)\n", conn.RemoteAddr(), slot) - srv.addPeer(conn, nil, slot) + srvlog.Debugf("Accepted conn %v (slot %d) - peer selector check after handshake", conn.RemoteAddr(), slot) + peer := &Peer{slot: slot} + srv.connectFunc(peer, conn) + srv.addPeer(peer) case <-srv.quit: return } @@ -313,77 +396,22 @@ func (srv *Server) removePortMapping(port int) { srv.NAT.DeletePortMapping("tcp", port, port) } -func (srv *Server) dialLoop() { - defer srv.wg.Done() - var ( - suggest chan *peerAddr - slot *int - slots = srv.peerSlots - ) - for { - select { - case i := <-slots: - // we need a peer in slot i, slot reserved - slot = &i - // now we can watch for candidate peers in the next loop - suggest = srv.peerConnect - // do not consume more until candidate peer is found - slots = nil - - case desc := <-suggest: - // candidate peer found, will dial out asyncronously - // if connection fails slot will be released - srvlog.Infof("dial %v (%v)", desc, *slot) - go srv.dialPeer(desc, *slot) - // we can watch if more peers needed in the next loop - slots = srv.peerSlots - // until then we dont care about candidate peers - suggest = nil - - case <-srv.quit: - // give back the currently reserved slot - if slot != nil { - srv.peerSlots <- *slot - } - return - } - } -} - -// connect to peer via dial out -func (srv *Server) dialPeer(desc *peerAddr, slot int) { - srvlog.Debugf("Dialing %v (slot %d)\n", desc, slot) - conn, err := srv.Dialer.Dial(desc.Network(), desc.String()) - if err != nil { - srvlog.Errorf("Dial error: %v", err) - srv.peerSlots <- slot - return - } - go srv.addPeer(conn, desc, slot) -} - // creates the new peer object and inserts it into its slot -func (srv *Server) addPeer(conn net.Conn, desc *peerAddr, slot int) *Peer { +func (srv *Server) addPeer(peer *Peer) { srv.lock.Lock() defer srv.lock.Unlock() - if !srv.running { - conn.Close() - srv.peerSlots <- slot // release slot - return nil - } - peer := srv.newPeerFunc(srv, conn, desc) - peer.slot = slot - srv.peers[slot] = peer + srvlog.Debugf("Add peer %v (slot %v)\n", peer, peer.slot) + srv.peers[peer.slot] = peer srv.peerCount++ go func() { peer.loop(); srv.peerDisconnect <- peer }() - return peer + return } // removes peer: sending disconnect msg, stop peer, remove rom list/table, release slot func (srv *Server) removePeer(peer *Peer) { srv.lock.Lock() defer srv.lock.Unlock() - srvlog.Debugf("Removing %v (slot %v)\n", peer, peer.slot) + srvlog.Debugf("Remove peer %v (slot %v)\n", peer, peer.slot) if srv.peers[peer.slot] != peer { srvlog.Warnln("Invalid peer to remove:", peer) return @@ -395,23 +423,29 @@ func (srv *Server) removePeer(peer *Peer) { srv.peerSlots <- peer.slot } -func (srv *Server) verifyPeer(addr *peerAddr) error { - if srv.Blacklist.Exists(addr.Pubkey) { - return errors.New("blacklisted") +func (srv *Server) verifyPeer(peer *Peer) error { + pubkey := peer.Pubkey() + if srv.Blacklist.Exists(pubkey) { + return newPeerError(errBlacklistedPeer, "") } - if bytes.Equal(srv.Identity.Pubkey()[1:], addr.Pubkey) { - return newPeerError(errPubkeyForbidden, "not allowed to connect to srv") + if bytes.Equal(srv.Identity.Pubkey()[1:], pubkey) { + return newPeerError(errSelfConnection, "") } srv.lock.RLock() - defer srv.lock.RUnlock() for _, peer := range srv.peers { if peer != nil { id := peer.Identity() - if id != nil && bytes.Equal(id.Pubkey(), addr.Pubkey) { - return errors.New("already connected") + if id != nil && bytes.Equal(id.Pubkey(), pubkey) { + return newPeerError(errConnectedPeer, "") } } } + srv.lock.RUnlock() + if peer.dialAddr == nil { + if err := srv.PeerSelector.AddPeer(peer); err != nil { + return newPeerError(errRejectedPeer, "%v", err) + } + } return nil } diff --git a/p2p/server_test.go b/p2p/server_test.go index ceb89e3f7f..2545aecfc8 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -9,12 +9,12 @@ import ( "time" ) -func startTestServer(t *testing.T, pf peerFunc) *Server { +func startTestServer(t *testing.T, cb func(*Peer, net.Conn)) *Server { server := &Server{ Identity: &peerId{}, MaxPeers: 10, ListenAddr: "127.0.0.1:0", - newPeerFunc: pf, + connectFunc: cb, } if err := server.Start(); err != nil { t.Fatalf("Could not start server: %v", err) @@ -27,16 +27,9 @@ func TestServerListen(t *testing.T) { // start the test server connected := make(chan *Peer) - srv := startTestServer(t, func(srv *Server, conn net.Conn, dialAddr *peerAddr) *Peer { - if conn == nil { - t.Error("peer func called with nil conn") - } - if dialAddr != nil { - t.Error("peer func called with non-nil dialAddr") - } - peer := newPeer(conn, nil, dialAddr) + srv := startTestServer(t, func(peer *Peer, conn net.Conn) { + peer.init(conn) connected <- peer - return peer }) defer close(connected) defer srv.Stop() @@ -50,10 +43,18 @@ func TestServerListen(t *testing.T) { select { case peer := <-connected: - if peer.conn.LocalAddr().String() != conn.RemoteAddr().String() { - t.Errorf("peer started with wrong conn: got %v, want %v", - peer.conn.LocalAddr(), conn.RemoteAddr()) + if peer.conn == nil { + t.Error("peer setup with nil conn") + } else { + if peer.conn.LocalAddr().String() != conn.RemoteAddr().String() { + t.Errorf("peer started with wrong conn: got %v, want %v", + peer.conn.LocalAddr(), conn.RemoteAddr()) + } } + if peer.dialAddr != nil { + t.Error("peer setup with non-nil dialAddr") + } + case <-time.After(1 * time.Second): t.Error("server did not accept within one second") } @@ -72,7 +73,7 @@ func TestServerDial(t *testing.T) { go func() { conn, err := listener.Accept() if err != nil { - t.Error("acccept error:", err) + t.Error("accept error:", err) } conn.Close() accepted <- conn @@ -80,28 +81,28 @@ func TestServerDial(t *testing.T) { // start the test server connected := make(chan *Peer) - srv := startTestServer(t, func(srv *Server, conn net.Conn, dialAddr *peerAddr) *Peer { - if conn == nil { - t.Error("peer func called with nil conn") - } - peer := newPeer(conn, nil, dialAddr) - connected <- peer - return peer + srv := startTestServer(t, func(peer *Peer, conn net.Conn) { + peer.init(conn) + go func() { connected <- peer }() }) defer close(connected) defer srv.Stop() // tell the server to connect. connAddr := newPeerAddr(listener.Addr(), nil) - srv.peerConnect <- connAddr + srv.AddPeer(connAddr) select { case conn := <-accepted: select { case peer := <-connected: - if peer.conn.RemoteAddr().String() != conn.LocalAddr().String() { - t.Errorf("peer started with wrong conn: got %v, want %v", - peer.conn.RemoteAddr(), conn.LocalAddr()) + if conn == nil { + t.Error("peer func called with nil conn") + } else { + if peer.conn.RemoteAddr().String() != conn.LocalAddr().String() { + t.Errorf("peer started with wrong conn: got %v, want %v", + peer.conn.RemoteAddr(), conn.LocalAddr()) + } } if peer.dialAddr != connAddr { t.Errorf("peer started with wrong dialAddr: got %v, want %v", @@ -119,11 +120,11 @@ func TestServerDial(t *testing.T) { func TestServerBroadcast(t *testing.T) { defer testlog(t).detach() var connected sync.WaitGroup - srv := startTestServer(t, func(srv *Server, c net.Conn, dialAddr *peerAddr) *Peer { - peer := newPeer(c, []Protocol{discard}, dialAddr) + srv := startTestServer(t, func(peer *Peer, conn net.Conn) { + peer.init(conn) + peer.protocols = []Protocol{discard} peer.startSubprotocols([]Cap{discard.cap()}) connected.Done() - return peer }) defer srv.Stop()