From 0ec75551265e113fb738924d123491046a17c2c1 Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Sun, 5 Jul 2026 22:08:06 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6 --- hub/internal/api/handler.go | 5 ++ hub/internal/api/wg.go | 85 +++++++++++++++++++- hub/internal/api/wg_test.go | 64 +++++++++++++++ hub/internal/store/store.go | 13 ++++ hub/internal/store/wg_operator.go | 103 +++++++++++++++++++++++++ hub/internal/store/wg_operator_test.go | 73 ++++++++++++++++++ 6 files changed, 342 insertions(+), 1 deletion(-) create mode 100644 hub/internal/store/wg_operator.go create mode 100644 hub/internal/store/wg_operator_test.go diff --git a/hub/internal/api/handler.go b/hub/internal/api/handler.go index 82d9045..df88d55 100644 --- a/hub/internal/api/handler.go +++ b/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": diff --git a/hub/internal/api/wg.go b/hub/internal/api/wg.go index 8bc5e88..9a61ac6 100644 --- a/hub/internal/api/wg.go +++ b/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", + }) +} diff --git a/hub/internal/api/wg_test.go b/hub/internal/api/wg_test.go index 1bfcfb5..cfc9818 100644 --- a/hub/internal/api/wg_test.go +++ b/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") diff --git a/hub/internal/store/store.go b/hub/internal/store/store.go index a4d5527..3e99f91 100644 --- a/hub/internal/store/store.go +++ b/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 { diff --git a/hub/internal/store/wg_operator.go b/hub/internal/store/wg_operator.go new file mode 100644 index 0000000..28042f4 --- /dev/null +++ b/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 +} diff --git a/hub/internal/store/wg_operator_test.go b/hub/internal/store/wg_operator_test.go new file mode 100644 index 0000000..071a8f2 --- /dev/null +++ b/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) + } +}