nfq: add support for batch verdicts

Normally, there is one verdict per packet, i.e., we receive a packet,
process it, and then tell the kernel what to do with that packet (eg.
DROP or ACCEPT).

recv(), packet id x
send verdict v, packet id x
recv(), packet id x+1
send verdict v, packet id x+1
[..]
recv(), packet id x+n
send verdict v, packet id x+n

An alternative is to process several packets from the queue, and then send
a batch-verdict.

recv(), packet id x
recv(), packet id x+1
[..]
recv(), packet id x+n
send batch verdict v, packet id x+n

A batch verdict affects all previous packets (packet_id <= x+n),
we thus only need to remember the last packet_id seen.

Caveats:
- can't modify payload
- verdict is applied to all packets
- nfmark (if set) will be set for all packets
- increases latency (packets remain queued by the kernel
  until batch verdict is sent).

To solve this, we only defer verdict for up to 20 packets and
send pending batch-verdict immediately if:
- no packets are currently queue
- current packet should be dropped
- current packet has different nfmark
- payload of packet was modified

This patch adds a configurable batch verdict support for workers runmode.
The batch verdicts are turned off by default.

Problem is that batch verdicts only work with kernels >= 3.1, i.e.
using newer libnetfilter_queue with an old kernel means non-working
suricata. So the functionnality has to be disabled by default.

diff --git a/configure.ac b/configure.ac
index fc32330fb5871f5281556f9c7dc31c6b5089f7d0..4dc9242288cda8e596d2d3346bf29cf1f045e99b 100644 (file)
--- a/configure.ac
+++ b/configure.ac
@@ -697,6 +697,7 @@ AC_INIT(configure.ac)
         AC_CHECK_LIB([netfilter_queue], [nfq_set_queue_maxlen],AC_DEFINE_UNQUOTED([HAVE_NFQ_MAXLEN],[1],[Found queue max length support in netfilter_queue]) ,,[-lnfnetlink])
         AC_CHECK_LIB([netfilter_queue], [nfq_set_verdict2],AC_DEFINE_UNQUOTED([HAVE_NFQ_SET_VERDICT2],[1],[Found nfq_set_verdict2 function in netfilter_queue]) ,,[-lnfnetlink])
         AC_CHECK_LIB([netfilter_queue], [nfq_set_queue_flags],AC_DEFINE_UNQUOTED([HAVE_NFQ_SET_QUEUE_FLAGS],[1],[Found nfq_set_queue_flags function in netfilter_queue]) ,,[-lnfnetlink])
+        AC_CHECK_LIB([netfilter_queue], [nfq_set_verdict_batch],AC_DEFINE_UNQUOTED([HAVE_NFQ_SET_VERDICT_BATCH],[1],[Found nfq_set_verdict_batch function in netfilter_queue]) ,,[-lnfnetlink])
 
         # check if the argument to nfq_get_payload is signed or unsigned
         AC_MSG_CHECKING([for signed nfq_get_payload payload argument])

diff --git a/src/source-nfq.c b/src/source-nfq.c
index 740fc1eb2279d43e899f37971549ddb2775ad614..6fdc3e391c6f974b68f6193edd6afac3cb5add66 100644 (file)
--- a/src/source-nfq.c
+++ b/src/source-nfq.c
@@ -163,6 +163,7 @@ typedef struct NFQCnf_ {
     uint32_t mask;
     uint32_t next_queue;
     uint32_t flags;
+    uint8_t batchcount;
 } NFQCnf;
 
 NFQCnf nfq_config;
@@ -238,7 +239,7 @@ void NFQInitConfig(char quiet)
         nfq_config.flags |= NFQ_FLAG_FAIL_OPEN;
 #else
         SCLogError(SC_ERR_NFQ_NOSUPPORT,
-                   "nfq.fail-open set but NFQ library has no support for it.");
+                   "nfq.%s set but NFQ library has no support for it.", "fail-open");
 #endif
     }
 
@@ -254,6 +255,21 @@ void NFQInitConfig(char quiet)
         nfq_config.next_queue = ((uint32_t)value) << 16;
     }
 
+    if ((ConfGetInt("nfq.batchcount", &value)) == 1) {
+#ifdef HAVE_NFQ_SET_VERDICT_BATCH
+        if (value > 255) {
+            SCLogError(SC_ERR_INVALID_ARGUMENT, "nfq.batchcount cannot exceed 255.");
+            value = 255;
+        }
+        if (value > 1)
+            nfq_config.batchcount = (uint8_t) (value - 1);
+#else
+        SCLogError(SC_ERR_NFQ_NOSUPPORT,
+                   "nfq.%s set but NFQ library has no support for it.", "batchcount");
+        exit(EXIT_FAILURE);
+#endif
+    }
+
     if (!quiet) {
         switch (nfq_config.mode) {
             case NFQ_ACCEPT_MODE:
@@ -272,6 +288,85 @@ void NFQInitConfig(char quiet)
 
 }
 
+static uint8_t NFQVerdictCacheLen(NFQQueueVars *t)
+{
+#ifdef HAVE_NFQ_SET_VERDICT_BATCH
+    return t->verdict_cache.len;
+#else
+    return 0;
+#endif
+}
+
+static void NFQVerdictCacheFlush(NFQQueueVars *t)
+{
+#ifdef HAVE_NFQ_SET_VERDICT_BATCH
+    int ret;
+    int iter = 0;
+
+    do {
+        if (t->verdict_cache.mark_valid)
+            ret = nfq_set_verdict_batch2(t->qh,
+                                         t->verdict_cache.packet_id,
+                                         t->verdict_cache.verdict,
+                                         t->verdict_cache.mark);
+        else
+            ret = nfq_set_verdict_batch(t->qh,
+                                        t->verdict_cache.packet_id,
+                                        t->verdict_cache.verdict);
+    } while ((ret < 0) && (iter++ < NFQ_VERDICT_RETRY_TIME));
+
+    if (ret < 0) {
+        SCLogWarning(SC_ERR_NFQ_SET_VERDICT, "nfq_set_verdict_batch failed");
+    } else {
+        t->verdict_cache.len = 0;
+        t->verdict_cache.mark_valid = 0;
+    }
+#endif
+}
+
+static int NFQVerdictCacheAdd(NFQQueueVars *t, Packet *p, uint32_t verdict)
+{
+#ifdef HAVE_NFQ_SET_VERDICT_BATCH
+    if (t->verdict_cache.maxlen == 0)
+        return -1;
+
+    if (p->flags & PKT_STREAM_MODIFIED || verdict == NF_DROP)
+        goto flush;
+
+    if (p->flags & PKT_MARK_MODIFIED) {
+        if (!t->verdict_cache.mark_valid) {
+            if (t->verdict_cache.len)
+                goto flush;
+            t->verdict_cache.mark_valid = 1;
+            t->verdict_cache.mark = p->nfq_v.mark;
+        } else if (t->verdict_cache.mark != p->nfq_v.mark) {
+            goto flush;
+        }
+    } else if (t->verdict_cache.mark_valid) {
+        goto flush;
+    }
+
+    if (t->verdict_cache.len == 0) {
+        t->verdict_cache.verdict = verdict;
+    } else if (t->verdict_cache.verdict != verdict)
+        goto flush;
+
+    /* same verdict, mark not set or identical -> can cache */
+    t->verdict_cache.packet_id = p->nfq_v.id;
+
+    if (t->verdict_cache.len >= t->verdict_cache.maxlen)
+        NFQVerdictCacheFlush(t);
+    else
+        t->verdict_cache.len++;
+    return 0;
+ flush:
+    /* can't cache. Flush current cache and signal caller it should send single verdict */
+    if (NFQVerdictCacheLen(t) > 0)
+        NFQVerdictCacheFlush(t);
+#endif
+    return -1;
+}
+
 static inline void NFQMutexInit(NFQQueueVars *nq)
 {
     char *active_runmode = RunmodeGetActive();
@@ -530,6 +625,14 @@ TmEcode NFQInitThread(NFQThreadVars *nfq_t, uint32_t queue_maxlen)
     }
 #endif
 
+#ifdef HAVE_NFQ_SET_VERDICT_BATCH
+    if (runmode_workers) {
+        nfq_q->verdict_cache.maxlen = nfq_config.batchcount;
+    } else if (nfq_config.batchcount) {
+        SCLogError(SC_ERR_INVALID_ARGUMENT, "nfq.batchcount is only valid in workers runmode.");
+    }
+#endif
+
     /* set a timeout to the socket so we can check for a signal
      * in case we don't get packets for a longer period. */
     tv.tv_sec = 1;
@@ -722,13 +825,16 @@ void *NFQGetThread(int number) {
 #ifndef OS_WIN32
 void NFQRecvPkt(NFQQueueVars *t, NFQThreadVars *tv) {
     int rv, ret;
+    int flag = NFQVerdictCacheLen(t) ? MSG_DONTWAIT : 0;
 
     /* XXX what happens on rv == 0? */
-    rv = recv(t->fd, tv->data, tv->datalen, 0);
+    rv = recv(t->fd, tv->data, tv->datalen, flag);
 
     if (rv < 0) {
         if (errno == EINTR || errno == EWOULDBLOCK) {
             /* no error on timeout */
+            if (flag)
+                NFQVerdictCacheFlush(t);
         } else {
 #ifdef COUNTERS
             NFQMutexLock(t);
@@ -929,6 +1035,12 @@ TmEcode NFQSetVerdict(Packet *p) {
 #endif /* COUNTERS */
     }
 
+    ret = NFQVerdictCacheAdd(t, p, verdict);
+    if (ret == 0) {
+        NFQMutexUnlock(t);
+        return TM_ECODE_OK;
+    }
+
     do {
         switch (nfq_config.mode) {
             default:

diff --git a/src/source-nfq.h b/src/source-nfq.h
index a1f1249e0c191717fa78881453dce0f34f01d1d7..b392add2094e70380f3ec123f8f987e35db31805 100644 (file)
--- a/src/source-nfq.h
+++ b/src/source-nfq.h
@@ -80,6 +80,14 @@ typedef struct NFQQueueVars_
     uint32_t accepted;
     uint32_t dropped;
     uint32_t replaced;
+    struct {
+        uint32_t packet_id; /* id of last processed packet */
+        uint32_t verdict;
+        uint32_t mark;
+        uint8_t mark_valid:1;
+        uint8_t len;
+        uint8_t maxlen;
+    } verdict_cache;
 
 } NFQQueueVars;
 

diff --git a/suricata.yaml.in b/suricata.yaml.in
index ebf09c6bb1218b234b7498c1997d5d5cd36ccbb4..88f521f4a65cd93c22819897afc1611fbcbe93a9 100644 (file)
--- a/suricata.yaml.in
+++ b/suricata.yaml.in
@@ -220,6 +220,8 @@ magic-file: @e_magic_file@
 # this mode, you need to set mode to 'repeat'
 # If you want packet to be sent to another queue after an ACCEPT decision
 # set mode to 'route' and set next-queue value.
+# On linux >= 3.1, you can set batchcount to a value > 1 to improve performance
+# by processing several packets before sending a verdict (worker runmode only).
 # On linux >= 3.6, you can set the fail-open option to yes to have the kernel
 # accept the packet if suricata is not able to keep pace.
 nfq:
@@ -227,6 +229,7 @@ nfq:
 #  repeat-mark: 1
 #  repeat-mask: 1
 #  route-queue: 2
+#  batchcount: 20
 #  fail-open: yes
 
 # af-packet support