Close connection directly when we failed to pack a data type by not sending the corrupted data to server

Author: Zhen

neo4j/bolt/connection.py

@@ -239,7 +239,12 @@ def append(self, signature, fields=(), response=None):
             log_info("C: INIT (%r, {...})", fields[0])
         else:
             raise ValueError("Unknown message signature")
-        self.packer.pack_struct(signature, fields)
+        try:
+            self.packer.pack_struct(signature, fields)
+        except ValueError as e:
+            # We failed to pack this message, therefore we close this connection to avoid sending corrupted data
+            self.close()
+            raise e
         self.output_buffer.chunk()
         self.output_buffer.chunk()
         self.responses.append(response)

neo4j/v1/api.py

@@ -612,7 +612,8 @@ def rollback(self):
     def close(self):
         """ Close this transaction, triggering either a COMMIT or a ROLLBACK.
         """
-        if not self.closed():
+        conn_closed = self.session._connection.closed()
+        if not self.closed() and not conn_closed:
             try:
                 self.sync()
             except CypherError:

test/integration/test_session.py

@@ -17,7 +17,6 @@
 # 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.
-
 from unittest import SkipTest
 from uuid import uuid4
 
@@ -596,7 +595,21 @@ def test_should_sync_after_rollback(self):
             assert len(buffer) == 1
             assert buffer[0][0] == 1
 
-    def test_errors_on_run(self):
+    def test_errors_on_write_transaction(self):
+        session = self.driver.session()
+        with self.assertRaises(ValueError):
+            session.write_transaction(lambda tx, uuid : tx.run("CREATE (a:Thing {uuid:$uuid})", uuid=uuid), uuid4())
+        session.close()
+
+    def test_errors_on_run_transaction(self):
+        session = self.driver.session()
+        tx = session.begin_transaction()
+        with self.assertRaises(ValueError):
+            tx.run("CREATE (a:Thing {uuid:$uuid})", uuid=uuid4())
+        tx.rollback()
+        session.close()
+
+    def test_errors_on_run_session(self):
         session = self.driver.session()
         session.close()
         with self.assertRaises(SessionError):

test/unit/test_packstream.py

@@ -24,6 +24,7 @@
 from io import BytesIO
 from math import pi
 from unittest import TestCase
+from uuid import uuid4
 
 from neo4j.bolt.io import MessageFrame as PyMessageFrame
 from neo4j.packstream.packer import Packer as PyPacker
@@ -273,19 +274,18 @@ def test_map_stream(self):
                                  (unpacked, unpacked_value))
 
     def test_illegal_signature(self):
-        try:
+        with self.assertRaises(ValueError):
             self.assert_packable((b"XXX", ()), b"\xB0XXX")
-        except ValueError:
-            assert True
-        else:
-            assert False
 
     def test_empty_struct(self):
         self.assert_packable((b"X", ()), b"\xB0X")
 
     def test_tiny_struct(self):
         self.assert_packable((b"Z", (u"A", 1)), b"\xB2Z\x81A\x01")
 
+    def test_illegal_uuid(self):
+        with self.assertRaises(ValueError):
+            self.assert_packable(uuid4(), b"\xB0XXX")
 
 try:
     from neo4j.bolt._io import MessageFrame as CMessageFrame