import demistomock as demisto from CommonServerPython import * from CommonServerUserPython import * from typing import Dict, Any, List, Optional, Tuple ''' CONSTANTS ''' VENDOR = 'Threat Crowd' DATE_FORMAT = '%Y-%m-%dT%H:%M:%SZ' # ISO8601 format with UTC, default in XSOAR ''' CLIENT CLASS ''' class Client(BaseClient): def __init__(self, base_url: str, verify: bool, proxy: bool, reliability: DBotScoreReliability, entry_limit: int): super().__init__(base_url=base_url, verify=verify, proxy=proxy) self.reliability = reliability self.entry_limit = entry_limit def _get_limit_for_command(self, command_limit: int = None): if command_limit: return None if command_limit == -1 else command_limit else: return self.entry_limit ''' HELPER FUNCTIONS ''' def _get_list_without_empty(list_to_change: list) -> list: if list_to_change: return [entry for entry in list_to_change if entry] return list_to_change def handle_resolutions(resolutions: List[dict], limit: Optional[int]) -> List[dict]: """ Gets a resolution section from response with following struct: [{"last_resolved": "2014-12-14", "domain": "example.com"}] return a sorted list truncated to limit, desc by time.""" resolutions = resolutions[:limit] resolutions.sort(key=lambda x: x['last_resolved'], reverse=True) return resolutions def _get_dbot_score(json_res: dict) -> Tuple[int, str]: """ Calculates DBot score according to https://github.com/AlienVault-OTX/ApiV2/blob/master/README.md#votes """ # checks that response is valid if json_res.get('response_code') == '1': votes = json_res.get('votes') if votes == -1: return Common.DBotScore.BAD, 'BAD' elif votes == 0: return Common.DBotScore.SUSPICIOUS, 'SUSPICIOUS' elif votes == 1: return Common.DBotScore.GOOD, 'GOOD' return Common.DBotScore.NONE, 'None' ''' COMMAND FUNCTIONS ''' def ip_command(client: Client, args: Dict[str, Any]) -> List[CommandResults]: command_results: List[CommandResults] = [] api_url = 'ip/report/' entries_limit = client._get_limit_for_command(arg_to_number(args.get('limit'), 'limit', False)) ips = argToList(args.get('ip')) for ip in ips: res = client._http_request(method='GET', url_suffix=api_url, params={'ip': ip}) # adding value to both outputs and raw results as it is not provided in the API response res['value'] = ip score, score_str = _get_dbot_score(res) dbot = Common.DBotScore( ip, DBotScoreType.IP, VENDOR, score, reliability=client.reliability) ip_object = Common.IP(ip, dbot) hashes = res.get('hashes', [])[:entries_limit] resolutions = handle_resolutions(res.get('resolutions', []), entries_limit) markdown = f"### Threat crowd report for ip {ip}: \n ### DBotScore: {score_str} \n" \ f"{tableToMarkdown('Resolutions', resolutions, removeNull=True)} \n " \ f"{tableToMarkdown('Hashes', hashes, headers='Hashes', removeNull=True)}" \ f"{tableToMarkdown('References', res.get('references'), removeNull=True, headers='References')}" outputs = { 'hashes': hashes, 'permalink': res.get('permalink'), 'resolutions': resolutions, 'references': res.get('references'), 'response_code': res.get('response_code'), 'votes': res.get('votes'), 'value': res.get('value') } # using res.copy() to avoid changing all previous entries's values. command_results.append(CommandResults( outputs_prefix='ThreatCrowd.IP', outputs=outputs, outputs_key_field='value', indicator=ip_object, readable_output=markdown, raw_response=res.copy() )) return command_results def email_command(client: Client, args: Dict[str, Any]) -> List[CommandResults]: command_results: List[CommandResults] = [] api_url = 'email/report/' emails = argToList(args.get('email')) for email in emails: res = client._http_request(method='GET', url_suffix=api_url, params={'email': email}) # adding value to both outputs and raw results as it is not provided in the API response res['value'] = email score, score_str = _get_dbot_score(res) dbot = Common.DBotScore( email, DBotScoreType.EMAIL, VENDOR, score, reliability=client.reliability) email_object = Common.EMAIL(email, dbot) markdown = f"### Threat crowd report for Email {email} \n " \ f"DBotScore: {score_str} \n {tableToMarkdown('Results', res, removeNull=True)}" # using res.copy() to avoid changing all previous entries's values. command_results.append(CommandResults( outputs_prefix='ThreatCrowd.Account', outputs=res.copy(), outputs_key_field='value', indicator=email_object, readable_output=markdown, raw_response=res.copy() )) return command_results def domain_command(client: Client, args: Dict[str, Any]) -> List[CommandResults]: command_results: List[CommandResults] = [] api_url = 'domain/report/' entries_limit = client._get_limit_for_command(arg_to_number(args.get('limit'), 'limit', False)) domains = argToList(args.get('domain')) for domain in domains: res = client._http_request(method='GET', url_suffix=api_url, params={'domain': domain}) # adding value to both outputs and raw results as it is not provided in the API response res['value'] = domain score, score_str = _get_dbot_score(res) dbot = Common.DBotScore( domain, DBotScoreType.DOMAIN, VENDOR, score, reliability=client.reliability) domain_object = Common.Domain(domain, dbot) markdown = f"### Threat crowd report for domain {domain} \n ### DBotScore: {score_str} \n" subdomains = res.get('subdomains')[:entries_limit] resolutions = handle_resolutions(res.get('resolutions', []), entries_limit) markdown += f'{tableToMarkdown("Resolutions", resolutions, removeNull=True)} \n ' \ f'{tableToMarkdown("Subdomains", subdomains, headers=["subdomains"])} \n' res_to_show = res.copy() res_to_show.pop("resolutions") res_to_show.pop("subdomains") markdown += f'{tableToMarkdown(" ", res_to_show, removeNull=True)}' outputs = { 'hashes': res.get('hashes'), 'permalink': res.get('permalink'), 'resolutions': resolutions, 'references': res.get('references'), 'response_code': res.get('response_code'), 'votes': res.get('votes'), 'subdomains': subdomains, 'value': res.get('value') } # using res.copy() to avoid changing all previous entries's values. command_results.append(CommandResults( outputs_prefix='ThreatCrowd.Domain', outputs=outputs, outputs_key_field='value', indicator=domain_object, readable_output=markdown, raw_response=res.copy() )) return command_results def antivirus_command(client: Client, args: Dict[str, Any]) -> List[CommandResults]: command_results: List[CommandResults] = [] api_url = 'antivirus/report/' antivirus_list = argToList(args.get('antivirus')) entries_limit = client._get_limit_for_command(arg_to_number(args.get('limit'), 'limit', False)) for antivirus in antivirus_list: res = client._http_request(method='GET', url_suffix=api_url, params={'antivirus': antivirus}) # adding value to both outputs and raw results as it is not provided in the API response res['value'] = antivirus res['hashes'] = res['hashes'][:entries_limit] markdown = tableToMarkdown(f"Threat crowd report for antivirus {antivirus}", res, removeNull=True) # using res.copy() to avoid changing all previous entries's values. command_results.append(CommandResults( outputs_prefix='ThreatCrowd.AntiVirus', outputs=res.copy(), outputs_key_field='value', readable_output=markdown, raw_response=res.copy() )) return command_results def file_command(client: Client, args: Dict[str, Any]) -> List[CommandResults]: command_results: List[CommandResults] = [] api_url = 'file/report/' files = argToList(args.get('file')) entries_limit = client._get_limit_for_command(arg_to_number(args.get('limit'), 'limit', False)) for file_hash in files: res = client._http_request(method='GET', url_suffix=api_url, params={'resource': file_hash}) # adding value to both outputs and raw results as it is not provided in the API response res['value'] = file_hash score, score_str = _get_dbot_score(res) dbot = Common.DBotScore( file_hash, DBotScoreType.FILE, VENDOR, score, reliability=client.reliability) file_object = Common.File(dbot, md5=res.get('md5'), sha1=res.get('sha1')) # removes empty entries returned by API res['scans'] = _get_list_without_empty(res.get('scans'))[:entries_limit] markdown = f"### Threat crowd report for File {file_hash}: \n ### DBotScore: {score_str} \n " \ f"{tableToMarkdown('Results', res)}".replace('
', '') # using res.copy() to avoid changing all previous entries's values. command_results.append(CommandResults( outputs_prefix='ThreatCrowd.File', outputs=res.copy(), outputs_key_field='value', indicator=file_object, readable_output=markdown, raw_response=res.copy() )) return command_results def test_module(client: Client) -> str: try: api_url = 'ip/report/' client._http_request(method='GET', url_suffix=api_url, params={'ip': '1.1.1.1'}) message = 'ok' except DemistoException as e: if 'Forbidden' in str(e) or 'Authorization' in str(e): message = 'Authorization Error: make sure API Key is correctly set.' else: raise e return message ''' MAIN FUNCTION ''' def main() -> None: command_functions = {'email': email_command, 'domain': domain_command, 'ip': ip_command, 'threat-crowd-antivirus': antivirus_command, 'file': file_command} command = demisto.command() demisto.debug(f'Command being called is {command}') try: params = demisto.params() base_url = params.get('server_url') verify_certificate = not params.get('insecure', False) proxy = params.get('proxy', False) entry_limit = arg_to_number(params.get('entry_limit'), 'entry_limit', True) if not entry_limit: raise Exception("Please Provide a limit for number of entries. To receive all entries use -1") reliability = params.get('integrationReliability') reliability = reliability if reliability else DBotScoreReliability.C if DBotScoreReliability.is_valid_type(reliability): reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(reliability) else: raise Exception("Please provide a valid value for the Source Reliability parameter.") client = Client( base_url=base_url, verify=verify_certificate, proxy=proxy, reliability=reliability, entry_limit=entry_limit ) if command == 'test-module': result = test_module(client) return_results(result) else: if command in command_functions: return_results(command_functions[command](client, demisto.args())) else: raise NotImplementedError(f'command {command} is not implemented.') # Log exceptions and return errors except Exception as e: return_error(f'Failed to execute {command} command.\nError:\n{str(e)}') ''' ENTRY POINT ''' if __name__ in ('__main__', '__builtin__', 'builtins'): main()