forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-21 20:26:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

cademlia changed to use hash of public key as address/nodeID

Browse source 
- rename NodeID, ID -> PublicKey
- add NodeID as 256 bit hash field to structs
- use NodeID in cademlia calculations and lookup target
- use PublicKey elsewhere
- add exportedd accessors for own and remote PublicKey and NodeID
This commit is contained in:
zelig 2015-02-15 00:48:00 +01:00
parent c29b01ce75
commit f5f69dff19
8 changed files with 198 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
eth/protocol.go
View file

@ -92,7 +92,7 @@ func EthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool)
// the main loop that handles incoming messages
// note RemovePeer in the post-disconnect hook
func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
id := peer.ID()
id := peer.PublicKey()
self := &ethProtocol{
txPool: txPool,
chainManager: chainManager,

13
p2p/crypto.go
View file

@ -1,7 +1,6 @@
package p2p
import (
// "binary"
"crypto/ecdsa"
"crypto/rand"
"fmt"
@ -37,26 +36,26 @@ func (self hexkey) String() string {
}
func encHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey, dial *discover.Node) (
remoteID discover.NodeID,
remotePublicKey discover.PublicKey,
sessionToken []byte,
err error,
) {
if dial == nil {
var remotePubkey []byte
sessionToken, remotePubkey, err = inboundEncHandshake(conn, prv, nil)
copy(remoteID[:], remotePubkey)
copy(remotePublicKey[:], remotePubkey)
} else {
remoteID = dial.ID
sessionToken, err = outboundEncHandshake(conn, prv, remoteID[:], nil)
remotePublicKey = dial.PublicKey
sessionToken, err = outboundEncHandshake(conn, prv, remotePublicKey[:], nil)
}
return remoteID, sessionToken, err
return remotePublicKey, sessionToken, err
}
// outboundEncHandshake negotiates a session token on conn.
// it should be called on the dialing side of the connection.
//
// privateKey is the local client's private key
// remotePublicKey is the remote peer's node ID
// remotePublicKey is the remote peer's node PublicKey
// sessionToken is the token from a previous session with this node.
func outboundEncHandshake(conn io.ReadWriter, prvKey *ecdsa.PrivateKey, remotePublicKey []byte, sessionToken []byte) (
newSessionToken []byte,

84
p2p/discover/node.go
View file

@ -1,6 +1,7 @@
package discover
import (
"crypto"
"crypto/ecdsa"
"crypto/elliptic"
"encoding/hex"
@ -14,16 +15,28 @@ import (
"strings"
"time"
ethcrypto "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/crypto/secp256k1"
"github.com/ethereum/go-ethereum/rlp"
)
const nodeIDBits = 512
const (
PublicKeyBits = 512
idBits = 256
)
var (
hashfunc = crypto.SHA256
// idSize = hashfunc.Size()
)
type NodeID [idBits / 8]byte
// Node represents a host on the network.
type Node struct {
ID NodeID
PublicKey PublicKey
IP net.IP
NodeID NodeID
DiscPort int // UDP listening port for discovery protocol
TCPPort int // TCP listening port for RLPx
@ -31,16 +44,22 @@ type Node struct {
active time.Time
}
func newNode(id NodeID, addr *net.UDPAddr) *Node {
func newNode(pubkey PublicKey, id NodeID, addr *net.UDPAddr) *Node {
return &Node{
ID: id,
PublicKey: pubkey,
IP: addr.IP,
DiscPort: addr.Port,
TCPPort: addr.Port,
active: time.Now(),
NodeID: id,
}
}
func Hash(id PublicKey) (a NodeID) {
copy(a[:], ethcrypto.Sha256(id[:]))
return
}
func (n *Node) isValid() bool {
// TODO: don't accept localhost, LAN addresses from internet hosts
return !n.IP.IsMulticast() && !n.IP.IsUnspecified() && n.TCPPort != 0 && n.DiscPort != 0
@ -52,7 +71,7 @@ func (n *Node) String() string {
addr := net.TCPAddr{IP: n.IP, Port: n.TCPPort}
u := url.URL{
Scheme: "enode",
User: url.User(fmt.Sprintf("%x", n.ID[:])),
User: url.User(fmt.Sprintf("%x", n.PublicKey[:])),
Host: addr.String(),
}
if n.DiscPort != n.TCPPort {
@ -65,7 +84,7 @@ func (n *Node) String() string {
//
// A node URL has scheme "enode".
//
// The hexadecimal node ID is encoded in the username portion of the
// The hexadecimal node PublicKey is encoded in the username portion of the
// URL, separated from the host by an @ sign. The hostname can only be
// given as an IP address, DNS domain names are not allowed. The port
// in the host name section is the TCP listening port. If the TCP and
@ -84,11 +103,12 @@ func ParseNode(rawurl string) (*Node, error) {
return nil, errors.New("invalid URL scheme, want \"enode\"")
}
if u.User == nil {
return nil, errors.New("does not contain node ID")
return nil, errors.New("does not contain node PublicKey")
}
if n.ID, err = HexID(u.User.String()); err != nil {
return nil, fmt.Errorf("invalid node ID (%v)", err)
if n.PublicKey, err = HexPublicKey(u.User.String()); err != nil {
return nil, fmt.Errorf("invalid node PublicKey (%v)", err)
}
n.NodeID = Hash(n.PublicKey)
ip, port, err := net.SplitHostPort(u.Host)
if err != nil {
return nil, fmt.Errorf("invalid host: %v", err)
@ -120,14 +140,14 @@ func MustParseNode(rawurl string) *Node {
}
func (n Node) EncodeRLP(w io.Writer) error {
return rlp.Encode(w, rpcNode{IP: n.IP.String(), Port: uint16(n.TCPPort), ID: n.ID})
return rlp.Encode(w, rpcNode{IP: n.IP.String(), Port: uint16(n.TCPPort), PublicKey: n.PublicKey})
}
func (n *Node) DecodeRLP(s *rlp.Stream) (err error) {
var ext rpcNode
if err = s.Decode(&ext); err == nil {
n.TCPPort = int(ext.Port)
n.DiscPort = int(ext.Port)
n.ID = ext.ID
n.PublicKey = ext.PublicKey
if n.IP = net.ParseIP(ext.IP); n.IP == nil {
return errors.New("invalid IP string")
}
@ -135,27 +155,27 @@ func (n *Node) DecodeRLP(s *rlp.Stream) (err error) {
return err
}
// NodeID is a unique identifier for each node.
// PublicKey is a unique identifier for each node.
// The node identifier is a marshaled elliptic curve public key.
type NodeID [nodeIDBits / 8]byte
type PublicKey [PublicKeyBits / 8]byte
// NodeID prints as a long hexadecimal number.
func (n NodeID) String() string {
// PublicKey prints as a long hexadecimal number.
func (n PublicKey) String() string {
return fmt.Sprintf("%#x", n[:])
}
// The Go syntax representation of a NodeID is a call to HexID.
func (n NodeID) GoString() string {
return fmt.Sprintf("discover.HexID(\"%#x\")", n[:])
// The Go syntax representation of a PublicKey is a call to HexPublicKey.
func (n PublicKey) GoString() string {
return fmt.Sprintf("discover.HexPublicKey(\"%#x\")", n[:])
}
// HexID converts a hex string to a NodeID.
// HexPublicKey converts a hex string to a PublicKey.
// The string may be prefixed with 0x.
func HexID(in string) (NodeID, error) {
func HexPublicKey(in string) (PublicKey, error) {
if strings.HasPrefix(in, "0x") {
in = in[2:]
}
var id NodeID
var id PublicKey
b, err := hex.DecodeString(in)
if err != nil {
return id, err
@ -166,19 +186,19 @@ func HexID(in string) (NodeID, error) {
return id, nil
}
// MustHexID converts a hex string to a NodeID.
// It panics if the string is not a valid NodeID.
func MustHexID(in string) NodeID {
id, err := HexID(in)
// MustHexPublicKey converts a hex string to a PublicKey.
// It panics if the string is not a valid PublicKey.
func MustHexPublicKey(in string) PublicKey {
id, err := HexPublicKey(in)
if err != nil {
panic(err)
}
return id
}
// PubkeyID returns a marshaled representation of the given public key.
func PubkeyID(pub *ecdsa.PublicKey) NodeID {
var id NodeID
// PublicKey returns a marshaled representation of the given public key.
func exportPublicKey(pub *ecdsa.PublicKey) PublicKey {
var id PublicKey
pbytes := elliptic.Marshal(pub.Curve, pub.X, pub.Y)
if len(pbytes)-1 != len(id) {
panic(fmt.Errorf("need %d bit pubkey, got %d bits", (len(id)+1)*8, len(pbytes)))
@ -187,9 +207,9 @@ func PubkeyID(pub *ecdsa.PublicKey) NodeID {
return id
}
// recoverNodeID computes the public key used to sign the
// recoverPublicKey computes the public key used to sign the
// given hash from the signature.
func recoverNodeID(hash, sig []byte) (id NodeID, err error) {
func recoverPublicKey(hash, sig []byte) (id PublicKey, err error) {
pubkey, err := secp256k1.RecoverPubkey(hash, sig)
if err != nil {
return id, err
@ -270,8 +290,8 @@ func logdist(a, b NodeID) int {
return len(a)*8 - lz
}
// randomID returns a random NodeID such that logdist(a, b) == n
func randomID(a NodeID, n int) (b NodeID) {
// randomNodeID returns a random NodeID such that logdist(a, b) == n
func randomNodeID(a NodeID, n int) (b NodeID) {
if n == 0 {
return a
}

58
p2p/discover/table.go
View file

@ -2,7 +2,7 @@
//
// The Node Discovery protocol provides a way to find RLPx nodes that
// can be connected to. It uses a Kademlia-like protocol to maintain a
// distributed database of the IDs and endpoints of all listening
// distributed database of the PublicKeys and endpoints of all listening
// nodes.
package discover
@ -16,7 +16,7 @@ import (
const (
alpha = 3 // Kademlia concurrency factor
bucketSize = 16 // Kademlia bucket size
nBuckets = nodeIDBits + 1 // Number of buckets
nBuckets = idBits + 1 // Number of buckets
)
type Table struct {
@ -43,17 +43,22 @@ type bucket struct {
entries []*Node
}
func newTable(t transport, ourID NodeID, ourAddr *net.UDPAddr) *Table {
tab := &Table{net: t, self: newNode(ourID, ourAddr)}
func newTable(t transport, ourPublicKey PublicKey, id NodeID, ourAddr *net.UDPAddr) *Table {
tab := &Table{net: t, self: newNode(ourPublicKey, id, ourAddr)}
for i := range tab.buckets {
tab.buckets[i] = new(bucket)
}
return tab
}
// Self returns the local node ID.
func (tab *Table) Self() NodeID {
return tab.self.ID
// PublicKey returns the local node PublicKey.
func (tab *Table) PublicKey() PublicKey {
return tab.self.PublicKey
}
// NodeID returns the local node NodeID.
func (tab *Table) NodeID() NodeID {
return tab.self.NodeID
}
// Close terminates the network listener.
@ -90,11 +95,11 @@ func (tab *Table) Lookup(target NodeID) []*Node {
// don't query further if we hit the target or ourself.
// unlikely to happen often in practice.
asked[target] = true
asked[tab.self.ID] = true
asked[tab.self.NodeID] = true
tab.mutex.Lock()
// update last lookup stamp (for refresh logic)
tab.buckets[logdist(tab.self.ID, target)].lastLookup = time.Now()
tab.buckets[logdist(tab.self.NodeID, target)].lastLookup = time.Now()
// generate initial result set
result := tab.closest(target, bucketSize)
tab.mutex.Unlock()
@ -103,8 +108,8 @@ func (tab *Table) Lookup(target NodeID) []*Node {
// ask the alpha closest nodes that we haven't asked yet
for i := 0; i < len(result.entries) && pendingQueries < alpha; i++ {
n := result.entries[i]
if !asked[n.ID] {
asked[n.ID] = true
if !asked[n.NodeID] {
asked[n.NodeID] = true
pendingQueries++
go func() {
result, _ := tab.net.findnode(n, target)
@ -120,8 +125,8 @@ func (tab *Table) Lookup(target NodeID) []*Node {
// wait for the next reply
for _, n := range <-reply {
cn := n
if !seen[n.ID] {
seen[n.ID] = true
if !seen[n.NodeID] {
seen[n.NodeID] = true
result.push(cn, bucketSize)
}
}
@ -142,13 +147,13 @@ func (tab *Table) refresh() {
}
tab.mutex.Unlock()
result := tab.Lookup(randomID(tab.self.ID, ld))
result := tab.Lookup(randomNodeID(tab.self.NodeID, ld))
if len(result) == 0 {
// bootstrap the table with a self lookup
tab.mutex.Lock()
tab.add(tab.nursery)
tab.mutex.Unlock()
tab.Lookup(tab.self.ID)
tab.Lookup(tab.self.NodeID)
// TODO: the Kademlia paper says that we're supposed to perform
// random lookups in all buckets further away than our closest neighbor.
}
@ -179,10 +184,11 @@ func (tab *Table) len() (n int) {
// bumpOrAdd updates the activity timestamp for the given node and
// attempts to insert the node into a bucket. The returned Node might
// not be part of the table. The caller must hold tab.mutex.
func (tab *Table) bumpOrAdd(node NodeID, from *net.UDPAddr) (n *Node) {
b := tab.buckets[logdist(tab.self.ID, node)]
if n = b.bump(node); n == nil {
n = newNode(node, from)
func (tab *Table) bumpOrAdd(pubkey PublicKey, from *net.UDPAddr) (n *Node) {
id := Hash(pubkey)
b := tab.buckets[logdist(tab.self.NodeID, id)]
if n = b.bump(id); n == nil {
n = newNode(pubkey, id, from)
if len(b.entries) == bucketSize {
tab.pingReplace(n, b)
} else {
@ -213,8 +219,8 @@ func (tab *Table) pingReplace(n *Node, b *bucket) {
// bump updates the activity timestamp for the given node.
// The caller must hold tab.mutex.
func (tab *Table) bump(node NodeID) {
tab.buckets[logdist(tab.self.ID, node)].bump(node)
func (tab *Table) bump(id NodeID) {
tab.buckets[logdist(tab.self.NodeID, id)].bump(id)
}
// add puts the entries into the table if their corresponding
@ -222,14 +228,14 @@ func (tab *Table) bump(node NodeID) {
func (tab *Table) add(entries []*Node) {
outer:
for _, n := range entries {
if n == nil || n.ID == tab.self.ID {
if n == nil || n.NodeID == tab.self.NodeID {
// skip bad entries. The RLP decoder returns nil for empty
// input lists.
continue
}
bucket := tab.buckets[logdist(tab.self.ID, n.ID)]
bucket := tab.buckets[logdist(tab.self.NodeID, n.NodeID)]
for i := range bucket.entries {
if bucket.entries[i].ID == n.ID {
if bucket.entries[i].NodeID == n.NodeID {
// already in bucket
continue outer
}
@ -242,7 +248,7 @@ outer:
func (b *bucket) bump(id NodeID) *Node {
for i, n := range b.entries {
if n.ID == id {
if n.NodeID == id {
n.active = time.Now()
// move it to the front
copy(b.entries[1:], b.entries[:i+1])
@ -263,7 +269,7 @@ type nodesByDistance struct {
// push adds the given node to the list, keeping the total size below maxElems.
func (h *nodesByDistance) push(n *Node, maxElems int) {
ix := sort.Search(len(h.entries), func(i int) bool {
return distcmp(h.target, h.entries[i].ID, n.ID) > 0
return distcmp(h.target, h.entries[i].NodeID, n.NodeID) > 0
})
if len(h.entries) < maxElems {
h.entries = append(h.entries, n)

53
p2p/discover/udp.go
View file

@ -73,7 +73,8 @@ type (
type rpcNode struct {
IP string
Port uint16
ID NodeID
PublicKey PublicKey
id NodeID
}
// udp implements the RPC protocol.
@ -99,7 +100,8 @@ type udp struct {
// to all the callback functions for that node.
type pending struct {
// these fields must match in the reply.
from NodeID
from PublicKey
id NodeID
ptype byte
// time when the request must complete
@ -117,7 +119,8 @@ type pending struct {
}
type reply struct {
from NodeID
from PublicKey
id NodeID
ptype byte
data interface{}
}
@ -150,7 +153,11 @@ func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface) (*Table
realaddr = &net.UDPAddr{IP: ext, Port: realaddr.Port}
}
}
udp.Table = newTable(udp, PubkeyID(&priv.PublicKey), realaddr)
pubkey := exportPublicKey(&priv.PublicKey)
hash := Hash(pubkey)
var id NodeID
copy(id[:], hash[:])
udp.Table = newTable(udp, pubkey, id, realaddr)
go udp.loop()
go udp.readLoop()
@ -167,7 +174,7 @@ func (t *udp) close() {
// ping sends a ping message to the given node and waits for a reply.
func (t *udp) ping(e *Node) error {
// TODO: maybe check for ReplyTo field in callback to measure RTT
errc := t.pending(e.ID, pongPacket, func(interface{}) bool { return true })
errc := t.pending(e.PublicKey, pongPacket, func(interface{}) bool { return true })
t.send(e, pingPacket, ping{
IP: t.self.IP.String(),
Port: uint16(t.self.TCPPort),
@ -181,7 +188,7 @@ func (t *udp) ping(e *Node) error {
func (t *udp) findnode(to *Node, target NodeID) ([]*Node, error) {
nodes := make([]*Node, 0, bucketSize)
nreceived := 0
errc := t.pending(to.ID, neighborsPacket, func(r interface{}) bool {
errc := t.pending(to.PublicKey, neighborsPacket, func(r interface{}) bool {
reply := r.(*neighbors)
for _, n := range reply.Nodes {
nreceived++
@ -202,9 +209,10 @@ func (t *udp) findnode(to *Node, target NodeID) ([]*Node, error) {
// pending adds a reply callback to the pending reply queue.
// see the documentation of type pending for a detailed explanation.
func (t *udp) pending(id NodeID, ptype byte, callback func(interface{}) bool) <-chan error {
func (t *udp) pending(pubkey PublicKey, ptype byte, callback func(interface{}) bool) <-chan error {
ch := make(chan error, 1)
p := &pending{from: id, ptype: ptype, callback: callback, errc: ch}
id := Hash(pubkey)
p := &pending{from: pubkey, id: id, ptype: ptype, callback: callback, errc: ch}
select {
case t.addpending <- p:
// loop will handle it
@ -339,13 +347,13 @@ func (t *udp) packetIn(from *net.UDPAddr, buf []byte) error {
if !bytes.Equal(hash, shouldhash) {
return errBadHash
}
fromID, err := recoverNodeID(crypto.Sha3(buf[headSize:]), sig)
fromPublicKey, err := recoverPublicKey(crypto.Sha3(buf[headSize:]), sig)
if err != nil {
return err
}
var req interface {
handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte) error
handle(t *udp, from *net.UDPAddr, fromPublicKey PublicKey, mac []byte) error
}
switch ptype := sigdata[0]; ptype {
case pingPacket:
@ -363,16 +371,16 @@ func (t *udp) packetIn(from *net.UDPAddr, buf []byte) error {
return err
}
log.DebugDetailf("<<< %v %T %v\n", from, req, req)
return req.handle(t, from, fromID, hash)
return req.handle(t, from, fromPublicKey, hash)
}
func (req *ping) handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte) error {
func (req *ping) handle(t *udp, from *net.UDPAddr, fromPublicKey PublicKey, mac []byte) error {
if expired(req.Expiration) {
return errExpired
}
t.mutex.Lock()
// Note: we're ignoring the provided IP address right now
n := t.bumpOrAdd(fromID, from)
n := t.bumpOrAdd(fromPublicKey, from)
if req.Port != 0 {
n.TCPPort = int(req.Port)
}
@ -385,24 +393,26 @@ func (req *ping) handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte) er
return nil
}
func (req *pong) handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte) error {
func (req *pong) handle(t *udp, from *net.UDPAddr, fromPublicKey PublicKey, mac []byte) error {
if expired(req.Expiration) {
return errExpired
}
id := Hash(fromPublicKey)
t.mutex.Lock()
t.bump(fromID)
t.bump(id)
t.mutex.Unlock()
t.replies <- reply{fromID, pongPacket, req}
t.replies <- reply{fromPublicKey, id, pongPacket, req}
return nil
}
func (req *findnode) handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte) error {
func (req *findnode) handle(t *udp, from *net.UDPAddr, fromPublicKey PublicKey, mac []byte) error {
if expired(req.Expiration) {
return errExpired
}
t.mutex.Lock()
e := t.bumpOrAdd(fromID, from)
e := t.bumpOrAdd(fromPublicKey, from)
closest := t.closest(req.Target, bucketSize).entries
t.mutex.Unlock()
@ -413,16 +423,17 @@ func (req *findnode) handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte
return nil
}
func (req *neighbors) handle(t *udp, from *net.UDPAddr, fromID NodeID, mac []byte) error {
func (req *neighbors) handle(t *udp, from *net.UDPAddr, fromPublicKey PublicKey, mac []byte) error {
if expired(req.Expiration) {
return errExpired
}
id := Hash(fromPublicKey)
t.mutex.Lock()
t.bump(fromID)
t.bump(id)
t.add(req.Nodes)
t.mutex.Unlock()
t.replies <- reply{fromID, neighborsPacket, req}
t.replies <- reply{fromPublicKey, id, neighborsPacket, req}
return nil
}

52
p2p/peer.go
View file

@ -39,7 +39,7 @@ type handshake struct {
Name string
Caps []Cap
ListenPort uint64
NodeID discover.NodeID
PublicKey discover.PublicKey
}
// Peer represents a connected remote node.
@ -52,6 +52,7 @@ type Peer struct {
name string
caps []Cap
ourPublicKey, remotePublicKey *discover.PublicKey
ourID, remoteID *discover.NodeID
ourName string
@ -72,17 +73,17 @@ type Peer struct {
}
// NewPeer returns a peer for testing purposes.
func NewPeer(id discover.NodeID, name string, caps []Cap) *Peer {
func NewPeer(remotePublicKey discover.PublicKey, name string, caps []Cap) *Peer {
conn, _ := net.Pipe()
peer := newPeer(conn, nil, "", nil, &id)
peer := newPeer(conn, nil, "", nil, nil, &remotePublicKey)
peer.setHandshakeInfo(name, caps)
close(peer.closed) // ensures Disconnect doesn't block
return peer
}
// ID returns the node's public key.
func (p *Peer) ID() discover.NodeID {
return *p.remoteID
// PublicKey returns the node's public key.
func (p *Peer) PublicKey() discover.PublicKey {
return *p.remotePublicKey
}
// Name returns the node name that the remote node advertised.
@ -115,6 +116,22 @@ func (p *Peer) LocalAddr() net.Addr {
return p.rw.LocalAddr()
}
func (p *Peer) OurPublicKey() *discover.PublicKey {
return p.ourPublicKey
}
func (p *Peer) OurID() *discover.NodeID {
return p.ourID
}
func (p *Peer) RemotePublicKey() *discover.PublicKey {
return p.remotePublicKey
}
func (p *Peer) RemoteID() *discover.NodeID {
return p.remoteID
}
// Disconnect terminates the peer connection with the given reason.
// It returns immediately and does not wait until the connection is closed.
func (p *Peer) Disconnect(reason DiscReason) {
@ -126,17 +143,20 @@ func (p *Peer) Disconnect(reason DiscReason) {
// String implements fmt.Stringer.
func (p *Peer) String() string {
return fmt.Sprintf("Peer %.8x %v", p.remoteID[:], p.RemoteAddr())
return fmt.Sprintf("Peer %.8x %v", p.remotePublicKey[:], p.RemoteAddr())
}
func newPeer(conn net.Conn, protocols []Protocol, ourName string, ourID, remoteID *discover.NodeID) *Peer {
logtag := fmt.Sprintf("Peer %.8x %v", remoteID[:], conn.RemoteAddr())
func newPeer(conn net.Conn, protocols []Protocol, ourName string, ourID *discover.NodeID, ourPublicKey, remotePublicKey *discover.PublicKey) *Peer {
remoteID := discover.Hash(*remotePublicKey)
logtag := fmt.Sprintf("Peer %.8x %v", remotePublicKey[:], conn.RemoteAddr())
return &Peer{
Logger: logger.NewLogger(logtag),
rw: newFrameRW(conn, msgWriteTimeout),
ourID: ourID,
ourPublicKey: ourPublicKey,
ourName: ourName,
remoteID: remoteID,
ourID: ourID,
remotePublicKey: remotePublicKey,
remoteID: &remoteID,
protocols: protocols,
running: make(map[string]*proto),
disc: make(chan DiscReason),
@ -160,7 +180,7 @@ func (p *Peer) run() DiscReason {
go func() { readErr <- p.readLoop() }()
if !p.noHandshake {
if err := writeProtocolHandshake(p.rw, p.ourName, *p.ourID, p.protocols); err != nil {
if err := writeProtocolHandshake(p.rw, p.ourName, *p.ourPublicKey, p.protocols); err != nil {
p.DebugDetailf("Protocol handshake error: %v\n", err)
p.rw.Close()
return DiscProtocolError
@ -277,8 +297,8 @@ func readProtocolHandshake(p *Peer, rw MsgReadWriter) error {
return newPeerError(errP2PVersionMismatch, "required version %d, received %d\n",
baseProtocolVersion, hs.Version)
}
if hs.NodeID == *p.remoteID {
return newPeerError(errPubkeyForbidden, "node ID mismatch")
if hs.PublicKey == *p.remotePublicKey {
return newPeerError(errPubkeyForbidden, "node PublicKey mismatch")
}
// TODO: remove Caps with empty name
p.setHandshakeInfo(hs.Name, hs.Caps)
@ -286,12 +306,12 @@ func readProtocolHandshake(p *Peer, rw MsgReadWriter) error {
return nil
}
func writeProtocolHandshake(w MsgWriter, name string, id discover.NodeID, ps []Protocol) error {
func writeProtocolHandshake(w MsgWriter, name string, PublicKey discover.PublicKey, ps []Protocol) error {
var caps []interface{}
for _, proto := range ps {
caps = append(caps, proto.cap())
}
return EncodeMsg(w, handshakeMsg, baseProtocolVersion, name, caps, 0, id)
return EncodeMsg(w, handshakeMsg, baseProtocolVersion, name, caps, 0, PublicKey)
}
// startProtocols starts matching named subprotocols.

36
p2p/server.go
View file

@ -89,7 +89,7 @@ type Server struct {
lock sync.RWMutex
running bool
listener net.Listener
peers map[discover.NodeID]*Peer
peers map[discover.PublicKey]*Peer
ntab *discover.Table
@ -99,7 +99,7 @@ type Server struct {
peerConnect chan *discover.Node
}
type handshakeFunc func(io.ReadWriter, *ecdsa.PrivateKey, *discover.Node) (discover.NodeID, []byte, error)
type handshakeFunc func(io.ReadWriter, *ecdsa.PrivateKey, *discover.Node) (discover.PublicKey, []byte, error)
type newPeerHook func(*Peer)
// Peers returns all connected peers.
@ -167,7 +167,7 @@ func (srv *Server) Start() (err error) {
return fmt.Errorf("Server.MaxPeers must be > 0")
}
srv.quit = make(chan struct{})
srv.peers = make(map[discover.NodeID]*Peer)
srv.peers = make(map[discover.PublicKey]*Peer)
srv.peerConnect = make(chan *discover.Node)
if srv.handshakeFunc == nil {
@ -276,7 +276,7 @@ func (srv *Server) dialLoop() {
go srv.findPeers()
dialed := make(chan *discover.Node)
dialing := make(map[discover.NodeID]bool)
dialing := make(map[discover.PublicKey]bool)
// TODO: limit number of active dials
// TODO: ensure only one findPeers goroutine is running
@ -293,13 +293,13 @@ func (srv *Server) dialLoop() {
// there is another check for this in addPeer,
// which runs after the handshake.
srv.lock.Lock()
_, isconnected := srv.peers[dest.ID]
_, isconnected := srv.peers[dest.PublicKey]
srv.lock.Unlock()
if isconnected || dialing[dest.ID] || dest.ID == srv.ntab.Self() {
if isconnected || dialing[dest.PublicKey] || dest.PublicKey == srv.ntab.PublicKey() {
continue
}
dialing[dest.ID] = true
dialing[dest.PublicKey] = true
srv.peerWG.Add(1)
go func() {
srv.dialNode(dest)
@ -309,7 +309,7 @@ func (srv *Server) dialLoop() {
}()
case dest := <-dialed:
delete(dialing, dest.ID)
delete(dialing, dest.PublicKey)
case <-srv.quit:
// TODO: maybe wait for active dials
@ -330,11 +330,11 @@ func (srv *Server) dialNode(dest *discover.Node) {
}
func (srv *Server) findPeers() {
far := srv.ntab.Self()
far := srv.ntab.NodeID()
for i := range far {
far[i] = ^far[i]
}
closeToSelf := srv.ntab.Lookup(srv.ntab.Self())
closeToSelf := srv.ntab.Lookup(srv.ntab.NodeID())
farFromSelf := srv.ntab.Lookup(far)
for i := 0; i < len(closeToSelf) || i < len(farFromSelf); i++ {
@ -350,15 +350,17 @@ func (srv *Server) findPeers() {
func (srv *Server) startPeer(conn net.Conn, dest *discover.Node) {
// TODO: handle/store session token
conn.SetDeadline(time.Now().Add(handshakeTimeout))
remoteID, _, err := srv.handshakeFunc(conn, srv.PrivateKey, dest)
remotePublicKey, _, err := srv.handshakeFunc(conn, srv.PrivateKey, dest)
if err != nil {
conn.Close()
srvlog.Debugf("Encryption Handshake with %v failed: %v", conn.RemoteAddr(), err)
return
}
ourID := srv.ntab.Self()
p := newPeer(conn, srv.Protocols, srv.Name, &ourID, &remoteID)
if ok, reason := srv.addPeer(remoteID, p); !ok {
ourPublicKey := srv.ntab.PublicKey()
ourID := srv.ntab.NodeID()
p := newPeer(conn, srv.Protocols, srv.Name, &ourID, &ourPublicKey, &remotePublicKey)
if ok, reason := srv.addPeer(remotePublicKey, p); !ok {
srvlog.DebugDetailf("Not adding %v (%v)\n", p, reason)
p.politeDisconnect(reason)
return
@ -373,7 +375,7 @@ func (srv *Server) startPeer(conn net.Conn, dest *discover.Node) {
srvlog.Debugf("Removed %v (%v)\n", p, discreason)
}
func (srv *Server) addPeer(id discover.NodeID, p *Peer) (bool, DiscReason) {
func (srv *Server) addPeer(id discover.PublicKey, p *Peer) (bool, DiscReason) {
srv.lock.Lock()
defer srv.lock.Unlock()
switch {
@ -385,7 +387,7 @@ func (srv *Server) addPeer(id discover.NodeID, p *Peer) (bool, DiscReason) {
return false, DiscAlreadyConnected
case srv.Blacklist.Exists(id[:]):
return false, DiscUselessPeer
case id == srv.ntab.Self():
case id == srv.ntab.PublicKey():
return false, DiscSelf
}
srv.peers[id] = p
@ -394,7 +396,7 @@ func (srv *Server) addPeer(id discover.NodeID, p *Peer) (bool, DiscReason) {
func (srv *Server) removePeer(p *Peer) {
srv.lock.Lock()
delete(srv.peers, *p.remoteID)
delete(srv.peers, *p.remotePublicKey)
srv.lock.Unlock()
srv.peerWG.Done()
}

2
xeth/types.go
View file

@ -214,7 +214,7 @@ func NewPeer(peer *p2p.Peer) *Peer {
return &Peer{
ref: peer,
Ip: fmt.Sprintf("%v", peer.RemoteAddr()),
Version: fmt.Sprintf("%v", peer.ID()),
Version: fmt.Sprintf("%v", peer.PublicKey()),
Caps: fmt.Sprintf("%v", caps),
}
}