123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410 |
- package analyzer;
- import analyzer.models.Client;
- import analyzer.models.Hashtag;
- import analyzer.models.Round;
- import analyzer.models.User;
- import java.io.IOException;
- import java.util.*;
- import java.util.concurrent.ExecutionException;
- import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Executors;
- import java.util.concurrent.Future;
- import java.util.concurrent.atomic.AtomicInteger;
- /**
- * This class calls the parsers running parallel, then performs intersection attacks from the loaded objects.
- */
- public class Analyzer {
- private static final int threadPoolSize = 8;
- public static final int roundLength = 1800;
- public static final int clientNo = 1000000;
- public static final String datasetName = "reddit"; //;
- public static final boolean coverTraffic = false;
- public static final int delay = 0;
- public static String childPath = "";
- public static String logType = ""; //"top-" "random_"
- public static String logPath = "";
- public static String clientLogPath = "";
- public static String serverLogPath = "";
- //TEST
- //private static final User toTraceUser = new User(306637085);
- private static final String toTraceHashtag = "InstantFollowBack";
- private static final String intersectingClientID = "client-937851";
- private static final long targetUserID = 398449825L;
- public static void main(String[] args) throws InterruptedException, ExecutionException {
- //a.traceHashtag();
- //traceClientHashtagLink();
- traceClientUserLink();
- }
- public Analyzer () throws InterruptedException {
- pathConfig();
- Thread slogs = new Thread(() -> {
- try {
- ServerLogParser.run();
- } catch (IOException e) {
- e.printStackTrace();
- }
- });
- Thread clogs = new Thread(ClientLogParser::run);
- slogs.start();
- clogs.start();
- slogs.join();
- clogs.join();
- //checkParsing();
- System.out.println("round length: " + Analyzer.roundLength);
- //new ServerLogParser(serverLogPath).traceUser();
- }
- private static void pathConfig() {
- childPath = coverTraffic ? "ct" + "\\" + coverTraffic : delay > 0 ? "delay" + "\\" + delay : roundLength + "";
- if(datasetName.equals("twitter")) {
- logType = "twitter\\random_";
- logPath = "C:\\Users\\Admin\\Desktop\\Skripts\\Thesis\\Repo\\local\\vm-mount\\logs\\" + logType + clientNo + "\\" + childPath;
- }
- else if(datasetName.equals("reddit")) {
- logType = "reddit\\";
- logPath = "C:\\Users\\Admin\\Desktop\\Skripts\\Thesis\\Repo\\local\\vm-mount\\logs\\" + logType + "\\" + "ct\\0\\";
- }
- clientLogPath = logPath + "\\clogs.txt";
- serverLogPath = logPath + "\\slogs.txt";
- }
- private static void traceClientUserLink() throws InterruptedException, ExecutionException {
- Analyzer privateInst = new Analyzer();
- int roundNum = privateInst.intersectUsers(ClientLogParser.clients.get(intersectingClientID), 1, true);
- System.out.println("roundNum: " + roundNum);
- }
- private static void traceClientHashtagLink() throws ExecutionException, InterruptedException {
- Analyzer privateInst = new Analyzer();
- List<Map.Entry<String, Integer>> intersection = privateInst.intersectHashtags(ClientLogParser.clients.get(intersectingClientID), 1000, true);
- Hashtag[] hashtags = new Hashtag[intersection.size()];
- for(int i = 0; i < hashtags.length; i++)
- hashtags[i] = ServerLogParser.hashtags.get(intersection.get(i).getKey());
- System.out.println();
- List<Map.Entry<Hashtag, Integer>> results = privateInst.calculatePointsGivenHashtags(ClientLogParser.clients.get(intersectingClientID), hashtags, 1, true);
- System.out.println("Result size: " + results.size());
- //User targetUser = ServerLogParser.users.get(targetUserID);
- //System.out.println();
- //targetUser.getHashtags().forEach((k, v) -> System.out.println(k.getName()));
- }
- public void traceHashtag() throws InterruptedException, ExecutionException {
- Hashtag staticTargetHashtag = ServerLogParser.hashtags.get(toTraceHashtag);
- Client staticTargetClient = ClientLogParser.clients.get(intersectingClientID);
- System.out.println(staticTargetHashtag.getName() + " size: " + staticTargetHashtag.getRoundMap().size());
- long startCalculatingMaxPoint = System.currentTimeMillis();
- calculatePointsAllClients(staticTargetHashtag, 0, true);
- long calculatingMaxPointTime = System.currentTimeMillis() - startCalculatingMaxPoint;
- System.out.println("CalculatingMaxPointTime: " + calculatingMaxPointTime + " ms");
- System.out.println();
- System.out.println(staticTargetClient.getId() + " " + staticTargetClient.postingTraces());
- //print traced hashtag
- System.out.println("Hashtag " + staticTargetHashtag.getName() + " " + staticTargetHashtag.hashtagTraces());
- long startIntersecting = System.currentTimeMillis();
- intersectHashtags(staticTargetClient, 0, true);
- long intersectingTime = System.currentTimeMillis() - startIntersecting;
- System.out.println("Intersecting Time: " + intersectingTime + " ms");
- }
- /**
- * This method calculates points between a given hashtag and all clients, and prints out these points if needed.
- * @param hashtag hashtag to calculate points against all clients
- * @param returnNum number of returned highest point 2-tuples (client, point)
- * @param printOut true if log print-out is needed
- * @return list of highest point 2-tuples (client, point)
- * @throws InterruptedException
- * @throws ExecutionException
- */
- public List<Map.Entry<Client, Integer>> calculatePointsAllClients(Hashtag hashtag, int returnNum, boolean printOut) throws InterruptedException, ExecutionException {
- Hashtable<Client, Integer> clientPointsWithHashtag = new Hashtable<>(ClientLogParser.clients.size(), 1);
- ClientLogParser.clients.values().forEach(c -> clientPointsWithHashtag.put(c, 0));
- ExecutorService executor = Executors.newFixedThreadPool(threadPoolSize);
- List<Future<?>> tasks = new ArrayList<>();
- for(Client c : clientPointsWithHashtag.keySet()) {
- Future<?> task = executor.submit(new PointCalculatorTask(c, hashtag, clientPointsWithHashtag, null));
- tasks.add(task);
- }
- for(Future<?> task : tasks)
- task.get();
- executor.shutdown();
- long startSortTime = System.currentTimeMillis();
- List<Map.Entry<Client, Integer>> sortedList = new ArrayList<>(clientPointsWithHashtag.entrySet());
- sortedList.sort(((o1, o2) -> o2.getValue().compareTo(o1.getValue())));
- long sortTime = System.currentTimeMillis() - startSortTime;
- if(printOut) {
- System.out.println("Points for hashtag " + hashtag.getName());
- System.out.println("Sorting time : " + sortTime + " ms");
- System.out.println("Client-ID\tPoints");
- int maxPrintOut = Math.min(20, sortedList.size());
- for(int i=0; i < maxPrintOut; i++) {
- Map.Entry<Client, Integer> e = sortedList.get(i);
- System.out.println(e.getKey().getId() + "\t" + e.getValue());
- }
- }
- if(returnNum > 0) {
- if(returnNum >= sortedList.size())
- return sortedList;
- else {
- List<Map.Entry<Client, Integer>> returnList = new ArrayList<>();
- int maxVal = sortedList.get(0).getValue();
- for (Map.Entry<Client, Integer> entry : sortedList) {
- if (entry.getValue().equals(maxVal)) {
- returnList.add(entry);
- } else {
- if (returnList.size() < returnNum) {
- maxVal = entry.getValue();
- returnList.add(entry);
- } else
- break;
- }
- }
- return returnList;
- }
- }
- return null;
- }
- /**
- * This method calculates points between a client and a set of hashtags and prints out these points if needed.
- * @param client client to calculate point against a set of hashtags
- * @param hashtags set of hashtags to calculate point against client
- * @param returnNum number of returned highest point 2-tuples (hashtag, point)
- * @param printOut true if log print-out is needed
- * @return list of highest point 2-tuples (hashtag, point)
- * @throws InterruptedException
- * @throws ExecutionException
- */
- public List<Map.Entry<Hashtag, Integer>> calculatePointsGivenHashtags(Client client, Hashtag[] hashtags, int returnNum, boolean printOut) throws InterruptedException, ExecutionException {
- Hashtable<Hashtag, Integer> hashtagPointMap = new Hashtable<>(hashtags.length, 1);
- Arrays.stream(hashtags).forEach(h -> hashtagPointMap.put(h, 0));
- ExecutorService executor = Executors.newFixedThreadPool(threadPoolSize);
- List<Future<?>> tasks = new ArrayList<>();
- for(Hashtag h : hashtags) {
- Future<?> task = executor.submit(new PointCalculatorTask(client, h, null, hashtagPointMap));
- tasks.add(task);
- }
- for(Future<?> task : tasks)
- task.get();
- executor.shutdown();
- long startSortTime = System.currentTimeMillis();
- List<Map.Entry<Hashtag, Integer>> sortedList = new ArrayList<>(hashtagPointMap.entrySet());
- sortedList.sort(((o1, o2) -> o2.getValue().compareTo(o1.getValue())));
- long sortTime = System.currentTimeMillis() - startSortTime;
- if(returnNum > 0) {
- if(returnNum >= sortedList.size())
- return sortedList;
- else {
- List<Map.Entry<Hashtag, Integer>> returnList = new ArrayList<>();
- int maxVal = sortedList.get(0).getValue();
- for (Map.Entry<Hashtag, Integer> entry : sortedList) {
- if (entry.getValue().equals(maxVal)) {
- returnList.add(entry);
- } else {
- if (returnList.size() < returnNum) {
- maxVal = entry.getValue();
- returnList.add(entry);
- } else
- break;
- }
- }
- if(printOut) {
- System.out.println("Points for Client " + client.getId());
- System.out.println("Sorting time : " + sortTime + " ms");
- System.out.println("Client-ID\tPoints");
- for(int i=0; i < returnList.size(); i++) {
- Map.Entry<Hashtag, Integer> e = returnList.get(i);
- System.out.println(e.getKey().getName() + "\t" + e.getValue());
- }
- }
- return returnList;
- }
- }
- return null;
- }
- /**
- * This method performs intersection attacks by calculating the correlation points from a given targetClient against hashtags based on active rounds
- * @param targetClient target client to perform the attack
- * @param returnNum number of returned 2-tuples of most relevant suspects
- * @param printOut true if log print-out is needed
- * @return list of highest point 2-tuples (hashtag, point)
- */
- public List<Map.Entry<String, Integer>> intersectHashtags(Client targetClient, int returnNum, boolean printOut) {
- //initialize a mapping of all possible hashtags with their no of occurrences
- Map<String, Integer> results = new Hashtable<>();
- List<Round> targetRounds = new ArrayList<>(targetClient.getRounds().keySet());
- targetRounds.sort(Comparator.comparingInt(Round::getNo));
- int targetNoRounds = targetRounds.size();
- int lastRound = 1;
- long lastResult = ServerLogParser.hashtags.size();
- for(int i = 0; i < targetNoRounds; i++) {
- results.forEach((k, v) -> results.replace(k, v - 1));
- //results.entrySet().removeIf(e -> (e.getValue() < 0));
- int targetRoundNo = targetRounds.get(i).getNo();
- List<Hashtag> hashtagsOfARound = new ArrayList<>(ServerLogParser.rounds.get(targetRoundNo).getHashtags().keySet());
- for(Hashtag hashtag: hashtagsOfARound) {
- if(results.containsKey(hashtag.getName()))
- results.replace(hashtag.getName(), results.get(hashtag.getName()) + 2);
- else
- results.put(hashtag.getName(), -(i + 1) + 1);
- }
- //long resultSize = Collections.frequency(results.values(), Collections.max(results.values(), Integer::compare));
- //for (int x = lastRound + 1; x < targetRoundNo; x++) {
- // System.out.print(lastResult+ ",");
- //}
- //lastRound = targetRoundNo;
- //lastResult = resultSize;
- //System.out.print(resultSize + ",");
- }
- //System.out.println();
- //results.entrySet().removeIf(e -> (e.getValue() < targetNoRound /2));
- List<Map.Entry<String, Integer>> sortedList = new ArrayList<>(results.entrySet());
- sortedList.sort(Map.Entry.comparingByValue((v1, v2) -> Integer.compare(v2, v1)));
- if(returnNum > 0) {
- if(returnNum >= sortedList.size())
- return sortedList;
- else {
- List<Map.Entry<String, Integer>> returnList = new ArrayList<>();
- int maxVal = sortedList.get(0).getValue();
- for (Map.Entry<String, Integer> entry : sortedList) {
- if (entry.getValue().equals(maxVal)) {
- returnList.add(entry);
- } else {
- if (returnList.size() < returnNum) {
- maxVal = entry.getValue();
- returnList.add(entry);
- } else
- break;
- }
- }
- if(printOut) {
- System.out.print(targetClient.getId() + ": ");
- System.out.println("no of rounds of " + targetClient.getId() + ": " + targetNoRounds);
- for(int i = 0; i < returnList.size(); i++) {
- System.out.print(returnList.get(i).getKey() + "|" + returnList.get(i).getValue() + " ");
- }
- //sortedList.forEach(e -> System.out.print(e.getKey() + "|" + e.getValue() + " "));
- System.out.println();
- }
- return returnList;
- }
- }
- return null;
- }
- /**
- * This method performs intersection attacks by matching the behaviour of targetClient with all other clients.
- * @param targetClient target client to perform the attacks
- * @param returnNum number of returned 2-tuples of most relevant suspects
- * @param printOut true if log print-out is needed
- * @return number of rounds needed until the attacks finish
- */
- public int intersectUsers(Client targetClient, int returnNum, boolean printOut) {
- int roundProcessed = -1;
- List<User> userList = new ArrayList<>(ServerLogParser.users.values());
- List<Round> roundList = new ArrayList<>(ServerLogParser.rounds.values());
- roundList.sort(Comparator.comparingInt(Round::getNo));
- for(Round r: roundList) {
- if(!targetClient.getRounds().containsKey(r)) {
- userList.removeIf(u -> u.getRounds().containsKey(r));
- } else {
- userList.removeIf(u -> !u.getRounds().containsKey(r) || !u.getRounds().get(r).equals(targetClient.getRounds().get(r)));
- }
- //System.out.print(userList.size() + ", ");
- if(userList.size() <= 1) {
- roundProcessed = r.getNo();
- break;
- }
- }
- //System.out.println();
- if(userList.size() != 1) {
- roundProcessed = -1;
- }
- if(printOut) {
- System.out.println("Suspect(s): ");
- userList.forEach(e -> System.out.println(e.getId()));
- System.out.println();
- System.out.println("rounds processed: " + roundProcessed);
- System.out.println();
- System.out.println();
- }
- return roundProcessed;
- }
- /**
- * This method checks the parsing processes, exits the JVM if there is an error.
- */
- private static void checkParsing() {
- if(ClientLogParser.clientRoundNo != ServerLogParser.serverRoundNo) {
- System.out.println("Parsing Error: Round numbers are not equals");
- System.exit(1);
- }
- if(ClientLogParser.clients.size() != ServerLogParser.users.size()) {
- System.out.println("Parsing Error: Client/User numbers are not equals");
- System.exit(1);
- }
- }
- /**
- * This class models the parallel execution of point calculations
- */
- private static class PointCalculatorTask implements Runnable {
- private final Client client;
- private final Hashtag hashtag;
- private final Hashtable<Client, Integer> clientMap;
- private final Hashtable<Hashtag, Integer> hashtagMap;
- private PointCalculatorTask(Client c, Hashtag hashtag, Hashtable<Client, Integer> clientMap, Hashtable<Hashtag, Integer> hashtagMap) {
- this.client = c;
- this.hashtag = hashtag;
- this.clientMap = clientMap;
- this.hashtagMap = hashtagMap;
- }
- @Override
- public void run() {
- calculatePoint();
- }
- private void calculatePoint() {
- Map<Round, Integer> clone = new Hashtable<>(client.getRounds());
- AtomicInteger n = new AtomicInteger();
- clone.keySet().retainAll(this.hashtag.getRoundMap().keySet());
- clone.forEach((k, v) -> n.addAndGet((Math.min(v, this.hashtag.getRoundMap().get(k)))));
- if(clientMap != null)
- clientMap.put(this.client, n.get());
- if(hashtagMap != null)
- hashtagMap.put(this.hashtag, n.get());
- }
- }
- }
|