| #!/usr/bin/python3 |
| |
| import argparse |
| import ipaddress |
| import json |
| import subprocess |
| import sys |
| import tempfile |
| |
| |
| class NetNamespace: |
| def __init__(self, name): |
| self._name = name |
| |
| def __enter__(self): |
| subprocess.check_call(['ip', 'netns', 'add', self._name]) |
| |
| def __exit__(self, *_): |
| subprocess.run(['ip', 'netns', 'del', self._name]) |
| |
| |
| class VethNetwork: |
| def __init__(self, ipv4_network, ipv6_network, local_iface, |
| other_ns, other_iface): |
| self._ip_networks = [ipv4_network, ipv6_network] |
| self._local_iface = local_iface |
| self._other_ns = other_ns |
| self._other_iface = other_iface |
| |
| def __enter__(self): |
| subprocess.check_call( |
| ['ip', 'link', 'add', self._local_iface, 'type', 'veth', |
| 'peer', 'name', self._other_iface]) |
| try: |
| subprocess.check_call( |
| ['ip', 'link', 'set', self._local_iface, 'up']) |
| for ip_network in self._ip_networks: |
| subprocess.check_call( |
| ['ip', 'addr', 'add', 'dev', self._local_iface, |
| f'{ ip_network[1] }/{ ip_network.prefixlen }']) |
| subprocess.check_call( |
| ['ip', 'link', 'set', self._other_iface, |
| 'netns', self._other_ns]) |
| except: |
| subprocess.run(['ip', 'link', 'del', self._local_iface]) |
| raise |
| |
| def __exit__(self, *_): |
| subprocess.run(['ip', 'link', 'del', self._local_iface]) |
| |
| |
| class DnsmasqRunning: |
| def __init__(self, ipv4_network, ipv6_network, iface): |
| self._iface = iface |
| self._ipv4_network = ipv4_network |
| self._ipv6_network = ipv6_network |
| assert self._ipv6_network.prefixlen == 64 # for SLAAC |
| |
| def __enter__(self): |
| config = tempfile.NamedTemporaryFile(mode='w') |
| try: |
| config_text = f'''\ |
| interface={ self._iface } |
| bind-interfaces |
| enable-ra |
| dhcp-range={ self._ipv4_network[2] },{ self._ipv4_network[-2] } |
| synth-domain=dhcp.example.com,{ self._ipv4_network[0] }/{ self._ipv4_network.prefixlen } |
| dhcp-option=option6:nis-domain,nisexample |
| dhcp-option=option6:domain-search,a.example.com,b.example.com |
| dhcp-option=option6:bootfile-url,tftp://boot.example.com/example |
| dhcp-range={ self._ipv6_network[2] },{ self._ipv6_network[-2] },slaac,64 |
| synth-domain=dhcpv6.example.com,{ self._ipv6_network[0] }/{ self._ipv6_network.prefixlen } |
| ''' |
| print('dnsmasq config:') |
| print(config_text) |
| config.write(config_text) |
| config.flush() |
| |
| self._dnsmasq_proc = subprocess.Popen( |
| ['dnsmasq', '-d', '-C', config.name, '--bootp-dynamic'], |
| stderr=sys.stderr) |
| self._config_file = config |
| except: |
| config.close() |
| raise |
| |
| def __exit__(self, *_): |
| self._dnsmasq_proc.terminate() |
| self._dnsmasq_proc.wait() |
| self._config_file.close() |
| |
| |
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument('command', nargs='+') |
| parser.add_argument('--ipv4-network', default='192.168.234.0/24') |
| parser.add_argument('--ipv6-network', default='fc00:c001:d00d:cafe::/64') |
| parser.add_argument('--server-iface', default='veth-dnsmasq') |
| parser.add_argument('--net-namespace', default='test-ipconfig') |
| parser.add_argument('--client-iface', default='veth-ipconfig') |
| args = parser.parse_args() |
| |
| ipv4_network = ipaddress.IPv4Network(args.ipv4_network) |
| ipv6_network = ipaddress.IPv6Network(args.ipv6_network) |
| with NetNamespace(args.net_namespace), \ |
| VethNetwork(ipv4_network, ipv6_network, args.server_iface, |
| args.net_namespace, args.client_iface), \ |
| DnsmasqRunning(ipv4_network, ipv6_network, args.server_iface): |
| subprocess.check_call( |
| ['ip', 'netns', 'exec', args.net_namespace] + args.command) |
| net_conf_name = f'/run/net-{ args.client_iface }.conf' |
| print(f'{net_conf_name}:') |
| subprocess.run(['cat', net_conf_name]) |
| print(f'ip addr:') |
| subprocess.check_call( |
| ['ip', 'netns', 'exec', args.net_namespace, 'ip', 'addr']) |
| |
| |
| if __name__ == '__main__': |
| main() |