Code/Projects/SmartPlugDetector/src/main/java/edu/uci/iotproject/Conversation.java

   1 package edu.uci.iotproject;
   2
   3 import edu.uci.iotproject.util.PcapPacketUtils;
   4 import org.pcap4j.core.PcapPacket;
   5 import org.pcap4j.packet.IpV4Packet;
   6 import org.pcap4j.packet.TcpPacket;
   7
   8 import java.util.*;
   9
  10 /**
  11  * Models a (TCP) conversation/connection/session/flow (packet's belonging to the same session between a client and a
  12  * server).
  13  * Holds a list of {@link PcapPacket}s identified as pertaining to the flow. Note that this list is <em>not</em>
  14  * considered when determining equality of two {@code Conversation} instances in order to allow for a
  15  * {@code Conversation} to function as a key in data structures such as {@link java.util.Map} and {@link java.util.Set}.
  16  * See {@link #equals(Object)} for the definition of equality.
  17  *
  18  * @author Janus Varmarken {@literal <jvarmark@uci.edu>}
  19  * @author Rahmadi Trimananda {@literal <rtrimana@uci.edu>}
  20  */
  21 public class Conversation {
  22
  23     /* Begin instance properties */
  24     /**
  25      * The IP of the host that is considered the client (i.e. the host that initiates the conversation)
  26      * in this conversation.
  27      */
  28     private final String mClientIp;
  29
  30     /**
  31      * The port number used by the host that is considered the client in this conversation.
  32      */
  33     private final int mClientPort;
  34
  35     /**
  36      * The IP of the host that is considered the server (i.e. is the responder) in this conversation.
  37      */
  38     private final String mServerIp;
  39
  40     /**
  41      * The port number used by the server in this conversation.
  42      */
  43     private final int mServerPort;
  44
  45     /**
  46      * The list of packets (with payload) pertaining to this conversation.
  47      */
  48     private final List<PcapPacket> mPackets;
  49
  50     /**
  51      * Contains the sequence numbers used thus far by the host that is considered the <em>client</em> in this
  52      * {@code Conversation}.
  53      * Used for filtering out retransmissions.
  54      */
  55     private final Set<Integer> mSeqNumbersClient;
  56
  57     /**
  58      * Contains the sequence numbers used thus far by the host that is considered the <em>server</em> in this
  59      * {@code Conversation}.
  60      * Used for filtering out retransmissions.
  61      */
  62     private final Set<Integer> mSeqNumbersSrv;
  63
  64     /**
  65      * List of SYN packets pertaining to this conversation.
  66      */
  67     private final List<PcapPacket> mSynPackets;
  68
  69     /**
  70      * List of pairs FINs and their corresponding ACKs associated with this conversation.
  71      */
  72     private final List<FinAckPair> mFinPackets;
  73
  74     /**
  75      * List of RST packets associated with this conversation.
  76      */
  77     private final List<PcapPacket> mRstPackets;
  78     /* End instance properties */
  79
  80     /**
  81      * Factory method for creating a {@code Conversation} from a {@link PcapPacket}.
  82      * @param pcapPacket The {@code PcapPacket} that wraps a TCP segment for which a {@code Conversation} is to be initiated.
  83      * @param clientIsSrc If {@code true}, the source address and source port found in the IP datagram and TCP segment
  84      *                    wrapped in the {@code PcapPacket} are regarded as pertaining to the client, and the destination
  85      *                    address and destination port are regarded as pertaining to the server---and vice versa if set
  86      *                    to {@code false}.
  87      * @return A {@code Conversation} initiated with ip:port for client and server according to the direction of the packet.
  88      */
  89     public static Conversation fromPcapPacket(PcapPacket pcapPacket, boolean clientIsSrc) {
  90         IpV4Packet ipPacket = pcapPacket.get(IpV4Packet.class);
  91         TcpPacket tcpPacket = pcapPacket.get(TcpPacket.class);
  92         String clientIp = clientIsSrc ? ipPacket.getHeader().getSrcAddr().getHostAddress() :
  93                 ipPacket.getHeader().getDstAddr().getHostAddress();
  94         String srvIp = clientIsSrc ? ipPacket.getHeader().getDstAddr().getHostAddress() :
  95                 ipPacket.getHeader().getSrcAddr().getHostAddress();
  96         int clientPort = clientIsSrc ? tcpPacket.getHeader().getSrcPort().valueAsInt() :
  97                 tcpPacket.getHeader().getDstPort().valueAsInt();
  98         int srvPort = clientIsSrc ? tcpPacket.getHeader().getDstPort().valueAsInt() :
  99                 tcpPacket.getHeader().getSrcPort().valueAsInt();
 100         return new Conversation(clientIp, clientPort, srvIp, srvPort);
 101     }
 102
 103     /**
 104      * Constructs a new {@code Conversation}.
 105      * @param clientIp The IP of the host that is considered the client (i.e. the host that initiates the conversation)
 106      *                 in the conversation.
 107      * @param clientPort The port number used by the client for the conversation.
 108      * @param serverIp The IP of the host that is considered the server (i.e. is the responder) in the conversation.
 109      * @param serverPort The port number used by the server for the conversation.
 110      */
 111     public Conversation(String clientIp, int clientPort, String serverIp, int serverPort) {
 112         this.mClientIp = clientIp;
 113         this.mClientPort = clientPort;
 114         this.mServerIp = serverIp;
 115         this.mServerPort = serverPort;
 116         this.mPackets = new ArrayList<>();
 117         this.mSeqNumbersClient = new HashSet<>();
 118         this.mSeqNumbersSrv = new HashSet<>();
 119         this.mSynPackets = new ArrayList<>();
 120         this.mFinPackets = new ArrayList<>();
 121         this.mRstPackets = new ArrayList<>();
 122     }
 123
 124     /**
 125      * Add a packet to the list of packets associated with this conversation.
 126      * @param packet The packet that is to be added to (associated with) this conversation.
 127      * @param ignoreRetransmissions Boolean value indicating if retransmissions should be ignored.
 128      *                              If set to {@code true}, {@code packet} will <em>not</em> be added to the
 129      *                              internal list of packets pertaining to this {@code Conversation}
 130      *                              <em>iff</em> the sequence number of {@code packet} was already
 131      *                              seen in a previous packet.
 132      */
 133     public void addPacket(PcapPacket packet, boolean ignoreRetransmissions) {
 134         // Precondition: verify that packet does indeed pertain to conversation.
 135         onAddPrecondition(packet);
 136         if (ignoreRetransmissions && isRetransmission(packet)) {
 137             // Packet is a retransmission. Ignore it.
 138             return;
 139         }
 140         // Select direction-dependent set of sequence numbers seen so far and update it with sequence number of new packet.
 141         addSeqNumber(packet);
 142         // Finally add packet to list of packets pertaining to this conversation.
 143         mPackets.add(packet);
 144         // Preserve order of packets in list: sort according to timestamp.
 145         if (mPackets.size() > 1 &&
 146                 mPackets.get(mPackets.size()-1).getTimestamp().isBefore(mPackets.get(mPackets.size()-2).getTimestamp())) {
 147             Collections.sort(mPackets, (o1, o2) -> {
 148                 if (o1.getTimestamp().isBefore(o2.getTimestamp())) { return -1; }
 149                 else if (o2.getTimestamp().isBefore(o1.getTimestamp())) { return 1; }
 150                 else { return 0; }
 151             });
 152         }
 153     }
 154
 155     /**
 156      * Get a list of packets pertaining to this {@code Conversation}.
 157      * The returned list is a read-only list.
 158      * @return the list of packets pertaining to this {@code Conversation}.
 159      */
 160     public List<PcapPacket> getPackets() {
 161         // Return read-only view to prevent external code from manipulating internal state (preserve invariant).
 162         return Collections.unmodifiableList(mPackets);
 163     }
 164
 165     /**
 166      * Records a TCP SYN packet as pertaining to this conversation (adds it to the the internal list).
 167      * Attempts to add duplicate SYN packets will be ignored, and the caller is made aware of the attempt to add a
 168      * duplicate by the return value being {@code false}.
 169      *
 170      * @param synPacket A {@link PcapPacket} wrapping a TCP SYN packet.
 171      * @return {@code true} if the packet was successfully added to this {@code Conversation}, {@code false} otherwise.
 172      */
 173     public boolean addSynPacket(PcapPacket synPacket) {
 174         onAddPrecondition(synPacket);
 175         final IpV4Packet synPacketIpSection = synPacket.get(IpV4Packet.class);
 176         final TcpPacket synPacketTcpSection = synPacket.get(TcpPacket.class);
 177         if (synPacketTcpSection == null || !synPacketTcpSection.getHeader().getSyn()) {
 178             throw new IllegalArgumentException("Not a SYN packet.");
 179         }
 180         // We are only interested in recording one copy of the two SYN packets (one SYN packet in each direction), i.e.,
 181         // we want to discard retransmitted SYN packets.
 182         if (mSynPackets.size() >= 2) {
 183             return false;
 184         }
 185         // Check the set of recorded SYN packets to see if we have already recorded a SYN packet going in the same
 186         // direction as the packet given in the argument.
 187         boolean matchingPrevSyn = mSynPackets.stream().anyMatch(p -> {
 188             IpV4Packet pIp = p.get(IpV4Packet.class);
 189             TcpPacket pTcp = p.get(TcpPacket.class);
 190             boolean srcAddrMatch = synPacketIpSection.getHeader().getSrcAddr().getHostAddress().
 191                     equals(pIp.getHeader().getSrcAddr().getHostAddress());
 192             boolean dstAddrMatch = synPacketIpSection.getHeader().getDstAddr().getHostAddress().
 193                     equals(pIp.getHeader().getDstAddr().getHostAddress());
 194             boolean srcPortMatch = synPacketTcpSection.getHeader().getSrcPort().valueAsInt() ==
 195                     pTcp.getHeader().getSrcPort().valueAsInt();
 196             boolean dstPortMatch = synPacketTcpSection.getHeader().getDstPort().valueAsInt() ==
 197                     pTcp.getHeader().getDstPort().valueAsInt();
 198             return srcAddrMatch && dstAddrMatch && srcPortMatch && dstPortMatch;
 199         });
 200         if (matchingPrevSyn) {
 201             return false;
 202         }
 203         // Update direction-dependent set of sequence numbers and record/log packet.
 204         addSeqNumber(synPacket);
 205         return mSynPackets.add(synPacket);
 206
 207         /*
 208         mSynPackets.stream().anyMatch(p -> {
 209             IpV4Packet pIp = p.get(IpV4Packet.class);
 210             TcpPacket pTcp = p.get(TcpPacket.class);
 211             boolean srcAddrMatch = synPacketIpSection.getHeader().getSrcAddr().getHostAddress().
 212                     equals(pIp.getHeader().getSrcAddr().getHostAddress());
 213             boolean dstAddrMatch = synPacketIpSection.getHeader().getDstAddr().getHostAddress().
 214                     equals(pIp.getHeader().getDstAddr().getHostAddress());
 215             boolean srcPortMatch = synPacketTcpSection.getHeader().getSrcPort().valueAsInt() ==
 216                     pTcp.getHeader().getSrcPort().valueAsInt();
 217             boolean dstPortMatch = synPacketTcpSection.getHeader().getDstPort().value() ==
 218                     pTcp.getHeader().getDstPort().value();
 219
 220             boolean fourTupleMatch = srcAddrMatch && dstAddrMatch && srcPortMatch && dstPortMatch;
 221
 222             boolean seqNoMatch = synPacketTcpSection.getHeader().getSequenceNumber() ==
 223                     pTcp.getHeader().getSequenceNumber();
 224
 225             if (fourTupleMatch && !seqNoMatch) {
 226                 // If the four tuple that identifies the conversation matches, but the sequence number is different,
 227                 // it means that this SYN packet is, in fact, an attempt to establish a **new** connection, and hence
 228                 // the given packet is NOT part of this conversation, even though the ip:port combinations are (by
 229                 // chance) selected such that they match this conversation.
 230                 throw new IllegalArgumentException("Attempt to add SYN packet that belongs to a different conversation " +
 231                         "(which is identified by the same four tuple as this conversation)");
 232             }
 233             return fourTupleMatch && seqNoMatch;
 234         });
 235         */
 236     }
 237
 238     /**
 239      * Get a list of SYN packets pertaining to this {@code Conversation}.
 240      * The returned list is a read-only list.
 241      * @return the list of SYN packets pertaining to this {@code Conversation}.
 242      */
 243     public List<PcapPacket> getSynPackets() {
 244         return Collections.unmodifiableList(mSynPackets);
 245     }
 246
 247     /**
 248      * Adds a TCP FIN packet to the list of TCP FIN packets associated with this conversation.
 249      * @param finPacket The TCP FIN packet that is to be added to (associated with) this conversation.
 250      */
 251     public void addFinPacket(PcapPacket finPacket) {
 252         // Precondition: verify that packet does indeed pertain to conversation.
 253         onAddPrecondition(finPacket);
 254         // TODO: should call addSeqNumber here?
 255         addSeqNumber(finPacket);
 256         mFinPackets.add(new FinAckPair(finPacket));
 257     }
 258
 259     /**
 260      * Attempt to ACK any FIN packets held by this conversation.
 261      * @param ackPacket The ACK for a FIN previously added to this conversation.
 262      */
 263     public void attemptAcknowledgementOfFin(PcapPacket ackPacket) {
 264         // Precondition: verify that the packet pertains to this conversation.
 265         onAddPrecondition(ackPacket);
 266         // Mark unack'ed FIN(s) that this ACK matches as ACK'ed (there might be more than one in case of retransmissions..?)
 267         mFinPackets.replaceAll(finAckPair -> !finAckPair.isAcknowledged() && finAckPair.isCorrespondingAckPacket(ackPacket) ? new FinAckPair(finAckPair.getFinPacket(), ackPacket) : finAckPair);
 268     }
 269
 270     /**
 271      * Retrieves an unmodifiable view of the list of {@link FinAckPair}s associated with this {@code Conversation}.
 272      * @return an unmodifiable view of the list of {@link FinAckPair}s associated with this {@code Conversation}.
 273      */
 274     public List<FinAckPair> getFinAckPairs() {
 275         return Collections.unmodifiableList(mFinPackets);
 276     }
 277
 278     /**
 279      * Get if this {@code Conversation} is considered to have been gracefully shut down.
 280      * A {@code Conversation} has been gracefully shut down if it contains a FIN+ACK pair for both directions
 281      * (client to server, and server to client).
 282      * @return {@code true} if the connection has been gracefully shut down, false otherwise.
 283      */
 284     public boolean isGracefullyShutdown() {
 285         //  The conversation has been gracefully shut down if we have recorded a FIN from both the client and the server which have both been ack'ed.
 286         return mFinPackets.stream().anyMatch(finAckPair -> finAckPair.isAcknowledged() && PcapPacketUtils.isSource(finAckPair.getFinPacket(), mClientIp, mClientPort)) &&
 287                 mFinPackets.stream().anyMatch(finAckPair -> finAckPair.isAcknowledged() && PcapPacketUtils.isSource(finAckPair.getFinPacket(), mServerIp, mServerPort));
 288     }
 289
 290     /**
 291      * Add a TCP segment for which the RST flag is set to this {@code Conversation}.
 292      * @param packet A {@link PcapPacket} wrapping a TCP segment pertaining to this {@code Conversation} for which the
 293      *               RST flag is set.
 294      */
 295     public void addRstPacket(PcapPacket packet) {
 296         /*
 297          * TODO:
 298          * When now also keeping track of RST packets, should we also...?
 299          * 1) Prevent later packets from being added once a RST segment has been added?
 300          * 2) Extend 'isGracefullyShutdown()' to also consider RST segments, or add another method, 'isShutdown()' that
 301          *    both considers FIN/ACK (graceful) as well as RST (abrupt/"ungraceful") shutdown?
 302          * 3) Should it be impossible to associate more than one RST segment with each Conversation?
 303          */
 304         onAddPrecondition(packet);
 305         TcpPacket tcpPacket = packet.get(TcpPacket.class);
 306         if (tcpPacket == null || !tcpPacket.getHeader().getRst()) {
 307             throw new IllegalArgumentException("not a RST packet");
 308         }
 309         mRstPackets.add(packet);
 310     }
 311
 312     /**
 313      * Get the TCP segments pertaining to this {@code Conversation} for which it was detected that the RST flag is set.
 314      * @return the TCP segments pertaining to this {@code Conversation} for which it was detected that the RST flag is
 315      *         set.
 316      */
 317     public List<PcapPacket> getRstPackets() {
 318         return Collections.unmodifiableList(mRstPackets);
 319     }
 320
 321     // =========================================================================================================
 322     // We simply reuse equals and hashCode methods of String.class to be able to use this class as a key
 323     // in a Map.
 324
 325     /**
 326      * <em>Note:</em> currently, equality is determined based on pairwise equality of the elements of the four tuple
 327      * ({@link #mClientIp}, {@link #mClientPort}, {@link #mServerIp}, {@link #mServerPort}) for {@code this} and
 328      * {@code obj}.
 329      * @param obj The object to test for equality with {@code this}.
 330      * @return {@code true} if {@code obj} is considered equal to {@code this} based on the definition of equality given above.
 331      */
 332     @Override
 333     public boolean equals(Object obj) {
 334         return obj instanceof Conversation && this.toString().equals(obj.toString());
 335     }
 336
 337     @Override
 338     public int hashCode() {
 339         return toString().hashCode();
 340     }
 341     // =========================================================================================================
 342
 343     @Override
 344     public String toString() {
 345         return String.format("%s:%d %s:%d", mClientIp, mClientPort, mServerIp, mServerPort);
 346     }
 347
 348     /**
 349      * Invoke to verify that the precondition holds when a caller attempts to add a packet to this {@code Conversation}.
 350      * An {@link IllegalArgumentException} is thrown if the precondition is violated.
 351      * @param packet the packet to be added to this {@code Conversation}
 352      */
 353     private void onAddPrecondition(PcapPacket packet) {
 354         // Apply precondition to preserve class invariant: all packets in mPackets must match the 4 tuple that
 355         // defines the conversation.
 356         IpV4Packet ipPacket = Objects.requireNonNull(packet.get(IpV4Packet.class));
 357         // For now we only support TCP flows.
 358         TcpPacket tcpPacket = Objects.requireNonNull(packet.get(TcpPacket.class));
 359         String ipSrc = ipPacket.getHeader().getSrcAddr().getHostAddress();
 360         String ipDst = ipPacket.getHeader().getDstAddr().getHostAddress();
 361         int srcPort = tcpPacket.getHeader().getSrcPort().valueAsInt();
 362         int dstPort = tcpPacket.getHeader().getDstPort().valueAsInt();
 363         String clientIp, serverIp;
 364         int clientPort, serverPort;
 365         if (ipSrc.equals(mClientIp)) {
 366             clientIp = ipSrc;
 367             clientPort = srcPort;
 368             serverIp = ipDst;
 369             serverPort = dstPort;
 370         } else {
 371             clientIp = ipDst;
 372             clientPort = dstPort;
 373             serverIp = ipSrc;
 374             serverPort = srcPort;
 375         }
 376         if (!(clientIp.equals(mClientIp) && clientPort == mClientPort &&
 377                 serverIp.equals(mServerIp) && serverPort == mServerPort)) {
 378             throw new IllegalArgumentException(
 379                     String.format("Attempt to add packet that does not pertain to %s",
 380                             Conversation.class.getSimpleName()));
 381         }
 382     }
 383
 384     /**
 385      * <p>
 386      *      Determines if the TCP packet contained in {@code packet} is a retransmission of a previously seen (logged)
 387      *      packet.
 388      * </p>
 389      *
 390      * <b>
 391      *     TODO:
 392      *     the current implementation, which uses a set of previously seen sequence numbers, will consider a segment
 393      *     with a reused sequence number---occurring as a result of sequence number wrap around for a very long-lived
 394      *     connection---as a retransmission (and may therefore end up discarding it even though it is in fact NOT a
 395      *     retransmission). Ideas?
 396      * </b>
 397      *
 398      * @param packet The packet.
 399      * @return {@code true} if {@code packet} was determined to be a retransmission, {@code false} otherwise.
 400      */
 401     public boolean isRetransmission(PcapPacket packet) {
 402         // Extract sequence number.
 403         int seqNo = packet.get(TcpPacket.class).getHeader().getSequenceNumber();
 404         switch (getDirection(packet)) {
 405             case CLIENT_TO_SERVER:
 406                 return mSeqNumbersClient.contains(seqNo);
 407             case SERVER_TO_CLIENT:
 408                 return mSeqNumbersSrv.contains(seqNo);
 409             default:
 410                 throw new AssertionError(String.format("Unexpected value of enum '%s'",
 411                         Direction.class.getSimpleName()));
 412         }
 413     }
 414
 415     /**
 416      * Extracts the TCP sequence number from {@code packet} and adds it to the proper set of sequence numbers by
 417      * analyzing the direction of the packet.
 418      * @param packet A TCP packet (wrapped in a {@code PcapPacket}) that was added to this conversation and whose
 419      *               sequence number is to be recorded as seen.
 420      */
 421     private void addSeqNumber(PcapPacket packet) {
 422         // Note: below check is redundant if client code is correct as the call to check the precondition should already
 423         // have been made by the addXPacket method that invokes this method. As such, the call below may be removed in
 424         // favor of speed, but the improvement will be minor, hence the added safety may be worth it.
 425         onAddPrecondition(packet);
 426         // Extract sequence number.
 427         int seqNo = packet.get(TcpPacket.class).getHeader().getSequenceNumber();
 428         // Determine direction of packet and add packet's sequence number to corresponding set of sequence numbers.
 429         switch (getDirection(packet)) {
 430             case CLIENT_TO_SERVER:
 431                 // Client to server packet.
 432                 mSeqNumbersClient.add(seqNo);
 433                 break;
 434             case SERVER_TO_CLIENT:
 435                 // Server to client packet.
 436                 mSeqNumbersSrv.add(seqNo);
 437                 break;
 438             default:
 439                 throw new AssertionError(String.format("Unexpected value of enum '%s'",
 440                         Direction.class.getSimpleName()));
 441         }
 442     }
 443
 444     /**
 445      * Determine the direction of {@code packet}. An {@link IllegalArgumentException} is thrown if {@code packet} does
 446      * not pertain to this conversation.
 447      *
 448      * @param packet The packet whose direction is to be determined.
 449      * @return A {@link Direction} indicating the direction of the packet.
 450      */
 451     public Direction getDirection(PcapPacket packet) {
 452         IpV4Packet ipPacket = packet.get(IpV4Packet.class);
 453         String ipSrc = ipPacket.getHeader().getSrcAddr().getHostAddress();
 454         String ipDst = ipPacket.getHeader().getDstAddr().getHostAddress();
 455         // Determine direction of packet.
 456         if (ipSrc.equals(mClientIp) && ipDst.equals(mServerIp)) {
 457             // Client to server packet.
 458             return Direction.CLIENT_TO_SERVER;
 459         } else if (ipSrc.equals(mServerIp) && ipDst.equals(mClientIp)) {
 460             // Server to client packet.
 461             return Direction.SERVER_TO_CLIENT;
 462         } else {
 463             throw new IllegalArgumentException("getDirection: packet not related to " + getClass().getSimpleName());
 464         }
 465     }
 466
 467     /**
 468      * Utility enum for expressing the direction of a packet pertaining to this {@code Conversation}.
 469      */
 470     public enum Direction {
 471
 472         CLIENT_TO_SERVER {
 473             @Override
 474             public String toCompactString() {
 475                 return "C->S";
 476             }
 477         },
 478         SERVER_TO_CLIENT {
 479             @Override
 480             public String toCompactString() {
 481                 return "S->C";
 482             }
 483         };
 484
 485
 486         /**
 487          * Get a compact string representation of this {@code Direction}.
 488          * @return a compact string representation of this {@code Direction}.
 489          */
 490         abstract public String toCompactString();
 491
 492     }
 493
 494 }