Stop providing a hardcoded CA bundle

dropbox/dropbox_client.py

@@ -181,8 +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
+        :param str ca_certs: a path to a file of concatenated CA certificates in PEM format.
+            Has the same meaning as when using :func:`ssl.wrap_socket`.
         """
 
         if not (oauth2_access_token or oauth2_refresh_token or (app_key and app_secret)):
@@ -590,7 +590,6 @@ def request_json_string(self,
                                headers=headers,
                                data=body,
                                stream=stream,
-                               verify=True,
                                timeout=timeout,
                                )
         self.raise_dropbox_error_for_resp(r)

dropbox/session.py

@@ -1,4 +1,3 @@
-import pkg_resources
 import os
 import ssl
 
@@ -32,21 +31,14 @@
 # This is the default longest time we'll block on receiving data from the server
 DEFAULT_TIMEOUT = 100
 
-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):
     _ca_certs = None
 
     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")
+        self._ca_certs = kwargs.pop("ca_certs", None)
         super(_SSLAdapter, self).__init__(*args, **kwargs)
 
     def init_poolmanager(self, connections, maxsize, block=False, **_):
@@ -59,8 +51,19 @@ def init_poolmanager(self, connections, maxsize, block=False, **_):
         )
 
 def pinned_session(pool_maxsize=8, ca_certs=None):
-    http_adapter = _SSLAdapter(pool_connections=4, pool_maxsize=pool_maxsize, ca_certs=ca_certs)
+    # always verify, use cert bundle if provided
+
     _session = requests.session()
-    _session.mount('https://', http_adapter)
 
+    # requests
+    if ca_certs is not None:
+        _session.verify = ca_certs
+    else:
+        _session.verify = True
+
+    # urllib3 within requests
+    http_adapter = _SSLAdapter(pool_connections=4, pool_maxsize=pool_maxsize, ca_certs=ca_certs)
+    _session.mount('https://', http_adapter)
     return _session
+
+SSLError = requests.exceptions.SSLError  # raised on verification errors

setup.py

@@ -54,7 +54,6 @@
     setup_requires=setup_requires,
     tests_require=test_reqs,
     packages=['dropbox'],
-    package_data={'dropbox': ['trusted-certs.crt']},
     zip_safe=False,
     author_email='[email protected]',
     author='Dropbox',

test/integration/expired-certs.crt

@@ -0,0 +1,78 @@
+# GeoTrust Global CA.pem
+# Certificate:
+#     Data:
+#         Version: 3 (0x2)
+#         Serial Number: 144470 (0x23456)
+#     Signature Algorithm: sha1WithRSAEncryption
+#         Issuer: C=US, O=GeoTrust Inc., CN=GeoTrust Global CA
+#         Validity
+#             Not Before: May 21 04:00:00 2002 GMT
+#             Not After : May 21 04:00:00 2022 GMT
+#         Subject: C=US, O=GeoTrust Inc., CN=GeoTrust Global CA
+#         Subject Public Key Info:
+#             Public Key Algorithm: rsaEncryption
+#                 Public-Key: (2048 bit)
+#                 Modulus:
+#                     00:da:cc:18:63:30:fd:f4:17:23:1a:56:7e:5b:df:
+#                     3c:6c:38:e4:71:b7:78:91:d4:bc:a1:d8:4c:f8:a8:
+#                     43:b6:03:e9:4d:21:07:08:88:da:58:2f:66:39:29:
+#                     bd:05:78:8b:9d:38:e8:05:b7:6a:7e:71:a4:e6:c4:
+#                     60:a6:b0:ef:80:e4:89:28:0f:9e:25:d6:ed:83:f3:
+#                     ad:a6:91:c7:98:c9:42:18:35:14:9d:ad:98:46:92:
+#                     2e:4f:ca:f1:87:43:c1:16:95:57:2d:50:ef:89:2d:
+#                     80:7a:57:ad:f2:ee:5f:6b:d2:00:8d:b9:14:f8:14:
+#                     15:35:d9:c0:46:a3:7b:72:c8:91:bf:c9:55:2b:cd:
+#                     d0:97:3e:9c:26:64:cc:df:ce:83:19:71:ca:4e:e6:
+#                     d4:d5:7b:a9:19:cd:55:de:c8:ec:d2:5e:38:53:e5:
+#                     5c:4f:8c:2d:fe:50:23:36:fc:66:e6:cb:8e:a4:39:
+#                     19:00:b7:95:02:39:91:0b:0e:fe:38:2e:d1:1d:05:
+#                     9a:f6:4d:3e:6f:0f:07:1d:af:2c:1e:8f:60:39:e2:
+#                     fa:36:53:13:39:d4:5e:26:2b:db:3d:a8:14:bd:32:
+#                     eb:18:03:28:52:04:71:e5:ab:33:3d:e1:38:bb:07:
+#                     36:84:62:9c:79:ea:16:30:f4:5f:c0:2b:e8:71:6b:
+#                     e4:f9
+#                 Exponent: 65537 (0x10001)
+#         X509v3 extensions:
+#             X509v3 Basic Constraints: critical
+#                 CA:TRUE
+#             X509v3 Subject Key Identifier:
+#                 C0:7A:98:68:8D:89:FB:AB:05:64:0C:11:7D:AA:7D:65:B8:CA:CC:4E
+#             X509v3 Authority Key Identifier:
+#                 keyid:C0:7A:98:68:8D:89:FB:AB:05:64:0C:11:7D:AA:7D:65:B8:CA:CC:4E
+#
+#     Signature Algorithm: sha1WithRSAEncryption
+#          35:e3:29:6a:e5:2f:5d:54:8e:29:50:94:9f:99:1a:14:e4:8f:
+#          78:2a:62:94:a2:27:67:9e:d0:cf:1a:5e:47:e9:c1:b2:a4:cf:
+#          dd:41:1a:05:4e:9b:4b:ee:4a:6f:55:52:b3:24:a1:37:0a:eb:
+#          64:76:2a:2e:2c:f3:fd:3b:75:90:bf:fa:71:d8:c7:3d:37:d2:
+#          b5:05:95:62:b9:a6:de:89:3d:36:7b:38:77:48:97:ac:a6:20:
+#          8f:2e:a6:c9:0c:c2:b2:99:45:00:c7:ce:11:51:22:22:e0:a5:
+#          ea:b6:15:48:09:64:ea:5e:4f:74:f7:05:3e:c7:8a:52:0c:db:
+#          15:b4:bd:6d:9b:e5:c6:b1:54:68:a9:e3:69:90:b6:9a:a5:0f:
+#          b8:b9:3f:20:7d:ae:4a:b5:b8:9c:e4:1d:b6:ab:e6:94:a5:c1:
+#          c7:83:ad:db:f5:27:87:0e:04:6c:d5:ff:dd:a0:5d:ed:87:52:
+#          b7:2b:15:02:ae:39:a6:6a:74:e9:da:c4:e7:bc:4d:34:1e:a9:
+#          5c:4d:33:5f:92:09:2f:88:66:5d:77:97:c7:1d:76:13:a9:d5:
+#          e5:f1:16:09:11:35:d5:ac:db:24:71:70:2c:98:56:0b:d9:17:
+#          b4:d1:e3:51:2b:5e:75:e8:d5:d0:dc:4f:34:ed:c2:05:66:80:
+#          a1:cb:e6:33
+-----BEGIN CERTIFICATE-----
+MIIDVDCCAjygAwIBAgIDAjRWMA0GCSqGSIb3DQEBBQUAMEIxCzAJBgNVBAYTAlVT
+MRYwFAYDVQQKEw1HZW9UcnVzdCBJbmMuMRswGQYDVQQDExJHZW9UcnVzdCBHbG9i
+YWwgQ0EwHhcNMDIwNTIxMDQwMDAwWhcNMjIwNTIxMDQwMDAwWjBCMQswCQYDVQQG
+EwJVUzEWMBQGA1UEChMNR2VvVHJ1c3QgSW5jLjEbMBkGA1UEAxMSR2VvVHJ1c3Qg
+R2xvYmFsIENBMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA2swYYzD9
+9BcjGlZ+W988bDjkcbd4kdS8odhM+KhDtgPpTSEHCIjaWC9mOSm9BXiLnTjoBbdq
+fnGk5sRgprDvgOSJKA+eJdbtg/OtppHHmMlCGDUUna2YRpIuT8rxh0PBFpVXLVDv
+iS2Aelet8u5fa9IAjbkU+BQVNdnARqN7csiRv8lVK83Qlz6cJmTM386DGXHKTubU
+1XupGc1V3sjs0l44U+VcT4wt/lAjNvxm5suOpDkZALeVAjmRCw7+OC7RHQWa9k0+
+bw8HHa8sHo9gOeL6NlMTOdReJivbPagUvTLrGAMoUgRx5aszPeE4uwc2hGKceeoW
+MPRfwCvocWvk+QIDAQABo1MwUTAPBgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBTA
+ephojYn7qwVkDBF9qn1luMrMTjAfBgNVHSMEGDAWgBTAephojYn7qwVkDBF9qn1l
+uMrMTjANBgkqhkiG9w0BAQUFAAOCAQEANeMpauUvXVSOKVCUn5kaFOSPeCpilKIn
+Z57QzxpeR+nBsqTP3UEaBU6bS+5Kb1VSsyShNwrrZHYqLizz/Tt1kL/6cdjHPTfS
+tQWVYrmm3ok9Nns4d0iXrKYgjy6myQzCsplFAMfOEVEiIuCl6rYVSAlk6l5PdPcF
+PseKUgzbFbS9bZvlxrFUaKnjaZC2mqUPuLk/IH2uSrW4nOQdtqvmlKXBx4Ot2/Un
+hw4EbNX/3aBd7YdStysVAq45pmp06drE57xNNB6pXE0zX5IJL4hmXXeXxx12E6nV
+5fEWCRE11azbJHFwLJhWC9kXtNHjUStedejV0NxPNO3CBWaAocvmMw==
+-----END CERTIFICATE-----

test/integration/test_dropbox.py

@@ -16,6 +16,7 @@
     from StringIO import StringIO as BytesIO
 
 from dropbox import (
+    create_session,
     Dropbox,
     DropboxOAuth2Flow,
     DropboxTeam,
@@ -39,6 +40,7 @@
     PathRoot,
     PathRoot_validator,
 )
+from dropbox.session import SSLError
 
 # Key Types
 REFRESH_TOKEN_KEY = "REFRESH_TOKEN"
@@ -65,34 +67,43 @@ def _value_from_env_or_die(env_name):
         sys.exit(1)
     return value
 
+_TRUSTED_CERTS_FILE = os.path.join(os.path.dirname(__file__), "trusted-certs.crt")
+_EXPIRED_CERTS_FILE = os.path.join(os.path.dirname(__file__), "expired-certs.crt")
+
+# enables testing both with and without a manually-provided CA bundle
+@pytest.fixture(params=[None, _TRUSTED_CERTS_FILE], ids=["no-pinning", "pinning"])
+def dbx_session(request):
+    return create_session(ca_certs=request.param)
+
 
 @pytest.fixture()
-def dbx_from_env():
+def dbx_from_env(dbx_session):
     oauth2_token = _value_from_env_or_die(format_env_name())
-    return Dropbox(oauth2_token)
+    return Dropbox(oauth2_token, session=dbx_session)
 
 
 @pytest.fixture()
-def refresh_dbx_from_env():
+def refresh_dbx_from_env(dbx_session):
     refresh_token = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, REFRESH_TOKEN_KEY))
     app_key = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_ID_KEY))
     app_secret = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_SECRET_KEY))
     return Dropbox(oauth2_refresh_token=refresh_token,
-                   app_key=app_key, app_secret=app_secret)
+                   app_key=app_key, app_secret=app_secret,
+                   session=dbx_session)
 
 
 @pytest.fixture()
-def dbx_team_from_env():
+def dbx_team_from_env(dbx_session):
     team_oauth2_token = _value_from_env_or_die(
         format_env_name(SCOPED_KEY, TEAM_KEY, ACCESS_TOKEN_KEY))
-    return DropboxTeam(team_oauth2_token)
+    return DropboxTeam(team_oauth2_token, session=dbx_session)
 
 
 @pytest.fixture()
-def dbx_app_auth_from_env():
+def dbx_app_auth_from_env(dbx_session):
     app_key = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_ID_KEY))
     app_secret = _value_from_env_or_die(format_env_name(SCOPED_KEY, USER_KEY, CLIENT_SECRET_KEY))
-    return Dropbox(app_key=app_key, app_secret=app_secret)
+    return Dropbox(app_key=app_key, app_secret=app_secret, session=dbx_session)
 
 
 @pytest.fixture()
@@ -110,7 +121,7 @@ def dbx_share_url_from_env():
 TIMESTAMP = str(datetime.datetime.utcnow())
 STATIC_FILE = "/test.txt"
 
-@pytest.fixture(scope='module', autouse=True)
+@pytest.fixture(scope='module')
 def pytest_setup():
     print("Setup")
     dbx = Dropbox(_value_from_env_or_die(format_env_name()))
@@ -125,47 +136,14 @@ def pytest_setup():
     except Exception:
         print("File not found")
 
-
 @pytest.mark.usefixtures(
+    "pytest_setup",
     "dbx_from_env",
     "refresh_dbx_from_env",
     "dbx_app_auth_from_env",
-    "dbx_share_url_from_env"
+    "dbx_share_url_from_env",
 )
 class TestDropbox:
-    def test_default_oauth2_urls(self):
-        flow_obj = DropboxOAuth2Flow('dummy_app_key', 'dummy_app_secret',
-            'http://localhost/dummy', 'dummy_session', 'dbx-auth-csrf-token')
-
-        assert re.match(
-            r'^https://{}/oauth2/authorize\?'.format(re.escape(session.WEB_HOST)),
-            flow_obj._get_authorize_url('http://localhost/redirect', 'state', 'legacy'),
-        )
-
-        assert flow_obj.build_url(
-            '/oauth2/authorize'
-        ) == 'https://{}/oauth2/authorize'.format(session.API_HOST)
-
-        assert flow_obj.build_url(
-            '/oauth2/authorize', host=session.WEB_HOST
-        ) == 'https://{}/oauth2/authorize'.format(session.WEB_HOST)
-
-    def test_bad_auth(self):
-        # Test malformed token
-        malformed_token_dbx = Dropbox(MALFORMED_TOKEN)
-        # TODO: backend is no longer returning `BadInputError`
-        # with pytest.raises(BadInputError,) as cm:
-        #     malformed_token_dbx.files_list_folder('')
-        # assert 'token is malformed' in cm.value.message
-        with pytest.raises(AuthError,):
-            malformed_token_dbx.files_list_folder('')
-
-        # Test reasonable-looking invalid token
-        invalid_token_dbx = Dropbox(INVALID_TOKEN)
-        with pytest.raises(AuthError) as cm:
-            invalid_token_dbx.files_list_folder('')
-        assert cm.value.error.is_invalid_access_token()
-
     def test_multi_auth(self, dbx_from_env, dbx_app_auth_from_env, dbx_share_url_from_env):
         # Test for user (with oauth token)
         preview_result, resp = dbx_from_env.files_get_thumbnail_v2(
@@ -280,7 +258,10 @@ def test_versioned_route(self, dbx_from_env):
         # Verify response type is of v2 route
         assert isinstance(resp, DeleteResult)
 
-@pytest.mark.usefixtures("dbx_team_from_env")
+@pytest.mark.usefixtures(
+    "pytest_setup",
+    "dbx_team_from_env",
+)
 class TestDropboxTeam:
     def test_team(self, dbx_team_from_env):
         dbx_team_from_env.team_groups_list()
@@ -310,3 +291,48 @@ def test_clone_when_team_linked(self, dbx_team_from_env):
         new_dbxt = dbx_team_from_env.clone()
         assert dbx_team_from_env is not new_dbxt
         assert isinstance(new_dbxt, dbx_team_from_env.__class__)
+
+def test_default_oauth2_urls():
+    flow_obj = DropboxOAuth2Flow('dummy_app_key', 'dummy_app_secret',
+        'http://localhost/dummy', 'dummy_session', 'dbx-auth-csrf-token')
+
+    assert re.match(
+        r'^https://{}/oauth2/authorize\?'.format(re.escape(session.WEB_HOST)),
+        flow_obj._get_authorize_url('http://localhost/redirect', 'state', 'legacy'),
+    )
+
+    assert flow_obj.build_url(
+        '/oauth2/authorize'
+    ) == 'https://{}/oauth2/authorize'.format(session.API_HOST)
+
+    assert flow_obj.build_url(
+        '/oauth2/authorize', host=session.WEB_HOST
+    ) == 'https://{}/oauth2/authorize'.format(session.WEB_HOST)
+
+def test_bad_auth(dbx_session):
+    # Test malformed token
+    malformed_token_dbx = Dropbox(MALFORMED_TOKEN, session=dbx_session)
+    # TODO: backend is no longer returning `BadInputError`
+    # with pytest.raises(BadInputError,) as cm:
+    #     malformed_token_dbx.files_list_folder('')
+    # assert 'token is malformed' in cm.value.message
+    with pytest.raises(AuthError):
+        malformed_token_dbx.files_list_folder('')
+
+    # Test reasonable-looking invalid token
+    invalid_token_dbx = Dropbox(INVALID_TOKEN, session=dbx_session)
+    with pytest.raises(AuthError) as cm:
+        invalid_token_dbx.files_list_folder('')
+    assert cm.value.error.is_invalid_access_token()
+
+def test_bad_pins():
+    # sanity-check: if we're pinning using expired pins, we should fail w/ an SSL error
+    _dbx = Dropbox("dummy_token", ca_certs=_EXPIRED_CERTS_FILE)
+    with pytest.raises(SSLError,):
+        _dbx.files_list_folder('')
+
+def test_bad_pins_session():
+    _session = create_session(ca_certs=_EXPIRED_CERTS_FILE)
+    _dbx = Dropbox("dummy_token2", session=_session)
+    with pytest.raises(SSLError,):
+        _dbx.files_list_folder('')