Add cluster image scanning to scan policies

app/models/clusters/cluster.rb

@@ -148,6 +148,7 @@ module Clusters
     scope :with_management_project, -> { where.not(management_project: nil) }
 
     scope :for_project_namespace, -> (namespace_id) { joins(:projects).where(projects: { namespace_id: namespace_id }) }
+    scope :with_name, -> (name) { where(name: name) }
 
     # with_application_prometheus scope is deprecated, and scheduled for removal
     # in %14.0. See https://gitlab.com/groups/gitlab-org/-/epics/4280

app/models/clusters/platforms/kubernetes.rb

@@ -137,6 +137,14 @@ module Clusters
         kubeclient.patch_ingress(ingress.name, data, namespace)
       end
 
+      def kubeconfig(namespace)
+        to_kubeconfig(
+          url: api_url,
+          namespace: namespace,
+          token: token,
+          ca_pem: ca_pem)
+      end
+
       private
 
       def default_namespace(project, environment_name:)
@@ -154,14 +162,6 @@ module Clusters
         ).execute
       end
 
-      def kubeconfig(namespace)
-        to_kubeconfig(
-          url: api_url,
-          namespace: namespace,
-          token: token,
-          ca_pem: ca_pem)
-      end
-
       def read_pods(namespace)
         kubeclient.get_pods(namespace: namespace).as_json
       rescue Kubeclient::ResourceNotFoundError

doc/user/application_security/policies/index.md

@@ -121,6 +121,16 @@ This rule enforces the defined actions and schedules a scan on the provided date
 | `type`     | `string` | `schedule` | The rule's type. |
 | `branches` | `array` of `string` | `*` or the branch's name | The branch the given policy applies to (supports wildcard). |
 | `cadence`  | `string` | CRON expression (for example, `0 0 * * *`) | A whitespace-separated string containing five fields that represents the scheduled time. |
+| `clusters` | `object` | | The cluster where the given policy will enforce running selected scans (only for `container_scanning`/`cluster_image_scanning` scans). The key of the object is the name of the Kubernetes cluster configured for your project in GitLab. In the optionally provided value of the object, you can precisely select Kubernetes resources that will be scanned. |
+
+#### `cluster` schema
+
+| Field        | Type                | Possible values          | Description |
+|--------------|---------------------|--------------------------|-------------|
+| `containers` | `array` of `string` | | The container name that will be scanned (only the first value is currently supported). |
+| `resources`  | `array` of `string` | | The resource name that will be scanned (only the first value is currently supported). |
+| `namespaces` | `array` of `string` | | The namespace that will be scanned (only the first value is currently supported). |
+| `kinds`      | `array` of `string` | `deployment`/`daemonset` | The resource kind that should be scanned (only the first value is currently supported). |
 
 ### `scan` action type
 
@@ -149,6 +159,9 @@ Note the following:
 - A secret detection scan runs in `normal` mode when executed as part of a pipeline, and in
   [`historic`](../secret_detection/index.md#full-history-secret-scan)
   mode when executed as part of a scheduled scan.
+- A container scanning and cluster image scanning scans configured for the `pipeline` rule type will ignore the cluster defined in the `clusters` object.
+  They will use predefined CI/CD variables defined for your project. Cluster selection with the `clusters` object is supported for the `schedule` rule type.
+  Cluster with name provided in `clusters` object must be created and configured for the project. To be able to successfully perform the `container_scanning`/`cluster_image_scanning` scans for the cluster you must follow instructions for the [Cluster Image Scanning feature](../cluster_image_scanning/index.md#prerequisites).
 
 Here's an example:
 
@@ -179,8 +192,8 @@ scan_execution_policy:
     scanner_profile: Scanner Profile C
     site_profile: Site Profile D
   - scan: secret_detection
-- name: Enforce Secret Detection in every default branch pipeline
-  description: This policy enforces pipeline configuration to have a job with Secret Detection scan for the default branch
+- name: Enforce Secret Detection and Container Scanning in every default branch pipeline
+  description: This policy enforces pipeline configuration to have a job with Secret Detection and Container Scanning scans for the default branch
   enabled: true
   rules:
   - type: pipeline
@@ -188,7 +201,25 @@ scan_execution_policy:
     - main
   actions:
   - scan: secret_detection
-```
+  - scan: container_scanning
+- name: Enforce Cluster Image Scanning on production-cluster every 24h
+  description: This policy enforces Cluster Image Scanning scan to run every 24 hours
+  enabled: true
+  rules:
+  - type: schedule
+    cadence: '15 3 * * *'
+    clusters:
+      production-cluster:
+        containers:
+        - database
+        resources:
+        - production-application
+        namespaces:
+        - production-namespace
+        kinds:
+        - deployment
+  actions:
+  - scan: cluster_image_scanning
 
 In this example:
 
@@ -196,7 +227,9 @@ In this example:
   `release/v1.2.1`), DAST scans run with `Scanner Profile A` and `Site Profile B`.
 - DAST and secret detection scans run every 10 minutes. The DAST scan runs with `Scanner Profile C`
   and `Site Profile D`.
-- Secret detection scans run for every pipeline executed on the `main` branch.
+- Secret detection and container scanning scans run for every pipeline executed on the `main` branch.
+- Cluster Image Scanning scan runs every 24h. The scan runs on the `production-cluster` cluster and fetches vulnerabilities
+  from the container with the name `database` configured for deployment with the name `production-application` in the `production-namepsace` namespace.
 
 ## Security Policy project selection
 

ee/app/models/security/orchestration_policy_configuration.rb

@@ -16,7 +16,7 @@ module Security
       schedule: 'schedule'
     }.freeze
 
-    SCAN_TYPES = %w[dast secret_detection].freeze
+    SCAN_TYPES = %w[dast secret_detection cluster_image_scanning container_scanning].freeze
     ON_DEMAND_SCANS = %w[dast].freeze
     AVAILABLE_POLICY_TYPES = %i{scan_execution_policy}.freeze
 

ee/app/models/security/orchestration_policy_rule_schedule.rb

@@ -33,14 +33,24 @@ module Security
     end
 
     def applicable_branches
-      configured_branches = policy&.dig(:rules, rule_index, :branches)
-      return [] if configured_branches.blank?
+      strong_memoize(:applicable_branches) do
+        configured_branches = policy&.dig(:rules, rule_index, :branches)
+        next [] if configured_branches.blank?
 
-      branch_names = security_orchestration_policy_configuration.project.repository.branches
+        branch_names = security_orchestration_policy_configuration.project.repository.branches
 
-      configured_branches
-        .flat_map { |pattern| RefMatcher.new(pattern).matching(branch_names).map(&:name) }
-        .uniq
+        configured_branches
+          .flat_map { |pattern| RefMatcher.new(pattern).matching(branch_names).map(&:name) }
+          .uniq
+      end
+    end
+
+    def applicable_clusters
+      policy&.dig(:rules, rule_index, :clusters)
+    end
+
+    def for_cluster?
+      applicable_clusters.present?
     end
 
     private

ee/app/services/security/security_orchestration_policies/ci_configuration_service.rb

@@ -4,13 +4,17 @@ module Security
   module SecurityOrchestrationPolicies
     class CiConfigurationService
       SCAN_TEMPLATES = {
-        'secret_detection' => 'Jobs/Secret-Detection'
+        'secret_detection' => 'Jobs/Secret-Detection',
+        'cluster_image_scanning' => 'Security/Cluster-Image-Scanning',
+        'container_scanning' => 'Security/Container-Scanning'
       }.freeze
 
       def execute(action, ci_variables)
         case action[:scan]
         when 'secret_detection'
           secret_detection_configuration(ci_variables)
+        when 'container_scanning', 'cluster_image_scanning'
+          scan_configuration(action[:scan], ci_variables)
         else
           error_script('Invalid Scan type')
         end
@@ -32,6 +36,14 @@ module Security
           .except(:extends)
       end
 
+      def scan_configuration(template, ci_variables)
+        ci_configuration = scan_template(template)
+
+        ci_configuration[template.to_sym]
+          .deep_merge(variables: ci_configuration[:variables].deep_merge(ci_variables).compact)
+          .except(:rules)
+      end
+
       def error_script(error_message)
         {
           'script' => "echo \"Error during Scan execution: #{error_message}\" && false",

ee/app/services/security/security_orchestration_policies/cluster_image_scanning_ci_variables_service.rb

+# frozen_string_literal: true
+
+module Security
+  module SecurityOrchestrationPolicies
+    class ClusterImageScanningCiVariablesService < ::BaseProjectService
+      SCAN_VARIABLES = {
+        'CLUSTER_IMAGE_SCANNING_DISABLED' => nil
+      }.freeze
+
+      def execute(action)
+        # TODO: Add support for multiple clusters. As for now, we only support the first cluster defined in the policy.
+        cluster_name, resource_filters = action[:clusters]&.first
+
+        [scan_variables(resource_filters), hidden_variable_attributes(cluster_name.to_s)]
+      end
+
+      private
+
+      def scan_variables(resource_filters)
+        return SCAN_VARIABLES if resource_filters.blank?
+
+        SCAN_VARIABLES.merge({
+          'CIS_CONTAINER_NAME' => resource_filter_value(resource_filters[:containers]),
+          'CIS_RESOURCE_NAME' => resource_filter_value(resource_filters[:resources]),
+          'CIS_RESOURCE_NAMESPACE' => resource_filter_value(resource_filters[:namespaces]),
+          'CIS_RESOURCE_KIND' => resource_filter_value(resource_filters[:kinds])
+        }.compact)
+      end
+
+      def hidden_variable_attributes(cluster_name)
+        cluster = project.all_clusters.enabled.with_name(cluster_name).first
+        return {} if cluster.blank?
+
+        deployment_platform = cluster.platform_kubernetes
+        return {} if deployment_platform.blank?
+
+        kubeconfig = deployment_platform.kubeconfig(Clusters::Kubernetes::GITLAB_SERVICE_ACCOUNT_NAMESPACE)
+
+        [{ key: 'CIS_KUBECONFIG', value: kubeconfig, variable_type: :file }]
+      end
+
+      def resource_filter_value(filter_values)
+        # TODO: Add support for multiple values in filter (modify analyzer to support that).
+        return if filter_values.blank?
+
+        filter_values
+          .compact
+          .first
+      end
+    end
+  end
+end

ee/app/services/security/security_orchestration_policies/create_pipeline_service.rb

@@ -7,12 +7,16 @@ module Security
         'secret_detection' => {
           'SECRET_DETECTION_HISTORIC_SCAN' => 'true',
           'SECRET_DETECTION_DISABLED' => nil
+        },
+        'container_scanning' => {
+          'CONTAINER_SCANNING_DISABLED' => nil
         }
       }.freeze
 
       def execute
         service = Ci::CreatePipelineService.new(project, current_user, ref: params[:branch])
-        result = service.execute(:security_orchestration_policy, content: ci_configuration.to_yaml)
+        ci_content, ci_hidden_variables = ci_configuration
+        result = service.execute(:security_orchestration_policy, content: ci_content.to_yaml, variables_attributes: ci_hidden_variables)
 
         pipeline = result.payload
         if pipeline.created_successfully?
@@ -25,9 +29,19 @@ module Security
       private
 
       def ci_configuration
-        ci_content = ::Security::SecurityOrchestrationPolicies::CiConfigurationService.new.execute(action, SCAN_VARIABLES[scan_type])
+        ci_variables, ci_hidden_variables = scan_variables
+        ci_content = ::Security::SecurityOrchestrationPolicies::CiConfigurationService.new.execute(action, ci_variables)
 
-        { "#{scan_type}" => ci_content }
+        [{ "#{scan_type}" => ci_content }, ci_hidden_variables]
+      end
+
+      def scan_variables
+        case scan_type.to_sym
+        when :cluster_image_scanning
+          ClusterImageScanningCiVariablesService.new(project: project).execute(action)
+        else
+          [SCAN_VARIABLES[scan_type].to_h, []]
+        end
       end
 
       def action

ee/app/services/security/security_orchestration_policies/rule_schedule_service.rb

@@ -7,7 +7,7 @@ module Security
         schedule.schedule_next_run!
 
         branches = schedule.applicable_branches
-        actions_for(schedule).each { |action| process_action(action, branches) }
+        actions_for(schedule).each { |action| process_action(action, schedule, branches) }
       end
 
       private
@@ -19,9 +19,11 @@ module Security
         policy[:actions]
       end
 
-      def process_action(action, branches)
-        case action[:scan]
+      def process_action(action, schedule, branches)
+        case action[:scan].to_s
         when 'secret_detection' then schedule_scan(action, branches)
+        when 'container_scanning' then schedule_container_scanning_scan(action, schedule, branches)
+        when 'cluster_image_scanning' then schedule_cluster_image_scanning_scan(action, schedule)
         when 'dast' then schedule_dast_on_demand_scan(action, branches)
         end
       end
@@ -34,6 +36,18 @@ module Security
         end
       end
 
+      def schedule_container_scanning_scan(action, schedule, branches)
+        if schedule.for_cluster?
+          schedule_cluster_image_scanning_scan(action, schedule)
+        else
+          schedule_scan(action, branches)
+        end
+      end
+
+      def schedule_cluster_image_scanning_scan(action, schedule)
+        schedule_scan(action.merge(scan: 'cluster_image_scanning', clusters: schedule.applicable_clusters), [container.default_branch_or_main])
+      end
+
       def schedule_dast_on_demand_scan(action, branches)
         dast_site_profile = find_dast_site_profile(container, action[:site_profile])
         dast_scanner_profile = find_dast_scanner_profile(container, action[:scanner_profile])

ee/app/services/security/security_orchestration_policies/scan_pipeline_service.rb

@@ -7,6 +7,9 @@ module Security
         secret_detection: {
           'SECRET_DETECTION_HISTORIC_SCAN' => 'false',
           'SECRET_DETECTION_DISABLED' => nil
+        },
+        container_scanning: {
+          'CONTAINER_SCANNING_DISABLED' => nil
         }
       }.freeze
 
@@ -29,7 +32,16 @@ module Security
       end
 
       def scan_configuration(action)
-        ::Security::SecurityOrchestrationPolicies::CiConfigurationService.new.execute(action, SCAN_VARIABLES[action[:scan].to_sym])
+        ::Security::SecurityOrchestrationPolicies::CiConfigurationService.new.execute(action, scan_variables(action))
+      end
+
+      def scan_variables(action)
+        case action[:scan].to_sym
+        when :cluster_image_scan
+          ClusterImageScanningCiVariablesService.new(project: project).execute(action).first
+        else
+          SCAN_VARIABLES[action[:scan].to_sym].to_h
+        end
       end
     end
   end

ee/app/validators/json_schemas/security_orchestration_policy.json

 {
   "required": [
-      "scan_execution_policy"
+    "scan_execution_policy"
   ],
   "type": "object",
   "properties": {
-      "scan_execution_policy": {
-          "type": "array",
-          "additionalItems": false,
-          "items": {
-              "maxItems": 5,
+    "scan_execution_policy": {
+      "type": "array",
+      "additionalItems": false,
+      "items": {
+        "maxItems": 5,
+        "required": [
+          "name",
+          "enabled",
+          "rules",
+          "actions"
+        ],
+        "type": "object",
+        "properties": {
+          "name": {
+            "minLength": 1,
+            "type": "string"
+          },
+          "description": {
+            "type": "string"
+          },
+          "enabled": {
+            "type": "boolean"
+          },
+          "rules": {
+            "type": "array",
+            "additionalItems": false,
+            "items": {
+              "type": "object",
+              "required": [],
+              "properties": {
+                "type": {
+                  "enum": [
+                    "pipeline",
+                    "schedule"
+                  ],
+                  "type": "string"
+                },
+                "branches": {
+                  "type": "array",
+                  "additionalItems": false,
+                  "items": {
+                    "minLength": 1,
+                    "type": "string"
+                  }
+                },
+                "cadence": {
+                  "type": "string"
+                },
+                "clusters": {
+                  "type": "object",
+                  "minProperties": 1,
+                  "maxProperties": 1,
+                  "patternProperties": {
+                    "\\A[a-z0-9]([-a-z0-9]*[a-z0-9])?\\z": {
+                      "type": "object",
+                      "properties": {
+                        "namespaces": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        },
+                        "resources": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        },
+                        "containers": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        },
+                        "kinds": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        }
+                      }
+                    }
+                  }
+                }
+              },
+              "if": {
+                "properties": {
+                  "type": {
+                    "const": "schedule"
+                  }
+                }
+              },
+              "then": {
+                "required": [
+                  "cadence"
+                ]
+              },
+              "additionalProperties": false
+            }
+          },
+          "actions": {
+            "type": "array",
+            "additionalItems": false,
+            "items": {
               "required": [
-                  "name",
-                  "enabled",
-                  "rules",
-                  "actions"
+                "scan"
               ],
               "type": "object",
               "properties": {
-                  "name": {
-                      "minLength": 1,
-                      "type": "string"
-                  },
-                  "description": {
-                      "type": "string"
+                "scan": {
+                  "enum": [
+                    "dast",
+                    "secret_detection",
+                    "container_scanning",
+                    "cluster_image_scanning"
+                  ],
+                  "type": "string"
+                },
+                "scanner_profile": {
+                  "type": "string"
+                },
+                "site_profile": {
+                  "type": [
+                    "string",
+                    "null"
+                  ]
+                }
+              },
+              "allOf": [
+                {
+                  "if": {
+                    "properties": {
+                      "scan": {
+                        "const": "dast"
+                      }
+                    }
                   },
-                  "enabled": {
-                      "type": "boolean"
+                  "then": {
+                    "required": [
+                      "site_profile"
+                    ],
+                    "maxProperties": 3
+                  }
+                },
+                {
+                  "if": {
+                    "properties": {
+                      "scan": {
+                        "const": "secret_detection"
+                      }
+                    }
                   },
-                  "rules": {
-                      "type": "array",
-                      "additionalItems": false,
-                      "items": {
-                          "type": "object",
-                          "required": [],
-                          "properties": {
-                              "type": {
-                                  "enum": [
-                                      "pipeline",
-                                      "schedule"
-                                  ],
-                                  "type": "string"
-                              },
-                              "branches": {
-                                  "type": "array",
-                                  "additionalItems": false,
-                                  "items": {
-                                      "minLength": 1,
-                                      "type": "string"
-                                  }
-                              },
-                             "cadence": {
-                                  "type": "string"
-                             }
-                          },
-                          "if": {
-                                "properties": {
-                                    "type": {
-                                        "const": "schedule"
-                                    }
-                                }
-                          },
-                          "then": {
-                                "required": ["cadence"]
-                          },
-                          "additionalProperties": false
+                  "then": {
+                    "maxProperties": 1
+                  }
+                },
+                {
+                  "if": {
+                    "properties": {
+                      "scan": {
+                        "const": "cluster_image_scanning"
                       }
+                    }
                   },
-                  "actions": {
-                      "type": "array",
-                      "additionalItems": false,
-                      "items": {
-                          "required": [
-                              "scan"
-                          ],
-                          "type": "object",
-                          "properties": {
-                              "scan": {
-                                  "enum": [
-                                      "dast",
-                                      "secret_detection"
-                                  ],
-                                  "type": "string"
-                              },
-                              "scanner_profile": {
-                                  "type": "string"
-                              },
-                              "site_profile": {
-                                  "type": [
-                                      "string",
-                                      "null"
-                                  ]
-                              }
-                          },
-                          "allOf": [
-                            {
-                              "if": {
-                                "properties": {
-                                  "scan": {
-                                    "const": "dast"
-                                  }
-                                }
-                              },
-                              "then": {
-                                "required": [
-                                  "site_profile"
-                                ],
-                                "maxProperties": 3
-                              }
-                            },
-                            {
-                              "if": {
-                                "properties": {
-                                  "scan": {
-                                    "const": "secret_detection"
-                                  }
-                                }
-                              },
-                              "then": {
-                                "maxProperties": 1
-                              }
-                            }
-                          ],
-                          "additionalProperties": false
+                  "then": {
+                    "maxProperties": 1
+                  }
+                },
+                {
+                  "if": {
+                    "properties": {
+                      "scan": {
+                        "const": "container_scanning"
                       }
+                    }
+                  },
+                  "then": {
+                    "maxProperties": 1
                   }
-              },
+                }
+              ],
               "additionalProperties": false
+            }
           }
+        },
+        "additionalProperties": false
       }
+    }
   },
   "additionalProperties": false
 }

ee/spec/models/security/orchestration_policy_rule_schedule_spec.rb

@@ -108,7 +108,7 @@ RSpec.describe Security::OrchestrationPolicyRuleSchedule do
   describe '#applicable_branches' do
     let_it_be_with_reload(:project) { create(:project, :repository) }
     let_it_be_with_reload(:policy_configuration) { create(:security_orchestration_policy_configuration, project: project) }
-    let_it_be_with_reload(:rule_schedule) do
+    let_it_be_with_refind(:rule_schedule) do
       create(:security_orchestration_policy_rule_schedule, security_orchestration_policy_configuration: policy_configuration)
     end
 
@@ -182,6 +182,90 @@ RSpec.describe Security::OrchestrationPolicyRuleSchedule do
     end
   end
 
+  describe '#applicable_clusters' do
+    let_it_be_with_reload(:project) { create(:project, :repository) }
+    let_it_be_with_reload(:policy_configuration) { create(:security_orchestration_policy_configuration, project: project) }
+
+    let!(:rule_schedule) do
+      create(:security_orchestration_policy_rule_schedule, security_orchestration_policy_configuration: policy_configuration, rule_index: rule_index)
+    end
+
+    let(:clusters) do
+      {
+        'production-cluster': {
+          namespaces: ['production-namespace']
+        }
+      }
+    end
+
+    let(:policy) do
+      build(:scan_execution_policy, rules: [
+        { type: 'schedule', clusters: clusters, cadence: '*/20 * * * *' },
+        { type: 'pipeline', branches: ['main'] }
+      ])
+    end
+
+    subject { rule_schedule.applicable_clusters }
+
+    before do
+      allow(rule_schedule).to receive(:policy).and_return(policy)
+    end
+
+    context 'when applicable rule contains clusters configuration' do
+      let(:rule_index) { 0 }
+
+      it { is_expected.to eq(clusters) }
+    end
+
+    context 'when applicable rule does not contain clusters configuration' do
+      let(:rule_index) { 1 }
+
+      it { is_expected.to be_nil }
+    end
+  end
+
+  describe '#for_cluster?' do
+    let_it_be_with_reload(:project) { create(:project, :repository) }
+    let_it_be_with_reload(:policy_configuration) { create(:security_orchestration_policy_configuration, project: project) }
+
+    let!(:rule_schedule) do
+      create(:security_orchestration_policy_rule_schedule, security_orchestration_policy_configuration: policy_configuration, rule_index: rule_index)
+    end
+
+    let(:clusters) do
+      {
+        'production-cluster': {
+          namespaces: ['production-namespace']
+        }
+      }
+    end
+
+    let(:policy) do
+      build(:scan_execution_policy, rules: [
+        { type: 'schedule', clusters: clusters, cadence: '*/20 * * * *' },
+        { type: 'pipeline', branches: ['main'] }
+      ])
+    end
+
+    subject { rule_schedule.for_cluster? }
+
+    before do
+      allow(rule_schedule).to receive(:policy).and_return(policy)
+    end
+
+    context 'when applicable rule contains clusters configuration' do
+      let(:rule_index) { 0 }
+
+      it { is_expected.to eq(true) }
+    end
+
+    context 'when applicable rule does not contain clusters configuration' do
+      let(:rule_index) { 1 }
+
+      it { is_expected.to eq(false) }
+    end
+  end
+
   describe '#set_next_run_at' do
     it_behaves_like 'handles set_next_run_at' do
       let(:schedule) { create(:security_orchestration_policy_rule_schedule, cron: '*/1 * * * *') }

ee/spec/services/security/security_orchestration_policies/ci_configuration_service_spec.rb

@@ -19,8 +19,8 @@ RSpec.describe Security::SecurityOrchestrationPolicies::CiConfigurationService d
       end
     end
 
-    context 'when scan type is secret_detection' do
-      context 'when action is valid' do
+    context 'when action is valid' do
+      context 'when scan type is secret_detection' do
         let_it_be(:action) { { scan: 'secret_detection' } }
         let_it_be(:template_name) { 'Jobs/Secret-Detection' }
 
@@ -58,18 +58,73 @@ RSpec.describe Security::SecurityOrchestrationPolicies::CiConfigurationService d
         end
       end
 
-      context 'when action is invalid' do
-        let_it_be(:action) { { scan: 'invalid_type' } }
+      context 'when scan type is cluster_image_scanning' do
+        let_it_be(:action) { { scan: 'cluster_image_scanning' } }
+        let_it_be(:template_name) { 'Security/Cluster-Image-Scanning' }
+        let_it_be(:ci_variables) { {} }
 
-        it 'returns prepared CI configuration with error script' do
+        it_behaves_like 'with template name for scan type'
+
+        it 'returns prepared CI configuration for Cluster Image Scanning' do
+          expected_configuration = {
+            image: '$CIS_ANALYZER_IMAGE',
+            stage: 'test',
+            allow_failure: true,
+            artifacts: {
+              reports: { cluster_image_scanning: 'gl-cluster-image-scanning-report.json' },
+              paths: ['gl-cluster-image-scanning-report.json']
+            },
+            dependencies: [],
+            script: ['/analyzer run'],
+            variables: {
+              CIS_ANALYZER_IMAGE: 'registry.gitlab.com/gitlab-org/security-products/analyzers/cluster-image-scanning:0'
+            }
+          }
+
+          expect(subject.deep_symbolize_keys).to eq(expected_configuration)
+        end
+      end
+
+      context 'when scan type is container_scanning' do
+        let_it_be(:action) { { scan: 'container_scanning' } }
+        let_it_be(:template_name) { 'Security/Container-Scanning' }
+        let_it_be(:ci_variables) { {} }
+
+        it_behaves_like 'with template name for scan type'
+
+        it 'returns prepared CI configuration for Container Scanning' do
           expected_configuration = {
-            'allow_failure' => true,
-            'script' => "echo \"Error during Scan execution: Invalid Scan type\" && false"
+            image: '$CS_ANALYZER_IMAGE',
+            stage: 'test',
+            allow_failure: true,
+            artifacts: {
+              reports: { container_scanning: 'gl-container-scanning-report.json' },
+              paths: ['gl-container-scanning-report.json']
+            },
+            dependencies: [],
+            script: ['gtcs scan'],
+            variables: {
+              CS_ANALYZER_IMAGE: 'registry.gitlab.com/security-products/container-scanning:4',
+              GIT_STRATEGY: 'none'
+            }
           }
 
-          expect(subject).to eq(expected_configuration)
+          expect(subject.deep_symbolize_keys).to eq(expected_configuration)
         end
       end
     end
+
+    context 'when action is invalid' do
+      let_it_be(:action) { { scan: 'invalid_type' } }
+
+      it 'returns prepared CI configuration with error script' do
+        expected_configuration = {
+          'allow_failure' => true,
+          'script' => "echo \"Error during Scan execution: Invalid Scan type\" && false"
+        }
+
+        expect(subject).to eq(expected_configuration)
+      end
+    end
   end
 end

ee/spec/services/security/security_orchestration_policies/cluster_image_scanning_ci_variables_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::SecurityOrchestrationPolicies::ClusterImageScanningCiVariablesService do
+  describe '#execute' do
+    let_it_be_with_reload(:project) { create(:project) }
+    let(:service) { described_class.new(project: project) }
+
+    let_it_be(:ci_variables) do
+      { 'CLUSTER_IMAGE_SCANNING_DISABLED' => nil }
+    end
+
+    let(:requested_cluster) { 'production' }
+    let(:action) do
+      {
+        clusters: {
+          requested_cluster => {
+            containers: %w[nginx falco],
+            resources: %w[nginx-www nginx-admin],
+            namespaces: %w[gitlab-production cluster-apps],
+            kinds: %w[deployment daemonset]
+          },
+          staging: {
+            containers: %w[falco],
+            resources: %w[nginx-admin],
+            namespaces: %w[cluster-apps],
+            kinds: %w[daemonset]
+          }
+        }
+      }
+    end
+
+    subject(:generated_variables) { service.execute(action) }
+
+    shared_examples 'with cluster image scanning resource filters' do
+      it 'generates CI variable values with first value for each resource filter' do
+        ci_variables, _ = generated_variables
+
+        expect(ci_variables).to eq(
+          'CLUSTER_IMAGE_SCANNING_DISABLED' => nil,
+          'CIS_CONTAINER_NAME' => 'nginx',
+          'CIS_RESOURCE_NAME' => 'nginx-www',
+          'CIS_RESOURCE_NAMESPACE' => 'gitlab-production',
+          'CIS_RESOURCE_KIND' => 'deployment'
+        )
+      end
+    end
+
+    shared_examples 'without variable attributes' do
+      it 'does not generate variable attributes for pipeline' do
+        _, variable_attributes = generated_variables
+
+        expect(variable_attributes).to eq({})
+      end
+    end
+
+    context 'when cluster was not found' do
+      it_behaves_like 'with cluster image scanning resource filters'
+      it_behaves_like 'without variable attributes'
+    end
+
+    context 'when cluster was found' do
+      let_it_be(:cluster) { create(:cluster, :with_environments, :provided_by_user, name: 'production') }
+      let_it_be(:project) { cluster.kubernetes_namespaces.first.project }
+
+      context 'when cluster with requested name does not exist' do
+        let(:requested_cluster) { 'gilab-managed-apps' }
+
+        it_behaves_like 'with cluster image scanning resource filters'
+        it_behaves_like 'without variable attributes'
+      end
+
+      context 'when cluster with requested name exists' do
+        it_behaves_like 'with cluster image scanning resource filters'
+
+        it 'generates variable attributes for pipeline with CIS_KUBECONFIG variable' do
+          _, variable_attributes = generated_variables
+
+          expect(variable_attributes).to include(hash_including(key: 'CIS_KUBECONFIG', variable_type: :file))
+        end
+      end
+    end
+  end
+end

ee/spec/services/security/security_orchestration_policies/create_pipeline_service_spec.rb

@@ -4,11 +4,14 @@ require 'spec_helper'
 
 RSpec.describe Security::SecurityOrchestrationPolicies::CreatePipelineService do
   describe '#execute' do
-    let_it_be(:project) { create(:project, :repository) }
+    let_it_be_with_reload(:project) { create(:project, :repository) }
     let_it_be(:current_user) { project.owner }
     let_it_be(:branch) { project.default_branch }
-    let_it_be(:action) { { scan: 'secret_detection' } }
-    let_it_be(:service) do
+
+    let(:action) { { scan: 'secret_detection' } }
+    let(:scan_type) { action[:scan] }
+
+    let(:service) do
       described_class.new(project: project, current_user: current_user, params: {
         action: action, branch: branch
       })
@@ -16,39 +19,43 @@ RSpec.describe Security::SecurityOrchestrationPolicies::CreatePipelineService do
 
     subject { service.execute }
 
-    context 'when scan type is valid' do
-      let(:status) { subject[:status] }
-      let(:pipeline) { subject[:payload] }
-      let(:message) { subject[:message] }
+    shared_examples 'valid security orchestration policy action' do
+      it 'returns a success status' do
+        expect(status).to eq(:success)
+      end
 
-      context 'when action is valid' do
-        it 'returns a success status' do
-          expect(status).to eq(:success)
-        end
+      it 'returns a pipeline' do
+        expect(pipeline).to be_a(Ci::Pipeline)
+      end
 
-        it 'returns a pipeline' do
-          expect(pipeline).to be_a(Ci::Pipeline)
-        end
+      it 'creates a pipeline' do
+        expect { subject }.to change(Ci::Pipeline, :count).by(1)
+      end
 
-        it 'creates a pipeline' do
-          expect { subject }.to change(Ci::Pipeline, :count).by(1)
-        end
+      it 'sets the pipeline ref to the branch' do
+        expect(pipeline.ref).to eq(branch)
+      end
 
-        it 'sets the pipeline ref to the branch' do
-          expect(pipeline.ref).to eq(branch)
-        end
+      it 'sets the source to security_orchestration_policy' do
+        expect(pipeline.source).to eq('security_orchestration_policy')
+      end
 
-        it 'sets the source to security_orchestration_policy' do
-          expect(pipeline.source).to eq('security_orchestration_policy')
-        end
+      it 'creates a stage' do
+        expect { subject }.to change(Ci::Stage, :count).by(1)
+      end
 
-        it 'creates a stage' do
-          expect { subject }.to change(Ci::Stage, :count).by(1)
-        end
+      it 'creates a build' do
+        expect { subject }.to change(Ci::Build, :count).by(1)
+      end
+    end
 
-        it 'creates a build' do
-          expect { subject }.to change(Ci::Build, :count).by(1)
-        end
+    context 'when scan type is valid' do
+      let(:status) { subject[:status] }
+      let(:pipeline) { subject[:payload] }
+      let(:message) { subject[:message] }
+
+      context 'when action is valid' do
+        it_behaves_like 'valid security orchestration policy action'
 
         it 'sets the build name to secret_detection' do
           build = pipeline.builds.first
@@ -69,6 +76,55 @@ RSpec.describe Security::SecurityOrchestrationPolicies::CreatePipelineService do
 
           expect(build.variables.to_runner_variables).to include(*expected_variables)
         end
+
+        context 'for cluster_image_scanning scan' do
+          let_it_be(:cluster) { create(:cluster, :provided_by_user, name: 'production') }
+          let_it_be(:cluster_project) { create(:cluster_project, cluster: cluster, project: project) }
+          let_it_be(:environment) { create(:environment, name: 'environment-name', project: project) }
+          let_it_be(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, cluster: cluster, cluster_project: cluster_project, project: project, environment: environment) }
+
+          let(:action) { { scan: 'cluster_image_scanning', clusters: { production: { namespaces: ['gitlab-namespace'] } } } }
+
+          it_behaves_like 'valid security orchestration policy action'
+
+          it 'sets the build name to cluster_image_scanning' do
+            build = pipeline.builds.first
+
+            expect(build.name).to eq('cluster_image_scanning')
+          end
+
+          it 'creates a build with appropriate variables' do
+            build = pipeline.builds.first
+
+            expected_variables = [
+              hash_including(
+                key: 'CIS_KUBECONFIG',
+                public: false,
+                masked: false
+              ),
+              {
+                key: 'CIS_RESOURCE_NAMESPACE',
+                masked: false,
+                public: true,
+                value: 'gitlab-namespace'
+              }
+            ]
+
+            expect(build.variables.to_runner_variables).to include(*expected_variables)
+          end
+        end
+
+        context 'for container_scanning scan' do
+          let(:action) { { scan: 'container_scanning' } }
+
+          it_behaves_like 'valid security orchestration policy action'
+
+          it 'sets the build name to container_scanning' do
+            build = pipeline.builds.first
+
+            expect(build.name).to eq('container_scanning')
+          end
+        end
       end
     end
   end

ee/spec/services/security/security_orchestration_policies/rule_schedule_service_spec.rb

@@ -10,7 +10,8 @@ RSpec.describe Security::SecurityOrchestrationPolicies::RuleScheduleService do
     let(:schedule) { create(:security_orchestration_policy_rule_schedule, security_orchestration_policy_configuration: policy_configuration) }
     let!(:scanner_profile) { create(:dast_scanner_profile, name: 'Scanner Profile', project: project) }
     let!(:site_profile) { create(:dast_site_profile, name: 'Site Profile', project: project) }
-    let(:policy) { build(:scan_execution_policy, enabled: true, rules: [{ type: 'schedule', branches: %w[master production], cadence: '*/20 * * * *' }]) }
+    let(:policy) { build(:scan_execution_policy, enabled: true, rules: [rule]) }
+    let(:rule) { { type: 'schedule', branches: %w[master production], cadence: '*/20 * * * *' } }
 
     subject(:service) { described_class.new(container: project, current_user: current_user) }
 
@@ -41,8 +42,82 @@ RSpec.describe Security::SecurityOrchestrationPolicies::RuleScheduleService do
     end
 
     context 'when scan type is secret_detection' do
-      it 'invokes Security::SecurityOrchestrationPolicies::CreatePipelineService' do
+      before do
         policy[:actions] = [{ scan: 'secret_detection' }]
+      end
+
+      it 'invokes Security::SecurityOrchestrationPolicies::CreatePipelineService' do
+        expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to receive(:new).twice.and_call_original
+
+        service.execute(schedule)
+      end
+    end
+
+    context 'when scan type is cluster_image_scanning' do
+      before do
+        policy[:actions] = [{ scan: 'cluster_image_scanning' }]
+      end
+
+      context 'when clusters are not defined in the rule' do
+        it 'invokes Security::SecurityOrchestrationPolicies::CreatePipelineService' do
+          expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to(
+            receive(:new)
+            .with(project: project, current_user: current_user, params: { action: policy[:actions].first.merge(clusters: nil), branch: project.default_branch_or_main })
+            .and_call_original)
+
+          service.execute(schedule)
+        end
+      end
+
+      context 'when clusters are defined in the rule' do
+        let(:rule) { { type: 'schedule', clusters: { production: {} }, cadence: '*/20 * * * *' } }
+
+        it 'invokes Security::SecurityOrchestrationPolicies::CreatePipelineService' do
+          expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to(
+            receive(:new)
+            .with(project: project, current_user: current_user, params: { action: policy[:actions].first.merge(clusters: { production: {} }), branch: project.default_branch_or_main })
+            .and_call_original)
+
+          service.execute(schedule)
+        end
+      end
+    end
+
+    context 'when scan type is container_scanning' do
+      before do
+        policy[:actions] = [{ scan: 'container_scanning' }]
+      end
+
+      context 'when clusters are not defined in the rule' do
+        it 'invokes Security::SecurityOrchestrationPolicies::CreatePipelineService for both branches' do
+          expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to(
+            receive(:new)
+            .with(project: project, current_user: current_user, params: { action: policy[:actions].first, branch: 'master' })
+            .and_call_original)
+
+          expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to(
+            receive(:new)
+            .with(project: project, current_user: current_user, params: { action: policy[:actions].first, branch: 'production' })
+            .and_call_original)
+
+          service.execute(schedule)
+        end
+      end
+
+      context 'when clusters are defined in the rule' do
+        let(:rule) { { type: 'schedule', clusters: { production: {} }, cadence: '*/20 * * * *' } }
+
+        it 'invokes Security::SecurityOrchestrationPolicies::CreatePipelineService for single cluster only' do
+          expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to(
+            receive(:new)
+            .with(project: project, current_user: current_user, params: { action: policy[:actions].first.merge(scan: 'cluster_image_scanning', clusters: { production: {} }), branch: project.default_branch_or_main })
+            .and_call_original)
+
+          service.execute(schedule)
+        end
+      end
+
+      it 'invokes Security::SecurityOrchestrationPolicies::CreatePipelineService' do
         expect(::Security::SecurityOrchestrationPolicies::CreatePipelineService).to receive(:new).twice.and_call_original
 
         service.execute(schedule)

ee/spec/services/security/security_orchestration_policies/scan_pipeline_service_spec.rb

@@ -36,11 +36,13 @@ RSpec.describe Security::SecurityOrchestrationPolicies::ScanPipelineService do
       let(:actions) do
         [
           { scan: 'secret_detection' },
-          { scan: 'dast', scanner_profile: 'Scanner Profile', site_profile: 'Site Profile' }
+          { scan: 'dast', scanner_profile: 'Scanner Profile', site_profile: 'Site Profile' },
+          { scan: 'cluster_image_scanning' },
+          { scan: 'container_scanning' }
         ]
       end
 
-      it_behaves_like 'creates scan jobs', 2, [:'secret-detection-0', :'dast-1']
+      it_behaves_like 'creates scan jobs', 4, [:'secret-detection-0', :'dast-1', :'cluster-image-scanning-2', :'container-scanning-3']
     end
 
     context 'when there are valid and invalid actions' do

spec/models/clusters/cluster_spec.rb

@@ -268,6 +268,16 @@ RSpec.describe Clusters::Cluster, :use_clean_rails_memory_store_caching do
     it { is_expected.to contain_exactly(cluster) }
   end
 
+  describe '.with_name' do
+    subject { described_class.with_name(name) }
+
+    let(:name) { 'this-cluster' }
+    let!(:cluster) { create(:cluster, :project, name: name) }
+    let!(:another_cluster) { create(:cluster, :project) }
+
+    it { is_expected.to contain_exactly(cluster) }
+  end
+
   describe 'validations' do
     subject { cluster.valid? }