enhance message sending with buffer capture for tracing

forward.h

@@ -87,9 +87,10 @@ int forward_reply( struct sip_msg* msg);
  * \param len - the length of the message to be sent
  * \return 0 if ok, -1 on error
  */
-static inline int msg_send( const struct socket_info* send_sock, int proto,
+static inline int msg_send_ex( const struct socket_info* send_sock, int proto,
 							union sockaddr_union* to, unsigned int id,
-							char* buf, int len, struct sip_msg* msg)
+							char* buf, int len, struct sip_msg* msg,
+							str *sent_buffer)
 {
 	str out_buff;
 	unsigned short port;
@@ -107,6 +108,10 @@ static inline int msg_send( const struct socket_info* send_sock, int proto,
 
 	out_buff.len = len;
 	out_buff.s = buf;
+	if (sent_buffer) {
+		sent_buffer->s = NULL;
+		sent_buffer->len = 0;
+	}
 
 	/* determine the send socket */
 	if (send_sock==0)
@@ -136,16 +141,32 @@ static inline int msg_send( const struct socket_info* send_sock, int proto,
 		goto error;
 	}
 
-	/* potentially allocated by the out raw processing */
-	if (out_buff.s != buf)
+	/* Return the final post-raw SIP buffer passed to the protocol transport.
+	 * Protocol-specific framing or encryption happens inside tran.send(), so
+	 * observers receive the plaintext SIP bytes.  An unchanged buffer is
+	 * borrowed from the caller; a pkg-allocated replacement is transferred. */
+	if (sent_buffer)
+		*sent_buffer = out_buff;
+	else if (out_buff.s != buf)
 		pkg_free(out_buff.s);
 
 	return 0;
 error:
 	if (out_buff.s != buf)
 		pkg_free(out_buff.s);
+	if (sent_buffer) {
+		sent_buffer->s = NULL;
+		sent_buffer->len = 0;
+	}
 	return -1;
 }
 
+static inline int msg_send( const struct socket_info* send_sock, int proto,
+							union sockaddr_union* to, unsigned int id,
+							char* buf, int len, struct sip_msg* msg)
+{
+	return msg_send_ex(send_sock, proto, to, id, buf, len, msg, NULL);
+}
+
 
 #endif

modules/tm/t_cancel.c

@@ -132,11 +132,15 @@ void cancel_branch( struct cell *t, int branch )
 		tcp_no_new_conn = 1;
 	backup_list = set_avp_list( &t->user_avps );
 	set_bavp_list(&TM_BRANCH(t,branch).user_avps);
-	if (SEND_BUFFER( crb )==0) {
-		if ( has_tran_tmcbs( t, TMCB_MSG_SENT_OUT) ) {
-			set_extra_tmcb_params( &crb->buffer, &crb->dst);
+	{
+		str sent_buffer = STR_NULL;
+		int observe_send = has_tran_tmcbs(t, TMCB_MSG_SENT_OUT);
+		if (SEND_BUFFER_CAPTURE(crb,
+				observe_send ? &sent_buffer : NULL) == 0 && observe_send) {
+			set_extra_tmcb_params(&sent_buffer, &crb->dst);
 			run_trans_callbacks( TMCB_MSG_SENT_OUT,
 				t, t->uas.request, 0, 0);
+			release_sent_buffer(&sent_buffer, crb->buffer.s);
 		}
 	}
 	set_avp_list(backup_list);
@@ -168,5 +172,3 @@ char *build_cancel(struct cell *Trans,unsigned int branch,
 	 * (by t_should_relay_response) which may lead into races ( building the
 	 * cancel versus handling a final response in a different process )*/
 }
-
-

modules/tm/t_funcs.c

@@ -62,11 +62,11 @@ int send_pr_buffer( struct retr_buf *rb, void *buf, int len,
 #ifdef EXTRA_DEBUG
 					char* file, const char *function, int line,
 #endif
-					void* ctx)
+					void* ctx, str *sent_buffer)
 {
 	if (buf && len && rb )
-		return msg_send( rb->dst.send_sock, rb->dst.proto, &rb->dst.to,
-				         rb->dst.proto_reserved1, buf, len, ctx);
+		return msg_send_ex( rb->dst.send_sock, rb->dst.proto, &rb->dst.to,
+				         rb->dst.proto_reserved1, buf, len, ctx, sent_buffer);
 	else {
 #ifdef EXTRA_DEBUG
 		LM_CRIT("sending an empty buffer from %s: %s (%d)\n",file,
@@ -280,4 +280,3 @@ int t_relay_to( struct sip_msg  *p_msg , struct proxy_l *proxy, int flags)
 done:
 	return ret;
 }
-

modules/tm/t_funcs.h

@@ -85,25 +85,47 @@ extern int noisy_ctimer;
 */
 #ifdef EXTRA_DEBUG
 int send_pr_buffer( struct retr_buf *rb,
-	void *buf, int len, char* file, const char *function, int line, void* ctx);
+	void *buf, int len, char* file, const char *function, int line, void* ctx,
+	str *sent_buffer);
 #define SEND_PR_BUFFER(_rb,_bf,_le ) \
-	send_pr_buffer( (_rb), (_bf), (_le), __FILE__,  __FUNCTION__, __LINE__, NULL)
+	send_pr_buffer( (_rb), (_bf), (_le), __FILE__,  __FUNCTION__, __LINE__, NULL, NULL)
+#define SEND_PR_BUFFER_CAPTURE(_rb,_bf,_le,_out) \
+	send_pr_buffer( (_rb), (_bf), (_le), __FILE__, __FUNCTION__, __LINE__, NULL, (_out))
 #define SEND_PR_CONTEXTS_BUFFER(_rb,_bf,_le, _ctx ) \
-	send_pr_buffer( (_rb), (_bf), (_le), __FILE__, __FUNCTION, __LINE__ ,_ctx)
+	send_pr_buffer( (_rb), (_bf), (_le), __FILE__, __FUNCTION__, __LINE__ ,_ctx, NULL)
 #else
-int send_pr_buffer( struct retr_buf *rb, void *buf, int len, void* ctx);
+int send_pr_buffer( struct retr_buf *rb, void *buf, int len, void* ctx,
+	str *sent_buffer);
 #define SEND_PR_BUFFER(_rb,_bf,_le ) \
-	send_pr_buffer( (_rb), (_bf), (_le), NULL)
+	send_pr_buffer( (_rb), (_bf), (_le), NULL, NULL)
+#define SEND_PR_BUFFER_CAPTURE(_rb,_bf,_le,_out) \
+	send_pr_buffer( (_rb), (_bf), (_le), NULL, (_out))
 #define SEND_PR_CONTEXTS_BUFFER(_rb,_bf,_le, _ctx ) \
-	send_pr_buffer( (_rb), (_bf), (_le), _ctx)
+	send_pr_buffer( (_rb), (_bf), (_le), _ctx, NULL)
 #endif
 
 #define SEND_BUFFER( _rb ) \
 	SEND_PR_BUFFER( (_rb) , (_rb)->buffer.s , (_rb)->buffer.len )
 
+#define SEND_BUFFER_CAPTURE( _rb, _out ) \
+	SEND_PR_BUFFER_CAPTURE( (_rb), (_rb)->buffer.s, (_rb)->buffer.len, (_out) )
+
 #define SEND_CONTEXTS_BUFFER( _rb, ctx) \
 	SEND_PR_CONTEXTS_BUFFER( (_rb) , (_rb)->buffer.s, (_rb)->buffer.len, ctx)
 
+/* msg_send_ex() lends the input buffer when raw processing leaves it
+ * unchanged, and transfers ownership of a pkg-allocated replacement.  A
+ * captured buffer may only be exposed synchronously before this function is
+ * called; callbacks which retain it must make their own copy.  Call this on
+ * every successful send for which capture was requested. */
+static inline void release_sent_buffer(str *sent_buffer, const char *input)
+{
+	if (sent_buffer->s && sent_buffer->s != input)
+		pkg_free(sent_buffer->s);
+	sent_buffer->s = NULL;
+	sent_buffer->len = 0;
+}
+
 
 #define UNREF_UNSAFE(_T_cell) do { \
 	((_T_cell)->ref_count--);\
@@ -226,4 +248,3 @@ int t_relay_to( struct sip_msg  *p_msg, struct proxy_l *proxy, int replicate);
 int tm_has_request_disponsition_no_cancel(struct sip_msg *msg);
 
 #endif
-

modules/tm/t_fwd.c

@@ -875,8 +875,11 @@ int t_forward_nonack( struct cell *t, struct sip_msg* p_msg ,
 	success_branch=0;
 	for (i=t->first_branch; i<t->nr_of_outgoings; i++) {
 		if ( BRANCH_BM_TST_IDX(added_branches, i) ) {
+			str sent_buffer = STR_NULL;
+			int observe_send;
 
 			uac = & TM_BRANCH( t, i);
+			observe_send = has_tran_tmcbs(t, TMCB_MSG_SENT_OUT);
 
 			if (uac->br_flags & tcp_no_new_conn_bflag)
 				tcp_no_new_conn = 1;
@@ -904,7 +907,8 @@ int t_forward_nonack( struct cell *t, struct sip_msg* p_msg ,
 							&uac->request.dst);
 					run_trans_callbacks(TMCB_PRE_SEND_BUFFER, t, p_msg, 0, i);
 
-					if (SEND_BUFFER( &uac->request)==0) {
+					if (SEND_BUFFER_CAPTURE(&uac->request,
+							observe_send ? &sent_buffer : NULL) == 0) {
 						reset_bavp_list();
 						ser_error = 0;
 						break;
@@ -940,11 +944,11 @@ int t_forward_nonack( struct cell *t, struct sip_msg* p_msg ,
 			set_kr(REQ_FWDED);
 
 			/* successfully sent out -> run callbacks */
-			if ( has_tran_tmcbs( t, TMCB_MSG_SENT_OUT) ) {
-				set_extra_tmcb_params( &uac->request.buffer,
-					&uac->request.dst);
+			if (observe_send) {
+				set_extra_tmcb_params(&sent_buffer, &uac->request.dst);
 				run_trans_callbacks( TMCB_MSG_SENT_OUT, t,
 					p_msg, 0, 0);
+				release_sent_buffer(&sent_buffer, uac->request.buffer.s);
 			}
 
 		}

modules/tm/t_reply.c

@@ -342,7 +342,8 @@ inline static int update_totag_set(struct cell *t, struct sip_msg *ok)
  *   - return 0
  *   - populate @ack_buf, for callback purposes, which *must* be SHM freed!
  */
-static int send_ack(struct sip_msg* rpl, struct cell *trans, int branch, str *ack_buf)
+static int send_ack(struct sip_msg* rpl, struct cell *trans, int branch,
+		str *ack_buf, str *sent_buf)
 {
 	str method = str_init(ACK);
 	str to;
@@ -371,7 +372,8 @@ static int send_ack(struct sip_msg* rpl, struct cell *trans, int branch, str *ac
 	set_bavp_list(&TM_BRANCH(trans,branch).user_avps);
 	backup_list = set_avp_list( &trans->user_avps );
 
-	rc = SEND_PR_BUFFER(&TM_BRANCH(trans,branch).request, ack_buf->s, ack_buf->len);
+	rc = SEND_PR_BUFFER_CAPTURE(&TM_BRANCH(trans,branch).request,
+		ack_buf->s, ack_buf->len, sent_buf);
 
 	set_avp_list(backup_list);
 	reset_bavp_list();
@@ -393,6 +395,8 @@ static int _reply_light( struct cell *trans, char* buf, unsigned int len,
 	unsigned int buf_len;
 	branch_bm_t cancel_bitmap = BRANCH_BM_ZERO;
 	str cb_s;
+	str sent_buffer = STR_NULL;
+	int observe_send;
 
 	if (!buf)
 	{
@@ -500,16 +504,18 @@ static int _reply_light( struct cell *trans, char* buf, unsigned int len,
 	}
 
 
-	if ( SEND_PR_BUFFER( rb, buf, len )==0 ) {
+	observe_send = has_tran_tmcbs(trans, TMCB_MSG_SENT_OUT);
+	if (SEND_PR_BUFFER_CAPTURE(rb, buf, len,
+			observe_send ? &sent_buffer : NULL) == 0) {
 		LM_DBG("reply sent out. buf=%p: %.9s..., "
 			"shmem=%p: %.9s\n", buf, buf, rb->buffer.s, rb->buffer.s );
 
-		if (has_tran_tmcbs(trans, TMCB_MSG_SENT_OUT) ) {
-			cb_s.s = buf;
-			cb_s.len = len;
+		if (observe_send) {
+			cb_s = sent_buffer;
 			set_extra_tmcb_params( &cb_s, &rb->dst);
 			run_trans_callbacks( TMCB_MSG_SENT_OUT, trans,
 				NULL, FAKED_REPLY, code);
+			release_sent_buffer(&sent_buffer, buf);
 		}
 		stats_trans_rpl( code, 1 /*local*/ );
 	}
@@ -1146,6 +1152,8 @@ int t_retransmit_reply( struct cell *t )
 	static char b[BUF_SIZE];
 	int len;
 	str cb_s;
+	str sent_buffer = STR_NULL;
+	int observe_send;
 
 	/* we need to lock the transaction as messages from
 	   upstream may change it continuously */
@@ -1175,16 +1183,18 @@ int t_retransmit_reply( struct cell *t )
 	if (t->uas.request && t->uas.request->flags & tcp_no_new_conn_rplflag)
 		tcp_no_new_conn = 1;
 
-	if (SEND_PR_BUFFER( & t->uas.response, b, len )==0) {
+	observe_send = has_tran_tmcbs(t, TMCB_MSG_SENT_OUT);
+	if (SEND_PR_BUFFER_CAPTURE(&t->uas.response, b, len,
+			observe_send ? &sent_buffer : NULL) == 0) {
 		/* success */
 		LM_DBG("buf=%p: %.9s..., shmem=%p: %.9s\n",b, b,
 			t->uas.response.buffer.s, t->uas.response.buffer.s );
-		if (has_tran_tmcbs( t, TMCB_MSG_SENT_OUT) ) {
-			cb_s.s = b;
-			cb_s.len = len;
+		if (observe_send) {
+			cb_s = sent_buffer;
 			set_extra_tmcb_params( &cb_s, &t->uas.response.dst);
 			run_trans_callbacks( TMCB_MSG_SENT_OUT, t,
 					NULL, FAKED_REPLY, t->uas.status);
+			release_sent_buffer(&sent_buffer, b);
 		}
 	}
 
@@ -1301,6 +1311,8 @@ enum rps relay_reply( struct cell *t, struct sip_msg *p_msg, int branch,
 	struct retr_buf *uas_rb;
 	str cb_s;
 	str text;
+	str sent_buffer = STR_NULL;
+	int observe_send;
 
 	/* keep compiler warnings about use of uninit vars silent */
 	res_len=0;
@@ -1442,17 +1454,19 @@ enum rps relay_reply( struct cell *t, struct sip_msg *p_msg, int branch,
 			tcp_no_new_conn = 1;
 
 		/* send it out*/
-		if (SEND_PR_BUFFER( uas_rb, buf, res_len)==0) {
+		observe_send = has_tran_tmcbs(t, TMCB_MSG_SENT_OUT);
+		if (SEND_PR_BUFFER_CAPTURE(uas_rb, buf, res_len,
+				observe_send ? &sent_buffer : NULL) == 0) {
 			/* success */
 			LM_DBG("sent buf=%p: %.9s..., shmem=%p: %.9s\n",
 				buf, buf, uas_rb->buffer.s, uas_rb->buffer.s );
 
-			if (has_tran_tmcbs( t, TMCB_MSG_SENT_OUT) ) {
-				cb_s.s = buf;
-				cb_s.len = res_len;
+			if (observe_send) {
+				cb_s = sent_buffer;
 				set_extra_tmcb_params( &cb_s, &uas_rb->dst);
 				run_trans_callbacks( TMCB_MSG_SENT_OUT, t,
 					NULL, relayed_msg, relayed_code);
+				release_sent_buffer(&sent_buffer, buf);
 			}
 		}
 
@@ -1657,6 +1671,8 @@ int reply_received( struct sip_msg  *p_msg )
 	unsigned int has_reply_route;
 	int old_route_type, ack_sent = 0;
 	str ack_buf;
+	str sent_ack_buf = STR_NULL;
+	int observe_ack;
 
 	set_t(T_UNDEFINED);
 
@@ -1709,9 +1725,11 @@ int reply_received( struct sip_msg  *p_msg )
 	/* acknowledge negative INVITE replies ASAP! (do it before detailed
 	 * on_reply processing, which may take very long, like if it
 	 * is attempted to establish a TCP connection to a fail-over dst */
+	observe_ack = has_tran_tmcbs(t, TMCB_MSG_SENT_OUT);
 	if (is_invite(t) && ((msg_status >= 300) ||
 	(is_local(t) && !no_autoack(t) && msg_status >= 200) )) {
-		if (!(ack_sent = (send_ack(p_msg, t, branch, &ack_buf)>=0)))
+		if (!(ack_sent = (send_ack(p_msg, t, branch, &ack_buf,
+				observe_ack ? &sent_ack_buf : NULL)>=0)))
 			LM_ERR("failed to send ACK (local=%s)\n", is_local(t)?"yes":"no");
 	}
 
@@ -1726,10 +1744,11 @@ int reply_received( struct sip_msg  *p_msg )
 			p_msg->REPLY_STATUS);
 
 	if (ack_sent) {
-		if ( has_tran_tmcbs( t, TMCB_MSG_SENT_OUT) ) {
-			set_extra_tmcb_params( &ack_buf, &uac->request.dst);
+		if (observe_ack) {
+			set_extra_tmcb_params(&sent_ack_buf, &uac->request.dst);
 			run_trans_callbacks( TMCB_MSG_SENT_OUT,
 				t, t->uas.request, 0, 0);
+			release_sent_buffer(&sent_ack_buf, ack_buf.s);
 		}
 		shm_free(ack_buf.s);
 	}

modules/tm/timer.c

@@ -295,18 +295,24 @@ inline static void retransmission_handler( struct timer_link *retr_tl )
 			LM_DBG("retransmission_handler : request resending"
 				" (t=%p, %.9s ... )\n", r_buf->my_T, r_buf->buffer.s);
 			set_t(r_buf->my_T);
-			if (SEND_BUFFER( r_buf )==0) {
-				if ( has_tran_tmcbs( r_buf->my_T, TMCB_MSG_SENT_OUT) ) {
-					set_extra_tmcb_params( &r_buf->buffer, &r_buf->dst);
+			{
+				str sent_buffer = STR_NULL;
+				int observe_send = has_tran_tmcbs(r_buf->my_T,
+					TMCB_MSG_SENT_OUT);
+				if (SEND_BUFFER_CAPTURE(r_buf,
+						observe_send ? &sent_buffer : NULL) == 0 &&
+						observe_send) {
+					set_extra_tmcb_params(&sent_buffer, &r_buf->dst);
 					run_trans_callbacks( TMCB_MSG_SENT_OUT, r_buf->my_T,
 						r_buf->my_T->uas.request, 0, 0);
+					release_sent_buffer(&sent_buffer, r_buf->buffer.s);
 				}
+			}
 			/*} else {
 				reset_timer( &r_buf->fr_timer );
 				fake_reply(r_buf->my_T, r_buf->branch, 503 );
 				return;
 			} */
-			}
 
 			set_t(T_UNDEFINED);
 			switch(r_buf->retr_list) {
@@ -1134,6 +1140,7 @@ void timer_routine(unsigned int ticks , void *set)
 
 
 
+
 void utimer_routine(utime_t uticks , void *set)
 {
 	struct timer_link *tl, *tmp_tl;
@@ -1166,4 +1173,3 @@ void utimer_routine(utime_t uticks , void *set)
 	    "now at %d%%+ capacity, inuse_transactions: %lu", (int)(TM_TIMER_LOAD_WARN*100),
 	    (unsigned long)get_stat_val(tm_trans_inuse));
 }
-