Implement Scheduling Logic for DAST On-demand Scheduler

Add new cron job app_sec_dast_profile_schedule_worker.

The cron job is configured to run every 15th minute of an hour.

ProfileScheduleWorker will periodically run all the runnable_schedules

Add new table dast_profile_schedules with the following three foreign keys
  dast_profile_id
  user_id
  project_id

Set up one to many relationship between dast_profile and
dast_profile_schedule.

Model dast_profile_schedule will hold the information about the
schedules of dast scans.

Add specs.

Issue link - https://gitlab.com/gitlab-org/gitlab/-/issues/330307

Changelog: added

config/initializers/1_settings.rb

@@ -701,6 +701,9 @@ Gitlab.ee do
   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'
+  Settings.cron_jobs['app_sec_dast_profile_schedule_worker'] ||= Settingslogic.new({})
+  Settings.cron_jobs['app_sec_dast_profile_schedule_worker']['cron'] ||= '7-59/15 * * * *'
+  Settings.cron_jobs['app_sec_dast_profile_schedule_worker']['job_class'] = 'AppSec::Dast::ProfileScheduleWorker'
 end
 
 #

db/migrate/20210713123345_create_dast_profile_schedule.rb

+# frozen_string_literal: true
+
+class CreateDastProfileSchedule < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+
+  INDEX_NAME = 'index_dast_profile_schedules_active_next_run_at'
+
+  def up
+    table_comment = {
+      owner: 'group::dynamic analysis', description: 'Scheduling for scans using DAST Profiles'
+    }
+
+    create_table_with_constraints :dast_profile_schedules, comment: table_comment.to_json do |t|
+      t.bigint :project_id, null: false
+      t.bigint :dast_profile_id, null: false
+      t.bigint :user_id
+
+      t.datetime_with_timezone :next_run_at, null: false
+      t.timestamps_with_timezone null: false
+      t.boolean :active, default: true, null: false
+      t.text :cron, null: false
+      t.text_limit :cron, 255
+
+      t.index %i[active next_run_at], name: INDEX_NAME
+      t.index %i[project_id dast_profile_id], unique: true
+      t.index :dast_profile_id
+      t.index :user_id
+    end
+  end
+
+  def down
+    with_lock_retries do
+      drop_table :dast_profile_schedules
+    end
+  end
+end

db/migrate/20210715074359_add_foreign_key_to_dast_profile_schedules_on_dast_profile.rb

+# frozen_string_literal: true
+
+class AddForeignKeyToDastProfileSchedulesOnDastProfile < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_foreign_key :dast_profile_schedules, :dast_profiles, column: :dast_profile_id, on_delete: :cascade
+  end
+
+  def down
+    with_lock_retries do
+      remove_foreign_key_if_exists :dast_profile_schedules, column: :dast_profile_id
+    end
+  end
+end

db/migrate/20210715074933_add_foreign_key_to_dast_profile_schedules_on_user.rb

+# frozen_string_literal: true
+
+class AddForeignKeyToDastProfileSchedulesOnUser < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_foreign_key :dast_profile_schedules, :users, column: :user_id, on_delete: :nullify
+  end
+
+  def down
+    with_lock_retries do
+      remove_foreign_key_if_exists :dast_profile_schedules, column: :user_id
+    end
+  end
+end

db/migrate/20210715075203_add_foreign_key_to_dast_profile_schedules_on_project.rb

+# frozen_string_literal: true
+
+class AddForeignKeyToDastProfileSchedulesOnProject < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_foreign_key :dast_profile_schedules, :projects, column: :project_id, on_delete: :cascade
+  end
+
+  def down
+    with_lock_retries do
+      remove_foreign_key_if_exists :dast_profile_schedules, column: :project_id
+    end
+  end
+end

db/schema_migrations/20210713123345

+d1226fdefe839aae4c7425924058e1944f883824c43a299b154bb6873d6c3855
\ No newline at end of file

db/schema_migrations/20210715074359

+7bc0654a97f85100df93b9dbbbdab374873f6d3d379a4393f718bad923b064ba
\ No newline at end of file

db/schema_migrations/20210715074933

+e29240947b2e0e6fa7c91643c5d1a2efa02ec062b5ccdffdf382dff993ab6225
\ No newline at end of file

db/schema_migrations/20210715075203

+9d29f4d776031e90cb42122146f65bb13e8778d223467a83dc311f4adab31565
\ No newline at end of file

db/structure.sql

@@ -12006,6 +12006,30 @@ CREATE SEQUENCE custom_emoji_id_seq
 
 ALTER SEQUENCE custom_emoji_id_seq OWNED BY custom_emoji.id;
 
+CREATE TABLE dast_profile_schedules (
+    id bigint NOT NULL,
+    project_id bigint NOT NULL,
+    dast_profile_id bigint NOT NULL,
+    user_id bigint,
+    next_run_at timestamp with time zone NOT NULL,
+    created_at timestamp with time zone NOT NULL,
+    updated_at timestamp with time zone NOT NULL,
+    active boolean DEFAULT true NOT NULL,
+    cron text NOT NULL,
+    CONSTRAINT check_86531ea73f CHECK ((char_length(cron) <= 255))
+);
+
+COMMENT ON TABLE dast_profile_schedules IS '{"owner":"group::dynamic analysis","description":"Scheduling for scans using DAST Profiles"}';
+
+CREATE SEQUENCE dast_profile_schedules_id_seq
+    START WITH 1
+    INCREMENT BY 1
+    NO MINVALUE
+    NO MAXVALUE
+    CACHE 1;
+
+ALTER SEQUENCE dast_profile_schedules_id_seq OWNED BY dast_profile_schedules.id;
+
 CREATE TABLE dast_profiles (
     id bigint NOT NULL,
     project_id bigint NOT NULL,
@@ -20024,6 +20048,8 @@ ALTER TABLE ONLY csv_issue_imports ALTER COLUMN id SET DEFAULT nextval('csv_issu
 
 ALTER TABLE ONLY custom_emoji ALTER COLUMN id SET DEFAULT nextval('custom_emoji_id_seq'::regclass);
 
+ALTER TABLE ONLY dast_profile_schedules ALTER COLUMN id SET DEFAULT nextval('dast_profile_schedules_id_seq'::regclass);
+
 ALTER TABLE ONLY dast_profiles ALTER COLUMN id SET DEFAULT nextval('dast_profiles_id_seq'::regclass);
 
 ALTER TABLE ONLY dast_scanner_profiles ALTER COLUMN id SET DEFAULT nextval('dast_scanner_profiles_id_seq'::regclass);
@@ -21295,6 +21321,9 @@ ALTER TABLE ONLY csv_issue_imports
 ALTER TABLE ONLY custom_emoji
     ADD CONSTRAINT custom_emoji_pkey PRIMARY KEY (id);
 
+ALTER TABLE ONLY dast_profile_schedules
+    ADD CONSTRAINT dast_profile_schedules_pkey PRIMARY KEY (id);
+
 ALTER TABLE ONLY dast_profiles_pipelines
     ADD CONSTRAINT dast_profiles_pipelines_pkey PRIMARY KEY (dast_profile_id, ci_pipeline_id);
 
@@ -23386,6 +23415,14 @@ CREATE UNIQUE INDEX index_custom_emoji_on_namespace_id_and_name ON custom_emoji
 
 CREATE UNIQUE INDEX index_daily_build_group_report_results_unique_columns ON ci_daily_build_group_report_results USING btree (project_id, ref_path, date, group_name);
 
+CREATE INDEX index_dast_profile_schedules_active_next_run_at ON dast_profile_schedules USING btree (active, next_run_at);
+
+CREATE INDEX index_dast_profile_schedules_on_dast_profile_id ON dast_profile_schedules USING btree (dast_profile_id);
+
+CREATE UNIQUE INDEX index_dast_profile_schedules_on_project_id_and_dast_profile_id ON dast_profile_schedules USING btree (project_id, dast_profile_id);
+
+CREATE INDEX index_dast_profile_schedules_on_user_id ON dast_profile_schedules USING btree (user_id);
+
 CREATE INDEX index_dast_profiles_on_dast_scanner_profile_id ON dast_profiles USING btree (dast_scanner_profile_id);
 
 CREATE INDEX index_dast_profiles_on_dast_site_profile_id ON dast_profiles USING btree (dast_site_profile_id);
@@ -26001,6 +26038,9 @@ ALTER TABLE ONLY project_access_tokens
 ALTER TABLE ONLY merge_requests
     ADD CONSTRAINT fk_6149611a04 FOREIGN KEY (assignee_id) REFERENCES users(id) ON DELETE SET NULL;
 
+ALTER TABLE ONLY dast_profile_schedules
+    ADD CONSTRAINT fk_61d52aa0e7 FOREIGN KEY (dast_profile_id) REFERENCES dast_profiles(id) ON DELETE CASCADE;
+
 ALTER TABLE ONLY events
     ADD CONSTRAINT fk_61fbf6ca48 FOREIGN KEY (group_id) REFERENCES namespaces(id) ON DELETE CASCADE;
 
@@ -26022,6 +26062,9 @@ ALTER TABLE ONLY merge_requests
 ALTER TABLE ONLY geo_event_log
     ADD CONSTRAINT fk_6ada82d42a FOREIGN KEY (container_repository_updated_event_id) REFERENCES geo_container_repository_updated_events(id) ON DELETE CASCADE;
 
+ALTER TABLE ONLY dast_profile_schedules
+    ADD CONSTRAINT fk_6cca0d8800 FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE;
+
 ALTER TABLE ONLY projects
     ADD CONSTRAINT fk_6e5c14658a FOREIGN KEY (pool_repository_id) REFERENCES pool_repositories(id) ON DELETE SET NULL;
 
@@ -26259,6 +26302,9 @@ ALTER TABLE ONLY ci_variables
 ALTER TABLE ONLY merge_request_metrics
     ADD CONSTRAINT fk_ae440388cc FOREIGN KEY (latest_closed_by_id) REFERENCES users(id) ON DELETE SET NULL;
 
+ALTER TABLE ONLY dast_profile_schedules
+    ADD CONSTRAINT fk_aef03d62e5 FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL;
+
 ALTER TABLE ONLY analytics_cycle_analytics_group_stages
     ADD CONSTRAINT fk_analytics_cycle_analytics_group_stages_group_value_stream_id FOREIGN KEY (group_value_stream_id) REFERENCES analytics_cycle_analytics_group_value_streams(id) ON DELETE CASCADE;
 

ee/app/models/dast/profile.rb

@@ -13,6 +13,8 @@ module Dast
     has_many :dast_profiles_pipelines, class_name: 'Dast::ProfilesPipeline', foreign_key: :dast_profile_id, inverse_of: :dast_profile
     has_many :ci_pipelines, class_name: 'Ci::Pipeline', through: :dast_profiles_pipelines
 
+    has_many :dast_profile_schedules, class_name: 'Dast::ProfileSchedule', foreign_key: :dast_profile_id, inverse_of: :dast_profile
+
     validates :description, length: { maximum: 255 }
     validates :name, length: { maximum: 255 }, uniqueness: { scope: :project_id }, presence: true
     validates :branch_name, length: { maximum: 255 }

ee/app/models/dast/profile_schedule.rb

+# frozen_string_literal: true
+
+class Dast::ProfileSchedule < ApplicationRecord
+  include CronSchedulable
+
+  self.table_name = 'dast_profile_schedules'
+
+  belongs_to :project
+  belongs_to :dast_profile, class_name: 'Dast::Profile', optional: false, inverse_of: :dast_profile_schedules
+  belongs_to :owner, class_name: 'User', optional: true, foreign_key: :user_id
+
+  validates :cron, presence: true
+  validates :next_run_at, presence: true
+
+  scope :with_project, -> { includes(:project) }
+  scope :with_profile, -> { includes(dast_profile: [:dast_site_profile, :dast_scanner_profile]) }
+  scope :with_owner, -> { includes(:owner) }
+  scope :active, -> { where(active: true) }
+
+  private
+
+  def cron_timezone
+    next_run_at.zone
+  end
+
+  def worker_cron_expression
+    Settings.cron_jobs['app_sec_dast_profile_schedule_worker']['cron']
+  end
+end

ee/app/workers/all_queues.yml

@@ -50,6 +50,15 @@
   :idempotent: true
   :tags:
   - :exclude_from_kubernetes
+- :name: cronjob:app_sec_dast_profile_schedule
+  :worker_name: AppSec::Dast::ProfileScheduleWorker
+  :feature_category: :dynamic_application_security_testing
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: cronjob:clear_shared_runners_minutes
   :worker_name: ClearSharedRunnersMinutesWorker
   :feature_category: :continuous_integration

ee/app/workers/app_sec/dast/profile_schedule_worker.rb

+# frozen_string_literal: true
+
+module AppSec
+  module Dast
+    class ProfileScheduleWorker
+      include ApplicationWorker
+      include CronjobQueue
+
+      deduplicate :until_executed, including_scheduled: true
+      idempotent!
+
+      feature_category :dynamic_application_security_testing
+
+      data_consistency :always
+
+      def perform
+        return unless Feature.enabled?(:dast_on_demand_scans_scheduler, default_enabled: :yaml)
+
+        dast_runnable_schedules.find_in_batches do |schedules|
+          schedules.each do |schedule|
+            with_context(project: schedule.project, user: schedule.owner) do
+              schedule.schedule_next_run!
+
+              response = service(schedule).execute
+              if response.error?
+                logger.info(structured_payload(message: response.message))
+              end
+            end
+          end
+        end
+      end
+
+      private
+
+      def dast_runnable_schedules
+        ::Dast::ProfileSchedule.with_project.with_profile.with_owner.runnable_schedules
+      end
+
+      def service(schedule)
+        ::DastOnDemandScans::CreateService.new(
+          container: schedule.project,
+          current_user: schedule.owner,
+          params: {
+            dast_site_profile: schedule.dast_profile.dast_site_profile,
+            dast_scanner_profile: schedule.dast_profile.dast_scanner_profile
+          }
+        )
+      end
+    end
+  end
+end

ee/config/feature_flags/development/dast_on_demand_scans_scheduler.yml

+---
+name: dast_on_demand_scans_scheduler
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/65327
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/328749
+milestone: '14.1'
+type: development
+group: group::dynamic analysis
+default_enabled: false

ee/spec/factories/dast/profile_schedules.rb

+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :dast_profile_schedule, class: 'Dast::ProfileSchedule' do
+    project
+    dast_profile
+    owner { association(:user) }
+    cron { '*/10 * * * *' }
+    next_run_at { Time.now }
+  end
+end

ee/spec/models/dast/profile_schedule_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Dast::ProfileSchedule, type: :model do
+  subject { create(:dast_profile_schedule) }
+
+  describe 'associations' do
+    it { is_expected.to belong_to(:project) }
+    it { is_expected.to belong_to(:dast_profile).class_name('Dast::Profile').required.inverse_of(:dast_profile_schedules) }
+    it { is_expected.to belong_to(:owner).class_name('User').with_foreign_key(:user_id) }
+  end
+
+  describe 'validations' do
+    it { is_expected.to be_valid }
+    it { is_expected.to validate_presence_of(:cron) }
+    it { is_expected.to validate_presence_of(:next_run_at) }
+  end
+
+  describe 'scopes' do
+    describe 'active' do
+      it 'includes the correct records' do
+        inactive_dast_profile_schedule = create(:dast_profile_schedule, active: false)
+
+        result = described_class.active
+
+        aggregate_failures do
+          expect(result).to include(subject)
+          expect(result).not_to include(inactive_dast_profile_schedule)
+        end
+      end
+    end
+
+    describe 'runnable_schedules' do
+      subject { described_class.runnable_schedules }
+
+      context 'when there are runnable schedules' do
+        let!(:profile_schedule) do
+          travel_to(1.day.ago) do
+            create(:dast_profile_schedule)
+          end
+        end
+
+        it 'returns the runnable schedule' do
+          is_expected.to eq([profile_schedule])
+        end
+      end
+
+      context 'when there are inactive schedules' do
+        let!(:profile_schedule) do
+          travel_to(1.day.ago) do
+            create(:dast_profile_schedule, active: false)
+          end
+        end
+
+        it 'returns an empty array' do
+          is_expected.to be_empty
+        end
+      end
+
+      context 'when there are no runnable schedules' do
+        let!(:profile_schedule) { }
+
+        it 'returns an empty array' do
+          is_expected.to be_empty
+        end
+      end
+
+      context 'when there are runnable schedules in future' do
+        let!(:profile_schedule) do
+          travel_to(1.day.from_now) do
+            create(:dast_profile_schedule)
+          end
+        end
+
+        it 'returns an empty array' do
+          is_expected.to be_empty
+        end
+      end
+    end
+  end
+
+  describe '#set_next_run_at' do
+    it_behaves_like 'handles set_next_run_at' do
+      let(:schedule) { create(:dast_profile_schedule, cron: '*/1 * * * *') }
+      let(:schedule_1) { create(:dast_profile_schedule) }
+      let(:schedule_2) { create(:dast_profile_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/models/dast/profile_spec.rb

@@ -14,6 +14,7 @@ RSpec.describe Dast::Profile, type: :model do
     it { is_expected.to have_many(:secret_variables).through(:dast_site_profile).class_name('Dast::SiteProfileSecretVariable') }
     it { is_expected.to have_many(:dast_profiles_pipelines).class_name('Dast::ProfilesPipeline').with_foreign_key(:dast_profile_id).inverse_of(:dast_profile) }
     it { is_expected.to have_many(:ci_pipelines).through(:dast_profiles_pipelines).class_name('Ci::Pipeline') }
+    it { is_expected.to have_many(:dast_profile_schedules).class_name('Dast::ProfileSchedule').with_foreign_key(:dast_profile_id).inverse_of(:dast_profile) }
   end
 
   describe 'validations' do

ee/spec/workers/app_sec/dast/profile_schedule_worker_spec.rb

+# frozen_string_literal: true
+require 'spec_helper'
+
+RSpec.describe AppSec::Dast::ProfileScheduleWorker do
+  include ExclusiveLeaseHelpers
+
+  let_it_be(:schedule) { create(:dast_profile_schedule) }
+
+  let(:worker) { described_class.new }
+  let(:logger) { worker.send(:logger) }
+  let(:service) { instance_double(::DastOnDemandScans::CreateService) }
+  let(:service_result) { ServiceResponse.success }
+
+  before do
+    allow(::DastOnDemandScans::CreateService)
+      .to receive(:new)
+      .and_return(service)
+    allow(service).to receive(:execute)
+      .and_return(service_result)
+  end
+
+  describe '#perform' do
+    subject { worker.perform }
+
+    context 'when feature flag is disabled' do
+      before do
+        stub_feature_flags(dast_on_demand_scans_scheduler: false)
+      end
+
+      it 'does not call runnable_schedules' do
+        expect(::Dast::ProfileSchedule).not_to receive(:runnable_schedules)
+        subject
+      end
+    end
+
+    context 'when multiple schedules exists' do
+      before do
+        schedule.update_column(:next_run_at, 1.minute.from_now)
+      end
+
+      def record_preloaded_queries
+        recorder = ActiveRecord::QueryRecorder.new { subject }
+        recorder.data.values.flat_map {|v| v[:occurrences]}.select do |query|
+          ['FROM "projects"', 'FROM "users"', 'FROM "dast_profile"', 'FROM "dast_profile_schedule"'].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(:dast_profile_schedule, 5) }
+        actual_count = record_preloaded_queries.count
+
+        expect(actual_count).to eq(expected_count)
+      end
+    end
+
+    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_found_instance_of(::Dast::ProfileSchedule) do |schedule|
+          expect(schedule).to receive(:schedule_next_run!)
+        end
+
+        expect(service).to receive(:execute)
+
+        subject
+      end
+    end
+
+    context 'when service returns an error' do
+      before do
+        schedule.update_column(:next_run_at, 1.minute.ago)
+      end
+
+      let(:error_message) { 'some message' }
+      let(:service_result) { ServiceResponse.error(message: error_message) }
+
+      it 'succeeds and logs the error' do
+        expect(logger)
+          .to receive(:info)
+          .with(a_hash_including('message' => error_message))
+
+        subject
+      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(::DastOnDemandScans::CreateService).not_to receive(:new)
+
+        subject
+      end
+    end
+  end
+end