tcpflow  1.6.1
About: tcpflow is a TCP/IP packet demultiplexer that captures data transmitted as part of TCP connections (flows), and stores the data in a way that is convenient for protocol analysis and debugging.
  Fossies Dox: tcpflow-1.6.1.tar.gz  ("unofficial" and yet experimental doxygen-generated source code documentation)  

tcpdemux.cpp
Go to the documentation of this file.
1 /**
2  *
3  * tcpdemux.cpp
4  * A tcpip demultiplier.
5  *
6  * This file is part of tcpflow by Simson Garfinkel,
7  * originally by Jeremy Elson <jelson@circlemud.org>
8  *
9  * This source code is under the GNU Public License (GPL). See
10  * LICENSE for details.
11  *
12  */
13 
14 #include "tcpflow.h"
15 #include "tcpip.h"
16 #include "tcpdemux.h"
17 
18 #include <iostream>
19 #include <sstream>
20 #include <vector>
21 
22 #ifdef HAVE_SYS_WAIT_H
23 #include <sys/wait.h>
24 #endif
25 
26 /* static */ uint32_t tcpdemux::max_saved_flows = 100;
27 /* static */ uint32_t tcpdemux::tcp_timeout = 0;
28 /* static */ int tcpdemux::tcp_subproc_max = 10;
29 /* static */ int tcpdemux::tcp_subproc = 0;
30 /* static */ int tcpdemux::tcp_alert_fd = -1;
31 /* static */ std::string tcpdemux::tcp_cmd = "";
32 
34 #ifdef HAVE_SQLITE3
35  db(),insert_flow(),
36 #endif
37  flow_sorter(0),tcp_processor(0),
38  outdir("."),flow_counter(0),packet_counter(0),
39  xreport(0),pwriter(0),max_open_flows(),max_fds(get_max_fds()-NUM_RESERVED_FDS),
40  unique_id(0),
41  flow_map(),open_flows(),saved_flow_map(),flow_fd_cache_map(0),
42  saved_flows(),start_new_connections(false),opt(),fs()
43 {
45 }
46 
48 {
49  DEBUG(1) ("ensuring pcap core");
51  flow_sorter = new pcap_writer();
52 }
53 
55 {
56 #ifdef HAVE_SQLITE3
57  int rc = sqlite3_open("test.db", &db);
58  if( rc ){
59  fprintf(stderr, "Can't open database: %s\n", sqlite3_errmsg(db));
60  db = 0;
61  }
62  /* Create SQL statement */
63  const char *sql = "CREATE TABLE connections ("
64  "starttime TEXT NOT NULL,"
65  "endtime TEXT NOT NULL,"
66  "src_ipn TEXT,"
67  "dst_ipn TEXT,"
68  "mac_daddr TEXT,"
69  "mac_saddr TEXT,"
70  "packets INTEGER,"
71  "srcport INTEGER,"
72  "dstport INTEGER,"
73  "hashdigest_md5 TEXT);";
74 
75  /* Execute SQL statement */
76  rc = sqlite3_exec(db, sql, callback, 0, &zErrMsg);
77  if( rc != SQLITE_OK ){
78  fprintf(stderr, "SQL error: %s\n", zErrMsg);
79  sqlite3_free(zErrMsg);
80  sqlite3_close(db);
81  return 0;
82  }
83  const char* zSql = "INSERT INTO connections (starttime,endtime,src_ipn,dst_ipn,mac_daddr,mac_saddr,packets,srcport,dstport,hashdigest_md5) VALUES (?,?,?,?,?,?,?,?,?,?)";
84  if(sqlite3_prepare_v2(db, zSql, strlen(zSql), &insert_stmt, NULL)!=SQLITE_OK ){
85  fprintf(stderr, "SQL prepare error");
86  db = 0;
87  insert_stmt=0;
88  return(0);
89  }
90 #endif
91 }
92 
93 void tcpdemux::write_flow_record(const std::string &starttime,const std::string &endtime,
94  const std::string &src_ipn,const std::string &dst_ipn,
95  const std::string &mac_daddr,const std::string &mac_saddr,
96  uint64_t packets,uint16_t srcport,uint16_t dstport,
97  const std::string &hashdigest_md5)
98 {
99 }
100 
101 
102 
104 {
105  static tcpdemux * theInstance = 0;
106  if(theInstance==0) theInstance = new tcpdemux();
107  return theInstance;
108 }
109 
110 /**
111  * find the flow that has been written to in the furthest past and close it.
112  */
114 {
115  tcpip *oldest_tcp = *open_flows.begin();
116  if(oldest_tcp) oldest_tcp->close_file();
117 }
118 
119 /* Open a file, closing one of the existing flows f necessary.
120  */
121 int tcpdemux::retrying_open(const std::string &filename,int oflag,int mask)
122 {
123  while(true){
124  //Packet index file reduces max_fds by 1/2 as the index files also take a fd
126  int fd = ::open(filename.c_str(),oflag,mask);
127  DEBUG(2)("retrying_open ::open(fn=%s,oflag=x%x,mask:x%x)=%d",filename.c_str(),oflag,mask,fd);
128  if(fd>=0){
129  /* Open was successful */
130  return fd;
131  }
132  DEBUG(2)("retrying_open ::open failed with errno=%d",errno);
133  if (errno != ENFILE && errno != EMFILE){
134  DEBUG(2)("retrying_open ::open failed with errno=%d (%s)",errno,strerror(errno));
135  return -1; // wonder what it was
136  }
137  DEBUG(5) ("too many open files -- contracting FD ring (size=%d)", (int)open_flows.size());
138  close_oldest_fd();
139  }
140 }
141 
142 /* Find previously a previously created flow state in the database.
143  */
145 {
146  flow_map_t::const_iterator it = flow_map.find(flow);
147  if (it==flow_map.end()){
148  return NULL; // flow not found
149  }
150  return it->second;
151 }
152 
153 /* Create a new flow state structure for a given flow.
154  * Puts the flow in the map.
155  * Returns a pointer to the new state.
156  *
157  * This is called by tcpdemux::process_tcp(). (Only place it is called)
158  *
159  * @param - pi - first packet seen on this connection.
160  *
161  * NOTE: We keep pointers to tcp structures in the map, rather than
162  * the structures themselves. This makes the map slightly more efficient,
163  * since it doesn't need to shuffle entire structures.
164  *
165  *
166  * TK: Note that the flow() is created on the stack and then used in new tcpip().
167  * This is resulting in an unnecessary copy.
168  */
169 
171 {
172  /* create space for the new state */
173  flow flow(flowa,flow_counter++,pi);
174 
175  tcpip *new_tcpip = new tcpip(*this,flow,isn);
176  new_tcpip->nsn = isn+1; // expected sequence number of the first byte
177  DEBUG(5) ("new flow %s. path: %s next seq num (nsn):%d",
178  flowa.str().c_str(),new_tcpip->flow_pathname.c_str(),new_tcpip->nsn);
179  flow_map[flow] = new_tcpip;
180  open_flows.reset(new_tcpip);
181  return new_tcpip;
182 }
183 
184 /**
185  * Remove a flow from the database.
186  * Close the flow file.
187  * Write to the report.xml object.
188  * Save in the sqlite database.
189  * This is the ONLY place where a tcpip object is deleted so there is no chance of finding it again.
190  *
191  * Flows are post-processed when a FIN is received and all bytes are received.
192  * If a FIN is received and bytes are outstanding, they are post-processed when the last byte is received.
193  * When the program shut down, all open flows are post-processed.
194  *
195  * Amended to trigger the packet/data location index sort as part of the post-processing. This sorts
196  * the (potentially out of order) index to make it simple for external applications. No processing is
197  * done if the (-I) index generation feature is turned off. --GDD
198  */
199 
201 {
202  std::stringstream xmladd; // for this <fileobject>
203  if(opt.post_processing && tcp->file_created && tcp->last_byte>0){
204  /**
205  * After the flow is finished, if more than a byte was
206  * written, then put it in an SBUF and process it. if we are
207  * doing post-processing. This is called from tcpip::~tcpip()
208  * in tcpip.cpp.
209  */
210 
211  /* Open the fd if it is not already open */
212  tcp->open_file();
213  if(tcp->fd>=0){
214  sbuf_t *sbuf = sbuf_t::map_file(tcp->flow_pathname,tcp->fd);
215  if(sbuf){
217  delete sbuf;
218  sbuf = 0;
219  }
220  }
221  }
222  tcp->close_file();
223  if(xreport) tcp->dump_xml(xreport,xmladd.str());
224  /**
225  * Before we delete the tcp structure, save information about the saved flow
226  */
227  save_flow(tcp);
228 
229  if(opt.store_output && tcp_alert_fd>=0){
230  std::stringstream ss;
231  ss << "close\t" << tcp->flow_pathname.c_str() << "\n";
232  const std::string &sso = ss.str();
233  if(write(tcp_alert_fd,sso.c_str(),sso.size()) != (int)sso.size()){
234  perror("write");
235  }
236  }
237 
238  if(tcp_cmd.size()>0 && tcp->flow_pathname.size()>0){
239  /* If we are at maximum number of subprocesses, wait for one to exit */
240  std::string cmd = tcp_cmd + " " + tcp->flow_pathname;
241 #ifdef HAVE_FORK
242  int status=0;
243  pid_t pid = 0;
244  while(tcp_subproc >= tcp_subproc_max){
245  pid = wait(&status);
246  tcp_subproc--;
247  }
248  /* Fork off a child */
249  pid = fork();
250  if(pid<0) die("Cannot fork child");
251  if(pid==0){
252  /* We are the child */
253  exit(system(cmd.c_str()));
254  }
255  tcp_subproc++;
256 #else
257  system(cmd.c_str());
258 #endif
259  }
260 
261  delete tcp;
262 }
263 
265 {
266  flow_map_t::iterator it = flow_map.find(flow);
267  if(it!=flow_map.end()){
268  post_process(it->second);
269  flow_map.erase(it);
270  }
271 }
272 
274 {
275 
276  DEBUG(10) ("Cleaning up flows");
277  for(flow_map_t::iterator it=flow_map.begin();it!=flow_map.end();it++){
278  post_process(it->second);
279  }
280  flow_map.clear();
281 
282  for(sparse_saved_flow_map_t::iterator it=flow_fd_cache_map.begin();it!=flow_fd_cache_map.end();it++){
283  delete it->second;
284  }
285  flow_fd_cache_map.clear();
286 }
287 
288 /****************************************************************
289  *** tcpdemultiplexer
290  ****************************************************************/
291 
292 /* Try to find the maximum number of FDs this system can have open */
293 unsigned int tcpdemux::get_max_fds(void)
294 {
295  int max_descs = 0;
296  const char *method=0;
297 
298  /* No longer users OPEN_MAX */
299 #if defined (HAVE_GETDTABLESIZE)
300  method = "getdtablesize";
301  max_descs = getdtablesize();
302 #elif defined(RLIMIT_NOFILE)
303  {
304  struct rlimit limit;
305  memset(&limit,0,sizeof(limit));
306 
307  method = "rlimit";
308  if (getrlimit(RLIMIT_NOFILE, &limit) < 0) {
309  perror("getrlimit");
310  exit(1);
311  }
312 
313  /* set the current to the maximum or specified value */
314  limit.rlim_cur = limit.rlim_max;
315 #ifdef OPEN_MAX
316  if(limit.rlim_cur > OPEN_MAX) limit.rlim_cur = OPEN_MAX;
317 #endif
318 
319  if (setrlimit(RLIMIT_NOFILE, &limit) < 0) {
320  perror("setrlimit");
321  exit(1);
322  }
323  max_descs = limit.rlim_max;
324 
325 #ifdef RLIM_INFINITY
326  if (limit.rlim_max == RLIM_INFINITY) max_descs = MAX_FD_GUESS * 4; /* pick a more reasonable max */
327 #endif
328  }
329 #elif defined (_SC_OPEN_MAX)
330  /* Okay, you don't have getrlimit() and you don't have OPEN_MAX.
331  * Time to try the POSIX sysconf() function. (See Stevens'
332  * _Advanced Programming in the UNIX Environment_). */
333  method = "POSIX sysconf";
334  errno = 0;
335  if ((max_descs = sysconf(_SC_OPEN_MAX)) < 0) {
336  if (errno == 0)
337  max_descs = MAX_FD_GUESS * 4;
338  else {
339  perror("calling sysconf");
340  exit(1);
341  }
342  }
343 
344  /* if everything has failed, we'll just take a guess */
345 #else
346  method = "MAX_FD_GUESS";
347  max_descs = MAX_FD_GUESS;
348 #endif
349  /* this must go here, after rlimit code */
350  DEBUG(10) ("found max FDs to be %d using %s", max_descs, method);
351  return max_descs;
352 }
353 
354 
355 /*
356  * open the packet save flow
357  */
358 void tcpdemux::save_unk_packets(const std::string &ofname,const std::string &ifname)
359 {
360  pwriter = pcap_writer::open_copy(ofname,ifname);
361 }
362 
363 /**
364  * save information on this flow needed to handle strangling packets
365  */
366 int c = 0;
368 {
369  /* First remove the oldest flow if we are in overload */
370  if(saved_flows.size()>0 && saved_flows.size()>max_saved_flows){
371  saved_flow *flow0 = saved_flows.at(0);
372  saved_flow_map.erase(flow0->addr); // remove from the map
373  saved_flows.erase(saved_flows.begin()); // remove from the vector
374  delete flow0; // and delete the saved flow
375  }
376 
377  /* Now save the flow */
378  saved_flow *sf = new saved_flow(tcp);
379  saved_flow_map[sf->addr] = sf;
380  saved_flows.push_back(sf);
381 }
382 
383 /**
384  * dissect_tcp():
385  *
386  * Called to process tcp pkts in a way that dissected or isolated pcap flows are
387  * emerging afterwards. Similar notions go into the direction of "sorting" pcap pkts
388  * as per flow context.
389  *
390  * Returns 0 if packet is processed, 1 if it is not processed, -1 if error
391  */
392 
393 int tcpdemux::dissect_tcp(const ipaddr &src, const ipaddr &dst,sa_family_t family,
394  const u_char *ip_data, uint32_t ip_payload_len,
395  const be13::packet_info &pi)
396 {
397  DEBUG(6) ("dissecting pkt *********************");
398  struct be13::tcphdr *tcp_header = (struct be13::tcphdr *) ip_data;
399  flow_addr this_flow(src,dst,ntohs(tcp_header->th_sport),
400  ntohs(tcp_header->th_dport),family);
401 
402  sparse_saved_flow_map_t::const_iterator it = flow_fd_cache_map.find(this_flow);
403  if(it!=flow_fd_cache_map.end()){
404  flow_sorter->update_sink(it->second->fcap);
405  }
406  else {
407  flow fn_gen_vehicle(this_flow, 0, pi); // impromptu flow name generator
408  std::string fn = fn_gen_vehicle.new_pcap_filename();
410  FILE *sink= flow_sorter->yield_sink();
411  sparse_saved_flow *ssf = new sparse_saved_flow(this_flow, sink);
412  flow_fd_cache_map[ssf->addr] = ssf;
413  }
414 
416 
417  return 0;
418 }
419 
420 /**
421  * process_tcp():
422  *
423  * Called to processes a tcp packet from either process_ip4() or process_ip6().
424  * The caller breaks out the ip addresses and finds the start of the tcp header.
425  *
426  * Skips but otherwise ignores TCP options.
427  *
428  * creates a new tcp connection if necessary, then asks the connection to either
429  * print the packet or store it.
430  *
431  * Returns 0 if packet is processed, 1 if it is not processed, -1 if error
432  */
433 
434 #define FLAG_SET(vector, flag) ((vector) & (flag))
435 
436 #pragma GCC diagnostic ignored "-Wcast-align"
437 #include "iptree.h"
438 
439 
440 
441 int tcpdemux::process_tcp(const ipaddr &src, const ipaddr &dst,sa_family_t family,
442  const u_char *ip_data, uint32_t ip_payload_len,
443  const be13::packet_info &pi)
444 {
445  if (ip_payload_len < sizeof(struct be13::tcphdr)) {
446  DEBUG(6) ("received truncated TCP segment! (%u<%u)",
447  (u_int)ip_payload_len,(u_int)sizeof(struct be13::tcphdr));
448  return 1;
449  }
450 
451  struct be13::tcphdr *tcp_header = (struct be13::tcphdr *) ip_data;
452 
453  /* fill in the flow_addr structure with info that identifies this flow */
454  flow_addr this_flow(src,dst,ntohs(tcp_header->th_sport),ntohs(tcp_header->th_dport),family);
455 
456  be13::tcp_seq seq = ntohl(tcp_header->th_seq);
457  bool syn_set = FLAG_SET(tcp_header->th_flags, TH_SYN);
458  bool ack_set = FLAG_SET(tcp_header->th_flags, TH_ACK);
459  bool fin_set = FLAG_SET(tcp_header->th_flags, TH_FIN);
460  bool rst_set = FLAG_SET(tcp_header->th_flags, TH_RST);
461 
462  /* calculate the total length of the TCP header including options */
463  u_int tcp_header_len = tcp_header->th_off * 4;
464 
465  /* Find the beginning of the tcp data.
466  */
467  const u_char *tcp_data = ip_data + tcp_header_len;
468 
469  /* figure out how much tcp data we have, taking into account tcp options */
470 
471  size_t tcp_datalen = (ip_payload_len > tcp_header_len) ? (ip_payload_len - tcp_header_len) : 0;
472 
473  /* see if we have state about this flow; if not, create it */
474  int32_t delta = 0; // from current position in tcp connection; must be SIGNED 32 bit!
475  tcpip *tcp = find_tcpip(this_flow);
476 
477  DEBUG(60)("%s%s%s%s tcp_header_len=%d tcp_datalen=%d seq=%u tcp=%p",
478  (syn_set?"SYN ":""),(ack_set?"ACK ":""),(fin_set?"FIN ":""),(rst_set?"RST ":""),(int)tcp_header_len,(int)tcp_datalen,(int)seq,tcp);
479 
480  /* If this_flow is not in the database and the start_new_connections flag is false, just return */
481  if(tcp==0 && start_new_connections==false) return 0;
482 
483  if(syn_set && tcp && tcp->syn_count>0 && tcp->pos>0){
484  std::cerr << "SYN TO IGNORE! SYN tcp="<<tcp << " flow="<<this_flow<<"\n";
485  return 1;
486  }
487 
488  if(tcp==0){
489  if(tcp_datalen==0){ // zero length packet
490  if(fin_set) return 0; // FIN on a connection that's unknown; safe to ignore
491  if(rst_set) return 0; // RST on a connection that's unknown; safe to ignore
492  if(syn_set==false && ack_set==false) return 0; // neither a SYN nor ACK; return
493  } else {
494  /* Data present on a flow that is not actively being demultiplexed.
495  * See if it is a saved flow. If so, see if the data in the packet
496  * matches what is on the disk. If so, return.
497  *
498  */
499  saved_flow_map_t::const_iterator it = saved_flow_map.find(this_flow);
500  if(it!=saved_flow_map.end()){
501  uint32_t offset = seq - it->second->isn - 1;
502  bool data_match = false;
503  int fd = open(it->second->saved_filename.c_str(),O_RDONLY | O_BINARY);
504  if(fd>0){
505  char *buf = (char *)malloc(tcp_datalen);
506  if(buf){
507  DEBUG(100)("lseek(fd,%" PRId64 ",SEEK_SET)",(int64_t)(offset));
508  lseek(fd,offset,SEEK_SET);
509  ssize_t r = read(fd,buf,tcp_datalen);
510  data_match = (r==(ssize_t)tcp_datalen) && memcmp(buf,tcp_data,tcp_datalen)==0;
511  free(buf);
512  }
513  close(fd);
514  }
515  DEBUG(60)("Packet matches saved flow. offset=%u len=%d filename=%s data match=%d\n",
516  (u_int)offset,(u_int)tcp_datalen,it->second->saved_filename.c_str(),(u_int)data_match);
517  if(data_match) return 0;
518  }
519  }
520  }
521 
522  /* flow is in the database; make sure the gap isn't too big.*/
523  if(tcp){
524  /* Compute delta based on next expected sequence number.
525  * If delta will be too much, start a new flow.
526  *
527  * NOTE: I hope we don't get a packet from the old flow when
528  * we are processing the new one. Perhaps we should be able to have
529  * multiple flows at the same time with the same quad, and they are
530  * at different window areas...
531  *
532  */
533  delta = seq - tcp->nsn; // notice that signed offset is calculated
534 
535  if(abs(delta) > opt.max_seek){
536  remove_flow(this_flow);
537  delta = 0;
538  tcp = 0;
539  }
540  }
541 
542  /* At this point, tcp may be NULL because:
543  * case 1 - It's a new connection and SYN IS SET; normal case
544  * case 2 - Extra packets on a now-closed connection
545  * case 3 - Packets for which the initial part of the connection was missed
546  * case 4 - It's a connecton that had a huge gap and was expired out of the databsae
547  *
548  * THIS IS THE ONLY PLACE THAT create_tcpip() is called.
549  */
550 
551  /* q: what if syn is set AND there is data? */
552  /* q: what if syn is set AND we already know about this connection? */
553 
554  if (tcp==NULL){
555 
556  /* Don't process if this is not a SYN and there is no data. */
557  if(syn_set==false && tcp_datalen==0) return 0;
558 
559  /* Check if this is the server->client flow related to a client->server flow that is being demultiplexed */
560  flow_addr reverse_flow(dst,src,ntohs(tcp_header->th_dport),ntohs(tcp_header->th_sport),family);
561  tcpip *reverse_tcp = find_tcpip(reverse_flow);
562  uint64_t uid;
563  if (reverse_tcp)
564  {
565  /* We found a matching client->server flow. Copy its session ID */
566  uid = reverse_tcp->myflow.session_id;
567  }
568  else
569  {
570  /* Assign a new unique ID */
571  uid = unique_id++;
572  }
573 
574  /* Create a new connection.
575  * delta will be 0, because it's a new connection!
576  */
577  be13::tcp_seq isn = syn_set ? seq : seq-1;
578  tcp = create_tcpip(this_flow, isn, pi);
579  tcp->myflow.session_id = uid;
580  }
581 
582  /* Now tcp is valid */
583  tcp->myflow.tlast = pi.ts; // most recently seen packet
585  tcp->myflow.len += pi.pcap_hdr->len;
586  tcp->myflow.caplen += pi.pcap_hdr->caplen;
587  tcp->myflow.packet_count++;
588 
589  // Does not seem consitent => Print a notice
590  // See also https://stackoverflow.com/q/1491660
591  if(pi.pcap_hdr->caplen != pi.pcap_hdr->len){
592  DEBUG(2)("Captured packet has a length caplen=%d different "
593  "from the un-truncated length len=%d provided by PCAP API",
594  pi.pcap_hdr->caplen, pi.pcap_hdr->len);
595  }
596 
597  /*
598  * 2012-10-24 slg - the first byte is sent at SEQ==ISN+1.
599  * The first byte in POSIX files have an LSEEK of 0.
600  * The original code overcame this issue by introducing an intentional off-by-one
601  * error with the statement tcp->isn++.
602  *
603  * With the new TCP state-machine we simply follow the spec.
604  *
605  * The new state machine works by examining the SYN and ACK packets
606  * in accordance with the TCP spec.
607  */
608  if(syn_set){
609  /* If the syn is set this is either a SYN or SYN-ACK. We use this information to set the direction
610  * flag, but that's it. The direction flag is only used for coloring.
611  */
612  if(tcp->syn_count>1){
613  DEBUG(2)("Multiple SYNs (%d) seen on connection %s",tcp->syn_count,tcp->flow_pathname.c_str());
614  }
615  tcp->syn_count++;
616  if( !ack_set ){
617  DEBUG(50) ("packet is handshake SYN"); /* First packet of three-way handshake */
618  tcp->dir = tcpip::dir_cs; // client->server
619  } else {
620  DEBUG(50) ("packet is handshake SYN/ACK"); /* second packet of three-way handshake */
621  tcp->dir = tcpip::dir_sc; // server->client
622  }
623  if(tcp_datalen>0){
624  tcp->violations++;
625  DEBUG(1) ("TCP PROTOCOL VIOLATION: SYN with data! (length=%d)",(int)tcp_datalen);
626  }
627  }
628  if(tcp_datalen==0) DEBUG(50) ("got TCP segment with no data"); // seems pointless to notify
629 
630  /* process any data.
631  * Notice that this typically won't be called for the SYN or SYN/ACK,
632  * since they both have no data by definition.
633  */
634  if (tcp_datalen>0){
635  if (opt.console_output) {
636  tcp->print_packet(tcp_data, tcp_datalen);
637  } else {
638  if (opt.store_output){
639  bool new_file = false;
640  if (tcp->fd < 0) new_file = true;
641 
642  tcp->store_packet(tcp_data, tcp_datalen, delta,pi.ts);
643 
644  if(new_file && tcp_alert_fd>=0){
645  std::stringstream ss;
646  ss << "open\t" << tcp->flow_pathname.c_str() << "\n";
647  const std::string &sso = ss.str();
648  if(write(tcp_alert_fd,sso.c_str(),sso.size())!=(int)sso.size()){
649  perror("write");
650  }
651  }
652  }
653  }
654  }
655 
656  if (rst_set){
657  remove_flow(this_flow); // take it out of the map
658  return 0;
659  }
660 
661  /* Count the FINs.
662  * If this is a fin, determine the size of the stream
663  */
664  if (fin_set){
665  tcp->fin_count++;
666  if(tcp->fin_count==1){
667  tcp->fin_size = (seq+tcp_datalen-tcp->isn)-1;
668  }
669  } else {
670  open_flows.move_to_end(tcp);
671  }
672 
673  /* If a fin was sent and we've seen all of the bytes, close the stream */
674  DEBUG(50)("%d>0 && %d == %d",tcp->fin_count,tcp->seen_bytes(),tcp->fin_size);
675 
676  if (tcp->fin_count>0 && tcp->seen_bytes() == tcp->fin_size){
677  DEBUG(50)("all bytes have been received; removing flow");
678  remove_flow(this_flow); // take it out of the map
679  }
680 
681  DEBUG(50)("fin_set=%d seq=%u fin_count=%d seq_count=%d len=%d isn=%u",
682  fin_set,seq,tcp->fin_count,tcp->syn_count,(int)tcp_datalen,tcp->isn);
683  return 0; // successfully processed
684 }
685 #pragma GCC diagnostic warning "-Wcast-align"
686 
687 
688 /* This is called when we receive an IPv4 datagram. We make sure that
689  * it's valid and contains a TCP segment; if so, we pass it to
690  * process_tcp() for further processing.
691  *
692  * Note: we currently don't know how to handle IP fragments. */
693 #pragma GCC diagnostic ignored "-Wcast-align"
694 
695 
696 
698 {
699  /* make sure that the packet is at least as long as the min IP header */
700  if (pi.ip_datalen < sizeof(struct be13::ip4)) {
701  DEBUG(6) ("received truncated IP datagram!");
702  return -1; // couldn't process
703  }
704 
705  const struct be13::ip4 *ip_header = (struct be13::ip4 *) pi.ip_data;
706 
707  DEBUG(100)("process_ip4. caplen=%d vlan=%d ip_p=%d",(int)pi.pcap_hdr->caplen,(int)pi.vlan(),(int)ip_header->ip_p);
708  if(debug>200){
709  sbuf_t sbuf(pos0_t(),(const uint8_t *)pi.ip_data,pi.ip_datalen,pi.ip_datalen,0,false, false,false);
710  sbuf.hex_dump(std::cerr);
711  }
712 
713  /* for now we're only looking for TCP; throw away everything else */
714  if (ip_header->ip_p != IPPROTO_TCP) {
715  DEBUG(50) ("got non-TCP frame -- IP proto %d", ip_header->ip_p);
716  return -1; // couldn't process
717  }
718 
719  /* check and see if we got everything. NOTE: we must use
720  * ip_total_len after this, because we may have captured bytes
721  * beyond the end of the packet (e.g. ethernet padding).
722  */
723  size_t ip_len = ntohs(ip_header->ip_len);
724  if (pi.ip_datalen < ip_len) {
725  DEBUG(6) ("warning: captured only %ld bytes of %ld-byte IP datagram",
726  (long) pi.ip_datalen, (long) ip_len);
727  }
728 
729  /* XXX - throw away everything but fragment 0; this version doesn't
730  * know how to do fragment reassembly.
731  */
732  if (ntohs(ip_header->ip_off) & 0x1fff) {
733  DEBUG(2) ("warning: throwing away IP fragment from X to X");
734  return -1;
735  }
736 
737  /* figure out where the IP header ends */
738  size_t ip_header_len = ip_header->ip_hl * 4;
739 
740  /* make sure there's some data */
741  if (ip_header_len > ip_len) {
742  DEBUG(6) ("received truncated IP datagram!");
743  return -1;
744  }
745 
746  /* do TCP processing, faking an ipv6 address */
747  uint16_t ip_payload_len = pi.ip_datalen - ip_header_len;
748  ipaddr src(ip_header->ip_src.addr);
749  ipaddr dst(ip_header->ip_dst.addr);
750  return (this->*tcp_processor)(src, dst ,AF_INET,
751  pi.ip_data + ip_header_len,
752  ip_payload_len,pi);
753 }
754 #pragma GCC diagnostic warning "-Wcast-align"
755 
756 
757 /* This is called when we receive an IPv6 datagram.
758  *
759  * Note: we don't support IPv6 extended headers
760  */
761 
762 /* These might be defined from an include file, so undef them to be sure */
763 
765 {
766  /* make sure that the packet is at least as long as the IPv6 header */
767  if (pi.ip_datalen < sizeof(struct be13::ip6_hdr)) {
768  DEBUG(6) ("received truncated IPv6 datagram!");
769  return -1;
770  }
771 
772  const struct be13::ip6_hdr *ip_header = (struct be13::ip6_hdr *) pi.ip_data;
773 
774  /* for now we're only looking for TCP; throw away everything else */
775  if (ip_header->ip6_ctlun.ip6_un1.ip6_un1_nxt != IPPROTO_TCP) {
776  DEBUG(50) ("got non-TCP frame -- IP proto %d", ip_header->ip6_ctlun.ip6_un1.ip6_un1_nxt);
777  return -1;
778  }
779 
780  /* do TCP processing */
781  uint16_t ip_payload_len = ntohs(ip_header->ip6_ctlun.ip6_un1.ip6_un1_plen);
782  ipaddr src(ip_header->ip6_src.addr.addr8);
783  ipaddr dst(ip_header->ip6_dst.addr.addr8);
784 
785  return (this->*tcp_processor)(src, dst ,AF_INET6,
786  pi.ip_data + sizeof(struct be13::ip6_hdr),
787  ip_payload_len,pi);
788 }
789 
790 /* This is called when we receive an IPv4 or IPv6 datagram.
791  * This function calls process_ip4 or process_ip6
792  * Returns 0 if packet is processed, 1 if it is not processed, -1 if error.
793  */
794 
795 #pragma GCC diagnostic ignored "-Wcast-align"
797 {
798  DEBUG(10)("process_pkt..............................................................................");
799  int r = 1; // not processed yet
800  switch(pi.ip_version()){
801  case 4:
802  r = process_ip4(pi);
803  break;
804  case 6:
805  r = process_ip6(pi);
806  break;
807  }
808  if(r!=0){ // packet not processed?
809  /* Write the packet if we didn't process it */
811  }
812 
813  /* Process the timeout, if there is any */
814  if(tcp_timeout){
815  /* Get a list of the flows that need to be closed. */
816  std::vector<flow_addr *> to_close;
817  for(flow_map_t::iterator it = flow_map.begin(); it!=flow_map.end(); it++){
818  tcpip &tcp = *(it->second);
819  uint32_t age = pi.ts.tv_sec - tcp.myflow.tlast.tv_sec;
820  if (age > tcp_timeout){
821  to_close.push_back(&tcp.myflow);
822  }
823  }
824  /* Close them. This removes the flows from the flow_map(), which is why we need
825  * to create the list first.
826  */
827  for(std::vector<flow_addr *>::iterator it = to_close.begin(); it!=to_close.end(); it++){
828  remove_flow(*(*it));
829  }
830  }
831  return r;
832 }
833 #pragma GCC diagnostic warning "-Wcast-align"
std::string str() const
Definition: tcpip.h:138
Definition: tcpip.h:156
uint64_t len
Definition: tcpip.h:183
uint64_t packet_count
Definition: tcpip.h:185
std::string new_pcap_filename()
Definition: flow.cpp:183
uint64_t session_id
Definition: tcpip.h:186
uint64_t caplen
Definition: tcpip.h:184
struct timeval tlast
Definition: tcpip.h:182
void move_to_end(T *node)
iterator begin()
void reset(T *node)
Definition: tcpip.h:25
FILE * yield_sink()
Definition: pcap_writer.h:97
void update_sink(FILE *sink)
Definition: pcap_writer.h:94
static pcap_writer * open_copy(const std::string &ofname, const std::string &ifname)
Definition: pcap_writer.h:72
void writepkt(const struct pcap_pkthdr *h, const u_char *p)
Definition: pcap_writer.h:81
void refresh_sink(const std::string &fname, const int pcap_dlt)
Definition: pcap_writer.h:90
Definition: sbuf.h:70
flow_addr addr
Definition: tcpip.h:365
Definition: sbuf.h:221
flow_addr addr
Definition: tcpip.h:375
bool post_processing
Definition: tcpdemux.h:112
bool output_packet_index
Definition: tcpdemux.h:122
int32_t max_seek
Definition: tcpdemux.h:124
void remove_flow(const flow_addr &flow)
Definition: tcpdemux.cpp:264
int retrying_open(const std::string &filename, int oflag, int mask)
Definition: tcpdemux.cpp:121
int dissect_tcp(const ipaddr &src, const ipaddr &dst, sa_family_t family, const u_char *tcp_data, uint32_t tcp_length, const be13::packet_info &pi)
Definition: tcpdemux.cpp:393
static int tcp_alert_fd
Definition: tcpdemux.h:87
void save_flow(tcpip *)
Definition: tcpdemux.cpp:367
intrusive_list< tcpip > open_flows
Definition: tcpdemux.h:139
int(tcpdemux::* tcp_processor)(const ipaddr &src, const ipaddr &dst, sa_family_t family, const u_char *tcp_data, uint32_t tcp_length, const be13::packet_info &pi)
Definition: tcpdemux.h:78
int process_ip4(const be13::packet_info &pi)
Definition: tcpdemux.cpp:697
uint64_t packet_counter
Definition: tcpdemux.h:131
options opt
Definition: tcpdemux.h:146
static uint32_t tcp_timeout
Definition: tcpdemux.h:83
saved_flow_map_t saved_flow_map
Definition: tcpdemux.h:141
bool start_new_connections
Definition: tcpdemux.h:144
uint64_t unique_id
Definition: tcpdemux.h:136
static int tcp_subproc_max
Definition: tcpdemux.h:85
int process_tcp(const ipaddr &src, const ipaddr &dst, sa_family_t family, const u_char *tcp_data, uint32_t tcp_length, const be13::packet_info &pi)
Definition: tcpdemux.cpp:441
dfxml_writer * xreport
Definition: tcpdemux.h:132
int process_pkt(const be13::packet_info &pi)
Definition: tcpdemux.cpp:796
pcap_writer * flow_sorter
Definition: tcpdemux.h:75
tcpip * find_tcpip(const flow_addr &flow)
Definition: tcpdemux.cpp:144
void remove_all_flows()
Definition: tcpdemux.cpp:273
void alter_processing_core()
Definition: tcpdemux.cpp:47
tcpdemux()
Definition: tcpdemux.cpp:33
void post_process(tcpip *tcp)
Definition: tcpdemux.cpp:200
class feature_recorder_set * fs
Definition: tcpdemux.h:147
static int tcp_subproc
Definition: tcpdemux.h:86
int process_ip6(const be13::packet_info &pi)
Definition: tcpdemux.cpp:764
void save_unk_packets(const std::string &wfname, const std::string &ifname)
Definition: tcpdemux.cpp:358
tcpip * create_tcpip(const flow_addr &flow, be13::tcp_seq isn, const be13::packet_info &pi)
Definition: tcpdemux.cpp:170
unsigned int max_fds
Definition: tcpdemux.h:135
void close_oldest_fd()
Definition: tcpdemux.cpp:113
static unsigned int get_max_fds(void)
Definition: tcpdemux.cpp:293
static std::string tcp_cmd
Definition: tcpdemux.h:84
void write_flow_record(const std::string &starttime, const std::string &endtime, const std::string &src_ipn, const std::string &dst_ipn, const std::string &mac_daddr, const std::string &mac_saddr, uint64_t packets, uint16_t srcport, uint16_t dstport, const std::string &hashdigest_md5)
Definition: tcpdemux.cpp:93
pcap_writer * pwriter
Definition: tcpdemux.h:133
sparse_saved_flow_map_t flow_fd_cache_map
Definition: tcpdemux.h:142
uint64_t flow_counter
Definition: tcpdemux.h:130
flow_map_t flow_map
Definition: tcpdemux.h:138
static uint32_t max_saved_flows
Definition: tcpdemux.h:149
static tcpdemux * getInstance()
Definition: tcpdemux.cpp:103
saved_flows_t saved_flows
Definition: tcpdemux.h:143
void openDB()
Definition: tcpdemux.cpp:54
Definition: tcpip.h:274
uint64_t last_packet_number
Definition: tcpip.h:319
uint32_t seen_bytes()
Definition: tcpip.cpp:47
uint64_t last_byte
Definition: tcpip.h:318
uint64_t violations
Definition: tcpip.h:321
uint32_t fin_size
Definition: tcpip.h:304
dir_t dir
Definition: tcpip.h:299
flow myflow
Definition: tcpip.h:298
int fd
Definition: tcpip.h:309
void dump_xml(class dfxml_writer *xmlreport, const std::string &xmladd)
Definition: tcpip.cpp:63
int open_file()
Definition: tcpip.cpp:177
uint64_t pos
Definition: tcpip.h:305
void store_packet(const u_char *data, uint32_t length, int32_t delta, struct timeval ts)
Definition: tcpip.cpp:475
uint32_t syn_count
Definition: tcpip.h:302
std::string flow_pathname
Definition: tcpip.h:308
bool file_created
Definition: tcpip.h:310
@ dir_sc
Definition: tcpip.h:279
@ dir_cs
Definition: tcpip.h:280
void close_file()
Definition: tcpip.cpp:128
be13::tcp_seq isn
Definition: tcpip.h:300
be13::tcp_seq nsn
Definition: tcpip.h:301
void print_packet(const u_char *data, uint32_t length)
Definition: tcpip.cpp:243
uint32_t fin_count
Definition: tcpip.h:303
#define O_BINARY
static int debug
void hex_dump(std::ostream &os, uint64_t start, uint64_t len) const
Definition: sbuf.cpp:121
static sbuf_t * map_file(const std::string &fname)
Definition: sbuf.cpp:22
#define TH_FIN
struct ip6_addr ip6_dst
#define TH_ACK
struct ip6_addr ip6_src
const u_char * pcap_data
const struct pcap_pkthdr * pcap_hdr
uint8_t addr8[16]
const size_t ip_datalen
union be13::ip6_addr::@0 addr
#define TH_RST
uint16_t ip_off
union be13::ip6_hdr::@1 ip6_ctlun
#define TH_SYN
static void process_sbuf(const class scanner_params &sp)
Definition: plugin.cpp:577
#define IPPROTO_TCP
const uint8_t *const ip_data
struct be13::ip6_hdr::@1::ip6_hdrctl ip6_un1
const struct timeval & ts
uint16_t ip_len
int ip_version() const
uint32_t tcp_seq
struct ip4_addr ip_src ip_dst
unsigned int uint32_t
Definition: core.h:40
uint32_t caplen
Definition: pcap_fake.h:37
uint32_t len
Definition: pcap_fake.h:38
int c
Definition: tcpdemux.cpp:366
#define FLAG_SET(vector, flag)
Definition: tcpdemux.cpp:434
dfxml_writer * xreport
Definition: tcpflow.cpp:242
#define SEEK_SET
Definition: tcpflow.h:202
void die(const char *fmt,...)
Definition: util.cpp:175
#define DEBUG(message_level)
Definition: tcpflow.h:273
#define MAX_FD_GUESS
Definition: tcpflow.h:244
#define NUM_RESERVED_FDS
Definition: tcpflow.h:251
unsigned short int sa_family_t
Definition: tcpip.h:16
unsigned short uint16_t
Definition: util.h:7
unsigned char uint8_t
Definition: util.h:6