update orca connector

content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.py

@@ -7,23 +7,21 @@
 from soar_sdk.SiemplifyConnectorsDataModel import AlertInfo
 from soar_sdk.SiemplifyUtils import output_handler, unix_now
 from TIPCommon import (
-    UNIX_FORMAT,
     extract_connector_param,
-    filter_old_alerts,
-    get_last_success_time,
     is_approaching_timeout,
     is_overflowed,
     read_ids,
-    save_timestamp,
     write_ids,
 )
 
 from ..core.constants import (
     BLACKLIST_FILTER,
     CONNECTOR_NAME,
+    CREATED_AT_LOOKBACK_HOURS,
     DEFAULT_LIMIT,
     DEFAULT_TIME_FRAME,
     POSSIBLE_SEVERITIES,
+    STORED_IDS_LIMIT,
     WHITELIST_FILTER,
 )
 from ..core.OrcaSecurityExceptions import OrcaSecurityInvalidParameterException
@@ -191,85 +189,140 @@ def main(is_test_run):
             siemplify_logger=siemplify.LOGGER,
         )
 
+        # The watermark tracks last_sync (DB write time), so alerts that become
+        # visible or eligible after creation (e.g. Orca Score populated later) still
+        # enter the fetch window. The CreatedAt bound keeps updates of alerts older
+        # than the lookback out of the window, preserving "new alerts only" semantics.
+        lookback_ms = CREATED_AT_LOOKBACK_HOURS * 60 * 60 * 1000
+        saved_timestamp = siemplify.fetch_timestamp()
+        if saved_timestamp:
+            # Tolerate downtime up to the CreatedAt lookback - alerts older than
+            # that age out of the fetch window anyway
+            last_sync_cursor = max(saved_timestamp, unix_now() - lookback_ms)
+        else:
+            last_sync_cursor = unix_now() - hours_backwards * 60 * 60 * 1000
+        created_at_start = unix_now() - lookback_ms
+        siemplify.LOGGER.info(f"Fetching alerts from last_sync cursor {last_sync_cursor}")
+
+        existing_ids_set = set(existing_ids)
         fetched_alerts = []
-        alerts = manager.get_alerts(
-            start_timestamp=get_last_success_time(
-                siemplify=siemplify,
-                offset_with_metric={"hours": hours_backwards},
-                time_format=UNIX_FORMAT,
-            ),
-            limit=fetch_limit,
-            lowest_severity=lowest_severity,
-            categories=category_filter,
-            title_filter=siemplify.whitelist,
-            title_filter_type=(BLACKLIST_FILTER if whitelist_as_a_blacklist else WHITELIST_FILTER),
-            alert_types=alert_type_filter,
-            lowest_score=lowest_score,
-        )
-
-        filtered_alerts = filter_old_alerts(siemplify, alerts, existing_ids, "alert_id")
-        siemplify.LOGGER.info(f"Fetched {len(filtered_alerts)} alerts")
-
-        if is_test_run:
-            siemplify.LOGGER.info("This is a TEST run. Only 1 alert will be processed.")
-            filtered_alerts = filtered_alerts[:1]
-
-        for alert in filtered_alerts:
-            try:
-                if is_approaching_timeout(connector_starting_time, script_timeout):
-                    siemplify.LOGGER.info("Timeout is approaching. Connector will gracefully exit")
-                    break
-
-                if len(processed_alerts) >= fetch_limit:
-                    # Provide slicing for the alerts amount.
-                    siemplify.LOGGER.info(
-                        "Reached max number of alerts cycle. No more alerts will be processed in this cycle."
-                    )
-                    break
-
-                siemplify.LOGGER.info(f"Started processing alert {alert.alert_id}")
-                alert.set_events()
-
-                # Update existing alerts
-                existing_ids.append(alert.alert_id)
-                fetched_alerts.append(alert)
-
-                alert_info = alert.get_alert_info(
-                    alert_info=AlertInfo(),
-                    environment_common=GetEnvironmentCommonFactory().create_environment_manager(
-                        siemplify, environment_field_name, environment_regex_pattern
-                    ),
-                    device_product_field=device_product_field,
-                )
+        watermark = 0
+        offset = 0
+        stop_fetching = False
+
+        while not stop_fetching:
+            alerts = manager.get_alerts(
+                start_timestamp=created_at_start,
+                limit=fetch_limit,
+                lowest_severity=lowest_severity,
+                categories=category_filter,
+                title_filter=siemplify.whitelist,
+                title_filter_type=(BLACKLIST_FILTER if whitelist_as_a_blacklist else WHITELIST_FILTER),
+                alert_types=alert_type_filter,
+                lowest_score=lowest_score,
+                last_sync_start_timestamp=last_sync_cursor,
+                start_at_index=offset,
+            )
+            siemplify.LOGGER.info(
+                f"Fetched page of {len(alerts)} alerts from last_sync cursor {last_sync_cursor}, offset {offset}"
+            )
 
-                if is_overflowed(siemplify, alert_info, is_test_run):
-                    siemplify.LOGGER.info(
-                        f"{alert_info.rule_generator}-{alert_info.ticket_id}-{alert_info.environment}"
-                        f"-{alert_info.device_product} found as overflow alert. Skipping..."
+            if is_test_run:
+                siemplify.LOGGER.info("This is a TEST run. Only 1 alert will be processed.")
+                alerts = alerts[:1]
+                stop_fetching = True
+
+            for alert in alerts:
+                try:
+                    if is_approaching_timeout(connector_starting_time, script_timeout):
+                        siemplify.LOGGER.info("Timeout is approaching. Connector will gracefully exit")
+                        stop_fetching = True
+                        break
+
+                    if alert.alert_id in existing_ids_set:
+                        # Already ingested in a previous run - advance the watermark
+                        # over it so the connector makes progress even when a page
+                        # contains only duplicates
+                        watermark = max(watermark, alert.last_sync_ms)
+                        continue
+
+                    if len(processed_alerts) >= fetch_limit:
+                        # Provide slicing for the alerts amount.
+                        siemplify.LOGGER.info(
+                            "Reached max number of alerts cycle. No more alerts will be processed in this cycle."
+                        )
+                        stop_fetching = True
+                        break
+
+                    siemplify.LOGGER.info(f"Started processing alert {alert.alert_id}")
+                    alert.set_events()
+
+                    # Update existing alerts
+                    existing_ids.append(alert.alert_id)
+                    existing_ids_set.add(alert.alert_id)
+                    fetched_alerts.append(alert)
+                    watermark = max(watermark, alert.last_sync_ms)
+
+                    alert_info = alert.get_alert_info(
+                        alert_info=AlertInfo(),
+                        environment_common=GetEnvironmentCommonFactory().create_environment_manager(
+                            siemplify, environment_field_name, environment_regex_pattern
+                        ),
+                        device_product_field=device_product_field,
                     )
-                    # If is overflowed we should skip
-                    continue
-
-                processed_alerts.append(alert_info)
-                siemplify.LOGGER.info(f"Alert {alert.alert_id} was created.")
-
-            except Exception as e:
-                siemplify.LOGGER.error(f"Failed to process alert {alert.alert_id}")
-                siemplify.LOGGER.exception(e)
 
-                if is_test_run:
-                    raise
-
-            siemplify.LOGGER.info(f"Finished processing alert {alert.alert_id}")
+                    if is_overflowed(siemplify, alert_info, is_test_run):
+                        siemplify.LOGGER.info(
+                            f"{alert_info.rule_generator}-{alert_info.ticket_id}-{alert_info.environment}"
+                            f"-{alert_info.device_product} found as overflow alert. Skipping..."
+                        )
+                        # If is overflowed we should skip
+                        continue
+
+                    processed_alerts.append(alert_info)
+                    siemplify.LOGGER.info(f"Alert {alert.alert_id} was created.")
+
+                except Exception as e:
+                    siemplify.LOGGER.error(f"Failed to process alert {alert.alert_id}")
+                    siemplify.LOGGER.exception(e)
+
+                    if is_test_run:
+                        raise
+
+                siemplify.LOGGER.info(f"Finished processing alert {alert.alert_id}")
+
+            if stop_fetching or len(alerts) < fetch_limit:
+                # A short page means there are no more alerts in the window
+                break
+
+            next_cursor = alerts[-1].last_sync_ms
+            if next_cursor <= last_sync_cursor:
+                # A full page within a single last_sync second (second-resolution
+                # ties) - the range start can't move, so page deeper with an offset.
+                # Never skip past the tie: rows beyond this page may be unseen.
+                # Tie ordering has no secondary sort key, so offsets are only
+                # meaningful within this run's back-to-back requests - do not carry
+                # the offset across runs. A row that shuffles out of view returns on
+                # its next last_sync rewrite; dedup absorbs the rest.
+                offset += len(alerts)
+            else:
+                last_sync_cursor = next_cursor
+                offset = 0
 
         if not is_test_run:
             siemplify.LOGGER.info("Saving existing ids.")
-            write_ids(siemplify, existing_ids)
-            save_timestamp(
-                siemplify=siemplify,
-                alerts=fetched_alerts,
-                timestamp_key="created_at_ms",
-            )
+            if len(existing_ids) > STORED_IDS_LIMIT:
+                siemplify.LOGGER.info(
+                    f"Alert ids cache exceeded {STORED_IDS_LIMIT} entries, oldest ids will be evicted. "
+                    f"If this repeats every run, duplicate cases are possible for re-synced alerts."
+                )
+            write_ids(siemplify, existing_ids, stored_ids_limit=STORED_IDS_LIMIT)
+
+            if watermark:
+                siemplify.LOGGER.info(f"Saving last_sync watermark: {watermark}")
+                siemplify.save_timestamp(new_timestamp=watermark)
+            else:
+                siemplify.LOGGER.info("Timestamp is not updated since no alerts were handled")
 
         siemplify.LOGGER.info(
             f"Alerts processed: {len(processed_alerts)} out of {len(fetched_alerts)}"

content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.yaml

@@ -111,9 +111,10 @@ parameters:
     default_value: 1
     type: integer
     description: Number of hours before the first connector iteration to retrieve
-        incidents from. This parameter applies to the initial connector iteration
-        after you enable the connector for the first time, or used as a fallback value
-        in cases where connector's last run timestamp expires.
+        incidents from. This parameter applies only to the initial connector iteration
+        after you enable the connector for the first time. On later iterations the
+        connector resumes from its last run position and tolerates downtime of up
+        to 6 hours.
     is_mandatory: false
     is_advanced: true
     mode: script

content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityManager.py

@@ -113,14 +113,17 @@ def get_alerts(
         title_filter_type: int | None = WHITELIST_FILTER,
         alert_types: list[str] | None = None,
         lowest_score: float | None = None,
+        last_sync_start_timestamp: int | None = None,
+        start_at_index: int = 0,
     ) -> list[Alert]:
         """Retrieve alerts from the API.
         Builds an alert query payload with the provided filters (severity, categories,
         title, type, score, etc.), sends it to the alerts endpoint, validates the
         response, and parses it into a list of Alert objects.
 
         Args:
-            start_timestamp (int): The start timestamp in milliseconds to fetch alerts.
+            start_timestamp (int): The start timestamp in milliseconds to filter
+            alerts by creation time.
             limit (int): The maximum number of alerts to fetch.
             lowest_severity (str): The lowest severity to filter by.
             categories (list[str]): List of categories to filter by.
@@ -129,13 +132,19 @@ def get_alerts(
             or BLACKLIST_FILTER. Defaults to WHITELIST_FILTER.
             alert_types (list[str]): List of alert types to filter by.
             lowest_score (float): The lowest score to filter by.
+            last_sync_start_timestamp (int): The start timestamp in milliseconds to
+            filter alerts by DB write time. When provided, results are ordered by
+            last_sync so it can be used as a pagination cursor.
+            start_at_index (int): The offset to start fetching results from. Used to
+            page within results sharing one last_sync value.
 
         Returns:
             list[Alert]: List of Alert objects.
         """
         url: str = self._get_full_url("get_alerts")
         payload: AlertQueryBuilder = (
-            AlertQueryBuilder(start_timestamp, limit)
+            AlertQueryBuilder(start_timestamp, limit, last_sync_start_timestamp)
+            .start_at_index(start_at_index)
             .with_severity(lowest_severity)
             .with_categories(categories)
             .with_title_filter(title_filter, title_filter_type)

content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityParser.py

@@ -35,6 +35,7 @@ def build_alert_object(raw_data):
             details=alert_data.get("Details", {}).get("value"),
             severity=alert_data.get("Severity", {}).get("value"),
             created_at=alert_data.get("CreatedAt", {}).get("value"),
+            last_sync=alert_data.get("last_sync", {}).get("value"),
             asset_name=asset_data.get("asset_name"),
             asset_type=asset_data.get("asset_type"),
             type_string=alert_data.get("AlertType", {}).get("value"),

content/response_integrations/third_party/partner/orca_security/core/constants.py

@@ -32,6 +32,12 @@
 CONNECTOR_NAME = f"{INTEGRATION_DISPLAY_NAME} - Alerts Connector"
 DEFAULT_TIME_FRAME = 1
 DEFAULT_LIMIT = 100
+# How far back in creation time an alert can still be ingested. Alerts can become
+# eligible for fetching after creation (e.g. Orca Score populated later), but
+# alerts older than this are treated as updates and never re-ingested.
+CREATED_AT_LOOKBACK_HOURS = 6
+# Dedup ID cache size - must cover all alerts fetched within the lookback window
+STORED_IDS_LIMIT = 5000
 DEFAULT_ASSET_LIMIT: int = 20
 DEFAULT_RESULTS_LIMIT: int = 1000
 DEFAULT_OFFSET: int = 0

content/response_integrations/third_party/partner/orca_security/core/datamodels.py

@@ -49,6 +49,7 @@ def __init__(
         asset_name,
         asset_type,
         type_string,
+        last_sync=None,
     ):
         super(Alert, self).__init__(raw_data)
         self.flat_raw_data = dict_to_flat(raw_data)
@@ -58,6 +59,10 @@ def __init__(
         self.severity = severity
         self.created_at = created_at
         self.created_at_ms = convert_string_to_unix_time(self.created_at)
+        self.last_sync = last_sync
+        # Fall back to creation time so the connector can still save a timestamp
+        # if the API response is missing last_sync
+        self.last_sync_ms = convert_string_to_unix_time(self.last_sync) if self.last_sync else self.created_at_ms
         self.type_string = type_string
         self.asset_name = asset_name or f"{self.title}-{self.alert_id}"
         self.asset_type = asset_type or f"{self.type_string}-{self.alert_id}"