1 """Wrappers for running external commands."""
3 import subprocess as subp
8 from .util import random_string
10 _netns_created: T.Set[str] = set()
11 _iface_created: T.Set[str] = set()
14 def _cleanup() -> None:
15 for namespace in _netns_created.copy():
16 netns_delete(namespace)
18 # Ignore errors since device may have been moved to a different netns
19 for iface in _iface_created.copy():
20 subp.run(["ip", "link", "delete", iface], check=False)
23 atexit.register(_cleanup)
26 def _netns_action(action: str, namespace: str) -> bool:
27 log.debug("%s network namespace %s", action, namespace)
29 res = subp.run(["ip", "netns", action, namespace], check=False)
31 log.error("could not %s netns %s", action, namespace)
33 log.debug("OK %s netns %s", action, namespace)
35 return not res.returncode
38 def netns_delete(namespace: str) -> bool:
39 """Remove a previously created network namespace."""
40 success = _netns_action("delete", namespace)
42 _netns_created.remove(namespace)
46 def netns_add(namespace: str) -> bool:
47 """Add a network namespace (which can be removed manually or automatically at exit)."""
48 success = _netns_action("add", namespace)
50 _netns_created.add(namespace)
54 def netns_exec(netns: str, *args: str, check: bool = False) -> subp.CompletedProcess:
55 """Execute command in the network namespace."""
56 return subp.run(["ip", "netns", "exec", netns, *args], check=check)
59 def ping(address: str, netns: T.Optional[str] = None) -> bool:
60 """Ping the address from inside the network namespace."""
61 args = ["ping", "-l1", "-W1", "-i0.1", "-c10", address]
63 proc = netns_exec(netns, *args)
65 proc = subp.run(args, check=False)
66 return proc.returncode == 0
69 def move_dev(netns: str, device: str, ip_addr: str) -> None:
70 """Move device to the network namespace."""
71 if netns not in _netns_created:
73 subp.run(["ip", "link", "set", device, "netns", netns], check=True)
74 netns_exec(netns, "ip", "addr", "add", ip_addr, "dev", device, check=True)
75 netns_exec(netns, "ip", "link", "set", device, "up", check=True)
78 def veth_add(name0: str, name1: str) -> None:
79 """Create a veth link pair."""
81 ["ip", "link", "add", name0, "type", "veth", "peer", "name", name1], check=True
83 _iface_created.add(name0)
86 def link_add(link_type: str) -> str:
87 """Create a virtual link."""
88 name = random_string(10)
89 if link_type in ("tun", "tap"):
90 subp.run(["ip", "tuntap", "add", "mode", link_type, "dev", name], check=True)
92 subp.run(["ip", "link", "add", name, "type", link_type], check=True)
93 _iface_created.add(name)