turn_usage.c

一个开源的sip源代码

/* $Id: turn_usage.c 1300 2007-05-25 06:11:35Z bennylp $ */
/* 
 * Copyright (C) 2003-2005 Benny Prijono <[e-mail]>
 *
 * 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 
 */
#include "server.h"

#define THIS_FILE   "turn_usage.c"

#define MAX_CLIENTS		8000
#define MAX_PEER_PER_CLIENT	16
#define START_PORT		2000
#define END_PORT		65530

/*
 * Forward declarations.
 */
struct turn_usage;
struct turn_client;

static void	   tu_on_rx_data(pj_stun_usage *usage,
				 void *pkt,
				 pj_size_t pkt_size,
				 const pj_sockaddr_t *src_addr,
				 unsigned src_addr_len);
static void	   tu_on_destroy(pj_stun_usage *usage);
static pj_status_t tu_sess_on_send_msg(pj_stun_session *sess,
				       const void *pkt,
				       pj_size_t pkt_size,
				       const pj_sockaddr_t *dst_addr,
				       unsigned addr_len);
static pj_status_t tu_sess_on_rx_request(pj_stun_session *sess,
					 const pj_uint8_t *pkt,
					 unsigned pkt_len,
					 const pj_stun_msg *msg,
					 const pj_sockaddr_t *src_addr,
					 unsigned src_addr_len);

static pj_status_t handle_binding_req(pj_stun_session *session,
				      const pj_stun_msg *msg,
				      const pj_sockaddr_t *src_addr,
				      unsigned src_addr_len);

static pj_status_t client_create(struct turn_usage *tu,
				 const pj_sockaddr_t *src_addr,
				 unsigned src_addr_len,
				 struct turn_client **p_client);
static pj_status_t client_destroy(struct turn_client *client,
				  pj_status_t reason);
static pj_status_t client_handle_stun_msg(struct turn_client *client,
					  const pj_stun_msg *msg,
					  const pj_sockaddr_t *src_addr,
					  unsigned src_addr_len);


struct turn_usage
{
    pj_pool_factory	*pf;
    pj_stun_config	*cfg;
    pj_ioqueue_t	*ioqueue;
    pj_timer_heap_t	*timer_heap;
    pj_pool_t		*pool;
    pj_mutex_t		*mutex;
    pj_stun_usage	*usage;
    int			 type;
    pj_stun_session	*default_session;
    pj_hash_table_t	*client_htable;
    pj_stun_auth_cred	*cred;
    pj_bool_t		 use_fingerprint;

    unsigned		 max_bw_kbps;
    unsigned		 max_lifetime;

    unsigned		 next_port;
};

struct peer;

struct turn_client
{
    char		 obj_name[PJ_MAX_OBJ_NAME];
    struct turn_usage	*tu;
    pj_pool_t		*pool;
    pj_stun_session	*session;
    pj_mutex_t		*mutex;

    pj_sockaddr_in	 client_src_addr;

    /* Socket and socket address of the allocated port */
    int			 sock_type;
    pj_sock_t		 sock;
    pj_ioqueue_key_t	*key;
    pj_sockaddr_in	 alloc_addr;

    /* Allocation properties */
    unsigned		 bw_kbps;
    unsigned		 lifetime;
    pj_timer_entry	 expiry_timer;


    /* Hash table to keep all peers, key-ed by their address */
    pj_hash_table_t	*peer_htable;

    /* Active destination, or sin_addr.s_addr will be zero if
     * no active destination is set.
     */
    struct peer		*active_peer;

    /* Current packet received/sent from/to the allocated port */
    pj_uint8_t		 pkt[4000];
    pj_sockaddr_in	 pkt_src_addr;
    int			 pkt_src_addr_len;
    pj_ioqueue_op_key_t	 pkt_read_key;
    pj_ioqueue_op_key_t  pkt_write_key;
};

struct peer
{
    struct turn_client   *client;
    pj_sockaddr_in	  addr;
};

struct session_data
{
    struct turn_usage	*tu;
    struct turn_client	*client;
};


/*
 * This is the only public API, to create and start the TURN usage.
 */
PJ_DEF(pj_status_t) pj_stun_turn_usage_create(pj_stun_server *srv,
					      int type,
					      const pj_str_t *ip_addr,
					      unsigned port,
					      pj_bool_t use_fingerprint,
					      pj_stun_usage **p_bu)
{
    pj_pool_t *pool;
    struct turn_usage *tu;
    pj_stun_server_info *si;
    pj_stun_usage_cb usage_cb;
    pj_stun_session_cb sess_cb;
    struct session_data *sd;
    pj_sockaddr_in local_addr;
    pj_status_t status;

    PJ_ASSERT_RETURN(srv && (type==PJ_SOCK_DGRAM||type==PJ_SOCK_STREAM),
		     PJ_EINVAL);
    si = pj_stun_server_get_info(srv);

    pool = pj_pool_create(si->pf, "turn%p", 4000, 4000, NULL);
    tu = PJ_POOL_ZALLOC_T(pool, struct turn_usage);
    tu->pool = pool;
    tu->type = type;
    tu->pf = si->pf;
    tu->cfg = &si->stun_cfg;
    tu->ioqueue = si->ioqueue;
    tu->timer_heap = si->timer_heap;
    tu->next_port = START_PORT;
    tu->max_bw_kbps = 64;
    tu->max_lifetime = 10 * 60;
    tu->use_fingerprint = use_fingerprint;

    status = pj_sockaddr_in_init(&local_addr, ip_addr, (pj_uint16_t)port);
    if (status != PJ_SUCCESS)
	return status;

    /* Create usage */
    pj_bzero(&usage_cb, sizeof(usage_cb));
    usage_cb.on_rx_data = &tu_on_rx_data;
    usage_cb.on_destroy = &tu_on_destroy;
    status = pj_stun_usage_create(srv, "turn%p", &usage_cb,
				  PJ_AF_INET, tu->type, 0,
				  &local_addr, sizeof(local_addr),
				  &tu->usage);
    if (status != PJ_SUCCESS) {
	pj_pool_release(pool);
	return status;
    }
    pj_stun_usage_set_user_data(tu->usage, tu);

    /* Init hash tables */
    tu->client_htable = pj_hash_create(tu->pool, MAX_CLIENTS);

    /* Create default session */
    pj_bzero(&sess_cb, sizeof(sess_cb));
    sess_cb.on_send_msg = &tu_sess_on_send_msg;
    sess_cb.on_rx_request = &tu_sess_on_rx_request;
    status = pj_stun_session_create(&si->stun_cfg, "turns%p", &sess_cb, 
				    use_fingerprint, &tu->default_session);
    if (status != PJ_SUCCESS) {
	pj_stun_usage_destroy(tu->usage);
	return status;
    }

    sd = PJ_POOL_ZALLOC_T(pool, struct session_data);
    sd->tu = tu;
    pj_stun_session_set_user_data(tu->default_session, sd);

    pj_stun_session_set_server_name(tu->default_session, NULL);

    /* Create mutex */
    status = pj_mutex_create_recursive(pool, "turn%p", &tu->mutex);
    if (status != PJ_SUCCESS) {
	pj_stun_usage_destroy(tu->usage);
	return status;
    }

    if (p_bu) {
	*p_bu = tu->usage;
    }

    return PJ_SUCCESS;
}


PJ_DEF(pj_status_t) pj_stun_turn_usage_set_credential(pj_stun_usage *turn,
						      const pj_stun_auth_cred *c)
{
    struct turn_usage *tu;
    tu = (struct turn_usage*) pj_stun_usage_get_user_data(turn);

    tu->cred = PJ_POOL_ZALLOC_T(tu->pool, pj_stun_auth_cred);
    pj_stun_auth_cred_dup(tu->pool, tu->cred, c);
    pj_stun_session_set_credential(tu->default_session, tu->cred);
    return PJ_SUCCESS;
}


/* 
 * This is a callback called by usage.c when the particular STUN usage
 * is to be destroyed.
 */
static void tu_on_destroy(pj_stun_usage *usage)
{
    struct turn_usage *tu;
    pj_hash_iterator_t hit, *it;

    tu = (struct turn_usage*) pj_stun_usage_get_user_data(usage);

    /* Destroy all clients */
    if (tu->client_htable) {
	it = pj_hash_first(tu->client_htable, &hit);
	while (it) {
	    struct turn_client *client;

	    client = (struct turn_client *)pj_hash_this(tu->client_htable, it);
	    client_destroy(client, PJ_SUCCESS);

	    it = pj_hash_first(tu->client_htable, &hit);
	}
    }

    pj_stun_session_destroy(tu->default_session);
    pj_mutex_destroy(tu->mutex);
    pj_pool_release(tu->pool);
}


/*
 * This is a callback called by the usage.c to notify the TURN usage, 
 * that incoming packet (may or may not be a STUN packet) is received
 * on the port where the TURN usage is listening.
 */
static void tu_on_rx_data(pj_stun_usage *usage,
			  void *pkt,
			  pj_size_t pkt_size,
			  const pj_sockaddr_t *src_addr,
			  unsigned src_addr_len)
{
    struct turn_usage *tu;
    struct turn_client *client;
    unsigned flags;
    pj_status_t status;

    /* Which usage instance is this */
    tu = (struct turn_usage*) pj_stun_usage_get_user_data(usage);

    /* Lookup client structure based on source address */
    client = (struct turn_client*) pj_hash_get(tu->client_htable, src_addr,
					       src_addr_len, NULL);

    /* STUN message decoding flag */
    flags = 0;
    if (tu->type == PJ_SOCK_DGRAM)
	flags |= PJ_STUN_IS_DATAGRAM;
    

    if (client) {
	status = pj_stun_msg_check((const pj_uint8_t*)pkt, pkt_size, flags);

	if (status == PJ_SUCCESS) {
	    /* Received STUN message */
	    status = pj_stun_session_on_rx_pkt(client->session, 
					       (pj_uint8_t*)pkt, pkt_size, 
					       flags, NULL, 
					       src_addr, src_addr_len);
	} else if (client->active_peer) {
	    /* Received non-STUN message and client has active destination */
	    pj_ssize_t sz = pkt_size;
	    pj_ioqueue_sendto(client->key, &client->pkt_write_key,
			      pkt, &sz, 0,
			      &client->active_peer->addr,
			      sizeof(client->active_peer->addr));
	} else {
	    /* Received non-STUN message and client doesn't have active 
	     * destination.
	     */
	    /* Ignore */
	}

    } else {
	/* Received packet (could be STUN or no) from new source */
	flags |= PJ_STUN_CHECK_PACKET;
	pj_stun_session_on_rx_pkt(tu->default_session, (pj_uint8_t*)pkt, 
				  pkt_size, flags, NULL, 
				  src_addr, src_addr_len);
    }
}


/*
 * This is a utility function provided by TU (Turn Usage) to reserve
 * or allocate internal port/socket. The allocation needs to be 
 * coordinated to minimize bind() collissions.
 */
static pj_status_t tu_alloc_port(struct turn_usage *tu, 
				 int type, 
				 unsigned rpp_bits, 
				 const pj_sockaddr_in *req_addr,
				 pj_sock_t *p_sock,
				 int *err_code)
{
    enum { RETRY = 100 };
    pj_sockaddr_in addr;
    pj_sock_t sock = PJ_INVALID_SOCKET;
    unsigned retry;
    pj_status_t status;

    if (req_addr && req_addr->sin_port != 0) {

	*err_code = PJ_STUN_SC_INVALID_PORT;

	/* Allocate specific port */
	status = pj_sock_socket(PJ_AF_INET, type, 0, &sock);
	if (status != PJ_SUCCESS)
	    return status;

	/* Bind */
	status = pj_sock_bind(sock, req_addr, sizeof(pj_sockaddr_in));
	if (status != PJ_SUCCESS) {
	    pj_sock_close(sock);
	    return status;
	}

	/* Success */
	*p_sock = sock;
	return PJ_SUCCESS;

    } else {
	status = -1;
	*err_code = PJ_STUN_SC_INSUFFICIENT_CAPACITY;

	if (req_addr && req_addr->sin_addr.s_addr) {
	    *err_code = PJ_STUN_SC_INVALID_IP_ADDR;
	    pj_memcpy(&addr, req_addr, sizeof(pj_sockaddr_in));
	} else {
	    pj_sockaddr_in_init(&addr, NULL, 0);
	}

	for (retry=0; retry<RETRY && sock == PJ_INVALID_SOCKET; ++retry) {
	    switch (rpp_bits) {
	    case 1:
		if ((tu->next_port & 0x01)==0)
		    tu->next_port++;
		break;
	    case 2:
	    case 3:
		if ((tu->next_port & 0x01)==1)
		    tu->next_port++;
		break;
	    }

	    status = pj_sock_socket(PJ_AF_INET, type, 0, &sock);
	    if (status != PJ_SUCCESS)
		return status;

	    addr.sin_port = pj_htons((pj_uint16_t)tu->next_port);

	    if (++tu->next_port > END_PORT)
		tu->next_port = START_PORT;

	    status = pj_sock_bind(sock, &addr, sizeof(addr));
	    if (status != PJ_SUCCESS) {
		pj_sock_close(sock);
		sock = PJ_INVALID_SOCKET;

		/* If client requested specific IP address, assume that
		 * bind failed because the IP address is not valid. We
		 * don't want to retry that since it will exhaust our
		 * port space.
		 */
		if (req_addr && req_addr->sin_addr.s_addr)
		    break;
	    }
	}

	if (sock == PJ_INVALID_SOCKET) {
	    return status;
	}

	*p_sock = sock;
	return PJ_SUCCESS;
    }
}


/* 
 * This callback is called by the TU's STUN session when it receives
 * a valid STUN message. This is called from tu_on_rx_data above.
 */
static pj_status_t tu_sess_on_rx_request(pj_stun_session *sess,
					 const pj_uint8_t *pkt,
					 unsigned pkt_len,
					 const pj_stun_msg *msg,
					 const pj_sockaddr_t *src_addr,
					 unsigned src_addr_len)
{
    struct session_data *sd;
    struct turn_client *client;
    pj_stun_tx_data *tdata;
    pj_status_t status;

    PJ_UNUSED_ARG(pkt);
    PJ_UNUSED_ARG(pkt_len);

    sd = (struct session_data*) pj_stun_session_get_user_data(sess);

    pj_assert(sd->client == NULL);

    if (msg->hdr.type == PJ_STUN_BINDING_REQUEST) {
	return handle_binding_req(sess, msg, src_addr, src_addr_len);

    } else if (msg->hdr.type != PJ_STUN_ALLOCATE_REQUEST) {
	if (PJ_STUN_IS_REQUEST(msg->hdr.type)) {
	    status = pj_stun_session_create_res(sess, msg, 
						PJ_STUN_SC_NO_BINDING,
						NULL, &tdata);
	    if (status==PJ_SUCCESS) {
		status = pj_stun_session_send_msg(sess, PJ_FALSE, 
						  src_addr, src_addr_len,
						  tdata);
	    }
	} else {
	    PJ_LOG(4,(THIS_FILE, 
		      "Received %s %s without matching Allocation, "
		      "ignored", pj_stun_get_method_name(msg->hdr.type),
		      pj_stun_get_class_name(msg->hdr.type)));
	}
	return PJ_SUCCESS;
    }