(feat.) add notification apis phase I

deepfence_server/apiDocs/docs.go

@@ -36,6 +36,7 @@ const (
 	tagSettings          = "Settings"
 	tagDiffAdd           = "Diff Add"
 	tagCompletion        = "Completion"
+	tagNotification      = "Notification"
 
 	securityName = "bearer_token"
 )

deepfence_server/apiDocs/operation.go

@@ -922,3 +922,18 @@ func (d *OpenAPIDocs) AddCompletionOperations() {
 		"Get Completion for Container fields", "Complete Container info",
 		http.StatusOK, []string{tagCompletion}, bearerToken, new(CompletionNodeFieldReq), new(CompletionNodeFieldRes))
 }
+
+func (d *OpenAPIDocs) AddNotificationOperations() {
+	d.AddOperation("getNotificationScans", http.MethodGet, "/deepfence/notification/scans",
+		"Get Notification", "Get Scans for Notification",
+		http.StatusOK, []string{tagNotification}, bearerToken, new(NotificationGetScanRequest), new(NotificationGetScanResponse))
+	d.AddOperation("markNotificationScansRead", http.MethodPost, "/deepfence/notification/scans/mark-read",
+		"Mark Notification Scans Read", "Mark Notification Scans Read",
+		http.StatusNoContent, []string{tagNotification}, bearerToken, new(NotificationMarkScanReadRequest), nil)
+	d.AddOperation("getNotificationRegistrySync", http.MethodGet, "/deepfence/notification/registry-sync",
+		"Get Notification", "Get Registry Sync for Notification",
+		http.StatusOK, []string{tagNotification}, bearerToken, nil, new([]RegistryAccount))
+	d.AddOperation("getNotificationIntegrationFailures", http.MethodGet, "/deepfence/notification/integration",
+		"Get Notification", "Get Integration Failures for Notification",
+		http.StatusOK, []string{tagNotification}, bearerToken, nil, new([]IntegrationListResp))
+}

deepfence_server/auth/policy.csv

@@ -92,3 +92,10 @@ p, admin, license, read
 p, admin, license, write
 p, admin, license, delete
 p, standard-user, license, read
+
+p, admin, notification, read
+p, admin, notification, write
+p, admin, notification, delete
+p, standard-user, notification, read
+p, standard-user, notification, write
+p, read-only-user, notification, read

deepfence_server/handler/integration.go

@@ -176,13 +176,19 @@ func (h *Handler) GetIntegrations(w http.ResponseWriter, r *http.Request) {
 			integrationStatus = integration.ErrorMsg.String
 		}
 
+		var lastSentTime string
+		if integration.LastSentTime.Valid {
+			lastSentTime = integration.LastSentTime.Time.String()
+		}
+
 		newIntegration := model.IntegrationListResp{
 			ID:               integration.ID,
 			IntegrationType:  integration.IntegrationType,
 			NotificationType: integration.Resource,
 			Config:           config,
 			Filters:          filters,
 			LastErrorMsg:     integrationStatus,
+			LastSentTime:     lastSentTime,
 		}
 
 		newIntegration.RedactSensitiveFieldsInConfig()

deepfence_server/handler/notification.go

@@ -0,0 +1,108 @@
+package handler
+
+import (
+	"net/http"
+
+	"github.com/deepfence/ThreatMapper/deepfence_server/model"
+	"github.com/deepfence/ThreatMapper/deepfence_server/reporters/notification"
+	"github.com/deepfence/ThreatMapper/deepfence_utils/log"
+	httpext "github.com/go-playground/pkg/v5/net/http"
+)
+
+func (h *Handler) GetScansHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	var req model.NotificationGetScanRequest
+
+	// parse request body
+	err := httpext.DecodeJSON(r, httpext.NoQueryParams, MaxPostRequestSize, &req)
+	if err != nil {
+		log.Error().Msgf("Error decoding request: %v", err)
+		h.respondError(err, w)
+		return
+	}
+
+	// TODO: check if status provided are valid
+
+	// get scans from db
+	scans, err := notification.GetScans(ctx, req.ScanTypes, req.Statuses)
+	if err != nil {
+		log.Error().Msgf("Error getting scans: %v", err)
+		h.respondError(err, w)
+		return
+	}
+
+	// respond with scans
+	err = httpext.JSON(w, http.StatusOK, scans)
+	if err != nil {
+		log.Error().Msgf("Error responding: %v", err)
+	}
+}
+
+func (h *Handler) MarkScansReadHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	var req model.NotificationMarkScanReadRequest
+
+	// parse request body
+	err := httpext.DecodeJSON(r, httpext.NoQueryParams, MaxPostRequestSize, &req)
+	if err != nil {
+		log.Error().Msgf("Error decoding request: %v", err)
+		h.respondError(err, w)
+		return
+	}
+
+	// mark scans as read
+	err = notification.MarkScansRead(ctx, req.ScanType, req.NodeIDs)
+	if err != nil {
+		log.Error().Msgf("Error marking scans as read: %v", err)
+		h.respondError(err, w)
+		return
+	}
+
+	// respond with success
+	err = httpext.JSON(w, http.StatusOK, nil)
+	if err != nil {
+		log.Error().Msgf("Error responding: %v", err)
+	}
+}
+
+/* Registry Sync Handlers */
+
+// GetRegistrySyncHandler returns the registries that are syncing
+func (h *Handler) GetRegistrySyncHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	// get registries that are syncing
+	registries, err := notification.GetRegistrySync(ctx)
+	if err != nil {
+		log.Error().Msgf("Error getting registries that are syncing: %v", err)
+		h.respondError(err, w)
+		return
+	}
+
+	// respond with registries
+	err = httpext.JSON(w, http.StatusOK, registries)
+	if err != nil {
+		log.Error().Msgf("Error responding: %v", err)
+	}
+}
+
+/* Integration Handlers */
+
+// GetIntegrationFailuresHandler returns the integrations that have failed
+func (h *Handler) GetIntegrationFailuresHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	// get integrations that have failed
+	integrations, err := notification.GetIntegrationFailures(ctx)
+	if err != nil {
+		log.Error().Msgf("Error getting integrations that have failed: %v", err)
+		h.respondError(err, w)
+		return
+	}
+
+	// respond with integrations
+	err = httpext.JSON(w, http.StatusOK, integrations)
+	if err != nil {
+		log.Error().Msgf("Error responding: %v", err)
+	}
+}

deepfence_server/main.go

@@ -320,6 +320,7 @@ func initializeOpenAPIDocs(openAPIDocs *apiDocs.OpenAPIDocs) {
 	openAPIDocs.AddDiffAddOperations()
 	openAPIDocs.AddCompletionOperations()
 	openAPIDocs.AddLicenseOperations()
+	openAPIDocs.AddNotificationOperations()
 }
 
 func initializeInternalOpenAPIDocs(openAPIDocs *apiDocs.OpenAPIDocs) {

deepfence_server/model/integration.go

@@ -103,6 +103,7 @@ type IntegrationListResp struct {
 	Config           map[string]interface{} `json:"config"`
 	Filters          IntegrationFilters     `json:"filters"`
 	LastErrorMsg     string                 `json:"last_error_msg"`
+	LastSentTime     string                 `json:"last_sent_time"`
 }
 
 func (i *IntegrationListReq) GetIntegrations(ctx context.Context, pgClient *postgresqlDb.Queries) ([]postgresqlDb.Integration, error) {

deepfence_server/model/notification.go

@@ -0,0 +1,49 @@
+package model
+
+type NotificationGetScanResponse struct {
+	VulnerabilityScan   []Scan `json:"vulnerability_scan"`
+	SecretScan          []Scan `json:"secret_scan"`
+	MalwareScan         []Scan `json:"malware_scan"`
+	ComplianceScan      []Scan `json:"compliance_scan"`
+	CloudComplianceScan []Scan `json:"cloud_compliance_scan"`
+}
+
+type Scan struct {
+	CreatedAt     int64  `json:"created_at"`
+	UpdatedAt     int64  `json:"updated_at"`
+	NodeID        string `json:"node_id"`
+	IsPriority    bool   `json:"is_priority"`
+	Status        string `json:"status"`
+	StatusMessage string `json:"status_message"`
+	TriggerAction string `json:"trigger_action"`
+	Retries       int64  `json:"retries"`
+}
+
+// TODO: later
+type TriggerAction struct {
+	ID             int    `json:"id"`
+	RequestPayload string `json:"request_payload"`
+}
+type RequestPayload struct {
+	NodeID   string `json:"node_id"`
+	NodeType int    `json:"node_type"`
+	BinArgs  struct {
+		NodeID     string `json:"node_id"`
+		NodeType   string `json:"node_type"`
+		RegistryID string `json:"registry_id"`
+		ScanID     string `json:"scan_id"`
+		ScanType   string `json:"scan_type"`
+	} `json:"bin_args"`
+}
+
+type NotificationGetScanRequest struct {
+	ScanTypes []string `json:"scan_types"`
+	Statuses  []string `json:"status"`
+	Page      int      `json:"page"`
+	Limit     int      `json:"limit"`
+}
+
+type NotificationMarkScanReadRequest struct {
+	ScanType string   `json:"scan_type"`
+	NodeIDs  []string `json:"node_ids"`
+}

deepfence_server/reporters/notification/integration.go

@@ -0,0 +1,69 @@
+package notification
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/deepfence/ThreatMapper/deepfence_server/model"
+	"github.com/deepfence/ThreatMapper/deepfence_utils/directory"
+	"github.com/deepfence/ThreatMapper/deepfence_utils/log"
+)
+
+// GetIntegrationFailures returns the integrations that have failed
+func GetIntegrationFailures(ctx context.Context) ([]model.IntegrationListResp, error) {
+	var failedIntegrations []model.IntegrationListResp
+	pgClient, err := directory.PostgresClient(ctx)
+	if err != nil {
+		return failedIntegrations, nil
+	}
+
+	integrations, err := pgClient.GetIntegrations(ctx)
+	if err != nil {
+		log.Error().Msgf("Error getting postgresCtx: %v", err)
+		return failedIntegrations, err
+	}
+
+	// filter out integrations that have errorMsg
+	for _, integration := range integrations {
+		if integration.ErrorMsg.Valid {
+			var config map[string]interface{}
+			var filters model.IntegrationFilters
+
+			err = json.Unmarshal(integration.Config, &config)
+			if err != nil {
+				log.Error().Msgf(err.Error())
+				continue
+			}
+			err = json.Unmarshal(integration.Filters, &filters)
+			if err != nil {
+				log.Error().Msgf(err.Error())
+				continue
+			}
+
+			var integrationStatus string
+			if integration.ErrorMsg.Valid {
+				integrationStatus = integration.ErrorMsg.String
+			}
+
+			var lastSentTime string
+			if integration.LastSentTime.Valid {
+				lastSentTime = integration.LastSentTime.Time.String()
+			}
+
+			newIntegration := model.IntegrationListResp{
+				ID:               integration.ID,
+				IntegrationType:  integration.IntegrationType,
+				NotificationType: integration.Resource,
+				Config:           config,
+				Filters:          filters,
+				LastErrorMsg:     integrationStatus,
+				LastSentTime:     lastSentTime,
+			}
+
+			newIntegration.RedactSensitiveFieldsInConfig()
+			failedIntegrations = append(failedIntegrations, newIntegration)
+		}
+	}
+
+	return failedIntegrations, nil
+}

deepfence_server/reporters/notification/registry.go

@@ -0,0 +1,62 @@
+package notification
+
+import (
+	"context"
+	"time"
+
+	"github.com/deepfence/ThreatMapper/deepfence_server/model"
+	"github.com/deepfence/ThreatMapper/deepfence_utils/directory"
+	"github.com/deepfence/ThreatMapper/deepfence_utils/log"
+	"github.com/neo4j/neo4j-go-driver/v5/neo4j"
+)
+
+// GetRegistrySync returns the registries that are syncing
+func GetRegistrySync(ctx context.Context) ([]model.RegistryAccount, error) {
+	registries := []model.RegistryAccount{}
+
+	driver, err := directory.Neo4jClient(ctx)
+	if err != nil {
+		return registries, err
+	}
+
+	log.Info().Msgf("Getting registries that are syncing")
+
+	session := driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeRead})
+	defer session.Close(ctx)
+
+	tx, err := session.BeginTransaction(ctx, neo4j.WithTxTimeout(30*time.Second))
+	if err != nil {
+		return registries, err
+	}
+	defer tx.Close(ctx)
+	query := `
+		MATCH (r:RegistryAccount)
+		WHERE r.syncing = true
+		RETURN r.name, r.node_id, r.registry_type, r.syncing
+	`
+	log.Debug().Msgf("Query: %s", query)
+	result, err := tx.Run(ctx, query, map[string]interface{}{})
+	if err != nil {
+		return registries, err
+	}
+
+	rec, err := result.Collect(ctx)
+	if err != nil {
+		return registries, err
+	}
+
+	if len(rec) == 0 {
+		return registries, nil
+	}
+
+	for _, record := range rec {
+		reg := model.RegistryAccount{}
+		reg.Name = record.Values[0].(string)
+		reg.ID = record.Values[1].(string)
+		reg.RegistryType = record.Values[2].(string)
+		reg.Syncing = record.Values[3].(bool)
+		registries = append(registries, reg)
+	}
+
+	return registries, nil
+}