Cloud scanner resources refresh progress (#2316)

(cherry picked from commit f1f3a4b)

.gitmodules

@@ -1,32 +1,32 @@
 [submodule "deepfence_agent/plugins/agent-plugins-grpc"]
 	path = deepfence_agent/plugins/agent-plugins-grpc
 	url = https://github.com/deepfence/agent-plugins-grpc
-	branch = release-2.3
+	branch = release-2.4
 [submodule "deepfence_agent/plugins/package-scanner"]
 	path = deepfence_agent/plugins/package-scanner
 	url = https://github.com/deepfence/package-scanner
-	branch = release-2.3
+	branch = release-2.4
 [submodule "deepfence_agent/plugins/SecretScanner"]
 	path = deepfence_agent/plugins/SecretScanner
 	url = https://github.com/deepfence/SecretScanner
-	branch = release-2.3
+	branch = release-2.4
 [submodule "deepfence_agent/plugins/compliance"]
 	path = deepfence_agent/plugins/compliance
 	url = https://github.com/deepfence/compliance
-	branch = release-2.3
+	branch = release-2.4
 [submodule "golang_deepfence_sdk"]
 	path = golang_deepfence_sdk
 	url = https://github.com/deepfence/golang_deepfence_sdk.git
-	branch = release-2.3
+	branch = release-2.4
 [submodule "deepfence_agent/plugins/YaraHunter"]
 	path = deepfence_agent/plugins/YaraHunter
 	url = https://github.com/deepfence/YaraHunter.git
-	branch = release-2.3
+	branch = release-2.4
 [submodule "deepfence_agent/plugins/yara-rules"]
 	path = deepfence_agent/plugins/yara-rules
 	url = https://github.com/deepfence/yara-rules
-	branch = release-2.3
+	branch = release-2.4
 [submodule "deepfence_agent/plugins/cloud-scanner"]
 	path = deepfence_agent/plugins/cloud-scanner
 	url = https://github.com/deepfence/cloud-scanner
-	branch = release-2.3
+	branch = release-2.4

Makefile

@@ -18,7 +18,7 @@ export IMAGE_REPOSITORY?=quay.io/deepfenceio
 export DF_IMG_TAG?=latest
 export STEAMPIPE_IMG_TAG?=0.23.x
 export IS_DEV_BUILD?=false
-export VERSION?=v2.3.1
+export VERSION?=v2.4.0
 export AGENT_BINARY_BUILD=$(DEEPFENCE_FARGATE_DIR)/build
 export AGENT_BINARY_BUILD_RELATIVE=deepfence_agent/agent-binary/build
 export AGENT_BINARY_DIST=$(DEEPFENCE_FARGATE_DIR)/dist

README.md

@@ -94,10 +94,10 @@ docker run -dit \
     -e MGMT_CONSOLE_URL="---CONSOLE-IP---" \
     -e MGMT_CONSOLE_PORT="443" \
     -e DEEPFENCE_KEY="---DEEPFENCE-API-KEY---" \
-    quay.io/deepfenceio/deepfence_agent_ce:2.3.1
+    quay.io/deepfenceio/deepfence_agent_ce:2.4.0
 ```
 
-Note: Image tag `quay.io/deepfenceio/deepfence_agent_ce:2.3.1-multiarch` is supported in amd64 and arm64/v8 architectures.
+Note: Image tag `quay.io/deepfenceio/deepfence_agent_ce:2.4.0-multiarch` is supported in amd64 and arm64/v8 architectures.
 
 On a Kubernetes platform, the sensors are installed using [helm chart](https://community.deepfence.io/threatmapper/docs/v2.3/sensors/kubernetes/)
 

deepfence_agent/Dockerfile.cloud-agent

@@ -14,7 +14,9 @@ ENV CHECKPOINT_DISABLE=true \
     DEEPFENCE_KEY="" \
     DF_ENABLE_CLOUD_NODE="true" \
     HOME_DIR="/home/deepfence" \
-    COMPLIANCE_MOD_PATH="/home/deepfence/steampipe"
+    STEAMPIPE_INSTALL_DIR="/home/deepfence/.steampipe" \
+    COMPLIANCE_MOD_PATH="/home/deepfence/steampipe" \
+    VERSION=$VERSION
 
 RUN apt-get update \
     && apt-get install -y --no-install-recommends bash ca-certificates nano logrotate sudo supervisor \

deepfence_agent/plugins/etc/run_shipper.sh

@@ -3,8 +3,8 @@
 export BATCH_SIZE=${1:-100}
 export TRUNCATE_SIZE=${2:-10}
 
-if [[ $DF_INSTALL_DIR == "/home/deepfence" ]]; then
-  exec /home/deepfence/bin/shipper --base-path="${DF_INSTALL_DIR:-/}" --truncate-size=$TRUNCATE_SIZE --routes=/home/deepfence/routes.yaml --batch-size=$BATCH_SIZE
+if [[ "${DF_INSTALL_DIR,,}" == *"/home/deepfence" ]]; then
+  exec $DF_INSTALL_DIR/bin/shipper --base-path="${DF_INSTALL_DIR:-/}" --truncate-size=$TRUNCATE_SIZE --routes=$DF_INSTALL_DIR/routes.yaml --batch-size=$BATCH_SIZE
 else
   exec $DF_INSTALL_DIR/home/deepfence/bin/shipper --base-path="${DF_INSTALL_DIR:-/}" --truncate-size=$TRUNCATE_SIZE --routes=$DF_INSTALL_DIR/home/deepfence/routes.yaml --batch-size=$BATCH_SIZE
 fi

deepfence_agent/start_cloud_agent.sh

@@ -48,7 +48,47 @@ launch_deepfenced() {
   tail -f $DF_INSTALL_DIR/var/log/supervisor/deepfenced*
 }
 
+check_persistent_volume() {
+  if [[ $DF_INSTALL_DIR != "/home/deepfence" ]]; then
+    echo "Copying postgres data to $DF_INSTALL_DIR"
+
+    clean_df_install_dir="false"
+    if [ -e "$DF_INSTALL_DIR"/.cloud_scanner_version ]; then
+      # Clean $DF_INSTALL_DIR if cloud scanner version is different
+      if [[ "$(cat "$DF_INSTALL_DIR"/.cloud_scanner_version)" != "$VERSION" ]]; then
+        clean_df_install_dir="true"
+      fi
+    else
+      clean_df_install_dir="true"
+    fi
+
+    if [[ $clean_df_install_dir == "true" ]]; then
+      rm -rf "${DF_INSTALL_DIR:?}"/* "${DF_INSTALL_DIR:?}"/.*
+      cp -R /home/deepfence/var "$DF_INSTALL_DIR"
+
+      mkdir -p "$STEAMPIPE_INSTALL_DIR"
+      cp -R /home/deepfence/.steampipe/config "$STEAMPIPE_INSTALL_DIR"/
+      cp -R /home/deepfence/.steampipe/plugins "$STEAMPIPE_INSTALL_DIR"/
+      cp -R /home/deepfence/.steampipe/db "$STEAMPIPE_INSTALL_DIR"/
+
+      # sudo -E -u deepfence /bin/bash
+      # export HOME=/home/deepfence
+      # steampipe service status
+    fi
+
+    cp -R /home/deepfence/bin /home/deepfence/routes.yaml /home/deepfence/run_shipper.sh /home/deepfence/supervisord.conf "$DF_INSTALL_DIR"
+    echo "$VERSION" > "$DF_INSTALL_DIR"/.cloud_scanner_version
+    chown -R deepfence: "$DF_INSTALL_DIR" "$STEAMPIPE_INSTALL_DIR"
+  fi
+
+  if [ ! -e "$DF_INSTALL_DIR"/.install_id ]; then
+    cat /proc/sys/kernel/random/uuid > "$DF_INSTALL_DIR"/.install_id
+    chown -R deepfence: "$DF_INSTALL_DIR"/.install_id
+  fi
+}
+
 main() {
+  check_persistent_volume
   launch_deepfenced
 }
 

deepfence_server/handler/cloud_node.go

@@ -56,14 +56,16 @@ func (h *Handler) RegisterCloudNodeAccountHandler(w http.ResponseWriter, r *http
 		}
 		orgNodeID := fmt.Sprintf("%s-%s-cloud-org", req.CloudProvider, orgAccountID)
 		orgAccountNode := map[string]interface{}{
-			"node_id":        orgNodeID,
-			"cloud_provider": model.PostureProviderOrgMap[req.CloudProvider],
-			"node_name":      orgAccountID,
-			"account_name":   req.AccountName,
-			"version":        req.Version,
-			"node_type":      req.CloudProvider,
+			"node_id":           orgNodeID,
+			"cloud_provider":    model.PostureProviderOrgMap[req.CloudProvider],
+			"node_name":         orgAccountID,
+			"account_name":      req.AccountName,
+			"version":           req.Version,
+			"node_type":         req.CloudProvider,
+			"installation_id":   req.InstallationID,
+			"persistent_volume": req.PersistentVolumeSupported,
 		}
-		err = model.UpsertCloudAccount(ctx, orgAccountNode, req.IsOrganizationDeployment, req.HostNodeID)
+		err = model.UpsertCloudAccount(ctx, orgAccountNode, req.IsOrganizationDeployment, req.HostNodeID, req.InitialRequest)
 		if err != nil {
 			log.Error().Msg(err.Error())
 			h.complianceError(w, err.Error())
@@ -78,16 +80,18 @@ func (h *Handler) RegisterCloudNodeAccountHandler(w http.ResponseWriter, r *http
 				return
 			}
 			monitoredNodes[i] = map[string]interface{}{
-				"node_id":         monitoredAccount.NodeID,
-				"cloud_provider":  req.CloudProvider,
-				"node_name":       monitoredAccount.AccountID,
-				"account_name":    monitoredAccount.AccountName,
-				"organization_id": orgNodeID,
-				"version":         req.Version,
-				"node_type":       req.CloudProvider,
+				"node_id":           monitoredAccount.NodeID,
+				"cloud_provider":    req.CloudProvider,
+				"node_name":         monitoredAccount.AccountID,
+				"account_name":      monitoredAccount.AccountName,
+				"organization_id":   orgNodeID,
+				"version":           req.Version,
+				"node_type":         req.CloudProvider,
+				"installation_id":   req.InstallationID,
+				"persistent_volume": req.PersistentVolumeSupported,
 			}
 		}
-		err = model.UpsertChildCloudAccounts(ctx, monitoredNodes, orgNodeID, req.HostNodeID)
+		err = model.UpsertChildCloudAccounts(ctx, monitoredNodes, orgNodeID, req.InstallationID, req.HostNodeID, req.InitialRequest)
 		if err != nil {
 			log.Error().Msgf("Error while upserting node: %+v", err)
 			h.complianceError(w, err.Error())
@@ -96,15 +100,17 @@ func (h *Handler) RegisterCloudNodeAccountHandler(w http.ResponseWriter, r *http
 	} else {
 		log.Debug().Msgf("Single account monitoring for node: %s", nodeID)
 		node := map[string]interface{}{
-			"node_id":        nodeID,
-			"cloud_provider": req.CloudProvider,
-			"node_name":      req.AccountID,
-			"account_name":   req.AccountName,
-			"version":        req.Version,
-			"node_type":      req.CloudProvider,
+			"node_id":           nodeID,
+			"cloud_provider":    req.CloudProvider,
+			"node_name":         req.AccountID,
+			"account_name":      req.AccountName,
+			"version":           req.Version,
+			"node_type":         req.CloudProvider,
+			"installation_id":   req.InstallationID,
+			"persistent_volume": req.PersistentVolumeSupported,
 		}
 		log.Debug().Msgf("Node for upsert: %+v", node)
-		err = model.UpsertCloudAccount(ctx, node, req.IsOrganizationDeployment, req.HostNodeID)
+		err = model.UpsertCloudAccount(ctx, node, req.IsOrganizationDeployment, req.HostNodeID, req.InitialRequest)
 		if err != nil {
 			log.Error().Msgf("Error while upserting node: %+v", err)
 			h.complianceError(w, err.Error())

deepfence_server/model/cloud_node.go

@@ -47,15 +47,18 @@ type CloudNodeMonitoredAccount struct {
 }
 
 type CloudNodeAccountRegisterReq struct {
-	NodeID                   string                      `json:"node_id" validate:"required" required:"true"`
-	AccountName              string                      `json:"account_name"`
-	HostNodeID               string                      `json:"host_node_id" validate:"required" required:"true"`
-	AccountID                string                      `json:"account_id" validate:"required" required:"true"`
-	CloudProvider            string                      `json:"cloud_provider" validate:"required,oneof=aws gcp azure" enum:"aws,gcp,azure" required:"true"`
-	IsOrganizationDeployment bool                        `json:"is_organization_deployment"`
-	MonitoredAccounts        []CloudNodeMonitoredAccount `json:"monitored_accounts"`
-	OrganizationAccountID    string                      `json:"organization_account_id"`
-	Version                  string                      `json:"version" validate:"required" required:"true"`
+	NodeID                    string                      `json:"node_id" validate:"required" required:"true"`
+	AccountName               string                      `json:"account_name"`
+	HostNodeID                string                      `json:"host_node_id" validate:"required" required:"true"`
+	AccountID                 string                      `json:"account_id" validate:"required" required:"true"`
+	CloudProvider             string                      `json:"cloud_provider" validate:"required,oneof=aws gcp azure" enum:"aws,gcp,azure" required:"true"`
+	IsOrganizationDeployment  bool                        `json:"is_organization_deployment"`
+	MonitoredAccounts         []CloudNodeMonitoredAccount `json:"monitored_accounts"`
+	OrganizationAccountID     string                      `json:"organization_account_id"`
+	Version                   string                      `json:"version" validate:"required" required:"true"`
+	PersistentVolumeSupported bool                        `json:"persistent_volume_supported"`
+	InstallationID            string                      `json:"installation_id" validate:"required" required:"true"`
+	InitialRequest            bool                        `json:"initial_request"`
 }
 
 type CloudNodeAccountsListReq struct {
@@ -84,6 +87,7 @@ type CloudNodeAccountInfo struct {
 	LastScanID           string           `json:"last_scan_id"`
 	LastScanStatus       string           `json:"last_scan_status"`
 	RefreshMessage       string           `json:"refresh_message"`
+	RefreshMetadata      string           `json:"refresh_metadata"`
 	RefreshStatus        string           `json:"refresh_status"`
 	RefreshStatusMap     map[string]int64 `json:"refresh_status_map"`
 	ScanStatusMap        map[string]int64 `json:"scan_status_map"`
@@ -207,7 +211,7 @@ type PostureProvider struct {
 	ResourceCount        int64   `json:"resource_count"`
 }
 
-func UpsertCloudAccount(ctx context.Context, nodeDetails map[string]interface{}, isOrganizationDeployment bool, hostNodeID string) error {
+func UpsertCloudAccount(ctx context.Context, nodeDetails map[string]interface{}, isOrganizationDeployment bool, hostNodeID string, initialRequest bool) error {
 
 	ctx, span := telemetry.NewSpan(ctx, "model", "upsert-cloud-compliance-node")
 	defer span.End()
@@ -227,9 +231,38 @@ func UpsertCloudAccount(ctx context.Context, nodeDetails map[string]interface{},
 	defer tx.Close(ctx)
 
 	var setRefreshStatusQuery string
+	var scheduleRefreshStatusQuery string
 	// Organization account node does not have refresh status. It is rather an aggregation of all child account statuses.
 	if !isOrganizationDeployment {
-		setRefreshStatusQuery = `ON CREATE SET n.refresh_status = '` + utils.ScanStatusStarting + `', n.refresh_message = ''`
+		setRefreshStatusQuery = `ON CREATE SET n.refresh_status = '` + utils.ScanStatusStarting + `', n.refresh_message = '', n.refresh_metadata = ''`
+
+		if initialRequest {
+			session2 := driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeRead})
+			defer session2.Close(ctx)
+
+			tx2, err := session2.BeginTransaction(ctx, neo4j.WithTxTimeout(30*time.Second))
+			if err != nil {
+				return err
+			}
+			defer tx2.Close(ctx)
+
+			res, err := tx2.Run(ctx,
+				`MATCH (n:CloudNode{node_id:$node_id}) RETURN coalesce(n.installation_id,''), coalesce(n.refresh_status,'')`,
+				map[string]interface{}{"node_id": nodeDetails["node_id"]})
+			if err != nil {
+				return err
+			}
+			rec, err := res.Single(ctx)
+			// ON CREATE will handle if Node does not exist
+			if err == nil {
+				if rec.Values[0].(string) != nodeDetails["installation_id"] {
+					scheduleRefreshStatusQuery = `, n.refresh_status = '` + utils.ScanStatusStarting + `', n.refresh_message = '', n.refresh_metadata = ''`
+				} else if rec.Values[1].(string) == utils.ScanStatusInProgress {
+					// Make it STARTING so that cloud scanner control request will pick it and continue
+					scheduleRefreshStatusQuery = `, n.refresh_status = '` + utils.ScanStatusStarting + `'`
+				}
+			}
+		}
 	}
 
 	if _, err = tx.Run(ctx, `
@@ -238,8 +271,8 @@ func UpsertCloudAccount(ctx context.Context, nodeDetails map[string]interface{},
 		MERGE (n:CloudNode{node_id:row.node_id})
 		`+setRefreshStatusQuery+`
 		MERGE (r) -[:HOSTS]-> (n)
-		SET n+= row, n.active = true, n.updated_at = TIMESTAMP(), n.version = row.version,
-		r.node_name=$host_node_id, r.active = true, r.agent_running=true, r.updated_at = TIMESTAMP()`,
+		SET n+= row, n.active = true, n.updated_at = TIMESTAMP(), n.version = row.version, r.node_name = $host_node_id, 
+	    r.active = true, r.agent_running = true, r.updated_at = TIMESTAMP()`+scheduleRefreshStatusQuery,
 		map[string]interface{}{
 			"param":        nodeDetails,
 			"host_node_id": hostNodeID,
@@ -250,7 +283,7 @@ func UpsertCloudAccount(ctx context.Context, nodeDetails map[string]interface{},
 	return tx.Commit(ctx)
 }
 
-func UpsertChildCloudAccounts(ctx context.Context, nodeDetails []map[string]interface{}, parentNodeID string, hostNodeID string) error {
+func UpsertChildCloudAccounts(ctx context.Context, nodeDetails []map[string]interface{}, parentNodeID string, installationID string, hostNodeID string, initialRequest bool) error {
 	ctx, span := telemetry.NewSpan(ctx, "model", "upsert-cloud-compliance-node")
 	defer span.End()
 
@@ -268,17 +301,59 @@ func UpsertChildCloudAccounts(ctx context.Context, nodeDetails []map[string]inte
 	}
 	defer tx.Close(ctx)
 
+	accountIDRefreshStatusMap := make(map[string]map[string]string)
+	if initialRequest {
+		session2 := driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeRead})
+		defer session2.Close(ctx)
+
+		tx2, err := session2.BeginTransaction(ctx, neo4j.WithTxTimeout(30*time.Second))
+		if err != nil {
+			return err
+		}
+		defer tx2.Close(ctx)
+
+		res, err := tx2.Run(ctx,
+			`MATCH (m:CloudNode{node_id:$node_id}) -[:IS_CHILD]-> (n:CloudNode)
+					RETURN coalesce(n.node_name,''), coalesce(n.installation_id,''), coalesce(n.refresh_status,'')`,
+			map[string]interface{}{"node_id": parentNodeID})
+		if err != nil {
+			return err
+		}
+		recs, err := res.Collect(ctx)
+		if err != nil {
+			return err
+		}
+		for _, rec := range recs {
+			accountID := rec.Values[0].(string)
+			if rec.Values[1].(string) != installationID {
+				accountIDRefreshStatusMap[accountID] = map[string]string{
+					"refresh_status": utils.ScanStatusStarting, "refresh_message": "", "refresh_metadata": ""}
+			} else if rec.Values[2].(string) == utils.ScanStatusInProgress {
+				// Make it STARTING so that cloud scanner control request will pick it and continue
+				accountIDRefreshStatusMap[accountID] = map[string]string{"refresh_status": utils.ScanStatusStarting}
+			}
+		}
+	}
+
+	for i, nodeDetail := range nodeDetails {
+		if refreshStatus, ok := accountIDRefreshStatusMap[nodeDetail["node_name"].(string)]; ok {
+			for k, v := range refreshStatus {
+				nodeDetails[i][k] = v
+			}
+		}
+	}
+
 	if _, err = tx.Run(ctx, `
 			MERGE (r:Node{node_id:$host_node_id, node_type: "cloud_agent"})
 			MERGE (m:CloudNode{node_id: $parent_node_id})
 			WITH r, m
 			UNWIND $param as row
 			MERGE (n:CloudNode{node_id:row.node_id})
-			ON CREATE SET n.refresh_status = '`+utils.ScanStatusStarting+`', n.refresh_message = ''
+			ON CREATE SET n.refresh_status = '`+utils.ScanStatusStarting+`', n.refresh_message = '', n.refresh_metadata = ''
 			MERGE (m) -[:IS_CHILD]-> (n)
 			MERGE (r) -[:HOSTS]-> (n)
-			SET n+= row, n.active = true, n.updated_at = TIMESTAMP(), n.version = row.version, 
-			r.active = true, r.agent_running=true, r.updated_at = TIMESTAMP()`,
+			SET n+= row, n.active = true, n.updated_at = TIMESTAMP(), n.version = row.version, r.active = true,
+			r.agent_running=true, r.updated_at = TIMESTAMP()`,
 		map[string]interface{}{
 			"param":          nodeDetails,
 			"parent_node_id": parentNodeID,
@@ -560,7 +635,7 @@ func (c *CloudAccountRefreshReq) SetCloudAccountRefresh(ctx context.Context) err
 	if _, err = tx.Run(ctx, `
 		UNWIND $batch as cloudNode
 		MATCH (m:CloudNode{node_id: cloudNode})
-		SET m.refresh_status = '`+utils.ScanStatusStarting+`', m.refresh_message = ''`,
+		SET m.refresh_status = '`+utils.ScanStatusStarting+`', m.refresh_message = '', m.refresh_metadata = ''`,
 		map[string]interface{}{
 			"batch": c.NodeIDs,
 		}); err != nil {