capture device node

CMakeLists.txt

@@ -72,6 +72,7 @@ option(INCLUDE_BRIDGE "link Bridge staticly to the core"  TRUE)
 if(LINUX) # todo (other platforms)
   option(INCLUDE_TUNDEVICE "link TunDevice staticly to the core"  TRUE)
   option(INCLUDE_RAWDEVICE "link RawDevice staticly to the core"  TRUE)
+  option(INCLUDE_CAPTURE "link CapTureDevice staticly to the core"  TRUE)
 endif()
 
 option(INCLUDE_OPENSSL_SERVER "link OpenSSlServer staticly to the core"  TRUE)
@@ -150,6 +151,14 @@ target_link_directories(Waterwall PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/tunnels/ada
 target_link_libraries(Waterwall RawDevice)
 endif()
 
+#capture device
+if (INCLUDE_CAPTURE)
+target_compile_definitions(Waterwall PUBLIC INCLUDE_CAPTURE=1)
+add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/tunnels/adapters/device/capture)
+target_link_directories(Waterwall PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/tunnels/adapters/device/capture)
+target_link_libraries(Waterwall CaptureDevice)
+endif()
+
 #layer3 receiver
 if (INCLUDE_LAYER3_RECEIVER)
 target_compile_definitions(Waterwall PUBLIC INCLUDE_LAYER3_RECEIVER=1)

core/static_tunnels.c

@@ -25,6 +25,10 @@
 #include "tunnels/adapters/device/raw/raw_device.h"
 #endif
 
+#ifdef INCLUDE_CAPTURE
+#include "tunnels/adapters/device/capture/caputre_device.h"
+#endif
+
 #ifdef INCLUDE_LAYER3_RECEIVER
 #include "tunnels/layer3/receiver/receiver.h"
 #endif
@@ -188,6 +192,10 @@ void loadStaticTunnelsIntoCore(void)
     USING(RawDevice);
 #endif
 
+#ifdef INCLUDE_CAPTURE
+    USING(CaptureDevice);
+#endif
+
 #ifdef INCLUDE_LAYER3_RECEIVER
     USING(Layer3Receiver);
 #endif

tunnels/adapters/device/capture/CMakeLists.txt

@@ -0,0 +1,13 @@
+
+add_library(CaptureDevice STATIC
+
+        caputre_device.c
+
+)
+
+target_link_libraries(CaptureDevice ww)
+
+target_include_directories(CaptureDevice PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/../../../shared/layer3)
+
+target_compile_definitions(CaptureDevice PRIVATE  CaptureDevice_VERSION=0.1)
+

tunnels/adapters/device/capture/caputre_device.c

@@ -0,0 +1,234 @@
+#include "caputre_device.h"
+#include "frand.h"
+#include "loggers/network_logger.h"
+#include "managers/signal_manager.h"
+#include "packet_types.h"
+#include "utils/jsonutils.h"
+#include "utils/procutils.h"
+#include "ww/devices/capture/capture.h"
+
+#define LOG_PACKET_INFO 1
+
+enum capturedevice_direction_dynamic_value_status
+{
+    kDvsIncoming = kDvsFirstOption,
+    kDvsOutgoing,
+    kDvsBoth
+};
+
+enum capturedevice_filter_type_dynamic_value_status
+{
+    kDvsSourceIp = kDvsFirstOption,
+    kDvsDestIp
+};
+
+static const char *ip_tables_enable_queue_mi  = "iptables -I INPUT -s %s -j NFQUEUE --queue-num %d";
+static const char *ip_tables_disable_queue_mi = "iptables -D INPUT -s %s -j NFQUEUE --queue-num %d";
+
+typedef struct capture_device_state_s
+{
+    capture_device_t *cdev;
+    line_t          **thread_lines;
+    char             *name;
+    char             *exitcmd;
+    uint32_t          queue_number;
+    uint32_t          except_fwmark;
+
+} capture_device_state_t;
+
+static void printIPPacketInfo(const unsigned char *buffer, unsigned int len)
+{
+    char  src_ip[INET6_ADDRSTRLEN];
+    char  dst_ip[INET6_ADDRSTRLEN];
+    char  logbuf[2048];
+    int   rem = sizeof(logbuf);
+    char *ptr = logbuf;
+    int   ret;
+
+    uint8_t version = buffer[0] >> 4;
+
+    if (version == 4)
+    {
+        struct ipv4header *ip_header = (struct ipv4header *) buffer;
+
+        inet_ntop(AF_INET, &ip_header->saddr, src_ip, INET_ADDRSTRLEN);
+        inet_ntop(AF_INET, &ip_header->daddr, dst_ip, INET_ADDRSTRLEN);
+
+        ret = snprintf(ptr, rem, "Received: => From %s to %s, Data: ", src_ip, dst_ip);
+    }
+    else if (version == 6)
+    {
+        struct ipv6header *ip6_header = (struct ipv6header *) buffer;
+
+        inet_ntop(AF_INET6, &ip6_header->saddr, src_ip, INET6_ADDRSTRLEN);
+        inet_ntop(AF_INET6, &ip6_header->daddr, dst_ip, INET6_ADDRSTRLEN);
+
+        ret = snprintf(ptr, rem, "Received:  From %s to %s, Data: ", src_ip, dst_ip);
+    }
+    else
+    {
+        ret = snprintf(ptr, rem, "Received: => Unknown IP version, Data: ");
+    }
+
+    ptr += ret;
+    rem -= ret;
+
+    for (int i = 0; i < (int) min(len, 240); i++)
+    {
+        ret = snprintf(ptr, rem, "%02x ", buffer[i]);
+        ptr += ret;
+        rem -= ret;
+    }
+    *ptr = '\0';
+
+    LOGD(logbuf);
+}
+
+static void upStream(tunnel_t *self, context_t *c)
+{
+    capture_device_state_t *state = TSTATE((tunnel_t *) self);
+
+    capture_device_t *cdev = state->cdev;
+    writeToCaptureDevce(cdev, c->payload);
+
+    dropContexPayload(c);
+    destroyContext(c);
+}
+
+static void downStream(tunnel_t *self, context_t *c)
+{
+    (void) (self);
+    (void) (c);
+    assert(false);
+
+    if (c->payload)
+    {
+        dropContexPayload(c);
+    }
+    destroyContext(c);
+}
+
+static void onIPPacketReceived(struct capture_device_s *cdev, void *userdata, shift_buffer_t *buf, tid_t tid)
+{
+    (void) cdev;
+    tunnel_t               *self  = userdata;
+    capture_device_state_t *state = TSTATE((tunnel_t *) self);
+
+#if LOG_PACKET_INFO
+    printIPPacketInfo(rawBuf(buf), bufLen(buf));
+#endif
+
+    // reuseBuffer(getWorkerBufferPool(tid), buf);
+
+    context_t *ctx = newContext(state->thread_lines[tid]);
+    ctx->payload   = buf;
+    self->up->upStream(self->up, ctx);
+}
+
+static void exitHook(void *userdata, int sig)
+{
+    (void) sig;
+    capture_device_state_t *state = TSTATE((tunnel_t *) userdata);
+    execCmd(state->exitcmd);
+}
+
+tunnel_t *newCaptureDevice(node_instance_context_t *instance_info)
+{
+    capture_device_state_t *state = globalMalloc(sizeof(capture_device_state_t));
+    memset(state, 0, sizeof(capture_device_state_t));
+
+    cJSON *settings = instance_info->node_settings_json;
+
+    if (! (cJSON_IsObject(settings) && settings->child != NULL))
+    {
+        LOGF("JSON Error: CaptureDevice->settings (object field) : The object was empty or invalid");
+        return NULL;
+    }
+
+    // not forced
+    getStringFromJsonObjectOrDefault(&(state->name), settings, "device-name", "unnamed-device");
+
+    dynamic_value_t directoin =
+        parseDynamicNumericValueFromJsonObject(settings, "direction", 3, "incoming", "outgoing", "bothdirections");
+
+    if ((int) directoin.status < kDvsIncoming)
+    {
+        LOGF("JSON Error: CaptureDevice->settings->direction (string field) : direction is not specified or invalid");
+        return NULL;
+    }
+    dynamic_value_t mode = parseDynamicNumericValueFromJsonObject(settings, "mode", 2, "source-ip", "dest-ip");
+    if ((int) mode.status < kDvsSourceIp)
+    {
+        LOGF("JSON Error: CaptureDevice->settings->mode (string field) : mode is not specified or invalid");
+        return NULL;
+    }
+    state->queue_number = 200 + (fastRand() % 200);
+    char *ipbuf         = NULL;
+    if (! getStringFromJsonObject(&ipbuf, settings, "ip"))
+    {
+        LOGF("JSON Error: CaptureDevice->settings->ip (string field) : mode is not specified or invalid");
+    }
+
+    char     *cmdbuf = globalMalloc(200);
+    tunnel_t *t      = newTunnel();
+
+    if ((int) directoin.status == kDvsIncoming)
+    {
+        if ((int) mode.status == kDvsSourceIp)
+        {
+            snprintf(cmdbuf, 100, ip_tables_enable_queue_mi, ipbuf, (int) state->queue_number);
+            if (execCmd(cmdbuf).exit_code != 0)
+            {
+                LOGF("CaptureDevicer: command failed: %s", cmdbuf);
+                return NULL;
+            }
+
+            state->exitcmd = cmdbuf;
+            snprintf(cmdbuf, 100, ip_tables_disable_queue_mi, ipbuf, (int) state->queue_number);
+            registerAtExitCallback(exitHook, t);
+        }
+        else
+        {
+            //todo
+        }
+    }
+
+    state->thread_lines = globalMalloc(sizeof(line_t *) * WORKERS_COUNT);
+    for (unsigned int i = 0; i < WORKERS_COUNT; i++)
+    {
+        state->thread_lines[i] = newLine(i);
+    }
+
+    state->cdev = createCaptureDevice(state->name, state->queue_number, t, onIPPacketReceived);
+
+    if (state->cdev == NULL)
+    {
+        LOGF("CaptureDevice: could not create device");
+        return NULL;
+    }
+    bringCaptureDeviceUP(state->cdev);
+
+    t->state      = state;
+    t->upStream   = &upStream;
+    t->downStream = &downStream;
+
+    return t;
+}
+
+api_result_t apiCaptureDevice(tunnel_t *self, const char *msg)
+{
+    (void) (self);
+    (void) (msg);
+    return (api_result_t) {0};
+}
+
+tunnel_t *destroyCaptureDevice(tunnel_t *self)
+{
+    (void) (self);
+    return NULL;
+}
+
+tunnel_metadata_t getMetadataCaptureDevice(void)
+{
+    return (tunnel_metadata_t) {.version = 0001, .flags = 0x0};
+}

tunnels/adapters/device/capture/caputre_device.h

@@ -0,0 +1,13 @@
+#pragma once
+#include "api.h"
+
+// 
+//      CaptureDevice
+// 
+
+//  this node will not join a chain , it will be used by other nodes (if they accept a device)
+
+tunnel_t *        newCaptureDevice(node_instance_context_t *instance_info);
+api_result_t      apiCaptureDevice(tunnel_t *self, const char *msg);
+tunnel_t *        destroyCaptureDevice(tunnel_t *self);
+tunnel_metadata_t getMetadataCaptureDevice(void);

ww/devices/capture/capture_linux.c

@@ -15,7 +15,6 @@
 #include <netinet/ip.h>
 #include <string.h>
 #include <sys/ioctl.h>
-#include <unistd.h>
 
 enum
 {
@@ -342,7 +341,7 @@ static HTHREAD_ROUTINE(routineWriteToCapture) // NOLINT
 
         if (nwrite < 0)
         {
-            LOGE("CaptureDevice: CaptureDevice: writing to Capture device failed");
+            LOGE("CaptureDevice: writing to Capture device failed");
             return 0;
         }
     }
@@ -356,11 +355,11 @@ void writeToCaptureDevce(capture_device_t *cdev, shift_buffer_t *buf)
     {
         if (closed)
         {
-            LOGE("CaptureDevice: CaptureDevice: write failed, channel was closed");
+            LOGE("CaptureDevice: write failed, channel was closed");
         }
         else
         {
-            LOGE("CaptureDevice: CaptureDevice: write failed, ring is full");
+            LOGE("CaptureDevice:write failed, ring is full");
         }
         reuseBufferThreadSafe(buf);
     }