Add TLS Validation

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)
+ - allow specification of server and client certificates (0.3.0)
  - initialize headers variable in do_request (0.2.31)
  - Make reproducible targz without mimetype (0.2.30)
  - don't include Content-Length header in upload_manifest (0.2.29)

oras/auth/__init__.py

@@ -14,14 +14,11 @@ class AuthenticationException(Exception):
     pass
 
 
-def get_auth_backend(
-    name="token", session=None, insecure=False, tls_verify=True, **kwargs
-):
+def get_auth_backend(name="token", session=None, insecure=False, **kwargs):
     backend = auth_backends.get(name)
     if not backend:
         raise ValueError(f"Authentication backend {backend} is not known.")
     backend = backend(**kwargs)
     backend.session = session or requests.Session()
     backend.prefix = "http" if insecure else "https"
-    backend._tls_verify = tls_verify
     return backend

oras/auth/base.py

@@ -20,7 +20,6 @@ class AuthBackend:
     """
 
     session: requests.Session
-    _tls_verify: bool
 
     def __init__(self, *args, **kwargs):
         self._auths: dict = {}

oras/auth/token.py

@@ -128,7 +128,7 @@ def request_token(self, h: auth_utils.authHeader) -> bool:
         headers["Authorization"] = "Basic %s" % self._basic_auth
 
         logger.debug(f"Requesting auth token for: {h}")
-        authResponse = self.session.get(h.realm, headers=headers, params=params, verify=self._tls_verify)  # type: ignore
+        authResponse = self.session.get(h.realm, headers=headers, params=params)  # type: ignore
 
         if authResponse.status_code != 200:
             logger.debug(f"Auth response was not successful: {authResponse.text}")
@@ -155,9 +155,7 @@ def request_anonymous_token(self, h: auth_utils.authHeader) -> bool:
             params["scope"] = h.scope
 
         logger.debug(f"Requesting anon token with params: {params}")
-        response = self.session.request(
-            "GET", h.realm, params=params, verify=self._tls_verify
-        )
+        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

oras/provider.py

@@ -50,6 +50,9 @@ def __init__(
         insecure: bool = False,
         tls_verify: bool = True,
         auth_backend: str = "token",
+        tls_cert: Optional[str] = None,
+        tls_key: Optional[str] = None,
+        ca_bundle: Optional[str] = None,
     ):
         """
         Create an ORAS client.
@@ -58,16 +61,23 @@ def __init__(
 
         :param hostname: the hostname of the registry to ping
         :type hostname: str
-        :param registry: if provided, use this custom provider instead of default
-        :type registry: oras.provider.Registry or None
         :param insecure: use http instead of https
         :type insecure: bool
+        :param tls_verify: perform tls verification
+        :type tls_verify: bool
+        :param auth_backend: name of the auth backend to use
+        :type auth_backend: str
+        :param tls_cert: path to the client certificate to use
+        :type tls_cert: str
+        :param tls_key: path to the client key to use
+        :type tls_key: str
+        :param ca_bundle: path to the CA bundle to use (enables tls verification)
+        :type ca_bundle: str
         """
         self.hostname: Optional[str] = hostname
         self.headers: dict = {}
         self.session: requests.Session = requests.Session()
         self.prefix: str = "http" if insecure else "https"
-        self._tls_verify = tls_verify
 
         if not tls_verify:
             requests.packages.urllib3.disable_warnings()  # type: ignore
@@ -77,10 +87,14 @@ def __init__(
         # trying to set further CSRF cookies (Harbor is such a case)
         self.session.cookies.set_policy(DefaultCookiePolicy(allowed_domains=[]))
 
+        self.session.verify = ca_bundle or tls_verify
+        if tls_cert and tls_key:
+            self.session.cert = (tls_cert, tls_key)
+        elif tls_cert:
+            self.session.cert = tls_cert
+
         # Get custom backend, pass on session to share
-        self.auth = oras.auth.get_auth_backend(
-            auth_backend, self.session, insecure, tls_verify=tls_verify
-        )
+        self.auth = oras.auth.get_auth_backend(auth_backend, self.session, insecure)
 
     def __repr__(self) -> str:
         return str(self)
@@ -994,7 +1008,6 @@ def do_request(
             json=json,
             headers=headers,
             stream=stream,
-            verify=self._tls_verify,
         )
 
         # A 401 response is a request for authentication, 404 is not found
@@ -1012,7 +1025,6 @@ def do_request(
             json=json,
             headers=headers,
             stream=stream,
-            verify=self._tls_verify,
         )
 
         # One retry if 403 denied (need new token?)
@@ -1027,7 +1039,6 @@ def do_request(
                 json=json,
                 headers=headers,
                 stream=stream,
-                verify=self._tls_verify,
             )
 
         return response

oras/version.py

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