NNTPGrab

/* 

    Copyright (C) 2005-2010  Erik van Pienbroek

    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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

*/

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include 

#include 

#include 

#include 

#include 

#include "nntpgrab_plugin.h"

#include "nntpgrab_utils.h"

#include "mongoose.h"

#ifdef WIN32

#include 

#include 

#include  /* freeaddrinfo for Win2k */

#else

#include 

#include 

#include 

#include 

#include 

#endif /* WIN32 */

/* Yes this is ugly, but it's required to avoid adding any changes to the original 

 * mongoose.c (we may need to update it from upstream in the future) */

#include "mongoose.c"

/* 

 * Setup listening socket on given port, return socket.

 */

static SOCKET

open_listening_port(struct mg_context *ctx, struct addrinfo *res, char *port, char **errmsg)

{

    SOCKET sock;

    int    on = 1;

    if ((sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol)) != INVALID_SOCKET &&

        setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) == 0 &&

        bind(sock, res->ai_addr, (int) res->ai_addrlen) == 0 &&

        listen(sock, 5) == 0) {

        /* Success */

        set_close_on_exec(sock);

    } else {

        /* Error */

        *errmsg = g_strdup_printf("%s(%s): %s", __func__, port, strerror(ERRNO));

        if (sock != INVALID_SOCKET)

            (void) closesocket(sock);

        sock = INVALID_SOCKET;

    }

    return (sock);

}

ngboolean

listen_on_port(struct mg_context *ctx, char *port)

{

    SOCKET        sock;

    int           is_ssl;

    struct socket *listener;

    struct addrinfo hints, *res, *ressave;

    int n;

    char *errmsg = NULL;

    close_all_listening_sockets(ctx);

    assert(ctx->num_listeners == 0);

    memset(&hints, 0, sizeof(struct addrinfo));

    hints.ai_flags    = AI_PASSIVE;

    hints.ai_family   = AF_UNSPEC;

    hints.ai_socktype = SOCK_STREAM;

    n = getaddrinfo(NULL, port, &hints, &res);

    if (n < 0) {

        fprintf(stderr,

                "getaddrinfo error:: [%s]\n",

#ifdef WIN32

                WSAGetLastError());

#else

                gai_strerror(n));

#endif

        return FALSE;

    }

    ressave=res;

    /* 

       Try open socket with each address getaddrinfo returned,

       until getting a valid listening socket.

    */

    while (res) {

        char *errmsg_tmp = NULL;

        /* is_ssl = vec.ptr[vec.len - 1] == 's' ? TRUE : FALSE; */

        is_ssl = FALSE;

        listener = ctx->listeners + ctx->num_listeners;

        if (ctx->num_listeners >= (int) (ARRAY_SIZE(ctx->listeners) - 1)) {

            cry(fc(ctx), "%s", "Too many listening sockets");

            return (FALSE);

        } else if ((sock = open_listening_port(ctx, res, port, &errmsg_tmp)) == INVALID_SOCKET) {

            if (errmsg) {

                g_free(errmsg);

            }

            errmsg = errmsg_tmp;

            res = res->ai_next;

            continue;

        } else if (is_ssl == TRUE && ctx->ssl_ctx == NULL) {

            (void) closesocket(sock);

            cry(fc(ctx), "cannot add SSL socket, please specify "

                "-ssl_cert option BEFORE -ports option");

            return (FALSE);

        } else {

            listener->sock = sock;

            listener->is_ssl = is_ssl;

            ctx->num_listeners++;

            /* Mongoose assumes that the listener->lsa is filled with information */

            memset(&listener->lsa, 0, sizeof(listener->lsa));

            listener->lsa.ai_family              = res->ai_family;

            listener->lsa.u.sin.sin_addr.s_addr  = htonl(INADDR_ANY);

            listener->lsa.len                    = (res->ai_family == AF_INET6 ? sizeof(listener->lsa.u.sin6) : sizeof(listener->lsa.u.sin));

            listener->lsa.u.sin.sin_family       = res->ai_family;

            listener->lsa.u.sin.sin_port         = htons((uint16_t) atoi(port));

        }

        res = res->ai_next;

    }

    freeaddrinfo(ressave);

    if (ctx->num_listeners == 0) {

        cry(fc(ctx), "%s", errmsg);

    }

    g_free(errmsg);

    return (TRUE);

}

static GList *active_connections = NULL;

static GStaticMutex connections_mutex = G_STATIC_MUTEX_INIT;

static GStaticMutex write_mutex = G_STATIC_MUTEX_INIT;

static NGPlugin *plugin_data_global = NULL;

/* The declaration for this function comes from jsonrpc.h but 

 * because of header conflicts on Win32 we need to put it here */

extern char* jsonrpc_process(char* request, void *user_data);

void

mongoose_hacks_set_plugin_data(NGPlugin *plugin_data)

{

    plugin_data_global = plugin_data;

}

static void

emit_num_active_connections_update(int num)

{

    const char *params[2];

    char tmp[16];

    memset(&tmp, 0, sizeof(tmp));

    snprintf(tmp, sizeof(tmp), "%i", num);

    params[0] = tmp;

    params[1] = NULL;

    g_return_if_fail(plugin_data_global != NULL);

    ng_plugin_emit_event(plugin_data_global, "num_active_connections_changed", params);

}

static void

add_conn_to_list(struct mg_connection *conn)

{

    int num;

    g_static_mutex_lock(&connections_mutex);

    active_connections = g_list_append(active_connections, conn);

    num = g_list_length(active_connections);

    g_static_mutex_unlock(&connections_mutex);

    emit_num_active_connections_update(num);

}

static void

del_conn_from_list(struct mg_connection *conn)

{

    int num;

    g_static_mutex_lock(&connections_mutex);

    active_connections = g_list_remove(active_connections, conn);

    num = g_list_length(active_connections);

    g_static_mutex_unlock(&connections_mutex);

    emit_num_active_connections_update(num);

}

void

jsonrpc_tcp_force_disconnect(void)

{

    GList *list;

    g_static_mutex_lock(&connections_mutex);

    list = active_connections;

    while (list) {

        struct mg_connection *conn = list->data;

        close_socket_gracefully(conn, conn->client.sock);

        list = g_list_next(list);

    }

    g_static_mutex_unlock(&connections_mutex);

}

/* 

 * Check whether full request is buffered. Return:

 *    0         if request is not yet fully buffered

 *   >0         actual request length, without last \r\n

 */

static int

get_request_len_line(const char *buf, size_t buflen)

{

    int i;

    for (i = 0; i < buflen; i++) {

        if ((buf[i] == '\r' && buf[i + 1] == '\n') ||

             buf[i] == '\n') {

            if (i == 0) {

                /* Ignore empty lines */

                if (buf[i] == '\r') {

                    memmove((void*) buf, (void*) buf + 2, buflen - 2);

                    buflen -= 2;

                } else {

                    memmove((void*) buf, (void*) buf + 1, buflen - 1);

                    buflen -= 1;

                }

                /* Start over again */

                i--;

                continue;

            }

            return i;

        }

    }

    return 0;

}

/* 

 * Keep reading the input (either opened file descriptor fd, or socket sock,

 * or SSL descriptor ssl) into buffer buf, until \r\n\r\n appears in the

 * buffer (which marks the end of HTTP request). Buffer buf may already

 * have some data. The length of the data is stored in nread.

 * Upon every read operation, increase nread by the number of bytes read.

 */

static int

read_request_line(FILE *fp, SOCKET sock, SSL *ssl, char *buf, int bufsiz, int *nread)

{

    int n, request_len;

    request_len = -1;

    while (*nread < bufsiz && request_len == -1) {

        n = pull(fp, sock, ssl, buf + *nread, bufsiz - *nread);

        if (n <= 0) {

            break;

        } else {

            *nread += n;

            if (*nread == bufsiz) {

                g_print(__FILE__ ":%i Too long request received from client. Disconnecting\n", __LINE__);

            }

            request_len = get_request_len_line(buf, (size_t) *nread);

            if (request_len == 0) {

                /* Ignore empty lines */

                request_len = -1;

            }

        }

    }

    return (request_len);

}

void

process_jsonrpc_tcp_connection(struct mg_connection *conn, char *line)

{

    int     api_version = 0;

#ifdef __APPLE__

    /* For some reason the memset below causes a segfault when the size is 512K */

    char    buf[256 * 1024];

#else

    char    buf[512 * 1024];

#endif

    int     request_len, nread;

    char   *response;

    /* Verify the API version */

    if (sscanf(line, "NNTPGrab - API version %i", &api_version) != 1 ||

        api_version != NNTPGRAB_PLUGIN_API_VERSION) {

        mg_printf(conn, "API mismatch, expected %i\r\n", NNTPGRAB_PLUGIN_API_VERSION);

        return;

    } else {

        mg_printf(conn, "OK\r\n");

    }

    /* Add the connection to a list of active connections so events can be emit to all connections */

    add_conn_to_list(conn);

    while (TRUE) {

        nread = 0;

        memset(buf, 0, sizeof(buf));

        request_len = read_request_line(NULL, conn->client.sock, conn->ssl, buf, sizeof(buf) - 1, &nread);

        if (request_len <= 0) {

            break; /* Remote end closed the connection */

        }

        /* 0-terminate the request */

        buf[request_len] = '\0';

        g_print(__FILE__ ":%i Received JSON-RPC over TCP request : %s\n", __LINE__, buf);

        response = jsonrpc_process(buf, conn);

        g_static_mutex_lock(&write_mutex);

        mg_write(conn, response, strlen(response));

        mg_write(conn, "\r\n", 2);

        g_static_mutex_unlock(&write_mutex);

        free(response);

    }

    /* All communication is finished,drop the connection from the list of active connections */

    del_conn_from_list(conn);

}

void

jsonrpc_tcp_emit_event_to_connection(const char *json_data, struct mg_connection *conn)

{

    int len;

    g_return_if_fail(json_data != NULL);

    g_return_if_fail(conn != NULL);

    len = strlen(json_data);

    g_static_mutex_lock(&write_mutex);

    mg_write(conn, json_data, len);

    mg_write(conn, "\r\n", 2);

    g_static_mutex_unlock(&write_mutex);

}

void

jsonrpc_tcp_emit_event(const char *json_data)

{

    GList *list;

    g_static_mutex_lock(&connections_mutex);

    list = active_connections;

    while (list) {

        struct mg_connection *conn = list->data;

        jsonrpc_tcp_emit_event_to_connection(json_data, conn);

        list = g_list_next(list);

    }

    g_static_mutex_unlock(&connections_mutex);

}