From fcf84a0c5c444fa29abc415d2efdf03dd9be0331 Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Sat, 4 Jul 2026 00:39:24 +0200 Subject: [PATCH] =?UTF-8?q?hub:=20S2=20store=20=E2=80=94=20host-bound=20WG?= =?UTF-8?q?=20peers=20(register/re-key/adopt),=20one-per-host=20index,=20B?= =?UTF-8?q?umpHostDesired?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit allocateWGPeerTx extracted from addWGPeerOnce (behavior-neutral; S1 tests unmodified+green). RegisterWGPeerForHost: idempotent / re-key-in-place-keep-ip / adopt-unbound / ErrWGPubkeyBoundElsewhere. Partial unique index enforces one bound peer per host. BumpHostDesired touches ONLY the generation. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6 --- hub/internal/store/store.go | 23 +++++ hub/internal/store/wg.go | 185 +++++++++++++++++++++++++++------- hub/internal/store/wg_test.go | 113 +++++++++++++++++++++ 3 files changed, 287 insertions(+), 34 deletions(-) diff --git a/hub/internal/store/store.go b/hub/internal/store/store.go index fec6980..4d5e563 100644 --- a/hub/internal/store/store.go +++ b/hub/internal/store/store.go @@ -391,6 +391,8 @@ func (s *Store) migrate() error { created_at DATETIME NOT NULL DEFAULT (datetime('now')), updated_at DATETIME NOT NULL DEFAULT (datetime('now')) ); + -- S2: one BOUND peer per host (partial index — S1's unbound admin/test rows unaffected). + CREATE UNIQUE INDEX IF NOT EXISTS idx_wg_peers_host ON wg_peers(host_id) WHERE host_id != ''; `) if err != nil { return err @@ -1459,6 +1461,27 @@ func (s *Store) SetHostDesired(hostID string, desiredJSON []byte) (int64, error) return gen, nil } +// BumpHostDesired advances a host's desired_generation WITHOUT touching desired_json (S2). Used +// when hub-OWNED served state changes (the merge-at-read wireguard block) — the stored operator +// blob is not the thing that moved, so SetHostDesired (which replaces it) must not be used. +// Returns the new generation; sql.ErrNoRows for an unknown host. +func (s *Store) BumpHostDesired(hostID string) (int64, error) { + res, err := s.db.Exec(` + UPDATE hosts SET desired_generation = desired_generation + 1, updated_at = datetime('now') + WHERE host_id = ?`, hostID) + if err != nil { + return 0, err + } + if n, _ := res.RowsAffected(); n == 0 { + return 0, sql.ErrNoRows // unknown host + } + var gen int64 + if err := s.db.QueryRow(`SELECT desired_generation FROM hosts WHERE host_id = ?`, hostID).Scan(&gen); err != nil { + return 0, err + } + return gen, nil +} + // SignedJob is one OPAQUE operator-signed destructive-op blob queued for a host (slice 10A). The // hub stores + serves the bytes; it never forges, opens, or executes them (10B owns verify+run). type SignedJob struct { diff --git a/hub/internal/store/wg.go b/hub/internal/store/wg.go index 7326d47..01107a9 100644 --- a/hub/internal/store/wg.go +++ b/hub/internal/store/wg.go @@ -20,6 +20,10 @@ 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") +// ErrWGPubkeyBoundElsewhere is returned by RegisterWGPeerForHost when the pubkey is already +// registered for a different owner — a key is never silently stolen (S2 Scenario C-c2). +var ErrWGPubkeyBoundElsewhere = errors.New("wg pubkey already registered elsewhere") + // 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 { @@ -103,49 +107,162 @@ func (s *Store) addWGPeerOnce(pubkey, hostID, note string) (string, bool, error) 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 - } + ip, err := allocateWGPeerTx(tx, pubkey, hostID, note) 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 } +// allocateWGPeerTx is the tx-scoped allocation core (S2 extraction — behavior-identical to the +// S1 inline code): load the endpoint, read the taken set, pick the lowest free host address, +// INSERT. Callers own the transaction and the pubkey-existence checks. +func allocateWGPeerTx(tx *sql.Tx, pubkey, hostID, note string) (string, error) { + 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 "", ErrWGEndpointUnset + } + if err != nil { + return "", err + } + + taken := map[string]bool{} + rows, err := tx.Query(`SELECT assigned_ip FROM wg_peers`) + if err != nil { + return "", err + } + for rows.Next() { + var ip string + if err := rows.Scan(&ip); err != nil { + rows.Close() + return "", err + } + taken[ip] = true + } + rows.Close() + if err := rows.Err(); err != nil { + return "", err + } + + ip, err := lowestFreeHost(subnet, pbsIP, taken) + if err != nil { + return "", err + } + + if _, err := tx.Exec(`INSERT INTO wg_peers (pubkey, assigned_ip, host_id, note) VALUES (?, ?, ?, ?)`, + pubkey, ip, hostID, note); err != nil { + return "", err + } + return ip, nil +} + +// RegisterWGPeerForHost binds a pubkey to a host (S2 — the box-facing registration core). +// One transaction; outcomes: +// - host already has this pubkey → (ip, changed=false): idempotent, NO generation/sync side effects +// - host has a DIFFERENT pubkey → re-key IN PLACE: pubkey replaced, /32 KEPT (stable tunnel +// addressing across rotation/DR); conflict with any existing row for the new pubkey → typed error +// - no row for host: pubkey exists bound elsewhere → ErrWGPubkeyBoundElsewhere; exists unbound +// (an S1 admin/test row) → ADOPT it (bind, keep its ip); absent → allocate the lowest free /32 +// +// The idx_wg_peers_host partial unique index backstops the one-bound-peer-per-host invariant. +func (s *Store) RegisterWGPeerForHost(hostID, pubkey string) (ip string, changed bool, err error) { + for attempt := 0; attempt < 2; attempt++ { + ip, changed, err = s.registerWGPeerForHostOnce(hostID, pubkey) + if err != nil && attempt == 0 && strings.Contains(err.Error(), "UNIQUE constraint failed: wg_peers.assigned_ip") { + continue // allocation race — retry once against the fresh taken set + } + return ip, changed, err + } + return ip, changed, err +} + +func (s *Store) registerWGPeerForHostOnce(hostID, pubkey string) (string, bool, error) { + if hostID == "" { + return "", false, fmt.Errorf("wg: RegisterWGPeerForHost needs a host_id") + } + tx, err := s.db.Begin() + if err != nil { + return "", false, err + } + defer tx.Rollback() + + // Existing binding for this host? + var curPK, curIP string + err = tx.QueryRow(`SELECT pubkey, assigned_ip FROM wg_peers WHERE host_id = ?`, hostID).Scan(&curPK, &curIP) + if err == nil { + if curPK == pubkey { + return curIP, false, nil // idempotent re-register + } + // Re-key in place. The new pubkey must not exist anywhere (bound OR unbound) — a key is + // never stolen, and adopting here would change the host's ip (stable-IP rule forbids it). + var n int + if err := tx.QueryRow(`SELECT COUNT(*) FROM wg_peers WHERE pubkey = ?`, pubkey).Scan(&n); err != nil { + return "", false, err + } + if n > 0 { + return "", false, ErrWGPubkeyBoundElsewhere + } + if _, err := tx.Exec(`UPDATE wg_peers SET pubkey = ?, updated_at = datetime('now') WHERE host_id = ?`, + pubkey, hostID); err != nil { + return "", false, err + } + if err := tx.Commit(); err != nil { + return "", false, err + } + return curIP, true, nil + } + if err != sql.ErrNoRows { + return "", false, err + } + + // No binding for this host — does the pubkey already exist? + var pkHost, pkIP string + err = tx.QueryRow(`SELECT host_id, assigned_ip FROM wg_peers WHERE pubkey = ?`, pubkey).Scan(&pkHost, &pkIP) + if err == nil { + if pkHost != "" { + return "", false, ErrWGPubkeyBoundElsewhere + } + // Unbound S1 row → adopt: bind it to the host, keep its ip. + if _, err := tx.Exec(`UPDATE wg_peers SET host_id = ?, updated_at = datetime('now') WHERE pubkey = ?`, + hostID, pubkey); err != nil { + return "", false, err + } + if err := tx.Commit(); err != nil { + return "", false, err + } + return pkIP, true, nil + } + if err != sql.ErrNoRows { + return "", false, err + } + + // Fresh registration → allocate. + ip, err := allocateWGPeerTx(tx, pubkey, hostID, "") + if err != nil { + return "", false, err + } + if err := tx.Commit(); err != nil { + return "", false, err + } + return ip, true, nil +} + +// GetWGPeerForHost returns the host's bound peer, or sql.ErrNoRows when none. +func (s *Store) GetWGPeerForHost(hostID string) (*WGPeer, error) { + var p WGPeer + err := s.db.QueryRow(`SELECT pubkey, assigned_ip, host_id, note FROM wg_peers WHERE host_id = ?`, hostID). + Scan(&p.Pubkey, &p.AssignedIP, &p.HostID, &p.Note) + if err != nil { + return nil, err + } + return &p, 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) { diff --git a/hub/internal/store/wg_test.go b/hub/internal/store/wg_test.go index 862e056..69a68d5 100644 --- a/hub/internal/store/wg_test.go +++ b/hub/internal/store/wg_test.go @@ -137,6 +137,119 @@ func TestRemoveWGPeer_UnknownIsNoRows(t *testing.T) { } } +// --- S2 Group A: host binding + generation bump --- + +func TestRegisterWGPeerForHost_FreshIdempotentAndRekey(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + if err := s.UpsertHost(&Host{HostID: "h1", CustomerID: "c1", APIKey: "k1"}); err != nil { + t.Fatal(err) + } + + // Fresh → allocates .2, changed=true. + ip, changed, err := s.RegisterWGPeerForHost("h1", "P1") + if err != nil || ip != "10.77.0.2" || !changed { + t.Fatalf("fresh register = %q changed=%v err=%v, want .2/true", ip, changed, err) + } + // Idempotent: same pubkey → same ip, changed=false. + ip, changed, err = s.RegisterWGPeerForHost("h1", "P1") + if err != nil || ip != "10.77.0.2" || changed { + t.Fatalf("idempotent register = %q changed=%v err=%v, want .2/false", ip, changed, err) + } + // Re-key: NEW pubkey → swapped in place, ip KEPT, changed=true. + ip, changed, err = s.RegisterWGPeerForHost("h1", "P2") + if err != nil || ip != "10.77.0.2" || !changed { + t.Fatalf("re-key = %q changed=%v err=%v, want .2 kept/true", ip, changed, err) + } + p, err := s.GetWGPeerForHost("h1") + if err != nil || p.Pubkey != "P2" || p.AssignedIP != "10.77.0.2" { + t.Fatalf("after re-key peer = %+v err=%v", p, err) + } + if n := peerCount(t, s); n != 1 { + t.Errorf("row count after re-key = %d, want 1", n) + } +} + +func TestRegisterWGPeerForHost_AdoptsUnboundRow(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + s.UpsertHost(&Host{HostID: "h1", CustomerID: "c1", APIKey: "k1"}) + // An S1 admin-added UNBOUND peer. + ip0, _, err := s.AddWGPeer("P1", "", "s1-test") + if err != nil { + t.Fatal(err) + } + ip, changed, err := s.RegisterWGPeerForHost("h1", "P1") + if err != nil || !changed || ip != ip0 { + t.Fatalf("adopt = %q changed=%v err=%v, want %q/true", ip, changed, err, ip0) + } + p, _ := s.GetWGPeerForHost("h1") + if p == nil || p.Pubkey != "P1" { + t.Fatalf("adopted peer = %+v", p) + } +} + +func TestRegisterWGPeerForHost_BoundElsewhereRefused(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + s.UpsertHost(&Host{HostID: "h1", CustomerID: "c1", APIKey: "k1"}) + s.UpsertHost(&Host{HostID: "h2", CustomerID: "c2", APIKey: "k2"}) + s.RegisterWGPeerForHost("h2", "P1") + + // Fresh-registration path: pubkey bound to h2 → refused. + if _, _, err := s.RegisterWGPeerForHost("h1", "P1"); err != ErrWGPubkeyBoundElsewhere { + t.Fatalf("register with h2's pubkey: err = %v, want ErrWGPubkeyBoundElsewhere", err) + } + if p, err := s.GetWGPeerForHost("h1"); err != sql.ErrNoRows { + t.Errorf("h1 gained a peer despite refusal: %+v (err=%v)", p, err) + } + // Re-key path: h1 registers its own, then tries to re-key to h2's pubkey → refused, binding intact. + s.RegisterWGPeerForHost("h1", "P9") + if _, _, err := s.RegisterWGPeerForHost("h1", "P1"); err != ErrWGPubkeyBoundElsewhere { + t.Fatalf("re-key onto h2's pubkey: err = %v, want ErrWGPubkeyBoundElsewhere", err) + } + p, _ := s.GetWGPeerForHost("h1") + if p == nil || p.Pubkey != "P9" { + t.Errorf("h1 binding disturbed by refused re-key: %+v", p) + } +} + +func TestWGPeers_OneBoundPerHostIndex(t *testing.T) { + s := newTestStore(t) + setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1") + s.UpsertHost(&Host{HostID: "h1", CustomerID: "c1", APIKey: "k1"}) + s.RegisterWGPeerForHost("h1", "P1") + // A second BOUND row for the same host via the S1 admin path → the partial index refuses. + if _, _, err := s.AddWGPeer("P2", "h1", ""); err == nil { + t.Fatal("second bound peer for h1 accepted — idx_wg_peers_host is not enforcing") + } + // Unbound rows are unaffected by the partial index. + if _, _, err := s.AddWGPeer("P3", "", ""); err != nil { + t.Fatalf("unbound add refused: %v", err) + } +} + +func TestBumpHostDesired(t *testing.T) { + s := newTestStore(t) + s.UpsertHost(&Host{HostID: "h1", CustomerID: "c1", APIKey: "k1"}) + gen, err := s.BumpHostDesired("h1") + if err != nil || gen != 1 { + t.Fatalf("bump #1 = %d, %v", gen, err) + } + gen, err = s.BumpHostDesired("h1") + if err != nil || gen != 2 { + t.Fatalf("bump #2 = %d, %v", gen, err) + } + if _, err := s.BumpHostDesired("ghost"); err != sql.ErrNoRows { + t.Errorf("unknown host bump err = %v, want ErrNoRows", err) + } + // desired_json untouched by the bump. + h, _ := s.GetHost("h1") + if h.DesiredJSON != "{}" && h.DesiredJSON != "" { + t.Errorf("desired_json moved on bump: %q", h.DesiredJSON) + } +} + func TestListWGPeers_DeterministicOrder(t *testing.T) { s := newTestStore(t) setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1")