1 files changed, 135 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_admin_challenge_bounds.py b/packages/meshbay-node/tests/test_admin_challenge_bounds.py
new file mode 100644
index 0000000..fd3b2f1
--- /dev/null
+++ b/packages/meshbay-node/tests/test_admin_challenge_bounds.py
@@ -0,0 +1,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