Extract Conversation to separate file (i.e. it is no longer an inner class). Add...

[pingpong.git] / Code / Projects / SmartPlugDetector / src / main / java / edu / uci / iotproject / FlowPatternFinder.java

diff --git a/Code/Projects/SmartPlugDetector/src/main/java/edu/uci/iotproject/FlowPatternFinder.java b/Code/Projects/SmartPlugDetector/src/main/java/edu/uci/iotproject/FlowPatternFinder.java
index df4d44769668ade3c0d9170b49718a1140d58e0a..9bbd99ce985a329270ab12cdef38d01b68350240 100644 (file)
--- a/Code/Projects/SmartPlugDetector/src/main/java/edu/uci/iotproject/FlowPatternFinder.java
+++ b/Code/Projects/SmartPlugDetector/src/main/java/edu/uci/iotproject/FlowPatternFinder.java
@@ -3,86 +3,159 @@ package edu.uci.iotproject;
 import org.pcap4j.core.NotOpenException;
 import org.pcap4j.core.PcapHandle;
 import org.pcap4j.core.PcapNativeException;
 import org.pcap4j.core.NotOpenException;
 import org.pcap4j.core.PcapHandle;
 import org.pcap4j.core.PcapNativeException;

+import org.pcap4j.core.PcapPacket;

 import org.pcap4j.packet.IpV4Packet;
 import org.pcap4j.packet.Packet;
 import org.pcap4j.packet.TcpPacket;
 import org.pcap4j.packet.IpV4Packet;
 import org.pcap4j.packet.Packet;
 import org.pcap4j.packet.TcpPacket;

+import org.pcap4j.packet.DnsPacket;

 
 import java.io.EOFException;
 
 import java.io.EOFException;

+import java.net.UnknownHostException;
+import java.time.Instant;

 import java.util.*;
 import java.util.concurrent.TimeoutException;
 import java.util.*;
 import java.util.concurrent.TimeoutException;

+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;

 
 /**
  * Provides functionality for searching for the presence of a {@link FlowPattern} in a PCAP trace.
 
 /**
  * Provides functionality for searching for the presence of a {@link FlowPattern} in a PCAP trace.

+ * We use 2 threads:
+ *  1) The first thread (main thread) collects conversations from the PCAP stream and put them into our data structure.
+ *  2) The second thread (checker thread) checks the collected conversation.

  *
  * @author Janus Varmarken
  *
  * @author Janus Varmarken

+ * @author Rahmadi Trimananda

  */
 public class FlowPatternFinder {
 
  */
 public class FlowPatternFinder {
 

-    private final Map<String, Set<String>> dnsMap;
-    private final Map<Conversation, List<Packet>> connections = new HashMap<>();
+    /* Class properties */
+    private Map<Conversation, List<PcapPacket>> connections;
+    private Queue<Conversation> conversations;
+    private DnsMap dnsMap;
+    private PcapHandle pcap;
+    private FlowPattern pattern;
+    private AtomicBoolean isEoF;
+    
+    
+    /* Constructor */
+    public FlowPatternFinder(PcapHandle _pcap, FlowPattern _pattern) {
+
+        this.connections = new ConcurrentHashMap<Conversation, List<PcapPacket>>();
+        this.conversations = new ConcurrentLinkedQueue<Conversation>();
+        this.dnsMap = new DnsMap();
+        this.isEoF = new AtomicBoolean(false);
+
+        // Get input parameters
+        this.pcap = _pcap;
+        this.pattern = _pattern;
+    }
+    
+    
+    public void start() {
+    
+        // Spawn the main thread
+        Thread mainThread = new Thread(new Runnable() {
+            public void run() {
+                findFlowPattern();
+            }
+        });
+        mainThread.start();
+
+        // Spawn the checker thread
+        Thread checkerThread = new Thread(new Runnable() {
+            public void run() {
+                find();
+            }
+        });
+        checkerThread.start();

 
 

-    public FlowPatternFinder(Map<String, Set<String>> dnsMap) {
-        this.dnsMap = Objects.requireNonNull(dnsMap);
+        /* TODO: Join the threads if we want it to be blocking
+        try {
+            mainThread.join();
+            checkerThread.join();
+        } catch(InterruptedException ex) {
+            ex.printStackTrace();
+        }*/
+        System.out.println("[ start ] Main and checker threads started!");

     }
 
     }
 

-    private static final Set<String> EMPTY_SET = Collections.unmodifiableSet(new HashSet<>());

 
 

-    // TODO clean up exceptions etc.
-    public void findFlowPattern(PcapHandle pcap, FlowPattern pattern)
-            throws PcapNativeException, NotOpenException, TimeoutException {
+    /**
+     * Find patterns based on the FlowPattern object (run by a thread)
+     */
+    private void findFlowPattern() {
+        int counter = 0;

         try {
         try {

-            Packet packet;
+            PcapPacket packet;
+            Set<Integer> seqNumberSet = new HashSet<Integer>();
+            int patternLength = pattern.getLength();

             while ((packet = pcap.getNextPacketEx()) != null) {
 
             while ((packet = pcap.getNextPacketEx()) != null) {
 

+                // Check if this is a valid DNS packet
+                dnsMap.validateAndAddNewEntry(packet);

                 // For now, we only work support pattern search in TCP over IPv4.
                 IpV4Packet ipPacket = packet.get(IpV4Packet.class);
                 TcpPacket tcpPacket = packet.get(TcpPacket.class);
                 // For now, we only work support pattern search in TCP over IPv4.
                 IpV4Packet ipPacket = packet.get(IpV4Packet.class);
                 TcpPacket tcpPacket = packet.get(TcpPacket.class);

-                if (ipPacket == null || tcpPacket == null) {
+                if (ipPacket == null || tcpPacket == null)

                     continue;
                     continue;

-                }

                 String srcAddress = ipPacket.getHeader().getSrcAddr().getHostAddress();
                 String dstAddress = ipPacket.getHeader().getDstAddr().getHostAddress();
                 int srcPort = tcpPacket.getHeader().getSrcPort().valueAsInt();
                 int dstPort = tcpPacket.getHeader().getDstPort().valueAsInt();
                 String srcAddress = ipPacket.getHeader().getSrcAddr().getHostAddress();
                 String dstAddress = ipPacket.getHeader().getDstAddr().getHostAddress();
                 int srcPort = tcpPacket.getHeader().getSrcPort().valueAsInt();
                 int dstPort = tcpPacket.getHeader().getDstPort().valueAsInt();

-                // Is this packet related to the pattern and coming from the cloud server?
-                boolean fromServer = dnsMap.getOrDefault(srcAddress, EMPTY_SET).contains(pattern.getHostname());
-                // Is this packet related to the pattern and going to the cloud server?
-                boolean fromClient = dnsMap.getOrDefault(dstAddress, EMPTY_SET).contains(pattern.getHostname());
-                if (!fromServer && !fromClient) {
-                    // Packet not related to pattern, skip it.
+                // Is this packet related to the pattern and coming to/from the cloud server?
+                boolean fromServer = dnsMap.isRelatedToCloudServer(srcAddress, pattern.getHostname());
+                boolean fromClient = dnsMap.isRelatedToCloudServer(dstAddress, pattern.getHostname());
+                if (!fromServer && !fromClient)  // Packet not related to pattern, skip it.

                     continue;
                     continue;

-                }
-                if (tcpPacket.getPayload() == null) {
-                    // We skip non-payload control packets as these are less predictable and should therefore not be
-                    // part of a signature (e.g. receiver can choose not to ACK immediately)
-                    continue;
-                }
-
+                if (tcpPacket.getPayload() == null) // We skip non-payload control packets as these are less predictable
+                    continue; 

                 // Identify conversations (connections/sessions) by the four-tuple (clientIp, clientPort, serverIp, serverPort).
                 // TODO: this is strictly not sufficient to differentiate one TCP session from another, but should suffice for now.
                 Conversation conversation = fromClient ? new Conversation(srcAddress, srcPort, dstAddress, dstPort) :
                         new Conversation(dstAddress, dstPort, srcAddress, srcPort);
                 // Identify conversations (connections/sessions) by the four-tuple (clientIp, clientPort, serverIp, serverPort).
                 // TODO: this is strictly not sufficient to differentiate one TCP session from another, but should suffice for now.
                 Conversation conversation = fromClient ? new Conversation(srcAddress, srcPort, dstAddress, dstPort) :
                         new Conversation(dstAddress, dstPort, srcAddress, srcPort);

-                List<Packet> listWrappedPacket = new ArrayList<>();
-                listWrappedPacket.add(packet);

                 // Create new conversation entry, or append packet to existing.
                 // Create new conversation entry, or append packet to existing.

-                connections.merge(conversation, listWrappedPacket, (v1, v2) -> {
-                    v1.addAll(v2);
-                    return v1;
-                });
+                List<PcapPacket> listPcapPacket = connections.get(conversation);
+                if (listPcapPacket == null) {
+                    listPcapPacket = new ArrayList<PcapPacket>();
+                    connections.put(conversation, listPcapPacket);
+                }
+                int seqNumber = packet.get(TcpPacket.class).getHeader().getSequenceNumber();
+                boolean retransmission = seqNumberSet.contains(seqNumber);
+                if (!retransmission) { // Do not add if retransmission -> avoid duplicate packets in flow
+                    listPcapPacket.add(packet);
+                    // End of conversation -> trigger thread to check
+                    if (listPcapPacket.size() == patternLength)
+                        conversations.add(conversation);
+                    seqNumberSet.add(seqNumber);
+                }

             }
         } catch (EOFException eofe) {
             }
         } catch (EOFException eofe) {

-            System.out.println("findFlowPattern: finished processing entire file");
-            find(pattern);
+            while (isEoF.compareAndSet(false, true) == false);  // Try to signal EoF!
+            System.out.println("[ findFlowPattern ] Finished processing entire PCAP stream!");
+        } catch (UnknownHostException |
+                 PcapNativeException  |
+                 NotOpenException     |
+                 TimeoutException ex) {
+            ex.printStackTrace();

         }
     }
         }
     }

+    

 
 

-    private void find(FlowPattern pattern) {
-        for (Conversation con : connections.keySet()) {
-            List<Packet> packets = connections.get(con);
-            if (packets.size() != pattern.getPacketOrder().size()) {
-                // Not a complete match if different number of packets.
-                continue;
-            }
+    /**
+     * Checker to match collected patterns (run by a thread)
+     */
+    private void find() {
+
+        while (isEoF.get() == false) {  // Continue until EoF
+            // Get the object from the queue
+            while(conversations.peek() == null) {  // Wait until queue is not empty
+                if (isEoF.get() == true)    // Return if EoF
+                    return;
+            }            
+            Conversation conversation = conversations.poll();
+            // Get the object and remove it from the Map (housekeeping)
+            List<PcapPacket> packets = connections.remove(conversation);

             boolean completeMatch = true;
             for (int i = 0; i < packets.size(); i++) {
                 TcpPacket tcpPacket = packets.get(i).get(TcpPacket.class);
             boolean completeMatch = true;
             for (int i = 0; i < packets.size(); i++) {
                 TcpPacket tcpPacket = packets.get(i).get(TcpPacket.class);


@@ -92,46 +165,16 @@ public class FlowPatternFinder {
                 }
             }
             if (completeMatch) {
                 }
             }
             if (completeMatch) {

-                System.out.println(String.format("found a complete match for %s", pattern.getPatternId()));
-            }
-        }
-    }
-
-    /**
-     * Immutable class used for identifying a conversation/connection/session/flow (packet's belonging to the same
-     * session between a client and a server).
-     */
-    private static class Conversation {
-
-        private final String clientIp;
-        private final int clientPort;
-        private final String serverIp;
-        private final int serverPort;
-
-        public Conversation(String clientIp, int clientPort, String serverIp, int serverPort) {
-            this.clientIp = clientIp;
-            this.clientPort = clientPort;
-            this.serverIp = serverIp;
-            this.serverPort = serverPort;
-        }
-
-
-        // =========================================================================================================
-        // We simply reuse equals and hashCode methods of String.class to be able to use this immutable class as a key
-        // in a Map.
-        @Override
-        public boolean equals(Object obj) {
-            return obj instanceof Conversation && this.toString().equals(obj.toString());
-        }
-        @Override
-        public int hashCode() {
-            return toString().hashCode();
-        }
-        // =========================================================================================================
-
-        @Override
-        public String toString() {
-            return String.format("%s:%d %s:%d", clientIp, clientPort, serverIp, serverPort);
+                PcapPacket firstPacketInFlow = packets.get(0);
+                System.out.println(
+                        String.format("[ find ] Detected a complete match of pattern '%s' at %s!",
+                                pattern.getPatternId(), firstPacketInFlow.getTimestamp().toString()));
+            } /*else {
+                PcapPacket firstPacketInFlow = packets.get(0);
+                System.out.println(
+                        String.format("[ detected a mismatch of pattern '%s' at %s]",
+                                pattern.getPatternId(), firstPacketInFlow.getTimestamp().toString()));
+            }*/

         }
     }
 
         }
     }