[Akamai] - Migrated SIEM data stream from HTTPJSON to CEL input

packages/akamai/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "2.29.0"
+  changes:
+    - description: Migrated siem data stream from HTTP JSON to CEL.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/15713
 - version: "2.28.1"
   changes:
     - description: Fixed time duration handling in request parameters to conform to API guidelines.

packages/akamai/data_stream/siem/_dev/test/system/test-emulator-config.yml

@@ -1,4 +1,4 @@
-input: httpjson
+input: cel
 service: akamai-siem-emulator
 vars: ~
 data_stream:
@@ -12,10 +12,9 @@ data_stream:
     access_token: at-6b8c7217-8748-490d-b0f5-bfeb72b2e7cd
     config_ids: 123456
     event_limit: 20
+    # The akamai-siem emulator does not limit the number of events or pages returned, so we set a large number of max_executions.
+    max_executions: 50000
     enable_request_tracer: true
 assert:
   # 12 hours at 5 minutes between events.
   hit_count: 144 #  = 12 * 60/5
-skip:
-  reason: "The fleet health status changes to degraded when the HTTPJSON template's value evaluation comes up empty, which leads to system test failures but does not interrupt the data flow."
-  link: https://github.com/elastic/beats/issues/45664

packages/akamai/data_stream/siem/agent/stream/cel.yml.hbs

@@ -0,0 +1,159 @@
+config_version: 2
+interval: {{interval}}
+resource:
+  url: {{api_host}}/siem/v1/configs/{{config_ids}}
+  {{#if ssl}}
+  ssl: {{ssl}}
+  {{/if}}
+  {{#if http_client_timeout}}
+  timeout: {{http_client_timeout}}
+  {{/if}}
+  {{#if proxy_url }}
+  proxy_url: {{proxy_url}}
+  {{/if}}
+  tracer:
+    enabled: {{enable_request_tracer}}
+    filename: "../../logs/cel/http-request-trace-*.ndjson"
+    maxbackups: 5
+{{#if max_executions}}
+max_executions: {{max_executions}}
+{{/if}}
+
+state:
+  client_token: {{client_token}}
+  access_token: {{access_token}}
+  client_secret: {{client_secret}}
+  initial_interval: {{initial_interval}}
+  event_limit: {{event_limit}}
+
+redact:
+  fields:
+    - client_secret
+    - access_token
+    - client_token  
+
+program: |-
+  state.with(
+    (
+      state.?cursor.recovery_mode.orValue(false) ?
+        {
+          "from": int(now - duration("12h")),
+          "to": int(now),
+        }
+      : state.?cursor.last_offset.hasValue() ?
+        {
+          "offset": state.cursor.last_offset,
+        }
+      :
+        {
+          "from": max(int(now - duration(state.initial_interval)), int(now - duration("12h"))),
+          "to": int(now),
+        }
+    ).as(params,
+      now.format("20060102T15:04:05-0700").as(timestamp,
+        uuid().as(nonce,
+          sprintf(
+            "EG1-HMAC-SHA256 client_token=%s;access_token=%s;timestamp=%s;nonce=%s;",
+            [state.client_token, state.access_token, timestamp, nonce]
+          ).as(sig_base,
+            base64(hmac(timestamp, "sha256", bytes(state.client_secret))).as(sig_key,
+              (
+                state.url.trim_right("/") + "?" + {
+                  "limit": [string(state.event_limit)],
+                  ?"from": (params.?from.orValue("") != "") ? optional.of([string(params.from)]) : optional.none(),
+                  ?"to": (params.?to.orValue("") != "") ? optional.of([string(params.to)]) : optional.none(),
+                  ?"offset": (params.?offset.orValue("") != "") ? optional.of([string(params.offset)]) : optional.none(),
+                }.format_query()
+              ).parse_url().as(u,
+                sprintf(
+                  "GET\t%s\t%s\t%s?%s\t\t\t%s",
+                  [
+                    string(u.Scheme),
+                    string(u.Host),
+                    string(u.Path),
+                    string(u.RawQuery),
+                    sig_base,
+                  ]
+                ).as(to_sign,
+                  base64(hmac(to_sign, "sha256", bytes(sig_key))).as(signature,
+                    sig_base + "signature=" + signature
+                  )
+                )
+              )
+            )
+          )
+        )
+      ).as(auth_header,
+        request(
+          "GET",
+          state.url.trim_right("/") + "?" + {
+            "limit": [string(state.event_limit)],
+            ?"from": (params.?from.orValue("") != "") ? optional.of([string(params.from)]) : optional.none(),
+            ?"to": (params.?to.orValue("") != "") ? optional.of([string(params.to)]) : optional.none(),
+            ?"offset": (params.?offset.orValue("") != "") ? optional.of([string(params.offset)]) : optional.none(),
+          }.format_query()
+        ).with(
+          {
+            "Header": {
+              "Authorization": [auth_header],
+            },
+          }
+        ).do_request().as(resp,
+          (resp.StatusCode == 200) ?
+            string(resp.Body).split("\n").filter(line, line != "").as(lines,
+              {
+                "events": lines.map(line, {"message": line}),
+                "cursor": {
+                  ?"last_offset": (lines.size() > 0) ?
+                    lines[lines.size() - 1].decode_json().as(lastEvent,
+                      lastEvent.?offset
+                    )
+                  :
+                    optional.none(),
+                  "recovery_mode": false,
+                },
+                "want_more": (lines.size() > 0) ?
+                  lines[lines.size() - 1].decode_json().as(lastEvent,
+                    has(lastEvent.offset) && lastEvent.offset != ""
+                  )
+                :
+                  false,
+              }
+            )
+          : (resp.StatusCode == 416) ?
+            {
+              "events": {
+                "error": {
+                  "code": string(resp.StatusCode),
+                  "id": string(resp.Status),
+                  "message": "GET " + state.url.trim_right("/") + (
+                    (size(resp.Body) != 0) ?
+                      string(resp.Body)
+                    :
+                      string(resp.Status) + " (" + string(resp.StatusCode) + ")"
+                  ),
+                },
+              },
+              "cursor": {"recovery_mode": true},
+              "want_more": true,
+            }
+          :
+            {
+              "events": {
+                "error": {
+                  "code": string(resp.StatusCode),
+                  "id": string(resp.Status),
+                  "message": "GET " + state.url.trim_right("/") + (
+                    (size(resp.Body) != 0) ?
+                      string(resp.Body)
+                    :
+                      string(resp.Status) + " (" + string(resp.StatusCode) + ")"
+                  ),
+                },
+              },
+              "want_more": false,
+            }
+        )
+      )
+    )
+  )

packages/akamai/data_stream/siem/manifest.yml

@@ -1,8 +1,8 @@
 type: logs
 title: Akamai SIEM Logs
 streams:
-  - input: httpjson
-    template_path: httpjson.yml.hbs
+  - input: cel
+    template_path: cel.yml.hbs
     title: Akamai SIEM logs
     description: Collect Akamai logs via the SIEM API
     vars:
@@ -76,6 +76,14 @@ streams:
         show_user: false
         title: Event Limit
         description: Defines the approximate maximum number of security events each fetch returns, in both offset and time-based modes. The default limit is 10000 and the maximum limit available is 600000. Listing an unlimited number of logs isn't possible. Expect requests to return a slightly higher number of security events than you set in the limit parameter, because data is stored in different buckets.
+      - name: max_executions
+        type: integer
+        title: Maximum Pages Per Interval
+        description: Maximum Pages Per Interval is the maximum number of pages that can be collected at each interval.
+        multi: false
+        required: false
+        show_user: false
+        default: 5000
       - name: proxy_url
         type: text
         title: Proxy URL

packages/akamai/manifest.yml

@@ -1,13 +1,13 @@
 name: akamai
 title: Akamai
-version: "2.28.1"
+version: "2.29.0"
 description: Collect logs from Akamai with Elastic Agent.
 type: integration
 format_version: "3.0.2"
 categories: [security, cdn_security]
 conditions:
   kibana:
-    version: "^8.13.0 || ^9.0.0"
+    version: "^8.18.0 || ^9.0.0"
 icons:
   - src: /img/akamai_logo.svg
     title: Akamai
@@ -18,7 +18,7 @@ policy_templates:
     title: Akamai logs
     description: Collect SIEM logs from Akamai
     inputs:
-      - type: httpjson
+      - type: cel
         title: "Collect Akamai SIEM logs via API"
         description: "Collecting SIEM logs from Akamai via API"
       - type: gcs