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

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

    Returns:
        str: Parsed forensic report data in flat CSV format, including headers
        """
    fields = ["feedback_type", "user_agent", "version", "original_envelope_id",
              "original_mail_from", "original_rcpt_to", "arrival_date",
              "arrival_date_utc", "subject", "message_id",
              "authentication_results", "dkim_domain", "source_ip_address",
              "source_country", "source_reverse_dns", "source_base_domain",
              "delivery_result", "auth_failure", "reported_domain",
              "authentication_mechanisms", "sample_headers_only"]

    if type(reports) == OrderedDict:
        reports = [reports]
    csv_file = StringIO()
    csv_writer = DictWriter(csv_file, fieldnames=fields)
    csv_writer.writeheader()
    for report in reports:
        row = report.copy()
        row["source_ip_address"] = report["source"]["ip_address"]
        row["source_reverse_dns"] = report["source"]["reverse_dns"]
        row["source_base_domain"] = report["source"]["base_domain"]
        row["source_country"] = report["source"]["country"]
        del row["source"]
        row["subject"] = report["parsed_sample"]["subject"]
        row["auth_failure"] = ",".join(report["auth_failure"])
        authentication_mechanisms = report["authentication_mechanisms"]
        row["authentication_mechanisms"] = ",".join(
            authentication_mechanisms)
        del row["sample"]
        del row["parsed_sample"]
        csv_writer.writerow(row)

    return csv_file.getvalue()
def parsed_forensic_reports_to_csv(reports):
    fields = ["feedback_type", "user_agent", "version", "original_envelope_id",
              "original_mail_from", "original_rcpt_to", "arrival_date",
              "arrival_date_utc", "subject", "message_id",
              "authentication_results", "dkim_domain", "source_ip_address",
              "source_country", "source_reverse_dns", "source_base_domain",
              "delivery_result", "auth_failure", "reported_domain",
              "authentication_mechanisms", "sample_headers_only"]

    if type(reports) == OrderedDict:
        reports = [reports]
    csv_file = StringIO()
    csv_writer = DictWriter(csv_file, fieldnames=fields)
    csv_writer.writeheader()
    for report in reports:
        row = report.copy()
        row["source_ip_address"] = report["source"]["ip_address"]
        row["source_reverse_dns"] = report["source"]["reverse_dns"]
        row["source_base_domain"] = report["source"]["base_domain"]
        row["source_country"] = report["source"]["country"]
        del row["source"]
        row["subject"] = report["parsed_sample"]["subject"]
        row["auth_failure"] = ",".join(report["auth_failure"])
        authentication_mechanisms = report["authentication_mechanisms"]
        row["authentication_mechanisms"] = ",".join(
            authentication_mechanisms)
        del row["sample"]
        del row["parsed_sample"]
        csv_writer.writerow(row)

    return csv_file.getvalue()
def parsed_forensic_reports_to_csv ( reports ) : fields = [ "feedback_type" , "user_agent" , "version" , "original_envelope_id" , "original_mail_from" , "original_rcpt_to" , "arrival_date" , "arrival_date_utc" , "subject" , "message_id" , "authentication_results" , "dkim_domain" , "source_ip_address" , "source_country" , "source_reverse_dns" , "source_base_domain" , "delivery_result" , "auth_failure" , "reported_domain" , "authentication_mechanisms" , "sample_headers_only" ] if type ( reports ) == OrderedDict : reports = [ reports ] csv_file = StringIO ( ) csv_writer = DictWriter ( csv_file , fieldnames = fields ) csv_writer . writeheader ( ) for report in reports : row = report . copy ( ) row [ "source_ip_address" ] = report [ "source" ] [ "ip_address" ] row [ "source_reverse_dns" ] = report [ "source" ] [ "reverse_dns" ] row [ "source_base_domain" ] = report [ "source" ] [ "base_domain" ] row [ "source_country" ] = report [ "source" ] [ "country" ] del row [ "source" ] row [ "subject" ] = report [ "parsed_sample" ] [ "subject" ] row [ "auth_failure" ] = "," . join ( report [ "auth_failure" ] ) authentication_mechanisms = report [ "authentication_mechanisms" ] row [ "authentication_mechanisms" ] = "," . join ( authentication_mechanisms ) del row [ "sample" ] del row [ "parsed_sample" ] csv_writer . writerow ( row ) return csv_file . getvalue ( )
Converts one or more parsed forensic reports to flat CSV format, including
    headers

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

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