Merge fad8757b89 into 22561a577c

This sets correct C++ standard for newer Protobuf installations as
FindProtobuf module only sets cxx_std_11

src/CMakeLists.txt

@@ -594,26 +594,26 @@ if (ENABLE_GOBGP_SUPPORT)
         target_link_libraries(gobgp_action absl::base absl::synchronization)
     endif()
 
-    # By default use module supplied by cmake to search for Protobuf 
-    set(FIND_PACKAGE_MODE_PROTOBUF "MODULE")
-
     if (DO_NOT_USE_SYSTEM_LIBRARIES_FOR_BUILD)
       # We add our custom path to Protobuf to top of search_list used by find_package: https://cmake.org/cmake/help/latest/variable/CMAKE_PREFIX_PATH.html
       # This approach has advantage over Protobuf_DIR which requires us to set direct path to cmake folder of custom built dependency
       # which resides in vendor specific folder with name lib which may be lib64 on CentOS platforms:
       # protobuf_21_12/lib/cmake/protobuf or  protobuf_21_12/lib64/cmake/protobuf on CentOS 
       list(APPEND CMAKE_PREFIX_PATH ${PROTOCOL_BUFFERS_CUSTOM_INSTALL_PATH}) 
-
-      # Switch to use to configuration supplied by custom Protobuf installation as it may be better
-      set(FIND_PACKAGE_MODE_PROTOBUF "CONFIG")
-
-      # Apparently it's required to set this flag because without this flag set it cannot find protoc when custom library path is in use
-      # https://github.com/protocolbuffers/protobuf/issues/1931
-      set(protobuf_MODULE_COMPATIBLE true)
     endif()
 
-    # https://cmake.org/cmake/help/latest/module/FindProtobuf.html
-    find_package(Protobuf ${FIND_PACKAGE_MODE_PROTOBUF} REQUIRED)
+    # Apparently it's required to set this flag because without this flag set it cannot find protoc when custom library path is in use
+    # https://github.com/protocolbuffers/protobuf/issues/1931
+    set(protobuf_MODULE_COMPATIBLE true)
+
+    # Switch to use to configuration supplied by custom Protobuf installation as it may be better
+    find_package(Protobuf CONFIG)
+
+    if (NOT Protobuf_FOUND)
+      # Fall back to module supplied by cmake to search for Protobuf
+      # https://cmake.org/cmake/help/latest/module/FindProtobuf.html
+      find_package(Protobuf MODULE REQUIRED)
+    endif()
 
     if (Protobuf_FOUND)
       message(STATUS "Found Protobuf ${Protobuf_VERSION}")

src/netflow_plugin/ipfix.hpp

@@ -45,9 +45,15 @@
 #define IPFIX_ENGINE_TYPE 38
 #define IPFIX_ENGINE_ID 39
 
+#define IPFIX_FRAGMENT_IDENTIFICATION 54
+
+#define IPFIX_SOURCE_MAC_ADDRESS 56
+
 #define IPFIX_FLOW_DIRECTION 61
 #define IPFIX_IPV6_NEXT_HOP 62
 
+#define IPFIX_DESTINATION_MAC_ADDRESS 80
+
 #define IPFIX_FORWARDING_STATUS 89
 
 #define IPFIX_FLOW_END_REASON 136
@@ -56,6 +62,12 @@
 #define IPFIX_FLOW_START_MILLISECONDS 152
 #define IPFIX_FLOW_END_MILLISECONDS 153
 
+// We use 8 byte encoding: https://datatracker.ietf.org/doc/html/rfc7011#section-6.1.10
+#define IPFIX_FLOW_START_NANOSECONDS 156
+#define IPFIX_FLOW_END_NANOSECONDS 157
+
+#define IPFIX_SAMPLING_SELECTOR_ALGORITHM 304
+
 #define IPFIX_SAMPLING_PACKET_INTERVAL 305
 
 #define IPFIX_SAMPLING_PACKET_SPACE 306
@@ -65,7 +77,8 @@
 #define IPFIX_SELECTOR_TOTAL_PACKETS_OBSERVED 318
 #define IPFIX_SELECTOR_TOTAL_PACKETS_SELECTED 319
 
-
+// Sampler types https://www.iana.org/assignments/psamp-parameters/psamp-parameters.xhtml
+#define IPFIX_SAMPLER_TYPE_SYSTEMATIC_COUNT_BASED_SAMPLING 1
 
 class __attribute__((__packed__)) ipfix_header_common_t {
     public:
@@ -118,5 +131,3 @@ class __attribute__((__packed__)) ipfix_options_header_t {
     uint16_t field_count       = 0;
     uint16_t scope_field_count = 0;
 };
-
-

src/netflow_plugin/ipfix_collector.cpp

@@ -1,3 +1,6 @@
+// That's not a module as we do refactoring right now in small steps
+// TODO: place make it proper module
+
 void update_ipfix_sampling_rate(uint32_t sampling_rate, const std::string& client_addres_in_string_format);
 
 // https://tools.ietf.org/html/rfc5101#page-18
@@ -172,8 +175,12 @@ bool process_ipfix_options_template(const uint8_t* pkt, size_t flowset_length, u
     bool updated                   = false;
     bool updated_existing_template = false;
 
-    add_update_peer_template(netflow_protocol_version_t::ipfix, global_ipfix_templates, source_id, template_id, client_addres_in_string_format,
-                             field_template, updated, updated_existing_template);
+    add_update_peer_template(netflow_protocol_version_t::ipfix, global_ipfix_templates, global_ipfix_templates_mutex, source_id,
+                             template_id, client_addres_in_string_format, field_template, updated, updated_existing_template);
+
+    // This code is not perfect from locks perspective as we read global_ipfix_templates without any locks below
+
+    // NB! Please be careful with changing name of variable as it's part of serialisation protocol
 
     if (updated_existing_template) {
         ipfix_template_data_updates++;
@@ -263,12 +270,10 @@ bool process_ipfix_template(const uint8_t* pkt, size_t flowset_length, uint32_t
         bool updated                   = false;
         bool updated_existing_template = false;
 
-        add_update_peer_template(netflow_protocol_version_t::ipfix, global_ipfix_templates, source_id, template_id, client_addres_in_string_format,
+        add_update_peer_template(netflow_protocol_version_t::ipfix, global_ipfix_templates,
+                                 global_ipfix_templates_mutex, source_id, template_id, client_addres_in_string_format,
                                  field_template, updated, updated_existing_template);
 
-        // If we have any changes for this template, let's flush them to disk
-        if (updated) {
-        }
 
         if (updated_existing_template) {
             ipfix_template_data_updates++;
@@ -278,12 +283,15 @@ bool process_ipfix_template(const uint8_t* pkt, size_t flowset_length, uint32_t
     return true;
 }
 
-
 bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const uint8_t* data, simple_packet_t& packet, netflow_meta_info_t& flow_meta) {
     switch (record_type) {
     case IPFIX_IN_BYTES:
         if (record_length > sizeof(packet.length)) {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IN_BYTES: " << record_length;
+            }
         } else {
             BE_COPY(packet.length);
 
@@ -299,6 +307,11 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
     case IPFIX_IN_PACKETS:
         if (record_length > sizeof(packet.number_of_packets)) {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IN_PACKETS: " << record_length;
+            }
+
         } else {
             BE_COPY(packet.number_of_packets);
 
@@ -309,6 +322,11 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
     case IPFIX_IN_PROTOCOL:
         if (record_length > sizeof(packet.protocol)) {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IN_PROTOCOL: " << record_length;
+            }
+
         } else {
             BE_COPY(packet.protocol);
 
@@ -327,12 +345,21 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             memcpy(&packet.flags, data + 1, 1);
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_TCP_FLAGS: " << record_length;
+            }
         }
 
         break;
     case IPFIX_L4_SRC_PORT:
         if (record_length > sizeof(packet.source_port)) {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_L4_SRC_PORT: " << record_length;
+            }
+
         } else {
             BE_COPY(packet.source_port);
 
@@ -344,6 +371,11 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
     case IPFIX_L4_DST_PORT:
         if (record_length > sizeof(packet.destination_port)) {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_L4_DST_PORT: " << record_length;
+            }
+
         } else {
             BE_COPY(packet.destination_port);
 
@@ -355,6 +387,11 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
     case IPFIX_IPV4_SRC_ADDR:
         if (record_length > sizeof(packet.src_ip)) {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IPV4_SRC_ADDR: " << record_length;
+            }
+
         } else {
             memcpy(&packet.src_ip, data, record_length);
         }
@@ -363,6 +400,11 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
     case IPFIX_IPV4_DST_ADDR:
         if (record_length > sizeof(packet.dst_ip)) {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IPV4_DST_ADDR: " << record_length;
+            }
+
         } else {
             memcpy(&packet.dst_ip, data, record_length);
         }
@@ -380,6 +422,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             // std::cout << "IP next hop: " << convert_ip_as_uint_to_string(ip_next_hop_ipv4) << std::endl;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IPV4_NEXT_HOP: " << record_length;
+            }
         }
 
         break;
@@ -396,6 +442,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             // std::cout << "BGP next hop: " << convert_ip_as_uint_to_string(bgp_next_hop_ipv4) << std::endl;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_BGP_NEXT_HOP_IPV4_ADDRESS: " << record_length;
+            }
         }
 
         break;
@@ -411,6 +461,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             // std::cout << "bgp next hop: " << print_ipv6_address(ipv6_next_hop) << std::endl;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IPV6_NEXT_HOP: " << record_length;
+            }
         }
 
 
@@ -432,6 +486,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             packet.src_asn = src_asn;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_SRC_AS: " << record_length;
+            }
         }
 
         break;
@@ -450,8 +508,36 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             packet.dst_asn = dst_asn;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_DST_AS: " << record_length;
+            }
         }
 
+        break;
+    case IPFIX_SOURCE_MAC_ADDRESS:
+        if (record_length == 6) {
+            // Copy it directly to packet structure
+            memcpy(&packet.source_mac, data, record_length);
+        } else {
+            ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Too large field for IPFIX_SOURCE_MAC_ADDRESS";
+            }
+        }
+        break;
+    case IPFIX_DESTINATION_MAC_ADDRESS:
+        if (record_length == 6) {
+            // Copy it directly to packet structure
+            memcpy(&packet.destination_mac, data, record_length);
+        } else {
+            ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Too large field for IPFIX_DESTINATION_MAC_ADDRESS";
+            }
+        }
         break;
     // According to https://www.iana.org/assignments/ipfix/ipfix.xhtml interfaces can be 4 byte only
     case IPFIX_INPUT_SNMP:
@@ -459,10 +545,20 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             uint32_t input_interface = 0;
             memcpy(&input_interface, data, record_length);
 
+            input_interface        = fast_ntoh(input_interface);
+            packet.input_interface = input_interface;
+        } else if (record_length == 2) {
+            uint16_t input_interface = 0;
+            memcpy(&input_interface, data, record_length);
+
             input_interface        = fast_ntoh(input_interface);
             packet.input_interface = input_interface;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_INPUT_SNMP: " << record_length;
+            }
         }
 
         break;
@@ -471,33 +567,51 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             uint32_t output_interface = 0;
             memcpy(&output_interface, data, record_length);
 
+            output_interface        = fast_ntoh(output_interface);
+            packet.output_interface = output_interface;
+        } else if (record_length == 2) {
+            uint16_t output_interface = 0;
+            memcpy(&output_interface, data, record_length);
+
             output_interface        = fast_ntoh(output_interface);
             packet.output_interface = output_interface;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_OUTPUT_SNMP: " << record_length;
+            }
         }
 
         break;
     case IPFIX_IPV6_SRC_ADDR:
         // It should be 16 bytes only
-            if (record_length == 16) {
-                memcpy(&packet.src_ipv6, data, record_length);
-                // Set protocol version to IPv6
-                packet.ip_protocol_version = 6;
-            } else {
-                ipfix_too_large_field++;
+        if (record_length == 16) {
+            memcpy(&packet.src_ipv6, data, record_length);
+            // Set protocol version to IPv6
+            packet.ip_protocol_version = 6;
+        } else {
+            ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IPV6_SRC_ADDR: " << record_length;
             }
+        }
 
         break;
     case IPFIX_IPV6_DST_ADDR:
         // It should be 16 bytes only
-            if (record_length == 16) {
-                memcpy(&packet.dst_ipv6, data, record_length);
-                // Set protocol version to IPv6
-                packet.ip_protocol_version = 6;
-            } else {
-                ipfix_too_large_field++;
+        if (record_length == 16) {
+            memcpy(&packet.dst_ipv6, data, record_length);
+            // Set protocol version to IPv6
+            packet.ip_protocol_version = 6;
+        } else {
+            ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_IPV6_DST_ADDR: " << record_length;
             }
+        }
         break;
     case IPFIX_FIRST_SWITCHED:
         // Mikrotik uses this encoding
@@ -510,6 +624,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             packet.flow_start = flow_started;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FIRST_SWITCHED: " << record_length;
+            }
         }
 
         break;
@@ -524,6 +642,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             packet.flow_end = flow_finished;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_LAST_SWITCHED: " << record_length;
+            }
         }
 
         break;
@@ -539,6 +661,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             packet.flow_start = flow_started;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FLOW_START_MILLISECONDS: " << record_length;
+            }
         }
 
         break;
@@ -553,9 +679,57 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             packet.flow_end = flow_finished;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FLOW_END_MILLISECONDS: " << record_length;
+            }
         }
 
         break;
+        // Netgate TNSR uses IPFIX_FLOW_START_NANOSECONDS and IPFIX_FLOW_END_NANOSECONDS
+    case IPFIX_FLOW_START_NANOSECONDS:
+        if (record_length == 8) {
+            uint64_t flow_started = 0;
+
+            memcpy(&flow_started, data, record_length);
+            flow_started = fast_ntoh(flow_started);
+
+            // We cast unsigned to signed and it may cause issues
+            packet.flow_start = flow_started;
+
+            // Convert to milliseconds
+            packet.flow_start = packet.flow_start / 1000000;
+        } else {
+            ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FLOW_START_NANOSECONDS: " << record_length;
+            }
+        }
+
+        break;
+    case IPFIX_FLOW_END_NANOSECONDS:
+        if (record_length == 8) {
+            uint64_t flow_finished = 0;
+
+            memcpy(&flow_finished, data, record_length);
+            flow_finished = fast_ntoh(flow_finished);
+
+            // We cast unsigned to signed and it may cause issues
+            packet.flow_end = flow_finished;
+
+            // Convert to milliseconds
+            packet.flow_end = packet.flow_end / 1000000;
+        } else {
+            ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FLOW_END_NANOSECONDS: " << record_length;
+            }
+        }
+
+        break;
+
     case IPFIX_FORWARDING_STATUS:
         // TODO: we did using theoretical information and did not test it at all
         // Documented here: https://www.iana.org/assignments/ipfix/ipfix.xhtml#forwarding-status
@@ -577,6 +751,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
         } else {
             // It must be exactly one byte
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FORWARDING_STATUS: " << record_length;
+            }
         }
 
         break;
@@ -588,47 +766,74 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             flow_meta.data_link_frame_size = fast_ntoh(datalink_frame_size);
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_DATALINK_FRAME_SIZE: " << record_length;
+            }
         }
 
         break;
     case IPFIX_DATALINK_FRAME_SECTION:
         // Element 315: https://www.iana.org/assignments/ipfix/ipfix.xhtml
 
-        // We create block here as gcc does not want to compile without it with error: error: jump to case label
-        {
         // It's packet header as is in variable length encoding
+        if (true) {
             ipfix_inline_headers++;
 
-            // This data is encoded using following approach: https://datatracker.ietf.org/doc/html/rfc7011#section-7
-            const uint8_t* field_length_ptr = (const uint8_t*)data;
+            // This packet is ended using IPFIX variable length encoding and it may have two possible ways of length encoding
+            // https://datatracker.ietf.org/doc/html/rfc7011#section-7
+            if (flow_meta.variable_field_length_encoding == variable_length_encoding_t::single_byte || flow_meta.variable_field_length_encoding == variable_length_encoding_t::two_byte) {
 
-            // Get rid of pointer syntax
-            uint8_t packet_header_length = *field_length_ptr;
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Packet header length: " << flow_meta.variable_field_length;
+                }
 
-            // logger << log4cpp::Priority::DEBUG << "Packet header length is " << int(packet_header_length);
+                if (flow_meta.variable_field_length != 0) {
+                    bool read_packet_length_from_ip_header = true;
 
-            // We will support only fields which are < 255 bytes length as it simpler way of encoding
-            if (packet_header_length != 0 && packet_header_length < 255) {
-                bool read_packet_length_from_ip_header = true;
+                    bool extract_tunnel_traffic = false;
 
-                bool extract_tunnel_traffic = false;
+                    const uint8_t* payload_shift = nullptr;
 
-                auto result =
-                    parse_raw_packet_to_simple_packet_full_ng((u_char*)(data + sizeof(uint8_t)), packet_header_length, packet_header_length,
-                                                           flow_meta.nested_packet, extract_tunnel_traffic, read_packet_length_from_ip_header);
+                    if (flow_meta.variable_field_length_encoding == variable_length_encoding_t::single_byte) {
+                        payload_shift = data + sizeof(uint8_t);
+                    } else if (flow_meta.variable_field_length_encoding == variable_length_encoding_t::two_byte) {
+                        payload_shift = data + sizeof(uint8_t) + sizeof(uint16_t);
+                    }
 
-                if (result != network_data_stuctures::parser_code_t::success) {
-                    // Cannot decode data
-                    ipfix_inline_header_parser_error++;
+                    auto result =
+                        parse_raw_packet_to_simple_packet_full_ng(payload_shift, flow_meta.variable_field_length,
+                                                                    flow_meta.variable_field_length, flow_meta.nested_packet,
+                                                                  extract_tunnel_traffic, read_packet_length_from_ip_header);
+
+                    if (result != network_data_stuctures::parser_code_t::success) {
+                        // Cannot decode data
+                        ipfix_inline_header_parser_error++;
+
+                        if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                            logger << log4cpp::Priority::DEBUG << "Cannot parse packet header with error: " << network_data_stuctures::parser_code_to_string(result);
+                        }
+
+                    } else {
+                        // Successfully decoded data
+                        ipfix_inline_header_parser_success++;
+
+                        flow_meta.nested_packet_parsed = true;
+                        // logger << log4cpp::Priority::DEBUG << "IPFIX inline extracted packet: " << print_simple_packet(flow_meta.nested_packet);
+                    }
                 } else {
-                    // Successfully decoded data
-                    ipfix_inline_header_parser_success++;
-                    
-                    flow_meta.nested_packet_parsed = true;
-                    // logger << log4cpp::Priority::DEBUG << "IPFIX inline extracted packet: " << print_simple_packet(flow_meta.nested_packet);
+                    ipfix_inline_encoding_error++;
+
+                    if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                        logger << log4cpp::Priority::DEBUG << "Zero length variable fields are not supported";
+                    }
                 }
             } else {
-                // It may be zero length value or it may be 255 which means that data is longer
+                ipfix_inline_encoding_error++;
+                
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Unknown variable field encoding type";
+                }
             }
         }
 
@@ -646,6 +851,10 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             // std::cout << "Flow direction: " << int(flow_direction) << std::endl;
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FLOW_DIRECTION: " << record_length;
+            }
         }
 
         break;
@@ -671,6 +880,61 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
             }
         } else {
             ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FLOW_END_REASON: " << record_length;
+            }
+        }
+
+        break;
+    case IPFIX_FRAGMENT_IDENTIFICATION:
+        //
+        // Specification: https://www.rfc-editor.org/rfc/rfc5102.html#section-5.4.23
+        //
+        // IPFIX uses 32 bit values to accommodate following cases:
+        //  - 16 bit IPv4 identification field https://www.rfc-editor.org/rfc/rfc791
+        //  - 32 bit IPv6 identification field https://en.wikipedia.org/wiki/IPv6_packet#Fragment
+        //
+        // Juniper uses it on J MX platforms but they do not have much information about it:
+        // https://www.juniper.net/documentation/us/en/software/junos/flow-monitoring/topics/concept/inline-sampling-overview.html
+        // I asked https://t.me/dgubin about it
+        //
+        // I did review of dump from J MX and I can confirm that values for IPv4 do not exceed maximum value for uint16_t (65535)
+        // 
+        // J MX is doing something fun with this field. I got dump in hands and in this dump of 42421 packets only 2337 have non zero value of this field.
+        // Clearly they violate RFC and do not populate this field unconditionally as RFC dictates.
+        //
+        // I see cases like this which is very likely non first fragment of fragmented series of packets as we do not have ports:
+        // Identification: 20203 ipv4:0 > ipv4:0 protocol: udp frag: 0  packets: 1 size: 352 bytes ip size: 352 bytes ttl: 0 sample ratio: 1
+        // 
+        // And I see packets like this which may be first packet in fragmented series of packets as we do indeed have ports here and packet length is high: 
+        // Identification: 2710 ipv4:53 > ipv4:45134 protocol: udp frag: 0  packets: 1 size: 1476 bytes ip size: 1476 bytes ttl: 0 sample ratio: 1
+        //
+        // And majority of packets looks this way:
+        // Identification: 0 ipv4:80 > ipv4:50179 protocol: tcp flags: ack frag: 0  packets: 1 size: 40 bytes ip size: 40 bytes ttl: 0 sample ratio: 1
+        //
+        // We clearly can distinguish first fragmented packet and non first fragmented packet
+        //
+        // TODO: this logic must be enabled via flag only as this is non RFC compliant behavior and we need to have confirmation from J
+        //
+        // We have this guide from J: https://www.juniper.net/documentation/us/en/software/junos/flow-monitoring/topics/concept/services-ipfix-flow-aggregation-ipv6-extended-attributes.html but it's written in exceptionally weird way and raises more questions then answers
+        //
+
+        // It's exactly 4 bytes
+        if (record_length == 4) {
+            uint32_t fragment_identification = 0;
+
+            memcpy(&fragment_identification, data, record_length);
+
+            fragment_identification = fast_ntoh(fragment_identification);
+
+            // logger << log4cpp::Priority::INFO << "Fragment identification: " << fragment_identification; 
+        } else {
+            ipfix_too_large_field++;
+
+            if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_FRAGMENT_IDENTIFICATION: " << record_length;
+            }
         }
 
         break;
@@ -679,7 +943,6 @@ bool ipfix_record_to_flow(uint32_t record_type, uint32_t record_length, const ui
     return true;
 }
 
-
 // Read options data packet with known template
 bool ipfix_options_flowset_to_store(const uint8_t* pkt,
                                     const ipfix_header_t* ipfix_header,
@@ -689,7 +952,16 @@ bool ipfix_options_flowset_to_store(const uint8_t* pkt,
     pkt += flow_template->option_scope_length;
 
     uint32_t sampling_rate = 0;
-    uint32_t offset        = 0;
+
+    // Field shift in memory
+    uint32_t offset = 0;
+
+    // Sampling algorithm for exotic sampling types
+    uint16_t sampling_selector_algorithm = 0;
+
+    // We use these fields to work with systematic count-based Sampling Selector on Nokia
+    uint32_t sampling_packet_space    = 0;
+    uint32_t sampling_packet_interval = 0;
 
     device_timeouts_t device_timeouts{};
 
@@ -699,46 +971,81 @@ bool ipfix_options_flowset_to_store(const uint8_t* pkt,
         // Time to extract sampling rate
         if (elem.record_type == IPFIX_SAMPLING_INTERVAL) {
             // RFC suggest that this field is 4 byte: https://www.iana.org/assignments/ipfix/ipfix.xhtml
-            if (elem.record_length > sizeof(sampling_rate)) {
-                logger << log4cpp::Priority::ERROR << "Unexpectedly big size for IPFIX_SAMPLING_INTERVAL: " << elem.record_length;
+            if (elem.record_length == 4) {
+                uint32_t current_sampling_rate = 0;
+                memcpy(&current_sampling_rate, data_shift, elem.record_length);
+
+                // TODO: we do not convert value to little endian as sampling update function expects big endian / network byte order
+
+                sampling_rate = current_sampling_rate;
+
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "4 byte encoded IPFIX_SAMPLING_INTERVAL sampling rate: " << sampling_rate
+                           << " from " << client_addres_in_string_format;
+                }
+            } else {
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Unexpectedly big size for IPFIX_SAMPLING_INTERVAL: " << elem.record_length;
+                }
+
                 ipfix_too_large_field++;
-
-                return false;
-            }
-
-            bool result = be_copy_function(data_shift, (uint8_t*)&sampling_rate, sizeof(sampling_rate), elem.record_length);
-
-            if (!result) {
-                logger << log4cpp::Priority::ERROR
-                       << "Prevented attempt to read outside of allowed memory region for IPFIX_SAMPLING_INTERVAL";
-                return false;
             }
         } else if (elem.record_type == IPFIX_SAMPLING_PACKET_INTERVAL) {
             // RFC suggest that this field is 4 byte: https://www.iana.org/assignments/ipfix/ipfix.xhtml
-            if (elem.record_length > sizeof(sampling_rate)) {
-                logger << log4cpp::Priority::ERROR
-                       << "Unexpectedly big size for IPFIX_SAMPLING_PACKET_INTERVAL: " << elem.record_length;
+            if (elem.record_length == 4) {
+                uint32_t current_sampling_packet_interval = 0;
+
+                memcpy(&current_sampling_packet_interval, data_shift, elem.record_length);
+
+                current_sampling_packet_interval = fast_ntoh(current_sampling_packet_interval);
+
+                // Well, we need this information to deal with systematic count-based Sampling Selector on Nokia
+                sampling_packet_interval = current_sampling_packet_interval;
+
+                // And we need this value to use as regular sampling rate on Cisco NSC
+                // We need to return it to big endian again we sampling logic in IPFIX uses big endian / network byte order
+                sampling_rate = fast_hton(sampling_packet_interval);
+            } else {
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG
+                           << "Unexpectedly big size for IPFIX_SAMPLING_PACKET_INTERVAL: " << elem.record_length;
+                }
+
                 ipfix_too_large_field++;
-
-                return false;
-            }
-
-            bool result = be_copy_function(data_shift, (uint8_t*)&sampling_rate, sizeof(sampling_rate), elem.record_length);
-
-            if (!result) {
-                logger << log4cpp::Priority::ERROR << "Prevented attempt to read outside of allowed memory region for IPFIX_SAMPLING_PACKET_INTERVAL";
-                return false;
             }
         } else if (elem.record_type == IPFIX_SAMPLING_PACKET_SPACE) {
-            uint32_t sampling_packet_space = 0;
-
             // RFC requires this field to be 4 byte long
             if (elem.record_length == 4) {
                 memcpy(&sampling_packet_space, data_shift, elem.record_length);
 
                 sampling_packet_space = fast_ntoh(sampling_packet_space);
             } else {
-                logger << log4cpp::Priority::DEBUG << "Unexpected size for IPFIX_SAMPLING_PACKET_SPACE: " << elem.record_length;
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Unexpected size for IPFIX_SAMPLING_PACKET_SPACE: " << elem.record_length;
+                }
+
+                ipfix_too_large_field++;
+
+                // We're OK to continue process, we should not stop it
+            }
+
+        } else if (elem.record_type == IPFIX_SAMPLING_SELECTOR_ALGORITHM) {
+            // RFC requires this field to be 2 byte long
+            // You can find all possible values for it here: https://www.iana.org/assignments/psamp-parameters/psamp-parameters.xhtml
+            if (elem.record_length == 2) {
+                memcpy(&sampling_selector_algorithm, data_shift, elem.record_length);
+
+                sampling_selector_algorithm = fast_ntoh(sampling_selector_algorithm);
+
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Decoded sampling selector algorithm " << sampling_selector_algorithm;
+                }
+            } else {
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG
+                           << "Unexpected size for IPFIX_SAMPLING_SELECTOR_ALGORITM: " << elem.record_length;
+                }
+
                 ipfix_too_large_field++;
 
                 // We're OK to continue process, we should not stop it
@@ -754,8 +1061,15 @@ bool ipfix_options_flowset_to_store(const uint8_t* pkt,
 
                 ipfix_active_flow_timeout_received++;
                 device_timeouts.active_timeout = active_timeout;
-                // logger << log4cpp::Priority::DEBUG << "Got active timeout: " << active_timeout << " seconds";
+                
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Got active timeout: " << active_timeout << " seconds";
+                }
             } else {
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Unexpected size for IPFIX_ACTIVE_TIMEOUT: " << elem.record_length;
+                }
+
                 ipfix_too_large_field++;
             }
 
@@ -769,8 +1083,15 @@ bool ipfix_options_flowset_to_store(const uint8_t* pkt,
 
                 ipfix_inactive_flow_timeout_received++;
                 device_timeouts.inactive_timeout = inactive_timeout;
-                // logger << log4cpp::Priority::DEBUG << "Got inactive timeout: " << inactive_timeout << " seconds";
+                
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Got inactive timeout: " << inactive_timeout << " seconds";
+                }
             } else {
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Unexpected size for IPFIX_INACTIVE_TIMEOUT: " << elem.record_length;
+                }
+
                 ipfix_too_large_field++;
             }
         }
@@ -778,6 +1099,35 @@ bool ipfix_options_flowset_to_store(const uint8_t* pkt,
         offset += elem.record_length;
     }
 
+    // Additional logic to deal with systematic count-based Sampling Selector on Nokia Nokia 7750 SR
+    // https://www.rfc-editor.org/rfc/rfc5476.html#section-6.5.2.1
+    // We check that sampler selected non zero number of packets as additional sanity check that we deal with this
+    // specific type of sampler and to avoid division by zero
+    if (sampling_selector_algorithm == IPFIX_SAMPLER_TYPE_SYSTEMATIC_COUNT_BASED_SAMPLING && sampling_packet_interval != 0) {
+        // We have seen following cases from Nokia:
+        // Packet space: 999 packet interval 1
+        // Packet space: 9999 packet interval 1
+        //
+        // Packet interval is the number of packets selected from whole packet space
+        //
+
+        //
+        // We never seen packet interval which is not set to 1 but I prefer to cover this case too
+        // For values of  packet interval after 1 we need to divide whole amount of observed packets
+        // (sampling_packet_space + sampling_packet_interval) by number of selected packets
+        //
+        uint32_t systematic_count_based_sampling_rate =
+            uint32_t(double(sampling_packet_space + sampling_packet_interval) / double(sampling_packet_interval));
+
+        // Update sampling rate
+        sampling_rate = fast_hton(systematic_count_based_sampling_rate);
+
+        if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+            logger << log4cpp::Priority::DEBUG << "Packet space: " << sampling_packet_space << " packet interval "
+                   << sampling_packet_interval << " sampling " << systematic_count_based_sampling_rate;
+        }
+    }
+
     update_ipfix_sampling_rate(sampling_rate, client_addres_in_string_format);
 
     // Update flow timeouts in our store
@@ -848,28 +1198,61 @@ bool ipfix_flowset_to_store(const uint8_t* pkt,
 
             if (*field_length_ptr == 0) {
                 logger << log4cpp::Priority::ERROR << "Zero length variable fields are not supported";
+                ipfix_inline_encoding_error++;
                 return false;
             }
 
             if (*field_length_ptr == 255) {
-                logger << log4cpp::Priority::ERROR << "Packet headers longer then 255 bytes are not supported, please reduce header length on device";
-                return false;
-            }
+                // 255 is special and it means that packet length is encoded in two following bytes
+                // Juniper PTX routers use this encoding even in case when packet length does not exceed 255 bytes
 
-            // Override field length with length extracted from leading byte
-            record_length = *field_length_ptr + sizeof(uint8_t);
+                // RFC reference https://datatracker.ietf.org/doc/html/rfc7011#page-37
+                // In this case, the first octet of the
+                // Length field MUST be 255, and the length is carried in the second and
+                // third octets, as shown in Figure S.
+
+                // Read 2 byte length by skipping placeholder byte with 255
+                const uint16_t* two_byte_field_length_ptr = (const uint16_t*)(pkt + offset + sizeof(uint8_t));
+
+                if (logger.getPriority() == log4cpp::Priority::DEBUG) {
+                    logger << log4cpp::Priority::DEBUG << "Two byte variable length encoding detected. Retrieved packet length: "
+                        << fast_ntoh(*two_byte_field_length_ptr);
+                }
+
+                // Pass variable payload length 
+                flow_meta.variable_field_length = fast_ntoh(*two_byte_field_length_ptr);
+
+                // Override field length with length extracted from two bytes + length of placeholder byte itself
+                record_length = flow_meta.variable_field_length + sizeof(uint8_t) + sizeof(uint16_t);;
+
+                // Pass variable payload length
+                flow_meta.variable_field_length = fast_ntoh(*two_byte_field_length_ptr);
+
+                // Specify length encoding type as it's required for payload retrieval process
+                flow_meta.variable_field_length_encoding = variable_length_encoding_t::two_byte;
+            } else {
+                // Pass variable payload length
+                flow_meta.variable_field_length = *field_length_ptr;
+
+                // Override field length with length extracted from leading byte
+                record_length = flow_meta.variable_field_length + sizeof(uint8_t);
+
+                // Specify length encoding type as it's required for payload retrieval process
+                flow_meta.variable_field_length_encoding = variable_length_encoding_t::single_byte;
+            }
         }
 
         // We do not need this check when we have only fixed length fields in template
         // but this function is versatile and must handle all cases.
         if (offset + record_length > flowset_maximum_length) {
-            logger << log4cpp::Priority::ERROR << "Attempt to read data after end of flowset";
+            logger << log4cpp::Priority::ERROR << "Attempt to read data after end of flowset. Offset: " << offset
+                << " record length: " << record_length << " flowset_maximum_length: " << flowset_maximum_length;
             return false;
         }
 
         bool ipfix_record_to_flow_result = ipfix_record_to_flow(record_type, record_length, pkt + offset, packet, flow_meta);
 
-        // In case of serious errors we stop loop completely:w
+        // In case of serious errors we stop loop completely
         if (!ipfix_record_to_flow_result) {
             return false;
         }
@@ -883,38 +1266,38 @@ bool ipfix_flowset_to_store(const uint8_t* pkt,
         override_packet_fields_from_nested_packet(packet, flow_meta.nested_packet);
     }
 
-    bool netflow_mark_zero_next_hop_and_zero_output_as_dropped = false;
 
-    if (netflow_mark_zero_next_hop_and_zero_output_as_dropped) {
+    if (false) {
         //
         // For Juniper routers we need fancy logic to mark packets as dropped as it does not use RFC compliant IPFIX field for it
         //
 
         //
-        // The only reliable information we have from Juniper documentation is about Netflow v9 
+        // The only reliable information we have from Juniper documentation is about Netflow v9
         // https://apps.juniper.net/feature-explorer/feature-info.html?fKey=7679&fn=Enhancements%20to%20inline%20flow%20monitoring
-        // and we have no idea how it behaves in IPFIX mode. 
-        // 
-        // I think previously we had Juniper routers which set output interface to zero and both bgp_next_hop_ipv4 and ip_next_hop_ipv4 to zero values to report
-        // dropped and we checked only bgp_next_hop_ipv4 to identify dropped traffic. It worked well enough until we got flows explained below where 
-        // bgp_next_hop_ipv4 is not 0.0.0.0 but ip_next_hop_ipv4 and output interface were set to zeroes.
+        // and we have no idea how it behaves in IPFIX mode.
         //
-        // In May 2023 got dumps in Google drive "MX10003 and MX 480 dropped traffic" which confirms that Juniper MX 10003 / MX480 with JUNOS 20.4R3-S4.8 encode
-        // it using zero output interface and zero ip_next_hop_ipv4. In same time these dumps have bgp_next_hop_ipv4 set to real non zero value of next router.
-        // To address this issue we added alternative section to check for zeroe 
+        // I think previously we had Juniper routers which set output interface to zero and both bgp_next_hop_ipv4 and
+        // ip_next_hop_ipv4 to zero values to report dropped and we checked only bgp_next_hop_ipv4 to identify dropped
+        // traffic. It worked well enough until we got flows explained below where bgp_next_hop_ipv4 is not 0.0.0.0 but
+        // ip_next_hop_ipv4 and output interface were set to zeroes.
+        //
+        // In May 2023 got dumps in Google drive "MX10003 and MX 480 dropped traffic" which confirms that Juniper MX
+        // 10003 / MX480 with JUNOS 20.4R3-S4.8 encode it using zero output interface and zero ip_next_hop_ipv4. In same
+        // time these dumps have bgp_next_hop_ipv4 set to real non zero value of next router. To address this issue we
+        // added alternative section to check for zeroe
         //
         // I posted question on LinkedIN: https://www.linkedin.com/feed/update/urn:li:activity:7062447441895141376/
         //
-            
+
         // We will apply it only if we have no forwarding_status in packet
         if (!flow_meta.received_forwarding_status) {
             // We need to confirm that TWO rules are TRUE:
             // - Output interface is 0
             // - Next hop for IPv4 is set and set to 0 OR next hop for IPv6 set and set to zero
             if (packet.output_interface == 0 &&
-                (
-                 (flow_meta.bgp_next_hop_ipv4_set && flow_meta.bgp_next_hop_ipv4 == 0) ||
-                 (flow_meta.ip_next_hop_ipv4_set  && flow_meta.ip_next_hop_ipv4 == 0 ) ||
+                ((flow_meta.bgp_next_hop_ipv4_set && flow_meta.bgp_next_hop_ipv4 == 0) ||
+                 (flow_meta.ip_next_hop_ipv4_set && flow_meta.ip_next_hop_ipv4 == 0) ||
                  (is_zero_ipv6_address(flow_meta.bgp_next_hop_ipv6) && flow_meta.bgp_next_hop_ipv6_set))) {
 
                 packet.forwarding_status = forwarding_status_t::dropped;
@@ -979,7 +1362,6 @@ bool ipfix_flowset_to_store(const uint8_t* pkt,
         packet.destination_port = 0;
     }
 
-
     // pass data to FastNetMon
     netflow_process_func_ptr(packet);
     return true;
@@ -987,7 +1369,19 @@ bool ipfix_flowset_to_store(const uint8_t* pkt,
 
 // That's kind of histogram emulation
 void increment_duration_counters_ipfix(int64_t duration) {
-    if (duration <= 15) {
+    if (duration == 0) {
+        ipfix_duration_0_seconds++;
+    } else if (duration <= 1) {
+        ipfix_duration_less_1_seconds++;
+    } else if (duration <= 2) {
+        ipfix_duration_less_2_seconds++;
+    } else if (duration <= 3) {
+        ipfix_duration_less_3_seconds++;
+    } else if (duration <= 5) {
+        ipfix_duration_less_5_seconds++;
+    } else if (duration <= 10) {
+        ipfix_duration_less_10_seconds++;
+    } else if (duration <= 15) {
         ipfix_duration_less_15_seconds++;
     } else if (duration <= 30) {
         ipfix_duration_less_30_seconds++;
@@ -1004,7 +1398,6 @@ void increment_duration_counters_ipfix(int64_t duration) {
     return;
 }
 
-
 bool process_ipfix_data(const uint8_t* pkt,
                         size_t flowset_length,
                         const ipfix_header_t* ipfix_header,
@@ -1025,8 +1418,8 @@ bool process_ipfix_data(const uint8_t* pkt,
 
     uint32_t flowset_id = ntohs(flowset_header->header.flowset_id);
 
-    const template_t* field_template =
-        peer_find_template(global_ipfix_templates, source_id, flowset_id, client_addres_in_string_format);
+    const template_t* field_template = peer_find_template(global_ipfix_templates, global_ipfix_templates_mutex,
+                                                          source_id, flowset_id, client_addres_in_string_format);
 
     if (field_template == NULL) {
         ipfix_packets_with_unknown_templates++;
@@ -1056,8 +1449,9 @@ bool process_ipfix_data(const uint8_t* pkt,
                 logger << log4cpp::Priority::DEBUG << "IPFIX variable field element was used";
             }
 
-            // This implementation is rather limited as we read only single flowset here
+            // TODO: This implementation is rather limited as we read only single flowset here
             // In many cases we use this stuff for IPFIX Inline monitoring and it uses just single flowset but it may change in future
+            // TODO: Juniper PTX uses multiple flowsets per packet
             ipfix_flowset_to_store(pkt + sizeof(ipfix_data_flowset_header_t), ipfix_header, current_flowset_length_no_header,
                                    field_template, client_ipv4_address, client_addres_in_string_format);
         } else {
@@ -1069,6 +1463,19 @@ bool process_ipfix_data(const uint8_t* pkt,
 
             uint32_t number_flowsets = (flowset_length - offset) / field_template->total_length;
 
+            // We need to calculate padding value
+            // IPFIX RFC explains it following way:
+            // https://datatracker.ietf.org/doc/html/rfc7011?ref=pavel.network#section-3.3.1
+            uint32_t flowset_padding = (flowset_length - offset) % field_template->total_length;
+
+            // Very likely data will be aligned by 4 byte boundaries and will have padding 1, 2, 3 bytes
+            // To be on safe side we assume that padding may be up to 7 bytes to achieve 8 byte boundaries
+            // All other values may be sign of some kind of issues. For example, it may be template conflicts
+            // https://pavel.network/its-just-wrong-to-update-ipfix-templates/
+            if (flowset_padding > 7) {
+                ipfix_flowsets_with_anomaly_padding++;
+            }
+
             if (number_flowsets > 0x4000) {
                 logger << log4cpp::Priority::ERROR << "Very high number of IPFIX data flowsets " << number_flowsets
                        << " Agent: " << client_addres_in_string_format
@@ -1080,10 +1487,8 @@ bool process_ipfix_data(const uint8_t* pkt,
             if (number_flowsets == 0) {
                 logger << log4cpp::Priority::ERROR << "Unexpected zero number of flowsets "
                        << " agent: " << client_addres_in_string_format
-                       << " flowset template length: " << field_template->total_length 
-                       << " flowset length " << flowset_length
-                       << " source_id " << source_id
-                       << " flowset_id: " << flowset_id;
+                       << " flowset template length: " << field_template->total_length << " flowset length "
+                       << flowset_length << " source_id " << source_id << " flowset_id: " << flowset_id;
 
                 return false;
             }
@@ -1115,7 +1520,6 @@ bool process_ipfix_data(const uint8_t* pkt,
     return true;
 }
 
-
 // Process IPFIX packet
 bool process_ipfix_packet(const uint8_t* packet,
                           uint32_t udp_packet_length,
@@ -1124,7 +1528,7 @@ bool process_ipfix_packet(const uint8_t* packet,
     const ipfix_header_t* ipfix_header = (const ipfix_header_t*)packet;
 
     if (udp_packet_length < sizeof(ipfix_header_t)) {
-        logger << log4cpp::Priority::ERROR << "Packet is too short to accommodate IPFIX header " << udp_packet_length
+        logger << log4cpp::Priority::ERROR << "Packet is too short to accomodate IPFIX header " << udp_packet_length
                << " bytes which requires at least " << sizeof(ipfix_header_t) << " bytes";
         return false;
     }
@@ -1232,7 +1636,6 @@ bool process_ipfix_packet(const uint8_t* packet,
     return true;
 }
 
-
 void update_ipfix_sampling_rate(uint32_t sampling_rate, const std::string& client_addres_in_string_format) {
     if (sampling_rate == 0) {
         return;
@@ -1259,7 +1662,6 @@ void update_ipfix_sampling_rate(uint32_t sampling_rate, const std::string& clien
 
             logger << log4cpp::Priority::INFO << "Learnt new IPFIX sampling rate " << new_sampling_rate << " for "
                    << client_addres_in_string_format;
-
         } else {
             auto old_sampling_rate = known_sampling_rate->second;
 
@@ -1268,10 +1670,10 @@ void update_ipfix_sampling_rate(uint32_t sampling_rate, const std::string& clien
 
                 ipfix_sampling_rate_changes++;
 
-                logger << log4cpp::Priority::INFO << "Detected IPFIX sampling rate change from " << old_sampling_rate << " to "
-                       << new_sampling_rate << " for " << client_addres_in_string_format;
-
+                logger << log4cpp::Priority::INFO << "Detected IPFIX sampling rate change from " << old_sampling_rate
+                       << " to " << new_sampling_rate << " for " << client_addres_in_string_format;
             }
         }
     }
 }
+

src/netflow_plugin/netflow_collector.cpp

@@ -80,6 +80,10 @@ class device_timeouts_t {
     bool operator==(const device_timeouts_t& rhs) const = default;
 };
 
+// Variable encoding may be single or two byte and we need to distinguish them explicitly
+enum class variable_length_encoding_t { unknown, single_byte, two_byte };
+
+
 // IPFIX per device timeouts
 std::mutex ipfix_per_device_flow_timeouts_mutex;
 std::map<std::string, device_timeouts_t> ipfix_per_device_flow_timeouts;
@@ -153,6 +157,9 @@ uint64_t ipfix_total_flows         = 0;
 std::string ipfix_total_ipv4_flows_desc = "Total number of IPFIX IPv4 flows (multiple in each packet)";
 uint64_t ipfix_total_ipv4_flows         = 0;
 
+std::string ipfix_flowsets_with_anomaly_padding_desc = "IPFIX flowsets with anomaly padding more then 7 bytes";
+uint64_t ipfix_flowsets_with_anomaly_padding         = 0;
+
 std::string ipfix_active_flow_timeout_received_desc = "Total number of received active IPFIX flow timeouts";
 uint64_t ipfix_active_flow_timeout_received         = 0;
 
@@ -244,6 +251,24 @@ uint64_t netflow9_duration_less_180_seconds         = 0;
 std::string netflow9_duration_exceed_180_seconds_desc = "Netflow v9 flows with duration more then 180 seconds";
 uint64_t netflow9_duration_exceed_180_seconds         = 0;
 
+std::string ipfix_duration_0_seconds_desc = "IPFIX flows with duration 0 seconds";
+uint64_t ipfix_duration_0_seconds         = 0;
+
+std::string ipfix_duration_less_1_seconds_desc = "IPFIX flows with duration less then 1 seconds";
+uint64_t ipfix_duration_less_1_seconds         = 0;
+
+std::string ipfix_duration_less_2_seconds_desc = "IPFIX flows with duration less then 2 seconds";
+uint64_t ipfix_duration_less_2_seconds         = 0;
+
+std::string ipfix_duration_less_3_seconds_desc = "IPFIX flows with duration less then 3 seconds";
+uint64_t ipfix_duration_less_3_seconds         = 0;
+
+std::string ipfix_duration_less_5_seconds_desc = "IPFIX flows with duration less then 5 seconds";
+uint64_t ipfix_duration_less_5_seconds         = 0;
+
+std::string ipfix_duration_less_10_seconds_desc = "IPFIX flows with duration less then 10 seconds";
+uint64_t ipfix_duration_less_10_seconds         = 0;
+
 std::string ipfix_duration_less_15_seconds_desc = "IPFIX flows with duration less then 15 seconds";
 uint64_t ipfix_duration_less_15_seconds         = 0;
 
@@ -373,6 +398,9 @@ uint64_t netflow_v9_lite_headers         = 0;
 std::string ipfix_inline_headers_desc = "Total number of headers in IPFIX received";
 uint64_t ipfix_inline_headers         = 0;
 
+std::string ipfix_inline_encoding_error_desc = "IPFIX inline encoding issues";
+uint64_t  ipfix_inline_encoding_error         = 0;
+
 std::string ipfix_packets_with_padding_desc = "Total number of IPFIX packets with padding";
 uint64_t ipfix_packets_with_padding         = 0;
 
@@ -387,7 +415,10 @@ uint64_t flowsets_per_packet_maximum_number = 256;
 // TODO: add per source uniq templates support
 process_packet_pointer netflow_process_func_ptr = NULL;
 
+std::mutex global_netflow9_templates_mutex;
 std::map<std::string, std::map<uint32_t, template_t>> global_netflow9_templates;
+
+std::mutex global_ipfix_templates_mutex;
 std::map<std::string, std::map<uint32_t, template_t>> global_ipfix_templates;
 
 std::vector<system_counter_t> get_netflow_stats() {
@@ -620,6 +651,7 @@ std::string get_netflow_protocol_version_as_string(const netflow_protocol_versio
 /* Prototypes */
 void add_update_peer_template(const netflow_protocol_version_t& netflow_version,
                               std::map<std::string, std::map<uint32_t, template_t>>& table_for_add,
+                              std::mutex& table_for_add_mutex,
                               uint32_t source_id,
                               uint32_t template_id,
                               const std::string& client_addres_in_string_format,
@@ -679,6 +711,11 @@ class netflow_meta_info_t {
 
     // Cisco ASA flow identifier
     uint64_t flow_id = 0;
+
+    variable_length_encoding_t variable_field_length_encoding = variable_length_encoding_t::unknown;
+
+    // Store variable field length here to avoid repeating parsing
+    uint16_t variable_field_length = 0;
 };
 
 int nf9_rec_to_flow(uint32_t record_type,
@@ -689,6 +726,7 @@ int nf9_rec_to_flow(uint32_t record_type,
                     netflow_meta_info_t& flow_meta);
 
 template_t* peer_find_template(std::map<std::string, std::map<uint32_t, template_t>>& table_for_lookup,
+                                      std::mutex& table_for_lookup_mutex,
                                       uint32_t source_id,
                                       uint32_t template_id,
                                       std::string client_addres_in_string_format) {
@@ -696,6 +734,8 @@ template_t* peer_find_template(std::map<std::string, std::map<uint32_t, template
     // We use source_id for distinguish multiple netflow agents with same IP
     std::string key = client_addres_in_string_format + "_" + std::to_string(source_id);
 
+    std::lock_guard<std::mutex> lock(table_for_lookup_mutex);
+
     auto itr = table_for_lookup.find(key);
 
     if (itr == table_for_lookup.end()) {
@@ -744,6 +784,7 @@ void override_packet_fields_from_nested_packet(simple_packet_t& packet, const si
 void add_update_peer_template(
                               const netflow_protocol_version_t& netflow_protocol_version,
                               std::map<std::string, std::map<uint32_t, template_t>>& table_for_add,
+                              std::mutex& table_for_add_mutex,
                               uint32_t source_id,
                               uint32_t template_id,
                               const std::string& client_address_in_string_format,
@@ -759,6 +800,9 @@ void add_update_peer_template(
             << " source id: " << source_id;  
     }
 
+    // We need to put lock on it
+    std::lock_guard<std::mutex> lock(table_for_add_mutex);
+
     auto itr = table_for_add.find(key);
 
     if (itr == table_for_add.end()) {

src/netflow_plugin/netflow_v9_collector.cpp

@@ -98,7 +98,7 @@ bool process_netflow_v9_options_template(const uint8_t* pkt, size_t flowset_leng
     bool updated                   = false;
     bool updated_existing_template = false;
 
-    add_update_peer_template(netflow_protocol_version_t::netflow_v9, global_netflow9_templates, source_id, template_id, client_addres_in_string_format,
+    add_update_peer_template(netflow_protocol_version_t::netflow_v9, global_netflow9_templates, global_netflow9_templates_mutex,  source_id, template_id, client_addres_in_string_format,
                              field_template, updated, updated_existing_template);
 
     if (updated_existing_template) {
@@ -190,7 +190,7 @@ bool process_netflow_v9_template(const uint8_t* pkt,
         bool updated                   = false;
         bool updated_existing_template = false;
 
-        add_update_peer_template(netflow_protocol_version_t::netflow_v9, global_netflow9_templates, source_id, template_id, client_addres_in_string_format,
+        add_update_peer_template(netflow_protocol_version_t::netflow_v9, global_netflow9_templates, global_netflow9_templates_mutex,  source_id, template_id, client_addres_in_string_format,
                                  field_template, updated, updated_existing_template);
 
         // If we have any changes for this template, let's flush them to disk
@@ -1366,7 +1366,7 @@ bool process_netflow_v9_data(const uint8_t* pkt,
 
     // We should find template here
     const template_t* field_template =
-        peer_find_template(global_netflow9_templates, source_id, flowset_id, client_addres_in_string_format);
+        peer_find_template(global_netflow9_templates, global_netflow9_templates_mutex, source_id, flowset_id, client_addres_in_string_format);
 
     if (field_template == NULL) {
         netflow9_packets_with_unknown_templates++;

src/network_data_structures.hpp

@@ -33,7 +33,7 @@ namespace network_data_stuctures {
 
 // We are using this structure as pretty interface for IPv4 address bytes in host byte order (little  endian)
 class __attribute__((__packed__)) ipv4_octets_form_little_endian_t {
-public:
+    public:
     uint8_t fourth = 0;
     uint8_t third  = 0;
     uint8_t second = 0;
@@ -130,7 +130,7 @@ class __attribute__((__packed__)) mpls_label_t {
     public:
     uint32_t label : 20 = 0, qos : 3 = 0, bottom_of_stack : 1 = 0, ttl : 8 = 0;
 
-    std::string print() {
+    std::string print() const {
         std::stringstream buffer;
 
         buffer << "label: " << uint32_t(label) << " "
@@ -145,7 +145,7 @@ class __attribute__((__packed__)) mpls_label_t {
 static_assert(sizeof(mpls_label_t) == 4, "Bad size for mpls_label_t");
 
 // In this class we keep vlan id, priority and cfi
-class __attribute__((__packed__))  ethernet_vlan_metadata_t {
+class __attribute__((__packed__)) ethernet_vlan_metadata_t {
     public:
     uint16_t vlan_id : 12, cfi : 1, priority : 3;
 };
@@ -157,12 +157,11 @@ static_assert(sizeof(ethernet_vlan_metadata_t) == 2, "Bad size for ethernet_vlan
 class __attribute__((__packed__)) ethernet_vlan_header_t {
     // We must not access these fields directly as it requires explicit byte order conversion
     private:
-        uint16_t vlan_metadata_as_integer = 0;
-        uint16_t ethertype = 0;
+    uint16_t vlan_metadata_as_integer = 0;
+    uint16_t ethertype                = 0;
 
     // We can access data in packet only using special methods which can do all required format conversions
     public:
-
     // Returns ethertype in host byte order
     uint16_t get_ethertype_host_byte_order() const {
         return fast_ntoh(ethertype);
@@ -230,7 +229,7 @@ class __attribute__((__packed__)) ethernet_header_t {
 
     public:
     // Returns ethertype in host byte order
-    uint16_t get_ethertype_host_byte_order() {
+    uint16_t get_ethertype_host_byte_order() const {
         return fast_ntoh(ethertype);
     }
 
@@ -270,7 +269,6 @@ class __attribute__((__packed__)) arp_header_t {
     uint32_t target_protocol_address = 0;
 
     public:
-
     uint8_t get_hardware_address_length() const {
         return hardware_address_length;
     }
@@ -310,9 +308,11 @@ class __attribute__((__packed__)) arp_header_t {
                << "protocol_address_length: " << uint32_t(get_protocol_address_length()) << " "
                << "operation: " << get_operation_host_byte_order() << " "
                << "sender_hardware_address: " << convert_mac_to_string(sender_hardware_address) << " "
-               << "sender_protocol_address: " << convert_ip_as_big_endian_to_string(get_sender_protocol_address_network_byte_order()) << " "
+               << "sender_protocol_address: "
+               << convert_ip_as_big_endian_to_string(get_sender_protocol_address_network_byte_order()) << " "
                << "target_hardware_address: " << convert_mac_to_string(target_hardware_address) << " "
-               << "target_protocol_address: " << convert_ip_as_big_endian_to_string(get_target_protocol_address_network_byte_order());
+               << "target_protocol_address: "
+               << convert_ip_as_big_endian_to_string(get_target_protocol_address_network_byte_order());
 
         return buffer.str();
     }
@@ -364,15 +364,15 @@ class __attribute__((__packed__)) udp_header_t {
     uint16_t get_source_port_host_byte_order() const {
         return fast_ntoh(source_port);
     }
-    
+
     uint16_t get_destination_port_host_byte_order() const {
         return fast_ntoh(destination_port);
     }
-    
+
     uint16_t get_length_host_byte_order() const {
         return fast_ntoh(length);
     }
-    
+
     uint16_t get_checksum_host_byte_order() const {
         return fast_ntoh(checksum);
     }
@@ -394,14 +394,13 @@ static_assert(sizeof(udp_header_t) == 8, "Bad size for udp_header_t");
 // https://datatracker.ietf.org/doc/html/rfc2784
 class __attribute__((__packed__)) gre_header_t {
     // We must not access these fields directly as they may need conversion
-   private:
+    private:
     // TODO: we have no pcaps where these fields are no zeros and we did not test this case
     // For some reasons PVS thinks that we did not initialised all members. I think they're not that great with bitfields
     uint16_t checksum : 1 = 0, reserved : 12 = 0, version : 3 = 0; //-V730
     uint16_t protocol_type = 0;
 
     public:
-
     uint16_t get_protocol_type_host_byte_order() const {
         return fast_ntoh(protocol_type);
     }
@@ -413,7 +412,7 @@ class __attribute__((__packed__)) gre_header_t {
     uint16_t get_checksum() const {
         return checksum;
     }
-    
+
     uint16_t get_version() const {
         return version;
     }
@@ -472,7 +471,6 @@ class __attribute__((__packed__)) cropped_tcp_header_only_ports_t {
     uint16_t destination_port = 0;
 
     public:
-
     uint16_t get_source_port_host_byte_order() const {
         return fast_ntoh(source_port);
     }
@@ -491,18 +489,16 @@ class __attribute__((__packed__)) tcp_header_t {
     uint16_t destination_port = 0;
     uint32_t sequence_number  = 0;
     uint32_t ack_number       = 0;
-    
+
     // Flags here encoded as tcp_flags_t
     uint16_t data_offset_and_flags_as_integer = 0;
 
     private:
-
     uint16_t window_size = 0;
     uint16_t checksum    = 0;
     uint16_t urgent      = 0;
 
     public:
-
     uint16_t get_source_port_host_byte_order() const {
         return fast_ntoh(source_port);
     }
@@ -510,23 +506,23 @@ class __attribute__((__packed__)) tcp_header_t {
     uint16_t get_destination_port_host_byte_order() const {
         return fast_ntoh(destination_port);
     }
-    
+
     uint32_t get_sequence_number_host_byte_order() const {
         return fast_ntoh(sequence_number);
     }
-    
+
     uint32_t get_ack_number_host_byte_order() const {
         return fast_ntoh(ack_number);
     }
-    
+
     uint16_t get_window_size_host_byte_order() const {
         return fast_ntoh(window_size);
     }
-    
+
     uint16_t get_checksum_host_byte_order() const {
         return fast_ntoh(checksum);
     }
-    
+
     uint16_t get_urgent_host_byte_order() const {
         return fast_ntoh(urgent);
     }
@@ -586,7 +582,7 @@ class __attribute__((__packed__)) tcp_header_t {
         return tcp_flags->ack == 1;
     }
 
-    bool get_urg ()const  {
+    bool get_urg() const {
         uint16_t data_offset_and_flags_as_integer_litle_endian = fast_ntoh(data_offset_and_flags_as_integer);
 
         tcp_flags_t* tcp_flags = (tcp_flags_t*)&data_offset_and_flags_as_integer_litle_endian;
@@ -594,7 +590,7 @@ class __attribute__((__packed__)) tcp_header_t {
         return tcp_flags->urg == 1;
     }
 
-    bool get_ece ()  const {
+    bool get_ece() const {
         uint16_t data_offset_and_flags_as_integer_litle_endian = fast_ntoh(data_offset_and_flags_as_integer);
 
         tcp_flags_t* tcp_flags = (tcp_flags_t*)&data_offset_and_flags_as_integer_litle_endian;
@@ -602,7 +598,7 @@ class __attribute__((__packed__)) tcp_header_t {
         return tcp_flags->ece == 1;
     }
 
-    bool get_cwr () const {
+    bool get_cwr() const {
         uint16_t data_offset_and_flags_as_integer_litle_endian = fast_ntoh(data_offset_and_flags_as_integer);
 
         tcp_flags_t* tcp_flags = (tcp_flags_t*)&data_offset_and_flags_as_integer_litle_endian;
@@ -618,7 +614,7 @@ class __attribute__((__packed__)) tcp_header_t {
         return tcp_flags->ns == 1;
     }
 
-    uint8_t get_reserved() const{
+    uint8_t get_reserved() const {
         uint16_t data_offset_and_flags_as_integer_litle_endian = fast_ntoh(data_offset_and_flags_as_integer);
 
         tcp_flags_t* tcp_flags = (tcp_flags_t*)&data_offset_and_flags_as_integer_litle_endian;
@@ -696,9 +692,9 @@ inline std::string convert_ipv6_in_byte_array_to_string(const uint8_t (&v6_addre
 */
 
 class __attribute__((__packed__)) ipv6_fragment_header_flags {
-     public:
-        // fragment_offset is a number of 8byte chunk
-        uint16_t more_fragments : 1, reserved2 : 2, fragment_offset : 13;
+    public:
+    // fragment_offset is a number of 8byte chunk
+    uint16_t more_fragments : 1, reserved2 : 2, fragment_offset : 13;
 };
 
 static_assert(sizeof(ipv6_fragment_header_flags) == 2, "Bad size for ipv6_header_flags_t");
@@ -717,7 +713,6 @@ class __attribute__((__packed__)) ipv6_extension_header_fragment_t {
     uint32_t identification = 0;
 
     public:
-
     uint16_t get_more_fragments() const {
         uint16_t flags_little_endian = fast_ntoh(fragmentation_and_flags_as_integer);
 
@@ -787,7 +782,7 @@ static_assert(sizeof(ipv6_header_flags_t) == 4, "Bad size for ipv6_header_flags_
 
 class __attribute__((__packed__)) ipv6_header_t {
     // We must not access these fields directly as they need decoding
-    private:    
+    private:
     // Multiple flags carried in ipv6_header_flags_t
     uint32_t version_and_traffic_class_as_integer = 0;
 
@@ -801,13 +796,13 @@ class __attribute__((__packed__)) ipv6_header_t {
 
     uint32_t get_flow_label() const {
         uint32_t version_and_traffic_class_little_endian = fast_ntoh(version_and_traffic_class_as_integer);
-    
+
         ipv6_header_flags_t* header_flags = (ipv6_header_flags_t*)&version_and_traffic_class_little_endian;
 
         return header_flags->flow_label;
     }
 
-    uint32_t get_traffic_class() const { 
+    uint32_t get_traffic_class() const {
         uint32_t version_and_traffic_class_little_endian = fast_ntoh(version_and_traffic_class_as_integer);
 
         ipv6_header_flags_t* header_flags = (ipv6_header_flags_t*)&version_and_traffic_class_little_endian;
@@ -835,7 +830,7 @@ class __attribute__((__packed__)) ipv6_header_t {
         return hop_limit;
     }
 
-    std::string print() const  {
+    std::string print() const {
         std::stringstream buffer;
 
         buffer << "version: " << get_version() << " "
@@ -856,8 +851,8 @@ static_assert(sizeof(ipv6_header_t) == 40, "Bad size for ipv6_header_t");
 // It's class for fragmentation flag representation. It's pretty useful in some cases
 class __attribute__((__packed__)) ipv4_header_fragmentation_flags_t {
     public:
-        // The offset value is the number of 8 byte blocks of data
-        uint16_t fragment_offset : 13, more_fragments_flag : 1, dont_fragment_flag : 1, reserved_flag : 1;
+    // The offset value is the number of 8 byte blocks of data
+    uint16_t fragment_offset : 13, more_fragments_flag : 1, dont_fragment_flag : 1, reserved_flag : 1;
 
     std::string print() {
 
@@ -876,7 +871,6 @@ static_assert(sizeof(ipv4_header_fragmentation_flags_t) == 2, "Bad size for ipv4
 class __attribute__((__packed__)) ipv4_header_t {
     // We must not access these fields directly as they need conversion
     private:
-
     uint8_t ihl : 4 = 0, version : 4 = 0;
     uint8_t ecn : 2 = 0, dscp : 6 = 0;
 
@@ -887,16 +881,15 @@ class __attribute__((__packed__)) ipv4_header_t {
     // There we have plenty of fragmentation specific fields encoded in ipv4_header_fragmentation_flags_t
     uint16_t fragmentation_details_as_integer = 0;
 
-    uint8_t ttl             = 0;
-    uint8_t protocol        = 0;
+    uint8_t ttl      = 0;
+    uint8_t protocol = 0;
 
-    uint16_t checksum       = 0;
+    uint16_t checksum = 0;
 
     uint32_t source_ip      = 0;
     uint32_t destination_ip = 0;
 
     public:
-
     ipv4_header_t()
     : ihl(0), version(0), ecn(0), dscp(0), total_length(0), identification(0), fragmentation_details_as_integer(0),
       ttl(0), protocol(0), checksum(0), source_ip(0), destination_ip(0) {
@@ -910,7 +903,7 @@ class __attribute__((__packed__)) ipv4_header_t {
         return fast_ntoh(total_length);
     }
 
-    bool is_fragmented() {
+    bool is_fragmented() const {
         if (this->get_more_fragments_flag()) {
             return true;
         }
@@ -923,20 +916,20 @@ class __attribute__((__packed__)) ipv4_header_t {
     }
 
     uint8_t get_ihl() const {
-    	return ihl;
-    }	
-     
-    uint8_t get_version() const {
-    	return version; 
+        return ihl;
     }
-    
+
+    uint8_t get_version() const {
+        return version;
+    }
+
     uint8_t get_ecn() const {
-    	return ecn;
-    } 
-    
+        return ecn;
+    }
+
     uint8_t get_dscp() const {
-    	return dscp;
-    } 
+        return dscp;
+    }
 
     uint16_t get_fragmentation_details_host_byte_order() {
         return fast_ntoh(fragmentation_details_as_integer);
@@ -965,10 +958,10 @@ class __attribute__((__packed__)) ipv4_header_t {
         uint16_t fragmenation_details_little_endian = fast_ntoh(fragmentation_details_as_integer);
 
         ipv4_header_fragmentation_flags_t* fragmentation_flags = (ipv4_header_fragmentation_flags_t*)&fragmenation_details_little_endian;
-    
+
         return fragmentation_flags->more_fragments_flag == 1;
     }
-     
+
     bool get_dont_fragment_flag() const {
         uint16_t fragmenation_details_little_endian = fast_ntoh(fragmentation_details_as_integer);
 
@@ -1009,7 +1002,7 @@ class __attribute__((__packed__)) ipv4_header_t {
         fragmentation_details_as_integer = fast_hton(fragmenation_details_little_endian);
     }
 
-    // Value is a number of 8 byte blocks 
+    // Value is a number of 8 byte blocks
     void set_fragment_offset_8byte_chunks(uint16_t value) {
         uint16_t fragmenation_details_little_endian = fast_ntoh(fragmentation_details_as_integer);
 
@@ -1032,7 +1025,7 @@ class __attribute__((__packed__)) ipv4_header_t {
     uint32_t get_source_ip_host_byte_order() const {
         return fast_ntoh(source_ip);
     }
-    
+
     uint32_t get_destination_ip_host_byte_order() const {
         return fast_ntoh(destination_ip);
     }
@@ -1082,7 +1075,8 @@ enum class parser_code_t {
     no_ipv6_support,
     no_ipv6_options_support,
     unknown_ethertype,
-    arp
+    arp,
+    too_many_nested_vlans,
 };
 
 std::string parser_code_to_string(parser_code_t code);

src/simple_packet_parser_ng.cpp

@@ -5,8 +5,8 @@
 #include "fast_library.hpp"
 
 #include <algorithm>
-#include <iterator>
 #include <cstring>
+#include <iterator>
 
 using namespace network_data_stuctures;
 
@@ -14,6 +14,9 @@ using namespace network_data_stuctures;
 // TODO: it's not working code yet
 bool decode_mpls = false;
 
+// We can strip only 3 nested vlans
+const uint32_t maximum_vlans_to_strip = 3;
+
 // Our own native function to convert wire packet into simple_packet_t
 parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
                                                         int length_before_sampling,
@@ -32,30 +35,44 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
         return parser_code_t::memory_violation;
     }
 
-    ethernet_header_t* ethernet_header = (ethernet_header_t*)local_pointer;
+    const ethernet_header_t* ethernet_header = (const ethernet_header_t*)local_pointer;
 
     // Copy Ethernet MAC addresses to packet structure using native C++ approach to avoid touching memory with memcpy
     std::copy(std::begin(ethernet_header->source_mac), std::end(ethernet_header->source_mac), std::begin(packet.source_mac));
 
-    std::copy(std::begin(ethernet_header->destination_mac), std::end(ethernet_header->destination_mac), std::begin(packet.destination_mac));
+    std::copy(std::begin(ethernet_header->destination_mac), std::end(ethernet_header->destination_mac),
+              std::begin(packet.destination_mac));
 
     local_pointer += sizeof(ethernet_header_t);
 
     // Copy ethertype as we may need to change it below
     uint16_t ethertype = ethernet_header->get_ethertype_host_byte_order();
 
-    if (ethertype == IanaEthertypeVLAN) {
-        // Return error if it shorter then vlan header
+    uint32_t number_of_stripped_vlans = 0;
+
+    // This loop will not start if ethertype is not VLAN
+    while (ethertype == IanaEthertypeVLAN) { 
+        // Return error if it's shorter than vlan header
         if (local_pointer + sizeof(ethernet_vlan_header_t) > end_pointer) {
             return parser_code_t::memory_violation;
         }
 
-        ethernet_vlan_header_t* ethernet_vlan_header = (ethernet_vlan_header_t*)local_pointer;
-
-        packet.vlan = ethernet_vlan_header->get_vlan_id_host_byte_order();
+        const ethernet_vlan_header_t* ethernet_vlan_header = (const ethernet_vlan_header_t*)local_pointer;
 
+        // We've agreed that this field keeps only outermost vlan
+        if (number_of_stripped_vlans == 0) {
+            packet.vlan = ethernet_vlan_header->get_vlan_id_host_byte_order();
+        }
+    
         local_pointer += sizeof(ethernet_vlan_header_t);
 
+        number_of_stripped_vlans++;
+
+        // We need to limit it to avoid possibility of attack which uses too many vlans tags to overload our parser
+        if (number_of_stripped_vlans > maximum_vlans_to_strip) {
+            return parser_code_t::too_many_nested_vlans;
+        }
+
         // Change ethertype to vlan's ethertype
         ethertype = ethernet_vlan_header->get_ethertype_host_byte_order();
     }
@@ -67,7 +84,7 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
                 return parser_code_t::memory_violation;
             }
 
-            mpls_label_t* mpls_label_header = (mpls_label_t*)local_pointer;
+            const mpls_label_t* mpls_label_header = (const mpls_label_t*)local_pointer;
 
             std::cout << "MPLS header: " << mpls_label_header->print() << std::endl;
 
@@ -90,7 +107,7 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
             return parser_code_t::memory_violation;
         }
 
-        ipv4_header_t* ipv4_header = (ipv4_header_t*)local_pointer;
+        const ipv4_header_t* ipv4_header = (const ipv4_header_t*)local_pointer;
 
         // Use network representation of addresses
         packet.src_ip = ipv4_header->get_source_ip_network_byte_order();
@@ -114,8 +131,8 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
         packet.ip_fragment_offset = ipv4_header->get_fragment_offset_bytes();
 
         // We keep these variables to maintain backward compatibility with parse_raw_packet_to_simple_packet_full()
-        packet.captured_payload_length      = length_before_sampling;
-        packet.payload_full_length = length_before_sampling;
+        packet.captured_payload_length = length_before_sampling;
+        packet.payload_full_length     = length_before_sampling;
 
         // Pointer to payload
         packet.payload_pointer = (void*)pointer;
@@ -145,7 +162,7 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
             return parser_code_t::memory_violation;
         }
 
-        ipv6_header_t* ipv6_header = (ipv6_header_t*)local_pointer;
+        const ipv6_header_t* ipv6_header = (const ipv6_header_t*)local_pointer;
 
         // TODO: we may use std::copy for it to avoid touching memory with memcpy
         memcpy(&packet.src_ipv6, ipv6_header->source_address, sizeof(packet.src_ipv6));
@@ -159,8 +176,8 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
         packet.ip_length = ipv6_header->get_payload_length();
 
         // We keep these variables to maintain backward compatibility with parse_raw_packet_to_simple_packet_full()
-        packet.captured_payload_length      = length_before_sampling;
-        packet.payload_full_length = length_before_sampling;
+        packet.captured_payload_length = length_before_sampling;
+        packet.payload_full_length     = length_before_sampling;
 
         // Pointer to payload
         packet.payload_pointer = (void*)pointer;
@@ -196,7 +213,7 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
 
             // We decided to parse only fragmentation header option as only this field may be found in the Wild
             if (protocol == IpProtocolNumberIPV6_FRAG) {
-                ipv6_extension_header_fragment_t* ipv6_extension_header_fragment = (ipv6_extension_header_fragment_t*)local_pointer;
+                const ipv6_extension_header_fragment_t* ipv6_extension_header_fragment = (const ipv6_extension_header_fragment_t*)local_pointer;
 
                 // If we received this header then we assume that packet was fragmented
                 packet.ip_fragmented = true;
@@ -224,18 +241,18 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
     if (protocol == IpProtocolNumberTCP) {
         if (local_pointer + sizeof(tcp_header_t) > end_pointer) {
             // We observed that Huawei routers may send only first 52 bytes of header in sFlow mode
-            // and it's not enough to accommodate whole TCP header which has length of 20 bytes and we can observe only first 14 bytes 
-            // and it happened in case of vlan presence
+            // and it's not enough to accommodate whole TCP header which has length of 20 bytes and we can observe only
+            // first 14 bytes and it happened in case of vlan presence
 
             // To offer better experience we will try retrieving only ports from TCP header as they're located in the beginning of packet
             if (local_pointer + sizeof(cropped_tcp_header_only_ports_t) > end_pointer) {
-                // Sadly we cannot even retrieve port numbers and we have to discard this packet 
+                // Sadly we cannot even retrieve port numbers and we have to discard this packet
                 // Idea of reporting this packet as TCP protocol without ports information is not reasonable
                 return parser_code_t::memory_violation;
             }
 
             // Use short TCP header which has only access to source and destination ports
-            cropped_tcp_header_only_ports_t* tcp_header = (cropped_tcp_header_only_ports_t*)local_pointer;
+            const cropped_tcp_header_only_ports_t* tcp_header = (const cropped_tcp_header_only_ports_t*)local_pointer;
 
             packet.source_port      = tcp_header->get_source_port_host_byte_order();
             packet.destination_port = tcp_header->get_destination_port_host_byte_order();
@@ -243,7 +260,7 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
             return parser_code_t::success;
         }
 
-        tcp_header_t* tcp_header = (tcp_header_t*)local_pointer;
+        const tcp_header_t* tcp_header = (const tcp_header_t*)local_pointer;
 
         packet.source_port      = tcp_header->get_source_port_host_byte_order();
         packet.destination_port = tcp_header->get_destination_port_host_byte_order();
@@ -257,7 +274,7 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
             return parser_code_t::memory_violation;
         }
 
-        udp_header_t* udp_header = (udp_header_t*)local_pointer;
+        const udp_header_t* udp_header = (const udp_header_t*)local_pointer;
 
         packet.source_port      = udp_header->get_source_port_host_byte_order();
         packet.destination_port = udp_header->get_destination_port_host_byte_order();
@@ -271,7 +288,7 @@ parser_code_t parse_raw_packet_to_simple_packet_full_ng(const uint8_t* pointer,
             return parser_code_t::memory_violation;
         }
 
-        gre_header_t* gre_header = (gre_header_t*)local_pointer;
+        const gre_header_t* gre_header = (const gre_header_t*)local_pointer;
 
         // Current version of parser does not handle these special codes and we just fail parsing process
         // These flags may extend length of GRE header and current logic is not ready to decode any of them
@@ -361,8 +378,8 @@ parser_code_t parse_raw_ipv4_packet_to_simple_packet_full_ng(const uint8_t* poin
     packet.ip_fragmented = ipv4_header->is_fragmented();
 
     // We keep these variables to maintain backward compatibility with parse_raw_packet_to_simple_packet_full()
-    packet.captured_payload_length      = length_before_sampling;
-    packet.payload_full_length = length_before_sampling;
+    packet.captured_payload_length = length_before_sampling;
+    packet.payload_full_length     = length_before_sampling;
 
     // Pointer to payload
     packet.payload_pointer = (void*)pointer;
@@ -409,4 +426,3 @@ parser_code_t parse_raw_ipv4_packet_to_simple_packet_full_ng(const uint8_t* poin
 
     return parser_code_t::success;
 }
-