package handlers import ( "encoding/json" "errors" "log" "net/http" "strings" "time" "blackbox/server/internal/auth" "blackbox/server/internal/models" "github.com/go-chi/chi/v5" "github.com/oklog/ulid/v2" "gorm.io/gorm" ) // SourceTypeDef describes a built-in source type for the catalog UI. type SourceTypeDef struct { Type string `json:"type"` Scope string `json:"scope"` Singleton bool `json:"singleton"` Name string `json:"name"` Description string `json:"description"` Mechanism string `json:"mechanism"` } var knownSourceTypes = []SourceTypeDef{ {Type: "docker", Scope: "agent", Singleton: true, Name: "Docker", Description: "Container lifecycle events from the local Docker socket", Mechanism: "agent · socket"}, {Type: "systemd", Scope: "agent", Singleton: true, Name: "Systemd", Description: "Service state changes for watched units via journald", Mechanism: "agent · journal"}, {Type: "filewatcher", Scope: "agent", Singleton: true, Name: "File Watcher", Description: "inotify events on watched config paths", Mechanism: "agent · inotify"}, {Type: "webhook_uptime_kuma", Scope: "server", Singleton: true, Name: "Uptime Kuma", Description: "Inbound webhook for Uptime Kuma monitor events", Mechanism: "server · http"}, {Type: "webhook_watchtower", Scope: "server", Singleton: true, Name: "Watchtower", Description: "Inbound webhook for Watchtower container update events", Mechanism: "server · http"}, {Type: "webhook_komodo", Scope: "server", Singleton: false, Name: "Komodo", Description: "Inbound webhook for Komodo deployment and automation events", Mechanism: "server · http"}, } var knownTypes = func() map[string]SourceTypeDef { m := make(map[string]SourceTypeDef, len(knownSourceTypes)) for _, t := range knownSourceTypes { m[t.Type] = t } return m }() type sourcesResponse struct { Server []models.DataSourceInstance `json:"server"` Nodes map[string]nodeSourcesResponse `json:"nodes"` Orphans []models.DataSourceInstance `json:"orphans"` } type nodeSourcesResponse struct { Capabilities []string `json:"capabilities"` AgentVersion string `json:"agent_version"` Status string `json:"status"` Sources []models.DataSourceInstance `json:"sources"` } func ListSources(db *gorm.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var instances []models.DataSourceInstance if err := db.Order("created_at ASC").Find(&instances).Error; err != nil { writeError(w, http.StatusInternalServerError, "failed to list sources") return } var nodes []models.Node if err := db.Find(&nodes).Error; err != nil { writeError(w, http.StatusInternalServerError, "failed to list nodes") return } now := time.Now() resp := sourcesResponse{ Server: []models.DataSourceInstance{}, Nodes: map[string]nodeSourcesResponse{}, Orphans: []models.DataSourceInstance{}, } for _, n := range nodes { var caps []string if err := json.Unmarshal([]byte(n.Capabilities), &caps); err != nil { log.Printf("ListSources: node %s has invalid capabilities JSON: %v", n.Name, err) caps = []string{"docker", "systemd", "filewatcher"} } else if len(caps) == 0 { log.Printf("ListSources: node %s has no stored capabilities yet; using legacy fallback", n.Name) caps = []string{"docker", "systemd", "filewatcher"} } resp.Nodes[n.Name] = nodeSourcesResponse{ Capabilities: caps, AgentVersion: n.AgentVersion, Status: effectiveNodeStatus(n, now), Sources: []models.DataSourceInstance{}, } } for _, inst := range instances { redacted := inst redacted.Config = redactConfig(inst.Type, inst.Config) if inst.Scope == "server" { resp.Server = append(resp.Server, redacted) continue } if inst.NodeID != nil { if nr, ok := resp.Nodes[*inst.NodeID]; ok { nr.Sources = append(nr.Sources, redacted) resp.Nodes[*inst.NodeID] = nr continue } } resp.Orphans = append(resp.Orphans, redacted) } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(resp); err != nil { log.Printf("ListSources encode: %v", err) } } } func ListSourceTypes() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(knownSourceTypes); err != nil { log.Printf("ListSourceTypes encode: %v", err) } } } type createSourceRequest struct { Type string `json:"type"` Scope string `json:"scope"` NodeID *string `json:"node_id"` Name string `json:"name"` Config json.RawMessage `json:"config"` Enabled *bool `json:"enabled"` } func CreateSource(db *gorm.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var req createSourceRequest if !decodeJSONBody(w, r, 64<<10, &req) { return } typeDef, ok := knownTypes[req.Type] if !ok { writeError(w, http.StatusBadRequest, "unknown source type: "+req.Type) return } if req.Type == "docker" { writeError(w, http.StatusBadRequest, "docker is a virtual source and cannot be created") return } req.Name = strings.TrimSpace(req.Name) if req.Name == "" { writeError(w, http.StatusBadRequest, "name is required") return } // Enforce scope from type definition (not caller-supplied) scope := typeDef.Scope // Enforce node_id: agent-scoped sources require it; server-scoped sources must not have it if scope == "agent" && (req.NodeID == nil || *req.NodeID == "") { writeError(w, http.StatusBadRequest, "node_id is required for agent-scoped sources") return } if scope == "server" && req.NodeID != nil && *req.NodeID != "" { writeError(w, http.StatusBadRequest, "node_id must not be set for server-scoped sources") return } // Enforce singleton: only one instance per type per node (or per server) if typeDef.Singleton { var count int64 q := db.Model(&models.DataSourceInstance{}).Where("type = ?", req.Type) if scope == "agent" { q = q.Where("node_id = ?", *req.NodeID) } else { q = q.Where("scope = ?", models.ScopeServer) } if err := q.Count(&count).Error; err != nil { writeError(w, http.StatusInternalServerError, "failed to check existing sources") return } if count > 0 { writeError(w, http.StatusConflict, "a source of this type already exists for this target") return } } obj := map[string]any{} if len(req.Config) > 0 { // Validate it's a JSON object if err := json.Unmarshal(req.Config, &obj); err != nil { writeError(w, http.StatusBadRequest, "config must be a JSON object") return } if obj == nil { writeError(w, http.StatusBadRequest, "config must be a JSON object") return } } if err := validateSourceConfig(req.Type, obj); err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } cfgJSON, err := json.Marshal(obj) if err != nil { writeError(w, http.StatusInternalServerError, "failed to serialize config") return } cfg := string(cfgJSON) enabled := true if req.Enabled != nil { enabled = *req.Enabled } now := time.Now().UTC() inst := models.DataSourceInstance{ ID: ulid.Make().String(), Type: req.Type, Scope: scope, NodeID: req.NodeID, Name: req.Name, Config: cfg, Enabled: enabled, CreatedAt: now, UpdatedAt: now, } if err := db.Select("*").Create(&inst).Error; err != nil { if isDuplicateKeyError(err) { writeError(w, http.StatusConflict, "a source of this type already exists for this target") return } writeError(w, http.StatusInternalServerError, "failed to create source") return } refreshWebhookSecretCacheIfNeeded(db, req.Type) claims, _ := auth.ClaimsFromContext(r.Context()) meta := map[string]interface{}{"type": inst.Type, "name": inst.Name, "scope": inst.Scope} if inst.NodeID != nil { meta["node_id"] = *inst.NodeID } WriteAuditLog(db, r, claims, "source.create", "data_source", inst.ID, meta) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusCreated) redacted := inst redacted.Config = redactConfig(inst.Type, inst.Config) if err := json.NewEncoder(w).Encode(redacted); err != nil { log.Printf("CreateSource encode: %v", err) } } } type updateSourceRequest struct { Name *string `json:"name"` Config json.RawMessage `json:"config"` Enabled *bool `json:"enabled"` } func UpdateSource(db *gorm.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") var inst models.DataSourceInstance if err := db.First(&inst, "id = ?", id).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { writeError(w, http.StatusNotFound, "source not found") return } log.Printf("UpdateSource: lookup failed for %s: %v", id, err) writeError(w, http.StatusInternalServerError, "database error") return } var req updateSourceRequest if !decodeJSONBody(w, r, 64<<10, &req) { return } if req.Name != nil { trimmedName := strings.TrimSpace(*req.Name) if trimmedName == "" { writeError(w, http.StatusBadRequest, "name is required") return } inst.Name = trimmedName } if len(req.Config) > 0 { var incoming map[string]any if err := json.Unmarshal(req.Config, &incoming); err != nil { writeError(w, http.StatusBadRequest, "config must be a JSON object") return } if incoming == nil { writeError(w, http.StatusBadRequest, "config must be a JSON object") return } mergedConfig, err := mergeSourceConfig(inst.Type, inst.Config, incoming) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } inst.Config = mergedConfig } if req.Enabled != nil { inst.Enabled = *req.Enabled } inst.UpdatedAt = time.Now().UTC() if err := db.Save(&inst).Error; err != nil { writeError(w, http.StatusInternalServerError, "failed to update source") return } refreshWebhookSecretCacheIfNeeded(db, inst.Type) claims, _ := auth.ClaimsFromContext(r.Context()) WriteAuditLog(db, r, claims, "source.update", "data_source", inst.ID, map[string]interface{}{ "type": inst.Type, "name": inst.Name, "enabled": inst.Enabled, }) w.Header().Set("Content-Type", "application/json") redacted := inst redacted.Config = redactConfig(inst.Type, inst.Config) if err := json.NewEncoder(w).Encode(redacted); err != nil { log.Printf("UpdateSource encode: %v", err) } } } func DeleteSource(db *gorm.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") var inst models.DataSourceInstance if err := db.First(&inst, "id = ?", id).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { writeError(w, http.StatusNotFound, "source not found") return } log.Printf("DeleteSource: lookup failed for %s: %v", id, err) writeError(w, http.StatusInternalServerError, "database error") return } result := db.Delete(&models.DataSourceInstance{}, "id = ?", id) if result.Error != nil { writeError(w, http.StatusInternalServerError, "failed to delete source") return } if result.RowsAffected == 0 { writeError(w, http.StatusNotFound, "source not found") return } refreshWebhookSecretCacheIfNeeded(db, inst.Type) claims, _ := auth.ClaimsFromContext(r.Context()) WriteAuditLog(db, r, claims, "source.delete", "data_source", id, map[string]interface{}{ "type": inst.Type, "name": inst.Name, }) w.WriteHeader(http.StatusNoContent) } } // redactConfig removes sensitive fields from source config before sending to clients. func redactConfig(sourceType, config string) string { keys := sensitiveKeysFor(sourceType) if len(keys) == 0 { return config } var m map[string]any if err := json.Unmarshal([]byte(config), &m); err != nil { return "{}" } for _, key := range keys { if _, ok := m[key]; ok { m[key] = "" } } out, err := json.Marshal(m) if err != nil { log.Printf("redactConfig: failed to marshal redacted %s config: %v", sourceType, err) return "{}" } return string(out) } // GetWebhookSecret returns the secret for an enabled webhook source type. func GetWebhookSecret(db *gorm.DB, sourceType string) string { var inst models.DataSourceInstance if err := db.Where("type = ? AND enabled = ?", sourceType, true).Order("created_at ASC").First(&inst).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return "" } log.Printf("GetWebhookSecret: sourceType=%s lookup failed: %v", sourceType, err) return "" } var cfg struct { Secret string `json:"secret"` } if err := json.Unmarshal([]byte(inst.Config), &cfg); err != nil || cfg.Secret == "" { return "" } return cfg.Secret } // GetAllWebhookSecrets returns a map of source ID → secret for all enabled instances of sourceType. func GetAllWebhookSecrets(db *gorm.DB, sourceType string) map[string]string { var instances []models.DataSourceInstance if err := db.Where("type = ? AND enabled = ?", sourceType, true).Find(&instances).Error; err != nil { log.Printf("GetAllWebhookSecrets: sourceType=%s lookup failed: %v", sourceType, err) return map[string]string{} } result := make(map[string]string, len(instances)) for _, inst := range instances { var cfg struct { Secret string `json:"secret"` } if err := json.Unmarshal([]byte(inst.Config), &cfg); err != nil || cfg.Secret == "" { continue } result[inst.ID] = cfg.Secret } return result } func sensitiveKeysFor(sourceType string) []string { switch { case strings.HasPrefix(sourceType, "webhook_"): return []string{"secret"} default: return nil } } func mergeSourceConfig(sourceType, existingConfig string, incoming map[string]any) (string, error) { if incoming == nil { return "", errors.New("config must be a JSON object") } // Defensive copy: this function mutates its working map (back-filling missing // keys and restoring blank sensitive fields). Copy incoming so callers are unaffected. work := make(map[string]any, len(incoming)) for k, v := range incoming { work[k] = v } incoming = work existing := map[string]any{} if existingConfig != "" { if err := json.Unmarshal([]byte(existingConfig), &existing); err != nil { existing = map[string]any{} } } for key, value := range existing { if _, ok := incoming[key]; !ok { incoming[key] = value } } if err := validateSourceConfig(sourceType, incoming); err != nil { return "", err } for _, key := range sensitiveKeysFor(sourceType) { incomingValue, ok := incoming[key] if !ok { continue } existingValue, existingOK := existing[key] if !existingOK { continue } existingString, existingIsString := existingValue.(string) if !existingIsString || existingString == "" { continue } incomingString, incomingIsString := incomingValue.(string) if incomingValue == nil || !incomingIsString || incomingString == "" { incoming[key] = existingString } } merged, err := json.Marshal(incoming) if err != nil { return "", err } return string(merged), nil } func validateSourceConfig(sourceType string, config map[string]any) error { if sourceType == "systemd" { rawValue, ok := config["units"] if !ok { return errors.New("units is required") } normalizedUnits, err := normalizeSystemdUnits(rawValue) if err != nil { return err } config["units"] = normalizedUnits } if strings.HasPrefix(sourceType, "webhook_") { rawValue, ok := config["secret"] if !ok { return errors.New("secret is required") } secret, ok := rawValue.(string) if !ok || strings.TrimSpace(secret) == "" { return errors.New("secret must be a non-empty string") } } if sourceType == "filewatcher" { rawValue, ok := config["redact_secrets"] if !ok { return errors.New("redact_secrets is required") } if _, ok := rawValue.(bool); !ok { return errors.New("redact_secrets must be a boolean") } } if sourceType == "webhook_komodo" { rawAllowed, ok := config["allowed_types"] if !ok { return errors.New("allowed_types is required") } allowedSlice, ok := rawAllowed.([]any) if !ok { return errors.New("allowed_types must be an array of strings") } if len(allowedSlice) == 0 { return errors.New("allowed_types must not be empty") } normalized := make([]any, 0, len(allowedSlice)) for _, v := range allowedSlice { s, ok := v.(string) if !ok { return errors.New("allowed_types must contain non-empty strings") } s = strings.ToLower(strings.TrimSpace(s)) if s == "" { return errors.New("allowed_types must contain non-empty strings") } normalized = append(normalized, s) } config["allowed_types"] = normalized if rawMap, ok := config["node_map"]; ok && rawMap != nil { nodeMap, ok := rawMap.(map[string]any) if !ok { return errors.New("node_map must be an object") } normMap := make(map[string]any, len(nodeMap)) for k, v := range nodeMap { k = strings.ToLower(strings.TrimSpace(k)) if k == "" { return errors.New("node_map keys must be non-empty strings") } if _, exists := normMap[k]; exists { return errors.New("node_map contains duplicate keys after normalization") } s, ok := v.(string) if !ok { return errors.New("node_map values must be non-empty strings") } s = strings.ToLower(strings.TrimSpace(s)) if s == "" { return errors.New("node_map values must be non-empty strings") } normMap[k] = s } config["node_map"] = normMap } } return nil } func normalizeSystemdUnits(rawValue any) ([]string, error) { var rawUnits []any switch units := rawValue.(type) { case []any: rawUnits = units case []string: rawUnits = make([]any, 0, len(units)) for _, unit := range units { rawUnits = append(rawUnits, unit) } default: return nil, errors.New("units must be an array of strings") } seen := make(map[string]struct{}, len(rawUnits)) clean := make([]string, 0, len(rawUnits)) for _, rawUnit := range rawUnits { unit, ok := rawUnit.(string) if !ok { return nil, errors.New("units must be an array of strings") } unit = strings.TrimSpace(unit) if unit == "" { continue } if !hasUnitTypeSuffix(unit) { unit += ".service" } if !validUnitName.MatchString(unit) { return nil, errors.New("invalid unit name: " + unit) } if _, dup := seen[unit]; dup { continue } seen[unit] = struct{}{} clean = append(clean, unit) } return clean, nil } func isDuplicateKeyError(err error) bool { if err == nil { return false } if errors.Is(err, gorm.ErrDuplicatedKey) { return true } msg := strings.ToLower(err.Error()) return strings.Contains(msg, "unique constraint failed") || strings.Contains(msg, "duplicate key value violates unique constraint") }