Merge pull request #176 from CESNET/flow-hash-plugin

Flow hash plugin

Author: SiskaPavel

Makefile.am

@@ -132,7 +132,9 @@ ipfixprobe_process_src=\
 		process/vlan.hpp \
 		process/vlan.cpp \
 		process/nettisa.hpp \
-		process/nettisa.cpp
+		process/nettisa.cpp \
+		process/flow_hash.hpp \
+		process/flow_hash.cpp
 
 if WITH_QUIC
 ipfixprobe_process_src+=\

README.md

@@ -643,6 +643,14 @@ List of fields exported together with basic flow fields on the interface by VLAN
 |:------------:|:------:|:--------------------------:|
 | VLAN_ID      | uint16 | Vlan ID (used in flow key) |
 
+### Flow Hash
+
+List of fields exported together with basic flow fields on interface by flow_hash plugin.
+
+| Output field       | Type   | Description                       |
+|:------------------:|:------:|:---------------------------------:|
+| FLOW_ID            | uint64 | Hash of the flow - unique flow id |
+
 ## Simplified function diagram
 Diagram below shows how `ipfixprobe` works.
 

include/ipfixprobe/flowifc.hpp

@@ -246,6 +246,8 @@ struct Record {
  * \brief Flow record struct constaining basic flow record data and extension headers.
  */
 struct Flow : public Record {
+   uint64_t flow_hash;
+
    struct timeval time_first;
    struct timeval time_last;
    uint64_t src_bytes;

include/ipfixprobe/ipfix-elements.hpp

@@ -80,6 +80,7 @@ namespace ipxp {
 #define INPUT_INTERFACE(F)            F(0,       10,    4,   &this->dir_bit_field)
 #define OUTPUT_INTERFACE(F)           F(0,       14,    2,   nullptr)
 #define FLOW_END_REASON(F)            F(0,      136,    1,   &flow.end_reason)
+#define FLOW_ID(F)                    F(0,      148,    8,   &flow.flow_hash)
 
 #define ETHERTYPE(F)                  F(0,      256,    2,   nullptr)
 
@@ -346,7 +347,7 @@ namespace ipxp {
    F(HTTP_CONTENT_TYPE) \
    F(HTTP_SERVER) \
    F(HTTP_SET_COOKIE_NAMES) \
-   F(HTTP_STATUS) 
+   F(HTTP_STATUS)
 
 #define IPFIX_RTSP_TEMPLATE(F) \
    F(RTSP_METHOD) \
@@ -529,6 +530,10 @@ namespace ipxp {
   F(NTS_TIME_DISTRIBUTION) \
   F(NTS_SWITCHING_RATIO)
 
+
+#define IPFIX_FLOW_HASH_TEMPLATE(F) \
+   F(FLOW_ID)
+
 #ifdef WITH_FLEXPROBE
 #define IPFIX_FLEXPROBE_DATA_TEMPLATE(F) F(FX_FRAME_SIGNATURE) F(FX_INPUT_INTERFACE)
 #define IPFIX_FLEXPROBE_TCP_TEMPLATE(F) F(FX_TCP_TRACKING)
@@ -573,7 +578,8 @@ namespace ipxp {
    IPFIX_FLEXPROBE_ENCR_TEMPLATE(F) \
    IPFIX_SSADETECTOR_TEMPLATE(F) \
    IPFIX_ICMP_TEMPLATE(F) \
-   IPFIX_NETTISA_TEMPLATE(F)
+   IPFIX_NETTISA_TEMPLATE(F) \
+   IPFIX_FLOW_HASH_TEMPLATE(F)
 
 /**
  * Helper macro, convert FIELD into its name as a C literal.

process/create_plugin.sh

@@ -78,7 +78,7 @@ struct RecordExt${PLUGIN_UPPER} : public RecordExt {
    }
 
 #ifdef WITH_NEMEA
-   virtual void fill_unirec(ur_template_t *tmplt, void *record)
+   void fill_unirec(ur_template_t *tmplt, void *record) override
    {
    }
 
@@ -88,7 +88,7 @@ struct RecordExt${PLUGIN_UPPER} : public RecordExt {
    }
 #endif
 
-   virtual int fill_ipfix(uint8_t *buffer, int size)
+   int fill_ipfix(uint8_t *buffer, int size) override
    {
       return 0;
    }

process/flow_hash.cpp

@@ -0,0 +1,77 @@
+/**
+ * \file flow_hash.cpp
+ * \brief Plugin for parsing flow_hash traffic.
+ * \author Jakub Antonín Štigler [email protected]
+ * \date 2023
+ */
+/*
+ * Copyright (C) 2023 CESNET
+ *
+ * LICENSE TERMS
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in
+ *    the documentation and/or other materials provided with the
+ *    distribution.
+ * 3. Neither the name of the Company nor the names of its contributors
+ *    may be used to endorse or promote products derived from this
+ *    software without specific prior written permission.
+ *
+ *
+ *
+ */
+
+#include <iostream>
+
+#include "flow_hash.hpp"
+
+namespace ipxp {
+
+int RecordExtFLOW_HASH::REGISTERED_ID = -1;
+
+__attribute__((constructor)) static void register_this_plugin()
+{
+    static PluginRecord rec = PluginRecord("flow_hash", [](){return new FLOW_HASHPlugin();});
+    register_plugin(&rec);
+    RecordExtFLOW_HASH::REGISTERED_ID = register_extension();
+}
+
+FLOW_HASHPlugin::FLOW_HASHPlugin()
+{
+}
+
+FLOW_HASHPlugin::~FLOW_HASHPlugin()
+{
+}
+
+void FLOW_HASHPlugin::init(const char *params)
+{
+}
+
+void FLOW_HASHPlugin::close()
+{
+}
+
+ProcessPlugin *FLOW_HASHPlugin::copy()
+{
+    return new FLOW_HASHPlugin(*this);
+}
+
+int FLOW_HASHPlugin::post_create(Flow &rec, const Packet &pkt)
+{
+    auto ext = new RecordExtFLOW_HASH();
+
+    ext->flow_hash = rec.flow_hash;
+
+    rec.add_extension(ext);
+
+    return 0;
+}
+
+}
+

process/flow_hash.hpp

@@ -0,0 +1,133 @@
+/**
+ * \file flow_hash.hpp
+ * \brief Plugin for parsing flow_hash traffic.
+ * \author Jakub Antonín Štigler [email protected]
+ * \date 2023
+ */
+/*
+ * Copyright (C) 2023 CESNET
+ *
+ * LICENSE TERMS
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in
+ *    the documentation and/or other materials provided with the
+ *    distribution.
+ * 3. Neither the name of the Company nor the names of its contributors
+ *    may be used to endorse or promote products derived from this
+ *    software without specific prior written permission.
+ *
+ *
+ *
+ */
+
+#ifndef IPXP_PROCESS_FLOW_HASH_HPP
+#define IPXP_PROCESS_FLOW_HASH_HPP
+
+#include <cstring>
+
+#ifdef WITH_NEMEA
+  #include "fields.h"
+#endif
+
+#include <string>
+#include <sstream>
+
+#include <ipfixprobe/process.hpp>
+#include <ipfixprobe/flowifc.hpp>
+#include <ipfixprobe/packet.hpp>
+#include <ipfixprobe/ipfix-elements.hpp>
+#include <ipfixprobe/byte-utils.hpp>
+
+namespace ipxp {
+
+#define FLOW_HASH_UNIREC_TEMPLATE "FLOW_ID"
+
+UR_FIELDS (
+    uint64 FLOW_ID
+)
+
+/**
+ * \brief Flow record extension header for storing parsed FLOW_HASH data.
+ */
+struct RecordExtFLOW_HASH : public RecordExt {
+    static int REGISTERED_ID;
+
+    // Value in host byte order
+    uint64_t flow_hash;
+
+    RecordExtFLOW_HASH() : RecordExt(REGISTERED_ID)
+    {
+        flow_hash = 0;
+    }
+
+#ifdef WITH_NEMEA
+   void fill_unirec(ur_template_t *tmplt, void *record) override
+    {
+        ur_set(tmplt, record, F_FLOW_ID, flow_hash);
+    }
+
+    const char *get_unirec_tmplt() const
+    {
+        return FLOW_HASH_UNIREC_TEMPLATE;
+    }
+#endif
+
+    int fill_ipfix(uint8_t *buffer, int size) override
+    {
+        constexpr int LEN = sizeof(flow_hash);
+
+        if (size < LEN) {
+            return -1;
+        }
+
+        // value is converted from host byte-order to network byte-order
+        *reinterpret_cast<decltype(flow_hash) *>(buffer) = swap_uint64(flow_hash);
+
+        return LEN;
+    }
+
+    const char **get_ipfix_tmplt() const
+    {
+        static const char *ipfix_template[] = {
+            IPFIX_FLOW_HASH_TEMPLATE(IPFIX_FIELD_NAMES)
+            NULL
+        };
+        return ipfix_template;
+    }
+
+    std::string get_text() const
+    {
+        std::ostringstream out;
+        out << std::hex << "flow_id=\"" << flow_hash << '"';
+
+        return out.str();
+    }
+};
+
+/**
+ * \brief Process plugin for parsing FLOW_HASH packets.
+ */
+class FLOW_HASHPlugin : public ProcessPlugin
+{
+public:
+    FLOW_HASHPlugin();
+    ~FLOW_HASHPlugin();
+    void init(const char *params);
+    void close();
+    OptionsParser *get_parser() const { return new OptionsParser("flow_hash", "Export flow hash as flow id"); }
+    std::string get_name() const { return "flow_hash"; }
+    RecordExt *get_ext() const { return new RecordExtFLOW_HASH(); }
+    ProcessPlugin *copy();
+
+    int post_create(Flow &rec, const Packet &pkt);
+};
+
+}
+#endif /* IPXP_PROCESS_FLOW_HASH_HPP */
+

storage/cache.cpp

@@ -107,6 +107,7 @@ void FlowRecord::create(const Packet &pkt, uint64_t hash)
 
    m_flow.time_first = pkt.ts;
    m_flow.time_last = pkt.ts;
+   m_flow.flow_hash = hash;
 
    memcpy(m_flow.src_mac, pkt.src_mac, 6);
    memcpy(m_flow.dst_mac, pkt.dst_mac, 6);