[Konflux] Support registry auth for bundle build

doozer/doozerlib/backend/konflux_olm_bundler.py

@@ -249,7 +249,12 @@ async def _replace_image_references(self, old_registry: str, content: str):
         # Get image infos for all found images
         for pullspec, (namespace, image_short_name, image_tag) in references.items():
             build_pullspec = f"{self.image_repo}:{image_short_name}-{image_tag}"
-            image_info_tasks.append(asyncio.create_task(util.oc_image_info__caching_async(build_pullspec)))
+            image_info_tasks.append(asyncio.create_task(
+                util.oc_image_info_async__caching(
+                    build_pullspec,
+                    registry_username=os.environ.get('KONFLUX_ART_IMAGES_USERNAME'),
+                    registry_password=os.environ.get('KONFLUX_ART_IMAGES_PASSWORD'),
+                )))
         image_infos = await asyncio.gather(*image_info_tasks)
 
         # Replace image references in the content

doozer/doozerlib/cli/scan_sources_konflux.py

@@ -28,7 +28,7 @@
 from doozerlib.runtime import Runtime
 from doozerlib.source_resolver import SourceResolver
 from artcommonlib.release_util import isolate_timestamp_in_release
-from doozerlib.util import oc_image_info__caching_async, isolate_el_version_in_brew_tag
+from doozerlib.util import oc_image_info_async__caching, isolate_el_version_in_brew_tag
 
 DEFAULT_THRESHOLD_HOURS = 6
 
@@ -605,7 +605,7 @@ async def get_builder_build_nvr(self, builder_image_name: str):
             builder_image_url = self.runtime.resolve_brew_image_url(builder_image_name)
 
         # Find and map the builder image NVR
-        latest_builder_image_info = Model(await oc_image_info__caching_async(builder_image_url))
+        latest_builder_image_info = Model(await oc_image_info_async__caching(builder_image_url))
         builder_info_labels = latest_builder_image_info.config.config.Labels
         builder_nvr_list = [builder_info_labels['com.redhat.component'], builder_info_labels['version'],
                             builder_info_labels['release']]

doozer/doozerlib/util.py

@@ -1,9 +1,11 @@
+import base64
 import copy
 import functools
 import json
 import os
 import pathlib
 import re
+import tempfile
 import urllib.parse
 from collections import deque
 from datetime import datetime
@@ -560,28 +562,81 @@ def oc_image_info__caching(pull_spec: str, go_arch: str = 'amd64') -> Dict:
     return oc_image_info(pull_spec, go_arch)
 
 
-@retry(reraise=True, wait=wait_fixed(3), stop=stop_after_attempt(3))
-async def oc_image_info_async(pull_spec: str, go_arch: str = 'amd64') -> Dict:
+async def oc_image_info_async(
+        pull_spec: str,
+        go_arch: str = 'amd64',
+        registry_config: Optional[str] = None,
+        registry_username: Optional[str] = None,
+        registry_password: Optional[str] = None,
+) -> Dict:
     """
     Returns a Dict of the parsed JSON output of `oc image info` for the specified
     pullspec. Filter by os because images can be multi-arch manifest lists
     (which cause oc image info to throw an error if not filtered).
-
-    Use oc_image_info__caching if you think the image won't change during the course of doozer's execution.
-    """
-    cmd = ['oc', 'image', 'info', f'--filter-by-os={go_arch}', '-o', 'json', pull_spec]
-    _, out, _ = await exectools.cmd_gather_async(cmd)
-    return json.loads(out)
+    This function will authenticate with the registry using the provided registry_config or username and password.
+
+    Use oc_image_info_async__caching if you think the image won't change during the course of doozer
+    execution.
+
+    :param pull_spec: The image pullspec to query.
+    :param go_arch: The Go architecture to filter by.
+    :param registry_config: The path to the registry config file.
+    :param registry_username: The username to authenticate with the registry.
+    :param registry_password: The password to authenticate with the registry.
+    :return: The parsed JSON output of `oc image info`.
+    """
+
+    async def _run_command(auth_file: Optional[str]):
+        options = [f'--filter-by-os={go_arch}', '-o', 'json']
+        if auth_file:
+            options.extend([f'--registry-config={auth_file}'])
+        cmd = ['oc', 'image', 'info'] + options + [pull_spec]
+        _, out, _ = await exectools.cmd_gather_async(cmd)
+        return json.loads(out)
+
+    if not registry_username and not registry_password:
+        return await _run_command(auth_file=registry_config)
+
+    if not registry_password:
+        raise ValueError('registry_password must be provided if auth is needed.')
+    if not registry_username:
+        auth = registry_password
+    else:
+        auth = base64.b64encode(f'{registry_username}:{registry_password}'.encode()).decode()
+    with tempfile.NamedTemporaryFile(mode='w', prefix="_doozer_") as registry_config_file:
+        registry_config_file.write(json.dumps({
+            'auths': {
+                pull_spec.split('/')[0]: {
+                    'auth': auth,
+                }
+            }
+        }))
+        registry_config_file.flush()
+        return await _run_command(auth_file=registry_config_file.name)
 
 
 @alru_cache
-async def oc_image_info__caching_async(pull_spec: str, go_arch: str = 'amd64') -> Dict:
+async def oc_image_info_async__caching(
+        pull_spec: str,
+        go_arch: str = 'amd64',
+        registry_config: Optional[str] = None,
+        registry_username: Optional[str] = None,
+        registry_password: Optional[str] = None) -> Dict:
     """
     Returns a Dict of the parsed JSON output of `oc image info` for the specified
-    pullspec. This function will cache that output per pullspec, so do not use it
+    pullspec. This will authenticate with the registry using the provided username and password.
+
+    This function will cache that output per pullspec, so do not use it
     if you expect the image to change during the course of doozer's execution.
+
+    :param pull_spec: The image pullspec to query.
+    :param go_arch: The Go architecture to filter by.
+    :param registry_config: The path to the registry config file.
+    :param registry_username: The username to authenticate with the registry.
+    :param registry_password: The password to authenticate with the registry.
+    :return: The parsed JSON output of `oc image info`.
     """
-    return await oc_image_info_async(pull_spec, go_arch)
+    return await oc_image_info_async(pull_spec, go_arch, registry_config, registry_username, registry_password)
 
 
 def infer_assembly_type(custom, assembly_name):

doozer/tests/backend/test_konflux_olm_bundler.py

@@ -67,7 +67,7 @@ def test_get_image_reference_pattern(self):
         self.assertEqual(match.group(1), "namespace/image")
         self.assertEqual(match.group(2), "tag")
 
-    @patch("doozerlib.util.oc_image_info__caching_async")
+    @patch("doozerlib.util.oc_image_info_async__caching")
     async def test_replace_image_references(self, mock_oc_image_info):
         old_registry = "registry.example.com"
         content = """