Abnormal Security Data Connector Pagination Support & Modifications in Defaults

Solutions/AbnormalSecurity/Data Connectors/SentinelFunctionsOrchestrator/soar_connector_async.py

@@ -48,7 +48,7 @@ def _get_header(self):
         return {
             "Authorization": f"Bearer {self.api_key}",
             "Soar-Integration-Origin": "AZURE SENTINEL",
-            "Azure-Sentinel-Version": "2024-10-03"
+            "Azure-Sentinel-Version": "2024-11-29"
         }
 
     def _get_filter_query(self, filter_param, gte_datetime=None, lte_datetime=None):

Solutions/AbnormalSecurity/Data Connectors/SentinelFunctionsOrchestrator/soar_connector_async_v2.py

@@ -36,7 +36,7 @@ def get_headers(ctx: Context) -> Dict[str, str]:
         "X-Abnormal-Trace-Id": str(ctx.TRACE_ID),
         "Authorization": f"Bearer {ctx.API_TOKEN}",
         "Soar-Integration-Origin": "AZURE SENTINEL",
-        "Azure-Sentinel-Version": "2024-10-03 V2",
+        "Azure-Sentinel-Version": "2024-11-29 V2",
     }
 
 
@@ -97,9 +97,9 @@ async def call_threat_campaigns_endpoint(
 
         threat_campaigns = set()
 
-        nextPageNumber = 1
-        while nextPageNumber:
-            params["pageNumber"] = nextPageNumber
+        pageNumber = 1
+        while pageNumber:
+            params["pageNumber"] = pageNumber
             endpoint = compute_url(ctx.BASE_URL, "/v1/threats", params)
             headers = get_headers(ctx)
 
@@ -112,9 +112,10 @@ async def call_threat_campaigns_endpoint(
             )
 
             nextPageNumber = response.get("nextPageNumber")
-            assert nextPageNumber is None or nextPageNumber > 0
+            assert nextPageNumber is None or nextPageNumber == pageNumber + 1
+            pageNumber = nextPageNumber
 
-            if nextPageNumber is None or nextPageNumber > ctx.MAX_PAGE_NUMBER:
+            if pageNumber is None or pageNumber > ctx.MAX_PAGE_NUMBER:
                 break
 
         return list(threat_campaigns)
@@ -130,9 +131,9 @@ async def call_cases_endpoint(
 
         case_ids = set()
 
-        nextPageNumber = 1
-        while nextPageNumber:
-            params["pageNumber"] = nextPageNumber
+        pageNumber = 1
+        while pageNumber:
+            params["pageNumber"] = pageNumber
             endpoint = compute_url(ctx.BASE_URL, "/v1/cases", params)
             headers = get_headers(ctx)
 
@@ -143,9 +144,10 @@ async def call_cases_endpoint(
             case_ids.update([case["caseId"] for case in response.get("cases", [])])
 
             nextPageNumber = response.get("nextPageNumber")
-            assert nextPageNumber is None or nextPageNumber > 0
+            assert nextPageNumber is None or nextPageNumber == pageNumber + 1
+            pageNumber = nextPageNumber
 
-            if nextPageNumber is None or nextPageNumber > ctx.MAX_PAGE_NUMBER:
+            if pageNumber is None or pageNumber > ctx.MAX_PAGE_NUMBER:
                 break
 
         return list(case_ids)
@@ -155,27 +157,43 @@ async def call_single_threat_endpoint(
     ctx: Context, threat_id: str, semaphore: asyncio.Semaphore
 ) -> List[str]:
     async with semaphore:
-        endpoint = compute_url(ctx.BASE_URL, f"/v1/threats/{threat_id}", params={})
-        headers = get_headers(ctx)
+        filtered_messages = []
 
-        response = await fetch_with_retries(url=endpoint, headers=headers)
+        pageNumber = 1
+        params = {"pageSize": ctx.SINGLE_THREAT_PAGE_SIZE}
+        while pageNumber:
+            params["pageNumber"] = pageNumber
+            print("Single Threat Params:", params)
+            endpoint = compute_url(ctx.BASE_URL, f"/v1/threats/{threat_id}", params=params)
+            headers = get_headers(ctx)
 
-        filtered_messages = []
-        for message in response["messages"]:
-            message_id = message["abxMessageId"]
-            remediation_time_str = message["remediationTimestamp"]
-
-            remediation_time = try_str_to_datetime(remediation_time_str)
-            if (
-                remediation_time >= ctx.CLIENT_FILTER_TIME_RANGE.start
-                and remediation_time < ctx.CLIENT_FILTER_TIME_RANGE.end
-            ):
-                filtered_messages.append(json.dumps(message, sort_keys=True))
-                logging.info(f"Successfully processed v2 threat message: {message_id}")
-            else:
-                logging.warning(f"Skipped processing v2 threat message: {message_id}")
-
-        return filtered_messages
+            response = await fetch_with_retries(url=endpoint, headers=headers)
+
+            for message in response["messages"]:
+                message_id = message["abxMessageId"]
+                remediation_time_str = message["remediationTimestamp"]
+
+                remediation_time = try_str_to_datetime(remediation_time_str)
+                if (
+                    remediation_time >= ctx.CLIENT_FILTER_TIME_RANGE.start
+                    and remediation_time < ctx.CLIENT_FILTER_TIME_RANGE.end
+                ):
+                    filtered_messages.append(json.dumps(message, sort_keys=True))
+                    logging.info(f"Successfully processed v2 threat message: {message_id}")
+                elif remediation_time < ctx.CLIENT_FILTER_TIME_RANGE.start:
+                    logging.info(f"Skipping further messages as remediationTime {remediation_time} of {message_id} < {ctx.CLIENT_FILTER_TIME_RANGE.start}")
+                    return list(set(filtered_messages))
+                else:
+                    logging.warning(f"Skipped processing v2 threat message: {message_id}")
+
+            nextPageNumber = response.get("nextPageNumber")
+            assert nextPageNumber is None or nextPageNumber == pageNumber + 1
+            pageNumber = nextPageNumber
+
+            if pageNumber is None or pageNumber > ctx.MAX_PAGE_NUMBER:
+                break
+
+        return list(set(filtered_messages))
 
 
 async def call_single_case_endpoint(

Solutions/AbnormalSecurity/Data Connectors/SentinelFunctionsOrchestrator/utils.py

@@ -55,6 +55,7 @@ class Context(BaseModel):
     LIMIT: timedelta
     NUM_CONCURRENCY: int
     MAX_PAGE_NUMBER: int
+    SINGLE_THREAT_PAGE_SIZE: int
     BASE_URL: str
     API_TOKEN: str
     TIME_RANGE: TimeRange
@@ -140,15 +141,16 @@ def get_context(stored_date_time: str) -> Context:
     BASE_URL = os.environ.get("API_HOST", "https://api.abnormalplatform.com/v1")
     API_TOKEN = os.environ["ABNORMAL_SECURITY_REST_API_TOKEN"]
     OUTAGE_TIME = timedelta(
-        minutes=int(os.environ.get("ABNORMAL_OUTAGE_TIME_MIN", "15"))
+        minutes=int(os.environ.get("ABNORMAL_OUTAGE_TIME_MIN", "45"))
     )
     LAG_ON_BACKEND = timedelta(
         seconds=int(os.environ.get("ABNORMAL_LAG_ON_BACKEND_SEC", "30"))
     )
     FREQUENCY = timedelta(minutes=int(os.environ.get("ABNORMAL_FREQUENCY_MIN", "5")))
     LIMIT = timedelta(minutes=int(os.environ.get("ABNORMAL_LIMIT_MIN", "6")))
-    NUM_CONCURRENCY = int(os.environ.get("ABNORMAL_NUM_CONCURRENCY", "5"))
-    MAX_PAGE_NUMBER = int(os.environ.get("ABNORMAL_MAX_PAGE_NUMBER", "3"))
+    NUM_CONCURRENCY = int(os.environ.get("ABNORMAL_NUM_CONCURRENCY", "2"))
+    MAX_PAGE_NUMBER = int(os.environ.get("ABNORMAL_MAX_PAGE_NUMBER", "6"))
+    SINGLE_THREAT_PAGE_SIZE = int(os.environ.get("ABNORMAL_SINGLE_THREAT_PAGE_SIZE", "40"))
 
     STORED_TIME = try_str_to_datetime(stored_date_time)
     CURRENT_TIME = try_str_to_datetime(datetime.now().strftime(TIME_FORMAT))
@@ -171,7 +173,8 @@ def get_context(stored_date_time: str) -> Context:
         CURRENT_TIME=CURRENT_TIME,
         LIMIT=LIMIT,
         TRACE_ID=uuid4(),
-        PYTHON_VERSION=sys.version
+        PYTHON_VERSION=sys.version,
+        SINGLE_THREAT_PAGE_SIZE=SINGLE_THREAT_PAGE_SIZE
     )