Cleanup - maybe working...

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)
 }

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

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;

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) {

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 {

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)

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 {
+				if p.State != StateRelayed {
-					t.Fatalf("state = %v, want StateRelayed", p.State())
+					t.Fatalf("state = %v, want StateRelayed", p.State)
 				}
 				dev.AssertAddPeer(t, 0, key)
 			},

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
 	}
 

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

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 {
+			if time.Since(p.LastHandshakeTime()) < 2*wginterface.ProbeKeepalive {
-				a.devAddDirect(p, p.WGEndpoint())
+				a.devPromote(p)
 			}
 
 		case StateDirect:
@@ -43,8 +43,8 @@ func (a *App) onTick() {
 		}
 	}
 
-	// Ensure we have a live relay.
+	// Ensure we have a live relay (if we're not public).
-	if a.relay == nil || !a.relay.Up() {
+	if !a.isPublic && (a.relay == nil || !a.relay.Up()) {
 		a.switchActiveRelay()
 	}
 }

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 {

peer/wginterface/manage.go

@@ -23,10 +23,9 @@ const (
 	SessionTimeout = 180 * time.Second
 )
 
-var (
+const ProbeKeepalive = 8 * time.Second
-	probeKeepalive = 5 * time.Second
+
-	zeroKeepalive  = time.Duration(0)
+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,
 		}},
 	})
 }