ParseEmailFilesV2

Source

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()

README