From 2b6c5eb29266ad0bc1d7f1371ed60243f6f73dd5 Mon Sep 17 00:00:00 2001 From: fearlessfe <505380967@qq.com> Date: Thu, 14 Mar 2024 22:59:21 +0800 Subject: [PATCH] feat: add cli and fix typ error --- cmd/shisui/config_test.go | 35 ++++ cmd/shisui/main.go | 209 ++++++++++++++--------- cmd/utils/flags.go | 67 ++++++++ internal/flags/categories.go | 38 +++-- node/defaults.go | 2 + p2p/discover/api.go | 26 +++ p2p/discover/portal_protocol.go | 50 +++--- portalnetwork/history/history_network.go | 1 + 8 files changed, 310 insertions(+), 118 deletions(-) create mode 100644 cmd/shisui/config_test.go diff --git a/cmd/shisui/config_test.go b/cmd/shisui/config_test.go new file mode 100644 index 0000000000..e22fac5dd5 --- /dev/null +++ b/cmd/shisui/config_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "flag" + "testing" + + "github.com/stretchr/testify/require" + "github.com/urfave/cli/v2" +) + +func TestGenConfig(t *testing.T) { + size := uint64(1000 * 1000) + flagSet := flag.NewFlagSet("test", 0) + flagSet.String("history.http.addr", "127.0.0.11", "test") + flagSet.String("history.http.port", "8888", "test") + flagSet.String("history.data.dir", "./test", "test") + flagSet.Uint64("history.data.capacity", size, "test") + flagSet.String("udp.addr", "172.23.50.11", "test") + flagSet.Int("udp.port", 9999, "test") + flagSet.Int("history.loglevel", 3, "test") + + command := &cli.Command{Name: "mycommand"} + + ctx := cli.NewContext(nil, flagSet, nil) + ctx.Command = command + + config, err := getPortalHistoryConfig(ctx) + require.NoError(t, err) + + require.Equal(t, config.DataCapacity, size) + require.Equal(t, config.DataDir, "./test") + require.Equal(t, config.LogLevel, 3) + require.Equal(t, config.RpcAddr, "127.0.0.11:8888") + require.Equal(t, config.Protocol.ListenAddr, ":9999") +} diff --git a/cmd/shisui/main.go b/cmd/shisui/main.go index f25d53e712..501a7b7761 100644 --- a/cmd/shisui/main.go +++ b/cmd/shisui/main.go @@ -2,124 +2,177 @@ package main import ( "crypto/ecdsa" - "crypto/x509" - "encoding/pem" - "errors" - "net/http" + "fmt" + "net" + "strings" + "os" + "github.com/ethereum/go-ethereum/cmd/utils" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/internal/flags" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/discover/portalwire" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/portalnetwork/history" "github.com/ethereum/go-ethereum/portalnetwork/storage/sqlite" - "github.com/ethereum/go-ethereum/rpc" + "github.com/urfave/cli/v2" ) +type PortalHistoryConfig struct { + Protocol *discover.PortalProtocolConfig + PrivateKey *ecdsa.PrivateKey + RpcAddr string + DataDir string + DataCapacity uint64 + LogLevel int +} + +var app = flags.NewApp("the go-portal-network command line interface") + +var ( + portalProtocolFlags = []cli.Flag{ + utils.ProtocolUDPListenAddrFlag, + utils.ProtocolUDPPortFlag, + } + historyRpcFlags = []cli.Flag{ + utils.HistoryHTTPListenAddrFlag, + utils.HistoryHTTPPortFlag, + utils.HistoryDataDirFlag, + utils.HistoryDataCapacityFlag, + utils.LogLevelFlag, + } + hiveTestFlags = []cli.Flag{ + utils.HiveBootNodeFlag, + utils.HiveClientPrivateKeyFlag, + utils.HiveLogLevelFlag, + } +) + +func init() { + app.Action = shisui + app.Flags = flags.Merge(portalProtocolFlags, historyRpcFlags) + flags.AutoEnvVars(app.Flags, "SHISUI") + + app.Flags = flags.Merge(app.Flags, hiveTestFlags) + flags.AutoEnvVars(hiveTestFlags, "HIVE") +} + func main() { + if err := app.Run(os.Args); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func shisui(ctx *cli.Context) error { + config, err := getPortalHistoryConfig(ctx) + if err != nil { + return nil + } + glogger := log.NewGlogHandler(log.NewTerminalHandler(os.Stderr, true)) - slogVerbosity := log.FromLegacyLevel(5) + slogVerbosity := log.FromLegacyLevel(config.LogLevel) glogger.Verbosity(slogVerbosity) log.SetDefault(log.NewLogger(glogger)) - var privateKey *ecdsa.PrivateKey - var err error - privateKeyHex := os.Getenv("HIVE_CLIENT_PRIVATE_KEY") - if privateKeyHex != "" { - keyBytes, err := hexutil.Decode("0x" + privateKeyHex) - if err != nil { - panic(err) - } - privateKey, err = crypto.ToECDSA(keyBytes) - if err != nil { - panic(err) - } - } else { - privateKey, err = crypto.GenerateKey() - if err != nil { - panic(err) - } - } - - config := discover.DefaultPortalProtocolConfig() - - bootNodeStr := os.Getenv("HIVE_BOOTNODE") - if bootNodeStr != "" { - bootNode := new(enode.Node) - err = bootNode.UnmarshalText([]byte(bootNodeStr)) - if err != nil { - panic(err) - } - config.BootstrapNodes = append(config.BootstrapNodes, bootNode) - } - - udpPort := os.Getenv("UDP_PORT") - - if udpPort != "" { - config.ListenAddr = ":" + udpPort - } - nodeId := enode.PubkeyToIDV4(&privateKey.PublicKey) - contentStorage, err := sqlite.NewContentStorage(1000*1000*1000, nodeId, "./") + nodeId := enode.PubkeyToIDV4(&config.PrivateKey.PublicKey) + contentStorage, err := sqlite.NewContentStorage(config.DataCapacity, nodeId, config.DataDir) if err != nil { - panic(err) + return err } contentQueue := make(chan *discover.ContentElement, 50) - protocol, err := discover.NewPortalProtocol(config, string(portalwire.HistoryNetwork), privateKey, contentStorage, contentQueue) + protocol, err := discover.NewPortalProtocol(config.Protocol, string(portalwire.HistoryNetwork), config.PrivateKey, contentStorage, contentQueue) if err != nil { - panic(err) + return err } accumulator, err := history.NewMasterAccumulator() if err != nil { - panic(err) + return err } historyNetwork := history.NewHistoryNetwork(protocol, &accumulator) err = historyNetwork.Start() if err != nil { - panic(err) + return err } + defer historyNetwork.Stop() - disv5 := discover.NewAPI(protocol.DiscV5) - portal := discover.NewPortalAPI(protocol) + discover.StartHistoryRpcServer(protocol, config.RpcAddr) - server := rpc.NewServer() - server.RegisterName("discv5", disv5) - server.RegisterName("portal", portal) - - tcpPort := os.Getenv("TCP_PORT") - - if tcpPort == "" { - tcpPort = "8545" - } - - httpServer := &http.Server{ - Addr: ":" + tcpPort, - Handler: server, - } - - httpServer.ListenAndServe() + return nil } -func ReadKeyFromFile(name string) (*ecdsa.PrivateKey, error) { - keyBytes, err := os.ReadFile(name) - if err != nil { - return nil, err +func getPortalHistoryConfig(ctx *cli.Context) (*PortalHistoryConfig, error) { + config := &PortalHistoryConfig{ + Protocol: discover.DefaultPortalProtocolConfig(), } - block, _ := pem.Decode(keyBytes) - if block == nil { - return nil, errors.New("failed to decode PEM block") + err := setPrivateKey(ctx, config) + if err != nil { + return config, err } - privateKey, err := x509.ParseECPrivateKey(block.Bytes) - if err != nil { - return nil, err + httpAddr := ctx.String(utils.HistoryHTTPListenAddrFlag.Name) + httpPort := ctx.String(utils.HistoryHTTPPortFlag.Name) + config.RpcAddr = net.JoinHostPort(httpAddr, httpPort) + config.DataDir = ctx.String(utils.HistoryDataDirFlag.Name) + config.DataCapacity = ctx.Uint64(utils.HistoryDataCapacityFlag.Name) + config.LogLevel = ctx.Int(utils.LogLevelFlag.Name) + port := ctx.String(utils.ProtocolUDPPortFlag.Name) + if !strings.HasPrefix(port, ":") { + config.Protocol.ListenAddr = ":" + port + } else { + config.Protocol.ListenAddr = port } - return privateKey, nil + + if ctx.IsSet(utils.ProtocolUDPListenAddrFlag.Name) { + ip := ctx.String(utils.ProtocolUDPListenAddrFlag.Name) + netIp := net.ParseIP(ip) + if netIp == nil { + return config, fmt.Errorf("invalid ip addr: %s", ip) + } + config.Protocol.NodeIP = netIp + } + + if ctx.IsSet(utils.HiveLogLevelFlag.Name) { + config.LogLevel = ctx.Int(utils.HiveLogLevelFlag.Name) + } + if ctx.IsSet(utils.HiveBootNodeFlag.Name) { + bootNode := new(enode.Node) + err = bootNode.UnmarshalText([]byte(ctx.String(utils.HiveBootNodeFlag.Name))) + if err != nil { + return config, err + } + config.Protocol.BootstrapNodes = append(config.Protocol.BootstrapNodes, bootNode) + } + return config, nil +} + +func setPrivateKey(ctx *cli.Context, config *PortalHistoryConfig) error { + var privateKey *ecdsa.PrivateKey + var err error + if ctx.IsSet(utils.HiveClientPrivateKeyFlag.Name) { + keyStr := ctx.String(utils.HiveClientPrivateKeyFlag.Name) + keyBytes, err := hexutil.Decode("0x" + keyStr) + if err != nil { + return err + } + privateKey, err = crypto.ToECDSA(keyBytes) + if err != nil { + return err + } + } else { + privateKey, err = crypto.GenerateKey() + if err != nil { + return err + } + } + config.PrivateKey = privateKey + return nil } diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index b38f33b8dd..92c4fc33c4 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -944,6 +944,73 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server. Value: metrics.DefaultConfig.InfluxDBOrganization, Category: flags.MetricsCategory, } + + HistoryHTTPListenAddrFlag = &cli.StringFlag{ + Name: "history.http.addr", + Usage: "HTTP-RPC server listening interface", + Value: node.DefaultHTTPHost, + Category: flags.PortalNetworkCategory, + } + + HistoryHTTPPortFlag = &cli.IntFlag{ + Name: "history.http.port", + Usage: "HTTP-RPC server listening port", + Value: node.DefaultHTTPPort, + Category: flags.PortalNetworkCategory, + } + + HistoryDataDirFlag = &cli.StringFlag{ + Name: "history.data.dir", + Usage: "data dir of where the data file located", + Value: "./", + Category: flags.PortalNetworkCategory, + } + + HistoryDataCapacityFlag = &cli.Uint64Flag{ + Name: "history.data.capacity", + Usage: "the capacity of the data stored, the unit is byte", + Value: 1000 * 1000 * 1000, // 1 GB + Category: flags.PortalNetworkCategory, + } + + ProtocolUDPListenAddrFlag = &cli.StringFlag{ + Name: "udp.addr", + Usage: "protocol UDP server listening interface", + Value: "", + Category: flags.PortalNetworkCategory, + } + + ProtocolUDPPortFlag = &cli.IntFlag{ + Name: "udp.port", + Usage: "protocol UDP server listening port", + Value: node.DefaultUDPPort, + Category: flags.PortalNetworkCategory, + } + + LogLevelFlag = &cli.IntFlag{ + Name: "history.loglevel", + Usage: "loglevel of portal network", + Value: node.DetaultLoglevel, + Category: flags.PortalNetworkCategory, + } + + HiveBootNodeFlag = &cli.StringFlag{ + Name: "bootnode", + Usage: "bootnode of p2p network with ENR format for portal hive test", + Category: flags.PortalNetworkHiveCategory, + } + + HiveClientPrivateKeyFlag = &cli.StringFlag{ + Name: "client.private.key", + Usage: "private key of current p2p node for portal hive test", + Category: flags.PortalNetworkHiveCategory, + } + + HiveLogLevelFlag = &cli.IntFlag{ + Name: "loglevel", + Usage: "loglevel for portal hive test", + Category: flags.PortalNetworkHiveCategory, + } ) var ( diff --git a/internal/flags/categories.go b/internal/flags/categories.go index d426add55b..2ee216d140 100644 --- a/internal/flags/categories.go +++ b/internal/flags/categories.go @@ -19,24 +19,26 @@ package flags import "github.com/urfave/cli/v2" const ( - EthCategory = "ETHEREUM" - BeaconCategory = "BEACON CHAIN" - DevCategory = "DEVELOPER CHAIN" - StateCategory = "STATE HISTORY MANAGEMENT" - TxPoolCategory = "TRANSACTION POOL (EVM)" - BlobPoolCategory = "TRANSACTION POOL (BLOB)" - PerfCategory = "PERFORMANCE TUNING" - AccountCategory = "ACCOUNT" - APICategory = "API AND CONSOLE" - NetworkingCategory = "NETWORKING" - MinerCategory = "MINER" - GasPriceCategory = "GAS PRICE ORACLE" - VMCategory = "VIRTUAL MACHINE" - LoggingCategory = "LOGGING AND DEBUGGING" - MetricsCategory = "METRICS AND STATS" - MiscCategory = "MISC" - TestingCategory = "TESTING" - DeprecatedCategory = "ALIASED (deprecated)" + EthCategory = "ETHEREUM" + BeaconCategory = "BEACON CHAIN" + DevCategory = "DEVELOPER CHAIN" + StateCategory = "STATE HISTORY MANAGEMENT" + TxPoolCategory = "TRANSACTION POOL (EVM)" + BlobPoolCategory = "TRANSACTION POOL (BLOB)" + PerfCategory = "PERFORMANCE TUNING" + AccountCategory = "ACCOUNT" + APICategory = "API AND CONSOLE" + NetworkingCategory = "NETWORKING" + MinerCategory = "MINER" + GasPriceCategory = "GAS PRICE ORACLE" + VMCategory = "VIRTUAL MACHINE" + LoggingCategory = "LOGGING AND DEBUGGING" + MetricsCategory = "METRICS AND STATS" + MiscCategory = "MISC" + TestingCategory = "TESTING" + DeprecatedCategory = "ALIASED (deprecated)" + PortalNetworkCategory = "PORTAL NETWORK" + PortalNetworkHiveCategory = "PORTAL NETWORK FOR HIVE TEST" ) func init() { diff --git a/node/defaults.go b/node/defaults.go index 307d9e186a..47511a98e8 100644 --- a/node/defaults.go +++ b/node/defaults.go @@ -34,6 +34,8 @@ const ( DefaultWSPort = 8546 // Default TCP port for the websocket RPC server DefaultAuthHost = "localhost" // Default host interface for the authenticated apis DefaultAuthPort = 8551 // Default port for the authenticated apis + DefaultUDPPort = 9009 // Default UDP port for the p2p network + DetaultLoglevel = 1 // Default loglevel for portal network, which is error level ) const ( diff --git a/p2p/discover/api.go b/p2p/discover/api.go index 73bbd8859b..e14ddddac0 100644 --- a/p2p/discover/api.go +++ b/p2p/discover/api.go @@ -2,10 +2,12 @@ package discover import ( "errors" + "net/http" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/p2p/discover/portalwire" "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/rpc" "github.com/holiman/uint256" ) @@ -71,6 +73,30 @@ type Enrs struct { Enrs []string `json:"enrs"` } +func StartHistoryRpcServer(protocol *PortalProtocol, addr string) error { + disv5 := NewAPI(protocol.DiscV5) + portal := NewPortalAPI(protocol) + + server := rpc.NewServer() + err := server.RegisterName("discv5", disv5) + if err != nil { + return err + } + err = server.RegisterName("portal", portal) + + if err != nil { + return err + } + + httpServer := &http.Server{ + Addr: addr, + Handler: server, + } + + httpServer.ListenAndServe() + return nil +} + func (d *DiscV5API) NodeInfo() *NodeInfo { n := d.DiscV5.LocalNode().Node() diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index a51f02a137..5fb34db6ac 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -120,12 +120,12 @@ type OfferRequestWithNode struct { Node *enode.Node } -type ContentInfoRes struct { +type ContentInfoResp struct { Content []byte UtpTransfer bool } -type traceContentInfoRes struct { +type traceContentInfoResp struct { Node *enode.Node Flag byte Content any @@ -135,8 +135,8 @@ type traceContentInfoRes struct { type PortalProtocolOption func(p *PortalProtocol) type PortalProtocolConfig struct { - BootstrapNodes []*enode.Node - + BootstrapNodes []*enode.Node + NodeIP net.IP ListenAddr string NetRestrict *netutil.Netlist NodeRadius *uint256.Int @@ -199,21 +199,27 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateK localNode := enode.NewLocalNode(nodeDB, privateKey) localNode.SetFallbackIP(net.IP{127, 0, 0, 1}) localNode.Set(tag) - addrs, err := net.InterfaceAddrs() - if err != nil { - return nil, err - } + if config.NodeIP != nil { + localNode.SetStaticIP(config.NodeIP) + } else { + addrs, err := net.InterfaceAddrs() - for _, address := range addrs { - // check ip addr is loopback addr - if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { - if ipnet.IP.To4() != nil { - localNode.SetStaticIP(ipnet.IP) - break + if err != nil { + return nil, err + } + + for _, address := range addrs { + // check ip addr is loopback addr + if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { + if ipnet.IP.To4() != nil { + localNode.SetStaticIP(ipnet.IP) + break + } } } } + closeCtx, cancelCloseCtx := context.WithCancel(context.Background()) protocol := &PortalProtocol{ @@ -1464,7 +1470,7 @@ func (p *PortalProtocol) collectTableNodes(rip net.IP, distances []uint, limit i func (p *PortalProtocol) ContentLookup(contentKey []byte) ([]byte, bool, error) { lookupContext, cancel := context.WithCancel(context.Background()) defer cancel() - resChan := make(chan *ContentInfoRes, 1) + resChan := make(chan *ContentInfoResp, 1) defer close(resChan) newLookup(lookupContext, p.table, p.Self().ID(), func(n *node) ([]*node, error) { return p.contentLookupWorker(unwrapNode(n), contentKey, resChan) @@ -1477,7 +1483,7 @@ func (p *PortalProtocol) ContentLookup(contentKey []byte) ([]byte, bool, error) return nil, false, ContentNotFound } -func (p *PortalProtocol) contentLookupWorker(n *enode.Node, contentKey []byte, resChan chan<- *ContentInfoRes) ([]*node, error) { +func (p *PortalProtocol) contentLookupWorker(n *enode.Node, contentKey []byte, resChan chan<- *ContentInfoResp) ([]*node, error) { wrapedNode := make([]*node, 0) flag, content, err := p.findContent(n, contentKey) if err != nil { @@ -1489,7 +1495,7 @@ func (p *PortalProtocol) contentLookupWorker(n *enode.Node, contentKey []byte, r if !ok { return wrapedNode, fmt.Errorf("failed to assert to raw content, value is: %v", content) } - res := &ContentInfoRes{ + res := &ContentInfoResp{ Content: content, } if flag == portalwire.ContentConnIdSelector { @@ -1511,10 +1517,10 @@ func (p *PortalProtocol) TraceContentLookup(contentKey []byte) (*TraceContentRes lookupContext, cancel := context.WithCancel(context.Background()) defer cancel() requestNodeChan := make(chan *enode.Node, 3) - resChan := make(chan *traceContentInfoRes, 3) + resChan := make(chan *traceContentInfoResp, 3) requestNode := make([]*enode.Node, 0) - requestRes := make(map[string]*traceContentInfoRes) + requestRes := make(map[string]*traceContentInfoResp) traceContentRes := &TraceContentResult{} @@ -1613,7 +1619,7 @@ func (p *PortalProtocol) TraceContentLookup(contentKey []byte) (*TraceContentRes return traceContentRes, nil } -func (p *PortalProtocol) traceContentLookupWorker(n *enode.Node, contentKey []byte, resChan chan<- *traceContentInfoRes) ([]*node, error) { +func (p *PortalProtocol) traceContentLookupWorker(n *enode.Node, contentKey []byte, resChan chan<- *traceContentInfoResp) ([]*node, error) { wrapedNode := make([]*node, 0) flag, content, err := p.findContent(n, contentKey) if err != nil { @@ -1625,7 +1631,7 @@ func (p *PortalProtocol) traceContentLookupWorker(n *enode.Node, contentKey []by if !ok { return wrapedNode, fmt.Errorf("failed to assert to raw content, value is: %v", content) } - res := &traceContentInfoRes{ + res := &traceContentInfoResp{ Node: n, Flag: flag, Content: content, @@ -1641,7 +1647,7 @@ func (p *PortalProtocol) traceContentLookupWorker(n *enode.Node, contentKey []by if !ok { return wrapedNode, fmt.Errorf("failed to assert to enrs content, value is: %v", content) } - resChan <- &traceContentInfoRes{Node: n, + resChan <- &traceContentInfoResp{Node: n, Flag: flag, Content: content, UtpTransfer: false} diff --git a/portalnetwork/history/history_network.go b/portalnetwork/history/history_network.go index 24f1191ac5..89d7d03450 100644 --- a/portalnetwork/history/history_network.go +++ b/portalnetwork/history/history_network.go @@ -82,6 +82,7 @@ func (h *HistoryNetwork) Start() error { return err } go h.processContentLoop(h.closeCtx) + h.log.Debug("history network start successfully") return nil }