代码拉取完成，页面将自动刷新
# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: broker\__init__.py
import time, uuid, logging, datetime, dateutil.rrule, copy
from dateutil.tz import tzutc
from dateutil.tz import tzlocal
import iso8601
from collections import defaultdict
from sqlalchemy import select as sql_select
from sqlalchemy import func as sql_func
from sqlalchemy import or_ as sql_or
from sqlalchemy import and_ as sql_and
from sqlalchemy import not_ as sql_not
from sqlalchemy import tuple_ as sql_tuple
from db import Connection
from db.tables.scan_sessions import ScanSessionRow, ScanSessionsTable
from db.tables.scans import ScanRow, ScansTable
from db.tables.targets import (
TargetRow,
TargetsTable,
TargetTypes,
TargetVerificationTypes,
)
from db.tables.profiles import ProfileRow, ProfilesTable
from db.tables.scan_session_jobs import ScanSessionJobRow, ScanSessionJobsTable
from db.tables.reports import ReportsTable
from db.tables.targets_allowed import TargetsAllowedRow, TargetsAllowedTable
from db.tables.targets_configurations import (
TargetConfigurationRow,
TargetConfigurationTable,
)
import broker.scan_session_job_helpers as job_helpers
from broker.scan_session_job_helpers import JobConstants, AbortReason
from broker.worker_manager.base import WorkerManager
from db.tables.events import create_event, EventResources
from db.views.targets_excluded_hours import (
TargetsWithExcludedHoursViewRow,
TargetsWithExcludedHoursViewTable,
)
from helpers.address import extract_domain_from_address
from helpers.licensing.features import (
Features,
is_report_id_compliance_id,
is_report_id_waf_export_id,
)
from helpers.updater import Updater
from db.tables.scan_session_vulns_stats import (
ScanSessionVulnsStatsRow,
ScanSessionVulnsStatsTable,
)
from helpers.excluded_hours import (
is_allowed_to_scan_now,
get_system_excluded_hours,
get_per_account_excluded_hours,
)
from scanners.scan_app.errors import *
from helpers.future_events.base import FutureEvents
from helpers.setup_types import *
from db.data.users import prepare_user_data, prepare_user_data_ext, get_owner_id
from db.tables.users import UserConfirmation
from helpers.target.tasks import do_check_target
from settings_provider import settings
from db.tables.events import EventRow
from helpers.constants.scans import *
from helpers.constants.jobs import ScanAppTypes
from helpers.licensing.features import BaseLicense
logger = logging.getLogger("service.broker")
lock_to_owner_id = None
def should_suppress_scan_starting():
if is_setup_type_on_premise():
user_license = BaseLicense.get_system_license()
if not user_license.is_active_license():
return True
system_update_object = settings.get("system_update_object", None)
if system_update_object:
if system_update_object.update_status in (
"waiting_for_scans",
"updating",
"restarting",
):
return True
return False
class Broker:
def __init__(
self,
shard_name,
worker_manager,
continuous_profile1=None,
continuous_profile2=None,
manual_browsing_storage=None,
scan_session_events=None,
max_scan_time=None,
max_job_preparation_time=datetime.timedelta(minutes=30),
max_job_abort_time=datetime.timedelta(minutes=30),
):
"""
:param shard_name:
"""
self.worker_manager = worker_manager
self.shard_name = shard_name
self.should_stop = False
self.profiles_cache = dict()
self.continuous_profile2 = continuous_profile2
self.continuous_profile1 = continuous_profile1
self.first_run = True
self.manual_browsing_storage = manual_browsing_storage
self.session_events = scan_session_events
self.max_scan_time = max_scan_time
self.max_job_preparation_time = max_job_preparation_time
self.max_job_abort_time = max_job_abort_time
def shutdown(self):
logger.debug("broker is shutting down")
self.should_stop = True
def loop(self):
"""
The broker does different task in a loop, each task should have different frequency
:return:
"""
self.should_stop = False
while not self.should_stop:
try:
self.abort_old_paused_jobs()
except Exception as e:
logger.exception("abort_old_paused_jobs:%s", e)
try:
self.restart_auto_paused_jobs()
except Exception as e:
logger.exception("restart_auto_paused_jobs:%s", e)
try:
self.worker_manager.process_queued_jobs()
except Exception as e:
logger.exception("worker_collection.process_queued_jobs:%s", e)
try:
if not should_suppress_scan_starting():
self.process_next_run()
except Exception as e:
logger.exception("process_next_run:%s", e)
try:
if not should_suppress_scan_starting():
self.start_queued_scan_session_jobs()
except Exception as e:
logger.exception("process_starting_jobs:%s", e)
try:
self.process_scan_session_jobs_in_progress()
self.first_run = False
except Exception as e:
logger.exception("process_about_to_abort_jobs:%s", e)
try:
self.process_finished_jobs()
except Exception as e:
logger.exception("process_finished_jobs:%s", e)
try:
self.worker_manager.workers_maintenance()
except Exception as e:
logger.exception("workers_maintenance:%s", e)
time.sleep(1)
try:
self.worker_manager.shutdown()
except Exception as e:
logger.exception("worker_manager.shut_down failed with %s", e)
scan_status_cache = settings.get("scan_status_cache")
if scan_status_cache:
try:
scan_status_cache.save_all(self.shard_name)
except Exception as e:
logger.exception("scan_status_cache.save_all failed with %s", e)
logger.info("Stop asked, broker is shutting down, bye.")
def __schedule_job_recheck(self, scan_session_job, time_delta):
if self.session_events:
self.session_events.add(
"%(scan_session_id)s:%(scanning_app)s" % scan_session_job,
time_delta,
namespace=self.shard_name,
)
def __remove_job_recheck_schedules(self, scheduled_job_check_events):
if self.session_events:
if scheduled_job_check_events:
self.session_events.delete_events(scheduled_job_check_events)
def __get_job_recheck_schedules(self, max_scheduled_events=20):
return self.session_events.get_many(
max_scheduled_events, namespace=self.shard_name
)
@property
def has_scheduled_job_checks_support(self):
return self.session_events is not None
def __create_scan_job_event(
self,
scan_session_job,
event_name,
job_status,
extended_job_status=None,
worker_id=None,
):
event_data = dict(
name=event_name,
owner_id=scan_session_job["owner_id"],
user_id=scan_session_job["creator_id"],
resource_type=EventResources.scan_session,
resource_id=scan_session_job["scan_session_id"],
data=dict(
scanning_app=scan_session_job["scanning_app"],
status=job_status,
extended_status=extended_job_status,
),
shard=self.shard_name,
)
if worker_id:
event_data["data"]["worker_id"] = worker_id
event = create_event(**event_data)
with Connection(self.shard_name) as (con):
con.execute(event)
def __cleanup_job(self, scan_session_job):
try:
self.worker_manager.delete_job(
scan_session_job["scan_session_id"], scan_session_job["scanning_app"]
)
except Exception:
logger.exception(
"delete job failed [%s:%s]",
scan_session_job["scan_session_id"],
scan_session_job["scanning_app"],
)
scan_status_cache = settings.get("scan_status_cache")
if scan_status_cache:
scan_status_cache.save_entry(
self.shard_name,
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
)
scan_status_cache.del_entry(
self.shard_name,
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
)
manual_browsing_storage = settings.get("manual_browsing_storage")
if manual_browsing_storage:
manual_browsing_storage.remove_item(
scan_session_job["scan_session_id"], scan_session_job["scanning_app"]
)
def __mark_job_failed(
self, scan_session_job, reason, scan_result=None, start_date=None
):
logger.debug(
"mark_job_failed %s:%s %s",
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
reason,
)
job_update_data = dict(
status=ScanStatusTypes.FAILED,
end_date=sql_func.current_timestamp(),
last_update=sql_func.current_timestamp(),
status_data=reason,
)
if start_date is not None:
job_update_data["start_date"] = start_date
with Connection(self.shard_name) as (db):
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app == scan_session_job["scanning_app"]
)
.where(ScanSessionJobRow.status.in_(list(ActiveScanTypesExt)))
.where(ScanSessionJobRow.result_processed.is_(False))
)
db.execute(q)
if scan_result:
extended_status = scan_result.get("extended_status")
if extended_status:
extended_status = extended_status.to_dict()
self._Broker__create_scan_job_event(
scan_session_job, "scan_job_failed", scan_result.status, extended_status
)
else:
self._Broker__create_scan_job_event(
scan_session_job,
"scan_job_failed",
"failed",
dict(fail_reason=reason["reason"], attachments=[]),
)
self._Broker__cleanup_job(scan_session_job)
def __mark_job_completed(self, scan_session_job, scan_result):
logger.debug(
"mark_job_completed %s:%s",
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
)
job_update_data = dict(
status="completed",
end_date=sql_func.current_timestamp(),
last_update=sql_func.current_timestamp(),
status_data=None,
)
with Connection(self.shard_name) as (db):
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app == scan_session_job["scanning_app"]
)
.where(ScanSessionJobRow.status.in_(list(ActiveScanTypes)))
.where(ScanSessionJobRow.result_processed.is_(False))
)
db.execute(q)
scan_session_job_stats = None
scan_status_cache = settings.get("scan_status_cache")
if scan_status_cache:
scan_session_job_stats = scan_status_cache.get_entry(
self.shard_name,
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
)
if scan_session_job_stats:
scan_session_job_stats = copy.deepcopy(scan_session_job_stats)
scan_status_cache.update_main_keys(
self.shard_name,
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
dict(progress=100),
)
targets_touched_by_scan_job = [scan_session_job["target_id"]]
if scan_session_job_stats:
for key in scan_session_job_stats.get("hosts", {}):
if key not in targets_touched_by_scan_job:
targets_touched_by_scan_job.append(key)
with Connection(self.shard_name) as (db):
q = (
(
TargetsTable.update(
values=dict(
last_scan_session_id=scan_session_job["scan_session_id"],
last_scan_id=scan_session_job["scan_id"],
)
)
)
.where(TargetRow.target_id.in_(targets_touched_by_scan_job))
.where(
sql_or(
TargetRow.last_scan_session_id
!= scan_session_job["scan_session_id"],
TargetRow.last_scan_session_id.is_(None),
)
)
)
db.execute(q)
extended_status = scan_result.get("extended_status")
if extended_status:
extended_status = extended_status.to_dict()
try:
self.worker_manager.download_scan_state_db_from_worker(scan_session_job)
except Exception:
logger.exception(
"download_scan_state_db_from_worker failed for %s",
scan_session_job["scan_session_id"],
)
self._Broker__create_scan_job_event(
scan_session_job, "scan_job_done", scan_result.status, extended_status
)
self._Broker__cleanup_job(scan_session_job)
def __mark_job_aborted(
self, scan_session_job, abort_reason=None, extended_status=None
):
if abort_reason is None:
abort_reason = scan_session_job["status_data"]
logger.debug(
"marking job aborted %s:%s %s",
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
abort_reason,
)
job_update_data = dict(
status=ScanStatusTypes.ABORTED,
end_date=sql_func.current_timestamp(),
last_update=sql_func.current_timestamp(),
)
if abort_reason:
job_update_data["status_data"] = abort_reason
with Connection(self.shard_name) as (db):
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app == scan_session_job["scanning_app"]
)
.where(ScanSessionJobRow.status.in_(list(ActiveScanTypesExt)))
.where(ScanSessionJobRow.result_processed.is_(False))
)
db.execute(q)
self._Broker__cleanup_job(scan_session_job)
self._Broker__create_scan_job_event(
scan_session_job,
"scan_job_aborted",
abort_reason,
extended_job_status=extended_status,
)
def __mark_job_paused(self, scan_session_job, scan_result):
logger.debug(
"marking job paused %s:%s",
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
)
job_update_data = dict(
status=ScanStatusTypes.PAUSED,
end_date=sql_func.current_timestamp(),
last_update=sql_func.current_timestamp(),
)
with Connection(self.shard_name) as (db):
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app == scan_session_job["scanning_app"]
)
.where(ScanSessionJobRow.status.in_(list(ActiveScanTypesExt)))
.where(ScanSessionJobRow.result_processed.is_(False))
)
db.execute(q)
extended_status = scan_result.get("extended_status")
if extended_status:
extended_status = extended_status.to_dict()
try:
self.worker_manager.download_scan_state_db_from_worker(scan_session_job)
except Exception:
logger.exception(
"download_scan_state_db_from_worker failed for %s",
scan_session_job["scan_session_id"],
)
self._Broker__cleanup_job(scan_session_job)
self._Broker__create_scan_job_event(
scan_session_job, "scan_job_paused", scan_result.status, extended_status
)
def __mark_job_executing(self, scan_session_job):
with Connection(self.shard_name) as (db):
job_update_data = dict(
last_update=sql_func.current_timestamp(),
status=ScanStatusTypes.PROCESSING,
)
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app == scan_session_job["scanning_app"]
)
.where(
ScanSessionJobRow.status.in_(
(
ScanStatusTypes.QUEUED,
ScanStatusTypes.STARTING,
ScanStatusTypes.RESUMING,
)
)
)
.where(ScanSessionJobRow.result_processed.is_(False))
)
result = db.execute(q)
if result.rowcount != 0:
q = (
(
ScanSessionsTable.update(
values=dict(status=ScanStatusTypes.PROCESSING)
)
)
.where(
ScanSessionRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionRow.status.in_(
(
ScanStatusTypes.QUEUED,
ScanStatusTypes.STARTING,
ScanStatusTypes.RESUMING,
)
)
)
)
db.execute(q)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.processing
)
def __abort_job(self, scan_session_job, abort_reason, update_status=True):
"""
Scan can be aborted in
:param scan_session_job:
:param abort_reason:
:param update_status: means we don't change scan status => aborting/aborted, the caller should handle this
:return:
"""
if scan_session_job["status"] not in (
ScanStatusTypes.STARTING,
ScanStatusTypes.PROCESSING,
ScanStatusTypes.PAUSING,
ScanStatusTypes.RESUMING,
):
if update_status:
return
logger.debug("should abort because %s", abort_reason)
try:
abort_success = self.worker_manager.abort_job(
scan_session_job["scan_session_id"],
scan_session_job["scanning_app"],
abort_reason,
)
if abort_success:
print("instant abort")
except Exception as e:
logger.exception("self.worker_manager.abort_job failed with %s", e)
if not update_status:
return
with Connection(self.shard_name) as (db):
job_update_data = dict(
status=ScanStatusTypes.ABORTING, status_data=abort_reason
)
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app
== scan_session_job["scanning_app"]
)
.where(
ScanSessionJobRow.status.in_(
(
ScanStatusTypes.STARTING,
ScanStatusTypes.PROCESSING,
ScanStatusTypes.ABORTING,
)
)
)
.where(ScanSessionJobRow.result_processed.is_(False))
)
db.execute(q)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.aborting
)
def __pause_job(self, scan_session_job, pause_reason):
"""
Scan can be aborted in
:param scan_session_job:
:param pause_reason:
:return:
"""
if scan_session_job["status"] not in (
ScanStatusTypes.STARTING,
ScanStatusTypes.PROCESSING,
ScanStatusTypes.PAUSING,
ScanStatusTypes.RESUMING,
):
return
logger.debug("should pause because %s", pause_reason)
try:
self.worker_manager.pause_job(
scan_session_job["scan_session_id"],
scan_session_job["scanning_app"],
pause_reason,
)
except Exception as e:
logger.exception("self.worker_manager.abort_job failed with %s", e)
with Connection(self.shard_name) as (db):
job_update_data = dict(
status=ScanStatusTypes.PAUSING, status_data=pause_reason
)
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app == scan_session_job["scanning_app"]
)
.where(
ScanSessionJobRow.status.in_(
(
ScanStatusTypes.STARTING,
ScanStatusTypes.PROCESSING,
ScanStatusTypes.ABORTING,
)
)
)
.where(ScanSessionJobRow.result_processed.is_(False))
)
db.execute(q)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.aborting
)
def restart_auto_paused_jobs(self):
"""
Searches any paused jobs with reason => ScanPauseReasonTypes.EXCLUDED_HOURS
:return:
"""
sf = ScanSessionJobsTable.join(
ScanSessionsTable,
ScanSessionJobRow.scan_session_id == ScanSessionRow.scan_session_id,
)
sf = sf.join(ScansTable, ScanSessionRow.scan_id == ScanRow.scan_id)
sf = sf.join(
TargetsWithExcludedHoursViewTable,
TargetsWithExcludedHoursViewRow.target_id == ScanRow.target_id,
)
base_query = (
sql_select(
(
ScanSessionJobRow.pause_requested,
ScanSessionJobRow.scan_session_id,
ScanSessionJobRow.scanning_app,
ScanRow.owner_id,
ScanRow.creator_id,
TargetsWithExcludedHoursViewRow.address,
TargetsWithExcludedHoursViewRow.description,
TargetsWithExcludedHoursViewRow.xh_data,
TargetsWithExcludedHoursViewRow.xh_time_offset,
)
)
.select_from(sf)
.where(ScanSessionJobRow.status == ScanStatusTypes.PAUSED)
.where(
ScanSessionJobRow.pause_requested["reason"].astext
== ScanPauseReasonTypes.EXCLUDED_HOURS
)
.where(ScanRow.deleted_at.is_(None))
.where(TargetsWithExcludedHoursViewRow.deleted_at.is_(None))
)
cursor = 0
limit = 20
with Connection(self.shard_name) as (db):
system_excluded_hours = None
if not has_feature(SetupFeatures.PER_ACCOUNT_EXCLUDED_HOURS):
system_excluded_hours = get_system_excluded_hours(db)
per_account_excluded_hours_cache = dict()
while True:
if self.should_stop:
break
q = (
base_query.limit(limit)
.offset(cursor)
.order_by(
ScanSessionJobRow.scan_session_id,
ScanSessionJobRow.scanning_app,
)
)
c = 0
for job_row in db.execute(q).fetchall():
if has_feature(SetupFeatures.PER_ACCOUNT_EXCLUDED_HOURS):
system_excluded_hours = per_account_excluded_hours_cache.get(
job_row.owner_id, -1
)
if system_excluded_hours == -1:
system_excluded_hours = get_per_account_excluded_hours(
db, job_row.owner_id
)
per_account_excluded_hours_cache[
job_row.owner_id
] = system_excluded_hours
resume = False
if job_row.xh_data:
if is_allowed_to_scan_now(
job_row.xh_data, job_row.xh_time_offset
):
resume = True
else:
if system_excluded_hours:
if is_allowed_to_scan_now(**system_excluded_hours):
resume = True
else:
resume = True
if resume:
query = (
(
ScanSessionJobsTable.update(
values=dict(
status=ScanStatusTypes.RESUMING,
result_processed=False,
pause_requested=False,
start_deadline=datetime.datetime.now(
tzlocal()
)
+ (datetime.timedelta(minutes=15)),
)
)
)
.where(
ScanSessionJobRow.scan_session_id
== job_row.scan_session_id
)
.where(
ScanSessionJobRow.scanning_app
== job_row.scanning_app
)
)
db.execute(query)
query = (
ScanSessionsTable.update(
values=dict(status=ScanStatusTypes.RESUMING)
)
).where(
ScanSessionRow.scan_session_id
== job_row.scan_session_id
)
db.execute(query)
c += 1
if c < limit:
break
cursor += limit
def abort_old_paused_jobs(self):
"""
Searches for paused jobs older than X hours. Marks any such job as aborted and propagates the abort if
necessary, meaning that the scan still has this session as current session
:return:
"""
sf = ScanSessionJobsTable.join(
ScanSessionsTable,
ScanSessionJobRow.scan_session_id == ScanSessionRow.scan_session_id,
)
sf = sf.join(ScansTable, ScanSessionRow.scan_id == ScanRow.scan_id)
q = (
sql_select(
(
ScanSessionJobRow.pause_requested,
ScanSessionJobRow.scan_session_id,
ScanSessionJobRow.scanning_app,
ScanRow.owner_id,
ScanRow.creator_id,
)
)
.select_from(sf)
.where(ScanSessionJobRow.status == ScanStatusTypes.PAUSED)
)
with Connection(self.shard_name) as (db):
for scan_session_job in db.execute(q).fetchall():
if self.should_stop:
break
if scan_session_job.pause_requested and iso8601.parse_date(
scan_session_job.pause_requested["date"]
) + (
datetime.timedelta(
hours=settings.get("max_paused_state_hours", 168)
)
) < datetime.datetime.now(
tzutc()
):
logger.warning(
"aborting paused scan %s", scan_session_job.scan_session_id
)
job_update_data = dict(
status=ScanStatusTypes.ABORTED,
end_date=sql_func.current_timestamp(),
last_update=sql_func.current_timestamp(),
status_data=AbortReason.pause_expired,
result_processed=False,
)
q = (
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job.scan_session_id
)
.where(
ScanSessionJobRow.scanning_app
== scan_session_job.scanning_app
)
.where(ScanSessionJobRow.status == ScanStatusTypes.PAUSED)
)
db.execute(q)
self._Broker__create_scan_job_event(
dict(scan_session_job),
"scan_job_aborted",
AbortReason.pause_expired,
)
def process_scan_session_job_in_progress(self, scan_session_job):
"""
Checks the status of a specific scan_session_job_id / scanning_app
:param scan_session_job: job descriptor
:return:
"""
logger.debug("processing %s", scan_session_job)
if scan_session_job["status"] not in list(ActiveScanTypes):
return
abort_reason = None
if scan_session_job["status"] in ActiveScanTypes.difference(
{ScanStatusTypes.ABORTING}
):
pass
if scan_session_job["abort_requested"]:
abort_reason = AbortReason.abort_requested
else:
if scan_session_job["deleted_at"]:
abort_reason = AbortReason.scan_deleted
else:
if scan_session_job["end_deadline_exceeded"]:
abort_reason = AbortReason.scan_timeout
else:
if (
scan_session_job["xh_data"] is None
and scan_session_job["system_excluded_hours"]
and not (
is_allowed_to_scan_now(
**scan_session_job["system_excluded_hours"]
)
)
):
if scan_session_job["status"] in (
ScanStatusTypes.ABORTING,
ScanStatusTypes.PAUSING,
):
pass
elif not settings.get("pause_scan_on_excluded_hours"):
logging.debug(
"abort scan_session:%s:%s because system excluded hours [%s]",
scan_session_job["scan_session_id"],
scan_session_job["scanning_app"],
scan_session_job["system_excluded_hours"]["name"],
)
abort_reason = AbortReason.excluded_hours
else:
if settings.get("pause_scan_on_excluded_hours"):
logging.debug(
"pause scan_session:%s:%s because system excluded hours [%s]",
scan_session_job["scan_session_id"],
scan_session_job["scanning_app"],
scan_session_job["system_excluded_hours"]["name"],
)
with Connection(self.shard_name) as (db):
query = (
ScanSessionJobsTable.update(
values=dict(
pause_requested=dict(
date=datetime.datetime.now(tzutc()),
reason=ScanPauseReasonTypes.EXCLUDED_HOURS,
)
)
)
).where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
db.execute(query)
query = (
ScanSessionsTable.update(
values=dict(status="pausing")
)
).where(
ScanSessionRow.scan_session_id
== scan_session_job["scan_session_id"]
)
db.execute(query)
try:
self._Broker__pause_job(
scan_session_job,
ScanPauseReasonTypes.EXCLUDED_HOURS,
)
except (JobNotFound, WorkerNotFound) as e:
logger.warning(
"__pause_job, job lost %s", e.str_message
)
if (
scan_session_job["status"]
== ScanStatusTypes.ABORTING
):
self._Broker__mark_job_aborted(scan_session_job)
else:
self._Broker__mark_job_failed(
scan_session_job,
dict(reason=e.str_message, extra=None),
)
return
else:
if not is_allowed_to_scan_now(
scan_session_job["xh_data"],
scan_session_job["xh_time_offset"],
):
if scan_session_job["status"] in (
ScanStatusTypes.ABORTING,
ScanStatusTypes.PAUSING,
):
pass
elif not settings.get("pause_scan_on_excluded_hours"):
logging.debug(
"abort scan_session:%s:%s because target excluded hours",
scan_session_job["scan_session_id"],
scan_session_job["scanning_app"],
)
abort_reason = AbortReason.excluded_hours
else:
if settings.get("pause_scan_on_excluded_hours"):
logging.debug(
"pause scan_session:%s:%s because target excluded hours",
scan_session_job["scan_session_id"],
scan_session_job["scanning_app"],
)
with Connection(self.shard_name) as (db):
query = (
ScanSessionJobsTable.update(
values=dict(
pause_requested=dict(
date=datetime.datetime.now(
tzutc()
),
reason=ScanPauseReasonTypes.EXCLUDED_HOURS,
)
)
)
).where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
db.execute(query)
query = (
ScanSessionsTable.update(
values=dict(status="pausing")
)
).where(
ScanSessionRow.scan_session_id
== scan_session_job["scan_session_id"]
)
db.execute(query)
try:
self._Broker__pause_job(
scan_session_job,
ScanPauseReasonTypes.EXCLUDED_HOURS,
)
except (JobNotFound, WorkerNotFound) as e:
logger.warning(
"__pause_job, job lost %s", e.str_message
)
if (
scan_session_job["status"]
== ScanStatusTypes.ABORTING
):
self._Broker__mark_job_aborted(
scan_session_job
)
else:
self._Broker__mark_job_failed(
scan_session_job,
dict(reason=e.str_message, extra=None),
)
return
else:
abort_reason = None
if abort_reason is not None:
self._Broker__abort_job(scan_session_job, abort_reason)
return
if scan_session_job["status"] in ActiveScanTypes.difference(
{ScanStatusTypes.PAUSING, ScanStatusTypes.ABORTING}
):
if scan_session_job["pause_requested"]:
self._Broker__pause_job(
scan_session_job, scan_session_job["pause_requested"]["reason"]
)
return
try:
result = self.worker_manager.check_single_job_status(
scan_session_job["scan_session_id"], scan_session_job["scanning_app"]
)
except (JobNotFound, WorkerNotFound) as e:
logger.warning("check_single_job_status, job lost %s", e.str_message)
if scan_session_job["status"] == ScanStatusTypes.ABORTING:
self._Broker__mark_job_aborted(scan_session_job)
else:
self._Broker__mark_job_failed(
scan_session_job, dict(reason=e.str_message, extra=None)
)
return
else:
logger.debug("check_single_job_status %s", result.to_dict())
if result.status == ScanApiStatusTypes.FAILED:
self._Broker__mark_job_failed(
scan_session_job,
dict(reason=JobConstants.status_crashed, extra=result.to_dict()),
result,
)
return
if result.status == ScanApiStatusTypes.FINISHED:
self._Broker__mark_job_completed(scan_session_job, result)
return
if result.status == ScanApiStatusTypes.ABORTED:
self._Broker__mark_job_aborted(
scan_session_job, extended_status=result.get("extended_status")
)
return
if result.status == ScanApiStatusTypes.PAUSED:
self._Broker__mark_job_paused(scan_session_job, result)
return
if result.status == ScanApiStatusTypes.EXECUTING:
logger.debug(
"mark_job_processing %s:%s",
scan_session_job["scanning_app"],
scan_session_job["scan_session_id"],
)
self._Broker__mark_job_executing(scan_session_job)
else:
if result.status == ScanApiStatusTypes.PREPARING:
if self.max_job_preparation_time is not None:
if (
scan_session_job["last_update_delta"]
> self.max_job_preparation_time
):
logger.error(
"job %s failed to start within %s, aborting",
scan_session_job,
self.max_job_preparation_time,
)
self._Broker__abort_job(
scan_session_job,
AbortReason.start_timeout,
update_status=False,
)
self._Broker__mark_job_failed(
scan_session_job, dict(reason=AbortReason.start_timeout)
)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.starting
)
else:
if result.status == ScanApiStatusTypes.ABORTING:
if self.max_job_abort_time is not None:
if (
scan_session_job["last_update_delta"]
> self.max_job_abort_time
):
logger.error(
"aborting job %s takes more than %s!!",
scan_session_job,
self.max_job_abort_time,
)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.aborting
)
else:
if result.status == ScanApiStatusTypes.PAUSING:
if self.max_job_abort_time is not None:
if (
scan_session_job["last_update_delta"]
> self.max_job_abort_time
):
logger.error(
"pausing job %s takes more than %s!!",
scan_session_job,
self.max_job_abort_time,
)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.pausing
)
else:
logger.debug("unknown status %s", result.status)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.default
)
def process_scan_session_jobs_in_progress(self):
select_from = ScanSessionJobsTable.join(
ScanSessionsTable,
ScanSessionJobRow.scan_session_id == ScanSessionRow.scan_session_id,
)
select_from = select_from.join(
ScansTable, ScanRow.scan_id == ScanSessionRow.scan_id
)
select_from = select_from.join(
TargetsWithExcludedHoursViewTable,
TargetsWithExcludedHoursViewRow.target_id == ScanRow.target_id,
)
sql_fetch_processing_jobs = (
sql_select(
(
ScanSessionJobRow,
ScanSessionRow.deleted_at,
ScanRow.owner_id,
ScanRow.creator_id,
ScanRow.scan_id,
ScanRow.target_id,
TargetsWithExcludedHoursViewRow.address,
TargetsWithExcludedHoursViewRow.description,
TargetsWithExcludedHoursViewRow.xh_data,
TargetsWithExcludedHoursViewRow.xh_time_offset,
(
sql_func.current_timestamp() > ScanSessionJobRow.end_deadline
).label("end_deadline_exceeded"),
(
sql_func.current_timestamp() - ScanSessionJobRow.last_update
).label("last_update_delta"),
)
)
.select_from(select_from)
.where(ScanSessionJobRow.result_processed.isnot(True))
.where(
ScanSessionJobRow.status.in_(
(
ScanStatusTypes.STARTING,
ScanStatusTypes.PROCESSING,
ScanStatusTypes.ABORTING,
ScanApiStatusTypes.PAUSING,
)
)
)
)
if self.first_run or not self.has_scheduled_job_checks_support:
scheduled_job_check_events = None
else:
scheduled_job_check_events = self._Broker__get_job_recheck_schedules()
items_in_work = [
x["event_id"].split(":") for x in scheduled_job_check_events
]
conditions = [
sql_and(
ScanSessionJobRow.status.in_(
(
ScanStatusTypes.STARTING,
ScanStatusTypes.PROCESSING,
ScanStatusTypes.RESUMING,
ScanApiStatusTypes.PAUSING,
)
),
ScanSessionJobRow.abort_requested.is_(True),
),
sql_and(
ScanSessionJobRow.status.in_(
(
ScanStatusTypes.STARTING,
ScanStatusTypes.PROCESSING,
ScanStatusTypes.RESUMING,
ScanApiStatusTypes.PAUSING,
)
),
sql_not(ScanSessionRow.deleted_at.is_(None)),
),
sql_and(
ScanSessionJobRow.last_update
< sql_func.current_timestamp() - (datetime.timedelta(hours=1))
),
]
if items_in_work:
conditions.append(
sql_tuple(
ScanSessionJobRow.scan_session_id,
ScanSessionJobRow.scanning_app,
).in_(items_in_work)
)
sql_fetch_processing_jobs = sql_fetch_processing_jobs.where(
sql_or(*conditions)
)
scan_session_jobs = []
try:
with Connection(self.shard_name) as (db):
system_excluded_hours = None
if not has_feature(SetupFeatures.PER_ACCOUNT_EXCLUDED_HOURS):
system_excluded_hours = get_system_excluded_hours(db)
jobs = db.execute(sql_fetch_processing_jobs).fetchall()
for scan_session_job in jobs:
scan_session_job = dict(scan_session_job)
if has_feature(SetupFeatures.PER_ACCOUNT_EXCLUDED_HOURS):
system_excluded_hours = get_per_account_excluded_hours(
db, scan_session_job["owner_id"]
)
scan_session_job["system_excluded_hours"] = system_excluded_hours
scan_session_jobs.append(scan_session_job)
except Exception as e:
logger.exception("failed to pull jobs %s", e)
else:
self._Broker__remove_job_recheck_schedules(scheduled_job_check_events)
for scan_session_job in scan_session_jobs:
try:
self.process_scan_session_job_in_progress(scan_session_job)
except Exception as e:
logger.exception(
"process_scan_session_job_in_progress failed with %s", e
)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.on_error_recheck
)
def process_next_run(self):
"""
next_run < sql_func.now() means the scan has reached a state where a session should be created
Sessions can have different jobs (for now ovs and/or wvs)
:return:
"""
select_from = ScansTable.join(
ProfilesTable, ProfileRow.profile_id == ScanRow.profile_id
).join(
TargetsWithExcludedHoursViewTable,
TargetsWithExcludedHoursViewRow.target_id == ScanRow.target_id,
)
sql_get_scans = (
sql_select(
(
ScanRow,
ProfileRow.jobs,
TargetsWithExcludedHoursViewRow.manual_intervention,
TargetsWithExcludedHoursViewRow.xh_data,
TargetsWithExcludedHoursViewRow.xh_time_offset,
TargetsWithExcludedHoursViewRow.address,
TargetsWithExcludedHoursViewRow.description,
TargetsWithExcludedHoursViewRow.cah,
)
)
.select_from(select_from)
.where(ScanRow.next_run < sql_func.now())
.where(ScanRow.deleted_at.is_(None))
.where(ScanRow.schedule_disabled.isnot(True))
)
if lock_to_owner_id:
sql_get_scans = sql_get_scans.where(ScanRow.owner_id == lock_to_owner_id)
manual_browsing_storage_items = []
addresses_currently_scanned = []
with Connection(self.shard_name) as (db):
system_excluded_hours = None
if not has_feature(SetupFeatures.PER_ACCOUNT_EXCLUDED_HOURS):
system_excluded_hours = get_system_excluded_hours(db)
max_scans_per_license = 1
if is_setup_type_on_premise():
system_license = BaseLicense.get_system_license()
max_scans_per_license = system_license.max_scans_per_license
if not max_scans_per_license:
max_scans_per_license = 1
users_license_cache = dict()
for scan in db.execute(sql_get_scans).fetchall():
if should_suppress_scan_starting():
break
if is_setup_type_aws():
max_scans_per_license = users_license_cache.get(scan.owner_id)
if max_scans_per_license is None:
user_data = prepare_user_data(db, user_id=scan.owner_id)
if user_data is not None:
user_license = BaseLicense.get_license_from_license_data(
user_data["license_data"]
)
max_scans_per_license = user_license.max_scans_per_license
if not max_scans_per_license:
max_scans_per_license = 1
else:
users_license_cache[
scan.owner_id
] = max_scans_per_license
else:
max_scans_per_license = 1
sf = ScanSessionsTable.join(
ScansTable, ScanRow.scan_id == ScanSessionRow.scan_id
)
q = (
sql_select(
(sql_func.count(ScanSessionRow.scan_session_id),)
)
.select_from(sf)
.where(ScanRow.owner_id == scan.owner_id)
.where(ScanSessionRow.deleted_at.is_(None))
.where(
ScanSessionRow.status.in_(list(TargetsInWorkScanStatus))
)
)
concurrent_scans = db.execute(q).scalar()
if concurrent_scans > max_scans_per_license:
pass
else:
if has_feature(SetupFeatures.PER_ACCOUNT_EXCLUDED_HOURS):
system_excluded_hours = get_per_account_excluded_hours(
db, scan.owner_id
)
if not scan.xh_data:
pass
if system_excluded_hours:
if not (is_allowed_to_scan_now(**system_excluded_hours)):
logging.debug(
"postponing scan:[%s] by system excluded hours",
scan.scan_id,
)
continue
else:
if not is_allowed_to_scan_now(
scan.xh_data, scan.xh_time_offset
):
logging.debug(
"postponing scan:[%s] by target:[%s] excluded hours",
scan.scan_id,
scan.target_id,
)
continue
query = (
sql_select((TargetRow.address, TargetRow.target_id))
.select_from(
TargetsAllowedTable.join(
TargetsTable,
TargetRow.target_id
== TargetsAllowedRow.allowed_target_id,
)
)
.where(TargetsAllowedRow.target_id == scan.target_id)
)
allowed_targets = dict()
for at in db.execute(query):
allowed_targets[
extract_domain_from_address(at.address)
] = at.target_id
if not allowed_targets:
allowed_targets = None
if scan.continuous:
start_deadline = (
scan.next_run
+ self.worker_manager.START_DEADLINE_CONTINUOUS
)
r = dateutil.rrule.rrulestr(scan.recurrence)
use_anniversary_profile = (
r._dtstart.isocalendar()[2] % 7
== datetime.datetime.now(tzlocal()).isocalendar()[2]
)
used_profile_id = scan.profile_id
if use_anniversary_profile:
if self.continuous_profile1:
used_profile_id = self.continuous_profile1
else:
if self.continuous_profile2:
used_profile_id = self.continuous_profile2
else:
start_deadline = (
scan.next_run
+ self.worker_manager.get_start_deadline(
scan.schedule_start_date,
scan.schedule_time_sensitive,
scan.recurrence,
)
)
used_profile_id = scan.profile_id
scan_session_id = str(uuid.uuid4())
new_scan_session = dict(
scan_session_id=scan_session_id,
scan_id=scan.scan_id,
target_id=scan.target_id,
status=ScanStatusTypes.QUEUED,
scan_key=str(uuid.uuid4()),
start_expected=scan.next_run,
allowed_targets=allowed_targets,
used_profile_id=used_profile_id,
)
scan_update = dict(
current_session_id=scan_session_id,
previous_session_id=scan.previous_session_id,
next_run=None,
)
new_scan_session_query = ScanSessionsTable.insert().values(
**new_scan_session
)
db.execute(new_scan_session_query)
update_scan_session_id = (
ScansTable.update(values=scan_update)
).where(ScanRow.scan_id == scan.scan_id)
db.execute(update_scan_session_id)
for scanning_app in scan.jobs:
scan_session_job = dict(
scan_session_id=scan_session_id,
scanning_app=scanning_app,
status=ScanStatusTypes.QUEUED,
start_deadline=start_deadline,
end_deadline=None,
result_processed=False,
abort_requested=False,
)
new_job_query = ScanSessionJobsTable.insert().values(
**scan_session_job
)
db.execute(new_job_query)
self._Broker__handle_manual_browsing(
db,
scan,
scan_session_id,
manual_browsing_storage_items,
scanning_app,
)
addresses_currently_scanned.append(scan.cah)
if has_feature(SetupFeatures.MANUAL_BROWSING):
for manual_browsing_storage_item in manual_browsing_storage_items:
self.manual_browsing_storage.init_item(**manual_browsing_storage_item)
def __handle_manual_browsing(
self, db, scan, scan_session_id, manual_browsing_storage_items, scanning_app
):
"""
Helper method for process_next_run & manual browsing
:param db:
:param scan:
:param scan_session_id:
:param manual_browsing_storage_items:
:param scanning_app:
:return:
"""
if not has_feature(SetupFeatures.MANUAL_BROWSING):
return
if not (
scanning_app == ScanAppTypes.WVS
and scan.manual_intervention
and scan.manual_intervention
and self.manual_browsing_storage
):
return
ui_session_id = self.manual_browsing_storage.get_ui_session_id(scan.scan_id)
if not ui_session_id:
logging.warning(
"manual intervention scan_session w/o ui session %s", scan_session_id
)
event_query = create_event(
"scan_mi_required_no_ui_session",
owner_id=scan.owner_id,
user_id=scan.creator_id,
resource_type=EventResources.scan,
resource_id=scan.scan_id,
)
db.execute(event_query)
return
if scan.schedule_start_date is not None or scan.recurrence is not None:
logging.warning(
"manual intervention flag for scheduled scan_session %s",
scan_session_id,
)
event_query = create_event(
"scan_mi_required_scheduled_scan",
owner_id=scan.owner_id,
user_id=scan.creator_id,
resource_type=EventResources.scan,
resource_id=scan.scan_id,
)
db.execute(event_query)
return
manual_browsing_storage_items.append(
dict(
target_id=scan.target_id,
scan_id=scan.scan_id,
scan_session_id=scan_session_id,
user_id=scan.creator_id,
scanning_app=scanning_app,
ui_session_id=ui_session_id,
address=scan.address,
description=scan.description,
)
)
def start_queued_scan_session_jobs(self):
"""
start any jobs with status=queued or status=resuming
:return:
"""
def update_scanning_job(job_update_data):
with Connection(self.shard_name) as (dbs):
dbs.execute(
(ScanSessionJobsTable.update(values=job_update_data))
.where(
ScanSessionJobRow.scan_session_id
== scan_session_job["scan_session_id"]
)
.where(
ScanSessionJobRow.scanning_app
== scan_session_job["scanning_app"]
)
.where(ScanSessionJobRow.result_processed.isnot(True))
.where(
ScanSessionJobRow.status.in_(
(ScanStatusTypes.QUEUED, ScanStatusTypes.RESUMING)
)
)
)
select_from = ScanSessionJobsTable.join(
ScanSessionsTable,
ScanSessionJobRow.scan_session_id == ScanSessionRow.scan_session_id,
)
select_from = select_from.join(
ScansTable, ScanRow.scan_id == ScanSessionRow.scan_id
)
select_from = select_from.join(
TargetsTable, TargetRow.target_id == ScanSessionRow.target_id
)
select_from = select_from.join(
ProfilesTable, ProfileRow.profile_id == ScanSessionRow.used_profile_id
)
sql_fetch_starting_jobs = (
sql_select(
(
ScanSessionJobRow,
ScanSessionRow.scan_id,
ScanSessionRow.deleted_at,
ScanSessionRow.allowed_targets,
ScanSessionRow.scan_key,
ProfileRow.jobs,
TargetRow.address,
TargetRow.target_id,
TargetRow.type.label("target_type"),
TargetRow.verification,
ScanRow.owner_id,
ScanRow.creator_id,
ScanRow.continuous,
ScanRow.schedule_start_date.label("scan_schedule_start_date"),
ScanRow.recurrence.label("scan_recurrence"),
)
)
.select_from(select_from)
.where(
ScanSessionJobRow.status.in_(
(ScanStatusTypes.QUEUED, ScanStatusTypes.RESUMING)
)
)
.where(ScanSessionJobRow.result_processed.isnot(True))
.order_by(ScanSessionJobRow.start_deadline)
)
if lock_to_owner_id:
sql_fetch_starting_jobs = sql_fetch_starting_jobs.where(
ScanRow.owner_id == lock_to_owner_id
)
sql_abort_jobs = (
(
ScanSessionJobsTable.update(
values=dict(
status=ScanStatusTypes.ABORTED,
start_date=sql_func.current_timestamp(),
end_date=sql_func.current_timestamp(),
status_data=AbortReason.abort_requested,
)
)
)
.where(
ScanSessionJobRow.status.in_(
(ScanStatusTypes.QUEUED, ScanStatusTypes.RESUMING)
)
)
.where(ScanSessionJobRow.abort_requested.is_(True))
)
jobs_to_process = []
with Connection(self.shard_name) as (db):
db.execute(sql_abort_jobs)
for scan_session_job in db.execute(sql_fetch_starting_jobs).fetchall():
jobs_to_process.append(dict(scan_session_job))
jobs_count = len(jobs_to_process)
if jobs_count:
logger.debug("%s jobs to start", jobs_count)
has_slots = dict()
for scan_session_job in jobs_to_process:
if should_suppress_scan_starting():
break
if scan_session_job["deleted_at"]:
update_scanning_job(
dict(
status=ScanStatusTypes.ABORTED,
start_date=sql_func.current_timestamp(),
end_date=sql_func.current_timestamp(),
status_data=AbortReason.scan_deleted,
)
)
continue
if is_setup_type_aws():
user = prepare_user_data_ext(
self.shard_name,
scan_session_job["owner_id"],
datetime.timedelta(hours=12),
)
if not user:
logger.error("user not found %s", scan_session_job["owner_id"])
update_scanning_job(
dict(
status=ScanStatusTypes.ABORTED,
start_date=sql_func.current_timestamp(),
end_date=sql_func.current_timestamp(),
status_data=AbortReason.user_deleted,
)
)
continue
if user["license_data"]["product_code"] == "OVSTRIAL":
if scan_session_job["scanning_app"] == ScanAppTypes.WVS:
scan_session_job["scanning_mode"] = "demo"
if scan_session_job["target_type"] == TargetTypes.NETWORK_ONLY:
if scan_session_job["scanning_app"] != ScanAppTypes.OVAS:
logger.warning(
"aborted web scan:%s on network only target for disabled user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.network_scan_only
)
continue
if not user["enabled"]:
logger.warning(
"aborted scan:%s for disabled user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.user_disabled
)
continue
if user["license_data"]["product_code"] == "OVS_FREE_MODE":
pass
if scan_session_job["scanning_app"] == ScanAppTypes.WVS:
logger.warning(
"aborted wvs scan:%s for OVS_FREE_MODE user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.license_error
)
continue
if scan_session_job["target_type"] == TargetTypes.DEMO:
logger.warning(
"aborted demo network scan:%s for OVS_FREE_MODE user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.license_error
)
continue
else:
if user["license_data"]["product_code"] == "OVS_EXPIRED":
logger.warning(
"aborted scan:%s for OVS_EXPIRED user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.license_error
)
continue
else:
if user["license_data"]["product_code"] == "OVSTRIAL":
pass
if scan_session_job["scanning_app"] == ScanAppTypes.OVAS:
if scan_session_job["target_type"] == TargetTypes.DEMO:
pass
else:
if user["confirmation"] not in (
UserConfirmation.AUTO_VALIDATED,
UserConfirmation.ADMIN_VALIDATED,
):
logger.warning(
"aborted scan:%s for OVSTRIAL not confirmed user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.user_not_validated
)
continue
else:
if scan_session_job["scanning_app"] == ScanAppTypes.WVS:
wvs_scan_count = 0
with Connection(self.shard_name) as (db):
sf = ScanSessionsTable.join(
ProfilesTable,
ProfileRow.profile_id
== ScanSessionRow.used_profile_id,
)
q = (
sql_select((ProfileRow.jobs,))
.select_from(sf)
.where(
ScanSessionRow.scan_id.in_(
sql_select((ScanRow.scan_id,)).where(
ScanRow.owner_id == get_owner_id(user)
)
)
)
)
for r in db.execute(q).fetchall():
if ScanAppTypes.WVS in r.jobs:
wvs_scan_count += 1
if wvs_scan_count > 5:
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.too_many_web_scans
)
continue
else:
if user["license_type"] == "customer":
user_license = BaseLicense.get_license_from_license_data(
user["license_data"]
)
if not user_license.is_active_license():
logger.warning(
"aborted scan:%s for expired license user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job, AbortReason.license_error
)
continue
hs = has_slots.get(scan_session_job["scanning_app"])
if hs is None:
hs = self.worker_manager.has_slots(
scan_session_job["scanning_app"]
)
has_slots[scan_session_job["scanning_app"]] = hs
if not hs:
logging.warning(
"no more %s workers", scan_session_job["scanning_app"]
)
continue
if has_feature(SetupFeatures.TARGET_CHECKING_MANDATORY):
if scan_session_job["target_type"] == TargetTypes.DEFAULT:
if scan_session_job["scanning_app"] == ScanAppTypes.WVS:
if (
scan_session_job["verification"]
!= TargetVerificationTypes.ADMIN
):
r = do_check_target(
self.shard_name,
scan_session_job["target_id"],
generate_notification=True,
)
if r["status"] != "success":
logger.warning(
"aborted scan:%s for target verification failed user:%s",
scan_session_job["scan_id"],
scan_session_job["owner_id"],
)
self._Broker__mark_job_aborted(
scan_session_job,
AbortReason.target_verification_failed,
)
continue
try:
worker_id = self.worker_manager.add_job(**scan_session_job)
except NoMoreWorkerSlots:
logger.debug(
"NoMoreWorkerSlots, aborting attempt_to_start_queued_jobs process"
)
has_slots[scan_session_job["scanning_app"]] = False
continue
except NoMoreWorkers:
logger.debug(
"NoMoreWorkers, aborting attempt_to_start_queued_jobs process"
)
has_slots[scan_session_job["scanning_app"]] = False
continue
except StartFailed as e:
self._Broker__mark_job_failed(
scan_session_job,
dict(reason=JobConstants.start_failed, extra=e.str_message),
start_date=sql_func.current_timestamp(),
)
continue
end_deadline = None
if self.max_scan_time:
t = "continuous" if scan_session_job["continuous"] else "generic"
d = self.max_scan_time.get(
scan_session_job["scanning_app"], {}
).get(t)
if d:
end_deadline = (datetime.datetime.now(tz=tzlocal())) + d
update_scanning_job(
dict(
status=ScanStatusTypes.STARTING,
start_date=sql_func.current_timestamp(),
last_update=sql_func.current_timestamp(),
end_deadline=end_deadline,
worker_id=worker_id,
)
)
self._Broker__create_scan_job_event(
scan_session_job,
"scan_job_starting",
"starting",
worker_id=worker_id,
)
self._Broker__schedule_job_recheck(
scan_session_job, job_helpers.CheckDeltaT.starting
)
with Connection(self.shard_name) as (db):
q = (
(
ScanSessionsTable.update(
values=dict(start_date=sql_func.current_timestamp())
)
)
.where(
sql_and(
ScanSessionRow.scan_session_id
== scan_session_job["scan_session_id"],
ScanSessionRow.start_date.is_(None),
)
)
.returning(ScanSessionRow.scan_session_id)
)
sid = db.execute(q).scalar()
if sid:
if not scan_session_job["continuous"]:
event_query = create_event(
"scan_started",
owner_id=scan_session_job["owner_id"],
user_id=scan_session_job["creator_id"],
resource_type=EventResources.scan,
resource_id=scan_session_job["scan_id"],
data=dict(
scan_session_id=scan_session_job["scan_session_id"],
worker_id=worker_id,
),
shard=self.shard_name,
)
db.execute(event_query)
def process_finished_jobs(self):
"""
get all completed jobs and process them
:return:
"""
select_from = (
ScanSessionJobsTable.join(
ScanSessionsTable,
ScanSessionRow.scan_session_id == ScanSessionJobRow.scan_session_id,
)
.join(ScansTable, ScanSessionRow.scan_id == ScanRow.scan_id)
.join(
ProfilesTable, ProfileRow.profile_id == ScanSessionRow.used_profile_id
)
.join(TargetsTable, TargetRow.target_id == ScanSessionRow.target_id)
.outerjoin(
TargetConfigurationTable,
sql_and(
TargetConfigurationRow.name == "continuous_mode",
TargetConfigurationRow.target_id == ScanRow.target_id,
),
)
.outerjoin(
ScanSessionVulnsStatsTable,
ScanSessionVulnsStatsRow.scan_session_id
== ScanSessionJobRow.scan_session_id,
)
)
sql_fetch_finished_unprocessed_jobs = (
sql_select(
(
ScanSessionJobRow,
ScanSessionRow.used_profile_id,
ScanSessionRow.event_level,
ScanRow.scan_id,
ScanRow.creator_id,
ScanRow.owner_id,
ScanRow.schedule_disabled,
ScanRow.continuous,
ScanRow.report_template_id,
ScanRow.recurrence,
ScanRow.target_id,
ScanSessionVulnsStatsRow.vuln_stats,
ProfileRow.jobs,
ProfileRow.name.label("profile_name"),
TargetConfigurationRow.value.label("continuous_mode_enabled"),
TargetRow.address.label("target_address"),
TargetRow.description.label("target_description"),
)
)
.select_from(select_from)
.where(ScanSessionJobRow.status.in_(list(FinishedScanJobsStatusTypes)))
.where(
sql_not(
ScanSessionRow.status.in_(
(ScanStatusTypes.ABORTED, ScanStatusTypes.FAILED)
)
)
)
.where(ScanRow.deleted_at.is_(None))
.where(ScanSessionRow.deleted_at.is_(None))
.where(ScanSessionJobRow.result_processed.is_(False))
)
scan_sessions = defaultdict(dict)
required_jobs = dict()
with Connection(self.shard_name) as (db):
for job in db.execute(sql_fetch_finished_unprocessed_jobs).fetchall():
scan_sessions[job.scan_session_id][job.scanning_app] = dict(job)
if job.scan_session_id not in required_jobs:
required_jobs[job.scan_session_id] = set(job.jobs.keys())
for scan_session_id, jobs in scan_sessions.items():
logger.debug("check %s end status", scan_session_id)
logger.debug("jobs: ", jobs)
if set(jobs.keys()) != required_jobs[scan_session_id]:
logger.debug("%s is not finished yet", scan_session_id)
continue
scan_descriptor = jobs[list(jobs.keys())[0]]
now = datetime.datetime.now(tz=tzlocal())
start_date = min(
[
job["start_date"] if job["start_date"] else now
for job in jobs.values()
]
)
end_date = max(
[job["end_date"] if job["end_date"] else now for job in jobs.values()]
)
event_level = scan_descriptor["event_level"]
if event_level is None:
event_level = 0
target_id = scan_descriptor["target_id"]
recurrence = scan_descriptor["recurrence"]
target_in_continuous_mode = scan_descriptor["continuous_mode_enabled"]
this_is_continuous_mode_scan = scan_descriptor["continuous"]
scan_id = scan_descriptor["scan_id"]
paused = ScanStatusTypes.PAUSED in [job["status"] for job in jobs.values()]
aborted = (
ScanStatusTypes.ABORTED in [job["status"] for job in jobs.values()]
and not paused
)
logger.debug("Paused: %s, Aborted: %s", paused, aborted)
with Connection(self.shard_name) as (db):
user_data = prepare_user_data(db, user_id=scan_descriptor["owner_id"])
if user_data is None:
logger.warning(
"user %s not found or disabled", scan_descriptor["owner_id"]
)
q = (
ScanSessionJobsTable.update(values=dict(result_processed=True))
).where(ScanSessionJobRow.scan_session_id == scan_session_id)
db.execute(q)
if this_is_continuous_mode_scan:
q = (
(TargetConfigurationTable.update(values=dict(value=None)))
.where(TargetConfigurationRow.target_id == target_id)
.where(TargetConfigurationRow.name == "continuous_mode")
)
db.execute(q)
continue
if is_setup_type_aws():
user_license = BaseLicense.get_license_from_license_data(
user_data["license_data"]
)
else:
if is_setup_type_on_premise():
user_license = BaseLicense.get_system_license()
else:
logger.error("unsupported setup type")
continue
failed_count = 0
for job in jobs.values():
if job["status"] == ScanStatusTypes.FAILED:
failed_count += 1
scan_session_update = dict(start_date=start_date, end_date=end_date)
if aborted:
scan_session_update["status"] = ScanStatusTypes.ABORTED
else:
if paused:
scan_session_update["status"] = ScanStatusTypes.PAUSED
else:
scan_session_update["status"] = ScanStatusTypes.COMPLETED
if failed_count == len(jobs):
scan_session_update["status"] = ScanStatusTypes.FAILED
if ScanJobEventLevel.CRITICAL in [
job["event_level"] for job in jobs.values()
]:
with Connection(self.shard_name) as (db):
q = (
sql_select((sql_func.count(EventRow.event_id),))
.where(EventRow.resource_id == scan_session_id)
.where(EventRow.type_id.in_((430, 431)))
)
if db.execute(q).scalar():
scan_session_update["status"] = ScanStatusTypes.FAILED
if paused:
if scan_session_update["status"] != ScanStatusTypes.FAILED:
with Connection(self.shard_name) as (db):
q = (
ScanSessionsTable.update(values=scan_session_update)
).where(
ScanSessionRow.scan_session_id == scan_session_id
)
db.execute(q)
q = (
ScanSessionJobsTable.update(
values=dict(result_processed=True)
)
).where(
ScanSessionJobRow.scan_session_id == scan_session_id
)
db.execute(q)
continue
if not aborted:
if has_feature(SetupFeatures.SALES_UPDATES):
if user_data["license_type"] == "trial":
if failed_count < len(jobs):
from helpers.sales.events import trial_scan_done
trial_scan_done(user_data)
if this_is_continuous_mode_scan:
if not target_in_continuous_mode:
with Connection(self.shard_name, create_transactions=True) as (
db
):
q = (
ScanSessionsTable.update(values=scan_session_update)
).where(ScanSessionRow.scan_session_id == scan_session_id)
db.execute(q)
q = (
ScanSessionJobsTable.update(
values=dict(result_processed=True)
)
).where(
ScanSessionJobRow.scan_session_id == scan_session_id
)
db.execute(q)
continue
disable_recurrence = False
if not user_license.has_feature(Features.CONTINUOUS_SCANS):
logger.debug(
"disable_recurrence %s continuous scan not allowed",
scan_session_id,
)
disable_recurrence = AbortReason.license_error
else:
if aborted:
status_data_list = [
job["status_data"] for job in jobs.values()
]
if AbortReason.user_deleted in status_data_list:
logger.debug(
"disable_recurrence %s user_deleted",
scan_session_id,
)
disable_recurrence = 1
else:
if AbortReason.user_disabled in status_data_list:
logger.debug(
"disable_recurrence %s user disabled",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.network_scan_only
in status_data_list
):
logger.debug(
"disable_recurrence %s network scan only",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.too_many_web_scans
in status_data_list
):
logger.debug(
"disable_recurrence %s too many web scans",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.license_error
in status_data_list
):
logger.debug(
"disable_recurrence %s license error",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.user_not_validated
in status_data_list
):
scan_session_update[
"status"
] = ScanStatusTypes.COMPLETED
else:
if (
AbortReason.start_timeout
in status_data_list
):
logger.debug(
"disable_recurrence %s start timeout",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.scan_deleted
in status_data_list
):
logger.debug(
"disable_recurrence %s scan deleted",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.abort_requested
in status_data_list
):
pass
else:
if (
AbortReason.excluded_hours
in status_data_list
):
pass
else:
if (
AbortReason.target_verification_failed
in status_data_list
):
pass
else:
if (
AbortReason.scan_timeout
in status_data_list
):
scan_session_update[
"status"
] = (
ScanStatusTypes.COMPLETED
)
scan_session_update[
"progress"
] = 100
else:
disable_recurrence = (
1
)
logger.warning(
"disabling recurrence for scan %s",
scan_session_id,
)
if disable_recurrence:
disable_cm_query = (
(
TargetConfigurationTable.update(
values=dict(
value=None
)
)
)
.where(
TargetConfigurationRow.target_id
== target_id
)
.where(
TargetConfigurationRow.name
== "continuous_mode"
)
)
event = None
if isinstance(
disable_recurrence,
str,
):
event = dict(
name="target_continuous_mode_disabled",
user_id=scan_descriptor[
"creator_id"
],
owner_id=scan_descriptor[
"owner_id"
],
resource_id=scan_descriptor[
"scan_id"
],
resource_type=EventResources.scan,
shard=self.shard_name,
data=dict(
reason=disable_recurrence,
target_id=target_id,
target_desc=(
scan_descriptor[
"target_address"
],
scan_descriptor[
"target_description"
],
),
),
)
with Connection(
self.shard_name,
create_transactions=True,
) as (
db
):
db.execute(
disable_cm_query
)
if event:
query = create_event(
**event
)
db.execute(
query
)
query = (
ScanSessionsTable.update(
values=scan_session_update
)
).where(
ScanSessionRow.scan_session_id
== scan_session_id
)
db.execute(
query
)
query = (
ScanSessionJobsTable.update(
values=dict(
result_processed=True
)
)
).where(
ScanSessionJobRow.scan_session_id
== scan_session_id
)
db.execute(
query
)
else:
rs = (
dateutil.rrule.rruleset()
)
rs.rrule(
dateutil.rrule.rrulestr(
recurrence
)
)
next_run = rs.after(
datetime.datetime.now(
tz=tzlocal()
),
True,
)
with Connection(
self.shard_name,
create_transactions=True,
) as (
db
):
query = (
ScanSessionsTable.update(
values=scan_session_update
)
).where(
ScanSessionRow.scan_session_id
== scan_session_id
)
db.execute(
query
)
if next_run:
query = (
ScansTable.update(
values=dict(
next_run=next_run
)
)
).where(
ScanRow.scan_id
== scan_id
)
db.execute(
query
)
query = (
ScanSessionJobsTable.update(
values=dict(
result_processed=True
)
)
).where(
ScanSessionJobRow.scan_session_id
== scan_session_id
)
db.execute(
query
)
disable_recurrence = False
generate_event = True
if aborted:
generate_event = False
status_data_list = [job["status_data"] for job in jobs.values()]
if AbortReason.user_deleted in status_data_list:
logger.debug("disable_recurrence %s user deleted", scan_session_id)
disable_recurrence = 1
else:
if AbortReason.user_disabled in status_data_list:
logger.debug(
"disable_recurrence %s user disabled", scan_session_id
)
disable_recurrence = 1
else:
if AbortReason.network_scan_only in status_data_list:
logger.debug(
"disable_recurrence %s network scans only",
scan_session_id,
)
disable_recurrence = 1
else:
if AbortReason.too_many_web_scans in status_data_list:
logger.debug(
"disable_recurrence %s too many web scans",
scan_session_id,
)
disable_recurrence = 1
else:
if AbortReason.license_error in status_data_list:
logger.debug(
"disable_recurrence %s license error",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.user_not_validated
in status_data_list
):
logger.debug(
"disable_recurrence %s user not validated",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.start_timeout
in status_data_list
):
logger.debug(
"disable_recurrence %s start timeout",
scan_session_id,
)
generate_event = True
else:
if (
AbortReason.scan_deleted
in status_data_list
):
logger.debug(
"disable_recurrence %s scan deleted",
scan_session_id,
)
disable_recurrence = 1
else:
if (
AbortReason.abort_requested
in status_data_list
):
pass
else:
if (
AbortReason.excluded_hours
in status_data_list
):
pass
else:
if (
AbortReason.target_verification_failed
in status_data_list
):
generate_event = True
else:
if (
AbortReason.scan_timeout
in status_data_list
):
scan_session_update[
"status"
] = (
ScanStatusTypes.COMPLETED
)
scan_session_update[
"progress"
] = 100
generate_event = True
else:
disable_recurrence = 1
logger.warning(
"disabling recurrence for scan %s",
scan_session_id,
)
event = dict(
name="scan_done"
if scan_session_update[
"status"
]
== "completed"
else "scan_failed",
user_id=scan_descriptor[
"creator_id"
],
owner_id=scan_descriptor[
"owner_id"
],
resource_id=scan_descriptor[
"scan_id"
],
resource_type=EventResources.scan,
shard=self.shard_name,
data=dict(
scan_session_id=scan_session_id,
target_id=scan_descriptor[
"target_id"
],
target_desc=(
scan_descriptor[
"target_address"
],
scan_descriptor[
"target_description"
],
),
failed_job_count=failed_count,
event_level=event_level,
started=start_date,
ended=end_date,
vuln_stats=scan_descriptor[
"vuln_stats"
],
profile_name=scan_descriptor[
"profile_name"
],
),
)
next_run = None
if (
scan_session_update[
"status"
]
== ScanStatusTypes.FAILED
):
if (
is_setup_type_on_premise()
):
pass
else:
logger.debug(
"disable_recurrence %s scan failed",
scan_session_id,
)
disable_recurrence = (
True
)
if disable_recurrence:
event["data"][
"recurrence_disabled"
] = True
else:
if scan_descriptor[
"recurrence"
]:
if (
scan_descriptor[
"schedule_disabled"
]
is not True
):
rs = (
dateutil.rrule.rruleset()
)
rs.rrule(
dateutil.rrule.rrulestr(
scan_descriptor[
"recurrence"
]
)
)
next_run = rs.after(
datetime.datetime.now(
tz=tzlocal()
),
True,
)
if next_run:
event[
"data"
][
"next_run"
] = next_run
with Connection(
self.shard_name,
create_transactions=True,
) as (db):
query = (
ScanSessionsTable.update(
values=scan_session_update
)
).where(
ScanSessionRow.scan_session_id
== scan_session_id
)
db.execute(query)
if next_run:
query = (
ScansTable.update(
values=dict(
next_run=next_run
)
)
).where(
ScanRow.scan_id
== scan_id
)
db.execute(
query
)
else:
if scan_descriptor[
"recurrence"
]:
query = (
ScansTable.update(
values=dict(
schedule_disabled=True
)
)
).where(
ScanRow.scan_id
== scan_id
)
db.execute(
query
)
if generate_event:
query = create_event(
**event
)
db.execute(
query
)
query = (
ScanSessionJobsTable.update(
values=dict(
result_processed=True
)
)
).where(
ScanSessionJobRow.scan_session_id
== scan_session_id
)
db.execute(query)
if scan_descriptor["report_template_id"]:
if scan_session_update["status"] == ScanStatusTypes.COMPLETED:
if not (
not user_license.has_feature(Features.COMPLIANCE_REPORTS)
and is_report_id_compliance_id(
scan_descriptor["report_template_id"]
)
or not user_license.has_feature(Features.EXPORT_WAF)
and is_report_id_waf_export_id(
scan_descriptor["report_template_id"]
)
):
report_id = str(uuid.uuid4())
query = ReportsTable.insert().values(
report_id=report_id,
owner_id=scan_descriptor["owner_id"],
creator_id=scan_descriptor["creator_id"],
source=dict(
list_type="scan_result",
id_list=[scan_session_id],
description=(";").join(
(
scan_descriptor["target_address"],
scan_descriptor["target_description"],
)
),
),
report_template_id=scan_descriptor[
"report_template_id"
],
source_type="scan_result",
)
with Connection(self.shard_name) as (db):
db.execute(query)
此处可能存在不合适展示的内容，页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容，可点击提交进行申诉，我们将尽快为您处理。