sip_transport_tls_ossl.c

基于sip协议的网络电话源码

	listener->factory.pool = NULL;
	pj_pool_release(pool);
    }

    return PJ_SUCCESS;
}


/***************************************************************************/
/*
 * TLS Transport
 */

/*
 * Prototypes.
 */
/* Called by transport manager to send message */
static pj_status_t tls_send_msg(pjsip_transport *transport, 
				pjsip_tx_data *tdata,
				const pj_sockaddr_t *rem_addr,
				int addr_len,
				void *token,
				void (*callback)(pjsip_transport *transport,
						 void *token, 
						 pj_ssize_t sent_bytes));

/* Called by transport manager to shutdown */
static pj_status_t tls_shutdown(pjsip_transport *transport);

/* Called by transport manager to destroy transport */
static pj_status_t tls_destroy_transport(pjsip_transport *transport);

/* Utility to destroy transport */
static pj_status_t tls_destroy(pjsip_transport *transport,
			       pj_status_t reason);

/* Callback from ioqueue on incoming packet */
static void on_read_complete(pj_ioqueue_key_t *key, 
                             pj_ioqueue_op_key_t *op_key, 
                             pj_ssize_t bytes_read);

/* Callback from ioqueue when packet is sent */
static void on_write_complete(pj_ioqueue_key_t *key, 
                              pj_ioqueue_op_key_t *op_key, 
                              pj_ssize_t bytes_sent);

/* Callback from ioqueue when connect completes */
static void on_connect_complete(pj_ioqueue_key_t *key, 
                                pj_status_t status);


/*
 * Common function to create TLS transport, called when pending accept() and
 * pending connect() complete.
 */
static pj_status_t tls_create( struct tls_listener *listener,
			       pj_pool_t *pool,
			       pj_sock_t sock, pj_bool_t is_server,
			       const pj_sockaddr_in *local,
			       const pj_sockaddr_in *remote,
			       struct tls_transport **p_tls)
{
    struct tls_transport *tls;
    pj_ioqueue_t *ioqueue;
    pj_ioqueue_callback tls_callback;
    int rc;
    pj_status_t status;
    

    PJ_ASSERT_RETURN(sock != PJ_INVALID_SOCKET, PJ_EINVAL);


    if (pool == NULL) {
	pool = pjsip_endpt_create_pool(listener->endpt, "tls",
				       POOL_TP_INIT, POOL_TP_INC);
	PJ_ASSERT_RETURN(pool != NULL, PJ_ENOMEM);
    }    

    /*
     * Create and initialize basic transport structure.
     */
    tls = pj_pool_zalloc(pool, sizeof(*tls));
    tls->sock = sock;
    tls->is_server = is_server;
    tls->listener = listener;
    pj_list_init(&tls->delayed_list);
    tls->base.pool = pool;

    pj_ansi_snprintf(tls->base.obj_name, PJ_MAX_OBJ_NAME, 
		     (is_server ? "tlss%p" :"tlsc%p"), tls);

    /* Initialize transport reference counter to 1 */
    status = pj_atomic_create(pool, 1, &tls->base.ref_cnt);
    if (status != PJ_SUCCESS) {
	goto on_error;
    }

    status = pj_lock_create_recursive_mutex(pool, "tls", &tls->base.lock);
    if (status != PJ_SUCCESS) {
	goto on_error;
    }

    tls->base.key.type = PJSIP_TRANSPORT_TLS;
    pj_memcpy(&tls->base.key.rem_addr, remote, sizeof(pj_sockaddr_in));
    tls->base.type_name = "tls";
    tls->base.flag = pjsip_transport_get_flag_from_type(PJSIP_TRANSPORT_TLS);

    tls->base.info = pj_pool_alloc(pool, 64);
    pj_ansi_snprintf(tls->base.info, 64, "TLS to %s:%d",
		     pj_inet_ntoa(remote->sin_addr), 
		     (int)pj_ntohs(remote->sin_port));

    tls->base.addr_len = sizeof(pj_sockaddr_in);
    pj_memcpy(&tls->base.local_addr, local, sizeof(pj_sockaddr_in));
    sockaddr_to_host_port(pool, &tls->base.local_name, local);
    sockaddr_to_host_port(pool, &tls->base.remote_name, remote);

    tls->base.endpt = listener->endpt;
    tls->base.tpmgr = listener->tpmgr;
    tls->base.send_msg = &tls_send_msg;
    tls->base.do_shutdown = &tls_shutdown;
    tls->base.destroy = &tls_destroy_transport;

    /* Create SSL connection object */
    tls->ssl = SSL_new(listener->ctx);
    if (tls->ssl == NULL) {
	ssl_report_error(tls->base.obj_name, 4, PJ_SUCCESS,
			 "Error creating SSL connection object");
	status = PJSIP_TLS_ESSLCONN;
	goto on_error;
    }

    /* Associate network socket with SSL connection object */
    rc = SSL_set_fd(tls->ssl, (int)sock);
    if (rc != 1) {
	ssl_report_error(tls->base.obj_name, 4, PJ_SUCCESS,
			 "Error calling SSL_set_fd");
	status = PJSIP_TLS_ESSLCONN;
	goto on_error;
    }

    /* Register socket to ioqueue */
    pj_bzero(&tls_callback, sizeof(pj_ioqueue_callback));
    tls_callback.on_read_complete = &on_read_complete;
    tls_callback.on_write_complete = &on_write_complete;
    tls_callback.on_connect_complete = &on_connect_complete;

    ioqueue = pjsip_endpt_get_ioqueue(listener->endpt);
    status = pj_ioqueue_register_sock(pool, ioqueue, sock, 
				      tls, &tls_callback, &tls->key);
    if (status != PJ_SUCCESS) {
	goto on_error;
    }

    /* Register transport to transport manager */
    status = pjsip_transport_register(listener->tpmgr, &tls->base);
    if (status != PJ_SUCCESS) {
	goto on_error;
    }

    tls->is_registered = PJ_TRUE;

    /* Done setting up basic transport. */
    *p_tls = tls;

    PJ_LOG(4,(tls->base.obj_name, "TLS %s transport created",
	      (tls->is_server ? "server" : "client")));

    return PJ_SUCCESS;

on_error:
    tls_destroy(&tls->base, status);
    return status;
}


/* Flush all delayed transmision once the socket is connected. 
 * Return non-zero if pending transmission list is empty after
 * the function returns.
 */
static pj_bool_t tls_flush_pending_tx(struct tls_transport *tls)
{
    pj_bool_t empty;

    pj_lock_acquire(tls->base.lock);
    while (!pj_list_empty(&tls->delayed_list)) {
	struct delayed_tdata *pending_tx;
	pjsip_tx_data *tdata;
	pj_ioqueue_op_key_t *op_key;
	pj_ssize_t size;
	pj_status_t status;

	pending_tx = tls->delayed_list.next;

	tdata = pending_tx->tdata_op_key->tdata;
	op_key = (pj_ioqueue_op_key_t*)pending_tx->tdata_op_key;

	/* send the txdata */
	status = ssl_write(tls, tdata);

	/* On EWOULDBLOCK, suspend further transmissions */
	if (status == PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {
	    break;
	}

	/* tdata has been transmitted (successfully or with failure).
	 * In any case, remove it from pending transmission list.
	 */
	pj_list_erase(pending_tx);

	/* Notify callback */
	if (status == PJ_SUCCESS)
	    size = tdata->buf.cur - tdata->buf.start;
	else
	    size = -status;

	on_write_complete(tls->key, op_key, size);

    }

    empty = pj_list_empty(&tls->delayed_list);

    pj_lock_release(tls->base.lock);

    return empty;
}


/* Called by transport manager to destroy transport */
static pj_status_t tls_destroy_transport(pjsip_transport *transport)
{
    struct tls_transport *tls = (struct tls_transport*)transport;

    /* Transport would have been unregistered by now since this callback
     * is called by transport manager.
     */
    tls->is_registered = PJ_FALSE;

    return tls_destroy(transport, tls->close_reason);
}


/* Destroy TLS transport */
static pj_status_t tls_destroy(pjsip_transport *transport, 
			       pj_status_t reason)
{
    struct tls_transport *tls = (struct tls_transport*)transport;

    if (tls->close_reason == 0)
	tls->close_reason = reason;

    if (tls->is_registered) {
	tls->is_registered = PJ_FALSE;
	pjsip_transport_destroy(transport);

	/* pjsip_transport_destroy will recursively call this function
	 * again.
	 */
	return PJ_SUCCESS;
    }

    /* Mark transport as closing */
    ++tls->is_closing;

    /* Cancel all delayed transmits */
    while (!pj_list_empty(&tls->delayed_list)) {
	struct delayed_tdata *pending_tx;
	pj_ioqueue_op_key_t *op_key;

	pending_tx = tls->delayed_list.next;
	pj_list_erase(pending_tx);

	op_key = (pj_ioqueue_op_key_t*)pending_tx->tdata_op_key;

	on_write_complete(tls->key, op_key, -reason);
    }

    if (tls->rdata.tp_info.pool) {
	pj_pool_release(tls->rdata.tp_info.pool);
	tls->rdata.tp_info.pool = NULL;
    }

    if (tls->key) {
	pj_ioqueue_unregister(tls->key);
	tls->key = NULL;
	tls->sock = PJ_INVALID_SOCKET;
    }

    if (tls->sock != PJ_INVALID_SOCKET) {
	pj_sock_close(tls->sock);
	tls->sock = PJ_INVALID_SOCKET;
    }

    if (tls->base.lock) {
	pj_lock_destroy(tls->base.lock);
	tls->base.lock = NULL;
    }

    if (tls->base.ref_cnt) {
	pj_atomic_destroy(tls->base.ref_cnt);
	tls->base.ref_cnt = NULL;
    }

    if (tls->ssl) {
	SSL_free(tls->ssl);
	tls->ssl = NULL;
    }

    if (tls->base.pool) {
	pj_pool_t *pool;

	if (reason != PJ_SUCCESS) {
	    char errmsg[PJ_ERR_MSG_SIZE];

	    pj_strerror(reason, errmsg, sizeof(errmsg));
	    PJ_LOG(4,(tls->base.obj_name, 
		      "TLS transport destroyed with reason %d: %s", 
		      reason, errmsg));

	} else {

	    PJ_LOG(4,(tls->base.obj_name, 
		      "TLS transport destroyed normally"));

	}

	pool = tls->base.pool;
	tls->base.pool = NULL;
	pj_pool_release(pool);
    }

    return PJ_SUCCESS;
}


/*
 * This utility function creates receive data buffers and start
 * asynchronous recv() operations from the socket. It is called after
 * accept() or connect() operation complete.
 */
static pj_status_t tls_start_read(struct tls_transport *tls)
{
    pj_pool_t *pool;
    pj_ssize_t size;
    pj_sockaddr_in *rem_addr;
    pj_status_t status;

    /* Init rdata */
    pool = pjsip_endpt_create_pool(tls->listener->endpt,
				   "rtd%p",
				   PJSIP_POOL_RDATA_LEN,
				   PJSIP_POOL_RDATA_INC);
    if (!pool) {
	ssl_report_error(tls->base.obj_name, 4, PJ_ENOMEM, 
			 "Unable to create pool for listener rxdata");
	return PJ_ENOMEM;
    }

    tls->rdata.tp_info.pool = pool;

    tls->rdata.tp_info.transport = &tls->base;
    tls->rdata.tp_info.tp_data = tls;
    tls->rdata.tp_info.op_key.rdata = &tls->rdata;
    pj_ioqueue_op_key_init(&tls->rdata.tp_info.op_key.op_key, 
			   sizeof(pj_ioqueue_op_key_t));

    tls->rdata.pkt_info.src_addr = tls->base.key.rem_addr;
    tls->rdata.pkt_info.src_addr_len = sizeof(pj_sockaddr_in);
    rem_addr = (pj_sockaddr_in*) &tls->base.key.rem_addr;
    pj_ansi_strcpy(tls->rdata.pkt_info.src_name,
		   pj_inet_ntoa(rem_addr->sin_addr));
    tls->rdata.pkt_info.src_port = pj_ntohs(rem_addr->sin_port);

    /* Here's the real trick with OpenSSL.
     * Since asynchronous socket operation with OpenSSL uses select() like
     * mechanism, it's not really compatible with PJLIB's ioqueue. So to
     * make them "talk" together, we simulate select() by using MSG_PEEK
     * when we call pj_ioqueue_recv().
     */
    size = 1;
    status = pj_ioqueue_recv(tls->key, &tls->rdata.tp_info.op_key.op_key,
			     tls->rdata.pkt_info.packet, &size,
			     PJ_IOQUEUE_ALWAYS_ASYNC | PJ_MSG_PEEK);
    if (status != PJ_SUCCESS && status != PJ_EPENDING) {
	ssl_report_error(tls->base.obj_name, 4, status,
			 "ioqueue_recv() error");
	return status;
    }

    return PJ_SUCCESS;
}



/* This callback is called by transport manager for the TLS factory
 * to create outgoing transport to the specified destination.
 */
static pj_status_t lis_create_transport(pjsip_tpfactory *factory,
					pjsip_tpmgr *mgr,
					pjsip_endpoint *endpt,
					const pj_sockaddr *rem_addr,
					int addr_len,
					pjsip_transport **p_transport)
{
    struct tls_listener *listener;
    struct tls_transport *tls;
    pj_sock_t sock;
    pj_sockaddr_in local_addr;
    pj_status_t status;

    /* Sanity checks */
    PJ_ASSERT_RETURN(factory && mgr && endpt && rem_addr &&
		     addr_len && p_transport, PJ_EINVAL);

    /* Check that address is a sockaddr_in */
    PJ_ASSERT_RETURN(rem_addr->sa_family == PJ_AF_INET &&
		     addr_len == sizeof(pj_sockaddr_in), PJ_EINVAL);


    listener = (struct tls_listener*)factory;

    
    /* Create socket */
    status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_STREAM, 0, &sock);
    if (status != PJ_SUCCESS)
	return status;

    /* Bind to any port */
    status = pj_sock_bind_in(sock, 0, 0);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    /* Get the local port */
    addr_len = sizeof(pj_sockaddr_in);
    status = pj_sock_getsockname(sock, &local_addr, &addr_len);
    if (status != PJ_SUCCESS) {
	pj_sock_close(sock);
	return status;
    }

    /* Initially set the address from the listener's address */
    local_addr.sin_addr.s_addr = 
	((pj_sockaddr_in*)&listener->factory.local_addr)->sin_addr.s_addr;

    /* Create the transport descriptor */
    status = tls_create(listener, NULL, sock, PJ_FALSE, &local_addr, 
			(pj_sockaddr_in*)rem_addr, &tls);
    if (status != PJ_SUCCESS)
	return status;


    /* Start asynchronous connect() operation */
    tls->has_pending_connect = PJ_TRUE;
    status = pj_ioqueue_connect(tls->key, rem_addr, sizeof(pj_sockaddr_in));
    if (status == PJ_SUCCESS) {

	/* Immediate socket connect() ! */
	tls->has_pending_connect = PJ_FALSE;

	/* Perform SSL_connect() */
	status = ssl_connect(tls);
	if (status != PJ_SUCCESS) {
	    tls_destroy(&tls->base, status);
	    return status;
	}

    } else if (status != PJ_EPENDING) {
	tls_destroy(&tls->base, status);
	return status;
    }

    /* Update (again) local address, just in case local address currently
     * set is different now that asynchronous connect() is started.
     */
    addr_len = sizeof(pj_sockaddr_in);
    if (pj_sock_getsockname(tls->sock, &local_addr, &addr_len)==PJ_SUCCESS) {
	pj_sockaddr_in *tp_addr = (pj_sockaddr_in*)&tls->base.local_addr;

	/* Some systems (like old Win32 perhaps) may not set local address
	 * properly before socket is fully connected.
	 */
	if (tp_addr->sin_addr.s_addr != local_addr.sin_addr.s_addr &&
	    local_addr.sin_addr.s_addr != 0) 
	{
	    tp_addr->sin_addr.s_addr = local_addr.sin_addr.s_addr;
	    tp_addr->sin_port = local_addr.sin_port;
	    sockaddr_to_host_port(tls->base.pool, &tls->base.local_name,
				  &local_addr);
	}
    }

    if (tls->has_pending_connect) {
	PJ_LOG(4,(tls->base.obj_name, 
		  "TLS transport %.*s:%d is connecting to %.*s:%d...",
		  (int)tls->base.local_name.host.slen,
		  tls->base.local_name.host.ptr,
		  tls->base.local_name.port,
		  (int)tls->base.remote_name.host.slen,
		  tls->base.remote_name.host.ptr,
		  tls->base.remote_name.port));
    }

    /* Done */
    *p_transport = &tls->base;

    return PJ_SUCCESS;
}


/*
 * This callback is called by ioqueue when pending accept() operation has
 * completed.
 */
static void on_accept_complete(	pj_ioqueue_key_t *key, 
				pj_ioqueue_op_key_t *op_key, 
				pj_sock_t sock, 
				pj_status_t status)
{
    struct tls_listener *listener;
    struct tls_transport *tls;
    struct pending_accept *accept_op;
    int err_cnt = 0;

    listener = pj_ioqueue_get_user_data(key);
    accept_op = (struct pending_accept*) op_key;

    /*
     * Loop while there is immediate connection or when there is error.
     */
    do {
	if (status == PJ_EPENDING) {
	    /*
	     * This can only happen when this function is called during
	     * initialization to kick off asynchronous accept().
	     */

	} else if (status != PJ_SUCCESS) {

	    /*
	     * Error in accept().
	     */
	    ssl_report_error(listener->factory.obj_name, 4, status,
			     "Error in asynchronous accept() completion");

	    /*
	     * Prevent endless accept() error loop by limiting the
	     * number of consecutive errors. Once the number of errors
	     * is equal to maximum, we treat this as permanent error, and
	     * we stop the accept() operation.
	     */
	    ++err_cnt;
	    if (err_cnt >= 10) {
		PJ_LOG(1, (listener->factory.obj_name, 
			   "Too many errors, listener stopping"));