parsedmarc/__init__.py
def parse_forensic_report(feedback_report, sample, msg_date,
                          nameservers=None, dns_timeout=2.0,
                          strip_attachment_payloads=False,
                          parallel=False):
    """
    Converts a DMARC forensic report and sample to a ``OrderedDict``

    Args:
        feedback_report (str): A message's feedback report as a string
        sample (str): The RFC 822 headers or RFC 822 message sample
        msg_date (str): The message's date header
        nameservers (list): A list of one or more nameservers to use
        (Cloudflare's public DNS resolvers by default)
        dns_timeout (float): Sets the DNS timeout in seconds
        strip_attachment_payloads (bool): Remove attachment payloads from
        forensic report results
        parallel (bool): Parallel processing

    Returns:
        OrderedDict: A parsed report and sample
    """
    delivery_results = ["delivered", "spam", "policy", "reject", "other"]

    try:
        parsed_report = OrderedDict()
        report_values = feedback_report_regex.findall(feedback_report)
        for report_value in report_values:
            key = report_value[0].lower().replace("-", "_")
            parsed_report[key] = report_value[1]

        if "arrival_date" not in parsed_report:
            if msg_date is None:
                raise InvalidForensicReport(
                    "Forensic sample is not a valid email")
            parsed_report["arrival_date"] = msg_date.isoformat()

        if "version" not in parsed_report:
            parsed_report["version"] = 1

        if "user_agent" not in parsed_report:
            parsed_report["user_agent"] = None

        if "delivery_result" not in parsed_report:
            parsed_report["delivery_result"] = None
        else:
            for delivery_result in delivery_results:
                if delivery_result in parsed_report["delivery_result"].lower():
                    parsed_report["delivery_result"] = delivery_result
                    break
        if parsed_report["delivery_result"] not in delivery_results:
            parsed_report["delivery_result"] = "other"

        arrival_utc = human_timestamp_to_datetime(
            parsed_report["arrival_date"], to_utc=True)
        arrival_utc = arrival_utc.strftime("%Y-%m-%d %H:%M:%S")
        parsed_report["arrival_date_utc"] = arrival_utc

        ip_address = parsed_report["source_ip"]
        parsed_report_source = get_ip_address_info(ip_address,
                                                   nameservers=nameservers,
                                                   timeout=dns_timeout,
                                                   parallel=parallel)
        parsed_report["source"] = parsed_report_source
        del parsed_report["source_ip"]

        if "identity_alignment" not in parsed_report:
            parsed_report["authentication_mechanisms"] = []
        elif parsed_report["identity_alignment"] == "none":
            parsed_report["authentication_mechanisms"] = []
            del parsed_report["identity_alignment"]
        else:
            auth_mechanisms = parsed_report["identity_alignment"]
            auth_mechanisms = auth_mechanisms.split(",")
            parsed_report["authentication_mechanisms"] = auth_mechanisms
            del parsed_report["identity_alignment"]

        if "auth_failure" not in parsed_report:
            parsed_report["auth_failure"] = "dmarc"
        auth_failure = parsed_report["auth_failure"].split(",")
        parsed_report["auth_failure"] = auth_failure

        optional_fields = ["original_envelope_id", "dkim_domain",
                           "original_mail_from", "original_rcpt_to"]
        for optional_field in optional_fields:
            if optional_field not in parsed_report:
                parsed_report[optional_field] = None

        parsed_sample = parse_email(
            sample,
            strip_attachment_payloads=strip_attachment_payloads)

        if "reported_domain" not in parsed_report:
            parsed_report["reported_domain"] = parsed_sample["from"]["domain"]

        sample_headers_only = False
        number_of_attachments = len(parsed_sample["attachments"])
        if number_of_attachments < 1 and parsed_sample["body"] is None:
            sample_headers_only = True
        if sample_headers_only and parsed_sample["has_defects"]:
            del parsed_sample["defects"]
            del parsed_sample["defects_categories"]
            del parsed_sample["has_defects"]
        parsed_report["sample_headers_only"] = sample_headers_only
        parsed_report["sample"] = sample
        parsed_report["parsed_sample"] = parsed_sample

        return parsed_report

    except KeyError as error:
        raise InvalidForensicReport("Missing value: {0}".format(
            error.__str__()))

    except Exception as error:
        raise InvalidForensicReport(
            "Unexpected error: {0}".format(error.__str__()))
def parse_forensic_report(feedback_report, sample, msg_date,
                          nameservers=None, dns_timeout=2.0,
                          strip_attachment_payloads=False,
                          parallel=False):
    """
    Converts a DMARC forensic report and sample to a ``OrderedDict``

    Args:
        feedback_report (str): A message's feedback report as a string
        sample (str): The RFC 822 headers or RFC 822 message sample
        msg_date (str): The message's date header
        nameservers (list): A list of one or more nameservers to use
        (Cloudflare's public DNS resolvers by default)
        dns_timeout (float): Sets the DNS timeout in seconds
        strip_attachment_payloads (bool): Remove attachment payloads from
        forensic report results
        parallel (bool): Parallel processing

    Returns:
        OrderedDict: A parsed report and sample
    """
    delivery_results = ["delivered", "spam", "policy", "reject", "other"]

    try:
        parsed_report = OrderedDict()
        report_values = feedback_report_regex.findall(feedback_report)
        for report_value in report_values:
            key = report_value[0].lower().replace("-", "_")
            parsed_report[key] = report_value[1]

        if "arrival_date" not in parsed_report:
            if msg_date is None:
                raise InvalidForensicReport(
                    "Forensic sample is not a valid email")
            parsed_report["arrival_date"] = msg_date.isoformat()

        if "version" not in parsed_report:
            parsed_report["version"] = 1

        if "user_agent" not in parsed_report:
            parsed_report["user_agent"] = None

        if "delivery_result" not in parsed_report:
            parsed_report["delivery_result"] = None
        else:
            for delivery_result in delivery_results:
                if delivery_result in parsed_report["delivery_result"].lower():
                    parsed_report["delivery_result"] = delivery_result
                    break
        if parsed_report["delivery_result"] not in delivery_results:
            parsed_report["delivery_result"] = "other"

        arrival_utc = human_timestamp_to_datetime(
            parsed_report["arrival_date"], to_utc=True)
        arrival_utc = arrival_utc.strftime("%Y-%m-%d %H:%M:%S")
        parsed_report["arrival_date_utc"] = arrival_utc

        ip_address = parsed_report["source_ip"]
        parsed_report_source = get_ip_address_info(ip_address,
                                                   nameservers=nameservers,
                                                   timeout=dns_timeout,
                                                   parallel=parallel)
        parsed_report["source"] = parsed_report_source
        del parsed_report["source_ip"]

        if "identity_alignment" not in parsed_report:
            parsed_report["authentication_mechanisms"] = []
        elif parsed_report["identity_alignment"] == "none":
            parsed_report["authentication_mechanisms"] = []
            del parsed_report["identity_alignment"]
        else:
            auth_mechanisms = parsed_report["identity_alignment"]
            auth_mechanisms = auth_mechanisms.split(",")
            parsed_report["authentication_mechanisms"] = auth_mechanisms
            del parsed_report["identity_alignment"]

        if "auth_failure" not in parsed_report:
            parsed_report["auth_failure"] = "dmarc"
        auth_failure = parsed_report["auth_failure"].split(",")
        parsed_report["auth_failure"] = auth_failure

        optional_fields = ["original_envelope_id", "dkim_domain",
                           "original_mail_from", "original_rcpt_to"]
        for optional_field in optional_fields:
            if optional_field not in parsed_report:
                parsed_report[optional_field] = None

        parsed_sample = parse_email(
            sample,
            strip_attachment_payloads=strip_attachment_payloads)

        if "reported_domain" not in parsed_report:
            parsed_report["reported_domain"] = parsed_sample["from"]["domain"]

        sample_headers_only = False
        number_of_attachments = len(parsed_sample["attachments"])
        if number_of_attachments < 1 and parsed_sample["body"] is None:
            sample_headers_only = True
        if sample_headers_only and parsed_sample["has_defects"]:
            del parsed_sample["defects"]
            del parsed_sample["defects_categories"]
            del parsed_sample["has_defects"]
        parsed_report["sample_headers_only"] = sample_headers_only
        parsed_report["sample"] = sample
        parsed_report["parsed_sample"] = parsed_sample

        return parsed_report

    except KeyError as error:
        raise InvalidForensicReport("Missing value: {0}".format(
            error.__str__()))

    except Exception as error:
        raise InvalidForensicReport(
            "Unexpected error: {0}".format(error.__str__()))
def parse_forensic_report ( feedback_report , sample , msg_date , nameservers = None , dns_timeout = 2.0 , strip_attachment_payloads = False , parallel = False ) : delivery_results = [ "delivered" , "spam" , "policy" , "reject" , "other" ] try : parsed_report = OrderedDict ( ) report_values = feedback_report_regex . findall ( feedback_report ) for report_value in report_values : key = report_value [ 0 ] . lower ( ) . replace ( "-" , "_" ) parsed_report [ key ] = report_value [ 1 ] if "arrival_date" not in parsed_report : if msg_date is None : raise InvalidForensicReport ( "Forensic sample is not a valid email" ) parsed_report [ "arrival_date" ] = msg_date . isoformat ( ) if "version" not in parsed_report : parsed_report [ "version" ] = 1 if "user_agent" not in parsed_report : parsed_report [ "user_agent" ] = None if "delivery_result" not in parsed_report : parsed_report [ "delivery_result" ] = None else : for delivery_result in delivery_results : if delivery_result in parsed_report [ "delivery_result" ] . lower ( ) : parsed_report [ "delivery_result" ] = delivery_result break if parsed_report [ "delivery_result" ] not in delivery_results : parsed_report [ "delivery_result" ] = "other" arrival_utc = human_timestamp_to_datetime ( parsed_report [ "arrival_date" ] , to_utc = True ) arrival_utc = arrival_utc . strftime ( "%Y-%m-%d %H:%M:%S" ) parsed_report [ "arrival_date_utc" ] = arrival_utc ip_address = parsed_report [ "source_ip" ] parsed_report_source = get_ip_address_info ( ip_address , nameservers = nameservers , timeout = dns_timeout , parallel = parallel ) parsed_report [ "source" ] = parsed_report_source del parsed_report [ "source_ip" ] if "identity_alignment" not in parsed_report : parsed_report [ "authentication_mechanisms" ] = [ ] elif parsed_report [ "identity_alignment" ] == "none" : parsed_report [ "authentication_mechanisms" ] = [ ] del parsed_report [ "identity_alignment" ] else : auth_mechanisms = parsed_report [ "identity_alignment" ] auth_mechanisms = auth_mechanisms . split ( "," ) parsed_report [ "authentication_mechanisms" ] = auth_mechanisms del parsed_report [ "identity_alignment" ] if "auth_failure" not in parsed_report : parsed_report [ "auth_failure" ] = "dmarc" auth_failure = parsed_report [ "auth_failure" ] . split ( "," ) parsed_report [ "auth_failure" ] = auth_failure optional_fields = [ "original_envelope_id" , "dkim_domain" , "original_mail_from" , "original_rcpt_to" ] for optional_field in optional_fields : if optional_field not in parsed_report : parsed_report [ optional_field ] = None parsed_sample = parse_email ( sample , strip_attachment_payloads = strip_attachment_payloads ) if "reported_domain" not in parsed_report : parsed_report [ "reported_domain" ] = parsed_sample [ "from" ] [ "domain" ] sample_headers_only = False number_of_attachments = len ( parsed_sample [ "attachments" ] ) if number_of_attachments < 1 and parsed_sample [ "body" ] is None : sample_headers_only = True if sample_headers_only and parsed_sample [ "has_defects" ] : del parsed_sample [ "defects" ] del parsed_sample [ "defects_categories" ] del parsed_sample [ "has_defects" ] parsed_report [ "sample_headers_only" ] = sample_headers_only parsed_report [ "sample" ] = sample parsed_report [ "parsed_sample" ] = parsed_sample return parsed_report except KeyError as error : raise InvalidForensicReport ( "Missing value: {0}" . format ( error . __str__ ( ) ) ) except Exception as error : raise InvalidForensicReport ( "Unexpected error: {0}" . format ( error . __str__ ( ) ) )
Converts a DMARC forensic report and sample to a ``OrderedDict``

    Args:
        feedback_report (str): A message's feedback report as a string
        sample (str): The RFC 822 headers or RFC 822 message sample
        msg_date (str): The message's date header
        nameservers (list): A list of one or more nameservers to use
        (Cloudflare's public DNS resolvers by default)
        dns_timeout (float): Sets the DNS timeout in seconds
        strip_attachment_payloads (bool): Remove attachment payloads from
        forensic report results
        parallel (bool): Parallel processing

    Returns:
        OrderedDict: A parsed report and sample
Converts a DMARC forensic report and sample to a OrderedDict