diff options
Diffstat (limited to 'framework/src/suricata/src/source-erf-dag.c')
| -rw-r--r-- | framework/src/suricata/src/source-erf-dag.c | 670 |
1 files changed, 0 insertions, 670 deletions
diff --git a/framework/src/suricata/src/source-erf-dag.c b/framework/src/suricata/src/source-erf-dag.c deleted file mode 100644 index f6640712..00000000 --- a/framework/src/suricata/src/source-erf-dag.c +++ /dev/null @@ -1,670 +0,0 @@ -/* Copyright (C) 2010-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 Endace Technology Limited. - * \author Jason MacLulich <jason.maclulich@endace.com> - * - * Support for reading ERF records from a DAG card. - * - * Only ethernet supported at this time. - */ - -#include "suricata-common.h" -#include "suricata.h" -#include "tm-threads.h" - -#include "util-privs.h" -#include "util-device.h" -#include "tmqh-packetpool.h" - -#ifndef HAVE_DAG - -TmEcode NoErfDagSupportExit(ThreadVars *, void *, void **); - -void -TmModuleReceiveErfDagRegister(void) -{ - tmm_modules[TMM_RECEIVEERFDAG].name = "ReceiveErfDag"; - tmm_modules[TMM_RECEIVEERFDAG].ThreadInit = NoErfDagSupportExit; - tmm_modules[TMM_RECEIVEERFDAG].Func = NULL; - tmm_modules[TMM_RECEIVEERFDAG].ThreadExitPrintStats = NULL; - tmm_modules[TMM_RECEIVEERFDAG].ThreadDeinit = NULL; - tmm_modules[TMM_RECEIVEERFDAG].RegisterTests = NULL; - tmm_modules[TMM_RECEIVEERFDAG].cap_flags = SC_CAP_NET_ADMIN; - tmm_modules[TMM_RECEIVEERFDAG].flags = TM_FLAG_RECEIVE_TM; -} - -void -TmModuleDecodeErfDagRegister(void) -{ - tmm_modules[TMM_DECODEERFDAG].name = "DecodeErfDag"; - tmm_modules[TMM_DECODEERFDAG].ThreadInit = NoErfDagSupportExit; - tmm_modules[TMM_DECODEERFDAG].Func = NULL; - tmm_modules[TMM_DECODEERFDAG].ThreadExitPrintStats = NULL; - tmm_modules[TMM_DECODEERFDAG].ThreadDeinit = NULL; - tmm_modules[TMM_DECODEERFDAG].RegisterTests = NULL; - tmm_modules[TMM_DECODEERFDAG].cap_flags = 0; - tmm_modules[TMM_DECODEERFDAG].flags = TM_FLAG_DECODE_TM; -} - -TmEcode -NoErfDagSupportExit(ThreadVars *tv, void *initdata, void **data) -{ - SCLogError(SC_ERR_DAG_NOSUPPORT, - "Error creating thread %s: you do not have support for DAG cards " - "enabled please recompile with --enable-dag", tv->name); - exit(EXIT_FAILURE); -} - -#else /* Implied we do have DAG support */ - -#include "source-erf-dag.h" -#include <dagapi.h> - -/* Minimum amount of data to read from the DAG at a time. */ -#define MINDATA 32768 - -/* Maximum time (us) to wait for MINDATA to be read. */ -#define MAXWAIT 20000 - -/* Poll interval in microseconds. */ -#define POLL_INTERVAL 1000; - -/* Number of bytes per loop to process before fetching more data. */ -#define BYTES_PER_LOOP (4 * 1024 * 1024) /* 4 MB */ - -extern int max_pending_packets; - -typedef struct ErfDagThreadVars_ { - ThreadVars *tv; - TmSlot *slot; - - int dagfd; - int dagstream; - char dagname[DAGNAME_BUFSIZE]; - - struct timeval maxwait, poll; /* Could possibly be made static */ - - LiveDevice *livedev; - - uint64_t bytes; - uint16_t packets; - uint16_t drops; - - /* Current location in the DAG stream input buffer. - */ - uint8_t *top; - uint8_t *btm; - -} ErfDagThreadVars; - -static inline TmEcode ProcessErfDagRecords(ErfDagThreadVars *ewtn, uint8_t *top, - uint32_t *pkts_read); -static inline TmEcode ProcessErfDagRecord(ErfDagThreadVars *ewtn, char *prec); -TmEcode ReceiveErfDagLoop(ThreadVars *, void *data, void *slot); -TmEcode ReceiveErfDagThreadInit(ThreadVars *, void *, void **); -void ReceiveErfDagThreadExitStats(ThreadVars *, void *); -TmEcode ReceiveErfDagThreadDeinit(ThreadVars *, void *); -TmEcode DecodeErfDagThreadInit(ThreadVars *, void *, void **); -TmEcode DecodeErfDagThreadDeinit(ThreadVars *tv, void *data); -TmEcode DecodeErfDag(ThreadVars *, Packet *, void *, PacketQueue *, - PacketQueue *); -void ReceiveErfDagCloseStream(int dagfd, int stream); - -/** - * \brief Register the ERF file receiver (reader) module. - */ -void -TmModuleReceiveErfDagRegister(void) -{ - tmm_modules[TMM_RECEIVEERFDAG].name = "ReceiveErfDag"; - tmm_modules[TMM_RECEIVEERFDAG].ThreadInit = ReceiveErfDagThreadInit; - tmm_modules[TMM_RECEIVEERFDAG].Func = NULL; - tmm_modules[TMM_RECEIVEERFDAG].PktAcqLoop = ReceiveErfDagLoop; - tmm_modules[TMM_RECEIVEERFDAG].ThreadExitPrintStats = - ReceiveErfDagThreadExitStats; - tmm_modules[TMM_RECEIVEERFDAG].ThreadDeinit = NULL; - tmm_modules[TMM_RECEIVEERFDAG].RegisterTests = NULL; - tmm_modules[TMM_RECEIVEERFDAG].cap_flags = 0; - tmm_modules[TMM_RECEIVEERFDAG].flags = TM_FLAG_RECEIVE_TM; -} - -/** - * \brief Register the ERF file decoder module. - */ -void -TmModuleDecodeErfDagRegister(void) -{ - tmm_modules[TMM_DECODEERFDAG].name = "DecodeErfDag"; - tmm_modules[TMM_DECODEERFDAG].ThreadInit = DecodeErfDagThreadInit; - tmm_modules[TMM_DECODEERFDAG].Func = DecodeErfDag; - tmm_modules[TMM_DECODEERFDAG].ThreadExitPrintStats = NULL; - tmm_modules[TMM_DECODEERFDAG].ThreadDeinit = DecodeErfDagThreadDeinit; - tmm_modules[TMM_DECODEERFDAG].RegisterTests = NULL; - tmm_modules[TMM_DECODEERFDAG].cap_flags = 0; - tmm_modules[TMM_DECODEERFDAG].flags = TM_FLAG_DECODE_TM; -} - -/** - * \brief Initialize the ERF receiver thread, generate a single - * ErfDagThreadVar structure for each thread, this will - * contain a DAG file descriptor which is read when the - * thread executes. - * - * \param tv Thread variable to ThreadVars - * \param initdata Initial data to the interface passed from the user, - * this is processed by the user. - * - * We assume that we have only a single name for the DAG - * interface. - * - * \param data data pointer gets populated with - * - */ -TmEcode -ReceiveErfDagThreadInit(ThreadVars *tv, void *initdata, void **data) -{ - SCEnter(); - int stream_count = 0; - - if (initdata == NULL) { - SCLogError(SC_ERR_INVALID_ARGUMENT, - "Error: No DAG interface provided."); - SCReturnInt(TM_ECODE_FAILED); - } - - ErfDagThreadVars *ewtn = SCMalloc(sizeof(ErfDagThreadVars)); - if (unlikely(ewtn == NULL)) { - SCLogError(SC_ERR_MEM_ALLOC, - "Failed to allocate memory for ERF DAG thread vars."); - exit(EXIT_FAILURE); - } - - memset(ewtn, 0, sizeof(*ewtn)); - - /* dag_parse_name will return a DAG device name and stream number - * to open for this thread. - */ - if (dag_parse_name(initdata, ewtn->dagname, DAGNAME_BUFSIZE, - &ewtn->dagstream) < 0) { - SCLogError(SC_ERR_INVALID_ARGUMENT, - "Failed to parse DAG interface: %s", - (char*)initdata); - SCFree(ewtn); - exit(EXIT_FAILURE); - } - - ewtn->livedev = LiveGetDevice(initdata); - if (ewtn->livedev == NULL) { - SCLogError(SC_ERR_INVALID_VALUE, "Unable to get %s live device", - (char *)initdata); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - - SCLogInfo("Opening DAG: %s on stream: %d for processing", - ewtn->dagname, ewtn->dagstream); - - if ((ewtn->dagfd = dag_open(ewtn->dagname)) < 0) { - SCLogError(SC_ERR_ERF_DAG_OPEN_FAILED, "Failed to open DAG: %s", - ewtn->dagname); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - - /* Check to make sure the card has enough available streams to - * support reading from the one specified. - */ - if ((stream_count = dag_rx_get_stream_count(ewtn->dagfd)) < 0) { - SCLogError(SC_ERR_ERF_DAG_OPEN_FAILED, - "Failed to open stream: %d, DAG: %s, could not query stream count", - ewtn->dagstream, ewtn->dagname); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - - /* Check to make sure we have enough rx streams to open the stream - * the user is asking for. - */ - if (ewtn->dagstream > stream_count * 2) { - SCLogError(SC_ERR_ERF_DAG_OPEN_FAILED, - "Failed to open stream: %d, DAG: %s, insufficient streams: %d", - ewtn->dagstream, ewtn->dagname, stream_count); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - - /* If we are transmitting into a soft DAG card then set the stream - * to act in reverse mode. - */ - if (0 != (ewtn->dagstream & 0x01)) { - /* Setting reverse mode for using with soft dag from daemon side */ - if (dag_set_mode(ewtn->dagfd, ewtn->dagstream, DAG_REVERSE_MODE)) { - SCLogError(SC_ERR_ERF_DAG_STREAM_OPEN_FAILED, - "Failed to set mode to DAG_REVERSE_MODE on stream: %d, DAG: %s", - ewtn->dagstream, ewtn->dagname); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - } - - if (dag_attach_stream(ewtn->dagfd, ewtn->dagstream, 0, 0) < 0) { - SCLogError(SC_ERR_ERF_DAG_STREAM_OPEN_FAILED, - "Failed to open DAG stream: %d, DAG: %s", - ewtn->dagstream, ewtn->dagname); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - - if (dag_start_stream(ewtn->dagfd, ewtn->dagstream) < 0) { - SCLogError(SC_ERR_ERF_DAG_STREAM_START_FAILED, - "Failed to start DAG stream: %d, DAG: %s", - ewtn->dagstream, ewtn->dagname); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - - SCLogInfo("Attached and started stream: %d on DAG: %s", - ewtn->dagstream, ewtn->dagname); - - /* - * Initialise DAG Polling parameters. - */ - timerclear(&ewtn->maxwait); - ewtn->maxwait.tv_usec = MAXWAIT; - timerclear(&ewtn->poll); - ewtn->poll.tv_usec = POLL_INTERVAL; - - /* 32kB minimum data to return -- we still restrict the number of - * pkts that are processed to a maximum of dag_max_read_packets. - */ - if (dag_set_stream_poll(ewtn->dagfd, ewtn->dagstream, MINDATA, - &(ewtn->maxwait), &(ewtn->poll)) < 0) { - SCLogError(SC_ERR_ERF_DAG_STREAM_SET_FAILED, - "Failed to set poll parameters for stream: %d, DAG: %s", - ewtn->dagstream, ewtn->dagname); - SCFree(ewtn); - SCReturnInt(TM_ECODE_FAILED); - } - - ewtn->packets = StatsRegisterCounter("capture.dag_packets", tv); - ewtn->drops = StatsRegisterCounter("capture.dag_drops", tv); - - ewtn->tv = tv; - *data = (void *)ewtn; - - SCLogInfo("Starting processing packets from stream: %d on DAG: %s", - ewtn->dagstream, ewtn->dagname); - - SCReturnInt(TM_ECODE_OK); -} - -/** - * \brief Receives packets from a DAG interface. - * - * \param tv pointer to ThreadVars - * \param data pointer to ErfDagThreadVars - * \param slot slot containing task information - * - * \retval TM_ECODE_OK on success - * \retval TM_ECODE_FAILED on failure - */ -TmEcode -ReceiveErfDagLoop(ThreadVars *tv, void *data, void *slot) -{ - SCEnter(); - - ErfDagThreadVars *dtv = (ErfDagThreadVars *)data; - uint32_t diff = 0; - int err; - uint8_t *top = NULL; - uint32_t pkts_read = 0; - TmSlot *s = (TmSlot *)slot; - - dtv->slot = s->slot_next; - - while (1) { - if (suricata_ctl_flags & (SURICATA_STOP | SURICATA_KILL)) { - SCReturnInt(TM_ECODE_OK); - } - - top = dag_advance_stream(dtv->dagfd, dtv->dagstream, &(dtv->btm)); - if (top == NULL) { - if (errno == EAGAIN) { - if (dtv->dagstream & 0x1) { - usleep(10 * 1000); - dtv->btm = dtv->top; - } - continue; - } else { - SCLogError(SC_ERR_ERF_DAG_STREAM_READ_FAILED, - "Failed to read from stream: %d, DAG: %s when " - "using dag_advance_stream", - dtv->dagstream, dtv->dagname); - SCReturnInt(TM_ECODE_FAILED); - } - } - - diff = top - dtv->btm; - if (diff == 0) { - continue; - } - - assert(diff >= dag_record_size); - - err = ProcessErfDagRecords(dtv, top, &pkts_read); - - if (err == TM_ECODE_FAILED) { - SCLogError(SC_ERR_ERF_DAG_STREAM_READ_FAILED, - "Failed to read from stream: %d, DAG: %s", - dtv->dagstream, dtv->dagname); - ReceiveErfDagCloseStream(dtv->dagfd, dtv->dagstream); - SCReturnInt(TM_ECODE_FAILED); - } - - StatsSyncCountersIfSignalled(tv); - - SCLogDebug("Read %d records from stream: %d, DAG: %s", - pkts_read, dtv->dagstream, dtv->dagname); - } - - SCReturnInt(TM_ECODE_OK); -} - -/** - * \brief Process a chunk of records read from a DAG interface. - * - * This function takes a pointer to buffer read from the DAG interface - * and processes it individual records. - */ -static inline TmEcode -ProcessErfDagRecords(ErfDagThreadVars *ewtn, uint8_t *top, uint32_t *pkts_read) -{ - SCEnter(); - - int err = 0; - dag_record_t *dr = NULL; - char *prec = NULL; - int rlen; - char hdr_type = 0; - int processed = 0; - - *pkts_read = 0; - - while (((top - ewtn->btm) >= dag_record_size) && - ((processed + dag_record_size) < BYTES_PER_LOOP)) { - - /* Make sure we have at least one packet in the packet pool, - * to prevent us from alloc'ing packets at line rate. */ - PacketPoolWait(); - - prec = (char *)ewtn->btm; - dr = (dag_record_t*)prec; - rlen = ntohs(dr->rlen); - hdr_type = dr->type; - - /* If we don't have enough data to finish processing this ERF - * record return and maybe next time we will. - */ - if ((top - ewtn->btm) < rlen) - SCReturnInt(TM_ECODE_OK); - - ewtn->btm += rlen; - processed += rlen; - - /* Only support ethernet at this time. */ - switch (hdr_type & 0x7f) { - case TYPE_PAD: - /* Skip. */ - continue; - case TYPE_DSM_COLOR_ETH: - case TYPE_COLOR_ETH: - case TYPE_COLOR_HASH_ETH: - /* In these types the color value overwrites the lctr - * (drop count). */ - break; - case TYPE_ETH: - if (dr->lctr) { - StatsAddUI64(ewtn->tv, ewtn->drops, ntohs(dr->lctr)); - } - break; - default: - SCLogError(SC_ERR_UNIMPLEMENTED, - "Processing of DAG record type: %d not implemented.", dr->type); - SCReturnInt(TM_ECODE_FAILED); - } - - err = ProcessErfDagRecord(ewtn, prec); - if (err != TM_ECODE_OK) { - SCReturnInt(TM_ECODE_FAILED); - } - - (*pkts_read)++; - } - - SCReturnInt(TM_ECODE_OK); -} - -/** - * \brief Process a DAG record into a TM packet buffer. - * \param prec pointer to a DAG record. - * \param - */ -static inline TmEcode -ProcessErfDagRecord(ErfDagThreadVars *ewtn, char *prec) -{ - SCEnter(); - - int wlen = 0; - int rlen = 0; - int hdr_num = 0; - char hdr_type = 0; - dag_record_t *dr = (dag_record_t*)prec; - erf_payload_t *pload; - Packet *p; - - hdr_type = dr->type; - wlen = ntohs(dr->wlen); - rlen = ntohs(dr->rlen); - - /* count extension headers */ - while (hdr_type & 0x80) { - if (rlen < (dag_record_size + (hdr_num * 8))) { - SCLogError(SC_ERR_UNIMPLEMENTED, - "Insufficient captured packet length."); - SCReturnInt(TM_ECODE_FAILED); - } - hdr_type = prec[(dag_record_size + (hdr_num * 8))]; - hdr_num++; - } - - /* Check that the whole frame was captured */ - if (rlen < (dag_record_size + (8 * hdr_num) + 2 + wlen)) { - SCLogInfo("Incomplete frame captured."); - SCReturnInt(TM_ECODE_OK); - } - - /* skip over extension headers */ - pload = (erf_payload_t *)(prec + dag_record_size + (8 * hdr_num)); - - p = PacketGetFromQueueOrAlloc(); - if (p == NULL) { - SCLogError(SC_ERR_MEM_ALLOC, - "Failed to allocate a Packet on stream: %d, DAG: %s", - ewtn->dagstream, ewtn->dagname); - SCReturnInt(TM_ECODE_FAILED); - } - PKT_SET_SRC(p, PKT_SRC_WIRE); - - SET_PKT_LEN(p, wlen); - p->datalink = LINKTYPE_ETHERNET; - - /* Take into account for link type Ethernet ETH frame starts - * after ther ERF header + pad. - */ - if (unlikely(PacketCopyData(p, pload->eth.dst, GET_PKT_LEN(p)))) { - TmqhOutputPacketpool(ewtn->tv, p); - SCReturnInt(TM_ECODE_FAILED); - } - - /* Convert ERF time to timeval - from libpcap. */ - uint64_t ts = dr->ts; - p->ts.tv_sec = ts >> 32; - ts = (ts & 0xffffffffULL) * 1000000; - ts += 0x80000000; /* rounding */ - p->ts.tv_usec = ts >> 32; - if (p->ts.tv_usec >= 1000000) { - p->ts.tv_usec -= 1000000; - p->ts.tv_sec++; - } - - StatsIncr(ewtn->tv, ewtn->packets); - ewtn->bytes += wlen; - - if (TmThreadsSlotProcessPkt(ewtn->tv, ewtn->slot, p) != TM_ECODE_OK) { - TmqhOutputPacketpool(ewtn->tv, p); - SCReturnInt(TM_ECODE_FAILED); - } - - SCReturnInt(TM_ECODE_OK); -} - -/** - * \brief Print some stats to the log at program exit. - * - * \param tv Pointer to ThreadVars. - * \param data Pointer to data, ErfFileThreadVars. - */ -void -ReceiveErfDagThreadExitStats(ThreadVars *tv, void *data) -{ - ErfDagThreadVars *ewtn = (ErfDagThreadVars *)data; - - (void)SC_ATOMIC_SET(ewtn->livedev->pkts, - StatsGetLocalCounterValue(tv, ewtn->packets)); - (void)SC_ATOMIC_SET(ewtn->livedev->drop, - StatsGetLocalCounterValue(tv, ewtn->drops)); - - SCLogInfo("Stream: %d; Bytes: %"PRIu64"; Packets: %"PRIu64 - "; Drops: %"PRIu64, - ewtn->dagstream, - ewtn->bytes, - StatsGetLocalCounterValue(tv, ewtn->packets), - StatsGetLocalCounterValue(tv, ewtn->drops)); -} - -/** - * \brief Deinitializes the DAG card. - * \param tv pointer to ThreadVars - * \param data pointer that gets cast into PcapThreadVars for ptv - */ -TmEcode -ReceiveErfDagThreadDeinit(ThreadVars *tv, void *data) -{ - SCEnter(); - - ErfDagThreadVars *ewtn = (ErfDagThreadVars *)data; - - ReceiveErfDagCloseStream(ewtn->dagfd, ewtn->dagstream); - - SCReturnInt(TM_ECODE_OK); -} - -void -ReceiveErfDagCloseStream(int dagfd, int stream) -{ - dag_stop_stream(dagfd, stream); - dag_detach_stream(dagfd, stream); - dag_close(dagfd); -} - -/** Decode ErfDag */ - -/** - * \brief This function passes off to link type decoders. - * - * DecodeErfDag reads packets from the PacketQueue and passes - * them off to the proper link type decoder. - * - * \param t pointer to ThreadVars - * \param p pointer to the current packet - * \param data pointer that gets cast into PcapThreadVars for ptv - * \param pq pointer to the current PacketQueue - */ -TmEcode -DecodeErfDag(ThreadVars *tv, Packet *p, void *data, PacketQueue *pq, - PacketQueue *postpq) -{ - SCEnter(); - DecodeThreadVars *dtv = (DecodeThreadVars *)data; - - /* XXX HACK: flow timeout can call us for injected pseudo packets - * see bug: https://redmine.openinfosecfoundation.org/issues/1107 */ - if (p->flags & PKT_PSEUDO_STREAM_END) - return TM_ECODE_OK; - - /* update counters */ - DecodeUpdatePacketCounters(tv, dtv, p); - - /* call the decoder */ - switch(p->datalink) { - case LINKTYPE_ETHERNET: - DecodeEthernet(tv, dtv, p, GET_PKT_DATA(p), GET_PKT_LEN(p), pq); - break; - default: - SCLogError(SC_ERR_DATALINK_UNIMPLEMENTED, - "Error: datalink type %" PRId32 - " not yet supported in module DecodeErfDag", - p->datalink); - break; - } - - PacketDecodeFinalize(tv, dtv, p); - - SCReturnInt(TM_ECODE_OK); -} - -TmEcode -DecodeErfDagThreadInit(ThreadVars *tv, 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); -} - -TmEcode -DecodeErfDagThreadDeinit(ThreadVars *tv, void *data) -{ - if (data != NULL) - DecodeThreadVarsFree(tv, data); - SCReturnInt(TM_ECODE_OK); -} - -#endif /* HAVE_DAG */ |