From ab246b2a90380bcb893001fbd4290ab5a6ce6500 Mon Sep 17 00:00:00 2001 From: jdl Date: Tue, 26 Aug 2025 15:33:27 +0200 Subject: [PATCH] WIP --- README.md | 30 ++- peer/bufferset.go | 1 + peer/cipher-control.go | 2 +- peer/cipher-control_test.go | 8 +- peer/cipher-data.go | 2 +- peer/cipher-data_test.go | 10 +- peer/cipher.go | 1 + peer/connreader.go | 115 ++------ peer/data-flow.dot | 15 +- peer/files.go | 8 +- peer/globals.go | 73 ++++- peer/header.go | 13 +- peer/header_test.go | 4 +- peer/hubpoller.go | 61 ++--- peer/ifreader.go | 58 +--- peer/main.go | 12 +- peer/main_test.go | 5 + peer/mcreader.go | 41 +-- peer/mcwriter.go | 6 +- peer/peer.go | 64 ++--- peer/peer_test.go | 4 +- peer/peersuper.go | 147 ---------- peer/relay.go | 1 + peer/relayhandler.go | 54 ++++ peer/remote.go | 350 ++++++++++++++++++++++++ peer/remotefsm.go | 446 +++++++++++++++++++++++++++++++ peer/remotestate-disconnected.go | 1 + peer/routingtable.go | 8 +- 28 files changed, 1093 insertions(+), 447 deletions(-) create mode 100644 peer/bufferset.go create mode 100644 peer/cipher.go create mode 100644 peer/main_test.go create mode 100644 peer/relay.go create mode 100644 peer/relayhandler.go create mode 100644 peer/remote.go create mode 100644 peer/remotefsm.go create mode 100644 peer/remotestate-disconnected.go diff --git a/README.md b/README.md index 29ae92c..2fa3eb3 100644 --- a/README.md +++ b/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 diff --git a/peer/bufferset.go b/peer/bufferset.go new file mode 100644 index 0000000..333deff --- /dev/null +++ b/peer/bufferset.go @@ -0,0 +1 @@ +package peer diff --git a/peer/cipher-control.go b/peer/cipher-control.go index bfecaeb..178ff97 100644 --- a/peer/cipher-control.go +++ b/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]) diff --git a/peer/cipher-control_test.go b/peer/cipher-control_test.go index 916d2ea..abeb8d5 100644 --- a/peer/cipher-control_test.go +++ b/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, diff --git a/peer/cipher-data.go b/peer/cipher-data.go index 9b229bb..5ce8555 100644 --- a/peer/cipher-data.go +++ b/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]) diff --git a/peer/cipher-data_test.go b/peer/cipher-data_test.go index ac9a03a..4a388f8 100644 --- a/peer/cipher-data_test.go +++ b/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, diff --git a/peer/cipher.go b/peer/cipher.go new file mode 100644 index 0000000..333deff --- /dev/null +++ b/peer/cipher.go @@ -0,0 +1 @@ +package peer diff --git a/peer/connreader.go b/peer/connreader.go index 4c156f4..36c8b8f 100644 --- a/peer/connreader.go +++ b/peer/connreader.go @@ -1,56 +1,34 @@ package peer import ( - "io" "log" + "net" "net/netip" - "sync/atomic" ) -type connReader struct { - // Input - readFromUDPAddrPort func([]byte) (int, netip.AddrPort, error) - - // 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 +type ConnReader struct { + Globals + conn *net.UDPConn + buf []byte } -func newConnReader( - readFromUDPAddrPort func([]byte) (int, netip.AddrPort, error), - writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error), - iface io.Writer, - handleControlMsg func(fromIP byte, pkt any), - 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 NewConnReader(g Globals, conn *net.UDPConn) *ConnReader { + return &ConnReader{ + Globals: g, + conn: conn, + buf: make([]byte, bufferSize), } } -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() - peer := rt.Peers[h.SourceIP] + remote := r.RemotePeers[h.SourceIP].Load() 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( - 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) { +func (r *ConnReader) logf(format string, args ...any) { log.Printf("[ConnReader] "+format, args...) } diff --git a/peer/data-flow.dot b/peer/data-flow.dot index 45b6f05..465f155 100644 --- a/peer/data-flow.dot +++ b/peer/data-flow.dot @@ -1,12 +1,11 @@ digraph d { - ifReader -> connWriter; - connReader -> ifWriter; - connReader -> connWriter; - connReader -> supervisor; - mcReader -> supervisor; - supervisor -> connWriter; - supervisor -> mcWriter; - hubPoller -> supervisor; + ifReader -> remote; + + connReader -> remote; + mcReader -> remote; + remote -> connWriter; + remote -> ifWriter; + hubPoller -> remote; connWriter [shape="box"]; mcWriter [shape="box"]; diff --git a/peer/files.go b/peer/files.go index a7d9566..e7e7022 100644 --- a/peer/files.go +++ b/peer/files.go @@ -8,8 +8,8 @@ import ( "vppn/m" ) -type localConfig struct { - PeerIP byte +type LocalConfig struct { + 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) } diff --git a/peer/globals.go b/peer/globals.go index 6dd26eb..3396e9c 100644 --- a/peer/globals.go +++ b/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 +} diff --git a/peer/header.go b/peer/header.go index fae3780..887c4dd 100644 --- a/peer/header.go +++ b/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 diff --git a/peer/header_test.go b/peer/header_test.go index 11e2f8f..f644a36 100644 --- a/peer/header_test.go +++ b/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) diff --git a/peer/hubpoller.go b/peer/hubpoller.go index 0082989..a0d79d0 100644 --- a/peer/hubpoller.go +++ b/peer/hubpoller.go @@ -10,22 +10,20 @@ import ( "vppn/m" ) -type hubPoller struct { - client *http.Client - req *http.Request - versions [256]int64 - localIP byte - netName string - handleControlMsg func(fromIP byte, msg any) +type HubPoller struct { + Globals + client *http.Client + req *http.Request + versions [256]int64 + netName string } -func newHubPoller( - localIP byte, +func NewHubPoller( + 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{ - client: client, - req: req, - localIP: localIP, - netName: netName, - handleControlMsg: handleControlMsg, + return &HubPoller{ + Globals: g, + client: client, + req: req, + netName: netName, }, 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 peer == nil || peer.Version != hp.versions[i] { - hp.handleControlMsg(byte(i), peerUpdateMsg{Peer: state.Peers[i]}) - if peer != nil { - hp.versions[i] = peer.Version - } - } + if i == int(hp.LocalPeerIP) { + continue + } + + 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 } } } diff --git a/peer/ifreader.go b/peer/ifreader.go index 2419758..ebade54 100644 --- a/peer/ifreader.go +++ b/peer/ifreader.go @@ -1,67 +1,35 @@ package peer import ( - "io" "log" - "net/netip" - "sync/atomic" ) -type ifReader struct { - iface io.Reader - writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error) - rt *atomic.Pointer[routingTable] - buf1 []byte - buf2 []byte +type IFReader struct { + Globals } -func newIFReader( - iface io.Reader, - writeToUDPAddrPort func([]byte, netip.AddrPort) (int, error), - rt *atomic.Pointer[routingTable], -) *ifReader { - return &ifReader{iface, writeToUDPAddrPort, rt, newBuf(), newBuf()} +func NewIFReader(g Globals) *IFReader { + return &IFReader{Globals: g} } -func (r *ifReader) Run() { - packet := newBuf() +func (r *IFReader) Run() { + 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 } - - 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) + r.RemotePeers[remoteIP].Load().SendDataTo(packet) } -func (r *ifReader) readNextPacket(buf []byte) []byte { - n, err := r.iface.Read(buf[:cap(buf)]) +func (r *IFReader) readNextPacket(buf []byte) []byte { + 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...) } diff --git a/peer/main.go b/peer/main.go index 819effb..bd20de6 100644 --- a/peer/main.go +++ b/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(&conf.HubAddress, "hub-address", "", "[REQUIRED] The hub address.") - flag.StringVar(&conf.APIKey, "api-key", "", "[REQUIRED] The node's API key.") + flag.StringVar(&args.NetName, "name", "", "[REQUIRED] The network name.") + flag.StringVar(&args.HubAddress, "hub-address", "", "[REQUIRED] The hub address.") + 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() } diff --git a/peer/main_test.go b/peer/main_test.go new file mode 100644 index 0000000..c759212 --- /dev/null +++ b/peer/main_test.go @@ -0,0 +1,5 @@ +package peer + +func newBuf() []byte { + return make([]byte, bufferSize) +} diff --git a/peer/mcreader.go b/peer/mcreader.go index 3c1086d..5bbfe71 100644 --- a/peer/mcreader.go +++ b/peer/mcreader.go @@ -3,27 +3,19 @@ package peer import ( "log" "net" - "sync/atomic" "time" ) -func runMCReader( - rt *atomic.Pointer[routingTable], - handleControlMsg func(destIP byte, msg any), -) { +func RunMCReader(g Globals) { for { - runMCReaderInner(rt, handleControlMsg) + runMCReaderInner(g) time.Sleep(broadcastErrorTimeoutInterval) } } -func runMCReaderInner( - rt *atomic.Pointer[routingTable], - handleControlMsg func(destIP byte, msg any), -) { +func runMCReaderInner(g Globals) { var ( - raw = newBuf() - buf = newBuf() + buf = make([]byte, bufferSize) 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] - h, ok := headerFromLocalDiscoveryPacket(raw) + buf = buf[:n] + 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] - 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) + g.RemotePeers[h.SourceIP].Load().HandleLocalDiscoveryPacket(h, remoteAddr, buf) } } diff --git a/peer/mcwriter.go b/peer/mcwriter.go index c20ed61..0b520d1 100644 --- a/peer/mcwriter.go +++ b/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) diff --git a/peer/peer.go b/peer/peer.go index ade03ce..56adea5 100644 --- a/peer/peer.go +++ b/peer/peer.go @@ -10,19 +10,14 @@ import ( "net/http" "net/netip" "net/url" - "sync" - "sync/atomic" "vppn/m" ) type peerMain struct { - conf localConfig - rt *atomic.Pointer[routingTable] - ifReader *ifReader - connReader *connReader - iface io.Writer - hubPoller *hubPoller - super *supervisor + Globals + ifReader *IFReader + connReader *ConnReader + hubPoller *HubPoller } 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) - rtPtr := &atomic.Pointer[routingTable]{} - rtPtr.Store(&rt) + g := NewGlobals(config, localAddr, conn, iface) - ifReader := newIFReader(iface, writeToUDPAddrPort, rtPtr) - 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) + hubPoller, err := NewHubPoller(g, args.NetName, args.HubAddress, args.APIKey) if err != nil { log.Fatalf("Failed to create hub poller: %v", err) } return &peerMain{ - conf: config, - rt: rtPtr, - iface: iface, - ifReader: ifReader, - connReader: connReader, + Globals: g, + ifReader: NewIFReader(g), + connReader: NewConnReader(g, conn), 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() { - go runMCWriter(p.conf.PeerIP, p.conf.PrivSignKey) - go runMCReader(p.rt, p.super.HandleControlMsg) + if !p.LocalAddrValid { + go RunMCWriter(p.LocalPeerIP, p.PrivSignKey) + 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.PeerIP = initResp.PeerIP + config := LocalConfig{} + config.LocalPeerIP = initResp.PeerIP config.Network = initResp.Network config.PubKey = keys.PubKey config.PrivKey = keys.PrivKey diff --git a/peer/peer_test.go b/peer/peer_test.go index 2c25812..863ca8f 100644 --- a/peer/peer_test.go +++ b/peer/peer_test.go @@ -14,8 +14,8 @@ type P struct { RT *atomic.Pointer[routingTable] Conn *TestUDPConn IFace *TestIFace - ConnReader *connReader - IFReader *ifReader + ConnReader *ConnReader + IFReader *IFReader } func NewPeerForTesting(n *TestNetwork, ip byte, addr netip.AddrPort) P { diff --git a/peer/peersuper.go b/peer/peersuper.go index 2ce6d03..333deff 100644 --- a/peer/peersuper.go +++ b/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(©) -} - -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) - } - } -} diff --git a/peer/relay.go b/peer/relay.go new file mode 100644 index 0000000..333deff --- /dev/null +++ b/peer/relay.go @@ -0,0 +1 @@ +package peer diff --git a/peer/relayhandler.go b/peer/relayhandler.go new file mode 100644 index 0000000..a8e9b3d --- /dev/null +++ b/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() +} diff --git a/peer/remote.go b/peer/remote.go new file mode 100644 index 0000000..bcbda78 --- /dev/null +++ b/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...) +} diff --git a/peer/remotefsm.go b/peer/remotefsm.go new file mode 100644 index 0000000..8097c46 --- /dev/null +++ b/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)) +} diff --git a/peer/remotestate-disconnected.go b/peer/remotestate-disconnected.go new file mode 100644 index 0000000..333deff --- /dev/null +++ b/peer/remotestate-disconnected.go @@ -0,0 +1 @@ +package peer diff --git a/peer/routingtable.go b/peer/routingtable.go index 3f0aac3..d29d4b0 100644 --- a/peer/routingtable.go +++ b/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