diff --git a/internal/server/server.go b/internal/server/server.go index 9f7cd5f..3c2587b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -922,6 +922,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("/api/netguard/zones", s.withAuth("netguard:read", s.handleNetGuardZones)) mux.HandleFunc("/api/netguard/zones/delete", s.withAuth("netguard:admin", s.handleDeleteGuardZone)) mux.HandleFunc("/api/netguard/nodes", s.withAuth("netguard:read", s.handleNetGuardNodes)) + mux.HandleFunc("/api/netguard/reality", s.withAuth("", s.handleNetGuardReality)) mux.HandleFunc("/api/netguard/bindings", s.withAuth("", s.handleNetGuardBindings)) mux.HandleFunc("/api/netguard/nodes/adopt", s.withAuth("", s.handleNetGuardAdopt)) mux.HandleFunc("/api/netguard/plan", s.withAuth("", s.handleNetGuardPlan)) @@ -938,6 +939,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("/api/agent/metrics", s.withAgentLimit(s.handleAgentMetrics)) mux.HandleFunc("/api/agent/proxy-usage", s.withAgentLimit(s.handleAgentProxyUsage)) mux.HandleFunc("/api/agent/singbox-inventory", s.withAgentLimit(s.handleAgentSingBoxInventory)) + mux.HandleFunc("/api/agent/guard-reality", s.withAgentLimit(s.handleAgentGuardReality)) mux.HandleFunc("/api/agent/tasks", s.withAgentLimit(s.handleAgentTasks)) mux.HandleFunc("/api/agent/task-result", s.withAgentLimit(s.handleAgentTaskResult)) mux.HandleFunc("/api/agent/terminal/sessions", s.withAgentLimit(s.handleAgentTerminalSessions)) diff --git a/internal/server/server_netguard_reality.go b/internal/server/server_netguard_reality.go new file mode 100644 index 0000000..7e4ddb7 --- /dev/null +++ b/internal/server/server_netguard_reality.go @@ -0,0 +1,461 @@ +package server + +import ( + "encoding/base64" + "errors" + "fmt" + "net/http" + "net/netip" + "sort" + "strconv" + "strings" + "time" + "unicode" + "unicode/utf8" + + "github.com/LatticeNet/lattice-sdk/model" + "github.com/LatticeNet/lattice-server/internal/id" + "github.com/LatticeNet/lattice-server/internal/rbac" + "github.com/LatticeNet/lattice-server/internal/store" +) + +const ( + guardRealityFutureSlack = 5 * time.Minute + guardRealityStaleAfter = 30 * time.Hour + guardRealityDefaultLimit = 100 + guardRealityMaxLimit = 500 + guardRealityMaxListeners = 4096 + guardRealityMaxInterfaces = 256 + guardRealityMaxIfaceAddresses = 64 + guardRealityMaxForeignTables = 512 + guardRealityMaxStringBytes = 256 + guardRealityMaxIfaceNameBytes = 64 +) + +type guardRealityResponse struct { + OK bool `json:"ok"` + NodeID string `json:"node_id"` + CollectedAt time.Time `json:"collected_at"` + ReceivedAt time.Time `json:"received_at"` + CollectedAtClamped bool `json:"collected_at_clamped"` +} + +type guardRealitySummary struct { + NodeID string `json:"node_id"` + SnapshotStatus string `json:"snapshot_status"` + CollectedAt *time.Time `json:"collected_at,omitempty"` + ReceivedAt *time.Time `json:"received_at,omitempty"` + StaleAfter *time.Time `json:"stale_after,omitempty"` + ManagedSHA *string `json:"managed_sha,omitempty"` + ListenerCount *int `json:"listener_count,omitempty"` + InterfaceCount *int `json:"interface_count,omitempty"` + ForeignTableCount *int `json:"foreign_table_count,omitempty"` +} + +type guardRealityListResponse struct { + Nodes []guardRealitySummary `json:"nodes"` + NextCursor string `json:"next_cursor,omitempty"` +} + +type guardRealityDetailResponse struct { + Node guardRealityDetail `json:"node"` +} + +type guardRealityDetail struct { + NodeID string `json:"node_id"` + SnapshotStatus string `json:"snapshot_status"` + Reality *model.GuardNodeReality `json:"reality"` + ReceivedAt *time.Time `json:"received_at"` + StaleAfter *time.Time `json:"stale_after"` +} + +func (s *Server) handleAgentGuardReality(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, errors.New("method not allowed")) + return + } + var req struct { + agentAuthRequest + Reality model.GuardNodeReality `json:"reality"` + } + if !decodeAgentJSON(w, r, &req) { + return + } + req.NodeID = strings.TrimSpace(req.NodeID) + if req.NodeID == "" { + writeError(w, http.StatusBadRequest, apiError(model.APIErrorBadRequest, "node_id is required")) + return + } + node, ok := s.authenticateAgentRequest(r, req.NodeID) + if !ok { + writeError(w, http.StatusUnauthorized, apiError(model.APIErrorInvalidNodeToken, "invalid node token")) + return + } + if realityNodeID := strings.TrimSpace(req.Reality.NodeID); realityNodeID != "" && realityNodeID != node.ID { + writeError(w, http.StatusBadRequest, apiError(model.APIErrorBadRequest, "reality node_id does not match authenticated node")) + return + } + req.Reality.NodeID = node.ID + receivedAt := s.now().UTC() + reality, clamped, err := normalizeGuardReality(req.Reality, receivedAt) + if err != nil { + writeError(w, http.StatusBadRequest, apiError(model.APIErrorBadRequest, err.Error())) + return + } + stored, _, err := s.store.UpsertGuardRealitySnapshot(node.LatticeIdentityUUID, store.GuardRealitySnapshot{ + Reality: reality, + ReceivedAt: receivedAt, + }) + if errors.Is(err, store.ErrGuardRealityDurabilityDegraded) { + if s.logger != nil { + s.logger.Printf("guard reality committed with degraded durability: node_id=%s: %v", node.ID, err) + } + err = nil + } + if errors.Is(err, store.ErrGuardRealityNodeChanged) { + writeError(w, http.StatusUnauthorized, apiError(model.APIErrorInvalidNodeToken, "invalid node token")) + return + } + if errors.Is(err, store.ErrGuardRealityStale) { + writeError(w, http.StatusConflict, apiError("guard_reality_stale", "guard reality snapshot is stale")) + return + } + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + s.recordRequestAudit(r, model.AuditEvent{ + ID: id.New("audit"), + Action: "netguard.reality.report", + Decision: "allow", + NodeID: node.ID, + Metadata: map[string]string{ + "listener_count": strconv.Itoa(len(reality.Listeners)), + "interface_count": strconv.Itoa(len(reality.Interfaces)), + "foreign_table_count": strconv.Itoa(len(reality.ForeignTables)), + }, + }) + writeJSON(w, http.StatusOK, guardRealityResponse{ + OK: true, + NodeID: node.ID, + CollectedAt: stored.Reality.CollectedAt, + ReceivedAt: stored.ReceivedAt, + CollectedAtClamped: clamped, + }) +} + +func (s *Server) handleNetGuardReality(w http.ResponseWriter, r *http.Request, p principal) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, errors.New("method not allowed")) + return + } + if !s.requireScope(w, p, "netguard:read") { + return + } + q := r.URL.Query() + nodeID := strings.TrimSpace(q.Get("node_id")) + if nodeID != "" { + if q.Get("cursor") != "" || q.Get("limit") != "" { + writeError(w, http.StatusBadRequest, apiError(model.APIErrorBadRequest, "node_id cannot be combined with pagination")) + return + } + if _, ok := s.store.Node(nodeID); !ok || !rbac.Allows(p.Principal, "netguard:read", nodeID) { + writeError(w, http.StatusNotFound, apiError(model.APIErrorNotFound, "not found")) + return + } + writeJSON(w, http.StatusOK, guardRealityDetailResponse{ + Node: s.guardRealityDetailForNode(nodeID, s.now().UTC()), + }) + return + } + limit, err := parseGuardRealityLimit(q.Get("limit")) + if err != nil { + writeError(w, http.StatusBadRequest, apiError(model.APIErrorBadRequest, err.Error())) + return + } + afterNodeID := "" + if cursor := strings.TrimSpace(q.Get("cursor")); cursor != "" { + afterNodeID, err = decodeGuardRealityCursor(cursor) + if err != nil { + writeError(w, http.StatusBadRequest, apiError(model.APIErrorBadRequest, "invalid cursor")) + return + } + } + nodes := s.visibleGuardRealityNodes(p) + start := sort.Search(len(nodes), func(i int) bool { + return nodes[i].ID > afterNodeID + }) + if afterNodeID == "" { + start = 0 + } + end := start + limit + if end > len(nodes) { + end = len(nodes) + } + now := s.now().UTC() + out := make([]guardRealitySummary, 0, end-start) + for _, node := range nodes[start:end] { + out = append(out, s.guardRealitySummaryForNode(node.ID, now)) + } + resp := guardRealityListResponse{Nodes: out} + if end < len(nodes) && len(out) > 0 { + resp.NextCursor = encodeGuardRealityCursor(out[len(out)-1].NodeID) + } + writeJSON(w, http.StatusOK, resp) +} + +func (s *Server) visibleGuardRealityNodes(p principal) []model.Node { + nodes := s.store.Nodes() + out := nodes[:0] + for _, node := range nodes { + if rbac.Allows(p.Principal, "netguard:read", node.ID) { + out = append(out, node) + } + } + return out +} + +func (s *Server) guardRealitySummaryForNode(nodeID string, now time.Time) guardRealitySummary { + snapshot, ok := s.store.GuardRealitySnapshot(nodeID) + if !ok { + return guardRealitySummary{NodeID: nodeID, SnapshotStatus: "unknown"} + } + status, staleAfter := guardRealityFreshness(snapshot, now) + managedSHA := snapshot.Reality.ManagedSHA + listenerCount := len(snapshot.Reality.Listeners) + interfaceCount := len(snapshot.Reality.Interfaces) + foreignTableCount := len(snapshot.Reality.ForeignTables) + collectedAt := snapshot.Reality.CollectedAt.UTC() + receivedAt := snapshot.ReceivedAt.UTC() + return guardRealitySummary{ + NodeID: nodeID, + SnapshotStatus: status, + CollectedAt: &collectedAt, + ReceivedAt: &receivedAt, + StaleAfter: &staleAfter, + ManagedSHA: &managedSHA, + ListenerCount: &listenerCount, + InterfaceCount: &interfaceCount, + ForeignTableCount: &foreignTableCount, + } +} + +func (s *Server) guardRealityDetailForNode(nodeID string, now time.Time) guardRealityDetail { + snapshot, ok := s.store.GuardRealitySnapshot(nodeID) + if !ok { + return guardRealityDetail{ + NodeID: nodeID, + SnapshotStatus: "unknown", + Reality: nil, + ReceivedAt: nil, + StaleAfter: nil, + } + } + status, staleAfter := guardRealityFreshness(snapshot, now) + reality := snapshot.Reality + receivedAt := snapshot.ReceivedAt.UTC() + return guardRealityDetail{ + NodeID: nodeID, + SnapshotStatus: status, + Reality: &reality, + ReceivedAt: &receivedAt, + StaleAfter: &staleAfter, + } +} + +func guardRealityFreshness(snapshot store.GuardRealitySnapshot, now time.Time) (string, time.Time) { + staleAfter := snapshot.Reality.CollectedAt.UTC().Add(guardRealityStaleAfter) + if !now.UTC().Before(staleAfter) { + return "stale", staleAfter + } + return "fresh", staleAfter +} + +func parseGuardRealityLimit(raw string) (int, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return guardRealityDefaultLimit, nil + } + limit, err := strconv.Atoi(raw) + if err != nil || limit < 1 || limit > guardRealityMaxLimit { + return 0, fmt.Errorf("limit must be 1-%d", guardRealityMaxLimit) + } + return limit, nil +} + +func encodeGuardRealityCursor(nodeID string) string { + return base64.RawURLEncoding.EncodeToString([]byte(nodeID)) +} + +func decodeGuardRealityCursor(raw string) (string, error) { + decoded, err := base64.RawURLEncoding.DecodeString(raw) + if err != nil || len(decoded) == 0 { + return "", errors.New("invalid cursor") + } + nodeID := string(decoded) + if !utf8.ValidString(nodeID) || strings.TrimSpace(nodeID) == "" { + return "", errors.New("invalid cursor") + } + return nodeID, nil +} + +func normalizeGuardReality(reality model.GuardNodeReality, receivedAt time.Time) (model.GuardNodeReality, bool, error) { + reality.NodeID = strings.TrimSpace(reality.NodeID) + if reality.NodeID == "" { + return model.GuardNodeReality{}, false, errors.New("node_id is required") + } + if reality.CollectedAt.IsZero() { + return model.GuardNodeReality{}, false, errors.New("collected_at is required") + } + collectedAtClamped := false + reality.CollectedAt = reality.CollectedAt.UTC() + if reality.CollectedAt.After(receivedAt.UTC().Add(guardRealityFutureSlack)) { + reality.CollectedAt = receivedAt.UTC() + collectedAtClamped = true + } + if len(reality.Listeners) > guardRealityMaxListeners { + return model.GuardNodeReality{}, false, fmt.Errorf("listeners must contain at most %d entries", guardRealityMaxListeners) + } + if len(reality.Interfaces) > guardRealityMaxInterfaces { + return model.GuardNodeReality{}, false, fmt.Errorf("interfaces must contain at most %d entries", guardRealityMaxInterfaces) + } + if len(reality.ForeignTables) > guardRealityMaxForeignTables { + return model.GuardNodeReality{}, false, fmt.Errorf("foreign_tables must contain at most %d entries", guardRealityMaxForeignTables) + } + if err := normalizeGuardRealityListeners(reality.Listeners); err != nil { + return model.GuardNodeReality{}, false, err + } + if err := normalizeGuardRealityInterfaces(reality.Interfaces); err != nil { + return model.GuardNodeReality{}, false, err + } + if err := normalizeGuardRealityForeignTables(reality.ForeignTables); err != nil { + return model.GuardNodeReality{}, false, err + } + managedSHA, err := normalizePrintableString(reality.ManagedSHA, "managed_sha", guardRealityMaxStringBytes, false) + if err != nil { + return model.GuardNodeReality{}, false, err + } + if managedSHA != "" && !isLowerHex64(managedSHA) { + return model.GuardNodeReality{}, false, errors.New("managed_sha must be empty or 64 lowercase hex characters") + } + nftVersion, err := normalizePrintableString(reality.NFTVersion, "nft_version", guardRealityMaxStringBytes, false) + if err != nil { + return model.GuardNodeReality{}, false, err + } + reality.ManagedSHA = managedSHA + reality.NFTVersion = nftVersion + return reality, collectedAtClamped, nil +} + +func normalizeGuardRealityListeners(listeners []model.GuardListener) error { + for i := range listeners { + protocol, err := normalizePrintableString(listeners[i].Protocol, "listeners.protocol", guardRealityMaxStringBytes, true) + if err != nil { + return err + } + if protocol != "tcp" && protocol != "udp" { + return errors.New("listeners.protocol must be tcp or udp") + } + if listeners[i].Port < 1 || listeners[i].Port > 65535 { + return errors.New("listeners.port must be 1-65535") + } + address, err := normalizeGuardRealityAddress(listeners[i].Address, "listeners.address", false) + if err != nil { + return err + } + process, err := normalizePrintableString(listeners[i].Process, "listeners.process", guardRealityMaxStringBytes, false) + if err != nil { + return err + } + listeners[i].Protocol = protocol + listeners[i].Address = address + listeners[i].Process = process + } + return nil +} + +func normalizeGuardRealityInterfaces(interfaces []model.GuardInterface) error { + for i := range interfaces { + name, err := normalizePrintableString(interfaces[i].Name, "interfaces.name", guardRealityMaxIfaceNameBytes, true) + if err != nil { + return err + } + if len(interfaces[i].Addresses) > guardRealityMaxIfaceAddresses { + return fmt.Errorf("interfaces.addresses must contain at most %d entries", guardRealityMaxIfaceAddresses) + } + for j := range interfaces[i].Addresses { + address, err := normalizeGuardRealityAddress(interfaces[i].Addresses[j], "interfaces.addresses", true) + if err != nil { + return err + } + interfaces[i].Addresses[j] = address + } + interfaces[i].Name = name + } + return nil +} + +func normalizeGuardRealityForeignTables(tables []string) error { + for i := range tables { + table, err := normalizePrintableString(tables[i], "foreign_tables", guardRealityMaxStringBytes, true) + if err != nil { + return err + } + tables[i] = table + } + return nil +} + +func normalizeGuardRealityAddress(value, field string, required bool) (string, error) { + value, err := normalizePrintableString(value, field, guardRealityMaxStringBytes, required) + if err != nil || value == "" { + return value, err + } + if strings.Contains(value, "/") { + prefix, err := netip.ParsePrefix(value) + if err != nil { + return "", fmt.Errorf("%s must be an IP address or prefix", field) + } + return prefix.String(), nil + } + addr, err := netip.ParseAddr(value) + if err != nil { + return "", fmt.Errorf("%s must be an IP address or prefix", field) + } + return addr.String(), nil +} + +func normalizePrintableString(value, field string, maxBytes int, required bool) (string, error) { + if !utf8.ValidString(value) { + return "", fmt.Errorf("%s must be valid UTF-8", field) + } + value = strings.TrimSpace(value) + if required && value == "" { + return "", fmt.Errorf("%s is required", field) + } + if len(value) > maxBytes { + return "", fmt.Errorf("%s must be at most %d bytes", field, maxBytes) + } + for _, r := range value { + if !unicode.IsPrint(r) { + return "", fmt.Errorf("%s must contain printable characters only", field) + } + } + return value, nil +} + +func isLowerHex64(value string) bool { + if len(value) != 64 { + return false + } + for _, b := range []byte(value) { + if b >= '0' && b <= '9' { + continue + } + if b >= 'a' && b <= 'f' { + continue + } + return false + } + return true +} diff --git a/internal/server/server_netguard_reality_test.go b/internal/server/server_netguard_reality_test.go new file mode 100644 index 0000000..b0ac619 --- /dev/null +++ b/internal/server/server_netguard_reality_test.go @@ -0,0 +1,439 @@ +package server + +import ( + "encoding/json" + "io" + "net/http" + "strings" + "testing" + "time" + + "github.com/LatticeNet/lattice-sdk/model" + "github.com/LatticeNet/lattice-server/internal/store" +) + +type guardRealitySummaryTest struct { + NodeID string `json:"node_id"` + SnapshotStatus string `json:"snapshot_status"` + CollectedAt *time.Time `json:"collected_at,omitempty"` + ReceivedAt *time.Time `json:"received_at,omitempty"` + StaleAfter *time.Time `json:"stale_after,omitempty"` + ManagedSHA string `json:"managed_sha,omitempty"` + ListenerCount *int `json:"listener_count,omitempty"` + InterfaceCount *int `json:"interface_count,omitempty"` + ForeignTableCount *int `json:"foreign_table_count,omitempty"` +} + +type guardRealityListTest struct { + Nodes []guardRealitySummaryTest `json:"nodes"` + NextCursor string `json:"next_cursor,omitempty"` +} + +type guardRealityDetailTest struct { + Node struct { + NodeID string `json:"node_id"` + SnapshotStatus string `json:"snapshot_status"` + Reality *model.GuardNodeReality `json:"reality"` + ReceivedAt *time.Time `json:"received_at"` + StaleAfter *time.Time `json:"stale_after"` + } `json:"node"` +} + +func newGuardRealityServerForTest(t *testing.T, now *time.Time) (*Server, http.Handler, *store.Store, []*http.Cookie, string) { + t.Helper() + st, err := store.Open("") + if err != nil { + t.Fatalf("open store: %v", err) + } + srv, err := New(Options{ + Store: st, + AdminPassword: testAdminPass, + DisableRenewalScheduler: true, + }) + if err != nil { + t.Fatalf("new server: %v", err) + } + srv.now = func() time.Time { return now.UTC() } + handler := srv.Handler() + cookies, csrf := loginSession(t, handler) + return srv, handler, st, cookies, csrf +} + +func guardRealityFixture(nodeID string, collectedAt time.Time) model.GuardNodeReality { + return model.GuardNodeReality{ + NodeID: nodeID, + Listeners: []model.GuardListener{{ + Protocol: "tcp", + Port: 22, + Address: "2001:db8::10", + Process: "sshd", + }}, + Interfaces: []model.GuardInterface{{ + Name: "ens3", + Addresses: []string{"2001:db8::10/128"}, + Up: true, + }}, + ManagedSHA: strings.Repeat("a", 64), + ForeignTables: []string{"inet docker"}, + NFTVersion: "nftables v1.0.9", + CollectedAt: collectedAt, + } +} + +func postGuardRealityForTest(t *testing.T, handler http.Handler, token, nodeID string, reality model.GuardNodeReality) *httptestResponse { + t.Helper() + body, err := json.Marshal(map[string]any{ + "node_id": nodeID, + "reality": reality, + "future_agent_field": map[string]any{ + "ignored": true, + }, + }) + if err != nil { + t.Fatalf("marshal guard reality body: %v", err) + } + rec := doAgentRaw(t, handler, http.MethodPost, "/api/agent/guard-reality", string(body), token) + return &httptestResponse{code: rec.Code, body: rec.Body.String()} +} + +type httptestResponse struct { + code int + body string +} + +func assertAPIErrorCodeFromBody(t *testing.T, body string, want string) { + t.Helper() + var out model.APIErrorResponse + if err := json.Unmarshal([]byte(body), &out); err != nil { + t.Fatalf("decode error body %q: %v", body, err) + } + if out.Error.Code != want { + t.Fatalf("error code = %q, want %q; body=%s", out.Error.Code, want, body) + } +} + +func TestNetGuardRealityAgentWriteAndReadContract(t *testing.T) { + now := time.Date(2026, 7, 31, 13, 0, 1, 0, time.UTC) + srv, handler, st, cookies, csrf := newGuardRealityServerForTest(t, &now) + + tokenA := enrollNamedNodeToken(t, handler, cookies, csrf, "node-a", "Node A") + enrollNamedNodeToken(t, handler, cookies, csrf, "node-b", "Node B") + + collectedAt := now.Add(-time.Second) + reality := guardRealityFixture("node-a", collectedAt) + resp := postGuardRealityForTest(t, handler, tokenA, "node-a", reality) + if resp.code != http.StatusOK { + t.Fatalf("agent write status = %d, body=%s", resp.code, resp.body) + } + var accepted struct { + OK bool `json:"ok"` + NodeID string `json:"node_id"` + CollectedAt time.Time `json:"collected_at"` + ReceivedAt time.Time `json:"received_at"` + CollectedAtClamped bool `json:"collected_at_clamped"` + } + if err := json.Unmarshal([]byte(resp.body), &accepted); err != nil { + t.Fatalf("decode accepted response: %v", err) + } + if !accepted.OK || accepted.NodeID != "node-a" { + t.Fatalf("unexpected accepted response: %+v", accepted) + } + if !accepted.CollectedAt.Equal(collectedAt) || !accepted.ReceivedAt.Equal(now) || accepted.CollectedAtClamped { + t.Fatalf("unexpected accepted timestamps: %+v", accepted) + } + stored, ok := st.GuardRealitySnapshot("node-a") + if !ok { + t.Fatalf("snapshot not persisted") + } + if stored.Reality.NodeID != "node-a" || stored.Reality.Listeners[0].Process != "sshd" { + t.Fatalf("unexpected persisted snapshot: %+v", stored.Reality) + } + foundAudit := false + for _, ev := range st.AuditEvents() { + if ev.Action != "netguard.reality.report" { + continue + } + foundAudit = true + if ev.NodeID != "node-a" { + t.Fatalf("reality audit node_id = %q, want node-a", ev.NodeID) + } + wantMetadata := map[string]string{ + "listener_count": "1", + "interface_count": "1", + "foreign_table_count": "1", + } + if len(ev.Metadata) != len(wantMetadata) { + t.Fatalf("reality audit metadata = %+v, want counts only", ev.Metadata) + } + for key, want := range wantMetadata { + if ev.Metadata[key] != want { + t.Fatalf("reality audit metadata[%s] = %q, want %q", key, ev.Metadata[key], want) + } + } + } + if !foundAudit { + t.Fatalf("missing netguard.reality.report audit: %+v", st.AuditEvents()) + } + + listRes := doJSON(t, handler, http.MethodGet, "/api/netguard/reality", "", cookies, csrf) + defer listRes.Body.Close() + if listRes.StatusCode != http.StatusOK { + t.Fatalf("list status = %d", listRes.StatusCode) + } + listRaw, err := io.ReadAll(listRes.Body) + if err != nil { + t.Fatalf("read list body: %v", err) + } + for _, forbidden := range []string{"sshd", "2001:db8::10", "2001:db8::10/128", "inet docker"} { + if strings.Contains(string(listRaw), forbidden) { + t.Fatalf("summary response leaked detail %q: %s", forbidden, string(listRaw)) + } + } + var list guardRealityListTest + if err := json.Unmarshal(listRaw, &list); err != nil { + t.Fatalf("decode list response: %v", err) + } + if len(list.Nodes) != 2 { + t.Fatalf("list node count = %d, want 2: %s", len(list.Nodes), string(listRaw)) + } + if list.Nodes[0].NodeID != "node-a" || list.Nodes[0].SnapshotStatus != "fresh" { + t.Fatalf("node-a summary = %+v", list.Nodes[0]) + } + if list.Nodes[0].ListenerCount == nil || *list.Nodes[0].ListenerCount != 1 { + t.Fatalf("node-a listener_count = %+v", list.Nodes[0].ListenerCount) + } + if list.Nodes[0].InterfaceCount == nil || *list.Nodes[0].InterfaceCount != 1 { + t.Fatalf("node-a interface_count = %+v", list.Nodes[0].InterfaceCount) + } + if list.Nodes[0].ForeignTableCount == nil || *list.Nodes[0].ForeignTableCount != 1 { + t.Fatalf("node-a foreign_table_count = %+v", list.Nodes[0].ForeignTableCount) + } + if list.Nodes[1].NodeID != "node-b" || list.Nodes[1].SnapshotStatus != "unknown" { + t.Fatalf("node-b summary = %+v", list.Nodes[1]) + } + if list.Nodes[1].CollectedAt != nil || list.Nodes[1].ListenerCount != nil { + t.Fatalf("unknown node exposed snapshot-derived fields: %+v", list.Nodes[1]) + } + + now = collectedAt.Add(30 * time.Hour) + srv.now = func() time.Time { return now.UTC() } + detailRes := doJSON(t, handler, http.MethodGet, "/api/netguard/reality?node_id=node-a", "", cookies, csrf) + defer detailRes.Body.Close() + if detailRes.StatusCode != http.StatusOK { + t.Fatalf("detail status = %d", detailRes.StatusCode) + } + var detail guardRealityDetailTest + if err := json.NewDecoder(detailRes.Body).Decode(&detail); err != nil { + t.Fatalf("decode detail: %v", err) + } + if detail.Node.NodeID != "node-a" || detail.Node.SnapshotStatus != "stale" { + t.Fatalf("detail status = %+v", detail.Node) + } + if detail.Node.Reality == nil || detail.Node.Reality.Listeners[0].Process != "sshd" { + t.Fatalf("detail did not include full normalized reality: %+v", detail.Node.Reality) + } + if detail.Node.StaleAfter == nil || !detail.Node.StaleAfter.Equal(collectedAt.Add(30*time.Hour)) { + t.Fatalf("stale_after = %+v, want %s", detail.Node.StaleAfter, collectedAt.Add(30*time.Hour)) + } +} + +func TestNetGuardRealityValidationAndStaleConflicts(t *testing.T) { + now := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + _, handler, _, cookies, csrf := newGuardRealityServerForTest(t, &now) + tokenA := enrollNamedNodeToken(t, handler, cookies, csrf, "node-a", "Node A") + tokenB := enrollNamedNodeToken(t, handler, cookies, csrf, "node-b", "Node B") + tokenC := enrollNamedNodeToken(t, handler, cookies, csrf, "node-c", "Node C") + + missingNodeID := string(mustJSON(t, map[string]any{"reality": guardRealityFixture("node-a", now)})) + rec := doAgentRaw(t, handler, http.MethodPost, "/api/agent/guard-reality", missingNodeID, tokenA) + if rec.Code != http.StatusBadRequest { + t.Fatalf("missing node_id status = %d, body=%s", rec.Code, rec.Body.String()) + } + assertAPIErrorCodeFromBody(t, rec.Body.String(), model.APIErrorBadRequest) + + mismatch := guardRealityFixture("other-node", now) + body, err := json.Marshal(map[string]any{"node_id": "node-a", "reality": mismatch}) + if err != nil { + t.Fatalf("marshal mismatch: %v", err) + } + rec = doAgentRaw(t, handler, http.MethodPost, "/api/agent/guard-reality", string(body), tokenA) + if rec.Code != http.StatusBadRequest { + t.Fatalf("mismatched node status = %d, body=%s", rec.Code, rec.Body.String()) + } + assertAPIErrorCodeFromBody(t, rec.Body.String(), model.APIErrorBadRequest) + + rawWithToken := `{"node_id":"node-a","token":"` + tokenA + `","reality":` + string(mustJSON(t, guardRealityFixture("node-a", now))) + `}` + rec = doAgentRaw(t, handler, http.MethodPost, "/api/agent/guard-reality", rawWithToken, "") + if rec.Code != http.StatusUnauthorized { + t.Fatalf("body token auth status = %d, body=%s", rec.Code, rec.Body.String()) + } + assertAPIErrorCodeFromBody(t, rec.Body.String(), model.APIErrorInvalidNodeToken) + + valid := guardRealityFixture("node-a", now) + rec = doAgentRaw(t, handler, http.MethodPost, "/api/agent/guard-reality", string(mustJSON(t, map[string]any{"node_id": "node-a", "reality": valid}))+" {}", tokenA) + if rec.Code != http.StatusBadRequest { + t.Fatalf("trailing JSON status = %d, body=%s", rec.Code, rec.Body.String()) + } + assertAPIErrorCodeFromBody(t, rec.Body.String(), model.APIErrorBadRequest) + + resp := postGuardRealityForTest(t, handler, tokenA, "node-a", valid) + if resp.code != http.StatusOK { + t.Fatalf("valid seed status = %d, body=%s", resp.code, resp.body) + } + older := guardRealityFixture("node-a", now.Add(-time.Second)) + resp = postGuardRealityForTest(t, handler, tokenA, "node-a", older) + if resp.code != http.StatusConflict { + t.Fatalf("older status = %d, body=%s", resp.code, resp.body) + } + assertAPIErrorCodeFromBody(t, resp.body, "guard_reality_stale") + + diffSameTime := guardRealityFixture("node-a", now) + diffSameTime.ManagedSHA = strings.Repeat("b", 64) + resp = postGuardRealityForTest(t, handler, tokenA, "node-a", diffSameTime) + if resp.code != http.StatusConflict { + t.Fatalf("same-time diff status = %d, body=%s", resp.code, resp.body) + } + assertAPIErrorCodeFromBody(t, resp.body, "guard_reality_stale") + + omittedEmpty := model.GuardNodeReality{ + NodeID: "node-c", + Interfaces: []model.GuardInterface{{Name: "lo"}}, + CollectedAt: now, + } + resp = postGuardRealityForTest(t, handler, tokenC, "node-c", omittedEmpty) + if resp.code != http.StatusOK { + t.Fatalf("omitted-empty seed status = %d, body=%s", resp.code, resp.body) + } + explicitEmpty := omittedEmpty + explicitEmpty.Listeners = []model.GuardListener{} + explicitEmpty.Interfaces[0].Addresses = []string{} + explicitEmpty.ForeignTables = []string{} + resp = postGuardRealityForTest(t, handler, tokenC, "node-c", explicitEmpty) + if resp.code != http.StatusOK { + t.Fatalf("explicit-empty retry status = %d, body=%s", resp.code, resp.body) + } + + future := guardRealityFixture("node-b", now.Add(10*time.Minute)) + resp = postGuardRealityForTest(t, handler, tokenB, "node-b", future) + if resp.code != http.StatusOK { + t.Fatalf("future-clamp status = %d, body=%s", resp.code, resp.body) + } + var accepted struct { + CollectedAt time.Time `json:"collected_at"` + CollectedAtClamped bool `json:"collected_at_clamped"` + } + if err := json.Unmarshal([]byte(resp.body), &accepted); err != nil { + t.Fatalf("decode future response: %v", err) + } + if !accepted.CollectedAt.Equal(now) || !accepted.CollectedAtClamped { + t.Fatalf("future clamp response = %+v, want collected_at=%s clamped=true", accepted, now) + } + + badProtocol := guardRealityFixture("node-b", now.Add(time.Minute)) + badProtocol.Listeners[0].Protocol = "icmp" + resp = postGuardRealityForTest(t, handler, tokenB, "node-b", badProtocol) + if resp.code != http.StatusBadRequest { + t.Fatalf("bad protocol status = %d, body=%s", resp.code, resp.body) + } + assertAPIErrorCodeFromBody(t, resp.body, model.APIErrorBadRequest) + + tooManyListeners := guardRealityFixture("node-b", now.Add(time.Minute)) + tooManyListeners.Listeners = make([]model.GuardListener, 4097) + for i := range tooManyListeners.Listeners { + tooManyListeners.Listeners[i] = model.GuardListener{Protocol: "tcp", Port: 1024 + i%1000} + } + resp = postGuardRealityForTest(t, handler, tokenB, "node-b", tooManyListeners) + if resp.code != http.StatusBadRequest { + t.Fatalf("too many listeners status = %d, body=%s", resp.code, resp.body) + } + assertAPIErrorCodeFromBody(t, resp.body, model.APIErrorBadRequest) +} + +func TestNetGuardRealityReadVisibilityAndPagination(t *testing.T) { + now := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + _, handler, _, cookies, csrf := newGuardRealityServerForTest(t, &now) + tokenA := enrollNamedNodeToken(t, handler, cookies, csrf, "node-a", "Node A") + enrollNamedNodeToken(t, handler, cookies, csrf, "node-b", "Node B") + tokenC := enrollNamedNodeToken(t, handler, cookies, csrf, "node-c", "Node C") + + if resp := postGuardRealityForTest(t, handler, tokenA, "node-a", guardRealityFixture("node-a", now)); resp.code != http.StatusOK { + t.Fatalf("seed node-a status=%d body=%s", resp.code, resp.body) + } + if resp := postGuardRealityForTest(t, handler, tokenC, "node-c", guardRealityFixture("node-c", now)); resp.code != http.StatusOK { + t.Fatalf("seed node-c status=%d body=%s", resp.code, resp.body) + } + + pat := createPAT(t, handler, cookies, csrf, []string{"netguard:read"}, []string{"node-b", "node-c"}) + first := doBearerJSON(t, handler, http.MethodGet, "/api/netguard/reality?limit=1", "", pat) + defer first.Body.Close() + if first.StatusCode != http.StatusOK { + t.Fatalf("first page status = %d", first.StatusCode) + } + var firstPage guardRealityListTest + if err := json.NewDecoder(first.Body).Decode(&firstPage); err != nil { + t.Fatalf("decode first page: %v", err) + } + if len(firstPage.Nodes) != 1 || firstPage.Nodes[0].NodeID != "node-b" || firstPage.Nodes[0].SnapshotStatus != "unknown" { + t.Fatalf("first page = %+v", firstPage) + } + if firstPage.NextCursor == "" { + t.Fatalf("first page missing next_cursor") + } + + second := doBearerJSON(t, handler, http.MethodGet, "/api/netguard/reality?cursor="+firstPage.NextCursor+"&limit=1", "", pat) + defer second.Body.Close() + if second.StatusCode != http.StatusOK { + t.Fatalf("second page status = %d", second.StatusCode) + } + var secondPage guardRealityListTest + if err := json.NewDecoder(second.Body).Decode(&secondPage); err != nil { + t.Fatalf("decode second page: %v", err) + } + if len(secondPage.Nodes) != 1 || secondPage.Nodes[0].NodeID != "node-c" || secondPage.Nodes[0].SnapshotStatus != "fresh" { + t.Fatalf("second page = %+v", secondPage) + } + if secondPage.NextCursor != "" { + t.Fatalf("final page next_cursor = %q, want empty", secondPage.NextCursor) + } + + hidden := doBearerJSON(t, handler, http.MethodGet, "/api/netguard/reality?node_id=node-a", "", pat) + defer hidden.Body.Close() + if hidden.StatusCode != http.StatusNotFound { + t.Fatalf("hidden detail status = %d", hidden.StatusCode) + } + body, err := io.ReadAll(hidden.Body) + if err != nil { + t.Fatalf("read hidden detail: %v", err) + } + assertAPIErrorCodeFromBody(t, string(body), model.APIErrorNotFound) + + unknown := doBearerJSON(t, handler, http.MethodGet, "/api/netguard/reality?node_id=node-b", "", pat) + defer unknown.Body.Close() + if unknown.StatusCode != http.StatusOK { + t.Fatalf("unknown detail status = %d", unknown.StatusCode) + } + var unknownDetail guardRealityDetailTest + if err := json.NewDecoder(unknown.Body).Decode(&unknownDetail); err != nil { + t.Fatalf("decode unknown detail: %v", err) + } + if unknownDetail.Node.NodeID != "node-b" || unknownDetail.Node.SnapshotStatus != "unknown" || unknownDetail.Node.Reality != nil || unknownDetail.Node.ReceivedAt != nil { + t.Fatalf("unknown detail = %+v", unknownDetail.Node) + } + + for _, path := range []string{ + "/api/netguard/reality?node_id=node-b&limit=1", + "/api/netguard/reality?cursor=not-base64", + "/api/netguard/reality?limit=501", + } { + res := doBearerJSON(t, handler, http.MethodGet, path, "", pat) + body, err := io.ReadAll(res.Body) + res.Body.Close() + if err != nil { + t.Fatalf("read %s body: %v", path, err) + } + if res.StatusCode != http.StatusBadRequest { + t.Fatalf("%s status = %d, body=%s", path, res.StatusCode, string(body)) + } + assertAPIErrorCodeFromBody(t, string(body), model.APIErrorBadRequest) + } +} diff --git a/internal/server/server_node_delete.go b/internal/server/server_node_delete.go index 8ef54cc..31a303a 100644 --- a/internal/server/server_node_delete.go +++ b/internal/server/server_node_delete.go @@ -40,6 +40,7 @@ type nodeDeleteSummary struct { Groups int `json:"groups"` Approvals int `json:"approvals"` Tunnels int `json:"tunnels"` + GuardRealitySnapshots int `json:"guard_reality_snapshots"` TerminalSessions int `json:"terminal_sessions"` // closed (delete) / active (plan) ProxyDriftCleared int `json:"proxy_drift_cleared"` // 0/1 LogStorePurged int `json:"log_store_purged"` // delete only @@ -57,7 +58,7 @@ func newNodeDeleteSummary(nodeID, name string, mutated bool, r store.NodeCascade AgentUpdatePolicies: r.AgentUpdatePolicies, ProxyNodeProfiles: r.ProxyNodeProfiles, ProxyUsageSnapshots: r.ProxyUsageSnapshots, MonitorsStripped: r.MonitorsStripped, MonitorResults: r.MonitorResults, LogSources: r.LogSources, Groups: r.Groups, - Approvals: r.Approvals, Tunnels: r.Tunnels, + Approvals: r.Approvals, Tunnels: r.Tunnels, GuardRealitySnapshots: r.GuardRealitySnapshots, } } @@ -191,31 +192,32 @@ func (s *Server) handleDeleteNode(w http.ResponseWriter, r *http.Request, p prin Action: "node.delete", Scope: "node:admin", Metadata: map[string]string{ - "node_name": name, - "tasks_stripped": strconv.Itoa(summary.TasksStripped), - "tasks_deleted": strconv.Itoa(summary.TasksDeleted), - "task_results": strconv.Itoa(summary.TaskResults), - "ddns": strconv.Itoa(summary.DDNSProfiles), - "machine_profiles": strconv.Itoa(summary.MachineProfiles), - "nft": strconv.Itoa(summary.NFTInputs), - "dns_deployments": strconv.Itoa(summary.DNSDeployments), - "net_policies": strconv.Itoa(summary.NetPolicies), - "net_peer_rules": strconv.Itoa(summary.NetPeerRulesStripped), - "group_policy_rules": strconv.Itoa(summary.GroupPolicyRulesStripped), - "geo_stripped": strconv.Itoa(summary.GeoRoutingStripped), - "geo_deleted": strconv.Itoa(summary.GeoRoutingDeleted), - "agent_updates": strconv.Itoa(summary.AgentUpdatePolicies), - "proxy_profiles": strconv.Itoa(summary.ProxyNodeProfiles), - "proxy_usage": strconv.Itoa(summary.ProxyUsageSnapshots), - "monitors_stripped": strconv.Itoa(summary.MonitorsStripped), - "monitor_results": strconv.Itoa(summary.MonitorResults), - "log_sources": strconv.Itoa(summary.LogSources), - "groups": strconv.Itoa(summary.Groups), - "approvals": strconv.Itoa(summary.Approvals), - "tunnels": strconv.Itoa(summary.Tunnels), - "terminal_sessions": strconv.Itoa(summary.TerminalSessions), - "proxy_drift_cleared": strconv.Itoa(summary.ProxyDriftCleared), - "log_purge_errors": strconv.Itoa(summary.LogStorePurgeErrs), + "node_name": name, + "tasks_stripped": strconv.Itoa(summary.TasksStripped), + "tasks_deleted": strconv.Itoa(summary.TasksDeleted), + "task_results": strconv.Itoa(summary.TaskResults), + "ddns": strconv.Itoa(summary.DDNSProfiles), + "machine_profiles": strconv.Itoa(summary.MachineProfiles), + "nft": strconv.Itoa(summary.NFTInputs), + "dns_deployments": strconv.Itoa(summary.DNSDeployments), + "net_policies": strconv.Itoa(summary.NetPolicies), + "net_peer_rules": strconv.Itoa(summary.NetPeerRulesStripped), + "group_policy_rules": strconv.Itoa(summary.GroupPolicyRulesStripped), + "geo_stripped": strconv.Itoa(summary.GeoRoutingStripped), + "geo_deleted": strconv.Itoa(summary.GeoRoutingDeleted), + "agent_updates": strconv.Itoa(summary.AgentUpdatePolicies), + "proxy_profiles": strconv.Itoa(summary.ProxyNodeProfiles), + "proxy_usage": strconv.Itoa(summary.ProxyUsageSnapshots), + "monitors_stripped": strconv.Itoa(summary.MonitorsStripped), + "monitor_results": strconv.Itoa(summary.MonitorResults), + "log_sources": strconv.Itoa(summary.LogSources), + "groups": strconv.Itoa(summary.Groups), + "approvals": strconv.Itoa(summary.Approvals), + "tunnels": strconv.Itoa(summary.Tunnels), + "guard_reality_snapshots": strconv.Itoa(summary.GuardRealitySnapshots), + "terminal_sessions": strconv.Itoa(summary.TerminalSessions), + "proxy_drift_cleared": strconv.Itoa(summary.ProxyDriftCleared), + "log_purge_errors": strconv.Itoa(summary.LogStorePurgeErrs), }, }) diff --git a/internal/server/server_node_delete_test.go b/internal/server/server_node_delete_test.go index a2ea1f8..52ad5a5 100644 --- a/internal/server/server_node_delete_test.go +++ b/internal/server/server_node_delete_test.go @@ -4,8 +4,10 @@ import ( "encoding/json" "net/http" "testing" + "time" "github.com/LatticeNet/lattice-sdk/model" + "github.com/LatticeNet/lattice-server/internal/store" ) func decodeNodeDeleteSummary(t *testing.T, res *http.Response) nodeDeleteSummary { @@ -28,6 +30,7 @@ func TestNodeDeletePlanIsNonMutating(t *testing.T) { if err := st.UpsertDDNSProfile(model.DDNSProfile{ID: "ddns-1", NodeID: nodeID, Provider: model.DDNSProviderCloudflare}); err != nil { t.Fatalf("seed ddns: %v", err) } + seedNodeGuardReality(t, st, nodeID) res := doJSON(t, handler, http.MethodPost, "/api/nodes/delete/plan", `{"node_id":"`+nodeID+`"}`, cookies, csrf) if res.StatusCode != http.StatusOK { @@ -43,10 +46,16 @@ func TestNodeDeletePlanIsNonMutating(t *testing.T) { if summary.DDNSProfiles != 1 { t.Fatalf("plan ddns_profiles = %d want 1", summary.DDNSProfiles) } + if summary.GuardRealitySnapshots != 1 { + t.Fatalf("plan guard_reality_snapshots = %d want 1", summary.GuardRealitySnapshots) + } // The node must still exist after a plan. if _, ok := st.Node(nodeID); !ok { t.Fatal("plan deleted the node") } + if _, ok := st.GuardRealitySnapshot(nodeID); !ok { + t.Fatal("plan deleted guard reality snapshot") + } } // TestNodeDeleteRemovesNodeAndAudits verifies a delete returns the summary, @@ -60,6 +69,7 @@ func TestNodeDeleteRemovesNodeAndAudits(t *testing.T) { if err := st.UpsertDDNSProfile(model.DDNSProfile{ID: "ddns-1", NodeID: nodeID, Provider: model.DDNSProviderCloudflare}); err != nil { t.Fatalf("seed ddns: %v", err) } + seedNodeGuardReality(t, st, nodeID) // A plan first: it must not write an audit row. planRes := doJSON(t, handler, http.MethodPost, "/api/nodes/delete/plan", `{"node_id":"`+nodeID+`"}`, cookies, csrf) @@ -80,6 +90,9 @@ func TestNodeDeleteRemovesNodeAndAudits(t *testing.T) { if summary.DDNSProfiles != 1 { t.Fatalf("delete ddns_profiles = %d want 1", summary.DDNSProfiles) } + if summary.GuardRealitySnapshots != 1 { + t.Fatalf("delete guard_reality_snapshots = %d want 1", summary.GuardRealitySnapshots) + } if _, ok := st.Node(nodeID); ok { t.Fatal("node survived delete") @@ -99,6 +112,9 @@ func TestNodeDeleteRemovesNodeAndAudits(t *testing.T) { if ev.Metadata["ddns"] != "1" { t.Fatalf("audit metadata ddns = %q want 1", ev.Metadata["ddns"]) } + if ev.Metadata["guard_reality_snapshots"] != "1" { + t.Fatalf("audit metadata guard_reality_snapshots = %q want 1", ev.Metadata["guard_reality_snapshots"]) + } // Idempotent: a second delete returns 404. res2 := doJSON(t, handler, http.MethodPost, "/api/nodes/delete", `{"node_id":"`+nodeID+`"}`, cookies, csrf) @@ -179,3 +195,18 @@ func lastDeleteAudit(st interface { } return last } + +func seedNodeGuardReality(t *testing.T, st *store.Store, nodeID string) { + t.Helper() + node, ok := st.Node(nodeID) + if !ok { + t.Fatal("seed node missing") + } + collectedAt := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + if _, _, err := st.UpsertGuardRealitySnapshot(node.LatticeIdentityUUID, store.GuardRealitySnapshot{ + Reality: model.GuardNodeReality{NodeID: nodeID, CollectedAt: collectedAt}, + ReceivedAt: collectedAt.Add(time.Second), + }); err != nil { + t.Fatalf("seed guard reality: %v", err) + } +} diff --git a/internal/store/bolt_migration_test.go b/internal/store/bolt_migration_test.go index db7cbd9..b000c56 100644 --- a/internal/store/bolt_migration_test.go +++ b/internal/store/bolt_migration_test.go @@ -3,6 +3,7 @@ package store import ( "os" "path/filepath" + "reflect" "strings" "testing" "time" @@ -13,7 +14,23 @@ import ( func seedMigrationState(now time.Time) State { st := emptyState() st.Users["u1"] = model.User{ID: "u1", Username: "admin", TOTPSecret: totpPlain, CreatedAt: now} - st.Nodes["node-a"] = model.Node{ID: "node-a", Name: "Node A", TokenHash: "node-token-hash", CreatedAt: now} + st.Nodes["node-a"] = model.Node{ID: "node-a", LatticeIdentityUUID: "generation-a", Name: "Node A", TokenHash: "node-token-hash", CreatedAt: now} + st.GuardRealitySnapshots["node-a"] = GuardRealitySnapshot{ + Reality: model.GuardNodeReality{ + NodeID: "node-a", + Listeners: []model.GuardListener{{ + Protocol: "tcp", Port: 443, Address: "2001:db8::10", Process: "edge-proxy", + }}, + Interfaces: []model.GuardInterface{{ + Name: "ens3", Addresses: []string{"2001:db8::10/128", "2001:db8::11/128"}, Up: true, + }}, + ManagedSHA: strings.Repeat("a", 64), + ForeignTables: []string{"inet docker", "inet podman"}, + NFTVersion: "nftables v1.0.9", + CollectedAt: now.Add(123456789 * time.Nanosecond), + }, + ReceivedAt: now.Add(987654321 * time.Nanosecond), + } st.DDNS["d1"] = model.DDNSProfile{ID: "d1", Name: "dns", Provider: "cloudflare", CFAPIToken: cfTokenPlain} st.NotifyChannels["ch1"] = model.NotifyChannel{ID: "ch1", Name: "tg", Kind: "telegram", Config: map[string]string{"bot_token": botTokenPlain}} st.OIDCProviders["oidc"] = model.OIDCProvider{ID: "oidc", DisplayName: "OIDC", ClientID: "client-id", ClientSecret: "oidc-secret", CreatedAt: now} @@ -31,7 +48,8 @@ func TestMigrateJSONToBoltAndExportBack(t *testing.T) { c := testCipher(t) now := time.Unix(1_700_000_001, 0).UTC() - if err := WriteJSONState(jsonPath, seedMigrationState(now), c, MigrationOptions{}); err != nil { + want := seedMigrationState(now) + if err := WriteJSONState(jsonPath, want, c, MigrationOptions{}); err != nil { t.Fatal(err) } rawJSON, err := os.ReadFile(jsonPath) @@ -70,6 +88,9 @@ func TestMigrateJSONToBoltAndExportBack(t *testing.T) { if got.Groups["grp1"].Members[0] != "node-a" || got.GroupPolicies["gnp1"].Rules[0].Ports[0] != 443 { t.Fatalf("migrated group state did not recover: %+v %+v", got.Groups["grp1"], got.GroupPolicies["gnp1"]) } + if !reflect.DeepEqual(got.GuardRealitySnapshots["node-a"], want.GuardRealitySnapshots["node-a"]) { + t.Fatalf("migrated guard reality did not recover:\n got=%+v\nwant=%+v", got.GuardRealitySnapshots["node-a"], want.GuardRealitySnapshots["node-a"]) + } if err := ExportBoltToJSON(boltPath, exportPath, c, MigrationOptions{}); err != nil { t.Fatal(err) @@ -88,6 +109,9 @@ func TestMigrateJSONToBoltAndExportBack(t *testing.T) { if back.Nodes["node-a"].Name != "Node A" || len(back.Audit) != 1 || back.NotifyChannels["ch1"].Config["bot_token"] != botTokenPlain || back.Groups["grp1"].Name != "Edge" || back.GroupPolicies["gnp1"].ScopeGroupID != "grp1" { t.Fatalf("exported JSON did not round-trip: %+v", back) } + if !reflect.DeepEqual(back.GuardRealitySnapshots["node-a"], want.GuardRealitySnapshots["node-a"]) { + t.Fatalf("exported guard reality did not round-trip:\n got=%+v\nwant=%+v", back.GuardRealitySnapshots["node-a"], want.GuardRealitySnapshots["node-a"]) + } } func TestMigrationRefusesToOverwriteTargets(t *testing.T) { diff --git a/internal/store/bolt_state.go b/internal/store/bolt_state.go index 816f2d3..c4e469b 100644 --- a/internal/store/bolt_state.go +++ b/internal/store/bolt_state.go @@ -49,6 +49,7 @@ var ( boltBucketMachineProfiles = []byte("machine_profiles") boltBucketMachineVendors = []byte("machine_vendors") boltBucketNFTInputs = []byte("nft_inputs") + boltBucketGuardReality = []byte("guard_reality_snapshots") boltBucketDNSDeployments = []byte("dns_deployments") boltBucketNetPolicies = []byte("net_policies") boltBucketGroups = []byte("groups") @@ -92,6 +93,7 @@ var boltStateBuckets = [][]byte{ boltBucketMachineProfiles, boltBucketMachineVendors, boltBucketNFTInputs, + boltBucketGuardReality, boltBucketDNSDeployments, boltBucketNetPolicies, boltBucketGroups, @@ -268,6 +270,9 @@ func (bs *BoltStateStore) ImportState(st State) error { if err := putMap(tx, boltBucketNFTInputs, persist.NFTInputs); err != nil { return err } + if err := putMap(tx, boltBucketGuardReality, persist.GuardRealitySnapshots); err != nil { + return err + } if err := putMap(tx, boltBucketDNSDeployments, persist.DNSDeployments); err != nil { return err } @@ -416,6 +421,9 @@ func (bs *BoltStateStore) ExportState() (State, error) { if err := readMap(tx, boltBucketNFTInputs, st.NFTInputs); err != nil { return err } + if err := readMap(tx, boltBucketGuardReality, st.GuardRealitySnapshots); err != nil { + return err + } if err := readMap(tx, boltBucketDNSDeployments, st.DNSDeployments); err != nil { return err } diff --git a/internal/store/cascade.go b/internal/store/cascade.go index cdb7dc4..3e22dcc 100644 --- a/internal/store/cascade.go +++ b/internal/store/cascade.go @@ -36,6 +36,7 @@ type NodeCascadeReport struct { Groups int `json:"groups"` // Members/LeaderID edited Approvals int `json:"approvals"` // NO existing primitive Tunnels int `json:"tunnels"` + GuardRealitySnapshots int `json:"guard_reality_snapshots"` // RemovedLogSourceIDs lists the log-source IDs whose records this delete // removed from the JSON store. The SERVER must call logStore.PurgeSource on // each (the log lines live in a separate bbolt db the store cannot reach). @@ -404,7 +405,15 @@ func (s *Store) buildNodeCascadeLocked(nodeID string, mutate bool) (NodeCascadeR } } - // Step 17: the node itself (embedded TokenHash/Metrics/HostFacts/Geo/etc all + // Step 17: latest low-trust NetGuard reality snapshot (node-owned). + if _, ok := s.state.GuardRealitySnapshots[nodeID]; ok { + report.GuardRealitySnapshots++ + if mutate { + delete(s.state.GuardRealitySnapshots, nodeID) + } + } + + // Step 18: the node itself (embedded TokenHash/Metrics/HostFacts/Geo/etc all // purged with the record). if mutate { delete(s.state.Nodes, nodeID) diff --git a/internal/store/cascade_test.go b/internal/store/cascade_test.go index be2fdeb..aaaee52 100644 --- a/internal/store/cascade_test.go +++ b/internal/store/cascade_test.go @@ -4,6 +4,7 @@ import ( "path/filepath" "reflect" "testing" + "time" "github.com/LatticeNet/lattice-sdk/model" ) @@ -19,12 +20,21 @@ func seedNodeCascade(t *testing.T) *Store { } const node = "node-target" const other = "node-other" - if err := s.UpsertNode(model.Node{ID: node, Name: "target"}); err != nil { + if err := s.UpsertNode(model.Node{ID: node, LatticeIdentityUUID: "generation-target", Name: "target"}); err != nil { t.Fatalf("upsert node: %v", err) } - if err := s.UpsertNode(model.Node{ID: other, Name: "other"}); err != nil { + if err := s.UpsertNode(model.Node{ID: other, LatticeIdentityUUID: "generation-other", Name: "other"}); err != nil { t.Fatalf("upsert other: %v", err) } + collectedAt := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + for nodeID, generation := range map[string]string{node: "generation-target", other: "generation-other"} { + if _, _, err := s.UpsertGuardRealitySnapshot(generation, GuardRealitySnapshot{ + Reality: model.GuardNodeReality{NodeID: nodeID, CollectedAt: collectedAt}, + ReceivedAt: collectedAt.Add(time.Second), + }); err != nil { + t.Fatalf("guard reality %s: %v", nodeID, err) + } + } // Step 1/2: a sole-target task (deleted) and a multi-target task (stripped), // each with a result for the target node. @@ -154,6 +164,7 @@ func expectedCascade() NodeCascadeReport { Groups: 1, Approvals: 1, Tunnels: 1, + GuardRealitySnapshots: 1, } } @@ -183,6 +194,9 @@ func TestPlanDeleteNodeMatchesDelete(t *testing.T) { if _, ok := s.NFTInputs("node-target"); !ok { t.Fatal("plan deleted nft inputs") } + if _, ok := s.GuardRealitySnapshot("node-target"); !ok { + t.Fatal("plan deleted guard reality snapshot") + } del, ok, err := s.DeleteNode("node-target") if err != nil { @@ -224,6 +238,12 @@ func TestDeleteNodeCascade(t *testing.T) { if _, ok := s.NetPolicy("node-target"); ok { t.Fatal("net policy survived") } + if _, ok := s.GuardRealitySnapshot("node-target"); ok { + t.Fatal("guard reality snapshot survived") + } + if _, ok := s.GuardRealitySnapshot("node-other"); !ok { + t.Fatal("bystander guard reality snapshot was deleted") + } if _, ok := s.AgentUpdatePolicy("node-target"); ok { t.Fatal("agent update policy survived") } diff --git a/internal/store/guard_reality.go b/internal/store/guard_reality.go new file mode 100644 index 0000000..91cb216 --- /dev/null +++ b/internal/store/guard_reality.go @@ -0,0 +1,172 @@ +package store + +import ( + "errors" + "fmt" + "reflect" + "sort" + "strings" + "time" + + "github.com/LatticeNet/lattice-sdk/model" +) + +// GuardRealitySnapshot stores the server-accepted, normalized latest reality +// report for one node. It deliberately contains operational facts only: raw +// request bytes, bearer credentials, stderr, key material, and secrets are +// forbidden from this collection. +type GuardRealitySnapshot struct { + Reality model.GuardNodeReality `json:"reality"` + ReceivedAt time.Time `json:"received_at"` +} + +// ErrGuardRealityStale is returned when a write would replace a newer snapshot +// or conflict with a different snapshot collected at the same instant. +var ErrGuardRealityStale = errors.New("guard reality snapshot is stale") + +// ErrGuardRealityNodeChanged is returned when the node authenticated by the +// handler no longer exists as the same immutable identity generation. +var ErrGuardRealityNodeChanged = errors.New("guard reality node identity changed") + +// ErrGuardRealityDurabilityDegraded means the atomic rename committed the +// snapshot, but syncing the parent directory failed. Callers must treat the +// snapshot as accepted while surfacing the durability warning operationally. +var ErrGuardRealityDurabilityDegraded = errors.New("guard reality committed with degraded durability") + +// UpsertGuardRealitySnapshot stores the latest normalized reality snapshot for +// a node. Same collected_at plus identical content is idempotent and does not +// rewrite received_at; same collected_at plus different content is a conflict. +func (s *Store) UpsertGuardRealitySnapshot(nodeIdentityUUID string, snapshot GuardRealitySnapshot) (GuardRealitySnapshot, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.ensureMaps() + snapshot = canonicalizeGuardRealitySnapshot(snapshot) + snapshot.Reality.CollectedAt = snapshot.Reality.CollectedAt.UTC() + snapshot.ReceivedAt = snapshot.ReceivedAt.UTC() + if snapshot.Reality.NodeID == "" { + return GuardRealitySnapshot{}, false, errors.New("node_id is required") + } + node, ok := s.state.Nodes[snapshot.Reality.NodeID] + if !ok || strings.TrimSpace(node.LatticeIdentityUUID) != strings.TrimSpace(nodeIdentityUUID) { + return GuardRealitySnapshot{}, false, ErrGuardRealityNodeChanged + } + if existing, ok := s.state.GuardRealitySnapshots[snapshot.Reality.NodeID]; ok { + existing = canonicalizeGuardRealitySnapshot(existing) + existing.Reality.CollectedAt = existing.Reality.CollectedAt.UTC() + existing.ReceivedAt = existing.ReceivedAt.UTC() + switch { + case snapshot.Reality.CollectedAt.Before(existing.Reality.CollectedAt): + return existing, false, ErrGuardRealityStale + case snapshot.Reality.CollectedAt.Equal(existing.Reality.CollectedAt): + if reflect.DeepEqual(snapshot.Reality, existing.Reality) { + return existing, false, nil + } + return existing, false, ErrGuardRealityStale + } + } + next := make(map[string]GuardRealitySnapshot, len(s.state.GuardRealitySnapshots)+1) + for nodeID, existing := range s.state.GuardRealitySnapshots { + next[nodeID] = existing + } + next[snapshot.Reality.NodeID] = snapshot + staged := s.state + staged.GuardRealitySnapshots = next + committed, err := s.persistState(s.jsonPersistStateFrom(staged)) + if !committed { + return GuardRealitySnapshot{}, false, err + } + s.state.GuardRealitySnapshots = next + if err != nil { + return cloneGuardRealitySnapshot(snapshot), true, fmt.Errorf("%w: %v", ErrGuardRealityDurabilityDegraded, err) + } + return cloneGuardRealitySnapshot(snapshot), true, nil +} + +// GuardRealitySnapshot returns a deep copy of one node's latest reality report. +func (s *Store) GuardRealitySnapshot(nodeID string) (GuardRealitySnapshot, bool) { + s.mu.Lock() + defer s.mu.Unlock() + s.ensureMaps() + snapshot, ok := s.state.GuardRealitySnapshots[nodeID] + if !ok { + return GuardRealitySnapshot{}, false + } + return cloneGuardRealitySnapshot(snapshot), true +} + +// GuardRealitySnapshots returns all snapshots sorted by node id. +func (s *Store) GuardRealitySnapshots() []GuardRealitySnapshot { + s.mu.Lock() + defer s.mu.Unlock() + s.ensureMaps() + out := make([]GuardRealitySnapshot, 0, len(s.state.GuardRealitySnapshots)) + for _, snapshot := range s.state.GuardRealitySnapshots { + out = append(out, cloneGuardRealitySnapshot(snapshot)) + } + sort.Slice(out, func(i, j int) bool { + return out[i].Reality.NodeID < out[j].Reality.NodeID + }) + return out +} + +func cloneGuardRealitySnapshot(snapshot GuardRealitySnapshot) GuardRealitySnapshot { + snapshot.Reality.Listeners = append([]model.GuardListener(nil), snapshot.Reality.Listeners...) + if snapshot.Reality.Interfaces != nil { + interfaces := make([]model.GuardInterface, len(snapshot.Reality.Interfaces)) + for i, iface := range snapshot.Reality.Interfaces { + iface.Addresses = append([]string(nil), iface.Addresses...) + interfaces[i] = iface + } + snapshot.Reality.Interfaces = interfaces + } + snapshot.Reality.ForeignTables = append([]string(nil), snapshot.Reality.ForeignTables...) + return snapshot +} + +func canonicalizeGuardRealitySnapshot(snapshot GuardRealitySnapshot) GuardRealitySnapshot { + snapshot = cloneGuardRealitySnapshot(snapshot) + if len(snapshot.Reality.Listeners) == 0 { + snapshot.Reality.Listeners = nil + } else { + sort.Slice(snapshot.Reality.Listeners, func(i, j int) bool { + a, b := snapshot.Reality.Listeners[i], snapshot.Reality.Listeners[j] + if a.Protocol != b.Protocol { + return a.Protocol < b.Protocol + } + if a.Port != b.Port { + return a.Port < b.Port + } + if a.Address != b.Address { + return a.Address < b.Address + } + return a.Process < b.Process + }) + } + if len(snapshot.Reality.Interfaces) == 0 { + snapshot.Reality.Interfaces = nil + } else { + for i := range snapshot.Reality.Interfaces { + if len(snapshot.Reality.Interfaces[i].Addresses) == 0 { + snapshot.Reality.Interfaces[i].Addresses = nil + } else { + sort.Strings(snapshot.Reality.Interfaces[i].Addresses) + } + } + sort.Slice(snapshot.Reality.Interfaces, func(i, j int) bool { + a, b := snapshot.Reality.Interfaces[i], snapshot.Reality.Interfaces[j] + if a.Name != b.Name { + return a.Name < b.Name + } + if a.Up != b.Up { + return !a.Up + } + return strings.Join(a.Addresses, "\x00") < strings.Join(b.Addresses, "\x00") + }) + } + if len(snapshot.Reality.ForeignTables) == 0 { + snapshot.Reality.ForeignTables = nil + } else { + sort.Strings(snapshot.Reality.ForeignTables) + } + return snapshot +} diff --git a/internal/store/guard_reality_test.go b/internal/store/guard_reality_test.go new file mode 100644 index 0000000..8088019 --- /dev/null +++ b/internal/store/guard_reality_test.go @@ -0,0 +1,577 @@ +package store + +import ( + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/LatticeNet/lattice-sdk/model" +) + +func TestGuardRealitySnapshotLatestOnlyAndCopies(t *testing.T) { + st, err := Open("") + if err != nil { + t.Fatalf("open store: %v", err) + } + if err := st.UpsertNode(model.Node{ID: "node-a", LatticeIdentityUUID: "generation-a"}); err != nil { + t.Fatalf("upsert node-a: %v", err) + } + if err := st.UpsertNode(model.Node{ID: "node-b", LatticeIdentityUUID: "generation-b"}); err != nil { + t.Fatalf("upsert node-b: %v", err) + } + + collectedAt := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + first := GuardRealitySnapshot{ + Reality: model.GuardNodeReality{ + NodeID: "node-b", + ManagedSHA: strings.Repeat("a", 64), + Listeners: []model.GuardListener{{ + Protocol: "tcp", + Port: 22, + Address: "2001:db8::10", + Process: "sshd", + }}, + Interfaces: []model.GuardInterface{{ + Name: "ens3", + Addresses: []string{"2001:db8::10/128"}, + Up: true, + }}, + ForeignTables: []string{"inet docker"}, + NFTVersion: "nftables v1.0.9", + CollectedAt: collectedAt, + }, + ReceivedAt: collectedAt.Add(time.Second), + } + + stored, changed, err := st.UpsertGuardRealitySnapshot("generation-b", first) + if err != nil { + t.Fatalf("upsert first snapshot: %v", err) + } + if !changed { + t.Fatalf("first snapshot was not marked changed") + } + if stored.Reality.NodeID != "node-b" { + t.Fatalf("stored node id = %q", stored.Reality.NodeID) + } + + first.Reality.Listeners[0].Process = "mutated-after-upsert" + got, ok := st.GuardRealitySnapshot("node-b") + if !ok { + t.Fatalf("snapshot missing after upsert") + } + if got.Reality.Listeners[0].Process != "sshd" { + t.Fatalf("snapshot aliases caller memory: process = %q", got.Reality.Listeners[0].Process) + } + got.Reality.Listeners[0].Process = "mutated-after-read" + gotAgain, ok := st.GuardRealitySnapshot("node-b") + if !ok { + t.Fatalf("snapshot missing after read") + } + if gotAgain.Reality.Listeners[0].Process != "sshd" { + t.Fatalf("read snapshot aliases store memory: process = %q", gotAgain.Reality.Listeners[0].Process) + } + + same := first + same.Reality.Listeners[0].Process = "sshd" + same.ReceivedAt = collectedAt.Add(2 * time.Second) + stored, changed, err = st.UpsertGuardRealitySnapshot("generation-b", same) + if err != nil { + t.Fatalf("idempotent upsert returned error: %v", err) + } + if changed { + t.Fatalf("idempotent upsert was marked changed") + } + if !stored.ReceivedAt.Equal(collectedAt.Add(time.Second)) { + t.Fatalf("idempotent upsert changed received_at to %s", stored.ReceivedAt) + } + + older := same + older.Reality.CollectedAt = collectedAt.Add(-time.Second) + if _, _, err := st.UpsertGuardRealitySnapshot("generation-b", older); !errors.Is(err, ErrGuardRealityStale) { + t.Fatalf("older snapshot error = %v, want ErrGuardRealityStale", err) + } + + diffSameTime := same + diffSameTime.Reality.ManagedSHA = strings.Repeat("b", 64) + if _, _, err := st.UpsertGuardRealitySnapshot("generation-b", diffSameTime); !errors.Is(err, ErrGuardRealityStale) { + t.Fatalf("same-time conflicting snapshot error = %v, want ErrGuardRealityStale", err) + } + + newer := same + newer.Reality.CollectedAt = collectedAt.Add(time.Minute) + newer.Reality.ManagedSHA = strings.Repeat("c", 64) + if _, changed, err := st.UpsertGuardRealitySnapshot("generation-b", newer); err != nil || !changed { + t.Fatalf("newer snapshot changed=%v err=%v, want changed nil-error", changed, err) + } + + nodeA := newer + nodeA.Reality.NodeID = "node-a" + if _, _, err := st.UpsertGuardRealitySnapshot("generation-a", nodeA); err != nil { + t.Fatalf("upsert node-a snapshot: %v", err) + } + all := st.GuardRealitySnapshots() + if len(all) != 2 { + t.Fatalf("snapshot count = %d, want 2", len(all)) + } + if all[0].Reality.NodeID != "node-a" || all[1].Reality.NodeID != "node-b" { + t.Fatalf("snapshots not sorted by node_id: %q, %q", all[0].Reality.NodeID, all[1].Reality.NodeID) + } +} + +func TestGuardRealitySnapshotPersistsPlaintextOperationalFacts(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + cipher := testCipher(t) + st, err := OpenWithCipher(path, cipher) + if err != nil { + t.Fatalf("open encrypted store: %v", err) + } + if err := st.UpsertNode(model.Node{ID: "node-a", LatticeIdentityUUID: "generation-a"}); err != nil { + t.Fatalf("upsert node-a: %v", err) + } + + collectedAt := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + if _, _, err := st.UpsertGuardRealitySnapshot("generation-a", GuardRealitySnapshot{ + Reality: model.GuardNodeReality{ + NodeID: "node-a", + ManagedSHA: strings.Repeat("a", 64), + Listeners: []model.GuardListener{{ + Protocol: "tcp", + Port: 443, + Address: "2001:db8::20", + Process: "edge-proxy", + }}, + CollectedAt: collectedAt, + }, + ReceivedAt: collectedAt.Add(time.Second), + }); err != nil { + t.Fatalf("upsert snapshot: %v", err) + } + + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read persisted state: %v", err) + } + if !strings.Contains(string(raw), "guard_reality_snapshots") { + t.Fatalf("persisted state missing guard_reality_snapshots collection") + } + if !strings.Contains(string(raw), "edge-proxy") { + t.Fatalf("operational snapshot facts were unexpectedly encrypted or omitted") + } + if strings.Contains(string(raw), "Bearer ") { + t.Fatalf("persisted state contains bearer credential material") + } + + reopened, err := OpenWithCipher(path, cipher) + if err != nil { + t.Fatalf("reopen encrypted store: %v", err) + } + got, ok := reopened.GuardRealitySnapshot("node-a") + if !ok { + t.Fatalf("reopened store missing guard reality snapshot") + } + if got.Reality.Listeners[0].Process != "edge-proxy" { + t.Fatalf("reopened process = %q, want edge-proxy", got.Reality.Listeners[0].Process) + } +} + +func TestGuardRealitySnapshotPersistFailureDoesNotPublish(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + cipher := testCipher(t) + st, err := OpenWithCipher(path, cipher) + if err != nil { + t.Fatalf("open store: %v", err) + } + for _, node := range []model.Node{ + {ID: "node-a", LatticeIdentityUUID: "generation-a"}, + {ID: "node-b", LatticeIdentityUUID: "generation-b"}, + } { + if err := st.UpsertNode(node); err != nil { + t.Fatalf("upsert %s: %v", node.ID, err) + } + } + + collectedAt := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + first := GuardRealitySnapshot{ + Reality: model.GuardNodeReality{ + NodeID: "node-a", + ManagedSHA: strings.Repeat("a", 64), + CollectedAt: collectedAt, + }, + ReceivedAt: collectedAt.Add(time.Second), + } + if _, _, err := st.UpsertGuardRealitySnapshot("generation-a", first); err != nil { + t.Fatalf("seed snapshot: %v", err) + } + + if err := os.Mkdir(path+".tmp", 0o700); err != nil { + t.Fatalf("install save-failure fixture: %v", err) + } + newer := first + newer.Reality.CollectedAt = collectedAt.Add(time.Minute) + newer.Reality.ManagedSHA = strings.Repeat("b", 64) + newer.ReceivedAt = collectedAt.Add(time.Minute + time.Second) + if _, _, err := st.UpsertGuardRealitySnapshot("generation-a", newer); err == nil { + t.Fatal("replacement unexpectedly survived forced persist failure") + } + if err := os.Mkdir(path+".tmp", 0o700); err != nil { + t.Fatalf("reinstall save-failure fixture: %v", err) + } + firstInsert := newer + firstInsert.Reality.NodeID = "node-b" + if _, _, err := st.UpsertGuardRealitySnapshot("generation-b", firstInsert); err == nil { + t.Fatal("first insert unexpectedly survived forced persist failure") + } + if err := st.ReadyCheck(); err != nil { + t.Fatalf("pre-rename failure degraded readiness: %v", err) + } + + got, ok := st.GuardRealitySnapshot("node-a") + if !ok || got.Reality.ManagedSHA != strings.Repeat("a", 64) { + t.Fatalf("live snapshot changed after failed persist: ok=%v snapshot=%+v", ok, got) + } + if _, ok := st.GuardRealitySnapshot("node-b"); ok { + t.Fatal("failed first insert was published to live state") + } + reopened, err := OpenWithCipher(path, cipher) + if err != nil { + t.Fatalf("reopen after failed persists: %v", err) + } + got, ok = reopened.GuardRealitySnapshot("node-a") + if !ok || got.Reality.ManagedSHA != strings.Repeat("a", 64) { + t.Fatalf("persisted snapshot changed after failed persist: ok=%v snapshot=%+v", ok, got) + } + if _, ok := reopened.GuardRealitySnapshot("node-b"); ok { + t.Fatal("failed first insert reached persisted state") + } + + if _, changed, err := st.UpsertGuardRealitySnapshot("generation-a", newer); err != nil || !changed { + t.Fatalf("retry replacement changed=%v err=%v", changed, err) + } + if _, changed, err := st.UpsertGuardRealitySnapshot("generation-b", firstInsert); err != nil || !changed { + t.Fatalf("retry first insert changed=%v err=%v", changed, err) + } + reopened, err = OpenWithCipher(path, cipher) + if err != nil { + t.Fatalf("reopen after retries: %v", err) + } + for nodeID, wantSHA := range map[string]string{ + "node-a": strings.Repeat("b", 64), + "node-b": strings.Repeat("b", 64), + } { + got, ok := reopened.GuardRealitySnapshot(nodeID) + if !ok || got.Reality.ManagedSHA != wantSHA { + t.Fatalf("retried snapshot %s: ok=%v snapshot=%+v", nodeID, ok, got) + } + } +} + +func TestOpenWithCipherSyncsOnlyExistingStateParentDirectory(t *testing.T) { + tests := []struct { + name string + create bool + contents []byte + wantCalls int + }{ + {name: "absent", wantCalls: 0}, + {name: "empty", create: true, wantCalls: 1}, + {name: "populated", create: true, contents: []byte("{}"), wantCalls: 1}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + if tt.create { + if err := os.WriteFile(path, tt.contents, 0o600); err != nil { + t.Fatalf("write state fixture: %v", err) + } + } + calls := 0 + syncedDir := "" + st, err := openWithCipher(path, testCipher(t), func(dir string) error { + calls++ + syncedDir = dir + return nil + }) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + if calls != tt.wantCalls { + t.Fatalf("startup parent sync calls = %d, want %d", calls, tt.wantCalls) + } + if tt.wantCalls == 0 && syncedDir != "" { + t.Fatalf("absent state synced directory %q", syncedDir) + } + if tt.wantCalls == 1 && syncedDir != filepath.Dir(path) { + t.Fatalf("startup parent sync directory = %q, want %q", syncedDir, filepath.Dir(path)) + } + if err := st.ReadyCheck(); err != nil { + t.Fatalf("successful startup path degraded readiness: %v", err) + } + }) + } +} + +func TestGuardRealitySnapshotPostRenameFailurePublishesCommittedState(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + cipher := testCipher(t) + st, err := OpenWithCipher(path, cipher) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + if err := st.UpsertNode(model.Node{ID: "node-a", LatticeIdentityUUID: "generation-a"}); err != nil { + t.Fatalf("upsert node: %v", err) + } + collectedAt := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + first := GuardRealitySnapshot{ + Reality: model.GuardNodeReality{ + NodeID: "node-a", + ManagedSHA: strings.Repeat("a", 64), + CollectedAt: collectedAt, + }, + ReceivedAt: collectedAt.Add(time.Second), + } + if _, _, err := st.UpsertGuardRealitySnapshot("generation-a", first); err != nil { + t.Fatalf("seed snapshot: %v", err) + } + if err := st.ReadyCheck(); err != nil { + t.Fatalf("healthy persistence degraded readiness: %v", err) + } + + newer := first + newer.Reality.CollectedAt = collectedAt.Add(time.Minute) + newer.Reality.ManagedSHA = strings.Repeat("b", 64) + newer.ReceivedAt = collectedAt.Add(time.Minute + time.Second) + st.syncParentDir = func(string) error { return errors.New("forced post-rename sync failure") } + stored, changed, err := st.UpsertGuardRealitySnapshot("generation-a", newer) + if !errors.Is(err, ErrGuardRealityDurabilityDegraded) || !changed { + t.Fatalf("post-rename result changed=%v err=%v", changed, err) + } + if err := st.ReadyCheck(); err == nil { + t.Fatal("post-rename sync failure left readiness healthy") + } + if stored.Reality.ManagedSHA != strings.Repeat("b", 64) { + t.Fatalf("returned committed snapshot = %+v", stored) + } + got, ok := st.GuardRealitySnapshot("node-a") + if !ok || got.Reality.ManagedSHA != strings.Repeat("b", 64) { + t.Fatalf("live state did not publish committed snapshot: ok=%v snapshot=%+v", ok, got) + } + startupSyncCalls := 0 + startupSyncDir := "" + reopened, err := openWithCipher(path, cipher, func(dir string) error { + startupSyncCalls++ + startupSyncDir = dir + return errors.New("forced startup parent sync failure") + }) + if err != nil { + t.Fatalf("reopen committed snapshot: %v", err) + } + t.Cleanup(func() { _ = reopened.Close() }) + got, ok = reopened.GuardRealitySnapshot("node-a") + if !ok || got.Reality.ManagedSHA != strings.Repeat("b", 64) { + t.Fatalf("reopened state did not contain committed snapshot: ok=%v snapshot=%+v", ok, got) + } + if err := reopened.ReadyCheck(); err == nil { + t.Fatal("restart cleared durability degradation without confirming parent directory sync") + } + if startupSyncCalls != 1 { + t.Fatalf("startup parent sync calls = %d, want 1", startupSyncCalls) + } + if startupSyncDir != filepath.Dir(path) { + t.Fatalf("startup parent sync directory = %q, want %q", startupSyncDir, filepath.Dir(path)) + } + reopenedStored, reopenedChanged, err := reopened.UpsertGuardRealitySnapshot("generation-a", newer) + if err != nil || reopenedChanged { + t.Fatalf("reopened committed retry changed=%v err=%v", reopenedChanged, err) + } + if !reopenedStored.ReceivedAt.Equal(newer.ReceivedAt) { + t.Fatalf("reopened committed retry received_at = %s, want %s", reopenedStored.ReceivedAt, newer.ReceivedAt) + } + if err := reopened.ReadyCheck(); err == nil { + t.Fatal("reopened idempotent retry cleared durability degradation without a parent sync") + } + if startupSyncCalls != 1 { + t.Fatalf("idempotent retry parent sync calls = %d, want 1", startupSyncCalls) + } + confirmed := newer + confirmed.Reality.CollectedAt = newer.Reality.CollectedAt.Add(time.Minute) + confirmed.Reality.ManagedSHA = strings.Repeat("c", 64) + confirmed.ReceivedAt = newer.ReceivedAt.Add(time.Minute) + recoverySyncCalls := 0 + recoverySyncDir := "" + reopened.syncParentDir = func(dir string) error { + recoverySyncCalls++ + recoverySyncDir = dir + return nil + } + reopenedStored, reopenedChanged, err = reopened.UpsertGuardRealitySnapshot("generation-a", confirmed) + if err != nil || !reopenedChanged { + t.Fatalf("startup-degraded recovery changed=%v err=%v", reopenedChanged, err) + } + if recoverySyncCalls != 1 { + t.Fatalf("recovery parent sync calls = %d, want 1", recoverySyncCalls) + } + if recoverySyncDir != filepath.Dir(path) { + t.Fatalf("recovery parent sync directory = %q, want %q", recoverySyncDir, filepath.Dir(path)) + } + if err := reopened.ReadyCheck(); err != nil { + t.Fatalf("confirmed recovery sync left startup readiness degraded: %v", err) + } + + confirmedSyncCalls := 0 + confirmedSyncDir := "" + confirmedOpen, err := openWithCipher(path, cipher, func(dir string) error { + confirmedSyncCalls++ + confirmedSyncDir = dir + return nil + }) + if err != nil { + t.Fatalf("reopen with confirmed parent sync: %v", err) + } + t.Cleanup(func() { _ = confirmedOpen.Close() }) + if confirmedSyncCalls != 1 { + t.Fatalf("confirmed startup parent sync calls = %d, want 1", confirmedSyncCalls) + } + if confirmedSyncDir != filepath.Dir(path) { + t.Fatalf("confirmed startup parent sync directory = %q, want %q", confirmedSyncDir, filepath.Dir(path)) + } + if err := confirmedOpen.ReadyCheck(); err != nil { + t.Fatalf("successful startup parent sync left readiness degraded: %v", err) + } + got, ok = confirmedOpen.GuardRealitySnapshot("node-a") + if !ok || got.Reality.ManagedSHA != strings.Repeat("c", 64) { + t.Fatalf("confirmed startup snapshot: ok=%v snapshot=%+v", ok, got) + } + + st.syncParentDir = syncDir + if err := os.Mkdir(path+".tmp", 0o700); err != nil { + t.Fatalf("install degraded save-failure fixture: %v", err) + } + if _, _, err := st.UpsertGuardRealitySnapshot("generation-a", confirmed); err == nil { + t.Fatal("pre-rename failure unexpectedly succeeded while durability was degraded") + } + if err := st.ReadyCheck(); err == nil { + t.Fatal("pre-rename failure cleared durability-degraded readiness") + } + + stored, changed, err = st.UpsertGuardRealitySnapshot("generation-a", newer) + if err != nil || changed { + t.Fatalf("committed retry changed=%v err=%v", changed, err) + } + if !stored.ReceivedAt.Equal(newer.ReceivedAt) { + t.Fatalf("committed retry received_at = %s, want %s", stored.ReceivedAt, newer.ReceivedAt) + } + if err := st.ReadyCheck(); err == nil { + t.Fatal("idempotent retry cleared durability-degraded readiness without a parent sync") + } + + stored, changed, err = st.UpsertGuardRealitySnapshot("generation-a", confirmed) + if err != nil || !changed { + t.Fatalf("confirmed durable update changed=%v err=%v", changed, err) + } + if err := st.ReadyCheck(); err != nil { + t.Fatalf("successful parent sync did not clear durability-degraded readiness: %v", err) + } + finalReopened, err := OpenWithCipher(path, cipher) + if err != nil { + t.Fatalf("reopen confirmed durable snapshot: %v", err) + } + t.Cleanup(func() { _ = finalReopened.Close() }) + got, ok = finalReopened.GuardRealitySnapshot("node-a") + if !ok || got.Reality.ManagedSHA != strings.Repeat("c", 64) { + t.Fatalf("reopened confirmed snapshot: ok=%v snapshot=%+v", ok, got) + } +} + +func TestGuardRealitySnapshotBindsNodeIdentityGeneration(t *testing.T) { + st, err := Open("") + if err != nil { + t.Fatalf("open store: %v", err) + } + if err := st.UpsertNode(model.Node{ID: "node-a", LatticeIdentityUUID: "generation-old"}); err != nil { + t.Fatalf("upsert old generation: %v", err) + } + snapshot := GuardRealitySnapshot{ + Reality: model.GuardNodeReality{ + NodeID: "node-a", + CollectedAt: time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC), + }, + ReceivedAt: time.Date(2026, 7, 31, 13, 0, 1, 0, time.UTC), + } + if _, _, err := st.UpsertGuardRealitySnapshot("generation-old", snapshot); err != nil { + t.Fatalf("upsert old generation snapshot: %v", err) + } + if _, ok, err := st.DeleteNode("node-a"); err != nil || !ok { + t.Fatalf("delete old generation: ok=%v err=%v", ok, err) + } + if err := st.UpsertNode(model.Node{ID: "node-a", LatticeIdentityUUID: "generation-new"}); err != nil { + t.Fatalf("upsert new generation: %v", err) + } + if _, _, err := st.UpsertGuardRealitySnapshot("generation-old", snapshot); !errors.Is(err, ErrGuardRealityNodeChanged) { + t.Fatalf("old generation error = %v, want ErrGuardRealityNodeChanged", err) + } + if _, ok := st.GuardRealitySnapshot("node-a"); ok { + t.Fatal("old generation report attached to replacement node") + } + if _, changed, err := st.UpsertGuardRealitySnapshot("generation-new", snapshot); err != nil || !changed { + t.Fatalf("new generation upsert changed=%v err=%v", changed, err) + } +} + +func TestGuardRealitySnapshotCanonicalizesEmptyAndSetOrder(t *testing.T) { + st, err := Open("") + if err != nil { + t.Fatalf("open store: %v", err) + } + for _, nodeID := range []string{"node-empty", "node-order"} { + if err := st.UpsertNode(model.Node{ID: nodeID, LatticeIdentityUUID: nodeID + "-generation"}); err != nil { + t.Fatalf("upsert %s: %v", nodeID, err) + } + } + collectedAt := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + empty := GuardRealitySnapshot{ + Reality: model.GuardNodeReality{NodeID: "node-empty", CollectedAt: collectedAt}, + ReceivedAt: collectedAt.Add(time.Second), + } + if _, _, err := st.UpsertGuardRealitySnapshot("node-empty-generation", empty); err != nil { + t.Fatalf("upsert omitted empties: %v", err) + } + empty.Reality.Listeners = []model.GuardListener{} + empty.Reality.Interfaces = []model.GuardInterface{} + empty.Reality.ForeignTables = []string{} + if _, changed, err := st.UpsertGuardRealitySnapshot("node-empty-generation", empty); err != nil || changed { + t.Fatalf("explicit empty retry changed=%v err=%v", changed, err) + } + + ordered := GuardRealitySnapshot{ + Reality: model.GuardNodeReality{ + NodeID: "node-order", + Listeners: []model.GuardListener{ + {Protocol: "udp", Port: 53, Address: "2001:db8::2"}, + {Protocol: "tcp", Port: 22, Address: "2001:db8::1"}, + }, + Interfaces: []model.GuardInterface{ + {Name: "eth1", Addresses: []string{"2001:db8::2/128", "2001:db8::1/128"}}, + {Name: "eth0", Up: true}, + }, + ForeignTables: []string{"inet z", "inet a"}, + CollectedAt: collectedAt, + }, + ReceivedAt: collectedAt.Add(time.Second), + } + if _, _, err := st.UpsertGuardRealitySnapshot("node-order-generation", ordered); err != nil { + t.Fatalf("upsert unordered snapshot: %v", err) + } + ordered.Reality.Listeners[0], ordered.Reality.Listeners[1] = ordered.Reality.Listeners[1], ordered.Reality.Listeners[0] + ordered.Reality.Interfaces[0], ordered.Reality.Interfaces[1] = ordered.Reality.Interfaces[1], ordered.Reality.Interfaces[0] + ordered.Reality.Interfaces[0].Addresses = []string{} + ordered.Reality.Interfaces[1].Addresses[0], ordered.Reality.Interfaces[1].Addresses[1] = ordered.Reality.Interfaces[1].Addresses[1], ordered.Reality.Interfaces[1].Addresses[0] + ordered.Reality.ForeignTables[0], ordered.Reality.ForeignTables[1] = ordered.Reality.ForeignTables[1], ordered.Reality.ForeignTables[0] + if _, changed, err := st.UpsertGuardRealitySnapshot("node-order-generation", ordered); err != nil || changed { + t.Fatalf("reordered retry changed=%v err=%v", changed, err) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index f82fb23..439401a 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -39,6 +39,8 @@ const metricsPersistenceInterval = 5 * time.Minute // avoiding a full snapshot rewrite for every unchanged probe cycle. const monitorResultPersistenceInterval = 5 * time.Minute +var errStoreDurabilityDegraded = errors.New("store durability degraded: parent directory sync not confirmed") + type State struct { Users map[string]model.User `json:"users"` Tokens map[string]model.Token `json:"tokens"` @@ -51,42 +53,43 @@ type State struct { // distinct collection from KV on purpose: KV is plaintext at rest AND readable // over GET /api/kv by any principal holding kv:read. A secret must have neither // property, so it gets its own map, its own cipher pass, and no HTTP handler. - PluginSecrets map[string]model.KVEntry `json:"plugin_secrets"` - Static map[string]model.StaticObject `json:"static"` - StorageBuckets map[string]model.StorageBucket `json:"storage_buckets"` - StorageBindings map[string]model.StorageBinding `json:"storage_bindings"` - StorageTokens map[string]model.StorageAccessToken `json:"storage_tokens"` - Workers map[string]model.WorkerScript `json:"workers"` - Plugins map[string]model.PluginInstallation `json:"plugins"` - Approvals map[string]model.Approval `json:"approvals"` - Sessions map[string]auth.Session `json:"sessions"` - DDNS map[string]model.DDNSProfile `json:"ddns"` - Monitors map[string]model.Monitor `json:"monitors"` - MonResults map[string][]model.MonitorResult `json:"monitor_results"` - LogSources map[string]model.LogSource `json:"log_sources"` - NotifyChannels map[string]model.NotifyChannel `json:"notify_channels"` - NotifyRules map[string]model.NotifyRule `json:"notify_rules"` - Tunnels map[string]model.TunnelProfile `json:"tunnels"` - MachineProfiles map[string]model.MachineProfile `json:"machine_profiles"` - MachineVendors map[string]model.MachineVendor `json:"machine_vendors"` - NFTInputs map[string]model.NFTInputs `json:"nft_inputs"` - SecurityGroups map[string]model.SecurityGroup `json:"security_groups"` - GuardZones map[string]model.GuardZone `json:"guard_zones"` - GuardBindings map[string]model.NodeGuardBinding `json:"guard_bindings"` - DNSDeployments map[string]model.DNSDeployment `json:"dns_deployments"` - NetPolicies map[string]model.NetPolicy `json:"net_policies"` - Groups map[string]model.Group `json:"groups"` - GroupPolicies map[string]model.GroupNetPolicy `json:"group_policies"` - GeoRouting map[string]model.GeoRouting `json:"geo_routing"` - AgentUpdates map[string]model.AgentUpdatePolicy `json:"agent_updates"` - ProxyInbounds map[string]model.ProxyInbound `json:"proxy_inbounds"` - ProxyUsers map[string]model.ProxyUser `json:"proxy_users"` - ProxyProfiles map[string]model.ProxyNodeProfile `json:"proxy_profiles"` - ProxyUsage map[string]model.ProxyUsageSnapshot `json:"proxy_usage"` - TOTPChallenges map[string]auth.TOTPChallenge `json:"totp_challenges"` - OIDCProviders map[string]model.OIDCProvider `json:"oidc_providers"` - OIDCIdentities map[string]model.OIDCIdentity `json:"oidc_identities"` - OIDCAuthStates map[string]auth.OIDCAuthState `json:"oidc_auth_states"` + PluginSecrets map[string]model.KVEntry `json:"plugin_secrets"` + Static map[string]model.StaticObject `json:"static"` + StorageBuckets map[string]model.StorageBucket `json:"storage_buckets"` + StorageBindings map[string]model.StorageBinding `json:"storage_bindings"` + StorageTokens map[string]model.StorageAccessToken `json:"storage_tokens"` + Workers map[string]model.WorkerScript `json:"workers"` + Plugins map[string]model.PluginInstallation `json:"plugins"` + Approvals map[string]model.Approval `json:"approvals"` + Sessions map[string]auth.Session `json:"sessions"` + DDNS map[string]model.DDNSProfile `json:"ddns"` + Monitors map[string]model.Monitor `json:"monitors"` + MonResults map[string][]model.MonitorResult `json:"monitor_results"` + LogSources map[string]model.LogSource `json:"log_sources"` + NotifyChannels map[string]model.NotifyChannel `json:"notify_channels"` + NotifyRules map[string]model.NotifyRule `json:"notify_rules"` + Tunnels map[string]model.TunnelProfile `json:"tunnels"` + MachineProfiles map[string]model.MachineProfile `json:"machine_profiles"` + MachineVendors map[string]model.MachineVendor `json:"machine_vendors"` + NFTInputs map[string]model.NFTInputs `json:"nft_inputs"` + SecurityGroups map[string]model.SecurityGroup `json:"security_groups"` + GuardZones map[string]model.GuardZone `json:"guard_zones"` + GuardBindings map[string]model.NodeGuardBinding `json:"guard_bindings"` + GuardRealitySnapshots map[string]GuardRealitySnapshot `json:"guard_reality_snapshots"` + DNSDeployments map[string]model.DNSDeployment `json:"dns_deployments"` + NetPolicies map[string]model.NetPolicy `json:"net_policies"` + Groups map[string]model.Group `json:"groups"` + GroupPolicies map[string]model.GroupNetPolicy `json:"group_policies"` + GeoRouting map[string]model.GeoRouting `json:"geo_routing"` + AgentUpdates map[string]model.AgentUpdatePolicy `json:"agent_updates"` + ProxyInbounds map[string]model.ProxyInbound `json:"proxy_inbounds"` + ProxyUsers map[string]model.ProxyUser `json:"proxy_users"` + ProxyProfiles map[string]model.ProxyNodeProfile `json:"proxy_profiles"` + ProxyUsage map[string]model.ProxyUsageSnapshot `json:"proxy_usage"` + TOTPChallenges map[string]auth.TOTPChallenge `json:"totp_challenges"` + OIDCProviders map[string]model.OIDCProvider `json:"oidc_providers"` + OIDCIdentities map[string]model.OIDCIdentity `json:"oidc_identities"` + OIDCAuthStates map[string]auth.OIDCAuthState `json:"oidc_auth_states"` // WebAuthnCreds holds registered passkeys keyed by store record id. The public // keys and credential ids are non-secret, so this map is persisted as-is (no // at-rest envelope like Users/Sessions carry). @@ -108,6 +111,8 @@ type Store struct { walAnchorPath string runtimeBoltHot *BoltStateStore // optional record-level sidecar for high-churn runtime domains runtimeBoltHotPath string + syncParentDir func(string) error + durabilityDegraded bool // guarded by mu; only a confirmed parent sync clears it } // Open loads (or initializes) the store at path, resolving the at-rest @@ -132,15 +137,25 @@ func Open(path string) (*Store, error) { // it after logging the resolved key source; tests use it to inject a known // cipher. A nil cipher disables encryption. func OpenWithCipher(path string, cph secret.Cipher) (*Store, error) { + return openWithCipher(path, cph, syncDir) +} + +// openWithCipher lets store tests inject the startup directory sync without +// widening the public constructor or persistence API. +func openWithCipher(path string, cph secret.Cipher, syncParentDir func(string) error) (*Store, error) { if cph == nil { cph = secret.Disabled() } + if syncParentDir == nil { + syncParentDir = syncDir + } s := &Store{ path: path, state: emptyState(), metricsPersistedAt: map[string]time.Time{}, monitorPersistedAt: map[string]time.Time{}, cipher: cph, + syncParentDir: syncParentDir, } if path == "" { return s, nil @@ -162,6 +177,7 @@ func OpenWithCipher(path string, cph secret.Cipher) (*Store, error) { return nil, err } if len(data) == 0 { + s.confirmParentDirDurability() return s, nil } if err := json.Unmarshal(data, &s.state); err != nil { @@ -173,9 +189,18 @@ func OpenWithCipher(path string, cph secret.Cipher) (*Store, error) { s.ensureMaps() s.seedMetricsPersistence() s.seedMonitorResultPersistence() + s.confirmParentDirDurability() return s, nil } +func (s *Store) confirmParentDirDurability() { + syncParentDir := s.syncParentDir + if syncParentDir == nil { + syncParentDir = syncDir + } + s.durabilityDegraded = syncParentDir(filepath.Dir(s.path)) != nil +} + // EnableRuntimeBoltHotStore moves high-churn runtime collections to a // record-level bbolt sidecar while keeping the Store API and in-memory read // model unchanged. It is intentionally opt-in so operators can canary the Phase @@ -337,48 +362,49 @@ func monitorResultPersistenceKey(monitorID, nodeID string) string { func emptyState() State { return State{ - Users: map[string]model.User{}, - Tokens: map[string]model.Token{}, - Nodes: map[string]model.Node{}, - Tasks: map[string]model.Task{}, - KV: map[string]model.KVEntry{}, - PluginSecrets: map[string]model.KVEntry{}, - Static: map[string]model.StaticObject{}, - StorageBuckets: map[string]model.StorageBucket{}, - StorageBindings: map[string]model.StorageBinding{}, - StorageTokens: map[string]model.StorageAccessToken{}, - Workers: map[string]model.WorkerScript{}, - Plugins: map[string]model.PluginInstallation{}, - Approvals: map[string]model.Approval{}, - Sessions: map[string]auth.Session{}, - DDNS: map[string]model.DDNSProfile{}, - Monitors: map[string]model.Monitor{}, - MonResults: map[string][]model.MonitorResult{}, - LogSources: map[string]model.LogSource{}, - NotifyChannels: map[string]model.NotifyChannel{}, - NotifyRules: map[string]model.NotifyRule{}, - Tunnels: map[string]model.TunnelProfile{}, - MachineProfiles: map[string]model.MachineProfile{}, - MachineVendors: map[string]model.MachineVendor{}, - NFTInputs: map[string]model.NFTInputs{}, - SecurityGroups: map[string]model.SecurityGroup{}, - GuardZones: map[string]model.GuardZone{}, - GuardBindings: map[string]model.NodeGuardBinding{}, - DNSDeployments: map[string]model.DNSDeployment{}, - NetPolicies: map[string]model.NetPolicy{}, - Groups: map[string]model.Group{}, - GroupPolicies: map[string]model.GroupNetPolicy{}, - GeoRouting: map[string]model.GeoRouting{}, - AgentUpdates: map[string]model.AgentUpdatePolicy{}, - ProxyInbounds: map[string]model.ProxyInbound{}, - ProxyUsers: map[string]model.ProxyUser{}, - ProxyProfiles: map[string]model.ProxyNodeProfile{}, - ProxyUsage: map[string]model.ProxyUsageSnapshot{}, - TOTPChallenges: map[string]auth.TOTPChallenge{}, - OIDCProviders: map[string]model.OIDCProvider{}, - OIDCIdentities: map[string]model.OIDCIdentity{}, - OIDCAuthStates: map[string]auth.OIDCAuthState{}, - WebAuthnCreds: map[string]auth.WebAuthnCredential{}, + Users: map[string]model.User{}, + Tokens: map[string]model.Token{}, + Nodes: map[string]model.Node{}, + Tasks: map[string]model.Task{}, + KV: map[string]model.KVEntry{}, + PluginSecrets: map[string]model.KVEntry{}, + Static: map[string]model.StaticObject{}, + StorageBuckets: map[string]model.StorageBucket{}, + StorageBindings: map[string]model.StorageBinding{}, + StorageTokens: map[string]model.StorageAccessToken{}, + Workers: map[string]model.WorkerScript{}, + Plugins: map[string]model.PluginInstallation{}, + Approvals: map[string]model.Approval{}, + Sessions: map[string]auth.Session{}, + DDNS: map[string]model.DDNSProfile{}, + Monitors: map[string]model.Monitor{}, + MonResults: map[string][]model.MonitorResult{}, + LogSources: map[string]model.LogSource{}, + NotifyChannels: map[string]model.NotifyChannel{}, + NotifyRules: map[string]model.NotifyRule{}, + Tunnels: map[string]model.TunnelProfile{}, + MachineProfiles: map[string]model.MachineProfile{}, + MachineVendors: map[string]model.MachineVendor{}, + NFTInputs: map[string]model.NFTInputs{}, + SecurityGroups: map[string]model.SecurityGroup{}, + GuardZones: map[string]model.GuardZone{}, + GuardBindings: map[string]model.NodeGuardBinding{}, + GuardRealitySnapshots: map[string]GuardRealitySnapshot{}, + DNSDeployments: map[string]model.DNSDeployment{}, + NetPolicies: map[string]model.NetPolicy{}, + Groups: map[string]model.Group{}, + GroupPolicies: map[string]model.GroupNetPolicy{}, + GeoRouting: map[string]model.GeoRouting{}, + AgentUpdates: map[string]model.AgentUpdatePolicy{}, + ProxyInbounds: map[string]model.ProxyInbound{}, + ProxyUsers: map[string]model.ProxyUser{}, + ProxyProfiles: map[string]model.ProxyNodeProfile{}, + ProxyUsage: map[string]model.ProxyUsageSnapshot{}, + TOTPChallenges: map[string]auth.TOTPChallenge{}, + OIDCProviders: map[string]model.OIDCProvider{}, + OIDCIdentities: map[string]model.OIDCIdentity{}, + OIDCAuthStates: map[string]auth.OIDCAuthState{}, + WebAuthnCreds: map[string]auth.WebAuthnCredential{}, WebAuthnChallenges: map[string]auth.WebAuthnChallenge{}, } @@ -468,6 +494,9 @@ func (st *State) ensureMaps() { if st.GuardBindings == nil { st.GuardBindings = map[string]model.NodeGuardBinding{} } + if st.GuardRealitySnapshots == nil { + st.GuardRealitySnapshots = map[string]GuardRealitySnapshot{} + } if st.DNSDeployments == nil { st.DNSDeployments = map[string]model.DNSDeployment{} } @@ -527,35 +556,52 @@ func (s *Store) ensureMaps() { } func (s *Store) Save() error { + _, err := s.persistState(s.jsonPersistState()) + return err +} + +// persistState writes the supplied state without changing the live read model. +// Callers that need commit-style publication can persist a staged copy and +// install it in s.state only after this returns successfully. +func (s *Store) persistState(st State) (committed bool, err error) { start := time.Now() - var err error defer func() { telemetry.ObserveStoreSave(time.Since(start), err) }() if s.path == "" { - return nil + return true, nil } // 0o700: this directory holds only the server's private state file and, // in the auto-generate case, the master key. It must match the 0o700 used // by secret.generateKeyFile so neither path can widen the other (MkdirAll // is a no-op once the directory exists, so the first creator's mode wins). if err = os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil { - return err + return false, err } - persist, err := encryptedState(s.jsonPersistState(), s.cipher) + persist, err := encryptedState(st, s.cipher) if err != nil { - return fmt.Errorf("encrypt state: %w", err) + return false, fmt.Errorf("encrypt state: %w", err) } data, err := json.MarshalIndent(persist, "", " ") if err != nil { - return err + return false, err } - err = syncedAtomicWrite(s.path, data, 0o600) - return err + syncParentDir := s.syncParentDir + if syncParentDir == nil { + syncParentDir = syncDir + } + committed, err = syncedAtomicWriteStatus(s.path, data, 0o600, syncParentDir) + if committed { + s.durabilityDegraded = err != nil + } + return committed, err } func (s *Store) jsonPersistState() State { - st := s.state + return s.jsonPersistStateFrom(s.state) +} + +func (s *Store) jsonPersistStateFrom(st State) State { if s.runtimeBoltHot == nil { return st } @@ -567,13 +613,17 @@ func (s *Store) jsonPersistState() State { return st } -// ReadyCheck verifies that the in-memory state is initialized and can still be -// serialized with the configured at-rest cipher. It does not write to disk or -// return state contents; callers use it for readiness probes. +// ReadyCheck verifies that persistence has no unresolved directory-sync +// failure and that the in-memory state can still be serialized with the +// configured at-rest cipher. It does not write to disk or return state +// contents; callers use it for readiness probes. func (s *Store) ReadyCheck() error { s.mu.Lock() defer s.mu.Unlock() s.ensureMaps() + if s.durabilityDegraded { + return errStoreDurabilityDegraded + } if s.path != "" { if _, err := os.Stat(s.path); err != nil && !errors.Is(err, os.ErrNotExist) { return fmt.Errorf("stat state file: %w", err) @@ -602,16 +652,23 @@ func (s *Store) ReadyCheck() error { // For the primary state file that holds all credentials/secrets, that window is // total data loss, so we close it the same way the audit WAL already does. func syncedAtomicWrite(path string, data []byte, perm os.FileMode) error { + _, err := syncedAtomicWriteStatus(path, data, perm, syncDir) + return err +} + +// syncedAtomicWriteStatus reports whether rename crossed the commit point even +// when the following parent-directory fsync fails. +func syncedAtomicWriteStatus(path string, data []byte, perm os.FileMode, syncParentDir func(string) error) (bool, error) { tmp := path + ".tmp" if err := writeSyncedFile(tmp, data, perm); err != nil { os.Remove(tmp) - return err + return false, err } if err := os.Rename(tmp, path); err != nil { os.Remove(tmp) - return err + return false, err } - return syncDir(filepath.Dir(path)) + return true, syncParentDir(filepath.Dir(path)) } // writeSyncedFile writes data to path (creating/truncating) and fsyncs the file