diff --git a/oci/remote/allocate_private_network.py b/oci/remote/allocate_private_network.py index df846201..fa1de3c1 100644 --- a/oci/remote/allocate_private_network.py +++ b/oci/remote/allocate_private_network.py @@ -16,6 +16,47 @@ def command_output(*command: str) -> str: return result.stdout +def _interface_files() -> list[Path]: + """Include pending Proxmox network edits and down interface definitions.""" + paths = [Path('/etc/network/interfaces'), Path('/etc/network/interfaces.new')] + paths.extend(Path('/etc/network/interfaces.d').glob('*')) + return [path for path in paths if path.is_file()] + + +def _configured_networks(text: str) -> list[ipaddress.IPv4Network]: + networks = [] + addresses = [] + netmask = None + inet = False + + def collect(): + for address in addresses: + value = address if '/' in address else address+'/'+(netmask or '32') + try: + interface = ipaddress.ip_interface(value) + if interface.version == 4: + networks.append(interface.network) + except ValueError: + continue + + for line in text.splitlines(): + words = line.partition('#')[0].split() + if not words: + continue + if words[0] == 'iface': + collect() + addresses = [] + netmask = None + inet = len(words) >= 4 and words[2] == 'inet' + elif inet and len(words) >= 2: + if words[0] == 'address': + addresses.append(words[1]) + elif words[0] == 'netmask': + netmask = words[1] + collect() + return networks + + def existing_bridges() -> set[str]: bridges: set[str] = set() for line in command_output("ip", "-o", "link", "show").splitlines(): @@ -25,15 +66,13 @@ def existing_bridges() -> set[str]: for path in Path("/etc/pve/lxc").glob("*.conf"): text = path.read_text(encoding="utf-8", errors="ignore") bridges.update(re.findall(r"(?:^|,)bridge=([^,\s]+)", text, re.MULTILINE)) - interface_paths = [Path("/etc/network/interfaces")] - interface_paths.extend(Path("/etc/network/interfaces.d").glob("*")) - for path in interface_paths: - if not path.is_file(): - continue + for path in _interface_files(): text = path.read_text(encoding="utf-8", errors="ignore") - bridges.update( - re.findall(r"^(?:auto|iface)\s+(vmbr\d+)\b", text, re.MULTILINE) - ) + for line in text.splitlines(): + words = line.partition('#')[0].split() + if words and words[0] in {'auto', 'iface'}: + names = words[1:] if words[0] == 'auto' else words[1:2] + bridges.update(name for name in names if re.fullmatch(r'vmbr\d+', name)) return bridges @@ -54,6 +93,8 @@ def existing_networks() -> list[ipaddress.IPv4Network]: networks.append(ipaddress.ip_interface(address).network) except ValueError: pass + for path in _interface_files(): + networks.extend(_configured_networks(path.read_text(encoding='utf-8', errors='ignore'))) return networks diff --git a/oci/tests/test_private_network_configured.py b/oci/tests/test_private_network_configured.py new file mode 100644 index 00000000..7daa4800 --- /dev/null +++ b/oci/tests/test_private_network_configured.py @@ -0,0 +1,74 @@ +"""Down and pending bridge configs must reserve their names and IPv4 subnets.""" +import importlib.util +import ipaddress +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / 'remote')) +spec = importlib.util.spec_from_file_location('configured_private_network', ROOT / 'remote/allocate_private_network.py') +allocator = importlib.util.module_from_spec(spec) +spec.loader.exec_module(allocator) + + +def deployment(): + return {'network': {'private_allocation':'automatic', 'private_subnet':'10.77.0.0/24', + 'private_bridge':'vmbr10', 'private_host_address':'10.77.0.1/24'}} + + +class ConfiguredPrivateNetworkTests(unittest.TestCase): + def test_interface_files_include_pending_main_and_include_fragments(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root/'interfaces').write_text('auto vmbr0\n') + (root/'interfaces.new').write_text('auto vmbr10\n') + (root/'interfaces.d').mkdir() + (root/'interfaces.d/down').write_text('iface vmbr11 inet static\n') + (root/'interfaces.d/nested').mkdir() + def fake_path(value): + return root / Path(value).relative_to('/etc/network') + with patch.object(allocator, 'Path', side_effect=fake_path): + files = allocator._interface_files() + self.assertEqual(set(files), {root/'interfaces',root/'interfaces.new',root/'interfaces.d/down'}) + + def test_down_and_pending_subnets_are_occupied_even_with_different_bridge_names(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + down = root/'interfaces' + down.write_text('iface vmbr999 inet static\n address 10.77.0.1\n netmask 255.255.255.0\n') + pending = root/'interfaces.new' + pending.write_text('auto vmbr11\niface vmbr11 inet static\n address 10.77.1.1/24\n') + with patch.object(allocator, '_interface_files', return_value=[down,pending]), \ + patch.object(allocator, 'command_output', return_value=''), \ + patch.object(allocator.Path, 'glob', return_value=[]): + bridges = allocator.existing_bridges() + occupied = allocator.existing_networks() + self.assertIn('vmbr11', bridges) + self.assertIn(ipaddress.ip_network('10.77.0.0/24'), occupied) + self.assertIn(ipaddress.ip_network('10.77.1.0/24'), occupied) + self.assertEqual(allocator.allocate_network(deployment(),bridges,occupied), + ('vmbr12',ipaddress.ip_network('10.77.2.0/24'))) + + def test_pending_bridge_name_reserves_name_without_an_address(self): + with tempfile.TemporaryDirectory() as directory: + pending = Path(directory)/'interfaces.new' + pending.write_text('auto vmbr0 vmbr10\niface vmbr10 inet manual\n') + with patch.object(allocator, '_interface_files', return_value=[pending]), \ + patch.object(allocator, 'command_output', return_value=''), \ + patch.object(allocator.Path, 'glob', return_value=[]): + bridges = allocator.existing_bridges() + self.assertEqual(allocator.allocate_network(deployment(),bridges,[]), + ('vmbr11',ipaddress.ip_network('10.77.1.0/24'))) + + def test_ipv6_and_invalid_addresses_do_not_reserve_ipv4_pool(self): + text = ('iface vmbr12 inet6 static\n address fd00::1/64\n' + 'iface vmbr20 inet static\n address invalid\n' + 'iface vmbr21 inet static\n address 10.77.8.1/24 # comment\n') + self.assertEqual(allocator._configured_networks(text), [ipaddress.ip_network('10.77.8.0/24')]) + + +if __name__ == '__main__': + unittest.main()