m365_defender,microsoft_defender_endpoint: fix handling of empty and error results in vulnerability data stream

packages/m365_defender/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "3.14.1"
+  changes:
+    - description: Fix agent handling of empty and error results in vulnerability data stream.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/14783
 - version: "3.14.0"
   changes:
     - description: |

packages/m365_defender/data_stream/vulnerability/agent/stream/cel.yml.hbs

@@ -29,55 +29,49 @@ auth.oauth2:
 {{/if}}
 
 state:
-  product_batch_size: 10000
+  config:
+    product_batch_size: 10000
+    machine_batch_size: 10000
+    vulnerabilities_batch_size: {{batch_size}}  
+    affected_machines_only: {{affected_machines_only}}
   product_skip: 0
-  batch_size: {{batch_size}}  
-  skip: 0
-  machine_batch_size: 10000
   machine_skip: 0
-  affected_machines_only: {{affected_machines_only}}
+  vulnerability_skip: 0
 redact:
   fields: ~
 program: |-
   state.with(
     (
+      // Get products.
       state.?is_all_products_fetched.orValue(false) ?
         {
           "products": state.products,
-          "product_batch_size": state.product_batch_size,
           "product_skip": 0,
           "is_all_products_fetched": state.is_all_products_fetched,
           ?"machines": state.?machines,
-          "machine_batch_size": state.machine_batch_size,
           "machine_skip": state.machine_skip,
           ?"is_all_machines_fetched": state.?is_all_machines_fetched,
           ?"vulnerabilities": state.?vulnerabilities,
-          "batch_size": state.batch_size,
-          "skip": state.skip,
+          "vulnerability_skip": state.vulnerability_skip,
           ?"is_all_vulnerabilities_fetched": state.?is_all_vulnerabilities_fetched,
-          "affected_machines_only": state.affected_machines_only,
         }
       :
         request(
           "GET",
           state.url.trim_right("/") + "/api/vulnerabilities/machinesVulnerabilities?" + {
-            "$top": [string(state.product_batch_size)],
+            "$top": [string(state.config.product_batch_size)],
             "$skip": [string(state.product_skip)],
           }.format_query()
         ).do_request().as(productResp, (productResp.StatusCode == 200) ?
           productResp.Body.decode_json().as(productBody,
             {
               "events": [{"message": "retry"}],
-              "products": (state.?products.orValue([]) + productBody.value).flatten(),
-              "product_batch_size": state.product_batch_size,
-              "product_skip": (size(productBody.value) > 0) ? (int(state.product_skip) + int(state.product_batch_size)) : 0,
-              "is_all_products_fetched": size(productBody.value) < int(state.product_batch_size),
               "want_more": true,
-              "machine_batch_size": state.machine_batch_size,
+              "products": (state.?products.orValue([]) + productBody.value).flatten(),
+              "product_skip": (size(productBody.value) > 0) ? (int(state.product_skip) + int(state.config.product_batch_size)) : 0,
+              "is_all_products_fetched": size(productBody.value) < int(state.config.product_batch_size),
               "machine_skip": state.machine_skip,
-              "batch_size": state.batch_size,
-              "skip": state.skip,
-              "affected_machines_only": state.affected_machines_only,
+              "vulnerability_skip": state.vulnerability_skip,
             }
           )
         :
@@ -102,51 +96,43 @@ program: |-
             "machine_skip": 0,
             "is_all_machines_fetched": false,
             "vulnerabilities": [],
-            "skip": 0,
+            "vulnerability_skip": 0,
             "is_all_vulnerabilities_fetched": false,
           }
         )
-    ).as(products, !products.?is_all_products_fetched.orValue(false) ?
-      products
-    : products.?is_all_machines_fetched.orValue(false) ?
+    ).as(res, !res.?is_all_products_fetched.orValue(false) ?
+      res
+    : res.?is_all_machines_fetched.orValue(false) ?
       {
-        "products": products.products,
-        "product_batch_size": products.product_batch_size,
+        "products": res.products,
         "product_skip": 0,
-        "is_all_products_fetched": products.is_all_products_fetched,
-        "machines": products.machines,
-        "machine_batch_size": products.machine_batch_size,
+        "is_all_products_fetched": res.is_all_products_fetched,
+        "machines": res.machines,
         "machine_skip": 0,
-        "is_all_machines_fetched": products.is_all_machines_fetched,
-        ?"vulnerabilities": products.?vulnerabilities,
-        "batch_size": products.batch_size,
-        "skip": products.skip,
-        ?"is_all_vulnerabilities_fetched": products.?is_all_vulnerabilities_fetched,
-        "affected_machines_only": products.affected_machines_only,
+        "is_all_machines_fetched": res.is_all_machines_fetched,
+        ?"vulnerabilities": res.?vulnerabilities,
+        "vulnerability_skip": res.vulnerability_skip,
+        ?"is_all_vulnerabilities_fetched": res.?is_all_vulnerabilities_fetched,
       }
     :
       request(
         "GET",
         state.url.trim_right("/") + "/api/machines?" + {
-          "$top": [string(products.machine_batch_size)],
-          "$skip": [string(products.machine_skip)],
+          "$top": [string(state.config.machine_batch_size)],
+          "$skip": [string(res.machine_skip)],
         }.format_query()
       ).do_request().as(machineResp, (machineResp.StatusCode == 200) ?
         machineResp.Body.decode_json().as(machineBody,
           {
             "events": [{"message": "retry"}],
-            "machines": (products.?machines.orValue([]) + machineBody.value).flatten(),
-            "machine_batch_size": products.machine_batch_size,
-            "machine_skip": (size(machineBody.value) > 0) ? (int(products.machine_skip) + int(products.machine_batch_size)) : 0,
-            "is_all_machines_fetched": size(machineBody.value) < int(products.machine_batch_size),
             "want_more": true,
-            "products": products.products,
-            "product_batch_size": products.product_batch_size,
+            "machines": (res.?machines.orValue([]) + machineBody.value).flatten(),
+            "machine_skip": (size(machineBody.value) > 0) ? (int(res.machine_skip) + int(state.config.machine_batch_size)) : 0,
+            "is_all_machines_fetched": size(machineBody.value) < int(state.config.machine_batch_size),
+            "products": res.products,
             "product_skip": 0,
-            "is_all_products_fetched": products.is_all_products_fetched,
-            "batch_size": products.batch_size,
-            "skip": products.skip,
-            "affected_machines_only": products.affected_machines_only,
+            "is_all_products_fetched": res.is_all_products_fetched,
+            "vulnerability_skip": res.vulnerability_skip,
           }
         )
       :
@@ -171,116 +157,117 @@ program: |-
           "machine_skip": 0,
           "is_all_machines_fetched": false,
           "vulnerabilities": [],
-          "skip": 0,
+          "vulnerability_skip": 0,
           "is_all_vulnerabilities_fetched": false,
         }
       )
-    ).as(products_with_machines, !products_with_machines.?is_all_machines_fetched.orValue(false) ?
-      products_with_machines
-    : products_with_machines.?is_all_vulnerability_fetched.orValue(false) ?
-      {
-        "products": products_with_machines.products,
-        "product_batch_size": products_with_machines.product_batch_size,
-        "product_skip": 0,
-        "is_all_products_fetched": products_with_machines.is_all_products_fetched,
-        "machines": products_with_machines.machines,
-        "machine_batch_size": products_with_machines.machine_batch_size,
-        "machine_skip": 0,
-        "is_all_machines_fetched": products_with_machines.is_all_machines_fetched,
-        "vulnerabilities": products_with_machines.vulnerabilities,
-        "batch_size": products_with_machines.batch_size,
-        "skip": 0,
-        "is_all_vulnerability_fetched": products_with_machines.is_all_vulnerability_fetched,
-        "affected_machines_only": products_with_machines.affected_machines_only,
-      }
-    :
-      request(
-        "GET",
-        state.url.trim_right("/") + "/api/vulnerabilities?" + {
-          "$top": [string(products_with_machines.batch_size)],
-          "$skip": [string(products_with_machines.skip)],
-        }.format_query()
-      ).do_request().as(vulnerabilityResp, (vulnerabilityResp.StatusCode == 200) ?
-        vulnerabilityResp.Body.decode_json().as(vulnerabilityBody,
+    ).as(res,
+      // Get products with machines.
+      !res.?is_all_machines_fetched.orValue(false) ?
+        res
+      : res.?is_all_vulnerability_fetched.orValue(false) ?
+        {
+          "products": res.products,
+          "product_skip": 0,
+          "is_all_products_fetched": res.is_all_products_fetched,
+          "machines": res.machines,
+          "machine_skip": 0,
+          "is_all_machines_fetched": res.is_all_machines_fetched,
+          "vulnerabilities": res.vulnerabilities,
+          "vulnerability_skip": 0,
+          "is_all_vulnerability_fetched": res.is_all_vulnerability_fetched,
+        }
+      :
+        request(
+          "GET",
+          state.url.trim_right("/") + "/api/vulnerabilities?" + {
+            "$top": [string(state.config.vulnerabilities_batch_size)],
+            "$skip": [string(res.vulnerability_skip)],
+          }.format_query()
+        ).do_request().as(vulnerabilityResp, (vulnerabilityResp.StatusCode == 200) ?
+          vulnerabilityResp.Body.decode_json().as(vulnerabilityBody,
+            {
+              "events": [{"message": "retry"}],
+              "want_more": true,
+              "vulnerabilities": (res.?vulnerabilities.orValue([]) + vulnerabilityBody.value).flatten(),
+              "vulnerability_skip": (size(vulnerabilityBody.value) > 0) ? (int(res.vulnerability_skip) + int(state.config.vulnerabilities_batch_size)) : 0,
+              "is_all_vulnerabilities_fetched": size(vulnerabilityBody.value) < int(state.config.vulnerabilities_batch_size),
+              "products": res.products,
+              "product_skip": 0,
+              "is_all_products_fetched": res.is_all_products_fetched,
+              "machines": res.machines,
+              "machine_skip": 0,
+              "is_all_machines_fetched": res.is_all_machines_fetched,
+            }
+          )
+        :
           {
-            "events": [{"message": "retry"}],
-            "vulnerabilities": (products_with_machines.?vulnerabilities.orValue([]) + vulnerabilityBody.value).flatten(),
-            "batch_size": state.batch_size,
-            "skip": (size(vulnerabilityBody.value) > 0) ? (int(products_with_machines.skip) + int(products_with_machines.batch_size)) : 0,
-            "is_all_vulnerabilities_fetched": size(vulnerabilityBody.value) < int(products_with_machines.batch_size),
-            "want_more": true,
-            "products": products_with_machines.products,
-            "product_batch_size": products_with_machines.product_batch_size,
+            "events": {
+              "error": {
+                "code": string(vulnerabilityResp.StatusCode),
+                "id": string(vulnerabilityResp.Status),
+                "message": "GET " + state.url.trim_right("/") + "/api/vulnerabilities" + (
+                  (size(vulnerabilityResp.Body) != 0) ?
+                    string(vulnerabilityResp.Body)
+                  :
+                    string(vulnerabilityResp.Status) + " (" + string(vulnerabilityResp.StatusCode) + ")"
+                ),
+              },
+            },
+            "want_more": false,
+            "products": [],
             "product_skip": 0,
-            "is_all_products_fetched": products_with_machines.is_all_products_fetched,
-            "machines": products_with_machines.machines,
-            "machine_batch_size": products_with_machines.machine_batch_size,
+            "is_all_products_fetched": false,
+            "machines": [],
             "machine_skip": 0,
-            "is_all_machines_fetched": products_with_machines.is_all_machines_fetched,
-            "affected_machines_only": products_with_machines.affected_machines_only,
+            "is_all_machines_fetched": false,
+            "vulnerabilities": [],
+            "vulnerability_skip": 0,
+            "is_all_vulnerabilities_fetched": false,
           }
         )
+    ).as(res,
+      // Collate data.
+      (!res.?is_all_vulnerabilities_fetched.orValue(false) || size(res.products) == 0) ?
+        res
       :
-        {
-          "events": {
-            "error": {
-              "code": string(vulnerabilityResp.StatusCode),
-              "id": string(vulnerabilityResp.Status),
-              "message": "GET " + state.url.trim_right("/") + "/api/vulnerabilities" + (
-                (size(vulnerabilityResp.Body) != 0) ?
-                  string(vulnerabilityResp.Body)
-                :
-                  string(vulnerabilityResp.Status) + " (" + string(vulnerabilityResp.StatusCode) + ")"
-              ),
-            },
-          },
-          "want_more": false,
-          "products": [],
-          "product_skip": 0,
-          "is_all_products_fetched": false,
-          "machines": [],
-          "machine_skip": 0,
-          "is_all_machines_fetched": false,
-          "vulnerabilities": [],
-          "skip": 0,
-          "is_all_vulnerabilities_fetched": false,
-        }
-      )
-    ).as(all_data, !all_data.?is_all_vulnerabilities_fetched.orValue(false) ?
-      all_data
-    :
-      all_data.products.map(p, all_data.machines.filter(m, m.id == p.machineId)[0].with(p)).as(mapped_products,
-        {
-          "vulnerability_with_machines": all_data.vulnerabilities.filter(v, v.exposedMachines > 0),
-          "vulnerability_without_machines": !all_data.affected_machines_only ?
-            all_data.vulnerabilities.filter(v, v.exposedMachines == 0)
+        res.products.map(p,
+          res.machines.filter(m, m.id == p.machineId)[?0].as(m, m.hasValue() ?
+            m.value().with(p)
           :
-            [],
-          "mapped_products": mapped_products,
-        }
-      ).as(final_data,
-        {
-          "events": (
-            final_data.vulnerability_with_machines.map(v,
-              final_data.mapped_products.filter(mp, mp.cveId == v.id).map(related_mapped_products,
-                {"message": v.with({"affectedMachine": related_mapped_products}).encode_json()}
+            {}
+          )
+        ).as(mapped_products,
+          {
+            "vulnerability_with_machines": res.vulnerabilities.filter(v, v.exposedMachines > 0),
+            "vulnerability_without_machines": state.config.affected_machines_only ?
+              []
+            :
+              res.vulnerabilities.filter(v, v.exposedMachines == 0),
+            "mapped_products": mapped_products,
+          }
+        ).as(final_data,
+          {
+            "events": (
+              final_data.vulnerability_with_machines.map(v,
+                final_data.mapped_products.map(related_mapped_products,
+                  has(related_mapped_products.cveId) && related_mapped_products.cveId == v.id,
+                  {
+                    "message": v.with({"affectedMachine": related_mapped_products}).encode_json(),
+                  }
+                )
+              ).flatten() + final_data.vulnerability_without_machines.map(v,
+                {
+                  "message": v.drop("affectedMachine").encode_json(),
+                }
               )
-            ).flatten() + final_data.vulnerability_without_machines.map(v,
-              {
-                "message": v.with({"affectedMachine": null}).encode_json(),
-              }
-            )
-          ).flatten(),
-          "product_batch_size": all_data.product_batch_size,
-          "product_skip": 0,
-          "machine_batch_size": all_data.machine_batch_size,
-          "machine_skip": 0,
-          "batch_size": all_data.batch_size,
-          "skip": 0,
-          "affected_machines_only": all_data.affected_machines_only,
-          "want_more": false,
-        }
-      )
+            ).flatten(),
+            "want_more": false,
+            "product_skip": 0,
+            "machine_skip": 0,
+            "vulnerability_skip": 0,
+          }
+        )
     )
   )
 tags:

packages/m365_defender/manifest.yml

@@ -1,7 +1,7 @@
 format_version: "3.4.0"
 name: m365_defender
 title: Microsoft Defender XDR
-version: "3.14.0"
+version: "3.14.1"
 description: Collect logs from Microsoft Defender XDR with Elastic Agent.
 categories:
   - "security"

packages/microsoft_defender_endpoint/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "2.42.1"
+  changes:
+    - description: Fix agent handling of empty and error results in vulnerability data stream.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/14783
 - version: "2.42.0"
   changes:
     - description: Add `process.name` ECS mapping in log data stream.