diff --git a/content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.py b/content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.py index f736a62cae..3281232f8f 100644 --- a/content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.py +++ b/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)}" diff --git a/content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.yaml b/content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.yaml index 3e1ec62a0a..c33621dc74 100644 --- a/content/response_integrations/third_party/partner/orca_security/connectors/AlertsConnector.yaml +++ b/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 diff --git a/content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityManager.py b/content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityManager.py index 776fca0cd7..d62a212f1b 100644 --- a/content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityManager.py +++ b/content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityManager.py @@ -113,6 +113,8 @@ 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, @@ -120,7 +122,8 @@ def get_alerts( 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) diff --git a/content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityParser.py b/content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityParser.py index 23a346da84..7559365a9b 100644 --- a/content/response_integrations/third_party/partner/orca_security/core/OrcaSecurityParser.py +++ b/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"), diff --git a/content/response_integrations/third_party/partner/orca_security/core/constants.py b/content/response_integrations/third_party/partner/orca_security/core/constants.py index 643547233e..3d93ef3fa6 100644 --- a/content/response_integrations/third_party/partner/orca_security/core/constants.py +++ b/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 diff --git a/content/response_integrations/third_party/partner/orca_security/core/datamodels.py b/content/response_integrations/third_party/partner/orca_security/core/datamodels.py index 5a14c0522b..b7929728ed 100644 --- a/content/response_integrations/third_party/partner/orca_security/core/datamodels.py +++ b/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}" diff --git a/content/response_integrations/third_party/partner/orca_security/core/query_builder.py b/content/response_integrations/third_party/partner/orca_security/core/query_builder.py index 5e0283790f..036095b822 100644 --- a/content/response_integrations/third_party/partner/orca_security/core/query_builder.py +++ b/content/response_integrations/third_party/partner/orca_security/core/query_builder.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime, timezone from typing import TYPE_CHECKING from soar_sdk.SiemplifyUtils import unix_now @@ -19,6 +20,10 @@ from typing import Any +def _ms_to_iso(timestamp_ms: int) -> str: + return datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc).isoformat() + + class BaseQueryBuilder: def __init__(self) -> None: self.payload: dict[str, Any] = { @@ -81,29 +86,53 @@ def __init__( self, start_timestamp: int | None = None, limit: int = DEFAULT_MAX_LIMIT, + last_sync_start_timestamp: int | None = None, ) -> None: super().__init__() self.payload["limit"] = limit self.payload["query"]["models"] = ["Alert"] - self.payload["order_by[]"] = ["CreatedAt"] + # last_sync is the DB write time - pagination cursor must match the order field + self.payload["order_by[]"] = ["last_sync"] if last_sync_start_timestamp is not None else ["CreatedAt"] if start_timestamp is not None: self.with_created_at_range(start_timestamp) + if last_sync_start_timestamp is not None: + self.with_last_sync_range(last_sync_start_timestamp) + def with_created_at_range(self, start_timestamp: int) -> AlertQueryBuilder: """Create a range filter for CreatedAt field. Args: - start_timestamp (int): The start timestamp to filter by. + start_timestamp (int): The start timestamp in milliseconds to filter by. Returns: AlertQueryBuilder: The instance of the builder. """ + # The "range" operator with ISO values compares full datetimes (inclusive + # on both ends). "date_range" must not be used for watermarking: it rounds + # the range start UP to the next UTC day boundary. self._add_filter( key="CreatedAt", - values=[start_timestamp, unix_now()], + values=[_ms_to_iso(start_timestamp), _ms_to_iso(unix_now())], + type_str="datetime", + operator="range", + ) + return self + + def with_last_sync_range(self, start_timestamp: int) -> AlertQueryBuilder: + """Create a range filter for last_sync field (time the alert row was + written to the database). + Args: + start_timestamp (int): The start timestamp in milliseconds to filter by. + + Returns: + AlertQueryBuilder: The instance of the builder. + """ + self._add_filter( + key="last_sync", + values=[_ms_to_iso(start_timestamp), _ms_to_iso(unix_now())], type_str="datetime", - operator="date_range", - value_type="days", + operator="range", ) return self diff --git a/content/response_integrations/third_party/partner/orca_security/pyproject.toml b/content/response_integrations/third_party/partner/orca_security/pyproject.toml index 22c4dda65d..99b2a9f906 100644 --- a/content/response_integrations/third_party/partner/orca_security/pyproject.toml +++ b/content/response_integrations/third_party/partner/orca_security/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "OrcaSecurity" -version = "15.0" +version = "16.0" description = "Orca Security offers a radical new, agentless approach to cloud security and compliance that provides 100% visibility and coverage of cloud configurations and workloads while eliminating the cost, organizational friction, and performance hits associated with agent-based solutions. In case of any queries, please reach out to support@orca.security." requires-python = ">=3.11,<3.12" dependencies = [ diff --git a/content/response_integrations/third_party/partner/orca_security/release_notes.yaml b/content/response_integrations/third_party/partner/orca_security/release_notes.yaml index 1ae586b61f..566727ec3d 100644 --- a/content/response_integrations/third_party/partner/orca_security/release_notes.yaml +++ b/content/response_integrations/third_party/partner/orca_security/release_notes.yaml @@ -175,3 +175,12 @@ item_type: Integration publish_time: '2026-07-17' ticket_number: '' +- description: Alerts Connector - Fixed intermittently missed alerts. The connector + now tracks its position using the alert's database write time (last_sync) with + precise timestamp filtering and cursor-based pagination, ingests alerts created + within the last 6 hours only, and deduplicates re-fetched alerts by ID. + version: 16.0 + item_name: Orca Security - Alerts Connector + item_type: Connector + publish_time: '2026-08-07' + ticket_number: ''