From c17273e27438e140d267bb8bbadc9d2379497c66 Mon Sep 17 00:00:00 2001 From: Enric Balletbo i Serra Date: Tue, 27 Jan 2026 08:30:59 +0100 Subject: [PATCH 1/3] feat(flasher): enhance flash CLI for partition-based and multiple image flashing The `flash` CLI command now supports more flexible image flashing options. Users can specify images to flash to specific partitions using the new `-t` option (e.g., `-t rootfs:rootfs.img`). The CLI now also supports flashing multiple images in a single command when using the `-t` option. The command now handles: - Flashing a single file to a default or specified block device (`flash image.img`) - Flashing multiple file-partition pairs to a default or specified block device (`flash -t rootfs:rootfs.img -t boot:boot.img --target emmc`) This is implemented with a new internal `_resolve_flash_parameters` helper to validate and parse the various CLI arguments before executing each flash operation. The core `flash` method is updated to correctly interpret the `partition` argument as a partition label when provided, constructing the appropriate `/dev/disk/by-partlabel/` path. Signed-off-by: Enric Balletbo i Serra --- .../jumpstarter-driver-flashers/README.md | 39 ++- .../jumpstarter_driver_flashers/client.py | 252 +++++++++++++++--- 2 files changed, 254 insertions(+), 37 deletions(-) diff --git a/python/packages/jumpstarter-driver-flashers/README.md b/python/packages/jumpstarter-driver-flashers/README.md index 947a297f1..220f6d292 100644 --- a/python/packages/jumpstarter-driver-flashers/README.md +++ b/python/packages/jumpstarter-driver-flashers/README.md @@ -113,23 +113,54 @@ Commands: ### flash ```shell -Usage: j storage flash [OPTIONS] FILE +Usage: j storage flash [OPTIONS] [FILE] - Flash image to DUT from file + Flash image(s) to DUT + + Usage examples: + + - Flash to default block device and target + + j storage flash image.img + + - Flash to specific block device (e.g., 'emmc') + + j storage flash image.img --target emmc + + - Flash to partition(s) on default block device + + j storage flash -t rootfs:rootfs.img + + - Flash to partition(s) on specific block device + + j storage flash --target emmc -t rootfs:rootfs.img -t boot:boot.img Options: - --partition TEXT + --target TEXT Block device to flash to (e.g., 'usd', + 'emmc'). If not provided, uses default + target. + -t TEXT Flash file to partition: + 'partition:filename'. Can be repeated for + multiple partitions. --os-image-checksum TEXT SHA256 checksum of OS image (direct value) --os-image-checksum-file FILE File containing SHA256 checksum of OS image --force-exporter-http Force use of exporter HTTP --force-flash-bundle TEXT Force use of a specific flasher OCI bundle - --console-debug Enable console debug mode --cacert FILE CA certificate to use for HTTPS --insecure-tls Skip TLS certificate verification --header TEXT Custom HTTP header in 'Key: Value' format --bearer TEXT Bearer token for HTTP authentication --oci-username TEXT OCI registry username (or OCI_USERNAME environment variable) --oci-password TEXT OCI registry password (or OCI_PASSWORD environment variable) + --retries INTEGER Number of retry attempts for flash operation + (default: 3) + --method [fls|shell] Method to use for flash operation (default: + fls) + --fls-version TEXT Download an specific fls version from the + github releases + --fls-binary-url TEXT Custom URL to download FLS binary from + (overrides --fls-version) + --console-debug Enable console debug mode --help Show this message and exit. ``` diff --git a/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py b/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py index 1faa37ac5..0a97c5026 100644 --- a/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py +++ b/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py @@ -113,6 +113,7 @@ def flash( # noqa: C901 path: PathBuf, *, partition: str | None = None, + block_device: str | None = None, operator: Operator | None = None, os_image_checksum: str | None = None, force_exporter_http: bool = False, @@ -209,6 +210,7 @@ def flash( # noqa: C901 try: self._perform_flash_operation( partition, + block_device, path, image_url, should_download_to_httpd, @@ -333,9 +335,86 @@ def _find_exception_in_chain(self, exception: Exception, target_type: type) -> E current = getattr(current, "__cause__", None) return None + def _resolve_and_prepare_target( + self, + partition: str | None, + block_device: str | None, + manifest, + console, + ) -> str: + """Resolve partition/target device and prepare flash target path. + + Resolves the target for flashing, with priority to manifest targets when there + is ambiguity. Supports two modes: + 1. partition matches a manifest target name - use it as target (if collision with + partition_label suspected, log warning mentioning both interpretations) + 2. partition is a partition label - resolve block_device or use default target, + then construct the /dev/disk/by-partlabel path + + Args: + partition: Either a target device name (from manifest.spec.targets) or a + partition label. If it matches a manifest target, it takes priority. + block_device: Optional block device name (e.g., 'usd', 'emmc'). Ignored if + partition matches a manifest target. If partition is a partition + label, this specifies which block_device to target; if omitted, + uses default target from call("get_default_target") or + manifest.spec.default_target. + manifest: Flasher manifest containing targets and default_target + console: Console object for device interaction + + Returns: + The flash target path (either device path from _get_target_device() or + /dev/disk/by-partlabel/ path) + + Raises: + ArgumentError: If no valid target can be determined + """ + # Check if partition is a valid target device name in the manifest + is_target_name = partition and partition in manifest.spec.targets + + if is_target_name: + # Collision detection: partition matches a manifest target + if block_device: + self.logger.warning( + f"Ambiguous partition argument '{partition}': matches both a " + f"manifest target name and potentially a partition label. " + f"Treating as manifest target (ignoring block_device='{block_device}'). " + f"To use '{partition}' as a partition label on block_device '{block_device}', " + f"rename the partition label or use only the -t option with partition:file pairs." + ) + target = partition + target_device = self._get_target_device(target, manifest, console) + self.logger.info(f"Using manifest target: {target} -> {target_device}") + return target_device + + # partition is a partition label (or None), not a manifest target + partition_label = partition + if block_device: + target = block_device + self.logger.debug(f"Using block_device '{block_device}' for partition label '{partition_label}'") + else: + target = self.call("get_default_target") or manifest.spec.default_target + if partition_label: + self.logger.debug(f"Using default target for partition label '{partition_label}'") + + if not target: + raise ArgumentError("No partition or default target specified") + + target_device = self._get_target_device(target, manifest, console) + + if partition_label: + flash_target = f"/dev/disk/by-partlabel/{partition_label}" + self.logger.info(f"Using partition label: {partition_label} on {target} -> {flash_target}") + else: + flash_target = target_device + self.logger.info(f"Using target block device: {target_device}") + + return flash_target + def _perform_flash_operation( self, partition: str | None, + block_device: str | None, path: PathBuf, image_url: str, should_download_to_httpd: bool, @@ -357,13 +436,8 @@ def _perform_flash_operation( """ with self._busybox() as console: manifest = self.manifest - target = partition or self.call("get_default_target") or manifest.spec.default_target - if not target: - raise ArgumentError("No partition or default target specified") + flash_target = self._resolve_and_prepare_target(partition, block_device, manifest, console) - target_device = self._get_target_device(target, manifest, console) - - self.logger.info(f"Using target block device: {target_device}") console.sendline(f"export dhcp_addr={self._dhcp_details.ip_address}") console.expect(manifest.spec.login.prompt, timeout=EXPECT_TIMEOUT_DEFAULT) console.sendline(f"export gw_addr={self._dhcp_details.gateway}") @@ -408,7 +482,7 @@ def _perform_flash_operation( manifest, path, image_url, - target_device, + flash_target, insecure_tls, stored_cacert, header_args, @@ -423,7 +497,7 @@ def _perform_flash_operation( manifest, path, image_url, - target_device, + flash_target, insecure_tls, stored_cacert, header_args, @@ -1286,6 +1360,71 @@ def _cleanup_fls_oci_credential_file(self, console, prompt: str, creds_file: str console.sendline(f"rm -f {shlex.quote(creds_file)}") console.expect(prompt, timeout=EXPECT_TIMEOUT_DEFAULT) + def _resolve_flash_parameters( + self, file: str | None, partitions: tuple[str, ...] | None, block_device: str | None + ) -> list[tuple[str, str | None, str | None]]: + """Resolve and validate flash parameters from CLI options. + + Supports multiple modes: + 1. Single file to default block device: + flash image.img + 2. Single file to specific block device: + flash image.img --target usd + 3. Multiple file-partition pairs: + flash -t rootfs:rootfs.img -t boot:boot.img + 4. Multiple file-partition pairs to specific block device: + flash --target emmc -t rootfs:rootfs.img -t boot:boot.img + + Args: + file: The image file argument (positional, optional). When provided alone, + flashes to the default block device. + partitions: The -t options in 'partition:file' format (repeatable). + Use this to specify partition and file together. + block_device: The --target option (block device name like 'usd', 'emmc'). + Can be used with both file and -t options. + + Returns: + list[tuple]: List of (image_file, target_partition, block_device) tuples + + Raises: + click.UsageError: If parameters are invalid or conflicting + """ + flash_ops: list[tuple[str, str | None, str | None]] = [] + + # Mode 1 & 2: Single file with optional block device + if file: + if partitions: + raise click.UsageError( + "Cannot specify FILE argument with -t options. " + "Use either 'flash image.img' or 'flash -t partition:file'" + ) + flash_ops.append((file, None, block_device)) + + # Mode 3 & 4: Multiple file-partition pairs with optional block device + elif partitions: + for spec in partitions: + if ':' not in spec: + raise click.UsageError( + f"Invalid flash spec format: '{spec}'. " + "Expected 'partition:filename'" + ) + partition_label, filename = spec.split(':', 1) + if not partition_label or not filename: + raise click.UsageError( + f"Invalid flash spec format: '{spec}'. " + "Both partition label and filename are required" + ) + flash_ops.append((filename, partition_label, block_device)) + + # No input provided + else: + raise click.UsageError( + "Must provide either FILE argument or -t options. " + "Use 'j storage flash --help' for usage examples" + ) + + return flash_ops + def cli(self): @driver_click_group(self) def base(): @@ -1293,8 +1432,18 @@ def base(): pass @base.command() - @click.argument("file") - @click.option("--target", type=str) + @click.argument("file", required=False) + @click.option( + "--target", + type=str, + help="Block device to flash to (e.g., 'usd', 'emmc'). If not provided, uses default target." + ) + @click.option( + "-t", + "partitions", + multiple=True, + help="Flash file to partition: 'partition:filename'. Can be repeated for multiple partitions.", + ) @click.option("--os-image-checksum", help="SHA256 checksum of OS image (direct value)") @click.option( "--os-image-checksum-file", @@ -1355,6 +1504,7 @@ def base(): def flash( file, target, + partitions, os_image_checksum, os_image_checksum_file, console_debug, @@ -1371,33 +1521,69 @@ def flash( fls_version, fls_binary_url, ): - """Flash image to DUT from file""" - if os_image_checksum_file and os.path.exists(os_image_checksum_file): - with open(os_image_checksum_file) as f: - os_image_checksum = f.read().strip().split()[0] - self.logger.info(f"Read checksum from file: {os_image_checksum}") + """Flash image(s) to DUT - self.set_console_debug(console_debug) + Usage examples: + + - Flash to default block device and target + + j storage flash image.img + + - Flash to specific block device (e.g., 'emmc') + + j storage flash image.img --target emmc + + - Flash to partition(s) on default block device - headers = self._parse_headers(header) if header else None - - self.flash( - file, - partition=target, - force_exporter_http=force_exporter_http, - force_flash_bundle=force_flash_bundle, - cacert_file=cacert, - insecure_tls=insecure_tls, - headers=headers, - bearer_token=bearer, - oci_username=oci_username, - oci_password=oci_password, - retries=retries, - method=method, - fls_version=fls_version, - fls_binary_url=fls_binary_url, + j storage flash -t rootfs:rootfs.img + + - Flash to partition(s) on specific block device + + j storage flash --target emmc -t rootfs:rootfs.img -t boot:boot.img + """ + # Validate and resolve flash parameters + flash_operations = self._resolve_flash_parameters( + file, partitions, target ) + # Setup common options + self.set_console_debug(console_debug) + headers_dict = self._parse_headers(header) if header else None + + # Load checksum from file if provided (used for all operations) + checksum = os_image_checksum + if os_image_checksum_file and os.path.exists(os_image_checksum_file): + with open(os_image_checksum_file) as f: + checksum = f.read().strip().split()[0] + self.logger.info(f"Read checksum from file: {checksum}") + + # Execute each flash operation + for idx, (image_file, target_partition, block_device) in enumerate(flash_operations): + op_num = f"{idx + 1}/{len(flash_operations)}" if len(flash_operations) > 1 else "" + op_desc = f"partition '{target_partition}'" if target_partition else "default target" + + self.logger.info(f"Flashing {op_num} {op_desc} with '{image_file}'".strip()) + + # Perform the flash operation + self.flash( + image_file, + partition=target_partition, + block_device=block_device, + os_image_checksum=checksum, + force_exporter_http=force_exporter_http, + force_flash_bundle=force_flash_bundle, + cacert_file=cacert, + insecure_tls=insecure_tls, + headers=headers_dict, + bearer_token=bearer, + oci_username=oci_username, + oci_password=oci_password, + retries=retries, + method=method, + fls_version=fls_version, + fls_binary_url=fls_binary_url, + ) + @base.command() @debug_console_option def bootloader_shell(console_debug): From c37ebd1bafcf27aec6d4d5befe6bc7cd25caccf8 Mon Sep 17 00:00:00 2001 From: Benny Zlotnik Date: Wed, 11 Feb 2026 14:31:30 +0200 Subject: [PATCH 2/3] convert paths to str Signed-off-by: Benny Zlotnik --- .../jumpstarter_driver_flashers/client.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py b/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py index 0a97c5026..87b41477b 100644 --- a/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py +++ b/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py @@ -654,7 +654,7 @@ def _flash_with_fls( # Flash the image creds_file = None with self._redaction_scope([oci_username, oci_password]): - if path.startswith("oci://") and oci_username: + if str(path).startswith("oci://") and oci_username: creds_file = self._setup_fls_oci_credential_file(console, prompt, oci_username, oci_password or "") fls_oci_auth_env = self._fls_oci_auth_env(path, creds_file) @@ -1311,7 +1311,7 @@ def _validate_oci_credentials( return username, password def _fls_oci_auth_env(self, path: PathBuf, creds_file: str | None) -> str: - if not path.startswith("oci://") or not creds_file: + if not str(path).startswith("oci://") or not creds_file: return "" return f"set -o allexport; . {shlex.quote(creds_file)}; set +o allexport;" From cea7d2370f7278a6337094ef3bd0877c79f0f117 Mon Sep 17 00:00:00 2001 From: Benny Zlotnik Date: Wed, 11 Feb 2026 14:49:10 +0200 Subject: [PATCH 3/3] add tests Signed-off-by: Benny Zlotnik --- .../client_test.py | 32 ++++++++++++++++--- 1 file changed, 28 insertions(+), 4 deletions(-) diff --git a/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client_test.py b/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client_test.py index 19acddadd..0f6f2420b 100644 --- a/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client_test.py +++ b/python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client_test.py @@ -1,5 +1,6 @@ import shlex from concurrent.futures import CancelledError +from pathlib import PosixPath import click import pytest @@ -86,6 +87,10 @@ def test_fls_oci_auth_env_empty_for_non_oci_paths(): env_args = client._fls_oci_auth_env("oci://quay.io/org/image:tag", None) assert env_args == "" + # PosixPath (converted by operator_for_path) must not crash + env_args = client._fls_oci_auth_env(PosixPath("/images/image.raw.xz"), "/tmp/fls_creds") + assert env_args == "" + def test_redact_sensitive_values_masks_username_and_password(): """Test that sensitive values are redacted from output.""" @@ -149,10 +154,10 @@ def get_url(self): captured = {} def capture_perform(*args): - captured["image_url"] = args[2] - captured["should_download_to_httpd"] = args[3] - captured["oci_username"] = args[13] - captured["oci_password"] = args[14] + captured["image_url"] = args[3] + captured["should_download_to_httpd"] = args[4] + captured["oci_username"] = args[14] + captured["oci_password"] = args[15] client._perform_flash_operation = capture_perform @@ -326,3 +331,22 @@ def test_categorize_exception_preserves_cause_for_wrapped_exceptions(): # IOError is an alias for OSError in Python 3 assert "OSError" in str(result) or "IOError" in str(result) assert "File not found" in str(result) + + +def test_resolve_flash_parameters(): + """Test flash parameter resolution for single file, partitions, and error cases""" + client = MockFlasherClient() + + assert client._resolve_flash_parameters("image.img", None, None) == [("image.img", None, None)] + assert client._resolve_flash_parameters("image.img", None, "emmc") == [("image.img", None, "emmc")] + assert client._resolve_flash_parameters(None, ("rootfs:rootfs.img", "boot:boot.img"), "emmc") == [ + ("rootfs.img", "rootfs", "emmc"), + ("boot.img", "boot", "emmc"), + ] + + with pytest.raises(click.UsageError): + client._resolve_flash_parameters("image.img", ("rootfs:rootfs.img",), None) + with pytest.raises(click.UsageError): + client._resolve_flash_parameters(None, None, None) + with pytest.raises(click.UsageError): + client._resolve_flash_parameters(None, ("rootfs_no_colon",), None)