/src/kamailio/src/core/tcp_read.c

Source (jump to first uncovered line)
/*
 * Copyright (C) 2001-2003 FhG Fokus
 *
 * 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., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 */

/** Kamailio core :: tcp readers processes, tcp read and pre-parse msg. functions.
 * @file tcp_read.c
 * @ingroup core
 * Module: @ref core
 */

#ifdef USE_TCP

#include <stdio.h>
#include <errno.h>
#include <string.h>


#include <sys/time.h>
#include <sys/types.h>
#include <sys/select.h>
#include <sys/socket.h>

#include <unistd.h>
#include <stdlib.h> /* for abort() */

#include "dprint.h"
#include "tcp_conn.h"
#include "tcp_read.h"
#include "tcp_stats.h"
#include "tcp_ev.h"
#include "pass_fd.h"
#include "globals.h"
#include "receive.h"
#include "timer.h"
#include "local_timer.h"
#include "ut.h"
#include "trim.h"
#include "pt.h"
#include "daemonize.h"
#include "cfg/cfg_struct.h"
#ifdef CORE_TLS
#include "tls/tls_server.h"
#else
#include "tls_hooks.h"
#endif /* CORE_TLS */
#ifdef USE_DST_BLOCKLIST
#include "dst_blocklist.h"
#endif /* USE_DST_BLOCKLIST */

#define HANDLE_IO_INLINE
#include "io_wait.h"
#include <fcntl.h> /* must be included after io_wait.h if SIGIO_RT is used */
#include "tsend.h"
#include "forward.h"
#include "events.h"
#include "stun.h"
#include "nonsip_hooks.h"

#ifdef READ_HTTP11
#define HTTP11CONTINUE "HTTP/1.1 100 Continue\r\nContent-Length: 0\r\n\r\n"
#define HTTP11CONTINUE_LEN (sizeof(HTTP11CONTINUE) - 1)
#endif

#define TCPCONN_TIMEOUT_MIN_RUN 1 /* run the timers each new tick */

/* types used in io_wait* */
enum fd_types
{
  F_NONE,
  F_TCPMAIN,
  F_TCPCONN
};

/* list of tcp connections handled by this process */
static struct tcp_connection *tcp_conn_lst = 0;
static io_wait_h io_w; /* io_wait handler*/
static int tcpmain_sock = -1;

static struct local_timer tcp_reader_ltimer;
static ticks_t tcp_reader_prev_ticks;

int is_msg_complete(struct tcp_req *r);

int ksr_tcp_accept_hep3 = 0;
int ksr_tcp_accept_haproxy = 0;

#define TCP_SCRIPT_MODE_CONTINUE (1 << 0)
int ksr_tcp_script_mode = 0;

/**
 * control cloning of TCP receive buffer
 * - needed for operations working directly inside the buffer
 *   (like msg_apply_changes())
 */
#define TCP_CLONE_RCVBUF
static int tcp_clone_rcvbuf = 0;

int tcp_set_clone_rcvbuf(int v)
{
  int r;
  r = tcp_clone_rcvbuf;
  tcp_clone_rcvbuf = v;
  return r;
}

int tcp_get_clone_rcvbuf(void)
{
  return tcp_clone_rcvbuf;
}

#ifdef READ_HTTP11
int tcp_http11_continue(struct tcp_connection *c)
{
  struct dest_info dst;
  char *p;
  struct msg_start fline;
  int ret;
  str msg;

  ret = 0;

  msg.s = c->req.start;
  msg.len = c->req.pos - c->req.start;
#ifdef READ_MSRP
  /* skip if MSRP message */
  if(c->req.flags & F_TCP_REQ_MSRP_FRAME)
    return 0;
#endif
  p = parse_first_line(msg.s, msg.len, &fline);
  if(p == NULL)
    return 0;

  if(fline.type != SIP_REQUEST)
    return 0;

  /* check if http request */
  if(fline.u.request.version.len < HTTP_VERSION_LEN
      || strncasecmp(
          fline.u.request.version.s, HTTP_VERSION, HTTP_VERSION_LEN))
    return 0;

  /* check for Expect header */
  if(str_casesearch_strz(&msg, "Expect: 100-continue") != NULL) {
    init_dst_from_rcv(&dst, &c->rcv);
    if(tcp_send(&dst, 0, HTTP11CONTINUE, HTTP11CONTINUE_LEN) < 0) {
      LM_ERR("HTTP/1.1 continue failed\n");
    }
  }
  /* check for Transfer-Encoding header */
  if(str_casesearch_strz(&msg, "Transfer-Encoding: chunked") != NULL) {
    c->req.flags |= F_TCP_REQ_BCHUNKED;
    ret = 1;
  }
  /* check for HTTP Via header
   * - HTTP Via format is different that SIP Via
   * - workaround: replace with Hia to be ignored by SIP parser
   */
  if((p = str_casesearch_strz(&msg, "\nVia:")) != NULL) {
    p++;
    *p = 'H';
    LM_DBG("replaced HTTP Via with Hia [[\n%.*s]]\n", msg.len, msg.s);
  }
  return ret;
}
#endif /* HTTP11 */


/** reads data from an existing tcp connection.
 * Side-effects: blocklisting, sets connection state to S_CONN_OK, tcp stats.
 * @param fd - connection file descriptor
 * @param c - tcp connection structure. c->state might be changed and
 *             receive info might be used for blocklisting.
 * @param buf - buffer where the received data will be stored.
 * @param b_size - buffer size.
 * @param flags - value/result - used to signal a seen or "forced" EOF on the
 *     connection (when it is known that no more data will come after the
 *     current socket buffer is emptied )=> return/signal EOF on the first
 *     short read (=> don't use it on POLLPRI, as OOB data will cause short
 *     reads even if there are still remaining bytes in the socket buffer)
 *     input: RD_CONN_FORCE_EOF  - force EOF after the first successful read
 *                                 (bytes_read >=0 )
 *     output: RD_CONN_SHORT_READ - if the read exhausted all the bytes
 *                                  in the socket read buffer.
 *             RD_CONN_EOF - if EOF detected (0 bytes read) or forced via
 *                           RD_CONN_FORCE_EOF.
 *             RD_CONN_REPEAT_READ - the read should be repeated immediately
 *                                   (used only by the tls code for now).
 *     Note: RD_CONN_SHORT_READ & RD_CONN_EOF _are_ not cleared internally,
 *           so one should clear them before calling this function.
 * @return number of bytes read, 0 on EOF or -1 on error,
 * on EOF it also sets c->state to S_CONN_EOF.
 * (to distinguish from reads that would block which could return 0)
 * RD_CONN_SHORT_READ is also set in *flags for short reads.
 * EOF checking should be done by checking the RD_CONN_EOF flag.
 */
int tcp_read_data(int fd, struct tcp_connection *c, char *buf, int b_size,
    rd_conn_flags_t *flags)
{
  int bytes_read;

again:
  bytes_read = read(fd, buf, b_size);

  if(likely(bytes_read != b_size)) {
    if(unlikely(bytes_read == -1)) {
      if(errno == EWOULDBLOCK || errno == EAGAIN) {
        bytes_read = 0; /* nothing has been read */
      } else if(errno == EINTR) {
        goto again;
      } else {
        if(unlikely(c->state == S_CONN_CONNECT)) {
          switch(errno) {
            case ECONNRESET:
#ifdef USE_DST_BLOCKLIST
              dst_blocklist_su(BLST_ERR_CONNECT, c->rcv.proto,
                  &c->rcv.src_su, &c->send_flags, 0);
#endif /* USE_DST_BLOCKLIST */
              TCP_EV_CONNECT_RST(errno, TCP_LADDR(c),
                  TCP_LPORT(c), TCP_PSU(c), TCP_PROTO(c));
              break;
            case ETIMEDOUT:
#ifdef USE_DST_BLOCKLIST
              dst_blocklist_su(BLST_ERR_CONNECT, c->rcv.proto,
                  &c->rcv.src_su, &c->send_flags, 0);
#endif /* USE_DST_BLOCKLIST */
              TCP_EV_CONNECT_TIMEOUT(errno, TCP_LADDR(c),
                  TCP_LPORT(c), TCP_PSU(c), TCP_PROTO(c));
              break;
            default:
              TCP_EV_CONNECT_ERR(errno, TCP_LADDR(c),
                  TCP_LPORT(c), TCP_PSU(c), TCP_PROTO(c));
          }
          TCP_STATS_CONNECT_FAILED();
        } else {
          switch(errno) {
            case ECONNRESET:
              TCP_STATS_CON_RESET();
#ifdef USE_DST_BLOCKLIST
              dst_blocklist_su(BLST_ERR_SEND, c->rcv.proto,
                  &c->rcv.src_su, &c->send_flags, 0);
#endif /* USE_DST_BLOCKLIST */
              break;
            case ETIMEDOUT:
#ifdef USE_DST_BLOCKLIST
              dst_blocklist_su(BLST_ERR_SEND, c->rcv.proto,
                  &c->rcv.src_su, &c->send_flags, 0);
#endif /* USE_DST_BLOCKLIST */
              break;
          }
        }
        LOG(cfg_get(core, core_cfg, corelog),
            "error reading: %s (%d) ([%s]:%u ->", strerror(errno),
            errno, ip_addr2a(&c->rcv.src_ip), c->rcv.src_port);
        LOG(cfg_get(core, core_cfg, corelog), "-> [%s]:%u)\n",
            ip_addr2a(&c->rcv.dst_ip), c->rcv.dst_port);
        if(errno == ETIMEDOUT) {
          c->event = TCP_CLOSED_TIMEOUT;
        } else if(errno == ECONNRESET) {
          c->event = TCP_CLOSED_RESET;
        }
        return -1;
      }
    } else if(unlikely((bytes_read == 0) || (*flags & RD_CONN_FORCE_EOF))) {
      LM_DBG("EOF on connection %p (state: %u, flags: %x) - FD %d,"
           " bytes %d, rd-flags %x ([%s]:%u -> [%s]:%u)",
          c, c->state, c->flags, fd, bytes_read, *flags,
          ip_addr2a(&c->rcv.src_ip), c->rcv.src_port,
          ip_addr2a(&c->rcv.dst_ip), c->rcv.dst_port);
      c->state = S_CONN_EOF;
      *flags |= RD_CONN_EOF;
      c->event = TCP_CLOSED_EOF;
    } else {
      if(unlikely(c->state == S_CONN_CONNECT
            || c->state == S_CONN_ACCEPT)) {
        TCP_STATS_ESTABLISHED(c->state);
        c->state = S_CONN_OK;
      }
    }
    /* short read */
    *flags |= RD_CONN_SHORT_READ;
  } else { /* else normal full read */
    if(unlikely(c->state == S_CONN_CONNECT || c->state == S_CONN_ACCEPT)) {
      TCP_STATS_ESTABLISHED(c->state);
      c->state = S_CONN_OK;
    }
  }
  return bytes_read;
}


/* reads next available bytes
 *   c- tcp connection used for reading, tcp_read changes also c->state on
 *      EOF and c->req.error on read error
 *   * flags - value/result - used to signal a seen or "forced" EOF on the
 *     connection (when it is known that no more data will come after the
 *     current socket buffer is emptied )=> return/signal EOF on the first
 *     short read (=> don't use it on POLLPRI, as OOB data will cause short
 *      reads even if there are still remaining bytes in the socket buffer)
 * return number of bytes read, 0 on EOF or -1 on error,
 * on EOF it also sets c->state to S_CONN_EOF.
 * (to distinguish from reads that would block which could return 0)
 * RD_CONN_SHORT_READ is also set in *flags for short reads.
 * sets also r->error */
int tcp_read(struct tcp_connection *c, rd_conn_flags_t *flags)
{
  int bytes_free, bytes_read;
  struct tcp_req *r;
  int fd;

  r = &c->req;
  fd = c->fd;
  bytes_free = r->b_size - (unsigned int)(r->pos - r->buf);

  if(unlikely(bytes_free <= 0)) {
    LM_ERR("buffer overrun, dropping ([%s]:%u -> [%s]:%u)\n",
        ip_addr2a(&c->rcv.src_ip), c->rcv.src_port,
        ip_addr2a(&c->rcv.dst_ip), c->rcv.dst_port);
    r->error = TCP_REQ_OVERRUN;
    return -1;
  }
  bytes_read = tcp_read_data(fd, c, r->pos, bytes_free, flags);
  if(unlikely(bytes_read < 0)) {
    r->error = TCP_READ_ERROR;
    return -1;
  }
#ifdef EXTRA_DEBUG
  LM_DBG("read %d bytes:\n%.*s\n", bytes_read, bytes_read, r->pos);
#endif
  r->pos += bytes_read;
  return bytes_read;
}


/* reads all headers (until double crlf), & parses the content-length header
 * (WARNING: inefficient, tries to reuse receive_msg but will go through
 * the headers twice [once here looking for Content-Length and for the end
 * of the headers and once in receive_msg]; a more speed efficient version will
 * result in either major code duplication or major changes to the receive code)
 * returns number of bytes read & sets r->state & r->body
 * when either r->body!=0 or r->state==H_BODY =>
 * all headers have been read. It should be called in a while loop.
 * returns < 0 if error or 0 if EOF */
int tcp_read_headers(struct tcp_connection *c, rd_conn_flags_t *read_flags)
{
  int bytes, remaining;
  char *p;
  struct tcp_req *r;
  unsigned int mc; /* magic cookie */
  unsigned short body_len;
  struct timeval tvnow;
  long long tvdiff;

#ifdef READ_MSRP
  char *mfline;
  str mtransid;
#endif

#define crlf_default_skip_case \
  case '\n':                 \
    r->state = H_LF;       \
    break;                 \
  default:                   \
    r->state = H_SKIP

#define content_len_beg_case                         \
  case ' ':                                        \
  case '\t':                                       \
    if(!TCP_REQ_HAS_CLEN(r))                     \
      r->state = H_STARTWS;                    \
    else                                         \
      r->state = H_SKIP;                       \
    /* not interested if we already found one */ \
    break;                                       \
  case 'C':                                        \
  case 'c':                                        \
    if(!TCP_REQ_HAS_CLEN(r))                     \
      r->state = H_CONT_LEN1;                  \
    else                                         \
      r->state = H_SKIP;                       \
    break;                                       \
  case 'l':                                        \
  case 'L':                                        \
    /* short form for Content-Length */          \
    if(!TCP_REQ_HAS_CLEN(r))                     \
      r->state = H_L_COLON;                    \
    else                                         \
      r->state = H_SKIP;                       \
    break

#define change_state(upper, lower, newstate) \
  switch(*p) {                             \
    case upper:                          \
    case lower:                          \
      r->state = (newstate);           \
      break;                           \
      crlf_default_skip_case;          \
  }

#define change_state_case(state0, upper, lower, newstate) \
  case state0:                                          \
    change_state(upper, lower, newstate);             \
    p++;                                              \
    break


  r = &c->req;
  if(r->parsed < r->buf || r->parsed > r->buf + r->b_size) {
    if(r->parsed < r->buf && (unsigned char)r->state == H_SKIP_EMPTY) {
      /* give it a chance to parse from beginning */
      LM_WARN("resetting parsed pointer (buf:%p parsed:%p bsize:%u)\n",
          r->buf, r->parsed, r->b_size);
      r->parsed = r->buf;
    } else {
      LM_ERR("out of bounds parsed pointer (buf:%p parsed:%p bsize:%u)\n",
          r->buf, r->parsed, r->b_size);
      r->parsed = r->buf;
      r->content_len = 0;
      r->error = TCP_REQ_BAD_LEN;
      r->state = H_SKIP; /* skip state now */
      return -1;
    }
  }
  /* if we still have some unparsed part, parse it first, don't do the read*/
  if(unlikely(r->parsed < r->pos)) {
    bytes = 0;
  } else {
#ifdef USE_TLS
    if(unlikely(c->type == PROTO_TLS))
      bytes = tls_read(c, read_flags);
    else
#endif
      bytes = tcp_read(c, read_flags);
    if(bytes <= 0)
      return bytes;
    gettimeofday(&tvnow, NULL);
    tvdiff = 1000000 * (tvnow.tv_sec - r->tvrstart.tv_sec)
         + (tvnow.tv_usec - r->tvrstart.tv_usec);
    if(tvdiff >= ksr_tcp_msg_read_timeout * 1000000) {
      LM_ERR("message reading timeout after %lld usec\n", tvdiff);
      r->parsed = r->buf;
      r->content_len = 0;
      r->error = TCP_REQ_BAD_LEN;
      r->state = H_SKIP; /* skip now */
      return -1;
    }
  }
  p = r->parsed;

  while(p < r->pos && r->error == TCP_REQ_OK) {
    switch((unsigned char)r->state) {
      case H_BODY: /* read the body*/
        remaining = r->pos - p;
        if(remaining > r->bytes_to_go)
          remaining = r->bytes_to_go;
        r->bytes_to_go -= remaining;
        p += remaining;
        if(r->bytes_to_go == 0) {
          r->flags |= F_TCP_REQ_COMPLETE;
          goto skip;
        }
        break;

      case H_SKIP:
        /* find lf, we are in this state if we are not interested
         * in anything till end of line*/
        p = q_memchr(p, '\n', r->pos - p);
        if(p) {
#ifdef READ_MSRP
          /* catch if it is MSRP or not with first '\n' */
          if(!((r->flags & F_TCP_REQ_MSRP_NO)
                 || (r->flags & F_TCP_REQ_MSRP_FRAME))) {
            if((r->pos - r->start) > 5
                && strncmp(r->start, "MSRP ", 5) == 0) {
              r->flags |= F_TCP_REQ_MSRP_FRAME;
            } else {
              r->flags |= F_TCP_REQ_MSRP_NO;
            }
          }
#endif
          p++;
          r->state = H_LF;
        } else {
          p = r->pos;
        }
        break;

      case H_LF:
        /* terminate on LF CR LF or LF LF */
        switch(*p) {
          case '\r':
            r->state = H_LFCR;
            break;
          case '\n':
            /* found LF LF */
            r->state = H_BODY;
#ifdef READ_HTTP11
            if(cfg_get(tcp, tcp_cfg, accept_no_cl) != 0)
              tcp_http11_continue(c);
#endif
            if(TCP_REQ_HAS_CLEN(r)) {
              r->body = p + 1;
              r->bytes_to_go = r->content_len;
              if(r->bytes_to_go == 0) {
                r->flags |= F_TCP_REQ_COMPLETE;
                p++;
                goto skip;
              }
            } else {
              if(cfg_get(tcp, tcp_cfg, accept_no_cl) != 0) {
#ifdef READ_MSRP
                /* if MSRP message */
                if(c->req.flags & F_TCP_REQ_MSRP_FRAME) {
                  r->body = p + 1;
                  /* at least 3 bytes: 0\r\n */
                  r->bytes_to_go = 3;
                  p++;
                  r->content_len = 0;
                  r->state = H_MSRP_BODY;
                  break;
                }
#endif

#ifdef READ_HTTP11
                if(TCP_REQ_BCHUNKED(r)) {
                  r->body = p + 1;
                  /* at least 3 bytes: 0\r\n */
                  r->bytes_to_go = 3;
                  p++;
                  r->content_len = 0;
                  r->state = H_HTTP11_CHUNK_START;
                  break;
                }
#endif
                r->body = p + 1;
                r->bytes_to_go = 0;
                r->flags |= F_TCP_REQ_COMPLETE;
                p++;
                goto skip;
              } else {
                LM_DBG("no clen, p=%X\n", *p);
                r->error = TCP_REQ_BAD_LEN;
              }
            }
            break;
          case '-':
            r->state = H_SKIP;
#ifdef READ_MSRP
            /* catch end of MSRP frame without body
             *     '-------sessid$\r\n'
             * follows headers without extra CRLF */
            if(r->flags & F_TCP_REQ_MSRP_FRAME) {
              p--;
              r->state = H_MSRP_BODY_END;
            }
#endif
            break;
            content_len_beg_case;
          default:
            r->state = H_SKIP;
        }
        p++;
        break;
      case H_LFCR:
        if(*p == '\n') {
          /* found LF CR LF */
          r->state = H_BODY;
#ifdef READ_HTTP11
          if(cfg_get(tcp, tcp_cfg, accept_no_cl) != 0)
            tcp_http11_continue(c);
#endif
          if(TCP_REQ_HAS_CLEN(r)) {
            r->body = p + 1;
            r->bytes_to_go = r->content_len;
            if(r->bytes_to_go == 0) {
              r->flags |= F_TCP_REQ_COMPLETE;
              p++;
              goto skip;
            }
          } else {
            if(cfg_get(tcp, tcp_cfg, accept_no_cl) != 0) {
#ifdef READ_MSRP
              /* if MSRP message */
              if(c->req.flags & F_TCP_REQ_MSRP_FRAME) {
                r->body = p + 1;
                /* at least 3 bytes: 0\r\n */
                r->bytes_to_go = 3;
                p++;
                r->content_len = 0;
                r->state = H_MSRP_BODY;
                break;
              }
#endif

#ifdef READ_HTTP11
              if(TCP_REQ_BCHUNKED(r)) {
                r->body = p + 1;
                /* at least 3 bytes: 0\r\n */
                r->bytes_to_go = 3;
                p++;
                r->content_len = 0;
                r->state = H_HTTP11_CHUNK_START;
                break;
              }
#endif
              r->body = p + 1;
              r->bytes_to_go = 0;
              r->flags |= F_TCP_REQ_COMPLETE;
              p++;
              goto skip;
            } else {
              LM_DBG("no clen, p=%X\n", *p);
              r->error = TCP_REQ_BAD_LEN;
            }
          }
        } else
          r->state = H_SKIP;
        p++;
        break;

      case H_STARTWS:
        switch(*p) {
          content_len_beg_case;
          crlf_default_skip_case;
        }
        p++;
        break;
      case H_SKIP_EMPTY:
        switch(*p) {
          case '\n':
            break;
          case '\r':
            if(cfg_get(tcp, tcp_cfg, crlf_ping)) {
              r->state = H_SKIP_EMPTY_CR_FOUND;
              r->start = p;
            }
            break;
          case ' ':
          case '\t':
            /* skip empty lines */
            break;
          case 'C':
          case 'c':
            r->state = H_CONT_LEN1;
            r->start = p;
            break;
          case 'l':
          case 'L':
            /* short form for Content-Length */
            r->state = H_L_COLON;
            r->start = p;
            break;
          default:
            /* stun test */
            if(unlikely(sr_event_enabled(SREV_STUN_IN))
                && (unsigned char)*p == 0x00) {
              r->state = H_STUN_MSG;
              /* body is used as pointer to the last used byte */
              r->body = p;
              r->content_len = 0;
              LM_DBG("stun msg detected\n");
            } else {
              r->state = H_SKIP;
            }
            r->start = p;
        };
        p++;
        break;

      case H_SKIP_EMPTY_CR_FOUND:
        if(*p == '\n') {
          r->state = H_SKIP_EMPTY_CRLF_FOUND;
          p++;
        } else {
          r->state = H_SKIP_EMPTY;
        }
        break;

      case H_SKIP_EMPTY_CRLF_FOUND:
        if(*p == '\r') {
          r->state = H_SKIP_EMPTY_CRLFCR_FOUND;
          p++;
        } else {
          r->state = H_SKIP_EMPTY;
        }
        break;

      case H_SKIP_EMPTY_CRLFCR_FOUND:
        if(*p == '\n') {
          r->state = H_PING_CRLF;
          r->flags |=
              F_TCP_REQ_HAS_CLEN
              | F_TCP_REQ_COMPLETE; /* hack to avoid error check */
          p++;
          goto skip;
        } else {
          r->state = H_SKIP_EMPTY;
        }
        break;

      case H_STUN_MSG:
        if((r->pos - r->body) >= sizeof(struct stun_hdr)) {
          /* copy second short from buffer where should be body
           * length
           */
          memcpy(&body_len, &r->start[sizeof(unsigned short)],
              sizeof(unsigned short));

          body_len = ntohs(body_len);

          /* check if there is valid magic cookie */
          memcpy(&mc, &r->start[sizeof(unsigned int)],
              sizeof(unsigned int));
          mc = ntohl(mc);
          /* using has_content_len as a flag if there should be
           * fingerprint or no
           */
          r->flags |= (mc == MAGIC_COOKIE) ? F_TCP_REQ_HAS_CLEN : 0;

          r->body += sizeof(struct stun_hdr);
          p = r->body;

          if(body_len > 0) {
            r->state = H_STUN_READ_BODY;
          } else {
            if(is_msg_complete(r) != 0) {
              goto skip;
            } else {
              /* set content_len to length of fingerprint */
              body_len = sizeof(struct stun_attr) + 20;
              /* 20 is SHA_DIGEST_LENGTH from openssl/sha.h */
            }
          }
          r->content_len = body_len;
        } else {
          p = r->pos;
        }
        break;

      case H_STUN_READ_BODY:
        /* check if the whole body was read */
        body_len = r->content_len;
        if((r->pos - r->body) >= body_len) {
          r->body += body_len;
          p = r->body;
          if(is_msg_complete(r) != 0) {
            r->content_len = 0;
            goto skip;
          } else {
            /* set content_len to length of fingerprint */
            body_len = sizeof(struct stun_attr) + 20;
            /* 20 is SHA_DIGEST_LENGTH from openssl/sha.h */
            r->content_len = body_len;
          }
        } else {
          p = r->pos;
        }
        break;

      case H_STUN_FP:
        /* content_len contains length of fingerprint in this place! */
        body_len = r->content_len;
        if((r->pos - r->body) >= body_len) {
          r->body += body_len;
          p = r->body;
          r->state = H_STUN_END;
          r->flags |=
              F_TCP_REQ_COMPLETE
              | F_TCP_REQ_HAS_CLEN; /* hack to avoid error check */
          r->content_len = 0;
          goto skip;
        } else {
          p = r->pos;
        }
        break;

        change_state_case(H_CONT_LEN1, 'O', 'o', H_CONT_LEN2);
        change_state_case(H_CONT_LEN2, 'N', 'n', H_CONT_LEN3);
        change_state_case(H_CONT_LEN3, 'T', 't', H_CONT_LEN4);
        change_state_case(H_CONT_LEN4, 'E', 'e', H_CONT_LEN5);
        change_state_case(H_CONT_LEN5, 'N', 'n', H_CONT_LEN6);
        change_state_case(H_CONT_LEN6, 'T', 't', H_CONT_LEN7);
        change_state_case(H_CONT_LEN7, '-', '_', H_CONT_LEN8);
        change_state_case(H_CONT_LEN8, 'L', 'l', H_CONT_LEN9);
        change_state_case(H_CONT_LEN9, 'E', 'e', H_CONT_LEN10);
        change_state_case(H_CONT_LEN10, 'N', 'n', H_CONT_LEN11);
        change_state_case(H_CONT_LEN11, 'G', 'g', H_CONT_LEN12);
        change_state_case(H_CONT_LEN12, 'T', 't', H_CONT_LEN13);
        change_state_case(H_CONT_LEN13, 'H', 'h', H_L_COLON);

      case H_L_COLON:
        switch(*p) {
          case ' ':
          case '\t':
            break; /* skip space */
          case ':':
            r->state = H_CONT_LEN_BODY;
            break;
            crlf_default_skip_case;
        };
        p++;
        break;

      case H_CONT_LEN_BODY:
        switch(*p) {
          case ' ':
          case '\t':
            break; /* eat space */
          case '0':
          case '1':
          case '2':
          case '3':
          case '4':
          case '5':
          case '6':
          case '7':
          case '8':
          case '9':
            r->state = H_CONT_LEN_BODY_PARSE;
            r->content_len = (*p - '0');
            break;
            /* note: review case of content-length
             * on different lines */
            crlf_default_skip_case;
        }
        p++;
        break;

      case H_CONT_LEN_BODY_PARSE:
        switch(*p) {
          case '0':
          case '1':
          case '2':
          case '3':
          case '4':
          case '5':
          case '6':
          case '7':
          case '8':
          case '9':
            r->content_len = r->content_len * 10 + (*p - '0');
            break;
          case '\r':
          case ' ':
          case '\t':
            if(r->content_len < 0) {
              LM_ERR("bad Content-Length header value %d in"
                   " state %d\n",
                  r->content_len, r->state);
              r->content_len = 0;
              r->error = TCP_REQ_BAD_LEN;
              r->state = H_SKIP; /* skip now */
            }
            r->state = H_SKIP;
            r->flags |= F_TCP_REQ_HAS_CLEN;
            break;
          case '\n':
            /* end of line, parse successful */
            if(r->content_len < 0) {
              LM_ERR("bad Content-Length header value %d in"
                   " state %d\n",
                  r->content_len, r->state);
              r->content_len = 0;
              r->error = TCP_REQ_BAD_LEN;
              r->state = H_SKIP; /* skip now */
            }
            r->state = H_LF;
            r->flags |= F_TCP_REQ_HAS_CLEN;
            break;
          default:
            LM_ERR("bad Content-Length header value, unexpected "
                 "char %c in state %d\n",
                *p, r->state);
            r->state = H_SKIP; /* try to find another?*/
        }
        p++;
        break;

#ifdef READ_HTTP11
      case H_HTTP11_CHUNK_START: /* start a new body chunk: SIZE\r\nBODY\r\n */
        r->chunk_size = 0;
        r->state = H_HTTP11_CHUNK_SIZE;
        break;
      case H_HTTP11_CHUNK_BODY: /* content of chunk */
        remaining = r->pos - p;
        if(remaining > r->bytes_to_go)
          remaining = r->bytes_to_go;
        r->bytes_to_go -= remaining;
        p += remaining;
        if(r->bytes_to_go == 0) {
          r->state = H_HTTP11_CHUNK_END;
          /* shift back body content */
          if(r->chunk_size > 0 && p - r->chunk_size > r->body) {
            memmove(r->body + r->content_len, p - r->chunk_size,
                r->chunk_size);
            r->content_len += r->chunk_size;
          }
          goto skip;
        }
        break;

      case H_HTTP11_CHUNK_END:
        switch(*p) {
          case '\r':
          case ' ':
          case '\t': /* skip */
            break;
          case '\n':
            r->state = H_HTTP11_CHUNK_START;
            break;
          default:
            LM_ERR("bad chunk, unexpected "
                 "char %c in state %d\n",
                *p, r->state);
            r->state = H_SKIP; /* try to find another?*/
        }
        p++;
        break;

      case H_HTTP11_CHUNK_SIZE:
        switch(*p) {
          case '0':
          case '1':
          case '2':
          case '3':
          case '4':
          case '5':
          case '6':
          case '7':
          case '8':
          case '9':
            r->chunk_size <<= 4;
            r->chunk_size += *p - '0';
            break;
          case 'a':
          case 'b':
          case 'c':
          case 'd':
          case 'e':
          case 'f':
            r->chunk_size <<= 4;
            r->chunk_size += *p - 'a' + 10;
            break;
          case 'A':
          case 'B':
          case 'C':
          case 'D':
          case 'E':
          case 'F':
            r->chunk_size <<= 4;
            r->chunk_size += *p - 'A' + 10;
            break;
          case '\r':
          case ' ':
          case '\t': /* skip */
            break;
          case '\n':
            /* end of line, parse successful */
            r->state = H_HTTP11_CHUNK_BODY;
            r->bytes_to_go = r->chunk_size;
            if(r->bytes_to_go == 0) {
              r->state = H_HTTP11_CHUNK_FINISH;
              r->flags |= F_TCP_REQ_COMPLETE;
              p++;
              goto skip;
            }
            break;
          default:
            LM_ERR("bad chunk size value, unexpected "
                 "char %c in state %d\n",
                *p, r->state);
            r->state = H_SKIP; /* try to find another?*/
        }
        p++;
        break;
#endif
#ifdef READ_MSRP
      case H_MSRP_BODY: /* body of msrp frame */
        /* find lf, we are in this state if we are not interested
         * in anything till end of line*/
        r->flags |= F_TCP_REQ_MSRP_BODY;
        p = q_memchr(p, '\n', r->pos - p);
        if(p) {
          p++;
          r->state = H_MSRP_BODY_LF;
        } else {
          p = r->pos;
        }
        break;
      case H_MSRP_BODY_LF: /* LF in body of msrp frame */
        switch(*p) {
          case '-':
            p--;
            r->state = H_MSRP_BODY_END;
            break;
          default:
            r->state = H_MSRP_BODY;
        }
        p++;
        break;
      case H_MSRP_BODY_END: /* end of body for msrp frame */
        /* find LF and check if it is end-line */
        p = q_memchr(p, '\n', r->pos - p);
        if(p) {
          /* check if it is end line '-------sessid$\r\n' */
          if(r->pos - r->start < 10) {
            LM_ERR("weird situation when reading MSRP frame"
                 " - continue reading\n");
            /* *p=='\n' */
            r->state = H_MSRP_BODY_LF;
            p++;
            break;
          }
          if(*(p - 1) != '\r') {
            /* not ending in '\r\n' - not end-line */
            /* *p=='\n' */
            r->state = H_MSRP_BODY_LF;
            p++;
            break;
          }
          /* locate transaction id in first line
           * -- first line exists, that's why we are here */
          mfline = q_memchr(r->start, '\n', r->pos - r->start);
          mtransid.s = q_memchr(
              r->start + 5 /* 'MSRP ' */, ' ', mfline - r->start);
          mtransid.len = mtransid.s - r->start - 5;
          mtransid.s = r->start + 5;
          trim(&mtransid);
          if(memcmp(mtransid.s,
                 p - 1 /*\r*/ - 1 /* '+'|'#'|'$' */
                     - mtransid.len,
                 mtransid.len)
              != 0) {
            /* no match on session id - not end-line */
            /* *p=='\n' */
            r->state = H_MSRP_BODY_LF;
            p++;
            break;
          }
          if(memcmp(p - 1 /*\r*/ - 1 /* '+'|'#'|'$' */ - mtransid.len
                     - 7 /* 7 x '-' */ - 1 /* '\n' */,
                 "\n-------", 8)
              != 0) {
            /* no match on "\n-------" - not end-line */
            /* *p=='\n' */
            r->state = H_MSRP_BODY_LF;
            p++;
            break;
          }
          r->state = H_MSRP_FINISH;
          r->flags |= F_TCP_REQ_COMPLETE;
          p++;
          goto skip;

        } else {
          p = r->pos;
        }
        break;
#endif

      default:
        LM_CRIT("unexpected state %d\n", r->state);
        abort();
    }
  }
skip:
  r->parsed = p;
  return bytes;
}


#ifdef READ_MSRP
int msrp_process_msg(char *tcpbuf, unsigned int len,
    struct receive_info *rcv_info, struct tcp_connection *con)
{
  int ret;
  tcp_event_info_t tev;
  sr_event_param_t evp = {0};

  ret = 0;
  LM_DBG("MSRP Message: [[>>>\n%.*s<<<]]\n", len, tcpbuf);
  if(likely(sr_event_enabled(SREV_TCP_MSRP_FRAME))) {
    memset(&tev, 0, sizeof(tcp_event_info_t));
    tev.type = SREV_TCP_MSRP_FRAME;
    tev.buf = tcpbuf;
    tev.len = len;
    tev.rcv = rcv_info;
    tev.con = con;
    evp.data = (void *)(&tev);
    ret = sr_event_exec(SREV_TCP_MSRP_FRAME, &evp);
  } else {
    LM_DBG("no callback registering for handling MSRP - dropping!\n");
  }
  return ret;
}
#endif

#ifdef READ_WS
static int tcp_read_ws(struct tcp_connection *c, rd_conn_flags_t *read_flags)
{
  int bytes;
  uint32_t size, pos, mask_present, len;
  char *p;
  struct tcp_req *r;

  r = &c->req;
#ifdef USE_TLS
  if(unlikely(c->type == PROTO_WSS))
    bytes = tls_read(c, read_flags);
  else
#endif
    bytes = tcp_read(c, read_flags);

  if(bytes < 0) {
    /* read error */
    return bytes;
  }
  if(r->parsed == r->pos) {
    /* nothing else to parse */
    return bytes;
  }
  if(r->parsed > r->pos) {
    LM_ERR("req buf pos (%p) before parsed (%p) [%d]\n", r->pos, r->parsed,
        bytes);
    return -1;
  }
  if(r->pos > r->buf + r->b_size) {
    LM_ERR("req pos (%p) over buf (%p / %u) - parsed (%p) [%d]\n", r->pos,
        r->buf, r->b_size, r->parsed, bytes);
    return -1;
  }
  if(r->buf > r->parsed) {
    LM_ERR("req parsed (%p) before buf (%p / %u) - pos (%p) [%d]\n",
        r->parsed, r->buf, r->b_size, r->pos, bytes);
    return -1;
  }

  size = r->pos - r->parsed;

  p = r->parsed;
  pos = 0;

  /*
   0                   1                   2                   3
   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  +-+-+-+-+-------+-+-------------+-------------------------------+
  |F|R|R|R| opcode|M| Payload len |    Extended payload length    |
  |I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
  |N|V|V|V|       |S|             |   (if payload len==126/127)   |
  | |1|2|3|       |K|             |                               |
  +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
  |     Extended payload length continued, if payload len == 127  |
  + - - - - - - - - - - - - - - - +-------------------------------+
  |                               |Masking-key, if MASK set to 1  |
  +-------------------------------+-------------------------------+
  | Masking-key (continued)       |          Payload Data         |
  +-------------------------------- - - - - - - - - - - - - - - - +
  :                     Payload Data continued ...                :
  + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
  |                     Payload Data continued ...                |
  +---------------------------------------------------------------+

  Do minimal parse required to make sure the full message has been
  received (websocket module will do full parse and validation).
  */

  /* Process first two bytes */
  if(size < pos + 2)
    goto skip;
  pos++;
  mask_present = p[pos] & 0x80;
  len = (p[pos] & 0xff) & ~0x80;
  pos++;

  /* Work out real length */
  if(len == 126) {
    /* 2 bytes store the payload size */
    if(size < pos + 2)
      goto skip;

    len = ((p[pos + 0] & 0xff) << 8) | ((p[pos + 1] & 0xff) << 0);
    pos += 2;
  } else if(len == 127) {
    /* 8 bytes store the payload size */
    if(size < pos + 8) {
      goto skip;
    }

    /* Only decoding the last four bytes of the length...
       This limits the size of WebSocket messages that can be
       handled to 2^32 - which should be plenty for SIP! */
    if((p[pos] & 0xff) != 0 || (p[pos + 1] & 0xff) != 0
        || (p[pos + 2] & 0xff) != 0 || (p[pos + 3] & 0xff) != 0) {
      LM_WARN("advertised length is too large (more than 2^32)\n");
      goto skip;
    }
    len = ((p[pos + 4] & 0xff) << 24) | ((p[pos + 5] & 0xff) << 16)
        | ((p[pos + 6] & 0xff) << 8) | ((p[pos + 7] & 0xff) << 0);
    pos += 8;
  }

  /* Skip mask */
  if(mask_present) {
    if(size < pos + 4)
      goto skip;
    pos += 4;
  }

  /* check if advertised length fits in read buffer */
  if(len >= r->b_size) {
    LM_WARN("advertised length (%u) greater than buffer size (%u)\n", len,
        r->b_size);
    goto skip;
  }
  /* Now check the whole message has been received */
  if(size < pos + len)
    goto skip;

  pos += len;
  r->flags |= F_TCP_REQ_COMPLETE;
  r->parsed = &p[pos];

skip:
  return bytes;
}

static int ws_process_msg(char *tcpbuf, unsigned int len,
    struct receive_info *rcv_info, struct tcp_connection *con)
{
  int ret;
  tcp_event_info_t tev;
  sr_event_param_t evp = {0};

  ret = 0;
  LM_DBG("WebSocket Message: [[>>>\n%.*s<<<]]\n", len, tcpbuf);
  if(likely(sr_event_enabled(SREV_TCP_WS_FRAME_IN))) {
    memset(&tev, 0, sizeof(tcp_event_info_t));
    tev.type = SREV_TCP_WS_FRAME_IN;
    tev.buf = tcpbuf;
    tev.len = len;
    tev.rcv = rcv_info;
    tev.con = con;
    evp.data = (void *)(&tev);
    ret = sr_event_exec(SREV_TCP_WS_FRAME_IN, &evp);
  } else {
    LM_DBG("no callback registering for handling WebSockets - dropping!\n");
  }
  return ret;
}
#endif

static int tcp_read_hep3(struct tcp_connection *c, rd_conn_flags_t *read_flags)
{
  int bytes;
  uint32_t size, len;
  char *p;
  struct tcp_req *r;

  r = &c->req;
#ifdef USE_TLS
  if(unlikely(c->type == PROTO_TLS)) {
    bytes = tls_read(c, read_flags);
  } else {
#endif
    bytes = tcp_read(c, read_flags);
#ifdef USE_TLS
  }
#endif

  if(bytes <= 0) {
    if(likely(r->parsed >= r->pos)) {
      LM_DBG("no new bytes to read, but still unparsed content\n");
      return 0;
    }
  }

  size = r->pos - r->parsed;

  p = r->parsed;

  /* Process first six bytes (HEP3 + 2 bytes the size)*/
  if(size < 6) {
    LM_DBG("not enough bytes to parse (%u)\n", size);
    goto skip;
  }

  if(p[0] != 'H' || p[1] != 'E' || p[2] != 'P' || p[3] != '3') {
    /* not hep3 */
    LM_DBG("not HEP3 packet header (%u): %c %c %c %c / %x %x %x %x\n", size,
        p[0], p[1], p[2], p[3], p[0], p[1], p[2], p[3]);
    goto skip;
  }
  r->flags |= F_TCP_REQ_HEP3;

  len = ((uint32_t)(p[4] & 0xff) << 8) + (p[5] & 0xff);

  /* check if advertised length fits in read buffer */
  if(len >= r->b_size) {
    LM_WARN("advertised length (%u) greater than buffer size (%u)\n", len,
        r->b_size);
    goto skip;
  }
  /* check the whole message has been received */
  if(size < len) {
    LM_DBG("incomplete HEP3 packet (%u / %u)\n", len, size);
    goto skip;
  }

  r->flags |= F_TCP_REQ_COMPLETE;
  r->parsed = p + len;
  LM_DBG("reading of HEP3 packet is complete (%u / %u)\n", len, size);

skip:
  return bytes;
}

static int hep3_process_msg(char *tcpbuf, unsigned int len,
    struct receive_info *rcv_info, struct tcp_connection *con)
{
  sip_msg_t msg;
  int ret;
  sr_event_param_t evp = {0};

  memset(&msg, 0, sizeof(sip_msg_t)); /* init everything to 0 */
  /* fill in msg */
  msg.buf = tcpbuf;
  msg.len = len;
  /* zero termination (termination of orig message below not that
   * useful as most of the work is done with scratch-pad; -jiri  */
  /* buf[len]=0; */ /* WARNING: zero term removed! */
  msg.rcv = *rcv_info;
  msg.id = msg_no;
  msg.pid = my_pid();
  msg.set_global_address = default_global_address;
  msg.set_global_port = default_global_port;

  if(likely(sr_msg_time == 1))
    msg_set_time(&msg);
  evp.data = (void *)(&msg);
  ret = sr_event_exec(SREV_RCV_NOSIP, &evp);
  LM_DBG("running hep3 handling event returned %d\n", ret);
  if(ret == NONSIP_MSG_DROP) {
    free_sip_msg(&msg);
    return 0;
  }
  if(ret < 0) {
    LM_ERR("error running hep3 handling event: %d\n", ret);
    free_sip_msg(&msg);
    return -1;
  }

  ret = receive_msg(msg.buf, msg.len, &msg.rcv);
  LM_DBG("running hep3-enclosed sip request route returned %d\n", ret);
  free_sip_msg(&msg);

  return ret;
}

/**
 * @brief wrapper around receive_msg() to clone the tcpbuf content
 *
 * When receiving over TCP, tcpbuf points inside the TCP stream buffer, but during
 * processing of config, msg->buf content might be changed and may corrupt
 * the content of the stream. Safer, make a clone of buf content in a local
 * buffer and give that to receive_msg() to link to msg->buf
 */
int receive_tcp_msg(char *tcpbuf, unsigned int len,
    struct receive_info *rcv_info, struct tcp_connection *con)
{
  int ret = 0;
#ifdef TCP_CLONE_RCVBUF
  static char *buf = NULL;
  static unsigned int bsize = 0;
  int blen;

  /* cloning is disabled via parameter */
  if(likely(tcp_clone_rcvbuf == 0)) {
#ifdef READ_MSRP
    if(unlikely(con->req.flags & F_TCP_REQ_MSRP_FRAME))
      return msrp_process_msg(tcpbuf, len, rcv_info, con);
#endif
#ifdef READ_WS
    if(unlikely(con->type == PROTO_WS || con->type == PROTO_WSS))
      return ws_process_msg(tcpbuf, len, rcv_info, con);
#endif
    if(unlikely(con->req.flags & F_TCP_REQ_HEP3))
      return hep3_process_msg(tcpbuf, len, rcv_info, con);

    ret = receive_msg(tcpbuf, len, rcv_info);
    if(ksr_tcp_script_mode & TCP_SCRIPT_MODE_CONTINUE) {
      return 0;
    }
    return ret;
  }

  /* min buffer size is BUF_SIZE */
  blen = len;
  if(blen < BUF_SIZE)
    blen = BUF_SIZE;

  /* allocate buffer when needed
   * - no buffer yet
   * - existing buffer too small (min size is BUF_SIZE - to accommodate most
   *   of SIP messages; expected larger for HTTP/XCAP)
   * - existing buffer too large (e.g., we got a too big message in the past,
   *   let's free it)
   *
   * - also, use system memory, not to eat from PKG (same as static buffer
   *   from PKG pov)
   */
  if(buf == NULL || bsize < blen || blen < bsize / 2) {
    if(buf != NULL)
      free(buf);
    buf = malloc(blen + 1);
    if(buf == 0) {
      SYS_MEM_ERROR;
      return -1;
    }
    bsize = blen;
  }

  memcpy(buf, tcpbuf, len);
  buf[len] = '\0';
#ifdef READ_MSRP
  if(unlikely(con->req.flags & F_TCP_REQ_MSRP_FRAME))
    return msrp_process_msg(buf, len, rcv_info, con);
#endif
#ifdef READ_WS
  if(unlikely(con->type == PROTO_WS || con->type == PROTO_WSS))
    return ws_process_msg(buf, len, rcv_info, con);
#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);
  if(ksr_tcp_script_mode & TCP_SCRIPT_MODE_CONTINUE) {
    return 0;
  }
  return ret;
#else /* TCP_CLONE_RCVBUF */
#ifdef READ_MSRP
  if(unlikely(con->req.flags & F_TCP_REQ_MSRP_FRAME))
    return msrp_process_msg(tcpbuf, len, rcv_info, con);
#endif
#ifdef READ_WS
  if(unlikely(con->type == PROTO_WS || con->type == PROTO_WSS))
    return ws_process_msg(tcpbuf, len, rcv_info, con);
#endif
  if(unlikely(con->req.flags & F_TCP_REQ_HEP3))
    return hep3_process_msg(tcpbuf, len, rcv_info, con);
  ret = receive_msg(tcpbuf, len, rcv_info);
  if(ksr_tcp_script_mode & TCP_SCRIPT_MODE_CONTINUE) {
    return 0;
  }
  return ret;
#endif /* TCP_CLONE_RCVBUF */
}

int tcp_read_req(struct tcp_connection *con, int *bytes_read,
    rd_conn_flags_t *read_flags)
{
  int bytes;
  int total_bytes;
  int resp;
  long size;
  struct tcp_req *req;
  struct dest_info dst;
  char c;
  int ret;

  bytes = -1;
  total_bytes = 0;
  resp = CONN_RELEASE;
  req = &con->req;
  if(req->tvrstart.tv_sec == 0) {
    gettimeofday(&req->tvrstart, NULL);
  }

again:
  if(likely(req->error == TCP_REQ_OK)) {
#ifdef READ_WS
    if(unlikely(con->type == PROTO_WS || con->type == PROTO_WSS)) {
      bytes = tcp_read_ws(con, read_flags);
    } else {
#endif
      if(unlikely(ksr_tcp_accept_hep3 != 0)) {
        bytes = tcp_read_hep3(con, read_flags);
        if(bytes >= 0) {
          if(!(con->req.flags & F_TCP_REQ_HEP3)) {
            /* not hep3, try to read headers */
            bytes = tcp_read_headers(con, read_flags);
          }
        }
      } else {
        bytes = tcp_read_headers(con, read_flags);
      }
#ifdef READ_WS
    }
#endif

    if(ksr_msg_recv_max_size <= (int)(req->parsed - req->start)) {
      LOG(cfg_get(core, core_cfg, corelog),
          "read message too large: %d - c: %p r: %p (%d)\n",
          (int)(req->parsed - req->start), con, req, bytes);
      resp = CONN_ERROR;
      goto end_req;
    }

    if(unlikely(bytes < 0)) {
      LOG(cfg_get(core, core_cfg, corelog),
          "ERROR: tcp_read_req: error reading - c: %p r: %p (%d)\n",
          con, req, bytes);
      resp = CONN_ERROR;
      goto end_req;
    }

#ifdef EXTRA_DEBUG
    /* if timeout state=0; goto end__req; */
    LM_DBG("read= %d bytes, parsed=%d, state=%d, error=%d\n", bytes,
        (int)(req->parsed - req->start), req->state, req->error);
    LM_DBG("last char=0x%02X, parsed msg=\n%.*s\n", *(req->parsed - 1),
        (int)(req->parsed - req->start), req->start);
#endif
    total_bytes += bytes;
    /* eof check:
       * is EOF if eof on fd and req.  not complete yet,
       * if req. is complete we might have a second unparsed
       * request after it, so postpone release_with_eof
       */
    if(unlikely((con->state == S_CONN_EOF) && (!TCP_REQ_COMPLETE(req)))) {
      LM_DBG("EOF\n");
      resp = CONN_EOF;
      goto end_req;
    }
  }
  if(unlikely(req->error != TCP_REQ_OK)) {
    if(req->buf != NULL && req->start != NULL && req->pos != NULL
        && req->pos >= req->buf && req->parsed >= req->start) {
      LM_ERR("bad request, state=%d, error=%d "
           "buf:\n%.*s\nparsed:\n%.*s\n",
          req->state, req->error, (int)(req->pos - req->buf),
          req->buf, (int)(req->parsed - req->start), req->start);
    } else {
      LM_ERR("bad request, state=%d, error=%d buf:%d - %p,"
           " parsed:%d - %p\n",
          req->state, req->error, (int)(req->pos - req->buf),
          req->buf, (int)(req->parsed - req->start), req->start);
    }
    LM_DBG("received from: port %d\n", con->rcv.src_port);
    print_ip("received from: ip", &con->rcv.src_ip, "\n");
    resp = CONN_ERROR;
    goto end_req;
  }
  if(likely(TCP_REQ_COMPLETE(req))) {
#ifdef EXTRA_DEBUG
    LM_DBG("end of header part\n");
    LM_DBG("received from: port %d\n", con->rcv.src_port);
    print_ip("received from: ip", &con->rcv.src_ip, "\n");
    LM_DBG("headers:\n%.*s.\n", (int)(req->body - req->start), req->start);
#endif
    if(likely(TCP_REQ_HAS_CLEN(req))) {
      LM_DBG("content-length=%d\n", req->content_len);
#ifdef EXTRA_DEBUG
      LM_DBG("body:\n%.*s\n", req->content_len, req->body);
#endif
    } else {
      if(cfg_get(tcp, tcp_cfg, accept_no_cl) == 0) {
        req->error = TCP_REQ_BAD_LEN;
        LM_ERR("content length not present or unparsable\n");
        resp = CONN_ERROR;
        goto end_req;
      }
    }
    /* if we are here everything is nice and ok*/
    resp = CONN_RELEASE;
    if(req->state != H_PING_CRLF) {
      req->dxstate |= KSR_TCP_REQSTATE_DATARECV;
    }
#ifdef EXTRA_DEBUG
    LM_DBG("receiving msg(%p, %d)\n", req->start,
        (int)(req->parsed - req->start));
#endif
    /* rcv.bind_address should always be !=0 */
    bind_address = con->rcv.bind_address;

    con->rcv.proto_reserved1 = con->id; /* copy the id */
    c = *req->parsed; /* ugly hack: zero term the msg & save the
                 previous char, req->parsed should be ok
                 because we always alloc BUF_SIZE+1 */
    *req->parsed = 0;

    if(req->state == H_PING_CRLF) {
      init_dst_from_rcv(&dst, &con->rcv);

      if(tcp_send(&dst, 0, CRLF, CRLF_LEN) < 0) {
        LM_ERR("CRLF ping: tcp_send() failed ([%s]:%u -> [%s]:%u)\n",
            ip_addr2a(&con->rcv.src_ip), con->rcv.src_port,
            ip_addr2a(&con->rcv.dst_ip), con->rcv.dst_port);
      }
      ret = 0;
    } else if(unlikely(req->state == H_STUN_END)) {
      /* stun request */
      ret = stun_process_msg(
          req->start, req->parsed - req->start, &con->rcv);
    } else
#ifdef READ_MSRP
      // if (unlikely(req->flags&F_TCP_REQ_MSRP_FRAME)){
      if(unlikely(req->state == H_MSRP_FINISH)) {
        /* msrp frame */
        ret = receive_tcp_msg(
            req->start, req->parsed - req->start, &con->rcv, con);
      } else
#endif
#ifdef READ_HTTP11
          if(unlikely(req->state == H_HTTP11_CHUNK_FINISH)) {
        /* http chunked request */
        req->body[req->content_len] = 0;
        ret = receive_tcp_msg(req->start,
            req->body + req->content_len - req->start, &con->rcv,
            con);
      } else
#endif
#ifdef READ_WS
          if(unlikely(con->type == PROTO_WS
                || con->type == PROTO_WSS)) {
        ret = receive_tcp_msg(
            req->start, req->parsed - req->start, &con->rcv, con);
      } else
#endif
        ret = receive_tcp_msg(
            req->start, req->parsed - req->start, &con->rcv, con);

    if(unlikely(ret < 0)) {
      *req->parsed = c;
      resp = CONN_ERROR;
      goto end_req;
    }
    *req->parsed = c;

    /* prepare for next request */
    size = req->pos - req->parsed;
    req->start = req->buf;
    req->body = 0;
    req->error = TCP_REQ_OK;
    req->state = H_SKIP_EMPTY;
    req->flags = 0;
    req->content_len = 0;
    req->bytes_to_go = 0;
    req->pos = req->buf + size;
    req->tvrstart.tv_sec = 0;
    req->tvrstart.tv_usec = 0;

    if(unlikely(size)) {
      gettimeofday(&req->tvrstart, NULL);
      memmove(req->buf, req->parsed, size);
      req->parsed = req->buf; /* fix req->parsed after using it */
#ifdef EXTRA_DEBUG
      LM_DBG("preparing for new request, kept %ld bytes\n", size);
#endif
      /*if we still have some unparsed bytes, try to parse them too*/
      goto again;
    } else if(unlikely(con->state == S_CONN_EOF)) {
      LM_DBG("EOF after reading complete request ([%s]:%u -> [%s]:%u)\n",
          ip_addr2a(&con->rcv.src_ip), con->rcv.src_port,
          ip_addr2a(&con->rcv.dst_ip), con->rcv.dst_port);
      resp = CONN_EOF;
    }
    req->parsed = req->buf; /* fix req->parsed */
  }

end_req:
  if(likely(bytes_read))
    *bytes_read = total_bytes;
  return resp;
}


void release_tcpconn(struct tcp_connection *c, long state, int unix_sock)
{
  long response[2];

  LM_DBG("releasing con %p, state %ld, fd=%d, id=%d ([%s]:%u -> [%s]:%u)\n",
      c, state, c->fd, c->id, ip_addr2a(&c->rcv.src_ip), c->rcv.src_port,
      ip_addr2a(&c->rcv.dst_ip), c->rcv.dst_port);
  LM_DBG("extra_data %p\n", c->extra_data);
  /* release req & signal the parent */
  c->reader_pid = 0; /* reset it */
  if(c->fd != -1) {
    close(c->fd);
    c->fd = -1;
  }
  /* errno==EINTR, EWOULDBLOCK a.s.o todo */
  response[0] = (long)c;
  response[1] = state;

  if(tsend_stream(unix_sock, (char *)response, sizeof(response), -1) <= 0)
    LM_ERR("tsend_stream failed\n");
}


static ticks_t tcpconn_read_timeout(ticks_t t, struct timer_ln *tl, void *data)
{
  struct tcp_connection *c;

  c = (struct tcp_connection *)data;
  /* or (struct tcp...*)(tl-offset(c->timer)) */

  if(likely(!(c->state < 0) && TICKS_LT(t, c->timeout))) {
    /* timeout extended, exit */
    return (ticks_t)(c->timeout - t);
  }
  /* if conn->state is ERROR or BAD => force timeout too */
  if(unlikely(io_watch_del(&io_w, c->fd, -1, IO_FD_CLOSING) < 0)) {
    LM_ERR("io_watch_del failed for %p"
         " id %d fd %d, state %d, flags %x, main fd %d"
         " ([%s]:%u -> [%s]:%u)\n",
        c, c->id, c->fd, c->state, c->flags, c->s,
        ip_addr2a(&c->rcv.src_ip), c->rcv.src_port,
        ip_addr2a(&c->rcv.dst_ip), c->rcv.dst_port);
  }
  if(tcp_conn_lst != NULL) {
    tcpconn_listrm(tcp_conn_lst, c, c_next, c_prev);
    c->event = TCP_CLOSED_TIMEOUT;
    release_tcpconn(
        c, (c->state < 0) ? CONN_ERROR : CONN_RELEASE, tcpmain_sock);
  }
  return 0;
}


/* handle io routine, based on the fd_map type
 * (it will be called from io_wait_loop* )
 * params:  fm  - pointer to a fd hash entry
 *          idx - index in the fd_array (or -1 if not known)
 * return: -1 on error, or when we are not interested any more on reads
 *            from this fd (e.g.: we are closing it )
 *          0 on EAGAIN or when by some other way it is known that no more
 *            io events are queued on the fd (the receive buffer is empty).
 *            Useful to detect when there are no more io events queued for
 *            sigio_rt, epoll_et, kqueue.
 *         >0 on successfull read from the fd (when there might be more io
 *            queued -- the receive buffer might still be non-empty)
 */
inline static int handle_io(struct fd_map *fm, short events, int idx)
{
  int ret;
  int n;
  rd_conn_flags_t read_flags;
  struct tcp_connection *con;
  int s;
  long resp;
  ticks_t t;
  fd_map_t *ee = NULL;

  /* update the local config */
  cfg_update();

  switch(fm->type) {
    case F_TCPMAIN:
    again:
      ret = n = receive_fd(fm->fd, &con, sizeof(con), &s, 0);
      LM_DBG("received n=%d con=%p, fd=%d\n", n, con, s);
      if(unlikely(n < 0)) {
        if(errno == EWOULDBLOCK || errno == EAGAIN) {
          ret = 0;
          break;
        } else if(errno == EINTR)
          goto again;
        else {
          LM_CRIT("read_fd: %s \n", strerror(errno));
          abort(); /* big error*/
        }
      }
      if(unlikely(n == 0)) {
        LM_ERR("0 bytes read\n");
        goto error;
      }
      if(unlikely(con == 0)) {
        LM_CRIT("null pointer\n");
        goto error;
      }
      con->fd = s;
      if(unlikely(s == -1)) {
        LM_ERR("read_fd: no fd read\n");
        goto con_error;
      }
      con->reader_pid = my_pid();
      if(unlikely(con == tcp_conn_lst)) {
        LM_CRIT("duplicate connection received: %p, id %d, fd %d, "
            "refcnt %d"
            " state %d (n=%d)\n",
            con, con->id, con->fd, atomic_get(&con->refcnt),
            con->state, n);
        goto con_error;
        break; /* try to recover */
      }
      if(unlikely(con->state == S_CONN_BAD)) {
        LM_WARN("received an already bad connection: %p id %d refcnt "
            "%d\n",
            con, con->id, atomic_get(&con->refcnt));
        goto con_error;
      }
      /* if we received the fd there is most likely data waiting to
       * be read => process it first to avoid extra sys calls */
      read_flags = ((con->flags & (F_CONN_EOF_SEEN | F_CONN_FORCE_EOF))
                 && !(con->flags & F_CONN_OOB_DATA))
                 ? RD_CONN_FORCE_EOF
                 : 0;
#ifdef USE_TLS
    repeat_1st_read:
#endif /* USE_TLS */
      resp = tcp_read_req(con, &n, &read_flags);
      if(unlikely(resp < 0)) {
        /* some error occurred, but on the new fd, not on the tcp
         * main fd, so keep the ret value */
        if(unlikely(resp != CONN_EOF))
          con->state = S_CONN_BAD;
        release_tcpconn(con, resp, tcpmain_sock);
        break;
      }
#ifdef USE_TLS
      /* repeat read if requested (for now only tls might do this) */
      if(unlikely(read_flags & RD_CONN_REPEAT_READ))
        goto repeat_1st_read;
#endif /* USE_TLS */

      /* must be before io_watch_add, io_watch_add might catch some
       * already existing events => might call handle_io and
       * handle_io might decide to del. the new connection =>
       * must be in the list */
      tcpconn_listadd(tcp_conn_lst, con, c_next, c_prev);
      t = get_ticks_raw();
      con->timeout = t + S_TO_TICKS(TCP_CHILD_TIMEOUT);
      /* re-activate the timer */
      con->timer.f = tcpconn_read_timeout;
      local_timer_reinit(&con->timer);
      local_timer_add(&tcp_reader_ltimer, &con->timer,
          S_TO_TICKS(TCP_CHILD_TIMEOUT), t);
      if(unlikely(io_watch_add(&io_w, s, POLLIN, F_TCPCONN, con) < 0)) {
        LM_CRIT("io_watch_add failed for %p id %d fd %d, state %d, "
            "flags %x,"
            " main fd %d, refcnt %d ([%s]:%u -> [%s]:%u)\n",
            con, con->id, con->fd, con->state, con->flags, con->s,
            atomic_get(&con->refcnt), ip_addr2a(&con->rcv.src_ip),
            con->rcv.src_port, ip_addr2a(&con->rcv.dst_ip),
            con->rcv.dst_port);
        ee = get_fd_map(&io_w, s);
        if(ee != 0 && ee->type == F_TCPCONN) {
          tcp_connection_t *ec;
          ec = (tcp_connection_t *)ee->data;
          LM_CRIT("existing tcp con %p id %d fd %d, state %d, flags "
              "%x,"
              " main fd %d, refcnt %d ([%s]:%u -> [%s]:%u)\n",
              ec, ec->id, ec->fd, ec->state, ec->flags, ec->s,
              atomic_get(&ec->refcnt), ip_addr2a(&ec->rcv.src_ip),
              ec->rcv.src_port, ip_addr2a(&ec->rcv.dst_ip),
              ec->rcv.dst_port);
        }
        if(tcp_conn_lst != NULL) {
          tcpconn_listrm(tcp_conn_lst, con, c_next, c_prev);
          local_timer_del(&tcp_reader_ltimer, &con->timer);
        }
        goto con_error;
      }
      break;
    case F_TCPCONN:
      con = (struct tcp_connection *)fm->data;
      if(unlikely(con->state == S_CONN_BAD)) {
        resp = CONN_ERROR;
        if(!(con->send_flags.f & SND_F_CON_CLOSE))
          LM_WARN("F_TCPCONN connection marked as bad: %p id %d fd %d"
              " refcnt %d ([%s]:%u -> [%s]:%u)\n",
              con, con->id, con->fd, atomic_get(&con->refcnt),
              ip_addr2a(&con->rcv.src_ip), con->rcv.src_port,
              ip_addr2a(&con->rcv.dst_ip), con->rcv.dst_port);
        goto read_error;
      }
      read_flags =
          ((
#ifdef POLLRDHUP
               (events & POLLRDHUP) |
#endif /* POLLRDHUP */
               (events & (POLLHUP | POLLERR))
               | (con->flags
                   & (F_CONN_EOF_SEEN | F_CONN_FORCE_EOF)))
              && !(events & POLLPRI))
              ? RD_CONN_FORCE_EOF
              : 0;
#ifdef USE_TLS
    repeat_read:
#endif /* USE_TLS */
      resp = tcp_read_req(con, &ret, &read_flags);
      if(unlikely(resp < 0)) {
      read_error:
        ret = -1; /* some error occurred */
        if(unlikely(io_watch_del(&io_w, con->fd, idx, IO_FD_CLOSING)
              < 0)) {
          LM_CRIT("io_watch_del failed for %p id %d fd %d,"
              " state %d, flags %x, main fd %d, refcnt %d"
              " ([%s]:%u -> [%s]:%u)\n",
              con, con->id, con->fd, con->state, con->flags,
              con->s, atomic_get(&con->refcnt),
              ip_addr2a(&con->rcv.src_ip), con->rcv.src_port,
              ip_addr2a(&con->rcv.dst_ip), con->rcv.dst_port);
        }
        if(tcp_conn_lst != NULL) {
          LM_DBG("removing from list %p id %d fd %d,"
               " state %d, flags %x, main fd %d, refcnt %d"
               " ([%s]:%u -> [%s]:%u)\n",
              con, con->id, con->fd, con->state, con->flags,
              con->s, atomic_get(&con->refcnt),
              ip_addr2a(&con->rcv.src_ip), con->rcv.src_port,
              ip_addr2a(&con->rcv.dst_ip), con->rcv.dst_port);
          tcpconn_listrm(tcp_conn_lst, con, c_next, c_prev);
          local_timer_del(&tcp_reader_ltimer, &con->timer);
          if(unlikely(resp != CONN_EOF))
            con->state = S_CONN_BAD;
          release_tcpconn(con, resp, tcpmain_sock);
        }
      } else {
#ifdef USE_TLS
        if(unlikely(read_flags & RD_CONN_REPEAT_READ))
          goto repeat_read;
#endif /* USE_TLS */
        /* update timeout */
        con->timeout = get_ticks_raw() + S_TO_TICKS(TCP_CHILD_TIMEOUT);
        /* ret= 0 (read the whole socket buffer) if short read
         * & !POLLPRI,  bytes read otherwise */
        ret &= (((read_flags & RD_CONN_SHORT_READ)
                && !(events & POLLPRI))
            - 1);
      }
      break;
    case F_NONE:
      LM_CRIT("empty fd map %p (%d): {%d, %d, %p}\n", fm,
          (int)(fm - io_w.fd_hash), fm->fd, fm->type, fm->data);
      goto error;
    default:
      LM_CRIT("unknown fd type %d\n", fm->type);
      goto error;
  }

  return ret;
con_error:
  con->state = S_CONN_BAD;
  release_tcpconn(con, CONN_ERROR, tcpmain_sock);
  return ret;
error:
  return -1;
}


inline static void tcp_reader_timer_run(void)
{
  ticks_t ticks;

  ticks = get_ticks_raw();
  if(unlikely((ticks - tcp_reader_prev_ticks) < TCPCONN_TIMEOUT_MIN_RUN))
    return;
  tcp_reader_prev_ticks = ticks;
  local_timer_run(&tcp_reader_ltimer, ticks);
}


void tcp_receive_loop(int unix_sock)
{

  /* init */
  tcpmain_sock = unix_sock; /* init com. socket */
  if(init_io_wait(&io_w, get_max_open_fds(), tcp_poll_method) < 0)
    goto error;
  tcp_reader_prev_ticks = get_ticks_raw();
  if(init_local_timer(&tcp_reader_ltimer, get_ticks_raw()) != 0)
    goto error;
  /* add the unix socket */
  if(io_watch_add(&io_w, tcpmain_sock, POLLIN, F_TCPMAIN, 0) < 0) {
    LM_CRIT("failed to add tcp main socket to the fd list\n");
    goto error;
  }

  /* initialize the config framework */
  if(cfg_child_init())
    goto error;

  /* main loop */
  switch(io_w.poll_method) {
    case POLL_POLL:
      while(1) {
        io_wait_loop_poll(&io_w, TCP_CHILD_SELECT_TIMEOUT, 0);
        tcp_reader_timer_run();
      }
      break;
#ifdef HAVE_SELECT
    case POLL_SELECT:
      while(1) {
        io_wait_loop_select(&io_w, TCP_CHILD_SELECT_TIMEOUT, 0);
        tcp_reader_timer_run();
      }
      break;
#endif
#ifdef HAVE_SIGIO_RT
    case POLL_SIGIO_RT:
      while(1) {
        io_wait_loop_sigio_rt(&io_w, TCP_CHILD_SELECT_TIMEOUT);
        tcp_reader_timer_run();
      }
      break;
#endif
#ifdef HAVE_EPOLL
    case POLL_EPOLL_LT:
      while(1) {
        io_wait_loop_epoll(&io_w, TCP_CHILD_SELECT_TIMEOUT, 0);
        tcp_reader_timer_run();
      }
      break;
    case POLL_EPOLL_ET:
      while(1) {
        io_wait_loop_epoll(&io_w, TCP_CHILD_SELECT_TIMEOUT, 1);
        tcp_reader_timer_run();
      }
      break;
#endif
#ifdef HAVE_KQUEUE
    case POLL_KQUEUE:
      while(1) {
        io_wait_loop_kqueue(&io_w, TCP_CHILD_SELECT_TIMEOUT, 0);
        tcp_reader_timer_run();
      }
      break;
#endif
#ifdef HAVE_DEVPOLL
    case POLL_DEVPOLL:
      while(1) {
        io_wait_loop_devpoll(&io_w, TCP_CHILD_SELECT_TIMEOUT, 0);
        tcp_reader_timer_run();
      }
      break;
#endif
    default:
      LM_CRIT("no support for poll method %s (%d)\n",
          poll_method_name(io_w.poll_method), io_w.poll_method);
      goto error;
  }
error:
  destroy_io_wait(&io_w);
  LM_CRIT("exiting...");
  ksr_exit(-1);
}


int is_msg_complete(struct tcp_req *r)
{
  if(TCP_REQ_HAS_CLEN(r)) {
    r->state = H_STUN_FP;
    return 0;
  } else {
    /* STUN message is complete */
    r->state = H_STUN_END;
    r->flags |= F_TCP_REQ_COMPLETE
          | F_TCP_REQ_HAS_CLEN; /* hack to avoid error check */
    return 1;
  }
}

#endif /* USE_TCP */

Coverage Report

Created: 2024-02-25 06:34