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

   1 #!/usr/bin/env python3
   2
   3 """Test binding to ports on localhost."""
   4
   5 import socket
   6 import sys
   7 import typing as T
   8
   9 from testlib import check, util
  10 from testlib.const import EXIT_SKIP
  11 from testlib.log import log
  12 from testlib.proc import Script
  13 from testlib.test import Test
  14
  15 # Call to close opened port
  16 Closer = T.Callable[[], None]
  17
  18
  19 def bind_random_port() -> T.Tuple[T.Optional[int], Closer]:
  20     """Bind to random port and return it, keeping the bind."""
  21     try:
  22         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  23         sock.bind(("127.0.0.1", 0))
  24         sock.listen()
  25         _, port = sock.getsockname()
  26         return port, sock.close
  27     except OSError:
  28         return None, sys.exit
  29
  30
  31 def test_bind_port(ctx: Test, ok_ports: T.List[int], bad_ports: T.List[int]) -> None:
  32     """Test binding to ports on localhost."""
  33
  34     foo = ctx.node(init="set LogLevel 1")
  35     foo.add_script(Script.TINC_UP)
  36     foo.add_script(Script.TINC_DOWN)
  37     log_path = foo.sub("log")
  38
  39     if ok_ports:
  40         log.info("check that tincd successfully binds to %s", ok_ports)
  41
  42         for port in ok_ports:
  43             foo.cmd("add", "BindToAddress", f"127.0.0.1 {port}")
  44
  45         proc = foo.tincd("-D")
  46         foo[Script.TINC_UP].wait()
  47         foo.cmd("stop")
  48         foo[Script.TINC_DOWN].wait()
  49         check.success(proc.wait())
  50
  51         foo_log = util.read_text(log_path)
  52
  53         for port in ok_ports:
  54             check.is_in(f"Listening on 127.0.0.1 port {port}", foo_log)
  55
  56     if bad_ports:
  57         log.info("check that tincd fails to bind to %s", bad_ports)
  58
  59         for port in bad_ports:
  60             foo.cmd("add", "BindToAddress", f"127.0.0.1 {port}")
  61
  62         util.remove_file(log_path)
  63         proc = foo.tincd("-D")
  64
  65         # Flush logs to the log file
  66         if ok_ports:
  67             foo[Script.TINC_UP].wait()
  68             foo.cmd("stop")
  69             foo[Script.TINC_DOWN].wait()
  70             check.success(proc.wait())
  71         else:
  72             check.failure(proc.wait())
  73
  74         foo_log = util.read_text(log_path)
  75
  76         for port in bad_ports:
  77             check.is_in(f"Can't bind to 127.0.0.1 port {port}", foo_log)
  78
  79         if not ok_ports:
  80             check.is_in("Unable to create any listening socket", foo_log)
  81
  82
  83 port0, close0 = bind_random_port()
  84 port1, close1 = bind_random_port()
  85
  86 if not port0 or not port1:
  87     log.info("could not bind ports, skipping test")
  88     sys.exit(EXIT_SKIP)
  89
  90 with Test("test binding with both ports unavailable") as context:
  91     test_bind_port(context, [], [port0, port1])
  92
  93 with Test("test binding to one free and one unavailable port") as context:
  94     close0()
  95     test_bind_port(context, [port0], [port1])
  96
  97 with Test("test binding to two free ports") as context:
  98     close1()
  99     test_bind_port(context, [port0, port1], [])