3 """Test tinc protocol security."""
8 from testlib import check
9 from testlib.log import log
10 from testlib.proc import Tinc, Script
11 from testlib.test import Test
16 async def recv(read: asyncio.StreamReader, out: T.List[bytes]) -> None:
17 """Receive data until connection is closed."""
18 while not read.at_eof():
19 rec = await read.read(1)
23 async def send(port: int, buf: str, delay: float = 0) -> bytes:
24 """Send data and receive response."""
25 raw = f"{buf}\n".encode("utf-8")
26 read, write = await asyncio.open_connection(host="localhost", port=port)
29 await asyncio.sleep(delay)
31 received: T.List[bytes] = []
34 await asyncio.wait_for(recv(read, received), timeout=1)
35 except asyncio.TimeoutError:
36 log.info('received: "%s"', received)
37 return b"".join(received)
39 raise RuntimeError("test should not have reached this line")
42 async def test_id_timeout(foo: Tinc) -> None:
43 """Test that peer does not send its ID before us."""
44 log.info("no ID sent by peer if we don't send ID before the timeout")
45 data = await send(foo.port, "0 bar 17.7", delay=TIMEOUT * 1.5)
49 async def test_tarpitted(foo: Tinc) -> None:
50 """Test that peer sends its ID if we send first and are in tarpit."""
51 log.info("ID sent if initiator sends first, but still tarpitted")
52 data = await send(foo.port, "0 bar 17.7")
53 check.has_prefix(data, f"0 {foo} 17.7".encode("utf-8"))
56 async def test_invalid_id_own(foo: Tinc) -> None:
57 """Test that peer does not accept its own ID."""
58 log.info("own ID not allowed")
59 data = await send(foo.port, f"0 {foo} 17.7")
63 async def test_invalid_id_unknown(foo: Tinc) -> None:
64 """Test that peer does not accept unknown ID."""
65 log.info("no unknown IDs allowed")
66 data = await send(foo.port, "0 baz 17.7")
70 async def test_null_metakey(foo: Tinc) -> None:
71 """Test that NULL metakey is not accepted."""
74 1 0 672 0 0 834188619F4D943FD0F4B1336F428BD4AC06171FEABA66BD2356BC9593F0ECD643F\
75 0E4B748C670D7750DFDE75DC9F1D8F65AB1026F5ED2A176466FBA4167CC567A2085ABD070C1545B\
76 180BDA86020E275EA9335F509C57786F4ED2378EFFF331869B856DDE1C05C461E4EECAF0E2FB97A\
77 F77B7BC2AD1B34C12992E45F5D1254BBF0C3FB224ABB3E8859594A83B6CA393ED81ECAC9221CE6B\
78 C71A727BCAD87DD80FC0834B87BADB5CB8FD3F08BEF90115A8DF1923D7CD9529729F27E1B8ABD83\
79 C4CF8818AE10257162E0057A658E265610B71F9BA4B365A20C70578FAC65B51B91100392171BA12\
80 A440A5E93C4AA62E0C9B6FC9B68F953514AAA7831B4B2C31C4
83 log.info("no NULL METAKEY allowed")
84 data = await send(foo.port, null_metakey)
88 def init(ctx: Test) -> Tinc:
89 """Initialize new test nodes."""
97 set PingTimeout {TIMEOUT}
103 foo.add_script(Script.SUBNET_UP)
105 foo[Script.SUBNET_UP].wait()
110 async def run_tests(ctx: Test) -> None:
114 log.info("getting into tarpit")
115 await test_id_timeout(foo)
117 log.info("starting other tests")
118 await asyncio.gather(
119 test_invalid_id_own(foo),
120 test_invalid_id_unknown(foo),
121 test_null_metakey(foo),
125 loop = asyncio.get_event_loop()
127 with Test("security") as context:
128 loop.run_until_complete(run_tests(context))