Improve RedHat OS detection (#1079)

Co-authored-by: Erik Schamper <[email protected]>

Author: JSCU-CNI

dissect/target/plugins/os/unix/_os.py

@@ -4,7 +4,7 @@
 import re
 import uuid
 from pathlib import Path
-from typing import Iterator
+from typing import Callable, Iterator
 
 from flow.record.fieldtypes import posix_path
 
@@ -49,7 +49,7 @@ def __init__(self, target: Target):
         super().__init__(target)
         self._add_mounts()
         self._add_devices()
-        self._hostname_dict = self._parse_hostname_string()
+        self._hostname, self._domain = self._parse_hostname_string()
         self._hosts_dict = self._parse_hosts_string()
         self._os_release = self._parse_os_release()
 
@@ -156,78 +156,83 @@ def architecture(self) -> str | None:
 
     @export(property=True)
     def hostname(self) -> str | None:
-        hosts_string = self._hosts_dict.get("hostname", "localhost")
-        return self._hostname_dict.get("hostname", hosts_string)
+        return self._hostname or self._hosts_dict.get("hostname", "localhost")
 
     @export(property=True)
     def domain(self) -> str | None:
-        domain = self._hostname_dict.get("domain", "localhost")
-        if domain == "localhost":
-            domain = self._hosts_dict["hostname", "localhost"]
-            if domain == self.hostname:
-                return domain  # domain likely not defined, so localhost is the domain.
-        return domain
+        if self._domain is None or self._domain == "localhost":
+            # fall back to /etc/hosts file
+            return self._hosts_dict.get("hostname")
+        return self._domain
 
     @export(property=True)
     def os(self) -> str:
         return OperatingSystem.UNIX.value
 
-    def _parse_rh_legacy(self, path: Path) -> str | None:
-        hostname = None
-        file_contents = path.open("rt").readlines()
-        for line in file_contents:
-            if not line.startswith("HOSTNAME"):
-                continue
-            _, _, hostname = line.rstrip().partition("=")
-        return hostname
-
-    def _parse_hostname_string(self, paths: list[str] | None = None) -> dict[str, str] | None:
-        """Returns a dict containing the hostname and domain name portion of the path(s) specified.
+    def _parse_hostname_string(
+        self, paths: list[tuple[str, Callable[[Path], str] | None]] | None = None
+    ) -> tuple[str | None, str | None]:
+        """Returns a tuple containing respectively the hostname and domain name portion of the path(s) specified.
 
         Args:
-            paths (list): list of paths
-        """
-        redhat_legacy_path = "/etc/sysconfig/network"
-        paths = paths or ["/etc/hostname", "/etc/HOSTNAME", "/proc/sys/kernel/hostname", redhat_legacy_path]
-        hostname_dict = {"hostname": None, "domain": None}
+            paths (list): list of tuples with paths and callables to parse the path or None
 
-        for path in paths:
-            path = self.target.fs.path(path)
-
-            if not path.exists():
+        Returns:
+            Tuple with hostname and domain strings.
+        """
+        hostname = None
+        domain = None
+
+        paths = paths or [
+            ("/etc/hostname", None),
+            ("/etc/HOSTNAME", None),
+            ("/proc/sys/kernel/hostname", None),
+            ("/etc/sysconfig/network", self._parse_rh_legacy),
+            ("/etc/hosts", self._parse_etc_hosts),  # fallback if no other hostnames are found
+        ]
+
+        for path, callable in paths:
+            if not (path := self.target.fs.path(path)).exists():
                 continue
 
-            if path.as_posix() == redhat_legacy_path:
-                hostname_string = self._parse_rh_legacy(path)
+            if callable:
+                hostname = callable(path)
             else:
-                hostname_string = path.open("rt").read().rstrip()
+                hostname = path.open("rt").read().rstrip()
+
+            if hostname and "." in hostname:
+                hostname, domain = hostname.split(".", maxsplit=1)
 
-            if hostname_string and "." in hostname_string:
-                hostname_string = hostname_string.split(".", maxsplit=1)
-                hostname_dict = {"hostname": hostname_string[0], "domain": hostname_string[1]}
-            elif hostname_string != "":
-                hostname_dict = {"hostname": hostname_string, "domain": None}
-            else:
-                hostname_dict = {"hostname": None, "domain": None}
             break  # break whenever a valid hostname is found
 
-        return hostname_dict
+        # Can be an empty string due to splitting of hostname and domain
+        return hostname or None, domain or None
+
+    def _parse_rh_legacy(self, path: Path) -> str | None:
+        hostname = None
+        file_contents = path.open("rt").readlines()
+        for line in file_contents:
+            if not line.startswith("HOSTNAME"):
+                continue
+            _, _, hostname = line.rstrip().partition("=")
+        return hostname
+
+    def _parse_etc_hosts(self, path: Path) -> str | None:
+        for line in path.open("rt"):
+            if line.startswith(("127.0.0.1 ", "::1 ")) and "localhost" not in line:
+                return line.split(" ")[1]
 
     def _parse_hosts_string(self, paths: list[str] | None = None) -> dict[str, str]:
         paths = paths or ["/etc/hosts"]
-        hosts_string = {"ip": None, "hostname": None}
+        hosts_string = {}
 
         for path in paths:
             for fs in self.target.filesystems:
                 if fs.exists(path):
                     for line in fs.path(path).open("rt").readlines():
-                        line = line.split()
-                        if not line:
+                        if not (line := line.split()):
                             continue
-
-                        if (line[0].startswith("127.0.") or line[0].startswith("::1")) and line[
-                            1
-                        ].lower() != "localhost":
+                        if line[0].startswith(("127.0.", "::1")):
                             hosts_string = {"ip": line[0], "hostname": line[1]}
         return hosts_string
 

dissect/target/plugins/os/unix/bsd/freebsd/_os.py

@@ -13,12 +13,16 @@ def __init__(self, target: Target):
 
     @classmethod
     def detect(cls, target: Target) -> Filesystem | None:
+        FREEBSD_PATHS = {
+            "/.sujournal",
+            "/entropy",
+            "/bin/freebsd-version",
+        }
+
         for fs in target.filesystems:
-            if fs.exists("/net") and (fs.exists("/.sujournal") or fs.exists("/entropy")):
+            if fs.exists("/net") and any(fs.exists(path) for path in FREEBSD_PATHS):
                 return fs
 
-        return None
-
     @export(property=True)
     def version(self) -> str | None:
         return self._os_release.get("USERLAND_VERSION")

dissect/target/plugins/os/unix/bsd/openbsd/_os.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from typing import Optional
-
 from dissect.target.filesystem import Filesystem
 from dissect.target.plugin import export
 from dissect.target.plugins.os.unix.bsd._os import BsdPlugin
@@ -11,20 +9,28 @@
 class OpenBsdPlugin(BsdPlugin):
     def __init__(self, target: Target):
         super().__init__(target)
-        self._hostname_dict = self._parse_hostname_string(["/etc/myname"])
+        self._hostname, self._domain = self._parse_hostname_string([("/etc/myname", None)])
 
     @classmethod
-    def detect(cls, target: Target) -> Optional[Filesystem]:
+    def detect(cls, target: Target) -> Filesystem | None:
+        OPENBSD_PATHS = {
+            "/bsd",
+            "/bsd.rd",
+            "/bsd.mp",
+        }
+
         for fs in target.filesystems:
-            if fs.exists("/bsd") or fs.exists("/bsd.rd") or fs.exists("/bsd.mp") or fs.exists("/bsd.mp"):
+            if any(fs.exists(path) for path in OPENBSD_PATHS):
                 return fs
 
+    @export(property=True)
+    def version(self) -> str | None:
         return None
 
     @export(property=True)
-    def version(self) -> Optional[str]:
-        return None
+    def hostname(self) -> str | None:
+        return self._hostname
 
     @export(property=True)
-    def hostname(self) -> Optional[str]:
-        return self._hostname_dict.get("hostname", None)
+    def domain(self) -> str | None:
+        return self._domain

dissect/target/plugins/os/unix/linux/redhat/_os.py

@@ -1,19 +1,26 @@
-from typing import Optional
+from __future__ import annotations
 
 from dissect.target.filesystem import Filesystem
 from dissect.target.plugins.os.unix.linux._os import LinuxPlugin
 from dissect.target.target import Target
 
 
 class RedHatPlugin(LinuxPlugin):
+    """RedHat, CentOS and Fedora Plugin."""
+
     def __init__(self, target: Target):
         super().__init__(target)
 
     @classmethod
-    def detect(cls, target: Target) -> Optional[Filesystem]:
-        # also applicable to centos (which is a red hat derivative)
-        for fs in target.filesystems:
-            if fs.exists("/etc/sysconfig/network-scripts"):
-                return fs
+    def detect(cls, target: Target) -> Filesystem | None:
+        REDHAT_PATHS = {
+            "/etc/centos-release",
+            "/etc/fedora-release",
+            "/etc/redhat-release",
+            "/etc/sysconfig/network-scripts",  # legacy detection
+        }
 
-        return None
+        for fs in target.filesystems:
+            for path in REDHAT_PATHS:
+                if fs.exists(path):
+                    return fs

dissect/target/plugins/os/unix/locale.py

@@ -84,15 +84,18 @@ def language(self) -> list[str]:
             "/etc/sysconfig/i18n",
         ]
 
-        found_languages = []
+        found_languages = set()
 
         for locale_path in locale_paths:
             if (path := self.target.fs.path(locale_path)).exists():
                 for line in path.open("rt"):
                     if "LANG=" in line:
-                        found_languages.append(normalize_language(line.replace("LANG=", "").strip().strip('"')))
+                        lang_str = line.partition("=")[-1].strip().strip('"')
+                        if lang_str == "C.UTF-8":  # Skip if no locales are installed.
+                            continue
+                        found_languages.add(normalize_language(lang_str))
 
-        return found_languages
+        return list(found_languages)
 
     @export(record=UnixKeyboardRecord)
     def keyboard(self) -> Iterator[UnixKeyboardRecord]:

tests/plugins/os/unix/bsd/freebsd/__init__.py

@@ -0,0 +1,20 @@
+from dissect.target.filesystem import VirtualFilesystem
+from dissect.target.plugin import OperatingSystem
+from dissect.target.plugins.os.unix.bsd.freebsd._os import FreeBsdPlugin
+from dissect.target.target import Target
+from tests._utils import absolute_path
+
+
+def test_bsd_freebsd_os_detection(target_bare: Target) -> None:
+    """test if we detect FreeBSD correctly."""
+
+    fs = VirtualFilesystem()
+    fs.makedirs("/net")
+    fs.map_file("/bin/freebsd-version", absolute_path("_data/plugins/os/unix/bsd/freebsd/freebsd-freebsd-version"))
+
+    target_bare.filesystems.add(fs)
+    target_bare.apply()
+
+    assert FreeBsdPlugin.detect(target_bare)
+    assert isinstance(target_bare._os, FreeBsdPlugin)
+    assert target_bare.os == OperatingSystem.BSD

tests/plugins/os/unix/bsd/freebsd/test__os.py

@@ -0,0 +1,21 @@
+from io import BytesIO
+
+from dissect.target.filesystem import VirtualFilesystem
+from dissect.target.plugin import OperatingSystem
+from dissect.target.plugins.os.unix.bsd.openbsd._os import OpenBsdPlugin
+from dissect.target.target import Target
+
+
+def test_bsd_openbsd_os_detection(target_bare: Target) -> None:
+    """test if we detect OpenBSD correctly."""
+
+    fs = VirtualFilesystem()
+    fs.map_file_fh("/etc/myname", BytesIO(b"hostname"))
+    fs.makedirs("/bsd")
+
+    target_bare.filesystems.add(fs)
+    target_bare.apply()
+
+    assert OpenBsdPlugin.detect(target_bare)
+    assert isinstance(target_bare._os, OpenBsdPlugin)
+    assert target_bare.os == OperatingSystem.BSD

tests/plugins/os/unix/bsd/openbsd/__init__.py

@@ -0,0 +1,31 @@
+from io import BytesIO
+
+import pytest
+
+from dissect.target.filesystem import VirtualFilesystem
+from dissect.target.plugin import OperatingSystem
+from dissect.target.plugins.os.unix.linux.redhat._os import RedHatPlugin
+from dissect.target.target import Target
+
+
+@pytest.mark.parametrize(
+    "file_name",
+    [
+        ("/etc/redhat-release"),
+        ("/etc/centos-release"),
+        ("/etc/fedora-release"),
+        ("/etc/sysconfig/network-scripts"),
+    ],
+)
+def test_unix_linux_redhat_os_detection(target_bare: Target, file_name: str) -> None:
+    """test if we detect RedHat OS correctly."""
+
+    fs = VirtualFilesystem()
+    fs.map_file_fh(file_name, BytesIO(b""))
+
+    target_bare.filesystems.add(fs)
+    target_bare.apply()
+
+    assert RedHatPlugin.detect(target_bare)
+    assert isinstance(target_bare._os, RedHatPlugin)
+    assert target_bare.os == OperatingSystem.LINUX