Merge remote-tracking branch 'github/master'

[stg.git] / projects / stargazer / plugins / capture / cap_nf / cap_nf.cpp

diff --git a/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp b/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp
index 1b84d4604adbc29e9a59bf13ba878688c9ebb1af..817ed7157d5cc9567e91c587da6c53436ebf2f92 100644 (file)
--- a/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp
+++ b/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp
@@ -27,49 +27,74 @@ $Revision: 1.11 $
 $Date: 2010/09/10 06:41:06 $
 $Author: faust $
 */
 $Date: 2010/09/10 06:41:06 $
 $Author: faust $
 */

-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-#include <unistd.h>
+
+#include "cap_nf.h"
+
+#include "stg/common.h"
+#include "stg/raw_ip_packet.h"
+#include "stg/traffcounter.h"
+
+#include <vector>

 
 #include <csignal>
 #include <cerrno>
 #include <cstring>
 
 
 #include <csignal>
 #include <cerrno>
 #include <cstring>
 

-#include <vector>
-
-#include "common.h" 
-#include "cap_nf.h"
-#include "raw_ip_packet.h"
-#include "traffcounter.h"
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>

 
 

-class CAP_NF_CREATOR {
-public:
-    CAP_NF_CREATOR()
-        : nf(new NF_CAP())
-        {
-        }
+namespace
+{

 
 

-    ~CAP_NF_CREATOR()
-        {
-        delete nf;
-        }
+struct NF_HEADER {
+    uint16_t version;   // Protocol version
+    uint16_t count;     // Flows count
+    uint32_t uptime;    // System uptime
+    uint32_t timestamp; // UNIX timestamp
+    uint32_t nsecs;     // Residual nanoseconds
+    uint32_t flowSeq;   // Sequence counter
+    uint8_t  eType;     // Engine type
+    uint8_t  eID;       // Engine ID
+    uint16_t sInterval; // Sampling mode and interval
+};
+
+struct NF_DATA {
+    uint32_t srcAddr;   // Flow source address
+    uint32_t dstAddr;   // Flow destination address
+    uint32_t nextHop;   // IP addres on next hop router
+    uint16_t inSNMP;    // SNMP index of input iface
+    uint16_t outSNMP;   // SNMP index of output iface
+    uint32_t packets;   // Packets in flow
+    uint32_t octets;    // Total number of bytes in flow
+    uint32_t timeStart; // Uptime on first packet in flow
+    uint32_t timeFinish;// Uptime on last packet in flow
+    uint16_t srcPort;   // Flow source port
+    uint16_t dstPort;   // Flow destination port
+    uint8_t  pad1;      // 1-byte padding
+    uint8_t  TCPFlags;  // Cumulative OR of TCP flags
+    uint8_t  proto;     // IP protocol type (tcp, udp, etc.)
+    uint8_t  tos;       // IP Type of Service (ToS)
+    uint16_t srcAS;     // Source BGP autonomous system number
+    uint16_t dstAS;     // Destination BGP autonomus system number
+    uint8_t  srcMask;   // Source address mask in "slash" notation
+    uint8_t  dstMask;   // Destination address mask in "slash" notation
+    uint16_t pad2;      // 2-byte padding
+};
+
+#define BUF_SIZE (sizeof(NF_HEADER) + 30 * sizeof(NF_DATA))

 
 

-    NF_CAP * GetCapturer() { return nf; }
-private:
-    NF_CAP * nf;
-} cnc;
+}

 
 

-PLUGIN * GetPlugin()
+extern "C" STG::Plugin* GetPlugin()

 {
 {

-return cnc.GetCapturer();
+    static NF_CAP plugin;
+    return &plugin;

 }
 
 NF_CAP::NF_CAP()
     : traffCnt(NULL),
 }
 
 NF_CAP::NF_CAP()
     : traffCnt(NULL),

-      tidTCP(0),
-      tidUDP(0),

       runningTCP(false),
       runningUDP(false),
       stoppedTCP(true),
       runningTCP(false),
       runningUDP(false),
       stoppedTCP(true),


@@ -77,20 +102,17 @@ NF_CAP::NF_CAP()
       portT(0),
       portU(0),
       sockTCP(-1),
       portT(0),
       portU(0),
       sockTCP(-1),

-      sockUDP(-1)
-{
-}
-
-NF_CAP::~NF_CAP()
+      sockUDP(-1),
+      logger(STG::PluginLogger::get("cap_nf"))

 {
 }
 
 int NF_CAP::ParseSettings()
 {
 {
 }
 
 int NF_CAP::ParseSettings()
 {

-std::vector<PARAM_VALUE>::iterator it;
+std::vector<STG::ParamValue>::iterator it;

 for (it = settings.moduleParams.begin(); it != settings.moduleParams.end(); ++it)
     {
 for (it = settings.moduleParams.begin(); it != settings.moduleParams.end(); ++it)
     {

-    if (it->param == "TCPPort")
+    if (it->param == "TCPPort" && !it->value.empty())

         {
         if (str2x(it->value[0], portT))
             {
         {
         if (str2x(it->value[0], portT))
             {


@@ -100,7 +122,7 @@ for (it = settings.moduleParams.begin(); it != settings.moduleParams.end(); ++it
             }
         continue;
         }
             }
         continue;
         }

-    if (it->param == "UDPPort")
+    if (it->param == "UDPPort" && !it->value.empty())

         {
         if (str2x(it->value[0], portU))
             {
         {
         if (str2x(it->value[0], portU))
             {


@@ -129,6 +151,7 @@ if (portU > 0)
         runningUDP = false;
         CloseUDP();
         errorStr = "Cannot create UDP thread";
         runningUDP = false;
         CloseUDP();
         errorStr = "Cannot create UDP thread";

+        logger("Cannot create UDP thread.");

         printfd(__FILE__, "Error: Cannot create UDP thread\n");
         return -1;
         }
         printfd(__FILE__, "Error: Cannot create UDP thread\n");
         return -1;
         }


@@ -144,6 +167,7 @@ if (portT > 0)
         {
         runningTCP = false;
         CloseTCP();
         {
         runningTCP = false;
         CloseTCP();

+        logger("Cannot create TCP thread.");

         errorStr = "Cannot create TCP thread";
         printfd(__FILE__, "Error: Cannot create TCP thread\n");
         return -1;
         errorStr = "Cannot create TCP thread";
         printfd(__FILE__, "Error: Cannot create TCP thread\n");
         return -1;


@@ -160,7 +184,8 @@ if (portU && !stoppedUDP)
     CloseUDP();
     for (int i = 0; i < 25 && !stoppedUDP; ++i)
         {
     CloseUDP();
     for (int i = 0; i < 25 && !stoppedUDP; ++i)
         {

-        usleep(200000);
+        struct timespec ts = {0, 200000000};
+        nanosleep(&ts, NULL);

         }
     if (stoppedUDP)
         {
         }
     if (stoppedUDP)
         {


@@ -171,10 +196,12 @@ if (portU && !stoppedUDP)
         if (pthread_kill(tidUDP, SIGUSR1))
             {
             errorStr = "Error sending signal to UDP thread";
         if (pthread_kill(tidUDP, SIGUSR1))
             {
             errorStr = "Error sending signal to UDP thread";

+            logger("Error sending sugnal to UDP thread.");

             printfd(__FILE__, "Error: Error sending signal to UDP thread\n");
             return -1;
             }
         printfd(__FILE__, "UDP thread NOT stopped\n");
             printfd(__FILE__, "Error: Error sending signal to UDP thread\n");
             return -1;
             }
         printfd(__FILE__, "UDP thread NOT stopped\n");

+        logger("Cannot stop UDP thread.");

         }
     }
 if (portT && !stoppedTCP)
         }
     }
 if (portT && !stoppedTCP)


@@ -182,7 +209,8 @@ if (portT && !stoppedTCP)
     CloseTCP();
     for (int i = 0; i < 25 && !stoppedTCP; ++i)
         {
     CloseTCP();
     for (int i = 0; i < 25 && !stoppedTCP; ++i)
         {

-        usleep(200000);
+        struct timespec ts = {0, 200000000};
+        nanosleep(&ts, NULL);

         }
     if (stoppedTCP)
         {
         }
     if (stoppedTCP)
         {


@@ -193,10 +221,12 @@ if (portT && !stoppedTCP)
         if (pthread_kill(tidTCP, SIGUSR1))
             {
             errorStr = "Error sending signal to TCP thread";
         if (pthread_kill(tidTCP, SIGUSR1))
             {
             errorStr = "Error sending signal to TCP thread";

+            logger("Error sending signal to TCP thread.");

             printfd(__FILE__, "Error: Error sending signal to TCP thread\n");
             return -1;
             }
         printfd(__FILE__, "TCP thread NOT stopped\n");
             printfd(__FILE__, "Error: Error sending signal to TCP thread\n");
             return -1;
             }
         printfd(__FILE__, "TCP thread NOT stopped\n");

+        logger("Cannot stop TCP thread.");

         }
     }
 return 0;
         }
     }
 return 0;


@@ -209,6 +239,7 @@ sockUDP = socket(PF_INET, SOCK_DGRAM, 0);
 if (sockUDP <= 0)
     {
     errorStr = "Error opening UDP socket";
 if (sockUDP <= 0)
     {
     errorStr = "Error opening UDP socket";

+    logger("Cannot create UDP socket: %s", strerror(errno));

     printfd(__FILE__, "Error: Error opening UDP socket\n");
     return true;
     }
     printfd(__FILE__, "Error: Error opening UDP socket\n");
     return true;
     }


@@ -218,6 +249,7 @@ sin.sin_addr.s_addr = inet_addr("0.0.0.0");
 if (bind(sockUDP, (struct sockaddr *)&sin, sizeof(sin)))
     {
     errorStr = "Error binding UDP socket";
 if (bind(sockUDP, (struct sockaddr *)&sin, sizeof(sin)))
     {
     errorStr = "Error binding UDP socket";

+    logger("Cannot bind UDP socket: %s", strerror(errno));

     printfd(__FILE__, "Error: Error binding UDP socket\n");
     return true;
     }
     printfd(__FILE__, "Error: Error binding UDP socket\n");
     return true;
     }


@@ -231,6 +263,7 @@ sockTCP = socket(PF_INET, SOCK_STREAM, 0);
 if (sockTCP <= 0)
     {
     errorStr = "Error opening TCP socket";
 if (sockTCP <= 0)
     {
     errorStr = "Error opening TCP socket";

+    logger("Cannot create TCP socket: %s", strerror(errno));

     printfd(__FILE__, "Error: Error opening TCP socket\n");
     return true;
     }
     printfd(__FILE__, "Error: Error opening TCP socket\n");
     return true;
     }


@@ -240,12 +273,14 @@ sin.sin_addr.s_addr = inet_addr("0.0.0.0");
 if (bind(sockTCP, (struct sockaddr *)&sin, sizeof(sin)))
     {
     errorStr = "Error binding TCP socket";
 if (bind(sockTCP, (struct sockaddr *)&sin, sizeof(sin)))
     {
     errorStr = "Error binding TCP socket";

+    logger("Cannot bind TCP socket: %s", strerror(errno));

     printfd(__FILE__, "Error: Error binding TCP socket\n");
     return true;
     }
 if (listen(sockTCP, 1))
     {
     errorStr = "Error listening on TCP socket";
     printfd(__FILE__, "Error: Error binding TCP socket\n");
     return true;
     }
 if (listen(sockTCP, 1))
     {
     errorStr = "Error listening on TCP socket";

+    logger("Cannot listen on TCP socket: %s", strerror(errno));

     printfd(__FILE__, "Error: Error listening TCP socket\n");
     return true;
     }
     printfd(__FILE__, "Error: Error listening TCP socket\n");
     return true;
     }


@@ -254,25 +289,33 @@ return false;
 
 void * NF_CAP::RunUDP(void * c)
 {
 
 void * NF_CAP::RunUDP(void * c)
 {

+sigset_t signalSet;
+sigfillset(&signalSet);
+pthread_sigmask(SIG_BLOCK, &signalSet, NULL);
+

 NF_CAP * cap = static_cast<NF_CAP *>(c);
 NF_CAP * cap = static_cast<NF_CAP *>(c);

-uint8_t buf[BUF_SIZE];
-int res;
-struct sockaddr_in sin;
-socklen_t slen;

 cap->stoppedUDP = false;
 while (cap->runningUDP)
     {
 cap->stoppedUDP = false;
 while (cap->runningUDP)
     {

-    if (!cap->WaitPackets(cap->sockUDP))
+    if (!WaitPackets(cap->sockUDP))

         {
         continue;
         }
 
     // Data
         {
         continue;
         }
 
     // Data

-    slen = sizeof(sin);
-    res = recvfrom(cap->sockUDP, buf, BUF_SIZE, 0, reinterpret_cast<struct sockaddr *>(&sin), &slen);
+    struct sockaddr_in sin;
+    socklen_t slen = sizeof(sin);
+    uint8_t buf[BUF_SIZE];
+    ssize_t res = recvfrom(cap->sockUDP, buf, BUF_SIZE, 0, reinterpret_cast<struct sockaddr *>(&sin), &slen);

     if (!cap->runningUDP)
         break;
 
     if (!cap->runningUDP)
         break;
 

+    if (res < 0)
+        {
+        cap->logger("recvfrom error: %s", strerror(errno));
+        continue;
+        }
+

     if (res == 0) // EOF
         {
         continue;
     if (res == 0) // EOF
         {
         continue;


@@ -296,43 +339,45 @@ return NULL;
 
 void * NF_CAP::RunTCP(void * c)
 {
 
 void * NF_CAP::RunTCP(void * c)
 {

+sigset_t signalSet;
+sigfillset(&signalSet);
+pthread_sigmask(SIG_BLOCK, &signalSet, NULL);
+

 NF_CAP * cap = static_cast<NF_CAP *>(c);
 NF_CAP * cap = static_cast<NF_CAP *>(c);

-uint8_t buf[BUF_SIZE];
-int res;
-int sd;
-struct sockaddr_in sin;
-socklen_t slen;

 cap->stoppedTCP = false;
 while (cap->runningTCP)
     {
 cap->stoppedTCP = false;
 while (cap->runningTCP)
     {

-    if (!cap->WaitPackets(cap->sockTCP))
+    if (!WaitPackets(cap->sockTCP))

         {
         continue;
         }
 
     // Data
         {
         continue;
         }
 
     // Data

-    slen = sizeof(sin);
-    sd = accept(cap->sockTCP, reinterpret_cast<struct sockaddr *>(&sin), &slen);
+    struct sockaddr_in sin;
+    socklen_t slen = sizeof(sin);
+    int sd = accept(cap->sockTCP, reinterpret_cast<struct sockaddr *>(&sin), &slen);

     if (!cap->runningTCP)
         break;
 
     if (sd <= 0)
         {
     if (!cap->runningTCP)
         break;
 
     if (sd <= 0)
         {

-        if (errno != EINTR)
-            {
-            cap->errorStr = "Error accepting connection";
-            printfd(__FILE__, "Error: Error accepting connection\n");
-            }
+        if (sd < 0)
+            cap->logger("accept error: %s", strerror(errno));

         continue;
         }
 
         continue;
         }
 

-    if (!cap->WaitPackets(sd))
+    if (!WaitPackets(sd))

         {
         close(sd);
         continue;
         }
 
         {
         close(sd);
         continue;
         }
 

-    res = recv(sd, buf, BUF_SIZE, MSG_WAITALL);
+    uint8_t buf[BUF_SIZE];
+    ssize_t res = recv(sd, buf, BUF_SIZE, MSG_WAITALL);
+
+    if (res < 0)
+        cap->logger("recv error: %s", strerror(errno));
+

     close(sd);
 
     if (!cap->runningTCP)
     close(sd);
 
     if (!cap->runningTCP)


@@ -347,11 +392,6 @@ while (cap->runningTCP)
     // Need to check actual data length and wait all data to receive
     if (res < 24)
         {
     // Need to check actual data length and wait all data to receive
     if (res < 24)
         {

-        if (errno != EINTR)
-            {
-            cap->errorStr = "Invalid data received";
-            printfd(__FILE__, "Error: Invalid data received through TCP\n");
-            }

         continue;
         }
 
         continue;
         }
 


@@ -361,9 +401,9 @@ cap->stoppedTCP = true;
 return NULL;
 }
 
 return NULL;
 }
 

-void NF_CAP::ParseBuffer(uint8_t * buf, int size)
+void NF_CAP::ParseBuffer(uint8_t * buf, ssize_t size)

 {
 {

-RAW_PACKET ip;
+STG::RawPacket ip;

 NF_HEADER * hdr = reinterpret_cast<NF_HEADER *>(buf);
 if (htons(hdr->version) != 5)
     {
 NF_HEADER * hdr = reinterpret_cast<NF_HEADER *>(buf);
 if (htons(hdr->version) != 5)
     {


@@ -387,43 +427,15 @@ for (int i = 0; i < packets; ++i)
     {
     NF_DATA * data = reinterpret_cast<NF_DATA *>(buf + 24 + i * 48);
 
     {
     NF_DATA * data = reinterpret_cast<NF_DATA *>(buf + 24 + i * 48);
 

-    ip.header.ipHeader.ip_v = 4;
-    ip.header.ipHeader.ip_hl = 5;
-    ip.header.ipHeader.ip_p = data->proto;
+    ip.rawPacket.header.ipHeader.ip_v = 4;
+    ip.rawPacket.header.ipHeader.ip_hl = 5;
+    ip.rawPacket.header.ipHeader.ip_p = data->proto;

     ip.dataLen = ntohl(data->octets);
     ip.dataLen = ntohl(data->octets);

-    ip.header.ipHeader.ip_src.s_addr = data->srcAddr;
-    ip.header.ipHeader.ip_dst.s_addr = data->dstAddr;
-    ip.header.sPort = data->srcPort;
-    ip.header.dPort = data->dstPort;
+    ip.rawPacket.header.ipHeader.ip_src.s_addr = data->srcAddr;
+    ip.rawPacket.header.ipHeader.ip_dst.s_addr = data->dstAddr;
+    ip.rawPacket.header.sPort = data->srcPort;
+    ip.rawPacket.header.dPort = data->dstPort;

 
 

-    traffCnt->Process(ip);
+    traffCnt->process(ip);

     }
 }
     }
 }

-
-bool NF_CAP::WaitPackets(int sd) const
-{
-fd_set rfds;
-FD_ZERO(&rfds);
-FD_SET(sd, &rfds);
-
-struct timeval tv;
-tv.tv_sec = 0;
-tv.tv_usec = 500000;
-
-int res = select(sd + 1, &rfds, NULL, NULL, &tv);
-if (res == -1) // Error
-    {
-    if (errno != EINTR)
-        {
-        printfd(__FILE__, "Error on select: '%s'\n", strerror(errno));
-        }
-    return false;
-    }
-
-if (res == 0) // Timeout
-    {
-    return false;
-    }
-
-return true;
-}