Fix TeamViewer timezone parsing

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

@@ -4,7 +4,6 @@
 
 from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.helpers.descriptor_extensions import UserRecordDescriptorExtension
-from dissect.target.helpers.fsutil import TargetPath
 from dissect.target.helpers.record import create_extended_descriptor
 from dissect.target.plugin import export
 from dissect.target.plugins.apps.remoteaccess.remoteaccess import (
@@ -13,7 +12,33 @@
 )
 from dissect.target.plugins.general.users import UserDetails
 
-START_PATTERN = re.compile(r"^(\d{2}|\d{4})/")
+RE_LOG = re.compile(
+    r"""
+        ^
+        (?P<date>(\d{2,4}\/)?\d{2}\/\d{2})                  # YYYY/MM/DD or YY/MM/DD or MM/DD
+        \s
+        (?P<time>\d{2}\:\d{2}\:\d{2}(?:[\.\:]\d{3})?)       # HH:MM:SS or HH:MM:SS.FFF or HH:MM:SS:FFF
+        \s+
+        (?P<message>.+)
+        $
+    """,
+    re.VERBOSE,
+)
+RE_START = re.compile(
+    r"""
+        ^Start\:
+        \s+
+        (?P<date>\S+)
+        \s
+        (?P<time>\S+)
+        (
+            \s
+            \((?P<timezone>\S+)\)                           # UTC+2:00
+        )?
+        $
+    """,
+    re.VERBOSE,
+)
 
 
 class TeamViewerPlugin(RemoteAccessPlugin):
@@ -45,82 +70,141 @@
     def __init__(self, target):
         super().__init__(target)
 
-        self.logfiles: list[list[TargetPath, UserDetails]] = []
+        self.logfiles: set[tuple[str, UserDetails | None]] = set()
 
         # Find system service log files.
         for log_glob in self.SYSTEM_GLOBS:
             for logfile in self.target.fs.glob(log_glob):
-                self.logfiles.append([logfile, None])
+                self.logfiles.add((logfile, None))
 
         # Find user log files.
         for user_details in self.target.user_details.all_with_home():
             for log_glob in self.USER_GLOBS:
                 for logfile in user_details.home_path.glob(log_glob):
-                    self.logfiles.append([logfile, user_details])
+                    self.logfiles.add((logfile, user_details))
 
     def check_compatible(self) -> None:
         if not len(self.logfiles):
-            raise UnsupportedPluginError("No Teamviewer logs found")
+            raise UnsupportedPluginError("No Teamviewer logs found on target")
 
     @export(record=RemoteAccessLogRecord)
     def logs(self) -> Iterator[RemoteAccessLogRecord]:
         """Yield TeamViewer client logs.
 
         TeamViewer is a commercial remote desktop application. An adversary may use it to gain persistence on a system.
         """
+
+        try:
+            target_tz = self.target.datetime.tzinfo
+
+        except Exception as e:
+            self.target.log.warning("Unable to determine target timezone, assuming UTC if no timezone is found in logs")
+            self.target.log.debug("", exc_info=e)
+            target_tz = timezone.utc
+
         for logfile, user_details in self.logfiles:
             logfile = self.target.fs.path(logfile)
 
             start_date = None
-            with logfile.open("rt") as file:
+            with logfile.open("rt") as fh:
                 while True:
                     try:
-                        line = file.readline()
-                    except UnicodeDecodeError:
+                        line = fh.readline()
+                    except UnicodeDecodeError as e:
+                        self.target.log.warning("Unable to parse log line in %s: %s", logfile, e)
+                        self.target.log.debug("", exc_info=e)
                         continue
 
-                    # End of file, quit while loop
                     if not line:
                         break
 
-                    line = line.strip()
-
-                    # Skip empty lines
-                    if not line:
+                    if not (line := line.strip()) or line.startswith("# "):
                         continue
-                    # Older logs first mention the start time and then leave out the year
+
                     if line.startswith("Start:"):
-                        start_date = datetime.strptime(line.split()[1], "%Y/%m/%d")
+                        try:
+                            start_date = parse_start(line)
+                        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)
+
+                        continue
 
-                    # Sometimes there are weird, mult-line/pretty print log messages.
-                    # We only parse the start line which starts with year (%Y/) or month (%m/)
-                    if not re.match(START_PATTERN, line):
+                    if not (match := RE_LOG.search(line)):
+                        self.target.log.warning("Skipping TeamViewer log line %r in %s", line, logfile)
                         continue
 
-                    ts_day, ts_time, message = line.split(" ", 2)
-                    ts_time = ts_time.split(".")[0]
+                    log = match.groupdict()
+                    date = log["date"]
+                    time = log["time"]
 
-                    # Correct for use of : as millisecond separator
-                    if ts_time.count(":") > 2:
-                        ts_time = ":".join(ts_time.split(":")[:3])
-                    # Correct for missing year in date
-                    if ts_day.count("/") == 1:
+                    # Older TeamViewer versions first mention the start time and then leave out the year,
+                    # so we have to correct for the missing year in date.
+                    if date.count("/") == 1:
                         if not start_date:
-                            self.target.log.debug("Missing year in log line, skipping line.")
+                            self.target.log.warning("Missing year in log line, skipping line %r in %s", line, logfile)
                             continue
-                        ts_day = f"{start_date.year}/{ts_day}"
+                        date = f"{start_date.year}/{log['date']}"
+
                     # Correct for year if short notation for 2000 is used
-                    if ts_day.count("/") == 2 and len(ts_day.split("/")[0]) == 2:
-                        ts_day = "20" + ts_day
+                    if date.count("/") == 2 and len(date.split("/")[0]) == 2:
+                        date = "20" + date
 
-                    timestamp = datetime.strptime(f"{ts_day} {ts_time}", "%Y/%m/%d %H:%M:%S").replace(
-                        tzinfo=timezone.utc
-                    )
+                    # Correct for ``:`` separator of milliseconds
+                    if time.count(":") == 3:
+                        hms, _, ms = time.rpartition(":")
+                        time = f"{hms}.{ms}"
+
+                    # Convert milliseconds to microseconds
+                    if "." in time:
+                        hms, _, ms = time.rpartition(".")
+                        time = f"{hms}.{ms:06}"
+                    else:
+                        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
+                        )
+                    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
 
                     yield self.RemoteAccessLogRecord(
                         ts=timestamp,
-                        message=message,
+                        message=log.get("message"),
                         source=logfile,
                         _target=self.target,
                         _user=user_details.user if user_details else None,
                     )
+
+
+def parse_start(line: str, tz_fallback: timezone = timezone.utc) -> datetime | None:
+    """TeamViewer ``Start`` messages can be formatted in different ways
+    and might contain the timezone offset of all timestamps.
+
+    .. code-block::
+
+        Start: 2021/11/11 12:34:56
+        Start: 2024/12/31 01:02:03.123 (UTC+2:00)
+    """
+
+    if match := RE_START.search(line):
+        dt = match.groupdict()
+
+        # Drop milliseconds
+        if "." in dt["time"]:
+            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:04d}"
+
+        start_date = datetime.strptime(
+            f"{dt['date']} {dt['time']}" + (f" {dt['timezone']}" if dt["timezone"] else ""),
+            "%Y/%m/%d %H:%M:%S" + (" %Z%z" if dt["timezone"] else ""),
+        )
+        return start_date

tests/plugins/apps/remoteaccess/test_teamviewer.py

@@ -1,10 +1,14 @@
 from datetime import datetime, timezone
+from io import BytesIO
+from textwrap import dedent
 
+from dissect.target.filesystem import VirtualFilesystem
 from dissect.target.plugins.apps.remoteaccess.teamviewer import TeamViewerPlugin
+from dissect.target.target import Target
 from tests._utils import absolute_path
 
 
-def test_teamviewer_plugin_global_log(target_win_users, fs_win):
+def test_teamviewer_global_log(target_win_users: Target, fs_win: VirtualFilesystem) -> None:
     teamviewer_logfile = absolute_path("_data/plugins/apps/remoteaccess/teamviewer/TestTeamviewer.log")
     target_logfile_name = "/sysvol/Program Files/TeamViewer/TestTeamviewer.log"
 
@@ -16,16 +20,15 @@ def test_teamviewer_plugin_global_log(target_win_users, fs_win):
     records = list(tvp.logs())
     assert len(records) == 4
 
-    record = records[0]
-    assert record.ts == datetime(2021, 11, 11, 12, 34, 56, tzinfo=timezone.utc)
-    assert record.message == "Strip the headers, trace the source!"
-    assert record.source == target_logfile_name
-    assert record.username is None
-    assert record.user_id is None
-    assert record.user_home is None
+    assert records[0].ts == datetime(2021, 11, 11, 12, 34, 56, tzinfo=timezone.utc)
+    assert records[0].message == "Strip the headers, trace the source!"
+    assert records[0].source == target_logfile_name
+    assert records[0].username is None
+    assert records[0].user_id is None
+    assert records[0].user_home is None
 
 
-def test_teamviewer_plugin_user_log(target_win_users, fs_win):
+def test_teamviewer_user_log(target_win_users: Target, fs_win: VirtualFilesystem) -> None:
     teamviewer_logfile = absolute_path("_data/plugins/apps/remoteaccess/teamviewer/TestTeamviewer.log")
     user_details = target_win_users.user_details.find(username="John")
     target_logfile_name = f"{user_details.home_path}/appdata/roaming/teamviewer/teamviewer_TEST_logfile.log"
@@ -38,16 +41,15 @@ def test_teamviewer_plugin_user_log(target_win_users, fs_win):
     records = list(tvp.logs())
     assert len(records) == 4
 
-    record = records[0]
-    assert record.ts == datetime(2021, 11, 11, 12, 34, 56, tzinfo=timezone.utc)
-    assert record.message == "Strip the headers, trace the source!"
-    assert record.source == target_logfile_name
-    assert record.username == user_details.user.name
-    assert record.user_id == user_details.user.sid
-    assert record.user_home == user_details.user.home
+    assert records[0].ts == datetime(2021, 11, 11, 12, 34, 56, tzinfo=timezone.utc)
+    assert records[0].message == "Strip the headers, trace the source!"
+    assert records[0].source == target_logfile_name
+    assert records[0].username == user_details.user.name
+    assert records[0].user_id == user_details.user.sid
+    assert records[0].user_home == user_details.user.home
 
 
-def test_teamviewer_plugin_special_date_parsing(target_win_users, fs_win):
+def test_teamviewer_special_date_parsing(target_win_users: Target, fs_win: VirtualFilesystem) -> None:
     teamviewer_logfile = absolute_path("_data/plugins/apps/remoteaccess/teamviewer/TestTeamviewer.log")
     user_details = target_win_users.user_details.find(username="John")
     target_logfile_name = f"{user_details.home_path}/appdata/roaming/teamviewer/teamviewer_TEST_logfile.log"
@@ -60,14 +62,42 @@ def test_teamviewer_plugin_special_date_parsing(target_win_users, fs_win):
     records = list(tvp.logs())
     assert len(records) == 4
 
-    record_2 = records[1]
-    assert record_2.ts == datetime(2021, 11, 11, 12, 35, 55, tzinfo=timezone.utc)
-    assert record_2.message == "Should be year 2021"
+    assert records[1].ts == datetime(2021, 11, 11, 12, 35, 55, 465000, tzinfo=timezone.utc)
+    assert records[1].message == "Should be year 2021"
 
-    record_3 = records[2]
-    assert record_3.ts == datetime(2021, 11, 11, 12, 36, 11, tzinfo=timezone.utc)
-    assert record_3.message == "Should discard the milliseconds properly"
+    assert records[2].ts == datetime(2021, 11, 11, 12, 36, 11, 111000, tzinfo=timezone.utc)
+    assert records[2].message == "Should discard the milliseconds properly"
 
-    record_4 = records[3]
-    assert record_4.ts == datetime(2021, 11, 11, 12, 37, 00, tzinfo=timezone.utc)
-    assert record_4.message == "Should be year 2021"
+    assert records[3].ts == datetime(2021, 11, 11, 12, 37, 0, 0, tzinfo=timezone.utc)
+    assert records[3].message == "Should be year 2021"
+
+
+def test_teamviewer_timezone(target_win_users: Target, fs_win: VirtualFilesystem) -> None:
+    """test if we correctly set the timezone in teamviewer logs."""
+
+    log = """
+    Start:          2024/12/31 01:02:03.123 (UTC+2:00)
+    2024/12/31 01:02:03.200  1234  5678 G1   LanguageControl: device language is 'enUS'
+    2024/12/31 01:02:03.300  1234  5678 G1   Example message 1
+    2024/12/31 01:02:03.400  1234  5678 G1   Example message 2
+    2024/12/31 01:02:03.500  1234  5678 G1   Example message 3
+    2024/12/31 01:02:03.600  1234  5678 G1   Example message 4
+    2024/12/31 01:02:03.700  1234  5678 G1!! Example message 5
+    2024/12/31 01:02:03.800  1234  5678 G1   TeamViewer is going offline!
+    2024/12/31 01:02:03.900  1234  5678 G1   NetworkControl shutdown done
+    """
+    fs_win.map_file_fh(
+        "Users/John/AppData/Roaming/TeamViewer/TeamViewer1337_Logfile.log", BytesIO(dedent(log).encode())
+    )
+
+    target_win_users.add_plugin(TeamViewerPlugin)
+
+    records = sorted(list(target_win_users.teamviewer.logs()), key=lambda r: r.ts)
+
+    assert len(records) == 8
+
+    # 01:02:03 with UTC+0200 becomes 23:02:03 UTC
+    assert records[0].ts == datetime(2024, 12, 30, 23, 2, 3, 200000, tzinfo=timezone.utc)
+    assert records[0].message == "1234  5678 G1   LanguageControl: device language is 'enUS'"
+    assert records[0].source == "C:\\Users\\John\\AppData\\Roaming\\TeamViewer\\TeamViewer1337_Logfile.log"
+    assert records[0].username == "John"