1// Copyright 2018 The go-ethereum Authors 2// This file is part of the go-ethereum library. 3// 4// The go-ethereum library is free software: you can redistribute it and/or modify 5// it under the terms of the GNU Lesser General Public License as published by 6// the Free Software Foundation, either version 3 of the License, or 7// (at your option) any later version. 8// 9// The go-ethereum library is distributed in the hope that it will be useful, 10// but WITHOUT ANY WARRANTY; without even the implied warranty of 11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 12// GNU Lesser General Public License for more details. 13// 14// You should have received a copy of the GNU Lesser General Public License 15// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. 16 17package enode 18 19import ( 20 "bytes" 21 "crypto/rand" 22 "encoding/binary" 23 "fmt" 24 "net" 25 "os" 26 "sync" 27 "time" 28 29 "github.com/ethereum/go-ethereum/rlp" 30 "github.com/syndtr/goleveldb/leveldb" 31 "github.com/syndtr/goleveldb/leveldb/errors" 32 "github.com/syndtr/goleveldb/leveldb/iterator" 33 "github.com/syndtr/goleveldb/leveldb/opt" 34 "github.com/syndtr/goleveldb/leveldb/storage" 35 "github.com/syndtr/goleveldb/leveldb/util" 36) 37 38// Keys in the node database. 39const ( 40 dbVersionKey = "version" // Version of the database to flush if changes 41 dbNodePrefix = "n:" // Identifier to prefix node entries with 42 dbLocalPrefix = "local:" 43 dbDiscoverRoot = "v4" 44 dbDiscv5Root = "v5" 45 46 // These fields are stored per ID and IP, the full key is "n:<ID>:v4:<IP>:findfail". 47 // Use nodeItemKey to create those keys. 48 dbNodeFindFails = "findfail" 49 dbNodePing = "lastping" 50 dbNodePong = "lastpong" 51 dbNodeSeq = "seq" 52 53 // Local information is keyed by ID only, the full key is "local:<ID>:seq". 54 // Use localItemKey to create those keys. 55 dbLocalSeq = "seq" 56) 57 58const ( 59 dbNodeExpiration = 24 * time.Hour // Time after which an unseen node should be dropped. 60 dbCleanupCycle = time.Hour // Time period for running the expiration task. 61 dbVersion = 9 62) 63 64var ( 65 errInvalidIP = errors.New("invalid IP") 66) 67 68var zeroIP = make(net.IP, 16) 69 70// DB is the node database, storing previously seen nodes and any collected metadata about 71// them for QoS purposes. 72type DB struct { 73 lvl *leveldb.DB // Interface to the database itself 74 runner sync.Once // Ensures we can start at most one expirer 75 quit chan struct{} // Channel to signal the expiring thread to stop 76} 77 78// OpenDB opens a node database for storing and retrieving infos about known peers in the 79// network. If no path is given an in-memory, temporary database is constructed. 80func OpenDB(path string) (*DB, error) { 81 if path == "" { 82 return newMemoryDB() 83 } 84 return newPersistentDB(path) 85} 86 87// newMemoryNodeDB creates a new in-memory node database without a persistent backend. 88func newMemoryDB() (*DB, error) { 89 db, err := leveldb.Open(storage.NewMemStorage(), nil) 90 if err != nil { 91 return nil, err 92 } 93 return &DB{lvl: db, quit: make(chan struct{})}, nil 94} 95 96// newPersistentNodeDB creates/opens a leveldb backed persistent node database, 97// also flushing its contents in case of a version mismatch. 98func newPersistentDB(path string) (*DB, error) { 99 opts := &opt.Options{OpenFilesCacheCapacity: 5} 100 db, err := leveldb.OpenFile(path, opts) 101 if _, iscorrupted := err.(*errors.ErrCorrupted); iscorrupted { 102 db, err = leveldb.RecoverFile(path, nil) 103 } 104 if err != nil { 105 return nil, err 106 } 107 // The nodes contained in the cache correspond to a certain protocol version. 108 // Flush all nodes if the version doesn't match. 109 currentVer := make([]byte, binary.MaxVarintLen64) 110 currentVer = currentVer[:binary.PutVarint(currentVer, int64(dbVersion))] 111 112 blob, err := db.Get([]byte(dbVersionKey), nil) 113 switch err { 114 case leveldb.ErrNotFound: 115 // Version not found (i.e. empty cache), insert it 116 if err := db.Put([]byte(dbVersionKey), currentVer, nil); err != nil { 117 db.Close() 118 return nil, err 119 } 120 121 case nil: 122 // Version present, flush if different 123 if !bytes.Equal(blob, currentVer) { 124 db.Close() 125 if err = os.RemoveAll(path); err != nil { 126 return nil, err 127 } 128 return newPersistentDB(path) 129 } 130 } 131 return &DB{lvl: db, quit: make(chan struct{})}, nil 132} 133 134// nodeKey returns the database key for a node record. 135func nodeKey(id ID) []byte { 136 key := append([]byte(dbNodePrefix), id[:]...) 137 key = append(key, ':') 138 key = append(key, dbDiscoverRoot...) 139 return key 140} 141 142// splitNodeKey returns the node ID of a key created by nodeKey. 143func splitNodeKey(key []byte) (id ID, rest []byte) { 144 if !bytes.HasPrefix(key, []byte(dbNodePrefix)) { 145 return ID{}, nil 146 } 147 item := key[len(dbNodePrefix):] 148 copy(id[:], item[:len(id)]) 149 return id, item[len(id)+1:] 150} 151 152// nodeItemKey returns the database key for a node metadata field. 153func nodeItemKey(id ID, ip net.IP, field string) []byte { 154 ip16 := ip.To16() 155 if ip16 == nil { 156 panic(fmt.Errorf("invalid IP (length %d)", len(ip))) 157 } 158 return bytes.Join([][]byte{nodeKey(id), ip16, []byte(field)}, []byte{':'}) 159} 160 161// splitNodeItemKey returns the components of a key created by nodeItemKey. 162func splitNodeItemKey(key []byte) (id ID, ip net.IP, field string) { 163 id, key = splitNodeKey(key) 164 // Skip discover root. 165 if string(key) == dbDiscoverRoot { 166 return id, nil, "" 167 } 168 key = key[len(dbDiscoverRoot)+1:] 169 // Split out the IP. 170 ip = key[:16] 171 if ip4 := ip.To4(); ip4 != nil { 172 ip = ip4 173 } 174 key = key[16+1:] 175 // Field is the remainder of key. 176 field = string(key) 177 return id, ip, field 178} 179 180func v5Key(id ID, ip net.IP, field string) []byte { 181 return bytes.Join([][]byte{ 182 []byte(dbNodePrefix), 183 id[:], 184 []byte(dbDiscv5Root), 185 ip.To16(), 186 []byte(field), 187 }, []byte{':'}) 188} 189 190// localItemKey returns the key of a local node item. 191func localItemKey(id ID, field string) []byte { 192 key := append([]byte(dbLocalPrefix), id[:]...) 193 key = append(key, ':') 194 key = append(key, field...) 195 return key 196} 197 198// fetchInt64 retrieves an integer associated with a particular key. 199func (db *DB) fetchInt64(key []byte) int64 { 200 blob, err := db.lvl.Get(key, nil) 201 if err != nil { 202 return 0 203 } 204 val, read := binary.Varint(blob) 205 if read <= 0 { 206 return 0 207 } 208 return val 209} 210 211// storeInt64 stores an integer in the given key. 212func (db *DB) storeInt64(key []byte, n int64) error { 213 blob := make([]byte, binary.MaxVarintLen64) 214 blob = blob[:binary.PutVarint(blob, n)] 215 return db.lvl.Put(key, blob, nil) 216} 217 218// fetchUint64 retrieves an integer associated with a particular key. 219func (db *DB) fetchUint64(key []byte) uint64 { 220 blob, err := db.lvl.Get(key, nil) 221 if err != nil { 222 return 0 223 } 224 val, _ := binary.Uvarint(blob) 225 return val 226} 227 228// storeUint64 stores an integer in the given key. 229func (db *DB) storeUint64(key []byte, n uint64) error { 230 blob := make([]byte, binary.MaxVarintLen64) 231 blob = blob[:binary.PutUvarint(blob, n)] 232 return db.lvl.Put(key, blob, nil) 233} 234 235// Node retrieves a node with a given id from the database. 236func (db *DB) Node(id ID) *Node { 237 blob, err := db.lvl.Get(nodeKey(id), nil) 238 if err != nil { 239 return nil 240 } 241 return mustDecodeNode(id[:], blob) 242} 243 244func mustDecodeNode(id, data []byte) *Node { 245 node := new(Node) 246 if err := rlp.DecodeBytes(data, &node.r); err != nil { 247 panic(fmt.Errorf("p2p/enode: can't decode node %x in DB: %v", id, err)) 248 } 249 // Restore node id cache. 250 copy(node.id[:], id) 251 return node 252} 253 254// UpdateNode inserts - potentially overwriting - a node into the peer database. 255func (db *DB) UpdateNode(node *Node) error { 256 if node.Seq() < db.NodeSeq(node.ID()) { 257 return nil 258 } 259 blob, err := rlp.EncodeToBytes(&node.r) 260 if err != nil { 261 return err 262 } 263 if err := db.lvl.Put(nodeKey(node.ID()), blob, nil); err != nil { 264 return err 265 } 266 return db.storeUint64(nodeItemKey(node.ID(), zeroIP, dbNodeSeq), node.Seq()) 267} 268 269// NodeSeq returns the stored record sequence number of the given node. 270func (db *DB) NodeSeq(id ID) uint64 { 271 return db.fetchUint64(nodeItemKey(id, zeroIP, dbNodeSeq)) 272} 273 274// Resolve returns the stored record of the node if it has a larger sequence 275// number than n. 276func (db *DB) Resolve(n *Node) *Node { 277 if n.Seq() > db.NodeSeq(n.ID()) { 278 return n 279 } 280 return db.Node(n.ID()) 281} 282 283// DeleteNode deletes all information associated with a node. 284func (db *DB) DeleteNode(id ID) { 285 deleteRange(db.lvl, nodeKey(id)) 286} 287 288func deleteRange(db *leveldb.DB, prefix []byte) { 289 it := db.NewIterator(util.BytesPrefix(prefix), nil) 290 defer it.Release() 291 for it.Next() { 292 db.Delete(it.Key(), nil) 293 } 294} 295 296// ensureExpirer is a small helper method ensuring that the data expiration 297// mechanism is running. If the expiration goroutine is already running, this 298// method simply returns. 299// 300// The goal is to start the data evacuation only after the network successfully 301// bootstrapped itself (to prevent dumping potentially useful seed nodes). Since 302// it would require significant overhead to exactly trace the first successful 303// convergence, it's simpler to "ensure" the correct state when an appropriate 304// condition occurs (i.e. a successful bonding), and discard further events. 305func (db *DB) ensureExpirer() { 306 db.runner.Do(func() { go db.expirer() }) 307} 308 309// expirer should be started in a go routine, and is responsible for looping ad 310// infinitum and dropping stale data from the database. 311func (db *DB) expirer() { 312 tick := time.NewTicker(dbCleanupCycle) 313 defer tick.Stop() 314 for { 315 select { 316 case <-tick.C: 317 db.expireNodes() 318 case <-db.quit: 319 return 320 } 321 } 322} 323 324// expireNodes iterates over the database and deletes all nodes that have not 325// been seen (i.e. received a pong from) for some time. 326func (db *DB) expireNodes() { 327 it := db.lvl.NewIterator(util.BytesPrefix([]byte(dbNodePrefix)), nil) 328 defer it.Release() 329 if !it.Next() { 330 return 331 } 332 333 var ( 334 threshold = time.Now().Add(-dbNodeExpiration).Unix() 335 youngestPong int64 336 atEnd = false 337 ) 338 for !atEnd { 339 id, ip, field := splitNodeItemKey(it.Key()) 340 if field == dbNodePong { 341 time, _ := binary.Varint(it.Value()) 342 if time > youngestPong { 343 youngestPong = time 344 } 345 if time < threshold { 346 // Last pong from this IP older than threshold, remove fields belonging to it. 347 deleteRange(db.lvl, nodeItemKey(id, ip, "")) 348 } 349 } 350 atEnd = !it.Next() 351 nextID, _ := splitNodeKey(it.Key()) 352 if atEnd || nextID != id { 353 // We've moved beyond the last entry of the current ID. 354 // Remove everything if there was no recent enough pong. 355 if youngestPong > 0 && youngestPong < threshold { 356 deleteRange(db.lvl, nodeKey(id)) 357 } 358 youngestPong = 0 359 } 360 } 361} 362 363// LastPingReceived retrieves the time of the last ping packet received from 364// a remote node. 365func (db *DB) LastPingReceived(id ID, ip net.IP) time.Time { 366 if ip = ip.To16(); ip == nil { 367 return time.Time{} 368 } 369 return time.Unix(db.fetchInt64(nodeItemKey(id, ip, dbNodePing)), 0) 370} 371 372// UpdateLastPingReceived updates the last time we tried contacting a remote node. 373func (db *DB) UpdateLastPingReceived(id ID, ip net.IP, instance time.Time) error { 374 if ip = ip.To16(); ip == nil { 375 return errInvalidIP 376 } 377 return db.storeInt64(nodeItemKey(id, ip, dbNodePing), instance.Unix()) 378} 379 380// LastPongReceived retrieves the time of the last successful pong from remote node. 381func (db *DB) LastPongReceived(id ID, ip net.IP) time.Time { 382 if ip = ip.To16(); ip == nil { 383 return time.Time{} 384 } 385 // Launch expirer 386 db.ensureExpirer() 387 return time.Unix(db.fetchInt64(nodeItemKey(id, ip, dbNodePong)), 0) 388} 389 390// UpdateLastPongReceived updates the last pong time of a node. 391func (db *DB) UpdateLastPongReceived(id ID, ip net.IP, instance time.Time) error { 392 if ip = ip.To16(); ip == nil { 393 return errInvalidIP 394 } 395 return db.storeInt64(nodeItemKey(id, ip, dbNodePong), instance.Unix()) 396} 397 398// FindFails retrieves the number of findnode failures since bonding. 399func (db *DB) FindFails(id ID, ip net.IP) int { 400 if ip = ip.To16(); ip == nil { 401 return 0 402 } 403 return int(db.fetchInt64(nodeItemKey(id, ip, dbNodeFindFails))) 404} 405 406// UpdateFindFails updates the number of findnode failures since bonding. 407func (db *DB) UpdateFindFails(id ID, ip net.IP, fails int) error { 408 if ip = ip.To16(); ip == nil { 409 return errInvalidIP 410 } 411 return db.storeInt64(nodeItemKey(id, ip, dbNodeFindFails), int64(fails)) 412} 413 414// FindFailsV5 retrieves the discv5 findnode failure counter. 415func (db *DB) FindFailsV5(id ID, ip net.IP) int { 416 if ip = ip.To16(); ip == nil { 417 return 0 418 } 419 return int(db.fetchInt64(v5Key(id, ip, dbNodeFindFails))) 420} 421 422// UpdateFindFailsV5 stores the discv5 findnode failure counter. 423func (db *DB) UpdateFindFailsV5(id ID, ip net.IP, fails int) error { 424 if ip = ip.To16(); ip == nil { 425 return errInvalidIP 426 } 427 return db.storeInt64(v5Key(id, ip, dbNodeFindFails), int64(fails)) 428} 429 430// localSeq retrieves the local record sequence counter, defaulting to the current 431// timestamp if no previous exists. This ensures that wiping all data associated 432// with a node (apart from its key) will not generate already used sequence nums. 433func (db *DB) localSeq(id ID) uint64 { 434 if seq := db.fetchUint64(localItemKey(id, dbLocalSeq)); seq > 0 { 435 return seq 436 } 437 return nowMilliseconds() 438} 439 440// storeLocalSeq stores the local record sequence counter. 441func (db *DB) storeLocalSeq(id ID, n uint64) { 442 db.storeUint64(localItemKey(id, dbLocalSeq), n) 443} 444 445// QuerySeeds retrieves random nodes to be used as potential seed nodes 446// for bootstrapping. 447func (db *DB) QuerySeeds(n int, maxAge time.Duration) []*Node { 448 var ( 449 now = time.Now() 450 nodes = make([]*Node, 0, n) 451 it = db.lvl.NewIterator(nil, nil) 452 id ID 453 ) 454 defer it.Release() 455 456seek: 457 for seeks := 0; len(nodes) < n && seeks < n*5; seeks++ { 458 // Seek to a random entry. The first byte is incremented by a 459 // random amount each time in order to increase the likelihood 460 // of hitting all existing nodes in very small databases. 461 ctr := id[0] 462 rand.Read(id[:]) 463 id[0] = ctr + id[0]%16 464 it.Seek(nodeKey(id)) 465 466 n := nextNode(it) 467 if n == nil { 468 id[0] = 0 469 continue seek // iterator exhausted 470 } 471 if now.Sub(db.LastPongReceived(n.ID(), n.IP())) > maxAge { 472 continue seek 473 } 474 for i := range nodes { 475 if nodes[i].ID() == n.ID() { 476 continue seek // duplicate 477 } 478 } 479 nodes = append(nodes, n) 480 } 481 return nodes 482} 483 484// reads the next node record from the iterator, skipping over other 485// database entries. 486func nextNode(it iterator.Iterator) *Node { 487 for end := false; !end; end = !it.Next() { 488 id, rest := splitNodeKey(it.Key()) 489 if string(rest) != dbDiscoverRoot { 490 continue 491 } 492 return mustDecodeNode(id[:], it.Value()) 493 } 494 return nil 495} 496 497// close flushes and closes the database files. 498func (db *DB) Close() { 499 close(db.quit) 500 db.lvl.Close() 501} 502