tport.c

Internet Phone, Chat, Conferencing

/*
 * This file is part of the Sofia-SIP package
 *
 * Copyright (C) 2005 Nokia Corporation.
 *
 * Contact: Pekka Pessi <[e-mail]>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public License
 * as published by the Free Software Foundation; either version 2.1 of
 * the License, or (at your option) any later version.
 *
 * This library 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 * 02110-1301 USA
 *
 */

/**@CFILE tport.c Transport interface implementation.
 *
 * See tport.docs for more detailed description of tport interface.
 *
 * @author Pekka Pessi <[e-mail]>
 * @author Ismo Puustinen <[e-mail]>
 * @author Tat Chan <[e-mail]>
 * @author Kai Vehmanen <[e-mail]>
 * @author Martti Mela <[e-mail]>
 *
 * @date Created: Thu Jul 20 12:54:32 2000 ppessi
 */

#include "config.h"

#include <sofia-sip/string0.h>
#include <stdlib.h>
#include <time.h>
#include <assert.h>
#include <errno.h>
#include <limits.h>

#include <sofia-sip/su.h>

#if HAVE_NETINET_TCP_H
#include <netinet/tcp.h>
#endif

#if !defined(EPROTONOSUPPORT) && defined(_WIN32)
#define EPROTONOSUPPORT WSAEPROTONOSUPPORT
#endif

#if !defined(ENOBUFS) && defined(_WIN32)
#define ENOBUFS WSAENOBUFS
#endif

#ifndef EBADMSG
#define EBADMSG 117 /* XXX -- mela: number? */
#endif

#if !defined(EMSGSIZE) && defined(_WIN32)
#define EMSGSIZE WSAEMSGSIZE
#endif

#if !defined(random) && defined(_WIN32)
#define random rand
#endif

#ifndef SU_DEBUG
#define SU_DEBUG 3
#endif

#define SU_LOG   tport_log

#include <sofia-sip/su_debug.h>
#include <sofia-sip/su_alloc.h>
#include <sofia-sip/su_tagarg.h>

#include <sofia-sip/su_localinfo.h>

#ifndef NONE
#define NONE ((void *)-1)
#endif

typedef struct tport_master tport_master_t;
typedef struct tport_nat_s tport_nat_t;

#define SU_WAKEUP_ARG_T struct tport_s
#define SU_TIMER_ARG_T  struct tport_master
#define SU_ROOT_MAGIC_T struct tport_threadpool
#define SU_MSG_ARG_T    union tport_su_msg_arg
#define STUN_MAGIC_T    tport_master_t


#include <sofia-sip/su_wait.h>

#include <sofia-sip/msg.h>
#include <sofia-sip/msg_addr.h>

#if HAVE_IP_RECVERR || HAVE_IPV6_RECVERR
#include <linux/types.h>
#include <linux/errqueue.h>
#include <sys/uio.h>
#endif

#if HAVE_SCTP && HAVE_NETINET_SCTP_H
#include <netinet/sctp.h>
#endif

#if HAVE_TLS
#include "tport_tls.h"
#else
typedef struct _tls_t tls_t;	/* dummy */
#endif

#if HAVE_SIGCOMP
#include <sigcomp.h>
#endif

#ifndef SOL_TCP
#define SOL_TCP IPPROTO_TCP
#endif

#ifndef IPPROTO_SCTP
#define IPPROTO_SCTP (132)
#endif

#if !defined(MSG_NOSIGNAL) || defined(__CYGWIN__)
#undef MSG_NOSIGNAL
#define MSG_NOSIGNAL (0)
#endif

#if HAVE_SOFIA_STUN
#include "sofia-sip/stun.h"
#include "sofia-sip/stun_tag.h"
#endif

#if HAVE_UPNP
#include "upnp_wrapper.h"
#endif

#include "sofia-sip/tport.h"
#include "sofia-sip/su_uniqueid.h"
#include <sofia-sip/rbtree.h>

#if HAVE_FUNC
#elif HAVE_FUNCTION
#define __func__ __FUNCTION__
#else
char const __func__[] = "tport";
#endif

typedef struct tport_pending_s tport_pending_t;
typedef struct tport_threadpool tport_threadpool_t;
typedef struct tport_sigcomp_handler tport_sigcomp_handler_t;
typedef struct tport_sigcomp tport_sigcomp_t;
typedef struct tport_primary tport_primary_t;

struct sigcomp_state_handler;
struct sigcomp_algorithm;
struct sigcomp_udvm;
struct sigcomp_magic;

#if HAVE_SIGCOMP

/** Per-socket SigComp data */
struct tport_sigcomp {
  struct sigcomp_udvm          *sc_udvm;
  struct sigcomp_compartment   *sc_cc;
  struct sigcomp_compressor    *sc_compressor;
  struct sigcomp_buffer        *sc_output;
  unsigned                      sc_compressed; 

  struct sigcomp_buffer        *sc_input;
  unsigned                      sc_copied;
  
  enum {
    format_is_unknown,
    format_is_sigcomp,
    format_is_noncomp
  } sc_infmt, sc_outfmt;
};

#endif

/** Transport parameters */
typedef struct {
  unsigned tpp_mtu;		/**< Maximum packet size */
  unsigned tpp_idle;		/**< Allowed connection idle time. */
  unsigned tpp_timeout;		/**< Allowed idle time for message. */
  unsigned tpp_sigcomp_lifetime;  /**< SigComp compartment lifetime  */
  unsigned tpp_thrpsize;	/**< Size of thread pool */

  unsigned tpp_thrprqsize;	/**< Length of per-thread recv queue */
  unsigned tpp_qsize;		/**< Size of queue */

  unsigned tpp_drop;		/**< Packet drop probablity */

  unsigned tpp_conn_orient:1;   /**< Connection-orienteded */
  unsigned tpp_sdwn_error:1;	/**< If true, shutdown is error. */
  unsigned :0;

} tport_params_t;


/** Transport object.
 *
 * A transport object can be used in three roles, to represent transport
 * list (aka master transport), to represent available transports (aka
 * primary transport) and to represent actual transport connections (aka
 * secondary transport).
 */
struct tport_s {
  su_home_t           tp_home[1];       /**< Memory home */

  int                 tp_refs;		/**< Number of references to tport */

  tport_t *tp_left, *tp_right, *tp_dad; /**< Links in tport tree */

  tport_master_t     *tp_master;        /**< Master transport */
  tport_primary_t    *tp_pri;           /**< Primary transport */

  tport_params_t     *tp_params;        /**< Transport parameters */

  tp_magic_t         *tp_magic; 	/**< Context provided by consumer */

  unsigned            tp_black:1;       /**< Used by red-black-tree */

  unsigned            tp_conn_orient:1;	/**< Is connection-oriented */
  unsigned            tp_connected : 1;	/**< Has real connection */
  unsigned            tp_reusable:1;    /**< Can this connection be reused */
  unsigned            tp_closed : 1;    /**< This transport is closed */
  /**< Remote end has sent FIN (2) or we should not just read */
  unsigned            tp_recv_close:2;
  /** We will send FIN (1) or have sent FIN (2) */
  unsigned            tp_send_close:2; 
  unsigned            tp_has_keepalive:1;

  unsigned:0;

  msg_t const        *tp_rlogged;       /**< Last logged when receiving */
  msg_t const        *tp_slogged;       /**< Last logged when sending */

  unsigned            tp_time;	        /**< When this transport was last used */

  tp_name_t           tp_name[1];	/**< Transport name.
					 * 
					 * This is either our name (if primary)
					 * or peer name (if secondary).
					 */

#define tp_protoname tp_name->tpn_proto
#define tp_canon     tp_name->tpn_canon
#define tp_host      tp_name->tpn_host
#define tp_port      tp_name->tpn_port
#define tp_ident     tp_name->tpn_ident

  su_socket_t  	      tp_socket;	/**< Socket of this tport*/
  int                 tp_index;		/**< Root registration index */
  int                 tp_events;        /**< Subscribed events */
  int                 tp_addrlen;       /**< Size fo tp_addr */
  su_sockaddr_t       tp_addr[1];	/**< Peer address */

#if HAVE_TLS
  tls_t              *tp_tls;
  char               *tp_tls_buffer;    /**< 2k Buffer  */
#endif

#if HAVE_SIGCOMP
  tport_sigcomp_t     tp_sigcomp[1];
#endif  

#if HAVE_SOFIA_STUN
#if 0
  stun_socket_t      *tp_stun_socket;
#endif
  su_socket_t         tp_stun_socket;
#endif

  /* ==== Receive queue ================================================== */

  msg_t   	     *tp_msg;		/**< Message being received */

  /* ==== Pending messages =============================================== */

  tport_pending_t    *tp_pending;       /**< Pending requests */
  tport_pending_t    *tp_released;      /**< Released pends */
  unsigned            tp_plen;          /**< Size of tp_pending */
  unsigned            tp_pused;         /**< Used pends */
  unsigned            tp_reported;      /**< Report counter */

  /* ==== Send queue ===================================================== */

  msg_t             **tp_queue;		/**< Messages being sent */
  unsigned short      tp_qhead;		/**< Head of queue */

  msg_iovec_t        *tp_unsent;	/**< Pointer to first unsent iovec */
  unsigned            tp_unsentlen;	/**< Number of unsent iovecs */

  msg_iovec_t        *tp_iov;		/**< Iovecs allocated for sending */
  unsigned            tp_iovlen;	/**< Number of allocated iovecs */

  /* ==== Statistics  ===================================================== */
  
  struct {
    uint64_t sent_bytes, sent_on_line, recv_bytes, recv_on_line;
    uint64_t sent_msgs, recv_msgs;
  } tp_stats;
};

/** Primary structure */
struct tport_primary {
  tport_t             pri_primary[1];   /**< Transport part */
#if DOX
  su_home_t           pri_home[1];
  tport_t           **pri_stail;	/**< End of secondary list */
#else
#define pri_home      pri_primary->tp_home
#define pri_master    pri_primary->tp_master
#define pri_protoname pri_primary->tp_name->tpn_proto
#endif

  char                pri_ident[16];
  tport_primary_t    *pri_next;	        /**< Next primary tport */

  tport_t            *pri_secondary;	/**< Secondary tports */

  tport_threadpool_t *pri_threadpool;   /**< Worker threads */
  unsigned            pri_thrpsize;
  unsigned            pri_family : 8;   /**< Network family (INET/INET6) */
  unsigned            pri_socktype : 8; /**< Socket type  */
  unsigned            pri_protocol : 8;	/**< IP protocol number  */

  unsigned            pri_natted:1;	/**< Using natted address  */

  unsigned:0;

  tport_params_t      pri_params[1];      /**< Transport parameters */

};

/** Master structure */
struct tport_master {
  tport_t             mr_master[1];
#if DOX
  su_home_t           mr_home[1];
#else
#define mr_home mr_master->tp_home
#endif

  int                 mr_stun_step_ready; /**< for stun's callback */

  tp_stack_t  	     *mr_stack;         /**< Transport consumer */
  tp_stack_class_t 
               const *mr_tpac;		/**< Methods provided by stack */
  int                 mr_log;	        /**< Do logging of parsed messages */
  su_root_t    	     *mr_root;		/**< SU root pointer */

  /**< Timer reclaiming unused connections and compartment */
  su_timer_t         *mr_timer;		
  /** File to dump received and sent data */
  FILE               *mr_dump_file;	

  tport_primary_t    *mr_primaries;     /**< List of primary contacts */

  tport_params_t      mr_params[1];
  
  unsigned            mr_boundserver:1; /**< Server has been bound */
  unsigned            mr_bindv6only:1; /**< We can bind separately to IPv6/4 */
  unsigned :0;

#if HAVE_SIGCOMP
  struct sigcomp_compartment   *mr_compartment;
#endif

  /* Delivery context */
  struct tport_delivery {
    tport_t              *d_tport;
    msg_t                *d_msg;
    tp_name_t             d_from[1];
    struct sigcomp_udvm **d_udvm;
  } mr_delivery[1];

  struct tport_nat_s {
    int initialized;
    int bound;
    int stun_enabled;
    char *external_ip_address;
#if HAVE_UPNP || HAVE_SOFIA_STUN
    int try_stun;
#endif
#if HAVE_UPNP
#endif
#if HAVE_SOFIA_STUN
    tport_master_t *tport;
    char *stun_server;
    /* stun_socket_t *stun_socket; */
    stun_handle_t *stun;
    su_socket_t stun_socket;
    su_sockaddr_t sockaddr;
#endif
  }                   mr_nat[1];
};

#define STACK_RECV(tp, msg, now)		       \
  (tp)->tp_master->mr_tpac->tpac_recv((tp)->tp_master->mr_stack, (tp), \
				      (msg), (tp)->tp_magic, (now))

#define STACK_ERROR(tp, errcode, dstname) \
  (tp)->tp_master->mr_tpac->tpac_error((tp)->tp_master->mr_stack, (tp), \
				       (errcode), (dstname))

#define STACK_SIGCOMP_ACCEPT(tp, msg)				  \
  (tp)->tp_master->mr_tpac->					  \
  tpac_sigcomp_accept((tp)->tp_master->mr_stack, (tp), (msg))

#define TP_STACK   tp_master->mr_stack

#define TP_SCTP_MSG_MAX (32768)

typedef long unsigned LU; 	/* for printf() and friends */

char const tport_sigcomp_name[] = "sigcomp";

#if HAVE_SIGCOMP
static inline
int msg_is_compressed(msg_t *msg)
{
  return msg && 
    (msg_addrinfo(msg)->ai_flags & TP_AI_COMPRESSED) == TP_AI_COMPRESSED;
}

static inline
void msg_mark_as_compressed(msg_t *msg)
{
  if (msg)
    msg_addrinfo(msg)->ai_flags |= TP_AI_COMPRESSED;
}

static struct sigcomp_udvm *tport_init_udvm(tport_t *self);
static int tport_recv_sigcomp_r(tport_t *, msg_t **, struct sigcomp_udvm *);
static struct sigcomp_compartment *tport_primary_compartment(tport_master_t *);
static inline void tport_try_accept_sigcomp(tport_t *self, msg_t *msg);

static
char const *tport_canonize_comp(char const *comp)
{
  if (comp && strcasecmp(comp, tport_sigcomp_name) == 0)
    return tport_sigcomp_name;
  return NULL;
}

/** Check if transport can receive compressed messages */
int tport_can_recv_sigcomp(tport_t const *self)
{
  return self && self->tp_sigcomp->sc_infmt != format_is_noncomp;
}

/** Check if transport can send compressed messages */
int tport_can_send_sigcomp(tport_t const *self)
{
  return self && self->tp_sigcomp->sc_outfmt != format_is_noncomp;
}

/** Check if transport supports sigcomp */
int tport_has_compression(tport_t const *self, char const *comp)
{
  return
    self && comp && 
    self->tp_name->tpn_comp == tport_canonize_comp(comp);
}

/** Set/reset compression */
int tport_set_compression(tport_t *self, char const *comp)
{
  if (self == NULL)
    ;
  else if (comp == NULL) {
    if (self->tp_sigcomp->sc_outfmt != format_is_sigcomp) {
      self->tp_name->tpn_comp = NULL;
      return 0;
    }
  }
  else {
    comp = tport_canonize_comp(comp);

    if (comp && self->tp_sigcomp->sc_outfmt != format_is_noncomp) {
      self->tp_name->tpn_comp = comp;
      return 0;