Fix: prevent hanging state by forwarding protocol-level parsing errors into the message stream. (#388)

Author: isoos

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 3.4.3
+
+- Fix: prevent hanging state by forwarding protocol-level parsing errors into the message stream.
+
 ## 3.4.2
 
 - Fix: When a transaction is rolled back, do not expose the exception on rollback, rather the original exception from the transaction.

lib/src/messages/server_messages.dart

@@ -182,6 +182,8 @@ class DataRowMessage extends ServerMessage {
         values.add(Uint8List(0));
       } else if (dataSize == -1) {
         values.add(null);
+      } else if (dataSize < -1) {
+        throw AssertionError('Bad data size for field $i: $dataSize');
       } else {
         final rawBytes = reader.read(dataSize);
         values.add(rawBytes);

lib/src/v3/protocol.dart

@@ -70,8 +70,12 @@ StreamTransformer<Uint8List, ServerMessage> _readMessages(
       }
 
       Future<void> handleChunk(Uint8List bytes) async {
-        await framer.addBytes(bytes);
-        emitFinishedMessages();
+        try {
+          await framer.addBytes(bytes);
+          emitFinishedMessages();
+        } catch (e, st) {
+          listener.addErrorSync(e, st);
+        }
       }
 
       // Don't cancel this subscription on error! If the listener wants that,

pubspec.yaml

@@ -1,6 +1,6 @@
 name: postgres
 description: PostgreSQL database driver. Supports statement reuse and binary protocol and connection pooling.
-version: 3.4.2
+version: 3.4.3
 homepage: https://github.com/isoos/postgresql-dart
 topics:
   - sql

test/unexpected_protocol_bytes_test.dart

@@ -0,0 +1,101 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:postgres/postgres.dart';
+import 'package:test/test.dart';
+
+import 'docker.dart';
+
+void main() {
+  withPostgresServer('unexpected protocol bytes', (server) {
+    late Connection conn;
+    late ServerSocket serverSocket;
+    bool sendGarbageResponse = false;
+
+    setUp(() async {
+      serverSocket = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
+      serverSocket.listen((socket) async {
+        final clientSocket = await Socket.connect(
+            InternetAddress.loopbackIPv4, await server.port);
+        late StreamSubscription socketSubs;
+        late StreamSubscription clientSubs;
+        socketSubs = socket.listen(clientSocket.add, onDone: () {
+          socketSubs.cancel();
+          clientSubs.cancel();
+          clientSocket.close();
+        }, onError: (e) {
+          socketSubs.cancel();
+          clientSubs.cancel();
+          clientSocket.close();
+        });
+        final pattern = [68, 0, 0, 0, 11, 0, 1, 0, 0, 0, 1];
+        clientSubs = clientSocket.listen((data) {
+          if (sendGarbageResponse) {
+            final i = _bytesIndexOf(data, pattern);
+            if (i >= 0) {
+              data[i + pattern.length - 4] = 255;
+              data[i + pattern.length - 3] = 255;
+              data[i + pattern.length - 2] = 255;
+              data[i + pattern.length - 1] = 250;
+            }
+            socket.add(data);
+          } else {
+            socket.add(data);
+          }
+        }, onDone: () {
+          socketSubs.cancel();
+          clientSubs.cancel();
+          clientSocket.close();
+        }, onError: (e) {
+          socketSubs.cancel();
+          clientSubs.cancel();
+          clientSocket.close();
+        });
+      });
+
+      final endpoint = await server.endpoint();
+      conn = await Connection.open(
+        Endpoint(
+          host: endpoint.host,
+          port: serverSocket.port,
+          database: endpoint.database,
+          password: endpoint.password,
+          username: endpoint.username,
+        ),
+        settings: ConnectionSettings(
+          sslMode: SslMode.disable,
+        ),
+      );
+    });
+
+    tearDown(() async {
+      await conn.close();
+      await serverSocket.close();
+    });
+
+    test('inject bad bytes', () async {
+      await conn.execute('SELECT 1;');
+      sendGarbageResponse = true;
+      await expectLater(
+          () => conn.execute('SELECT 2;', queryMode: QueryMode.simple),
+          throwsA(isA<PgException>()));
+      expect(conn.isOpen, false);
+    });
+  });
+}
+
+int _bytesIndexOf(List<int> data, List<int> pattern) {
+  for (var i = 0; i < data.length - pattern.length; i++) {
+    var matches = true;
+    for (var j = 0; j < pattern.length; j++) {
+      if (data[i + j] != pattern[j]) {
+        matches = false;
+        break;
+      }
+    }
+    if (matches) {
+      return i;
+    }
+  }
+  return -1;
+}