sip_transport_tls_ossl.c

一个开源SIP协议栈

    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"));
	    }

	} else {
	    pj_pool_t *pool;
	    struct pending_accept *new_op;

	    if (sock == PJ_INVALID_SOCKET) {
		sock = accept_op->new_sock;
	    }

	    if (sock == PJ_INVALID_SOCKET) {
		pj_assert(!"Should not happen. status should be error");
		goto next_accept;
	    }

	    PJ_LOG(4,(listener->factory.obj_name, 
		      "TLS listener %.*s:%d: got incoming TCP connection "
		      "from %s:%d, sock=%d",
		      (int)listener->factory.addr_name.host.slen,
		      listener->factory.addr_name.host.ptr,
		      listener->factory.addr_name.port,
		      pj_inet_ntoa(accept_op->remote_addr.sin_addr),
		      pj_ntohs(accept_op->remote_addr.sin_port),
		      sock));

	    /* Create new accept_opt */
	    pool = pjsip_endpt_create_pool(listener->endpt, "tlss%p", 
					   POOL_TP_INIT, POOL_TP_INC);
	    new_op = pj_pool_zalloc(pool, sizeof(struct pending_accept));
	    new_op->pool = pool;
	    new_op->listener = listener;
	    new_op->index = accept_op->index;
	    pj_ioqueue_op_key_init(&new_op->op_key, sizeof(new_op->op_key));
	    listener->accept_op[accept_op->index] = new_op;

	    /* 
	     * Incoming connections!
	     * Create TLS transport for the new socket.
	     */
	    status = tls_create( listener, accept_op->pool, sock, PJ_TRUE,
				 &accept_op->local_addr, 
				 &accept_op->remote_addr, &tls);
	    if (status == PJ_SUCCESS) {
		/* Complete SSL_accept() */
		status = ssl_accept(tls);
	    }

	    if (status == PJ_SUCCESS) {
		/* Start asynchronous read from the socket */
		status = tls_start_read(tls);
	    }

	    if (status != PJ_SUCCESS) {
		ssl_report_error(tls->base.obj_name, 4, status,
				 "Error creating incoming TLS transport");
		pjsip_transport_shutdown(&tls->base);
	    }

	    accept_op = new_op;
	}

next_accept:
	/*
	 * Start the next asynchronous accept() operation.
	 */
	accept_op->addr_len = sizeof(pj_sockaddr_in);
	accept_op->new_sock = PJ_INVALID_SOCKET;

	status = pj_ioqueue_accept(listener->key, 
				   &accept_op->op_key,
				   &accept_op->new_sock,
				   &accept_op->local_addr,
				   &accept_op->remote_addr,
				   &accept_op->addr_len);

	/*
	 * Loop while we have immediate connection or when there is error.
	 */

    } while (status != PJ_EPENDING);
}


/* 
 * 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)
{
    struct tls_transport *tls = pj_ioqueue_get_user_data(key);
    pjsip_tx_data_op_key *tdata_op_key = (pjsip_tx_data_op_key*)op_key;

    tdata_op_key->tdata = NULL;

    /* Check for error/closure */
    if (bytes_sent <= 0) {
	pj_status_t status;

	ssl_report_error(tls->base.obj_name, 4, -bytes_sent, 
		         "TLS send() error");

	status = (bytes_sent == 0) ? PJ_STATUS_FROM_OS(OSERR_ENOTCONN) :
				     -bytes_sent;
	if (tls->close_reason==PJ_SUCCESS) tls->close_reason = status;
	pjsip_transport_shutdown(&tls->base);
    }

    if (tdata_op_key->callback) {
	/*
	 * Notify sip_transport.c that packet has been sent.
	 */