/* $Id: seas_action.c 2346 2007-06-08 16:56:40Z eliasbaixas $ * * Copyright (C) 2006-2007 VozTelecom Sistemas S.L * * This file is part of openser, a free SIP server. * * openser 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 * * openser 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 #include #include #include #include #include #include /*assert*/ #include "../../mem/mem.h" #include "../../dprint.h" #include "../../str.h" #include "../../pt.h"/*process_count*/ #include "../../ip_addr.h" #include "../../tags.h" #include "../../error.h" #include "../../parser/hf.h" #include "../../parser/parse_fline.h" #include "../../parser/parser_f.h"/*find_not_quoted*/ #include "../../parser/parse_to.h" #include "../../parser/parse_from.h" #include "../../parser/parse_cseq.h" #include "../../parser/parse_content.h" #include "../../parser/parse_rr.h"/*parse_rr*/ #include "../../parser/parse_via.h"/*parse_via*/ #include "../../parser/parse_param.h"/*parse_params*/ #include "../../parser/parse_uri.h" /*parse_uri*/ #include "../../parser/msg_parser.h" #include "encode_msg.h" #include "../tm/t_lookup.h" #include "../tm/h_table.h" #include "../tm/dlg.h" #include "seas.h" #include "statistics.h" #include "seas_action.h" #include "seas_error.h" #include "ha.h" #define MAX_HEADER 1024 #define SPIRAL_HDR "X-WeSIP-SPIRAL: true" #define SPIRAL_HDR_LEN (sizeof(SPIRAL_HDR)-1) #define RECORD_ROUTE "Record-Route: " #define RECORD_ROUTE_LEN (sizeof(RECORD_ROUTE)-1) #define VIA "Via: " #define VIA_LEN (sizeof(VIA)-1) extern char *seas_tag_suffix; extern char seas_tags[]; pid_t my_parent; extern int fifo_pid; static inline int shm_str_dup(str* _d, str* _s); static inline struct sip_msg *parse_ac_msg(hdr_flags_t flags,char *start,int len); static inline void free_sip_msg_lite(struct sip_msg *my_msg); static inline int calculate_hooks(dlg_t* _d); static inline int process_input(int fd); static inline int process_pings(struct ha *the_table); static inline int ac_jain_pong(as_p the_as,char *action,int len); int process_pong(struct ha *the_table,unsigned int seqno); int print_local_uri(as_p as,char processor_id,char *where,int len); int dispatch_actions() { int fd,n,ret,timeout,elapsed_ms; static int ktimeout; struct pollfd fds[1]; struct timeval last,now; /* now the process_no is set, I delete the pt (process_table) global var, because it confuses * LOG() */ pt=0; fd=my_as->u.as.action_fd; fds[0].fd=fd; fds[0].events=POLLIN|POLLHUP; fds[0].revents=0; my_parent=getppid(); snprintf(whoami,MAX_WHOAMI_LEN,"[%.*s] Action dispatcher",my_as->name.len,my_as->name.s); if(jain_ping_timeout && servlet_ping_timeout) ktimeout=jain_ping_timeoutu.as.ac_buffer.s = pkg_malloc(AS_BUF_SIZE))==0){ LOG(L_ERR,"unable to alloc pkg mem for the action buffer\n"); return -1; } my_as->u.as.ac_buffer.len=0; if(use_ha){ timeout=ktimeout; while(1){ gettimeofday(&last,NULL); print_pingtable(&my_as->u.as.jain_pings,-1,1); if(0>(n=poll(fds,1,timeout))){ if(errno==EINTR){ gettimeofday(&last,NULL); continue; }else if(errno==EBADF){ LOG(L_ERR,"EBADF !!\n"); }else{ LOG(L_ERR,"on poll\n"); } }else if(n==0){/*timeout*/ if (0>(ret=process_pings(&my_as->u.as.jain_pings))) { return ret; } timeout=ktimeout; }else{ /*events*/ if (0>(ret=process_input(fd))) { return ret; } gettimeofday(&now,NULL); elapsed_ms=((now.tv_sec-last.tv_sec)*1000)+((now.tv_usec-last.tv_usec)/1000); if(elapsed_ms(ret=process_pings(&my_as->u.as.jain_pings))){ return ret; } timeout=ktimeout; } } fds[0].events=POLLIN|POLLHUP; fds[0].revents=0; } }else{ do{ ret=process_input(fd); }while(ret>=0); } return 0; } static inline int process_input(int fd) { int j,k; k=AS_BUF_SIZE-(my_as->u.as.ac_buffer.len); again: if(0>(j=read(fd,my_as->u.as.ac_buffer.s+my_as->u.as.ac_buffer.len,k))){ if(errno==EINTR) goto again; LOG(L_ERR,"reading data for as %.*s (%s)\n",my_as->name.len,my_as->name.s,strerror(errno)); return -1; }else if(j==0){ pkg_free(my_as->u.as.ac_buffer.s); close(fd); LOG(L_ERR,"read 0 bytes from AS:%.*s\n",my_as->name.len,my_as->name.s); /** we return, so we will exit, so our parent (Event Dispatcher) will receive a sigchld and know * it should tear down the corresponding AS * what still is not clear is what will happen to events that were put in the pipe... */ return -2; } (my_as->u.as.ac_buffer.len)+=j; LOG(L_DBG,"read %d bytes from AS action socket (total = %d)\n",j,my_as->u.as.ac_buffer.len); if(use_stats) receivedplus(); if(my_as->u.as.ac_buffer.len>5){ process_action(&my_as->u.as); LOG(L_DBG,"(Action dispatched,buffer.len=%d)\n",my_as->u.as.ac_buffer.len); } return 0; } /** * The ha structure (high availability) uses a circular (ring) buffer. A linked * list could be used, but it would involve a lot of shm_malloc/free, and this * would involve a lot of shm-lock_get/release, which would interfere a lot * with all the SER processes. With a this ring buffer, the lock_get/release only * involve the SEAS processes. * This function scans the ping structures in the buffer, computing the elapsed time * from when the ping was sent, so if the ping has timed out, it increases the * timed_out_pings counter. All the timed-out pings are removed from the buffer (the * begin index is incremented). Because the pings are added always at the end * of the buffer, they will always be ordered in increasing time, so when we find one ping * that has not timed out, the following pings will neither be. * */ static inline int process_pings(struct ha *the_table) { int i,k,elapsed; struct ping *tmp; struct timeval now; tmp=NULL; gettimeofday(&now,NULL); if(the_table->count==0) return 0; lock_get(the_table->mutex); { print_pingtable(the_table,-1,0); for(i=0;icount;i++){ k=(the_table->begin+i)%the_table->size; tmp=the_table->pings+k; elapsed=(now.tv_sec-tmp->sent.tv_sec)*1000+(now.tv_usec-tmp->sent.tv_usec)/1000; if(elapsed>the_table->timeout){ LOG(L_DBG,"ping timed out %d\n",tmp->id); the_table->timed_out_pings++; }else{ the_table->begin=k; the_table->count-=i; break; } } } lock_release(the_table->mutex); return 0; } /* Because TransactionModule organizes statistics based on process_no, * and process_no are only assigned to SER processes (not to Action dispatchers like us ;) * we have to simulate we are the FIFO process, so TM thinks that the transactions WE put * are put by the fifo process... static inline void set_process_no() { int pcnt,i; pcnt=process_count(); for(i=0;iac_buffer.s[0]<<24)|(the_as->ac_buffer.s[1]<<16)|(the_as->ac_buffer.s[2]<<8)|((the_as->ac_buffer.s[3])&0xFF); /*yeah, it comes in network byte order*/ /*if ac_len > BUF_SIZE then a flag should be put on the AS so that the whole length * of the action is skipped, until a mechanism for handling big packets is implemented*/ if(use_stats) stats_reply(); if(ac_len>AS_BUF_SIZE){ LOG(L_WARN,"BUG:Action too big (%d)!!! should be skipped and an error returned!\n",ac_len); return -1; } while (the_as->ac_buffer.len>=ac_len) { LOG(L_DBG,"Processing action %d bytes long\n",ac_len); switch(the_as->ac_buffer.s[4]){ case REPLY_PROV: case REPLY_FIN: LOG(L_DBG,"Processing a REPLY action from AS (length=%d): %.*s\n",ac_len,the_as->name.len,the_as->name.s); ac_reply(the_as,the_as->ac_buffer.s+5,ac_len-5); break; case UAC_REQ: LOG(L_DBG,"Processing an UAC REQUEST action from AS (length=%d): %.*s\n",ac_len,the_as->name.len,the_as->name.s); ac_uac_req(the_as,the_as->ac_buffer.s+5,ac_len-5); break; case AC_CANCEL: LOG(L_DBG,"Processing a CANCEL REQUEST action from AS (length=%d): %.*s\n",ac_len,the_as->name.len,the_as->name.s); ac_cancel(the_as,the_as->ac_buffer.s+5,ac_len-5); break; case SL_MSG: LOG(L_DBG,"Processing a STATELESS MESSAGE action from AS (length=%d): %.*s\n",ac_len,the_as->name.len,the_as->name.s); ac_sl_msg(the_as,the_as->ac_buffer.s+5,ac_len-5); break; case JAIN_PONG: LOG(L_DBG,"Processing a PONG\n"); ac_jain_pong(the_as,the_as->ac_buffer.s+5,ac_len-5); break; default: LOG(L_DBG,"Processing a UNKNOWN TYPE action from AS (length=%d): %.*s\n",ac_len,the_as->name.len,the_as->name.s); break; } memmove(the_as->ac_buffer.s,the_as->ac_buffer.s+ac_len,(the_as->ac_buffer.len)-ac_len); (the_as->ac_buffer.len)-=ac_len; if(the_as->ac_buffer.len>5){ ac_len=(the_as->ac_buffer.s[0]<<24)|(the_as->ac_buffer.s[1]<<16)|(the_as->ac_buffer.s[2]<<8)|((the_as->ac_buffer.s[3])&0xFF); }else{ return 0; } } return 0; } static inline int ac_jain_pong(as_p the_as,char *action,int len) { unsigned int seqno,flags; int k; k=0; net2hostL(flags,action,k); net2hostL(seqno,action,k); process_pong(&the_as->jain_pings,seqno); return 0; } int process_pong(struct ha *the_table,unsigned int seqno) { int i,k,elapsed; struct ping *tmp; struct timeval now; gettimeofday(&now,NULL); tmp=NULL; if(the_table->count==0) return 0; lock_get(the_table->mutex); print_pingtable(the_table,-1,0); for(i=0;icount;i++){ k=(the_table->begin+i)%the_table->size; tmp=the_table->pings+k; if(tmp->id == seqno){ elapsed=(now.tv_sec-tmp->sent.tv_sec)*1000+(now.tv_usec-tmp->sent.tv_usec)/1000; LOG(L_DBG,"Ping-Pong delay: %d (timeout was:%d)\n",elapsed,the_table->timeout); if(elapsed>the_table->timeout){ /*if this ping has timed out, all the more-ancient pings will also be * timed out*/ the_table->timed_out_pings+=i; }/*anyway, when we find a ping in the table, we remove all the pings that are more ancient (if there are any..)*/ the_table->count-=(i+1); the_table->begin=(k+1)%the_table->size; break; } } lock_release(the_table->mutex); return 0; } /** * ac_cancel: * @param1: the Application Server structure which sent this action * @param2: the action payload * @param3: len the length of the payload * * This function cancels a previously initiated UAC Transaction. * it receives the HashIndex and Label of the cell being cancelled * and invokes t_cancel_uac from the transactionModule API which * cancels the transaction. * * Returns: * */ int ac_cancel(as_p the_as,char *action,int len) { unsigned int flags,ret,cancelled_hashIdx,cancelled_label,i; char processor_id; struct sip_msg *my_msg; struct as_uac_param *the_param; int k,retval,uac_id; str headers,body; body.s=headers.s=NULL; my_msg=NULL; the_param=NULL; i=k=0; net2hostL(flags,action,k); net2hostL(uac_id,action,k); processor_id=action[k++]; net2hostL(cancelled_hashIdx,action,k); net2hostL(cancelled_label,action,k); if(!(headers.s=pkg_malloc(MAX_HEADER))){ LOG(L_ERR,"Out of Memory!!"); goto error; } headers.len=0; if(!(my_msg=pkg_malloc(sizeof(struct sip_msg)))){ LOG(L_ERR,"out of memory!\n"); goto error; } memset(my_msg,0,sizeof(struct sip_msg)); my_msg->buf=action+k; my_msg->len=len-k; LOG(L_DBG,"Action UAC Message: uac_id:%d processor_id=%d, message:[%.*s]\n",uac_id,processor_id,len-4,&action[4]); if(parse_msg(action+k,len-k,my_msg)<0){ LOG(L_ERR,"parsing sip_msg"); goto error; } if(my_msg->first_line.type==SIP_REPLY){ LOG(L_ERR,"trying to create a UAC with a SIP response!!\n"); goto error; } if(parse_headers(my_msg,HDR_EOH_F,0)==-1){ LOG(L_ERR,"parsing headers\n"); goto error; } if(0>(headers.len=extract_allowed_headers(my_msg,1,-1,HDR_CONTENTLENGTH_F|HDR_ROUTE_F|HDR_TO_F|HDR_FROM_F|HDR_CALLID_F|HDR_CSEQ_F,headers.s,MAX_HEADER))) { LOG(L_ERR,"Unable to extract allowed headers!!\n"); goto error; } headers.s[headers.len]=0; /*let's get the body*/ i=(unsigned int)get_content_length(my_msg); if(i!=0){ if(!(body.s=pkg_malloc(i))){ LOG(L_ERR,"Out of Memory!"); goto error; } memcpy(body.s,get_body(my_msg),i); body.len=i; LOG(L_DBG,"Trying to construct a Sip Request with: body:%d[%s] headers:%d[%s]\n",\ body.len,body.s,headers.len,headers.s); }else{ body.s=NULL; body.len=0; } if(!(the_param=shm_malloc(sizeof(struct as_uac_param)))){ LOG(L_ERR,"out of shared memory\n"); goto error; } the_param->who=my_as; the_param->uac_id=uac_id; the_param->processor_id=processor_id; ret=seas_f.tmb.t_cancel_uac(&headers,&body,cancelled_hashIdx,cancelled_label,uac_cb,(void*)the_param); if (ret == 0) { LOG(L_ERR, "Error on t_cancel_uac\n"); as_action_fail_resp(uac_id,SE_CANCEL,SE_CANCEL_MSG,SE_CANCEL_MSG_LEN); goto error; }else{ the_param->label=ret; } retval=0; goto exit; error: retval = -1; if(the_param) shm_free(the_param); exit: if(headers.s) pkg_free(headers.s); if(body.s) pkg_free(headers.s); if(my_msg){ if(my_msg->headers) free_hdr_field_lst(my_msg->headers); pkg_free(my_msg); } return retval; } int recordroute_diff(struct sip_msg *req,struct sip_msg *resp) { struct hdr_field *hf; rr_t *rr1; int i,j,k; i=j=k=0; /* count how many record-route bodies come in the response*/ /* this does not work, I think because of siblings for(hf=resp->record_route;hf;hf=hf->sibling,j=0){ */ for(hf=resp->headers;hf;hf=hf->next,j=0){ if(hf->type != HDR_RECORDROUTE_T) continue; if(!hf->parsed){ if(0>parse_rr(hf)) goto error; j=1; } for(rr1=hf->parsed;rr1;rr1=rr1->next){ i++; } if(j){ free_rr((rr_t**)(void*)&hf->parsed); hf->parsed=NULL; } } /* for(hf=req->record_route;hf;hf=hf->sibling,j=0){ */ for(hf=req->headers;hf;hf=hf->next,j=0){ if(hf->type != HDR_RECORDROUTE_T) continue; if(!hf->parsed){ if(0>parse_rr(hf)) goto error; j=1; } for(rr1=hf->parsed;rr1;rr1=rr1->next){ k++; } if(j){ free_rr((rr_t**)(void*)&hf->parsed); hf->parsed=NULL; } } return i-k; error: return -1; } int via_diff(struct sip_msg *req,struct sip_msg *resp) { struct hdr_field *hf; struct via_body *vb; int i,j,k; i=j=k=0; /* count how many via bodies come in the response*/ for(hf=resp->h_via1;hf;hf=hf->sibling){ if(!hf->parsed){ if((vb=pkg_malloc(sizeof(struct via_body)))==0){ LOG(L_ERR,"Out of mem in via_diff!!\n"); return -1; } memset(vb,0,sizeof(struct via_body)); if(parse_via(hf->body.s,hf->body.s+hf->body.len+1,vb)==0){ LOG(L_ERR,"Unable to parse via in via_diff!\n"); pkg_free(vb); return -1; } hf->parsed=vb; j=1; } for(vb=hf->parsed;vb;vb=vb->next){ i++; } if(j){ free_via_list((struct via_body*)hf->parsed); hf->parsed=NULL; j=0; } } j=0; /* count how many via bodies were in the orig. request*/ for(hf=req->h_via1;hf;hf=hf->sibling){ if(!hf->parsed){ if((vb=pkg_malloc(sizeof(struct via_body)))==0){ goto error; } memset(vb,0,sizeof(struct via_body)); if(parse_via(hf->body.s,hf->body.s+hf->body.len+1,vb)==0){ goto error; } hf->parsed=vb; j=1; } for(vb=hf->parsed;vb;vb=vb->next){ k++; } if(j){ free_via_list((struct via_body*)hf->parsed); hf->parsed=NULL; j=0; } } return i-k; error: return -1; } /** * ac_reply: UAS transaction Reply action. It replies to an incoming request with a response. * @the_as: The App Server that sent this action. * * function description * * Returns: what */ int ac_reply(as_p the_as,char *action,int len) { unsigned int flags,hash_index,label,contentlength; struct cell *c; struct sip_msg *my_msg; struct to_body *tb; str new_header,body,totag; char *ttag; int i,k,retval; static char headers[MAX_HEADER]; contentlength=0; ttag=NULL; my_msg=NULL; i=k=0; net2hostL(flags,action,k); net2hostL(hash_index,action,k); net2hostL(label,action,k); if(seas_f.tmb.t_lookup_ident(&c,hash_index,label)<0){ LOG(L_ERR,"Failed to t_lookup_ident hash_idx=%d,label=%d\n",hash_index,label); goto error; } if(use_stats) action_stat(c); if(c->uas.status>=200){ LOG(L_ERR,"ac_reply: trying to reply to a \"%.*s\" transaction" "that is already in completed state\n",REQ_LINE(c->uas.request).method.len,REQ_LINE(c->uas.request).method.s); goto error; } if (!(my_msg=parse_ac_msg(HDR_EOH_F,action+k,len-k))) { LOG(L_ERR,"Failed to parse_ac_msg hash_idx=%d,label=%d\n",hash_index,label); goto error; } tb=(struct to_body*)my_msg->to->parsed; if(tb->tag_value.s && tb->tag_value.len){ totag=tb->tag_value; }else{ totag.s=NULL; totag.len=0; /*if(!(ttag=pkg_malloc(TOTAG_VALUE_LEN))){ LOG(L_ERR,"Out of memory !!!\n"); goto error; } totag.s=ttag; calc_crc_suffix(c->uas.request,seas_tag_suffix); LOG(L_DBG,"seas_tags = %.*s\n",TOTAG_VALUE_LEN,seas_tags); memcpy(totag.s,seas_tags,TOTAG_VALUE_LEN); totag.len=TOTAG_VALUE_LEN;*/ } LOG(L_DBG,"Using totag=[%.*s]\n",totag.len,totag.s); if(my_msg->content_length) contentlength=(unsigned int)(long)my_msg->content_length->parsed; if(0>(i=recordroute_diff(c->uas.request,my_msg))){/*not likely..*/ LOG(L_DBG,"Seems that request had more RecordRoutes than response...\n"); goto error; }else LOG(L_DBG,"Recordroute Diff = %d\n",i); if(0>(i=extract_allowed_headers(my_msg,0,i,HDR_VIA_F|HDR_TO_F|HDR_FROM_F|HDR_CSEQ_F|HDR_CALLID_F|HDR_CONTENTLENGTH_F,headers,MAX_HEADER))){ LOG(L_ERR,"ac_reply() filtering headers !\n"); goto error; } headers[i]=0; new_header.s=headers; new_header.len=i; /* If it is INVITE and response is success (>=200 && <300), we mark it as local so that * SER does NOT retransmit the final response (by default, SER retransmit local UAS final * responses...*/ if(is_invite(c) && my_msg->first_line.u.reply.statuscode>=200 && my_msg->first_line.u.reply.statuscode<300) c->flags |= T_IS_LOCAL_FLAG; /*WARNING casting unsigned int to int*/ body.len=contentlength; body.s=get_body(my_msg); LOG(L_DBG,"Trying to construct a SipReply with: ReasonPhrase:[%.*s] body:[%.*s] headers:[%.*s] totag:[%.*s]\n",\ my_msg->first_line.u.reply.reason.len,my_msg->first_line.u.reply.reason.s,\ body.len,body.s,new_header.len,new_header.s,totag.len,totag.s); /* t_reply_with_body un-ref-counts the transaction, so dont use it anymore*/ if(seas_f.tmb.t_reply_with_body(c,my_msg->first_line.u.reply.statuscode,&(my_msg->first_line.u.reply.reason),&body,&new_header,&totag)<0){ LOG(L_ERR,"Failed to t_reply\n"); goto error; } retval=0; goto exit; error: retval = -1; exit: if(ttag) pkg_free(ttag); if(my_msg){ free_sip_msg_lite(my_msg); pkg_free(my_msg); } return retval; } static inline struct sip_msg *parse_ac_msg(hdr_flags_t flags,char *start,int len) { struct sip_msg *my_msg; my_msg=NULL; if(!(my_msg=pkg_malloc(sizeof(struct sip_msg)))){ LOG(L_ERR,"ac_reply: out of memory!\n"); goto error; } memset(my_msg,0,sizeof(struct sip_msg)); my_msg->buf=start; my_msg->len=len; LOG(L_DBG,"Action Message:[%.*s]\n",len,start); if(0>parse_msg(start,len,my_msg)){ LOG(L_ERR,"parse_ac_msg: parsing sip_msg"); goto error; } if(0>parse_headers(my_msg,flags,0)){ LOG(L_ERR,"parse_ac_msg: parsing headers\n"); goto error; } return my_msg; error: if(my_msg){ free_sip_msg_lite(my_msg); pkg_free(my_msg); } return NULL; } /* Actions are composed as follows: * (the action length and type as always= 5 bytes) * * TODO performance speedup: instead of using * dynamically allocated memory for headers,body,totag,reason and my_msg * use static buffers. * */ int ac_sl_msg(as_p the_as,char *action,int len) { char processor_id; struct sip_msg *my_msg; str *uri; struct proxy_l *proxy; rr_t *my_route; int i,k,retval; unsigned int flags; enum sip_protos proto; my_msg=NULL; i=k=0; net2hostL(flags,action,k); processor_id=action[k++]; proxy=0; if(!(my_msg = parse_ac_msg(HDR_EOH_F,action+k,len-k))){ LOG(L_ERR,"out of memory!\n"); goto error; } if(my_msg->first_line.type == SIP_REQUEST) LOG(L_DBG,"forwarding request:\"%.*s\" statelessly \n",my_msg->first_line.u.request.method.len+1+\ my_msg->first_line.u.request.uri.len,my_msg->first_line.u.request.method.s); else LOG(L_DBG,"forwarding reply:\"%.*s\" statelessly \n",my_msg->first_line.u.reply.status.len+1+\ my_msg->first_line.u.reply.reason.len,my_msg->first_line.u.reply.status.s); if (my_msg->route) { if (parse_rr(my_msg->route) < 0) { LOG(L_ERR, "Error while parsing Route body\n"); goto error; } my_route = (rr_t*)my_msg->route->parsed; uri=&(my_route->nameaddr.uri); }else{ uri = GET_RURI(my_msg); } my_msg->force_send_socket=grep_sock_info(&my_msg->via1->host,my_msg->via1->port,my_msg->via1->proto); /* or also could be: my_msg->force_send_socket=the_as->binds[processor_id].bind_address; not sure which is better... */ proxy=uri2proxy(uri,PROTO_NONE); if (proxy==0) { LOG(L_ERR,"unable to create proxy from URI \n"); goto error; } proto=proxy->proto; /* uri2proxy set it correctly */ //TODO my_msg->recvd if(0>forward_sl_request(my_msg,proxy,proto)) goto error; retval=0; goto exit; error: retval = -1; exit: if(proxy){ free_proxy(proxy); pkg_free(proxy); } if(my_msg){ free_sip_msg_lite(my_msg); pkg_free(my_msg); } return retval; } static inline void free_sip_msg_lite(struct sip_msg *my_msg) { if(my_msg){ /**should do the same as in free_sip_msg() but w/o freeing my_msg->buf*/ if (my_msg->new_uri.s) { pkg_free(my_msg->new_uri.s); my_msg->new_uri.len=0; } if (my_msg->dst_uri.s) { pkg_free(my_msg->dst_uri.s); my_msg->dst_uri.len=0; } if (my_msg->path_vec.s) { pkg_free(my_msg->path_vec.s);my_msg->path_vec.len=0; } if (my_msg->headers) free_hdr_field_lst(my_msg->headers); if (my_msg->add_rm) free_lump_list(my_msg->add_rm); if (my_msg->body_lumps) free_lump_list(my_msg->body_lumps); /* this is not in lump_struct.h, and anyhow it's not supposed to be any lumps * in our messages... or is it? if (my_msg->reply_lump) free_reply_lump(my_msg->reply_lump); */ } } int forward_sl_request(struct sip_msg *msg,struct proxy_l *proxy,int proto) { union sockaddr_union *to; struct socket_info *send_sock; int ret; to = (union sockaddr_union*)pkg_malloc(sizeof(union sockaddr_union)); ret = -1; hostent2su(to, &proxy->host, proxy->addr_idx, (proxy->port)?proxy->port:SIP_PORT); do { send_sock=get_send_socket(msg, to, proto); if (send_sock==0){ LOG(L_ERR, "ERROR:seas:forward_msg: cannot forward to af %d, " "proto %d no corresponding listening socket\n", to->s.sa_family, proto); continue; } DBG("seas:sl_msg:Sending:\n%.*s.\n", (int)msg->len,msg->buf); if (msg_send(send_sock, proto, to, 0, msg->buf,msg->len)<0){ LOG(L_ERR,"ERROR:seas:forward_msg: Error sending message !!\n"); continue; } ret = 0; break; }while( get_next_su( proxy, to, 0)==0 ); pkg_free(to); return ret; } /*Actions are composed as follows: * (the action length and type as always= 5 bytes) * 4:uac_id * * int request(str* method, str* req_uri, str* to, str* from, str* headers, str* body, transaction_cb c, void* cp) * TODO performance speedup: instead of using * dynamically allocated memory for headers,body,totag,reason and my_msg * use static buffers. * */ int ac_uac_req(as_p the_as,char *action,int len) { unsigned int flags,cseq; char err_buf[MAX_REASON_LEN],processor_id; struct sip_msg *my_msg; struct to_body *fb,*tb; struct cseq_body *cseqb; struct as_uac_param *the_param; dlg_t *my_dlg; int i,k,retval,uac_id,sip_error,ret,err_ret; long clen; str headers,body,fake_uri,*tmp; headers.s=body.s=fake_uri.s=NULL; my_dlg=NULL; my_msg=NULL; the_param=NULL; i=k=clen=0; net2hostL(flags,action,k); net2hostL(uac_id,action,k); processor_id=action[k++]; if(!(headers.s=pkg_malloc(MAX_HEADER))){ LOG(L_ERR,"Out of Memory!!"); goto error; } headers.len=0; LOG(L_DBG,"Action UAC Message: uac_id:%d processor_id=%d\n",uac_id,processor_id); if (!(my_msg = parse_ac_msg(HDR_EOH_F,action+k,len-k))) { LOG(L_ERR,"out of memory!\n"); goto error; } if(my_msg->first_line.type==SIP_REPLY){ LOG(L_ERR,"trying to create a UAC with a SIP response!!\n"); goto error; } if(parse_headers(my_msg,HDR_EOH_F,0)==-1){ LOG(L_ERR,"ERROR:seas:ac_uac_req:parsing headers\n"); goto error; } if(parse_from_header(my_msg)<0){ LOG(L_ERR,"parsing from header ! \n"); goto error; } if(check_transaction_quadruple(my_msg)==0){ as_action_fail_resp(uac_id,SE_UAC,"Headers missing (to,from,call-id,cseq)?",0); LOG(L_ERR,"Headers missing (to,from,call-id,cseq)?"); goto error; } if(!(get_from(my_msg)) || !(get_from(my_msg)->tag_value.s) || !(get_from(my_msg)->tag_value.len)){ as_action_fail_resp(uac_id,SE_UAC,"From tag missing",0); LOG(L_ERR,"From tag missing"); goto error; } fb=my_msg->from->parsed; tb=my_msg->to->parsed; cseqb=my_msg->cseq->parsed; if(0!=(str2int(&cseqb->number,&cseq))){ LOG(L_DBG,"unable to parse CSeq\n"); goto error; } if(my_msg->first_line.u.request.method_value != METHOD_ACK && my_msg->first_line.u.request.method_value != METHOD_CANCEL) { /** we trick req_within */ cseq--; } tmp=&(fb->uri); if((*(tmp->s - 1) == '<') && (*(tmp->s + tmp->len)=='>')){ tmp->s--; tmp->len += 2; } tmp=&(tb->uri); if((*(tmp->s - 1) == '<') && (*(tmp->s + tmp->len)=='>')){ tmp->s--; tmp->len += 2; } if(seas_f.tmb.new_dlg_uac(&(my_msg->callid->body),&(fb->tag_value),cseq,\ &(fb->uri),&(tb->uri),&my_dlg) < 0) { as_action_fail_resp(uac_id,SE_UAC,"Error creating new dialog",0); LOG(L_ERR,"Error while creating new dialog\n"); goto error; } if(tb->tag_value.s && tb->tag_value.len) shm_str_dup(&my_dlg->id.rem_tag,&tb->tag_value); /**Awful hack: to be able to set our own CSeq, from_tag and call-ID we have * to use req_within instead of req_outside (it sets it's own CSeq,Call-ID * and ftag), so we have to simulate that the dialog is already in completed * state so... */ server_signature=0; my_dlg->state = DLG_CONFIRMED; if(0>(headers.len=extract_allowed_headers(my_msg,1,-1,HDR_CONTENTLENGTH_F|HDR_ROUTE_F|HDR_TO_F|HDR_FROM_F|HDR_CALLID_F|HDR_CSEQ_F,headers.s,MAX_HEADER))) { LOG(L_ERR,"Unable to extract allowed headers!!\n"); goto error; } headers.s[headers.len]=0; /*let's get the body*/ if(my_msg->content_length) clen=(long)get_content_length(my_msg); if(clen!=0){ if(!(body.s=pkg_malloc(clen))){ LOG(L_ERR,"Out of Memory!"); goto error; } memcpy(body.s,get_body(my_msg),clen); body.len=clen; body.s[clen]=0; LOG(L_DBG,"Trying to construct a Sip Request with: body:%d[%.*s] headers:%d[%.*s]\n",\ body.len,body.len,body.s,headers.len,headers.len,headers.s); /*t_reply_with_body un-ref-counts the transaction, so dont use it anymore*/ }else{ body.s=NULL; body.len=0; } /*Now... create the UAC !! * it would be great to know the hash_index and the label that have been assigned * to our newly created cell, but t_uac does not leave any way for us to know... * only that when that transaction transitions its state (ie. a response is received, * a timeout is reached, etc...) the callback will be called with the given parameter. * * So the only way we have to know who we are, is passing as a parameter a structure with * 2 pointers: one to the app_server and the other, the identifier of the UAC (uac_id). * */ if(!(the_param=shm_malloc(sizeof(struct as_uac_param)))){ LOG(L_ERR,"out of shared memory\n"); goto error; } the_param->who=my_as; the_param->uac_id=uac_id; the_param->processor_id=processor_id; shm_str_dup(&my_dlg->rem_target,&my_msg->first_line.u.request.uri); if (my_msg->route) { if (parse_rr(my_msg->route) < 0) { LOG(L_ERR, "Error while parsing Route body\n"); goto error; } /* TODO route_set should be a shm copy of my_msg->route->parsed */ my_dlg->route_set=(rr_t*)my_msg->route->parsed; /** this SHOULD be: shm_duplicate_rr(&my_dlg->route_set,my_msg->route->parsed); * but it will last more... */ } calculate_hooks(my_dlg); if(flags & SPIRAL_FLAG){ memcpy(headers.s+headers.len,SPIRAL_HDR CRLF,SPIRAL_HDR_LEN + CRLF_LEN); headers.len+=SPIRAL_HDR_LEN+CRLF_LEN; headers.s[headers.len]=0; fake_uri.s=pkg_malloc(200); fake_uri.len=print_local_uri(the_as,processor_id,fake_uri.s,200); if(fake_uri.len<0){ LOG(L_ERR,"printing local uri\n"); goto error; } my_dlg->hooks.next_hop=&fake_uri; } my_dlg->T_flags=T_NO_AUTOACK_FLAG|T_PASS_PROVISIONAL_FLAG ; ret=seas_f.tmb.t_request_within(&(my_msg->first_line.u.request.method),&headers,&body,my_dlg,uac_cb,(void *)the_param); /** now undo all the fakes we have put in my_dlg*/ /*because my_dlg->route_set should be shm but we fake it (its pkg_mem)*/ my_dlg->route_set=(rr_t *)0; if (ret <= 0) { err_ret = err2reason_phrase(ret,&sip_error,err_buf, sizeof(err_buf), "SEAS/UAC"); LOG(L_ERR,"Error on request_within %s\n",err_buf ); if(err_ret > 0) { as_action_fail_resp(uac_id,ret,err_buf,0); }else{ as_action_fail_resp(uac_id,E_UNSPEC,"500 SEAS/UAC error",0); } goto error; } retval=0; goto exit; error: retval = -1; if(the_param) shm_free(the_param); exit: seas_f.tmb.free_dlg(my_dlg); if(headers.s) pkg_free(headers.s); if(body.s) pkg_free(body.s); if(fake_uri.s) pkg_free(fake_uri.s); if(my_msg){ if(my_msg->headers) free_hdr_field_lst(my_msg->headers); pkg_free(my_msg); } return retval; } /** * len MUST be >0 */ int print_local_uri(as_p as,char processor_id,char *where,int len) { int i; struct socket_info *si; str proto; proto.s=NULL; proto.len=0; for(i=0;ibound_processor[i]==processor_id) break; } if(i==MAX_BINDS){ LOG(L_DBG,"processor ID not found\n"); return -1; } si=as->binds[i]; switch(si->proto){ case PROTO_UDP: proto.s=""; proto.len=0; break; case PROTO_TCP: proto.s=TRANSPORT_PARAM "TCP"; proto.len=TRANSPORT_PARAM_LEN + 3; break; case PROTO_TLS: proto.s=TRANSPORT_PARAM "TLS"; proto.len=TRANSPORT_PARAM_LEN + 3; break; case PROTO_SCTP: proto.s=TRANSPORT_PARAM "SCTP"; proto.len=TRANSPORT_PARAM_LEN + 4; break; } switch(si->address.af){ case AF_INET: i=snprintf(where,len,"sip:%d.%d.%d.%d:%u%.*s",si->address.u.addr[0],si->address.u.addr[1],\ si->address.u.addr[2],si->address.u.addr[3],si->port_no,proto.len,proto.s); break; case AF_INET6: i=snprintf(where,len,"sip:[%x:%x:%x:%x:%x:%x:%x:%x]:%u%.*s", htons(si->address.u.addr16[0]), htons(si->address.u.addr16[1]),\ htons(si->address.u.addr16[2]), htons(si->address.u.addr16[3]), htons(si->address.u.addr16[4]), htons(si->address.u.addr16[5]),\ htons(si->address.u.addr16[6]), htons(si->address.u.addr16[7]),si->port_no,proto.len,proto.s); break; default: LOG(L_ERR,"address family unknown\n"); return -1; } if(i>len){ LOG(L_ERR,"Output was truncated!!\n"); return -1; }else if(i<0){ LOG(L_ERR,"Error on snprintf\n"); return i; } return i; } /* !!! COPIED FROM MODULES/TM !! * This function skips name part * uri parsed by parse_contact must be used * (the uri must not contain any leading or * trailing part and if angle bracket were * used, right angle bracket must be the * last character in the string) * * _s will be modified so it should be a tmp * copy */ void get_raw_uri(str* _s) { char* aq; if (_s->s[_s->len - 1] == '>') { aq = find_not_quoted(_s, '<'); _s->len -= aq - _s->s + 2; _s->s = aq + 1; } } /* !!! COPIED FROM MODULES/TM !! * Calculate dialog hooks * * This is copied from modules/tm/dlg.c * * Maybe a reference to the original function in TM * could be reached via handlers or whatever... */ static inline int calculate_hooks(dlg_t* _d) { str* uri; struct sip_uri puri; param_hooks_t hooks; param_t* params; if (_d->route_set) { uri = &_d->route_set->nameaddr.uri; if (parse_uri(uri->s, uri->len, &puri) < 0) { LOG(L_ERR, "Error while parsing URI\n"); return -1; } if (parse_params(&puri.params, CLASS_URI, &hooks, ¶ms) < 0) { LOG(L_ERR, "Error while parsing parameters\n"); return -2; } free_params(params); if (hooks.uri.lr) { if (_d->rem_target.s) _d->hooks.request_uri = &_d->rem_target; else _d->hooks.request_uri = &_d->rem_uri; _d->hooks.next_hop = &_d->route_set->nameaddr.uri; _d->hooks.first_route = _d->route_set; } else { _d->hooks.request_uri = &_d->route_set->nameaddr.uri; _d->hooks.next_hop = _d->hooks.request_uri; _d->hooks.first_route = _d->route_set->next; _d->hooks.last_route = &_d->rem_target; } } else { if (_d->rem_target.s) _d->hooks.request_uri = &_d->rem_target; else _d->hooks.request_uri = &_d->rem_uri; _d->hooks.next_hop = _d->hooks.request_uri; } if ((_d->hooks.request_uri) && (_d->hooks.request_uri->s) && (_d->hooks.request_uri->len)) { _d->hooks.ru.s = _d->hooks.request_uri->s; _d->hooks.ru.len = _d->hooks.request_uri->len; _d->hooks.request_uri = &_d->hooks.ru; get_raw_uri(_d->hooks.request_uri); } if ((_d->hooks.next_hop) && (_d->hooks.next_hop->s) && (_d->hooks.next_hop->len)) { _d->hooks.nh.s = _d->hooks.next_hop->s; _d->hooks.nh.len = _d->hooks.next_hop->len; _d->hooks.next_hop = &_d->hooks.nh; get_raw_uri(_d->hooks.next_hop); } return 0; } /** * Strips the topmost via headers. * Leaves only the topmost Record-Route headers. * */ int extract_allowed_headers(struct sip_msg *my_msg,int strip_top_vias,int allow_top_Rroutes,hdr_flags_t forbidden_hdrs,char *headers,int headers_len) { struct hdr_field *hf; rr_t *rb; struct via_body *vb; int len,k,rtcnt,i; len=0; rtcnt=allow_top_Rroutes; rb=NULL; vb=NULL; for(hf=my_msg->headers;hf;hf=hf->next){ if(forbidden_hdrs & HDR_T2F(hf->type)){ LOG(L_DBG,"Skipping header (%.*s)\n",hf->name.len,hf->name.s); continue; }else if(hf->type==HDR_VIA_T && strip_top_vias > 0){ /** All vias MUST be parsed !!*/ for(i=0,vb=hf->parsed;vb;vb=vb->next,i++); if(i<=strip_top_vias){ LOG(L_DBG,"Stripping vias [%.*s]\n",hf->len,hf->name.s); /** skip this via header*/ strip_top_vias-=i; }else{ assert(i>1); vb=hf->parsed; while(strip_top_vias--) vb=vb->next; k= (hf->name.s + hf->len) - vb->name.s; LOG(L_DBG,"Stripping vias [%.*s]\n",(int)(vb->name.s-hf->name.s), hf->name.s); if(k+VIA_LENname.s,k); len+=k; }else{ LOG(L_ERR,"Out Of Space !!\n"); goto error; } } }else if(hf->type==HDR_RECORDROUTE_T && rtcnt>=0){ if(rtcnt==0) continue; if(!hf->parsed && 0>parse_rr(hf)){ LOG(L_ERR,"parsing Record-Route:\"%.*s\"\n",hf->body.len,hf->body.s); goto error; } for(i=0,rb=hf->parsed;rb;rb=rb->next,i++); if(i<=rtcnt){ if((len+hf->len)len,hf->name.s); memcpy(headers+len,hf->name.s,hf->len); len+=hf->len; }else{ LOG(L_ERR,"Unable to keep recordroute (not enough space left in headers) Discarding \"%.*s\" \n",hf->name.len,hf->name.s); goto error; } /** is this dangerous ? because the rtcnt is the control variable for this conditional 'if' * so if I change rtcnt value in one of the statements... what then ??? */ rtcnt-=i; }else{ assert(rtcnt>0); rb=hf->parsed; while(--rtcnt) rb=rb->next; k= (((rb->nameaddr.name.s) + rb->len)-hf->name.s) ; if(len+k+CRLF_LENname.s,k); LOG(L_DBG,"Allowing RecordRoute [%.*s\r\n]\n",k,hf->name.s); len+=k; memcpy(headers+len,CRLF,CRLF_LEN); len+=CRLF_LEN; }else{ LOG(L_ERR,"Out Of Space !!\n"); goto error; } } if(hf->parsed){ free_rr((rr_t **)(void*)(&hf->parsed)); hf->parsed=NULL; } }else{ if((len+hf->len)name.s,hf->len); len+=hf->len; }else{ LOG(L_WARN,"Too many headers. Discarding \"%.*s\" \n",hf->name.len,hf->name.s); } } }/*for*/ return len; error: return -1; } /* * Make a copy of a str structure using shm_malloc */ static inline int shm_str_dup(str* _d, str* _s) { _d->s = shm_malloc(_s->len); if (!_d->s) { LOG(L_ERR, "No memory left\n"); return -1; } memcpy(_d->s, _s->s, _s->len); _d->len = _s->len; return 0; } /** * ERROR action responses are composed of: * 4: the length of the event * 1: the event type (AC_RES_FAIL) * 4: NBO of the uac-action-request identification (uac_id) * 4: the sip_error code in NBO. * 1: (unsigned) the length of the string. * N: the string * */ int as_action_fail_resp(int uac_id,int sip_error,char *err_buf,int i) { char msg[14+MAX_REASON_LEN]; unsigned int k; k=4; if(i==0) i=strlen(err_buf); if(i>MAX_REASON_LEN){ LOG(L_ERR,"Error Reason bigger than MAX_REASON_LEN\n"); return -1; } msg[k++]=AC_RES_FAIL; uac_id=htonl(uac_id); memcpy(msg+k,&uac_id,4); k+=4; sip_error=htonl(sip_error); memcpy(msg+k,&sip_error,4); k+=4; msg[k++]=(char)(unsigned char)i; memcpy(msg+k,err_buf,i); k+=i; k=htonl(k); memcpy(msg,&k,4); k=ntohl(k); write(my_as->u.as.action_fd,msg,k); return 0; } /** * This function will be called from a SER process when a reply is received for * the transaction. The SER processes only have acces to the EventDispatcher * fifo (not to the ActionDispatcher) so EventDispatcher will be the one who * will send the event to the AppServer. * TODO WARNING !!! there's a clear MEMORY LEAK here, see exit: at the bottom of * the function... it should free ev_info !!!!!!!! * I have disabled the free() because It may be that we receive a retransmitted 200 OK * if the ACK gets lost, that 200 OK will make SER invoke this callback a second,third, etc time... * */ void uac_cb(struct cell* t, int type,struct tmcb_params *rcvd_params) { as_msg_p my_as_ev=0; int mylen,code,i; struct as_uac_param *ev_info; struct sip_msg *msg; char *buffer; ev_info=(struct as_uac_param*)*rcvd_params->param; msg=rcvd_params->rpl; code=rcvd_params->code; buffer=0; if(!ev_info || !ev_info->who){ return; } LOG(L_DBG,"Reply to UAC Transaction for AS:%.*s code: %d\n",ev_info->who->name.len,ev_info->who->name.s,code); LOG(L_DBG,"Transaction %p Nr_of_outgoings:%d is_Local:%c\n",t,t->nr_of_outgoings,is_local(t)?'y':'n'); for(i=0;inr_of_outgoings;i++) LOG(L_DBG,"UAC[%d].last_received=%d\n",i,t->uac[i].last_received); if(!(my_as_ev=shm_malloc(sizeof(as_msg_t)))){ LOG(L_ERR,"Out of shared mem\n"); goto error; } if(!(buffer=create_as_action_reply(t,rcvd_params,ev_info->uac_id,ev_info->processor_id,&mylen))){ LOG(L_ERR,":seas:uac_cb() Unable to encode message\n"); goto error; } my_as_ev->as = ev_info->who; my_as_ev->msg = buffer; my_as_ev->len = mylen; my_as_ev->type = RES_IN; my_as_ev->transaction = t; if(write(write_pipe,&my_as_ev,sizeof(as_msg_p))<=0){ goto error; } goto exit; error: if(my_as_ev){ shm_free(my_as_ev); } if(buffer) shm_free(buffer); exit: return ; } char* create_as_action_reply(struct cell *c,struct tmcb_params *params,int uac_id,char processor_id,int *evt_len) { int i; unsigned int code,flags; unsigned short int port; unsigned int k,len; char *buffer; struct sip_msg *msg; if(!(buffer=shm_malloc(ENCODED_MSG_SIZE))){ LOG(L_ERR,"create_as_action_reply Out Of Memory !!\n"); return 0; } msg=0; *evt_len=0; flags=0; if(params->rpl==FAKED_REPLY) flags=FAKED_REPLY_FLAG; /*length*/ k=4; /*type*/ buffer[k++]=(unsigned char)RES_IN; /*processor id*/ buffer[k++]=processor_id; /*flags (by now, not used)*/ flags=htonl(flags); memcpy(buffer+k,&flags,4); k+=4; /*recv info*/ if(!(params->rpl == FAKED_REPLY)) { msg=params->rpl; /*protocol should be UDP,TCP,TLS or whatever*/ buffer[k++]=(unsigned char)msg->rcv.proto; /*src ip len + src ip*/ len=msg->rcv.src_ip.len; buffer[k++]=(unsigned char)len; memcpy(buffer+k,&(msg->rcv.src_ip.u),len); k+=len; /*dst ip len + dst ip*/ len=msg->rcv.dst_ip.len; buffer[k++]=(unsigned char)len; memcpy(buffer+k,&(msg->rcv.dst_ip.u),len); k+=len; /*src port */ port=htons(msg->rcv.src_port); memcpy(buffer+k,&port,2); k+=2; /*dst port */ port=htons(msg->rcv.dst_port); memcpy(buffer+k,&port,2); k+=2; }else{ /*protocol*/ buffer[k++]=0; /*src ip len*/ buffer[k++]=0; /*dst ip len*/ buffer[k++]=0; /*skip src port and dst port*/ buffer[k++]=0; buffer[k++]=0; buffer[k++]=0; buffer[k++]=0; } /*hash_index*/ i=htonl(c->hash_index); memcpy(buffer+k,&i,4); k+=4; /*label*/ i=(!strncmp(c->method.s,"CANCEL",6)) ? \ htonl(((struct as_uac_param*)*params->param)->label) : \ htonl(c->label); memcpy(buffer+k,&i,4); k+=4; /*uac_id*/ uac_id=htonl(uac_id); memcpy(buffer+k,&uac_id,4); k+=4; /*code*/ code=htonl(params->code); memcpy(buffer+k,&code,4); k+=4; /*length of event (hdr+payload-4), copied at the beginning*/ if(params->rpl != FAKED_REPLY) { if((i=encode_msg(msg,buffer+k,ENCODED_MSG_SIZE-k))<0){ LOG(L_ERR,"Unable to encode msg\n"); goto error; } k+=i; } *evt_len=k; k=htonl(k); memcpy(buffer,&k,4); return buffer; error: return 0; }