app/vppn
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 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 -3597
44 changed files with 0 additions and 3597 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 589aa08866 - Show all commits

16
node/README.md
View File

@ -1,16 +0,0 @@
# VPPN Peer Code
## Refactoring for Testability
* [x] connWriter
* [x] mcWriter
* [x] ifWriter
* [ ] ifReader (testing)
* [ ] connReader
* [ ] mcReader
* [ ] hubPoller
* [ ] supervisor
## Updates
* [ ] Send timing info w/ syn/ack packets

71
node/addrdiscovery.go
View File

@ -1,71 +0,0 @@
package node
import (
"log"
"net/netip"
"runtime/debug"
"sort"
"time"
)
type pubAddrStore struct {
lastSeen map[netip.AddrPort]time.Time
addrList []netip.AddrPort
}
func newPubAddrStore() *pubAddrStore {
return &pubAddrStore{
lastSeen: map[netip.AddrPort]time.Time{},
addrList: make([]netip.AddrPort, 0, 32),
}
}
func (store *pubAddrStore) Store(add netip.AddrPort) {
if localPub {
log.Printf("OOPS: Local pub but storage attempt: %s", debug.Stack())
return
}
if !add.IsValid() {
return
}
if _, exists := store.lastSeen[add]; !exists {
store.addrList = append(store.addrList, add)
}
store.lastSeen[add] = time.Now()
store.sort()
}
func (store *pubAddrStore) Get() (addrs [8]netip.AddrPort) {
if localPub {
addrs[0] = localAddr
return
}
copy(addrs[:], store.addrList)
return
}
func (store *pubAddrStore) Clean() {
if localPub {
return
}
for ip, lastSeen := range store.lastSeen {
if time.Since(lastSeen) > timeoutInterval {
delete(store.lastSeen, ip)
}
}
store.addrList = store.addrList[:0]
for ip := range store.lastSeen {
store.addrList = append(store.addrList, ip)
}
store.sort()
}
func (store *pubAddrStore) sort() {
sort.Slice(store.addrList, func(i, j int) bool {
return store.lastSeen[store.addrList[j]].Before(store.lastSeen[store.addrList[i]])
})
}

29
node/addrdiscovery_test.go
View File

@ -1,29 +0,0 @@
package node
import (
"net/netip"
"testing"
"time"
)
func TestPubAddrStore(t *testing.T) {
s := newPubAddrStore()
l := []netip.AddrPort{
netip.AddrPortFrom(netip.AddrFrom4([4]byte{0, 1, 2, 3}), 20),
netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 2, 3}), 21),
netip.AddrPortFrom(netip.AddrFrom4([4]byte{2, 1, 2, 3}), 22),
}
for i := range l {
s.Store(l[i])
time.Sleep(time.Millisecond)
}
s.Clean()
l2 := s.Get()
if l2[0] != l[2] || l2[1] != l[1] || l2[2] != l[0] {
t.Fatal(l, l2)
}
}

21
node/bitset.go
View File

@ -1,21 +0,0 @@
package node
const bitSetSize = 512 // Multiple of 64.
type bitSet [bitSetSize / 64]uint64
func (bs *bitSet) Set(i int) {
bs[i/64] |= 1 << (i % 64)
}
func (bs *bitSet) Clear(i int) {
bs[i/64] &= ^(1 << (i % 64))
}
func (bs *bitSet) ClearAll() {
clear(bs[:])
}
func (bs *bitSet) Get(i int) bool {
return bs[i/64]&(1<<(i%64)) != 0
}

48
node/bitset_test.go
View File

@ -1,48 +0,0 @@
package node
import (
"math/rand"
"testing"
)
func TestBitSet(t *testing.T) {
state := make([]bool, bitSetSize)
for i := range state {
state[i] = rand.Float32() > 0.5
}
bs := bitSet{}
for i := range state {
if state[i] {
bs.Set(i)
}
}
for i := range state {
if bs.Get(i) != state[i] {
t.Fatal(i, state[i], bs.Get(i))
}
}
for i := range state {
if rand.Float32() > 0.5 {
state[i] = false
bs.Clear(i)
}
}
for i := range state {
if bs.Get(i) != state[i] {
t.Fatal(i, state[i], bs.Get(i))
}
}
bs.ClearAll()
for i := range state {
if bs.Get(i) {
t.Fatal(i, bs.Get(i))
}
}
}

26
node/cipher-control.go
View File

@ -1,26 +0,0 @@
package node
import "golang.org/x/crypto/nacl/box"
type controlCipher struct {
sharedKey [32]byte
}
func newControlCipher(privKey, pubKey []byte) *controlCipher {
shared := [32]byte{}
box.Precompute(&shared, (*[32]byte)(pubKey), (*[32]byte)(privKey))
return &controlCipher{shared}
}
func (cc *controlCipher) Encrypt(h header, data, out []byte) []byte {
const s = controlHeaderSize
out = out[:s+controlCipherOverhead+len(data)]
h.Marshal(out[:s])
box.SealAfterPrecomputation(out[s:s], data, (*[24]byte)(out[:s]), &cc.sharedKey)
return out
}
func (cc *controlCipher) Decrypt(encrypted, out []byte) (data []byte, ok bool) {
const s = controlHeaderSize
return box.OpenAfterPrecomputation(out[:0], encrypted[s:], (*[24]byte)(encrypted[:s]), &cc.sharedKey)
}

122
node/cipher-control_test.go
View File

@ -1,122 +0,0 @@
package node
import (
"bytes"
"crypto/rand"
"reflect"
"testing"
"golang.org/x/crypto/nacl/box"
)
func newControlCipherForTesting() (c1, c2 *controlCipher) {
pubKey1, privKey1, err := box.GenerateKey(rand.Reader)
if err != nil {
panic(err)
}
pubKey2, privKey2, err := box.GenerateKey(rand.Reader)
if err != nil {
panic(err)
}
return newControlCipher(privKey1[:], pubKey2[:]),
newControlCipher(privKey2[:], pubKey1[:])
}
func TestControlCipher(t *testing.T) {
c1, c2 := newControlCipherForTesting()
maxSizePlaintext := make([]byte, bufferSize-controlHeaderSize-controlCipherOverhead)
rand.Read(maxSizePlaintext)
testCases := [][]byte{
make([]byte, 0),
{1},
{255},
{1, 2, 3, 4, 5},
[]byte("Hello world"),
maxSizePlaintext,
}
for _, plaintext := range testCases {
h1 := header{
StreamID: controlStreamID,
Counter: 235153,
SourceIP: 4,
DestIP: 88,
}
encrypted := make([]byte, bufferSize)
encrypted = c1.Encrypt(h1, plaintext, encrypted)
h2 := header{}
h2.Parse(encrypted)
if !reflect.DeepEqual(h1, h2) {
t.Fatal(h1, h2)
}
decrypted, ok := c2.Decrypt(encrypted, make([]byte, bufferSize))
if !ok {
t.Fatal(ok)
}
if !bytes.Equal(decrypted, plaintext) {
t.Fatal("not equal")
}
}
}
func TestControlCipher_ShortCiphertext(t *testing.T) {
c1, _ := newControlCipherForTesting()
shortText := make([]byte, controlHeaderSize+controlCipherOverhead-1)
rand.Read(shortText)
_, ok := c1.Decrypt(shortText, make([]byte, bufferSize))
if ok {
t.Fatal(ok)
}
}
func BenchmarkControlCipher_Encrypt(b *testing.B) {
c1, _ := newControlCipherForTesting()
h1 := header{
Counter: 235153,
SourceIP: 4,
DestIP: 88,
}
plaintext := make([]byte, bufferSize-controlHeaderSize-controlCipherOverhead)
rand.Read(plaintext)
encrypted := make([]byte, bufferSize)
b.ResetTimer()
for i := 0; i < b.N; i++ {
encrypted = c1.Encrypt(h1, plaintext, encrypted)
}
}
func BenchmarkControlCipher_Decrypt(b *testing.B) {
c1, c2 := newControlCipherForTesting()
h1 := header{
Counter: 235153,
SourceIP: 4,
DestIP: 88,
}
plaintext := make([]byte, bufferSize-controlHeaderSize-controlCipherOverhead)
rand.Read(plaintext)
encrypted := make([]byte, bufferSize)
encrypted = c1.Encrypt(h1, plaintext, encrypted)
decrypted := make([]byte, bufferSize)
b.ResetTimer()
for i := 0; i < b.N; i++ {
decrypted, _ = c2.Decrypt(encrypted, decrypted)
}
}

60
node/cipher-data.go
View File

@ -1,60 +0,0 @@
package node
import (
"crypto/aes"
"crypto/cipher"
"crypto/rand"
)
type dataCipher struct {
key [32]byte
aead cipher.AEAD
}
func newDataCipher() *dataCipher {
key := [32]byte{}
if _, err := rand.Read(key[:]); err != nil {
panic(err)
}
return newDataCipherFromKey(key)
}
func newDataCipherFromKey(key [32]byte) *dataCipher {
block, err := aes.NewCipher(key[:])
if err != nil {
panic(err)
}
aead, err := cipher.NewGCM(block)
if err != nil {
panic(err)
}
return &dataCipher{key: key, aead: aead}
}
func (sc *dataCipher) Key() [32]byte {
return sc.key
}
func (sc *dataCipher) Encrypt(h header, data, out []byte) []byte {
const s = dataHeaderSize
out = out[:s+dataCipherOverhead+len(data)]
h.Marshal(out[:s])
sc.aead.Seal(out[s:s], out[:s], data, nil)
return out
}
func (sc *dataCipher) Decrypt(encrypted, out []byte) (data []byte, ok bool) {
const s = dataHeaderSize
if len(encrypted) < s+dataCipherOverhead {
ok = false
return
}
var err error
data, err = sc.aead.Open(out[:0], encrypted[:s], encrypted[s:], nil)
ok = err == nil
return
}

141
node/cipher-data_test.go
View File

@ -1,141 +0,0 @@
package node
import (
"bytes"
"crypto/rand"
mrand "math/rand/v2"
"reflect"
"testing"
)
func TestDataCipher(t *testing.T) {
maxSizePlaintext := make([]byte, bufferSize-dataHeaderSize-dataCipherOverhead)
rand.Read(maxSizePlaintext)
testCases := [][]byte{
make([]byte, 0),
{1},
{255},
{1, 2, 3, 4, 5},
[]byte("Hello world"),
maxSizePlaintext,
}
for _, plaintext := range testCases {
h1 := header{
StreamID: dataStreamID,
Counter: 235153,
SourceIP: 4,
DestIP: 88,
}
encrypted := make([]byte, bufferSize)
dc1 := newDataCipher()
encrypted = dc1.Encrypt(h1, plaintext, encrypted)
h2 := header{}
h2.Parse(encrypted)
dc2 := newDataCipherFromKey(dc1.Key())
decrypted, ok := dc2.Decrypt(encrypted, make([]byte, bufferSize-dataHeaderSize))
if !ok {
t.Fatal(ok)
}
if !bytes.Equal(plaintext, decrypted) {
t.Fatal("not equal")
}
if !reflect.DeepEqual(h1, h2) {
t.Fatalf("%v != %v", h1, h2)
}
}
}
func TestDataCipher_ModifyCiphertext(t *testing.T) {
maxSizePlaintext := make([]byte, bufferSize-dataHeaderSize-dataCipherOverhead)
rand.Read(maxSizePlaintext)
testCases := [][]byte{
make([]byte, 0),
{1},
{255},
{1, 2, 3, 4, 5},
[]byte("Hello world"),
maxSizePlaintext,
}
for _, plaintext := range testCases {
h1 := header{
Counter: 235153,
SourceIP: 4,
DestIP: 88,
}
encrypted := make([]byte, bufferSize)
dc1 := newDataCipher()
encrypted = dc1.Encrypt(h1, plaintext, encrypted)
encrypted[mrand.IntN(len(encrypted))]++
dc2 := newDataCipherFromKey(dc1.Key())
_, ok := dc2.Decrypt(encrypted, make([]byte, bufferSize-dataHeaderSize))
if ok {
t.Fatal(ok)
}
}
}
func TestDataCipher_ShortCiphertext(t *testing.T) {
dc1 := newDataCipher()
shortText := make([]byte, dataHeaderSize+dataCipherOverhead-1)
rand.Read(shortText)
_, ok := dc1.Decrypt(shortText, make([]byte, bufferSize))
if ok {
t.Fatal(ok)
}
}
func BenchmarkDataCipher_Encrypt(b *testing.B) {
h1 := header{
Counter: 235153,
SourceIP: 4,
DestIP: 88,
}
plaintext := make([]byte, bufferSize-dataHeaderSize-dataCipherOverhead)
rand.Read(plaintext)
encrypted := make([]byte, bufferSize)
dc1 := newDataCipher()
b.ResetTimer()
for i := 0; i < b.N; i++ {
encrypted = dc1.Encrypt(h1, plaintext, encrypted)
}
}
func BenchmarkDataCipher_Decrypt(b *testing.B) {
h1 := header{
Counter: 235153,
SourceIP: 4,
DestIP: 88,
}
plaintext := make([]byte, bufferSize-dataHeaderSize-dataCipherOverhead)
rand.Read(plaintext)
encrypted := make([]byte, bufferSize)
dc1 := newDataCipher()
encrypted = dc1.Encrypt(h1, plaintext, encrypted)
decrypted := make([]byte, bufferSize)
b.ResetTimer()
for i := 0; i < b.N; i++ {
decrypted, _ = dc1.Decrypt(encrypted, decrypted)
}
}

13
node/cipher-discovery.go
View File

@ -1,13 +0,0 @@
package node
/*
func signData(privKey *[64]byte, h header, data, out []byte) []byte {
out = out[:headerSize]
h.Marshal(out)
return sign.Sign(out, data, privKey)
}
func openData(pubKey *[32]byte, signed, out []byte) (data []byte, ok bool) {
return sign.Open(out[:0], signed[headerSize:], pubKey)
}
*/

11
node/config.go
View File

@ -1,11 +0,0 @@
package node
import "vppn/m"
type localConfig struct {
m.PeerConfig
PubKey []byte
PrivKey []byte
PubSignKey []byte
PrivSignKey []byte
}

3
node/conn.go
View File

@ -1,3 +0,0 @@
package node
// ----------------------------------------------------------------------------

146
node/connwriter.go
View File

@ -1,146 +0,0 @@
package node
import (
"log"
"net/netip"
"sync"
"sync/atomic"
"time"
)
// ----------------------------------------------------------------------------
type peerRoute struct {
IP 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.
PubSignKey []byte
ControlCipher *controlCipher
DataCipher *dataCipher
RemoteAddr netip.AddrPort // Remote address if directly connected.
}
// ----------------------------------------------------------------------------
type udpAddrPortWriter interface {
WriteToUDPAddrPort([]byte, netip.AddrPort) (int, error)
}
type marshaller interface {
Marshal([]byte) []byte
}
// ----------------------------------------------------------------------------
type connWriter struct {
localIP byte
conn udpAddrPortWriter
// For sending control packets.
cBuf1 []byte
cBuf2 []byte
// For sending data packets.
dBuf1 []byte
dBuf2 []byte
counters [256]uint64
// Lock around for sending on UDP Conn.
wLock sync.Mutex
}
func newConnWriter(conn udpAddrPortWriter, localIP byte) *connWriter {
w := &connWriter{
localIP: localIP,
conn: conn,
cBuf1: make([]byte, bufferSize),
cBuf2: make([]byte, bufferSize),
dBuf1: make([]byte, bufferSize),
dBuf2: make([]byte, bufferSize),
}
for i := range w.counters {
w.counters[i] = uint64(time.Now().Unix()<<30 + 1)
}
return w
}
// Not safe for concurrent use. Should only be called by supervisor.
func (w *connWriter) SendControlPacket(pkt marshaller, route *peerRoute) {
buf := w.encryptControlPacket(pkt, route)
w.writeTo(buf, route.RemoteAddr)
}
// Relay control packet. Routes must not be nil.
func (w *connWriter) RelayControlPacket(pkt marshaller, route, relay *peerRoute) {
buf := w.encryptControlPacket(pkt, route)
w.relayPacket(buf, w.cBuf1, route, relay)
}
// Encrypted packet will occupy cBuf2.
func (w *connWriter) encryptControlPacket(pkt marshaller, route *peerRoute) []byte {
buf := pkt.Marshal(w.cBuf1)
h := header{
StreamID: controlStreamID,
Counter: atomic.AddUint64(&w.counters[route.IP], 1),
SourceIP: w.localIP,
DestIP: route.IP,
}
return route.ControlCipher.Encrypt(h, buf, w.cBuf2)
}
// Not safe for concurrent use. Should only be called by ifReader.
func (w *connWriter) SendDataPacket(pkt []byte, route *peerRoute) {
h := header{
StreamID: dataStreamID,
Counter: atomic.AddUint64(&w.counters[route.IP], 1),
SourceIP: w.localIP,
DestIP: route.IP,
}
enc := route.DataCipher.Encrypt(h, pkt, w.dBuf1)
w.writeTo(enc, route.RemoteAddr)
}
// Relay a data packet. Routes must not be nil.
func (w *connWriter) RelayDataPacket(pkt []byte, route, relay *peerRoute) {
h := header{
StreamID: dataStreamID,
Counter: atomic.AddUint64(&w.counters[route.IP], 1),
SourceIP: w.localIP,
DestIP: route.IP,
}
enc := route.DataCipher.Encrypt(h, pkt, w.dBuf1)
w.relayPacket(enc, w.dBuf2, route, relay)
}
// Safe for concurrent use. Should only be called by connReader.
//
// 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) relayPacket(data, buf []byte, route, relay *peerRoute) {
h := header{
StreamID: dataStreamID,
Counter: atomic.AddUint64(&w.counters[relay.IP], 1),
SourceIP: w.localIP,
DestIP: route.IP,
}
enc := relay.DataCipher.Encrypt(h, data, buf)
w.writeTo(enc, relay.RemoteAddr)
}
func (w *connWriter) writeTo(packet []byte, addr netip.AddrPort) {
w.wLock.Lock()
if _, err := w.conn.WriteToUDPAddrPort(packet, addr); err != nil {
log.Printf("Failed to write to UDP port: %v", err)
}
w.wLock.Unlock()
}

248
node/connwriter_test.go
View File

@ -1,248 +0,0 @@
package node
import (
"bytes"
"net/netip"
"testing"
)
// ----------------------------------------------------------------------------
type testUDPPacket struct {
Addr netip.AddrPort
Data []byte
}
type testUDPAddrPortWriter struct {
written []testUDPPacket
}
func (w *testUDPAddrPortWriter) WriteToUDPAddrPort(b []byte, addr netip.AddrPort) (int, error) {
w.written = append(w.written, testUDPPacket{
Addr: addr,
Data: bytes.Clone(b),
})
return len(b), nil
}
func (w *testUDPAddrPortWriter) Written() []testUDPPacket {
out := w.written
w.written = []testUDPPacket{}
return out
}
// ----------------------------------------------------------------------------
type testPacket string
func (p testPacket) Marshal(b []byte) []byte {
b = b[:len(p)]
copy(b, []byte(p))
return b
}
// ----------------------------------------------------------------------------
func testConnWriter_getTestRoutes() (local, remote, relayLocal, relayRemote *peerRoute) {
localKeys := generateKeys()
remoteKeys := generateKeys()
local = &peerRoute{
IP: 2,
Up: true,
Relay: false,
PubSignKey: remoteKeys.PubSignKey,
ControlCipher: newControlCipher(localKeys.PrivKey, remoteKeys.PubKey),
DataCipher: newDataCipher(),
RemoteAddr: netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 2}), 100),
}
remote = &peerRoute{
IP: 1,
Up: true,
Relay: false,
PubSignKey: localKeys.PubSignKey,
ControlCipher: newControlCipher(remoteKeys.PrivKey, localKeys.PubKey),
DataCipher: local.DataCipher,
RemoteAddr: netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 1}), 100),
}
rLocalKeys := generateKeys()
rRemoteKeys := generateKeys()
relayLocal = &peerRoute{
IP: 3,
Up: true,
Relay: true,
Direct: true,
PubSignKey: rRemoteKeys.PubSignKey,
ControlCipher: newControlCipher(rLocalKeys.PrivKey, rRemoteKeys.PubKey),
DataCipher: newDataCipher(),
RemoteAddr: netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 3}), 100),
}
relayRemote = &peerRoute{
IP: 1,
Up: true,
Relay: false,
Direct: true,
PubSignKey: rLocalKeys.PubSignKey,
ControlCipher: newControlCipher(rRemoteKeys.PrivKey, rLocalKeys.PubKey),
DataCipher: relayLocal.DataCipher,
RemoteAddr: netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 1, 1, 1}), 100),
}
return
}
// ----------------------------------------------------------------------------
// Testing if we can send a control packet directly to the remote route.
func TestConnWriter_SendControlPacket_direct(t *testing.T) {
route, rRoute, _, _ := testConnWriter_getTestRoutes()
route.Direct = true
writer := &testUDPAddrPortWriter{}
w := newConnWriter(writer, rRoute.IP)
in := testPacket("hello world!")
w.SendControlPacket(in, route)
out := writer.Written()
if len(out) != 1 {
t.Fatal(out)
}
if out[0].Addr != route.RemoteAddr {
t.Fatal(out[0])
}
dec, ok := rRoute.ControlCipher.Decrypt(out[0].Data, make([]byte, 1024))
if !ok {
t.Fatal(ok)
}
if string(dec) != string(in) {
t.Fatal(dec)
}
}
// Testing if we can relay a packet via an intermediary.
func TestConnWriter_RelayControlPacket_relay(t *testing.T) {
route, rRoute, relay, rRelay := testConnWriter_getTestRoutes()
writer := &testUDPAddrPortWriter{}
w := newConnWriter(writer, rRoute.IP)
in := testPacket("hello world!")
w.RelayControlPacket(in, route, relay)
out := writer.Written()
if len(out) != 1 {
t.Fatal(out)
}
if out[0].Addr != relay.RemoteAddr {
t.Fatal(out[0])
}
dec, ok := rRelay.DataCipher.Decrypt(out[0].Data, make([]byte, 1024))
if !ok {
t.Fatal(ok)
}
dec2, ok := rRoute.ControlCipher.Decrypt(dec, make([]byte, 1024))
if !ok {
t.Fatal(ok)
}
if string(dec2) != string(in) {
t.Fatal(dec2)
}
}
// Testing that we can send a data packet directly to a remote route.
func TestConnWriter_SendDataPacket_direct(t *testing.T) {
route, rRoute, _, _ := testConnWriter_getTestRoutes()
route.Direct = true
writer := &testUDPAddrPortWriter{}
w := newConnWriter(writer, rRoute.IP)
in := []byte("hello world!")
w.SendDataPacket(in, route)
out := writer.Written()
if len(out) != 1 {
t.Fatal(out)
}
if out[0].Addr != route.RemoteAddr {
t.Fatal(out[0])
}
dec, ok := rRoute.DataCipher.Decrypt(out[0].Data, make([]byte, 1024))
if !ok {
t.Fatal(ok)
}
if !bytes.Equal(dec, in) {
t.Fatal(dec)
}
}
// Testing that we can relay a data packet via a relay.
func TestConnWriter_RelayDataPacket_relay(t *testing.T) {
route, rRoute, relay, rRelay := testConnWriter_getTestRoutes()
writer := &testUDPAddrPortWriter{}
w := newConnWriter(writer, rRoute.IP)
in := []byte("Hello world!")
w.RelayDataPacket(in, route, relay)
out := writer.Written()
if len(out) != 1 {
t.Fatal(out)
}
if out[0].Addr != relay.RemoteAddr {
t.Fatal(out[0])
}
dec, ok := rRelay.DataCipher.Decrypt(out[0].Data, make([]byte, 1024))
if !ok {
t.Fatal(ok)
}
dec2, ok := rRoute.DataCipher.Decrypt(dec, make([]byte, 1024))
if !ok {
t.Fatal(ok)
}
if !bytes.Equal(dec2, in) {
t.Fatal(dec2)
}
}
// Testing that we can send an already encrypted packet.
func TestConnWriter_SendEncryptedDataPacket(t *testing.T) {
route, rRoute, _, _ := testConnWriter_getTestRoutes()
writer := &testUDPAddrPortWriter{}
w := newConnWriter(writer, rRoute.IP)
in := []byte("Hello world!")
w.SendEncryptedDataPacket(in, route)
out := writer.Written()
if len(out) != 1 {
t.Fatal(out)
}
if out[0].Addr != route.RemoteAddr {
t.Fatal(out[0])
}
if !bytes.Equal(out[0].Data, in) {
t.Fatal(out[0])
}
}

30
node/crypto.go
View File

@ -1,30 +0,0 @@
package node
import (
"crypto/rand"
"log"
"golang.org/x/crypto/nacl/box"
"golang.org/x/crypto/nacl/sign"
)
type cryptoKeys struct {
PubKey []byte
PrivKey []byte
PubSignKey []byte
PrivSignKey []byte
}
func generateKeys() cryptoKeys {
pubKey, privKey, err := box.GenerateKey(rand.Reader)
if err != nil {
log.Fatalf("Failed to generate encryption keys: %v", err)
}
pubSignKey, privSignKey, err := sign.GenerateKey(rand.Reader)
if err != nil {
log.Fatalf("Failed to generate signing keys: %v", err)
}
return cryptoKeys{pubKey[:], privKey[:], pubSignKey[:], privSignKey[:]}
}

14
node/data-flow.dot
View File

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

76
node/dupcheck.go
View File

@ -1,76 +0,0 @@
package node
type dupCheck struct {
bitSet
head int
tail int
headCounter uint64
tailCounter uint64 // Also next expected counter value.
}
func newDupCheck(headCounter uint64) *dupCheck {
return &dupCheck{
headCounter: headCounter,
tailCounter: headCounter + 1,
tail: 1,
}
}
func (dc *dupCheck) IsDup(counter uint64) bool {
// Before head => it's late, say it's a dup.
if counter < dc.headCounter {
return true
}
// It's within the counter bounds.
if counter < dc.tailCounter {
index := (int(counter-dc.headCounter) + dc.head) % bitSetSize
if dc.Get(index) {
return true
}
dc.Set(index)
return false
}
// It's more than 1 beyond the tail.
delta := counter - dc.tailCounter
// Full clear.
if delta >= bitSetSize-1 {
dc.ClearAll()
dc.Set(0)
dc.tail = 1
dc.head = 2
dc.tailCounter = counter + 1
dc.headCounter = dc.tailCounter - bitSetSize + 1
return false
}
// Clear if necessary.
for i := 0; i < int(delta); i++ {
dc.put(false)
}
dc.put(true)
return false
}
func (dc *dupCheck) put(set bool) {
if set {
dc.Set(dc.tail)
} else {
dc.Clear(dc.tail)
}
dc.tail = (dc.tail + 1) % bitSetSize
dc.tailCounter++
if dc.head == dc.tail {
dc.head = (dc.head + 1) % bitSetSize
dc.headCounter++
}
}

54
node/dupcheck_test.go
View File

@ -1,54 +0,0 @@
package node
import (
"testing"
)
func TestDupCheck(t *testing.T) {
dc := newDupCheck(0)
for i := range bitSetSize {
if dc.IsDup(uint64(i)) {
t.Fatal("!")
}
}
type TestCase struct {
Counter uint64
Dup bool
}
testCases := []TestCase{
{0, true},
{1, true},
{2, true},
{3, true},
{63, true},
{256, true},
{510, true},
{511, true},
{512, false},
{0, true},
{512, true},
{513, false},
{517, false},
{512, true},
{513, true},
{514, false},
{515, false},
{516, false},
{517, true},
{2512, false},
{2000, true},
{2001, false},
{4000, false},
{4000 - 512, true}, // Too old.
{4000 - 511, false}, // Just in the window.
}
for i, tc := range testCases {
if ok := dc.IsDup(tc.Counter); ok != tc.Dup {
t.Fatal(i, ok, tc)
}
}
}

82
node/files.go
View File

@ -1,82 +0,0 @@
package node
import (
"encoding/json"
"log"
"os"
"path/filepath"
"vppn/m"
)
func configDir(netName string) string {
d, err := os.UserHomeDir()
if err != nil {
log.Fatalf("Failed to get user home directory: %v", err)
}
return filepath.Join(d, ".vppn", netName)
}
func peerConfigPath(netName string) string {
return filepath.Join(configDir(netName), "peer-config.json")
}
func peerStatePath(netName string) string {
return filepath.Join(configDir(netName), "peer-state.json")
}
func storeJson(x any, outPath string) error {
outDir := filepath.Dir(outPath)
_ = os.MkdirAll(outDir, 0700)
tmpPath := outPath + ".tmp"
buf, err := json.Marshal(x)
if err != nil {
return err
}
f, err := os.Create(tmpPath)
if err != nil {
return err
}
if _, err := f.Write(buf); err != nil {
f.Close()
return err
}
if err := f.Sync(); err != nil {
f.Close()
return err
}
if err := f.Close(); err != nil {
return err
}
return os.Rename(tmpPath, outPath)
}
func storePeerConfig(netName string, pc localConfig) error {
return storeJson(pc, peerConfigPath(netName))
}
func storeNetworkState(netName string, ps m.NetworkState) error {
return storeJson(ps, peerStatePath(netName))
}
func loadJson(dataPath string, ptr any) error {
data, err := os.ReadFile(dataPath)
if err != nil {
return err
}
return json.Unmarshal(data, ptr)
}
func loadPeerConfig(netName string) (pc localConfig, err error) {
return pc, loadJson(peerConfigPath(netName), &pc)
}
func loadNetworkState(netName string) (ps m.NetworkState, err error) {
return ps, loadJson(peerStatePath(netName), &ps)
}

8
node/globalfuncs.go
View File

@ -1,8 +0,0 @@
package node
func getRelayRoute() *peerRoute {
if ip := relayIP.Load(); ip != nil {
return routingTable[*ip].Load()
}
return nil
}

63
node/globals.go
View File

@ -1,63 +0,0 @@
package node
import (
"net"
"net/netip"
"net/url"
"sync/atomic"
)
const (
bufferSize = 1536
if_mtu = 1200
if_queue_len = 2048
controlCipherOverhead = 16
dataCipherOverhead = 16
signOverhead = 64
)
var multicastAddr = net.UDPAddrFromAddrPort(netip.AddrPortFrom(
netip.AddrFrom4([4]byte{224, 0, 0, 157}),
4560))
var (
hubURL *url.URL
apiKey string
// Configuration for this peer.
netName string
localIP byte
localPub bool
localAddr netip.AddrPort
privKey []byte
privSignKey []byte
// TODO: Doesn't need to be global.
// Duplicate checkers for incoming packets.
dupChecks [256]*dupCheck = func() (out [256]*dupCheck) {
for i := range out {
out[i] = newDupCheck(0)
}
return
}()
// TODO: Doesn't need to be global .
// Messages for the supervisor.
messages = make(chan any, 1024)
// TODO: Doesn't need to be global .
// Global routing table.
routingTable [256]*atomic.Pointer[peerRoute] = func() (out [256]*atomic.Pointer[peerRoute]) {
for i := range out {
out[i] = &atomic.Pointer[peerRoute]{}
out[i].Store(&peerRoute{})
}
return
}()
// Managed by the relayManager.
relayIP = &atomic.Pointer[byte]{}
// TODO: Only used by supervisor: can make local there.
publicAddrs = newPubAddrStore()
)

49
node/header.go
View File

@ -1,49 +0,0 @@
package node
import "unsafe"
// ----------------------------------------------------------------------------
const (
headerSize = 12
controlStreamID = 2
controlHeaderSize = 24
dataStreamID = 1
dataHeaderSize = 12
)
type header struct {
Version byte
StreamID byte
SourceIP byte
DestIP byte
Counter uint64 // Init with time.Now().Unix << 30 to ensure monotonic.
}
func parseHeader(b []byte) (h header, ok bool) {
if len(b) < headerSize {
return
}
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
}
func (h *header) Parse(b []byte) {
h.Version = b[0]
h.StreamID = b[1]
h.SourceIP = b[2]
h.DestIP = b[3]
h.Counter = *(*uint64)(unsafe.Pointer(&b[4]))
}
func (h *header) Marshal(buf []byte) {
buf[0] = h.Version
buf[1] = h.StreamID
buf[2] = h.SourceIP
buf[3] = h.DestIP
*(*uint64)(unsafe.Pointer(&buf[4])) = h.Counter
}

21
node/header_test.go
View File

@ -1,21 +0,0 @@
package node
import "testing"
func TestHeaderMarshalParse(t *testing.T) {
nIn := header{
StreamID: 23,
Counter: 3212,
SourceIP: 34,
DestIP: 200,
}
buf := make([]byte, headerSize)
nIn.Marshal(buf)
nOut := header{}
nOut.Parse(buf)
if nIn != nOut {
t.Fatal(nIn, nOut)
}
}

92
node/hubpoller.go
View File

@ -1,92 +0,0 @@
package node
import (
"encoding/json"
"io"
"log"
"net/http"
"time"
"vppn/m"
)
type hubPoller struct {
client *http.Client
req *http.Request
versions [256]int64
}
func newHubPoller() *hubPoller {
u := *hubURL
u.Path = "/peer/fetch-state/"
client := &http.Client{Timeout: 8 * time.Second}
req := &http.Request{
Method: http.MethodGet,
URL: &u,
Header: http.Header{},
}
req.SetBasicAuth("", apiKey)
return &hubPoller{
client: client,
req: req,
}
}
func (hp *hubPoller) Run() {
defer panicHandler()
state, err := loadNetworkState(netName)
if err != nil {
log.Printf("Failed to load network state: %v", err)
log.Printf("Polling hub...")
hp.pollHub()
} else {
hp.applyNetworkState(state)
}
for range time.Tick(64 * time.Second) {
hp.pollHub()
}
}
func (hp *hubPoller) pollHub() {
var state m.NetworkState
resp, err := hp.client.Do(hp.req)
if err != nil {
log.Printf("Failed to fetch peer state: %v", err)
return
}
body, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
if err != nil {
log.Printf("Failed to read body from hub: %v", err)
return
}
if err := json.Unmarshal(body, &state); err != nil {
log.Printf("Failed to unmarshal response from hub: %v\n%s", err, body)
return
}
hp.applyNetworkState(state)
if err := storeNetworkState(netName, state); err != nil {
log.Printf("Failed to store network state: %v", err)
}
}
func (hp *hubPoller) applyNetworkState(state m.NetworkState) {
for i, peer := range state.Peers {
if i != int(localIP) {
if peer == nil || peer.Version != hp.versions[i] {
messages <- peerUpdateMsg{PeerIP: byte(i), Peer: state.Peers[i]}
if peer != nil {
hp.versions[i] = peer.Version
}
}
}
}
}

102
node/ifreader.go
View File

@ -1,102 +0,0 @@
package node
import (
"io"
"log"
"sync/atomic"
)
type ifReader struct {
iface io.Reader
routes [256]*atomic.Pointer[peerRoute]
relay *atomic.Pointer[peerRoute]
sendDataPacket func(pkt []byte, route *peerRoute)
relayDataPacket func(pkt []byte, route, relay *peerRoute)
}
func newIFReader(
iface io.Reader,
routes [256]*atomic.Pointer[peerRoute],
relay *atomic.Pointer[peerRoute],
sendDataPacket func(pkt []byte, route *peerRoute),
relayDackPacket func(pkt []byte, route, relay *peerRoute),
) *ifReader {
return &ifReader{
iface: iface,
routes: routes,
relay: relay,
sendDataPacket: sendDataPacket,
}
}
func (r *ifReader) Run() {
var (
packet = make([]byte, bufferSize)
remoteIP byte
ok bool
)
for {
packet = r.readNextPacket(packet)
if remoteIP, ok = r.parsePacket(packet); ok {
r.sendPacket(packet, remoteIP)
}
}
}
func (r *ifReader) sendPacket(pkt []byte, remoteIP byte) {
route := r.routes[remoteIP].Load()
if !route.Up {
log.Printf("Route not connected: %d", remoteIP)
return
}
// Direct path => early return.
if route.Direct {
r.sendDataPacket(pkt, route)
return
}
if relay := r.relay.Load(); relay != nil && relay.Up {
r.relayDataPacket(pkt, route, relay)
}
}
// Get next packet, returning packet, and destination ip.
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)
}
return buf[:n]
}
func (r *ifReader) parsePacket(buf []byte) (byte, bool) {
n := len(buf)
if n == 0 {
return 0, false
}
version := buf[0] >> 4
switch version {
case 4:
if n < 20 {
log.Printf("Short IPv4 packet: %d", len(buf))
return 0, false
}
return buf[19], true
case 6:
if len(buf) < 40 {
log.Printf("Short IPv6 packet: %d", len(buf))
return 0, false
}
return buf[39], true
default:
log.Printf("Invalid IP packet version: %v", version)
return 0, false
}
}

117
node/ifreader_test.go
View File

@ -1,117 +0,0 @@
package node
import (
"bytes"
"net"
"sync/atomic"
"testing"
)
// Test that we parse IPv4 packets correctly.
func TestIFReader_parsePacket_ipv4(t *testing.T) {
r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil, nil)
pkt := make([]byte, 1234)
pkt[0] = 4 << 4
pkt[19] = 128
if ip, ok := r.parsePacket(pkt); !ok || ip != 128 {
t.Fatal(ip, ok)
}
}
// Test that we parse IPv6 packets correctly.
func TestIFReader_parsePacket_ipv6(t *testing.T) {
r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil, nil)
pkt := make([]byte, 1234)
pkt[0] = 6 << 4
pkt[39] = 42
if ip, ok := r.parsePacket(pkt); !ok || ip != 42 {
t.Fatal(ip, ok)
}
}
// Test that empty packets work as expected.
func TestIFReader_parsePacket_emptyPacket(t *testing.T) {
r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil, nil)
pkt := make([]byte, 0)
if ip, ok := r.parsePacket(pkt); ok {
t.Fatal(ip, ok)
}
}
// Test that invalid IP versions fail.
func TestIFReader_parsePacket_invalidIPVersion(t *testing.T) {
r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil, nil)
for i := byte(1); i < 16; i++ {
if i == 4 || i == 6 {
continue
}
pkt := make([]byte, 1234)
pkt[0] = i << 4
if ip, ok := r.parsePacket(pkt); ok {
t.Fatal(i, ip, ok)
}
}
}
// Test that short IPv4 packets fail.
func TestIFReader_parsePacket_shortIPv4(t *testing.T) {
r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil, nil)
pkt := make([]byte, 19)
pkt[0] = 4 << 4
if ip, ok := r.parsePacket(pkt); ok {
t.Fatal(ip, ok)
}
}
// Test that short IPv6 packets fail.
func TestIFReader_parsePacket_shortIPv6(t *testing.T) {
r := newIFReader(nil, [256]*atomic.Pointer[peerRoute]{}, nil, nil, nil)
pkt := make([]byte, 39)
pkt[0] = 6 << 4
if ip, ok := r.parsePacket(pkt); ok {
t.Fatal(ip, ok)
}
}
// 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, nil)
defer in.Close()
defer out.Close()
go in.Write([]byte("hello world!"))
pkt := r.readNextPacket(make([]byte, bufferSize))
if !bytes.Equal(pkt, []byte("hello world!")) {
t.Fatalf("%s", pkt)
}
}
// Testing that we can send a packet directly.
func TestIFReader_sendPacket_direct(t *testing.T) {
// TODO
}
// Testing that we don't send a packet if route isn't up.
func TestIFReader_sendPacket_directNotUp(t *testing.T) {
// TODO
}
// Testing that we can send a packet via a relay.
func TestIFReader_sendPacket_relayed(t *testing.T) {
// TODO
}
// Testing that we don't try to send on a nil relay IP.

5
node/ifwriter.go
View File

@ -1,5 +0,0 @@
package node
import "io"
type ifWriter io.Writer

177
node/interface.go
View File

@ -1,177 +0,0 @@
package node
import (
"fmt"
"io"
"log"
"net"
"os"
"syscall"
"golang.org/x/sys/unix"
)
// Get next packet, returning packet, ip, and possible error.
func readNextPacket(iface io.ReadWriteCloser, buf []byte) ([]byte, byte, error) {
var (
version byte
ip byte
)
for {
n, err := iface.Read(buf[:cap(buf)])
if err != nil {
return nil, ip, err
}
buf = buf[:n]
version = buf[0] >> 4
switch version {
case 4:
if n < 20 {
log.Printf("Short IPv4 packet: %d", len(buf))
continue
}
ip = buf[19]
case 6:
if len(buf) < 40 {
log.Printf("Short IPv6 packet: %d", len(buf))
continue
}
ip = buf[39]
default:
log.Printf("Invalid IP packet version: %v", version)
continue
}
return buf, ip, nil
}
}
func openInterface(network []byte, localIP byte, name string) (io.ReadWriteCloser, error) {
if len(network) != 4 {
return nil, fmt.Errorf("expected network to be 4 bytes, got %d", len(network))
}
ip := net.IPv4(network[0], network[1], network[2], localIP)
//////////////////////////
// Create TUN Interface //
//////////////////////////
tunFD, err := syscall.Open("/dev/net/tun", syscall.O_RDWR|unix.O_CLOEXEC, 0600)
if err != nil {
return nil, fmt.Errorf("failed to open TUN device: %w", err)
}
// New interface request.
req, err := unix.NewIfreq(name)
if err != nil {
return nil, fmt.Errorf("failed to create new TUN interface request: %w", err)
}
// Flags:
//
// IFF_NO_PI => don't add packet info data to packets sent to the interface.
// IFF_TUN => create a TUN device handling IP packets.
req.SetUint16(unix.IFF_NO_PI | unix.IFF_TUN)
err = unix.IoctlIfreq(tunFD, unix.TUNSETIFF, req)
if err != nil {
return nil, fmt.Errorf("failed to set TUN device settings: %w", err)
}
// Name may not be exactly the same?
name = req.Name()
/////////////
// Set MTU //
/////////////
// We need a socket file descriptor to set other options for some reason.
sockFD, err := unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, unix.IPPROTO_IP)
if err != nil {
return nil, fmt.Errorf("failed to open socket: %w", err)
}
defer unix.Close(sockFD)
req, err = unix.NewIfreq(name)
if err != nil {
return nil, fmt.Errorf("failed to create MTU interface request: %w", err)
}
req.SetUint32(if_mtu)
if err = unix.IoctlIfreq(sockFD, unix.SIOCSIFMTU, req); err != nil {
return nil, fmt.Errorf("failed to set interface MTU: %w", err)
}
//////////////////////
// Set Queue Length //
//////////////////////
req, err = unix.NewIfreq(name)
if err != nil {
return nil, fmt.Errorf("failed to create IP interface request: %w", err)
}
req.SetUint16(if_queue_len)
if err = unix.IoctlIfreq(sockFD, unix.SIOCSIFTXQLEN, req); err != nil {
return nil, fmt.Errorf("failed to set interface queue length: %w", err)
}
/////////////////////
// Set IP and Mask //
/////////////////////
req, err = unix.NewIfreq(name)
if err != nil {
return nil, fmt.Errorf("failed to create IP interface request: %w", err)
}
if err := req.SetInet4Addr(ip.To4()); err != nil {
return nil, fmt.Errorf("failed to set interface request IP: %w", err)
}
if err = unix.IoctlIfreq(sockFD, unix.SIOCSIFADDR, req); err != nil {
return nil, fmt.Errorf("failed to set interface IP: %w", err)
}
// SET MASK - must happen after setting address.
req, err = unix.NewIfreq(name)
if err != nil {
return nil, fmt.Errorf("failed to create mask interface request: %w", err)
}
if err := req.SetInet4Addr(net.IPv4(255, 255, 255, 0).To4()); err != nil {
return nil, fmt.Errorf("failed to set interface request mask: %w", err)
}
if err := unix.IoctlIfreq(sockFD, unix.SIOCSIFNETMASK, req); err != nil {
return nil, fmt.Errorf("failed to set interface mask: %w", err)
}
////////////////////////
// Bring Interface Up //
////////////////////////
req, err = unix.NewIfreq(name)
if err != nil {
return nil, fmt.Errorf("failed to create up interface request: %w", err)
}
// Get current flags.
if err = unix.IoctlIfreq(sockFD, unix.SIOCGIFFLAGS, req); err != nil {
return nil, fmt.Errorf("failed to get interface flags: %w", err)
}
flags := req.Uint16() | unix.IFF_UP | unix.IFF_RUNNING
// Set UP flag / broadcast flags.
req.SetUint16(flags)
if err = unix.IoctlIfreq(sockFD, unix.SIOCSIFFLAGS, req); err != nil {
return nil, fmt.Errorf("failed to set interface up: %w", err)
}
return os.NewFile(uintptr(tunFD), "tun"), nil
}

97
node/localdiscovery.go
View File

@ -1,97 +0,0 @@
package node
import (
"log"
"net"
"time"
"golang.org/x/crypto/nacl/sign"
)
func localDiscovery() {
conn, err := net.ListenMulticastUDP("udp", nil, multicastAddr)
if err != nil {
log.Printf("Failed to bind to multicast address: %v", err)
return
}
go sendLocalDiscovery(conn)
go recvLocalDiscovery(conn)
}
func sendLocalDiscovery(conn *net.UDPConn) {
var (
buf1 = make([]byte, bufferSize)
buf2 = make([]byte, bufferSize)
)
for range time.Tick(16 * time.Second) {
signed := buildLocalDiscoveryPacket(buf1, buf2)
if _, err := conn.WriteToUDP(signed, multicastAddr); err != nil {
log.Printf("Failed to write multicast UDP packet: %v", err)
}
}
}
func recvLocalDiscovery(conn *net.UDPConn) {
var (
raw = make([]byte, bufferSize)
buf = make([]byte, bufferSize)
)
for {
n, remoteAddr, err := conn.ReadFromUDPAddrPort(raw[:bufferSize])
if err != nil {
log.Fatalf("Failed to read from UDP port (multicast): %v", err)
}
raw = raw[:n]
h, ok := openLocalDiscoveryPacket(raw, buf)
if !ok {
log.Printf("Failed to open discovery packet?")
continue
}
msg := controlMsg[localDiscoveryPacket]{
SrcIP: h.SourceIP,
SrcAddr: remoteAddr,
Packet: localDiscoveryPacket{},
}
select {
case messages <- msg:
default:
log.Printf("Dropping local discovery message.")
}
}
}
func buildLocalDiscoveryPacket(buf1, buf2 []byte) []byte {
h := header{
StreamID: controlStreamID,
Counter: 0,
SourceIP: localIP,
DestIP: 255,
}
out := buf1[:headerSize]
h.Marshal(out)
return sign.Sign(buf2[:0], out, (*[64]byte)(privSignKey))
}
func openLocalDiscoveryPacket(raw, buf []byte) (h header, ok bool) {
if len(raw) != headerSize+signOverhead {
ok = false
return
}
h.Parse(raw[signOverhead:])
route := routingTable[h.SourceIP].Load()
if route == nil || route.PubSignKey == nil {
log.Printf("Missing signing key: %d", h.SourceIP)
ok = false
return
}
_, ok = sign.Open(buf[:0], raw, (*[32]byte)(route.PubSignKey))
return
}

35
node/localdiscovery_test.go
View File

@ -1,35 +0,0 @@
package node
import (
"bytes"
"crypto/rand"
"testing"
"golang.org/x/crypto/nacl/sign"
)
func TestLocalDiscoveryPacketSigning(t *testing.T) {
localIP = 32
var (
buf1 = make([]byte, bufferSize)
buf2 = make([]byte, bufferSize)
pubSignKey, privSigKey, _ = sign.GenerateKey(rand.Reader)
)
privSignKey = privSigKey[:]
route := routingTable[localIP].Load()
route.IP = byte(localIP)
route.PubSignKey = pubSignKey[:]
routingTable[localIP].Store(route)
out := buildLocalDiscoveryPacket(buf1, buf2)
h, ok := openLocalDiscoveryPacket(bytes.Clone(out), buf1)
if !ok {
t.Fatal(h, ok)
}
if h.StreamID != controlStreamID || h.SourceIP != localIP || h.DestIP != 255 {
t.Fatal(h)
}
}

320
node/main.go
View File

@ -1,320 +0,0 @@
package node
import (
"bytes"
"crypto/rand"
"encoding/json"
"flag"
"fmt"
"io"
"log"
"net"
"net/http"
"net/netip"
"net/url"
"os"
"runtime/debug"
"time"
"vppn/m"
"golang.org/x/crypto/nacl/box"
"golang.org/x/crypto/nacl/sign"
)
func panicHandler() {
if r := recover(); r != nil {
log.Fatalf("\n %v\n\nstacktrace from panic: %s\n", r, string(debug.Stack()))
}
}
func Main() {
defer panicHandler()
var hubAddress string
flag.StringVar(&netName, "name", "", "[REQUIRED] The network name.")
flag.StringVar(&hubAddress, "hub-address", "", "[REQUIRED] The hub address.")
flag.StringVar(&apiKey, "api-key", "", "[REQUIRED] The node's API key.")
flag.Parse()
if netName == "" || hubAddress == "" || apiKey == "" {
flag.Usage()
os.Exit(1)
}
var err error
hubURL, err = url.Parse(hubAddress)
if err != nil {
log.Fatalf("Failed to parse hub address: %v", err)
}
main()
}
func initPeerWithHub() {
encPubKey, encPrivKey, err := box.GenerateKey(rand.Reader)
if err != nil {
log.Fatalf("Failed to generate encryption keys: %v", err)
}
signPubKey, signPrivKey, err := sign.GenerateKey(rand.Reader)
if err != nil {
log.Fatalf("Failed to generate signing keys: %v", err)
}
initURL := *hubURL
initURL.Path = "/peer/init/"
args := m.PeerInitArgs{
EncPubKey: encPubKey[:],
PubSignKey: signPubKey[:],
}
buf := &bytes.Buffer{}
if err := json.NewEncoder(buf).Encode(args); err != nil {
log.Fatalf("Failed to encode init args: %v", err)
}
req, err := http.NewRequest(http.MethodPost, initURL.String(), buf)
if err != nil {
log.Fatalf("Failed to construct request: %v", err)
}
req.SetBasicAuth("", apiKey)
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatalf("Failed to init with hub: %v", err)
}
defer resp.Body.Close()
data, err := io.ReadAll(resp.Body)
if err != nil {
log.Fatalf("Failed to read response body: %v", err)
}
peerConfig := localConfig{}
if err := json.Unmarshal(data, &peerConfig.PeerConfig); err != nil {
log.Fatalf("Failed to parse configuration: %v\n%s", err, data)
}
peerConfig.PubKey = encPubKey[:]
peerConfig.PrivKey = encPrivKey[:]
peerConfig.PubSignKey = signPubKey[:]
peerConfig.PrivSignKey = signPrivKey[:]
if err := storePeerConfig(netName, peerConfig); err != nil {
log.Fatalf("Failed to store configuration: %v", err)
}
log.Print("Initialization successful.")
}
// ----------------------------------------------------------------------------
func main() {
config, err := loadPeerConfig(netName)
if err != nil {
log.Printf("Failed to load configuration: %v", err)
log.Printf("Initializing...")
initPeerWithHub()
config, err = loadPeerConfig(netName)
if err != nil {
log.Fatalf("Failed to load configuration: %v", err)
}
}
iface, err := openInterface(config.Network, config.PeerIP, netName)
if err != nil {
log.Fatalf("Failed to open interface: %v", err)
}
myAddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf(":%d", config.Port))
if err != nil {
log.Fatalf("Failed to resolve UDP address: %v", err)
}
conn, err := net.ListenUDP("udp", myAddr)
if err != nil {
log.Fatalf("Failed to open UDP port: %v", err)
}
conn.SetReadBuffer(1024 * 1024 * 8)
conn.SetWriteBuffer(1024 * 1024 * 8)
localIP = config.PeerIP
ip, ok := netip.AddrFromSlice(config.PublicIP)
if ok {
localPub = true
localAddr = netip.AddrPortFrom(ip, config.Port)
}
privKey = config.PrivKey
privSignKey = config.PrivSignKey
if !localPub {
go relayManager()
go localDiscovery()
}
go func() {
for range time.Tick(pingInterval) {
messages <- pingTimerMsg{}
}
}()
sender := newPacketSender(conn)
go startPeerSuper(routingTable, messages, sender)
go newHubPoller().Run()
go readFromConn(conn, iface, sender)
readFromIFace(iface, sender)
}
// ----------------------------------------------------------------------------
func readFromConn(conn *net.UDPConn, iface io.ReadWriteCloser, sender dataPacketSender) {
defer panicHandler()
var (
remoteAddr netip.AddrPort
n int
err error
buf = make([]byte, bufferSize)
decBuf = make([]byte, bufferSize)
data []byte
h header
)
for {
n, remoteAddr, err = conn.ReadFromUDPAddrPort(buf[:bufferSize])
if err != nil {
log.Fatalf("Failed to read from UDP port: %v", err)
}
remoteAddr = netip.AddrPortFrom(remoteAddr.Addr().Unmap(), remoteAddr.Port())
data = buf[:n]
if n < headerSize {
continue // Packet it soo short.
}
h.Parse(data)
switch h.StreamID {
case controlStreamID:
handleControlPacket(remoteAddr, h, data, decBuf)
case dataStreamID:
handleDataPacket(h, data, decBuf, iface, sender)
default:
log.Printf("Unknown stream ID: %d", h.StreamID)
}
}
}
func handleControlPacket(addr netip.AddrPort, h header, data, decBuf []byte) {
route := routingTable[h.SourceIP].Load()
if route.ControlCipher == nil {
//log.Printf("Not connected (control).")
return
}
if h.DestIP != localIP {
log.Printf("Incorrect destination IP on control packet: %#v", h)
return
}
out, ok := route.ControlCipher.Decrypt(data, decBuf)
if !ok {
log.Printf("Failed to decrypt control packet.")
return
}
if len(out) == 0 {
log.Printf("Empty control packet from: %d", h.SourceIP)
return
}
if dupChecks[h.SourceIP].IsDup(h.Counter) {
log.Printf("[%03d] Duplicate control packet: %d", h.SourceIP, h.Counter)
return
}
msg, err := parseControlMsg(h.SourceIP, addr, out)
if err != nil {
log.Printf("Failed to parse control packet: %v", err)
return
}
select {
case messages <- msg:
default:
log.Printf("Dropping control packet.")
}
}
func handleDataPacket(h header, data []byte, decBuf []byte, iface ifWriter, sender dataPacketSender) {
route := routingTable[h.SourceIP].Load()
if !route.Up {
log.Printf("Not connected (recv).")
return
}
dec, ok := route.DataCipher.Decrypt(data, decBuf)
if !ok {
log.Printf("Failed to decrypt data packet.")
return
}
if dupChecks[h.SourceIP].IsDup(h.Counter) {
log.Printf("[%03d] Duplicate data packet: %d", h.SourceIP, h.Counter)
return
}
if h.DestIP == localIP {
if _, err := iface.Write(dec); err != nil {
log.Fatalf("Failed to write to interface: %v", err)
}
return
}
destRoute := routingTable[h.DestIP].Load()
if !destRoute.Up {
log.Printf("Not connected (relay): %d", destRoute.IP)
return
}
sender.SendEncryptedDataPacket(dec, destRoute.RemoteAddr)
}
// ----------------------------------------------------------------------------
func readFromIFace(iface io.ReadWriteCloser, sender dataPacketSender) {
var (
packet = make([]byte, bufferSize)
remoteIP byte
err error
)
for {
packet, remoteIP, err = readNextPacket(iface, packet)
if err != nil {
log.Fatalf("Failed to read from interface: %v", err)
}
route := routingTable[remoteIP].Load()
if !route.Up {
log.Printf("Route not connected: %d", remoteIP)
continue
}
sender.SendDataPacket(packet, *route)
}
}

37
node/main_test.go
View File

@ -1,37 +0,0 @@
package node
import (
"crypto/rand"
"log"
"golang.org/x/crypto/nacl/box"
"golang.org/x/crypto/nacl/sign"
)
type testPeer struct {
IP byte
PubKey []byte
PrivKey []byte
PubSignKey []byte
PrivSignKey []byte
}
func newTestPeer(ip byte) testPeer {
encPubKey, encPrivKey, err := box.GenerateKey(rand.Reader)
if err != nil {
log.Fatalf("Failed to generate encryption keys: %v", err)
}
signPubKey, signPrivKey, err := sign.GenerateKey(rand.Reader)
if err != nil {
log.Fatalf("Failed to generate signing keys: %v", err)
}
return testPeer{
IP: ip,
PubKey: encPubKey[:],
PrivKey: encPrivKey[:],
PubSignKey: signPubKey[:],
PrivSignKey: signPrivKey[:],
}
}

62
node/mcwriter.go
View File

@ -1,62 +0,0 @@
package node
import (
"log"
"net"
"golang.org/x/crypto/nacl/sign"
)
// ----------------------------------------------------------------------------
type udpWriter interface {
WriteToUDP([]byte, *net.UDPAddr) (int, error)
}
// ----------------------------------------------------------------------------
func createLocalDiscoveryPacket(localIP byte, signingKey []byte) []byte {
h := header{
SourceIP: localIP,
DestIP: 255,
}
buf := make([]byte, headerSize)
h.Marshal(buf)
out := make([]byte, headerSize+signOverhead)
return sign.Sign(out[:0], buf, (*[64]byte)(signingKey))
}
func headerFromLocalDiscoveryPacket(pkt []byte) (h header, ok bool) {
if len(pkt) != headerSize+signOverhead {
return
}
h.Parse(pkt[signOverhead:])
ok = true
return
}
func verifyLocalDiscoveryPacket(pkt, buf []byte, pubSignKey []byte) bool {
_, ok := sign.Open(buf[:0], pkt, (*[32]byte)(pubSignKey))
return ok
}
// ----------------------------------------------------------------------------
type mcWriter struct {
conn udpWriter
discoveryPacket []byte
}
func newMCWriter(conn udpWriter, localIP byte, signingKey []byte) *mcWriter {
return &mcWriter{
conn: conn,
discoveryPacket: createLocalDiscoveryPacket(localIP, signingKey),
}
}
func (w *mcWriter) SendLocalDiscovery() {
if _, err := w.conn.WriteToUDP(w.discoveryPacket, multicastAddr); err != nil {
log.Printf("Failed to write multicast UDP packet: %v", err)
}
}

102
node/mcwriter_test.go
View File

@ -1,102 +0,0 @@
package node
import (
"bytes"
"net"
"testing"
)
// ----------------------------------------------------------------------------
// Testing that we can create and verify a local discovery packet.
func TestVerifyLocalDiscoveryPacket_valid(t *testing.T) {
keys := generateKeys()
created := createLocalDiscoveryPacket(55, keys.PrivSignKey)
header, ok := headerFromLocalDiscoveryPacket(created)
if !ok {
t.Fatal(ok)
}
if header.SourceIP != 55 || header.DestIP != 255 {
t.Fatal(header)
}
if !verifyLocalDiscoveryPacket(created, make([]byte, 1024), keys.PubSignKey) {
t.Fatal("Not valid")
}
}
// Testing that we don't try to parse short packets.
func TestVerifyLocalDiscoveryPacket_tooShort(t *testing.T) {
keys := generateKeys()
created := createLocalDiscoveryPacket(55, keys.PrivSignKey)
_, ok := headerFromLocalDiscoveryPacket(created[:len(created)-1])
if ok {
t.Fatal(ok)
}
}
// Testing that modifying a packet makes it invalid.
func TestVerifyLocalDiscoveryPacket_invalid(t *testing.T) {
keys := generateKeys()
created := createLocalDiscoveryPacket(55, keys.PrivSignKey)
buf := make([]byte, 1024)
for i := range created {
modified := bytes.Clone(created)
modified[i]++
if verifyLocalDiscoveryPacket(modified, buf, keys.PubSignKey) {
t.Fatal("Verification should have failed.")
}
}
}
// ----------------------------------------------------------------------------
type testUDPWriter struct {
written [][]byte
}
func (w *testUDPWriter) WriteToUDP(b []byte, addr *net.UDPAddr) (int, error) {
w.written = append(w.written, bytes.Clone(b))
return len(b), nil
}
func (w *testUDPWriter) Written() [][]byte {
out := w.written
w.written = [][]byte{}
return out
}
// ----------------------------------------------------------------------------
// Testing that the mcWriter sends local discovery packets as expected.
func TestMCWriter_SendLocalDiscovery(t *testing.T) {
keys := generateKeys()
writer := &testUDPWriter{}
mcw := newMCWriter(writer, 42, keys.PrivSignKey)
mcw.SendLocalDiscovery()
out := writer.Written()
if len(out) != 1 {
t.Fatal(out)
}
pkt := out[0]
header, ok := headerFromLocalDiscoveryPacket(pkt)
if !ok {
t.Fatal(ok)
}
if header.SourceIP != 42 || header.DestIP != 255 {
t.Fatal(header)
}
if !verifyLocalDiscoveryPacket(pkt, make([]byte, 1024), keys.PubSignKey) {
t.Fatal("Verification should succeed.")
}
}

58
node/messages.go
View File

@ -1,58 +0,0 @@
package node
import (
"net/netip"
"vppn/m"
)
// ----------------------------------------------------------------------------
type controlMsg[T any] struct {
SrcIP byte
SrcAddr netip.AddrPort
// TODO: RecvdAt int64 // Unixmilli.
Packet T
}
func parseControlMsg(srcIP byte, srcAddr netip.AddrPort, buf []byte) (any, error) {
switch buf[0] {
case packetTypeSyn:
packet, err := parseSynPacket(buf)
return controlMsg[synPacket]{
SrcIP: srcIP,
SrcAddr: srcAddr,
Packet: packet,
}, err
case packetTypeAck:
packet, err := parseAckPacket(buf)
return controlMsg[ackPacket]{
SrcIP: srcIP,
SrcAddr: srcAddr,
Packet: packet,
}, err
case packetTypeProbe:
packet, err := parseProbePacket(buf)
return controlMsg[probePacket]{
SrcIP: srcIP,
SrcAddr: srcAddr,
Packet: packet,
}, err
default:
return nil, errUnknownPacketType
}
}
// ----------------------------------------------------------------------------
type peerUpdateMsg struct {
PeerIP byte
Peer *m.Peer
}
// ----------------------------------------------------------------------------
type pingTimerMsg struct{}

190
node/packets-util.go
View File

@ -1,190 +0,0 @@
package node
import (
"net/netip"
"sync/atomic"
"time"
"unsafe"
)
var traceIDCounter uint64 = uint64(time.Now().Unix()<<30) + 1
func newTraceID() uint64 {
return atomic.AddUint64(&traceIDCounter, 1)
}
// ----------------------------------------------------------------------------
type binWriter struct {
b []byte
i int
}
func newBinWriter(buf []byte) *binWriter {
buf = buf[:cap(buf)]
return &binWriter{buf, 0}
}
func (w *binWriter) Bool(b bool) *binWriter {
if b {
return w.Byte(1)
}
return w.Byte(0)
}
func (w *binWriter) Byte(b byte) *binWriter {
w.b[w.i] = b
w.i++
return w
}
func (w *binWriter) SharedKey(key [32]byte) *binWriter {
copy(w.b[w.i:w.i+32], key[:])
w.i += 32
return w
}
func (w *binWriter) Uint16(x uint16) *binWriter {
*(*uint16)(unsafe.Pointer(&w.b[w.i])) = x
w.i += 2
return w
}
func (w *binWriter) Uint64(x uint64) *binWriter {
*(*uint64)(unsafe.Pointer(&w.b[w.i])) = x
w.i += 8
return w
}
func (w *binWriter) Int64(x int64) *binWriter {
*(*int64)(unsafe.Pointer(&w.b[w.i])) = x
w.i += 8
return w
}
func (w *binWriter) AddrPort(addrPort netip.AddrPort) *binWriter {
w.Bool(addrPort.IsValid())
addr := addrPort.Addr().As16()
copy(w.b[w.i:w.i+16], addr[:])
w.i += 16
return w.Uint16(addrPort.Port())
}
func (w *binWriter) AddrPortArray(l [8]netip.AddrPort) *binWriter {
for _, addrPort := range l {
w.AddrPort(addrPort)
}
return w
}
func (w *binWriter) Build() []byte {
return w.b[:w.i]
}
// ----------------------------------------------------------------------------
type binReader struct {
b []byte
i int
err error
}
func newBinReader(buf []byte) *binReader {
return &binReader{b: buf}
}
func (r *binReader) hasBytes(n int) bool {
if r.err != nil || (len(r.b)-r.i) < n {
r.err = errMalformedPacket
return false
}
return true
}
func (r *binReader) Bool(b *bool) *binReader {
var bb byte
r.Byte(&bb)
*b = bb != 0
return r
}
func (r *binReader) Byte(b *byte) *binReader {
if !r.hasBytes(1) {
return r
}
*b = r.b[r.i]
r.i++
return r
}
func (r *binReader) SharedKey(x *[32]byte) *binReader {
if !r.hasBytes(32) {
return r
}
*x = ([32]byte)(r.b[r.i : r.i+32])
r.i += 32
return r
}
func (r *binReader) Uint16(x *uint16) *binReader {
if !r.hasBytes(2) {
return r
}
*x = *(*uint16)(unsafe.Pointer(&r.b[r.i]))
r.i += 2
return r
}
func (r *binReader) Uint64(x *uint64) *binReader {
if !r.hasBytes(8) {
return r
}
*x = *(*uint64)(unsafe.Pointer(&r.b[r.i]))
r.i += 8
return r
}
func (r *binReader) Int64(x *int64) *binReader {
if !r.hasBytes(8) {
return r
}
*x = *(*int64)(unsafe.Pointer(&r.b[r.i]))
r.i += 8
return r
}
func (r *binReader) AddrPort(x *netip.AddrPort) *binReader {
if !r.hasBytes(19) {
return r
}
var (
valid bool
port uint16
)
r.Bool(&valid)
addr := netip.AddrFrom16(([16]byte)(r.b[r.i : r.i+16])).Unmap()
r.i += 16
r.Uint16(&port)
if valid {
*x = netip.AddrPortFrom(addr, port)
} else {
*x = netip.AddrPort{}
}
return r
}
func (r *binReader) AddrPortArray(x *[8]netip.AddrPort) *binReader {
for i := range x {
r.AddrPort(&x[i])
}
return r
}
func (r *binReader) Error() error {
return r.err
}

56
node/packets-util_test.go
View File

@ -1,56 +0,0 @@
package node
import (
"net/netip"
"reflect"
"testing"
)
func TestBinWriteRead(t *testing.T) {
buf := make([]byte, 1024)
type Item struct {
Type byte
TraceID uint64
Addrs [8]netip.AddrPort
DestAddr netip.AddrPort
}
in := Item{
1,
2,
[8]netip.AddrPort{},
netip.AddrPortFrom(netip.AddrFrom4([4]byte{1, 2, 3, 4}), 22),
}
in.Addrs[0] = netip.AddrPortFrom(netip.AddrFrom4([4]byte{0, 1, 2, 3}), 20)
in.Addrs[2] = netip.AddrPortFrom(netip.AddrFrom4([4]byte{2, 3, 4, 5}), 22)
in.Addrs[3] = netip.AddrPortFrom(netip.AddrFrom4([4]byte{2, 3, 4, 3}), 23)
in.Addrs[4] = netip.AddrPortFrom(netip.AddrFrom4([4]byte{2, 3, 4, 4}), 24)
in.Addrs[5] = netip.AddrPortFrom(netip.AddrFrom4([4]byte{2, 3, 4, 5}), 25)
in.Addrs[6] = netip.AddrPortFrom(netip.AddrFrom4([4]byte{2, 3, 4, 6}), 26)
in.Addrs[7] = netip.AddrPortFrom(netip.AddrFrom4([4]byte{7, 8, 9, 7}), 27)
buf = newBinWriter(buf).
Byte(in.Type).
Uint64(in.TraceID).
AddrPort(in.DestAddr).
AddrPortArray(in.Addrs).
Build()
out := Item{}
err := newBinReader(buf).
Byte(&out.Type).
Uint64(&out.TraceID).
AddrPort(&out.DestAddr).
AddrPortArray(&out.Addrs).
Error()
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(in, out) {
t.Fatal(in, out)
}
}

130
node/packets.go
View File

@ -1,130 +0,0 @@
package node
import (
"errors"
"net/netip"
)
var (
errMalformedPacket = errors.New("malformed packet")
errUnknownPacketType = errors.New("unknown packet type")
)
const (
packetTypeSyn = iota + 1
packetTypeSynAck
packetTypeAck
packetTypeProbe
packetTypeAddrDiscovery
)
// ----------------------------------------------------------------------------
type synPacket struct {
TraceID uint64 // TraceID to match response w/ request.
// TODO: SentAt int64 // Unixmilli.
SharedKey [32]byte // Our shared key.
Direct bool
PossibleAddrs [8]netip.AddrPort // Possible public addresses of the sender.
}
func (p synPacket) Marshal(buf []byte) []byte {
return newBinWriter(buf).
Byte(packetTypeSyn).
Uint64(p.TraceID).
SharedKey(p.SharedKey).
Bool(p.Direct).
AddrPort(p.PossibleAddrs[0]).
AddrPort(p.PossibleAddrs[1]).
AddrPort(p.PossibleAddrs[2]).
AddrPort(p.PossibleAddrs[3]).
AddrPort(p.PossibleAddrs[4]).
AddrPort(p.PossibleAddrs[5]).
AddrPort(p.PossibleAddrs[6]).
AddrPort(p.PossibleAddrs[7]).
Build()
}
func parseSynPacket(buf []byte) (p synPacket, err error) {
err = newBinReader(buf[1:]).
Uint64(&p.TraceID).
SharedKey(&p.SharedKey).
Bool(&p.Direct).
AddrPort(&p.PossibleAddrs[0]).
AddrPort(&p.PossibleAddrs[1]).
AddrPort(&p.PossibleAddrs[2]).
AddrPort(&p.PossibleAddrs[3]).
AddrPort(&p.PossibleAddrs[4]).
AddrPort(&p.PossibleAddrs[5]).
AddrPort(&p.PossibleAddrs[6]).
AddrPort(&p.PossibleAddrs[7]).
Error()
return
}
// ----------------------------------------------------------------------------
type ackPacket struct {
TraceID uint64
ToAddr netip.AddrPort
PossibleAddrs [8]netip.AddrPort // Possible public addresses of the sender.
}
func (p ackPacket) Marshal(buf []byte) []byte {
return newBinWriter(buf).
Byte(packetTypeAck).
Uint64(p.TraceID).
AddrPort(p.ToAddr).
AddrPort(p.PossibleAddrs[0]).
AddrPort(p.PossibleAddrs[1]).
AddrPort(p.PossibleAddrs[2]).
AddrPort(p.PossibleAddrs[3]).
AddrPort(p.PossibleAddrs[4]).
AddrPort(p.PossibleAddrs[5]).
AddrPort(p.PossibleAddrs[6]).
AddrPort(p.PossibleAddrs[7]).
Build()
}
func parseAckPacket(buf []byte) (p ackPacket, err error) {
err = newBinReader(buf[1:]).
Uint64(&p.TraceID).
AddrPort(&p.ToAddr).
AddrPort(&p.PossibleAddrs[0]).
AddrPort(&p.PossibleAddrs[1]).
AddrPort(&p.PossibleAddrs[2]).
AddrPort(&p.PossibleAddrs[3]).
AddrPort(&p.PossibleAddrs[4]).
AddrPort(&p.PossibleAddrs[5]).
AddrPort(&p.PossibleAddrs[6]).
AddrPort(&p.PossibleAddrs[7]).
Error()
return
}
// ----------------------------------------------------------------------------
// A probeReqPacket is sent from a client to a server to determine if direct
// UDP communication can be used.
type probePacket struct {
TraceID uint64
}
func (p probePacket) Marshal(buf []byte) []byte {
return newBinWriter(buf).
Byte(packetTypeProbe).
Uint64(p.TraceID).
Build()
}
func parseProbePacket(buf []byte) (p probePacket, err error) {
err = newBinReader(buf[1:]).
Uint64(&p.TraceID).
Error()
return
}
// ----------------------------------------------------------------------------
type localDiscoveryPacket struct{}

1
node/packets_test.go
View File

@ -1 +0,0 @@
package node

127
node/packetsender.go
View File

@ -1,127 +0,0 @@
package node
import (
"log"
"net"
"net/netip"
"sync"
"sync/atomic"
"time"
)
type controlPacketSender interface {
SendControlPacket(pkt marshaller, route peerRoute)
}
type dataPacketSender interface {
SendDataPacket(pkt []byte, route peerRoute)
SendEncryptedDataPacket(pkt []byte, addr netip.AddrPort)
}
// ----------------------------------------------------------------------------
type packetSender struct {
conn *net.UDPConn
// For sending control packets.
cLock sync.Mutex
cBuf1 []byte
cBuf2 []byte
// For sending data packets.
dBuf1 []byte
dBuf2 []byte
counters [256]uint64
// Lock around for sending on UDP Conn.
wLock sync.Mutex
}
func newPacketSender(conn *net.UDPConn) *packetSender {
ps := &packetSender{
conn: conn,
cBuf1: make([]byte, bufferSize),
cBuf2: make([]byte, bufferSize),
dBuf1: make([]byte, bufferSize),
dBuf2: make([]byte, bufferSize),
}
for i := range ps.counters {
ps.counters[i] = uint64(time.Now().Unix()<<30 + 1)
}
return ps
}
// Safe for concurrent use.
func (sender *packetSender) SendControlPacket(pkt marshaller, route peerRoute) {
sender.cLock.Lock()
defer sender.cLock.Unlock()
buf := pkt.Marshal(sender.cBuf1)
h := header{
StreamID: controlStreamID,
Counter: atomic.AddUint64(&sender.counters[route.IP], 1),
SourceIP: localIP,
DestIP: route.IP,
}
buf = route.ControlCipher.Encrypt(h, buf, sender.cBuf2)
if route.Direct {
sender.writeTo(buf, route.RemoteAddr)
return
}
sender.relayPacket(route.IP, buf, sender.cBuf1)
}
// Not safe for concurrent use.
func (sender *packetSender) SendDataPacket(pkt []byte, route peerRoute) {
h := header{
StreamID: dataStreamID,
Counter: atomic.AddUint64(&sender.counters[route.IP], 1),
SourceIP: localIP,
DestIP: route.IP,
}
enc := route.DataCipher.Encrypt(h, pkt, sender.dBuf1)
if route.Direct {
sender.writeTo(enc, route.RemoteAddr)
return
}
sender.relayPacket(route.IP, enc, sender.dBuf2)
}
func (sender *packetSender) SendEncryptedDataPacket(pkt []byte, addr netip.AddrPort) {
sender.writeTo(pkt, addr)
}
func (sender *packetSender) relayPacket(destIP byte, data, buf []byte) {
ip := relayIP.Load()
if ip == nil {
return
}
relayRoute := routingTable[*ip].Load()
if relayRoute == nil || !relayRoute.Up || !relayRoute.Relay {
return
}
h := header{
StreamID: dataStreamID,
Counter: atomic.AddUint64(&sender.counters[relayRoute.IP], 1),
SourceIP: localIP,
DestIP: destIP,
}
enc := relayRoute.DataCipher.Encrypt(h, data, buf)
sender.writeTo(enc, relayRoute.RemoteAddr)
}
func (sender *packetSender) writeTo(packet []byte, addr netip.AddrPort) {
sender.wLock.Lock()
if _, err := sender.conn.WriteToUDPAddrPort(packet, addr); err != nil {
log.Printf("Failed to write to UDP port: %v", err)
}
sender.wLock.Unlock()
}

41
node/relaymanager.go
View File

@ -1,41 +0,0 @@
package node
import (
"log"
"math/rand"
"time"
)
// TODO: Make part of main loop on ping timer
func relayManager() {
time.Sleep(2 * time.Second)
updateRelayRoute()
for range time.Tick(8 * time.Second) {
relay := getRelayRoute()
if relay == nil || !relay.Up || !relay.Relay {
updateRelayRoute()
}
}
}
func updateRelayRoute() {
possible := make([]*peerRoute, 0, 8)
for i := range routingTable {
route := routingTable[i].Load()
if !route.Up || !route.Relay {
continue
}
possible = append(possible, route)
}
if len(possible) == 0 {
log.Printf("No relay available.")
relayIP.Store(nil)
return
}
ip := possible[rand.Intn(len(possible))].IP
log.Printf("New relay IP: %d", ip)
relayIP.Store(&ip)
}

59
node/shared.go
View File

@ -1,59 +0,0 @@
package node
import (
"net/netip"
"sync/atomic"
)
type sharedState struct {
// Immutable:
HubAddress string
APIKey string
NetName string
LocalIP byte
LocalPub bool
LocalAddr netip.AddrPort
PrivKey []byte
PrivSignKey []byte
// Mutable:
Routes [256]*atomic.Pointer[peerRoute]
RelayIP *atomic.Pointer[byte]
// Messages for supervisor main loop.
Messages chan any
}
func newSharedState(
netName,
hubAddress,
apiKey string,
conf localConfig,
) (
ss sharedState,
) {
ss.HubAddress = hubAddress
ss.APIKey = apiKey
ss.NetName = netName
ss.LocalIP = conf.PeerIP
ip, ok := netip.AddrFromSlice(conf.PublicIP)
if ok {
ss.LocalPub = true
ss.LocalAddr = netip.AddrPortFrom(ip, conf.Port)
}
ss.PrivKey = conf.PrivKey
ss.PrivSignKey = conf.PrivSignKey
for i := range ss.Routes {
ss.Routes[i] = &atomic.Pointer[peerRoute]{}
ss.Routes[i].Store(&peerRoute{})
}
ss.RelayIP = &atomic.Pointer[byte]{}
ss.Messages = make(chan any, 1024)
return
}

16
node/shared_test.go
View File

@ -1,16 +0,0 @@
package node
import "vppn/m"
// TODO:
var sharedStateForTesting = func() sharedState {
ss := newSharedState(
"testNet",
"http://localhost:39499",
"123",
localConfig{
PeerConfig: m.PeerConfig{},
})
return ss
}

421
node/supervisor.go
View File

@ -1,421 +0,0 @@
package node
import (
"fmt"
"log"
"net/netip"
"strings"
"sync/atomic"
"time"
"vppn/m"
"git.crumpington.com/lib/go/ratelimiter"
)
const (
pingInterval = 8 * time.Second
timeoutInterval = 30 * time.Second
)
// ----------------------------------------------------------------------------
func startPeerSuper(
routingTable [256]*atomic.Pointer[peerRoute],
messages chan any,
sender controlPacketSender,
) {
peers := [256]peerState{}
for i := range peers {
data := &peerStateData{
sender: sender,
published: routingTable[i],
remoteIP: byte(i),
limiter: ratelimiter.New(ratelimiter.Config{
FillPeriod: 20 * time.Millisecond,
MaxWaitCount: 1,
}),
}
peers[i] = data.OnPeerUpdate(nil)
}
go runPeerSuper(peers, messages)
}
func runPeerSuper(peers [256]peerState, messages chan any) {
for raw := range messages {
switch msg := raw.(type) {
case peerUpdateMsg:
peers[msg.PeerIP] = peers[msg.PeerIP].OnPeerUpdate(msg.Peer)
case controlMsg[synPacket]:
peers[msg.SrcIP].OnSyn(msg)
case controlMsg[ackPacket]:
peers[msg.SrcIP].OnAck(msg)
case controlMsg[probePacket]:
peers[msg.SrcIP].OnProbe(msg)
case controlMsg[localDiscoveryPacket]:
peers[msg.SrcIP].OnLocalDiscovery(msg)
case pingTimerMsg:
publicAddrs.Clean()
for i := range peers {
if newState := peers[i].OnPingTimer(); newState != nil {
peers[i] = newState
}
}
default:
log.Printf("WARNING: unknown message type: %+v", msg)
}
}
}
// ----------------------------------------------------------------------------
type peerState interface {
OnPeerUpdate(*m.Peer) peerState
OnSyn(controlMsg[synPacket])
OnAck(controlMsg[ackPacket])
OnProbe(controlMsg[probePacket])
OnLocalDiscovery(controlMsg[localDiscoveryPacket])
OnPingTimer() peerState
}
// ----------------------------------------------------------------------------
type peerStateData struct {
sender controlPacketSender
// The purpose of this state machine is to manage this published data.
published *atomic.Pointer[peerRoute]
staged peerRoute // Local copy of shared data. See publish().
// Immutable data.
remoteIP byte // Remote VPN IP.
// Mutable peer data.
peer *m.Peer
remotePub bool
// For logging. Set per-state.
client bool
// We rate limit per remote endpoint because if we don't we tend to lose
// packets.
limiter *ratelimiter.Limiter
}
// ----------------------------------------------------------------------------
func (s *peerStateData) sendControlPacket(pkt interface{ Marshal([]byte) []byte }) {
s._sendControlPacket(pkt, s.staged)
}
func (s *peerStateData) sendControlPacketTo(pkt interface{ Marshal([]byte) []byte }, addr netip.AddrPort) {
if !addr.IsValid() {
s.logf("ERROR: Attepted to send packet to invalid address: %v", addr)
return
}
route := s.staged
route.Direct = true
route.RemoteAddr = addr
s._sendControlPacket(pkt, route)
}
func (s *peerStateData) _sendControlPacket(pkt interface{ Marshal([]byte) []byte }, route peerRoute) {
if err := s.limiter.Limit(); err != nil {
s.logf("Not sending control packet: rate limited.") // Shouldn't happen.
return
}
s.sender.SendControlPacket(pkt, route)
}
// ----------------------------------------------------------------------------
func (s *peerStateData) publish() {
data := s.staged
s.published.Store(&data)
}
func (s *peerStateData) logf(format string, args ...any) {
b := strings.Builder{}
b.WriteString(fmt.Sprintf("%30s: ", s.peer.Name))
if s.client {
b.WriteString("CLIENT | ")
} else {
b.WriteString("SERVER | ")
}
if s.staged.Direct {
b.WriteString("DIRECT | ")
} else {
b.WriteString("RELAYED | ")
}
if s.staged.Up {
b.WriteString("UP | ")
} else {
b.WriteString("DOWN | ")
}
log.Printf(b.String()+format, args...)
}
// ----------------------------------------------------------------------------
func (s *peerStateData) OnPeerUpdate(peer *m.Peer) peerState {
defer s.publish()
if peer == nil {
return enterStateDisconnected(s)
}
s.peer = peer
s.staged = peerRoute{
IP: s.remoteIP,
PubSignKey: peer.PubSignKey,
// TODO: privKey global.
ControlCipher: newControlCipher(privKey, peer.PubKey),
DataCipher: newDataCipher(),
}
s.remotePub = false
if ip, isValid := netip.AddrFromSlice(peer.PublicIP); isValid {
s.remotePub = true
s.staged.Relay = peer.Relay
s.staged.Direct = true
s.staged.RemoteAddr = netip.AddrPortFrom(ip, peer.Port)
} else if localPub {
s.staged.Direct = true
}
if s.remotePub == localPub {
// TODO: localIP is global
if localIP < s.remoteIP {
return enterStateServer(s)
}
return enterStateClient(s)
}
if s.remotePub {
return enterStateClient(s)
}
return enterStateServer(s)
}
// ----------------------------------------------------------------------------
type stateDisconnected struct {
*peerStateData
}
func enterStateDisconnected(s *peerStateData) peerState {
s.peer = nil
s.staged = peerRoute{}
s.publish()
return &stateDisconnected{s}
}
func (s *stateDisconnected) OnSyn(controlMsg[synPacket]) {}
func (s *stateDisconnected) OnAck(controlMsg[ackPacket]) {}
func (s *stateDisconnected) OnProbe(controlMsg[probePacket]) {}
func (s *stateDisconnected) OnLocalDiscovery(controlMsg[localDiscoveryPacket]) {}
func (s *stateDisconnected) OnPingTimer() peerState {
return nil
}
// ----------------------------------------------------------------------------
type stateServer struct {
*stateDisconnected
lastSeen time.Time
synTraceID uint64
}
func enterStateServer(s *peerStateData) peerState {
s.client = false
return &stateServer{stateDisconnected: &stateDisconnected{s}}
}
func (s *stateServer) OnSyn(msg controlMsg[synPacket]) {
s.lastSeen = time.Now()
p := msg.Packet
// Before we can respond to this packet, we need to make sure the
// route is setup properly.
//
// The client will update the syn's TraceID whenever there's a change.
// The server will follow the client's request.
if p.TraceID != s.synTraceID || !s.staged.Up {
s.synTraceID = p.TraceID
s.staged.Up = true
s.staged.Direct = p.Direct
s.staged.DataCipher = newDataCipherFromKey(p.SharedKey)
s.staged.RemoteAddr = msg.SrcAddr
s.publish()
s.logf("Got syn.")
}
// Always respond.
ack := ackPacket{
TraceID: p.TraceID,
ToAddr: s.staged.RemoteAddr,
PossibleAddrs: publicAddrs.Get(),
}
s.sendControlPacket(ack)
if s.staged.Direct {
return
}
// Not direct => send probes.
for _, addr := range p.PossibleAddrs {
if !addr.IsValid() {
break
}
s.sendControlPacketTo(probePacket{TraceID: newTraceID()}, addr)
}
}
func (s *stateServer) OnProbe(msg controlMsg[probePacket]) {
if !msg.SrcAddr.IsValid() {
s.logf("Invalid probe address.")
return
}
s.sendControlPacketTo(probePacket{TraceID: msg.Packet.TraceID}, msg.SrcAddr)
}
func (s *stateServer) OnPingTimer() peerState {
if time.Since(s.lastSeen) > timeoutInterval && s.staged.Up {
s.staged.Up = false
s.publish()
s.logf("Connection timeout.")
}
return nil
}
// ----------------------------------------------------------------------------
type stateClient struct {
*stateDisconnected
lastSeen time.Time
syn synPacket
ack ackPacket
probes map[uint64]netip.AddrPort
localDiscoveryAddr netip.AddrPort
}
func enterStateClient(s *peerStateData) peerState {
s.client = true
ss := &stateClient{
stateDisconnected: &stateDisconnected{s},
probes: map[uint64]netip.AddrPort{},
}
ss.syn = synPacket{
TraceID: newTraceID(),
SharedKey: s.staged.DataCipher.Key(),
Direct: s.staged.Direct,
PossibleAddrs: publicAddrs.Get(),
}
ss.sendControlPacket(ss.syn)
return ss
}
func (s *stateClient) sendProbeTo(addr netip.AddrPort) {
probe := probePacket{TraceID: newTraceID()}
s.probes[probe.TraceID] = addr
s.sendControlPacketTo(probe, addr)
}
func (s *stateClient) OnAck(msg controlMsg[ackPacket]) {
if msg.Packet.TraceID != s.syn.TraceID {
s.logf("Ack has incorrect trace ID")
return
}
s.ack = msg.Packet
s.lastSeen = time.Now()
if !s.staged.Up {
s.staged.Up = true
s.logf("Got ack.")
s.publish()
}
// Store possible public address if we're not a public node.
// TODO: localPub is global, publicAddrs is global.
if !localPub && s.remotePub {
publicAddrs.Store(msg.Packet.ToAddr)
}
}
func (s *stateClient) OnProbe(msg controlMsg[probePacket]) {
if s.staged.Direct {
return
}
addr, ok := s.probes[msg.Packet.TraceID]
if !ok {
return
}
s.staged.RemoteAddr = addr
s.staged.Direct = true
s.publish()
s.syn.TraceID = newTraceID()
s.syn.Direct = true
s.syn.PossibleAddrs = [8]netip.AddrPort{}
s.sendControlPacket(s.syn)
s.logf("Established direct connection to %s.", s.staged.RemoteAddr.String())
}
func (s *stateClient) OnLocalDiscovery(msg controlMsg[localDiscoveryPacket]) {
if s.staged.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.
s.localDiscoveryAddr = netip.AddrPortFrom(msg.SrcAddr.Addr(), s.peer.Port)
}
func (s *stateClient) OnPingTimer() peerState {
if time.Since(s.lastSeen) > timeoutInterval {
if s.staged.Up {
s.logf("Connection timeout.")
}
return s.OnPeerUpdate(s.peer)
}
s.sendControlPacket(s.syn)
if s.staged.Direct {
return nil
}
clear(s.probes)
for _, addr := range s.ack.PossibleAddrs {
if !addr.IsValid() {
break
}
s.sendProbeTo(addr)
}
if s.localDiscoveryAddr.IsValid() {
s.sendProbeTo(s.localDiscoveryAddr)
s.localDiscoveryAddr = netip.AddrPort{}
}
return nil
}