add support for registry creds

Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com>

Author: bennyz

python/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/client.py

@@ -1,3 +1,4 @@
+import os
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Dict, Optional
@@ -215,12 +216,33 @@ def _flash_operation():
 
         return self._execute_flash_operation(_flash_operation)
 
+    def _read_oci_credentials(self):
+        """Read OCI registry credentials from environment variables.
+
+        Returns:
+            Tuple of (username, password), both None if not set.
+
+        Raises:
+            click.ClickException: If only one of username/password is set.
+        """
+        username = os.environ.get("OCI_USERNAME")
+        password = os.environ.get("OCI_PASSWORD")
+
+        if bool(username) != bool(password):
+            raise click.ClickException(
+                "OCI authentication requires both OCI_USERNAME and OCI_PASSWORD environment variables"
+            )
+
+        return username, password
+
     def _flash_oci_auto_impl(
         self,
         oci_url: str,
         partitions: Dict[str, str] | None = None,
     ):
         """Core implementation of OCI flash without wrapper logic."""
+        oci_username, oci_password = self._read_oci_credentials()
+
         self.logger.info("Checking for fastboot devices on Exporter...")
         detection_result = self.call("detect_fastboot_device", 5, 2.0)
 
@@ -230,7 +252,10 @@ def _flash_oci_auto_impl(
         device_id = detection_result["device_id"]
         self.logger.info(f"Found fastboot device: {device_id}")
 
-        flash_result = self.call("flash_oci_image", oci_url, partitions)
+        flash_result = self.call(
+            "flash_oci_image", oci_url, partitions,
+            oci_username, oci_password,
+        )
 
         # Display FLS output to user
         if flash_result.get("status") == "success" and flash_result.get("output"):
@@ -360,6 +385,9 @@ def flash(path, target_specs):
               # OCI with explicit partition->filename mapping
               j storage flash -t boot_a:boot.img oci://registry.com/image:tag
 
+              # OCI with registry credentials (via env vars)
+              OCI_USERNAME=user OCI_PASSWORD=pass j storage flash oci://registry.com/image:tag
+
               # Single file to partition
               j storage flash /local/boot.img --target boot_a
 
@@ -368,6 +396,11 @@ def flash(path, target_specs):
 
               # HTTP URLs
               j storage flash -t boot_a:https://example.com/boot.img
+
+            \b
+            Environment variables:
+              OCI_USERNAME  Registry username for private OCI images
+              OCI_PASSWORD  Registry password for private OCI images
             """
             self._execute_flash_command(path, target_specs)
 

python/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 import subprocess
 import time
 from dataclasses import dataclass, field
@@ -220,21 +221,8 @@ def flash_with_fastboot(self, device_id: str, partitions: Dict[str, str]):
             self.logger.warning(f"stdout: {e.stdout}")
             self.logger.warning(f"stderr: {e.stderr}")
 
-    @export
-    def flash_oci_image(
-        self,
-        oci_url: str,
-        partitions: Dict[str, str] | None = None,
-    ):
-        """Flash OCI image using FLS fastboot CLI
-
-        Args:
-            oci_url: OCI image reference (e.g., "quay.io/bzlotnik/ridesx-image:latest")
-            partitions: Optional mapping of partition -> filename inside OCI image
-        """
-        if not oci_url.startswith("oci://"):
-            raise ValueError(f"OCI URL must start with oci://, got: {oci_url}")
-
+    def _build_fls_command(self, oci_url, partitions):
+        """Build FLS fastboot command and environment."""
         fls_binary = get_fls_binary(
             fls_version=self.fls_version,
             fls_binary_url=self.fls_custom_binary_url,
@@ -251,13 +239,47 @@ def flash_oci_image(
                     )
                 fls_cmd.extend(["-t", f"{partition_name}:{filename}"])
 
-        # Align fastboot timeout with driver timeout
         fls_cmd.extend(["--timeout", str(self.flash_timeout)])
+        return fls_cmd
+
+    @export
+    def flash_oci_image(
+        self,
+        oci_url: str,
+        partitions: Dict[str, str] | None = None,
+        oci_username: str | None = None,
+        oci_password: str | None = None,
+    ):
+        """Flash OCI image using FLS fastboot CLI
+
+        Args:
+            oci_url: OCI image reference (e.g., "quay.io/bzlotnik/ridesx-image:latest")
+            partitions: Optional mapping of partition -> filename inside OCI image
+            oci_username: Registry username for OCI authentication
+            oci_password: Registry password for OCI authentication
+        """
+        if not oci_url.startswith("oci://"):
+            raise ValueError(f"OCI URL must start with oci://, got: {oci_url}")
+
+        if bool(oci_username) != bool(oci_password):
+            raise ValueError("OCI authentication requires both --username and --password")
+
+        fls_cmd = self._build_fls_command(oci_url, partitions)
+
+        fls_env = os.environ.copy()
+        if oci_username and oci_password:
+            fls_env["FLS_REGISTRY_USERNAME"] = oci_username
+            fls_env["FLS_REGISTRY_PASSWORD"] = oci_password
 
         self.logger.info(f"Running FLS fastboot: {' '.join(fls_cmd)}")
+        if oci_username:
+            self.logger.info("Using OCI registry credentials from environment")
 
         try:
-            result = subprocess.run(fls_cmd, capture_output=True, text=True, check=True, timeout=self.flash_timeout)
+            result = subprocess.run(
+                fls_cmd, capture_output=True, text=True,
+                check=True, timeout=self.flash_timeout, env=fls_env,
+            )
 
             self.logger.info("FLS fastboot auto-detection completed successfully")
             self.logger.debug(f"FLS stdout: {result.stdout}")
@@ -267,10 +289,11 @@ def flash_oci_image(
             return {"status": "success", "output": result.stdout}
 
         except subprocess.CalledProcessError as e:
-            self.logger.error(f"FLS fastboot auto-detection failed - return code: {e.returncode}")
+            self.logger.error(f"FLS fastboot failed - return code: {e.returncode}")
             self.logger.error(f"stdout: {e.stdout}")
             self.logger.error(f"stderr: {e.stderr}")
-            raise RuntimeError(f"FLS fastboot auto-detection failed: {e}") from e
+            output = (e.stderr or e.stdout or "").strip()
+            raise RuntimeError(f"FLS fastboot failed: {output}") from e
 
         except subprocess.TimeoutExpired:
             self.logger.error("FLS fastboot auto-detection timed out")

python/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver_test.py

@@ -515,7 +515,7 @@ def test_flash_oci_image_error_cases(temp_storage_dir, ridesx_driver):
                 error.stderr = "Flash failed"
                 mock_subprocess.side_effect = error
 
-                with pytest.raises(DriverError, match="FLS fastboot auto-detection failed"):
+                with pytest.raises(DriverError, match="FLS fastboot failed: Flash failed"):
                     client.call("flash_oci_image", "oci://image:tag", None)
 
                 # TimeoutExpired
@@ -531,6 +531,67 @@ def test_flash_oci_image_error_cases(temp_storage_dir, ridesx_driver):
                     client.call("flash_oci_image", "oci://image:tag", None)
 
 
+def test_flash_oci_image_with_credentials(temp_storage_dir, ridesx_driver):
+    """Test that OCI credentials are passed via env vars to FLS"""
+    with serve(ridesx_driver) as client:
+        with patch("jumpstarter_driver_ridesx.driver.get_fls_binary", return_value="fls"):
+            with patch("subprocess.run") as mock_subprocess:
+                mock_result = MagicMock()
+                mock_result.stdout = "Flashing complete"
+                mock_result.stderr = ""
+                mock_result.returncode = 0
+                mock_subprocess.return_value = mock_result
+
+                result = client.call(
+                    "flash_oci_image", "oci://quay.io/private/image:tag", None, "myuser", "mypass"
+                )
+
+                assert result["status"] == "success"
+                # Credentials should NOT appear in the command args
+                call_args = mock_subprocess.call_args[0][0]
+                assert "-u" not in call_args
+                assert "-p" not in call_args
+                assert "myuser" not in call_args
+                assert "mypass" not in call_args
+                # Credentials should be passed via env vars
+                call_kwargs = mock_subprocess.call_args[1]
+                env = call_kwargs["env"]
+                assert env["FLS_REGISTRY_USERNAME"] == "myuser"
+                assert env["FLS_REGISTRY_PASSWORD"] == "mypass"
+
+
+def test_flash_oci_image_partial_credentials_rejected(temp_storage_dir, ridesx_driver):
+    """Test that providing only username or only password is rejected"""
+    from jumpstarter.client.core import DriverError
+
+    with serve(ridesx_driver) as client:
+        with pytest.raises(DriverError, match="OCI authentication requires both"):
+            client.call("flash_oci_image", "oci://image:tag", None, "myuser", None)
+
+        with pytest.raises(DriverError, match="OCI authentication requires both"):
+            client.call("flash_oci_image", "oci://image:tag", None, None, "mypass")
+
+
+def test_flash_oci_image_no_credentials(temp_storage_dir, ridesx_driver):
+    """Test that omitting credentials works (anonymous access)"""
+    with serve(ridesx_driver) as client:
+        with patch("jumpstarter_driver_ridesx.driver.get_fls_binary", return_value="fls"):
+            with patch("subprocess.run") as mock_subprocess:
+                mock_result = MagicMock()
+                mock_result.stdout = "Flashing complete"
+                mock_result.stderr = ""
+                mock_result.returncode = 0
+                mock_subprocess.return_value = mock_result
+
+                result = client.call("flash_oci_image", "oci://image:tag", None, None, None)
+
+                assert result["status"] == "success"
+                call_kwargs = mock_subprocess.call_args[1]
+                env = call_kwargs["env"]
+                assert "FLS_REGISTRY_USERNAME" not in env
+                assert "FLS_REGISTRY_PASSWORD" not in env
+
+
 def test_flash_oci_image_requires_oci_scheme(temp_storage_dir, ridesx_driver):
     """Test that only oci:// URLs are accepted"""
     from jumpstarter.client.core import DriverError