QinQ support and Source optimization

README.md

@@ -173,7 +173,8 @@ These plugins extract protocol-specific or behavioral information from packets a
 | [`smtp`](./src/plugins/process/smtp/README.md)             | extracts SMTP envelope data (from, to, subject, etc.)        |
 | [`ssaDetector`](./src/plugins/process/ssaDetector/README.md) | performs simple anomaly detection based on traffic patterns |
 | [`ssdp`](./src/plugins/process/ssdp/README.md)             | parses SSDP (UPnP discovery) protocol                        |
-| [`vlan`](./src/plugins/process/vlan/README.md)             | extracts VLAN IDs and QinQ encapsulation                     |
+| [`vlan`](./src/plugins/process/vlan/README.md)             | extracts VLAN IDs                                            |
+| [`qinq`](./src/plugins/process/qinq/README.md)             | extracts QinQ outer and inner VLAN IDs.                      |
 
 ---
 ### Output Plugins

include/ipfixprobe/ipfix-elements.hpp

@@ -93,6 +93,8 @@ namespace ipxp {
 #define ETHERTYPE(F) F(0, 256, 2, nullptr)
 
 #define VLAN_ID(F) F(0, 58, 2, nullptr)
+#define DOT1Q_VLAN_ID(F) F(0, 243, 2, nullptr)
+#define DOT1Q_CUSTOMER_VLAN_ID(F) F(0, 245, 2, nullptr)
 
 #define L2_SRC_MAC(F) F(0, 56, 6, flow.src_mac)
 #define L2_DST_MAC(F) F(0, 80, 6, flow.dst_mac)
@@ -564,6 +566,10 @@ namespace ipxp {
 
 #define IPFIX_VLAN_TEMPLATE(F) F(VLAN_ID)
 
+#define IPFIX_QINQ_TEMPLATE(F)                                                                     \
+    F(DOT1Q_VLAN_ID)                                                                               \
+	F(DOT1Q_CUSTOMER_VLAN_ID)
+
 #define IPFIX_NETTISA_TEMPLATE(F)                                                                  \
 	F(NTS_MEAN)                                                                                    \
 	F(NTS_MIN)                                                                                     \
@@ -615,6 +621,7 @@ namespace ipxp {
 	IPFIX_SSADETECTOR_TEMPLATE(F)                                                                  \
 	IPFIX_ICMP_TEMPLATE(F)                                                                         \
 	IPFIX_VLAN_TEMPLATE(F)                                                                         \
+	IPFIX_QINQ_TEMPLATE(F)                                                                         \
 	IPFIX_NETTISA_TEMPLATE(F)                                                                      \
 	IPFIX_FLOW_HASH_TEMPLATE(F)
 

include/ipfixprobe/packet.hpp

@@ -60,6 +60,7 @@ struct Packet : public Record {
 	ipaddr_t src_ip;
 	ipaddr_t dst_ip;
 	uint32_t vlan_id;
+	uint32_t vlan_id2;
 	uint32_t frag_id;
 	uint16_t frag_off;
 	bool more_fragments;
@@ -119,6 +120,7 @@ struct Packet : public Record {
 		, src_ip({0})
 		, dst_ip({0})
 		, vlan_id(0)
+		, vlan_id2(0)
 		, frag_id(0)
 		, frag_off(0)
 		, more_fragments(false)

init/config2args.py

@@ -336,6 +336,40 @@ def process_storage(config):
             cache_params.append(f"s={cache['size_exponent']}")
         if "line_size_exponent" in cache:
             cache_params.append(f"l={cache['line_size_exponent']}")
+        if "source_optimization" in cache:
+            so_value = "true" if cache['source_optimization'] else "false"
+            cache_params.append(f"so={so_value}")
+        if "source_optimization_network" in cache:
+            son_networks = cache.get("source_optimization_network")
+            if son_networks:
+                # Handle source_optimization_network - can be a list of dicts or a single dict
+                if isinstance(son_networks, dict):
+                    # If it's a single dict with main/exclude, convert to list
+                    son_networks = [son_networks]
+                elif not isinstance(son_networks, list):
+                    son_networks = [son_networks]
+
+                # Generate -son arguments for each network group
+                for network_group in son_networks:
+                    if isinstance(network_group, dict):
+                        main = network_group.get("main")
+                        exclude = network_group.get("exclude", "")
+
+                        if main:
+                            # Combine main network with exclusions
+                            networks = [main.strip()]
+                            if exclude:
+                                # Handle exclude as comma-separated string
+                                if isinstance(exclude, str):
+                                    excludes = [e.strip() for e in exclude.split(",")]
+                                    networks.extend(excludes)
+                                elif isinstance(exclude, list):
+                                    networks.extend(exclude)
+
+                            cache_params.append(f"son={','.join(networks)}")
+                    elif isinstance(network_group, str):
+                        # If it's just a string, use it directly
+                        cache_params.append(f"son={network_group}")
         if cache_params:
             params.append(f"{';'.join(cache_params)}")
 

init/schema.json

@@ -412,6 +412,27 @@
             "line_size_exponent": {
               "type": "integer",
               "minimum": 1
+            },
+            "source_optimization": {
+              "type": "boolean"
+            },
+            "source_optimization_network": {
+              "type": "array",
+              "items": {
+                "type": "object",
+                "properties": {
+                  "main": {
+                    "type": "string"
+                  },
+                  "exclude": {
+                    "type": "string"
+                  }
+                },
+                "required": [
+                  "main"
+                ],
+                "additionalProperties": false
+              }
             }
           },
           "additionalProperties": false

src/plugins/input/parser/parser.cpp

@@ -114,7 +114,7 @@ inline uint16_t parse_eth_hdr(const u_char* data_ptr, uint16_t data_len, Packet*
 
 	// set the default value in case there is no VLAN ID
 	pkt->vlan_id = 0;
-
+    pkt->vlan_id2 = 0;
 	if (ethertype == ETH_P_8021AD || ethertype == ETH_P_8021Q) {
 		if (4 > data_len - hdr_len) {
 			throw "Parser detected malformed packet";
@@ -138,7 +138,8 @@ inline uint16_t parse_eth_hdr(const u_char* data_ptr, uint16_t data_len, Packet*
 		if (4 > data_len - hdr_len) {
 			throw "Parser detected malformed packet";
 		}
-		DEBUG_CODE(uint16_t vlan = ntohs(*(uint16_t*) (data_ptr + hdr_len)));
+		uint16_t vlan = ntohs(*(uint16_t*) (data_ptr + hdr_len));
+		pkt->vlan_id2 = vlan & 0x0FFF;
 		DEBUG_MSG("\t802.1q field:\n");
 		DEBUG_MSG("\t\tPriority:\t%u\n", ((vlan & 0xE000) >> 12));
 		DEBUG_MSG("\t\tCFI:\t\t%u\n", ((vlan & 0x1000) >> 11));
@@ -773,6 +774,9 @@ void parse_packet(
 	if (pkt->vlan_id) {
 		stats.vlan_packets++;
 	}
+	if( pkt->vlan_id2) {
+		stats.vlan_packets++;
+	}
 
 	if (pkt->ethertype == ETH_P_IP) {
 		stats.ipv4_packets++;
@@ -804,6 +808,9 @@ void parse_packet(
 	pkt->payload = pkt->packet + data_offset;
 
 	stats.vlan_stats[pkt->vlan_id].update(*pkt);
+	if( pkt->vlan_id2) {
+		stats.vlan_stats[pkt->vlan_id2].update(*pkt);
+	}
 
 	DEBUG_MSG("Payload length:\t%u\n", pkt->payload_len);
 	DEBUG_MSG("Packet parser exits: packet parsed\n");

src/plugins/process/CMakeLists.txt

@@ -21,6 +21,7 @@ add_subdirectory(smtp)
 add_subdirectory(quic)
 add_subdirectory(tls)
 add_subdirectory(http)
+add_subdirectory(qinq)
 
 if (ENABLE_PROCESS_EXPERIMENTAL)
 	add_subdirectory(sip)

src/plugins/process/qinq/CMakeLists.txt

@@ -0,0 +1,27 @@
+project(ipfixprobe-process-qinq VERSION 1.0.0 DESCRIPTION "ipfixprobe-process-qinq plugin")
+
+add_library(ipfixprobe-process-qinq MODULE
+	src/qinq.cpp
+	src/qinq.hpp
+)
+
+set_target_properties(ipfixprobe-process-qinq PROPERTIES
+	CXX_VISIBILITY_PRESET hidden
+	VISIBILITY_INLINES_HIDDEN YES
+)
+
+target_include_directories(ipfixprobe-process-qinq PRIVATE
+	${CMAKE_SOURCE_DIR}/include/
+)
+
+if(ENABLE_NEMEA)
+	target_link_libraries(ipfixprobe-process-qinq PRIVATE
+		-Wl,--whole-archive ipfixprobe-nemea-fields -Wl,--no-whole-archive
+		unirec::unirec
+		trap::trap
+	)
+endif()
+
+install(TARGETS ipfixprobe-process-qinq
+	LIBRARY DESTINATION "${INSTALL_DIR_LIB}/ipfixprobe/process/"
+)

src/plugins/process/qinq/README.md

@@ -0,0 +1 @@
+QinQ process plugin for parsing QinQ traffic, outputs outer and inner VLAN IDs.

src/plugins/process/qinq/src/qinq.cpp

@@ -0,0 +1,56 @@
+/**
+ * @file
+ * @brief Plugin for parsing basicplus traffic.
+ * @author Jakub Antonín Štigler xstigl00@xstigl00@stud.fit.vut.cz
+ * @author Pavel Siska <siska@cesnet.cz>
+ * @date 2025
+ *
+ * Copyright (c) 2025 CESNET
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "qinq.hpp"
+
+#include <iostream>
+
+#include <ipfixprobe/pluginFactory/pluginManifest.hpp>
+#include <ipfixprobe/pluginFactory/pluginRegistrar.hpp>
+
+namespace ipxp {
+
+static const PluginManifest qinqPluginManifest = {
+	.name = "qinq",
+	.description = "QinQ process plugin for parsing QinQ traffic, outputs outer and inner VLAN IDs.",
+	.pluginVersion = "1.0.0",
+	.apiVersion = "1.0.0",
+	.usage =
+		[]() {
+			OptionsParser parser("qinq", "Parse qinq traffic");
+			parser.usage(std::cout);
+		},
+};
+
+QinQPlugin::QinQPlugin(const std::string& params, int pluginID)
+	: ProcessPlugin(pluginID)
+{
+	init(params.c_str());
+}
+
+ProcessPlugin* QinQPlugin::copy()
+{
+	return new QinQPlugin(*this);
+}
+
+int QinQPlugin::post_create(Flow& rec, const Packet& pkt)
+{
+	auto ext = new RecordExtQinQ(m_pluginID);
+	ext->vlan_id = pkt.vlan_id;
+	ext->vlan_id2 = pkt.vlan_id2;
+	rec.add_extension(ext);
+	return 0;
+}
+
+static const PluginRegistrar<QinQPlugin, ProcessPluginFactory> qinqRegistrar(qinqPluginManifest);
+
+} // namespace ipxp

src/plugins/process/qinq/src/qinq.hpp

@@ -0,0 +1,104 @@
+/**
+ * @file
+ * @brief Plugin for parsing basicplus traffic.
+ * @author Jakub Antonín Štigler xstigl00@xstigl00@stud.fit.vut.cz
+ * @author Pavel Siska <siska@cesnet.cz>
+ * @date 2025
+ *
+ * Copyright (c) 2025 CESNET
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#pragma once
+
+#include <cstring>
+
+#ifdef WITH_NEMEA
+#include "fields.h"
+#endif
+
+#include <cstdint>
+#include <sstream>
+#include <string>
+
+#include <ipfixprobe/flowifc.hpp>
+#include <ipfixprobe/ipfix-elements.hpp>
+#include <ipfixprobe/packet.hpp>
+#include <ipfixprobe/processPlugin.hpp>
+
+namespace ipxp {
+
+#define QINQ_UNIREC_TEMPLATE "DOT1Q_VLAN_ID,DOT1Q_CUSTOMER_VLAN_ID"
+
+UR_FIELDS(
+	uint16 VLAN_ID;
+	uint16 VLAN_ID2;
+)
+
+
+/**
+ * \brief Flow record extension header for storing parsed QinQ data.
+ */
+struct RecordExtQinQ : public RecordExt {
+	// vlan id is in the host byte order
+	uint16_t vlan_id;
+	uint16_t vlan_id2;
+	RecordExtQinQ(int pluginID)
+		: RecordExt(pluginID)
+		, vlan_id(0)
+		, vlan_id2(0)
+	{
+	}
+
+#ifdef WITH_NEMEA
+	virtual void fill_unirec(ur_template_t* tmplt, void* record)
+	{
+		ur_set(tmplt, record, F_VLAN_ID, vlan_id);
+		ur_set(tmplt, record, F_VLAN_ID2, vlan_id2);
+	}
+
+	const char* get_unirec_tmplt() const { return QINQ_UNIREC_TEMPLATE; }
+#endif
+
+	virtual int fill_ipfix(uint8_t* buffer, int size)
+	{
+		const int LEN = sizeof(vlan_id);		
+		const int LEN2 = sizeof(vlan_id2);
+		if( size < (LEN + LEN2) ) {
+			return LEN;
+		}
+		*reinterpret_cast<uint16_t*>(buffer) = htons(vlan_id);
+		*reinterpret_cast<uint16_t*>(buffer + LEN) = htons(vlan_id2);
+		return (LEN + LEN2);
+	}
+
+	const char** get_ipfix_tmplt() const
+	{
+		static const char* ipfix_qinq_template[] = {IPFIX_QINQ_TEMPLATE(IPFIX_FIELD_NAMES) NULL};
+	    return ipfix_qinq_template;		
+	}
+
+	std::string get_text() const
+	{
+		std::ostringstream out;
+		out << "DOT1Q_VLAN_ID=\"" << vlan_id << "\", DOT1Q_CUSTOMER_VLAN_ID=\"" << vlan_id2 << "\"";
+		return out.str();
+	}
+};
+
+/**
+ * \brief Process plugin for parsing VLAN packets.
+ */
+class QinQPlugin : public ProcessPlugin {
+public:
+	QinQPlugin(const std::string& params, int pluginID);
+	OptionsParser* get_parser() const { return new OptionsParser("qinq", "Parse QinQ traffic"); }
+	std::string get_name() const { return "qinq"; }
+	RecordExt* get_ext() const { return new RecordExtQinQ(m_pluginID); }
+	ProcessPlugin* copy();
+
+	int post_create(Flow& rec, const Packet& pkt);
+};
+
+} // namespace ipxp

src/plugins/process/vlan/src/vlan.hpp

@@ -39,7 +39,7 @@ UR_FIELDS(uint16 VLAN_ID)
 struct RecordExtVLAN : public RecordExt {
 	// vlan id is in the host byte order
 	uint16_t vlan_id;
-
+	uint16_t vlan_id2;
 	RecordExtVLAN(int pluginID)
 		: RecordExt(pluginID)
 		, vlan_id(0)
@@ -69,8 +69,8 @@ struct RecordExtVLAN : public RecordExt {
 
 	const char** get_ipfix_tmplt() const
 	{
-		static const char* ipfix_template[] = {IPFIX_VLAN_TEMPLATE(IPFIX_FIELD_NAMES) NULL};
-		return ipfix_template;
+		static const char* ipfix_vlan_template[] = {IPFIX_VLAN_TEMPLATE(IPFIX_FIELD_NAMES) NULL};
+		return ipfix_vlan_template;
 	}
 
 	std::string get_text() const

src/plugins/storage/cache/CMakeLists.txt

@@ -12,6 +12,8 @@ add_library(ipfixprobe-storage-cache MODULE
 	src/fragmentationCache/fragmentationTable.hpp
 	src/fragmentationCache/ringBuffer.hpp
 	src/fragmentationCache/timevalUtils.hpp
+	src/sourceOptimization.cpp
+	src/sourceOptimization.hpp
 	src/xxhash.c
 	src/xxhash.h
 )