1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
#!/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()
|
