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

   1 #!/usr/bin/env python3
   2
   3 """Test binding to interfaces and addresses."""
   4
   5 import json
   6 import socket
   7 import subprocess as subp
   8 import sys
   9 import typing as T
  10
  11 from testlib import check, util
  12 from testlib.const import EXIT_SKIP
  13 from testlib.log import log
  14 from testlib.test import Test
  15
  16 util.require_command("ss", "-nlup")
  17 util.require_command("ip", "--json", "addr")
  18
  19
  20 def connect_tcp(address: str, port: int) -> None:
  21     """Check that a TCP connection to (address, port) works."""
  22
  23     family = socket.AF_INET if "." in address else socket.AF_INET6
  24
  25     with socket.socket(family, socket.SOCK_STREAM) as sock:
  26         sock.connect((address, port))
  27
  28
  29 def get_interfaces() -> T.List[T.Tuple[str, T.List[str]]]:
  30     """Get a list of network interfaces with assigned addresses."""
  31
  32     output = subp.run(
  33         ["ip", "--json", "addr"], check=True, encoding="utf-8", stdout=subp.PIPE
  34     ).stdout
  35
  36     result: T.List[T.Tuple[str, T.List[str]]] = []
  37
  38     for line in json.loads(output):
  39         if not "UP" in line["flags"]:
  40             continue
  41         local: T.List[str] = []
  42         for addr in line["addr_info"]:
  43             if addr["family"] in ("inet", "inet6"):
  44                 local.append(addr["local"])
  45         if local:
  46             result.append((line["ifname"], local))
  47
  48     return result
  49
  50
  51 INTERFACES = get_interfaces()
  52
  53
  54 def get_udp_listen(pid: int) -> T.List[str]:
  55     """Get a list of the currently listening UDP sockets."""
  56
  57     listen = subp.run(["ss", "-nlup"], check=True, stdout=subp.PIPE, encoding="utf-8")
  58     addresses: T.List[str] = []
  59
  60     for line in listen.stdout.splitlines():
  61         if f"pid={pid}," in line:
  62             _, _, _, addr, _ = line.split(maxsplit=4)
  63             addresses.append(addr)
  64
  65     return addresses
  66
  67
  68 def test_bind_interface(ctx: Test) -> None:
  69     """Test BindToInterface."""
  70
  71     devname, addresses = INTERFACES[0]
  72     log.info("using interface %s, addresses (%s)", devname, addresses)
  73
  74     init = f"""
  75         set BindToInterface {devname}
  76         set LogLevel 5
  77     """
  78     foo = ctx.node(init=init)
  79     foo.start()
  80
  81     log.info("check that tincd opened UDP sockets")
  82     listen = get_udp_listen(foo.pid)
  83     check.is_in(f"%{devname}:{foo.port}", *listen)
  84
  85     log.info("check TCP sockets")
  86     for addr in addresses:
  87         connect_tcp(addr, foo.port)
  88
  89
  90 def test_bind_address(ctx: Test, kind: str) -> None:
  91     """Test BindToAddress or ListenAddress."""
  92
  93     _, addresses = INTERFACES[0]
  94
  95     log.info("create and start tincd node")
  96     foo = ctx.node(init="set LogLevel 10")
  97     for addr in addresses:
  98         foo.cmd("add", kind, addr)
  99     foo.start()
 100
 101     log.info("check for correct log message")
 102     for addr in addresses:
 103         check.in_file(foo.sub("log"), f"Listening on {addr}")
 104
 105     log.info("test TCP connections")
 106     for addr in addresses:
 107         connect_tcp(addr, foo.port)
 108
 109     log.info("check that tincd opened UDP sockets")
 110     listen = get_udp_listen(foo.pid)
 111     for addr in addresses:
 112         check.is_in(addr, *listen)
 113         check.is_in(f":{foo.port}", *listen)
 114     check.equals(len(addresses), len(listen))
 115
 116
 117 if not INTERFACES:
 118     log.info("interface list is empty, skipping test")
 119     sys.exit(EXIT_SKIP)
 120
 121 with Test("test ListenAddress") as context:
 122     test_bind_address(context, "ListenAddress")
 123
 124 with Test("test BindToAddress") as context:
 125     test_bind_address(context, "BindToAddress")
 126
 127 with Test("test BindToInterface") as context:
 128     test_bind_interface(context)