io.c

Go to the documentation of this file.

/*
 *   This program 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: 7c6edcd36d2ebd58261d56de325a0e3d462ffb48 $
 * @file lib/redis/io.c
 * @brief Common functions for interacting with Redis via hiredis
 *
 * @copyright 2019 The FreeRADIUS server project
 * @copyright 2019 Network RADIUS SAS (legal@networkradius.com)
 *
 * @author Arran Cudbard-Bell (a.cudbardb@freeradius.org)
 */
 
#include <freeradius-devel/redis/io.h>
#include <freeradius-devel/util/debug.h>
 
#ifdef HAVE_REDIS_SSL
#include <hiredis/hiredis_ssl.h>
#endif
 
typedef struct {
        fr_redis_io_conf_t const        *io_conf;
#ifdef HAVE_REDIS_SSL
        SSL_CTX                         *ssl_ctx;
#endif
} redis_conn_uctx_t;
 
/** Called by hiredis to indicate the connection is dead
 *
 */
static void _redis_disconnected(redisAsyncContext const *ac, UNUSED int status)
{
        connection_t            *conn = talloc_get_type_abort(ac->data, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        /*
         *      Some code paths result in hiredis calling this callback after
         *      we have freed the handle, which would result in a use after free.
         *      or we do not actually want to reconnnect.
         */
        switch (conn->state) {
        case CONNECTION_STATE_CLOSED:
        case CONNECTION_STATE_SHUTDOWN:
                return;
 
        default:
                break;
        }
 
        /*
         *      redisAsyncFree was called with a live
         *      connection, but inside the talloc
         *      destructor of the fr_redis_handle_t.
         *
         *      Don't signal the connection state
         *      machine that it needs reconnecting,
         *      the connection is being destroyed.
         */
        if (h->freeing) return;
 
        DEBUG4("Signalled by hiredis, connection disconnected");
 
        connection_signal_reconnect(conn, CONNECTION_FAILED);
}
 
/** Callback for verifying the results of SELECT
 *
 */
static void _redis_select_result(struct redisAsyncContext *ac, void *data, UNUSED void *privdata)
{
        connection_t    *conn = talloc_get_type_abort(ac->data, connection_t);
        redisReply      *reply = data;
 
        if (!reply) {
                ERROR("Failed selecting database: %s", ac->errstr);
        error:
                connection_signal_reconnect(conn, CONNECTION_FAILED);
                return;
        }
 
        switch (reply->type) {
        case REDIS_REPLY_STATUS:
                if (strcmp(reply->str, "OK") != 0) {
                        ERROR("Failed selecting database: %s", reply->str);
                        goto error;
                }
                break;  /* else it's OK */
 
        case REDIS_REPLY_ERROR:
                ERROR("Failed selecting database: %s", reply->str);
                goto error;
 
        default:
                ERROR("Unexpected reply of type %s to SELECT",
                      fr_table_str_by_value(redis_reply_types, reply->type, "<UNKNOWN>"));
                goto error;
        }
 
        connection_signal_connected(conn);
}
 
/** Callback for verifying the results of AUTH
 *
 */
static void _redis_auth_result(struct redisAsyncContext *ac, void *data, UNUSED void *privdata)
{
        connection_t            *conn = talloc_get_type_abort(ac->data, connection_t);
        redisReply              *reply = data;
        redis_conn_uctx_t       *conn_uctx = connection_uctx_get(conn);
        uint32_t                database = conn_uctx->io_conf->database;
 
        if (!reply) {
                ERROR("Failed authenticating: %s", ac->errstr);
        error:
                connection_signal_reconnect(conn, CONNECTION_FAILED);
                return;
        }
 
        switch (reply->type) {
        case REDIS_REPLY_STATUS:
                if (strcmp(reply->str, "OK") != 0) {
                        ERROR("Failed authenticating: %s", reply->str);
                        goto error;
                }
                break;  /* else it's OK */
 
        case REDIS_REPLY_ERROR:
                ERROR("Failed authenticating: %s", reply->str);
                goto error;
 
        default:
                ERROR("Unexpected reply of type %s to AUTH",
                      fr_table_str_by_value(redis_reply_types, reply->type, "<UNKNOWN>"));
                goto error;
        }
 
        if (database) {
                DEBUG3("Executing: SELECT %d", database);
                if (redisAsyncCommand(ac, _redis_select_result, NULL, "SELECT %d", database) != REDIS_OK) goto error;
                return;
        }
 
        connection_signal_connected(conn);
}
 
/** Called by hiredis to indicate the connection is live
 *
 */
static void _redis_connected_nc(redisAsyncContext *ac, UNUSED int status)
{
        connection_t                    *conn = talloc_get_type_abort(ac->data, connection_t);
        redis_conn_uctx_t               *conn_uctx = connection_uctx_get(conn);
        fr_redis_io_conf_t const        *io_conf = conn_uctx->io_conf;
 
        DEBUG4("Signalled by hiredis, connection is open");
 
#ifdef HAVE_REDIS_SSL
        if (io_conf->use_tls) {
                fr_tls_session_t *tls_session = fr_tls_session_alloc_client(conn, conn_uctx->ssl_ctx);
                DEBUG2("%s - Using tls", io_conf->log_prefix);
                if (!tls_session) {
                        fr_tls_strerror_printf("%s - [%s]", io_conf->log_prefix, conn->name);
                        ERROR("%s - Failed to allocate TLS session", io_conf->log_prefix);
                        connection_signal_reconnect(conn, CONNECTION_FAILED);
                        return;
                }
 
                // redisInitiateSSL() takes ownership of SSL object on success
                SSL_up_ref(tls_session->ssl);
                if (redisInitiateSSL(&ac->c, tls_session->ssl) != REDIS_OK) {
                        ERROR("%s - Failed to initiate SSL: %s", io_conf->log_prefix, ac->c.errstr);
                        SSL_free(tls_session->ssl);
                        connection_signal_reconnect(conn, CONNECTION_FAILED);
                        return;
                }
        }
#endif
 
        /*
         *      If auth is configured, send the appropriate AUTH command
         *      before marking the connection as connected.
         */
        if (io_conf->password) {
                if (io_conf->username) {
                        DEBUG3("Executing: AUTH %s %s", io_conf->username, io_conf->password);
                        if (redisAsyncCommand(ac, _redis_auth_result, NULL, "AUTH %s %s",
                                              io_conf->username, io_conf->password) != REDIS_OK) {
                        error:
                                ERROR("Failed executing command");
                                connection_signal_reconnect(conn, CONNECTION_FAILED);
                        }
                        return;
                }
                DEBUG3("Executing: AUTH %s", io_conf->password);
                if (redisAsyncCommand(ac, _redis_auth_result, NULL, "AUTH %s", io_conf->password) != REDIS_OK) goto error;
                return;
        }
 
        if (io_conf->database) {
                DEBUG3("Executing: SELECT %d", io_conf->database);
                if (redisAsyncCommand(ac, _redis_select_result, NULL, "SELECT %d", io_conf->database) != REDIS_OK) goto error;
                return;
        }
 
        connection_signal_connected(conn);
}
 
#ifndef HAVE_REDIS_CALLBACKNC
/*
 *      Hiredis prior to 1.1.0 only has the callback with const context
 *      We need it un-const so we can update the connection status.
 */
static void _redis_connected(redisAsyncContext const *ac, int status)
{
        redisAsyncContext       *our_ac = UNCONST(redisAsyncContext *, ac);
 
        _redis_connected_nc(our_ac, status);
}
#endif
 
/** Redis FD became readable
 *
 */
static void _redis_io_service_readable(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
{
        connection_t const      *conn = talloc_get_type_abort_const(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - FD %i now readable", h, fd);
 
        redisAsyncHandleRead(h->ac);
}
 
/** Redis FD became writable
 *
 */
static void _redis_io_service_writable(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
{
        connection_t const      *conn = talloc_get_type_abort_const(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - FD %i now writable", h, fd);
 
        redisAsyncHandleWrite(h->ac);
}
 
static void _redis_io_common(connection_t *conn, fr_redis_handle_t *h, bool read, bool write);
 
/** Redis FD errored - Automatically removes registered events
 *
 */
static void _redis_io_service_errored(UNUSED fr_event_list_t *el, int fd, UNUSED int flags,
                                      int fd_errno, void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        ERROR("%s handle %p - FD %i errored: %s", conn->name, h, fd, fr_syserror(fd_errno));
 
        /*
         *      Ensure fd events are removed - hiredis doesn't clean up quickly.
         */
        _redis_io_common(conn, h, false, false);
 
        /*
         *      Connection state machine will handle reconnecting
         */
        connection_signal_reconnect(conn, CONNECTION_FAILED);
}
 
/** Deal with the method hiredis uses to register/unregister interest in a file descriptor
 *
 */
static void _redis_io_common(connection_t *conn, fr_redis_handle_t *h, bool read, bool write)
{
        redisContext            *c = &(h->ac->c);
        fr_event_list_t         *el = conn->el;
 
        /*
         *      hiredis doesn't even attempt to dedup registration
         *      requests *sigh*...
         */
        if ((h->read_set == read) && (h->write_set == write)) return;
 
        if (!read && !write) {
                DEBUG4("redis handle %p - De-registering FD %i", h, c->fd);
 
                if (fr_event_fd_delete_handle(h->fd_ev)) {
                        PERROR("redis handle %p - De-registration failed for FD %i", h, c->fd);
                }
                h->fd_ev = NULL;
                return;
        }
 
        DEBUG4("redis handle %p - Registered for %s%serror events on FD %i",
               h, read ? "read+" : "", write ? "write+" : "", c->fd);
 
        if (fr_event_fd_insert(h, &h->fd_ev, el, c->fd,
                               read ? _redis_io_service_readable : NULL,
                               write ? _redis_io_service_writable : NULL,
                               _redis_io_service_errored,
                               conn) < 0) {
                PERROR("redis handle %p - Registration failed for %s%serror events on FD %i",
                       h, read ? "read+" : "", write ? "write+" : "", c->fd);
                return;
        }
 
        h->read_set = read;
        h->write_set = write;
}
 
/** Register FD for reads
 *
 */
static void _redis_io_add_read(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, true, h->write_set);
}
 
/** De-register FD for reads
 *
 */
static void _redis_io_del_read(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, false, h->write_set);
}
 
/** Register FD for writes
 *
 */
static void _redis_io_add_write(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, h->read_set, true);
}
 
/** De-register FD for writes
 *
 */
static void _redis_io_del_write(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, h->read_set, false);
}
 
#ifdef HAVE_REDIS_TIMEOUT
/** Connection timer expired
 *
 */
static void _redis_io_service_timer_expired(UNUSED fr_timer_list_t *tl, UNUSED fr_time_t now, void *uctx)
{
        connection_t const      *conn = talloc_get_type_abort_const(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - Timeout", h);
 
        redisAsyncHandleTimeout(h->ac);
}
 
/** Modify the connection I/O timer
 *
 */
static void _redis_io_timer_modify(void *uctx, struct timeval tv)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
        fr_time_delta_t         timeout;
 
        timeout = fr_time_delta_from_timeval(&tv);
 
        DEBUG4("redis handle %p - Timeout in %pV seconds", h, fr_box_time_delta(timeout));
 
        if (fr_timer_in(h, conn->el->tl, &h->timer_ev, timeout,
                        false, _redis_io_service_timer_expired, conn) < 0) {
                PERROR("redis timeout %p - Failed adding timeout", h);
        }
}
#endif
 
/** Handle freeing the redisAsyncContext
 *
 * delRead and delWrite don't seem to be called when the redisAsyncContext is freed
 *
 * As the IO events must be removed from the event loop *before* the FD is closed
 * and as the IO events will only be automatically de-registered when when the
 * fr_redis_handle_t is freed.
 *
 * Unfortunately the destructor for the fr_redis_handle_t will be run before
 * the IO events are de-registered, which'll free the redisAsycCtx, which'll close
 * the FD.
 *
 * This means there'd be a brief period of time between the FD is closed, and
 * it being removed from the event loop.
 *
 * We use the cleanup callback (which is called before the FD is closed) to remove
 * the events now, and ensure there's no chance of issues.
 */
static void _redis_io_free(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        /*
         *      Some code paths result in hiredis calling this callback after
         *      we have freed the handle, which would result in a use after free.
         */
        if (conn->state == CONNECTION_STATE_CLOSED) return;
 
        DEBUG4("redis handle %p - Freed", h);
 
        _redis_io_common(conn, h, false, false);
        h->ac = NULL;
}
 
/** Configures async I/O callbacks for an existing redisAsyncContext
 *
 */
static int fr_redis_io_setup(redisAsyncContext *ac, connection_t const *conn)
{
        if (ac->ev.data != NULL) return REDIS_ERR;
 
        ac->ev.addRead = _redis_io_add_read;
        ac->ev.delRead = _redis_io_del_read;
        ac->ev.addWrite = _redis_io_add_write;
        ac->ev.delWrite = _redis_io_del_write;
#ifdef HAVE_REDIS_TIMEOUT
        ac->ev.scheduleTimer = _redis_io_timer_modify;
#endif
        ac->ev.cleanup = _redis_io_free;
        memcpy(&ac->ev.data, &conn, sizeof(ac->ev.data));
 
        return REDIS_OK;
}
 
/** Free the redis async context when the handle is freed
 *
 */
static int _redis_handle_free(fr_redis_handle_t *h)
{
        /*
         *      Don't fire the reconnect callback if we're
         *      freeing the handle.
         */
        h->freeing = true;
        if (h->ac) redisAsyncFree(h->ac);
 
        return 0;
}
 
/** Callback for the initialise state
 *
 * Should attempt to open a non-blocking connection and return it in h_out.
 *
 * @param[out] h_out    Where to write the new handle
 * @param[in] conn      This connection.  Opaque, should only be used for
 *                      signalling the connection state machine.
 * @param[in] uctx      User context.
 * @return
 *      - #CONNECTION_STATE_CONNECTING  if a file descriptor was successfully created.
 *      - #CONNECTION_STATE_FAILED              if we could not open a valid handle.
 */
CC_NO_UBSAN(function) /* UBSAN: false positive - public vs private connection_t trips --fsanitize=function*/
static connection_state_t _redis_io_connection_init(void **h_out, connection_t *conn, void *uctx)
{
        redis_conn_uctx_t       *conn_uctx = uctx;
        char const              *host = conn_uctx->io_conf->hostname;
        uint16_t                port = conn_uctx->io_conf->port;
        fr_redis_handle_t       *h;
        int                     ret;
 
        /*
         *      Allocate a structure to wrap the
         *      redis async context.
         */
        MEM(h = talloc_zero(conn, fr_redis_handle_t));
        talloc_set_destructor(h, _redis_handle_free);
 
        h->ac = redisAsyncConnect(host, port);
        if (!h->ac) {
                ERROR("Failed allocating handle for %s:%u", host, port);
                return CONNECTION_STATE_FAILED;
        }
 
        if (h->ac->err) {
                ERROR("Failed allocating handle for %s:%u: %s", host, port, h->ac->errstr);
        error:
                redisAsyncFree(h->ac);
                h->ac = NULL;
                return CONNECTION_STATE_FAILED;
        }
 
        /*
         *      Store the connection in private data,
         *      so we can use it for signalling.
         *
         *      Comments in the redis src indicate
         *      it doesn't mess with this.
         */
        memcpy(&h->ac->data, &conn, sizeof(h->ac->data));
 
        /*
         *      Handle has to be associated with the
         *      conn in case I/O handlers want to get
         *      at it.
         */
        *h_out = h;
 
        /*
         *      Install the I/O service functions
         *
         *      Event library must be set first
         *      before calling redisAsyncSetConnectCallback
         *      as just setting the callback triggers
         *      I/O events to be registered.
         */
        fr_redis_io_setup(h->ac, conn);
 
        /*
         *      Setup callbacks so we're notified
         *      when the connection state changes.
         *
         *      We then signal the connection state
         *      machine, to let it handle
         *      reconnecting.
         */
#ifdef HAVE_REDIS_CALLBACKNC
        ret = redisAsyncSetConnectCallbackNC(h->ac, _redis_connected_nc);
#else
        ret = redisAsyncSetConnectCallback(h->ac, _redis_connected);
#endif
        if (ret != REDIS_OK) {
                ERROR("Failed setting connected callback: Error %i", ret);
                goto error;
        }
 
        ret = redisAsyncSetDisconnectCallback(h->ac, _redis_disconnected);
        if (ret != REDIS_OK) {
                ERROR("Failed setting disconnected callback: Error %i", ret);
                goto error;
        }
 
        fr_dlist_talloc_init(&h->ignore, fr_redis_sqn_ignore_t, entry);
 
        return CONNECTION_STATE_CONNECTING;
}
 
/** Gracefully signal that the connection should shutdown
 *
 */
static connection_state_t _redis_io_connection_shutdown(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx)
{
        fr_redis_handle_t       *our_h = talloc_get_type_abort(h, fr_redis_handle_t);
 
        redisAsyncDisconnect(our_h->ac);        /* Should not free the handle */
 
        return CONNECTION_STATE_SHUTDOWN;
}
 
/** Notification that the connection has errored and must be closed
 *
 * This should be used to close the file descriptor.  It is assumed
 * that the file descriptor is invalid after this callback has been executed.
 *
 * If this callback does not close the file descriptor, the server will leak
 * file descriptors.
 *
 * @param[in] el        to remove event handlers from.
 * @param[in] h         to close.
 * @param[in] uctx      User context.
 */
static void _redis_io_connection_close(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx)
{
        fr_redis_handle_t       *our_h = talloc_get_type_abort(h, fr_redis_handle_t);
 
        /*
         *      The destructor will free the redisAsyncContext
         *      which'll close the connection, after removing
         *      it from the event loop.
         */
        talloc_free(our_h);
}
 
/** Allocate an async redis I/O connection
 *
 */
connection_t *fr_redis_connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, connection_conf_t const *conn_conf,
                                        fr_redis_io_conf_t const *io_conf,
#ifdef HAVE_REDIS_SSL
                                        SSL_CTX *ssl_ctx,
#endif
                                        char const *log_prefix)
{
        connection_t            *conn;
        redis_conn_uctx_t       *uctx;
 
        MEM(uctx = talloc(ctx, redis_conn_uctx_t));
        uctx->io_conf = io_conf;
#ifdef HAVE_REDIS_SSL
        uctx->ssl_ctx = ssl_ctx;
#endif
 
        /*
         *      We don't specify an open callback
         *      as hiredis handles switching over
         *      all the I/O handlers internally
         *      within hireds, and calls us when
         *      the connection is open.
         */
        conn = connection_alloc(ctx, el,
                                   &(connection_funcs_t){
                                        .init = _redis_io_connection_init,
                                        .close = _redis_io_connection_close,
                                        .shutdown = _redis_io_connection_shutdown
                                   },
                                   conn_conf,
                                   log_prefix,
                                   uctx);
        if (!conn) return NULL;
 
        return conn;
}
 
/** Return the redisAsyncContext associated with the connection
 *
 * This is needed to issue commands to the redis server.
 *
 * @param[in] conn      To retrieve async ctx from.
 * @return The async ctx.
 */
redisAsyncContext *fr_redis_connection_get_async_ctx(connection_t *conn)
{
        fr_redis_handle_t       *h = conn->h;
        return h->ac;
}