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
   8 from testlib import check, util
   9 from testlib.log import log
  10 from testlib.const import RUN_ACCESS_CHECKS
  11 from testlib.proc import Tinc
  12 from testlib.test import Test
  13
  14 FAKE_INVITE = "localhost:65535/pVOZMJGm3MqTvTu0UnhMGb2cfuqygiu79MdnERnGYdga5v8C"
  15
  16
  17 def init(ctx: Test) -> Tinc:
  18     """Initialize a node."""
  19
  20     node = ctx.node()
  21     stdin = f"""
  22         init {node}
  23         set Port 0
  24         set Address localhost
  25         set DeviceType dummy
  26     """
  27     node.cmd(stdin=stdin)
  28     return node
  29
  30
  31 def test_invite(foo: Tinc) -> None:
  32     """Test successful 'invite'."""
  33
  34     foo.cmd("set", "Mode", "switch")
  35     foo.cmd("set", "Broadcast", "mst")
  36     foo.start()
  37
  38     log.info("test successful invitation")
  39     out, _ = foo.cmd("invite", "quux")
  40     check.is_in(f"localhost:{foo.port}/", out)
  41
  42     for filename in os.listdir(foo.sub("invitations")):
  43         content = util.read_text(foo.sub(f"invitations/{filename}"))
  44         if filename == "ed25519_key.priv":
  45             check.is_in("-----BEGIN ED25519 PRIVATE KEY-----", content)
  46         else:
  47             check.is_in("Broadcast = mst", content)
  48             check.is_in("Mode = switch", content)
  49             check.is_in("Address = localhost", content)
  50             check.is_in("Name = quux", content)
  51             check.is_in(f"NetName = {foo}", content)
  52             check.is_in(f"ConnectTo = {foo}", content)
  53
  54
  55 def test_invite_errors(foo: Tinc) -> None:
  56     """Test invite error conditions."""
  57
  58     log.info("invite node with tincd stopped")
  59     _, err = foo.cmd("invite", "foobar", code=1)
  60     check.is_in("Could not open pid file", err)
  61
  62     log.info("start node %s", foo)
  63     foo.start()
  64
  65     log.info("invite without arguments")
  66     _, err = foo.cmd("invite", code=1)
  67     check.is_in("Not enough arguments", err)
  68
  69     log.info("invite with too many arguments")
  70     _, err = foo.cmd("invite", "foo", "bar", code=1)
  71     check.is_in("Too many arguments", err)
  72
  73     log.info("invite with invalid name")
  74     _, err = foo.cmd("invite", "!@#", code=1)
  75     check.is_in("Invalid name for node", err)
  76
  77     log.info("invite existing node")
  78     _, err = foo.cmd("invite", foo.name, code=1)
  79     check.is_in("already exists", err)
  80
  81     if RUN_ACCESS_CHECKS:
  82         log.info("bad permissions on invitations are fixed")
  83         invites = foo.sub("invitations")
  84         os.chmod(invites, 0)
  85         out, _ = foo.cmd("invite", "foobar")
  86         check.has_prefix(out, "localhost:")
  87
  88         log.info("invitations directory is created with bad permissions on parent")
  89         shutil.rmtree(invites)
  90         os.chmod(foo.work_dir, 0o500)
  91         out, _ = foo.cmd("invite", "foobar")
  92         check.has_prefix(out, "localhost:")
  93         check.true(os.access(invites, os.W_OK))
  94
  95         log.info("fully block access to configuration directory")
  96         work_dir = foo.sub("test_no_access")
  97         os.mkdir(work_dir, mode=0)
  98         _, err = foo.cmd("-c", work_dir, "invite", "foobar", code=1)
  99         check.is_in("Could not open", err)
 100
 101
 102 def test_join_errors(foo: Tinc) -> None:
 103     """Test join error conditions."""
 104
 105     log.info("try joining with redundant arguments")
 106     _, err = foo.cmd("join", "bar", "quux", code=1)
 107     check.is_in("Too many arguments", err)
 108
 109     log.info("try joining with existing configuration")
 110     _, err = foo.cmd("join", FAKE_INVITE, code=1)
 111     check.is_in("already exists", err)
 112
 113     log.info("try running without an invite URL")
 114     work_dir = foo.sub("test_no_invite")
 115     join = foo.tinc("-c", work_dir, "join")
 116     _, err = join.communicate(input="")
 117     check.equals(1, join.returncode)
 118     check.is_in("Error while reading", err)
 119
 120     log.info("try using an invalid invite")
 121     work_dir = foo.sub("test_invalid_invite")
 122     _, err = foo.cmd("-c", work_dir, "join", FAKE_INVITE, code=1)
 123     check.is_in("Could not connect to", err)
 124
 125     if RUN_ACCESS_CHECKS:
 126         log.info("bad permissions on configuration directory are fixed")
 127         work_dir = foo.sub("wd_access_test")
 128         os.mkdir(work_dir, mode=400)
 129         _, err = foo.cmd("-c", work_dir, "join", FAKE_INVITE, code=1)
 130         check.is_in("Could not connect to", err)
 131         check.true(os.access(work_dir, mode=os.W_OK))
 132
 133
 134 with Test("run invite success tests") as context:
 135     test_invite(init(context))
 136
 137 with Test("run invite error tests") as context:
 138     test_invite_errors(init(context))
 139
 140 with Test("run join tests") as context:
 141     test_join_errors(init(context))