sip_transport_tls_ossl.c

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

	    }

	} 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.
	 */
	tdata_op_key->callback(&tls->base, tdata_op_key->token, bytes_sent);
    }
}


/* Add tdata to pending list */
static void add_pending_tx(struct tls_transport *tls,
			   pjsip_tx_data *tdata)
{
    struct delayed_tdata *delayed_tdata;

    delayed_tdata = pj_pool_alloc(tdata->pool, 
				  sizeof(*delayed_tdata));
    delayed_tdata->tdata_op_key = &tdata->op_key;
    pj_list_push_back(&tls->delayed_list, delayed_tdata);
}


/* 
 * This callback is called by transport manager to send SIP 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))
{
    struct tls_transport *tls = (struct tls_transport*)transport;
    pj_ssize_t size;
    pj_bool_t delayed = PJ_FALSE;
    pj_status_t status = PJ_SUCCESS;

    /* Sanity check */
    PJ_ASSERT_RETURN(transport && tdata, PJ_EINVAL);

    /* Check that there's no pending operation associated with the tdata */
    PJ_ASSERT_RETURN(tdata->op_key.tdata == NULL, PJSIP_EPENDINGTX);
    
    /* Check the address is supported */
    PJ_ASSERT_RETURN(rem_addr && addr_len==sizeof(pj_sockaddr_in), PJ_EINVAL);



    /* Init op key. */
    tdata->op_key.tdata = tdata;
    tdata->op_key.token = token;
    tdata->op_key.callback = callback;

    /* If asynchronous connect() has not completed yet, just put the
     * transmit data in the pending transmission list since we can not
     * use the socket yet.
     */
    if (tls->has_pending_connect) {

	/*
	 * Looks like connect() is still in progress. Check again (this time
	 * with holding the lock) to be sure.
	 */
	pj_lock_acquire(tls->base.lock);

	if (tls->has_pending_connect) {
	    /*
	     * connect() is still in progress. Put the transmit data to
	     * the delayed list.
	     */
	    add_pending_tx(tls, tdata);
	    status = PJ_EPENDING;

	    /* Prevent ssl_write() to be called below */
	    delayed = PJ_TRUE;
	}

	pj_lock_release(tls->base.lock);
    } 
    
    if (!delayed) {

	pj_bool_t no_pending_tx;

	/* Make sure that we've flushed pending tx first so that
	 * stream is in order.
	 */
	no_pending_tx = tls_flush_pending_tx(tls);

	/* Send data immediately with SSL_write() if we don't have
	 * pending data in our list.
	 */
	if (no_pending_tx) {

	    status = ssl_write(tls, tdata);

	    /* On EWOULDBLOCK, put this tdata in the list */
	    if (status == PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {
		add_pending_tx(tls, tdata);
		status = PJ_EPENDING;
	    }

	    if (status != PJ_EPENDING) {
		/* Not pending (could be immediate success or error) */
		tdata->op_key.tdata = NULL;

		/* Shutdown transport on closure/errors */
		if (status != PJ_SUCCESS) {
		    size = -status;

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

		    if (tls->close_reason==PJ_SUCCESS) tls->close_reason = status;
		    pjsip_transport_shutdown(&tls->base);
		}
	    }

	} else {
	    /* We have pending data in our list, so queue the txdata
	     * in the pending tx list.
	     */
	    add_pending_tx(tls, tdata);
	    status = PJ_EPENDING;
	}
    }

    return status;
}


/* 
 * This callback is called by transport manager to shutdown transport.
 * This normally is only used by UDP transport.
 */
static pj_status_t tls_shutdown(pjsip_transport *transport)
{
    struct tls_transport *tls = (struct tls_transport*)transport;

    /* Shutdown SSL */
    if (!tls->ssl_shutdown_called) {
	/* Release our reference counter and shutdown SSL */
	pjsip_transport_dec_ref(transport);
	SSL_shutdown(tls->ssl);
	tls->ssl_shutdown_called = PJ_TRUE;

	PJ_LOG(4,(transport->obj_name, "TLS transport shutdown"));
    }

    return PJ_SUCCESS;
}


/* 
 * Callback from ioqueue that an incoming data is received from the socket.
 */
static void on_read_complete(pj_ioqueue_key_t *key, 
                             pj_ioqueue_op_key_t *op_key, 
                             pj_ssize_t bytes_read_unused)
{
    enum { MAX_IMMEDIATE_PACKET = 10 };
    pjsip_rx_data_op_key *rdata_op_key = (pjsip_rx_data_op_key*) op_key;
    pjsip_rx_data *rdata = rdata_op_key->rdata;
    struct tls_transport *tls = 
	(struct tls_transport*)rdata->tp_info.transport;
    int i;
    pj_status_t status;

    /* Don't do anything if transport is closing. */
    if (tls->is_closing) {
	tls->is_closing++;
	return;
    }


    /* Recall that we use MSG_PEEK when calling ioqueue_recv(), so
     * when this callback is called, data has not actually been read
     * from socket buffer.
     */

    for (i=0;; ++i) {
	pj_uint32_t flags;

	/* Read data from SSL connection */
	status = ssl_read(tls);

	if (status == PJ_SUCCESS) {
	    /*
	     * We have packet!
	     */
	    pj_size_t size_eaten;

	    /* Init pkt_info part. */
	    rdata->pkt_info.zero = 0;
	    pj_gettimeofday(&rdata->pkt_info.timestamp);

	    /* Report to transport manager.
	     * The transport manager will tell us how many bytes of the packet
	     * have been processed (as valid SIP message).
	     */
	    size_eaten = 
		pjsip_tpmgr_receive_packet(rdata->tp_info.transport->tpmgr, 
					   rdata);

	    pj_assert(size_eaten <= (pj_size_t)rdata->pkt_info.len);

	    /* Move unprocessed data to the front of the buffer */
	    if (size_eaten>0 && size_eaten<(pj_size_t)rdata->pkt_info.len) {
		pj_memmove(rdata->pkt_info.packet,
			   rdata->pkt_info.packet + size_eaten,
			   rdata->pkt_info.len - size_eaten);
	    }
	    
	    rdata->pkt_info.len -= size_eaten;

	} else if (status == PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {

	    /* Ignore EWOULDBLOCK error (?) */

	} else {

	    /* For other errors, treat as transport being closed */
	    ssl_report_error(tls->base.obj_name, 4, status,
			     "Error reading SSL stream");
	    
	    /* We can not destroy the transport since high level objects may
	     * still keep reference to this transport. So we can only 
	     * instruct transport manager to gracefully start the shutdown
	     * procedure for this transport.
	     */
	    if (tls->close_reason==PJ_SUCCESS) 
		tls->close_reason = status;
	    pjsip_transport_shutdown(&tls->base);

	    return;
	}

	/* Reset pool. */
	pj_pool_reset(rdata->tp_info.pool);

	/* If we have pending data in SSL buffer, read it. */
	if (SSL_pending(tls->ssl)) {

	    /* Check that we have enough space in buffer */
	    if (rdata->pkt_info.len >= PJSIP_MAX_PKT_LEN-1) {
		PJ_LOG(4,(tls->base.obj_name, 
			  "Incoming packet dropped from tls:%.*s:%d "
			  "because it's too big (%d bytes)",
			  (int)tls->base.remote_name.host.slen,
			  tls->base.remote_name.host.ptr,
			  tls->base.remote_name.port,
			  rdata->pkt_info.len));
		rdata->pkt_info.len = 0;
	    }

	    continue;
	}

	/* If we've reached maximum number of packets received on a single
	 * poll, force the next reading to be asynchronous.
	 */
	if (i >= MAX_IMMEDIATE_PACKET) {
	    /* Receive quota reached. Force ioqueue_recv() to 
	     * return PJ_EPENDING 
	     */
	    flags = PJ_IOQUEUE_ALWAYS_ASYNC;
	} else {
	    /* This doesn't work too well when IOCP is used, as it seems that
	     * IOCP refuses to do MSG_PEEK when WSARecv() is called without
	     * OVERLAP structure. With OpenSSL it gives this error:
	     * error:1408F10B:SSL routines:SSL3_GET_RECORD:wrong version 
	     * number
	     *
	     * flags = 0 
	     */
	    flags = PJ_IOQUEUE_ALWAYS_ASYNC;
	}

	/* Read next packet from the network. Remember, we need to use
	 * MSG_PEEK or otherwise the packet will be eaten by us!
	 */
	bytes_read_unused = 1;
	status = pj_ioqueue_recv(key, op_key, 
				 rdata->pkt_info.packet+rdata->pkt_info.len,
				 &bytes_read_unused, flags | PJ_MSG_PEEK);

	if (status == PJ_SUCCESS) {

	    /* Continue loop. */
	    pj_assert(i < MAX_IMMEDIATE_PACKET);

	} else if (status == PJ_EPENDING) {
	    break;

	} else {
	    /* Socket error */

	    /* We can not destroy the transport since high level objects may
	     * still keep reference to this transport. So we can only 
	     * instruct transport manager to gracefully start the shutdown
	     * procedure for this transport.
	     */
	    if (tls->close_reason==PJ_SUCCESS) tls->close_reason = status;
	    pjsip_transport_shutdown(&tls->base);

	    return;
	}
    }
}


/* 
 * Callback from ioqueue when asynchronous connect() operation completes.
 */
static void on_connect_complete(pj_ioqueue_key_t *key, 
                                pj_status_t status)
{
    struct tls_transport *tls;
    pj_sockaddr_in addr;
    int addrlen;

    tls = pj_ioqueue_get_user_data(key);

    /* Check connect() status */
    if (status != PJ_SUCCESS) {

	/* Mark that pending connect() operation has completed. */
	tls->has_pending_connect = PJ_FALSE;

	ssl_report_error(tls->base.obj_name, 4, status,
			 "Error connecting to %.*s:%d",
			 (int)tls->base.remote_name.host.slen,
			 tls->base.remote_name.host.ptr,
			 tls->base.remote_name.port);

	/* 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, -status);
	}

	/* We can not destroy the transport since high level objects may
	 * still keep reference to this transport. So we can only 
	 * instruct transport manager to gracefully start the shutdown
	 * procedure for this transport.
	 */
	if (tls->close_reason==PJ_SUCCESS) tls->close_reason = status;
	pjsip_transport_shutdown(&tls->base);
	return;
    }

    PJ_LOG(4,(tls->base.obj_name, 
	      "TCP transport %.*s:%d is connected 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));


    /* Update (again) local address, just in case local address currently
     * set is different now that the socket is connected (could happen
     * on some systems, like old Win32 probably?).
     */
    addrlen = sizeof(pj_sockaddr_in);
    if (pj_sock_getsockname(tls->sock, &addr, &addrlen)==PJ_SUCCESS) {
	pj_sockaddr_in *tp_addr = (pj_sockaddr_in*)&tls->base.local_addr;

	if (tp_addr->sin_addr.s_addr != addr.sin_addr.s_addr) {
	    tp_addr->sin_addr.s_addr = addr.sin_addr.s_addr;
	    tp_addr->sin_port = addr.sin_port;
	    sockaddr_to_host_port(tls->base.pool, &tls->base.local_name,
				  tp_addr);
	}
    }

    /* Perform SSL_connect() */
    status = ssl_connect(tls);
    if (status != PJ_SUCCESS) {

	/* 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, -status);
	}

	if (tls->close_reason==PJ_SUCCESS) tls->close_reason = status;
	pjsip_transport_shutdown(&tls->base);
	return;
    }

    /* Mark that pending connect() operation has completed. */
    tls->has_pending_connect = PJ_FALSE;

    /* Start pending read */
    status = tls_start_read(tls);
    if (status != PJ_SUCCESS) {

	/* 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, -status);
	}


	/* We can not destroy the transport since high level objects may
	 * still keep reference to this transport. So we can only 
	 * instruct transport manager to gracefully start the shutdown
	 * procedure for this transport.
	 */
	if (tls->close_reason==PJ_SUCCESS) tls->close_reason = status;
	pjsip_transport_shutdown(&tls->base);
	return;
    }

    /* Flush all pending send operations */
    tls_flush_pending_tx(tls);
}

#endif	/* PJSIP_HAS_TLS_TRANSPORT */