3 """Test invite/join error conditions."""
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
15 FAKE_INVITE = "localhost:65535/pVOZMJGm3MqTvTu0UnhMGb2cfuqygiu79MdnERnGYdga5v8C"
18 def test_invite(foo: Tinc) -> None:
19 """Test successful 'invite'."""
21 foo.cmd("set", "Mode", "switch")
22 foo.cmd("set", "Broadcast", "mst")
25 log.info("test successful invitation")
26 out, _ = foo.cmd("invite", "quux")
27 check.is_in(f"localhost:{foo.port}/", out)
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)
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)
42 def test_invite_errors(foo: Tinc) -> None:
43 """Test invite error conditions."""
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)
49 log.info("start node %s", foo)
52 log.info("invite without arguments")
53 _, err = foo.cmd("invite", code=1)
54 check.is_in("Not enough arguments", err)
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)
60 log.info("invite with invalid name")
61 _, err = foo.cmd("invite", "!@#", code=1)
62 check.is_in("Invalid name for node", err)
64 log.info("invite existing node")
65 _, err = foo.cmd("invite", foo.name, code=1)
66 check.is_in("already exists", err)
69 log.info("bad permissions on invitations are fixed")
70 invites = foo.sub("invitations")
72 out, _ = foo.cmd("invite", "foobar")
73 check.has_prefix(out, "localhost:")
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))
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)
89 def test_join_errors(foo: Tinc) -> None:
90 """Test join error conditions."""
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)
96 log.info("try joining with existing configuration")
97 _, err = foo.cmd("join", FAKE_INVITE, code=1)
98 check.is_in("already exists", err)
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)
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)
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))
121 def resolve(address: str) -> bool:
122 """Try to resolve domain and return True if successful."""
124 return len(socket.gethostbyname(address)) > 0
125 except socket.gaierror:
129 def test_broken_invite(ctx: Test) -> None:
130 """Test joining using a broken invitation."""
132 foo, bar = ctx.node(init="set Address 127.0.0.1"), ctx.node()
137 "localhost/" + ("x" * 47),
138 "localhost/" + ("x" * 49),
139 "[::1/QWNVAevHNSHyMk1qarlZAQOB5swl3Ptu1yGCMSZrzKWpBUMv",
141 _, err = bar.cmd("join", url, code=1)
142 check.is_in("Invalid invitation URL", err)
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)
153 timeout_err = "Timed out waiting for the server"
154 conn_err = "Could not connect to inviter"
155 server_err = "Please try again"
157 bad_url = f"127.0.0.1:{foo.port}/jkhjAi0LGVP0o6TN7aa_7xjqM9qTb_DUxBpk6UuLEF4ubDLX"
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)
164 url, _ = foo.cmd("invite", "bar")
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)
171 log.info("remove invitation directory")
172 shutil.rmtree(foo.sub("invitations"))
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)
179 foo.add_script(Script.TINC_DOWN)
181 foo[Script.TINC_DOWN].wait()
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)
188 with Test("run invite success tests") as context:
189 test_invite(context.node(init=True))
191 with Test("run invite error tests") as context:
192 test_invite_errors(context.node(init=True))
194 with Test("run join tests") as context:
195 test_join_errors(context.node(init=True))
197 with Test("broken invitation") as context:
198 test_broken_invite(context)