ws_listener.cpp

Go to the documentation of this file.

/* SPDX-License-Identifier: MPL-2.0 */
 
#include "precompiled.hpp"
#include <new>
 
#include <string>
#include <stdio.h>
 
#include "ws_listener.hpp"
#include "io_thread.hpp"
#include "config.hpp"
#include "err.hpp"
#include "ip.hpp"
#include "tcp.hpp"
#include "socket_base.hpp"
#include "address.hpp"
#include "ws_engine.hpp"
#include "session_base.hpp"
 
#ifdef ZMQ_HAVE_WSS
#include "wss_engine.hpp"
#include "wss_address.hpp"
#endif
 
#ifndef ZMQ_HAVE_WINDOWS
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/tcp.h>
#include <netinet/in.h>
#include <netdb.h>
#include <fcntl.h>
#ifdef ZMQ_HAVE_VXWORKS
#include <sockLib.h>
#endif
#endif
 
#ifdef ZMQ_HAVE_OPENVMS
#include <ioctl.h>
#endif
 
zmq::ws_listener_t::ws_listener_t (io_thread_t *io_thread_,
                                   socket_base_t *socket_,
                                   const options_t &options_,
                                   bool wss_) :
    stream_listener_base_t (io_thread_, socket_, options_), _wss (wss_)
{
#ifdef ZMQ_HAVE_WSS
    if (_wss) {
        int rc = gnutls_certificate_allocate_credentials (&_tls_cred);
        zmq_assert (rc == GNUTLS_E_SUCCESS);
 
        gnutls_datum_t cert = {(unsigned char *) options_.wss_cert_pem.c_str (),
                               (unsigned int) options_.wss_cert_pem.length ()};
        gnutls_datum_t key = {(unsigned char *) options_.wss_key_pem.c_str (),
                              (unsigned int) options_.wss_key_pem.length ()};
        rc = gnutls_certificate_set_x509_key_mem (_tls_cred, &cert, &key,
                                                  GNUTLS_X509_FMT_PEM);
        zmq_assert (rc == GNUTLS_E_SUCCESS);
    }
#endif
}
 
zmq::ws_listener_t::~ws_listener_t ()
{
#ifdef ZMQ_HAVE_WSS
    if (_wss)
        gnutls_certificate_free_credentials (_tls_cred);
#endif
}
 
void zmq::ws_listener_t::in_event ()
{
    const fd_t fd = accept ();
 
    //  If connection was reset by the peer in the meantime, just ignore it.
    //  TODO: Handle specific errors like ENFILE/EMFILE etc.
    if (fd == retired_fd) {
        _socket->event_accept_failed (
          make_unconnected_bind_endpoint_pair (_endpoint), zmq_errno ());
        return;
    }
 
    int rc = tune_tcp_socket (fd);
    rc = rc | tune_tcp_maxrt (fd, options.tcp_maxrt);
    if (rc != 0) {
        _socket->event_accept_failed (
          make_unconnected_bind_endpoint_pair (_endpoint), zmq_errno ());
        return;
    }
 
    //  Create the engine object for this connection.
    create_engine (fd);
}
 
std::string zmq::ws_listener_t::get_socket_name (zmq::fd_t fd_,
                                                 socket_end_t socket_end_) const
{
    std::string socket_name;
 
#ifdef ZMQ_HAVE_WSS
    if (_wss)
        socket_name = zmq::get_socket_name<wss_address_t> (fd_, socket_end_);
    else
#endif
        socket_name = zmq::get_socket_name<ws_address_t> (fd_, socket_end_);
 
    return socket_name + _address.path ();
}
 
int zmq::ws_listener_t::create_socket (const char *addr_)
{
    tcp_address_t address;
    _s = tcp_open_socket (addr_, options, true, true, &address);
    if (_s == retired_fd) {
        return -1;
    }
 
    //  TODO why is this only done for the listener?
    make_socket_noninheritable (_s);
 
    //  Allow reusing of the address.
    int flag = 1;
    int rc;
#ifdef ZMQ_HAVE_WINDOWS
    //  TODO this was changed for Windows from SO_REUSEADDRE to
    //  SE_EXCLUSIVEADDRUSE by 0ab65324195ad70205514d465b03d851a6de051c,
    //  so the comment above is no longer correct; also, now the settings are
    //  different between listener and connecter with a src address.
    //  is this intentional?
    rc = setsockopt (_s, SOL_SOCKET, SO_EXCLUSIVEADDRUSE,
                     reinterpret_cast<const char *> (&flag), sizeof (int));
    wsa_assert (rc != SOCKET_ERROR);
#elif defined ZMQ_HAVE_VXWORKS
    rc =
      setsockopt (_s, SOL_SOCKET, SO_REUSEADDR, (char *) &flag, sizeof (int));
    errno_assert (rc == 0);
#else
    rc = setsockopt (_s, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof (int));
    errno_assert (rc == 0);
#endif
 
    //  Bind the socket to the network interface and port.
#if defined ZMQ_HAVE_VXWORKS
    rc = bind (_s, (sockaddr *) _address.addr (), _address.addrlen ());
#else
    rc = bind (_s, address.addr (), address.addrlen ());
#endif
#ifdef ZMQ_HAVE_WINDOWS
    if (rc == SOCKET_ERROR) {
        errno = wsa_error_to_errno (WSAGetLastError ());
        goto error;
    }
#else
    if (rc != 0)
        goto error;
#endif
 
    //  Listen for incoming connections.
    rc = listen (_s, options.backlog);
#ifdef ZMQ_HAVE_WINDOWS
    if (rc == SOCKET_ERROR) {
        errno = wsa_error_to_errno (WSAGetLastError ());
        goto error;
    }
#else
    if (rc != 0)
        goto error;
#endif
 
    return 0;
 
error:
    const int err = errno;
    close ();
    errno = err;
    return -1;
}
 
int zmq::ws_listener_t::set_local_address (const char *addr_)
{
    if (options.use_fd != -1) {
        //  in this case, the addr_ passed is not used and ignored, since the
        //  socket was already created by the application
        _s = options.use_fd;
    } else {
        const int rc = _address.resolve (addr_, true, options.ipv6);
        if (rc != 0)
            return -1;
 
        //  remove the path, otherwise resolving the port will fail with wildcard
        const char *delim = strrchr (addr_, '/');
        std::string host_address;
        if (delim) {
            host_address = std::string (addr_, delim - addr_);
        } else {
            host_address = addr_;
        }
 
        if (create_socket (host_address.c_str ()) == -1)
            return -1;
    }
 
    _endpoint = get_socket_name (_s, socket_end_local);
 
    _socket->event_listening (make_unconnected_bind_endpoint_pair (_endpoint),
                              _s);
    return 0;
}
 
zmq::fd_t zmq::ws_listener_t::accept ()
{
    //  The situation where connection cannot be accepted due to insufficient
    //  resources is considered valid and treated by ignoring the connection.
    //  Accept one connection and deal with different failure modes.
    zmq_assert (_s != retired_fd);
 
    struct sockaddr_storage ss;
    memset (&ss, 0, sizeof (ss));
#if defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_VXWORKS
    int ss_len = sizeof (ss);
#else
    socklen_t ss_len = sizeof (ss);
#endif
#if defined ZMQ_HAVE_SOCK_CLOEXEC && defined HAVE_ACCEPT4
    fd_t sock = ::accept4 (_s, reinterpret_cast<struct sockaddr *> (&ss),
                           &ss_len, SOCK_CLOEXEC);
#else
    const fd_t sock =
      ::accept (_s, reinterpret_cast<struct sockaddr *> (&ss), &ss_len);
#endif
 
    if (sock == retired_fd) {
#if defined ZMQ_HAVE_WINDOWS
        const int last_error = WSAGetLastError ();
        wsa_assert (last_error == WSAEWOULDBLOCK || last_error == WSAECONNRESET
                    || last_error == WSAEMFILE || last_error == WSAENOBUFS);
#elif defined ZMQ_HAVE_ANDROID
        errno_assert (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR
                      || errno == ECONNABORTED || errno == EPROTO
                      || errno == ENOBUFS || errno == ENOMEM || errno == EMFILE
                      || errno == ENFILE || errno == EINVAL);
#else
        errno_assert (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR
                      || errno == ECONNABORTED || errno == EPROTO
                      || errno == ENOBUFS || errno == ENOMEM || errno == EMFILE
                      || errno == ENFILE);
#endif
        return retired_fd;
    }
 
    make_socket_noninheritable (sock);
 
    if (zmq::set_nosigpipe (sock)) {
#ifdef ZMQ_HAVE_WINDOWS
        const int rc = closesocket (sock);
        wsa_assert (rc != SOCKET_ERROR);
#else
        int rc = ::close (sock);
        errno_assert (rc == 0);
#endif
        return retired_fd;
    }
 
    // Set the IP Type-Of-Service priority for this client socket
    if (options.tos != 0)
        set_ip_type_of_service (sock, options.tos);
 
    // Set the protocol-defined priority for this client socket
    if (options.priority != 0)
        set_socket_priority (sock, options.priority);
 
    return sock;
}
 
void zmq::ws_listener_t::create_engine (fd_t fd_)
{
    const endpoint_uri_pair_t endpoint_pair (
      get_socket_name (fd_, socket_end_local),
      get_socket_name (fd_, socket_end_remote), endpoint_type_bind);
 
    i_engine *engine = NULL;
    if (_wss)
#ifdef ZMQ_HAVE_WSS
        engine = new (std::nothrow)
          wss_engine_t (fd_, options, endpoint_pair, _address, false, _tls_cred,
                        std::string ());
#else
        zmq_assert (false);
#endif
    else
        engine = new (std::nothrow)
          ws_engine_t (fd_, options, endpoint_pair, _address, false);
 
    alloc_assert (engine);
 
    //  Choose I/O thread to run connecter in. Given that we are already
    //  running in an I/O thread, there must be at least one available.
    io_thread_t *io_thread = choose_io_thread (options.affinity);
    zmq_assert (io_thread);
 
    //  Create and launch a session object.
    session_base_t *session =
      session_base_t::create (io_thread, false, _socket, options, NULL);
    errno_assert (session);
    session->inc_seqnum ();
    launch_child (session);
    send_attach (session, engine, false);
 
    _socket->event_accepted (endpoint_pair, fd_);
}