feat(hub): operator OOB peer + oob_peer_ip desired-state merge (H1 Part 1)

store.SetOperatorOOBPeer/GetOperatorOOBPeer (empty-host_id wg_peers row, explicit
/32, validated in-subnet/not-reserved/not-taken, last-write-wins rotation).
PUT/GET /admin/wg/operator-peer (global key). mergeWireguard adds oob_peer_ip when
an operator peer exists (absent = byte-identical). BumpAllHostGenerations forces
fleet re-fetch. Non-hollow tests both sides.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6

hub/internal/api/handler.go

@@ -215,6 +215,11 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		h.handleAdminDeleteWGPeer(w, r)
 	case r.Method == http.MethodGet && path == "/admin/wg/peers":
 		h.handleAdminListWGPeers(w, r)
+	// H1: the fleet operator OOB peer (register/rotate at an explicit /32) + read-back.
+	case r.Method == http.MethodPut && path == "/admin/wg/operator-peer":
+		h.handleAdminSetOperatorPeer(w, r)
+	case r.Method == http.MethodGet && path == "/admin/wg/operator-peer":
+		h.handleAdminGetOperatorPeer(w, r)
 	case r.Method == http.MethodPost && path == "/event":
 		h.handleEvent(w, r)
 	case r.Method == http.MethodPost && path == "/mail":

hub/internal/api/wg.go

@@ -309,7 +309,7 @@ func (h *Handler) mergeWireguard(hostID, desired string) string {
 		h.logger.Printf("[ERROR] wg merge %s: stored desired_json unparsable: %v (serving unmerged)", hostID, err)
 		return desired
 	}
-	doc["wireguard"] = map[string]interface{}{
+	wgBlock := map[string]interface{}{
 		"endpoint": map[string]interface{}{
 			"dns_name":      ep.DNSName,
 			"wg_port":       ep.WGPort,
@@ -319,6 +319,16 @@ func (h *Handler) mergeWireguard(hostID, desired string) string {
 		"pubkey":      peer.Pubkey,
 		"assigned_ip": peer.AssignedIP + "/32",
 	}
+	// H1 [OF-1]: if a fleet operator OOB peer is registered, deliver its tunnel /32 as oob_peer_ip so
+	// the agent RENDERS it into wg-felhom AllowedIPs (survives self-heal). Absent → key omitted →
+	// byte-identical pass-through for pre-H1 fleets. A lookup failure is fail-safe (serve without OOB,
+	// never break the control channel).
+	if op, oerr := h.store.GetOperatorOOBPeer(); oerr != nil {
+		h.logger.Printf("[WARN] wg merge %s: operator OOB peer lookup failed: %v (serving without oob_peer_ip)", hostID, oerr)
+	} else if op != nil {
+		wgBlock["oob_peer_ip"] = op.AssignedIP // bare IPv4, e.g. "10.77.0.250"
+	}
+	doc["wireguard"] = wgBlock
 	out, err := json.Marshal(doc)
 	if err != nil {
 		h.logger.Printf("[ERROR] wg merge %s: re-marshal: %v (serving unmerged)", hostID, err)
@@ -409,3 +419,76 @@ func (h *Handler) handleAdminListWGPeers(w http.ResponseWriter, r *http.Request)
 	w.Header().Set("Content-Type", "application/json")
 	json.NewEncoder(w).Encode(map[string]interface{}{"peers": out})
 }
+
+// handleAdminSetOperatorPeer — PUT /admin/wg/operator-peer (global key). Registers/rotates the fleet
+// operator OOB peer at an EXPLICIT /32 (so the endpoint's static forward chain can hardcode it),
+// pushes the full peer list to the endpoint (so the operator's handshake is accepted), and bumps
+// EVERY host's generation so agents re-fetch + re-render oob_peer_ip into wg-felhom AllowedIPs (H1).
+func (h *Handler) handleAdminSetOperatorPeer(w http.ResponseWriter, r *http.Request) {
+	_, _, isGlobal, ok := h.checkAuthHost(r)
+	if !ok || !isGlobal {
+		http.Error(w, "Forbidden: global key required", http.StatusForbidden)
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
+	if err != nil {
+		http.Error(w, "Bad request", http.StatusBadRequest)
+		return
+	}
+	var req struct {
+		Pubkey     string `json:"pubkey"`
+		AssignedIP string `json:"assigned_ip"` // bare IPv4, e.g. "10.77.0.250"
+	}
+	if err := json.Unmarshal(body, &req); err != nil {
+		http.Error(w, "Invalid payload: body must be JSON", http.StatusBadRequest)
+		return
+	}
+	if err := validateWGPubkey(req.Pubkey); err != nil {
+		http.Error(w, "Invalid payload: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+	if err := h.store.SetOperatorOOBPeer(req.Pubkey, req.AssignedIP); err != nil {
+		if err == store.ErrWGEndpointUnset {
+			http.Error(w, "wg endpoint not configured", http.StatusConflict)
+			return
+		}
+		http.Error(w, "Invalid operator peer: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+	syncStatus := h.syncAfterMutation(r.Context())
+	bumped, berr := h.store.BumpAllHostGenerations()
+	if berr != nil {
+		h.logger.Printf("[WARN] operator peer set but generation bump failed: %v", berr)
+	}
+	h.logger.Printf("[INFO] operator OOB peer set: %s -> %s/32 (sync=%s, %d host generations bumped)",
+		req.Pubkey, req.AssignedIP, syncStatus, bumped)
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(map[string]interface{}{
+		"pubkey": req.Pubkey, "assigned_ip": req.AssignedIP + "/32", "sync": syncStatus, "hosts_bumped": bumped,
+	})
+}
+
+// handleAdminGetOperatorPeer — GET /admin/wg/operator-peer (global key). Read-back for the runbook
+// (the operator /32 the endpoint's static forward chain must allow). 404 when none registered.
+func (h *Handler) handleAdminGetOperatorPeer(w http.ResponseWriter, r *http.Request) {
+	_, _, isGlobal, ok := h.checkAuthHost(r)
+	if !ok || !isGlobal {
+		http.Error(w, "Forbidden: global key required", http.StatusForbidden)
+		return
+	}
+	op, err := h.store.GetOperatorOOBPeer()
+	if err != nil {
+		h.logger.Printf("[ERROR] get operator peer: %v", err)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	if op == nil {
+		http.Error(w, "No operator OOB peer registered", http.StatusNotFound)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]interface{}{
+		"pubkey": op.Pubkey, "assigned_ip": op.AssignedIP + "/32",
+	})
+}

hub/internal/api/wg_test.go

@@ -385,6 +385,70 @@ func TestDesiredState_WireguardMergeMatchesGolden(t *testing.T) {
 	}
 }
 
+// H1: with a fleet operator OOB peer registered, a host's served wireguard block carries oob_peer_ip
+// (the operator's /32) so the agent renders it into AllowedIPs. Without it, no key (byte-identical —
+// covered by TestDesiredState_WireguardMergeMatchesGolden).
+func TestDesiredState_IncludesOOBPeerIPWhenOperatorRegistered(t *testing.T) {
+	h, st, _ := newTestHandler(t)
+	seedHost(t, st, "h1", "c1", "HKEY1")
+	putGoldenEndpoint(t, h)
+	h.SetWGSyncer(&fakeWGSyncer{})
+	if rr := do(h, http.MethodPost, "/hosts/h1/wg", "HKEY1", `{"pubkey":"`+testPK(1)+`"}`); rr.Code != 200 {
+		t.Fatalf("register: %d", rr.Code)
+	}
+
+	// no operator peer yet → served block has NO oob_peer_ip
+	got := servedWG(t, h)
+	if _, has := got["oob_peer_ip"]; has {
+		t.Fatalf("oob_peer_ip present with no operator peer: %v", got)
+	}
+
+	// register the operator peer (global key)
+	if rr := do(h, http.MethodPut, "/admin/wg/operator-peer", globalKey,
+		`{"pubkey":"`+opTestPubkey+`","assigned_ip":"10.77.0.250"}`); rr.Code != 200 {
+		t.Fatalf("set operator peer: %d %s", rr.Code, servedBody(t, h))
+	}
+	got = servedWG(t, h)
+	if got["oob_peer_ip"] != "10.77.0.250" {
+		t.Fatalf("served oob_peer_ip = %v, want 10.77.0.250", got["oob_peer_ip"])
+	}
+
+	// read-back route
+	if rr := do(h, http.MethodGet, "/admin/wg/operator-peer", globalKey, ""); rr.Code != 200 {
+		t.Fatalf("operator-peer read-back: %d", rr.Code)
+	}
+	// a per-host key cannot set the operator peer
+	if rr := do(h, http.MethodPut, "/admin/wg/operator-peer", "HKEY1",
+		`{"pubkey":"`+opTestPubkey+`","assigned_ip":"10.77.0.248"}`); rr.Code != 403 {
+		t.Fatalf("host key setting operator peer must be 403, got %d", rr.Code)
+	}
+}
+
+const opTestPubkey = "cdmN4U+fjR18zBk+SKoceJQyz9HgA9+hN8/FiKF1u0o="
+
+func servedWG(t *testing.T, h *Handler) map[string]any {
+	t.Helper()
+	rr := do(h, http.MethodGet, "/hosts/h1/desired-state", "HKEY1", "")
+	if rr.Code != 200 {
+		t.Fatalf("GET desired-state: %d", rr.Code)
+	}
+	var got struct {
+		DesiredState json.RawMessage `json:"desired_state"`
+	}
+	json.Unmarshal(rr.Body.Bytes(), &got)
+	var doc map[string]any
+	json.Unmarshal(got.DesiredState, &doc)
+	wg, _ := doc["wireguard"].(map[string]any)
+	if wg == nil {
+		t.Fatalf("no wireguard block in served state: %s", got.DesiredState)
+	}
+	return wg
+}
+
+func servedBody(t *testing.T, h *Handler) string {
+	return do(h, http.MethodGet, "/hosts/h1/desired-state", "HKEY1", "").Body.String()
+}
+
 func TestAdminSetDesiredState_RejectsWireguardKey(t *testing.T) {
 	h, st, _ := newTestHandler(t)
 	seedHost(t, st, "h1", "c1", "HKEY1")

hub/internal/store/store.go

@@ -1500,6 +1500,19 @@ func (s *Store) BumpHostDesired(hostID string) (int64, error) {
 	return gen, nil
 }
 
+// BumpAllHostGenerations advances EVERY host's desired_generation (TASK H1). Used when a FLEET-WIDE
+// hub-owned served-state value changes — the operator OOB peer's /32 flows into every host's
+// merge-at-read wireguard block (oob_peer_ip), so every agent must re-fetch + re-render. Returns the
+// number of hosts bumped.
+func (s *Store) BumpAllHostGenerations() (int64, error) {
+	res, err := s.db.Exec(`UPDATE hosts SET desired_generation = desired_generation + 1, updated_at = datetime('now')`)
+	if err != nil {
+		return 0, err
+	}
+	n, _ := res.RowsAffected()
+	return n, 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 {

hub/internal/store/wg_operator.go

@@ -0,0 +1,103 @@
+package store
+
+import (
+	"database/sql"
+	"fmt"
+	"net/netip"
+)
+
+// OperatorOOBNote marks the single fleet-wide operator OOB peer among wg_peers (TASK H1). It is an
+// UNBOUND peer (host_id ''), so it rides the existing empty-host_id support + the partial-unique
+// idx_wg_peers_host (which only constrains non-empty host_id) without collision [OF-4]. It is pushed
+// to the endpoint by peersync like any peer (so the endpoint accepts the operator's handshake), and
+// its assigned_ip becomes each box's desired-state oob_peer_ip (rendered into wg-felhom AllowedIPs).
+const OperatorOOBNote = "operator-oob"
+
+// GetOperatorOOBPeer returns the fleet operator OOB peer, or (nil, nil) if none is registered.
+func (s *Store) GetOperatorOOBPeer() (*WGPeer, error) {
+	var p WGPeer
+	err := s.db.QueryRow(
+		`SELECT pubkey, assigned_ip, host_id, note, created_at FROM wg_peers WHERE note = ? LIMIT 1`,
+		OperatorOOBNote).
+		Scan(&p.Pubkey, &p.AssignedIP, &p.HostID, &p.Note, &p.CreatedAt)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &p, nil
+}
+
+// SetOperatorOOBPeer registers/rotates THE operator OOB peer at an EXPLICIT tunnel address (so the
+// endpoint's STATIC forward chain can hardcode the operator /32). Last-write-wins: any prior
+// operator-oob row is replaced (key rotation / IP change). Validated: assignedIP is an IPv4 host
+// address inside the endpoint's tunnel_subnet, not the network/broadcast/pbs_tunnel_ip, and not
+// already taken by a DIFFERENT (customer) peer. host_id stays '' (unbound) so it is never counted as
+// a customer host.
+func (s *Store) SetOperatorOOBPeer(pubkey, assignedIP string) error {
+	ip, err := netip.ParseAddr(assignedIP)
+	if err != nil || !ip.Is4() {
+		return fmt.Errorf("operator oob: assigned_ip %q is not an IPv4 address", assignedIP)
+	}
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	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
+	}
+	pfx, err := netip.ParsePrefix(subnet)
+	if err != nil {
+		return fmt.Errorf("operator oob: endpoint tunnel_subnet %q unparseable: %w", subnet, err)
+	}
+	if !pfx.Contains(ip) {
+		return fmt.Errorf("operator oob: assigned_ip %s is not inside tunnel_subnet %s", assignedIP, subnet)
+	}
+	if ip == pfx.Masked().Addr() || ip.String() == pbsIP || ip == lastAddr(pfx) {
+		return fmt.Errorf("operator oob: assigned_ip %s is a reserved address (network/broadcast/pbs)", assignedIP)
+	}
+	// The IP must be free OR already this operator peer's (idempotent re-set at the same IP).
+	var otherPubkey string
+	err = tx.QueryRow(`SELECT pubkey FROM wg_peers WHERE assigned_ip = ? AND note != ?`, assignedIP, OperatorOOBNote).
+		Scan(&otherPubkey)
+	if err == nil {
+		return fmt.Errorf("operator oob: assigned_ip %s is already taken by peer %s", assignedIP, shortKey(otherPubkey))
+	}
+	if err != sql.ErrNoRows {
+		return err
+	}
+	// The pubkey must not be bound to a host (an operator peer is never a customer key).
+	var boundHost string
+	err = tx.QueryRow(`SELECT host_id FROM wg_peers WHERE pubkey = ? AND host_id != ''`, pubkey).Scan(&boundHost)
+	if err == nil {
+		return fmt.Errorf("operator oob: pubkey is already bound to host %s — refusing", boundHost)
+	}
+	if err != sql.ErrNoRows {
+		return err
+	}
+
+	if _, err := tx.Exec(`DELETE FROM wg_peers WHERE note = ?`, OperatorOOBNote); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(`INSERT INTO wg_peers (pubkey, assigned_ip, host_id, note) VALUES (?, ?, '', ?)`,
+		pubkey, assignedIP, OperatorOOBNote); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+func shortKey(k string) string {
+	if len(k) > 12 {
+		return k[:12] + "…"
+	}
+	return k
+}

hub/internal/store/wg_operator_test.go

@@ -0,0 +1,73 @@
+package store
+
+import "testing"
+
+const opPubkey = "cdmN4U+fjR18zBk+SKoceJQyz9HgA9+hN8/FiKF1u0o="
+
+func TestOperatorOOBPeer_RoundTripRotateAndUnbound(t *testing.T) {
+	s := newTestStore(t)
+	setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1")
+
+	// none → (nil,nil)
+	if op, err := s.GetOperatorOOBPeer(); err != nil || op != nil {
+		t.Fatalf("no operator peer: got %+v / %v", op, err)
+	}
+
+	// register at an explicit /32
+	if err := s.SetOperatorOOBPeer(opPubkey, "10.77.0.250"); err != nil {
+		t.Fatalf("SetOperatorOOBPeer: %v", err)
+	}
+	op, err := s.GetOperatorOOBPeer()
+	if err != nil || op == nil {
+		t.Fatalf("GetOperatorOOBPeer: %+v / %v", op, err)
+	}
+	if op.AssignedIP != "10.77.0.250" || op.Pubkey != opPubkey {
+		t.Fatalf("operator peer = %+v", op)
+	}
+	if op.HostID != "" || op.Note != OperatorOOBNote {
+		t.Fatalf("operator peer must be UNBOUND (host_id '') + note operator-oob: %+v", op)
+	}
+	// it IS in the peer list (so peersync pushes it to the endpoint)
+	if peerCount(t, s) != 1 {
+		t.Fatalf("operator peer must appear in ListWGPeers")
+	}
+
+	// rotate: new key + new IP replaces the old (last-write-wins, still one operator row)
+	const opPubkey2 = "tQ6eJC9y8pSKtam44DaSrrtYBZYUvZG2z1stS6PsaD4="
+	if err := s.SetOperatorOOBPeer(opPubkey2, "10.77.0.249"); err != nil {
+		t.Fatalf("rotate: %v", err)
+	}
+	op, _ = s.GetOperatorOOBPeer()
+	if op.Pubkey != opPubkey2 || op.AssignedIP != "10.77.0.249" {
+		t.Fatalf("rotate did not replace: %+v", op)
+	}
+	if peerCount(t, s) != 1 {
+		t.Fatalf("rotate must keep exactly one operator peer, got %d", peerCount(t, s))
+	}
+}
+
+func TestOperatorOOBPeer_ValidationRejectsReservedAndTaken(t *testing.T) {
+	s := newTestStore(t)
+	setTestEndpoint(t, s, "10.77.0.0/24", "10.77.0.1")
+
+	for _, bad := range []struct{ ip, why string }{
+		{"10.77.0.1", "pbs_tunnel_ip"},
+		{"10.77.0.0", "network address"},
+		{"10.77.0.255", "broadcast"},
+		{"10.88.0.5", "outside subnet"},
+		{"not-an-ip", "not an ip"},
+	} {
+		if err := s.SetOperatorOOBPeer(opPubkey, bad.ip); err == nil {
+			t.Errorf("SetOperatorOOBPeer(%q) accepted but should reject (%s)", bad.ip, bad.why)
+		}
+	}
+
+	// a customer peer holds 10.77.0.2 → operator can't take it
+	if _, _, err := s.AddWGPeer("PCUST", "h1", ""); err != nil {
+		t.Fatalf("AddWGPeer: %v", err)
+	}
+	custIP, _ := s.GetWGPeerForHost("h1")
+	if err := s.SetOperatorOOBPeer(opPubkey, custIP.AssignedIP); err == nil {
+		t.Fatalf("operator peer must not take a customer's IP %s", custIP.AssignedIP)
+	}
+}