From ced5157dc9f7b76e2a253dca9998fca4e42278ed Mon Sep 17 00:00:00 2001 From: Gurvinder Singh Date: Wed, 2 Sep 2009 20:43:59 +0300 Subject: [PATCH] Flow get state protocol specific --- src/eidps.c | 4 ++- src/flow.c | 83 ++++++++++++++++++++++++++++-------------------- src/flow.h | 8 +++++ src/stream-tcp.c | 45 ++++++++++++++++++++------ src/stream-tcp.h | 1 + 5 files changed, 96 insertions(+), 45 deletions(-) diff --git a/src/eidps.c b/src/eidps.c index a7352b0d33..03c6585cf7 100644 --- a/src/eidps.c +++ b/src/eidps.c @@ -1182,7 +1182,6 @@ int main(int argc, char **argv) TmModuleLogHttplogRegister(); TmModuleLogHttplogIPv4Register(); TmModuleLogHttplogIPv6Register(); - TmModuleStreamTcpRegister(); TmModuleDebugList(); #ifdef UNITTESTS @@ -1213,6 +1212,7 @@ int main(int argc, char **argv) AlpDetectRegisterTests(); ConfRegisterTests(); TmqhFlowRegisterTests(); + StreamTcpRegisterTests(); FlowRegisterTests(); uint32_t failed = UtRunTests(); UtCleanup(); @@ -1273,6 +1273,8 @@ int main(int argc, char **argv) /* Spawn the flow manager thread */ FlowManagerThreadSpawn(); + TmModuleStreamTcpRegister(); + /* Spawn the L7 App Detect thread */ AppLayerDetectProtoThreadSpawn(); diff --git a/src/flow.c b/src/flow.c index b548d6a77a..62a32f99c0 100644 --- a/src/flow.c +++ b/src/flow.c @@ -45,6 +45,7 @@ int FlowSetProtoEmergencyTimeout(uint8_t , uint32_t ,uint32_t ); static int FlowGetProtoMapping(uint8_t ); static int FlowClearMemory(Flow *,uint8_t ); int FlowSetProtoFreeFunc(uint8_t, void (*Free)(void *)); +int FlowSetProtoFlowStateFunc (uint8_t , int (*GetFlowState)(void *)); /** \brief Update the flows position in the queue's * \param f Flow to requeue. * @@ -113,16 +114,31 @@ static int FlowPrune (FlowQueue *q, struct timeval *ts) uint8_t proto_map; proto_map = FlowGetProtoMapping(f->proto); - if (!(FlowUpdateSpareFlows()) && (flow_flags & FLOW_EMERGENCY)) { - if (f->flags & FLOW_EST_LIST) - timeout = protocols[proto_map].emerg_est_timeout; - else - timeout = protocols[proto_map].emerg_new_timeout; + if (flow_flags & FLOW_EMERGENCY) { + if (protocols[proto_map].GetFlowState != NULL) { + if ((protocols[proto_map].GetFlowState(f->stream)) == FLOW_STATE_ESTABLISHED) + timeout = protocols[proto_map].emerg_est_timeout; + else + timeout = protocols[proto_map].emerg_new_timeout; + } else { + if (f->flags & FLOW_EST_LIST) + timeout = protocols[proto_map].emerg_est_timeout; + else + timeout = protocols[proto_map].emerg_new_timeout; + } + } else { - if (f->flags & FLOW_EST_LIST) - timeout = protocols[proto_map].est_timeout; - else - timeout = protocols[proto_map].new_timeout; + if (protocols[proto_map].GetFlowState != NULL) { + if ((protocols[proto_map].GetFlowState(f->stream)) == FLOW_STATE_ESTABLISHED) + timeout = protocols[proto_map].emerg_est_timeout; + else + timeout = protocols[proto_map].emerg_new_timeout; + } else { + if (f->flags & FLOW_EST_LIST) + timeout = protocols[proto_map].emerg_est_timeout; + else + timeout = protocols[proto_map].emerg_new_timeout; + } } DEBUGPRINT("got lock, now check: %" PRId64 "+%" PRIu32 "=(%" PRId64 ") < %" PRId64 "", f->lastts.tv_sec, @@ -447,11 +463,11 @@ void *FlowManagerThread(void *td) TimeGet(&ts); DEBUGPRINT("ts %" PRId64 "", ts.tv_sec); - /* see if we still have enough spare flows + /* see if we still have enough spare flows */ if (!(FlowUpdateSpareFlows()) && emerg == TRUE) { - timeout_new = flow_config.emerg_timeout_new; - timeout_est = flow_config.emerg_timeout_est; - }*/ + /*timeout_new = flow_config.emerg_timeout_new; + timeout_est = flow_config.emerg_timeout_est;*/ + } /* prune new list */ nowcnt = FlowPruneFlows(&flow_new_q, &ts); @@ -516,25 +532,28 @@ void FlowInitProtocols(void) { protocols[FLOW_PROTO_DEFAULT].emerg_new_timeout = FLOW_DEFAULT_EMERG_NEW_TIMEOUT; protocols[FLOW_PROTO_DEFAULT].emerg_est_timeout = FLOW_DEFAULT_EMERG_EST_TIMEOUT; protocols[FLOW_PROTO_DEFAULT].Freefunc = NULL; + protocols[FLOW_PROTO_DEFAULT].GetFlowState = NULL; /*TCP*/ protocols[FLOW_PROTO_TCP].new_timeout = FLOW_IPPROTO_TCP_NEW_TIMEOUT; protocols[FLOW_PROTO_TCP].est_timeout = FLOW_IPPROTO_TCP_EST_TIMEOUT; protocols[FLOW_PROTO_TCP].emerg_new_timeout = FLOW_IPPROTO_TCP_EMERG_NEW_TIMEOUT; protocols[FLOW_PROTO_TCP].emerg_est_timeout = FLOW_IPPROTO_TCP_EMERG_EST_TIMEOUT; protocols[FLOW_PROTO_TCP].Freefunc = NULL; + protocols[FLOW_PROTO_TCP].GetFlowState = NULL; /*UDP*/ protocols[FLOW_PROTO_UDP].new_timeout = FLOW_IPPROTO_UDP_NEW_TIMEOUT; protocols[FLOW_PROTO_UDP].est_timeout = FLOW_IPPROTO_UDP_EST_TIMEOUT; protocols[FLOW_PROTO_UDP].emerg_new_timeout = FLOW_IPPROTO_UDP_EMERG_NEW_TIMEOUT; protocols[FLOW_PROTO_UDP].emerg_est_timeout = FLOW_IPPROTO_UDP_EMERG_EST_TIMEOUT; protocols[FLOW_PROTO_UDP].Freefunc = NULL; + protocols[FLOW_PROTO_UDP].GetFlowState = NULL; /*ICMP*/ protocols[FLOW_PROTO_ICMP].new_timeout = FLOW_IPPROTO_ICMP_NEW_TIMEOUT; protocols[FLOW_PROTO_ICMP].est_timeout = FLOW_IPPROTO_ICMP_EST_TIMEOUT; protocols[FLOW_PROTO_ICMP].emerg_new_timeout = FLOW_IPPROTO_ICMP_EMERG_NEW_TIMEOUT; protocols[FLOW_PROTO_ICMP].emerg_est_timeout = FLOW_IPPROTO_ICMP_EMERG_EST_TIMEOUT; protocols[FLOW_PROTO_ICMP].Freefunc = NULL; - + protocols[FLOW_PROTO_ICMP].GetFlowState = NULL; } static int FlowClearMemory(Flow* f, uint8_t proto_map) { @@ -555,6 +574,15 @@ int FlowSetProtoFreeFunc (uint8_t proto, void (*Free)(void *)) { return 1; } +int FlowSetProtoFlowStateFunc (uint8_t proto, int (*GetFlowState)(void *)) { + + uint8_t proto_map; + proto_map = FlowGetProtoMapping(proto); + + protocols[proto_map].GetFlowState = GetFlowState; + return 1; +} + int FlowSetProtoTimeout(uint8_t proto, uint32_t new_timeout, uint32_t est_timeout) { uint8_t proto_map; @@ -562,7 +590,7 @@ int FlowSetProtoTimeout(uint8_t proto, uint32_t new_timeout, uint32_t est_timeou protocols[proto_map].new_timeout = new_timeout; protocols[proto_map].est_timeout = est_timeout; - printf("The time out is %"PRIu32"\n",protocols[FLOW_PROTO_TCP].est_timeout); + return 1; } @@ -695,11 +723,7 @@ static int FlowTest03 (void) { memset(&fb, 0, sizeof(FlowBucket)); TimeGet(&ts); - f.flags = FLOW_EST_LIST; - /*The value should be more than 3600s but as the FlowInitConfig() - is called Decodeppptests(), it reinitalize the flow timeout values to - defaults.*/ - f.lastts.tv_sec = ts.tv_sec - 500; + f.lastts.tv_sec = ts.tv_sec - 5000; f.stream = &ssn; f.fb = &fb; f.proto = IPPROTO_TCP; @@ -736,11 +760,7 @@ static int FlowTest04 (void) { ssn.client = client; ssn.server = client; ssn.state = TCP_ESTABLISHED; - f.flags = FLOW_EST_LIST; - /*The value should be more than 3600s but as the FlowInitConfig() - is called Decodeppptests(), it reinitalize the flow timeout values to - defaults.*/ - f.lastts.tv_sec = ts.tv_sec - 500; + f.lastts.tv_sec = ts.tv_sec - 5000; f.stream = &ssn; f.fb = &fb; f.proto = IPPROTO_TCP; @@ -765,11 +785,8 @@ static int FlowTest05 (void) { memset(&fb, 0, sizeof(FlowBucket)); TimeGet(&ts); - f.flags = FLOW_EST_LIST; - /*The value should be more than 300s but as the FlowInitConfig() - is called Decodeppptests(), it reinitalize the flow timeout values to - defaults.*/ - f.lastts.tv_sec = ts.tv_sec - 150; + ssn.state = TCP_SYN_SENT; + f.lastts.tv_sec = ts.tv_sec - 300; f.stream = &ssn; f.fb = &fb; f.proto = IPPROTO_TCP; @@ -807,11 +824,7 @@ static int FlowTest06 (void) { ssn.client = client; ssn.server = client; ssn.state = TCP_ESTABLISHED; - f.flags = FLOW_EST_LIST; - /*The value should be more than 300s but as the FlowInitConfig() - is called Decodeppptests(), it reinitalize the flow timeout values to - defaults.*/ - f.lastts.tv_sec = ts.tv_sec - 150; + f.lastts.tv_sec = ts.tv_sec - 5000; f.stream = &ssn; f.fb = &fb; f.proto = IPPROTO_TCP; diff --git a/src/flow.h b/src/flow.h index 34ac9e629b..4331ad7cdc 100644 --- a/src/flow.h +++ b/src/flow.h @@ -91,12 +91,19 @@ enum { FLOW_PROTO_ICMP, }; +enum { + FLOW_STATE_NEW = 0, + FLOW_STATE_ESTABLISHED, + FLOW_STATE_CLOSED, +}; + typedef struct Protocols_ { uint32_t new_timeout; uint32_t est_timeout; uint32_t emerg_new_timeout; uint32_t emerg_est_timeout; void (*Freefunc)(void *); + int (*GetFlowState)(void *); }Protocols; void FlowHandlePacket (ThreadVars *, Packet *); @@ -113,6 +120,7 @@ void FlowRegisterTests (void); int FlowSetProtoTimeout(uint8_t ,uint32_t ,uint32_t ); int FlowSetProtoEmergencyTimeout(uint8_t ,uint32_t ,uint32_t ); int FlowSetProtoFreeFunc (uint8_t , void (*Free)(void *)); +int FlowSetProtoFlowStateFunc (uint8_t , int (*GetFlowState)(void *)); #endif /* __FLOW_H__ */ diff --git a/src/stream-tcp.c b/src/stream-tcp.c index 98431259f5..f5237a5686 100644 --- a/src/stream-tcp.c +++ b/src/stream-tcp.c @@ -47,6 +47,7 @@ void StreamTcpRegisterTests (void); void StreamTcpReturnStreamSegments (TcpStream *); void StreamTcpInitConfig(char); extern void StreamTcpSegmentReturntoPool(TcpSegment *); +int StreamTcpGetFlowState(void *); #define STREAMTCP_DEFAULT_SESSIONS 262144 #define STREAMTCP_DEFAULT_PREALLOC 32768 @@ -167,6 +168,7 @@ void StreamTcpInitConfig(char quiet) { FlowSetProtoTimeout(IPPROTO_TCP, STREAMTCP_NEW_TIMEOUT, STREAMTCP_EST_TIMEOUT); FlowSetProtoEmergencyTimeout(IPPROTO_TCP, STREAMTCP_EMERG_NEW_TIMEOUT, STREAMTCP_EMERG_EST_TIMEOUT); FlowSetProtoFreeFunc(IPPROTO_TCP, StreamTcpSessionPoolFree); + FlowSetProtoFlowStateFunc(IPPROTO_TCP, StreamTcpGetFlowState); } /** \brief The function is used to to fetch a TCP session from the @@ -424,7 +426,7 @@ static int StreamTcpPacketStateSynSent(ThreadVars *tv, Packet *p, StreamTcpThrea if(ValidReset(ssn, p)){ if(SEQ_EQ(TCP_GET_SEQ(p), ssn->client.isn) && SEQ_EQ(TCP_GET_WINDOW(p), 0) && SEQ_EQ(TCP_GET_ACK(p), (ssn->client.isn + 1))) { ssn->state = TCP_CLOSED; - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } } else return -1; @@ -507,7 +509,7 @@ static int StreamTcpPacketStateSynRecv(ThreadVars *tv, Packet *p, StreamTcpThrea case TH_RST|TH_ACK: if(ValidReset(ssn, p)) { ssn->state = TCP_CLOSED; - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } else return -1; break; @@ -672,7 +674,7 @@ static int StreamTcpPacketStateEstablished(ThreadVars *tv, Packet *p, StreamTcpT printf("StreamTcpPacketStateEstablished (%p): =+ next SEQ %" PRIu32 ", last ACK %" PRIu32 "\n", ssn, ssn->client.next_seq, ssn->server.last_ack); #endif - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } else { #ifdef DEBUG printf("StreamTcpPacketStateEstablished (%p): Reset received and state changed to TCP_CLOSED\n", ssn); @@ -694,7 +696,7 @@ static int StreamTcpPacketStateEstablished(ThreadVars *tv, Packet *p, StreamTcpT printf("StreamTcpPacketStateEstablished (%p): =+ next SEQ %" PRIu32 ", last ACK %" PRIu32 "\n", ssn, ssn->server.next_seq, ssn->client.last_ack); #endif - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } } else return -1; @@ -909,7 +911,7 @@ static int StreamTcpPacketStateFinWait1(ThreadVars *tv, Packet *p, StreamTcpThre printf("StreamTcpPacketStateFinWait1 (%p): Reset received state changed to TCP_CLOSED\n", ssn); #endif ssn->state = TCP_CLOSED; - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } else return -1; @@ -1002,7 +1004,7 @@ static int StreamTcpPacketStateFinWait2(ThreadVars *tv, Packet *p, StreamTcpThre printf("StreamTcpPacketStateFinWait2 (%p): Reset received state changed to TCP_CLOSED\n", ssn); #endif ssn->state = TCP_CLOSED; - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } else return -1; @@ -1254,7 +1256,7 @@ static int StreamTcpPakcetStateLastAck(ThreadVars *tv, Packet *p, StreamTcpThrea printf("StreamTcpPacketStateLastAck (%p): =+ next SEQ %" PRIu32 ", last ACK %" PRIu32 "\n", ssn, ssn->client.next_seq, ssn->server.last_ack); #endif - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } break; default: @@ -1306,7 +1308,7 @@ static int StreamTcpPacketStateTimeWait(ThreadVars *tv, Packet *p, StreamTcpThre printf("StreamTcpPacketStateTimeWait (%p): =+ next SEQ %" PRIu32 ", last ACK %" PRIu32 "\n", ssn, ssn->client.next_seq, ssn->server.last_ack); #endif - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } else { #ifdef DEBUG printf("StreamTcpPacketStateTimeWait (%p): pkt (%" PRIu32 ") is to client: SEQ %" PRIu32 ", ACK %" PRIu32 "\n", @@ -1333,7 +1335,7 @@ static int StreamTcpPacketStateTimeWait(ThreadVars *tv, Packet *p, StreamTcpThre printf("StreamTcpPacketStateTimeWait (%p): =+ next SEQ %" PRIu32 ", last ACK %" PRIu32 "\n", ssn, ssn->server.next_seq, ssn->client.last_ack); #endif - StreamTcpSessionPktFree(p); + //StreamTcpSessionPktFree(p); } break; default: @@ -1596,6 +1598,30 @@ static int ValidReset(TcpSession *ssn, Packet *p) { return 0; } +int StreamTcpGetFlowState(void *s) { + TcpSession *ssn = (TcpSession *)s; + + switch(ssn->state) { + case 0: + case TCP_SYN_SENT: + case TCP_SYN_RECV: + case TCP_LISTEN: + return FLOW_STATE_NEW; + case TCP_ESTABLISHED: + return FLOW_STATE_ESTABLISHED; + case TCP_FIN_WAIT1: + case TCP_FIN_WAIT2: + case TCP_CLOSING: + case TCP_LAST_ACK: + case TCP_TIME_WAIT: + case TCP_CLOSE_WAIT: + case TCP_CLOSED: + return FLOW_STATE_CLOSED; + } + return FLOW_STATE_CLOSED; +} + + #ifdef UNITTESTS /** @@ -1615,6 +1641,7 @@ static int StreamTcpTest01 (void) { f.stream = &ssn1; p.flow = &f; + StreamTcpInitConfig(TRUE); TcpSession *ssn = StreamTcpNewSession(&p); if (ssn == NULL) { printf("Session can not be allocated \n"); diff --git a/src/stream-tcp.h b/src/stream-tcp.h index 6b97195a71..aa99225c24 100644 --- a/src/stream-tcp.h +++ b/src/stream-tcp.h @@ -16,6 +16,7 @@ typedef struct TcpStreamCnf_ { TcpStreamCnf stream_config; void TmModuleStreamTcpRegister (void); void StreamTcpInitConfig (char); +void StreamTcpRegisterTests (void); #endif /* __STREAM_TCP_H__ */