ntopng/src/SyslogCollectorInterface.cpp

/*
 *
 * (C) 2013-23 - ntop.org
 *
 *
 * This program 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 3 of the License, or
 * (at your option) any later version.
 *
 * This program 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 "ntop_includes.h"

#ifndef HAVE_NEDGE

/* **************************************************** */

bool SyslogCollectorInterface::openSocket(syslog_socket *ss,
                                          const char *server_address,
                                          int server_port, int protocol) {
  struct sockaddr_in listen_addr;
  int reuse = 1;

  ntop->getTrace()->traceEvent(
      TRACE_NORMAL, "Starting %s syslog collector on %s:%d",
      protocol == SOCK_DGRAM ? "UDP" : "TCP", server_address, server_port);

  ss->sock =
      Utils::openSocket(AF_INET, protocol, 0, "SyslogCollectorInterface");

  if (ss->sock < 0) {
    ntop->getTrace()->traceEvent(TRACE_ERROR, "socket error");
    return false;
  }

  /* Allow to re-bind in case previous instance died */
  if (setsockopt(ss->sock, SOL_SOCKET, SO_REUSEADDR,
#ifdef WIN32
                 (const char *)
#endif
                 &reuse,
                 sizeof(reuse)) != 0) {
    ntop->getTrace()->traceEvent(TRACE_ERROR, "setsockopt error");
    return false;
  }

  memset(&listen_addr, 0, sizeof(listen_addr));

  listen_addr.sin_family = AF_INET;
  listen_addr.sin_addr.s_addr = inet_addr(server_address);
  listen_addr.sin_port = htons(server_port);

  if (::bind(ss->sock, (struct sockaddr *)&listen_addr,
             sizeof(struct sockaddr)) != 0) {
    ntop->getTrace()->traceEvent(TRACE_ERROR, "bind error");
    return false;
  }

  if (protocol == SOCK_STREAM) {
    if (listen(ss->sock, MAX_SYSLOG_SUBSCRIBERS) != 0) {
      ntop->getTrace()->traceEvent(TRACE_ERROR, "listen error");
      return false;
    }

    memset(tcp_connections, 0, sizeof(tcp_connections));
  }

  ntop->getTrace()->traceEvent(
      TRACE_NORMAL, "Accepting %s connections on %s:%d",
      protocol == SOCK_DGRAM ? "UDP" : "TCP", server_address, server_port);

  return true;
}

/* **************************************************** */

void SyslogCollectorInterface::closeSocket(syslog_socket *ss, int protocol) {
  close(ss->sock);

  if (protocol == SOCK_STREAM) {
    for (int i = 0; i < MAX_SYSLOG_SUBSCRIBERS; ++i)
      if (tcp_connections[i].socket != 0)
        Utils::closeSocket(tcp_connections[i].socket);
  }
}

/* **************************************************** */

SyslogCollectorInterface::SyslogCollectorInterface(const char *_endpoint)
    : SyslogParserInterface(_endpoint) {
  char *tmp, *pos, *port, *address, *protocol;
  const char *server_address;
  int server_port;

  udp_socket.enable = true;
  tcp_socket.enable = true;

  endpoint = strdup(_endpoint);

  if (endpoint == NULL) throw("memory allocation error");

  tmp = strdup(_endpoint);

  if (tmp == NULL) throw("memory allocation error");

  /*
   * Interface name format:
   * syslog://<ip>:<port>[@{udp,tcp}]
   */

  if (strncmp(tmp, (char *)"syslog://", 9) == 0) {
    address = &tmp[9];
  } else {
    address = tmp;
  }

  pos = strchr(address, '@');

  if (pos != NULL) {
    pos[0] = '\0';
    pos++;
    protocol = pos;

    if (strcmp(protocol, "udp") == 0)
      tcp_socket.enable = false;
    else if (strcmp(protocol, "tcp") == 0)
      udp_socket.enable = false;
  }

  port = strchr(address, ':');

  if (port != NULL) {
    port[0] = '\0';
    port++;
    server_port = atoi(port);
  } else {
    throw("bad tcp bind address format");
  }

  if (strcmp(address, "*") == 0) {
    /* any address */
    server_address = "0.0.0.0";
  } else {
    server_address = address;
  }

  if (udp_socket.enable)
    if (!openSocket(&udp_socket, server_address, server_port, SOCK_DGRAM))
      throw("Error opening socket");

  if (tcp_socket.enable)
    if (!openSocket(&tcp_socket, server_address, server_port, SOCK_STREAM))
      throw("Error opening socket");

  free(tmp);
}

/* **************************************************** */

SyslogCollectorInterface::~SyslogCollectorInterface() {
  if (udp_socket.enable) closeSocket(&udp_socket, SOCK_DGRAM);

  if (tcp_socket.enable) closeSocket(&tcp_socket, SOCK_STREAM);

  free(endpoint);
}

/* **************************************************** */

/* set FDs and returns the max sock */
int SyslogCollectorInterface::initFDSetsSocket(syslog_socket *ss,
                                               fd_set *read_fds,
                                               fd_set *write_fds,
                                               fd_set *except_fds,
                                               int protocol) {
  int high_sock = ss->sock;

  FD_SET(ss->sock, read_fds);
  FD_SET(ss->sock, except_fds);

  if (protocol == SOCK_STREAM) {
    for (int i = 0; i < MAX_SYSLOG_SUBSCRIBERS; ++i) {
      if (tcp_connections[i].socket != 0) {
        FD_SET(tcp_connections[i].socket, read_fds);
        FD_SET(tcp_connections[i].socket, except_fds);
        if (tcp_connections[i].socket > high_sock)
          high_sock = tcp_connections[i].socket;
      }
    }
  }

  return high_sock;
}

/* **************************************************** */

/* set FDs and returns the max sock */
int SyslogCollectorInterface::initFDSets(fd_set *read_fds, fd_set *write_fds,
                                         fd_set *except_fds) {
  int high_sock = 0;

  FD_ZERO(read_fds);
  FD_ZERO(write_fds);
  FD_ZERO(except_fds);

  if (udp_socket.enable) {
    high_sock = initFDSetsSocket(&udp_socket, read_fds, write_fds, except_fds,
                                 SOCK_DGRAM);
  }

  if (tcp_socket.enable) {
    int sock = initFDSetsSocket(&tcp_socket, read_fds, write_fds, except_fds,
                                SOCK_STREAM);
    if (sock > high_sock) high_sock = sock;
  }

  return high_sock;
}

/* **************************************************** */

int SyslogCollectorInterface::handleNewConnection() {
  char client_ipv4_str[INET_ADDRSTRLEN];
  struct sockaddr_in client_addr;
  socklen_t client_len = sizeof(client_addr);
  int new_client_sock;
  int i;

  memset(&client_addr, 0, sizeof(client_addr));

  new_client_sock =
      accept(tcp_socket.sock, (struct sockaddr *)&client_addr, &client_len);

  if (new_client_sock < 0) {
    ntop->getTrace()->traceEvent(TRACE_ERROR, "accept() failure");
    return -1;
  }

  inet_ntop(AF_INET, &client_addr.sin_addr, client_ipv4_str, INET_ADDRSTRLEN);

  ntop->getTrace()->traceEvent(TRACE_NORMAL, "Incoming connection from %s:%d",
                               client_ipv4_str, client_addr.sin_port);

  for (i = 0; i < MAX_SYSLOG_SUBSCRIBERS; ++i) {
    if (tcp_connections[i].socket == 0) {
      tcp_connections[i].socket = new_client_sock;
      tcp_connections[i].address = client_addr;
      snprintf(tcp_connections[i].ip_str, sizeof(tcp_connections[i].ip_str),
               "%s", client_ipv4_str);
      return 0;
    }
  }

  ntop->getTrace()->traceEvent(
      TRACE_NORMAL, "Too many connections. Closing connection from %s:%d",
      client_ipv4_str, client_addr.sin_port);

  close(new_client_sock);

  return -1;
}

/* **************************************************** */

void SyslogCollectorInterface::closeConnection(syslog_client *client) {
  ntop->getTrace()->traceEvent(TRACE_NORMAL,
                               "Closing client socket for %s:%d\n",
                               client->ip_str, client->address.sin_port);

  close(client->socket);
  client->socket = 0;
}

/* **************************************************** */

#ifdef USE_RECVLINE
int SyslogCollectorInterface::recvLine(int socket, char *buffer, size_t n) {
  ssize_t num_read;
  size_t tot_read = 0;
  char c;

  while (tot_read < n - 1) {
    num_read = read(socket, &c, 1);

    if (num_read == -1) {
      if (errno == EINTR)
        continue;
      else
        return -1;
    } else if (num_read == 0) { /* EOF */
      if (tot_read == 0)
        return 0;
      else
        break;
    } else {
      if (tot_read < n - 1) {
        tot_read++;
        *buffer++ = c;
      }
      if (c == '\n') break;
    }
  }

  *buffer = '\0';

  return tot_read;
}
#endif

/* **************************************************** */

int SyslogCollectorInterface::receive(int socket, char *client_ip,
                                      bool use_recvfrom) {
  char buffer[8192];
  int len, received_total = 0;
  int buffer_size = sizeof(buffer) - 1;
  char *line, *pos;
  struct sockaddr_in client_addr;
  socklen_t client_addr_len = sizeof(client_addr);
  char ip_str[INET_ADDRSTRLEN];

  // Note: do not loop to handle other clients in round robin
  // and also handle purgeIdle
  // do {

  if (use_recvfrom)
    len = recvfrom(socket,
#ifndef WIN32
                   (void *)
#endif
                       buffer,
                   buffer_size,
#ifndef WIN32
                   MSG_DONTWAIT
#else
                   0
#endif
                   ,
                   (struct sockaddr *)&client_addr, &client_addr_len);
  else
#ifdef USE_RECVLINE
    /* Read single line to avoid splitting lines across chunks */
    len = recvLine(socket, (char *)buffer, buffer_size);
#else
    len = recv(socket, (char *)buffer, buffer_size,
#ifndef WIN32
               MSG_DONTWAIT
#else
               0
#endif
    );
#endif

  if (len < 0) {
    if (errno == EAGAIN || errno == EWOULDBLOCK) {
      ntop->getTrace()->traceEvent(TRACE_INFO, "Client is not ready (%d)",
                                   errno);
      goto done;
    } else {
      ntop->getTrace()->traceEvent(TRACE_ERROR, "Client error");
      return -1;
    }

  } else if (len == 0) {
    ntop->getTrace()->traceEvent(TRACE_NORMAL, "Client shutdown");
    return -1;

  } else if (len > 0) {
    if (client_ip == NULL && use_recvfrom) {
      inet_ntop(AF_INET, &client_addr.sin_addr, ip_str, sizeof(ip_str));
      client_ip = ip_str;
    }
    received_total += len;
    buffer[len] = '\0';
    line = strtok_r(buffer, "\n", &pos);
    while (line) {
      recvStats.num_flows += parseLog(line, client_ip);
      line = strtok_r(NULL, "\n", &pos);
    }
  }

  //} while (len > 0);

done:

  ntop->getTrace()->traceEvent(TRACE_INFO, "Total received bytes: %u",
                               received_total);

  return 0;
}

/* **************************************************** */

int SyslogCollectorInterface::receiveFromClient(syslog_client *client) {
  ntop->getTrace()->traceEvent(TRACE_INFO, "Trying to receive from %s:%d",
                               client->ip_str, client->address.sin_port);

  return receive(client->socket, client->ip_str, false);
}

/* **************************************************** */

void SyslogCollectorInterface::collect_events() {
  u_int32_t max_num_polls_before_purge = MAX_SYSLOG_POLLS_BEFORE_PURGE;
  fd_set read_fds, write_fds, except_fds;
  struct timeval timeout;
  time_t now, next_purge_idle = time(NULL) + FLOW_PURGE_FREQUENCY;
  ;
  int high_sock;
  int i, rc;

  ntop->getTrace()->traceEvent(TRACE_NORMAL, "Collecting events on %s", ifname);

  while (isRunning()) {
    while (idle()) {
      purgeIdle(time(NULL));
      sleep(1);
      if (ntop->getGlobals()->isShutdown()) return;
    }

    high_sock = initFDSets(&read_fds, &write_fds, &except_fds);

    timeout.tv_sec = MAX_SYSLOG_POLL_WAIT_MS / 1000;
    timeout.tv_usec = (MAX_SYSLOG_POLL_WAIT_MS % 1000) * 1000;

    rc = select(high_sock + 1, &read_fds, &write_fds, &except_fds, &timeout);

    now = time(NULL);
    max_num_polls_before_purge--;
    if (rc == 0 || now >= next_purge_idle || max_num_polls_before_purge == 0) {
      purgeIdle(now);
      next_purge_idle = now + FLOW_PURGE_FREQUENCY;
      max_num_polls_before_purge = MAX_SYSLOG_POLLS_BEFORE_PURGE;
    }

    if (rc > 0) {
      if (udp_socket.enable) {
        if (FD_ISSET(udp_socket.sock, &read_fds)) {
          if (receive(udp_socket.sock, NULL, true) != 0)
            ntop->getTrace()->traceEvent(TRACE_ERROR,
                                         "Error receiving from UDP socket fd");
        }

        if (FD_ISSET(udp_socket.sock, &except_fds))
          ntop->getTrace()->traceEvent(TRACE_ERROR,
                                       "Exception on listen UDP socket fd");
      }

      if (tcp_socket.enable) {
        if (FD_ISSET(tcp_socket.sock, &read_fds)) {
          handleNewConnection();
        }

        if (FD_ISSET(tcp_socket.sock, &except_fds))
          ntop->getTrace()->traceEvent(TRACE_ERROR,
                                       "Exception on listen TCP socket fd");

        for (i = 0; i < MAX_SYSLOG_SUBSCRIBERS; ++i) {
          if (tcp_connections[i].socket != 0 &&
              FD_ISSET(tcp_connections[i].socket, &read_fds)) {
            if (receiveFromClient(&tcp_connections[i]) != 0) {
              closeConnection(&tcp_connections[i]);
              continue;
            }
          }

          if (tcp_connections[i].socket != 0 &&
              FD_ISSET(tcp_connections[i].socket, &except_fds)) {
            ntop->getTrace()->traceEvent(TRACE_ERROR,
                                         "Exception on TCP client fd");
            closeConnection(&tcp_connections[i]);
          }
        }
      }
    }
  }

  ntop->getTrace()->traceEvent(TRACE_NORMAL, "Flow collection is over.");
}

/* **************************************************** */

static void *messagePollLoop(void *ptr) {
  SyslogCollectorInterface *iface = (SyslogCollectorInterface *)ptr;

  /* Wait until the initialization completes */
  while (iface->isStartingUp()) sleep(1);

  iface->collect_events();

  return (NULL);
}

/* **************************************************** */

void SyslogCollectorInterface::startPacketPolling() {
  pthread_create(&pollLoop, NULL, messagePollLoop, (void *)this);
  pollLoopCreated = true;

  SyslogParserInterface::startPacketPolling();
}

/* **************************************************** */

void SyslogCollectorInterface::shutdown() {
  void *res;

  if (running) {
    NetworkInterface::shutdown();
    pthread_join(pollLoop, &res);
  }
}

/* **************************************************** */

bool SyslogCollectorInterface::set_packet_filter(char *filter) {
  ntop->getTrace()->traceEvent(
      TRACE_ERROR, "No filter can be set on a collector interface. Ignored %s",
      filter);
  return (false);
}

/* **************************************************** */

void SyslogCollectorInterface::lua(lua_State *vm) {
  SyslogParserInterface::lua(vm);

  lua_push_bool_table_entry(vm, "isSyslog", true);

  lua_newtable(vm);
  lua_push_uint64_table_entry(vm, "flows", recvStats.num_flows);
  lua_pushstring(vm, "syslogRecvStats");
  lua_insert(vm, -2);
  lua_settable(vm, -3);
}

#endif