3 """Test binding to interfaces and addresses."""
7 import subprocess as subp
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
16 util.require_command("ss", "-nlup")
17 util.require_command("ip", "--json", "addr")
20 def connect_tcp(address: str, port: int) -> None:
21 """Check that a TCP connection to (address, port) works."""
23 family = socket.AF_INET if "." in address else socket.AF_INET6
25 with socket.socket(family, socket.SOCK_STREAM) as sock:
26 sock.connect((address, port))
29 def get_interfaces() -> T.List[T.Tuple[str, T.List[str]]]:
30 """Get a list of network interfaces with assigned addresses."""
33 ["ip", "--json", "addr"], check=True, encoding="utf-8", stdout=subp.PIPE
36 result: T.List[T.Tuple[str, T.List[str]]] = []
38 for line in json.loads(output):
39 if not "UP" in line["flags"]:
41 local: T.List[str] = []
42 for addr in line["addr_info"]:
43 if addr["family"] in ("inet", "inet6"):
44 local.append(addr["local"])
46 result.append((line["ifname"], local))
51 INTERFACES = get_interfaces()
54 def get_udp_listen(pid: int) -> T.List[str]:
55 """Get a list of the currently listening UDP sockets."""
57 listen = subp.run(["ss", "-nlup"], check=True, stdout=subp.PIPE, encoding="utf-8")
58 addresses: T.List[str] = []
60 for line in listen.stdout.splitlines():
61 if f"pid={pid}," in line:
62 _, _, _, addr, _ = line.split(maxsplit=4)
63 addresses.append(addr)
68 def test_bind_interface(ctx: Test) -> None:
69 """Test BindToInterface."""
71 devname, addresses = INTERFACES[0]
72 log.info("using interface %s, addresses (%s)", devname, addresses)
75 set BindToInterface {devname}
78 foo = ctx.node(init=init)
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)
85 log.info("check TCP sockets")
86 for addr in addresses:
87 connect_tcp(addr, foo.port)
90 def test_bind_address(ctx: Test, kind: str) -> None:
91 """Test BindToAddress or ListenAddress."""
93 _, addresses = INTERFACES[0]
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)
101 log.info("check for correct log message")
102 for addr in addresses:
103 check.in_file(foo.sub("log"), f"Listening on {addr}")
105 log.info("test TCP connections")
106 for addr in addresses:
107 connect_tcp(addr, foo.port)
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))
118 log.info("interface list is empty, skipping test")
121 with Test("test ListenAddress") as context:
122 test_bind_address(context, "ListenAddress")
124 with Test("test BindToAddress") as context:
125 test_bind_address(context, "BindToAddress")
127 with Test("test BindToInterface") as context:
128 test_bind_interface(context)