From: Victor Julien <victor@inliniac.net>
Date: Fri, 4 Apr 2014 10:51:26 +0000 (+0200)
Subject: output-streaming: a Log API for streaming data
X-Git-Tag: suricata-2.1beta2~171
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=9d9ef983dde8a404be7a0007a6c1fae7f5126595;p=thirdparty%2Fsuricata.git

output-streaming: a Log API for streaming data

This patch adds a new Log API for streaming data such as TCP reassembled
data and HTTP body data. It could also replace Filedata API.

Each time a new chunk of data is available, the callback will be called.
---

diff --git a/src/Makefile.am b/src/Makefile.am
index 7a419cf9e8..20f729313e 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -226,6 +226,7 @@ output-json-http.c output-json-http.h \
 output-json-ssh.c output-json-ssh.h \
 output-json-tls.c output-json-tls.h \
 output-packet.c output-packet.h \
+output-streaming.c output-streaming.h \
 output-tx.c output-tx.h \
 output-json.c output-json.h \
 packet-queue.c packet-queue.h \
diff --git a/src/output-streaming.c b/src/output-streaming.c
new file mode 100644
index 0000000000..cf76a68b13
--- /dev/null
+++ b/src/output-streaming.c
@@ -0,0 +1,285 @@
+/* Copyright (C) 2007-2014 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 Victor Julien <victor@inliniac.net>
+ *
+ * Logger for stremaing data
+ */
+
+#include "suricata-common.h"
+#include "tm-modules.h"
+#include "output-streaming.h"
+#include "app-layer.h"
+#include "app-layer-parser.h"
+#include "conf.h"
+#include "util-profiling.h"
+
+typedef struct OutputLoggerThreadStore_ {
+    void *thread_data;
+    struct OutputLoggerThreadStore_ *next;
+} OutputLoggerThreadStore;
+
+/** per thread data for this module, contains a list of per thread
+ *  data for the packet loggers. */
+typedef struct OutputLoggerThreadData_ {
+    OutputLoggerThreadStore *store;
+} OutputLoggerThreadData;
+
+/* logger instance, a module + a output ctx,
+ * it's perfectly valid that have multiple instances of the same
+ * log module (e.g. http.log) with different output ctx'. */
+typedef struct OutputStreamingLogger_ {
+    StreamingLogger LogFunc;
+    OutputCtx *output_ctx;
+    struct OutputStreamingLogger_ *next;
+    const char *name;
+    TmmId module_id;
+} OutputStreamingLogger;
+
+static OutputStreamingLogger *list = NULL;
+
+int OutputRegisterStreamingLogger(const char *name, StreamingLogger LogFunc, OutputCtx *output_ctx)
+{
+    int module_id = TmModuleGetIdByName(name);
+    if (module_id < 0)
+        return -1;
+
+    OutputStreamingLogger *op = SCMalloc(sizeof(*op));
+    if (op == NULL)
+        return -1;
+    memset(op, 0x00, sizeof(*op));
+
+    op->LogFunc = LogFunc;
+    op->output_ctx = output_ctx;
+    op->name = name;
+    op->module_id = (TmmId) module_id;
+
+    if (list == NULL)
+        list = op;
+    else {
+        OutputStreamingLogger *t = list;
+        while (t->next)
+            t = t->next;
+        t->next = op;
+    }
+
+    SCLogDebug("OutputRegisterTxLogger happy");
+    return 0;
+}
+
+typedef struct StreamerCallbackData_ {
+    OutputStreamingLogger *logger;
+    OutputLoggerThreadStore *store;
+    ThreadVars *tv;
+    Packet *p;
+} StreamerCallbackData;
+
+int Streamer(void *cbdata, Flow *f, uint8_t *data, uint32_t data_len, uint8_t flags)
+{
+    StreamerCallbackData *streamer_cbdata = (StreamerCallbackData *)cbdata;
+    BUG_ON(streamer_cbdata == NULL);
+    OutputStreamingLogger *logger = streamer_cbdata->logger;
+    OutputLoggerThreadStore *store = streamer_cbdata->store;
+    ThreadVars *tv = streamer_cbdata->tv;
+#ifdef PROFILING
+    Packet *p = streamer_cbdata->p;
+#endif
+    BUG_ON(logger == NULL);
+    BUG_ON(store == NULL);
+
+    while (logger && store) {
+        BUG_ON(logger->LogFunc == NULL);
+
+        SCLogDebug("logger %p", logger);
+        PACKET_PROFILING_TMM_START(p, logger->module_id);
+        logger->LogFunc(tv, store->thread_data, (const Flow *)f, data, data_len, flags);
+        PACKET_PROFILING_TMM_END(p, logger->module_id);
+
+        logger = logger->next;
+        store = store->next;
+
+        BUG_ON(logger == NULL && store != NULL);
+        BUG_ON(logger != NULL && store == NULL);
+    }
+
+    return 0;
+}
+
+static TmEcode OutputStreamingLog(ThreadVars *tv, Packet *p, void *thread_data, PacketQueue *pq, PacketQueue *postpq)
+{
+    BUG_ON(thread_data == NULL);
+    BUG_ON(list == NULL);
+
+    OutputLoggerThreadData *op_thread_data = (OutputLoggerThreadData *)thread_data;
+    OutputStreamingLogger *logger = list;
+    OutputLoggerThreadStore *store = op_thread_data->store;
+
+//    StreamerCallbackData streamer_cbdata = { logger, store, tv, p };
+
+    BUG_ON(logger == NULL && store != NULL);
+    BUG_ON(logger != NULL && store == NULL);
+    BUG_ON(logger == NULL && store == NULL);
+
+    uint8_t flags = 0;
+    Flow * const f = p->flow;
+
+    /* no flow, no streaming */
+    if (f == NULL) {
+        SCReturnInt(TM_ECODE_OK);
+    }
+
+    if (p->flowflags & FLOW_PKT_TOCLIENT)
+        flags |= OUTPUT_STREAMING_FLAG_TOCLIENT;
+    else
+        flags |= OUTPUT_STREAMING_FLAG_TOSERVER;
+
+//    int file_close = (p->flags & PKT_PSEUDO_STREAM_END) ? 1 : 0;
+//    int file_trunc = 0;
+
+    FLOWLOCK_WRLOCK(f);
+
+
+                logger = list;
+                store = op_thread_data->store;
+                while (logger && store) {
+                    BUG_ON(logger->LogFunc == NULL);
+
+                    SCLogDebug("logger %p", logger);
+                    PACKET_PROFILING_TMM_START(p, logger->module_id);
+                    //logger->LogFunc(tv, store->thread_data, (const Packet *)p, (const File *)ff,
+                    //        (const FileData *)write_ffd, flags);
+                    PACKET_PROFILING_TMM_END(p, logger->module_id);
+
+                    logger = logger->next;
+                    store = store->next;
+
+                    BUG_ON(logger == NULL && store != NULL);
+                    BUG_ON(logger != NULL && store == NULL);
+                }
+
+    FLOWLOCK_UNLOCK(f);
+    return TM_ECODE_OK;
+}
+
+/** \brief thread init for the tx logger
+ *  This will run the thread init functions for the individual registered
+ *  loggers */
+static TmEcode OutputStreamingLogThreadInit(ThreadVars *tv, void *initdata, void **data) {
+    OutputLoggerThreadData *td = SCMalloc(sizeof(*td));
+    if (td == NULL)
+        return TM_ECODE_FAILED;
+    memset(td, 0x00, sizeof(*td));
+
+    *data = (void *)td;
+
+    SCLogDebug("OutputStreamingLogThreadInit happy (*data %p)", *data);
+
+    OutputStreamingLogger *logger = list;
+    while (logger) {
+        TmModule *tm_module = TmModuleGetByName((char *)logger->name);
+        if (tm_module == NULL) {
+            SCLogError(SC_ERR_INVALID_ARGUMENT,
+                    "TmModuleGetByName for %s failed", logger->name);
+            exit(EXIT_FAILURE);
+        }
+
+        if (tm_module->ThreadInit) {
+            void *retptr = NULL;
+            if (tm_module->ThreadInit(tv, (void *)logger->output_ctx, &retptr) == TM_ECODE_OK) {
+                OutputLoggerThreadStore *ts = SCMalloc(sizeof(*ts));
+/* todo */      BUG_ON(ts == NULL);
+                memset(ts, 0x00, sizeof(*ts));
+
+                /* store thread handle */
+                ts->thread_data = retptr;
+
+                if (td->store == NULL) {
+                    td->store = ts;
+                } else {
+                    OutputLoggerThreadStore *tmp = td->store;
+                    while (tmp->next != NULL)
+                        tmp = tmp->next;
+                    tmp->next = ts;
+                }
+
+                SCLogDebug("%s is now set up", logger->name);
+            }
+        }
+
+        logger = logger->next;
+    }
+
+    return TM_ECODE_OK;
+}
+
+static TmEcode OutputStreamingLogThreadDeinit(ThreadVars *tv, void *thread_data) {
+    OutputLoggerThreadData *op_thread_data = (OutputLoggerThreadData *)thread_data;
+    OutputLoggerThreadStore *store = op_thread_data->store;
+    OutputStreamingLogger *logger = list;
+
+    while (logger && store) {
+        TmModule *tm_module = TmModuleGetByName((char *)logger->name);
+        if (tm_module == NULL) {
+            SCLogError(SC_ERR_INVALID_ARGUMENT,
+                    "TmModuleGetByName for %s failed", logger->name);
+            exit(EXIT_FAILURE);
+        }
+
+        if (tm_module->ThreadDeinit) {
+            tm_module->ThreadDeinit(tv, store->thread_data);
+        }
+
+        logger = logger->next;
+        store = store->next;
+    }
+
+    return TM_ECODE_OK;
+}
+
+static void OutputStreamingLogExitPrintStats(ThreadVars *tv, void *thread_data) {
+    OutputLoggerThreadData *op_thread_data = (OutputLoggerThreadData *)thread_data;
+    OutputLoggerThreadStore *store = op_thread_data->store;
+    OutputStreamingLogger *logger = list;
+
+    while (logger && store) {
+        TmModule *tm_module = TmModuleGetByName((char *)logger->name);
+        if (tm_module == NULL) {
+            SCLogError(SC_ERR_INVALID_ARGUMENT,
+                    "TmModuleGetByName for %s failed", logger->name);
+            exit(EXIT_FAILURE);
+        }
+
+        if (tm_module->ThreadExitPrintStats) {
+            tm_module->ThreadExitPrintStats(tv, store->thread_data);
+        }
+
+        logger = logger->next;
+        store = store->next;
+    }
+}
+
+void TmModuleStreamingLoggerRegister (void) {
+    tmm_modules[TMM_STREAMINGLOGGER].name = "__streaming_logger__";
+    tmm_modules[TMM_STREAMINGLOGGER].ThreadInit = OutputStreamingLogThreadInit;
+    tmm_modules[TMM_STREAMINGLOGGER].Func = OutputStreamingLog;
+    tmm_modules[TMM_STREAMINGLOGGER].ThreadExitPrintStats = OutputStreamingLogExitPrintStats;
+    tmm_modules[TMM_STREAMINGLOGGER].ThreadDeinit = OutputStreamingLogThreadDeinit;
+    tmm_modules[TMM_STREAMINGLOGGER].cap_flags = 0;
+}
diff --git a/src/output-streaming.h b/src/output-streaming.h
new file mode 100644
index 0000000000..9bf44bcd45
--- /dev/null
+++ b/src/output-streaming.h
@@ -0,0 +1,45 @@
+/* Copyright (C) 2007-2014 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 Victor Julien <victor@inliniac.net>
+ *
+ * AppLayer Filedata Logger Output registration functions
+ */
+
+#ifndef __OUTPUT_STREAMING_H__
+#define __OUTPUT_STREAMING_H__
+
+#include "decode.h"
+#include "util-file.h"
+
+#define OUTPUT_STREAMING_FLAG_OPEN      0x01
+#define OUTPUT_STREAMING_FLAG_CLOSE     0x02
+#define OUTPUT_STREAMING_FLAG_TOSERVER  0x04
+#define OUTPUT_STREAMING_FLAG_TOCLIENT  0x08
+
+/** filedata logger function pointer type */
+typedef int (*StreamingLogger)(ThreadVars *, void *thread_data,
+        const Flow *f, const uint8_t *data, uint32_t data_len, uint8_t flags);
+
+int OutputRegisterStreamingLogger(const char *name, StreamingLogger LogFunc, OutputCtx *);
+
+void TmModuleStreamingLoggerRegister (void);
+
+#endif /* __OUTPUT_STREAMING_H__ */
diff --git a/src/output.c b/src/output.c
index e4cc85fcb3..5359825773 100644
--- a/src/output.c
+++ b/src/output.c
@@ -424,6 +424,76 @@ error:
     exit(EXIT_FAILURE);
 }
 
+/**
+ * \brief Register a streaming data output module.
+ *
+ * This function will register an output module so it can be
+ * configured with the configuration file.
+ *
+ * \retval Returns 0 on success, -1 on failure.
+ */
+void
+OutputRegisterStreamingModule(const char *name, const char *conf_name,
+    OutputCtx *(*InitFunc)(ConfNode *), StreamingLogger StreamingLogFunc)
+{
+    if (unlikely(StreamingLogFunc == NULL)) {
+        goto error;
+    }
+
+    OutputModule *module = SCCalloc(1, sizeof(*module));
+    if (unlikely(module == NULL)) {
+        goto error;
+    }
+
+    module->name = name;
+    module->conf_name = conf_name;
+    module->InitFunc = InitFunc;
+    module->StreamingLogFunc = StreamingLogFunc;
+    TAILQ_INSERT_TAIL(&output_modules, module, entries);
+
+    SCLogDebug("Streaming logger \"%s\" registered.", name);
+    return;
+error:
+    SCLogError(SC_ERR_FATAL, "Fatal error encountered. Exiting...");
+    exit(EXIT_FAILURE);
+}
+
+/**
+ * \brief Register a streaming data output sub-module.
+ *
+ * This function will register an output module so it can be
+ * configured with the configuration file.
+ *
+ * \retval Returns 0 on success, -1 on failure.
+ */
+void
+OutputRegisterStreamingSubModule(const char *parent_name, const char *name,
+    const char *conf_name, OutputCtx *(*InitFunc)(ConfNode *, OutputCtx *),
+    StreamingLogger StreamingLogFunc)
+{
+    if (unlikely(StreamingLogFunc == NULL)) {
+        goto error;
+    }
+
+    OutputModule *module = SCCalloc(1, sizeof(*module));
+    if (unlikely(module == NULL)) {
+        goto error;
+    }
+
+    module->name = name;
+    module->conf_name = conf_name;
+    module->parent_name = parent_name;
+    module->InitSubFunc = InitFunc;
+    module->StreamingLogFunc = StreamingLogFunc;
+    TAILQ_INSERT_TAIL(&output_modules, module, entries);
+
+    SCLogDebug("Streaming logger \"%s\" registered.", name);
+    return;
+error:
+    SCLogError(SC_ERR_FATAL, "Fatal error encountered. Exiting...");
+    exit(EXIT_FAILURE);
+}
+
 /**
  * \brief Get an output module by name.
  *
diff --git a/src/output.h b/src/output.h
index a097cbc99c..37ffea03bd 100644
--- a/src/output.h
+++ b/src/output.h
@@ -35,6 +35,7 @@
 #include "output-file.h"
 #include "output-filedata.h"
 #include "output-flow.h"
+#include "output-streaming.h"
 
 typedef struct OutputModule_ {
     const char *name;
@@ -49,6 +50,7 @@ typedef struct OutputModule_ {
     FileLogger FileLogFunc;
     FiledataLogger FiledataLogFunc;
     FlowLogger FlowLogFunc;
+    StreamingLogger StreamingLogFunc;
     AppProto alproto;
 
     TAILQ_ENTRY(OutputModule_) entries;
@@ -88,6 +90,12 @@ void OutputRegisterFlowSubModule(const char *parent_name, const char *name,
     const char *conf_name, OutputCtx *(*InitFunc)(ConfNode *, OutputCtx *),
     FlowLogger FlowLogFunc);
 
+void OutputRegisterStreamingModule(const char *name, const char *conf_name,
+    OutputCtx *(*InitFunc)(ConfNode *), StreamingLogger StreamingLogFunc);
+void OutputRegisterStreamingSubModule(const char *parent_name, const char *name,
+    const char *conf_name, OutputCtx *(*InitFunc)(ConfNode *, OutputCtx *),
+    StreamingLogger StreamingLogFunc);
+
 OutputModule *OutputGetModuleByConfName(const char *name);
 void OutputDeregisterAll(void);
 
diff --git a/src/runmodes.c b/src/runmodes.c
index c84165873f..d29dedac91 100644
--- a/src/runmodes.c
+++ b/src/runmodes.c
@@ -434,6 +434,7 @@ static TmModule *pkt_logger_module = NULL;
 static TmModule *tx_logger_module = NULL;
 static TmModule *file_logger_module = NULL;
 static TmModule *filedata_logger_module = NULL;
+static TmModule *streaming_logger_module = NULL;
 
 /**
  * Cleanup the run mode.
@@ -585,6 +586,27 @@ static void SetupOutput(const char *name, OutputModule *module, OutputCtx *outpu
             TAILQ_INSERT_TAIL(&RunModeOutputs, runmode_output, entries);
             SCLogDebug("__filedata_logger__ added");
         }
+    } else if (module->StreamingLogFunc) {
+        SCLogDebug("%s is a streaming logger", module->name);
+        OutputRegisterStreamingLogger(module->name, module->StreamingLogFunc, output_ctx);
+
+        /* need one instance of the streaming logger module */
+        if (streaming_logger_module == NULL) {
+            streaming_logger_module = TmModuleGetByName("__streaming_logger__");
+            if (streaming_logger_module == NULL) {
+                SCLogError(SC_ERR_INVALID_ARGUMENT,
+                        "TmModuleGetByName for __streaming_logger__ failed");
+                exit(EXIT_FAILURE);
+            }
+
+            RunModeOutput *runmode_output = SCCalloc(1, sizeof(RunModeOutput));
+            if (unlikely(runmode_output == NULL))
+                return;
+            runmode_output->tm_module = streaming_logger_module;
+            runmode_output->output_ctx = NULL;
+            TAILQ_INSERT_TAIL(&RunModeOutputs, runmode_output, entries);
+            SCLogDebug("__streaming_logger__ added");
+        }
     } else {
         SCLogDebug("%s is a regular logger", module->name);
 
diff --git a/src/suricata.c b/src/suricata.c
index db130392b2..be073b2f76 100644
--- a/src/suricata.c
+++ b/src/suricata.c
@@ -167,6 +167,7 @@
 #include "output-tx.h"
 #include "output-file.h"
 #include "output-filedata.h"
+#include "output-streaming.h"
 
 #include "util-privs.h"
 
@@ -891,6 +892,7 @@ void RegisterAllModules()
     TmModuleTxLoggerRegister();
     TmModuleFileLoggerRegister();
     TmModuleFiledataLoggerRegister();
+    TmModuleStreamingLoggerRegister();
     TmModuleDebugList();
     /* nflog */
     TmModuleReceiveNFLOGRegister();
diff --git a/src/tm-modules.c b/src/tm-modules.c
index 40ce0dfc07..ae4ecf1f31 100644
--- a/src/tm-modules.c
+++ b/src/tm-modules.c
@@ -251,6 +251,7 @@ const char * TmModuleTmmIdToString(TmmId id)
         CASE_CODE (TMM_TXLOGGER);
         CASE_CODE (TMM_FILELOGGER);
         CASE_CODE (TMM_FILEDATALOGGER);
+        CASE_CODE (TMM_STREAMINGLOGGER);
         CASE_CODE (TMM_JSONALERTLOG);
         CASE_CODE (TMM_JSONDROPLOG);
         CASE_CODE (TMM_JSONDNSLOG);
diff --git a/src/tm-threads-common.h b/src/tm-threads-common.h
index de869747ba..311dd909fc 100644
--- a/src/tm-threads-common.h
+++ b/src/tm-threads-common.h
@@ -82,6 +82,7 @@ typedef enum {
     TMM_TXLOGGER,
     TMM_FILELOGGER,
     TMM_FILEDATALOGGER,
+    TMM_STREAMINGLOGGER,
     TMM_JSONALERTLOG,
     TMM_JSONDROPLOG,
     TMM_JSONHTTPLOG,