From 71410204128801adcad71a64974ac005d757e8ae Mon Sep 17 00:00:00 2001 From: jdl Date: Tue, 25 Feb 2025 18:14:39 +0100 Subject: [PATCH] wip --- peer/peerstates.go | 510 ------------------------------------ peer/peerstates_test.go | 2 +- peer/peersuper.go | 4 +- peer/state-clientdirect.go | 85 ------ peer/state-clientinit.go | 16 +- peer/state-clientrelayed.go | 163 ------------ peer/state-server.go | 19 +- peer/statedata.go | 91 ++++++- 8 files changed, 107 insertions(+), 783 deletions(-) delete mode 100644 peer/peerstates.go delete mode 100644 peer/state-clientdirect.go delete mode 100644 peer/state-clientrelayed.go diff --git a/peer/peerstates.go b/peer/peerstates.go deleted file mode 100644 index 6c52f55..0000000 --- a/peer/peerstates.go +++ /dev/null @@ -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) -} -*/ diff --git a/peer/peerstates_test.go b/peer/peerstates_test.go index 15f7d18..26ebacd 100644 --- a/peer/peerstates_test.go +++ b/peer/peerstates_test.go @@ -27,7 +27,7 @@ func NewPeerStateTestHarness() *PeerStateTestHarness { keys := generateKeys() - state := &pState{ + state := &peerData{ publish: func(rp remotePeer) { h.Published = rp }, diff --git a/peer/peersuper.go b/peer/peersuper.go index ec8c741..2ce6d03 100644 --- a/peer/peersuper.go +++ b/peer/peersuper.go @@ -39,7 +39,7 @@ func newSupervisor( pubAddrs := newPubAddrStore(routes.LocalAddr) for i := range s.peers { - state := &pState{ + state := &peerData{ publish: s.publish, sendControlPacket: s.send, pingTimer: time.NewTicker(timeoutInterval), @@ -121,7 +121,7 @@ type peerSuper struct { pingTimer *time.Ticker } -func newPeerSuper(state *pState, pingTimer *time.Ticker) *peerSuper { +func newPeerSuper(state *peerData, pingTimer *time.Ticker) *peerSuper { return &peerSuper{ messages: make(chan any, 8), state: initPeerState(state, nil), diff --git a/peer/state-clientdirect.go b/peer/state-clientdirect.go deleted file mode 100644 index c6c552d..0000000 --- a/peer/state-clientdirect.go +++ /dev/null @@ -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 -} diff --git a/peer/state-clientinit.go b/peer/state-clientinit.go index 674d63e..f34854a 100644 --- a/peer/state-clientinit.go +++ b/peer/state-clientinit.go @@ -5,13 +5,13 @@ import ( "time" ) -type stateClientInit2 struct { +type stateClientInit struct { *peerData startedAt time.Time traceID uint64 } -func enterStateClientInit2(data *peerData) peerState { +func enterStateClientInit(data *peerData) peerState { ip, ipValid := netip.AddrFromSlice(data.peer.PublicIP) data.staged.Up = false @@ -24,7 +24,7 @@ func enterStateClientInit2(data *peerData) peerState { data.publish(data.staged) - state := &stateClientInit2{ + state := &stateClientInit{ peerData: data, startedAt: time.Now(), traceID: newTraceID(), @@ -37,11 +37,11 @@ func enterStateClientInit2(data *peerData) peerState { return state } -func (s *stateClientInit2) logf(str string, args ...any) { +func (s *stateClientInit) logf(str string, args ...any) { s.peerData.logf("INIT | "+str, args...) } -func (s *stateClientInit2) OnMsg(raw any) peerState { +func (s *stateClientInit) OnMsg(raw any) peerState { switch msg := raw.(type) { case peerUpdateMsg: 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 { s.logf("Invalid trace ID on INIT.") return s @@ -64,7 +64,7 @@ func (s *stateClientInit2) onInit(msg controlMsg[packetInit]) peerState { return enterStateClient(s.peerData) } -func (s *stateClientInit2) onPing() peerState { +func (s *stateClientInit) onPing() peerState { if time.Since(s.startedAt) > timeoutInterval { s.logf("Init timeout. Assuming version 1.") return enterStateClient(s.peerData) @@ -74,7 +74,7 @@ func (s *stateClientInit2) onPing() peerState { return s } -func (s *stateClientInit2) sendInit() { +func (s *stateClientInit) sendInit() { s.traceID = newTraceID() init := packetInit{ TraceID: s.traceID, diff --git a/peer/state-clientrelayed.go b/peer/state-clientrelayed.go deleted file mode 100644 index b51398d..0000000 --- a/peer/state-clientrelayed.go +++ /dev/null @@ -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) -} diff --git a/peer/state-server.go b/peer/state-server.go index 4543a60..723af5c 100644 --- a/peer/state-server.go +++ b/peer/state-server.go @@ -5,13 +5,13 @@ import ( "time" ) -type stateServer2 struct { +type stateServer struct { *peerData lastSeen time.Time synTraceID uint64 // Last syn trace ID. } -func enterStateServer2(data *peerData) peerState { +func enterStateServer(data *peerData) peerState { data.staged.Up = false data.staged.Relay = false data.staged.Direct = false @@ -24,16 +24,16 @@ func enterStateServer2(data *peerData) peerState { data.pingTimer.Reset(pingInterval) - state := &stateServer2{peerData: data} + state := &stateServer{peerData: data} state.logf("==> Server") return state } -func (s *stateServer2) logf(str string, args ...any) { +func (s *stateServer) logf(str string, args ...any) { s.peerData.logf("SRVR | "+str, args...) } -func (s *stateServer2) OnMsg(raw any) peerState { +func (s *stateServer) OnMsg(raw any) peerState { switch msg := raw.(type) { case peerUpdateMsg: 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.Direct = msg.Packet.Direct s.staged.DirectAddr = msg.SrcAddr @@ -70,7 +70,7 @@ func (s *stateServer2) onInit(msg controlMsg[packetInit]) peerState { return s } -func (s *stateServer2) onSyn(msg controlMsg[packetSyn]) peerState { +func (s *stateServer) onSyn(msg controlMsg[packetSyn]) peerState { s.lastSeen = time.Now() p := msg.Packet @@ -100,6 +100,7 @@ func (s *stateServer2) onSyn(msg controlMsg[packetSyn]) peerState { return s } + // Send probes if not a direct connection. for _, addr := range msg.Packet.PossibleAddrs { if !addr.IsValid() { break @@ -111,7 +112,7 @@ func (s *stateServer2) onSyn(msg controlMsg[packetSyn]) peerState { return s } -func (s *stateServer2) onProbe(msg controlMsg[packetProbe]) peerState { +func (s *stateServer) onProbe(msg controlMsg[packetProbe]) peerState { if msg.SrcAddr.IsValid() { s.logf("Probe response %v...", 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 } -func (s *stateServer2) onPingTimer() peerState { +func (s *stateServer) onPingTimer() peerState { if time.Since(s.lastSeen) > timeoutInterval && s.staged.Up { s.staged.Up = false s.publish(s.staged) diff --git a/peer/statedata.go b/peer/statedata.go index 44330fa..0ea0929 100644 --- a/peer/statedata.go +++ b/peer/statedata.go @@ -1,11 +1,92 @@ package peer import ( + "fmt" + "log" "net/netip" + "strings" + "time" "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 { data.peer = peer @@ -16,13 +97,13 @@ func initPeerState(data *peerData, peer *m.Peer) peerState { if _, isValid := netip.AddrFromSlice(peer.PublicIP); isValid { 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 { - return enterStateServer2(data) + return enterStateServer(data) } - return enterStateClientInit2(data) + return enterStateClientInit(data) }