tinc-vpn.org Git - tinc/blob - test/integration/testlib/external.py

   1 """Wrappers for running external commands."""
   2
   3 import subprocess as subp
   4 import atexit
   5 import typing as T
   6
   7 from .log import log
   8 from .util import random_string
   9
  10 _netns_created: T.Set[str] = set()
  11 _iface_created: T.Set[str] = set()
  12
  13
  14 def _cleanup() -> None:
  15     for namespace in _netns_created.copy():
  16         netns_delete(namespace)
  17
  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)
  21
  22
  23 atexit.register(_cleanup)
  24
  25
  26 def _netns_action(action: str, namespace: str) -> bool:
  27     log.debug("%s network namespace %s", action, namespace)
  28
  29     res = subp.run(["ip", "netns", action, namespace], check=False)
  30     if res.returncode:
  31         log.error("could not %s netns %s", action, namespace)
  32     else:
  33         log.debug("OK %s netns %s", action, namespace)
  34
  35     return not res.returncode
  36
  37
  38 def netns_delete(namespace: str) -> bool:
  39     """Remove a previously created network namespace."""
  40     success = _netns_action("delete", namespace)
  41     if success:
  42         _netns_created.remove(namespace)
  43     return success
  44
  45
  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)
  49     if success:
  50         _netns_created.add(namespace)
  51     return success
  52
  53
  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)
  57
  58
  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]
  62     if netns:
  63         proc = netns_exec(netns, *args)
  64     else:
  65         proc = subp.run(args, check=False)
  66     return proc.returncode == 0
  67
  68
  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:
  72         netns_add(netns)
  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)
  76
  77
  78 def veth_add(name0: str, name1: str) -> None:
  79     """Create a veth link pair."""
  80     subp.run(
  81         ["ip", "link", "add", name0, "type", "veth", "peer", "name", name1], check=True
  82     )
  83     _iface_created.add(name0)
  84
  85
  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)
  91     else:
  92         subp.run(["ip", "link", "add", name, "type", link_type], check=True)
  93     _iface_created.add(name)
  94     return name