Update changelog

- Ignore duplicate aggregate DMARC reports with the same `org_name` and `report_id` seen within the same hour ([#539](https://github.com/domainaware/parsedmarc/issues/539))
- Fix saving SMTP TLS reports to OpenSearch (PR #585 closed issue #576)
- Add 303 entries to `base_reverse_dns_map.csv`

.gitignore

@@ -136,3 +136,6 @@ samples/private
 
 *.html
 *.sqlite-journal
+
+parsedmarc.ini
+scratch.py

.vscode/settings.json

@@ -70,6 +70,7 @@
         "modindex",
         "msgconvert",
         "msgraph",
+        "MSSP",
         "Munge",
         "ndjson",
         "newkey",

CHANGELOG.md

@@ -1,6 +1,23 @@
 Changelog
 =========
 
+8.17.0
+------
+
+- Ignore duplicate aggregate DMARC reports with the same `org_name` and `report_id` seen within the same hour (Fixes [#539](https://github.com/domainaware/parsedmarc/issues/539))
+- Fix saving SMTP TLS reports to OpenSearch (PR #585 closed issue #576)
+- Add 303 entries to `base_reverse_dns_map.csv`
+
+8.16.1
+------
+
+- Failed attempt to ignore aggregate DMARC reports seen within a period of one hour (#535)
+
+8.16.0
+------
+
+- Add a `since` option to only search for emails since a certain time (PR #527)
+
 8.15.4
 ------
 

build.sh

@@ -16,6 +16,7 @@ make html
 touch build/html/.nojekyll
 cp -rf build/html/* ../../parsedmarc-docs/
 cd ..
+./sortmaps.py
 python3 tests.py
 rm -rf dist/ build/
 hatch build

docker-compose.yml

@@ -28,3 +28,30 @@ services:
       interval: 10s
       timeout: 10s
       retries: 24
+
+  opensearch:
+    image: opensearchproject/opensearch:2.18.0
+    environment:
+      - network.host=127.0.0.1
+      - http.host=0.0.0.0
+      - node.name=opensearch
+      - discovery.type=single-node
+      - cluster.name=parsedmarc-cluster
+      - discovery.seed_hosts=opensearch
+      - bootstrap.memory_lock=true
+      - OPENSEARCH_INITIAL_ADMIN_PASSWORD=${OPENSEARCH_INITIAL_ADMIN_PASSWORD}
+    ports:
+      - 127.0.0.1:9201:9200
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "curl -s -XGET http://localhost:9201/_cluster/health?pretty | grep status | grep -q '\\(green\\|yellow\\)'"
+        ]
+      interval: 10s
+      timeout: 10s
+      retries: 24

parsedmarc/__init__.py

@@ -28,14 +28,18 @@ from lxml import etree
 from mailsuite.smtp import send_email
 
 from parsedmarc.log import logger
-from parsedmarc.mail import MailboxConnection, IMAPConnection, \
-    MSGraphConnection, GmailConnection
+from parsedmarc.mail import (
+    MailboxConnection,
+    IMAPConnection,
+    MSGraphConnection,
+    GmailConnection,
+)
 from parsedmarc.utils import get_base_domain, get_ip_address_info
 from parsedmarc.utils import is_outlook_msg, convert_outlook_msg
 from parsedmarc.utils import parse_email
 from parsedmarc.utils import timestamp_to_human, human_timestamp_to_datetime
 
-__version__ = "8.15.4"
+__version__ = "8.17.0"
 
 logger.debug("parsedmarc v{0}".format(__version__))
 
@@ -50,6 +54,7 @@ MAGIC_XML = b"\x3c\x3f\x78\x6d\x6c\x20"
 MAGIC_JSON = b"\7b"
 
 IP_ADDRESS_CACHE = ExpiringDict(max_len=10000, max_age_seconds=14400)
+SEEN_AGGREGATE_REPORT_IDS = ExpiringDict(max_len=100000000, max_age_seconds=3600)
 REVERSE_DNS_MAP = dict()
 
 
@@ -1466,7 +1471,17 @@ def get_dmarc_reports_from_mbox(
                     strip_attachment_payloads=sa,
                 )
                 if parsed_email["report_type"] == "aggregate":
-                    aggregate_reports.append(parsed_email["report"])
+                    report_org = parsed_email["report"]["report_metadata"]["org_name"]
+                    report_id = parsed_email["report"]["report_metadata"]["report_id"]
+                    report_key = f"{report_org}_{report_id}"
+                    if report_key not in SEEN_AGGREGATE_REPORT_IDS:
+                        SEEN_AGGREGATE_REPORT_IDS[report_key] = True
+                        aggregate_reports.append(parsed_email["report"])
+                    else:
+                        logger.debug(
+                            "Skipping duplicate aggregate report "
+                            f"from {report_org} with ID: {report_id}"
+                        )
                 elif parsed_email["report_type"] == "forensic":
                     forensic_reports.append(parsed_email["report"])
                 elif parsed_email["report_type"] == "smtp_tls":
@@ -1484,23 +1499,25 @@ def get_dmarc_reports_from_mbox(
     )
 
 
-def get_dmarc_reports_from_mailbox(connection: MailboxConnection,
-                                   reports_folder="INBOX",
-                                   archive_folder="Archive",
-                                   delete=False,
-                                   test=False,
-                                   ip_db_path=None,
-                                   always_use_local_files=False,
-                                   reverse_dns_map_path=None,
-                                   reverse_dns_map_url=None,
-                                   offline=False,
-                                   nameservers=None,
-                                   dns_timeout=6.0,
-                                   strip_attachment_payloads=False,
-                                   results=None,
-                                   batch_size=10,
-                                   since=None,
-                                   create_folders=True):
+def get_dmarc_reports_from_mailbox(
+    connection: MailboxConnection,
+    reports_folder="INBOX",
+    archive_folder="Archive",
+    delete=False,
+    test=False,
+    ip_db_path=None,
+    always_use_local_files=False,
+    reverse_dns_map_path=None,
+    reverse_dns_map_url=None,
+    offline=False,
+    nameservers=None,
+    dns_timeout=6.0,
+    strip_attachment_payloads=False,
+    results=None,
+    batch_size=10,
+    since=None,
+    create_folders=True,
+):
     """
     Fetches and parses DMARC reports from a mailbox
 
@@ -1564,42 +1581,44 @@ def get_dmarc_reports_from_mailbox(connection: MailboxConnection,
 
     if since:
         _since = 1440  # default one day
-        if re.match(r'\d+[mhd]$', since):
-            s = re.split(r'(\d+)', since)
-            if s[2] == 'm':
+        if re.match(r"\d+[mhd]$", since):
+            s = re.split(r"(\d+)", since)
+            if s[2] == "m":
                 _since = int(s[1])
-            elif s[2] == 'h':
-                _since = int(s[1])*60
-            elif s[2] == 'd':
-                _since = int(s[1])*60*24
-            elif s[2] == 'w':
-                _since = int(s[1])*60*24*7
+            elif s[2] == "h":
+                _since = int(s[1]) * 60
+            elif s[2] == "d":
+                _since = int(s[1]) * 60 * 24
+            elif s[2] == "w":
+                _since = int(s[1]) * 60 * 24 * 7
         else:
-            logger.warning("Incorrect format for \'since\' option. \
+            logger.warning(
+                "Incorrect format for 'since' option. \
                            Provided value:{0}, Expected values:(5m|3h|2d|1w). \
-                           Ignoring option, fetching messages for last 24hrs" \
-                           "SMTP does not support a time or timezone in since." \
-                           "See https://www.rfc-editor.org/rfc/rfc3501#page-52"
-                           .format(since))
+                           Ignoring option, fetching messages for last 24hrs"
+                "SMTP does not support a time or timezone in since."
+                "See https://www.rfc-editor.org/rfc/rfc3501#page-52".format(since)
+            )
 
         if isinstance(connection, IMAPConnection):
-            logger.debug("Only days and weeks values in \'since\' option are \
-                         considered for IMAP conections. Examples: 2d or 1w")
+            logger.debug(
+                "Only days and weeks values in 'since' option are \
+                         considered for IMAP conections. Examples: 2d or 1w"
+            )
             since = (datetime.utcnow() - timedelta(minutes=_since)).date()
             current_time = datetime.utcnow().date()
         elif isinstance(connection, MSGraphConnection):
-            since = (datetime.utcnow() - timedelta(minutes=_since)) \
-                .isoformat() + 'Z'
-            current_time = datetime.utcnow().isoformat() + 'Z'
+            since = (datetime.utcnow() - timedelta(minutes=_since)).isoformat() + "Z"
+            current_time = datetime.utcnow().isoformat() + "Z"
         elif isinstance(connection, GmailConnection):
-            since = (datetime.utcnow() - timedelta(minutes=_since)) \
-                .strftime('%s')
-            current_time = datetime.utcnow().strftime('%s')
+            since = (datetime.utcnow() - timedelta(minutes=_since)).strftime("%s")
+            current_time = datetime.utcnow().strftime("%s")
         else:
             pass
 
-    messages = connection.fetch_messages(reports_folder, batch_size=batch_size,
-                                         since=since)
+    messages = connection.fetch_messages(
+        reports_folder, batch_size=batch_size, since=since
+    )
     total_messages = len(messages)
     logger.debug("Found {0} messages in {1}".format(len(messages), reports_folder))
 
@@ -1612,16 +1631,16 @@ def get_dmarc_reports_from_mailbox(connection: MailboxConnection,
 
     for i in range(message_limit):
         msg_uid = messages[i]
-        logger.debug("Processing message {0} of {1}: UID {2}".format(
-            i+1, message_limit, msg_uid
-        ))
+        logger.debug(
+            "Processing message {0} of {1}: UID {2}".format(
+                i + 1, message_limit, msg_uid
+            )
+        )
         if isinstance(mailbox, MSGraphConnection):
             if test:
-                msg_content = connection.fetch_message(msg_uid,
-                                                       mark_read=False)
+                msg_content = connection.fetch_message(msg_uid, mark_read=False)
             else:
-                msg_content = connection.fetch_message(msg_uid,
-                                                       mark_read=True)
+                msg_content = connection.fetch_message(msg_uid, mark_read=True)
         else:
             msg_content = connection.fetch_message(msg_uid)
         try:
@@ -1639,7 +1658,16 @@ def get_dmarc_reports_from_mailbox(connection: MailboxConnection,
                 keep_alive=connection.keepalive,
             )
             if parsed_email["report_type"] == "aggregate":
-                aggregate_reports.append(parsed_email["report"])
+                report_org = parsed_email["report"]["report_metadata"]["org_name"]
+                report_id = parsed_email["report"]["report_metadata"]["report_id"]
+                report_key = f"{report_org}_{report_id}"
+                if report_key not in SEEN_AGGREGATE_REPORT_IDS:
+                    SEEN_AGGREGATE_REPORT_IDS[report_key] = True
+                    aggregate_reports.append(parsed_email["report"])
+                else:
+                    logger.debug(
+                        "Skipping duplicate aggregate report " f"with ID: {report_id}"
+                    )
                 aggregate_report_msg_uids.append(msg_uid)
             elif parsed_email["report_type"] == "forensic":
                 forensic_reports.append(parsed_email["report"])
@@ -1755,8 +1783,9 @@ def get_dmarc_reports_from_mailbox(connection: MailboxConnection,
     )
 
     if current_time:
-        total_messages = len(connection.fetch_messages(reports_folder,
-                                                       since=current_time))
+        total_messages = len(
+            connection.fetch_messages(reports_folder, since=current_time)
+        )
     else:
         total_messages = len(connection.fetch_messages(reports_folder))
 

parsedmarc/cli.py

@@ -46,6 +46,7 @@ from parsedmarc.mail.graph import AuthMethod
 
 from parsedmarc.log import logger
 from parsedmarc.utils import is_mbox, get_reverse_dns
+from parsedmarc import SEEN_AGGREGATE_REPORT_IDS
 
 formatter = logging.Formatter(
     fmt="%(levelname)8s:%(filename)s:%(lineno)d:%(message)s",
@@ -714,8 +715,7 @@ def _main():
             if "batch_size" in mailbox_config:
                 opts.mailbox_batch_size = mailbox_config.getint("batch_size")
             if "check_timeout" in mailbox_config:
-                opts.mailbox_check_timeout = mailbox_config.getint(
-                    "check_timeout")
+                opts.mailbox_check_timeout = mailbox_config.getint("check_timeout")
             if "since" in mailbox_config:
                 opts.mailbox_since = mailbox_config["since"]
 
@@ -1419,7 +1419,17 @@ def _main():
             logger.error("Failed to parse {0} - {1}".format(result[1], result[0]))
         else:
             if result[0]["report_type"] == "aggregate":
-                aggregate_reports.append(result[0]["report"])
+                report_org = result[0]["report"]["report_metadata"]["org_name"]
+                report_id = result[0]["report"]["report_metadata"]["report_id"]
+                report_key = f"{report_org}_{report_id}"
+                if report_key not in SEEN_AGGREGATE_REPORT_IDS:
+                    SEEN_AGGREGATE_REPORT_IDS[report_key] = True
+                    aggregate_reports.append(result[0]["report"])
+                else:
+                    logger.debug(
+                        "Skipping duplicate aggregate report "
+                        f"from {report_org} with ID: {report_id}"
+                    )
             elif result[0]["report_type"] == "forensic":
                 forensic_reports.append(result[0]["report"])
             elif result[0]["report_type"] == "smtp_tls":

parsedmarc/mail/gmail.py

@@ -69,8 +69,7 @@ class GmailConnection(MailboxConnection):
             else:
                 raise e
 
-    def _fetch_all_message_ids(self, reports_label_id, page_token=None,
-                               since=None):
+    def _fetch_all_message_ids(self, reports_label_id, page_token=None, since=None):
         if since:
             results = (
                 self.service.users()
@@ -80,7 +79,7 @@ class GmailConnection(MailboxConnection):
                     includeSpamTrash=self.include_spam_trash,
                     labelIds=[reports_label_id],
                     pageToken=page_token,
-                    q=f'after:{since}',
+                    q=f"after:{since}",
                 )
                 .execute()
             )
@@ -107,10 +106,11 @@ class GmailConnection(MailboxConnection):
 
     def fetch_messages(self, reports_folder: str, **kwargs) -> List[str]:
         reports_label_id = self._find_label_id_for_label(reports_folder)
-        since = kwargs.get('since')
+        since = kwargs.get("since")
         if since:
-            return [id for id in self._fetch_all_message_ids(reports_label_id,
-                                                             since=since)]
+            return [
+                id for id in self._fetch_all_message_ids(reports_label_id, since=since)
+            ]
         else:
             return [id for id in self._fetch_all_message_ids(reports_label_id)]
 

parsedmarc/mail/graph.py

@@ -146,24 +146,21 @@ class MSGraphConnection(MailboxConnection):
     def fetch_messages(self, folder_name: str, **kwargs) -> List[str]:
         """Returns a list of message UIDs in the specified folder"""
         folder_id = self._find_folder_id_from_folder_path(folder_name)
-        url = f'/users/{self.mailbox_name}/mailFolders/' \
-              f'{folder_id}/messages'
-        since = kwargs.get('since')
+        url = f"/users/{self.mailbox_name}/mailFolders/" f"{folder_id}/messages"
+        since = kwargs.get("since")
         if not since:
             since = None
-        batch_size = kwargs.get('batch_size')
+        batch_size = kwargs.get("batch_size")
         if not batch_size:
             batch_size = 0
         emails = self._get_all_messages(url, batch_size, since)
-        return [email['id'] for email in emails]
+        return [email["id"] for email in emails]
 
     def _get_all_messages(self, url, batch_size, since):
         messages: list
-        params = {
-            '$select': 'id'
-        }
+        params = {"$select": "id"}
         if since:
-            params['$filter'] = f'receivedDateTime ge {since}'
+            params["$filter"] = f"receivedDateTime ge {since}"
         if batch_size and batch_size > 0:
             params["$top"] = batch_size
         else:
@@ -173,11 +170,10 @@ class MSGraphConnection(MailboxConnection):
             raise RuntimeError(f"Failed to fetch messages {result.text}")
         messages = result.json()["value"]
         # Loop if next page is present and not obtained message limit.
-        while '@odata.nextLink' in result.json() and (
-                since is not None or (
-                batch_size == 0 or
-                batch_size - len(messages) > 0)):
-            result = self._client.get(result.json()['@odata.nextLink'])
+        while "@odata.nextLink" in result.json() and (
+            since is not None or (batch_size == 0 or batch_size - len(messages) > 0)
+        ):
+            result = self._client.get(result.json()["@odata.nextLink"])
             if result.status_code != 200:
                 raise RuntimeError(f"Failed to fetch messages {result.text}")
             messages.extend(result.json()["value"])
@@ -193,12 +189,13 @@ class MSGraphConnection(MailboxConnection):
             )
 
     def fetch_message(self, message_id: str, **kwargs):
-        url = f'/users/{self.mailbox_name}/messages/{message_id}/$value'
+        url = f"/users/{self.mailbox_name}/messages/{message_id}/$value"
         result = self._client.get(url)
         if result.status_code != 200:
-            raise RuntimeWarning(f"Failed to fetch message"
-                                 f"{result.status_code}: {result.json()}")
-        mark_read = kwargs.get('mark_read')
+            raise RuntimeWarning(
+                f"Failed to fetch message" f"{result.status_code}: {result.json()}"
+            )
+        mark_read = kwargs.get("mark_read")
         if mark_read:
             self.mark_message_read(message_id)
         return result.text

parsedmarc/mail/imap.py

@@ -39,9 +39,9 @@ class IMAPConnection(MailboxConnection):
 
     def fetch_messages(self, reports_folder: str, **kwargs):
         self._client.select_folder(reports_folder)
-        since = kwargs.get('since')
+        since = kwargs.get("since")
         if since:
-            return self._client.search([u'SINCE', since])
+            return self._client.search(["SINCE", since])
         else:
             return self._client.search()
 

parsedmarc/opensearch.py

@@ -202,13 +202,15 @@ class _SMTPTLSPolicyDoc(InnerDoc):
         receiving_ip,
         receiving_mx_helo,
         failed_session_count,
+        sending_mta_ip=None,
         receiving_mx_hostname=None,
         additional_information_uri=None,
         failure_reason_code=None,
     ):
-        self.failure_details.append(
+        _details = _SMTPTLSFailureDetailsDoc(
             result_type=result_type,
             ip_address=ip_address,
+            sending_mta_ip=sending_mta_ip,
             receiving_mx_hostname=receiving_mx_hostname,
             receiving_mx_helo=receiving_mx_helo,
             receiving_ip=receiving_ip,
@@ -216,9 +218,10 @@ class _SMTPTLSPolicyDoc(InnerDoc):
             additional_information=additional_information_uri,
             failure_reason_code=failure_reason_code,
         )
+        self.failure_details.append(_details)
 
 
-class _SMTPTLSFailureReportDoc(Document):
+class _SMTPTLSReportDoc(Document):
     class Index:
         name = "smtp_tls"
 
@@ -499,6 +502,7 @@ def save_aggregate_report_to_opensearch(
             index = "{0}_{1}".format(index, index_suffix)
         if index_prefix:
             index = "{0}{1}".format(index_prefix, index)
+
         index = "{0}-{1}".format(index, index_date)
         index_settings = dict(
             number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
@@ -685,7 +689,7 @@ def save_smtp_tls_report_to_opensearch(
             AlreadySaved
     """
     logger.info("Saving aggregate report to OpenSearch")
-    org_name = report["org_name"]
+    org_name = report["organization_name"]
     report_id = report["report_id"]
     begin_date = human_timestamp_to_datetime(report["begin_date"], to_utc=True)
     end_date = human_timestamp_to_datetime(report["end_date"], to_utc=True)
@@ -741,11 +745,11 @@ def save_smtp_tls_report_to_opensearch(
         number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
     )
 
-    smtp_tls_doc = _SMTPTLSFailureReportDoc(
-        organization_name=report["organization_name"],
-        date_range=[report["date_begin"], report["date_end"]],
-        date_begin=report["date_begin"],
-        date_end=report["date_end"],
+    smtp_tls_doc = _SMTPTLSReportDoc(
+        org_name=report["organization_name"],
+        date_range=[report["begin_date"], report["end_date"]],
+        date_begin=report["begin_date"],
+        date_end=report["end_date"],
         contact_info=report["contact_info"],
         report_id=report["report_id"],
     )
@@ -760,32 +764,48 @@ def save_smtp_tls_report_to_opensearch(
         policy_doc = _SMTPTLSPolicyDoc(
             policy_domain=policy["policy_domain"],
             policy_type=policy["policy_type"],
+            succesful_session_count=policy["successful_session_count"],
+            failed_session_count=policy["failed_session_count"],
             policy_string=policy_strings,
             mx_host_patterns=mx_host_patterns,
         )
         if "failure_details" in policy:
-            failure_details = policy["failure_details"]
-            receiving_mx_hostname = None
-            additional_information_uri = None
-            failure_reason_code = None
-            if "receiving_mx_hostname" in failure_details:
-                receiving_mx_hostname = failure_details["receiving_mx_hostname"]
-            if "additional_information_uri" in failure_details:
-                additional_information_uri = failure_details[
-                    "additional_information_uri"
-                ]
-            if "failure_reason_code" in failure_details:
-                failure_reason_code = failure_details["failure_reason_code"]
-            policy_doc.add_failure_details(
-                result_type=failure_details["result_type"],
-                ip_address=failure_details["ip_address"],
-                receiving_ip=failure_details["receiving_ip"],
-                receiving_mx_helo=failure_details["receiving_mx_helo"],
-                failed_session_count=failure_details["failed_session_count"],
-                receiving_mx_hostname=receiving_mx_hostname,
-                additional_information_uri=additional_information_uri,
-                failure_reason_code=failure_reason_code,
-            )
+            for failure_detail in policy["failure_details"]:
+                receiving_mx_hostname = None
+                additional_information_uri = None
+                failure_reason_code = None
+                ip_address = None
+                receiving_ip = None
+                receiving_mx_helo = None
+                sending_mta_ip = None
+
+                if "receiving_mx_hostname" in failure_detail:
+                    receiving_mx_hostname = failure_detail["receiving_mx_hostname"]
+                if "additional_information_uri" in failure_detail:
+                    additional_information_uri = failure_detail[
+                        "additional_information_uri"
+                    ]
+                if "failure_reason_code" in failure_detail:
+                    failure_reason_code = failure_detail["failure_reason_code"]
+                if "ip_address" in failure_detail:
+                    ip_address = failure_detail["ip_address"]
+                if "receiving_ip" in failure_detail:
+                    receiving_ip = failure_detail["receiving_ip"]
+                if "receiving_mx_helo" in failure_detail:
+                    receiving_mx_helo = failure_detail["receiving_mx_helo"]
+                if "sending_mta_ip" in failure_detail:
+                    sending_mta_ip = failure_detail["sending_mta_ip"]
+                policy_doc.add_failure_details(
+                    result_type=failure_detail["result_type"],
+                    ip_address=ip_address,
+                    receiving_ip=receiving_ip,
+                    receiving_mx_helo=receiving_mx_helo,
+                    failed_session_count=failure_detail["failed_session_count"],
+                    sending_mta_ip=sending_mta_ip,
+                    receiving_mx_hostname=receiving_mx_hostname,
+                    additional_information_uri=additional_information_uri,
+                    failure_reason_code=failure_reason_code,
+                )
         smtp_tls_doc.policies.append(policy_doc)
 
     create_indexes([index], index_settings)

pyproject.toml

@@ -1,6 +1,6 @@
 [build-system]
 requires = [
-    "hatchling>=1.8.1",
+    "hatchling>=1.27.0",
 ]
 build-backend = "hatchling.build"
 
@@ -59,7 +59,7 @@ dependencies = [
 
 [project.optional-dependencies]
 build = [
-    "hatch",
+    "hatch>=1.14.0",
     "myst-parser[linkify]",
     "nose",
     "pytest",

sortmaps.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+
+import os
+import glob
+import csv
+
+
+maps_dir = os.path.join("parsedmarc", "resources", "maps")
+csv_files = glob.glob(os.path.join(maps_dir, "*.csv"))
+
+
+def sort_csv(filepath, column=0):
+    with open(filepath, mode="r", newline="") as infile:
+        reader = csv.reader(infile)
+        header = next(reader)
+        sorted_rows = sorted(reader, key=lambda row: row[column])
+
+    with open(filepath, mode="w", newline="\n") as outfile:
+        writer = csv.writer(outfile)
+        writer.writerow(header)
+        writer.writerows(sorted_rows)
+
+
+for csv_file in csv_files:
+    sort_csv(csv_file)