Implement new TCP stream follower mechanism

.travis.yml

@@ -15,6 +15,7 @@ addons:
     packages:
     - libpcap-dev
     - libssl-dev
+    - libboost-all-dev
 
 before_script:
   - mkdir build

CMakeLists.txt

@@ -56,7 +56,6 @@ IF(WIN32)
     ADD_DEFINITIONS(-DNOMINMAX)
 ENDIF(WIN32)
 
-
 INCLUDE(ExternalProject)
 
 # *******************
@@ -97,6 +96,22 @@ IF(LIBTINS_ENABLE_DOT11)
     ENDIF(LIBTINS_ENABLE_WPA2)
 ENDIF(LIBTINS_ENABLE_DOT11)
 
+OPTION(LIBTINS_ENABLE_ACK_TRACKER "Enable TCP ACK tracking support" ON)
+IF(LIBTINS_ENABLE_ACK_TRACKER AND HAVE_CXX11)
+    FIND_PACKAGE(Boost)
+    IF (Boost_FOUND)
+        MESSAGE(STATUS "Enabling TCP ACK tracking support.")
+        INCLUDE_DIRECTORIES(Boost_INCLUDE_DIRS)
+        SET(HAVE_ACK_TRACKER ON)
+    ELSE()
+        MESSAGE(WARNING "Disabling ACK tracking support as boost.icl was not found")
+        SET(HAVE_ACK_TRACKER OFF)
+    ENDIF()
+ELSE()
+    SET(HAVE_ACK_TRACKER OFF)
+    MESSAGE(STATUS "Disabling ACK tracking support")
+ENDIF()
+
 # Use pcap_sendpacket to send l2 packets rather than raw sockets
 IF(WIN32)
     SET(USE_PCAP_SENDPACKET_DEFAULT ON)

cmake/appveyor.yml

@@ -37,7 +37,7 @@ after_build:
 - 7z a libtins-%platform%-%Configuration%.zip libtins
 test_script:
 - cd c:\projects\libtins\build
-- ctest -C %Configuration%
+- ctest -C %Configuration% -V
 deploy_script:
 - ps: Push-AppveyorArtifact "install\libtins-$env:Platform-$env:Configuration.zip"
 skip_commits:

examples/CMakeLists.txt

@@ -1,5 +1,6 @@
 FIND_PACKAGE(libtins QUIET)
 FIND_PACKAGE(Threads QUIET)
+FIND_PACKAGE(Boost COMPONENTS regex)
 
 IF(libtins_FOUND)
     SET(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/examples)
@@ -15,12 +16,18 @@ IF(libtins_FOUND)
             dns_queries
             dns_spoof
             dns_stats
+            stream_dump
             icmp_responses
             interfaces_info
             tcp_connection_close
             traceroute
             wps_detect
         )
+        IF(Boost_REGEX_FOUND)
+            SET(LIBTINS_CXX11_EXAMPLES ${LIBTINS_CXX11_EXAMPLES} http_requests)
+        ELSE()
+            MESSAGE(WARNING "Disabling HTTP requests example since boost.regex was not found")
+        ENDIF()
     ELSE(HAVE_CXX11)
         MESSAGE(WARNING "Disabling some examples since C++11 support is disabled.")
     ENDIF(HAVE_CXX11)
@@ -40,10 +47,15 @@ IF(libtins_FOUND)
         ADD_EXECUTABLE(arpmonitor EXCLUDE_FROM_ALL arpmonitor.cpp)
         ADD_EXECUTABLE(dns_queries EXCLUDE_FROM_ALL dns_queries.cpp)
         ADD_EXECUTABLE(dns_spoof EXCLUDE_FROM_ALL dns_spoof.cpp)
+        ADD_EXECUTABLE(stream_dump EXCLUDE_FROM_ALL stream_dump.cpp)
         ADD_EXECUTABLE(icmp_responses EXCLUDE_FROM_ALL icmp_responses.cpp)
         ADD_EXECUTABLE(interfaces_info EXCLUDE_FROM_ALL interfaces_info.cpp)
         ADD_EXECUTABLE(tcp_connection_close EXCLUDE_FROM_ALL tcp_connection_close.cpp)
         ADD_EXECUTABLE(wps_detect EXCLUDE_FROM_ALL wps_detect.cpp)
+        IF (Boost_REGEX_FOUND)
+            ADD_EXECUTABLE(http_requests EXCLUDE_FROM_ALL http_requests.cpp)
+            TARGET_LINK_LIBRARIES(http_requests ${Boost_LIBRARIES})
+        ENDIF()
     ENDIF(HAVE_CXX11)
 
     ADD_EXECUTABLE(beacon_display EXCLUDE_FROM_ALL beacon_display.cpp)

examples/http_requests.cpp

@@ -0,0 +1,145 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <string>
+#include <iostream>
+#include <stdexcept>
+#include <boost/regex.hpp>
+#include "tins/tcp_ip/stream_follower.h"
+#include "tins/sniffer.h"
+
+using std::string;
+using std::cout;
+using std::cerr;
+using std::endl;
+using std::exception;
+
+using boost::regex;
+using boost::match_results;
+
+using Tins::Packet;
+using Tins::Sniffer;
+using Tins::SnifferConfiguration;
+using Tins::TCPIP::Stream;
+using Tins::TCPIP::StreamFollower;
+
+// This example captures and follows TCP streams seen on port 80. It will
+// wait until both the client and server send data and then apply a regex
+// to both payloads, extrating some information and printing it.
+
+// Don't buffer more than 3kb of data in either request/response
+const size_t MAX_PAYLOAD = 3 * 1024;
+// The regex to be applied on the request. This will extract the HTTP
+// method being used, the request's path and the Host header value.
+regex request_regex("([\\w]+) ([^ ]+).+\r\nHost: ([\\d\\w\\.-]+)\r\n");
+// The regex to be applied on the response. This finds the response code.
+regex response_regex("HTTP/[^ ]+ ([\\d]+)");
+
+void on_server_data(Stream& stream) {
+    match_results<Stream::payload_type::const_iterator> client_match;
+    match_results<Stream::payload_type::const_iterator> server_match;
+    const Stream::payload_type& client_payload = stream.client_payload();
+    const Stream::payload_type& server_payload = stream.server_payload();
+    // Run the regexes on client/server payloads
+    bool valid = regex_search(server_payload.begin(), server_payload.end(),
+                              server_match, response_regex) &&
+                 regex_search(client_payload.begin(), client_payload.end(),
+                              client_match, request_regex);
+    // If we matched both the client and the server regexes
+    if (valid) {
+        // Extract all fields
+        string method = string(client_match[1].first, client_match[1].second);
+        string url = string(client_match[2].first, client_match[2].second);
+        string host = string(client_match[3].first, client_match[3].second);
+        string response_code = string(server_match[1].first, server_match[1].second);
+        // Now print them
+        cout << method << " http://" << host << url << " -> " << response_code << endl;
+
+        // Once we've seen the first request on this stream, ignore it
+        stream.ignore_client_data();
+        stream.ignore_server_data();
+    }
+    
+    // Just in case the server returns invalid data, stop at 3kb
+    if (stream.server_payload().size() > MAX_PAYLOAD) {
+        stream.ignore_server_data();
+    }
+}
+
+void on_client_data(Stream& stream) {
+    // Don't hold more than 3kb of data from the client's flow
+    if (stream.client_payload().size() > MAX_PAYLOAD) {
+        stream.ignore_client_data();
+    }
+}
+
+void on_new_connection(Stream& stream) {
+    stream.client_data_callback(&on_client_data);
+    stream.server_data_callback(&on_server_data);
+    // Don't automatically cleanup the stream's data, as we'll manage
+    // the buffer ourselves and let it grow until we see a full request
+    // and response
+    stream.auto_cleanup_payloads(false);
+}
+
+int main(int argc, char* argv[]) {
+    if (argc != 2) {
+        cout << "Usage: " << argv[0] << " <interface>" << endl;
+        return 1;
+    }
+
+    try {
+        // Construct the sniffer configuration object
+        SnifferConfiguration config;
+        // Only capture TCP traffic sent from/to port 80
+        config.set_filter("tcp port 80");
+        // Construct the sniffer we'll use
+        Sniffer sniffer(argv[1], config);
+
+        cout << "Starting capture on interface " << argv[1] << endl;
+
+        // Now construct the stream follower
+        StreamFollower follower;
+        // We just need to specify the callback to be executed when a new 
+        // stream is captured. In this stream, you should define which callbacks
+        // will be executed whenever new data is sent on that stream 
+        // (see on_new_connection)
+        follower.new_stream_callback(&on_new_connection);
+        // Now start capturing. Every time there's a new packet, call 
+        // follower.process_packet
+        sniffer.sniff_loop([&](Packet& packet) {
+            follower.process_packet(packet);
+            return true;
+        });
+    }
+    catch (exception& ex) {
+        cerr << "Error: " << ex.what() << endl;
+        return 1;
+    }
+}

examples/stream_dump.cpp

@@ -0,0 +1,161 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <iostream>
+#include <sstream>
+#include "tins/tcp_ip/stream_follower.h"
+#include "tins/sniffer.h"
+#include "tins/packet.h"
+#include "tins/ip_address.h"
+#include "tins/ipv6_address.h"
+
+using std::cout;
+using std::cerr;
+using std::endl;
+using std::bind;
+using std::string;
+using std::ostringstream;
+using std::exception;
+
+using Tins::Sniffer;
+using Tins::SnifferConfiguration;
+using Tins::Packet;
+using Tins::TCPIP::StreamFollower;
+using Tins::TCPIP::Stream;
+
+// This example takes an interface and a port as an argument and
+// it listens for TCP streams on the given interface and port.
+// It will reassemble TCP streams and show the traffic sent by
+// both the client and the server.
+
+// Convert the client endpoint to a readable string
+string client_endpoint(const Stream& stream) {
+    ostringstream output;
+    // Use the IPv4 or IPv6 address depending on which protocol the 
+    // connection uses
+    if (stream.is_v6()) {
+        output << stream.client_addr_v6();
+    }
+    else {
+        output << stream.client_addr_v4();
+    }
+    output << ":" << stream.client_port();
+    return output.str();
+}
+
+// Convert the server endpoint to a readable string
+string server_endpoint(const Stream& stream) {
+    ostringstream output;
+    if (stream.is_v6()) {
+        output << stream.server_addr_v6();
+    }
+    else {
+        output << stream.server_addr_v4();
+    }
+    output << ":" << stream.server_port();
+    return output.str();
+}
+
+// Concat both endpoints to get a readable stream identifier
+string stream_identifier(const Stream& stream) {
+    ostringstream output;
+    output << client_endpoint(stream) << " - " << server_endpoint(stream);
+    return output.str();
+}
+
+// Whenever there's new client data on the stream, this callback is executed.
+void on_client_data(Stream& stream) {
+    // Construct a string out of the contents of the client's payload
+    string data(stream.client_payload().begin(), stream.client_payload().end());
+    // Now print it, prepending some information about the stream
+    cout << client_endpoint(stream) << " >> " 
+         << server_endpoint(stream) << ": " << endl << data << endl;
+}
+
+// Whenever there's new server data on the stream, this callback is executed.
+// This does the same thing as on_client_data
+void on_server_data(Stream& stream) {
+    string data(stream.server_payload().begin(), stream.server_payload().end());
+    cout << server_endpoint(stream) << " >> " 
+         << client_endpoint(stream) << ": " << endl << data << endl;
+}
+
+// When a connection is closed, this callback is executed.
+void on_connection_closed(Stream& stream) {
+    cout << "[+] Connection closed: " << stream_identifier(stream) << endl;
+}
+
+// When a new connection is captured, this callback will be executed.
+void on_new_connection(Stream& stream) {
+    // Print some information about the new connection
+    cout << "[+] New connection " << stream_identifier(stream) << endl;
+    // Now configure the callbacks on it.
+    // First, we want on_client_data to be called every time there's new client data
+    stream.client_data_callback(&on_client_data);
+    // Same thing for server data, but calling on_server_data
+    stream.server_data_callback(&on_server_data);
+    // When the connection is closed, call on_connection_closed
+    stream.stream_closed_callback(&on_connection_closed);
+}
+
+int main(int argc, char* argv[]) {
+    if (argc != 3) {
+        cout << "Usage: " << argv[0] << " <interface> <port>" << endl;
+        return 1;
+    }
+
+    try {
+        // Construct the sniffer configuration object
+        SnifferConfiguration config;
+        // Only capture TCP traffic sent from/to the given port
+        config.set_filter("tcp port " + to_string(stoi(string(argv[2]))));
+        // Construct the sniffer we'll use
+        Sniffer sniffer(argv[1], config);
+
+        cout << "Starting capture on interface " << argv[1] << endl;
+
+        // Now construct the stream follower
+        StreamFollower follower;
+        // We just need to specify the callback to be executed when a new 
+        // stream is captured. In this stream, you should define which callbacks
+        // will be executed whenever new data is sent on that stream 
+        // (see on_new_connection)
+        follower.new_stream_callback(&on_new_connection);
+        // Now start capturing. Every time there's a new packet, call 
+        // follower.process_packet
+        sniffer.sniff_loop([&](Packet& packet) {
+            follower.process_packet(packet);
+            return true;
+        });
+    }
+    catch (exception& ex) {
+        cerr << "Error: " << ex.what() << endl;
+        return 1;
+    }
+}

include/tins/CMakeLists.txt

@@ -5,3 +5,4 @@ INSTALL(
     COMPONENT Headers
 )
 ADD_SUBDIRECTORY(dot11)
+ADD_SUBDIRECTORY(tcp_ip)

include/tins/config.h.in

@@ -13,4 +13,7 @@
 /* Use pcap_sendpacket to send l2 packets */
 #cmakedefine HAVE_PACKET_SENDER_PCAP_SENDPACKET
 
+/* Have TCP ACK tracking */
+#cmakedefine HAVE_ACK_TRACKER
+
 #endif // TINS_CONFIG_H

include/tins/exceptions.h

@@ -200,6 +200,17 @@ public:
     }
 };
 
+/**
+ * \brief Exception thrown when a feature has been disabled
+ * at compile time.
+ */
+class feature_disabled : public exception_base {
+public:
+    const char* what() const throw() {
+        return "Feature disabled";
+    }
+};
+
 /**
  * \brief Exception thrown when a payload is too large to fit
  * into a PDUOption.
@@ -273,6 +284,36 @@ public:
     }
 };
 
+/**
+ * \brief Exception thrown when a stream is not found
+ */
+class stream_not_found : public exception_base {
+public:
+    const char* what() const throw() {
+        return "Stream not found";
+    }
+};
+
+/**
+ * \brief Exception thrown when a required callback for an object is not set
+ */
+class callback_not_set : public exception_base {
+public:
+    const char* what() const throw() {
+        return "Callback not set";
+    }
+};
+
+/**
+ * \brief Exception thrown when an invalid packet is provided to some function
+ */
+class invalid_packet : public exception_base {
+public:
+    const char* what() const throw() {
+        return "Invalid packet";
+    }
+};
+
 namespace Crypto {
 namespace WPA2 {
     /**

include/tins/internals.h

@@ -177,6 +177,9 @@ bool decrement(HWAddress<n>& addr) {
     return decrement_buffer(addr);
 }
 
+// Compares sequence numbers as defined by RFC 1982.
+int seq_compare(uint32_t seq1, uint32_t seq2);
+
 IPv4Address last_address_from_mask(IPv4Address addr, IPv4Address mask);
 IPv6Address last_address_from_mask(IPv6Address addr, const IPv6Address& mask);
 template<size_t n>

include/tins/packet.h

@@ -148,11 +148,19 @@ public:
     /**
      * \brief Constructs a Packet from a PDU* and a Timestamp.
      * 
-     * The PDU* is cloned using PDU::clone.
+     * The PDU is cloned using PDU::clone.
      */
     Packet(const PDU* apdu, const Timestamp& tstamp) 
     : pdu_(apdu->clone()), ts_(tstamp) { }
 
+    /**
+     * \brief Constructs a Packet from a PDU& and a Timestamp.
+     * 
+     * The PDU is cloned using PDU::clone.
+     */
+    Packet(const PDU& apdu, const Timestamp& tstamp) 
+    : pdu_(apdu.clone()), ts_(tstamp) { }
+
     /**
      * \brief Constructs a Packet from a PDU* and a Timestamp.
      * 

include/tins/tcp_ip/CMakeLists.txt

@@ -0,0 +1,6 @@
+FILE(GLOB INCLUDE_FILES "*.h")
+INSTALL(
+    FILES ${INCLUDE_FILES} 
+    DESTINATION include/tins/tcp_ip
+    COMPONENT Headers
+)

include/tins/tcp_ip/ack_tracker.h

@@ -0,0 +1,155 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef TINS_TCP_IP_ACK_TRACKER_H
+#define TINS_TCP_IP_ACK_TRACKER_H
+
+#include "../config.h"
+
+#ifdef HAVE_ACK_TRACKER
+
+#include <vector>
+#include <boost/icl/interval_set.hpp>
+
+namespace Tins {
+
+class PDU;
+
+namespace TCPIP {
+
+/**
+ * \brief Represents an acknowledged segment range
+ *
+ * The interval represented by this range is a closed interval [first, last].
+ */
+class AckedRange {
+public:
+    typedef boost::icl::discrete_interval<uint32_t> interval_type;
+
+    /**
+     * \brief Constructs an acked range
+     *
+     * \param first The first acked byte
+     * \param last The last acked byte (inclusive)
+     */
+    AckedRange(uint32_t first, uint32_t last);
+
+    /**
+     * \brief Gets the next acked interval in this range
+     *
+     * If has_next() == false, then this returns an empty interval
+     */
+    interval_type next();
+
+    /**
+     * Indicates whether there is still some non-consumed acked-interval in this
+     * range
+     */
+    bool has_next() const;
+
+    /**
+     * Gets the first index acked by this range
+     */
+    uint32_t first() const;
+
+    /**
+     * Gets the last index acked by this range
+     */
+    uint32_t last() const;
+private:
+    uint32_t first_;
+    uint32_t last_;
+};
+
+/**
+ * \brief Allows tracking acknowledged intervals in a TCP stream
+ */
+class AckTracker {
+public:
+    /**
+     * The type used to store ACKed intervals
+     */
+    typedef boost::icl::interval_set<uint32_t> interval_set_type;
+
+    /**
+     * Default constructor
+     */
+    AckTracker();
+
+    /**
+     * \brief Construct an instance using some attributes
+     *
+     * \param intial_ack The initial ACK number to use
+     * \param use_sack Indicate whether to use Selective ACKs to track ACK numbers
+     */
+    AckTracker(uint32_t initial_ack, bool use_sack = true);
+
+    /**
+     * \brief Process a packet
+     */
+    void process_packet(const PDU& packet);
+
+    /**
+     * \brief Indicates whether Selective ACKs should be processed
+     */
+    void use_sack();
+
+    /**
+     * Retrieves the current ACK number in this tracker
+     */
+    uint32_t ack_number() const;
+
+    /**
+     * \brief Retrieves all acked intervals by Selective ACKs
+     */
+    const interval_set_type& acked_intervals() const; 
+
+    /**
+     * \brief Indicates whether the given segment has been already ACKed
+     *
+     * \param sequence_number The segment's sequence number
+     * \param length The segment's length
+     */
+    bool is_segment_acked(uint32_t sequence_number, uint32_t length) const;
+private:
+    void process_sack(const std::vector<uint32_t>& sack);
+    void cleanup_sacked_intervals(uint32_t old_ack, uint32_t new_ack);
+
+    interval_set_type acked_intervals_;
+    uint32_t ack_number_;
+    bool use_sack_;
+};
+
+} // TCPIP
+} // Tins
+
+#endif // HAVE_ACK_TRACKER
+
+#endif // TINS_TCP_IP_ACK_TRACKER_H
+

include/tins/tcp_ip/flow.h

@@ -0,0 +1,336 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef TINS_TCP_IP_FLOW_H
+#define TINS_TCP_IP_FLOW_H
+
+#include "../cxxstd.h"
+
+// This classes use C++11 features
+#if TINS_IS_CXX11
+
+#include <vector>
+#include <array>
+#include <map>
+#include <functional>
+#include <stdint.h>
+#include "../macros.h"
+#include "ack_tracker.h"
+#include "../hw_address.h"
+
+namespace Tins {
+
+class PDU;
+class TCP;
+class IPv4Address;
+class IPv6Address;
+
+namespace TCPIP {
+
+/**
+ * \brief Represents an unidirectional TCP flow between 2 endpoints
+ *
+ * This class will keep the state for all the traffic sent by
+ * one of the peers in a TCP connection. This contains the sequence number,
+ * payload ready to be read and buffered payload, along with some other
+ * properties of the flow.
+ *
+ * A TCP stream (see class Stream) is made out of 2 Flows, so you should 
+ * probably have a look at that class first.
+ *
+ * You shouldn't normally need to interact with this class. Stream already
+ * provides proxys to most of its Flow's attributes.
+ */
+class TINS_API Flow {
+public:
+    /**
+     * \brief Enum that indicates the state of this flow.
+     * 
+     * Note that although similar, this is not mapped to a TCP state-machine 
+     * state. This is mostly used internally to know which packets the flow is
+     * expecting and to know when it's done sending data.
+     */
+    enum State {
+        UNKNOWN,
+        SYN_SENT,
+        ESTABLISHED,
+        FIN_SENT,
+        RST_SENT
+    };
+
+    /** 
+     * The type used to store the payload
+     */
+    typedef std::vector<uint8_t> payload_type;
+
+    /**
+     * The type used to store the buffered payload
+     */
+    typedef std::map<uint32_t, payload_type> buffered_payload_type;
+
+    /**
+     * The type used to store the callback called when new data is available
+     */
+    typedef std::function<void(Flow&)> data_available_callback_type;
+
+    /**
+     * \brief The type used to store the callback called when data is buffered
+     *
+     * The arguments are the flow, the sequence number and payload that will
+     * be buffered.
+     */
+    typedef std::function<void(Flow&,
+                               uint32_t,
+                               const payload_type&)> flow_packet_callback_type;
+
+    /** 
+     * Construct a Flow from an IPv4 address
+     *
+     * \param dst_address This flow's destination address
+     * \param dst_port This flow's destination port
+     * \param sequence_number The initial sequence number to be used 
+     */
+    Flow(const IPv4Address& dst_address, uint16_t dst_port,
+         uint32_t sequence_number);
+    
+    /** 
+     * Construct a Flow from an IPv6 address
+     *
+     * \param dst_address This flow's destination address
+     * \param dst_port This flow's destination port
+     * \param sequence_number The initial sequence number to be used 
+     */
+    Flow(const IPv6Address& dst_address, uint16_t dst_port,
+         uint32_t sequence_number);
+
+    /**
+     * \brief Sets the callback that will be executed when data is readable
+     *
+     * Whenever this flow has readable data, this callback will be executed.
+     * By readable, this means that there's non-out-of-order data captured.
+     *
+     * \param callback The callback to be executed   
+     */
+    void data_callback(const data_available_callback_type& callback);
+
+    /**
+     * \brief Sets the callback that will be executed when out of order data arrives
+     *
+     * Whenever this flow receives out-of-order data, this callback will be
+     * executed.
+     * 
+     * \param callback The callback to be executed
+     */
+    void out_of_order_callback(const flow_packet_callback_type& callback);
+
+    /**
+     * \brief Processes a packet.
+     *
+     * If this packet contains data and starts or overlaps with the current
+     * sequence number, then the data will be appended to this flow's payload
+     * and the data_callback will be executed.
+     *
+     * If this packet contains out-of-order data, it will be buffered and the
+     * buffering_callback will be executed.
+     *
+     * \param pdu The packet to be processed
+     * \sa Flow::data_callback
+     * \sa Flow::buffering_callback
+     */
+    void process_packet(PDU& pdu);
+
+    /**
+     * Indicates whether this flow uses IPv6 addresses
+     */
+    bool is_v6() const;
+
+    /**
+     * \brief Indicates whether this flow is finished
+     *
+     * A finished is considered to be finished if either it sent a
+     * packet with the FIN or RST flags on. 
+     */
+    bool is_finished() const;
+
+    /**
+     * \brief Indicates whether a packet belongs to this flow
+     *
+     * Since Flow represents a unidirectional stream, this will only check
+     * the destination endpoint and not the source one.
+     *
+     * \param packet The packet to be checked
+     */
+    bool packet_belongs(const PDU& packet) const;
+
+    /**
+     * \brief Retrieves the IPv4 destination address
+     *
+     * Note that it's only safe to execute this method if is_v6() == false
+     */
+    IPv4Address dst_addr_v4() const;
+
+    /**
+     * \brief Retrieves the IPv6 destination address
+     *
+     * Note that it's only safe to execute this method if is_v6() == true
+     */
+    IPv6Address dst_addr_v6() const;
+
+    /** 
+     * Retrieves this flow's destination port
+     */
+    uint16_t dport() const;
+
+    /** 
+     * Retrieves this flow's payload (const)
+     */
+    const payload_type& payload() const;
+
+    /** 
+     * Retrieves this flow's destination port
+     */
+    payload_type& payload();
+
+    /** 
+     * Retrieves this flow's state
+     */
+    State state() const;
+
+    /** 
+     * Retrieves this flow's sequence number
+     */
+    uint32_t sequence_number() const;
+
+    /** 
+     * Retrieves this flow's buffered payload (const)
+     */
+    const buffered_payload_type& buffered_payload() const;
+
+    /** 
+     * Retrieves this flow's buffered payload
+     */
+    buffered_payload_type& buffered_payload();
+
+    /**
+     * Retrieves this flow's total buffered bytes
+     */
+    uint32_t total_buffered_bytes() const;
+
+    /**
+     * Sets the state of this flow
+     *
+     * \param new_state The new state of this flow
+     */
+    void state(State new_state);
+
+    /**
+     * \brief Sets whether this flow should ignore data packets
+     *
+     * If the data packets are ignored then the flow will just be 
+     * followed to keep track of its state.
+     */
+    void ignore_data_packets();
+
+    /**
+     * \brief Returns the MSS for this Flow.
+     *
+     * If the MSS option wasn't provided by the peer, -1 is returned
+     */
+    int mss() const;
+
+    /**
+     * \brief Indicates whether this Flow supports selective acknowledgements
+     */
+    bool sack_permitted() const;
+
+    /** 
+     * \brief Enables tracking of ACK numbers
+     *
+     * This requires having the boost.icl library. If the library is not installed
+     * or ACK tracking was disabled when compiling the library, then this method
+     * will throw an exception.
+     */
+    void enable_ack_tracking();
+
+    /**
+     * \brief Indicates whether ACK number tracking is enabled
+     */
+    bool ack_tracking_enabled() const;
+
+    #ifdef HAVE_ACK_TRACKER
+    /**
+     * Retrieves the ACK tracker for this Flow (const)
+     */
+    const AckTracker& ack_tracker() const;
+
+    /**
+     * Retrieves the ACK tracker for this Flow
+     */
+    AckTracker& ack_tracker();
+    #endif // HAVE_ACK_TRACKER
+private:
+    // Compress all flags into just one struct using bitfields 
+    struct flags {
+        flags() : ignore_data_packets(0), sack_permitted(0), ack_tracking(0) {
+
+        }
+
+        uint32_t is_v6:1,
+                 ignore_data_packets:1,
+                 sack_permitted:1,
+                 ack_tracking:1;
+    };
+
+    void store_payload(uint32_t seq, payload_type payload);
+    buffered_payload_type::iterator erase_iterator(buffered_payload_type::iterator iter);
+    void update_state(const TCP& tcp);
+    void initialize();
+
+    payload_type payload_;
+    buffered_payload_type buffered_payload_;
+    uint32_t seq_number_;
+    uint32_t total_buffered_bytes_;
+    std::array<uint8_t, 16> dest_address_;
+    uint16_t dest_port_;
+    data_available_callback_type on_data_callback_;
+    flow_packet_callback_type on_out_of_order_callback_;
+    State state_;
+    int mss_;
+    flags flags_;
+    #ifdef HAVE_ACK_TRACKER
+    AckTracker ack_tracker_;
+    #endif // HAVE_ACK_TRACKER
+};
+
+} // TCPIP
+} // TINS
+
+#endif // TINS_IS_CXX11
+#endif // TINS_TCP_IP_FLOW_H
+

include/tins/tcp_ip/stream.h

@@ -0,0 +1,398 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef TINS_TCP_IP_STREAM_H
+#define TINS_TCP_IP_STREAM_H
+
+#include "../cxxstd.h"
+
+// This classes use C++11 features
+#if TINS_IS_CXX11
+
+#include <vector>
+#include <array>
+#include <map>
+#include <functional>
+#include <chrono>
+#include <stdint.h>
+#include "../macros.h"
+#include "../hw_address.h"
+#include "flow.h"
+
+namespace Tins {
+
+class PDU;
+class TCP;
+class IPv4Address;
+class IPv6Address;
+
+namespace TCPIP {
+
+/** 
+ * \brief Represents a TCP stream
+ *
+ * A TCP stream is made out of 2 Flows, one in each direction, plus 
+ * some other attributes and callbacks.
+ *
+ * This class works using callbacks. Whenever the stream is created, you should
+ * set at least the client/server callbacks so you are notified whenever the
+ * client/server has sent data. Note that setting these is not mandatory, so
+ * you can subscribe to just the callbacks you need.
+ *
+ * \sa Stream::auto_cleanup_payloads
+ */
+class TINS_API Stream {
+public:
+    /**
+     * The type used to store payloads
+     */
+    typedef Flow::payload_type payload_type;
+
+    /** 
+     * The type used to represent timestamps
+     */
+    typedef std::chrono::microseconds timestamp_type;
+    
+    /**
+     * The type used for callbacks
+     */
+    typedef std::function<void(Stream&)> stream_callback_type;
+
+    /**
+     * The type used for packet-triggered callbacks
+     *
+     * /sa Flow::buffering_callback
+     */
+    typedef std::function<void(Stream&,
+                               uint32_t,
+                               const payload_type&)> stream_packet_callback_type;
+
+    /**
+     * The type used to store hardware addresses
+     */
+    typedef HWAddress<6> hwaddress_type;
+
+
+    /**
+     * \brief Constructs a TCP stream using the provided packet.
+     * 
+     * \param initial_packet The first packet of the stream
+     * \param ts The first packet's timestamp
+     */
+    Stream(PDU& initial_packet, const timestamp_type& ts = timestamp_type());
+
+    /**
+     * \brief Processes this packet.
+     *
+     * This will forward the packet appropriately to the client
+     * or server flow.
+     *
+     * \param packet The packet to be processed
+     * \param ts The packet's timestamp
+     */
+    void process_packet(PDU& packet, const timestamp_type& ts);
+
+    /**
+     * \brief Processes this packet.
+     *
+     * This will forward the packet appropriately to the client
+     * or server flow.
+     *
+     * \param packet The packet to be processed
+     */
+    void process_packet(PDU& packet);
+
+    /**
+     * Getter for the client flow
+     */
+    Flow& client_flow();
+
+    /**
+     * Getter for the client flow (const)
+     */
+    const Flow& client_flow() const;
+
+    /**
+     * Getter for the server flow
+     */
+    Flow& server_flow();
+
+    /**
+     * Getter for the server flow (const)
+     */
+    const Flow& server_flow() const;
+
+    /**
+     * \brief Indicates whether this stream is finished.
+     *
+     * This stream is finished if either peer sent a packet with
+     * the RST flag on, or both peers sent a FIN.
+     */
+    bool is_finished() const;
+
+    /**
+     * Indicates whether this packet uses IPv6 addresses
+     */
+    bool is_v6() const;
+
+    /**
+     * \brief Retrieves the client's IPv4 address
+     *
+     * Note that it's only valid to call this method if is_v6() == false
+     */
+    IPv4Address client_addr_v4() const;
+
+    /**
+     * \brief Retrieves the client's IPv6 address
+     *
+     * Note that it's only valid to call this method if is_v6() == true
+     */
+    IPv6Address client_addr_v6() const;
+
+    /**
+     * \brief Retrieves the client's hardware address.
+     *
+     * Note that this is not the actual hardware address of the client, but
+     * just the address seen from packets coming from it. If the client
+     * is on another network, then this will be the address of the last
+     * device (switch, route, etc) the packet went through.
+     */
+    const hwaddress_type& client_hw_addr() const;
+
+    /**
+     * \brief Retrieves the server's hardware address.
+     *
+     * Note that this is not the actual hardware address of the server, but
+     * just the address seen from packets coming from it. If the server
+     * is on another network, then this will be the address of the last
+     * device (switch, route, etc) the packet went through.
+     */
+    const hwaddress_type& server_hw_addr() const;
+
+    /**
+     * \brief Retrieves the server's IPv4 address
+     *
+     * Note that it's only valid to call this method if is_v6() == false
+     */
+    IPv4Address server_addr_v4() const;
+
+    /**
+     * \brief Retrieves the server's IPv6 address
+     *
+     * Note that it's only valid to call this method if is_v6() == true
+     */
+    IPv6Address server_addr_v6() const;
+
+    /**
+     * Getter for the client's port
+     */
+    uint16_t client_port() const;
+
+    /**
+     * Getter for the server's port
+     */
+    uint16_t server_port() const;
+
+    /**
+     * Getter for the client's payload (const)
+     */
+    const payload_type& client_payload() const;
+
+    /**
+     * Getter for the client's payload
+     */
+    payload_type& client_payload();
+
+    /**
+     * Getter for the server's payload (const)
+     */
+    const payload_type& server_payload() const;
+
+    /**
+     * Getter for the server's payload
+     */
+    payload_type& server_payload();
+
+    /**
+     * Getter for the creation time of this stream
+     */
+    const timestamp_type& create_time() const;
+
+    /**
+     * Getter for the last seen time of this stream
+     */
+    const timestamp_type& last_seen() const;
+
+    /**
+     * \brief Sets the callback to be executed when the stream is closed
+     *
+     * \param callback The callback to be set
+     */
+    void stream_closed_callback(const stream_callback_type& callback);
+
+    /**
+     * \brief Sets the callback to be executed when there's client data
+     *
+     * \sa Flow::data_callback
+     * \param callback The callback to be set
+     */
+    void client_data_callback(const stream_callback_type& callback);
+
+    /**
+     * \brief Sets the callback to be executed when there's server data
+     *
+     * \sa Flow::data_callback
+     * \param callback The callback to be set
+     */
+    void server_data_callback(const stream_callback_type& callback);
+
+    /**
+     * \brief Sets the callback to be executed when there's new buffered 
+     * client data
+     *
+     * \sa Flow::buffering_callback
+     * \param callback The callback to be set
+     */
+    void client_out_of_order_callback(const stream_packet_callback_type& callback);
+
+    /**
+     * \brief Sets the callback to be executed when there's new buffered 
+     * client data
+     *
+     * \sa Flow::buffering_callback
+     * \param callback The callback to be set
+     */
+    void server_out_of_order_callback(const stream_packet_callback_type& callback);
+
+    /**
+     * \brief Indicates that the data packets sent by the client should be 
+     * ignored
+     *
+     * \sa Flow::ignore_data_packets
+     */
+    void ignore_client_data();
+
+    /**
+     * \brief Indicates that the data packets sent by the server should be 
+     * ignored
+     *
+     * \sa Flow::ignore_data_packets
+     */
+    void ignore_server_data();
+
+    /**
+     * \brief Sets the internal callbacks. 
+     *
+     * This shouldn't normally need to be called except if you're constructing
+     * this object and then moving it around before persisting it somewhere.  
+     */
+    void setup_flows_callbacks();
+
+    /**
+     * \brief Indicates whether each flow's payloads should be automatically
+     * erased.
+     *
+     * If this property is true, then whenever there's new data for a stream,
+     * the appropriate callback will be executed and then the payload will be 
+     * erased. 
+     *
+     * If this property is false, then the payload <b>will not</b> be erased
+     * and the user is responsible for clearing the payload vector. 
+     *
+     * Setting this property to false is useful if it's desired to hold all 
+     * of the data sent on the stream before processing it. Note that this
+     * can lead to the memory growing a lot.
+     *
+     * This property is true by default. 
+     *
+     * \param value The value to be set for this property
+     */
+    void auto_cleanup_payloads(bool value);
+
+    /**
+     * \brief Indicates whether the client flow's payloads should be 
+     * automatically erased.
+     *
+     * \sa auto_cleanup_payloads
+     */
+    void auto_cleanup_client_data(bool value);
+
+    /**
+     * \brief Indicates whether the server flow's payloads should be 
+     * automatically erased.
+     *
+     * \sa auto_cleanup_payloads
+     */
+    void auto_cleanup_server_data(bool value);
+
+    /**
+     * Enables tracking of acknowledged segments
+     *
+     * \sa Flow::enable_ack_tracking
+     */
+    void enable_ack_tracking();
+
+    /**
+     * \brief Indicates whether ACK number tracking is enabled for this stream
+     */
+    bool ack_tracking_enabled() const;
+private:
+    static Flow extract_client_flow(const PDU& packet);
+    static Flow extract_server_flow(const PDU& packet);
+
+    void on_client_flow_data(const Flow& flow);
+    void on_server_flow_data(const Flow& flow);
+    void on_client_out_of_order(const Flow& flow,
+                                uint32_t seq,
+                                const payload_type& payload);
+    void on_server_out_of_order(const Flow& flow,
+                                uint32_t seq,
+                                const payload_type& payload);
+
+    Flow client_flow_;
+    Flow server_flow_;
+    stream_callback_type on_stream_closed_;
+    stream_callback_type on_client_data_callback_;
+    stream_callback_type on_server_data_callback_;
+    stream_packet_callback_type on_client_out_of_order_callback_;
+    stream_packet_callback_type on_server_out_of_order_callback_;
+    hwaddress_type client_hw_addr_;
+    hwaddress_type server_hw_addr_;
+    timestamp_type create_time_;
+    timestamp_type last_seen_;
+    bool auto_cleanup_client_;
+    bool auto_cleanup_server_;
+};
+
+} // TCPIP
+} // Tins
+
+#endif // TINS_IS_CXX11 
+
+#endif // TINS_TCP_IP_STREAM_H

include/tins/tcp_ip/stream_follower.h

@@ -0,0 +1,251 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef TINS_TCP_IP_STREAM_FOLLOWER_H
+#define TINS_TCP_IP_STREAM_FOLLOWER_H
+
+#include "../cxxstd.h"
+
+// This classes use C++11 features
+#if TINS_IS_CXX11
+
+#include <map>
+#include "stream.h"
+
+namespace Tins {
+
+class PDU;
+class TCP;
+class IPv4Address;
+class IPv6Address;
+class Packet;
+
+namespace TCPIP {
+
+/**
+ * \brief Represents a class that follows TCP and reassembles streams
+ *
+ * This class processes packets and whenever it detects a new connection
+ * being open, it starts tracking it. This will follow all data sent by
+ * each peer and make it available to the user in a simple way.
+ *
+ * In order to use this class, just create an instance and set the 
+ * new stream callback to some function that you want:
+ *
+ * \code
+ * void on_new_stream(TCPStream& stream) {
+ *     // Do something with it.
+ *     // This is the perfect time to set the stream's client/server 
+ *     // write callbacks so you are notified whenever there's new
+ *     // data on the stream
+ * }
+ *
+ * // Create it 
+ * StreamFollower follower;
+ * // Set the callback
+ * follower.new_stream_callback(&on_new_stream);
+ * \endcode
+ */
+class TINS_API StreamFollower {
+public:
+    /**
+     * \brief The type used for callbacks
+     */
+    typedef Stream::stream_callback_type stream_callback_type;
+
+    /**
+     * Enum to indicate the reason why a stream was terminated
+     */
+    enum TerminationReason {
+        TIMEOUT, ///< The stream was terminated due to a timeout
+        BUFFERED_DATA, ///< The stream was terminated because it had too much buffered data
+        SACKED_SEGMENTS ///< The stream was terminated because it had too many SACKed segments
+    };
+
+    /**
+     * \brief The type used for stream termination callbacks
+     *
+     * \sa StreamFollower::stream_termination_callback
+     */
+    typedef std::function<void(Stream&, TerminationReason)> stream_termination_callback_type;
+
+    /**
+     * \brief Unique identifies a stream. 
+     *
+     * This struct is used to track TCP streams. It keeps track of minimum and maximum
+     * addresses/ports in a stream to match packets coming from any of the 2 endpoints
+     * into the same object.
+     */
+    struct stream_id {
+        /**
+         * The type used to store each endpoint's address
+         */
+        typedef std::array<uint8_t, 16> address_type;
+
+        /**
+         * Default constructor
+         */
+        stream_id();
+
+        /**
+         * Constructs a stream_id
+         *
+         * \param client_addr Client's address
+         * \param client_port Port's port
+         * \param server_addr Server's address
+         * \param server_port Server's port
+         */
+        stream_id(const address_type& client_addr, uint16_t client_port,
+                  const address_type& server_addr, uint16_t server_port);
+
+        address_type min_address;
+        address_type max_address;
+        uint16_t min_address_port;
+        uint16_t max_address_port;
+
+        bool operator<(const stream_id& rhs) const;
+
+        static size_t hash(const stream_id& id);
+    };
+
+    /** 
+     * Default constructor
+     */
+    StreamFollower();
+
+    /** 
+     * \brief Processes a packet
+     *
+     * This will detect if this packet belongs to an existing stream 
+     * and process it, or if it belongs to a new one, in which case it
+     * starts tracking it.
+     *
+     * \param packet The packet to be processed
+     */
+    void process_packet(PDU& packet);
+
+    /** 
+     * \brief Processes a packet
+     *
+     * This will detect if this packet belongs to an existing stream 
+     * and process it, or if it belongs to a new one, in which case it
+     * starts tracking it.
+     *
+     * \param packet The packet to be processed
+     */
+    void process_packet(Packet& packet);
+
+    /**
+     * \brief Sets the callback to be executed when a new stream is captured.
+     *
+     * Whenever a new stream is captured, the provided callback will be 
+     * executed.
+     *
+     * \param callback The callback to be set
+     */
+    void new_stream_callback(const stream_callback_type& callback);
+
+    /**
+     * \brief Sets the stream termination callback
+     *
+     * A stream is terminated when either:
+     *
+     * * It contains too much buffered data.
+     * * No packets have been seen for some time interval.
+     *
+     * \param callback The callback to be executed on stream termination
+     * \sa StreamFollower::stream_keep_alive
+     */
+    void stream_termination_callback(const stream_termination_callback_type& callback);
+
+    /**
+     * \brief Sets the maximum time a stream will be followed without capturing
+     * packets that belong to it.
+     *
+     * \param keep_alive The maximum time to keep unseen streams
+     */
+    template <typename Rep, typename Period>
+    void stream_keep_alive(const std::chrono::duration<Rep, Period>& keep_alive) {
+        stream_keep_alive_ = keep_alive;
+    }
+
+    /**
+     * Finds the stream identified by the provided arguments.
+     *
+     * \param client_addr The client's address
+     * \param client_port The client's port
+     * \param server_addr The server's address
+     * \param server_addr The server's port
+     */
+    Stream& find_stream(const IPv4Address& client_addr, uint16_t client_port,
+                        const IPv4Address& server_addr, uint16_t server_port);
+
+    /**
+     * Finds the stream identified by the provided arguments.
+     *
+     * \param client_addr The client's address
+     * \param client_port The client's port
+     * \param server_addr The server's address
+     * \param server_addr The server's port
+     */
+    Stream& find_stream(const IPv6Address& client_addr, uint16_t client_port,
+                        const IPv6Address& server_addr, uint16_t server_port);
+private:
+    typedef Stream::timestamp_type timestamp_type;
+
+    static const size_t DEFAULT_MAX_BUFFERED_CHUNKS;
+    static const size_t DEFAULT_MAX_SACKED_INTERVALS;
+    static const uint32_t DEFAULT_MAX_BUFFERED_BYTES;
+    static const timestamp_type DEFAULT_KEEP_ALIVE;
+
+    typedef std::map<stream_id, Stream> streams_type;
+
+    static stream_id make_stream_id(const PDU& packet);
+    Stream& find_stream(const stream_id& id);
+    static stream_id::address_type serialize(IPv4Address address);
+    static stream_id::address_type serialize(const IPv6Address& address);
+    void process_packet(PDU& packet, const timestamp_type& ts);
+    void cleanup_streams(const timestamp_type& now);
+
+    streams_type streams_;
+    stream_callback_type on_new_connection_;
+    stream_termination_callback_type on_stream_termination_;
+    size_t max_buffered_chunks_;
+    uint32_t max_buffered_bytes_;
+    timestamp_type last_cleanup_;
+    timestamp_type stream_keep_alive_;
+    bool attach_to_flows_;
+};
+
+} // TCPIP
+} // Tins
+
+#endif // TINS_IS_CXX11
+
+#endif // TINS_TCP_IP_STREAM_FOLLOWER_H

src/CMakeLists.txt

@@ -53,6 +53,10 @@ ADD_LIBRARY(
     snap.cpp
     sniffer.cpp
     tcp.cpp
+    tcp_ip/ack_tracker.cpp
+    tcp_ip/flow.cpp
+    tcp_ip/stream.cpp
+    tcp_ip/stream_follower.cpp
     tcp_stream.cpp
     udp.cpp
     utils.cpp

src/internals.cpp

@@ -349,6 +349,20 @@ bool decrement(IPv6Address& addr) {
     return decrement_buffer(addr);
 }
 
+int seq_compare(uint32_t seq1, uint32_t seq2) {
+    // As defined by RFC 1982 - 2 ^ (SERIAL_BITS - 1)
+    static const uint32_t seq_number_diff = 2147483648U;
+    if (seq1 == seq2) {
+        return 0;
+    }
+    if (seq1 < seq2) {
+        return (seq2 - seq1 < seq_number_diff) ? -1 : 1;
+    }
+    else {
+        return (seq1 - seq2 > seq_number_diff) ? -1 : 1;
+    }
+}
+
 IPv4Address last_address_from_mask(IPv4Address addr, IPv4Address mask) {
     uint32_t addr_int = Endian::be_to_host<uint32_t>(addr),
              mask_int = Endian::be_to_host<uint32_t>(mask);

src/tcp_ip/ack_tracker.cpp

@@ -0,0 +1,174 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "tcp_ip/ack_tracker.h"
+
+#ifdef HAVE_ACK_TRACKER
+
+#include <limits>
+#include "tcp.h"
+#include "internals.h"
+
+using std::vector;
+using std::numeric_limits;
+
+using boost::icl::interval_bounds;
+using boost::icl::contains;
+
+using Tins::Internals::seq_compare;
+
+namespace Tins {
+namespace TCPIP {
+
+uint32_t interval_end(const AckedRange::interval_type& interval) {
+    if (interval.bounds() == interval_bounds::right_open()) {
+        return interval.upper() - 1;
+    }
+    else {
+        return interval.upper();
+    }
+}
+
+// AckedRange
+
+AckedRange::AckedRange(uint32_t first, uint32_t last) 
+: first_(first), last_(last) {
+
+}
+
+AckedRange::interval_type AckedRange::next() {
+    uint32_t interval_first = first_;
+    // Regular case
+    if (first_ <= last_) {
+        first_ = last_ + 1;
+        return interval_type::closed(interval_first, last_);
+    }
+    else {
+        // Range wraps around 
+        first_ = 0;
+        return interval_type::closed(interval_first, numeric_limits<uint32_t>::max());
+    }
+}
+
+bool AckedRange::has_next() const {
+    return seq_compare(first_, last_) <= 0;
+}
+
+uint32_t AckedRange::first() const {
+    return first_;
+}
+
+uint32_t AckedRange::last() const {
+    return last_;
+}
+
+// AckTracker
+
+AckTracker::AckTracker()
+: ack_number_(0), use_sack_(false) {
+    
+}
+
+AckTracker::AckTracker(uint32_t initial_ack, bool use_sack)
+: ack_number_(initial_ack), use_sack_(use_sack) {
+
+}
+
+void AckTracker::process_packet(const PDU& packet) {
+    const TCP* tcp = packet.find_pdu<TCP>();
+    if (!tcp) {
+        return;
+    }
+    if (seq_compare(tcp->ack_seq(), ack_number_) > 0) {
+        cleanup_sacked_intervals(ack_number_, tcp->ack_seq());
+        ack_number_ = tcp->ack_seq();
+    }
+    if (use_sack_) {
+        const TCP::option* sack_option = tcp->search_option(TCP::SACK);
+        if (sack_option) {
+            TCP::sack_type sack = sack_option->to<TCP::sack_type>();
+            process_sack(sack);
+        }
+    }
+}
+
+void AckTracker::process_sack(const vector<uint32_t>& sack) {
+    for (size_t i = 1; i < sack.size(); i += 2) {
+        // Left edge must be lower than right edge
+        if (seq_compare(sack[i - 1], sack[i]) < 0) {
+            AckedRange range(sack[i - 1], sack[i] - 1);
+            // If this range starts after our current ack number
+            if (seq_compare(range.first(), ack_number_) > 0) {
+                while (range.has_next()) {
+                    acked_intervals_.insert(range.next());
+                }
+            }
+        }
+    }
+}
+
+void AckTracker::cleanup_sacked_intervals(uint32_t old_ack, uint32_t new_ack) {
+    AckedRange range(old_ack, new_ack);
+    while (range.has_next()) {
+        acked_intervals_.erase(range.next());
+    }
+}
+
+void AckTracker::use_sack() {
+    use_sack_ = true;
+}
+
+uint32_t AckTracker::ack_number() const {
+    return ack_number_;
+}
+
+const AckTracker::interval_set_type& AckTracker::acked_intervals() const {
+    return acked_intervals_;
+}
+
+bool AckTracker::is_segment_acked(uint32_t sequence_number, uint32_t length) const {
+    if (length == 0) {
+        return true;
+    }
+    AckedRange range(sequence_number, sequence_number + length - 1);
+    while (range.has_next()) {
+        AckedRange::interval_type interval = range.next();
+        const int comparison = seq_compare(interval_end(interval), ack_number_);
+        // Only check for SACKed intervals if the segment finishes after our ACK number
+        if (comparison >= 0 && !contains(acked_intervals_, interval)) {
+            return false;
+        }
+    }
+    return true;
+}
+
+} // TCPIP
+} // Tins
+
+#endif // HAVE_ACK_TRACKER

src/tcp_ip/flow.cpp

@@ -0,0 +1,339 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "tcp_ip/flow.h"
+
+#if TINS_IS_CXX11
+
+#include <limits>
+#include <algorithm>
+#include "memory.h"
+#include "ip_address.h"
+#include "ipv6_address.h"
+#include "tcp.h"
+#include "ip.h"
+#include "ipv6.h"
+#include "rawpdu.h"
+#include "internals.h"
+#include "exceptions.h"
+#include "memory_helpers.h"
+
+using std::make_pair;
+using std::bind;
+using std::pair;
+using std::runtime_error;
+using std::numeric_limits;
+using std::max;
+using std::swap;
+
+using Tins::Memory::OutputMemoryStream;
+using Tins::Memory::InputMemoryStream;
+using Tins::Internals::seq_compare;
+
+namespace Tins {
+namespace TCPIP {
+
+Flow::Flow(const IPv4Address& dest_address, uint16_t dest_port,
+           uint32_t sequence_number) 
+: seq_number_(sequence_number), dest_port_(dest_port) {
+    OutputMemoryStream output(dest_address_.data(), dest_address_.size());
+    output.write(dest_address);
+    flags_.is_v6 = false;
+    initialize();
+}
+
+Flow::Flow(const IPv6Address& dest_address, uint16_t dest_port,
+           uint32_t sequence_number) 
+: seq_number_(sequence_number), dest_port_(dest_port) {
+    OutputMemoryStream output(dest_address_.data(), dest_address_.size());
+    output.write(dest_address);
+    flags_.is_v6 = true;
+    initialize();
+}
+
+void Flow::initialize() {
+    total_buffered_bytes_ = 0;
+    state_ = UNKNOWN;
+    mss_ = -1;
+}
+
+void Flow::data_callback(const data_available_callback_type& callback) {
+    on_data_callback_ = callback;
+}
+
+void Flow::out_of_order_callback(const flow_packet_callback_type& callback) {
+    on_out_of_order_callback_ = callback;
+}
+
+void Flow::process_packet(PDU& pdu) {
+    TCP* tcp = pdu.find_pdu<TCP>();
+    RawPDU* raw = pdu.find_pdu<RawPDU>(); 
+    // Update the internal state first
+    if (tcp) {
+        update_state(*tcp);
+        #ifdef HAVE_ACK_TRACKER
+        if (flags_.ack_tracking) {
+            ack_tracker_.process_packet(*tcp);
+        }
+        #endif // HAVE_ACK_TRACKER
+    }
+    if (flags_.ignore_data_packets) {
+        return;
+    }
+    if (!tcp || !raw) {
+        return;
+    }
+    const uint32_t chunk_end = tcp->seq() + raw->payload_size();
+    // If the end of the chunk ends after our current sequence number, process it.
+    if (seq_compare(chunk_end, seq_number_) >= 0) {
+        bool added_some = false;
+        uint32_t seq = tcp->seq();
+        // If we're going to buffer this and we have a buffering callback, execute it
+        if (seq > seq_number_ && on_out_of_order_callback_) {
+            on_out_of_order_callback_(*this, seq, raw->payload());
+        }
+
+        // If it starts before our sequence number, slice it
+        if (seq_compare(seq, seq_number_) < 0) {
+            const uint32_t diff = seq_number_ - seq;
+            raw->payload().erase(
+                raw->payload().begin(),
+                raw->payload().begin() + diff
+            );
+            seq = seq_number_;
+        }
+        // Store this payload
+        store_payload(seq, move(raw->payload()));
+        // Keep looping while the fragments seq is lower or equal to our seq
+        buffered_payload_type::iterator iter = buffered_payload_.find(seq_number_);
+        while (iter != buffered_payload_.end() && seq_compare(iter->first, seq_number_) <= 0) {
+            // Does this fragment start before our sequence number?
+            if (seq_compare(iter->first, seq_number_) < 0) {
+                uint32_t fragment_end = iter->first + iter->second.size();
+                int comparison = seq_compare(fragment_end, seq_number_);
+                // Does it end after our sequence number? 
+                if (comparison > 0) {
+                    // Then slice it
+                    payload_type& payload = iter->second;
+                    // First update this counter
+                    total_buffered_bytes_ -= payload.size();
+                    payload.erase(
+                        payload.begin(),
+                        payload.begin() + (seq_number_ - iter->first)
+                    );
+                    store_payload(seq_number_, move(iter->second));
+                    iter = erase_iterator(iter);
+                }
+                else {
+                    // Otherwise, we've seen this part of the payload. Erase it.
+                    iter = erase_iterator(iter);
+                }
+            }
+            else {
+                // They're equal. Add this payload.
+                payload_.insert(
+                    payload_.end(),
+                    iter->second.begin(), 
+                    iter->second.end()
+                );
+                seq_number_ += iter->second.size();
+                iter = erase_iterator(iter);
+                added_some = true;
+            }
+        }
+        if (added_some) {
+            if (on_data_callback_) {
+                on_data_callback_(*this);
+            }
+        }
+    }
+    else if (on_out_of_order_callback_) {
+        on_out_of_order_callback_(*this, tcp->seq(), raw->payload());
+    }
+}
+
+void Flow::store_payload(uint32_t seq, payload_type payload) {
+    buffered_payload_type::iterator iter = buffered_payload_.find(seq);
+    // New segment, store it
+    if (iter == buffered_payload_.end()) {
+        total_buffered_bytes_ += payload.size();
+        buffered_payload_.insert(make_pair(seq, move(payload)));
+    }
+    else if (iter->second.size() < payload.size()) {
+        // Increment by the diff between sizes
+        total_buffered_bytes_ += (payload.size() - iter->second.size());
+        // If we already have payload on this position but it's a shorter
+        // chunk than the new one, replace it
+        iter->second = move(payload);
+    }
+}
+
+Flow::buffered_payload_type::iterator Flow::erase_iterator(buffered_payload_type::iterator iter) {
+    buffered_payload_type::iterator output = iter;
+    total_buffered_bytes_ -= iter->second.size();
+    ++output;
+    buffered_payload_.erase(iter);
+    if (output == buffered_payload_.end()) {
+        output = buffered_payload_.begin();
+    }
+    return output;
+}
+
+void Flow::update_state(const TCP& tcp) {
+    if ((tcp.flags() & TCP::FIN) != 0) {
+        state_ = FIN_SENT;
+    }
+    else if ((tcp.flags() & TCP::RST) != 0) {
+        state_ = RST_SENT;
+    }
+    else if (state_ == SYN_SENT && (tcp.flags() & TCP::ACK) != 0) {
+        state_ = ESTABLISHED;
+        seq_number_++;
+    }
+    else if (state_ == UNKNOWN && (tcp.flags() & TCP::SYN) != 0) {
+        state_ = SYN_SENT;
+        seq_number_ = tcp.seq();
+        const TCP::option* mss_option = tcp.search_option(TCP::MSS);
+        if (mss_option) {
+            mss_ = mss_option->to<uint16_t>();
+        }
+        flags_.sack_permitted = tcp.has_sack_permitted();
+    }
+}
+
+bool Flow::is_v6() const {
+    return flags_.is_v6;
+}
+
+bool Flow::is_finished() const {
+    return state_ == FIN_SENT || state_ == RST_SENT;
+}
+
+bool Flow::packet_belongs(const PDU& packet) const {
+    if (is_v6()) {
+        const IPv6* ip = packet.find_pdu<IPv6>();
+        if (!ip || ip->dst_addr() != dst_addr_v6()) {
+            return false;
+        }
+    }
+    else {
+        const IP* ip = packet.find_pdu<IP>();
+        if (!ip || ip->dst_addr() != dst_addr_v4()) {
+            return false;
+        }
+    }
+    const TCP* tcp = packet.find_pdu<TCP>();
+    return tcp && tcp->dport() == dport();
+}
+
+IPv4Address Flow::dst_addr_v4() const {
+    InputMemoryStream stream(dest_address_.data(), dest_address_.size());
+    return stream.read<IPv4Address>();
+}
+
+IPv6Address Flow::dst_addr_v6() const {
+    InputMemoryStream stream(dest_address_.data(), dest_address_.size());
+    return stream.read<IPv6Address>();
+}
+
+uint16_t Flow::dport() const {
+    return dest_port_;
+}
+
+const Flow::payload_type& Flow::payload() const {
+    return payload_;
+}
+
+Flow::State Flow::state() const {
+    return state_;
+}
+
+uint32_t Flow::sequence_number() const {
+    return seq_number_;
+}
+
+const Flow::buffered_payload_type& Flow::buffered_payload() const {
+    return buffered_payload_;
+}
+
+Flow::buffered_payload_type& Flow::buffered_payload() {
+    return buffered_payload_;
+}
+
+uint32_t Flow::total_buffered_bytes() const {
+    return total_buffered_bytes_;
+}
+
+Flow::payload_type& Flow::payload() {
+    return payload_;
+}
+
+void Flow::state(State new_state) {
+    state_ = new_state;
+}
+
+void Flow::ignore_data_packets() {
+    flags_.ignore_data_packets = true;
+}
+
+int Flow::mss() const {
+    return mss_;
+}
+
+bool Flow::sack_permitted() const {
+    return flags_.sack_permitted;
+}
+
+void Flow::enable_ack_tracking() {
+    #ifdef HAVE_ACK_TRACKER
+    flags_.ack_tracking = 1;
+    #else
+    throw feature_disabled();
+    #endif
+}
+
+bool Flow::ack_tracking_enabled() const {
+    return flags_.ack_tracking;
+}
+
+#ifdef HAVE_ACK_TRACKER
+const AckTracker& Flow::ack_tracker() const {
+    return ack_tracker_;
+}
+
+AckTracker& Flow::ack_tracker() {
+    return ack_tracker_;
+}
+
+#endif // HAVE_ACK_TRACKER
+
+} // TCPIP
+} // Tins
+
+#endif // TINS_IS_CXX11

src/tcp_ip/stream.cpp

@@ -0,0 +1,310 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "tcp_ip/stream.h"
+
+#if TINS_IS_CXX11
+
+#include <limits>
+#include <algorithm>
+#include "memory.h"
+#include "ip_address.h"
+#include "ipv6_address.h"
+#include "tcp.h"
+#include "ip.h"
+#include "ipv6.h"
+#include "ethernetII.h"
+#include "rawpdu.h"
+#include "exceptions.h"
+#include "memory_helpers.h"
+
+using std::make_pair;
+using std::bind;
+using std::pair;
+using std::runtime_error;
+using std::numeric_limits;
+using std::max;
+using std::swap;
+
+using Tins::Memory::OutputMemoryStream;
+using Tins::Memory::InputMemoryStream;
+
+namespace Tins {
+namespace TCPIP {
+
+Stream::Stream(PDU& packet, const timestamp_type& ts) 
+: client_flow_(extract_client_flow(packet)),
+  server_flow_(extract_server_flow(packet)), create_time_(ts), 
+  last_seen_(ts), auto_cleanup_client_(true), auto_cleanup_server_(true) {
+    // Update client flow state
+    client_flow().process_packet(packet);
+    const EthernetII* eth = packet.find_pdu<EthernetII>();
+    if (eth) {
+        client_hw_addr_ = eth->src_addr();
+        server_hw_addr_ = eth->dst_addr();
+    }
+}
+
+void Stream::process_packet(PDU& packet, const timestamp_type& ts) {
+    last_seen_ = ts;
+    if (client_flow_.packet_belongs(packet)) {
+        client_flow_.process_packet(packet);
+    }
+    else if (server_flow_.packet_belongs(packet)) {
+        server_flow_.process_packet(packet);
+    }
+    if (is_finished() && on_stream_closed_) {
+        on_stream_closed_(*this);
+    }
+}
+
+void Stream::process_packet(PDU& packet) {
+    return process_packet(packet, timestamp_type(0));
+}
+
+Flow& Stream::client_flow() {
+    return client_flow_;
+}
+
+const Flow& Stream::client_flow() const {
+    return client_flow_;
+}
+
+Flow& Stream::server_flow() {
+    return server_flow_;
+}
+
+const Flow& Stream::server_flow() const {
+    return server_flow_;
+}
+
+void Stream::stream_closed_callback(const stream_callback_type& callback) {
+    on_stream_closed_ = callback;
+}
+
+void Stream::client_data_callback(const stream_callback_type& callback) {
+    on_client_data_callback_ = callback;
+}
+
+void Stream::server_data_callback(const stream_callback_type& callback) {
+    on_server_data_callback_ = callback;
+}
+
+void Stream::client_out_of_order_callback(const stream_packet_callback_type& callback) {
+    on_client_out_of_order_callback_ = callback;
+}
+
+void Stream::server_out_of_order_callback(const stream_packet_callback_type& callback) {
+    on_server_out_of_order_callback_ = callback;
+}
+
+void Stream::ignore_client_data() {
+    client_flow().ignore_data_packets();
+}
+
+void Stream::ignore_server_data() {
+    server_flow().ignore_data_packets();
+}
+
+bool Stream::is_finished() const {
+    const Flow::State client_state = client_flow_.state();
+    const Flow::State server_state = server_flow_.state();
+    // If either peer sent a RST then the stream is done
+    if (client_state == Flow::RST_SENT || server_state == Flow::RST_SENT) {
+        return true;
+    }
+    else {
+        // Otherwise, only finish if both sent a FIN
+        return client_state == Flow::FIN_SENT && server_state == Flow::FIN_SENT;
+    }
+}
+
+bool Stream::is_v6() const {
+    return server_flow().is_v6();
+}
+
+IPv4Address Stream::client_addr_v4() const {
+    return server_flow().dst_addr_v4();
+}
+
+IPv6Address Stream::client_addr_v6() const {
+    return server_flow().dst_addr_v6();
+}
+
+const Stream::hwaddress_type& Stream::client_hw_addr() const {
+    return client_hw_addr_;
+}
+
+const Stream::hwaddress_type& Stream::server_hw_addr() const {
+    return server_hw_addr_;
+}
+
+IPv4Address Stream::server_addr_v4() const {
+    return client_flow().dst_addr_v4();
+}
+
+IPv6Address Stream::server_addr_v6() const {
+    return client_flow().dst_addr_v6();
+}
+
+uint16_t Stream::client_port() const {
+    return server_flow().dport();
+}
+
+uint16_t Stream::server_port() const {
+    return client_flow().dport();
+}
+
+const Stream::payload_type& Stream::client_payload() const {
+    return client_flow().payload();
+}
+
+Stream::payload_type& Stream::client_payload() {
+    return client_flow().payload();
+}
+
+const Stream::payload_type& Stream::server_payload() const {
+    return server_flow().payload();
+}
+
+Stream::payload_type& Stream::server_payload() {
+    return server_flow().payload();
+}
+
+const Stream::timestamp_type& Stream::create_time() const {
+    return create_time_;
+}
+
+const Stream::timestamp_type& Stream::last_seen() const {
+    return last_seen_;
+}
+
+Flow Stream::extract_client_flow(const PDU& packet) {
+    const TCP* tcp = packet.find_pdu<TCP>();
+    if (!tcp) {
+        throw invalid_packet();
+    }
+    if (const IP* ip = packet.find_pdu<IP>()) {
+        return Flow(ip->dst_addr(), tcp->dport(), tcp->seq());
+    }
+    else if (const IPv6* ip = packet.find_pdu<IPv6>()) {
+        return Flow(ip->dst_addr(), tcp->dport(), tcp->seq());
+    }
+    else {
+        throw invalid_packet();
+    }
+}
+
+Flow Stream::extract_server_flow(const PDU& packet) {
+    const TCP* tcp = packet.find_pdu<TCP>();
+    if (!tcp) {
+        throw invalid_packet();
+    }
+    if (const IP* ip = packet.find_pdu<IP>()) {
+        return Flow(ip->src_addr(), tcp->sport(), tcp->ack_seq());
+    }
+    else if (const IPv6* ip = packet.find_pdu<IPv6>()) {
+        return Flow(ip->src_addr(), tcp->sport(), tcp->ack_seq());
+    }
+    else {
+        throw invalid_packet();
+    }
+}
+
+void Stream::setup_flows_callbacks() {
+    using namespace std::placeholders;
+
+    client_flow_.data_callback(bind(&Stream::on_client_flow_data, this, _1));
+    server_flow_.data_callback(bind(&Stream::on_server_flow_data, this, _1));
+    client_flow_.out_of_order_callback(bind(&Stream::on_client_out_of_order,
+                                            this, _1, _2, _3));
+    server_flow_.out_of_order_callback(bind(&Stream::on_server_out_of_order,
+                                            this, _1, _2, _3));
+}
+
+void Stream::auto_cleanup_payloads(bool value) {
+    auto_cleanup_client_data(value);
+    auto_cleanup_server_data(value);
+}
+
+void Stream::auto_cleanup_client_data(bool value) {
+    auto_cleanup_client_ = value;
+}
+
+void Stream::auto_cleanup_server_data(bool value) {
+    auto_cleanup_client_ = value;
+}
+
+void Stream::enable_ack_tracking() {
+    client_flow().enable_ack_tracking();
+    server_flow().enable_ack_tracking();
+}
+
+bool Stream::ack_tracking_enabled() const {
+    return client_flow().ack_tracking_enabled() && server_flow().ack_tracking_enabled();
+}
+
+void Stream::on_client_flow_data(const Flow& /*flow*/) {
+    if (on_client_data_callback_) {
+        on_client_data_callback_(*this);
+    }
+    if (auto_cleanup_client_) {
+        client_payload().clear();
+    }
+}
+
+void Stream::on_server_flow_data(const Flow& /*flow*/) {
+    if (on_server_data_callback_) {
+        on_server_data_callback_(*this);
+    }
+    if (auto_cleanup_server_) {
+        server_payload().clear();
+    }
+}
+
+void Stream::on_client_out_of_order(const Flow& flow,
+                                 uint32_t seq,
+                                 const payload_type& payload) {
+    if (on_client_out_of_order_callback_) {
+        on_client_out_of_order_callback_(*this, seq, payload);
+    }
+}
+
+void Stream::on_server_out_of_order(const Flow& flow,
+                                 uint32_t seq,
+                                 const payload_type& payload) {
+    if (on_server_out_of_order_callback_) {
+        on_server_out_of_order_callback_(*this, seq, payload);
+    }
+}
+
+} // TCPIP
+} // Tins
+
+#endif // TINS_IS_CXX11

src/tcp_ip/stream_follower.cpp

@@ -0,0 +1,269 @@
+/*
+ * Copyright (c) 2016, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ * 
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "tcp_ip/stream_follower.h"
+
+#if TINS_IS_CXX11
+
+#include <limits>
+#include <algorithm>
+#include "memory.h"
+#include "ip_address.h"
+#include "ipv6_address.h"
+#include "tcp.h"
+#include "ip.h"
+#include "ipv6.h"
+#include "rawpdu.h"
+#include "packet.h"
+#include "exceptions.h"
+#include "memory_helpers.h"
+
+using std::make_pair;
+using std::bind;
+using std::pair;
+using std::runtime_error;
+using std::numeric_limits;
+using std::max;
+using std::swap;
+using std::chrono::system_clock;
+using std::chrono::minutes;
+using std::chrono::duration_cast;
+
+using Tins::Memory::OutputMemoryStream;
+using Tins::Memory::InputMemoryStream;
+
+namespace Tins {
+namespace TCPIP {
+
+const size_t StreamFollower::DEFAULT_MAX_BUFFERED_CHUNKS = 512;
+const size_t StreamFollower::DEFAULT_MAX_SACKED_INTERVALS = 1024;
+const uint32_t StreamFollower::DEFAULT_MAX_BUFFERED_BYTES = 3 * 1024 * 1024; // 3MB
+const StreamFollower::timestamp_type StreamFollower::DEFAULT_KEEP_ALIVE = minutes(5);
+
+StreamFollower::StreamFollower() 
+: max_buffered_chunks_(DEFAULT_MAX_BUFFERED_CHUNKS),
+  max_buffered_bytes_(DEFAULT_MAX_BUFFERED_BYTES), last_cleanup_(0),
+  stream_keep_alive_(DEFAULT_KEEP_ALIVE), attach_to_flows_(false) {
+
+}
+
+void StreamFollower::process_packet(PDU& packet) {
+    // Use current time
+    const system_clock::duration ts = system_clock::now().time_since_epoch();
+    process_packet(packet, duration_cast<timestamp_type>(ts));
+}
+
+void StreamFollower::process_packet(Packet& packet) {
+    process_packet(*packet.pdu(), packet.timestamp());
+}
+
+void StreamFollower::process_packet(PDU& packet, const timestamp_type& ts) {
+    const TCP* tcp = packet.find_pdu<TCP>();
+    if (!tcp) {
+        return;
+    }
+    stream_id identifier = make_stream_id(packet);
+    streams_type::iterator iter = streams_.find(identifier);
+    bool process = true;
+    if (iter == streams_.end()) {
+        // Start tracking if they're either SYNs or they contain data (attach
+        // to an already running flow).
+        if (tcp->flags() == TCP::SYN || (attach_to_flows_ && tcp->find_pdu<RawPDU>() != 0)) {
+            iter = streams_.insert(make_pair(identifier, Stream(packet, ts))).first;
+            iter->second.setup_flows_callbacks();
+            if (on_new_connection_) {
+                on_new_connection_(iter->second);
+            }
+            else {
+                throw callback_not_set();
+            }
+            if (tcp->flags() == TCP::SYN) {
+                process = false;
+            }
+            else {
+                // Otherwise, assume the connection is established
+                iter->second.client_flow().state(Flow::ESTABLISHED);
+                iter->second.server_flow().state(Flow::ESTABLISHED);
+            }
+        }
+        else {
+            process = false;
+        }
+    }
+    // We'll process it if we had already seen this stream or if we just attached to
+    // it and it contains payload
+    if (process) {
+        Stream& stream = iter->second;
+        stream.process_packet(packet, ts);
+        // Check for different potential termination 
+        size_t total_chunks = stream.client_flow().buffered_payload().size() + 
+                              stream.server_flow().buffered_payload().size();
+        uint32_t total_buffered_bytes = stream.client_flow().total_buffered_bytes() +
+                                        stream.server_flow().total_buffered_bytes();
+        bool terminate_stream = total_chunks > max_buffered_chunks_ || 
+                                total_buffered_bytes > max_buffered_bytes_;
+        TerminationReason reason = BUFFERED_DATA;
+        #ifdef HAVE_ACK_TRACKER
+        if (!terminate_stream) {
+            uint32_t count = 0;
+            count += stream.client_flow().ack_tracker().acked_intervals().iterative_size();
+            count += stream.server_flow().ack_tracker().acked_intervals().iterative_size();
+            terminate_stream = count > DEFAULT_MAX_SACKED_INTERVALS;
+            reason = SACKED_SEGMENTS;
+        }
+        #endif // HAVE_ACK_TRACKER
+        if (stream.is_finished() || terminate_stream) {
+            // If we're terminating the stream, execute the termination callback
+            if (terminate_stream && on_stream_termination_) {
+                on_stream_termination_(stream, reason);
+            }
+            streams_.erase(iter);
+        }
+    }
+    if (last_cleanup_ + stream_keep_alive_ <= ts) {
+        cleanup_streams(ts);
+    }
+}
+
+void StreamFollower::new_stream_callback(const stream_callback_type& callback) {
+    on_new_connection_ = callback;
+}
+
+void StreamFollower::stream_termination_callback(const stream_termination_callback_type& callback) {
+    on_stream_termination_ = callback;
+}
+
+Stream& StreamFollower::find_stream(const IPv4Address& client_addr, uint16_t client_port,
+                                    const IPv4Address& server_addr, uint16_t server_port) {
+    stream_id identifier(serialize(client_addr), client_port,
+                         serialize(server_addr), server_port);
+    return find_stream(identifier);
+}
+
+Stream& StreamFollower::find_stream(const IPv6Address& client_addr, uint16_t client_port,
+                                    const IPv6Address& server_addr, uint16_t server_port) {
+    stream_id identifier(serialize(client_addr), client_port,
+                         serialize(server_addr), server_port);
+    return find_stream(identifier);
+}
+
+StreamFollower::stream_id StreamFollower::make_stream_id(const PDU& packet) {
+    const TCP* tcp = packet.find_pdu<TCP>();
+    if (!tcp) {
+        throw invalid_packet();
+    }
+    if (const IP* ip = packet.find_pdu<IP>()) {
+        return stream_id(serialize(ip->src_addr()), tcp->sport(),
+                         serialize(ip->dst_addr()), tcp->dport());
+    }
+    else if (const IPv6* ip = packet.find_pdu<IPv6>()) {
+        return stream_id(serialize(ip->src_addr()), tcp->sport(),
+                         serialize(ip->dst_addr()), tcp->dport());
+    }
+    else {
+        throw invalid_packet();
+    }
+}
+
+Stream& StreamFollower::find_stream(const stream_id& id) {
+    streams_type::iterator iter = streams_.find(id);
+    if (iter == streams_.end()) {
+        throw stream_not_found();
+    }
+    else {
+        return iter->second;
+    }
+}
+
+StreamFollower::stream_id::address_type StreamFollower::serialize(IPv4Address address) {
+    stream_id::address_type addr;
+    OutputMemoryStream output(addr.data(), addr.size());
+    addr.fill(0);
+    output.write(address);
+    return addr; 
+}
+
+StreamFollower::stream_id::address_type StreamFollower::serialize(const IPv6Address& address) {
+    stream_id::address_type addr;
+    OutputMemoryStream output(addr.data(), addr.size());
+    addr.fill(0);
+    output.write(address);
+    return addr;
+}
+
+void StreamFollower::cleanup_streams(const timestamp_type& now) {
+    streams_type::iterator iter = streams_.begin();
+    while (iter != streams_.end()) {
+        if (iter->second.last_seen() + stream_keep_alive_ <= now) {
+            // If we have a termination callback, execute it
+            if (on_stream_termination_) {
+                on_stream_termination_(iter->second, TIMEOUT);
+            }
+            streams_.erase(iter++);
+        }
+        else {
+            ++iter;
+        }
+    }
+    last_cleanup_ = now;
+}
+
+// stream_id
+
+StreamFollower::stream_id::stream_id() 
+: min_address_port(0), max_address_port(0) {
+    min_address.fill(0);
+    max_address.fill(0);
+}
+
+StreamFollower::stream_id::stream_id(const address_type& client_addr,
+                                     uint16_t client_port,
+                                     const address_type& server_addr,
+                                     uint16_t server_port) 
+: min_address(client_addr), max_address(server_addr), min_address_port(client_port),
+  max_address_port(server_port) {
+    if (min_address > max_address) {
+        swap(min_address, max_address);
+        swap(min_address_port, max_address_port);
+    }
+    else if (min_address == max_address && min_address_port > max_address_port) {
+        // If the address is the same, just sort ports
+        swap(min_address_port, max_address_port);
+    }
+}
+
+bool StreamFollower::stream_id::operator<(const stream_id& rhs) const {
+    return tie(min_address, min_address_port, max_address, max_address_port) <
+        tie(rhs.min_address, rhs.min_address_port, rhs.max_address, rhs.max_address_port);
+}
+
+} // TCPIP
+} // Tins
+
+#endif // TINS_IS_CXX11

tests/src/CMakeLists.txt

@@ -83,6 +83,7 @@ ADD_CUSTOM_TARGET(
     SLLTest
     SNAPTest
     STPTest
+    TCPIPTest
     TCPTest
     TCPStreamTest
     UDPTest
@@ -128,6 +129,7 @@ ADD_EXECUTABLE(SLLTest EXCLUDE_FROM_ALL sll.cpp)
 ADD_EXECUTABLE(SNAPTest EXCLUDE_FROM_ALL snap.cpp)
 ADD_EXECUTABLE(STPTest EXCLUDE_FROM_ALL stp.cpp)
 ADD_EXECUTABLE(TCPTest EXCLUDE_FROM_ALL tcp.cpp)
+ADD_EXECUTABLE(TCPIPTest EXCLUDE_FROM_ALL tcp_ip.cpp)
 ADD_EXECUTABLE(TCPStreamTest EXCLUDE_FROM_ALL tcp_stream.cpp)
 ADD_EXECUTABLE(UDPTest EXCLUDE_FROM_ALL udp.cpp)
 ADD_EXECUTABLE(UtilsTest EXCLUDE_FROM_ALL utils.cpp)
@@ -208,6 +210,7 @@ ADD_TEST(SLL SLLTest)
 ADD_TEST(SNAP SNAPTest)
 ADD_TEST(STP STPTest)
 ADD_TEST(TCP TCPTest)
+ADD_TEST(TCPIP TCPIPTest)
 ADD_TEST(TCPStream TCPStreamTest)
 ADD_TEST(UDP UDPTest)
 ADD_TEST(Utils UtilsTest)

tests/src/tcp_ip.cpp

@@ -0,0 +1,693 @@
+#include "cxxstd.h"
+
+#if TINS_IS_CXX11
+
+#include <gtest/gtest.h>
+#include <iostream>
+#include <algorithm>
+#include <string>
+#include <limits>
+#include <cassert>
+#include "tcp_ip/stream_follower.h"
+#include "tcp.h"
+#include "ip.h"
+#include "ip_address.h"
+#include "ipv6_address.h"
+#include "exceptions.h"
+#include "ethernetII.h"
+#include "rawpdu.h"
+#include "packet.h"
+#include "utils.h"
+#include "config.h"
+#ifdef HAVE_ACK_TRACKER
+    #include "tcp_ip/ack_tracker.h"
+#endif // HAVE_ACK_TRACKER
+
+using namespace std;
+using namespace std::chrono;
+using namespace Tins;
+using namespace Tins::TCPIP;
+
+class FlowTest : public testing::Test {
+public:
+    struct order_element {
+        order_element(size_t payload_index, uint32_t payload_size) 
+        : payload_index(payload_index),payload_size(payload_size) {
+
+        }
+
+        size_t payload_index;
+        uint32_t payload_size;
+    };
+
+    static const size_t num_packets = 20;
+    static EthernetII packets[], overlapped_packets1[], 
+                      overlapped_packets2[], overlapped_packets3[],
+                      overlapped_packets4[], overlapped_packets5[];
+    static const string payload;
+    typedef vector<order_element> ordering_info_type;
+    
+    void cumulative_flow_data_handler(Flow& flow);
+    void on_new_stream(Stream& stream);
+    void cumulative_stream_client_data_handler(Stream& stream);
+    void cumulative_stream_server_data_handler(Stream& stream);
+    void out_of_order_handler(Flow& session, uint32_t seq, Flow::payload_type payload);
+    void run_test(uint32_t initial_seq, const ordering_info_type& chunks, 
+                  const string& payload);
+    void run_test(uint32_t initial_seq, const ordering_info_type& chunks);
+    void run_tests(const ordering_info_type& chunks, const string& payload);
+    void run_tests(const ordering_info_type& chunks);
+    ordering_info_type split_payload(const string& payload, uint32_t chunk_size);
+    string merge_chunks(const vector<Flow::payload_type>& chunks);
+    vector<EthernetII> chunks_to_packets(uint32_t initial_seq,
+                                         const ordering_info_type& chunks, 
+                                         const string& payload);
+    vector<EthernetII> three_way_handshake(uint32_t client_seq, uint32_t server_seq,
+                                           IPv4Address client_addr, uint16_t client_port,
+                                           IPv4Address server_addr, uint16_t server_port);
+    void set_endpoints(vector<EthernetII>& packets, IPv4Address src_addr,
+                       uint16_t src_port, IPv4Address dst_addr,
+                       uint16_t dst_port);
+    
+    vector<Flow::payload_type> flow_payload_chunks;
+    vector<pair<uint32_t, Flow::payload_type> > flow_out_of_order_chunks;
+    vector<Flow::payload_type> stream_client_payload_chunks;
+    vector<Flow::payload_type> stream_server_payload_chunks;
+};
+
+const string FlowTest::payload = "Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
+                                 "Sed at aliquam arcu. Sed at iaculis magna. Nam ut dolor "
+                                 "eget velit mattis posuere ut non dui. Aliquam faucibus "
+                                 "erat pretium ligula tincidunt eget tristique justo placerat. "
+                                 "Phasellus turpis tellus, ornare ultricies egestas vitae, "
+                                 "mollis sed neque. Sed et libero in nunc pharetra auctor ut "
+                                 "a eros. Mauris quis faucibus nibh. \nLorem ipsum dolor sit "
+                                 "amet, consectetur adipiscing elit. Sed at aliquam arcu. "
+                                 "Sed at iaculis magna. Nam ut dolor eget velit mattis "
+                                 "posuere ut non dui. Aliquam faucibus erat pretium ligula "
+                                 "tincidunt eget tristique justo placerat. Phasellus turpis "
+                                 "tellus, ornare ultricies egestas vitae, mollis sed neque. "
+                                 "Sed et libero in nunc pharetra auctor ut a eros. Mauris "
+                                 "quis faucibus nibh. \n\n\nCurabitur sem erat, bibendum "
+                                 "quis condimentum ut, imperdiet at est. Duis sagittis rhoncus "
+                                 "felis at ultricies. In libero urna, dignissim eu elementum "
+                                 "quis, consectetur a neque. Praesent leo sem, cursus sed lobortis "
+                                 "sit amet, ornare ac augue. Mauris tristique semper ipsum at "
+                                 "consequat. Sed fringilla dolor ut lacus sagittis quis ultricies "
+                                 "leo vulputate. Maecenas dignissim imperdiet justo. Cras libero "
+                                 "odio, vehicula et adipiscing quis, luctus vel ante. \nAliquam "
+                                 "imperdiet est quis nunc malesuada eget convallis tellus "
+                                 "ullamcorper. Vivamus ullamcorper eros sit amet odio sollicitudin "
+                                 "rutrum. Donec pellentesque faucibus nulla, ut fringilla risus "
+                                 "aliquam eget. Sed et ante mi. Morbi a turpis et tellus dapibus "
+                                 "iaculis. Etiam faucibus tellus sed metus consequat rutrum. "
+                                 "Fusce sit amet nulla massa, tempus vulputate sem. Cras tincidunt "
+                                 "quam in libero rutrum interdum. Aliquam quam sapien, facilisis "
+                                 "at vestibulum et, venenatis id mauris. Morbi rutrum gravida "
+                                 "ultricies. \nAenean et justo ut libero euismod sollicitudin. "
+                                 "Nullam enim dui, iaculis vitae bibendum et, commodo in tellus. "
+                                 "Nullam eget purus mi, a ullamcorper lorem. Suspendisse potenti. "
+                                 "Duis ac justo ut leo euismod gravida sit amet at lectus. Lorem "
+                                 "ipsum dolor sit amet, consectetur adipiscing elit. Maecenas sed "
+                                 "arcu vitae nisi sollicitudin gravida. Nulla facilisis nibh turpis. "
+                                 "Maecenas quis imperdiet arcu. Sed sit amet nulla urna, at "
+                                 "vestibulum mauris. Suspendisse quis elit dui. Class aptent taciti "
+                                 "sociosqu ad litora torquent per conubia nostra, per inceptos "
+                                 "himenaeos. \n";
+
+void FlowTest::cumulative_flow_data_handler(Flow& flow) {
+    flow_payload_chunks.push_back(flow.payload());
+    flow.payload().clear();
+}
+
+void FlowTest::on_new_stream(Stream& stream) {
+    using std::placeholders::_1;
+    stream.client_data_callback(bind(&FlowTest::cumulative_stream_client_data_handler,
+                                     this, _1));
+}
+
+void FlowTest::cumulative_stream_client_data_handler(Stream& stream) {
+    stream_client_payload_chunks.push_back(stream.client_flow().payload());
+}
+
+void FlowTest::cumulative_stream_server_data_handler(Stream& stream) {
+    stream_server_payload_chunks.push_back(stream.server_flow().payload());
+}
+
+void FlowTest::out_of_order_handler(Flow& session, uint32_t seq, Flow::payload_type payload) {
+    flow_out_of_order_chunks.push_back(make_pair(seq, move(payload)));
+}
+
+void FlowTest::run_test(uint32_t initial_seq, const ordering_info_type& chunks, 
+                        const string& payload) {
+    using std::placeholders::_1;
+    flow_payload_chunks.clear();
+
+    Flow flow(IPv4Address("1.2.3.4"), 22, initial_seq);
+    flow.data_callback(bind(&FlowTest::cumulative_flow_data_handler, this, _1));
+    vector<EthernetII> packets = chunks_to_packets(initial_seq, chunks, payload);
+    for (size_t i = 0; i < packets.size(); ++i) {
+        flow.process_packet(packets[i]);
+    }
+    string flow_payload = merge_chunks(flow_payload_chunks);
+    EXPECT_EQ(payload, string(flow_payload.begin(), flow_payload.end()));
+    EXPECT_EQ(0, flow.total_buffered_bytes());
+    EXPECT_TRUE(flow.buffered_payload().empty());
+}
+
+void FlowTest::run_test(uint32_t initial_seq, const ordering_info_type& chunks) {
+    run_test(initial_seq, chunks, payload);
+}
+
+void FlowTest::run_tests(const ordering_info_type& chunks, const string& payload) {
+    run_test(0, chunks, payload);
+    run_test(20, chunks, payload);
+    run_test(numeric_limits<uint32_t>::max() / 2, chunks, payload);
+    run_test(numeric_limits<uint32_t>::max() - 2, chunks, payload);
+    run_test(numeric_limits<uint32_t>::max() - 5, chunks, payload);
+    run_test(numeric_limits<uint32_t>::max() - 10, chunks, payload);
+    run_test(numeric_limits<uint32_t>::max() - 34, chunks, payload);
+    run_test(numeric_limits<uint32_t>::max() - 31, chunks, payload);
+}
+
+void FlowTest::run_tests(const ordering_info_type& chunks) {
+    run_tests(chunks, payload);
+}
+
+FlowTest::ordering_info_type FlowTest::split_payload(const string& payload,
+                                                     uint32_t chunk_size) {
+    ordering_info_type output;
+    uint32_t chunk_count = payload.size() / chunk_size;
+    for (uint32_t i = 0; i < chunk_count; ++i) {
+        output.push_back(order_element(i * chunk_size, chunk_size));
+    }
+    if (chunk_count * chunk_size < payload.size()) {
+        uint32_t index = chunk_count * chunk_size;
+        output.push_back(order_element(index, payload.size() - index));
+    }
+    return output;
+}
+
+string FlowTest::merge_chunks(const vector<Flow::payload_type>& chunks) {
+    string output;
+    for (size_t i = 0; i < chunks.size(); ++i) {
+        Flow::payload_type this_chunk = chunks[i];
+        output += string(this_chunk.begin(), this_chunk.end());
+    }
+    return output;
+}
+
+vector<EthernetII> FlowTest::chunks_to_packets(uint32_t initial_seq,
+                                               const ordering_info_type& chunks, 
+                                               const string& payload) {
+    vector<EthernetII> output;
+    for (size_t i = 0; i < chunks.size(); ++i) {
+        const order_element& element = chunks[i];
+        assert(element.payload_index + element.payload_size <= payload.size());
+        TCP tcp;
+        RawPDU raw(payload.begin() + element.payload_index, 
+                   payload.begin() + element.payload_index + element.payload_size);
+        tcp.seq(initial_seq + element.payload_index);
+        output.push_back(EthernetII() / IP() / tcp / raw);
+    }
+    return output;
+}
+
+vector<EthernetII> FlowTest::three_way_handshake(uint32_t client_seq, uint32_t server_seq,
+                                                 IPv4Address client_addr, uint16_t client_port,
+                                                 IPv4Address server_addr, uint16_t server_port) {
+    vector<EthernetII> output;
+    output.push_back(EthernetII() / IP(server_addr, client_addr) / TCP(server_port, client_port));
+    output.push_back(EthernetII() / IP(client_addr, server_addr) / TCP(client_port, server_port));
+    output.push_back(EthernetII() / IP(server_addr, client_addr) / TCP(server_port, client_port));
+    output[0].rfind_pdu<TCP>().flags(TCP::SYN);
+    output[0].rfind_pdu<TCP>().seq(client_seq);
+    output[1].rfind_pdu<TCP>().flags(TCP::SYN | TCP::ACK);
+    output[1].rfind_pdu<TCP>().seq(server_seq);
+    output[1].rfind_pdu<TCP>().ack_seq(client_seq + 1);
+    output[2].rfind_pdu<TCP>().flags(TCP::ACK);
+    output[2].rfind_pdu<TCP>().seq(client_seq + 1);
+    output[2].rfind_pdu<TCP>().ack_seq(server_seq + 1);
+    return output;
+}
+
+void FlowTest::set_endpoints(vector<EthernetII>& packets, IPv4Address src_addr,
+                             uint16_t src_port, IPv4Address dst_addr,
+                             uint16_t dst_port) {
+    for (size_t i = 0; i < packets.size(); ++i) {
+        packets[i].rfind_pdu<IP>().src_addr(src_addr);
+        packets[i].rfind_pdu<IP>().dst_addr(dst_addr);
+        packets[i].rfind_pdu<TCP>().sport(src_port);
+        packets[i].rfind_pdu<TCP>().dport(dst_port);
+    }
+}
+
+TEST_F(FlowTest, ReassembleStreamPlain) {
+    ordering_info_type chunks = split_payload(payload, 5);
+    run_tests(chunks);
+}
+
+TEST_F(FlowTest, ReassembleStreamReordering) {
+    ordering_info_type chunks = split_payload(payload, 5);
+    // e.g. input [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+    // after this it's [2, 1, 0, 3, 6, 5, 4, 7, 10, 9, 8]
+    for (size_t i = 0; i < chunks.size(); i += 4) {
+        if (i + 2 < chunks.size()) {
+            swap(chunks[i], chunks[i + 2]);
+        }
+    }
+    run_tests(chunks);
+}
+
+TEST_F(FlowTest, ReassembleStreamReversed) {
+    ordering_info_type chunks = split_payload(payload, 5);
+    reverse(chunks.begin(), chunks.end());
+    run_tests(chunks);
+}
+
+TEST_F(FlowTest, Overlapping) {
+    string payload = "Hello world. This is a payload";
+    ordering_info_type chunks;
+    // "Hello "
+    chunks.push_back(order_element(0, 6));
+    // "ello Wo"
+    chunks.push_back(order_element(1, 7));
+    // "lo World"
+    chunks.push_back(order_element(3, 8));
+    chunks.push_back(order_element(10, payload.size() - 10));
+    chunks.push_back(order_element(9, 1));
+    run_tests(chunks, payload);
+
+    reverse(chunks.begin(), chunks.end());
+    run_tests(chunks, payload);
+
+    swap(chunks[2], chunks[4]);
+    run_tests(chunks, payload);
+}
+
+TEST_F(FlowTest, IgnoreDataPackets) {
+    using std::placeholders::_1;
+
+    ordering_info_type chunks = split_payload(payload, 5);
+    Flow flow(IPv4Address("1.2.3.4"), 22, 0);
+    flow.data_callback(bind(&FlowTest::cumulative_flow_data_handler, this, _1));
+    flow.ignore_data_packets();
+    vector<EthernetII> packets = chunks_to_packets(0, chunks, payload);
+    for (size_t i = 0; i < packets.size(); ++i) {
+        flow.process_packet(packets[i]);
+    }
+    EXPECT_TRUE(flow_payload_chunks.empty());
+}
+
+TEST_F(FlowTest, OutOfOrderCallback) {
+    using namespace std::placeholders;
+
+    ordering_info_type chunks = split_payload(payload, 5);
+    Flow flow(IPv4Address("1.2.3.4"), 22, 0);
+    flow.out_of_order_callback(bind(&FlowTest::out_of_order_handler, this, _1, _2, _3));
+    vector<EthernetII> packets = chunks_to_packets(0, chunks, payload);
+    reverse(packets.begin(), packets.end());
+    // Copy, as Flow::process_packet takes ownership of the payload
+    vector<EthernetII> original_packets = packets;
+    for (size_t i = 0; i < packets.size(); ++i) {
+        flow.process_packet(packets[i]);
+    }
+    // All elements should be out of order except the first
+    // one (last one in reverse order)
+    ASSERT_EQ(original_packets.size() - 1, flow_out_of_order_chunks.size());
+    for (size_t i = 0; i < original_packets.size() - 1; ++i) {
+        // Compare sequence number
+        EXPECT_EQ(
+            original_packets[i].rfind_pdu<TCP>().seq(),
+            flow_out_of_order_chunks[i].first
+        );
+        // Compare payload
+        EXPECT_EQ(
+            original_packets[i].rfind_pdu<RawPDU>().payload(),
+            flow_out_of_order_chunks[i].second
+        );
+    }
+}
+
+// Stream follower tests
+
+TEST_F(FlowTest, StreamFollower_ThreeWayHandshake) {
+    using std::placeholders::_1;
+
+    vector<EthernetII> packets = three_way_handshake(29, 60, "1.2.3.4", 22, "4.3.2.1", 25);
+    packets[0].src_addr("00:01:02:03:04:05");
+    packets[0].dst_addr("05:04:03:02:01:00");
+    StreamFollower follower;
+    follower.new_stream_callback(bind(&FlowTest::on_new_stream, this, _1));
+
+    Stream::timestamp_type ts(10000);
+    Stream::timestamp_type create_time = ts;
+    for (size_t i = 0; i < packets.size(); ++i) {
+        if (i != 0) {
+            ts += milliseconds(100);
+        }
+        Packet packet(packets[i], ts);
+        follower.process_packet(packet);
+    }
+    Stream& stream = follower.find_stream(IPv4Address("1.2.3.4"), 22,
+                                          IPv4Address("4.3.2.1"), 25);
+    EXPECT_EQ(Flow::ESTABLISHED, stream.client_flow().state());
+    EXPECT_EQ(Flow::SYN_SENT, stream.server_flow().state());
+    EXPECT_EQ(30, stream.client_flow().sequence_number());
+    EXPECT_EQ(60, stream.server_flow().sequence_number());
+    EXPECT_EQ(IPv4Address("4.3.2.1"), stream.client_flow().dst_addr_v4());
+    EXPECT_EQ(25, stream.client_flow().dport());
+    EXPECT_EQ(IPv4Address("1.2.3.4"), stream.server_flow().dst_addr_v4());
+    EXPECT_EQ(22, stream.server_flow().dport());
+    EXPECT_EQ(IPv4Address("1.2.3.4"), stream.client_addr_v4());
+    EXPECT_EQ(IPv4Address("4.3.2.1"), stream.server_addr_v4());
+    EXPECT_EQ(HWAddress<6>("00:01:02:03:04:05"), stream.client_hw_addr());
+    EXPECT_EQ(HWAddress<6>("05:04:03:02:01:00"), stream.server_hw_addr());
+    EXPECT_EQ(22, stream.client_port());
+    EXPECT_EQ(25, stream.server_port());
+    EXPECT_EQ(create_time, stream.create_time());
+    EXPECT_EQ(ts, stream.last_seen());
+
+    IP server_packet = IP("1.2.3.4", "4.3.2.1") / TCP(22, 25);
+    server_packet.rfind_pdu<TCP>().flags(TCP::ACK);
+    follower.process_packet(server_packet);
+
+    EXPECT_EQ(Flow::ESTABLISHED, stream.server_flow().state());
+    EXPECT_EQ(61, stream.server_flow().sequence_number());
+}
+
+TEST_F(FlowTest, StreamFollower_TCPOptions) {
+    using std::placeholders::_1;
+
+    vector<EthernetII> packets = three_way_handshake(29, 60, "1.2.3.4", 22, "4.3.2.1", 25);
+    // Client's mss is 1220
+    packets[0].rfind_pdu<TCP>().mss(1220);
+    // Server's mss is 1460
+    packets[1].rfind_pdu<TCP>().mss(1460);
+    // Server supports SACK
+    packets[1].rfind_pdu<TCP>().sack_permitted();
+    StreamFollower follower;
+    follower.new_stream_callback(bind(&FlowTest::on_new_stream, this, _1));
+    for (size_t i = 0; i < packets.size(); ++i) {
+        follower.process_packet(packets[i]);
+    }
+    Stream& stream = follower.find_stream(IPv4Address("1.2.3.4"), 22,
+                                          IPv4Address("4.3.2.1"), 25);
+    EXPECT_EQ(1220, stream.client_flow().mss());
+    EXPECT_EQ(1460, stream.server_flow().mss());
+    EXPECT_FALSE(stream.client_flow().sack_permitted());
+    EXPECT_TRUE(stream.server_flow().sack_permitted());
+}
+
+TEST_F(FlowTest, StreamFollower_CleanupWorks) {
+    using std::placeholders::_1;
+
+    bool timed_out = false;
+    vector<EthernetII> packets = three_way_handshake(29, 60, "1.2.3.4", 22, "4.3.2.1", 25);
+    StreamFollower follower;
+    follower.new_stream_callback(bind(&FlowTest::on_new_stream, this, _1));
+    follower.stream_termination_callback([&](Stream&, StreamFollower::TerminationReason reason) {
+        timed_out = (reason == StreamFollower::TIMEOUT);
+    });
+    packets[2].rfind_pdu<IP>().src_addr("6.6.6.6");
+    auto base_time = duration_cast<Stream::timestamp_type>(system_clock::now().time_since_epoch());
+    Packet packet1(packets[0], base_time);   
+    Packet packet2(packets[1], base_time + seconds(50));   
+    Packet packet3(packets[2], base_time + minutes(10));   
+    follower.process_packet(packet1);
+    Stream& stream = follower.find_stream(IPv4Address("1.2.3.4"), 22,
+                                          IPv4Address("4.3.2.1"), 25);
+    EXPECT_EQ(base_time, stream.create_time());
+    follower.process_packet(packet2);
+    follower.process_packet(packet3);
+    // At this point, it should be cleaned up
+    EXPECT_THROW(
+        follower.find_stream(IPv4Address("1.2.3.4"), 22, IPv4Address("4.3.2.1"), 25), 
+        stream_not_found
+    );
+    EXPECT_TRUE(timed_out);
+}
+
+TEST_F(FlowTest, StreamFollower_RSTClosesStream) {
+    using std::placeholders::_1;
+
+    vector<EthernetII> packets = three_way_handshake(29, 60, "1.2.3.4", 22, "4.3.2.1", 25);
+    StreamFollower follower;
+    follower.new_stream_callback(bind(&FlowTest::on_new_stream, this, _1));
+    for (size_t i = 0; i < packets.size(); ++i) {
+        follower.process_packet(packets[i]);
+    }
+    Stream stream = follower.find_stream(IPv4Address("1.2.3.4"), 22,
+                                         IPv4Address("4.3.2.1"), 25);
+
+    IP server_packet = IP("1.2.3.4", "4.3.2.1") / TCP(22, 25);
+    server_packet.rfind_pdu<TCP>().flags(TCP::RST);
+    stream.process_packet(server_packet);
+
+    EXPECT_EQ(Flow::RST_SENT, stream.server_flow().state());
+    EXPECT_TRUE(stream.is_finished());
+}
+
+TEST_F(FlowTest, StreamFollower_FINClosesStream) {
+    using std::placeholders::_1;
+
+    vector<EthernetII> packets = three_way_handshake(29, 60, "1.2.3.4", 22, "4.3.2.1", 25);
+    StreamFollower follower;
+    follower.new_stream_callback(bind(&FlowTest::on_new_stream, this, _1));
+    for (size_t i = 0; i < packets.size(); ++i) {
+        follower.process_packet(packets[i]);
+    }
+    Stream stream = follower.find_stream(IPv4Address("1.2.3.4"), 22,
+                                         IPv4Address("4.3.2.1"), 25);
+
+    IP server_packet = IP("1.2.3.4", "4.3.2.1") / TCP(22, 25);
+    server_packet.rfind_pdu<TCP>().flags(TCP::FIN | TCP::ACK);
+    stream.process_packet(server_packet);
+
+    EXPECT_EQ(Flow::FIN_SENT, stream.server_flow().state());
+    EXPECT_FALSE(stream.is_finished());
+
+    IP client_packet = IP("4.3.2.1", "1.2.3.4") / TCP(25, 22);
+    client_packet.rfind_pdu<TCP>().flags(TCP::FIN | TCP::ACK);
+    stream.process_packet(client_packet);
+
+    EXPECT_EQ(Flow::FIN_SENT, stream.client_flow().state());
+    EXPECT_TRUE(stream.is_finished());
+}
+
+TEST_F(FlowTest, StreamFollower_StreamIsRemovedWhenFinished) {
+    using std::placeholders::_1;
+
+    vector<EthernetII> packets = three_way_handshake(29, 60, "1.2.3.4", 22, "4.3.2.1", 25);
+    StreamFollower follower;
+    follower.new_stream_callback(bind(&FlowTest::on_new_stream, this, _1));
+    for (size_t i = 0; i < packets.size(); ++i) {
+        follower.process_packet(packets[i]);
+    }
+    IP server_packet = IP("1.2.3.4", "4.3.2.1") / TCP(22, 25);
+    server_packet.rfind_pdu<TCP>().flags(TCP::RST);
+    follower.process_packet(server_packet);
+
+    // We shouldn't be able to find it
+    EXPECT_THROW(
+        follower.find_stream(IPv4Address("1.2.3.4"), 22, IPv4Address("4.3.2.1"), 25), 
+        stream_not_found
+    );
+}
+
+TEST_F(FlowTest, StreamFollower_FollowStream) {
+    using std::placeholders::_1;
+
+    vector<EthernetII> packets = three_way_handshake(29, 60, "1.2.3.4", 22, "4.3.2.1", 25);
+    ordering_info_type chunks = split_payload(payload, 5);
+    vector<EthernetII> chunk_packets = chunks_to_packets(30 /*initial_seq*/, chunks, payload);
+    set_endpoints(chunk_packets, "1.2.3.4", 22, "4.3.2.1", 25);
+    packets.insert(packets.end(), chunk_packets.begin(), chunk_packets.end());
+    StreamFollower follower;
+    follower.new_stream_callback(bind(&FlowTest::on_new_stream, this, _1));
+    for (size_t i = 0; i < packets.size(); ++i) {
+        follower.process_packet(packets[i]);
+    }
+    EXPECT_EQ(chunk_packets.size(), stream_client_payload_chunks.size());
+    EXPECT_EQ(payload, merge_chunks(stream_client_payload_chunks));
+}
+
+
+#ifdef HAVE_ACK_TRACKER
+
+using namespace boost;
+using namespace boost::icl;
+
+class AckTrackerTest : public testing::Test {
+public:
+    typedef AckedRange::interval_type interval_type;
+private:
+
+};
+
+vector<uint32_t> make_sack() {
+    return vector<uint32_t>();
+}
+
+template <typename... SackEdges>
+vector<uint32_t> make_sack(pair<uint32_t, uint32_t> head, SackEdges&&... tail) {
+    vector<uint32_t> output = make_sack(tail...);
+    output.push_back(head.first);
+    output.push_back(head.second);
+    return output;
+}
+
+TCP make_tcp_ack(uint32_t ack_number) {
+    TCP output;
+    output.ack_seq(ack_number);
+    return output;
+}
+
+template <typename... SackEdges>
+TCP make_tcp_ack(uint32_t ack_number, SackEdges&&... rest) {
+    TCP output = make_tcp_ack(ack_number);
+    vector<uint32_t> sack = make_sack(rest...);
+    output.sack(sack);
+    return output;
+}
+
+// This section compares ranges using 
+//
+// EXPECT_TRUE(r1 == r2)
+//
+// Since otherwise gtest fails to compile when trying to print an interval
+// to a std::ostream
+TEST_F(AckTrackerTest, AckedRange_1) {
+    AckedRange range(0, 100);
+    EXPECT_TRUE(range.has_next());
+    EXPECT_TRUE(interval_type::closed(0, 100) == range.next());
+    EXPECT_FALSE(range.has_next());
+}
+
+TEST_F(AckTrackerTest, AckedRange_2) {
+    AckedRange range(2, 3);
+    EXPECT_TRUE(range.has_next());
+    EXPECT_TRUE(interval_type::closed(2, 3) == range.next());
+    EXPECT_FALSE(range.has_next());
+}
+
+TEST_F(AckTrackerTest, AckedRange_3) {
+    AckedRange range(0, 0);
+    EXPECT_TRUE(range.has_next());
+    EXPECT_TRUE(interval_type::right_open(0, 1) == range.next());
+    EXPECT_FALSE(range.has_next());
+}
+
+TEST_F(AckTrackerTest, AckedRange_4) {
+    uint32_t maximum = numeric_limits<uint32_t>::max();
+    AckedRange range(maximum, maximum);
+    EXPECT_TRUE(range.has_next());
+    EXPECT_TRUE(interval_type::left_open(maximum - 1, maximum) == range.next());
+    EXPECT_FALSE(range.has_next());
+}
+
+TEST_F(AckTrackerTest, AckedRange_WrapAround) {
+    uint32_t first = numeric_limits<uint32_t>::max() - 5;
+    AckedRange range(first, 100);
+    EXPECT_TRUE(range.has_next());
+    EXPECT_TRUE(
+        interval_type::closed(first, numeric_limits<uint32_t>::max()) ==
+        range.next()
+    );
+    EXPECT_TRUE(range.has_next());
+    EXPECT_TRUE(interval_type::closed(0, 100) == range.next());
+    EXPECT_FALSE(range.has_next());
+}
+
+TEST_F(AckTrackerTest, AckingTcp1) {
+    AckTracker tracker(0, false);
+    EXPECT_EQ(0, tracker.ack_number());
+    tracker.process_packet(make_tcp_ack(100));
+    EXPECT_EQ(100, tracker.ack_number());
+    EXPECT_TRUE(tracker.is_segment_acked(0, 10));
+    EXPECT_TRUE(tracker.is_segment_acked(50, 10));
+    EXPECT_TRUE(tracker.is_segment_acked(99, 1));
+    EXPECT_FALSE(tracker.is_segment_acked(90, 20));
+    EXPECT_FALSE(tracker.is_segment_acked(99, 2));
+    tracker.process_packet(make_tcp_ack(50));
+    EXPECT_EQ(100, tracker.ack_number());
+    tracker.process_packet(make_tcp_ack(150));
+    EXPECT_EQ(150, tracker.ack_number());
+    tracker.process_packet(make_tcp_ack(200));
+    EXPECT_EQ(200, tracker.ack_number());
+}
+
+TEST_F(AckTrackerTest, AckingTcp2) {
+    uint32_t maximum = numeric_limits<uint32_t>::max();
+    AckTracker tracker(maximum - 10, false);
+    EXPECT_EQ(maximum - 10, tracker.ack_number());   
+    tracker.process_packet(make_tcp_ack(maximum - 3));
+    EXPECT_EQ(maximum - 3, tracker.ack_number());   
+    tracker.process_packet(make_tcp_ack(maximum));
+    EXPECT_EQ(maximum, tracker.ack_number());   
+    tracker.process_packet(make_tcp_ack(5));
+    EXPECT_EQ(5, tracker.ack_number());   
+}
+
+TEST_F(AckTrackerTest, AckingTcp3) {
+    uint32_t maximum = numeric_limits<uint32_t>::max();
+    AckTracker tracker(maximum - 10, false);
+    tracker.process_packet(make_tcp_ack(5));
+    EXPECT_EQ(5, tracker.ack_number());   
+}
+
+TEST_F(AckTrackerTest, AckingTcp_Sack1) {
+    AckTracker tracker(0, true);
+    tracker.process_packet(make_tcp_ack(0, make_pair(2, 5), make_pair(9, 11)));
+    EXPECT_EQ(3 + 2, tracker.acked_intervals().size());
+    EXPECT_TRUE(tracker.is_segment_acked(2, 3));
+    EXPECT_TRUE(tracker.is_segment_acked(9, 2));
+    EXPECT_FALSE(tracker.is_segment_acked(2, 9));
+
+    tracker.process_packet(make_tcp_ack(9));
+    EXPECT_EQ(1, tracker.acked_intervals().size());
+
+    tracker.process_packet(make_tcp_ack(15));
+    EXPECT_EQ(0, tracker.acked_intervals().size());
+}
+
+TEST_F(AckTrackerTest, AckingTcp_Sack2) {
+    uint32_t maximum = numeric_limits<uint32_t>::max();
+    AckTracker tracker(maximum - 10, true);
+    tracker.process_packet(make_tcp_ack(
+        maximum - 10,
+        make_pair(maximum - 3, maximum),
+        make_pair(0, 10)
+    ));
+    EXPECT_EQ(3 + 10, tracker.acked_intervals().size());
+    EXPECT_TRUE(tracker.is_segment_acked(maximum - 12, 2));
+    EXPECT_TRUE(tracker.is_segment_acked(maximum - 2, 1));
+    EXPECT_TRUE(tracker.is_segment_acked(2, 3));
+    EXPECT_FALSE(tracker.is_segment_acked(maximum - 10, 10));
+
+    tracker.process_packet(make_tcp_ack(maximum - 2));
+    EXPECT_EQ(1 + 10, tracker.acked_intervals().size());
+
+    tracker.process_packet(make_tcp_ack(5));
+    EXPECT_EQ(4, tracker.acked_intervals().size());
+
+    tracker.process_packet(make_tcp_ack(15));
+    EXPECT_EQ(0, tracker.acked_intervals().size());
+}
+
+TEST_F(AckTrackerTest, AckingTcp_Sack3) {
+    uint32_t maximum = numeric_limits<uint32_t>::max();
+    AckTracker tracker(maximum - 10, true);
+    tracker.process_packet(make_tcp_ack(
+        maximum - 10,
+        make_pair(maximum - 3, 5)
+    ));
+    EXPECT_EQ(9, tracker.acked_intervals().size());
+
+    tracker.process_packet(make_tcp_ack(maximum));
+    EXPECT_EQ(5, tracker.acked_intervals().size());
+}
+
+#endif // HAVE_ACK_TRACKER
+
+#endif // TINS_IS_CXX11