guess_wire_crypt()

Author: nakagami

firebirdsql/aio/fbcore.py

@@ -652,27 +652,29 @@ async def _async_parse_connect_response(self):
                     b''
                 )
                 (h, oid, buf) = await self._async_op_response()
-                guessed_wire_crypt = self._guess_wire_crypt(buf)
+                enc_plugin, nonce = guess_wire_crypt(buf)
             else:
-                guessed_wire_crypt = None
+                enc_plugin = nonce = None
 
-            if guessed_wire_crypt and self.wire_crypt and session_key:
-                self._op_crypt(guessed_wire_crypt[0])
-                if guessed_wire_crypt[0] == b'Arc4':
+            if enc_plugin and self.wire_crypt and session_key:
+                self._op_crypt(enc_plugin)
+                if enc_plugin == b'Arc4':
                     self.sock.set_translator(
                         ARC4.new(session_key), ARC4.new(session_key))
-                elif guessed_wire_crypt[0] == b'ChaCha':
+                elif enc_plugin == b'ChaCha':
                     k = hashlib.sha256(session_key).digest()
                     self.sock.set_translator(
-                        ChaCha20.new(k, guessed_wire_crypt[1]),
-                        ChaCha20.new(k, guessed_wire_crypt[1]),
+                        ChaCha20.new(k, nonce),
+                        ChaCha20.new(k, nonce),
                     )
                 else:
                     raise OperationalError(
-                        'Unknown wirecrypt plugin %s' % (guessed_wire_crypt)
+                        'Unknown wirecrypt plugin %s' % (enc_plugin)
                     )
-                (h, oid, buf) = await self._async_op_response()
-            else:   # use later _op_attach() and _op_create()
+                (h, oid, buf) = self._op_response()
+            else:
+                # no matched wire encription plugin
+                # self.auth_data use _op_attach() and _op_create()
                 self.auth_data = auth_data
         else:
             assert op_code == self.op_accept

firebirdsql/fbcore.py

@@ -755,7 +755,7 @@ def _parse_connect_response(self):
                     b''
                 )
                 (h, oid, buf) = self._op_response()
-                enc_plugin, nonce = self._guess_wire_crypt(buf)
+                enc_plugin, nonce = guess_wire_crypt(buf)
             else:
                 enc_plugin = nonce = None
 

firebirdsql/utils.py

@@ -167,6 +167,37 @@ def parse_dsn(dsn, host=None, port=None, database=None, user=None, password=None
     return host, port, database, user, password
 
 
+def guess_wire_crypt(b):
+    available_plugins = []
+    plugin_nonce = []
+    i = 0
+    while i < len(b):
+        t = b[i] if PYTHON_MAJOR_VER == 3 else ord(b[i])
+        i += 1
+        ln = b[i] if PYTHON_MAJOR_VER == 3 else ord(b[i])
+        i += 1
+        v = b[i:i+ln]
+        i += ln
+        if t == 0:
+            assert v == b"Symmetric"
+        elif t == 1:
+            available_plugins = v.split()
+        elif t == 3:
+            plugin_nonce.append(v)
+
+#    if b'ChaCha64' in available_plugins:
+#        for s in plugin_nonce:
+#            if s[:9] == b"ChaCha64\x00":
+#                return (b'ChaCha64', s[9:])
+    if b'ChaCha' in available_plugins:
+        for s in plugin_nonce:
+            if s[:7] == b"ChaCha\x00":
+                return (b'ChaCha', s[7:7 + 12])
+    elif b'Arc4' in available_plugins:
+        return (b'Arc4', None)
+    return None, None
+
+
 class RowMapping(Mapping):
     """dict like interface to result rows
     """

firebirdsql/wireprotocol.py

@@ -465,36 +465,6 @@ def _op_crypt(self, algo):
         p.pack_bytes(b'Symmetric')
         self.sock.send(p.get_buffer())
 
-    def _guess_wire_crypt(self, b):
-        available_plugins = []
-        plugin_nonce = []
-        i = 0
-        while i < len(b):
-            t = b[i] if PYTHON_MAJOR_VER == 3 else ord(b[i])
-            i += 1
-            ln = b[i] if PYTHON_MAJOR_VER == 3 else ord(b[i])
-            i += 1
-            v = b[i:i+ln]
-            i += ln
-            if t == 0:
-                assert v == b"Symmetric"
-            elif t == 1:
-                available_plugins = v.split()
-            elif t == 3:
-                plugin_nonce.append(v)
-
-#        if b'ChaCha64' in available_plugins:
-#            for s in plugin_nonce:
-#                if s[:9] == b"ChaCha64\x00":
-#                    return (b'ChaCha64', s[9:])
-        if b'ChaCha' in available_plugins:
-            for s in plugin_nonce:
-                if s[:7] == b"ChaCha\x00":
-                    return (b'ChaCha', s[7:7 + 12])
-        elif b'Arc4' in available_plugins:
-            return (b'Arc4', None)
-        return None, None
-
     @wire_operation
     def _op_attach(self, timezone):
         dpb = bs([isc_dpb_version1])