WIP: Add flexray support via vector xllib

.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-win32
+
+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 make lint-win32

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

fr_dump.py

@@ -0,0 +1,33 @@
+import asyncio
+import sys
+
+assert sys.platform == "win32"
+
+from gallia.transports import RawFlexrayTransport, TargetURI
+
+
+async def main() -> None:
+    url = TargetURI("flexray-raw:")
+    tp = await RawFlexrayTransport.connect(url, None)
+    tp.add_block_all_filter()
+    tp.set_acceptance_filter(33, 33)
+    tp.set_acceptance_filter(59, 59)
+    tp.activate_channel()
+
+    while True:
+        frame = await tp.read_frame()
+        # Filter out null frames. :)
+        if frame.data[0] == 0x00:
+            continue
+
+        # print(f"raw event: {frame.raw}")
+        print(f"   -> slot_id: {frame.slot_id}; data: {frame.data.hex()}")
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except TimeoutError:
+        pass
+    except KeyboardInterrupt:
+        pass

fr_send.py

@@ -0,0 +1,29 @@
+import asyncio
+import sys
+
+assert sys.platform == "win32"
+
+from gallia.log import Loglevel, setup_logging
+from gallia.transports import FlexRayTPLegacyTransport, TargetURI
+
+
+async def main() -> None:
+    setup_logging(level=Loglevel.TRACE, no_volatile_info=True)
+
+    # TODO: rename dst->target
+    url = TargetURI(
+        "flexray-tp-legacy://?dst_slot_id=59&src_slot_id=33&dst_address=0x307c&src_address=0x6100&payload_rx_start_index=0&payload_rx_end_index=12"
+    )
+    tp = await FlexRayTPLegacyTransport.connect(url, None)
+
+    await tp.write(bytes.fromhex("3e 00"))
+
+    data = await tp.read()
+    print(data.hex())
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass

pyproject.toml

@@ -48,6 +48,7 @@ platformdirs = ">=2.6,<5.0"
 exitcode = "^0.1.0"
 psutil = ">=5.9.4,<7.0.0"
 httpx = {extras = ["http2"], version = ">=0.26,<0.28"}
+more-itertools = "^10.3.0"
 
 [tool.poetry.group.dev.dependencies]
 Sphinx = ">=5.2,<8.0"
@@ -104,6 +105,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

@@ -4,7 +4,6 @@
 
 import argparse
 import asyncio
-import fcntl
 import os
 import os.path
 import shutil
@@ -18,7 +17,7 @@
 from pathlib import Path
 from subprocess import CalledProcessError, run
 from tempfile import gettempdir
-from typing import cast
+from typing import Protocol, cast
 
 import exitcode
 import msgspec
@@ -72,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.
@@ -304,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 (OSError, NotADirectoryError) as e:
+            logger.warn(f"symlink error: {e}")
 
     def prepare_artifactsdir(
         self,
@@ -350,28 +399,11 @@ def prepare_artifactsdir(
 
         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 (p := args.lock_file) is not None:
             try:
-                self._aquire_flock(p)
+                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
@@ -542,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()
 
@@ -561,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=True),
+            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",
+    ]