Change the iis plugin log_dirs to check for the presence of W3SVC* directories (#1460)

The previous option went through too many log files, even those that are
not compatible with the IIS plugin. So this commit makes it more
specific.

The commit also includes:
- Continuing to the next file if it fails to parse a line
- Skip trying to parse the config file if it doesn't exist
- Avoid a possibility of calling a None value instead of a function

Author: Miauwkeru

dissect/target/plugins/apps/webserver/iis.py

@@ -66,12 +66,11 @@ class IISLogsPlugin(WebserverPlugin):
 
     APPLICATION_HOST_CONFIG = "%windir%/system32/inetsrv/config/applicationHost.config"
 
-    DEFAULT_LOG_PATHS = (
-        "%windir%\\System32\\LogFiles\\W3SVC*\\*.log",
-        "sysvol\\Windows.old\\Windows\\System32\\LogFiles\\W3SVC*\\*.log",
-        "sysvol\\inetpub\\logs\\LogFiles\\*.log",
-        "sysvol\\inetpub\\logs\\LogFiles\\W3SVC*\\*.log",
-        "sysvol\\Resources\\Directory\\*\\LogFiles\\Web\\W3SVC*\\*.log",
+    DEFAULT_LOG_DIRS = (
+        "%windir%\\System32\\LogFiles\\W3SVC*",
+        "sysvol\\Windows.old\\Windows\\System32\\LogFiles\\W3SVC*",
+        "sysvol\\inetpub\\logs\\LogFiles",
+        "sysvol\\Resources\\Directory\\*\\LogFiles\\Web\\W3SVC*",
     )
 
     __namespace__ = "iis"
@@ -91,28 +90,12 @@ def log_dirs(self) -> dict[str, set[Path]]:
         if (sysvol_files := self.target.fs.path("sysvol/files")).exists():
             dirs["auto"].add(sysvol_files)
 
-        try:
-            xml_data = ElementTree.fromstring(self.config.read_bytes(), forbid_dtd=True)
-            for log_file_element in xml_data.findall("*/sites/*/logFile"):
-                log_format = log_file_element.get("logFormat") or "W3C"
-                if log_dir := log_file_element.get("directory"):
-                    if log_format not in dirs:
-                        self.target.log.warning("Unsupported log format %s, skipping %s", log_format, log_dir)
-                        continue
-                    dirs[log_format].add(self.target.resolve(log_dir))
-
-        except (ElementTree.ParseError, FileNotFoundError) as e:
-            self.target.log.warning("Error while parsing %s", self.config)
-            self.target.log.debug("", exc_info=e)
+        if self.config.exists():
+            self._read_config_log_paths(dirs)
 
-        for log_path in self.DEFAULT_LOG_PATHS:
-            try:
-                # later on we use */*.log to collect the files, so we need to move up 2 levels
-                log_path = self.target.expand_env(log_path)
-                log_dir = self.target.fs.path(log_path).parents[1]
-            except IndexError:
-                self.target.log.info("Incompatible path found: %s", log_path)
-                continue
+        for log_dir in self.DEFAULT_LOG_DIRS:
+            log_dir = self.target.expand_env(log_dir)
+            log_dir = self.target.fs.path(log_dir)
 
             if not has_glob_magic(str(log_dir)) and log_dir.exists():
                 dirs["auto"].add(log_dir)
@@ -132,6 +115,21 @@ def _get_paths(self) -> Iterator[Path]:
     def _get_auxiliary_paths(self) -> Iterator[Path]:
         yield from {self.config}
 
+    def _read_config_log_paths(self, dirs: dict[str, set[str]]) -> None:
+        try:
+            xml_data = ElementTree.fromstring(self.config.read_bytes(), forbid_dtd=True)
+            for log_file_element in xml_data.findall("*/sites/*/logFile"):
+                log_format = log_file_element.get("logFormat") or "W3C"
+                if log_dir := log_file_element.get("directory"):
+                    if log_format not in dirs:
+                        self.target.log.warning("Unsupported log format %s, skipping %s", log_format, log_dir)
+                        continue
+                    dirs[log_format].add(self.target.resolve(log_dir))
+
+        except (ElementTree.ParseError, FileNotFoundError) as e:
+            self.target.log.warning("Error while parsing %s", self.config)
+            self.target.log.debug("", exc_info=e)
+
     @export(record=BasicRecordDescriptor)
     def logs(self) -> Iterator[TargetRecordDescriptor]:
         """Return contents of IIS (v7 and above) log files.
@@ -142,22 +140,28 @@ def logs(self) -> Iterator[TargetRecordDescriptor]:
         Supported log formats: IIS, W3C.
         """
 
+        # We handle direct files here because _get_paths cannot select (filter) on the type of logfile.
+        if self.target.is_direct:
+            for log_file in self.get_paths():
+                yield from parse_autodetect_format_log(self.target, log_file)
+            # If we use the direct loader, there are no other files available.
+            return
+
         parsers = {
             "W3C": parse_w3c_format_log,
             "IIS": parse_iis_format_log,
             "auto": parse_autodetect_format_log,
         }
 
-        for format in ("IIS", "W3C", "auto"):
+        for format, parser in parsers.items():
             for log_dir in self.log_dirs.get(format, ()):
-                for log_file in log_dir.glob("*/*.log"):
+                for log_file in log_dir.rglob("*.log"):
                     self.target.log.info("Processing IIS log file %s in %s format", log_file, format)
-                    yield from parsers[format](self.target, log_file)
-
-        # We handle direct files here because _get_paths cannot select (filter) on the type of logfile.
-        if self.target.is_direct:
-            for log_file in self.get_paths():
-                yield from parse_autodetect_format_log(self.target, log_file)
+                    try:
+                        yield from parser(self.target, log_file)
+                    except Exception as e:
+                        self.target.log.error("Issue processing log file %s in %s format", log_file, format)  # noqa TRY400
+                        self.target.log.debug("", exc_info=e)
 
     @export(record=WebserverAccessLogRecord)
     def access(self) -> Iterator[WebserverAccessLogRecord]:
@@ -245,6 +249,7 @@ def parse_w3c_format_log(target: Target, path: Path) -> Iterator[TargetRecordDes
 
         if not record_descriptor:
             target.log.warning("Comment line with the fields defined should come before the values, skipping: %r", line)
+            continue
 
         raw = replace_dash_with_none(dict(zip(fields, values, strict=False)))
 

tests/plugins/apps/webserver/test_iis.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timedelta, timezone
+from io import BytesIO
 from pathlib import Path
 from typing import TYPE_CHECKING
 from unittest.mock import mock_open, patch
@@ -219,6 +221,26 @@ def test_iis_access_noconfig(
     assert len(results) > 0
 
 
+def test_iis_failed_to_parse_log(
+    target_win_tzinfo: Target,
+    fs_win: VirtualFilesystem,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    map_dir = "inetpub/logs/LogFiles/W3SVC1"
+
+    data_dir = absolute_path("_data/plugins/apps/webserver/iis/iis-logs-w3c/W3SVC1")
+    fs_win.map_dir(map_dir, data_dir)
+    fs_win.map_file_fh(f"{map_dir}/01_logfile.log", BytesIO(b"a,b,c,d,e,f,g,h,i,j,k,l,m,n,o,p"))
+    target_win_tzinfo.add_plugin(iis.IISLogsPlugin)
+    with caplog.at_level(logging.ERROR, target_win_tzinfo.log.name):
+        assert list(target_win_tzinfo.iis.access())
+
+    assert any(
+        f"Issue processing log file sysvol/{map_dir}/01_logfile.log in auto format" in message
+        for message in caplog.messages
+    )
+
+
 def test_iis_direct_mode() -> None:
     data_path = absolute_path("_data/plugins/apps/webserver/iis/iis-logs-iis/W3SVC1/u_in211001.log")