siptrace.c

Go to the documentation of this file.

/* 
 * $Id$ 
 *
 * siptrace module - helper module to trace sip messages
 *
 * Copyright (C) 2006 Voice Sistem S.R.L.
 * Copyright (C) 2009 Daniel-Constantin Mierla (asipto.com)
 *
 * This file is part of Kamailio, a free SIP server.
 *
 * Kamailio 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
 *
 * Kamailio 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 <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include "../../sr_module.h"
#include "../../dprint.h"
#include "../../ut.h"
#include "../../ip_addr.h"
#include "../../mem/mem.h"
#include "../../mem/shm_mem.h"
#include "../../lib/kmi/mi.h"
#include "../../lib/srdb1/db.h"
#include "../../parser/parse_content.h"
#include "../../parser/parse_from.h"
#include "../../parser/parse_cseq.h"
#include "../../pvar.h"
#include "../../modules/tm/tm_load.h"
#include "../../modules/sl/sl.h"
#include "../../str.h"
#include "../../onsend.h"

#include "../../modules/sipcapture/sipcapture.h"

#ifdef STATISTICS
#include "../../lib/kcore/statistics.h"
#endif

MODULE_VERSION

struct _siptrace_data {
        struct usr_avp *avp;
        int_str avp_value;
        struct search_state state;
        str body;
        str callid;
        str method;
        str status;
        char *dir;
        str fromtag;
        str fromip;
        str toip;
        char toip_buff[IP_ADDR_MAX_STR_SIZE+12];
        char fromip_buff[IP_ADDR_MAX_STR_SIZE+12];
        struct timeval tv;
#ifdef STATISTICS
        stat_var *stat;
#endif
};

struct tm_binds tmb;

sl_api_t slb;

/* module function prototypes */
static int mod_init(void);
static int child_init(int rank);
static void destroy(void);
static int sip_trace(struct sip_msg*, char*, char*);

static int sip_trace_store_db(struct _siptrace_data* sto);
static int trace_send_duplicate(char *buf, int len);

static void trace_onreq_in(struct cell* t, int type, struct tmcb_params *ps);
static void trace_onreq_out(struct cell* t, int type, struct tmcb_params *ps);
static void trace_onreply_in(struct cell* t, int type, struct tmcb_params *ps);
static void trace_onreply_out(struct cell* t, int type, struct tmcb_params *ps);
static void trace_sl_onreply_out(sl_cbp_t *slcb);
static void trace_sl_ack_in(sl_cbp_t *slcb);

static int trace_send_hep_duplicate(str *body, str *from, str *to);
static int pipport2su (char *pipport, union sockaddr_union *tmp_su, unsigned int *proto);


static struct mi_root* sip_trace_mi(struct mi_root* cmd, void* param );

static str db_url             = str_init(DEFAULT_RODB_URL);
static str siptrace_table     = str_init("sip_trace");
static str date_column        = str_init("time_stamp");  /* 00 */
static str callid_column      = str_init("callid");      /* 01 */
static str traced_user_column = str_init("traced_user"); /* 02 */
static str msg_column         = str_init("msg");         /* 03 */
static str method_column      = str_init("method");      /* 04 */
static str status_column      = str_init("status");      /* 05 */
static str fromip_column      = str_init("fromip");      /* 06 */
static str toip_column        = str_init("toip");        /* 07 */
static str fromtag_column     = str_init("fromtag");     /* 08 */
static str direction_column   = str_init("direction");   /* 09 */
static str time_us_column     = str_init("time_us");     /* 10 */

#define NR_KEYS 11

#define XHEADERS_BUFSIZE 512

int trace_flag = -1;
int trace_on   = 0;
int trace_sl_acks = 1;

int trace_to_database = 1;
int trace_delayed = 0;

int hep_version = 1;
int hep_capture_id = 1;

int xheaders_write = 0;
int xheaders_read = 0;

str    dup_uri_str      = {0, 0};
struct sip_uri *dup_uri = 0;

int *trace_on_flag = NULL;
int *trace_to_database_flag = NULL;

int *xheaders_write_flag = NULL;
int *xheaders_read_flag = NULL;

static unsigned short traced_user_avp_type = 0;
static int_str traced_user_avp;
static str traced_user_avp_str = {NULL, 0};

static unsigned short trace_table_avp_type = 0;
static int_str trace_table_avp;
static str trace_table_avp_str = {NULL, 0};

static str trace_local_ip = {NULL, 0};

int hep_mode_on = 0;

db1_con_t *db_con = NULL;               
db_func_t db_funcs;                     
static cmd_export_t cmds[] = {
        {"sip_trace", (cmd_function)sip_trace, 0, 0, 0, ANY_ROUTE},
        {"sip_trace", (cmd_function)sip_trace, 1, 0, 0, ANY_ROUTE},
        {0, 0, 0, 0, 0, 0}
};


static param_export_t params[] = {
        {"db_url",             STR_PARAM, &db_url.s             },
        {"table",              STR_PARAM, &siptrace_table.s     },
        {"date_column",        STR_PARAM, &date_column.s        },
        {"callid_column",      STR_PARAM, &callid_column.s      },
        {"traced_user_column", STR_PARAM, &traced_user_column.s },
        {"msg_column",         STR_PARAM, &msg_column.s         },
        {"method_column",      STR_PARAM, &method_column.s      },
        {"status_column",      STR_PARAM, &status_column.s      },
        {"fromip_column",      STR_PARAM, &fromip_column.s      },
        {"toip_column",        STR_PARAM, &toip_column.s        },
        {"fromtag_column",     STR_PARAM, &fromtag_column.s     },
        {"direction_column",   STR_PARAM, &direction_column.s   },
        {"trace_flag",         INT_PARAM, &trace_flag           },
        {"trace_on",           INT_PARAM, &trace_on             },
        {"traced_user_avp",    STR_PARAM, &traced_user_avp_str.s},
        {"trace_table_avp",    STR_PARAM, &trace_table_avp_str.s},
        {"duplicate_uri",      STR_PARAM, &dup_uri_str.s        },
        {"trace_to_database",  INT_PARAM, &trace_to_database    },
        {"trace_local_ip",     STR_PARAM, &trace_local_ip.s     },
        {"trace_sl_acks",      INT_PARAM, &trace_sl_acks        },
        {"xheaders_write",     INT_PARAM, &xheaders_write       },
        {"xheaders_read",      INT_PARAM, &xheaders_read        },
        {"hep_mode_on",        INT_PARAM, &hep_mode_on          },       
        {"hep_version",        INT_PARAM, &hep_version          },
        {"hep_capture_id",     INT_PARAM, &hep_capture_id       },              
        {"trace_delayed",      INT_PARAM, &trace_delayed        },
        {0, 0, 0}
};

static mi_export_t mi_cmds[] = {
        { "sip_trace", sip_trace_mi,   0,  0,  0 },
        { 0, 0, 0, 0, 0}
};


#ifdef STATISTICS
stat_var* siptrace_req;
stat_var* siptrace_rpl;

stat_export_t siptrace_stats[] = {
        {"traced_requests" ,  0,  &siptrace_req  },
        {"traced_replies"  ,  0,  &siptrace_rpl  },
        {0,0,0}
};
#endif

struct module_exports exports = {
        "siptrace", 
        DEFAULT_DLFLAGS, 
        cmds,       
        params,     
#ifdef STATISTICS
        siptrace_stats,  
#else
        0,          
#endif
        mi_cmds,    
        0,          
        0,          
        mod_init,   
        0,          
        destroy,    
        child_init  
};


static int mod_init(void)
{
        pv_spec_t avp_spec;
        sl_cbelem_t slcb;

#ifdef STATISTICS
        /* register statistics */
        if (register_module_stats(exports.name, siptrace_stats)!=0)
        {
                LM_ERR("failed to register core statistics\n");
                return -1;
        }
#endif

        if(register_mi_mod(exports.name, mi_cmds)!=0)
        {
                LM_ERR("failed to register MI commands\n");
                return -1;
        }

        db_url.len = strlen(db_url.s);
        siptrace_table.len = strlen(siptrace_table.s);
        date_column.len = strlen(date_column.s);
        callid_column.len = strlen(callid_column.s);
        traced_user_column.len = strlen(traced_user_column.s);
        msg_column.len = strlen(msg_column.s);
        method_column.len = strlen(method_column.s);
        status_column.len = strlen(status_column.s);
        fromip_column.len = strlen(fromip_column.s);
        toip_column.len = strlen(toip_column.s);
        fromtag_column.len = strlen(fromtag_column.s);
        direction_column.len = strlen(direction_column.s);
        if (traced_user_avp_str.s)
                traced_user_avp_str.len = strlen(traced_user_avp_str.s);
        if (trace_table_avp_str.s)
                trace_table_avp_str.len = strlen(trace_table_avp_str.s);
        if (dup_uri_str.s)
                dup_uri_str.len = strlen(dup_uri_str.s);
        if (trace_local_ip.s)
                trace_local_ip.len = strlen(trace_local_ip.s);

        if (trace_flag<0 || trace_flag>(int)MAX_FLAG)
        {
                LM_ERR("invalid trace flag %d\n", trace_flag);
                return -1;
        }
        trace_flag = 1<<trace_flag;

        trace_to_database_flag = (int*)shm_malloc(sizeof(int));
        if(trace_to_database_flag==NULL) {
                LM_ERR("no more shm memory left\n");
                return -1;
        }

        *trace_to_database_flag = trace_to_database;

        /* Find a database module if needed */
        if(trace_to_database_flag!=NULL && *trace_to_database_flag!=0) {
                if (db_bind_mod(&db_url, &db_funcs))
                {
                        LM_ERR("unable to bind database module\n");
                        return -1;
                }
                if (trace_to_database_flag && !DB_CAPABILITY(db_funcs, DB_CAP_INSERT))
                {
                        LM_ERR("database modules does not provide all functions needed"
                                        " by module\n");
                        return -1;
                }
        }

        if(hep_version != 1 && hep_version != 2) {
  
                  LM_ERR("unsupported version of HEP");
                  return -1;
        }                                          


        trace_on_flag = (int*)shm_malloc(sizeof(int));
        if(trace_on_flag==NULL) {
                LM_ERR("no more shm memory left\n");
                return -1;
        }

        *trace_on_flag = trace_on;

        xheaders_write_flag = (int*)shm_malloc(sizeof(int));
        xheaders_read_flag = (int*)shm_malloc(sizeof(int));
        if (!(xheaders_write_flag && xheaders_read_flag)) {
                LM_ERR("no more shm memory left\n");
                return -1;
        }
        *xheaders_write_flag = xheaders_write;
        *xheaders_read_flag = xheaders_read;

        /* register callbacks to TM */
        if (load_tm_api(&tmb)!=0) {
                LM_WARN("can't load tm api. Will not install tm callbacks.\n");
        } else if(tmb.register_tmcb( 0, 0, TMCB_REQUEST_IN, trace_onreq_in, 0, 0) <=0) {
                LM_ERR("can't register trace_onreq_in\n");
                return -1;
        }

        /* bind the SL API */
        if (sl_load_api(&slb)!=0) {
                LM_WARN("cannot bind to SL API. Will not install sl callbacks.\n");
        } else {
                /* register sl callbacks */
                memset(&slcb, 0, sizeof(sl_cbelem_t));

                slcb.type = SLCB_REPLY_READY;
                slcb.cbf  = trace_sl_onreply_out;
                if (slb.register_cb(&slcb) != 0) {
                        LM_ERR("can't register for SLCB_REPLY_READY\n");
                        return -1;
                }

                if(trace_sl_acks)
                {
                        slcb.type = SLCB_ACK_FILTERED;
                        slcb.cbf  = trace_sl_ack_in;
                        if (slb.register_cb(&slcb) != 0) {
                                LM_ERR("can't register for SLCB_ACK_FILTERED\n");
                                return -1;
                        }
                }
        }

        if(dup_uri_str.s!=0)
        {
                dup_uri_str.len = strlen(dup_uri_str.s);
                dup_uri = (struct sip_uri *)pkg_malloc(sizeof(struct sip_uri));
                if(dup_uri==0)
                {
                        LM_ERR("no more pkg memory left\n");
                        return -1;
                }
                memset(dup_uri, 0, sizeof(struct sip_uri));
                if(parse_uri(dup_uri_str.s, dup_uri_str.len, dup_uri)<0)
                {
                        LM_ERR("bad dup uri\n");
                        return -1;
                }
        }

        if(traced_user_avp_str.s && traced_user_avp_str.len > 0)
        {
                if (pv_parse_spec(&traced_user_avp_str, &avp_spec)==0
                                || avp_spec.type!=PVT_AVP)
                {
                        LM_ERR("malformed or non AVP %.*s AVP definition\n",
                                        traced_user_avp_str.len, traced_user_avp_str.s);
                        return -1;
                }

                if(pv_get_avp_name(0, &avp_spec.pvp, &traced_user_avp,
                                        &traced_user_avp_type)!=0)
                {
                        LM_ERR("[%.*s] - invalid AVP definition\n",
                                        traced_user_avp_str.len, traced_user_avp_str.s);
                        return -1;
                }
        } else {
                traced_user_avp.n = 0;
                traced_user_avp_type = 0;
        }
        if(trace_table_avp_str.s && trace_table_avp_str.len > 0)
        {
                if (pv_parse_spec(&trace_table_avp_str, &avp_spec)==0
                                || avp_spec.type!=PVT_AVP)
                {
                        LM_ERR("malformed or non AVP %.*s AVP definition\n",
                                        trace_table_avp_str.len, trace_table_avp_str.s);
                        return -1;
                }

                if(pv_get_avp_name(0, &avp_spec.pvp, &trace_table_avp,
                                        &trace_table_avp_type)!=0)
                {
                        LM_ERR("[%.*s] - invalid AVP definition\n",
                                        trace_table_avp_str.len, trace_table_avp_str.s);
                        return -1;
                }
        } else {
                trace_table_avp.n = 0;
                trace_table_avp_type = 0;
        }

        return 0;
}


static int child_init(int rank)
{
        if (rank==PROC_INIT || rank==PROC_MAIN || rank==PROC_TCP_MAIN)
                return 0; /* do nothing for the main process */

        if(trace_to_database_flag!=NULL && *trace_to_database_flag!=0) {
                db_con = db_funcs.init(&db_url);
                if (!db_con)
                {
                        LM_ERR("unable to connect to database. Please check configuration.\n");
                        return -1;
                }
        }

        return 0;
}


static void destroy(void)
{
        if(trace_to_database_flag!=NULL && *trace_to_database_flag!=0) {
                if (db_con!=NULL)
                        db_funcs.close(db_con);
        }

        if (trace_on_flag)
                shm_free(trace_on_flag);

}

static inline int siptrace_copy_proto(int proto, char *buf)
{
        if(buf==0)
                return -1;
        if(proto==PROTO_TCP) {
                strcpy(buf, "tcp:");
        } else if(proto==PROTO_TLS) {
                strcpy(buf, "tls:");
        } else if(proto==PROTO_SCTP) {
                strcpy(buf, "sctp:");
        } else {
                strcpy(buf, "udp:");
        }
        return 0;
}

static inline str* siptrace_get_table(void)
{
        static int_str         avp_value;
        struct usr_avp *avp;

        if(trace_table_avp.n==0)
                return &siptrace_table;

        avp = NULL;
        if(trace_table_avp.n!=0)
                avp=search_first_avp(trace_table_avp_type, trace_table_avp, &avp_value,
                                0);

        if(avp==NULL || !is_avp_str_val(avp) || avp_value.s.len<=0)
                return &siptrace_table;

        return &avp_value.s;
}

static int sip_trace_prepare(sip_msg_t *msg)
{
        if(parse_from_header(msg)==-1 || msg->from==NULL || get_from(msg)==NULL) {
                LM_ERR("cannot parse FROM header\n");
                goto error;
        }

        if(parse_headers(msg, HDR_CALLID_F, 0)!=0 || msg->callid==NULL
                        || msg->callid->body.s==NULL) {
                LM_ERR("cannot parse call-id\n");
                goto error;
        }

        return 0;
error:
        return -1;
}

// Appends x-headers to the message in sto->body containing data from sto
static int sip_trace_xheaders_write(struct _siptrace_data *sto)
{
        char* buf = NULL;
        int bytes_written = 0;
        char* eoh = NULL;
        int eoh_offset = 0;
        char* new_eoh = NULL;

        if(xheaders_write_flag==NULL || *xheaders_write_flag==0)
                return 0;

        // Memory for the message with some additional headers.
        // It gets free()ed in sip_trace_xheaders_free().
        buf = pkg_malloc(sto->body.len + XHEADERS_BUFSIZE);
        if (buf == NULL) {
                LM_ERR("sip_trace_xheaders_write: out of memory\n");
                return -1;
        }

        // Copy the whole message to buf first; it must be \0-terminated for
        // strstr() to work. Then search for the end-of-header sequence.
        memcpy(buf, sto->body.s, sto->body.len);
        buf[sto->body.len] = '\0';
        eoh = strstr(buf, "\r\n\r\n");
        if (eoh == NULL) {
                LM_ERR("sip_trace_xheaders_write: malformed message\n");
                goto error;
        }
        eoh += 2; // the first \r\n belongs to the last header => skip it

        // Write the new headers a the end-of-header position. This overwrites
        // the \r\n terminating the old headers and the beginning of the message
        // body. Both will be recovered later.
        bytes_written = snprintf(eoh, XHEADERS_BUFSIZE,
                        "X-Siptrace-Fromip: %.*s\r\n"
                        "X-Siptrace-Toip: %.*s\r\n"
                        "X-Siptrace-Time: %llu %llu\r\n"
                        "X-Siptrace-Method: %.*s\r\n"
                        "X-Siptrace-Dir: %s\r\n",
                        sto->fromip.len, sto->fromip.s,
                        sto->toip.len, sto->toip.s,
                        (unsigned long long)sto->tv.tv_sec, (unsigned long long)sto->tv.tv_usec,
                        sto->method.len, sto->method.s,
                        sto->dir);
        if (bytes_written >= XHEADERS_BUFSIZE) {
                LM_ERR("sip_trace_xheaders_write: string too long\n");
                goto error;
        }

        // Copy the \r\n terminating the old headers and the message body from the
        // old buffer in sto->body.s to the new end-of-header in buf.
        eoh_offset = eoh - buf;
        new_eoh = eoh + bytes_written;
        memcpy(new_eoh, sto->body.s + eoh_offset, sto->body.len - eoh_offset);

        // Change sto to point to the new buffer.
        sto->body.s = buf;
        sto->body.len += bytes_written;
        return 0;
error:
        if(buf != NULL)
                pkg_free(buf);
        return -1;
}

// Parses x-headers, saves the data back to sto, and removes the x-headers
// from the message in sto->buf
static int sip_trace_xheaders_read(struct _siptrace_data *sto)
{
        char* searchend = NULL;
        char* eoh = NULL;
        char* xheaders = NULL;
        long long unsigned int tv_sec, tv_usec;

        if(xheaders_read_flag==NULL || *xheaders_read_flag==0)
                return 0;

        // Find the end-of-header marker \r\n\r\n
        searchend = sto->body.s + sto->body.len - 3;
        eoh = memchr(sto->body.s, '\r', searchend - eoh);
        while (eoh != NULL && eoh < searchend) {
                if (memcmp(eoh, "\r\n\r\n", 4) == 0)
                        break;
                eoh = memchr(eoh + 1, '\r', searchend - eoh);
        }
        if (eoh == NULL) {
                LM_ERR("sip_trace_xheaders_read: malformed message\n");
                return -1;
        }

        // Find x-headers: eoh will be overwritten by \0 to allow the use of
        // strstr(). The byte at eoh will later be recovered, when the
        // message body is shifted towards the beginning of the message
        // to remove the x-headers.
        *eoh = '\0';
        xheaders = strstr(sto->body.s, "\r\nX-Siptrace-Fromip: ");
        if (xheaders == NULL) {
                LM_ERR("sip_trace_xheaders_read: message without x-headers "
                                "from %.*s, callid %.*s\n",
                                sto->fromip.len, sto->fromip.s, sto->callid.len, sto->callid.s);
                return -1;
        }

        // Allocate memory for new strings in sto
        // (gets free()ed in sip_trace_xheaders_free() )
        sto->fromip.s = pkg_malloc(51);
        sto->toip.s = pkg_malloc(51);
        sto->method.s = pkg_malloc(51);
        sto->dir = pkg_malloc(4);
        if (!(sto->fromip.s && sto->toip.s && sto->method.s && sto->dir)) {
                LM_ERR("sip_trace_xheaders_read: out of memory\n");
                goto erroraftermalloc;
        }

        // Parse the x-headers: scanf()
        if (sscanf(xheaders, "\r\n"
                                "X-Siptrace-Fromip: %50s\r\n"
                                "X-Siptrace-Toip: %50s\r\n"
                                "X-Siptrace-Time: %llu %llu\r\n"
                                "X-Siptrace-Method: %50s\r\n"
                                "X-Siptrace-Dir: %3s",
                                sto->fromip.s, sto->toip.s,
                                &tv_sec, &tv_usec,
                                sto->method.s,
                                sto->dir) == EOF) {
                LM_ERR("sip_trace_xheaders_read: malformed x-headers\n");
                goto erroraftermalloc;
        }
        sto->fromip.len = strlen(sto->fromip.s);
        sto->toip.len = strlen(sto->toip.s);
        sto->tv.tv_sec = (time_t)tv_sec;
        sto->tv.tv_usec = (suseconds_t)tv_usec;
        sto->method.len = strlen(sto->method.s);

        // Remove the x-headers: the message body is shifted towards the beginning
        // of the message, overwriting the x-headers. Before that, the byte at eoh
        // is recovered.
        *eoh = '\r';
        memmove(xheaders, eoh, sto->body.len - (eoh - sto->body.s));
        sto->body.len -= eoh - xheaders;

        return 0;

erroraftermalloc:
        if (sto->fromip.s)
                pkg_free(sto->fromip.s);
        if (sto->toip.s)
                pkg_free(sto->toip.s);
        if (sto->method.s)
                pkg_free(sto->method.s);
        if (sto->dir)
                pkg_free(sto->dir);
        return -1;
}

// Frees the memory allocated by sip_trace_xheaders_{write,read}
static int sip_trace_xheaders_free(struct _siptrace_data *sto)
{
        if (xheaders_write_flag != NULL && *xheaders_write_flag != 0) {
                if(sto->body.s)
                        pkg_free(sto->body.s);
        }

        if (xheaders_read_flag != NULL && *xheaders_read_flag != 0) {
                if(sto->fromip.s)
                        pkg_free(sto->fromip.s);
                if(sto->toip.s)
                        pkg_free(sto->toip.s);
                if(sto->dir)
                        pkg_free(sto->dir);
        }

        return 0;
}

static int sip_trace_store(struct _siptrace_data *sto)
{
        if(sto==NULL)
        {
                LM_DBG("invalid parameter\n");
                return -1;
        }

        gettimeofday(&sto->tv, NULL);

        if (sip_trace_xheaders_read(sto) != 0)
                return -1;
        int ret = sip_trace_store_db(sto);

        if (sip_trace_xheaders_write(sto) != 0)
                return -1;

        if(hep_mode_on) trace_send_hep_duplicate(&sto->body, &sto->fromip, &sto->toip);
        else trace_send_duplicate(sto->body.s, sto->body.len);

        if (sip_trace_xheaders_free(sto) != 0)
                return -1;

        return ret;
}

static int sip_trace_store_db(struct _siptrace_data *sto)
{
        if(trace_to_database_flag==NULL || *trace_to_database_flag==0)
                goto done;

        db_key_t db_keys[NR_KEYS];
        db_val_t db_vals[NR_KEYS];

        db_keys[0] = &msg_column;
        db_vals[0].type = DB1_BLOB;
        db_vals[0].nul = 0;
        db_vals[0].val.blob_val = sto->body;

        db_keys[1] = &callid_column;
        db_vals[1].type = DB1_STR;
        db_vals[1].nul = 0;
        db_vals[1].val.str_val = sto->callid;

        db_keys[2] = &method_column;
        db_vals[2].type = DB1_STR;
        db_vals[2].nul = 0;
        db_vals[2].val.str_val = sto->method;

        db_keys[3] = &status_column;
        db_vals[3].type = DB1_STR;
        db_vals[3].nul = 0;
        db_vals[3].val.str_val = sto->status;

        db_keys[4] = &fromip_column;
        db_vals[4].type = DB1_STR;
        db_vals[4].nul = 0;
        db_vals[4].val.str_val = sto->fromip;

        db_keys[5] = &toip_column;
        db_vals[5].type = DB1_STR;
        db_vals[5].nul = 0;
        db_vals[5].val.str_val = sto->toip;

        db_keys[6] = &date_column;
        db_vals[6].type = DB1_DATETIME;
        db_vals[6].nul = 0;
        db_vals[6].val.time_val = sto->tv.tv_sec;

        db_keys[7] = &direction_column;
        db_vals[7].type = DB1_STRING;
        db_vals[7].nul = 0;
        db_vals[7].val.string_val = sto->dir;

        db_keys[8] = &fromtag_column;
        db_vals[8].type = DB1_STR;
        db_vals[8].nul = 0;
        db_vals[8].val.str_val = sto->fromtag;

        db_keys[9] = &traced_user_column;
        db_vals[9].type = DB1_STR;
        db_vals[9].nul = 0;

        db_keys[10] = &time_us_column;
        db_vals[10].type = DB1_INT;
        db_vals[10].nul = 0;
        db_vals[10].val.int_val = sto->tv.tv_usec;

        db_funcs.use_table(db_con, siptrace_get_table());

        if(trace_on_flag!=NULL && *trace_on_flag!=0) {
                db_vals[9].val.str_val.s   = "";
                db_vals[9].val.str_val.len = 0;

                LM_DBG("storing info...\n");
                if(trace_delayed!=0 && db_funcs.insert_delayed!=NULL)
                {
                        if(db_funcs.insert_delayed(db_con, db_keys, db_vals, NR_KEYS)<0) {
                                LM_ERR("error storing trace\n");
                                goto error;
                        }
                } else {
                        if(db_funcs.insert(db_con, db_keys, db_vals, NR_KEYS) < 0) {
                                LM_ERR("error storing trace\n");
                                goto error;
                        }
                }
#ifdef STATISTICS
                update_stat(sto->stat, 1);
#endif
        }

        if(sto->avp==NULL)
                goto done;

        db_vals[9].val.str_val = sto->avp_value.s;

        LM_DBG("storing info...\n");
        if(trace_delayed!=0 && db_funcs.insert_delayed!=NULL)
        {
                if(db_funcs.insert_delayed(db_con, db_keys, db_vals, NR_KEYS) < 0) {
                        LM_ERR("error storing trace\n");
                        goto error;
                }
        } else {
                if(db_funcs.insert(db_con, db_keys, db_vals, NR_KEYS) < 0) {
                        LM_ERR("error storing trace\n");
                        goto error;
                }
        }

        sto->avp = search_next_avp(&sto->state, &sto->avp_value);
        while(sto->avp!=NULL) {
                db_vals[9].val.str_val = sto->avp_value.s;

                LM_DBG("storing info...\n");
                if(trace_delayed!=0 && db_funcs.insert_delayed!=NULL)
                {
                        if(db_funcs.insert_delayed(db_con, db_keys, db_vals, NR_KEYS) < 0) {
                                LM_ERR("error storing trace\n");
                                goto error;
                        }
                } else {
                        if(db_funcs.insert(db_con, db_keys, db_vals, NR_KEYS) < 0) {
                                LM_ERR("error storing trace\n");
                                goto error;
                        }
                }
                sto->avp = search_next_avp(&sto->state, &sto->avp_value);
        }

done:
        return 1;
error:
        return -1;
}

static int sip_trace(struct sip_msg *msg, char *dir, char *s2)
{
        struct _siptrace_data sto;
        struct onsend_info *snd_inf = NULL;

        if(msg==NULL) {
                LM_DBG("nothing to trace\n");
                return -1;
        }
        memset(&sto, 0, sizeof(struct _siptrace_data));

        if(traced_user_avp.n!=0)
                sto.avp=search_first_avp(traced_user_avp_type, traced_user_avp,
                                &sto.avp_value, &sto.state);

        if((sto.avp==NULL) && (trace_on_flag==NULL || *trace_on_flag==0)) {
                LM_DBG("trace off...\n");
                return -1;
        }
        if(sip_trace_prepare(msg)<0)
                return -1;

        sto.callid = msg->callid->body;

        if(msg->first_line.type==SIP_REQUEST) {
                sto.method = msg->first_line.u.request.method;
        } else {
                if(parse_headers(msg, HDR_CSEQ_F, 0) != 0 || msg->cseq==NULL
                                || msg->cseq->parsed==NULL) {
                        LM_ERR("cannot parse cseq header\n");
                        return -1;
                }
                sto.method = get_cseq(msg)->method;
        }

        if(msg->first_line.type==SIP_REPLY) {
                sto.status = msg->first_line.u.reply.status;
        } else {
                sto.status.s = "";
                sto.status.len = 0;
        }

        snd_inf=get_onsend_info();
        if(snd_inf==NULL) {
                sto.body.s = msg->buf;
                sto.body.len = msg->len;

                siptrace_copy_proto(msg->rcv.proto, sto.fromip_buff);
                strcat(sto.fromip_buff, ip_addr2a(&msg->rcv.src_ip));
                strcat(sto.fromip_buff,":");
                strcat(sto.fromip_buff, int2str(msg->rcv.src_port, NULL));
                sto.fromip.s = sto.fromip_buff;
                sto.fromip.len = strlen(sto.fromip_buff);

                siptrace_copy_proto(msg->rcv.proto, sto.toip_buff);
                strcat(sto.toip_buff, ip_addr2a(&msg->rcv.dst_ip));
                strcat(sto.toip_buff,":");
                strcat(sto.toip_buff, int2str(msg->rcv.dst_port, NULL));
                sto.toip.s = sto.toip_buff;
                sto.toip.len = strlen(sto.toip_buff);

                sto.dir = (dir)?dir:"in";
        } else {
                sto.body.s   = snd_inf->buf;
                sto.body.len = snd_inf->len;

                strncpy(sto.fromip_buff, snd_inf->send_sock->sock_str.s,
                                snd_inf->send_sock->sock_str.len);
                sto.fromip.s = sto.fromip_buff;
                sto.fromip.len = strlen(sto.fromip_buff);

                siptrace_copy_proto(snd_inf->send_sock->proto, sto.toip_buff);
                strcat(sto.toip_buff, suip2a(snd_inf->to, sizeof(*snd_inf->to)));
                strcat(sto.toip_buff,":");
                strcat(sto.toip_buff, int2str((int)su_getport(snd_inf->to), NULL));
                sto.toip.s = sto.toip_buff;
                sto.toip.len = strlen(sto.toip_buff);

                sto.dir = "out";
        }

        sto.fromtag = get_from(msg)->tag_value;

#ifdef STATISTICS
        if(msg->first_line.type==SIP_REPLY) {
                sto.stat = siptrace_rpl;
        } else {
                sto.stat = siptrace_req;
        }
#endif
        return sip_trace_store(&sto);
}

#define trace_is_off(_msg) \
        (trace_on_flag==NULL || *trace_on_flag==0 || \
         ((_msg)->flags&trace_flag)==0)

static void trace_onreq_in(struct cell* t, int type, struct tmcb_params *ps)
{
        struct sip_msg* msg;
        int_str         avp_value;
        struct usr_avp* avp;

        if(t==NULL || ps==NULL)
        {
                LM_DBG("no uas request, local transaction\n");
                return;
        }

        msg = ps->req;
        if(msg==NULL)
        {
                LM_DBG("no uas request, local transaction\n");
                return;
        }

        avp = NULL;
        if(traced_user_avp.n!=0)
                avp=search_first_avp(traced_user_avp_type, traced_user_avp, &avp_value,
                                0);

        if((avp==NULL) && trace_is_off(msg))
        {
                LM_DBG("trace off...\n");
                return;
        }

        if(parse_from_header(msg)==-1 || msg->from==NULL || get_from(msg)==NULL)
        {
                LM_ERR("cannot parse FROM header\n");
                return;
        }

        if(parse_headers(msg, HDR_CALLID_F, 0)!=0)
        {
                LM_ERR("cannot parse call-id\n");
                return;
        }

        if(tmb.register_tmcb(0, t, TMCB_REQUEST_SENT, trace_onreq_out, 0, 0) <=0)
        {
                LM_ERR("can't register trace_onreq_out\n");
                return;
        }
        if(tmb.register_tmcb(0, t, TMCB_RESPONSE_IN, trace_onreply_in, 0, 0) <=0)
        {
                LM_ERR("can't register trace_onreply_in\n");
                return;
        }

        if(tmb.register_tmcb(0, t, TMCB_RESPONSE_SENT, trace_onreply_out, 0, 0)<=0)
        {
                LM_ERR("can't register trace_onreply_out\n");
                return;
        }
}

static void trace_onreq_out(struct cell* t, int type, struct tmcb_params *ps)
{
        struct _siptrace_data sto;
        sip_msg_t *msg;
        struct ip_addr to_ip;
        int len;
        struct dest_info *dst;

        if(t==NULL || ps==NULL) {
                LM_DBG("very weird\n");
                return;
        }

        if(ps->flags&TMCB_RETR_F) {
                LM_DBG("retransmission\n");
                return;
        }
        msg=ps->req;
        if(msg==NULL) {
                /* check if it is outgoing cancel, t is INVITE
                 * and send_buf starts with CANCEL */
                if(is_invite(t) && ps->send_buf.len>7
                                && strncmp(ps->send_buf.s, "CANCEL ", 7)==0) {
                        msg = t->uas.request;
                        if(msg==NULL) {
                                LM_DBG("no uas msg for INVITE transaction\n");
                                return;
                        } else {
                                LM_DBG("recording CANCEL based on INVITE transaction\n");
                        }
                } else {
                        LM_DBG("no uas msg, local transaction\n");
                        return;
                }
        }
        memset(&sto, 0, sizeof(struct _siptrace_data));

        if(traced_user_avp.n!=0)
                sto.avp=search_first_avp(traced_user_avp_type, traced_user_avp,
                                &sto.avp_value, &sto.state);

        if((sto.avp==NULL) && trace_is_off(msg) ) {
                LM_DBG("trace off...\n");
                return;
        }

        if(sip_trace_prepare(msg)<0)
                return;

        if(ps->send_buf.len>0) {
                sto.body = ps->send_buf;
        } else {
                sto.body.s   = "No request buffer";
                sto.body.len = sizeof("No request buffer")-1;
        }

        sto.callid = msg->callid->body;

        if(ps->send_buf.len>10) {
                sto.method.s = ps->send_buf.s;
                sto.method.len = 0;
                while(sto.method.len<ps->send_buf.len) {
                        if(ps->send_buf.s[sto.method.len]==' ')
                                break;
                        sto.method.len++;
                }
                if(sto.method.len==ps->send_buf.len)
                        sto.method.len = 10;
        } else {
                sto.method = t->method;
        }

        sto.status.s = "";
        sto.status.len = 0;

        memset(&to_ip, 0, sizeof(struct ip_addr));
        dst = ps->dst;

        if (trace_local_ip.s && trace_local_ip.len > 0) {
                sto.fromip = trace_local_ip;
        } else {
                if(dst==0 || dst->send_sock==0 || dst->send_sock->sock_str.s==0) {
                        siptrace_copy_proto(msg->rcv.proto, sto.fromip_buff);
                        strcat(sto.fromip_buff, ip_addr2a(&msg->rcv.dst_ip));
                        strcat(sto.fromip_buff,":");
                        strcat(sto.fromip_buff, int2str(msg->rcv.dst_port, NULL));
                        sto.fromip.s = sto.fromip_buff;
                        sto.fromip.len = strlen(sto.fromip_buff);
                } else {
                        sto.fromip = dst->send_sock->sock_str;
                }
        }

        if(dst==0) {
                sto.toip.s = "any:255.255.255.255";
                sto.toip.len = 19;
        } else {
                su2ip_addr(&to_ip, &dst->to);
                siptrace_copy_proto(dst->proto, sto.toip_buff);
                strcat(sto.toip_buff, ip_addr2a(&to_ip));
                strcat(sto.toip_buff, ":");
                strcat(sto.toip_buff,
                                int2str((unsigned long)su_getport(&dst->to), &len));
                sto.toip.s = sto.toip_buff;
                sto.toip.len = strlen(sto.toip_buff);
        }

        sto.dir = "out";

        sto.fromtag = get_from(msg)->tag_value;

#ifdef STATISTICS
        sto.stat = siptrace_req;
#endif

        sip_trace_store(&sto);
        return;
}

static void trace_onreply_in(struct cell* t, int type, struct tmcb_params *ps)
{
        struct _siptrace_data sto;
        sip_msg_t *msg;
        sip_msg_t *req;
        char statusbuf[8];

        if(t==NULL || t->uas.request==0 || ps==NULL) {
                LM_DBG("no uas request, local transaction\n");
                return;
        }

        req = ps->req;
        msg = ps->rpl;
        if(msg==NULL || req==NULL) {
                LM_DBG("no reply\n");
                return;
        }
        memset(&sto, 0, sizeof(struct _siptrace_data));

        if(traced_user_avp.n!=0)
                sto.avp=search_first_avp(traced_user_avp_type, traced_user_avp,
                                &sto.avp_value, &sto.state);

        if((sto.avp==NULL) &&  trace_is_off(req)) {
                LM_DBG("trace off...\n");
                return;
        }

        if(sip_trace_prepare(msg)<0)
                return;

        sto.body.s = msg->buf;
        sto.body.len = msg->len;

        sto.callid = msg->callid->body;

        sto.method = t->method;

        strcpy(statusbuf, int2str(ps->code, &sto.status.len));
        sto.status.s = statusbuf;

        siptrace_copy_proto(msg->rcv.proto, sto.fromip_buff);
        strcat(sto.fromip_buff, ip_addr2a(&msg->rcv.src_ip));
        strcat(sto.fromip_buff,":");
        strcat(sto.fromip_buff, int2str(msg->rcv.src_port, NULL));
        sto.fromip.s = sto.fromip_buff;
        sto.fromip.len = strlen(sto.fromip_buff);

        if(trace_local_ip.s && trace_local_ip.len > 0) {
                sto.toip = trace_local_ip;
        } else {
                siptrace_copy_proto(msg->rcv.proto, sto.toip_buff);
                strcat(sto.toip_buff, ip_addr2a(&msg->rcv.dst_ip));
                strcat(sto.toip_buff,":");
                strcat(sto.toip_buff, int2str(msg->rcv.dst_port, NULL));
                sto.toip.s = sto.toip_buff;
                sto.toip.len = strlen(sto.toip_buff);
        }

        sto.dir = "in";

        sto.fromtag = get_from(msg)->tag_value;
#ifdef STATISTICS
        sto.stat = siptrace_rpl;
#endif

        sip_trace_store(&sto);
        return;
}

static void trace_onreply_out(struct cell* t, int type, struct tmcb_params *ps)
{
        struct _siptrace_data sto;
        int faked = 0;
        struct sip_msg* msg;
        struct sip_msg* req;
        struct ip_addr to_ip;
        int len;
        char statusbuf[8];
        struct dest_info *dst;

        if (t==NULL || t->uas.request==0 || ps==NULL) {
                LM_DBG("no uas request, local transaction\n");
                return;
        }

        if(ps->flags&TMCB_RETR_F) {
                LM_DBG("retransmission\n");
                return;
        }
        memset(&sto, 0, sizeof(struct _siptrace_data));
        if(traced_user_avp.n!=0)
                sto.avp=search_first_avp(traced_user_avp_type, traced_user_avp,
                                &sto.avp_value, &sto.state);

        if((sto.avp==NULL) &&  trace_is_off(t->uas.request)) {
                LM_DBG("trace off...\n");
                return;
        }

        req = ps->req;
        msg = ps->rpl;
        if(msg==NULL || msg==FAKED_REPLY) {
                msg = t->uas.request;
                faked = 1;
        }

        if(sip_trace_prepare(msg)<0)
                return;

        if(faked==0) {
                if(ps->send_buf.len>0) {
                        sto.body = ps->send_buf;
                } else if(t->uas.response.buffer!=NULL) {
                        sto.body.s = t->uas.response.buffer;
                        sto.body.len = t->uas.response.buffer_len;
                } else if(msg->len>0) {
                        sto.body.s = msg->buf;
                        sto.body.len = msg->len;
                } else {
                        sto.body.s = "No reply buffer";
                        sto.body.len = sizeof("No reply buffer")-1;
                }
        } else {
                if(ps->send_buf.len>0) {
                        sto.body = ps->send_buf;
                } else if(t->uas.response.buffer!=NULL) {
                        sto.body.s = t->uas.response.buffer;
                        sto.body.len = t->uas.response.buffer_len;
                } else {
                        sto.body.s = "No reply buffer";
                        sto.body.len = sizeof("No reply buffer")-1;
                }
        }

        sto.callid = msg->callid->body;
        sto.method = t->method;

        if(trace_local_ip.s && trace_local_ip.len > 0) {
                sto.fromip = trace_local_ip;
        } else {
                siptrace_copy_proto(msg->rcv.proto, sto.fromip_buff);
                strcat(sto.fromip_buff, ip_addr2a(&req->rcv.dst_ip));
                strcat(sto.fromip_buff,":");
                strcat(sto.fromip_buff, int2str(req->rcv.dst_port, NULL));
                sto.fromip.s = sto.fromip_buff;
                sto.fromip.len = strlen(sto.fromip_buff);
        }

        strcpy(statusbuf, int2str(ps->code, &sto.status.len));
        sto.status.s = statusbuf;

        memset(&to_ip, 0, sizeof(struct ip_addr));
        dst = ps->dst;
        if(dst==0) {
                sto.toip.s = "any:255.255.255.255";
                sto.toip.len = 19;
        } else {
                su2ip_addr(&to_ip, &dst->to);
                siptrace_copy_proto(dst->proto, sto.toip_buff);
                strcat(sto.toip_buff, ip_addr2a(&to_ip));
                strcat(sto.toip_buff, ":");
                strcat(sto.toip_buff,
                                int2str((unsigned long)su_getport(&dst->to), &len));
                sto.toip.s = sto.toip_buff;
                sto.toip.len = strlen(sto.toip_buff);
        }

        sto.dir = "out";
        sto.fromtag = get_from(msg)->tag_value;

#ifdef STATISTICS
        sto.stat = siptrace_rpl;
#endif

        sip_trace_store(&sto);
        return;
}

static void trace_sl_ack_in(sl_cbp_t *slcbp)
{
        sip_msg_t *req;
        LM_DBG("storing ack...\n");
        req = slcbp->req;
        sip_trace(req, 0, 0);
}

static void trace_sl_onreply_out(sl_cbp_t *slcbp)
{
        sip_msg_t *req;
        struct _siptrace_data sto;
        struct sip_msg* msg;
        struct ip_addr to_ip;
        int len;
        char statusbuf[5];

        if(slcbp==NULL || slcbp->req==NULL)
        {
                LM_ERR("bad parameters\n");
                return;
        }
        req = slcbp->req;

        memset(&sto, 0, sizeof(struct _siptrace_data));
        if(traced_user_avp.n!=0)
                sto.avp=search_first_avp(traced_user_avp_type, traced_user_avp,
                                &sto.avp_value, &sto.state);

        if((sto.avp==NULL) && trace_is_off(req)) {
                LM_DBG("trace off...\n");
                return;
        }

        msg = req;

        if(sip_trace_prepare(msg)<0)
                return;

        sto.body.s = (slcbp->reply)?slcbp->reply->s:"";
        sto.body.len = (slcbp->reply)?slcbp->reply->len:0;

        sto.callid = msg->callid->body;
        sto.method = msg->first_line.u.request.method;

        if(trace_local_ip.len > 0) {
                sto.fromip = trace_local_ip;
        } else {
                siptrace_copy_proto(msg->rcv.proto, sto.fromip_buff);
                strcat(sto.fromip_buff, ip_addr2a(&req->rcv.dst_ip));
                strcat(sto.fromip_buff,":");
                strcat(sto.fromip_buff, int2str(req->rcv.dst_port, NULL));
                sto.fromip.s = sto.fromip_buff;
                sto.fromip.len = strlen(sto.fromip_buff);
        }

        strcpy(statusbuf, int2str(slcbp->code, &sto.status.len));
        sto.status.s = statusbuf;

        memset(&to_ip, 0, sizeof(struct ip_addr));
        if(slcbp->dst==0)
        {
                sto.toip.s = "any:255.255.255.255";
                sto.toip.len = 19;
        } else {
                su2ip_addr(&to_ip, &slcbp->dst->to);
                siptrace_copy_proto(req->rcv.proto, sto.toip_buff);
                strcat(sto.toip_buff, ip_addr2a(&to_ip));
                strcat(sto.toip_buff, ":");
                strcat(sto.toip_buff,
                                int2str((unsigned long)su_getport(&slcbp->dst->to), &len));
                sto.toip.s = sto.toip_buff;
                sto.toip.len = strlen(sto.toip_buff);
        }

        sto.dir = "out";
        sto.fromtag = get_from(msg)->tag_value;

#ifdef STATISTICS
        sto.stat = siptrace_rpl;
#endif

        sip_trace_store(&sto);
        return;
}


static struct mi_root* sip_trace_mi(struct mi_root* cmd_tree, void* param )
{
        struct mi_node* node;

        struct mi_node *rpl; 
        struct mi_root *rpl_tree ; 

        node = cmd_tree->node.kids;
        if(node == NULL) {
                rpl_tree = init_mi_tree( 200, MI_SSTR(MI_OK));
                if (rpl_tree == 0)
                        return 0;
                rpl = &rpl_tree->node;

                if (*trace_on_flag == 0 ) {
                        node = add_mi_node_child(rpl,0,0,0,MI_SSTR("off"));
                } else if (*trace_on_flag == 1) {
                        node = add_mi_node_child(rpl,0,0,0,MI_SSTR("on"));
                }
                return rpl_tree ;
        }
        if(trace_on_flag==NULL)
                return init_mi_tree( 500, MI_SSTR(MI_INTERNAL_ERR));

        if ( node->value.len==2 && (node->value.s[0]=='o'
                                || node->value.s[0]=='O') &&
                        (node->value.s[1]=='n'|| node->value.s[1]=='N')) {
                *trace_on_flag = 1;
                return init_mi_tree( 200, MI_SSTR(MI_OK));
        } else if ( node->value.len==3 && (node->value.s[0]=='o'
                                || node->value.s[0]=='O')
                        && (node->value.s[1]=='f'|| node->value.s[1]=='F')
                        && (node->value.s[2]=='f'|| node->value.s[2]=='F')) {
                *trace_on_flag = 0;
                return init_mi_tree( 200, MI_SSTR(MI_OK));
        } else {
                return init_mi_tree( 400, MI_SSTR(MI_BAD_PARM));
        }
}

static int trace_send_duplicate(char *buf, int len)
{
        struct dest_info dst;
        struct proxy_l * p;

        if(buf==NULL || len <= 0)
                return -1;

        if(dup_uri_str.s==0 || dup_uri==NULL)
                return 0;

        init_dest_info(&dst);
        /* create a temporary proxy*/
        dst.proto = PROTO_UDP;
        p=mk_proxy(&dup_uri->host, (dup_uri->port_no)?dup_uri->port_no:SIP_PORT,
                        dst.proto);
        if (p==0)
        {
                LM_ERR("bad host name in uri\n");
                return -1;
        }

        hostent2su(&dst.to, &p->host, p->addr_idx, (p->port)?p->port:SIP_PORT);

        dst.send_sock=get_send_socket(0, &dst.to, dst.proto);
        if (dst.send_sock==0)
        {
                LM_ERR("can't forward to af %d, proto %d no corresponding"
                                " listening socket\n", dst.to.s.sa_family, dst.proto);
                goto error;
        }

        if (msg_send(&dst, buf, len)<0)
        {
                LM_ERR("cannot send duplicate message\n");
                goto error;
        }

        free_proxy(p); /* frees only p content, not p itself */
        pkg_free(p);
        return 0;
error:
        free_proxy(p); /* frees only p content, not p itself */
        pkg_free(p);
        return -1;
}

static int trace_send_hep_duplicate(str *body, str *from, str *to)
{
        struct dest_info dst;
        struct proxy_l * p=NULL /* make gcc happy */;
        void* buffer = NULL;
        union sockaddr_union from_su;
        union sockaddr_union to_su;
        unsigned int len, buflen, proto;
        struct hep_hdr hdr;
        struct hep_iphdr hep_ipheader;
        struct hep_timehdr hep_time;
        struct timeval tvb;
        struct timezone tz;
                         
#if USE_IPV6
        struct hep_ip6hdr hep_ip6header;
#endif

        if(body->s==NULL || body->len <= 0)
                return -1;

        if(dup_uri_str.s==0 || dup_uri==NULL)
                return 0;


        gettimeofday( &tvb, &tz );
        

        /* message length */
        len = body->len 
#if USE_IPV6
                + sizeof(struct hep_ip6hdr)
#else
                + sizeof(struct hep_iphdr)          
#endif
                + sizeof(struct hep_hdr) + sizeof(struct hep_timehdr);;


        /* The packet is too big for us */
        if (unlikely(len>BUF_SIZE)){
                goto error;
        }

        /* Convert proto:ip:port to sockaddress union SRC IP */
        if (pipport2su(from->s, &from_su, &proto)==-1 || (pipport2su(to->s, &to_su, &proto)==-1))
                goto error;

        /* check if from and to are in the same family*/
        if(from_su.s.sa_family != to_su.s.sa_family) {
                LOG(L_ERR, "ERROR: trace_send_hep_duplicate: interworking detected ?\n");
                goto error;
        }

        init_dest_info(&dst);
        /* create a temporary proxy*/
        dst.proto = PROTO_UDP;
        p=mk_proxy(&dup_uri->host, (dup_uri->port_no)?dup_uri->port_no:SIP_PORT,
                        dst.proto);
        if (p==0)
        {
                LM_ERR("bad host name in uri\n");
                goto error;
        }

        hostent2su(&dst.to, &p->host, p->addr_idx, (p->port)?p->port:SIP_PORT);

        dst.send_sock=get_send_socket(0, &dst.to, dst.proto);
        if (dst.send_sock==0)
        {
                LM_ERR("can't forward to af %d, proto %d no corresponding"
                                " listening socket\n", dst.to.s.sa_family, dst.proto);
                goto error;
        }

        /* Version && proto && length */
        hdr.hp_l = sizeof(struct hep_hdr);
        hdr.hp_v = hep_version;
        hdr.hp_p = proto;

        /* AND the last */
        if (from_su.s.sa_family==AF_INET){
                /* prepare the hep headers */

                hdr.hp_f = AF_INET;
                hdr.hp_sport = htons(from_su.sin.sin_port);
                hdr.hp_dport = htons(to_su.sin.sin_port);

                hep_ipheader.hp_src = from_su.sin.sin_addr;
                hep_ipheader.hp_dst = to_su.sin.sin_addr;

                len = sizeof(struct hep_iphdr);
        }
#ifdef USE_IPV6
        else if (from_su.s.sa_family==AF_INET6){
                /* prepare the hep6 headers */

                hdr.hp_f = AF_INET6;

                hdr.hp_sport = htons(from_su.sin6.sin6_port);
                hdr.hp_dport = htons(to_su.sin6.sin6_port);

                hep_ip6header.hp6_src = from_su.sin6.sin6_addr;
                hep_ip6header.hp6_dst = to_su.sin6.sin6_addr;

                len = sizeof(struct hep_ip6hdr);
        }
#endif /* USE_IPV6 */
        else {
                LOG(L_ERR, "ERROR: trace_send_hep_duplicate: Unsupported protocol family\n");
                goto error;;
        }

        hdr.hp_l +=len;
        len += sizeof(struct hep_hdr) + body->len;
        buffer = (void *)pkg_malloc(len+1);
        if (buffer==0){
                LOG(L_ERR, "ERROR: trace_send_hep_duplicate: out of memory\n");
                goto error;
        }

        /* Copy job */
        memset(buffer, '\0', len+1);

        /* copy hep_hdr */
        memcpy((void*)buffer, &hdr, sizeof(struct hep_hdr));
        buflen = sizeof(struct hep_hdr);

        /* hep_ip_hdr */
        if(from_su.s.sa_family==AF_INET) {
                memcpy((void*)buffer + buflen, &hep_ipheader, sizeof(struct hep_iphdr));
                buflen += sizeof(struct hep_iphdr);
        }
#if USE_IPV6
        else {
                memcpy((void*)buffer+buflen, &hep_ip6header, sizeof(struct hep_ip6hdr));
                buflen += sizeof(struct hep_ip6hdr);
        }
#endif /* USE_IPV6 */

        if(hep_version == 2) {

                hep_time.tv_sec = tvb.tv_sec;
                hep_time.tv_usec = tvb.tv_usec;
                hep_time.captid = hep_capture_id;

                memcpy((void*)buffer+buflen, &hep_time, sizeof(struct hep_timehdr));
                buflen += sizeof(struct hep_timehdr);
        }

        /* PAYLOAD */
        memcpy((void*)(buffer + buflen) , (void*)body->s, body->len);
        buflen +=body->len;

        if (msg_send(&dst, buffer, buflen)<0)
        {
                LM_ERR("cannot send hep duplicate message\n");
                goto error;
        }

        free_proxy(p); /* frees only p content, not p itself */
        pkg_free(p);
        pkg_free(buffer);
        return 0;
error:
        if(p)
        {
                free_proxy(p); /* frees only p content, not p itself */
                pkg_free(p);
        }
        if(buffer) pkg_free(buffer);
        return -1;
}

static int pipport2su (char *pipport, union sockaddr_union *tmp_su, unsigned int *proto)
{
        unsigned int port_no, cutlen = 4;
        struct ip_addr *ip;
        char *p, *host_s;
        str port_str, host_uri;
        unsigned len = 0;
        char tmp_piport[256];

        /*parse protocol */
        if(strncmp(pipport, "udp:",4) == 0) *proto = IPPROTO_UDP;
        else if(strncmp(pipport, "tcp:",4) == 0) *proto = IPPROTO_TCP;
        else if(strncmp(pipport, "tls:",4) == 0) *proto = IPPROTO_IDP; /* fake proto type */
#ifdef USE_SCTP
        else if(strncmp(pipport, "sctp:",5) == 0) cutlen = 5, *proto = IPPROTO_SCTP;
#endif
        else if(strncmp(pipport, "any:",4) == 0) *proto = IPPROTO_UDP;
        else {
                LM_ERR("bad protocol %s\n", pipport);
                return -1;
        }
        
        if((len = strlen(pipport)) >= 256) {
                LM_ERR("too big pipport\n");
                goto error;
        }

        /* our tmp string */
        strncpy(tmp_piport, pipport, len+1);

        len = 0;

        /*separate proto and host */
        p = tmp_piport+cutlen;
        if( (*(p)) == '\0') {
                LM_ERR("malformed ip address\n");
                goto error;
        }
        host_s=p;

        if( (p = strrchr(p+1, ':')) == 0 ) {
                LM_DBG("no port specified\n");
                port_no = 0;
        }
        else {
                /*the address contains a port number*/
                *p = '\0';
                p++;
                port_str.s = p;
                port_str.len = strlen(p);
                LM_DBG("the port string is %s\n", p);
                if(str2int(&port_str, &port_no) != 0 ) {
                        LM_ERR("there is not a valid number port\n");
                        goto error;
                }
                *p = '\0';
        }
        
        /* now IPv6 address has no brakets. It should be fixed! */
        if (host_s[0] == '[') {
                len = strlen(host_s + 1) - 1;
                if(host_s[len+1] != ']') {
                        LM_ERR("bracket not closed\n");
                        goto error;
                }
                memmove(host_s, host_s + 1, len);
                host_s[len] = '\0';
        }

        host_uri.s = host_s;
        host_uri.len = strlen(host_s);

        /* check if it's an ip address */
        if (((ip=str2ip(&host_uri))!=0)
#ifdef  USE_IPV6
                        || ((ip=str2ip6(&host_uri))!=0)
#endif
           ) {
                ip_addr2su(tmp_su, ip, ntohs(port_no));
                return 0;       

        }

error:
        return -1;
}