Fix aggregate-report timestamp skew on non-UTC hosts (#819) (#821)

record["interval_begin"]/["interval_end"] are UTC wall-clock strings
(produced in __init__.py by strftime() on a datetime already converted
via human_timestamp_to_datetime(..., to_utc=True)), but the
Elasticsearch and OpenSearch per-record save loops and the Splunk HEC
aggregate-report event builder re-parsed them without assume_utc=True,
so on non-UTC hosts they were misinterpreted as local time and shifted
by the host's UTC offset -- shifting the stored date_begin/date_end,
the daily/monthly index date, and the Splunk event time. Verified by
parsing a real sample report under TZ=Europe/Warsaw vs TZ=UTC and by
reproducing the exact skew via human_timestamp_to_unix_timestamp().

This is the same class of bug fixed for arrival_date_utc in #811/#812
(commit cdda5da); the assume_utc keyword already exists on
human_timestamp_to_datetime()/human_timestamp_to_unix_timestamp() and
is reused here rather than reimplemented.

The issue also proposed changes to the report-level begin_date/end_date
parses in elastic.py/opensearch.py (~line 455) and to postgres.py.
Both were investigated and left unchanged: the report-level strings
are genuinely host-local time (from timestamp_to_human() ->
datetime.fromtimestamp()), so their existing no-assume_utc round-trip
is already correct on a single host -- postgres.py already gets this
right via two distinct helpers (_naive_local_to_timestamptz vs
_ensure_utc_suffix). Adding assume_utc to the report-level parses
would introduce a skew rather than fix one.

Fixes https://github.com/domainaware/parsedmarc/issues/819

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

CHANGELOG.md

@@ -10,6 +10,7 @@
 
 - **`parse_email()` no longer crashes with `KeyError: 'Headers'` on messages whose `From` header is present but empty/unparseable** (e.g. a bare `From:` line). mailparser omits `"from"` from `mail_json` for such messages, and the fallback read `parsed_email["Headers"]` — a key that is never set; the parsed headers are stored under lowercase `"headers"` (see the assignment at the top of `parse_email()`). The fallback now reads the correct key and treats an empty parsed header list the same as a missing header, yielding `from=None`.
 - **A failed IPinfo API token probe no longer logs "IPinfo API configured".** `configure_ipinfo_api(..., probe=True)` documents that non-fatal probe errors are logged as warnings with the token still accepted, but `_ipinfo_api_lookup()` returns `None` on network errors instead of raising, so the probe's exception handler never fired and a probe that couldn't reach the API logged the success message. The probe now checks the lookup result and logs a warning when verification failed. Invalid tokens (401/403) still raise `InvalidIPinfoAPIKey`.
+- **Aggregate-report record timestamps are no longer skewed by the host's UTC offset in the Elasticsearch, OpenSearch, and Splunk HEC outputs** ([#819](https://github.com/domainaware/parsedmarc/issues/819)). A record's `interval_begin`/`interval_end` are UTC wall-clock strings (converted to UTC at parse time), but the ES/OpenSearch save paths re-parsed them as local time, shifting the stored `date_begin`/`date_end` fields and the daily/monthly index date by the host's UTC offset on any non-UTC host, and the Splunk HEC output computed the aggregate event `time` the same way. The three call sites now pass `assume_utc=True`, the same treatment `arrival_date_utc` received in the #811 fix. The report-level `begin_date`/`end_date` parses (ES/OpenSearch dedup query, PostgreSQL output) were investigated and left unchanged — those strings are genuinely local time, so their existing round-trip is already correct.
 
 ## 10.2.1
 

parsedmarc/elastic.py

@@ -510,8 +510,12 @@ def save_aggregate_report_to_elasticsearch(
     )
 
     for record in aggregate_report["records"]:
-        begin_date = human_timestamp_to_datetime(record["interval_begin"], to_utc=True)
-        end_date = human_timestamp_to_datetime(record["interval_end"], to_utc=True)
+        begin_date = human_timestamp_to_datetime(
+            record["interval_begin"], to_utc=True, assume_utc=True
+        )
+        end_date = human_timestamp_to_datetime(
+            record["interval_end"], to_utc=True, assume_utc=True
+        )
         normalized_timespan = record["normalized_timespan"]
 
         if monthly_indexes:

parsedmarc/opensearch.py

@@ -510,8 +510,12 @@ def save_aggregate_report_to_opensearch(
     )
 
     for record in aggregate_report["records"]:
-        begin_date = human_timestamp_to_datetime(record["interval_begin"], to_utc=True)
-        end_date = human_timestamp_to_datetime(record["interval_end"], to_utc=True)
+        begin_date = human_timestamp_to_datetime(
+            record["interval_begin"], to_utc=True, assume_utc=True
+        )
+        end_date = human_timestamp_to_datetime(
+            record["interval_end"], to_utc=True, assume_utc=True
+        )
         normalized_timespan = record["normalized_timespan"]
 
         if monthly_indexes:

parsedmarc/splunk.py

@@ -122,8 +122,10 @@ class HECClient(object):
                     new_report["spf_results"] = record["auth_results"]["spf"]
 
                 data["sourcetype"] = "dmarc:aggregate"
+                # interval_begin 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(
-                    new_report["interval_begin"]
+                    new_report["interval_begin"], assume_utc=True
                 )
                 data["time"] = timestamp
                 data["event"] = new_report.copy()

tests/test_elastic.py

@@ -578,6 +578,30 @@ class TestSaveAggregateReport(unittest.TestCase):
         search_index = mock_search_cls.call_args.kwargs["index"]
         self.assertIn("cust_dmarc_aggregate_tenant_a*", search_index)
 
+    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
+    def test_interval_dates_are_utc_regardless_of_host_timezone(self):
+        """interval_begin/interval_end are UTC wall-clock strings (already
+        converted to UTC at parse time in __init__.py); the index-date
+        bucketing and stored date_begin/date_end must use their true UTC
+        epoch on any host. Regression test for
+        https://github.com/domainaware/parsedmarc/issues/819: the naive
+        parse used to shift the stored epoch (and therefore the index
+        date) by the host's UTC offset."""
+        force_tz(self)
+        with (
+            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+            patch("parsedmarc.elastic._AggregateReportDoc") as mock_doc_cls,
+        ):
+            mock_index_cls.return_value.exists.return_value = True
+            save_aggregate_report_to_elasticsearch(_aggregate_report())
+            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
+        self.assertIn("dmarc_aggregate-2024-01-15", index_calls)
+        # Fixture begin_date/interval_begin is 2024-01-15 00:00:00 UTC.
+        self.assertEqual(
+            mock_doc_cls.call_args.kwargs["date_begin"].timestamp(), 1705276800
+        )
+
 
 class TestAggregateDocPassedDmarc(unittest.TestCase):
     """The _AggregateReportDoc.save() override derives passed_dmarc — the

tests/test_opensearch.py

@@ -578,6 +578,30 @@ class TestSaveAggregateReport(unittest.TestCase):
         search_index = mock_search_cls.call_args.kwargs["index"]
         self.assertIn("cust_dmarc_aggregate_tenant_a*", search_index)
 
+    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
+    def test_interval_dates_are_utc_regardless_of_host_timezone(self):
+        """interval_begin/interval_end are UTC wall-clock strings (already
+        converted to UTC at parse time in __init__.py); the index-date
+        bucketing and stored date_begin/date_end must use their true UTC
+        epoch on any host. Regression test for
+        https://github.com/domainaware/parsedmarc/issues/819: the naive
+        parse used to shift the stored epoch (and therefore the index
+        date) by the host's UTC offset."""
+        force_tz(self)
+        with (
+            patch("parsedmarc.opensearch.Search", return_value=_empty_search()),
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+            patch("parsedmarc.opensearch._AggregateReportDoc") as mock_doc_cls,
+        ):
+            mock_index_cls.return_value.exists.return_value = True
+            save_aggregate_report_to_opensearch(_aggregate_report())
+            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
+        self.assertIn("dmarc_aggregate-2024-01-15", index_calls)
+        # Fixture begin_date/interval_begin is 2024-01-15 00:00:00 UTC.
+        self.assertEqual(
+            mock_doc_cls.call_args.kwargs["date_begin"].timestamp(), 1705276800
+        )
+
 
 class TestAggregateDocPassedDmarc(unittest.TestCase):
     """The _AggregateReportDoc.save() override derives passed_dmarc — the

tests/test_splunk.py

@@ -294,6 +294,21 @@ class TestSaveAggregateReportsToSplunk(unittest.TestCase):
             client.save_aggregate_reports_to_splunk(_aggregate_report())
         self.assertIn("network", str(ctx.exception))
 
+    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
+    def test_event_time_treats_interval_begin_as_utc(self):
+        """interval_begin 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/819: the naive
+        parse used to shift the epoch by the host's UTC offset."""
+        force_tz(self)
+        client = _client()
+        client.session = MagicMock()
+        client.session.post.return_value = _ok_response()
+        client.save_aggregate_reports_to_splunk(_aggregate_report())
+        event_wrapper = json.loads(client.session.post.call_args.kwargs["data"].strip())
+        self.assertEqual(event_wrapper["time"], 1704067200)
+
 
 class TestSaveFailureReportsToSplunk(unittest.TestCase):
     def test_sends_one_event_per_report(self):