Permit SSL to be disabled gracefully in older Python versions

Author: technige

examples/test_examples.py

@@ -19,9 +19,9 @@
 # limitations under the License.
 
 
-from unittest import skip
+from unittest import skip, skipUnless
 
-from neo4j.v1 import TRUST_ON_FIRST_USE, TRUST_SIGNED_CERTIFICATES
+from neo4j.v1 import TRUST_ON_FIRST_USE, TRUST_SIGNED_CERTIFICATES, SSL_AVAILABLE
 from test.util import ServerTestCase
 
 # Do not change the contents of this tagged section without good reason*
@@ -74,11 +74,13 @@ def test_configuration(self):
         # end::configuration[]
         return driver
 
+    @skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
     def test_tls_require_encryption(self):
         # tag::tls-require-encryption[]
         driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"), encrypted=True)
         # end::tls-require-encryption[]
 
+    @skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
     def test_tls_trust_on_first_use(self):
         # tag::tls-trust-on-first-use[]
         driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "password"), encrypted=True, trust=TRUST_ON_FIRST_USE)

neo4j/v1/connection.py

@@ -29,14 +29,14 @@
 from os.path import dirname, isfile
 from select import select
 from socket import create_connection, SHUT_RDWR, error as SocketError
-from ssl import HAS_SNI, SSLError
 from struct import pack as struct_pack, unpack as struct_unpack, unpack_from as struct_unpack_from
 
 from .constants import DEFAULT_PORT, DEFAULT_USER_AGENT, KNOWN_HOSTS, MAGIC_PREAMBLE, \
     TRUST_DEFAULT, TRUST_ON_FIRST_USE
 from .compat import hex2
 from .exceptions import ProtocolError
 from .packstream import Packer, Unpacker
+from .ssl_compat import SSL_AVAILABLE, HAS_SNI, SSLError
 
 
 # Signature bytes for each message type
@@ -385,7 +385,7 @@ def connect(host, port=None, ssl_context=None, **config):
             raise
 
     # Secure the connection if an SSL context has been provided
-    if ssl_context:
+    if ssl_context and SSL_AVAILABLE:
         if __debug__: log_info("~~ [SECURE] %s", host)
         try:
             s = ssl_context.wrap_socket(s, server_hostname=host if HAS_SNI else None)

neo4j/v1/constants.py

@@ -22,6 +22,7 @@
 from os.path import expanduser, join
 
 from ..meta import version
+from .ssl_compat import SSL_AVAILABLE
 
 
 DEFAULT_PORT = 7687
@@ -31,7 +32,7 @@
 
 MAGIC_PREAMBLE = 0x6060B017
 
-ENCRYPTED_DEFAULT = True
+ENCRYPTED_DEFAULT = SSL_AVAILABLE
 
 TRUST_ON_FIRST_USE = 0
 TRUST_SIGNED_CERTIFICATES = 1

neo4j/v1/session.py

@@ -29,12 +29,12 @@ class which can be used to obtain `Driver` instances that are used for
 from __future__ import division
 
 from collections import deque, namedtuple
-from ssl import SSLContext, PROTOCOL_SSLv23, OP_NO_SSLv2, CERT_REQUIRED
 
 from .compat import integer, string, urlparse
 from .connection import connect, Response, RUN, PULL_ALL
 from .constants import ENCRYPTED_DEFAULT, TRUST_DEFAULT, TRUST_SIGNED_CERTIFICATES
 from .exceptions import CypherError
+from .ssl_compat import SSL_AVAILABLE, SSLContext, PROTOCOL_SSLv23, OP_NO_SSLv2, CERT_REQUIRED
 from .types import hydrated
 
 
@@ -102,6 +102,8 @@ def __init__(self, url, **config):
         self.encrypted = encrypted = config.get("encrypted", ENCRYPTED_DEFAULT)
         self.trust = trust = config.get("trust", TRUST_DEFAULT)
         if encrypted:
+            if not SSL_AVAILABLE:
+                raise RuntimeError("Bolt over TLS is only available in Python 2.7.9+ and Python 3.3+")
             ssl_context = SSLContext(PROTOCOL_SSLv23)
             ssl_context.options |= OP_NO_SSLv2
             if trust >= TRUST_SIGNED_CERTIFICATES:

neo4j/v1/ssl_compat.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+
+# Copyright (c) 2002-2016 "Neo Technology,"
+# Network Engine for Objects in Lund AB [http://neotechnology.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.
+
+try:
+    from ssl import SSLContext, PROTOCOL_SSLv23, OP_NO_SSLv2, CERT_REQUIRED, HAS_SNI, SSLError
+except ImportError:
+    SSL_AVAILABLE = False
+    SSLContext = None
+    PROTOCOL_SSLv23 = None
+    OP_NO_SSLv2 = None
+    CERT_REQUIRED = None
+    HAS_SNI = None
+    SSLError = None
+else:
+    SSL_AVAILABLE = True

test/test_session.py

@@ -21,11 +21,13 @@
 
 from socket import socket
 from ssl import SSLSocket
+from unittest import skipUnless
 
 from mock import patch
+
 from neo4j.v1.constants import TRUST_ON_FIRST_USE
 from neo4j.v1.exceptions import CypherError
-from neo4j.v1.session import GraphDatabase, basic_auth, Record
+from neo4j.v1.session import GraphDatabase, basic_auth, Record, SSL_AVAILABLE
 from neo4j.v1.types import Node, Relationship, Path
 
 from test.util import ServerTestCase
@@ -90,10 +92,6 @@ def test_sessions_are_not_reused_if_still_in_use(self):
 
 class SecurityTestCase(ServerTestCase):
 
-    def test_default_session_uses_tofu(self):
-        driver = GraphDatabase.driver("bolt://localhost")
-        assert driver.trust == TRUST_ON_FIRST_USE
-
     def test_insecure_session_uses_normal_socket(self):
         driver = GraphDatabase.driver("bolt://localhost",  auth=auth_token, encrypted=False)
         session = driver.session()
@@ -102,6 +100,7 @@ def test_insecure_session_uses_normal_socket(self):
         assert connection.der_encoded_server_certificate is None
         session.close()
 
+    @skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
     def test_tofu_session_uses_secure_socket(self):
         driver = GraphDatabase.driver("bolt://localhost", auth=auth_token, encrypted=True, trust=TRUST_ON_FIRST_USE)
         session = driver.session()
@@ -110,6 +109,7 @@ def test_tofu_session_uses_secure_socket(self):
         assert connection.der_encoded_server_certificate is not None
         session.close()
 
+    @skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
     def test_tofu_session_trusts_certificate_after_first_use(self):
         driver = GraphDatabase.driver("bolt://localhost",  auth=auth_token, encrypted=True, trust=TRUST_ON_FIRST_USE)
         session = driver.session()