ti_threatconnect: fix handling of missing cursor.last_timestamp (elastic#13235)

It is possible to get into a state where the program expects there to be
a last_timestamp in the cursor but none exists due to previous data
being present but empty. Fix this by falling back to the look-back time
if the last_timestamp is missing.

Also reorganise the code to make the logic less opaque.

Author: efd6

packages/ti_threatconnect/changelog.yml

@@ -1,5 +1,10 @@
 # newer versions go on top
 # WARNING: this version number needs to be kept up to date in the transform!
+- version: "1.9.3"
+  changes:
+    - description: Prevent agent failure when last time stamp is missing in cursor.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/13235
 - version: "1.9.2"
   changes:
     - description: Add a compatibility pipeline to the transform.

packages/ti_threatconnect/data_stream/indicator/agent/stream/cel.yml.hbs

@@ -71,10 +71,10 @@ redact:
 # milliseconds and if it ended in 0, that 0 would be dropped and TC TQL then would error on the timestamp.
 program: |
   ['lastModified GEQ "'+(
-      !state.want_more ?
-          state.?cursor.last_timestamp.orValue((now - duration(state.initial_interval)).format(time_layout.RFC3339))
-      :
-          state.?cursor.first_timestamp.orValue("")
+          state.want_more ?
+              state.?cursor.first_timestamp.orValue("")
+          :
+              state.?cursor.last_timestamp.orValue((now - duration(state.initial_interval)).format(time_layout.RFC3339))
       )+'"'+(
           state.?tql_filter.orValue("") != "" ?
               " AND "+state.tql_filter.trim(" ")
@@ -127,23 +127,24 @@ program: |
                       state.url.trim_right("/")
               ),
               "cursor": {
+                  "first_timestamp": (
+                      !(has(body.data) && has(state.?cursor.first_timestamp)) ?
+                          // We don't have any data or a first_timestamp. Limit to look-back.
+                          string(now - duration(state.initial_interval))
+                      : (has(body.next) && body.next != null && body.next != "") ?
+                          // want_more is true, so limit to first timestamp.
+                          state.cursor.first_timestamp
+                      :
+                          // We have data, but want_more is false, limit to last available
+                          // timestamp falling back to look-back.
+                          state.?cursor.last_timestamp.orValue(string(now - duration(state.initial_interval)))
+                  ),
                   ?"last_timestamp": (
                       has(body.data) && body.data.size() > 0 ?
                           optional.of(body.data.map(e, timestamp(e.lastModified)).max() + duration("1s"))
                       :
                           state.?cursor.last_timestamp
                   ),
-                  "first_timestamp": (
-                      has(body.data) && state.?cursor.first_timestamp.orValue(null) != null ?
-                          (
-                              has(body.next) && body.next != null && body.next != "" ?
-                                  state.cursor.first_timestamp
-                              :
-                                  state.cursor.last_timestamp
-                          )
-                      :
-                          string(now - duration(state.initial_interval))
-                  ),
               }
           })
       :

packages/ti_threatconnect/elasticsearch/transform/latest/transform.yml

@@ -9,8 +9,8 @@ source:
 # us that ability in order to prevent having duplicate IoC data and prevent query
 # time field type conflicts.
 dest:
-  index: "logs-ti_threatconnect_latest.dest_indicator-5"
-  pipeline: "1.9.2-tactics_compatibility"
+  index: "logs-ti_threatconnect_latest.dest_indicator-6"
+  pipeline: "1.9.3-tactics_compatibility"
   aliases:
     - alias: "logs-ti_threatconnect_latest.indicator"
       move_on_creation: true
@@ -33,4 +33,4 @@ _meta:
   managed: true
   # Bump this version to delete, reinstall, and restart the transform during package.
   # Version bump is needed if there is any code change in transform.
-  fleet_transform_version: 0.5.0
+  fleet_transform_version: 0.6.0

packages/ti_threatconnect/manifest.yml

@@ -2,7 +2,7 @@
 format_version: 3.0.3
 name: ti_threatconnect
 title: ThreatConnect
-version: "1.9.2"
+version: "1.9.3"
 description: Collects Indicators from ThreatConnect using the Elastic Agent and saves them as logs inside Elastic
 type: integration
 categories: