from lea import Lea from Attack.MembersMgmtCommAttack import MessageType from Attack.MembersMgmtCommAttack import Message # needed because of machine inprecision. E.g A time difference of 0.1s is stored as >0.1s EPS_TOLERANCE = 1e-13 # works for a difference of 0.1, no less class CommunicationProcessor(): """ Class to process parsed input CSV/XML data and retrieve a mapping or other information. """ def __init__(self, packets:list, mtypes:dict, nat:bool): self.packets = packets self.mtypes = mtypes self.nat = nat def set_mapping(self, packets: list, mapped_ids: dict): """ Set the selected mapping for this communication processor. :param packets: all packets contained in the mapped time frame :param mapped_ids: the chosen IDs """ self.packets = packets self.local_init_ids = set(mapped_ids.keys()) def find_interval_most_comm(self, number_ids: int, max_int_time: float): """ Finds a time interval of the given seconds where the given number of IDs commuicate the most. If NAT is active, the most communication is restricted to the most communication by the given number of initiating IDs. If NAT is inactive, the intervall the most overall communication, that has at least the given number of initiating IDs in it, is chosen. :param number_ids: The number of IDs that are to be considered :param max_int_time: A short description of the attack. :return: A triple consisting of the IDs, as well as start and end idx with respect to the given packets. """ packets = self.packets mtypes = self.mtypes def get_nez_comm_counts(comm_counts: dict): """ Filters out all msg_counts that have 0 as value """ nez_comm_counts = dict() for id_ in comm_counts.keys(): count = comm_counts[id_] if count > 0: nez_comm_counts[id_] = count return nez_comm_counts def greater_than(a: float, b: float): """ A greater than operator desgined to handle slight machine inprecision up to EPS_TOLERANCE. :return: True if a > b, otherwise False """ return b - a < -EPS_TOLERANCE def change_comm_counts(comm_counts: dict, idx: int, add=True): """ Changes the communication count, stored in comm_counts, of the initiating ID with respect to the packet specified by the given index. If add is True, 1 is added to the value, otherwise 1 is subtracted. """ change = 1 if add else -1 mtype = mtypes[int(packets[idx]["Type"])] id_src, id_dst = packets[idx]["Src"], packets[idx]["Dst"] if mtype in {MessageType.SALITY_HELLO, MessageType.SALITY_NL_REQUEST}: if id_src in comm_counts: comm_counts[id_src] += change elif change > 0: comm_counts[id_src] = 1 elif mtype in {MessageType.SALITY_HELLO_REPLY, MessageType.SALITY_NL_REPLY}: if id_dst in comm_counts: comm_counts[id_dst] += change elif change > 0: comm_counts[id_dst] = 1 def get_comm_count_first_ids(comm_counts: list): """ Finds the IDs that communicate among themselves the most with respect to the given message counts. :param msg_counts: a sorted list of message counts where each entry is a tuple of key and value :return: The picked IDs and their total message count as a tuple """ # if order of most messages is important, use an additional list picked_ids = {} total_comm_count = 0 # iterate over every message count for i, comm in enumerate(comm_counts): count_picked_ids = len(picked_ids) # if enough IDs have been found, stop if count_picked_ids >= number_ids: break picked_ids[comm[0]] = comm[1] total_comm_count += comm[1] return picked_ids, total_comm_count # first find all possible intervals that contain enough IDs that initiate communication idx_low, idx_high = 0, 0 comm_counts = {} possible_intervals = [] general_comm_sum, cur_highest_sum = 0, 0 # Iterate over all packets from start to finish and process the info of each packet # If time of packet within time interval, update the message count for this communication # If time of packet exceeds time interval, substract from the message count for this communication # Similar to a Sliding Window approach while True: if idx_high < len(packets): cur_int_time = float(packets[idx_high]["Time"]) - float(packets[idx_low]["Time"]) # if current interval time exceeds time interval, save the message counts if appropriate, or stop if no more packets if greater_than(cur_int_time, max_int_time) or idx_high >= len(packets): # get all message counts for communications that took place in the current intervall nez_comm_counts = get_nez_comm_counts(comm_counts) # if we have enough IDs as specified by the caller, mark as possible interval if len(nez_comm_counts) >= number_ids: if self.nat: possible_intervals.append((nez_comm_counts, idx_low, idx_high-1)) elif general_comm_sum >= cur_highest_sum: cur_highest_sum = general_comm_sum possible_intervals.append({"IDs": nez_comm_counts, "CommSum": general_comm_sum, "Start": idx_low, "End": idx_high-1}) general_comm_sum = 0 if idx_high >= len(packets): break # let idx_low "catch up" so that the current interval time fits into the interval time specified by the caller while greater_than(cur_int_time, max_int_time): change_comm_counts(comm_counts, idx_low, add=False) idx_low += 1 cur_int_time = float(packets[idx_high]["Time"]) - float(packets[idx_low]["Time"]) # consume the new packet at idx_high and process its information change_comm_counts(comm_counts, idx_high) idx_high += 1 general_comm_sum += 1 if self.nat: # now find the interval in which as many IDs as specified communicate the most in the given time interval summed_intervals = [] sum_intervals_idxs = [] cur_highest_sum = 0 # for every interval compute the sum of id_counts of the first most communicative IDs and eventually find # the interval(s) with most communication and its IDs # on the side also store the communication count of the individual IDs for j, interval in enumerate(possible_intervals): comm_counts = interval[0].items() sorted_comm_counts = sorted(comm_counts, key=lambda x: x[1], reverse=True) picked_ids, comm_sum = get_comm_count_first_ids(sorted_comm_counts) if comm_sum == cur_highest_sum: summed_intervals.append({"IDs": picked_ids, "CommSum": comm_sum, "Start": interval[1], "End": interval[2]}) elif comm_sum > cur_highest_sum: summed_intervals = [] summed_intervals.append({"IDs": picked_ids, "CommSum": comm_sum, "Start": interval[1], "End": interval[2]}) cur_highest_sum = comm_sum return summed_intervals else: return possible_intervals def det_id_roles_and_msgs(self): """ Determine the role of every mapped ID. The role can be initiator, responder or both. On the side also connect corresponding messages together to quickly find out which reply belongs to which request and vice versa. :return: a triple as (initiator IDs, responder IDs, messages) """ mtypes = self.mtypes # setup initial variables and their values respnd_ids = set() # msgs --> the filtered messages, msg_id --> an increasing ID to give every message an artificial primary key msgs, msg_id = [], 0 # keep track of previous request to find connections prev_reqs = {} local_init_ids = self.local_init_ids external_init_ids = set() # process every packet individually for packet in self.packets: id_src, id_dst, msg_type, time = packet["Src"], packet["Dst"], int(packet["Type"]), float(packet["Time"]) # if if either one of the IDs is not mapped, continue if (id_src not in local_init_ids) and (id_dst not in local_init_ids): continue # convert message type number to enum type msg_type = mtypes[msg_type] # process a request if msg_type in {MessageType.SALITY_HELLO, MessageType.SALITY_NL_REQUEST}: if not self.nat and id_dst in local_init_ids and id_src not in local_init_ids: external_init_ids.add(id_src) elif id_src not in local_init_ids: continue else: # process ID's role respnd_ids.add(id_dst) # convert the abstract message into a message object to handle it better msg_str = "{0}-{1}".format(id_src, id_dst) msg = Message(msg_id, id_src, id_dst, msg_type, time) msgs.append(msg) prev_reqs[msg_str] = msg_id msg_id += 1 # process a reply elif msg_type in {MessageType.SALITY_HELLO_REPLY, MessageType.SALITY_NL_REPLY}: if not self.nat and id_src in local_init_ids and id_dst not in local_init_ids: # process ID's role external_init_ids.add(id_dst) elif id_dst not in local_init_ids: continue else: # process ID's role respnd_ids.add(id_src) # convert the abstract message into a message object to handle it better msg_str = "{0}-{1}".format(id_dst, id_src) # find the request message ID for this response and set its reference index refer_idx = prev_reqs[msg_str] msgs[refer_idx].refer_msg_id = msg_id msg = Message(msg_id, id_src, id_dst, msg_type, time, refer_idx) msgs.append(msg) # remove the request to this response from storage del(prev_reqs[msg_str]) msg_id += 1 # store the retrieved information in this object for later use self.respnd_ids = sorted(respnd_ids) self.external_init_ids = sorted(external_init_ids) self.messages = msgs # return the retrieved information return self.local_init_ids, self.external_init_ids, self.respnd_ids, self.messages def det_ext_and_local_ids(self, prob_rspnd_local: int): """ Map the given IDs to a locality (i.e. local or external} considering the given probabilities. :param comm_type: the type of communication (i.e. local, external or mixed) :param prob_rspnd_local: the probabilty that a responder is local """ external_ids = set() local_ids = self.local_init_ids.copy() # set up probabilistic chooser rspnd_locality = Lea.fromValFreqsDict({"local": prob_rspnd_local*100, "external": (1-prob_rspnd_local)*100}) for id_ in self.external_init_ids: external_ids.add(id_) # determine responder localities for id_ in self.respnd_ids: if id_ in local_ids or id_ in external_ids: continue pos = rspnd_locality.random() if pos == "local": local_ids.add(id_) elif pos == "external": external_ids.add(id_) self.local_ids, self.external_ids = local_ids, external_ids return self.local_ids, self.external_ids