From ee9714e593dae89fabd3ba823b762a8f3ace9d78 Mon Sep 17 00:00:00 2001 From: Angelo Mirabella Date: Wed, 3 Apr 2024 01:06:39 -0700 Subject: [PATCH] libsuricata: add library runmode Add library source and runmode modules. Reorganized library example to create a worker thread and replay a pcap file using the library mode. No API layer is added at this stage. Edits by Jason Ish: - fix guard - add copyright/license headers Ticket: #7240 --- examples/lib/simple/Makefile.example.in | 8 +- examples/lib/simple/main.c | 75 +++++++++- src/Makefile.am | 4 + src/runmode-lib.c | 154 ++++++++++++++++++++ src/runmode-lib.h | 57 ++++++++ src/runmodes.c | 7 + src/runmodes.h | 1 + src/source-lib.c | 180 ++++++++++++++++++++++++ src/source-lib.h | 51 +++++++ src/suricata.c | 9 ++ src/suricata.h | 3 + src/tm-modules.c | 1 + src/tm-threads-common.h | 1 + src/tm-threads.c | 106 +++++++++++++- src/tm-threads.h | 2 + 15 files changed, 650 insertions(+), 9 deletions(-) create mode 100644 src/runmode-lib.c create mode 100644 src/runmode-lib.h create mode 100644 src/source-lib.c create mode 100644 src/source-lib.h diff --git a/examples/lib/simple/Makefile.example.in b/examples/lib/simple/Makefile.example.in index 724beccdb1..c9da97560e 100644 --- a/examples/lib/simple/Makefile.example.in +++ b/examples/lib/simple/Makefile.example.in @@ -1,12 +1,16 @@ LIBSURICATA_CONFIG ?= @CONFIGURE_PREFIX@/bin/libsuricata-config -SURICATA_LIBS = `$(LIBSURICATA_CONFIG) --libs` +SURICATA_LIBS = `$(LIBSURICATA_CONFIG) --libs --static` SURICATA_CFLAGS := `$(LIBSURICATA_CONFIG) --cflags` +# Currently the Suricata logging system requires this to be even for +# plugins. +CPPFLAGS += "-D__SCFILENAME__=\"$(*F)\"" + all: simple simple: main.c - $(CC) -o $@ $^ $(CFLAGS) $(SURICATA_CFLAGS) $(SURICATA_LIBS) + $(CC) -o $@ $^ $(CPPFLAGS) $(CFLAGS) $(SURICATA_CFLAGS) $(SURICATA_LIBS) clean: rm -f simple diff --git a/examples/lib/simple/main.c b/examples/lib/simple/main.c index f9c09fb0f5..7e6900e5f4 100644 --- a/examples/lib/simple/main.c +++ b/examples/lib/simple/main.c @@ -16,14 +16,62 @@ */ #include "suricata.h" +#include "conf.h" +#include "pcap.h" +#include "runmode-lib.h" +#include "source-lib.h" +#include "threadvars.h" + +/* Suricata worker thread in library mode. + The functions should be wrapped in an API layer. */ +static void *SimpleWorker(void *arg) +{ + char *pcap_file = (char *)arg; + + /* Create worker. */ + ThreadVars *tv = RunModeCreateWorker(); + if (!tv) { + pthread_exit(NULL); + } + + /* Start worker. */ + if (RunModeSpawnWorker(tv) != 0) { + pthread_exit(NULL); + } + + /* Replay pcap. */ + pcap_t *fp = pcap_open_offline(pcap_file, NULL); + if (fp == NULL) { + pthread_exit(NULL); + } + + int datalink = pcap_datalink(fp); + struct pcap_pkthdr pkthdr; + const u_char *packet; + while ((packet = pcap_next(fp, &pkthdr)) != NULL) { + if (TmModuleLibHandlePacket(tv, packet, datalink, pkthdr.ts, pkthdr.len, 0, 0, NULL) != 0) { + pthread_exit(NULL); + } + } + pcap_close(fp); + + /* Cleanup. */ + RunModeDestroyWorker(tv); + pthread_exit(NULL); +} int main(int argc, char **argv) { SuricataPreInit(argv[0]); /* Parse command line options. This is optional, you could - * directly configure Suricata through the Conf API. */ - SCParseCommandLine(argc, argv); + * directly configure Suricata through the Conf API. + The last argument is the PCAP file to replay. */ + SCParseCommandLine(argc - 1, argv); + + /* Set lib runmode. There is currently no way to set it via + the Conf API. */ + SuricataSetLibRunmode(); /* Validate/finalize the runmode. */ if (SCFinalizeRunMode() != TM_ECODE_OK) { @@ -47,12 +95,27 @@ int main(int argc, char **argv) exit(EXIT_FAILURE); } + /* Set "offline" runmode to replay a pcap in library mode. */ + if (!ConfSetFromString("runmode=offline", 1)) { + exit(EXIT_FAILURE); + } + SuricataInit(); - SuricataPostInit(); - /* Suricata is now running, but we enter a loop to keep it running - * until it shouldn't be running anymore. */ - SuricataMainLoop(); + /* Create and start worker on its own thread, passing the PCAP file + as argument. This needs to be done in between SuricataInit and + SuricataPostInit. */ + pthread_t worker; + if (pthread_create(&worker, NULL, SimpleWorker, argv[argc - 1]) != 0) { + exit(EXIT_FAILURE); + } + + /* Need to introduce a little sleep to allow the worker thread to + initialize before SuricataPostInit invokes TmThreadContinueThreads(). + This should be handle at the API level. */ + usleep(100); + + SuricataPostInit(); /* Shutdown engine. */ SuricataShutdown(); diff --git a/src/Makefile.am b/src/Makefile.am index 09ee015d23..849c716337 100755 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -405,6 +405,7 @@ noinst_HEADERS = \ runmode-erf-dag.h \ runmode-erf-file.h \ runmode-ipfw.h \ + runmode-lib.h \ runmode-netmap.h \ runmode-nflog.h \ runmode-nfq.h \ @@ -422,6 +423,7 @@ noinst_HEADERS = \ source-erf-dag.h \ source-erf-file.h \ source-ipfw.h \ + source-lib.h \ source-netmap.h \ source-nflog.h \ source-nfq.h \ @@ -983,6 +985,7 @@ libsuricata_c_a_SOURCES = \ runmode-erf-dag.c \ runmode-erf-file.c \ runmode-ipfw.c \ + runmode-lib.c \ runmode-netmap.c \ runmode-nflog.c \ runmode-nfq.c \ @@ -999,6 +1002,7 @@ libsuricata_c_a_SOURCES = \ source-erf-dag.c \ source-erf-file.c \ source-ipfw.c \ + source-lib.c \ source-netmap.c \ source-nflog.c \ source-nfq.c \ diff --git a/src/runmode-lib.c b/src/runmode-lib.c new file mode 100644 index 0000000000..1441dabe4b --- /dev/null +++ b/src/runmode-lib.c @@ -0,0 +1,154 @@ +/* Copyright (C) 2023-2024 Open Information Security Foundation + * + * You can copy, redistribute or modify this Program under the terms of + * the GNU General Public License version 2 as published by the Free + * Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * version 2 along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301, USA. + */ + +/** \file + * + * \author Angelo Mirabella + * + * Library runmode. + */ +#include "suricata-common.h" +#include "runmode-lib.h" +#include "runmodes.h" +#include "tm-threads.h" +#include "util-device.h" + +static int g_thread_id = 0; + +/** \brief register runmodes for suricata as a library */ +void RunModeIdsLibRegister(void) +{ + RunModeRegisterNewRunMode(RUNMODE_LIB, "offline", "Library offline mode (pcap replaying)", + RunModeIdsLibOffline, NULL); + RunModeRegisterNewRunMode(RUNMODE_LIB, "live", "Library live mode", RunModeIdsLibLive, NULL); + return; +} + +/** \brief runmode for offline packet processing (pcap files) */ +int RunModeIdsLibOffline(void) +{ + TimeModeSetOffline(); + + return 0; +} + +/** \brief runmode for live packet processing */ +int RunModeIdsLibLive(void) +{ + TimeModeSetLive(); + + return 0; +} + +const char *RunModeLibGetDefaultMode(void) +{ + return "live"; +} + +/** \brief create a "fake" worker thread in charge of processing the packets. + * + * This method just creates a context representing the worker, which is handled from the library + * client. No actual thread (pthread_t) is created. + * + * \return Pointer to ThreadVars structure representing the worker thread */ +void *RunModeCreateWorker(void) +{ + char tname[TM_THREAD_NAME_MAX]; + TmModule *tm_module = NULL; + snprintf(tname, sizeof(tname), "%s#%02d", thread_name_workers, ++g_thread_id); + + ThreadVars *tv = TmThreadCreatePacketHandler( + tname, "packetpool", "packetpool", "packetpool", "packetpool", "lib"); + if (tv == NULL) { + SCLogError("TmThreadsCreate failed"); + return NULL; + } + + tm_module = TmModuleGetByName("DecodeLib"); + if (tm_module == NULL) { + SCLogError("TmModuleGetByName DecodeLib failed"); + return NULL; + } + TmSlotSetFuncAppend(tv, tm_module, NULL); + + tm_module = TmModuleGetByName("FlowWorker"); + if (tm_module == NULL) { + SCLogError("TmModuleGetByName for FlowWorker failed"); + return NULL; + } + TmSlotSetFuncAppend(tv, tm_module, NULL); + + TmThreadAppend(tv, tv->type); + + return tv; +} + +/** \brief start the "fake" worker. + * + * This method performs all the initialization tasks. + */ +int RunModeSpawnWorker(void *td) +{ + ThreadVars *tv = (ThreadVars *)td; + + if (TmThreadLibSpawn(tv) != TM_ECODE_OK) { + SCLogError("TmThreadLibSpawn failed"); + return -1; + } + + TmThreadsSetFlag(tv, THV_RUNNING); + return 0; +} + +/** \brief destroy a worker thread */ +void RunModeDestroyWorker(void *td) +{ + ThreadVars *tv = (ThreadVars *)td; + TmSlot *s = tv->tm_slots; + TmEcode r; + TmSlot *slot = NULL; + + StatsSyncCounters(tv); + + TmThreadsSetFlag(tv, THV_FLOW_LOOP); + + /* process all pseudo packets the flow timeout may throw at us */ + TmThreadTimeoutLoop(tv, s); + + TmThreadsSetFlag(tv, THV_RUNNING_DONE); + TmThreadWaitForFlag(tv, THV_DEINIT); + + PacketPoolDestroy(); + + for (slot = s; slot != NULL; slot = slot->slot_next) { + if (slot->SlotThreadExitPrintStats != NULL) { + slot->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(slot->slot_data)); + } + + if (slot->SlotThreadDeinit != NULL) { + r = slot->SlotThreadDeinit(tv, SC_ATOMIC_GET(slot->slot_data)); + if (r != TM_ECODE_OK) { + break; + } + } + } + + tv->stream_pq = NULL; + --g_thread_id; + SCLogDebug("%s ending", tv->name); + TmThreadsSetFlag(tv, THV_CLOSED); +} diff --git a/src/runmode-lib.h b/src/runmode-lib.h new file mode 100644 index 0000000000..c33cb4e65b --- /dev/null +++ b/src/runmode-lib.h @@ -0,0 +1,57 @@ +/* Copyright (C) 2023-2024 Open Information Security Foundation + * + * You can copy, redistribute or modify this Program under the terms of + * the GNU General Public License version 2 as published by the Free + * Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * version 2 along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301, USA. + */ + +/** \file + * + * \author Angelo Mirabella + * + * Library runmode. + */ + +#ifndef SURICATA_RUNMODE_LIB_H +#define SURICATA_RUNMODE_LIB_H + +/** \brief register runmodes for suricata as a library */ +void RunModeIdsLibRegister(void); + +/** \brief runmode for live packet processing */ +int RunModeIdsLibLive(void); + +/** \brief runmode for offline packet processing (pcap files) */ +int RunModeIdsLibOffline(void); + +/** \brief runmode default mode (live) */ +const char *RunModeLibGetDefaultMode(void); + +/** \brief create a "fake" worker thread in charge of processing the packets. + * + * This method just creates a context representing the worker, which is handled from the library + * client. No actual thread (pthread_t) is created. + * + * \return Pointer to ThreadVars structure representing the worker thread */ +void *RunModeCreateWorker(void); + +/** \brief start the "fake" worker. + * + * This method performs all the initialization tasks. + */ +int RunModeSpawnWorker(void *); + +/** \brief destroy a worker thread */ +void RunModeDestroyWorker(void *); + +#endif /* SURICATA_RUNMODE_LIB_H */ diff --git a/src/runmodes.c b/src/runmodes.c index 87f1e5cef4..d478e1c658 100644 --- a/src/runmodes.c +++ b/src/runmodes.c @@ -36,6 +36,7 @@ #include "runmode-erf-dag.h" #include "runmode-erf-file.h" #include "runmode-ipfw.h" +#include "runmode-lib.h" #include "runmode-netmap.h" #include "runmode-nflog.h" #include "runmode-nfq.h" @@ -160,6 +161,8 @@ static const char *RunModeTranslateModeToName(int runmode) #else return "DPDK(DISABLED)"; #endif + case RUNMODE_LIB: + return "LIB"; default: FatalError("Unknown runtime mode. Aborting"); @@ -232,6 +235,7 @@ void RunModeRegisterRunModes(void) RunModeUnixSocketRegister(); RunModeIpsWinDivertRegister(); RunModeDpdkRegister(); + RunModeIdsLibRegister(); #ifdef UNITTESTS UtRunModeRegister(); #endif @@ -344,6 +348,9 @@ static const char *RunModeGetConfOrDefault(int capture_mode, const char *capture custom_mode = RunModeDpdkGetDefaultMode(); break; #endif + case RUNMODE_LIB: + custom_mode = RunModeLibGetDefaultMode(); + break; default: return NULL; } diff --git a/src/runmodes.h b/src/runmodes.h index 18fd5886d3..4466d69b9e 100644 --- a/src/runmodes.h +++ b/src/runmodes.h @@ -37,6 +37,7 @@ enum RunModes { RUNMODE_AFXDP_DEV, RUNMODE_NETMAP, RUNMODE_DPDK, + RUNMODE_LIB, RUNMODE_UNITTEST, RUNMODE_UNIX_SOCKET, RUNMODE_WINDIVERT, diff --git a/src/source-lib.c b/src/source-lib.c new file mode 100644 index 0000000000..6c304fc300 --- /dev/null +++ b/src/source-lib.c @@ -0,0 +1,180 @@ +/* Copyright (C) 2023-2024 Open Information Security Foundation + * + * You can copy, redistribute or modify this Program under the terms of + * the GNU General Public License version 2 as published by the Free + * Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * version 2 along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301, USA. + */ + +/** \file + * + * \author Angelo Mirabella + * + * LIB packet and stream decoding support + * + */ + +#include "suricata-common.h" +#include "source-lib.h" +#include "util-device.h" + +static TmEcode DecodeLibThreadInit(ThreadVars *tv, const void *initdata, void **data); +static TmEcode DecodeLibThreadDeinit(ThreadVars *tv, void *data); +static TmEcode DecodeLib(ThreadVars *tv, Packet *p, void *data); + +/* Set time to the first packet timestamp when replaying a PCAP. */ +static bool time_set = false; + +/** \brief register a "Decode" module for suricata as a library. + * + * The "Decode" module is the first module invoked when processing a packet */ +void TmModuleDecodeLibRegister(void) +{ + tmm_modules[TMM_DECODELIB].name = "DecodeLib"; + tmm_modules[TMM_DECODELIB].ThreadInit = DecodeLibThreadInit; + tmm_modules[TMM_DECODELIB].Func = DecodeLib; + tmm_modules[TMM_DECODELIB].ThreadExitPrintStats = NULL; + tmm_modules[TMM_DECODELIB].ThreadDeinit = DecodeLibThreadDeinit; + tmm_modules[TMM_DECODELIB].cap_flags = 0; + tmm_modules[TMM_DECODELIB].flags = TM_FLAG_DECODE_TM; +} + +/** \brief initialize the "Decode" module. + * + * \param tv Pointer to the per-thread structure. + * \param initdata Pointer to initialization context. + * \param data Pointer to the initialized context. + * \return Error code. + */ +TmEcode DecodeLibThreadInit(ThreadVars *tv, const void *initdata, void **data) +{ + SCEnter(); + DecodeThreadVars *dtv = NULL; + + dtv = DecodeThreadVarsAlloc(tv); + + if (dtv == NULL) + SCReturnInt(TM_ECODE_FAILED); + + DecodeRegisterPerfCounters(dtv, tv); + + *data = (void *)dtv; + + SCReturnInt(TM_ECODE_OK); +} + +/** \brief deinitialize the "Decode" module. + * + * \param tv Pointer to the per-thread structure. + * \param data Pointer to the context. + * \return Error code. + */ +TmEcode DecodeLibThreadDeinit(ThreadVars *tv, void *data) +{ + if (data != NULL) + DecodeThreadVarsFree(tv, data); + + time_set = false; + SCReturnInt(TM_ECODE_OK); +} + +/** \brief main decoding function. + * + * This method receives a packet and tries to identify layer 2 to 4 layers. + * + * \param tv Pointer to the per-thread structure. + * \param p Pointer to the packet. + * \param data Pointer to the context. + * \return Error code. + */ +TmEcode DecodeLib(ThreadVars *tv, Packet *p, void *data) +{ + SCEnter(); + DecodeThreadVars *dtv = (DecodeThreadVars *)data; + + BUG_ON(PKT_IS_PSEUDOPKT(p)); + + /* update counters */ + DecodeUpdatePacketCounters(tv, dtv, p); + + /* If suri has set vlan during reading, we increase vlan counter */ + if (p->vlan_idx) { + StatsIncr(tv, dtv->counter_vlan); + } + + /* call the decoder */ + DecodeLinkLayer(tv, dtv, p->datalink, p, GET_PKT_DATA(p), GET_PKT_LEN(p)); + + PacketDecodeFinalize(tv, dtv, p); + + SCReturnInt(TM_ECODE_OK); +} + +/** \brief process a single packet. + * + * \param tv Pointer to the per-thread structure. + * \param data Pointer to the raw packet. + * \param datalink Datalink type. + * \param ts Timeval structure. + * \param len Packet length. + * \param tenant_id Tenant id of the detection engine to use. + * \param flags Packet flags (packet checksum, rule profiling...). + * \param iface Sniffing interface this packet comes from (can be NULL). + * \return Error code. + */ +int TmModuleLibHandlePacket(ThreadVars *tv, const uint8_t *data, int datalink, struct timeval ts, + uint32_t len, uint32_t tenant_id, uint32_t flags, const char *iface) +{ + + /* If the packet is NULL, consider it as a read timeout. */ + if (data == NULL) { + TmThreadsSetFlag(tv, THV_CAPTURE_INJECT_PKT); + TmThreadsCaptureHandleTimeout(tv, NULL); + SCReturnInt(TM_ECODE_OK); + } + + Packet *p = PacketGetFromQueueOrAlloc(); + if (unlikely(p == NULL)) { + SCReturnInt(TM_ECODE_FAILED); + } + + /* If we are processing a PCAP and it is the first packet we need to set the timestamp. */ + SCTime_t timestamp = SCTIME_FROM_TIMEVAL(&ts); + if (!time_set && !TimeModeIsLive()) { + TmThreadsInitThreadsTimestamp(timestamp); + time_set = true; + } + + PKT_SET_SRC(p, PKT_SRC_WIRE); + p->ts = timestamp; + p->datalink = datalink; + p->tenant_id = tenant_id; + p->flags |= flags; + + /* Set the sniffing interface. */ + if (iface) { + p->livedev = LiveGetDevice(iface); + } + + if (PacketSetData(p, data, len) == -1) { + TmqhOutputPacketpool(tv, p); + SCReturnInt(TM_ECODE_FAILED); + } + + SCLogDebug("pktlen: %" PRIu32 " (pkt %p, pkt data %p)", GET_PKT_LEN(p), p, GET_PKT_DATA(p)); + + if (TmThreadsSlotProcessPkt(tv, tv->tm_slots, p) != TM_ECODE_OK) { + SCReturnInt(TM_ECODE_FAILED); + } + + SCReturnInt(TM_ECODE_OK); +} diff --git a/src/source-lib.h b/src/source-lib.h new file mode 100644 index 0000000000..3f0594bd8e --- /dev/null +++ b/src/source-lib.h @@ -0,0 +1,51 @@ +/* Copyright (C) 2023-2024 Open Information Security Foundation + * + * You can copy, redistribute or modify this Program under the terms of + * the GNU General Public License version 2 as published by the Free + * Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * version 2 along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301, USA. + */ + +/** \file + * + * \author Angelo Mirabella + * + * LIB packet and stream decoding support + * + */ + +#ifndef SURICATA_SOURCE_LIB_H +#define SURICATA_SOURCE_LIB_H + +#include "tm-threads.h" + +/** \brief register a "Decode" module for suricata as a library. + * + * The "Decode" module is the first module invoked when processing a packet */ +void TmModuleDecodeLibRegister(void); + +/** \brief process a single packet. + * + * \param tv Pointer to the per-thread structure. + * \param data Pointer to the raw packet. + * \param datalink Datalink type. + * \param ts Timeval structure. + * \param len Packet length. + * \param tenant_id Tenant id of the detection engine to use. + * \param flags Packet flags (packet checksum, rule profiling...). + * \param iface Sniffing interface this packet comes from (can be NULL). + * \return Error code. + */ +int TmModuleLibHandlePacket(ThreadVars *tv, const uint8_t *data, int datalink, struct timeval ts, + uint32_t len, uint32_t tenant_id, uint32_t flags, const char *iface); + +#endif /* SURICATA_SOURCE_LIB_H */ diff --git a/src/suricata.c b/src/suricata.c index 9977c54d82..7c3b05f3f8 100644 --- a/src/suricata.c +++ b/src/suricata.c @@ -90,6 +90,7 @@ #include "source-nfq-prototypes.h" #include "source-nflog.h" #include "source-ipfw.h" +#include "source-lib.h" #include "source-pcap.h" #include "source-pcap-file.h" #include "source-pcap-file-helper.h" @@ -954,6 +955,9 @@ void RegisterAllModules(void) /* Dpdk */ TmModuleReceiveDPDKRegister(); TmModuleDecodeDPDKRegister(); + + /* Library */ + TmModuleDecodeLibRegister(); } TmEcode SCLoadYamlConfig(void) @@ -3063,3 +3067,8 @@ void SuricataPostInit(void) } SCPledge(); } + +void SuricataSetLibRunmode(void) +{ + suricata.run_mode = RUNMODE_LIB; +} diff --git a/src/suricata.h b/src/suricata.h index 2448cb9f73..311671cb79 100644 --- a/src/suricata.h +++ b/src/suricata.h @@ -227,4 +227,7 @@ int WindowsInitService(int argc, char **argv); const char *GetProgramVersion(void); +/* Library only methods. */ +void SuricataSetLibRunmode(void); + #endif /* SURICATA_SURICATA_H */ diff --git a/src/tm-modules.c b/src/tm-modules.c index 8f9a2022e1..df3dc3c808 100644 --- a/src/tm-modules.c +++ b/src/tm-modules.c @@ -182,6 +182,7 @@ const char * TmModuleTmmIdToString(TmmId id) CASE_CODE(TMM_DECODEPCAPFILE); CASE_CODE(TMM_RECEIVEDPDK); CASE_CODE(TMM_DECODEDPDK); + CASE_CODE(TMM_DECODELIB); CASE_CODE (TMM_RECEIVEPLUGIN); CASE_CODE (TMM_DECODEPLUGIN); CASE_CODE (TMM_RESPONDREJECT); diff --git a/src/tm-threads-common.h b/src/tm-threads-common.h index eb88168b02..cd2ace6740 100644 --- a/src/tm-threads-common.h +++ b/src/tm-threads-common.h @@ -64,6 +64,7 @@ typedef enum { TMM_RECEIVEWINDIVERT, TMM_VERDICTWINDIVERT, TMM_DECODEWINDIVERT, + TMM_DECODELIB, TMM_FLOWMANAGER, TMM_FLOWRECYCLER, diff --git a/src/tm-threads.c b/src/tm-threads.c index dca77281ce..75541f341c 100644 --- a/src/tm-threads.c +++ b/src/tm-threads.c @@ -163,7 +163,7 @@ TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p, TmSlot *slot) * is run until the flow engine kills the thread and the queue is * empty. */ -static int TmThreadTimeoutLoop(ThreadVars *tv, TmSlot *s) +int TmThreadTimeoutLoop(ThreadVars *tv, TmSlot *s) { TmSlot *fw_slot = tv->tm_flowworker; int r = TM_ECODE_OK; @@ -379,6 +379,89 @@ bool TmThreadsWaitForUnpause(ThreadVars *tv) return true; } +static void *TmThreadsLib(void *td) +{ + ThreadVars *tv = (ThreadVars *)td; + TmSlot *s = tv->tm_slots; + TmEcode r = TM_ECODE_OK; + TmSlot *slot = NULL; + + /* Set the thread name */ + SCSetThreadName(tv->name); + + if (tv->thread_setup_flags != 0) + TmThreadSetupOptions(tv); + + /* Drop the capabilities for this thread */ + SCDropCaps(tv); + + PacketPoolInit(); + + /* check if we are setup properly */ + if (s == NULL || tv->tmqh_in == NULL || tv->tmqh_out == NULL) { + SCLogError("TmSlot or ThreadVars badly setup: s=%p, tmqh_in=%p," + " tmqh_out=%p", + s, tv->tmqh_in, tv->tmqh_out); + TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); + return NULL; + } + + for (slot = s; slot != NULL; slot = slot->slot_next) { + if (slot->SlotThreadInit != NULL) { + void *slot_data = NULL; + r = slot->SlotThreadInit(tv, slot->slot_initdata, &slot_data); + if (r != TM_ECODE_OK) { + if (r == TM_ECODE_DONE) { + EngineDone(); + TmThreadsSetFlag(tv, THV_CLOSED | THV_INIT_DONE | THV_RUNNING_DONE); + goto error; + } else { + TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); + goto error; + } + } + (void)SC_ATOMIC_SET(slot->slot_data, slot_data); + } + + /* if the flowworker module is the first, get the threads input queue */ + if (slot == (TmSlot *)tv->tm_slots && (slot->tm_id == TMM_FLOWWORKER)) { + tv->stream_pq = tv->inq->pq; + tv->tm_flowworker = slot; + SCLogDebug("pre-stream packetqueue %p (inq)", tv->stream_pq); + tv->flow_queue = FlowQueueNew(); + if (tv->flow_queue == NULL) { + TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); + return NULL; + } + /* setup a queue */ + } else if (slot->tm_id == TMM_FLOWWORKER) { + tv->stream_pq_local = SCCalloc(1, sizeof(PacketQueue)); + if (tv->stream_pq_local == NULL) + FatalError("failed to alloc PacketQueue"); + SCMutexInit(&tv->stream_pq_local->mutex_q, NULL); + tv->stream_pq = tv->stream_pq_local; + tv->tm_flowworker = slot; + SCLogDebug("pre-stream packetqueue %p (local)", tv->stream_pq); + tv->flow_queue = FlowQueueNew(); + if (tv->flow_queue == NULL) { + TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); + return NULL; + } + } + } + StatsSetupPrivate(tv); + + TmThreadsSetFlag(tv, THV_INIT_DONE); + + TmThreadsWaitForUnpause(tv); + + return NULL; + +error: + tv->stream_pq = NULL; + return NULL; +} + static void *TmThreadsSlotVar(void *td) { ThreadVars *tv = (ThreadVars *)td; @@ -622,6 +705,8 @@ static TmEcode TmThreadSetSlots(ThreadVars *tv, const char *name, void *(*fn_p)( tv->tm_func = TmThreadsManagement; } else if (strcmp(name, "command") == 0) { tv->tm_func = TmThreadsManagement; + } else if (strcmp(name, "lib") == 0) { + tv->tm_func = TmThreadsLib; } else if (strcmp(name, "custom") == 0) { if (fn_p == NULL) goto error; @@ -1708,6 +1793,25 @@ TmEcode TmThreadSpawn(ThreadVars *tv) return TM_ECODE_OK; } +/** + * \brief Spawns a "fake" lib thread associated with the ThreadVars instance tv + * + * \retval TM_ECODE_OK on success and TM_ECODE_FAILED on failure + */ +TmEcode TmThreadLibSpawn(ThreadVars *tv) +{ + if (tv->tm_func == NULL) { + printf("ERROR: no thread function set\n"); + return TM_ECODE_FAILED; + } + + tv->tm_func((void *)tv); + + TmThreadWaitForFlag(tv, THV_INIT_DONE | THV_RUNNING_DONE); + + return TM_ECODE_OK; +} + /** * \brief Initializes the mutex and condition variables for this TV * diff --git a/src/tm-threads.h b/src/tm-threads.h index 9c644a31be..13ce78d51d 100644 --- a/src/tm-threads.h +++ b/src/tm-threads.h @@ -94,6 +94,8 @@ ThreadVars *TmThreadCreateMgmtThreadByName(const char *name, const char *module, ThreadVars *TmThreadCreateCmdThreadByName(const char *name, const char *module, int mucond); TmEcode TmThreadSpawn(ThreadVars *); +TmEcode TmThreadLibSpawn(ThreadVars *); +int TmThreadTimeoutLoop(ThreadVars *tv, TmSlot *s); void TmThreadKillThreadsFamily(int family); void TmThreadKillThreads(void); void TmThreadClearThreadsFamily(int family);