#! /usr/bin/python3 # SPDX-License-Identifier: GPL-2.0-or-later # # avocado/pasta.py - Set up test environments for pasta # # Copyright Red Hat # Author: David Gibson import ipaddress import itertools import json import os import os.path import common PASTA = './pasta' def base_config_net(x, ifname, ip4, net4, ip6, net6): x('ip link set lo up', sudo=True) x('ip -4 addr add {}/{} dev {}'.format(ip4, net4.prefixlen, ifname), sudo=True) x('ip -6 addr add {}/{} dev {}'.format(ip6, net6.prefixlen, ifname), sudo=True) x('ip link set {} up'.format(ifname), sudo=True) class PastaBaseTest(common.BaseTest): IFNAME = 'testif' IP4_NET = ipaddress.IPv4Network('192.0.2.0/24') (GW_IP4, OUTER_IP4) = itertools.islice(IP4_NET.hosts(), 2) IP6_NET = ipaddress.IPv6Network('2001:db8:9a55::/112') (GW_IP6, OUTER_IP6) = itertools.islice(IP6_NET.hosts(), 2) INBOUND_FWD = 10002 OUTBOUND_FWD = 10003 def outerx(self, cmd, **kwargs): return self.outer.system_output(cmd, **kwargs) def innerx(self, cmd, **kwargs): return self.inner.system_output(cmd, **kwargs) def setUp(self): super().setUp() # Create the testing namespaces self.sandbox = common.NsToolUnshare(self.workdir, 'sandbox', '-Ucnpf --mount-proc') self.gw = common.NsToolUnshare(self.workdir, 'gw', '-n', parent=self.sandbox) self.outer = common.NsToolUnshare(self.workdir, 'outer', '-n', parent=self.sandbox) self.inner = common.NsToolUnshare(self.workdir, 'inner', '-Ucnpf --mount-proc', parent=self.sandbox) # Create the link from gw <-> outer gw_ifname = 'gw-{}'.format(self.IFNAME) self.sandbox.system_output('ip link add {} type veth peer name {}' .format(self.IFNAME, gw_ifname), sudo=True) pid = self.gw.relative_pid(self.sandbox) self.sandbox.system_output( 'ip link set {} netns {}'.format(gw_ifname, pid), sudo=True) pid = self.outer.relative_pid(self.sandbox) self.sandbox.system_output( 'ip link set {} netns {}'.format(self.IFNAME, pid), sudo=True) # Basic network configuration on gw and outer base_config_net(self.gw.system_output, gw_ifname, self.GW_IP4, self.IP4_NET, self.GW_IP6, self.IP6_NET) base_config_net(self.outerx, self.IFNAME, self.OUTER_IP4, self.IP4_NET, self.OUTER_IP6, self.IP6_NET) # Get the gateway's link-local address self.gw_ll = common.slaac_wait(self.gw.system_output, gw_ifname) # Make gw the default route for outer self.outerx( 'ip -4 route add default via {}'.format(self.GW_IP4), sudo=True) self.outerx('ip -6 route add default via {} dev {}' .format(self.gw_ll, self.IFNAME), sudo=True) def tearDown(self): # PID namespace means shutting down the sandbox will kill # everything else del(self.sandbox) super().tearDown() class PastaUnconfiguredTest(PastaBaseTest): def setUp(self): super().setUp() pidfile = os.path.join(self.workdir, 'pasta.pid') relpid = self.inner.relative_pid(self.sandbox) pastacmd = ('{} -f -t {} -T {} -u {} -U {} -P {} {}' .format(PASTA, self.INBOUND_FWD, self.OUTBOUND_FWD, self.INBOUND_FWD, self.OUTBOUND_FWD, pidfile, relpid)) self.pasta = self.outer.subprocess(pastacmd) self.pasta.start() # Pasta is ready once the pid file is written while not os.path.exists(pidfile) or not open(pidfile).read(): pass # PID of pasta relative to outer ns self.pastapid = int(open(pidfile).read()) # Shut down pasta and check for errors def shutdown(self): # We can't use self.pasta.stop() or self.pasta.terminate() # because that will just kill the nstool and won't propagate # the signal to pasta itself self.outer.system_output('kill -TERM {}'.format(self.pastapid)) rc = self.pasta.wait(timeout=1.0) self.assertEqual(rc, 0) def test_ifname(self): ifs = json.loads(self.innerx('ip -j link show')) ifnames = set([ifi['ifname'] for ifi in ifs]) self.assertEquals(ifnames, set(('lo', self.IFNAME))) self.shutdown() class PastaConfiguredTest(PastaBaseTest): def setUp(self): super().setUp() pidfile = os.path.join(self.workdir, 'pasta.pid') relpid = self.inner.relative_pid(self.sandbox) pastacmd = ('{} -f -t {} -T {} -u {} -U {} -P {} --config-net {}' .format(PASTA, self.INBOUND_FWD, self.OUTBOUND_FWD, self.INBOUND_FWD, self.OUTBOUND_FWD, pidfile, relpid)) self.pasta = self.outer.subprocess(pastacmd) self.pasta.start() # Pasta is ready once the pid file is written while not os.path.exists(pidfile) or not open(pidfile).read(): pass # PID of pasta relative to sandbox ns self.pastapid = int(open(pidfile).read()) # Shut down pasta and check for errors def shutdown(self): # We can't use self.pasta.stop() or self.pasta.terminate() # because that will just kill the nstool and won't propagate # the signal to pasta itself self.sandbox.system_output('kill -TERM {}'.format(self.pastapid)) rc = self.pasta.wait(timeout=1.0) self.assertEqual(rc, 0) class ConfigNetTest(PastaConfiguredTest): def test_addr4(self): ifinfo = self.innerx('ip -4 -j addr show {}'.format(self.IFNAME)) ifinfo = json.loads(ifinfo)[0] self.assertEquals(len(ifinfo['addr_info']), 1) adinfo = ifinfo['addr_info'][0] addr = ipaddress.ip_address(adinfo['local']) self.assertEquals(addr, self.OUTER_IP4) self.shutdown() def test_addr6(self): ifinfo = self.innerx('ip -6 -j addr show {}'.format(self.IFNAME)) ifinfo = json.loads(ifinfo)[0] for adinfo in ifinfo['addr_info']: if adinfo['scope'] == 'global' and not 'dynamic' in adinfo: global_addr = ipaddress.ip_address(adinfo['local']) self.assertEquals(global_addr, self.OUTER_IP6) self.shutdown() def test_route4(self): rinfo = self.innerx('ip -j -4 route show') rinfo = json.loads(rinfo) for route in rinfo: if route['dst'] != 'default': continue gateway = ipaddress.ip_address(route['gateway']) self.assertEquals(gateway, self.GW_IP4) self.shutdown() def test_route6(self): rinfo = self.innerx('ip -j -6 route show') rinfo = json.loads(rinfo) for route in rinfo: if route['dst'] != 'default': continue gateway = ipaddress.ip_address(route['gateway']) self.assertEquals(gateway, self.gw_ll) self.shutdown() def test_mtu(self): ifinfo = self.innerx('ip -j link show {}'.format(self.IFNAME)) ifinfo = json.loads(ifinfo)[0] self.assertEquals(ifinfo['mtu'], 65520) self.shutdown()