forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 01:13:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge 84a43f7f45 into 732f5468d3

Browse source
This commit is contained in:
Felföldi Zsolt 2017-12-09 22:51:22 +00:00 committed by GitHub
commit 19998a0954
22 changed files with 885 additions and 153 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
cmd/bootnode/main.go
View file

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

1
cmd/faucet/faucet.go
View file

@ -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,
},

10
cmd/utils/flags.go
View file

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

39
les/randselect.go → common/randselect.go
View file

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

10
les/randselect_test.go → common/randselect_test.go
View file

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

12
les/distributor.go
View file

@ -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

39
les/serverpool.go
View file

@ -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

1
mobile/geth.go
View file

@ -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(),

7
node/defaults.go
View file

@ -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(),
},
}

44
p2p/discover/udp.go
View file

@ -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:
}
}
}
}

196
p2p/discv5/encrypt.go Normal file
View file

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

104
p2p/discv5/encrypt_test.go Normal file
View file

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

32
p2p/discv5/net.go
View file

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

2
p2p/discv5/net_test.go
View file

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

35
p2p/discv5/node.go
View file

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

207
p2p/discv5/pow.go Normal file
View file

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

25
p2p/discv5/pow_test.go Normal file
View file

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

2
p2p/discv5/sim_test.go
View file

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

2
p2p/discv5/ticket.go
View file

@ -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:

167
p2p/discv5/udp.go
View file

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

3
p2p/discv5/udp_test.go
View file

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

75
p2p/server.go
View file

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