Passing custom sock to client (MPTCP and other socket flags)

[Pithikos]

Currently it's possible to pass a custom socket object on the server side. This is handy in my case since we want to use Multi-Path TCP (MPTCP).

async def on_connection(*args, **kwargs):
    print("Connected")

def create_mptcp_sock(server_ip, server_port):
    """
    Create MultiPath TCP socket
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_MPTCP)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.setblocking(False)
    sock.bind((server_ip, server_port))
    sock.listen(0)
    return sock

sock = create_mptcp_sock(server_ip, server_port)
async with websockets.serve(on_connection, sock=sock) as server:
    ...

However when I pass a sock similarly in the client side, I get an exception when the client tries to connect.

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_MPTCP)
sock.setblocking(False)

async with websockets.connect(uri, sock=sock) as ws:
    ....

Traceback:

Traceback (most recent call last):
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/client.py", line 139, in read_http_response
    status_code, reason, headers = await read_response(self.reader)
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/http.py", line 120, in read_response
    status_line = await read_line(stream)
                  ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/http.py", line 194, in read_line
    line = await stream.readline()
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/asyncio/streams.py", line 568, in readline
    line = await self.readuntil(sep)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/asyncio/streams.py", line 660, in readuntil
    await self._wait_for_data('readuntil')
  File "/usr/lib/python3.12/asyncio/streams.py", line 545, in _wait_for_data
    await self._waiter
  File "/usr/lib/python3.12/asyncio/selector_events.py", line 1075, in write
    n = self._sock.send(data)
        ^^^^^^^^^^^^^^^^^^^^^
BrokenPipeError: [Errno 32] Broken pipe

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/alex/workspace/projectx/controller/controller.py", line 819, in <module>
    asyncio.run(connection_loop(state))
  File "/usr/lib/python3.12/asyncio/runners.py", line 194, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/asyncio/base_events.py", line 687, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/alex/workspace/projectx/controller/controller.py", line 729, in connection_loop
    async with websockets.connect(
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/client.py", line 629, in __aenter__
    return await self
           ^^^^^^^^^^
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/client.py", line 647, in __await_impl_timeout__
    return await self.__await_impl__()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/client.py", line 654, in __await_impl__
    await protocol.handshake(
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/client.py", line 319, in handshake
    status_code, response_headers = await self.read_http_response()
                                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/alex/workspace/projectx/.venv/lib/python3.12/site-packages/websockets/legacy/client.py", line 141, in read_http_response
    raise InvalidMessage("did not receive a valid HTTP response") from exc
websockets.exceptions.InvalidMessage: did not receive a valid HTTP response

Is there a way to achieve this with the current library?

[aaugustin]

Could you try the following:

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_MPTCP)
sock.setblocking(False)
reader, writer = await asyncio.open_connection(<host>, <port>, sock=sock)
writer.write(
    b"GET / HTTP/1.1\r\n"
    b"Host: <host>:<port>\r\n"
    b"Upgrade: websocket\r\n"
    b"Connection: Upgrade\r\n"
    b"Sec-WebSocket-Key: 9c55e0/siQ6tJPCs/QR8ZA==\r\n"
    b"Sec-WebSocket-Version: 13\r\n"
    b"\r\n"
)
while True:
    print(await reader.readline())

If that works, I'll investigate further.

If that doesn't work, then the problem is in asyncio, not in websockets, as this example didn't involve websockets.

NB: you didn't specify if uri is a ws:// or a wss:// URI. That could play a role.

[Pithikos]

@aaugustin thanks for the prompt reply. It seems asyncio.open_connection does not accept both sock and host/port at the same time. Ran into other issues too. But the below seems to work.

    loop = asyncio.get_running_loop()

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setblocking(False)

    await loop.sock_connect(sock, (server_ip, server_port))
    reader, writer = await asyncio.open_connection(sock=sock)
    writer.write(
        b"GET / HTTP/1.1\r\n"
        b"Host: <host>:<port>\r\n"
        b"Upgrade: websocket\r\n"
        b"Connection: Upgrade\r\n"
        b"Sec-WebSocket-Key: 9c55e0/siQ6tJPCs/QR8ZA==\r\n"
        b"Sec-WebSocket-Version: 13\r\n"
        b"\r\n"
    )
    while True:
        print(await reader.readline())

[aaugustin]

OK - this rules out an asyncio issue.

I noticed that you aren't using the latest version of websockets. As a consequence you get the legacy implementation by default. You could try updating websockets. I don't expect that it will make a difference — but it does, that's good news!

If you want to investigate further, I recommend turning on debug logs and looking for a clue of why the connection fails. Most likely the connection gets closed before websockets can read anything and the trackback you posted above is just a consequence of that.

Else, I'll try to reproduce — as far as I understand, it's possible to create a MPTCP server and client locally; it's just a matter of constructing the sockets as shown in your code samples above?

[aaugustin]

Taking a closer look, I think that your problem is that you aren't connecting the client socket to the server:

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_MPTCP)
await loop.sock_connect(sock, (server_ip, server_port))  # this line is missing from your original example
async with websockets.connect(uri, sock=sock) as ws:
    ...

It was also missing for my example and you fixed it :-) The correct version should work for websockets just like for asyncio.

For context, sock gives you full control over the TCP-level connection. websockets will only start TLS if it's a wss:// connection and then start the WebSocket handshake. It will not connect the socket.

[Pithikos]

Thanks.

Unfortunately I don't think that solves it. Namely I still get the HTTP handshake error in that case with v13.1 (websockets.legacy.exceptions.InvalidMessage: did not receive a valid HTTP response).

Also I can't upgrade to v14.0 since I get other errors then ('ClientConnection' object has no attribute 'transport'). I'm using Python3.12.

[aaugustin]

I tried to reproduce but I'm on macOS which doesn't support IPPROTO_MPTCP :-(

I'm surprised that you managed to make it work with asyncio.open_connection(sock=sock) but not with websockets. Indeed, websockets doesn't do anything special. It uses asyncio.create_connection which is a lower-level API than asyncio.open_connection but does basically the same.

[aaugustin]

Also I can't upgrade to v14.0 since I get other errors then ('ClientConnection' object has no attribute 'transport'). I'm using Python3.12.

This private attribute still exists in 14.0 and it's initialized at the same point as in the legacy implementation so I'm surprised that you're getting a different behavior.

[aaugustin]

IMO the next step is to enable debug logs, like I suggested above. Perhaps you'll get a traceback that will give us a clue about what's happening.

I may try to reproduce this on Linux at some point.