import copy import demistomock as demisto # noqa: F401 import urllib3 from CommonServerPython import * # noqa: F401 QUERIES = { "tag": "get_taginfo", "signature": "get_siginfo", "file_type": "get_file_type", "clamav": "get_clamavinfo", "imphash": "get_imphash", "yara_rule": "get_yarainfo", "issuer_cn": "get_issuerinfo", } EXCEPTIONS_MESSAGES = { "illegal_sha256_hash": "Illegal SHA256 hash provided.", "file_not_found": "The file was not found or is unknown to MalwareBazaar.", "hash_not_found": "The file (hash) you wanted to query is unknown to MalwareBazaar.", "illegal_hash": "The hash you provided is not a valid SHA256 hash.", "user_blacklisted": "Your API key is blacklisted.", "no_results": "Your query yield no results.", "not_found": "Tha value you wanted to query is unknown to MalwareBazaar.", "illegal": "The text you provided is not valid.", } VENDOR_NAME = "MalwareBazaar" LIST_HEADERS = ["md5_hash", "sha256_hash", "sha1_hash", "file_name", "file_type", "file_size", "tags", "first_seen", "last_seen"] FILE_HEADERS = [ "md5_hash", "sha256_hash", "sha1_hash", "file_name", "file_type", "file_size", "tags", "first_seen", "last_seen", "signature", "ssdeep", "reporter", "imphash", "yara_rules_names", ] class Client(BaseClient): def __init__(self, server_url, verify, proxy, headers, api_key): self.api_key = api_key super().__init__(base_url=server_url, verify=verify, proxy=proxy, headers=headers) def file_request(self, hash): response = self._http_request("POST", files={"query": (None, "get_info"), "hash": (None, hash)}) return response def malwarebazaar_download_sample_request(self, sha256_hash): response = self._http_request( "POST", files={"query": (None, "get_file"), "sha256_hash": (None, sha256_hash)}, resp_type="response" ) return response def malwarebazaar_comment_add_request(self, sha256_hash, comment): if self.api_key is None: raise Exception("API Key is required for this command") response = self._http_request( "POST", files={"query": (None, "add_comment"), "sha256_hash": (None, sha256_hash), "comment": (None, comment)}, ) return response def malwarebazaar_samples_list_request(self, sample_input, value, limit, query): files = { "query": (None, query), sample_input: (None, value), } if sample_input != "issuer_cn": files.update({"limit": (None, limit)}) response = self._http_request("POST", files=files) return response def file_process(hash, reliability, raw_response, response_data) -> CommandResults: """ creates CommandResults for every file in the list inserted to file_command Args: hash: raw_response: response_data: Returns: CommandResults for the relevant file """ dbot_score = Common.DBotScore( indicator=hash, indicator_type=DBotScoreType.FILE, integration_name=VENDOR_NAME, score=Common.DBotScore.BAD, reliability=reliability, malicious_description=response_data.get("comment"), ) signature = response_data.get("signature") relationship = EntityRelationship( name="indicator-of", entity_a=hash, entity_a_type="File", entity_b=signature, entity_b_type=FeedIndicatorType.indicator_type_by_server_version("STIX Malware"), source_reliability=reliability, brand=VENDOR_NAME, ) table_name = f"{VENDOR_NAME} File reputation for: {hash}" humam_readable_data = copy.deepcopy(response_data) humam_readable_data.update({"yara_rules_names": []}) rules = humam_readable_data.get("yara_rules", []) rules = rules if rules else [] for rule in rules: humam_readable_data.get("yara_rules_names").append(rule.get("rule_name")) md = tableToMarkdown( table_name, t=humam_readable_data, headerTransform=string_to_table_header, removeNull=True, headers=FILE_HEADERS ) file_object = Common.File( md5=response_data.get("md5_hash"), sha256=response_data.get("sha256_hash"), sha1=response_data.get("sha1_hash"), size=response_data.get("file_size"), file_type=response_data.get("file_type"), dbot_score=dbot_score, relationships=[relationship], ) return CommandResults( outputs_prefix="MalwareBazaar.File", outputs_key_field="md5_hash", outputs=response_data, raw_response=raw_response, indicator=file_object, relationships=[relationship], readable_output=md, ) def check_query_status(response, is_list_command=False, sample_type=None): """ checks whether the request to the API returned with the proper result Args: sample_type: string, type of sample (tag, signature, etc.) is_list_command: bool response: response from API """ not_found_error = "_not_found" illegal_error = "illegal_" query_status = response.get("query_status") if query_status != "ok" and query_status != "success": if is_list_command: if query_status == sample_type + not_found_error: raise Exception(EXCEPTIONS_MESSAGES.get("not_found")) if query_status == sample_type + illegal_error: raise Exception(EXCEPTIONS_MESSAGES.get("illegal")) if query_status in EXCEPTIONS_MESSAGES: raise Exception(EXCEPTIONS_MESSAGES.get(query_status)) else: raise Exception(query_status) def file_command(client: Client, args: Dict[str, Any]) -> List[CommandResults]: """ Args: client: args: file - list of files hash Returns: file reputation for the given hashes """ reliability = demisto.params().get("integrationReliability", DBotScoreReliability.A) 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.") file_list = argToList(args.get("file")) command_results: List[CommandResults] = [] for hash in file_list: raw_response = client.file_request(hash) if raw_response.get("query_status") == "hash_not_found": command_results.append(create_indicator_result_with_dbotscore_unknown(hash, DBotScoreType.FILE, reliability)) else: check_query_status(raw_response) response_data = raw_response.get("data")[0] if file_name := response_data.get("file_name"): response_data["file_name"] = "" if file_name == "file" else file_name command_results.append(file_process(hash, reliability, raw_response, response_data)) return command_results def malwarebazaar_download_sample_command(client: Client, args: Dict[str, Any]) -> CommandResults: """ Args: client: args: sha256_hash of file Returns: zip file contains the malware sample from MalwareBazaar """ sha256_hash = args.get("sha256_hash") response = client.malwarebazaar_download_sample_request(sha256_hash) filename = f"{sha256_hash}.zip" return fileResult(filename, response.content) def malwarebazaar_comment_add_command(client: Client, args: Dict[str, Any]) -> CommandResults: """ Args: client: args: sha256_hash of file, comment to add in context of this file Returns: query status of the request to MalwareBazaar (success or error) """ sha256_hash = args.get("sha256_hash") comment = args.get("comment") response = client.malwarebazaar_comment_add_request(sha256_hash, comment) check_query_status(response) readable_output = f"Comment added to {sha256_hash} malware sample successfully" outputs = { "sha256_hash": sha256_hash, "comment": comment, } return CommandResults( outputs_prefix="MalwareBazaar.MalwarebazaarCommentAdd", outputs_key_field="sha256_hash", outputs=outputs, readable_output=readable_output, raw_response=response, ) def malwarebazaar_samples_list_command(client: Client, args: Dict[str, Any]) -> CommandResults: """ Args: client: args: sample_type - {clamav, file_type, imphash, signature, tag, yara_rule} sample_value limit (optional) - number of results (default 50) Returns: query results from API """ sample_input = args.get("sample_type") or "" value = args.get("sample_value") limit = arg_to_number(args.get("limit")) if "limit" in args else None page = arg_to_number(args.get("page")) if "page" in args else None page_size = arg_to_number(args.get("page_size")) if "page_size" in args else None # # if limit was provided, request limit results from api, else, use pagination (if nothing is used 50 results will # # be requested as default) if limit is None: if page is not None and page_size is not None: if page <= 0: raise Exception("Chosen page number must be greater than 0") limit = page_size * page else: limit = 50 # # 1000 is the maximal value we can get from tha API limit = min(limit, 1000) query = QUERIES.get(sample_input) response = client.malwarebazaar_samples_list_request(sample_input, value, limit, query) check_query_status(response, True, args.get("sample_type")) response_data = response.get("data") # take required results from response if pagination by page and page_size if page is not None and page_size is not None: response_data = response_data[-1 * page_size :] readable_output = tableToMarkdown( "Sample List", t=response_data, headerTransform=string_to_table_header, removeNull=True, headers=LIST_HEADERS ) return CommandResults( outputs_prefix="MalwareBazaar.MalwarebazaarSamplesList", outputs_key_field="sha256_hash", readable_output=readable_output, outputs=response_data, raw_response=response, ) def test_module(client: Client) -> None: if client.api_key: response = client.malwarebazaar_comment_add_request( "094fd325049b8a9cf6d3e5ef2a6d4cc6a567d7d49c35f8bb8dd9e3c6acf3d78d", "test comment" ) else: response = client.malwarebazaar_samples_list_request("tag", "TrickBot", "2", QUERIES.get("tag")) check_query_status(response) return_results("ok") def main() -> None: params: Dict[str, Any] = demisto.params() args: Dict[str, Any] = demisto.args() url = params.get("url") api_key = params.get("credentials", {}).get("password") verify_certificate: bool = not params.get("insecure", False) proxy = params.get("proxy", False) command = demisto.command() demisto.debug(f"Command being called is {command}") try: if not api_key: raise ValueError("Missing required parameter Auth Key. Please set this parameter in the instance configuration.") urllib3.disable_warnings() client: Client = Client( urljoin(url, "/api/v1/"), verify_certificate, proxy, headers={"Auth-Key": api_key}, api_key=api_key ) commands = { "file": file_command, "malwarebazaar-download-sample": malwarebazaar_download_sample_command, "malwarebazaar-comment-add": malwarebazaar_comment_add_command, "malwarebazaar-samples-list": malwarebazaar_samples_list_command, } if command == "test-module": test_module(client) elif command in commands: return_results(commands[command](client, args)) else: raise NotImplementedError(f"{command} command is not implemented.") except Exception as e: return_error(str(e)) if __name__ in ["__main__", "builtin", "builtins"]: main()