parsedmarc/__init__.py
def parsed_aggregate_reports_to_csv(reports):
    """
    Converts one or more parsed aggregate reports to flat CSV format, including
    headers

    Args:
        reports: A parsed aggregate report or list of parsed aggregate reports

    Returns:
        str: Parsed aggregate report data in flat CSV format, including headers
    """

    def to_str(obj):
        return str(obj).lower()

    fields = ["xml_schema", "org_name", "org_email",
              "org_extra_contact_info", "report_id", "begin_date", "end_date",
              "errors", "domain", "adkim", "aspf", "p", "sp", "pct", "fo",
              "source_ip_address", "source_country", "source_reverse_dns",
              "source_base_domain", "count", "disposition", "dkim_alignment",
              "spf_alignment", "policy_override_reasons",
              "policy_override_comments", "envelope_from", "header_from",
              "envelope_to", "dkim_domains", "dkim_selectors", "dkim_results",
              "spf_domains", "spf_scopes", "spf_results"]

    csv_file_object = StringIO(newline="\n")
    writer = DictWriter(csv_file_object, fields)
    writer.writeheader()

    if type(reports) == OrderedDict:
        reports = [reports]

    for report in reports:
        xml_schema = report["xml_schema"]
        org_name = report["report_metadata"]["org_name"]
        org_email = report["report_metadata"]["org_email"]
        org_extra_contact = report["report_metadata"]["org_extra_contact_info"]
        report_id = report["report_metadata"]["report_id"]
        begin_date = report["report_metadata"]["begin_date"]
        end_date = report["report_metadata"]["end_date"]
        errors = "|".join(report["report_metadata"]["errors"])
        domain = report["policy_published"]["domain"]
        adkim = report["policy_published"]["adkim"]
        aspf = report["policy_published"]["aspf"]
        p = report["policy_published"]["p"]
        sp = report["policy_published"]["sp"]
        pct = report["policy_published"]["pct"]
        fo = report["policy_published"]["fo"]

        report_dict = dict(xml_schema=xml_schema, org_name=org_name,
                           org_email=org_email,
                           org_extra_contact_info=org_extra_contact,
                           report_id=report_id, begin_date=begin_date,
                           end_date=end_date, errors=errors, domain=domain,
                           adkim=adkim, aspf=aspf, p=p, sp=sp, pct=pct, fo=fo)

        for record in report["records"]:
            row = report_dict
            row["source_ip_address"] = record["source"]["ip_address"]
            row["source_country"] = record["source"]["country"]
            row["source_reverse_dns"] = record["source"]["reverse_dns"]
            row["source_base_domain"] = record["source"]["base_domain"]
            row["count"] = record["count"]
            row["disposition"] = record["policy_evaluated"]["disposition"]
            row["spf_alignment"] = record["policy_evaluated"]["spf"]
            row["dkim_alignment"] = record["policy_evaluated"]["dkim"]
            policy_override_reasons = list(map(
                lambda r: r["type"],
                record["policy_evaluated"]
                ["policy_override_reasons"]))
            policy_override_comments = list(map(
                lambda r: r["comment"] or "none",
                record["policy_evaluated"]
                ["policy_override_reasons"]))
            row["policy_override_reasons"] = ",".join(
                policy_override_reasons)
            row["policy_override_comments"] = "|".join(
                policy_override_comments)
            row["envelope_from"] = record["identifiers"]["envelope_from"]
            row["header_from"] = record["identifiers"]["header_from"]
            envelope_to = record["identifiers"]["envelope_to"]
            row["envelope_to"] = envelope_to
            dkim_domains = []
            dkim_selectors = []
            dkim_results = []
            for dkim_result in record["auth_results"]["dkim"]:
                dkim_domains.append(dkim_result["domain"])
                if "selector" in dkim_result:
                    dkim_selectors.append(dkim_result["selector"])
                dkim_results.append(dkim_result["result"])
            row["dkim_domains"] = ",".join(map(to_str, dkim_domains))
            row["dkim_selectors"] = ",".join(map(to_str, dkim_selectors))
            row["dkim_results"] = ",".join(map(to_str, dkim_results))
            spf_domains = []
            spf_scopes = []
            spf_results = []
            for spf_result in record["auth_results"]["spf"]:
                spf_domains.append(spf_result["domain"])
                spf_scopes.append(spf_result["scope"])
                spf_results.append(spf_result["result"])
            row["spf_domains"] = ",".join(map(to_str, spf_domains))
            row["spf_scopes"] = ",".join(map(to_str, spf_scopes))
            row["spf_results"] = ",".join(map(to_str, dkim_results))

            writer.writerow(row)
            csv_file_object.flush()

    return csv_file_object.getvalue()
def parsed_aggregate_reports_to_csv(reports):
    """
    Converts one or more parsed aggregate reports to flat CSV format, including
    headers

    Args:
        reports: A parsed aggregate report or list of parsed aggregate reports

    Returns:
        str: Parsed aggregate report data in flat CSV format, including headers
    """

    def to_str(obj):
        return str(obj).lower()

    fields = ["xml_schema", "org_name", "org_email",
              "org_extra_contact_info", "report_id", "begin_date", "end_date",
              "errors", "domain", "adkim", "aspf", "p", "sp", "pct", "fo",
              "source_ip_address", "source_country", "source_reverse_dns",
              "source_base_domain", "count", "disposition", "dkim_alignment",
              "spf_alignment", "policy_override_reasons",
              "policy_override_comments", "envelope_from", "header_from",
              "envelope_to", "dkim_domains", "dkim_selectors", "dkim_results",
              "spf_domains", "spf_scopes", "spf_results"]

    csv_file_object = StringIO(newline="\n")
    writer = DictWriter(csv_file_object, fields)
    writer.writeheader()

    if type(reports) == OrderedDict:
        reports = [reports]

    for report in reports:
        xml_schema = report["xml_schema"]
        org_name = report["report_metadata"]["org_name"]
        org_email = report["report_metadata"]["org_email"]
        org_extra_contact = report["report_metadata"]["org_extra_contact_info"]
        report_id = report["report_metadata"]["report_id"]
        begin_date = report["report_metadata"]["begin_date"]
        end_date = report["report_metadata"]["end_date"]
        errors = "|".join(report["report_metadata"]["errors"])
        domain = report["policy_published"]["domain"]
        adkim = report["policy_published"]["adkim"]
        aspf = report["policy_published"]["aspf"]
        p = report["policy_published"]["p"]
        sp = report["policy_published"]["sp"]
        pct = report["policy_published"]["pct"]
        fo = report["policy_published"]["fo"]

        report_dict = dict(xml_schema=xml_schema, org_name=org_name,
                           org_email=org_email,
                           org_extra_contact_info=org_extra_contact,
                           report_id=report_id, begin_date=begin_date,
                           end_date=end_date, errors=errors, domain=domain,
                           adkim=adkim, aspf=aspf, p=p, sp=sp, pct=pct, fo=fo)

        for record in report["records"]:
            row = report_dict
            row["source_ip_address"] = record["source"]["ip_address"]
            row["source_country"] = record["source"]["country"]
            row["source_reverse_dns"] = record["source"]["reverse_dns"]
            row["source_base_domain"] = record["source"]["base_domain"]
            row["count"] = record["count"]
            row["disposition"] = record["policy_evaluated"]["disposition"]
            row["spf_alignment"] = record["policy_evaluated"]["spf"]
            row["dkim_alignment"] = record["policy_evaluated"]["dkim"]
            policy_override_reasons = list(map(
                lambda r: r["type"],
                record["policy_evaluated"]
                ["policy_override_reasons"]))
            policy_override_comments = list(map(
                lambda r: r["comment"] or "none",
                record["policy_evaluated"]
                ["policy_override_reasons"]))
            row["policy_override_reasons"] = ",".join(
                policy_override_reasons)
            row["policy_override_comments"] = "|".join(
                policy_override_comments)
            row["envelope_from"] = record["identifiers"]["envelope_from"]
            row["header_from"] = record["identifiers"]["header_from"]
            envelope_to = record["identifiers"]["envelope_to"]
            row["envelope_to"] = envelope_to
            dkim_domains = []
            dkim_selectors = []
            dkim_results = []
            for dkim_result in record["auth_results"]["dkim"]:
                dkim_domains.append(dkim_result["domain"])
                if "selector" in dkim_result:
                    dkim_selectors.append(dkim_result["selector"])
                dkim_results.append(dkim_result["result"])
            row["dkim_domains"] = ",".join(map(to_str, dkim_domains))
            row["dkim_selectors"] = ",".join(map(to_str, dkim_selectors))
            row["dkim_results"] = ",".join(map(to_str, dkim_results))
            spf_domains = []
            spf_scopes = []
            spf_results = []
            for spf_result in record["auth_results"]["spf"]:
                spf_domains.append(spf_result["domain"])
                spf_scopes.append(spf_result["scope"])
                spf_results.append(spf_result["result"])
            row["spf_domains"] = ",".join(map(to_str, spf_domains))
            row["spf_scopes"] = ",".join(map(to_str, spf_scopes))
            row["spf_results"] = ",".join(map(to_str, dkim_results))

            writer.writerow(row)
            csv_file_object.flush()

    return csv_file_object.getvalue()
def parsed_aggregate_reports_to_csv ( reports ) : def to_str ( obj ) : return str ( obj ) . lower ( ) fields = [ "xml_schema" , "org_name" , "org_email" , "org_extra_contact_info" , "report_id" , "begin_date" , "end_date" , "errors" , "domain" , "adkim" , "aspf" , "p" , "sp" , "pct" , "fo" , "source_ip_address" , "source_country" , "source_reverse_dns" , "source_base_domain" , "count" , "disposition" , "dkim_alignment" , "spf_alignment" , "policy_override_reasons" , "policy_override_comments" , "envelope_from" , "header_from" , "envelope_to" , "dkim_domains" , "dkim_selectors" , "dkim_results" , "spf_domains" , "spf_scopes" , "spf_results" ] csv_file_object = StringIO ( newline = "\n" ) writer = DictWriter ( csv_file_object , fields ) writer . writeheader ( ) if type ( reports ) == OrderedDict : reports = [ reports ] for report in reports : xml_schema = report [ "xml_schema" ] org_name = report [ "report_metadata" ] [ "org_name" ] org_email = report [ "report_metadata" ] [ "org_email" ] org_extra_contact = report [ "report_metadata" ] [ "org_extra_contact_info" ] report_id = report [ "report_metadata" ] [ "report_id" ] begin_date = report [ "report_metadata" ] [ "begin_date" ] end_date = report [ "report_metadata" ] [ "end_date" ] errors = "|" . join ( report [ "report_metadata" ] [ "errors" ] ) domain = report [ "policy_published" ] [ "domain" ] adkim = report [ "policy_published" ] [ "adkim" ] aspf = report [ "policy_published" ] [ "aspf" ] p = report [ "policy_published" ] [ "p" ] sp = report [ "policy_published" ] [ "sp" ] pct = report [ "policy_published" ] [ "pct" ] fo = report [ "policy_published" ] [ "fo" ] report_dict = dict ( xml_schema = xml_schema , org_name = org_name , org_email = org_email , org_extra_contact_info = org_extra_contact , report_id = report_id , begin_date = begin_date , end_date = end_date , errors = errors , domain = domain , adkim = adkim , aspf = aspf , p = p , sp = sp , pct = pct , fo = fo ) for record in report [ "records" ] : row = report_dict row [ "source_ip_address" ] = record [ "source" ] [ "ip_address" ] row [ "source_country" ] = record [ "source" ] [ "country" ] row [ "source_reverse_dns" ] = record [ "source" ] [ "reverse_dns" ] row [ "source_base_domain" ] = record [ "source" ] [ "base_domain" ] row [ "count" ] = record [ "count" ] row [ "disposition" ] = record [ "policy_evaluated" ] [ "disposition" ] row [ "spf_alignment" ] = record [ "policy_evaluated" ] [ "spf" ] row [ "dkim_alignment" ] = record [ "policy_evaluated" ] [ "dkim" ] policy_override_reasons = list ( map ( lambda r : r [ "type" ] , record [ "policy_evaluated" ] [ "policy_override_reasons" ] ) ) policy_override_comments = list ( map ( lambda r : r [ "comment" ] or "none" , record [ "policy_evaluated" ] [ "policy_override_reasons" ] ) ) row [ "policy_override_reasons" ] = "," . join ( policy_override_reasons ) row [ "policy_override_comments" ] = "|" . join ( policy_override_comments ) row [ "envelope_from" ] = record [ "identifiers" ] [ "envelope_from" ] row [ "header_from" ] = record [ "identifiers" ] [ "header_from" ] envelope_to = record [ "identifiers" ] [ "envelope_to" ] row [ "envelope_to" ] = envelope_to dkim_domains = [ ] dkim_selectors = [ ] dkim_results = [ ] for dkim_result in record [ "auth_results" ] [ "dkim" ] : dkim_domains . append ( dkim_result [ "domain" ] ) if "selector" in dkim_result : dkim_selectors . append ( dkim_result [ "selector" ] ) dkim_results . append ( dkim_result [ "result" ] ) row [ "dkim_domains" ] = "," . join ( map ( to_str , dkim_domains ) ) row [ "dkim_selectors" ] = "," . join ( map ( to_str , dkim_selectors ) ) row [ "dkim_results" ] = "," . join ( map ( to_str , dkim_results ) ) spf_domains = [ ] spf_scopes = [ ] spf_results = [ ] for spf_result in record [ "auth_results" ] [ "spf" ] : spf_domains . append ( spf_result [ "domain" ] ) spf_scopes . append ( spf_result [ "scope" ] ) spf_results . append ( spf_result [ "result" ] ) row [ "spf_domains" ] = "," . join ( map ( to_str , spf_domains ) ) row [ "spf_scopes" ] = "," . join ( map ( to_str , spf_scopes ) ) row [ "spf_results" ] = "," . join ( map ( to_str , dkim_results ) ) writer . writerow ( row ) csv_file_object . flush ( ) return csv_file_object . getvalue ( )
Converts one or more parsed aggregate reports to flat CSV format, including
    headers

    Args:
        reports: A parsed aggregate report or list of parsed aggregate reports

    Returns:
        str: Parsed aggregate report data in flat CSV format, including headers
Converts one or more parsed aggregate reports to flat CSV format including headers