From b0ff07aad66d4d801fb00803dbe2f88112e1e630 Mon Sep 17 00:00:00 2001 From: jdl Date: Tue, 9 Jun 2026 20:17:21 +0200 Subject: [PATCH] Cleanup - maybe working... --- cmd/vppn/main.go | 39 ++++++++++++++++++++------ hub/api/api.go | 10 +++++++ hub/api/migrations/2024-11-30-init.sql | 2 +- peer/device.go | 5 ++++ peer/multicast.go | 5 +++- peer/new.go | 1 + peer/on_hub_test.go | 4 +-- peer/on_multicast.go | 2 +- peer/on_ping.go | 2 +- peer/on_tick.go | 10 +++---- peer/remote.go | 11 +------- peer/wginterface/manage.go | 10 +++---- 12 files changed, 66 insertions(+), 35 deletions(-) diff --git a/cmd/vppn/main.go b/cmd/vppn/main.go index bd16125..8dd105f 100644 --- a/cmd/vppn/main.go +++ b/cmd/vppn/main.go @@ -5,8 +5,11 @@ import ( "log" "os" "path/filepath" + "strings" "vppn/peer" + + "git.crumpington.com/lib/go/flock" ) func main() { @@ -14,24 +17,34 @@ func main() { name := flag.String("name", "", "network name (required)") hub := flag.String("hub", "", "hub base URL (required)") - apiKey := flag.String("api-key", "", "API key (required)") flag.Parse() - if *name == "" || *hub == "" || *apiKey == "" { + if *name == "" || *hub == "" { flag.Usage() os.Exit(1) } - // TODO: Acquire flock on lock file. + apiKey, err := loadAPIKey(*name) + if err != nil { + log.Fatalf("api key: %v", err) + } - statePath := networkStatePath(*name) + // Directory existence is guaranteed by the apikey file read above. + lockFile, err := flock.TryLock(vppnPath(*name, "lock")) + if err != nil { + log.Fatalf("lock: %v", err) + } + if lockFile == nil { + log.Fatalf("already running for network %q", *name) + } + defer flock.Unlock(lockFile) - state, err := peer.LoadOrInit(statePath, *hub, *apiKey) + state, err := peer.LoadOrInit(vppnPath(*name, "state.json"), *hub, apiKey) if err != nil { log.Fatalf("init: %v", err) } - app, err := peer.New(state, *hub, *apiKey, *name) + app, err := peer.New(state, *hub, apiKey, *name) if err != nil { log.Fatalf("start: %v", err) } @@ -41,10 +54,18 @@ func main() { } } -func networkStatePath(name string) string { +func loadAPIKey(name string) (string, error) { + data, err := os.ReadFile(vppnPath(name, "apikey")) + if err != nil { + return "", err + } + return strings.TrimSpace(string(data)), nil +} + +func vppnPath(name, file string) string { home, err := os.UserHomeDir() if err != nil { - return filepath.Join(".vppn", name, "state.json") + return filepath.Join(".vppn", name, file) } - return filepath.Join(home, ".vppn", name, "state.json") + return filepath.Join(home, ".vppn", name, file) } diff --git a/hub/api/api.go b/hub/api/api.go index 7b656a6..bc62a77 100644 --- a/hub/api/api.go +++ b/hub/api/api.go @@ -157,6 +157,16 @@ func (a *API) Peer_Init(peer *Peer, args m.PeerInitArgs) error { a.lock.Lock() defer a.lock.Unlock() + // Re-read from DB inside the lock — the caller's copy was fetched before + // we held the lock, so it may be stale under concurrent requests. + current, err := db.Peer_Get(a.db, peer.NetworkID, peer.PeerIP) + if err != nil { + return err + } + if len(current.WGPubKey) != 0 { + return errors.New("peer already initialized") + } + peer.Version = idgen.NextID(0) peer.WGPubKey = args.WGPubKey peer.SignPubKey = args.SignPubKey diff --git a/hub/api/migrations/2024-11-30-init.sql b/hub/api/migrations/2024-11-30-init.sql index f2f2a79..63d9d45 100644 --- a/hub/api/migrations/2024-11-30-init.sql +++ b/hub/api/migrations/2024-11-30-init.sql @@ -20,6 +20,6 @@ CREATE TABLE peers ( Port INTEGER NOT NULL, Relay INTEGER NOT NULL DEFAULT 0, -- Boolean if peer will forward packets. WGPubKey BLOB NOT NULL, - SignPubKey BLOB NOT NULL + SignPubKey BLOB NOT NULL, PRIMARY KEY(NetworkID, PeerIP) ) WITHOUT ROWID; diff --git a/peer/device.go b/peer/device.go index e175278..6630813 100644 --- a/peer/device.go +++ b/peer/device.go @@ -39,22 +39,27 @@ func (a *App) devPeers() []wgtypes.Peer { func (a *App) devAddPeer(p *Peer) { devRetry(p.VPNIP, "AddPeer", func() error { return a.dev.AddPeer(p.PubKey()) }) + p.State = StateRelayed } func (a *App) devAddDirect(p *Peer, endpoint netip.AddrPort) { devRetry(p.VPNIP, "AddDirect", func() error { return a.dev.AddDirect(p.PubKey(), endpoint, p.VPNIP) }) + p.State = StateDirect } func (a *App) devSetRelay(p *Peer, endpoint netip.AddrPort) { devRetry(p.VPNIP, "SetRelay", func() error { return a.dev.SetRelay(p.PubKey(), endpoint, a.vpnNet) }) + p.State = StateDirect } func (a *App) devPromote(p *Peer) { devRetry(p.VPNIP, "Promote", func() error { return a.dev.Promote(p.PubKey(), p.VPNIP) }) + p.State = StateDirect } func (a *App) devAddProbe(p *Peer, endpoint netip.AddrPort) { devRetry(p.VPNIP, "AddProbe", func() error { return a.dev.AddProbe(p.PubKey(), endpoint) }) + p.State = StateProbing } func (a *App) devRemove(p *Peer) { diff --git a/peer/multicast.go b/peer/multicast.go index 01d225b..648f9db 100644 --- a/peer/multicast.go +++ b/peer/multicast.go @@ -13,7 +13,7 @@ import ( ) const ( - mcBeaconLen = 35 // 1 VPN IP byte + 32 WG pubkey + 2 WG listen port + mcBeaconLen = 35 // 1 VPN IP byte + 32 WG pubkey + 2 WG listen port mcSignedBeaconLen = sign.Overhead + mcBeaconLen // 64-byte nacl/sign prefix + payload mcBroadcastInterval = 32 * time.Second mcErrorRetryInterval = 16 * time.Second @@ -78,6 +78,9 @@ func runMCReaderInner(vpnNet netip.Prefix, selfVPNIP netip.Addr, ch chan<- Multi conn.SetReadDeadline(time.Now().Add(32 * time.Second)) n, src, err := conn.ReadFromUDPAddrPort(buf) if err != nil { + if ne, ok := err.(net.Error); ok && ne.Timeout() { + continue + } return fmt.Errorf("read: %w", err) } if n != mcSignedBeaconLen { diff --git a/peer/new.go b/peer/new.go index ffe0154..8795e37 100644 --- a/peer/new.go +++ b/peer/new.go @@ -17,6 +17,7 @@ func New( hubURL, apiKey string, ifaceName string, ) (*App, error) { + a4 := state.VPNIP.As4() if err := wginterface.Create(ifaceName, a4[:], 24); err != nil { return nil, fmt.Errorf("create WG interface: %w", err) diff --git a/peer/on_hub_test.go b/peer/on_hub_test.go index b02d3db..98c7c17 100644 --- a/peer/on_hub_test.go +++ b/peer/on_hub_test.go @@ -41,8 +41,8 @@ func TestOnAddPeer(t *testing.T) { if a.peersByIP[peerVPNIP] == nil { t.Fatal("not in peersByIP") } - if p.State() != StateRelayed { - t.Fatalf("state = %v, want StateRelayed", p.State()) + if p.State != StateRelayed { + t.Fatalf("state = %v, want StateRelayed", p.State) } dev.AssertAddPeer(t, 0, key) }, diff --git a/peer/on_multicast.go b/peer/on_multicast.go index 000a375..1476443 100644 --- a/peer/on_multicast.go +++ b/peer/on_multicast.go @@ -28,7 +28,7 @@ func (a *App) onMulticastDiscovery(e MulticastEvent) { return } - if peer.IsPublic || peer.State() == StateDirect { + if peer.IsPublic || peer.State == StateDirect { return } diff --git a/peer/on_ping.go b/peer/on_ping.go index 48e35e2..2be8d0a 100644 --- a/peer/on_ping.go +++ b/peer/on_ping.go @@ -34,7 +34,7 @@ func (a *App) onPing(e PingEvent) { // 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 peer.State == StateDirect { if dst := e.ping.Dst; dst.IsValid() { if dst.Addr().Is4() { a.selfV4 = dst diff --git a/peer/on_tick.go b/peer/on_tick.go index 3786df3..c258cb0 100644 --- a/peer/on_tick.go +++ b/peer/on_tick.go @@ -27,12 +27,12 @@ func (a *App) onTick() { a.sendPing(p, now) } - switch p.State() { + switch p.State { case StateProbing: // Promote probing peers to direct once alive (direct path confirmed // working). - if time.Since(p.LastHandshakeTime()) < wginterface.SessionTimeout { - a.devAddDirect(p, p.WGEndpoint()) + if time.Since(p.LastHandshakeTime()) < 2*wginterface.ProbeKeepalive { + a.devPromote(p) } case StateDirect: @@ -43,8 +43,8 @@ func (a *App) onTick() { } } - // Ensure we have a live relay. - if a.relay == nil || !a.relay.Up() { + // Ensure we have a live relay (if we're not public). + if !a.isPublic && (a.relay == nil || !a.relay.Up()) { a.switchActiveRelay() } } diff --git a/peer/remote.go b/peer/remote.go index d7fd56d..0e8924a 100644 --- a/peer/remote.go +++ b/peer/remote.go @@ -26,6 +26,7 @@ type Peer struct { Endpoint4 netip.AddrPort // Reported IPv4 endpoint. Endpoint6 netip.AddrPort // Reported IPv6 endpoint. RTT time.Duration // Round-trip time. + State PeerState // Current routing state; updated on each devXxx call. Role control.Role // Client initiates pings; server responds. SignPubKey [32]byte // nacl/sign public key for verifying multicast beacons. } @@ -35,16 +36,6 @@ func (p *Peer) PubKey() wgtypes.Key { return p.wgPeer.PublicKey } -func (p *Peer) State() PeerState { - if len(p.wgPeer.AllowedIPs) > 0 { - return StateDirect - } - if p.wgPeer.Endpoint == nil { - return StateRelayed - } - return StateProbing -} - func (p *Peer) WGEndpoint() netip.AddrPort { ep := p.wgPeer.Endpoint if ep == nil { diff --git a/peer/wginterface/manage.go b/peer/wginterface/manage.go index 90789cd..4f8f9fe 100644 --- a/peer/wginterface/manage.go +++ b/peer/wginterface/manage.go @@ -23,10 +23,9 @@ const ( SessionTimeout = 180 * time.Second ) -var ( - probeKeepalive = 5 * time.Second - zeroKeepalive = time.Duration(0) -) +const ProbeKeepalive = 8 * time.Second + +var zeroKeepalive = time.Duration(0) // Device wraps a wgctrl client bound to a named WireGuard interface. type Device struct { @@ -116,13 +115,14 @@ func (d *Device) SetRelay(pubKey wgtypes.Key, endpoint netip.AddrPort, network n // AddProbe adds a peer with no AllowedIPs and a 5s keepalive. WireGuard will // attempt handshakes without routing any traffic through this peer yet. func (d *Device) AddProbe(pubKey wgtypes.Key, endpoint netip.AddrPort) error { + keepalive := ProbeKeepalive return d.client.ConfigureDevice(d.name, wgtypes.Config{ Peers: []wgtypes.PeerConfig{{ PublicKey: pubKey, Endpoint: net.UDPAddrFromAddrPort(endpoint), AllowedIPs: []net.IPNet{}, ReplaceAllowedIPs: true, - PersistentKeepaliveInterval: &probeKeepalive, + PersistentKeepaliveInterval: &keepalive, }}, }) }