Okta 4.2.1 Release: Improved Monitor Logs Task

plugins/okta/.CHECKSUM

@@ -1,7 +1,7 @@
 {
-	"spec": "2a124bebc54c6ed4bbefa8caa038ce0f",
-	"manifest": "96a067bc255b4bd4eab6fb3dfae79354",
-	"setup": "f41cee058863b3ed32c6048d876f1ea1",
+	"spec": "3ad1604efd5761d7129ee19e728efb5d",
+	"manifest": "ae6c3d90c00d8b25576f218a99b62763",
+	"setup": "f0f0aa00f602f9aa8da621bfe91e7107",
 	"schemas": [
 		{
 			"identifier": "add_user_to_group/schema.py",

plugins/okta/bin/komand_okta

@@ -6,7 +6,7 @@ from sys import argv
 
 Name = "Okta"
 Vendor = "rapid7"
-Version = "4.2.0"
+Version = "4.2.1"
 Description = "Secure identity management and single sign-on to any application"
 
 

plugins/okta/help.md

@@ -1612,6 +1612,7 @@ by Okta themselves, or constructed by the plugin based on the information it has
 
 # Version History
 
+* 4.2.1 - Monitor Logs task: filter previously returned log events | only update time checkpoint when an event is returned | update timestamp format | set cutoff time of 24 hours.
 * 4.2.0 - Monitor Logs task: return raw logs data without cleaning and use last log time as checkpoint in time for next run.
 * 4.1.1 - Monitor Logs task: strip http/https in hostname
 * 4.1.0 - New action Get User Groups | Update to latest SDK version

plugins/okta/komand_okta/tasks/monitor_logs/task.py

@@ -28,19 +28,23 @@ def run(self, params={}, state={}):  # pylint: disable=unused-argument
         parameters = {}
         try:
             now = self.get_current_time() - timedelta(minutes=1)  # allow for latency of this being triggered
-            now_iso = now.isoformat()
+            now_iso = self.get_iso(now)
+            last_24_hours = self.get_iso(now - timedelta(hours=24))  # cut off point - never query beyond 24 hours
             next_page_link = state.get(self.NEXT_PAGE_LINK)
             if not state:
                 self.logger.info("First run")
-                last_24_hours = now - timedelta(hours=24)
-                parameters = {"since": last_24_hours.isoformat(), "until": now_iso, "limit": 1000}
-                state[self.LAST_COLLECTION_TIMESTAMP] = now_iso
+                parameters = {"since": last_24_hours, "until": now_iso, "limit": 1000}
+                state[self.LAST_COLLECTION_TIMESTAMP] = last_24_hours  # we only change this once we get new events
             else:
                 if next_page_link:
                     state.pop(self.NEXT_PAGE_LINK)
                     self.logger.info("Getting the next page of results...")
                 else:
-                    parameters = {"since": state.get(self.LAST_COLLECTION_TIMESTAMP), "until": now_iso, "limit": 1000}
+                    parameters = {
+                        "since": self.get_since(state, last_24_hours),
+                        "until": now_iso,
+                        "limit": 1000,
+                    }
                     self.logger.info("Subsequent run...")
             try:
                 self.logger.info(f"Calling Okta with parameters={parameters} and next_page={next_page_link}")
@@ -49,11 +53,12 @@ def run(self, params={}, state={}):  # pylint: disable=unused-argument
                     if not next_page_link
                     else self.connection.api_client.get_next_page(next_page_link)
                 )
-                next_page_link, new_logs = self.get_next_page_link(new_logs_resp.headers), new_logs_resp.json()
+                next_page_link = self.get_next_page_link(new_logs_resp.headers)
+                new_logs = self.get_events(new_logs_resp.json(), state.get(self.LAST_COLLECTION_TIMESTAMP))
                 if next_page_link:
                     state[self.NEXT_PAGE_LINK] = next_page_link
                     has_more_pages = True
-                state[self.LAST_COLLECTION_TIMESTAMP] = self.get_last_collection_timestamp(now_iso, new_logs)
+                state[self.LAST_COLLECTION_TIMESTAMP] = self.get_last_collection_timestamp(new_logs, state)
                 return new_logs, state, has_more_pages, 200, None
             except ApiException as error:
                 return [], state, False, error.status_code, error
@@ -64,22 +69,94 @@ def run(self, params={}, state={}):  # pylint: disable=unused-argument
     def get_current_time():
         return datetime.now(timezone.utc)
 
+    @staticmethod
+    def get_iso(time: datetime) -> str:
+        """
+        Match the timestamp format used in old collector code and the format that Okta uses for 'published' to allow
+        comparison in `get_events`. e.g. '2023-10-02T15:43:51.450Z'
+        :param time: newly formatted time string value.
+        :return: formatted time string.
+        """
+        return time.isoformat("T", "milliseconds").replace("+00:00", "Z")
+
+    def get_since(self, state: dict, cut_off: str) -> str:
+        """
+        If the customer has paused this task for an extended amount of time we don't want start polling events that
+        exceed 24 hours ago. Check if the saved state is beyond this and revert to use the last 24 hours time.
+        :param state: saved state to check and update the time being used.
+        :param cut_off: string time of now - 24 hours.
+        :return: updated time string to use in the parameters.
+        """
+        saved_time = state.get(self.LAST_COLLECTION_TIMESTAMP)
+
+        if saved_time < cut_off:
+            self.logger.info(
+                f"Saved state {saved_time} exceeds the cut off (24 hours)." f" Reverting to use time: {cut_off}"
+            )
+            state[self.LAST_COLLECTION_TIMESTAMP] = cut_off
+
+        return state[self.LAST_COLLECTION_TIMESTAMP]
+
     @staticmethod
     def get_next_page_link(headers: dict) -> str:
+        """
+        Find the next page of results link from the response headers. Header example: `link: <url>; rel="next"`
+        :param headers: response headers from the request to Okta.
+        :return: next page link if available.
+        """
         links = headers.get("link").split(", ")
         next_link = None
         for link in links:
             matched_link = re.match("<(.*?)>", link) if 'rel="next"' in link else None
             next_link = matched_link.group(1) if matched_link else None
         return next_link
 
-    def get_last_collection_timestamp(self, now: str, new_logs: list) -> str:
+    def get_events(self, logs: list, time: str) -> list:
+        """
+        In the collector code we would iterate over all events and drop any that match the 'since' parameter to make
+        sure that we don't double ingest the same event from the previous run (see `get_last_collection_timestamp`).
+        :param logs: response json including all returned events from Okta.
+        :param time: 'since' parameter being used to query Okta.
+        :return: filtered_logs: removed any events that matched the query start time.
+        """
+
+        # If Okta returns only 1 event (no new events occurred) returned in a previous run we want to remove this.
+        if len(logs) == 1 and logs[0].get("published") == time:
+            self.logger.info("No new events found since last execution.")
+            return []
+
+        log = "Returning {filtered} log event(s) from this iteration."
+        pop_index, filtered_logs = 0, logs
+
+        for index, event in enumerate(logs):
+            published = event.get("published")
+            if published and published > time:
+                pop_index = index
+                break
+        if pop_index:
+            filtered_logs = logs[pop_index:]
+            log += f" Removed {pop_index} event log(s) that should have been returned in previous iteration."
+        self.logger.info(log.format(filtered=len(filtered_logs)))
+        return filtered_logs
+
+    def get_last_collection_timestamp(self, new_logs: list, state: dict) -> str:
+        """
+        Mirror the behaviour in collector code to save the TS of the last parsed event as the 'since' time checkpoint.
+        If no new events found then we want to keep the current checkpoint the same.
+        :param new_logs: event logs returned from Okta.
+        :param state: access state dictionary to get the current checkpoint in time if no new logs.
+        :return: new time value to save as the checkpoint to query 'since' on the next run.
+        """
         new_ts = ""
-        # Mirror the behaviour in collector code to save the TS of the last parsed event as the 'since' time checkpoint.
         if new_logs:  # make sure that logs were returned from Okta otherwise will get index error
             new_ts = new_logs[-1].get("published")
+            self.logger.info(f"Saving the last record's published timestamp ({new_ts}) as checkpoint.")
         if not new_ts:
-            self.logger.warn(f'No published record to use as last timestamp, reverting to use "now" ({now})')
-            new_ts = now
+            state_time = state.get(self.LAST_COLLECTION_TIMESTAMP)
+            self.logger.warning(
+                f"No record to use as last timestamp, will not move checkpoint forward. "
+                f"Keeping value of {state_time}"
+            )
+            new_ts = state_time
 
         return new_ts

plugins/okta/plugin.spec.yaml

@@ -13,7 +13,7 @@ sdk:
   version: 5
   user: nobody
 description: Secure identity management and single sign-on to any application
-version: 4.2.0
+version: 4.2.1
 connection_version: 4
 resources:
   source_url: https://github.com/rapid7/insightconnect-plugins/tree/master/plugins/okta

plugins/okta/setup.py

@@ -3,7 +3,7 @@
 
 
 setup(name="okta-rapid7-plugin",
-      version="4.2.0",
+      version="4.2.1",
       description="Secure identity management and single sign-on to any application",
       author="rapid7",
       author_email="",

plugins/okta/unit_test/expected/get_logs.json.exp

@@ -26,7 +26,7 @@
       "outcome": {
         "result": "SUCCESS"
       },
-      "published": "2023-04-27T07:49:21.764Z",
+      "published": "2023-04-27T08:49:21.764Z",
       "securityContext": {
         "asNumber": 123456,
         "asOrg": "test",
@@ -97,7 +97,7 @@
       "outcome": {
         "result": "SUCCESS"
       },
-      "published": "2023-04-27T07:49:21.777Z",
+      "published": "2023-04-27T09:49:21.777Z",
       "securityContext": {
         "asNumber": 12345,
         "asOrg": "test",
@@ -144,7 +144,7 @@
     }
   ],
   "state": {
-    "last_collection_timestamp": "2023-04-27T07:49:21.777Z",
+    "last_collection_timestamp": "2023-04-27T09:49:21.777Z",
     "next_page_link": "https://example.com/nextLink?q=next"
   },
   "has_more_pages": true,

plugins/okta/unit_test/expected/get_logs_empty_resp.json.exp

@@ -0,0 +1,8 @@
+{
+    "logs": [],
+    "state": {
+        "last_collection_timestamp": "2023-04-27T08:33:46.123Z"
+    },
+  "has_more_pages": false,
+  "status_code": 200
+}

plugins/okta/unit_test/expected/get_logs_filtered.json.exp

@@ -0,0 +1,81 @@
+{
+  "logs": [
+    {
+      "actor": {
+      "id": "12345",
+      "type": "User",
+      "alternateId": "user@example.com",
+      "displayName": "User 2"
+    },
+    "client": {
+      "userAgent": {
+        "rawUserAgent": "python-requests/2.26.0",
+        "os": "Unknown",
+        "browser": "UNKNOWN"
+      },
+      "zone": "null",
+      "device": "Unknown",
+      "ipAddress": "198.51.100.1",
+      "geographicalContext": {}
+    },
+    "authenticationContext": {
+      "externalSessionId": "12345"
+    },
+    "displayMessage": "Clear user session",
+    "eventType": "user.session.clear",
+    "outcome": {
+      "result": "SUCCESS"
+    },
+    "published": "2023-04-27T09:49:21.777Z",
+    "securityContext": {
+      "asNumber": 12345,
+      "asOrg": "test",
+      "isp": "test",
+      "domain": "example.com",
+      "isProxy": false
+    },
+    "severity": "INFO",
+    "debugContext": {
+      "debugData": {
+        "requestId": "12345",
+        "dtHash": "11111ecd0ecfb444ee1fcb9687ba8b174a3c8d251ce927e6016b871bc0222222",
+        "requestUri": "/api/v1/users/12345/lifecycle/suspend",
+        "url": "/api/v1/users/12345/lifecycle/suspend?"
+      }
+    },
+    "legacyEventType": "core.user_auth.session_clear",
+    "transaction": {
+      "type": "WEB",
+      "id": "12345",
+      "detail": {
+        "requestApiTokenId": "12345"
+      }
+    },
+    "uuid": "9de5069c-5afe-602b-2ea0-a04b66beb2c0",
+    "version": "0",
+    "request": {
+      "ipChain": [
+        {
+          "ip": "198.51.100.1",
+          "geographicalContext": {},
+          "version": "V4"
+        }
+      ]
+    },
+    "target": [
+      {
+        "id": "12345",
+        "type": "User",
+        "alternateId": "user@example.com",
+        "displayName": "User Test123"
+      }
+    ]
+  }
+  ],
+  "state": {
+    "last_collection_timestamp": "2023-04-27T09:49:21.777Z",
+    "next_page_link": "https://example.com/nextLink?q=next"
+  },
+  "has_more_pages": true,
+  "status_code": 200
+}

plugins/okta/unit_test/expected/get_logs_next_empty.json.exp

@@ -0,0 +1,8 @@
+{
+    "logs": [],
+    "state": {
+        "last_collection_timestamp": "2023-04-27T07:49:21.777Z"
+    },
+  "has_more_pages": false,
+  "status_code": 200
+}

plugins/okta/unit_test/inputs/monitor_logs_next_page.json.inp

@@ -1,4 +1,4 @@
 {
-  "last_collection_timestamp": "2023-04-27T08:34:46",
+  "last_collection_timestamp": "2023-04-27T07:49:21.777Z",
   "next_page_link": "https://example.com/nextLink?q=next"
 }

plugins/okta/unit_test/inputs/monitor_logs_with_state.json.inp

@@ -1,3 +1,3 @@
 {
-  "last_collection_timestamp": "2023-04-27T08:33:46"
+  "last_collection_timestamp": "2023-04-27T08:33:46.123Z"
 }

plugins/okta/unit_test/responses/get_logs.json.resp

@@ -25,7 +25,7 @@
     "outcome": {
       "result": "SUCCESS"
     },
-    "published": "2023-04-27T07:49:21.764Z",
+    "published": "2023-04-27T08:49:21.764Z",
     "securityContext": {
       "asNumber": 123456,
       "asOrg": "test",
@@ -96,7 +96,7 @@
     "outcome": {
       "result": "SUCCESS"
     },
-    "published": "2023-04-27T07:49:21.777Z",
+    "published": "2023-04-27T09:49:21.777Z",
     "securityContext": {
       "asNumber": 12345,
       "asOrg": "test",

plugins/okta/unit_test/responses/get_logs_empty_response.resp

@@ -0,0 +1 @@
+[]