123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231 |
- from lea import Lea
- from Attack.MembersMgmtCommAttack import MessageType
- from Attack.MembersMgmtCommAttack import Message
- EPS_TOLERANCE = 1e-13
- 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
- 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):
- """
- Creates an instance of CommunicationProcessor.
- :param packets: the list of abstract packets
- :param mtypes: a dict containing an int to EnumType mapping of MessageTypes
- :param nat: whether NAT is present in this network
- """
- 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)
- def find_interval_most_comm(self, number_ids: int, max_int_time: float):
- """
- Finds the time interval(s) of the given seconds with the most overall communication (i.e. requests and responses)
- that has at least number_ids communication initiators in it.
- :param number_ids: The number of initiator IDs that have to exist in the interval(s)
- :param max_int_time: The maximum time period of the interval
- :return: A list of triples, where each triple contains the initiator IDs, the start index and end index
- of the respective interval in that order. The indices are with respect to self.packets
- """
-
- packets = self.packets
- mtypes = self.mtypes
- idx_low, idx_high = 0, 0
- comm_sum = 0
- cur_highest_sum = 0
- init_ids = []
- possible_intervals = []
-
-
- while True:
- if idx_high < len(packets):
- cur_int_time = float(packets[idx_high]["Time"]) - float(packets[idx_low]["Time"])
-
-
- if greater_than(cur_int_time, max_int_time) or idx_high >= len(packets):
- interval_ids = set(init_ids)
-
- if len(interval_ids) >= number_ids:
- interval = {"IDs": sorted(interval_ids), "Start": idx_low, "End": idx_high-1}
-
- if comm_sum > cur_highest_sum:
- possible_intervals = [interval]
- cur_highest_sum = comm_sum
-
- elif comm_sum == cur_highest_sum:
- possible_intervals.append(interval)
-
- if idx_high >= len(packets):
- break
-
- while greater_than(cur_int_time, max_int_time):
- cur_packet = packets[idx_low]
-
-
- if mtypes[int(cur_packet["Type"])] != MessageType.TIMEOUT:
- comm_sum -= 1
- del init_ids[0]
- idx_low += 1
- cur_int_time = float(packets[idx_high]["Time"]) - float(packets[idx_low]["Time"])
-
- cur_packet = packets[idx_high]
- cur_mtype = mtypes[int(cur_packet["Type"])]
-
- if MessageType.is_request(cur_mtype):
- init_ids.append(cur_packet["Src"])
- comm_sum += 1
-
- elif MessageType.is_response(cur_mtype):
- init_ids.append(cur_packet["Dst"])
- comm_sum += 1
- idx_high += 1
- 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
-
- respnd_ids = set()
-
- msgs, msg_id = [], 0
-
- prev_reqs = {}
- local_init_ids = self.local_init_ids
- external_init_ids = set()
-
- for packet in self.packets:
- id_src, id_dst, msg_type, time = packet["Src"], packet["Dst"], int(packet["Type"]), float(packet["Time"])
- lineno = packet.get("LineNumber", -1)
-
- if (id_src not in local_init_ids) and (id_dst not in local_init_ids):
- continue
-
- msg_type = mtypes[msg_type]
-
- 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:
-
- respnd_ids.add(id_dst)
-
- msg_str = "{0}-{1}".format(id_src, id_dst)
- msg = Message(msg_id, id_src, id_dst, msg_type, time, line_no = lineno)
- msgs.append(msg)
- prev_reqs[msg_str] = msg_id
- msg_id += 1
-
- 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:
-
- external_init_ids.add(id_dst)
- elif id_dst not in local_init_ids:
- continue
- else:
-
- respnd_ids.add(id_src)
-
- msg_str = "{0}-{1}".format(id_dst, id_src)
-
- 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, lineno)
- msgs.append(msg)
-
- del(prev_reqs[msg_str])
- msg_id += 1
- elif msg_type == MessageType.TIMEOUT and id_src in local_init_ids and not self.nat:
-
- msg_str = "{0}-{1}".format(id_dst, id_src)
-
- refer_idx = prev_reqs.get(msg_str)
- if refer_idx is not None:
- msgs[refer_idx].refer_msg_id = msg_id
- if msgs[refer_idx].type == MessageType.SALITY_NL_REQUEST:
- msg = Message(msg_id, id_src, id_dst, MessageType.SALITY_NL_REPLY, time, refer_idx, lineno)
- else:
- msg = Message(msg_id, id_src, id_dst, MessageType.SALITY_HELLO_REPLY, time, refer_idx, lineno)
- msgs.append(msg)
-
- del(prev_reqs[msg_str])
- msg_id += 1
-
- self.respnd_ids = sorted(respnd_ids)
- self.external_init_ids = sorted(external_init_ids)
- self.messages = msgs
-
- 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=0):
- """
- 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()
-
-
- 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_)
-
- 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
|