From 84a43f7f45fe4880b3cb199e3f43a3a75178b628 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Tue, 21 Nov 2017 20:27:13 +0100 Subject: [PATCH] p2p/discv5: implement new connection/encryption layer --- p2p/discv5/encrypt.go | 196 +++++++++++++++++++++++++++++++++++ p2p/discv5/encrypt_test.go | 104 +++++++++++++++++++ p2p/discv5/net.go | 29 +++--- p2p/discv5/net_test.go | 2 +- p2p/discv5/node.go | 35 +++++++ p2p/discv5/pow.go | 207 +++++++++++++++++++++++++++++++++++++ p2p/discv5/pow_test.go | 25 +++++ p2p/discv5/sim_test.go | 2 +- p2p/discv5/udp.go | 151 ++++++++++++++++++++++----- p2p/discv5/udp_test.go | 3 +- 10 files changed, 712 insertions(+), 42 deletions(-) create mode 100644 p2p/discv5/encrypt.go create mode 100644 p2p/discv5/encrypt_test.go create mode 100644 p2p/discv5/pow.go create mode 100644 p2p/discv5/pow_test.go 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 0d2293b1a2..94cdf2c554 100644 --- a/p2p/discv5/net.go +++ b/p2p/discv5/net.go @@ -430,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") @@ -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") } 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/udp.go b/p2p/discv5/udp.go index 99baee9515..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,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 { - 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: @@ -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) - 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 }