First release of open core

2026-04-02 10:57:36 -04:00
parent 1c94f12d1c
commit 084c1321fc
101 changed files with 8812 additions and 17 deletions
--- a/pkg/ingest/ingest.go
+++ b/pkg/ingest/ingest.go
@@ -0,0 +1,163 @@
+package ingest
+
+import (
+	"crypto/sha256"
+	"encoding/csv"
+	"encoding/hex"
+	"encoding/json"
+	"log"
+	"net/http"
+	"strconv"
+
+	"epigas.gitea.cloud/RiskRancher/core/pkg/domain"
+)
+
+func (h *Handler) HandleIngest(w http.ResponseWriter, r *http.Request) {
+	decoder := json.NewDecoder(r.Body)
+	_, err := decoder.Token()
+	if err != nil {
+		http.Error(w, "Invalid JSON payload: expected array", http.StatusBadRequest)
+		return
+	}
+
+	type groupKey struct {
+		Source string
+		Asset  string
+	}
+	groupedTickets := make(map[groupKey][]domain.Ticket)
+	for decoder.More() {
+		var ticket domain.Ticket
+		if err := decoder.Decode(&ticket); err != nil {
+			http.Error(w, "Error parsing ticket object", http.StatusBadRequest)
+			return
+		}
+
+		if ticket.Status == "" {
+			ticket.Status = "Waiting to be Triaged"
+		}
+
+		if ticket.DedupeHash == "" {
+			hashInput := ticket.Source + "|" + ticket.AssetIdentifier + "|" + ticket.Title
+			hash := sha256.Sum256([]byte(hashInput))
+			ticket.DedupeHash = hex.EncodeToString(hash[:])
+		}
+
+		key := groupKey{
+			Source: ticket.Source,
+			Asset:  ticket.AssetIdentifier,
+		}
+		groupedTickets[key] = append(groupedTickets[key], ticket)
+	}
+
+	_, err = decoder.Token()
+	if err != nil {
+		http.Error(w, "Invalid JSON payload termination", http.StatusBadRequest)
+		return
+	}
+
+	for key, batch := range groupedTickets {
+		err := h.Store.ProcessIngestionBatch(r.Context(), key.Source, key.Asset, batch)
+		if err != nil {
+			log.Printf("🔥 Ingestion DB Error for Asset %s: %v", key.Asset, err)
+			h.Store.LogSync(r.Context(), key.Source, "Failed", len(batch), err.Error())
+			http.Error(w, "Database error processing batch", http.StatusInternalServerError)
+			return
+		} else {
+			h.Store.LogSync(r.Context(), key.Source, "Success", len(batch), "")
+		}
+	}
+
+	w.WriteHeader(http.StatusCreated)
+}
+
+func (h *Handler) HandleCSVIngest(w http.ResponseWriter, r *http.Request) {
+	if err := r.ParseMultipartForm(10 << 20); err != nil {
+		http.Error(w, "Failed to parse form", http.StatusBadRequest)
+		return
+	}
+
+	adapterIDStr := r.FormValue("adapter_id")
+	adapterID, err := strconv.Atoi(adapterIDStr)
+	if err != nil {
+		http.Error(w, "Invalid adapter_id", http.StatusBadRequest)
+		return
+	}
+
+	adapter, err := h.Store.GetAdapterByID(r.Context(), adapterID)
+	if err != nil {
+		http.Error(w, "Adapter mapping not found", http.StatusNotFound)
+		return
+	}
+
+	file, _, err := r.FormFile("file")
+	if err != nil {
+		http.Error(w, "Failed to read file payload", http.StatusBadRequest)
+		return
+	}
+	defer file.Close()
+
+	reader := csv.NewReader(file)
+	records, err := reader.ReadAll()
+	if err != nil || len(records) < 2 {
+		http.Error(w, "Invalid or empty CSV format", http.StatusBadRequest)
+		return
+	}
+
+	headers := records[0]
+	headerMap := make(map[string]int)
+	for i, h := range headers {
+		headerMap[h] = i
+	}
+
+	type groupKey struct {
+		Source string
+		Asset  string
+	}
+	groupedTickets := make(map[groupKey][]domain.Ticket)
+
+	for _, row := range records[1:] {
+		ticket := domain.Ticket{
+			Source: adapter.SourceName,
+			Status: "Waiting to be Triaged",
+		}
+
+		if idx, ok := headerMap[adapter.MappingTitle]; ok && idx < len(row) {
+			ticket.Title = row[idx]
+		}
+		if idx, ok := headerMap[adapter.MappingAsset]; ok && idx < len(row) {
+			ticket.AssetIdentifier = row[idx]
+		}
+		if idx, ok := headerMap[adapter.MappingSeverity]; ok && idx < len(row) {
+			ticket.Severity = row[idx]
+		}
+		if idx, ok := headerMap[adapter.MappingDescription]; ok && idx < len(row) {
+			ticket.Description = row[idx]
+		}
+		if adapter.MappingRemediation != "" {
+			if idx, ok := headerMap[adapter.MappingRemediation]; ok && idx < len(row) {
+				ticket.RecommendedRemediation = row[idx]
+			}
+		}
+
+		if ticket.Title != "" && ticket.AssetIdentifier != "" {
+			hashInput := ticket.Source + "|" + ticket.AssetIdentifier + "|" + ticket.Title
+			hash := sha256.Sum256([]byte(hashInput))
+			ticket.DedupeHash = hex.EncodeToString(hash[:])
+			key := groupKey{Source: ticket.Source, Asset: ticket.AssetIdentifier}
+			groupedTickets[key] = append(groupedTickets[key], ticket)
+		}
+	}
+
+	for key, batch := range groupedTickets {
+		err := h.Store.ProcessIngestionBatch(r.Context(), key.Source, key.Asset, batch)
+		if err != nil {
+			log.Printf("🔥 CSV Ingestion Error for Asset %s: %v", key.Asset, err)
+			h.Store.LogSync(r.Context(), key.Source, "Failed", len(batch), err.Error())
+			http.Error(w, "Database error processing CSV batch", http.StatusInternalServerError)
+			return
+		} else {
+			h.Store.LogSync(r.Context(), key.Source, "Success", len(batch), "")
+		}
+	}
+	w.WriteHeader(http.StatusCreated)
+}