diff --git a/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC.py b/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC.py index 6133b6455dd..19ca6d5fd1e 100644 --- a/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC.py +++ b/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC.py @@ -13,6 +13,7 @@ DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" FETCH_INTERVAL_DEFAULT = 1 MAX_FETCH_DEFAULT = 10 +MAX_LOOK_BACK_DAYS = 1 SAAS_NAMES = ["office365_emails", "google_mail"] SAAS_APPS_TO_SAAS_NAMES = {"Microsoft Exchange": "office365_emails", "Gmail": "google_mail"} SEVERITY_VALUES = {"critical": 5, "high": 4, "medium": 3, "low": 2, "very low": 1} @@ -163,8 +164,13 @@ def test_api(self) -> dict[str, Any]: return self._call_api("GET", url_suffix="scopes") def restore_requests( - self, start_date: str, saas: str, include_denied: Optional[bool], include_accepted: Optional[bool] - ) -> dict[str, Any]: + self, + start_date: str, + saas: str, + include_denied: Optional[bool], + include_accepted: Optional[bool], + min_results: int, + ) -> list[dict[str, Any]]: denied_attr_op = "is" if include_denied else "isNot" accepted_attr_op = "is" if include_accepted else "isNot" fifteen_days_ago = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=15)).isoformat() @@ -182,8 +188,23 @@ def restore_requests( {"saasAttrName": "entityPayload.isRestored", "saasAttrOp": accepted_attr_op, "saasAttrValue": "true"}, ], } - payload = {"requestData": request_data} - return self._call_api("POST", url_suffix="search/query", json_data=payload) + + entries: list[dict[str, Any]] = [] + for _ in range(20): + result = self._call_api("POST", url_suffix="search/query", json_data={"requestData": request_data}) + entries.extend(result.get("responseData") or []) + + envelope = result.get("responseEnvelope") or {} + scroll_id = envelope.get("scrollId") + total = envelope.get("recordsNumber") or 0 + # The same scroll id is returned for every page, so it cannot be used to detect the end of the scroll. + if not scroll_id or len(entries) >= total or len(entries) >= min_results: + break + request_data["scrollId"] = scroll_id + else: + demisto.debug(f"Stopped paging restore requests for {saas} after 20 pages") + + return entries def query_events( self, @@ -628,6 +649,11 @@ def fetch_incidents(client: Client, params: dict): counter = 0 incidents: List[dict[str, Any]] = [] + demisto.debug( + f"fetch-incidents window {last_fetch} -> {now_15.isoformat()} | saas={saas_apps} states={states} " + f"severities={severities} threat_types={threat_types} max_fetch={max_fetch}" + ) + result = client.query_events( start_date=last_fetch, end_date=now_15.isoformat(), @@ -637,6 +663,7 @@ def fetch_incidents(client: Client, params: dict): threat_types=threat_types, ) events = result["responseData"] + demisto.debug(f"fetch-incidents query returned {len(events)} events") for event in events: if (occurred := event.get("eventCreated")) <= last_fetch: @@ -666,6 +693,11 @@ def fetch_incidents(client: Client, params: dict): else: last_run["last_fetch"] = (now_15 - timedelta(minutes=fetch_interval)).isoformat() + demisto.debug( + f"fetch-incidents created {len(incidents)} incidents from {len(events)} events " + f"(truncated={counter == max_fetch}) | next last_fetch={last_run['last_fetch']}" + ) + demisto.setLastRun(last_run) demisto.incidents(incidents) @@ -674,56 +706,78 @@ def fetch_restore_requests(client: Client, params: dict): first_fetch: str = params.get("first_fetch", "1 hour") saas_apps: List[str] = [SAAS_APPS_TO_SAAS_NAMES[x] for x in argToList(params.get("saas_apps"))] or SAAS_NAMES max_fetch: int = arg_to_number(params.get("max_fetch")) or MAX_FETCH_DEFAULT - fetch_interval: int = arg_to_number(params.get("incidentFetchInterval")) or FETCH_INTERVAL_DEFAULT + max_lookup_time = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=MAX_LOOK_BACK_DAYS)).isoformat() - now = datetime.now(timezone.utc).replace(tzinfo=None) # We get current time before processing - last_run = demisto.getLastRun() - if not (last_fetch := last_run.get("last_rr_fetch")): - if last_fetch := dateparser.parse(first_fetch, date_formats=[DATE_FORMAT]): - last_fetch = last_fetch.isoformat() - else: - raise DemistoException("Could not get last restore request fetch") + if not (first_fetch_dt := dateparser.parse(first_fetch, date_formats=[DATE_FORMAT])): + raise DemistoException("Could not get last restore request fetch") + default_cursor = first_fetch_dt.isoformat() - counter = 0 - incidents: List[dict[str, Any]] = [] + last_run = demisto.getLastRun() + stored_cursor = last_run.get("last_rr_fetch") + # Pre-1.1.16 instances stored a single cursor shared by every saas app; seed each one from it on upgrade. + if isinstance(stored_cursor, str): + cursors: dict[str, str] = dict.fromkeys(saas_apps, stored_cursor) + else: + cursors = dict(stored_cursor or {}) include_denied_rr: Optional[bool] = arg_to_bool(params.get("include_denied_requests")) include_accepted_rr: Optional[bool] = arg_to_bool(params.get("include_accepted_requests")) + + demisto.debug( + f"fetch restore-requests cursors={ {saas: cursors.get(saas) for saas in saas_apps} } max_fetch={max_fetch} " + f"include_denied={include_denied_rr} include_accepted={include_accepted_rr}" + ) + + candidates: List[tuple[str, str, dict, dict]] = [] for saas in saas_apps: - result = client.restore_requests(last_fetch, saas, include_denied_rr, include_accepted_rr) - for restore_request in result["responseData"]: - entity_info = restore_request.get("entityInfo") - entity_payload = restore_request.get("entityPayload") + saas_cursor = max(cursors.get(saas) or default_cursor, max_lookup_time) + + for restore_request in client.restore_requests(saas_cursor, saas, include_denied_rr, include_accepted_rr, max_fetch): + entity_info = restore_request.get("entityInfo") or {} + entity_payload = restore_request.get("entityPayload") or {} if entity_payload.get("emailSplit") == "split": # is master email, skipping continue - if (occurred := entity_payload.get("restoreRequestTime")) <= last_fetch: + occurred = entity_payload.get("restoreRequestTime") + if not occurred or occurred <= saas_cursor: continue - count_field = "count_restore_request" - count = last_run.get(count_field, 0) + 1 - last_run[count_field] = count + candidates.append((occurred, saas, entity_info, entity_payload)) - entity_payload["entityId"] = entity_info.get("entityId") - incidents.append( - { - "dbotMirrorId": entity_info.get("entityId"), - "details": entity_payload.get("restoreCommentary"), - "name": f"Threat: Restore Request {count}", - "occurred": occurred, - "rawJSON": json.dumps(entity_payload), - } - ) + # Every response is ascending on its own, so this merges the per saas streams to keep the oldest records on + # truncation and to make the last incident the newest one. + candidates.sort(key=lambda candidate: candidate[0]) - if max_fetch == (counter := counter + 1): - break + incidents: List[dict[str, Any]] = [] + count_field = "count_restore_request" + count = last_run.get(count_field, 0) + for occurred, saas, entity_info, entity_payload in candidates[:max_fetch]: + count += 1 + entity_payload["entityId"] = entity_info.get("entityId") + incidents.append( + { + "dbotMirrorId": entity_info.get("entityId"), + "details": entity_payload.get("restoreCommentary"), + "name": f"Threat: Restore Request {count}", + "occurred": occurred, + "rawJSON": json.dumps(entity_payload), + } + ) + # Candidates are ascending, so the final write per saas app is that app's newest emitted record. Saas apps + # that emitted nothing keep their previous cursor instead of being dragged forward by a busier app. + cursors[saas] = occurred if incidents: - last_run["last_rr_fetch"] = incidents[-1]["occurred"] - else: - last_run["last_rr_fetch"] = (now - timedelta(minutes=fetch_interval)).isoformat() + last_run[count_field] = count + + last_run["last_rr_fetch"] = cursors + + demisto.debug( + f"fetch restore-requests created {len(incidents)} incidents from {len(candidates)} candidates " + f"(truncated={len(candidates) > max_fetch}) | next cursors={cursors}" + ) demisto.setLastRun(last_run) demisto.incidents(incidents) diff --git a/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC_test.py b/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC_test.py index 617b150e2a4..72341368513 100644 --- a/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC_test.py +++ b/Packs/CheckPointHEC/Integrations/CheckPointHEC/CheckPointHEC_test.py @@ -1,8 +1,12 @@ import json +from datetime import datetime, timedelta, UTC import demistomock as demisto import pytest from CheckPointHEC import ( + MAX_LOOK_BACK_DAYS, + SAAS_APPS_TO_SAAS_NAMES, + SAAS_NAMES, Client, checkpointhec_create_anomaly_exception, checkpointhec_create_ap_exception, @@ -289,6 +293,221 @@ def test_fetch_restore_requests(mocker): demisto_incidents.assert_called_once() +def _restore_requests_client(): + return Client( + base_url="https://smart-api-example-1-us.avanan-example.net", + client_id="****", + client_secret="****", + verify=True, + proxy=False, + ) + + +def _hours_ago(hours: int, suffix: str = "Z"): + """Restore request timestamps have to be recent, otherwise the look back clamp filters them out.""" + return (datetime.now(UTC).replace(tzinfo=None) - timedelta(hours=hours)).isoformat() + suffix + + +def _restore_request_entry(entity_id: str, occurred, **payload): + return { + "entityInfo": {"entityId": entity_id}, + "entityPayload": {"restoreRequestTime": occurred, "restoreCommentary": "please restore", **payload}, + } + + +def _restore_requests_response(entries: list, scroll_id: str = "", total: int = None): + return { + "responseEnvelope": { + "recordsNumber": len(entries) if total is None else total, + "scrollId": scroll_id, + }, + "responseData": entries, + } + + +def test_fetch_restore_requests_empty_keeps_cursor(mocker): + """An empty result must leave the cursor untouched, otherwise requests that are not yet searchable are skipped.""" + client = _restore_requests_client() + mocker.patch.object(Client, "_call_api", return_value=_restore_requests_response([])) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": "2023-06-30T00:00:00"}) + demisto_incidents = mocker.patch.object(demisto, "incidents") + set_last_run = mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day", "saas_apps": "Microsoft Exchange"}) + + exchange = SAAS_APPS_TO_SAAS_NAMES["Microsoft Exchange"] + assert set_last_run.call_args[0][0]["last_rr_fetch"] == {exchange: "2023-06-30T00:00:00"} + demisto_incidents.assert_called_once_with([]) + + +def test_fetch_restore_requests_cursor_is_per_saas(mocker): + """Each saas app must advance to its own newest record, not to the newest record across every app.""" + client = _restore_requests_client() + first_saas, second_saas = SAAS_NAMES + newer, older = _hours_ago(1), _hours_ago(2) + mocker.patch.object( + Client, + "_call_api", + side_effect=[ + _restore_requests_response([_restore_request_entry("newer", newer)]), + _restore_requests_response([_restore_request_entry("older", older)]), + ], + ) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": _hours_ago(3, "")}) + demisto_incidents = mocker.patch.object(demisto, "incidents") + set_last_run = mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day"}) + + incidents = demisto_incidents.call_args[0][0] + assert [incident["dbotMirrorId"] for incident in incidents] == ["older", "newer"] + assert set_last_run.call_args[0][0]["last_rr_fetch"] == {first_saas: newer, second_saas: older} + + +def test_fetch_restore_requests_queries_each_saas_from_its_own_cursor(mocker): + """A stored per saas cursor must scope that app's query window, so a busy app cannot skip a quiet app's records.""" + client = _restore_requests_client() + first_saas, second_saas = SAAS_NAMES + ahead, behind = _hours_ago(1, ""), _hours_ago(5, "") + call_api = mocker.patch.object(Client, "_call_api", return_value=_restore_requests_response([])) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": {first_saas: ahead, second_saas: behind}}) + mocker.patch.object(demisto, "incidents") + mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day"}) + + start_dates = [ + next( + f["saasAttrValue"] + for f in call.kwargs["json_data"]["requestData"]["entityExtendedFilter"] + if f["saasAttrName"] == "entityPayload.restoreRequestTime" + ) + for call in call_api.call_args_list + ] + assert start_dates == [ahead, behind] + + +def test_fetch_restore_requests_clamps_stale_cursor(mocker): + """A cursor older than the look back limit must be clamped so the query window stays bounded.""" + client = _restore_requests_client() + call_api = mocker.patch.object(Client, "_call_api", return_value=_restore_requests_response([])) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": "2020-01-01T00:00:00"}) + mocker.patch.object(demisto, "incidents") + mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day", "saas_apps": "Microsoft Exchange"}) + + extended_filter = call_api.call_args.kwargs["json_data"]["requestData"]["entityExtendedFilter"] + start_date = next(f["saasAttrValue"] for f in extended_filter if f["saasAttrName"] == "entityPayload.restoreRequestTime") + expected = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=MAX_LOOK_BACK_DAYS) + assert abs((datetime.fromisoformat(start_date) - expected).total_seconds()) < 60 + + +def test_fetch_restore_requests_max_fetch_across_saas(mocker): + """max_fetch must be enforced across all saas apps, and the remainder left for the next fetch.""" + client = _restore_requests_client() + first_saas, second_saas = SAAS_NAMES + seeded = _hours_ago(3, "") + newer, older = _hours_ago(1), _hours_ago(2) + mocker.patch.object( + Client, + "_call_api", + side_effect=[ + _restore_requests_response([_restore_request_entry("newer", newer)]), + _restore_requests_response([_restore_request_entry("older", older)]), + ], + ) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": seeded}) + demisto_incidents = mocker.patch.object(demisto, "incidents") + set_last_run = mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day", "max_fetch": "1"}) + + incidents = demisto_incidents.call_args[0][0] + assert [incident["dbotMirrorId"] for incident in incidents] == ["older"] + # Truncation dropped the first app's record, so only the app we emitted for may advance. + assert set_last_run.call_args[0][0]["last_rr_fetch"] == {first_saas: seeded, second_saas: older} + + +def test_fetch_restore_requests_skips_missing_request_time(mocker): + """A restore request without a restoreRequestTime must be skipped instead of failing the whole fetch.""" + client = _restore_requests_client() + mocker.patch.object( + Client, + "_call_api", + return_value=_restore_requests_response( + [ + _restore_request_entry("no-time", None), + _restore_request_entry("valid", _hours_ago(1)), + ] + ), + ) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": _hours_ago(3, "")}) + demisto_incidents = mocker.patch.object(demisto, "incidents") + mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day", "saas_apps": "Microsoft Exchange"}) + + incidents = demisto_incidents.call_args[0][0] + assert [incident["dbotMirrorId"] for incident in incidents] == ["valid"] + + +def test_fetch_restore_requests_follows_scroll(mocker): + """Results beyond the first page must be retrieved by sending the scroll id back.""" + client = _restore_requests_client() + call_api = mocker.patch.object( + Client, + "_call_api", + side_effect=[ + _restore_requests_response( + [ + _restore_request_entry("first", _hours_ago(3)), + _restore_request_entry("second", _hours_ago(2)), + ], + scroll_id="abc", + total=3, + ), + # The server returns the same scroll id for every page of a scroll. + _restore_requests_response([_restore_request_entry("third", _hours_ago(1))], scroll_id="abc", total=3), + ], + ) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": _hours_ago(5, "")}) + demisto_incidents = mocker.patch.object(demisto, "incidents") + mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day", "saas_apps": "Microsoft Exchange"}) + + assert call_api.call_count == 2 + assert call_api.call_args_list[1].kwargs["json_data"]["requestData"]["scrollId"] == "abc" + incidents = demisto_incidents.call_args[0][0] + assert [incident["dbotMirrorId"] for incident in incidents] == ["first", "second", "third"] + + +def test_fetch_restore_requests_stops_paging_at_max_fetch(mocker): + """Since pages arrive oldest first, paging must stop once enough records are held.""" + client = _restore_requests_client() + call_api = mocker.patch.object( + Client, + "_call_api", + return_value=_restore_requests_response( + [ + _restore_request_entry("first", _hours_ago(2)), + _restore_request_entry("second", _hours_ago(1)), + ], + scroll_id="abc", + total=50, + ), + ) + mocker.patch.object(demisto, "getLastRun", return_value={"last_rr_fetch": _hours_ago(3, "")}) + demisto_incidents = mocker.patch.object(demisto, "incidents") + mocker.patch.object(demisto, "setLastRun") + + fetch_restore_requests(client, {"first_fetch": "1 day", "saas_apps": "Microsoft Exchange", "max_fetch": "1"}) + + call_api.assert_called_once() + assert [incident["dbotMirrorId"] for incident in demisto_incidents.call_args[0][0]] == ["first"] + + def test_checkpointhec_get_entity_success(mocker): client = Client( base_url="https://smart-api-example-1-us.avanan-example.net", diff --git a/Packs/CheckPointHEC/ReleaseNotes/1_1_17.md b/Packs/CheckPointHEC/ReleaseNotes/1_1_17.md new file mode 100644 index 00000000000..e9c86e5df96 --- /dev/null +++ b/Packs/CheckPointHEC/ReleaseNotes/1_1_17.md @@ -0,0 +1,5 @@ +#### Integrations + +##### Check Point Harmony Email and Collaboration (HEC) + +- Fixed an issue where restore requests could be missed when the fetch cursor advanced past records that were not yet available in the search index. \ No newline at end of file diff --git a/Packs/CheckPointHEC/pack_metadata.json b/Packs/CheckPointHEC/pack_metadata.json index 4e94756d618..2e053981030 100644 --- a/Packs/CheckPointHEC/pack_metadata.json +++ b/Packs/CheckPointHEC/pack_metadata.json @@ -2,7 +2,7 @@ "name": "Check Point Harmony Email and Collaboration (HEC)", "description": "The Best Way to Protect Enterprise Email & Collaboration from phishing, malware, account takeover, data loss, etc.", "support": "partner", - "currentVersion": "1.1.16", + "currentVersion": "1.1.17", "author": "Check Point Harmony Email & Collaboration (HEC)", "url": "https://supportcenter.checkpoint.com/", "email": "EmailSecurity_Support@checkpoint.com",