app/vppn
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 9 Wiki Activity

refactor-for-testability #3

Merged
johnnylee merged 26 commits from refactor-for-testability into main 2025-03-01 20:02:27 +00:00
Conversation 0 Commits 26 Files Changed 68 +250 -177
13 changed files with 250 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 8c618616dd - Show all commits

44
peer/connreader.go
View File

@@ -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
}
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)
}

77
peer/connreader_test.go
View File

@@ -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)
}
}

36
peer/connwriter.go
View File

@@ -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) {

28
peer/connwriter_test.go
View File

@@ -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])
}

31
peer/crypto.go
View File

@@ -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
}

41
peer/crypto_test.go
View File

@@ -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 {

7
peer/header.go
View File

@@ -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) {

22
peer/ifreader.go
View File

@@ -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)
}
}

40
peer/ifreader_test.go
View File

@@ -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

15
peer/interfaces.go
View File

@@ -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)
}

57
peer/mcreader.go Normal file
View File

@@ -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,
})
}

9
peer/mcwriter.go
View File

@@ -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,

10
peer/state.go
View File

@@ -5,23 +5,23 @@ import (
"time"
)
type peerRoute struct {
type RemotePeer struct {
IP byte // VPN IP of peer (last byte).
Up bool // True if data can be sent on the route.
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),