[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:
@@ -103,6 +102,43 @@ inputs:
                 )
               )
             ).as(state,
+              // Drop listing links whose startTime has aged past the API's accepted
+              // window. The Office 365 Management API rejects requests whose startTime
+              // is more than 7 days in the past with AF20055; a link that was 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, so the only way out is to prune up front. Once pruned,
+              // the bottom branch generates a fresh link clamped to the valid
+              // window on the next iteration.
+              (now() - duration(state.base.maximum_age)).as(oldest_allowed_start,
+                state.with(
+                  {
+                    "cursor": state.cursor.with(
+                      {
+                        "todo_links": state.cursor.todo_links.filter(link,
+                          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
+                              :
+                                st[0].parse_time(time_layout.RFC3339) >= oldest_allowed_start
+                            )
+                          )
+                        ),
+                      }
+                    ),
+                  }
+                )
+              )
+            ).as(state,
 
               (state.todo_types.size() == 0) ?
                 // ------------------------------------------------------------------------

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:
@@ -93,6 +92,43 @@ inputs:
                 )
               )
             ).as(state,
+              // Drop listing links whose startTime has aged past the API's accepted
+              // window. The Office 365 Management API rejects requests whose startTime
+              // is more than 7 days in the past with AF20055; a link that was 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, so the only way out is to prune up front. Once pruned,
+              // the bottom branch generates a fresh link clamped to the valid
+              // window on the next iteration.
+              (now() - duration(state.base.maximum_age)).as(oldest_allowed_start,
+                state.with(
+                  {
+                    "cursor": state.cursor.with(
+                      {
+                        "todo_links": state.cursor.todo_links.filter(link,
+                          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
+                              :
+                                st[0].parse_time(time_layout.RFC3339) >= oldest_allowed_start
+                            )
+                          )
+                        ),
+                      }
+                    ),
+                  }
+                )
+              )
+            ).as(state,
 
               (state.todo_types.size() == 0) ?
                 // ------------------------------------------------------------------------

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

@@ -174,6 +174,43 @@ program: |-
       )
     )
   ).as(state,
+    // Drop listing links whose startTime has aged past the API's accepted
+    // window. The Office 365 Management API rejects requests whose startTime
+    // is more than 7 days in the past with AF20055; a link that was 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, so the only way out is to prune up front. Once pruned,
+    // the bottom branch generates a fresh link clamped to the valid
+    // window on the next iteration.
+    (now() - duration(state.base.maximum_age)).as(oldest_allowed_start,
+      state.with(
+        {
+          "cursor": state.cursor.with(
+            {
+              "todo_links": state.cursor.todo_links.filter(link,
+                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
+                    :
+                      st[0].parse_time(time_layout.RFC3339) >= oldest_allowed_start
+                  )
+                )
+              ),
+            }
+          ),
+        }
+      )
+    )
+  ).as(state,
 
     (state.todo_types.size() == 0) ?
       // ------------------------------------------------------------------------

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"