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

   1 #!/usr/bin/env python3
   2
   3 """Test invite/join error conditions."""
   4
   5 import os
   6 import shutil
   7 import socket
   8
   9 from testlib import check, util
  10 from testlib.log import log
  11 from testlib.const import RUN_ACCESS_CHECKS
  12 from testlib.proc import Tinc, Script
  13 from testlib.test import Test
  14
  15 FAKE_INVITE = "localhost:65535/pVOZMJGm3MqTvTu0UnhMGb2cfuqygiu79MdnERnGYdga5v8C"
  16
  17
  18 def test_invite(foo: Tinc) -> None:
  19     """Test successful 'invite'."""
  20
  21     foo.cmd("set", "Mode", "switch")
  22     foo.cmd("set", "Broadcast", "mst")
  23     foo.start()
  24
  25     log.info("test successful invitation")
  26     out, _ = foo.cmd("invite", "quux")
  27     check.is_in(f"localhost:{foo.port}/", out)
  28
  29     for filename in os.listdir(foo.sub("invitations")):
  30         content = util.read_text(foo.sub(f"invitations/{filename}"))
  31         if filename == "ed25519_key.priv":
  32             check.is_in("-----BEGIN ED25519 PRIVATE KEY-----", content)
  33         else:
  34             check.is_in("Broadcast = mst", content)
  35             check.is_in("Mode = switch", content)
  36             check.is_in("Address = localhost", content)
  37             check.is_in("Name = quux", content)
  38             check.is_in(f"NetName = {foo}", content)
  39             check.is_in(f"ConnectTo = {foo}", content)
  40
  41
  42 def test_invite_errors(foo: Tinc) -> None:
  43     """Test invite error conditions."""
  44
  45     log.info("invite node with tincd stopped")
  46     _, err = foo.cmd("invite", "foobar", code=1)
  47     check.is_in("Could not open pid file", err)
  48
  49     log.info("start node %s", foo)
  50     foo.start()
  51
  52     log.info("invite without arguments")
  53     _, err = foo.cmd("invite", code=1)
  54     check.is_in("Not enough arguments", err)
  55
  56     log.info("invite with too many arguments")
  57     _, err = foo.cmd("invite", "foo", "bar", code=1)
  58     check.is_in("Too many arguments", err)
  59
  60     log.info("invite with invalid name")
  61     _, err = foo.cmd("invite", "!@#", code=1)
  62     check.is_in("Invalid name for node", err)
  63
  64     log.info("invite existing node")
  65     _, err = foo.cmd("invite", foo.name, code=1)
  66     check.is_in("already exists", err)
  67
  68     if RUN_ACCESS_CHECKS:
  69         log.info("bad permissions on invitations are fixed")
  70         invites = foo.sub("invitations")
  71         os.chmod(invites, 0)
  72         out, _ = foo.cmd("invite", "foobar")
  73         check.has_prefix(out, "localhost:")
  74
  75         log.info("invitations directory is created with bad permissions on parent")
  76         shutil.rmtree(invites)
  77         os.chmod(foo.work_dir, 0o500)
  78         out, _ = foo.cmd("invite", "foobar")
  79         check.has_prefix(out, "localhost:")
  80         check.true(os.access(invites, os.W_OK))
  81
  82         log.info("fully block access to configuration directory")
  83         work_dir = foo.sub("test_no_access")
  84         os.mkdir(work_dir, mode=0)
  85         _, err = foo.cmd("-c", work_dir, "invite", "foobar", code=1)
  86         check.is_in("Could not open", err)
  87
  88
  89 def test_join_errors(foo: Tinc) -> None:
  90     """Test join error conditions."""
  91
  92     log.info("try joining with redundant arguments")
  93     _, err = foo.cmd("join", "bar", "quux", code=1)
  94     check.is_in("Too many arguments", err)
  95
  96     log.info("try joining with existing configuration")
  97     _, err = foo.cmd("join", FAKE_INVITE, code=1)
  98     check.is_in("already exists", err)
  99
 100     log.info("try running without an invite URL")
 101     work_dir = foo.sub("test_no_invite")
 102     join = foo.tinc("-c", work_dir, "join")
 103     _, err = join.communicate(input="")
 104     check.equals(1, join.returncode)
 105     check.is_in("Error while reading", err)
 106
 107     log.info("try using an invalid invite")
 108     work_dir = foo.sub("test_invalid_invite")
 109     _, err = foo.cmd("-c", work_dir, "join", FAKE_INVITE, code=1)
 110     check.is_in("Could not connect to", err)
 111
 112     if RUN_ACCESS_CHECKS:
 113         log.info("bad permissions on configuration directory are fixed")
 114         work_dir = foo.sub("wd_access_test")
 115         os.mkdir(work_dir, mode=400)
 116         _, err = foo.cmd("-c", work_dir, "join", FAKE_INVITE, code=1)
 117         check.is_in("Could not connect to", err)
 118         check.true(os.access(work_dir, mode=os.W_OK))
 119
 120
 121 def resolve(address: str) -> bool:
 122     """Try to resolve domain and return True if successful."""
 123     try:
 124         return len(socket.gethostbyname(address)) > 0
 125     except socket.gaierror:
 126         return False
 127
 128
 129 def test_broken_invite(ctx: Test) -> None:
 130     """Test joining using a broken invitation."""
 131
 132     foo, bar = ctx.node(init="set Address 127.0.0.1"), ctx.node()
 133     foo.start()
 134
 135     for url in (
 136         "localhost",
 137         "localhost/" + ("x" * 47),
 138         "localhost/" + ("x" * 49),
 139         "[::1/QWNVAevHNSHyMk1qarlZAQOB5swl3Ptu1yGCMSZrzKWpBUMv",
 140     ):
 141         _, err = bar.cmd("join", url, code=1)
 142         check.is_in("Invalid invitation URL", err)
 143
 144     # This can fail for those with braindead DNS servers that resolve
 145     # everything to show spam search results.
 146     # https://datatracker.ietf.org/doc/html/rfc6761#section-6.4
 147     if not resolve("tinc.invalid"):
 148         log.info("test invitation with an invalid domain")
 149         url = "tinc.invalid/QWNVAevHNSHyMk1qarlZAQOB5swl3Ptu1yGCMSZrzKWpBUMv"
 150         _, err = bar.cmd("join", url, code=1)
 151         check.is_in("Error looking up tinc.invalid", err)
 152
 153     timeout_err = "Timed out waiting for the server"
 154     conn_err = "Could not connect to inviter"
 155     server_err = "Please try again"
 156
 157     bad_url = f"127.0.0.1:{foo.port}/jkhjAi0LGVP0o6TN7aa_7xjqM9qTb_DUxBpk6UuLEF4ubDLX"
 158
 159     log.info("test invitation created by another server before invite is created")
 160     _, err = bar.cmd("join", bad_url, code=1, timeout=10)
 161     check.is_in(timeout_err, err)
 162     check.is_in(conn_err, err)
 163
 164     url, _ = foo.cmd("invite", "bar")
 165     url = url.strip()
 166
 167     log.info("test invitation created by another server after invite is created")
 168     _, err = bar.cmd("join", bad_url, code=1)
 169     check.is_in("Peer has an invalid key", err)
 170
 171     log.info("remove invitation directory")
 172     shutil.rmtree(foo.sub("invitations"))
 173
 174     log.info("test when invitation file is missing")
 175     _, err = bar.cmd("join", url, code=1, timeout=10)
 176     check.is_in(timeout_err, err)
 177     check.is_in(server_err, err)
 178
 179     foo.add_script(Script.TINC_DOWN)
 180     foo.cmd("stop")
 181     foo[Script.TINC_DOWN].wait()
 182
 183     foo_log = util.read_text(foo.sub("log"))
 184     check.is_in("we don't have an invitation key", foo_log)
 185     check.is_in("tried to use non-existing invitation", foo_log)
 186
 187
 188 with Test("run invite success tests") as context:
 189     test_invite(context.node(init=True))
 190
 191 with Test("run invite error tests") as context:
 192     test_invite_errors(context.node(init=True))
 193
 194 with Test("run join tests") as context:
 195     test_join_errors(context.node(init=True))
 196
 197 with Test("broken invitation") as context:
 198     test_broken_invite(context)