diff --git a/cmd/bootnode/main.go b/cmd/bootnode/main.go index e1734d89ac..ecfc6fc24e 100644 --- a/cmd/bootnode/main.go +++ b/cmd/bootnode/main.go @@ -21,6 +21,7 @@ import ( "crypto/ecdsa" "flag" "fmt" + "net" "os" "github.com/ethereum/go-ethereum/cmd/utils" @@ -96,12 +97,32 @@ func main() { } } + addr, err := net.ResolveUDPAddr("udp", *listenAddr) + if err != nil { + utils.Fatalf("-ResolveUDPAddr: %v", err) + } + conn, err := net.ListenUDP("udp", addr) + if err != nil { + utils.Fatalf("-ListenUDP: %v", err) + } + + realaddr := conn.LocalAddr().(*net.UDPAddr) + if natm != nil { + if !realaddr.IP.IsLoopback() { + go nat.Map(natm, nil, "udp", realaddr.Port, realaddr.Port, "ethereum discovery") + } + // TODO: react to external IP changes over time. + if ext, err := natm.ExternalIP(); err == nil { + realaddr = &net.UDPAddr{IP: ext, Port: realaddr.Port} + } + } + if *runv5 { - if _, err := discv5.ListenUDP(nodeKey, *listenAddr, natm, "", restrictList); err != nil { + if _, err := discv5.ListenUDP(nodeKey, conn, realaddr, "", restrictList); err != nil { utils.Fatalf("%v", err) } } else { - if _, err := discover.ListenUDP(nodeKey, *listenAddr, natm, "", restrictList); err != nil { + if _, err := discover.ListenUDP(nodeKey, conn, realaddr, nil, "", restrictList); err != nil { utils.Fatalf("%v", err) } } diff --git a/cmd/faucet/faucet.go b/cmd/faucet/faucet.go index 328029fdf6..affb7ebdfb 100644 --- a/cmd/faucet/faucet.go +++ b/cmd/faucet/faucet.go @@ -223,7 +223,6 @@ func newFaucet(genesis *core.Genesis, port int, enodes []*discv5.Node, network u NoDiscovery: true, DiscoveryV5: true, ListenAddr: fmt.Sprintf(":%d", port), - DiscoveryV5Addr: fmt.Sprintf(":%d", port+1), MaxPeers: 25, BootstrapNodesV5: enodes, }, diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 30edf199c9..92d4ac4c4c 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -635,14 +635,6 @@ func setListenAddress(ctx *cli.Context, cfg *p2p.Config) { } } -// setDiscoveryV5Address creates a UDP listening address string from set command -// line flags for the V5 discovery protocol. -func setDiscoveryV5Address(ctx *cli.Context, cfg *p2p.Config) { - if ctx.GlobalIsSet(ListenPortFlag.Name) { - cfg.DiscoveryV5Addr = fmt.Sprintf(":%d", ctx.GlobalInt(ListenPortFlag.Name)+1) - } -} - // setNAT creates a port mapper from command line flags. func setNAT(ctx *cli.Context, cfg *p2p.Config) { if ctx.GlobalIsSet(NATFlag.Name) { @@ -793,7 +785,6 @@ func SetP2PConfig(ctx *cli.Context, cfg *p2p.Config) { setNodeKey(ctx, cfg) setNAT(ctx, cfg) setListenAddress(ctx, cfg) - setDiscoveryV5Address(ctx, cfg) setBootstrapNodes(ctx, cfg) setBootstrapNodesV5(ctx, cfg) @@ -829,7 +820,6 @@ func SetP2PConfig(ctx *cli.Context, cfg *p2p.Config) { // --dev mode can't use p2p networking. cfg.MaxPeers = 0 cfg.ListenAddr = ":0" - cfg.DiscoveryV5Addr = ":0" cfg.NoDiscovery = true cfg.DiscoveryV5 = false } diff --git a/les/randselect.go b/common/randselect.go similarity index 80% rename from les/randselect.go rename to common/randselect.go index 1a9d0695bd..3b1ef8c9c5 100644 --- a/les/randselect.go +++ b/common/randselect.go @@ -14,44 +14,43 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -// Package les implements the Light Ethereum Subprotocol. -package les +package common import ( "math/rand" ) // wrsItem interface should be implemented by any entries that are to be selected from -// a weightedRandomSelect set. Note that recalculating monotonously decreasing item -// weights on-demand (without constantly calling update) is allowed +// a WeightedRandomSelect set. Note that recalculating monotonously decreasing item +// weights on-demand (without constantly calling Update) is allowed type wrsItem interface { Weight() int64 } -// weightedRandomSelect is capable of weighted random selection from a set of items -type weightedRandomSelect struct { +// WeightedRandomSelect is capable of weighted random selection from a set of items +type WeightedRandomSelect struct { root *wrsNode idx map[wrsItem]int } -// newWeightedRandomSelect returns a new weightedRandomSelect structure -func newWeightedRandomSelect() *weightedRandomSelect { - return &weightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[wrsItem]int)} +// newWeightedRandomSelect returns a new WeightedRandomSelect structure +func NewWeightedRandomSelect() *WeightedRandomSelect { + return &WeightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[wrsItem]int)} } -// update updates an item's weight, adds it if it was non-existent or removes it if +// Update updates an item's weight, adds it if it was non-existent or removes it if // the new weight is zero. Note that explicitly updating decreasing weights is not necessary. -func (w *weightedRandomSelect) update(item wrsItem) { +func (w *WeightedRandomSelect) Update(item wrsItem) { w.setWeight(item, item.Weight()) } -// remove removes an item from the set -func (w *weightedRandomSelect) remove(item wrsItem) { +// Remove removes an item from the set +func (w *WeightedRandomSelect) Remove(item wrsItem) { w.setWeight(item, 0) } // setWeight sets an item's weight to a specific value (removes it if zero) -func (w *weightedRandomSelect) setWeight(item wrsItem, weight int64) { +func (w *WeightedRandomSelect) setWeight(item wrsItem, weight int64) { idx, ok := w.idx[item] if ok { w.root.setWeight(idx, weight) @@ -72,17 +71,17 @@ func (w *weightedRandomSelect) setWeight(item wrsItem, weight int64) { } } -// choose randomly selects an item from the set, with a chance proportional to its +// Choose randomly selects an item from the set, with a chance proportional to its // current weight. If the weight of the chosen element has been decreased since the // last stored value, returns it with a newWeight/oldWeight chance, otherwise just // updates its weight and selects another one -func (w *weightedRandomSelect) choose() wrsItem { +func (w *WeightedRandomSelect) Choose() wrsItem { for { if w.root.sumWeight == 0 { return nil } val := rand.Int63n(w.root.sumWeight) - choice, lastWeight := w.root.choose(val) + choice, lastWeight := w.root.Choose(val) weight := choice.Weight() if weight != lastWeight { w.setWeight(choice, weight) @@ -156,14 +155,14 @@ func (n *wrsNode) setWeight(idx int, weight int64) int64 { return diff } -// choose recursively selects an item from the tree and returns it along with its weight -func (n *wrsNode) choose(val int64) (wrsItem, int64) { +// Choose recursively selects an item from the tree and returns it along with its weight +func (n *wrsNode) Choose(val int64) (wrsItem, int64) { for i, w := range n.weights { if val < w { if n.level == 0 { return n.items[i].(wrsItem), n.weights[i] } else { - return n.items[i].(*wrsNode).choose(val) + return n.items[i].(*wrsNode).Choose(val) } } else { val -= w diff --git a/les/randselect_test.go b/common/randselect_test.go similarity index 93% rename from les/randselect_test.go rename to common/randselect_test.go index 9ae7726ddd..5b23faa4df 100644 --- a/les/randselect_test.go +++ b/common/randselect_test.go @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -package les +package common import ( "math/rand" @@ -36,15 +36,15 @@ func (t *testWrsItem) Weight() int64 { func TestWeightedRandomSelect(t *testing.T) { testFn := func(cnt int) { - s := newWeightedRandomSelect() + s := NewWeightedRandomSelect() w := -1 list := make([]testWrsItem, cnt) for i := range list { list[i] = testWrsItem{idx: i, widx: &w} - s.update(&list[i]) + s.Update(&list[i]) } w = rand.Intn(cnt) - c := s.choose() + c := s.Choose() if c == nil { t.Errorf("expected item, got nil") } else { @@ -53,7 +53,7 @@ func TestWeightedRandomSelect(t *testing.T) { } } w = -2 - if s.choose() != nil { + if s.Choose() != nil { t.Errorf("expected nil, got item") } } diff --git a/les/distributor.go b/les/distributor.go index 159fa4c73f..eeb42418ad 100644 --- a/les/distributor.go +++ b/les/distributor.go @@ -23,6 +23,8 @@ import ( "errors" "sync" "time" + + "github.com/ethereum/go-ethereum/common" ) // ErrNoPeers is returned if no peers capable of serving a queued request are available @@ -161,7 +163,7 @@ func (d *requestDistributor) loop() { } } -// selectPeerItem represents a peer to be selected for a request by weightedRandomSelect +// selectPeerItem represents a peer to be selected for a request by WeightedRandomSelect type selectPeerItem struct { peer distPeer req *distReq @@ -182,7 +184,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { bestPeer distPeer bestReq *distReq bestWait time.Duration - sel *weightedRandomSelect + sel *common.WeightedRandomSelect ) d.peerLock.RLock() @@ -198,9 +200,9 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { wait, bufRemain := peer.waitBefore(cost) if wait == 0 { if sel == nil { - sel = newWeightedRandomSelect() + sel = common.NewWeightedRandomSelect() } - sel.update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) + sel.Update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) } else { if bestReq == nil || wait < bestWait { bestPeer = peer @@ -220,7 +222,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { } if sel != nil { - c := sel.choose().(selectPeerItem) + c := sel.Choose().(selectPeerItem) return c.peer, c.req, 0 } return bestPeer, bestReq, bestWait diff --git a/les/serverpool.go b/les/serverpool.go index dc1ea6bf02..c7bc6cf131 100644 --- a/les/serverpool.go +++ b/les/serverpool.go @@ -27,6 +27,7 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" @@ -114,7 +115,7 @@ type serverPool struct { adjustStats chan poolStatAdjust knownQueue, newQueue poolEntryQueue - knownSelect, newSelect *weightedRandomSelect + knownSelect, newSelect *common.WeightedRandomSelect knownSelected, newSelected int fastDiscover bool } @@ -129,8 +130,8 @@ func newServerPool(db ethdb.Database, quit chan struct{}, wg *sync.WaitGroup) *s timeout: make(chan *poolEntry, 1), adjustStats: make(chan poolStatAdjust, 100), enableRetry: make(chan *poolEntry, 1), - knownSelect: newWeightedRandomSelect(), - newSelect: newWeightedRandomSelect(), + knownSelect: common.NewWeightedRandomSelect(), + newSelect: common.NewWeightedRandomSelect(), fastDiscover: true, } pool.knownQueue = newPoolEntryQueue(maxKnownEntries, pool.removeEntry) @@ -183,8 +184,8 @@ func (pool *serverPool) connect(p *peer, ip net.IP, port uint16) *poolEntry { entry.lastConnected = addr entry.addr = make(map[string]*poolEntryAddress) entry.addr[addr.strKey()] = addr - entry.addrSelect = *newWeightedRandomSelect() - entry.addrSelect.update(addr) + entry.addrSelect = *common.NewWeightedRandomSelect() + entry.addrSelect.Update(addr) return entry } @@ -352,7 +353,7 @@ func (pool *serverPool) findOrNewNode(id discover.NodeID, ip net.IP, port uint16 entry = &poolEntry{ id: id, addr: make(map[string]*poolEntryAddress), - addrSelect: *newWeightedRandomSelect(), + addrSelect: *common.NewWeightedRandomSelect(), shortRetry: shortRetryCnt, } pool.entries[id] = entry @@ -373,7 +374,7 @@ func (pool *serverPool) findOrNewNode(id discover.NodeID, ip net.IP, port uint16 entry.addr[addr.strKey()] = addr } addr.lastSeen = now - entry.addrSelect.update(addr) + entry.addrSelect.Update(addr) if !entry.known { pool.newQueue.setLatest(entry) } @@ -400,7 +401,7 @@ func (pool *serverPool) loadNodes() { "timeout", fmt.Sprintf("%v/%v", e.timeoutStats.avg, e.timeoutStats.weight)) pool.entries[e.id] = e pool.knownQueue.setLatest(e) - pool.knownSelect.update((*knownEntry)(e)) + pool.knownSelect.Update((*knownEntry)(e)) } } @@ -421,8 +422,8 @@ func (pool *serverPool) saveNodes() { // Note that it is called by the new/known queues from which the entry has already // been removed so removing it from the queues is not necessary. func (pool *serverPool) removeEntry(entry *poolEntry) { - pool.newSelect.remove((*discoveredEntry)(entry)) - pool.knownSelect.remove((*knownEntry)(entry)) + pool.newSelect.Remove((*discoveredEntry)(entry)) + pool.knownSelect.Remove((*knownEntry)(entry)) entry.removed = true delete(pool.entries, entry.id) } @@ -451,8 +452,8 @@ func (pool *serverPool) setRetryDial(entry *poolEntry) { // updateCheckDial is called when an entry can potentially be dialed again. It updates // its selection weights and checks if new dials can/should be made. func (pool *serverPool) updateCheckDial(entry *poolEntry) { - pool.newSelect.update((*discoveredEntry)(entry)) - pool.knownSelect.update((*knownEntry)(entry)) + pool.newSelect.Update((*discoveredEntry)(entry)) + pool.knownSelect.Update((*knownEntry)(entry)) pool.checkDial() } @@ -461,7 +462,7 @@ func (pool *serverPool) updateCheckDial(entry *poolEntry) { func (pool *serverPool) checkDial() { fillWithKnownSelects := !pool.fastDiscover for pool.knownSelected < targetKnownSelect { - entry := pool.knownSelect.choose() + entry := pool.knownSelect.Choose() if entry == nil { fillWithKnownSelects = false break @@ -469,7 +470,7 @@ func (pool *serverPool) checkDial() { pool.dial((*poolEntry)(entry.(*knownEntry)), true) } for pool.knownSelected+pool.newSelected < targetServerCount { - entry := pool.newSelect.choose() + entry := pool.newSelect.Choose() if entry == nil { break } @@ -480,7 +481,7 @@ func (pool *serverPool) checkDial() { // is over, we probably won't find more in the near future so select more // known entries if possible for pool.knownSelected < targetServerCount { - entry := pool.knownSelect.choose() + entry := pool.knownSelect.Choose() if entry == nil { break } @@ -501,7 +502,7 @@ func (pool *serverPool) dial(entry *poolEntry, knownSelected bool) { } else { pool.newSelected++ } - addr := entry.addrSelect.choose().(*poolEntryAddress) + addr := entry.addrSelect.Choose().(*poolEntryAddress) log.Debug("Dialing new peer", "lesaddr", entry.id.String()+"@"+addr.strKey(), "set", len(entry.addr), "known", knownSelected) entry.dialed = addr go func() { @@ -548,7 +549,7 @@ type poolEntry struct { id discover.NodeID addr map[string]*poolEntryAddress lastConnected, dialed *poolEntryAddress - addrSelect weightedRandomSelect + addrSelect common.WeightedRandomSelect lastDiscovered mclock.AbsTime known, knownSelected bool @@ -582,8 +583,8 @@ func (e *poolEntry) DecodeRLP(s *rlp.Stream) error { e.id = entry.ID e.addr = make(map[string]*poolEntryAddress) e.addr[addr.strKey()] = addr - e.addrSelect = *newWeightedRandomSelect() - e.addrSelect.update(addr) + e.addrSelect = *common.NewWeightedRandomSelect() + e.addrSelect.Update(addr) e.lastConnected = addr e.connectStats = entry.CStat e.delayStats = entry.DStat diff --git a/mobile/geth.go b/mobile/geth.go index 7b39faadec..7e3b8f4915 100644 --- a/mobile/geth.go +++ b/mobile/geth.go @@ -116,7 +116,6 @@ func NewNode(datadir string, config *NodeConfig) (stack *Node, _ error) { P2P: p2p.Config{ NoDiscovery: true, DiscoveryV5: true, - DiscoveryV5Addr: ":0", BootstrapNodesV5: config.BootstrapNodes.nodes, ListenAddr: ":0", NAT: nat.Any(), diff --git a/node/defaults.go b/node/defaults.go index 848f08e05c..d4e1486834 100644 --- a/node/defaults.go +++ b/node/defaults.go @@ -41,10 +41,9 @@ var DefaultConfig = Config{ WSPort: DefaultWSPort, WSModules: []string{"net", "web3"}, P2P: p2p.Config{ - ListenAddr: ":30303", - DiscoveryV5Addr: ":30304", - MaxPeers: 25, - NAT: nat.Any(), + ListenAddr: ":30303", + MaxPeers: 25, + NAT: nat.Any(), }, } diff --git a/p2p/discover/udp.go b/p2p/discover/udp.go index f9eb99ee36..60436952d8 100644 --- a/p2p/discover/udp.go +++ b/p2p/discover/udp.go @@ -210,17 +210,15 @@ type reply struct { matched chan<- bool } +// ReadPacket is sent to the unhandled channel when it could not be processed +type ReadPacket struct { + Data []byte + Addr *net.UDPAddr +} + // ListenUDP returns a new table that listens for UDP packets on laddr. -func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, nodeDBPath string, netrestrict *netutil.Netlist) (*Table, error) { - addr, err := net.ResolveUDPAddr("udp", laddr) - if err != nil { - return nil, err - } - conn, err := net.ListenUDP("udp", addr) - if err != nil { - return nil, err - } - tab, _, err := newUDP(priv, conn, natm, nodeDBPath, netrestrict) +func ListenUDP(priv *ecdsa.PrivateKey, conn conn, realaddr *net.UDPAddr, unhandled chan ReadPacket, nodeDBPath string, netrestrict *netutil.Netlist) (*Table, error) { + tab, _, err := newUDP(priv, conn, realaddr, unhandled, nodeDBPath, netrestrict) if err != nil { return nil, err } @@ -228,7 +226,7 @@ func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, nodeDBP return tab, nil } -func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, nodeDBPath string, netrestrict *netutil.Netlist) (*Table, *udp, error) { +func newUDP(priv *ecdsa.PrivateKey, c conn, realaddr *net.UDPAddr, unhandled chan ReadPacket, nodeDBPath string, netrestrict *netutil.Netlist) (*Table, *udp, error) { udp := &udp{ conn: c, priv: priv, @@ -237,16 +235,6 @@ func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, nodeDBPath strin gotreply: make(chan reply), addpending: make(chan *pending), } - realaddr := c.LocalAddr().(*net.UDPAddr) - if natm != nil { - if !realaddr.IP.IsLoopback() { - go nat.Map(natm, udp.closing, "udp", realaddr.Port, realaddr.Port, "ethereum discovery") - } - // TODO: react to external IP changes over time. - if ext, err := natm.ExternalIP(); err == nil { - realaddr = &net.UDPAddr{IP: ext, Port: realaddr.Port} - } - } // TODO: separate TCP port udp.ourEndpoint = makeEndpoint(realaddr, uint16(realaddr.Port)) tab, err := newTable(udp, PubkeyID(&priv.PublicKey), realaddr, nodeDBPath) @@ -256,7 +244,7 @@ func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, nodeDBPath strin udp.Table = tab go udp.loop() - go udp.readLoop() + go udp.readLoop(unhandled) return udp.Table, udp, nil } @@ -492,8 +480,11 @@ func encodePacket(priv *ecdsa.PrivateKey, ptype byte, req interface{}) ([]byte, } // readLoop runs in its own goroutine. it handles incoming UDP packets. -func (t *udp) readLoop() { +func (t *udp) readLoop(unhandled chan ReadPacket) { defer t.conn.Close() + if unhandled != nil { + defer close(unhandled) + } // Discovery packets are defined to be no larger than 1280 bytes. // Packets larger than this size will be cut at the end and treated // as invalid because their hash won't match. @@ -509,7 +500,12 @@ func (t *udp) readLoop() { log.Debug("UDP read error", "err", err) return } - t.handlePacket(from, buf[:nbytes]) + if t.handlePacket(from, buf[:nbytes]) != nil && unhandled != nil { + select { + case unhandled <- ReadPacket{buf[:nbytes], from}: + default: + } + } } } diff --git a/p2p/discv5/encrypt.go b/p2p/discv5/encrypt.go new file mode 100644 index 0000000000..9b87978e3b --- /dev/null +++ b/p2p/discv5/encrypt.go @@ -0,0 +1,196 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package discv5 + +import ( + "bytes" + "crypto/aes" + "crypto/cipher" + "crypto/ecdsa" + crand "crypto/rand" + "encoding/binary" + "math/rand" + "sync" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/crypto/ecies" +) + +type symmEncryption interface { + encode(packet []byte) []byte + decode(encPacket []byte) []byte + maxDecodedLength() int +} + +type Aes256Encryption struct { + blockCipher cipher.Block + randGen *rand.Rand + randLock sync.Mutex + maxLength int +} + +func newEcdhAes256Encryption(privKey *ecdsa.PrivateKey, pubKey *ecdsa.PublicKey, maxEncodedLength int) (*Aes256Encryption, error) { + x, _ := crypto.S256().ScalarMult(pubKey.X, pubKey.Y, privKey.D.Bytes()) + key := x.Bytes() + return newAes256Encryption(key, maxEncodedLength) +} + +func newAes256Encryption(key []byte, maxEncodedLength int) (*Aes256Encryption, error) { + if len(key) != 32 { + panic(nil) + } + if cipher, err := aes.NewCipher(key); err == nil { + var seedArr [8]byte + crand.Read(seedArr[:]) + seed := int64(binary.BigEndian.Uint64(seedArr[:])) + randGen := rand.New(rand.NewSource(seed)) + return &Aes256Encryption{blockCipher: cipher, randGen: randGen, maxLength: maxEncodedLength}, nil + } else { + return nil, err + } +} + +const ( + cipherIvLength = 16 + maxPadding = 32 + tailSize = 8 +) + +func (e *Aes256Encryption) encode(packet []byte) []byte { + length := len(packet) + maxpad := e.maxLength - length - cipherIvLength - tailSize + if maxpad < 1 { + // packet is too large, should be checked by caller + panic(nil) + } + if maxpad > maxPadding { + maxpad = maxPadding + } + e.randLock.Lock() + padding := e.randGen.Intn(maxpad) + 1 + encLength := cipherIvLength + padding + length + tailSize + dest := make([]byte, encLength) + e.randGen.Read(dest[:cipherIvLength]) + dest[cipherIvLength] = byte(padding - 1) + if padding > 1 { + e.randGen.Read(dest[cipherIvLength+1 : cipherIvLength+padding]) + } + e.randLock.Unlock() + copy(dest[cipherIvLength+padding:encLength-tailSize], packet) + integrityHash := crypto.Keccak256(dest[cipherIvLength : encLength-tailSize]) + copy(dest[encLength-tailSize:], integrityHash[:tailSize]) + cipher.NewCFBEncrypter(e.blockCipher, dest[:cipherIvLength]).XORKeyStream(dest[cipherIvLength:], dest[cipherIvLength:]) + return dest +} + +func (e *Aes256Encryption) decode(encPacket []byte) []byte { + paddedLength := len(encPacket) - cipherIvLength + dest := make([]byte, paddedLength) + cipher.NewCFBDecrypter(e.blockCipher, encPacket[:cipherIvLength]).XORKeyStream(dest, encPacket[cipherIvLength:]) + padding := int(dest[0]) + 1 + if padding > paddedLength-tailSize { + return nil + } + // check packet integrity and reject if tail does not match hash + if !bytes.Equal(dest[paddedLength-tailSize:], crypto.Keccak256(dest[:paddedLength-tailSize])[:tailSize]) { + return nil + } + return dest[padding : paddedLength-tailSize] +} + +func (e *Aes256Encryption) maxDecodedLength() int { + return e.maxLength - cipherIvLength - tailSize - 1 +} + +const ( + rpMinLength = 100 + rpMaxLength = 1000 +) + +func newReconnectSeedAndHash() (int64, common.Hash) { + var seedArr [8]byte + crand.Read(seedArr[:]) + seed := int64(binary.BigEndian.Uint64(seedArr[:])) + hash := crypto.Keccak256Hash(reconnectPacket(seed)) + return seed, hash +} + +func reconnectPacket(seed int64) []byte { + r := rand.New(rand.NewSource(seed)) + length := rpMinLength + r.Intn(rpMaxLength-rpMinLength+1) + rp := make([]byte, length) + r.Read(rp) + return rp +} + +type asymmEncryption interface { + encode(packet []byte, pubKey *ecdsa.PublicKey) ([]byte, error) // will receive an ENR record instead of an ECDSA pubkey + decode(encPacket []byte) []byte + maxDecodedLength() int +} + +type EciesEncryption struct { + privKey *ecies.PrivateKey + randGen *rand.Rand + randLock sync.Mutex + maxEncLength, maxDecLength int +} + +func newEciesEncryption(privKey *ecdsa.PrivateKey, maxEncodedLength int) *EciesEncryption { + var seedArr [8]byte + crand.Read(seedArr[:]) + seed := int64(binary.BigEndian.Uint64(seedArr[:])) + randGen := rand.New(rand.NewSource(seed)) + privateKey := ecies.ImportECDSA(privKey) + testEnc, err := ecies.Encrypt(randGen, &privateKey.PublicKey, []byte{42}, nil, nil) + if err != nil { + panic(err) + } + maxDecodedLength := maxEncodedLength - len(testEnc) + 1 + + return &EciesEncryption{ + privKey: privateKey, + randGen: randGen, + maxEncLength: maxEncodedLength, + maxDecLength: maxDecodedLength, + } +} + +func (e *EciesEncryption) encode(packet []byte, pubKey *ecdsa.PublicKey) ([]byte, error) { + if len(packet) > e.maxDecLength { + panic(nil) + } + //TODO add random padding + enc, err := ecies.Encrypt(e.randGen, ecies.ImportECDSAPublic(pubKey), packet, nil, nil) + if len(enc) > e.maxEncLength { + panic(nil) + } + return enc, err +} + +func (e *EciesEncryption) decode(encPacket []byte) []byte { + dec, err := e.privKey.Decrypt(e.randGen, encPacket, nil, nil) + if err != nil { + return nil + } + return dec +} + +func (e *EciesEncryption) maxDecodedLength() int { + return e.maxDecLength +} diff --git a/p2p/discv5/encrypt_test.go b/p2p/discv5/encrypt_test.go new file mode 100644 index 0000000000..2436533b0f --- /dev/null +++ b/p2p/discv5/encrypt_test.go @@ -0,0 +1,104 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package discv5 + +import ( + "bytes" + "crypto/ecdsa" + "crypto/rand" + "testing" + + "github.com/ethereum/go-ethereum/crypto" +) + +const testMaxPacketLen = 1000 + +func testPacket(t *testing.T, encA, encB symmEncryption, packetLen int) { + packet := make([]byte, packetLen) + rand.Read(packet[:]) + enc := encA.encode(packet) + if len(enc) > testMaxPacketLen { + t.Errorf("Encoded packet is too long") + } + dec := encB.decode(enc) + if !bytes.Equal(packet, dec) { + t.Errorf("Decoded packet does not match original (packet = %x size = %d enc = %x encSize = %d dec = %x decSize = %d)", packet, len(packet), enc, len(enc), dec, len(dec)) + } +} + +func TestEcdhAes256Encryption(t *testing.T) { + privKeyA, err := ecdsa.GenerateKey(crypto.S256(), rand.Reader) + if err != nil { + panic(err) + } + pubKeyA := &privKeyA.PublicKey + privKeyB, err := ecdsa.GenerateKey(crypto.S256(), rand.Reader) + if err != nil { + panic(err) + } + pubKeyB := &privKeyB.PublicKey + + encA, err := newEcdhAes256Encryption(privKeyA, pubKeyB, testMaxPacketLen) + if err != nil { + panic(err) + } + encB, err := newEcdhAes256Encryption(privKeyB, pubKeyA, testMaxPacketLen) + if err != nil { + panic(err) + } + + maxDecLen := encA.maxDecodedLength() + for i := 0; i <= maxDecLen; i++ { + testPacket(t, encA, encB, i) + testPacket(t, encB, encA, i) + } +} + +func testPacketAsymm(t *testing.T, encA, encB asymmEncryption, pubKeyB *ecdsa.PublicKey, packetLen int) { + packet := make([]byte, packetLen) + rand.Read(packet[:]) + enc, _ := encA.encode(packet, pubKeyB) + if len(enc) > testMaxPacketLen { + t.Errorf("Encoded packet is too long") + } + dec := encB.decode(enc) + if !bytes.Equal(packet, dec) { + t.Errorf("Decoded packet does not match original (packet = %x size = %d enc = %x encSize = %d dec = %x decSize = %d)", packet, len(packet), enc, len(enc), dec, len(dec)) + } +} + +func TestEciesEncryption(t *testing.T) { //TODO fix this + privKeyA, err := ecdsa.GenerateKey(crypto.S256(), rand.Reader) + if err != nil { + panic(err) + } + pubKeyA := &privKeyA.PublicKey + privKeyB, err := ecdsa.GenerateKey(crypto.S256(), rand.Reader) + if err != nil { + panic(err) + } + pubKeyB := &privKeyB.PublicKey + + encA := newEciesEncryption(privKeyA, testMaxPacketLen) + encB := newEciesEncryption(privKeyB, testMaxPacketLen) + + maxDecLen := encA.maxDecodedLength() + for i := 0; i <= maxDecLen; i++ { + testPacketAsymm(t, encA, encB, pubKeyB, i) + testPacketAsymm(t, encB, encA, pubKeyA, i) + } +} diff --git a/p2p/discv5/net.go b/p2p/discv5/net.go index a39cfcc645..94cdf2c554 100644 --- a/p2p/discv5/net.go +++ b/p2p/discv5/net.go @@ -29,7 +29,6 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto/sha3" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/p2p/nat" "github.com/ethereum/go-ethereum/p2p/netutil" "github.com/ethereum/go-ethereum/rlp" ) @@ -141,7 +140,7 @@ type timeoutEvent struct { node *Node } -func newNetwork(conn transport, ourPubkey ecdsa.PublicKey, natm nat.Interface, dbPath string, netrestrict *netutil.Netlist) (*Network, error) { +func newNetwork(conn transport, ourPubkey ecdsa.PublicKey, dbPath string, netrestrict *netutil.Netlist) (*Network, error) { ourID := PubkeyID(&ourPubkey) var db *nodeDB @@ -431,17 +430,18 @@ loop: //fmt.Println("read", pkt.ev) debugLog("<-net.read") n := net.internNode(&pkt) - prestate := n.state - status := "ok" - if err := net.handle(n, pkt.ev, &pkt); err != nil { - status = err.Error() + if n.serialReplayFilter.accept(pkt.serialNo) { + prestate := n.state + status := "ok" + if err := net.handle(n, pkt.ev, &pkt); err != nil { + status = err.Error() + } + log.Trace("", "msg", log.Lazy{Fn: func() string { + return fmt.Sprintf("<<< (%d) %v from %x@%v: %v -> %v (%v)", + net.tab.count, pkt.ev, pkt.remoteID[:8], pkt.remoteAddr, prestate, n.state, status) + }}) + // TODO: persist state if n.state goes >= known, delete if it goes <= known } - log.Trace("", "msg", log.Lazy{Fn: func() string { - return fmt.Sprintf("<<< (%d) %v from %x@%v: %v -> %v (%v)", - net.tab.count, pkt.ev, pkt.remoteID[:8], pkt.remoteAddr, prestate, n.state, status) - }}) - // TODO: persist state if n.state goes >= known, delete if it goes <= known - // State transition timeouts. case timeout := <-net.timeout: debugLog("<-net.timeout") @@ -722,7 +722,7 @@ func (net *Network) internNode(pkt *ingressPacket) *Node { n.TCP = uint16(pkt.remoteAddr.Port) return n } - n := NewNode(pkt.remoteID, pkt.remoteAddr.IP, uint16(pkt.remoteAddr.Port), uint16(pkt.remoteAddr.Port)) + n := pkt.newNode n.state = unknown net.nodes[pkt.remoteID] = n return n @@ -781,6 +781,7 @@ type nodeNetGuts struct { deferredQueries []*findnodeQuery // queries that can't be sent yet pendingNeighbours *findnodeQuery // current query, waiting for reply queryTimeouts int + serialReplayFilter serialReplayFilter } func (n *nodeNetGuts) deferQuery(q *findnodeQuery) { @@ -1215,9 +1216,10 @@ func (net *Network) handleQueryEvent(n *Node, ev nodeEvent, pkt *ingressPacket) func (net *Network) checkTopicRegister(data *topicRegister) (*pong, error) { var pongpkt ingressPacket - if err := decodePacket(data.Pong, &pongpkt); err != nil { + panic(nil) + /*if err := decodePacket(data.Pong, &pongpkt); err != nil { return nil, err - } + }*/ if pongpkt.ev != pongPacket { return nil, errors.New("is not pong packet") } diff --git a/p2p/discv5/net_test.go b/p2p/discv5/net_test.go index bd234f5ba6..369282ca9c 100644 --- a/p2p/discv5/net_test.go +++ b/p2p/discv5/net_test.go @@ -28,7 +28,7 @@ import ( func TestNetwork_Lookup(t *testing.T) { key, _ := crypto.GenerateKey() - network, err := newNetwork(lookupTestnet, key.PublicKey, nil, "", nil) + network, err := newNetwork(lookupTestnet, key.PublicKey, "", nil) if err != nil { t.Fatal(err) } diff --git a/p2p/discv5/node.go b/p2p/discv5/node.go index 2db7a508f0..c001775237 100644 --- a/p2p/discv5/node.go +++ b/p2p/discv5/node.go @@ -45,6 +45,8 @@ type Node struct { // These fields are not supposed to be used off the // Network.loop goroutine. nodeNetGuts + + nodeUDPfields } // NewNode creates a new node. It is mostly meant to be used for @@ -431,3 +433,36 @@ func hashAtDistance(a common.Hash, n int) (b common.Hash) { } return b } + +// serialReplayFilter belongs to known connections and filters decrypted incoming +// packets by serial number. Since UDP does not guarantee to keep the ordering of +// packets, it does not require serial numbers to arrive in a strictly monotonic +// order. bitMask & (2**(highest-serialNo)) is set if serialNo has already been +// received. Serials older than highest-63 are always rejected. +// Note: a new introduction packet can reset the sender's serial number and the +// recipient should accept it even if it still remembers the sender. +type serialReplayFilter struct { + highest, bitMask uint64 +} + +func (f *serialReplayFilter) accept(sn uint64) bool { + if sn > f.highest { + shift := sn - f.highest + if shift < 64 { + f.bitMask = (f.bitMask << shift) + 1 + } else { + f.bitMask = 1 + } + f.highest = sn + return true + } + shift := f.highest - sn + if shift < 64 { + bit := (uint64(1) << shift) + if (f.bitMask & bit) == 0 { + f.bitMask += bit + return true + } + } + return false +} diff --git a/p2p/discv5/pow.go b/p2p/discv5/pow.go new file mode 100644 index 0000000000..08ad02f593 --- /dev/null +++ b/p2p/discv5/pow.go @@ -0,0 +1,207 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package discv5 + +import ( + "encoding/binary" + "math/rand" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/crypto" +) + +const powSize = 8 + +type pow interface { + valid(packetHash common.Hash) bool +} + +type simplePoW struct { + compare uint64 +} + +func newSimplePoW(difficulty float64) *simplePoW { + compare := ^uint64(0) + if difficulty > 1 { + compare = uint64(float64(compare) / difficulty) + } + return &simplePoW{compare} +} + +func (s *simplePoW) valid(packetHash common.Hash) bool { + return binary.BigEndian.Uint64(packetHash[0:powSize]) <= s.compare +} + +func findPoW(targetHash common.Hash, packet []byte, pow pow, maxCount int) bool { + hashBytes := targetHash.Bytes() + data := append(hashBytes, packet...) + nonceBytes := data[len(hashBytes) : len(hashBytes)+powSize] + rand.Read(nonceBytes) + nonce := binary.BigEndian.Uint64(nonceBytes) + for i := 0; i < maxCount; i++ { + packetHash := crypto.Keccak256Hash(data) + if pow.valid(packetHash) { + binary.BigEndian.PutUint64(packet[:powSize], nonce) + return true + } + nonce++ + binary.BigEndian.PutUint64(nonceBytes, nonce) + } + return false +} + +// powRequest represents a PoW to be calculated for an outgoing message +// PoWs are processed by powProcessor and are selected by common.WeightedRandomSelect +// for processing (powRequest implements wrsItem). +type powRequest struct { + targetHash common.Hash + packet []byte + pow pow + weight int64 + done chan bool + // these fields are set by the processor + timeout mclock.AbsTime + next *powRequest +} + +func (p *powRequest) Weight() int64 { + return p.weight +} + +const ( + powQueueTimeout = time.Second * 10 + powTryCount = 1000000 + powCpuRatio = 0.1 +) + +// powProcessor starts a global processing loop for PoWs that ensures that only a +// certain percentage of a single CPU's time is assigned for PoW search globally +func powProcessor() chan *powRequest { + wrs := common.NewWeightedRandomSelect() + powCh := make(chan *powRequest, 100) + go func() { + var ( + first, last *powRequest + removeFirst, processNext <-chan time.Time + ) + + for { + select { + case pr, ok := <-powCh: + if !ok { + return + } + wrs.Update(pr) + pr.timeout = mclock.Now() + mclock.AbsTime(powQueueTimeout) + if first == nil { + first = pr + removeFirst = time.After(powQueueTimeout) + } + if last != nil { + last.next = pr + } + last = pr + if processNext == nil { + processNext = time.After(0) + } + case <-removeFirst: + wrs.Remove(first) + select { + case first.done <- false: + default: + } + first = first.next + if first != nil { + removeFirst = time.After(time.Duration(first.timeout - mclock.Now())) + } + case <-processNext: + p := wrs.Choose() + if p != nil { + pr := p.(*powRequest) + start := mclock.Now() + if findPoW(pr.targetHash, pr.packet, pr.pow, powTryCount) { + wrs.Remove(pr) + select { + case pr.done <- true: + default: + } + } + d := time.Duration(mclock.Now() - start) + processNext = time.After(d * (1/powCpuRatio - 1)) + } + } + } + }() + return powCh +} + +// hashReplayFilter rejects replayed packets by packet hash, remembering only the +// recent received packet hashes. Intro packets are filtered by hash after checking +// their PoW. +// Note: general packets are also filtered by hash first even though they are later +// filtered by the node specific serial filter too in order to avoid decryption costs +// in case of packet resending. This is realized with a separate instance of +// hashReplayFilter so that processed intro packets are remembered for as long as possible. +type hashReplayFilter struct { + indexToHash map[uint64]common.Hash + hashToIndex map[common.Hash]uint64 + nextIndex, deleteIndex uint64 +} + +const hashReplayFilterSize = 10000 + +func newHashReplayFilter() *hashReplayFilter { + return &hashReplayFilter{ + indexToHash: make(map[uint64]common.Hash), + hashToIndex: make(map[common.Hash]uint64), + } +} + +func (f *hashReplayFilter) accept(hash common.Hash) bool { + if oldIndex, ok := f.hashToIndex[hash]; ok { + // pow already known, move to the front of the queue and reject + if f.nextIndex != oldIndex { + f.hashToIndex[hash] = f.nextIndex + delete(f.indexToHash, oldIndex) + f.indexToHash[f.nextIndex] = hash + f.nextIndex++ + } + return false + } + // pow not seen recently, add to the front of the queue and accept + f.hashToIndex[hash] = f.nextIndex + f.indexToHash[f.nextIndex] = hash + f.nextIndex++ + // delete least recently received hash if entry count has reached the limit + if len(f.indexToHash) > hashReplayFilterSize { + for { + if hash, ok := f.indexToHash[f.deleteIndex]; ok { + delete(f.indexToHash, f.deleteIndex) + delete(f.hashToIndex, hash) + f.deleteIndex++ + break + } + f.deleteIndex++ + if f.deleteIndex >= f.nextIndex { + panic(nil) + } + } + } + return true +} diff --git a/p2p/discv5/pow_test.go b/p2p/discv5/pow_test.go new file mode 100644 index 0000000000..593d7d9abb --- /dev/null +++ b/p2p/discv5/pow_test.go @@ -0,0 +1,25 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package discv5 + +import ( + "testing" +) + +func TestSimplePoW(t *testing.T) { + //TODO write this +} diff --git a/p2p/discv5/sim_test.go b/p2p/discv5/sim_test.go index bf57872e2d..543faecd48 100644 --- a/p2p/discv5/sim_test.go +++ b/p2p/discv5/sim_test.go @@ -282,7 +282,7 @@ func (s *simulation) launchNode(log bool) *Network { addr := &net.UDPAddr{IP: ip, Port: 30303} transport := &simTransport{joinTime: time.Now(), sender: id, senderAddr: addr, sim: s, priv: key} - net, err := newNetwork(transport, key.PublicKey, nil, "", nil) + net, err := newNetwork(transport, key.PublicKey, "", nil) if err != nil { panic("cannot launch new node: " + err.Error()) } diff --git a/p2p/discv5/ticket.go b/p2p/discv5/ticket.go index 48dd114f06..fa52254776 100644 --- a/p2p/discv5/ticket.go +++ b/p2p/discv5/ticket.go @@ -643,7 +643,7 @@ func (s *ticketStore) gotTopicNodes(from *Node, hash common.Hash, nodes []rpcNod if ip.IsUnspecified() || ip.IsLoopback() { ip = from.IP } - n := NewNode(node.ID, ip, node.UDP-1, node.TCP-1) // subtract one from port while discv5 is running in test mode on UDPport+1 + n := NewNode(node.ID, ip, node.UDP, node.TCP) select { case chn <- n: default: diff --git a/p2p/discv5/udp.go b/p2p/discv5/udp.go index 26087cd8e5..d8b86711cc 100644 --- a/p2p/discv5/udp.go +++ b/p2p/discv5/udp.go @@ -19,6 +19,7 @@ package discv5 import ( "bytes" "crypto/ecdsa" + "encoding/binary" "errors" "fmt" "net" @@ -44,6 +45,8 @@ var ( errTimeout = errors.New("RPC timeout") errClockWarp = errors.New("reply deadline too far in the future") errClosed = errors.New("socket closed") + errDecryptFailed = errors.New("decryption failed") + errPacketReplay = errors.New("packet replay") ) // Timeouts @@ -59,6 +62,30 @@ const ( // RPC request structures type ( + reconn struct{} + + newping struct { + TimeStamp uint64 + } // will be renamed to ping + + update struct { + From rpcEndpoint // will be replaced by ENR + ID NodeID // will be replaced by ENR + ReconnectHash common.Hash + TimeStamp uint64 + } + + ack struct { + To rpcEndpoint + ReplyTo common.Hash // packetHash of ping/update/intro/reconn packet + TimeStamp uint64 + } + + accept struct { + ack + update + } + ping struct { Version uint From, To rpcEndpoint @@ -146,9 +173,11 @@ type ( ) const ( - macSize = 256 / 8 - sigSize = 520 / 8 - headSize = macSize + sigSize // space of packet frame data + macSize = 256 / 8 + sigSize = 520 / 8 + headSize = macSize + sigSize // space of packet frame data + introPoWdiff = 1000000 + decryptPoWdiff = 10 ) // Neighbors replies are sent across multiple packets to @@ -218,6 +247,8 @@ type ingressPacket struct { hash []byte data interface{} // one of the RPC structs rawData []byte + newNode *Node + serialNo uint64 } type conn interface { @@ -227,6 +258,10 @@ type conn interface { LocalAddr() net.Addr } +type nodeUDPfields struct { + symmEncryption symmEncryption +} + // udp implements the RPC protocol. type udp struct { conn conn @@ -234,15 +269,22 @@ type udp struct { ourEndpoint rpcEndpoint nat nat.Interface net *Network + + addressLookup map[string]*Node + rpHashLookup map[common.Hash]*Node + introPow, decryptPow pow + powProcessCh chan *powRequest + introHashFilter, generalHashFilter *hashReplayFilter + asymmEncryption asymmEncryption } // ListenUDP returns a new table that listens for UDP packets on laddr. -func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, nodeDBPath string, netrestrict *netutil.Netlist) (*Network, error) { - transport, err := listenUDP(priv, laddr) +func ListenUDP(priv *ecdsa.PrivateKey, conn conn, realaddr *net.UDPAddr, nodeDBPath string, netrestrict *netutil.Netlist) (*Network, error) { + transport, err := listenUDP(priv, conn, realaddr) if err != nil { return nil, err } - net, err := newNetwork(transport, priv.PublicKey, natm, nodeDBPath, netrestrict) + net, err := newNetwork(transport, priv.PublicKey, nodeDBPath, netrestrict) if err != nil { return nil, err } @@ -251,16 +293,20 @@ func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, nodeDBP return net, nil } -func listenUDP(priv *ecdsa.PrivateKey, laddr string) (*udp, error) { - addr, err := net.ResolveUDPAddr("udp", laddr) - if err != nil { - return nil, err - } - conn, err := net.ListenUDP("udp", addr) - if err != nil { - return nil, err - } - return &udp{conn: conn, priv: priv, ourEndpoint: makeEndpoint(addr, uint16(addr.Port))}, nil +func listenUDP(priv *ecdsa.PrivateKey, conn conn, realaddr *net.UDPAddr) (*udp, error) { + return &udp{ + conn: conn, + priv: priv, + ourEndpoint: makeEndpoint(realaddr, uint16(realaddr.Port)), + addressLookup: make(map[string]*Node), + rpHashLookup: make(map[common.Hash]*Node), + introPow: newSimplePoW(introPoWdiff), + decryptPow: newSimplePoW(decryptPoWdiff), + powProcessCh: powProcessor(), + introHashFilter: newHashReplayFilter(), + generalHashFilter: newHashReplayFilter(), + asymmEncryption: newEciesEncryption(priv, 1280), + }, nil } func (t *udp) localAddr() *net.UDPAddr { @@ -269,6 +315,7 @@ func (t *udp) localAddr() *net.UDPAddr { func (t *udp) Close() { t.conn.Close() + close(t.powProcessCh) } func (t *udp) send(remote *Node, ptype nodeEvent, data interface{}) (hash []byte) { @@ -405,7 +452,7 @@ func (t *udp) readLoop() { func (t *udp) handlePacket(from *net.UDPAddr, buf []byte) error { pkt := ingressPacket{remoteAddr: from} - if err := decodePacket(buf, &pkt); err != nil { + if err := t.decodePacket(buf, &pkt); err != nil { log.Debug(fmt.Sprintf("Bad packet from %v: %v", from, err)) //fmt.Println("bad packet", err) return err @@ -414,25 +461,70 @@ func (t *udp) handlePacket(from *net.UDPAddr, buf []byte) error { return nil } -func decodePacket(buffer []byte, pkt *ingressPacket) error { - if len(buffer) < headSize+1 { +func (t *udp) decodePacket(buffer []byte, pkt *ingressPacket) error { + // calculate packet hash to check reconnect packet or PoW + targetHash := t.net.tab.self.sha + packetHash := crypto.Keccak256Hash(append(targetHash.Bytes(), buffer...)) + pkt.hash = packetHash[:] + if node, ok := t.rpHashLookup[packetHash]; ok { + pkt.remoteID = node.ID + pkt.data = new(reconn) + delete(t.rpHashLookup, packetHash) + return nil + } + + address := pkt.remoteAddr.String() + node := t.addressLookup[address] + + if node != nil && t.decryptPow.valid(packetHash) { + if !t.generalHashFilter.accept(packetHash) { + return errPacketReplay + } + if packet := node.symmEncryption.decode(buffer[powSize:]); packet != nil { + pkt.remoteID = node.ID + return t.decodeDecryptedPacket(packet, pkt) + } + } + if t.introPow.valid(packetHash) { + if !t.introHashFilter.accept(packetHash) { + return errPacketReplay + } + if packet := t.asymmEncryption.decode(buffer[powSize:]); packet != nil { + if err := t.decodeDecryptedPacket(packet, pkt); err != nil { + return err + } + if u, ok := pkt.data.(*update); ok { + if node == nil { + remotePubKey, err := u.ID.Pubkey() + if err != nil { + return err + } + enc, err := newEcdhAes256Encryption(t.priv, remotePubKey, 1280) + if err != nil { + return err + } + pkt.remoteID = u.ID + node = NewNode(pkt.remoteID, pkt.remoteAddr.IP, uint16(pkt.remoteAddr.Port), uint16(pkt.remoteAddr.Port)) + node.symmEncryption = enc + pkt.newNode = node + t.addressLookup[address] = node + } + } else { + return errDecryptFailed + } + } + } + return errUnknownNode +} + +func (t *udp) decodeDecryptedPacket(buffer []byte, pkt *ingressPacket) error { + if len(buffer) < 9 { return errPacketTooSmall } - buf := make([]byte, len(buffer)) - copy(buf, buffer) - hash, sig, sigdata := buf[:macSize], buf[macSize:headSize], buf[headSize:] - shouldhash := crypto.Keccak256(buf[macSize:]) - if !bytes.Equal(hash, shouldhash) { - return errBadHash - } - fromID, err := recoverNodeID(crypto.Keccak256(buf[headSize:]), sig) - if err != nil { - return err - } - pkt.rawData = buf - pkt.hash = hash - pkt.remoteID = fromID - switch pkt.ev = nodeEvent(sigdata[0]); pkt.ev { + + pkt.serialNo = binary.BigEndian.Uint64(buffer[:8]) + pkt.rawData = buffer + switch pkt.ev = nodeEvent(buffer[8]); pkt.ev { case pingPacket: pkt.data = new(ping) case pongPacket: @@ -450,9 +542,8 @@ func decodePacket(buffer []byte, pkt *ingressPacket) error { case topicNodesPacket: pkt.data = new(topicNodes) default: - return fmt.Errorf("unknown packet type: %d", sigdata[0]) + return fmt.Errorf("unknown packet type: %d", buffer[0]) } - s := rlp.NewStream(bytes.NewReader(sigdata[1:]), 0) - err = s.Decode(pkt.data) - return err + s := rlp.NewStream(bytes.NewReader(buffer[9:]), 0) + return s.Decode(pkt.data) } diff --git a/p2p/discv5/udp_test.go b/p2p/discv5/udp_test.go index 7d31815947..caaeb19a3d 100644 --- a/p2p/discv5/udp_test.go +++ b/p2p/discv5/udp_test.go @@ -377,7 +377,8 @@ func TestForwardCompatibility(t *testing.T) { t.Fatalf("invalid hex: %s", test.input) } var pkt ingressPacket - if err := decodePacket(input, &pkt); err != nil { + var udp *udp //TODO fix this + if err := udp.decodePacket(input, &pkt); err != nil { t.Errorf("did not accept packet %s\n%v", test.input, err) continue } diff --git a/p2p/server.go b/p2p/server.go index 922df55ba5..2cff94ea5b 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -78,9 +78,6 @@ type Config struct { // protocol should be started or not. DiscoveryV5 bool `toml:",omitempty"` - // Listener address for the V5 discovery protocol UDP traffic. - DiscoveryV5Addr string `toml:",omitempty"` - // Name sets the node name of this server. // Use common.MakeName to create a name that follows existing conventions. Name string `toml:"-"` @@ -354,6 +351,32 @@ func (srv *Server) Stop() { srv.loopWG.Wait() } +// sharedUDPConn implements a shared connection. Write sends messages to the underlying connection while read returns +// messages that were found unprocessable and sent to the unhandled channel by the primary listener. +type sharedUDPConn struct { + *net.UDPConn + unhandled chan discover.ReadPacket +} + +// ReadFromUDP implements discv5.conn +func (s *sharedUDPConn) ReadFromUDP(b []byte) (n int, addr *net.UDPAddr, err error) { + packet, ok := <-s.unhandled + if !ok { + return 0, nil, fmt.Errorf("Connection was closed") + } + l := len(packet.Data) + if l > len(b) { + l = len(b) + } + copy(b[:l], packet.Data[:l]) + return l, packet.Addr, nil +} + +// Close implements discv5.conn +func (s *sharedUDPConn) Close() error { + return nil +} + // Start starts running the server. // Servers can not be re-used after stopping. func (srv *Server) Start() (err error) { @@ -388,9 +411,43 @@ func (srv *Server) Start() (err error) { srv.peerOp = make(chan peerOpFunc) srv.peerOpDone = make(chan struct{}) + var ( + conn *net.UDPConn + sconn *sharedUDPConn + realaddr *net.UDPAddr + unhandled chan discover.ReadPacket + ) + + if !srv.NoDiscovery || srv.DiscoveryV5 { + addr, err := net.ResolveUDPAddr("udp", srv.ListenAddr) + if err != nil { + return err + } + conn, err = net.ListenUDP("udp", addr) + if err != nil { + return err + } + + realaddr = conn.LocalAddr().(*net.UDPAddr) + if srv.NAT != nil { + if !realaddr.IP.IsLoopback() { + go nat.Map(srv.NAT, srv.quit, "udp", realaddr.Port, realaddr.Port, "ethereum discovery") + } + // TODO: react to external IP changes over time. + if ext, err := srv.NAT.ExternalIP(); err == nil { + realaddr = &net.UDPAddr{IP: ext, Port: realaddr.Port} + } + } + } + + if !srv.NoDiscovery && srv.DiscoveryV5 { + unhandled = make(chan discover.ReadPacket, 100) + sconn = &sharedUDPConn{conn, unhandled} + } + // node table if !srv.NoDiscovery { - ntab, err := discover.ListenUDP(srv.PrivateKey, srv.ListenAddr, srv.NAT, srv.NodeDatabase, srv.NetRestrict) + ntab, err := discover.ListenUDP(srv.PrivateKey, conn, realaddr, unhandled, srv.NodeDatabase, srv.NetRestrict) if err != nil { return err } @@ -401,7 +458,15 @@ func (srv *Server) Start() (err error) { } if srv.DiscoveryV5 { - ntab, err := discv5.ListenUDP(srv.PrivateKey, srv.DiscoveryV5Addr, srv.NAT, "", srv.NetRestrict) //srv.NodeDatabase) + var ( + ntab *discv5.Network + err error + ) + if sconn != nil { + ntab, err = discv5.ListenUDP(srv.PrivateKey, sconn, realaddr, "", srv.NetRestrict) //srv.NodeDatabase) + } else { + ntab, err = discv5.ListenUDP(srv.PrivateKey, conn, realaddr, "", srv.NetRestrict) //srv.NodeDatabase) + } if err != nil { return err }