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 8c4fdc2dd350cf9dbc9306d59273cf39446d2366..b4555f7c010f441ae11854af6f06b85f1ef28476 100644 (file)
--- a/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp
+++ b/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp
@@ -27,64 +27,94 @@ $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 <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>

 
 

-#include "stg/common.h"
-#include "stg/raw_ip_packet.h"
-#include "stg/traffcounter.h"
-#include "stg/plugin_creator.h"
-#include "cap_nf.h"
+using STG::NF_CAP;

 
 namespace
 {
 
 namespace
 {

-PLUGIN_CREATOR<NF_CAP> cnc;
-}

 
 

-extern "C" PLUGIN * GetPlugin();
+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))
+
+}

 
 

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

 {
 {

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

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

-      settings(),
-      tidTCP(),
-      tidUDP(),
-      runningTCP(false),
-      runningUDP(false),

       stoppedTCP(true),
       stoppedUDP(true),
       portT(0),
       portU(0),
       sockTCP(-1),
       sockUDP(-1),
       stoppedTCP(true),
       stoppedUDP(true),
       portT(0),
       portU(0),
       sockTCP(-1),
       sockUDP(-1),

-      errorStr(),
-      logger(GetPluginLogger(GetStgLogger(), "cap_nf"))
-{
-}
-
-NF_CAP::~NF_CAP()
+      logger(PluginLogger::get("cap_nf"))

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

-std::vector<PARAM_VALUE>::iterator it;
+std::vector<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))
             {


@@ -94,7 +124,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))
             {


@@ -117,16 +147,7 @@ if (portU > 0)
         {
         return -1;
         }
         {
         return -1;
         }

-    runningUDP = true;
-    if (pthread_create(&tidUDP, NULL, RunUDP, this))
-        {
-        runningUDP = false;
-        CloseUDP();
-        errorStr = "Cannot create UDP thread";
-        logger("Cannot create UDP thread.");
-        printfd(__FILE__, "Error: Cannot create UDP thread\n");
-        return -1;
-        }
+    m_threadUDP = std::jthread([this](auto token){ RunUDP(std::move(token)); });

     }
 if (portT > 0)
     {
     }
 if (portT > 0)
     {


@@ -134,23 +155,15 @@ if (portT > 0)
         {
         return -1;
         }
         {
         return -1;
         }

-    runningTCP = true;
-    if (pthread_create(&tidTCP, NULL, RunTCP, this))
-        {
-        runningTCP = false;
-        CloseTCP();
-        logger("Cannot create TCP thread.");
-        errorStr = "Cannot create TCP thread";
-        printfd(__FILE__, "Error: Cannot create TCP thread\n");
-        return -1;
-        }
+    m_threadTCP = std::jthread([this](auto token){ RunTCP(std::move(token)); });

     }
 return 0;
 }
 
 int NF_CAP::Stop()
 {
     }
 return 0;
 }
 
 int NF_CAP::Stop()
 {

-runningTCP = runningUDP = false;
+m_threadTCP.request_stop();
+m_threadUDP.request_stop();

 if (portU && !stoppedUDP)
     {
     CloseUDP();
 if (portU && !stoppedUDP)
     {
     CloseUDP();


@@ -161,17 +174,11 @@ if (portU && !stoppedUDP)
         }
     if (stoppedUDP)
         {
         }
     if (stoppedUDP)
         {

-        pthread_join(tidUDP, NULL);
+        m_threadUDP.join();

         }
     else
         {
         }
     else
         {

-        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;
-            }
+        m_threadUDP.detach();

         printfd(__FILE__, "UDP thread NOT stopped\n");
         logger("Cannot stop UDP thread.");
         }
         printfd(__FILE__, "UDP thread NOT stopped\n");
         logger("Cannot stop UDP thread.");
         }


@@ -186,17 +193,11 @@ if (portT && !stoppedTCP)
         }
     if (stoppedTCP)
         {
         }
     if (stoppedTCP)
         {

-        pthread_join(tidTCP, NULL);
+        m_threadTCP.join();

         }
     else
         {
         }
     else
         {

-        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;
-            }
+        m_threadTCP.detach();

         printfd(__FILE__, "TCP thread NOT stopped\n");
         logger("Cannot stop TCP thread.");
         }
         printfd(__FILE__, "TCP thread NOT stopped\n");
         logger("Cannot stop TCP thread.");
         }


@@ -218,7 +219,7 @@ if (sockUDP <= 0)
 sin.sin_family = AF_INET;
 sin.sin_port = htons(portU);
 sin.sin_addr.s_addr = inet_addr("0.0.0.0");
 sin.sin_family = AF_INET;
 sin.sin_port = htons(portU);
 sin.sin_addr.s_addr = inet_addr("0.0.0.0");

-if (bind(sockUDP, (struct sockaddr *)&sin, sizeof(sin)))
+if (bind(sockUDP, reinterpret_cast<const sockaddr*>(&sin), sizeof(sin)))

     {
     errorStr = "Error binding UDP socket";
     logger("Cannot bind UDP socket: %s", strerror(errno));
     {
     errorStr = "Error binding UDP socket";
     logger("Cannot bind UDP socket: %s", strerror(errno));


@@ -242,7 +243,7 @@ if (sockTCP <= 0)
 sin.sin_family = AF_INET;
 sin.sin_port = htons(portT);
 sin.sin_addr.s_addr = inet_addr("0.0.0.0");
 sin.sin_family = AF_INET;
 sin.sin_port = htons(portT);
 sin.sin_addr.s_addr = inet_addr("0.0.0.0");

-if (bind(sockTCP, (struct sockaddr *)&sin, sizeof(sin)))
+if (bind(sockTCP, reinterpret_cast<const sockaddr*>(&sin), sizeof(sin)))

     {
     errorStr = "Error binding TCP socket";
     logger("Cannot bind TCP socket: %s", strerror(errno));
     {
     errorStr = "Error binding TCP socket";
     logger("Cannot bind TCP socket: %s", strerror(errno));


@@ -259,17 +260,16 @@ if (listen(sockTCP, 1))
 return false;
 }
 
 return false;
 }
 

-void * NF_CAP::RunUDP(void * c)
+void NF_CAP::RunUDP(std::stop_token token) noexcept

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

-NF_CAP * cap = static_cast<NF_CAP *>(c);
-cap->stoppedUDP = false;
-while (cap->runningUDP)
+stoppedUDP = false;
+while (!token.stop_requested())

     {
     {

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

         {
         continue;
         }
         {
         continue;
         }


@@ -278,13 +278,13 @@ while (cap->runningUDP)
     struct sockaddr_in sin;
     socklen_t slen = sizeof(sin);
     uint8_t buf[BUF_SIZE];
     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)
+    ssize_t res = recvfrom(sockUDP, buf, BUF_SIZE, 0, reinterpret_cast<struct sockaddr *>(&sin), &slen);
+    if (token.stop_requested())

         break;
 
     if (res < 0)
         {
         break;
 
     if (res < 0)
         {

-        cap->logger("recvfrom error: %s", strerror(errno));
+        logger("recvfrom error: %s", strerror(errno));

         continue;
         }
 
         continue;
         }
 


@@ -297,29 +297,27 @@ while (cap->runningUDP)
         {
         if (errno != EINTR)
             {
         {
         if (errno != EINTR)
             {

-            cap->errorStr = "Invalid data received";
+            errorStr = "Invalid data received";

             printfd(__FILE__, "Error: Invalid data received through UDP\n");
             }
         continue;
         }
 
             printfd(__FILE__, "Error: Invalid data received through UDP\n");
             }
         continue;
         }
 

-    cap->ParseBuffer(buf, res);
+    ParseBuffer(buf, res);

     }
     }

-cap->stoppedUDP = true;
-return NULL;
+stoppedUDP = true;

 }
 
 }
 

-void * NF_CAP::RunTCP(void * c)
+void NF_CAP::RunTCP(std::stop_token token) noexcept

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

-NF_CAP * cap = static_cast<NF_CAP *>(c);
-cap->stoppedTCP = false;
-while (cap->runningTCP)
+stoppedTCP = false;
+while (!token.stop_requested())

     {
     {

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

         {
         continue;
         }
         {
         continue;
         }


@@ -327,14 +325,14 @@ while (cap->runningTCP)
     // Data
     struct sockaddr_in sin;
     socklen_t slen = sizeof(sin);
     // Data
     struct sockaddr_in sin;
     socklen_t slen = sizeof(sin);

-    int sd = accept(cap->sockTCP, reinterpret_cast<struct sockaddr *>(&sin), &slen);
-    if (!cap->runningTCP)
+    int sd = accept(sockTCP, reinterpret_cast<struct sockaddr *>(&sin), &slen);
+    if (token.stop_requested())

         break;
 
     if (sd <= 0)
         {
         if (sd < 0)
         break;
 
     if (sd <= 0)
         {
         if (sd < 0)

-            cap->logger("accept error: %s", strerror(errno));
+            logger("accept error: %s", strerror(errno));

         continue;
         }
 
         continue;
         }
 


@@ -348,11 +346,11 @@ while (cap->runningTCP)
     ssize_t res = recv(sd, buf, BUF_SIZE, MSG_WAITALL);
 
     if (res < 0)
     ssize_t res = recv(sd, buf, BUF_SIZE, MSG_WAITALL);
 
     if (res < 0)

-        cap->logger("recv error: %s", strerror(errno));
+        logger("recv error: %s", strerror(errno));

 
     close(sd);
 
 
     close(sd);
 

-    if (!cap->runningTCP)
+    if (token.stop_requested())

         break;
 
     if (res == 0) // EOF
         break;
 
     if (res == 0) // EOF


@@ -367,15 +365,14 @@ while (cap->runningTCP)
         continue;
         }
 
         continue;
         }
 

-    cap->ParseBuffer(buf, res);
+    ParseBuffer(buf, res);

     }
     }

-cap->stoppedTCP = true;
-return NULL;
+stoppedTCP = true;

 }
 
 void NF_CAP::ParseBuffer(uint8_t * buf, ssize_t size)
 {
 }
 
 void NF_CAP::ParseBuffer(uint8_t * buf, ssize_t size)
 {

-RAW_PACKET ip;
+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)
     {


@@ -408,6 +405,6 @@ for (int i = 0; i < packets; ++i)
     ip.rawPacket.header.sPort = data->srcPort;
     ip.rawPacket.header.dPort = data->dstPort;
 
     ip.rawPacket.header.sPort = data->srcPort;
     ip.rawPacket.header.dPort = data->dstPort;
 

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

     }
 }
     }
 }