Propagate the tls_verify parameter to the auth backends

Signed-off-by: Jonathan GAYVALLET <jonathan.gayvallet@orange.com>

Author: Meallia

oras/auth/__init__.py

@@ -14,11 +14,14 @@ class AuthenticationException(Exception):
     pass
 
 
-def get_auth_backend(name="token", session=None, insecure=False, **kwargs):
+def get_auth_backend(
+    name="token", session=None, insecure=False, tls_verify=True, **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

@@ -5,6 +5,8 @@
 
 from typing import Optional
 
+import requests
+
 import oras.auth.utils as auth_utils
 import oras.container
 import oras.decorator as decorator
@@ -17,6 +19,9 @@ class AuthBackend:
     Generic (and default) auth backend.
     """
 
+    session: requests.Session
+    _tls_verify: bool
+
     def __init__(self, *args, **kwargs):
         self._auths: dict = {}
         self.prefix: str = "https"

oras/auth/token.py

@@ -127,7 +127,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)  # type: ignore
+        authResponse = self.session.get(h.realm, headers=headers, params=params, verify=self._tls_verify)  # type: ignore
 
         if authResponse.status_code != 200:
             logger.debug(f"Auth response was not successful: {authResponse.text}")
@@ -154,7 +154,9 @@ 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)
+        response = self.session.request(
+            "GET", h.realm, params=params, verify=self._tls_verify
+        )
         if response.status_code != 200:
             logger.debug(f"Response for anon token failed: {response.text}")
             return

oras/decorator.py

@@ -5,6 +5,8 @@
 import time
 from functools import wraps
 
+import requests.exceptions
+
 import oras.auth
 from oras.logger import logger
 
@@ -50,6 +52,8 @@ def inner(*args, **kwargs):
                     return res
                 except oras.auth.AuthenticationException as e:
                     raise e
+                except requests.exceptions.SSLError:
+                    raise
                 except Exception as e:
                     sleep = timeout + 3**attempt
                     logger.info(f"Retrying in {sleep} seconds - error: {e}")

oras/provider.py

@@ -78,7 +78,9 @@ def __init__(
         self.session.cookies.set_policy(DefaultCookiePolicy(allowed_domains=[]))
 
         # Get custom backend, pass on session to share
-        self.auth = oras.auth.get_auth_backend(auth_backend, self.session, insecure)
+        self.auth = oras.auth.get_auth_backend(
+            auth_backend, self.session, insecure, tls_verify=tls_verify
+        )
 
     def __repr__(self) -> str:
         return str(self)

oras/tests/snakeoil.crt

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDATCCAemgAwIBAgIUVD3r8T8WSZDh9kfPNus7RgWf+mIwDQYJKoZIhvcNAQEL
+BQAwFzEVMBMGA1UEAwwMeWQtb3I2MjkyNDkyMB4XDTI1MDEwNzE5MzgyN1oXDTM1
+MDEwNTE5MzgyN1owFzEVMBMGA1UEAwwMeWQtb3I2MjkyNDkyMIIBIjANBgkqhkiG
+9w0BAQEFAAOCAQ8AMIIBCgKCAQEAoiQ4lEoFDmoCFCLWH9gRyebWlnv15OPdykwV
+N2P7nLf+fFFOyQgE59jc2gVw9TeOVmnXxncDOYQrLKikHko8vGKg+daT1DLIufjz
+ijOtdUz9Zgd0rt5Ar7ocnjxkOYJRSW9cBd51O6jpUGN3r1BziQMdvIywDSw+BX2F
+IMLvg2tlHuMb1dR0GHemwqwwdzAsw9QJB3MbGSfVsH4/QQxRsQhpq/5D3WyZJPlJ
+/tYmJhUPVVF94ZLAhqfDueWj+6LaiDUVLQ0CUml8S33Q9570e8JvAYYHwveRIkw/
+W7lVoKu0OnkCKrKWjqgweBfszVz6ARTd8QPCCZO6At99RLjZBwIDAQABo0UwQzAJ
+BgNVHRMEAjAAMBcGA1UdEQQQMA6CDHlkLW9yNjI5MjQ5MjAdBgNVHQ4EFgQUSNhV
+pFANLtcSQsnuT4qrZ7TdADUwDQYJKoZIhvcNAQELBQADggEBABNaYMotGee4GT4J
+6lw6eRq/Zzy6xN1n0iSJW4QJvPkbOoFa+CBRtQ/Vk5yiKOB3yX7SOdilm5yzQbBd
+QKsYgEhkPi8F/LN3UYDgE69soIBJv9Fx/EcmbsqLZqz60xSuwtVcqtJ9MuewZhZm
+5ny2zbCNQxK6JehscFKVeR/W088jfWmtp4kIQIdK9h4301dAeNu8Si35HGtQupIw
++jliOQtOFu3nikbBH+k8wwi4/2tMtEAsKbr83ixfkjkkivT7B5A9NNHNpNTzYw/T
+HGQaoWWIMf6Zcea33lM087raIC83qU/CuJYiNVvUkw7CnV2P/yg/FpCxsP0AudGC
+4FLB77A=
+-----END CERTIFICATE-----

oras/tests/test_oras.py

@@ -4,8 +4,12 @@
 
 import os
 import shutil
+import tempfile
+import time
+from pathlib import Path
 
 import pytest
+from requests.exceptions import SSLError
 
 import oras.client
 
@@ -227,3 +231,49 @@ def test_custom_docker_config_path(tmp_path, registry, credentials, target_dir):
     assert "artifact.txt" in os.listdir(files[0])
 
     client.logout(registry)
+
+
+@pytest.fixture
+def empty_request_ca():
+    old_ca = os.environ.get("REQUESTS_CA_BUNDLE", None)
+    try:
+        # we're setting a fake CA since an empty one won't work
+        os.environ["REQUESTS_CA_BUNDLE"] = str(Path(__file__).parent / "snakeoil.crt")
+        yield
+    finally:
+        if old_ca is not None:
+            os.environ["REQUESTS_CA_BUNDLE"] = old_ca
+        else:
+            del os.environ["REQUESTS_CA_BUNDLE"]
+
+
+def test_ssl_no_verify(empty_request_ca):
+    """
+    Make sure the client works without a CA file and tls_verify set to False
+    """
+    client = oras.client.OrasClient(
+        hostname="ghcr.io", insecure=False, tls_verify=False
+    )
+    client.get_tags("channel-mirrors/conda-forge/linux-aarch64/arrow-cpp", N=1)
+
+
+def test_ssl_verify_fails_if_bad_ca(empty_request_ca):
+    """
+    Make sure the client fails without a CA file and tls_verify set to True
+    """
+    client = oras.client.OrasClient(hostname="ghcr.io", insecure=False, tls_verify=True)
+
+    with pytest.raises(SSLError):
+        client.get_tags("channel-mirrors/conda-forge/linux-aarch64/arrow-cpp", N=1)
+
+
+def test_ssl_verify_fails_fast_if_bad_ca(empty_request_ca):
+    """
+    The client should fail fast in case of SSL errors
+    """
+    client = oras.client.OrasClient(hostname="ghcr.io", insecure=False, tls_verify=True)
+    st = time.monotonic()
+    with pytest.raises(SSLError):
+        client.get_tags("channel-mirrors/conda-forge/linux-aarch64/arrow-cpp", N=1)
+    et = time.monotonic()
+    assert et - st < 5

oras/version.py

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