import demistomock as demisto import urllib3 from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import from CommonServerUserPython import * # noqa # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # ISO8601 format with UTC, default in XSOAR VENDOR = "orca" PRODUCT = "security" DEFAULT_PAGE_SIZE = 100 # Serving Layer API default page size (SQL LIMIT) TEST_MODULE_LIMIT = 1 # Minimal page size for the connectivity test DEFAULT_GET_EVENTS_LIMIT = 1000 # Fallback cap for the manual get-events command DEFAULT_GET_EVENTS_START = "1 minute" # Fallback lower bound for the manual get-events command """ CLIENT CLASS """ class Client(BaseClient): def __init__(self, server_url: str, headers: dict, proxy: bool = False, verify: bool = False): super().__init__(base_url=server_url, verify=verify, proxy=proxy, headers=headers) def get_alerts_request(self, page_size: int, start_index: int, last_fetch: str, end_time: str) -> dict: """Retrieve a single page of alerts using the Serving Layer API. The Serving Layer API uses offset-based pagination: ``start_at_index`` is a plain 0-based SQL OFFSET (not a cursor/token) and ``limit`` maps to SQL LIMIT. The caller is responsible for advancing ``start_index`` between pages and detecting the end of results (a page returning fewer rows than ``page_size`` or an empty ``data`` array). Events are filtered and ordered by ``LastUpdated`` (the sync timestamp), which is monotonic under Orca's eventual consistency - unlike ``CreatedAt``. The window is two-sided: ``LastUpdated >= last_fetch AND LastUpdated < end_time``. The ``end_time`` upper bound is fixed at the start of the run so events synced mid-run are not half-fetched (they are picked up whole on the next run). ``gte``/``lt`` operators are used without ``value_type`` for second-level precision (per Orca guidance). Args: page_size: int - Maximum number of records to return in this page (SQL LIMIT). start_index: int - 0-based offset for this page (SQL OFFSET). last_fetch: str - Inclusive lower bound for ``LastUpdated`` (ISO 8601). **MUST be a valid ISO 8601 string (e.g., "2023-10-27T10:00:00Z").** The API is very strict about this format. end_time: str - Exclusive upper bound for ``LastUpdated`` (ISO 8601), fixed at run start. Returns: A dictionary with the alerts details (keys: ``status``, ``data``, ``from_cache``). """ payload = { "query": { "models": ["Alert"], "type": "object_set", "with": { "type": "operation", "operator": "and", "values": [ { "key": "LastUpdated", "values": [last_fetch], "type": "datetime", "operator": "gte", }, { "key": "LastUpdated", "values": [end_time], "type": "datetime", "operator": "lt", }, ], }, }, "limit": page_size, "start_at_index": start_index, "order_by[]": ["LastUpdated"], "select": [ "AlertId", "AlertType", "OrcaScore", "RiskLevel", "RuleSource", "ScoreVector", "Category", "Inventory.Name", "CloudAccount.Name", "CloudAccount.CloudProvider", "Source", "Status", "CreatedAt", "LastUpdated", "LastSeen", "Labels", ], } demisto.debug(f"[Client] get_alerts (Serving Layer API) request payload: {json.dumps(payload)}") return self._http_request(method="POST", url_suffix="/serving-layer/query", json_data=payload) """ HELPER FUNCTIONS """ def safe_parse_datetime(raw_value: Optional[str], field_name: str = "timestamp", context: str = "") -> Optional[datetime]: """Parse ``raw_value`` with ``arg_to_datetime``, returning ``None`` if missing/unparseable. ``arg_to_datetime`` raises on unparseable input; this wrapper catches that and logs the full traceback so callers never crash on a single bad timestamp. """ if not raw_value: return None try: return arg_to_datetime(arg=raw_value) except Exception as e: demisto.debug( f"[safe_parse_datetime] arg_to_datetime failed parsing '{field_name}'{f' for {context}' if context else ''} " f"with value '{raw_value}'. Error: {e}\n{traceback.format_exc()}" ) return None def add_time_key_to_alerts(alerts: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Adds the _time key to the alerts with improved logging and clarity. This function mutates the 'alerts' list in place and also returns it. ``_time`` is set from the alert's ``LastUpdated`` (the sync timestamp), so updated alerts re-land in XSIAM at their update time. If ``LastUpdated`` is missing/unparseable, we fall back to ``CreatedAt``, and finally to the current time. Args: alerts: A list of alert dictionaries to process. Returns: The mutated list of alerts, now including the '_time' key. """ now_utc = datetime.now(timezone.utc) fallback_time_str = now_utc.strftime(DATE_FORMAT) if not alerts: return alerts for alert in alerts: alert_data = alert.get("data", {}) alert_id = alert_data.get("AlertId", {}).get("value") or alert.get("id", "UNKNOWN_ID") last_updated_str = alert_data.get("LastUpdated", {}).get("value") create_time_str = alert_data.get("CreatedAt", {}).get("value") # Prefer LastUpdated (sync time); fall back to CreatedAt if it is missing/unparseable. event_time = None for field_name, raw_value in (("LastUpdated", last_updated_str), ("CreatedAt", create_time_str)): event_time = safe_parse_datetime(raw_value, field_name, f"AlertId: {alert_id}") if event_time: break if event_time: alert["_time"] = event_time.strftime(DATE_FORMAT) else: demisto.info( f"[add_time_key_to_alerts] Could not parse or find 'LastUpdated'/'CreatedAt' value for AlertId: {alert_id}. " f"Raw values were LastUpdated='{last_updated_str}', CreatedAt='{create_time_str}'. " f"Setting '_time' to {fallback_time_str}." ) alert["_time"] = fallback_time_str demisto.debug(f"[add_time_key_to_alerts] Processed AlertId: {alert_id}, final _time: {alert.get('_time')}") return alerts def add_entry_status_to_alerts(alerts: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Adds the ``_ENTRY_STATUS`` key to each alert, classifying whether it is new or updated. This function mutates the 'alerts' list in place and also returns it. Logic: - "new" when ``LastUpdated`` == ``CreatedAt`` (never modified since creation), or when either timestamp is missing/unparseable (safe default). - "updated" when ``LastUpdated`` > ``CreatedAt`` (modified after creation). Args: alerts: A list of alert dictionaries to process. Returns: The mutated list of alerts, now including the '_ENTRY_STATUS' key. """ if not alerts: return alerts for alert in alerts: alert_data = alert.get("data", {}) alert_id = alert_data.get("AlertId", {}).get("value") or alert.get("id", "UNKNOWN_ID") created_time = safe_parse_datetime(alert_data.get("CreatedAt", {}).get("value"), "CreatedAt", f"AlertId: {alert_id}") last_updated_time = safe_parse_datetime( alert_data.get("LastUpdated", {}).get("value"), "LastUpdated", f"AlertId: {alert_id}" ) if created_time and last_updated_time and last_updated_time > created_time: alert["_ENTRY_STATUS"] = "updated" else: alert["_ENTRY_STATUS"] = "new" demisto.debug(f"[add_entry_status_to_alerts] Processed AlertId: {alert_id}, _ENTRY_STATUS: {alert.get('_ENTRY_STATUS')}") return alerts """ COMMAND FUNCTIONS """ def orca_test_module(client: Client, last_fetch: str, end_time: str) -> str: """Test the connection to Orca Security. Args: client: client - An Orca client. last_fetch: str - The time and date of the last fetch alert. end_time: str - The exclusive upper bound for the fetch window (run start). Returns: 'ok' if the connection was successful, else throws exception. """ try: client.get_alerts_request(TEST_MODULE_LIMIT, 0, last_fetch, end_time) return "ok" except DemistoException as e: if "Error in API call [404] - Not Found" in e.message: raise Exception('Error in API call [404] - Not Found\n{"error": "URL is invalid"}') else: raise Exception(e.message) def get_alerts( client: Client, max_fetch: int, last_fetch: str, end_time: str, start_offset: int = 0, page_size: int = DEFAULT_PAGE_SIZE, ) -> list: """Retrieve alerts, paginating through all available pages for the given window. The Serving Layer API is offset-based: there is no next-page token. We start at ``start_at_index`` = ``start_offset`` and, after each page, advance the offset by the number of rows actually returned (not blindly by ``page_size``). Pagination stops when a page returns fewer rows than ``page_size`` (or an empty ``data`` array) - the normal end-of-results signal - or when we reach ``max_fetch`` total events for this run. ``start_offset`` lets the caller resume mid-second: when a single ``LastUpdated`` second holds more events than ``max_fetch``, one run cannot consume the whole second, so the cursor timestamp cannot advance. To avoid wedging the collector on that second WITHOUT exceeding ``max_fetch``, the caller persists how many boundary-second events were already consumed (``lastRunOffset``) and passes it here so the next run continues where the previous one stopped. The fetch window is two-sided on ``LastUpdated``: ``>= last_fetch AND < end_time``. Args: client: client - An Orca client. max_fetch: int - The maximum number of events to collect in this run (across all pages). last_fetch: str - Inclusive lower bound for ``LastUpdated``. end_time: str - Exclusive upper bound for ``LastUpdated`` (fixed at run start). start_offset: int - SQL OFFSET to start from (resume point within the boundary second). page_size: int - The maximum number of events to request per page. Returns: A list of alerts (at most ``max_fetch`` items). """ alerts: list = [] start_index = start_offset pages_fetched = 0 while len(alerts) < max_fetch: # Don't request more than we still need for this run. current_page_size = min(page_size, max_fetch - len(alerts)) response = client.get_alerts_request(current_page_size, start_index, last_fetch, end_time) page = response.get("data", []) demisto.debug(f"[get_alerts] page: {start_index=} , {current_page_size=} , returned={len(page)}") if not page: # Empty page - no more results. break pages_fetched += 1 alerts.extend(page) # Advance the SQL OFFSET by the number of rows actually returned. start_index += len(page) if len(page) < current_page_size: # Short page - this was the last page of results. break # Summary line: confirms pagination is looping (vs the old single-page behavior). demisto.info(f"[get_alerts] summary: fetched {len(alerts)} event(s) across {pages_fetched} page(s) from {start_offset=}") # If we stopped because we hit max_fetch (rather than exhausting the window), a backlog # remains for the next run. This is the key signal to watch during the tenant catch-up: # it means the window still has more events than max_fetch could carry this cycle. if len(alerts) >= max_fetch: demisto.info( f"[get_alerts] reached max_fetch cap ({max_fetch}); more events remain in the window " f"[LastUpdated >= {last_fetch} AND < {end_time}]. They will be collected on the next run." ) return alerts def normalize_last_updated(raw_value: Optional[str]) -> Optional[str]: """Normalize a ``LastUpdated`` string to ``DATE_FORMAT`` (``...Z``). The API returns ``+00:00`` offsets while the persisted cursor uses ``...Z``; both boundary comparisons (dedup_alerts, get_boundary_ids) must normalize both sides so the same instant compares equal. Returns ``None`` if missing/unparseable. """ parsed = safe_parse_datetime(raw_value, "LastUpdated") return parsed.strftime(DATE_FORMAT) if parsed else None def add_one_second(cursor: str) -> str: """Return ``cursor`` advanced by one second, in ``DATE_FORMAT`` (``cursor`` if unparseable). Steps the cursor past a drained boundary second so the inclusive lower bound does not re-read it. Safe because every event at that second was already fetched via the offset walk. """ parsed = safe_parse_datetime(cursor, "cursor") if not parsed: return cursor return (parsed + timedelta(seconds=1)).strftime(DATE_FORMAT) def dedup_alerts(alerts: List[Dict[str, Any]], seen_ids: List[str], boundary_time: Optional[str]) -> List[Dict[str, Any]]: """Remove alerts already returned at the previous window's upper boundary. Because the lower bound is inclusive (``LastUpdated >= last_fetch``), any alert whose ``LastUpdated`` equals the previous run's boundary second would be re-fetched. We drop exactly those alerts whose ``AlertId`` was already emitted at that boundary second. Timestamps are normalized via :func:`normalize_last_updated` before comparison so that the API's ``+00:00`` form matches the persisted ``...Z`` cursor. Args: alerts: Alerts fetched this run (ordered by ``LastUpdated`` ascending). seen_ids: AlertIds emitted at ``boundary_time`` on the previous run. boundary_time: The previous run's boundary ``LastUpdated`` value (== this run's ``last_fetch``). Returns: The alerts list with boundary-second duplicates removed. """ if not alerts or not seen_ids or not boundary_time: return alerts normalized_boundary = normalize_last_updated(boundary_time) seen = set(seen_ids) deduped = [ alert for alert in alerts if not ( normalize_last_updated(alert.get("data", {}).get("LastUpdated", {}).get("value")) == normalized_boundary and alert.get("data", {}).get("AlertId", {}).get("value") in seen ) ] demisto.debug(f"[dedup_alerts] removed {len(alerts) - len(deduped)} boundary duplicate(s) at {normalized_boundary}") return deduped def get_boundary_ids(alerts: List[Dict[str, Any]], boundary_time: str) -> List[str]: """Collect AlertIds whose ``LastUpdated`` equals the given boundary second. These IDs are persisted so the next run can skip them (see :func:`dedup_alerts`). Timestamps are normalized via :func:`normalize_last_updated` before comparison so that the API's ``+00:00`` form matches the persisted ``...Z`` cursor - without this the list comes back empty and the cursor stalls on the boundary second. Args: alerts: Alerts emitted this run. boundary_time: The newest ``LastUpdated`` value (this run's boundary). Returns: A list of AlertIds sharing ``boundary_time`` as their ``LastUpdated``. """ normalized_boundary = normalize_last_updated(boundary_time) return [ alert_id for alert in alerts if normalize_last_updated(alert.get("data", {}).get("LastUpdated", {}).get("value")) == normalized_boundary and (alert_id := alert.get("data", {}).get("AlertId", {}).get("value")) ] """ MAIN FUNCTION """ def fetch_events( client: Client, start_time: str, last_run_ids: list, last_run_offset: int, max_fetch: int, end_time: str, ) -> None: """Scheduled fetch: pull alerts, dedup, push to XSIAM, and advance the persisted cursor. ``start_time`` is the persisted cursor (inclusive lower bound on ``LastUpdated``) from the previous run's ``lastRun``. """ # Explicit fetch window for this run (lower inclusive -> upper exclusive), for diagnostics. demisto.debug(f"[Fetch Events] Fetch window: LastUpdated >= {start_time} AND < {end_time} ({max_fetch=} {last_run_offset=})") alerts = get_alerts(client, max_fetch, start_time, end_time, start_offset=last_run_offset) # Capture the RAW page size and newest timestamp BEFORE dedup: the wedge guard must key # off these, since dedup can empty a full page and hide the "full page, same second" signal. raw_fetched_count = len(alerts) newest_before_dedup = alerts[-1].get("data", {}).get("LastUpdated", {}).get("value") if alerts else None # Drop alerts already emitted at the previous run's boundary second (inclusive lower bound). alerts = dedup_alerts(alerts, last_run_ids, start_time) if alerts: alerts = add_time_key_to_alerts(alerts) alerts = add_entry_status_to_alerts(alerts) # Distribution of new vs updated: proves the LastUpdated cursor now captures updates. updated_count = sum(1 for alert in alerts if alert.get("_ENTRY_STATUS") == "updated") demisto.debug( f"[Fetch Events] Entry status distribution: new={len(alerts) - updated_count}, " f"updated={updated_count}, total={len(alerts)}" ) demisto.debug(f"[Fetch Events] before send_events_to_xsiam {VENDOR=} {PRODUCT=} {alerts=}") send_events_to_xsiam(alerts, VENDOR, PRODUCT) demisto.debug(f"[Fetch Events] after send_events_to_xsiam {VENDOR=} {PRODUCT=} {alerts=}") # Advance the cursor using the newest RAW timestamp (dedup may have emptied the list). new_last_fetch = normalize_last_updated(newest_before_dedup) or start_time page_was_full = raw_fetched_count == max_fetch same_second = new_last_fetch == normalize_last_updated(start_time) if raw_fetched_count == 0 and last_run_offset > 0: # Drained-second terminator: an empty page while resuming at a non-zero offset proves # the boundary second is fully consumed. Step the cursor +1s and clear the offset so # the inclusive lower bound does not re-read it forever (infinite duplicate loop). new_last_fetch = add_one_second(start_time) new_last_run_offset = 0 new_last_run_ids: list[str] = [] demisto.info( f"[Fetch Events] Boundary second {start_time} fully drained (empty page at offset " f"{last_run_offset}); stepping cursor to {new_last_fetch} and resetting offset." ) elif page_was_full and same_second: # Wedge guard: a full raw page still on the same second means that second holds more # than max_fetch events. Keep the timestamp but advance an in-second offset so the # next run reads the next slice. new_last_run_offset = last_run_offset + raw_fetched_count # Same second: accumulate (union) ids so earlier slices aren't lost when the offset resets. current_ids = get_boundary_ids(alerts, new_last_fetch) if alerts else [] new_last_run_ids = list(set(last_run_ids + current_ids)) demisto.info( f"[Fetch Events] Boundary second {new_last_fetch} not fully drained; advancing lastRunOffset " f"to {new_last_run_offset} (max_fetch={max_fetch})" ) else: # Normal path: page not full or crossed into a newer second. Reset the offset. new_last_run_offset = 0 current_ids = get_boundary_ids(alerts, new_last_fetch) if alerts else [] if same_second: # Offset reset while still on the same second: union so accumulated ids survive the reset. new_last_run_ids = list(set(last_run_ids + current_ids)) else: # Second advanced: only the new second's ids matter. new_last_run_ids = current_ids if alerts else last_run_ids current_last_run = { "lastRun": new_last_fetch, "lastRunIds": new_last_run_ids, "lastRunOffset": new_last_run_offset, } demisto.setLastRun(current_last_run) demisto.debug( f"[Fetch Events] Persisted lastRun={new_last_fetch} with {len(new_last_run_ids)} boundary id(s), " f"lastRunOffset={new_last_run_offset}" ) demisto.debug(f"[Fetch Events] {current_last_run=}") def get_events_command(client: Client, args: dict) -> None: """Manual debug command: pull alerts for an explicit window and optionally push them. Driven entirely by its own ``start_time`` / ``end_time`` / ``limit`` arguments - it does NOT read or write the persisted ``lastRun`` cursor, so it is safe to run on demand without disrupting the scheduled fetch. When an argument is omitted, a sensible default window/cap is used. """ start = safe_parse_datetime(args.get("start_time") or DEFAULT_GET_EVENTS_START, "start_time") start_time = start.strftime(DATE_FORMAT) if start else "" end = safe_parse_datetime(args.get("end_time"), "end_time") end_time = end.strftime(DATE_FORMAT) if end else datetime.now(timezone.utc).strftime(DATE_FORMAT) limit = arg_to_number(args.get("limit")) or DEFAULT_GET_EVENTS_LIMIT should_push_events = argToBoolean(args.get("should_push_events", False)) demisto.debug(f"[Get Events] Fetch window: LastUpdated >= {start_time} AND < {end_time} ({limit=})") alerts = get_alerts(client, limit, start_time, end_time) # Normalize before returning so the CommandResults outputs carry the parsed events (_time / _ENTRY_STATUS). if alerts: alerts = add_time_key_to_alerts(alerts) alerts = add_entry_status_to_alerts(alerts) return_results( CommandResults( readable_output=tableToMarkdown(t=alerts, name=f"{VENDOR} - {PRODUCT} events", removeNull=True), outputs_prefix=f"{VENDOR.capitalize()}.Events", outputs=alerts, raw_response=alerts, ) ) if should_push_events and alerts: demisto.debug(f"[Get Events] before send_events_to_xsiam {VENDOR=} {PRODUCT=} count={len(alerts)}") send_events_to_xsiam(alerts, VENDOR, PRODUCT) demisto.debug(f"[Get Events] after send_events_to_xsiam {VENDOR=} {PRODUCT=} count={len(alerts)}") def main() -> None: command = demisto.command() api_token = demisto.params().get("credentials", {}).get("password") server_url = f"{demisto.params().get('server_url')}/api" # Event Collectors default to now - 1 minute when no first_fetch is configured. first_fetch = demisto.params().get("first_fetch") or "1 minute" max_fetch = arg_to_number(demisto.params().get("max_fetch")) or 1000 verify_certificate = not demisto.params().get("insecure", False) proxy = demisto.params().get("proxy", False) # How much time before the first fetch to retrieve events first_fetch_time = arg_to_datetime(arg=first_fetch, arg_name="First fetch time", required=True) first_fetch_time = first_fetch_time.strftime(DATE_FORMAT) if first_fetch_time else "" # Fixed exclusive upper bound for this run, so events synced mid-run aren't half-fetched. end_time = datetime.now(timezone.utc).strftime(DATE_FORMAT) demisto.debug(f"[Main] {first_fetch_time=} {end_time=}") demisto.info(f"[Main] Orca Security. Command being called is {command}") try: headers: dict = {"Authorization": f"Token {api_token}"} client = Client(server_url=server_url, verify=verify_certificate, headers=headers, proxy=proxy) last_run = demisto.getLastRun() if not last_run: demisto.debug(f"[Fetch Events] first run {last_run=}") last_fetch = first_fetch_time last_run_ids: list = [] last_run_offset = 0 else: last_fetch = last_run.get("lastRun") last_run_ids = last_run.get("lastRunIds", []) # In-second resume point: how many events at the boundary second were already consumed. # Non-zero only when a single LastUpdated second holds more events than max_fetch. last_run_offset = last_run.get("lastRunOffset", 0) demisto.debug(f"[Fetch Events] Isn't the first run {last_fetch=} {last_run_ids=} {last_run_offset=}") if command == "test-module": return_results(orca_test_module(client, last_fetch, end_time)) elif command == "fetch-events": fetch_events(client, last_fetch, last_run_ids, last_run_offset, max_fetch, end_time) elif command == "orca-security-get-events": get_events_command(client, demisto.args()) else: raise NotImplementedError("This command is not implemented yet.") # Log exceptions and return errors except Exception as e: return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e!s}") """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()