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

   1 #!/usr/bin/env python3
   2
   3 """Test tinc protocol security."""
   4
   5 import asyncio
   6 import typing as T
   7
   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
  13 TIMEOUT = 2
  14
  15
  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)
  20         out.append(rec)
  21
  22
  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)
  27
  28     if delay:
  29         await asyncio.sleep(delay)
  30
  31     received: T.List[bytes] = []
  32     try:
  33         write.write(raw)
  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)
  38
  39     raise RuntimeError("test should not have reached this line")
  40
  41
  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)
  46     check.false(data)
  47
  48
  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"))
  54
  55
  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")
  60     check.false(data)
  61
  62
  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")
  67     check.false(data)
  68
  69
  70 async def test_null_metakey(foo: Tinc) -> None:
  71     """Test that NULL metakey is not accepted."""
  72     null_metakey = f"""
  73 0 {foo} 17.0\
  74 1 0 672 0 0 834188619F4D943FD0F4B1336F428BD4AC06171FEABA66BD2356BC9593F0ECD643F\
  75 0E4B748C670D7750DFDE75DC9F1D8F65AB1026F5ED2A176466FBA4167CC567A2085ABD070C1545B\
  76 180BDA86020E275EA9335F509C57786F4ED2378EFFF331869B856DDE1C05C461E4EECAF0E2FB97A\
  77 F77B7BC2AD1B34C12992E45F5D1254BBF0C3FB224ABB3E8859594A83B6CA393ED81ECAC9221CE6B\
  78 C71A727BCAD87DD80FC0834B87BADB5CB8FD3F08BEF90115A8DF1923D7CD9529729F27E1B8ABD83\
  79 C4CF8818AE10257162E0057A658E265610B71F9BA4B365A20C70578FAC65B51B91100392171BA12\
  80 A440A5E93C4AA62E0C9B6FC9B68F953514AAA7831B4B2C31C4
  81 """.strip()
  82
  83     log.info("no NULL METAKEY allowed")
  84     data = await send(foo.port, null_metakey)
  85     check.false(data)
  86
  87
  88 def init(ctx: Test) -> Tinc:
  89     """Initialize new test nodes."""
  90     foo = ctx.node()
  91
  92     stdin = f"""
  93         init {foo}
  94         set Port 0
  95         set DeviceType dummy
  96         set Address localhost
  97         set PingTimeout {TIMEOUT}
  98         set AutoConnect no
  99         set Subnet 10.96.96.1
 100     """
 101     foo.cmd(stdin=stdin)
 102
 103     foo.add_script(Script.SUBNET_UP)
 104     foo.start()
 105     foo[Script.SUBNET_UP].wait()
 106
 107     return foo
 108
 109
 110 async def run_tests(ctx: Test) -> None:
 111     """Run all tests."""
 112     foo = init(ctx)
 113
 114     log.info("getting into tarpit")
 115     await test_id_timeout(foo)
 116
 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),
 122     )
 123
 124
 125 loop = asyncio.get_event_loop()
 126
 127 with Test("security") as context:
 128     loop.run_until_complete(run_tests(context))