diff --git a/go.mod b/go.mod index a9022efa80..b4f5b129de 100644 --- a/go.mod +++ b/go.mod @@ -54,6 +54,7 @@ require ( github.com/mattn/go-isatty v0.0.17 github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416 github.com/olekukonko/tablewriter v0.0.5 + github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7 github.com/rs/cors v1.7.0 @@ -71,7 +72,7 @@ require ( golang.org/x/sys v0.13.0 golang.org/x/text v0.13.0 golang.org/x/time v0.3.0 - golang.org/x/tools v0.13.0 + golang.org/x/tools v0.14.0 gopkg.in/natefinch/lumberjack.v2 v2.0.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -127,7 +128,6 @@ require ( github.com/mmcloughlin/addchain v0.4.0 // indirect github.com/naoina/go-stringutil v0.1.0 // indirect github.com/opentracing/opentracing-go v1.1.0 // indirect - github.com/optimism-java/utp-go v0.0.0-20231024092003-1dd76611b5f2 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_golang v1.12.0 // indirect @@ -140,10 +140,9 @@ require ( github.com/tklauser/go-sysconf v0.3.12 // indirect github.com/tklauser/numcpus v0.6.1 // indirect github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect - go.uber.org/atomic v1.7.0 // indirect - go.uber.org/multierr v1.6.0 // indirect - go.uber.org/zap v1.19.0 // indirect - golang.org/x/mod v0.12.0 // indirect + go.uber.org/multierr v1.11.0 // indirect + go.uber.org/zap v1.26.0 // indirect + golang.org/x/mod v0.13.0 // indirect golang.org/x/net v0.17.0 // indirect google.golang.org/protobuf v1.27.1 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect diff --git a/go.sum b/go.sum index 5f370a931b..1d99102ab7 100644 --- a/go.sum +++ b/go.sum @@ -98,8 +98,6 @@ github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= -github.com/bits-and-blooms/bitset v1.7.0 h1:YjAGVd3XmtK9ktAbX8Zg2g2PwLIMjGREZJHlV4j7NEo= -github.com/bits-and-blooms/bitset v1.7.0/go.mod h1:gIdJ4wp64HaoK2YrL1Q5/N7Y16edYb8uY+O0FJTyyDA= github.com/bits-and-blooms/bitset v1.10.0 h1:ePXTeiPEazB5+opbv5fr8umg2R/1NlzgDsyepwsSr88= github.com/bits-and-blooms/bitset v1.10.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= github.com/btcsuite/btcd/btcec/v2 v2.2.0 h1:fzn1qaOt32TuLjFlkzYSsBC35Q3KUjT1SwPxiMSCF5k= @@ -499,8 +497,8 @@ github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE= github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo= github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU= github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= -github.com/optimism-java/utp-go v0.0.0-20231024092003-1dd76611b5f2 h1:1H9unjvDxqDVTF9rh99sfolMtp8zYCUNAq+aoeGCxLA= -github.com/optimism-java/utp-go v0.0.0-20231024092003-1dd76611b5f2/go.mod h1:ohCuwoc66lfiNpo2Wk22zV07IbB0gte8+TYiclv9an4= +github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98 h1:uxUbd8LFc24XetNFjTu9Kp9MqF2zKF92UMbcDuPxYZ8= +github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0= diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index cd4b0ddbf7..df6bd861f8 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -7,11 +7,13 @@ import ( "encoding/binary" "errors" "fmt" + "io" "net" "sort" "time" "github.com/VictoriaMetrics/fastcache" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover/portalwire" "github.com/ethereum/go-ethereum/p2p/enode" @@ -43,9 +45,11 @@ const ( portalFindnodesResultLimit = 32 - defaultUTPAcceptTimeout = 15 * time.Second + defaultUTPConnectTimeout = 15 * time.Second defaultUTPWriteTimeout = 60 * time.Second + + defaultUTPReadTimeout = 60 * time.Second ) type PortalProtocolConfig struct { @@ -94,7 +98,7 @@ type PortalProtocol struct { storage Storage } -func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateKey *ecdsa.PrivateKey) (*PortalProtocol, error) { +func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateKey *ecdsa.PrivateKey, storage Storage) (*PortalProtocol, error) { nodeDB, err := enode.OpenDB(config.NodeDBPath) if err != nil { return nil, err @@ -117,6 +121,7 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateK cancelCloseCtx: cancelCloseCtx, localNode: localNode, validSchemes: enode.ValidSchemes, + storage: storage, } return protocol, nil @@ -161,9 +166,8 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) { p.utpPackets = make(chan *utp.UdpMessage, 10) p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithCustomHandler( func(buf []byte, addr *net.UDPAddr) (int, error) { - var a [32]byte - // todo need to find enode.ID by addr - _, err := p.DiscV5.TalkRequestToID(a, addr, portalwire.UTPNetwork, buf) + id := crypto.Keccak256([]byte(addr.String())) + _, err := p.DiscV5.TalkRequestToID(enode.ID(id), addr, portalwire.UTPNetwork, buf) return 0, err }, func() ([]byte, *net.UDPAddr, error) { @@ -211,6 +215,44 @@ func (p *PortalProtocol) setupDiscV5AndTable() error { return nil } +func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) { + enrSeq := p.DiscV5.LocalNode().Seq() + radiusBytes, err := p.nodeRadius.MarshalSSZ() + if err != nil { + return 0, err + } + customPayload := &portalwire.PingPongCustomData{ + Radius: radiusBytes, + } + + customPayloadBytes, err := customPayload.MarshalSSZ() + if err != nil { + return 0, err + } + + pingRequest := &portalwire.Ping{ + EnrSeq: enrSeq, + CustomPayload: customPayloadBytes, + } + + p.log.Trace("Sending ping request", "protocol", p.protocolId, "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest) + pingRequestBytes, err := pingRequest.MarshalSSZ() + if err != nil { + return 0, err + } + + talkRequestBytes := make([]byte, 0, len(pingRequestBytes)+1) + talkRequestBytes = append(talkRequestBytes, portalwire.PING) + talkRequestBytes = append(talkRequestBytes, pingRequestBytes...) + + talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes) + + if err != nil { + p.replaceNode(node) + } + return p.processPong(node, talkResp) +} + func (p *PortalProtocol) findNodes(node *enode.Node, distances []uint) ([]*enode.Node, error) { distancesBytes := make([][2]byte, len(distances)) for i, distance := range distances { @@ -241,44 +283,148 @@ func (p *PortalProtocol) findNodes(node *enode.Node, distances []uint) ([]*enode return p.processNodes(node, talkResp, distances) } -func (p *PortalProtocol) processNodes(target *enode.Node, resp []byte, distances []uint) ([]*enode.Node, error) { - var ( - nodes []*enode.Node - seen = make(map[enode.ID]struct{}) - err error - verified = 0 - ) +func (p *PortalProtocol) findContent(node *enode.Node, contentKey []byte) (byte, interface{}, error) { + findContent := &portalwire.FindContent{ + ContentKey: contentKey, + } + p.log.Trace("Sending find content request", "id", node.ID(), "findContent", findContent) + findContentBytes, err := findContent.MarshalSSZ() + if err != nil { + p.log.Error("failed to marshal find content request", "err", err) + return 0xff, nil, err + } + + talkRequestBytes := make([]byte, 0, len(findContentBytes)+1) + talkRequestBytes = append(talkRequestBytes, portalwire.FINDCONTENT) + talkRequestBytes = append(talkRequestBytes, findContentBytes...) + + talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes) + if err != nil { + p.log.Error("failed to send find content request", "err", err) + return 0xff, nil, err + } + + return p.processContent(node, talkResp) +} + +func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte, interface{}, error) { + if resp[0] != portalwire.CONTENT { + return 0xff, nil, fmt.Errorf("invalid content response") + } + + switch resp[1] { + case portalwire.ContentRawSelector: + content := &portalwire.Content{} + err := content.UnmarshalSSZ(resp[2:]) + if err != nil { + return 0xff, nil, err + } + + p.log.Trace("Received content response", "id", target.ID(), "content", content) + return resp[1], content.Content, nil + case portalwire.ContentConnIdSelector: + connIdMsg := &portalwire.ConnectionId{} + err := connIdMsg.UnmarshalSSZ(resp[2:]) + if err != nil { + return 0xff, nil, err + } + + p.log.Trace("Received content response", "id", target.ID(), "connIdMsg", connIdMsg) + rctx, rcancel := context.WithTimeout(context.Background(), defaultUTPConnectTimeout) + laddr := p.utp.Addr().(*utp.Addr) + raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()} + connId := binary.BigEndian.Uint16(connIdMsg.Id[:]) + conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(rctx), utp.WithConnId(uint32(connId))) + if err != nil { + rcancel() + return 0xff, nil, err + } + + err = conn.SetReadDeadline(time.Now().Add(defaultUTPReadTimeout)) + if err != nil { + rcancel() + return 0xff, nil, err + } + // Read ALL the data from the connection until EOF and return it + data := make([]byte, 0) + for { + buf := make([]byte, 1024) + var n int + n, err = conn.Read(buf) + if err != nil { + rcancel() + if errors.Is(err, io.EOF) { + p.log.Trace("Received content response", "id", target.ID(), "data", data, "size", n) + return resp[1], data, nil + } + + p.log.Error("failed to read from utp connection", "err", err) + return 0xff, nil, err + } + data = append(data, buf[:n]...) + } + case portalwire.ContentEnrsSelector: + enrs := &portalwire.Enrs{} + err := enrs.UnmarshalSSZ(resp[2:]) + + if err != nil { + return 0xff, nil, err + } + + p.log.Trace("Received content response", "id", target.ID(), "enrs", enrs) + + nodes := p.filterNodes(target, enrs.Enrs, nil) + return resp[1], nodes, nil + default: + return 0xff, nil, fmt.Errorf("invalid content response") + } +} + +func (p *PortalProtocol) processNodes(target *enode.Node, resp []byte, distances []uint) ([]*enode.Node, error) { if resp[0] != portalwire.NODES { return nil, fmt.Errorf("invalid nodes response") } nodesResp := &portalwire.Nodes{} - err = nodesResp.UnmarshalSSZ(resp[1:]) + err := nodesResp.UnmarshalSSZ(resp[1:]) if err != nil { return nil, err } p.table.addVerifiedNode(wrapNode(target)) - var n *enode.Node - for _, b := range nodesResp.Enrs { + nodes := p.filterNodes(target, nodesResp.Enrs, distances) + + return nodes, nil +} + +func (p *PortalProtocol) filterNodes(target *enode.Node, enrs [][]byte, distances []uint) []*enode.Node { + var ( + nodes []*enode.Node + seen = make(map[enode.ID]struct{}) + err error + verified = 0 + n *enode.Node + ) + + for _, b := range enrs { record := &enr.Record{} err = rlp.DecodeBytes(b, record) if err != nil { - p.log.Debug("Invalid record in nodes response", "id", target.ID(), "err", err) + p.log.Error("Invalid record in nodes response", "id", target.ID(), "err", err) continue } n, err = p.verifyResponseNode(target, record, distances, seen) if err != nil { - p.log.Debug("Invalid record in nodes response", "id", target.ID(), "err", err) + p.log.Error("Invalid record in nodes response", "id", target.ID(), "err", err) continue } verified++ nodes = append(nodes, n) } - p.log.Trace("Received nodes response", "id", target.ID(), "total", nodesResp.Total, "verified", verified, "nodes", nodes) - return nodes, nil + p.log.Trace("Received nodes response", "id", target.ID(), "total", len(enrs), "verified", verified, "nodes", nodes) + return nodes } func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (uint64, error) { @@ -309,16 +455,16 @@ func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (uint64, e } func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte { - if node := p.DiscV5.getNode(id); node != nil { - p.table.addSeenNode(wrapNode(node)) + if n := p.DiscV5.getNode(id); n != nil { + p.table.addSeenNode(wrapNode(n)) } p.utpPackets <- &utp.UdpMessage{Buf: msg, Addr: addr} return []byte("") } func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte { - if node := p.DiscV5.getNode(id); node != nil { - p.table.addSeenNode(wrapNode(node)) + if n := p.DiscV5.getNode(id); n != nil { + p.table.addSeenNode(wrapNode(n)) } msgCode := msg[0] @@ -459,16 +605,16 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque contentId := p.storage.ContentId(request.ContentKey) if contentId == nil { - return nil, fmt.Errorf("content not found") + return nil, ContentNotFound } var content []byte content, err = p.storage.Get(request.ContentKey, contentId) - if err != nil { + if err != nil && !errors.Is(err, ContentNotFound) { return nil, err } - if content == nil { + if errors.Is(err, ContentNotFound) { closestNodes := p.findNodesCloseToContent(contentId) for i, n := range closestNodes { if n.ID() == id { @@ -500,9 +646,21 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque return talkRespBytes, nil } else if len(content) <= maxPayloadSize { - contentMsgBytes := make([]byte, 0, len(content)+1) + rawContentMsg := &portalwire.Content{ + Content: content, + } + + p.log.Trace("Sending raw content response", "protocol", p.protocolId, "source", addr, "content", rawContentMsg) + + var rawContentMsgBytes []byte + rawContentMsgBytes, err = rawContentMsg.MarshalSSZ() + if err != nil { + return nil, err + } + + contentMsgBytes := make([]byte, 0, len(rawContentMsgBytes)+1) contentMsgBytes = append(contentMsgBytes, portalwire.ContentRawSelector) - contentMsgBytes = append(contentMsgBytes, content...) + contentMsgBytes = append(contentMsgBytes, rawContentMsgBytes...) talkRespBytes := make([]byte, 0, len(contentMsgBytes)+1) talkRespBytes = append(talkRespBytes, portalwire.CONTENT) @@ -515,7 +673,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque connIdSend := connId.SendId() go func() { - ctx, cancel := context.WithTimeout(context.Background(), defaultUTPAcceptTimeout) + ctx, cancel := context.WithTimeout(context.Background(), defaultUTPConnectTimeout) var conn *utp.Conn conn, err = p.utp.AcceptUTPContext(ctx, connIdSend) if err != nil { @@ -604,44 +762,6 @@ func (p *PortalProtocol) verifyResponseNode(sender *enode.Node, r *enr.Record, d return n, nil } -func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) { - enrSeq := p.DiscV5.LocalNode().Seq() - radiusBytes, err := p.nodeRadius.MarshalSSZ() - if err != nil { - return 0, err - } - customPayload := &portalwire.PingPongCustomData{ - Radius: radiusBytes, - } - - customPayloadBytes, err := customPayload.MarshalSSZ() - if err != nil { - return 0, err - } - - pingRequest := &portalwire.Ping{ - EnrSeq: enrSeq, - CustomPayload: customPayloadBytes, - } - - p.log.Trace("Sending ping request", "protocol", p.protocolId, "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest) - pingRequestBytes, err := pingRequest.MarshalSSZ() - if err != nil { - return 0, err - } - - talkRequestBytes := make([]byte, 0, len(pingRequestBytes)+1) - talkRequestBytes = append(talkRequestBytes, portalwire.PING) - talkRequestBytes = append(talkRequestBytes, pingRequestBytes...) - - talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes) - - if err != nil { - p.replaceNode(node) - } - return p.processPong(node, talkResp) -} - func (p *PortalProtocol) replaceNode(node *enode.Node) { p.table.mutex.Lock() defer p.table.mutex.Unlock() diff --git a/p2p/discover/portal_protocol_test.go b/p2p/discover/portal_protocol_test.go index 6bfa29e1d3..240ec33935 100644 --- a/p2p/discover/portal_protocol_test.go +++ b/p2p/discover/portal_protocol_test.go @@ -1,10 +1,12 @@ package discover import ( + "crypto/rand" "fmt" "testing" "time" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/internal/testlog" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover/portalwire" @@ -13,6 +15,26 @@ import ( "golang.org/x/exp/slices" ) +type MockStorage struct { + db map[string][]byte +} + +func (m *MockStorage) ContentId(contentKey []byte) []byte { + return crypto.Keccak256(contentKey) +} + +func (m *MockStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) { + if content, ok := m.db[string(contentId)]; ok { + return content, nil + } + return nil, ContentNotFound +} + +func (m *MockStorage) Put(contentKey []byte, content []byte) error { + m.db[string(m.ContentId(contentKey))] = content + return nil +} + func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol, error) { conf := DefaultPortalProtocolConfig() if addr != "" { @@ -21,7 +43,7 @@ func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol if bootNodes != nil { conf.BootstrapNodes = bootNodes } - portalProtocol, err := NewPortalProtocol(conf, portalwire.HistoryNetwork, newkey()) + portalProtocol, err := NewPortalProtocol(conf, portalwire.HistoryNetwork, newkey(), &MockStorage{db: make(map[string][]byte)}) if err != nil { return nil, err } @@ -76,4 +98,32 @@ func TestPortalWireProtocol(t *testing.T) { slices.ContainsFunc(node3.table.Nodes(), func(n *enode.Node) bool { return n.ID() == node2.localNode.Node().ID() }) + + err = node1.storage.Put([]byte("test_key"), []byte("test_value")) + assert.NoError(t, err) + + flag, content, err := node2.findContent(node1.localNode.Node(), []byte("test_key")) + assert.NoError(t, err) + assert.Equal(t, portalwire.ContentRawSelector, flag) + assert.Equal(t, []byte("test_value"), content) + + flag, content, err = node2.findContent(node3.localNode.Node(), []byte("test_key")) + assert.NoError(t, err) + assert.Equal(t, portalwire.ContentEnrsSelector, flag) + assert.Equal(t, 1, len(content.([]*enode.Node))) + assert.Equal(t, node1.localNode.Node().ID(), content.([]*enode.Node)[0].ID()) + + // create a byte slice of length 1199 and fill it with random data + // this will be used as a test content + largeTestContent := make([]byte, 1199) + _, err = rand.Read(largeTestContent) + assert.NoError(t, err) + + err = node1.storage.Put([]byte("large_test_key"), largeTestContent) + assert.NoError(t, err) + + //flag, content, err = node2.findContent(node1.localNode.Node(), []byte("large_test_key")) + //assert.NoError(t, err) + //assert.Equal(t, portalwire.ContentConnIdSelector, flag) + //assert.Equal(t, largeTestContent, content) } diff --git a/p2p/discover/portal_storage.go b/p2p/discover/portal_storage.go index d603cf5f79..3b9023be6a 100644 --- a/p2p/discover/portal_storage.go +++ b/p2p/discover/portal_storage.go @@ -1,5 +1,9 @@ package discover +import "fmt" + +var ContentNotFound = fmt.Errorf("content not found") + type Storage interface { ContentId(contentKey []byte) []byte