Add ca_certs argument for oauth and dropbox client (#385)

stanislau-arkhipenka · Stanislau Arkhipenka · Brent1LT · web-flow · commit f3f8729ebd1a · 2022-06-30T10:32:49.000-07:00
* Add ca_certs argument for oauth and dropbox client

* Replace FileNotFoundError with AttributeError

New error better represents actual problem (CA not set)
And also compatible with python2.7

Co-authored-by: Stanislau Arkhipenka &lt;stas@stanislau.arkhipenka@gmail.com&gt;
Co-authored-by: Brent Bumann &lt;brent.bumann@aol.com&gt;
diff --git a/dropbox/dropbox_client.py b/dropbox/dropbox_client.py
@@ -97,7 +97,7 @@ def __init__(self, request_id, obj_result):
         self.request_id = request_id
         self.obj_result = obj_result
 
-def create_session(max_connections=8, proxies=None):
+def create_session(max_connections=8, proxies=None, ca_certs=None):
     """
     Creates a session object that can be used by multiple :class:`Dropbox` and
     :class:`DropboxTeam` instances. This lets you share a connection pool
@@ -112,7 +112,7 @@ def create_session(max_connections=8, proxies=None):
         for more details.
     """
     # We only need as many pool_connections as we have unique hostnames.
-    session = pinned_session(pool_maxsize=max_connections)
+    session = pinned_session(pool_maxsize=max_connections, ca_certs=ca_certs)
     if proxies:
         session.proxies = proxies
     return session
@@ -151,7 +151,8 @@ def __init__(self,
                  oauth2_access_token_expiration=None,
                  app_key=None,
                  app_secret=None,
-                 scope=None,):
+                 scope=None,
+                 ca_certs=None):
         """
         :param str oauth2_access_token: OAuth2 access token for making client
             requests.
@@ -180,6 +181,8 @@ def __init__(self,
             Not required if PKCE was used to authorize the token
         :param list scope: list of scopes to request on refresh.  If left blank,
             refresh will request all available scopes for application
+        :param str ca_certs: path to CA certificate. If left blank, default certificate location \
+            will be used
         """
 
         if not (oauth2_access_token or oauth2_refresh_token or (app_key and app_secret)):
@@ -212,7 +215,7 @@ def __init__(self,
                                         .format(session))
             self._session = session
         else:
-            self._session = create_session()
+            self._session = create_session(ca_certs=ca_certs)
         self._headers = headers
 
         base_user_agent = 'OfficialDropboxPythonSDKv2/' + __version__
diff --git a/dropbox/oauth.py b/dropbox/oauth.py
@@ -119,7 +119,8 @@ def __repr__(self):
 class DropboxOAuth2FlowBase(object):
 
     def __init__(self, consumer_key, consumer_secret=None, locale=None, token_access_type=None,
-                 scope=None, include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT):
+                 scope=None, include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT,
+                 ca_certs=None):
         if scope is not None and (len(scope) == 0 or not isinstance(scope, list)):
             raise BadInputException("Scope list must be of type list")
         if token_access_type is not None and token_access_type not in TOKEN_ACCESS_TYPES:
@@ -134,7 +135,7 @@ def __init__(self, consumer_key, consumer_secret=None, locale=None, token_access
         self.consumer_secret = consumer_secret
         self.locale = locale
         self.token_access_type = token_access_type
-        self.requests_session = pinned_session()
+        self.requests_session = pinned_session(ca_certs=ca_certs)
         self.scope = scope
         self.include_granted_scopes = include_granted_scopes
         self._timeout = timeout
@@ -273,7 +274,8 @@ class DropboxOAuth2FlowNoRedirect(DropboxOAuth2FlowBase):
     """
 
     def __init__(self, consumer_key, consumer_secret=None, locale=None, token_access_type=None,
-                 scope=None, include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT):  # noqa: E501;
+                 scope=None, include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT,
+                 ca_certs=None):  # noqa: E501;
         """
         Construct an instance.
 
@@ -306,6 +308,8 @@ def __init__(self, consumer_key, consumer_secret=None, locale=None, token_access
             client will wait for any single packet from the server. After the timeout the client
             will give up on connection. If `None`, client will wait forever. Defaults
             to 100 seconds.
+        :param str ca_cert: path to CA certificate. If left blank, default certificate location \
+            will be used
         """
         super(DropboxOAuth2FlowNoRedirect, self).__init__(
             consumer_key=consumer_key,
@@ -315,7 +319,8 @@ def __init__(self, consumer_key, consumer_secret=None, locale=None, token_access
             scope=scope,
             include_granted_scopes=include_granted_scopes,
             use_pkce=use_pkce,
-            timeout=timeout
+            timeout=timeout,
+            ca_certs=ca_certs
         )
 
     def start(self):
@@ -360,7 +365,8 @@ class DropboxOAuth2Flow(DropboxOAuth2FlowBase):
     def __init__(self, consumer_key, redirect_uri, session,
                  csrf_token_session_key, consumer_secret=None, locale=None,
                  token_access_type=None, scope=None,
-                 include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT):
+                 include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT,
+                 ca_certs=None):
         """
         Construct an instance.
 
@@ -399,6 +405,8 @@ def __init__(self, consumer_key, redirect_uri, session,
         :param Optional[float] timeout: Maximum duration in seconds that client will wait for any
             single packet from the server. After the timeout the client will give up on connection.
             If `None`, client will wait forever. Defaults to 100 seconds.
+        :param str ca_cert: path to CA certificate. If left blank, default certificate location \
+            will be used
         """
 
         super(DropboxOAuth2Flow, self).__init__(
@@ -409,7 +417,8 @@ def __init__(self, consumer_key, redirect_uri, session,
             scope=scope,
             include_granted_scopes=include_granted_scopes,
             use_pkce=use_pkce,
-            timeout=timeout
+            timeout=timeout,
+            ca_certs=ca_certs
         )
         self.redirect_uri = redirect_uri
         self.session = session
diff --git a/dropbox/session.py b/dropbox/session.py
@@ -32,22 +32,33 @@
 # This is the default longest time we'll block on receiving data from the server
 DEFAULT_TIMEOUT = 100
 
-_TRUSTED_CERT_FILE = pkg_resources.resource_filename(__name__, 'trusted-certs.crt')
+try:
+    _TRUSTED_CERT_FILE = pkg_resources.resource_filename(__name__, 'trusted-certs.crt')
+except NotImplementedError:  # Package is used inside python archive
+    _TRUSTED_CERT_FILE = None
+
 
 # TODO(kelkabany): We probably only want to instantiate this once so that even
 # if multiple Dropbox objects are instantiated, they all share the same pool.
 class _SSLAdapter(HTTPAdapter):
+
+    def __init__(self, *args, **kwargs):
+        self._ca_certs = kwargs.pop("ca_certs", None) or _TRUSTED_CERT_FILE
+        if not self._ca_certs:
+            raise AttributeError("CA certificate not set")
+        super(_SSLAdapter, self).__init__(*args, **kwargs)
+
     def init_poolmanager(self, connections, maxsize, block=False, **_):
         self.poolmanager = PoolManager(
             num_pools=connections,
             maxsize=maxsize,
             block=block,
             cert_reqs=ssl.CERT_REQUIRED,
-            ca_certs=_TRUSTED_CERT_FILE,
+            ca_certs=self._ca_certs,
         )
 
-def pinned_session(pool_maxsize=8):
-    http_adapter = _SSLAdapter(pool_connections=4, pool_maxsize=pool_maxsize)
+def pinned_session(pool_maxsize=8, ca_certs=None):
+    http_adapter = _SSLAdapter(pool_connections=4, pool_maxsize=pool_maxsize, ca_certs=ca_certs)
     _session = requests.session()
     _session.mount('https://', http_adapter)
 
diff --git a/test/unit/test_dropbox_unit.py b/test/unit/test_dropbox_unit.py
@@ -22,6 +22,7 @@
 TEAM_MEMBER_ID = 'dummy_team_member_id'
 SCOPE_LIST = ['files.metadata.read', 'files.metadata.write']
 EXPIRATION = datetime.utcnow() + timedelta(seconds=EXPIRES_IN)
+CA_CERTS = "/dummy/path/ca.crt"
 
 EXPIRATION_BUFFER = timedelta(minutes=5)
 
@@ -85,6 +86,10 @@ def test_authorization_url(self):
                                     assert 'code_challenge_method' not in authorization_url
                                     assert 'code_challenge' not in authorization_url
 
+    def test_authorization_with_ca_certs(self):
+        DropboxOAuth2Flow(APP_KEY, APP_SECRET, 'http://localhost/dummy', 'dummy_session',
+                          'dbx-auth-csrf-token', ca_certs=CA_CERTS)
+
     def test_authorization_url_legacy_default(self):
         flow_obj = DropboxOAuth2Flow(APP_KEY, APP_SECRET, 'http://localhost/dummy',
                                      'dummy_session', 'dbx-auth-csrf-token')
@@ -233,6 +238,14 @@ def session_instance(self, mocker):
         mocker.patch.object(session_obj, 'post', return_value=post_response)
         return session_obj
 
+    @pytest.fixture(scope='function')
+    def session_instance_with_ca_certs(self, mocker):
+        session_obj = create_session(ca_certs=CA_CERTS)
+        post_response = mock.MagicMock(status_code=200)
+        post_response.json.return_value = {"access_token": ACCESS_TOKEN, "expires_in": EXPIRES_IN}
+        mocker.patch.object(session_obj, 'post', return_value=post_response)
+        return session_obj
+
     @pytest.fixture(scope='function')
     def invalid_grant_session_instance(self, mocker):
         session_obj = create_session()
@@ -366,6 +379,10 @@ def test_check_refresh_with_invalid_grant(self, invalid_grant_session_instance):
             assert invalid_grant_session_instance.post.call_count == 1
             assert e.error.is_invalid_access_token()
 
+    def test_check_Dropbox_with_ca_certs(self, session_instance_with_ca_certs):
+        Dropbox(oauth2_access_token=ACCESS_TOKEN, oauth2_access_token_expiration=EXPIRATION,
+                session=session_instance_with_ca_certs)
+
     def test_team_client_refresh(self, session_instance):
         dbx = DropboxTeam(oauth2_refresh_token=REFRESH_TOKEN,
                       app_key=APP_KEY,
