Playbooks (solves #628)

.github/ISSUE_TEMPLATE/new_analyzer.md

@@ -14,7 +14,8 @@ assignees: ''
 ## Type of analyzer
 **this can be observable, file, and docker**
 
+
 ## Why should we use it
 
 
-## Possible implementation
+## Possible implementation

.github/release_template.md

@@ -16,3 +16,4 @@ WARNING: The release will be live within an hour!
 - [ ] Wait for [dockerHub](https://hub.docker.com/repository/docker/intelowlproject/intelowl) to finish the builds
 - [ ] Merge the PR to the `master` branch. **Note:** Only use "Merge and commit" as the merge strategy and not "Squash and merge". Using "Squash and merge" makes history between branches misaligned.
 - [ ] Remove the "wait" statement in the release description.
+- [ ] If the analyzer is free, Please add it in the `FREE_TO_USE_ANALYZERS` playbook in `playbook_config.json`

.github/workflows/pull_request_automation.yml

@@ -136,6 +136,10 @@ jobs:
         run: |
           docker/scripts/coverage_test.sh tests.analyzers_manager.test_file_scripts
 
+      - name: "Test: Playbooks Manager"
+        run: |
+          docker/scripts/coverage_test.sh tests.playbooks_manager.test_controller
+
       - name: "Coverage: generate xml and transfer from docker container to host"
         run: |
           docker exec intelowl_uwsgi coverage combine

api_app/analyzers_manager/classes.py

@@ -93,7 +93,9 @@ def _validate_result(self, result, level=0, max_recursion=190):
             result = 9223372036854775807
         return result
 
-    def before_run(self):
+    def before_run(self, *args, **kwargs):
+        parent_playbook = kwargs.get("parent_playbook", "")
+        self.add_parent_playbook(parent_playbook=parent_playbook)
         self.report.update_status(status=self.report.Status.RUNNING)
 
     def after_run(self):
@@ -136,8 +138,8 @@ def __post__init__(self):
             self.observable_classification = self._job.observable_classification
         return super(ObservableAnalyzer, self).__post__init__()
 
-    def before_run(self):
-        super().before_run()
+    def before_run(self, *args, **kwargs):
+        super().before_run(**kwargs)
         logger.info(
             f"STARTED analyzer: {self.__repr__()} -> "
             f"Observable: {self.observable_name}."
@@ -184,8 +186,8 @@ def __post__init__(self):
         self.file_mimetype = self._job.file_mimetype
         return super(FileAnalyzer, self).__post__init__()
 
-    def before_run(self):
-        super().before_run()
+    def before_run(self, *args, **kwargs):
+        super().before_run(**kwargs)
         logger.info(
             f"STARTED analyzer: {self.__repr__()} -> "
             f"File: ({self.filename}, md5: {self.md5})"

api_app/analyzers_manager/constants.py

@@ -20,3 +20,12 @@ class ObservableTypes(models.TextChoices):
     DOMAIN = "domain"
     HASH = "hash"
     GENERIC = "generic"
+
+
+class AllTypes(models.TextChoices):
+    IP = "ip"
+    URL = "url"
+    DOMAIN = "domain"
+    HASH = "hash"
+    GENERIC = "generic"
+    FILE = "file"

api_app/analyzers_manager/controller.py

@@ -4,16 +4,13 @@
 import logging
 from typing import Dict, List
 
-from celery import chord, uuid
-from django.conf import settings
+from celery import chord
 from django.utils.module_loading import import_string
 from rest_framework.exceptions import ValidationError
 
+from intel_owl import tasks
 from intel_owl.celery import app as celery_app
-from intel_owl.consts import DEFAULT_QUEUE
 
-from ..exceptions import AlreadyFailedJobException
-from ..helpers import get_now
 from ..models import Job
 from .classes import DockerBasedAnalyzer
 from .dataclasses import AnalyzerConfig
@@ -27,115 +24,32 @@ def start_analyzers(
     analyzers_to_execute: List[str],
     runtime_configuration: Dict[str, Dict] = None,
 ) -> None:
-    from intel_owl import tasks
 
     # we should not use mutable objects as default to avoid unexpected issues
     if runtime_configuration is None:
         runtime_configuration = {}
 
-    # to store the celery task signatures
-    task_signatures = []
+    cleaned_result = AnalyzerConfig.stack_analyzers(
+        job_id=job_id,
+        analyzers_to_execute=analyzers_to_execute,
+        runtime_configuration=runtime_configuration,
+    )
 
-    # get analyzer config
-    analyzer_dataclasses = AnalyzerConfig.all()
-
-    # get job
-    job = Job.objects.get(pk=job_id)
-    job.update_status(Job.Status.RUNNING)  # set job status to running
-
-    # loop over and create task signatures
-    for a_name in analyzers_to_execute:
-        # get corresponding dataclass
-        config = analyzer_dataclasses[a_name]
-
-        # if disabled or unconfigured (this check is bypassed in STAGE_CI)
-        if not config.is_ready_to_use and not settings.STAGE_CI:
-            logger.info(f"skipping execution of analyzer {a_name}, job_id {job_id}")
-            continue
-
-        # get runtime_configuration if any specified for this analyzer
-        runtime_params = runtime_configuration.get(a_name, {})
-        # gen new task_id
-        task_id = uuid()
-        # construct arguments
-        args = [
-            job_id,
-            config.asdict(),
-            {"runtime_configuration": runtime_params, "task_id": task_id},
-        ]
-        # get celery queue
-        queue = config.config.queue
-        if queue not in settings.CELERY_QUEUES:
-            logger.warning(
-                f"Analyzer {a_name} has a wrong queue: {queue}."
-                f" Setting to `{DEFAULT_QUEUE}`"
-            )
-            queue = DEFAULT_QUEUE
-        # get soft_time_limit
-        soft_time_limit = config.config.soft_time_limit
-        # create task signature and add to list
-        task_signatures.append(
-            tasks.run_analyzer.signature(
-                args,
-                {},
-                queue=queue,
-                soft_time_limit=soft_time_limit,
-                task_id=task_id,
-            )
-        )
+    task_signatures = cleaned_result[0]
 
     # fire the analyzers in a grouped celery task
     # also link the callback to be executed
     # canvas docs: https://docs.celeryproject.org/en/stable/userguide/canvas.html
     runner = chord(task_signatures)
     cb_signature = tasks.post_all_analyzers_finished.signature(
-        [job.pk, runtime_configuration], immutable=True
+        [job_id, runtime_configuration], immutable=True
     )
+
     runner(cb_signature)
 
     return None
 
 
-def job_cleanup(job: Job) -> None:
-    logger.info(f"[STARTING] job_cleanup for <-- {job.__repr__()}.")
-    status_to_set = job.Status.RUNNING
-
-    try:
-        if job.status == job.Status.FAILED:
-            raise AlreadyFailedJobException()
-
-        stats = job.get_analyzer_reports_stats()
-
-        logger.info(f"[REPORT] {job.__repr__()}, status:{job.status}, reports:{stats}")
-
-        if len(job.analyzers_to_execute) == stats["all"]:
-            if stats["running"] > 0 or stats["pending"] > 0:
-                status_to_set = job.Status.RUNNING
-            elif stats["success"] == stats["all"]:
-                status_to_set = job.Status.REPORTED_WITHOUT_FAILS
-            elif stats["failed"] == stats["all"]:
-                status_to_set = job.Status.FAILED
-            elif stats["failed"] >= 1 or stats["killed"] >= 1:
-                status_to_set = job.Status.REPORTED_WITH_FAILS
-            elif stats["killed"] == stats["all"]:
-                status_to_set = job.Status.KILLED
-
-    except AlreadyFailedJobException:
-        logger.error(
-            f"[REPORT] {job.__repr__()}, status: failed. Do not process the report"
-        )
-
-    except Exception as e:
-        logger.exception(f"job_id: {job.pk}, Error: {e}")
-        job.append_error(str(e), save=False)
-
-    finally:
-        if not (job.status == job.Status.FAILED and job.finished_analysis_time):
-            job.finished_analysis_time = get_now()
-        job.status = status_to_set
-        job.save(update_fields=["status", "errors", "finished_analysis_time"])
-
-
 def set_failed_analyzer(
     job_id: int, name: str, err_msg, **report_defaults
 ) -> AnalyzerReport:
@@ -153,7 +67,7 @@ def set_failed_analyzer(
 
 
 def run_analyzer(
-    job_id: int, config_dict: dict, report_defaults: dict
+    job_id: int, config_dict: dict, report_defaults: dict, parent_playbook
 ) -> AnalyzerReport:
     aconfig = AnalyzerConfig.from_dict(config_dict)
     try:
@@ -164,7 +78,7 @@ def run_analyzer(
             raise Exception(f"Class: {cls_path} couldn't be imported")
         # else
         instance = klass(config=aconfig, job_id=job_id, report_defaults=report_defaults)
-        report = instance.start()
+        report = instance.start(parent_playbook=parent_playbook)
     except Exception as e:
         report = set_failed_analyzer(job_id, aconfig.name, str(e), **report_defaults)
 
@@ -180,7 +94,7 @@ def post_all_analyzers_finished(job_id: int, runtime_configuration: dict) -> Non
     # get job instance
     job = Job.objects.get(pk=job_id)
     # execute some callbacks
-    job_cleanup(job)
+    job.job_cleanup()
     # fire connectors when job finishes with success
     # avoid re-triggering of connectors (case: recurring analyzer run)
     if job.status == Job.Status.REPORTED_WITHOUT_FAILS and (

api_app/analyzers_manager/dataclasses.py

@@ -1,9 +1,16 @@
 # This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
 # See the file 'LICENSE' for copying permission.
 import dataclasses
+import logging
 import typing
 
+from celery import uuid
+from celery.canvas import Signature
+from django.conf import settings
+
 from api_app.core.dataclasses import AbstractConfig
+from api_app.models import Job
+from intel_owl.consts import DEFAULT_QUEUE
 
 from .constants import HashChoices, TypeChoices
 from .serializers import AnalyzerConfigSerializer
@@ -12,6 +19,8 @@
     "AnalyzerConfig",
 ]
 
+logger = logging.getLogger(__name__)
+
 
 @dataclasses.dataclass
 class AnalyzerConfig(AbstractConfig):
@@ -31,7 +40,6 @@ class AnalyzerConfig(AbstractConfig):
     run_hash_type: typing.Literal["md5", "sha256"] = HashChoices.MD5
 
     # utils
-
     @property
     def is_type_observable(self) -> bool:
         return self.type == TypeChoices.OBSERVABLE
@@ -104,3 +112,81 @@ def all(cls) -> typing.Dict[str, "AnalyzerConfig"]:
     def filter(cls, names: typing.List[str]) -> typing.Dict[str, "AnalyzerConfig"]:
         all_analyzer_config = cls.all()
         return {name: ac for name, ac in all_analyzer_config.items() if name in names}
+
+    @staticmethod
+    def runnable_analyzers(analyzers_to_execute: typing.List[str]) -> typing.List[str]:
+        analyzer_dataclass = AnalyzerConfig.all()
+        return [
+            analyzer
+            for analyzer in analyzers_to_execute
+            if analyzer_dataclass.get(analyzer)
+        ]
+
+    @classmethod
+    def stack_analyzers(
+        cls,
+        job_id: int,
+        analyzers_to_execute: typing.List[str],
+        runtime_configuration: typing.Dict[str, typing.Dict] = None,
+        parent_playbook="",
+    ) -> typing.Tuple[typing.List[Signature], typing.List[str]]:
+        from intel_owl import tasks
+
+        # to store the celery task signatures
+        task_signatures = []
+        analyzers_used = []
+
+        analyzers_to_run = cls.runnable_analyzers(
+            analyzers_to_execute=analyzers_to_execute
+        )
+
+        analyzer_dataclasses = cls.all()
+
+        # get job
+        job = Job.objects.get(pk=job_id)
+        job.update_status(Job.Status.RUNNING)  # set job status to running
+
+        # loop over and create task signatures
+        for a_name in analyzers_to_run:
+            # get corresponding dataclass
+            config = analyzer_dataclasses.get(a_name, None)
+
+            # if disabled or unconfigured (this check is bypassed in STAGE_CI)
+            if not config.is_ready_to_use and not settings.STAGE_CI:
+                logger.info(f"skipping execution of analyzer {a_name}, job_id {job_id}")
+                continue
+
+            # get runtime_configuration if any specified for this analyzer
+            runtime_params = runtime_configuration.get(a_name, {})
+            # gen new task_id
+            task_id = uuid()
+            # construct arguments
+            args = [
+                job_id,
+                config.asdict(),
+                {"runtime_configuration": runtime_params, "task_id": task_id},
+                parent_playbook,
+            ]
+            # get celery queue
+            queue = config.config.queue
+            if queue not in settings.CELERY_QUEUES:
+                logger.warning(
+                    f"Analyzer {a_name} has a wrong queue."
+                    f" Setting to `{DEFAULT_QUEUE}`"
+                )
+                queue = DEFAULT_QUEUE
+            # get soft_time_limit
+            soft_time_limit = config.config.soft_time_limit
+            # create task signature and add to list
+            task_signatures.append(
+                tasks.run_analyzer.signature(
+                    args,
+                    {},
+                    queue=queue,
+                    soft_time_limit=soft_time_limit,
+                    task_id=task_id,
+                )
+            )
+            analyzers_used.append(a_name)
+
+        return task_signatures, analyzers_used

api_app/analyzers_manager/migrations/0002_analyzerreport_parent_playbook.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.14 on 2022-10-07 18:52
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("analyzers_manager", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="analyzerreport",
+            name="parent_playbook",
+            field=models.CharField(blank=True, default="", max_length=128),
+        ),
+    ]

api_app/analyzers_manager/serializers.py

@@ -29,6 +29,7 @@ class Meta:
             "end_time",
             "runtime_configuration",
             "type",
+            "parent_playbook",
         )