X-Git-Url: https://git.stg.codes/stg.git/blobdiff_plain/a9c940996bac5c61168f06ef61bc20225fb7a65e..0c097ef3435d2a45c1ee4ac80f8bd3f254fb8df5:/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 9ed50756..b4555f7c 100644 --- a/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp +++ b/projects/stargazer/plugins/capture/cap_nf/cap_nf.cpp @@ -27,58 +27,94 @@ $Revision: 1.11 $ $Date: 2010/09/10 06:41:06 $ $Author: faust $ */ + +#include "cap_nf.h" + +#include "stg/common.h" +#include "stg/raw_ip_packet.h" +#include "stg/traffcounter.h" + +#include + +#include +#include +#include + #include #include #include #include -#include -#include -#include -#include +using STG::NF_CAP; -#include +namespace +{ -#include "stg/common.h" -#include "stg/raw_ip_packet.h" -#include "stg/traffcounter.h" -#include "stg/plugin_creator.h" -#include "cap_nf.h" +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_CREATOR cnc; +} -PLUGIN * GetPlugin() +extern "C" STG::Plugin* GetPlugin() { -return cnc.GetPlugin(); + static NF_CAP plugin; + return &plugin; } 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), - errorStr() -{ -} - -NF_CAP::~NF_CAP() + logger(PluginLogger::get("cap_nf")) { } int NF_CAP::ParseSettings() { -std::vector::iterator it; +std::vector::iterator 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)) { @@ -88,7 +124,7 @@ for (it = settings.moduleParams.begin(); it != settings.moduleParams.end(); ++it } continue; } - if (it->param == "UDPPort") + if (it->param == "UDPPort" && !it->value.empty()) { if (str2x(it->value[0], portU)) { @@ -111,15 +147,7 @@ if (portU > 0) { return -1; } - runningUDP = true; - if (pthread_create(&tidUDP, NULL, RunUDP, this)) - { - runningUDP = false; - CloseUDP(); - errorStr = "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) { @@ -127,42 +155,32 @@ if (portT > 0) { return -1; } - runningTCP = true; - if (pthread_create(&tidTCP, NULL, RunTCP, this)) - { - runningTCP = false; - CloseTCP(); - 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() { -runningTCP = runningUDP = false; +m_threadTCP.request_stop(); +m_threadUDP.request_stop(); if (portU && !stoppedUDP) { CloseUDP(); for (int i = 0; i < 25 && !stoppedUDP; ++i) { - usleep(200000); + struct timespec ts = {0, 200000000}; + nanosleep(&ts, NULL); } if (stoppedUDP) { - pthread_join(tidUDP, NULL); + m_threadUDP.join(); } else { - if (pthread_kill(tidUDP, SIGUSR1)) - { - errorStr = "Error sending signal 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."); } } if (portT && !stoppedTCP) @@ -170,21 +188,18 @@ if (portT && !stoppedTCP) CloseTCP(); for (int i = 0; i < 25 && !stoppedTCP; ++i) { - usleep(200000); + struct timespec ts = {0, 200000000}; + nanosleep(&ts, NULL); } if (stoppedTCP) { - pthread_join(tidTCP, NULL); + m_threadTCP.join(); } else { - if (pthread_kill(tidTCP, SIGUSR1)) - { - errorStr = "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."); } } return 0; @@ -197,15 +212,17 @@ sockUDP = socket(PF_INET, SOCK_DGRAM, 0); 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; } 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(&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; } @@ -219,48 +236,58 @@ sockTCP = socket(PF_INET, SOCK_STREAM, 0); 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; } 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(&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"; + logger("Cannot listen on TCP socket: %s", strerror(errno)); printfd(__FILE__, "Error: Error listening TCP socket\n"); return true; } return false; } -void * NF_CAP::RunUDP(void * c) +void NF_CAP::RunUDP(std::stop_token token) noexcept { -NF_CAP * cap = static_cast(c); -uint8_t buf[BUF_SIZE]; -int res; -struct sockaddr_in sin; -socklen_t slen; -cap->stoppedUDP = false; -while (cap->runningUDP) +sigset_t signalSet; +sigfillset(&signalSet); +pthread_sigmask(SIG_BLOCK, &signalSet, NULL); + +stoppedUDP = false; +while (!token.stop_requested()) { - if (!WaitPackets(cap->sockUDP)) + if (!WaitPackets(sockUDP)) { continue; } // Data - slen = sizeof(sin); - res = recvfrom(cap->sockUDP, buf, BUF_SIZE, 0, reinterpret_cast(&sin), &slen); - if (!cap->runningUDP) + struct sockaddr_in sin; + socklen_t slen = sizeof(sin); + uint8_t buf[BUF_SIZE]; + ssize_t res = recvfrom(sockUDP, buf, BUF_SIZE, 0, reinterpret_cast(&sin), &slen); + if (token.stop_requested()) break; + if (res < 0) + { + logger("recvfrom error: %s", strerror(errno)); + continue; + } + if (res == 0) // EOF { continue; @@ -270,47 +297,42 @@ while (cap->runningUDP) { if (errno != EINTR) { - cap->errorStr = "Invalid data received"; + errorStr = "Invalid data received"; 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 { -NF_CAP * cap = static_cast(c); -uint8_t buf[BUF_SIZE]; -int res; -int sd; -struct sockaddr_in sin; -socklen_t slen; -cap->stoppedTCP = false; -while (cap->runningTCP) +sigset_t signalSet; +sigfillset(&signalSet); +pthread_sigmask(SIG_BLOCK, &signalSet, NULL); + +stoppedTCP = false; +while (!token.stop_requested()) { - if (!WaitPackets(cap->sockTCP)) + if (!WaitPackets(sockTCP)) { continue; } // Data - slen = sizeof(sin); - sd = accept(cap->sockTCP, reinterpret_cast(&sin), &slen); - if (!cap->runningTCP) + struct sockaddr_in sin; + socklen_t slen = sizeof(sin); + int sd = accept(sockTCP, reinterpret_cast(&sin), &slen); + if (token.stop_requested()) break; if (sd <= 0) { - if (errno != EINTR) - { - cap->errorStr = "Error accepting connection"; - printfd(__FILE__, "Error: Error accepting connection\n"); - } + if (sd < 0) + logger("accept error: %s", strerror(errno)); continue; } @@ -320,10 +342,15 @@ while (cap->runningTCP) 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) + logger("recv error: %s", strerror(errno)); + close(sd); - if (!cap->runningTCP) + if (token.stop_requested()) break; if (res == 0) // EOF @@ -335,23 +362,17 @@ while (cap->runningTCP) // 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; } - cap->ParseBuffer(buf, res); + ParseBuffer(buf, res); } -cap->stoppedTCP = true; -return NULL; +stoppedTCP = true; } -void NF_CAP::ParseBuffer(uint8_t * buf, int size) +void NF_CAP::ParseBuffer(uint8_t * buf, ssize_t size) { -RAW_PACKET ip; +RawPacket ip; NF_HEADER * hdr = reinterpret_cast(buf); if (htons(hdr->version) != 5) { @@ -384,6 +405,6 @@ for (int i = 0; i < packets; ++i) ip.rawPacket.header.sPort = data->srcPort; ip.rawPacket.header.dPort = data->dstPort; - traffCnt->Process(ip); + traffCnt->process(ip); } }