chore: Remove psutil dependency

The psutil dependency is used at exactly one location in the codebase.
The psutil library claims to be portable, however the broadcast feature
is [not supported](https://github.com/giampaolo/psutil/blob/8d943015ffce86de31d9494ffa3a1eae7dd91719/psutil/arch/windows/net.c#L210) on Windows.

Since we support Linux as Tier-1 and Windows as Tier-2, let's remove the
dependency and use the always available system tooling for this.

pyproject.toml

@@ -43,7 +43,6 @@ construct = "^2.10"
 msgspec = ">=0.11,<0.19"
 pydantic = "^2.0"
 platformdirs = ">=2.6,<5.0"
-psutil = ">=5.9.4,<7.0.0"
 more-itertools = "^10.3.0"
 pydantic-argparse = { path = "vendor/pydantic-argparse", develop = true }
 
@@ -56,7 +55,6 @@ pytest = ">=7.1,<9.0"
 pytest-asyncio = ">=0.20,<0.25"
 python-lsp-server = "^1.5"
 types-aiofiles = ">=23.1,<25.0"
-types-psutil = ">=5.9.5.10,<7.0.0.0"
 types-tabulate = "^0.9"
 myst-parser = ">=3.0.1,<4.1"
 sphinx-rtd-theme = ">=1,<3"

src/gallia/commands/discover/doip.py

@@ -9,7 +9,6 @@
 from urllib.parse import parse_qs, urlparse
 
 import aiofiles
-import psutil
 
 from gallia.command import AsyncScript
 from gallia.command.base import AsyncScriptConfig
@@ -31,6 +30,7 @@
     TimingAndCommunicationParameters,
     VehicleAnnouncementMessage,
 )
+from gallia.utils import AddrInfo, net_if_addrs
 
 logger = get_logger(__name__)
 
@@ -490,33 +490,40 @@ async def read_diag_request_custom(self, conn: DoIPConnection) -> tuple[int | No
                 continue
             return (None, payload.UserData)
 
-    async def run_udp_discovery(self) -> list[tuple[str, int]]:
-        all_ips = []
-        found = []
+    @staticmethod
+    def get_broadcast_addrs() -> list[AddrInfo]:
+        out = []
+        for iface in net_if_addrs():
+            if iface.is_up() or not iface.can_broadcast():
+                continue
 
-        for iface in psutil.net_if_addrs().values():
-            for ip in iface:
-                # we only work with broadcastable IPv4
-                if ip.family != socket.AF_INET or ip.broadcast is None:
+            for addr in iface.addr_info:
+                # We only work with broadcastable IPv4.
+                if not addr.is_v4() or addr.broadcast is None:
                     continue
-                all_ips.append(ip)
+                out.append(addr)
+        return out
+
+    async def run_udp_discovery(self) -> list[tuple[str, int]]:
+        addrs = self.get_broadcast_addrs()
+        found = []
 
-        for ip in all_ips:
-            logger.info(f"[💌] Sending DoIP VehicleIdentificationRequest to {ip.broadcast}")
+        for addr in addrs:
+            logger.info(f"[💌] Sending DoIP VehicleIdentificationRequest to {addr.broadcast}")
             sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
             sock.setblocking(False)
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
-            sock.bind((ip.address, 0))
+            sock.bind((addr.local, 0))
             loop = asyncio.get_running_loop()
 
             hdr = GenericHeader(0xFF, PayloadTypes.VehicleIdentificationRequestMessage, 0x00)
-            await loop.sock_sendto(sock, hdr.pack(), (ip.broadcast, 13400))
+            await loop.sock_sendto(sock, hdr.pack(), (addr.broadcast, 13400))
             try:
                 while True:
-                    data, addr = await asyncio.wait_for(loop.sock_recvfrom(sock, 1024), 2)
+                    data, from_addr = await asyncio.wait_for(loop.sock_recvfrom(sock, 1024), 2)
                     info = VehicleAnnouncementMessage.unpack(data[8:])
                     logger.notice(f"[💝]: {addr} responded: {info}")
-                    found.append(addr)
+                    found.append(from_addr)
             except TimeoutError:
                 logger.info("[💔] Reached timeout...")
                 continue

src/gallia/utils.py

@@ -8,8 +8,10 @@
 import contextvars
 import importlib.util
 import ipaddress
+import json
 import logging
 import re
+import subprocess
 import sys
 from collections.abc import Awaitable, Callable
 from pathlib import Path
@@ -18,6 +20,8 @@
 from urllib.parse import urlparse
 
 import aiofiles
+import pydantic
+from pydantic.networks import IPvAnyAddress
 
 from gallia.log import Loglevel, get_logger
 
@@ -26,6 +30,9 @@
     from gallia.transports import TargetURI
 
 
+logger = get_logger(__name__)
+
+
 def auto_int(arg: str) -> int:
     return int(arg, 0)
 
@@ -302,3 +309,53 @@ def handle_task_error(fut: asyncio.Future[Any]) -> None:
     except BaseException as e:
         # Info level is enough, since our aim is only to consume the stack trace
         logger.info(f"{task_name if task_name is not None else 'Task'} ended with error: {e!r}")
+
+
+class AddrInfo(pydantic.BaseModel):
+    family: str
+    local: IPvAnyAddress
+    prefixlen: int
+    broadcast: IPvAnyAddress | None = None
+    scope: str
+    label: str | None = None
+    valid_life_time: int
+    preferred_life_time: int
+
+    def is_v4(self) -> bool:
+        return self.family == "inet"
+
+
+class Interface(pydantic.BaseModel):
+    ifindex: int
+    ifname: str
+    flags: list[str]
+    mtu: int
+    qdisc: str
+    operstate: str
+    group: str
+    link_type: str
+    address: str | None = None
+    broadcast: str | None = None
+    addr_info: list[AddrInfo]
+
+    def is_up(self) -> bool:
+        return self.operstate == "UP"
+
+    def can_broadcast(self) -> bool:
+        return "BROADCAST" in self.flags
+
+
+def net_if_addrs() -> list[Interface]:
+    if sys.platform != "linux":
+        raise NotImplementedError("net_if_addrs() is only supported on Linux platforms")
+
+    p = subprocess.run(["ip", "-j", "address", "show"], capture_output=True, check=True)
+
+    try:
+        return [Interface(**item) for item in json.loads(p.stdout.decode())]
+    except pydantic.ValidationError as e:
+        logger.error("BUG: A special case for `ip -j address show` is not handled!")
+        logger.error("Please report a bug including the following json string.")
+        logger.error("https://github.com/Fraunhofer-AISEC/gallia/issues")
+        logger.error(e.json())
+        raise

tests/pytest/test_net_if.py

@@ -0,0 +1,9 @@
+# SPDX-FileCopyrightText: AISEC Pentesting Team
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from gallia.utils import net_if_addrs
+
+
+def test_net_if_addrs() -> None:
+    net_if_addrs()