Apply a few improvements and fixes

.github/workflows/main.yml

@@ -1,6 +1,9 @@
 name: Python package
 
-on: [push, pull_request]
+on:
+  push:
+    branches: [main]
+  pull_request:
 
 jobs:
   lint:
@@ -23,6 +26,8 @@ jobs:
   test:
     name: Test Python ${{ matrix.python-version }} ${{ matrix.os }}
     runs-on: ${{ matrix.os }}
+    needs: lint
+    continue-on-error: ${{ matrix.python-version == '3.14' }}
     strategy:
       fail-fast: false
       matrix:
@@ -37,9 +42,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
           cache: 'pip'
       - name: Install dependencies
-        run: |
-          python -m pip install --upgrade pip
-          pip install ruff
+        run: python -m pip install --upgrade pip
         shell: bash
       - name: Build and install
         run: |

src/snagfactory/board.py

@@ -4,6 +4,7 @@
 from snagrecover.utils import prettify_usb_addr, parse_usb_path
 import sys
 import os
+import contextlib
 import os.path
 import logging
 import logging.handlers
@@ -181,9 +182,6 @@ def get_recovery_config(board):
 
 
 def run_recovery(config, soc_family, log_queue):
-	sys.stdout = open(os.devnull, "w")
-	sys.stderr = open(os.devnull, "w")
-
 	import snagrecover.config
 
 	snagrecover.config.recovery_config = config
@@ -201,10 +199,12 @@ def run_recovery(config, soc_family, log_queue):
 
 	recovery = snagrecover.utils.get_recovery(soc_family)
 
-	try:
-		recovery()
-	except Exception as e:
-		logger.error(f"Caught exception from snagrecover: {e}")
-		sys.exit(-1)
+	with open(os.devnull, "w") as devnull:
+		with contextlib.redirect_stdout(devnull), contextlib.redirect_stderr(devnull):
+			try:
+				recovery()
+			except Exception as e:
+				logger.error(f"Caught exception from snagrecover: {e}")
+				sys.exit(-1)
 
 	logger.handlers.clear()

src/snagfactory/session.py

@@ -208,6 +208,8 @@ def load_log_section(self, marker: str, logs: list):
 		if marker == "summary:":
 			pattern = re.compile("summary: (\d+) done (\d+) failed (\d+) other")
 			match = pattern.match(logs[0])
+			if match is None:
+				raise ValueError(f"Could not parse summary line: {logs[0]!r}")
 			self.nb_done = int(match.groups()[0])
 			self.nb_failed = int(match.groups()[1])
 			self.nb_other = int(match.groups()[2])
@@ -220,6 +222,8 @@ def load_log_section(self, marker: str, logs: list):
 			pattern = re.compile("([\w:]+) at ([\d\-\.]+): (\w+)")
 			for log in logs[1:-1]:
 				match = pattern.match(log)
+				if match is None:
+					raise ValueError(f"Could not parse results line: {log!r}")
 				usb_ids = match.groups()[0]
 				path = match.groups()[1]
 				phase = BoardPhase[match.groups()[2]]
@@ -234,6 +238,8 @@ def load_log_section(self, marker: str, logs: list):
 		elif marker == "BOARD LOG":
 			pattern = re.compile("BOARD LOG ([\d\-\.]+):")
 			match = pattern.match(logs[0])
+			if match is None:
+				raise ValueError(f"Could not parse BOARD LOG header: {logs[0]!r}")
 			path = match.groups()[0]
 			if path not in self.board_dict:
 				raise KeyError(

src/snagflash/fastboot.py

@@ -85,13 +85,36 @@ def fastboot(args):
 	if args.protocol == "fastboot-uboot" and args.fastboot_cmd != []:
 		cli_error("The '-f' option is not available with the fastboot_uboot protocol!")
 
+	allowed_cmds = {
+		"getvar",
+		"download",
+		"erase",
+		"flash",
+		"boot",
+		"continue",
+		"reboot",
+		"reboot_bootloader",
+		"powerdown",
+		"ucmd",
+		"acmd",
+		"oem_run",
+		"oem_format",
+		"oem_partconf",
+		"oem_bootbus",
+		"reset",
+		"flash_sparse",
+	}
+
 	for cmd in args.fastboot_cmd:
 		cmd = cmd.split(":", 1)
 		cmd, cmd_args = cmd[0], cmd[1:]
 		cmd = cmd.replace("-", "_")
-		logger.info(f"Sending command {cmd} with args {cmd_args}")
+		if cmd not in allowed_cmds:
+			logger.error(f"Unknown fastboot command: {cmd}")
+			sys.exit(-1)
 		if cmd == "continue":
 			cmd = "fbcontinue"
+		logger.info(f"Sending command {cmd} with args {cmd_args}")
 		try:
 			getattr(fast, cmd)(*cmd_args)
 		except Exception as e:

src/snagflash/ums.py

@@ -56,6 +56,7 @@ def bmap_copy(filepath: str, dev, src_size: int):
 	mapfile = None
 	logger.info(f"Looking for {mappath}...")
 	gen_bmap = True
+	mapfileb = None
 	if os.path.exists(mappath):
 		logger.info(f"Found bmap file {mappath}")
 		gen_bmap = False
@@ -66,6 +67,8 @@ def bmap_copy(filepath: str, dev, src_size: int):
 		hdr = mapfileb.read(34)
 		if hdr == b"-----BEGIN PGP SIGNED MESSAGE-----":
 			logger.info("Warning: bmap file is clearsigned, skipping...")
+			mapfileb.close()
+			mapfileb = None
 			gen_bmap = True
 		else:
 			mapfileb.seek(0)
@@ -81,12 +84,15 @@ def bmap_copy(filepath: str, dev, src_size: int):
 		creator.generate(True)
 		mapfileb = open(mapfile.name, "rb")
 
-	with open_compressed_file(filepath, "rb") as src_file:
-		writer = BmapCopy.BmapBdevCopy(src_file, dev, mapfileb, src_size)
-		writer.copy(False, True)
-	mapfileb.close()
-	if mapfile is not None:
-		mapfile.close()
+	try:
+		with open_compressed_file(filepath, "rb") as src_file:
+			writer = BmapCopy.BmapBdevCopy(src_file, dev, mapfileb, src_size)
+			writer.copy(False, True)
+	finally:
+		if mapfileb is not None:
+			mapfileb.close()
+		if mapfile is not None:
+			mapfile.close()
 
 
 def write_raw(args):

src/snagrecover/protocols/fastboot.py

@@ -65,9 +65,6 @@ def __init__(self, dev: usb.core.Device, timeout: int = 10000):
 				is_bulk = (
 					ep.bmAttributes & usb.ENDPOINT_TYPE_MASK
 				) == usb.ENDPOINT_TYPE_BULK
-				is_in = (
-					ep.bmAttributes & usb.ENDPOINT_TYPE_MASK
-				) == usb.ENDPOINT_TYPE_BULK
 				if not is_bulk:
 					continue
 				is_in = (ep.bEndpointAddress & usb.ENDPOINT_DIR_MASK) == usb.ENDPOINT_IN

src/snagrecover/protocols/fel.py

@@ -55,9 +55,6 @@ def __init__(self, dev: usb.core.Device, timeout: int):
 				is_bulk = (
 					ep.bmAttributes & usb.ENDPOINT_TYPE_MASK
 				) == usb.ENDPOINT_TYPE_BULK
-				is_in = (
-					ep.bmAttributes & usb.ENDPOINT_TYPE_MASK
-				) == usb.ENDPOINT_TYPE_BULK
 				if not is_bulk:
 					continue
 				is_in = (ep.bEndpointAddress & usb.ENDPOINT_DIR_MASK) == usb.ENDPOINT_IN

src/snagrecover/protocols/hid.py

@@ -220,11 +220,11 @@ def set_report(self, report_id: int, data: bytes):
 
 	def libusb_read(self, length: int, timeout: int):
 		logger.debug(f"HID libusb read length: {length}")
-		data = self.intr_in.read(length + 1)[1:]
-		if isinstance(data, int) or data is None:
-			raise HIDError("Failed to read {length + 1} bytes from HID device")
+		raw = self.intr_in.read(length + 1)
+		if isinstance(raw, int) or raw is None:
+			raise HIDError(f"Failed to read {length + 1} bytes from HID device")
 
-		return bytes(data)
+		return bytes(raw[1:])
 
 	def close(self):
 		if self.hidraw:

src/snagrecover/protocols/imx_sdp.py

@@ -61,6 +61,7 @@
 from snagrecover.protocols.hid import HIDDevice, HIDError
 from usb.core import USBError
 import struct
+import time
 
 
 class SDPCommand:
@@ -224,31 +225,41 @@ def _process_dcd_write_data(self, addr, value_mask, param):
 
 		return self.write32(addr, value)
 
+	DCD_CHECK_TIMEOUT_S = 30.0
+	DCD_CHECK_POLL_INTERVAL_S = 0.01
+
 	def _process_dcd_check_data(self, addr, mask, param):
 		logger.debug("dcd check: addr=%08x mask=%08x param=%2x", addr, mask, param)
 		is_mask = bool(param & (1 << 3))
 		is_set = bool(param & (1 << 4))
 
-		while True:
+		deadline = time.monotonic() + __class__.DCD_CHECK_TIMEOUT_S
+		value = 0
+		while time.monotonic() < deadline:
 			value = self.read32(addr)
 			logger.debug("    check: value=%08x", value)
 			if (is_mask, is_set) == (False, False):
 				if (value & mask) == 0:
-					break
+					return True
 
 			if (is_mask, is_set) == (False, True):
 				if (value & mask) == mask:
-					break
+					return True
 
 			if (is_mask, is_set) == (True, False):
 				if (value & mask) != mask:
-					break
+					return True
 
 			if (is_mask, is_set) == (True, True):
 				if (value & mask) != 0:
-					break
+					return True
 
-		return True
+			time.sleep(__class__.DCD_CHECK_POLL_INTERVAL_S)
+
+		raise TimeoutError(
+			f"DCD check timed out after {__class__.DCD_CHECK_TIMEOUT_S}s: "
+			f"addr={addr:#010x} mask={mask:#010x} value={value:#010x}"
+		)
 
 	def write_blob(
 		self, blob: bytes, addr: int, offset: int, size: int, write_dcd: bool = False