o365: fix error propagation within cel program (#15445)

efd6 · web-flow · commit 6c447acff122 · 2025-09-25T08:11:59.000+09:30
The logic for handling error propagation from failed subscriptions
requests was incorrect. The downstream logic was expecting to find the
error in the root of state, but the subscription request block was
leaving it in the work object.

Later error handling was incorrectly leaving any error state in the root
of state, when it should have been placed under events.

In the case that a subscription has failed, there may not be an entry in
the last_for list for the subscription, resulting in a look-up failure
during assessment for completion of the hour-window steps.

Finally, the postamble logic was brittle to failed requests, assuming
that the todo_type and todo_content list would be present, which is not
necessarily the case when a subscription has failed.

This has been manually tested against the mock committed in the deploy
directory (with removal of authentication for convenience) with the
following outcomes:

only one, invalid, subscription:
"""
-- data.json --
{
	"url": "http://localhost:9001",
	"want_more": false,
	"base": {
		"tenant_id": "test-cel-tenant-id",
		"list_contents_start_time": "12h",
		"batch_interval": "1h",
		"maximum_age": "167h55m",
		"content_types": "Audit.TypeRequiringAdditionalPermissions"
	}
}
-- out.json --
{
	"base": {
		"batch_interval": "1h",
		"content_types": "Audit.TypeRequiringAdditionalPermissions",
		"list_contents_start_time": "12h",
		"maximum_age": "167h55m",
		"tenant_id": "test-cel-tenant-id"
	},
	"events": {
		"error": {
			"code": "401",
			"id": "401 Unauthorized",
			"message": "POST /activity/feed/subscriptions/start?contentType=Audit.TypeRequiringAdditionalPermissions: {\"error\":{\"code\":\"AF10001\",\"message\":\"The permission set (...) sent in the request does not include the expected permission.\"}}"
		}
	},
	"url": "http://localhost:9001",
	"want_more": false,
	"work": {
		"curr_type": "Audit.TypeRequiringAdditionalPermissions",
		"todo_type": []
	}
}
"""

start with an invalid subscription:
"""
-- data.json --
{
	"url": "http://localhost:9001",
	"want_more": false,
	"base": {
		"tenant_id": "test-cel-tenant-id",
		"list_contents_start_time": "12h",
		"batch_interval": "1h",
		"maximum_age": "167h55m",
		"content_types": "Audit.TypeRequiringAdditionalPermissions, Audit.SharePoint"
	}
}
-- out.json --
{
	"base": {
		"batch_interval": "1h",
		"content_types": "Audit.TypeRequiringAdditionalPermissions, Audit.SharePoint",
		"list_contents_start_time": "12h",
		"maximum_age": "167h55m",
		"tenant_id": "test-cel-tenant-id"
	},
	"events": {
		"error": {
			"code": "401",
			"id": "401 Unauthorized",
			"message": "POST /activity/feed/subscriptions/start?contentType=Audit.TypeRequiringAdditionalPermissions: {\"error\":{\"code\":\"AF10001\",\"message\":\"The permission set (...) sent in the request does not include the expected permission.\"}}"
		}
	},
	"url": "http://localhost:9001",
	"want_more": true,
	"work": {
		"curr_type": "Audit.TypeRequiringAdditionalPermissions",
		"todo_type": [
			"Audit.SharePoint"
		]
	}
}
{
	"base": {
		"batch_interval": "1h",
		"content_types": "Audit.TypeRequiringAdditionalPermissions, Audit.SharePoint",
		"list_contents_start_time": "12h",
		"maximum_age": "167h55m",
		"tenant_id": "test-cel-tenant-id"
	},
	"cursor": {
		"last_for": {
			"audit.sharepoint": "2025-09-23T13:07:36.915764483Z"
		}
	},
	"events": [
		{
… more …
"""

end with an invalid subscription:
"""
-- data.json --
{
	"url": "http://localhost:9001",
	"want_more": false,
	"base": {
		"tenant_id": "test-cel-tenant-id",
		"list_contents_start_time": "12h",
		"batch_interval": "1h",
		"maximum_age": "167h55m",
		"content_types": "Audit.SharePoint, Audit.TypeRequiringAdditionalPermissions"
	}
}
-- out.json --
… more before …
			"enabled": true,
			"type": "Audit.SharePoint"
		},
		"todo_content": [],
		"todo_type": [
			"Audit.TypeRequiringAdditionalPermissions"
		]
	}
}
{
	"base": {
		"batch_interval": "1h",
		"content_types": "Audit.SharePoint, Audit.TypeRequiringAdditionalPermissions",
		"list_contents_start_time": "12h",
		"maximum_age": "167h55m",
		"tenant_id": "test-cel-tenant-id"
	},
	"cursor": {
		"last_for": {
			"audit.sharepoint": "2025-09-24T00:10:30.808672557Z"
		}
	},
	"events": {
		"error": {
			"code": "401",
			"id": "401 Unauthorized",
			"message": "POST /activity/feed/subscriptions/start?contentType=Audit.TypeRequiringAdditionalPermissions: {\"error\":{\"code\":\"AF10001\",\"message\":\"The permission set (...) sent in the request does not include the expected permission.\"}}"
		}
	},
	"url": "http://localhost:9001",
	"want_more": false,
	"work": {
		"curr_type": "Audit.TypeRequiringAdditionalPermissions",
		"todo_type": []
	}
}
"""
diff --git a/packages/o365/changelog.yml b/packages/o365/changelog.yml
@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "2.29.2"
+  changes:
+    - description: Fix handling of error propagation within agent CEL program.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/15445
 - version: "2.29.1"
   changes:
     - description: Fix handling of error conditions when requesting work continuation.
diff --git a/packages/o365/data_stream/audit/agent/stream/cel.yml.hbs b/packages/o365/data_stream/audit/agent/stream/cel.yml.hbs
@@ -234,15 +234,17 @@ program: |-
         )
       :
         {
-          "error": {
-            "code": string(list_resp.StatusCode),
-            "id": string(list_resp.Status),
-            "message": "GET " + state.work.next_list + ": " + (
-              (size(list_resp.Body) != 0) ?
-                string(list_resp.Body)
-              :
-                string(list_resp.Status) + " (" + string(list_resp.StatusCode) + ")"
-            ),
+          "events": {
+            "error": {
+              "code": string(list_resp.StatusCode),
+              "id": string(list_resp.Status),
+              "message": "GET " + state.work.next_list + ": " + (
+                (size(list_resp.Body) != 0) ?
+                  string(list_resp.Body)
+                :
+                  string(list_resp.Status) + " (" + string(list_resp.StatusCode) + ")"
+              ),
+            },
           },
         }
       )
@@ -314,9 +316,9 @@ program: |-
       ).as(state, !has(state.base) ?
         // No current work item above, so finish.
         {}
-      : has(state.error) ?
+      : has(state.?work.sub.error) ?
         // Getting subscription detail failed.
-        state
+        state.work.sub.as(err, state.drop("work.sub").with({"events": err}))
       :
         (
           // This exists purely to rewrite the cursor from the original
@@ -440,15 +442,17 @@ program: |-
                   )
                 :
                   {
-                    "error": {
-                      "code": string(list_resp.StatusCode),
-                      "id": string(list_resp.Status),
-                      "message": "GET /activity/feed/subscriptions/content?contentType=" + state.work.curr_type + ": " + (
-                        (size(list_resp.Body) != 0) ?
-                          string(list_resp.Body)
-                        :
-                          string(list_resp.Status) + " (" + string(list_resp.StatusCode) + ")"
-                      ),
+                    "events": {
+                      "error": {
+                        "code": string(list_resp.StatusCode),
+                        "id": string(list_resp.Status),
+                        "message": "GET /activity/feed/subscriptions/content?contentType=" + state.work.curr_type + ": " + (
+                          (size(list_resp.Body) != 0) ?
+                            string(list_resp.Body)
+                          :
+                            string(list_resp.Status) + " (" + string(list_resp.StatusCode) + ")"
+                        ),
+                      },
                     },
                   }
                 )
@@ -461,8 +465,12 @@ program: |-
     (
       // Ensure that we bring the current type up to the current time,
       // even if we did not get any content for the query period.
-      has(state.?work.curr_type) && state.?cursor.last_for.optMap(l,
-        timestamp(l[state.work.curr_type.to_lower()]) < now - duration("1h")
+      has(state.?work.curr_type) && state.?cursor.last_for.optMap(last_for,
+        state.work.curr_type.to_lower().as(curr_type, curr_type in last_for ?
+            timestamp(last_for[curr_type]) < now - duration("1h")
+          :
+            false
+        )
       ).orValue(false) && !state.work.todo_type.exists(t, t == state.work.curr_type)
     ) ?
       state.with(
@@ -476,17 +484,14 @@ program: |-
     state.with(
       {
         "want_more": state.work.as(w,
-          size(w.todo_type) != 0 || size(w.todo_content) != 0 || w.?next_list.orValue("") != ""
+          size(w.?todo_type.orValue([])) != 0 || size(w.?todo_content.orValue([])) != 0 || w.?next_list.orValue("") != ""
         ),
       }
     )
   ).as(state,
     // Make sure we complete the remaining work if we got
     // no events but the work lists are not empty.
-    // We do not need to put a dummy event in in the
-    // case that we have errored, since the error will
-    // be raised to an event by the input.
-    (state.want_more && !has(state.error) && type(state.events) == list && size(state.events) == 0) ?
+    (state.want_more && type(state.events) == list && size(state.events) == 0) ?
       state.with(
         {
           "events": [{"retry": true}],
diff --git a/packages/o365/manifest.yml b/packages/o365/manifest.yml
@@ -1,6 +1,6 @@
 name: o365
 title: Microsoft Office 365
-version: "2.29.1"
+version: "2.29.2"
 description: Collect logs from Microsoft Office 365 with Elastic Agent.
 type: integration
 format_version: "3.2.3"
