Merge pull request #180 from CESNET/fragmentation-cache

Introduce fragmentation cache 
Cache merge releated fragmented packets to single flow

Author: SiskaPavel

Makefile.am

@@ -60,6 +60,13 @@ ipfixprobe_input_src+=\
 endif
 
 ipfixprobe_storage_src=\
+		storage/fragmentationCache/ringBuffer.hpp \
+		storage/fragmentationCache/timevalUtils.hpp \
+		storage/fragmentationCache/fragmentationKeyData.hpp \
+		storage/fragmentationCache/fragmentationTable.hpp \
+		storage/fragmentationCache/fragmentationTable.cpp \
+		storage/fragmentationCache/fragmentationCache.hpp \
+		storage/fragmentationCache/fragmentationCache.cpp \
 		storage/cache.cpp \
 		storage/cache.hpp \
 		storage/xxhash.c \

include/ipfixprobe/flowifc.hpp

@@ -48,6 +48,7 @@
 
 #include <arpa/inet.h>
 #include "ipaddr.hpp"
+#include <string>
 
 namespace ipxp {
 

include/ipfixprobe/packet.hpp

@@ -61,6 +61,9 @@ struct Packet : public Record {
    ipaddr_t    src_ip;
    ipaddr_t    dst_ip;
    uint32_t    vlan_id;
+   uint32_t    frag_id;
+   uint16_t    frag_off;
+   bool        more_fragments;
 
    uint16_t    src_port;
    uint16_t    dst_port;
@@ -107,6 +110,7 @@ struct Packet : public Record {
       dst_mac(), src_mac(), ethertype(0),
       ip_len(0), ip_payload_len(0), ip_version(0), ip_ttl(0),
       ip_proto(0), ip_tos(0), ip_flags(0), src_ip({0}), dst_ip({0}), vlan_id(0),
+      frag_id(0), frag_off(0), more_fragments(false),
       src_port(0), dst_port(0), tcp_flags(0), tcp_window(0),
       tcp_options(0), tcp_mss(0), tcp_seq(0), tcp_ack(0), mplsTop(0),
       packet(nullptr), packet_len(0), packet_len_wire(0),

init/ipfixprobed

@@ -54,11 +54,23 @@ if [ -e "$CONFFILE" ]; then
    if [ ! -z ${ACTIVE_TIMEOUT+x} ]; then
       CACHE_ACTIVET_PARAM=";active=${ACTIVE_TIMEOUT}"
    fi
-   CACHE_INACTIVET_PARAM=""
+   CACHE_INACTIVE_PARAM=""
    if [ ! -z ${INACTIVE_TIMEOUT+x} ]; then
-      CACHE_INACTIVET_PARAM=";inactive=${INACTIVE_TIMEOUT}"
+      CACHE_INACTIVE_PARAM=";inactive=${INACTIVE_TIMEOUT}"
    fi
-   storage="-s cache;${CACHE_SIZE_PARAM}${CACHE_ACTIVET_PARAM}${CACHE_INACTIVET_PARAM}"
+   CACHE_FRAG_ENABLE_PARAM=""
+   if [ ! -z ${FRAG_CACHE_ENABLE+x} ]; then
+      CACHE_FRAG_ENABLE_PARAM=";frag-enable=${FRAG_CACHE_ENABLE}"
+   fi
+   CACHE_FRAG_SIZE=""
+   if [ ! -z ${FRAG_CACHE_SIZE+x} ]; then
+      CACHE_FRAG_SIZE=";frag-size=${FRAG_CACHE_SIZE}"
+   fi
+   CACHE_FRAG_TIMEOUT=""
+   if [ ! -z ${FRAG_CACHE_TIMEOUT+x} ]; then
+      CACHE_FRAG_TIMEOUT=";frag-timeout=${FRAG_CACHE_TIMEOUT}"
+   fi
+   storage="-s cache;${CACHE_SIZE_PARAM}${CACHE_ACTIVET_PARAM}${CACHE_INACTIVE_PARAM}${CACHE_FRAG_ENABLE_PARAM}${CACHE_FRAG_SIZE}${CACHE_FRAG_TIMEOUT}"
    process=""
    if `declare -p PROCESS > /dev/null 2>/dev/null`; then
       # list of input plugins

init/link0.conf.example

@@ -146,6 +146,15 @@ CACHE_SIZE=17
 ACTIVE_TIMEOUT=300
 INACTIVE_TIMEOUT=65
 
+# Enable/disable fragmentation cache (true, false)
+FRAG_CACHE_ENABLE=true
+
+# size of fragmentation cache
+FRAG_CACHE_SIZE=10007
+
+# timeout in seconds for fragments in fragmentation cache
+FRAG_CACHE_TIMEOUT=3
+
 #
 #  $$$$$$\              $$\                           $$\
 # $$  __$$\             $$ |                          $$ |

input/headers.hpp

@@ -46,6 +46,15 @@
 
 namespace ipxp {
 
+struct ip6_frag {
+   uint8_t ip_proto;
+   uint8_t reserved;
+   uint16_t frag_off;
+#define IPV6_FRAGMENT_OFFSET 0xFFF8
+#define IPV6_MORE_FRAGMENTS 0x1
+   uint32_t frag_id;
+} __attribute__((packed));
+
 struct grehdr {
    uint16_t flags;
 #define GRE_CHECKSUM 0x8000

input/parser.cpp

@@ -51,6 +51,10 @@ static uint32_t s_total_pkts = 0;
 #define DEBUG_CODE(code)
 #endif
 
+// masks for iphdr::frag_off
+#define IPV4_MORE_FRAGMENTS 0x2000
+#define IPV4_FRAGMENT_OFFSET 0x1FFF
+
 /**
  * \brief Parse specific fields from ETHERNET frame header.
  * \param [in] data_ptr Pointer to begin of header.
@@ -316,6 +320,9 @@ inline uint16_t parse_ipv4_hdr(const u_char *data_ptr, uint16_t data_len, Packet
    pkt->ip_flags = (ntohs(ip->frag_off) & 0xE000) >> 13;
    pkt->src_ip.v4 = ip->saddr;
    pkt->dst_ip.v4 = ip->daddr;
+   pkt->frag_id = ntohs(ip->id);
+   pkt->frag_off = ntohs(ip->frag_off) & IPV4_FRAGMENT_OFFSET;
+   pkt->more_fragments = ntohs(ip->frag_off) & IPV4_MORE_FRAGMENTS;
 
    DEBUG_MSG("IPv4 header:\n");
    DEBUG_MSG("\tHDR version:\t%u\n",   ip->version);
@@ -335,7 +342,7 @@ inline uint16_t parse_ipv4_hdr(const u_char *data_ptr, uint16_t data_len, Packet
 }
 
 /**
- * \brief Skip IPv6 extension headers.
+ * \brief Skip/parse IPv6 extension headers.
  * \param [in] data_ptr Pointer to begin of header.
  * \param [in] data_len Length of packet data in `data_ptr`.
  * \param [out] pkt Pointer to Packet structure where parsed fields will be stored.
@@ -347,7 +354,7 @@ uint16_t skip_ipv6_ext_hdrs(const u_char *data_ptr, uint16_t data_len, Packet *p
    uint8_t next_hdr = pkt->ip_proto;
    uint16_t hdrs_len = 0;
 
-   /* Skip extension headers... */
+   /* Skip/parse extension headers... */
    while (1) {
       if ((int)sizeof(struct ip6_ext) > data_len - hdrs_len) {
          throw "Parser detected malformed packet";
@@ -361,7 +368,13 @@ uint16_t skip_ipv6_ext_hdrs(const u_char *data_ptr, uint16_t data_len, Packet *p
       } else if (next_hdr == IPPROTO_AH) {
          hdrs_len += (ext->ip6e_len << 2) - 2;
       } else if (next_hdr == IPPROTO_FRAGMENT) {
-         hdrs_len += 8;
+         // extract the fragmentation info
+         auto *frag = reinterpret_cast<const ip6_frag *>(data_ptr + hdrs_len);
+         pkt->frag_id = ntohl(frag->frag_id);
+         pkt->frag_off = ntohs(frag->frag_off) & IPV6_FRAGMENT_OFFSET;
+         pkt->more_fragments = ntohs(frag->frag_off) & IPV6_MORE_FRAGMENTS;
+
+         hdrs_len += sizeof(ip6_frag);
       } else if (next_hdr == IPPROTO_MH) {
          hdrs_len += (ext->ip6e_len << 3) + 8;
          // Mobility header can't have payload now but may have it in the future

storage/cache.cpp

@@ -164,7 +164,9 @@ void FlowRecord::update(const Packet &pkt, bool src)
 NHTFlowCache::NHTFlowCache() :
    m_cache_size(0), m_line_size(0), m_line_mask(0), m_line_new_idx(0),
    m_qsize(0), m_qidx(0), m_timeout_idx(0), m_active(0), m_inactive(0),
-   m_split_biflow(false), m_keylen(0), m_key(), m_key_inv(), m_flow_table(nullptr), m_flow_records(nullptr)
+   m_split_biflow(false), m_enable_fragmentation_cache(true), m_keylen(0),
+   m_key(), m_key_inv(), m_flow_table(nullptr), m_flow_records(nullptr),
+   m_fragmentation_cache(0, 0)
 {
 }
 
@@ -213,6 +215,15 @@ void NHTFlowCache::init(const char *params)
    }
 
    m_split_biflow = parser.m_split_biflow;
+   m_enable_fragmentation_cache = parser.m_enable_fragmentation_cache;
+
+   if (m_enable_fragmentation_cache) {
+      try {
+         m_fragmentation_cache = FragmentationCache(parser.m_frag_cache_size, parser.m_frag_cache_timeout);
+      } catch (std::bad_alloc &e) {
+         throw PluginError("not enough memory for fragment cache allocation");
+      }
+   }
 
 #ifdef FLOW_CACHE_STATS
    m_empty = 0;
@@ -301,6 +312,10 @@ int NHTFlowCache::put_pkt(Packet &pkt)
 {
    int ret = plugins_pre_create(pkt);
 
+   if (m_enable_fragmentation_cache) {
+      try_to_fill_ports_to_fragmented_packet(pkt);
+   }
+
    if (!create_hash_key(pkt)) { // saves key value and key length into attributes NHTFlowCache::key and NHTFlowCache::m_keylen
       return 0;
    }
@@ -455,6 +470,11 @@ int NHTFlowCache::put_pkt(Packet &pkt)
    return 0;
 }
 
+void NHTFlowCache::try_to_fill_ports_to_fragmented_packet(Packet& packet)
+{
+   m_fragmentation_cache.process_packet(packet);
+}
+
 uint8_t NHTFlowCache::get_export_reason(Flow &flow)
 {
    if ((flow.src_tcp_flags | flow.dst_tcp_flags) & (0x01 | 0x04)) {

storage/cache.hpp

@@ -39,6 +39,8 @@
 #include <ipfixprobe/flowifc.hpp>
 #include <ipfixprobe/utils.hpp>
 
+#include "fragmentationCache/fragmentationCache.hpp"
+
 namespace ipxp {
 
 struct __attribute__((packed)) flow_key_v4_t {
@@ -93,10 +95,15 @@ class CacheOptParser : public OptionsParser
    uint32_t m_active;
    uint32_t m_inactive;
    bool m_split_biflow;
+   bool m_enable_fragmentation_cache;
+   std::size_t m_frag_cache_size;
+   time_t m_frag_cache_timeout;
 
    CacheOptParser() : OptionsParser("cache", "Storage plugin implemented as a hash table"),
       m_cache_size(1 << DEFAULT_FLOW_CACHE_SIZE), m_line_size(1 << DEFAULT_FLOW_LINE_SIZE),
-      m_active(DEFAULT_ACTIVE_TIMEOUT), m_inactive(DEFAULT_INACTIVE_TIMEOUT), m_split_biflow(false)
+      m_active(DEFAULT_ACTIVE_TIMEOUT), m_inactive(DEFAULT_INACTIVE_TIMEOUT), m_split_biflow(false),
+      m_enable_fragmentation_cache(true), m_frag_cache_size(10007), // Prime for better distribution in hash table
+      m_frag_cache_timeout(3)
    {
       register_option("s", "size", "EXPONENT", "Cache size exponent to the power of two",
          [this](const char *arg){try {unsigned exp = str2num<decltype(exp)>(arg);
@@ -121,6 +128,33 @@ class CacheOptParser : public OptionsParser
          OptionFlags::RequiredArgument);
       register_option("S", "split", "", "Split biflows into uniflows",
          [this](const char *arg){ m_split_biflow = true; return true;}, OptionFlags::NoArgument);
+      register_option("fe", "frag-enable", "true|false", "Enable/disable fragmentation cache. Enabled (true) by default.",
+         [this](const char *arg){
+            if (strcmp(arg, "true") == 0) {
+               m_enable_fragmentation_cache = true;
+            } else if (strcmp(arg, "false") == 0) {
+               m_enable_fragmentation_cache = false;
+            } else {
+               return false;
+            }
+            return true;
+         }, OptionFlags::RequiredArgument);
+      register_option("fs", "frag-size", "size", "Size of fragmentation cache, must be at least 1. Default value is 10007.", [this](const char *arg) {
+         try {
+            m_frag_cache_size = str2num<decltype(m_frag_cache_size)>(arg);
+         } catch(std::invalid_argument &e) {
+            return false;
+         }
+         return m_frag_cache_size > 0;
+      });
+      register_option("ft", "frag-timeout", "TIME", "Timeout of fragments in fragmentation cache in seconds. Default value is 3.", [this](const char *arg) {
+         try {
+            m_frag_cache_timeout = str2num<decltype(m_frag_cache_timeout)>(arg);
+         } catch(std::invalid_argument &e) {
+            return false;
+         }
+         return true;
+      });
    }
 };
 
@@ -177,12 +211,16 @@ class NHTFlowCache : public StoragePlugin
    uint32_t m_active;
    uint32_t m_inactive;
    bool m_split_biflow;
+   bool m_enable_fragmentation_cache;
    uint8_t m_keylen;
    char m_key[MAX_KEY_LENGTH];
    char m_key_inv[MAX_KEY_LENGTH];
    FlowRecord **m_flow_table;
    FlowRecord *m_flow_records;
 
+   FragmentationCache m_fragmentation_cache;
+
+   void try_to_fill_ports_to_fragmented_packet(Packet& packet);
    void flush(Packet &pkt, size_t flow_index, int ret, bool source_flow);
    bool create_hash_key(Packet &pkt);
    void export_flow(size_t index);

storage/fragmentationCache/fragmentationCache.cpp

@@ -0,0 +1,89 @@
+/**
+ * \file
+ * \author Pavel Siska <siska@cesnet.cz>
+ * \author Jakub Antonín Štigler xstigl00@stud.fit.vut.cz
+ * \brief Contains implementation of the FragmentationCache class for managing fragmented packet
+ * data using a fragmentation table.
+ *
+ * The FragmentationCache class handles the processing and management of fragmented network packets.
+ * It utilizes a fragmentation table to store and retrieve necessary data for processing fragmented
+ * packets.
+ */
+/*
+ * Copyright (C) 2023 CESNET
+ *
+ * LICENSE TERMS
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in
+ *    the documentation and/or other materials provided with the
+ *    distribution.
+ * 3. Neither the name of the Company nor the names of its contributors
+ *    may be used to endorse or promote products derived from this
+ *    software without specific prior written permission.
+ */
+
+#include "../xxhash.h"
+#include "fragmentationCache.hpp"
+#include "timevalUtils.hpp"
+
+#include <cstring>
+
+namespace ipxp {
+
+FragmentationCache::FragmentationCache(std::size_t table_size, time_t timeout_in_seconds)
+    : m_timeout({timeout_in_seconds, 0})
+    , m_fragmentation_table(table_size)
+{
+}
+
+void FragmentationCache::process_packet(Packet& packet)
+{
+    if (!is_packet_fragmented(packet)) {
+        return; // Packet is not fragmented, no further action needed.
+    }
+    process_fragmented_packet(packet);
+}
+
+void FragmentationCache::process_fragmented_packet(Packet& packet) noexcept
+{
+    if (is_packet_first_fragment(packet)) {
+        m_fragmentation_table.insert(packet);
+    } else {
+        auto fragmentation_data = m_fragmentation_table.find(packet);
+        if (fragmentation_data) {
+            fill_missing_packet_data(packet, *fragmentation_data);
+        }
+    }
+}
+
+void FragmentationCache::fill_missing_packet_data(
+    Packet& packet,
+    const FragmentationData& fragmentation_data) noexcept
+{
+    if (!is_fragmentation_data_timedouted(packet, fragmentation_data)) {
+        fill_ports_to_packet(packet, fragmentation_data);
+    }
+}
+
+bool FragmentationCache::is_fragmentation_data_timedouted(
+    const Packet& packet,
+    const FragmentationData& data) const noexcept
+{
+    return packet.ts > data.timestamp + m_timeout;
+}
+
+void FragmentationCache::fill_ports_to_packet(
+    Packet& packet,
+    const FragmentationData& fragmentation_data) const noexcept
+{
+    packet.src_port = fragmentation_data.source_port;
+    packet.dst_port = fragmentation_data.destination_port;
+}
+
+} // namespace ipxp