conn: create from socket fd

This patch adds the ability to create Tarantool connection
using an existing socket fd.

The authentication [1] might have already occured when we establish
such a connection. If that's the case, there is no need to pass
'user' argument.

1. https://www.tarantool.io/en/doc/latest/dev_guide/internals/iproto/authentication/

Closes #304

Author: askalt

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### Added
+- The ability to connect to the Tarantool using an existing socket fd (#304).
+
 ## 1.1.2 - 2023-09-20
 
 ### Fixed

tarantool/connection.py

@@ -623,7 +623,7 @@ def __init__(self, host, port,
             Unix sockets.
         :type host: :obj:`str` or :obj:`None`
 
-        :param port: Server port or Unix socket path.
+        :param port: Server port, socket fd or Unix socket path.
         :type port: :obj:`int` or :obj:`str`
 
         :param user: User name for authentication on the Tarantool
@@ -897,10 +897,33 @@ def connect_basic(self):
         :meta private:
         """
 
-        if self.host is None:
-            self.connect_unix()
-        else:
+        if self.host is not None:
             self.connect_tcp()
+            return
+        if isinstance(self.port, int):
+            self.connect_socket_fd()
+            return
+        self.connect_unix()
+
+    def connect_socket_fd(self):
+        """
+        Establish a connection using an existing socket fd.
+
+        :raise: :exc:`~tarantool.error.NetworkError`
+
+        :meta private:
+        """
+
+        try:
+            # If old socket already exists - close it and re-create.
+            self.connected = True
+            if self._socket:
+                self._socket.close()
+            self._socket = socket.socket(fileno=self.port)
+            self._socket.settimeout(self.socket_timeout)
+        except socket.error as exc:
+            self.connected = False
+            raise NetworkError(exc) from exc
 
     def connect_tcp(self):
         """

tarantool/mesh_connection.py

@@ -139,23 +139,23 @@ def format_error(address, err):
             result[key] = val
 
     if isinstance(result['port'], int):
-        # Looks like an inet address.
+        # Looks like an inet address or socket fd.
 
         # Validate host.
-        if 'host' not in result or result['host'] is None:
-            return format_error(result,
-                                'host is mandatory for an inet result')
-        if not isinstance(result['host'], str):
-            return format_error(result,
-                                'host must be a string for an inet result')
+        if ('host' in result and result['host'] is not None
+                and not isinstance(result['host'], str)):
+            return format_error(result, 'host must be a string for an inet result, '
+                                        'or None for a socket fd result')
 
         # Validate port.
         if not isinstance(result['port'], int):
             return format_error(result,
-                                'port must be an int for an inet result')
-        if result['port'] < 1 or result['port'] > 65535:
-            return format_error(result, 'port must be in range [1, 65535] '
-                                        'for an inet result')
+                                'port must be an int for an inet/socket fd result')
+        if 'host' in result and isinstance(result['host'], str):
+            # Check that the port is correct.
+            if result['port'] < 1 or result['port'] > 65535:
+                return format_error(result, 'port must be in range [1, 65535] '
+                                            'for an inet result')
 
         # Looks okay.
         return result, None
@@ -447,7 +447,8 @@ def __init__(self, host=None, port=None,
             # Don't change user provided arguments.
             addrs = addrs[:]
 
-        if host and port:
+        if port:
+            # host can be None in the case of socket fd or Unix socket.
             addrs.insert(0, {'host': host,
                              'port': port,
                              'transport': transport,

test/suites/lib/skip.py

@@ -306,3 +306,14 @@ def skip_or_run_iproto_basic_features_test(func):
 
     return skip_or_run_test_tarantool(func, '2.10.0',
                                       'does not support iproto ID and iproto basic features')
+
+
+def skip_or_run_box_session_new_tests(func):
+    """
+    Decorator to skip or run tests that use box.session.new.
+
+    Tarantool supports box.session.new only in current master since
+    commit 324872a.
+    See https://github.com/tarantool/tarantool/issues/8801.
+    """
+    return skip_or_run_test_tarantool(func, '3.0.0', 'does not support box.session.new')

test/suites/test_socket_fd.py

@@ -0,0 +1,102 @@
+"""
+This module tests work with connection over socket fd.
+"""
+# pylint: disable=missing-class-docstring,missing-function-docstring
+
+import socket
+import unittest
+
+import tarantool
+from .lib.skip import skip_or_run_box_session_new_tests
+from .lib.tarantool_server import TarantoolServer, find_port
+from .utils import assert_admin_success
+
+
+class TestSuiteSocketFD(unittest.TestCase):
+    EVAL_USER = "return box.session.user()"
+
+    @classmethod
+    def setUpClass(cls):
+        cls.srv = TarantoolServer()
+        cls.srv.script = 'test/suites/box.lua'
+        cls.srv.start()
+        cls.tcp_port = find_port()
+
+        resp = cls.srv.admin("""
+            local socket = require('socket')
+
+            box.cfg{}
+            box.schema.user.create('test', {password = 'test', if_not_exists = true})
+            box.schema.user.grant('test', 'read,write,execute,create', 'universe',
+                                   nil, {if_not_exists = true})
+            box.schema.user.grant('guest', 'execute', 'universe',
+                                   nil, {if_not_exists = true})
+
+            socket.tcp_server('localhost', %d, function(s)
+                box.session.new({
+                    type = 'binary',
+                    fd = s:fd(),
+                    user = 'test',
+                })
+                s:detach()
+            end)
+
+            return true
+        """ % cls.tcp_port)
+        assert_admin_success(resp)
+
+    @skip_or_run_box_session_new_tests
+    def setUp(self):
+        # Prevent a remote tarantool from clean our session.
+        if self.srv.is_started():
+            self.srv.touch_lock()
+
+    def get_tt_sock(self):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.connect(("localhost", self.tcp_port))
+        return sock
+
+    def test_01_socket_fd_connect(self):
+        sock = self.get_tt_sock()
+        conn = tarantool.connect(host=None, port=sock.fileno())
+        sock.detach()
+        try:
+            self.assertSequenceEqual(conn.eval(self.EVAL_USER), ["test"])
+        finally:
+            conn.close()
+
+    def test_02_socket_fd_re_auth(self):
+        sock = self.get_tt_sock()
+        conn = tarantool.connect(host=None, port=sock.fileno(), user="guest")
+        sock.detach()
+        try:
+            self.assertSequenceEqual(conn.eval(self.EVAL_USER), ["guest"])
+        finally:
+            conn.close()
+
+    def test_03_socket_fd_mesh(self):
+        sock = self.get_tt_sock()
+        pool = tarantool.ConnectionPool(
+            addrs=[{'host': None, 'port': sock.fileno()}]
+        )
+        sock.detach()
+        try:
+            self.assertSequenceEqual(pool.eval(self.EVAL_USER, mode=tarantool.Mode.ANY), ["test"])
+        finally:
+            pool.close()
+
+    def test_04_socket_fd_pool(self):
+        sock = self.get_tt_sock()
+        mesh = tarantool.MeshConnection(
+            host=None, port=sock.fileno()
+        )
+        sock.detach()
+        try:
+            self.assertSequenceEqual(mesh.eval(self.EVAL_USER), ["test"])
+        finally:
+            mesh.close()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.srv.stop()
+        cls.srv.clean()