1 files changed, 109 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py
index a5a48e4..88e426d 100644
--- a/packages/meshbay-node/tests/test_roster_pairing.py
+++ b/packages/meshbay-node/tests/test_roster_pairing.py
@@ -805,3 +805,112 @@ async def test_each_group_gets_its_own_key(tmp_path, roster):
     assert await roster.is_authorized("group-a", "member") is True
     assert await roster.is_authorized("group-b", "member") is False, (
         "membership of one group must not admit anyone to another")
+
+
+# ── Removing a directory ─────────────────────────────────────────────────────
+
+async def _dir_session(tmp_path, roster):
+    """A session with a shared root and an operator paired, ready for admin ops."""
+    session = _session(tmp_path, roster, group_id="g1", gek=generate_gek())
+    session._admin_ops = {}
+    session._ctx["has_admin_authority"] = True
+    return session
+
+
+async def test_a_directory_with_anything_in_it_is_refused(tmp_path, roster):
+    session = await _dir_session(tmp_path, roster)
+    full = tmp_path / "shared" / "full"
+    full.mkdir()
+    (full / "keep.txt").write_text("still here")
+
+    session._do_dir_delete({"dir": "full"})
+
+    assert _last(session).get("detail") == "Directory is not empty"
+    assert full.exists() and (full / "keep.txt").exists()
+
+
+async def test_no_challenge_is_issued_without_an_operator(tmp_path, roster):
+    """Fails closed, and says so, rather than asking for a signature nobody can give."""
+    session = await _dir_session(tmp_path, roster)
+    session._ctx["has_admin_authority"] = False
+    (tmp_path / "shared" / "empty").mkdir()
+
+    session._do_dir_delete({"dir": "empty"})
+
+    assert _last(session).get("detail") == "No authorized key for deletion"
+    assert (tmp_path / "shared" / "empty").exists()
+
+
+async def test_the_shared_root_itself_is_not_a_target(tmp_path, roster):
+    session = await _dir_session(tmp_path, roster)
+    for attempt in ("", ".", "/", "../shared"):
+        session._do_dir_delete({"dir": attempt})
+        assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
+    assert (tmp_path / "shared").is_dir()
+
+
+async def test_escaping_the_shared_root_is_refused(tmp_path, roster):
+    session = await _dir_session(tmp_path, roster)
+    outside = tmp_path / "outside"
+    outside.mkdir()
+
+    for attempt in ("../outside", "../../outside", "sub/../../outside"):
+        session._do_dir_delete({"dir": attempt})
+        assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
+    assert outside.is_dir(), "a path leaving the shared root removed a directory"
+
+
+async def test_an_empty_directory_needs_a_signature_and_then_goes(tmp_path, roster):
+    """The whole round trip: challenge, operator signature, removal."""
+    from meshbay_common.adminop import admin_transcript
+
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
+    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
+
+    session = await _dir_session(tmp_path, roster)
+    (tmp_path / "shared" / "gone").mkdir()
+
+    session._do_dir_delete({"dir": "gone"})
+    challenge = _last(session)
+    assert challenge["type"] == "admin_challenge"
+    assert challenge["op"] == "dir_delete"
+    assert challenge["subject"] == "gone"
+
+    transcript = admin_transcript(
+        op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1",
+        subject="gone", nonce=base64.b64decode(challenge["nonce"]),
+        ts=challenge["ts"])
+    await session._admin_exec_dir_delete(
+        session._admin_ops.pop(challenge["op_id"]) if session._admin_ops
+        else {"op": "dir_delete", "subject": "gone"},
+        transcript, sk_ed.sign(transcript))
+
+    assert _last(session)["type"] == "dir_delete_ack"
+    assert not (tmp_path / "shared" / "gone").exists()
+
+
+async def test_someone_elses_signature_does_not_remove_it(tmp_path, roster):
+    from meshbay_common.adminop import admin_transcript
+
+    sk_op, pk_op, pk_x = _keypair()
+    await roster.pin_identity("grenet", "grenet", pk_op, pk_x, "code")
+    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
+
+    sk_member, pk_member, pk_x_m = _keypair()
+    await roster.pin_identity("mallory", "mallory", pk_member, pk_x_m, "code")
+    await roster.set_member("g1", "mallory", ROLE_MEMBER, "active", "grenet")
+
+    session = await _dir_session(tmp_path, roster)
+    (tmp_path / "shared" / "theirs").mkdir()
+
+    transcript = admin_transcript(
+        op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1",
+        subject="theirs", nonce=b"\x22" * 32, ts=int(time.time()))
+    await session._admin_exec_dir_delete(
+        {"op": "dir_delete", "subject": "theirs"},
+        transcript, sk_member.sign(transcript))
+
+    assert _last(session).get("detail") == "Signature verification failed"
+    assert (tmp_path / "shared" / "theirs").is_dir(), (
+        "a member's signature removed a directory — only the operator may")