diff --git a/PLAYBOOKS/TI_Enrichment_By_AlienVaultOTX.py b/PLAYBOOKS/TI_Enrichment_By_AlienVaultOTX.py index cb5bb8b..944160b 100644 --- a/PLAYBOOKS/TI_Enrichment_By_AlienVaultOTX.py +++ b/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 diff --git a/PLAYBOOKS/TI_Enrichment_By_Mock.py b/PLAYBOOKS/TI_Enrichment_By_Mock.py index 6aa5c7d..5ea19c7 100644 --- a/PLAYBOOKS/TI_Enrichment_By_Mock.py +++ b/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() diff --git a/PLUGINS/AlienVaultOTX/alienvaultotx.py b/PLUGINS/AlienVaultOTX/alienvaultotx.py index 34f8d9b..1d1c88a 100644 --- a/PLUGINS/AlienVaultOTX/alienvaultotx.py +++ b/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: