← Back
Editing: ip_map.py
# Copyright 2020 Red Hat, Inc. Jake Hunsaker <jhunsake@redhat.com> # This file is part of the sos project: https://github.com/sosreport/sos # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # version 2 of the GNU General Public License. # # See the LICENSE file in the source distribution for further information. import ipaddress from sos.cleaner.mappings import SoSMap class SoSIPMap(SoSMap): """A mapping store for IP addresses Each IP address added to this map is chcked for subnet membership. If that subnet already exists in the map, then IP addresses are deterministically generated sequentially within that subnet. For example, if a given IP is matched to subnet 192.168.1.0/24 then 192.168.1 may be obfuscated to 100.11.12.0/24. Each IP address in the original 192.168.1.0/24 subnet will then be assigned an address in 100.11.12.0/24 sequentially, such as 100.11.12.1, 100.11.12.2, etc... Internally, the ipaddress library is used to manipulate the address objects however, when retrieved by SoSCleaner any values will be strings. """ ignore_matches = [ r'127.*', r'::1', r'0\.(.*)?', r'1\.(.*)?', r'8.8.8.8', r'8.8.4.4', r'169.254.*', r'255.*' ] _networks = {} obfuscated_ips = set() network_first_octet = 100 skip_network_octets = ['127', '169', '172', '192'] compile_regexes = False # counter for obfuscating a single IP address; the value stands for # 172.17.0.0; we use a private block of IP addresses and ignore # 172.16.0.0/16 block as those addresses are more often used in real # (an attempt to prevent confusion) _saddr_cnt = 2886795264 def conf_update(self, config): """Override the base conf_update() so that we can add items into obfuscated_ips. """ for value in config.values(): self.obfuscated_ips.add(value.split('/', maxsplit=1)[0]) super().conf_update(config) def insert_to_dataset(self, item, value): self.obfuscated_ips.add(value.split('/', maxsplit=1)[0]) self.dataset[item] = value def ip_in_dataset(self, ipaddr): """There are multiple ways in which an ip address could be handed to us in a way where we're matching against a previously obfuscated address. Here, match the ip address to any of the obfuscated addresses we've already created """ addr_str = str(ipaddr).split('/', maxsplit=1)[0] return addr_str in self.obfuscated_ips def get(self, item): """Ensure that when requesting an obfuscated address, we return a str object instead of an IPv(4|6)Address object """ filt_start = ('/', '=', ']', ')') if item.startswith(filt_start): item = item.lstrip(''.join(filt_start)) if item in self.dataset: return self.dataset[item] if self.ignore_item(item) or self.ip_in_dataset(item): return item # it's not in there, but let's make sure we haven't previously added # an address with a CIDR notation and we're now looking for it without # that notation if '/' not in item: for key, value in self.dataset.items(): if key.startswith(item): return value.split('/')[0] # fallback to the default map behavior of adding it fresh return self.add(item) def set_ip_cidr_from_existing_subnet(self, addr): """Determine if a given address is in a subnet of an already obfuscated network and if it is, then set the address' network to the network object we're tracking. This allows us to match ip addresses with or without a CIDR notation and maintain proper network relationships. """ nets = [] for net in self._networks: if addr.ip == net.broadcast_address: addr.network = net return if addr.ip in net: nets.append(net) # assign the address to the smallest network that was matched. This is # necessary due to certain files specifying addresses that cause the # ipaddress library to create artificially huge subnets that will # include the actual subnets used by the system if nets: nets.sort(key=lambda n: n.prefixlen, reverse=True) addr.network = nets[0] def sanitize_item(self, item): """Given an IP address, sanitize it to an obfuscated network or host address as appropriate """ try: addr = ipaddress.ip_interface(item) except ValueError: # not an IP, add it to the skip list to avoid flooding logs self.ignore_matches.append(item) raise network = addr.network if str(network.netmask) == '255.255.255.255': # check to see if this IP is in a subnet of an already obfuscated # network and if it has, replace the default /32 netmask that # ipaddress applies to no CIDR-notated addresses self.set_ip_cidr_from_existing_subnet(addr) else: # we have a CIDR notation, so generate an obfuscated network # address and then generate an IP address within that network's # range self.sanitize_network(network) return self.sanitize_ipaddr(addr) def sanitize_network(self, network): """Obfuscate the network address provided, and if there are host bits in the address then obfuscate those as well """ # check if the address is in a network we've already encountered if network not in self._networks: self._new_obfuscated_network(network) def sanitize_ipaddr(self, addr): """Obfuscate the IP address within the known obfuscated network """ # get the obfuscated network object if addr.network in self._networks: _obf_network = self._networks[addr.network] # if the plain address is the broadcast address for it's own # network, then assign the broadcast address for the obfuscated # network if addr.ip == addr.network.broadcast_address: return str(_obf_network.broadcast_address) # otherwise within that obfuscated network grab the next available # address from it for _ip in _obf_network.hosts(): if not self.ip_in_dataset(_ip): # the ipaddress module does not assign the network's # netmask to hosts in the hosts() generator for some reason return f"{str(_ip)}/{_obf_network.prefixlen}" # ip is a single ip address without the netmask return self._new_obfuscated_single_address() def _new_obfuscated_single_address(self): # increment the counter and ignore *.0 and *.255 addresses self._saddr_cnt += 1 while self._saddr_cnt % 256 in (0, 255): self._saddr_cnt += 1 # split the counter value to four octets (i.e. % 256) to get an # obfuscated IP address _addr = f"{self._saddr_cnt >> 24}.{(self._saddr_cnt >> 16) % 256}." \ f"{(self._saddr_cnt >> 8) % 256}.{self._saddr_cnt % 256}" if _addr in self.dataset.values(): return self._new_obfuscated_single_address() return _addr def _new_obfuscated_network(self, network): """Generate an obfuscated network address for the network address given which will allow us to maintain network relationships without divulging actual network details Positional arguments: :param network: An ipaddress.IPv{4|6)Network object """ _obf_network = None if isinstance(network, ipaddress.IPv4Network): if self.network_first_octet in self.skip_network_octets: self.network_first_octet += 1 _obf_address = f"{self.network_first_octet}.0.0.0" _obf_mask = network.with_netmask.split('/')[1] _obf_network = ipaddress.IPv4Network(f"{_obf_address}/{_obf_mask}") self.network_first_octet += 1 if isinstance(network, ipaddress.IPv6Network): # TODO: define this pass if _obf_network: _obf_network_s = str(_obf_network) self._networks[network] = _obf_network self.dataset[str(network)] = _obf_network_s self.obfuscated_ips.add(_obf_network_s.split('/', maxsplit=1)[0])
Save File
Cancel