feat: Add an HTTP CONNECT proxy for the blockchain node

cmd/masa-node/main.go

@@ -14,6 +14,7 @@
 	"github.com/masa-finance/masa-oracle/pkg/api"
 	"github.com/masa-finance/masa-oracle/pkg/config"
 	"github.com/masa-finance/masa-oracle/pkg/db"
+	"github.com/masa-finance/masa-oracle/pkg/network"
 	"github.com/masa-finance/masa-oracle/pkg/staking"
 )
 
@@ -111,6 +112,17 @@
 		logrus.Info("API server is disabled")
 	}
 
+	if cfg.ProxyEnabled {
+		proxy, err := network.NewProxy(masaNode.Host, cfg.ProxyListenAddr, cfg.ProxyListenPort, cfg.ProxyTargetPort)
+		if err != nil {
+			logrus.Fatalf("[-] Error creating HTTP CONNECT proxy: %v", err)
+		}
+
+		go func(ctx context.Context) {
+			proxy.Start(ctx)
+		}(ctx)
+	}
+
 	// Get the multiaddress and IP address of the node
 	multiAddrs, err := masaNode.GetP2PMultiAddrs()
 	if err != nil {

node/options.go

@@ -22,6 +22,8 @@
 	IsTelegramScraper bool
 	IsWebScraper      bool
 
+	IsProxy bool
+
 	Bootnodes            []string
 	RandomIdentity       bool
 	Services             []func(ctx context.Context, node *OracleNode)
@@ -87,6 +89,10 @@
 	o.IsWebScraper = true
 }
 
+var IsProxy = func(o *NodeOption) {
+	o.IsProxy = true
+}
+
 func (a *NodeOption) Apply(opts ...Option) {
 	for _, opt := range opts {
 		opt(a)

node/oracle_node.go

@@ -228,7 +228,11 @@
 		go p(node.Context, node)
 	}
 
-	go myNetwork.Discover(node.Context, node.Options.Bootnodes, node.Host, node.DHT, node.Protocol)
+	protocols := []protocol.ID{node.Protocol}
+	if node.Options.IsProxy {
+		protocols = append(protocols, myNetwork.ProxyProtocol)
+	}
+	go myNetwork.Discover(node.Context, node.Options.Bootnodes, node.Host, node.DHT, protocols)
 
 	nodeData := node.NodeTracker.GetNodeData(node.Host.ID().String())
 	if nodeData == nil {

pkg/config/app.go

@@ -58,6 +58,10 @@
 	TelegramScraper    bool   `mapstructure:"telegramScraper"`
 	WebScraper         bool   `mapstructure:"webScraper"`
 	APIEnabled         bool   `mapstructure:"api_enabled"`
+	ProxyEnabled       bool   `mapstructure:"proxy_enabled"`
+	ProxyListenAddr    string `mapstructure:"proxyListenAddr"`
+	ProxyListenPort    uint16 `mapstructure:"proxyListenPort"`
+	ProxyTargetPort    uint16 `mapstructure:"proxyTargetPort"`
 
 	KeyManager   *masacrypto.KeyManager
 	TelegramStop bg.StopFunc
@@ -128,6 +132,12 @@
 	viper.SetDefault(PrivKeyFile, DefaultPrivKeyFile)
 
 	viper.SetDefault(APIEnabled, false)
+
+	viper.SetDefault(ProxyEnabled, true)
+	viper.SetDefault(ProxyListenAddr, "127.0.0.1")
+	viper.SetDefault(ProxyListenPort, "8888")
+	// TODO What is the default Cosmos port?
+	viper.SetDefault(ProxyTargetPort, "8080")
 }
 
 // setFileConfig loads configuration from a YAML file.
@@ -190,6 +200,10 @@
 	pflag.BoolVar(&c.WebScraper, "webScraper", viper.GetBool(WebScraper), "Web Scraper")
 	pflag.BoolVar(&c.Faucet, "faucet", viper.GetBool(Faucet), "Faucet")
 	pflag.BoolVar(&c.APIEnabled, "api-enabled", viper.GetBool(APIEnabled), "Enable API server")
+	pflag.BoolVar(&c.ProxyEnabled, "proxy-enabled", viper.GetBool(ProxyEnabled), "Enable CONNECT proxy")
+	pflag.StringVar(&c.ProxyListenAddr, "proxyListenAddr", viper.GetString(ProxyListenAddr), "Proxy listen address")
+	pflag.Uint16Var(&c.ProxyListenPort, "proxyListenPort", viper.GetUint16(ProxyListenPort), "Proxy listen port")
+	pflag.Uint16Var(&c.ProxyTargetPort, "proxyTargetPort", viper.GetUint16(ProxyTargetPort), "Proxy target port")
 
 	pflag.Parse()
 

pkg/config/constants.go

@@ -30,15 +30,20 @@ const (
 	BlockTopic           = "blockTopic"
 	Rendezvous           = "masa-mdns"
 	PageSize             = 25
+	DefaultPrivKeyFile   = "masa_oracle_key"
 
-	TwitterUsername    = "TWITTER_USERNAME"
-	TwitterPassword    = "TWITTER_PASSWORD"
-	Twitter2FaCode     = "TWITTER_2FA_CODE"
-	DiscordBotToken    = "DISCORD_BOT_TOKEN"
-	TwitterScraper     = "TWITTER_SCRAPER"
-	DiscordScraper     = "DISCORD_SCRAPER"
-	TelegramScraper    = "TELEGRAM_SCRAPER"
-	WebScraper         = "WEB_SCRAPER"
-	APIEnabled         = "API_ENABLED"
-	DefaultPrivKeyFile = "masa_oracle_key"
+	TwitterUsername = "TWITTER_USERNAME"
+	TwitterPassword = "TWITTER_PASSWORD"
+	Twitter2FaCode  = "TWITTER_2FA_CODE"
+	DiscordBotToken = "DISCORD_BOT_TOKEN"
+	TwitterScraper  = "TWITTER_SCRAPER"
+	DiscordScraper  = "DISCORD_SCRAPER"
+	TelegramScraper = "TELEGRAM_SCRAPER"
+	WebScraper      = "WEB_SCRAPER"
+	APIEnabled      = "API_ENABLED"
+
+	ProxyEnabled    = "PROXY_ENABLED"
+	ProxyListenAddr = "PROXY_LISTEN_ADDR"
+	ProxyListenPort = "PROXY_LISTEN_PORT"
+	ProxyTargetPort = "PROXY_TARGET_PORT"
 )

pkg/config/options.go

@@ -64,6 +64,10 @@
 		masaNodeOptions = append(masaNodeOptions, node.IsWebScraper)
 	}
 
+	if cfg.ProxyEnabled {
+		masaNodeOptions = append(masaNodeOptions, node.IsProxy)
+	}
+
 	workHandlerManager := workers.NewWorkHandlerManager(workerManagerOptions...)
 	blockChainEventTracker := node.NewBlockChain()
 	pubKeySub := &pubsub.PublicKeySubscriptionHandler{}

pkg/network/discover.go

@@ -22,26 +22,34 @@
 // It initializes discovery via the DHT and advertises this node.
 // It runs discovery in a loop with a ticker, re-advertising and finding new peers.
 // For each discovered peer, it checks if already connected, and if not, dials them.
-func Discover(ctx context.Context, bootNodes []string, host host.Host, dht *dht.IpfsDHT, protocol protocol.ID) {
+func Discover(ctx context.Context, bootNodes []string, host host.Host, dht *dht.IpfsDHT, protocols []protocol.ID) {
 	var routingDiscovery *routing.RoutingDiscovery
-	protocolString := string(protocol)
-	logrus.Infof("[+] Discovering peers for protocol: %s", protocolString)
-
-	routingDiscovery = routing.NewRoutingDiscovery(dht)
-
-	// Advertise node right away, then it will re-advertise with each ticker interval
-	logrus.Infof("[+] Attempting to advertise protocol: %s", protocolString)
-	_, err := routingDiscovery.Advertise(ctx, protocolString)
-	if err != nil {
-		logrus.Debugf("[-] Failed to advertise protocol: %v", err)
-	} else {
-		logrus.Infof("[+] Successfully advertised protocol %s", protocolString)
+
+	protos := []string{}
+	for _, p := range protocols {
+		protos = append(protos, string(p))
+	}
+
+	for _, p := range protos {
+		logrus.Infof("[+] Discovering peers for protocol: %s", p)
+
+		routingDiscovery = routing.NewRoutingDiscovery(dht)
+
+		// Advertise node right away, then it will re-advertise with each ticker interval
+		logrus.Infof("[+] Attempting to advertise protocol: %s", p)
+		_, err := routingDiscovery.Advertise(ctx, p)
+		if err != nil {
+			logrus.Debugf("[-] Failed to advertise protocol: %v", err)
+		} else {
+			logrus.Infof("[+] Successfully advertised protocol %s", p)
+		}
 	}
 
 	ticker := time.NewTicker(time.Minute * 1)
 	defer ticker.Stop()
 
 	var peerChan <-chan peer.AddrInfo
+	var err error
 
 	for {
 		select {
@@ -56,66 +64,70 @@
 			logrus.Debug("[-] Searching for other peers...")
 			routingDiscovery = routing.NewRoutingDiscovery(dht)
 
-			// Advertise this node
-			logrus.Debugf("[-] Attempting to advertise protocol: %s", protocolString)
-			_, err := routingDiscovery.Advertise(ctx, protocolString)
-			if err != nil {
-				logrus.Debugf("[-] Failed to advertise protocol with error %v", err)
-
-				// Network retry when connectivity is temporarily lost using NewExponentialBackOff
-				expBackOff := backoff.NewExponentialBackOff()
-				expBackOff.MaxElapsedTime = time.Second * 10
-				err := backoff.Retry(func() error {
-					peerChan, err = routingDiscovery.FindPeers(ctx, protocolString)
-					return err
-				}, expBackOff)
-				if err != nil {
-					logrus.Warningf("[-] Retry failed to find peers: %v", err)
-				}
-
-			} else {
-				logrus.Infof("[+] Successfully advertised protocol: %s", protocolString)
-			}
+			for _, protocolString := range protos {
+				// Advertise this node
+				logrus.Debugf("[-] Attempting to advertise protocol: %s", protocolString)
+				if _, err := routingDiscovery.Advertise(ctx, protocolString); err != nil {
+					logrus.Debugf("[-] Failed to advertise protocol with error %v", err)
+
+					// Network retry when connectivity is temporarily lost using NewExponentialBackOff
+					expBackOff := backoff.NewExponentialBackOff()
+					expBackOff.MaxElapsedTime = time.Second * 10
+					err = backoff.Retry(func() error {
+						peerChan, err = routingDiscovery.FindPeers(ctx, protocolString)
+						return err
+					}, expBackOff)
+					if err != nil {
+						logrus.Warningf("[-] Retry failed to find peers: %v", err)
+					}
 
-			// Use the routing discovery to find peers.
-			peerChan, err = routingDiscovery.FindPeers(ctx, protocolString)
-			if err != nil {
-				logrus.Errorf("[-] Failed to find peers: %v", err)
-			} else {
-				logrus.Debug("[+] Successfully started finding peers")
-			}
-			select {
-			case availPeer, ok := <-peerChan:
-				if !ok {
-					logrus.Info("[+] Peer channel closed, restarting discovery")
-					break
-				}
-				// validating proper peers to connect to
-				availPeerAddrInfo := peer.AddrInfo{
-					ID:    availPeer.ID,
-					Addrs: availPeer.Addrs,
+				} else {
+					logrus.Infof("[+] Successfully advertised protocol: %s", protocolString)
 				}
-				if availPeerAddrInfo.ID == host.ID() {
-					logrus.Debugf("Skipping connect to self: %s", availPeerAddrInfo.ID.String())
-					continue
+
+				// Use the routing discovery to find peers.
+				peerChan, err = routingDiscovery.FindPeers(ctx, protocolString)
+				if err != nil {
+					logrus.Errorf("[-] Failed to find peers: %v", err)
+				} else {
+					logrus.Debug("[+] Successfully started finding peers")
 				}
-				if len(availPeerAddrInfo.Addrs) == 0 {
-					for _, bn := range bootNodes {
-						bootNode := strings.Split(bn, "/")[len(strings.Split(bn, "/"))-1]
-						if availPeerAddrInfo.ID.String() != bootNode {
-							logrus.Warningf("Skipping connect to non bootnode peer with no multiaddress: %s", availPeerAddrInfo.ID.String())
-							continue
+
+				select {
+				case availPeer, ok := <-peerChan:
+					if !ok {
+						logrus.Info("[+] Peer channel closed, restarting discovery")
+						break
+					}
+					// validating proper peers to connect to
+					availPeerAddrInfo := peer.AddrInfo{
+						ID:    availPeer.ID,
+						Addrs: availPeer.Addrs,
+					}
+					if availPeerAddrInfo.ID == host.ID() {
+						logrus.Debugf("Skipping connect to self: %s", availPeerAddrInfo.ID.String())
+						continue
+					}
+					if len(availPeerAddrInfo.Addrs) == 0 {
+						for _, bn := range bootNodes {
+							bootNode := strings.Split(bn, "/")[len(strings.Split(bn, "/"))-1]
+							if availPeerAddrInfo.ID.String() != bootNode {
+								logrus.Warningf("Skipping connect to non bootnode peer with no multiaddress: %s", availPeerAddrInfo.ID.String())
+								continue
+							}
 						}
 					}
-				}
-				logrus.Infof("[+] Available Peer: %s", availPeer.String())
-
-				if host.Network().Connectedness(availPeer.ID) != network.Connected {
-					if isConnectedToBootnode(host, bootNodes) {
-						_, err := host.Network().DialPeer(ctx, availPeer.ID)
-						if err != nil {
-							logrus.Warningf("[-] Failed to connect to peer %s, will retry...", availPeer.ID.String())
-							continue
+					logrus.Infof("[+] Available Peer: %s", availPeer.String())
+
+					if host.Network().Connectedness(availPeer.ID) != network.Connected {
+						if isConnectedToBootnode(host, bootNodes) {
+							_, err := host.Network().DialPeer(ctx, availPeer.ID)
+							if err != nil {
+								logrus.Warningf("[-] Failed to connect to peer %s, will retry...", availPeer.ID.String())
+								continue
+							} else {
+								logrus.Infof("[+] Connected to peer %s", availPeer.ID.String())
+							}
 						} else {
 							for _, bn := range bootNodes {
 								if len(bn) > 0 {
@@ -125,10 +137,10 @@
 							}
 						}
 					}
+				case <-ctx.Done():
+					logrus.Info("[-] Stopping peer discovery")
+					return
 				}
-			case <-ctx.Done():
-				logrus.Info("[-] Stopping peer discovery")
-				return
 			}
 		}
 	}

pkg/network/mdns.go

@@ -17,7 +17,7 @@ type discoveryNotifee struct {
 func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
 	pe := PeerEvent{
 		AddrInfo:   pi,
-		Action:     "PeerFound",
+		Action:     PeerFound,
 		Source:     "mdns",
 		Rendezvous: n.Rendezvous,
 	}