feat: use trivy filesystem scanner (#872)

Allow Trivy plugin to scan images cached on cluster nodes using
filesystem scanning as opposed to image scanning from a remote
registry. This requires scheduling scan jobs on the same node
as the scanned application.

pkg/kube/object.go

@@ -300,6 +300,8 @@ func (o *ObjectResolver) ObjectFromObjectRef(ctx context.Context, workload Objec
 }
 
 var ErrReplicaSetNotFound = errors.New("replicaset not found")
+var ErrNoRunningPods = errors.New("no active pods for controller")
+var ErrUnSupportedKind = errors.New("unsupported workload kind")
 
 // ReportOwner resolves the owner of a security report for the specified object.
 func (o *ObjectResolver) ReportOwner(ctx context.Context, obj client.Object) (client.Object, error) {
@@ -445,6 +447,61 @@ func (o *ObjectResolver) GetRelatedReplicasetName(ctx context.Context, object Ob
 	return "", fmt.Errorf("can only get related ReplicaSet for Deployment or Pod, not %q", string(object.Kind))
 }
 
+// GetNodeName will return the nodeName from one of the running pod of any kubernetes kind
+// if there are no running pods then ErrNoRunningPods will be returned.
+// if there are not active replicaset for deployment then ErrReplicaSetNotFound will be returned.
+func (o *ObjectResolver) GetNodeName(ctx context.Context, obj client.Object) (string, error) {
+	switch obj.(type) {
+	case *corev1.Pod:
+		return (obj.(*corev1.Pod)).Spec.NodeName, nil
+	case *appsv1.Deployment:
+		replicaSet, err := o.ReplicaSetByDeployment(ctx, obj.(*appsv1.Deployment))
+		if err != nil {
+			return "", err
+		}
+		pods, err := o.getActivePodsByLabelSelector(ctx, obj.GetNamespace(), replicaSet.Spec.Selector.MatchLabels)
+		if err != nil {
+			return "", err
+		}
+		return pods[0].Spec.NodeName, nil
+	case *appsv1.ReplicaSet:
+		pods, err := o.getActivePodsByLabelSelector(ctx, obj.GetNamespace(), obj.(*appsv1.ReplicaSet).Spec.Selector.MatchLabels)
+		if err != nil {
+			return "", err
+		}
+		return pods[0].Spec.NodeName, nil
+	case *corev1.ReplicationController:
+		pods, err := o.getActivePodsByLabelSelector(ctx, obj.GetNamespace(), obj.(*corev1.ReplicationController).Spec.Selector)
+		if err != nil {
+			return "", err
+		}
+		return pods[0].Spec.NodeName, nil
+	case *appsv1.StatefulSet:
+		pods, err := o.getActivePodsByLabelSelector(ctx, obj.GetNamespace(), obj.(*appsv1.StatefulSet).Spec.Selector.MatchLabels)
+		if err != nil {
+			return "", err
+		}
+		return pods[0].Spec.NodeName, nil
+	case *appsv1.DaemonSet:
+		pods, err := o.getActivePodsByLabelSelector(ctx, obj.GetNamespace(), obj.(*appsv1.DaemonSet).Spec.Selector.MatchLabels)
+		if err != nil {
+			return "", err
+		}
+		return pods[0].Spec.NodeName, nil
+	case *batchv1beta1.CronJob:
+		//Todo handle cronjob
+		return "", ErrUnSupportedKind
+	case *batchv1.Job:
+		pods, err := o.getActivePodsByLabelSelector(ctx, obj.GetNamespace(), obj.(*batchv1.Job).Spec.Selector.MatchLabels)
+		if err != nil {
+			return "", err
+		}
+		return pods[0].Spec.NodeName, nil
+	default:
+		return "", ErrUnSupportedKind
+	}
+}
+
 func (o *ObjectResolver) getActiveReplicaSetByDeployment(ctx context.Context, object ObjectRef) (string, error) {
 	deploy := &appsv1.Deployment{}
 	err := o.Client.Get(ctx, types.NamespacedName{Namespace: object.Namespace, Name: object.Name}, deploy)
@@ -505,3 +562,27 @@ func (o *ObjectResolver) IsActiveReplicaSet(ctx context.Context, workloadObj cli
 	}
 	return true, nil
 }
+
+func (o *ObjectResolver) GetPodsByLabelSelector(ctx context.Context, namespace string,
+	labelSelector labels.Set) ([]corev1.Pod, error) {
+	podList := &corev1.PodList{}
+	err := o.Client.List(ctx, podList, client.InNamespace(namespace),
+		client.MatchingLabelsSelector{Selector: labels.SelectorFromSet(labelSelector)})
+	if err != nil {
+		return podList.Items, fmt.Errorf("listing pods in namespace %s for labelselector %v: %w", namespace,
+			labelSelector, err)
+	}
+	return podList.Items, err
+}
+
+func (o *ObjectResolver) getActivePodsByLabelSelector(ctx context.Context, namespace string,
+	labelSelector labels.Set) ([]corev1.Pod, error) {
+	pods, err := o.GetPodsByLabelSelector(ctx, namespace, labelSelector)
+	if err != nil {
+		return pods, err
+	}
+	if len(pods) == 0 {
+		return pods, ErrNoRunningPods
+	}
+	return pods, nil
+}

pkg/operator/controller/vulnerabilityreport.go

@@ -4,6 +4,7 @@ import (
 	. "github.com/aquasecurity/starboard/pkg/operator/predicate"
 
 	"context"
+	"errors"
 	"fmt"
 	"reflect"
 
@@ -17,7 +18,7 @@ import (
 	batchv1 "k8s.io/api/batch/v1"
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
+	k8sapierror "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -92,7 +93,7 @@ func (r *VulnerabilityReportReconciler) reconcileWorkload(workloadKind kube.Kind
 		log.V(1).Info("Getting workload from cache")
 		workloadObj, err := r.ObjectFromObjectRef(ctx, workloadPartial)
 		if err != nil {
-			if errors.IsNotFound(err) {
+			if k8sapierror.IsNotFound(err) {
 				log.V(1).Info("Ignoring cached workload that must have been deleted")
 				return ctrl.Result{}, nil
 			}
@@ -205,7 +206,7 @@ func (r *VulnerabilityReportReconciler) hasActiveScanJob(ctx context.Context, ow
 	job := &batchv1.Job{}
 	err := r.Get(ctx, client.ObjectKey{Namespace: r.Config.Namespace, Name: jobName}, job)
 	if err != nil {
-		if errors.IsNotFound(err) {
+		if k8sapierror.IsNotFound(err) {
 			return false, nil, nil
 		}
 		return false, nil, fmt.Errorf("getting job from cache: %w", err)
@@ -217,6 +218,8 @@ func (r *VulnerabilityReportReconciler) hasActiveScanJob(ctx context.Context, ow
 }
 
 func (r *VulnerabilityReportReconciler) submitScanJob(ctx context.Context, owner client.Object) error {
+	log := r.Logger.WithValues("kind", owner.GetObjectKind().GroupVersionKind().Kind,
+		"name", owner.GetName(), "namespace", owner.GetNamespace())
 	credentials, err := r.CredentialsByWorkload(ctx, owner)
 	if err != nil {
 		return err
@@ -248,11 +251,20 @@ func (r *VulnerabilityReportReconciler) submitScanJob(ctx context.Context, owner
 		WithCredentials(credentials).
 		Get()
 
+	if err != nil {
+		if errors.Is(err, kube.ErrReplicaSetNotFound) || errors.Is(err, kube.ErrNoRunningPods) ||
+			errors.Is(err, kube.ErrUnSupportedKind) {
+			log.V(1).Info("ignoring vulnerability scan", "reason", err)
+			return nil
+		}
+		return fmt.Errorf("constructing scan job: %w", err)
+	}
+
 	for _, secret := range secrets {
 		secret.Namespace = r.PluginContext.GetNamespace()
 		err = r.Client.Create(ctx, secret)
 		if err != nil {
-			if errors.IsAlreadyExists(err) {
+			if k8sapierror.IsAlreadyExists(err) {
 				return nil
 			}
 			return fmt.Errorf("creating secret used by scan job failed: %s: %w", secret.Namespace+"/"+secret.Name, err)
@@ -261,7 +273,7 @@ func (r *VulnerabilityReportReconciler) submitScanJob(ctx context.Context, owner
 
 	err = r.Client.Create(ctx, scanJob)
 	if err != nil {
-		if errors.IsAlreadyExists(err) {
+		if k8sapierror.IsAlreadyExists(err) {
 			// TODO Delete secrets that were created in the previous step. Alternatively we can delete them on schedule.
 			return nil
 		}
@@ -289,7 +301,7 @@ func (r *VulnerabilityReportReconciler) reconcileJobs() reconcile.Func {
 		job := &batchv1.Job{}
 		err := r.Client.Get(ctx, req.NamespacedName, job)
 		if err != nil {
-			if errors.IsNotFound(err) {
+			if k8sapierror.IsNotFound(err) {
 				log.V(1).Info("Ignoring cached job that must have been deleted")
 				return ctrl.Result{}, nil
 			}
@@ -325,7 +337,7 @@ func (r *VulnerabilityReportReconciler) processCompleteScanJob(ctx context.Conte
 
 	owner, err := r.ObjectFromObjectRef(ctx, ownerRef)
 	if err != nil {
-		if errors.IsNotFound(err) {
+		if k8sapierror.IsNotFound(err) {
 			log.V(1).Info("Report owner must have been deleted", "owner", owner)
 			return r.deleteJob(ctx, job)
 		}
@@ -414,7 +426,7 @@ func (r *VulnerabilityReportReconciler) processFailedScanJob(ctx context.Context
 func (r *VulnerabilityReportReconciler) deleteJob(ctx context.Context, job *batchv1.Job) error {
 	err := r.Client.Delete(ctx, job, client.PropagationPolicy(metav1.DeletePropagationBackground))
 	if err != nil {
-		if errors.IsNotFound(err) {
+		if k8sapierror.IsNotFound(err) {
 			return nil
 		}
 		return fmt.Errorf("deleting job: %w", err)

pkg/plugin/factory.go

@@ -72,7 +72,7 @@ func (r *Resolver) GetVulnerabilityPlugin() (vulnerabilityreport.Plugin, starboa
 
 	switch scanner {
 	case starboard.Trivy:
-		return trivy.NewPlugin(ext.NewSystemClock(), ext.NewGoogleUUIDGenerator()), pluginContext, nil
+		return trivy.NewPlugin(ext.NewSystemClock(), ext.NewGoogleUUIDGenerator(), r.client), pluginContext, nil
 	case starboard.Aqua:
 		return aqua.NewPlugin(ext.NewGoogleUUIDGenerator(), r.buildInfo), pluginContext, nil
 	}