p2p/discv5: implement new connection/encryption layer

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 <http://www.gnu.org/licenses/>.
+
+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
+}

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 <http://www.gnu.org/licenses/>.
+
+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)
+	}
+}

p2p/discv5/net.go

@@ -430,6 +430,7 @@ loop:
 			//fmt.Println("read", pkt.ev)
 			debugLog("<-net.read")
 			n := net.internNode(&pkt)
+			if n.serialReplayFilter.accept(pkt.serialNo) {
 				prestate := n.state
 				status := "ok"
 				if err := net.handle(n, pkt.ev, &pkt); err != nil {
@@ -440,7 +441,7 @@ loop:
 						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")
@@ -721,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
@@ -780,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) {
@@ -1214,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")
 	}

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)
 	}

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
+}

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 <http://www.gnu.org/licenses/>.
+
+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
+}

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 <http://www.gnu.org/licenses/>.
+
+package discv5
+
+import (
+	"testing"
+)
+
+func TestSimplePoW(t *testing.T) {
+	//TODO write this
+}

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, "<no database>", nil)
+	net, err := newNetwork(transport, key.PublicKey, "<no database>", nil)
 	if err != nil {
 		panic("cannot launch new node: " + err.Error())
 	}

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
@@ -149,6 +176,8 @@ const (
 	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,6 +269,13 @@ 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.
@@ -252,7 +294,19 @@ func ListenUDP(priv *ecdsa.PrivateKey, conn conn, realaddr *net.UDPAddr, nodeDBP
 }
 
 func listenUDP(priv *ecdsa.PrivateKey, conn conn, realaddr *net.UDPAddr) (*udp, error) {
-	return &udp{conn: conn, priv: priv, ourEndpoint: makeEndpoint(realaddr, uint16(realaddr.Port))}, nil
+	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 {
@@ -261,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) {
@@ -397,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
@@ -406,25 +461,70 @@ func (t *udp) handlePacket(from *net.UDPAddr, buf []byte) error {
 	return nil
 }
 
-func decodePacket(buffer []byte, pkt *ingressPacket) error {
+func (t *udp) decodePacket(buffer []byte, pkt *ingressPacket) error {
-	if len(buffer) < headSize+1 {
+	// calculate packet hash to check reconnect packet or PoW
-		return errPacketTooSmall
+	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
 	}
-	buf := make([]byte, len(buffer))
+
-	copy(buf, buffer)
+	address := pkt.remoteAddr.String()
-	hash, sig, sigdata := buf[:macSize], buf[macSize:headSize], buf[headSize:]
+	node := t.addressLookup[address]
-	shouldhash := crypto.Keccak256(buf[macSize:])
+
-	if !bytes.Equal(hash, shouldhash) {
+	if node != nil && t.decryptPow.valid(packetHash) {
-		return errBadHash
+		if !t.generalHashFilter.accept(packetHash) {
+			return errPacketReplay
 		}
-	fromID, err := recoverNodeID(crypto.Keccak256(buf[headSize:]), sig)
+		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
 					}
-	pkt.rawData = buf
+					enc, err := newEcdhAes256Encryption(t.priv, remotePubKey, 1280)
-	pkt.hash = hash
+					if err != nil {
-	pkt.remoteID = fromID
+						return err
-	switch pkt.ev = nodeEvent(sigdata[0]); pkt.ev {
+					}
+					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
+	}
+
+	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:
@@ -442,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)
+	s := rlp.NewStream(bytes.NewReader(buffer[9:]), 0)
-	err = s.Decode(pkt.data)
+	return s.Decode(pkt.data)
-	return err
 }

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
 		}