Merge pull request #747 from MISP/semayellow-feature/dev-01-anyrun-sandbox-integration

Semayellow feature/dev 01 anyrun sandbox integration

Author: adulau

misp_modules/lib/anyrun_sandbox/__init__.py

@@ -0,0 +1,2 @@
+class Config:
+    INTEGRATION: str = 'MISP:0.1'

misp_modules/lib/anyrun_sandbox/config.py

@@ -0,0 +1,147 @@
+import json
+from http import HTTPStatus
+from pymisp import MISPEvent, MISPObject
+
+from anyrun import RunTimeException
+from anyrun.connectors.sandbox.operation_systems import WindowsConnector, LinuxConnector, AndroidConnector
+
+
+class AnyRunParser:
+    """
+    Implements functionality for parsing ANY.RUN Sandbox reports into MISP objects and attributes
+    """
+    def __init__(
+        self,
+        config: dict[str, str],
+        analysis_uuid: str,
+        connector: WindowsConnector | LinuxConnector | AndroidConnector
+    ) -> None:
+        self._event = MISPEvent()
+        self._connector = connector
+        self._analysis_uuid = analysis_uuid
+        self._config = config
+
+    def generate_results(self) -> dict[str, dict]:
+        """
+        Enriches the results dictionary with ANY.RUN report objects
+
+        :return: MISP results dictionary
+        """
+        self._check_analysis_completion()
+
+        summary = self._connector.get_analysis_report(self._analysis_uuid)
+
+        self._add_report()
+
+        if self._check_option("IOCs"):
+            self._add_indicators()
+
+        if self._check_option("Tags"):
+            self._add_tags(summary)
+
+        if self._check_option("MITRE"):
+            self._add_mitre_galaxies(summary)
+
+        results = json.loads(self._event.to_json())
+
+        return {
+            "results": {
+                key: results[key]
+                for key in ("Attribute", "Object", "Tag")
+                if (key in results and results[key])
+            }
+        }
+
+    def _check_analysis_completion(self) -> None:
+        """
+        Checks if ANY.RUN Sandbox analysis is completed
+
+        :raises RunTimeException: If analysis is not completed
+        """
+        status = self._connector.get_analysis_report(self._analysis_uuid).get("data").get("status")
+
+        if status != "done":
+            raise RunTimeException(
+                f"Analysis is running. Please, wait a few minutes to request a report",
+                HTTPStatus.BAD_REQUEST
+            )
+
+    def _add_indicators(self) -> None:
+        """
+        Converts ANY.RUN indicators to the MISP attributes
+        """
+        if iocs := self._connector.get_analysis_report(self._analysis_uuid, report_format="ioc"):
+            for ioc in iocs:
+                if ioc.get("type") in ("domain", "ip", "sha256") and ioc.get("reputation") in (1, 2):
+                    ioc_type = ioc.get("type")
+                    ioc_reputation = {1: "Suspicious", 2: "Malicious"}.get(ioc.get("reputation"))
+                    attribute = self._event.add_attribute(
+                        type=ioc_type if ioc_type in ("domain", "sha256") else "ip-dst",
+                        value=ioc.get("ioc"),
+                        categories="Network activity" if ioc_type in ("ip", "domain") else "Payload delivery",
+                        comment=f"{ioc_reputation} IoC from https://app.any.run/tasks/{self._analysis_uuid}."
+                    )
+
+                    attribute.add_tag("ANY.RUN Sandbox")
+
+    def _add_tags(self, summary: dict) -> None:
+        """
+        Converts ANY.RUN analysis tags to the MISP tags
+
+        :param summary: ANY.RUN Sandbox report
+        """
+        self._event.add_tag("ANY.RUN Sandbox")
+        if tags := summary.get("data").get("analysis").get("tags"):
+            for tag in tags:
+                self._event.add_tag(tag.get("tag"))
+
+    def _add_mitre_galaxies(self, summary: dict) -> None:
+        """
+        Converts ANY.RUN analysis mitre techniques to the MISP Galaxies
+
+        :param summary: ANY.RUN Sandbox report
+        """
+        if mitre_techniques := summary.get("data").get("mitre"):
+            for entry in mitre_techniques:
+                if entry:
+                    mitre_name = entry.get("name")
+                    mitre_id = entry.get("id")
+                    self._event.add_tag(f'misp-galaxy:mitre-attack-pattern="{mitre_name} - {mitre_id}"')
+
+    def _add_report(self) -> None:
+        """
+        Converts ANY.RUN analysis HTML report and external references to the MISP attributes
+
+        :return:
+        """
+        report = self._connector.get_analysis_report(self._analysis_uuid, report_format="html")
+
+        self._event.add_attribute(
+            type="text",
+            value=f"ANY.RUN Sandbox verdict: {self._connector.get_analysis_verdict(self._analysis_uuid)}",
+            categories="Other",
+            comment="ANYRUN Sandbox Analysis verdict."
+        )
+
+        self._event.add_attribute(
+            type="link",
+            value=f"https://app.any.run/tasks/{self._analysis_uuid}",
+            categories="External analysis",
+            comment="ANYRUN Sandbox Analysis report."
+        )
+
+        self._event.add_attribute(
+            type="attachment",
+            value="report.html",
+            data=report.encode(),
+            comment="ANYRUN Sandbox Analysis report."
+        )
+
+    def _check_option(self, option_name: str) -> bool:
+        """
+        Checks if option is active
+
+        :param option_name: Option name to check
+        :return: True if option is enabled else False
+        """
+        return bool(int(self._config.get(option_name)))

misp_modules/lib/anyrun_sandbox/parser.py

@@ -0,0 +1,164 @@
+import io
+import base64
+import zipfile
+
+from anyrun import RunTimeException
+from anyrun.connectors import SandboxConnector
+from anyrun.connectors.sandbox.operation_systems import WindowsConnector, LinuxConnector, AndroidConnector
+
+from anyrun_sandbox.config import Config
+
+
+class AnyRunSubmitter:
+    """
+    Implements functionality for sending MISP entities to the ANY.RUN sandbox
+    """
+    def __init__(self, request: dict) -> None:
+        self._request = request
+        self._config: dict = request.get("config")
+
+    def submit(self) -> str:
+        """
+        Configures ANY.RUN Sandbox environment and executes the analysis
+
+        :return: ANY.RUN Sandbox analysis uuid
+        """
+        self._parse_config()
+        self._sanitize_config()
+
+        if self._os_type == "windows":
+            with SandboxConnector.windows(self._token, Config.INTEGRATION) as connector:
+                analysis_uuid = self._process_analysis(connector)
+        elif self._os_type in ("ubuntu", "debian"):
+            with SandboxConnector.linux(self._token, Config.INTEGRATION) as connector:
+                analysis_uuid = self._process_analysis(connector)
+        elif self._os_type == "android":
+            with SandboxConnector.android(self._token, Config.INTEGRATION) as connector:
+                analysis_uuid = self._process_analysis(connector)
+        else:
+            raise RunTimeException(
+                f"Received invalid OS type: {self._os_type}. Supports: windows, ubuntu, debian, android"
+            )
+
+        return analysis_uuid
+
+    def _parse_config(self) -> None:
+        """
+        Configures analysis options according to the chosen environment
+
+        """
+        self._token = self._config.pop("api_key", "")
+        self._os_type = self._config.pop("os_type", "")
+
+        if not any((self._token, self._os_type)):
+            raise RunTimeException(f"ANYRUN Sandbox API-KEY and OS type must be specified.")
+
+        if "url" in self._request:
+            self._prepare_url_params()
+            self._config["obj_url"] = self._request.get("url")
+        elif "malware-sample" in self._request:
+            self._prepare_file_params()
+            self._prepare_file_content("malware-sample")
+        elif "attachment" in self._request:
+            self._prepare_file_params()
+            self._prepare_file_content("attachment")
+        else:
+            raise RunTimeException("Received invalid Object. Supports: url, attachment, malware-sample.")
+
+    def _process_analysis(self, connector: WindowsConnector | LinuxConnector | AndroidConnector) -> str:
+        """
+        Executes analysis
+
+        :param connector: Instance of the ANY.RUN connector
+        :return: ANY.RUN Sandbox analysis uuid
+        """
+        connector.check_authorization()
+
+        if "obj_url" in self._config:
+            analysis_uuid = connector.run_url_analysis(**self._config)
+        else:
+            analysis_uuid = connector.run_file_analysis(**self._config)
+        return analysis_uuid
+
+    def _prepare_url_params(self) -> None:
+        """
+        Prepares analysis configuration for the url submission
+        """
+        if self._os_type != "windows":
+            self._config.pop("env_version", "")
+            self._config.pop("env_bitness", "")
+            self._config.pop("env_type", "")
+
+        self._config.pop("obj_ext_extension", "")
+        self._config.pop("obj_ext_startfolder", "")
+        self._config.pop("obj_ext_cmd", "")
+        self._config.pop("obj_force_elevation", "")
+        self._config.pop("run_as_root", "")
+
+
+    def _prepare_file_params(self) -> None:
+        """
+        Prepares analysis configuration for the file submission
+        """
+        self._config.pop("obj_ext_browser", "")
+
+        if self._os_type == "windows":
+            self._config.pop("run_as_root", "")
+        elif self._os_type in ("ubuntu", "debian"):
+            self._config.pop("env_version", "")
+            self._config.pop("env_bitness", "")
+            self._config.pop("env_type", "")
+            self._config.pop("obj_force_elevation", "")
+            self._config["env_os"] = self._os_type
+        elif self._os_type == "android":
+            self._config.pop("env_version", "")
+            self._config.pop("env_bitness", "")
+            self._config.pop("env_type", "")
+            self._config.pop("obj_force_elevation", "")
+            self._config.pop("obj_ext_startfolder", "")
+            self._config.pop("run_as_root", "")
+
+    def _prepare_file_content(self, sample_type: str) -> None:
+        """
+        Prepares file content to the analysis
+
+        :param sample_type: Attachment or malware-sample MISP entity
+        """
+        if sample_type == "attachment":
+            filename = self._request.get("attachment")
+            file_content = self._extract_file_content(self._request.get("data"))
+        else:
+            filename = self._request.get("malware-sample").split("|", 1)[0]
+            file_content = self._extract_file_content(self._request.get("data"), is_encoded=True)
+
+        self._config["filename"] = filename
+        self._config["file_content"] = file_content
+
+    def _sanitize_config(self) -> None:
+        """
+        Removes empty parameters from the analysis configuration
+        """
+        temp_config = dict()
+
+        for key, value in self._config.items():
+            if value is not None:
+                temp_config[key] = value
+
+        self._config = temp_config
+
+    @staticmethod
+    def _extract_file_content(file_content: str, is_encoded: bool = False) -> bytes:
+        """
+        Extracts file content from the **malware-sample** MISP entity
+
+        :param file_content: Base64 file content
+        :param is_encoded: Marks if **malware-sample** or **attachment** entity received
+        :return: File bytes payload
+        """
+        data = base64.b64decode(file_content)
+
+        if is_encoded:
+            with zipfile.ZipFile(io.BytesIO(data)) as file:
+                data = file.read(file.namelist()[0], pwd=b"infected")
+
+        return data