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
|
"""
What a connection may leave waiting for a signature (docs/MESHBAY_DESIGN.md §13.5b).
Anyone authenticated can ask for an admin challenge — the signature is checked
later — so a member who never answers must not make the node keep every request.
Measured before the bound: 200 `root_add` of 1 MiB each from a plain member held
200 pending operations and ~400 MiB for the life of the connection.
"""
import struct
import time
import msgpack
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_node.transport.webrtc.admin import MAX_ADMIN_OP_BYTES, MAX_PENDING_ADMIN_OPS
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
GROUP = "g" * 32
class _Channel:
readyState = "open"
def __init__(self):
self.sent = []
def send(self, data: bytes) -> None:
(n,) = struct.unpack(">I", data[:4])
self.sent.append(msgpack.unpackb(data[4:4 + n], raw=False))
class _PC:
connectionState = "connected"
iceConnectionState = "connected"
remoteDescription = None
localDescription = None
sctp = None
def _member_session():
"""An authenticated member — not the operator — on a node that has one."""
ctx = {"sk_node": Ed25519PrivateKey.from_private_bytes(b"\x01" * 32),
"groups": {GROUP: {}}, "has_admin_authority": True}
s = WebRTCPeerSession(_PC(), ctx, peer_id="peer")
s._channel = _Channel()
s._audit = lambda *a, **k: None
s._user_id, s._group_id = "member-1", GROUP
return s
def _root_add(s, path: str) -> dict:
s._dispatch_message({"type": "root_add", "group_id": GROUP, "path": path})
return s._channel.sent[-1]
def test_a_member_cannot_pile_up_challenges():
s = _member_session()
for i in range(MAX_PENDING_ADMIN_OPS):
assert _root_add(s, f"/srv/{i}")["type"] == "admin_challenge"
refused = _root_add(s, "/srv/one-too-many")
assert refused["type"] == "error" and refused["code"] == "too_many_pending"
assert len(s._admin_ops) == MAX_PENDING_ADMIN_OPS
def test_an_oversized_request_is_not_kept():
s = _member_session()
refused = _root_add(s, "x" * (MAX_ADMIN_OP_BYTES + 1))
assert refused["type"] == "error" and refused["code"] == "too_large"
assert s._admin_ops == {}
def test_an_expired_challenge_frees_its_place():
s = _member_session()
for i in range(MAX_PENDING_ADMIN_OPS):
_root_add(s, f"/srv/{i}")
for pending in s._admin_ops.values():
pending["ts"] -= 10_000
assert _root_add(s, "/srv/after-expiry")["type"] == "admin_challenge"
assert len(s._admin_ops) == 1
def test_answering_a_challenge_frees_its_place():
s = _member_session()
for i in range(MAX_PENDING_ADMIN_OPS):
_root_add(s, f"/srv/{i}")
op_id = next(iter(s._admin_ops))
s._dispatch_message({"type": "admin_response", "op_id": op_id, "signature": "!!"})
assert len(s._admin_ops) == MAX_PENDING_ADMIN_OPS - 1
assert _root_add(s, "/srv/next")["type"] == "admin_challenge"
assert all(time.time() - p["ts"] < 5 for p in s._admin_ops.values())
# ── What a challenge covers (docs/MESHBAY_DESIGN.md §5.4) ────────────────────
#
# The signature covers the subject and nothing else of a request, so every value
# the executor acts on has to be in it.
def test_root_add_signs_whether_members_may_write():
from meshbay_common.adminop import root_add_subject
s = _member_session()
s._dispatch_message({"type": "root_add", "group_id": GROUP, "path": "/srv/drop",
"name": "Drop", "writable": True, "removable": False})
challenge = s._channel.sent[-1]
assert challenge["subject"] == root_add_subject("/srv/drop", "Drop", "generic",
True, False)
assert challenge["subject"] != root_add_subject("/srv/drop", "Drop", "generic",
False, False)
def test_group_attach_signs_the_directory_it_exposes():
from meshbay_common.adminop import group_attach_subject
s = _member_session()
s._dispatch_message({"type": "group_attach", "name": "photos",
"shared_dir": "/home/me/Photos"})
assert s._channel.sent[-1]["subject"] == group_attach_subject(
"photos", "/home/me/Photos", True)
def test_invite_create_signs_the_name_it_records():
from meshbay_common.adminop import invite_create_subject
s = _member_session()
s._ctx["roster"] = object() # only its presence is checked before the challenge
s._dispatch_message({"type": "invite_create", "group_id": GROUP,
"user_id": "u-1", "username": "alice"})
assert s._channel.sent[-1]["subject"] == invite_create_subject("u-1", "alice")
def test_tmdb_config_signs_the_token_without_writing_it():
from meshbay_common.adminop import tmdb_config_subject
s = _member_session()
s._dispatch_message({"type": "tmdb_config", "token": "secret-token",
"language": "fr-FR"})
subject = s._channel.sent[-1]["subject"]
assert subject == tmdb_config_subject("secret-token", "fr-FR")
assert "secret-token" not in subject
|