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

   1 """Classes for working with compiled instances of tinc and tincd binaries."""
   2
   3 import os
   4 import random
   5 import tempfile
   6 import typing as T
   7 import subprocess as subp
   8 from enum import Enum
   9 from platform import system
  10
  11 from . import check, path
  12 from .log import log
  13 from .script import TincScript, Script, ScriptType
  14 from .template import make_script, make_cmd_wrap
  15 from .util import random_string, random_port
  16
  17 # Does the OS support all addresses in 127.0.0.0/8 without additional configuration?
  18 _FULL_LOCALHOST_SUBNET = system() in ("Linux", "Windows")
  19
  20 # Path to the system temporary directory.
  21 _TEMPDIR = tempfile.gettempdir()
  22
  23
  24 def _make_wd(name: str) -> str:
  25     work_dir = os.path.join(path.TEST_WD, "data", name)
  26     os.makedirs(work_dir, exist_ok=True)
  27     return work_dir
  28
  29
  30 def _random_octet() -> int:
  31     return random.randint(1, 254)
  32
  33
  34 def _rand_localhost() -> str:
  35     """Generate random IP in subnet 127.0.0.0/8 for operating systems that support
  36     it without additional configuration. For all others, return 127.0.0.1.
  37     """
  38     if _FULL_LOCALHOST_SUBNET:
  39         return f"127.{_random_octet()}.{_random_octet()}.{_random_octet()}"
  40     return "127.0.0.1"
  41
  42
  43 class Feature(Enum):
  44     """Optional features supported by both tinc and tincd."""
  45
  46     COMP_LZ4 = "comp_lz4"
  47     COMP_LZO = "comp_lzo"
  48     COMP_ZLIB = "comp_zlib"
  49     CURSES = "curses"
  50     JUMBOGRAMS = "jumbograms"
  51     LEGACY_PROTOCOL = "legacy_protocol"
  52     LIBGCRYPT = "libgcrypt"
  53     MINIUPNPC = "miniupnpc"
  54     OPENSSL = "openssl"
  55     READLINE = "readline"
  56     TUNEMU = "tunemu"
  57     SANDBOX = "sandbox"
  58     UML = "uml"
  59     VDE = "vde"
  60
  61
  62 class Tinc:
  63     """Thin wrapper around Popen that simplifies running tinc/tincd
  64     binaries by passing required arguments, checking exit codes, etc.
  65     """
  66
  67     name: str
  68     address: str
  69     _work_dir: str
  70     _pid_file: str
  71     _port: T.Optional[int]
  72     _scripts: T.Dict[str, TincScript]
  73     _procs: T.List[subp.Popen]
  74
  75     def __init__(self, name: str = "", addr: str = "") -> None:
  76         self.name = name if name else random_string(10)
  77         self.address = addr if addr else _rand_localhost()
  78         self._work_dir = _make_wd(self.name)
  79         os.makedirs(self._work_dir, exist_ok=True)
  80         self._pid_file = os.path.join(_TEMPDIR, f"tinc_{self.name}")
  81         self._port = None
  82         self._scripts = {}
  83         self._procs = []
  84
  85     def randomize_port(self) -> int:
  86         """Use a random port for this node."""
  87         self._port = random_port()
  88         return self._port
  89
  90     @property
  91     def pid_file(self) -> str:
  92         """Get the path to the pid file."""
  93         return self._pid_file
  94
  95     def read_port(self) -> int:
  96         """Read port used by tincd from its pidfile and update the _port field."""
  97         log.debug("reading pidfile at %s", self.pid_file)
  98
  99         with open(self.pid_file, "r", encoding="utf-8") as f:
 100             content = f.read()
 101         log.debug("found data %s", content)
 102
 103         _, _, _, token, port = content.split()
 104         check.equals("port", token)
 105
 106         self._port = int(port)
 107         return self._port
 108
 109     @property
 110     def port(self) -> int:
 111         """Port that tincd is listening on."""
 112         assert self._port is not None
 113         return self._port
 114
 115     def __str__(self) -> str:
 116         return self.name
 117
 118     def __getitem__(self, script: ScriptType) -> TincScript:
 119         if isinstance(script, Script):
 120             script = script.name
 121         return self._scripts[script]
 122
 123     def __enter__(self):
 124         return self
 125
 126     def __exit__(self, exc_type, exc_val, exc_tb):
 127         self.cleanup()
 128
 129     @property
 130     def features(self) -> T.List[Feature]:
 131         """List of features supported by tinc and tincd."""
 132         tinc, _ = self.cmd("--version")
 133         tincd, _ = self.tincd("--version").communicate(timeout=5)
 134         prefix, features = "Features: ", []
 135
 136         for out in tinc, tincd:
 137             for line in out.splitlines():
 138                 if not line.startswith(prefix):
 139                     continue
 140                 tokens = line[len(prefix) :].split()
 141                 for token in tokens:
 142                     features.append(Feature(token))
 143                 break
 144
 145         log.info('supported features: "%s"', features)
 146         return features
 147
 148     @property
 149     def _common_args(self) -> T.List[str]:
 150         return [
 151             "--net",
 152             self.name,
 153             "--config",
 154             self.work_dir,
 155             "--pidfile",
 156             self.pid_file,
 157         ]
 158
 159     def sub(self, *paths: str) -> str:
 160         """Return path to a subdirectory within the working dir for this node."""
 161         return os.path.join(self._work_dir, *paths)
 162
 163     @property
 164     def work_dir(self):
 165         """Node's working directory."""
 166         return self._work_dir
 167
 168     @property
 169     def script_up(self) -> str:
 170         """Name of the hosts/XXX-up script for this node."""
 171         return f"hosts/{self.name}-up"
 172
 173     @property
 174     def script_down(self) -> str:
 175         """Name of the hosts/XXX-down script for this node."""
 176         return f"hosts/{self.name}-down"
 177
 178     def cleanup(self) -> None:
 179         """Terminate all tinc and tincd processes started from this instance."""
 180         log.info("running node cleanup for %s", self)
 181
 182         try:
 183             self.cmd("stop")
 184         except (AssertionError, ValueError):
 185             log.info("unsuccessfully tried to stop node %s", self)
 186
 187         for proc in self._procs:
 188             if proc.returncode is not None:
 189                 log.debug("PID %d exited, skipping", proc.pid)
 190             else:
 191                 log.info("PID %d still running, stopping", proc.pid)
 192                 try:
 193                     proc.kill()
 194                 except OSError as ex:
 195                     log.error("could not kill PID %d", proc.pid, exc_info=ex)
 196
 197             log.debug("waiting on %d to prevent zombies", proc.pid)
 198             try:
 199                 proc.wait()
 200             except OSError as ex:
 201                 log.error("waiting on %d failed", proc.pid, exc_info=ex)
 202
 203         self._procs.clear()
 204
 205     def start(self, *args: str) -> int:
 206         """Start the node, wait for it to call tinc-up, and get the port it's
 207         listening on from the pid file. Don't use this method unless you need
 208         to know the port tincd is running on. Call .cmd("start"), it's faster.
 209
 210         Reading pidfile and setting the port cannot be done from tinc-up because
 211         you can't send tinc commands to yourself there — the daemon doesn't
 212         respond to them until tinc-up is finished. The port field on this Tinc
 213         instance is updated to reflect the correct port. If tinc-up is missing,
 214         this command creates a new one, and then disables it.
 215         """
 216         new_script = Script.TINC_UP.name not in self._scripts
 217         if new_script:
 218             self.add_script(Script.TINC_UP)
 219
 220         tinc_up = self[Script.TINC_UP]
 221         self.cmd(*args, "start")
 222         tinc_up.wait()
 223
 224         if new_script:
 225             tinc_up.disable()
 226
 227         self._port = self.read_port()
 228         self.cmd("set", "Port", str(self._port))
 229
 230         return self._port
 231
 232     def cmd(
 233         self, *args: str, code: T.Optional[int] = 0, stdin: T.Optional[T.AnyStr] = None
 234     ) -> T.Tuple[str, str]:
 235         """Run command through tinc, writes `stdin` to it (if the argument is not None),
 236         check its return code (if the argument is not None), and return (stdout, stderr).
 237         """
 238         proc = self.tinc(*args, binary=isinstance(stdin, bytes))
 239         log.debug('tinc %s: PID %d, in "%s", want code %s', self, proc.pid, stdin, code)
 240
 241         out, err = proc.communicate(stdin, timeout=60)
 242         res = proc.returncode
 243         self._procs.remove(proc)
 244         log.debug('tinc %s: code %d, out "%s", err "%s"', self, res, out, err)
 245
 246         if code is not None:
 247             check.equals(code, res)
 248
 249         return out if out else "", err if err else ""
 250
 251     def tinc(self, *args: str, binary=False) -> subp.Popen:
 252         """Start tinc with the specified arguments."""
 253         args = tuple(filter(bool, args))
 254         cmd = [path.TINC_PATH, *self._common_args, *args]
 255         log.debug('starting tinc %s: "%s"', self.name, " ".join(cmd))
 256         # pylint: disable=consider-using-with
 257         proc = subp.Popen(
 258             cmd,
 259             cwd=self._work_dir,
 260             stdin=subp.PIPE,
 261             stdout=subp.PIPE,
 262             stderr=subp.PIPE,
 263             encoding=None if binary else "utf-8",
 264         )
 265         self._procs.append(proc)
 266         return proc
 267
 268     def tincd(self, *args: str, env: T.Optional[T.Dict[str, str]] = None) -> subp.Popen:
 269         """Start tincd with the specified arguments."""
 270         args = tuple(filter(bool, args))
 271         cmd = [
 272             path.TINCD_PATH,
 273             *self._common_args,
 274             "--logfile",
 275             self.sub("log"),
 276             "-d5",
 277             *args,
 278         ]
 279         log.debug('starting tincd %s: "%s"', self.name, " ".join(cmd))
 280         if env is not None:
 281             env = {**os.environ, **env}
 282         # pylint: disable=consider-using-with
 283         proc = subp.Popen(
 284             cmd,
 285             cwd=self._work_dir,
 286             stdin=subp.PIPE,
 287             stdout=subp.PIPE,
 288             stderr=subp.PIPE,
 289             encoding="utf-8",
 290             env=env,
 291         )
 292         self._procs.append(proc)
 293         return proc
 294
 295     def add_script(self, script: ScriptType, source: str = "") -> TincScript:
 296         """Create a script with the passed Python source code.
 297         The source must either be empty, or start indentation with 4 spaces.
 298         If the source is empty, the created script can be used to receive notifications.
 299         """
 300         rel_path = script if isinstance(script, str) else script.value
 301         check.not_in(rel_path, self._scripts)
 302
 303         full_path = os.path.join(self._work_dir, rel_path)
 304         tinc_script = TincScript(self.name, rel_path, full_path)
 305
 306         log.debug("creating script %s at %s", script, full_path)
 307         with open(full_path, "w", encoding="utf-8") as f:
 308             content = make_script(self.name, rel_path, source)
 309             f.write(content)
 310
 311         if os.name == "nt":
 312             log.debug("creating .cmd script wrapper at %s", full_path)
 313             win_content = make_cmd_wrap(full_path)
 314             with open(f"{full_path}.cmd", "w", encoding="utf-8") as f:
 315                 f.write(win_content)
 316         else:
 317             os.chmod(full_path, 0o755)
 318
 319         if isinstance(script, Script):
 320             self._scripts[script.name] = tinc_script
 321         self._scripts[rel_path] = tinc_script
 322
 323         return tinc_script