diff options
author | Ashlee Young <ashlee@onosfw.com> | 2015-09-09 22:21:41 -0700 |
---|---|---|
committer | Ashlee Young <ashlee@onosfw.com> | 2015-09-09 22:21:41 -0700 |
commit | 8879b125d26e8db1a5633de5a9c692eb2d1c4f83 (patch) | |
tree | c7259d85a991b83dfa85ab2e339360669fc1f58e /framework/src/suricata/src/output-streaming.c | |
parent | 13d05bc8458758ee39cb829098241e89616717ee (diff) |
suricata checkin based on commit id a4bce14770beee46a537eda3c3f6e8e8565d5d0a
Change-Id: I9a214fa0ee95e58fc640e50bd604dac7f42db48f
Diffstat (limited to 'framework/src/suricata/src/output-streaming.c')
-rw-r--r-- | framework/src/suricata/src/output-streaming.c | 469 |
1 files changed, 469 insertions, 0 deletions
diff --git a/framework/src/suricata/src/output-streaming.c b/framework/src/suricata/src/output-streaming.c new file mode 100644 index 00000000..d416cbc3 --- /dev/null +++ b/framework/src/suricata/src/output-streaming.c @@ -0,0 +1,469 @@ +/* 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 streaming data + */ + +#include "suricata-common.h" +#include "tm-modules.h" +#include "output-streaming.h" +#include "app-layer.h" +#include "app-layer-parser.h" +#include "app-layer-htp.h" +#include "util-print.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; + uint32_t loggers; +} 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; + enum OutputStreamingType type; +} OutputStreamingLogger; + +static OutputStreamingLogger *list = NULL; + +int OutputRegisterStreamingLogger(const char *name, StreamingLogger LogFunc, + OutputCtx *output_ctx, enum OutputStreamingType type ) +{ + 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; + op->type = type; + + 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; + enum OutputStreamingType type; +} StreamerCallbackData; + +int Streamer(void *cbdata, Flow *f, uint8_t *data, uint32_t data_len, uint64_t tx_id, 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); + + if (logger->type == streamer_cbdata->type) { + SCLogDebug("logger %p", logger); + PACKET_PROFILING_TMM_START(p, logger->module_id); + logger->LogFunc(tv, store->thread_data, (const Flow *)f, data, data_len, tx_id, 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; +} + +/** \brief Http Body Iterator for logging + * + * Global logic: + * + * - For each tx + * - For each body chunk + * - Invoke Streamer + */ + +int HttpBodyIterator(Flow *f, int close, void *cbdata, uint8_t iflags) +{ + SCLogDebug("called with %p, %d, %p, %02x", f, close, cbdata, iflags); + + HtpState *s = f->alstate; + if (s != NULL && s->conn != NULL) { + int tx_progress_done_value_ts = + AppLayerParserGetStateProgressCompletionStatus(IPPROTO_TCP, + ALPROTO_HTTP, STREAM_TOSERVER); + int tx_progress_done_value_tc = + AppLayerParserGetStateProgressCompletionStatus(IPPROTO_TCP, + ALPROTO_HTTP, STREAM_TOCLIENT); + + // for each tx + uint64_t tx_id = 0; + uint64_t total_txs = AppLayerParserGetTxCnt(f->proto, f->alproto, f->alstate); + SCLogDebug("s->conn %p", s->conn); + for (tx_id = 0; tx_id < total_txs; tx_id++) { // TODO optimization store log tx + htp_tx_t *tx = AppLayerParserGetTx(f->proto, f->alproto, f->alstate, tx_id); + if (tx != NULL) { + int tx_done = 0; + int tx_logged = 0; + + int tx_progress_ts = AppLayerParserGetStateProgress( + IPPROTO_TCP, ALPROTO_HTTP, tx, FlowGetDisruptionFlags(f, STREAM_TOSERVER)); + if (tx_progress_ts >= tx_progress_done_value_ts) { + int tx_progress_tc = AppLayerParserGetStateProgress( + IPPROTO_TCP, ALPROTO_HTTP, tx, FlowGetDisruptionFlags(f, STREAM_TOCLIENT)); + if (tx_progress_tc >= tx_progress_done_value_tc) { + tx_done = 1; + } + } + + SCLogDebug("tx %p", tx); + HtpTxUserData *htud = (HtpTxUserData *) htp_tx_get_user_data(tx); + if (htud != NULL) { + SCLogDebug("htud %p", htud); + HtpBody *body = NULL; + if (iflags & OUTPUT_STREAMING_FLAG_TOCLIENT) + body = &htud->request_body; + else if (iflags & OUTPUT_STREAMING_FLAG_TOSERVER) + body = &htud->response_body; + + if (body == NULL) { + SCLogDebug("no body"); + goto next; + } + if (body->first == NULL) { + SCLogDebug("no body chunks"); + goto next; + } + if (body->last->logged == 1) { + SCLogDebug("all logged already"); + goto next; + } + + // for each chunk + HtpBodyChunk *chunk = body->first; + for ( ; chunk != NULL; chunk = chunk->next) { + if (chunk->logged) { + SCLogDebug("logged %d", chunk->logged); + continue; + } + + uint8_t flags = iflags | OUTPUT_STREAMING_FLAG_TRANSACTION; + if (chunk->stream_offset == 0) + flags |= OUTPUT_STREAMING_FLAG_OPEN; + /* if we need to close and we're at the last segment in the list + * we add the 'close' flag so the logger can close up. */ + if ((tx_done || close) && chunk->next == NULL) { + flags |= OUTPUT_STREAMING_FLAG_CLOSE; + } + + // invoke Streamer + Streamer(cbdata, f, chunk->data, (uint32_t)chunk->len, tx_id, flags); + //PrintRawDataFp(stdout, chunk->data, chunk->len); + chunk->logged = 1; + tx_logged = 1; + } + + next: + /* if we need to close we need to invoke the Streamer for sure. If we + * logged no chunks, we call the Streamer with NULL data so it can + * close up. */ + if (tx_logged == 0 && (close||tx_done)) { + Streamer(cbdata, f, NULL, 0, tx_id, + OUTPUT_STREAMING_FLAG_CLOSE|OUTPUT_STREAMING_FLAG_TRANSACTION); + } + } + } + } + } + + + return 0; +} + +int StreamIterator(Flow *f, TcpStream *stream, int close, void *cbdata, uint8_t iflags) +{ + SCLogDebug("called with %p, %d, %p, %02x", f, close, cbdata, iflags); + int logged = 0; + + /* optimization: don't iterate list if we've logged all, + * so check the last segment's flags */ + if (stream->seg_list_tail != NULL && + (!(stream->seg_list_tail->flags & SEGMENTTCP_FLAG_LOGAPI_PROCESSED))) + { + TcpSegment *seg = stream->seg_list; + while (seg) { + uint8_t flags = iflags; + + if (seg->flags & SEGMENTTCP_FLAG_LOGAPI_PROCESSED) { + seg = seg->next; + continue; + } + + if (SEQ_GT(seg->seq + seg->payload_len, stream->last_ack)) { + SCLogDebug("seg not (fully) acked yet"); + break; + } + + if (seg->seq == stream->isn + 1) + flags |= OUTPUT_STREAMING_FLAG_OPEN; + /* if we need to close and we're at the last segment in the list + * we add the 'close' flag so the logger can close up. */ + if (close && seg->next == NULL) + flags |= OUTPUT_STREAMING_FLAG_CLOSE; + + Streamer(cbdata, f, seg->payload, (uint32_t)seg->payload_len, 0, flags); + + seg->flags |= SEGMENTTCP_FLAG_LOGAPI_PROCESSED; + + seg = seg->next; + + logged = 1; + } + } + + /* if we need to close we need to invoke the Streamer for sure. If we + * logged no segments, we call the Streamer with NULL data so it can + * close up. */ + if (logged == 0 && close) { + Streamer(cbdata, f, NULL, 0, 0, OUTPUT_STREAMING_FLAG_CLOSE); + } + + 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 , 0}; + + 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; + + FLOWLOCK_WRLOCK(f); + + if (op_thread_data->loggers & (1<<STREAMING_TCP_DATA)) { + TcpSession *ssn = f->protoctx; + if (ssn) { + int close = (ssn->state >= TCP_CLOSED); + close |= ((p->flags & PKT_PSEUDO_STREAM_END) ? 1 : 0); + SCLogDebug("close ? %s", close ? "yes" : "no"); + + TcpStream *stream = flags & OUTPUT_STREAMING_FLAG_TOSERVER ? &ssn->client : &ssn->server; + + streamer_cbdata.type = STREAMING_TCP_DATA; + StreamIterator(p->flow, stream, close, (void *)&streamer_cbdata, flags); + } + } + if (op_thread_data->loggers & (1<<STREAMING_HTTP_BODIES)) { + if (f->alproto == ALPROTO_HTTP && f->alstate != NULL) { + int close = 0; + TcpSession *ssn = f->protoctx; + if (ssn) { + close = (ssn->state >= TCP_CLOSED); + close |= ((p->flags & PKT_PSEUDO_STREAM_END) ? 1 : 0); + } + SCLogDebug("close ? %s", close ? "yes" : "no"); + streamer_cbdata.type = STREAMING_HTTP_BODIES; + HttpBodyIterator(f, close, (void *)&streamer_cbdata, flags); + } + } + + 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; + } + + SCLogInfo("%s is now set up", logger->name); + } + } + + td->loggers |= (1<<logger->type); + + 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; +} + +void OutputStreamingShutdown(void) +{ + OutputStreamingLogger *logger = list; + while (logger) { + OutputStreamingLogger *next_logger = logger->next; + SCFree(logger); + logger = next_logger; + } + list = NULL; +} |