diff --git a/hub/cmd/hub/main.go b/hub/cmd/hub/main.go index e828730..297557a 100644 --- a/hub/cmd/hub/main.go +++ b/hub/cmd/hub/main.go @@ -20,6 +20,7 @@ import ( "gitea.dooplex.hu/admin/felhom-hub/internal/notify" "gitea.dooplex.hu/admin/felhom-hub/internal/store" "gitea.dooplex.hu/admin/felhom-hub/internal/web" + "gitea.dooplex.hu/admin/felhom-hub/internal/wgsync" "gopkg.in/yaml.v3" ) @@ -316,6 +317,37 @@ func main() { apiHandler.SetLatestVersionProvider(versionChecker) } + // S1 offsite connectivity: the WG peer-sync reconciler (internal/wgsync). Config from env + // (mirror of the Resend/registry pattern): the SSH private key from the mounted Secret file, + // the (non-secret) pinned host key from plain env. Any piece missing → disabled with an INFO + // log; the /admin/wg mutations still work against the DB and report sync:"disabled". + { + wgAddr := os.Getenv("WG_ENDPOINT_SSH_ADDR") + wgUser := os.Getenv("WG_ENDPOINT_SSH_USER") + if wgUser == "" { + wgUser = "felhom-peersync" + } + wgKeyFile := os.Getenv("WG_ENDPOINT_SSH_KEY_FILE") + wgHostKey := os.Getenv("WG_ENDPOINT_SSH_HOSTKEY") + if wgAddr != "" && wgKeyFile != "" && wgHostKey != "" { + keyPEM, err := os.ReadFile(wgKeyFile) + if err != nil { + logger.Printf("[ERROR] WG peer-sync disabled: read key file %s: %v", wgKeyFile, err) + } else if wgClient, err := wgsync.New(wgsync.Config{ + Addr: wgAddr, User: wgUser, PrivateKey: keyPEM, HostKeyLine: wgHostKey, + }, logger); err != nil { + logger.Printf("[ERROR] WG peer-sync disabled: %v", err) + } else { + wgReconciler := wgsync.NewReconciler(dataStore, wgClient, logger) + go wgReconciler.Run(ctx) + apiHandler.SetWGSyncer(wgReconciler) + logger.Printf("[INFO] WG peer-sync enabled (endpoint %s, user %s)", wgAddr, wgUser) + } + } else { + logger.Printf("[INFO] WG peer-sync disabled (endpoint not configured)") + } + } + // Session cleanup — removes expired sessions every hour go webServer.CleanupSessions(ctx) diff --git a/hub/internal/api/handler.go b/hub/internal/api/handler.go index ef6e07f..1ac7bb0 100644 --- a/hub/internal/api/handler.go +++ b/hub/internal/api/handler.go @@ -49,6 +49,10 @@ type Handler struct { mailSender mailrelay.Sender mailLimiter *mailRateLimiter mailFromAllow map[string]bool + + // S1 offsite connectivity: the wgsync reconciler seam (internal/api/wg.go). nil = peer-sync + // disabled — mutations still persist, responses carry sync:"disabled". + wgSyncer WGSyncer } // SetLatestVersionProvider wires the registry version checker so the controller report ACK can @@ -187,6 +191,18 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { case r.Method == http.MethodPost && strings.HasPrefix(path, "/admin/hosts/") && strings.HasSuffix(path, "/jobs"): hostID := strings.TrimSuffix(strings.TrimPrefix(path, "/admin/hosts/"), "/jobs") h.handleAdminEnqueueJob(w, r, hostID) + // S1 offsite connectivity — WG endpoint record + peer registry (global key only, api/wg.go). + // DELETE carries the pubkey in the body: base64 '/'+'+' keep pubkeys out of URL paths. + case r.Method == http.MethodPut && path == "/admin/wg/endpoint": + h.handleAdminSetWGEndpoint(w, r) + case r.Method == http.MethodGet && path == "/admin/wg/endpoint": + h.handleAdminGetWGEndpoint(w, r) + case r.Method == http.MethodPost && path == "/admin/wg/peers": + h.handleAdminAddWGPeer(w, r) + case r.Method == http.MethodDelete && path == "/admin/wg/peers": + h.handleAdminDeleteWGPeer(w, r) + case r.Method == http.MethodGet && path == "/admin/wg/peers": + h.handleAdminListWGPeers(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 new file mode 100644 index 0000000..0afcb34 --- /dev/null +++ b/hub/internal/api/wg.go @@ -0,0 +1,264 @@ +package api + +// S1 offsite connectivity (doc 06 §3.2/§5): the operator admin surface for the WG endpoint +// record + peer registry. GLOBAL key ONLY on every route (the handleAdminSetDesiredState gate) — +// a per-host key must never author the peer list; the box-facing registration path is S2. +// DELETE takes the pubkey in the JSON body: WG pubkeys are std base64 ('/' and '+'), so a pubkey +// NEVER appears in a URL path — and no, URL-escaping is not the fix (see the S1 spec §8). + +import ( + "context" + "database/sql" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "net/netip" + "time" + + "gitea.dooplex.hu/admin/felhom-hub/internal/store" +) + +// WGSyncer is the reconciler seam (satisfied by *wgsync.Reconciler; tests inject a fake). nil = +// peer-sync disabled: mutations still hit the DB (the source of truth) and report sync:"disabled". +type WGSyncer interface { + SyncNow(ctx context.Context) error + Trigger() +} + +// SetWGSyncer wires the wgsync reconciler (mirror of SetLatestVersionProvider; nil-safe). +func (h *Handler) SetWGSyncer(s WGSyncer) { + h.wgSyncer = s +} + +// validateWGPubkey enforces the exact WG public-key shape: 44 chars of std base64 decoding to +// 32 bytes. Anything else is rejected before any allocation. +func validateWGPubkey(pk string) error { + if len(pk) != 44 { + return fmt.Errorf("pubkey must be 44 base64 chars, got %d", len(pk)) + } + raw, err := base64.StdEncoding.DecodeString(pk) + if err != nil { + return fmt.Errorf("pubkey is not valid base64: %v", err) + } + if len(raw) != 32 { + return fmt.Errorf("pubkey must decode to 32 bytes, got %d", len(raw)) + } + return nil +} + +// syncAfterMutation runs an inline sync after a peer mutation. The DB write already happened — +// it is the source of truth — so a push failure is REPORTED, not rolled back: the reconciler's +// next tick converges the endpoint (Scenario D). +func (h *Handler) syncAfterMutation(ctx context.Context) string { + if h.wgSyncer == nil { + return "disabled" + } + syncCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if err := h.wgSyncer.SyncNow(syncCtx); err != nil { + h.logger.Printf("[ERROR] wgsync: inline push after mutation failed: %v (reconciler will retry)", err) + h.wgSyncer.Trigger() + return "deferred: " + err.Error() + } + return "ok" +} + +// handleAdminSetWGEndpoint — PUT /admin/wg/endpoint. Upserts the endpoint record. +func (h *Handler) handleAdminSetWGEndpoint(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 { + EndpointID string `json:"endpoint_id"` + DNSName string `json:"dns_name"` + WGPort int `json:"wg_port"` + ServerPubkey string `json:"server_pubkey"` + TunnelSubnet string `json:"tunnel_subnet"` + PBSTunnelIP string `json:"pbs_tunnel_ip"` + } + if err := json.Unmarshal(body, &req); err != nil { + http.Error(w, "Invalid payload: body must be JSON", http.StatusBadRequest) + return + } + if req.DNSName == "" { + http.Error(w, "Invalid payload: dns_name required", http.StatusBadRequest) + return + } + if req.WGPort < 1 || req.WGPort > 65535 { + http.Error(w, "Invalid payload: wg_port must be 1-65535", http.StatusBadRequest) + return + } + if err := validateWGPubkey(req.ServerPubkey); err != nil { + http.Error(w, "Invalid payload: server_pubkey: "+err.Error(), http.StatusBadRequest) + return + } + prefix, err := netip.ParsePrefix(req.TunnelSubnet) + if err != nil { + http.Error(w, "Invalid payload: tunnel_subnet must be CIDR", http.StatusBadRequest) + return + } + pbsAddr, err := netip.ParseAddr(req.PBSTunnelIP) + if err != nil || !prefix.Contains(pbsAddr) { + http.Error(w, "Invalid payload: pbs_tunnel_ip must be an address inside tunnel_subnet", http.StatusBadRequest) + return + } + if err := h.store.SetWGEndpoint(&store.WGEndpoint{ + EndpointID: req.EndpointID, DNSName: req.DNSName, WGPort: req.WGPort, + ServerPubkey: req.ServerPubkey, TunnelSubnet: req.TunnelSubnet, PBSTunnelIP: req.PBSTunnelIP, + }); err != nil { + h.logger.Printf("[ERROR] set wg endpoint: %v", err) + http.Error(w, "Internal error", http.StatusInternalServerError) + return + } + h.logger.Printf("[INFO] wg endpoint set: %s (%s:%d, subnet %s)", req.DNSName, req.DNSName, req.WGPort, req.TunnelSubnet) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"status":"ok"}`)) +} + +// handleAdminGetWGEndpoint — GET /admin/wg/endpoint. +func (h *Handler) handleAdminGetWGEndpoint(w http.ResponseWriter, r *http.Request) { + _, _, isGlobal, ok := h.checkAuthHost(r) + if !ok || !isGlobal { + http.Error(w, "Forbidden: global key required", http.StatusForbidden) + return + } + e, err := h.store.GetWGEndpoint() + if err == sql.ErrNoRows { + http.Error(w, "wg endpoint not configured", http.StatusNotFound) + return + } + if err != nil { + h.logger.Printf("[ERROR] get wg endpoint: %v", err) + http.Error(w, "Internal error", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "endpoint_id": e.EndpointID, "dns_name": e.DNSName, "wg_port": e.WGPort, + "server_pubkey": e.ServerPubkey, "tunnel_subnet": e.TunnelSubnet, "pbs_tunnel_ip": e.PBSTunnelIP, + }) +} + +// handleAdminAddWGPeer — POST /admin/wg/peers. Allocates a /32 (idempotent on pubkey) and +// pushes the full list inline (sync semantics: ok | deferred | disabled). +func (h *Handler) handleAdminAddWGPeer(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"` + HostID string `json:"host_id"` + Note string `json:"note"` + } + 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 + } + ip, existed, err := h.store.AddWGPeer(req.Pubkey, req.HostID, req.Note) + if err == store.ErrWGEndpointUnset { + http.Error(w, "wg endpoint not configured", http.StatusConflict) + return + } + if err == store.ErrWGSubnetExhausted { + http.Error(w, "tunnel subnet exhausted", http.StatusConflict) + return + } + if err != nil { + h.logger.Printf("[ERROR] add wg peer: %v", err) + http.Error(w, "Internal error", http.StatusInternalServerError) + return + } + syncStatus := h.syncAfterMutation(r.Context()) + h.logger.Printf("[INFO] wg peer added: %s -> %s/32 (existed=%v, sync=%s)", req.Pubkey, ip, existed, syncStatus) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]interface{}{ + "pubkey": req.Pubkey, "assigned_ip": ip + "/32", "existed": existed, "sync": syncStatus, + }) +} + +// handleAdminDeleteWGPeer — DELETE /admin/wg/peers, pubkey in the JSON body (never the URL). +// Unknown pubkey → 404 with NO sync (nothing changed). Known → delete + inline push: the pushed +// full list no longer contains the peer, so revocation lands with the push (Scenario B). +func (h *Handler) handleAdminDeleteWGPeer(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"` + } + 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 + } + err = h.store.RemoveWGPeer(req.Pubkey) + if err == sql.ErrNoRows { + http.Error(w, "Unknown pubkey", http.StatusNotFound) + return + } + if err != nil { + h.logger.Printf("[ERROR] remove wg peer: %v", err) + http.Error(w, "Internal error", http.StatusInternalServerError) + return + } + syncStatus := h.syncAfterMutation(r.Context()) + h.logger.Printf("[INFO] wg peer removed: %s (sync=%s)", req.Pubkey, syncStatus) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "sync": syncStatus}) +} + +// handleAdminListWGPeers — GET /admin/wg/peers. The verification surface (S2 builds UI on top). +func (h *Handler) handleAdminListWGPeers(w http.ResponseWriter, r *http.Request) { + _, _, isGlobal, ok := h.checkAuthHost(r) + if !ok || !isGlobal { + http.Error(w, "Forbidden: global key required", http.StatusForbidden) + return + } + peers, err := h.store.ListWGPeers() + if err != nil { + h.logger.Printf("[ERROR] list wg peers: %v", err) + http.Error(w, "Internal error", http.StatusInternalServerError) + return + } + out := make([]map[string]interface{}, 0, len(peers)) + for _, p := range peers { + out = append(out, map[string]interface{}{ + "pubkey": p.Pubkey, "assigned_ip": p.AssignedIP + "/32", "host_id": p.HostID, "note": p.Note, + }) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{"peers": out}) +} diff --git a/hub/internal/api/wg_test.go b/hub/internal/api/wg_test.go new file mode 100644 index 0000000..11b92a1 --- /dev/null +++ b/hub/internal/api/wg_test.go @@ -0,0 +1,212 @@ +package api + +// Group B — WG admin API auth + validation (Scenario C). Non-hollow: asserts store effects and +// fake-syncer call counts, not just statuses. + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "net/http" + "testing" + + "gitea.dooplex.hu/admin/felhom-hub/internal/store" +) + +// fakeWGSyncer counts SyncNow/Trigger calls; err scripts the SyncNow result. +type fakeWGSyncer struct { + syncCalls int + triggerCalls int + err error +} + +func (f *fakeWGSyncer) SyncNow(ctx context.Context) error { f.syncCalls++; return f.err } +func (f *fakeWGSyncer) Trigger() { f.triggerCalls++ } + +// testPK returns a VALID WG-shaped pubkey (44 std-base64 chars, 32 bytes) unique per fill byte. +func testPK(fill byte) string { + return base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{fill}, 32)) +} + +func putTestEndpoint(t *testing.T, h *Handler) { + t.Helper() + body := `{"dns_name":"ep0.example","wg_port":443,"server_pubkey":"` + testPK(9) + `",` + + `"tunnel_subnet":"10.77.0.0/24","pbs_tunnel_ip":"10.77.0.1"}` + rr := do(h, http.MethodPut, "/admin/wg/endpoint", globalKey, body) + if rr.Code != http.StatusOK { + t.Fatalf("PUT endpoint = %d: %s", rr.Code, rr.Body.String()) + } +} + +func TestWGPeers_PerHostKeyForbidden(t *testing.T) { + h, st, _ := newTestHandler(t) + st.UpsertHost(&store.Host{HostID: "h1", CustomerID: "c1", APIKey: "HKEY"}) + putTestEndpoint(t, h) + fake := &fakeWGSyncer{} + h.SetWGSyncer(fake) + + for _, m := range []string{http.MethodPost, http.MethodDelete, http.MethodGet} { + rr := do(h, m, "/admin/wg/peers", "HKEY", `{"pubkey":"`+testPK(1)+`"}`) + if rr.Code != http.StatusForbidden { + t.Errorf("%s with per-host key = %d, want 403", m, rr.Code) + } + } + rr := do(h, http.MethodPut, "/admin/wg/endpoint", "HKEY", `{}`) + if rr.Code != http.StatusForbidden { + t.Errorf("PUT endpoint with per-host key = %d, want 403", rr.Code) + } + peers, _ := st.ListWGPeers() + if len(peers) != 0 { + t.Errorf("rows created despite 403: %d", len(peers)) + } + if fake.syncCalls != 0 { + t.Errorf("sync ran despite 403: %d calls", fake.syncCalls) + } +} + +func TestWGPeers_AddHappyPath(t *testing.T) { + h, st, _ := newTestHandler(t) + putTestEndpoint(t, h) + fake := &fakeWGSyncer{} + h.SetWGSyncer(fake) + + rr := do(h, http.MethodPost, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(1)+`","note":"test"}`) + if rr.Code != http.StatusOK { + t.Fatalf("POST = %d: %s", rr.Code, rr.Body.String()) + } + var resp struct { + Pubkey string `json:"pubkey"` + AssignedIP string `json:"assigned_ip"` + Existed bool `json:"existed"` + Sync string `json:"sync"` + } + json.Unmarshal(rr.Body.Bytes(), &resp) + if resp.AssignedIP != "10.77.0.2/32" || resp.Existed || resp.Sync != "ok" { + t.Errorf("resp = %+v, want .2/32 existed=false sync=ok", resp) + } + if fake.syncCalls != 1 { + t.Errorf("sync calls = %d, want 1", fake.syncCalls) + } + peers, _ := st.ListWGPeers() + if len(peers) != 1 || peers[0].AssignedIP != "10.77.0.2" || peers[0].Note != "test" { + t.Errorf("stored peers = %+v", peers) + } +} + +func TestWGPeers_SyncDeferredOnPushFailure(t *testing.T) { + h, _, _ := newTestHandler(t) + putTestEndpoint(t, h) + fake := &fakeWGSyncer{err: context.DeadlineExceeded} + h.SetWGSyncer(fake) + + rr := do(h, http.MethodPost, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(1)+`"}`) + if rr.Code != http.StatusOK { + t.Fatalf("POST = %d (DB write is source of truth; push failure must not fail the request)", rr.Code) + } + var resp struct { + Sync string `json:"sync"` + } + json.Unmarshal(rr.Body.Bytes(), &resp) + if len(resp.Sync) < 8 || resp.Sync[:8] != "deferred" { + t.Errorf("sync = %q, want deferred:...", resp.Sync) + } + if fake.triggerCalls != 1 { + t.Errorf("Trigger calls = %d, want 1 (reconciler retry requested)", fake.triggerCalls) + } +} + +func TestWGPeers_SyncDisabledWhenUnwired(t *testing.T) { + h, _, _ := newTestHandler(t) + putTestEndpoint(t, h) + // no SetWGSyncer — nil seam + rr := do(h, http.MethodPost, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(1)+`"}`) + if rr.Code != http.StatusOK { + t.Fatalf("POST = %d", rr.Code) + } + var resp struct { + Sync string `json:"sync"` + } + json.Unmarshal(rr.Body.Bytes(), &resp) + if resp.Sync != "disabled" { + t.Errorf("sync = %q, want disabled", resp.Sync) + } +} + +func TestWGPeers_BadPubkeyRejected(t *testing.T) { + h, st, _ := newTestHandler(t) + putTestEndpoint(t, h) + bad := []string{ + "not-base64", + base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{1}, 16)), // 24 chars + "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!", // 44 chars, not base64 + base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{1}, 33)), // 44 chars but 33 bytes + } + for _, pk := range bad { + rr := do(h, http.MethodPost, "/admin/wg/peers", globalKey, `{"pubkey":"`+pk+`"}`) + if rr.Code != http.StatusBadRequest { + t.Errorf("pubkey %q = %d, want 400", pk, rr.Code) + } + } + peers, _ := st.ListWGPeers() + if len(peers) != 0 { + t.Errorf("allocation happened for bad pubkey: %+v", peers) + } +} + +func TestWGPeers_NoEndpointIs409(t *testing.T) { + h, _, _ := newTestHandler(t) + rr := do(h, http.MethodPost, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(1)+`"}`) + if rr.Code != http.StatusConflict { + t.Errorf("POST without endpoint = %d, want 409", rr.Code) + } +} + +func TestWGEndpoint_PBSOutsideSubnetRejected(t *testing.T) { + h, _, _ := newTestHandler(t) + body := `{"dns_name":"ep0.example","wg_port":443,"server_pubkey":"` + testPK(9) + `",` + + `"tunnel_subnet":"10.77.0.0/24","pbs_tunnel_ip":"10.88.0.1"}` + rr := do(h, http.MethodPut, "/admin/wg/endpoint", globalKey, body) + if rr.Code != http.StatusBadRequest { + t.Errorf("PUT with pbs outside subnet = %d, want 400", rr.Code) + } + rr = do(h, http.MethodGet, "/admin/wg/endpoint", globalKey, "") + if rr.Code != http.StatusNotFound { + t.Errorf("GET after rejected PUT = %d, want 404 (nothing stored)", rr.Code) + } +} + +func TestWGPeers_DeleteUnknown404NoSync(t *testing.T) { + h, _, _ := newTestHandler(t) + putTestEndpoint(t, h) + fake := &fakeWGSyncer{} + h.SetWGSyncer(fake) + rr := do(h, http.MethodDelete, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(7)+`"}`) + if rr.Code != http.StatusNotFound { + t.Errorf("DELETE unknown = %d, want 404", rr.Code) + } + if fake.syncCalls != 0 { + t.Errorf("sync ran on 404 delete: %d", fake.syncCalls) + } +} + +func TestWGPeers_DeleteKnownRemovesAndSyncs(t *testing.T) { + h, st, _ := newTestHandler(t) + putTestEndpoint(t, h) + fake := &fakeWGSyncer{} + h.SetWGSyncer(fake) + do(h, http.MethodPost, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(1)+`"}`) + do(h, http.MethodPost, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(2)+`"}`) + + rr := do(h, http.MethodDelete, "/admin/wg/peers", globalKey, `{"pubkey":"`+testPK(1)+`"}`) + if rr.Code != http.StatusOK { + t.Fatalf("DELETE = %d: %s", rr.Code, rr.Body.String()) + } + peers, _ := st.ListWGPeers() + if len(peers) != 1 || peers[0].Pubkey != testPK(2) { + t.Errorf("peers after delete = %+v, want only p2", peers) + } + if fake.syncCalls != 3 { // 2 adds + 1 delete + t.Errorf("sync calls = %d, want 3", fake.syncCalls) + } +} diff --git a/hub/internal/wgsync/client.go b/hub/internal/wgsync/client.go new file mode 100644 index 0000000..da3c511 --- /dev/null +++ b/hub/internal/wgsync/client.go @@ -0,0 +1,132 @@ +// Package wgsync pushes the hub's WG peer registry to the offsite endpoint (S1, doc 06 §5). +// It is the structural sibling of internal/cloudflare: the hub holds the credential and drives +// external infra; the endpoint stays a dumb, runbook-provisioned box. Transport is SSH with a +// PINNED host key (the internal/pbs pin posture — exact-match or refuse; there is no insecure +// fallback), to a forced-command reconcile script server-side, so even this credential's theft +// bounds the attacker to "mutate the peer list" (doc 06 §3.1 blast radius). +package wgsync + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log" + "net" + "strings" + "time" + + "golang.org/x/crypto/ssh" +) + +// Config configures the SSH push client. All values come from the deployment env / mounted +// Secret (operator infra — never from a customer record). +type Config struct { + Addr string // "host:22" + User string // "felhom-peersync" + PrivateKey []byte // PEM private key (from the mounted Secret file) + HostKeyLine string // single authorized_keys-format line of the endpoint's host pubkey + Timeout time.Duration // default 30s +} + +// Client is a pinned-host-key SSH pusher. Construct with New (parses keys up front). +type Client struct { + addr string + user string + signer ssh.Signer + hostKey ssh.PublicKey + timeout time.Duration + logger *log.Logger +} + +// New builds a Client, failing early on an unparsable private key or host-key line. +func New(cfg Config, logger *log.Logger) (*Client, error) { + if cfg.Addr == "" || cfg.User == "" { + return nil, fmt.Errorf("wgsync: Addr and User are required") + } + signer, err := ssh.ParsePrivateKey(cfg.PrivateKey) + if err != nil { + return nil, fmt.Errorf("wgsync: parse private key: %w", err) + } + hostKey, _, _, _, err := ssh.ParseAuthorizedKey([]byte(cfg.HostKeyLine)) + if err != nil { + return nil, fmt.Errorf("wgsync: parse host key line: %w", err) + } + timeout := cfg.Timeout + if timeout == 0 { + timeout = 30 * time.Second + } + if logger == nil { + logger = log.Default() + } + return &Client{ + addr: cfg.Addr, user: cfg.User, signer: signer, + hostKey: hostKey, timeout: timeout, logger: logger, + }, nil +} + +// pushResponse is the peersync script's stdout contract ({"status":"ok","applied":N}). +type pushResponse struct { + Status string `json:"status"` + Applied int `json:"applied"` +} + +// Push sends the payload to the endpoint's forced-command script over one SSH session and +// verifies the script's ok-response. The host key is pinned (ssh.FixedHostKey) — a wrong key is +// a refused connection, never a prompt or a fallback. +func (c *Client) Push(ctx context.Context, payload []byte) error { + sshCfg := &ssh.ClientConfig{ + User: c.user, + Auth: []ssh.AuthMethod{ssh.PublicKeys(c.signer)}, + HostKeyCallback: ssh.FixedHostKey(c.hostKey), + Timeout: c.timeout, + } + dialer := net.Dialer{Timeout: c.timeout} + conn, err := dialer.DialContext(ctx, "tcp", c.addr) + if err != nil { + return fmt.Errorf("wgsync: dial %s: %w", c.addr, err) + } + // Hand the ssh handshake a deadline too — DialContext's ctx stops applying after Dial. + if dl, ok := ctx.Deadline(); ok { + conn.SetDeadline(dl) + } else { + conn.SetDeadline(time.Now().Add(c.timeout)) + } + sconn, chans, reqs, err := ssh.NewClientConn(conn, c.addr, sshCfg) + if err != nil { + conn.Close() + return fmt.Errorf("wgsync: ssh handshake %s: %w", c.addr, err) + } + client := ssh.NewClient(sconn, chans, reqs) + defer client.Close() + conn.SetDeadline(time.Time{}) // handshake done; session I/O below is bounded by the same conn + if dl, ok := ctx.Deadline(); ok { + conn.SetDeadline(dl) + } + + session, err := client.NewSession() + if err != nil { + return fmt.Errorf("wgsync: session: %w", err) + } + defer session.Close() + + var stdout, stderr bytes.Buffer + session.Stdin = bytes.NewReader(payload) + session.Stdout = &stdout + session.Stderr = &stderr + + // The server's authorized_keys forced command overrides this string, but it MUST be set: + // some sshd configs log the requested command, and it documents intent on the wire. + if err := session.Run("felhom-peersync"); err != nil { + return fmt.Errorf("wgsync: remote peersync failed: %w (stderr: %s)", + err, strings.TrimSpace(stderr.String())) + } + + var resp pushResponse + if err := json.Unmarshal(bytes.TrimSpace(stdout.Bytes()), &resp); err != nil || resp.Status != "ok" { + return fmt.Errorf("wgsync: malformed peersync response %q (parse err: %v, stderr: %s)", + strings.TrimSpace(stdout.String()), err, strings.TrimSpace(stderr.String())) + } + c.logger.Printf("[INFO] wgsync: pushed %d peers to %s", resp.Applied, c.addr) + return nil +} diff --git a/hub/internal/wgsync/client_test.go b/hub/internal/wgsync/client_test.go new file mode 100644 index 0000000..c585931 --- /dev/null +++ b/hub/internal/wgsync/client_test.go @@ -0,0 +1,222 @@ +package wgsync + +// Group C — the SSH push client against an IN-PROCESS x/crypto/ssh server (Scenarios A/B/C-c4). +// The server captures the exact stdin payload, so tests assert the pushed BYTES; the host-key +// tests prove the pin both ways (correct key accepted, wrong key refused). + +import ( + "bytes" + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/pem" + "io" + "log" + "net" + "strings" + "sync" + "testing" + "time" + + "golang.org/x/crypto/ssh" +) + +// testKeys generates an ed25519 keypair and returns (PEM private key, ssh.Signer). +func testKeys(t *testing.T) ([]byte, ssh.Signer) { + t.Helper() + _, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("ed25519: %v", err) + } + block, err := ssh.MarshalPrivateKey(priv, "") + if err != nil { + t.Fatalf("MarshalPrivateKey: %v", err) + } + signer, err := ssh.NewSignerFromKey(priv) + if err != nil { + t.Fatalf("NewSignerFromKey: %v", err) + } + return pem.EncodeToMemory(block), signer +} + +// testServer is a single-shot in-process SSH server handling one exec session. +type testServer struct { + addr string + mu sync.Mutex + captured []byte // stdin the "script" received + cmd string // the exec command string requested +} + +// startTestServer runs an SSH server that accepts clientSigner's key, serves with hostSigner, +// reads all stdin, replies stdoutResp/stderrResp and exitStatus. +func startTestServer(t *testing.T, hostSigner ssh.Signer, clientSigner ssh.Signer, + stdoutResp, stderrResp string, exitStatus uint32) *testServer { + t.Helper() + cfg := &ssh.ServerConfig{ + PublicKeyCallback: func(conn ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) { + if bytes.Equal(key.Marshal(), clientSigner.PublicKey().Marshal()) { + return &ssh.Permissions{}, nil + } + return nil, io.EOF + }, + } + cfg.AddHostKey(hostSigner) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { ln.Close() }) + srv := &testServer{addr: ln.Addr().String()} + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + sconn, chans, reqs, err := ssh.NewServerConn(conn, cfg) + if err != nil { + return // e.g. the wrong-host-key client aborts the handshake + } + defer sconn.Close() + go ssh.DiscardRequests(reqs) + for newCh := range chans { + if newCh.ChannelType() != "session" { + newCh.Reject(ssh.UnknownChannelType, "unsupported") + continue + } + ch, chReqs, err := newCh.Accept() + if err != nil { + continue + } + go func() { + for req := range chReqs { + if req.Type == "exec" { + var p struct{ Command string } + ssh.Unmarshal(req.Payload, &p) + srv.mu.Lock() + srv.cmd = p.Command + srv.mu.Unlock() + req.Reply(true, nil) + data, _ := io.ReadAll(ch) // the pushed payload (client EOFs stdin) + srv.mu.Lock() + srv.captured = data + srv.mu.Unlock() + if stderrResp != "" { + ch.Stderr().Write([]byte(stderrResp)) + } + if stdoutResp != "" { + ch.Write([]byte(stdoutResp)) + } + ch.SendRequest("exit-status", false, ssh.Marshal(struct{ Status uint32 }{exitStatus})) + ch.Close() + return + } + req.Reply(false, nil) + } + }() + } + }() + return srv +} + +func hostKeyLine(t *testing.T, s ssh.Signer) string { + t.Helper() + return strings.TrimSpace(string(ssh.MarshalAuthorizedKey(s.PublicKey()))) +} + +func newTestClient(t *testing.T, addr, hostKey string, clientPEM []byte) *Client { + t.Helper() + c, err := New(Config{ + Addr: addr, User: "felhom-peersync", PrivateKey: clientPEM, + HostKeyLine: hostKey, Timeout: 5 * time.Second, + }, log.New(io.Discard, "", 0)) + if err != nil { + t.Fatalf("wgsync.New: %v", err) + } + return c +} + +func TestPush_PayloadDeliveredExactly(t *testing.T) { + clientPEM, clientSigner := testKeys(t) + _, hostSigner := testKeys(t) + srv := startTestServer(t, hostSigner, clientSigner, `{"status":"ok","applied":1}`, "", 0) + c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM) + + payload := []byte(`{"version":1,"interface":"wg0","peers":[{"pubkey":"PK1","allowed_ip":"10.77.0.2/32"}]}`) + if err := c.Push(context.Background(), payload); err != nil { + t.Fatalf("Push: %v", err) + } + srv.mu.Lock() + defer srv.mu.Unlock() + if !bytes.Equal(srv.captured, payload) { + t.Errorf("server captured %q, want the exact payload %q", srv.captured, payload) + } + if srv.cmd != "felhom-peersync" { + t.Errorf("exec command = %q, want felhom-peersync (documents intent even under forced command)", srv.cmd) + } +} + +func TestPush_RemoteFailureSurfacesStderr(t *testing.T) { + clientPEM, clientSigner := testKeys(t) + _, hostSigner := testKeys(t) + srv := startTestServer(t, hostSigner, clientSigner, "", "boom", 1) + c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM) + + err := c.Push(context.Background(), []byte(`{}`)) + if err == nil { + t.Fatal("Push succeeded against exit-1 server") + } + if !strings.Contains(err.Error(), "boom") { + t.Errorf("error %q does not carry the remote stderr", err) + } +} + +func TestPush_MalformedResponseIsError(t *testing.T) { + clientPEM, clientSigner := testKeys(t) + _, hostSigner := testKeys(t) + srv := startTestServer(t, hostSigner, clientSigner, "garbage-not-json", "", 0) + c := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM) + + if err := c.Push(context.Background(), []byte(`{}`)); err == nil { + t.Fatal("Push accepted a malformed script response — it must not guess") + } +} + +func TestPush_WrongHostKeyRefused(t *testing.T) { + clientPEM, clientSigner := testKeys(t) + _, hostSigner := testKeys(t) + _, otherSigner := testKeys(t) // NOT the server's key + srv := startTestServer(t, hostSigner, clientSigner, `{"status":"ok","applied":0}`, "", 0) + c := newTestClient(t, srv.addr, hostKeyLine(t, otherSigner), clientPEM) + + err := c.Push(context.Background(), []byte(`{}`)) + if err == nil { + t.Fatal("Push succeeded against a server with the WRONG host key — the pin is dead") + } + if !strings.Contains(err.Error(), "host key") && !strings.Contains(err.Error(), "handshake") { + t.Errorf("error %q is not a host-key refusal", err) + } + srv.mu.Lock() + defer srv.mu.Unlock() + if len(srv.captured) != 0 { + t.Errorf("payload leaked to a mis-keyed server: %q", srv.captured) + } +} + +func TestNew_BadInputsFailEarly(t *testing.T) { + clientPEM, _ := testKeys(t) + _, hostSigner := testKeys(t) + hk := hostKeyLine(t, hostSigner) + logger := log.New(io.Discard, "", 0) + + if _, err := New(Config{Addr: "x:22", User: "u", PrivateKey: []byte("not-a-key"), HostKeyLine: hk}, logger); err == nil { + t.Error("bad private key accepted") + } + if _, err := New(Config{Addr: "x:22", User: "u", PrivateKey: clientPEM, HostKeyLine: "not a key line"}, logger); err == nil { + t.Error("bad host key line accepted") + } + if _, err := New(Config{User: "u", PrivateKey: clientPEM, HostKeyLine: hk}, logger); err == nil { + t.Error("missing addr accepted") + } +} diff --git a/hub/internal/wgsync/reconciler.go b/hub/internal/wgsync/reconciler.go new file mode 100644 index 0000000..48ae98d --- /dev/null +++ b/hub/internal/wgsync/reconciler.go @@ -0,0 +1,106 @@ +package wgsync + +import ( + "context" + "encoding/json" + "fmt" + "log" + "time" + + "gitea.dooplex.hu/admin/felhom-hub/internal/store" +) + +// syncer is the push seam — satisfied by *Client; tests inject a fake to count calls and +// capture payloads without SSH. +type syncer interface { + Push(ctx context.Context, payload []byte) error +} + +// Reconciler keeps the endpoint's WG peer list converged on the hub DB (the source of truth). +// DECLARATIVE: every push carries the FULL desired list — never deltas — so endpoint drift +// (a manually-added peer, a missed earlier push) is erased on the next push, and a failed push +// self-heals on the next tick. This full-list property is load-bearing (Scenario D drift +// repair) — do not "optimize" it into deltas. +type Reconciler struct { + store *store.Store + sync syncer + trigger chan struct{} + interval time.Duration // tick period; tests shrink it + logger *log.Logger +} + +// NewReconciler builds a Reconciler over the store and a syncer (the SSH client in production). +func NewReconciler(st *store.Store, sync syncer, logger *log.Logger) *Reconciler { + if logger == nil { + logger = log.Default() + } + return &Reconciler{ + store: st, + sync: sync, + trigger: make(chan struct{}, 1), + interval: 5 * time.Minute, + logger: logger, + } +} + +// Trigger requests an immediate sync from Run's loop. Non-blocking: a pending trigger already +// covers this request (the full list is pushed either way). +func (r *Reconciler) Trigger() { + select { + case r.trigger <- struct{}{}: + default: + } +} + +// payload builds the versioned, deterministic sync document from the store. +// {"version":1,"interface":"wg0","peers":[{"pubkey":"...","allowed_ip":"/32"}]} +func (r *Reconciler) payload() ([]byte, error) { + peers, err := r.store.ListWGPeers() + if err != nil { + return nil, fmt.Errorf("list peers: %w", err) + } + type wirePeer struct { + Pubkey string `json:"pubkey"` + AllowedIP string `json:"allowed_ip"` + } + doc := struct { + Version int `json:"version"` + Interface string `json:"interface"` + Peers []wirePeer `json:"peers"` + }{Version: 1, Interface: "wg0", Peers: make([]wirePeer, 0, len(peers))} + for _, p := range peers { + doc.Peers = append(doc.Peers, wirePeer{Pubkey: p.Pubkey, AllowedIP: p.AssignedIP + "/32"}) + } + return json.Marshal(doc) +} + +// SyncNow builds the current full-list payload and pushes it. Called inline by the mutation +// handlers (short ctx) and by Run's loop. +func (r *Reconciler) SyncNow(ctx context.Context) error { + payload, err := r.payload() + if err != nil { + return err + } + return r.sync.Push(ctx, payload) +} + +// Run loops until ctx is done: an explicit Trigger OR the periodic tick pushes the full list. +// The periodic push happens even with zero mutations — that is the drift repair. Errors are +// logged and retried on the next signal; Run never exits early. +func (r *Reconciler) Run(ctx context.Context) { + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-r.trigger: + case <-ticker.C: + } + syncCtx, cancel := context.WithTimeout(ctx, 60*time.Second) + if err := r.SyncNow(syncCtx); err != nil { + r.logger.Printf("[ERROR] wgsync reconcile: %v (will retry on next tick)", err) + } + cancel() + } +} diff --git a/hub/internal/wgsync/reconciler_test.go b/hub/internal/wgsync/reconciler_test.go new file mode 100644 index 0000000..065442c --- /dev/null +++ b/hub/internal/wgsync/reconciler_test.go @@ -0,0 +1,206 @@ +package wgsync + +// Group D — reconciler (Scenario D). Non-hollow: asserts the exact pushed payloads (full list, +// deterministic order, absent-after-remove), retry-after-failure, and the no-mutation drift- +// repair tick. + +import ( + "context" + "encoding/json" + "io" + "log" + "path/filepath" + "sync" + "testing" + "time" + + "gitea.dooplex.hu/admin/felhom-hub/internal/store" +) + +type fakeSyncer struct { + mu sync.Mutex + payloads [][]byte + err error + notify chan struct{} +} + +func newFakeSyncer() *fakeSyncer { return &fakeSyncer{notify: make(chan struct{}, 16)} } + +func (f *fakeSyncer) Push(ctx context.Context, payload []byte) error { + f.mu.Lock() + cp := append([]byte(nil), payload...) + f.payloads = append(f.payloads, cp) + err := f.err + f.mu.Unlock() + select { + case f.notify <- struct{}{}: + default: + } + return err +} + +func (f *fakeSyncer) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.payloads) +} + +func (f *fakeSyncer) last() []byte { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.payloads) == 0 { + return nil + } + return f.payloads[len(f.payloads)-1] +} + +func waitPush(t *testing.T, f *fakeSyncer) { + t.Helper() + select { + case <-f.notify: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for a push") + } +} + +func newWGStore(t *testing.T) *store.Store { + t.Helper() + s, err := store.New(filepath.Join(t.TempDir(), "test.db"), log.New(io.Discard, "", 0)) + if err != nil { + t.Fatalf("store.New: %v", err) + } + t.Cleanup(func() { s.Close() }) + if err := s.SetWGEndpoint(&store.WGEndpoint{ + DNSName: "ep0.example", WGPort: 443, ServerPubkey: "SPK", + TunnelSubnet: "10.77.0.0/24", PBSTunnelIP: "10.77.0.1", + }); err != nil { + t.Fatalf("SetWGEndpoint: %v", err) + } + return s +} + +func TestReconciler_TriggerPushesFullList(t *testing.T) { + st := newWGStore(t) + st.AddWGPeer("PKA", "", "") + st.AddWGPeer("PKB", "", "") + fake := newFakeSyncer() + r := NewReconciler(st, fake, log.New(io.Discard, "", 0)) + r.interval = time.Hour // tick out of the picture + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go r.Run(ctx) + + r.Trigger() + waitPush(t, fake) + if fake.count() != 1 { + t.Fatalf("push count = %d, want exactly 1", fake.count()) + } + var doc struct { + Version int `json:"version"` + Interface string `json:"interface"` + Peers []struct { + Pubkey string `json:"pubkey"` + AllowedIP string `json:"allowed_ip"` + } `json:"peers"` + } + if err := json.Unmarshal(fake.last(), &doc); err != nil { + t.Fatalf("payload not JSON: %v", err) + } + if doc.Version != 1 || doc.Interface != "wg0" || len(doc.Peers) != 2 { + t.Fatalf("payload = %s", fake.last()) + } + if doc.Peers[0].Pubkey != "PKA" || doc.Peers[0].AllowedIP != "10.77.0.2/32" || + doc.Peers[1].Pubkey != "PKB" || doc.Peers[1].AllowedIP != "10.77.0.3/32" { + t.Errorf("peers = %+v (order/content)", doc.Peers) + } +} + +func TestReconciler_FailureRetriedOnTick(t *testing.T) { + st := newWGStore(t) + st.AddWGPeer("PKA", "", "") + fake := newFakeSyncer() + fake.err = context.DeadlineExceeded // every push fails + r := NewReconciler(st, fake, log.New(io.Discard, "", 0)) + r.interval = 30 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go r.Run(ctx) + + r.Trigger() + waitPush(t, fake) // the failing triggered push + waitPush(t, fake) // the tick retry — Run must not have exited on the error + if fake.count() < 2 { + t.Fatalf("push count = %d, want >=2 (retry after failure)", fake.count()) + } +} + +func TestReconciler_TickPushesWithoutMutations(t *testing.T) { + st := newWGStore(t) // zero peers, zero mutations + fake := newFakeSyncer() + r := NewReconciler(st, fake, log.New(io.Discard, "", 0)) + r.interval = 30 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go r.Run(ctx) + + waitPush(t, fake) // pure tick — this IS the drift repair + var doc struct { + Peers []interface{} `json:"peers"` + } + if err := json.Unmarshal(fake.last(), &doc); err != nil { + t.Fatalf("payload not JSON: %v", err) + } + if len(doc.Peers) != 0 { + t.Errorf("empty registry must push an empty (not absent) peer list: %s", fake.last()) + } +} + +func TestReconciler_RemovedPeerAbsentFromNextPayload(t *testing.T) { + st := newWGStore(t) + st.AddWGPeer("PKA", "", "") + st.AddWGPeer("PKB", "", "") + fake := newFakeSyncer() + r := NewReconciler(st, fake, log.New(io.Discard, "", 0)) + + if err := r.SyncNow(context.Background()); err != nil { + t.Fatalf("SyncNow: %v", err) + } + if err := st.RemoveWGPeer("PKA"); err != nil { + t.Fatalf("RemoveWGPeer: %v", err) + } + if err := r.SyncNow(context.Background()); err != nil { + t.Fatalf("SyncNow 2: %v", err) + } + last := string(fake.last()) + // Assert the NEGATIVE: the removed pubkey is gone from the pushed bytes. + if json.Valid([]byte(last)) == false { + t.Fatalf("payload not JSON: %s", last) + } + if contains := jsonContainsPubkey(t, []byte(last), "PKA"); contains { + t.Errorf("removed peer PKA still in payload: %s", last) + } + if contains := jsonContainsPubkey(t, []byte(last), "PKB"); !contains { + t.Errorf("surviving peer PKB missing from payload: %s", last) + } +} + +func jsonContainsPubkey(t *testing.T, payload []byte, pk string) bool { + t.Helper() + var doc struct { + Peers []struct { + Pubkey string `json:"pubkey"` + } `json:"peers"` + } + if err := json.Unmarshal(payload, &doc); err != nil { + t.Fatalf("payload parse: %v", err) + } + for _, p := range doc.Peers { + if p.Pubkey == pk { + return true + } + } + return false +}