[O365] - Prune stale listing URLs from cursor to recover from long outages

packages/o365/_dev/deploy/docker/config.yml

@@ -532,3 +532,125 @@ rules:
               }
           ]
           `}}
+  # Stale-NextPageUri test rules
+  # Exercise the pre-flight filter that drops queued listing links whose
+  # startTime has aged past maximum_age. The first page is fresh; the
+  # NextPageUri it returns has a hardcoded 2019 startTime so the filter
+  # must drop it on the next evaluation rather than follow it. The
+  # corresponding responder is defined below and is only reached if the
+  # filter regresses.
+  - path: /api/v1.0/test-cel-tenant-id/activity/feed/subscriptions/start
+    methods: [POST]
+    query_params:
+      contentType: "Audit.Exchange"
+      PublisherIdentifier: test-cel-tenant-id
+    request_headers:
+      Authorization:
+        - "Bearer CELtoken"
+    responses:
+      - status_code: 200
+        headers:
+          Content-Type:
+            - "application/json"
+        body: |-
+          {{ minify_json `
+          {
+              "contentType": "Audit.Exchange",
+              "status": "enabled",
+              "webhook": null
+          }
+          `}}
+  - path: /api/v1.0/test-cel-tenant-id/activity/feed/subscriptions/content
+    methods: [GET]
+    query_params:
+      contentType: "Audit.Exchange"
+      startTime: "{startTime:.*}"
+      endTime: "{endTime:.*}"
+      PublisherIdentifier: test-cel-tenant-id
+    request_headers:
+      Authorization:
+        - "Bearer CELtoken"
+    responses:
+      - status_code: 200
+        headers:
+          Content-Type:
+            - "application/json"
+          NextPageUri:
+            - 'http://{{ hostname }}:{{ env "PORT" }}/api/v1.0/test-cel-tenant-id/activity/feed/subscriptions/content?contenttype=Audit.Exchange&starttime=2019-01-01T00:00:00Z&endtime=2019-01-01T01:00:00Z&nextpage=aged-page-that-should-be-dropped'
+        body: |-
+          [{"contentType": "Audit.Exchange","contentId": "celexchfresh","contentUri": "http://{{ hostname }}:{{ env "PORT" }}/api/v1.0/celexch/activity/feed/audit/celexchfresh","contentCreated": "{{ .request.vars.endTime }}","contentExpiration": "2199-05-30T17:35:00.000Z"}]
+  # This path MUST NOT be called when the pre-flight filter is working:
+  # its startTime is from 2019 and falls outside any sensible maximum_age.
+  # If it is called, the aged-content responder below will serve an event
+  # and hit_count will be 2 instead of 1, failing the assertion.
+  - path: /api/v1.0/test-cel-tenant-id/activity/feed/subscriptions/content
+    methods: [GET]
+    query_params:
+      contenttype: "Audit.Exchange"
+      starttime: "2019-01-01T00:00:00Z"
+      endtime: "2019-01-01T01:00:00Z"
+      nextpage: "aged-page-that-should-be-dropped"
+    request_headers:
+      Authorization:
+        - "Bearer CELtoken"
+    responses:
+      - status_code: 200
+        headers:
+          Content-Type:
+            - "application/json"
+        body: |-
+          [{"contentType": "Audit.Exchange","contentId": "celexchstale","contentUri": "http://{{ hostname }}:{{ env "PORT" }}/api/v1.0/celexch/activity/feed/audit/celexchstale","contentCreated": "2019-01-01T00:30:00Z","contentExpiration": "2199-05-30T17:35:00.000Z"}]
+  - path: /api/v1.0/celexch/activity/feed/audit/celexchfresh
+    methods: [GET]
+    request_headers:
+      Authorization:
+        - "Bearer CELtoken"
+    responses:
+      - status_code: 200
+        headers:
+          Content-Type:
+            - "application/json"
+        body: |-
+          {{ minify_json `
+          [
+              {
+                  "CreationTime": "2020-02-07T16:43:53",
+                  "Id": "exch-fresh-event-1",
+                  "Operation": "Send",
+                  "OrganizationId": "b86ab9d4-fcf1-4b11-8a06-7a8f91b47fbd",
+                  "RecordType": 2,
+                  "UserKey": "fresh-key",
+                  "UserType": 0,
+                  "Version": 1,
+                  "Workload": "Exchange",
+                  "UserId": "fresh@testsiem.onmicrosoft.com"
+              }
+          ]
+          `}}
+  - path: /api/v1.0/celexch/activity/feed/audit/celexchstale
+    methods: [GET]
+    request_headers:
+      Authorization:
+        - "Bearer CELtoken"
+    responses:
+      - status_code: 200
+        headers:
+          Content-Type:
+            - "application/json"
+        body: |-
+          {{ minify_json `
+          [
+              {
+                  "CreationTime": "2019-01-01T00:30:00",
+                  "Id": "exch-stale-event-1",
+                  "Operation": "Send",
+                  "OrganizationId": "b86ab9d4-fcf1-4b11-8a06-7a8f91b47fbd",
+                  "RecordType": 2,
+                  "UserKey": "stale-key",
+                  "UserType": 0,
+                  "Version": 1,
+                  "Workload": "Exchange",
+                  "UserId": "stale@testsiem.onmicrosoft.com"
+              }
+          ]
+          `}}

packages/o365/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "3.8.1"
+  changes:
+    - description: Prune queued listing links whose startTime is already outside the API's accepted window so that prolonged periods without successful fetches (e.g. auth failures during a credential rotation, or agent downtime longer than seven days) self-heal on resume instead of looping on AF20055.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/18510
 - version: "3.8.0"
   changes:
     - description: Increase field limits to avoid unindexed ECS fields.

packages/o365/data_stream/audit/_dev/test/policy/test-cel-all.expected

@@ -17,7 +17,6 @@ inputs:
             token_url: https://login.microsoftonline.com/tenant-id/oauth2/v2.0/token
           data_stream:
             dataset: o365.audit
-            type: logs
           interval: 3m
           max_executions: 10000
           processors:
@@ -83,16 +82,44 @@ inputs:
                 }
               )
             ).as(state,
-              // Filter stale cursor data that references content types no longer in config.
-              // Listing links have the content type embedded in the URL query string, so
-              // links for removed types would produce API errors if not pruned here.
+              // Filter stale cursor data before acting on it.
               state.base.content_types.split(",").map(t, t.trim_space()).as(configured_types,
                 state.with(
                   {
                     "cursor": state.?cursor.orValue({}).with(
                       {
+                        // Listing links have the content type embedded in the URL
+                        // query string, so links for removed types would produce API
+                        // errors if not pruned here.
+                        //
+                        // Links whose startTime has aged past maximum_age are also
+                        // pruned. The Office 365 Management API rejects requests
+                        // whose startTime is more than 7 days in the past with
+                        // AF20055; a link queued before a prolonged period without
+                        // successful fetches (agent downtime, auth failures,
+                        // credential rotation gaps, etc.) may have aged past that
+                        // boundary while parked in the cursor. The listing-error
+                        // branch cannot evict it reliably because its map-shaped
+                        // error return is treated by the runtime as a freeze signal
+                        // that discards cursor mutations. Once pruned, the bottom
+                        // branch generates a fresh link clamped to the valid window
+                        // on the next iteration.
                         "todo_links": state.?cursor.todo_links.orValue([]).filter(link,
                           configured_types.exists(ct, link.to_lower().contains("contenttype=" + ct.to_lower()))
+                          &&
+                          link.parse_url().RawQuery.parse_query().as(q,
+                            // Listing URLs generated by this program use "startTime",
+                            // but NextPageUri values returned by the API have been
+                            // seen with the lowercase "starttime" variant (see
+                            // #15325), so accept either. If no startTime can be
+                            // extracted, keep the link and let the API itself decide.
+                            q.?startTime.orValue(q.?starttime.orValue([])).as(st,
+                              (st.size() == 0) ?
+                                true
+                              :
+                                now() - st[0].parse_time(time_layout.RFC3339) <= duration(state.base.maximum_age)
+                            )
+                          )
                         ),
                         "todo_content": state.?cursor.todo_content.orValue([]).filter(item,
                           configured_types.exists(ct, ct.to_lower() == item.?contentType.orValue("").to_lower())

packages/o365/data_stream/audit/_dev/test/policy/test-cel-default.expected

@@ -17,7 +17,6 @@ inputs:
             token_url: https://login.microsoftonline.com/tenant-id/oauth2/v2.0/token
           data_stream:
             dataset: o365.audit
-            type: logs
           interval: 3m
           max_executions: 10000
           processors:
@@ -73,16 +72,44 @@ inputs:
                 }
               )
             ).as(state,
-              // Filter stale cursor data that references content types no longer in config.
-              // Listing links have the content type embedded in the URL query string, so
-              // links for removed types would produce API errors if not pruned here.
+              // Filter stale cursor data before acting on it.
               state.base.content_types.split(",").map(t, t.trim_space()).as(configured_types,
                 state.with(
                   {
                     "cursor": state.?cursor.orValue({}).with(
                       {
+                        // Listing links have the content type embedded in the URL
+                        // query string, so links for removed types would produce API
+                        // errors if not pruned here.
+                        //
+                        // Links whose startTime has aged past maximum_age are also
+                        // pruned. The Office 365 Management API rejects requests
+                        // whose startTime is more than 7 days in the past with
+                        // AF20055; a link queued before a prolonged period without
+                        // successful fetches (agent downtime, auth failures,
+                        // credential rotation gaps, etc.) may have aged past that
+                        // boundary while parked in the cursor. The listing-error
+                        // branch cannot evict it reliably because its map-shaped
+                        // error return is treated by the runtime as a freeze signal
+                        // that discards cursor mutations. Once pruned, the bottom
+                        // branch generates a fresh link clamped to the valid window
+                        // on the next iteration.
                         "todo_links": state.?cursor.todo_links.orValue([]).filter(link,
                           configured_types.exists(ct, link.to_lower().contains("contenttype=" + ct.to_lower()))
+                          &&
+                          link.parse_url().RawQuery.parse_query().as(q,
+                            // Listing URLs generated by this program use "startTime",
+                            // but NextPageUri values returned by the API have been
+                            // seen with the lowercase "starttime" variant (see
+                            // #15325), so accept either. If no startTime can be
+                            // extracted, keep the link and let the API itself decide.
+                            q.?startTime.orValue(q.?starttime.orValue([])).as(st,
+                              (st.size() == 0) ?
+                                true
+                              :
+                                now() - st[0].parse_time(time_layout.RFC3339) <= duration(state.base.maximum_age)
+                            )
+                          )
                         ),
                         "todo_content": state.?cursor.todo_content.orValue([]).filter(item,
                           configured_types.exists(ct, ct.to_lower() == item.?contentType.orValue("").to_lower())

packages/o365/data_stream/audit/_dev/test/policy/test-o365-all.expected

@@ -20,7 +20,6 @@ inputs:
             - DLP.All
           data_stream:
             dataset: o365.audit
-            type: logs
           key: /tmp/path_to/key
           key_passphrase: ${SECRET_1}
           processors:

packages/o365/data_stream/audit/_dev/test/system/test-cel-stale-nextpage-config.yml

@@ -0,0 +1,34 @@
+input: cel
+service: o365-cel
+vars: ~
+policy_template: o365
+data_stream:
+  vars:
+    url: http://{{Hostname}}:{{Port}}
+    token_url: http://{{Hostname}}:{{Port}}
+    preserve_original_event: true
+    client_id: test-cel-client-id
+    client_secret: test-cel-client-secret
+    azure_tenant_id: test-cel-tenant-id
+    content_types: "Audit.Exchange"
+    initial_interval: 30m
+    maximum_age: 1h
+    enable_request_tracer: true
+# Exercises the pre-flight filter that drops queued listing links whose
+# startTime has aged past maximum_age (see packages/o365/changelog.yml 3.8.1).
+# The mock's Audit.Exchange first page returns a NextPageUri with a 2019
+# startTime; with a 1h maximum_age that NextPageUri must be pruned by the
+# filter rather than followed.
+#
+# initial_interval (30m) is deliberately less than maximum_age (1h) so the
+# fresh first listing URL's startTime (now - 30m) sits comfortably inside
+# the window (now - 1h) with 30 minutes of buffer against iteration delay.
+# This mirrors the package default, where initial_interval (167h) is 55m
+# less than maximum_age (167h55m); setting the two equal would place the
+# fresh URL exactly at the boundary and let iteration-time drift put it
+# outside.
+#
+# The first-page event ingests normally, the NextPageUri event must not.
+# hit_count is therefore 1, not 2.
+assert:
+  hit_count: 1

packages/o365/data_stream/audit/agent/stream/cel.yml.hbs

@@ -154,16 +154,44 @@ program: |-
       }
     )
   ).as(state,
-    // Filter stale cursor data that references content types no longer in config.
-    // Listing links have the content type embedded in the URL query string, so
-    // links for removed types would produce API errors if not pruned here.
+    // Filter stale cursor data before acting on it.
     state.base.content_types.split(",").map(t, t.trim_space()).as(configured_types,
       state.with(
         {
           "cursor": state.?cursor.orValue({}).with(
             {
+              // Listing links have the content type embedded in the URL
+              // query string, so links for removed types would produce API
+              // errors if not pruned here.
+              //
+              // Links whose startTime has aged past maximum_age are also
+              // pruned. The Office 365 Management API rejects requests
+              // whose startTime is more than 7 days in the past with
+              // AF20055; a link queued before a prolonged period without
+              // successful fetches (agent downtime, auth failures,
+              // credential rotation gaps, etc.) may have aged past that
+              // boundary while parked in the cursor. The listing-error
+              // branch cannot evict it reliably because its map-shaped
+              // error return is treated by the runtime as a freeze signal
+              // that discards cursor mutations. Once pruned, the bottom
+              // branch generates a fresh link clamped to the valid window
+              // on the next iteration.
               "todo_links": state.?cursor.todo_links.orValue([]).filter(link,
                 configured_types.exists(ct, link.to_lower().contains("contenttype=" + ct.to_lower()))
+                &&
+                link.parse_url().RawQuery.parse_query().as(q,
+                  // Listing URLs generated by this program use "startTime",
+                  // but NextPageUri values returned by the API have been
+                  // seen with the lowercase "starttime" variant (see
+                  // #15325), so accept either. If no startTime can be
+                  // extracted, keep the link and let the API itself decide.
+                  q.?startTime.orValue(q.?starttime.orValue([])).as(st,
+                    (st.size() == 0) ?
+                      true
+                    :
+                      now() - st[0].parse_time(time_layout.RFC3339) <= duration(state.base.maximum_age)
+                  )
+                )
               ),
               "todo_content": state.?cursor.todo_content.orValue([]).filter(item,
                 configured_types.exists(ct, ct.to_lower() == item.?contentType.orValue("").to_lower())

packages/o365/manifest.yml

@@ -1,6 +1,6 @@
 name: o365
 title: Microsoft Office 365
-version: "3.8.0"
+version: "3.8.1"
 description: Collect logs from Microsoft Office 365 with Elastic Agent.
 type: integration
 format_version: "3.2.3"