suricata
tmqh-flow.c
Go to the documentation of this file.
1 /* Copyright (C) 2007-2020 Open Information Security Foundation
2  *
3  * You can copy, redistribute or modify this Program under the terms of
4  * the GNU General Public License version 2 as published by the Free
5  * Software Foundation.
6  *
7  * This program is distributed in the hope that it will be useful,
8  * but WITHOUT ANY WARRANTY; without even the implied warranty of
9  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10  * GNU General Public License for more details.
11  *
12  * You should have received a copy of the GNU General Public License
13  * version 2 along with this program; if not, write to the Free Software
14  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
15  * 02110-1301, USA.
16  */
17 
18 /**
19  * \file
20  *
21  * \author Victor Julien <victor@inliniac.net>
22  * \author Anoop Saldanha <anoopsaldanha@gmail.com>
23  *
24  * Simple output queue handler that makes sure all packets of the same flow
25  * are sent to the same queue. We support different kind of q handlers. Have
26  * a look at "autofp-scheduler" conf to further understand the various q
27  * handlers we provide.
28  */
29 
30 #include "suricata.h"
31 #include "packet-queue.h"
32 #include "decode.h"
33 #include "threads.h"
34 #include "threadvars.h"
35 #include "tmqh-flow.h"
36 #include "flow-hash.h"
37 
38 #include "tm-queuehandlers.h"
39 
40 #include "conf.h"
41 
42 Packet *TmqhInputFlow(ThreadVars *t);
43 void TmqhOutputFlowHash(ThreadVars *t, Packet *p);
44 void TmqhOutputFlowIPPair(ThreadVars *t, Packet *p);
45 static void TmqhOutputFlowFTPHash(ThreadVars *t, Packet *p);
46 void *TmqhOutputFlowSetupCtx(const char *queue_str);
47 void TmqhOutputFlowFreeCtx(void *ctx);
48 void TmqhFlowRegisterTests(void);
49 
50 void TmqhFlowRegister(void)
51 {
52  tmqh_table[TMQH_FLOW].name = "flow";
53  tmqh_table[TMQH_FLOW].InHandler = TmqhInputFlow;
54  tmqh_table[TMQH_FLOW].OutHandlerCtxSetup = TmqhOutputFlowSetupCtx;
55  tmqh_table[TMQH_FLOW].OutHandlerCtxFree = TmqhOutputFlowFreeCtx;
56  tmqh_table[TMQH_FLOW].RegisterTests = TmqhFlowRegisterTests;
57 
58  const char *scheduler = NULL;
59  if (SCConfGetNonNull("autofp-scheduler", &scheduler) == 1) {
60  if (strcasecmp(scheduler, "round-robin") == 0) {
61  SCLogNotice("using flow hash instead of round robin");
62  tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowHash;
63  } else if (strcasecmp(scheduler, "active-packets") == 0) {
64  SCLogNotice("using flow hash instead of active packets");
65  tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowHash;
66  } else if (strcasecmp(scheduler, "hash") == 0) {
67  tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowHash;
68  } else if (strcasecmp(scheduler, "ippair") == 0) {
69  tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowIPPair;
70  } else if (strcasecmp(scheduler, "ftp-hash") == 0) {
71  tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowFTPHash;
72  } else {
73  SCLogError("Invalid entry \"%s\" "
74  "for autofp-scheduler in conf. Killing engine.",
75  scheduler);
76  exit(EXIT_FAILURE);
77  }
78  } else {
79  tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowHash;
80  }
81 }
82 
83 void TmqhFlowPrintAutofpHandler(void)
84 {
85 #define PRINT_IF_FUNC(f, msg) \
86  if (tmqh_table[TMQH_FLOW].OutHandler == (f)) \
87  SCLogConfig("AutoFP mode using \"%s\" flow load balancer", (msg))
88 
89  PRINT_IF_FUNC(TmqhOutputFlowHash, "Hash");
90  PRINT_IF_FUNC(TmqhOutputFlowIPPair, "IPPair");
91  PRINT_IF_FUNC(TmqhOutputFlowFTPHash, "FTPHash");
92 
93 #undef PRINT_IF_FUNC
94 }
95 
96 /* same as 'simple' */
97 Packet *TmqhInputFlow(ThreadVars *tv)
98 {
99  PacketQueue *q = tv->inq->pq;
100 
101  StatsSyncCountersIfSignalled(&tv->stats);
102 
103  SCMutexLock(&q->mutex_q);
104  if (q->len == 0) {
105  /* if we have no packets in queue, wait... */
106  SCCondWait(&q->cond_q, &q->mutex_q);
107  }
108 
109  if (q->len > 0) {
110  Packet *p = PacketDequeue(q);
111  SCMutexUnlock(&q->mutex_q);
112  return p;
113  } else {
114  /* return NULL if we have no pkt. Should only happen on signals. */
115  SCMutexUnlock(&q->mutex_q);
116  return NULL;
117  }
118 }
119 
120 static int StoreQueueId(TmqhFlowCtx *ctx, char *name)
121 {
122  void *ptmp;
123  Tmq *tmq = TmqGetQueueByName(name);
124  if (tmq == NULL) {
125  tmq = TmqCreateQueue(name);
126  if (tmq == NULL)
127  return -1;
128  }
129  tmq->writer_cnt++;
130 
131  if (ctx->queues == NULL) {
132  ctx->size = 1;
133  ctx->queues = SCCalloc(1, ctx->size * sizeof(TmqhFlowMode));
134  if (ctx->queues == NULL) {
135  return -1;
136  }
137  } else {
138  ctx->size++;
139  ptmp = SCRealloc(ctx->queues, ctx->size * sizeof(TmqhFlowMode));
140  if (ptmp == NULL) {
141  SCFree(ctx->queues);
142  ctx->queues = NULL;
143  return -1;
144  }
145  ctx->queues = ptmp;
146 
147  memset(ctx->queues + (ctx->size - 1), 0, sizeof(TmqhFlowMode));
148  }
149  ctx->queues[ctx->size - 1].q = tmq->pq;
150 
151  return 0;
152 }
153 
154 /**
155  * \brief setup the queue handlers ctx
156  *
157  * Parses a comma separated string "queuename1,queuename2,etc"
158  * and sets the ctx up to devide flows over these queue's.
159  *
160  * \param queue_str comma separated string with output queue names
161  *
162  * \retval ctx queues handlers ctx or NULL in error
163  */
164 void *TmqhOutputFlowSetupCtx(const char *queue_str)
165 {
166  if (queue_str == NULL || strlen(queue_str) == 0)
167  return NULL;
168 
169  SCLogDebug("queue_str %s", queue_str);
170 
171  TmqhFlowCtx *ctx = SCCalloc(1, sizeof(TmqhFlowCtx));
172  if (unlikely(ctx == NULL))
173  return NULL;
174 
175  char *str = SCStrdup(queue_str);
176  if (unlikely(str == NULL)) {
177  goto error;
178  }
179  char *tstr = str;
180 
181  /* parse the comma separated string */
182  do {
183  char *comma = strchr(tstr,',');
184  if (comma != NULL) {
185  *comma = '\0';
186  char *qname = tstr;
187  int r = StoreQueueId(ctx,qname);
188  if (r < 0)
189  goto error;
190  } else {
191  char *qname = tstr;
192  int r = StoreQueueId(ctx,qname);
193  if (r < 0)
194  goto error;
195  }
196  tstr = comma ? (comma + 1) : comma;
197  } while (tstr != NULL);
198 
199  SCFree(str);
200  return (void *)ctx;
201 
202 error:
203  SCFree(ctx);
204  if (str != NULL)
205  SCFree(str);
206  return NULL;
207 }
208 
209 void TmqhOutputFlowFreeCtx(void *ctx)
210 {
211  TmqhFlowCtx *fctx = (TmqhFlowCtx *)ctx;
212 
213  SCLogPerf("AutoFP - Total flow handler queues - %" PRIu16,
214  fctx->size);
215  SCFree(fctx->queues);
216  SCFree(fctx);
217 }
218 
219 void TmqhOutputFlowHash(ThreadVars *tv, Packet *p)
220 {
221  uint32_t qid;
222  TmqhFlowCtx *ctx = (TmqhFlowCtx *)tv->outctx;
223 
224  if (p->flags & PKT_WANTS_FLOW) {
225  uint32_t hash = p->flow_hash;
226  qid = hash % ctx->size;
227  } else {
228  qid = ctx->last++;
229 
230  if (ctx->last == ctx->size)
231  ctx->last = 0;
232  }
233 
234  PacketQueue *q = ctx->queues[qid].q;
235  SCMutexLock(&q->mutex_q);
236  PacketEnqueue(q, p);
237  SCCondSignal(&q->cond_q);
238  SCMutexUnlock(&q->mutex_q);
239 }
240 
241 /**
242  * \brief select the queue to output based on IP address pair.
243  *
244  * \param tv thread vars.
245  * \param p packet.
246  */
247 void TmqhOutputFlowIPPair(ThreadVars *tv, Packet *p)
248 {
249  uint32_t addr_hash = 0;
250 
251  TmqhFlowCtx *ctx = (TmqhFlowCtx *)tv->outctx;
252 
253  if (p->src.family == AF_INET6) {
254  for (int i = 0; i < 4; i++) {
255  addr_hash += p->src.addr_data32[i] + p->dst.addr_data32[i];
256  }
257  } else {
258  addr_hash = p->src.addr_data32[0] + p->dst.addr_data32[0];
259  }
260 
261  uint32_t qid = addr_hash % ctx->size;
262  PacketQueue *q = ctx->queues[qid].q;
263  SCMutexLock(&q->mutex_q);
264  PacketEnqueue(q, p);
265  SCCondSignal(&q->cond_q);
266  SCMutexUnlock(&q->mutex_q);
267 }
268 
269 static void TmqhOutputFlowFTPHash(ThreadVars *tv, Packet *p)
270 {
271  uint32_t qid;
272  TmqhFlowCtx *ctx = (TmqhFlowCtx *)tv->outctx;
273 
274  if (p->flags & PKT_WANTS_FLOW) {
275  uint32_t hash = p->flow_hash;
276  if (PacketIsTCP(p) && ((p->sp >= 1024 && p->dp >= 1024) || p->dp == 21 || p->sp == 21 ||
277  p->dp == 20 || p->sp == 20)) {
278  hash = FlowGetIpPairProtoHash(p);
279  }
280  qid = hash % ctx->size;
281  } else {
282  qid = ctx->last++;
283 
284  if (ctx->last == ctx->size)
285  ctx->last = 0;
286  }
287 
288  PacketQueue *q = ctx->queues[qid].q;
289  SCMutexLock(&q->mutex_q);
290  PacketEnqueue(q, p);
291  SCCondSignal(&q->cond_q);
292  SCMutexUnlock(&q->mutex_q);
293 }
294 
295 #ifdef UNITTESTS
296 
297 static int TmqhOutputFlowSetupCtxTest01(void)
298 {
299  TmqResetQueues();
300 
301  Tmq *tmq1 = TmqCreateQueue("queue1");
302  FAIL_IF_NULL(tmq1);
303  Tmq *tmq2 = TmqCreateQueue("queue2");
304  FAIL_IF_NULL(tmq2);
305  Tmq *tmq3 = TmqCreateQueue("another");
306  FAIL_IF_NULL(tmq3);
307  Tmq *tmq4 = TmqCreateQueue("yetanother");
308  FAIL_IF_NULL(tmq4);
309 
310  const char *str = "queue1,queue2,another,yetanother";
311  void *ctx = TmqhOutputFlowSetupCtx(str);
312  FAIL_IF_NULL(ctx);
313 
314  TmqhFlowCtx *fctx = (TmqhFlowCtx *)ctx;
315 
316  FAIL_IF_NOT(fctx->size == 4);
317 
318  FAIL_IF_NULL(fctx->queues);
319 
320  FAIL_IF_NOT(fctx->queues[0].q == tmq1->pq);
321  FAIL_IF_NOT(fctx->queues[1].q == tmq2->pq);
322  FAIL_IF_NOT(fctx->queues[2].q == tmq3->pq);
323  FAIL_IF_NOT(fctx->queues[3].q == tmq4->pq);
324 
325  TmqhOutputFlowFreeCtx(fctx);
326  TmqResetQueues();
327  PASS;
328 }
329 
330 static int TmqhOutputFlowSetupCtxTest02(void)
331 {
332  TmqResetQueues();
333 
334  Tmq *tmq1 = TmqCreateQueue("queue1");
335  FAIL_IF_NULL(tmq1);
336  Tmq *tmq2 = TmqCreateQueue("queue2");
337  FAIL_IF_NULL(tmq2);
338  Tmq *tmq3 = TmqCreateQueue("another");
339  FAIL_IF_NULL(tmq3);
340  Tmq *tmq4 = TmqCreateQueue("yetanother");
341  FAIL_IF_NULL(tmq4);
342 
343  const char *str = "queue1";
344  void *ctx = TmqhOutputFlowSetupCtx(str);
345  FAIL_IF_NULL(ctx);
346 
347  TmqhFlowCtx *fctx = (TmqhFlowCtx *)ctx;
348 
349  FAIL_IF_NOT(fctx->size == 1);
350 
351  FAIL_IF_NULL(fctx->queues);
352 
353  FAIL_IF_NOT(fctx->queues[0].q == tmq1->pq);
354  TmqhOutputFlowFreeCtx(fctx);
355  TmqResetQueues();
356 
357  PASS;
358 }
359 
360 static int TmqhOutputFlowSetupCtxTest03(void)
361 {
362  TmqResetQueues();
363 
364  const char *str = "queue1,queue2,another,yetanother";
365  void *ctx = TmqhOutputFlowSetupCtx(str);
366  FAIL_IF_NULL(ctx);
367 
368  TmqhFlowCtx *fctx = (TmqhFlowCtx *)ctx;
369 
370  FAIL_IF_NOT(fctx->size == 4);
371 
372  FAIL_IF_NULL(fctx->queues);
373 
374  Tmq *tmq1 = TmqGetQueueByName("queue1");
375  FAIL_IF_NULL(tmq1);
376  Tmq *tmq2 = TmqGetQueueByName("queue2");
377  FAIL_IF_NULL(tmq2);
378  Tmq *tmq3 = TmqGetQueueByName("another");
379  FAIL_IF_NULL(tmq3);
380  Tmq *tmq4 = TmqGetQueueByName("yetanother");
381  FAIL_IF_NULL(tmq4);
382 
383  FAIL_IF_NOT(fctx->queues[0].q == tmq1->pq);
384  FAIL_IF_NOT(fctx->queues[1].q == tmq2->pq);
385  FAIL_IF_NOT(fctx->queues[2].q == tmq3->pq);
386  FAIL_IF_NOT(fctx->queues[3].q == tmq4->pq);
387 
388  TmqhOutputFlowFreeCtx(fctx);
389  TmqResetQueues();
390  PASS;
391 }
392 
393 #endif /* UNITTESTS */
394 
395 void TmqhFlowRegisterTests(void)
396 {
397 #ifdef UNITTESTS
398  UtRegisterTest("TmqhOutputFlowSetupCtxTest01",
399  TmqhOutputFlowSetupCtxTest01);
400  UtRegisterTest("TmqhOutputFlowSetupCtxTest02",
401  TmqhOutputFlowSetupCtxTest02);
402  UtRegisterTest("TmqhOutputFlowSetupCtxTest03",
403  TmqhOutputFlowSetupCtxTest03);
404 #endif
405 }
Tmq_::writer_cnt
uint16_t writer_cnt
Definition: tm-queues.h:35
FAIL_IF_NULL
#define FAIL_IF_NULL(expr)
Fail a test if expression evaluates to NULL.
Definition: util-unittest.h:89
StatsSyncCountersIfSignalled
void StatsSyncCountersIfSignalled(StatsThreadContext *stats)
Definition: counters.c:481
TmqhOutputFlowIPPair
void TmqhOutputFlowIPPair(ThreadVars *t, Packet *p)
select the queue to output based on IP address pair.
Definition: tmqh-flow.c:247
unlikely
#define unlikely(expr)
Definition: util-optimize.h:35
UtRegisterTest
void UtRegisterTest(const char *name, int(*TestFn)(void))
Register unit test.
Definition: util-unittest.c:101
TmqhFlowPrintAutofpHandler
void TmqhFlowPrintAutofpHandler(void)
Definition: tmqh-flow.c:83
ThreadVars_::outctx
void * outctx
Definition: threadvars.h:103
SCLogDebug
#define SCLogDebug(...)
Definition: util-debug.h:282
PacketEnqueue
void PacketEnqueue(PacketQueue *q, Packet *p)
Definition: packet-queue.c:175
name
const char * name
Definition: detect-engine-proto.c:47
PacketQueue_
simple fifo queue for packets with mutex and cond Calling the mutex or triggering the cond is respons...
Definition: packet-queue.h:49
Packet_::flags
uint32_t flags
Definition: decode.h:562
threads.h
Tmq_::pq
PacketQueue * pq
Definition: tm-queues.h:36
TmqhFlowRegisterTests
void TmqhFlowRegisterTests(void)
Definition: tmqh-flow.c:395
ctx
struct Thresholds ctx
Tmqh_::OutHandler
void(* OutHandler)(ThreadVars *, Packet *)
Definition: tm-queuehandlers.h:40
TmqhFlowMode_
Definition: tmqh-flow.h:27
PKT_WANTS_FLOW
#define PKT_WANTS_FLOW
Definition: decode.h:1341
flow-hash.h
packet-queue.h
SCMutexLock
#define SCMutexLock(mut)
Definition: threads-debug.h:117
p
Packet * p
Definition: fuzz_dataset.c:30
FAIL_IF_NOT
#define FAIL_IF_NOT(expr)
Fail a test if expression evaluates to false.
Definition: util-unittest.h:82
TmqhFlowRegister
void TmqhFlowRegister(void)
Definition: tmqh-flow.c:50
Tmqh_::InHandler
Packet *(* InHandler)(ThreadVars *)
Definition: tm-queuehandlers.h:38
TmqhOutputFlowFreeCtx
void TmqhOutputFlowFreeCtx(void *ctx)
Definition: tmqh-flow.c:209
PacketQueue_::mutex_q
SCMutex mutex_q
Definition: packet-queue.h:56
TMQH_FLOW
@ TMQH_FLOW
Definition: tm-queuehandlers.h:31
decode.h
TmqhFlowCtx_::size
uint16_t size
Definition: tmqh-flow.h:35
PASS
#define PASS
Pass the test.
Definition: util-unittest.h:105
TmqhFlowCtx_::queues
TmqhFlowMode * queues
Definition: tmqh-flow.h:38
SCCondWait
#define SCCondWait
Definition: threads-debug.h:142
SCMutexUnlock
#define SCMutexUnlock(mut)
Definition: threads-debug.h:120
SCConfGetNonNull
int SCConfGetNonNull(const char *name, const char **vptr)
Retrieve the non-null value of a configuration node.
Definition: conf.c:380
ThreadVars_
Per thread variable structure.
Definition: threadvars.h:57
TmqGetQueueByName
Tmq * TmqGetQueueByName(const char *name)
Definition: tm-queues.c:60
Tmqh_::RegisterTests
void(* RegisterTests)(void)
Definition: tm-queuehandlers.h:43
Packet_::sp
Port sp
Definition: decode.h:523
TmqhOutputFlowSetupCtx
void * TmqhOutputFlowSetupCtx(const char *queue_str)
setup the queue handlers ctx
Definition: tmqh-flow.c:164
Tmqh_::OutHandlerCtxFree
void(* OutHandlerCtxFree)(void *)
Definition: tm-queuehandlers.h:42
TmqhFlowMode_::q
PacketQueue * q
Definition: tmqh-flow.h:28
Packet_
Definition: decode.h:516
Tmqh_::name
const char * name
Definition: tm-queuehandlers.h:37
conf.h
TmqhOutputFlowHash
void TmqhOutputFlowHash(ThreadVars *t, Packet *p)
Definition: tmqh-flow.c:219
PacketQueue_::cond_q
SCCondT cond_q
Definition: packet-queue.h:57
SCRealloc
#define SCRealloc(ptr, sz)
Definition: util-mem.h:50
FlowGetIpPairProtoHash
uint32_t FlowGetIpPairProtoHash(const Packet *p)
Definition: flow-hash.c:151
tm-queuehandlers.h
PRINT_IF_FUNC
#define PRINT_IF_FUNC(f, msg)
SCCondSignal
#define SCCondSignal
Definition: threads-debug.h:140
ThreadVars_::inq
Tmq * inq
Definition: threadvars.h:88
SCLogPerf
#define SCLogPerf(...)
Definition: util-debug.h:241
PacketQueue_::len
uint32_t len
Definition: packet-queue.h:52
SCStrdup
#define SCStrdup(s)
Definition: util-mem.h:56
tmqh_table
Tmqh tmqh_table[TMQH_SIZE]
Definition: tm-queuehandlers.c:37
tv
ThreadVars * tv
Definition: fuzz_decodepcapfile.c:34
PacketDequeue
Packet * PacketDequeue(PacketQueue *q)
Definition: packet-queue.c:216
threadvars.h
Packet_::flow_hash
uint32_t flow_hash
Definition: decode.h:568
str
#define str(s)
Definition: suricata-common.h:313
Tmqh_::OutHandlerCtxSetup
void *(* OutHandlerCtxSetup)(const char *)
Definition: tm-queuehandlers.h:41
SCLogError
#define SCLogError(...)
Macro used to log ERROR messages.
Definition: util-debug.h:274
TmqCreateQueue
Tmq * TmqCreateQueue(const char *name)
SCFree
#define SCFree(p)
Definition: util-mem.h:61
TmqhInputFlow
Packet * TmqhInputFlow(ThreadVars *t)
Definition: tmqh-flow.c:97
tmqh-flow.h
suricata.h
Address_::family
char family
Definition: decode.h:114
Packet_::dst
Address dst
Definition: decode.h:521
Tmq_
Definition: tm-queues.h:30
SCLogNotice
#define SCLogNotice(...)
Macro used to log NOTICE messages.
Definition: util-debug.h:250
Packet_::dp
Port dp
Definition: decode.h:531
SCCalloc
#define SCCalloc(nm, sz)
Definition: util-mem.h:53
ThreadVars_::stats
StatsThreadContext stats
Definition: threadvars.h:120
TmqhFlowCtx_
Ctx for the flow queue handler.
Definition: tmqh-flow.h:34
Packet_::src
Address src
Definition: decode.h:520
TmqResetQueues
void TmqResetQueues(void)
Definition: tm-queues.c:81