:gear: Add egress service

Allow nodes flagged as egress to handle traffic to outside. Clients can
bind a local http proxy server to forward requests to egress nodes.

cmd/dns.go

@@ -0,0 +1,88 @@
+// Copyright © 2021 Ettore Di Giacinto <[email protected]>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along
+// with this program; if not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"time"
+
+	"github.com/mudler/edgevpn/pkg/node"
+	"github.com/mudler/edgevpn/pkg/services"
+	"github.com/urfave/cli"
+)
+
+func DNS() cli.Command {
+	return cli.Command{
+		Name:        "dns",
+		Usage:       "Starts a local dns server",
+		Description: `Start a local dns server which uses the blockchain to resolve addresses`,
+		UsageText:   "edgevpn dns",
+		Flags: append(CommonFlags,
+			&cli.StringFlag{
+				Name:   "listen",
+				Usage:  "DNS listening address. Empty to disable dns server",
+				EnvVar: "DNSADDRESS",
+				Value:  "",
+			},
+			&cli.BoolTFlag{
+				Name:   "dns-forwarder",
+				Usage:  "Enables dns forwarding",
+				EnvVar: "DNSFORWARD",
+			},
+			&cli.IntFlag{
+				Name:   "dns-cache-size",
+				Usage:  "DNS LRU cache size",
+				EnvVar: "DNSCACHESIZE",
+				Value:  200,
+			},
+			&cli.StringSliceFlag{
+				Name:   "dns-forward-server",
+				Usage:  "List of DNS forward server, e.g. 8.8.8.8:53, 192.168.1.1:53 ...",
+				EnvVar: "DNSFORWARDSERVER",
+				Value:  &cli.StringSlice{"8.8.8.8:53", "1.1.1.1:53"},
+			},
+		),
+		Action: func(c *cli.Context) error {
+			o, _, ll := cliToOpts(c)
+
+			dns := c.String("listen")
+			// Adds DNS Server
+			o = append(o,
+				services.DNS(dns,
+					c.Bool("dns-forwarder"),
+					c.StringSlice("dns-forward-server"),
+					c.Int("dns-cache-size"),
+				)...)
+
+			e, err := node.New(o...)
+			if err != nil {
+				return err
+			}
+
+			displayStart(ll)
+
+			ctx := context.Background()
+			// Start the node to the network, using our ledger
+			if err := e.Start(ctx); err != nil {
+				return err
+			}
+
+			for {
+				time.Sleep(1 * time.Second)
+			}
+		},
+	}
+}

cmd/main.go

@@ -101,6 +101,17 @@ func MainFlags() []cli.Flag {
 			Usage:  "Enables dns forwarding",
 			EnvVar: "DNSFORWARD",
 		},
+		&cli.BoolFlag{
+			Name:   "egress",
+			Usage:  "Enables nodes for egress",
+			EnvVar: "EGRESS",
+		},
+		&cli.IntFlag{
+			Name:   "egress-announce-time",
+			Usage:  "Egress announce time (s)",
+			EnvVar: "EGRESSANNOUNCE",
+			Value:  200,
+		},
 		&cli.IntFlag{
 			Name:   "dns-cache-size",
 			Usage:  "DNS LRU cache size",
@@ -159,11 +170,14 @@ func Main() func(c *cli.Context) error {
 		}
 		o, vpnOpts, ll := cliToOpts(c)
 
+		// Egress and DHCP needs the Alive service
+		// DHCP needs alive services enabled to all nodes, also those with a static IP.
 		o = append(o,
 			services.Alive(
 				time.Duration(c.Int("aliveness-healthcheck-interval"))*time.Second,
 				time.Duration(c.Int("aliveness-healthcheck-scrub-interval"))*time.Second,
 				time.Duration(c.Int("aliveness-healthcheck-max-interval"))*time.Second)...)
+
 		if c.Bool("dhcp") {
 			// Adds DHCP server
 			address, _, err := net.ParseCIDR(c.String("address"))
@@ -175,6 +189,10 @@ func Main() func(c *cli.Context) error {
 			vpnOpts = append(vpnOpts, vO...)
 		}
 
+		if c.Bool("egress") {
+			o = append(o, services.Egress(time.Duration(c.Int("egress-announce-time"))*time.Second)...)
+		}
+
 		dns := c.String("dns")
 		if dns != "" {
 			// Adds DNS Server

cmd/proxy.go

@@ -0,0 +1,77 @@
+// Copyright © 2021 Ettore Di Giacinto <[email protected]>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along
+// with this program; if not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"time"
+
+	"github.com/mudler/edgevpn/api"
+	"github.com/mudler/edgevpn/pkg/node"
+	"github.com/mudler/edgevpn/pkg/services"
+	"github.com/urfave/cli"
+)
+
+func Proxy() cli.Command {
+	return cli.Command{
+		Name:        "proxy",
+		Usage:       "Starts a local http proxy server to egress nodes",
+		Description: `Start a proxy locally, providing an ingress point for the network.`,
+		UsageText:   "edgevpn proxy",
+		Flags: append(CommonFlags,
+			&cli.StringFlag{
+				Name:   "listen",
+				Value:  ":8080",
+				Usage:  "Listening address",
+				EnvVar: "PROXYLISTEN",
+			},
+			&cli.IntFlag{
+				Name:   "interval",
+				Usage:  "proxy announce time interval",
+				EnvVar: "PROXYINTERVAL",
+				Value:  120,
+			},
+			&cli.IntFlag{
+				Name:   "dead-interval",
+				Usage:  "interval (in seconds) wether detect egress nodes offline",
+				EnvVar: "PROXYDEADINTERVAL",
+				Value:  600,
+			},
+		),
+		Action: func(c *cli.Context) error {
+			o, _, ll := cliToOpts(c)
+
+			o = append(o, services.Proxy(
+				time.Duration(c.Int("interval"))*time.Second,
+				time.Duration(c.Int("dead-interval"))*time.Second,
+				c.String("listen"))...)
+			e, err := node.New(o...)
+			if err != nil {
+				return err
+			}
+
+			displayStart(ll)
+
+			ctx := context.Background()
+			// Start the node to the network, using our ledger
+			if err := e.Start(ctx); err != nil {
+				return err
+			}
+
+			return api.API(ctx, c.String("listen"), 5*time.Second, 20*time.Second, e)
+		},
+	}
+}

cmd/service.go

@@ -118,19 +118,24 @@ to the service over the network`,
 				return err
 			}
 			o, _, ll := cliToOpts(c)
-			e, err := node.New(o...)
+			e, err := node.New(
+				append(o,
+					node.WithNetworkService(
+						services.ConnectNetworkService(
+							time.Duration(c.Int("ledger-announce-interval"))*time.Second,
+							name,
+							address,
+						),
+					),
+				)...,
+			)
 			if err != nil {
 				return err
 			}
 			displayStart(ll)
 
-			// Join the node to the network, using our ledger
-			if err := e.Start(context.Background()); err != nil {
-				return err
-			}
-
-			ledger, _ := e.Ledger()
-			return services.ConnectToService(context.Background(), ledger, e, ll, time.Duration(c.Int("ledger-announce-interval"))*time.Second, name, address)
+			// starts the node
+			return e.Start(context.Background())
 		},
 	}
 }

main.go

@@ -41,7 +41,9 @@ func main() {
 			cmd.ServiceAdd(),
 			cmd.ServiceConnect(),
 			cmd.FileReceive(),
+			cmd.Proxy(),
 			cmd.FileSend(),
+			cmd.DNS(),
 		},
 
 		Action: cmd.Main(),

pkg/protocol/protocol.go

@@ -23,6 +23,7 @@ const (
 	EdgeVPN         Protocol = "/edgevpn/0.1"
 	ServiceProtocol Protocol = "/edgevpn/service/0.1"
 	FileProtocol    Protocol = "/edgevpn/file/0.1"
+	EgressProtocol  Protocol = "/edgevpn/egress/0.1"
 )
 
 const (
@@ -32,6 +33,7 @@ const (
 	UsersLedgerKey    = "users"
 	HealthCheckKey    = "healthcheck"
 	DNSKey            = "dns"
+	EgressService     = "egress"
 )
 
 type Protocol string

pkg/services/egress.go

@@ -0,0 +1,224 @@
+// Copyright © 2021-2022 Ettore Di Giacinto <[email protected]>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along
+// with this program; if not, see <http://www.gnu.org/licenses/>.
+
+package services
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"log"
+	"math/rand"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/network"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/mudler/edgevpn/pkg/blockchain"
+	"github.com/mudler/edgevpn/pkg/node"
+	"github.com/mudler/edgevpn/pkg/protocol"
+	"github.com/mudler/edgevpn/pkg/types"
+)
+
+func egressHandler(n *node.Node, b *blockchain.Ledger) func(stream network.Stream) {
+	return func(stream network.Stream) {
+		// Remember to close the stream when we are done.
+		defer stream.Close()
+
+		// Retrieve current ID for ip in the blockchain
+		_, found := b.GetKey(protocol.UsersLedgerKey, stream.Conn().RemotePeer().String())
+		// If mismatch, update the blockchain
+		if !found {
+			//		ll.Debugf("Reset '%s': not found in the ledger", stream.Conn().RemotePeer().String())
+			stream.Reset()
+			return
+		}
+
+		// Create a new buffered reader, as ReadRequest needs one.
+		// The buffered reader reads from our stream, on which we
+		// have sent the HTTP request (see ServeHTTP())
+		buf := bufio.NewReader(stream)
+		// Read the HTTP request from the buffer
+		req, err := http.ReadRequest(buf)
+		if err != nil {
+			stream.Reset()
+			log.Println(err)
+			return
+		}
+		defer req.Body.Close()
+
+		// We need to reset these fields in the request
+		// URL as they are not maintained.
+		req.URL.Scheme = "http"
+		hp := strings.Split(req.Host, ":")
+		if len(hp) > 1 && hp[1] == "443" {
+			req.URL.Scheme = "https"
+		} else {
+			req.URL.Scheme = "http"
+		}
+		req.URL.Host = req.Host
+
+		outreq := new(http.Request)
+		*outreq = *req
+
+		// We now make the request
+		//fmt.Printf("Making request to %s\n", req.URL)
+		resp, err := http.DefaultTransport.RoundTrip(outreq)
+		if err != nil {
+			stream.Reset()
+			log.Println(err)
+			return
+		}
+
+		// resp.Write writes whatever response we obtained for our
+		// request back to the stream.
+		resp.Write(stream)
+	}
+}
+
+// ProxyService starts a local http proxy server which redirects requests to egresses into the network
+// It takes a deadtime to consider hosts which are alive within a time window
+func ProxyService(announceTime time.Duration, listenAddr string, deadtime time.Duration) node.NetworkService {
+	return func(ctx context.Context, c node.Config, n *node.Node, b *blockchain.Ledger) error {
+
+		ps := &proxyService{
+			host:       n,
+			listenAddr: listenAddr,
+			deadTime:   deadtime,
+		}
+
+		// Announce ourselves so nodes accepts our connection
+		b.Announce(
+			ctx,
+			announceTime,
+			func() {
+				// Retrieve current ID for ip in the blockchain
+				_, found := b.GetKey(protocol.UsersLedgerKey, n.Host().ID().String())
+				// If mismatch, update the blockchain
+				if !found {
+					updatedMap := map[string]interface{}{}
+					updatedMap[n.Host().ID().String()] = &types.User{
+						PeerID:    n.Host().ID().String(),
+						Timestamp: time.Now().String(),
+					}
+					b.Add(protocol.UsersLedgerKey, updatedMap)
+				}
+			},
+		)
+
+		go ps.Serve()
+		return nil
+	}
+}
+
+type proxyService struct {
+	host       *node.Node
+	listenAddr string
+	deadTime   time.Duration
+}
+
+func (p *proxyService) Serve() error {
+	return http.ListenAndServe(p.listenAddr, p)
+}
+
+func (p *proxyService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+
+	l, err := p.host.Ledger()
+	if err != nil {
+		//fmt.Printf("no ledger")
+		return
+	}
+
+	egress := l.CurrentData()[protocol.EgressService]
+	nodes := AvailableNodes(l, p.deadTime)
+
+	availableEgresses := []string{}
+	for _, n := range nodes {
+		for e := range egress {
+			if e == n {
+				availableEgresses = append(availableEgresses, e)
+			}
+		}
+	}
+
+	chosen := availableEgresses[rand.Intn(len(availableEgresses)-1)]
+
+	//fmt.Printf("proxying request for %s to peer %s\n", r.URL, chosen)
+	// We need to send the request to the remote libp2p peer, so
+	// we open a stream to it
+	stream, err := p.host.Host().NewStream(context.Background(), peer.ID(chosen), protocol.EgressProtocol.ID())
+	// If an error happens, we write an error for response.
+	if err != nil {
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	defer stream.Close()
+
+	// r.Write() writes the HTTP request to the stream.
+	err = r.Write(stream)
+	if err != nil {
+		stream.Reset()
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusServiceUnavailable)
+		return
+	}
+
+	// Now we read the response that was sent from the dest
+	// peer
+	buf := bufio.NewReader(stream)
+	resp, err := http.ReadResponse(buf, r)
+	if err != nil {
+		stream.Reset()
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusServiceUnavailable)
+		return
+	}
+
+	// Copy any headers
+	for k, v := range resp.Header {
+		for _, s := range v {
+			w.Header().Add(k, s)
+		}
+	}
+
+	// Write response status and headers
+	w.WriteHeader(resp.StatusCode)
+
+	// Finally copy the body
+	io.Copy(w, resp.Body)
+	resp.Body.Close()
+}
+
+func EgressService(announceTime time.Duration) node.NetworkService {
+	return func(ctx context.Context, c node.Config, n *node.Node, b *blockchain.Ledger) error {
+		b.AnnounceUpdate(ctx, announceTime, protocol.EgressService, n.Host().ID().String(), "ok")
+		return nil
+	}
+}
+
+func Egress(announceTime time.Duration) []node.Option {
+	return []node.Option{
+		node.WithNetworkService(EgressService(announceTime)),
+		node.WithStreamHandler(protocol.EgressProtocol, egressHandler),
+	}
+}
+
+func Proxy(announceTime, deadtime time.Duration, listenAddr string) []node.Option {
+	return []node.Option{
+		node.WithNetworkService(ProxyService(announceTime, listenAddr, deadtime)),
+	}
+}

pkg/services/services.go

@@ -94,87 +94,91 @@ func RegisterService(ll log.StandardLogger, announcetime time.Duration, serviceI
 		node.WithNetworkService(ExposeNetworkService(announcetime, serviceID))}
 }
 
-func ConnectToService(ctx context.Context, ledger *blockchain.Ledger, node *node.Node, ll log.StandardLogger, announcetime time.Duration, serviceID string, srcaddr string) error {
-	// Open local port for listening
-	l, err := net.Listen("tcp", srcaddr)
-	if err != nil {
-		return err
-	}
-	ll.Info("Binding local port on", srcaddr)
-
-	// Announce ourselves so nodes accepts our connection
-	ledger.Announce(
-		ctx,
-		announcetime,
-		func() {
-			// Retrieve current ID for ip in the blockchain
-			_, found := ledger.GetKey(protocol.UsersLedgerKey, node.Host().ID().String())
-			// If mismatch, update the blockchain
-			if !found {
-				updatedMap := map[string]interface{}{}
-				updatedMap[node.Host().ID().String()] = &types.User{
-					PeerID:    node.Host().ID().String(),
-					Timestamp: time.Now().String(),
-				}
-				ledger.Add(protocol.UsersLedgerKey, updatedMap)
-			}
-		},
-	)
-
-	defer l.Close()
-	for {
-		select {
-		case <-ctx.Done():
-			return errors.New("context canceled")
-		default:
-			// Listen for an incoming connection.
-			conn, err := l.Accept()
-			if err != nil {
-				ll.Error("Error accepting: ", err.Error())
-				continue
-			}
+// ConnectNetworkService returns a network service that binds to a service
+func ConnectNetworkService(announcetime time.Duration, serviceID string, srcaddr string) node.NetworkService {
+	return func(ctx context.Context, c node.Config, node *node.Node, ledger *blockchain.Ledger) error {
+		// Open local port for listening
+		l, err := net.Listen("tcp", srcaddr)
+		if err != nil {
+			return err
+		}
+		//	ll.Info("Binding local port on", srcaddr)
 
-			ll.Info("New connection from", l.Addr().String())
-			// Handle connections in a new goroutine, forwarding to the p2p service
-			go func() {
+		// Announce ourselves so nodes accepts our connection
+		ledger.Announce(
+			ctx,
+			announcetime,
+			func() {
 				// Retrieve current ID for ip in the blockchain
-				existingValue, found := ledger.GetKey(protocol.ServicesLedgerKey, serviceID)
-				service := &types.Service{}
-				existingValue.Unmarshal(service)
+				_, found := ledger.GetKey(protocol.UsersLedgerKey, node.Host().ID().String())
 				// If mismatch, update the blockchain
 				if !found {
-					conn.Close()
-					ll.Debugf("service '%s' not found on blockchain", serviceID)
-					return
+					updatedMap := map[string]interface{}{}
+					updatedMap[node.Host().ID().String()] = &types.User{
+						PeerID:    node.Host().ID().String(),
+						Timestamp: time.Now().String(),
+					}
+					ledger.Add(protocol.UsersLedgerKey, updatedMap)
 				}
+			},
+		)
 
-				// Decode the Peer
-				d, err := peer.Decode(service.PeerID)
+		defer l.Close()
+		for {
+			select {
+			case <-ctx.Done():
+				return errors.New("context canceled")
+			default:
+				// Listen for an incoming connection.
+				conn, err := l.Accept()
 				if err != nil {
-					conn.Close()
-					ll.Debugf("could not decode peer '%s'", service.PeerID)
-					return
+					//	ll.Error("Error accepting: ", err.Error())
+					continue
 				}
 
-				// Open a stream
-				stream, err := node.Host().NewStream(ctx, d, protocol.ServiceProtocol.ID())
-				if err != nil {
-					conn.Close()
-					ll.Debugf("could not open stream '%s'", err.Error())
-					return
-				}
-				ll.Debugf("(service %s) Redirecting", serviceID, l.Addr().String())
+				//	ll.Info("New connection from", l.Addr().String())
+				// Handle connections in a new goroutine, forwarding to the p2p service
+				go func() {
+					// Retrieve current ID for ip in the blockchain
+					existingValue, found := ledger.GetKey(protocol.ServicesLedgerKey, serviceID)
+					service := &types.Service{}
+					existingValue.Unmarshal(service)
+					// If mismatch, update the blockchain
+					if !found {
+						conn.Close()
+						//	ll.Debugf("service '%s' not found on blockchain", serviceID)
+						return
+					}
+
+					// Decode the Peer
+					d, err := peer.Decode(service.PeerID)
+					if err != nil {
+						conn.Close()
+						//	ll.Debugf("could not decode peer '%s'", service.PeerID)
+						return
+					}
+
+					// Open a stream
+					stream, err := node.Host().NewStream(ctx, d, protocol.ServiceProtocol.ID())
+					if err != nil {
+						conn.Close()
+						//	ll.Debugf("could not open stream '%s'", err.Error())
+						return
+					}
+					//	ll.Debugf("(service %s) Redirecting", serviceID, l.Addr().String())
 
-				closer := make(chan struct{}, 2)
-				go copyStream(closer, stream, conn)
-				go copyStream(closer, conn, stream)
-				<-closer
+					closer := make(chan struct{}, 2)
+					go copyStream(closer, stream, conn)
+					go copyStream(closer, conn, stream)
+					<-closer
 
-				stream.Close()
-				conn.Close()
-				ll.Infof("(service %s) Done handling %s", serviceID, l.Addr().String())
-			}()
+					stream.Close()
+					conn.Close()
+					//	ll.Infof("(service %s) Done handling %s", serviceID, l.Addr().String())
+				}()
+			}
 		}
+
 	}
 }
 

pkg/services/services_test.go

@@ -58,16 +58,18 @@ var _ = Describe("Expose services", func() {
 
 	logg := logger.New(log.LevelFatal)
 	l := node.Logger(logg)
+	serviceUUID := "test"
 
-	e2, _ := node.New(node.FromBase64(true, true, token), node.WithStore(&blockchain.MemoryStore{}), l)
+	e2, _ := node.New(
+
+		node.WithNetworkService(ConnectNetworkService(1*time.Second, serviceUUID, "127.0.0.1:9999")),
+		node.FromBase64(true, true, token), node.WithStore(&blockchain.MemoryStore{}), l)
 
 	Context("Service sharing", func() {
 		It("expose services and can connect to them", func() {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
-			serviceUUID := "test"
-
 			opts := RegisterService(logg, 1*time.Second, serviceUUID, "142.250.184.35:80")
 			opts = append(opts, node.FromBase64(true, true, token), node.WithStore(&blockchain.MemoryStore{}), l)
 			e, _ := node.New(opts...)
@@ -76,11 +78,8 @@ var _ = Describe("Expose services", func() {
 			// redirects to google:80
 
 			e.Start(ctx)
-			e2.Start(ctx)
-
-			ll, _ := e2.Ledger()
 
-			go ConnectToService(ctx, ll, e2, logg, 1*time.Second, serviceUUID, "127.0.0.1:9999")
+			go e2.Start(ctx)
 
 			Eventually(func() string {
 				return get("http://127.0.0.1:9999")