+++ /dev/null
-package discover
-
-import (
- "bytes"
- "crypto/ecdsa"
- "errors"
- "fmt"
- "net"
- "time"
-
- log "github.com/sirupsen/logrus"
- "github.com/tendermint/go-crypto"
- "github.com/tendermint/go-wire"
-
- "github.com/vapor/common"
- "github.com/vapor/p2p/netutil"
-)
-
-const Version = 4
-
-// Errors
-var (
- errPacketTooSmall = errors.New("too small")
- errBadPrefix = errors.New("bad prefix")
- errExpired = errors.New("expired")
- errUnsolicitedReply = errors.New("unsolicited reply")
- errUnknownNode = errors.New("unknown node")
- errTimeout = errors.New("RPC timeout")
- errClockWarp = errors.New("reply deadline too far in the future")
- errClosed = errors.New("socket closed")
-)
-
-// Timeouts
-const (
- respTimeout = 1 * time.Second
- queryDelay = 1000 * time.Millisecond
- expiration = 20 * time.Second
-
- ntpFailureThreshold = 32 // Continuous timeouts after which to check NTP
- ntpWarningCooldown = 10 * time.Minute // Minimum amount of time to pass before repeating NTP warning
- driftThreshold = 10 * time.Second // Allowed clock drift before warning user
-)
-
-// ReadPacket is sent to the unhandled channel when it could not be processed
-type ReadPacket struct {
- Data []byte
- Addr *net.UDPAddr
-}
-
-// Config holds Table-related settings.
-type Config struct {
- // These settings are required and configure the UDP listener:
- PrivateKey *ecdsa.PrivateKey
-
- // These settings are optional:
- AnnounceAddr *net.UDPAddr // local address announced in the DHT
- NodeDBPath string // if set, the node database is stored at this filesystem location
- //NetRestrict *netutil.Netlist // network whitelist
- Bootnodes []*Node // list of bootstrap nodes
- Unhandled chan<- ReadPacket // unhandled packets are sent on this channel
-}
-
-// RPC request structures
-type (
- ping struct {
- Version uint
- From, To rpcEndpoint
- Expiration uint64
-
- // v5
- Topics []Topic
-
- // Ignore additional fields (for forward compatibility).
- Rest []byte
- }
-
- // pong is the reply to ping.
- pong struct {
- // This field should mirror the UDP envelope address
- // of the ping packet, which provides a way to discover the
- // the external address (after NAT).
- To rpcEndpoint
-
- ReplyTok []byte // This contains the hash of the ping packet.
- Expiration uint64 // Absolute timestamp at which the packet becomes invalid.
-
- // v5
- TopicHash common.Hash
- TicketSerial uint32
- WaitPeriods []uint32
-
- // Ignore additional fields (for forward compatibility).
- Rest []byte
- }
-
- // findnode is a query for nodes close to the given target.
- findnode struct {
- Target NodeID // doesn't need to be an actual public key
- Expiration uint64
- // Ignore additional fields (for forward compatibility).
- Rest []byte
- }
-
- // findnode is a query for nodes close to the given target.
- findnodeHash struct {
- Target common.Hash
- Expiration uint64
- // Ignore additional fields (for forward compatibility).
- Rest []byte
- }
-
- // reply to findnode
- neighbors struct {
- Nodes []rpcNode
- Expiration uint64
- // Ignore additional fields (for forward compatibility).
- Rest []byte
- }
-
- topicRegister struct {
- Topics []Topic
- Idx uint
- Pong []byte
- }
-
- topicQuery struct {
- Topic Topic
- Expiration uint64
- }
-
- // reply to topicQuery
- topicNodes struct {
- Echo common.Hash
- Nodes []rpcNode
- }
-
- rpcNode struct {
- IP net.IP // len 4 for IPv4 or 16 for IPv6
- UDP uint16 // for discovery protocol
- TCP uint16 // for RLPx protocol
- ID NodeID
- }
-
- rpcEndpoint struct {
- IP net.IP // len 4 for IPv4 or 16 for IPv6
- UDP uint16 // for discovery protocol
- TCP uint16 // for RLPx protocol
- }
-)
-
-var (
- versionPrefix = []byte("bytom discovery")
- versionPrefixSize = len(versionPrefix)
- nodeIDSize = 32
- sigSize = 520 / 8
- headSize = versionPrefixSize + nodeIDSize + sigSize // space of packet frame data
-)
-
-// Neighbors replies are sent across multiple packets to
-// stay below the 1280 byte limit. We compute the maximum number
-// of entries by stuffing a packet until it grows too large.
-var maxNeighbors = func() int {
- p := neighbors{Expiration: ^uint64(0)}
- maxSizeNode := rpcNode{IP: make(net.IP, 16), UDP: ^uint16(0), TCP: ^uint16(0)}
- for n := 0; ; n++ {
- p.Nodes = append(p.Nodes, maxSizeNode)
- var size int
- var err error
- b := new(bytes.Buffer)
- wire.WriteJSON(p, b, &size, &err)
- if err != nil {
- // If this ever happens, it will be caught by the unit tests.
- panic("cannot encode: " + err.Error())
- }
- if headSize+size+1 >= 1280 {
- return n
- }
- }
-}()
-
-var maxTopicNodes = func() int {
- p := topicNodes{}
- maxSizeNode := rpcNode{IP: make(net.IP, 16), UDP: ^uint16(0), TCP: ^uint16(0)}
- for n := 0; ; n++ {
- p.Nodes = append(p.Nodes, maxSizeNode)
- var size int
- var err error
- b := new(bytes.Buffer)
- wire.WriteJSON(p, b, &size, &err)
- if err != nil {
- // If this ever happens, it will be caught by the unit tests.
- panic("cannot encode: " + err.Error())
- }
- if headSize+size+1 >= 1280 {
- return n
- }
- }
-}()
-
-func makeEndpoint(addr *net.UDPAddr, tcpPort uint16) rpcEndpoint {
- ip := addr.IP.To4()
- if ip == nil {
- ip = addr.IP.To16()
- }
- return rpcEndpoint{IP: ip, UDP: uint16(addr.Port), TCP: tcpPort}
-}
-
-func (e1 rpcEndpoint) equal(e2 rpcEndpoint) bool {
- return e1.UDP == e2.UDP && e1.TCP == e2.TCP && e1.IP.Equal(e2.IP)
-}
-
-func nodeFromRPC(sender *net.UDPAddr, rn rpcNode) (*Node, error) {
- if err := netutil.CheckRelayIP(sender.IP, rn.IP); err != nil {
- return nil, err
- }
- n := NewNode(rn.ID, rn.IP, rn.UDP, rn.TCP)
- err := n.validateComplete()
- return n, err
-}
-
-func nodeToRPC(n *Node) rpcNode {
- return rpcNode{ID: n.ID, IP: n.IP, UDP: n.UDP, TCP: n.TCP}
-}
-
-type ingressPacket struct {
- remoteID NodeID
- remoteAddr *net.UDPAddr
- ev nodeEvent
- hash []byte
- data interface{} // one of the RPC structs
- rawData []byte
-}
-
-type conn interface {
- ReadFromUDP(b []byte) (n int, addr *net.UDPAddr, err error)
- WriteToUDP(b []byte, addr *net.UDPAddr) (n int, err error)
- Close() error
- LocalAddr() net.Addr
-}
-
-// udp implements the RPC protocol.
-type udp struct {
- conn conn
- priv *crypto.PrivKeyEd25519
- ourEndpoint rpcEndpoint
- //nat nat.Interface
- net *Network
-}
-
-// ListenUDP returns a new table that listens for UDP packets on laddr.
-func ListenUDP(priv *crypto.PrivKeyEd25519, 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.PubKey().Unwrap().(crypto.PubKeyEd25519), nodeDBPath, netrestrict)
- if err != nil {
- return nil, err
- }
- log.Info("UDP listener up v5", "net", net.tab.self)
- transport.net = net
- go transport.readLoop()
- return net, nil
-}
-
-func listenUDP(priv *crypto.PrivKeyEd25519, conn conn, realaddr *net.UDPAddr) (*udp, error) {
- return &udp{conn: conn, priv: priv, ourEndpoint: makeEndpoint(realaddr, uint16(realaddr.Port))}, nil
-}
-
-func (t *udp) localAddr() *net.UDPAddr {
- return t.conn.LocalAddr().(*net.UDPAddr)
-}
-
-func (t *udp) Close() {
- t.conn.Close()
-}
-
-func (t *udp) send(remote *Node, ptype nodeEvent, data interface{}) (hash []byte) {
- hash, _ = t.sendPacket(remote.ID, remote.addr(), byte(ptype), data)
- return hash
-}
-
-func (t *udp) sendPing(remote *Node, toaddr *net.UDPAddr, topics []Topic) (hash []byte) {
- hash, _ = t.sendPacket(remote.ID, toaddr, byte(pingPacket), ping{
- Version: Version,
- From: t.ourEndpoint,
- To: makeEndpoint(toaddr, uint16(toaddr.Port)), // TODO: maybe use known TCP port from DB
- Expiration: uint64(time.Now().Add(expiration).Unix()),
- Topics: topics,
- })
- return hash
-}
-
-func (t *udp) sendFindnode(remote *Node, target NodeID) {
- t.sendPacket(remote.ID, remote.addr(), byte(findnodePacket), findnode{
- Target: target,
- Expiration: uint64(time.Now().Add(expiration).Unix()),
- })
-}
-
-func (t *udp) sendNeighbours(remote *Node, results []*Node) {
- // Send neighbors in chunks with at most maxNeighbors per packet
- // to stay below the 1280 byte limit.
- p := neighbors{Expiration: uint64(time.Now().Add(expiration).Unix())}
- for i, result := range results {
- p.Nodes = append(p.Nodes, nodeToRPC(result))
- if len(p.Nodes) == maxNeighbors || i == len(results)-1 {
- t.sendPacket(remote.ID, remote.addr(), byte(neighborsPacket), p)
- p.Nodes = p.Nodes[:0]
- }
- }
-}
-
-func (t *udp) sendFindnodeHash(remote *Node, target common.Hash) {
- t.sendPacket(remote.ID, remote.addr(), byte(findnodeHashPacket), findnodeHash{
- Target: common.Hash(target),
- Expiration: uint64(time.Now().Add(expiration).Unix()),
- })
-}
-
-func (t *udp) sendTopicRegister(remote *Node, topics []Topic, idx int, pong []byte) {
- t.sendPacket(remote.ID, remote.addr(), byte(topicRegisterPacket), topicRegister{
- Topics: topics,
- Idx: uint(idx),
- Pong: pong,
- })
-}
-
-func (t *udp) sendTopicNodes(remote *Node, queryHash common.Hash, nodes []*Node) {
- p := topicNodes{Echo: queryHash}
- var sent bool
- for _, result := range nodes {
- if result.IP.Equal(t.net.tab.self.IP) || netutil.CheckRelayIP(remote.IP, result.IP) == nil {
- p.Nodes = append(p.Nodes, nodeToRPC(result))
- }
- if len(p.Nodes) == maxTopicNodes {
- t.sendPacket(remote.ID, remote.addr(), byte(topicNodesPacket), p)
- p.Nodes = p.Nodes[:0]
- sent = true
- }
- }
- if !sent || len(p.Nodes) > 0 {
- t.sendPacket(remote.ID, remote.addr(), byte(topicNodesPacket), p)
- }
-}
-
-func (t *udp) sendPacket(toid NodeID, toaddr *net.UDPAddr, ptype byte, req interface{}) (hash []byte, err error) {
- //fmt.Println("sendPacket", nodeEvent(ptype), toaddr.String(), toid.String())
- packet, hash, err := encodePacket(t.priv, ptype, req)
- if err != nil {
- //fmt.Println(err)
- return hash, err
- }
- log.Debug(fmt.Sprintf(">>> %v to %x@%v", nodeEvent(ptype), toid[:8], toaddr))
- if _, err = t.conn.WriteToUDP(packet, toaddr); err != nil {
- log.Info(fmt.Sprint("UDP send failed:", err))
- }
- return hash, err
-}
-
-// zeroed padding space for encodePacket.
-var headSpace = make([]byte, headSize)
-
-func encodePacket(priv *crypto.PrivKeyEd25519, ptype byte, req interface{}) (p, hash []byte, err error) {
- b := new(bytes.Buffer)
- b.Write(headSpace)
- b.WriteByte(ptype)
- var size int
- wire.WriteJSON(req, b, &size, &err)
- if err != nil {
- log.Error(fmt.Sprint("error encoding packet:", err))
- return nil, nil, err
- }
- packet := b.Bytes()
- nodeID := priv.PubKey().Unwrap().(crypto.PubKeyEd25519)
- sig := priv.Sign(common.BytesToHash(packet[headSize:]).Bytes())
- copy(packet, versionPrefix)
- copy(packet[versionPrefixSize:], nodeID[:])
- copy(packet[versionPrefixSize+nodeIDSize:], sig.Bytes())
-
- hash = common.BytesToHash(packet[versionPrefixSize:]).Bytes()
- return packet, hash, nil
-}
-
-// readLoop runs in its own goroutine. it injects ingress UDP packets
-// into the network loop.
-func (t *udp) readLoop() {
- defer t.conn.Close()
- // 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.
- buf := make([]byte, 1280)
- for {
- nbytes, from, err := t.conn.ReadFromUDP(buf)
- if netutil.IsTemporaryError(err) {
- // Ignore temporary read errors.
- log.Debug(fmt.Sprintf("Temporary read error: %v", err))
- continue
- } else if err != nil {
- // Shut down the loop for permament errors.
- log.Debug(fmt.Sprintf("Read error: %v", err))
- return
- }
- t.handlePacket(from, buf[:nbytes])
- }
-}
-
-func (t *udp) handlePacket(from *net.UDPAddr, buf []byte) error {
- pkt := ingressPacket{remoteAddr: from}
- if err := decodePacket(buf, &pkt); err != nil {
- log.Debug(fmt.Sprintf("Bad packet from %v: %v", from, err))
- //fmt.Println("bad packet", err)
- return err
- }
- t.net.reqReadPacket(pkt)
- return nil
-}
-
-func decodePacket(buffer []byte, pkt *ingressPacket) error {
- if len(buffer) < headSize+1 {
- return errPacketTooSmall
- }
- buf := make([]byte, len(buffer))
- copy(buf, buffer)
- prefix, fromID, sigdata := buf[:versionPrefixSize], buf[versionPrefixSize:versionPrefixSize+nodeIDSize], buf[headSize:]
- if !bytes.Equal(prefix, versionPrefix) {
- return errBadPrefix
- }
- pkt.rawData = buf
- pkt.hash = common.BytesToHash(buf[versionPrefixSize:]).Bytes()
- pkt.remoteID = ByteID(fromID)
- switch pkt.ev = nodeEvent(sigdata[0]); pkt.ev {
- case pingPacket:
- pkt.data = new(ping)
- case pongPacket:
- pkt.data = new(pong)
- case findnodePacket:
- pkt.data = new(findnode)
- case neighborsPacket:
- pkt.data = new(neighbors)
- case findnodeHashPacket:
- pkt.data = new(findnodeHash)
- case topicRegisterPacket:
- pkt.data = new(topicRegister)
- case topicQueryPacket:
- pkt.data = new(topicQuery)
- case topicNodesPacket:
- pkt.data = new(topicNodes)
- default:
- return fmt.Errorf("unknown packet type: %d", sigdata[0])
- }
- var err error
- wire.ReadJSON(pkt.data, sigdata[1:], &err)
- if err != nil {
- log.Error("wire readjson err:", err)
- }
-
- return err
-}
OSDN Git Service
