# Copyright 2013 The Emscripten Authors. All rights reserved.
# Emscripten is available under two separate licenses, the MIT license and the
# University of Illinois/NCSA Open Source License. Both these licenses can be
# found in the LICENSE file.
import socket
import socketserver
import threading
from functools import wraps
if __name__ == '__main__':
raise Exception('do not run this file directly; do something like: test/runner sockets_node')
import common
from common import EMCC, NON_ZERO, RunnerCore, create_file
from decorators import (
also_with_proxy_to_pthread,
crossplatform,
parameterized,
requires_dev_dependency,
requires_native_clang,
test_file,
)
from sockets_common import (
EMTEST_SKIP_NODE_DEV_PACKAGES,
EMTEST_SKIP_PYTHON_DEV_PACKAGES,
HAS_IPV6_LOOPBACK,
CompiledServerHarness,
EchoHandler,
WebsockifyServerHarness,
requires_python_dev_packages,
)
def requires_jspi_node(func):
# require_jspi() falls back to the d8/SpiderMonkey shells when node is too
# old, but these tests need node (real sockets via node:net), so handle the
# node setup directly. The new JSPI API requires node >= 24, so skip below
# that.
assert callable(func)
@wraps(func)
def decorated(self, *args, **kwargs):
if not common.check_node_version(24):
self.skipTest('JSPI requires node >= 24')
if not common.check_node_version(26):
self.node_args += ['--experimental-wasm-stack-switching']
self.cflags += ['-Wno-experimental']
self.set_setting('JSPI')
return func(self, *args, **kwargs)
return decorated
class sockets_node(RunnerCore):
# Note: in the WebsockifyServerHarness and CompiledServerHarness tests below, explicitly use
# consecutive server listen ports, because server teardown might not occur deterministically
# (python dtor time) and is a bit racy.
# WebsockifyServerHarness uses two port numbers, x and x-1, so increment it by two.
# CompiledServerHarness only uses one. If adding new tests, increment the used port
# addresses below (using ports below 32768 to avoid Linux ephemeral client ports).
@crossplatform
@parameterized({
'native': (WebsockifyServerHarness, 25160, ['-DTEST_DGRAM=0']),
'tcp': (CompiledServerHarness, 25162, ['-DTEST_DGRAM=0', '-sEXPORT_ES6', '--extern-post-js', test_file('modularize_post_js.js')]),
'udp': (CompiledServerHarness, 25164, ['-DTEST_DGRAM=1']),
'pthread': (CompiledServerHarness, 25166, ['-pthread', '-sPROXY_TO_PTHREAD']),
})
def test_nodejs_sockets_echo(self, harness_class, port, args):
if harness_class == WebsockifyServerHarness and common.EMTEST_LACKS_NATIVE_CLANG:
self.skipTest('requires native clang')
if harness_class == WebsockifyServerHarness and EMTEST_SKIP_PYTHON_DEV_PACKAGES:
self.skipTest('requires python websockify and EMTEST_SKIP_PYTHON_DEV_PACKAGES=1')
if harness_class == CompiledServerHarness and EMTEST_SKIP_NODE_DEV_PACKAGES:
self.skipTest('requires node ws and EMTEST_SKIP_NODE_DEV_PACKAGES=1')
# Basic test of node client against both a Websockified and compiled echo server.
with harness_class(test_file('sockets/test_sockets_echo_server.c'), args, port) as harness:
expected = 'do_msg_read: read 14 bytes'
self.do_runf('sockets/test_sockets_echo_client.c', expected, cflags=[f'-DSOCKK={harness.listen_port}', *args])
@requires_dev_dependency('ws')
def test_nodejs_sockets_connect_failure(self):
self.do_runf('sockets/test_sockets_echo_client.c', r'connect failed: (Connection refused|Host is unreachable)', regex=True, cflags=['-DSOCKK=666'], assert_returncode=NON_ZERO)
@requires_dev_dependency('ws')
def test_nodejs_sockets_listen_inuse(self):
server = socketserver.TCPServer(('127.0.0.1', 0), EchoHandler)
port = server.server_address[1]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
self.do_runf('sockets/test_sockets_listen_inuse.c', 'done\n', args=[str(port)])
finally:
server.shutdown()
server.server_close()
thread.join()
def _run_against_echo_server(self, src):
# Start a loopback TCP echo server on an ephemeral port and run the test
# against it, passing the port as argv[1].
server = socketserver.TCPServer(('127.0.0.1', 0), EchoHandler)
port = server.server_address[1]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
self.do_runf(src, 'done\n', cflags=['-sNODERAWSOCKETS'], args=[str(port)])
finally:
server.shutdown()
server.server_close()
thread.join()
# The proxy_to_pthread variant proves the backend works when socket syscalls
# are proxied to the main thread: with PROXY_TO_PTHREAD, main() runs on a
# worker and every socket call funnels to the main thread where node:net lives.
@also_with_proxy_to_pthread
def test_noderawsockets_echo(self):
# With -sNODERAWSOCKETS the client does a non-blocking connect, send and
# recv over a real OS socket against a loopback echo server we run here.
self._run_against_echo_server('sockets/test_tcp_echo.c')
def test_noderawsockets_client_bind(self):
# A client that bind()s an explicit source port has it honored by connect(),
# and the plain client path never realizes a private tcp_wrap handle. We
# allocate a free source port here and pass it alongside the echo server's.
# Reserve a free loopback port for the client's bound source port.
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('127.0.0.1', 0))
src_port = s.getsockname()[1]
s.close()
server = socketserver.TCPServer(('127.0.0.1', 0), EchoHandler)
port = server.server_address[1]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
self.do_runf('sockets/test_tcp_client_bind.c', 'done\n', cflags=['-sNODERAWSOCKETS'], args=[str(port), str(src_port)])
finally:
server.shutdown()
server.server_close()
thread.join()
def test_noderawsockets_connect_getsockname(self):
# getsockname() immediately after a non-blocking connect() on an unbound
# client reports the source address and ephemeral port synchronously
# (kernel semantics: both are assigned at connect(), not when the connection
# completes), and neither changes once connected.
if not self.try_require_node_version(26, 7):
self.skipTest('requires in-tick BoundSocket connect (node >= 26.7)')
self.do_runf('sockets/test_tcp_connect_getsockname.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
def test_noderawsockets_client_semantics(self):
# EISCONN on a second connect, shutdown(SHUT_WR) leaving reads working,
# EPIPE on a write after that, and POLLHUP after a full shutdown(SHUT_RDWR).
self._run_against_echo_server('sockets/test_tcp_client_semantics.c')
def test_noderawsockets_refused(self):
# A connect to a loopback port with nothing listening reports ECONNREFUSED.
self.do_runf('sockets/test_tcp_refused.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
def test_noderawsockets_backpressure(self):
# A sink server that accepts but never reads, so the client's writes fill
# the buffers and send() reports EAGAIN rather than buffering unboundedly.
done = threading.Event()
class SinkHandler(socketserver.BaseRequestHandler):
def handle(self):
done.wait(30) # hold the connection open without ever reading
server = socketserver.TCPServer(('127.0.0.1', 0), SinkHandler)
port = server.server_address[1]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
self.do_runf('sockets/test_tcp_backpressure.c', 'done\n', cflags=['-sNODERAWSOCKETS'], args=[str(port)])
finally:
done.set()
server.shutdown()
server.server_close()
thread.join()
@also_with_proxy_to_pthread
def test_noderawsockets_server(self):
# Self-contained loopback accept+echo, exercising bind(:0)+getsockname
# (synchronous ephemeral port), listen, accept, non-blocking connect, send
# and recv over real OS sockets via the tcp_wrap server path.
self.do_runf('sockets/test_tcp_server.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
@also_with_proxy_to_pthread
def test_noderawsockets_peek(self):
# recv(MSG_PEEK) must leave the data buffered: a peek returns the bytes, the
# socket stays readable, and the following plain recv returns them again.
self.do_runf('sockets/test_tcp_peek.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
# AF_UNIX is gated behind NODERAWFS: the socket path lives in the host
# filesystem, which only stays coherent with the program's own file syscalls
# (bind's parent dir, getsockname, unlink) when the FS is the host FS.
@also_with_proxy_to_pthread
def test_noderawsockets_unix_server(self):
# Self-contained named AF_UNIX (pathname) loopback accept+echo: bind(path),
# listen, getsockname (the bound path), accept, getpeername, non-blocking
# connect-by-path, send and recv over a real node pipe.
self.do_runf('sockets/test_unix_server.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sNODERAWFS'])
def test_noderawsockets_unix_refused(self):
# A connect to an AF_UNIX path with no socket file reports ENOENT.
self.do_runf('sockets/test_unix_refused.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sNODERAWFS'])
def test_noderawsockets_unix_bind_inuse(self):
# Binding an already-bound AF_UNIX path fails synchronously with EADDRINUSE.
self.do_runf('sockets/test_unix_bind_inuse.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sNODERAWFS'])
def test_noderawsockets_server_autobind(self):
# listen() without a prior bind() must auto-bind an ephemeral port and
# getsockname() must report it (POSIX), then accept+echo as usual.
self.do_runf('sockets/test_tcp_server.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-DNO_EXPLICIT_BIND'])
def test_noderawsockets_tcp_ipv6(self):
# Self-contained IPv6 TCP loopback accept+echo over ::1: bind(:0)+getsockname,
# listen, accept, non-blocking connect, send/recv on AF_INET6 sockets.
if not HAS_IPV6_LOOPBACK:
self.skipTest('no IPv6 loopback available')
self.do_runf('sockets/test_tcp_ipv6.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
def test_noderawsockets_udp_ipv6(self):
# Self-contained IPv6 UDP loopback echo over ::1 on AF_INET6 sockets.
if not HAS_IPV6_LOOPBACK:
self.skipTest('no IPv6 loopback available')
self.do_runf('sockets/test_udp_ipv6.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
def test_noderawsockets_dns(self):
# getaddrinfo() resolves numeric addresses synchronously. A hostname needs
# a node:dns lookup, and with no stack able to wait on it is EAI_AGAIN.
self.do_runf('sockets/test_dns.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-DNO_WAIT'])
def test_noderawsockets_dns_blocking(self):
# A hostname blocks on the node:dns lookup, returning every address as a
# linked list: main() is proxied to a worker, which awaits the resolution
# through the sync proxy.
self.do_runf('sockets/test_dns.c', 'done\n',
cflags=['-sNODERAWSOCKETS', '-pthread', '-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME'])
@requires_jspi_node
def test_noderawsockets_dns_blocking_jspi(self):
# Same, but getaddrinfo() suspends the wasm stack under JSPI.
self.do_runf('sockets/test_dns.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME'])
def test_noderawsockets_dns_blocking_asyncify(self):
# Same, unwinding the wasm stack under ASYNCIFY.
self.do_runf('sockets/test_dns.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sASYNCIFY', '-sEXIT_RUNTIME'])
def test_noderawsockets_epoll_socket_blocking(self):
# A blocking epoll_wait() on a socket is woken by an incoming datagram
# through the unified readiness wait-queue (the SOCKFS.emit bridge), with
# main() proxied to a worker so the wait can suspend.
self.do_runf('sockets/test_epoll_socket_blocking.c', 'done\n',
cflags=['-sNODERAWSOCKETS', '-pthread', '-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME'])
@requires_jspi_node
def test_noderawsockets_epoll_socket_blocking_jspi(self):
# Same, but the blocking epoll_wait() suspends the wasm stack under JSPI.
self.do_runf('sockets/test_epoll_socket_blocking.c', 'done\n',
cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME'])
def test_noderawsockets_epoll_rdhup(self):
# A blocking epoll_wait reports EPOLLRDHUP when the TCP peer half-closes its
# write side (FIN), distinct from a full EPOLLHUP, and only when requested.
self.do_runf('sockets/test_epoll_rdhup.c', 'done\n',
cflags=['-sNODERAWSOCKETS', '-pthread', '-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME'])
def test_noderawsockets_nonblock_flags(self):
# socket()/accept4() SOCK_NONBLOCK, FIONBIO, no listener flag inheritance on
# accept, non-blocking connect EINPROGRESS (TCP and AF_UNIX), and a warning
# when a blocking fd would-blocks.
out = self.do_runf('sockets/test_nonblock_flags.c', 'done\n',
cflags=['-sNODERAWSOCKETS', '-sNODERAWFS', '-sASSERTIONS', '-pthread', '-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME'])
self.assertContained('a blocking socket operation would block', out)
@requires_jspi_node
def test_noderawsockets_epoll_rdhup_jspi(self):
# Same, but the blocking calls suspend the wasm stack under JSPI.
self.do_runf('sockets/test_epoll_rdhup.c', 'done\n',
cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME'])
@also_with_proxy_to_pthread
def test_noderawsockets_udp(self):
# Self-contained loopback UDP echo: the server binds(:0)+getsockname for its
# ephemeral port, the client sends a datagram, the server echoes it back.
self.do_runf('sockets/test_udp_echo.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
def test_noderawsockets_udp_recvmsg(self):
# recvmsg scatters a datagram across multiple iovecs at the right offsets
# and updates msg_namelen/msg_controllen/msg_flags in the caller's msghdr.
self.do_runf('sockets/test_udp_recvmsg.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
def test_noderawsockets_mmsg(self):
# sendmmsg batches two datagrams out, recvmmsg receives them back in one
# call, updating msg_len per message.
self.do_runf('sockets/test_udp_mmsg.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
@parameterized({
'': ([],),
'unref': (['-DMODE_UNREF'],),
})
def test_noderawsockets_epoll_callback(self, cflags):
# An epoll listener callback woken repeatedly by arriving datagrams on a
# real socket via the SOCKFS -> wait-queue bridge, with no ASYNCIFY/JSPI.
# MODE_UNREF exits without waiting for delivery.
self.do_runf('sockets/test_epoll_callback.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME', *cflags])
def test_noderawsockets_epoll_callback_force_exit(self):
# emscripten_force_exit with a listener registered on an armed socket:
# FS.quit closes the epoll on the way out, removing the listener.
self.do_runf('sockets/test_epoll_callback_force_exit.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME'])
@also_with_proxy_to_pthread
def test_noderawsockets_udp_connect(self):
# Connected UDP: sendto() with an address gives EISCONN, send() reaches the
# peer, and datagrams from a non-peer socket are filtered out.
self.do_runf('sockets/test_udp_connect.c', 'done\n', cflags=['-sNODERAWSOCKETS'])
@also_with_proxy_to_pthread
def test_noderawsockets_udp_sockopts(self):
# UDP socket options: IP_MULTICAST_TTL/LOOP and their IPv6 counterparts
# round-trip through set/getsockopt, with POSIX defaults readable before
# any set, and SO_REUSEADDR/SO_REUSEPORT are applied at bind so sockets can
# share a port, and multicast group joins reach the kernel. EXIT_RUNTIME so the plain synchronous main()
# tears down the proxy worker on return (otherwise noExitRuntime keeps the
# worker, and thus node, alive under PROXY_TO_PTHREAD).
self.do_runf('sockets/test_udp_sockopts.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME'])
@also_with_proxy_to_pthread
def test_noderawsockets_socket_options(self):
# Socket metadata/options on a fresh socket: fstat reports S_ISSOCK, the
# read-only identity options, SO_LINGER/SO_RCVTIMEO/SO_SNDTIMEO struct
# round-trips, flag options, and symmetric ENOPROTOOPT for unknown options.
self.do_runf('sockets/test_socket_options.c', 'done\n',
cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME'])
@requires_native_clang
@requires_python_dev_packages
@requires_dev_dependency('ws')
def test_nodejs_sockets_echo_subprotocol(self):
# Test against a Websockified server with compile time configured WebSocket subprotocol. We use a Websockified
# server because as long as the subprotocol list contains binary it will configure itself to accept binary
# This test also checks that the connect url contains the correct subprotocols.
with WebsockifyServerHarness(test_file('sockets/test_sockets_echo_server.c'), [], 25168):
self.run_process([EMCC, '-Werror', test_file('sockets/test_sockets_echo_client.c'), '-o', 'client.js', '-sSOCKET_DEBUG', '-sWEBSOCKET_SUBPROTOCOL="base64, binary"', '-DSOCKK=25168'])
out = self.run_js('client.js')
self.assertContained('do_msg_read: read 14 bytes', out)
self.assertContained(['connect: ws://127.0.0.1:25168, base64,binary', 'connect: ws://127.0.0.1:25168/, base64,binary'], out)
@requires_native_clang
@requires_python_dev_packages
@requires_dev_dependency('ws')
def test_nodejs_sockets_echo_subprotocol_runtime(self):
# Test against a Websockified server with runtime WebSocket configuration. We specify both url and subprotocol.
# In this test we have *deliberately* used the wrong port '-DSOCKK=12345' to configure the echo_client.c, so
# the connection would fail without us specifying a valid WebSocket URL in the configuration.
create_file('websocket_pre.js', '''
var Module = {
websocket: {
url: 'ws://localhost:25170/testA/testB',
subprotocol: 'text, base64, binary',
}
};
''')
with WebsockifyServerHarness(test_file('sockets/test_sockets_echo_server.c'), [], 25170):
self.run_process([EMCC, '-Werror', test_file('sockets/test_sockets_echo_client.c'), '-o', 'client.js', '--pre-js=websocket_pre.js', '-sSOCKET_DEBUG', '-DSOCKK=12345'])
out = self.run_js('client.js')
self.assertContained('do_msg_read: read 14 bytes', out)
self.assertContained('connect: ws://localhost:25170/testA/testB, text,base64,binary', out)
class sockets_node64(sockets_node):
def setUp(self):
super().setUp()
self.cflags.append('-m64')
self.require_wasm64()