Add MacNetworkPlugin (DIS-3024)

dissect/target/helpers/record.py

@@ -175,7 +175,7 @@ def DynamicDescriptor(types):  # noqa
     [
         *COMMON_INTERFACE_ELEMENTS,
         ("varint", "vlan"),
-        ("string", "proxy"),
+        ("net.ipaddress[]", "proxy"),
         ("varint", "interface_service_order"),
     ],
 )

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

@@ -40,24 +40,8 @@ def hostname(self) -> Optional[str]:
     @export(property=True)
     def ips(self) -> Optional[list[str]]:
         ips = set()
-
-        # Static configured IP-addresses
-        if (preferences := self.target.fs.path(self.SYSTEM)).exists():
-            network = plistlib.load(preferences.open()).get("NetworkServices")
-
-            for interface in network.values():
-                for addresses in [interface.get("IPv4"), interface.get("IPv6")]:
-                    ips.update(addresses.get("Addresses", []))
-
-        # IP-addresses configured by DHCP
-        if (dhcp := self.target.fs.path("/private/var/db/dhcpclient/leases")).exists():
-            for lease in dhcp.iterdir():
-                if lease.is_file():
-                    lease = plistlib.load(lease.open())
-
-                    if ip := lease.get("IPAddress"):
-                        ips.add(ip)
-
+        for ip in self.target.network.ips():
+            ips.add(str(ip))
         return list(ips)
 
     @export(property=True)

dissect/target/plugins/os/unix/bsd/osx/network.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import plistlib
+from typing import Iterator
+
+from dissect.target.helpers.record import MacInterfaceRecord
+from dissect.target.plugin import internal
+from dissect.target.plugins.general.network import NetworkPlugin
+
+
+class MacNetworkPlugin(NetworkPlugin):
+    SYSTEM = "/Library/Preferences/SystemConfiguration/preferences.plist"
+    DHCP = "/private/var/db/dhcpclient/leases"
+    plistlease = {}
+    plistnetwork = {}
+
+    def _plistlease(self) -> None:
+        if (dhcp := self.target.fs.path(self.DHCP)).exists():
+            for lease in dhcp.iterdir():
+                if lease.is_file():
+                    self.plistlease = plistlib.load(lease.open())
+
+    def _plistnetwork(self) -> None:
+        if (preferences := self.target.fs.path(self.SYSTEM)).exists():
+            self.plistnetwork = plistlib.load(preferences.open())
+
+    @internal
+    def _interfaces(self) -> Iterator[MacInterfaceRecord]:
+        if not self.plistlease:
+            self._plistlease()
+
+        if not self.plistnetwork:
+            self._plistnetwork()
+
+        dhcp_ip = self.plistlease.get("IPAddress")
+
+        current_set = self.plistnetwork.get("CurrentSet")
+        sets = self.plistnetwork.get("Sets", {})
+        for name, _set in sets.items():
+            if f"/Sets/{name}" == current_set:
+                item = _set
+                for key in ["Network", "Global", "IPv4", "ServiceOrder"]:
+                    item = item.get(key, {})
+                service_order = item
+                break
+
+        network = self.plistnetwork.get("NetworkServices", {})
+        vlans = self.plistnetwork.get("VirtualNetworkInterfaces", {}).get("VLAN", {})
+        vlan_lookup = {}
+        for key, vlan in vlans.items():
+            vlan_lookup[key] = vlan.get("Tag")
+
+        for _id, interface in network.items():
+            dns = set()
+            gateways = set()
+            proxies = set()
+            ips = set()
+            record = MacInterfaceRecord(_target=self.target)
+            record.source = "NetworkServices"
+            device = interface.get("Interface", {})
+            record.name = device.get("DeviceName")
+            record.type = device.get("Type")
+            record.vlan = vlan_lookup.get(record.name)
+            record.interface_service_order = service_order.index(_id) if _id in service_order else None
+            try:
+                record.enabled = not interface.get("__INACTIVE__", False)
+                for setting, value in interface.get("Proxies", {}).items():
+                    if setting.endswith("Proxy"):
+                        proxies.add(value)
+                record.proxy = proxies
+                for addr in interface.get("DNS", {}).get("ServerAddresses", {}):
+                    dns.add(addr)
+                for addresses in [interface.get("IPv4", {}), interface.get("IPv6", {})]:
+                    if router := addresses.get("Router"):
+                        gateways.add(router)
+                    for addr in addresses.get("Addresses", []):
+                        ips.add(addr)
+                    if dhcp_ip and addresses.get("ConfigMethod", "") == "DHCP":
+                        ips.add(dhcp_ip)
+                record.ip = list(ips)
+                record.dns = list(dns)
+                record.gateway = list(gateways)
+            except Exception as message:
+                self.target.log.warning("Error reading configuration for network device %s: %s.", record.name, message)
+                continue
+
+            yield record

tests/plugins/os/unix/bsd/osx/test_network.py

@@ -0,0 +1,160 @@
+import copy
+
+import pytest
+
+from dissect.target.plugins.os.unix.bsd.osx.network import MacNetworkPlugin
+from dissect.target.target import Target
+
+fake_plist = {
+    "CurrentSet": "/Sets/1",
+    "NetworkServices": {
+        "1": {
+            "DNS": {"ServerAddresses": ["8.8.8.8"]},
+            "IPv4": {
+                "Addresses": ["192.122.13.34"],
+                "Router": "8.8.8.8",
+            },
+            "Interface": {
+                "DeviceName": "en0",
+                "Type": "Ethernet",
+            },
+            "Proxies": {
+                "GopherProxy": "9.9.9.9",
+            },
+        },
+    },
+    "Sets": {
+        "1": {
+            "Network": {
+                "Global": {"IPv4": {"ServiceOrder": ["1"]}},
+            },
+        },
+    },
+    "VirtualNetworkInterfaces": {"VLAN": {"vlan0": {"Tag": 2}}},
+}
+
+
+def vlan0(fake_plist: dict) -> dict:
+    fake_plist = copy.deepcopy(fake_plist)
+    fake_plist["NetworkServices"]["1"]["Interface"].update({"DeviceName": "vlan0"})
+    return fake_plist
+
+
+def inactive(fake_plist: dict) -> dict:
+    fake_plist = copy.deepcopy(fake_plist)
+    fake_plist["NetworkServices"]["1"].update({"__INACTIVE__": True})
+    return fake_plist
+
+
+def ipv6(fake_plist: dict) -> dict:
+    fake_plist = copy.deepcopy(fake_plist)
+    del fake_plist["NetworkServices"]["1"]["IPv4"]
+    fake_plist["NetworkServices"]["1"]["IPv6"] = {"Addresses": ["::1"]}
+    return fake_plist
+
+
+def reorder(fake_plist: dict) -> dict:
+    fake_plist = copy.deepcopy(fake_plist)
+    fake_plist["Sets"]["1"]["Network"]["Global"]["IPv4"]["ServiceOrder"] = ["2", "1"]
+    return fake_plist
+
+
+def double(fake_plist: dict) -> dict:
+    fake_plist = copy.deepcopy(fake_plist)
+    fake_plist["NetworkServices"]["2"] = fake_plist["NetworkServices"]["1"]
+    return fake_plist
+
+
+def dhcp(fake_plist: dict) -> dict:
+    fake_plist = copy.deepcopy(fake_plist)
+    fake_plist["NetworkServices"]["1"]["IPv4"].update({"ConfigMethod": "DHCP"})
+    return fake_plist
+
+
+@pytest.mark.parametrize(
+    "lease,netinfo,expected,count",
+    [
+        ({"IPAddress": None}, {"CurrentSet": {}}, [], 0),
+        (
+            {},
+            fake_plist,
+            [
+                (0, "hostname", ["dummys Mac"]),
+                (0, "domain", ["None"]),
+                (0, "name", ["en0"]),
+                (0, "type", ["Ethernet"]),
+                (0, "ip", ["192.122.13.34"]),
+                (0, "proxy", ["9.9.9.9"]),
+                (0, "gateway", ["8.8.8.8"]),
+                (0, "dns", ["8.8.8.8"]),
+                (0, "vlan", ["None"]),
+                (0, "enabled", ["True"]),
+                (0, "interface_service_order", ["0"]),
+                (0, "mac", ["None"]),
+                (0, "vlan", ["None"]),
+            ],
+            1,
+        ),
+        (
+            {"IPAddress": "10.0.0.2"},
+            dhcp(fake_plist),
+            [
+                (0, "ip", sorted(["10.0.0.2", "192.122.13.34"])),
+            ],
+            1,
+        ),
+        (
+            {},
+            vlan0(fake_plist),
+            [
+                (0, "vlan", ["2"]),
+            ],
+            1,
+        ),
+        (
+            {},
+            inactive(fake_plist),
+            [
+                (0, "enabled", ["False"]),
+            ],
+            1,
+        ),
+        (
+            {},
+            ipv6(fake_plist),
+            [
+                (0, "ip", ["::1"]),
+            ],
+            1,
+        ),
+        (
+            {},
+            reorder(fake_plist),
+            [
+                (0, "interface_service_order", ["1"]),
+            ],
+            1,
+        ),
+        (
+            {},
+            double(fake_plist),
+            [
+                (0, "enabled", ["True"]),
+                (1, "enabled", ["True"]),
+            ],
+            2,
+        ),
+    ],
+)
+def test_macos_network(target_osx: Target, lease: dict, netinfo: dict, expected: dict, count: int) -> None:
+    network = MacNetworkPlugin(target_osx)
+    network.plistlease = lease
+    network.plistnetwork = netinfo
+    interfaces = list(network.interfaces())
+    assert len(interfaces) == count
+    for index, key, value in expected:
+        attr = getattr(interfaces[index], key)
+        if not isinstance(attr, list):
+            attr = [attr]
+        attr = list(sorted(map(str, attr)))
+        assert attr == value