Make pyright happy

oras/auth/__init__.py

@@ -1,17 +1,29 @@
+from typing import Dict, Literal, Optional
+
 import requests
 
+from oras.auth.base import AuthBackend
 from oras.logger import logger
 
 from .basic import BasicAuth
 from .token import TokenAuth
 
-auth_backends = {"token": TokenAuth, "basic": BasicAuth}
+AuthBackendName = Literal["token", "basic"]
+
+auth_backends: Dict[AuthBackendName, type[AuthBackend]] = {
+    "token": TokenAuth,
+    "basic": BasicAuth,
+}
 
 
-def get_auth_backend(name="token", session=None, **kwargs):
+def get_auth_backend(
+    name: AuthBackendName = "token",
+    session: Optional[requests.Session] = None,
+    **kwargs,
+):
     backend = auth_backends.get(name)
     if not backend:
-        logger.exit(f"Authentication backend {backend} is not known.")
-    backend = backend(**kwargs)
-    backend.session = session or requests.Session()
+        return logger.exit(f"Authentication backend {backend} is not known.")
+    _session = session or requests.Session()
+    backend = backend(session=_session, **kwargs)
     return backend

oras/auth/base.py

@@ -2,37 +2,49 @@
 __copyright__ = "Copyright The ORAS Authors."
 __license__ = "Apache-2.0"
 
+import abc
+from typing import Dict, Optional, Tuple
 
-from typing import Optional
+import requests
 
 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
 
 
-class AuthBackend:
+class AuthBackend(abc.ABC):
     """
     Generic (and default) auth backend.
     """
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, session: requests.Session):
         self._auths: dict = {}
+        self.session = session
+        self.headers: Dict[str, str] = {}
+
+    @abc.abstractmethod
+    def authenticate_request(
+        self,
+        original: requests.Response,
+        headers: Optional[dict[str, str]] = None,
+        refresh=False,
+    ) -> Tuple[dict[str, str], bool]:
+        pass
 
-    def get_auth_header(self):
-        raise NotImplementedError
-
-    def get_container(self, name: container_type) -> oras.container.Container:
+    def set_header(self, name: str, value: str):
         """
-        Courtesy function to get a container from a URI.
+        Courtesy function to set a header
 
-        :param name: unique resource identifier to parse
-        :type name: oras.container.Container or str
+        :param name: header name to set
+        :type name: str
+        :param value: header value to set
+        :type value: str
         """
-        if isinstance(name, oras.container.Container):
-            return name
-        return oras.container.Container(name, registry=self.hostname)
+        self.headers.update({name: value})
+
+    def get_auth_header(self):
+        raise NotImplementedError
 
     def logout(self, hostname: str):
         """
@@ -81,8 +93,9 @@ def _load_auth(self, hostname: str) -> bool:
             return True
         return False
 
-    @decorator.ensure_container
-    def load_configs(self, container: container_type, configs: Optional[list] = None):
+    def load_configs(
+        self, container: oras.container.Container, configs: Optional[list] = None
+    ):
         """
         Load configs to discover credentials for a specific container.
 
@@ -96,7 +109,7 @@ def load_configs(self, container: container_type, configs: Optional[list] = None
         """
         if not self._auths:
             self._auths = auth_utils.load_configs(configs)
-        for registry in oras.utils.iter_localhosts(container.registry):  # type: ignore
+        for registry in oras.utils.iter_localhosts(container.registry):
             if self._load_auth(registry):
                 return
 
@@ -120,37 +133,3 @@ def set_basic_auth(self, username: str, password: str):
         :type password: str
         """
         self._basic_auth = auth_utils.get_basic_auth(username, password)
-
-    def request_anonymous_token(self, h: auth_utils.authHeader, headers: dict) -> bool:
-        """
-        Given no basic auth, fall back to trying to request an anonymous token.
-
-        Returns: boolean if headers have been updated with token.
-        """
-        if not h.realm:
-            logger.debug("Request anonymous token: no realm provided, exiting early")
-            return headers, False
-
-        params = {}
-        if h.service:
-            params["service"] = h.service
-        if h.scope:
-            params["scope"] = h.scope
-
-        logger.debug(f"Final params are {params}")
-        response = self.session.request("GET", h.realm, params=params)
-        if response.status_code != 200:
-            logger.debug(f"Response for anon token failed: {response.text}")
-            return headers, False
-
-        # From https://docs.docker.com/registry/spec/auth/token/ section
-        # We can get token OR access_token OR both (when both they are identical)
-        data = response.json()
-        token = data.get("token") or data.get("access_token")
-
-        # Update the headers but not self.token (expects Basic)
-        if token:
-            headers["Authorization"] = {"Authorization": "Bearer %s" % token}
-
-        logger.debug("Warning: no token or access_token present in response.")
-        return headers, False

oras/auth/basic.py

@@ -3,6 +3,7 @@
 __license__ = "Apache-2.0"
 
 import os
+from typing import Optional, Tuple
 
 import requests
 
@@ -14,10 +15,10 @@ class BasicAuth(AuthBackend):
     Generic (and default) auth backend.
     """
 
-    def __init__(self):
+    def __init__(self, session: requests.Session):
         username = os.environ.get("ORAS_USER")
         password = os.environ.get("ORAS_PASS")
-        super().__init__()
+        super().__init__(session=session)
         if username and password:
             self.set_basic_auth(username, password)
 
@@ -28,8 +29,11 @@ def get_auth_header(self):
         return {"Authorization": "Basic %s" % self._basic_auth}
 
     def authenticate_request(
-        self, original: requests.Response, headers: dict, refresh=False
-    ):
+        self,
+        original: requests.Response,
+        headers: Optional[dict[str, str]] = None,
+        refresh=False,
+    ) -> Tuple[dict[str, str], bool]:
         """
         Authenticate Request
         Given a response, look for a Www-Authenticate header to parse.

oras/auth/token.py

@@ -2,6 +2,8 @@
 __copyright__ = "Copyright The ORAS Authors."
 __license__ = "Apache-2.0"
 
+from typing import Optional, Tuple
+
 import requests
 
 import oras.auth.utils as auth_utils
@@ -15,9 +17,9 @@ class TokenAuth(AuthBackend):
     Token (OAuth2) style auth.
     """
 
-    def __init__(self):
+    def __init__(self, session: requests.Session):
         self.token = None
-        super().__init__()
+        super().__init__(session=session)
 
     def _logout(self):
         self.token = None
@@ -44,8 +46,11 @@ def reset_basic_auth(self):
             self.set_header("Authorization", "Basic %s" % self._basic_auth)
 
     def authenticate_request(
-        self, original: requests.Response, headers: dict, refresh=False
-    ):
+        self,
+        original: requests.Response,
+        headers: Optional[dict[str, str]] = None,
+        refresh=False,
+    ) -> Tuple[dict[str, str], bool]:
         """
         Authenticate Request
         Given a response, look for a Www-Authenticate header to parse.
@@ -55,6 +60,8 @@ def authenticate_request(
         :param original: original response to get the Www-Authenticate header
         :type original: requests.Response
         """
+        _headers = headers or {}
+
         if refresh:
             self.token = None
 
@@ -63,12 +70,12 @@ def authenticate_request(
             logger.debug(
                 "Www-Authenticate not found in original response, cannot authenticate."
             )
-            return headers, False
+            return _headers, False
 
         # If we have a token, set auth header (base64 encoded user/pass)
         if self.token:
-            headers["Authorization"] = "Bearer %s" % self.token
-            return headers, True
+            _headers["Authorization"] = "Bearer %s" % self.token
+            return _headers, True
 
         h = auth_utils.parse_auth_header(authHeaderRaw)
 
@@ -78,23 +85,23 @@ def authenticate_request(
         if anon_token:
             logger.debug("Successfully obtained anonymous token!")
             self.token = anon_token
-            headers["Authorization"] = "Bearer %s" % self.token
-            return headers, True
+            _headers["Authorization"] = "Bearer %s" % self.token
+            return _headers, True
 
         # Next try for logged in token
         token = self.request_token(h)
         if token:
             self.token = token
-            headers["Authorization"] = "Bearer %s" % self.token
-            return headers, True
+            _headers["Authorization"] = "Bearer %s" % self.token
+            return _headers, True
 
         logger.error(
             "This endpoint requires a token. Please use "
             "basic auth with a username or password."
         )
-        return headers, False
+        return _headers, False
 
-    def request_token(self, h: auth_utils.authHeader) -> bool:
+    def request_token(self, h: auth_utils.authHeader):
         """
         Request an authenticated token and save for later.s
         """
@@ -113,16 +120,18 @@ def request_token(self, h: auth_utils.authHeader) -> bool:
                 }
             )
 
+        assert h.realm is not None, "realm must be defined"
+
         # Ensure the realm starts with http
-        if not h.realm.startswith("http"):  # type: ignore
-            h.realm = f"{self.prefix}://{h.realm}"
+        if not h.realm.startswith("http"):
+            h.realm = f"http://{h.realm}"  # TODO: Should this be htts
 
         # If the www-authenticate included a scope, honor it!
         if h.scope:
             logger.debug(f"Scope: {h.scope}")
             params["scope"] = h.scope
 
-        authResponse = self.session.get(h.realm, headers=headers, params=params)  # type: ignore
+        authResponse = self.session.get(h.realm, headers=headers, params=params)
         if authResponse.status_code != 200:
             logger.debug(f"Auth response was not successful: {authResponse.text}")
             return
@@ -131,11 +140,9 @@ def request_token(self, h: auth_utils.authHeader) -> bool:
         info = authResponse.json()
         return info.get("token") or info.get("access_token")
 
-    def request_anonymous_token(self, h: auth_utils.authHeader) -> bool:
+    def request_anonymous_token(self, h: auth_utils.authHeader) -> Optional[str]:
         """
         Given no basic auth, fall back to trying to request an anonymous token.
-
-        Returns: boolean if headers have been updated with token.
         """
         if not h.realm:
             logger.debug("Request anonymous token: no realm provided, exiting early")

oras/auth/utils.py

@@ -18,17 +18,16 @@ def load_configs(configs: Optional[List[str]] = None):
     :param configs: list of configuration paths to load, defaults to None
     :type configs: optional list
     """
-    configs = configs or []
+    _configs = configs or []
     default_config = oras.utils.find_docker_config()
 
     # Add the default docker config
     if default_config:
-        configs.append(default_config)
-    configs = set(configs)  # type: ignore
+        _configs.append(default_config)
 
     # Load configs until we find our registry hostname
     auths = {}
-    for config in configs:
+    for config in set(_configs):
         if not os.path.exists(config):
             logger.warning(f"{config} does not exist.")
             continue

oras/container.py

@@ -108,7 +108,7 @@ def parse(self, name: str):
             raise ValueError(
                 f"{name} does not match a recognized registry unique resource identifier. Try <registry>/<namespace>/<repository>:<tag|digest>"
             )
-        items = match.groupdict()  # type: ignore
+        items = match.groupdict()
         self.repository = items["repository"]
         self.registry = items["registry"] or self.registry
         self.namespace = items["namespace"]