From 199f774ed3738fba3b6f8cceae35bd552299e07f Mon Sep 17 00:00:00 2001 From: jdl Date: Sat, 6 Jun 2026 18:46:46 +0200 Subject: [PATCH] WIP --- peer/app.go | 110 +++++++++++++++++++++++ peer/app_test.go | 37 ++++++++ peer/control.go | 150 +++++++++++++++++++++++++++++++ peer/device.go | 48 ++++++++++ peer/endpointreporter.go | 88 ++++++++++++++++++ peer/fake_wgdevice_test.go | 67 ++++++++++++++ peer/holepunch.go | 155 ++++++++++++++++++++++++++++++++ peer/interfaces.go | 18 ++++ peer/on_hub.go | 115 ++++++++++++++++++++++++ peer/on_multicast.go | 27 ++++++ peer/on_ping.go | 66 ++++++++++++++ peer/on_tick.go | 50 +++++++++++ peer/ping.go | 9 ++ peer/remote.go | 46 ++++++++++ peer/wgdev.go | 177 +++++++++++++++++++++++++++++++++++++ peer/wginterface/manage.go | 5 ++ 16 files changed, 1168 insertions(+) create mode 100644 peer/app.go create mode 100644 peer/app_test.go create mode 100644 peer/control.go create mode 100644 peer/device.go create mode 100644 peer/endpointreporter.go create mode 100644 peer/fake_wgdevice_test.go create mode 100644 peer/holepunch.go create mode 100644 peer/interfaces.go create mode 100644 peer/on_hub.go create mode 100644 peer/on_multicast.go create mode 100644 peer/on_ping.go create mode 100644 peer/on_tick.go create mode 100644 peer/ping.go create mode 100644 peer/remote.go create mode 100644 peer/wgdev.go diff --git a/peer/app.go b/peer/app.go new file mode 100644 index 0000000..5102beb --- /dev/null +++ b/peer/app.go @@ -0,0 +1,110 @@ +package peer + +import ( + "net" + "net/netip" + "os" + "os/signal" + "syscall" + "time" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" + + "vppn/peer/control" + "vppn/peer/wginterface" +) + +var _ WGDevice = (*wginterface.Device)(nil) // compile-time check: Device satisfies WGDevice + +const ( + ControlPort = 4561 + PingInterval = 8 * time.Second + TimeoutInterval = 30 * time.Second +) + +// HubPeer is a peer entry as reported by the hub poller. +type HubPeer struct { + PubKey wgtypes.Key + VPNIP netip.Addr + IsRelay bool + IsPublic bool + EndpointV4 netip.AddrPort // zero if none + EndpointV6 netip.AddrPort // zero if none +} + +type PingEvent struct { + srcVPNIP netip.Addr + ping control.Ping +} + +type MulticastEvent struct { + pubKey wgtypes.Key + vpnIP netip.Addr + endpoint netip.AddrPort +} + +// App is the peer application. All mutable state lives here and is +// accessed only from the Run goroutine. +type App struct { + // Identity + vpnIP netip.Addr + vpnNet netip.Prefix + privKey wgtypes.Key + pubKey wgtypes.Key + isRelay bool + isPublic bool + + // Infrastructure + dev WGDevice + controlConn *net.UDPConn + + // Peer state + relay *Peer + peersByKey map[wgtypes.Key]*Peer + peersByIP map[netip.Addr]*Peer + + // Our own external endpoints, learned from Dst fields in incoming pings + selfV4 netip.AddrPort + selfV6 netip.AddrPort + + // Monotonically increasing ID for outbound pings (client role only) + nextPingID int64 + + // Event channels fed by background goroutines + hubAddCh <-chan HubPeer + hubRemoveCh <-chan wgtypes.Key + pingCh <-chan PingEvent + multicastCh <-chan MulticastEvent +} + +// Run is the main event loop. It runs until SIGTERM/SIGINT. +func (a *App) Run() error { + ticker := time.NewTicker(PingInterval) + defer ticker.Stop() + + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGTERM, syscall.SIGINT) + defer signal.Stop(sig) + + for { + select { + case p := <-a.hubAddCh: + a.onAddPeer(p) + case key := <-a.hubRemoveCh: + a.onRemovePeer(key) + case e := <-a.pingCh: + a.onPing(e) + case e := <-a.multicastCh: + a.onMulticastDiscovery(e) + case <-ticker.C: + a.onTick() + case <-sig: + return a.onShutdown() + } + } +} + + +func (a *App) onShutdown() error { + return wginterface.Delete(a.dev.Name()) +} diff --git a/peer/app_test.go b/peer/app_test.go new file mode 100644 index 0000000..8d909d7 --- /dev/null +++ b/peer/app_test.go @@ -0,0 +1,37 @@ +package peer + +import ( + "net/netip" + "testing" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +// newTestApp returns a minimal App wired to a fakeWGDevice. +// vpnIP is the local VPN address (e.g. "10.0.0.1"). +// isPublic / isRelay describe the local node's role. +func newTestApp(t *testing.T, vpnIP string, isPublic, isRelay bool) (*App, *fakeWGDevice) { + t.Helper() + privKey, err := wgtypes.GeneratePrivateKey() + if err != nil { + t.Fatalf("generate key: %v", err) + } + ip := netip.MustParseAddr(vpnIP) + dev := &fakeWGDevice{} + a := &App{ + vpnIP: ip, + vpnNet: netip.MustParsePrefix("10.0.0.0/24"), + privKey: privKey, + pubKey: privKey.PublicKey(), + isPublic: isPublic, + isRelay: isRelay, + dev: dev, + peersByKey: make(map[wgtypes.Key]*Peer), + peersByIP: make(map[netip.Addr]*Peer), + hubAddCh: make(chan HubPeer), + hubRemoveCh: make(chan wgtypes.Key), + pingCh: make(chan PingEvent), + multicastCh: make(chan MulticastEvent), + } + return a, dev +} diff --git a/peer/control.go b/peer/control.go new file mode 100644 index 0000000..cb929a7 --- /dev/null +++ b/peer/control.go @@ -0,0 +1,150 @@ +package peer + +import ( + "encoding/binary" + "fmt" + "log" + "net" + "net/netip" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +const ( + controlPort = uint16(4561) + controlMsgLen = 7 // 1 type + 4 IPv4 + 2 port + msgYourEndpt = uint8(1) + msgMyEndpt = uint8(2) +) + +type ControlServer struct { + localPeerIP byte + network []byte + conn *net.UDPConn + hp *HolePunch + netName string +} + +func NewControlServer(g Globals, hp *HolePunch, netName string) (*ControlServer, error) { + vpnIP := netip.AddrFrom4([4]byte{ + g.Network[0], g.Network[1], g.Network[2], g.LocalPeerIP, + }) + listenAddr := net.UDPAddrFromAddrPort(netip.AddrPortFrom(vpnIP, controlPort)) + conn, err := net.ListenUDP("udp4", listenAddr) + if err != nil { + return nil, fmt.Errorf("listen on control port: %w", err) + } + return &ControlServer{ + localPeerIP: g.LocalPeerIP, + network: g.Network, + conn: conn, + hp: hp, + netName: netName, + }, nil +} + +func (cs *ControlServer) Run() { + buf := make([]byte, 64) + for { + n, src, err := cs.conn.ReadFromUDPAddrPort(buf) + if err != nil { + log.Printf("[Control] read: %v", err) + continue + } + if n < controlMsgLen { + continue + } + cs.handle(buf[:n], src) + } +} + +func (cs *ControlServer) handle(msg []byte, src netip.AddrPort) { + msgType, ep, ok := decodeControlMsg(msg) + if !ok { + return + } + switch msgType { + case msgYourEndpt: + cs.onYourEndpoint(ep) + case msgMyEndpt: + cs.onMyEndpoint(src, ep) + } +} + +// onYourEndpoint is called when the relay tells us our external WG endpoint. +// We broadcast MsgMyEndpoint to all known peers so they can attempt direct +// connections to us. +func (cs *ControlServer) onYourEndpoint(ourEndpoint netip.AddrPort) { + log.Printf("[Control] external endpoint: %v", ourEndpoint) + + state, err := loadNetworkState(cs.netName) + if err != nil { + log.Printf("[Control] load state: %v", err) + return + } + + for _, p := range state.Peers { + if p == nil || p.PeerIP == cs.localPeerIP || len(p.WGPubKey) != wgtypes.KeyLen { + continue + } + peerVPNIP := netip.AddrFrom4([4]byte{ + cs.network[0], cs.network[1], cs.network[2], p.PeerIP, + }) + cs.sendMsg(msgMyEndpt, peerVPNIP, ourEndpoint) + } +} + +// onMyEndpoint is called when a peer tells us their external WG endpoint. +// We start a probe-before-commit attempt to reach them directly. +func (cs *ControlServer) onMyEndpoint(src netip.AddrPort, theirEndpoint netip.AddrPort) { + peerIPByte := src.Addr().As4()[3] + + state, err := loadNetworkState(cs.netName) + if err != nil { + log.Printf("[Control] load state: %v", err) + return + } + + peer := state.Peers[peerIPByte] + if peer == nil || len(peer.WGPubKey) != wgtypes.KeyLen { + return + } + + pubKey, err := wgtypes.NewKey(peer.WGPubKey) + if err != nil { + return + } + + cs.hp.OnEndpointLearned(peerIPByte, pubKey, theirEndpoint, false) +} + +// SendYourEndpoint sends MsgYourEndpoint to a peer, informing them of their +// own external WG endpoint. Used by the relay's endpoint reporter (Phase 6). +func (cs *ControlServer) SendYourEndpoint(peerVPNIP netip.Addr, theirEndpoint netip.AddrPort) { + cs.sendMsg(msgYourEndpt, peerVPNIP, theirEndpoint) +} + +func (cs *ControlServer) sendMsg(msgType uint8, peerVPNIP netip.Addr, ep netip.AddrPort) { + dst := net.UDPAddrFromAddrPort(netip.AddrPortFrom(peerVPNIP, controlPort)) + if _, err := cs.conn.WriteTo(encodeControlMsg(msgType, ep), dst); err != nil { + log.Printf("[Control] send to %v: %v", peerVPNIP, err) + } +} + +func encodeControlMsg(msgType uint8, ep netip.AddrPort) []byte { + msg := make([]byte, controlMsgLen) + msg[0] = msgType + a4 := ep.Addr().Unmap().As4() + copy(msg[1:5], a4[:]) + binary.BigEndian.PutUint16(msg[5:7], ep.Port()) + return msg +} + +func decodeControlMsg(msg []byte) (msgType uint8, ep netip.AddrPort, ok bool) { + if len(msg) < controlMsgLen { + return 0, netip.AddrPort{}, false + } + ip := netip.AddrFrom4([4]byte(msg[1:5])) + port := binary.BigEndian.Uint16(msg[5:7]) + return msg[0], netip.AddrPortFrom(ip, port), true +} diff --git a/peer/device.go b/peer/device.go new file mode 100644 index 0000000..5627d69 --- /dev/null +++ b/peer/device.go @@ -0,0 +1,48 @@ +package peer + +import ( + "log" + "net/netip" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +func (a *App) devPeers() []wgtypes.Peer { + peers, err := a.dev.Peers() + if err != nil { + log.Fatalf("Failed to get peers %v: %v", a.vpnIP, err) + } + return peers +} + +func (a *App) devAddDirect(p *Peer, endpoint netip.AddrPort) { + if err := a.dev.AddDirect(p.PubKey, endpoint, p.VPNIP); err != nil { + log.Fatalf("Failed to add peer %v: %v", p.VPNIP, err) + } +} + +func (a *App) devSetRelay(p *Peer, endpoint netip.AddrPort) { + if err := a.dev.SetRelay(p.PubKey, endpoint, a.vpnNet); err != nil { + log.Fatalf("Failed to add relay %v: %v", p.VPNIP, err) + } +} + +func (a *App) devPromote(p *Peer) { + if err := a.dev.Promote(p.PubKey, p.VPNIP); err != nil { + log.Fatalf("Failed to promot peer %v: %v", p.VPNIP, err) + } +} + +func (a *App) devAddProbe(p *Peer, endpoint netip.AddrPort) { + if err := a.dev.AddProbe(p.PubKey, endpoint); err != nil { + log.Fatalf("Failed to add probe %v: %v", p.VPNIP, err) + } +} + +func (a *App) devRemove(p *Peer) { + if p.State != StateRelayed { + if err := a.dev.RemovePeer(p.PubKey); err != nil { + log.Fatalf("Failed to remove peer %v: %v", p.VPNIP, err) + } + } +} diff --git a/peer/endpointreporter.go b/peer/endpointreporter.go new file mode 100644 index 0000000..fe626b7 --- /dev/null +++ b/peer/endpointreporter.go @@ -0,0 +1,88 @@ +package peer + +import ( + "log" + "net/netip" + "time" + + "golang.zx2c4.com/wireguard/wgctrl" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +const reporterInterval = 5 * time.Second + +// EndpointReporter runs on relay peers only. It polls wgctrl every 5s and, +// for each peer whose LastHandshakeTime has changed, sends MsgYourEndpoint to +// that peer's VPN IP so it learns its own external WG endpoint. +type EndpointReporter struct { + client *wgctrl.Client + devName string + network []byte + netName string + control *ControlServer + lastTimes map[wgtypes.Key]time.Time +} + +func NewEndpointReporter(g Globals, cs *ControlServer, netName string) *EndpointReporter { + return &EndpointReporter{ + client: g.WGClient, + devName: g.WGDevName, + network: g.Network, + netName: netName, + control: cs, + lastTimes: make(map[wgtypes.Key]time.Time), + } +} + +func (er *EndpointReporter) Run() { + for range time.Tick(reporterInterval) { + er.poll() + } +} + +func (er *EndpointReporter) poll() { + dev, err := er.client.Device(er.devName) + if err != nil { + log.Printf("[EndpointReporter] get device: %v", err) + return + } + + state, err := loadNetworkState(er.netName) + if err != nil { + log.Printf("[EndpointReporter] load state: %v", err) + return + } + + // Build WGPubKey → VPN IP byte index from current network state. + keyToIP := make(map[wgtypes.Key]byte, len(dev.Peers)) + for _, p := range state.Peers { + if p == nil || len(p.WGPubKey) != wgtypes.KeyLen { + continue + } + key, err := wgtypes.NewKey(p.WGPubKey) + if err == nil { + keyToIP[key] = p.PeerIP + } + } + + for _, p := range dev.Peers { + if p.Endpoint == nil || p.LastHandshakeTime.IsZero() { + continue + } + if p.LastHandshakeTime == er.lastTimes[p.PublicKey] { + continue + } + + peerIPByte, ok := keyToIP[p.PublicKey] + if !ok { + continue + } + + er.lastTimes[p.PublicKey] = p.LastHandshakeTime + + peerVPNIP := netip.AddrFrom4([4]byte{ + er.network[0], er.network[1], er.network[2], peerIPByte, + }) + er.control.SendYourEndpoint(peerVPNIP, p.Endpoint.AddrPort()) + } +} diff --git a/peer/fake_wgdevice_test.go b/peer/fake_wgdevice_test.go new file mode 100644 index 0000000..b2651fa --- /dev/null +++ b/peer/fake_wgdevice_test.go @@ -0,0 +1,67 @@ +package peer + +import ( + "net/netip" + "sync" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +// fakeWGDevice records every call made to it. It is safe to read Calls after +// the event loop has processed the event under test (single-threaded loop +// means no extra synchronisation needed, but the mutex guards concurrent test +// helpers if needed). +type fakeWGDevice struct { + mu sync.Mutex + Calls []fakeCall + peers []wgtypes.Peer +} + +type fakeCall struct { + Method string + PubKey wgtypes.Key + Endpoint netip.AddrPort + VPNiP netip.Addr + Network netip.Prefix +} + +func (f *fakeWGDevice) record(c fakeCall) { + f.mu.Lock() + f.Calls = append(f.Calls, c) + f.mu.Unlock() +} + +func (f *fakeWGDevice) Name() string { return "wg-test" } + +func (f *fakeWGDevice) Peers() ([]wgtypes.Peer, error) { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]wgtypes.Peer, len(f.peers)) + copy(out, f.peers) + return out, nil +} + +func (f *fakeWGDevice) AddDirect(pubKey wgtypes.Key, endpoint netip.AddrPort, vpnIP netip.Addr) error { + f.record(fakeCall{Method: "AddDirect", PubKey: pubKey, Endpoint: endpoint, VPNiP: vpnIP}) + return nil +} + +func (f *fakeWGDevice) SetRelay(pubKey wgtypes.Key, endpoint netip.AddrPort, network netip.Prefix) error { + f.record(fakeCall{Method: "SetRelay", PubKey: pubKey, Endpoint: endpoint, Network: network}) + return nil +} + +func (f *fakeWGDevice) AddProbe(pubKey wgtypes.Key, endpoint netip.AddrPort) error { + f.record(fakeCall{Method: "AddProbe", PubKey: pubKey, Endpoint: endpoint}) + return nil +} + +func (f *fakeWGDevice) Promote(pubKey wgtypes.Key, vpnIP netip.Addr) error { + f.record(fakeCall{Method: "Promote", PubKey: pubKey, VPNiP: vpnIP}) + return nil +} + +func (f *fakeWGDevice) RemovePeer(pubKey wgtypes.Key) error { + f.record(fakeCall{Method: "RemovePeer", PubKey: pubKey}) + return nil +} diff --git a/peer/holepunch.go b/peer/holepunch.go new file mode 100644 index 0000000..2b7e96e --- /dev/null +++ b/peer/holepunch.go @@ -0,0 +1,155 @@ +package peer + +import ( + "log" + "net/netip" + "sync" + "time" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +const ( + probeWait = 15 * time.Second + backoffStart = 5 * time.Minute + backoffMax = time.Hour +) + +type probeState struct { + pubKey wgtypes.Key + endpoint netip.AddrPort + probing bool + direct bool + backoff time.Duration + backoffAt time.Time +} + +type HolePunch struct { + Globals + mu sync.Mutex + peers [256]*probeState +} + +func NewHolePunch(g Globals) *HolePunch { + return &HolePunch{Globals: g} +} + +// OnEndpointLearned is called when a peer's external WG endpoint becomes known, +// either from the VPN control channel (MsgMyEndpoint) or from the hub poller. +// fromHub=true resets any existing backoff so the hub-reported change is acted +// on immediately. +func (hp *HolePunch) OnEndpointLearned(peerIP byte, pubKey wgtypes.Key, endpoint netip.AddrPort, fromHub bool) { + hp.mu.Lock() + defer hp.mu.Unlock() + + ps := hp.peers[peerIP] + if ps == nil { + ps = &probeState{pubKey: pubKey} + hp.peers[peerIP] = ps + } + + if fromHub { + ps.backoff = 0 + ps.backoffAt = time.Time{} + ps.direct = false + ps.pubKey = pubKey + } + ps.endpoint = endpoint + + if ps.probing || ps.direct { + return + } + if ps.backoff > 0 && time.Now().Before(ps.backoffAt) { + return + } + + ps.probing = true + go hp.runProbe(peerIP) +} + +func (hp *HolePunch) runProbe(peerIP byte) { + hp.mu.Lock() + ps := hp.peers[peerIP] + if ps == nil { + hp.mu.Unlock() + return + } + pubKey := ps.pubKey + endpoint := ps.endpoint + hp.mu.Unlock() + + vpnIP := netip.AddrFrom4([4]byte{ + hp.Network[0], hp.Network[1], hp.Network[2], peerIP, + }) + + probeStart := time.Now() + if err := addProbeEntry(hp.WGClient, hp.WGDevName, pubKey, endpoint); err != nil { + log.Printf("[HolePunch] addProbeEntry peer %d: %v", peerIP, err) + hp.finishProbe(peerIP, false) + return + } + + time.Sleep(probeWait) + + _, handshakeTime, err := getPeerEndpoint(hp.WGClient, hp.WGDevName, pubKey) + if err == nil && handshakeTime.After(probeStart) { + if err := promoteToDirect(hp.WGClient, hp.WGDevName, pubKey, vpnIP); err != nil { + log.Printf("[HolePunch] promoteToDirect peer %d: %v", peerIP, err) + } + hp.finishProbe(peerIP, true) + return + } + + // Probe failed — remove entry and schedule backoff retry. + if err := removePeerEntry(hp.WGClient, hp.WGDevName, pubKey); err != nil { + log.Printf("[HolePunch] removePeerEntry peer %d: %v", peerIP, err) + } + + hp.mu.Lock() + ps = hp.peers[peerIP] + var delay time.Duration + if ps != nil { + if ps.backoff == 0 { + ps.backoff = backoffStart + } else { + ps.backoff = min(ps.backoff*2, backoffMax) + } + ps.backoffAt = time.Now().Add(ps.backoff) + ps.probing = false + delay = ps.backoff + } + hp.mu.Unlock() + + if delay > 0 { + go func() { + time.Sleep(delay) + hp.retryProbe(peerIP) + }() + } +} + +func (hp *HolePunch) finishProbe(peerIP byte, success bool) { + hp.mu.Lock() + defer hp.mu.Unlock() + ps := hp.peers[peerIP] + if ps == nil { + return + } + ps.probing = false + if success { + ps.direct = true + ps.backoff = 0 + } +} + +func (hp *HolePunch) retryProbe(peerIP byte) { + hp.mu.Lock() + ps := hp.peers[peerIP] + if ps == nil || ps.probing || ps.direct { + hp.mu.Unlock() + return + } + ps.probing = true + hp.mu.Unlock() + hp.runProbe(peerIP) +} diff --git a/peer/interfaces.go b/peer/interfaces.go new file mode 100644 index 0000000..be13ac5 --- /dev/null +++ b/peer/interfaces.go @@ -0,0 +1,18 @@ +package peer + +import ( + "net/netip" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +// WGDevice is the subset of wginterface.Device used by App. +type WGDevice interface { + Name() string + Peers() ([]wgtypes.Peer, error) + AddDirect(pubKey wgtypes.Key, endpoint netip.AddrPort, vpnIP netip.Addr) error + SetRelay(pubKey wgtypes.Key, endpoint netip.AddrPort, network netip.Prefix) error + AddProbe(pubKey wgtypes.Key, endpoint netip.AddrPort) error + Promote(pubKey wgtypes.Key, vpnIP netip.Addr) error + RemovePeer(pubKey wgtypes.Key) error +} diff --git a/peer/on_hub.go b/peer/on_hub.go new file mode 100644 index 0000000..3c83d86 --- /dev/null +++ b/peer/on_hub.go @@ -0,0 +1,115 @@ +package peer + +import ( + "log" + "net/netip" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" + + "vppn/peer/control" +) + +func (a *App) onAddPeer(p HubPeer) { + if _, exists := a.peersByKey[p.PubKey]; exists { + a.onRemovePeer(p.PubKey) + } + + peer := &Peer{ + PubKey: p.PubKey, + VPNIP: p.VPNIP, + IsRelay: p.IsRelay, + IsPublic: p.IsPublic, + Endpoint4: p.EndpointV4, + Endpoint6: p.EndpointV6, + Role: roleFor(a.isPublic, a.vpnIP, p), + State: StateRelayed, + } + + endpoint := peer.PreferredEndpoint() + if peer.IsPublic && !endpoint.IsValid() { + // The peer is misconfigured. + // TODO: Log here. + return + } + + a.peersByKey[p.PubKey] = peer + a.peersByIP[peer.VPNIP] = peer + + if !peer.IsPublic { + return + } + + peer.WGEndpoint = endpoint + peer.State = StateDirect + a.devAddDirect(peer, endpoint) +} + +func (a *App) onRemovePeer(key wgtypes.Key) { + peer, exists := a.peersByKey[key] + if !exists { + return + } + a.devRemove(peer) + delete(a.peersByKey, key) + delete(a.peersByIP, peer.VPNIP) + + if peer == a.relay { + a.relay = nil + a.switchActiveRelay() + } +} + +// switchActiveRelay promotes the lowest-latency relay peer to active. +func (a *App) switchActiveRelay() { + if a.relay != nil { + a.devAddDirect(a.relay, a.relay.WGEndpoint) + a.relay = nil + } + + var best *Peer + for _, p := range a.peersByKey { + if !p.CanRelay() { + continue + } + + if best == nil || betterRelay(p, best) { + best = p + } + } + if best == nil { + log.Printf("no relay available") + return + } + + a.devSetRelay(best, best.WGEndpoint) + a.relay = best +} + +// betterRelay reports whether a is a better relay candidate than b. +// Prefers lower RTT; treats zero RTT (no measurement yet) as worst case. +func betterRelay(a, b *Peer) bool { + if a.RTT == 0 { + return false + } + if b.RTT == 0 { + return true + } + return a.RTT < b.RTT +} + +func preferredEndpoint(v4, v6 netip.AddrPort) netip.AddrPort { + if v4.IsValid() { + return v4 + } + return v6 +} + +func roleFor(selfIsPublic bool, selfIP netip.Addr, p HubPeer) control.Role { + if !selfIsPublic && p.IsPublic { + return control.Client + } + if selfIsPublic && !p.IsPublic { + return control.Server + } + return control.RoleFor(selfIP, p.VPNIP) +} diff --git a/peer/on_multicast.go b/peer/on_multicast.go new file mode 100644 index 0000000..bb6c99d --- /dev/null +++ b/peer/on_multicast.go @@ -0,0 +1,27 @@ +package peer + +import "net/netip" + +func (a *App) onMulticastDiscovery(e MulticastEvent) { + if a.isPublic { + return + } + + peer, ok := a.peersByKey[e.pubKey] + if !ok { + return + } + + if peer.IsPublic || peer.State == StateDirect { + return + } + + var v4, v6 netip.AddrPort + if e.endpoint.Addr().Is4() { + v4 = e.endpoint + } else { + v6 = e.endpoint + } + + a.addProbe(peer, v4, v6) +} diff --git a/peer/on_ping.go b/peer/on_ping.go new file mode 100644 index 0000000..3bd41c6 --- /dev/null +++ b/peer/on_ping.go @@ -0,0 +1,66 @@ +package peer + +import ( + "net/netip" + "time" + + "vppn/peer/control" +) + +func (a *App) onPing(e PingEvent) { + peer, ok := a.peersByIP[e.srcVPNIP] + if !ok { + // TODO: Log here. + return + } + + now := time.Now() + peer.LastPing = now + peer.Up = true + + // If we're the server, respond. + if peer.Role == control.Server { + a.sendPing(peer, e.ping.ID, e.ping.PingTS) + } + + // Compute RTT from server echo. + if peer.Role == control.Client { + peer.RTT = now.Sub(time.Unix(0, e.ping.PingTS)) + } + + // If we're public, nothing more to do. + if a.isPublic { + return + } + + // We can only learn our own endpoint from directly-connected peers — Dst + // is the sender's observation of our WG handshake source. + if peer.State == StateDirect { + if dst := e.ping.Dst; dst.IsValid() { + if dst.Addr().Is4() { + a.selfV4 = dst + } else { + a.selfV6 = dst + } + } + return + } + + a.addProbe(peer, e.ping.SrcV4, e.ping.SrcV6) +} + +func (a *App) addProbe(peer *Peer, v4, v6 netip.AddrPort) { + endpoint := preferredEndpoint(v4, v6) + if !endpoint.IsValid() || endpoint == peer.WGEndpoint { + return + } + + peer.Endpoint4 = v4 + peer.Endpoint6 = v6 + peer.WGEndpoint = endpoint + + if peer.State == StateRelayed { + peer.State = StateProbing + } + a.devAddProbe(peer, endpoint) +} diff --git a/peer/on_tick.go b/peer/on_tick.go new file mode 100644 index 0000000..5673162 --- /dev/null +++ b/peer/on_tick.go @@ -0,0 +1,50 @@ +package peer + +import ( + "log" + "time" + + "vppn/peer/control" +) + +func (a *App) onTick() { + wgPeers := a.devPeers() + + a.nextPingID++ + now := time.Now().UnixNano() + + // Update Up values. + for _, p := range a.peersByKey { + p.Up = p.Alive() + } + + for _, wgPeer := range wgPeers { + p, ok := a.peersByKey[wgPeer.PublicKey] + if !ok { + log.Fatalf("Wireguard peer not in index: %v", wgPeer) + } + + // Send pings to peers where we're the client. + if p.Role == control.Client { + a.sendPing(p, a.nextPingID, now) + } + + // Promote probing peers to direct once alive (direct path confirmed + // working). + if p.State == StateProbing && time.Since(wgPeer.LastHandshakeTime) < 2*PingInterval { + p.State = StateDirect + a.devAddDirect(p, p.WGEndpoint) + } + + // Demote stale non-public direct peers back to probing. + if p.State == StateDirect && !p.IsPublic && !p.Up { + p.State = StateProbing + a.devAddProbe(p, p.WGEndpoint) + } + } + + // Ensure we have a live relay. + if a.relay == nil || !a.relay.Up { + a.switchActiveRelay() + } +} diff --git a/peer/ping.go b/peer/ping.go new file mode 100644 index 0000000..5126ebf --- /dev/null +++ b/peer/ping.go @@ -0,0 +1,9 @@ +package peer + +import "net/netip" + +func (a *App) sendPing(p *Peer, id, ts int64) { + _ = id + _ = ts + _ = netip.AddrPort{} +} diff --git a/peer/remote.go b/peer/remote.go new file mode 100644 index 0000000..f8553fd --- /dev/null +++ b/peer/remote.go @@ -0,0 +1,46 @@ +package peer + +import ( + "net/netip" + "time" + + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" + + "vppn/peer/control" +) + +type PeerState string + +const ( + StateRelayed = PeerState("RELAY") + StateProbing = PeerState("PROBE") + StateDirect = PeerState("DIRECT") +) + +type Peer struct { + PubKey wgtypes.Key // WireGuard public key. + VPNIP netip.Addr // VPN IP address. + IsRelay bool // Peer is a relay. + IsPublic bool // Peer has a public IP. + Endpoint4 netip.AddrPort // Reported IPv4 endpoint. + Endpoint6 netip.AddrPort // Reported IPv6 endpoint. + ObservedEndpoint netip.AddrPort // If we're public: WG handshake source endpoint. + LastPing time.Time // Time of last ping received. + RTT time.Duration // Round-trip time. + Role control.Role // Client initiates pings; server responds. + State PeerState // Current connection state. + Up bool // Is the peer alive. + WGEndpoint netip.AddrPort // Endpoint currently in WG (direct/probe). +} + +func (p *Peer) Alive() bool { + return time.Since(p.LastPing) < TimeoutInterval +} + +func (p *Peer) CanRelay() bool { + return p.IsRelay && p.Up && p.WGEndpoint.IsValid() +} + +func (p *Peer) PreferredEndpoint() netip.AddrPort { + return preferredEndpoint(p.Endpoint4, p.Endpoint6) +} diff --git a/peer/wgdev.go b/peer/wgdev.go new file mode 100644 index 0000000..b537684 --- /dev/null +++ b/peer/wgdev.go @@ -0,0 +1,177 @@ +package peer + +import ( + "fmt" + "net" + "net/netip" + "os" + "time" + + "github.com/vishvananda/netlink" + "golang.zx2c4.com/wireguard/wgctrl" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +func createWGDevice(name string, privKey wgtypes.Key, listenPort int, vpnIP netip.Addr, network []byte) (*wgctrl.Client, error) { + if len(network) != 4 { + return nil, fmt.Errorf("expected 4-byte network, got %d", len(network)) + } + + la := netlink.NewLinkAttrs() + la.Name = name + if err := netlink.LinkAdd(&netlink.GenericLink{LinkAttrs: la, LinkType: "wireguard"}); err != nil { + return nil, fmt.Errorf("add wireguard link: %w", err) + } + + link, err := netlink.LinkByName(name) + if err != nil { + _ = destroyWGDevice(name) + return nil, fmt.Errorf("get wireguard link: %w", err) + } + + a4 := vpnIP.As4() + if err := netlink.AddrAdd(link, &netlink.Addr{ + IPNet: &net.IPNet{ + IP: net.IP(a4[:]), + Mask: net.CIDRMask(24, 32), + }, + }); err != nil { + _ = destroyWGDevice(name) + return nil, fmt.Errorf("add VPN address: %w", err) + } + + if err := netlink.LinkSetUp(link); err != nil { + _ = destroyWGDevice(name) + return nil, fmt.Errorf("set link up: %w", err) + } + + client, err := wgctrl.New() + if err != nil { + _ = destroyWGDevice(name) + return nil, fmt.Errorf("new wgctrl client: %w", err) + } + + cfg := wgtypes.Config{ + PrivateKey: &privKey, + ListenPort: &listenPort, + } + if err := client.ConfigureDevice(name, cfg); err != nil { + client.Close() + _ = destroyWGDevice(name) + return nil, fmt.Errorf("configure wireguard: %w", err) + } + + return client, nil +} + +func destroyWGDevice(name string) error { + link, err := netlink.LinkByName(name) + if err != nil { + return fmt.Errorf("get link %q: %w", name, err) + } + return netlink.LinkDel(link) +} + +// applyBaseConfig adds the relay peer with /24 AllowedIPs, making it the +// fallback route for all VPN traffic. +func applyBaseConfig(client *wgctrl.Client, devName string, relayPubKey wgtypes.Key, relayEndpoint netip.AddrPort, network []byte) error { + if len(network) != 4 { + return fmt.Errorf("expected 4-byte network, got %d", len(network)) + } + keepalive := 25 * time.Second + cfg := wgtypes.Config{ + Peers: []wgtypes.PeerConfig{{ + PublicKey: relayPubKey, + Endpoint: net.UDPAddrFromAddrPort(relayEndpoint), + AllowedIPs: []net.IPNet{{ + IP: net.IP{network[0], network[1], network[2], 0}, + Mask: net.CIDRMask(24, 32), + }}, + ReplaceAllowedIPs: true, + PersistentKeepaliveInterval: &keepalive, + }}, + } + return client.ConfigureDevice(devName, cfg) +} + +// addProbeEntry adds a peer with no AllowedIPs and a 5s keepalive so WireGuard +// attempts handshakes without routing any traffic through it yet. +func addProbeEntry(client *wgctrl.Client, devName string, pubKey wgtypes.Key, endpoint netip.AddrPort) error { + keepalive := 5 * time.Second + cfg := wgtypes.Config{ + Peers: []wgtypes.PeerConfig{{ + PublicKey: pubKey, + Endpoint: net.UDPAddrFromAddrPort(endpoint), + AllowedIPs: []net.IPNet{}, + ReplaceAllowedIPs: true, + PersistentKeepaliveInterval: &keepalive, + }}, + } + return client.ConfigureDevice(devName, cfg) +} + +// addDirectPeer adds a peer with a known endpoint and /32 AllowedIPs in one +// step, for use when the hub reports a peer with a stable public endpoint. +func addDirectPeer(client *wgctrl.Client, devName string, pubKey wgtypes.Key, endpoint netip.AddrPort, vpnIP netip.Addr) error { + a4 := vpnIP.As4() + cfg := wgtypes.Config{ + Peers: []wgtypes.PeerConfig{{ + PublicKey: pubKey, + Endpoint: net.UDPAddrFromAddrPort(endpoint), + AllowedIPs: []net.IPNet{{ + IP: net.IP(a4[:]), + Mask: net.CIDRMask(32, 32), + }}, + ReplaceAllowedIPs: true, + }}, + } + return client.ConfigureDevice(devName, cfg) +} + +// promoteToDirect upgrades a probe entry to a /32 AllowedIPs entry, causing +// WireGuard to prefer the direct path over the relay's /24 route. +func promoteToDirect(client *wgctrl.Client, devName string, pubKey wgtypes.Key, vpnIP netip.Addr) error { + a4 := vpnIP.As4() + cfg := wgtypes.Config{ + Peers: []wgtypes.PeerConfig{{ + PublicKey: pubKey, + AllowedIPs: []net.IPNet{{ + IP: net.IP(a4[:]), + Mask: net.CIDRMask(32, 32), + }}, + ReplaceAllowedIPs: true, + }}, + } + return client.ConfigureDevice(devName, cfg) +} + +func removePeerEntry(client *wgctrl.Client, devName string, pubKey wgtypes.Key) error { + cfg := wgtypes.Config{ + Peers: []wgtypes.PeerConfig{{ + PublicKey: pubKey, + Remove: true, + }}, + } + return client.ConfigureDevice(devName, cfg) +} + +func enableForwarding(ifaceName string) error { + path := fmt.Sprintf("/proc/sys/net/ipv4/conf/%s/forwarding", ifaceName) + return os.WriteFile(path, []byte("1\n"), 0644) +} + +func getPeerEndpoint(client *wgctrl.Client, devName string, pubKey wgtypes.Key) (netip.AddrPort, time.Time, error) { + dev, err := client.Device(devName) + if err != nil { + return netip.AddrPort{}, time.Time{}, fmt.Errorf("get device: %w", err) + } + for _, p := range dev.Peers { + if p.PublicKey == pubKey { + if p.Endpoint == nil { + return netip.AddrPort{}, p.LastHandshakeTime, nil + } + return p.Endpoint.AddrPort(), p.LastHandshakeTime, nil + } + } + return netip.AddrPort{}, time.Time{}, fmt.Errorf("peer %v not found in device %s", pubKey, devName) +} diff --git a/peer/wginterface/manage.go b/peer/wginterface/manage.go index bdb1b5b..07f4e57 100644 --- a/peer/wginterface/manage.go +++ b/peer/wginterface/manage.go @@ -48,6 +48,11 @@ func (d *Device) Close() error { return d.client.Close() } +// Name returns the interface name. +func (d *Device) Name() string { + return d.name +} + // Configure sets the device's private key and UDP listen port. func (d *Device) Configure(privKey wgtypes.Key, listenPort int) error { return d.client.ConfigureDevice(d.name, wgtypes.Config{