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
|
"""Sign-up and passphrase reset are gated by the hub's own proof of work.
The gate replaced reCAPTCHA, which sent every visitor's address and browser
details to Google and needed a consent the sign-up form could not ask for
freely. What it must still be: unconditional when enabled — the old gate was
once skipped for any request carrying an `auth_key`, which every real client
sends — and an answer the hub issued, in date, and used once.
The last test runs the page's own solver (`static/captcha.js`) against a
challenge from `captcha.issue`, so the two halves cannot drift apart unseen.
"""
import hashlib
import json
import shutil
import subprocess
import time
from pathlib import Path
import pytest
from meshbay_hub import captcha
from meshbay_hub.config import load_config
STATIC = Path(captcha.__file__).parent / "static"
@pytest.fixture
def captcha_on(client, monkeypatch):
"""The check on, with little enough work that the tests solve it at once."""
from meshbay_hub.api.users import _cfg
monkeypatch.setattr(_cfg.captcha, "enabled", True)
monkeypatch.setattr(_cfg.captcha, "max_number", 2000)
monkeypatch.setattr(captcha, "_spent", {})
def _solve(ch: dict) -> str:
for n in range(ch["max_number"]):
if hashlib.sha256(f"{ch['prefix']}{n}".encode()).hexdigest() == ch["challenge"]:
return f"{ch['prefix']}{n}.{ch['signature']}"
raise AssertionError("no number matches the challenge")
async def _token(client) -> str:
r = await client.get("/v1/users/captcha")
assert r.status_code == 200, r.text
return _solve(r.json())
def _body(**over):
b = {"username": "newbie_test", "email": "newbie@t.com", "auth_key": "a" * 44}
b.update(over)
return b
@pytest.mark.asyncio
async def test_missing_answer_rejected_even_with_auth_key(client, captcha_on):
r = await client.post("/v1/users/register", json=_body())
assert r.status_code == 400
assert r.json()["detail"] == "captcha_required"
@pytest.mark.asyncio
async def test_solved_challenge_accepted(client, captcha_on):
r = await client.post("/v1/users/register",
json=_body(captcha_token=await _token(client)))
assert r.status_code == 201, r.text
@pytest.mark.asyncio
async def test_an_answer_is_good_for_one_form(client, captcha_on):
token = await _token(client)
r = await client.post("/v1/users/register", json=_body(captcha_token=token))
assert r.status_code == 201, r.text
r = await client.post("/v1/users/register", json=_body(
username="another_test", email="another@t.com", captcha_token=token))
assert r.status_code == 400
assert r.json()["detail"] == "captcha_failed"
@pytest.mark.asyncio
async def test_a_wrong_number_is_refused(client, captcha_on):
nonce, expires, number, sig = (await _token(client)).split(".")
forged = f"{nonce}.{expires}.{int(number) + 1}.{sig}"
r = await client.post("/v1/users/register", json=_body(captcha_token=forged))
assert r.json()["detail"] == "captcha_failed"
@pytest.mark.asyncio
async def test_a_challenge_the_hub_did_not_sign_is_refused(client, captcha_on):
"""Somebody minting their own easy challenge has no key to sign it with."""
forged = f"{'00' * 12}.{int(time.time()) + 60}.0.{'0' * 64}"
r = await client.post("/v1/users/register", json=_body(captcha_token=forged))
assert r.json()["detail"] == "captcha_failed"
@pytest.mark.asyncio
async def test_an_expired_answer_is_refused(client, captcha_on, monkeypatch):
token = await _token(client)
later = time.time() + captcha.CHALLENGE_TTL + 1
monkeypatch.setattr(captcha.time, "time", lambda: later)
r = await client.post("/v1/users/register", json=_body(captcha_token=token))
assert r.json()["detail"] == "captcha_failed"
@pytest.mark.asyncio
async def test_password_reset_request_is_gated_too(client, captcha_on):
body = {"username": "nobody_test", "email": "nobody@t.com"}
r = await client.post("/v1/users/password/reset-request", json=body)
assert r.json()["detail"] == "captcha_required"
r = await client.post("/v1/users/password/reset-request",
json={**body, "captcha_token": await _token(client)})
assert r.status_code == 200, r.text
@pytest.mark.asyncio
async def test_turned_off_it_issues_nothing_and_checks_nothing(client):
# The test configuration has it off (conftest.py).
r = await client.get("/v1/users/captcha")
assert r.status_code == 404
assert r.json()["detail"] == "captcha_disabled"
r = await client.post("/v1/users/register", json=_body())
assert r.status_code == 201
def test_on_unless_the_configuration_says_otherwise(tmp_path):
cfg_file = tmp_path / "hub.toml"
cfg_file.write_text('[server]\nport = 8000\n', encoding="utf-8")
assert load_config(cfg_file).captcha.enabled
cfg_file.write_text('[captcha]\nenabled = false\n', encoding="utf-8")
assert not load_config(cfg_file).captcha.enabled
def test_old_recaptcha_keys_are_ignored_with_a_warning(tmp_path, monkeypatch):
# Caught at the logger rather than through caplog: other tests in the suite
# reconfigure logging, and caplog then sees nothing.
from meshbay_hub import config
warned = []
monkeypatch.setattr(config.log, "warning", lambda msg, *a: warned.append(msg))
cfg_file = tmp_path / "hub.toml"
cfg_file.write_text('[captcha]\nsite_key = "6Lsite"\nsecret_key = "6Lsecret"\n',
encoding="utf-8")
cfg = load_config(cfg_file)
assert cfg.captcha.enabled
assert any("no longer read" in m for m in warned)
@pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")
@pytest.mark.asyncio
async def test_the_page_solver_answers_the_hub(client, captcha_on, tmp_path):
ch = (await client.get("/v1/users/captcha")).json()
(tmp_path / "package.json").write_text('{"type":"module"}', encoding="utf-8")
(tmp_path / "captcha.js").write_text(
(STATIC / "captcha.js").read_text(encoding="utf-8"), encoding="utf-8")
(tmp_path / "case.js").write_text(
"import { solve } from './captcha.js';\n"
f"console.log(JSON.stringify(await solve({json.dumps(ch)})));\n", encoding="utf-8")
proc = subprocess.run(["node", "case.js"], cwd=tmp_path,
capture_output=True, text=True, encoding="utf-8")
assert proc.returncode == 0, proc.stderr
token = json.loads(proc.stdout)
assert token == _solve(ch)
r = await client.post("/v1/users/register", json=_body(captcha_token=token))
assert r.status_code == 201, r.text
|