SSL

Author: julien-lang

docs/advanced/iron_python.rst

@@ -22,9 +22,7 @@ v3.0.20 can be used with IronPython with a little bit of added work:
 
 - If you encounter any SSL errors like
   ``unknown field: SERIALNUMBER=0123456789`` or ``:SSL3_GET_SERVER_CERTIFICATE:certificate verify failed``.
-  For now you can workaround this problem by disabling ssl certificate validation which we've
-  encountered some intermittent issues with. Set ``NO_SSL_VALIDATION = True`` for either case.
-  See :const:`shotgun_api3.shotgun.NO_SSL_VALIDATION`
+  Inject you own CA cert <<TODO>>
 
 - If you encounter ``LookupError: unknown encoding: idna``, you can force utf-8 by changing
   iri2uri.py ~ln 71 from ``authority = authority.encode('idna')`` to

shotgun_api3/shotgun.py

@@ -77,14 +77,6 @@
 
 SHOTGUN_API_DISABLE_ENTITY_OPTIMIZATION = False
 
-NO_SSL_VALIDATION = False
-"""
-Turns off hostname matching validation for SSL certificates
-
-Sometimes there are cases where certificate validation should be disabled. For example, if you
-have a self-signed internal certificate that isn't included in our certificate bundle, you may
-not require the added security provided by enforcing this.
-"""
 
 # ----------------------------------------------------------------------------
 # Version
@@ -350,12 +342,11 @@ def __init__(self):
 
         self.py_version = ".".join(str(x) for x in sys.version_info[:2])
 
-        # extract the OpenSSL version if we can. The version is only available in Python 2.7 and
-        # only if we successfully imported ssl
+        # extract the OpenSSL version if we can.
         self.ssl_version = "unknown"
         try:
             self.ssl_version = ssl.OPENSSL_VERSION
-        except (AttributeError, NameError):
+        except AttributeError:
             pass
 
     def __str__(self):
@@ -424,7 +415,6 @@ def __init__(self, sg):
         self.proxy_pass = None
         self.session_token = None
         self.authorization = None
-        self.no_ssl_validation = False
         self.localized = False
 
     def set_server_params(self, base_url):
@@ -633,7 +623,6 @@ def __init__(
         self.config.session_token = session_token
         self.config.sudo_as_login = sudo_as_login
         self.config.convert_datetimes_to_utc = convert_datetimes_to_utc
-        self.config.no_ssl_validation = NO_SSL_VALIDATION
         self.config.raw_http_proxy = http_proxy
 
         try:
@@ -2281,14 +2270,10 @@ def reset_user_agent(self):
             ua_platform = self.client_caps.platform.capitalize()
 
         # create ssl validation string based on settings
-        validation_str = "validate"
-        if self.config.no_ssl_validation:
-            validation_str = "no-validate"
-
         self._user_agents = [
-            "shotgun-json (%s)" % __version__,
-            "Python %s (%s)" % (self.client_caps.py_version, ua_platform),
-            "ssl %s (%s)" % (self.client_caps.ssl_version, validation_str),
+            f"shotgun-json ({__version__})",
+            f"Python {self.client_caps.py_version} ({ua_platform})",
+            f"ssl {self.client_caps.ssl_version}",
         ]
 
     def set_session_uuid(self, session_uuid):
@@ -3560,8 +3545,14 @@ def _build_opener(self, handler):
         Build urllib2 opener with appropriate proxy handler.
         """
         handlers = []
-        if self.__ca_certs and not NO_SSL_VALIDATION:
-            handlers.append(CACertsHTTPSHandler(self.__ca_certs))
+        if self.__ca_certs:
+            handlers.append(
+                urllib.request.HTTPSHandler(
+                    context=ssl.create_default_context(
+                        cafile=self.__ca_certs,
+                    ),
+                ),
+            )
 
         if self.config.proxy_handler:
             handlers.append(self.config.proxy_handler)
@@ -3630,23 +3621,6 @@ def _get_certs_file(cls, ca_certs):
             cert_file = os.path.join(cur_dir, "lib", "certifi", "cacert.pem")
             return cert_file
 
-    def _turn_off_ssl_validation(self):
-        """
-        Turn off SSL certificate validation.
-        """
-        global NO_SSL_VALIDATION
-        self.config.no_ssl_validation = True
-        NO_SSL_VALIDATION = True
-        # reset ssl-validation in user-agents
-        self._user_agents = [
-            (
-                "ssl %s (no-validate)" % self.client_caps.ssl_version
-                if ua.startswith("ssl ")
-                else ua
-            )
-            for ua in self._user_agents
-        ]
-
     # Deprecated methods from old wrapper
     def schema(self, entity_type):
         """
@@ -3849,59 +3823,6 @@ def _make_call(self, verb, path, body, headers):
             attempt += 1
             try:
                 return self._http_request(verb, path, body, req_headers)
-            except ssl.SSLEOFError as e:
-                # SG-34910 - EOF occurred in violation of protocol (_ssl.c:2426)
-                # This issue seems to be related to proxy and keep alive.
-                # It looks like, sometimes, the proxy drops the connection on
-                # the TCP/TLS level despites the keep-alive. So we need to close
-                # the connection and make a new attempt.
-                LOG.debug("SSLEOFError: {}".format(e))
-                self._close_connection()
-                if attempt == max_rpc_attempts:
-                    LOG.debug("Request failed.  Giving up after %d attempts." % attempt)
-                    raise
-                # This is the exact same block as the "except Exception" bellow.
-                # We need to do it here because the next except will match it
-                # otherwise and will not re-attempt.
-                # When we drop support of Python 2 and we will probably drop the
-                # next except, we might want to remove this except too.
-            except ssl_error_classes as e:
-                # Test whether the exception is due to the fact that this is an older version of
-                # Python that cannot validate certificates encrypted with SHA-2. If it is, then
-                # fall back on disabling the certificate validation and try again - unless the
-                # SHOTGUN_FORCE_CERTIFICATE_VALIDATION environment variable has been set by the
-                # user. In that case we simply raise the exception. Any other exceptions simply
-                # get raised as well.
-                #
-                # For more info see:
-                # https://www.shotgridsoftware.com/blog/important-ssl-certificate-renewal-and-sha-2/
-                #
-                # SHA-2 errors look like this:
-                #   [Errno 1] _ssl.c:480: error:0D0C50A1:asn1 encoding routines:ASN1_item_verify:
-                #   unknown message digest algorithm
-                #
-                # Any other exceptions simply get raised.
-                if (
-                    "unknown message digest algorithm" not in str(e)
-                    or "SHOTGUN_FORCE_CERTIFICATE_VALIDATION" in os.environ
-                ):
-                    raise
-
-                if self.config.no_ssl_validation is False:
-                    LOG.warning(
-                        "SSL Error: this Python installation is incompatible with "
-                        "certificates signed with SHA-2. Disabling certificate validation. "
-                        "For more information, see https://www.shotgridsoftware.com/blog/"
-                        "important-ssl-certificate-renewal-and-sha-2/"
-                    )
-                    self._turn_off_ssl_validation()
-                    # reload user agent to reflect that we have turned off ssl validation
-                    req_headers["user-agent"] = "; ".join(self._user_agents)
-
-                self._close_connection()
-                if attempt == max_rpc_attempts:
-                    LOG.debug("Request failed.  Giving up after %d attempts." % attempt)
-                    raise
             except Exception:
                 self._close_connection()
                 if attempt == max_rpc_attempts:
@@ -4160,14 +4081,12 @@ def _get_connection(self):
                 timeout=self.config.timeout_secs,
                 ca_certs=self.__ca_certs,
                 proxy_info=pi,
-                disable_ssl_certificate_validation=self.config.no_ssl_validation,
             )
         else:
             self._connection = Http(
                 timeout=self.config.timeout_secs,
                 ca_certs=self.__ca_certs,
                 proxy_info=None,
-                disable_ssl_certificate_validation=self.config.no_ssl_validation,
             )
 
         return self._connection

tests/test_api.py

@@ -276,7 +276,7 @@ def test_upload_to_sg(self, mock_send_form):
         Upload an attachment tests for _upload_to_sg()
         """
         self.sg.server_info["s3_direct_uploads_enabled"] = False
-        mock_send_form.method.assert_called_once()
+
         mock_send_form.return_value = "1\n:123\nasd"
         this_dir, _ = os.path.split(__file__)
         u_path = os.path.abspath(
@@ -289,6 +289,8 @@ def test_upload_to_sg(self, mock_send_form):
             "attachments",
             tag_list="monkeys, everywhere, send, help",
         )
+
+        mock_send_form.assert_called_once()
         mock_send_form_args, _ = mock_send_form.call_args
         display_name_to_send = mock_send_form_args[1].get("display_name", "")
         self.assertTrue(isinstance(upload_id, int))
@@ -311,7 +313,7 @@ def test_upload_to_sg(self, mock_send_form):
             display_name_to_send.startswith("b'") and display_name_to_send.endswith("'")
         )
 
-        mock_send_form.method.assert_called_once()
+        # mock_send_form.method.assert_called_once() ## CANOT work because was already called earlier....
         mock_send_form.return_value = "2\nIt can't be upload"
         self.assertRaises(
             shotgun_api3.ShotgunError,
@@ -674,7 +676,6 @@ def share_thumbnail_retry(*args, **kwargs):
     def test_share_thumbnail_not_ready(self, mock_send_form):
         """throw an exception if trying to share a transient thumbnail"""
 
-        mock_send_form.method.assert_called_once()
         mock_send_form.return_value = (
             "2"
             "\nsource_entity image is a transient thumbnail that cannot be shared. "
@@ -692,7 +693,7 @@ def test_share_thumbnail_not_ready(self, mock_send_form):
     def test_share_thumbnail_returns_error(self, mock_send_form):
         """throw an exception if server returns an error code"""
 
-        mock_send_form.method.assert_called_once()
+        # mock_send_form.method.assert_called_once()  # never worked.....
         mock_send_form.return_value = "1\nerror message.\n"
 
         self.assertRaises(
@@ -2245,78 +2246,7 @@ def my_side_effect2(*args, **kwargs):
         finally:
             self.sg.config.rpc_attempt_interval = bak_rpc_attempt_interval
 
-    @unittest.mock.patch("shotgun_api3.shotgun.Http.request")
-    def test_sha2_error(self, mock_request):
-        # Simulate the exception raised with SHA-2 errors
-        mock_request.side_effect = ssl.SSLError(
-            "[Errno 1] _ssl.c:480: error:0D0C50A1:asn1 "
-            "encoding routines:ASN1_item_verify: unknown message digest "
-            "algorithm"
-        )
-
-        # save the original state
-        original_env_val = os.environ.pop("SHOTGUN_FORCE_CERTIFICATE_VALIDATION", None)
-
-        # ensure we're starting with the right values
-        self.sg.reset_user_agent()
-
-        # ensure the initial settings are correct. These will be different depending on whether
-        # the ssl module imported successfully or not.
-        if "ssl" in sys.modules:
-            self.assertFalse(self.sg.config.no_ssl_validation)
-            self.assertFalse(shotgun_api3.shotgun.NO_SSL_VALIDATION)
-            self.assertTrue("(validate)" in " ".join(self.sg._user_agents))
-            self.assertFalse("(no-validate)" in " ".join(self.sg._user_agents))
-        else:
-            self.assertTrue(self.sg.config.no_ssl_validation)
-            self.assertTrue(shotgun_api3.shotgun.NO_SSL_VALIDATION)
-            self.assertFalse("(validate)" in " ".join(self.sg._user_agents))
-            self.assertTrue("(no-validate)" in " ".join(self.sg._user_agents))
-
-        try:
-            self.sg.info()
-        except ssl.SSLError:
-            # ensure the api has reset the values in the correct fallback behavior
-            self.assertTrue(self.sg.config.no_ssl_validation)
-            self.assertTrue(shotgun_api3.shotgun.NO_SSL_VALIDATION)
-            self.assertFalse("(validate)" in " ".join(self.sg._user_agents))
-            self.assertTrue("(no-validate)" in " ".join(self.sg._user_agents))
-
-        if original_env_val is not None:
-            os.environ["SHOTGUN_FORCE_CERTIFICATE_VALIDATION"] = original_env_val
-
-    @unittest.mock.patch("shotgun_api3.shotgun.Http.request")
-    def test_sha2_error_with_strict(self, mock_request):
-        # Simulate the exception raised with SHA-2 errors
-        mock_request.side_effect = ssl.SSLError(
-            "[Errno 1] _ssl.c:480: error:0D0C50A1:asn1 "
-            "encoding routines:ASN1_item_verify: unknown message digest "
-            "algorithm"
-        )
-
-        # save the original state
-        original_env_val = os.environ.pop("SHOTGUN_FORCE_CERTIFICATE_VALIDATION", None)
-        os.environ["SHOTGUN_FORCE_CERTIFICATE_VALIDATION"] = "1"
-
-        # ensure we're starting with the right values
-        self.sg.config.no_ssl_validation = False
-        shotgun_api3.shotgun.NO_SSL_VALIDATION = False
-        self.sg.reset_user_agent()
-
-        try:
-            self.sg.info()
-        except ssl.SSLError:
-            # ensure the api has NOT reset the values in the fallback behavior because we have
-            # set the env variable to force validation
-            self.assertFalse(self.sg.config.no_ssl_validation)
-            self.assertFalse(shotgun_api3.shotgun.NO_SSL_VALIDATION)
-            self.assertFalse("(no-validate)" in " ".join(self.sg._user_agents))
-            self.assertTrue("(validate)" in " ".join(self.sg._user_agents))
-
-        if original_env_val is not None:
-            os.environ["SHOTGUN_FORCE_CERTIFICATE_VALIDATION"] = original_env_val
-
-    @unittest.mock.patch.object(urllib.request.OpenerDirector, "open")
+    @unittest.mock.patch.object(urllib.request.OpenerDirector, 'open')
     def test_sanitized_auth_params(self, mock_open):
         # Simulate the server blowing up and giving us a 500 error
         mock_open.side_effect = urllib.error.HTTPError("url", 500, "message", {}, None)

tests/test_client.py

@@ -28,7 +28,8 @@
 import shotgun_api3.lib.httplib2 as httplib2
 import shotgun_api3 as api
 from shotgun_api3.shotgun import ServerCapabilities, SG_TIMEZONE
-from . import base
+import shotgun_api3 as api
+import shotgun_api3.lib.httplib2 as httplib2
 
 
 def b64encode(val):
@@ -268,12 +269,11 @@ def test_user_agent(self):
         args, _ = self.sg._http_request.call_args
         (_, _, _, headers) = args
         ssl_validate_lut = {True: "no-validate", False: "validate"}
-        expected = "shotgun-json (%s); Python %s (%s); ssl %s (%s)" % (
+        expected = "shotgun-json (%s); Python %s (%s); ssl %s" % (
             api.__version__,
             client_caps.py_version,
             client_caps.platform.capitalize(),
             client_caps.ssl_version,
-            ssl_validate_lut[config.no_ssl_validation],
         )
         self.assertEqual(expected, headers.get("user-agent"))
 
@@ -282,12 +282,11 @@ def test_user_agent(self):
         self.sg.info()
         args, _ = self.sg._http_request.call_args
         (_, _, _, headers) = args
-        expected = "shotgun-json (%s); Python %s (%s); ssl %s (%s); test-agent" % (
+        expected = "shotgun-json (%s); Python %s (%s); ssl %s; test-agent" % (
             api.__version__,
             client_caps.py_version,
             client_caps.platform.capitalize(),
             client_caps.ssl_version,
-            ssl_validate_lut[config.no_ssl_validation],
         )
         self.assertEqual(expected, headers.get("user-agent"))
 
@@ -296,12 +295,11 @@ def test_user_agent(self):
         self.sg.info()
         args, _ = self.sg._http_request.call_args
         (_, _, _, headers) = args
-        expected = "shotgun-json (%s); Python %s (%s); ssl %s (%s)" % (
+        expected = "shotgun-json (%s); Python %s (%s); ssl %s" % (
             api.__version__,
             client_caps.py_version,
             client_caps.platform.capitalize(),
             client_caps.ssl_version,
-            ssl_validate_lut[config.no_ssl_validation],
         )
         self.assertEqual(expected, headers.get("user-agent"))
 
@@ -325,7 +323,7 @@ def test_network_retry(self):
             )
             # Ensure that sleep was called with the retry interval between each attempt
             attempt_interval = self.sg.config.rpc_attempt_interval / 1000.0
-            calls = [unittest.mock.callargs(((attempt_interval,), {}))]
+            calls = [(attempt_interval,)]
             calls *= self.sg.config.max_rpc_attempts - 1
             self.assertTrue(
                 mock_sleep.call_args_list == calls,