SNOW-90415 : Caught uncaught exception.

SNOW-91743 : Updated connection timeouts for Cache Server and OCSP Responders.
SNOW-82276 : Removed support for OLD OCSP URL for Privatelink in SnowSQL.
SNOW-90234 : OOB Telemetry for Revoked Certificates is updated to be sent with Urgent flag.

Author: sfc-gh-stakeda

connection.py

@@ -73,8 +73,6 @@
     HeartBeatTimer, get_time_millis)
 from .util_text import split_statements, construct_hostname, parse_account
 
-from snowflake.connector.network import APPLICATION_SNOWSQL
-
 
 def DefaultConverterClass():
     if PY_ISSUE_23517 or IS_WINDOWS:
@@ -612,36 +610,11 @@ def __set_error_attributes(self):
     @staticmethod
     def setup_ocsp_privatelink(app, hostname):
         SnowflakeConnection.OCSP_ENV_LOCK.acquire()
-        if app is APPLICATION_SNOWSQL:
-            ocsp_cache_server = u'http://ocsp{}/ocsp_response_cache.json'.format(
-                hostname[hostname.index('.'):])
-            '''
-             Check if user has configured a custom OCSP Cache Server URL
-             '''
-            if 'SF_OCSP_RESPONSE_CACHE_SERVER_URL' not in os.environ:
-                os.environ[
-                    'SF_OCSP_RESPONSE_CACHE_SERVER_URL'] = ocsp_cache_server
-            else:
-                if "ocsp_response_cache" not in os.environ['SF_OCSP_RESPONSE_CACHE_SERVER_URL']:
-                    if not os.environ['SF_OCSP_RESPONSE_CACHE_SERVER_URL']. \
-                            startswith("http://"):
-                        ocsp_cache_server = "http://{0}/{1}".format(
-                            os.environ['SF_OCSP_RESPONSE_CACHE_SERVER_URL'],
-                            "ocsp_response_cache.json")
-                    else:
-                        ocsp_cache_server = "{0}/{1}".format(
-                            os.environ['SF_OCSP_RESPONSE_CACHE_SERVER_URL'],
-                            "ocsp_response_cache.json")
-                else:
-                    ocsp_cache_server = os.environ['SF_OCSP_RESPONSE_CACHE_SERVER_URL']
-
-                os.environ['SF_OCSP_RESPONSE_CACHE_SERVER_URL'] = ocsp_cache_server
-        else:
-            ocsp_cache_server = \
-                u'http://ocsp.{}/ocsp_response_cache.json'.format(
-                    hostname)
-            os.environ[
-                'SF_OCSP_RESPONSE_CACHE_SERVER_URL'] = ocsp_cache_server
+        ocsp_cache_server = \
+            u'http://ocsp.{}/ocsp_response_cache.json'.format(
+                hostname)
+        os.environ[
+            'SF_OCSP_RESPONSE_CACHE_SERVER_URL'] = ocsp_cache_server
         logger.debug(u"OCSP Cache Server is updated: %s", ocsp_cache_server)
         SnowflakeConnection.OCSP_ENV_LOCK.release()
 

ocsp_asn1crypto.py

@@ -25,6 +25,7 @@
 from snowflake.connector.ocsp_snowflake import SnowflakeOCSP
 from collections import OrderedDict
 from snowflake.connector.ssd_internal_keys import ret_wildcard_hkey
+from os import getenv
 
 logger = getLogger(__name__)
 
@@ -223,11 +224,16 @@ def is_valid_time(self, cert_id, ocsp_response):
     def process_ocsp_response(self, issuer, cert_id, ocsp_response):
         try:
             res = OCSPResponse.load(ocsp_response)
+            if self.test_mode is not None:
+                ocsp_load_failure = getenv("SF_TEST_OCSP_FORCE_BAD_OCSP_RESPONSE")
+                if ocsp_load_failure is not None:
+                    raise RevocationCheckError("Force fail")
         except Exception:
             raise RevocationCheckError(
                 msg='Invalid OCSP Response',
                 errno=ER_INVALID_OCSP_RESPONSE
             )
+
         if res['response_status'].native != 'successful':
             raise RevocationCheckError(
                 msg="Invalid Status: {0}".format(res['response_status'].native),
@@ -256,6 +262,7 @@ def process_ocsp_response(self, issuer, cert_id, ocsp_response):
                 ocsp_cert['tbs_certificate'])
 
             cert_valid, debug_msg = self.check_cert_time_validity(cur_time, ocsp_cert)
+
             if not cert_valid:
                 raise RevocationCheckError(
                     msg=debug_msg,
@@ -277,6 +284,15 @@ def process_ocsp_response(self, issuer, cert_id, ocsp_response):
 
         single_response = tbs_response_data['responses'][0]
         cert_status = single_response['cert_status'].name
+        if self.test_mode is not None:
+            test_cert_status = getenv("SF_TEST_OCSP_CERT_STATUS")
+            if test_cert_status == 'revoked':
+                cert_status = 'revoked'
+            elif test_cert_status == 'unknown':
+                cert_status = 'unknown'
+            elif test_cert_status == 'good':
+                cert_status = 'good'
+
         try:
             if cert_status == 'good':
                 self._process_good_status(single_response, cert_id, ocsp_response)

ocsp_pyasn1.py

@@ -11,6 +11,7 @@
 from datetime import datetime
 from logging import getLogger
 from threading import Lock
+from os import getenv
 
 import pyasn1
 from Cryptodome.Hash import SHA256, SHA384, SHA1, SHA512
@@ -438,6 +439,10 @@ def is_valid_time(self, cert_id, ocsp_response):
     def process_ocsp_response(self, issuer, cert_id, ocsp_response):
         try:
             res = der_decoder.decode(ocsp_response, OCSPResponse())[0]
+            if self.test_mode is not None:
+                ocsp_load_failure = getenv("SF_TEST_OCSP_FORCE_BAD_OCSP_RESPONSE")
+                if ocsp_load_failure is not None:
+                    raise RevocationCheckError("Force fail")
         except Exception:
             raise RevocationCheckError(
                 msg='Invalid OCSP Response',
@@ -502,6 +507,16 @@ def process_ocsp_response(self, issuer, cert_id, ocsp_response):
 
         single_response = tbs_response_data.getComponentByName('responses')[0]
         cert_status = single_response.getComponentByName('certStatus')
+
+        if self.test_mode is not None:
+            test_cert_status = getenv("SF_TEST_OCSP_CERT_STATUS")
+            if test_cert_status == 'revoked':
+                cert_status = 'revoked'
+            elif test_cert_status == 'unknown':
+                cert_status = 'unknown'
+            elif test_cert_status == 'good':
+                cert_status = 'good'
+
         try:
             if cert_status.getName() == 'good':
                 self._process_good_status(single_response, cert_id, ocsp_response)