Use oras registry_config and registry url parse to finalize login

Author: yma955

charon/config.py

@@ -35,6 +35,9 @@ def __init__(self, data: Dict):
         self.__client_key: str = data.get("client_key", None)
         self.__client_key_pass_file: str = data.get("client_key_pass_file", None)
         self.__root_ca: str = data.get("root_ca", "/etc/pki/tls/certs/ca-bundle.crt")
+        self.__quay_radas_registry_config: str = data.get(
+            "quay_radas_registry_config", os.path.join(os.getenv("HOME", ""), ".oras/config.json")
+        )
         self.__radas_sign_timeout_retry_count: int = data.get("radas_sign_timeout_retry_count", 10)
         self.__radas_sign_timeout_retry_interval: int = data.get(
             "radas_sign_timeout_retry_interval", 60
@@ -62,6 +65,11 @@ def validate(self) -> bool:
         if self.__root_ca and not os.access(self.__root_ca, os.R_OK):
             logger.error("The root ca file is not valid!")
             return False
+        if self.__quay_radas_registry_config and not os.access(
+            self.__quay_radas_registry_config, os.R_OK
+        ):
+            logger.error("The quay registry config for oras is not valid!")
+            return False
         return True
 
     def umb_target(self) -> str:
@@ -91,6 +99,9 @@ def client_key_password(self) -> str:
     def root_ca(self) -> str:
         return self.__root_ca
 
+    def quay_radas_registry_config(self) -> str:
+        return self.__quay_radas_registry_config
+
     def radas_sign_timeout_retry_count(self) -> int:
         return self.__radas_sign_timeout_retry_count
 

charon/pkgs/oras_client.py

@@ -18,6 +18,7 @@
 import logging
 from charon.config import get_config
 from typing import List
+from urllib.parse import urlparse
 
 logger = logging.getLogger(__name__)
 
@@ -31,21 +32,24 @@ def __init__(self):
         self.conf = get_config()
         self.client = oras.client.OrasClient()
 
-    def login_if_needed(self) -> None:
+    def login_if_needed(self, registry: str) -> None:
         """
-        If quay_radas_auth_enabled is true, call login to authenticate.
+        If quay_radas_registry_config is provided, call login to authenticate.
         """
+        if not registry.startswith("http://") and not registry.startswith("https://"):
+            registry = "https://" + registry
+        registry = urlparse(registry).netloc
 
-        if self.conf and self.conf.is_quay_radas_auth_enabled():
-            logger.info("Logging in to registry.")
+        rconf = self.conf.get_radas_config() if self.conf else None
+        if rconf and rconf.quay_radas_registry_config():
+            logger.info("Logging in to registry: %s", registry)
             res = self.client.login(
-                hostname=self.conf.get_quay_radas_registry(),
-                username=self.conf.get_quay_radas_username(),
-                password=self.conf.get_quay_radas_password(),
+                hostname=registry,
+                config_path=rconf.quay_radas_registry_config(),
             )
             logger.info(res)
         else:
-            logger.info("Registry auth not enabled, skip login.")
+            logger.info("Registry config is not provided, skip login.")
 
     def pull(self, result_reference_url: str, sign_result_loc: str) -> List[str]:
         """
@@ -58,7 +62,7 @@ def pull(self, result_reference_url: str, sign_result_loc: str) -> List[str]:
         """
         files = []
         try:
-            self.login_if_needed()
+            self.login_if_needed(registry=result_reference_url)
             files = self.client.pull(target=result_reference_url, outdir=sign_result_loc)
             logger.info("Pull file from %s to %s", result_reference_url, sign_result_loc)
         except Exception as e: