diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 463 |
1 files changed, 390 insertions, 73 deletions
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 aaf1ecf..938ce3b 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -33,6 +33,7 @@ import time from pathlib import Path from typing import Any +import blake3 import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel @@ -64,6 +65,8 @@ from meshbay_common.adminop import ( OP_MEMBER_UPLOAD, OP_APPS_ENABLED, OP_SET_SCAN_SETTINGS, + OP_TMDB_CONFIG, + OP_VIDEO_ROOT, OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, @@ -84,10 +87,15 @@ from meshbay_common.join import ( join_transcript, ) from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes -from meshbay_common.protocol import MNP +from meshbay_common.protocol import MNP, index_entry_wire from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import ops +# Re-imported under its original name: every call site and existing test in +# this module still refers to it as `_probe_video`. The implementation lives +# in media_probe.py so the indexer package (imported just above) can call it +# too, for index-time enrichment, without a circular import. +from meshbay_node.media_probe import probe_video as _probe_video from meshbay_node.roots import ( RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, ) @@ -163,59 +171,6 @@ STREAM_CREDIT_TIMEOUT = 120 # charged for a slot within this, rather than within the timeout. STREAM_CREDIT_POLL = 3 -_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} - - -async def _probe_video(path: str) -> tuple[str | None, float, bool]: - """ - Probe video file with ffprobe, return (MSE codec string, duration, - has_audio). - - The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or - absent — never the source's real audio codec — because _stream_video_ - inner always transcodes audio to AAC and never copies it: MSE in every - mainstream browser only decodes AAC/Opus, and a source codec outside - that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at - worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the - fragmented MP4 header ("Cannot write moov atom before EAC3 packets - parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video stays - whatever it actually is: it is always copied, never transcoded. - """ - import json as _json - proc = await asyncio.create_subprocess_exec( - "ffprobe", "-v", "error", - "-show_entries", "stream=codec_name,profile,level,codec_type", - "-show_entries", "format=duration", - "-of", "json", path, - stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - ) - stdout, _ = await proc.communicate() - info = _json.loads(stdout) - duration = float(info.get("format", {}).get("duration", 0)) - - v_codec = "" - has_audio = False - for s in info.get("streams", []): - if s.get("codec_type") == "video" and not v_codec: - cn = s.get("codec_name", "") - if cn == "h264": - p = _H264_PROFILES.get(s.get("profile", "High"), "64") - lvl = int(s.get("level", 40)) - v_codec = f"avc1.{p}00{lvl:02x}" - elif cn == "hevc": - v_codec = "hev1.1.6.L93.B0" - elif cn == "vp9": - v_codec = "vp09.00.10.08" - elif cn == "av1": - v_codec = "av01.0.01M.08" - elif s.get("codec_type") == "audio": - has_audio = True - - if not v_codec: - return None, duration, has_audio - codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec - return codec, duration, has_audio - def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) @@ -417,6 +372,12 @@ class WebRTCPeerSession: self._do_apps_enabled(msg) elif mtype == MNP.SET_SCAN_SETTINGS: self._do_set_scan_settings(msg) + elif mtype == MNP.TMDB_CONFIG: + self._do_tmdb_config(msg) + elif mtype == MNP.VIDEO_ROOT: + self._do_video_root(msg) + elif mtype == MNP.MEDIA_META_REQ: + self._spawn(self._do_media_meta_request(msg)) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: @@ -667,6 +628,20 @@ class WebRTCPeerSession: # setting (or one whose context has not loaded it yet) hides # nothing. "enabled_apps": list(self._group_ctx().get("enabled_apps") or []), + # Which folder the Videos app treats as its entry point for + # this group — "" means the whole group index. + "video_root": self._group_ctx().get("video_root") or "", + # Node-wide (not per-group), same "read once, kept current by + # the signed op" shape — surfaced here rather than only via + # tmdb_config_ack so a client that connects after the operator + # already configured it does not have to wait for a live change + # to find out (docs/mediacenter.md §5.5). + "tmdb_enabled": bool( + self._ctx.get("daemon_state", {}).get("tmdb_enabled", True)), + "tmdb_token_customized": bool( + self._ctx.get("daemon_state", {}).get("tmdb_token_customized", False)), + "tmdb_language": str( + self._ctx.get("daemon_state", {}).get("tmdb_language") or ""), # So a client that connects mid-scan shows the indexing state # immediately, instead of waiting for the next periodic # INDEX_PROGRESS push. Never a path or filename — see @@ -1623,10 +1598,13 @@ class WebRTCPeerSession: except Exception: pass - # Every "application" a group can show — Chat and Files today. Videos, - # Music, Photos join this set (and apps.js's registry, client-side) when - # they land; nothing else about this handler changes. - ALLOWED_APPS = frozenset({"chat", "files"}) + # Every "application" a group can show. Music, Photos join this set (and + # apps.js's registry, client-side) when they land; nothing else about + # this handler changes. DEFAULT_APPS (roster.py) deliberately does not + # include "video" — it is the first app with outbound third-party + # network calls (once TMDB is on), so an operator opts a group in + # explicitly rather than getting it for free (docs/mediacenter.md §5.6). + ALLOWED_APPS = frozenset({"chat", "files", "video"}) def _do_apps_enabled(self, msg: dict) -> None: """ @@ -1677,6 +1655,125 @@ class WebRTCPeerSession: except Exception: pass + def _do_tmdb_config(self, msg: dict) -> None: + """ + Turn TMDB lookups on/off node-wide, optionally set (or clear) a + custom API token, and optionally set the language TMDB is queried + in (e.g. "fr-FR") — one for the whole node, same reasoning as the + token: one shared cache, not a per-viewer request. Signed like the + rest: this turns on outbound third-party network traffic the node + did not have before the Videos app (docs/mediacenter.md §5.5, §8) + — an unsigned toggle would let any member turn on egress the + operator never agreed to. + """ + enabled = msg.get("enabled") + if not isinstance(enabled, bool): + self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) + return + token = msg.get("token") + if token is not None and not isinstance(token, str): + self._send({"type": "error", "detail": "Invalid 'token'"}) + return + language = msg.get("language") + if language is not None and not isinstance(language, str): + self._send({"type": "error", "detail": "Invalid 'language'"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + # The subject is the signed, audited, human-shown string — it must + # never contain the token itself (it would end up in the audit log + # in plaintext). The actual token travels only in `payload`, which + # is node-side context, never re-sent or re-verified from the wire. + # The language is not a secret, so it travels in the subject itself. + subject = (f"enabled={enabled},custom_token={'yes' if token else 'no'}," + f"language={language or 'default'}") + self._issue_admin_challenge( + OP_TMDB_CONFIG, subject, + payload={"enabled": enabled, "token": token, "language": language}, + group_id="") + + async def _admin_exec_tmdb_config( + 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"tmdb_config:{pending['subject']}") + return + p = pending.get("payload") or {} + try: + result = await self._run_op( + ops.set_tmdb_config, p.get("enabled", True), p.get("token"), p.get("language")) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("tmdb_config", pending["subject"]) + + # Node-wide setting: every connected peer in every group is told, not + # just this group's peers (unlike apps_enabled/member_upload). + notice = { + "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION, + "enabled": result["enabled"], "token_customized": result["token_customized"], + "language": result["language"], + } + for gctx in self._ctx.get("groups", {}).values(): + for session in list(gctx.get("_peers", {}).values()): + try: + session._send(notice) + except Exception: + pass + + def _do_video_root(self, msg: dict) -> None: + """ + Which folder (possibly a subfolder of a shared root) the Videos app + treats as its entry point for this group. Signed like apps_enabled: + it decides what every member's Videos tab shows. + + An empty path is always accepted (it means "the whole group index", + today's behaviour). A non-empty path must resolve to a real, + currently-readable directory — validated against the group's own + roots the same way directory creation/deletion already is, so a + stale or mistyped path is refused before a signature is even asked + for. + """ + path = msg.get("path") + if not isinstance(path, str): + self._send({"type": "error", "detail": "Missing or invalid 'path'"}) + return + path = path.strip("/") + if path: + ctx = self._group_ctx() + resolved = ctx["roots"].resolve(path) if ctx.get("roots") else None + if not resolved or not resolved.is_dir(): + self._send({"type": "error", "detail": "Not a directory in this group"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_VIDEO_ROOT, path) + + async def _admin_exec_video_root( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + path = 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"video_root:{path}") + return + try: + await self._run_op(ops.set_video_root, self._group_id or "", path) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("video_root", path) + + notice = {"type": MNP.VIDEO_ROOT_ACK, "v": MNP_VERSION, "path": path} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + # Reconcile's backstop and the watchdog debounce (indexer.py # DirectoryIndexer) — how hard the node works on the operator's own # disk, not a member-facing permission. Signed for the same reason as @@ -2138,14 +2235,7 @@ class WebRTCPeerSession: def _do_index_sync(self) -> None: ctx = self._group_ctx() idx = ctx["index"] - entries = [ - { - "id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at, - "uploader_id": e.uploader_id, - } - for e in idx.entries - ] + entries = [index_entry_wire(e) for e in idx.entries] self._send({ "type": MNP.INDEX_SYNC, "v": MNP_VERSION, @@ -2189,12 +2279,38 @@ class WebRTCPeerSession: continue return sorted(out)[:2000] + async def _try_serve_thumbnail( + self, thumb_hash: str, chunk_index: int, gek: bytes | None, + ) -> dict | None: + """ + docs/mediacenter.md §5.3: a thumbnail is served through the same + chunked file_req path as a real file, resolved against the media + cache instead of the index when the id doesn't match a file. Always + a single chunk in practice (a thumbnail-sized JPEG never approaches + CHUNK_SIZE) — a request for any chunk beyond 0 is just a miss. + """ + media_cache = self._ctx.get("media_cache") + if media_cache is None or chunk_index != 0: + return None + jpeg = await media_cache.get_thumb(thumb_hash) + if jpeg is None: + return None + return _encrypt_chunk_bytes( + self._ctx["sk_node"], gek, jpeg, 0, + bytes.fromhex(thumb_hash), thumb_hash, + ) + async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) if not entry: + thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek")) + if thumb is not None: + log.debug("file_req file_id=%s chunk=%s: served as thumbnail", file_id[:16], chunk_index) + self._send(thumb) + return log.warning("File not found: %s", file_id[:16]) self._send({"type": "error", "detail": "File not found"}) return @@ -2235,6 +2351,191 @@ class WebRTCPeerSession: if chunk_index == 0: self._audit("file_download", entry.name) + @staticmethod + async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None: + """ + Downloads a TMDB poster/backdrop once, caches it under its own + blake3 like a video thumbnail (docs/mediacenter.md §5.4), and + returns the hash a client then fetches via the normal file_req/ + chunk path (§5.3) — no client ever contacts image.tmdb.org directly. + + Checked by the synthetic `tmdb:{poster_path}` id *before* touching + the network: without this, every `media_meta_req` for an + already-cached file re-downloaded the same poster from TMDB (found + live — a poster grid re-fetched both a show's poster and backdrop + from TMDB on every single visit, real added latency and needless + outbound traffic for an image that never changes). + """ + if not poster_path: + return None + synthetic_id = f"tmdb:{poster_path}" + cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id) + if cached_hash is not None: + return cached_hash + content = await tmdb_client.fetch_image(tmdb_client.poster_url(poster_path)) + if content is None: + return None + thumb_hash = blake3.blake3(content).hexdigest() + await media_cache.put_thumb(thumb_hash, synthetic_id, content) + return thumb_hash + + async def _do_media_meta_request(self, msg: dict) -> None: + """ + docs/mediacenter.md §5.4: TMDB metadata for one path, resolved from + the group's index (root+relpath the client already knows from + index_sync/index_delta — never a raw filesystem path off the wire). + """ + path = msg.get("path") + log.debug("media_meta_req path=%r", path) + if not isinstance(path, str) or not path: + self._send({"type": "error", "detail": "Missing path"}) + return + ctx = self._group_ctx() + entry = ctx["index"].get_entry_by_path(path) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + media_cache = self._ctx.get("media_cache") + tmdb_client = self._ctx.get("tmdb_client") + if media_cache is None or tmdb_client is None: + self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, + "path": path, "confidence": 0}) + return + + is_show = entry.season is not None and entry.episode is not None + media_type = "tv" if is_show else "movie" + + cached = await media_cache.get_file_tmdb(entry.id) + meta = None + tmdb_id = None + if cached is not None: + tmdb_id, media_type = cached + meta = await media_cache.get_tmdb_meta(tmdb_id, media_type) + + if meta is None: + result, ratio = await self._tmdb_search(tmdb_client, entry, is_show) + if result is None or ratio < 0.6: + self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, + "path": path, "confidence": 0}) + return + tmdb_id = str(result["id"]) + meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, result) + await media_cache.set_file_tmdb(entry.id, tmdb_id, media_type) + await media_cache.set_tmdb_meta(tmdb_id, media_type, meta) + + poster_thumb_hash = await self._fetch_and_cache_poster( + media_cache, tmdb_client, meta.get("poster_path")) + backdrop_thumb_hash = await self._fetch_and_cache_poster( + media_cache, tmdb_client, meta.get("backdrop_path")) + log.debug("media_meta_req path=%r: replying tmdb_id=%s poster=%s backdrop=%s", + path, tmdb_id, poster_thumb_hash, backdrop_thumb_hash) + + resp = { + "type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "path": path, + "tmdb_id": tmdb_id, "title": meta.get("title"), + "original_title": meta.get("original_title"), + "overview": meta.get("overview"), + "poster_thumb_hash": poster_thumb_hash, + "backdrop_thumb_hash": backdrop_thumb_hash, + "release_date": meta.get("release_date"), + "first_air_date": meta.get("first_air_date"), + "genres": meta.get("genres", []), + "vote_average": meta.get("vote_average"), + "runtime": meta.get("runtime"), + "cast": meta.get("cast", []), + "director": meta.get("director"), + "confidence": meta.get("confidence", 1.0), + } + if is_show: + resp["season"] = entry.season + resp["episode"] = entry.episode + self._send(resp) + + async def _tmdb_search(self, tmdb_client, entry, is_show: bool): + """ + §3.3's retry ladder: the parsed title first, then a couple of + generic, non-per-title fallbacks — never re-ranking TMDB's own + top result locally (§3.3's last row). + """ + from meshbay_node.indexer import title_parse + + if is_show: + title = entry.display_title or title_parse.naive_title(entry.name) + result, ratio = await tmdb_client.search_tv(title) + if result is None or ratio < 0.6: + naive = title_parse.naive_title(entry.name) + if naive != title: + result, ratio = await tmdb_client.search_tv(naive) + return result, ratio + + parsed = title_parse.parse_movie_filename(entry.name) + title = entry.display_title or parsed.display_title or parsed.naive_title + result, ratio = await tmdb_client.search_movie(title, parsed.year) + if result is not None and ratio >= 0.6: + return result, ratio + for candidate in filter(None, [parsed.alt_title, parsed.naive_title, + *title_parse.sequel_variants(title)]): + if candidate == title: + continue + result2, ratio2 = await tmdb_client.search_movie(candidate, parsed.year) + if result2 is not None and ratio2 > ratio: + result, ratio = result2, ratio2 + if ratio >= 0.6: + break + return result, ratio + + @staticmethod + async def _tmdb_build_meta(tmdb_client, tmdb_id: str, media_type: str, result: dict) -> dict: + """ + `result` (the search hit) only carries `genre_ids` and no `runtime` + at all — the full details endpoint is the actual source for those, + falling back to the search result for anything details somehow + lacks (never expected in practice, just avoids a KeyError-shaped + surprise if TMDB's response ever varies). + """ + details = (await tmdb_client.tv_details(tmdb_id) if media_type == "tv" + else await tmdb_client.movie_details(tmdb_id)) or result + # TMDB doesn't fall back server-side for a field with no translation + # in the configured language — it returns "" (or an empty list) for + # it, not the English text (confirmed live: a French query left + # `overview` empty for a title TMDB has no French translation for). + # The TMDB website covers exactly this gap client-side, by falling + # back to English per field rather than discarding an otherwise-good + # localized response over one empty one — mirrored here the same + # way, at field granularity, not by abandoning the whole response. + if not details.get("overview") or not details.get("poster_path") or not details.get("genres"): + fallback = (await tmdb_client.tv_details(tmdb_id, language="en-US") if media_type == "tv" + else await tmdb_client.movie_details(tmdb_id, language="en-US")) or {} + details = {**fallback, **{k: v for k, v in details.items() if v not in (None, "", [])}} + credits = (await tmdb_client.tv_credits(tmdb_id) if media_type == "tv" + else await tmdb_client.movie_credits(tmdb_id)) + cast = [{"name": c.get("name"), "character": c.get("character")} + for c in (credits or {}).get("cast", [])[:10]] + director = None + if media_type == "movie": + director = next( + (c.get("name") for c in (credits or {}).get("crew", []) + if c.get("job") == "Director"), None) + runtime = details.get("runtime") + if runtime is None and media_type == "tv": + episode_run_times = details.get("episode_run_time") or [] + runtime = episode_run_times[0] if episode_run_times else None + return { + "title": details.get("title") or details.get("name"), + "original_title": details.get("original_title") or details.get("original_name"), + "overview": details.get("overview"), + "poster_path": details.get("poster_path"), + "backdrop_path": details.get("backdrop_path"), + "release_date": details.get("release_date"), + "first_air_date": details.get("first_air_date"), + "genres": [g.get("name") for g in details.get("genres", []) if g.get("name")], + "vote_average": details.get("vote_average"), + "runtime": runtime, + "cast": cast, + "director": director, + } + def _do_stream_segment(self, msg: dict) -> None: self._spawn(self._do_stream_segment_async(msg)) @@ -2738,6 +3039,12 @@ class WebRTCPeerSession: elif pending["op"] == OP_SET_SCAN_SETTINGS: self._spawn( self._admin_exec_set_scan_settings(pending, transcript, sig_bytes)) + elif pending["op"] == OP_TMDB_CONFIG: + self._spawn( + self._admin_exec_tmdb_config(pending, transcript, sig_bytes)) + elif pending["op"] == OP_VIDEO_ROOT: + self._spawn( + self._admin_exec_video_root(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_ADD: self._spawn( self._admin_exec_root_add(pending, transcript, sig_bytes)) @@ -3009,7 +3316,7 @@ class WebRTCPeerSession: file_hash = bytes.fromhex(entry.id) try: - codec_str, duration, has_audio = await _probe_video(str(file_path)) + codec_str, duration, has_audio, _width, _height = await _probe_video(str(file_path)) except Exception as e: self._send({"type": "error", "detail": f"Probe failed: {e}"}) return @@ -3220,18 +3527,14 @@ class WebRTCPeerSession: await self._pc.close() -def _read_and_encrypt( +def _encrypt_chunk_bytes( sk_node: Ed25519PrivateKey, gek: bytes, - file_path: Path, + plaintext: bytes, chunk_index: int, file_hash: bytes, file_id: str = "", ) -> dict: - with open(file_path, "rb") as f: - f.seek(chunk_index * CHUNK_SIZE) - plaintext = f.read(CHUNK_SIZE) - ckey = chunk_key_aes(gek, file_hash, chunk_index) nonce, ct = encrypt_chunk_aes(ckey, plaintext) @@ -3249,6 +3552,20 @@ def _read_and_encrypt( } +def _read_and_encrypt( + sk_node: Ed25519PrivateKey, + gek: bytes, + file_path: Path, + chunk_index: int, + file_hash: bytes, + file_id: str = "", +) -> dict: + with open(file_path, "rb") as f: + f.seek(chunk_index * CHUNK_SIZE) + plaintext = f.read(CHUNK_SIZE) + return _encrypt_chunk_bytes(sk_node, gek, plaintext, chunk_index, file_hash, file_id) + + class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. |