WIP windows support

.github/workflows/linters-win32.yml

@@ -0,0 +1,46 @@
+# SPDX-FileCopyrightText: AISEC Pentesting Team
+#
+# SPDX-License-Identifier: CC0-1.0
+
+# https://github.com/actions/setup-python/issues/374
+
+name: linters
+
+on:
+  push:
+    branches:
+    - master
+  pull_request:
+    paths:
+    - 'pyproject.toml'
+    - 'poetry.lock'
+    - '.github/**'
+    - 'src/**'
+    - 'tests/**'
+
+jobs:
+  linters:
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.11', '3.12']
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+    - uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+    - uses: Gr1N/setup-poetry@v9
+    - uses: actions/cache@v4
+      with:
+        path: ~/.cache/pypoetry/virtualenvs
+        key: ${{ runner.os }}-poetry-${{ hashFiles('poetry.lock') }}-${{ hashFiles('pyproject.toml') }}
+
+    - name: Install Dependencies
+      run: |
+        poetry install
+    - name: Run mypy
+      run: |
+        poetry run mypy-win32 src tests

Makefile

@@ -28,6 +28,11 @@ lint:
 	ruff format --check src tests
 	reuse lint
 
+.PHONY: lint-win32
+lint-win32:
+	mypy --platform win32 --exclude "gallia\/log\.py" --exclude "hr" src tests
+	ruff check src tests
+
 .PHONY: fmt
 fmt:
 	ruff check --fix-only src tests

pyproject.toml

@@ -104,6 +104,7 @@ select = [
     "UP",   # pyupgrade
 ]
 ignore = [
+    "E402",     # Module level import not at top of file
     "E501",     # line length
     "PLR2004",  # magic value used in comparison
     "PLR0911",  # too many return statements

src/gallia/command/base.py

@@ -17,17 +17,14 @@
 from pathlib import Path
 from subprocess import CalledProcessError, run
 from tempfile import gettempdir
-from typing import cast
-
-if sys.platform != "win32":
-    import fcntl
-    from gallia.dumpcap import Dumpcap
+from typing import Protocol, cast
 
 import exitcode
 import msgspec
 
 from gallia.config import Config
 from gallia.db.handler import DBHandler
+from gallia.dumpcap import Dumpcap
 from gallia.log import add_zst_log_handler, get_logger, tz
 from gallia.plugins import load_transport
 from gallia.powersupply import PowerSupply, PowerSupplyURI
@@ -74,7 +71,54 @@ def json(self) -> str:
 logger = get_logger("gallia.base")
 
 
-class BaseCommand(ABC):
+if sys.platform.startswith("linux") or sys.platform == "darwin":
+    import fcntl
+
+    class Flockable(Protocol):
+        @property
+        def _lock_file_fd(self) -> int | None: ...
+
+    class FlockMixin:
+        def _open_lockfile(self, path: Path) -> int | None:
+            if not path.exists():
+                path.touch()
+
+            logger.notice("opening lockfile…")
+            return os.open(path, os.O_RDONLY)
+
+        def _aquire_flock(self: Flockable) -> None:
+            assert self._lock_file_fd is not None
+
+            try:
+                # First do a non blocking flock. If waiting is required,
+                # log a message and do a blocking wait afterwards.
+                fcntl.flock(self._lock_file_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                logger.notice("waiting for flock…")
+                fcntl.flock(self._lock_file_fd, fcntl.LOCK_EX)
+            logger.info("Acquired lock. Continuing…")
+
+        def _release_flock(self: Flockable) -> None:
+            assert self._lock_file_fd is not None
+            fcntl.flock(self._lock_file_fd, fcntl.LOCK_UN)
+            os.close(self._lock_file_fd)
+
+
+if sys.platform == "win32":
+
+    class FlockMixin:
+        def _open_lockfile(self, path: Path) -> int | None:
+            logger.warn("lockfile in windows is not supported")
+            return None
+
+        def _aquire_flock(self) -> None:
+            pass
+
+        def _release_flock(self) -> None:
+            pass
+
+
+class BaseCommand(FlockMixin, ABC):
     """BaseCommand is the baseclass for all gallia commands.
     This class can be used in standalone scripts via the
     gallia command line interface facility.
@@ -306,7 +350,10 @@ def _add_latest_link(self, path: Path) -> None:
 
         symlink = path.joinpath("LATEST")
         symlink.unlink(missing_ok=True)
-        symlink.symlink_to(latest_dir)
+        try:
+            symlink.symlink_to(latest_dir)
+        except NotImplementedError as e:
+            logger.warn(f"symlink error: {e}")
 
     def prepare_artifactsdir(
         self,
@@ -346,40 +393,20 @@ def prepare_artifactsdir(
             artifacts_dir.mkdir(parents=True)
 
             self._dump_environment(artifacts_dir.joinpath(FileNames.ENV.value))
-
-            if sys.platform != "win32":
-                self._add_latest_link(command_dir)
+            self._add_latest_link(command_dir)
 
             return artifacts_dir.absolute()
 
         raise ValueError("base_dir or force_path must be different from None")
 
-    def _aquire_flock(self, path: Path) -> None:
-        if not path.exists():
-            path.touch()
-        self._lock_file_fd = os.open(path, os.O_RDONLY)
-        try:
-            # First do a non blocking flock. If waiting is required,
-            # log a message and do a blocking wait afterwards.
-            fcntl.flock(self._lock_file_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except BlockingIOError:
-            logger.notice(f"Waiting for flock: {path}")
-            fcntl.flock(self._lock_file_fd, fcntl.LOCK_EX)
-        logger.info("Acquired lock. Continuing.")
-
-    def _release_flock(self) -> None:
-        assert self._lock_file_fd
-        fcntl.flock(self._lock_file_fd, fcntl.LOCK_UN)
-        os.close(self._lock_file_fd)
-
     def entry_point(self, args: Namespace) -> int:
-        if sys.platform != "win32":
-            if (p := args.lock_file) is not None:
-                try:
-                    self._aquire_flock(p)
-                except OSError as e:
-                    logger.critical(f"Unable to lock {p}: {e}")
-                    return exitcode.OSFILE
+        if (p := args.lock_file) is not None:
+            try:
+                self._lock_file_fd = self._open_lockfile(p)
+                self._aquire_flock()
+            except OSError as e:
+                logger.critical(f"Unable to lock {p}: {e}")
+                return exitcode.OSFILE
 
         if self.HAS_ARTIFACTS_DIR:
             self.artifacts_dir = self.prepare_artifactsdir(
@@ -438,9 +465,8 @@ def entry_point(self, args: Namespace) -> int:
         if args.hooks:
             self.run_hook(HookVariant.POST, args, exit_code)
 
-        if sys.platform != "win32":
-            if self._lock_file_fd is not None:
-                self._release_flock()
+        if self._lock_file_fd is not None:
+            self._release_flock()
 
         return exit_code
 
@@ -548,7 +574,7 @@ async def setup(self, args: Namespace) -> None:
                 self.parser.error("--dumpcap specified but `dumpcap` is not available")
             self.dumpcap = await Dumpcap.start(args.target, self.artifacts_dir)
             if self.dumpcap is None:
-                logger.error("Dumpcap could not be started!")
+                logger.error("`dumpcap` could not be started!")
             else:
                 await self.dumpcap.sync()
 
@@ -567,7 +593,10 @@ def configure_class_parser(self) -> None:
         group.add_argument(
             "--dumpcap",
             action=argparse.BooleanOptionalAction,
-            default=self.config.get_value("gallia.scanner.dumpcap", default=sys.platform == "linux"),
+            default=self.config.get_value(
+                "gallia.scanner.dumpcap",
+                default=sys.platform == "linux",
+            ),
             help="Enable/Disable creating a pcap file",
         )
 

src/gallia/commands/__init__.py

@@ -2,11 +2,10 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import sys
+
 from gallia.command.base import BaseCommand
 from gallia.commands.discover.doip import DoIPDiscoverer
-# from gallia.commands.discover.find_xcp import FindXCP
-# from gallia.commands.discover.uds.isotp import IsotpDiscoverer
-# from gallia.commands.fuzz.uds.pdu import PDUFuzzer
 from gallia.commands.primitive.generic.pdu import GenericPDUPrimitive
 from gallia.commands.primitive.uds.dtc import DTCPrimitive
 from gallia.commands.primitive.uds.ecu_reset import ECUResetPrimitive
@@ -19,21 +18,15 @@
 from gallia.commands.primitive.uds.vin import VINPrimitive
 from gallia.commands.primitive.uds.wdbi import WriteByIdentifierPrimitive
 from gallia.commands.primitive.uds.wmba import WMBAPrimitive
-# from gallia.commands.primitive.uds.xcp import SimpleTestXCP
 from gallia.commands.scan.uds.identifiers import ScanIdentifiers
 from gallia.commands.scan.uds.memory import MemoryFunctionsScanner
 from gallia.commands.scan.uds.reset import ResetScanner
 from gallia.commands.scan.uds.sa_dump_seeds import SASeedsDumper
 from gallia.commands.scan.uds.services import ServicesScanner
 from gallia.commands.scan.uds.sessions import SessionsScanner
-# from gallia.commands.script.vecu import VirtualECU
 
 registry: list[type[BaseCommand]] = [
-    # SimpleTestXCP,
     DoIPDiscoverer,
-    # IsotpDiscoverer,
-    # FindXCP,
-    # PDUFuzzer,
     MemoryFunctionsScanner,
     ReadByIdentifierPrimitive,
     ResetScanner,
@@ -51,9 +44,51 @@
     GenericPDUPrimitive,
     SendPDUPrimitive,
     WMBAPrimitive,
-    # VirtualECU,
     WriteByIdentifierPrimitive,
-    # SimpleTestXCP,
 ]
 
-__all__ = [x.__name__ for x in registry]
+__all__ = [
+    "DoIPDiscoverer",
+    "DMemoryFunctionsScanner",
+    "DReadByIdentifierPrimitive",
+    "DResetScanner",
+    "DSASeedsDumper",
+    "DScanIdentifiers",
+    "DSessionsScanner",
+    "DServicesScanner",
+    "DDTCPrimitive",
+    "DECUResetPrimitive",
+    "DVINPrimitive",
+    "DIOCBIPrimitive",
+    "DPingPrimitive",
+    "DRMBAPrimitive",
+    "DRTCLPrimitive",
+    "DGenericPDUPrimitive",
+    "DSendPDUPrimitive",
+    "DWMBAPrimitive",
+    "DWriteByIdentifierPrimitive",
+]
+
+
+if sys.platform.startswith("linux"):
+    from gallia.commands.discover.find_xcp import FindXCP
+    from gallia.commands.discover.uds.isotp import IsotpDiscoverer
+    from gallia.commands.fuzz.uds.pdu import PDUFuzzer
+    from gallia.commands.primitive.uds.xcp import SimpleTestXCP
+    from gallia.commands.script.vecu import VirtualECU
+
+    registry += [
+        FindXCP,
+        IsotpDiscoverer,
+        PDUFuzzer,
+        SimpleTestXCP,
+        VirtualECU,
+    ]
+
+    __all__ += [
+        "FindXCP",
+        "IsotpDiscoverer",
+        "PDUFuzzer",
+        "SimpleTestXCP",
+        "VirtualECU",
+    ]

src/gallia/commands/discover/find_xcp.py

@@ -4,8 +4,11 @@
 
 import socket
 import struct
+import sys
 from argparse import ArgumentParser, Namespace
 
+assert sys.platform.startswith("linux"), "unsupported platform"
+
 from gallia.command import AsyncScript
 from gallia.config import Config
 from gallia.log import get_logger

src/gallia/commands/discover/uds/isotp.py

@@ -3,9 +3,12 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import asyncio
+import sys
 from argparse import Namespace
 from binascii import unhexlify
 
+assert sys.platform.startswith("linux"), "unsupported platform"
+
 from gallia.command import UDSDiscoveryScanner
 from gallia.log import get_logger
 from gallia.services.uds import NegativeResponse, UDSClient, UDSRequest

src/gallia/commands/fuzz/uds/pdu.py

@@ -5,8 +5,11 @@
 import asyncio
 import binascii
 import random
+import sys
 from argparse import Namespace
 
+assert sys.platform.startswith("linux"), "unsupported platform"
+
 from gallia.command import UDSScanner
 from gallia.log import get_logger
 from gallia.services.uds.core.client import UDSRequestConfig

src/gallia/commands/primitive/uds/xcp.py

@@ -2,8 +2,11 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import sys
 from argparse import ArgumentParser, Namespace
 
+assert sys.platform.startswith("linux"), "unsupported platform"
+
 from gallia.command import Scanner
 from gallia.config import Config
 from gallia.plugins import load_transport