Fix failure-report timestamp skew on non-UTC hosts in ES/OpenSearch/Splunk outputs (#812)

* Fix failure-report timestamp skew on non-UTC hosts in ES/OS/Splunk sinks

arrival_date_utc is a UTC wall-clock string (generated in
parse_failure_report via an aware-UTC strftime), but elastic.py,
opensearch.py, and splunk.py parsed it back into a naive datetime and
called .timestamp(), which per the Python docs interprets naive values
as local time (https://docs.python.org/3/library/datetime.html#datetime.datetime.timestamp).
On any non-UTC host the epoch stored as the ES/OpenSearch arrival_date
field, used in the failure-report dedup match query, and sent as the
Splunk HEC event time was therefore off by the host's UTC offset
(verified -3600 s under TZ=Europe/Warsaw in January).

Add an assume_utc keyword to human_timestamp_to_datetime() /
human_timestamp_to_unix_timestamp() that attaches timezone.utc to naive
parses, and use it at the three arrival_date_utc call sites. Aware
inputs (explicit offsets) are unaffected; all other callers keep the
existing local-time semantics, whose round-trip with timestamp_to_human
is self-consistent on a single host (the broader local-time output
question is tracked separately in issue #811 bug 2).

The three new sink regression tests fail on the unfixed code
(verified by stashing the source changes) and force TZ=Europe/Warsaw
via time.tzset() so they catch the skew even on UTC CI runners.

Fixes half of https://github.com/domainaware/parsedmarc/issues/811.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Deduplicate TZ-forcing test boilerplate; fix unix-timestamp docstring

Extract the repeated TZ=Europe/Warsaw + time.tzset() setup/cleanup from
the four timestamp regression tests into a shared tests/tzutil.py
force_tz() helper, and correct human_timestamp_to_unix_timestamp()'s
docstring, which said the return type was float while the function
returns int.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: MISAPOR LAB <misapor@lab.misapor.pl>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Sean Whalen <44679+seanthegeek@users.noreply.github.com>

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Bug fixes
+
+- **Failure-report timestamps are no longer skewed by the host's UTC offset in the Elasticsearch, OpenSearch, and Splunk HEC outputs** ([#811](https://github.com/domainaware/parsedmarc/issues/811), bug 1). `arrival_date_utc` is a UTC wall-clock string, but the three sinks parsed it into a naive `datetime` and called `.timestamp()`, which per the Python docs interprets naive values as *local* time — so on any non-UTC host, the epoch stored as the ES/OpenSearch `arrival_date` field, used in the failure-report dedup query, and sent as the Splunk HEC event `time` was off by the host's UTC offset (1–2 h for most of Europe). `human_timestamp_to_datetime()` / `human_timestamp_to_unix_timestamp()` gained an `assume_utc` keyword that attaches `timezone.utc` to naive parses, and the `arrival_date_utc` consumers now use it.
+
 ## 10.2.0
 
 ### Changes

parsedmarc/elastic.py

@@ -660,7 +660,12 @@ def save_failure_report_to_elasticsearch(
     for original_header in original_headers:
         headers[original_header.lower()] = original_headers[original_header]
 
-    arrival_date = human_timestamp_to_datetime(failure_report["arrival_date_utc"])
+    # arrival_date_utc is a UTC wall-clock string; without assume_utc the
+    # naive .timestamp() below would interpret it as local time and skew
+    # the epoch by the host's UTC offset.
+    arrival_date = human_timestamp_to_datetime(
+        failure_report["arrival_date_utc"], assume_utc=True
+    )
     arrival_date_epoch_milliseconds = int(arrival_date.timestamp() * 1000)
 
     if index_suffix is not None:

parsedmarc/opensearch.py

@@ -660,7 +660,12 @@ def save_failure_report_to_opensearch(
     for original_header in original_headers:
         headers[original_header.lower()] = original_headers[original_header]
 
-    arrival_date = human_timestamp_to_datetime(failure_report["arrival_date_utc"])
+    # arrival_date_utc is a UTC wall-clock string; without assume_utc the
+    # naive .timestamp() below would interpret it as local time and skew
+    # the epoch by the host's UTC offset.
+    arrival_date = human_timestamp_to_datetime(
+        failure_report["arrival_date_utc"], assume_utc=True
+    )
     arrival_date_epoch_milliseconds = int(arrival_date.timestamp() * 1000)
 
     if index_suffix is not None:

parsedmarc/splunk.py

@@ -163,7 +163,11 @@ class HECClient(object):
         for report in failure_reports:
             data = self._common_data.copy()
             data["sourcetype"] = "dmarc:failure"
-            timestamp = human_timestamp_to_unix_timestamp(report["arrival_date_utc"])
+            # arrival_date_utc is a UTC wall-clock string; assume_utc keeps
+            # it from being re-interpreted as local time on non-UTC hosts.
+            timestamp = human_timestamp_to_unix_timestamp(
+                report["arrival_date_utc"], assume_utc=True
+            )
             data["time"] = timestamp
             data["event"] = report.copy()
             json_str += "{0}\n".format(json.dumps(data))

parsedmarc/utils.py

@@ -359,7 +359,7 @@ def timestamp_to_human(timestamp: int) -> str:
 
 
 def human_timestamp_to_datetime(
-    human_timestamp: str, *, to_utc: bool = False
+    human_timestamp: str, *, to_utc: bool = False, assume_utc: bool = False
 ) -> datetime:
     """
     Converts a human-readable timestamp into a Python ``datetime`` object
@@ -367,6 +367,11 @@ def human_timestamp_to_datetime(
     Args:
         human_timestamp (str): A timestamp string
         to_utc (bool): Convert the timestamp to UTC
+        assume_utc (bool): Treat a timestamp that carries no UTC offset as
+            UTC wall-clock time instead of local time. Pass this when the
+            string is known to be UTC (e.g. an ``arrival_date_utc`` value);
+            otherwise naive results are interpreted as local time by
+            ``datetime.astimezone()`` / ``datetime.timestamp()``.
 
     Returns:
         datetime: The converted timestamp
@@ -376,21 +381,29 @@ def human_timestamp_to_datetime(
     human_timestamp = parenthesis_regex.sub("", human_timestamp)
 
     dt = parse_date(human_timestamp)
+    if assume_utc and dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
     return dt.astimezone(timezone.utc) if to_utc else dt
 
 
-def human_timestamp_to_unix_timestamp(human_timestamp: str) -> int:
+def human_timestamp_to_unix_timestamp(
+    human_timestamp: str, *, assume_utc: bool = False
+) -> int:
     """
     Converts a human-readable timestamp into a UNIX timestamp
 
     Args:
         human_timestamp (str): A timestamp in `YYYY-MM-DD HH:MM:SS`` format
+        assume_utc (bool): Treat a timestamp that carries no UTC offset as
+            UTC wall-clock time instead of local time
 
     Returns:
-        float: The converted timestamp
+        int: The converted timestamp
     """
     human_timestamp = human_timestamp.replace("T", " ")
-    return int(human_timestamp_to_datetime(human_timestamp).timestamp())
+    return int(
+        human_timestamp_to_datetime(human_timestamp, assume_utc=assume_utc).timestamp()
+    )
 
 
 _IP_DB_PATH: str | None = None

tests/test_elastic.py

@@ -6,6 +6,7 @@ transformation logic — document construction, index naming, deduplication
 queries, error wrapping — without needing a running Elasticsearch cluster.
 """
 
+import time
 import unittest
 from unittest.mock import MagicMock, call, patch
 
@@ -21,6 +22,7 @@ from parsedmarc.elastic import (
     save_smtp_tls_report_to_elasticsearch,
     set_hosts,
 )
+from tests.tzutil import force_tz
 
 
 # ---------------------------------------------------------------------------
@@ -660,6 +662,25 @@ class TestSaveFailureReport(unittest.TestCase):
             index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
         self.assertIn("dmarc_failure-2024-01", index_calls)
 
+    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
+    def test_arrival_date_epoch_is_utc_regardless_of_host_timezone(self):
+        """arrival_date_utc is a UTC wall-clock string; the epoch-ms
+        value stored in the document (and used in the dedup query) must
+        be its true UTC epoch on any host. Regression test for
+        https://github.com/domainaware/parsedmarc/issues/811 (bug 1):
+        the naive parse used to shift the stored epoch by the host's
+        UTC offset."""
+        force_tz(self)
+
+        with (
+            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
+            patch("parsedmarc.elastic.Index"),
+            patch("parsedmarc.elastic._FailureReportDoc") as mock_doc_cls,
+        ):
+            save_failure_report_to_elasticsearch(_failure_report())
+        # Fixture arrival_date_utc is 2024-01-01 00:00:00 UTC.
+        self.assertEqual(mock_doc_cls.call_args.kwargs["arrival_date"], 1704067200000)
+
     def test_failure_search_index_with_suffix_and_prefix(self):
         """When both suffix and prefix are set, the dedup search
         pattern joins them onto BOTH dmarc_failure* and

tests/test_opensearch.py

@@ -6,6 +6,7 @@ transformation logic — document construction, index naming, deduplication
 queries, error wrapping — without needing a running OpenSearch cluster.
 """
 
+import time
 import unittest
 from unittest.mock import MagicMock, call, patch
 
@@ -21,6 +22,7 @@ from parsedmarc.opensearch import (
     save_smtp_tls_report_to_opensearch,
     set_hosts,
 )
+from tests.tzutil import force_tz
 
 
 # ---------------------------------------------------------------------------
@@ -658,6 +660,25 @@ class TestSaveFailureReport(unittest.TestCase):
             index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
         self.assertIn("dmarc_failure-2024-01", index_calls)
 
+    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
+    def test_arrival_date_epoch_is_utc_regardless_of_host_timezone(self):
+        """arrival_date_utc is a UTC wall-clock string; the epoch-ms
+        value stored in the document (and used in the dedup query) must
+        be its true UTC epoch on any host. Regression test for
+        https://github.com/domainaware/parsedmarc/issues/811 (bug 1):
+        the naive parse used to shift the stored epoch by the host's
+        UTC offset."""
+        force_tz(self)
+
+        with (
+            patch("parsedmarc.opensearch.Search", return_value=_empty_search()),
+            patch("parsedmarc.opensearch.Index"),
+            patch("parsedmarc.opensearch._FailureReportDoc") as mock_doc_cls,
+        ):
+            save_failure_report_to_opensearch(_failure_report())
+        # Fixture arrival_date_utc is 2024-01-01 00:00:00 UTC.
+        self.assertEqual(mock_doc_cls.call_args.kwargs["arrival_date"], 1704067200000)
+
     def test_failure_search_index_with_suffix_and_prefix(self):
         """When both suffix and prefix are set, the dedup search
         pattern joins them onto BOTH dmarc_failure* and

tests/test_splunk.py

@@ -1,10 +1,12 @@
 """Tests for parsedmarc.splunk"""
 
 import json
+import time
 import unittest
 from unittest.mock import MagicMock
 
 from parsedmarc.splunk import HECClient, SplunkError
+from tests.tzutil import force_tz
 
 
 def _aggregate_report():
@@ -323,6 +325,24 @@ class TestSaveFailureReportsToSplunk(unittest.TestCase):
         client.save_failure_reports_to_splunk([])
         client.session.post.assert_not_called()
 
+    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
+    def test_event_time_treats_arrival_date_utc_as_utc(self):
+        """arrival_date_utc is a UTC wall-clock string; the HEC event
+        `time` must be its true UTC epoch regardless of the host
+        timezone. Regression test for
+        https://github.com/domainaware/parsedmarc/issues/811 (bug 1):
+        the naive parse used to shift the epoch by the host's UTC
+        offset (-3600 s under Europe/Warsaw in January)."""
+        force_tz(self)
+
+        client = _client()
+        client.session = MagicMock()
+        client.session.post.return_value = _ok_response()
+        client.save_failure_reports_to_splunk(_failure_report())
+        event = json.loads(client.session.post.call_args.kwargs["data"].strip())
+        # Fixture arrival_date_utc is 2024-01-01 00:00:00 UTC.
+        self.assertEqual(event["time"], 1704067200)
+
     def test_non_zero_response_code_raises_splunk_error(self):
         client = _client()
         client.session = MagicMock()

tests/test_utils.py

@@ -2,6 +2,7 @@
 
 import os
 import tempfile
+import time
 import unittest
 from datetime import datetime, timezone
 from tempfile import NamedTemporaryFile
@@ -13,6 +14,7 @@ from expiringdict import ExpiringDict
 
 import parsedmarc
 import parsedmarc.utils
+from tests.tzutil import force_tz
 
 
 class Test(unittest.TestCase):
@@ -550,6 +552,58 @@ class TestUtilsDnsCaching(unittest.TestCase):
             self.assertIsNone(result)
 
 
+@unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
+class TestTimestampAssumeUtc(unittest.TestCase):
+    """Timestamp helpers must not re-interpret known-UTC strings as local
+    time. Per the Python docs, naive ``datetime.timestamp()`` and
+    ``datetime.astimezone()`` assume the naive value is *local* time
+    (https://docs.python.org/3/library/datetime.html#datetime.datetime.timestamp),
+    so a UTC wall-clock string like ``arrival_date_utc`` parsed naive comes
+    out skewed by the host's UTC offset. ``assume_utc=True`` attaches
+    ``timezone.utc`` instead. Regression tests for
+    https://github.com/domainaware/parsedmarc/issues/811 (bug 1)."""
+
+    # 2024-01-15 12:00:00 UTC
+    UTC_STRING = "2024-01-15 12:00:00"
+    TRUE_EPOCH = 1705320000
+
+    def setUp(self):
+        # Fixed non-UTC zone (UTC+1 in January) so the local-time
+        # misinterpretation this guards against would shift the epoch.
+        force_tz(self)
+
+    def testAssumeUtcYieldsAwareUtcDatetime(self):
+        dt = parsedmarc.utils.human_timestamp_to_datetime(
+            self.UTC_STRING, assume_utc=True
+        )
+        self.assertEqual(dt.tzinfo, timezone.utc)
+        self.assertEqual(int(dt.timestamp()), self.TRUE_EPOCH)
+
+    def testWithoutAssumeUtcNaiveIsLocal(self):
+        """Documents the default: a naive parse followed by .timestamp()
+        uses local time — off by one hour under Europe/Warsaw in January.
+        This is the behavior arrival_date_utc consumers must avoid."""
+        dt = parsedmarc.utils.human_timestamp_to_datetime(self.UTC_STRING)
+        self.assertIsNone(dt.tzinfo)
+        self.assertEqual(int(dt.timestamp()), self.TRUE_EPOCH - 3600)
+
+    def testUnixTimestampHelperAssumeUtc(self):
+        self.assertEqual(
+            parsedmarc.utils.human_timestamp_to_unix_timestamp(
+                self.UTC_STRING, assume_utc=True
+            ),
+            self.TRUE_EPOCH,
+        )
+
+    def testAssumeUtcDoesNotOverrideExplicitOffset(self):
+        """A timestamp that carries its own offset keeps it; assume_utc
+        only applies to naive parses."""
+        dt = parsedmarc.utils.human_timestamp_to_datetime(
+            "2024-01-15 13:00:00 +0100", assume_utc=True
+        )
+        self.assertEqual(int(dt.timestamp()), self.TRUE_EPOCH)
+
+
 class TestUtilsIpDbPaths(unittest.TestCase):
     """Tests for IP database path validation"""
 

tests/tzutil.py

@@ -0,0 +1,30 @@
+"""Shared timezone-forcing helper for timestamp regression tests."""
+
+import os
+import time
+import unittest
+
+
+def force_tz(testcase: unittest.TestCase, tz: str = "Europe/Warsaw") -> None:
+    """Set the process timezone to *tz* for the duration of *testcase*.
+
+    Registers a cleanup that restores the previous ``TZ`` value.
+    Requires POSIX ``time.tzset()``; guard callers with
+    ``@unittest.skipUnless(hasattr(time, "tzset"), ...)``.
+
+    The default zone is fixed and non-UTC (UTC+1 in January, UTC+2 in
+    summer), so code that wrongly interprets a UTC wall-clock string as
+    local time produces a shifted epoch under it.
+    """
+    old_tz = os.environ.get("TZ")
+    os.environ["TZ"] = tz
+    time.tzset()
+
+    def restore() -> None:
+        if old_tz is None:
+            os.environ.pop("TZ", None)
+        else:
+            os.environ["TZ"] = old_tz
+        time.tzset()
+
+    testcase.addCleanup(restore)