oci: add OciCredentials model and harden credential resolution

Introduce a Pydantic OciCredentials model that normalizes empty/whitespace
strings and enforces both-or-neither validation. Update all callers to use
the new model. Add tests for digest parsing, Docker Hub bare images, and
whitespace env var handling.

Assisted-by: claude-opus-4.6
Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com>

Author: bennyz

python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import base64
 import hashlib
 import json
@@ -12,8 +14,12 @@
 from dataclasses import dataclass
 from pathlib import Path, PosixPath
 from queue import Queue
+from typing import TYPE_CHECKING
 from urllib.parse import urlparse
 
+if TYPE_CHECKING:
+    from jumpstarter.common.oci import OciCredentials
+
 import click
 import pexpect
 import requests
@@ -137,7 +143,9 @@ def flash(  # noqa: C901
 
         if headers:
             headers = self._validate_header_dict(headers)
-        oci_username, oci_password = self._resolve_oci_credentials(path, oci_username, oci_password)
+        oci_creds = self._resolve_oci_credentials(path, oci_username, oci_password)
+        oci_username = oci_creds.username
+        oci_password = oci_creds.password.get_secret_value() if oci_creds.password else None
         should_download_to_httpd = True
         image_url = ""
         original_http_url = None
@@ -1291,34 +1299,25 @@ def _validate_bearer_token(self, token: str | None) -> str | None:
 
         return token
 
-    def _validate_oci_credentials(self, username: str | None, password: str | None) -> tuple[str | None, str | None]:
-        if username is not None:
-            username = username.strip()
-        if password is not None:
-            password = password.strip()
+    def _resolve_oci_credentials(
+        self, path: PathBuf, username: str | None, password: str | None
+    ) -> "OciCredentials":
+        from pydantic import SecretStr, ValidationError
 
-        if username == "":
-            username = None
-        if password == "":
-            password = None
+        from jumpstarter.common.oci import OciCredentials, resolve_oci_credentials
 
-        if bool(username) != bool(password):
+        try:
+            creds = OciCredentials(username=username, password=SecretStr(password) if password is not None else None)
+        except ValidationError as err:
             raise click.ClickException(
                 "OCI authentication requires both OCI_USERNAME and OCI_PASSWORD "
                 "environment variables (or both oci_username and oci_password arguments)"
-            )
-
-        return username, password
-
-    def _resolve_oci_credentials(
-        self, path: PathBuf, username: str | None, password: str | None
-    ) -> tuple[str | None, str | None]:
-        if username is None and password is None and path.startswith("oci://"):
-            from jumpstarter.common.oci import resolve_oci_credentials
+            ) from err
 
-            username, password = resolve_oci_credentials(str(path))
+        if not creds.is_authenticated and path.startswith("oci://"):
+            return resolve_oci_credentials(str(path))
 
-        return self._validate_oci_credentials(username, password)
+        return creds
 
     def _fls_oci_auth_env(self, path: PathBuf, creds_file: str | None) -> str:
         if not str(path).startswith("oci://") or not creds_file:

python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client_test.py

@@ -47,24 +47,25 @@ def test_validate_bearer_token_fails_invalid():
         client._validate_bearer_token('token"with"quotes')
 
 
-def test_validate_oci_credentials_fails_when_partial():
-    """Test OCI credential validation fails when only one value is provided"""
+def test_resolve_oci_credentials_fails_when_partial():
+    """Test OCI credential resolution fails when only one value is provided"""
     client = MockFlasherClient()
 
     with pytest.raises(click.ClickException, match="OCI authentication requires both"):
-        client._validate_oci_credentials("myuser", None)
+        client._resolve_oci_credentials("oci://quay.io/org/image:tag", "myuser", None)
 
     with pytest.raises(click.ClickException, match="OCI authentication requires both"):
-        client._validate_oci_credentials(None, "mypassword")
+        client._resolve_oci_credentials("oci://quay.io/org/image:tag", None, "mypassword")
 
 
-def test_validate_oci_credentials_accepts_pair_and_strips_whitespace():
-    """Test OCI credential validation accepts full username/password pair"""
+def test_resolve_oci_credentials_accepts_pair_and_strips_whitespace():
+    """Test OCI credential resolution accepts full username/password pair and strips whitespace"""
     client = MockFlasherClient()
 
-    username, password = client._validate_oci_credentials(" myuser ", " mypassword ")
-    assert username == "myuser"
-    assert password == "mypassword"
+    creds = client._resolve_oci_credentials("oci://quay.io/org/image:tag", " myuser ", " mypassword ")
+    assert creds.username == "myuser"
+    assert creds.password is not None
+    assert creds.password.get_secret_value() == "mypassword"
 
 
 def test_resolve_oci_credentials_reads_env_for_oci_path(monkeypatch):
@@ -73,9 +74,10 @@ def test_resolve_oci_credentials_reads_env_for_oci_path(monkeypatch):
     monkeypatch.setenv("OCI_USERNAME", "env-user")
     monkeypatch.setenv("OCI_PASSWORD", "env-pass")
 
-    username, password = client._resolve_oci_credentials("oci://quay.io/org/image:tag", None, None)
-    assert username == "env-user"
-    assert password == "env-pass"
+    creds = client._resolve_oci_credentials("oci://quay.io/org/image:tag", None, None)
+    assert creds.username == "env-user"
+    assert creds.password is not None
+    assert creds.password.get_secret_value() == "env-pass"
 
 
 def test_resolve_oci_credentials_ignores_env_for_non_oci_path(monkeypatch):
@@ -84,30 +86,53 @@ def test_resolve_oci_credentials_ignores_env_for_non_oci_path(monkeypatch):
     monkeypatch.setenv("OCI_USERNAME", "env-user")
     monkeypatch.setenv("OCI_PASSWORD", "env-pass")
 
-    username, password = client._resolve_oci_credentials("https://example.com/image.raw.xz", None, None)
-    assert username is None
-    assert password is None
+    creds = client._resolve_oci_credentials("https://example.com/image.raw.xz", None, None)
+    assert creds.username is None
+    assert creds.password is None
 
 
 def test_resolve_oci_credentials_partial_env_falls_through_to_auth_file(monkeypatch):
     """Partial env vars should fall through to auth file lookup, not error."""
     from unittest.mock import patch
 
+    from pydantic import SecretStr
+
+    from jumpstarter.common.oci import OciCredentials
+
     client = MockFlasherClient()
     monkeypatch.setenv("OCI_USERNAME", "env-user")
     monkeypatch.delenv("OCI_PASSWORD", raising=False)
 
-    # When auth file has no match, result is (None, None) — no error
-    with patch("jumpstarter.common.oci.read_auth_file_credentials", return_value=(None, None)):
-        username, password = client._resolve_oci_credentials("oci://quay.io/org/image:tag", None, None)
-        assert username is None
-        assert password is None
+    # When auth file has no match, result is unauthenticated — no error
+    with patch("jumpstarter.common.oci.read_auth_file_credentials", return_value=OciCredentials()):
+        creds = client._resolve_oci_credentials("oci://quay.io/org/image:tag", None, None)
+        assert creds.username is None
+        assert creds.password is None
 
     # When auth file has a match, those credentials are used
-    with patch("jumpstarter.common.oci.read_auth_file_credentials", return_value=("fileuser", "filepass")):
-        username, password = client._resolve_oci_credentials("oci://quay.io/org/image:tag", None, None)
-        assert username == "fileuser"
-        assert password == "filepass"
+    with patch(
+        "jumpstarter.common.oci.read_auth_file_credentials",
+        return_value=OciCredentials(username="fileuser", password=SecretStr("filepass")),
+    ):
+        creds = client._resolve_oci_credentials("oci://quay.io/org/image:tag", None, None)
+        assert creds.username == "fileuser"
+        assert creds.password is not None
+        assert creds.password.get_secret_value() == "filepass"
+
+
+def test_resolve_oci_credentials_normalizes_empty_strings():
+    """Empty-string username/password should be treated as absent and fall back to resolve."""
+    from unittest.mock import patch
+
+    from jumpstarter.common.oci import OciCredentials
+
+    client = MockFlasherClient()
+
+    with patch("jumpstarter.common.oci.resolve_oci_credentials", return_value=OciCredentials()) as mock_resolve:
+        creds = client._resolve_oci_credentials("oci://quay.io/org/image:tag", "", "")
+        assert creds.username is None
+        assert creds.password is None
+        mock_resolve.assert_called_once()
 
 
 def test_fls_oci_auth_env_sources_credentials_file():

python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver.py

@@ -105,19 +105,17 @@ async def flash_oci(
         if not oci_url.startswith("oci://"):
             raise ValueError(f"OCI URL must start with oci://, got: {oci_url}")
 
-        # If explicit credentials were provided, validate immediately
+        from pydantic import SecretStr
+
+        from jumpstarter.common.oci import OciCredentials, resolve_oci_credentials
+
         if oci_username or oci_password:
-            if bool(oci_username) != bool(oci_password):
-                raise ValueError("OCI authentication requires both username and password")
+            creds = OciCredentials(username=oci_username, password=SecretStr(oci_password) if oci_password is not None else None)
         else:
-            # Fall back to env vars, then container auth files
-            from jumpstarter.common.oci import resolve_oci_credentials
-
-            oci_username, oci_password = resolve_oci_credentials(oci_url)
-            if oci_username and oci_password:
-                self.logger.info("Using OCI registry credentials from environment or auth file")
-            elif oci_username or oci_password:
-                raise ValueError("OCI authentication requires both username and password")
+            creds = resolve_oci_credentials(oci_url)
+
+        oci_username = creds.username
+        oci_password = creds.password.get_secret_value() if creds.password else None
 
         target_path = str(self.parent.validate_partition(partition))
 

python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py

@@ -14,6 +14,7 @@
 
 from jumpstarter_driver_qemu.driver import Qemu, QemuFlasher
 
+from jumpstarter.common.oci import OciCredentials
 from jumpstarter.common.utils import serve
 
 
@@ -316,7 +317,7 @@ async def test_flash_oci_no_credentials():
     # Ensure OCI env vars are not set so driver doesn't pick them up
     env_clean = {k: v for k, v in os.environ.items() if k not in ("OCI_USERNAME", "OCI_PASSWORD")}
     with patch.dict(os.environ, env_clean, clear=True):
-        with patch("jumpstarter.common.oci.read_auth_file_credentials", return_value=(None, None)):
+        with patch("jumpstarter.common.oci.read_auth_file_credentials", return_value=OciCredentials()):
             with patch("jumpstarter_driver_qemu.driver.get_fls_binary", return_value="fls"):
                 with patch(
                     "asyncio.create_subprocess_exec", new_callable=AsyncMock, return_value=mock_process

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

@@ -279,20 +279,15 @@ def _read_oci_credentials(self, oci_url: str):
         """Read OCI registry credentials from environment variables or auth files."""
         from jumpstarter.common.oci import resolve_oci_credentials
 
-        username, password = resolve_oci_credentials(oci_url)
-
-        if bool(username) != bool(password):
-            raise click.ClickException("OCI authentication requires both username and password")
-
-        return username, password
+        return resolve_oci_credentials(oci_url)
 
     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(oci_url)
+        creds = self._read_oci_credentials(oci_url)
 
         self.logger.info("Checking for fastboot devices on Exporter...")
         detection_result = self.call("detect_fastboot_device", 5, 2.0)
@@ -307,8 +302,8 @@ def _flash_oci_auto_impl(
             "flash_oci_image",
             oci_url,
             partitions,
-            oci_username,
-            oci_password,
+            creds.username,
+            creds.password.get_secret_value() if creds.password else None,
         )
 
         # Display FLS output to user

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

@@ -6,6 +6,7 @@
 from jumpstarter_driver_pyserial.driver import PySerial
 
 from .driver import RideSXDriver
+from jumpstarter.common.oci import OciCredentials
 from jumpstarter.common.utils import serve
 
 
@@ -57,20 +58,21 @@ def test_validate_partition_mappings(ridesx_client):
 
 def test_flash_oci_auto_success(ridesx_client):
     """Test successful flash_oci_auto call"""
-    with patch.object(ridesx_client, "call") as mock_call:
-        mock_call.side_effect = [
-            None,  # boot_to_fastboot call
-            {"status": "device_found", "device_id": "ABC123"},
-            {"status": "success"},
-        ]
+    with patch("jumpstarter.common.oci.resolve_oci_credentials", return_value=OciCredentials()):
+        with patch.object(ridesx_client, "call") as mock_call:
+            mock_call.side_effect = [
+                None,  # boot_to_fastboot call
+                {"status": "device_found", "device_id": "ABC123"},
+                {"status": "success"},
+            ]
 
-        result = ridesx_client.flash_oci_auto("oci://quay.io/org/image:tag")
+            result = ridesx_client.flash_oci_auto("oci://quay.io/org/image:tag")
 
-        assert result == {"status": "success"}
-        # Verify flash_oci_image was called with the OCI URL
-        flash_call = mock_call.call_args_list[2]
-        assert flash_call[0][0] == "flash_oci_image"
-        assert flash_call[0][1] == "oci://quay.io/org/image:tag"
+            assert result == {"status": "success"}
+            # Verify flash_oci_image was called with the OCI URL
+            flash_call = mock_call.call_args_list[2]
+            assert flash_call[0][0] == "flash_oci_image"
+            assert flash_call[0][1] == "oci://quay.io/org/image:tag"
 
 
 def test_flash_oci_auto_error_cases(ridesx_client):
@@ -84,11 +86,12 @@ def test_flash_oci_auto_error_cases(ridesx_client):
         ridesx_client.flash_oci_auto("quay.io/org/image:tag")
 
     # No device found
-    with patch.object(ridesx_client, "call") as mock_call:
-        mock_call.return_value = {"status": "no_device_found", "device_id": None}
+    with patch("jumpstarter.common.oci.resolve_oci_credentials", return_value=OciCredentials()):
+        with patch.object(ridesx_client, "call") as mock_call:
+            mock_call.return_value = {"status": "no_device_found", "device_id": None}
 
-        with pytest.raises(click.ClickException, match="No fastboot devices found"):
-            ridesx_client.flash_oci_auto("oci://image:tag")
+            with pytest.raises(click.ClickException, match="No fastboot devices found"):
+                ridesx_client.flash_oci_auto("oci://image:tag")
 
 
 # _execute_flash_command Tests