JoeSandbox New Solution

Solutions/JoeSandbox/Data Connectors/JoeSandbox/app.py

@@ -0,0 +1,83 @@
+"""
+Main Function
+"""
+
+# pylint: disable=logging-fstring-interpolation
+import logging
+from datetime import datetime, timedelta, timezone
+from json import loads
+from traceback import format_exc
+
+import azure.functions as func
+
+from .const import IOC_LIST, joe_config
+from .utils import (IOC_MAPPING_FUNCTION, joe_api, parse_analysis_data, state,
+                    submit_indicator, DATE_FORMAT)
+
+def main(mytimer: func.TimerRequest) -> None:
+    """
+    Timer-triggered Azure Function to interact with the JoeSandbox API.
+
+    This function fetches recent analyses from JoeSandbox, processes finished results,
+    extracts IOCs, and submits them to Microsoft Sentinel. It handles pagination, API
+    errors, and maintains state using checkpointing.
+
+    Parameters
+    ----------
+    mytimer : func.TimerRequest
+        Timer trigger object.
+    """
+    try:
+        if mytimer.past_due:
+            logging.info("The timer is past due!")
+            return
+
+        last_run_value = state.get()
+        if last_run_value is not None:
+            last_run = str(last_run_value)
+            logging.info(f"Last run timestamp: {last_run}")
+        else:
+            last_run = (
+                datetime.now(timezone.utc) - timedelta(days=int(joe_config.INITIAL_FETCH_DAYS))
+            ).strftime(DATE_FORMAT)
+            logging.info(f"No checkpoint found. Using initial fetch date: {last_run}")
+
+        current_time = datetime.now(timezone.utc).strftime(DATE_FORMAT)
+
+        verdicts = joe_config.JOE_ANALYSIS_VERDICTS.split(" & ")
+        logging.info(f"Configured analysis verdicts: {verdicts}")
+
+        analysis_list = joe_api.get_analysis_list(last_run, verdicts)
+        if not analysis_list:
+            logging.info("No analyses returned.")
+            return
+
+        logging.info(f"Number of analyses returned: {len(analysis_list)}")
+        all_indicators = []
+        for analysis in analysis_list:
+            webid = analysis.get("webid", "")
+
+            _, file_data = joe_api.download_analysis(webid, "irjsonfixed")
+            if not file_data:
+                logging.warning(f"No file data returned for analysis {webid}")
+                continue
+
+            json_file_data = loads(file_data)
+
+            analysis_info = joe_api.get_analysis_info(webid)
+            if not analysis_info or analysis_info.get("status") != "finished":
+                logging.info(f"Analysis {webid} is not marked as finished. Skipping.")
+                continue
+
+            iocs = parse_analysis_data(json_file_data) or {}
+            for key, values in iocs.items():
+                if key in IOC_LIST and key in IOC_MAPPING_FUNCTION:
+                    indicators = IOC_MAPPING_FUNCTION[key](values, analysis_info)
+                    all_indicators.extend(indicators)
+        submit_indicator(all_indicators)
+        state.post(current_time)
+
+    except Exception as ex:
+        logging.error("An unexpected error occurred.")
+        logging.error(f"Error: {ex}")
+        logging.error(f"Traceback: {format_exc()}")

Solutions/JoeSandbox/Data Connectors/JoeSandbox/const.py

@@ -0,0 +1,75 @@
+"""
+Constant File
+"""
+
+from dataclasses import dataclass
+from os import environ
+
+# pylint: disable=invalid-name
+
+
+@dataclass
+class JoeConfig:
+    """
+    JoeSandbox Configuration
+    """
+
+    API_KEY: str
+    BASE_URL: str
+    JOE_ANALYSIS_VERDICTS: str
+    INITIAL_FETCH_DAYS: str
+    VALID_UNTIL: str
+    API_URL: str = ""
+    CONNECTOR_NAME: str = "JoeSandboxThreatIntelligence:1.0.0"
+    RETRIES: int = 5
+    TIMEOUT = 300
+
+    def __post_init__(self):
+        self.BASE_URL = self.BASE_URL.rstrip("/")
+        self.API_URL = f"{self.BASE_URL}/api"
+
+
+joe_config = JoeConfig(
+    API_KEY=environ.get("JoeSandboxAPIKey", ""),
+    BASE_URL=environ.get("JoeSandboxBaseURL", "https://jbxcloud.joesecurity.org"),
+    JOE_ANALYSIS_VERDICTS=environ.get("JoeAnalysisVerdict", "Malicious & Suspicious"),
+    INITIAL_FETCH_DAYS=environ.get("JoeSandboxInitialFetchDate", ""),
+    VALID_UNTIL=environ.get("IndicatorExpirationInDays", "30"),
+)
+
+
+@dataclass
+class APIConfig:
+    """
+    Microsoft API Configurations
+    """
+
+    APPLICATION_ID: str
+    APPLICATION_SECRET: str
+    AUTH_URL: str
+    URL: str
+    RESOURCE_APPLICATION_ID_URI: str = "https://management.azure.com"
+    USER_AGENT: str = "MSSentinelJoeSandboxIntelligenceSentinel:1.0.0"
+    SLEEP: int = 60
+    TIMEOUT: int = 300
+    MAX_TI_INDICATORS_PER_REQUEST: int = 100
+
+
+SENTINEL_API = APIConfig(
+    APPLICATION_ID=environ.get("AzureClientID", ""),
+    APPLICATION_SECRET=environ.get("AzureClientSecret", ""),
+    AUTH_URL=f"https://login.microsoftonline.com/{environ.get('AzureTenantID', '')}/oauth2/token",
+    URL=f"https://api.ti.sentinel.azure.com/workspaces/{environ.get('AzureWorkspaceID', '')}/"
+    f"threat-intelligence-stix-objects:upload?api-version=2024-02-01-preview",
+)
+
+RETRY_STATUS_CODE = [500, 501, 502, 503, 504, 429]
+CONFIDENCE = {"malicious": "100", "suspicious": "75"}
+HASH_TYPE_LIST = [
+    ("MD5", "md5_hash"),
+    ("SHA-1", "sha1_hash"),
+    ("SHA-256", "sha256_hash"),
+]
+IOC_LIST = ["domains", "ips", "urls", "files"]
+DATE_FORMAT = "%Y-%m-%d"
+UTC_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"

Solutions/JoeSandbox/Data Connectors/JoeSandbox/function.json

@@ -0,0 +1,11 @@
+{
+    "scriptFile": "app.py",
+    "bindings": [
+        {
+            "name": "mytimer",
+            "type": "timerTrigger",
+            "direction": "in",
+            "schedule": "%Polling%"
+        }
+    ]
+}