XSUP-78726: Generic API Event Collector - fix pagination crash, integer ID cursor, and unresolved placeholders

Packs/GenericAPIEventCollector/Integrations/GenericAPIEventCollector/GenericAPIEventCollector.py

@@ -150,14 +150,20 @@ def recursive_replace(org_dict: dict[Any, Any] | None, substitutions: list[tuple
     # Create a deep copy of the dictionary to avoid modifying the original
     copy_dict = copy.deepcopy(org_dict)
 
-    for key, value in copy_dict.items():
+    for key, value in list(copy_dict.items()):
         if isinstance(value, dict):
             # If value is a dictionary, recursively call this function
             copy_dict[key] = recursive_replace(value, substitutions)
         elif isinstance(value, str):
             # If value is a string, perform substitutions
             for old, new in substitutions:
                 value = value.replace(old, new)
+            # A placeholder with no value yet (e.g. @last_fetched_id before the first stored ID) would be sent
+            # literally and rejected by the API, so the parameter is omitted instead.
+            if unresolved := [p.value for p in PlaceHolders if p.value in value]:
+                demisto.debug(f"Omitting parameter {key}: no value yet for {', '.join(unresolved)}")
+                del copy_dict[key]
+                continue
             copy_dict[key] = value
 
     return copy_dict
@@ -253,11 +259,15 @@ def fetch_events(
             if id_type.lower() not in ALL_ID_TYPES:
                 return_error(f"ID type {id_type} but must be one of {', '.join(ALL_ID_TYPES)}")
                 return {}, []
+            id_type_lower = id_type.lower()
             demisto.debug(f"ID type:{id_type}")
         else:
             return_error("ID type was not specified")
             return {}, []
 
+    # Stop paginating once the configured per-fetch limit is reached; the next fetch continues from the saved cursor.
+    fetch_limit: int = min(arg_to_number(params.get("limit")) or MAX_INCIDENTS_TO_FETCH, MAX_INCIDENTS_TO_FETCH)
+    previous_page_value: Any = None
     while pagination_needed:
         raw_events = client.search_events(
             endpoint=endpoint, http_method=http_method, request_data=request_data, ok_codes=ok_codes
@@ -266,15 +276,35 @@ def fetch_events(
         all_events_list.extend(events_list)
         demisto.debug(f"{len(all_events_list)} events fetched")
         pagination_needed, next_page_value = is_pagination_needed(raw_events, pagination_logic)
+        # Stop on an empty page or an empty/unchanged cursor, otherwise APIs that always return
+        # the pagination flag (e.g. a sequence cursor) would loop forever.
+        if pagination_needed and (
+            not events_list
+            or not next_page_value
+            or next_page_value == previous_page_value
+            or len(all_events_list) >= fetch_limit
+        ):
+            demisto.debug(f"Stopping pagination: page size {len(events_list)}, next page value {next_page_value}")
+            pagination_needed = False
         if pagination_needed:
-            request_json = {pagination_logic.pagination_field_name: next_page_value}
-            request_data = RequestData(request_data.request_data, request_json, request_data.query_params)
+            previous_page_value = next_page_value
+            # pagination_field_name is a dot-separated path in the response; the request uses its last key.
+            page_param = {pagination_logic.pagination_field_name[-1]: next_page_value}
+            if http_method.upper() == "GET":
+                request_data = RequestData(
+                    request_data.request_data, request_data.request_json, {**(request_data.query_params or {}), **page_param}
+                )
+            else:
+                request_data = RequestData(
+                    request_data.request_data, {**(request_data.request_json or {}), **page_param}, request_data.query_params
+                )
 
     # endregion
 
     # region Collect all events based on their last fetch time.
     latest_created_datetime: datetime = last_fetched_datetime
     last_fetched_id: Any | None = last_run.get(PlaceHolders.LAST_FETCHED_ID.value)
+    previous_id_int = arg_to_number(last_fetched_id) if id_type_lower == IdTypes.INTEGER.value else None
     returned_event_list: list[dict[str, Any]] = []
     for event in all_events_list:
         try:
@@ -283,21 +313,35 @@ def fetch_events(
             demisto.error(f"Error parsing timestamp for event: {event} exception: {e}")
             continue
         event["_time"] = incident_created_dt.isoformat()
+        current_id: Any = dict_safe_get(event, id_keys) if id_keys else None
+        # Some APIs wrap the ID in a one-item list, e.g. Group-IB breacheddb addInfo.updateTimestampLong.
+        if isinstance(current_id, list) and len(current_id) == 1:
+            current_id = current_id[0]
+        try:
+            current_id_int = arg_to_number(current_id) if id_type_lower == IdTypes.INTEGER.value else None
+        except ValueError as e:
+            # Skip the event so a non-numeric ID is never saved as the integer cursor.
+            demisto.error(f"Skipping event with a non-integer ID {current_id!r}: {e}")
+            continue
+
+        # An integer ID above the last fetched ID is new even if its timestamp is older, e.g. feeds
+        # ordered by an update sequence where the timestamp is the original content date. With no stored ID
+        # yet (first fetch), every event with an ID is new, otherwise the first batch is dropped by the time filter.
+        is_new_by_id = current_id_int is not None and (previous_id_int is None or current_id_int > previous_id_int)
 
         # to prevent duplicates, we are only adding events with creation_time > last fetched incident
-        if incident_created_dt > last_fetched_datetime:
+        if incident_created_dt > last_fetched_datetime or is_new_by_id:
             demisto.debug(f"Adding event with creation time: {incident_created_dt}")
             returned_event_list.append(event)
             latest_created_datetime = max(latest_created_datetime, incident_created_dt)
         else:
             demisto.debug(f"This event is to old to pull, creation time: {incident_created_dt}")
 
-        # Handle the last event id.
-        if id_keys:
-            current_id: Any = dict_safe_get(event, id_keys)
+        # Handle the last event id. An event without an ID must not wipe the stored cursor.
+        if id_keys and current_id is not None:
             demisto.debug(f"Current event id: {current_id}")
-            if last_fetched_id is not None and id_type_lower == IdTypes.INTEGER.value:
-                last_fetched_id = str(max(int(last_fetched_id), int(current_id)))  # noqa
+            if current_id_int is not None and last_fetched_id is not None:
+                last_fetched_id = str(max(int(last_fetched_id), current_id_int))
             else:
                 # We assume the last event contains the last id.
                 last_fetched_id = current_id

Packs/GenericAPIEventCollector/Integrations/GenericAPIEventCollector/GenericAPIEventCollector_test.py

@@ -67,6 +67,19 @@ def test_recursive_replace_with_none():
     assert recursive_replace(None, []) is None
 
 
+def test_recursive_replace_omits_unresolved_placeholders():
+    """
+    Given: query params referencing @last_fetched_id while the last run has no stored ID yet.
+    When: recursive_replace substitutes only the placeholders that have values.
+    Then: the unresolved parameter is omitted instead of being sent literally (regression test, XSUP-78726).
+    """
+    query_params = {"limit": "@fetch_size_limit", "seqUpdate": "@last_fetched_id", "nested": {"cursor": "@last_fetched_id"}}
+
+    result = recursive_replace(query_params, [("@fetch_size_limit", "10")])
+
+    assert result == {"limit": "10", "nested": {}}
+
+
 def test_get_time_field_from_event_to_dt():
     event = {"timestamp": "2023-05-01T12:00:00Z"}
     config = TimestampFieldConfig(["timestamp"], "%Y-%m-%dT%H:%M:%SZ")
@@ -944,6 +957,242 @@ def search_events(self, **kwargs):
     assert "NOT-A-TIMESTAMP" in mock_error.call_args[0][0]
 
 
+class RecordingClient:
+    """Returns the given pages in order and records the request of every call."""
+
+    def __init__(self, pages):
+        self.pages = list(pages)
+        self.requests: list[RequestData] = []
+
+    def search_events(self, **kwargs):
+        self.requests.append(kwargs["request_data"])
+        return self.pages.pop(0) if self.pages else {"items": []}
+
+
+def run_fetch(client, params, last_run, http_method="GET"):
+    return fetch_events(
+        client=client,
+        params=params,
+        last_run=last_run,
+        first_fetch_datetime=datetime(2026, 10, 11),
+        endpoint="/e",
+        http_method=http_method,
+        ok_codes=[200],
+        events_keys=["items"],
+        timestamp_field_config=TimestampFieldConfig(["datetime"], "%Y-%m-%dT%H:%M:%S"),
+    )
+
+
+def test_fetch_events_pagination_sends_cursor_as_query_param():
+    """
+    Given: pagination configured with a cursor field that is a list (as built by argToList).
+    When: fetch_events paginates a GET endpoint.
+    Then: no "unhashable type: 'list'" error, the cursor is sent as a query parameter on the next
+          request, and pagination stops on an empty page (regression test, XSUP-78726).
+    """
+    client = RecordingClient(
+        [
+            {"items": [{"datetime": "2026-10-12T00:00:00", "seqUpdate": 5}], "seqUpdate": 5},
+            {"items": [], "seqUpdate": 5},
+        ]
+    )
+    params = {"pagination_needed": "true", "pagination_field_name": "seqUpdate", "pagination_flag": "seqUpdate"}
+
+    _, events = run_fetch(client, params, {"@first_fetch_datetime": "2026-10-11T00:00:00"})
+
+    assert len(events) == 1
+    assert len(client.requests) == 2
+    assert client.requests[1].query_params == {"seqUpdate": 5}
+
+
+def test_fetch_events_pagination_stops_when_cursor_does_not_advance():
+    """
+    Given: an API that always returns the pagination flag with the same cursor.
+    When: fetch_events paginates.
+    Then: it stops instead of looping forever.
+    """
+    page = {"items": [{"datetime": "2026-10-12T00:00:00"}], "next": "abc"}
+    client = RecordingClient([page, page, page])
+    params = {"pagination_needed": "true", "pagination_field_name": "next", "pagination_flag": "next"}
+
+    run_fetch(client, params, {"@first_fetch_datetime": "2026-10-11T00:00:00"}, http_method="POST")
+
+    assert len(client.requests) == 2
+    assert client.requests[1].request_json == {"next": "abc"}
+
+
+@pytest.mark.parametrize("empty_cursor", ["", 0])
+def test_fetch_events_pagination_stops_on_empty_cursor(empty_cursor):
+    """
+    Given: an API that returns the pagination flag with an empty cursor ("" or 0) on the first page.
+    When: fetch_events paginates.
+    Then: no extra request is sent with the empty cursor.
+    """
+    client = RecordingClient([{"items": [{"datetime": "2026-10-12T00:00:00"}], "more": True, "next": empty_cursor}])
+    params = {"pagination_needed": "true", "pagination_field_name": "next", "pagination_flag": "more"}
+
+    run_fetch(client, params, {"@first_fetch_datetime": "2026-10-11T00:00:00"})
+
+    assert len(client.requests) == 1
+
+
+def test_fetch_events_integer_id_first_fetch_accepts_events_with_old_timestamps():
+    """
+    Given: ID type "integer" and no stored ID yet (first fetch).
+    When: the API returns events whose timestamps are older than the first fetch time.
+    Then: they are collected and the cursor is set to the highest ID.
+    """
+    client = RecordingClient(
+        [{"items": [{"datetime": "2014-06-04T01:28:00", "seq": 7}, {"datetime": "2020-01-01T00:00:00", "seq": 9}]}]
+    )
+    params = {"id_keys": "seq", "id_type": "integer"}
+
+    next_run, events = run_fetch(client, params, {"@first_fetch_datetime": "2026-10-11T00:00:00"})
+
+    assert [event["seq"] for event in events] == [7, 9]
+    assert next_run["@last_fetched_id"] == "9"
+
+
+@pytest.mark.parametrize("id_type", ["integer", "string"])
+def test_fetch_events_event_without_id_keeps_cursor(id_type):
+    """
+    Given: a stored cursor and a batch whose last event has no ID.
+    When: fetch_events runs.
+    Then: the stored cursor is kept instead of being wiped.
+    """
+    client = RecordingClient([{"items": [{"datetime": "2026-10-12T00:00:00", "seq": 5}, {"datetime": "2026-10-12T00:00:01"}]}])
+    params = {"id_keys": "seq", "id_type": id_type}
+    last_run = {"@first_fetch_datetime": "2026-10-11T00:00:00", "@last_fetched_id": "3"}
+
+    next_run, _ = run_fetch(client, params, last_run)
+
+    assert next_run["@last_fetched_id"] == "5"
+
+
+def test_fetch_events_pagination_stops_at_configured_limit():
+    """
+    Given: a limit of 2 and an API that keeps returning pages with an advancing cursor.
+    When: fetch_events paginates.
+    Then: it stops once the limit is reached instead of paging up to MAX_INCIDENTS_TO_FETCH.
+    """
+    client = RecordingClient([{"items": [{"datetime": "2026-10-12T00:00:00"}], "next": str(page)} for page in range(5)])
+    params = {"limit": "2", "pagination_needed": "true", "pagination_field_name": "next", "pagination_flag": "next"}
+
+    run_fetch(client, params, {"@first_fetch_datetime": "2026-10-11T00:00:00"})
+
+    assert len(client.requests) == 2
+
+
+@patch("GenericAPIEventCollector.demisto.error")
+def test_fetch_events_skips_event_with_non_integer_id(mock_error):
+    """
+    Given: ID type "integer" and one event whose ID is not a number.
+    When: fetch_events runs.
+    Then: that event is skipped and logged, the other events are kept, and the cursor stays numeric.
+    """
+    client = RecordingClient(
+        [
+            {
+                "items": [
+                    {"datetime": "2026-10-12T00:00:00", "seq": 5},
+                    {"datetime": "2026-10-12T00:00:01", "seq": "not-a-number"},
+                ]
+            }
+        ]
+    )
+    params = {"id_keys": "seq", "id_type": "integer"}
+
+    next_run, events = run_fetch(client, params, {"@first_fetch_datetime": "2026-10-11T00:00:00"})
+
+    assert [event["seq"] for event in events] == [5]
+    assert next_run["@last_fetched_id"] == "5"
+    mock_error.assert_called_once()
+
+
+def test_fetch_events_integer_id_cursor_accepts_events_with_old_timestamps():
+    """
+    Given: a feed ordered by an integer update sequence whose timestamp is the original content date
+           (Group-IB darkweb items), and a last run holding the previous sequence.
+    When: fetch_events runs with ID type "integer".
+    Then: events with a higher sequence are ingested even though their timestamp is old, events at
+          or below the stored sequence are dropped, and the cursor becomes the maximum sequence
+          even when the API returns items out of order (regression test, XSUP-78726).
+    """
+    client = RecordingClient(
+        [
+            {
+                "items": [
+                    {"datetime": "2014-06-04T01:28:00", "seqUpdate": 300},
+                    {"datetime": "2025-03-09T17:15:00", "seqUpdate": 100},
+                    {"datetime": "2023-12-23T15:28:00", "seqUpdate": 200},
+                ]
+            }
+        ]
+    )
+    params = {"id_keys": "seqUpdate", "id_type": "integer"}
+    last_run = {
+        "@first_fetch_datetime": "2026-10-11T00:00:00",
+        "@last_fetched_datetime": "2026-10-11T00:00:00",
+        "@last_fetched_id": "100",
+    }
+
+    next_run, events = run_fetch(client, params, last_run)
+
+    assert [event["seqUpdate"] for event in events] == [300, 200]
+    assert next_run["@last_fetched_id"] == "300"
+    assert next_run["@last_fetched_datetime"] == "2026-10-11T00:00:00"
+
+
+def test_fetch_events_integer_id_wrapped_in_one_item_list():
+    """
+    Given: an integer ID returned as a one-item list (Group-IB breacheddb addInfo.updateTimestampLong).
+    When: fetch_events runs with ID type "integer".
+    Then: the list is unwrapped, so the fetch does not fail with "is not a valid number" and the cursor
+          is saved as a plain number (regression test, XSUP-78726).
+    """
+    client = RecordingClient(
+        [
+            {
+                "items": [
+                    {"datetime": "2020-01-18T12:50:49", "addInfo": {"updateTimestampLong": [1579333849000000]}},
+                    {"datetime": "2020-01-18T17:25:52", "addInfo": {"updateTimestampLong": [1579350352000000]}},
+                ]
+            }
+        ]
+    )
+    params = {"id_keys": "addInfo.updateTimestampLong", "id_type": "integer"}
+    last_run = {
+        "@first_fetch_datetime": "2026-10-11T00:00:00",
+        "@last_fetched_datetime": "2026-10-11T00:00:00",
+        "@last_fetched_id": "1579333849000000",
+    }
+
+    next_run, events = run_fetch(client, params, last_run)
+
+    assert len(events) == 1
+    assert next_run["@last_fetched_id"] == "1579350352000000"
+
+
+def test_fetch_events_string_id_keeps_time_filter():
+    """
+    Given: ID type "string".
+    When: fetch_events runs with events older than the last fetched time.
+    Then: the time filter still applies (string IDs are not comparable), and the last event's ID is kept.
+    """
+    client = RecordingClient([{"items": [{"datetime": "2020-01-01T00:00:00", "id": "b"}]}])
+    params = {"id_keys": "id", "id_type": "string"}
+    last_run = {
+        "@first_fetch_datetime": "2026-10-11T00:00:00",
+        "@last_fetched_datetime": "2026-10-11T00:00:00",
+        "@last_fetched_id": "a",
+    }
+
+    next_run, events = run_fetch(client, params, last_run)
+
+    assert events == []
+    assert next_run["@last_fetched_id"] == "b"
+
+
 def test_derive_authorize_url_swaps_last_path_segment():
     """
     Given: a Microsoft Entra style token endpoint.

Packs/GenericAPIEventCollector/ReleaseNotes/1_1_2.md

@@ -0,0 +1,16 @@
+
+#### Integrations
+
+##### Generic API Event Collector (Beta)
+
+- Fixed an issue where fetching events failed with the *unhashable type: 'list'* error when *Is pagination needed* was enabled. The next page value is now sent as a query parameter for GET requests and in the JSON body for POST requests.
+- Fixed an issue where pagination could loop indefinitely when the API kept returning the same next page value.
+- Fixed an issue where the *integer* ID type was ignored, so the last fetched ID was the ID of the last event returned instead of the highest ID.
+- Fixed an issue where events with an ID higher than the last fetched ID were not collected if their timestamp was older than the last fetch time when the ID type is *integer*.
+- Fixed an issue where a request parameter referencing a placeholder with no stored value yet (for example, *@last_fetched_id* before the first ID was saved) was sent as literal text. Such parameters are now omitted from the request.
+- Fixed an issue where fetching events failed with the *is not a valid number* error when the ID type is *integer* and the API returns the ID inside a single-item list.
+- Fixed an issue where an event with a non-numeric ID failed the whole fetch when the ID type is *integer*. Such events are now skipped and logged.
+- Fixed an issue where pagination ignored the *Number of incidents to fetch per fetch* parameter and continued up to 10,000 events.
+- Fixed an issue where pagination sent an extra request when the API returned an empty next page value.
+- Fixed an issue where the first fetch dropped events with timestamps older than the first fetch time when the ID type is *integer*.
+- Fixed an issue where an event without an ID cleared the last fetched ID, causing the next fetch to start over.