topos: update sip headers for initial request forwarding

modules/topos/topos_mod.c

@@ -55,6 +55,7 @@
 
 #include "../../modules/sanity/api.h"
 
+#include "tps_storage.h"
 #include "tps_msg.h"
 
 MODULE_VERSION
@@ -131,6 +132,10 @@ static int mod_init(void)
 			goto error;
 		}
 	}
+	if(tps_storage_lock_set_init()<0) {
+		LM_ERR("failed to initialize locks set\n");
+		return -1;
+	}
 
 	if(sruid_init(&_tps_sruid, '-', "tpsh", SRUID_INC)<0)
 		return -1;
@@ -176,6 +181,7 @@ static void destroy(void)
 		tpsdbf.close(_tps_db_handle);
 		_tps_db_handle = 0;
 	}
+	tps_storage_lock_set_destroy();
 }
 
 /**

modules/topos/tps_msg.c

@@ -43,9 +43,13 @@
 #include "../../parser/contact/parse_contact.h"
 #include "../../parser/parse_refer_to.h"
 #include "tps_msg.h"
+#include "tps_storage.h"
 
 extern int _tps_param_mask_callid;
 
+/**
+ *
+ */
 int tps_skip_rw(char *s, int len)
 {
 	while(len>0)
@@ -58,6 +62,9 @@ int tps_skip_rw(char *s, int len)
 	return 0;
 }
 
+/**
+ *
+ */
 struct via_param *tps_get_via_param(struct via_body *via, str *name)
 {
 	struct via_param *p;
@@ -70,6 +77,9 @@ struct via_param *tps_get_via_param(struct via_body *via, str *name)
 	return NULL;
 }
 
+/**
+ *
+ */
 int tps_get_param_value(str *in, str *name, str *value)
 {
 	param_t* params = NULL;
@@ -93,6 +103,79 @@ int tps_get_param_value(str *in, str *name, str *value)
 
 }
 
+/**
+ *
+ */
+int tps_remove_headers(sip_msg_t *msg, uint32_t hdr)
+{
+	struct hdr_field *hf;
+	struct lump* l;
+
+	parse_headers(msg, HDR_EOH_F, 0);
+	for (hf=msg->headers; hf; hf=hf->next) {
+		if (hdr!=hf->type)
+			continue;
+		l=del_lump(msg, hf->name.s-msg->buf, hf->len, 0);
+		if (l==0) {
+			LM_ERR("no memory\n");
+			return -1;
+		}
+	}
+	return 0;
+}
+
+/**
+ *
+ */
+int tps_add_headers(sip_msg_t *msg, str *hname, str *hbody, int hpos)
+{
+	struct lump* anchor;
+	str hs;
+
+	parse_headers(msg, HDR_EOH_F, 0);
+	if(hpos == 0) { /* append */
+		/* after last header */
+		anchor = anchor_lump(msg, msg->unparsed - msg->buf, 0, 0);
+	} else { /* insert */
+		/* before first header */
+		anchor = anchor_lump(msg, msg->headers->name.s - msg->buf, 0, 0);
+	}
+
+	if(anchor == 0) {
+		LM_ERR("can't get anchor\n");
+		return -1;
+	}
+
+	hs.len = hname->len + 2 + hbody->len;
+	hs.s  = (char*)pkg_malloc(hs.len + 3);
+	if (hs.s==NULL) {
+		LM_ERR("no pkg memory left\n");
+		return -1;
+	}
+	memcpy(hs.s, hname->s, hname->len);
+	hs.s[hname->len] = ':';
+	hs.s[hname->len+1] = ' ';
+	memcpy(hs.s + hname->len + 2, hbody->s, hbody->len);
+
+	/* add end of header if not present */
+	if(hs.s[hname->len + 2 + hbody->len]!='\n') {
+		hs.s[hname->len + 2 + hbody->len] = '\r';
+		hs.s[hname->len + 2 + hbody->len+1] = '\n';
+		hs.len += 2;
+	}
+
+	if (insert_new_lump_before(anchor, hs.s, hs.len, 0) == 0) {
+		LM_ERR("can't insert lump\n");
+		pkg_free(hs.s);
+		return -1;
+	}
+
+	return 0;
+}
+
+/**
+ *
+ */
 int tps_get_uri_param_value(str *uri, str *name, str *value)
 {
 	struct sip_uri puri;
@@ -103,6 +186,9 @@ int tps_get_uri_param_value(str *uri, str *name, str *value)
 	return tps_get_param_value(&puri.params, name, value);
 }
 
+/**
+ *
+ */
 int tps_get_uri_type(str *uri, int *mode, str *value)
 {
 	struct sip_uri puri;
@@ -135,6 +221,9 @@ int tps_get_uri_type(str *uri, int *mode, str *value)
 	return 1; /* encode */
 }
 
+/**
+ *
+ */
 char* tps_msg_update(sip_msg_t *msg, unsigned int *olen)
 {
 	struct dest_info dst;
@@ -145,6 +234,9 @@ char* tps_msg_update(sip_msg_t *msg, unsigned int *olen)
 			olen, &dst, BUILD_NO_LOCAL_VIA|BUILD_NO_VIA1_UPDATE);
 }
 
+/**
+ *
+ */
 int tps_route_direction(sip_msg_t *msg)
 {
 	rr_t *rr;
@@ -187,6 +279,9 @@ int tps_route_direction(sip_msg_t *msg)
 	return 0;
 }
 
+/**
+ *
+ */
 int tps_skip_msg(sip_msg_t *msg)
 {
 	if (msg->cseq==NULL || get_cseq(msg)==NULL) {
@@ -205,6 +300,10 @@ int tps_skip_msg(sip_msg_t *msg)
  */
 int tps_request_received(sip_msg_t *msg, int dialog, int direction)
 {
+	if(dialog==0) {
+		/* nothing to do for initial request */
+		return 0;
+	}
 	return 0;
 }
 
@@ -216,12 +315,211 @@ int tps_response_received(sip_msg_t *msg)
 	return 0;
 }
 
+/**
+ *
+ */
+int tps_pack_request(sip_msg_t *msg, tps_data_t *ptsd)
+{
+	hdr_field_t *hdr;
+	via_body_t *via;
+	rr_t *rr;
+	int i;
+	int vlen;
+
+	if(ptsd->cp==NULL) {
+		ptsd->cp = ptsd->cbuf;
+	}
+	i = 0;
+	for(hdr=msg->h_via1; hdr; hdr=next_sibling_hdr(hdr)) {
+		for(via=(struct via_body*)hdr->parsed; via; via=via->next) {
+			i++;
+			vlen = tps_skip_rw(via->name.s, via->bsize);
+			if(ptsd->cp + vlen + 2 >= ptsd->cbuf + TPS_DATA_SIZE) {
+				LM_ERR("no more spage to pack via headers\n");
+				return -1;
+			}
+			if(i>1) {
+				*ptsd->cp = ',';
+				ptsd->cp++;
+				if(i>2) {
+					ptsd->x_via2.len++;
+				}
+			}
+			memcpy(ptsd->cp, via->name.s, vlen);
+			if(i==1) {
+				ptsd->x_via1.s = ptsd->cp;
+				ptsd->x_via1.len = vlen;
+				if(via->branch!=NULL) {
+					ptsd->x_vbranch1.s = ptsd->x_via1.s + (via->branch->value.s - via->name.s);
+					ptsd->x_vbranch1.len = via->branch->value.len;
+				}
+			} else {
+				if(i==2) {
+					ptsd->x_via2.s = ptsd->cp;
+				}
+				ptsd->x_via2.len += vlen;
+			}
+			ptsd->cp += vlen;
+		}
+	}
+	LM_DBG("compacted headers - x_via1: [%.*s](%d) - x_via2: [%.*s](%d)"
+			" - x_vbranch1: [%.*s](%d)\n",
+			ptsd->x_via1.len, ZSW(ptsd->x_via1.s), ptsd->x_via1.len,
+			ptsd->x_via2.len, ZSW(ptsd->x_via2.s), ptsd->x_via2.len,
+			ptsd->x_vbranch1.len, ZSW(ptsd->x_vbranch1.s), ptsd->x_vbranch1.len);
+
+	i = 0;
+	ptsd->a_rr.len = 0;
+	for(hdr=msg->record_route; hdr; hdr=next_sibling_hdr(hdr)) {
+		if (parse_rr(hdr) < 0) {
+			LM_ERR("failed to parse RR\n");
+			return -1;
+		}
+
+		for(rr =(rr_t*)hdr->parsed; rr; rr=rr->next) {
+			i++;
+			if(ptsd->cp + rr->nameaddr.uri.len + 4 >= ptsd->cbuf + TPS_DATA_SIZE) {
+				LM_ERR("no more spage to pack rr headers\n");
+				return -1;
+			}
+			if(i>1) {
+				*ptsd->cp = ',';
+				ptsd->cp++;
+				ptsd->a_rr.len++;
+			}
+			*ptsd->cp = '<';
+			if(i==1) {
+				ptsd->a_rr.s = ptsd->cp;
+			}
+			ptsd->cp++;
+			ptsd->a_rr.len++;
+
+			memcpy(ptsd->cp, rr->nameaddr.uri.s, rr->nameaddr.uri.len);
+			if(i==1) {
+				ptsd->bs_contact.s = ptsd->cp;
+				ptsd->bs_contact.len = rr->nameaddr.uri.len;
+				if(strnstr(ptsd->bs_contact.s, ";r2=on",
+							ptsd->bs_contact.len)==NULL) {
+					LM_DBG("single record routing by proxy\n");
+					ptsd->as_contact.s = ptsd->cp;
+					ptsd->as_contact.len = rr->nameaddr.uri.len;
+				}
+			} else {
+				if(i==2 && ptsd->as_contact.len==0) {
+					LM_DBG("double record routing by proxy\n");
+					ptsd->as_contact.s = ptsd->cp;
+					ptsd->as_contact.len = rr->nameaddr.uri.len;
+				}
+			}
+			ptsd->a_rr.len += rr->nameaddr.uri.len;
+			ptsd->cp += rr->nameaddr.uri.len;
+			*ptsd->cp = '>';
+			ptsd->cp++;
+			ptsd->a_rr.len++;
+		}
+
+	}
+	LM_DBG("compacted headers - a_rr: [%.*s](%d) - b_rr: [%.*s](%d)\n",
+			ptsd->a_rr.len, ZSW(ptsd->a_rr.s), ptsd->a_rr.len,
+			ptsd->b_rr.len, ZSW(ptsd->b_rr.s), ptsd->b_rr.len);
+	LM_DBG("compacted headers - as_contact: [%.*s](%d) - bs_contact: [%.*s](%d)\n",
+			ptsd->as_contact.len, ZSW(ptsd->as_contact.s), ptsd->as_contact.len,
+			ptsd->bs_contact.len, ZSW(ptsd->bs_contact.s), ptsd->bs_contact.len);
+	return 0;
+}
+
+/**
+ *
+ */
+int tps_reinsert_via(sip_msg_t *msg, tps_data_t *ptsd, str *hbody)
+{
+	str hname = str_init("Via");
+	
+	if(tps_add_headers(msg, &hname, hbody, 1)<0) {
+		return -1;
+	}
+
+	return 0;
+}
+
+/**
+ *
+ */
+int tps_reinsert_contact(sip_msg_t *msg, tps_data_t *ptsd, str *hbody)
+{
+	str hname = str_init("Contact");
+	
+	if(tps_add_headers(msg, &hname, hbody, 0)<0) {
+		return -1;
+	}
+
+	return 0;
+}
+
+
 /**
  *
  */
 int tps_request_sent(sip_msg_t *msg, int dialog, int direction, int local)
 {
+	tps_data_t mtsd;
+	tps_data_t stsd;
+	tps_data_t *ptsd;
+	str lkey;
+
+	memset(&mtsd, 0, sizeof(tps_data_t));
+	memset(&stsd, 0, sizeof(tps_data_t));
+	ptsd = &mtsd;
+
+	if(tps_pack_request(msg, &mtsd)<0) {
+		LM_ERR("failed to extract and pack the headers\n");
+		return -1;
+	}
+
+	if(direction==TPS_DIR_DOWNSTREAM) {
+		lkey = get_from(msg)->tag_value;
+	} else {
+		lkey = get_to(msg)->tag_value;
+	}
+	tps_storage_lock_get(&lkey);
+	if(dialog==0) {
+		if(tps_storage_record(msg, ptsd)<0) {
+			goto error;
+		}
+	}
+
+	/* local generated requests */
+	if(local) {
+		/* ACK and CANCEL go downstream */
+		if(get_cseq(msg)->method_id==METHOD_ACK
+				|| get_cseq(msg)->method_id==METHOD_CANCEL
+				|| local==2) {
+			// th_mask_callid(&msg);
+			goto done;
+		} else {
+			/* should be for upstream */
+			goto done;
+		}
+	}
+
+	tps_remove_headers(msg, HDR_RECORDROUTE_T);
+	tps_remove_headers(msg, HDR_CONTACT_T);
+	tps_remove_headers(msg, HDR_VIA_T);
+
+	tps_reinsert_via(msg, ptsd, &ptsd->x_via1);
+	if(direction==TPS_DIR_UPSTREAM) {
+		tps_reinsert_contact(msg, ptsd, &ptsd->as_contact);
+	} else {
+		tps_reinsert_contact(msg, ptsd, &ptsd->bs_contact);
+	}
+
+done:
+	tps_storage_lock_release(&lkey);
 	return 0;
+
+error:
+	tps_storage_lock_release(&lkey);
+	return -1;
 }
 
 /**

modules/topos/tps_storage.c

@@ -27,11 +27,78 @@
  */
 
 #include <string.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
 
 #include "../../dprint.h"
+#include "../../hashes.h"
+#include "../../locking.h"
+#include "../../parser/parse_uri.h"
+
+#include "../../lib/srdb1/db.h"
+#include "../../lib/srutils/sruid.h"
 
 #include "tps_storage.h"
 
+extern sruid_t _tps_sruid;
+
+#define TPS_STORAGE_LOCK_SIZE	1<<8
+static gen_lock_set_t *_tps_storage_lock_set = NULL;
+
+/**
+ *
+ */
+int tps_storage_lock_set_init(void)
+{
+	_tps_storage_lock_set = lock_set_alloc(TPS_STORAGE_LOCK_SIZE);
+	if(_tps_storage_lock_set==NULL
+			|| lock_set_init(_tps_storage_lock_set)==NULL) {
+		LM_ERR("cannot initiate lock set\n");
+		return -1;
+	}
+	return 0;
+}
+
+/**
+ *
+ */
+int tps_storage_lock_get(str *lkey)
+{
+	uint32_t pos;
+	pos = core_case_hash(lkey, 0, TPS_STORAGE_LOCK_SIZE);
+	LM_DBG("tps lock get: %u\n", pos);
+	lock_set_get(_tps_storage_lock_set, pos);
+	return 1;
+}
+
+/**
+ *
+ */
+int tps_storage_lock_release(str *lkey)
+{
+	uint32_t pos;
+	pos = core_case_hash(lkey, 0, TPS_STORAGE_LOCK_SIZE);
+	LM_DBG("tps lock release: %u\n", pos);
+	lock_set_release(_tps_storage_lock_set, pos);
+	return 1;
+}
+
+/**
+ *
+ */
+int tps_storage_lock_set_destroy(void)
+{
+	if(_tps_storage_lock_set!=NULL) {
+		lock_set_destroy(_tps_storage_lock_set);
+		lock_set_dealloc(_tps_storage_lock_set);
+	}
+	_tps_storage_lock_set = NULL;
+	return 0;
+}
+
 /**
  *
  */
@@ -80,3 +147,94 @@ int tps_storage_branch_rm(sip_msg_t *msg, tps_data_t *td)
 {
 	return 0;
 }
+
+/**
+ *
+ */
+int tps_storage_fill_contact(sip_msg_t *msg, tps_data_t *td, int dir)
+{
+	str sv;
+	sip_uri_t puri;
+
+	sruid_next(&_tps_sruid);
+
+	if(dir==TPS_DIR_DOWNSTREAM) {
+		sv = td->bs_contact;
+	} else {
+		sv = td->as_contact;
+	}
+	if(td->cp + 8 + (2*_tps_sruid.uid.len) + sv.len >= td->cbuf + TPS_DATA_SIZE) {
+		LM_ERR("insufficient data buffer\n");
+		return -1;
+	}
+	if (parse_uri(sv.s, sv.len, &puri) < 0) {
+		LM_ERR("failed to parse the uri\n");
+		return -1;
+	}
+	if(dir==TPS_DIR_DOWNSTREAM) {
+		td->b_uuid.s = td->cp;
+		*td->cp = 'b';
+		td->cp++;
+		memcpy(td->cp, _tps_sruid.uid.s, _tps_sruid.uid.len);
+		td->cp += _tps_sruid.uid.len;
+		td->b_uuid.len = td->cp - td->b_uuid.s;
+
+		td->bs_contact.s = td->cp;
+	} else {
+		td->a_uuid.s = td->cp;
+		*td->cp = 'a';
+		td->cp++;
+		memcpy(td->cp, _tps_sruid.uid.s, _tps_sruid.uid.len);
+		td->cp += _tps_sruid.uid.len;
+		td->a_uuid.len = td->cp - td->a_uuid.s;
+
+		td->as_contact.s = td->cp;
+	}
+	*td->cp = '<';
+	td->cp++;
+	if(dir==TPS_DIR_DOWNSTREAM) {
+		*td->cp = 'b';
+	} else {
+		*td->cp = 'a';
+	}
+	td->cp++;
+	memcpy(td->cp, _tps_sruid.uid.s, _tps_sruid.uid.len);
+	td->cp += _tps_sruid.uid.len;
+	*td->cp = '@';
+	td->cp++;
+	memcpy(td->cp, puri.host.s, puri.host.len);
+	td->cp += puri.host.len;
+	if(puri.port.len>0) {
+		*td->cp = ':';
+		td->cp++;
+		memcpy(td->cp, puri.port.s, puri.port.len);
+		td->cp += puri.port.len;
+	}
+	if(puri.transport_val.len>0) {
+		memcpy(td->cp, ";transport=", 11);
+		td->cp += 11;
+		memcpy(td->cp, puri.transport_val.s, puri.transport_val.len);
+		td->cp += puri.transport_val.len;
+	}
+
+	*td->cp = '>';
+	td->cp++;
+	if(dir==TPS_DIR_DOWNSTREAM) {
+		td->bs_contact.len = td->cp - td->bs_contact.s;
+	} else {
+		td->as_contact.len = td->cp - td->as_contact.s;
+	}
+	return 0;
+}
+
+/**
+ *
+ */
+int tps_storage_record(sip_msg_t *msg, tps_data_t *td)
+{
+	int ret;
+
+	ret = tps_storage_fill_contact(msg, td, TPS_DIR_DOWNSTREAM);
+	if(ret<0) return ret;
+	return tps_storage_fill_contact(msg, td, TPS_DIR_UPSTREAM);
+}

modules/topos/tps_storage.h

@@ -31,6 +31,9 @@
 
 #include "../../parser/msg_parser.h"
 
+#define TPS_DIR_DOWNSTREAM	0
+#define TPS_DIR_UPSTREAM	1
+
 #define TPS_DATA_SIZE	4096
 typedef struct tps_data {
 	char cbuf[TPS_DATA_SIZE];
@@ -42,6 +45,11 @@ typedef struct tps_data {
 	str b_rr;
 	str a_contact;
 	str b_contact;
+	str as_contact;
+	str bs_contact;
+	str x_via1;
+	str x_via2;
+	str x_vbranch1;
 } tps_data_t;
 
 int tps_storage_dialog_find(sip_msg_t *msg, tps_data_t *td);
@@ -52,4 +60,11 @@ int tps_storage_branch_find(sip_msg_t *msg, tps_data_t *td);
 int tps_storage_branch_save(sip_msg_t *msg, tps_data_t *td);
 int tps_storage_branch_rm(sip_msg_t *msg, tps_data_t *td);
 
+int tps_storage_record(sip_msg_t *msg, tps_data_t *td);
+
+int tps_storage_lock_set_init(void);
+int tps_storage_lock_get(str *lkey);
+int tps_storage_lock_release(str *lkey);
+int tps_storage_lock_set_destroy(void);
+
 #endif