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 +107 -783
8 changed files with 107 additions and 783 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 7141020412 - Show all commits

510
peer/peerstates.go
View File

@@ -1,510 +0,0 @@
package peer
import (
"fmt"
"log"
"net/netip"
"strings"
"time"
"vppn/m"
"git.crumpington.com/lib/go/ratelimiter"
)
type peerState interface {
OnMsg(raw any) peerState
}
// ----------------------------------------------------------------------------
type pState struct {
// Output.
publish func(remotePeer)
sendControlPacket func(remotePeer, marshaller)
pingTimer *time.Ticker
// Immutable data.
localIP byte
remoteIP byte
privKey []byte
localAddr netip.AddrPort // If valid, then local peer is publicly accessible.
pubAddrs *pubAddrStore
// The purpose of this state machine is to manage the RemotePeer object,
// publishing it as necessary.
staged remotePeer // Local copy of shared data. See publish().
// Mutable peer data.
peer *m.Peer
// We rate limit per remote endpoint because if we don't we tend to lose
// packets.
limiter *ratelimiter.Limiter
}
/*
func (s *pState) OnPeerUpdate(peer *m.Peer) peerState {
defer func() {
// Don't defer directly otherwise s.staged will be evaluated immediately
// and won't reflect changes made in the function.
s.publish(s.staged)
}()
s.peer = peer
s.staged.localIP = s.localIP
s.staged.Up = false
s.staged.Relay = false
s.staged.Direct = false
s.staged.DirectAddr = netip.AddrPort{}
s.staged.PubSignKey = nil
s.staged.ControlCipher = nil
s.staged.DataCipher = nil
if peer == nil {
return enterStateDisconnected(s)
}
s.staged.IP = peer.PeerIP
s.staged.PubSignKey = peer.PubSignKey
s.staged.ControlCipher = newControlCipher(s.privKey, peer.PubKey)
s.staged.DataCipher = newDataCipher()
if ip, isValid := netip.AddrFromSlice(peer.PublicIP); isValid {
s.staged.Relay = peer.Relay
s.staged.Direct = true
s.staged.DirectAddr = netip.AddrPortFrom(ip, peer.Port)
if s.localAddr.IsValid() && s.localIP < s.remoteIP {
return enterStateServer(s)
}
return enterStateClientinit(s)
}
if s.localAddr.IsValid() {
s.staged.Direct = true
return enterStateServer(s)
}
if s.localIP < s.remoteIP {
return enterStateServer(s)
}
return enterStateClientinit(s)
}
*/
func (s *pState) logf(format string, args ...any) {
b := strings.Builder{}
name := ""
if s.peer != nil {
name = s.peer.Name
}
b.WriteString(fmt.Sprintf("%03d", s.remoteIP))
b.WriteString(fmt.Sprintf("%30s: ", name))
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 *pState) SendTo(pkt marshaller, addr netip.AddrPort) {
if !addr.IsValid() {
return
}
route := s.staged
route.Direct = true
route.DirectAddr = addr
s.Send(route, pkt)
}
func (s *pState) Send(peer remotePeer, pkt marshaller) {
if err := s.limiter.Limit(); err != nil {
s.logf("Rate limited.")
return
}
s.sendControlPacket(peer, pkt)
}
// ----------------------------------------------------------------------------
/*
type stateDisconnected struct{ *pState }
func enterStateDisconnected(s *pState) peerState {
s.pingTimer.Stop()
return &stateDisconnected{pState: s}
}
func (s *stateDisconnected) OnMsg(raw any) peerState {
switch msg := raw.(type) {
case peerUpdateMsg:
return s.OnPeerUpdate(msg.Peer)
default:
// TODO: Log.
return s
}
}
func (s *stateDisconnected) OnSyn(controlMsg[packetSyn]) peerState { return s }
func (s *stateDisconnected) OnAck(controlMsg[packetAck]) {}
func (s *stateDisconnected) OnProbe(controlMsg[packetProbe]) peerState { return s }
func (s *stateDisconnected) OnLocalDiscovery(controlMsg[packetLocalDiscovery]) {}
func (s *stateDisconnected) OnPingTimer() peerState { return s }
// ----------------------------------------------------------------------------
type stateServer struct {
*stateDisconnected
lastSeen time.Time
synTraceID uint64
}
func enterStateServer(s *pState) peerState {
s.logf("==> Server")
s.pingTimer.Reset(pingInterval)
return &stateServer{stateDisconnected: &stateDisconnected{pState: s}}
}
func (s *stateServer) OnMsg(rawMsg any) peerState {
switch msg := rawMsg.(type) {
case peerUpdateMsg:
return s.OnPeerUpdate(msg.Peer)
case controlMsg[packetInit]:
return s.OnInit(msg)
case controlMsg[packetSyn]:
return s.OnSyn(msg)
case controlMsg[packetProbe]:
return s.OnProbe(msg)
case pingTimerMsg:
return s.OnPingTimer()
default:
// TODO: Log
return s
}
}
func (s *stateServer) OnInit(msg controlMsg[packetInit]) peerState {
s.logf("Responding to INIT.")
route := s.staged
route.Direct = msg.Packet.Direct
route.DirectAddr = msg.SrcAddr
s.Send(route, packetInit{
TraceID: msg.Packet.TraceID,
Direct: route.Direct,
Version: version,
})
return s
}
func (s *stateServer) OnSyn(msg controlMsg[packetSyn]) peerState {
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.DirectAddr = msg.SrcAddr
s.publish(s.staged)
s.logf("Got SYN.")
}
// Always respond.
ack := packetAck{
TraceID: p.TraceID,
ToAddr: s.staged.DirectAddr,
PossibleAddrs: s.pubAddrs.Get(),
}
s.Send(s.staged, ack)
if p.Direct {
return s
}
for _, addr := range msg.Packet.PossibleAddrs {
if !addr.IsValid() {
break
}
s.SendTo(packetProbe{TraceID: newTraceID()}, addr)
}
return s
}
func (s *stateServer) OnProbe(msg controlMsg[packetProbe]) peerState {
if msg.SrcAddr.IsValid() {
s.SendTo(packetProbe{TraceID: msg.Packet.TraceID}, msg.SrcAddr)
}
return s
}
func (s *stateServer) OnPingTimer() peerState {
if time.Since(s.lastSeen) > timeoutInterval && s.staged.Up {
s.staged.Up = false
s.publish(s.staged)
s.logf("Timeout.")
}
return s
}
// ----------------------------------------------------------------------------
type stateClientInit struct {
*stateDisconnected
startedAt time.Time
traceID uint64
}
func enterStateClientinit(s *pState) peerState {
s.logf("==> ClientInit")
s.pingTimer.Reset(pingInterval)
state := &stateClientInit{
stateDisconnected: &stateDisconnected{s},
startedAt: time.Now(),
traceID: newTraceID(),
}
state.Send(s.staged, packetInit{
TraceID: state.traceID,
Direct: s.staged.Direct,
Version: version,
})
return state
}
func (s *stateClientInit) OnMsg(raw any) peerState {
switch msg := raw.(type) {
case peerUpdateMsg:
return s.OnPeerUpdate(msg.Peer)
case controlMsg[packetInit]:
return s.onInit(msg)
case pingTimerMsg:
return s.onPing()
default:
return s
}
}
func (s *stateClientInit) onInit(msg controlMsg[packetInit]) peerState {
if msg.Packet.TraceID != s.traceID {
s.logf("Invalid trace ID on INIT.")
return s
}
s.logf("Got INIT version %d.", msg.Packet.Version)
return s.nextState()
}
func (s *stateClientInit) onPing() peerState {
if time.Since(s.startedAt) > timeoutInterval {
s.logf("Init timeout. Assuming version 1.")
return s.nextState()
}
s.traceID = newTraceID()
s.Send(s.staged, packetInit{
TraceID: s.traceID,
Direct: s.staged.Direct,
Version: version,
})
return s
}
func (s *stateClientInit) nextState() peerState {
if s.staged.Direct {
return enterStateClientDirect(s.pState)
}
return enterStateClientRelayed(s.pState)
}
// ----------------------------------------------------------------------------
type stateClientDirect struct {
*stateDisconnected
lastSeen time.Time
syn packetSyn
}
func enterStateClientDirect(s *pState) peerState {
s.logf("==> ClientDirect")
s.pingTimer.Reset(pingInterval)
return newStateClientDirect(s)
}
func newStateClientDirect(s *pState) *stateClientDirect {
state := &stateClientDirect{
stateDisconnected: &stateDisconnected{s},
lastSeen: time.Now(), // Avoid immediate timeout.
}
state.syn = packetSyn{
TraceID: newTraceID(),
SharedKey: s.staged.DataCipher.Key(),
Direct: s.staged.Direct,
PossibleAddrs: s.pubAddrs.Get(),
}
state.Send(s.staged, state.syn)
return state
}
func (s *stateClientDirect) OnMsg(raw any) peerState {
switch msg := raw.(type) {
case peerUpdateMsg:
return s.OnPeerUpdate(msg.Peer)
case controlMsg[packetAck]:
s.OnAck(msg)
return s
case pingTimerMsg:
if next := s.onPingTimer(); next != nil {
return next
}
return s
default:
// TODO: Log
return s
}
}
func (s *stateClientDirect) OnAck(msg controlMsg[packetAck]) {
if msg.Packet.TraceID != s.syn.TraceID {
return
}
s.lastSeen = time.Now()
if !s.staged.Up {
s.staged.Up = true
s.publish(s.staged)
s.logf("Got ACK.")
}
s.pubAddrs.Store(msg.Packet.ToAddr)
}
func (s *stateClientDirect) onPingTimer() peerState {
if time.Since(s.lastSeen) > timeoutInterval {
if s.staged.Up {
s.staged.Up = false
s.publish(s.staged)
s.logf("Timeout.")
}
return s.OnPeerUpdate(s.peer)
}
s.Send(s.staged, s.syn)
return nil
}
// ----------------------------------------------------------------------------
type stateClientRelayed struct {
*stateClientDirect
ack packetAck
probes map[uint64]netip.AddrPort // TODO: something better
localDiscoveryAddr netip.AddrPort // TODO: Remove
}
func enterStateClientRelayed(s *pState) peerState {
s.logf("==> ClientRelayed")
s.pingTimer.Reset(pingInterval)
return &stateClientRelayed{
stateClientDirect: newStateClientDirect(s),
probes: map[uint64]netip.AddrPort{},
}
}
func (s *stateClientRelayed) OnMsg(raw any) peerState {
switch msg := raw.(type) {
case peerUpdateMsg:
return s.OnPeerUpdate(msg.Peer)
case controlMsg[packetAck]:
s.OnAck(msg)
return s
case controlMsg[packetProbe]:
return s.OnProbe(msg)
case controlMsg[packetLocalDiscovery]:
s.OnLocalDiscovery(msg)
return s
case pingTimerMsg:
return s.OnPingTimer()
default:
// TODO: Log
return s
}
}
func (s *stateClientRelayed) OnAck(msg controlMsg[packetAck]) {
s.ack = msg.Packet
s.stateClientDirect.OnAck(msg)
// TODO: Send probes now.
}
func (s *stateClientRelayed) OnProbe(msg controlMsg[packetProbe]) peerState {
addr, ok := s.probes[msg.Packet.TraceID]
if !ok {
return s
}
s.staged.DirectAddr = addr
s.staged.Direct = true
s.publish(s.staged)
return enterStateClientDirect(s.stateClientDirect.pState)
}
func (s *stateClientRelayed) OnLocalDiscovery(msg controlMsg[packetLocalDiscovery]) {
// 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)
// TODO: s.sendProbeTo(s.localDiscoveryAddr)
}
func (s *stateClientRelayed) OnPingTimer() peerState {
if next := s.stateClientDirect.onPingTimer(); next != nil {
return next
}
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 s
}
func (s *stateClientRelayed) sendProbeTo(addr netip.AddrPort) {
probe := packetProbe{TraceID: newTraceID()}
s.probes[probe.TraceID] = addr
s.SendTo(probe, addr)
}
*/

2
peer/peerstates_test.go
View File

@@ -27,7 +27,7 @@ func NewPeerStateTestHarness() *PeerStateTestHarness {
keys := generateKeys() keys := generateKeys()
state := &pState{ state := &peerData{
publish: func(rp remotePeer) { publish: func(rp remotePeer) {
h.Published = rp h.Published = rp
}, },

4
peer/peersuper.go
View File

@@ -39,7 +39,7 @@ func newSupervisor(
pubAddrs := newPubAddrStore(routes.LocalAddr) pubAddrs := newPubAddrStore(routes.LocalAddr)
for i := range s.peers { for i := range s.peers {
state := &pState{ state := &peerData{
publish: s.publish, publish: s.publish,
sendControlPacket: s.send, sendControlPacket: s.send,
pingTimer: time.NewTicker(timeoutInterval), pingTimer: time.NewTicker(timeoutInterval),
@@ -121,7 +121,7 @@ type peerSuper struct {
pingTimer *time.Ticker pingTimer *time.Ticker
} }
func newPeerSuper(state *pState, pingTimer *time.Ticker) *peerSuper { func newPeerSuper(state *peerData, pingTimer *time.Ticker) *peerSuper {
return &peerSuper{ return &peerSuper{
messages: make(chan any, 8), messages: make(chan any, 8),
state: initPeerState(state, nil), state: initPeerState(state, nil),

85
peer/state-clientdirect.go
View File

@@ -1,85 +0,0 @@
package peer
import (
"net/netip"
"time"
)
type stateClientDirect2 struct {
*peerData
lastSeen time.Time
syn packetSyn
}
func enterStateClientDirect2(data *peerData, directAddr netip.AddrPort) peerState {
data.staged.Relay = data.peer.Relay
data.staged.Direct = true
data.staged.DirectAddr = directAddr
data.publish(data.staged)
state := &stateClientDirect2{
peerData: data,
lastSeen: time.Now(),
syn: packetSyn{
TraceID: newTraceID(),
SharedKey: data.staged.DataCipher.Key(),
Direct: true,
},
}
state.Send(state.staged, state.syn)
data.pingTimer.Reset(pingInterval)
state.logf("==> ClientDirect")
return state
}
func (s *stateClientDirect2) logf(str string, args ...any) {
s.peerData.logf("CLNT | "+str, args...)
}
func (s *stateClientDirect2) OnMsg(raw any) peerState {
switch msg := raw.(type) {
case peerUpdateMsg:
return initPeerState(s.peerData, msg.Peer)
case controlMsg[packetAck]:
return s.onAck(msg)
case pingTimerMsg:
return s.onPingTimer()
case controlMsg[packetLocalDiscovery]:
return s
default:
s.logf("Ignoring message: %v", raw)
return s
}
}
func (s *stateClientDirect2) onAck(msg controlMsg[packetAck]) peerState {
if msg.Packet.TraceID != s.syn.TraceID {
return s
}
s.lastSeen = time.Now()
if !s.staged.Up {
s.staged.Up = true
s.publish(s.staged)
s.logf("Got ACK.")
}
s.pubAddrs.Store(msg.Packet.ToAddr)
return s
}
func (s *stateClientDirect2) onPingTimer() peerState {
if time.Since(s.lastSeen) > timeoutInterval {
if s.staged.Up {
s.logf("Timeout.")
}
return initPeerState(s.peerData, s.peer)
}
s.Send(s.staged, s.syn)
return s
}

16
peer/state-clientinit.go
View File

@@ -5,13 +5,13 @@ import (
"time" "time"
) )
type stateClientInit2 struct { type stateClientInit struct {
*peerData *peerData
startedAt time.Time startedAt time.Time
traceID uint64 traceID uint64
} }
func enterStateClientInit2(data *peerData) peerState { func enterStateClientInit(data *peerData) peerState {
ip, ipValid := netip.AddrFromSlice(data.peer.PublicIP) ip, ipValid := netip.AddrFromSlice(data.peer.PublicIP)
data.staged.Up = false data.staged.Up = false
@@ -24,7 +24,7 @@ func enterStateClientInit2(data *peerData) peerState {
data.publish(data.staged) data.publish(data.staged)
state := &stateClientInit2{ state := &stateClientInit{
peerData: data, peerData: data,
startedAt: time.Now(), startedAt: time.Now(),
traceID: newTraceID(), traceID: newTraceID(),
@@ -37,11 +37,11 @@ func enterStateClientInit2(data *peerData) peerState {
return state return state
} }
func (s *stateClientInit2) logf(str string, args ...any) { func (s *stateClientInit) logf(str string, args ...any) {
s.peerData.logf("INIT | "+str, args...) s.peerData.logf("INIT | "+str, args...)
} }
func (s *stateClientInit2) OnMsg(raw any) peerState { func (s *stateClientInit) OnMsg(raw any) peerState {
switch msg := raw.(type) { switch msg := raw.(type) {
case peerUpdateMsg: case peerUpdateMsg:
return initPeerState(s.peerData, msg.Peer) return initPeerState(s.peerData, msg.Peer)
@@ -55,7 +55,7 @@ func (s *stateClientInit2) OnMsg(raw any) peerState {
} }
} }
func (s *stateClientInit2) onInit(msg controlMsg[packetInit]) peerState { func (s *stateClientInit) onInit(msg controlMsg[packetInit]) peerState {
if msg.Packet.TraceID != s.traceID { if msg.Packet.TraceID != s.traceID {
s.logf("Invalid trace ID on INIT.") s.logf("Invalid trace ID on INIT.")
return s return s
@@ -64,7 +64,7 @@ func (s *stateClientInit2) onInit(msg controlMsg[packetInit]) peerState {
return enterStateClient(s.peerData) return enterStateClient(s.peerData)
} }
func (s *stateClientInit2) onPing() peerState { func (s *stateClientInit) onPing() peerState {
if time.Since(s.startedAt) > timeoutInterval { if time.Since(s.startedAt) > timeoutInterval {
s.logf("Init timeout. Assuming version 1.") s.logf("Init timeout. Assuming version 1.")
return enterStateClient(s.peerData) return enterStateClient(s.peerData)
@@ -74,7 +74,7 @@ func (s *stateClientInit2) onPing() peerState {
return s return s
} }
func (s *stateClientInit2) sendInit() { func (s *stateClientInit) sendInit() {
s.traceID = newTraceID() s.traceID = newTraceID()
init := packetInit{ init := packetInit{
TraceID: s.traceID, TraceID: s.traceID,

163
peer/state-clientrelayed.go
View File

@@ -1,163 +0,0 @@
package peer
import (
"net/netip"
"time"
)
type sentProbe struct {
SentAt time.Time
Addr netip.AddrPort
}
type stateClientRelayed2 struct {
*peerData
lastSeen time.Time
syn packetSyn
probes map[uint64]sentProbe
}
func enterStateClient(data *peerData) peerState {
ip, ipValid := netip.AddrFromSlice(data.peer.PublicIP)
data.staged.Relay = data.peer.Relay && ipValid
data.staged.Direct = ipValid
data.staged.DirectAddr = netip.AddrPortFrom(ip, data.peer.Port)
data.publish(data.staged)
state := &stateClientRelayed2{
peerData: data,
lastSeen: time.Now(),
syn: packetSyn{
TraceID: newTraceID(),
SharedKey: data.staged.DataCipher.Key(),
Direct: data.staged.Direct,
PossibleAddrs: data.pubAddrs.Get(),
},
probes: map[uint64]sentProbe{},
}
state.Send(state.staged, state.syn)
data.pingTimer.Reset(pingInterval)
state.logf("==> Client")
return state
}
func (s *stateClientRelayed2) logf(str string, args ...any) {
s.peerData.logf("CLNT | "+str, args...)
}
func (s *stateClientRelayed2) OnMsg(raw any) peerState {
switch msg := raw.(type) {
case peerUpdateMsg:
return initPeerState(s.peerData, msg.Peer)
case controlMsg[packetAck]:
s.onAck(msg)
case controlMsg[packetProbe]:
return s.onProbe(msg)
case controlMsg[packetLocalDiscovery]:
s.onLocalDiscovery(msg)
case pingTimerMsg:
return s.onPingTimer()
default:
s.logf("Ignoring message: %v", raw)
}
return s
}
func (s *stateClientRelayed2) onAck(msg controlMsg[packetAck]) {
if msg.Packet.TraceID != s.syn.TraceID {
return
}
s.lastSeen = time.Now()
if !s.staged.Up {
s.staged.Up = true
s.publish(s.staged)
s.logf("Got ACK.")
}
if s.staged.Direct {
s.pubAddrs.Store(msg.Packet.ToAddr)
return
}
// Relayed below.
s.cleanProbes()
for _, addr := range msg.Packet.PossibleAddrs {
if !addr.IsValid() {
break
}
s.sendProbeTo(addr)
}
}
func (s *stateClientRelayed2) onPingTimer() peerState {
if time.Since(s.lastSeen) > timeoutInterval {
if s.staged.Up {
s.logf("Timeout.")
}
return initPeerState(s.peerData, s.peer)
}
s.Send(s.staged, s.syn)
return s
}
func (s *stateClientRelayed2) onProbe(msg controlMsg[packetProbe]) peerState {
if s.staged.Direct {
return s
}
s.cleanProbes()
sent, ok := s.probes[msg.Packet.TraceID]
if !ok {
return s
}
s.staged.Direct = true
s.staged.DirectAddr = sent.Addr
s.publish(s.staged)
s.syn.TraceID = newTraceID()
s.syn.Direct = true
s.Send(s.staged, s.syn)
s.logf("Successful probe.")
return s
}
func (s *stateClientRelayed2) onLocalDiscovery(msg controlMsg[packetLocalDiscovery]) {
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.
addr := netip.AddrPortFrom(msg.SrcAddr.Addr(), s.peer.Port)
s.sendProbeTo(addr)
}
func (s *stateClientRelayed2) cleanProbes() {
for key, sent := range s.probes {
if time.Since(sent.SentAt) > pingInterval {
delete(s.probes, key)
}
}
}
func (s *stateClientRelayed2) sendProbeTo(addr netip.AddrPort) {
probe := packetProbe{TraceID: newTraceID()}
s.probes[probe.TraceID] = sentProbe{
SentAt: time.Now(),
Addr: addr,
}
s.logf("Probing %v...", addr)
s.SendTo(probe, addr)
}

19
peer/state-server.go
View File

@@ -5,13 +5,13 @@ import (
"time" "time"
) )
type stateServer2 struct { type stateServer struct {
*peerData *peerData
lastSeen time.Time lastSeen time.Time
synTraceID uint64 // Last syn trace ID. synTraceID uint64 // Last syn trace ID.
} }
func enterStateServer2(data *peerData) peerState { func enterStateServer(data *peerData) peerState {
data.staged.Up = false data.staged.Up = false
data.staged.Relay = false data.staged.Relay = false
data.staged.Direct = false data.staged.Direct = false
@@ -24,16 +24,16 @@ func enterStateServer2(data *peerData) peerState {
data.pingTimer.Reset(pingInterval) data.pingTimer.Reset(pingInterval)
state := &stateServer2{peerData: data} state := &stateServer{peerData: data}
state.logf("==> Server") state.logf("==> Server")
return state return state
} }
func (s *stateServer2) logf(str string, args ...any) { func (s *stateServer) logf(str string, args ...any) {
s.peerData.logf("SRVR | "+str, args...) s.peerData.logf("SRVR | "+str, args...)
} }
func (s *stateServer2) OnMsg(raw any) peerState { func (s *stateServer) OnMsg(raw any) peerState {
switch msg := raw.(type) { switch msg := raw.(type) {
case peerUpdateMsg: case peerUpdateMsg:
return initPeerState(s.peerData, msg.Peer) return initPeerState(s.peerData, msg.Peer)
@@ -53,7 +53,7 @@ func (s *stateServer2) OnMsg(raw any) peerState {
} }
} }
func (s *stateServer2) onInit(msg controlMsg[packetInit]) peerState { func (s *stateServer) onInit(msg controlMsg[packetInit]) peerState {
s.staged.Up = false s.staged.Up = false
s.staged.Direct = msg.Packet.Direct s.staged.Direct = msg.Packet.Direct
s.staged.DirectAddr = msg.SrcAddr s.staged.DirectAddr = msg.SrcAddr
@@ -70,7 +70,7 @@ func (s *stateServer2) onInit(msg controlMsg[packetInit]) peerState {
return s return s
} }
func (s *stateServer2) onSyn(msg controlMsg[packetSyn]) peerState { func (s *stateServer) onSyn(msg controlMsg[packetSyn]) peerState {
s.lastSeen = time.Now() s.lastSeen = time.Now()
p := msg.Packet p := msg.Packet
@@ -100,6 +100,7 @@ func (s *stateServer2) onSyn(msg controlMsg[packetSyn]) peerState {
return s return s
} }
// Send probes if not a direct connection.
for _, addr := range msg.Packet.PossibleAddrs { for _, addr := range msg.Packet.PossibleAddrs {
if !addr.IsValid() { if !addr.IsValid() {
break break
@@ -111,7 +112,7 @@ func (s *stateServer2) onSyn(msg controlMsg[packetSyn]) peerState {
return s return s
} }
func (s *stateServer2) onProbe(msg controlMsg[packetProbe]) peerState { func (s *stateServer) onProbe(msg controlMsg[packetProbe]) peerState {
if msg.SrcAddr.IsValid() { if msg.SrcAddr.IsValid() {
s.logf("Probe response %v...", msg.SrcAddr) s.logf("Probe response %v...", msg.SrcAddr)
s.SendTo(packetProbe{TraceID: msg.Packet.TraceID}, msg.SrcAddr) s.SendTo(packetProbe{TraceID: msg.Packet.TraceID}, msg.SrcAddr)
@@ -119,7 +120,7 @@ func (s *stateServer2) onProbe(msg controlMsg[packetProbe]) peerState {
return s return s
} }
func (s *stateServer2) onPingTimer() peerState { func (s *stateServer) onPingTimer() peerState {
if time.Since(s.lastSeen) > timeoutInterval && s.staged.Up { if time.Since(s.lastSeen) > timeoutInterval && s.staged.Up {
s.staged.Up = false s.staged.Up = false
s.publish(s.staged) s.publish(s.staged)

91
peer/statedata.go
View File

@@ -1,11 +1,92 @@
package peer package peer
import ( import (
"fmt"
"log"
"net/netip" "net/netip"
"strings"
"time"
"vppn/m" "vppn/m"
"git.crumpington.com/lib/go/ratelimiter"
) )
type peerData = pState type peerState interface {
OnMsg(raw any) peerState
}
// ----------------------------------------------------------------------------
type peerData struct {
// Output.
publish func(remotePeer)
sendControlPacket func(remotePeer, marshaller)
pingTimer *time.Ticker
// Immutable data.
localIP byte
remoteIP byte
privKey []byte
localAddr netip.AddrPort // If valid, then local peer is publicly accessible.
pubAddrs *pubAddrStore
// The purpose of this state machine is to manage the RemotePeer object,
// publishing it as necessary.
staged remotePeer // Local copy of shared data. See publish().
// Mutable peer data.
peer *m.Peer
// We rate limit per remote endpoint because if we don't we tend to lose
// packets.
limiter *ratelimiter.Limiter
}
func (s *peerData) logf(format string, args ...any) {
b := strings.Builder{}
name := ""
if s.peer != nil {
name = s.peer.Name
}
b.WriteString(fmt.Sprintf("%03d", s.remoteIP))
b.WriteString(fmt.Sprintf("%30s: ", name))
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 *peerData) SendTo(pkt marshaller, addr netip.AddrPort) {
if !addr.IsValid() {
return
}
route := s.staged
route.Direct = true
route.DirectAddr = addr
s.Send(route, pkt)
}
func (s *peerData) Send(peer remotePeer, pkt marshaller) {
if err := s.limiter.Limit(); err != nil {
s.logf("Rate limited.")
return
}
s.sendControlPacket(peer, pkt)
}
func initPeerState(data *peerData, peer *m.Peer) peerState { func initPeerState(data *peerData, peer *m.Peer) peerState {
data.peer = peer data.peer = peer
@@ -16,13 +97,13 @@ func initPeerState(data *peerData, peer *m.Peer) peerState {
if _, isValid := netip.AddrFromSlice(peer.PublicIP); isValid { if _, isValid := netip.AddrFromSlice(peer.PublicIP); isValid {
if data.localAddr.IsValid() && data.localIP < data.remoteIP { if data.localAddr.IsValid() && data.localIP < data.remoteIP {
return enterStateServer2(data) return enterStateServer(data)
} }
return enterStateClientInit2(data) return enterStateClientInit(data)
} }
if data.localAddr.IsValid() || data.localIP < data.remoteIP { if data.localAddr.IsValid() || data.localIP < data.remoteIP {
return enterStateServer2(data) return enterStateServer(data)
} }
return enterStateClientInit2(data) return enterStateClientInit(data)
} }