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

   1 #!/usr/bin/env python3
   2
   3 """Test legacy protocol support (tinc 1.0)."""
   4
   5 import typing as T
   6
   7 from testlib import check, cmd
   8 from testlib.log import log
   9 from testlib.proc import Tinc, Script
  10 from testlib.test import Test
  11
  12 TIMEOUT = 2
  13
  14
  15 def init(ctx: Test) -> T.Tuple[Tinc, Tinc]:
  16     """Initialize new test nodes."""
  17     foo, bar = ctx.node(), ctx.node()
  18
  19     stdin = f"""
  20         init {foo}
  21         set Port 0
  22         set DeviceType dummy
  23         set Address localhost
  24         add Subnet 10.98.98.1
  25         set PingTimeout {TIMEOUT}
  26     """
  27     foo.cmd(stdin=stdin)
  28     foo.start()
  29
  30     stdin = f"""
  31         init {bar}
  32         set Port 0
  33         set Address localhost
  34         set DeviceType dummy
  35         add Subnet 10.98.98.2
  36         set PingTimeout {TIMEOUT}
  37         set MaxTimeout {TIMEOUT}
  38     """
  39     bar.cmd(stdin=stdin)
  40
  41     cmd.exchange(foo, bar)
  42     bar.cmd("add", "ConnectTo", foo.name)
  43
  44     foo.add_script(bar.script_up)
  45     bar.add_script(foo.script_up)
  46
  47     return foo, bar
  48
  49
  50 def run_keys_test(foo: Tinc, bar: Tinc, empty: bool) -> None:
  51     """Check that EC public keys match the expected values."""
  52     bar.cmd("start")
  53
  54     foo[bar.script_up].wait()
  55     bar[foo.script_up].wait()
  56
  57     check.nodes(foo, 2)
  58     check.nodes(bar, 2)
  59
  60     foo_bar, _ = foo.cmd("get", f"{bar.name}.Ed25519PublicKey", code=None)
  61     log.info('got key foo/bar "%s"', foo_bar)
  62
  63     bar_foo, _ = bar.cmd("get", f"{foo.name}.Ed25519PublicKey", code=None)
  64     log.info('got key bar/foo "%s"', bar_foo)
  65
  66     assert not foo_bar == empty
  67     assert not bar_foo == empty
  68
  69
  70 with Test("foo 1.1, bar 1.1") as context:
  71     foo_node, bar_node = init(context)
  72     run_keys_test(foo_node, bar_node, empty=False)
  73
  74 with Test("foo 1.1, bar 1.0") as context:
  75     foo_node, bar_node = init(context)
  76     bar_node.cmd("set", "ExperimentalProtocol", "no")
  77     foo_node.cmd("del", f"{bar_node}.Ed25519PublicKey")
  78     bar_node.cmd("del", f"{foo_node}.Ed25519PublicKey")
  79     run_keys_test(foo_node, bar_node, empty=True)
  80
  81 with Test("bar 1.0 must not be allowed to connect") as context:
  82     foo_node, bar_node = init(context)
  83     bar_node.cmd("set", "ExperimentalProtocol", "no")
  84
  85     bar_up = bar_node.add_script(Script.SUBNET_UP)
  86     bar_node.cmd("start")
  87     bar_up.wait()
  88
  89     assert not foo_node[bar_node.script_up].wait(TIMEOUT * 2)
  90     check.nodes(foo_node, 1)
  91     check.nodes(bar_node, 1)