From 8c618616dd4fd8084da862a309d33def1198a3a4 Mon Sep 17 00:00:00 2001 From: jdl Date: Fri, 31 Jan 2025 21:32:47 +0100 Subject: [PATCH] wip --- peer/connreader.go | 48 ++++++++++--------------- peer/connreader_test.go | 77 ++++++++++++++++++++++++++++++++++------- peer/connwriter.go | 36 +++++++++---------- peer/connwriter_test.go | 28 +++++++-------- peer/crypto.go | 31 +++++++++-------- peer/crypto_test.go | 41 ++++++---------------- peer/header.go | 7 ++-- peer/ifreader.go | 22 ++++++------ peer/ifreader_test.go | 40 ++++++++++----------- peer/interfaces.go | 15 +++++--- peer/mcreader.go | 57 ++++++++++++++++++++++++++++++ peer/mcwriter.go | 9 ----- peer/state.go | 16 ++++----- 13 files changed, 250 insertions(+), 177 deletions(-) create mode 100644 peer/mcreader.go diff --git a/peer/connreader.go b/peer/connreader.go index 757b37c..b127030 100644 --- a/peer/connreader.go +++ b/peer/connreader.go @@ -12,11 +12,10 @@ type connReader struct { sender encryptedPacketSender super controlMsgHandler localIP byte - routes [256]*atomic.Pointer[peerRoute] + peers [256]*atomic.Pointer[RemotePeer] - buf []byte - decBuf []byte - dupChecks [256]*dupCheck + buf []byte + decBuf []byte } func newConnReader( @@ -25,7 +24,7 @@ func newConnReader( sender encryptedPacketSender, super controlMsgHandler, localIP byte, - routes [256]*atomic.Pointer[peerRoute], + peers [256]*atomic.Pointer[RemotePeer], ) *connReader { return &connReader{ conn: conn, @@ -33,15 +32,9 @@ func newConnReader( sender: sender, super: super, localIP: localIP, - routes: routes, + peers: peers, buf: make([]byte, bufferSize), decBuf: make([]byte, bufferSize), - dupChecks: func() (out [256]*dupCheck) { - for i := range out { - out[i] = newDupCheck(0) - } - return - }(), } } @@ -69,19 +62,16 @@ func (r *connReader) handleNextPacket() { remoteAddr = netip.AddrPortFrom(remoteAddr.Addr().Unmap(), remoteAddr.Port()) buf = buf[:n] - h, ok := parseHeader(buf) - if !ok { - return - } + h := parseHeader(buf) - route := r.routes[h.SourceIP].Load() + peer := r.peers[h.SourceIP].Load() switch h.StreamID { case controlStreamID: - r.handleControlPacket(route, remoteAddr, h, buf) + r.handleControlPacket(peer, remoteAddr, h, buf) case dataStreamID: - r.handleDataPacket(route, h, buf) + r.handleDataPacket(peer, h, buf) default: r.logf("Unknown stream ID: %d", h.StreamID) @@ -89,12 +79,12 @@ func (r *connReader) handleNextPacket() { } func (r *connReader) handleControlPacket( - route *peerRoute, + peer *RemotePeer, addr netip.AddrPort, h header, enc []byte, ) { - if route.ControlCipher == nil { + if peer.ControlCipher == nil { return } @@ -103,7 +93,7 @@ func (r *connReader) handleControlPacket( return } - msg, err := decryptControlPacket(route, addr, h, enc, r.decBuf) + msg, err := decryptControlPacket(peer, addr, h, enc, r.decBuf) if err != nil { r.logf("Failed to decrypt control packet: %v", err) return @@ -112,13 +102,13 @@ func (r *connReader) handleControlPacket( r.super.HandleControlMsg(msg) } -func (r *connReader) handleDataPacket(route *peerRoute, h header, enc []byte) { - if !route.Up { +func (r *connReader) handleDataPacket(peer *RemotePeer, h header, enc []byte) { + if !peer.Up { r.logf("Not connected (recv).") return } - data, err := decryptDataPacket(route, h, enc, r.decBuf) + data, err := decryptDataPacket(peer, h, enc, r.decBuf) if err != nil { r.logf("Failed to decrypt data packet: %v", err) return @@ -131,11 +121,11 @@ func (r *connReader) handleDataPacket(route *peerRoute, h header, enc []byte) { return } - destRoute := r.routes[h.DestIP].Load() - if !destRoute.Up { - r.logf("Not connected (relay): %d", destRoute.IP) + destPeer := r.peers[h.DestIP].Load() + if !destPeer.Up { + r.logf("Not connected (relay): %d", destPeer.IP) return } - r.sender.SendEncryptedDataPacket(data, destRoute) + r.sender.SendEncryptedDataPacket(data, destPeer) } diff --git a/peer/connreader_test.go b/peer/connreader_test.go index fe8e6cb..39da83c 100644 --- a/peer/connreader_test.go +++ b/peer/connreader_test.go @@ -2,6 +2,7 @@ package peer import ( "bytes" + "crypto/rand" "net/netip" "reflect" "sync/atomic" @@ -19,14 +20,14 @@ func (w *mockIfWriter) Write(b []byte) (int, error) { type mockEncryptedPacket struct { Packet []byte - Route *peerRoute + Route *RemotePeer } type mockEncryptedPacketSender struct { Sent []mockEncryptedPacket } -func (m *mockEncryptedPacketSender) SendEncryptedDataPacket(pkt []byte, route *peerRoute) { +func (m *mockEncryptedPacketSender) SendEncryptedDataPacket(pkt []byte, route *RemotePeer) { m.Sent = append(m.Sent, mockEncryptedPacket{ Packet: bytes.Clone(pkt), Route: route, @@ -65,8 +66,8 @@ type connReaderTestHarness struct { R *connReader WRemote *connWriter WRelayRemote *connWriter - Remote *peerRoute - RelayRemote *peerRoute + Remote *RemotePeer + RelayRemote *RemotePeer IFace *mockIfWriter Sender *mockEncryptedPacketSender Super *mockControlMsgHandler @@ -75,10 +76,10 @@ type connReaderTestHarness struct { // Peer 2 is indirect, peer 3 is direct. func newConnReadeTestHarness() (h connReaderTestHarness) { pipe := newUDPPipe() - routes := [256]*atomic.Pointer[peerRoute]{} + routes := [256]*atomic.Pointer[RemotePeer]{} for i := range routes { - routes[i] = &atomic.Pointer[peerRoute]{} - routes[i].Store(&peerRoute{}) + routes[i] = &atomic.Pointer[RemotePeer]{} + routes[i].Store(&RemotePeer{}) } local, remote, relayLocal, relayRemote := testConnWriter_getTestRoutes() @@ -255,8 +256,6 @@ func TestConnReader_handleControlPacket_duplicate(t *testing.T) { } } -/* - // Testing that we can receive a data packet. func TestConnReader_handleDataPacket(t *testing.T) { h := newConnReadeTestHarness() @@ -285,7 +284,7 @@ func TestConnReader_handleDataPacket_routeDown(t *testing.T) { rand.Read(pkt) h.WRemote.SendDataPacket(pkt, h.Remote) - route := h.R.routes[2].Load() + route := h.R.peers[2].Load() route.Up = false h.R.handleNextPacket() @@ -294,9 +293,61 @@ func TestConnReader_handleDataPacket_routeDown(t *testing.T) { t.Fatal(h.IFace.Written) } } -*/ + // Testing that a duplicate data packet is ignored. +func TestConnReader_handleDataPacket_duplicate(t *testing.T) { + h := newConnReadeTestHarness() -// Testing that we send a relayed data packet. + pkt := make([]byte, 123) -// Testing that a relayed data packet is ignored if destination isn't up. + h.WRemote.SendDataPacket(pkt, h.Remote) + *h.Remote.Counter = *h.Remote.Counter - 1 + h.WRemote.SendDataPacket(pkt, h.Remote) + + h.R.handleNextPacket() + h.R.handleNextPacket() + + if len(h.IFace.Written) != 1 { + t.Fatal(h.IFace.Written) + } + + if !bytes.Equal(pkt, h.IFace.Written[0]) { + t.Fatal(h.IFace.Written) + } +} + +// Testing that we can relay a data packet. +func TestConnReader_handleDataPacket_relay(t *testing.T) { + h := newConnReadeTestHarness() + + pkt := make([]byte, 1024) + rand.Read(pkt) + + h.RelayRemote.IP = 3 + h.WRemote.RelayDataPacket(pkt, h.RelayRemote, h.Remote) + h.R.handleNextPacket() + + if len(h.Sender.Sent) != 1 { + t.Fatal(h.Sender.Sent) + } + +} + +// Testing that we drop a relayed packet if destination is down. +func TestConnReader_handleDataPacket_relayDown(t *testing.T) { + h := newConnReadeTestHarness() + + pkt := make([]byte, 1024) + rand.Read(pkt) + + h.RelayRemote.IP = 3 + relay := h.R.peers[3].Load() + relay.Up = false + + h.WRemote.RelayDataPacket(pkt, h.RelayRemote, h.Remote) + h.R.handleNextPacket() + + if len(h.Sender.Sent) != 0 { + t.Fatal(h.Sender.Sent) + } +} diff --git a/peer/connwriter.go b/peer/connwriter.go index 928b2a0..7daa567 100644 --- a/peer/connwriter.go +++ b/peer/connwriter.go @@ -37,29 +37,29 @@ func newConnWriter(conn udpWriter, localIP byte) *connWriter { } // Not safe for concurrent use. Should only be called by supervisor. -func (w *connWriter) SendControlPacket(pkt marshaller, route *peerRoute) { - enc := encryptControlPacket(w.localIP, route, pkt, w.cBuf1, w.cBuf2) - w.writeTo(enc, route.RemoteAddr) +func (w *connWriter) SendControlPacket(pkt marshaller, peer *RemotePeer) { + enc := encryptControlPacket(w.localIP, peer, pkt, w.cBuf1, w.cBuf2) + w.writeTo(enc, peer.DirectAddr) } -// Relay control packet. Route must not be nil. -func (w *connWriter) RelayControlPacket(pkt marshaller, route, relay *peerRoute) { - enc := encryptControlPacket(w.localIP, route, pkt, w.cBuf1, w.cBuf2) - enc = encryptDataPacket(w.localIP, route.IP, relay, enc, w.cBuf1) - w.writeTo(enc, relay.RemoteAddr) +// Relay control packet. Peer must not be nil. +func (w *connWriter) RelayControlPacket(pkt marshaller, peer, relay *RemotePeer) { + enc := encryptControlPacket(w.localIP, peer, pkt, w.cBuf1, w.cBuf2) + enc = encryptDataPacket(w.localIP, peer.IP, relay, enc, w.cBuf1) + w.writeTo(enc, relay.DirectAddr) } // Not safe for concurrent use. Should only be called by ifReader. -func (w *connWriter) SendDataPacket(pkt []byte, route *peerRoute) { - enc := encryptDataPacket(w.localIP, route.IP, route, pkt, w.dBuf1) - w.writeTo(enc, route.RemoteAddr) +func (w *connWriter) SendDataPacket(pkt []byte, peer *RemotePeer) { + enc := encryptDataPacket(w.localIP, peer.IP, peer, pkt, w.dBuf1) + w.writeTo(enc, peer.DirectAddr) } -// Relay a data packet. Route must not be nil. -func (w *connWriter) RelayDataPacket(pkt []byte, route, relay *peerRoute) { - enc := encryptDataPacket(w.localIP, route.IP, route, pkt, w.dBuf1) - enc = encryptDataPacket(w.localIP, route.IP, relay, enc, w.dBuf2) - w.writeTo(enc, relay.RemoteAddr) +// Relay a data packet. Peer must not be nil. +func (w *connWriter) RelayDataPacket(pkt []byte, peer, relay *RemotePeer) { + enc := encryptDataPacket(w.localIP, peer.IP, peer, pkt, w.dBuf1) + enc = encryptDataPacket(w.localIP, peer.IP, relay, enc, w.dBuf2) + w.writeTo(enc, relay.DirectAddr) } // Safe for concurrent use. Should only be called by connReader. @@ -67,8 +67,8 @@ func (w *connWriter) RelayDataPacket(pkt []byte, route, relay *peerRoute) { // This function will send pkt to the peer directly. This is used when a peer // is acting as a relay and is forwarding already encrypted data for another // peer. -func (w *connWriter) SendEncryptedDataPacket(pkt []byte, route *peerRoute) { - w.writeTo(pkt, route.RemoteAddr) +func (w *connWriter) SendEncryptedDataPacket(pkt []byte, peer *RemotePeer) { + w.writeTo(pkt, peer.DirectAddr) } func (w *connWriter) writeTo(packet []byte, addr netip.AddrPort) { diff --git a/peer/connwriter_test.go b/peer/connwriter_test.go index 14f128e..d8c0365 100644 --- a/peer/connwriter_test.go +++ b/peer/connwriter_test.go @@ -43,46 +43,46 @@ func (p testPacket) Marshal(b []byte) []byte { // ---------------------------------------------------------------------------- -func testConnWriter_getTestRoutes() (local, remote, relayLocal, relayRemote *peerRoute) { +func testConnWriter_getTestRoutes() (local, remote, relayLocal, relayRemote *RemotePeer) { localKeys := generateKeys() remoteKeys := generateKeys() - local = newPeerRoute(2) + local = NewRemotePeer(2) local.Up = true local.Relay = false local.PubSignKey = remoteKeys.PubSignKey local.ControlCipher = newControlCipher(localKeys.PrivKey, remoteKeys.PubKey) local.DataCipher = newDataCipher() - local.RemoteAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 2}), 100) + local.DirectAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 2}), 100) - remote = newPeerRoute(1) + remote = NewRemotePeer(1) remote.Up = true remote.Relay = false remote.PubSignKey = localKeys.PubSignKey remote.ControlCipher = newControlCipher(remoteKeys.PrivKey, localKeys.PubKey) remote.DataCipher = local.DataCipher - remote.RemoteAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 1}), 100) + remote.DirectAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 1}), 100) rLocalKeys := generateKeys() rRemoteKeys := generateKeys() - relayLocal = newPeerRoute(3) + relayLocal = NewRemotePeer(3) relayLocal.Up = true relayLocal.Relay = true relayLocal.Direct = true relayLocal.PubSignKey = rRemoteKeys.PubSignKey relayLocal.ControlCipher = newControlCipher(rLocalKeys.PrivKey, rRemoteKeys.PubKey) relayLocal.DataCipher = newDataCipher() - relayLocal.RemoteAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 3}), 100) + relayLocal.DirectAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 3}), 100) - relayRemote = newPeerRoute(1) + relayRemote = NewRemotePeer(1) relayRemote.Up = true relayRemote.Relay = false relayRemote.Direct = true relayRemote.PubSignKey = rLocalKeys.PubSignKey relayRemote.ControlCipher = newControlCipher(rRemoteKeys.PrivKey, rLocalKeys.PubKey) relayRemote.DataCipher = relayLocal.DataCipher - relayRemote.RemoteAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 1}), 100) + relayRemote.DirectAddr = netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 1}), 100) return } @@ -104,7 +104,7 @@ func TestConnWriter_SendControlPacket_direct(t *testing.T) { t.Fatal(out) } - if out[0].Addr != route.RemoteAddr { + if out[0].Addr != route.DirectAddr { t.Fatal(out[0]) } @@ -132,7 +132,7 @@ func TestConnWriter_RelayControlPacket_relay(t *testing.T) { t.Fatal(out) } - if out[0].Addr != relay.RemoteAddr { + if out[0].Addr != relay.DirectAddr { t.Fatal(out[0]) } @@ -167,7 +167,7 @@ func TestConnWriter_SendDataPacket_direct(t *testing.T) { t.Fatal(out) } - if out[0].Addr != route.RemoteAddr { + if out[0].Addr != route.DirectAddr { t.Fatal(out[0]) } @@ -196,7 +196,7 @@ func TestConnWriter_RelayDataPacket_relay(t *testing.T) { t.Fatal(out) } - if out[0].Addr != relay.RemoteAddr { + if out[0].Addr != relay.DirectAddr { t.Fatal(out[0]) } @@ -230,7 +230,7 @@ func TestConnWriter_SendEncryptedDataPacket(t *testing.T) { t.Fatal(out) } - if out[0].Addr != route.RemoteAddr { + if out[0].Addr != route.DirectAddr { t.Fatal(out[0]) } diff --git a/peer/crypto.go b/peer/crypto.go index 3bc970f..f9c61db 100644 --- a/peer/crypto.go +++ b/peer/crypto.go @@ -33,40 +33,40 @@ func generateKeys() cryptoKeys { // ---------------------------------------------------------------------------- -// Route must have a ControlCipher. +// Peer must have a ControlCipher. func encryptControlPacket( localIP byte, - route *peerRoute, + peer *RemotePeer, pkt marshaller, tmp []byte, out []byte, ) []byte { h := header{ StreamID: controlStreamID, - Counter: atomic.AddUint64(route.Counter, 1), + Counter: atomic.AddUint64(peer.Counter, 1), SourceIP: localIP, - DestIP: route.IP, + DestIP: peer.IP, } tmp = pkt.Marshal(tmp) - return route.ControlCipher.Encrypt(h, tmp, out) + return peer.ControlCipher.Encrypt(h, tmp, out) } -// Returns a controlMsg[PacketType]. Route must have ControlCipher. +// Returns a controlMsg[PacketType]. Peer must have a non-nil ControlCipher. // // This function also drops packets with duplicate sequence numbers. func decryptControlPacket( - route *peerRoute, + peer *RemotePeer, fromAddr netip.AddrPort, h header, encrypted []byte, tmp []byte, ) (any, error) { - out, ok := route.ControlCipher.Decrypt(encrypted, tmp) + out, ok := peer.ControlCipher.Decrypt(encrypted, tmp) if !ok { return nil, errDecryptionFailed } - if route.DupCheck.IsDup(h.Counter) { + if peer.DupCheck.IsDup(h.Counter) { return nil, errDuplicateSeqNum } @@ -83,31 +83,32 @@ func decryptControlPacket( func encryptDataPacket( localIP byte, destIP byte, - route *peerRoute, + peer *RemotePeer, data []byte, out []byte, ) []byte { h := header{ StreamID: dataStreamID, - Counter: atomic.AddUint64(route.Counter, 1), + Counter: atomic.AddUint64(peer.Counter, 1), SourceIP: localIP, DestIP: destIP, } - return route.DataCipher.Encrypt(h, data, out) + return peer.DataCipher.Encrypt(h, data, out) } +// Decrypts and de-dups incoming data packets. func decryptDataPacket( - route *peerRoute, + peer *RemotePeer, h header, encrypted []byte, out []byte, ) ([]byte, error) { - dec, ok := route.DataCipher.Decrypt(encrypted, out) + dec, ok := peer.DataCipher.Decrypt(encrypted, out) if !ok { return nil, errDecryptionFailed } - if route.DupCheck.IsDup(h.Counter) { + if peer.DupCheck.IsDup(h.Counter) { return nil, errDuplicateSeqNum } diff --git a/peer/crypto_test.go b/peer/crypto_test.go index 29ee377..c93b87f 100644 --- a/peer/crypto_test.go +++ b/peer/crypto_test.go @@ -9,16 +9,16 @@ import ( "testing" ) -func newRoutePairForTesting() (*peerRoute, *peerRoute) { +func newRoutePairForTesting() (*RemotePeer, *RemotePeer) { keys1 := generateKeys() keys2 := generateKeys() - r1 := newPeerRoute(1) + r1 := NewRemotePeer(1) r1.PubSignKey = keys1.PubSignKey r1.ControlCipher = newControlCipher(keys1.PrivKey, keys2.PubKey) r1.DataCipher = newDataCipher() - r2 := newPeerRoute(2) + r2 := NewRemotePeer(2) r2.PubSignKey = keys2.PubSignKey r2.ControlCipher = newControlCipher(keys2.PrivKey, keys1.PubKey) r2.DataCipher = r1.DataCipher @@ -40,10 +40,7 @@ func TestDecryptControlPacket(t *testing.T) { } enc := encryptControlPacket(r1.IP, r2, in, tmp, out) - h, ok := parseHeader(enc) - if !ok { - t.Fatal(h, ok) - } + h := parseHeader(enc) iMsg, err := decryptControlPacket(r2, netip.AddrPort{}, h, enc, tmp) if err != nil { @@ -74,10 +71,7 @@ func TestDecryptControlPacket_decryptionFailed(t *testing.T) { } enc := encryptControlPacket(r1.IP, r2, in, tmp, out) - h, ok := parseHeader(enc) - if !ok { - t.Fatal(h, ok) - } + h := parseHeader(enc) for i := range enc { x := bytes.Clone(enc) @@ -103,10 +97,7 @@ func TestDecryptControlPacket_duplicate(t *testing.T) { } enc := encryptControlPacket(r1.IP, r2, in, tmp, out) - h, ok := parseHeader(enc) - if !ok { - t.Fatal(h, ok) - } + h := parseHeader(enc) if _, err := decryptControlPacket(r2, netip.AddrPort{}, h, enc, tmp); err != nil { t.Fatal(err) @@ -128,10 +119,7 @@ func TestDecryptControlPacket_invalidPacket(t *testing.T) { in := testPacket("hello!") enc := encryptControlPacket(r1.IP, r2, in, tmp, out) - h, ok := parseHeader(enc) - if !ok { - t.Fatal(h, ok) - } + h := parseHeader(enc) _, err := decryptControlPacket(r2, netip.AddrPort{}, h, enc, tmp) if !errors.Is(err, errUnknownPacketType) { @@ -149,10 +137,7 @@ func TestDecryptDataPacket(t *testing.T) { rand.Read(data) enc := encryptDataPacket(r1.IP, r2.IP, r2, data, out) - h, ok := parseHeader(enc) - if !ok { - t.Fatal(h, ok) - } + h := parseHeader(enc) out, err := decryptDataPacket(r1, h, bytes.Clone(enc), out) if err != nil { @@ -174,10 +159,7 @@ func TestDecryptDataPacket_incorrectCipher(t *testing.T) { rand.Read(data) enc := encryptDataPacket(r1.IP, r2.IP, r2, data, bytes.Clone(out)) - h, ok := parseHeader(enc) - if !ok { - t.Fatal(h, ok) - } + h := parseHeader(enc) r1.DataCipher = newDataCipher() _, err := decryptDataPacket(r1, h, enc, bytes.Clone(out)) @@ -196,10 +178,7 @@ func TestDecryptDataPacket_duplicate(t *testing.T) { rand.Read(data) enc := encryptDataPacket(r1.IP, r2.IP, r2, data, bytes.Clone(out)) - h, ok := parseHeader(enc) - if !ok { - t.Fatal(h, ok) - } + h := parseHeader(enc) _, err := decryptDataPacket(r1, h, enc, bytes.Clone(out)) if err != nil { diff --git a/peer/header.go b/peer/header.go index 08698dd..fae3780 100644 --- a/peer/header.go +++ b/peer/header.go @@ -20,16 +20,13 @@ type header struct { Counter uint64 // Init with time.Now().Unix << 30 to ensure monotonic. } -func parseHeader(b []byte) (h header, ok bool) { - if len(b) < headerSize { - return - } +func parseHeader(b []byte) (h header) { h.Version = b[0] h.StreamID = b[1] h.SourceIP = b[2] h.DestIP = b[3] h.Counter = *(*uint64)(unsafe.Pointer(&b[4])) - return h, true + return h } func (h *header) Parse(b []byte) { diff --git a/peer/ifreader.go b/peer/ifreader.go index 61627a2..79ff441 100644 --- a/peer/ifreader.go +++ b/peer/ifreader.go @@ -8,20 +8,20 @@ import ( type ifReader struct { iface io.Reader - routes [256]*atomic.Pointer[peerRoute] - relay *atomic.Pointer[peerRoute] + peers [256]*atomic.Pointer[RemotePeer] + relay *atomic.Pointer[RemotePeer] sender dataPacketSender } func newIFReader( iface io.Reader, - routes [256]*atomic.Pointer[peerRoute], - relay *atomic.Pointer[peerRoute], + peers [256]*atomic.Pointer[RemotePeer], + relay *atomic.Pointer[RemotePeer], sender dataPacketSender, ) *ifReader { return &ifReader{ iface: iface, - routes: routes, + peers: peers, relay: relay, sender: sender, } @@ -43,20 +43,20 @@ func (r *ifReader) Run() { } func (r *ifReader) sendPacket(pkt []byte, remoteIP byte) { - route := r.routes[remoteIP].Load() - if !route.Up { - log.Printf("Route not connected: %d", remoteIP) + peer := r.peers[remoteIP].Load() + if !peer.Up { + log.Printf("Peer not connected: %d", remoteIP) return } // Direct path => early return. - if route.Direct { - r.sender.SendDataPacket(pkt, route) + if peer.Direct { + r.sender.SendDataPacket(pkt, peer) return } if relay := r.relay.Load(); relay != nil && relay.Up { - r.sender.RelayDataPacket(pkt, route, relay) + r.sender.RelayDataPacket(pkt, peer, relay) } } diff --git a/peer/ifreader_test.go b/peer/ifreader_test.go index c5efb30..e8c5683 100644 --- a/peer/ifreader_test.go +++ b/peer/ifreader_test.go @@ -10,7 +10,7 @@ import ( // Test that we parse IPv4 packets correctly. func TestIFReader_parsePacket_ipv4(t *testing.T) { - r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil) + r := newIFReader(nil, [256]*atomic.Pointer[RemotePeer]{}, nil, nil) pkt := make([]byte, 1234) pkt[0] = 4 << 4 @@ -23,7 +23,7 @@ func TestIFReader_parsePacket_ipv4(t *testing.T) { // Test that we parse IPv6 packets correctly. func TestIFReader_parsePacket_ipv6(t *testing.T) { - r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil) + r := newIFReader(nil, [256]*atomic.Pointer[RemotePeer]{}, nil, nil) pkt := make([]byte, 1234) pkt[0] = 6 << 4 @@ -36,7 +36,7 @@ func TestIFReader_parsePacket_ipv6(t *testing.T) { // Test that empty packets work as expected. func TestIFReader_parsePacket_emptyPacket(t *testing.T) { - r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil) + r := newIFReader(nil, [256]*atomic.Pointer[RemotePeer]{}, nil, nil) pkt := make([]byte, 0) if ip, ok := r.parsePacket(pkt); ok { @@ -46,7 +46,7 @@ func TestIFReader_parsePacket_emptyPacket(t *testing.T) { // Test that invalid IP versions fail. func TestIFReader_parsePacket_invalidIPVersion(t *testing.T) { - r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil) + r := newIFReader(nil, [256]*atomic.Pointer[RemotePeer]{}, nil, nil) for i := byte(1); i < 16; i++ { if i == 4 || i == 6 { @@ -63,7 +63,7 @@ func TestIFReader_parsePacket_invalidIPVersion(t *testing.T) { // Test that short IPv4 packets fail. func TestIFReader_parsePacket_shortIPv4(t *testing.T) { - r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil) + r := newIFReader(nil, [256]*atomic.Pointer[RemotePeer]{}, nil, nil) pkt := make([]byte, 19) pkt[0] = 4 << 4 @@ -75,7 +75,7 @@ func TestIFReader_parsePacket_shortIPv4(t *testing.T) { // Test that short IPv6 packets fail. func TestIFReader_parsePacket_shortIPv6(t *testing.T) { - r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil) + r := newIFReader(nil, [256]*atomic.Pointer[RemotePeer]{}, nil, nil) pkt := make([]byte, 39) pkt[0] = 6 << 4 @@ -88,7 +88,7 @@ func TestIFReader_parsePacket_shortIPv6(t *testing.T) { // Test that we can read a packet. func TestIFReader_readNextpacket(t *testing.T) { in, out := net.Pipe() - r := newIFReader(out, [256]*atomic.Pointer[peerRoute]{}, nil, nil) + r := newIFReader(out, [256]*atomic.Pointer[RemotePeer]{}, nil, nil) defer in.Close() defer out.Close() @@ -105,22 +105,22 @@ func TestIFReader_readNextpacket(t *testing.T) { type sentPacket struct { Relayed bool Packet []byte - Route peerRoute - Relay peerRoute + Route RemotePeer + Relay RemotePeer } type sendPacketTestHarness struct { Packets []sentPacket } -func (h *sendPacketTestHarness) SendDataPacket(pkt []byte, route *peerRoute) { +func (h *sendPacketTestHarness) SendDataPacket(pkt []byte, route *RemotePeer) { h.Packets = append(h.Packets, sentPacket{ Packet: bytes.Clone(pkt), Route: *route, }) } -func (h *sendPacketTestHarness) RelayDataPacket(pkt []byte, route, relay *peerRoute) { +func (h *sendPacketTestHarness) RelayDataPacket(pkt []byte, route, relay *RemotePeer) { h.Packets = append(h.Packets, sentPacket{ Relayed: true, Packet: bytes.Clone(pkt), @@ -132,12 +132,12 @@ func (h *sendPacketTestHarness) RelayDataPacket(pkt []byte, route, relay *peerRo func newIFReaderForSendPacketTesting() (*ifReader, *sendPacketTestHarness) { h := &sendPacketTestHarness{} - routes := [256]*atomic.Pointer[peerRoute]{} + routes := [256]*atomic.Pointer[RemotePeer]{} for i := range routes { - routes[i] = &atomic.Pointer[peerRoute]{} - routes[i].Store(&peerRoute{}) + routes[i] = &atomic.Pointer[RemotePeer]{} + routes[i].Store(&RemotePeer{}) } - relay := &atomic.Pointer[peerRoute]{} + relay := &atomic.Pointer[RemotePeer]{} r := newIFReader(nil, routes, relay, h) return r, h } @@ -146,7 +146,7 @@ func newIFReaderForSendPacketTesting() (*ifReader, *sendPacketTestHarness) { func TestIFReader_sendPacket_direct(t *testing.T) { r, h := newIFReaderForSendPacketTesting() - route := r.routes[2].Load() + route := r.peers[2].Load() route.Up = true route.Direct = true @@ -172,7 +172,7 @@ func TestIFReader_sendPacket_direct(t *testing.T) { func TestIFReader_sendPacket_directNotUp(t *testing.T) { r, h := newIFReaderForSendPacketTesting() - route := r.routes[2].Load() + route := r.peers[2].Load() route.Direct = true in := []byte("hello world") @@ -187,11 +187,11 @@ func TestIFReader_sendPacket_directNotUp(t *testing.T) { func TestIFReader_sendPacket_relayed(t *testing.T) { r, h := newIFReaderForSendPacketTesting() - route := r.routes[2].Load() + route := r.peers[2].Load() route.Up = true route.Direct = false - relay := r.routes[3].Load() + relay := r.peers[3].Load() r.relay.Store(relay) relay.Up = true relay.Direct = true @@ -219,7 +219,7 @@ func TestIFReader_sendPacket_relayed(t *testing.T) { func TestIFReader_sendPacket_nilRealy(t *testing.T) { r, h := newIFReaderForSendPacketTesting() - route := r.routes[2].Load() + route := r.peers[2].Load() route.Up = true route.Direct = false diff --git a/peer/interfaces.go b/peer/interfaces.go index 84f9c99..0d826c3 100644 --- a/peer/interfaces.go +++ b/peer/interfaces.go @@ -1,6 +1,9 @@ package peer -import "net/netip" +import ( + "net" + "net/netip" +) type udpReader interface { ReadFromUDPAddrPort(b []byte) (n int, addr netip.AddrPort, err error) @@ -15,14 +18,18 @@ type marshaller interface { } type dataPacketSender interface { - SendDataPacket(pkt []byte, route *peerRoute) - RelayDataPacket(pkt []byte, route, relay *peerRoute) + SendDataPacket(pkt []byte, peer *RemotePeer) + RelayDataPacket(pkt []byte, peer, relay *RemotePeer) } type encryptedPacketSender interface { - SendEncryptedDataPacket(pkt []byte, route *peerRoute) + SendEncryptedDataPacket(pkt []byte, peer *RemotePeer) } type controlMsgHandler interface { HandleControlMsg(pkt any) } + +type mcUDPWriter interface { + WriteToUDP([]byte, *net.UDPAddr) (int, error) +} diff --git a/peer/mcreader.go b/peer/mcreader.go new file mode 100644 index 0000000..7d5c959 --- /dev/null +++ b/peer/mcreader.go @@ -0,0 +1,57 @@ +package peer + +import ( + "log" + "sync/atomic" +) + +type mcReader struct { + conn udpReader + super controlMsgHandler + peers [256]*atomic.Pointer[RemotePeer] + + incoming []byte + buf []byte +} + +func newMCReader( + conn udpReader, + super controlMsgHandler, + peers [256]*atomic.Pointer[RemotePeer], +) *mcReader { + return &mcReader{conn, super, peers, newBuf(), newBuf()} +} + +func (r *mcReader) Run() { + for { + r.handleNextPacket() + } +} + +func (r *mcReader) handleNextPacket() { + incoming := r.incoming[:bufferSize] + n, remoteAddr, err := r.conn.ReadFromUDPAddrPort(incoming) + if err != nil { + log.Fatalf("Failed to read from UDP multicast port: %v", err) + } + incoming = incoming[:n] + + h, ok := headerFromLocalDiscoveryPacket(incoming) + if !ok { + return + } + + peer := r.peers[h.SourceIP].Load() + if peer == nil || peer.PubSignKey == nil { + return + } + + if !verifyLocalDiscoveryPacket(incoming, r.buf, peer.PubSignKey) { + return + } + + r.super.HandleControlMsg(controlMsg[localDiscoveryPacket]{ + SrcIP: h.SourceIP, + SrcAddr: remoteAddr, + }) +} diff --git a/peer/mcwriter.go b/peer/mcwriter.go index db9a76b..a8b55e9 100644 --- a/peer/mcwriter.go +++ b/peer/mcwriter.go @@ -2,19 +2,10 @@ package peer import ( "log" - "net" "golang.org/x/crypto/nacl/sign" ) -// ---------------------------------------------------------------------------- - -type mcUDPWriter interface { - WriteToUDP([]byte, *net.UDPAddr) (int, error) -} - -// ---------------------------------------------------------------------------- - func createLocalDiscoveryPacket(localIP byte, signingKey []byte) []byte { h := header{ SourceIP: localIP, diff --git a/peer/state.go b/peer/state.go index 2ef248b..d6589fe 100644 --- a/peer/state.go +++ b/peer/state.go @@ -5,23 +5,23 @@ import ( "time" ) -type peerRoute struct { - IP byte // VPN IP of peer (last byte). - Up bool // True if data can be sent on the route. - Relay bool // True if the peer is a relay. - Direct bool // True if this is a direct connection. +type RemotePeer struct { + IP byte // VPN IP of peer (last byte). + Up bool // True if data can be sent on the peer. + Relay bool // True if the peer is a relay. + Direct bool // True if this is a direct connection. + DirectAddr netip.AddrPort // Remote address if directly connected. PubSignKey []byte ControlCipher *controlCipher DataCipher *dataCipher - RemoteAddr netip.AddrPort // Remote address if directly connected. Counter *uint64 // For sending to. Atomic access only. DupCheck *dupCheck // For receiving from. Not safe for concurrent use. } -func newPeerRoute(ip byte) *peerRoute { +func NewRemotePeer(ip byte) *RemotePeer { counter := uint64(time.Now().Unix()<<30 + 1) - return &peerRoute{ + return &RemotePeer{ IP: ip, Counter: &counter, DupCheck: newDupCheck(0),