WIP

README.md

@@ -1,5 +1,32 @@
 # vppn: Virtual Potentially Private Network
 
+## TO DO
+
+* Double buffering in IFReader and ConnReader ?
+* Replace time-based counter with startup counter
+  * 16 byte startupCounter
+  * (startupCount << 48) + counter
+  * pass startup count to newRoutingTable function (or global?)
+  * write / increment on startup
+* Clean up state machine - one class w/
+  * type stateFunc func(msg any) stateFunc
+  * "init" funcs: func enterDisconnected() stateFunc
+* ~~Idea: Use a bufferSet object to manager buffers. In function calls, buffers
+  should get used up, can panic if we run out of buffers to ensure we never
+  allocate~~
+* Idea: bufferSize should be large enough to split and use parts of the
+  buffer for encryption, etc. Yes.
+
+### Peer State Messages
+
+* peerUpdateMsg
+* packetInit
+* packetSyn
+* packetAck
+* packetProbe
+* packetLocalDiscovery
+* pingTimerMsg
+
 ## Hub Server Configuration
 
 ```
@@ -9,7 +36,6 @@ adduser user
 # Enable ssh.
 cp -r ~/.ssh /home/user/
 chown -R user:user /home/user/.ssh
-
 ```
 
 Upload `hub` executable:
@@ -56,7 +82,6 @@ Install the binary somewhere, for example `~/bin/vppn`.
 
 Create systemd file in `/etc/systemd/system/vppn.service`.
 
-
 ```
 [Service]
 AmbientCapabilities=CAP_NET_BIND_SERVICE CAP_NET_ADMIN
@@ -73,7 +98,6 @@ WantedBy=multi-user.target
 
 Add and start the service:
 
-
 ```
 systemctl daemon-reload
 systemctl enable vppn

peer/bufferset.go

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

peer/cipher-control.go

@@ -12,7 +12,7 @@ func newControlCipher(privKey, pubKey []byte) *controlCipher {
 	return &controlCipher{shared}
 }
 
-func (cc *controlCipher) Encrypt(h header, data, out []byte) []byte {
+func (cc *controlCipher) Encrypt(h Header, data, out []byte) []byte {
 	const s = controlHeaderSize
 	out = out[:s+controlCipherOverhead+len(data)]
 	h.Marshal(out[:s])

peer/cipher-control_test.go

@@ -40,7 +40,7 @@ func TestControlCipher(t *testing.T) {
 	}
 
 	for _, plaintext := range testCases {
-		h1 := header{
+		h1 := Header{
 			StreamID: controlStreamID,
 			Counter:  235153,
 			SourceIP: 4,
@@ -51,7 +51,7 @@ func TestControlCipher(t *testing.T) {
 
 		encrypted = c1.Encrypt(h1, plaintext, encrypted)
 
-		h2 := header{}
+		h2 := Header{}
 		h2.Parse(encrypted)
 		if !reflect.DeepEqual(h1, h2) {
 			t.Fatal(h1, h2)
@@ -80,7 +80,7 @@ func TestControlCipher_ShortCiphertext(t *testing.T) {
 
 func BenchmarkControlCipher_Encrypt(b *testing.B) {
 	c1, _ := newControlCipherForTesting()
-	h1 := header{
+	h1 := Header{
 		Counter:  235153,
 		SourceIP: 4,
 		DestIP:   88,
@@ -100,7 +100,7 @@ func BenchmarkControlCipher_Encrypt(b *testing.B) {
 func BenchmarkControlCipher_Decrypt(b *testing.B) {
 	c1, c2 := newControlCipherForTesting()
 
-	h1 := header{
+	h1 := Header{
 		Counter:  235153,
 		SourceIP: 4,
 		DestIP:   88,

peer/cipher-data.go

@@ -38,7 +38,7 @@ func (sc *dataCipher) Key() [32]byte {
 	return sc.key
 }
 
-func (sc *dataCipher) Encrypt(h header, data, out []byte) []byte {
+func (sc *dataCipher) Encrypt(h Header, data, out []byte) []byte {
 	const s = dataHeaderSize
 	out = out[:s+dataCipherOverhead+len(data)]
 	h.Marshal(out[:s])

peer/cipher-data_test.go

@@ -22,7 +22,7 @@ func TestDataCipher(t *testing.T) {
 	}
 
 	for _, plaintext := range testCases {
-		h1 := header{
+		h1 := Header{
 			StreamID: dataStreamID,
 			Counter:  235153,
 			SourceIP: 4,
@@ -33,7 +33,7 @@ func TestDataCipher(t *testing.T) {
 
 		dc1 := newDataCipher()
 		encrypted = dc1.Encrypt(h1, plaintext, encrypted)
-		h2 := header{}
+		h2 := Header{}
 		h2.Parse(encrypted)
 
 		dc2 := newDataCipherFromKey(dc1.Key())
@@ -67,7 +67,7 @@ func TestDataCipher_ModifyCiphertext(t *testing.T) {
 	}
 
 	for _, plaintext := range testCases {
-		h1 := header{
+		h1 := Header{
 			Counter:  235153,
 			SourceIP: 4,
 			DestIP:   88,
@@ -99,7 +99,7 @@ func TestDataCipher_ShortCiphertext(t *testing.T) {
 }
 
 func BenchmarkDataCipher_Encrypt(b *testing.B) {
-	h1 := header{
+	h1 := Header{
 		Counter:  235153,
 		SourceIP: 4,
 		DestIP:   88,
@@ -118,7 +118,7 @@ func BenchmarkDataCipher_Encrypt(b *testing.B) {
 }
 
 func BenchmarkDataCipher_Decrypt(b *testing.B) {
-	h1 := header{
+	h1 := Header{
 		Counter:  235153,
 		SourceIP: 4,
 		DestIP:   88,

peer/cipher.go

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

peer/connreader.go

@@ -1,56 +1,34 @@
 package peer
 
 import (
-	"io"
 	"log"
+	"net"
 	"net/netip"
-	"sync/atomic"
 )
 
-type connReader struct {
+type ConnReader struct {
-	// Input
+	Globals
-	readFromUDPAddrPort func([]byte) (int, netip.AddrPort, error)
+	conn *net.UDPConn
-
+	buf  []byte
-	// Output
-	writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error)
-	iface              io.Writer
-	handleControlMsg   func(fromIP byte, pkt any)
-
-	localIP byte
-	rt      *atomic.Pointer[routingTable]
-
-	buf    []byte
-	decBuf []byte
 }
 
-func newConnReader(
+func NewConnReader(g Globals, conn *net.UDPConn) *ConnReader {
-	readFromUDPAddrPort func([]byte) (int, netip.AddrPort, error),
+	return &ConnReader{
-	writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error),
+		Globals: g,
-	iface io.Writer,
+		conn:    conn,
-	handleControlMsg func(fromIP byte, pkt any),
+		buf:     make([]byte, bufferSize),
-	rt *atomic.Pointer[routingTable],
-) *connReader {
-	return &connReader{
-		readFromUDPAddrPort: readFromUDPAddrPort,
-		writeToUDPAddrPort:  writeToUDPAddrPort,
-		iface:               iface,
-		handleControlMsg:    handleControlMsg,
-		localIP:             rt.Load().LocalIP,
-		rt:                  rt,
-		buf:                 newBuf(),
-		decBuf:              newBuf(),
 	}
 }
 
-func (r *connReader) Run() {
+func (r *ConnReader) Run() {
 	for {
 		r.handleNextPacket()
 	}
 }
 
-func (r *connReader) handleNextPacket() {
+func (r *ConnReader) handleNextPacket() {
 	buf := r.buf[:bufferSize]
-	n, remoteAddr, err := r.readFromUDPAddrPort(buf)
+	n, remoteAddr, err := r.conn.ReadFromUDPAddrPort(buf)
 	if err != nil {
 		log.Fatalf("Failed to read from UDP port: %v", err)
 	}
@@ -64,77 +42,18 @@ func (r *connReader) handleNextPacket() {
 	buf = buf[:n]
 	h := parseHeader(buf)
 
-	rt := r.rt.Load()
+	remote := r.RemotePeers[h.SourceIP].Load()
-	peer := rt.Peers[h.SourceIP]
 
 	switch h.StreamID {
 	case controlStreamID:
-		r.handleControlPacket(remoteAddr, peer, h, buf)
+		remote.handleControlPacket(h, remoteAddr, buf)
 	case dataStreamID:
-		r.handleDataPacket(rt, peer, h, buf)
+		remote.handleDataPacket(h, buf)
 	default:
 		r.logf("Unknown stream ID: %d", h.StreamID)
 	}
 }
 
-func (r *connReader) handleControlPacket(
+func (r *ConnReader) logf(format string, args ...any) {
-	remoteAddr netip.AddrPort,
-	peer remotePeer,
-	h header,
-	enc []byte,
-) {
-	if peer.ControlCipher == nil {
-		r.logf("No control cipher for peer: %d", h.SourceIP)
-		return
-	}
-
-	if h.DestIP != r.localIP {
-		r.logf("Incorrect destination IP on control packet: %d", h.DestIP)
-		return
-	}
-
-	msg, err := peer.DecryptControlPacket(remoteAddr, h, enc, r.decBuf)
-	if err != nil {
-		r.logf("Failed to decrypt control packet: %v", err)
-		return
-	}
-
-	r.handleControlMsg(h.SourceIP, msg)
-}
-
-func (r *connReader) handleDataPacket(
-	rt *routingTable,
-	peer remotePeer,
-	h header,
-	enc []byte,
-) {
-	if !peer.Up {
-		r.logf("Not connected (recv).")
-		return
-	}
-
-	data, err := peer.DecryptDataPacket(h, enc, r.decBuf)
-	if err != nil {
-		r.logf("Failed to decrypt data packet: %v", err)
-		return
-	}
-
-	if h.DestIP == r.localIP {
-		if _, err := r.iface.Write(data); err != nil {
-			log.Fatalf("Failed to write to interface: %v", err)
-		}
-		return
-	}
-
-	remote := rt.Peers[h.DestIP]
-	if !remote.Direct {
-		r.logf("Unable to relay data to %d.", h.DestIP)
-		return
-	}
-
-	r.writeToUDPAddrPort(data, remote.DirectAddr)
-}
-
-func (r *connReader) logf(format string, args ...any) {
 	log.Printf("[ConnReader] "+format, args...)
 }

peer/data-flow.dot

@@ -1,12 +1,11 @@
 digraph d {
-    ifReader   -> connWriter;
+    ifReader   -> remote;
-    connReader -> ifWriter;
+
-    connReader -> connWriter;
+    connReader -> remote;
-    connReader -> supervisor;
+    mcReader   -> remote;
-    mcReader   -> supervisor;
+    remote -> connWriter;
-    supervisor -> connWriter;
+    remote -> ifWriter;
-    supervisor -> mcWriter;
+    hubPoller  -> remote;
-    hubPoller  -> supervisor;
 
     connWriter [shape="box"];
     mcWriter [shape="box"];

peer/files.go

@@ -8,8 +8,8 @@ import (
 	"vppn/m"
 )
 
-type localConfig struct {
+type LocalConfig struct {
-	PeerIP      byte
+	LocalPeerIP byte `json:"PeerIP"`
 	Network     []byte
 	PubKey      []byte
 	PrivKey     []byte
@@ -65,7 +65,7 @@ func storeJson(x any, outPath string) error {
 	return os.Rename(tmpPath, outPath)
 }
 
-func storePeerConfig(netName string, pc localConfig) error {
+func storePeerConfig(netName string, pc LocalConfig) error {
 	return storeJson(pc, peerConfigPath(netName))
 }
 
@@ -82,7 +82,7 @@ func loadJson(dataPath string, ptr any) error {
 	return json.Unmarshal(data, ptr)
 }
 
-func loadPeerConfig(netName string) (pc localConfig, err error) {
+func loadPeerConfig(netName string) (pc LocalConfig, err error) {
 	return pc, loadJson(peerConfigPath(netName), &pc)
 }
 

peer/globals.go

@@ -1,15 +1,18 @@
 package peer
 
 import (
+	"io"
 	"net"
 	"net/netip"
+	"sync"
+	"sync/atomic"
 	"time"
 )
 
 const (
 	version = 1
 
-	bufferSize = 1536
+	bufferSize = 8192 // Enough for data packets and encryption buffers.
 
 	if_mtu       = 1200
 	if_queue_len = 2048
@@ -28,10 +31,70 @@ var multicastAddr = net.UDPAddrFromAddrPort(netip.AddrPortFrom(
 	netip.AddrFrom4([4]byte{224, 0, 0, 157}),
 	4560))
 
-func newBuf() []byte {
-	return make([]byte, bufferSize)
-}
-
 type marshaller interface {
 	Marshal([]byte) []byte
 }
+
+// ----------------------------------------------------------------------------
+
+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/header.go

@@ -6,13 +6,14 @@ import "unsafe"
 
 const (
 	headerSize        = 12
-	controlStreamID   = 2
 	controlHeaderSize = 24
-	dataStreamID      = 1
 	dataHeaderSize    = 12
+
+	dataStreamID    = 1
+	controlStreamID = 2
 )
 
-type header struct {
+type Header struct {
 	Version  byte
 	StreamID byte
 	SourceIP byte
@@ -20,7 +21,7 @@ type header struct {
 	Counter  uint64 // Init with time.Now().Unix << 30 to ensure monotonic.
 }
 
-func parseHeader(b []byte) (h header) {
+func parseHeader(b []byte) (h Header) {
 	h.Version = b[0]
 	h.StreamID = b[1]
 	h.SourceIP = b[2]
@@ -29,7 +30,7 @@ func parseHeader(b []byte) (h header) {
 	return h
 }
 
-func (h *header) Parse(b []byte) {
+func (h *Header) Parse(b []byte) {
 	h.Version = b[0]
 	h.StreamID = b[1]
 	h.SourceIP = b[2]
@@ -37,7 +38,7 @@ func (h *header) Parse(b []byte) {
 	h.Counter = *(*uint64)(unsafe.Pointer(&b[4]))
 }
 
-func (h *header) Marshal(buf []byte) {
+func (h *Header) Marshal(buf []byte) {
 	buf[0] = h.Version
 	buf[1] = h.StreamID
 	buf[2] = h.SourceIP

peer/header_test.go

@@ -3,7 +3,7 @@ package peer
 import "testing"
 
 func TestHeaderMarshalParse(t *testing.T) {
-	nIn := header{
+	nIn := Header{
 		StreamID: 23,
 		Counter:  3212,
 		SourceIP: 34,
@@ -13,7 +13,7 @@ func TestHeaderMarshalParse(t *testing.T) {
 	buf := make([]byte, headerSize)
 	nIn.Marshal(buf)
 
-	nOut := header{}
+	nOut := Header{}
 	nOut.Parse(buf)
 	if nIn != nOut {
 		t.Fatal(nIn, nOut)

peer/hubpoller.go

@@ -10,22 +10,20 @@ import (
 	"vppn/m"
 )
 
-type hubPoller struct {
+type HubPoller struct {
-	client           *http.Client
+	Globals
-	req              *http.Request
+	client   *http.Client
-	versions         [256]int64
+	req      *http.Request
-	localIP          byte
+	versions [256]int64
-	netName          string
+	netName  string
-	handleControlMsg func(fromIP byte, msg any)
 }
 
-func newHubPoller(
+func NewHubPoller(
-	localIP byte,
+	g Globals,
 	netName,
 	hubURL,
 	apiKey string,
-	handleControlMsg func(byte, any),
+) (*HubPoller, error) {
-) (*hubPoller, error) {
 	u, err := url.Parse(hubURL)
 	if err != nil {
 		return nil, err
@@ -41,20 +39,19 @@ func newHubPoller(
 	}
 	req.SetBasicAuth("", apiKey)
 
-	return &hubPoller{
+	return &HubPoller{
-		client:           client,
+		Globals: g,
-		req:              req,
+		client:  client,
-		localIP:          localIP,
+		req:     req,
-		netName:          netName,
+		netName: netName,
-		handleControlMsg: handleControlMsg,
 	}, nil
 }
 
-func (hp *hubPoller) logf(s string, args ...any) {
+func (hp *HubPoller) logf(s string, args ...any) {
 	log.Printf("[HubPoller] "+s, args...)
 }
 
-func (hp *hubPoller) Run() {
+func (hp *HubPoller) Run() {
 	state, err := loadNetworkState(hp.netName)
 	if err != nil {
 		hp.logf("Failed to load network state: %v", err)
@@ -69,7 +66,7 @@ func (hp *hubPoller) Run() {
 	}
 }
 
-func (hp *hubPoller) pollHub() {
+func (hp *HubPoller) pollHub() {
 	var state m.NetworkState
 
 	resp, err := hp.client.Do(hp.req)
@@ -89,22 +86,26 @@ func (hp *hubPoller) pollHub() {
 		return
 	}
 
-	hp.applyNetworkState(state)
-
 	if err := storeNetworkState(hp.netName, state); err != nil {
 		hp.logf("Failed to store network state: %v", err)
 	}
+
+	hp.applyNetworkState(state)
 }
 
-func (hp *hubPoller) applyNetworkState(state m.NetworkState) {
+func (hp *HubPoller) applyNetworkState(state m.NetworkState) {
 	for i, peer := range state.Peers {
-		if i != int(hp.localIP) {
+		if i == int(hp.LocalPeerIP) {
-			if peer == nil || peer.Version != hp.versions[i] {
+			continue
-				hp.handleControlMsg(byte(i), peerUpdateMsg{Peer: state.Peers[i]})
+		}
-				if peer != nil {
+
-					hp.versions[i] = peer.Version
+		if peer != nil && peer.Version == hp.versions[i] {
-				}
+			continue
-			}
+		}
+
+		hp.RemotePeers[i].Load().HandlePeerUpdate(peerUpdateMsg{Peer: state.Peers[i]})
+		if peer != nil {
+			hp.versions[i] = peer.Version
 		}
 	}
 }

peer/ifreader.go

@@ -1,67 +1,35 @@
 package peer
 
 import (
-	"io"
 	"log"
-	"net/netip"
-	"sync/atomic"
 )
 
-type ifReader struct {
+type IFReader struct {
-	iface              io.Reader
+	Globals
-	writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error)
-	rt                 *atomic.Pointer[routingTable]
-	buf1               []byte
-	buf2               []byte
 }
 
-func newIFReader(
+func NewIFReader(g Globals) *IFReader {
-	iface io.Reader,
+	return &IFReader{Globals: g}
-	writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error),
-	rt *atomic.Pointer[routingTable],
-) *ifReader {
-	return &ifReader{iface, writeToUDPAddrPort, rt, newBuf(), newBuf()}
 }
 
-func (r *ifReader) Run() {
+func (r *IFReader) Run() {
-	packet := newBuf()
+	packet := make([]byte, bufferSize)
 	for {
 		r.handleNextPacket(packet)
 	}
 }
 
-func (r *ifReader) handleNextPacket(packet []byte) {
+func (r *IFReader) handleNextPacket(packet []byte) {
 	packet = r.readNextPacket(packet)
 	remoteIP, ok := r.parsePacket(packet)
 	if !ok {
 		return
 	}
-
+	r.RemotePeers[remoteIP].Load().SendDataTo(packet)
-	rt := r.rt.Load()
-	peer := rt.Peers[remoteIP]
-	if !peer.Up {
-		r.logf("Peer %d not up.", peer.IP)
-		return
-	}
-
-	enc := peer.EncryptDataPacket(peer.IP, packet, r.buf1)
-	if peer.Direct {
-		r.writeToUDPAddrPort(enc, peer.DirectAddr)
-		return
-	}
-
-	relay, ok := rt.GetRelay()
-	if !ok {
-		r.logf("Relay not available for peer %d.", peer.IP)
-		return
-	}
-
-	enc = relay.EncryptDataPacket(peer.IP, enc, r.buf2)
-	r.writeToUDPAddrPort(enc, relay.DirectAddr)
 }
 
-func (r *ifReader) readNextPacket(buf []byte) []byte {
+func (r *IFReader) readNextPacket(buf []byte) []byte {
-	n, err := r.iface.Read(buf[:cap(buf)])
+	n, err := r.IFace.Read(buf[:cap(buf)])
 	if err != nil {
 		log.Fatalf("Failed to read from interface: %v", err)
 	}
@@ -69,7 +37,9 @@ func (r *ifReader) readNextPacket(buf []byte) []byte {
 	return buf[:n]
 }
 
-func (r *ifReader) parsePacket(buf []byte) (byte, bool) {
+// parsePacket returns the VPN ip for the packet, and a boolean indicating
+// success.
+func (r *IFReader) parsePacket(buf []byte) (byte, bool) {
 	n := len(buf)
 	if n == 0 {
 		return 0, false
@@ -98,6 +68,6 @@ func (r *ifReader) parsePacket(buf []byte) (byte, bool) {
 	}
 }
 
-func (*ifReader) logf(s string, args ...any) {
+func (*IFReader) logf(s string, args ...any) {
 	log.Printf("[IFReader] "+s, args...)
 }

peer/main.go

@@ -6,18 +6,18 @@ import (
 )
 
 func Main() {
-	conf := mainArgs{}
+	args := mainArgs{}
 
-	flag.StringVar(&conf.NetName, "name", "", "[REQUIRED] The network name.")
+	flag.StringVar(&args.NetName, "name", "", "[REQUIRED] The network name.")
-	flag.StringVar(&conf.HubAddress, "hub-address", "", "[REQUIRED] The hub address.")
+	flag.StringVar(&args.HubAddress, "hub-address", "", "[REQUIRED] The hub address.")
-	flag.StringVar(&conf.APIKey, "api-key", "", "[REQUIRED] The node's API key.")
+	flag.StringVar(&args.APIKey, "api-key", "", "[REQUIRED] The node's API key.")
 	flag.Parse()
 
-	if conf.NetName == "" || conf.HubAddress == "" || conf.APIKey == "" {
+	if args.NetName == "" || args.HubAddress == "" || args.APIKey == "" {
 		flag.Usage()
 		os.Exit(1)
 	}
 
-	peer := newPeerMain(conf)
+	peer := newPeerMain(args)
 	peer.Run()
 }

peer/main_test.go

@@ -0,0 +1,5 @@
+package peer
+
+func newBuf() []byte {
+	return make([]byte, bufferSize)
+}

peer/mcreader.go

@@ -3,27 +3,19 @@ package peer
 import (
 	"log"
 	"net"
-	"sync/atomic"
 	"time"
 )
 
-func runMCReader(
+func RunMCReader(g Globals) {
-	rt *atomic.Pointer[routingTable],
-	handleControlMsg func(destIP byte, msg any),
-) {
 	for {
-		runMCReaderInner(rt, handleControlMsg)
+		runMCReaderInner(g)
 		time.Sleep(broadcastErrorTimeoutInterval)
 	}
 }
 
-func runMCReaderInner(
+func runMCReaderInner(g Globals) {
-	rt *atomic.Pointer[routingTable],
-	handleControlMsg func(destIP byte, msg any),
-) {
 	var (
-		raw  = newBuf()
+		buf  = make([]byte, bufferSize)
-		buf  = newBuf()
 		logf = func(s string, args ...any) {
 			log.Printf("[MCReader] "+s, args...)
 		}
@@ -37,35 +29,20 @@ func runMCReaderInner(
 
 	for {
 		conn.SetReadDeadline(time.Now().Add(32 * time.Second))
-		n, remoteAddr, err := conn.ReadFromUDPAddrPort(raw[:bufferSize])
+		n, remoteAddr, err := conn.ReadFromUDPAddrPort(buf[:bufferSize])
 		if err != nil {
 			logf("Failed to read from UDP port): %v", err)
 			return
 		}
 
-		raw = raw[:n]
+		buf = buf[:n]
-		h, ok := headerFromLocalDiscoveryPacket(raw)
+		h, ok := headerFromLocalDiscoveryPacket(buf)
 		if !ok {
 			logf("Failed to open discovery packet?")
 			continue
 		}
+		log.Printf("Got local discovery from %v: %v", remoteAddr, h)
 
-		peer := rt.Load().Peers[h.SourceIP]
+		g.RemotePeers[h.SourceIP].Load().HandleLocalDiscoveryPacket(h, remoteAddr, buf)
-		if peer.PubSignKey == nil {
-			logf("No signing key for peer %d.", h.SourceIP)
-			continue
-		}
-
-		if !verifyLocalDiscoveryPacket(raw, buf, peer.PubSignKey) {
-			logf("Invalid signature from peer: %d", h.SourceIP)
-			continue
-		}
-
-		msg := controlMsg[packetLocalDiscovery]{
-			SrcIP:   h.SourceIP,
-			SrcAddr: remoteAddr,
-		}
-		logf("Got discovery packet from peer %d.", h.SourceIP)
-		handleControlMsg(h.SourceIP, msg)
 	}
 }

peer/mcwriter.go

@@ -9,7 +9,7 @@ import (
 )
 
 func createLocalDiscoveryPacket(localIP byte, signingKey []byte) []byte {
-	h := header{
+	h := Header{
 		SourceIP: localIP,
 		DestIP:   255,
 	}
@@ -19,7 +19,7 @@ func createLocalDiscoveryPacket(localIP byte, signingKey []byte) []byte {
 	return sign.Sign(out[:0], buf, (*[64]byte)(signingKey))
 }
 
-func headerFromLocalDiscoveryPacket(pkt []byte) (h header, ok bool) {
+func headerFromLocalDiscoveryPacket(pkt []byte) (h Header, ok bool) {
 	if len(pkt) != headerSize+signOverhead {
 		return
 	}
@@ -36,7 +36,7 @@ func verifyLocalDiscoveryPacket(pkt, buf []byte, pubSignKey []byte) bool {
 
 // ----------------------------------------------------------------------------
 
-func runMCWriter(localIP byte, signingKey []byte) {
+func RunMCWriter(localIP byte, signingKey []byte) {
 	discoveryPacket := createLocalDiscoveryPacket(localIP, signingKey)
 
 	conn, err := net.ListenMulticastUDP("udp", nil, multicastAddr)

peer/peer.go

@@ -10,19 +10,14 @@ import (
 	"net/http"
 	"net/netip"
 	"net/url"
-	"sync"
-	"sync/atomic"
 	"vppn/m"
 )
 
 type peerMain struct {
-	conf       localConfig
+	Globals
-	rt         *atomic.Pointer[routingTable]
+	ifReader   *IFReader
-	ifReader   *ifReader
+	connReader *ConnReader
-	connReader *connReader
+	hubPoller  *HubPoller
-	iface      io.Writer
-	hubPoller  *hubPoller
-	super      *supervisor
 }
 
 type mainArgs struct {
@@ -53,13 +48,14 @@ func newPeerMain(args mainArgs) *peerMain {
 		log.Fatalf("Failed to load network state: %v", err)
 	}
 
-	iface, err := openInterface(config.Network, config.PeerIP, args.NetName)
+	iface, err := openInterface(config.Network, config.LocalPeerIP, args.NetName)
 	if err != nil {
 		log.Fatalf("Failed to open interface: %v", err)
 	}
 
-	localPeer := state.Peers[config.PeerIP]
+	localPeer := state.Peers[config.LocalPeerIP]
 
+	log.Printf("XXXXX %v %v", config.LocalPeerIP, localPeer)
 	myAddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf(":%d", localPeer.Port))
 	if err != nil {
 		log.Fatalf("Failed to resolve UDP address: %v", err)
@@ -74,59 +70,43 @@ func newPeerMain(args mainArgs) *peerMain {
 	conn.SetReadBuffer(1024 * 1024 * 8)
 	conn.SetWriteBuffer(1024 * 1024 * 8)
 
-	// Wrap write function - this is necessary to avoid starvation.
-	writeLock := sync.Mutex{}
-	writeToUDPAddrPort := func(b []byte, addr netip.AddrPort) (n int, err error) {
-		writeLock.Lock()
-		n, err = conn.WriteToUDPAddrPort(b, addr)
-		if err != nil {
-			logf("Failed to write packet: %v", err)
-		}
-		writeLock.Unlock()
-		return n, err
-	}
-
 	var localAddr netip.AddrPort
 	ip, localAddrValid := netip.AddrFromSlice(localPeer.PublicIP)
 	if localAddrValid {
 		localAddr = netip.AddrPortFrom(ip, localPeer.Port)
 	}
 
-	rt := newRoutingTable(localPeer.PeerIP, localAddr)
+	g := NewGlobals(config, localAddr, conn, iface)
-	rtPtr := &atomic.Pointer[routingTable]{}
-	rtPtr.Store(&rt)
 
-	ifReader := newIFReader(iface, writeToUDPAddrPort, rtPtr)
+	hubPoller, err := NewHubPoller(g, args.NetName, args.HubAddress, args.APIKey)
-	super := newSupervisor(writeToUDPAddrPort, rtPtr, config.PrivKey)
-	connReader := newConnReader(conn.ReadFromUDPAddrPort, writeToUDPAddrPort, iface, super.HandleControlMsg, rtPtr)
-	hubPoller, err := newHubPoller(config.PeerIP, args.NetName, args.HubAddress, args.APIKey, super.HandleControlMsg)
 	if err != nil {
 		log.Fatalf("Failed to create hub poller: %v", err)
 	}
 
 	return &peerMain{
-		conf:       config,
+		Globals:    g,
-		rt:         rtPtr,
+		ifReader:   NewIFReader(g),
-		iface:      iface,
+		connReader: NewConnReader(g, conn),
-		ifReader:   ifReader,
-		connReader: connReader,
 		hubPoller:  hubPoller,
-		super:      super,
 	}
 }
 
 func (p *peerMain) Run() {
+	for i := range p.RemotePeers {
+		remote := p.RemotePeers[i].Load()
+		go newRemoteFSM(remote).Run()
+	}
 
 	go p.ifReader.Run()
 	go p.connReader.Run()
-	p.super.Start()
 
-	if !p.rt.Load().LocalAddr.IsValid() {
+	if !p.LocalAddrValid {
-		go runMCWriter(p.conf.PeerIP, p.conf.PrivSignKey)
+		go RunMCWriter(p.LocalPeerIP, p.PrivSignKey)
-		go runMCReader(p.rt, p.super.HandleControlMsg)
+		go RunMCReader(p.Globals)
 	}
 
 	go p.hubPoller.Run()
+
 	select {}
 }
 
@@ -171,8 +151,8 @@ func initPeerWithHub(args mainArgs) {
 		log.Fatalf("Failed to parse configuration: %v\n%s", err, data)
 	}
 
-	config := localConfig{}
+	config := LocalConfig{}
-	config.PeerIP = initResp.PeerIP
+	config.LocalPeerIP = initResp.PeerIP
 	config.Network = initResp.Network
 	config.PubKey = keys.PubKey
 	config.PrivKey = keys.PrivKey

peer/peer_test.go

@@ -14,8 +14,8 @@ type P struct {
 	RT         *atomic.Pointer[routingTable]
 	Conn       *TestUDPConn
 	IFace      *TestIFace
-	ConnReader *connReader
+	ConnReader *ConnReader
-	IFReader   *ifReader
+	IFReader   *IFReader
 }
 
 func NewPeerForTesting(n *TestNetwork, ip byte, addr netip.AddrPort) P {

peer/peersuper.go

@@ -1,148 +1 @@
 package peer
-
-import (
-	"net/netip"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"git.crumpington.com/lib/go/ratelimiter"
-)
-
-type supervisor struct {
-	writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error)
-	staged             routingTable
-	shared             *atomic.Pointer[routingTable]
-	peers              [256]*peerSuper
-	lock               sync.Mutex
-
-	buf1 []byte
-	buf2 []byte
-}
-
-func newSupervisor(
-	writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error),
-	rt *atomic.Pointer[routingTable],
-	privKey []byte,
-) *supervisor {
-
-	routes := rt.Load()
-
-	s := &supervisor{
-		writeToUDPAddrPort: writeToUDPAddrPort,
-		staged:             *routes,
-		shared:             rt,
-		buf1:               newBuf(),
-		buf2:               newBuf(),
-	}
-
-	pubAddrs := newPubAddrStore(routes.LocalAddr)
-
-	for i := range s.peers {
-		state := &peerData{
-			publish:           s.publish,
-			sendControlPacket: s.send,
-			pingTimer:         time.NewTicker(timeoutInterval),
-			localIP:           routes.LocalIP,
-			remoteIP:          byte(i),
-			privKey:           privKey,
-			localAddr:         routes.LocalAddr,
-			pubAddrs:          pubAddrs,
-			staged:            routes.Peers[i],
-			limiter: ratelimiter.New(ratelimiter.Config{
-				FillPeriod:   20 * time.Millisecond,
-				MaxWaitCount: 1,
-			}),
-		}
-		s.peers[i] = newPeerSuper(state, state.pingTimer)
-	}
-
-	return s
-}
-
-func (s *supervisor) Start() {
-	for i := range s.peers {
-		go s.peers[i].Run()
-	}
-}
-
-func (s *supervisor) HandleControlMsg(destIP byte, msg any) {
-	s.peers[destIP].HandleControlMsg(msg)
-}
-
-func (s *supervisor) send(peer remotePeer, pkt marshaller) {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-
-	enc := peer.EncryptControlPacket(pkt, s.buf1, s.buf2)
-	if peer.Direct {
-		s.writeToUDPAddrPort(enc, peer.DirectAddr)
-		return
-	}
-
-	relay, ok := s.staged.GetRelay()
-	if !ok {
-		return
-	}
-
-	enc = relay.EncryptDataPacket(peer.IP, enc, s.buf1)
-	s.writeToUDPAddrPort(enc, relay.DirectAddr)
-}
-
-func (s *supervisor) publish(rp remotePeer) {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-
-	s.staged.Peers[rp.IP] = rp
-	s.ensureRelay()
-	copy := s.staged
-	s.shared.Store(&copy)
-}
-
-func (s *supervisor) ensureRelay() {
-	if _, ok := s.staged.GetRelay(); ok {
-		return
-	}
-
-	// TODO: Random selection? Something else?
-	for _, peer := range s.staged.Peers {
-		if peer.Up && peer.Direct && peer.Relay {
-			s.staged.RelayIP = peer.IP
-			return
-		}
-	}
-}
-
-// ----------------------------------------------------------------------------
-
-type peerSuper struct {
-	messages  chan any
-	state     peerState
-	pingTimer *time.Ticker
-}
-
-func newPeerSuper(state *peerData, pingTimer *time.Ticker) *peerSuper {
-	return &peerSuper{
-		messages:  make(chan any, 8),
-		state:     initPeerState(state, nil),
-		pingTimer: pingTimer,
-	}
-}
-
-func (s *peerSuper) HandleControlMsg(msg any) {
-	select {
-	case s.messages <- msg:
-	default:
-	}
-}
-
-func (s *peerSuper) Run() {
-	for {
-		select {
-		case <-s.pingTimer.C:
-			s.state = s.state.OnMsg(pingTimerMsg{})
-		case raw := <-s.messages:
-			s.state = s.state.OnMsg(raw)
-		}
-	}
-}

peer/relay.go

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

peer/relayhandler.go

@@ -0,0 +1,54 @@
+package peer
+
+import (
+	"log"
+	"sync"
+	"sync/atomic"
+)
+
+type relayHandler struct {
+	lock   sync.Mutex
+	relays map[byte]*Remote
+	relay  atomic.Pointer[Remote]
+}
+
+func newRelayHandler() *relayHandler {
+	return &relayHandler{
+		relays: make(map[byte]*Remote, 256),
+	}
+}
+
+func (h *relayHandler) Add(r *Remote) {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	h.relays[r.RemotePeerIP] = r
+
+	if h.relay.Load() == nil {
+		log.Printf("Setting Relay: %v", r.conf().Peer.Name)
+		h.relay.Store(r)
+	}
+}
+
+func (h *relayHandler) Remove(r *Remote) {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	log.Printf("Removing relay %d...", r.RemotePeerIP)
+	delete(h.relays, r.RemotePeerIP)
+
+	if h.relay.Load() == r {
+		// Remove current relay.
+		h.relay.Store(nil)
+
+		// Find new relay.
+		for _, r := range h.relays {
+			h.relay.Store(r)
+			break
+		}
+	}
+}
+
+func (h *relayHandler) Load() *Remote {
+	return h.relay.Load()
+}

peer/remote.go

@@ -0,0 +1,350 @@
+package peer
+
+import (
+	"fmt"
+	"log"
+	"net/netip"
+	"strings"
+	"sync/atomic"
+	"time"
+	"vppn/m"
+
+	"git.crumpington.com/lib/go/ratelimiter"
+)
+
+// ----------------------------------------------------------------------------
+// The remoteConfig is the shared, immutable configuration for a remote
+// peer. It's read and written atomically. See remote.config.
+// ----------------------------------------------------------------------------
+
+type remoteConfig struct {
+	Up            bool           // True if peer is up and we can send data.
+	Server        bool           // True if role is server.
+	Direct        bool           // True if this is a direct connection.
+	DirectAddr    netip.AddrPort // Remote address if directly connected.
+	ControlCipher *controlCipher
+	DataCipher    *dataCipher
+	Peer          *m.Peer
+}
+
+// CanRelay returns true if the remote configuration is able to relay packets.
+// to other hosts.
+func (rc remoteConfig) CanRelay() bool {
+	return rc.Up && rc.Direct && rc.Peer.Relay
+}
+
+// A Remote represents a remote peer and contains functions for handling
+// incoming control, data, and multicast packets, peer udpates, as well as
+// sending, forwarding, and relaying packets.
+type Remote struct {
+	Globals
+	RemotePeerIP byte // Immutable.
+
+	limiter     *ratelimiter.Limiter
+	dupCheck    *dupCheck
+	sendCounter uint64 // init to startupCount << 48. Atomic access only.
+
+	// config should be accessed via conf() and updateConf(...) methods.
+	config   atomic.Pointer[remoteConfig]
+	messages chan any
+}
+
+func newRemote(g Globals, remotePeerIP byte) *Remote {
+	r := &Remote{
+		Globals:      g,
+		RemotePeerIP: remotePeerIP,
+		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),
+	}
+	r.config.Store(&remoteConfig{})
+	return r
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *Remote) conf() remoteConfig {
+	return *(r.config.Load())
+}
+
+func (r *Remote) updateConf(conf remoteConfig) {
+	old := r.config.Load()
+	r.config.Store(&conf)
+
+	if !old.CanRelay() && conf.CanRelay() {
+		r.RelayHandler.Add(r)
+	}
+
+	if old.CanRelay() && !conf.CanRelay() {
+		r.RelayHandler.Remove(r)
+	}
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *Remote) sendUDP(b []byte, addr netip.AddrPort) {
+	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)
+	}
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *Remote) encryptData(conf remoteConfig, packet []byte) []byte {
+	h := Header{
+		StreamID: dataStreamID,
+		Counter:  atomic.AddUint64(&r.sendCounter, 1),
+		SourceIP: r.Globals.LocalPeerIP,
+		DestIP:   r.RemotePeerIP,
+	}
+	return conf.DataCipher.Encrypt(h, packet, packet[len(packet):cap(packet)])
+}
+
+func (r *Remote) encryptControl(conf remoteConfig, packet []byte) []byte {
+	h := Header{
+		StreamID: controlStreamID,
+		Counter:  atomic.AddUint64(&r.sendCounter, 1),
+		SourceIP: r.LocalPeerIP,
+		DestIP:   r.RemotePeerIP,
+	}
+	return conf.ControlCipher.Encrypt(h, packet, packet[len(packet):cap(packet)])
+}
+
+// ----------------------------------------------------------------------------
+
+// SendDataTo sends a data packet to the remote, called by the IFReader.
+func (r *Remote) SendDataTo(data []byte) {
+	conf := r.conf()
+	if !conf.Up {
+		r.logf("Cannot send: link down")
+		return
+	}
+
+	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) {
+	relay := r.RelayHandler.Load()
+
+	if relay == nil {
+		r.logf("Connot send: no relay")
+		return
+	}
+
+	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) {
+	conf := r.conf()
+	if !conf.Up || !conf.Direct {
+		r.logf("Cannot relay: not up or not a direct connection")
+		return
+	}
+	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...")
+	relay := r.RelayHandler.Load()
+
+	if relay == nil {
+		r.logf("Connot send: no relay")
+		return
+	}
+
+	relay.relayData(r.encryptControl(conf, data))
+}
+
+func (r *Remote) forwardPacket(data []byte) {
+	conf := r.conf()
+	if !conf.Up || !conf.Direct {
+		r.logf("Cannot forward to %d: not a direct connection", conf.Peer.PeerIP)
+		return
+	}
+	r.sendUDP(data, conf.DirectAddr)
+}
+
+// ----------------------------------------------------------------------------
+
+// HandlePacket is called by the ConnReader to handle an incoming packet.
+func (r *Remote) HandlePacket(h Header, srcAddr netip.AddrPort, data []byte) {
+	switch h.StreamID {
+	case controlStreamID:
+		r.handleControlPacket(h, srcAddr, data)
+	case dataStreamID:
+		r.handleDataPacket(h, data)
+	default:
+		r.logf("Unknown stream ID: %d", h.StreamID)
+	}
+}
+
+// Handle a control packet. Decrypt, verify, etc.
+func (r *Remote) handleControlPacket(h Header, srcAddr netip.AddrPort, data []byte) {
+	conf := r.conf()
+	if conf.ControlCipher == nil {
+		r.logf("No control cipher")
+		return
+	}
+
+	dec, ok := conf.ControlCipher.Decrypt(data, data[len(data):cap(data)])
+	if !ok {
+		r.logf("Failed to decrypt control packet")
+		return
+	}
+
+	if r.dupCheck.IsDup(h.Counter) {
+		r.logf("Dropping control packet as duplicate: %d", h.Counter)
+		return
+	}
+
+	msg, err := parseControlMsg(h.SourceIP, srcAddr, dec)
+	if err != nil {
+		r.logf("Failed to parse control packet: %v", err)
+		return
+	}
+
+	select {
+	case r.messages <- msg:
+	default:
+		r.logf("Dropping control message")
+	}
+}
+
+func (r *Remote) handleDataPacket(h Header, data []byte) {
+	conf := r.conf()
+	if conf.DataCipher == nil {
+		return
+	}
+
+	dec, ok := conf.DataCipher.Decrypt(data, data[len(data):cap(data)])
+	if !ok {
+		r.logf("Failed to decrypt data packet")
+		return
+	}
+
+	if r.dupCheck.IsDup(h.Counter) {
+		r.logf("Dropping data packet as duplicate: %d", h.Counter)
+		return
+	}
+
+	// For local.
+	if h.DestIP == r.LocalPeerIP {
+		if _, err := r.IFace.Write(dec); err != nil {
+			log.Fatalf("Failed to write to interface: %v", err)
+		}
+		return
+	}
+
+	// Forward.
+	dest := r.RemotePeers[h.DestIP].Load()
+	dest.forwardPacket(dec)
+}
+
+// ----------------------------------------------------------------------------
+
+// HandleLocalDiscoveryPacket is called by the MCReader.
+func (r *Remote) HandleLocalDiscoveryPacket(h Header, srcAddr netip.AddrPort, data []byte) {
+	conf := r.conf()
+	if conf.Peer.PubSignKey == nil {
+		r.logf("No signing key for discovery packet.")
+		return
+	}
+
+	if !verifyLocalDiscoveryPacket(data, data[len(data):cap(data)], conf.Peer.PubSignKey) {
+		r.logf("Invalid signature on discovery packet.")
+		return
+	}
+
+	msg := controlMsg[packetLocalDiscovery]{
+		SrcIP:   h.SourceIP,
+		SrcAddr: srcAddr,
+	}
+	r.logf("Got local discovery packet from %v.", srcAddr)
+
+	select {
+	case r.messages <- msg:
+	default:
+		r.logf("Dropping discovery message.")
+	}
+}
+
+// ----------------------------------------------------------------------------
+
+// HandlePeerUpdate is called by the HubPoller when it gets a new version of
+// the associated peer configuration.
+func (r *Remote) HandlePeerUpdate(msg peerUpdateMsg) {
+	r.messages <- msg
+}
+
+// ----------------------------------------------------------------------------
+
+func (s *Remote) logf(format string, args ...any) {
+	conf := s.conf()
+
+	b := strings.Builder{}
+	name := ""
+	if conf.Peer != nil {
+		name = conf.Peer.Name
+	}
+	b.WriteString(fmt.Sprintf("%03d", s.RemotePeerIP))
+
+	b.WriteString(fmt.Sprintf("%30s: ", name))
+
+	if conf.Server {
+		b.WriteString("SERVER | ")
+	} else {
+		b.WriteString("CLIENT | ")
+	}
+
+	if conf.Direct {
+		b.WriteString("DIRECT  | ")
+	} else {
+		b.WriteString("RELAYED | ")
+	}
+
+	if conf.Up {
+		b.WriteString("UP   | ")
+	} else {
+		b.WriteString("DOWN | ")
+	}
+
+	log.Printf(b.String()+format, args...)
+}

peer/remotefsm.go

@@ -0,0 +1,446 @@
+package peer
+
+import (
+	"bytes"
+	"net/netip"
+	"time"
+	"vppn/m"
+)
+
+type stateFunc func(msg any) stateFunc
+
+type remoteFSM struct {
+	*Remote
+
+	pingTimer *time.Ticker
+	lastSeen  time.Time
+	traceID   uint64
+	probes    map[uint64]sentProbe
+	sharedKey [32]byte
+
+	buf []byte
+}
+
+func newRemoteFSM(r *Remote) *remoteFSM {
+	fsm := &remoteFSM{
+		Remote:    r,
+		pingTimer: time.NewTicker(timeoutInterval),
+		probes:    map[uint64]sentProbe{},
+		buf:       make([]byte, bufferSize),
+	}
+	fsm.pingTimer.Stop()
+	return fsm
+}
+
+func (r *remoteFSM) Run() {
+	go func() {
+		for range r.pingTimer.C {
+			r.messages <- pingTimerMsg{}
+		}
+	}()
+	state := r.enterDisconnected()
+	for msg := range r.messages {
+		state = state(msg)
+	}
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *remoteFSM) enterDisconnected() stateFunc {
+	r.updateConf(remoteConfig{})
+	return r.stateDisconnected
+}
+
+func (r *remoteFSM) stateDisconnected(iMsg any) stateFunc {
+	switch msg := iMsg.(type) {
+	case peerUpdateMsg:
+		return r.enterPeerUpdating(msg.Peer)
+	case controlMsg[packetInit]:
+		r.logf("Unexpected INIT")
+	case controlMsg[packetSyn]:
+		r.logf("Unexpected SYN")
+	case controlMsg[packetAck]:
+		r.logf("Unexpected ACK")
+	case controlMsg[packetProbe]:
+		r.logf("Unexpected probe")
+	case controlMsg[packetLocalDiscovery]:
+		// Ignore
+	case pingTimerMsg:
+		r.logf("Unexpected ping")
+	default:
+		r.logf("Ignoring message: %#v", iMsg)
+	}
+
+	return r.stateDisconnected
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *remoteFSM) enterPeerUpdating(peer *m.Peer) stateFunc {
+	if peer == nil {
+		return r.enterDisconnected()
+	}
+
+	conf := remoteConfig{
+		Peer:          peer,
+		ControlCipher: newControlCipher(r.PrivKey, peer.PubKey),
+	}
+	r.updateConf(conf)
+
+	if _, isValid := netip.AddrFromSlice(peer.PublicIP); isValid {
+		if r.LocalAddrValid && r.LocalPeerIP < peer.PeerIP {
+			return r.enterServer()
+		}
+		return r.enterClientInit()
+	}
+
+	if r.LocalAddrValid || r.LocalPeerIP < peer.PeerIP {
+		return r.enterServer()
+	}
+
+	return r.enterClientInit()
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *remoteFSM) enterServer() stateFunc {
+
+	conf := r.conf()
+	conf.Server = true
+	r.updateConf(conf)
+	r.logf("==> Server")
+
+	r.pingTimer.Reset(pingInterval)
+	r.lastSeen = time.Now()
+	clear(r.sharedKey[:])
+	return r.stateServer
+}
+
+func (r *remoteFSM) stateServer(iMsg any) stateFunc {
+	switch msg := iMsg.(type) {
+	case peerUpdateMsg:
+		return r.enterPeerUpdating(msg.Peer)
+	case controlMsg[packetInit]:
+		r.stateServer_onInit(msg)
+	case controlMsg[packetSyn]:
+		r.stateServer_onSyn(msg)
+	case controlMsg[packetAck]:
+		r.logf("Unexpected ACK")
+	case controlMsg[packetProbe]:
+		r.stateServer_onProbe(msg)
+	case controlMsg[packetLocalDiscovery]:
+		// Ignore
+	case pingTimerMsg:
+		r.stateServer_onPingTimer()
+	default:
+		r.logf("Unexpected message: %#v", iMsg)
+	}
+
+	return r.stateServer
+}
+
+func (r *remoteFSM) stateServer_onInit(msg controlMsg[packetInit]) {
+	conf := r.conf()
+	conf.Up = false
+	conf.Direct = msg.Packet.Direct
+	conf.DirectAddr = msg.SrcAddr
+	r.updateConf(conf)
+
+	init := packetInit{
+		TraceID: msg.Packet.TraceID,
+		Direct:  conf.Direct,
+		Version: version,
+	}
+
+	r.sendControl(conf, init.Marshal(r.buf))
+}
+
+func (r *remoteFSM) stateServer_onSyn(msg controlMsg[packetSyn]) {
+	r.logf("Got SYN: %v", msg.Packet)
+	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()
+	if !conf.Up || conf.Direct != p.Direct {
+		r.logf("Got SYN.")
+	}
+
+	conf.Up = true
+	conf.Direct = p.Direct
+	conf.DirectAddr = msg.SrcAddr
+
+	// Update data cipher if the key has changed.
+	if !bytes.Equal(r.sharedKey[:], p.SharedKey[:]) {
+		conf.DataCipher = newDataCipherFromKey(p.SharedKey)
+		copy(r.sharedKey[:], p.SharedKey[:])
+	}
+
+	r.updateConf(conf)
+
+	r.sendControl(conf, packetAck{
+		TraceID:       p.TraceID,
+		ToAddr:        conf.DirectAddr,
+		PossibleAddrs: r.PubAddrs.Get(),
+	}.Marshal(r.buf))
+
+	if p.Direct {
+		return
+	}
+
+	// Send probes if not a direct connection.
+	for _, addr := range msg.Packet.PossibleAddrs {
+		if !addr.IsValid() {
+			break
+		}
+		r.logf("Probing %v...", addr)
+		r.sendControlToAddr(packetProbe{TraceID: newTraceID()}.Marshal(r.buf), addr)
+	}
+}
+
+func (r *remoteFSM) stateServer_onProbe(msg controlMsg[packetProbe]) {
+	if !msg.SrcAddr.IsValid() {
+		return
+	}
+
+	data := packetProbe{TraceID: msg.Packet.TraceID}.Marshal(r.buf)
+	r.sendControlToAddr(data, msg.SrcAddr)
+}
+
+func (r *remoteFSM) stateServer_onPingTimer() {
+	conf := r.conf()
+	if time.Since(r.lastSeen) > timeoutInterval && conf.Up {
+		conf.Up = false
+		r.updateConf(conf)
+		r.logf("Timeout.")
+	}
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *remoteFSM) enterClientInit() stateFunc {
+	conf := r.conf()
+	ip, ipValid := netip.AddrFromSlice(conf.Peer.PublicIP)
+
+	conf.Up = false
+	conf.Server = false
+	conf.Direct = ipValid
+	conf.DirectAddr = netip.AddrPortFrom(ip, conf.Peer.Port)
+	conf.DataCipher = newDataCipher()
+
+	r.updateConf(conf)
+	r.logf("==> ClientInit")
+
+	r.lastSeen = time.Now()
+	r.pingTimer.Reset(pingInterval)
+	r.stateClientInit_sendInit()
+	return r.stateClientInit
+}
+
+func (r *remoteFSM) stateClientInit(iMsg any) stateFunc {
+	switch msg := iMsg.(type) {
+	case peerUpdateMsg:
+		return r.enterPeerUpdating(msg.Peer)
+	case controlMsg[packetInit]:
+		return r.stateClientInit_onInit(msg)
+	case controlMsg[packetSyn]:
+		r.logf("Unexpected SYN")
+	case controlMsg[packetAck]:
+		r.logf("Unexpected ACK")
+	case controlMsg[packetProbe]:
+		// Ignore
+	case controlMsg[packetLocalDiscovery]:
+		// Ignore
+	case pingTimerMsg:
+		return r.stateClientInit_onPing()
+	default:
+		r.logf("Unexpected message: %#v", iMsg)
+	}
+
+	return r.stateClientInit
+}
+
+func (r *remoteFSM) stateClientInit_sendInit() {
+	conf := r.conf()
+	r.traceID = newTraceID()
+	init := packetInit{
+		TraceID: r.traceID,
+		Direct:  conf.Direct,
+		Version: version,
+	}
+	r.sendControl(conf, init.Marshal(r.buf))
+}
+
+func (r *remoteFSM) stateClientInit_onInit(msg controlMsg[packetInit]) stateFunc {
+	if msg.Packet.TraceID != r.traceID {
+		r.logf("Invalid trace ID on INIT.")
+		return r.stateClientInit
+	}
+	r.logf("Got INIT version %d.", msg.Packet.Version)
+	return r.enterClient()
+}
+
+func (r *remoteFSM) stateClientInit_onPing() stateFunc {
+	if time.Since(r.lastSeen) < timeoutInterval {
+		r.stateClientInit_sendInit()
+		return r.stateClientInit
+	}
+
+	// Direct connect failed. Try indirect.
+	conf := r.conf()
+
+	if conf.Direct {
+		conf.Direct = false
+		r.updateConf(conf)
+		r.lastSeen = time.Now()
+		r.stateClientInit_sendInit()
+		r.logf("Direct connection failed. Attempting indirect connection.")
+		return r.stateClientInit
+	}
+
+	// Indirect failed. Re-enter init state.
+	r.logf("Timeout.")
+	return r.enterClientInit()
+}
+
+// ----------------------------------------------------------------------------
+
+func (r *remoteFSM) enterClient() stateFunc {
+	conf := r.conf()
+	r.probes = make(map[uint64]sentProbe, 8)
+
+	r.traceID = newTraceID()
+	r.stateClient_sendSyn(conf)
+
+	r.pingTimer.Reset(pingInterval)
+	r.logf("==> Client")
+	return r.stateClient
+}
+
+func (r *remoteFSM) stateClient(iMsg any) stateFunc {
+	switch msg := iMsg.(type) {
+	case peerUpdateMsg:
+		return r.enterPeerUpdating(msg.Peer)
+	case controlMsg[packetAck]:
+		r.stateClient_onAck(msg)
+	case controlMsg[packetProbe]:
+		r.stateClient_onProbe(msg)
+	case controlMsg[packetLocalDiscovery]:
+		r.stateClient_onLocalDiscovery(msg)
+	case pingTimerMsg:
+		return r.stateClient_onPingTimer()
+	default:
+		r.logf("Ignoring message: %v", iMsg)
+	}
+	return r.stateClient
+}
+
+func (r *remoteFSM) stateClient_onAck(msg controlMsg[packetAck]) {
+	if msg.Packet.TraceID != r.traceID {
+		return
+	}
+
+	r.lastSeen = time.Now()
+
+	conf := r.conf()
+	if !conf.Up {
+		conf.Up = true
+		r.updateConf(conf)
+		r.logf("Got ACK.")
+	}
+
+	if conf.Direct {
+		r.PubAddrs.Store(msg.Packet.ToAddr)
+		return
+	}
+
+	// Relayed.
+
+	r.stateClient_cleanProbes()
+
+	for _, addr := range msg.Packet.PossibleAddrs {
+		if !addr.IsValid() {
+			break
+		}
+		r.stateClient_sendProbeTo(addr)
+	}
+}
+
+func (r *remoteFSM) stateClient_cleanProbes() {
+	for key, sent := range r.probes {
+		if time.Since(sent.SentAt) > pingInterval {
+			delete(r.probes, key)
+		}
+	}
+}
+
+func (r *remoteFSM) stateClient_sendProbeTo(addr netip.AddrPort) {
+	probe := packetProbe{TraceID: newTraceID()}
+	r.probes[probe.TraceID] = sentProbe{
+		SentAt: time.Now(),
+		Addr:   addr,
+	}
+	r.logf("Probing %v...", addr)
+	r.sendControlToAddr(probe.Marshal(r.buf), addr)
+}
+
+func (r *remoteFSM) stateClient_onProbe(msg controlMsg[packetProbe]) {
+	conf := r.conf()
+	if conf.Direct {
+		return
+	}
+
+	r.stateClient_cleanProbes()
+
+	sent, ok := r.probes[msg.Packet.TraceID]
+	if !ok {
+		return
+	}
+
+	conf.Direct = true
+	conf.DirectAddr = sent.Addr
+	r.updateConf(conf)
+
+	r.traceID = newTraceID()
+	r.stateClient_sendSyn(conf)
+	r.logf("Successful probe to %v.", sent.Addr)
+}
+
+func (r *remoteFSM) stateClient_onLocalDiscovery(msg controlMsg[packetLocalDiscovery]) {
+	conf := r.conf()
+	if conf.Direct {
+		return
+	}
+
+	// The source port will be the multicast port, so we'll have to
+	// construct the correct address using the peer's listed port.
+	addr := netip.AddrPortFrom(msg.SrcAddr.Addr(), conf.Peer.Port)
+	r.stateClient_sendProbeTo(addr)
+}
+
+func (r *remoteFSM) stateClient_onPingTimer() stateFunc {
+	conf := r.conf()
+
+	if time.Since(r.lastSeen) > timeoutInterval {
+		if conf.Up {
+			r.logf("Timeout.")
+		}
+		return r.enterClientInit()
+	}
+
+	r.traceID = newTraceID()
+	r.stateClient_sendSyn(conf)
+	return r.stateClient
+}
+
+func (r *remoteFSM) stateClient_sendSyn(conf remoteConfig) {
+	syn := packetSyn{
+		TraceID:       r.traceID,
+		SharedKey:     conf.DataCipher.Key(),
+		Direct:        conf.Direct,
+		PossibleAddrs: r.PubAddrs.Get(),
+	}
+
+	r.sendControl(conf, syn.Marshal(r.buf))
+}

peer/remotestate-disconnected.go

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

peer/routingtable.go

@@ -34,7 +34,7 @@ type remotePeer struct {
 }
 
 func (p remotePeer) EncryptDataPacket(destIP byte, data, out []byte) []byte {
-	h := header{
+	h := Header{
 		StreamID: dataStreamID,
 		Counter:  atomic.AddUint64(p.counter, 1),
 		SourceIP: p.localIP,
@@ -44,7 +44,7 @@ func (p remotePeer) EncryptDataPacket(destIP byte, data, out []byte) []byte {
 }
 
 // Decrypts and de-dups incoming data packets.
-func (p remotePeer) DecryptDataPacket(h header, enc, out []byte) ([]byte, error) {
+func (p remotePeer) DecryptDataPacket(h Header, enc, out []byte) ([]byte, error) {
 	dec, ok := p.DataCipher.Decrypt(enc, out)
 	if !ok {
 		return nil, errDecryptionFailed
@@ -60,7 +60,7 @@ func (p remotePeer) DecryptDataPacket(h header, enc, out []byte) ([]byte, error)
 // Peer must have a ControlCipher.
 func (p remotePeer) EncryptControlPacket(pkt marshaller, tmp, out []byte) []byte {
 	tmp = pkt.Marshal(tmp)
-	h := header{
+	h := Header{
 		StreamID: controlStreamID,
 		Counter:  atomic.AddUint64(p.counter, 1),
 		SourceIP: p.localIP,
@@ -73,7 +73,7 @@ func (p remotePeer) EncryptControlPacket(pkt marshaller, tmp, out []byte) []byte
 // Returns a controlMsg[PacketType]. Peer must have a non-nil ControlCipher.
 //
 // This function also drops packets with duplicate sequence numbers.
-func (p remotePeer) DecryptControlPacket(fromAddr netip.AddrPort, h header, enc, tmp []byte) (any, error) {
+func (p remotePeer) DecryptControlPacket(fromAddr netip.AddrPort, h Header, enc, tmp []byte) (any, error) {
 	out, ok := p.ControlCipher.Decrypt(enc, tmp)
 	if !ok {
 		return nil, errDecryptionFailed