Adjust teamviewer plugin to account for daylight savings time in timestamps

dissect/target/plugins/apps/remoteaccess/teamviewer.py

@@ -132,17 +132,29 @@ def logs(self) -> Iterator[RemoteAccessLogRecord]:
             logfile = self.target.fs.path(logfile)
 
             start_date = None
+            prev_timestamp = None
+            fold = 0
             for line in logfile.open("rt", errors="replace"):
                 if not (line := line.strip()) or line.startswith("# "):
                     continue
 
                 if line.startswith("Start:"):
                     try:
                         start_date = parse_start(line)
+                        fold = 0
                     except Exception as e:
                         self.target.log.warning("Failed to parse Start message %r in %s", line, logfile)
                         self.target.log.debug("", exc_info=e)
 
+                    if start_date is None:
+                        continue
+
+                    # See whether the utcoffset with the two different timezones are the same
+                    target_start_date = start_date.replace(tzinfo=target_tz)
+                    if target_start_date.utcoffset() == start_date.utcoffset():
+                        # Adjust the start_date so it uses the timezone known to target throughtout
+                        start_date = target_start_date
+
                     continue
 
                 if not (match := RE_LOG.search(line)):
@@ -178,14 +190,23 @@ def logs(self) -> Iterator[RemoteAccessLogRecord]:
                     time += ".000000"
 
                 try:
-                    timestamp = datetime.strptime(f"{date} {time}", "%Y/%m/%d %H:%M:%S.%f").replace(
-                        tzinfo=start_date.tzinfo if start_date else target_tz
-                    )
+                    tz_info = start_date.tzinfo if start_date else target_tz
+                    timestamp = datetime.strptime(f"{date} {time}", "%Y/%m/%d %H:%M:%S.%f").replace(tzinfo=tz_info)
                 except Exception as e:
                     self.target.log.warning("Unable to parse timestamp %r in file %s", line, logfile)
                     self.target.log.debug("", exc_info=e)
                     timestamp = 0
 
+                if timestamp and prev_timestamp and prev_timestamp > timestamp:
+                    # We might currently be in a grey area where the dst period ended.
+                    # replace the fold value of the timestamp
+                    fold = 1
+
+                if timestamp:
+                    timestamp = timestamp.replace(fold=fold)
+
+                prev_timestamp = timestamp
+
                 yield self.RemoteAccessLogRecord(
                     ts=timestamp,
                     message=log.get("message"),
@@ -248,6 +269,7 @@ def parse_start(line: str) -> datetime | None:
 
         Start: 2021/11/11 12:34:56
         Start: 2024/12/31 01:02:03.123 (UTC+2:00)
+        Start: 2025/01/01 12:28:41.436 (UTC)
     """
     if match := RE_START.search(line):
         dt = match.groupdict()
@@ -257,13 +279,21 @@ def parse_start(line: str) -> datetime | None:
             dt["time"] = dt["time"].rsplit(".")[0]
 
         # Format timezone, e.g. "UTC+2:00" to "UTC+0200"
-        if dt["timezone"]:
-            name, operator, amount = re.split(r"(\+|\-)", dt["timezone"])
-            amount = int(amount.replace(":", ""))
-            dt["timezone"] = f"{name}{operator}{amount:0>4d}"
+        if timezone := dt["timezone"]:
+            identifier = " %Z%z"
+            # Handle just UTC timezone
+            if timezone.lower() == "utc":
+                timezone = " UTC+00:00"
+            else:
+                name, operator, amount = re.split(r"(\+|\-)", timezone)
+                amount = int(amount.replace(":", ""))
+                timezone = f" {name}{operator}{amount:0>4d}"
+        else:
+            timezone = ""
+            identifier = ""
 
         return datetime.strptime(  # noqa: DTZ007
-            f"{dt['date']} {dt['time']}" + (f" {dt['timezone']}" if dt["timezone"] else ""),
-            "%Y/%m/%d %H:%M:%S" + (" %Z%z" if dt["timezone"] else ""),
+            f"{dt['date']} {dt['time']}{timezone}",
+            f"%Y/%m/%d %H:%M:%S{identifier}",
         )
     return None

pyproject.toml

@@ -35,7 +35,7 @@ dependencies = [
     "dissect.regf>=3.13,<4",
     "dissect.util>=3,<4",
     "dissect.volume>=3.17,<4",
-    "flow.record~=3.21.0",
+    "flow.record>=3.22.dev10", #TODO: Update at release
     "structlog",
 ]
 dynamic = ["version"]

tests/helpers/test_modifier.py

@@ -63,11 +63,11 @@ def test_hash_path_records_with_paths(
     ):
         hashed_record = hash_function(target_win, record)
 
-    assert hashed_record.name == "test"
-    assert len(hashed_record.records) == expected_records
-    assert hashed_record.records[0] == record
+    assert hashed_record.__name__ == "test"
+    assert len(hashed_record.__records__) == expected_records
+    assert hashed_record.__records__[0] == record
 
-    for name, _record in zip(path_field_names, hashed_record.records[1:], strict=False):
+    for name, _record in zip(path_field_names, hashed_record.__records__[1:], strict=False):
         assert getattr(_record, f"{name}_resolved") is not None
         assert getattr(_record, f"{name}_digest").__dict__ == digest(HASHES).__dict__
 
@@ -151,6 +151,6 @@ def test_resolved_modifier(record: Record, target_win: Target, resolve_function:
 
     resolved_record = resolve_function(target_win, record)
 
-    for _record in resolved_record.records[1:]:
+    for _record in resolved_record.__records__[1:]:
         assert _record.name_resolved is not None
         assert not hasattr(_record, "name_digest")

tests/plugins/apps/remoteaccess/test_teamviewer.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from io import BytesIO
 from textwrap import dedent
 from typing import TYPE_CHECKING
+from unittest.mock import patch
 
-from dissect.target.plugins.apps.remoteaccess.teamviewer import TeamViewerPlugin
+import pytest
+
+from dissect.target.plugins.apps.remoteaccess.teamviewer import TeamViewerPlugin, parse_start
 from tests._utils import absolute_path
 
 if TYPE_CHECKING:
@@ -135,3 +138,52 @@ def test_teamviewer_incoming(target_win_users: Target, fs_win: VirtualFilesystem
     assert records[1].user == "Server"
     assert records[1].connection_type == "RemoteControl"
     assert records[1].connection_id == "{4BF22BA7-32BA-4F64-8755-97E6E45F9883}"
+
+
+def test_teamviewer_daylight_savings_time(target_win_tzinfo: Target, fs_win: VirtualFilesystem) -> None:
+    """Test whether the teamviewer plugin handles dst correctly."""
+    log = """
+    Start:              2025/10/26 02:50:32.134 (UTC+2:00)
+    2025/10/26 02:50:32.300  1234  5678 G1   Example DST timestamp
+    2025/10/26 02:00:03.400  1234  5678 G1   Example non DST timestamp
+    2025/10/26 02:30:03.400  1234  5678 G1   Example continued timestamp
+    Start:              2025/10/27 01:02:03.123 (UTC+1:00)
+    2025/10/27 01:02:03.500  1234  5678 G1   Example non DST timestamp
+    """
+    fs_win.map_file_fh("Program Files/TeamViewer/Teamviewer_Log.log", BytesIO(dedent(log).encode()))
+    # set timezone to something that has a dst time record
+    eu_timezone = target_win_tzinfo.datetime.tz("W. Europe Standard Time")
+    target_win_tzinfo.add_plugin(TeamViewerPlugin)
+
+    with patch.object(target_win_tzinfo.datetime, "_tzinfo", eu_timezone):
+        records = list(target_win_tzinfo.teamviewer.logs())
+    assert len(records) == 4
+
+    assert records[0].ts.astimezone(timezone.utc) == datetime(2025, 10, 26, 0, 50, 32, 300000, tzinfo=timezone.utc)
+    assert records[1].ts.astimezone(timezone.utc) == datetime(2025, 10, 26, 1, 0, 3, 400000, tzinfo=timezone.utc)
+    assert records[2].ts.astimezone(timezone.utc) == datetime(2025, 10, 26, 1, 30, 3, 400000, tzinfo=timezone.utc)
+    assert records[3].ts.astimezone(timezone.utc) == datetime(2025, 10, 27, 0, 2, 3, 500000, tzinfo=timezone.utc)
+
+
+@pytest.mark.parametrize(
+    argnames=("line", "expected_date"),
+    argvalues=[
+        pytest.param(
+            "Start: 2021/11/11 12:34:56",
+            datetime(2021, 11, 11, 12, 34, 56),  # noqa DTZ001
+            id="Parse withouth timezone",
+        ),
+        pytest.param(
+            "Start: 2024/12/31 01:02:03.123 (UTC+2:00)",
+            datetime(2024, 12, 31, 1, 2, 3, tzinfo=timezone(timedelta(seconds=7200))),
+            id="Parse (UTC+2:00)",
+        ),
+        pytest.param(
+            "Start: 2025/01/01 12:28:41.436 (UTC)",
+            datetime(2025, 1, 1, 12, 28, 41, tzinfo=timezone.utc),
+            id="Parse UTC without offset",
+        ),
+    ],
+)
+def test_teamviewer_parse_start(line: str, expected_date: datetime) -> None:
+    assert parse_start(line) == expected_date

tests/plugins/os/windows/test_tasks.py

@@ -171,8 +171,8 @@ def assert_at_task_grouped_padding(at_task_grouped: GroupedRecord) -> None:
 
 
 def assert_at_task_grouped_monthlydow(at_task_grouped: GroupedRecord) -> None:
-    assert at_task_grouped.records[0].enabled
-    assert at_task_grouped.records[1].trigger_enabled
+    assert at_task_grouped.__records__[0].enabled
+    assert at_task_grouped.__records__[1].trigger_enabled
     assert at_task_grouped.start_boundary == datetime.fromisoformat("2023-05-11 00:00:00+00:00")
     assert at_task_grouped.end_boundary == datetime.fromisoformat("2023-05-20 00:00:00+00:00")
     assert at_task_grouped.repetition_interval == "PT1M"
@@ -186,8 +186,8 @@ def assert_at_task_grouped_monthlydow(at_task_grouped: GroupedRecord) -> None:
 
 
 def assert_at_task_grouped_weekly(at_task_grouped: GroupedRecord) -> None:
-    assert at_task_grouped.records[0].enabled
-    assert at_task_grouped.records[1].trigger_enabled
+    assert at_task_grouped.__records__[0].enabled
+    assert at_task_grouped.__records__[1].trigger_enabled
     assert at_task_grouped.end_boundary == datetime.fromisoformat("2023-05-27 00:00:00+00:00")
     assert at_task_grouped.execution_time_limit == "P3D"
     assert at_task_grouped.repetition_duration == "PT1H"
@@ -201,8 +201,8 @@ def assert_at_task_grouped_weekly(at_task_grouped: GroupedRecord) -> None:
 
 
 def assert_at_task_grouped_monthly_date(at_task_grouped: GroupedRecord) -> None:
-    assert at_task_grouped.records[0].enabled
-    assert at_task_grouped.records[1].trigger_enabled
+    assert at_task_grouped.__records__[0].enabled
+    assert at_task_grouped.__records__[1].trigger_enabled
     assert at_task_grouped.day_of_month == [15]
     assert at_task_grouped.months_of_year == ["March", "May", "June", "July", "August", "October"]
     assert at_task_grouped.end_boundary == datetime.fromisoformat("2023-05-29 00:00:00+00:00")
@@ -214,8 +214,8 @@ def assert_at_task_grouped_monthly_date(at_task_grouped: GroupedRecord) -> None:
 
 
 def assert_xml_task_trigger_properties(xml_task: GroupedRecord) -> None:
-    assert xml_task.records[0].enabled
-    assert xml_task.records[1].trigger_enabled
+    assert xml_task.__records__[0].enabled
+    assert xml_task.__records__[1].trigger_enabled
     assert xml_task.days_between_triggers == 1
     assert xml_task.start_boundary == datetime.fromisoformat("2023-05-12 00:00:00+00:00")