PcapAddressOperations.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262
  1. from random import choice
  2. from ID2TLib import Statistics
  3. from ID2TLib.IPv4 import IPAddress
  4. is_ipv4 = IPAddress.is_ipv4
  5. class PcapAddressOperations():
  6. def __init__(self, statistics: Statistics, uncertain_ip_mult: int=3):
  7. """
  8. Initializes a pcap information extractor that uses the provided statistics for its operations.
  9. :param statistics: The statistics of the pcap file
  10. :param uncertain_ip_mult: the mutliplier to create new address space when the remaining observed space has been drained
  11. """
  12. self.statistics = statistics
  13. self.UNCERTAIN_IPSPACE_MULTIPLIER = uncertain_ip_mult
  14. self._init_ipaddress_ops()
  15. def get_probable_router_mac(self):
  16. """
  17. Returns the most probable router MAC address based on the most used MAC address in the statistics.
  18. :return: the MAC address
  19. """
  20. self.probable_router_mac, count = self.statistics.process_db_query("most_used(macAddress)", print_results=False)[0]
  21. return self.probable_router_mac # and count as a measure of certainty?
  22. def pcap_contains_priv_ips(self):
  23. """
  24. Returns if the provided traffic contains private IPs.
  25. :return: True if the provided traffic contains private IPs, otherwise False
  26. """
  27. return self.contains_priv_ips
  28. def get_local_address_range(self):
  29. """
  30. Returns a tuple with the start and end of the observed local IP range.
  31. :return: The IP range as tuple
  32. """
  33. return str(self.min_local_ip), str(self.max_local_ip)
  34. def get_count_rem_local_ips(self):
  35. """
  36. Returns the number of local IPs in the pcap file that have not aldready been returned by get_existing_local_ips.
  37. :return: the not yet assigned local IPs
  38. """
  39. return len(self.remaining_local_ips)
  40. def get_existing_local_ips(self, count: int=1):
  41. """
  42. Returns the given number of local IPs that are existent in the pcap file.
  43. :param count: the number of local IPs to return
  44. :return: the chosen local IPs
  45. """
  46. if count > len(self.remaining_local_ips):
  47. print("Warning: There are no more {} local IPs in the .pcap file. Returning all remaining local IPs.".format(count))
  48. total = min(len(self.remaining_local_ips), count)
  49. retr_local_ips = []
  50. local_ips = self.remaining_local_ips
  51. for _ in range(0, total):
  52. random_local_ip = choice(sorted(local_ips))
  53. retr_local_ips.append(str(random_local_ip))
  54. local_ips.remove(random_local_ip)
  55. return retr_local_ips
  56. def get_new_local_ips(self, count: int=1):
  57. """
  58. Returns in the pcap not existent local IPs that are in proximity of the observed local IPs. IPs can be returned
  59. that are either between the minimum and maximum observed IP and are therefore considered certain
  60. or that are above the observed maximum address, are more likely to not belong to the local network
  61. and are therefore considered uncertain.
  62. :param count: the number of new local IPs to return
  63. :return: the newly created local IP addresses
  64. """
  65. # add more unused local ips to the pool, if needed
  66. while len(self.unused_local_ips) < count and self.expand_unused_local_ips() == True:
  67. pass
  68. unused_local_ips = self.unused_local_ips
  69. uncertain_local_ips = self.uncertain_local_ips
  70. count_certain = min(count, len(unused_local_ips))
  71. retr_local_ips = []
  72. for _ in range(0, count_certain):
  73. random_local_ip = choice(sorted(unused_local_ips))
  74. retr_local_ips.append(str(random_local_ip))
  75. unused_local_ips.remove(random_local_ip)
  76. # retrieve uncertain local ips
  77. if count_certain < count:
  78. count_uncertain = count - count_certain
  79. # check if new uncertain IPs have to be created
  80. if len(uncertain_local_ips) < count_uncertain:
  81. ipspace_multiplier = self.UNCERTAIN_IPSPACE_MULTIPLIER
  82. max_new_ip = self.max_uncertain_local_ip.to_int() + ipspace_multiplier * count_uncertain
  83. count_new_ips = max_new_ip - self.max_uncertain_local_ip.to_int()
  84. # create ipspace_multiplier * count_uncertain new uncertain local IP addresses
  85. last_gen_ip = None
  86. for i in range(1, count_new_ips + 1):
  87. ip = IPAddress.from_int(self.max_uncertain_local_ip.to_int() + i)
  88. # exclude the definite broadcast address
  89. if self.priv_ip_segment:
  90. if ip.to_int() >= self.priv_ip_segment.last_address().to_int():
  91. break
  92. uncertain_local_ips.add(ip)
  93. last_gen_ip = ip
  94. self.max_uncertain_local_ip = last_gen_ip
  95. # choose the uncertain IPs to return
  96. total_uncertain = min(count_uncertain, len(uncertain_local_ips))
  97. for _ in range(0, total_uncertain):
  98. random_local_ip = choice(sorted(uncertain_local_ips))
  99. retr_local_ips.append(str(random_local_ip))
  100. uncertain_local_ips.remove(random_local_ip)
  101. return retr_local_ips
  102. def get_existing_external_ips(self, count: int=1):
  103. """
  104. Returns the given number of external IPs that are existent in the pcap file.
  105. :param count: the number of external IPs to return
  106. :return: the chosen external IPs
  107. """
  108. if not (len(self.external_ips) > 0):
  109. print("Warning: .pcap does not contain any external ips.")
  110. return []
  111. total = min(len(self.remaining_external_ips), count)
  112. retr_external_ips = []
  113. external_ips = self.remaining_external_ips
  114. for _ in range(0, total):
  115. random_external_ip = choice(sorted(external_ips))
  116. retr_external_ips.append(str(random_external_ip))
  117. external_ips.remove(random_external_ip)
  118. return retr_external_ips
  119. def _init_ipaddress_ops(self):
  120. """
  121. Load and process data needed to perform functions on the IP addresses contained in the statistics
  122. """
  123. # retrieve local and external IPs
  124. all_ips_str = set(self.statistics.process_db_query("all(ipAddress)", print_results=False))
  125. external_ips_str = set(self.statistics.process_db_query("ipAddress(macAddress=%s)" % self.get_probable_router_mac(), print_results=False)) # including router
  126. local_ips_str = all_ips_str - external_ips_str
  127. external_ips = set()
  128. local_ips = set()
  129. self.contains_priv_ips = False
  130. self.priv_ip_segment = None
  131. # convert local IP strings to IPv4.IPAddress representation
  132. for ip in local_ips_str:
  133. if is_ipv4(ip):
  134. ip = IPAddress.parse(ip)
  135. if ip.is_private() and not self.contains_priv_ips:
  136. self.contains_priv_ips = True
  137. self.priv_ip_segment = ip.get_private_segment()
  138. # exclude local broadcast address and other special addresses
  139. if (not str(ip) == "255.255.255.255") and (not ip.is_localhost()) and (not ip.is_multicast()) and (not ip.is_reserved()) and (not ip.is_zero_conf()):
  140. local_ips.add(ip)
  141. # convert external IP strings to IPv4.IPAddress representation
  142. for ip in external_ips_str:
  143. if is_ipv4(ip):
  144. ip = IPAddress.parse(ip)
  145. # if router MAC can definitely be mapped to local/private IP, add it to local_ips (because at first it is stored in external_ips, see above)
  146. # this depends on whether the local network is identified by a private IP address range or not.
  147. if ip.is_private():
  148. local_ips.add(ip)
  149. # exclude local broadcast address and other special addresses
  150. elif (not str(ip) == "255.255.255.255") and (not ip.is_localhost()) and (not ip.is_multicast()) and (not ip.is_reserved()) and (not ip.is_zero_conf()):
  151. external_ips.add(ip)
  152. # save the certain unused local IPs of the network
  153. # to do that, divide the unused local Addressspace into chunks of (chunks_size) Addresses
  154. # initally only the first chunk will be used, but more chunks can be added to the pool of unused_local_ips if needed
  155. self.min_local_ip, self.max_local_ip = min(local_ips), max(local_ips)
  156. local_ip_range = (self.max_local_ip.to_int()) - (self.min_local_ip.to_int() + 1)
  157. if local_ip_range < 0:
  158. # for min,max pairs like (1,1), (1,2) there is no free address in between, but for (1,1) local_ip_range may be -1, because 1-(1+1)=-1
  159. local_ip_range = 0
  160. # chunk size can be adjusted if needed
  161. self.chunk_size = 200
  162. self.current_chunk = 1
  163. if local_ip_range < self.chunk_size:
  164. # there are not more than chunk_size unused IP Addresses to begin with
  165. self.chunks = 0
  166. self.chunk_remainder = local_ip_range
  167. else:
  168. # determine how many chunks of (chunk_size) Addresses there are and the save the remainder
  169. self.chunks = local_ip_range // self.chunk_size
  170. self.chunk_remainder = local_ip_range % self.chunk_size
  171. # add the first chunk of IP Addresses
  172. self.unused_local_ips = set()
  173. self.expand_unused_local_ips()
  174. # save the gathered information for efficient later use
  175. self.external_ips = frozenset(external_ips)
  176. self.remaining_external_ips = external_ips
  177. self.max_uncertain_local_ip = self.max_local_ip
  178. self.local_ips = frozenset(local_ips)
  179. self.remaining_local_ips = local_ips
  180. self.uncertain_local_ips = set()
  181. def expand_unused_local_ips(self):
  182. """
  183. expands the set of unused_local_ips by one chunk_size
  184. to illustrate this algorithm: suppose we have a chunksize of 100 and an Address space of 1 to 1000 (1 and 1000 are unused too), we then have 10 chunks
  185. every time this method is called, one chunk (100 Addresses) is added, each chunk starts at the base_address + the number of its chunk
  186. then, every chunk_amounth'th Address is added. Therefore for 10 chunks, every 10th address is added
  187. For the above example for the first, second and last call, we get the following IPs, respectively:
  188. first Call: 1+0, 1+10, 1+20, 1+30, ..., 1+990
  189. second Call: 2+0, 2+10, 2+20, 2+30, ..., 2+990
  190. ten'th Call: 10+0, 10+10, 10+20, 10+30, ..., 10+990
  191. :return: False if there are no more available unusd local IP Addresses, True otherwise
  192. """
  193. if self.current_chunk == self.chunks+1:
  194. # all chunks are used up, therefore add the remainder
  195. remainder_base_addr = self.min_local_ip.to_int() + self.chunks*self.chunk_size + 1
  196. for i in range(0,self.chunk_remainder):
  197. ip = IPAddress.from_int(remainder_base_addr + i)
  198. self.unused_local_ips.add(ip)
  199. self.current_chunk = self.current_chunk + 1
  200. return True
  201. elif self.current_chunk <= self.chunks:
  202. # add another chunk
  203. # choose IPs from the whole address space, that is available
  204. base_address = self.min_local_ip.to_int() + self.current_chunk
  205. for i in range(0,self.chunk_size):
  206. ip = IPAddress.from_int(base_address + i*self.chunks)
  207. self.unused_local_ips.add(ip)
  208. self.current_chunk = self.current_chunk + 1
  209. return True
  210. else:
  211. # no free IPs remaining
  212. return False