src/openvpn/mtcp.c

/*
 *  OpenVPN -- An application to securely tunnel IP networks
 *             over a single TCP/UDP port, with support for SSL/TLS-based
 *             session authentication and key exchange,
 *             packet encryption, packet authentication, and
 *             packet compression.
 *
 *  Copyright (C) 2002-2018 OpenVPN Inc <sales@openvpn.net>
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License version 2
 *  as published by the Free Software Foundation.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License along
 *  with this program; if not, write to the Free Software Foundation, Inc.,
 *  51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#elif defined(_MSC_VER)
#include "config-msvc.h"
#endif

#include "syshead.h"

#if P2MP_SERVER

#include "multi.h"
#include "forward-inline.h"

#include "memdbg.h"

#ifdef HAVE_SYS_INOTIFY_H
#include <sys/inotify.h>
#endif

/*
 * TCP States
 */
#define TA_UNDEF                 0
#define TA_SOCKET_READ           1
#define TA_SOCKET_READ_RESIDUAL  2
#define TA_SOCKET_WRITE          3
#define TA_SOCKET_WRITE_READY    4
#define TA_SOCKET_WRITE_DEFERRED 5
#define TA_TUN_READ              6
#define TA_TUN_WRITE             7
#define TA_INITIAL               8
#define TA_TIMEOUT               9
#define TA_TUN_WRITE_TIMEOUT     10

/*
 * Special tags passed to event.[ch] functions
 */
#define MTCP_SOCKET      ((void *)1)
#define MTCP_TUN         ((void *)2)
#define MTCP_SIG         ((void *)3) /* Only on Windows */
#ifdef ENABLE_MANAGEMENT
#define MTCP_MANAGEMENT ((void *)4)
#endif

#ifdef ENABLE_ASYNC_PUSH
#define MTCP_FILE_CLOSE_WRITE ((void *)5)
#endif

#define MTCP_N           ((void *)16) /* upper bound on MTCP_x */

struct ta_iow_flags
{
    unsigned int flags;
    unsigned int ret;
    unsigned int tun;
    unsigned int sock;
};

static const char *
pract(int action)
{
    switch (action)
    {
        case TA_UNDEF:
            return "TA_UNDEF";

        case TA_SOCKET_READ:
            return "TA_SOCKET_READ";

        case TA_SOCKET_READ_RESIDUAL:
            return "TA_SOCKET_READ_RESIDUAL";

        case TA_SOCKET_WRITE:
            return "TA_SOCKET_WRITE";

        case TA_SOCKET_WRITE_READY:
            return "TA_SOCKET_WRITE_READY";

        case TA_SOCKET_WRITE_DEFERRED:
            return "TA_SOCKET_WRITE_DEFERRED";

        case TA_TUN_READ:
            return "TA_TUN_READ";

        case TA_TUN_WRITE:
            return "TA_TUN_WRITE";

        case TA_INITIAL:
            return "TA_INITIAL";

        case TA_TIMEOUT:
            return "TA_TIMEOUT";

        case TA_TUN_WRITE_TIMEOUT:
            return "TA_TUN_WRITE_TIMEOUT";

        default:
            return "?";
    }
}

static struct multi_instance *
multi_create_instance_tcp(struct multi_context *m)
{
    struct gc_arena gc = gc_new();
    struct multi_instance *mi = NULL;
    struct hash *hash = m->hash;

    mi = multi_create_instance(m, NULL);
    if (mi)
    {
        struct hash_element *he;
        const uint32_t hv = hash_value(hash, &mi->real);
        struct hash_bucket *bucket = hash_bucket(hash, hv);

        he = hash_lookup_fast(hash, bucket, &mi->real, hv);

        if (he)
        {
            struct multi_instance *oldmi = (struct multi_instance *) he->value;
            msg(D_MULTI_LOW, "MULTI TCP: new incoming client address matches existing client address -- new client takes precedence");
            oldmi->did_real_hash = false;
            multi_close_instance(m, oldmi, false);
            he->key = &mi->real;
            he->value = mi;
        }
        else
        {
            hash_add_fast(hash, bucket, &mi->real, hv, mi);
        }

        mi->did_real_hash = true;
    }

#ifdef ENABLE_DEBUG
    if (mi)
    {
        dmsg(D_MULTI_DEBUG, "MULTI TCP: instance added: %s", mroute_addr_print(&mi->real, &gc));
    }
    else
    {
        dmsg(D_MULTI_DEBUG, "MULTI TCP: new client instance failed");
    }
#endif

    gc_free(&gc);
    ASSERT(!(mi && mi->halt));
    return mi;
}

bool
multi_tcp_instance_specific_init(struct multi_context *m, struct multi_instance *mi)
{
    /* buffer for queued TCP socket output packets */
    mi->tcp_link_out_deferred = mbuf_init(m->top.options.n_bcast_buf);

    ASSERT(mi->context.c2.link_socket);
    ASSERT(mi->context.c2.link_socket->info.lsa);
    ASSERT(mi->context.c2.link_socket->mode == LS_MODE_TCP_ACCEPT_FROM);
    ASSERT(mi->context.c2.link_socket->info.lsa->actual.dest.addr.sa.sa_family == AF_INET
           || mi->context.c2.link_socket->info.lsa->actual.dest.addr.sa.sa_family == AF_INET6
           );
    if (!mroute_extract_openvpn_sockaddr(&mi->real, &mi->context.c2.link_socket->info.lsa->actual.dest, true))
    {
        msg(D_MULTI_ERRORS, "MULTI TCP: TCP client address is undefined");
        return false;
    }
    return true;
}

void
multi_tcp_instance_specific_free(struct multi_instance *mi)
{
    mbuf_free(mi->tcp_link_out_deferred);
}

struct multi_tcp *
multi_tcp_init(int maxevents, int *maxclients)
{
    struct multi_tcp *mtcp;
    const int extra_events = BASE_N_EVENTS;

    ASSERT(maxevents >= 1);
    ASSERT(maxclients);

    ALLOC_OBJ_CLEAR(mtcp, struct multi_tcp);
    mtcp->maxevents = maxevents + extra_events;
    mtcp->es = event_set_init(&mtcp->maxevents, 0);
    wait_signal(mtcp->es, MTCP_SIG);
    ALLOC_ARRAY(mtcp->esr, struct event_set_return, mtcp->maxevents);
    *maxclients = max_int(min_int(mtcp->maxevents - extra_events, *maxclients), 1);
    msg(D_MULTI_LOW, "MULTI: TCP INIT maxclients=%d maxevents=%d", *maxclients, mtcp->maxevents);
    return mtcp;
}

void
multi_tcp_delete_event(struct multi_tcp *mtcp, event_t event)
{
    if (mtcp && mtcp->es)
    {
        event_del(mtcp->es, event);
    }
}

void
multi_tcp_free(struct multi_tcp *mtcp)
{
    if (mtcp)
    {
        event_free(mtcp->es);
        if (mtcp->esr)
        {
            free(mtcp->esr);
        }
        free(mtcp);
    }
}

void
multi_tcp_dereference_instance(struct multi_tcp *mtcp, struct multi_instance *mi)
{
    struct link_socket *ls = mi->context.c2.link_socket;
    if (ls && mi->socket_set_called)
    {
        event_del(mtcp->es, socket_event_handle(ls));
    }
    mtcp->n_esr = 0;
}

static inline void
multi_tcp_set_global_rw_flags(struct multi_context *m, struct multi_instance *mi)
{
    if (mi)
    {
        mi->socket_set_called = true;
        socket_set(mi->context.c2.link_socket,
                   m->mtcp->es,
                   mbuf_defined(mi->tcp_link_out_deferred) ? EVENT_WRITE : EVENT_READ,
                   mi,
                   &mi->tcp_rwflags);
    }
}

static inline int
multi_tcp_wait(const struct context *c,
               struct multi_tcp *mtcp)
{
    int status;
    socket_set_listen_persistent(c->c2.link_socket, mtcp->es, MTCP_SOCKET);
    tun_set(c->c1.tuntap, mtcp->es, EVENT_READ, MTCP_TUN, &mtcp->tun_rwflags);
#ifdef ENABLE_MANAGEMENT
    if (management)
    {
        management_socket_set(management, mtcp->es, MTCP_MANAGEMENT, &mtcp->management_persist_flags);
    }
#endif

#ifdef ENABLE_ASYNC_PUSH
    /* arm inotify watcher */
    event_ctl(mtcp->es, c->c2.inotify_fd, EVENT_READ, MTCP_FILE_CLOSE_WRITE);
#endif

    status = event_wait(mtcp->es, &c->c2.timeval, mtcp->esr, mtcp->maxevents);
    update_time();
    mtcp->n_esr = 0;
    if (status > 0)
    {
        mtcp->n_esr = status;
    }
    return status;
}

static inline struct context *
multi_tcp_context(struct multi_context *m, struct multi_instance *mi)
{
    if (mi)
    {
        return &mi->context;
    }
    else
    {
        return &m->top;
    }
}

static bool
multi_tcp_process_outgoing_link_ready(struct multi_context *m, struct multi_instance *mi, const unsigned int mpp_flags)
{
    struct mbuf_item item;
    bool ret = true;
    ASSERT(mi);

    /* extract from queue */
    if (mbuf_extract_item(mi->tcp_link_out_deferred, &item)) /* ciphertext IP packet */
    {
        dmsg(D_MULTI_TCP, "MULTI TCP: transmitting previously deferred packet");

        ASSERT(mi == item.instance);
        mi->context.c2.to_link = item.buffer->buf;
        ret = multi_process_outgoing_link_dowork(m, mi, mpp_flags);
        if (!ret)
        {
            mi = NULL;
        }
        mbuf_free_buf(item.buffer);
    }
    return ret;
}

static bool
multi_tcp_process_outgoing_link(struct multi_context *m, bool defer, const unsigned int mpp_flags)
{
    struct multi_instance *mi = multi_process_outgoing_link_pre(m);
    bool ret = true;

    if (mi)
    {
        if (defer || mbuf_defined(mi->tcp_link_out_deferred))
        {
            /* save to queue */
            struct buffer *buf = &mi->context.c2.to_link;
            if (BLEN(buf) > 0)
            {
                struct mbuf_buffer *mb = mbuf_alloc_buf(buf);
                struct mbuf_item item;

                set_prefix(mi);
                dmsg(D_MULTI_TCP, "MULTI TCP: queuing deferred packet");
                item.buffer = mb;
                item.instance = mi;
                mbuf_add_item(mi->tcp_link_out_deferred, &item);
                mbuf_free_buf(mb);
                buf_reset(buf);
                ret = multi_process_post(m, mi, mpp_flags);
                if (!ret)
                {
                    mi = NULL;
                }
                clear_prefix();
            }
        }
        else
        {
            ret = multi_process_outgoing_link_dowork(m, mi, mpp_flags);
            if (!ret)
            {
                mi = NULL;
            }
        }
    }
    return ret;
}

static int
multi_tcp_wait_lite(struct multi_context *m, struct multi_instance *mi, const int action, bool *tun_input_pending)
{
    struct context *c = multi_tcp_context(m, mi);
    unsigned int looking_for = 0;

    dmsg(D_MULTI_DEBUG, "MULTI TCP: multi_tcp_wait_lite a=%s mi=" ptr_format,
         pract(action),
         (ptr_type)mi);

    tv_clear(&c->c2.timeval); /* ZERO-TIMEOUT */

    switch (action)
    {
        case TA_TUN_READ:
            looking_for = TUN_READ;
            tun_input_pending = NULL;
            io_wait(c, IOW_READ_TUN);
            break;

        case TA_SOCKET_READ:
            looking_for = SOCKET_READ;
            tun_input_pending = NULL;
            io_wait(c, IOW_READ_LINK);
            break;

        case TA_TUN_WRITE:
            looking_for = TUN_WRITE;
            tun_input_pending = NULL;
            c->c2.timeval.tv_sec = 1; /* For some reason, the Linux 2.2 TUN/TAP driver hits this timeout */
            perf_push(PERF_PROC_OUT_TUN_MTCP);
            io_wait(c, IOW_TO_TUN);
            perf_pop();
            break;

        case TA_SOCKET_WRITE:
            looking_for = SOCKET_WRITE;
            io_wait(c, IOW_TO_LINK|IOW_READ_TUN_FORCE);
            break;

        default:
            msg(M_FATAL, "MULTI TCP: multi_tcp_wait_lite, unhandled action=%d", action);
    }

    if (tun_input_pending && (c->c2.event_set_status & TUN_READ))
    {
        *tun_input_pending = true;
    }

    if (c->c2.event_set_status & looking_for)
    {
        return action;
    }
    else
    {
        switch (action)
        {
            /* TCP socket output buffer is full */
            case TA_SOCKET_WRITE:
                return TA_SOCKET_WRITE_DEFERRED;

            /* TUN device timed out on accepting write */
            case TA_TUN_WRITE:
                return TA_TUN_WRITE_TIMEOUT;
        }

        return TA_UNDEF;
    }
}

static struct multi_instance *
multi_tcp_dispatch(struct multi_context *m, struct multi_instance *mi, const int action)
{
    const unsigned int mpp_flags = MPP_PRE_SELECT|MPP_RECORD_TOUCH;
    struct multi_instance *touched = mi;
    m->mpp_touched = &touched;

    dmsg(D_MULTI_DEBUG, "MULTI TCP: multi_tcp_dispatch a=%s mi=" ptr_format,
         pract(action),
         (ptr_type)mi);

    switch (action)
    {
        case TA_TUN_READ:
            read_incoming_tun(&m->top);
            if (!IS_SIG(&m->top))
            {
                multi_process_incoming_tun(m, mpp_flags);
            }
            break;

        case TA_SOCKET_READ:
        case TA_SOCKET_READ_RESIDUAL:
            ASSERT(mi);
            ASSERT(mi->context.c2.link_socket);
            set_prefix(mi);
            read_incoming_link(&mi->context);
            clear_prefix();
            if (!IS_SIG(&mi->context))
            {
                multi_process_incoming_link(m, mi, mpp_flags);
                if (!IS_SIG(&mi->context))
                {
                    stream_buf_read_setup(mi->context.c2.link_socket);
                }
            }
            break;

        case TA_TIMEOUT:
            multi_process_timeout(m, mpp_flags);
            break;

        case TA_TUN_WRITE:
            multi_process_outgoing_tun(m, mpp_flags);
            break;

        case TA_TUN_WRITE_TIMEOUT:
            multi_process_drop_outgoing_tun(m, mpp_flags);
            break;

        case TA_SOCKET_WRITE_READY:
            ASSERT(mi);
            multi_tcp_process_outgoing_link_ready(m, mi, mpp_flags);
            break;

        case TA_SOCKET_WRITE:
            multi_tcp_process_outgoing_link(m, false, mpp_flags);
            break;

        case TA_SOCKET_WRITE_DEFERRED:
            multi_tcp_process_outgoing_link(m, true, mpp_flags);
            break;

        case TA_INITIAL:
            ASSERT(mi);
            multi_tcp_set_global_rw_flags(m, mi);
            multi_process_post(m, mi, mpp_flags);
            break;

        default:
            msg(M_FATAL, "MULTI TCP: multi_tcp_dispatch, unhandled action=%d", action);
    }

    m->mpp_touched = NULL;
    return touched;
}

static int
multi_tcp_post(struct multi_context *m, struct multi_instance *mi, const int action)
{
    struct context *c = multi_tcp_context(m, mi);
    int newaction = TA_UNDEF;

#define MTP_NONE         0
#define MTP_TUN_OUT      (1<<0)
#define MTP_LINK_OUT     (1<<1)
    unsigned int flags = MTP_NONE;

    if (TUN_OUT(c))
    {
        flags |= MTP_TUN_OUT;
    }
    if (LINK_OUT(c))
    {
        flags |= MTP_LINK_OUT;
    }

    switch (flags)
    {
        case MTP_TUN_OUT|MTP_LINK_OUT:
        case MTP_TUN_OUT:
            newaction = TA_TUN_WRITE;
            break;

        case MTP_LINK_OUT:
            newaction = TA_SOCKET_WRITE;
            break;

        case MTP_NONE:
            if (mi && socket_read_residual(c->c2.link_socket))
            {
                newaction = TA_SOCKET_READ_RESIDUAL;
            }
            else
            {
                multi_tcp_set_global_rw_flags(m, mi);
            }
            break;

        default:
        {
            struct gc_arena gc = gc_new();
            msg(M_FATAL, "MULTI TCP: multi_tcp_post bad state, mi=%s flags=%d",
                multi_instance_string(mi, false, &gc),
                flags);
            gc_free(&gc);
            break;
        }
    }

    dmsg(D_MULTI_DEBUG, "MULTI TCP: multi_tcp_post %s -> %s",
         pract(action),
         pract(newaction));

    return newaction;
}

static void
multi_tcp_action(struct multi_context *m, struct multi_instance *mi, int action, bool poll)
{
    bool tun_input_pending = false;

    do
    {
        dmsg(D_MULTI_DEBUG, "MULTI TCP: multi_tcp_action a=%s p=%d",
             pract(action),
             poll);

        /*
         * If TA_SOCKET_READ_RESIDUAL, it means we still have pending
         * input packets which were read by a prior TCP recv.
         *
         * Otherwise do a "lite" wait, which means we wait with 0 timeout
         * on I/O events only related to the current instance, not
         * the big list of events.
         *
         * On our first pass, poll will be false because we already know
         * that input is available, and to call io_wait would be redundant.
         */
        if (poll && action != TA_SOCKET_READ_RESIDUAL)
        {
            const int orig_action = action;
            action = multi_tcp_wait_lite(m, mi, action, &tun_input_pending);
            if (action == TA_UNDEF)
            {
                msg(M_FATAL, "MULTI TCP: I/O wait required blocking in multi_tcp_action, action=%d", orig_action);
            }
        }

        /*
         * Dispatch the action
         */
        {
            struct multi_instance *touched = multi_tcp_dispatch(m, mi, action);

            /*
             * Signal received or TCP connection
             * reset by peer?
             */
            if (touched && IS_SIG(&touched->context))
            {
                if (mi == touched)
                {
                    mi = NULL;
                }
                multi_close_instance_on_signal(m, touched);
            }
        }

        /*
         * If dispatch produced any pending output
         * for a particular instance, point to
         * that instance.
         */
        if (m->pending)
        {
            mi = m->pending;
        }

        /*
         * Based on the effects of the action,
         * such as generating pending output,
         * possibly transition to a new action state.
         */
        action = multi_tcp_post(m, mi, action);

        /*
         * If we are finished processing the original action,
         * check if we have any TUN input.  If so, transition
         * our action state to processing this input.
         */
        if (tun_input_pending && action == TA_UNDEF)
        {
            action = TA_TUN_READ;
            mi = NULL;
            tun_input_pending = false;
            poll = false;
        }
        else
        {
            poll = true;
        }

    } while (action != TA_UNDEF);
}

static void
multi_tcp_process_io(struct multi_context *m)
{
    struct multi_tcp *mtcp = m->mtcp;
    int i;

    for (i = 0; i < mtcp->n_esr; ++i)
    {
        struct event_set_return *e = &mtcp->esr[i];

        /* incoming data for instance? */
        if (e->arg >= MTCP_N)
        {
            struct multi_instance *mi = (struct multi_instance *) e->arg;
            if (mi)
            {
                if (e->rwflags & EVENT_WRITE)
                {
                    multi_tcp_action(m, mi, TA_SOCKET_WRITE_READY, false);
                }
                else if (e->rwflags & EVENT_READ)
                {
                    multi_tcp_action(m, mi, TA_SOCKET_READ, false);
                }
            }
        }
        else
        {
#ifdef ENABLE_MANAGEMENT
            if (e->arg == MTCP_MANAGEMENT)
            {
                ASSERT(management);
                management_io(management);
            }
            else
#endif
            /* incoming data on TUN? */
            if (e->arg == MTCP_TUN)
            {
                if (e->rwflags & EVENT_WRITE)
                {
                    multi_tcp_action(m, NULL, TA_TUN_WRITE, false);
                }
                else if (e->rwflags & EVENT_READ)
                {
                    multi_tcp_action(m, NULL, TA_TUN_READ, false);
                }
            }
            /* new incoming TCP client attempting to connect? */
            else if (e->arg == MTCP_SOCKET)
            {
                struct multi_instance *mi;
                ASSERT(m->top.c2.link_socket);
                socket_reset_listen_persistent(m->top.c2.link_socket);
                mi = multi_create_instance_tcp(m);
                if (mi)
                {
                    multi_tcp_action(m, mi, TA_INITIAL, false);
                }
            }
            /* signal received? */
            else if (e->arg == MTCP_SIG)
            {
                get_signal(&m->top.sig->signal_received);
            }
#ifdef ENABLE_ASYNC_PUSH
            else if (e->arg == MTCP_FILE_CLOSE_WRITE)
            {
                multi_process_file_closed(m, MPP_PRE_SELECT | MPP_RECORD_TOUCH);
            }
#endif
        }
        if (IS_SIG(&m->top))
        {
            break;
        }
    }
    mtcp->n_esr = 0;

    /*
     * Process queued mbuf packets destined for TCP socket
     */
    {
        struct multi_instance *mi;
        while (!IS_SIG(&m->top) && (mi = mbuf_peek(m->mbuf)) != NULL)
        {
            multi_tcp_action(m, mi, TA_SOCKET_WRITE, true);
        }
    }
}

/*
 * Top level event loop for single-threaded operation.
 * TCP mode.
 */
void
tunnel_server_tcp(struct context *top)
{
    struct multi_context multi;
    int status;

    top->mode = CM_TOP;
    context_clear_2(top);

    /* initialize top-tunnel instance */
    init_instance_handle_signals(top, top->es, CC_HARD_USR1_TO_HUP);
    if (IS_SIG(top))
    {
        return;
    }

    /* initialize global multi_context object */
    multi_init(&multi, top, true, MC_SINGLE_THREADED);

    /* initialize our cloned top object */
    multi_top_init(&multi, top);

    /* initialize management interface */
    init_management_callback_multi(&multi);

    /* finished with initialization */
    initialization_sequence_completed(top, ISC_SERVER); /* --mode server --proto tcp-server */

#ifdef ENABLE_ASYNC_PUSH
    multi.top.c2.inotify_fd = inotify_init();
    if (multi.top.c2.inotify_fd < 0)
    {
        msg(D_MULTI_ERRORS | M_ERRNO, "MULTI: inotify_init error");
    }
#endif

    /* per-packet event loop */
    while (true)
    {
        perf_push(PERF_EVENT_LOOP);

        /* wait on tun/socket list */
        multi_get_timeout(&multi, &multi.top.c2.timeval);
        status = multi_tcp_wait(&multi.top, multi.mtcp);
        MULTI_CHECK_SIG(&multi);

        /* check on status of coarse timers */
        multi_process_per_second_timers(&multi);

        /* timeout? */
        if (status > 0)
        {
            /* process the I/O which triggered select */
            multi_tcp_process_io(&multi);
            MULTI_CHECK_SIG(&multi);
        }
        else if (status == 0)
        {
            multi_tcp_action(&multi, NULL, TA_TIMEOUT, false);
        }

        perf_pop();
    }

#ifdef ENABLE_ASYNC_PUSH
    close(top->c2.inotify_fd);
#endif

    /* shut down management interface */
    uninit_management_callback();

    /* save ifconfig-pool */
    multi_ifconfig_pool_persist(&multi, true);

    /* tear down tunnel instance (unless --persist-tun) */
    multi_uninit(&multi);
    multi_top_free(&multi);
    close_instance(top);
}

#endif /* if P2MP_SERVER */