Add tests for Bolt fix for broken DateTime encoding (#470) (#476)

The structures with signature `0x46` and `0x66` are being replaced by `0x49` and `0x69`. This new structures changes the meaning of seconds and nano seconds from `adjusted Unix epoch`  to `UTC`.

This changes have with goal of avoiding un-existing or ambiguous ZonedDateTime to be received or sent over Bolt.

Bolt v4.3 and v4.4 were patched to support this feature if the server supports the patch.

Co-authored-by: Rouven Bauer <[email protected]>
Co-authored-by: grant lodge <[email protected]>
Co-authored-by: Florent Biville <[email protected]>
Co-authored-by: Dmitriy Tverdiakov <[email protected]>

Author: 5 people

boltstub/bolt_protocol.py

@@ -5,12 +5,18 @@
     ServerExit,
 )
 from .packstream import Structure
-from .simple_jolt import dumps_simple as jolt_dumps
+from .simple_jolt import v1 as jolt_v1
+from .simple_jolt import v2 as jolt_v2
 from .util import (
     hex_repr,
     recursive_subclasses,
 )
 
+jolt_package = {
+    1: jolt_v1,
+    2: jolt_v2,
+}
+
 
 def get_bolt_protocol(version):
     if version is None:
@@ -27,33 +33,21 @@ def get_bolt_protocol(version):
 def verify_script_messages(script):
     protocol = get_bolt_protocol(script.context.bolt_version)
     for line in script.client_lines:
-        if line.parsed[0] not in protocol.messages["C"].values():
-            raise BoltUnknownMessageError(
-                "Unsupported client message {} for BOLT version {}. "
-                "Must be one of {}".format(
-                    line.parsed[0], script.context.bolt_version,
-                    list(protocol.messages["C"].values())
-                ),
-                line
-            )
+        # will raise an exception if the message is unknown or the body
+        # cannot be decoded
+        protocol.translate_client_line(line)
     for line in script.server_lines:
-        if line.parsed[0] is None:
+        if line.is_command:
             continue  # this server line contains a command, not a message
-        if line.parsed[0] not in protocol.messages["S"].values():
-            raise BoltUnknownMessageError(
-                "Unsupported server message {} for BOLT version {}. "
-                "Must be one of {}".format(
-                    line.parsed[0], script.context.bolt_version,
-                    list(protocol.messages["S"].values())
-                ),
-                line
-            )
+        # will raise an exception if the message is unknown
+        protocol.translate_server_line(line)
 
 
 class TranslatedStructure(Structure):
-    def __init__(self, name, tag, *fields):
+    def __init__(self, name, tag, *fields, packstream_version):
         # Verified is false as this class in only used for message structs
-        super().__init__(tag, *fields, verified=False)
+        super().__init__(tag, *fields, packstream_version=packstream_version,
+                         verified=False)
         self.name = name
 
     def __repr__(self):
@@ -62,7 +56,8 @@ def __repr__(self):
 
     def __str__(self):
         return self.name + " {}".format(" ".join(
-            map(jolt_dumps, self.fields_to_jolt_types())
+            map(jolt_package[self.packstream_version].dumps_simple,
+                self.fields_to_jolt_types())
         ))
 
     def __eq__(self, other):
@@ -80,6 +75,8 @@ class BoltProtocol:
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = None
+
     messages = {
         "C": {},
         "S": {},
@@ -94,25 +91,55 @@ def decode_versions(cls, b):
             for minor in range(spec_minor, spec_minor - range_ - 1, -1):
                 yield major, minor
 
+    @classmethod
+    def translate_client_line(cls, client_line):
+        if not client_line.jolt_parsed:
+            client_line.parse_jolt(cls.get_jolt_package())
+        name, fields = client_line.jolt_parsed
+        try:
+            tag = next(tag_ for tag_, name_ in cls.messages["C"].items()
+                       if name == name_)
+        except StopIteration:
+            raise BoltUnknownMessageError(
+                "Unsupported client message {} for BOLT version {}. "
+                "Must be one of {}".format(
+                    name, cls.protocol_version,
+                    list(cls.messages["C"].values())
+                ),
+                client_line
+            )
+        return TranslatedStructure(
+            name, tag, *fields, packstream_version=cls.packstream_version
+        )
+
     @classmethod
     def translate_server_line(cls, server_line):
-        name, fields = server_line.parsed
+        if not server_line.jolt_parsed:
+            server_line.parse_jolt(cls.get_jolt_package())
+        name, fields = server_line.jolt_parsed
         try:
             tag = next(tag_ for tag_, name_ in cls.messages["S"].items()
                        if name == name_)
-        except StopIteration as e:
-            raise ValueError(
-                "Unknown response message type {} in Bolt version {}".format(
-                    name, ".".join(map(str, cls.protocol_version))
-                )
-            ) from e
-        return TranslatedStructure(name, tag, *fields)
+        except StopIteration:
+            raise BoltUnknownMessageError(
+                "Unsupported server message {} for BOLT version {}. "
+                "Must be one of {}".format(
+                    name, cls.protocol_version,
+                    list(cls.messages["S"].values())
+                ),
+                server_line
+            )
+        return TranslatedStructure(
+            name, tag, *fields, packstream_version=cls.packstream_version
+        )
 
     @classmethod
     def translate_structure(cls, structure: Structure):
         try:
-            return TranslatedStructure(cls.messages["C"][structure.tag],
-                                       structure.tag, *structure.fields)
+            return TranslatedStructure(
+                cls.messages["C"][structure.tag], structure.tag,
+                *structure.fields, packstream_version=cls.packstream_version
+            )
         except KeyError:
             raise ServerExit(
                 "Unknown response message type {} in Bolt version {}".format(
@@ -121,6 +148,10 @@ def translate_structure(cls, structure: Structure):
                 )
             )
 
+    @classmethod
+    def get_jolt_package(cls):
+        return jolt_package[cls.packstream_version]
+
 
 class Bolt1Protocol(BoltProtocol):
 
@@ -129,6 +160,8 @@ class Bolt1Protocol(BoltProtocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = 1
+
     messages = {
         "C": {
             b"\x01": "INIT",
@@ -159,11 +192,15 @@ def decode_versions(cls, b):
     @classmethod
     def get_auto_response(cls, request: TranslatedStructure):
         if request.tag == b"\x01":
-            return TranslatedStructure("SUCCESS", b"\x70", {
-                "server": cls.server_agent,
-            })
+            return TranslatedStructure(
+                "SUCCESS", b"\x70", {"server": cls.server_agent},
+                packstream_version=cls.packstream_version
+            )
         else:
-            return TranslatedStructure("SUCCESS", b"\x70", {})
+            return TranslatedStructure(
+                "SUCCESS", b"\x70", {},
+                packstream_version=cls.packstream_version
+            )
 
 
 class Bolt2Protocol(Bolt1Protocol):
@@ -173,16 +210,22 @@ class Bolt2Protocol(Bolt1Protocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = 1
+
     server_agent = "Neo4j/3.4.0"
 
     @classmethod
     def get_auto_response(cls, request: TranslatedStructure):
         if request.tag == b"\x01":
-            return TranslatedStructure("SUCCESS", b"\x70", {
-                "server": cls.server_agent,
-            })
+            return TranslatedStructure(
+                "SUCCESS", b"\x70", {"server": cls.server_agent},
+                packstream_version=cls.packstream_version
+            )
         else:
-            return TranslatedStructure("SUCCESS", b"\x70", {})
+            return TranslatedStructure(
+                "SUCCESS", b"\x70", {},
+                packstream_version=cls.packstream_version
+            )
 
 
 class Bolt3Protocol(Bolt2Protocol):
@@ -192,6 +235,8 @@ class Bolt3Protocol(Bolt2Protocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = 1
+
     messages = {
         "C": {
             b"\x01": "HELLO",
@@ -217,12 +262,16 @@ class Bolt3Protocol(Bolt2Protocol):
     @classmethod
     def get_auto_response(cls, request: TranslatedStructure):
         if request.tag == b"\x01":
-            return TranslatedStructure("SUCCESS", b"\x70", {
-                "connection_id": "bolt-0",
-                "server": cls.server_agent,
-            })
+            return TranslatedStructure(
+                "SUCCESS", b"\x70",
+                {"connection_id": "bolt-0", "server": cls.server_agent},
+                packstream_version=cls.packstream_version
+            )
         else:
-            return TranslatedStructure("SUCCESS", b"\x70", {})
+            return TranslatedStructure(
+                "SUCCESS", b"\x70", {},
+                packstream_version=cls.packstream_version
+            )
 
 
 class Bolt4x0Protocol(Bolt3Protocol):
@@ -232,6 +281,8 @@ class Bolt4x0Protocol(Bolt3Protocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = 1
+
     messages = {
         "C": {
             b"\x01": "HELLO",
@@ -265,12 +316,16 @@ def decode_versions(cls, b):
     @classmethod
     def get_auto_response(cls, request: TranslatedStructure):
         if request.tag == b"\x01":
-            return TranslatedStructure("SUCCESS", b"\x70", {
-                "connection_id": "bolt-0",
-                "server": cls.server_agent,
-            })
+            return TranslatedStructure(
+                "SUCCESS", b"\x70",
+                {"connection_id": "bolt-0", "server": cls.server_agent},
+                packstream_version=cls.packstream_version
+            )
         else:
-            return TranslatedStructure("SUCCESS", b"\x70", {})
+            return TranslatedStructure(
+                "SUCCESS", b"\x70", {},
+                packstream_version=cls.packstream_version
+            )
 
 
 class Bolt4x1Protocol(Bolt4x0Protocol):
@@ -280,6 +335,8 @@ class Bolt4x1Protocol(Bolt4x0Protocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = 1
+
     messages = {
         "C": {
             b"\x01": "HELLO",
@@ -305,13 +362,19 @@ class Bolt4x1Protocol(Bolt4x0Protocol):
     @classmethod
     def get_auto_response(cls, request: TranslatedStructure):
         if request.tag == b"\x01":
-            return TranslatedStructure("SUCCESS", b"\x70", {
-                "connection_id": "bolt-0",
-                "server": cls.server_agent,
-                "routing": None,
-            })
+            return TranslatedStructure(
+                "SUCCESS", b"\x70",
+                {
+                    "connection_id": "bolt-0", "server": cls.server_agent,
+                    "routing": None,
+                },
+                packstream_version=cls.packstream_version
+            )
         else:
-            return TranslatedStructure("SUCCESS", b"\x70", {})
+            return TranslatedStructure(
+                "SUCCESS", b"\x70", {},
+                packstream_version=cls.packstream_version
+            )
 
 
 class Bolt4x2Protocol(Bolt4x1Protocol):
@@ -321,6 +384,8 @@ class Bolt4x2Protocol(Bolt4x1Protocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = {(4, 1)}
 
+    packstream_version = 1
+
     server_agent = "Neo4j/4.2.0"
 
 
@@ -331,6 +396,8 @@ class Bolt4x3Protocol(Bolt4x2Protocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = 1
+
     messages = {
         "C": {
             b"\x01": "HELLO",
@@ -370,4 +437,18 @@ class Bolt4x4Protocol(Bolt4x3Protocol):
     # allow the server to negotiate other bolt versions
     equivalent_versions = set()
 
+    packstream_version = 1
+
     server_agent = "Neo4j/4.4.0"
+
+
+class Bolt5x0Protocol(Bolt4x4Protocol):
+
+    protocol_version = (5, 0)
+    version_aliases = set()
+    # allow the server to negotiate other bolt versions
+    equivalent_versions = set()
+
+    packstream_version = 2
+
+    server_agent = "Neo4j/5.0.0"

boltstub/channel.py

@@ -8,10 +8,13 @@
 
 
 class Channel:
+    # This class is the glue between a stub script, the socket, and the bolt
+    # protocol.
+
     def __init__(self, wire, bolt_version, log_cb=None, handshake_data=None):
         self.wire = wire
-        self.stream = PackStream(wire)
         self.bolt_protocol = get_bolt_protocol(bolt_version)
+        self.stream = PackStream(wire, self.bolt_protocol.packstream_version)
         self.log = log_cb
         self.handshake_data = handshake_data
         self._buffered_msg = None
@@ -69,6 +72,9 @@ def version_handshake(self):
         self.wire.send()
         self._log("S: <HANDSHAKE> %s", hex_repr(response))
 
+    def match_client_line(self, client_line, msg):
+        return client_line.match_message(msg.name, msg.fields)
+
     def send_raw(self, b):
         self.log("%s", hex_repr(b))
         self.wire.write(b)