tmp

PLAYBOOKS/TI_Enrichment_By_AlienVaultOTX.py

@@ -3,11 +3,37 @@ import json
 from Lib.api import is_ipaddress
 from Lib.baseplaybook import BasePlaybook
 from PLUGINS.AlienVaultOTX.alienvaultotx import AlienVaultOTX
-from PLUGINS.SIRP.sirpapi import Artifact
-from PLUGINS.SIRP.sirpcoremodel import EnrichmentModel, ArtifactModel
+from PLUGINS.SIRP.sirpapi import Artifact, Case
+from PLUGINS.SIRP.sirpcoremodel import EnrichmentModel, ArtifactModel, ArtifactType
 from PLUGINS.SIRP.sirpextramodel import PlaybookJobStatus, PlaybookModel
 
 
+TI_ENRICHMENT_TYPE = "Threat Intelligence"
+TI_PROVIDER = "OTX"
+
+
+def _query_ip(value: str) -> dict:
+    if not is_ipaddress(value):
+        return {"error": "Invalid IP address format."}
+    return AlienVaultOTX.query_ip(value)
+
+
+def _query_hash(value: str) -> dict:
+    return AlienVaultOTX.query_file(value)
+
+
+def _query_url(value: str) -> dict:
+    return AlienVaultOTX.query_url(value)
+
+
+OTX_QUERY_HANDLERS = {
+    ArtifactType.IP_ADDRESS: _query_ip,
+    ArtifactType.HASH: _query_hash,
+    ArtifactType.URL_STRING: _query_url,
+    ArtifactType.UNIFORM_RESOURCE_LOCATOR: _query_url,
+}
+
+
 class Playbook(BasePlaybook):
     NAME = "TI Enrichment By AlienVaultOTX"
     DESC = "TI Enrichment By AlienVaultOTX"
@@ -15,34 +41,101 @@ class Playbook(BasePlaybook):
     def __init__(self):
         super().__init__()  # do not delete this code
 
+    @staticmethod
+    def _normalize_artifact_type(artifact_type):
+        if isinstance(artifact_type, ArtifactType):
+            return artifact_type
+        try:
+            return ArtifactType(artifact_type)
+        except ValueError:
+            return artifact_type
+
+    def _query_ti(self, artifact) -> dict:
+        artifact_type = self._normalize_artifact_type(artifact.type)
+        handler = OTX_QUERY_HANDLERS.get(artifact_type)
+        if not handler:
+            return {
+                "error": "Unsupported type.",
+                "artifact_type": str(artifact.type),
+                "supported_types": [artifact_type.value for artifact_type in OTX_QUERY_HANDLERS],
+            }
+        return handler(artifact.value or "")
+
+    @staticmethod
+    def _update_artifact_enrichment(artifact, ti_result: dict):
+        enrichments = artifact.enrichments or []
+        for enrichment in enrichments:
+            if enrichment.type == TI_ENRICHMENT_TYPE and enrichment.provider == TI_PROVIDER:
+                enrichment.data = json.dumps(ti_result)
+                break
+        else:
+            enrichment = EnrichmentModel(
+                name="TI Enrichment",
+                type=TI_ENRICHMENT_TYPE,
+                provider=TI_PROVIDER,
+                value=artifact.value,
+                data=json.dumps(ti_result),
+            )
+            enrichments.append(enrichment)
+        model_tmp = ArtifactModel(row_id=artifact.row_id, enrichments=enrichments)
+        Artifact.update(model_tmp)
+
+    @staticmethod
+    def _collect_unique_artifacts(case):
+        artifacts = {}
+        artifact_refs = 0
+        for alert in case.alerts or []:
+            for artifact in alert.artifacts or []:
+                artifact_refs += 1
+                if artifact and artifact.row_id and artifact.row_id not in artifacts:
+                    artifacts[artifact.row_id] = artifact
+        return artifact_refs, artifacts
+
     def run(self):
         try:
-            artifact = Artifact.get(self.param_source_row_id)
-            self.logger.info(f"Querying threat intelligence for : {artifact}")
+            case_row_id = self.param_source_row_id
+            case = Case.get(case_row_id, lazy_load=False)
+            if not case:
+                message = f"Case not found. row_id: {case_row_id}"
+                self.logger.error(message)
+                self.update_playbook_status(PlaybookJobStatus.FAILED, message)
+                return
 
-            if artifact.type in ["IP Address"]:
-                if is_ipaddress(artifact.value):
-                    ti_result = AlienVaultOTX().query_ip(artifact.value)
-                else:
-                    ti_result = {"error": "Invalid IP address format."}
-            elif artifact.type in ["Hash"]:
-                ti_result = AlienVaultOTX().query_file(artifact.value)
-            else:
-                ti_result = {"error": "Unsupported type."}
+            artifact_refs, artifacts = self._collect_unique_artifacts(case)
+            stats = {
+                "alerts": len(case.alerts or []),
+                "artifacts": artifact_refs,
+                "unique": len(artifacts),
+                "enriched": 0,
+                "unsupported": 0,
+                "invalid": 0,
+                "errors": 0,
+            }
 
-            enrichments = artifact.enrichments
-            for enrichment in enrichments:
-                if enrichment.type == "Threat Intelligence" and enrichment.provider == "OTX":
-                    enrichment.data = json.dumps(ti_result)
-                    break
-            else:
-                enrichment = EnrichmentModel(name="TI Enrichment", type="Threat Intelligence", provider="OTX", value=artifact.value,
-                                             data=json.dumps(ti_result))
-                enrichments.append(enrichment)
-            model_tmp = ArtifactModel(row_id=artifact.row_id, enrichments=enrichments)
-            Artifact.update(model_tmp)
+            for artifact in artifacts.values():
+                try:
+                    self.logger.info(f"Querying threat intelligence for artifact: {artifact}")
+                    ti_result = self._query_ti(artifact)
+                    if ti_result.get("error") == "Unsupported type.":
+                        stats["unsupported"] += 1
+                    elif ti_result.get("error") == "Invalid IP address format.":
+                        stats["invalid"] += 1
+                    self._update_artifact_enrichment(artifact, ti_result)
+                    stats["enriched"] += 1
+                except Exception as e:
+                    stats["errors"] += 1
+                    self.logger.exception(
+                        f"Error during TI enrichment for artifact row_id={artifact.row_id}, "
+                        f"type={artifact.type}, value={artifact.value}: {e}"
+                    )
 
-            self.update_playbook_status(PlaybookJobStatus.SUCCESS, "Threat intelligence enrichment completed.")
+            message = (
+                "Threat intelligence enrichment completed. "
+                f"alerts={stats['alerts']}, artifacts={stats['artifacts']}, unique={stats['unique']}, "
+                f"enriched={stats['enriched']}, unsupported={stats['unsupported']}, "
+                f"invalid={stats['invalid']}, errors={stats['errors']}"
+            )
+            self.update_playbook_status(PlaybookJobStatus.SUCCESS, message)
         except Exception as e:
             self.logger.exception(e)
             self.update_playbook_status(PlaybookJobStatus.FAILED, f"Error during TI enrichment: {e}")
@@ -55,7 +148,7 @@ if __name__ == "__main__":
 
     os.environ.setdefault("DJANGO_SETTINGS_MODULE", "ASP.settings")
     django.setup()
-    model = PlaybookModel(source_row_id='73ed8a06-38e9-4d03-8d17-74b578f0cafa')
+    model = PlaybookModel(source_row_id='3a22cbbf-5b33-4727-aa99-0ab8f763c196')
     module = Playbook()
     module._playbook_model = model
 

PLAYBOOKS/TI_Enrichment_By_Mock.py

@@ -1,9 +1,56 @@
 import json
 
 from Lib.baseplaybook import BasePlaybook
-from PLUGINS.SIRP.sirpapi import Artifact
+from PLUGINS.SIRP.sirpapi import Artifact, Case
+from PLUGINS.SIRP.sirpcoremodel import EnrichmentModel, ArtifactModel, ArtifactType
 from PLUGINS.SIRP.sirpextramodel import PlaybookJobStatus, PlaybookModel
-from PLUGINS.SIRP.sirpcoremodel import EnrichmentModel, ArtifactModel
+
+TI_ENRICHMENT_TYPE = "Threat Intelligence"
+TI_PROVIDER = "MockTIProvider"
+
+
+def _mock_ip_result(value: str) -> dict:
+    return {
+        "malicious": True,
+        "score": 85,
+        "indicator_type": "ip",
+        "indicator": value,
+        "description": "This IP is associated with known malicious activities.",
+        "source": "MockTIProvider",
+        "last_seen": "2024-10-01T12:34:56Z",
+    }
+
+
+def _mock_hash_result(value: str) -> dict:
+    return {
+        "malicious": True,
+        "score": 90,
+        "indicator_type": "file",
+        "indicator": value,
+        "description": "This file hash is associated with known malware samples.",
+        "source": "MockTIProvider",
+        "last_seen": "2024-10-01T12:34:56Z",
+    }
+
+
+def _mock_url_result(value: str) -> dict:
+    return {
+        "malicious": True,
+        "score": 80,
+        "indicator_type": "url",
+        "indicator": value,
+        "description": "This URL is associated with suspicious or malicious activity.",
+        "source": "MockTIProvider",
+        "last_seen": "2024-10-01T12:34:56Z",
+    }
+
+
+MOCK_QUERY_HANDLERS = {
+    ArtifactType.IP_ADDRESS: _mock_ip_result,
+    ArtifactType.HASH: _mock_hash_result,
+    ArtifactType.URL_STRING: _mock_url_result,
+    ArtifactType.UNIFORM_RESOURCE_LOCATOR: _mock_url_result,
+}
 
 
 class Playbook(BasePlaybook):
@@ -13,28 +60,100 @@ class Playbook(BasePlaybook):
     def __init__(self):
         super().__init__()  # do not delete this code
 
-    def run(self):
-        artifact = Artifact.get(self.param_source_row_id)
+    @staticmethod
+    def _normalize_artifact_type(artifact_type):
+        if isinstance(artifact_type, ArtifactType):
+            return artifact_type
+        try:
+            return ArtifactType(artifact_type)
+        except ValueError:
+            return artifact_type
 
-        # Simulate querying a threat intelligence database. In a real application, this should call an external API or database.
-        if artifact.type not in ["IP Address", "Hash"]:
-            self.update_playbook_status(PlaybookJobStatus.FAILED, "Unsupported type. Please use 'IP Address', 'Hash'.")
-            return
-        else:
-            ti_result = {"malicious": True, "score": 85, "description": "This IP is associated with known malicious activities.", "source": "ThreatIntelDB",
-                         "last_seen": "2024-10-01T12:34:56Z"}
-        enrichments = artifact.enrichments
+    def _query_ti(self, artifact) -> dict:
+        artifact_type = self._normalize_artifact_type(artifact.type)
+        handler = MOCK_QUERY_HANDLERS.get(artifact_type)
+        if not handler:
+            return {
+                "error": "Unsupported type.",
+                "artifact_type": str(artifact.type),
+                "supported_types": [artifact_type.value for artifact_type in MOCK_QUERY_HANDLERS],
+            }
+        return handler(artifact.value or "")
+
+    @staticmethod
+    def _update_artifact_enrichment(artifact, ti_result: dict):
+        enrichments = artifact.enrichments or []
         for enrichment in enrichments:
-            if enrichment.type == "Threat Intelligence" and enrichment.provider == "MockTIProvider":
+            if enrichment.type == TI_ENRICHMENT_TYPE and enrichment.provider == TI_PROVIDER:
                 enrichment.data = json.dumps(ti_result)
                 break
         else:
-            enrichment = EnrichmentModel(name="Mock TI Enrichment", type="Mock Threat Intelligence", provider="MockTIProvider", value=artifact.value,
-                                         data=json.dumps(ti_result))
+            enrichment = EnrichmentModel(
+                name="Mock TI Enrichment",
+                type=TI_ENRICHMENT_TYPE,
+                provider=TI_PROVIDER,
+                value=artifact.value,
+                data=json.dumps(ti_result),
+            )
             enrichments.append(enrichment)
         model_tmp = ArtifactModel(row_id=artifact.row_id, enrichments=enrichments)
         Artifact.update(model_tmp)
-        self.update_playbook_status(PlaybookJobStatus.SUCCESS, "Threat intelligence enrichment completed.")
+
+    @staticmethod
+    def _collect_unique_artifacts(case):
+        artifacts = {}
+        artifact_refs = 0
+        for alert in case.alerts or []:
+            for artifact in alert.artifacts or []:
+                artifact_refs += 1
+                if artifact and artifact.row_id and artifact.row_id not in artifacts:
+                    artifacts[artifact.row_id] = artifact
+        return artifact_refs, artifacts
+
+    def run(self):
+        try:
+            case_row_id = self.param_source_row_id
+            case = Case.get(case_row_id, lazy_load=False)
+            if not case:
+                message = f"Case not found. row_id: {case_row_id}"
+                self.logger.error(message)
+                self.update_playbook_status(PlaybookJobStatus.FAILED, message)
+                return
+
+            artifact_refs, artifacts = self._collect_unique_artifacts(case)
+            stats = {
+                "alerts": len(case.alerts or []),
+                "artifacts": artifact_refs,
+                "unique": len(artifacts),
+                "enriched": 0,
+                "unsupported": 0,
+                "errors": 0,
+            }
+
+            for artifact in artifacts.values():
+                try:
+                    self.logger.info(f"Mock threat intelligence enrichment for artifact: {artifact}")
+                    ti_result = self._query_ti(artifact)
+                    if ti_result.get("error") == "Unsupported type.":
+                        stats["unsupported"] += 1
+                    self._update_artifact_enrichment(artifact, ti_result)
+                    stats["enriched"] += 1
+                except Exception as e:
+                    stats["errors"] += 1
+                    self.logger.exception(
+                        f"Error during mock TI enrichment for artifact row_id={artifact.row_id}, "
+                        f"type={artifact.type}, value={artifact.value}: {e}"
+                    )
+
+            message = (
+                "Mock threat intelligence enrichment completed. "
+                f"alerts={stats['alerts']}, artifacts={stats['artifacts']}, unique={stats['unique']}, "
+                f"enriched={stats['enriched']}, unsupported={stats['unsupported']}, errors={stats['errors']}"
+            )
+            self.update_playbook_status(PlaybookJobStatus.SUCCESS, message)
+        except Exception as e:
+            self.logger.exception(e)
+            self.update_playbook_status(PlaybookJobStatus.FAILED, f"Error during mock TI enrichment: {e}")
         return
 
 
@@ -44,7 +163,7 @@ if __name__ == "__main__":
 
     os.environ.setdefault("DJANGO_SETTINGS_MODULE", "ASP.settings")
     django.setup()
-    model = PlaybookModel(source_row_id='250b934c-360b-426e-8a4a-3de022efa99e')
+    model = PlaybookModel(source_row_id='44958bcb-31ab-4fdf-85e7-60e02f9677f2')
     module = Playbook()
     module._playbook_model = model
     module.run()

PLUGINS/AlienVaultOTX/alienvaultotx.py

@@ -5,6 +5,10 @@ import requests
 from PLUGINS.AlienVaultOTX.CONFIG import API_KEY, HTTP_PROXY
 
 
+MAX_PULSE_SUMMARIES = 5
+MAX_LIST_ITEMS = 12
+
+
 class AlienVaultOTX(object):
     headers = {
         "accept": "application/json",
@@ -24,19 +28,16 @@ class AlienVaultOTX(object):
             parts = indicator.split('.')
             if all(0 <= int(part) <= 255 for part in parts):
                 result = cls.query_ip(indicator)
-                result['indicator_type'] = 'ip'
                 return result
 
         if re.match(r'^[a-fA-F0-9]{32}$|^[a-fA-F0-9]{40}$|^[a-fA-F0-9]{64}$', indicator):
             result = cls.query_file(indicator)
-            result['indicator_type'] = 'file'
             return result
 
         url_pattern = r'^(https?://|ftp://|www\.)'
         domain_pattern = r'\.'
         if re.match(url_pattern, indicator, re.IGNORECASE) or (re.search(domain_pattern, indicator) and '/' in indicator):
             result = cls.query_url(indicator)
-            result['indicator_type'] = 'url'
             return result
 
         return {
@@ -63,8 +64,10 @@ class AlienVaultOTX(object):
 
         url = f"{cls.base_url}/indicators/IPv4/{ip}/general"
         req_result = cls._get(url)
+        if req_result.get("error"):
+            return cls.summarize_result(req_result, "ip", ip)
         req_result["reputation_score"] = cls.calculate_reputation_score(req_result)
-        return req_result
+        return cls.summarize_result(req_result, "ip", ip)
 
     @classmethod
     def query_url(cls, url: str) -> dict:
@@ -87,9 +90,7 @@ class AlienVaultOTX(object):
             otx_url = f"{cls.base_url}/indicators/url/{encoded_url}/general"
 
             result = cls._get(otx_url)
-            if result and not result.get('error'):
-                result['original_url'] = url
-            return result
+            return cls.summarize_result(result, "url", url)
         except Exception as e:
             return {"error": str(e)}
 
@@ -121,8 +122,124 @@ class AlienVaultOTX(object):
 
         url = f"{cls.base_url}/indicators/file/{file_hash}/general"
         req_result = cls._get(url)
+        if req_result.get("error"):
+            return cls.summarize_result(req_result, "file", file_hash)
         req_result["reputation_score"] = cls.calculate_reputation_score(req_result)
-        return req_result
+        return cls.summarize_result(req_result, "file", file_hash)
+
+    @classmethod
+    def summarize_result(cls, attributes: dict, indicator_type: str, indicator: str) -> dict:
+        if attributes.get("error"):
+            return {
+                "indicator": indicator,
+                "indicator_type": indicator_type,
+                "provider": "AlienVault OTX",
+                "error": attributes.get("error"),
+            }
+
+        pulse_info = attributes.get("pulse_info") or {}
+        pulses = pulse_info.get("pulses") or []
+        reputation_score = attributes.get("reputation_score")
+        pulse_count = pulse_info.get("count", len(pulses))
+
+        summary = {
+            "indicator": indicator,
+            "indicator_type": indicator_type,
+            "provider": "AlienVault OTX",
+            "risk_level": cls._risk_level(reputation_score, pulse_count),
+            "reputation_score": reputation_score,
+            "pulse_count": pulse_count,
+            "tags": cls._limit_list(cls._unique_items(tag for pulse in pulses for tag in pulse.get("tags", []))),
+            "attack_techniques": cls._limit_list(cls._extract_attack_techniques(pulses)),
+            "malware_families": cls._limit_list(cls._extract_related_values(pulse_info, "malware_families")),
+            "adversaries": cls._limit_list(cls._extract_related_values(pulse_info, "adversary")),
+            "industries": cls._limit_list(cls._extract_related_values(pulse_info, "industries")),
+            "validation": cls._compact_named_items(attributes.get("validation") or []),
+            "false_positive": cls._compact_named_items(attributes.get("false_positive") or []),
+            "pulses": cls._compact_pulses(pulses),
+        }
+
+        network_context = cls._network_context(attributes)
+        if network_context:
+            summary["network_context"] = network_context
+
+        return summary
+
+    @staticmethod
+    def _unique_items(items) -> list:
+        unique = []
+        for item in items:
+            if item and item not in unique:
+                unique.append(item)
+        return unique
+
+    @staticmethod
+    def _limit_list(items: list, limit: int = MAX_LIST_ITEMS) -> list:
+        return items[:limit]
+
+    @classmethod
+    def _extract_attack_techniques(cls, pulses: list) -> list:
+        techniques = []
+        for pulse in pulses:
+            for attack in pulse.get("attack_ids", []) or []:
+                display_name = attack.get("display_name") or attack.get("name") or attack.get("id")
+                if display_name:
+                    techniques.append(display_name)
+        return cls._unique_items(techniques)
+
+    @classmethod
+    def _extract_related_values(cls, pulse_info: dict, key: str) -> list:
+        related = pulse_info.get("related") or {}
+        values = []
+        for source in ("alienvault", "other"):
+            values.extend((related.get(source) or {}).get(key, []) or [])
+        return cls._unique_items(values)
+
+    @classmethod
+    def _compact_named_items(cls, items: list) -> list:
+        compact = []
+        for item in items:
+            if isinstance(item, dict):
+                value = item.get("name") or item.get("source") or item.get("description")
+                if value:
+                    compact.append(value)
+            elif item:
+                compact.append(item)
+        return cls._limit_list(cls._unique_items(compact))
+
+    @classmethod
+    def _compact_pulses(cls, pulses: list) -> list:
+        compact = []
+        for pulse in pulses[:MAX_PULSE_SUMMARIES]:
+            compact.append({
+                "name": pulse.get("name"),
+                "description": pulse.get("description"),
+                "tags": cls._limit_list(pulse.get("tags", [])),
+                "attack_techniques": cls._limit_list(cls._extract_attack_techniques([pulse])),
+                "malware_families": cls._limit_list(pulse.get("malware_families", []) or []),
+                "adversary": pulse.get("adversary"),
+                "created": pulse.get("created"),
+                "modified": pulse.get("modified"),
+                "tlp": pulse.get("TLP"),
+            })
+        return compact
+
+    @staticmethod
+    def _network_context(attributes: dict) -> dict:
+        context = {}
+        for field in ("asn", "country_name", "country_code", "region", "city"):
+            if attributes.get(field):
+                context[field] = attributes.get(field)
+        return context
+
+    @staticmethod
+    def _risk_level(reputation_score, pulse_count: int) -> str:
+        score = reputation_score or 0
+        if score >= 50 or pulse_count >= 5:
+            return "high"
+        if score >= 20 or pulse_count > 0:
+            return "medium"
+        return "low"
 
     @classmethod
     def _get(cls, url: str) -> dict: