AttackController.py 9.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219
  1. import importlib
  2. import sys
  3. import difflib
  4. import pkgutil
  5. import typing
  6. import Attack.AttackParameters as atkParam
  7. import Core.LabelManager as LabelManager
  8. import Core.Statistics as Statistics
  9. import ID2TLib.Label as Label
  10. import ID2TLib.PcapFile as PcapFile
  11. import ID2TLib.Utility as Util
  12. class AttackController:
  13. def __init__(self, pcap_file: PcapFile.PcapFile, statistics_class: Statistics, label_manager: LabelManager):
  14. """
  15. Creates a new AttackController. The controller manages the attack injection, including the PCAP writing.
  16. :param pcap_file: The source .pcap file to run the attack on.
  17. :param statistics_class: A Statistics Object.
  18. :param label_manager: A LabelManager Object.
  19. """
  20. self.statistics = statistics_class
  21. self.pcap_file = pcap_file
  22. self.label_mgr = label_manager
  23. self.current_attack = None
  24. self.added_attacks = []
  25. self.seed = None
  26. self.total_packets = 0
  27. self.additional_files = []
  28. def set_seed(self, seed: int) -> None:
  29. """
  30. Sets rng seed.
  31. :param seed: rng seed
  32. """
  33. self.seed = seed
  34. def get_seed(self) -> typing.Union[int, None]:
  35. """
  36. Gets rng seed.
  37. :return: The current rng seed
  38. """
  39. return self.seed
  40. @staticmethod
  41. def choose_attack(input_name):
  42. """"
  43. Finds the attack best matching to input_name
  44. :param input_name: The name of the attack the user put in
  45. :return: The best matching attack in case one was found
  46. """
  47. import Attack
  48. # Find all attacks, exclude some classes
  49. package = Attack
  50. available_attacks = []
  51. for _, name, __ in pkgutil.iter_modules(package.__path__):
  52. if name != 'BaseAttack' and name != 'AttackParameters':
  53. available_attacks.append(name)
  54. highest_sim = 0.0
  55. highest_sim_attack = None
  56. for attack in available_attacks:
  57. # Check if attack name is accurate
  58. if input_name == attack:
  59. return attack
  60. # Compares input with one of the available attacks
  61. # Makes comparison with lowercase version with generic endings removed
  62. # List of generic attack name endings can be found in ID2TLib.Utility
  63. counter_check = attack.lower()
  64. if not any(ending in input_name for ending in Util.generic_attack_names):
  65. counter_check = Util.remove_generic_ending(counter_check)
  66. similarity = difflib.SequenceMatcher(None, input_name.lower(), counter_check).ratio()
  67. # Exact match, return appropriate attack name
  68. if similarity == 1.0:
  69. return attack
  70. # Found more likely match
  71. if similarity > highest_sim:
  72. highest_sim = similarity
  73. highest_sim_attack = attack
  74. # Found no exactly matching attack name, print best match and exit
  75. if highest_sim >= 0.6:
  76. print('Found no attack of name ' + input_name + '. The closest match was ' + highest_sim_attack +
  77. '. Use ./id2t -l for a list of available attacks.')
  78. exit(1)
  79. # Found no reasonably matching attack name, recommend -l and exit
  80. else:
  81. print('Found no attack of name ' + input_name + ' or one similar to it.'
  82. ' Use ./id2t -l for an overview of available attacks.')
  83. exit(1)
  84. def create_attack(self, attack_name: str, seed=None):
  85. """
  86. Creates dynamically a new class instance based on the given attack_name.
  87. :param attack_name: The name of the attack, must correspond to the attack's class name.
  88. :param seed: random seed for param generation
  89. :return: None
  90. """
  91. print("\nCreating attack instance of \033[1m" + attack_name + "\033[0m")
  92. # Load attack class
  93. attack_module = importlib.import_module("Attack." + attack_name)
  94. attack_class = getattr(attack_module, attack_name)
  95. # Instantiate the desired attack
  96. self.current_attack = attack_class()
  97. # Initialize the parameters of the attack with defaults or user supplied values.
  98. self.current_attack.set_statistics(self.statistics)
  99. if seed is not None:
  100. self.current_attack.set_seed(seed=seed)
  101. # Unset the user-specified-flag for all parameters set in init_params
  102. for k, v in self.current_attack.params.items():
  103. self.current_attack.params[k] = self.current_attack.ValuePair(v.value, False)
  104. # Record the attack
  105. self.added_attacks.append(self.current_attack)
  106. def process_attack(self, attack: str, params: str, time=False):
  107. """
  108. Takes as input the name of an attack (classname) and the attack parameters as string. Parses the string of
  109. attack parameters, creates the attack by writing the attack packets and returns the path of the written pcap.
  110. :param attack: The classname of the attack to inject.
  111. :param params: The parameters for attack customization, see attack class for supported params.
  112. :param time: Measure packet generation time or not.
  113. :return: The file path to the created pcap file.
  114. """
  115. attack = self.choose_attack(attack)
  116. self.create_attack(attack, self.seed)
  117. print("Validating and adding attack parameters.")
  118. # Add attack parameters if provided
  119. params_dict = []
  120. if isinstance(params, list) and params:
  121. # Convert attack param list into dictionary
  122. for entry in params:
  123. params_dict.append(entry.split('='))
  124. params_dict = dict(params_dict)
  125. # Check if Parameter.INJECT_AT_TIMESTAMP and Parameter.INJECT_AFTER_PACKET are provided at the same time
  126. # if TRUE: delete Parameter.INJECT_AT_TIMESTAMP (lower priority) and use Parameter.INJECT_AFTER_PACKET
  127. if (atkParam.Parameter.INJECT_AFTER_PACKET.value in params_dict) and (
  128. atkParam.Parameter.INJECT_AT_TIMESTAMP.value in params_dict):
  129. print("CONFLICT: Parameters", atkParam.Parameter.INJECT_AT_TIMESTAMP.value, "and",
  130. atkParam.Parameter.INJECT_AFTER_PACKET.value,
  131. "given at the same time. Ignoring", atkParam.Parameter.INJECT_AT_TIMESTAMP.value, "and using",
  132. atkParam.Parameter.INJECT_AFTER_PACKET.value, "instead to derive the timestamp.")
  133. del params_dict[atkParam.Parameter.INJECT_AT_TIMESTAMP.value]
  134. # Extract attack_note parameter, if not provided returns an empty string
  135. key_attack_note = "attack.note"
  136. attack_note = params_dict.get(key_attack_note, "")
  137. params_dict.pop(key_attack_note, None) # delete entry if found, otherwise return an empty string
  138. # Pass paramters to attack controller
  139. self.set_params(params_dict)
  140. else:
  141. attack_note = "This attack used only (random) default parameters."
  142. self.current_attack.init_params()
  143. # Write attack into pcap file
  144. print("Generating attack packets...", end=" ")
  145. sys.stdout.flush() # force python to print text immediately
  146. if time:
  147. self.current_attack.set_start_time()
  148. self.current_attack.generate_attack_packets()
  149. if time:
  150. self.current_attack.set_finish_time()
  151. duration = self.current_attack.get_packet_generation_time()
  152. attack_result = self.current_attack.generate_attack_pcap()
  153. self.total_packets = attack_result[0]
  154. temp_attack_pcap_path = attack_result[1]
  155. if len(attack_result) == 3:
  156. # Extract the list of additional files, if available
  157. self.additional_files += attack_result[2]
  158. print("done. (total: " + str(self.total_packets) + " pkts", end="")
  159. if time:
  160. print(" in ", duration, " seconds", end="")
  161. print(".)")
  162. # Store label into LabelManager
  163. label = Label.Label(attack, self.get_attack_start_utime(), self.get_attack_end_utime(),
  164. self.seed, self.current_attack.params, attack_note)
  165. self.label_mgr.add_labels(label)
  166. return temp_attack_pcap_path, duration
  167. def get_attack_start_utime(self):
  168. """
  169. :return: The start time (timestamp of first packet) of the attack as unix timestamp.
  170. """
  171. return self.current_attack.attack_start_utime
  172. def get_attack_end_utime(self):
  173. """
  174. :return: The end time (timestamp of last packet) of the attack as unix timestamp.
  175. """
  176. return self.current_attack.attack_end_utime
  177. def set_params(self, params: dict):
  178. """
  179. Sets the attack's parameters.
  180. :param params: The parameters in a dictionary: {parameter_name: parameter_value}
  181. :return: None
  182. """
  183. for param_key, param_value in params.items():
  184. self.current_attack.add_param_value(param_key, param_value, user_specified=True)