[abnormal_security] Update pagination termination condition (#10860)

Update the pagination termination condition for the threat data stream so that pagination stops whenever nextPageNumber is not found in the response body. This should apply to both the /v1/threats and /v1/threats/:id endpoints.
elastic · Aug 27, 2024 · c577fbb · c577fbb
1 parent db5288a
commit c577fbb
Show file tree

Hide file tree

Showing 7 changed files with 33 additions and 86 deletions.
diff --git a/packages/abnormal_security/_dev/deploy/docker/files/config.yml b/packages/abnormal_security/_dev/deploy/docker/files/config.yml
@@ -323,23 +323,7 @@ rules:
           Content-Type:
             - 'application/json'
         body: |-
-          {"threats":[{"threatId":"284712ab-6d8b-47b3-89d3-a314efef79e2"}],"pageNumber":2,"nextPageNumber":3}
-  - path: /v1/threats
-    methods: ['GET']
-    query_params:
-      filter: "{filter:.*}"
-      pageNumber: 3
-      pageSize: 1
-    request_headers:
-      Authorization:
-        - "Bearer xxxx"
-    responses:
-      - status_code: 200
-        headers:
-          Content-Type:
-            - 'application/json'
-        body: |-
-          {"threats":[],"pageNumber":3,"nextPageNumber":4}
+          {"threats":[{"threatId":"284712ab-6d8b-47b3-89d3-a314efef79e2"}]}
   - path: /v1/threats/184712ab-6d8b-47b3-89d3-a314efef79e2
     methods: ['GET']
     query_params:
@@ -402,26 +386,9 @@ rules:
                     "https://www.example.com/"
                   ]
                 }
-              ],
-              "pageNumber": 1,
-              "nextPageNumber": 2
+              ]
           }
           `}}
-  - path: /v1/threats/184712ab-6d8b-47b3-89d3-a314efef79e2
-    methods: ['GET']
-    query_params:
-      pageNumber: 2
-      pageSize: 1
-    request_headers:
-      Authorization:
-        - "Bearer xxxx"
-    responses:
-      - status_code: 200
-        headers:
-          Content-Type:
-            - 'application/json'
-        body: |-
-          {"threatId":"184712ab-6d8b-47b3-89d3-a314efef79e2","messages":[],"pageNumber":2,"nextPageNumber":3}
   - path: /v1/threats/284712ab-6d8b-47b3-89d3-a314efef79e2
     methods: ['GET']
     query_params:
@@ -483,23 +450,6 @@ rules:
                         "https://www.example.com/"
                     ]
                 }
-              ],
-              "pageNumber": 1,
-              "nextPageNumber": 2
+              ]
           }
           `}}
-  - path: /v1/threats/284712ab-6d8b-47b3-89d3-a314efef79e2
-    methods: ['GET']
-    query_params:
-      pageNumber: 2
-      pageSize: 1
-    request_headers:
-      Authorization:
-        - "Bearer xxxx"
-    responses:
-      - status_code: 200
-        headers:
-          Content-Type:
-            - 'application/json'
-        body: |-
-          {"threatId":"284712ab-6d8b-47b3-89d3-a314efef79e2","messages":[],"pageNumber":2,"nextPageNumber":3}
diff --git a/packages/abnormal_security/changelog.yml b/packages/abnormal_security/changelog.yml
@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "0.1.1"
+  changes:
+    - description: Update pagination termination condition in threat data stream.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/10860
 - version: "0.1.0"
   changes:
     - description: Initial release.

diff --git a/packages/abnormal_security/data_stream/threat/agent/stream/cel.yml.hbs b/packages/abnormal_security/data_stream/threat/agent/stream/cel.yml.hbs
@@ -50,7 +50,7 @@ program: |
           }
         }).do_request().as(resp, resp.StatusCode == 200 ?
             bytes(resp.Body).decode_json().as(body, {
-              "worklist": body.threats.map(e, e.threatId),
+              "worklist": body,
               "next": 0,
             })
           :
@@ -73,9 +73,9 @@ program: |
       ))
   ).as(state, state.with(
     !has(state.worklist) ? state : // Exit early due to GET failure.
-      state.next < size(state.worklist) ?
+      has(state.worklist.threats) && size(state.worklist.threats) > 0 ?
         request("GET",
-          state.url.trim_right("/") + "/v1/threats/" + string(state.worklist[state.next]) + "?" + {
+          state.url.trim_right("/") + "/v1/threats/" + string(state.worklist.threats[state.next].threatId) + "?" + {
             "pageSize": [string(state.page_size)],
             "pageNumber": [string(state.child_next_page)]
           }.format_query()
@@ -85,30 +85,25 @@ program: |
           }
         }).do_request().as(resp, resp.StatusCode == 200 ?
           bytes(resp.Body).decode_json().as(body,{
-            "events": (
-              size(body.messages) > 0 ?
-                body.messages.map(e,{
-                  "message": e.encode_json(),
-                })
-              :
-                [{"message":"retry"}]
-            ),
+            "events": body.messages.map(e,{
+              "message": e.encode_json(),
+            }),
             "cursor": {
               "last_timestamp": state.end_time
             },
-            "worklist": int(state.next) + 1 < size(state.worklist) || size(body.messages) > 0 ? state.worklist : [],
-            "child_next_page": size(body.messages) > 0 ? int(state.child_next_page) + 1 : 1,
+            "worklist": int(state.next) + 1 < size(state.worklist.threats) || has(body.nextPageNumber) ? state.worklist : {},
+            "child_next_page": has(body.nextPageNumber) ? body.nextPageNumber : 1,
             "next": (
-              size(body.messages) > 0 ?
+              has(body.nextPageNumber) ?
                 state.next
               :
-                int(state.next) + 1 < size(state.worklist) ?
+                int(state.next) + 1 < size(state.worklist.threats) ?
                   int(state.next) + 1
                 :
                   0
             ),
-            "next_page": int(state.next) + 1 < size(state.worklist) || size(body.messages) > 0 ? state.next_page : int(state.next_page) + 1,
-            "want_more": true,
+            "next_page": int(state.next) + 1 < size(state.worklist.threats) || has(body.nextPageNumber) ? state.next_page : has(state.worklist.nextPageNumber) ? state.worklist.nextPageNumber : 1,
+            "want_more": int(state.next) + 1 < size(state.worklist.threats) || has(body.nextPageNumber) || has(state.worklist.nextPageNumber),
           })
         :
           {

diff --git a/packages/abnormal_security/data_stream/threat/elasticsearch/ingest_pipeline/default.yml b/packages/abnormal_security/data_stream/threat/elasticsearch/ingest_pipeline/default.yml
@@ -9,9 +9,6 @@ processors:
       tag: data_collection_error
       if: ctx.error?.message != null && ctx.message == null && ctx.event?.original == null
       message: error message set and no data to process.
-  - drop:
-      if: ctx.message == 'retry'
-      tag: drop_retry_events
   - rename:
       field: message
       tag: rename_message_to_event_original

diff --git a/packages/abnormal_security/data_stream/threat/sample_event.json b/packages/abnormal_security/data_stream/threat/sample_event.json
@@ -45,22 +45,22 @@
         }
     },
     "agent": {
-        "ephemeral_id": "3cfaa9dc-bca8-4e29-a807-77b68709b731",
-        "id": "7aaba523-565c-4597-bc42-59135436336b",
-        "name": "docker-fleet-agent",
+        "ephemeral_id": "b66f399f-ba1c-4fe5-af82-9ca7a0204545",
+        "id": "e2eadaf0-613d-41d9-913c-96125e06487a",
+        "name": "elastic-agent-55334",
         "type": "filebeat",
         "version": "8.13.0"
     },
     "data_stream": {
         "dataset": "abnormal_security.threat",
-        "namespace": "37330",
+        "namespace": "45319",
         "type": "logs"
     },
     "ecs": {
         "version": "8.11.0"
     },
     "elastic_agent": {
-        "id": "7aaba523-565c-4597-bc42-59135436336b",
+        "id": "e2eadaf0-613d-41d9-913c-96125e06487a",
         "snapshot": false,
         "version": "8.13.0"
     },
@@ -88,7 +88,7 @@
         ],
         "dataset": "abnormal_security.threat",
         "id": "2260288475997441000",
-        "ingested": "2024-08-08T06:53:48Z",
+        "ingested": "2024-08-23T05:40:07Z",
         "kind": "enrichment",
         "original": "{\"abxMessageId\":2260288475997441000,\"abxPortalUrl\":\"https://portal.abnormalsecurity.com/home/threat-center/remediation-history/3456765434567654\",\"attachmentCount\":0,\"attachmentNames\":[],\"attackStrategy\":\"Unknown Sender\",\"attackType\":\"Spam\",\"attackVector\":\"Link\",\"attackedParty\":\"Employee (Other)\",\"autoRemediated\":true,\"ccEmails\":[],\"fromAddress\":\"[email protected]\",\"fromName\":\"john\",\"impersonatedParty\":\"None / Others\",\"internetMessageId\":\"\\u003cAZz8NUMEST-qmuz77_koic@example\\u003e\",\"isRead\":false,\"postRemediated\":false,\"receivedTime\":\"2024-07-17T23:25:38Z\",\"recipientAddress\":\"[email protected]\",\"remediationStatus\":\"Auto-Remediated\",\"remediationTimestamp\":\"2024-07-17T23:25:45.73564Z\",\"replyToEmails\":[],\"returnPath\":\"[email protected]\",\"senderDomain\":\"example.com\",\"senderIpAddress\":\"81.2.69.142\",\"sentTime\":\"2024-07-17T23:25:29Z\",\"subject\":\"YoU.have.𝗪𝟬0𝗡𝗡 a K0baIt 215-piece_ToooI_Set_Noo0wW..#GBOB\",\"summaryInsights\":[\"Abnormal Email Body HTML\",\"Invisible characters found in Email\",\"Suspicious Link\",\"Unusual Sender\",\"Unusual Sender Domain\"],\"threatId\":\"bf255f2d-a2ad-3f50-5075-fdcc24308bbd\",\"toAddresses\":[\"[email protected]\"],\"urlCount\":1,\"urls\":[\"https://www.example.com/\"]}",
         "reference": "https://portal.abnormalsecurity.com/home/threat-center/remediation-history/3456765434567654",

diff --git a/packages/abnormal_security/docs/README.md b/packages/abnormal_security/docs/README.md
@@ -498,22 +498,22 @@ An example event for `threat` looks as following:
         }
     },
     "agent": {
-        "ephemeral_id": "3cfaa9dc-bca8-4e29-a807-77b68709b731",
-        "id": "7aaba523-565c-4597-bc42-59135436336b",
-        "name": "docker-fleet-agent",
+        "ephemeral_id": "b66f399f-ba1c-4fe5-af82-9ca7a0204545",
+        "id": "e2eadaf0-613d-41d9-913c-96125e06487a",
+        "name": "elastic-agent-55334",
         "type": "filebeat",
         "version": "8.13.0"
     },
     "data_stream": {
         "dataset": "abnormal_security.threat",
-        "namespace": "37330",
+        "namespace": "45319",
         "type": "logs"
     },
     "ecs": {
         "version": "8.11.0"
     },
     "elastic_agent": {
-        "id": "7aaba523-565c-4597-bc42-59135436336b",
+        "id": "e2eadaf0-613d-41d9-913c-96125e06487a",
         "snapshot": false,
         "version": "8.13.0"
     },
@@ -541,7 +541,7 @@ An example event for `threat` looks as following:
         ],
         "dataset": "abnormal_security.threat",
         "id": "2260288475997441000",
-        "ingested": "2024-08-08T06:53:48Z",
+        "ingested": "2024-08-23T05:40:07Z",
         "kind": "enrichment",
         "original": "{\"abxMessageId\":2260288475997441000,\"abxPortalUrl\":\"https://portal.abnormalsecurity.com/home/threat-center/remediation-history/3456765434567654\",\"attachmentCount\":0,\"attachmentNames\":[],\"attackStrategy\":\"Unknown Sender\",\"attackType\":\"Spam\",\"attackVector\":\"Link\",\"attackedParty\":\"Employee (Other)\",\"autoRemediated\":true,\"ccEmails\":[],\"fromAddress\":\"[email protected]\",\"fromName\":\"john\",\"impersonatedParty\":\"None / Others\",\"internetMessageId\":\"\\u003cAZz8NUMEST-qmuz77_koic@example\\u003e\",\"isRead\":false,\"postRemediated\":false,\"receivedTime\":\"2024-07-17T23:25:38Z\",\"recipientAddress\":\"[email protected]\",\"remediationStatus\":\"Auto-Remediated\",\"remediationTimestamp\":\"2024-07-17T23:25:45.73564Z\",\"replyToEmails\":[],\"returnPath\":\"[email protected]\",\"senderDomain\":\"example.com\",\"senderIpAddress\":\"81.2.69.142\",\"sentTime\":\"2024-07-17T23:25:29Z\",\"subject\":\"YoU.have.𝗪𝟬0𝗡𝗡 a K0baIt 215-piece_ToooI_Set_Noo0wW..#GBOB\",\"summaryInsights\":[\"Abnormal Email Body HTML\",\"Invisible characters found in Email\",\"Suspicious Link\",\"Unusual Sender\",\"Unusual Sender Domain\"],\"threatId\":\"bf255f2d-a2ad-3f50-5075-fdcc24308bbd\",\"toAddresses\":[\"[email protected]\"],\"urlCount\":1,\"urls\":[\"https://www.example.com/\"]}",
         "reference": "https://portal.abnormalsecurity.com/home/threat-center/remediation-history/3456765434567654",

diff --git a/packages/abnormal_security/manifest.yml b/packages/abnormal_security/manifest.yml
@@ -1,7 +1,7 @@
 format_version: 3.2.1
 name: abnormal_security
 title: Abnormal Security
-version: 0.1.0
+version: 0.1.1
 description: Collect logs from Abnormal Security with Elastic Agent.
 type: integration
 categories: