rlm_tacacs_tcp.c

Go to the documentation of this file.

/*
 *   This program is is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or (at
 *   your option) any later version.
 *
 *   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 St, Fifth Floor, Boston, MA 02110-1301, USA
 */
 
/**
 * $Id: e4741f8621d0f24b4e17fb64f614985f19307149 $
 * @file rlm_tacacs_tcp.c
 * @brief TACACS+ transport
 *
 * @copyright 2023 Network RADIUS SAS (legal@networkradius.com)
 */
RCSID("$Id: e4741f8621d0f24b4e17fb64f614985f19307149 $")
 
#include <freeradius-devel/io/application.h>
#include <freeradius-devel/io/listen.h>
#include <freeradius-devel/io/pair.h>
#include <freeradius-devel/missing.h>
#include <freeradius-devel/server/connection.h>
#include <freeradius-devel/util/debug.h>
#include <freeradius-devel/util/heap.h>
#include <freeradius-devel/util/udp.h>
 
#include <sys/socket.h>
#include <sys/uio.h>
 
#include "rlm_tacacs.h"
 
/** Static configuration for the module.
 *
 */
typedef struct {
        rlm_tacacs_t            *parent;                //!< rlm_tacacs instance.
        CONF_SECTION            *config;
 
        fr_ipaddr_t             dst_ipaddr;             //!< IP of the home server.
        fr_ipaddr_t             src_ipaddr;             //!< IP we open our socket on.
        uint16_t                dst_port;               //!< Port of the home server.
        char const              *secret;                //!< Shared secret.
        size_t                  secretlen;              //!< length of secret
 
        char const              *interface;             //!< Interface to bind to.
 
        uint32_t                recv_buff;              //!< How big the kernel's receive buffer should be.
        uint32_t                send_buff;              //!< How big the kernel's send buffer should be.
 
        uint32_t                max_packet_size;        //!< Maximum packet size.
        uint16_t                max_send_coalesce;      //!< Maximum number of packets to coalesce into one mmsg call.
 
        bool                    recv_buff_is_set;       //!< Whether we were provided with a recv_buf
        bool                    send_buff_is_set;       //!< Whether we were provided with a send_buf
 
        trunk_conf_t            *trunk_conf;            //!< trunk configuration
} rlm_tacacs_tcp_t;
 
typedef struct {
        fr_event_list_t         *el;                    //!< Event list.
 
        rlm_tacacs_tcp_t const  *inst;                  //!< our instance
 
        trunk_t         *trunk;                 //!< trunk handler
} udp_thread_t;
 
typedef struct {
        trunk_request_t *treq;
        rlm_rcode_t             rcode;                  //!< from the transport
} udp_result_t;
 
typedef struct udp_request_s udp_request_t;
 
typedef struct {
        uint8_t                 *read;                  //!< where we read data from
        uint8_t                 *write;                 //!< where we write data to
        uint8_t                 *end;                   //!< end of the buffer
        uint8_t                 *data;                  //!< actual data
} tcp_buffer_t;
 
/** Track the handle, which is tightly correlated with the FD
 *
 */
typedef struct {
        char const              *name;                  //!< From IP PORT to IP PORT.
        char const              *module_name;           //!< the module that opened the connection
 
        int                     fd;                     //!< File descriptor.
 
        trunk_request_t         **coalesced;            //!< Outbound coalesced requests.
 
        size_t                  send_buff_actual;       //!< What we believe the maximum SO_SNDBUF size to be.
                                                        ///< We don't try and encode more packet data than this
                                                        ///< in one go.
 
        rlm_tacacs_tcp_t const  *inst;                  //!< Our module instance.
        udp_thread_t            *thread;
 
        uint32_t                session_id;             //!< for TACACS+ "security".
 
        uint32_t                max_packet_size;        //!< Our max packet size. may be different from the parent.
 
        fr_ipaddr_t             src_ipaddr;             //!< Source IP address.  May be altered on bind
                                                        //!< to be the actual IP address packets will be
                                                        //!< sent on.  This is why we can't use the inst
                                                        //!< src_ipaddr field.
        uint16_t                src_port;               //!< Source port specific to this connection.
                                                        //!< @todo - not set by socket_client_tcp()
 
        tcp_buffer_t            recv;                   //!< receive buffer
        tcp_buffer_t            send;                   //!< send buffer
 
        int                     id;                     //!< starts at 1.
        int                     active;                 //!< active packets
        trunk_request_t         *tracking[UINT8_MAX];   //!< all sequential!
 
        fr_time_t               mrs_time;               //!< Most recent sent time which had a reply.
        fr_time_t               last_reply;             //!< When we last received a reply.
        fr_time_t               first_sent;             //!< first time we sent a packet since going idle
        fr_time_t               last_sent;              //!< last time we sent a packet.
        fr_time_t               last_idle;              //!< last time we had nothing to do
 
        fr_timer_t      *zombie_ev;             //!< Zombie timeout.
 
        trunk_connection_t      *tconn;                 //!< trunk connection
} udp_handle_t;
 
 
/** Connect request_t to local tracking structure
 *
 */
struct udp_request_s {
        uint32_t                priority;               //!< copied from request->async->priority
        fr_time_t               recv_time;              //!< copied from request->async->recv_time
 
        uint8_t                 code;                   //!< Packet code.
        uint8_t                 id;                     //!< Last ID assigned to this packet.
        bool                    outstanding;            //!< are we waiting for a reply?
 
        uint8_t                 *packet;                //!< Packet we write to the network.
        size_t                  packet_len;             //!< Length of the packet.
 
        fr_timer_t      *ev;                    //!< timer for retransmissions
        fr_retry_t              retry;                  //!< retransmission timers
};
 
static const conf_parser_t module_config[] = {
        { FR_CONF_OFFSET_TYPE_FLAGS("ipaddr", FR_TYPE_COMBO_IP_ADDR, 0, rlm_tacacs_tcp_t, dst_ipaddr), },
        { FR_CONF_OFFSET_TYPE_FLAGS("ipv4addr", FR_TYPE_IPV4_ADDR, 0, rlm_tacacs_tcp_t, dst_ipaddr) },
        { FR_CONF_OFFSET_TYPE_FLAGS("ipv6addr", FR_TYPE_IPV6_ADDR, 0, rlm_tacacs_tcp_t, dst_ipaddr) },
 
        { FR_CONF_OFFSET("port", rlm_tacacs_tcp_t, dst_port) },
 
        { FR_CONF_OFFSET("secret", rlm_tacacs_tcp_t, secret) }, /* can be NULL */
 
        { FR_CONF_OFFSET("interface", rlm_tacacs_tcp_t, interface) },
 
        { FR_CONF_OFFSET_IS_SET("recv_buff", FR_TYPE_UINT32, 0, rlm_tacacs_tcp_t, recv_buff) },
        { FR_CONF_OFFSET_IS_SET("send_buff", FR_TYPE_UINT32, 0, rlm_tacacs_tcp_t, send_buff) },
 
        { FR_CONF_OFFSET("max_packet_size", rlm_tacacs_tcp_t, max_packet_size), .dflt = STRINGIFY(FR_MAX_PACKET_SIZE) },
        { FR_CONF_OFFSET("max_send_coalesce", rlm_tacacs_tcp_t, max_send_coalesce), .dflt = "1024" },
 
        { FR_CONF_OFFSET_TYPE_FLAGS("src_ipaddr", FR_TYPE_COMBO_IP_ADDR, 0, rlm_tacacs_tcp_t, src_ipaddr) },
        { FR_CONF_OFFSET_TYPE_FLAGS("src_ipv4addr", FR_TYPE_IPV4_ADDR, 0, rlm_tacacs_tcp_t, src_ipaddr) },
        { FR_CONF_OFFSET_TYPE_FLAGS("src_ipv6addr", FR_TYPE_IPV6_ADDR, 0, rlm_tacacs_tcp_t, src_ipaddr) },
 
        CONF_PARSER_TERMINATOR
};
 
static fr_dict_t const *dict_tacacs;
 
extern fr_dict_autoload_t rlm_tacacs_tcp_dict[];
fr_dict_autoload_t rlm_tacacs_tcp_dict[] = {
        { .out = &dict_tacacs, .proto = "tacacs" },
        { NULL }
};
 
static fr_dict_attr_t const *attr_packet_type;
static fr_dict_attr_t const *attr_packet_hdr;
static fr_dict_attr_t const *attr_session_id;
 
extern fr_dict_attr_autoload_t rlm_tacacs_tcp_dict_attr[];
fr_dict_attr_autoload_t rlm_tacacs_tcp_dict_attr[] = {
        { .out = &attr_packet_type, .name = "Packet-Type", .type = FR_TYPE_UINT32, .dict = &dict_tacacs },
        { .out = &attr_packet_hdr, .name = "Packet", .type = FR_TYPE_STRUCT, .dict = &dict_tacacs },
        { .out = &attr_session_id, .name = "Packet.Session-ID", .type = FR_TYPE_UINT32, .dict = &dict_tacacs },
        { NULL }
};
 
/** Clear out any connection specific resources from a udp request
 *
 */
static void udp_request_reset(udp_handle_t *h, udp_request_t *u)
{
        u->packet = NULL;
 
        fr_assert(h->active > 0);
        fr_assert(h->tracking[u->id] != NULL);
        fr_assert(h->tracking[u->id]->preq == u);
 
        h->tracking[u->id] = NULL;
        u->outstanding = false;
        h->active--;
 
        if (u->ev) (void)fr_timer_delete(&u->ev);
 
        /*
         *      We've sent 255 packets, and received all replies.  Shut the connection down.
         *
         *      Welcome to the insanity that is TACACS+.
         */
        if ((h->active == 0) && (h->id > 255)) {
                trunk_connection_signal_reconnect(h->tconn, CONNECTION_EXPIRED);
        }
}
 
 
/** Free a connection handle, closing associated resources
 *
 */
static int _udp_handle_free(udp_handle_t *h)
{
        fr_assert(h->fd >= 0);
 
        fr_event_fd_delete(h->thread->el, h->fd, FR_EVENT_FILTER_IO);
 
        if (shutdown(h->fd, SHUT_RDWR) < 0) {
                DEBUG3("%s - Failed shutting down connection %s: %s",
                       h->module_name, h->name, fr_syserror(errno));
        }
 
        if (close(h->fd) < 0) {
                DEBUG3("%s - Failed closing connection %s: %s",
                       h->module_name, h->name, fr_syserror(errno));
        }
 
        h->fd = -1;
 
        DEBUG("%s - Connection closed - %s", h->module_name, h->name);
 
        return 0;
}
 
/** Initialise a new outbound connection
 *
 * @param[out] h_out    Where to write the new file descriptor.
 * @param[in] conn      to initialise.
 * @param[in] uctx      A #udp_thread_t
 */
static connection_state_t conn_init(void **h_out, connection_t *conn, void *uctx)
{
        int                     fd;
        udp_handle_t            *h;
        udp_thread_t            *thread = talloc_get_type_abort(uctx, udp_thread_t);
 
        MEM(h = talloc_zero(conn, udp_handle_t));
        h->thread = thread;
        h->inst = thread->inst;
        h->module_name = h->inst->parent->name;
        h->src_ipaddr = h->inst->src_ipaddr;
        h->src_port = 0;
        h->max_packet_size = h->inst->max_packet_size;
        h->last_idle = fr_time();
 
        h->id = 1;              /* clients send odd sequence numbers */
        h->session_id = fr_rand();
 
        /*
         *      Initialize the buffer of coalesced packets we're doing to write.
         */
        h->coalesced = talloc_zero_array(h, trunk_request_t *, h->inst->max_send_coalesce);
 
        /*
         *      Open the outgoing socket.
         */
        fd = fr_socket_client_tcp(NULL, &h->src_ipaddr, &h->inst->dst_ipaddr, h->inst->dst_port, true);
        if (fd < 0) {
                PERROR("%s - Failed opening socket", h->module_name);
                talloc_free(h);
                return CONNECTION_STATE_FAILED;
        }
 
        /*
         *      Set the connection name.
         */
        h->name = fr_asprintf(h, "proto tcp local %pV port %u remote %pV port %u",
                              fr_box_ipaddr(h->src_ipaddr), h->src_port,
                              fr_box_ipaddr(h->inst->dst_ipaddr), h->inst->dst_port);
 
        talloc_set_destructor(h, _udp_handle_free);
 
#ifdef SO_RCVBUF
        if (h->inst->recv_buff_is_set) {
                int opt;
 
                opt = h->inst->recv_buff;
                if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &opt, sizeof(int)) < 0) {
                        WARN("%s - Failed setting 'SO_RCVBUF': %s", h->module_name, fr_syserror(errno));
                }
        }
#endif
 
#ifdef SO_SNDBUF
        {
                int opt;
                socklen_t socklen = sizeof(int);
 
                if (h->inst->send_buff_is_set) {
                        opt = h->inst->send_buff;
                        if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &opt, sizeof(int)) < 0) {
                                WARN("%s - Failed setting 'SO_SNDBUF', write performance may be sub-optimal: %s",
                                     h->module_name, fr_syserror(errno));
                        }
                }
 
                if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &opt, &socklen) < 0) {
                        WARN("%s - Failed getting 'SO_SNDBUF', write performance may be sub-optimal: %s",
                             h->module_name, fr_syserror(errno));
 
                        /*
                         *      This controls how many packets we attempt
                         *      to send at once.  Nothing bad happens if
                         *      we get it wrong, but the user may see
                         *      ENOBUFS errors at high packet rates.
                         *
                         *      Since this is TACACS, we have small
                         *      packets and a maximum of 255 packets
                         *      per connection.  So don't set this too large.
                         */
                        if (h->inst->send_buff_is_set) {
                                h->send_buff_actual = h->inst->send_buff;
                        } else {
                                h->send_buff_actual = h->max_packet_size * h->inst->max_send_coalesce;
                                if (h->send_buff_actual > 256*1024) h->send_buff_actual = 256*1024;
                        }
 
                        WARN("%s - Max coalesced outbound data will be %zu bytes", h->module_name,
                             h->send_buff_actual);
                } else {
#ifdef __linux__
                        /*
                         *      Linux doubles the buffer when you set it
                         *      to account for "overhead".
                         */
                        h->send_buff_actual = ((size_t)opt) / 2;
#else
                        h->send_buff_actual = (size_t)opt;
#endif
                }
        }
#else
        h->send_buff_actual = h->inst->send_buff_is_set ?
                              h->inst_send_buff : h->max_packet_size * h->inst->max_send_coalesce;
 
        WARN("%s - Modifying 'SO_SNDBUF' value is not supported on this system, "
             "write performance may be sub-optimal", h->module_name);
        WARN("%s - Max coalesced outbound data will be %zu bytes", h->module_name, h->inst->send_buff_actual);
#endif
 
        /*
         *      Allow receiving of 2 max-sized packets.  In practice, most packets will be less than this.
         */
        MEM(h->recv.data = talloc_array(h, uint8_t, h->max_packet_size * 2));
        h->recv.read = h->recv.write = h->recv.data;
        h->recv.end = h->recv.data + h->max_packet_size * 2;
 
        /*
         *      Use the system SO_SNDBUF for how many packets to send at once.  In most circumstances the
         *      packets are small, and widely separated in time, and we really only need a very small buffer.
         */
        MEM(h->send.data = talloc_array(h, uint8_t, h->send_buff_actual));
        h->send.read = h->send.write = h->send.data;
        h->send.end = h->send.data + h->send_buff_actual;
 
        h->fd = fd;
 
        /*
         *      Signal the connection
         *      as open as soon as it becomes writable.
         */
        connection_signal_on_fd(conn, fd);
 
        *h_out = h;
 
        // @todo - initialize the tracking memory, etc.
        // i.e. histograms (or hyperloglog) of packets, so we can see
        // which connections / home servers are fast / slow.
 
        return CONNECTION_STATE_CONNECTING;
}
 
/** Shutdown/close a file descriptor
 *
 */
static void conn_close(UNUSED fr_event_list_t *el, void *handle, UNUSED void *uctx)
{
        udp_handle_t *h = talloc_get_type_abort(handle, udp_handle_t);
 
        /*
         *      There's tracking entries still allocated
         *      this is bad, they should have all been
         *      released.
         */
        fr_assert(!h->active);
 
        DEBUG4("Freeing rlm_tacacs_tcp handle %p", handle);
 
        talloc_free(h);
}
 
CC_NO_UBSAN(function) /* UBSAN: false positive - public vs private connection_t trips --fsanitize=function*/
static connection_t *thread_conn_alloc(trunk_connection_t *tconn, fr_event_list_t *el,
                                          connection_conf_t const *conf,
                                          char const *log_prefix, void *uctx)
{
        connection_t            *conn;
        udp_thread_t            *thread = talloc_get_type_abort(uctx, udp_thread_t);
 
        conn = connection_alloc(tconn, el,
                                   &(connection_funcs_t){
                                        .init = conn_init,
                                        .close = conn_close,
                                   },
                                   conf,
                                   log_prefix,
                                   thread);
        if (!conn) {
                PERROR("%s - Failed allocating state handler for new connection", thread->inst->parent->name);
                return NULL;
        }
 
        return conn;
}
 
/** Connection errored
 *
 * We were signalled by the event loop that a fatal error occurred on this connection.
 *
 * @param[in] el        The event list signalling.
 * @param[in] fd        that errored.
 * @param[in] flags     El flags.
 * @param[in] fd_errno  The nature of the error.
 * @param[in] uctx      The trunk connection handle (tconn).
 */
static void conn_error(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, int fd_errno, void *uctx)
{
        trunk_connection_t      *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
        connection_t            *conn = tconn->conn;
        udp_handle_t            *h = talloc_get_type_abort(conn->h, udp_handle_t);
 
        ERROR("%s - Connection %s failed: %s", h->module_name, h->name, fr_syserror(fd_errno));
 
        connection_signal_reconnect(conn, CONNECTION_FAILED);
}
 
CC_NO_UBSAN(function) /* UBSAN: false positive - public vs private connection_t trips --fsanitize=function*/
static void thread_conn_notify(trunk_connection_t *tconn, connection_t *conn,
                               fr_event_list_t *el,
                               trunk_connection_event_t notify_on, UNUSED void *uctx)
{
        udp_handle_t            *h = talloc_get_type_abort(conn->h, udp_handle_t);
        fr_event_fd_cb_t        read_fn = NULL;
        fr_event_fd_cb_t        write_fn = NULL;
 
        switch (notify_on) {
        case TRUNK_CONN_EVENT_NONE:
                return;
 
        case TRUNK_CONN_EVENT_READ:
                read_fn = trunk_connection_callback_readable;
                break;
 
        case TRUNK_CONN_EVENT_WRITE:
                write_fn = trunk_connection_callback_writable;
                break;
 
        case TRUNK_CONN_EVENT_BOTH:
                read_fn = trunk_connection_callback_readable;
                write_fn = trunk_connection_callback_writable;
                break;
 
        }
 
        if (fr_event_fd_insert(h, NULL, el, h->fd,
                               read_fn,
                               write_fn,
                               conn_error,
                               tconn) < 0) {
                PERROR("%s - Failed inserting FD event", h->module_name);
 
                /*
                 *      May free the connection!
                 */
                trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
        }
}
 
/*
 *  Return negative numbers to put 'a' at the top of the heap.
 *  Return positive numbers to put 'b' at the top of the heap.
 *
 *  We want the value with the lowest timestamp to be prioritized at
 *  the top of the heap.
 */
static int8_t request_prioritise(void const *one, void const *two)
{
        udp_request_t const *a = one;
        udp_request_t const *b = two;
        int8_t ret;
 
        /*
         *      Larger priority is more important.
         */
        ret = CMP(a->priority, b->priority);
        if (ret != 0) return ret;
 
        /*
         *      Smaller timestamp (i.e. earlier) is more important.
         */
        return CMP_PREFER_SMALLER(fr_time_unwrap(a->recv_time), fr_time_unwrap(b->recv_time));
}
 
/** Decode response packet data, extracting relevant information and validating the packet
 *
 * @param[in] ctx                       to allocate pairs in.
 * @param[out] reply                    Pointer to head of pair list to add reply attributes to.
 * @param[out] response_code            The type of response packet.
 * @param[in] h                         connection handle.
 * @param[in] request                   the request.
 * @param[in] u                         UDP request.
 * @param[in] data                      to decode.
 * @param[in] data_len                  Length of input data.
 * @return
 *      - <0 on error
 *      - >0 for how many bytes were decoded
 */
static ssize_t decode(TALLOC_CTX *ctx, fr_pair_list_t *reply, uint8_t *response_code,
                      udp_handle_t *h, request_t *request, udp_request_t *u,
                      uint8_t *data, size_t data_len)
{
        rlm_tacacs_tcp_t const *inst = h->thread->inst;
        ssize_t                 packet_len;
        int                     code;
 
        *response_code = 0;     /* Initialise to keep the rest of the code happy */
 
        /*
         *      Check the session ID here, because we've lost the original packet.
         */
        if (h->session_id != fr_nbo_to_uint32(data + 4)) {
                REDEBUG("Session ID %08x does not match expected number %08x",
                        fr_nbo_to_uint32(data + 4), h->session_id);
        }
 
        /*
         *      Decode the attributes, in the context of the reply.
         *      This only fails if the packet is strangely malformed,
         *      or if we run out of memory.
         */
        packet_len = fr_tacacs_decode(ctx, reply, NULL, data, data_len, NULL, inst->secret, inst->secretlen, &code);
        if (packet_len < 0) {
                RPEDEBUG("Failed decoding TACACS+ reply packet");
                fr_pair_list_free(reply);
                return -1;
        }
 
        RDEBUG("Received %s ID %d length %ld reply packet on connection %s",
               fr_tacacs_packet_names[code], code, packet_len, h->name);
        log_request_pair_list(L_DBG_LVL_2, request, NULL, reply, NULL);
 
        *response_code = code;
 
        /*
         *      Fixup retry times
         */
        if (fr_time_gt(u->retry.start, h->mrs_time)) h->mrs_time = u->retry.start;
 
        return packet_len;
}
 
static int encode(udp_handle_t *h, request_t *request, udp_request_t *u)
{
        ssize_t                 packet_len;
        rlm_tacacs_tcp_t const *inst = h->inst;
        fr_pair_t               *hdr, *vp;
 
        fr_assert(inst->parent->allowed[u->code]);
        fr_assert(!u->packet);
        fr_assert(!u->outstanding);
 
        /*
         *      Encode the packet in the outbound buffer.
         */
        u->packet = h->send.write;
 
        /*
         *      Set the session ID, if it hasn't already been set.
         */
        hdr = fr_pair_find_by_da(&request->request_pairs, NULL, attr_packet_hdr);
        if (!hdr) hdr = request->request_ctx;
 
        vp = fr_pair_find_by_da_nested(&hdr->vp_group, NULL, attr_session_id);
        if (!vp) {
                MEM(vp = fr_pair_afrom_da(hdr, attr_session_id));
 
                vp->vp_uint32 = h->session_id;
                fr_pair_append(&hdr->vp_group, vp);
                fr_pair_list_sort(&hdr->vp_group, fr_pair_cmp_by_parent_num);
        }
 
        /*
         *      Encode the packet.
         */
        packet_len = fr_tacacs_encode(&FR_DBUFF_TMP(u->packet, (size_t) inst->max_packet_size), NULL,
                                      inst->secret, inst->secretlen, request->reply->code, &request->request_pairs);
        if (packet_len < 0) {
                RPERROR("Failed encoding packet");
                return -1;
        }
 
        /*
         *      Update the ID and the actual packet length;
         */
        u->packet[1] = u->id;
        u->packet_len = packet_len;
        u->outstanding = true;
 
//      fr_tacacs_packet_log_hex(&default_log, u->packet);
 
        return 0;
}
 
 
/** Revive a connection after "revive_interval"
 *
 */
static void revive_timeout(UNUSED fr_timer_list_t *tl, UNUSED fr_time_t now, void *uctx)
{
        trunk_connection_t      *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
        udp_handle_t            *h = talloc_get_type_abort(tconn->conn->h, udp_handle_t);
 
        INFO("%s - Reviving connection %s", h->module_name, h->name);
        trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
}
 
/** Mark a connection dead after "zombie_interval"
 *
 */
static void zombie_timeout(fr_timer_list_t *tl, fr_time_t now, void *uctx)
{
        trunk_connection_t      *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
        udp_handle_t            *h = talloc_get_type_abort(tconn->conn->h, udp_handle_t);
 
        INFO("%s - No replies during 'zombie_period', marking connection %s as dead", h->module_name, h->name);
 
        /*
         *      Don't use this connection, and re-queue all of its
         *      requests onto other connections.
         */
        trunk_connection_signal_inactive(tconn);
        (void) trunk_connection_requests_requeue(tconn, TRUNK_REQUEST_STATE_ALL, 0, false);
 
        /*
         *      Revive the connection after a time.
         */
        if (fr_timer_at(h, tl, &h->zombie_ev,
                        fr_time_add(now, h->inst->parent->revive_interval), false, revive_timeout, h) < 0) {
                ERROR("Failed inserting revive timeout for connection");
                trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
        }
}
 
 
/** See if the connection is zombied.
 *
 *      We check for zombie when major events happen:
 *
 *      1) request hits its final timeout
 *      2) request timer hits, and it needs to be retransmitted
 *      3) a DUP packet comes in, and the request needs to be retransmitted
 *      4) we're sending a packet.
 *
 *  There MIGHT not be retries configured, so we MUST check for zombie
 *  when any new packet comes in.  Similarly, there MIGHT not be new
 *  packets, but retries are configured, so we have to check there,
 *  too.
 *
 *  Also, the socket might not be writable for a while.  There MIGHT
 *  be a long time between getting the timer / DUP signal, and the
 *  request finally being written to the socket.  So we need to check
 *  for zombie at BOTH the timeout and the mux / write function.
 *
 * @return
 *      - true if the connection is zombie.
 *      - false if the connection is not zombie.
 */
static bool check_for_zombie(fr_timer_list_t *tl, trunk_connection_t *tconn, fr_time_t now, fr_time_t last_sent)
{
        udp_handle_t    *h = talloc_get_type_abort(tconn->conn->h, udp_handle_t);
 
        /*
         *      If we're already zombie, don't go to zombie
         *
         */
        if (h->zombie_ev) return true;
 
        if (fr_time_eq(now, fr_time_wrap(0))) now = fr_time();
 
        /*
         *      We received a reply since this packet was sent, the connection isn't zombie.
         */
        if (fr_time_gteq(h->last_reply, last_sent)) return false;
 
        /*
         *      If we've seen ANY response in the allowed window, then the connection is still alive.
         */
        if (fr_time_gt(last_sent, fr_time_wrap(0)) &&
            (fr_time_lt(fr_time_add(last_sent, h->inst->parent->response_window), now))) return false;
 
        /*
         *      Mark the connection as inactive, but keep sending
         *      packets on it.
         */
        WARN("%s - Entering Zombie state - connection %s", h->module_name, h->name);
        trunk_connection_signal_inactive(tconn);
 
        if (fr_timer_at(h, tl, &h->zombie_ev, fr_time_add(now, h->inst->parent->zombie_period),
                        false, zombie_timeout, h) < 0) {
                ERROR("Failed inserting zombie timeout for connection");
                trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
        }
 
        return true;
}
 
/** Handle retries.
 *
 *  Note that with TCP we don't actually retry on this particular connection, but the retry timer allows us to
 *  fail over from one connection to another when a connection fails.
 */
static void request_retry(fr_timer_list_t *tl, fr_time_t now, void *uctx)
{
        trunk_request_t         *treq = talloc_get_type_abort(uctx, trunk_request_t);
        udp_request_t           *u = talloc_get_type_abort(treq->preq, udp_request_t);
        udp_result_t            *r = talloc_get_type_abort(treq->rctx, udp_result_t);
        request_t               *request = treq->request;
        trunk_connection_t      *tconn = treq->tconn;
 
        fr_assert(treq->state == TRUNK_REQUEST_STATE_SENT);             /* No other states should be timing out */
        fr_assert(treq->preq);                                          /* Must still have a protocol request */
        fr_assert(tconn);
 
        switch (fr_retry_next(&u->retry, now)) {
        /*
         *      Queue the request for retransmission.
         *
         *      @todo - set up "next" timer here, instead of in
         *      request_mux() ?  That way we can catch the case of
         *      packets sitting in the queue for extended periods of
         *      time, and still run the timers.
         */
        case FR_RETRY_CONTINUE:
                trunk_request_requeue(treq);
                return;
 
        case FR_RETRY_MRD:
                REDEBUG("Reached maximum_retransmit_duration (%pVs > %pVs), failing request",
                        fr_box_time_delta(fr_time_sub(now, u->retry.start)), fr_box_time_delta(u->retry.config->mrd));
                break;
 
        case FR_RETRY_MRC:
                REDEBUG("Reached maximum_retransmit_count (%u > %u), failing request",
                        u->retry.count, u->retry.config->mrc);
                break;
        }
 
        r->rcode = RLM_MODULE_FAIL;
        trunk_request_signal_complete(treq);
 
        check_for_zombie(tl, tconn, now, u->retry.start);
}
 
CC_NO_UBSAN(function) /* UBSAN: false positive - public vs private connection_t trips --fsanitize=function*/
static void request_mux(fr_event_list_t *el,
                        trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
{
        udp_handle_t            *h = talloc_get_type_abort(conn->h, udp_handle_t);
        rlm_tacacs_tcp_t const  *inst = h->inst;
        ssize_t                 sent;
        uint16_t                i, queued;
        uint8_t const           *written;
        uint8_t                 *partial;
 
        /*
         *      Encode multiple packets in preparation for transmission with write()
         */
        for (i = 0, queued = 0; (i < inst->max_send_coalesce); i++) {
                trunk_request_t *treq;
                udp_request_t           *u;
                request_t               *request;
 
                if (unlikely(trunk_connection_pop_request(&treq, tconn) < 0)) return;
 
                /*
                 *      No more requests to send
                 */
                if (!treq) break;
 
                /*
                 *      The partial write MUST be the first one popped off of the request list.
                 *
                 *      If we have a partial packet, then we know that there's partial data in the output
                 *      buffer.  However, the request MAY still be freed or timed out before we can write the
                 *      data.  As a result, we ignore the udp_request_t, and just keep writing the data.
                 */
                if (treq->state == TRUNK_REQUEST_STATE_PARTIAL) {
                        fr_assert(h->send.read == h->send.data);
                        fr_assert(h->send.write > h->send.read);
 
                        fr_assert(i == 0);
 
                        h->coalesced[0] = treq;
                        goto next;
                }
 
                /*
                 *      The request must still be pending.
                 */
                fr_assert(treq->state == TRUNK_REQUEST_STATE_PENDING);
 
                request = treq->request;
                u = talloc_get_type_abort(treq->preq, udp_request_t);
 
                /*
                 *      We'd like to retransmit the packet on this connection, but it's TCP so we don't.
                 *
                 *      The retransmission timers are really there to move the packet to a new connection if
                 *      the current connection is dead.
                 */
                if (u->outstanding) continue;
 
                /*
                 *      Not enough room for a full-sized packet, stop encoding packets
                 */
                if ((h->send.end - h->send.write) < inst->max_packet_size) {
                        break;
                }
 
                /*
                 *      Start retransmissions from when the socket is writable.
                 */
                fr_retry_init(&u->retry, fr_time(), &h->inst->parent->retry);
                fr_assert(fr_time_delta_ispos(u->retry.rt));
                fr_assert(fr_time_gt(u->retry.next, fr_time_wrap(0)));
 
                /*
                 *      Set up the packet for encoding.
                 */
                u->id = h->id;
                h->tconn = tconn;
 
                h->tracking[u->id] = treq;
                h->id += 2;
                h->active++;
 
                RDEBUG("Sending %s ID %d length %ld over connection %s",
                       fr_tacacs_packet_names[u->code], u->id, u->packet_len, h->name);
 
                if (encode(h, request, u) < 0) {
                        /*
                         *      Need to do this because request_conn_release
                         *      may not be called.
                         */
                        udp_request_reset(h, u);
                        trunk_request_signal_fail(treq);
                        continue;
                }
                RHEXDUMP3(u->packet, u->packet_len, "Encoded packet");
 
                log_request_pair_list(L_DBG_LVL_2, request, NULL, &request->request_pairs, NULL);
 
                /*
                 *      Remember that we've encoded this packet.
                 */
                h->coalesced[queued] = treq;
                h->send.write += u->packet_len;
 
                fr_assert(h->send.write <= h->send.end);
 
                /*
                 *      If we just hit this limit, stop using the connection.
                 *
                 *      When we've received all replies (or timeouts), we'll close the connections.
                 */
                if (h->id > 255) {
                        trunk_connection_signal_inactive(tconn);
                }
 
        next:
                /*
                 *      Tell the trunk API that this request is now in
                 *      the "sent" state.  And we don't want to see
                 *      this request again. The request hasn't actually
                 *      been sent, but it's the only way to get at the
                 *      next entry in the heap.
                 */
                trunk_request_signal_sent(treq);
                queued++;
        }
 
        if (queued == 0) return;
 
        /*
         *      Verify nothing accidentally freed the connection handle
         */
        (void)talloc_get_type_abort(h, udp_handle_t);
 
        /*
         *      Send the packets as one system call.
         */
        sent = write(h->fd, h->send.read, h->send.write - h->send.read);
        if (sent < 0) {         /* Error means no messages were sent */
                /*
                 *      Temporary conditions
                 */
                switch (errno) {
#if defined(EWOULDBLOCK) && (EWOULDBLOCK != EAGAIN)
                case EWOULDBLOCK:       /* No outbound packet buffers, maybe? */
#endif
                case EAGAIN:            /* No outbound packet buffers, maybe? */
                case EINTR:             /* Interrupted by signal */
                case ENOBUFS:           /* No outbound packet buffers, maybe? */
                case ENOMEM:            /* malloc failure in kernel? */
                        WARN("%s - Failed sending data over connection %s: %s",
                             h->module_name, h->name, fr_syserror(errno));
                        sent = 0;
                        break;
 
                /*
                 *      Will re-queue any 'sent' requests, so we don't
                 *      have to do any cleanup.
                 */
                default:
                        ERROR("%s - Failed sending data over connection %s: %s",
                              h->module_name, h->name, fr_syserror(errno));
                        trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
                        return;
                }
        }
 
        written = h->send.read + sent;
        partial = h->send.read;
 
        /*
         *      For all messages that were actually sent by writev()
         *      start the request timer.
         */
        for (i = 0; i < queued; i++) {
                trunk_request_t *treq = h->coalesced[i];
                udp_request_t           *u;
                request_t               *request;
 
                /*
                 *      We *think* we sent this, but we might not had :(
                 */
                fr_assert(treq->state == TRUNK_REQUEST_STATE_SENT);
 
                request = treq->request;
                u = talloc_get_type_abort(treq->preq, udp_request_t);
 
                /*
                 *      This packet ends before the piece we've
                 *      written, so we've written all of it.
                 */
                if (u->packet + u->packet_len <= written) {
                        h->last_sent = u->retry.start;
                        if (fr_time_lteq(h->first_sent, h->last_idle)) h->first_sent = h->last_sent;
 
                        if (fr_timer_at(u, el->tl, &u->ev, u->retry.next, false, request_retry, treq) < 0) {
                                RERROR("Failed inserting retransmit timeout for connection");
                                trunk_request_signal_fail(treq);
                        }
 
                        /*
                         *      If the packet doesn't get a response, then the timer will hit
                         *      and will retransmit.
                         */
                        u->outstanding = true;
                        continue;
                }
 
                /*
                 *      The packet starts before the piece we've written, BUT ends after the written piece.
                 *
                 *      We only wrote part of this packet, remember the partial packet we wrote.  Note that
                 *      we only track the packet data, and not the udp_request_t.  The underlying request (and
                 *      u) may disappear at any time, even if there's still data in the buffer.
                 *
                 *      Then, signal that isn't a partial packet, and stop processing the queue, as we know
                 *      that the next packet wasn't written.
                 */
                if (u->packet < written) {
                        size_t skip = written - u->packet;
                        size_t left = u->packet_len - skip;
 
                        fr_assert(u->packet + u->packet_len > written);
 
                        memmove(h->send.data, u->packet, left);
 
                        fr_assert(h->send.read == h->send.data);
                        partial = h->send.data + left;
                        u->outstanding = true;
 
                        trunk_request_signal_partial(h->coalesced[i]);
                        continue;
                }
 
                /*
                 *      The packet starts after the piece we've written, so we haven't written any of it.
                 *
                 *      Requests that weren't sent get re-enqueued.  Which means that they get re-encoded, but
                 *      oh well.
                 *
                 *      The cancel logic runs as per-normal and cleans up
                 *      the request ready for sending again...
                 */
                trunk_request_requeue(h->coalesced[i]);
                fr_assert(!u->outstanding); /* must have called udp_request_requeue() */
        }
 
        /*
         *      Remember where to write the next packet.  Either at the start of the buffer, or after the one
         *      which was partially written.
         */
        h->send.write = partial;
}
 
static void request_demux(UNUSED fr_event_list_t *el, trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
{
        udp_handle_t            *h = talloc_get_type_abort(conn->h, udp_handle_t);
        bool                    do_read = true;
 
        DEBUG3("%s - Reading data for connection %s", h->module_name, h->name);
 
        while (true) {
                ssize_t slen;
                size_t                  available, used, packet_len;
 
                trunk_request_t *treq;
                request_t               *request;
                udp_request_t           *u;
                udp_result_t            *r;
                uint8_t                 code = 0;
                fr_pair_list_t          reply;
 
                /*
                 *      Ensure that we can read at least one max-sized packet.
                 *
                 *      If not, move the trailing bytes to the start of the buffer, and reset the read/write
                 *      pointers to the start of the buffer.  Note that the read buffer has to be at least 2x
                 *      max_packet_size.
                 */
                available = h->recv.end - h->recv.read;
                if (available < h->inst->max_packet_size) {
                        fr_assert(h->recv.data + h->inst->max_packet_size < h->recv.read);
 
                        used = h->recv.write - h->recv.read;
 
                        memcpy(h->recv.data, h->recv.read, used);
                        h->recv.read = h->recv.data;
                        h->recv.write = h->recv.read + used;
                }
 
                /*
                 *      Read as much data as possible.
                 *
                 *      We don't need to call read() on every round through the loop.  Instead, we call it
                 *      only when this function first gets called, OR if the read stopped at the end of the
                 *      buffer.
                 *
                 *      This allows us to read a large amount of data at once, and then process multiple
                 *      packets without calling read() too many times.
                 */
                if (do_read) {
                        slen = read(h->fd, h->recv.write, h->recv.end - h->recv.write);
                        if (slen < 0) {
                                if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) return;
 
                                ERROR("%s - Failed reading response from socket: %s",
                                      h->module_name, fr_syserror(errno));
                                trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
                                return;
                        }
 
                        h->recv.write += slen;
                        do_read = (h->recv.write == h->recv.end);
                }
 
                used = h->recv.write - h->recv.read;
 
                /*
                 *      We haven't received a full header, read more or return.
                 */
                if (used < sizeof(fr_tacacs_packet_hdr_t)) {
                        if (do_read) continue;
                        return;
                }
 
                /*
                 *      The packet contains a 4 octet length in the
                 *      header, but the header bytes aren't included
                 *      in the 4 octet length field.
                 */
                packet_len = fr_nbo_to_uint32(h->recv.read + 8) + FR_HEADER_LENGTH;
 
                /*
                 *      The packet is too large, reject it.
                 */
                if (packet_len > h->inst->max_packet_size) {
                        ERROR("%s - Packet is larger than max_packet_size",
                              h->module_name);
                        trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
                        return;
                }
 
                /*
                 *      We haven't received the full packet, read more or return.
                 */
                if (used < packet_len) {
                        if (do_read) continue;
                        return;
                }
 
                fr_assert(h->recv.read + packet_len <= h->recv.end);
 
                /*
                 *      TACACS+ doesn't care about packet codes.  All packet of the codes share the same ID
                 *      space.
                 */
                treq = h->tracking[h->recv.read[1]];
                if (!treq) {
                        WARN("%s - Ignoring reply with ID %i that arrived too late",
                             h->module_name, h->recv.data[1]);
 
                        h->recv.read += packet_len;
                        continue;
                }
 
                treq = talloc_get_type_abort(treq, trunk_request_t);
                request = treq->request;
                fr_assert(request != NULL);
                u = talloc_get_type_abort(treq->preq, udp_request_t);
                r = talloc_get_type_abort(treq->rctx, udp_result_t);
 
                fr_pair_list_init(&reply);
 
                /*
                 *      Validate and decode the incoming packet
                 */
                slen = decode(request->reply_ctx, &reply, &code, h, request, u, h->recv.read, packet_len);
                if (slen < 0) {
                        // @todo - give real decode error?
                        trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
                        return;
                }
                h->recv.read += packet_len;
 
                /*
                 *      Only valid packets are processed.
                 */
                h->last_reply = fr_time();
 
                treq->request->reply->code = code;
 
                // @todo - check various random locations for status of the reply: error, etc.
                r->rcode = RLM_MODULE_OK;
//              r->rcode = radius_code_to_rcode[code];
                fr_pair_list_append(&request->reply_pairs, &reply);
                trunk_request_signal_complete(treq);
        }
}
 
/** Remove the request from any tracking structures
 *
 * Frees encoded packets if the request is being moved to a new connection
 */
static void request_cancel(connection_t *conn, void *preq_to_reset,
                           trunk_cancel_reason_t reason, UNUSED void *uctx)
{
        udp_request_t   *u = talloc_get_type_abort(preq_to_reset, udp_request_t);
 
        /*
         *      Request has been requeued on the same
         *      connection due to timeout or DUP signal.
         */
        if (reason == TRUNK_CANCEL_REASON_REQUEUE) {
                udp_handle_t            *h = talloc_get_type_abort(conn->h, udp_handle_t);
 
                udp_request_reset(h, u);
        }
 
        /*
         *      Other cancellations are dealt with by
         *      request_conn_release as the request is removed
         *      from the trunk.
         */
}
 
/** Clear out anything associated with the handle from the request
 *
 */
static void request_conn_release(connection_t *conn, void *preq_to_reset, UNUSED void *uctx)
{
        udp_request_t           *u = talloc_get_type_abort(preq_to_reset, udp_request_t);
        udp_handle_t            *h = talloc_get_type_abort(conn->h, udp_handle_t);
 
        if (u->packet) udp_request_reset(h, u);
 
        /*
         *      If there are no outstanding tracking entries
         *      allocated then the connection is "idle".
         *
         *      @todo - enable idle timeout?
         */
        if (!h->active) h->last_idle = fr_time();
}
 
/** Write out a canned failure
 *
 */
static void request_fail(request_t *request, NDEBUG_UNUSED void *preq, void *rctx,
                         NDEBUG_UNUSED trunk_request_state_t state, UNUSED void *uctx)
{
        udp_result_t            *r = talloc_get_type_abort(rctx, udp_result_t);
#ifndef NDEBUG
        udp_request_t           *u = talloc_get_type_abort(preq, udp_request_t);
#endif
 
        fr_assert(!u->ev);      /* Dealt with by request_conn_release */
 
        fr_assert(state != TRUNK_REQUEST_STATE_INIT);
 
        r->rcode = RLM_MODULE_FAIL;
        r->treq = NULL;
 
        unlang_interpret_mark_runnable(request);
}
 
/** Response has already been written to the rctx at this point
 *
 */
static void request_complete(request_t *request, NDEBUG_UNUSED void *preq, void *rctx, UNUSED void *uctx)
{
        udp_result_t            *r = talloc_get_type_abort(rctx, udp_result_t);
#ifndef NDEBUG
        udp_request_t           *u = talloc_get_type_abort(preq, udp_request_t);
#endif
 
        fr_assert(!u->packet && !u->ev);        /* Dealt with by request_conn_release */
 
        r->treq = NULL;
 
        unlang_interpret_mark_runnable(request);
}
 
/** Explicitly free resources associated with the protocol request
 *
 */
static void request_free(UNUSED request_t *request, void *preq_to_free, UNUSED void *uctx)
{
        udp_request_t           *u = talloc_get_type_abort(preq_to_free, udp_request_t);
 
        fr_assert(!u->packet && !u->ev);        /* Dealt with by request_conn_release */
 
        talloc_free(u);
}
 
/** Resume execution of the request, returning the rcode set during trunk execution
 *
 */
static unlang_action_t mod_resume(rlm_rcode_t *p_result, module_ctx_t const *mctx, UNUSED request_t *request)
{
        udp_result_t    *r = talloc_get_type_abort(mctx->rctx, udp_result_t);
        rlm_rcode_t     rcode = r->rcode;
 
        talloc_free(r);
 
        RETURN_MODULE_RCODE(rcode);
}
 
static void mod_signal(module_ctx_t const *mctx, UNUSED request_t *request, fr_signal_t action)
{
//      udp_thread_t            *t = talloc_get_type_abort(mctx->thread, udp_thread_t);
        udp_result_t            *r = talloc_get_type_abort(mctx->rctx, udp_result_t);
 
        /*
         *      If we don't have a treq associated with the
         *      rctx it's likely because the request was
         *      scheduled, but hasn't yet been resumed, and
         *      has received a signal, OR has been resumed
         *      and immediately cancelled as the event loop
         *      is exiting, in which case
         *      unlang_request_is_scheduled will return false
         *      (don't use it).
         */
        if (!r->treq) {
                talloc_free(r);
                return;
        }
 
        switch (action) {
        /*
         *      The request is being cancelled, tell the
         *      trunk so it can clean up the treq.
         */
        case FR_SIGNAL_CANCEL:
                trunk_request_signal_cancel(r->treq);
                r->treq = NULL;
                talloc_free(r);         /* Should be freed soon anyway, but better to be explicit */
                return;
 
        /*
         *      Requeue the request on the same connection
         *      causing a "retransmission" if the request
         *      has already been sent out.
         */
        case FR_SIGNAL_DUP:
                /*
                 *      Retransmit the current request on the same connection.
                 *
                 *      If it's zombie, we still resend it.  If the
                 *      connection is dead, then a callback will move
                 *      this request to a new connection.
                 */
                trunk_request_requeue(r->treq);
                return;
 
        default:
                return;
        }
}
 
#ifndef NDEBUG
/** Free a udp_result_t
 *
 * Allows us to set break points for debugging.
 */
static int _udp_result_free(udp_result_t *r)
{
        trunk_request_t *treq;
        udp_request_t           *u;
 
        if (!r->treq) return 0;
 
        treq = talloc_get_type_abort(r->treq, trunk_request_t);
        u = talloc_get_type_abort(treq->preq, udp_request_t);
 
        fr_assert_msg(!u->ev, "udp_result_t freed with active timer");
 
        return 0;
}
#endif
 
static unlang_action_t mod_enqueue(rlm_rcode_t *p_result, void **rctx_out, UNUSED void *instance, void *thread, request_t *request)
{
        udp_thread_t                    *t = talloc_get_type_abort(thread, udp_thread_t);
        udp_result_t                    *r;
        udp_request_t                   *u;
        trunk_request_t         *treq;
        trunk_enqueue_t         q;
 
        fr_assert(FR_TACACS_PACKET_CODE_VALID(request->packet->code));
 
        treq = trunk_request_alloc(t->trunk, request);
        if (!treq) RETURN_MODULE_FAIL;
 
        MEM(r = talloc_zero(request, udp_result_t));
#ifndef NDEBUG
        talloc_set_destructor(r, _udp_result_free);
#endif
 
        /*
         *      Can't use compound literal - const issues.
         */
        MEM(u = talloc_zero(treq, udp_request_t));
        u->code = request->packet->code;
        u->priority = request->async->priority;
        u->recv_time = request->async->recv_time;
 
        r->rcode = RLM_MODULE_FAIL;
 
        q = trunk_request_enqueue(&treq, t->trunk, request, u, r);
        if (q < 0) {
                fr_assert(!u->packet);  /* Should not have been fed to the muxer */
                trunk_request_free(&treq);              /* Return to the free list */
        fail:
                talloc_free(r);
                RETURN_MODULE_FAIL;
        }
 
        /*
         *      All destinations are down.
         */
        if (q == TRUNK_ENQUEUE_IN_BACKLOG) {
                RDEBUG("All destinations are down - cannot send packet");
                goto fail;
        }
 
        r->treq = treq; /* Remember for signalling purposes */
 
        *rctx_out = r;
 
        return UNLANG_ACTION_YIELD;
}
 
/** Instantiate thread data for the submodule.
 *
 */
static int mod_thread_instantiate(module_thread_inst_ctx_t const *mctx)
{
        rlm_tacacs_tcp_t                *inst = talloc_get_type_abort(mctx->mi->data, rlm_tacacs_tcp_t);
        udp_thread_t                    *thread = talloc_get_type_abort(mctx->thread, udp_thread_t);
 
        static trunk_io_funcs_t io_funcs = {
                                                .connection_alloc = thread_conn_alloc,
                                                .connection_notify = thread_conn_notify,
                                                .request_prioritise = request_prioritise,
                                                .request_mux = request_mux,
                                                .request_demux = request_demux,
                                                .request_conn_release = request_conn_release,
                                                .request_complete = request_complete,
                                                .request_fail = request_fail,
                                                .request_cancel = request_cancel,
                                                .request_free = request_free
                                        };
 
        inst->trunk_conf = &inst->parent->trunk_conf;
 
        inst->trunk_conf->req_pool_headers = 2; /* One for the request, one for the buffer */
        inst->trunk_conf->req_pool_size = sizeof(udp_request_t) + inst->max_packet_size;
 
        thread->el = mctx->el;
        thread->inst = inst;
        thread->trunk = trunk_alloc(thread, mctx->el, &io_funcs,
                                       inst->trunk_conf, inst->parent->name, thread, false);
        if (!thread->trunk) return -1;
 
        /*
         *      Empty secrets don't exist
         */
        if (inst->secret && !*inst->secret) {
                talloc_const_free(inst->secret);
                inst->secret = NULL;
        }
 
        if (inst->secret) inst->secretlen = talloc_array_length(inst->secret) - 1;
 
        return 0;
}
 
static int mod_instantiate(module_inst_ctx_t const *mctx)
{
        rlm_tacacs_t            *parent = talloc_get_type_abort(mctx->mi->parent->data, rlm_tacacs_t);
        rlm_tacacs_tcp_t        *inst = talloc_get_type_abort(mctx->mi->data, rlm_tacacs_tcp_t);
        CONF_SECTION            *conf = mctx->mi->conf;
 
        if (!parent) {
                ERROR("IO module cannot be instantiated directly");
                return -1;
        }
 
        inst->parent = parent;
 
        /*
         *      Always need at least one mmsgvec
         */
        if (inst->max_send_coalesce == 0) inst->max_send_coalesce = 1;
 
        /*
         *      Ensure that we have a destination address.
         */
        if (inst->dst_ipaddr.af == AF_UNSPEC) {
                cf_log_err(conf, "A value must be given for 'ipaddr'");
                return -1;
        }
 
        /*
         *      If src_ipaddr isn't set, make sure it's INADDR_ANY, of
         *      the same address family as dst_ipaddr.
         */
        if (inst->src_ipaddr.af == AF_UNSPEC) {
                memset(&inst->src_ipaddr, 0, sizeof(inst->src_ipaddr));
 
                inst->src_ipaddr.af = inst->dst_ipaddr.af;
 
                if (inst->src_ipaddr.af == AF_INET) {
                        inst->src_ipaddr.prefix = 32;
                } else {
                        inst->src_ipaddr.prefix = 128;
                }
        }
 
        else if (inst->src_ipaddr.af != inst->dst_ipaddr.af) {
                cf_log_err(conf, "The 'ipaddr' and 'src_ipaddr' configuration items must "
                           "be both of the same address family");
                return -1;
        }
 
        if (!inst->dst_port) {
                cf_log_err(conf, "A value must be given for 'port'");
                return -1;
        }
 
        /*
         *      Clamp max_packet_size first before checking recv_buff and send_buff
         */
        FR_INTEGER_BOUND_CHECK("max_packet_size", inst->max_packet_size, >=, ((255 + (int) sizeof(fr_tacacs_packet_t)) & 0xffffff00));
        FR_INTEGER_BOUND_CHECK("max_packet_size", inst->max_packet_size, <=, 65535);
 
 
        if (inst->recv_buff_is_set) {
                FR_INTEGER_BOUND_CHECK("recv_buff", inst->recv_buff, >=, inst->max_packet_size);
                FR_INTEGER_BOUND_CHECK("recv_buff", inst->recv_buff, <=, (1 << 30));
        }
 
        if (inst->send_buff_is_set) {
                FR_INTEGER_BOUND_CHECK("send_buff", inst->send_buff, >=, inst->max_packet_size);
                FR_INTEGER_BOUND_CHECK("send_buff", inst->send_buff, <=, (1 << 30));
        }
 
 
        return 0;
}
 
extern rlm_tacacs_io_t rlm_tacacs_tcp;
rlm_tacacs_io_t rlm_tacacs_tcp = {
        .common = {
                .magic                  = MODULE_MAGIC_INIT,
                .name                   = "tacacs_tcp",
                .inst_size              = sizeof(rlm_tacacs_tcp_t),
 
                .thread_inst_size       = sizeof(udp_thread_t),
                .thread_inst_type       = "udp_thread_t",
 
                .config                 = module_config,
                .instantiate            = mod_instantiate,
                .thread_instantiate     = mod_thread_instantiate,
        },
        .enqueue                = mod_enqueue,
        .signal                 = mod_signal,
        .resume                 = mod_resume,
};