mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-17 01:13:45 +00:00
p2p, p2p/discover, p2p/discv5: implement UDP port sharing
This commit is contained in:
parent
8c33ac10bf
commit
c9a1981292
9 changed files with 128 additions and 49 deletions
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -237,12 +237,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 +251,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 {
|
||||
|
|
@ -376,6 +368,7 @@ func encodePacket(priv *ecdsa.PrivateKey, ptype byte, req interface{}) (p, hash
|
|||
// add the hash to the front. Note: this doesn't protect the
|
||||
// packet in any way.
|
||||
hash = crypto.Keccak256(packet[macSize:])
|
||||
hash[0]++ // guarantee incompatibility between v4 and temporary v5 packet formats
|
||||
copy(packet, hash)
|
||||
return packet, hash, nil
|
||||
}
|
||||
|
|
@ -422,6 +415,7 @@ func decodePacket(buffer []byte, pkt *ingressPacket) error {
|
|||
copy(buf, buffer)
|
||||
hash, sig, sigdata := buf[:macSize], buf[macSize:headSize], buf[headSize:]
|
||||
shouldhash := crypto.Keccak256(buf[macSize:])
|
||||
shouldhash[0]++ // guarantee incompatibility between v4 and temporary v5 packet formats
|
||||
if !bytes.Equal(hash, shouldhash) {
|
||||
return errBadHash
|
||||
}
|
||||
|
|
|
|||
|
|
@ -354,6 +354,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 +414,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 +461,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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -54,7 +54,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",
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue