Extend DAST scan execution policy to support scheduled execution

app/models/ci/pipeline_schedule.rb

@@ -5,7 +5,7 @@ module Ci
     extend Gitlab::Ci::Model
     include Importable
     include StripAttribute
-    include Schedulable
+    include CronSchedulable
     include Limitable
     include EachBatch
 
@@ -51,36 +51,14 @@ module Ci
       update_attribute(:active, false)
     end
 
-    ##
-    # The `next_run_at` column is set to the actual execution date of `PipelineScheduleWorker`.
-    # This way, a schedule like `*/1 * * * *` won't be triggered in a short interval
-    # when PipelineScheduleWorker runs irregularly by Sidekiq Memory Killer.
-    def set_next_run_at
-      now = Time.zone.now
-      ideal_next_run = ideal_next_run_from(now)
-
-      self.next_run_at = if ideal_next_run == cron_worker_next_run_from(now)
-                           ideal_next_run
-                         else
-                           cron_worker_next_run_from(ideal_next_run)
-                         end
-    end
-
     def job_variables
       variables&.map(&:to_runner_variable) || []
     end
 
     private
 
-    def ideal_next_run_from(start_time)
-      Gitlab::Ci::CronParser.new(cron, cron_timezone)
-                            .next_time_from(start_time)
-    end
-
-    def cron_worker_next_run_from(start_time)
-      Gitlab::Ci::CronParser.new(Settings.cron_jobs['pipeline_schedule_worker']['cron'],
-                                Time.zone.name)
-                            .next_time_from(start_time)
+    def worker_cron_expression
+      Settings.cron_jobs['pipeline_schedule_worker']['cron']
     end
   end
 end

app/models/concerns/cron_schedulable.rb

+# frozen_string_literal: true
+
+module CronSchedulable
+  extend ActiveSupport::Concern
+  include Schedulable
+
+  ##
+  # The `next_run_at` column is set to the actual execution date of worker that
+  # triggers the schedule. This way, a schedule like `*/1 * * * *` won't be triggered
+  # in a short interval when the worker runs irregularly by Sidekiq Memory Killer.
+  def set_next_run_at
+    now = Time.zone.now
+    ideal_next_run = ideal_next_run_from(now)
+
+    self.next_run_at = if ideal_next_run == cron_worker_next_run_from(now)
+                         ideal_next_run
+                       else
+                         cron_worker_next_run_from(ideal_next_run)
+                       end
+  end
+
+  private
+
+  def ideal_next_run_from(start_time)
+    next_time_from(start_time, cron, cron_timezone)
+  end
+
+  def cron_worker_next_run_from(start_time)
+    next_time_from(start_time, worker_cron_expression, Time.zone.name)
+  end
+
+  def next_time_from(start_time, cron, cron_timezone)
+    Gitlab::Ci::CronParser
+      .new(cron, cron_timezone)
+      .next_time_from(start_time)
+  end
+
+  def worker_cron_expression
+    raise NotImplementedError
+  end
+end

config/initializers/1_settings.rb

@@ -691,6 +691,12 @@ Gitlab.ee do
   Settings.cron_jobs['vulnerability_historical_statistics_deletion_worker'] ||= Settingslogic.new({})
   Settings.cron_jobs['vulnerability_historical_statistics_deletion_worker']['cron'] ||= '15 3 * * *'
   Settings.cron_jobs['vulnerability_historical_statistics_deletion_worker']['job_class'] = 'Vulnerabilities::HistoricalStatistics::DeletionWorker'
+  Settings.cron_jobs['security_create_orchestration_policy_worker'] ||= Settingslogic.new({})
+  Settings.cron_jobs['security_create_orchestration_policy_worker']['cron'] ||= '*/10 * * * *'
+  Settings.cron_jobs['security_create_orchestration_policy_worker']['job_class'] = 'Security::CreateOrchestrationPolicyWorker'
+  Settings.cron_jobs['security_orchestration_policy_rule_schedule_worker'] ||= Settingslogic.new({})
+  Settings.cron_jobs['security_orchestration_policy_rule_schedule_worker']['cron'] ||= '*/15 * * * *'
+  Settings.cron_jobs['security_orchestration_policy_rule_schedule_worker']['job_class'] = 'Security::OrchestrationPolicyRuleScheduleWorker'
 end
 
 #

ee/app/models/security/orchestration_policy_configuration.rb

@@ -2,6 +2,7 @@
 
 module Security
   class OrchestrationPolicyConfiguration < ApplicationRecord
+    include EachBatch
     include Gitlab::Utils::StrongMemoize
 
     self.table_name = 'security_orchestration_policy_configurations'
@@ -10,6 +11,11 @@ module Security
     POLICY_SCHEMA_PATH = 'ee/app/validators/json_schemas/security_orchestration_policy.json'
     POLICY_LIMIT = 5
 
+    RULE_TYPES = {
+      pipeline: 'pipeline',
+      schedule: 'schedule'
+    }.freeze
+
     ON_DEMAND_SCANS = %w[dast].freeze
 
     belongs_to :project, inverse_of: :security_orchestration_policy_configuration
@@ -24,6 +30,10 @@ module Security
     validates :security_policy_management_project, presence: true
 
     scope :for_project, -> (project_id) { where(project_id: project_id) }
+    scope :with_outdated_configuration, -> do
+      joins(:security_policy_management_project)
+        .where(arel_table[:configured_at].lt(Project.arel_table[:last_repository_updated_at]).or(arel_table[:configured_at].eq(nil)))
+    end
 
     def enabled?
       ::Feature.enabled?(:security_orchestration_policies_configuration, project)
@@ -60,6 +70,16 @@ module Security
       active_policy_names_with_dast_profiles.dig(:scanner_profiles, profile_name)
     end
 
+    def policy_last_updated_by
+      strong_memoize(:policy_last_updated_by) do
+        policy_repo.last_commit_for_path(default_branch_or_main, POLICY_PATH)&.author
+      end
+    end
+
+    def delete_all_schedules
+      rule_schedules.delete_all(:delete_all)
+    end
+
     private
 
     def policy_repo
@@ -107,7 +127,7 @@ module Security
 
     def applicable_for_branch?(policy, ref)
       policy[:rules].any? do |rule|
-        rule[:type] == 'pipeline' && rule[:branches].any? { |branch| RefMatcher.new(branch).matches?(ref) }
+        rule[:type] == RULE_TYPES[:pipeline] && rule[:branches].any? { |branch| RefMatcher.new(branch).matches?(ref) }
       end
     end
   end

ee/app/models/security/orchestration_policy_rule_schedule.rb

@@ -2,6 +2,8 @@
 
 module Security
   class OrchestrationPolicyRuleSchedule < ApplicationRecord
+    include CronSchedulable
+
     self.table_name = 'security_orchestration_policy_rule_schedules'
 
     belongs_to :owner, class_name: 'User', foreign_key: 'user_id'
@@ -13,5 +15,27 @@ module Security
     validates :security_orchestration_policy_configuration, presence: true
     validates :cron, presence: true
     validates :policy_index, presence: true
+
+    scope :runnable_schedules, -> { where("next_run_at < ?", Time.zone.now) }
+    scope :with_owner, -> { includes(:owner) }
+    scope :with_configuration_and_project, -> do
+      includes(
+        security_orchestration_policy_configuration: [:project, :security_policy_management_project]
+      )
+    end
+
+    def policy
+      security_orchestration_policy_configuration.active_policies.at(policy_index)
+    end
+
+    private
+
+    def cron_timezone
+      Time.zone.name
+    end
+
+    def worker_cron_expression
+      Settings.cron_jobs['security_orchestration_policy_rule_schedule_worker']['cron']
+    end
   end
 end

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

+# frozen_string_literal: true
+
+module Security
+  module SecurityOrchestrationPolicies
+    class ProcessRuleService
+      def initialize(policy_configuration:, policy_index:, policy:)
+        @policy_configuration = policy_configuration
+        @policy_index = policy_index
+        @policy = policy
+      end
+
+      def execute
+        policy_configuration.delete_all_schedules
+        create_new_schedule_rules
+        policy_configuration.update!(configured_at: Time.current)
+      end
+
+      private
+
+      attr_reader :policy_configuration, :policy_index, :policy
+
+      def create_new_schedule_rules
+        return unless policy_configuration.enabled?
+
+        policy[:rules]
+          .select { |rule| rule[:type] == Security::OrchestrationPolicyConfiguration::RULE_TYPES[:schedule] }
+          .each do |rule|
+          Security::OrchestrationPolicyRuleSchedule
+            .create!(
+              security_orchestration_policy_configuration: policy_configuration,
+              policy_index: policy_index,
+              cron: rule[:cadence],
+              owner: policy_configuration.policy_last_updated_by
+            )
+        end
+      end
+    end
+  end
+end

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

+# frozen_string_literal: true
+
+module Security
+  module SecurityOrchestrationPolicies
+    class RuleScheduleService < BaseContainerService
+      def execute(schedule)
+        schedule.schedule_next_run!
+        actions_for(schedule)
+          .each { |action| process_action(action) }
+      end
+
+      private
+
+      def actions_for(schedule)
+        policy = schedule.policy
+        return [] if policy.blank?
+
+        policy[:actions]
+      end
+
+      def process_action(action)
+        case action[:scan]
+        when 'dast' then schedule_dast_on_demand_scan(action)
+        end
+      end
+
+      def schedule_dast_on_demand_scan(action)
+        dast_site_profile = find_dast_site_profile(container, action[:site_profile])
+        dast_scanner_profile = find_dast_scanner_profile(container, action[:scanner_profile])
+
+        ::DastOnDemandScans::CreateService.new(
+          container: container,
+          current_user: current_user,
+          params: {
+            dast_site_profile: dast_site_profile,
+            dast_scanner_profile: dast_scanner_profile
+          }
+        ).execute
+      end
+
+      def find_dast_site_profile(project, dast_site_profile)
+        DastSiteProfilesFinder.new(project_id: project.id, name: dast_site_profile).execute.first
+      end
+
+      def find_dast_scanner_profile(project, dast_scanner_profile)
+        return unless dast_scanner_profile
+
+        DastScannerProfilesFinder.new(project_ids: [project.id], name: dast_scanner_profile).execute.first
+      end
+    end
+  end
+end

ee/app/validators/json_schemas/security_orchestration_policy.json

@@ -36,7 +36,8 @@
                           "properties": {
                               "type": {
                                   "enum": [
-                                      "pipeline"
+                                      "pipeline",
+                                      "schedule"
                                   ],
                                   "type": "string"
                               },
@@ -47,7 +48,20 @@
                                       "minLength": 1,
                                       "type": "string"
                                   }
+                              },
+                             "cadence": {
+                                  "type": "string"
+                             }
+                          },
+                          "if": {
+                                "properties": {
+                                    "type": {
+                                        "const": "schedule"
                                     }
+                                }
+                          },
+                          "then": {
+                                "required": ["cadence"]
                           },
                           "additionalProperties": false
                       }

ee/app/workers/all_queues.yml

@@ -368,6 +368,24 @@
   :weight: 1
   :idempotent:
   :tags: []
+- :name: cronjob:security_create_orchestration_policy
+  :worker_name: Security::CreateOrchestrationPolicyWorker
+  :feature_category: :security_orchestration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent:
+  :tags: []
+- :name: cronjob:security_orchestration_policy_rule_schedule
+  :worker_name: Security::OrchestrationPolicyRuleScheduleWorker
+  :feature_category: :security_orchestration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent:
+  :tags: []
 - :name: cronjob:sync_seat_link
   :worker_name: SyncSeatLinkWorker
   :feature_category: :license

ee/app/workers/security/create_orchestration_policy_worker.rb

+# frozen_string_literal: true
+
+module Security
+  class CreateOrchestrationPolicyWorker # rubocop:disable Scalability/IdempotentWorker
+    include ApplicationWorker
+    # rubocop:disable Scalability/CronWorkerContext
+    # This worker does not perform work scoped to a context
+    include CronjobQueue
+    # rubocop:enable Scalability/CronWorkerContext
+
+    feature_category :security_orchestration
+
+    def perform
+      Security::OrchestrationPolicyConfiguration.with_outdated_configuration.each_batch do |configurations|
+        configurations.each do |configuration|
+          unless configuration.policy_configuration_valid?
+            configuration.delete_all_schedules
+            next
+          end
+
+          configuration.active_policies.each_with_index do |policy, policy_index|
+            Security::SecurityOrchestrationPolicies::ProcessRuleService
+              .new(policy_configuration: configuration, policy_index: policy_index, policy: policy)
+              .execute
+          end
+        end
+      end
+    end
+  end
+end

ee/app/workers/security/orchestration_policy_rule_schedule_worker.rb

+# frozen_string_literal: true
+
+module Security
+  class OrchestrationPolicyRuleScheduleWorker # rubocop:disable Scalability/IdempotentWorker
+    include ApplicationWorker
+    # rubocop:disable Scalability/CronWorkerContext
+    # This worker does not perform work scoped to a context
+    include CronjobQueue
+    # rubocop:enable Scalability/CronWorkerContext
+
+    feature_category :security_orchestration
+
+    def perform
+      Security::OrchestrationPolicyRuleSchedule.with_configuration_and_project.with_owner.runnable_schedules.find_in_batches do |schedules|
+        schedules.each do |schedule|
+          with_context(project: schedule.security_orchestration_policy_configuration.project, user: schedule.owner) do
+            Security::SecurityOrchestrationPolicies::RuleScheduleService
+              .new(container: schedule.security_orchestration_policy_configuration.project, current_user: schedule.owner)
+              .execute(schedule)
+          end
+        end
+      end
+    end
+  end
+end

ee/spec/models/security/orchestration_policy_configuration_spec.rb

@@ -5,7 +5,7 @@ require 'spec_helper'
 RSpec.describe Security::OrchestrationPolicyConfiguration do
   let_it_be(:security_policy_management_project) { create(:project, :repository) }
 
-  let!(:security_orchestration_policy_configuration) do
+  let(:security_orchestration_policy_configuration) do
     create(:security_orchestration_policy_configuration, security_policy_management_project: security_policy_management_project)
   end
 
@@ -28,9 +28,9 @@ RSpec.describe Security::OrchestrationPolicyConfiguration do
   end
 
   describe '.for_project' do
-    let!(:security_orchestration_policy_configuration_1) { create(:security_orchestration_policy_configuration) }
-    let!(:security_orchestration_policy_configuration_2) { create(:security_orchestration_policy_configuration) }
-    let!(:security_orchestration_policy_configuration_3) { create(:security_orchestration_policy_configuration) }
+    let_it_be(:security_orchestration_policy_configuration_1) { create(:security_orchestration_policy_configuration) }
+    let_it_be(:security_orchestration_policy_configuration_2) { create(:security_orchestration_policy_configuration) }
+    let_it_be(:security_orchestration_policy_configuration_3) { create(:security_orchestration_policy_configuration) }
 
     subject { described_class.for_project([security_orchestration_policy_configuration_2.project, security_orchestration_policy_configuration_3.project]) }
 
@@ -39,6 +39,18 @@ RSpec.describe Security::OrchestrationPolicyConfiguration do
     end
   end
 
+  describe '.with_outdated_configuration' do
+    let!(:security_orchestration_policy_configuration_1) { create(:security_orchestration_policy_configuration, configured_at: nil) }
+    let!(:security_orchestration_policy_configuration_2) { create(:security_orchestration_policy_configuration, configured_at: Time.zone.now - 1.hour) }
+    let!(:security_orchestration_policy_configuration_3) { create(:security_orchestration_policy_configuration, configured_at: Time.zone.now + 1.hour) }
+
+    subject { described_class.with_outdated_configuration }
+
+    it 'returns configuration with outdated configurations' do
+      is_expected.to contain_exactly(security_orchestration_policy_configuration_1, security_orchestration_policy_configuration_2)
+    end
+  end
+
   describe '#enabled?' do
     subject { security_orchestration_policy_configuration.enabled? }
 
@@ -365,4 +377,37 @@ RSpec.describe Security::OrchestrationPolicyConfiguration do
       expect(security_orchestration_policy_configuration.active_policy_names_with_dast_scanner_profile('Scanner Profile')).to contain_exactly('Run DAST in every pipeline')
     end
   end
+
+  describe '#policy_last_updated_by' do
+    let(:commit) { create(:commit, author: security_policy_management_project.owner) }
+
+    subject(:policy_last_updated_by) { security_orchestration_policy_configuration.policy_last_updated_by }
+
+    before do
+      allow(security_policy_management_project).to receive(:repository).and_return(repository)
+      allow(repository).to receive(:last_commit_for_path).with(default_branch, Security::OrchestrationPolicyConfiguration::POLICY_PATH).and_return(commit)
+    end
+
+    context 'when last commit to policy file exists' do
+      it { is_expected.to eq(security_policy_management_project.owner) }
+    end
+
+    context 'when last commit to policy file does not exist' do
+      let(:commit) {}
+
+      it { is_expected.to be_nil }
+    end
+  end
+
+  describe '#delete_all_schedules' do
+    let(:rule_schedule) { create(:security_orchestration_policy_rule_schedule, security_orchestration_policy_configuration: security_orchestration_policy_configuration) }
+
+    subject(:delete_all_schedules) { security_orchestration_policy_configuration.delete_all_schedules }
+
+    it 'deletes all schedules belonging to configuration' do
+      delete_all_schedules
+
+      expect(security_orchestration_policy_configuration.rule_schedules).to be_empty
+    end
+  end
 end

ee/spec/models/security/orchestration_policy_rule_schedule_spec.rb

@@ -16,4 +16,103 @@ RSpec.describe Security::OrchestrationPolicyRuleSchedule do
     it { is_expected.to validate_presence_of(:cron) }
     it { is_expected.to validate_presence_of(:policy_index) }
   end
+
+  describe '.runnable_schedules' do
+    subject { described_class.runnable_schedules }
+
+    context 'when there are runnable schedules' do
+      let!(:policy_rule_schedule) do
+        travel_to(1.day.ago) do
+          create(:security_orchestration_policy_rule_schedule)
+        end
+      end
+
+      it 'returns the runnable schedule' do
+        is_expected.to eq([policy_rule_schedule])
+      end
+    end
+
+    context 'when there are no runnable schedules' do
+      let!(:policy_rule_schedule) { }
+
+      it 'returns an empty array' do
+        is_expected.to be_empty
+      end
+    end
+
+    context 'when there are runnable schedules in future' do
+      let!(:policy_rule_schedule) do
+        travel_to(1.day.from_now) do
+          create(:security_orchestration_policy_rule_schedule)
+        end
+      end
+
+      it 'returns an empty array' do
+        is_expected.to be_empty
+      end
+    end
+  end
+
+  describe '#policy' do
+    let(:rule_schedule) { create(:security_orchestration_policy_rule_schedule) }
+    let(:policy_yaml) { { scan_execution_policy: [policy] }.to_yaml }
+
+    subject { rule_schedule.policy }
+
+    before do
+      allow_next_instance_of(Repository) do |repository|
+        allow(repository).to receive(:blob_data_at).and_return(policy_yaml)
+      end
+    end
+
+    context 'when policy is present' do
+      let(:policy) do
+        {
+          name: 'Scheduled DAST 1',
+          description: 'This policy runs DAST for every 20 mins',
+          enabled: true,
+          rules: [{ type: 'schedule', branches: %w[production], cadence: '*/20 * * * *' }],
+          actions: [
+            { scan: 'dast', site_profile: 'Site Profile', scanner_profile: 'Scanner Profile' }
+          ]
+        }
+      end
+
+      it { is_expected.to eq(policy) }
+    end
+
+    context 'when policy is not present' do
+      let(:policy_yaml) { nil }
+
+      it { is_expected.to be_nil }
+    end
+
+    context 'when policy is not enabled' do
+      let(:policy) do
+        {
+          name: 'Scheduled DAST 1',
+          description: 'This policy runs DAST for every 20 mins',
+          enabled: false,
+          rules: [{ type: 'schedule', branches: %w[production], cadence: '*/20 * * * *' }],
+          actions: [
+            { scan: 'dast', site_profile: 'Site Profile', scanner_profile: 'Scanner Profile' }
+          ]
+        }
+      end
+
+      it { is_expected.to be_nil }
+    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 * * * *') }
+      let(:schedule_1) { create(:security_orchestration_policy_rule_schedule) }
+      let(:schedule_2) { create(:security_orchestration_policy_rule_schedule) }
+      let(:new_cron) { '0 0 1 1 *' }
+
+      let(:ideal_next_run_at) { schedule.send(:ideal_next_run_from, Time.zone.now) }
+      let(:cron_worker_next_run_at) { schedule.send(:cron_worker_next_run_from, Time.zone.now) }
+    end
+  end
 end

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

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::SecurityOrchestrationPolicies::ProcessRuleService do
+  describe '#execute' do
+    let_it_be(:policy_configuration) { create(:security_orchestration_policy_configuration) }
+    let_it_be(:owner) { create(:user) }
+    let_it_be(:schedule) do
+      travel_to(1.day.ago) do
+        create(:security_orchestration_policy_rule_schedule, security_orchestration_policy_configuration: policy_configuration)
+      end
+    end
+
+    let(:policy) do
+      {
+        name: 'Scheduled DAST',
+        description: 'This policy runs DAST for every 15 mins',
+        enabled: true,
+        rules: [{ type: 'schedule', branches: %w[production], cadence: '*/15 * * * *' }],
+        actions: [
+          { scan: 'dast', site_profile: 'Site Profile', scanner_profile: 'Scanner Profile' }
+        ]
+      }
+    end
+
+    subject(:service) { described_class.new(policy_configuration: policy_configuration, policy_index: 0, policy: policy) }
+
+    before do
+      allow(policy_configuration).to receive(:policy_last_updated_by).and_return(owner)
+    end
+
+    context 'when security_orchestration_policies_configuration feature is enabled and policy is scheduled' do
+      it 'creates new schedule' do
+        service.execute
+
+        new_schedule = Security::OrchestrationPolicyRuleSchedule.first
+        expect(policy_configuration.configured_at).not_to be_nil
+        expect(Security::OrchestrationPolicyRuleSchedule.count).to eq(1)
+        expect(new_schedule.id).not_to eq(schedule.id)
+        expect(new_schedule.next_run_at).to be > schedule.next_run_at
+      end
+    end
+
+    context 'when security_orchestration_policies_configuration feature is disabled' do
+      before do
+        stub_feature_flags(security_orchestration_policies_configuration: false)
+      end
+
+      it 'deletes schedules' do
+        expect { service.execute }.to change(Security::OrchestrationPolicyRuleSchedule, :count).by(-1)
+        expect(policy_configuration.configured_at).not_to be_nil
+      end
+    end
+
+    context 'when policy is not of type scheduled' do
+      let(:policy) do
+        {
+          name: 'Run DAST in every pipeline',
+          description: 'This policy enforces to run DAST for every pipeline within the project',
+          enabled: false,
+          rules: [{ type: 'pipeline', branches: %w[production] }],
+          actions: [
+            { scan: 'dast', site_profile: 'Site Profile', scanner_profile: 'Scanner Profile' }
+          ]
+        }
+      end
+
+      it 'deletes schedules' do
+        expect { service.execute }.to change(Security::OrchestrationPolicyRuleSchedule, :count).by(-1)
+        expect(policy_configuration.configured_at).not_to be_nil
+      end
+    end
+  end
+end

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

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::SecurityOrchestrationPolicies::RuleScheduleService do
+  describe '#execute' do
+    let(:project) { create(:project, :repository) }
+    let(:current_user) { project.users.first }
+    let(:policy_configuration) { create(:security_orchestration_policy_configuration, project: project) }
+    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) do
+      {
+        name: 'Run DAST in every pipeline',
+        description: 'This policy enforces to run DAST for every pipeline within the project',
+        enabled: true,
+        rules: [{ type: 'schedule', branches: %w[production], cadence: '*/20 * * * *' }],
+        actions: [
+          { scan: 'dast', site_profile: 'Site Profile', scanner_profile: 'Scanner Profile' }
+        ]
+      }
+    end
+
+    subject(:service) { described_class.new(container: project, current_user: current_user) }
+
+    shared_examples 'does not execute DAST on demand-scan' do
+      it 'does not create a DAST on demand-scan pipeline but updates next_run_at' do
+        expect { service.execute(schedule) }.to change(Ci::Pipeline, :count).by(0)
+
+        expect(schedule.next_run_at).to be > Time.zone.now
+      end
+    end
+
+    before do
+      stub_licensed_features(security_on_demand_scans: true)
+
+      allow_next_instance_of(Security::OrchestrationPolicyConfiguration) do |instance|
+        allow(instance).to receive(:active_policies).and_return([policy])
+      end
+    end
+
+    context 'when policy actions exists' do
+      it 'creates a DAST on demand-scan pipeline and updates next_run_at' do
+        expect { service.execute(schedule) }.to change(Ci::Pipeline, :count).by(1)
+
+        expect(schedule.next_run_at).to be > Time.zone.now
+      end
+    end
+
+    context 'when policy actions does not exist' do
+      let(:policy) do
+        {
+          name: 'Run DAST in every pipeline',
+          description: 'This policy enforces to run DAST for every pipeline within the project',
+          enabled: true,
+          rules: [{ type: 'schedule', branches: %w[production], cadence: '*/20 * * * *' }],
+          actions: []
+        }
+      end
+
+      it_behaves_like 'does not execute DAST on demand-scan'
+    end
+
+    context 'when policy scan type is invalid' do
+      let(:policy) do
+        {
+          name: 'Run DAST in every pipeline',
+          description: 'This policy enforces to run DAST for every pipeline within the project',
+          enabled: true,
+          rules: [{ type: 'schedule', branches: %w[production], cadence: '*/20 * * * *' }],
+          actions: [
+            { scan: 'invalid' }
+          ]
+        }
+      end
+
+      it_behaves_like 'does not execute DAST on demand-scan'
+    end
+
+    context 'when policy does not exist' do
+      let(:policy) { nil }
+
+      it_behaves_like 'does not execute DAST on demand-scan'
+    end
+  end
+end

ee/spec/workers/security/create_orchestration_policy_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::CreateOrchestrationPolicyWorker do
+  describe '#perform' do
+    let_it_be(:configuration) { create(:security_orchestration_policy_configuration) }
+    let_it_be(:schedule) { create(:security_orchestration_policy_rule_schedule, security_orchestration_policy_configuration: configuration) }
+
+    before do
+      allow_next_instance_of(Repository) do |repository|
+        allow(repository).to receive(:blob_data_at).and_return({ scan_execution_policy: active_policies }.to_yaml)
+      end
+    end
+
+    subject(:worker) { described_class.new }
+
+    context 'when policy is valid' do
+      let(:active_policies) do
+        [
+          {
+            name: 'Scheduled DAST 1',
+            description: 'This policy runs DAST for every 20 mins',
+            enabled: true,
+            rules: [{ type: 'schedule', branches: %w[production], cadence: '*/20 * * * *' }],
+            actions: [
+              { scan: 'dast', site_profile: 'Site Profile', scanner_profile: 'Scanner Profile' }
+            ]
+          },
+          {
+            name: 'Scheduled DAST 2',
+            description: 'This policy runs DAST for every 20 mins',
+            enabled: true,
+            rules: [{ type: 'schedule', branches: %w[production], cadence: '*/20 * * * *' }],
+            actions: [
+              { scan: 'dast', site_profile: 'Site Profile', scanner_profile: 'Scanner Profile' }
+            ]
+          }
+        ]
+      end
+
+      it 'executes the process rule service' do
+        active_policies.each_with_index do |policy, policy_index|
+          expect_next_instance_of(Security::SecurityOrchestrationPolicies::ProcessRuleService,
+                                  policy_configuration: configuration, policy_index: policy_index, policy: policy) do |service|
+            expect(service).to receive(:execute)
+          end
+        end
+
+        expect { worker.perform }.not_to change(Security::OrchestrationPolicyRuleSchedule, :count)
+      end
+    end
+
+    context 'when policy is invalid' do
+      let(:active_policies) do
+        [
+          {
+            key: 'invalid',
+            label: 'invalid'
+          }
+        ]
+      end
+
+      it 'does not execute process rule service' do
+        expect(Security::SecurityOrchestrationPolicies::ProcessRuleService).not_to receive(:new)
+
+        expect { worker.perform }.to change(Security::OrchestrationPolicyRuleSchedule, :count).by(-1)
+      end
+    end
+  end
+end

ee/spec/workers/security/orchestration_policy_rule_schedule_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::OrchestrationPolicyRuleScheduleWorker do
+  describe '#perform' do
+    let_it_be(:schedule) { create(:security_orchestration_policy_rule_schedule) }
+
+    subject(:worker) { described_class.new }
+
+    context 'when schedule exists' do
+      before do
+        schedule.update_column(:next_run_at, 1.minute.ago)
+      end
+
+      it 'executes the rule schedule service' do
+        expect_next_instance_of(Security::SecurityOrchestrationPolicies::RuleScheduleService,
+                                container: schedule.security_orchestration_policy_configuration.project, current_user: schedule.owner) do |service|
+          expect(service).to receive(:execute)
+        end
+
+        worker.perform
+      end
+    end
+
+    context 'when schedule does not exist' do
+      before do
+        schedule.update_column(:next_run_at, 1.minute.from_now)
+      end
+
+      it 'executes the rule schedule service' do
+        expect(Security::SecurityOrchestrationPolicies::RuleScheduleService).not_to receive(:new)
+
+        worker.perform
+      end
+    end
+
+    context 'when multiple schedules exists' do
+      before do
+        schedule.update_column(:next_run_at, 1.minute.ago)
+      end
+
+      def record_preloaded_queries
+        recorder = ActiveRecord::QueryRecorder.new { worker.perform }
+        recorder.data.values.flat_map {|v| v[:occurrences]}.select do |query|
+          ['FROM "projects"', 'FROM "users"', 'FROM "security_orchestration_policy_configurations"'].any? do |s|
+            query.include?(s)
+          end
+        end
+      end
+
+      it 'preloads configuration, project and owner to avoid N+1 queries' do
+        expected_count = record_preloaded_queries.count
+
+        travel_to(30.minutes.ago) { create_list(:security_orchestration_policy_rule_schedule, 5) }
+        actual_count = record_preloaded_queries.count
+
+        expect(actual_count).to eq(expected_count)
+      end
+    end
+  end
+end

spec/models/ci/pipeline_schedule_spec.rb

@@ -126,16 +126,6 @@ RSpec.describe Ci::PipelineSchedule do
       end
     end
 
-    context 'when pipeline schedule runs every minute' do
-      let(:pipeline_schedule) { create(:ci_pipeline_schedule, :every_minute) }
-
-      it "updates next_run_at to the sidekiq worker's execution time" do
-        travel_to(Time.zone.parse("2019-06-01 12:18:00+0000")) do
-          expect(pipeline_schedule.next_run_at).to eq(cron_worker_next_run_at)
-        end
-      end
-    end
-
     context 'when there are two different pipeline schedules in different time zones' do
       let(:pipeline_schedule_1) { create(:ci_pipeline_schedule, :weekly, cron_timezone: 'Eastern Time (US & Canada)') }
       let(:pipeline_schedule_2) { create(:ci_pipeline_schedule, :weekly, cron_timezone: 'UTC') }
@@ -144,24 +134,6 @@ RSpec.describe Ci::PipelineSchedule do
         expect(pipeline_schedule_1.next_run_at).not_to eq(pipeline_schedule_2.next_run_at)
       end
     end
-
-    context 'when there are two different pipeline schedules in the same time zones' do
-      let(:pipeline_schedule_1) { create(:ci_pipeline_schedule, :weekly, cron_timezone: 'UTC') }
-      let(:pipeline_schedule_2) { create(:ci_pipeline_schedule, :weekly, cron_timezone: 'UTC') }
-
-      it 'sets the sames next_run_at' do
-        expect(pipeline_schedule_1.next_run_at).to eq(pipeline_schedule_2.next_run_at)
-      end
-    end
-
-    context 'when updates cron of exsisted pipeline schedule' do
-      let(:new_cron) { '0 0 1 1 *' }
-
-      it 'updates next_run_at automatically' do
-        expect { pipeline_schedule.update!(cron: new_cron) }
-          .to change { pipeline_schedule.next_run_at }
-      end
-    end
   end
 
   describe '#schedule_next_run!' do

spec/models/concerns/cron_schedulable_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe CronSchedulable do
+  let(:ideal_next_run_at) { schedule.send(:ideal_next_run_from, Time.zone.now) }
+  let(:cron_worker_next_run_at) { schedule.send(:cron_worker_next_run_from, Time.zone.now) }
+
+  context 'for ci_pipeline_schedule' do
+    let(:schedule) { create(:ci_pipeline_schedule, :every_minute) }
+    let(:schedule_1) { create(:ci_pipeline_schedule, :weekly, cron_timezone: 'UTC') }
+    let(:schedule_2) { create(:ci_pipeline_schedule, :weekly, cron_timezone: 'UTC') }
+    let(:new_cron) { '0 0 1 1 *' }
+
+    it_behaves_like 'handles set_next_run_at'
+  end
+end

spec/support/shared_examples/models/concerns/cron_schedulable_shared_examples.rb

+# frozen_string_literal: true
+
+RSpec.shared_examples 'handles set_next_run_at' do
+  context 'when schedule runs every minute' do
+    it "updates next_run_at to the worker's execution time" do
+      travel_to(1.day.ago) do
+        expect(schedule.next_run_at).to eq(cron_worker_next_run_at)
+      end
+    end
+  end
+
+  context 'when there are two different schedules in the same time zones' do
+    it 'sets the sames next_run_at' do
+      expect(schedule_1.next_run_at).to eq(schedule_2.next_run_at)
+    end
+  end
+
+  context 'when cron is updated for existing schedules' do
+    it 'updates next_run_at automatically' do
+      expect { schedule.update!(cron: new_cron) }.to change { schedule.next_run_at }
+    end
+  end
+end