WIP: Cleanup

peer/bufferset.go

@@ -0,0 +1 @@
+package peer

peer/cipher.go

@@ -0,0 +1 @@
+package peer

peer/globals.go

@@ -31,6 +31,7 @@ var multicastAddr = net.UDPAddrFromAddrPort(netip.AddrPortFrom(
 	netip.AddrFrom4([4]byte{224, 0, 0, 157}),
 	4560))
 
+<<<<<<< HEAD
 // ----------------------------------------------------------------------------
 
 type Globals struct {
@@ -105,5 +106,73 @@ func NewGlobals(
 		g.RemotePeers[i].Store(newRemote(g, byte(i)))
 	}
 
+	return g
+=======
+type marshaller interface {
+	Marshal([]byte) []byte
+>>>>>>> 69f2536 (WIP)
+}
+
+// ----------------------------------------------------------------------------
+
+type Globals struct {
+	LocalConfig // Embed, immutable.
+
+	// Local public address (if available). Immutable.
+	LocalAddr netip.AddrPort
+
+	// True if local public address is valid. Immutable.
+	LocalAddrValid bool
+
+	// All remote peers by VPN IP.
+	RemotePeers [256]*atomic.Pointer[Remote]
+
+	// Discovered public addresses.
+	PubAddrs *pubAddrStore
+
+	// Attempts to ensure that we have a relay available.
+	RelayHandler *relayHandler
+
+	// Send UDP - Global function to write UDP packets.
+	SendUDP func(b []byte, addr netip.AddrPort) (n int, err error)
+
+	// Global TUN interface.
+	IFace io.ReadWriteCloser
+}
+
+func NewGlobals(
+	localConfig LocalConfig,
+	localAddr netip.AddrPort,
+	conn *net.UDPConn,
+	iface io.ReadWriteCloser,
+) (g Globals) {
+	g.LocalConfig = localConfig
+
+	g.LocalAddr = localAddr
+	g.LocalAddrValid = localAddr.IsValid()
+
+	g.PubAddrs = newPubAddrStore(localAddr)
+
+	g.RelayHandler = newRelayHandler()
+
+	// Use a lock here avoids starvation, at least on my Linux machine.
+	sendLock := sync.Mutex{}
+	g.SendUDP = func(b []byte, addr netip.AddrPort) (int, error) {
+		sendLock.Lock()
+		n, err := conn.WriteToUDPAddrPort(b, addr)
+		sendLock.Unlock()
+		return n, err
+	}
+
+	g.IFace = iface
+
+	for i := range g.RemotePeers {
+		g.RemotePeers[i] = &atomic.Pointer[Remote]{}
+	}
+
+	for i := range g.RemotePeers {
+		g.RemotePeers[i].Store(newRemote(g, byte(i)))
+	}
+
 	return g
 }

peer/mcreader.go

@@ -41,6 +41,7 @@ func runMCReaderInner(g Globals) {
 			logf("Failed to open discovery packet?")
 			continue
 		}
+		log.Printf("Got local discovery from %v: %v", remoteAddr, h)
 
 		g.RemotePeers[h.SourceIP].Load().HandleLocalDiscoveryPacket(h, remoteAddr, buf)
 	}

peer/peer.go

@@ -11,7 +11,10 @@ import (
 	"net/http"
 	"net/netip"
 	"net/url"
+<<<<<<< HEAD
 	"os"
+=======
+>>>>>>> 69f2536 (WIP)
 	"vppn/m"
 )
 
@@ -50,6 +53,7 @@ func newPeerMain(args mainArgs) *peerMain {
 		log.Fatalf("Failed to load network state: %v", err)
 	}
 
+<<<<<<< HEAD
 	startupCount, err := loadStartupCount(args.NetName)
 	if err != nil {
 		if !os.IsNotExist(err) {
@@ -66,6 +70,8 @@ func newPeerMain(args mainArgs) *peerMain {
 		log.Fatalf("Failed to write startup count: %v", err)
 	}
 
+=======
+>>>>>>> 69f2536 (WIP)
 	iface, err := openInterface(config.Network, config.LocalPeerIP, args.NetName)
 	if err != nil {
 		log.Fatalf("Failed to open interface: %v", err)
@@ -93,7 +99,11 @@ func newPeerMain(args mainArgs) *peerMain {
 		localAddr = netip.AddrPortFrom(ip, localPeer.Port)
 	}
 
+<<<<<<< HEAD
 	g := NewGlobals(config, startupCount, localAddr, conn, iface)
+=======
+	g := NewGlobals(config, localAddr, conn, iface)
+>>>>>>> 69f2536 (WIP)
 
 	hubPoller, err := NewHubPoller(g, args.NetName, args.HubAddress, args.APIKey)
 	if err != nil {

peer/peerfsm.dot

@@ -0,0 +1,28 @@
+digraph d {
+        disconnected -> peerUpdating;
+        peerUpdating -> disconnected;
+        peerUpdating -> server;
+        peerUpdating -> clientInit;
+        server -> peerUpdating;
+        clientInit -> peerUpdating;
+        clientInit -> clientInit;
+        clientInit -> client;
+        client -> clientInit;
+        client -> peerUpdating;
+
+        clientInitializing -> clientSyncing;
+        clientSyncing -> clientInitializing;
+        clientSyncing -> clientUpIndirect;
+        clientSyncing -> clientUpDirect;
+        clientUpIndirect -> clientUpDirect;
+        clientUpIndirect -> clientInitializing;
+        clientUpDirect -> clientInitializing;
+
+        serverInitializing -> serverSyncing;
+        serverSyncing -> serverInitializing;
+        serverSyncing -> serverUpIndirect;
+        serverSyncing -> serverUpDirect;
+        serverUpIndirect -> serverUpDirect;
+        serverUpIndirect -> serverInitializing;
+        serverUpDirect -> serverInitializing;
+}

peer/peersuper.go

@@ -0,0 +1 @@
+package peer

peer/relay.go

@@ -0,0 +1 @@
+package peer

peer/remote.go

@@ -6,7 +6,14 @@ import (
 	"net/netip"
 	"strings"
 	"sync/atomic"
+<<<<<<< HEAD
 	"vppn/m"
+=======
+	"time"
+	"vppn/m"
+
+	"git.crumpington.com/lib/go/ratelimiter"
+>>>>>>> 69f2536 (WIP)
 )
 
 // ----------------------------------------------------------------------------
@@ -37,6 +44,10 @@ type Remote struct {
 	Globals
 	RemotePeerIP byte // Immutable.
 
+<<<<<<< HEAD
+=======
+	limiter     *ratelimiter.Limiter
+>>>>>>> 69f2536 (WIP)
 	dupCheck    *dupCheck
 	sendCounter uint64 // init to startupCount << 48. Atomic access only.
 
@@ -49,9 +60,19 @@ func newRemote(g Globals, remotePeerIP byte) *Remote {
 	r := &Remote{
 		Globals:      g,
 		RemotePeerIP: remotePeerIP,
+<<<<<<< HEAD
 		dupCheck:     newDupCheck(0),
 		sendCounter:  (uint64(g.StartupCount) << 48) + 1,
 		messages:     make(chan any, 8),
+=======
+		limiter: ratelimiter.New(ratelimiter.Config{
+			FillPeriod:   20 * time.Millisecond,
+			MaxWaitCount: 1,
+		}),
+		dupCheck:    newDupCheck(0),
+		sendCounter: uint64(time.Now().Unix()<<30) + 1,
+		messages:    make(chan any, 8),
+>>>>>>> 69f2536 (WIP)
 	}
 	r.config.Store(&remoteConfig{})
 	return r
@@ -79,19 +100,36 @@ func (r *Remote) updateConf(conf remoteConfig) {
 // ----------------------------------------------------------------------------
 
 func (r *Remote) sendUDP(b []byte, addr netip.AddrPort) {
+<<<<<<< HEAD
 	if _, err := r.SendUDP(b, addr); err != nil {
 		r.logf("Failed to send UDP packet: %v", err)
+=======
+	if err := r.limiter.Limit(); err != nil {
+		r.logf("Rate limiter")
+		return
+	}
+	if _, err := r.SendUDP(b, addr); err != nil {
+		r.logf("Failed to send URP packet: %v", err)
+>>>>>>> 69f2536 (WIP)
 	}
 }
 
 // ----------------------------------------------------------------------------
 
+<<<<<<< HEAD
 func (r *Remote) encryptData(conf remoteConfig, destIP byte, packet []byte) []byte {
+=======
+func (r *Remote) encryptData(conf remoteConfig, packet []byte) []byte {
+>>>>>>> 69f2536 (WIP)
 	h := Header{
 		StreamID: dataStreamID,
 		Counter:  atomic.AddUint64(&r.sendCounter, 1),
 		SourceIP: r.Globals.LocalPeerIP,
+<<<<<<< HEAD
 		DestIP:   destIP,
+=======
+		DestIP:   r.RemotePeerIP,
+>>>>>>> 69f2536 (WIP)
 	}
 	return conf.DataCipher.Encrypt(h, packet, packet[len(packet):cap(packet)])
 }
@@ -116,6 +154,7 @@ func (r *Remote) SendDataTo(data []byte) {
 		return
 	}
 
+<<<<<<< HEAD
 	// Direct:
 
 	if conf.Direct {
@@ -124,6 +163,17 @@ func (r *Remote) SendDataTo(data []byte) {
 	}
 
 	// Relayed:
+=======
+	if conf.Direct {
+		r.sendDataDirect(conf, data)
+	} else {
+		r.sendDataRelayed(conf, data)
+	}
+}
+
+// sendDataRelayed sends data to the remote via the relay.
+func (r *Remote) sendDataRelayed(conf remoteConfig, data []byte) {
+>>>>>>> 69f2536 (WIP)
 	relay := r.RelayHandler.Load()
 
 	if relay == nil {
@@ -131,15 +181,29 @@ func (r *Remote) SendDataTo(data []byte) {
 		return
 	}
 
+<<<<<<< HEAD
 	relay.relayData(conf.Peer.PeerIP, r.encryptData(conf, conf.Peer.PeerIP, data))
 }
 
 func (r *Remote) relayData(toIP byte, enc []byte) {
+=======
+	relay.relayData(r.encryptData(conf, data))
+}
+
+// sendDataDirect sends data to the remote directly.
+func (r *Remote) sendDataDirect(conf remoteConfig, data []byte) {
+	r.logf("Sending data direct...")
+	r.sendUDP(r.encryptData(conf, data), conf.DirectAddr)
+}
+
+func (r *Remote) relayData(enc []byte) {
+>>>>>>> 69f2536 (WIP)
 	conf := r.conf()
 	if !conf.Up || !conf.Direct {
 		r.logf("Cannot relay: not up or not a direct connection")
 		return
 	}
+<<<<<<< HEAD
 	r.sendUDP(r.encryptData(conf, toIP, enc), conf.DirectAddr)
 }
 
@@ -154,6 +218,32 @@ func (r *Remote) sendControl(conf remoteConfig, data []byte) {
 
 	// Relayed:
 
+=======
+	r.sendDataDirect(conf, enc)
+}
+
+func (r *Remote) sendControl(conf remoteConfig, data []byte) {
+	if conf.Direct {
+		r.sendControlDirect(conf, data)
+	} else {
+		r.sendControlRelayed(conf, data)
+	}
+}
+
+func (r *Remote) sendControlToAddr(buf []byte, addr netip.AddrPort) {
+	enc := r.encryptControl(r.conf(), buf)
+	r.sendUDP(enc, addr)
+}
+
+func (r *Remote) sendControlDirect(conf remoteConfig, data []byte) {
+	r.logf("Sending control direct...")
+	enc := r.encryptControl(conf, data)
+	r.sendUDP(enc, conf.DirectAddr)
+}
+
+func (r *Remote) sendControlRelayed(conf remoteConfig, data []byte) {
+	r.logf("Sending control relayed...")
+>>>>>>> 69f2536 (WIP)
 	relay := r.RelayHandler.Load()
 
 	if relay == nil {
@@ -161,12 +251,16 @@ func (r *Remote) sendControl(conf remoteConfig, data []byte) {
 		return
 	}
 
+<<<<<<< HEAD
 	relay.relayData(conf.Peer.PeerIP, r.encryptControl(conf, data))
 }
 
 func (r *Remote) sendControlToAddr(buf []byte, addr netip.AddrPort) {
 	enc := r.encryptControl(r.conf(), buf)
 	r.sendUDP(enc, addr)
+=======
+	relay.relayData(r.encryptControl(conf, data))
+>>>>>>> 69f2536 (WIP)
 }
 
 func (r *Remote) forwardPacket(data []byte) {
@@ -244,9 +338,13 @@ func (r *Remote) handleDataPacket(h Header, data []byte) {
 	// For local.
 	if h.DestIP == r.LocalPeerIP {
 		if _, err := r.IFace.Write(dec); err != nil {
+<<<<<<< HEAD
 			// This could be a malformed packet from a peer, so we don't crash if it
 			// happens.
 			r.logf("Failed to write to interface: %v", err)
+=======
+			log.Fatalf("Failed to write to interface: %v", err)
+>>>>>>> 69f2536 (WIP)
 		}
 		return
 	}
@@ -275,6 +373,10 @@ func (r *Remote) HandleLocalDiscoveryPacket(h Header, srcAddr netip.AddrPort, da
 		SrcIP:   h.SourceIP,
 		SrcAddr: srcAddr,
 	}
+<<<<<<< HEAD
+=======
+	r.logf("Got local discovery packet from %v.", srcAddr)
+>>>>>>> 69f2536 (WIP)
 
 	select {
 	case r.messages <- msg:

peer/remotefsm.go

@@ -9,11 +9,14 @@ import (
 
 type stateFunc func(msg any) stateFunc
 
+<<<<<<< HEAD
 type sentProbe struct {
 	SentAt time.Time
 	Addr   netip.AddrPort
 }
 
+=======
+>>>>>>> 69f2536 (WIP)
 type remoteFSM struct {
 	*Remote
 
@@ -161,13 +164,23 @@ func (r *remoteFSM) stateServer_onInit(msg controlMsg[packetInit]) {
 }
 
 func (r *remoteFSM) stateServer_onSyn(msg controlMsg[packetSyn]) {
+<<<<<<< HEAD
+=======
+	r.logf("Got SYN: %v", msg.Packet)
+>>>>>>> 69f2536 (WIP)
 	r.lastSeen = time.Now()
 	p := msg.Packet
 
 	// Before we can respond to this packet, we need to make sure the
 	// route is setup properly.
 	conf := r.conf()
+<<<<<<< HEAD
 	logSyn := !conf.Up || conf.Direct != p.Direct
+=======
+	if !conf.Up || conf.Direct != p.Direct {
+		r.logf("Got SYN.")
+	}
+>>>>>>> 69f2536 (WIP)
 
 	conf.Up = true
 	conf.Direct = p.Direct
@@ -181,10 +194,13 @@ func (r *remoteFSM) stateServer_onSyn(msg controlMsg[packetSyn]) {
 
 	r.updateConf(conf)
 
+<<<<<<< HEAD
 	if logSyn {
 		r.logf("Got SYN.")
 	}
 
+=======
+>>>>>>> 69f2536 (WIP)
 	r.sendControl(conf, packetAck{
 		TraceID:       p.TraceID,
 		ToAddr:        conf.DirectAddr,
@@ -201,7 +217,11 @@ func (r *remoteFSM) stateServer_onSyn(msg controlMsg[packetSyn]) {
 			break
 		}
 		r.logf("Probing %v...", addr)
+<<<<<<< HEAD
 		r.sendControlToAddr(packetProbe{TraceID: r.NewTraceID()}.Marshal(r.buf), addr)
+=======
+		r.sendControlToAddr(packetProbe{TraceID: newTraceID()}.Marshal(r.buf), addr)
+>>>>>>> 69f2536 (WIP)
 	}
 }
 
@@ -269,7 +289,11 @@ func (r *remoteFSM) stateClientInit(iMsg any) stateFunc {
 
 func (r *remoteFSM) stateClientInit_sendInit() {
 	conf := r.conf()
+<<<<<<< HEAD
 	r.traceID = r.NewTraceID()
+=======
+	r.traceID = newTraceID()
+>>>>>>> 69f2536 (WIP)
 	init := packetInit{
 		TraceID: r.traceID,
 		Direct:  conf.Direct,
@@ -316,7 +340,11 @@ func (r *remoteFSM) enterClient() stateFunc {
 	conf := r.conf()
 	r.probes = make(map[uint64]sentProbe, 8)
 
+<<<<<<< HEAD
 	r.traceID = r.NewTraceID()
+=======
+	r.traceID = newTraceID()
+>>>>>>> 69f2536 (WIP)
 	r.stateClient_sendSyn(conf)
 
 	r.pingTimer.Reset(pingInterval)
@@ -382,7 +410,11 @@ func (r *remoteFSM) stateClient_cleanProbes() {
 }
 
 func (r *remoteFSM) stateClient_sendProbeTo(addr netip.AddrPort) {
+<<<<<<< HEAD
 	probe := packetProbe{TraceID: r.NewTraceID()}
+=======
+	probe := packetProbe{TraceID: newTraceID()}
+>>>>>>> 69f2536 (WIP)
 	r.probes[probe.TraceID] = sentProbe{
 		SentAt: time.Now(),
 		Addr:   addr,
@@ -408,7 +440,11 @@ func (r *remoteFSM) stateClient_onProbe(msg controlMsg[packetProbe]) {
 	conf.DirectAddr = sent.Addr
 	r.updateConf(conf)
 
+<<<<<<< HEAD
 	r.traceID = r.NewTraceID()
+=======
+	r.traceID = newTraceID()
+>>>>>>> 69f2536 (WIP)
 	r.stateClient_sendSyn(conf)
 	r.logf("Successful probe to %v.", sent.Addr)
 }
@@ -435,6 +471,10 @@ func (r *remoteFSM) stateClient_onPingTimer() stateFunc {
 		return r.enterClientInit()
 	}
 
+<<<<<<< HEAD
+=======
+	r.traceID = newTraceID()
+>>>>>>> 69f2536 (WIP)
 	r.stateClient_sendSyn(conf)
 	return r.stateClient
 }

peer/remotestate-disconnected.go

@@ -0,0 +1 @@
+package peer