conn.c

来自「mms client」· C语言 代码 · 共 1,478 行 · 第 1/3 页

/* conn.c - implement Connection type
 *
 * This file implements the interface defined in conn.h.
 *
 * Richard Braakman
 *
 * SSL client implementation contributed by
 * Jarkko Kovala <[e-mail]>
 *
 * SSL server implementation contributed by
 * Stipe Tolj <[e-mail]> for Wapme Systems AG
 */

/* TODO: unlocked_close() on error */
/* TODO: have I/O functions check if connection is open */
/* TODO: have conn_open_tcp do a non-blocking connect() */

#include <signal.h>
#include <unistd.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>

#include "gwlib/gwlib.h"

#ifdef HAVE_LIBSSL
#include <openssl/ssl.h>
#include <openssl/err.h>

SSL_CTX *global_ssl_context = NULL;
SSL_CTX *global_server_ssl_context = NULL;
X509 *ssl_public_cert;
RSA *ssl_private_key;
#endif /* HAVE_LIBSSL */

typedef unsigned long (*CRYPTO_CALLBACK_PTR)(void);

/*
 * This used to be 4096.  It is now 0 so that callers don't have to
 * deal with the complexities of buffering (i.e. deciding when to
 * flush) unless they want to.
 * FIXME: Figure out how to combine buffering sensibly with use of
 * conn_register.
 */
#define DEFAULT_OUTPUT_BUFFERING 0
#define SSL_CONN_TIMEOUT         30

struct Connection
{
    /* We use separate locks for input and ouput fields, so that
     * read and write activities don't have to get in each other's
     * way.  If you need both, then acquire the outlock first. */
    Mutex *inlock;
    Mutex *outlock;
    volatile sig_atomic_t claimed;
#ifndef NO_GWASSERT
    long claiming_thread;
#endif

    /* fd value is read-only and is not locked */
    int fd;

  /* socket state */
  enum {yes,no} connected;

    /* Protected by outlock */
    Octstr *outbuf;
    long outbufpos;   /* start of unwritten data in outbuf */

    /* Try to buffer writes until there are this many octets to send.
     * Set it to 0 to get an unbuffered connection. */
    unsigned int output_buffering;

    /* Protected by inlock */
    Octstr *inbuf;
    long inbufpos;    /* start of unread data in inbuf */

    int read_eof;     /* we encountered eof on read */
    int read_error;   /* we encountered error on read */

    /* Protected by both locks when updating, so you need only one
     * of the locks when reading. */
    FDSet *registered;
    conn_callback_t *callback;
    void *callback_data;
    /* Protected by inlock */
    int listening_pollin;
    /* Protected by outlock */
    int listening_pollout;

#ifdef HAVE_LIBSSL
    SSL *ssl;
    X509 *peer_certificate;
    Mutex *ssl_mutex;
#endif /* HAVE_LIBSSL */
};

static void unlocked_register_pollin(Connection *conn, int onoff);
static void unlocked_register_pollout(Connection *conn, int onoff);

/* There are a number of functions that play with POLLIN and POLLOUT flags.
 * The general rule is that we always want to poll for POLLIN except when
 * we have detected eof (which may be reported as eternal POLLIN), and
 * we want to poll for POLLOUT only if there's data waiting in the
 * output buffer.  If output buffering is set, we may not want to poll for
 * POLLOUT if there's not enough data waiting, which is why we have
 * unlocked_try_write. */

/* Macros to get more information for debugging purposes */
#define unlock_in(conn) unlock_in_real(conn, __FILE__, __LINE__, __func__)
#define unlock_out(conn) unlock_out_real(conn, __FILE__, __LINE__, __func__)

/* Lock a Connection's read direction, if the Connection is unclaimed */
static void lock_in(Connection *conn)
{
    gw_assert(conn != NULL);

    if (conn->claimed)
        gw_assert(gwthread_self() == conn->claiming_thread);
    else
        mutex_lock(conn->inlock);
}

/* Unlock a Connection's read direction, if the Connection is unclaimed */
static void unlock_in_real(Connection *conn, char *file, int line, const char *func)
{
    int ret;
    gw_assert(conn != NULL);

    if (!conn->claimed) {
        if ((ret = mutex_unlock(conn->inlock)) != 0) {
            panic(0, "%s:%ld: %s: Mutex unlock failed. " \
		             "(Called from %s:%ld:%s.)", \
			         __FILE__, (long) __LINE__, __func__, \
			         file, (long) line, func);
        }
     }
}

/* Lock a Connection's write direction, if the Connection is unclaimed */
static void lock_out(Connection *conn)
{
    gw_assert(conn != NULL);

    if (conn->claimed)
        gw_assert(gwthread_self() == conn->claiming_thread);
    else
        mutex_lock(conn->outlock);
}

/* Unlock a Connection's write direction, if the Connection is unclaimed */
static void unlock_out_real(Connection *conn, char *file, int line, const char *func)
{
    int ret;
    gw_assert(conn != NULL);

    if (!conn->claimed) {
        if ((ret = mutex_unlock(conn->outlock)) != 0) {
            panic(0, "%s:%ld: %s: Mutex unlock failed. " \
		             "(Called from %s:%ld:%s.)", \
			         __FILE__, (long) __LINE__, __func__, \
			         file, (long) line, func);
        }
     }
}

/* Return the number of bytes in the Connection's output buffer */
static long unlocked_outbuf_len(Connection *conn)
{
    return octstr_len(conn->outbuf) - conn->outbufpos;
}

/* Return the number of bytes in the Connection's input buffer */
static long unlocked_inbuf_len(Connection *conn)
{
    return octstr_len(conn->inbuf) - conn->inbufpos;
}

/* Send as much data as can be sent without blocking.  Return the number
 * of bytes written, or -1 in case of error. */
static long unlocked_write(Connection *conn)
{
    long ret;

#ifdef HAVE_LIBSSL
    if (conn->ssl != NULL) {
        mutex_lock(conn->ssl_mutex);
        ret = SSL_write(conn->ssl, 
                        octstr_get_cstr(conn->outbuf) + conn->outbufpos, 
                        octstr_len(conn->outbuf) - conn->outbufpos);

        if (ret < 0) {
            int SSL_error = SSL_get_error(conn->ssl, ret); 

            if (SSL_error == SSL_ERROR_WANT_READ) {
                SSL_read(conn->ssl, NULL, 0);
                mutex_unlock(conn->ssl_mutex);
                return 0;
            } else if (SSL_error == SSL_ERROR_WANT_WRITE) {
                mutex_unlock(conn->ssl_mutex);
                return 0;
            } else {
                error(0, "SSL write failed: OpenSSL error %d: %s", 
                      SSL_error, ERR_error_string(SSL_error, NULL));
                mutex_unlock(conn->ssl_mutex);
                return -1;
            }
        }
        mutex_unlock(conn->ssl_mutex);
    } else
#endif /* HAVE_LIBSSL */
        ret = octstr_write_data(conn->outbuf, conn->fd, conn->outbufpos);

    if (ret < 0)
        return -1;

    conn->outbufpos += ret;

    /* Heuristic: Discard the already-written data if it's more than
     * half of the total.  This should keep the buffer size small
     * without wasting too many cycles on moving data around. */
    if (conn->outbufpos > octstr_len(conn->outbuf) / 2) {
        octstr_delete(conn->outbuf, 0, conn->outbufpos);
        conn->outbufpos = 0;
    }

    if (conn->registered)
        unlocked_register_pollout(conn, unlocked_outbuf_len(conn) > 0);

    return ret;
}

/* Try to empty the output buffer without blocking.  Return 0 for success,
 * 1 if there is still data left in the buffer, and -1 for errors. */
static int unlocked_try_write(Connection *conn)
{
    long len;

    len = unlocked_outbuf_len(conn);
    if (len == 0)
        return 0;

    if (len < (long) conn->output_buffering)
        return 1;

    if (unlocked_write(conn) < 0)
        return -1;

    if (unlocked_outbuf_len(conn) > 0)
        return 1;

    return 0;
}

/* Read whatever data is currently available, up to an internal maximum. */
static void unlocked_read(Connection *conn)
{
    unsigned char buf[4096];
    long len;

    if (conn->inbufpos > 0) {
        octstr_delete(conn->inbuf, 0, conn->inbufpos);
        conn->inbufpos = 0;
    }

#ifdef HAVE_LIBSSL
    if (conn->ssl != NULL) {
        mutex_lock(conn->ssl_mutex);
        len = SSL_read(conn->ssl, buf, sizeof(buf));

        if (len < 0) { 
            int SSL_error = SSL_get_error(conn->ssl, len);

            if (SSL_error == SSL_ERROR_WANT_WRITE) {
                len = SSL_write(conn->ssl, NULL, 0);
                mutex_unlock(conn->ssl_mutex);
                return;
            } else if (SSL_error == SSL_ERROR_WANT_READ) {
                mutex_unlock(conn->ssl_mutex);
                return;
            } else
                error(0, "SSL read failed: OpenSSL error %d: %s",
                      SSL_error, ERR_error_string(SSL_error, NULL));
        }
        mutex_unlock(conn->ssl_mutex);
    } else
#endif /* HAVE_LIBSSL */
        len = read(conn->fd, buf, sizeof(buf));

    if (len < 0) {
        if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
            return;
        error(errno, "Error reading from fd %d:", conn->fd);
        conn->read_error = 1;
        if (conn->registered)
            unlocked_register_pollin(conn, 0);
        return;
    } else if (len == 0) {
        conn->read_eof = 1;
        if (conn->registered)
            unlocked_register_pollin(conn, 0);
    } else {
        octstr_append_data(conn->inbuf, buf, len);
    }
}

/* Cut "length" octets from the input buffer and return them as an Octstr */
static Octstr *unlocked_get(Connection *conn, long length)
{
    Octstr *result = NULL;

    gw_assert(unlocked_inbuf_len(conn) >= length);
    result = octstr_copy(conn->inbuf, conn->inbufpos, length);
    conn->inbufpos += length;

    return result;
}

/* Tell the fdset whether we are interested in POLLIN events, but only
 * if the status changed.  (Calling fdset_listen can be expensive if
 * it requires synchronization with the polling thread.)
 * We must already have the inlock.
 */
static void unlocked_register_pollin(Connection *conn, int onoff)
{
    gw_assert(conn->registered);

    if (onoff == 1 && !conn->listening_pollin) {
        /* Turn it on */
        conn->listening_pollin = 1;
        fdset_listen(conn->registered, conn->fd, POLLIN, POLLIN);
    } else if (onoff == 0 && conn->listening_pollin) {
        /* Turn it off */
        conn->listening_pollin = 0;
        fdset_listen(conn->registered, conn->fd, POLLIN, 0);
    }
}

/* Tell the fdset whether we are interested in POLLOUT events, but only
 * if the status changed.  (Calling fdset_listen can be expensive if
 * it requires synchronization with the polling thread.)
 * We must already have the outlock.
 */
static void unlocked_register_pollout(Connection *conn, int onoff)
{
    gw_assert(conn->registered);

    if (onoff == 1 && !conn->listening_pollout) {
        /* Turn it on */
        conn->listening_pollout = 1;
        fdset_listen(conn->registered, conn->fd, POLLOUT, POLLOUT);
    } else if (onoff == 0 && conn->listening_pollout) {
        /* Turn it off */
        conn->listening_pollout = 0;
        fdset_listen(conn->registered, conn->fd, POLLOUT, 0);
    }
}

#ifdef HAVE_LIBSSL

Connection *conn_open_ssl(Octstr *host, int port, Octstr *certkeyfile,
			  Octstr *our_host)
{
    Connection *ret;
    int SSL_ret = 0;
    int connected = 0;
    time_t timeout;

    /* open the TCP connection */
    if (!(ret = conn_open_tcp(host, port, our_host))) {
        return NULL;
    }

    ret->ssl = SSL_new(global_ssl_context);
    ret->ssl_mutex = mutex_create();
    
    SSL_set_connect_state(ret->ssl);
    if (certkeyfile != NULL) {
        SSL_use_certificate_file(ret->ssl, octstr_get_cstr(certkeyfile),
                                 SSL_FILETYPE_PEM);
        SSL_use_PrivateKey_file(ret->ssl, octstr_get_cstr(certkeyfile), 
                                SSL_FILETYPE_PEM);
        if (SSL_check_private_key(ret->ssl) != 1) {
            error(0, "conn_open_ssl: private key isn't consistent with the "
                     "certificate from file %s (or failed reading the file)",
                  octstr_get_cstr(certkeyfile));
            goto error;
        }
    }
    
    SSL_set_fd(ret->ssl, ret->fd);

    /*
     * The current thread's error queue must be empty before
     * the TLS/SSL I/O operation is attempted, or SSL_get_error()
     * will not work reliably.
     */
    ERR_clear_error();

    /*
     * make the socket is non-blocking while we do SSL_connect
     */
    if (socket_set_blocking(ret->fd, 0) < 0) {
        goto error;
    }

    /* record current time */
    timeout = time(NULL);

    while(!connected && (timeout + SSL_CONN_TIMEOUT > time(NULL))) {
        /* Attempt to connect as long as the timeout hasn't run down */
        SSL_ret = SSL_connect(ret->ssl);
        switch(SSL_get_error(ret->ssl,SSL_ret)) {
            case SSL_ERROR_WANT_READ:
            case SSL_ERROR_WANT_WRITE:
                /* non-blocking socket wants more time to read or write */
                gwthread_sleep(0.01F);
                continue;
            default:
                /* we're connected to the server successfuly */
                connected++;
        }
    }

    if (!connected) {
        /* connection timed out - this probably means that something is terrible wrong */
        int SSL_error = SSL_get_error (ret->ssl, SSL_ret);
        error(0,"SSL connection timeout: OpenSSL error %d: %s",
                SSL_error, ERR_error_string(SSL_error, NULL));
        goto error;
    }
    
    /* 
     * XXX - restore the non-blocking state
     * we don't need this since we use non-blocking operations
     * anyway before doing SSL_connect(), right?!
     */
    /*
    if (socket_set_blocking(ret->fd, 0) < 0) {
        goto error;
    }
    */

    if (SSL_ret != 1) {
        int SSL_error = SSL_get_error (ret->ssl, SSL_ret); 
        error(0, "SSL connect failed: OpenSSL error %d: %s", 
                  SSL_error, ERR_error_string(SSL_error, NULL));
        goto error;
    }
    
    return ret;
error:
    conn_destroy(ret);
    return NULL;
}

#endif /* HAVE_LIBSSL */

Connection *conn_open_tcp(Octstr *host, int port, Octstr *our_host)
{
    return conn_open_tcp_with_port(host, port, 0, our_host);
}

Connection *conn_open_tcp_nb(Octstr *host, int port, Octstr *our_host)
{
  return conn_open_tcp_nb_with_port(host, port, 0, our_host);
}

Connection *conn_open_tcp_nb_with_port(Octstr *host, int port, int our_port,
				       Octstr *our_host)
{
  int sockfd;
  int done = -1;
  Connection *c;
  
  sockfd = tcpip_connect_nb_to_server_with_port(octstr_get_cstr(host), port,
						our_port, our_host == NULL ?
						NULL : octstr_get_cstr(our_host), &done);
  if (sockfd < 0)
    return NULL;
  c = conn_wrap_fd(sockfd, 0);
  if (done != 0) {
    c->connected = no;
  }
  return c;
}

int conn_is_connected(Connection *conn) 
{
  if (conn->connected == yes) return 0;
  return -1;