aboutsummaryrefslogtreecommitdiffstats
path: root/p2p
diff options
context:
space:
mode:
Diffstat (limited to 'p2p')
-rw-r--r--p2p/discover/udp.go44
-rw-r--r--p2p/discover/udp_test.go3
-rw-r--r--p2p/discv5/net.go3
-rw-r--r--p2p/discv5/net_test.go2
-rw-r--r--p2p/discv5/sim_test.go2
-rw-r--r--p2p/discv5/ticket.go2
-rw-r--r--p2p/discv5/udp.go46
-rw-r--r--p2p/server.go75
8 files changed, 114 insertions, 63 deletions
diff --git a/p2p/discover/udp.go b/p2p/discover/udp.go
index f9eb99ee3..60436952d 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 21e8b561d..b81caf839 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 cd9981584..f9baf126f 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 bd234f5ba..369282ca9 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 bf57872e2..543faecd4 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, "<no database>", nil)
+ net, err := newNetwork(transport, key.PublicKey, "<no database>", nil)
if err != nil {
panic("cannot launch new node: " + err.Error())
}
diff --git a/p2p/discv5/ticket.go b/p2p/discv5/ticket.go
index b45ec4d2b..023c5000d 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 26087cd8e..e92152076 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 922df55ba..2cff94ea5 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
}
generated by cgit (git 2.34.1) at 2024-11-14 07:55:56 +0800