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

   1 package edu.uci.iotproject;
   2
   3 import edu.uci.iotproject.comparison.ComparisonFunctions;
   4 import edu.uci.iotproject.comparison.CompleteMatchPatternComparisonResult;
   5 import edu.uci.iotproject.comparison.PatternComparisonTask;
   6 import edu.uci.iotproject.trafficreassembly.layer3.Conversation;
   7 import org.pcap4j.core.NotOpenException;
   8 import org.pcap4j.core.PcapHandle;
   9 import org.pcap4j.core.PcapNativeException;
  10 import org.pcap4j.core.PcapPacket;
  11 import org.pcap4j.packet.DnsPacket;
  12 import org.pcap4j.packet.IpV4Packet;
  13 import org.pcap4j.packet.TcpPacket;
  14
  15 import java.io.*;
  16 import java.net.UnknownHostException;
  17 import java.text.DateFormat;
  18 import java.text.SimpleDateFormat;
  19 import java.util.*;
  20 import java.util.concurrent.*;
  21
  22
  23 /**
  24  * <p>Provides functionality for searching for the presence of a {@link FlowPattern} in a PCAP trace.</p>
  25  *
  26  * <p>
  27  * The (entire) PCAP trace is traversed and parsed on one thread (specifically, the thread that calls
  28  * {@link #findFlowPattern()}). This thread builds a {@link DnsMap} using the DNS packets present in the trace and uses
  29  * that {@code DnsMap} to reassemble {@link Conversation}s that <em>potentially</em> match the provided
  30  * {@link FlowPattern} (in that one end/party of said conversations matches the hostname(s) specified by the given
  31  * {@code FlowPattern}).
  32  * These potential matches are then examined on background worker thread(s) to determine if they are indeed a (complete)
  33  * match of the provided {@code FlowPattern}.
  34  * </p>
  35  *
  36  * @author Janus Varmarken {@literal <jvarmark@uci.edu>}
  37  * @author Rahmadi Trimananda {@literal <rtrimana@uci.edu>}
  38  */
  39 public class FlowPatternFinder {
  40
  41     /* Begin class properties */
  42     /**
  43      * {@link ExecutorService} responsible for parallelizing pattern searches.
  44      * Declared as static to allow for reuse of threads across different instances of {@code FlowPatternFinder} and to
  45      * avoid the overhead of initializing a new thread pool for each {@code FlowPatternFinder} instance.
  46      */
  47     private static final ExecutorService EXECUTOR_SERVICE = Executors.newCachedThreadPool();
  48     /* End class properties */
  49
  50     /* Begin instance properties */
  51     /**
  52      * Holds a set of {@link Conversation}s that <em>potentially</em> match {@link #mPattern} since each individual
  53      * {@code Conversation} is communication with the hostname identified by {@code mPattern.getHostname()}.
  54      * Note that due to limitations of the {@link Set} interface (specifically, there is no {@code get(T t)} method),
  55      * we have to resort to a {@link Map} (in which keys map to themselves) to "mimic" a set with {@code get(T t)}
  56      * functionality.
  57      *
  58      * @see <a href="https://stackoverflow.com/questions/7283338/getting-an-element-from-a-set">this question on StackOverflow.com</a>
  59      */
  60     private final Map<Conversation, Conversation> mConversations;
  61
  62     /**
  63      * Holds a list of trigger times.
  64      */
  65     private final List<Long> mTriggerTimes;
  66     private static int triggerListCounter;
  67
  68     private final DnsMap mDnsMap;
  69     private final PcapHandle mPcap;
  70     private final FlowPattern mPattern;
  71     private final ConversationPair mConvPair;
  72     private final String FILE = "./devices/tplink_switch/datapoints.csv";
  73     //private final String REF_FILE = "./devices/dlink_switch/dlink-june-26-2018.timestamps";
  74     private final String REF_FILE = "./devices/tplink_switch/tplink-june-14-2018.timestamps";
  75     //private final String REF_FILE = "./devices/tplink_switch/tplink-feb-13-2018.timestamps";
  76     // Router time is in CET and we use PST for the trigger times
  77     // Difference is 7 hours x 3600 x 1000ms = 25,200,000ms
  78     private final long TIME_OFFSET = 25200000;
  79
  80     private final List<Future<CompleteMatchPatternComparisonResult>> mPendingComparisons = new ArrayList<>();
  81     /* End instance properties */
  82
  83     /**
  84      * Constructs a new {@code FlowPatternFinder}.
  85      * @param pcap an <em>open</em> {@link PcapHandle} that provides access to the trace that is to be examined.
  86      * @param pattern the {@link FlowPattern} to search for.
  87      */
  88     public FlowPatternFinder(PcapHandle pcap, FlowPattern pattern) {
  89         this.mConversations = new HashMap<>();
  90         this.mTriggerTimes = readTriggerTimes(REF_FILE);
  91         triggerListCounter = 0;
  92         this.mDnsMap = new DnsMap();
  93         this.mPcap = Objects.requireNonNull(pcap,
  94                 String.format("Argument of type '%s' cannot be null", PcapHandle.class.getSimpleName()));
  95         this.mPattern = Objects.requireNonNull(pattern,
  96                 String.format("Argument of type '%s' cannot be null", FlowPattern.class.getSimpleName()));
  97         this.mConvPair = new ConversationPair(FILE, ConversationPair.Direction.DEVICE_TO_SERVER);
  98     }
  99
 100
 101     private List<Long> readTriggerTimes(String refFileName) {
 102
 103         List<Long> listTriggerTimes = new ArrayList<>();
 104         try {
 105             File file = new File(refFileName);
 106             BufferedReader br = new BufferedReader(new FileReader(file));
 107             String s;
 108             while ((s = br.readLine()) != null) {
 109                 listTriggerTimes.add(timeToMillis(s, false));
 110             }
 111         } catch (IOException e) {
 112             e.printStackTrace();
 113         }
 114         System.out.println("List has: " + listTriggerTimes.size());
 115
 116         return listTriggerTimes;
 117     }
 118
 119     /**
 120      * Starts the pattern search.
 121      */
 122     public void start() {
 123
 124         //findFlowPattern();
 125         findSignatureBasedOnTimestamp();
 126     }
 127
 128     /**
 129      * Find patterns based on the FlowPattern object (run by a thread)
 130      */
 131     private void findFlowPattern() {
 132         try {
 133             PcapPacket packet;
 134 //            TODO: The new comparison method is pending
 135 //            TODO: For now, just compare using one hostname and one list per FlowPattern
 136 //            List<String> hostnameList = mPattern.getHostnameList();
 137 //            int hostIndex = 0;
 138             while ((packet = mPcap.getNextPacketEx()) != null) {
 139                 // Let DnsMap handle DNS packets.
 140                 if (packet.get(DnsPacket.class) != null) {
 141                     // Check if this is a valid DNS packet
 142                     mDnsMap.validateAndAddNewEntry(packet);
 143                     continue;
 144                 }
 145                 // For now, we only work support pattern search in TCP over IPv4.
 146                 final IpV4Packet ipPacket = packet.get(IpV4Packet.class);
 147                 final TcpPacket tcpPacket = packet.get(TcpPacket.class);
 148                 if (ipPacket == null || tcpPacket == null) {
 149                     continue;
 150                 }
 151
 152                 String srcAddress = ipPacket.getHeader().getSrcAddr().getHostAddress();
 153                 String dstAddress = ipPacket.getHeader().getDstAddr().getHostAddress();
 154                 int srcPort = tcpPacket.getHeader().getSrcPort().valueAsInt();
 155                 int dstPort = tcpPacket.getHeader().getDstPort().valueAsInt();
 156                 // Is this packet related to the pattern; i.e. is it going to (or coming from) the cloud server?
 157                 boolean fromServer = mDnsMap.isRelatedToCloudServer(srcAddress, mPattern.getHostname());
 158                 boolean fromClient = mDnsMap.isRelatedToCloudServer(dstAddress, mPattern.getHostname());
 159 //                String currentHostname = hostnameList.get(hostIndex);
 160 //                boolean fromServer = mDnsMap.isRelatedToCloudServer(srcAddress, currentHostname);
 161 //                boolean fromClient = mDnsMap.isRelatedToCloudServer(dstAddress, currentHostname);
 162                 if (!fromServer && !fromClient) {
 163                     // Packet not related to pattern, skip it.
 164                     continue;
 165                 }
 166
 167                 // Conversations (connections/sessions) are identified by the four-tuple
 168                 // (clientIp, clientPort, serverIp, serverPort) (see Conversation Javadoc).
 169                 // Create "dummy" conversation for looking up an existing entry.
 170                 Conversation conversation = fromClient ? new Conversation(srcAddress, srcPort, dstAddress, dstPort) :
 171                         new Conversation(dstAddress, dstPort, srcAddress, srcPort);
 172                 // Add the packet so that the "dummy" conversation can be immediately added to the map if no entry
 173                 // exists for the conversation that the current packet belongs to.
 174                 if (tcpPacket.getHeader().getFin()) {
 175                     // Record FIN packets.
 176                     conversation.addFinPacket(packet);
 177                 }
 178                 if (tcpPacket.getPayload() != null) {
 179                     // Record regular payload packets.
 180                     conversation.addPacket(packet, true);
 181                 }
 182                 // Note: does not make sense to call attemptAcknowledgementOfFin here as the new packet has no FINs
 183                 // in its list, so if this packet is an ACK, it would not be added anyway.
 184                 // Need to retain a final reference to get access to the packet in the lambda below.
 185                 final PcapPacket finalPacket = packet;
 186                 // Add the new conversation to the map if an equal entry is not already present.
 187                 // If an existing entry is already present, the current packet is simply added to that conversation.
 188                 mConversations.merge(conversation, conversation, (existingEntry, toMerge) -> {
 189                     // toMerge may not have any payload packets if the current packet is a FIN packet.
 190                     if (toMerge.getPackets().size() > 0) {
 191                         existingEntry.addPacket(toMerge.getPackets().get(0), true);
 192                     }
 193                     if (toMerge.getFinAckPairs().size() > 0) {
 194                         // Add the FIN packet to the existing entry.
 195                         existingEntry.addFinPacket(toMerge.getFinAckPairs().get(0).getFinPacket());
 196                     }
 197                     if (finalPacket.get(TcpPacket.class).getHeader().getAck()) {
 198                         existingEntry.attemptAcknowledgementOfFin(finalPacket);
 199                     }
 200                     return existingEntry;
 201                 });
 202                 // Refresh reference to point to entry in map (in case packet was added to existing entry).
 203                 conversation = mConversations.get(conversation);
 204                 if (conversation.isGracefullyShutdown()) {
 205                     // Conversation terminated gracefully, so we can now start analyzing it.
 206                     // Remove the Conversation from the map and start the analysis.
 207                     // Any future packets identified by the same four tuple will be tied to a new Conversation instance.
 208                     mConversations.remove(conversation);
 209                     // Create comparison task and send to executor service.
 210                     PatternComparisonTask<CompleteMatchPatternComparisonResult> comparisonTask =
 211                             new PatternComparisonTask<>(conversation, mPattern, ComparisonFunctions.SUB_SEQUENCE_COMPLETE_MATCH);
 212                     mPendingComparisons.add(EXECUTOR_SERVICE.submit(comparisonTask));
 213                     // Increment hostIndex to find the next
 214
 215                 }
 216             }
 217         } catch (EOFException eofe) {
 218             // TODO should check for leftover conversations in map here and fire tasks for those.
 219             // TODO [cont'd] such tasks may be present if connections did not terminate gracefully or if there are longlived connections.
 220             System.out.println("[ findFlowPattern ] Finished processing entire PCAP stream!");
 221             System.out.println("[ findFlowPattern ] Now waiting for comparisons to finish...");
 222             // Wait for all comparisons to finish, then output their results to std.out.
 223             for(Future<CompleteMatchPatternComparisonResult> comparisonTask : mPendingComparisons) {
 224                 try {
 225                     // Blocks until result is ready.
 226                     CompleteMatchPatternComparisonResult comparisonResult = comparisonTask.get();
 227                     if (comparisonResult.getResult()) {
 228                         System.out.println(comparisonResult.getTextualDescription());
 229                     }
 230                 } catch (InterruptedException|ExecutionException e) {
 231                     e.printStackTrace();
 232                 }
 233             }
 234         } catch (UnknownHostException |
 235                  PcapNativeException  |
 236                  NotOpenException     |
 237                  TimeoutException ex) {
 238             ex.printStackTrace();
 239         }
 240     }
 241
 242     /**
 243      * Find patterns based on the FlowPattern object (run by a thread)
 244      */
 245     private void findSignatureBasedOnTimestamp() {
 246         try {
 247             PcapPacket packet;
 248 //            TODO: The new comparison method is pending
 249 //            TODO: For now, just compare using one hostname and one list per FlowPattern
 250             while ((packet = mPcap.getNextPacketEx()) != null) {
 251                 // Let DnsMap handle DNS packets.
 252                 if (packet.get(DnsPacket.class) != null) {
 253                     // Check if this is a valid DNS packet
 254                     mDnsMap.validateAndAddNewEntry(packet);
 255                     continue;
 256                 }
 257                 // For now, we only work support pattern search in TCP over IPv4.
 258                 final IpV4Packet ipPacket = packet.get(IpV4Packet.class);
 259                 final TcpPacket tcpPacket = packet.get(TcpPacket.class);
 260                 if (ipPacket == null || tcpPacket == null) {
 261                     continue;
 262                 }
 263
 264                 String srcAddress = ipPacket.getHeader().getSrcAddr().getHostAddress();
 265                 String dstAddress = ipPacket.getHeader().getDstAddr().getHostAddress();
 266                 int srcPort = tcpPacket.getHeader().getSrcPort().valueAsInt();
 267                 int dstPort = tcpPacket.getHeader().getDstPort().valueAsInt();
 268                 //System.out.println("Timestamp packet: " + packet.getTimestamp());
 269                 // Is this packet related to the pattern; i.e. is it going to (or coming from) the cloud server?
 270                 boolean fromServer = mDnsMap.isRelatedToCloudServer(srcAddress, mPattern.getHostname());
 271                 boolean fromClient = mDnsMap.isRelatedToCloudServer(dstAddress, mPattern.getHostname());
 272                 if (!fromServer && !fromClient) {
 273                     // Packet not related to pattern, skip it.
 274                     continue;
 275                 }
 276                 // Record the conversation pairs
 277                 if (tcpPacket.getPayload() != null && checkTimeStamp(packet)) {
 278                 //if (tcpPacket.getPayload() != null) {
 279                     mConvPair.writeConversationPair(packet, fromClient, fromServer);
 280                 }
 281             }
 282         } catch (EOFException eofe) {
 283             triggerListCounter = 0;
 284             mConvPair.close();
 285             System.out.println("[ findFlowPattern ] ConversationPair writer closed!");
 286             System.out.println("[ findFlowPattern ] Frequencies of data points:");
 287             mConvPair.printListFrequency();
 288         } catch (UnknownHostException |
 289                 PcapNativeException  |
 290                 NotOpenException     |
 291                 TimeoutException ex) {
 292             ex.printStackTrace();
 293         }
 294     }
 295
 296     private boolean checkTimeStamp(PcapPacket packet) {
 297
 298         // Extract time from the packet's timestamp
 299         String timeStamp = packet.getTimestamp().toString();
 300         String timeString = timeStamp.substring(timeStamp.indexOf("T") + 1, timeStamp.indexOf("."));
 301         // Timestamps are in CET (ahead of PST) so it should be deducted by TIME_OFFSET
 302         long time = timeToMillis(timeString, true) - TIME_OFFSET;
 303         //long time = timeToMillis(timeString, true);
 304
 305         //System.out.println("Gets here: " + time + " trigger time: " + mTriggerTimes.get(triggerListCounter));
 306
 307         // We accept packets that are at most 3 seconds away from the trigger time
 308         if ((mTriggerTimes.get(triggerListCounter) <= time) &&
 309                 (time <= mTriggerTimes.get(triggerListCounter) + 3000)) {
 310             //System.out.println("Gets here 1: " + timeString + " index: " + triggerListCounter);
 311             return true;
 312         } else {
 313             // Handle the case that the timestamp is > 3000, but < next timestamp
 314             // in the list. We ignore these packets.
 315             if (time < mTriggerTimes.get(triggerListCounter)) {
 316                 // Timestamp is smaller than trigger, ignore!
 317                 //System.out.println("Gets here 2: " + timeString + " index: " + triggerListCounter);
 318                 return false;
 319             } else { // Timestamp is greater than trigger, increment!
 320                 triggerListCounter = triggerListCounter + 1;
 321                 //System.out.println("Gets here 3: " + timeString + " index: " + triggerListCounter);
 322                 //return false;
 323                 return checkTimeStamp(packet);
 324             }
 325         }
 326
 327         //System.out.println("Timestamp: " + timeToMillis(time, true));
 328         //String time2 = "21:38:08";
 329         //System.out.println("Timestamp: " + timeToMillis(time2, true));
 330     }
 331
 332     /**
 333      * A private function that returns time in milliseconds.
 334      * @param time The time in the form of String.
 335      * @param is24Hr If true, then this is in 24-hour format.
 336      */
 337     private long timeToMillis(String time, boolean is24Hr) {
 338
 339         String format = null;
 340         if (is24Hr) {
 341             format = "hh:mm:ss";
 342         } else { // 12 Hr format
 343             format = "hh:mm:ss aa";
 344         }
 345         DateFormat sdf = new SimpleDateFormat(format);
 346         Date date = null;
 347         try {
 348             date = sdf.parse(time);
 349         } catch(Exception e) {
 350             e.printStackTrace();
 351         }
 352         if (date == null)
 353             return 0;
 354         return date.getTime();
 355     }
 356
 357 }