diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/quic_server.py | 40 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 301 |
2 files changed, 270 insertions, 71 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 73e668c..ce6fe17 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -35,6 +35,7 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived, StreamReset from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION +from meshbay_node.roots import RootSet, entry_abs_path from meshbay_common.handshake import ( NONCE_LEN, ROLE_CLIENT, @@ -98,6 +99,37 @@ class Denylist: log.info("Denied jti: %s", jti[:8]) self._save() + def entries(self) -> dict[str, list[str]]: + """What is currently refused, for the operator to inspect (14.10).""" + return { + "users": sorted(self.user_ids), + "groups": sorted(self.group_ids), + "jtis": sorted(self.jtis), + } + + def clear(self, subject: str = "") -> int: + """ + Drop everything, or one identifier. Returns how many entries went. + + Not silent by design: clearing re-admits whoever it was keeping out, and + the count is what tells the operator whether they undid one revocation + or all of them. + """ + before = len(self.user_ids) + len(self.group_ids) + len(self.jtis) + if subject: + self.user_ids.discard(subject) + self.group_ids.discard(subject) + self.jtis.discard(subject) + else: + self.user_ids.clear() + self.group_ids.clear() + self.jtis.clear() + after = len(self.user_ids) + len(self.group_ids) + len(self.jtis) + removed = before - after + if removed: + self._save() + return removed + def _load(self) -> None: if not self._path or not self._path.exists(): return @@ -352,7 +384,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return @@ -379,7 +411,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return @@ -506,7 +538,7 @@ class QuicChunkServer: sk_node: Ed25519PrivateKey, hub_pk_pem: bytes, gek: bytes, - shared_root: Path, + roots: RootSet, index: GroupIndex, host: str = "::", # listen IPv4 + IPv6 (dual-stack Linux) port: int = 19000, @@ -519,7 +551,7 @@ class QuicChunkServer: "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, - "shared_root": shared_root, + "roots": roots, "index": index, } if groups: diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 6b87e31..4ec841f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -60,6 +60,8 @@ from meshbay_common.adminop import ( OP_FILE_DELETE, OP_INVITE_CREATE, OP_MEMBER_REVOKE, + OP_GEK_ROTATE, + OP_MEMBER_UNPIN, admin_transcript, ) from meshbay_common.crypto import pk_to_b64, wrap_gek_aes @@ -72,6 +74,8 @@ from meshbay_common.join import ( from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex +from meshbay_node import ops +from meshbay_node.roots import RootSet, entry_abs_path from meshbay_node.roster import DEFAULT_INVITE_TTL log = logging.getLogger(__name__) @@ -156,37 +160,47 @@ def _free_name(directory: Path, filename: str) -> str: raise FileExistsError(filename) -def safe_subdir(shared_root: Path, rel: str) -> Path | None: +def safe_subdir(roots: RootSet, rel: str) -> Path | None: """ - Resolve a client-supplied directory under the shared root, or refuse. + Resolve a client-supplied directory inside one of the group's roots, or refuse. - Uploads land where the member is looking now rather than in a per-user - quarantine, so the path arrives from the wire and every part of it has to be - checked: each segment against the same allowlist as filenames, and the - resolved result against the root. `..`, absolute paths, symlinks pointing - out, and anything with a separator in a segment are all refused here rather - than in the caller, so there is one place to get it right. + The path arrives from the wire, so every part is checked: the first segment + must name a root that is readable right now, each later segment against the + same allowlist as filenames, and the resolved result against that root's + directory. `..`, absolute paths, symlinks pointing out, and anything with a + separator in a segment are all refused here rather than in the caller, so + there is one place to get it right. + + The virtual root itself — `""` — is deliberately **not** resolvable. It is + not a directory on anyone's disk: a file cannot be written there and a + directory cannot be created there, because it belongs to no volume. Callers + that used to receive the shared root for an empty path now receive None, + which is the honest answer. The quarantine was the fix for C5a; what actually mattered in it — no overwrite, a name allowlist, and confinement — is kept by this plus the caller's existing checks. """ - rel = (rel or "").strip().strip("/") - if not rel: - return shared_root - parts = [seg for seg in rel.split("/") if seg not in ("", ".")] + found = roots.split(rel or "") + if found is None: + return None + root, tail = found + if not root.available: + return None + parts = [seg for seg in tail.split("/") if seg not in ("", ".")] if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts): return None try: - target = (shared_root / Path(*parts)).resolve() - root = shared_root.resolve() + target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve() + base = root.path.resolve() except OSError: return None - if target != root and root not in target.parents: + if target != base and base not in target.parents: return None return target + def _extract_dtls_fingerprint(sdp: str) -> bytes: """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes.""" for line in sdp.splitlines(): @@ -439,6 +453,10 @@ class WebRTCPeerSession: self._do_invite_create(msg) elif mtype == MNP.MEMBER_REVOKE: self._do_member_revoke(msg) + elif mtype == MNP.MEMBER_UNPIN: + self._do_member_unpin(msg) + elif mtype == MNP.GEK_ROTATE: + self._do_gek_rotate(msg) elif mtype == MNP.KEYPAIR_BUNDLE_STORE: self._spawn(self._do_keypair_bundle_store(msg)) elif mtype == MNP.KEYPAIR_BUNDLE_DELETE: @@ -1023,10 +1041,9 @@ class WebRTCPeerSession: but it writes to the operator's disk, so it is audited like one. """ ctx = self._group_ctx() - shared_root = ctx.get("shared_root") - if not shared_root: - self._send({"type": "error", "detail": "No shared directory", - "filename": filename}) + roots: RootSet | None = ctx.get("roots") + if not roots: + self._send({"type": "error", "detail": "No shared directory"}) return name = str(msg.get("name", "")).strip() @@ -1034,12 +1051,21 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "Invalid directory name"}) return - parent = safe_subdir(shared_root, msg.get("dir") or "") + # The virtual root is not a directory on anyone's disk, so a member + # cannot create one there — that would be adding a root, which is the + # operator's configuration and not a file operation. + parent_rel = (msg.get("dir") or "").strip("/") + if not parent_rel: + self._send({"type": "error", + "detail": "Choose a folder to create this in"}) + return + + parent = safe_subdir(roots, parent_rel) if parent is None or not parent.is_dir(): self._send({"type": "error", "detail": "Invalid directory"}) return - target = safe_subdir(shared_root, f"{(msg.get('dir') or '').strip('/')}/{name}") + target = safe_subdir(roots, f"{parent_rel}/{name}") if target is None: self._send({"type": "error", "detail": "Invalid directory"}) return @@ -1048,14 +1074,20 @@ class WebRTCPeerSession: return target.mkdir(parents=False) - log.info("Directory created by %s: %s", self._user_id[:8], - target.relative_to(shared_root)) - self._audit("dir_create", str(target.relative_to(shared_root))) + virtual = roots.virtual_of(target) or f"{parent_rel}/{name}" + log.info("Directory created by %s: %s", self._user_id[:8], virtual) + self._audit("dir_create", virtual) self._send({ "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION, - "dir": str(target.relative_to(shared_root)), + "dir": virtual, }) + @staticmethod + def _names_a_root(roots: RootSet, rel: str) -> bool: + """True when `rel` is a bare root name rather than something inside one.""" + found = roots.split(rel or "") + return found is not None and not found[1] + def _do_dir_delete(self, msg: dict) -> None: """ Remove an empty directory, for the node operator. @@ -1068,14 +1100,17 @@ class WebRTCPeerSession: operator deletes the files first and sees what they are losing. """ ctx = self._group_ctx() - shared_root = ctx.get("shared_root") - if not shared_root: - self._send({"type": "error", "detail": "No shared directory", - "filename": filename}) + roots: RootSet | None = ctx.get("roots") + if not roots: + self._send({"type": "error", "detail": "No shared directory"}) return - target = safe_subdir(shared_root, msg.get("dir") or "") - if target is None or target == shared_root: + rel = (msg.get("dir") or "").strip("/") + target = safe_subdir(roots, rel) + # A root itself is not deletable here: removing one is a configuration + # change, and doing it through a file operation would leave the group + # config naming a directory nobody can reach. + if target is None or self._names_a_root(roots, rel): self._send({"type": "error", "detail": "Invalid directory"}) return if not target.is_dir(): @@ -1089,16 +1124,17 @@ class WebRTCPeerSession: return self._issue_admin_challenge( - OP_DIR_DELETE, str(target.relative_to(shared_root))) + OP_DIR_DELETE, roots.virtual_of(target) or rel) async def _admin_exec_dir_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: rel = pending["subject"] ctx = self._group_ctx() - shared_root = ctx.get("shared_root") - target = safe_subdir(shared_root, rel) if shared_root else None - if target is None or target == shared_root or not target.is_dir(): + roots: RootSet | None = ctx.get("roots") + target = safe_subdir(roots, rel) if roots else None + if (target is None or self._names_a_root(roots, rel) + or not target.is_dir()): self._send({"type": "error", "detail": "Not a directory"}) return @@ -1145,6 +1181,95 @@ class WebRTCPeerSession: return self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id) + def _do_gek_rotate(self, msg: dict) -> None: + """ + Ask for a new group key. Operator only, and signed. + + This is what actually removes a revoked member's access: revocation + stops the node serving the *next* key, and they still hold the current + one. The node generates the replacement itself — nothing arriving here + contributes key material, which is what the C5b rule is about. + """ + if not self._group_id: + self._send({"type": "error", "detail": "No group on this connection"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_GEK_ROTATE, self._group_id) + + async def _admin_exec_gek_rotate( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"gek_rotate:{pending['subject'][:8]}") + return + try: + result = await self._run_op( + ops.set_gek, pending["subject"], rotate=True) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("gek_rotate", pending["subject"]) + self._send({ + "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION, + "group_id": pending["subject"], + "authorized_members": result.get("authorized_members", 0), + # Said plainly, because rotating is the step people skip: content + # already downloaded stays readable to whoever holds it. + "note": "members re-receive the key on their next connect; content " + "already downloaded is unaffected", + }) + + def _do_member_unpin(self, msg: dict) -> None: + """Forget a pinned identity, so someone can pair again with a new key.""" + user_id = str(msg.get("user_id", "")).strip() + if not user_id: + self._send({"type": "error", "detail": "Missing user_id"}) + return + if user_id == self._user_id: + # Unpinning yourself over the connection your pin authorizes would + # end that connection's authority mid-operation. + self._send({"type": "error", "detail": "Cannot unpin yourself"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_MEMBER_UNPIN, user_id) + + async def _admin_exec_member_unpin( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + user_id = pending["subject"] + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"member_unpin:{user_id[:8]}") + return + try: + await self._run_op(ops.unpin_member, user_id) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("member_unpin", user_id) + self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION, + "user_id": user_id}) + + async def _run_op(self, fn, *args, **kwargs): + """ + Call an operation from `meshbay_node.ops` with the daemon's own view. + + The transport carries its own context and the loopback API carries the + daemon state; they overlap but are not the same dict. Handing the MNP + path a *second* set of lookups is exactly how two implementations of one + operation start disagreeing — C1 and C6 one size down — so the daemon + publishes its state here and both adapters call the same function. + """ + state = self._ctx.get("daemon_state") + if state is None: + raise ops.OpError("Node state not available", status=503) + return await fn(state, *args, **kwargs) + async def _admin_exec_member_revoke( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: @@ -1285,24 +1410,39 @@ class WebRTCPeerSession: # Directories are not index entries, so the client used to infer them # from file paths — which means a folder someone just created, or one # they emptied, simply did not exist as far as the UI was concerned. - "dirs": self._list_dirs(ctx.get("shared_root")), + "dirs": self._list_dirs(ctx.get("roots")), + # Which top-level folders are roots, and whether each is readable. + # A frozen root's files stay listed, so without this a member cannot + # tell "the drive is unplugged" from "it is all still there". + "roots": ctx["roots"].describe() if ctx.get("roots") else [], }) @staticmethod - def _list_dirs(shared_root: Path | None) -> list[str]: - """Directories under the shared root, relative and sorted.""" - if not shared_root: - return [] - out = [] - try: - for path in sorted(shared_root.rglob("*")): - if path.is_dir() and not path.name.startswith("."): - rel = path.relative_to(shared_root) - if not any(part.startswith(".") for part in rel.parts): - out.append(str(rel)) - except OSError: + def _list_dirs(roots: RootSet | None) -> list[str]: + """ + Every directory in the group, as members address them, sorted. + + Each root appears as a directory in its own right, so a root holding no + files yet is still somewhere a member can navigate to and upload into. + An unavailable root is listed too — its content is frozen, not gone, and + hiding it would look exactly like deletion. + """ + if not roots: return [] - return out[:2000] + out: list[str] = [] + for root in roots: + out.append(root.name) + if not root.available: + continue + try: + for path in sorted(root.path.rglob("*")): + if path.is_dir() and not path.name.startswith("."): + rel = path.relative_to(root.path) + if not any(part.startswith(".") for part in rel.parts): + out.append(f"{root.name}/{rel.as_posix()}") + except OSError: + continue + return sorted(out)[:2000] async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() @@ -1314,7 +1454,7 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -1372,7 +1512,7 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -1542,20 +1682,41 @@ class WebRTCPeerSession: "filename": filename}) return - shared_root = ctx.get("shared_root") - if not shared_root: - self._send({"type": "error", "detail": "No shared directory", + roots: RootSet | None = ctx.get("roots") + upload_root = roots.upload_root if roots else None + if upload_root is None: + # Refused, never guessed. With several roots, picking one would send + # a member's file to a disk the operator did not intend, and that is + # discovered weeks later. + self._send({"type": "error", + "detail": "No upload folder is configured for this group", + "filename": filename}) + return + if not upload_root.available: + # The designated root's volume is absent. Falling back to another + # root would scatter uploads across disks depending on what happened + # to be plugged in. + self._send({"type": "error", + "detail": f"The upload folder ({upload_root.name}) is " + f"currently unavailable", "filename": filename}) return - # One destination, chosen here and not by the client: uploads/ at the root - # of the shared directory. C5a is still honoured — the name passed the - # allowlist above, and an existing file is never replaced, which was the - # real defect (overwriting a file also made the attacker its recorded - # uploader, and therefore able to delete it). - rel_dir = UPLOAD_DIR_NAME - target_dir = shared_root / UPLOAD_DIR_NAME - target_dir.mkdir(parents=True, exist_ok=True) + # One destination, chosen by the operator and not by the client: + # uploads/ inside the group's designated root. C5a is still honoured — + # the name passed the allowlist above, and an existing file is never + # replaced, which was the real defect (overwriting a file also made the + # attacker its recorded uploader, and therefore able to delete it). + rel_dir = f"{upload_root.name}/{UPLOAD_DIR_NAME}" + target_dir = upload_root.path / UPLOAD_DIR_NAME + try: + target_dir.mkdir(parents=True, exist_ok=True) + except OSError as e: + log.warning("Cannot create upload folder in root %r: %s", + upload_root.name, e) + self._send({"type": "error", "detail": "Upload folder unavailable", + "filename": filename}) + return upload_key = f"{rel_dir}/{filename}" state = self._uploads.get(upload_key) @@ -1789,6 +1950,12 @@ class WebRTCPeerSession: elif pending["op"] == OP_INVITE_CREATE: self._spawn( self._admin_exec_invite_create(pending, transcript, sig_bytes)) + elif pending["op"] == OP_GEK_ROTATE: + self._spawn( + self._admin_exec_gek_rotate(pending, transcript, sig_bytes)) + elif pending["op"] == OP_MEMBER_UNPIN: + self._spawn( + self._admin_exec_member_unpin(pending, transcript, sig_bytes)) else: self._send({"type": "error", "detail": "Unknown admin operation"}) @@ -1865,7 +2032,7 @@ class WebRTCPeerSession: }) def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if file_path.exists(): file_path.unlink() log.info("File deleted: %s", entry.name) @@ -2049,7 +2216,7 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -2265,7 +2432,7 @@ class WebRTCTransport: Manages WebRTC peer connections for browser clients. Usage: - transport = WebRTCTransport(sk_node, hub_pk_pem, gek, shared_root, index) + transport = WebRTCTransport(sk_node, hub_pk_pem, gek, roots, index) answer_sdp = await transport.handle_offer(offer_sdp, peer_id) # Return answer_sdp to the browser via hub signaling """ @@ -2275,7 +2442,7 @@ class WebRTCTransport: sk_node: Ed25519PrivateKey, hub_pk_pem: bytes, gek: bytes, - shared_root: Path, + roots: RootSet, index: GroupIndex, groups: dict[str, dict] | None = None, denylist: Any | None = None, @@ -2286,7 +2453,7 @@ class WebRTCTransport: "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, - "shared_root": shared_root, + "roots": roots, "index": index, "_peers": {}, # None means "the operator said nothing" — the default applies. It |