{microsoft_defender_endpoint, m365_defender}.vulnerability: New API implementation

packages/m365_defender/_dev/deploy/docker/docker-compose.yml

@@ -32,6 +32,7 @@ services:
       - 8080
     volumes:
       - ./vulnerability-http-mock-config.yml:/config.yml
+      - ./download-vulnerability.log.gz:/download-vulnerability.log.gz
     environment:
       PORT: 8080
     command:

packages/m365_defender/changelog.yml

@@ -1,4 +1,16 @@
 # newer versions go on top
+- version: "5.0.0"
+  changes:
+    - description: |
+        Fetch vulnerability data using SoftwareVulnerabilitiesExport API endpoint.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/15603
+    - description: |
+        The following fields are no longer available in the new implementation: "cloud.provider", "cloud.resource_id", 
+        "cloud.instance.id", "host.geo", "host.ip", "host.risk.calculated_level", "related.ip", 
+        "vulnerability.description", "vulnerability.published_date", "vulnerability.score.version".
+      type: breaking-change
+      link: https://github.com/elastic/integrations/pull/15603
 - version: "4.2.0"
   changes:
     - description: Prevent updating fleet health status to degraded.

packages/m365_defender/data_stream/vulnerability/_dev/test/system/test-default-config.yml

@@ -8,8 +8,9 @@ vars:
   azure_tenant_id: tenant_id
 data_stream:
   vars:
-    batch_size: 2
+    sas_valid_hours: 2h
     preserve_original_event: true
     preserve_duplicate_custom_fields: true
+    enable_request_tracer: true
 assert:
-  hit_count: 5
+  hit_count: 4

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

@@ -14,261 +14,123 @@ resource.ssl: {{ssl}}
 resource.timeout: {{http_client_timeout}}
 {{/if}}
 resource.url: {{url}}
-auth.oauth2:
-  provider: azure
-  client.id: {{client_id}}
-  client.secret: {{client_secret}}
-  scopes:
+state:
+  sas_valid_hours: {{sas_valid_hours}}
+  token_url: {{token_url}}/{{azure_tenant_id}}/oauth2/v2.0/token
+  client_id: {{client_id}}
+  client_secret: {{client_secret}}
+  token_scopes:
 {{#each token_scopes as |token_scope|}}
     - {{token_scope}}
 {{/each}}
-{{#if token_url}}
-  token_url: {{token_url}}/{{azure_tenant_id}}/oauth2/v2.0/token
-{{else if azure_tenant_id}}
-  azure.tenant_id: {{azure_tenant_id}}
-{{/if}}
-
-state:
-  config:
-    product_batch_size: 10000
-    machine_batch_size: 10000
-    vulnerabilities_batch_size: {{batch_size}}  
-    affected_machines_only: {{affected_machines_only}}
-  product_skip: 0
-  machine_skip: 0
-  vulnerability_skip: 0
 redact:
-  fields: ~
+  fields:
+    - client_id
+    - client_secret
+    - token.access_token
 program: |-
   state.with(
-    (
-      // Get products.
-      state.?is_all_products_fetched.orValue(false) ?
-        {
-          "products": state.products,
-          "product_skip": 0,
-          "is_all_products_fetched": state.is_all_products_fetched,
-          ?"machines": state.?machines,
-          "machine_skip": state.machine_skip,
-          ?"is_all_machines_fetched": state.?is_all_machines_fetched,
-          ?"vulnerabilities": state.?vulnerabilities,
-          "vulnerability_skip": state.vulnerability_skip,
-          ?"is_all_vulnerabilities_fetched": state.?is_all_vulnerabilities_fetched,
-        }
-      :
-        request(
-          "GET",
-          state.url.trim_right("/") + "/api/vulnerabilities/machinesVulnerabilities?" + {
-            "$top": [string(state.config.product_batch_size)],
-            "$skip": [string(int(state.product_skip))],
-          }.format_query()
-        ).do_request().as(productResp, (productResp.StatusCode == 200) ?
-          productResp.Body.decode_json().as(productBody,
-            {
-              "events": [{"message": "retry"}],
-              "want_more": true,
-              "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,
-              "vulnerability_skip": state.vulnerability_skip,
-            }
-          )
-        :
-          {
-            "events": {
-              "error": {
-                "code": string(productResp.StatusCode),
-                "id": string(productResp.Status),
-                "message": "GET " + state.url.trim_right("/") + "/api/vulnerabilities/machinesVulnerabilities" + (
-                  (size(productResp.Body) != 0) ?
-                    string(productResp.Body)
-                  :
-                    string(productResp.Status) + " (" + string(productResp.StatusCode) + ")"
-                ),
-              },
-            },
-            "want_more": false,
-            "products": [],
-            "product_skip": 0,
-            "is_all_products_fetched": false,
-            "machines": [],
-            "machine_skip": 0,
-            "is_all_machines_fetched": false,
-            "vulnerabilities": [],
-            "vulnerability_skip": 0,
-            "is_all_vulnerabilities_fetched": false,
-          }
-        )
-    ).as(res, !res.?is_all_products_fetched.orValue(false) ?
-      res
-    : res.?is_all_machines_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": res.vulnerability_skip,
-        ?"is_all_vulnerabilities_fetched": res.?is_all_vulnerabilities_fetched,
-      }
-    :
+    state.?work_list.orValue([]).size() > 0 ?
       request(
-        "GET",
-        state.url.trim_right("/") + "/api/machines?" + {
-          "$top": [string(state.config.machine_batch_size)],
-          "$skip": [string(int(res.machine_skip))],
-        }.format_query()
-      ).do_request().as(machineResp, (machineResp.StatusCode == 200) ?
-        machineResp.Body.decode_json().as(machineBody,
-          {
-            "events": [{"message": "retry"}],
-            "want_more": true,
-            "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": res.is_all_products_fetched,
-            "vulnerability_skip": res.vulnerability_skip,
-          }
-        )
+        "GET", 
+        state.work_list[0]
+      ).do_request().as(resp, resp.StatusCode == 200 ?
+        resp.Body.mime("application/gzip").decode_json_stream().map(v,
+          {"message": dyn(v.encode_json())}
+        ).as(events, {
+          "events": events,
+          // Keep polling if more work.
+          "want_more": state.work_list.size() > 1,
+          "work_list": tail(state.work_list),
+        })
       :
+        // It is possible that download URLs have expired, so ignore remaining work_list and return error.
         {
           "events": {
             "error": {
-              "code": string(machineResp.StatusCode),
-              "id": string(machineResp.Status),
-              "message": "GET " + state.url.trim_right("/") + "/api/machines" + (
-                (size(machineResp.Body) != 0) ?
-                  string(machineResp.Body)
+              "code": string(resp.StatusCode),
+              "id": string(resp.Status),
+              "message": "GET "+ state.work_list[0] + ":" + (
+                size(resp.Body) != 0 ?
+                  string(resp.Body)
                 :
-                  string(machineResp.Status) + " (" + string(machineResp.StatusCode) + ")"
+                  string(resp.Status) + ' (' + string(resp.StatusCode) + ')'
               ),
             },
           },
           "want_more": false,
-          "products": [],
-          "product_skip": 0,
-          "is_all_products_fetched": false,
-          "machines": [],
-          "machine_skip": 0,
-          "is_all_machines_fetched": false,
-          "vulnerabilities": [],
-          "vulnerability_skip": 0,
-          "is_all_vulnerabilities_fetched": false,
         }
       )
-    ).as(res,
-      // Get products with machines.
-      !res.?is_all_machines_fetched.orValue(false) ?
-        res
-      : res.?is_all_vulnerability_fetched.orValue(false) ?
+    :
+      // Periodic poll. No work_list, so get new token and work_list.
+      post_request(state.token_url.trim_right("/"), "application/x-www-form-urlencoded",
         {
-          "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,
-        }
+          "grant_type": ["client_credentials"],
+          "client_id": [state.client_id],
+          "client_secret": [state.client_secret],
+          "scope": state.token_scopes,
+        }.format_query()
+      ).do_request().as(auth, auth.StatusCode == 200 ?
+        auth.Body.decode_json()
       :
+        {
+          "events": {
+            "error": {
+              "code": string(auth.StatusCode),
+              "id": string(auth.Status),
+              "message": "POST /oauth2/v2.0/token :" +(
+                size(auth.Body) != 0 ?
+                  string(auth.Body)
+                :
+                  string(auth.Status) + ' (' + string(auth.StatusCode) + ')'
+              ),
+            },
+          },
+          "want_more": false,
+        }
+      ).as(token, !has(token.access_token) ? token :
         request(
           "GET",
-          state.url.trim_right("/") + "/api/vulnerabilities?" + {
-            "$top": [string(state.config.vulnerabilities_batch_size)],
-            "$skip": [string(int(res.vulnerability_skip))],
+          state.url.trim_right("/") + "/api/machines/SoftwareVulnerabilitiesExport?" + {
+            "sasValidHours": [string(duration(state.sas_valid_hours).getHours())],
           }.format_query()
-        ).do_request().as(vulnerabilityResp, (vulnerabilityResp.StatusCode == 200) ?
-          vulnerabilityResp.Body.decode_json().as(vulnerabilityBody,
+        ).with({
+          "Header":{
+            "Authorization": ["Bearer " + string(token.access_token)],
+          }
+        }).do_request().as(resp, resp.StatusCode == 200 ?
+          resp.Body.decode_json().as(exportBody, exportBody.?exportFiles.orValue([]).size() == 0 ?
+            // Nothing to download. Don't poll again.
+            {
+              "events": [],
+              "want_more": false,
+            }
+          :
+            // Return new work_list to download.
             {
-              "events": [{"message": "retry"}],
+              "events": [{"message":"retry"}],
+              "work_list": exportBody.exportFiles,
               "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": {
               "error": {
-                "code": string(vulnerabilityResp.StatusCode),
-                "id": string(vulnerabilityResp.Status),
-                "message": "GET " + state.url.trim_right("/") + "/api/vulnerabilities" + (
-                  (size(vulnerabilityResp.Body) != 0) ?
-                    string(vulnerabilityResp.Body)
+                "code": string(resp.StatusCode),
+                "id": string(resp.Status),
+                "message": "GET /api/machines/SoftwareVulnerabilitiesExport :" + (
+                  size(resp.Body) != 0 ?
+                    string(resp.Body)
                   :
-                    string(vulnerabilityResp.Status) + " (" + string(vulnerabilityResp.StatusCode) + ")"
+                    string(resp.Status) + ' (' + string(resp.StatusCode) + ')'
                 ),
               },
             },
             "want_more": false,
-            "products": [],
-            "product_skip": 0,
-            "is_all_products_fetched": false,
-            "machines": [],
-            "machine_skip": 0,
-            "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
-      :
-        res.products.map(p,
-          res.machines.filter(m, m.id == p.machineId)[?0].as(m, m.hasValue() ?
-            m.value().with(p)
-          :
-            {}
-          )
-        ).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(),
-            "want_more": false,
-            "product_skip": 0,
-            "machine_skip": 0,
-            "vulnerability_skip": 0,
           }
         )
-    )
+      )
   )
 tags:
 {{#if preserve_original_event}}