try loading trusted certs from a list of fallbacks

src/OpenSSL/SSL.py

@@ -1,3 +1,5 @@
+import os
+import re
 import socket
 from sys import platform
 from functools import wraps, partial
@@ -130,6 +132,19 @@ class _buffer(object):
 SSL_CB_HANDSHAKE_START = _lib.SSL_CB_HANDSHAKE_START
 SSL_CB_HANDSHAKE_DONE = _lib.SSL_CB_HANDSHAKE_DONE
 
+# Taken from https://golang.org/src/crypto/x509/root_linux.go
+_CERTIFICATE_FILE_LOCATIONS = [
+    "/etc/ssl/certs/ca-certificates.crt",  # Debian/Ubuntu/Gentoo etc.
+    "/etc/pki/tls/certs/ca-bundle.crt",  # Fedora/RHEL 6
+    "/etc/ssl/ca-bundle.pem",  # OpenSUSE
+    "/etc/pki/tls/cacert.pem",  # OpenELEC
+    "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem",  # CentOS/RHEL 7
+]
+
+_CERTIFICATE_PATH_LOCATIONS = [
+    "/etc/ssl/certs",  # SLES10/SLES11
+]
+
 
 class Error(Exception):
     """
@@ -699,8 +714,98 @@ def set_default_verify_paths(self):
 
         :return: None
         """
+        # This function will attempt to load certs from both a cafile and
+        # capath that are set at compile time. However, it will first check
+        # environment variables and, if present, load those paths instead
         set_result = _lib.SSL_CTX_set_default_verify_paths(self._context)
         _openssl_assert(set_result == 1)
+        # After attempting to set default_verify_paths we need to know whether
+        # to go down the fallback path.
+        # First we'll check to see if any env vars have been set. If so,
+        # we won't try to do anything else because the user has set the path
+        # themselves.
+        dir_env_var = _ffi.string(
+            _lib.X509_get_default_cert_dir_env()
+        ).decode("ascii")
+        file_env_var = _ffi.string(
+            _lib.X509_get_default_cert_file_env()
+        ).decode("ascii")
+        if not self._verify_env_vars_set(dir_env_var, file_env_var):
+            # If no env vars are set next we want to see if any certs were
+            # loaded. For a cafile this is simple and we can just ask how many
+            # objects are present. However, the cert directory (capath) is
+            # lazily loaded and num will always be zero so we need to check if
+            # the dir exists and has valid file names in it to cover that case.
+            num = self._check_num_store_objects()
+            default_dir = _ffi.string(_lib.X509_get_default_cert_dir())
+            if num == 0 and not self._default_dir_exists(default_dir):
+                # No certs and no default dir, let's load our fallbacks
+                self._fallback_default_verify_paths(
+                    _CERTIFICATE_FILE_LOCATIONS,
+                    _CERTIFICATE_PATH_LOCATIONS
+                )
+
+    def _verify_env_vars_set(self, dir_env_var, file_env_var):
+        """
+        Check to see if the default cert dir/file environment vars are present.
+
+        :return: bool
+        """
+        return (
+            os.environ.get(file_env_var, None) is not None or
+            os.environ.get(dir_env_var, None) is not None
+        )
+
+    def _default_dir_exists(self, default_dir):
+        """
+        Check to see if the default cert dir exists and has filenames in a
+        valid form.
+
+        :return: bool
+        """
+        try:
+            l = os.listdir(default_dir)
+            # the dir exists, but we need to know if there are any valid
+            # certs in there. OpenSSL requires a hashed naming scheme of the
+            # form: [0-9a-f]{8}\.[0-9]
+            # Arguably this is overkill and we could just assume that if the
+            # dir exists it's fine.
+            return any(
+                [re.match(b'^[0-9a-f]{8}\.[0-9]', x) is not None for x in l]
+            )
+        except OSError:
+            return False
+
+    def _check_num_store_objects(self):
+        """
+        Checks how many objects are present in the Context's X509Store.
+
+        :return: int
+        """
+        store = self.get_cert_store()
+        sk_obj = _lib.X509_STORE_get0_objects(store._store)
+        _openssl_assert(sk_obj != _ffi.NULL)
+        return _lib.sk_X509_OBJECT_num(sk_obj)
+
+    def _fallback_default_verify_paths(self, file_path, dir_path):
+        """
+        Default verify paths are based on the compiled version of OpenSSL.
+        However, when pyca/cryptography is compiled as a manylinux1 wheel
+        that compiled location can potentially be wrong. So, like Go, we
+        will try a predefined set of paths and attempt to load roots
+        from there.
+
+        :return: None
+        """
+        for cafile in file_path:
+            if os.path.isfile(cafile):
+                self.load_verify_locations(cafile)
+                break
+
+        for capath in dir_path:
+            if os.path.isdir(capath):
+                self.load_verify_locations(None, capath)
+                break
 
     def use_certificate_chain_file(self, certfile):
         """

tests/test_ssl.py

@@ -6,6 +6,7 @@
 """
 
 import datetime
+import os
 import sys
 import uuid
 
@@ -57,7 +58,7 @@
     Context, ContextType, Session, Connection, ConnectionType, SSLeay_version)
 from OpenSSL.SSL import _make_requires
 
-from OpenSSL._util import lib as _lib
+from OpenSSL._util import ffi as _ffi, lib as _lib
 
 from OpenSSL.SSL import (
     OP_NO_QUERY_MTU, OP_COOKIE_EXCHANGE, OP_NO_TICKET, OP_NO_COMPRESSION,
@@ -1108,6 +1109,95 @@ def test_load_verify_locations_wrong_args(self):
         with pytest.raises(TypeError):
             context.load_verify_locations(object(), object())
 
+    @pytest.mark.skipif(
+        platform != "linux",
+        reason="Loading fallback paths is a linux-specific behavior to "
+        "accommodate pyca/cryptography manylinux1 wheels"
+    )
+    def test_fallback_default_verify_paths(self, monkeypatch):
+        """
+        Test that we load certificates successfully on linux from the fallback
+        path.
+        """
+        context = Context(TLSv1_METHOD)
+        monkeypatch.setattr(
+            _lib, "SSL_CTX_set_default_verify_paths", lambda x: 1
+        )
+        monkeypatch.setattr(context, "_default_dir_exists", lambda x: False)
+        context.set_default_verify_paths()
+        num = context._check_num_store_objects()
+        assert num != 0
+
+    def test_verify_env_vars(self):
+        """
+        Test that we return True/False appropriately if the env vars are set.
+        """
+        context = Context(TLSv1_METHOD)
+        original = dict(os.environ)
+        try:
+            dir_var = "CUSTOM_DIR_VAR"
+            file_var = "CUSTOM_FILE_VAR"
+            os.environ[dir_var] = "value"
+            assert context._verify_env_vars_set(dir_var, file_var) is True
+            os.environ[file_var] = "value"
+            assert context._verify_env_vars_set(dir_var, file_var) is True
+            del os.environ[dir_var]
+            del os.environ[file_var]
+            assert context._verify_env_vars_set(dir_var, file_var) is False
+        finally:
+            os.environ.clear()
+            os.environ.update(original)
+
+    def test_verify_no_fallback_if_env_vars_set(self, monkeypatch):
+        """
+        Test that we don't use the fallback path if env vars are set.
+        """
+        context = Context(TLSv1_METHOD)
+        monkeypatch.setattr(
+            _lib, "SSL_CTX_set_default_verify_paths", lambda x: 1
+        )
+        original = dict(os.environ)
+        dir_env_var = _ffi.string(
+            _lib.X509_get_default_cert_dir_env()
+        ).decode("ascii")
+        file_env_var = _ffi.string(
+            _lib.X509_get_default_cert_file_env()
+        ).decode("ascii")
+        try:
+            os.environ[file_env_var] = "nonsense-value"
+            os.environ[dir_env_var] = "nonsense-value"
+            context.set_default_verify_paths()
+            num = context._check_num_store_objects()
+            assert num == 0
+        finally:
+            os.environ.clear()
+            os.environ.update(original)
+
+    def _fallback_path_is_not_file_or_dir(self, monkeypatch):
+        """
+        Test that when passed empty arrays or paths that do not exist no
+        errors are raised.
+        """
+        context = Context(TLSv1_METHOD)
+        context._fallback_default_verify_paths([], [])
+        context._fallback_default_verify_paths(
+            ["/not/a/file"], ["/not/a/dir"]
+        )
+
+    def test_default_dir_doesnt_exist(self):
+        """
+        Test that we raise catch OSError for both non-existent dirs and paths
+        that are not directories.
+        """
+        context = Context(TLSv1_METHOD)
+        assert context._default_dir_exists(
+            b"/nonexistent/path/probably"
+        ) is False
+
+        assert context._default_dir_exists(
+            os.path.dirname(os.path.abspath(__file__)).encode("ascii")
+        ) is False
+
     @pytest.mark.skipif(
         platform == "win32",
         reason="set_default_verify_paths appears not to work on Windows.  "