[PLGN-423]Updated timestamp handling (#2012)

* Updated timestamp handling

* Updated for timestamp handling

* Added changelog to help file

* Updated linting

* Updated timestamp handling logic

* Updated initialisation and handling of previous hashes

---------

Co-authored-by: Dympna Laverty <dympna_laverty@rapid7.com>

plugins/duo_admin/.CHECKSUM

@@ -1,7 +1,7 @@
 {
-	"spec": "77beeb58071c98d64a6a6e4c1915ebd3",
-	"manifest": "01bcc6c2542999d60ebb034e89c8b1d8",
-	"setup": "cca5518b93d244c3428e098cc5a83523",
+	"spec": "7539a3a6c37bc080ff7c6ce06e35f18c",
+	"manifest": "444a5c0139f94e0618426c4d3fd2b168",
+	"setup": "588b6772ffcf7bd56254cddc2ba929c7",
 	"schemas": [
 		{
 			"identifier": "add_user/schema.py",

plugins/duo_admin/bin/komand_duo_admin

@@ -6,7 +6,7 @@ from sys import argv
 
 Name = "Duo Admin API"
 Vendor = "rapid7"
-Version = "4.2.0"
+Version = "4.2.1"
 Description = "Duo is a trusted access solution for organizations. The Duo Admin plugin for Rapid7 InsightConnect allows users to manage and administrate their Duo organization"
 
 

plugins/duo_admin/help.md

@@ -1021,6 +1021,7 @@ A User ID can be obtained by passing a username to the Get User Status action.
 
 # Version History
 
+* 4.2.1 - Monitor Logs task: updated timestamp handling
 * 4.2.0 - Monitor Logs task: removed formatting of task output
 * 4.1.1 - Monitor Logs task: strip http/https in hostname, fix problem with generating header signature
 * 4.1.0 - Update to latest plugin SDK

plugins/duo_admin/komand_duo_admin/tasks/monitor_logs/task.py

@@ -11,6 +11,9 @@
 
 class MonitorLogs(insightconnect_plugin_runtime.Task):
     LAST_COLLECTION_TIMESTAMP = "last_collection_timestamp"
+    ADMIN_LOGS_LAST_LOG_TIMESTAMP = "admin_logs_last_log_timestamp"
+    AUTH_LOGS_LAST_LOG_TIMESTAMP = "auth_logs_last_log_timestamp"
+    TRUST_MONITOR_LAST_LOG_TIMESTAMP = "trust_monitor_last_log_timestamp"
     STATUS_CODE = "status_code"
     ADMIN_LOGS_NEXT_PAGE_PARAMS = "admin_logs_next_page_params"
     AUTH_LOGS_NEXT_PAGE_PARAMS = "auth_logs_next_page_params"
@@ -28,96 +31,153 @@ def __init__(self):
             state=MonitorLogsState(),
         )
 
+    def get_parameters_for_query(self, log_type, now, last_log_timestamp, next_page_params):
+        get_next_page = False
+        last_two_minutes = now - timedelta(minutes=2)
+        if not last_log_timestamp:
+            self.logger.info(f"First run for {log_type}")
+            last_24_hours = now - timedelta(hours=24)
+            if log_type != "Admin logs":
+                mintime = self.convert_to_milliseconds(last_24_hours)
+                maxtime = self.convert_to_milliseconds(last_two_minutes)
+            else:
+                # Use seconds for admin log endpoint
+                mintime = self.convert_to_seconds(last_24_hours)
+                maxtime = self.convert_to_seconds(last_two_minutes)
+
+        else:
+            if next_page_params:
+                self.logger.info("Getting the next page of results...")
+                get_next_page = True
+            else:
+                self.logger.info(f"Subsequent run for {log_type}")
+            if log_type != "Admin logs":
+                mintime = int(last_log_timestamp * 1000)
+                maxtime = self.convert_to_milliseconds(last_two_minutes)
+            else:
+                # Use seconds for admin log endpoint
+                # mintime = int(last_log_timestamp / 1000)
+                mintime = last_log_timestamp
+                maxtime = self.convert_to_seconds(last_two_minutes)
+
+        self.logger.info(f"Retrieve data from {mintime} to {maxtime}. Get next page is set to {get_next_page}")
+        return mintime, maxtime, get_next_page
+
     # pylint: disable=unused-argument
     def run(self, params={}, state={}):  # noqa: C901
         self.connection.admin_api.toggle_rate_limiting = False
         has_more_pages = False
-        get_next_page = False
+
         try:
             now = self.get_current_time()
-            last_minute = now - timedelta(minutes=1)
             last_collection_timestamp = state.get(self.LAST_COLLECTION_TIMESTAMP)
             trust_monitor_next_page_params = state.get(self.TRUST_MONITOR_NEXT_PAGE_PARAMS)
             auth_logs_next_page_params = state.get(self.AUTH_LOGS_NEXT_PAGE_PARAMS)
             admin_logs_next_page_params = state.get(self.ADMIN_LOGS_NEXT_PAGE_PARAMS)
 
-            if not last_collection_timestamp:
-                self.logger.info("First run")
-                last_24_hours = now - timedelta(hours=24)
-                mintime_in_milliseconds = self.convert_to_milliseconds(last_24_hours)
-                maxtime_in_milliseconds = self.convert_to_milliseconds(last_minute)
-                mintime_in_seconds = self.convert_to_seconds(last_24_hours)
-                maxtime_in_seconds = self.convert_to_seconds(last_minute)
+            if last_collection_timestamp:
+                # Previously only one timestamp was held (the end of the collection window)
+                # This has been superceded by a latest timestamp per log type
+                self.logger.info("Backwards compatibility - update all timestamps to the last known timestamp")
+                trust_monitor_last_log_timestamp = (
+                    auth_logs_last_log_timestamp
+                ) = admin_logs_last_log_timestamp = last_collection_timestamp
+                # Update the old last collection timestamp to None so it is not considered in future runs
+                state[self.LAST_COLLECTION_TIMESTAMP] = None
             else:
-                if trust_monitor_next_page_params or auth_logs_next_page_params or admin_logs_next_page_params:
-                    self.logger.info("Getting the next page of results...")
-                    get_next_page = True
-                else:
-                    self.logger.info("Subsequent run")
-                mintime_in_milliseconds = last_collection_timestamp
-                maxtime_in_milliseconds = self.convert_to_milliseconds(last_minute)
-                mintime_in_seconds = int(mintime_in_milliseconds / 1000)
-                maxtime_in_seconds = int(maxtime_in_milliseconds / 1000)
+                trust_monitor_last_log_timestamp = state.get(self.TRUST_MONITOR_LAST_LOG_TIMESTAMP)
+                auth_logs_last_log_timestamp = state.get(self.AUTH_LOGS_LAST_LOG_TIMESTAMP)
+                admin_logs_last_log_timestamp = state.get(self.ADMIN_LOGS_LAST_LOG_TIMESTAMP)
+                self.logger.info(
+                    f"Previous timestamps retrieved. "
+                    f"Auth {auth_logs_last_log_timestamp}. "
+                    f"Admin: {admin_logs_last_log_timestamp}. "
+                    f"Trust monitor {trust_monitor_last_log_timestamp}."
+                )
             try:
                 new_logs = []
-                if not get_next_page:
-                    state[self.LAST_COLLECTION_TIMESTAMP] = maxtime_in_milliseconds
 
                 previous_trust_monitor_event_hashes = state.get(self.PREVIOUS_TRUST_MONITOR_EVENT_HASHES, [])
                 previous_admin_log_hashes = state.get(self.PREVIOUS_ADMIN_LOG_HASHES, [])
                 previous_auth_log_hashes = state.get(self.PREVIOUS_AUTH_LOG_HASHES, [])
-                new_trust_monitor_event_hashes = []
-                new_admin_log_hashes = []
-                new_auth_log_hashes = []
+                new_trust_monitor_event_hashes, new_admin_log_hashes, new_auth_log_hashes = [], [], []
+
+                # Get trust monitor events
+                mintime, maxtime, get_next_page = self.get_parameters_for_query(
+                    "Trust monitor events", now, trust_monitor_last_log_timestamp, trust_monitor_next_page_params
+                )
 
                 if (get_next_page and trust_monitor_next_page_params) or not get_next_page:
                     trust_monitor_events, trust_monitor_next_page_params = self.get_trust_monitor_event(
-                        mintime_in_milliseconds, maxtime_in_milliseconds, trust_monitor_next_page_params
+                        mintime, maxtime, trust_monitor_next_page_params
                     )
                     new_trust_monitor_events, new_trust_monitor_event_hashes = self.compare_hashes(
                         previous_trust_monitor_event_hashes, trust_monitor_events
                     )
                     new_logs.extend(new_trust_monitor_events)
+                    state[self.TRUST_MONITOR_LAST_LOG_TIMESTAMP] = self.get_highest_timestamp(
+                        trust_monitor_last_log_timestamp, new_trust_monitor_events
+                    )
+                    self.logger.info(f"{len(new_trust_monitor_events)} trust monitor events retrieved")
+                if new_trust_monitor_event_hashes:
+                    state[self.PREVIOUS_TRUST_MONITOR_EVENT_HASHES] = new_trust_monitor_event_hashes
+
+                if trust_monitor_next_page_params:
+                    state[self.TRUST_MONITOR_NEXT_PAGE_PARAMS] = trust_monitor_next_page_params
+                    has_more_pages = True
+                elif state.get(self.TRUST_MONITOR_NEXT_PAGE_PARAMS):
+                    state.pop(self.TRUST_MONITOR_NEXT_PAGE_PARAMS)
+
+                # Get admin logs
+                mintime, maxtime, get_next_page = self.get_parameters_for_query(
+                    "Admin logs", now, admin_logs_last_log_timestamp, admin_logs_next_page_params
+                )
+
                 if (get_next_page and admin_logs_next_page_params) or not get_next_page:
                     admin_logs, admin_logs_next_page_params = self.get_admin_logs(
-                        mintime_in_seconds, maxtime_in_seconds, admin_logs_next_page_params
+                        mintime, maxtime, admin_logs_next_page_params
                     )
                     new_admin_logs, new_admin_log_hashes = self.compare_hashes(previous_admin_log_hashes, admin_logs)
                     new_logs.extend(new_admin_logs)
+                    state[self.ADMIN_LOGS_LAST_LOG_TIMESTAMP] = self.get_highest_timestamp(
+                        admin_logs_last_log_timestamp, new_admin_logs
+                    )
+                    self.logger.info(f"{len(new_admin_logs)} admin logs retrieved")
+
+                if new_admin_log_hashes:
+                    state[self.PREVIOUS_ADMIN_LOG_HASHES] = new_admin_log_hashes
+
+                if admin_logs_next_page_params:
+                    state[self.ADMIN_LOGS_NEXT_PAGE_PARAMS] = admin_logs_next_page_params
+                    has_more_pages = True
+                elif state.get(self.ADMIN_LOGS_NEXT_PAGE_PARAMS):
+                    state.pop(self.ADMIN_LOGS_NEXT_PAGE_PARAMS)
+
+                # Get auth logs
+                mintime, maxtime, get_next_page = self.get_parameters_for_query(
+                    "Auth logs", now, auth_logs_last_log_timestamp, auth_logs_next_page_params
+                )
+
                 if (get_next_page and auth_logs_next_page_params) or not get_next_page:
                     auth_logs, auth_logs_next_page_params = self.get_auth_logs(
-                        mintime_in_milliseconds, maxtime_in_milliseconds, auth_logs_next_page_params
+                        mintime, maxtime, auth_logs_next_page_params
                     )
                     new_auth_logs, new_auth_log_hashes = self.compare_hashes(previous_auth_log_hashes, auth_logs)
+                    # Grab the most recent timestamp and save it to use as min time for next run
                     new_logs.extend(new_auth_logs)
+                    state[self.AUTH_LOGS_LAST_LOG_TIMESTAMP] = self.get_highest_timestamp(
+                        auth_logs_last_log_timestamp, new_auth_logs
+                    )
+                    self.logger.info(f"{len(new_auth_logs)} auth logs retrieved")
 
-                state[self.PREVIOUS_TRUST_MONITOR_EVENT_HASHES] = (
-                    previous_trust_monitor_event_hashes
-                    if not new_trust_monitor_event_hashes and get_next_page
-                    else new_trust_monitor_event_hashes
-                )
-                state[self.PREVIOUS_ADMIN_LOG_HASHES] = (
-                    previous_admin_log_hashes if not new_admin_log_hashes and get_next_page else new_admin_log_hashes
-                )
-                state[self.PREVIOUS_AUTH_LOG_HASHES] = (
-                    previous_auth_log_hashes if not new_auth_log_hashes and get_next_page else new_auth_log_hashes
-                )
+                if new_auth_log_hashes:
+                    state[self.PREVIOUS_AUTH_LOG_HASHES] = new_auth_log_hashes
 
-                if trust_monitor_next_page_params:
-                    state[self.TRUST_MONITOR_NEXT_PAGE_PARAMS] = trust_monitor_next_page_params
-                    has_more_pages = True
-                elif state.get(self.TRUST_MONITOR_NEXT_PAGE_PARAMS):
-                    state.pop(self.TRUST_MONITOR_NEXT_PAGE_PARAMS)
                 if auth_logs_next_page_params:
                     state[self.AUTH_LOGS_NEXT_PAGE_PARAMS] = auth_logs_next_page_params
                     has_more_pages = True
                 elif state.get(self.AUTH_LOGS_NEXT_PAGE_PARAMS):
                     state.pop(self.AUTH_LOGS_NEXT_PAGE_PARAMS)
-                if admin_logs_next_page_params:
-                    state[self.ADMIN_LOGS_NEXT_PAGE_PARAMS] = admin_logs_next_page_params
-                    has_more_pages = True
-                elif state.get(self.ADMIN_LOGS_NEXT_PAGE_PARAMS):
-                    state.pop(self.ADMIN_LOGS_NEXT_PAGE_PARAMS)
 
                 return new_logs, state, has_more_pages, 200, None
             except ApiException as error:
@@ -164,14 +224,32 @@ def compare_hashes(self, previous_logs_hashes: list, new_logs: list):
             if hash_ not in previous_logs_hashes:
                 new_logs_hashes.append(hash_)
                 logs_to_return.append(log)
+        self.logger.info(
+            f"Original number of logs:{len(new_logs)}. Number of logs after de-duplication:{len(logs_to_return)}"
+        )
         return logs_to_return, new_logs_hashes
 
+    def get_highest_timestamp(self, last_recorded_highest_timestamp, logs):
+        if last_recorded_highest_timestamp:
+            highest_timestamp = last_recorded_highest_timestamp
+        else:
+            highest_timestamp = 0
+        for log in logs:
+            log_timestamp = log.get("timestamp")
+            if log_timestamp and log_timestamp > highest_timestamp:
+                highest_timestamp = log_timestamp
+        self.logger.info(f"Highest timestamp set to {highest_timestamp}")
+        return highest_timestamp
+
     def get_auth_logs(self, mintime: int, maxtime: int, next_page_params: dict) -> list:
+        self.logger.info(f"Get auth logs: mintime:{mintime}, maxtime:{maxtime}, next_page_params:{next_page_params}")
         parameters = (
             next_page_params
             if next_page_params
             else {"mintime": str(mintime), "maxtime": str(maxtime), "limit": str(1000)}
         )
+        parameters.update({"sort": "ts:asc"})
+        self.logger.info(f"Parameters for get auth logs set to {parameters}")
         response = self.connection.admin_api.get_auth_logs(parameters).get("response", {})
         metadata = response.get("metadata") or {}
         next_offset = metadata.get("next_offset")
@@ -180,10 +258,13 @@ def get_auth_logs(self, mintime: int, maxtime: int, next_page_params: dict) -> l
         else:
             parameters = {}
         auth_logs = self.add_log_type_field(response.get("authlogs", []), "authentication")
+        self.logger.info(f"Get auth logs: parameters to return {parameters}. Return {len(auth_logs)} logs")
         return auth_logs, parameters
 
     def get_admin_logs(self, mintime: int, maxtime: int, next_page_params: dict) -> list:
+        self.logger.info(f"Get admin logs: mintime:{mintime}, maxtime:{maxtime}, next_page_params:{next_page_params}")
         parameters = {"mintime": next_page_params.get("mintime") if next_page_params else str(mintime)}
+        self.logger.info(f"Parameters for get admin logs set to {parameters}")
         response = self.connection.admin_api.get_admin_logs(parameters).get("response", [])
         last_item = response[-1] if response and isinstance(response, list) else None
 
@@ -198,6 +279,7 @@ def get_admin_logs(self, mintime: int, maxtime: int, next_page_params: dict) ->
                 logs_to_return.append(log)
 
         admin_logs = self.add_log_type_field(logs_to_return, "administrator")
+        self.logger.info(f"Parameters to return from get admin logs set to {parameters}. Return {len(admin_logs)} logs")
         return admin_logs, parameters
 
     def get_trust_monitor_event(self, mintime: int, maxtime: int, next_page_params: dict) -> list:
@@ -206,6 +288,7 @@ def get_trust_monitor_event(self, mintime: int, maxtime: int, next_page_params:
             if next_page_params
             else {"mintime": str(mintime), "maxtime": str(maxtime), "limit": str(200)}
         )
+        self.logger.info(f"Parameters for get trust monitor events set to {parameters}")
         response = self.connection.admin_api.get_trust_monitor_events(parameters).get("response", {})
         offset = response.get("metadata", {}).get("next_offset")
         if offset:

plugins/duo_admin/plugin.spec.yaml

@@ -13,7 +13,7 @@ sdk:
   version: 5
   user: nobody
 description: Duo is a trusted access solution for organizations. The Duo Admin plugin for Rapid7 InsightConnect allows users to manage and administrate their Duo organization
-version: 4.2.0
+version: 4.2.1
 connection_version: 4
 resources:
   source_url: https://github.com/rapid7/insightconnect-plugins/tree/master/plugins/duo_admin

plugins/duo_admin/setup.py

@@ -3,7 +3,7 @@
 
 
 setup(name="duo_admin-rapid7-plugin",
-      version="4.2.0",
+      version="4.2.1",
       description="Duo is a trusted access solution for organizations. The Duo Admin plugin for Rapid7 InsightConnect allows users to manage and administrate their Duo organization",
       author="rapid7",
       author_email="",