hub: S1 wgsync (pinned-SSH push + declarative reconciler) + /admin/wg API + env wiring
internal/wgsync: x/crypto/ssh client with ssh.FixedHostKey pin (no insecure fallback), forced-command exec, ok/applied response contract; Reconciler pushes the FULL peer list on Trigger or 5-min tick (drift repair by construction). internal/api/wg.go: PUT/GET /admin/wg/endpoint + POST/DELETE/GET /admin/wg/peers, global-key-only, pubkey in body (base64 vs URL), sync ok|deferred|disabled. main.go: WG_ENDPOINT_SSH_* env wiring, disabled-with-INFO when unconfigured. Groups B/C/D tests incl. in-process SSH server; red-proofs b/c/d run + reverted. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
This commit is contained in:
@@ -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)
|
||||
|
||||
|
||||
@@ -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":
|
||||
|
||||
@@ -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})
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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":"<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()
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user