diff --git a/src/core/msg_translator.c b/src/core/msg_translator.c index b54eded3f89..b2e119fb19f 100644 --- a/src/core/msg_translator.c +++ b/src/core/msg_translator.c @@ -111,6 +111,7 @@ extern char version[]; extern int version_len; +extern int ksr_tcp_accept_haproxy; str _ksr_xavp_via_params = STR_NULL; str _ksr_xavp_via_fields = STR_NULL; @@ -558,6 +559,9 @@ static inline int lumps_len( int recv_proto_id = PROTO_NONE; int recv_af = 0; struct socket_info *send_sock; + char ip_buf[MAX_URI_SIZE] = {0}; + str _recv_address_str = STR_NULL; + str _recv_port_str = STR_NULL; #ifdef USE_COMP @@ -828,7 +832,16 @@ static inline int lumps_len( send_proto_id = send_sock->proto; } /* init recv_address_str, recv_port_str & recv_port_no */ - if(msg->rcv.bind_address) { + if(ksr_tcp_accept_haproxy && msg->rcv.proto_reserved2) { + ip_addr2sbuf(&msg->rcv.dst_ip, ip_buf, MAX_URI_SIZE); + recv_address_str = &_recv_address_str; + recv_address_str->s = ip_buf; + recv_address_str->len = strlen(ip_buf); + recv_port_str = &_recv_port_str; + recv_port_str->s = int2str(msg->rcv.dst_port, &recv_port_str->len); + recv_port_no = msg->rcv.dst_port; + recv_proto_id = msg->rcv.bind_address->proto; + } else if(msg->rcv.bind_address) { if(msg->rcv.bind_address->useinfo.name.len > 0) { recv_address_str = &(msg->rcv.bind_address->useinfo.name); recv_af = msg->rcv.bind_address->useinfo.af; @@ -959,6 +972,9 @@ void process_lumps(struct sip_msg *msg, struct lump *lumps, char *new_buf, int recv_proto_id = PROTO_NONE; int recv_af = 0; struct socket_info *send_sock; + char ip_buf[MAX_URI_SIZE] = {0}; + str _recv_address_str = STR_NULL; + str _recv_port_str = STR_NULL; #ifdef USE_COMP #define RCVCOMP_PARAM_ADD \ @@ -1356,7 +1372,16 @@ void process_lumps(struct sip_msg *msg, struct lump *lumps, char *new_buf, send_proto_id = send_sock->proto; } /* init recv_address_str, recv_port_str & recv_port_no */ - if(msg->rcv.bind_address) { + if(ksr_tcp_accept_haproxy && msg->rcv.proto_reserved2) { + ip_addr2sbuf(&msg->rcv.dst_ip, ip_buf, MAX_URI_SIZE); + recv_address_str = &_recv_address_str; + recv_address_str->s = ip_buf; + recv_address_str->len = strlen(ip_buf); + recv_port_str = &_recv_port_str; + recv_port_str->s = int2str(msg->rcv.dst_port, &recv_port_str->len); + recv_port_no = msg->rcv.dst_port; + recv_proto_id = msg->rcv.bind_address->proto; + } else if(msg->rcv.bind_address) { if(msg->rcv.bind_address->useinfo.name.len > 0) { recv_address_str = &(msg->rcv.bind_address->useinfo.name); recv_af = msg->rcv.bind_address->useinfo.af; @@ -2845,6 +2870,49 @@ int branch_builder(unsigned int hash_index, return size; } +static int is_haproxy( + struct dest_info *send_info, struct receive_info *haproxy_rcv) +{ + int port; + struct ip_addr ip; + union sockaddr_union *from = NULL; + union sockaddr_union local_addr; + struct tcp_connection *con = NULL; + int haproxy = 0; + + if(!send_info || !haproxy_rcv) { + LM_ERR("wrong arguments\n"); + return 0; + } + + if(unlikely(send_info->send_flags.f & SND_F_FORCE_SOCKET + && send_info->send_sock)) { + local_addr = send_info->send_sock->su; + su_setport(&local_addr, 0); /* any local port will do */ + from = &local_addr; + } + + port = su_getport(&send_info->to); + if(likely(port)) { + su2ip_addr(&ip, &send_info->to); + con = tcpconn_get(send_info->id, &ip, port, from, 0); + } else if(likely(send_info->id)) + con = tcpconn_get(send_info->id, 0, 0, 0, 0); + else { + LM_CRIT("null_id & to\n"); + return 0; + } + + if(con == NULL) { + LM_DBG("TCP/TLS connection (id: %d) for is not found\n", send_info->id); + return 0; + } + + *haproxy_rcv = con->rcv; + haproxy = con->rcv.proto_reserved2; + tcpconn_put(con); + return haproxy; +} /* uses only the send_info->send_socket, send_info->proto, send_info->id and * send_info->comp (so that a send_info used for sending can be passed @@ -2872,6 +2940,11 @@ char *via_builder(unsigned int *len, sip_msg_t *msg, struct tcp_connection *con = NULL; sr_xavp_t *rxavp = NULL; str xname; + char ip_buf[MAX_URI_SIZE] = {0}; + str haproxy_address_str = STR_NULL; + str haproxy_port_str = STR_NULL; + struct receive_info haproxy_rcv; + int haproxy = is_haproxy(send_info, &haproxy_rcv); send_sock = send_info->send_sock; /* use pre-set address in via, the outbound socket alias or address one */ @@ -2883,6 +2956,11 @@ char *via_builder(unsigned int *len, sip_msg_t *msg, if(rxavp != NULL) { address_str = &rxavp->val.v.s; } + } else if(ksr_tcp_accept_haproxy && haproxy) { + ip_addr2sbuf(&haproxy_rcv.dst_ip, ip_buf, MAX_URI_SIZE); + address_str = &haproxy_address_str; + address_str->s = ip_buf; + address_str->len = strlen(ip_buf); } if(address_str == NULL) { if(hp && hp->host->len) @@ -2900,6 +2978,9 @@ char *via_builder(unsigned int *len, sip_msg_t *msg, if(rxavp != NULL) { port_str = &rxavp->val.v.s; } + } else if(ksr_tcp_accept_haproxy && haproxy) { + port_str = &haproxy_port_str; + port_str->s = int2str(haproxy_rcv.dst_port, &port_str->len); } if(port_str == NULL) { if(hp && hp->port->len) @@ -2918,6 +2999,8 @@ char *via_builder(unsigned int *len, sip_msg_t *msg, if(rxavp != NULL) { proto = get_valid_proto_id(&rxavp->val.v.s); } + } else if(ksr_tcp_accept_haproxy && haproxy) { + proto = haproxy_rcv.proto; } if(proto == PROTO_NONE) { if(send_sock->useinfo.proto != PROTO_NONE) { @@ -3442,6 +3525,7 @@ int sip_msg_update_buffer(sip_msg_t *msg, str *obuf) msg->id = tmp.id; msg->pid = tmp.pid; msg->rcv = tmp.rcv; + msg->haproxy_rcv = tmp.haproxy_rcv; msg->set_global_address = tmp.set_global_address; msg->set_global_port = tmp.set_global_port; msg->flags = tmp.flags; diff --git a/src/core/parser/msg_parser.h b/src/core/parser/msg_parser.h index 45e7eb1a714..62a7aa46f3d 100644 --- a/src/core/parser/msg_parser.h +++ b/src/core/parser/msg_parser.h @@ -396,6 +396,8 @@ typedef struct sip_msg char *unparsed; /*!< here we stopped parsing*/ struct receive_info rcv; /*!< source & dest ip, ports, proto a.s.o*/ + struct receive_info + haproxy_rcv; /*!< source & dest ip, ports, proto a.s.o for the message from haproxy*/ char *buf; /*!< scratch pad, holds a modified message, * via, etc. point into it */ diff --git a/src/core/receive.c b/src/core/receive.c index 702e94437ca..52fb4a930b5 100644 --- a/src/core/receive.c +++ b/src/core/receive.c @@ -292,7 +292,8 @@ int ksr_evrt_pre_routing(sip_msg_t *msg) * WARNING: buf must be 0 terminated (buf[len]=0) or some things might * break (e.g.: modules/textops) */ -int receive_msg(char *buf, unsigned int len, receive_info_t *rcv_info) +int _receive_msg(char *buf, unsigned int len, receive_info_t *rcv_info, + receive_info_t *haproxy_rcv_info) { struct sip_msg *msg = NULL; struct run_act_ctx ctx; @@ -361,6 +362,10 @@ int receive_msg(char *buf, unsigned int len, receive_info_t *rcv_info) msg->set_global_address = default_global_address; msg->set_global_port = default_global_port; + if(haproxy_rcv_info) { + msg->haproxy_rcv = *haproxy_rcv_info; + } + if(likely(sr_msg_time == 1)) msg_set_time(msg); diff --git a/src/core/receive.h b/src/core/receive.h index 2f0e662fb9b..ecd764f9458 100644 --- a/src/core/receive.h +++ b/src/core/receive.h @@ -30,7 +30,9 @@ #include "ip_addr.h" -int receive_msg(char *buf, unsigned int len, struct receive_info *ri); +#define receive_msg(_buf, _len, _ri) _receive_msg(_buf, _len, _ri, NULL) +int _receive_msg(char *buf, unsigned int len, struct receive_info *ri, + struct receive_info *haproxy_ri); int sip_check_fline(char *buf, unsigned int len); unsigned int inc_msg_no(void); void ksr_msg_env_reset(void); diff --git a/src/core/tcp_conn.h b/src/core/tcp_conn.h index 4534439f6b2..81e68946380 100644 --- a/src/core/tcp_conn.h +++ b/src/core/tcp_conn.h @@ -272,6 +272,7 @@ typedef struct tcp_connection enum tcp_closed_reason event; /* connection close reason */ int reader_pid; /* pid of the active reader process */ struct receive_info rcv; /* src & dst ip, ports, proto a.s.o*/ + struct receive_info haproxy_rcv; /* haproxy real connection*/ ksr_coninfo_t cinfo; /* connection info (e.g., for haproxy ) */ struct tcp_req req; /* request data */ atomic_t refcnt; diff --git a/src/core/tcp_main.c b/src/core/tcp_main.c index 757329a8543..ee1c54ea5ef 100644 --- a/src/core/tcp_main.c +++ b/src/core/tcp_main.c @@ -1002,6 +1002,9 @@ int tcpconn_read_haproxy(struct tcp_connection *c) src_ip = &c->rcv.src_ip; dst_ip = &c->rcv.dst_ip; + c->haproxy_rcv = c->rcv; + c->haproxy_rcv.proto = PROTO_TCP; + if(bytes >= 16 && memcmp(&hdr.v2, v2sig, 12) == 0 && (hdr.v2.ver_cmd & 0xF0) == 0x20) { LM_DBG("received PROXY protocol v2 header\n"); @@ -1195,7 +1198,8 @@ struct tcp_connection *tcpconn_new(int sock, union sockaddr_union *su, int state) { struct tcp_connection *c; - int rd_b_size, ret; + int rd_b_size; + int ret = -1; rd_b_size = cfg_get(tcp, tcp_cfg, rd_buf_size); c = shm_malloc(sizeof(struct tcp_connection) + rd_b_size); @@ -1243,7 +1247,7 @@ struct tcp_connection *tcpconn_new(int sock, union sockaddr_union *su, init_tcp_req(&c->req, (char *)c + sizeof(struct tcp_connection), rd_b_size); c->id = (*connection_id)++; c->rcv.proto_reserved1 = 0; /* this will be filled before receive_message*/ - c->rcv.proto_reserved2 = 0; + c->rcv.proto_reserved2 = !ret ? 1 : 0; /* haproxy msg marker */ c->state = state; c->initstate = state; c->extra_data = 0; @@ -1620,10 +1624,10 @@ inline static struct tcp_connection *tcpconn_add(struct tcp_connection *c) } if(unlikely(c->cinfo.dst_ip.af && !ip_addr_any(&c->cinfo.dst_ip) && !ip_addr_cmp(&c->rcv.dst_ip, &c->cinfo.dst_ip))) { - _tcpconn_add_alias_unsafe(c, c->rcv.src_port, &c->cinfo.dst_ip, 0, - new_conn_alias_flags); - _tcpconn_add_alias_unsafe(c, c->rcv.src_port, &c->cinfo.dst_ip, - c->cinfo.dst_port, new_conn_alias_flags); + _tcpconn_add_alias_unsafe(c, c->haproxy_rcv.src_port, + &c->cinfo.dst_ip, 0, new_conn_alias_flags); + _tcpconn_add_alias_unsafe(c, c->haproxy_rcv.src_port, + &c->cinfo.dst_ip, c->cinfo.dst_port, new_conn_alias_flags); } /* ignore add_alias errors, there are some valid cases when one @@ -1737,8 +1741,10 @@ struct tcp_connection *_tcpconn_find( print_ip("ip=", &a->parent->rcv.src_ip, "\n"); #endif if((a->parent->state != S_CONN_BAD) && (port == a->port) - && ((l_port == 0) || (l_port == a->parent->rcv.dst_port)) - && (ip_addr_cmp(ip, &a->parent->rcv.src_ip)) + && ((l_port == 0) || (l_port == a->parent->rcv.dst_port) + || (l_port == a->parent->cinfo.dst_port)) + && (ip_addr_cmp(ip, &a->parent->rcv.src_ip) + || ip_addr_cmp(ip, &a->parent->cinfo.src_ip)) && (is_local_ip_any || ip_addr_cmp(l_ip, &a->parent->rcv.dst_ip) || ip_addr_cmp(l_ip, &a->parent->cinfo.dst_ip))) { @@ -1850,7 +1856,10 @@ inline static int _tcpconn_add_alias_unsafe(struct tcp_connection *c, int port, a = 0; is_local_ip_any = ip_addr_any(l_ip); if(likely(c)) { - hash = tcp_addr_hash(&c->rcv.src_ip, port, l_ip, l_port); + hash = tcp_addr_hash(c->rcv.proto_reserved2 ? &c->haproxy_rcv.src_ip + : &c->rcv.src_ip, + port, l_ip, l_port); + /* search the aliases for an already existing one */ for(a = tcpconn_aliases_hash[hash], nxt = 0; a; a = nxt) { nxt = a->next; diff --git a/src/core/tcp_read.c b/src/core/tcp_read.c index 8fa6566c14b..13ff1051dde 100644 --- a/src/core/tcp_read.c +++ b/src/core/tcp_read.c @@ -1446,7 +1446,7 @@ int receive_tcp_msg(char *tcpbuf, unsigned int len, #endif if(unlikely(con->req.flags & F_TCP_REQ_HEP3)) return hep3_process_msg(tcpbuf, len, rcv_info, con); - ret = receive_msg(buf, len, rcv_info); + ret = _receive_msg(buf, len, rcv_info, &con->haproxy_rcv); if(ksr_tcp_script_mode & TCP_SCRIPT_MODE_CONTINUE) { return 0; } diff --git a/src/modules/path/path.c b/src/modules/path/path.c index 8fde991260c..7835c821124 100644 --- a/src/modules/path/path.c +++ b/src/modules/path/path.c @@ -86,7 +86,9 @@ static int handleOutbound(sip_msg_t *_m, str *user, path_param_t *param) if(path_obb.use_outbound != NULL && path_obb.use_outbound(_m)) { struct via_body *via; - if(path_obb.encode_flow_token(user, &_m->rcv) != 0) { + if(path_obb.encode_flow_token( + user, _m->rcv.proto_reserved2 ? &_m->haproxy_rcv : &_m->rcv) + != 0) { LM_ERR("encoding outbound flow-token\n"); return -1; } diff --git a/src/modules/rr/loose.c b/src/modules/rr/loose.c index 62a944a1ce8..0d43d39a348 100644 --- a/src/modules/rr/loose.c +++ b/src/modules/rr/loose.c @@ -553,8 +553,9 @@ static inline int process_outbound(struct sip_msg *_m, str flow_token) } else if(ret == -1) { LM_INFO("failed to decode flow token\n"); return -1; - } else if(!ip_addr_cmp(&rcv->src_ip, &_m->rcv.src_ip) - || rcv->src_port != _m->rcv.src_port) { + } else if(!_m->rcv.proto_reserved2 + && (!ip_addr_cmp(&rcv->src_ip, &_m->rcv.src_ip) + || rcv->src_port != _m->rcv.src_port)) { LM_DBG("\"incoming\" request found. Using flow-token for " "routing\n"); diff --git a/src/modules/rr/record.c b/src/modules/rr/record.c index 36fdb2612c4..80605aad162 100644 --- a/src/modules/rr/record.c +++ b/src/modules/rr/record.c @@ -407,7 +407,9 @@ int record_route(struct sip_msg *_m, str *params) } } } else if(use_ob == 1) { - if(rr_obb.encode_flow_token(&user, &_m->rcv) != 0) { + if(rr_obb.encode_flow_token( + &user, _m->rcv.proto_reserved2 ? &_m->haproxy_rcv : &_m->rcv) + != 0) { LM_ERR("encoding outbound flow-token\n"); return -1; } @@ -545,7 +547,9 @@ int record_route_preset(struct sip_msg *_m, str *_data) } } } else if(use_ob == 1) { - if(rr_obb.encode_flow_token(&user, &_m->rcv) != 0) { + if(rr_obb.encode_flow_token( + &user, _m->rcv.proto_reserved2 ? &_m->haproxy_rcv : &_m->rcv) + != 0) { LM_ERR("encoding outbound flow-token\n"); return -1; } @@ -851,7 +855,9 @@ int record_route_advertised_address(struct sip_msg *_m, str *_data) } } } else if(use_ob == 1) { - if(rr_obb.encode_flow_token(&user, &_m->rcv) != 0) { + if(rr_obb.encode_flow_token( + &user, _m->rcv.proto_reserved2 ? &_m->haproxy_rcv : &_m->rcv) + != 0) { LM_ERR("encoding outbound flow-token\n"); return -1; } diff --git a/src/modules/siptrace/siptrace.c b/src/modules/siptrace/siptrace.c index 561e6ac4b90..18c021c8fc8 100644 --- a/src/modules/siptrace/siptrace.c +++ b/src/modules/siptrace/siptrace.c @@ -42,6 +42,7 @@ #include "../../core/rpc.h" #include "../../core/rpc_lookup.h" #include "../../lib/srdb1/db.h" +#include "../../core/tcp_conn.h" #include "../../core/parser/parse_content.h" #include "../../core/parser/parse_from.h" #include "../../core/parser/parse_cseq.h" @@ -204,6 +205,8 @@ int trace_dialog_ack = 1; int trace_dialog_spiral = 1; static int spiral_tracked; +extern int ksr_tcp_accept_haproxy; + int pv_parse_siptrace_name(pv_spec_t *sp, str *in); int pv_get_siptrace(sip_msg_t *msg, pv_param_t *param, pv_value_t *res); @@ -2408,6 +2411,8 @@ int siptrace_net_data_sent(sr_event_param_t *evp) int proto; int evcb_ret; int ret = 0; + int dst_to_updated = 0; + int found = 0; if(evp->data == 0) return -1; @@ -2431,33 +2436,77 @@ int siptrace_net_data_sent(sr_event_param_t *evp) sto.body.s = nd->data.s; sto.body.len = nd->data.len; - if(unlikely(new_dst.send_sock == 0)) { - LM_WARN("no sending socket found\n"); - strcpy(sto.fromip_buff, SIPTRACE_ANYADDR); - sto.fromip.len = SIPTRACE_ANYADDR_LEN; - proto = PROTO_UDP; - } else { - if(new_dst.send_sock->sock_str.len >= SIPTRACE_ADDR_MAX - 1) { - LM_ERR("socket string is too large: %d\n", + if(ksr_tcp_accept_haproxy && new_dst.proto == PROTO_TCP) { + tcp_connection_t *con = NULL; + unsigned short dest_port = su_getport(&new_dst.to); + if(likely(new_dst.id)) { + con = tcpconn_get(new_dst.id, 0, 0, 0, 0); + } else if(likely(dest_port)) { + ip_addr_t ip; + su2ip_addr(&ip, &new_dst.to); + con = tcpconn_get( + new_dst.id, &ip, dest_port, &new_dst.send_sock->su, 0); + } + + if(con == NULL) { + LM_WARN("TCP connection could not be found\n"); + } else if(con->rcv.proto_reserved2) { + sto.fromip.len = snprintf(sto.fromip_buff, SIPTRACE_ADDR_MAX, + "%s:%s:%d", siptrace_proto_name(con->rcv.proto), + ip_addr2a(&con->rcv.dst_ip), (int)con->rcv.dst_port); + proto = PROTO_TCP; + sto.toip.len = snprintf(sto.toip_buff, SIPTRACE_ADDR_MAX, + "%s:%s:%d", siptrace_proto_name(proto), + ip_addr2a(&con->rcv.src_ip), (int)con->rcv.src_port); + if(sto.toip.len < 0 || sto.toip.len >= SIPTRACE_ADDR_MAX) { + LM_ERR("failed to format toip buffer (%d)\n", sto.toip.len); + sto.toip.s = SIPTRACE_ANYADDR; + sto.toip.len = SIPTRACE_ANYADDR_LEN; + } else { + sto.toip.s = sto.toip_buff; + } + dst_to_updated = 1; + found = 1; + tcpconn_put(con); + } else { + LM_WARN("Current TCP connection is not a connection to haproxy " + "server\n"); + tcpconn_put(con); + } + } + + if(!found) { + if(unlikely(new_dst.send_sock == 0)) { + LM_WARN("no sending socket found\n"); + strcpy(sto.fromip_buff, SIPTRACE_ANYADDR); + sto.fromip.len = SIPTRACE_ANYADDR_LEN; + proto = PROTO_UDP; + } else { + if(new_dst.send_sock->sock_str.len >= SIPTRACE_ADDR_MAX - 1) { + LM_ERR("socket string is too large: %d\n", + new_dst.send_sock->sock_str.len); + return -1; + } + strncpy(sto.fromip_buff, new_dst.send_sock->sock_str.s, new_dst.send_sock->sock_str.len); - return -1; + sto.fromip.len = new_dst.send_sock->sock_str.len; + proto = new_dst.send_sock->proto; } - strncpy(sto.fromip_buff, new_dst.send_sock->sock_str.s, - new_dst.send_sock->sock_str.len); - sto.fromip.len = new_dst.send_sock->sock_str.len; - proto = new_dst.send_sock->proto; } sto.fromip.s = sto.fromip_buff; - sto.toip.len = snprintf(sto.toip_buff, SIPTRACE_ADDR_MAX, "%s:%s:%d", - siptrace_proto_name(proto), suip2a(&new_dst.to, sizeof(new_dst.to)), - (int)su_getport(&new_dst.to)); - if(sto.toip.len < 0 || sto.toip.len >= SIPTRACE_ADDR_MAX) { - LM_ERR("failed to format toip buffer (%d)\n", sto.toip.len); - sto.toip.s = SIPTRACE_ANYADDR; - sto.toip.len = SIPTRACE_ANYADDR_LEN; - } else { - sto.toip.s = sto.toip_buff; + if(!dst_to_updated) { + sto.toip.len = snprintf(sto.toip_buff, SIPTRACE_ADDR_MAX, "%s:%s:%d", + siptrace_proto_name(proto), + suip2a(&new_dst.to, sizeof(new_dst.to)), + (int)su_getport(&new_dst.to)); + if(sto.toip.len < 0 || sto.toip.len >= SIPTRACE_ADDR_MAX) { + LM_ERR("failed to format toip buffer (%d)\n", sto.toip.len); + sto.toip.s = SIPTRACE_ANYADDR; + sto.toip.len = SIPTRACE_ANYADDR_LEN; + } else { + sto.toip.s = sto.toip_buff; + } } sto.dir = "out";