mirror of
https://github.com/hyprspace/hyprspace.git
synced 2026-09-12 19:51:07 +05:00
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
This commit is contained in:
Max
GitHub
parent
8f5778dffe
commit
e994260bdb
397
cli/up.go
397
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])
|
||||
}
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
432
node/node.go
Normal file
432
node/node.go
Normal file
@ -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
|
||||
}
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@ -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():
|
||||
|
||||
@ -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():
|
||||
|
||||
@ -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()
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user