1 package edu.uci.iotproject.detection;
3 import edu.uci.iotproject.analysis.TriggerTrafficExtractor;
4 import edu.uci.iotproject.analysis.UserAction;
5 import edu.uci.iotproject.io.PcapHandleReader;
6 import edu.uci.iotproject.util.PrintUtils;
7 import org.jgrapht.GraphPath;
8 import org.jgrapht.alg.shortestpath.DijkstraShortestPath;
9 import org.jgrapht.graph.DefaultWeightedEdge;
10 import org.jgrapht.graph.SimpleDirectedWeightedGraph;
11 import org.pcap4j.core.*;
13 import java.time.Duration;
14 import java.time.ZoneId;
15 import java.time.format.DateTimeFormatter;
16 import java.time.format.FormatStyle;
18 import java.util.function.Consumer;
21 * Detects an event signature that spans one or multiple TCP connections.
23 * @author Janus Varmarken {@literal <jvarmark@uci.edu>}
24 * @author Rahmadi Trimananda {@literal <rtrimana@uci.edu>}
26 public class SignatureDetector implements PacketListener, ClusterMatcher.ClusterMatchObserver {
29 public static void main(String[] args) throws PcapNativeException, NotOpenException {
30 // String path = "/scratch/July-2018"; // Rahmadi
31 String path = "/Users/varmarken/temp/UCI IoT Project/experiments"; // Janus
32 final String inputPcapFile = path + "/2018-08/kwikset-doorlock/kwikset3.wlan1.local.pcap";
33 final String onSignatureFile = path + "/2018-08/kwikset-doorlock/onSignature-Kwikset-Doorlock-phone.sig";
34 final String offSignatureFile = path + "/2018-08/kwikset-doorlock/offSignature-Kwikset-Doorlock-phone.sig";
36 List<List<List<PcapPacket>>> onSignature = PrintUtils.deserializeSignatureFromFile(onSignatureFile);
37 List<List<List<PcapPacket>>> offSignature = PrintUtils.deserializeSignatureFromFile(offSignatureFile);
39 SignatureDetector onDetector = new SignatureDetector(onSignature, null);
40 SignatureDetector offDetector = new SignatureDetector(offSignature, null);
42 final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofLocalizedDateTime(FormatStyle.MEDIUM).
43 withLocale(Locale.US).withZone(ZoneId.of("America/Los_Angeles"));
45 // Outputs information about a detected event to std.out
46 final Consumer<UserAction> outputter = ua -> {
47 String eventDescription;
48 switch (ua.getType()) {
50 eventDescription = "ON";
53 eventDescription = "OFF";
56 throw new AssertionError("unhandled event type");
58 String output = String.format("[ !!! %s SIGNATURE DETECTED at %s !!! ]",
59 eventDescription, dateTimeFormatter.format(ua.getTimestamp()));
60 System.out.println(output);
63 // Let's create observers that construct a UserAction representing the detected event.
64 final List<UserAction> detectedEvents = new ArrayList<>();
65 onDetector.addObserver((searched, match) -> {
66 PcapPacket firstPkt = match.get(0).get(0);
67 detectedEvents.add(new UserAction(UserAction.Type.TOGGLE_ON, firstPkt.getTimestamp()));
69 offDetector.addObserver((searched, match) -> {
70 PcapPacket firstPkt = match.get(0).get(0);
71 detectedEvents.add(new UserAction(UserAction.Type.TOGGLE_OFF, firstPkt.getTimestamp()));
76 handle = Pcaps.openOffline(inputPcapFile, PcapHandle.TimestampPrecision.NANO);
77 } catch (PcapNativeException pne) {
78 handle = Pcaps.openOffline(inputPcapFile);
80 PcapHandleReader reader = new PcapHandleReader(handle, p -> true, onDetector, offDetector);
81 reader.readFromHandle();
83 // TODO: need a better way of triggering detection than this...
84 onDetector.mClusterMatchers.forEach(cm -> cm.performDetection());
85 offDetector.mClusterMatchers.forEach(cm -> cm.performDetection());
87 // Sort the list of detected events by timestamp to make it easier to compare it line-by-line with the trigger
89 Collections.sort(detectedEvents, Comparator.comparing(UserAction::getTimestamp));
90 // Output the detected events
91 detectedEvents.forEach(outputter);
95 * The signature that this {@link SignatureDetector} is searching for.
97 private final List<List<List<PcapPacket>>> mSignature;
100 * The {@link ClusterMatcher}s in charge of detecting each individual sequence of packets that together make up the
103 private final List<ClusterMatcher> mClusterMatchers;
106 * For each {@code i} ({@code i >= 0 && i < pendingMatches.length}), {@code pendingMatches[i]} holds the matches
107 * found by the {@link ClusterMatcher} at {@code mClusterMatchers.get(i)} that have yet to be "consumed", i.e.,
108 * have yet to be included in a signature detected by this {@link SignatureDetector} (a signature can be encompassed
109 * of multiple packet sequences occurring shortly after one another on multiple connections).
111 private final List<List<PcapPacket>>[] pendingMatches;
114 * Maps a {@link ClusterMatcher} to its corresponding index in {@link #pendingMatches}.
116 private final Map<ClusterMatcher, Integer> mClusterMatcherIds;
118 private final List<SignatureDetectionObserver> mObservers = new ArrayList<>();
120 public SignatureDetector(List<List<List<PcapPacket>>> searchedSignature, String routerWanIp) {
121 // note: doesn't protect inner lists from changes :'(
122 mSignature = Collections.unmodifiableList(searchedSignature);
123 // Generate corresponding/appropriate ClusterMatchers based on the provided signature
124 List<ClusterMatcher> clusterMatchers = new ArrayList<>();
125 for (List<List<PcapPacket>> cluster : mSignature) {
126 clusterMatchers.add(new ClusterMatcher(cluster, routerWanIp, this));
128 mClusterMatchers = Collections.unmodifiableList(clusterMatchers);
131 pendingMatches = new List[mClusterMatchers.size()];
132 for (int i = 0; i < pendingMatches.length; i++) {
133 pendingMatches[i] = new ArrayList<>();
135 Map<ClusterMatcher, Integer> clusterMatcherIds = new HashMap<>();
136 for (int i = 0; i < mClusterMatchers.size(); i++) {
137 clusterMatcherIds.put(mClusterMatchers.get(i), i);
139 mClusterMatcherIds = Collections.unmodifiableMap(clusterMatcherIds);
142 public void addObserver(SignatureDetectionObserver observer) {
143 mObservers.add(observer);
146 public boolean removeObserver(SignatureDetectionObserver observer) {
147 return mObservers.remove(observer);
151 public void gotPacket(PcapPacket packet) {
152 // simply delegate packet reception to all ClusterMatchers.
153 mClusterMatchers.forEach(cm -> cm.gotPacket(packet));
157 public void onMatch(ClusterMatcher clusterMatcher, List<PcapPacket> match) {
158 // Add the match at the corresponding index
159 pendingMatches[mClusterMatcherIds.get(clusterMatcher)].add(match);
160 checkSignatureMatch3();
164 // // No need to check for signature presence until all ClusterMatchers have found a match.
165 // if (Arrays.stream(pendingMatches).noneMatch(l -> l.isEmpty())) {
166 // // There's potentially a signature match...
167 // // TODO need to check if all matches are within X seconds of one another
169 // List<List<PcapPacket>> signatureMatch = new ArrayList<>();
170 // for (int i = 0; i < pendingMatches.length; i++) {
171 // if (signatureMatch.size() != i) {
172 // // Didn't manage to add sequence at previous index to signature match, so not a signature match.
173 // // TODO: clear array?
177 // // Special case with no preceding sequence as this is the first sequence of the signature.
179 // signatureMatch.add(pendingMatches[i].get(0)); // TODO: pick earliest or latest match?
181 // // Fetch the sequence in the signature that precedes this sequence
182 // List<PcapPacket> prev = signatureMatch.get(i-1);
183 // // And get a hold of it's latest packet; note that a match should never be empty so .get() is safe.
184 // PcapPacket prevLatestPkt = prev.stream().max(Comparator.comparing(PcapPacket::getTimestamp)).get();
186 // * Do any of the matches of the sequence at the current index of the signature lie later in time
187 // * than the match of the sequence that precedes it? If so, we are good and can proceed, otherwise we
188 // * do not have a signature match.
190 // Optional<List<PcapPacket>> curr = pendingMatches[i].stream().filter(pkts -> pkts.stream().allMatch(
191 // pkt -> pkt.getTimestamp().isAfter(prevLatestPkt.getTimestamp()))).findFirst();
192 // if (curr.isPresent()) {
193 // // So far so good, keep going.
194 // signatureMatch.add(curr.get());
196 // // Bummer, not a signature match.
197 // // TODO: clear array?
202 // // If we make it out of the loop, it means that we have managed to construct a match of the signature.
203 // // Notify observers of the match.
204 // // TODO: clear array? At the very least we need to remove those entries that we used for this match so they are not reused later.
205 // mObservers.forEach(obs -> obs.onSignatureDetected(mSignature, signatureMatch));
210 private void checkSignatureMatch3() {
211 // << Graph-based approach using Balint's idea. >>
212 // This implementation assumes that the packets in the inner lists (the sequences) are ordered by asc timestamp.
214 // There cannot be a signature match until each ClusterMatcher has found a match of its respective sequence.
215 if (Arrays.stream(pendingMatches).noneMatch(l -> l.isEmpty())) {
217 final SimpleDirectedWeightedGraph<Vertex, DefaultWeightedEdge> graph =
218 new SimpleDirectedWeightedGraph<>(DefaultWeightedEdge.class);
219 // Add a vertex for each match found by all ClusterMatchers
220 // And maintain an array to keep track of what cluster matcher each vertex corresponds to
221 final List<Vertex>[] vertices = new List[pendingMatches.length];
222 for (int i = 0; i < pendingMatches.length; i++) {
223 vertices[i] = new ArrayList<>();
224 for (List<PcapPacket> sequence : pendingMatches[i]) {
225 Vertex v = new Vertex(sequence);
226 vertices[i].add(v); // retain reference for later when we are to add edges
227 graph.addVertex(v); // add to vertex to graph
230 // Add dummy source and sink vertices to facilitate search.
231 final Vertex source = new Vertex(null);
232 final Vertex sink = new Vertex(null);
233 graph.addVertex(source);
234 graph.addVertex(sink);
235 // The source is connected to all vertices that wrap the sequences detected by ClusterMatcher at index 0.
236 // Note: zero cost edges as this is just a dummy link to facilitate search from a common start node.
237 for (Vertex v : vertices[0]) {
238 DefaultWeightedEdge edge = graph.addEdge(source, v);
239 graph.setEdgeWeight(edge, 0.0);
241 // Similarly, all vertices that wrap the sequences detected by the last ClusterMatcher of the signature
242 // are connected to the sink node.
243 for (Vertex v : vertices[vertices.length-1]) {
244 DefaultWeightedEdge edge = graph.addEdge(v, sink);
245 graph.setEdgeWeight(edge, 0.0);
247 // Now link sequences detected by ClusterMatcher at index i to sequences detected by ClusterMatcher at index
248 // i+1 if they obey the timestamp constraint (i.e., that the latter is later in time than the former).
249 for (int i = 0; i < vertices.length; i++) {
251 if (j < vertices.length) {
252 for (Vertex iv : vertices[i]) {
253 PcapPacket ivLast = iv.sequence.get(iv.sequence.size()-1);
254 for (Vertex jv : vertices[j]) {
255 PcapPacket jvFirst = jv.sequence.get(jv.sequence.size()-1);
256 if (ivLast.getTimestamp().isBefore(jvFirst.getTimestamp())) {
257 DefaultWeightedEdge edge = graph.addEdge(iv, jv);
258 // The weight is the duration of the i'th sequence plus the duration between the i'th
259 // and i+1'th sequence.
260 Duration d = Duration.
261 between(iv.sequence.get(0).getTimestamp(), jvFirst.getTimestamp());
262 // Unfortunately weights are double values, so must convert from long to double.
263 // TODO: need nano second precision? If so, use d.toNanos().
264 // TODO: risk of overflow when converting from long to double..?
265 graph.setEdgeWeight(edge, Long.valueOf(d.toMillis()).doubleValue());
267 // Alternative version if we cannot assume that sequences are ordered by timestamp:
268 // if (iv.sequence.stream().max(Comparator.comparing(PcapPacket::getTimestamp)).get()
269 // .getTimestamp().isBefore(jv.sequence.stream().min(
270 // Comparator.comparing(PcapPacket::getTimestamp)).get().getTimestamp())) {
277 // Graph construction complete, run shortest-path to find a (potential) signature match.
278 DijkstraShortestPath<Vertex, DefaultWeightedEdge> dijkstra = new DijkstraShortestPath<>(graph);
279 GraphPath<Vertex, DefaultWeightedEdge> shortestPath = dijkstra.getPath(source, sink);
280 if (shortestPath != null) {
281 // The total weight is the duration between the first packet of the first sequence and the last packet
282 // of the last sequence, so we simply have to compare the weight against the timeframe that we allow
283 // the signature to span. For now we just use the inclusion window we defined for training purposes.
284 // Note however, that we must convert back from double to long as the weight is stored as a double in
286 if (((long)shortestPath.getWeight()) < TriggerTrafficExtractor.INCLUSION_WINDOW_MILLIS) {
287 // There's a signature match!
288 // Extract the match from the vertices
289 List<List<PcapPacket>> signatureMatch = new ArrayList<>();
290 for(Vertex v : shortestPath.getVertexList()) {
291 if (v == source || v == sink) {
292 // Skip the dummy source and sink nodes.
295 signatureMatch.add(v.sequence);
296 // As there is a one-to-one correspondence between vertices[] and pendingMatches[], we know that
297 // the sequence we've "consumed" for index i of the matched signature is also at index i in
298 // pendingMatches. We must remove it from pendingMatches so that we don't use it to construct
299 // another signature match in a later call.
300 pendingMatches[signatureMatch.size()-1].remove(v.sequence);
302 // Declare success: notify observers
303 mObservers.forEach(obs -> obs.onSignatureDetected(mSignature,
304 Collections.unmodifiableList(signatureMatch)));
310 private void checkSignatureMatch2() {
312 * In this implementation, we assume that the packets in the inner lists (the sequences) are ordered by
313 * timestamp (ascending) AND that the outer list is ordered by timestamp of the most recent packet of each inner
314 * list (i.e., the last packet of the inner list).
316 if (Arrays.stream(pendingMatches).noneMatch(l -> l.isEmpty())) {
318 * The signature match consisting of one (or a set of) sequence(s) observed on (potentially multiple)
319 * separate TCP connections. The signature match is reconstructed from the matches found by the individual
320 * ClusterMatchers that each look for a separate sequence of packets occurring on one TCP connection.
321 * Invariant used below: if all entries are non-null, we have a match; initially all entries are null.
323 List<PcapPacket>[] signatureMatch = new List[pendingMatches.length];
324 // List<List<PcapPacket>> signatureMatch = new ArrayList<>();
326 * For the first sequence of the signature, we give preference to the later samples as that strategy makes
327 * it more likely that the full set of sequences that make up the signature fit in the time window that
328 * dictates the maximum time between the sequences of the signature.
330 for (int i = pendingMatches[0].size()-1; i >= 0; i--) {
331 signatureMatch[0] = pendingMatches[0].get(i);
332 // Having selected the most recent sequence
333 for (int j = 1; j < pendingMatches.length; j++) {
334 List<List<PcapPacket>> entry = pendingMatches[j];
342 // First sort by duration
343 Stream<List<PcapPacket>> sortedByDuration = pendingMatches[0].stream().sorted((l1, l2) -> {
344 Instant l1Max = l1.get(l1.size()-1).getTimestamp();
345 Instant l1Min = l1.get(0).getTimestamp();
346 Instant l2Max = l2.get(l2.size()-1).getTimestamp();
347 Instant l2Min = l2.get(0).getTimestamp();
348 Duration l1Duration = Duration.between(l1Min, l1Max);
349 Duration l2Duration = Duration.between(l2Min, l2Max);
351 return l1Duration.compareTo(l2Duration);
353 for (int i = 1; i < pendingMatches.length; i++) {
354 pendingMatches[i].stream()
362 private void checkSignatureMatch() {
363 // There cannot be a signature match until each ClusterMatcher has found a match of its respective sequence.
364 if (Arrays.stream(pendingMatches).noneMatch(l -> l.isEmpty())) {
365 List<List<PcapPacket>> sigMatch = new ArrayList<>();
366 for (int i = 0; i < pendingMatches.length; i++) {
367 if (i + 1 < pendingMatches.length) {
368 // We want to select the current element that is the latest, yet lies before the next element.
369 // Start by fetching the matches at the next index.
370 List<List<PcapPacket>> nextIdxMatches = pendingMatches[i+1];
371 // Create a stream that contains the minimum packet timestamp of each inner list of nextIdMatches
372 Stream<PcapPacket> nextMinTimestamps = nextIdxMatches.stream().
373 map(l -> l.stream().min(Comparator.comparing(PcapPacket::getTimestamp)).get());
374 // Create a stream that contains the maximum packet timestamps of each inner list of current index
375 Stream<PcapPacket> currMaxTimestamps = pendingMatches[i].stream().
376 map(ps -> ps.stream().max(Comparator.comparing(PcapPacket::getTimestamp)).get());
377 currMaxTimestamps.filter(p1 -> nextMinTimestamps.anyMatch(p2 -> p2.getTimestamp().isAfter(p1.getTimestamp())));
381 //pendingMatches[i].stream().filter(ps -> ps.stream().map(p1 -> ))
386 pendingMatches[i].stream().filter(ps -> ps.stream().allMatch(p -> p.getTimestamp().isBefore(
390 pendingMatches[i].stream().filter(ps -> ps.stream().allMatch(p -> p.getTimestamp().isBefore(
394 Stream<PcapPacket> currMaxTimestamps = pendingMatches[i].stream().
395 map(ps -> ps.stream().max(Comparator.comparing(PcapPacket::getTimestamp)));
398 // pendingMatches[i].stream().filter(ps -> ps.stream().allMatch(p -> p.getTimestamp().isBefore(
399 // // which match (item) in 'next' do we consider?
408 interface SignatureDetectionObserver {
409 // TODO: add argument that points to the packets matching the signature
410 void onSignatureDetected(List<List<List<PcapPacket>>> searchedSignature,
411 List<List<PcapPacket>> matchingTraffic);
415 * Encapsulates a {@code List<PcapPacket>} so as to allow the list to be used as a vertex in a graph while avoiding
416 * the expensive {@link AbstractList#equals(Object)} calls when adding vertices to the graph.
417 * Using this wrapper makes the incurred {@code equals(Object)} calls delegate to {@link Object#equals(Object)}
418 * instead of {@link AbstractList#equals(Object)}. The net effect is a faster implementation, but the graph will not
419 * recognize two lists that contain the same items--from a value and not reference point of view--as the same
420 * vertex. However, this is fine for our purposes -- in fact restricting it to reference equality seems more
423 private static class Vertex {
424 private final List<PcapPacket> sequence;
425 private Vertex(List<PcapPacket> wrappedSequence) {
426 sequence = wrappedSequence;