bundlewrap/libs/tools.py

106 lines
3.2 KiB
Python
Raw Permalink Normal View History

2023-02-05 16:30:58 +00:00
from ipaddress import IPv4Address, IPv4Network, ip_address, ip_network
2020-10-16 20:19:45 +00:00
2023-02-05 16:30:58 +00:00
from bundlewrap.exceptions import BundleError, NoSuchGroup, NoSuchNode
from bundlewrap.utils.text import bold, red
from bundlewrap.utils.ui import io
2023-02-05 16:30:58 +00:00
def resolve_identifier(repo, identifier, linklocal=False, only_physical=False):
2020-10-16 20:19:45 +00:00
"""
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}'
)
2020-10-16 20:19:45 +00:00
found_ips = set()
for node in nodes:
for interface, config in node.metadata.get('interfaces', {}).items():
if only_physical and not (
interface.startswith('bond') or
interface.startswith('br') or
interface.startswith('en') or
interface.startswith('et') or
interface == 'default' # dummy nodes use these
):
continue
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))
2020-10-16 20:19:45 +00:00
if node.metadata.get('external_ipv4', None):
2020-10-16 20:19:45 +00:00
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 ip.is_link_local and not linklocal:
continue
if isinstance(ip, IPv4Address):
ip_dict['ipv4'].add(ip)
else:
ip_dict['ipv6'].add(ip)
return ip_dict
2021-04-01 14:27:31 +00:00
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}')