1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
|
"""
The browser's half of an invitation link (docs/MESHBAY_DESIGN.md §3.4).
Three properties, each run against the shipped code rather than restated:
- **one shape.** The hub writes a link when it mails one (`invite_url`), the
page writes one when it shows one (`buildInviteLink`), and the page reads both
(`parseInvite`). A disagreement is a link that opens on nothing.
- **the code leaves the address at once, and the tab keeps it.** Run in node
against a stand-in `window`: `captureFromLocation` rewrites the address and
stores what it read, and a malformed link is cleaned out without being kept.
- **the code goes to the node the link names, and to no other.** The transport's
`_linkJoinRefusal` is what stops it; this runs it.
The rest are read from the source, which is the evidence there is for them: the
hub is never handed the code except when the inviter ticked the mail box, the
capture is the first thing `app.js` loads, and signing out forgets the
invitation.
"""
import base64
import json
import re
import shutil
import subprocess
from pathlib import Path
import pytest
from meshbay_hub import mail as mail_mod
from meshbay_hub.api import invite_links
from spa_source import transport_source
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
LINK_JS = STATIC / "invite-link.js"
pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")
GROUP = "0f8fad5b-d9cb-469f-a165-70867728950e"
TICKET = "AbCdEfGhIjKlMnOpQr-_12"
NODE_PK_STD = base64.b64encode(bytes(range(32))).decode() # has '+', '/', '='
CODE = "K7P2-9WQX"
def _module_body() -> str:
"""invite-link.js with its import, its exports and its load-time capture
removed — the functions as shipped, runnable against a stand-in window."""
src = LINK_JS.read_text(encoding="utf-8")
src = re.sub(r"^import .*?;\n", "", src, flags=re.M)
src = src.replace("export function", "function")
tail = "\ncaptureFromLocation();\nwindow.addEventListener('hashchange', captureFromLocation);\n"
assert src.endswith(tail), "invite-link.js no longer ends with its load-time capture"
return src[: -len(tail)]
def _run(tmp_path, script: str):
harness = tmp_path / "h.js"
harness.write_text(script)
out = subprocess.run(["node", str(harness)], capture_output=True, text=True, timeout=60)
assert out.returncode == 0, out.stderr
return json.loads(out.stdout)
_WINDOW = r"""
const store = new Map();
globalThis.sessionStorage = {
getItem: (k) => (store.has(k) ? store.get(k) : null),
setItem: (k, v) => store.set(k, String(v)),
removeItem: (k) => store.delete(k),
};
const replaced = [];
globalThis.window = {
location: { hash: '', pathname: '/', search: '' },
history: { replaceState: (_s, _t, url) => replaced.push(url) },
addEventListener() {},
};
const platform = { hubOrigin: () => 'https://hub.example' };
"""
def test_one_shape_between_the_hub_and_the_page(tmp_path, monkeypatch):
monkeypatch.setattr(mail_mod, "_hub_url", "https://hub.example")
n = NODE_PK_STD.replace("+", "-").replace("/", "_").rstrip("=")
from_hub = invite_links.invite_url(GROUP, TICKET, n, CODE)
got = _run(tmp_path, _WINDOW + _module_body() + f"""
const fields = {{ g: '{GROUP}', t: '{TICKET}',
n: nodePkForLink('{NODE_PK_STD}'), c: '{CODE}' }};
process.stdout.write(JSON.stringify({{
parsed: parseInvite({json.dumps(from_hub)}),
built: buildInviteLink('https://hub.example', fields),
back: nodePkFromLink(fields.n),
lower: parseInvite({json.dumps(from_hub.replace(CODE, CODE.lower()))}),
}}));
""")
assert got["parsed"] == {"g": GROUP, "t": TICKET, "n": n, "c": CODE}
assert got["built"] == from_hub
assert got["back"] == NODE_PK_STD, "the key the transport compares must come back exact"
assert got["lower"]["c"] == CODE
def test_the_cli_writes_the_same_link(monkeypatch):
"""`meshbay-node member invite --link` builds its link on the node, from the
hub address it was configured with; it must be the hub's own shape."""
from meshbay_node.ops import _invite_url
monkeypatch.setattr(mail_mod, "_hub_url", "https://hub.example")
n = NODE_PK_STD.replace("+", "-").replace("/", "_").rstrip("=")
assert (_invite_url("https://hub.example/", GROUP, TICKET, NODE_PK_STD, CODE)
== invite_links.invite_url(GROUP, TICKET, n, CODE))
@pytest.mark.parametrize("tamper", [
lambda u: u.replace("v=1", "v=2"),
lambda u: u.replace(CODE, "K7P2-9WQ"),
lambda u: u.replace(CODE, "K7P2-9WQX<script>"),
lambda u: u.replace(TICKET, TICKET + "x"),
lambda u: u.replace(GROUP, "../../admin"),
lambda u: u.replace("&n=", "&m="),
])
def test_anything_but_that_shape_is_not_an_invitation(tmp_path, tamper):
good = f"https://hub.example/#/invite?v=1&g={GROUP}&t={TICKET}&n={'A' * 43}&c={CODE}"
got = _run(tmp_path, _WINDOW + _module_body()
+ f"process.stdout.write(JSON.stringify(parseInvite({json.dumps(tamper(good))})));")
assert got is None
def test_the_code_leaves_the_address_and_stays_in_the_tab(tmp_path):
good = f"#/invite?v=1&g={GROUP}&t={TICKET}&n={'A' * 43}&c={CODE}"
got = _run(tmp_path, _WINDOW + _module_body() + f"""
window.location.hash = {json.dumps(good)};
const first = captureFromLocation();
const kept = loadPending();
window.location.hash = '#/invite?v=1&g=nope';
captureFromLocation();
const afterBad = loadPending();
clearPending();
process.stdout.write(JSON.stringify({{
first: Boolean(first), replaced, kept, afterBad, cleared: loadPending(),
}}));
""")
assert got["first"] is True
assert got["replaced"] == ["/#/invite", "/#/invite"], (
"both the good link and the malformed one must be taken out of the address")
assert got["kept"]["c"] == CODE and got["kept"]["g"] == GROUP
assert got["afterBad"]["t"] == TICKET, "a malformed link must not replace a good one"
assert got["cleared"] is None
def test_a_link_code_goes_to_the_node_the_link_names_and_no_other(tmp_path):
src = transport_source()
fn = re.search(r"^function _linkJoinRefusal\(.*?^\}", src, re.M | re.S)
assert fn, "transport.js no longer has _linkJoinRefusal"
got = _run(tmp_path, fn.group(0) + """
const r = (...a) => { const e = _linkJoinRefusal(...a); return e ? e.reason : null; };
process.stdout.write(JSON.stringify([
r('KEY', 'K7P2-9WQX', 'KEY', true),
r('KEY', 'K7P2-9WQX', 'OTHER', true),
r('KEY', 'K7P2-9WQX', 'KEY', false),
r(undefined, 'K7P2-9WQX', 'OTHER', false),
r('KEY', null, 'OTHER', false),
]));
""")
assert got == [None, "link_other_node", "link_node_unproved", None, None]
# ── Read from the source ─────────────────────────────────────────────────────
def _code(name: str) -> str:
"""The file without its comments — prose about the code is not the code."""
src = (STATIC / name).read_text(encoding="utf-8")
src = re.sub(r"/\*.*?\*/", "", src, flags=re.S)
return "\n".join(line for line in src.splitlines()
if not line.strip().startswith("//"))
def test_the_capture_is_the_first_thing_the_app_loads():
imports = re.findall(r"^import .*? from '([^']+)';", _code("app.js"), re.M | re.S)
assert imports and imports[0] == "./invite-link.js"
def test_the_invitation_page_never_sends_the_code_to_the_hub():
page = _code("invite-page.js")
assert "inv.t" in page, "the check below is looking at the wrong names"
assert not re.search(r"\binv\.c\b|\binv\[.c.\]", page), (
"invite-page.js reads the code; only the ticket is its to send")
def test_the_members_tab_sends_the_code_only_for_the_mail():
settings = _code("group-settings.js")
sends = [m.start() for m in re.finditer(r"code: node\.code", settings)]
assert len(sends) == 1
before = settings[settings.rfind("\n", 0, sends[0] - 200):sends[0]]
assert "inviteByEmail ?" in before, "the code reaches the hub only when the box asks"
def test_signing_out_forgets_the_invitation():
app = _code("app.js")
logout = app[app.index("logout: () => {"):]
logout = logout[:logout.index("},")]
assert "clearPending()" in logout
def test_the_group_page_moves_on_from_another_host():
page = _code("group-page.js")
loop = page[page.index("for (const n of nodesData.nodes)"):]
loop = loop[:loop.index("if (!transport)")]
assert "link_other_node" in loop
|