TCP - Add ipfix decoder

Author: BonnyAD9

src/plugins/input/tcp/CMakeLists.txt

@@ -7,6 +7,7 @@ add_library(tcp-input MODULE
     src/Connection.cpp
     src/Epoll.cpp
     src/ClientManager.cpp
+    src/IpfixDecoder.cpp
 )
 
 if (CMAKE_HOST_SYSTEM_NAME STREQUAL "FreeBSD" OR CMAKE_HOST_SYSTEM_NAME STREQUAL "OpenBSD")

src/plugins/input/tcp/src/IpfixDecoder.cpp

@@ -0,0 +1,86 @@
+/**
+ * \file
+ * \author Jakub Antonín Štigler <[email protected]>
+ * \brief IPFIX decoder for tcp plugin (header file)
+ * \date 2024
+ *
+ * Copyright: (C) 2023 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "IpfixDecoder.hpp"
+
+#include <stdexcept> // runtime_error
+#include <string>    // string
+#include <errno.h>   // errno, EWOULDBLOCK, EAGAIN
+#include <stddef.h>  // size_t
+
+#include <unistd.h> // read
+
+#include <ipfixcol2.h> // fds_ipfix_msg_hdr, ipx_strerror
+
+#include "DecodeBuffer.hpp" // DecodeBuffer
+#include "read_until_n.hpp" // read_until_n
+
+namespace tcp_in {
+
+DecodeBuffer &IpfixDecoder::decode() {
+    while (!m_decoded.enough_data()) {
+        // Read the header of the message.
+        if (!read_header()) {
+            // There is not enough data to read the whole header.
+            break;
+        }
+
+        // Read the body of the message.
+        if (!read_body()) {
+            // There is not enough data to read the whole body of the message.
+            break;
+        }
+    }
+
+    if (m_decoded.is_eof_reached() && m_part_readed.size() != 0) {
+        throw std::runtime_error("Received incomplete message.");
+    }
+
+    return m_decoded;
+}
+
+bool IpfixDecoder::read_header() {
+    if (m_msg_size != 0) {
+        // The header is already read, but the message body is incomplete.
+        return true;
+    }
+
+    // Read the header
+    if (!read_until_n(sizeof(fds_ipfix_msg_hdr))) {
+        // There is not enough data to read the whole header.
+        return false;
+    }
+
+    // The header has been read, load the size of the message.
+    auto hdr = reinterpret_cast<fds_ipfix_msg_hdr *>(m_part_readed.data());
+    m_msg_size = ntohs(hdr->length);
+    return true;
+}
+
+bool IpfixDecoder::read_body() {
+    // Read the body.
+    if (!read_until_n(m_msg_size)) {
+        // There is not enough data to read the whole body
+        return false;
+    }
+
+    // Whole message has been read. Add it to the decode buffer.
+    m_decoded.add(std::move(m_part_readed));
+    m_part_readed = ByteVector();
+    m_msg_size = 0;
+    return true;
+}
+
+bool IpfixDecoder::read_until_n(size_t n) {
+    return ::read_until_n(n, m_fd, m_part_readed, m_decoded);
+}
+
+} // namespace tcp_in
+

src/plugins/input/tcp/src/IpfixDecoder.hpp

@@ -0,0 +1,56 @@
+/**
+ * \file
+ * \author Jakub Antonín Štigler <[email protected]>
+ * \brief IPFIX decoder for tcp plugin (header file)
+ * \date 2024
+ *
+ * Copyright: (C) 2023 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#pragma once
+
+#include <cstdint> // uint16_t
+#include <cstddef> // size_t
+
+#include "Decoder.hpp"      // Decoder
+#include "DecodeBuffer.hpp" // DecodeBuffer
+#include "ByteVector.hpp"   // ByteVector
+
+namespace tcp_in {
+
+/** Identifies data for which this decoder should be used. */
+constexpr uint16_t IPFIX_MAGIC = 10;
+
+/** Decoder for basic IPFX data. */
+class IpfixDecoder : public Decoder {
+public:
+    /**
+     * @brief Creates ipfix decoder.
+     * @param fd TCP connection file descriptor.
+     */
+    IpfixDecoder(int fd) : m_fd(fd), m_decoded(), m_part_readed(), m_msg_size(0) {}
+
+    virtual DecodeBuffer &decode() override;
+
+    virtual const char *get_name() const override {
+        return "IPFIX";
+    }
+
+private:
+    /** returns true if there was enough data to read the header or if the header is already read */
+    bool read_header();
+    /** returns true if there was enough data to read the body */
+    bool read_body();
+    /** returns true if there was enough data to read to the given amount */
+    bool read_until_n(size_t n);
+
+    int m_fd;
+    DecodeBuffer m_decoded;
+
+    ByteVector m_part_readed;
+    /** Final size of the whole IPFIX message. When 0 IPIFX header is not fully read. */
+    size_t m_msg_size;
+};
+
+} // namespace tcp_in

src/plugins/input/tcp/src/read_until_n.hpp

@@ -0,0 +1,68 @@
+/**
+ * \file
+ * \author Jakub Antonín Štigler <[email protected]>
+ * \brief Function for reading from file descriptor (header file)
+ * \date 2024
+ *
+ * Copyright: (C) 2023 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#pragma once
+
+#include <cstddef>   // size_t
+#include <cerrno>    // errno, EWOULDBLOCK, EAGAIN
+#include <string>    // std::string
+#include <stdexcept> // std::runtime_error
+
+#include <sys/socket.h> // recv
+
+#include <ipfixcol2.h> // ipx_strerror
+
+/**
+ * @brief Reads to vector from file descriptor until the vector has the desired number of bytes
+ *
+ * @tparam Vec vector like object
+ * @tparam EofSig `EofSig::signal_eof()` is called when eof is reached
+ * @param n number of bytes that should be in the vector
+ * @param fd file descriptor to read from
+ * @param result where to read the bytes to
+ * @param eofSignaler has method to signal eof
+ * @return true if the vector has at least `n` bytes after the call
+ * @throws when fails to read from the file descriptor
+ */
+template<typename Vec, typename EofSig>
+bool read_until_n(size_t n, int fd, Vec &result, EofSig &eofSignaler) {
+    auto filled = result.size();
+    if (filled >= n) {
+        return true;
+    }
+
+    auto remaining = n - filled;
+    result.resize(n);
+
+    int res = recv(fd, result.data() + filled, remaining, 0);
+    if (res == -1) {
+        result.resize(filled);
+        int err = errno;
+        if (err == EWOULDBLOCK || err == EAGAIN) {
+            return false;
+        }
+
+        const char *err_str;
+        ipx_strerror(err, err_str);
+        throw std::runtime_error("Failed to read from descriptor: " + std::string(err_str));
+    }
+
+    result.resize(filled + res);
+
+    if (res == 0) {
+        eofSignaler.signal_eof();
+    }
+
+    if (static_cast<size_t>(res) != remaining) {
+        return false;
+    }
+
+    return true;
+}