diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_member_upload_policy.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_member_upload_policy.py | 176 |
1 files changed, 0 insertions, 176 deletions
diff --git a/packages/meshbay-node/tests/test_member_upload_policy.py b/packages/meshbay-node/tests/test_member_upload_policy.py deleted file mode 100644 index b1dc0cb..0000000 --- a/packages/meshbay-node/tests/test_member_upload_policy.py +++ /dev/null @@ -1,176 +0,0 @@ -""" -The operator can close uploading to everyone but themselves. - -The point of these tests is the difference between a hidden button and a closed -door. The interface stops offering the control, which is a courtesy to the -people who are not trying; **the node refuses the upload**, which is the part -that holds against someone who is. A member who kept an old tab open, or who -speaks MNP directly, gets the same answer as everyone else. - -Two further things are worth holding: - -* the setting is changed by a **signed** operator instruction. A node that took - it from an unsigned message would let any member turn it back on, and the - control would be a suggestion; -* it is stored on the **node**, not the hub. A hub that could decide who may - write to the operator's disk is a hub with authority over the node, which is - the thing this whole design is arranged to avoid. -""" - -import base64 -from pathlib import Path - -import pytest -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey - -from meshbay_common.adminop import OP_MEMBER_UPLOAD -from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.roster import Roster -from meshbay_node.transport.webrtc_server import WebRTCPeerSession - -from conftest import one_root - -pytestmark = pytest.mark.asyncio - - -def _session(tmp_path: Path, user_id: str, *, member_upload: bool, - operator: str | None = None) -> WebRTCPeerSession: - shared_root = tmp_path / "shared" - shared_root.mkdir(exist_ok=True) - index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) - ctx = { - "roots": one_root(shared_root), - "index": index, - "sk_node": index.sk_node, - "member_upload": member_upload, - "node_user_id": operator, - } - session = WebRTCPeerSession.__new__(WebRTCPeerSession) - session._ctx = ctx - session._group_id = None - session._user_id = user_id - session._pk_user = "" - session._uploads = {} - session.sent = [] - session._send = session.sent.append - session._audit = lambda *a, **k: None - return session - - -def _upload(session, filename="clip.mp4", body=b"bytes"): - session._do_file_upload({ - "filename": filename, "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(body).decode(), - }) - - -def _uploads_dir(session) -> Path: - return session._ctx["roots"].upload_root.path / "uploads" - - -# ── The door, not the button ──────────────────────────────────────────────── - -async def test_a_member_cannot_upload_when_it_is_turned_off(tmp_path): - session = _session(tmp_path, "member-1", member_upload=False, - operator="the-operator") - _upload(session) - - assert not (_uploads_dir(session) / "clip.mp4").exists(), ( - "the file was written even though uploading is off — the setting is " - "decorative and the hidden button was the whole control") - refusal = [m for m in session.sent if m.get("type") == "error"] - assert refusal and refusal[0].get("code") == "member_upload_off" - - -async def test_the_operator_can_still_upload(tmp_path): - """Otherwise turning it off locks the operator out of their own node, and - the only way back is a config file and a restart.""" - session = _session(tmp_path, "the-operator", member_upload=False, - operator="the-operator") - _upload(session) - - assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes" - - -async def test_members_upload_normally_when_it_is_on(tmp_path): - session = _session(tmp_path, "member-1", member_upload=True, - operator="the-operator") - _upload(session) - - assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes" - - -async def test_a_node_that_never_heard_of_the_setting_still_accepts_uploads(tmp_path): - """An existing node's context has no such key. The absence must read as - "allowed", or upgrading the node silently closes every group.""" - session = _session(tmp_path, "member-1", member_upload=True, - operator="the-operator") - del session._ctx["member_upload"] - _upload(session) - - assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes" - - -# ── Who may change it ─────────────────────────────────────────────────────── - -async def test_changing_it_needs_a_signature(tmp_path): - """ - The request only ever produces a challenge. Nothing is applied until a - signature over the transcript verifies — the same path as removing a member. - """ - session = _session(tmp_path, "member-1", member_upload=True, - operator="the-operator") - session._has_admin_authority = lambda: True - issued = [] - session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) - - session._do_member_upload({"allowed": False}) - - assert issued == [(OP_MEMBER_UPLOAD, "off")] - assert session._ctx["member_upload"] is True, "applied before it was signed" - - -async def test_the_subject_names_the_outcome_not_the_operation(tmp_path): - """The operator is shown the subject before signing. "member_upload" tells - them nothing; "off" tells them what they are about to do.""" - session = _session(tmp_path, "op", member_upload=False, operator="op") - session._has_admin_authority = lambda: True - issued = [] - session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) - - session._do_member_upload({"allowed": True}) - - assert issued == [(OP_MEMBER_UPLOAD, "on")] - - -async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): - session = _session(tmp_path, "member-1", member_upload=True, - operator="the-operator") - session._has_admin_authority = lambda: False - - session._do_member_upload({"allowed": False}) - - assert [m for m in session.sent if m.get("type") == "error"] - - -# ── Where it is stored ────────────────────────────────────────────────────── - -async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): - roster = Roster(db_path=tmp_path / "roster.db") - await roster.open() - try: - assert await roster.member_upload_allowed("g1") is True, ( - "absent must mean allowed, or an upgrade closes every group") - await roster.set_member_upload("g1", False, set_by="op") - assert await roster.member_upload_allowed("g1") is False - finally: - await roster.close() - - reopened = Roster(db_path=tmp_path / "roster.db") - await reopened.open() - try: - assert await reopened.member_upload_allowed("g1") is False - assert await reopened.member_upload_allowed("g2") is True, ( - "one group's setting must not answer for another") - finally: - await reopened.close() |