parsedmarc/__init__.py
def _parse_report_record(record, nameservers=None, dns_timeout=2.0,
                         parallel=False):
    """
    Converts a record from a DMARC aggregate report into a more consistent
    format

    Args:
        record (OrderedDict): The record to convert
        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

    Returns:
        OrderedDict: The converted record
    """
    if nameservers is None:
        nameservers = ["1.1.1.1", "1.0.0.1",
                       "2606:4700:4700::1111", "2606:4700:4700::1001",
                       ]
    record = record.copy()
    new_record = OrderedDict()
    new_record_source = get_ip_address_info(record["row"]["source_ip"],
                                            cache=IP_ADDRESS_CACHE,
                                            nameservers=nameservers,
                                            timeout=dns_timeout,
                                            parallel=parallel)
    new_record["source"] = new_record_source
    new_record["count"] = int(record["row"]["count"])
    policy_evaluated = record["row"]["policy_evaluated"].copy()
    new_policy_evaluated = OrderedDict([("disposition", "none"),
                                        ("dkim", "fail"),
                                        ("spf", "fail"),
                                        ("policy_override_reasons", [])
                                        ])
    if "disposition" in policy_evaluated:
        new_policy_evaluated["disposition"] = policy_evaluated["disposition"]
        if new_policy_evaluated["disposition"].strip().lower() == "pass":
            new_policy_evaluated["disposition"] = "none"
    if "dkim" in policy_evaluated:
        new_policy_evaluated["dkim"] = policy_evaluated["dkim"]
    if "spf" in policy_evaluated:
        new_policy_evaluated["spf"] = policy_evaluated["spf"]
    reasons = []
    spf_aligned = policy_evaluated["spf"] == "pass"
    dkim_aligned = policy_evaluated["dkim"] == "pass"
    dmarc_aligned = spf_aligned or dkim_aligned
    new_record["alignment"] = dict()
    new_record["alignment"]["spf"] = spf_aligned
    new_record["alignment"]["dkim"] = dkim_aligned
    new_record["alignment"]["dmarc"] = dmarc_aligned
    if "reason" in policy_evaluated:
        if type(policy_evaluated["reason"]) == list:
            reasons = policy_evaluated["reason"]
        else:
            reasons = [policy_evaluated["reason"]]
    for reason in reasons:
        if "comment" not in reason:
            reason["comment"] = None
    new_policy_evaluated["policy_override_reasons"] = reasons
    new_record["policy_evaluated"] = new_policy_evaluated
    new_record["identifiers"] = record["identifiers"].copy()
    new_record["auth_results"] = OrderedDict([("dkim", []), ("spf", [])])
    if record["auth_results"] is not None:
        auth_results = record["auth_results"].copy()
        if "spf" not in auth_results:
            auth_results["spf"] = []
        if "dkim" not in auth_results:
            auth_results["dkim"] = []
    else:
        auth_results = new_record["auth_results"].copy()

    if type(auth_results["dkim"]) != list:
        auth_results["dkim"] = [auth_results["dkim"]]
    for result in auth_results["dkim"]:
        if "domain" in result and result["domain"] is not None:
            new_result = OrderedDict([("domain", result["domain"])])
            if "selector" in result and result["selector"] is not None:
                new_result["selector"] = result["selector"]
            else:
                new_result["selector"] = "none"
            if "result" in result and result["result"] is not None:
                new_result["result"] = result["result"]
            else:
                new_result["result"] = "none"
            new_record["auth_results"]["dkim"].append(new_result)

    if type(auth_results["spf"]) != list:
        auth_results["spf"] = [auth_results["spf"]]
    for result in auth_results["spf"]:
        new_result = OrderedDict([("domain", result["domain"])])
        if "scope" in result and result["scope"] is not None:
            new_result["scope"] = result["scope"]
        else:
            new_result["scope"] = "mfrom"
        if "result" in result and result["result"] is not None:
            new_result["result"] = result["result"]
        else:
            new_result["result"] = "none"
        new_record["auth_results"]["spf"].append(new_result)

    if "envelope_from" not in new_record["identifiers"]:
        envelope_from = None
        if len(auth_results["spf"]) > 0:
            envelope_from = new_record["auth_results"]["spf"][-1]["domain"]
        if envelope_from is not None:
            envelope_from = str(envelope_from).lower()
        new_record["identifiers"]["envelope_from"] = envelope_from

    elif new_record["identifiers"]["envelope_from"] is None:
        if len(auth_results["spf"]) > 0:
            envelope_from = new_record["auth_results"]["spf"][-1]["domain"]
            if envelope_from is not None:
                envelope_from = str(envelope_from).lower()
            new_record["identifiers"]["envelope_from"] = envelope_from

    envelope_to = None
    if "envelope_to" in new_record["identifiers"]:
        envelope_to = new_record["identifiers"]["envelope_to"]
        del new_record["identifiers"]["envelope_to"]

    new_record["identifiers"]["envelope_to"] = envelope_to

    return new_record
def _parse_report_record(record, nameservers=None, dns_timeout=2.0,
                         parallel=False):
    """
    Converts a record from a DMARC aggregate report into a more consistent
    format

    Args:
        record (OrderedDict): The record to convert
        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

    Returns:
        OrderedDict: The converted record
    """
    if nameservers is None:
        nameservers = ["1.1.1.1", "1.0.0.1",
                       "2606:4700:4700::1111", "2606:4700:4700::1001",
                       ]
    record = record.copy()
    new_record = OrderedDict()
    new_record_source = get_ip_address_info(record["row"]["source_ip"],
                                            cache=IP_ADDRESS_CACHE,
                                            nameservers=nameservers,
                                            timeout=dns_timeout,
                                            parallel=parallel)
    new_record["source"] = new_record_source
    new_record["count"] = int(record["row"]["count"])
    policy_evaluated = record["row"]["policy_evaluated"].copy()
    new_policy_evaluated = OrderedDict([("disposition", "none"),
                                        ("dkim", "fail"),
                                        ("spf", "fail"),
                                        ("policy_override_reasons", [])
                                        ])
    if "disposition" in policy_evaluated:
        new_policy_evaluated["disposition"] = policy_evaluated["disposition"]
        if new_policy_evaluated["disposition"].strip().lower() == "pass":
            new_policy_evaluated["disposition"] = "none"
    if "dkim" in policy_evaluated:
        new_policy_evaluated["dkim"] = policy_evaluated["dkim"]
    if "spf" in policy_evaluated:
        new_policy_evaluated["spf"] = policy_evaluated["spf"]
    reasons = []
    spf_aligned = policy_evaluated["spf"] == "pass"
    dkim_aligned = policy_evaluated["dkim"] == "pass"
    dmarc_aligned = spf_aligned or dkim_aligned
    new_record["alignment"] = dict()
    new_record["alignment"]["spf"] = spf_aligned
    new_record["alignment"]["dkim"] = dkim_aligned
    new_record["alignment"]["dmarc"] = dmarc_aligned
    if "reason" in policy_evaluated:
        if type(policy_evaluated["reason"]) == list:
            reasons = policy_evaluated["reason"]
        else:
            reasons = [policy_evaluated["reason"]]
    for reason in reasons:
        if "comment" not in reason:
            reason["comment"] = None
    new_policy_evaluated["policy_override_reasons"] = reasons
    new_record["policy_evaluated"] = new_policy_evaluated
    new_record["identifiers"] = record["identifiers"].copy()
    new_record["auth_results"] = OrderedDict([("dkim", []), ("spf", [])])
    if record["auth_results"] is not None:
        auth_results = record["auth_results"].copy()
        if "spf" not in auth_results:
            auth_results["spf"] = []
        if "dkim" not in auth_results:
            auth_results["dkim"] = []
    else:
        auth_results = new_record["auth_results"].copy()

    if type(auth_results["dkim"]) != list:
        auth_results["dkim"] = [auth_results["dkim"]]
    for result in auth_results["dkim"]:
        if "domain" in result and result["domain"] is not None:
            new_result = OrderedDict([("domain", result["domain"])])
            if "selector" in result and result["selector"] is not None:
                new_result["selector"] = result["selector"]
            else:
                new_result["selector"] = "none"
            if "result" in result and result["result"] is not None:
                new_result["result"] = result["result"]
            else:
                new_result["result"] = "none"
            new_record["auth_results"]["dkim"].append(new_result)

    if type(auth_results["spf"]) != list:
        auth_results["spf"] = [auth_results["spf"]]
    for result in auth_results["spf"]:
        new_result = OrderedDict([("domain", result["domain"])])
        if "scope" in result and result["scope"] is not None:
            new_result["scope"] = result["scope"]
        else:
            new_result["scope"] = "mfrom"
        if "result" in result and result["result"] is not None:
            new_result["result"] = result["result"]
        else:
            new_result["result"] = "none"
        new_record["auth_results"]["spf"].append(new_result)

    if "envelope_from" not in new_record["identifiers"]:
        envelope_from = None
        if len(auth_results["spf"]) > 0:
            envelope_from = new_record["auth_results"]["spf"][-1]["domain"]
        if envelope_from is not None:
            envelope_from = str(envelope_from).lower()
        new_record["identifiers"]["envelope_from"] = envelope_from

    elif new_record["identifiers"]["envelope_from"] is None:
        if len(auth_results["spf"]) > 0:
            envelope_from = new_record["auth_results"]["spf"][-1]["domain"]
            if envelope_from is not None:
                envelope_from = str(envelope_from).lower()
            new_record["identifiers"]["envelope_from"] = envelope_from

    envelope_to = None
    if "envelope_to" in new_record["identifiers"]:
        envelope_to = new_record["identifiers"]["envelope_to"]
        del new_record["identifiers"]["envelope_to"]

    new_record["identifiers"]["envelope_to"] = envelope_to

    return new_record
def _parse_report_record ( record , nameservers = None , dns_timeout = 2.0 , parallel = False ) : if nameservers is None : nameservers = [ "1.1.1.1" , "1.0.0.1" , "2606:4700:4700::1111" , "2606:4700:4700::1001" , ] record = record . copy ( ) new_record = OrderedDict ( ) new_record_source = get_ip_address_info ( record [ "row" ] [ "source_ip" ] , cache = IP_ADDRESS_CACHE , nameservers = nameservers , timeout = dns_timeout , parallel = parallel ) new_record [ "source" ] = new_record_source new_record [ "count" ] = int ( record [ "row" ] [ "count" ] ) policy_evaluated = record [ "row" ] [ "policy_evaluated" ] . copy ( ) new_policy_evaluated = OrderedDict ( [ ( "disposition" , "none" ) , ( "dkim" , "fail" ) , ( "spf" , "fail" ) , ( "policy_override_reasons" , [ ] ) ] ) if "disposition" in policy_evaluated : new_policy_evaluated [ "disposition" ] = policy_evaluated [ "disposition" ] if new_policy_evaluated [ "disposition" ] . strip ( ) . lower ( ) == "pass" : new_policy_evaluated [ "disposition" ] = "none" if "dkim" in policy_evaluated : new_policy_evaluated [ "dkim" ] = policy_evaluated [ "dkim" ] if "spf" in policy_evaluated : new_policy_evaluated [ "spf" ] = policy_evaluated [ "spf" ] reasons = [ ] spf_aligned = policy_evaluated [ "spf" ] == "pass" dkim_aligned = policy_evaluated [ "dkim" ] == "pass" dmarc_aligned = spf_aligned or dkim_aligned new_record [ "alignment" ] = dict ( ) new_record [ "alignment" ] [ "spf" ] = spf_aligned new_record [ "alignment" ] [ "dkim" ] = dkim_aligned new_record [ "alignment" ] [ "dmarc" ] = dmarc_aligned if "reason" in policy_evaluated : if type ( policy_evaluated [ "reason" ] ) == list : reasons = policy_evaluated [ "reason" ] else : reasons = [ policy_evaluated [ "reason" ] ] for reason in reasons : if "comment" not in reason : reason [ "comment" ] = None new_policy_evaluated [ "policy_override_reasons" ] = reasons new_record [ "policy_evaluated" ] = new_policy_evaluated new_record [ "identifiers" ] = record [ "identifiers" ] . copy ( ) new_record [ "auth_results" ] = OrderedDict ( [ ( "dkim" , [ ] ) , ( "spf" , [ ] ) ] ) if record [ "auth_results" ] is not None : auth_results = record [ "auth_results" ] . copy ( ) if "spf" not in auth_results : auth_results [ "spf" ] = [ ] if "dkim" not in auth_results : auth_results [ "dkim" ] = [ ] else : auth_results = new_record [ "auth_results" ] . copy ( ) if type ( auth_results [ "dkim" ] ) != list : auth_results [ "dkim" ] = [ auth_results [ "dkim" ] ] for result in auth_results [ "dkim" ] : if "domain" in result and result [ "domain" ] is not None : new_result = OrderedDict ( [ ( "domain" , result [ "domain" ] ) ] ) if "selector" in result and result [ "selector" ] is not None : new_result [ "selector" ] = result [ "selector" ] else : new_result [ "selector" ] = "none" if "result" in result and result [ "result" ] is not None : new_result [ "result" ] = result [ "result" ] else : new_result [ "result" ] = "none" new_record [ "auth_results" ] [ "dkim" ] . append ( new_result ) if type ( auth_results [ "spf" ] ) != list : auth_results [ "spf" ] = [ auth_results [ "spf" ] ] for result in auth_results [ "spf" ] : new_result = OrderedDict ( [ ( "domain" , result [ "domain" ] ) ] ) if "scope" in result and result [ "scope" ] is not None : new_result [ "scope" ] = result [ "scope" ] else : new_result [ "scope" ] = "mfrom" if "result" in result and result [ "result" ] is not None : new_result [ "result" ] = result [ "result" ] else : new_result [ "result" ] = "none" new_record [ "auth_results" ] [ "spf" ] . append ( new_result ) if "envelope_from" not in new_record [ "identifiers" ] : envelope_from = None if len ( auth_results [ "spf" ] ) > 0 : envelope_from = new_record [ "auth_results" ] [ "spf" ] [ - 1 ] [ "domain" ] if envelope_from is not None : envelope_from = str ( envelope_from ) . lower ( ) new_record [ "identifiers" ] [ "envelope_from" ] = envelope_from elif new_record [ "identifiers" ] [ "envelope_from" ] is None : if len ( auth_results [ "spf" ] ) > 0 : envelope_from = new_record [ "auth_results" ] [ "spf" ] [ - 1 ] [ "domain" ] if envelope_from is not None : envelope_from = str ( envelope_from ) . lower ( ) new_record [ "identifiers" ] [ "envelope_from" ] = envelope_from envelope_to = None if "envelope_to" in new_record [ "identifiers" ] : envelope_to = new_record [ "identifiers" ] [ "envelope_to" ] del new_record [ "identifiers" ] [ "envelope_to" ] new_record [ "identifiers" ] [ "envelope_to" ] = envelope_to return new_record
Converts a record from a DMARC aggregate report into a more consistent
    format

    Args:
        record (OrderedDict): The record to convert
        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

    Returns:
        OrderedDict: The converted record
Converts a record from a DMARC aggregate report into a more consistent format