tinc-vpn.org Git - tinc/blob - test/integration/invite.py

   1 #!/usr/bin/env python3
   2 # pylint: disable=import-outside-toplevel
   3
   4 """Test tinc peer invitations."""
   5
   6 import time
   7 import subprocess as subp
   8
   9 from testlib import check, util
  10 from testlib.proc import Tinc
  11 from testlib.log import log
  12 from testlib.test import Test
  13
  14
  15 def run_port0_test(ctx: Test) -> None:
  16     """Checks that tinc invite fails if called with Port 0 and tincd stopped."""
  17     foo = ctx.node(init=True)
  18     _, err = foo.cmd("invite", "bar", code=1)
  19     check.is_in("Please start tincd", err)
  20
  21
  22 def init(ctx: Test) -> Tinc:
  23     """Initialize a node."""
  24     foo = ctx.node()
  25     stdin = f"""
  26         init {foo}
  27         set Port 12345
  28         set Address localhost
  29         set DeviceType dummy
  30         set Mode switch
  31         set Broadcast no
  32     """
  33     foo.cmd(stdin=stdin)
  34     return foo
  35
  36
  37 def run_expiration_test(ctx: Test) -> None:
  38     """Make sure that invites can't be used after expiration date."""
  39
  40     foo, bar = init(ctx), ctx.node()
  41     foo.cmd("set", "InvitationExpire", "1")
  42     foo.start()
  43
  44     url, _ = foo.cmd("invite", bar.name)
  45     url = url.strip()
  46     time.sleep(2)
  47
  48     try:
  49         bar.cmd("join", url, code=1, timeout=1)
  50     except subp.TimeoutExpired:
  51         pass
  52
  53     foo.cmd("stop")
  54     foo_log = util.read_text(foo.sub("log"))
  55     check.is_in("tried to use expired invitation", foo_log)
  56
  57
  58 def run_invite_test(ctx: Test, start_before_invite: bool) -> None:
  59     """Run tests. If start_before_invite is True,
  60     tincd is started *before* creating invitation, and vice versa.
  61     """
  62     foo = init(ctx)
  63     bar = ctx.node()
  64
  65     if start_before_invite:
  66         foo.cmd("set", "Port", "0")
  67         port = foo.start()
  68
  69     log.info("create invitation")
  70     foo_invite, _ = foo.cmd("invite", bar.name)
  71     assert foo_invite
  72     foo_invite = foo_invite.strip()
  73
  74     if not start_before_invite:
  75         foo.cmd("set", "Port", "0")
  76         port = foo.start()
  77         foo_invite = foo_invite.replace(":12345/", f":{port}/")
  78
  79     log.info("join second node with %s", foo_invite)
  80     bar.cmd("join", foo_invite)
  81     bar.cmd("set", "Port", "0")
  82
  83     if not start_before_invite:
  84         log.info("%s thinks %s is using port 0, updating", bar, foo)
  85         bar.cmd("set", f"{foo}.Port", str(port))
  86
  87     log.info("compare configs")
  88     check.files_eq(foo.sub("hosts", foo.name), bar.sub("hosts", foo.name))
  89
  90     log.info("compare keys")
  91
  92     prefix = "Ed25519PublicKey"
  93     foo_key = util.find_line(foo.sub("hosts", bar.name), prefix)
  94     bar_key = util.find_line(bar.sub("hosts", bar.name), prefix)
  95     check.equals(foo_key, bar_key)
  96
  97     log.info("checking Mode")
  98     bar_mode, _ = bar.cmd("get", "Mode")
  99     check.equals("switch", bar_mode.strip())
 100
 101     log.info("checking Broadcast")
 102     bar_bcast, _ = bar.cmd("get", "Broadcast")
 103     check.equals("no", bar_bcast.strip())
 104
 105     log.info("checking ConnectTo")
 106     bar_conn, _ = bar.cmd("get", "ConnectTo")
 107     check.equals(foo.name, bar_conn.strip())
 108
 109     log.info("configuring %s", bar.name)
 110     bar.cmd("set", "DeviceType", "dummy")
 111
 112     log.info("adding scripts")
 113     foo.add_script(bar.script_up)
 114     bar.add_script(foo.script_up)
 115
 116     log.info("starting %s", bar.name)
 117     bar.cmd("start")
 118
 119     log.info("waiting for nodes to come up")
 120     foo[bar.script_up].wait()
 121     bar[foo.script_up].wait()
 122
 123     log.info("checking required nodes")
 124     check.nodes(foo, 2)
 125     check.nodes(bar, 2)
 126
 127
 128 with Test("fail with Port 0 and tincd not running") as context:
 129     run_port0_test(context)
 130
 131 with Test("offline mode") as context:
 132     run_invite_test(context, start_before_invite=False)
 133
 134 with Test("online mode") as context:
 135     run_invite_test(context, start_before_invite=True)
 136
 137 with Test("invite expiration") as context:
 138     run_expiration_test(context)