diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_admin_challenge_bounds.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_admin_challenge_bounds.py | 45 |
1 files changed, 13 insertions, 32 deletions
diff --git a/packages/meshbay-node/tests/test_admin_challenge_bounds.py b/packages/meshbay-node/tests/test_admin_challenge_bounds.py index fd3b2f1..9dcb750 100644 --- a/packages/meshbay-node/tests/test_admin_challenge_bounds.py +++ b/packages/meshbay-node/tests/test_admin_challenge_bounds.py @@ -3,8 +3,9 @@ What a connection may leave waiting for a signature (docs/MESHBAY_DESIGN.md §13 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. +Measured before the bound: 200 `root_add` (an op since removed) of 1 MiB each +from a plain member held 200 pending operations and ~400 MiB for the life of +the connection. """ import struct @@ -48,23 +49,24 @@ def _member_session(): return s -def _root_add(s, path: str) -> dict: - s._dispatch_message({"type": "root_add", "group_id": GROUP, "path": path}) +def _ask(s, path: str) -> dict: + """A signed op whose subject is whatever the caller sends.""" + s._dispatch_message({"type": "chat_directory", "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 _ask(s, f"/srv/{i}")["type"] == "admin_challenge" + refused = _ask(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)) + refused = _ask(s, "x" * (MAX_ADMIN_OP_BYTES + 1)) assert refused["type"] == "error" and refused["code"] == "too_large" assert s._admin_ops == {} @@ -72,21 +74,21 @@ def test_an_oversized_request_is_not_kept(): 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}") + _ask(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 _ask(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}") + _ask(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 _ask(s, "/srv/next")["type"] == "admin_challenge" assert all(time.time() - p["ts"] < 5 for p in s._admin_ops.values()) @@ -95,27 +97,6 @@ def test_answering_a_challenge_frees_its_place(): # 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() |