[SOAR-17745] Duo admin Improve handling of 429s in task code (#3048)

* Updating plugin.spec (#3047)

* Duo admin Improve handling of 429s in task code

* Linting the code

* Version bump

* Adjusting styling

* Resolving static code analysis

* Resolving plugin's validation

* CR fixes

* Additional test case

* Bumping the SDK and linting the code

---------

Co-authored-by: rmurray-r7 <[email protected]>

plugins/duo_admin/.CHECKSUM

@@ -1,7 +1,7 @@
 {
-	"spec": "4a94a77c37f17820e8768bb2850f6ee5",
-	"manifest": "672f0da4df4edb87ab669d69e435c5c7",
-	"setup": "8a8919e13bd1afe4849427d3dae6dbf4",
+	"spec": "814e81dd40fcf4d884c984e0c451201e",
+	"manifest": "b4029998fe0d9bcc85c8016d009d19b3",
+	"setup": "5965f3fd331d7855550e5afad6ea8956",
 	"schemas": [
 		{
 			"identifier": "add_user/schema.py",

plugins/duo_admin/Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=linux/amd64 rapid7/insightconnect-python-3-plugin:6.2.2
+FROM --platform=linux/amd64 rapid7/insightconnect-python-3-plugin:6.2.3
 
 LABEL organization=rapid7
 LABEL sdk=python

plugins/duo_admin/bin/komand_duo_admin

@@ -6,7 +6,7 @@ from sys import argv
 
 Name = "Duo Admin API"
 Vendor = "rapid7"
-Version = "5.0.2"
+Version = "5.0.3"
 Description = "[Duo](https://duo.com/)'s Trusted Access platform verifies the identity of your users with two-factor authentication and security health of their devices before they connect to the apps they use. Using the Duo plugin for InsightConnect will allow Duo user management within automation workflows"
 
 

plugins/duo_admin/help.md

@@ -950,6 +950,7 @@ Example output:
 
 # Version History
 
+* 5.0.3 - Bump the SDK to version 6.2.3 | Update Task `monitor_logs` to delay retry if a rate limit error is returned from Duo Admin
 * 5.0.2 - Updated SDK to the latest version (v6.2.2) | Address vulnerabilities
 * 5.0.1 - Update to enable Plugin as FedRAMP ready | Update SDK (`6.1.2`)
 * 5.0.0 - Updated to include latest SDK v5.5.5 | Removing Unused fields from User Object

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

@@ -1,21 +1,25 @@
-from typing import Dict
+from datetime import datetime, timedelta, timezone
+from time import time
+from typing import Any, Dict, Tuple, Union
 
 import insightconnect_plugin_runtime
 from insightconnect_plugin_runtime.exceptions import PluginException
-
-from .schema import MonitorLogsInput, MonitorLogsOutput, MonitorLogsState, Component, Input
+from insightconnect_plugin_runtime.helper import hash_sha1
 
 # Custom imports below
+from komand_duo_admin.util.constants import Assistance
 from komand_duo_admin.util.exceptions import ApiException
-from datetime import datetime, timedelta, timezone
-from hashlib import sha1
+from komand_duo_admin.util.util import Utils
+
+from .schema import MonitorLogsInput, MonitorLogsOutput, MonitorLogsState, Component, Input
 
 ADMIN_LOGS_LOG_TYPE = "Admin logs"
 AUTH_LOGS_LOG_TYPE = "Auth logs"
 TRUST_MONITOR_EVENTS_LOG_TYPE = "Trust monitor events"
 INITIAL_CUTOFF_HOURS = 24
 MAX_CUTOFF_HOURS = 168
 API_CUTOFF_HOURS = 4320
+RATE_LIMIT_DELAY = 600
 
 
 class MonitorLogs(insightconnect_plugin_runtime.Task):
@@ -30,6 +34,7 @@ class MonitorLogs(insightconnect_plugin_runtime.Task):
     PREVIOUS_ADMIN_LOG_HASHES = "previous_admin_log_hashes"
     PREVIOUS_AUTH_LOG_HASHES = "previous_auth_log_hashes"
     PREVIOUS_TRUST_MONITOR_EVENT_HASHES = "previous_trust_monitor_event_hashes"
+    RATE_LIMIT_DATETIME = "rate_limit_datetime"
 
     def __init__(self):
         super(self.__class__, self).__init__(
@@ -103,7 +108,46 @@ def get_parameters_for_query(
         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
 
+    def check_rate_limit(self, state: Dict) -> Union[PluginException, None]:
+        rate_limited = state.get(self.RATE_LIMIT_DATETIME)
+        now = time()
+        if rate_limited:
+            rate_limit_string = Utils.convert_epoch_to_readable(rate_limited)
+            log_msg = f"Rate limit value stored in state: {rate_limit_string}. "
+            if rate_limited > now:
+                log_msg += "Still within rate limiting period, skipping task execution..."
+                self.logger.info(log_msg)
+                error = PluginException(
+                    cause=PluginException.causes.get(PluginException.Preset.RATE_LIMIT),
+                    assistance=Assistance.RATE_LIMIT,
+                )
+                return error
+
+            log_msg += "However no longer in rate limiting period, so task can be executed..."
+            del state[self.RATE_LIMIT_DATETIME]
+            self.logger.info(log_msg)
+
+    def check_rate_limit_error(
+        self, error: ApiException, status_code: int, state: dict, rate_limit_delay: int
+    ) -> Tuple[int, Any]:
+        if status_code == 429:
+            new_run_time = time() + rate_limit_delay  # default to wait 10 minutes before the next run
+            try:
+                new_run_time_string = Utils.convert_epoch_to_readable(new_run_time)
+                self.logger.error(f"A rate limit error has occurred, task will resume after {new_run_time_string}")
+                state[self.RATE_LIMIT_DATETIME] = new_run_time
+            except Exception as err:
+                self.logger.error(
+                    f"Unable to calculate new run time, no rate limiting applied to the state. Error: {repr(err)}",
+                    exc_info=True,
+                )
+            return 200, None
+        return status_code, error
+
     def run(self, params={}, state={}, custom_config={}):  # noqa: C901
+        rate_limit_delay = custom_config.get("rate_limit_delay", RATE_LIMIT_DELAY)
+        if rate_limited := self.check_rate_limit(state):
+            return [], state, False, 429, rate_limited
         self.connection.admin_api.toggle_rate_limiting = False
         has_more_pages = False
         backward_comp_first_run = False
@@ -257,7 +301,8 @@ def run(self, params={}, state={}, custom_config={}):  # noqa: C901
                 state[self.PREVIOUS_TRUST_MONITOR_EVENT_HASHES] = []
                 state[self.PREVIOUS_ADMIN_LOG_HASHES] = []
                 state[self.PREVIOUS_AUTH_LOG_HASHES] = []
-                return [], state, False, error.status_code, error
+                status_code, error = self.check_rate_limit_error(error, error.status_code, state, rate_limit_delay)
+                return [], state, False, status_code, error
         except Exception as error:
             self.logger.info(f"An Exception has been raised. Error: {error}")
             state[self.PREVIOUS_TRUST_MONITOR_EVENT_HASHES] = []
@@ -283,18 +328,11 @@ def add_log_type_field(logs: list, value: str) -> list:
             log["log_type"] = value
         return logs
 
-    @staticmethod
-    def sha1(log: dict) -> str:
-        hash_ = sha1()  # nosec B303
-        for key, value in log.items():
-            hash_.update(f"{key}{value}".encode("utf-8"))
-        return hash_.hexdigest()
-
     def compare_hashes(self, previous_logs_hashes: list, new_logs: list):
         new_logs_hashes = []
         logs_to_return = []
         for log in new_logs:
-            hash_ = self.sha1(log)
+            hash_ = hash_sha1(log)
             if hash_ not in previous_logs_hashes:
                 new_logs_hashes.append(hash_)
                 logs_to_return.append(log)

plugins/duo_admin/komand_duo_admin/util/constants.py

@@ -19,6 +19,7 @@ class Assistance:
     VERIFY_INPUT = (
         "Verify your input is correct and not malformed and try again. If the issue persists, please contact support."
     )
+    RATE_LIMIT = "Task will resume collection of logs after the rate limiting period has expired."
 
 
 class PossibleInputs:

plugins/duo_admin/komand_duo_admin/util/util.py

@@ -0,0 +1,7 @@
+from datetime import datetime
+
+
+class Utils:
+    @staticmethod
+    def convert_epoch_to_readable(epoch_time: float) -> str:
+        return datetime.utcfromtimestamp(epoch_time).strftime("%Y-%m-%d %H:%M:%S")

plugins/duo_admin/plugin.spec.yaml

@@ -11,7 +11,7 @@ status: []
 supported_versions: ["Duo Admin API 2024-09-17"]
 sdk:
   type: full
-  version: 6.2.2
+  version: 6.2.3
   user: nobody
 description: "[Duo](https://duo.com/)'s Trusted Access platform verifies the identity of your users with two-factor authentication and
 security health of their devices before they connect to the apps they use. Using the Duo plugin for InsightConnect will allow Duo user management within automation workflows"
@@ -29,7 +29,7 @@ key_features:
 requirements:
 - "Two secret keys - `integration key` and `secret key`"
 - "`API hostname`"
-version: 5.0.2
+version: 5.0.3
 connection_version: 4
 resources:
   source_url: https://github.com/rapid7/insightconnect-plugins/tree/master/plugins/duo_admin
@@ -50,6 +50,7 @@ references:
 troubleshooting:
   - "Many actions in this plugin take a User ID as input. A User ID is not the username - instead it's a unique identifier e.g. DU9I6T0F7R2S1J4XZHHA. A User ID can be obtained by passing a username to the Get User Status action."
 version_history:
+- "5.0.3 - Bump the SDK to version 6.2.3 | Update Task `monitor_logs` to delay retry if a rate limit error is returned from Duo Admin"
 - "5.0.2 - Updated SDK to the latest version (v6.2.2) | Address vulnerabilities"
 - "5.0.1 - Update to enable Plugin as FedRAMP ready | Update SDK (`6.1.2`)"
 - "5.0.0 - Updated to include latest SDK v5.5.5 | Removing Unused fields from User Object"

plugins/duo_admin/setup.py

@@ -3,7 +3,7 @@
 
 
 setup(name="duo_admin-rapid7-plugin",
-      version="5.0.2",
+      version="5.0.3",
       description="[Duo](https://duo.com/)'s Trusted Access platform verifies the identity of your users with two-factor authentication and security health of their devices before they connect to the apps they use. Using the Duo plugin for InsightConnect will allow Duo user management within automation workflows",
       author="rapid7",
       author_email="",

plugins/duo_admin/unit_test/expected/monitor_logs_rate_limit.json.exp

@@ -0,0 +1,10 @@
+{
+  "logs": [],
+  "state": {
+    "previous_admin_log_hashes": [],
+    "previous_auth_log_hashes": [],
+    "rate_limit_datetime": 3050853836.506636,
+    "previous_trust_monitor_event_hashes": []
+},
+  "status_code": 429
+}

plugins/duo_admin/unit_test/inputs/monitor_logs_with_rate_limit.json.inp

@@ -0,0 +1,6 @@
+{
+  "previous_admin_log_hashes": [],
+  "previous_auth_log_hashes": [],
+  "rate_limit_datetime": 3050853836.506636,
+  "previous_trust_monitor_event_hashes": []
+}

plugins/duo_admin/unit_test/test_monitor_logs.py

@@ -1,5 +1,6 @@
 import sys
 import os
+from time import time
 
 sys.path.append(os.path.abspath("../"))
 
@@ -21,7 +22,7 @@
 class TestMonitorLogs(TestCase):
     @classmethod
     def setUpClass(cls) -> None:
-        cls.action = Util.default_connector(MonitorLogs())
+        cls.task = Util.default_connector(MonitorLogs())
         cls.custom_config = {"cutoff": {"date": "2023-04-30T08:34:46.000Z"}, "lookback": "2023-04-30T08:34:46.000Z"}
 
     @parameterized.expand(
@@ -37,6 +38,12 @@ def setUpClass(cls) -> None:
                     "lookback": "2023-04-30T08:34:46.000Z",
                 },
             ],
+            [
+                "with_rate_limit",
+                Util.read_file_to_dict("inputs/monitor_logs_with_rate_limit.json.inp"),
+                Util.read_file_to_dict("expected/monitor_logs_rate_limit.json.exp"),
+                {},
+            ],
             [
                 "with_state",
                 Util.read_file_to_dict("inputs/monitor_logs_with_state.json.inp"),
@@ -94,9 +101,26 @@ def test_monitor_logs(
         expected,
         config,
     ):
-        actual, actual_state, has_more_pages, status_code, _ = self.action.run(
-            state=current_state, custom_config=config
-        )
+        actual, actual_state, has_more_pages, status_code, _ = self.task.run(state=current_state, custom_config=config)
         self.assertEqual(actual, expected.get("logs"))
         self.assertEqual(actual_state, expected.get("state"))
         self.assertEqual(status_code, expected.get("status_code"))
+
+    def test_monitor_logs_with_rate_limit_whole_flow(
+        self, mock_request, mock_request_instance, mock_get_headers, mock_get_time
+    ):
+        future_time_state = {"rate_limit_datetime": time() + 600}
+        passed_time_state = {"rate_limit_datetime": time() - 600}
+
+        actual, new_state, has_more_pages, status_code, _ = self.task.run(state=future_time_state, custom_config={})
+
+        self.assertEqual(actual, [])
+        self.assertEqual(future_time_state, new_state)
+        self.assertEqual(has_more_pages, False)
+        self.assertEqual(status_code, 429)
+
+        actual_2, new_state_2, _, status_code_2, _ = self.task.run(state=passed_time_state, custom_config={})
+
+        self.assertTrue(actual_2)
+        self.assertTrue(new_state_2)
+        self.assertEqual(status_code_2, 200)