diff --git a/cmd/bootnode/main.go b/cmd/bootnode/main.go index e1734d89ac..ecfc6fc24e 100644 --- a/cmd/bootnode/main.go +++ b/cmd/bootnode/main.go @@ -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) } } diff --git a/cmd/faucet/faucet.go b/cmd/faucet/faucet.go index e92924fc9c..99527f9d1e 100644 --- a/cmd/faucet/faucet.go +++ b/cmd/faucet/faucet.go @@ -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, }, diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 3766ea4a60..89dcd230c4 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -636,14 +636,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) { @@ -794,7 +786,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) @@ -830,7 +821,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 } diff --git a/les/backend.go b/les/backend.go index 7180b81d76..798e44e85c 100644 --- a/les/backend.go +++ b/les/backend.go @@ -221,9 +221,8 @@ func (s *LightEthereum) Start(srvr *p2p.Server) error { s.startBloomHandlers() log.Warn("Light client mode is an experimental feature") s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.networkId) - // search the topic belonging to the oldest supported protocol because - // servers always advertise all supported protocols - protocolVersion := ClientProtocolVersions[len(ClientProtocolVersions)-1] + // clients are searching for the first advertised protocol in the list + protocolVersion := AdvertiseProtocolVersions[0] s.serverPool.start(srvr, lesTopic(s.blockchain.Genesis().Hash(), protocolVersion)) s.protocolManager.Start() return nil diff --git a/les/protocol.go b/les/protocol.go index 05e6654d6e..6a7354d1c2 100644 --- a/les/protocol.go +++ b/les/protocol.go @@ -41,8 +41,9 @@ const ( // Supported versions of the les protocol (first is primary) var ( - ClientProtocolVersions = []uint{lpv2, lpv1} - ServerProtocolVersions = []uint{lpv2, lpv1} + ClientProtocolVersions = []uint{lpv2, lpv1} + ServerProtocolVersions = []uint{lpv2, lpv1} + AdvertiseProtocolVersions = []uint{lpv2} // clients are searching for the first advertised protocol in the list ) // Number of implemented message corresponding to different protocol versions. diff --git a/les/server.go b/les/server.go index d8f93cd87b..ec2e44fecc 100644 --- a/les/server.go +++ b/les/server.go @@ -56,8 +56,8 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { return nil, err } - lesTopics := make([]discv5.Topic, len(ServerProtocolVersions)) - for i, pv := range ServerProtocolVersions { + lesTopics := make([]discv5.Topic, len(AdvertiseProtocolVersions)) + for i, pv := range AdvertiseProtocolVersions { lesTopics[i] = lesTopic(eth.BlockChain().Genesis().Hash(), pv) } diff --git a/mobile/geth.go b/mobile/geth.go index 7b39faadec..7e3b8f4915 100644 --- a/mobile/geth.go +++ b/mobile/geth.go @@ -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(), diff --git a/node/defaults.go b/node/defaults.go index 848f08e05c..d4e1486834 100644 --- a/node/defaults.go +++ b/node/defaults.go @@ -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(), }, } diff --git a/p2p/discover/udp.go b/p2p/discover/udp.go index f9eb99ee36..60436952d8 100644 --- a/p2p/discover/udp.go +++ b/p2p/discover/udp.go @@ -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: + } + } } } diff --git a/p2p/discover/udp_test.go b/p2p/discover/udp_test.go index 21e8b561da..b81caf8392 100644 --- a/p2p/discover/udp_test.go +++ b/p2p/discover/udp_test.go @@ -70,7 +70,8 @@ func newUDPTest(t *testing.T) *udpTest { remotekey: newkey(), remoteaddr: &net.UDPAddr{IP: net.IP{10, 0, 1, 99}, Port: 30303}, } - test.table, test.udp, _ = newUDP(test.localkey, test.pipe, nil, "", nil) + realaddr := test.pipe.LocalAddr().(*net.UDPAddr) + test.table, test.udp, _ = newUDP(test.localkey, test.pipe, realaddr, nil, "", nil) return test } diff --git a/p2p/discv5/net.go b/p2p/discv5/net.go index cd9981584d..f9baf126f1 100644 --- a/p2p/discv5/net.go +++ b/p2p/discv5/net.go @@ -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" ) @@ -134,7 +133,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 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/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/ticket.go b/p2p/discv5/ticket.go index b45ec4d2be..023c5000d2 100644 --- a/p2p/discv5/ticket.go +++ b/p2p/discv5/ticket.go @@ -642,7 +642,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: diff --git a/p2p/discv5/udp.go b/p2p/discv5/udp.go index 26087cd8e5..e921520768 100644 --- a/p2p/discv5/udp.go +++ b/p2p/discv5/udp.go @@ -37,7 +37,7 @@ const Version = 4 // Errors var ( errPacketTooSmall = errors.New("too small") - errBadHash = errors.New("bad hash") + errBadPrefix = errors.New("bad prefix") errExpired = errors.New("expired") errUnsolicitedReply = errors.New("unsolicited reply") errUnknownNode = errors.New("unknown node") @@ -145,10 +145,11 @@ type ( } ) -const ( - macSize = 256 / 8 - sigSize = 520 / 8 - headSize = macSize + sigSize // space of packet frame data +var ( + versionPrefix = []byte("temporary discovery v5") + versionPrefixSize = len(versionPrefix) + sigSize = 520 / 8 + headSize = versionPrefixSize + sigSize // space of packet frame data ) // Neighbors replies are sent across multiple packets to @@ -237,12 +238,12 @@ type udp struct { } // 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 +252,8 @@ 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))}, nil } func (t *udp) localAddr() *net.UDPAddr { @@ -372,11 +365,9 @@ func encodePacket(priv *ecdsa.PrivateKey, ptype byte, req interface{}) (p, hash log.Error(fmt.Sprint("could not sign packet:", err)) return nil, nil, err } - copy(packet[macSize:], sig) - // add the hash to the front. Note: this doesn't protect the - // packet in any way. - hash = crypto.Keccak256(packet[macSize:]) - copy(packet, hash) + copy(packet, versionPrefix) + copy(packet[versionPrefixSize:], sig) + hash = crypto.Keccak256(packet[versionPrefixSize:]) return packet, hash, nil } @@ -420,17 +411,16 @@ func decodePacket(buffer []byte, pkt *ingressPacket) error { } 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 + prefix, sig, sigdata := buf[:versionPrefixSize], buf[versionPrefixSize:headSize], buf[headSize:] + if !bytes.Equal(prefix, versionPrefix) { + return errBadPrefix } fromID, err := recoverNodeID(crypto.Keccak256(buf[headSize:]), sig) if err != nil { return err } pkt.rawData = buf - pkt.hash = hash + pkt.hash = crypto.Keccak256(buf[versionPrefixSize:]) pkt.remoteID = fromID switch pkt.ev = nodeEvent(sigdata[0]); pkt.ev { case pingPacket: diff --git a/p2p/server.go b/p2p/server.go index 922df55ba5..2cff94ea5b 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -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 } diff --git a/params/bootnodes.go b/params/bootnodes.go index ecb1acd4f1..849b569200 100644 --- a/params/bootnodes.go +++ b/params/bootnodes.go @@ -56,7 +56,7 @@ var RinkebyV5Bootnodes = []string{ // DiscoveryV5Bootnodes are the enode URLs of the P2P bootstrap nodes for the // experimental RLPx v5 topic-discovery network. var DiscoveryV5Bootnodes = []string{ - "enode://0cc5f5ffb5d9098c8b8c62325f3797f56509bff942704687b6530992ac706e2cb946b90a34f1f19548cd3c7baccbcaea354531e5983c7d1bc0dee16ce4b6440b@40.118.3.223:30305", - "enode://1c7a64d76c0334b0418c004af2f67c50e36a3be60b5e4790bdac0439d21603469a85fad36f2473c9a80eb043ae60936df905fa28f1ff614c3e5dc34f15dcd2dc@40.118.3.223:30308", - "enode://85c85d7143ae8bb96924f2b54f1b3e70d8c4d367af305325d30a61385a432f247d2c75c45c6b4a60335060d072d7f5b35dd1d4c45f76941f62a4f83b6e75daaf@40.118.3.223:30309", + "enode://0cc5f5ffb5d9098c8b8c62325f3797f56509bff942704687b6530992ac706e2cb946b90a34f1f19548cd3c7baccbcaea354531e5983c7d1bc0dee16ce4b6440b@40.118.3.223:30304", + "enode://1c7a64d76c0334b0418c004af2f67c50e36a3be60b5e4790bdac0439d21603469a85fad36f2473c9a80eb043ae60936df905fa28f1ff614c3e5dc34f15dcd2dc@40.118.3.223:30306", + "enode://85c85d7143ae8bb96924f2b54f1b3e70d8c4d367af305325d30a61385a432f247d2c75c45c6b4a60335060d072d7f5b35dd1d4c45f76941f62a4f83b6e75daaf@40.118.3.223:30307", }