rtpproxy_relay.c

Siproxd is a proxy/masquerading daemon for the SIP protocol. It handles registrations of SIP client

/*
    Copyright (C) 2003-2005  Thomas Ries <[e-mail]>

    This file is part of Siproxd.
    
    Siproxd 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.
    
    Siproxd is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warrantry 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 Siproxd; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
*/

#include "config.h"

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/time.h>

#include <sys/socket.h>
#include <netinet/in.h>
#include <signal.h>

#ifdef HAVE_PTHREAD_SETSCHEDPARAM
   #include <sched.h>
#endif

#include <osipparser2/osip_parser.h>

#include "siproxd.h"
#include "rtpproxy.h"
#include "log.h"

#if !defined(SOL_IP)
#define SOL_IP IPPROTO_IP
#endif

static char const ident[]="$Id: rtpproxy_relay.c,v 1.41 2006/06/11 20:29:34 hb9xar Exp $";

/* configuration storage */
extern struct siproxd_config configuration;

/*
 * table to remember all active rtp proxy streams
 */
rtp_proxytable_t rtp_proxytable[RTPPROXY_SIZE];

/*
 * Mutex for thread synchronization (locking when accessing common 
 * data structures -> rtp_proxytable[]).
 *
 * use a 'fast' mutex for synchronizing - as these are portable... 
 */
static pthread_mutex_t rtp_proxytable_mutex = PTHREAD_MUTEX_INITIALIZER;

/* thread id of RTP proxy */
static pthread_t rtpproxy_tid=0;

/* master fd_set */
static fd_set master_fdset;
static int    master_fd_max;

/* forward declarations */
static void *rtpproxy_main(void *i);
static int rtp_recreate_fdset(void);
void rtpproxy_kill( void );
static void sighdl_alm(int sig) {/* just wake up from select() */};


/*
 * initialize and create rtp_relay proxy thread
 *
 * RETURNS
 *	STS_SUCCESS on success
 */
int rtp_relay_init( void ) {
   int sts;
   int arg=0;
   struct sigaction sigact;

   atexit(rtpproxy_kill);  /* cancel RTP thread at exit */

   /* clean proxy table */
   memset (rtp_proxytable, 0, sizeof(rtp_proxytable));

   /* initialize fd set for RTP proxy thread */
   FD_ZERO(&master_fdset); /* start with an empty fdset */
   master_fd_max=-1;

   /* install signal handler for SIGALRM - used to wake up
      the rtpproxy thread from select() hibernation */
   sigact.sa_handler = sighdl_alm;
   sigemptyset(&sigact.sa_mask);
   sigact.sa_flags=0;
   sigaction(SIGALRM, &sigact, NULL);

   DEBUGC(DBCLASS_RTP,"create thread");
   sts=pthread_create(&rtpproxy_tid, NULL, rtpproxy_main, (void *)&arg);
   DEBUGC(DBCLASS_RTP,"created, sts=%i", sts);

   /* set realtime scheduling - if started by root */
#ifdef HAVE_PTHREAD_SETSCHEDPARAM
   {
      int uid,euid;
      struct sched_param schedparam;

#ifndef _CYGWIN
      uid=getuid();
      euid=geteuid();
      DEBUGC(DBCLASS_RTP,"uid=%i, euid=%i", uid, euid);
      if (uid != euid) seteuid(0);

      if (geteuid()==0) {
#endif

#if defined(HAVE_SCHED_GET_PRIORITY_MAX) && defined(HAVE_SCHED_GET_PRIORITY_MIN)
         int pmin, pmax;
         /* place ourself at 1/3 of the available priority space */
         pmin=sched_get_priority_min(SCHED_RR);
         pmax=sched_get_priority_max(SCHED_RR);
         schedparam.sched_priority=pmin+(pmax-pmin)/3;
         DEBUGC(DBCLASS_RTP,"pmin=%i, pmax=%i, using p=%i", pmin, pmax,
                schedparam.sched_priority);
#else
         /* just taken a number out of thin air */
         schedparam.sched_priority=10;
         DEBUGC(DBCLASS_RTP,"using p=%i", schedparam.sched_priority);
#endif
         sts=pthread_setschedparam(rtpproxy_tid, SCHED_RR, &schedparam);
         if (sts != 0) {
            ERROR("pthread_setschedparam failed: %s", strerror(errno));
         }
#ifndef _CYGWIN
      } else {
         INFO("Unable to use realtime scheduling for RTP proxy");
         INFO("You may want to start siproxd as root and switch UID afterwards");
      }
      if (uid != euid)  seteuid(euid);
#endif
   }
#endif

   return STS_SUCCESS;
}


/*
 * main() of rtpproxy
 */
static void *rtpproxy_main(void *arg) {
   struct timeval tv;
   fd_set fdset;
   int fd_max;
   time_t t, last_t=0;
   int i, sts;
   int num_fd;
   osip_call_id_t callid;
   static char rtp_buff[RTP_BUFFER_SIZE];
   int count;

   memcpy(&fdset, &master_fdset, sizeof(fdset));
   fd_max=master_fd_max;

   /* loop forever... */
   for (;;) {

      tv.tv_sec = 5;
      tv.tv_usec = 0;

      num_fd=select(fd_max+1, &fdset, NULL, NULL, &tv);
      pthread_testcancel();
      if ((num_fd<0) && (errno==EINTR)) {
         /*
          * wakeup due to a change in the proxy table:
          * lock mutex, copy master FD set and unlock
          */
         pthread_mutex_lock(&rtp_proxytable_mutex);
         memcpy(&fdset, &master_fdset, sizeof(fdset));
         fd_max=master_fd_max;
         pthread_mutex_unlock(&rtp_proxytable_mutex);
         continue;
      }

      time(&t);

      /*
       * LOCK the MUTEX
       */
      pthread_mutex_lock(&rtp_proxytable_mutex);

      /* check for data available and send to destination */
      for (i=0;(i<RTPPROXY_SIZE) && (num_fd>0);i++) {
         if ( (rtp_proxytable[i].rtp_rx_sock != 0) && 
            FD_ISSET(rtp_proxytable[i].rtp_rx_sock, &fdset) ) {
            /* yup, have some data to send */
            num_fd--;

	    /* read from sock rtp_proxytable[i].sock*/
            count=read(rtp_proxytable[i].rtp_rx_sock, rtp_buff, RTP_BUFFER_SIZE);

            /* check if something went banana */
            if (count < 0) {
               /*
                * It has been seen on linux 2.2.x systems that for some
                * reason (ICMP issue? -> below) inside the RTP relay, select()
                * claims that a certain file descriptor has data available to
                * read, a subsequent call to read() or recv() then does block!!
                * So lets make the FD's we are going to use non-blocking, so
                * we will at least survive and not run into a deadlock.
                * 
                * We catch this here with this workaround (pronounce "HACK")
                * and hope that next time we pass by it will be ok again.
                */
               if (errno == EAGAIN) {
                  /* I may want to remove this WARNing */
                  WARN("read() [fd=%i, %s:%i] would block, but select() "
                       "claimed to be readable!",
                       rtp_proxytable[i].rtp_rx_sock,
                       utils_inet_ntoa(rtp_proxytable[i].local_ipaddr),
                       rtp_proxytable[i].local_port);
                  continue;
               }

               /*
                * I *MAY* receive ICMP destination unreachable messages when I
                * try to send RTP traffic to a destination that is in HOLD
                * (better: is not listening on the UDP port where I send
                * my RTP data to).
                * So I should *not* do this - or ignore errors originating
                * by this -> ECONNREFUSED
                *
                * Note: This error is originating from a previous send() on the
                *       same socket and has nothing to do with the read() we have
                *       done above!
                */
               if (errno != ECONNREFUSED) {
                  /* some other error that I probably want to know about */
                  int j;
                  WARN("read() [fd=%i, %s:%i] returned error [%i:%s]",
                  rtp_proxytable[i].rtp_rx_sock,
                  utils_inet_ntoa(rtp_proxytable[i].local_ipaddr),
                  rtp_proxytable[i].local_port, errno, strerror(errno));
                  for (j=0; j<RTPPROXY_SIZE;j++) {
                     DEBUGC(DBCLASS_RTP, "%i - rx:%i tx:%i %s@%s dir:%i "
                            "lp:%i, rp:%i rip:%s",
                            j,
                            rtp_proxytable[j].rtp_rx_sock,
                            rtp_proxytable[j].rtp_tx_sock,
                            rtp_proxytable[j].callid_number,
                            rtp_proxytable[j].callid_host,
                            rtp_proxytable[j].direction,
                            rtp_proxytable[j].local_port,
                            rtp_proxytable[j].remote_port,
                            utils_inet_ntoa(rtp_proxytable[j].remote_ipaddr));
                  } /* for j */
              } /* if errno != ECONNREFUSED */
            } /* count < 0 */

            /*
             * forwarding an RTP packet only makes sense if we really
             * have got some data in it (count > 0)
             */
            if (count > 0) {
               /* find the corresponding TX socket */
               if (rtp_proxytable[i].rtp_tx_sock == 0) {
                  int j;
                  int rtp_direction = rtp_proxytable[i].direction;
                  int media_stream_no = rtp_proxytable[i].media_stream_no;

                  callid.number = rtp_proxytable[i].callid_number;
                  callid.host = rtp_proxytable[i].callid_host;

                  for (j=0;(j<RTPPROXY_SIZE);j++) {
                     char *client_id = rtp_proxytable[i].client_id;
                     osip_call_id_t cid;
                     cid.number = rtp_proxytable[j].callid_number;
                     cid.host = rtp_proxytable[j].callid_host;

                     /* match on:
                      * - same call ID
                      * - same media stream
                      * - opposite direction
                      * - different client ID
                      */
                     if ( (rtp_proxytable[j].rtp_rx_sock != 0) &&
                          (compare_callid(&callid, &cid) == STS_SUCCESS) &&
                          (media_stream_no == rtp_proxytable[j].media_stream_no) &&
                          (rtp_direction != rtp_proxytable[j].direction) &&
                          (strcmp(rtp_proxytable[j].client_id, client_id) != 0) ) {
                        rtp_proxytable[i].rtp_tx_sock = rtp_proxytable[j].rtp_rx_sock;
                        rtp_proxytable[i].opposite_entry=j+1;
                        DEBUGC(DBCLASS_RTP, "connected entry %i (fd=%i) <-> entry %i (fd=%i)",
                               j, rtp_proxytable[j].rtp_rx_sock,
                               i, rtp_proxytable[i].rtp_rx_sock);
                        break;
                     }
                  }
               } /* rtp_tx_sock == 0 */

               if (rtp_proxytable[i].rtp_tx_sock != 0) {
                  /* write to dest via socket rtp_tx_sock */
                  struct sockaddr_in dst_addr;
                  dst_addr.sin_family = AF_INET;
                  memcpy(&dst_addr.sin_addr.s_addr,
                         &rtp_proxytable[i].remote_ipaddr, 
                         sizeof(struct in_addr));
                  dst_addr.sin_port= htons(rtp_proxytable[i].remote_port);

                  sts = sendto(rtp_proxytable[i].rtp_tx_sock, rtp_buff,
                               count, 0, (const struct sockaddr *)&dst_addr,
                               (socklen_t)sizeof(dst_addr));
                  if (sts == -1) {
                     if (errno != ECONNREFUSED) {
                        ERROR("sendto() [%s:%i size=%i] call failed: %s",
                        utils_inet_ntoa(rtp_proxytable[i].remote_ipaddr),
                        rtp_proxytable[i].remote_port, count, strerror(errno));

                    /* if sendto() fails with bad filedescriptor,
                     * this means that the opposite stream has been
                     * canceled or timed out.
                     * we should then cancel this stream as well.*/

                    WARN("stopping opposite stream");
                    /* don't lock the mutex, as we own the lock */
                    callid.number=rtp_proxytable[i].callid_number;
                    callid.host=rtp_proxytable[i].callid_host;
                    rtp_relay_stop_fwd(&callid, rtp_proxytable[i].direction, -1, 1);
                     }
                  }
               }
            } /* count > 0 */

            /* update timestamp of last usage for both (RX and TX) entries.
             * This allows silence (no data) on one stream without breaking
             * the connection after the RTP timeout */
            rtp_proxytable[i].timestamp=t;
            if (rtp_proxytable[i].opposite_entry > 0) {
               rtp_proxytable[rtp_proxytable[i].opposite_entry-1].timestamp=t;
            }
         }
      } /* for i */

      /*
       * age and clean rtp_proxytable (check every 10 seconds)
       */
      if (t > (last_t+10) ) {
         last_t = t;
	 for (i=0;i<RTPPROXY_SIZE; i++) {
            if ( (rtp_proxytable[i].rtp_rx_sock != 0) &&
		 ((rtp_proxytable[i].timestamp+configuration.rtp_timeout)<t)) {
               /* this one has expired, clean it up */
               callid.number=rtp_proxytable[i].callid_number;
               callid.host=rtp_proxytable[i].callid_host;
               INFO("RTP stream %s@%s (media=%i) has expired",
                    callid.number, callid.host,
                    rtp_proxytable[i].media_stream_no);
               DEBUGC(DBCLASS_RTP,"RTP stream rx_sock=%i tx_sock=%i "
                      "%s@%s (idx=%i) has expired",
                      rtp_proxytable[i].rtp_rx_sock,
                      rtp_proxytable[i].rtp_tx_sock,
                      callid.number, callid.host, i);
               /* Don't lock the mutex, as we own the lock already here */
               /* Only stop the stream we caught is timeout and not everything.
                * This may be a multiple stream conversation (audio/video) and
                * just one (unused?) has timed out. Seen with VoIPEX PBX! */
               rtp_relay_stop_fwd(&callid, rtp_proxytable[i].direction,
                                  rtp_proxytable[i].media_stream_no, 1);
	    }
	 }
      } /* if (t>...) */

      /* copy master FD set */
      memcpy(&fdset, &master_fdset, sizeof(fdset));
      fd_max=master_fd_max;

      /*
       * UNLOCK the MUTEX
       */
      pthread_mutex_unlock(&rtp_proxytable_mutex);
   } /* for(;;) */

   return NULL;
}


/*
 * start an rtp stream on the proxy
 *
 * RETURNS
 *	STS_SUCCESS on success
 *	STS_FAILURE on error
 */
int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id,
                         int rtp_direction,
                         int media_stream_no, struct in_addr local_ipaddr,
                         int *local_port, struct in_addr remote_ipaddr,
                         int remote_port) {
   static int prev_used_port = 0;
   int num_ports;
   int i2, i, j;
   int sock, port;
   int freeidx;
   int sts=STS_SUCCESS;
   int tos;
   osip_call_id_t cid;
   

   if (callid == NULL) {
      ERROR("rtp_relay_start_fwd: callid is NULL!");
      return STS_FAILURE;
   }

   if (client_id == NULL) {
      ERROR("rtp_relay_start_fwd: did not get a client ID!");
      return STS_FAILURE;
   }

   /*