import html import mimetypes from pathlib import Path import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 from parse_emails.parse_emails import EmailParser logger = logging.getLogger("parse-email") # type: ignore[assignment] logger.addHandler(DemistoHandler) # type: ignore[attr-defined] def html_unescape(html_body: str) -> str: """ Unescape HTML entities in the raw HTML string returned by the email parser. The ``parse_emails`` library returns the HTML body verbatim from the email source, which means HTML entities such as ``&`` (used inside attribute values like ``href``) are **not** decoded. When indicator-extraction runs against this raw HTML it sees ``https://example.com?a=1&b=2`` instead of the real URL ``https://example.com?a=1&b=2``, causing missed or broken indicators. This function applies a single ``html.unescape()`` pass so that all named and numeric character references are replaced with their Unicode equivalents before the HTML is stored in context and consumed by downstream automations. Args: html_body (str): Raw HTML content as returned by the email parser. Returns: str: HTML string with all character references decoded. """ return html.unescape(html_body) def remove_bom(file_path: str, file_type: str, file_name: str) -> tuple[str, Optional[str], str]: """ Removes the Byte Order Mark (BOM) from a file, saves the cleaned content, and returns the path to the cleaned file, its MIME type, and its file name. If no BOM, keep the previous behaviour. """ path = Path(file_path) content = path.read_bytes() if content.startswith(b"\xef\xbb\xbf"): content = content[3:] # Write the cleaned content to a new file or overwrite the original file cleaned_file_path = path.with_name("cleaned_" + path.name) cleaned_file_path.write_bytes(content) # Get the MIME type mime_type, _ = mimetypes.guess_type(cleaned_file_path) # Get the file name file_name = cleaned_file_path.name return str(cleaned_file_path), mime_type, file_name else: # keep the exists behaviour (without BOM) demisto.info(f"BOM not detected in file: {file_name}, {file_type=}") return file_path, file_type, file_name def data_to_md(email_data, email_file_name=None, parent_email_file=None, print_only_headers=False) -> str: """ create Markdown with the data. Args: email_data (dict): all the email data. email_file_name (str): the email file name. parent_email_file (str): the parent email file name (for attachment mail). print_only_headers (bool): Whether to only the headers. Returns: str: the parsed Markdown """ if email_data is None: return "No data extracted from email" md = "### Results:\n" if email_file_name: md = f"### {email_file_name}\n" if print_only_headers: return tableToMarkdown(f"Email Headers: {email_file_name}", email_data.get("HeadersMap")) if parent_email_file: md += f"### Containing email: {parent_email_file}\n" md += f"""* From:\t{email_data.get('From') or ""}\n""" md += f"""* To:\t{email_data.get('To') or ""}\n""" md += f"""* CC:\t{email_data.get('CC') or ""}\n""" md += f"""* BCC:\t{email_data.get('BCC') or ""}\n""" md += f"""* Subject:\t{email_data.get('Subject') or ""}\n""" if email_data.get("Text"): text = email_data["Text"].replace("<", "[").replace(">", "]") md += f'* Body/Text:\t{text or ""}\n' if email_data.get("HTML"): md += f"""* Body/HTML:\t{email_data['HTML'] or ""}\n""" md += f"""* Attachments:\t{email_data.get('Attachments') or ""}\n""" md += "\n\n" + tableToMarkdown("HeadersMap", email_data.get("HeadersMap")) return md def save_file(file_name, file_content) -> str: """ save attachment to the war room and return the file internal path. Args: file_name (str): The name of the file to be created. file_content (str/bytes): the file data. Returns: str: the file internal path """ created_file = fileResult(file_name, file_content) file_id = created_file.get("FileID") attachment_internal_path = demisto.investigation().get("id") + "_" + file_id return_results(created_file) return attachment_internal_path def extract_file_info(entry_id: str) -> tuple: """ extract from the entry id the file_type, file_path and file_name. Args: entry_id (str): The entry id. Returns: file_type(str): the file mime type. file_path(str): the file path. file_name(str):the file name. """ file_type = "" file_path = "" file_name = "" try: result = demisto.executeCommand("getFilePath", {"id": entry_id}) if is_error(result): return_error(get_error(result)) file_path = result[0]["Contents"]["path"] file_name = result[0]["Contents"]["name"] dt_file_type = demisto.dt(demisto.context(), f"File(val.EntryID=='{entry_id}').Type") file_type = dt_file_type[0] if isinstance(dt_file_type, list) else dt_file_type dt_file_info = demisto.dt(demisto.context(), f"File(val.EntryID=='{entry_id}').Info") file_info = dt_file_info[0] if isinstance(dt_file_info, list) else dt_file_info demisto.debug(f"Context values: {dt_file_type=}, {file_type=}, {dt_file_info=}, {file_info=}, {file_name=}") if file_type in ("eml", "txt") and file_info and ("rfc" in file_info.lower() or "ascii" in file_info.lower()): demisto.debug(f"{file_type=} seems wrong, changing it to {file_info=}") file_type = file_info if ( file_name and file_name.lower().endswith(".eml") and file_type and ("iso-8859" in file_type.lower() or "mime entity" in file_type.lower()) ): demisto.debug(f"Detected EML file misclassified as text ({file_type}). Forcing RFC822 parsing.") file_type = "RFC 822 mail text" except Exception as ex: return_error( "Failed to load file entry with entry id: {}. Error: {}".format( entry_id, str(ex) + "\n\nTrace:\n" + traceback.format_exc() ) ) demisto.debug(f"extract_file_info returning {file_type=}, {file_path=}, {file_name=}") return file_type, file_path, file_name def parse_nesting_level(nesting_level_to_return, output): if nesting_level_to_return == "Outer file": # return only the outer email info return [output[0]] elif nesting_level_to_return == "Inner file": # the last file in list it is the inner attached file return [output[-1]] return output def main(): args = demisto.args() entry_id = args.get("entryid") max_depth = arg_to_number(args.get("max_depth", "3")) if not max_depth or max_depth < 1: return_error("Minimum max_depth is 1, the script will parse just the top email") parse_only_headers = argToBoolean(args.get("parse_only_headers", "false")) forced_encoding = args.get("forced_encoding") default_encoding = args.get("default_encoding") nesting_level_to_return = args.get("nesting_level_to_return", "All files") file_type, file_path, file_name = extract_file_info(entry_id) demisto.debug(f"{file_type=}, {file_path=}, {file_name=}") # Remove BOM and parse the email cleaned_file_path, file_type, file_name = remove_bom(file_path, file_type, file_name) try: email_parser = EmailParser( file_path=cleaned_file_path, max_depth=max_depth, parse_only_headers=parse_only_headers, file_info=file_type, forced_encoding=forced_encoding, default_encoding=default_encoding, file_name=file_name, ) output = email_parser.parse() demisto.debug(f"{output=}") results = [] if isinstance(output, dict): output = [output] elif output and nesting_level_to_return != "All files": output = parse_nesting_level(nesting_level_to_return, output) for email in output: if email.get("AttachmentsData"): for attachment in email.get("AttachmentsData"): if name := attachment.get("Name"): if content := attachment.get("FileData"): attachment["FilePath"] = save_file(name, content) del attachment["FileData"] else: attachment["FileData"] = None # probably a wrapper and we can ignore the outer "email" if email.get("Format") == "multipart/signed" and all(not email.get(field) for field in ["To", "From", "Subject"]): continue if isinstance(email.get("HTML"), bytes): email["HTML"] = email.get("HTML").decode("utf-8") if html_body := email.get("HTML"): email["HTMLUnescape"] = html_unescape(html_body) results.append( CommandResults( outputs_prefix="Email", outputs=email, readable_output=data_to_md( email, file_name, email.get("ParentFileName", None), print_only_headers=parse_only_headers ), raw_response=email, ) ) return_results(results) except Exception as e: return_error(str(e) + "\n\nTrace:\n" + traceback.format_exc()) if __name__ in ("__main__", "__builtin__", "builtins"): main()