Re-design SSL config options

CHANGELOG.md

@@ -15,8 +15,8 @@
   or call the `dbms.components` procedure instead.
 - SSL configuration options have been changed:
   - `trust` has been deprecated and will be removed in a future release.  
-    Use `trusted_certificates` instead which expects `None` or a `list`. See the
-    API documentation for more details.
+    Use `trusted_certificates` instead.
+    See the API documentation for more details.
 - `neo4j.time` module:
   - `Duration`
     - The constructor does not accept `subseconds` anymore.  

docs/source/api.rst

@@ -330,25 +330,14 @@ Specify how to determine the authenticity of encryption certificates provided by
 This setting does not have any effect if ``encrypted`` is set to ``False`` or a
 custom ``ssl_context`` is configured.
 
-:Type: :class:`list` or :const:`None`
+:Type: :class:`.TrustSystemCAs`, :class:`.TrustAll`, or :class:`.TrustCustomCAs`
+:Default: :const:`neo4j.TrustSystemCAs()`
 
-**None** (default)
-    Trust server certificates that can be verified against the system
-    certificate authority. This option is primarily intended for use with
-    full certificates.
+.. autoclass:: neo4j.TrustSystemCAs
 
-**[] (empty list)**
-    Trust any server certificate. This ensures that communication
-    is encrypted but does not verify the server certificate against a
-    certificate authority. This option is primarily intended for use with
-    the default auto-generated server certificate.
+.. autoclass:: neo4j.TrustAll
 
-**["<path>", ...]**
-    Trust server certificates that can be verified against the certificate
-    authority at the specified paths. This option is primarily intended for
-    self-signed and custom certificates.
-
-:Default: :const:`None`
+.. autoclass:: neo4j.TrustCustomCAs
 
 .. versionadded:: 5.0
 

neo4j/__init__.py

@@ -22,8 +22,8 @@
     "AsyncBoltDriver",
     "AsyncDriver",
     "AsyncGraphDatabase",
-    "AsyncNeo4jDriver",
     "AsyncManagedTransaction",
+    "AsyncNeo4jDriver",
     "AsyncResult",
     "AsyncSession",
     "AsyncTransaction",
@@ -58,6 +58,9 @@
     "Transaction",
     "TRUST_ALL_CERTIFICATES",
     "TRUST_SYSTEM_CA_SIGNED_CERTIFICATES",
+    "TrustAll",
+    "TrustCustomCAs",
+    "TrustSystemCAs",
     "unit_of_work",
     "Version",
     "WorkspaceConfig",
@@ -79,6 +82,11 @@
     AsyncSession,
     AsyncTransaction,
 )
+from ._conf import (
+    TrustAll,
+    TrustCustomCAs,
+    TrustSystemCAs,
+)
 from ._sync.driver import (
     BoltDriver,
     Driver,

neo4j/_async/driver.py

@@ -16,9 +16,11 @@
 # limitations under the License.
 
 
-import asyncio
-
 from .._async_compat.util import AsyncUtil
+from .._conf import (
+    TrustAll,
+    TrustStore,
+)
 from ..addressing import Address
 from ..api import (
     READ_ACCESS,
@@ -31,10 +33,6 @@
     SessionConfig,
     WorkspaceConfig,
 )
-from ..exceptions import (
-    ServiceUnavailable,
-    SessionExpired,
-)
 from ..meta import (
     deprecation_warn,
     experimental,
@@ -66,7 +64,6 @@ def driver(cls, uri, *, auth=None, **config):
             DRIVER_NEO4j,
             parse_neo4j_uri,
             parse_routing_context,
-            SECURITY_TYPE_NOT_SECURE,
             SECURITY_TYPE_SECURE,
             SECURITY_TYPE_SELF_SIGNED_CERTIFICATE,
             URI_SCHEME_BOLT,
@@ -94,6 +91,17 @@ def driver(cls, uri, *, auth=None, **config):
                     )
                 )
 
+        if ("trusted_certificates" in config.keys()
+            and not isinstance(config["trusted_certificates"],
+                               TrustStore)):
+            raise ConnectionError(
+                "The config setting `trusted_certificates` must be of type "
+                "neo4j.TrustAll, neo4j.TrustCustomCAs, or"
+                "neo4j.TrustSystemCAs but was {}".format(
+                    type(config["trusted_certificates"])
+                )
+            )
+
         if (security_type in [SECURITY_TYPE_SELF_SIGNED_CERTIFICATE, SECURITY_TYPE_SECURE]
             and ("encrypted" in config.keys()
                  or "trust" in config.keys()
@@ -125,7 +133,7 @@ def driver(cls, uri, *, auth=None, **config):
             config["encrypted"] = True
         elif security_type == SECURITY_TYPE_SELF_SIGNED_CERTIFICATE:
             config["encrypted"] = True
-            config["trusted_certificates"] = []
+            config["trusted_certificates"] = TrustAll()
 
         if driver_type == DRIVER_BOLT:
             if parse_routing_context(parsed.query):

neo4j/_conf.py

@@ -0,0 +1,80 @@
+# Copyright (c) "Neo4j"
+# Neo4j Sweden AB [http://neo4j.com]
+#
+# This file is part of Neo4j.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class TrustStore:
+    # Base class for trust stores. For internal type-checking only.
+    pass
+
+
+class TrustSystemCAs(TrustStore):
+    """Used to configure the driver to trust system CAs (default).
+
+    Trust server certificates that can be verified against the system
+    certificate authority. This option is primarily intended for use with
+    full certificates.
+
+    For example::
+
+        driver = neo4j.GraphDatabase.driver(
+            url, auth=auth, trusted_certificates=neo4j.TrustSystemCAs()
+        )
+    """
+    pass
+
+
+class TrustAll(TrustStore):
+    """Used to configure the driver to trust all certificates.
+
+    Trust any server certificate. This ensures that communication
+    is encrypted but does not verify the server certificate against a
+    certificate authority. This option is primarily intended for use with
+    the default auto-generated server certificate.
+
+
+    For example::
+
+        driver = neo4j.GraphDatabase.driver(
+            url, auth=auth, trusted_certificates=neo4j.TrustAll()
+        )
+    """
+    pass
+
+
+class TrustCustomCAs(TrustStore):
+    """Used to configure the driver to trust custom CAs.
+
+    Trust server certificates that can be verified against the certificate
+    authority at the specified paths. This option is primarily intended for
+    self-signed and custom certificates.
+
+    :param certificates (str): paths to the certificates to trust.
+        Those are not the certificates you expect to see from the server but
+        the CA certificates you expect to be used to sign the server's
+        certificate.
+
+    For example::
+
+        driver = neo4j.GraphDatabase.driver(
+            url, auth=auth,
+            trusted_certificates=neo4j.TrustCustomCAs(
+                "/path/to/ca1.crt", "/path/to/ca2.crt",
+            )
+        )
+    """
+    def __init__(self, *certificates):
+        self.certs = certificates

neo4j/_sync/driver.py

@@ -16,9 +16,11 @@
 # limitations under the License.
 
 
-import asyncio
-
 from .._async_compat.util import Util
+from .._conf import (
+    TrustAll,
+    TrustStore,
+)
 from ..addressing import Address
 from ..api import (
     READ_ACCESS,
@@ -31,10 +33,6 @@
     SessionConfig,
     WorkspaceConfig,
 )
-from ..exceptions import (
-    ServiceUnavailable,
-    SessionExpired,
-)
 from ..meta import (
     deprecation_warn,
     experimental,
@@ -66,7 +64,6 @@ def driver(cls, uri, *, auth=None, **config):
             DRIVER_NEO4j,
             parse_neo4j_uri,
             parse_routing_context,
-            SECURITY_TYPE_NOT_SECURE,
             SECURITY_TYPE_SECURE,
             SECURITY_TYPE_SELF_SIGNED_CERTIFICATE,
             URI_SCHEME_BOLT,
@@ -94,6 +91,17 @@ def driver(cls, uri, *, auth=None, **config):
                     )
                 )
 
+        if ("trusted_certificates" in config.keys()
+            and not isinstance(config["trusted_certificates"],
+                               TrustStore)):
+            raise ConnectionError(
+                "The config setting `trusted_certificates` must be of type "
+                "neo4j.TrustAll, neo4j.TrustCustomCAs, or"
+                "neo4j.TrustSystemCAs but was {}".format(
+                    type(config["trusted_certificates"])
+                )
+            )
+
         if (security_type in [SECURITY_TYPE_SELF_SIGNED_CERTIFICATE, SECURITY_TYPE_SECURE]
             and ("encrypted" in config.keys()
                  or "trust" in config.keys()
@@ -125,7 +133,7 @@ def driver(cls, uri, *, auth=None, **config):
             config["encrypted"] = True
         elif security_type == SECURITY_TYPE_SELF_SIGNED_CERTIFICATE:
             config["encrypted"] = True
-            config["trusted_certificates"] = []
+            config["trusted_certificates"] = TrustAll()
 
         if driver_type == DRIVER_BOLT:
             if parse_routing_context(parsed.query):

neo4j/conf.py

@@ -18,8 +18,12 @@
 
 from abc import ABCMeta
 from collections.abc import Mapping
-from warnings import warn
 
+from ._conf import (
+    TrustAll,
+    TrustCustomCAs,
+    TrustSystemCAs,
+)
 from .api import (
     DEFAULT_DATABASE,
     TRUST_ALL_CERTIFICATES,
@@ -205,9 +209,9 @@ def __iter__(self):
 
 def _trust_to_trusted_certificates(pool_config, trust):
     if trust == TRUST_SYSTEM_CA_SIGNED_CERTIFICATES:
-        pool_config.trusted_certificates = None
+        pool_config.trusted_certificates = TrustSystemCAs()
     elif trust == TRUST_ALL_CERTIFICATES:
-        pool_config.trusted_certificates = []
+        pool_config.trusted_certificates = TrustAll()
 
 
 class PoolConfig(Config):
@@ -241,12 +245,13 @@ class PoolConfig(Config):
     # Specify whether to use an encrypted connection between the driver and server.
 
     #: SSL Certificates to Trust
-    trusted_certificates = None
+    trusted_certificates = TrustSystemCAs()
     # Specify how to determine the authenticity of encryption certificates
     # provided by the Neo4j instance on connection.
-    # * None: Use system trust store. (default)
-    # * []: Trust any certificate.
-    # * ["<path>", ...]: Trust the specified certificate(s).
+    # * `neo4j.TrustSystemCAs()`: Use system trust store. (default)
+    # * `neo4j.TrustAll()`: Trust any certificate.
+    # * `neo4j.TrustCustomCAs("<path>", ...)`:
+    #       Trust the specified certificate(s).
 
     #: Custom SSL context to use for wrapping sockets
     ssl_context = None
@@ -296,26 +301,25 @@ def get_ssl_context(self):
         ssl_context.options |= ssl.OP_NO_TLSv1      # Python 3.2
         ssl_context.options |= ssl.OP_NO_TLSv1_1    # Python 3.4
 
-        if self.trusted_certificates is None:
+        if isinstance(self.trusted_certificates, TrustAll):
+            # trust any certificate
+            ssl_context.check_hostname = False
+            # https://docs.python.org/3.7/library/ssl.html#ssl.CERT_NONE
+            ssl_context.verify_mode = ssl.CERT_NONE
+        elif isinstance(self.trusted_certificates, TrustCustomCAs):
+            # trust the specified certificate(s)
+            ssl_context.check_hostname = True
+            ssl_context.verify_mode = ssl.CERT_REQUIRED
+            for cert in self.trusted_certificates.certs:
+                ssl_context.load_verify_locations(cert)
+        else:
+            # default
             # trust system CA certificates
             ssl_context.check_hostname = True
             ssl_context.verify_mode = ssl.CERT_REQUIRED
             # Must be load_default_certs, not set_default_verify_paths to
             # work on Windows with system CAs.
             ssl_context.load_default_certs()
-        else:
-            self.trusted_certificates = tuple(self.trusted_certificates)
-            if not self.trusted_certificates:
-                # trust any certificate
-                ssl_context.check_hostname = False
-                # https://docs.python.org/3.7/library/ssl.html#ssl.CERT_NONE
-                ssl_context.verify_mode = ssl.CERT_NONE
-            else:
-                # trust the specified certificate(s)
-                ssl_context.check_hostname = True
-                ssl_context.verify_mode = ssl.CERT_REQUIRED
-                for cert in self.trusted_certificates:
-                    ssl_context.load_verify_locations(cert)
 
         return ssl_context
 

testkitbackend/_async/requests.py

@@ -103,12 +103,14 @@ async def NewDriver(backend, data):
     if "encrypted" in data:
         kwargs["encrypted"] = data["encrypted"]
     if "trustedCertificates" in data:
-        kwargs["trusted_certificates"] = data["trustedCertificates"]
-        if isinstance(kwargs["trusted_certificates"], list):
-            kwargs["trusted_certificates"] = [
-                "/usr/local/share/custom-ca-certificates/" + cert
-                for cert in kwargs["trusted_certificates"]
-            ]
+        if data["trustedCertificates"] is None:
+            kwargs["trusted_certificates"] = neo4j.TrustSystemCAs()
+        elif not data["trustedCertificates"]:
+            kwargs["trusted_certificates"] = neo4j.TrustAll()
+        else:
+            cert_paths = ("/usr/local/share/custom-ca-certificates/" + cert
+                          for cert in data["trustedCertificates"])
+            kwargs["trusted_certificates"] = neo4j.TrustCustomCAs(*cert_paths)
 
     data.mark_item_as_read("domainNameResolverRegistered")
     driver = neo4j.AsyncGraphDatabase.driver(