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 from testlib import check, util
   7 from testlib.log import log
   8 from testlib.test import Test
   9
  10
  11 def run_port0_test(ctx: Test) -> None:
  12     """Checks that tinc invite fails if called with Port 0 and tincd stopped."""
  13     foo = ctx.node()
  14     stdin = f"""
  15         init {foo}
  16         set Port 0
  17         set Address localhost
  18         set DeviceType dummy
  19     """
  20     foo.cmd(stdin=stdin)
  21     _, err = foo.cmd("invite", "bar", code=1)
  22     check.is_in("Please start tincd", err)
  23
  24
  25 def run_invite_test(ctx: Test, start_before_invite: bool) -> None:
  26     """Run tests. If start_before_invite is True,
  27     tincd is started *before* creating invitation, and vice versa.
  28     """
  29     foo, bar = ctx.node(), ctx.node()
  30     stdin = f"""
  31         init {foo}
  32         set Port 12345
  33         set Address localhost
  34         set DeviceType dummy
  35         set Mode switch
  36         set Broadcast no
  37     """
  38     foo.cmd(stdin=stdin)
  39
  40     if start_before_invite:
  41         foo.cmd("set", "Port", "0")
  42         port = foo.start()
  43
  44     log.info("create invitation")
  45     foo_invite, _ = foo.cmd("invite", bar.name)
  46     assert foo_invite
  47     foo_invite = foo_invite.strip()
  48
  49     if not start_before_invite:
  50         foo.cmd("set", "Port", "0")
  51         port = foo.start()
  52         foo_invite = foo_invite.replace(":12345/", f":{port}/")
  53
  54     log.info("join second node with %s", foo_invite)
  55     bar.cmd("join", foo_invite)
  56     bar.cmd("set", "Port", "0")
  57
  58     if not start_before_invite:
  59         log.info("%s thinks %s is using port 0, updating", bar, foo)
  60         bar.cmd("set", f"{foo}.Port", str(port))
  61
  62     log.info("compare configs")
  63     check.files_eq(foo.sub("hosts", foo.name), bar.sub("hosts", foo.name))
  64
  65     log.info("compare keys")
  66
  67     prefix = "Ed25519PublicKey"
  68     foo_key = util.find_line(foo.sub("hosts", bar.name), prefix)
  69     bar_key = util.find_line(bar.sub("hosts", bar.name), prefix)
  70     check.equals(foo_key, bar_key)
  71
  72     log.info("checking Mode")
  73     bar_mode, _ = bar.cmd("get", "Mode")
  74     check.equals("switch", bar_mode.strip())
  75
  76     log.info("checking Broadcast")
  77     bar_bcast, _ = bar.cmd("get", "Broadcast")
  78     check.equals("no", bar_bcast.strip())
  79
  80     log.info("checking ConnectTo")
  81     bar_conn, _ = bar.cmd("get", "ConnectTo")
  82     check.equals(foo.name, bar_conn.strip())
  83
  84     log.info("configuring %s", bar.name)
  85     bar.cmd("set", "DeviceType", "dummy")
  86
  87     log.info("adding scripts")
  88     foo.add_script(bar.script_up)
  89     bar.add_script(foo.script_up)
  90
  91     log.info("starting %s", bar.name)
  92     bar.cmd("start")
  93
  94     log.info("waiting for nodes to come up")
  95     foo[bar.script_up].wait()
  96     bar[foo.script_up].wait()
  97
  98     log.info("checking required nodes")
  99     check.nodes(foo, 2)
 100     check.nodes(bar, 2)
 101
 102
 103 with Test("fail with Port 0 and tincd not running") as context:
 104     run_port0_test(context)
 105
 106 with Test("offline mode") as context:
 107     run_invite_test(context, start_before_invite=False)
 108
 109 with Test("online mode") as context:
 110     run_invite_test(context, start_before_invite=True)