From 7c729d2d533b6cfc120434c852a9de6992b2d09e Mon Sep 17 00:00:00 2001 From: Anoop Saldanha Date: Fri, 16 Sep 2011 16:38:43 +0530 Subject: [PATCH] some more code cleanup + comments added --- src/Makefile.am | 1 + src/flow-timeout.c | 598 +++++++++++++++++++++++++++++++++++++++++++++ src/flow-timeout.h | 31 +++ src/flow.c | 541 +--------------------------------------- src/tm-threads.c | 13 - 5 files changed, 635 insertions(+), 549 deletions(-) create mode 100644 src/flow-timeout.c create mode 100644 src/flow-timeout.h diff --git a/src/Makefile.am b/src/Makefile.am index ac6966da79..5149e38be8 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -42,6 +42,7 @@ decode-tcp.c decode-tcp.h \ decode-udp.c decode-udp.h \ decode-sctp.c decode-sctp.h \ flow.c flow.h \ +flow-timeout.c flow-timeout.h \ flow-queue.c flow-queue.h \ flow-hash.c flow-hash.h \ flow-util.c flow-util.h \ diff --git a/src/flow-timeout.c b/src/flow-timeout.c new file mode 100644 index 0000000000..01279c7f58 --- /dev/null +++ b/src/flow-timeout.c @@ -0,0 +1,598 @@ +/* Copyright (C) 2007-2011 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 Anoop Saldanha + */ + +#include "suricata-common.h" +#include "suricata.h" +#include "decode.h" +#include "conf.h" +#include "threadvars.h" +#include "tm-threads.h" +#include "runmodes.h" + +#include "util-random.h" +#include "util-time.h" + +#include "flow.h" +#include "flow-queue.h" +#include "flow-hash.h" +#include "flow-util.h" +#include "flow-var.h" +#include "flow-private.h" + +#include "stream-tcp-private.h" +#include "stream-tcp-reassemble.h" +#include "stream-tcp.h" + +#include "util-unittest.h" +#include "util-unittest-helper.h" +#include "util-byte.h" + +#include "util-debug.h" +#include "util-privs.h" + +#include "detect.h" +#include "detect-engine-state.h" +#include "stream.h" + +#include "app-layer-parser.h" + +static TmSlot *stream_pseudo_pkt_stream_tm_slot = NULL; +static ThreadVars *stream_pseudo_pkt_stream_TV = NULL; + +static TmSlot *stream_pseudo_pkt_detect_tm_slot = NULL; +static ThreadVars *stream_pseudo_pkt_detect_TV = NULL; +static ThreadVars *stream_pseudo_pkt_detect_prev_TV = NULL; + +static TmSlot *stream_pseudo_pkt_decode_tm_slot = NULL; +static ThreadVars *stream_pseudo_pkt_decode_TV = NULL; + +/** + * \brief Used to kill flow manager thread(s). + * + * \todo Kinda hackish since it uses the tv name to identify flow manager + * thread. We need an all weather identification scheme. + */ +static inline void FlowKillFlowManagerThread(void) +{ + ThreadVars *tv = NULL; + int cnt = 0; + + SCMutexLock(&tv_root_lock); + + /* flow manager thread(s) is/are a part of mgmt threads */ + tv = tv_root[TVT_MGMT]; + + while (tv != NULL) { + if (strcasecmp(tv->name, "FlowManagerThread") == 0) { + TmThreadsSetFlag(tv, THV_KILL); + TmThreadsSetFlag(tv, THV_DEINIT); + + /* be sure it has shut down */ + while (!TmThreadsCheckFlag(tv, THV_CLOSED)) { + usleep(100); + } + cnt++; + } + tv = tv->next; + } + + /* not possible, unless someone decides to rename FlowManagerThread */ + if (cnt == 0) { + SCMutexUnlock(&tv_root_lock); + abort(); + } + + SCMutexUnlock(&tv_root_lock); + return; +} + +/** + * \internal + * \brief Pseudo packet setup for flow forced reassembly. + * + * \param direction Direction of the packet. 0 indicates toserver and 1 + * indicates toclient. + * \param f Pointer to the flow. + * \param ssn Pointer to the tcp session. + * \param dummy Indicates to create a dummy pseudo packet. Not all pseudo + * packets need to force reassembly, in which case we just + * set dummy ack/seq values. + */ +static inline Packet *FlowForceReassemblyPseudoPacketSetup(int direction, + Flow *f, + TcpSession *ssn, + int dummy) +{ + Packet *p = PacketGetFromAlloc(); + if (p == NULL) + return NULL; + + p->proto = IPPROTO_TCP; + p->flow = f; + FlowIncrUsecnt(f); + p->flags |= PKT_STREAM_EST; + p->flags |= PKT_STREAM_EOF; + p->flags |= PKT_HAS_FLOW; + p->flags |= PKT_PSEUDO_STREAM_END; + if (direction == 0) + p->flowflags |= FLOW_PKT_TOSERVER; + else + p->flowflags |= FLOW_PKT_TOCLIENT; + p->flowflags |= FLOW_PKT_ESTABLISHED; + if (direction == 0) { + COPY_ADDRESS(&f->src, &p->src); + COPY_ADDRESS(&f->dst, &p->dst); + p->sp = f->sp; + p->dp = f->dp; + } else { + COPY_ADDRESS(&f->src, &p->dst); + COPY_ADDRESS(&f->dst, &p->src); + p->sp = f->dp; + p->dp = f->sp; + } + p->payload = NULL; + p->payload_len = 0; + if (f->src.family == AF_INET) { + /* set the ip header */ + p->ip4h = (IPV4Hdr *)p->pkt; + /* version 4 and length 20 bytes for the tcp header */ + p->ip4h->ip_verhl = 0x45; + p->ip4h->ip_tos = 0; + p->ip4h->ip_len = htons(40); + p->ip4h->ip_id = 0; + p->ip4h->ip_off = 0; + p->ip4h->ip_ttl = 64; + p->ip4h->ip_proto = IPPROTO_TCP; + //p->ip4h->ip_csum = + if (direction == 0) { + p->ip4h->ip_src.s_addr = f->src.addr_data32[0]; + p->ip4h->ip_dst.s_addr = f->dst.addr_data32[0]; + } else { + p->ip4h->ip_src.s_addr = f->dst.addr_data32[0]; + p->ip4h->ip_dst.s_addr = f->src.addr_data32[0]; + } + + /* set the tcp header */ + p->tcph = (TCPHdr *)((uint8_t *)p->pkt + 20); + + } else { + /* set the ip header */ + p->ip6h = (IPV6Hdr *)p->pkt; + /* version 6 */ + p->ip6h->s_ip6_vfc = 0x60; + p->ip6h->s_ip6_flow = 0; + p->ip6h->s_ip6_nxt = IPPROTO_TCP; + p->ip6h->s_ip6_plen = htons(20); + p->ip6h->s_ip6_hlim = 64; + if (direction == 0) { + p->ip6h->ip6_src[0] = f->src.addr_data32[0]; + p->ip6h->ip6_src[1] = f->src.addr_data32[0]; + p->ip6h->ip6_src[2] = f->src.addr_data32[0]; + p->ip6h->ip6_src[3] = f->src.addr_data32[0]; + p->ip6h->ip6_dst[0] = f->dst.addr_data32[0]; + p->ip6h->ip6_dst[1] = f->dst.addr_data32[0]; + p->ip6h->ip6_dst[2] = f->dst.addr_data32[0]; + p->ip6h->ip6_dst[3] = f->dst.addr_data32[0]; + } else { + p->ip6h->ip6_src[0] = f->dst.addr_data32[0]; + p->ip6h->ip6_src[1] = f->dst.addr_data32[0]; + p->ip6h->ip6_src[2] = f->dst.addr_data32[0]; + p->ip6h->ip6_src[3] = f->dst.addr_data32[0]; + p->ip6h->ip6_dst[0] = f->src.addr_data32[0]; + p->ip6h->ip6_dst[1] = f->src.addr_data32[0]; + p->ip6h->ip6_dst[2] = f->src.addr_data32[0]; + p->ip6h->ip6_dst[3] = f->src.addr_data32[0]; + } + + /* set the tcp header */ + p->tcph = (TCPHdr *)((uint8_t *)p->pkt + 40); + } + + p->tcph->th_offx2 = 0x50; + p->tcph->th_flags |= TH_ACK; + p->tcph->th_win = 10; + p->tcph->th_urp = 0; + + /* to server */ + if (direction == 0) { + p->tcph->th_sport = htons(f->sp); + p->tcph->th_dport = htons(f->dp); + + if (dummy) { + p->tcph->th_seq = htonl(ssn->client.next_seq); + p->tcph->th_ack = htonl(ssn->server.last_ack); + } else { + p->tcph->th_seq = htonl(ssn->client.next_seq); + p->tcph->th_ack = htonl(ssn->server.seg_list_tail->seq + + ssn->server.seg_list_tail->payload_len); + } + + /* to client */ + } else { + p->tcph->th_sport = htons(f->dp); + p->tcph->th_dport = htons(f->sp); + + if (dummy) { + p->tcph->th_seq = htonl(ssn->server.next_seq); + p->tcph->th_ack = htonl(ssn->client.last_ack); + } else { + p->tcph->th_seq = htonl(ssn->server.next_seq); + p->tcph->th_ack = htonl(ssn->client.seg_list_tail->seq + + ssn->client.seg_list_tail->payload_len); + } + } + + if (f->src.family == AF_INET) { + p->tcph->th_sum = TCPCalculateChecksum((uint16_t *)&(p->ip4h->ip_src), + (uint16_t *)p->tcph, 20); + } else { + p->tcph->th_sum = TCPCalculateChecksum((uint16_t *)&(p->ip6h->ip6_src), + (uint16_t *)p->tcph, 20); + } + + return p; +} + +/** + * \internal + * \brief Forces reassembly for flow if it needs it. + * + * The function requires flow to be locked beforehand. + * + * \param f Pointer to the flow. + * + * \retval 0 This flow doesn't need any reassembly processing; 1 otherwise. + */ +int FlowForceReassemblyForFlowV2(Flow *f) +{ + TcpSession *ssn; + + int client_ok = 1; + int server_ok = 1; + + /* looks like we have no flows in this queue */ + if (f == NULL || f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE) { + return 0; + } + + /* Get the tcp session for the flow */ + ssn = (TcpSession *)f->protoctx; + /* \todo Also skip flows that shouldn't be inspected */ + if (ssn == NULL) { + return 0; + } + + if (!StreamHasUnprocessedSegments(ssn, 0)) { + client_ok = 0; + } + if (!StreamHasUnprocessedSegments(ssn, 1)) { + server_ok = 0; + } + + /* nothing to do */ + if (client_ok == 0 && server_ok == 0) { + return 0; + } + + /* move this unlock after the strream reassemble call */ + SCSpinUnlock(&f->fb->s); + + Packet *p1 = NULL, *p2 = NULL, *p3 = NULL; + + /* The packets we use are based on what segments in what direction are + * unprocessed. + * p1 if we have client segments for reassembly purpose only. If we + * have no server segments p2 can be a toserver packet with dummy + * seq/ack, and if we have server segments p2 has to carry out reassembly + * for server segment as well, in which case we will also need a p3 in the + * toclient which is now dummy since all we need it for is detection */ + + /* insert a pseudo packet in the toserver direction */ + if (client_ok == 1) { + p1 = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 0); + if (p1 == NULL) { + return 1; + } + + if (server_ok == 1) { + p2 = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 0); + if (p2 == NULL) { + TmqhOutputPacketpool(NULL,p1); + return 1; + } + + p3 = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 0); + if (p3 == NULL) { + TmqhOutputPacketpool(NULL, p1); + TmqhOutputPacketpool(NULL, p2); + return 1; + } + } else { + p2 = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 1); + if (p2 == NULL) { + TmqhOutputPacketpool(NULL, p1); + return 1; + } + } + } else { + p1 = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 0); + if (p1 == NULL) { + return 1; + } + + p2 = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 1); + if (p2 == NULL) { + TmqhOutputPacketpool(NULL, p1); + return 1; + } + } + f->flags |= FLOW_TIMEOUT_REASSEMBLY_DONE; + + SCMutexLock(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq.mutex_q); + PacketEnqueue(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq, p1); + PacketEnqueue(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq, p2); + if (p3 != NULL) + PacketEnqueue(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq, p3); + SCMutexUnlock(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq.mutex_q); + if (stream_pseudo_pkt_decode_TV->inq != NULL) { + SCCondSignal(&trans_q[stream_pseudo_pkt_decode_TV->inq->id].cond_q); + } + + return 1; +} + +/** + * \internal + * \brief Forces reassembly for flows that need it. + * + * Please note we don't use locks anywhere. This function is to be + * called right when the engine is not doing anything. + * + * \param q The queue to process flows from. + */ +static inline void FlowForceReassemblyForQ(FlowQueue *q) +{ + Flow *f; + TcpSession *ssn; + int client_ok; + int server_ok; + + /* no locks needed, since the engine is virtually dead. + * We are the kings here */ + + /* get the topmost flow from the QUEUE */ + f = q->top; + + /* we need to loop through all the flows in the queue */ + while (f != NULL) { + /* We use this packet just for reassembly purpose */ + Packet reassemble_p; + memset(&reassemble_p, 0, sizeof(Packet)); + + client_ok = 0; + server_ok = 0; + + /* Get the tcp session for the flow */ + ssn = (TcpSession *)f->protoctx; + + /* \todo Also skip flows that shouldn't be inspected */ + if (ssn == NULL) { + f = f->lnext; + continue; + } + + /* ah ah! We have some unattended toserver segments */ + if (StreamHasUnprocessedSegments(ssn, 0)) { + client_ok = 1; + + StreamTcpThread *stt = stream_pseudo_pkt_stream_tm_slot->slot_data; + + ssn->client.last_ack = (ssn->client.seg_list_tail->seq + + ssn->client.seg_list_tail->payload_len); + reassemble_p.flow = f; + reassemble_p.flags |= PKT_PSEUDO_STREAM_END; + reassemble_p.flowflags = FLOW_PKT_TOCLIENT; + StreamTcpReassembleHandleSegment(stream_pseudo_pkt_detect_TV, + stt->ra_ctx, ssn, &ssn->server, + &reassemble_p, NULL); + StreamTcpReassembleProcessAppLayer(stt->ra_ctx); + } + /* oh oh! We have some unattended toclient segments */ + if (StreamHasUnprocessedSegments(ssn, 1)) { + server_ok = 1; + StreamTcpThread *stt = stream_pseudo_pkt_stream_tm_slot->slot_data; + + ssn->server.last_ack = (ssn->server.seg_list_tail->seq + + ssn->server.seg_list_tail->payload_len); + reassemble_p.flow = f; + reassemble_p.flags |= PKT_PSEUDO_STREAM_END; + reassemble_p.flowflags = FLOW_PKT_TOSERVER; + StreamTcpReassembleHandleSegment(stream_pseudo_pkt_detect_TV, + stt->ra_ctx, ssn, &ssn->client, + &reassemble_p, NULL); + StreamTcpReassembleProcessAppLayer(stt->ra_ctx); + } + + /* insert a pseudo packet in the toserver direction */ + if (client_ok == 1) { + Packet *p = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 1); + if (p == NULL) { + return; + } + + if (stream_pseudo_pkt_detect_prev_TV != NULL) { + stream_pseudo_pkt_detect_prev_TV-> + tmqh_out(stream_pseudo_pkt_detect_prev_TV, p); + } else { + TmSlot *s = stream_pseudo_pkt_detect_tm_slot; + while (s != NULL) { + s->SlotFunc(NULL, p, s->slot_data, &s->slot_pre_pq, + &s->slot_post_pq); + s = s->slot_next; + } + + if (stream_pseudo_pkt_detect_TV != NULL) { + stream_pseudo_pkt_detect_TV-> + tmqh_out(stream_pseudo_pkt_detect_TV, p); + } + } + } /* if (ssn->client.seg_list != NULL) */ + if (server_ok == 1) { + Packet *p = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 1); + if (p == NULL) { + return; + } + + if (stream_pseudo_pkt_detect_prev_TV != NULL) { + stream_pseudo_pkt_detect_prev_TV-> + tmqh_out(stream_pseudo_pkt_detect_prev_TV, p); + } else { + TmSlot *s = stream_pseudo_pkt_detect_tm_slot; + while (s != NULL) { + s->SlotFunc(NULL, p, s->slot_data, &s->slot_pre_pq, + &s->slot_post_pq); + s = s->slot_next; + } + + if (stream_pseudo_pkt_detect_TV != NULL) { + stream_pseudo_pkt_detect_TV-> + tmqh_out(stream_pseudo_pkt_detect_TV, p); + } + } + } /* if (ssn->server.seg_list != NULL) */ + + /* next flow in the queue */ + f = f->lnext; + } /* while (f != NULL) */ + + return; +} + +/** + * \brief Force reassembly for all the flows that have unprocessed segments. + */ +void FlowForceReassembly(void) +{ + /* Do remember. We need to have packet acquire disabled by now */ + + /** ----- Part 1 ----- **/ + /* First we need to kill the flow manager thread */ + FlowKillFlowManagerThread(); + + /** ----- Part 2 ----- **/ + /* Check if all threads are idle. We need this so that we have all + * packets freeds. As a consequence, no flows are in use */ + + SCMutexLock(&tv_root_lock); + + /* all receive threads are part of packet processing threads */ + ThreadVars *tv = tv_root[TVT_PPT]; + + /* we are doing this in order receive -> decode -> ... -> log */ + while (tv != NULL) { + if (tv->inq != NULL) { + /* we wait till we dry out all the inq packets, before we + * kill this thread. Do note that you should have disabled + * packet acquire by now using TmThreadDisableReceiveThreads()*/ + if (!(strlen(tv->inq->name) == strlen("packetpool") && + strcasecmp(tv->inq->name, "packetpool") == 0)) { + PacketQueue *q = &trans_q[tv->inq->id]; + while (q->len != 0) { + usleep(100); + } + } + } + tv = tv->next; + } + + SCMutexUnlock(&tv_root_lock); + + /** ----- Part 3 ----- **/ + /* Carry out flow reassembly for unattended flows */ + FlowForceReassemblyForQ(&flow_new_q[FLOW_PROTO_TCP]); + FlowForceReassemblyForQ(&flow_est_q[FLOW_PROTO_TCP]); + FlowForceReassemblyForQ(&flow_close_q[FLOW_PROTO_TCP]); + + return; +} + +void FlowForceReassemblySetup(void) +{ + /* get StreamTCP TM's slot and TV containing this slot */ + stream_pseudo_pkt_stream_tm_slot = TmSlotGetSlotForTM(TMM_STREAMTCP); + if (stream_pseudo_pkt_stream_tm_slot == NULL) { + /* yes, this is fatal! */ + SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " + "retrieve the slot for STREAMTCP TM"); + exit(EXIT_FAILURE); + } + stream_pseudo_pkt_stream_TV = + TmThreadsGetTVContainingSlot(stream_pseudo_pkt_stream_tm_slot); + if (stream_pseudo_pkt_stream_TV == NULL) { + /* yes, this is fatal! */ + SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " + "retrieve the TV containing STREAMTCP TM slot"); + exit(EXIT_FAILURE); + } + + /* get detect TM's slot and TV containing this slot */ + stream_pseudo_pkt_detect_tm_slot = TmSlotGetSlotForTM(TMM_DETECT); + if (stream_pseudo_pkt_detect_tm_slot == NULL) { + /* yes, this is fatal! */ + SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " + "retrieve a slot for DETECT TM"); + exit(EXIT_FAILURE); + } + stream_pseudo_pkt_detect_TV = + TmThreadsGetTVContainingSlot(stream_pseudo_pkt_detect_tm_slot); + if (stream_pseudo_pkt_detect_TV == NULL) { + /* yes, this is fatal! */ + SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " + "retrieve the TV containing the Detect TM slot"); + exit(EXIT_FAILURE); + } + if (stream_pseudo_pkt_detect_TV->tm_slots == stream_pseudo_pkt_detect_tm_slot) { + stream_pseudo_pkt_detect_prev_TV = stream_pseudo_pkt_detect_TV->prev; + } + if (stream_pseudo_pkt_detect_TV->next == NULL) { + stream_pseudo_pkt_detect_TV = NULL; + } + + stream_pseudo_pkt_decode_tm_slot = TmThreadGetFirstTmSlotForPartialPattern("decode"); + if (stream_pseudo_pkt_decode_tm_slot == NULL) { + /* yes, this is fatal! */ + SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " + "retrieve the slot for DECODE TM"); + exit(EXIT_FAILURE); + } + stream_pseudo_pkt_decode_TV = + TmThreadsGetTVContainingSlot(stream_pseudo_pkt_decode_tm_slot); + if (stream_pseudo_pkt_decode_TV == NULL) { + /* yes, this is fatal! */ + SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " + "retrieve the TV containing the Decode TM slot"); + exit(EXIT_FAILURE); + } + + return; +} diff --git a/src/flow-timeout.h b/src/flow-timeout.h new file mode 100644 index 0000000000..0da6762c67 --- /dev/null +++ b/src/flow-timeout.h @@ -0,0 +1,31 @@ +/* Copyright (C) 2007-2011 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 Anoop Saldanha + */ + +#ifndef __FLOW_TIMEOUT_H__ +#define __FLOW_TIMEOUT_H__ + +int FlowForceReassemblyForFlowV2(Flow *); +void FlowForceReassembly(void); +void FlowForceReassemblySetup(void); + +#endif /* __FLOW_TIMEOUT_H__ */ diff --git a/src/flow.c b/src/flow.c index 73c3c24174..84a681f7c2 100644 --- a/src/flow.c +++ b/src/flow.c @@ -43,6 +43,7 @@ #include "flow-util.h" #include "flow-var.h" #include "flow-private.h" +#include "flow-timeout.h" #include "stream-tcp-private.h" #include "stream-tcp-reassemble.h" @@ -85,16 +86,6 @@ int FlowKill(FlowQueue *); /* Run mode selected at suricata.c */ extern int run_mode; -static TmSlot *stream_pseudo_pkt_stream_tm_slot = NULL; -static ThreadVars *stream_pseudo_pkt_stream_TV = NULL; - -static TmSlot *stream_pseudo_pkt_detect_tm_slot = NULL; -static ThreadVars *stream_pseudo_pkt_detect_TV = NULL; -static ThreadVars *stream_pseudo_pkt_detect_prev_TV = NULL; - -static TmSlot *stream_pseudo_pkt_decode_tm_slot = NULL; -static ThreadVars *stream_pseudo_pkt_decode_TV = NULL; - /** \brief Initialize the l7data ptr in the Flow session used by the L7 Modules * for data storage. * @@ -199,253 +190,6 @@ static uint64_t prune_no_timeout = 0; static uint64_t prune_usecnt = 0; #endif -/** - * \internal - * \brief Pseudo packet setup for flow forced reassembly. - * - * \param direction Direction of the packet. 0 indicates toserver and 1 - * indicates toclient. - * \param f Pointer to the flow. - * \param ssn Pointer to the tcp session. - * \param dummy Indicates to create a dummy pseudo packet. Not all pseudo - * packets need to force reassembly, in which case we just - * set dummy ack/seq values. - */ -static inline Packet *FlowForceReassemblyPseudoPacketSetup(int direction, - Flow *f, - TcpSession *ssn, - int dummy) -{ - Packet *p = PacketGetFromAlloc(); - if (p == NULL) - return NULL; - - p->proto = IPPROTO_TCP; - p->flow = f; - FlowIncrUsecnt(f); - p->flags |= PKT_STREAM_EST; - p->flags |= PKT_STREAM_EOF; - p->flags |= PKT_HAS_FLOW; - p->flags |= PKT_PSEUDO_STREAM_END; - if (direction == 0) - p->flowflags |= FLOW_PKT_TOSERVER; - else - p->flowflags |= FLOW_PKT_TOCLIENT; - p->flowflags |= FLOW_PKT_ESTABLISHED; - if (direction == 0) { - COPY_ADDRESS(&f->src, &p->src); - COPY_ADDRESS(&f->dst, &p->dst); - p->sp = f->sp; - p->dp = f->dp; - } else { - COPY_ADDRESS(&f->src, &p->dst); - COPY_ADDRESS(&f->dst, &p->src); - p->sp = f->dp; - p->dp = f->sp; - } - p->payload = NULL; - p->payload_len = 0; - if (f->src.family == AF_INET) { - /* set the ip header */ - p->ip4h = (IPV4Hdr *)p->pkt; - /* version 4 and length 20 bytes for the tcp header */ - p->ip4h->ip_verhl = 0x45; - p->ip4h->ip_tos = 0; - p->ip4h->ip_len = htons(40); - p->ip4h->ip_id = 0; - p->ip4h->ip_off = 0; - p->ip4h->ip_ttl = 64; - p->ip4h->ip_proto = IPPROTO_TCP; - //p->ip4h->ip_csum = - if (direction == 0) { - p->ip4h->ip_src.s_addr = f->src.addr_data32[0]; - p->ip4h->ip_dst.s_addr = f->dst.addr_data32[0]; - } else { - p->ip4h->ip_src.s_addr = f->dst.addr_data32[0]; - p->ip4h->ip_dst.s_addr = f->src.addr_data32[0]; - } - - /* set the tcp header */ - p->tcph = (TCPHdr *)((uint8_t *)p->pkt + 20); - - } else { - /* set the ip header */ - p->ip6h = (IPV6Hdr *)p->pkt; - /* version 6 */ - p->ip6h->s_ip6_vfc = 0x60; - p->ip6h->s_ip6_flow = 0; - p->ip6h->s_ip6_nxt = IPPROTO_TCP; - p->ip6h->s_ip6_plen = htons(20); - p->ip6h->s_ip6_hlim = 64; - if (direction == 0) { - p->ip6h->ip6_src[0] = f->src.addr_data32[0]; - p->ip6h->ip6_src[1] = f->src.addr_data32[0]; - p->ip6h->ip6_src[2] = f->src.addr_data32[0]; - p->ip6h->ip6_src[3] = f->src.addr_data32[0]; - p->ip6h->ip6_dst[0] = f->dst.addr_data32[0]; - p->ip6h->ip6_dst[1] = f->dst.addr_data32[0]; - p->ip6h->ip6_dst[2] = f->dst.addr_data32[0]; - p->ip6h->ip6_dst[3] = f->dst.addr_data32[0]; - } else { - p->ip6h->ip6_src[0] = f->dst.addr_data32[0]; - p->ip6h->ip6_src[1] = f->dst.addr_data32[0]; - p->ip6h->ip6_src[2] = f->dst.addr_data32[0]; - p->ip6h->ip6_src[3] = f->dst.addr_data32[0]; - p->ip6h->ip6_dst[0] = f->src.addr_data32[0]; - p->ip6h->ip6_dst[1] = f->src.addr_data32[0]; - p->ip6h->ip6_dst[2] = f->src.addr_data32[0]; - p->ip6h->ip6_dst[3] = f->src.addr_data32[0]; - } - - /* set the tcp header */ - p->tcph = (TCPHdr *)((uint8_t *)p->pkt + 40); - } - - p->tcph->th_offx2 = 0x50; - p->tcph->th_flags |= TH_ACK; - p->tcph->th_win = 10; - p->tcph->th_urp = 0; - - /* to server */ - if (direction == 0) { - p->tcph->th_sport = htons(f->sp); - p->tcph->th_dport = htons(f->dp); - - if (dummy) { - p->tcph->th_seq = htonl(ssn->client.next_seq); - p->tcph->th_ack = htonl(ssn->server.last_ack); - } else { - p->tcph->th_seq = htonl(ssn->client.next_seq); - p->tcph->th_ack = htonl(ssn->server.seg_list_tail->seq + - ssn->server.seg_list_tail->payload_len); - } - - /* to client */ - } else { - p->tcph->th_sport = htons(f->dp); - p->tcph->th_dport = htons(f->sp); - - if (dummy) { - p->tcph->th_seq = htonl(ssn->server.next_seq); - p->tcph->th_ack = htonl(ssn->client.last_ack); - } else { - p->tcph->th_seq = htonl(ssn->server.next_seq); - p->tcph->th_ack = htonl(ssn->client.seg_list_tail->seq + - ssn->client.seg_list_tail->payload_len); - } - } - - if (f->src.family == AF_INET) { - p->tcph->th_sum = TCPCalculateChecksum((uint16_t *)&(p->ip4h->ip_src), - (uint16_t *)p->tcph, 20); - } else { - p->tcph->th_sum = TCPCalculateChecksum((uint16_t *)&(p->ip6h->ip6_src), - (uint16_t *)p->tcph, 20); - } - - return p; -} - -/** - * \internal - * \brief Forces reassembly for flow if it needs it. - * - * The function requires flow to be locked beforehand. - * - * \param f Pointer to the flow. - * - * \retval 0 This flow doesn't need any reassembly processing; 1 otherwise. - */ -static inline int FlowForceReassemblyForFlowV2(Flow *f) -{ - TcpSession *ssn; - - int client_ok = 1; - int server_ok = 1; - - /* looks like we have no flows in this queue */ - if (f == NULL || f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE) { - return 0; - } - - /* Get the tcp session for the flow */ - ssn = (TcpSession *)f->protoctx; - /* \todo Also skip flows that shouldn't be inspected */ - if (ssn == NULL) { - return 0; - } - - if (!StreamHasUnprocessedSegments(ssn, 0)) { - client_ok = 0; - } - if (!StreamHasUnprocessedSegments(ssn, 1)) { - server_ok = 0; - } - - /* nothing to do */ - if (client_ok == 0 && server_ok == 0) { - return 0; - } - - /* move this unlock after the strream reassemble call */ - SCSpinUnlock(&f->fb->s); - - Packet *p1 = NULL, *p2 = NULL, *p3 = NULL; - - /* insert a pseudo packet in the toserver direction */ - if (client_ok == 1) { - p1 = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 0); - if (p1 == NULL) { - return 1; - } - - if (server_ok == 1) { - p2 = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 0); - if (p2 == NULL) { - TmqhOutputPacketpool(NULL,p1); - return 1; - } - - p3 = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 0); - if (p3 == NULL) { - TmqhOutputPacketpool(NULL, p1); - TmqhOutputPacketpool(NULL, p2); - return 1; - } - } else { - p2 = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 1); - if (p2 == NULL) { - TmqhOutputPacketpool(NULL, p1); - return 1; - } - } - } else { - p1 = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 0); - if (p1 == NULL) { - return 1; - } - - p2 = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 1); - if (p2 == NULL) { - TmqhOutputPacketpool(NULL, p1); - return 1; - } - } - f->flags |= FLOW_TIMEOUT_REASSEMBLY_DONE; - - SCMutexLock(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq.mutex_q); - PacketEnqueue(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq, p1); - PacketEnqueue(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq, p2); - if (p3 != NULL) - PacketEnqueue(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq, p3); - SCMutexUnlock(&stream_pseudo_pkt_decode_tm_slot->slot_post_pq.mutex_q); - if (stream_pseudo_pkt_decode_TV->inq != NULL) { - SCCondSignal(&trans_q[stream_pseudo_pkt_decode_TV->inq->id].cond_q); - } - - return 1; -} - /** FlowPrune * * Inspect top (last recently used) flow from the queue and see if @@ -453,9 +197,10 @@ static inline int FlowForceReassemblyForFlowV2(Flow *f) * * Use trylock here so prevent us from blocking the packet handling. * - * \param q flow queue to prune - * \param ts current time - * \param timeout timeout to enforce + * \param q Flow queue to prune + * \param ts Current time + * \param timeout Timeout to enforce + * \param try_cnt Tries to prune the first try_cnt no of flows in the q * * \retval 0 on error, failed block, nothing to prune * \retval 1 on successfully pruned one @@ -1313,282 +1058,6 @@ void FlowShutdown(void) return; } -/** - * \internal - * \brief Forces reassembly for flows that need it. - * - * Please note we don't use locks anywhere. This function is to be - * called right when the engine is not doing anything. - * - * \param q The queue to process flows from. - */ -static inline void FlowForceReassemblyForQ(FlowQueue *q) -{ - Flow *f; - TcpSession *ssn; - int client_ok; - int server_ok; - - /* no locks needed, since the engine is virtually dead. - * We are the kings here */ - - /* get the topmost flow from the QUEUE */ - f = q->top; - - /* we need to loop through all the flows in the queue */ - while (f != NULL) { - /* We use this packet just for reassembly purpose */ - Packet reassemble_p; - memset(&reassemble_p, 0, sizeof(Packet)); - - client_ok = 0; - server_ok = 0; - - /* Get the tcp session for the flow */ - ssn = (TcpSession *)f->protoctx; - - /* \todo Also skip flows that shouldn't be inspected */ - if (ssn == NULL) { - f = f->lnext; - continue; - } - - /* ah ah! We have some unattended toserver segments */ - if (StreamHasUnprocessedSegments(ssn, 0)) { - client_ok = 1; - - StreamTcpThread *stt = stream_pseudo_pkt_stream_tm_slot->slot_data; - - ssn->client.last_ack = (ssn->client.seg_list_tail->seq + - ssn->client.seg_list_tail->payload_len); - reassemble_p.flow = f; - reassemble_p.flags |= PKT_PSEUDO_STREAM_END; - reassemble_p.flowflags = FLOW_PKT_TOCLIENT; - StreamTcpReassembleHandleSegment(stream_pseudo_pkt_detect_TV, - stt->ra_ctx, ssn, &ssn->server, - &reassemble_p, NULL); - StreamTcpReassembleProcessAppLayer(stt->ra_ctx); - } - /* oh oh! We have some unattended toclient segments */ - if (StreamHasUnprocessedSegments(ssn, 1)) { - server_ok = 1; - StreamTcpThread *stt = stream_pseudo_pkt_stream_tm_slot->slot_data; - - ssn->server.last_ack = (ssn->server.seg_list_tail->seq + - ssn->server.seg_list_tail->payload_len); - reassemble_p.flow = f; - reassemble_p.flags |= PKT_PSEUDO_STREAM_END; - reassemble_p.flowflags = FLOW_PKT_TOSERVER; - StreamTcpReassembleHandleSegment(stream_pseudo_pkt_detect_TV, - stt->ra_ctx, ssn, &ssn->client, - &reassemble_p, NULL); - StreamTcpReassembleProcessAppLayer(stt->ra_ctx); - } - - /* insert a pseudo packet in the toserver direction */ - if (client_ok == 1) { - Packet *p = FlowForceReassemblyPseudoPacketSetup(0, f, ssn, 1); - if (p == NULL) { - return; - } - - if (stream_pseudo_pkt_detect_prev_TV != NULL) { - stream_pseudo_pkt_detect_prev_TV-> - tmqh_out(stream_pseudo_pkt_detect_prev_TV, p); - } else { - TmSlot *s = stream_pseudo_pkt_detect_tm_slot; - while (s != NULL) { - s->SlotFunc(NULL, p, s->slot_data, &s->slot_pre_pq, - &s->slot_post_pq); - s = s->slot_next; - } - - if (stream_pseudo_pkt_detect_TV != NULL) { - stream_pseudo_pkt_detect_TV-> - tmqh_out(stream_pseudo_pkt_detect_TV, p); - } - } - } /* if (ssn->client.seg_list != NULL) */ - if (server_ok == 1) { - Packet *p = FlowForceReassemblyPseudoPacketSetup(1, f, ssn, 1); - if (p == NULL) { - return; - } - - if (stream_pseudo_pkt_detect_prev_TV != NULL) { - stream_pseudo_pkt_detect_prev_TV-> - tmqh_out(stream_pseudo_pkt_detect_prev_TV, p); - } else { - TmSlot *s = stream_pseudo_pkt_detect_tm_slot; - while (s != NULL) { - s->SlotFunc(NULL, p, s->slot_data, &s->slot_pre_pq, - &s->slot_post_pq); - s = s->slot_next; - } - - if (stream_pseudo_pkt_detect_TV != NULL) { - stream_pseudo_pkt_detect_TV-> - tmqh_out(stream_pseudo_pkt_detect_TV, p); - } - } - } /* if (ssn->server.seg_list != NULL) */ - - /* next flow in the queue */ - f = f->lnext; - } /* while (f != NULL) */ - - return; -} - -/** - * \brief Used to kill flow manager thread(s). - * - * \todo Kinda hackish since it uses the tv name to identify flow manager - * thread. We need an all weather identification scheme. - */ -static inline void FlowKillFlowManagerThread(void) -{ - ThreadVars *tv = NULL; - int cnt = 0; - - SCMutexLock(&tv_root_lock); - - /* flow manager thread(s) is/are a part of mgmt threads */ - tv = tv_root[TVT_MGMT]; - - while (tv != NULL) { - if (strcasecmp(tv->name, "FlowManagerThread") == 0) { - TmThreadsSetFlag(tv, THV_KILL); - TmThreadsSetFlag(tv, THV_DEINIT); - - /* be sure it has shut down */ - while (!TmThreadsCheckFlag(tv, THV_CLOSED)) { - usleep(100); - } - cnt++; - } - tv = tv->next; - } - - /* not possible, unless someone decides to rename FlowManagerThread */ - if (cnt == 0) { - SCMutexUnlock(&tv_root_lock); - abort(); - } - - SCMutexUnlock(&tv_root_lock); - return; -} - -/** - * \brief Force reassembly for all the flows that have unprocessed segments. - */ -void FlowForceReassembly(void) -{ - /* Do remember. We need to have packet acquire disabled by now */ - - /** ----- Part 1 ----- **/ - /* First we need to kill the flow manager thread */ - FlowKillFlowManagerThread(); - - /** ----- Part 2 ----- **/ - /* Check if all threads are idle. We need this so that we have all - * packets freeds. As a consequence, no flows are in use */ - - SCMutexLock(&tv_root_lock); - - /* all receive threads are part of packet processing threads */ - ThreadVars *tv = tv_root[TVT_PPT]; - - /* we are doing this in order receive -> decode -> ... -> log */ - while (tv != NULL) { - if (tv->inq != NULL) { - /* we wait till we dry out all the inq packets, before we - * kill this thread. Do note that you should have disabled - * packet acquire by now using TmThreadDisableReceiveThreads()*/ - if (!(strlen(tv->inq->name) == strlen("packetpool") && - strcasecmp(tv->inq->name, "packetpool") == 0)) { - PacketQueue *q = &trans_q[tv->inq->id]; - while (q->len != 0) { - usleep(100); - } - } - } - tv = tv->next; - } - - SCMutexUnlock(&tv_root_lock); - - /** ----- Part 3 ----- **/ - /* Carry out flow reassembly for unattended flows */ - FlowForceReassemblyForQ(&flow_new_q[FLOW_PROTO_TCP]); - FlowForceReassemblyForQ(&flow_est_q[FLOW_PROTO_TCP]); - FlowForceReassemblyForQ(&flow_close_q[FLOW_PROTO_TCP]); - - return; -} - -void FlowForceReassemblySetup(void) -{ - /* get StreamTCP TM's slot and TV containing this slot */ - stream_pseudo_pkt_stream_tm_slot = TmSlotGetSlotForTM(TMM_STREAMTCP); - if (stream_pseudo_pkt_stream_tm_slot == NULL) { - /* yes, this is fatal! */ - SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " - "retrieve the slot for STREAMTCP TM"); - exit(EXIT_FAILURE); - } - stream_pseudo_pkt_stream_TV = - TmThreadsGetTVContainingSlot(stream_pseudo_pkt_stream_tm_slot); - if (stream_pseudo_pkt_stream_TV == NULL) { - /* yes, this is fatal! */ - SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " - "retrieve the TV containing STREAMTCP TM slot"); - exit(EXIT_FAILURE); - } - - /* get detect TM's slot and TV containing this slot */ - stream_pseudo_pkt_detect_tm_slot = TmSlotGetSlotForTM(TMM_DETECT); - if (stream_pseudo_pkt_detect_tm_slot == NULL) { - /* yes, this is fatal! */ - SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " - "retrieve a slot for DETECT TM"); - exit(EXIT_FAILURE); - } - stream_pseudo_pkt_detect_TV = - TmThreadsGetTVContainingSlot(stream_pseudo_pkt_detect_tm_slot); - if (stream_pseudo_pkt_detect_TV == NULL) { - /* yes, this is fatal! */ - SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " - "retrieve the TV containing the Detect TM slot"); - exit(EXIT_FAILURE); - } - if (stream_pseudo_pkt_detect_TV->tm_slots == stream_pseudo_pkt_detect_tm_slot) { - stream_pseudo_pkt_detect_prev_TV = stream_pseudo_pkt_detect_TV->prev; - } - if (stream_pseudo_pkt_detect_TV->next == NULL) { - stream_pseudo_pkt_detect_TV = NULL; - } - - stream_pseudo_pkt_decode_tm_slot = TmThreadGetFirstTmSlotForPartialPattern("decode"); - if (stream_pseudo_pkt_decode_tm_slot == NULL) { - /* yes, this is fatal! */ - SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " - "retrieve the slot for DECODE TM"); - exit(EXIT_FAILURE); - } - stream_pseudo_pkt_decode_TV = - TmThreadsGetTVContainingSlot(stream_pseudo_pkt_decode_tm_slot); - if (stream_pseudo_pkt_decode_TV == NULL) { - /* yes, this is fatal! */ - SCLogError(SC_ERR_TM_MODULES_ERROR, "Looks like we have failed to " - "retrieve the TV containing the Decode TM slot"); - exit(EXIT_FAILURE); - } - - return; -} - /** \brief Thread that manages the various queue's and removes timed out flows. * \param td ThreadVars casted to void ptr * diff --git a/src/tm-threads.c b/src/tm-threads.c index 264c073067..ef4c89e4ef 100644 --- a/src/tm-threads.c +++ b/src/tm-threads.c @@ -840,19 +840,6 @@ void TmSlotSetFuncAppend(ThreadVars *tv, TmModule *tm, void *data) } } -#ifdef PROFILING - TmModule *prof_tm; - int prof_tm_id = 0; - - for (prof_tm_id = 0; prof_tm_id < TMM_SIZE; prof_tm_id++) { - prof_tm = &tmm_modules[prof_tm_id]; - - if (prof_tm == tm) { - slot->tm_id = prof_tm_id; - break; - } - } -#endif return; }