68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
import ipaddress
|
|
from operator import attrgetter
|
|
|
|
import dns.name
|
|
|
|
|
|
class FilterModule:
|
|
def filters(self):
|
|
return {
|
|
"add_origin": add_origin,
|
|
"add_origin_keys": add_origin_keys,
|
|
"ip_filter": ip_filter,
|
|
"remove_domain_suffix": remove_domain_suffix,
|
|
"ipaddr_sort": ipaddr_sort,
|
|
}
|
|
|
|
|
|
def first_addr(addresses, ipv4 = True):
|
|
version = ipaddress.IPv4Address if ipv4 else ipaddress.IPv6Address
|
|
for addr in addresses:
|
|
parsed = ipaddress.ip_address(xx)
|
|
if isinstance(parsed, version):
|
|
return parsed
|
|
raise ValueError("missing address")
|
|
|
|
|
|
def ip_filter(addresses, networks):
|
|
if isinstance(addresses, dict):
|
|
return {k: ip_filter(v, networks) for k, v in addresses.items()}
|
|
ip_networks = [ipaddress.ip_network(n) for n in networks]
|
|
ip_addresses = [ipaddress.ip_address(a) for a in addresses]
|
|
return [str(a) for a in ip_addresses if any(a in n for n in ip_networks)]
|
|
|
|
|
|
def add_origin(name, origin="."):
|
|
return dns.name.from_text(name, dns.name.from_text(origin)).to_text()
|
|
|
|
|
|
def add_origin_keys(dct, origin="."):
|
|
return {add_origin(k, origin): v for k, v in dct.items()}
|
|
|
|
|
|
def remove_domain_suffix(name):
|
|
parent = dns.name.from_text(name).parent()
|
|
return parent.to_text()
|
|
|
|
|
|
def ipaddr_sort(addrs, types, unknown_after=True):
|
|
check_types = {
|
|
"global": attrgetter("is_global"),
|
|
"link-local": attrgetter("is_link_local"),
|
|
"loopback": attrgetter("is_loopback"),
|
|
"multicast": attrgetter("is_multicast"),
|
|
"private": attrgetter("is_private"),
|
|
"reserved": attrgetter("is_reserved"),
|
|
"site_local": attrgetter("is_site_local"),
|
|
"unspecified": attrgetter("is_unspecified"),
|
|
}
|
|
|
|
def addr_weight(addr):
|
|
if isinstance(addr, str):
|
|
addr = ipaddress.ip_address(addr.split("/")[0])
|
|
for index, ty in enumerate(types):
|
|
if check_types[ty](ipaddress.ip_address(addr)):
|
|
return index
|
|
return len(types) if unknown_after else -1
|
|
|
|
return sorted(addrs, key=addr_weight)
|