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
12 from testlib.feature import SANDBOX_LEVEL
17 async def recv(read: asyncio.StreamReader, out: T.List[bytes]) -> None:
18 """Receive data until connection is closed."""
19 while not read.at_eof():
20 rec = await read.read(1)
24 async def send(port: int, buf: str, delay: float = 0) -> bytes:
25 """Send data and receive response."""
26 raw = f"{buf}\n".encode("utf-8")
27 read, write = await asyncio.open_connection(host="localhost", port=port)
30 await asyncio.sleep(delay)
32 received: T.List[bytes] = []
35 await asyncio.wait_for(recv(read, received), timeout=1)
36 except asyncio.TimeoutError:
37 log.info('received: "%s"', received)
38 return b"".join(received)
40 raise RuntimeError("test should not have reached this line")
43 async def test_id_timeout(foo: Tinc) -> None:
44 """Test that peer does not send its ID before us."""
45 log.info("no ID sent by peer if we don't send ID before the timeout")
46 data = await send(foo.port, "0 bar 17.7", delay=TIMEOUT * 1.5)
50 async def test_tarpitted(foo: Tinc) -> None:
51 """Test that peer sends its ID if we send first and are in tarpit."""
52 log.info("ID sent if initiator sends first, but still tarpitted")
53 data = await send(foo.port, "0 bar 17.7")
54 check.has_prefix(data, f"0 {foo} 17.7".encode("utf-8"))
57 async def test_invalid_id_own(foo: Tinc) -> None:
58 """Test that peer does not accept its own ID."""
59 log.info("own ID not allowed")
60 data = await send(foo.port, f"0 {foo} 17.7")
64 async def test_invalid_id_unknown(foo: Tinc) -> None:
65 """Test that peer does not accept unknown ID."""
66 log.info("no unknown IDs allowed")
67 data = await send(foo.port, "0 baz 17.7")
71 async def test_null_metakey(foo: Tinc) -> None:
72 """Test that NULL metakey is not accepted."""
75 1 0 672 0 0 834188619F4D943FD0F4B1336F428BD4AC06171FEABA66BD2356BC9593F0ECD643F\
76 0E4B748C670D7750DFDE75DC9F1D8F65AB1026F5ED2A176466FBA4167CC567A2085ABD070C1545B\
77 180BDA86020E275EA9335F509C57786F4ED2378EFFF331869B856DDE1C05C461E4EECAF0E2FB97A\
78 F77B7BC2AD1B34C12992E45F5D1254BBF0C3FB224ABB3E8859594A83B6CA393ED81ECAC9221CE6B\
79 C71A727BCAD87DD80FC0834B87BADB5CB8FD3F08BEF90115A8DF1923D7CD9529729F27E1B8ABD83\
80 C4CF8818AE10257162E0057A658E265610B71F9BA4B365A20C70578FAC65B51B91100392171BA12\
81 A440A5E93C4AA62E0C9B6FC9B68F953514AAA7831B4B2C31C4
84 log.info("no NULL METAKEY allowed")
85 data = await send(foo.port, null_metakey)
89 def init(ctx: Test) -> Tinc:
90 """Initialize new test nodes."""
98 set PingTimeout {TIMEOUT}
100 set Subnet 10.96.96.1
101 set Sandbox {SANDBOX_LEVEL}
105 foo.add_script(Script.SUBNET_UP)
107 foo[Script.SUBNET_UP].wait()
112 async def run_tests(ctx: Test) -> None:
116 log.info("getting into tarpit")
117 await test_id_timeout(foo)
119 log.info("starting other tests")
120 await asyncio.gather(
121 test_invalid_id_own(foo),
122 test_invalid_id_unknown(foo),
123 test_null_metakey(foo),
127 loop = asyncio.get_event_loop()
129 with Test("security") as context:
130 loop.run_until_complete(run_tests(context))