Load auth configs before making requests to container registry

CHANGELOG.md

@@ -14,6 +14,7 @@ and **Merged pull requests**. Critical items to know are:
 The versions coincide with releases on pip. Only major versions will be released as tags on Github.
 
 ## [0.0.x](https://github.com/oras-project/oras-py/tree/main) (0.0.x)
+ - Add authentication config loading before making requests to container registry (0.2.34)
  - fix 'get_manifest()' method with adding 'load_configs()' calling (0.2.33)
  - fix 'Provider' method signature to allow custom CA-Bundles (0.2.32)
  - initialize headers variable in do_request (0.2.31)

oras/auth/base.py

@@ -10,6 +10,7 @@
 import oras.auth.utils as auth_utils
 import oras.container
 import oras.decorator as decorator
+import oras.utils
 from oras.logger import logger
 from oras.types import container_type
 
@@ -87,7 +88,7 @@ def _load_auth(self, hostname: str) -> bool:
             return True
         return False
 
-    @decorator.ensure_container
+    @decorator.ensure_container()
     def load_configs(self, container: container_type, configs: Optional[list] = None):
         """
         Load configs to discover credentials for a specific container.
@@ -106,6 +107,26 @@ def load_configs(self, container: container_type, configs: Optional[list] = None
             if self._load_auth(registry):
                 return
 
+    def ensure_auth_for_container(self, container: container_type):
+        """
+        Ensure authentication is loaded for a specific container's registry.
+        This assumes auths have already been loaded via load_configs or __init__.
+
+        :param container: the parsed container URI with components
+        :type container: oras.container.Container
+        """
+        # At this point, container should already be a Container object
+        # since the decorators handle conversion
+        if not isinstance(container, oras.container.Container):
+            raise ValueError(
+                "Container must be a Container object when ensure_auth_for_container is called"
+            )
+
+        # Try to load auth for this container's registry
+        for registry in oras.utils.iter_localhosts(container.registry):  # type: ignore
+            if self._load_auth(registry):
+                return
+
     def set_token_auth(self, token: str):
         """
         Set token authentication.

oras/decorator.py

@@ -11,21 +11,73 @@
 from oras.logger import logger
 
 
-def ensure_container(func):
+def ensure_container(container_arg_index=0):
     """
-    Ensure the first argument is a container, and not a string.
+    Ensure the specified argument is a container, and not a string.
+
+    :param container_arg_index: The index of the container argument (0 for first arg, 1 for second arg)
+    :type container_arg_index: int
+
+    Usage examples:
+        @ensure_container()     # Container is first argument (default)
+        @ensure_container(0)    # Container is first argument (explicit)
+        @ensure_container(1)    # Container is second argument
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(cls, *args, **kwargs):
+            if "container" in kwargs:
+                kwargs["container"] = cls.get_container(kwargs["container"])
+            elif len(args) > container_arg_index:
+                # Convert the specified argument to a container
+                container = cls.get_container(args[container_arg_index])
+                # Rebuild args tuple with the converted container
+                args = (
+                    args[:container_arg_index]
+                    + (container,)
+                    + args[container_arg_index + 1 :]
+                )
+            return func(cls, *args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
+def ensure_auth(container_arg_index=0):
     """
+    Ensure authentication is loaded for the container's registry.
+    This decorator should be applied after @ensure_container.
+
+    :param container_arg_index: The index of the container argument (0 for first arg, 1 for second arg)
+    :type container_arg_index: int
+
+    Usage examples:
+        @ensure_auth()          # Container is first argument (default)
+        @ensure_auth(0)         # Container is first argument (explicit)
+        @ensure_auth(1)         # Container is second argument
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(cls, *args, **kwargs):
+            # Get the container from the specified argument position
+            container = None
+            if "container" in kwargs:
+                container = kwargs["container"]
+            elif len(args) > container_arg_index:
+                container = args[container_arg_index]
+
+            if container and hasattr(cls, "auth"):
+                # Load auth for this specific container's registry without reloading configs
+                cls.auth.ensure_auth_for_container(container)
 
-    @wraps(func)
-    def wrapper(cls, *args, **kwargs):
-        if "container" in kwargs:
-            kwargs["container"] = cls.get_container(kwargs["container"])
-        elif args:
-            container = cls.get_container(args[0])
-            args = (container, *args[1:])
-        return func(cls, *args, **kwargs)
+            return func(cls, *args, **kwargs)
 
-    return wrapper
+        return wrapper
+
+    return decorator
 
 
 def retry(attempts=5, timeout=2):

oras/provider.py

@@ -16,6 +16,7 @@
 import requests
 
 import oras.auth
+import oras.auth.utils
 import oras.container
 import oras.decorator as decorator
 import oras.defaults
@@ -84,6 +85,10 @@ def __init__(
             auth_backend, self.session, insecure, tls_verify=tls_verify
         )
 
+        # Load all authentication configs once during initialization
+        # This avoids re-reading the docker config file for each operation
+        self.auth._auths = oras.auth.utils.load_configs()
+
     def __repr__(self) -> str:
         return str(self)
 
@@ -249,6 +254,8 @@ def _parse_manifest_ref(self, ref: str) -> Tuple[str, str]:
             path_content.content = oras.defaults.unknown_config_media_type
         return path_content.path, path_content.content
 
+    @decorator.ensure_container(1)
+    @decorator.ensure_auth(1)
     def upload_blob(
         self,
         blob: str,
@@ -276,7 +283,6 @@ def upload_blob(
         :type chunk_size: int
         """
         blob = os.path.abspath(blob)
-        container = self.get_container(container)
 
         if self.blob_exists(layer, container):
             logger.debug(f'layer already exists: {layer["digest"]}')
@@ -306,7 +312,8 @@ def upload_blob(
             response.status_code = 200
         return response
 
-    @decorator.ensure_container
+    @decorator.ensure_container()
+    @decorator.ensure_auth()
     def delete_tag(self, container: container_type, tag: str) -> bool:
         """
         Delete a tag for a container.
@@ -340,7 +347,8 @@ def delete_tag(self, container: container_type, tag: str) -> bool:
             raise RuntimeError("Delete was not successful: {response.json()}")
         return True
 
-    @decorator.ensure_container
+    @decorator.ensure_container()
+    @decorator.ensure_auth()
     def get_tags(self, container: container_type, N=None) -> List[str]:
         """
         Retrieve tags for a package.
@@ -404,7 +412,8 @@ def _do_paginated_request(
             # use link + base url to continue with next page
             url = urllib.parse.urljoin(base_url, link)
 
-    @decorator.ensure_container
+    @decorator.ensure_container()
+    @decorator.ensure_auth()
     def get_blob(
         self,
         container: container_type,
@@ -440,7 +449,8 @@ def get_container(self, name: container_type) -> oras.container.Container:
         return oras.container.Container(name, registry=self.hostname)
 
     # Functions to be deprecated in favor of exposed ones
-    @decorator.ensure_container
+    @decorator.ensure_container()
+    @decorator.ensure_auth()
     def _download_blob(
         self, container: container_type, digest: str, outfile: str
     ) -> str:
@@ -485,7 +495,8 @@ def _upload_blob(
         )
         return self.upload_blob(blob, container, layer, do_chunked)
 
-    @decorator.ensure_container
+    @decorator.ensure_container()
+    @decorator.ensure_auth()
     def download_blob(
         self, container: container_type, digest: str, outfile: str
     ) -> str:
@@ -516,6 +527,8 @@ def download_blob(
             raise e
         return outfile
 
+    @decorator.ensure_container(1)
+    @decorator.ensure_auth(1)
     def put_upload(
         self,
         blob: str,
@@ -561,6 +574,8 @@ def put_upload(
             )
         return response
 
+    @decorator.ensure_container(1)
+    @decorator.ensure_auth(1)
     def blob_exists(self, layer: dict, container: oras.container.Container) -> bool:
         """
         Check if a layer already exists in the registry.
@@ -600,6 +615,8 @@ def _get_location(
             session_url = f"{prefix}{session_url}"
         return session_url
 
+    @decorator.ensure_container(1)
+    @decorator.ensure_auth(1)
     def chunked_upload(
         self,
         blob: str,
@@ -688,6 +705,8 @@ def _parse_response_errors(self, response: requests.Response):
         except Exception:
             pass
 
+    @decorator.ensure_container(1)
+    @decorator.ensure_auth(1)
     def upload_manifest(
         self,
         manifest: dict,
@@ -751,9 +770,13 @@ def push(
         """
         container = self.get_container(target)
         files = files or []
-        self.auth.load_configs(
-            container, configs=[config_path] if config_path else None
-        )
+
+        # If a custom config path is provided, load those configs
+        if config_path:
+            self.auth.load_configs(container, configs=[config_path])
+        else:
+            # Use the already loaded auths with ensure_auth pattern
+            self.auth.ensure_auth_for_container(container)
 
         # Prepare a new manifest
         manifest = oras.oci.NewManifest()
@@ -891,9 +914,13 @@ def pull(
         :type target: str
         """
         container = self.get_container(target)
-        self.auth.load_configs(
-            container, configs=[config_path] if config_path else None
-        )
+
+        # If a custom config path is provided, load those configs
+        if config_path:
+            self.auth.load_configs(container, configs=[config_path])
+        else:
+            # Use the already loaded auths with ensure_auth pattern
+            self.auth.ensure_auth_for_container(container)
         manifest = self.get_manifest(container, allowed_media_type)
         outdir = outdir or oras.utils.get_tmpdir()
         overwrite = overwrite
@@ -932,7 +959,8 @@ def pull(
             files.append(outfile)
         return files
 
-    @decorator.ensure_container
+    @decorator.ensure_container()
+    @decorator.ensure_auth()
     def get_manifest(
         self,
         container: container_type,
@@ -946,10 +974,6 @@ def get_manifest(
         :param allowed_media_type: one or more allowed media types
         :type allowed_media_type: str
         """
-        # Load authentication configs for the container's registry
-        # This ensures credentials are available for authenticated registries
-        self.auth.load_configs(container)
-
         if not allowed_media_type:
             allowed_media_type = [oras.defaults.default_manifest_media_type]
         headers = {"Accept": ";".join(allowed_media_type)}

oras/version.py

@@ -2,7 +2,7 @@
 __copyright__ = "Copyright The ORAS Authors."
 __license__ = "Apache-2.0"
 
-__version__ = "0.2.33"
+__version__ = "0.2.34"
 AUTHOR = "Vanessa Sochat"
 EMAIL = "vsoch@users.noreply.github.com"
 NAME = "oras"