diff options
Diffstat (limited to 'packages/meshbay-cli/tests/fake_node.py')
| -rw-r--r-- | packages/meshbay-cli/tests/fake_node.py | 29 |
1 files changed, 28 insertions, 1 deletions
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index f9c552c..552bd39 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -23,7 +23,7 @@ from fake_hub import USER_ID from meshbay_cli.client.keys import Identity from meshbay_cli.client.session import GroupSession, NodeRefusal from meshbay_common import chatbox -from meshbay_common.adminop import OP_FILE_DELETE, admin_transcript +from meshbay_common.adminop import OP_DIR_DELETE, OP_FILE_DELETE, admin_transcript from meshbay_common.device import ( device_add_transcript, device_code_hash, @@ -322,11 +322,38 @@ class FakeChannel: "subject": msg["file_id"], "nonce": base64.b64encode(nonce).decode(), "ts": 1_700_000_000, "node_pk": NODE_PK, "group_id": GROUP_ID} + def _dir_delete(self, msg): + """Empty folders only, the operator's only, never a root — the node's rule.""" + n, path = self.node, msg["dir"].strip("/") + if path not in n.dirs or "/" not in path: + return {"type": "error", "detail": "Invalid directory"} + if any(d.startswith(path + "/") for d in n.dirs) or \ + any(e["path"] == path or e["path"].startswith(path + "/") + for e in n.files.values()): + return {"type": "error", "detail": "Directory is not empty"} + if n.role != "operator": + return {"type": "error", "detail": "No authorized key for deletion"} + op_id, nonce = secrets.token_hex(8), secrets.token_bytes(32) + n.admin_ops[op_id] = {"op": OP_DIR_DELETE, "subject": path, "nonce": nonce, + "ts": 1_700_000_000} + return {"type": "admin_challenge", "op_id": op_id, "op": OP_DIR_DELETE, + "subject": path, "nonce": base64.b64encode(nonce).decode(), + "ts": 1_700_000_000, "node_pk": NODE_PK, "group_id": GROUP_ID} + def _admin_response(self, msg): n = self.node op = n.admin_ops.pop(msg["op_id"], None) if op is None: return {"type": "error", "detail": "No pending admin operation"} + if op.get("op") == OP_DIR_DELETE: + transcript = admin_transcript(op=OP_DIR_DELETE, node_pk_b64=NODE_PK, + group_id=GROUP_ID, subject=op["subject"], + nonce=op["nonce"], ts=op["ts"]) + if not (_signed_by_a_device(n, transcript, msg["signature"]) + and n.role == "operator"): + return {"type": "error", "detail": "Signature verification failed"} + n.dirs.discard(op["subject"]) + return {"type": "dir_delete_ack", "dir": op["subject"]} entry = n.files.get(op["subject"]) transcript = admin_transcript(op=OP_FILE_DELETE, node_pk_b64=NODE_PK, group_id=GROUP_ID, subject=op["subject"], |