[proofpoint_tap] Set timestamp to max of {click,message}Time and threatTime (#9982)

Prevent new events from appearing in the past by using the time the event was triggered as the `@timestamp`. An event can be triggered when a click occurs, a message is delivered/blocked, or new threat is discovered. When a new threat is discovered the previous behavior made the event appear to have happened in the past because we used the clickTime or messageTime as the timestamp which might have occurred weeks ago.

The API docs state, "the time an event is created is always after either of these two times:

- the time that the message was sent or the time click occurred (messageTime or clickTime)
- the time that the threat referenced by the message or click was recognized and condemned by Proofpoint (threatsInfoMap/threatTime)"

I suspect this does not occur for `clicks_blocked` because click was already condemned, but in case they send an update if a new threat matches the previous click I'll include the same change there.

Fixes #9967

Author: andrewkroh

packages/proofpoint_tap/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "1.19.0"
+  changes:
+    - description: Set @timestamp based on the event trigger which is either the messageTime/clickTime or the threatTime.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/9982
 - version: "1.18.1"
   changes:
     - description: Prevent dropped events due to an insufficiently unique deduplication key.  The document `_id` computation now uses a hash over the `event.original` value.

packages/proofpoint_tap/data_stream/clicks_blocked/_dev/test/pipeline/test-clicks-blocked.log-expected.json

@@ -57,6 +57,7 @@
                 "clicks_blocked": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "phish",
+                    "click_time": "2022-03-21T07:52:11.000Z",
                     "threat": {
                         "id": "3xx97xx852c66a7xx761450xxxxxx9f4ffaxxxxxxxxxxxxxxx7a76481xx",
                         "status": "active",
@@ -156,6 +157,7 @@
                 "clicks_blocked": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "phish",
+                    "click_time": "2022-03-30T07:22:52.000Z",
                     "threat": {
                         "id": "fdxxxxxxxxxxxcc34aff1aefxbx3xx7xb7xfxcxx1xxxxxxxx98780b5xxxexbx5xc32c",
                         "status": "active",
@@ -254,6 +256,7 @@
                 "clicks_blocked": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "spam",
+                    "click_time": "2022-03-30T07:10:19.000Z",
                     "threat": {
                         "id": "eaxxxxxxxxxxxx6376xxxxxxxxxxx1cba65xxx9x7xxxxxxxxxxfbbxx4x0",
                         "status": "active",
@@ -353,6 +356,7 @@
                 "clicks_blocked": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "malware",
+                    "click_time": "2022-03-30T10:11:12.000Z",
                     "threat": {
                         "id": "502bxxxxxxxxxxx70513b6cxxxxxxxxxxxxebc7fc699xxxxxxxxxxxxxxxxd5f",
                         "status": "active",
@@ -452,6 +456,7 @@
                 "clicks_blocked": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "spam",
+                    "click_time": "2022-03-30T10:01:01.000Z",
                     "threat": {
                         "id": "47580xdx0x2x5x2xfx8x3x3x7x7xxxxcx6x7x4x4x1xexcx5cx9x3xfxfxxx1",
                         "status": "active",

packages/proofpoint_tap/data_stream/clicks_blocked/elasticsearch/ingest_pipeline/default.yml

@@ -55,13 +55,19 @@ processors:
       if: ctx.email?.to?.address instanceof String
   - date:
       field: json.clickTime
+      target_field: proofpoint_tap.clicks_blocked.click_time
       if: ctx.json?.clickTime != null && ctx.json.clickTime != ''
       formats:
         - ISO8601
       on_failure:
         - append:
             field: error.message
             value: '{{{_ingest.on_failure_message}}}'
+  - set:
+      field: '@timestamp'
+      copy_from: proofpoint_tap.clicks_blocked.click_time
+      ignore_failure: true
+      ignore_empty_value: true
   - rename:
       field: json.id
       target_field: event.id
@@ -126,6 +132,18 @@ processors:
         - append:
             field: error.message
             value: '{{{_ingest.on_failure_message}}}'
+  - script:
+      lang: painless
+      description: Set the @timestamp to the maximum of clickTime and threatTime.
+      tag: timestamp-is-maximum
+      if: ctx.proofpoint_tap?.clicks_blocked?.threat?.time instanceof String
+      ignore_failure: true
+      source: |
+        def ts = Instant.parse(ctx['@timestamp']);
+        def threatTime = Instant.parse(ctx.proofpoint_tap.clicks_blocked.threat.time);
+        if (threatTime.isAfter(ts)) {
+          ctx['@timestamp'] = ctx.proofpoint_tap.clicks_blocked.threat.time;
+        }
   - uri_parts:
       field: json.url
       keep_original: false

packages/proofpoint_tap/data_stream/clicks_blocked/fields/fields.yml

@@ -7,6 +7,9 @@
         - name: campaign_id
           type: keyword
           description: An identifier for the campaign of which the threat is a member, if available at the time of the query. Threats can be linked to campaigns even after these events are retrieved.
+        - name: click_time
+          type: date
+          description: The time the user clicked on the URL.
         - name: classification
           type: keyword
           description: The threat category of the malicious URL.

packages/proofpoint_tap/data_stream/clicks_permitted/_dev/test/pipeline/test-clicks-permitted.log-expected.json

@@ -1,7 +1,7 @@
 {
     "expected": [
         {
-            "@timestamp": "2016-06-24T19:17:44.000Z",
+            "@timestamp": "2016-06-24T19:17:46.000Z",
             "destination": {
                 "as": {
                     "number": 29518,
@@ -57,6 +57,7 @@
                 "clicks_permitted": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "MALWARE",
+                    "click_time": "2016-06-24T19:17:44.000Z",
                     "threat": {
                         "id": "61f7622xx1x6x7x1x4xxxxxxxxxxx4xdbaxxxxxxxxxxx5xex3xbxxxxxdxfx5xxxxx0",
                         "status": "active",
@@ -99,7 +100,7 @@
             }
         },
         {
-            "@timestamp": "2022-03-21T20:39:37.000Z",
+            "@timestamp": "2022-03-30T10:05:57.000Z",
             "destination": {
                 "as": {
                     "number": 29518,
@@ -155,6 +156,7 @@
                 "clicks_permitted": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "phish",
+                    "click_time": "2022-03-21T20:39:37.000Z",
                     "threat": {
                         "id": "92c17aaxxxxxxxxxx07xx7xxxx9xexcx3x3xxxxxx8xx3xxxx",
                         "status": "active",
@@ -254,6 +256,7 @@
                 "clicks_permitted": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "spam",
+                    "click_time": "2022-03-30T10:51:53.000Z",
                     "threat": {
                         "id": "xxxxxxbx1cxcxx0xcx5xxxxdx5xex8xbx7xxxeexxxxxxxx9",
                         "status": "cleared",
@@ -352,6 +355,7 @@
                 "clicks_permitted": {
                     "campaign_id": "46x01x8x-x899-404x-xxx9-111xx393d1x7",
                     "classification": "phish",
+                    "click_time": "2022-03-30T00:56:14.000Z",
                     "threat": {
                         "id": "xxxdxxdx6x7x6xxxxx5xxx837ex4x4xcx8xcxxxexxx2xxxxxx5",
                         "status": "active",

packages/proofpoint_tap/data_stream/clicks_permitted/elasticsearch/ingest_pipeline/default.yml

@@ -55,13 +55,19 @@ processors:
       if: ctx.email?.to?.address instanceof String
   - date:
       field: json.clickTime
+      target_field: proofpoint_tap.clicks_permitted.click_time
       if: ctx.json?.clickTime != null && ctx.json.clickTime != ''
       formats:
         - ISO8601
       on_failure:
         - append:
             field: error.message
             value: '{{{_ingest.on_failure_message}}}'
+  - set:
+      field: '@timestamp'
+      copy_from: proofpoint_tap.clicks_permitted.click_time
+      ignore_failure: true
+      ignore_empty_value: true
   - rename:
       field: json.id
       target_field: event.id
@@ -126,6 +132,18 @@ processors:
         - append:
             field: error.message
             value: '{{{_ingest.on_failure_message}}}'
+  - script:
+      lang: painless
+      description: Set the @timestamp to the maximum of clickTime and threatTime.
+      tag: timestamp-is-maximum
+      if: ctx.proofpoint_tap?.clicks_permitted?.threat?.time instanceof String
+      ignore_failure: true
+      source: |
+        def ts = Instant.parse(ctx['@timestamp']);
+        def threatTime = Instant.parse(ctx.proofpoint_tap.clicks_permitted.threat.time);
+        if (threatTime.isAfter(ts)) {
+          ctx['@timestamp'] = ctx.proofpoint_tap.clicks_permitted.threat.time;
+        }
   - uri_parts:
       field: json.url
       keep_original: false

packages/proofpoint_tap/data_stream/clicks_permitted/fields/fields.yml

@@ -7,6 +7,9 @@
         - name: campaign_id
           type: keyword
           description: An identifier for the campaign of which the threat is a member, if available at the time of the query. Threats can be linked to campaigns even after these events are retrieved.
+        - name: click_time
+          type: date
+          description: The time the user clicked on the URL.
         - name: classification
           type: keyword
           description: The threat category of the malicious URL.

packages/proofpoint_tap/data_stream/message_blocked/_dev/test/pipeline/test-message-blocked.log-expected.json

@@ -1,7 +1,7 @@
 {
     "expected": [
         {
-            "@timestamp": "2022-01-01T00:45:55.050Z",
+            "@timestamp": "2022-01-01T02:03:40.050Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -126,7 +126,7 @@
             ]
         },
         {
-            "@timestamp": "2022-01-01T01:25:59.059Z",
+            "@timestamp": "2022-01-01T10:10:02.020Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -250,7 +250,7 @@
             ]
         },
         {
-            "@timestamp": "2022-01-01T04:51:56.269Z",
+            "@timestamp": "2022-01-01T11:06:50.580Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -381,7 +381,7 @@
             ]
         },
         {
-            "@timestamp": "2022-01-01T00:25:20.010Z",
+            "@timestamp": "2022-01-01T05:02:48.832Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -581,7 +581,7 @@
             ]
         },
         {
-            "@timestamp": "2022-01-01T00:00:00.000Z",
+            "@timestamp": "2022-01-01T03:02:25.092Z",
             "ecs": {
                 "version": "8.11.0"
             },

packages/proofpoint_tap/data_stream/message_blocked/elasticsearch/ingest_pipeline/default.yml

@@ -488,6 +488,19 @@ processors:
           ignore_missing: true
       ignore_failure: true
       if: ctx.proofpoint_tap?.message_blocked?.threat_info_map instanceof List
+  - script:
+      lang: painless
+      description: Set the @timestamp to the maximum of messageTime and threatTime.
+      tag: timestamp-is-maximum
+      if: ctx.proofpoint_tap?.message_blocked?.threat_info_map instanceof List
+      ignore_failure: true
+      source: |
+        def ts = Instant.parse(ctx['@timestamp']);
+        for (item in ctx.proofpoint_tap.message_blocked.threat_info_map) {
+          if (item?.threat?.time instanceof String && Instant.parse(item.threat.time).isAfter(ts)) {
+            ctx['@timestamp'] = item.threat.time;
+          }
+        }
   - remove:
       field: event.original
       if: ctx.tags?.contains('preserve_original_event') != true

packages/proofpoint_tap/data_stream/message_delivered/_dev/test/pipeline/test-message-delivered.log-expected.json

@@ -1,7 +1,7 @@
 {
     "expected": [
         {
-            "@timestamp": "2022-01-05T10:05:56.020Z",
+            "@timestamp": "2022-03-29T18:48:33.000Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -88,7 +88,7 @@
             ]
         },
         {
-            "@timestamp": "2022-01-01T00:00:00.000Z",
+            "@timestamp": "2022-05-01T19:00:00.653Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -158,7 +158,7 @@
             ]
         },
         {
-            "@timestamp": "2022-01-01T00:00:00.000Z",
+            "@timestamp": "2022-11-25T13:05:05.592Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -234,7 +234,7 @@
             ]
         },
         {
-            "@timestamp": "2022-01-01T00:00:00.000Z",
+            "@timestamp": "2022-10-11T00:23:02.519Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -310,7 +310,7 @@
             ]
         },
         {
-            "@timestamp": "2022-03-15T15:00:20.000Z",
+            "@timestamp": "2022-04-01T18:44:01.050Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -458,7 +458,7 @@
             ]
         },
         {
-            "@timestamp": "2021-09-28T16:28:59.490Z",
+            "@timestamp": "2022-04-01T23:14:30.450Z",
             "ecs": {
                 "version": "8.11.0"
             },
@@ -762,7 +762,7 @@
             ]
         },
         {
-            "@timestamp": "2022-03-24T13:24:57.000Z",
+            "@timestamp": "2022-04-01T20:56:13.000Z",
             "ecs": {
                 "version": "8.11.0"
             },

packages/proofpoint_tap/data_stream/message_delivered/elasticsearch/ingest_pipeline/default.yml

@@ -467,6 +467,19 @@ processors:
           ignore_missing: true
       ignore_failure: true
       if: ctx.proofpoint_tap?.message_delivered?.threat_info_map instanceof List
+  - script:
+      lang: painless
+      description: Set the @timestamp to the maximum of messageTime and threatTime.
+      tag: timestamp-is-maximum
+      if: ctx.proofpoint_tap?.message_delivered?.threat_info_map instanceof List
+      ignore_failure: true
+      source: |
+        def ts = Instant.parse(ctx['@timestamp']);
+        for (item in ctx.proofpoint_tap.message_delivered.threat_info_map) {
+          if (item?.threat?.time instanceof String && Instant.parse(item.threat.time).isAfter(ts)) {
+            ctx['@timestamp'] = item.threat.time;
+          }
+        }
   - remove:
       field: event.original
       if: ctx.tags?.contains('preserve_original_event') != true

packages/proofpoint_tap/docs/README.md

@@ -222,6 +222,7 @@ An example event for `clicks_blocked` looks as following:
 | log.offset | Log offset | long |
 | proofpoint_tap.clicks_blocked.campaign_id | An identifier for the campaign of which the threat is a member, if available at the time of the query. Threats can be linked to campaigns even after these events are retrieved. | keyword |
 | proofpoint_tap.clicks_blocked.classification | The threat category of the malicious URL. | keyword |
+| proofpoint_tap.clicks_blocked.click_time | The time the user clicked on the URL. | date |
 | proofpoint_tap.clicks_blocked.sender_ip | The IP address of the sender. | ip |
 | proofpoint_tap.clicks_blocked.threat.id | The unique identifier associated with this threat. It can be used to query the forensics and campaign endpoints. | keyword |
 | proofpoint_tap.clicks_blocked.threat.status | The current state of the threat. | keyword |
@@ -459,6 +460,7 @@ An example event for `clicks_permitted` looks as following:
 | log.offset | Log offset | long |
 | proofpoint_tap.clicks_permitted.campaign_id | An identifier for the campaign of which the threat is a member, if available at the time of the query. Threats can be linked to campaigns even after these events are retrieved. | keyword |
 | proofpoint_tap.clicks_permitted.classification | The threat category of the malicious URL. | keyword |
+| proofpoint_tap.clicks_permitted.click_time | The time the user clicked on the URL. | date |
 | proofpoint_tap.clicks_permitted.sender_ip | The IP address of the sender. | ip |
 | proofpoint_tap.clicks_permitted.threat.id | The unique identifier associated with this threat. It can be used to query the forensics and campaign endpoints. | keyword |
 | proofpoint_tap.clicks_permitted.threat.status | The current state of the threat. | keyword |

packages/proofpoint_tap/manifest.yml

@@ -1,7 +1,7 @@
 format_version: "3.0.3"
 name: proofpoint_tap
 title: Proofpoint TAP
-version: "1.18.1"
+version: "1.19.0"
 description: Collect logs from Proofpoint TAP with Elastic Agent.
 type: integration
 categories: