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

Merge f5f69dff19 into 7299eb72e0

Browse source
This commit is contained in:
Viktor Trón 2015-02-15 03:00:32 +00:00
commit 0c9d8f681a
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 // the main loop that handles incoming messages
// note RemovePeer in the post-disconnect hook // note RemovePeer in the post-disconnect hook
func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
id := peer.ID() id := peer.PublicKey()
self := &ethProtocol{ self := &ethProtocol{
txPool: txPool, txPool: txPool,
chainManager: chainManager, chainManager: chainManager,

13
p2p/crypto.go
View file

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

94
p2p/discover/node.go
View file

@ -1,6 +1,7 @@
package discover package discover
import ( import (
"crypto"
"crypto/ecdsa" "crypto/ecdsa"
"crypto/elliptic" "crypto/elliptic"
"encoding/hex" "encoding/hex"
@ -14,16 +15,28 @@ import (
"strings" "strings"
"time" "time"
ethcrypto "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/crypto/secp256k1" "github.com/ethereum/go-ethereum/crypto/secp256k1"
"github.com/ethereum/go-ethereum/rlp" "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. // Node represents a host on the network.
type Node struct { type Node struct {
ID NodeID PublicKey PublicKey
IP net.IP IP net.IP
NodeID NodeID
DiscPort int // UDP listening port for discovery protocol DiscPort int // UDP listening port for discovery protocol
TCPPort int // TCP listening port for RLPx TCPPort int // TCP listening port for RLPx
@ -31,16 +44,22 @@ type Node struct {
active time.Time active time.Time
} }
func newNode(id NodeID, addr *net.UDPAddr) *Node { func newNode(pubkey PublicKey, id NodeID, addr *net.UDPAddr) *Node {
return &Node{ return &Node{
ID: id, PublicKey: pubkey,
IP: addr.IP, IP: addr.IP,
DiscPort: addr.Port, DiscPort: addr.Port,
TCPPort: addr.Port, TCPPort: addr.Port,
active: time.Now(), active: time.Now(),
NodeID: id,
} }
} }
func Hash(id PublicKey) (a NodeID) {
copy(a[:], ethcrypto.Sha256(id[:]))
return
}
func (n *Node) isValid() bool { func (n *Node) isValid() bool {
// TODO: don't accept localhost, LAN addresses from internet hosts // TODO: don't accept localhost, LAN addresses from internet hosts
return !n.IP.IsMulticast() && !n.IP.IsUnspecified() && n.TCPPort != 0 && n.DiscPort != 0 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} addr := net.TCPAddr{IP: n.IP, Port: n.TCPPort}
u := url.URL{ u := url.URL{
Scheme: "enode", Scheme: "enode",
User: url.User(fmt.Sprintf("%x", n.ID[:])), User: url.User(fmt.Sprintf("%x", n.PublicKey[:])),
Host: addr.String(), Host: addr.String(),
} }
if n.DiscPort != n.TCPPort { if n.DiscPort != n.TCPPort {
@ -65,7 +84,7 @@ func (n *Node) String() string {
// //
// A node URL has scheme "enode". // 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 // 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 // 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 // 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\"") return nil, errors.New("invalid URL scheme, want \"enode\"")
} }
if u.User == nil { 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 { if n.PublicKey, err = HexPublicKey(u.User.String()); err != nil {
return nil, fmt.Errorf("invalid node ID (%v)", err) return nil, fmt.Errorf("invalid node PublicKey (%v)", err)
} }
n.NodeID = Hash(n.PublicKey)
ip, port, err := net.SplitHostPort(u.Host) ip, port, err := net.SplitHostPort(u.Host)
if err != nil { if err != nil {
return nil, fmt.Errorf("invalid host: %v", err) 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 { 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) { func (n *Node) DecodeRLP(s *rlp.Stream) (err error) {
var ext rpcNode var ext rpcNode
if err = s.Decode(&ext); err == nil { if err = s.Decode(&ext); err == nil {
n.TCPPort = int(ext.Port) n.TCPPort = int(ext.Port)
n.DiscPort = 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 { if n.IP = net.ParseIP(ext.IP); n.IP == nil {
return errors.New("invalid IP string") return errors.New("invalid IP string")
} }
@ -135,27 +155,27 @@ func (n *Node) DecodeRLP(s *rlp.Stream) (err error) {
return err 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. // 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. // PublicKey prints as a long hexadecimal number.
func (n NodeID) String() string { func (n PublicKey) String() string {
return fmt.Sprintf("%#x", n[:]) return fmt.Sprintf("%#x", n[:])
} }
// The Go syntax representation of a NodeID is a call to HexID. // The Go syntax representation of a PublicKey is a call to HexPublicKey.
func (n NodeID) GoString() string { func (n PublicKey) GoString() string {
return fmt.Sprintf("discover.HexID(\"%#x\")", n[:]) 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. // The string may be prefixed with 0x.
func HexID(in string) (NodeID, error) { func HexPublicKey(in string) (PublicKey, error) {
if strings.HasPrefix(in, "0x") { if strings.HasPrefix(in, "0x") {
in = in[2:] in = in[2:]
} }
var id NodeID var id PublicKey
b, err := hex.DecodeString(in) b, err := hex.DecodeString(in)
if err != nil { if err != nil {
return id, err return id, err
@ -166,19 +186,19 @@ func HexID(in string) (NodeID, error) {
return id, nil return id, nil
} }
// MustHexID converts a hex string to a NodeID. // MustHexPublicKey converts a hex string to a PublicKey.
// It panics if the string is not a valid NodeID. // It panics if the string is not a valid PublicKey.
func MustHexID(in string) NodeID { func MustHexPublicKey(in string) PublicKey {
id, err := HexID(in) id, err := HexPublicKey(in)
if err != nil { if err != nil {
panic(err) panic(err)
} }
return id return id
} }
// PubkeyID returns a marshaled representation of the given public key. // PublicKey returns a marshaled representation of the given public key.
func PubkeyID(pub *ecdsa.PublicKey) NodeID { func exportPublicKey(pub *ecdsa.PublicKey) PublicKey {
var id NodeID var id PublicKey
pbytes := elliptic.Marshal(pub.Curve, pub.X, pub.Y) pbytes := elliptic.Marshal(pub.Curve, pub.X, pub.Y)
if len(pbytes)-1 != len(id) { if len(pbytes)-1 != len(id) {
panic(fmt.Errorf("need %d bit pubkey, got %d bits", (len(id)+1)*8, len(pbytes))) 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 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. // 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) pubkey, err := secp256k1.RecoverPubkey(hash, sig)
if err != nil { if err != nil {
return id, err return id, err
@ -270,8 +290,8 @@ func logdist(a, b NodeID) int {
return len(a)*8 - lz return len(a)*8 - lz
} }
// randomID returns a random NodeID such that logdist(a, b) == n // randomNodeID returns a random NodeID such that logdist(a, b) == n
func randomID(a NodeID, n int) (b NodeID) { func randomNodeID(a NodeID, n int) (b NodeID) {
if n == 0 { if n == 0 {
return a return a
} }

62
p2p/discover/table.go
View file

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

57
p2p/discover/udp.go
View file

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

72
p2p/peer.go
View file

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

36
p2p/server.go
View file

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

2
xeth/types.go
View file

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