Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
evidence_output = [ReportText]
NAME = 'HTTPAccessLogAnalysisJob'
def create_tasks(self, evidence):
"""Create task.
Args:
evidence: List of evidence objects to process
Returns:
A list of tasks to schedule.
"""
evidence = [e for e in evidence if e.artifact_name in ACCESS_LOG_ARTIFACTS]
return [wordpress.WordpressAccessLogAnalysisTask() for _ in evidence]
manager.JobsManager.RegisterJobs(
[HTTPAccessLogExtractionJob, HTTPAccessLogAnalysisJob])
def setup(self, jobs_blacklist=None, jobs_whitelist=None, *args, **kwargs):
"""Does setup of Task manager and its dependencies.
Args:
jobs_blacklist (list): Jobs that will be excluded from running
jobs_whitelist (list): The only Jobs will be included to run
"""
self._backend_setup(*args, **kwargs)
job_names = jobs_manager.JobsManager.GetJobNames()
if jobs_blacklist or jobs_whitelist:
log.info(
'Filtering Jobs with whitelist {0!s} and blacklist {1!s}'.format(
jobs_whitelist, jobs_blacklist))
job_names = jobs_manager.JobsManager.FilterJobNames(
job_names, jobs_blacklist, jobs_whitelist)
self.jobs = [job for _, job in jobs_manager.JobsManager.GetJobs(job_names)]
log.debug('Registered job list: {0:s}'.format(str(job_names)))