Improve robustness of Sophos plugin

dissect/target/plugins/apps/av/sophos.py

@@ -70,20 +70,21 @@
         systems, the details field might contain a lot of text, it might
         contain stracktraces etc.
         """
-        try:
-            fh = self.target.fs.path(self.LOG_SOPHOS_HITMAN).open("rb")
-            db = sqlite3.SQLite3(fh)
-            alerts = list(filter(lambda t: t.name == "Alerts", db.tables()))[0]
-            for alert in alerts.rows():
-                yield HitmanAlertRecord(
-                    ts=wintimestamp(alert.Timestamp),  # already utc
-                    alert=alert.AlertType,
-                    description=alert.Description,
-                    details=alert.Details,
-                    _target=self.target,
-                )
-        except Exception as error:
-            self.target.log.error(f"Error occurred during reading alerts: {error}.")
+        if self.target.fs.path(self.LOG_SOPHOS_HITMAN).exists():
+            try:
+                fh = self.target.fs.path(self.LOG_SOPHOS_HITMAN).open("rb")
+                db = sqlite3.SQLite3(fh)
+                alerts = list(filter(lambda t: t.name == "Alerts", db.tables()))[0]
+                for alert in alerts.rows():
+                    yield HitmanAlertRecord(
+                        ts=wintimestamp(alert.Timestamp),  # already utc
+                        alert=alert.AlertType,
+                        description=alert.Description,
+                        details=alert.Details,
+                        _target=self.target,
+                    )
+            except Exception as error:
+                self.target.log.error("Error occurred during reading alerts: %r.", error)
 
     @export(record=SophosLogRecord)
     def sophoshomelogs(self) -> Iterator[SophosLogRecord]:
@@ -98,22 +99,30 @@
             path (path): Path to the infected file (if available).
 
         """
-        log = self.target.fs.path(self.LOG_SOPHOS_HOME).open("rt", 0, "utf-16le")
-        while line := log.readline():
-            if line.find(self.MARKER_INFECTION) > -1:
-                try:
-                    ts, json_data = line.split(" ", maxsplit=2)
-                    details = json.loads(json_data)
-
-                    path_to_infected_file = None
-                    if targets := details.get("targets", None):
-                        path_to_infected_file = targets[0].get("file_path", None)
-
-                    yield SophosLogRecord(
-                        ts=ts,
-                        description=details.get("threat_name", details),
-                        path=self.target.fs.path(path_to_infected_file),
-                        _target=self.target,
-                    )
-                except Exception as error:
-                    self.target.log.warning(f"Error: {error} on log line: {line}.")
+        if self.target.fs.path(self.LOG_SOPHOS_HOME).exists():
+            for line in self.target.fs.path(self.LOG_SOPHOS_HOME).open("rt", 0, "utf-16le"):
+                if line.find(self.MARKER_INFECTION) > -1:
+                    try:
+                        ts, json_data = line.split(" ", maxsplit=2)
+                        details = json.loads(json_data)
+
+                        path_to_infected_file = None
+                        if targets := details.get("targets", None):
+                            path_to_infected_file = targets[0].get("file_path", None)
+
+                        # < 3.9.4.1
+                        if path := details.get("file_path"):
+                            path_to_infected_file = path
+                        # > 3.10.3
+                        elif targets := details.get("targets", None):
+                            path_to_infected_file = targets[0].get("file_path", None)
+                            path = (self.target.fs.path(path_to_infected_file),)
+
+                        yield SophosLogRecord(
+                            ts=ts,
+                            description=details.get("threat_name", details),
+                            path=self.target.fs.path(path_to_infected_file),
+                            _target=self.target,
+                        )
+                    except Exception as error:
+                        self.target.log.warning("Error: %r on log line: %s.", error, line)

tests/plugins/apps/av/test_sophos.py

@@ -24,8 +24,21 @@ def test_sophos_hitman_plugin_log(target_win: Target, fs_win: VirtualFilesystem)
     assert log.details.find("LOVE-LETTER-FOR-YOU.TXT.vbs") > -1
 
 
-def test_sophos_home_plugin_log(target_win: Target, fs_win: VirtualFilesystem) -> None:
-    log_file = absolute_path("_data/plugins/apps/av/sophos/Clean.log")
+def test_sophos_home_plugin_log_3_9_4_1(target_win: Target, fs_win: VirtualFilesystem) -> None:
+    log_file = absolute_path("_data/plugins/apps/av/sophos/Clean-3.9.4.1.log")
+    fs_win.map_file("ProgramData/Sophos/Clean/Logs/Clean.log", log_file)
+    target_win.add_plugin(SophosPlugin)
+    logs = list(target_win.sophos.sophoshomelogs())
+    assert len(logs) == 1
+    log = logs[0]
+    assert log.ts == dt("2023-06-14T10:46:56.235Z")
+    assert isinstance(log, type(SophosLogRecord()))
+    assert log.description == "EICAR-AV-Test"
+    assert str(log.path) == "C:\\eicar_com.zip"
+
+
+def test_sophos_home_plugin_log_3_10_3(target_win: Target, fs_win: VirtualFilesystem) -> None:
+    log_file = absolute_path("_data/plugins/apps/av/sophos/Clean-3.10.3.log")
     fs_win.map_file("ProgramData/Sophos/Clean/Logs/Clean.log", log_file)
     target_win.add_plugin(SophosPlugin)
     logs = list(target_win.sophos.sophoshomelogs())