bundlewrap/libs/tools.py

93 lines
2.8 KiB
Python

from ipaddress import IPv4Address, IPv4Network, ip_address, ip_network
from bundlewrap.exceptions import BundleError, NoSuchGroup, NoSuchNode
from bundlewrap.utils.text import bold, red
from bundlewrap.utils.ui import io
def resolve_identifier(repo, identifier):
"""
Try to resolve an identifier (group or node). Return a set of ip
addresses valid for this identifier.
"""
try:
nodes = {repo.get_node(identifier)}
except NoSuchNode:
try:
nodes = repo.nodes_in_group(identifier)
except NoSuchGroup:
try:
ip = ip_network(identifier)
if isinstance(ip, IPv4Network):
return {'ipv4': {ip}, 'ipv6': set()}
else:
return {'ipv4': set(), 'ipv6': {ip}}
except ValueError:
try:
return repo.libs.firewall.named_networks[identifier]
except KeyError:
raise BundleError(
f'libs.tools.resolve_identifier(): Could not resolve {identifier}'
)
found_ips = set()
for node in nodes:
for interface, config in node.metadata.get('interfaces', {}).items():
for ip in config.get('ips', set()):
if '/' in ip:
found_ips.add(ip_address(ip.split('/')[0]))
else:
found_ips.add(ip_address(ip))
if node.metadata.get('external_ipv4', None):
found_ips.add(ip_address(node.metadata.get('external_ipv4')))
try:
found_ips.add(ip_address(node.hostname))
except ValueError:
pass
ip_dict = {
'ipv4': set(),
'ipv6': set(),
}
for ip in found_ips:
if isinstance(ip, IPv4Address):
ip_dict['ipv4'].add(ip)
else:
ip_dict['ipv6'].add(ip)
return ip_dict
def remove_more_specific_subnets(input_subnets) -> list:
final_subnets = []
for subnet in sorted(input_subnets):
source = ip_network(subnet)
if not source in final_subnets:
subnet_found = False
for dest_subnet in final_subnets:
if source.subnet_of(dest_subnet):
subnet_found = True
if not subnet_found:
final_subnets.append(source)
out = []
for net in final_subnets:
out.append(str(net))
return out
def require_bundle(node, bundle, hint=''):
# It's considered bad style to use assert statements outside of tests.
# That's why this little helper function exists, so we have an easy
# way of defining bundle requirements in other bundles.
if not node.has_bundle(bundle):
raise BundleError(f'{node.name} requires bundle {bundle}, but wasn\'t found! {hint}')