hub: S2 store — host-bound WG peers (register/re-key/adopt), one-per-host index, BumpHostDesired
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
This commit is contained in:
@@ -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 {
|
||||
|
||||
+151
-34
@@ -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) {
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user