RDBC-788 Secured Changes API

Author: poissoncorp

ravendb/changes/database_changes.py

@@ -1,6 +1,9 @@
+import ssl
 from threading import Lock
 from typing import TYPE_CHECKING, Dict
 
+from websocket import WebSocket
+
 from ravendb.changes.observers import Observable
 from ravendb.changes.types import (
     DocumentChange,
@@ -66,16 +69,16 @@ def do_work(self):
         while not self._closed:
             try:
                 if not self.client_websocket.connected:
-                    # todo: certificates
-                    # if self._request_executor.certificate:
-                    #    if isinstance(self._request_executor.certificate, tuple):
-                    #        (crt, key) = self._request_executor.certificate
-                    #        self.client_websocket.sock_opt.sslopt.update({"certfile": crt, "keyfile": key})
-                    #    else:
-                    #        self.client_websocket.sock_opt.sslopt.update(
-                    #            {"ca_certs": self._request_executor.certificate}
-                    #        )
-                    self.client_websocket.connect(url)
+                    # Get certificate and wrap socket into secured socket
+                    if self._request_executor.certificate_path:
+                        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
+                        ssl_context.load_cert_chain(self._request_executor.certificate_path)
+                        self.client_websocket = WebSocket(sslopt={"context": ssl_context})
+
+                        self.client_websocket.connect(url, suppress_origin=True)
+                    else:
+                        self.client_websocket.connect(url)
+
                     for observables in self._observables.values():
                         for observer in observables.values():
                             observer.set(self._executor.submit(observer.on_connect))

ravendb/serverwide/operations/common.py

@@ -143,7 +143,8 @@ def __init__(
 
         def create_request(self, node: ServerNode) -> requests.Request:
             url = (
-                f"{node.url}/admin/databases?name={self.__database_name}&replicationFactor={self.__replication_factor}"
+                f"{node.url}/admin/databases?name={self.__database_name}"
+                f"&replicationFactor={self.__replication_factor}&?raft-request-id={self.get_raft_unique_request_id}"
             )
 
             request = requests.Request("PUT")

ravendb/tests/jvm_migrated_tests/client_tests/indexing_tests/test_indexes_from_client.py

@@ -173,7 +173,7 @@ def test_can_stop_and_start(self):
 
         self.assertEqual(IndexRunningStatus.RUNNING, status.status)
 
-        self.assertEquals(1, len(status.indexes))
+        self.assertEqual(1, len(status.indexes))
 
         self.assertEqual(IndexRunningStatus.RUNNING, status.indexes[0].status)
 

ravendb/tests/jvm_migrated_tests/issues_tests/test_ravenDB_6967.py

@@ -78,7 +78,7 @@ def test_can_delete_index_errors(self):
         index_errors3 = self.store.maintenance.send(GetIndexErrorsOperation("Index3"))
 
         self.assertGreater(sum([len(x.errors) for x in index_errors1]), 0)
-        self.assertEquals(sum([len(x.errors) for x in index_errors2]), 0)
+        self.assertEqual(sum([len(x.errors) for x in index_errors2]), 0)
         self.assertGreater(sum([len(x.errors) for x in index_errors3]), 0)
 
         self.store.maintenance.send(DeleteIndexErrorsOperation())
@@ -87,8 +87,8 @@ def test_can_delete_index_errors(self):
         index_errors2 = self.store.maintenance.send(GetIndexErrorsOperation("Index2"))
         index_errors3 = self.store.maintenance.send(GetIndexErrorsOperation("Index3"))
 
-        self.assertEquals(sum([len(x.errors) for x in index_errors1]), 0)
-        self.assertEquals(sum([len(x.errors) for x in index_errors2]), 0)
-        self.assertEquals(sum([len(x.errors) for x in index_errors3]), 0)
+        self.assertEqual(sum([len(x.errors) for x in index_errors1]), 0)
+        self.assertEqual(sum([len(x.errors) for x in index_errors2]), 0)
+        self.assertEqual(sum([len(x.errors) for x in index_errors3]), 0)
 
         RavenTestHelper.assert_no_index_errors(self.store)

ravendb/tests/jvm_migrated_tests/server_tests/documents/notifications/test_changes.py

@@ -3,7 +3,7 @@
 
 from ravendb import AbstractIndexCreationTask, SetIndexesPriorityOperation
 from ravendb.changes.observers import ActionObserver
-from ravendb.changes.types import DocumentChange, IndexChange
+from ravendb.changes.types import DocumentChange, IndexChange, DocumentChangeType
 from ravendb.documents.indexes.definitions import IndexPriority
 from ravendb.infrastructure.entities import User
 from ravendb.infrastructure.orders import Order
@@ -235,3 +235,62 @@ def __ev(value: DocumentChange):
         self.assertEqual("users/2", document_changes[1].key)
 
         close_action()
+
+    def test_changes_with_https(self):
+        event = Event()
+        changes_list = []
+        exception = None
+
+        def _on_error(e):
+            nonlocal exception
+            exception = e
+
+        changes = self.secured_document_store.changes(on_error=_on_error)
+        observable = changes.for_document("users/1")
+
+        def __ev(value: DocumentChange):
+            changes_list.append(value)
+            event.set()
+
+        observer = ActionObserver(__ev)
+        close_action = observable.subscribe_with_observer(observer)
+        try:
+            observable.ensure_subscribe_now()
+        except Exception:
+            raise exception
+
+        with self.secured_document_store.open_session() as session:
+            user = User()
+            session.store(user, "users/1")
+            session.save_changes()
+
+        event.wait(2)
+        document_change = changes_list[0]
+        self.assertIsNotNone(document_change)
+        self.assertEqual("users/1", document_change.key)
+        self.assertEqual(DocumentChangeType.PUT, document_change.type_of_change)
+
+        changes_list.clear()
+
+        try:
+            event.wait(1)
+        except Exception:
+            pass
+
+        self.assertEqual(0, len(changes_list))
+        close_action()
+        # at this point we should be unsubscribed from changes on 'users/1'
+
+        with self.secured_document_store.open_session() as session:
+            user = User()
+            user.name = "another name"
+            session.store(user, "users/1")
+            session.save_changes()
+
+        # it should be empty
+        try:
+            event.wait(1)
+        except Exception:
+            pass
+
+        self.assertEqual(0, len(changes_list))

ravendb/util/tcp_utils.py

@@ -17,11 +17,13 @@ def connect(
     ) -> socket.socket:
         hostname, port = url_string.replace("tcp://", "").split(":")
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
         is_ssl_socket = server_certificate_base64 and client_certificate_pem_path
-        if server_certificate_base64 and client_certificate_pem_path:
+        if is_ssl_socket:
             context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
             context.load_cert_chain(client_certificate_pem_path, password=certificate_private_key_password)
             s = context.wrap_socket(s)
+
         s.connect((hostname, int(port)))
         if is_ssl_socket and base64.b64decode(server_certificate_base64) != s.getpeercert(True):
             raise ConnectionError("Failed to validate public server certificate.")