import demistomock as demisto from CommonServerPython import * from CommonServerUserPython import * ''' IMPORTS ''' import requests import traceback from typing import Dict, Optional, List, Any, Callable, Collection # Disable insecure warnings requests.packages.urllib3.disable_warnings() ''' CONSTANTS ''' # Integration information INTEGRATION_NAME = 'illuminate' INTEGRATION_CONTEXT_BRAND = 'Illuminate' MALICIOUS_DATA: Dict[str, str] = { 'Vendor': 'illuminate', 'Description': 'illuminate has determined that this indicator is malicious via internal analysis.' } ''' HELPER FUNCTIONS ''' class IdNamePair(object): def __init__(self, unique_id: int, name: str): self.id = unique_id self.name = name def __str__(self): return f'id = {self.id}, name = {self.name}' class EnrichmentOutput(object): def __init__(self, illuminate_context_data: dict, raw_data: dict, indicator_type: str) -> None: self.illuminate_context_data = illuminate_context_data self.raw_data = raw_data self.indicator_type = indicator_type self.reputation_context: dict = {} def get_human_readable_output(self) -> str: human_readable_data = self.illuminate_context_data.copy() human_readable_data['Actors'] = [IdNamePair(d['id'], d['name']) for d in human_readable_data['Actors']] human_readable_data['Malwares'] = [IdNamePair(d['id'], d['name']) for d in human_readable_data['Malwares']] return tableToMarkdown( t=human_readable_data, name=f'{INTEGRATION_NAME} {self.indicator_type.capitalize()} Information', removeNull=True ) def build_illuminate_context(self) -> dict: return { f'{INTEGRATION_CONTEXT_BRAND}.{self.indicator_type.capitalize()}(val.ID && val.ID === obj.ID)': self.illuminate_context_data } def generate_reputation_context( self, primary_key: str, indicator_value: str, indicator_type: str, reputation_key: str, extra_context: Optional[dict] = None ): if self.has_context_data(): reputation_context: Dict[str, Any] = {primary_key: indicator_value} if extra_context is not None: reputation_context.update(extra_context) malicious = Client.is_indicator_malicious(self.raw_data) if malicious: reputation_context['Malicious'] = MALICIOUS_DATA self.add_reputation_context( f'{reputation_key}(val.{primary_key} && val.{primary_key} === obj.{primary_key})', reputation_context ) self.add_reputation_context('DBotScore', { 'Indicator': indicator_value, 'Score': 3 if malicious else 1, 'Type': indicator_type, 'Vendor': INTEGRATION_NAME }) def build_all_context(self) -> dict: all_context = {} all_context.update(self.build_illuminate_context()) if len(self.reputation_context) > 0: all_context.update(self.reputation_context) return all_context def return_outputs(self): # We need to use the underlying demisto.results function call rather than using return_outputs because # we need to add the IgnoreAutoExtract key to ensure that our illuminate links are not marked as indicators entry = { "Type": entryTypes["note"], "HumanReadable": self.get_human_readable_output(), "ContentsFormat": formats["json"], "Contents": self.raw_data, "EntryContext": self.build_all_context(), "IgnoreAutoExtract": True } demisto.results(entry) def add_illuminate_context(self, key: str, data: Any): self.illuminate_context_data[key] = data def add_reputation_context(self, key: str, context: dict): self.reputation_context[key] = context def has_context_data(self): return len(self.illuminate_context_data) > 0 class Client(BaseClient): def __init__(self, server: str, username: str, password: str, insecure: bool, proxy: bool): # NB: 404 is a valid response since that just means no entries, and we want the UI to respect that and show "No Entries" super().__init__( base_url=f'https://{server}/api/1_0/', verify=not insecure, proxy=proxy, auth=(username, password), ok_codes=(200, 404) ) def indicator_search(self, indicator_type: str, indicator: str) -> dict: params = {'type': indicator_type, 'value': indicator} return self._http_request(method='GET', url_suffix='indicator/match', params=params) def perform_test_request(self): data: dict = self._http_request(method='GET', url_suffix='') if data['links'] is None: raise DemistoException('Invalid URL or Credentials. JSON structure not recognized') def enrich_indicator(self, indicator: str, indicator_type: str) -> EnrichmentOutput: raw_data: dict = self.indicator_search(indicator_type, indicator) if raw_data is None: return EnrichmentOutput({}, {}, indicator_type) context_data = self.get_context_from_response(raw_data) return EnrichmentOutput(context_data, raw_data, indicator_type) @staticmethod def get_data_key(data: dict, key: str) -> Optional[Any]: return None if key not in data else data[key] @staticmethod def get_nested_data_key(data: dict, key: str, nested_key: str) -> Optional[Any]: top_level = Client.get_data_key(data, key) return None if top_level is None or nested_key not in top_level else top_level[nested_key] @staticmethod def get_data_key_as_date(data: dict, key: str, fmt: str) -> Optional[str]: value = Client.get_data_key(data, key) return None if value is None else datetime.fromtimestamp(value / 1000.0).strftime(fmt) @staticmethod def get_data_key_as_list(data: dict, key: str) -> List[Any]: data_list = Client.get_data_key(data, key) return [] if data_list is None or not isinstance(data[key], (list,)) else data_list @staticmethod def get_data_key_as_list_of_values(data: dict, key: str, value_key: str) -> List[Any]: data_list = Client.get_data_key_as_list(data, key) return [value_data[value_key] for value_data in data_list] @staticmethod def get_data_key_as_list_of_dicts(data: dict, key: str, dict_creator: Callable) -> Collection[Any]: data_list = Client.get_data_key_as_list(data, key) return {} if len(data_list) == 0 else [dict_creator(value_data) for value_data in data_list] @staticmethod def is_indicator_malicious(data: dict) -> bool: benign = Client.get_nested_data_key(data, 'benign', 'value') return False if benign is None or benign is True else True @staticmethod def get_context_from_response(data: dict) -> dict: result_dict = { 'ID': Client.get_data_key(data, 'id'), 'Indicator': Client.get_nested_data_key(data, 'value', 'name'), 'EvidenceCount': Client.get_data_key(data, 'reportCount'), 'Active': Client.get_data_key(data, 'active'), 'HitCount': Client.get_data_key(data, 'hitCount'), 'ConfidenceLevel': Client.get_nested_data_key(data, 'confidenceLevel', 'value'), 'FirstHit': Client.get_data_key_as_date(data, 'firstHit', '%Y-%m-%d'), 'LastHit': Client.get_data_key_as_date(data, 'lastHit', '%Y-%m-%d'), 'ReportedDates': Client.get_data_key_as_list_of_values(data, 'reportedDates', 'date'), 'ActivityDates': Client.get_data_key_as_list_of_values(data, 'activityDates', 'date'), 'Malwares': Client.get_data_key_as_list_of_dicts(data, 'malwares', lambda d: {'id': d['id'], 'name': d['name']}), 'Actors': Client.get_data_key_as_list_of_dicts(data, 'actors', lambda d: {'id': d['id'], 'name': d['name']}), 'Benign': Client.get_nested_data_key(data, 'benign', 'value'), 'IlluminateLink': None } links_list = Client.get_data_key_as_list(data, 'links') result_dict['IlluminateLink'] = next(( link['href'].replace("api/1_0/indicator/", "indicators/") for link in links_list if 'rel' in link and link['rel'] == 'self' and 'href' in link ), None) return result_dict def build_client(demisto_params: dict) -> Client: server: str = str(demisto_params.get('server')) proxy: bool = demisto_params.get('proxy', False) insecure: bool = demisto_params.get('insecure', False) credentials: dict = demisto_params.get('credentials', {}) username: str = str(credentials.get('identifier')) password: str = str(credentials.get('password')) return Client(server, username, password, insecure, proxy) ''' COMMAND EXECUTION ''' def perform_test_module(client: Client): client.perform_test_request() def domain_command(client: Client, args: dict) -> List[EnrichmentOutput]: domains: List[str] = argToList(args.get('domain')) enrichment_data_list: List[EnrichmentOutput] = [] for domain in domains: enrichment_data: EnrichmentOutput = client.enrich_indicator(domain, 'domain') if enrichment_data.has_context_data(): extra_context = {} ip_resolution = Client.get_nested_data_key(enrichment_data.raw_data, 'ipResolution', 'name') if ip_resolution is not None: enrichment_data.add_illuminate_context('IpResolution', ip_resolution) extra_context['DNS'] = ip_resolution enrichment_data.generate_reputation_context('Name', domain, 'domain', 'Domain', extra_context) enrichment_data_list.append(enrichment_data) return enrichment_data_list def email_command(client: Client, args: dict) -> List[EnrichmentOutput]: emails: List[str] = argToList(args.get('email')) enrichment_data_list: List[EnrichmentOutput] = [] for email in emails: enrichment_data: EnrichmentOutput = client.enrich_indicator(email, 'email') if enrichment_data.has_context_data(): enrichment_data.generate_reputation_context('From', email, 'email', 'Email') enrichment_data_list.append(enrichment_data) return enrichment_data_list def ip_command(client: Client, args: dict) -> List[EnrichmentOutput]: ips: List[str] = argToList(args.get('ip')) enrichment_data_list: List[EnrichmentOutput] = [] for ip in ips: enrichment_data: EnrichmentOutput = client.enrich_indicator(ip, 'ip') if enrichment_data.has_context_data(): enrichment_data.generate_reputation_context('Address', ip, 'ip', 'IP') enrichment_data_list.append(enrichment_data) return enrichment_data_list def file_command(client: Client, args: dict) -> List[EnrichmentOutput]: files: List[str] = argToList(args.get('file')) enrichment_data_list: List[EnrichmentOutput] = [] for file in files: enrichment_data: EnrichmentOutput = client.enrich_indicator(file, 'file') if enrichment_data.has_context_data(): hash_type = get_hash_type(file) if hash_type != 'Unknown': enrichment_data.generate_reputation_context(hash_type.upper(), file, 'file', 'File') enrichment_data_list.append(enrichment_data) return enrichment_data_list def illuminate_enrich_string_command(client: Client, args: dict) -> List[EnrichmentOutput]: strings: List[str] = argToList(args.get('string')) enrichment_data_list: List[EnrichmentOutput] = [] for string in strings: enrichment_data_list.append(client.enrich_indicator(string, 'string')) return enrichment_data_list def illuminate_enrich_ipv6_command(client: Client, args: dict) -> List[EnrichmentOutput]: ips: List[str] = argToList(args.get('ip')) enrichment_data_list: List[EnrichmentOutput] = [] for ip in ips: enrichment_data_list.append(client.enrich_indicator(ip, 'ipv6')) return enrichment_data_list def illuminate_enrich_mutex_command(client: Client, args: dict) -> List[EnrichmentOutput]: mutexes: List[str] = argToList(args.get('mutex')) enrichment_data_list: List[EnrichmentOutput] = [] for mutex in mutexes: enrichment_data_list.append(client.enrich_indicator(mutex, 'mutex')) return enrichment_data_list def illuminate_enrich_http_request_command(client: Client, args: dict) -> List[EnrichmentOutput]: http_requests: List[str] = argToList(args.get('http-request')) enrichment_data_list: List[EnrichmentOutput] = [] for http_request in http_requests: enrichment_data_list.append(client.enrich_indicator(http_request, 'httpRequest')) return enrichment_data_list def url_command(client: Client, args: dict) -> List[EnrichmentOutput]: urls: List[str] = argToList(args.get('url')) enrichment_data_list: List[EnrichmentOutput] = [] for url in urls: enrichment_data: EnrichmentOutput = client.enrich_indicator(url, 'url') if enrichment_data.has_context_data(): enrichment_data.generate_reputation_context('Data', url, 'url', 'URL') enrichment_data_list.append(enrichment_data) return enrichment_data_list ''' EXECUTION ''' def main(): commands = { 'domain': domain_command, 'email': email_command, 'file': file_command, 'ip': ip_command, 'url': url_command, 'illuminate-enrich-string': illuminate_enrich_string_command, 'illuminate-enrich-ipv6': illuminate_enrich_ipv6_command, 'illuminate-enrich-mutex': illuminate_enrich_mutex_command, 'illuminate-enrich-http-request': illuminate_enrich_http_request_command } command: str = demisto.command() LOG(f'command is {command}') try: client = build_client(demisto.params()) if command == 'test-module': perform_test_module(client) demisto.results('ok') elif command in commands: enrichment_outputs: List[EnrichmentOutput] = commands[command](client, demisto.args()) [e.return_outputs() for e in enrichment_outputs] except Exception as e: err_msg = f'Error in {INTEGRATION_NAME} Integration [{e}]\nTrace:\n{traceback.format_exc()}' return_error(err_msg, error=e) if __name__ in ['__main__', 'builtin', 'builtins']: main()