From b18f6aee1bf9f3ef3def8aae714039a95c052f1a Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Fri, 3 Jul 2026 23:32:39 +0200 Subject: [PATCH] =?UTF-8?q?hub:=20S1=20store=20=E2=80=94=20wg=5Fendpoints/?= =?UTF-8?q?wg=5Fpeers=20tables=20+=20/32=20allocator=20(doc=2006=20=C2=A73?= =?UTF-8?q?.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Additive migration; AddWGPeer = one tx, idempotent on pubkey, lowest-free-host allocation skipping network/pbs_tunnel_ip/broadcast, UNIQUE(assigned_ip) race backstop with one internal retry; typed ErrWGEndpointUnset/ErrWGSubnetExhausted. Group-A tests + red-proof (allocator-ignores-rows mutation fails 3 tests). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6 --- hub/internal/store/store.go | 29 +++++ hub/internal/store/wg.go | 218 ++++++++++++++++++++++++++++++++++ hub/internal/store/wg_test.go | 155 ++++++++++++++++++++++++ 3 files changed, 402 insertions(+) create mode 100644 hub/internal/store/wg.go create mode 100644 hub/internal/store/wg_test.go diff --git a/hub/internal/store/store.go b/hub/internal/store/store.go index 18c595f..fec6980 100644 --- a/hub/internal/store/store.go +++ b/hub/internal/store/store.go @@ -367,6 +367,35 @@ func (s *Store) migrate() error { return err } + // S1 offsite connectivity (doc 06 §3.2): the WG endpoint record + peer registry. The hub is the + // source of truth; the endpoint converges on it (SSH push, internal/wgsync). No `status` column + // on wg_peers — presence in the table IS the desired state; the doc-06 `status` field belongs to + // the S2 host-join. assigned_ip stores the BARE IP (no /32 — that's presentation, appended by + // the API layer and the sync payload). UNIQUE(assigned_ip) is the allocator's race backstop. + _, err = s.db.Exec(` + CREATE TABLE IF NOT EXISTS wg_endpoints ( + endpoint_id TEXT PRIMARY KEY, + dns_name TEXT NOT NULL, + wg_port INTEGER NOT NULL, + server_pubkey TEXT NOT NULL, + tunnel_subnet TEXT NOT NULL, + pbs_tunnel_ip TEXT NOT NULL, + created_at DATETIME NOT NULL DEFAULT (datetime('now')), + updated_at DATETIME NOT NULL DEFAULT (datetime('now')) + ); + CREATE TABLE IF NOT EXISTS wg_peers ( + pubkey TEXT PRIMARY KEY, + assigned_ip TEXT NOT NULL UNIQUE, + host_id TEXT NOT NULL DEFAULT '', + note TEXT NOT NULL DEFAULT '', + created_at DATETIME NOT NULL DEFAULT (datetime('now')), + updated_at DATETIME NOT NULL DEFAULT (datetime('now')) + ); + `) + if err != nil { + return err + } + return nil } diff --git a/hub/internal/store/wg.go b/hub/internal/store/wg.go new file mode 100644 index 0000000..7326d47 --- /dev/null +++ b/hub/internal/store/wg.go @@ -0,0 +1,218 @@ +package store + +// S1 offsite connectivity (doc 06 §3.2): the WG endpoint record + the per-peer /32 allocator. +// The hub DB is the source of truth for the endpoint's peer list; internal/wgsync pushes the +// FULL list to the endpoint (declarative — the endpoint converges, drift is erased). These +// methods never touch SSH; they are pure persistence + allocation. + +import ( + "database/sql" + "errors" + "fmt" + "net/netip" + "strings" +) + +// ErrWGEndpointUnset is returned by AddWGPeer when no endpoint record exists yet — allocation +// needs the tunnel subnet, so peers cannot be registered before PUT /admin/wg/endpoint. +var ErrWGEndpointUnset = errors.New("wg endpoint not configured") + +// ErrWGSubnetExhausted is returned when the tunnel subnet has no free host address left. +var ErrWGSubnetExhausted = errors.New("tunnel subnet exhausted") + +// WGEndpoint is the public endpoint's coordinates (doc 06 §3.2 "endpoint record"). One row +// expected ("ep0") for now; the schema allows more for a later multi-endpoint world. +type WGEndpoint struct { + EndpointID string + DNSName string + WGPort int + ServerPubkey string + TunnelSubnet string // CIDR, e.g. "10.77.0.0/24" + PBSTunnelIP string // the endpoint's own in-tunnel address, e.g. "10.77.0.1" +} + +// WGPeer is one registered peer (doc 06 §3.2 "per-host peer entry"). HostID is inert until the +// S2 host-join; presence of the row IS the desired state (no status column by design). +type WGPeer struct { + Pubkey string + AssignedIP string // bare IP — the /32 suffix is presentation, not storage + HostID string + Note string +} + +// SetWGEndpoint upserts the (single expected) endpoint record. +func (s *Store) SetWGEndpoint(e *WGEndpoint) error { + if e.EndpointID == "" { + e.EndpointID = "ep0" + } + _, err := s.db.Exec(` + INSERT INTO wg_endpoints (endpoint_id, dns_name, wg_port, server_pubkey, tunnel_subnet, pbs_tunnel_ip) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(endpoint_id) DO UPDATE SET + dns_name = excluded.dns_name, wg_port = excluded.wg_port, + server_pubkey = excluded.server_pubkey, tunnel_subnet = excluded.tunnel_subnet, + pbs_tunnel_ip = excluded.pbs_tunnel_ip, updated_at = datetime('now')`, + e.EndpointID, e.DNSName, e.WGPort, e.ServerPubkey, e.TunnelSubnet, e.PBSTunnelIP) + return err +} + +// GetWGEndpoint returns the endpoint record, or sql.ErrNoRows when unset. With multiple rows +// (not expected in S1) the lowest endpoint_id wins, deterministically. +func (s *Store) GetWGEndpoint() (*WGEndpoint, error) { + var e WGEndpoint + err := s.db.QueryRow(` + SELECT endpoint_id, dns_name, wg_port, server_pubkey, tunnel_subnet, pbs_tunnel_ip + FROM wg_endpoints ORDER BY endpoint_id LIMIT 1`). + Scan(&e.EndpointID, &e.DNSName, &e.WGPort, &e.ServerPubkey, &e.TunnelSubnet, &e.PBSTunnelIP) + if err != nil { + return nil, err + } + return &e, nil +} + +// AddWGPeer registers a peer pubkey and allocates the lowest free host address in the tunnel +// subnet — skipping the network address, the endpoint's own pbs_tunnel_ip, and (v4) the +// broadcast. Idempotent: an already-registered pubkey returns its existing IP with existed=true +// and allocates nothing. One transaction; the UNIQUE(assigned_ip) constraint backstops a +// concurrent-allocation race (the loser retries once internally). +func (s *Store) AddWGPeer(pubkey, hostID, note string) (assignedIP string, existed bool, err error) { + for attempt := 0; attempt < 2; attempt++ { + assignedIP, existed, err = s.addWGPeerOnce(pubkey, hostID, note) + if err != nil && attempt == 0 && strings.Contains(err.Error(), "UNIQUE constraint failed: wg_peers.assigned_ip") { + continue // race with a concurrent insert — re-read the assigned set and retry once + } + return assignedIP, existed, err + } + return assignedIP, existed, err +} + +func (s *Store) addWGPeerOnce(pubkey, hostID, note string) (string, bool, error) { + tx, err := s.db.Begin() + if err != nil { + return "", false, err + } + defer tx.Rollback() + + // Idempotency: existing pubkey → its IP, no allocation. + var existingIP string + err = tx.QueryRow(`SELECT assigned_ip FROM wg_peers WHERE pubkey = ?`, pubkey).Scan(&existingIP) + if err == nil { + return existingIP, true, nil + } + if err != sql.ErrNoRows { + return "", false, err + } + + var subnet, pbsIP string + err = tx.QueryRow(`SELECT tunnel_subnet, pbs_tunnel_ip FROM wg_endpoints ORDER BY endpoint_id LIMIT 1`). + Scan(&subnet, &pbsIP) + if err == sql.ErrNoRows { + return "", false, ErrWGEndpointUnset + } + if err != nil { + return "", false, err + } + + taken := map[string]bool{} + rows, err := tx.Query(`SELECT assigned_ip FROM wg_peers`) + if err != nil { + return "", false, err + } + for rows.Next() { + var ip string + if err := rows.Scan(&ip); err != nil { + rows.Close() + return "", false, err + } + taken[ip] = true + } + rows.Close() + if err := rows.Err(); err != nil { + return "", false, err + } + + ip, err := lowestFreeHost(subnet, pbsIP, taken) + if err != nil { + return "", false, err + } + + if _, err := tx.Exec(`INSERT INTO wg_peers (pubkey, assigned_ip, host_id, note) VALUES (?, ?, ?, ?)`, + pubkey, ip, hostID, note); err != nil { + return "", false, err + } + if err := tx.Commit(); err != nil { + return "", false, err + } + return ip, false, nil +} + +// lowestFreeHost walks the subnet's host addresses from lowest and returns the first not in +// `taken`, skipping the network address, the endpoint's own address, and (v4) the broadcast. +func lowestFreeHost(subnet, endpointIP string, taken map[string]bool) (string, error) { + prefix, err := netip.ParsePrefix(subnet) + if err != nil { + return "", fmt.Errorf("bad tunnel_subnet %q: %w", subnet, err) + } + prefix = prefix.Masked() + last := lastAddr(prefix) + for a := prefix.Addr().Next(); prefix.Contains(a); a = a.Next() { + if a.Is4() && a == last { + break // v4 broadcast — never allocated + } + ip := a.String() + if ip == endpointIP || taken[ip] { + continue + } + return ip, nil + } + return "", ErrWGSubnetExhausted +} + +// lastAddr returns the highest address in the prefix (the v4 broadcast). +func lastAddr(p netip.Prefix) netip.Addr { + b := p.Addr().AsSlice() + bits := p.Bits() + for i := range b { + hostBits := 8*(i+1) - bits + if hostBits <= 0 { + continue + } + if hostBits > 8 { + hostBits = 8 + } + b[i] |= byte(0xFF >> (8 - hostBits)) //nolint:gosec // hostBits is 1..8 here + } + a, _ := netip.AddrFromSlice(b) + return a +} + +// RemoveWGPeer deletes a peer registration. sql.ErrNoRows when the pubkey is unknown. +func (s *Store) RemoveWGPeer(pubkey string) error { + res, err := s.db.Exec(`DELETE FROM wg_peers WHERE pubkey = ?`, pubkey) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return sql.ErrNoRows + } + return nil +} + +// ListWGPeers returns all registered peers ordered by assigned_ip — a deterministic order so +// the sync payload bytes are stable for identical registries. +func (s *Store) ListWGPeers() ([]WGPeer, error) { + rows, err := s.db.Query(`SELECT pubkey, assigned_ip, host_id, note FROM wg_peers ORDER BY assigned_ip`) + if err != nil { + return nil, err + } + defer rows.Close() + var peers []WGPeer + for rows.Next() { + var p WGPeer + if err := rows.Scan(&p.Pubkey, &p.AssignedIP, &p.HostID, &p.Note); err != nil { + return nil, err + } + peers = append(peers, p) + } + return peers, rows.Err() +} diff --git a/hub/internal/store/wg_test.go b/hub/internal/store/wg_test.go new file mode 100644 index 0000000..862e056 --- /dev/null +++ b/hub/internal/store/wg_test.go @@ -0,0 +1,155 @@ +package store + +// Group A — the /32 allocator (S1, doc 06 §3.2). Non-hollow: every test asserts allocation +// EFFECTS (exact IPs, row counts, typed errors), not just nil-errors. + +import ( + "database/sql" + "testing" +) + +func setTestEndpoint(t *testing.T, s *Store, subnet, pbsIP string) { + t.Helper() + err := s.SetWGEndpoint(&WGEndpoint{ + DNSName: "ep0.example", WGPort: 443, ServerPubkey: "SPK", + TunnelSubnet: subnet, PBSTunnelIP: pbsIP, + }) + if err != nil { + t.Fatalf("SetWGEndpoint: %v", err) + } +} + +func peerCount(t *testing.T, s *Store) int { + t.Helper() + peers, err := s.ListWGPeers() + if err != nil { + t.Fatalf("ListWGPeers: %v", err) + } + return len(peers) +} + +func TestWGEndpoint_UpsertAndGet(t *testing.T) { + s := newTestStore(t) + if _, err := s.GetWGEndpoint(); err != sql.ErrNoRows { + t.Fatalf("unset endpoint: err = %v, want sql.ErrNoRows", err) + } + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + e, err := s.GetWGEndpoint() + if err != nil || e.EndpointID != "ep0" || e.TunnelSubnet != "10.77.0.0/24" || e.WGPort != 443 { + t.Fatalf("GetWGEndpoint = %+v, %v", e, err) + } + // Upsert overwrites, same row. + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + e2, err := s.GetWGEndpoint() + if err != nil || e2.EndpointID != "ep0" { + t.Fatalf("after upsert: %+v, %v", e2, err) + } +} + +func TestAddWGPeer_SequentialAllocationSkipsNetPBSAndBroadcast(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + + // .0 network and .1 endpoint are skipped → first three peers get .2 .3 .4 exactly. + want := []string{"10.77.0.2", "10.77.0.3", "10.77.0.4"} + for i, pk := range []string{"p1", "p2", "p3"} { + ip, existed, err := s.AddWGPeer(pk, "", "") + if err != nil || existed { + t.Fatalf("add %s: ip=%q existed=%v err=%v", pk, ip, existed, err) + } + if ip != want[i] { + t.Errorf("peer %s ip = %q, want %q", pk, ip, want[i]) + } + } +} + +func TestAddWGPeer_FreedIPReused(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + for _, pk := range []string{"p1", "p2", "p3"} { + s.AddWGPeer(pk, "", "") + } + if err := s.RemoveWGPeer("p2"); err != nil { + t.Fatalf("RemoveWGPeer(p2): %v", err) + } + ip, existed, err := s.AddWGPeer("p4", "", "") + if err != nil || existed { + t.Fatalf("add p4: %v existed=%v", err, existed) + } + if ip != "10.77.0.3" { + t.Errorf("p4 ip = %q, want the freed lowest 10.77.0.3", ip) + } +} + +func TestAddWGPeer_IdempotentOnExistingPubkey(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + ip1, _, err := s.AddWGPeer("p1", "", "") + if err != nil { + t.Fatalf("first add: %v", err) + } + before := peerCount(t, s) + ip2, existed, err := s.AddWGPeer("p1", "", "") + if err != nil || !existed { + t.Fatalf("re-add: err=%v existed=%v, want idempotent hit", err, existed) + } + if ip2 != ip1 { + t.Errorf("re-add ip = %q, want unchanged %q", ip2, ip1) + } + if after := peerCount(t, s); after != before { + t.Errorf("row count changed on idempotent re-add: %d -> %d", before, after) + } +} + +func TestAddWGPeer_NoEndpoint(t *testing.T) { + s := newTestStore(t) + _, _, err := s.AddWGPeer("p1", "", "") + if err != ErrWGEndpointUnset { + t.Fatalf("err = %v, want ErrWGEndpointUnset", err) + } + if n := peerCount(t, s); n != 0 { + t.Errorf("peer rows created without endpoint: %d", n) + } +} + +func TestAddWGPeer_SubnetExhausted(t *testing.T) { + s := newTestStore(t) + // /30: .0 network, .1 endpoint, .2 the only host, .3 broadcast. + setTestEndpoint(t, s, "10.77.0.0/30", "10.77.0.1") + ip, _, err := s.AddWGPeer("p1", "", "") + if err != nil || ip != "10.77.0.2" { + t.Fatalf("p1: ip=%q err=%v, want the single free host .2", ip, err) + } + _, _, err = s.AddWGPeer("p2", "", "") + if err != ErrWGSubnetExhausted { + t.Fatalf("p2 err = %v, want ErrWGSubnetExhausted", err) + } + if n := peerCount(t, s); n != 1 { + t.Errorf("partial row on exhaustion: count = %d, want 1", n) + } +} + +func TestRemoveWGPeer_UnknownIsNoRows(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + if err := s.RemoveWGPeer("ghost"); err != sql.ErrNoRows { + t.Fatalf("err = %v, want sql.ErrNoRows", err) + } +} + +func TestListWGPeers_DeterministicOrder(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + s.AddWGPeer("pA", "h1", "note-a") + s.AddWGPeer("pB", "", "") + peers, err := s.ListWGPeers() + if err != nil || len(peers) != 2 { + t.Fatalf("ListWGPeers: %v (%d)", err, len(peers)) + } + if peers[0].Pubkey != "pA" || peers[0].AssignedIP != "10.77.0.2" || peers[0].HostID != "h1" { + t.Errorf("peers[0] = %+v", peers[0]) + } + if peers[1].AssignedIP != "10.77.0.3" { + t.Errorf("peers[1] = %+v", peers[1]) + } +}