rtpp_command.c

大名鼎鼎的 RTP Proxy 源代码 在OpenSER 中成熟应用的

/*
 * Copyright (c) 2004-2006 Maxim Sobolev <[e-mail]>
 * Copyright (c) 2006-2007 Sippy Software, Inc., http://www.sippysoft.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * $Id: rtpp_command.c,v 1.6 2008/04/03 23:05:41 sobomax Exp $
 *
 */

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/uio.h>
#include <netinet/in.h>
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <netdb.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "rtpp_command.h"
#include "rtpp_log.h"
#include "rtpp_record.h"
#include "rtpp_session.h"
#include "rtpp_util.h"

struct proto_cap proto_caps[] = {
    /*
     * The first entry must be basic protocol version and isn't shown
     * as extension on -v.
     */
    { "20040107", "Basic RTP proxy functionality" },
    { "20050322", "Support for multiple RTP streams and MOH" },
    { "20060704", "Support for extra parameter in the V command" },
    { "20071116", "Support for RTP re-packetization" },
    { "20071218", "Support for forking (copying) RTP stream" },
    { "20080403", "Support for RTP statistics querying" },
    { NULL, NULL }
};

static int create_twinlistener(struct cfg *, struct sockaddr *, int, int *);
static int create_listener(struct cfg *, struct sockaddr *, int, int *, int *);
static int handle_delete(struct cfg *, char *, char *, char *, int);
static void handle_noplay(struct cfg *, struct rtpp_session *, int);
static int handle_play(struct cfg *, struct rtpp_session *, int, char *, char *, int);
static void handle_record(struct cfg *, struct rtpp_session *, int, char *);
static void handle_query(struct cfg *, int, struct sockaddr_storage *,
  socklen_t, char *, struct rtpp_session *, int);

static int
create_twinlistener(struct cfg *cf, struct sockaddr *ia, int port, int *fds)
{
    struct sockaddr_storage iac;
    int rval, i, flags;

    fds[0] = fds[1] = -1;

    rval = -1;
    for (i = 0; i < 2; i++) {
	fds[i] = socket(ia->sa_family, SOCK_DGRAM, 0);
	if (fds[i] == -1) {
	    rtpp_log_ewrite(RTPP_LOG_ERR, cf->glog, "can't create %s socket",
	      (ia->sa_family == AF_INET) ? "IPv4" : "IPv6");
	    goto failure;
	}
	memcpy(&iac, ia, SA_LEN(ia));
	satosin(&iac)->sin_port = htons(port);
	if (bind(fds[i], sstosa(&iac), SA_LEN(ia)) != 0) {
	    if (errno != EADDRINUSE && errno != EACCES) {
		rtpp_log_ewrite(RTPP_LOG_ERR, cf->glog, "can't bind to the %s port %d",
		  (ia->sa_family == AF_INET) ? "IPv4" : "IPv6", port);
	    } else {
		rval = -2;
	    }
	    goto failure;
	}
	port++;
	if ((ia->sa_family == AF_INET) &&
	  (setsockopt(fds[i], IPPROTO_IP, IP_TOS, &cf->tos, sizeof(cf->tos)) == -1))
	    rtpp_log_ewrite(RTPP_LOG_ERR, cf->glog, "unable to set TOS to %d", cf->tos);
	flags = fcntl(fds[i], F_GETFL);
	fcntl(fds[i], F_SETFL, flags | O_NONBLOCK);
    }
    return 0;

failure:
    for (i = 0; i < 2; i++)
	if (fds[i] != -1) {
	    close(fds[i]);
	    fds[i] = -1;
	}
    return rval;
}

static int
create_listener(struct cfg *cf, struct sockaddr *ia,  int startport,
  int *port, int *fds)
{
    int i, init, rval;

    for (i = 0; i < 2; i++)
	fds[i] = -1;

    init = 0;
    if (startport < cf->port_min || startport > cf->port_max)
	startport = cf->port_min;
    for (*port = startport; *port != startport || init == 0; (*port) += 2) {
	init = 1;
	rval = create_twinlistener(cf, ia, *port, fds);
	if (rval != 0) {
	    if (rval == -1)
		break;
	    if (*port >= cf->port_max)
		*port = cf->port_min - 2;
	    continue;
	}
	return 0;
    }
    return -1;
}

static void
doreply(struct cfg *cf, int fd, char *buf, int len,
  struct sockaddr_storage *raddr, socklen_t rlen)
{

    buf[len] = '\0';
    rtpp_log_write(RTPP_LOG_DBUG, cf->glog, "sending reply \"%s\"", buf);
    if (cf->umode == 0) {
	write(fd, buf, len);
    } else {
	while (sendto(fd, buf, len, 0, sstosa(raddr),
	  rlen) == -1 && errno == ENOBUFS);
    }
}

static void
reply_number(struct cfg *cf, int fd, struct sockaddr_storage *raddr,
  socklen_t rlen, char *cookie, int number)
{
    int len;
    char buf[1024 * 8];

    if (cookie != NULL)
	len = sprintf(buf, "%s %d\n", cookie, number);
    else {
	len = sprintf(buf, "%d\n", number);
    }
    doreply(cf, fd, buf, len, raddr, rlen);
}

static void
reply_ok(struct cfg *cf, int fd, struct sockaddr_storage *raddr,
  socklen_t rlen, char *cookie)
{

    reply_number(cf, fd, raddr, rlen, cookie, 0);
}

static void
reply_port(struct cfg *cf, int fd, struct sockaddr_storage *raddr,
  socklen_t rlen, char *cookie, int lport, struct sockaddr **lia)
{
    int len;
    char buf[1024 * 8], *cp;

    cp = buf;
    len = 0;
    if (cookie != NULL) {
	len = sprintf(cp, "%s ", cookie);
	cp += len;
    }
    if (lia[0] == NULL || ishostnull(lia[0]))
	len += sprintf(cp, "%d\n", lport);
    else
	len += sprintf(cp, "%d %s%s\n", lport, addr2char(lia[0]),
	  (lia[0]->sa_family == AF_INET) ? "" : " 6");
    doreply(cf, fd, buf, len, raddr, rlen);
}

static void
reply_error(struct cfg *cf, int fd, struct sockaddr_storage *raddr,
  socklen_t rlen, char *cookie, int ecode)
{
    int len;
    char buf[1024 * 8];

    if (cookie != NULL)
	len = sprintf(buf, "%s E%d\n", cookie, ecode);
    else
	len = sprintf(buf, "E%d\n", ecode);
    doreply(cf, fd, buf, len, raddr, rlen);
}

static void
handle_nomem(struct cfg *cf, int fd, struct sockaddr_storage *raddr,
  socklen_t rlen, char *cookie, int ecode, struct sockaddr **ia, int *fds,
  struct rtpp_session *spa, struct rtpp_session *spb)
{
    int i;

    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "can't allocate memory");
    for (i = 0; i < 2; i++)
	if (ia[i] != NULL)
	    free(ia[i]);
    if (spa != NULL) {
	if (spa->call_id != NULL)
	    free(spa->call_id);
	free(spa);
    }
    if (spb != NULL)
	free(spb);
    for (i = 0; i < 2; i++)
	if (fds[i] != -1)
	    close(fds[i]);
    reply_error(cf, fd, raddr, rlen, cookie, ecode);
}

int
handle_command(struct cfg *cf, int controlfd)
{
    int len, argc, i, j, pidx, asymmetric;
    int external, pf, lidx, playcount, weak;
    int fds[2], lport, n;
    socklen_t rlen;
    char buf[1024 * 8];
    char *cp, *call_id, *from_tag, *to_tag, *addr, *port, *cookie;
    char *pname, *codecs, *recording_name;
    struct rtpp_session *spa, *spb;
    char **ap, *argv[10];
    const char *rname;
    struct sockaddr *ia[2], *lia[2];
    struct sockaddr_storage raddr;
    int requested_nsamples;
    enum {DELETE, RECORD, PLAY, NOPLAY, COPY, UPDATE, LOOKUP, QUERY} op;

    requested_nsamples = -1;
    ia[0] = ia[1] = NULL;
    spa = spb = NULL;
    lia[0] = lia[1] = cf->bindaddr[0];
    lidx = 1;
    fds[0] = fds[1] = -1;
    recording_name = NULL;

    if (cf->umode == 0) {
		for (;;) {
		    len = read(controlfd, buf, sizeof(buf) - 1);						//从控制通道获取控制指令
		    if (len != -1 || (errno != EAGAIN && errno != EINTR))
			break;
		    sched_yield();
		}
    } else {
		rlen = sizeof(raddr);
		len = recvfrom(controlfd, buf, sizeof(buf) - 1, 0,
		  sstosa(&raddr), &rlen);
    }
    if (len == -1) {
			if (errno != EAGAIN && errno != EINTR)
			    rtpp_log_ewrite(RTPP_LOG_ERR, cf->glog, "can't read from control socket");
		return -1;
    }
    buf[len] = '\0';

    rtpp_log_write(RTPP_LOG_DBUG, cf->glog, "received command \"%s\"", buf);

    cp = buf;																	//cp  控制指令
    argc = 0;
    memset(argv, 0, sizeof(argv));
    for (ap = argv; (*ap = strsep(&cp, "\r\n\t ")) != NULL;)					//控制指令采用\r\n\t 进行分割
		if (**ap != '\0') {														//放入argc，argv
		    argc++;
		    if (++ap >= &argv[10])
			break;
		}
	    cookie = NULL;
	    if (argc < 1 || (cf->umode != 0 && argc < 2)) {
		rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
		reply_error(cf, controlfd, &raddr, rlen, cookie, 0);
		return 0;
    }

    /* Stream communication mode doesn't use cookie */
    if (cf->umode != 0) {
	cookie = argv[0];															//argv[0] = Cookie
	for (i = 1; i < argc; i++)													//执行指令
	    argv[i - 1] = argv[i];
	argc--;
	argv[argc] = NULL;
    } else {
	cookie = NULL;
    }

    addr = port = NULL;
    switch (argv[0][0]) {
    case 'u':
    case 'U':
		op = UPDATE;
		rname = "update/create";
	break;

    case 'l':
    case 'L':
	op = LOOKUP;
	rname = "lookup";
	break;

    case 'd':
    case 'D':
	op = DELETE;
	rname = "delete";
	break;

    case 'p':
    case 'P':
		/* P callid pname codecs from_tag to_tag */
		op = PLAY;
		rname = "play";
		playcount = 1;
		pname = argv[2];
		codecs = argv[3];
		break;

    case 'r':
    case 'R':
		op = RECORD;
		rname = "record";
		break;

    case 'c':
    case 'C':
		op = COPY;
		rname = "copy";
		break;

    case 's':
    case 'S':
		op = NOPLAY;
		rname = "noplay";
		
		break;

    case 'v':
    case 'V':
		if (argv[0][1] == 'F' || argv[0][1] == 'f') {
		    int i, known;
		    /*
		     * Wait for protocol version datestamp and check whether we
		     * know it.
		     */
		    if (argc != 2 && argc != 3) {
			rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
			reply_error(cf, controlfd, &raddr, rlen, cookie, 2);
			return 0;
		    }
		    for (known = i = 0; proto_caps[i].pc_id != NULL; ++i) {
			if (!strcmp(argv[1], proto_caps[i].pc_id)) {
			    known = 1;
			    break;
			}
		    }
		    reply_number(cf, controlfd, &raddr, rlen, cookie, known);
		    return 0;
		}
		if (argc != 1 && argc != 2) {
		    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
		    reply_error(cf, controlfd, &raddr, rlen, cookie, 2);
		    return 0;
		}
		/* This returns base version. */
		reply_number(cf, controlfd, &raddr, rlen, cookie, CPROTOVER);
		return 0;

    case 'i':
    case 'I':														//列出所有的sessions
		if (cookie == NULL)
		    len = sprintf(buf, "sessions created: %llu\nactive sessions: %d\n"
		      "active streams: %d\n", cf->sessions_created,
		      cf->sessions_active, cf->nsessions / 2);
		else
		    len = sprintf(buf, "%s sessions created: %llu\nactive sessions: %d\n"
		      "active streams: %d\n", cookie, cf->sessions_created,
		      cf->sessions_active, cf->nsessions / 2);
		
		for (i = 1; i < cf->nsessions; i++) {
		    char addrs[4][256];

		    spa = cf->sessions[i];
		    if (spa == NULL || spa->sidx[0] != i)
			continue;
		    /* RTCP twin session */
		    if (spa->rtcp == NULL) {
				spb = spa->rtp;
				buf[len++] = '\t';
		    } else {
				spb = spa->rtcp;
				buf[len++] = '\t';
				buf[len++] = 'C';
				buf[len++] = ' ';
		    }

		    addr2char_r(spb->laddr[1], addrs[0], sizeof(addrs[0]));
			
		    if (spb->addr[1] == NULL) {
				strcpy(addrs[1], "NONE");
		    } else {
				sprintf(addrs[1], "%s:%d", addr2char(spb->addr[1]),
			  	addr2port(spb->addr[1]));
		    }
			
		    addr2char_r(spb->laddr[0], addrs[2], sizeof(addrs[2]));
			
		    if (spb->addr[0] == NULL) {
				strcpy(addrs[3], "NONE");
		    } else {
				sprintf(addrs[3], "%s:%d", addr2char(spb->addr[0]),
			  	addr2port(spb->addr[0]));
		    }

		    len += sprintf(buf + len,
		      "%s/%s: caller = %s:%d/%s, callee = %s:%d/%s, "
		      "stats = %lu/%lu/%lu/%lu, ttl = %d\n",
		      spb->call_id, spb->tag, addrs[0], spb->ports[1], addrs[1],
		      addrs[2], spb->ports[0], addrs[3], spa->pcount[0], spa->pcount[1],
		      spa->pcount[2], spa->pcount[3], spb->ttl);
		    if (len + 512 > sizeof(buf)) {
			doreply(cf, controlfd, buf, len, &raddr, rlen);
			len = 0;
		    }
			
		}
		if (len > 0)
		    doreply(cf, controlfd, buf, len, &raddr, rlen);;
		return 0;
	break;

    case 'q':
    case 'Q':
	op = QUERY;
	rname = "query";
	break;

    default:
	rtpp_log_write(RTPP_LOG_ERR, cf->glog, "unknown command");
	reply_error(cf, controlfd, &raddr, rlen, cookie, 3);
	return 0;
    }
	
    call_id = argv[1];			
    if (op == UPDATE || op == LOOKUP || op == PLAY) {
	if (argc < 5 || argc > 6) {
	    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
	    reply_error(cf, controlfd, &raddr, rlen, cookie, 4);
	    return 0;
	}
	from_tag = argv[4];
	to_tag = argv[5];
	if (op == PLAY && argv[0][1] != '\0')
	    playcount = atoi(argv[0] + 1);
    }
    if (op == COPY) {
		if (argc < 4 || argc > 5) {
		    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
		    reply_error(cf, controlfd, &raddr, rlen, cookie, 1);
		    return 0;
		}
		recording_name = argv[2];
		from_tag = argv[3];
		to_tag = argv[4];
    }
    if (op == DELETE || op == RECORD || op == NOPLAY || op == QUERY) {
			if (argc < 3 || argc > 4) {
			    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
			    reply_error(cf, controlfd, &raddr, rlen, cookie, 1);
			    return 0;
			}
		from_tag = argv[2];
		to_tag = argv[3];