From e994260bdb9da287933b3b41061f7e53d99bfb20 Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 30 Nov 2025 14:28:51 +0100 Subject: [PATCH] Node refactor (#88) * node: extract most of the logic from cli/up.go * node: close interface after bringing it down * synchronize shutdowns with a WaitGroup * p2p: use Connect instead of DialPeer * rpc: defer Close --- cli/up.go | 397 ++------------------------------------------ dns/server.go | 14 +- node/node.go | 432 ++++++++++++++++++++++++++++++++++++++++++++++++ p2p/discover.go | 17 +- p2p/metrics.go | 5 +- p2p/pex.go | 5 +- rpc/server.go | 7 +- 7 files changed, 482 insertions(+), 395 deletions(-) create mode 100644 node/node.go diff --git a/cli/up.go b/cli/up.go index e93af12..474c845 100644 --- a/cli/up.go +++ b/cli/up.go @@ -2,53 +2,14 @@ package cli import ( "context" - "encoding/binary" - "errors" "fmt" - "io/fs" - "net" - "net/http" + "log" "os" "os/signal" - "path/filepath" - "sync" "syscall" - "time" "github.com/DataDrake/cli-ng/v2/cmd" - "github.com/hyprspace/hyprspace/config" - hsdns "github.com/hyprspace/hyprspace/dns" - "github.com/hyprspace/hyprspace/p2p" - hsrpc "github.com/hyprspace/hyprspace/rpc" - "github.com/hyprspace/hyprspace/svc" - "github.com/hyprspace/hyprspace/tun" - dht "github.com/libp2p/go-libp2p-kad-dht" - "github.com/libp2p/go-libp2p/core/event" - "github.com/libp2p/go-libp2p/core/host" - "github.com/libp2p/go-libp2p/core/network" - "github.com/libp2p/go-libp2p/core/peer" - "github.com/multiformats/go-multiaddr" - "github.com/prometheus/client_golang/prometheus/promhttp" - "github.com/yl2chen/cidranger" -) - -type MuxStream struct { - Stream *network.Stream - Lock *sync.Mutex -} - -var ( - cfg *config.Config - node host.Host - // iface is the tun device used to pass packets between - // Hyprspace and the user's machine. - tunDev *tun.TUN - // activeStreams is a map of active streams to a peer - activeStreams map[peer.ID]MuxStream - // context - ctx context.Context - // context cancel function - ctxCancel func() + hsnode "github.com/hyprspace/hyprspace/node" ) // Up creates and brings up a Hyprspace Interface. @@ -72,359 +33,27 @@ func UpRun(r *cmd.Root, c *cmd.Sub) { configPath = "/etc/hyprspace/" + ifName + ".json" } - // Read in configuration from file. - cfg2, err := config.Read(configPath) - checkErr(err) - cfg2.Interface = ifName - cfg = cfg2 + node := hsnode.New(context.Background(), configPath, ifName) + checkErr(node.Run()) - fmt.Println("[+] Creating TUN Device") - - // Create new TUN device - tunDev, err = tun.New( - cfg.Interface, - tun.Address(cfg.BuiltinAddr4.String()+"/32"), - tun.Address(cfg.BuiltinAddr6.String()+"/128"), - tun.MTU(1420), - ) - if err != nil { - checkErr(err) - } - allRoutes4, err := cfg.PeerLookup.ByRoute.CoveredNetworks(*cidranger.AllIPv4) - if err != nil { - checkErr(err) - } - allRoutes6, err := cfg.PeerLookup.ByRoute.CoveredNetworks(*cidranger.AllIPv6) - if err != nil { - checkErr(err) - } - var routeOpts []tun.Option - - for _, r := range allRoutes4 { - routeOpts = append(routeOpts, tun.Route(r.Network())) - } - for _, r := range allRoutes6 { - routeOpts = append(routeOpts, tun.Route(r.Network())) - } - - // Setup System Context - ctx, ctxCancel = context.WithCancel(context.Background()) - - fmt.Println("[+] Creating LibP2P Node") - - // Create P2P Node - host, dht, err := p2p.CreateNode( - ctx, - cfg.PrivateKey, - cfg.ListenAddresses, - streamHandler, - p2p.NewClosedCircuitRelayFilter(cfg.Peers), - p2p.NewRecursionGater(cfg), - cfg.Peers, - ) - checkErr(err) - host.SetStreamHandler(p2p.PeXProtocol, p2p.NewPeXStreamHandler(host, cfg)) - node = host - - for _, p := range cfg.Peers { - host.ConnManager().Protect(p.ID, "/hyprspace/peer") - } - - fmt.Println("[+] Setting Up Node Discovery via DHT") - - // Setup P2P Discovery - go p2p.Discover(ctx, host, dht, cfg.Peers) - - // Configure path for lock - lockPath := filepath.Join(filepath.Dir(cfg.Path), cfg.Interface+".lock") - - // PeX - go p2p.PeXService(ctx, host, cfg) - - // Route metrics and latency - go p2p.RouteMetricsService(ctx, host, cfg) - - // Register the application to listen for signals - go signalHandler(ctx, host, lockPath, dht) - - // Log about various events - go eventLogger(ctx, host) - - // RPC server - go hsrpc.RpcServer(ctx, multiaddr.StringCast(fmt.Sprintf("/unix/run/hyprspace-rpc.%s.sock", cfg.Interface)), host, *cfg, *tunDev) - - // Magic DNS server - go hsdns.MagicDnsServer(ctx, *cfg, node) - - // metrics endpoint - metricsPort, ok := os.LookupEnv("HYPRSPACE_METRICS_PORT") - if ok { - metricsTuple := fmt.Sprintf("127.0.0.1:%s", metricsPort) - http.Handle("/metrics", promhttp.Handler()) - go func() { - http.ListenAndServe(metricsTuple, nil) - }() - fmt.Printf("[+] Listening for metrics scrape requests on http://%s/metrics\n", metricsTuple) - } - - serviceNet := svc.NewServiceNetwork(host, cfg, tunDev) - - for name, addr := range cfg.Services { - proxy, err := svc.ProxyTo(addr) - checkErr(err) - serviceNet.Register( - name, - proxy, - ) - } - - var svcNetIds [][4]byte - for _, p := range cfg.Peers { - svcNetIds = append(svcNetIds, config.MkNetID(p.ID)) - } - svcNetIds = append(svcNetIds, config.MkNetID(host.ID())) - for _, netId := range svcNetIds { - addr := make([]byte, 16) - copy(addr, serviceNet.NetworkRange.IP) - copy(addr[10:], netId[:]) - mask1, mask0 := serviceNet.NetworkRange.Mask.Size() - routeOpts = append(routeOpts, tun.Route(net.IPNet{ - IP: addr, - Mask: net.CIDRMask(mask1+32, mask0), - })) - } - - // Write lock to filesystem to indicate an existing running daemon. - err = os.WriteFile(lockPath, []byte(fmt.Sprint(os.Getpid())), os.ModePerm) - checkErr(err) - - // Bring Up TUN Device - err = tunDev.Up() - if err != nil { - checkErr(errors.New("unable to bring up tun device")) - } - checkErr(tunDev.Apply(routeOpts...)) - - fmt.Println("[+] Network setup complete") - - // + ----------------------------------------+ - // | Listen For New Packets on TUN Interface | - // + ----------------------------------------+ - - // Initialize active streams map and packet byte array. - activeStreams = make(map[peer.ID]MuxStream) - for { - var packet = make([]byte, 1420) - // Read in a packet from the tun device. - plen, err := tunDev.Iface.Read(packet) - if errors.Is(err, fs.ErrClosed) { - fmt.Println("[-] Interface closed") - <-ctx.Done() - time.Sleep(1 * time.Second) - return - } else if err != nil { - fmt.Println(err) - continue - } - - var dstIP net.IP - proto := packet[0] & 0xf0 - - if proto == 0x40 { - dstIP = net.IP(packet[16:20]) - if cfg.BuiltinAddr4.Equal(dstIP) { - continue - } - } else if proto == 0x60 { - dstIP = net.IP(packet[24:40]) - if cfg.BuiltinAddr6.Equal(dstIP) { - continue - } else if serviceNet.NetworkRange.Contains(dstIP) { - // Are you TCP because your protocol is 6, or is your protocol 6 because you are TCP? - if packet[6] == 0x06 { - port := uint16(packet[42])*256 + uint16(packet[43]) - if serviceNet.EnsureListener([16]byte(packet[24:40]), port) { - count, err := (*serviceNet.Tun).Write([][]byte{packet}, 0) - if count == 0 { - fmt.Printf("[!] To service network: %s\n", err) - } - } - } - continue - } - } else { - continue - } - var dst peer.ID - - // Check route table for destination address. - route, found := cfg.FindRouteForIP(dstIP) - - if found { - dst = route.Target.ID - go sendPacket(dst, packet, plen) - } - } -} - -func sendPacket(dst peer.ID, packet []byte, plen int) { - // Check if we already have an open connection to the destination peer. - ms, ok := activeStreams[dst] - if ok { - if func() bool { - ms.Lock.Lock() - defer ms.Lock.Unlock() - // Write out the packet's length to the libp2p stream to ensure - // we know the full size of the packet at the other end. - err := binary.Write(*ms.Stream, binary.LittleEndian, uint16(plen)) - if err == nil { - // Write the packet out to the libp2p stream. - // If everyting succeeds continue on to the next packet. - _, err = (*ms.Stream).Write(packet[:plen]) - if err == nil { - err := (*ms.Stream).SetWriteDeadline(time.Now().Add(25 * time.Second)) - if err == nil { - return true - } - } - } - // If we encounter an error when writing to a stream we should - // close that stream and delete it from the active stream map. - (*ms.Stream).Close() - delete(activeStreams, dst) - return false - }() { - return - } - } - - stream, err := node.NewStream(ctx, dst, p2p.Protocol) - if err != nil { - fmt.Println("[!] Failed to open stream to " + dst.String() + ": " + err.Error()) - go p2p.Rediscover() - return - } - err = stream.SetWriteDeadline(time.Now().Add(25 * time.Second)) - if err != nil { - fmt.Println("[!] Failed to set write deadline: " + err.Error()) - stream.Close() - return - } - // Write packet length - err = binary.Write(stream, binary.LittleEndian, uint16(plen)) - if err != nil { - stream.Close() - return - } - // Write the packet - _, err = stream.Write(packet[:plen]) - if err != nil { - stream.Close() - return - } - - // If all succeeds when writing the packet to the stream - // we should reuse this stream by adding it active streams map. - activeStreams[dst] = MuxStream{ - Stream: &stream, - Lock: &sync.Mutex{}, - } -} - -func signalHandler(ctx context.Context, host host.Host, lockPath string, dht *dht.IpfsDHT) { exitCh := make(chan os.Signal, 1) rebootstrapCh := make(chan os.Signal, 1) - signal.Notify(exitCh, syscall.SIGINT, syscall.SIGTERM) + signal.Notify(exitCh, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGQUIT) signal.Notify(rebootstrapCh, syscall.SIGUSR1) for { select { - case <-ctx.Done(): - return case <-rebootstrapCh: fmt.Println("[-] Rebootstrapping on SIGUSR1") - host.ConnManager().TrimOpenConns(context.Background()) - <-dht.ForceRefresh() - p2p.Rediscover() + node.Rebootstrap() case <-exitCh: - // Shut the node down - err := host.Close() - checkErr(err) - - // Remove daemon lock from file system. - err = os.Remove(lockPath) - checkErr(err) - - fmt.Println("Received signal, shutting down...") - - tunDev.Iface.Close() - err = tunDev.Down() - checkErr(err) - ctxCancel() + fmt.Println("[-] Shutting down...") + go func() { + <-exitCh + log.Fatal("Terminating immediately.") + }() + checkErr(node.Stop()) + os.Exit(0) } } } - -func eventLogger(ctx context.Context, host host.Host) { - subCon, err := host.EventBus().Subscribe(new(event.EvtPeerConnectednessChanged)) - checkErr(err) - for { - select { - case <-ctx.Done(): - return - case ev := <-subCon.Out(): - evt := ev.(event.EvtPeerConnectednessChanged) - for _, vpnPeer := range cfg.Peers { - if vpnPeer.ID == evt.Peer { - if evt.Connectedness == network.Connected { - for _, c := range host.Network().ConnsToPeer(evt.Peer) { - fmt.Printf("[+] Connected to %s/p2p/%s\n", c.RemoteMultiaddr().String(), evt.Peer.String()) - } - } else if evt.Connectedness == network.NotConnected { - fmt.Printf("[!] Disconnected from %s\n", evt.Peer.String()) - } - break - } - } - } - } -} - -func streamHandler(stream network.Stream) { - // If the remote node ID isn't in the list of known nodes don't respond. - if _, ok := config.FindPeer(cfg.Peers, stream.Conn().RemotePeer()); !ok { - stream.Reset() - return - } - var packet = make([]byte, 1420) - var packetSize = make([]byte, 2) - for { - // Read the incoming packet's size as a binary value. - _, err := stream.Read(packetSize) - if err != nil { - stream.Close() - return - } - - // Decode the incoming packet's size from binary. - size := binary.LittleEndian.Uint16(packetSize) - - // Read in the packet until completion. - var plen uint16 = 0 - for plen < size { - tmp, err := stream.Read(packet[plen:size]) - plen += uint16(tmp) - if err != nil { - stream.Close() - return - } - } - err = stream.SetWriteDeadline(time.Now().Add(25 * time.Second)) - if err != nil { - fmt.Println("[!] Failed to set write deadline: " + err.Error()) - stream.Close() - return - } - _, _ = tunDev.Iface.Write(packet[:size]) - } -} diff --git a/dns/server.go b/dns/server.go index f65442b..66b3ba0 100644 --- a/dns/server.go +++ b/dns/server.go @@ -6,7 +6,9 @@ import ( "net" "os" "strings" + "sync" "syscall" + "time" "github.com/hyprspace/hyprspace/config" "github.com/iguanesolutions/go-systemd/v5/resolved" @@ -108,7 +110,9 @@ func writeResponse(msg *dns.Msg, q dns.Question, p peer.ID, addr net.IP) { }) } -func MagicDnsServer(ctx context.Context, config config.Config, node host.Host) { +func MagicDnsServer(ctx context.Context, wg *sync.WaitGroup, config config.Config, node host.Host) { + wg.Add(1) + defer wg.Done() dns.HandleFunc(domainSuffix(config), func(w dns.ResponseWriter, r *dns.Msg) { m := new(dns.Msg) m.SetReply(r) @@ -175,6 +179,7 @@ func MagicDnsServer(ctx context.Context, config config.Config, node host.Host) { w.WriteMsg(m) }) + var servers = make([]*dns.Server, 0) dnsServerAddrBytes := []byte{127, 80, 01, 53} var dnsServerPort uint16 = 5380 for i, b := range []byte(config.Interface) { @@ -194,6 +199,7 @@ func MagicDnsServer(ctx context.Context, config config.Config, node host.Host) { fmt.Printf("[!] DNS server error: %s, %s\n", server.Net, err.Error()) } }(sv) + servers = append(servers, sv) } conn, err := resolved.NewConn() @@ -230,4 +236,10 @@ func MagicDnsServer(ctx context.Context, config config.Config, node host.Host) { return } } + + <-ctx.Done() + for _, s := range servers { + shutdownCtx, _ := context.WithDeadline(ctx, time.Now().Add(5*time.Second)) + s.ShutdownContext(shutdownCtx) + } } diff --git a/node/node.go b/node/node.go new file mode 100644 index 0000000..bf714fb --- /dev/null +++ b/node/node.go @@ -0,0 +1,432 @@ +package node + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "io/fs" + "net" + "net/http" + "os" + "path/filepath" + "sync" + "time" + + "github.com/hyprspace/hyprspace/config" + hsdns "github.com/hyprspace/hyprspace/dns" + "github.com/hyprspace/hyprspace/p2p" + hsrpc "github.com/hyprspace/hyprspace/rpc" + "github.com/hyprspace/hyprspace/svc" + "github.com/hyprspace/hyprspace/tun" + "github.com/libp2p/go-cidranger" + dht "github.com/libp2p/go-libp2p-kad-dht" + "github.com/libp2p/go-libp2p/core/event" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/multiformats/go-multiaddr" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +type SharedStream struct { + Stream *network.Stream + Lock *sync.Mutex +} + +type Node struct { + cfg *config.Config + p2p host.Host + dht *dht.IpfsDHT + tunDev *tun.TUN + activeStreams map[peer.ID]SharedStream + ctx context.Context + cancel func() + lockPath string + configPath string + interfaceName string + wg *sync.WaitGroup +} + +func New(ctx context.Context, configPath string, ifName string) Node { + innerCtx, ctxCancel := context.WithCancel(ctx) + + return Node{ + cfg: &config.Config{}, + p2p: nil, + tunDev: &tun.TUN{}, + activeStreams: map[peer.ID]SharedStream{}, + ctx: innerCtx, + cancel: ctxCancel, + configPath: configPath, + interfaceName: ifName, + } +} + +func (node *Node) Run() error { + // Read in configuration from file. + cfg2, err := config.Read(node.configPath) + if err != nil { + return err + } + + cfg2.Interface = node.interfaceName + node.cfg = cfg2 + + fmt.Println("[+] Creating TUN Device") + + // Create new TUN device + node.tunDev, err = tun.New( + node.cfg.Interface, + tun.Address(node.cfg.BuiltinAddr4.String()+"/32"), + tun.Address(node.cfg.BuiltinAddr6.String()+"/128"), + tun.MTU(1420), + ) + if err != nil { + return err + } + allRoutes4, err := node.cfg.PeerLookup.ByRoute.CoveredNetworks(*cidranger.AllIPv4) + if err != nil { + return err + } + allRoutes6, err := node.cfg.PeerLookup.ByRoute.CoveredNetworks(*cidranger.AllIPv6) + if err != nil { + return err + } + var routeOpts []tun.Option + + for _, r := range allRoutes4 { + routeOpts = append(routeOpts, tun.Route(r.Network())) + } + for _, r := range allRoutes6 { + routeOpts = append(routeOpts, tun.Route(r.Network())) + } + + fmt.Println("[+] Creating LibP2P Node") + + // Create P2P Node + node.p2p, node.dht, err = p2p.CreateNode( + node.ctx, + node.cfg.PrivateKey, + node.cfg.ListenAddresses, + node.streamHandler, + p2p.NewClosedCircuitRelayFilter(node.cfg.Peers), + p2p.NewRecursionGater(node.cfg), + node.cfg.Peers, + ) + if err != nil { + return err + } + node.p2p.SetStreamHandler(p2p.PeXProtocol, p2p.NewPeXStreamHandler(node.p2p, node.cfg)) + + for _, p := range node.cfg.Peers { + node.p2p.ConnManager().Protect(p.ID, "/hyprspace/peer") + } + + node.wg = &sync.WaitGroup{} + + fmt.Println("[+] Setting Up Node Discovery via DHT") + + // Setup P2P Discovery + go p2p.Discover(node.ctx, node.wg, node.p2p, node.dht, node.cfg.Peers) + + // Configure path for lock + node.lockPath = filepath.Join(filepath.Dir(node.cfg.Path), node.cfg.Interface+".lock") + + // PeX + go p2p.PeXService(node.ctx, node.wg, node.p2p, node.cfg) + + // Route metrics and latency + go p2p.RouteMetricsService(node.ctx, node.wg, node.p2p, node.cfg) + + // Log about various events + err = node.eventLogger(node.ctx, node.p2p) + if err != nil { + return err + } + + // RPC server + go hsrpc.RpcServer(node.ctx, node.wg, multiaddr.StringCast(fmt.Sprintf("/unix/run/hyprspace-rpc.%s.sock", node.cfg.Interface)), node.p2p, *node.cfg, *node.tunDev) + + // Magic DNS server + go hsdns.MagicDnsServer(node.ctx, node.wg, *node.cfg, node.p2p) + + // metrics endpoint + metricsPort, ok := os.LookupEnv("HYPRSPACE_METRICS_PORT") + if ok { + metricsTuple := fmt.Sprintf("127.0.0.1:%s", metricsPort) + http.Handle("/metrics", promhttp.Handler()) + go func() { + http.ListenAndServe(metricsTuple, nil) + }() + fmt.Printf("[+] Listening for metrics scrape requests on http://%s/metrics\n", metricsTuple) + } + + serviceNet := svc.NewServiceNetwork(node.p2p, node.cfg, node.tunDev) + + for name, addr := range node.cfg.Services { + proxy, err := svc.ProxyTo(addr) + if err != nil { + return err + } + serviceNet.Register( + name, + proxy, + ) + } + + var svcNetIds [][4]byte + for _, p := range node.cfg.Peers { + svcNetIds = append(svcNetIds, config.MkNetID(p.ID)) + } + svcNetIds = append(svcNetIds, config.MkNetID(node.p2p.ID())) + for _, netId := range svcNetIds { + addr := make([]byte, 16) + copy(addr, serviceNet.NetworkRange.IP) + copy(addr[10:], netId[:]) + mask1, mask0 := serviceNet.NetworkRange.Mask.Size() + routeOpts = append(routeOpts, tun.Route(net.IPNet{ + IP: addr, + Mask: net.CIDRMask(mask1+32, mask0), + })) + } + + // Write lock to filesystem to indicate an existing running daemon. + err = os.WriteFile(node.lockPath, []byte(fmt.Sprint(os.Getpid())), os.ModePerm) + if err != nil { + return err + } + + // Bring Up TUN Device + err = node.tunDev.Up() + if err != nil { + return errors.New("unable to bring up tun device: " + err.Error()) + } + err = node.tunDev.Apply(routeOpts...) + if err != nil { + return err + } + + fmt.Println("[+] Network setup complete") + + // Initialize active streams map and packet byte array. + node.activeStreams = make(map[peer.ID]SharedStream) + go func() { + for { + var packet = make([]byte, 1420) + // Read in a packet from the tun device. + plen, err := node.tunDev.Iface.Read(packet) + if errors.Is(err, fs.ErrClosed) { + fmt.Println("[-] Interface closed") + <-node.ctx.Done() + time.Sleep(1 * time.Second) + return + } else if err != nil { + fmt.Println(err) + continue + } + + var dstIP net.IP + proto := packet[0] & 0xf0 + + if proto == 0x40 { + dstIP = net.IP(packet[16:20]) + if node.cfg.BuiltinAddr4.Equal(dstIP) { + continue + } + } else if proto == 0x60 { + dstIP = net.IP(packet[24:40]) + if node.cfg.BuiltinAddr6.Equal(dstIP) { + continue + } else if serviceNet.NetworkRange.Contains(dstIP) { + // Are you TCP because your protocol is 6, or is your protocol 6 because you are TCP? + if packet[6] == 0x06 { + port := uint16(packet[42])*256 + uint16(packet[43]) + if serviceNet.EnsureListener([16]byte(packet[24:40]), port) { + count, err := (*serviceNet.Tun).Write([][]byte{packet}, 0) + if count == 0 { + fmt.Printf("[!] To service network: %s\n", err) + } + } + } + continue + } + } else { + continue + } + var dst peer.ID + + // Check route table for destination address. + route, found := node.cfg.FindRouteForIP(dstIP) + + if found { + dst = route.Target.ID + go node.sendPacket(dst, packet, plen) + } + } + }() + return nil +} + +func (node *Node) streamHandler(stream network.Stream) { + // If the remote node ID isn't in the list of known nodes don't respond. + if _, ok := config.FindPeer(node.cfg.Peers, stream.Conn().RemotePeer()); !ok { + stream.Reset() + return + } + var packet = make([]byte, 1420) + var packetSize = make([]byte, 2) + for { + // Read the incoming packet's size as a binary value. + _, err := stream.Read(packetSize) + if err != nil { + stream.Close() + return + } + + // Decode the incoming packet's size from binary. + size := binary.LittleEndian.Uint16(packetSize) + + // Read in the packet until completion. + var plen uint16 = 0 + for plen < size { + tmp, err := stream.Read(packet[plen:size]) + plen += uint16(tmp) + if err != nil { + stream.Close() + return + } + } + err = stream.SetWriteDeadline(time.Now().Add(25 * time.Second)) + if err != nil { + fmt.Println("[!] Failed to set write deadline: " + err.Error()) + stream.Close() + return + } + _, _ = node.tunDev.Iface.Write(packet[:size]) + } +} + +func (node *Node) sendPacket(dst peer.ID, packet []byte, plen int) { + // Check if we already have an open connection to the destination peer. + ms, ok := node.activeStreams[dst] + if ok { + if func() bool { + ms.Lock.Lock() + defer ms.Lock.Unlock() + // Write out the packet's length to the libp2p stream to ensure + // we know the full size of the packet at the other end. + err := binary.Write(*ms.Stream, binary.LittleEndian, uint16(plen)) + if err == nil { + // Write the packet out to the libp2p stream. + // If everyting succeeds continue on to the next packet. + _, err = (*ms.Stream).Write(packet[:plen]) + if err == nil { + err := (*ms.Stream).SetWriteDeadline(time.Now().Add(25 * time.Second)) + if err == nil { + return true + } + } + } + // If we encounter an error when writing to a stream we should + // close that stream and delete it from the active stream map. + (*ms.Stream).Close() + delete(node.activeStreams, dst) + return false + }() { + return + } + } + + stream, err := node.p2p.NewStream(node.ctx, dst, p2p.Protocol) + if err != nil { + fmt.Println("[!] Failed to open stream to " + dst.String() + ": " + err.Error()) + go p2p.Rediscover() + return + } + err = stream.SetWriteDeadline(time.Now().Add(25 * time.Second)) + if err != nil { + fmt.Println("[!] Failed to set write deadline: " + err.Error()) + stream.Close() + return + } + // Write packet length + err = binary.Write(stream, binary.LittleEndian, uint16(plen)) + if err != nil { + stream.Close() + return + } + // Write the packet + _, err = stream.Write(packet[:plen]) + if err != nil { + stream.Close() + return + } + + // If all succeeds when writing the packet to the stream + // we should reuse this stream by adding it active streams map. + node.activeStreams[dst] = SharedStream{ + Stream: &stream, + Lock: &sync.Mutex{}, + } +} + +func (node *Node) eventLogger(ctx context.Context, host host.Host) error { + subCon, err := host.EventBus().Subscribe(new(event.EvtPeerConnectednessChanged)) + if err != nil { + return err + } + go func() { + for { + select { + case <-ctx.Done(): + node.wg.Done() + return + case ev := <-subCon.Out(): + evt := ev.(event.EvtPeerConnectednessChanged) + for _, vpnPeer := range node.cfg.Peers { + if vpnPeer.ID == evt.Peer { + if evt.Connectedness == network.Connected { + for _, c := range host.Network().ConnsToPeer(evt.Peer) { + fmt.Printf("[+] Connected to %s/p2p/%s\n", c.RemoteMultiaddr().String(), evt.Peer.String()) + } + } else if evt.Connectedness == network.NotConnected { + fmt.Printf("[!] Disconnected from %s\n", evt.Peer.String()) + } + break + } + } + } + } + }() + return nil +} + +func (node *Node) Rebootstrap() { + node.p2p.ConnManager().TrimOpenConns(context.Background()) + <-node.dht.ForceRefresh() + p2p.Rediscover() +} + +func (node *Node) Stop() error { + err := node.p2p.Close() + if err != nil { + return err + } + + err = os.Remove(node.lockPath) + if err != nil { + return err + } + + fmt.Println("Received signal, shutting down...") + + err = node.tunDev.Down() + if err != nil { + return err + } + node.tunDev.Iface.Close() + node.cancel() + node.wg.Wait() + return nil +} diff --git a/p2p/discover.go b/p2p/discover.go index 2d54888..5ab9cfb 100644 --- a/p2p/discover.go +++ b/p2p/discover.go @@ -3,22 +3,27 @@ package p2p import ( "context" "fmt" + "sync" "time" "github.com/hyprspace/hyprspace/config" dht "github.com/libp2p/go-libp2p-kad-dht" "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/multiformats/go-multiaddr" ) var discoverNow = make(chan bool) // Discover starts up a DHT based discovery system finding and adding nodes with the same rendezvous string. -func Discover(ctx context.Context, h host.Host, dht *dht.IpfsDHT, peers []config.Peer) { +func Discover(ctx context.Context, wg *sync.WaitGroup, h host.Host, dht *dht.IpfsDHT, peers []config.Peer) { dur := time.Second * 1 ticker := time.NewTicker(dur) defer ticker.Stop() + wg.Add(1) + defer wg.Done() for { select { case <-ctx.Done(): @@ -31,7 +36,10 @@ func Discover(ctx context.Context, h host.Host, dht *dht.IpfsDHT, peers []config connectedToAny := false for _, p := range peers { if h.Network().Connectedness(p.ID) != network.Connected { - _, err := h.Network().DialPeer(ctx, p.ID) + err := h.Connect(ctx, peer.AddrInfo{ + ID: p.ID, + Addrs: []multiaddr.Multiaddr{}, + }) if err != nil { continue } @@ -47,10 +55,7 @@ func Discover(ctx context.Context, h host.Host, dht *dht.IpfsDHT, peers []config dur = time.Second * 10 ticker.Reset(dur) } else { - dur = dur * 2 - if dur >= time.Second*60 { - dur = time.Second * 60 - } + dur = min(dur*2, time.Minute) ticker.Reset(dur) } } diff --git a/p2p/metrics.go b/p2p/metrics.go index bd82da5..b34e155 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "sync" "time" "github.com/hyprspace/hyprspace/config" @@ -13,12 +14,14 @@ import ( "github.com/libp2p/go-libp2p/p2p/protocol/ping" ) -func RouteMetricsService(ctx context.Context, host host.Host, cfg *config.Config) { +func RouteMetricsService(ctx context.Context, wg *sync.WaitGroup, host host.Host, cfg *config.Config) { subCon, err := host.EventBus().Subscribe(new(event.EvtPeerConnectednessChanged)) if err != nil { log.Fatal(err) } fmt.Println("[-] Route metrics service ready") + wg.Add(1) + defer wg.Done() for { select { case <-ctx.Done(): diff --git a/p2p/pex.go b/p2p/pex.go index bb816f2..79f5409 100644 --- a/p2p/pex.go +++ b/p2p/pex.go @@ -7,6 +7,7 @@ import ( "io" "log" "strings" + "sync" "time" "github.com/hyprspace/hyprspace/config" @@ -111,12 +112,14 @@ func RequestPeX(ctx context.Context, host host.Host, peers []peer.ID) (addrInfos return addrInfos, nil } -func PeXService(ctx context.Context, host host.Host, cfg *config.Config) { +func PeXService(ctx context.Context, wg *sync.WaitGroup, host host.Host, cfg *config.Config) { subCon, err := host.EventBus().Subscribe(new(event.EvtPeerConnectednessChanged)) if err != nil { log.Fatal(err) } fmt.Println("[-] PeX service ready") + wg.Add(1) + defer wg.Done() for { select { case <-ctx.Done(): diff --git a/rpc/server.go b/rpc/server.go index eb8b523..ad228f2 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -7,6 +7,7 @@ import ( "log" "net" "net/rpc" + "sync" "syscall" "github.com/hyprspace/hyprspace/config" @@ -163,7 +164,9 @@ func (hsr *HyprspaceRPC) Peers(args *Args, reply *PeersReply) error { return nil } -func RpcServer(ctx context.Context, ma multiaddr.Multiaddr, host host.Host, config config.Config, tunDev tun.TUN) { +func RpcServer(ctx context.Context, wg *sync.WaitGroup, ma multiaddr.Multiaddr, host host.Host, config config.Config, tunDev tun.TUN) { + wg.Add(1) + defer wg.Done() hsr := HyprspaceRPC{host, config, tunDev} rpc.Register(&hsr) @@ -184,8 +187,8 @@ func RpcServer(ctx context.Context, ma multiaddr.Multiaddr, host host.Host, conf } fmt.Println("[-] RPC server ready") + defer l.Close() go rpc.Accept(l) <-ctx.Done() fmt.Println("[-] Closing RPC server") - l.Close() }