From 45f4c1bc84bc69d908860a50618fbd14fe332224 Mon Sep 17 00:00:00 2001 From: foxyz Date: Wed, 7 Oct 2026 01:57:04 +0300 Subject: [PATCH] Jellyfin refs on the wire + guest relay (proto 7) - Media can be a Jellyfin ref {serverId, itemId, mediaSourceId, title, subtitle, runtimeSecs}; the server never stores or broadcasts a URL or token for it. - Optional server-side Jellyfin credential (serve map / --jellyfin file) enables GET /j///: range-faithful proxy of the active item only, exact client header fields, upstream aborted on disconnect, bytes counted. welcome.relayUrl / state.relay advertise it. - Members report jf access (own account vs relay). - /m now sends the real Content-Type (Relic overrode it). Co-Authored-By: Claude Opus 5.5 --- docs/ARCHITECTURE.md | 43 +- docs/PROTOCOL.md | 63 ++- packages/core/lib/src/config.dart | 136 +++++- packages/core/lib/src/constants.dart | 2 +- packages/core/lib/src/protocol.dart | 132 +++++- packages/core/lib/src/sync_engine.dart | 13 +- packages/core/test/jellyfin_ref_test.dart | 220 ++++++++++ packages/server/bin/server.dart | 17 + packages/server/lib/server.dart | 1 + packages/server/lib/src/jellyfin_relay.dart | 160 +++++++ packages/server/lib/src/media_endpoint.dart | 16 +- packages/server/lib/src/server_hub.dart | 73 +++- packages/server/lib/src/session.dart | 3 + packages/server/lib/src/sync_server.dart | 43 +- .../server/test/jellyfin_relay_live_test.dart | 91 ++++ packages/server/test/jellyfin_relay_test.dart | 408 ++++++++++++++++++ packages/server/test/server_auth_test.dart | 4 +- packages/server/test/session_test.dart | 4 +- 18 files changed, 1363 insertions(+), 66 deletions(-) create mode 100644 packages/core/test/jellyfin_ref_test.dart create mode 100644 packages/server/lib/src/jellyfin_relay.dart create mode 100644 packages/server/test/jellyfin_relay_live_test.dart create mode 100644 packages/server/test/jellyfin_relay_test.dart diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 675ac2d..db131a8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -8,7 +8,7 @@ Plain overview first: [HOW-IT-WORKS.md](HOW-IT-WORKS.md). Wire format: | Path | What | Rule | |---|---|---| | `packages/core` | protocol messages, sync engine (`SyncEngine`), constants, config, Range parsing, logging, Jellyfin client | pure Dart, no Flutter | -| `packages/server` | Relic HTTP + WebSocket server (`SyncServer`, `ServerHub`, `Session`, media endpoint), the `serve()` entrypoint, `bin/server.dart` (dedicated server) | pure Dart, no Flutter | +| `packages/server` | Relic HTTP + WebSocket server (`SyncServer`, `ServerHub`, `Session`, media endpoint, Jellyfin relay), the `serve()` entrypoint, `bin/server.dart` (dedicated server) | pure Dart, no Flutter | | `apps/desktop` | Flutter app: UI, `SyncClient`, `SessionCoordinator`, `PlaybackController` (media_kit/mpv), `ServerIsolate`, updater | ships with `watch-together-server` | | `tool` | harnesses: `range_logger`, `shaper`, `fake_viewer`, `e2e_check` | dev only | | `deploy` | systemd unit for the dedicated server | | @@ -40,7 +40,7 @@ all sockets and returns, and the isolate exits. - **Dedicated server**: `bin/server.dart` compiled with `dart compile exe` and shipped next to the app as `watch-together-server`. It parses flags (`--port`, `--secret`, `--name`, `--media`, `--advertise-host`, - `--config`; default port 8080) into the same map and calls `serve`; + `--jellyfin `, `--config`; default port 8080) into the same map and calls `serve`; SIGINT and SIGTERM stop it. A Flutter binary cannot run without a display, which is why the server is its own build. @@ -137,7 +137,7 @@ Additive wire changes only; bump `proto` and PROTOCOL.md in the same commit. ## Media -Two sources, same sync. +Three sources, same sync. - **Link**: the media field holds an http(s) URL. It travels as `media.url` in `swap` and `state`, and as `mediaUrl` in `welcome`. Each app plays it @@ -153,6 +153,12 @@ Two sources, same sync. (`bytesServed`); nothing is stored. The host plays through loopback like everyone else. +- **Jellyfin item**: the media is a ref (`{serverId, itemId, ...}`, wire in + [PROTOCOL.md](PROTOCOL.md) "Jellyfin media"); the server never stores or + sends a URL or token for it. Each app resolves it with its own account and + direct-plays the static stream; guests use the relay (see "Jellyfin + relay"). + Range table (`resolveMediaRequest`, tested byte-exact): | Request | Response | @@ -278,12 +284,41 @@ Gotchas (Jellyfin 10.11): failure). - Jellyfin overwrites a token's stored device name and app version with the header's. Only use tokens this app obtained itself; a relay using a - viewer's token must send that viewer's client fields. + viewer's token must send that viewer's client fields (the server relay + sends exactly its credential's fields). - Static streams (`/Videos/{id}/stream?static=true`) serve the original file with byte ranges. The share URL has no credentials; each side adds its own `api_key` (`authorizeUrl`). - Ticks: 10,000,000 per second. +## Jellyfin relay + +For viewers without an account on the item's Jellyfin +(`packages/server/lib/src/jellyfin_relay.dart`). The server may hold one +`JellyfinCredential` (config key `jellyfin`: `baseUrl`, `serverId`, `token`, +`client`, `device`, `deviceId`, `version`; primitives, so it crosses the +isolate boundary in the `serve()` map). The hosting app fills it with its +own app-obtained token and that token's device fields; the dedicated server +reads it from `--jellyfin ` (YAML/JSON with those keys, keeps the +token off the command line) or a `jellyfin:` map in `--config`. An empty +`serverId` is looked up at start (`/System/Info/Public`); if that fails the +relay stays off. The token is never logged (`toString` and config parse +errors leave it out). + +- Available (`state.relay`) only while the current media is a ref whose + `serverId` equals the credential's. Only that item is relayed, only to a + valid, unkicked viewer token: no open proxy. +- Each request opens `/Videos/{id}/stream?static=true&mediaSourceId=` with + `Authorization: MediaBrowser` built from exactly the configured fields + (Jellyfin rewrites a token's device fields from the header), forwards + `Range` untouched and `Accept-Encoding: identity`, and returns Jellyfin's + status and range headers. Bytes stream through (no buffering) and count + toward `bytesServed` as sent. A viewer disconnect cancels the upstream + response, which closes that connection (tested: `HttpClientRequest.abort` + does nothing once the response has arrived; cancelling the response + subscription does). +- No progress reporting in the server: watched state is per client. + ## Desktop UI One `HomeScreen` (`screens/home/`) with a single top bar (`top_bar.dart`): diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index bfc350c..cea4ef0 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -1,4 +1,4 @@ -# Protocol (proto = 6) +# Protocol (proto = 7) Wire format only. Behavior (gate, clock, timeouts) is in [ARCHITECTURE.md](ARCHITECTURE.md). Source of truth: @@ -21,19 +21,22 @@ with different `proto` refuse each other at `hello` (`proto mismatch`). - `GET /health` returns JSON (`ok`, `name`, `proto`, `viewers`, `bytesServed`, `uptimeSecs`; unauthenticated). `GET /m///` serves the host's file (Range table in ARCHITECTURE.md). + `GET /j///` relays the current Jellyfin media (see + "Jellyfin media" below). ## Client to server ``` -{"t":"hello","proto":6,"secret":"","user":{"name":"","avatar":""}, +{"t":"hello","proto":7,"secret":"","user":{"name":"","avatar":""}, "media":{"name":"","size":0,"duration":0,"url":null},"hostKey":"...", "resume":""} {"t":"ready","ready":true,"source":"manual"} source: manual | auto {"t":"force","target":"all"|"","ready":true} {"t":"seek","position":123.4,"seq":3} {"t":"swap","media":{"name":"","size":0,"duration":0,"url":"..."}} +{"t":"swap","media":{"name":"","size":0,"duration":0,"jellyfin":{}}} {"t":"report","position":123.1,"buffered":6.2,"eventId":41, - "duration":1440.5,"mediaId":2} + "duration":1440.5,"mediaId":2,"jf":{"":true}} {"t":"ping","c":1767225600000} {"t":"chat","text":"..."} {"t":"kick","target":"","reason":""} @@ -58,23 +61,30 @@ Field notes: - `report.duration` / `report.mediaId` (optional, omitted when 0): the open media's length in seconds and the `mediaId` it belongs to. The server takes the first one whose `mediaId` is current while its media duration is 0. +- `report.jf` (optional): Jellyfin access per `ServerId`, `true` = this + client plays the item with its own account, `false` = through the relay. + Sticky: a report without `jf` keeps the last one. At most 8 entries, keys + up to 64 characters; others are dropped. - `ping.c`: the client's clock in epoch ms; answered with `pong`. - `user.avatar`: base64 image, at most 256 KiB of characters. `name` is trimmed, empty becomes `?`, cut to 32 characters. - `chat.text` is cut to 500 characters. - `swap.media.url` is the link; `name`, `size`, `duration` are informational. + `swap.media.jellyfin` makes it Jellyfin media (see below); then `url` is + dropped. - Frames before `welcome` (other than `hello`) are ignored; unparsable or unknown frames are ignored silently. ## Server to client ``` -{"t":"welcome","token":"...","memberId":"m1","mediaUrl":"...","state":{}} +{"t":"welcome","token":"...","memberId":"m1","mediaUrl":"...","relayUrl":"...", + "state":{}} {"t":"state","eventId":7,"transport":"playing"|"held"|"finished","position":12.3, "serverTime":"","mediaId":1, - "media":{name,size,duration,url?}|absent, - "members":[{"id","name","avatar","ready","inSet","buffered","stale"}], - "seekBy":"m1","seekSeq":3} + "media":{name,size,duration,url?,jellyfin?}|absent, + "members":[{"id","name","avatar","ready","inSet","buffered","stale","jf"?}], + "seekBy":"m1","seekSeq":3,"relay":true} {"t":"pong","c":1767225600000,"s":1767225600012} {"t":"feed","actor":"","action":"","target":"","reason":"","at":""} {"t":"chat","text":"...","from":""} @@ -86,7 +96,9 @@ Field notes: media URL. `mediaUrl` is omitted when there is no media; it is a built `/m/...` URL for a host file (host part = what the client dialled), otherwise the shared link. - `welcome.state` has the same shape as `state`. + `welcome.state` has the same shape as `state`. `relayUrl` is present only + when `state.relay` is: this member's `/j/...` URL for the current + Jellyfin media (private like `token`). - `state` is sent every second and on every change. `position` is the value at `serverTime` (server clock); clients extrapolate while `transport` is `playing`, on the server's clock (see `pong`). `finished`: the media @@ -94,7 +106,9 @@ Field notes: `eventId` increases on seek, swap and transport changes. `mediaId` increases on swap. `media.duration` is 0 until a client reports it. `seekBy` (member id) and `seekSeq` are present only while the current - `eventId` came from a seek. + `eventId` came from a seek. `relay` (omitted when false): the server can + relay the current Jellyfin media. `members[].jf` is the member's last + `report.jf`, omitted when it never sent one. - `pong.c` echoes the ping's `c`; `pong.s` is the server's clock in epoch ms (the clock `serverTime` uses). With `t3` the local receive time, the offset is `s - (c + t3) / 2`, accurate to half the round trip. @@ -107,6 +121,37 @@ Field notes: `feed` with `action:"chat"`. Render the `chat` frame only; the reference client drops the textless feed entry. +## Jellyfin media + +``` + = {"serverId":"","itemId":"<32 hex>", + "mediaSourceId":"...","title":"Pilot","subtitle":"Show - S01E01", + "runtimeSecs":1440.5} +``` + +- `serverId`, `itemId` required; `mediaSourceId`, `subtitle`, `runtimeSecs` + optional (omitted when absent); `title` always sent (may be empty). + `runtimeSecs` is informational: the sync clock still uses the duration a + client reports. +- The server keeps only the ref: for Jellyfin media no URL or token is stored + or broadcast (a `url` sent along is dropped). `serverId`, `itemId` and + `mediaSourceId` must match `[A-Za-z0-9-]{1,64}`, otherwise `bad swap media`. + `media.name` is the swap's name, else the ref's `title`; names are + sanitized like link names. +- Each client resolves the ref with its own account on that server. A client + without one plays the relay URL when `state.relay` is true: `welcome.relayUrl`, + or after a swap built like the `/m` one from the host it dialled: + `http:///j/default//`. +- `GET /j///`: the server fetches + `/Videos//stream?static=true&mediaSourceId=` (the original + file) with its own credential and streams it through. `Range` is forwarded + as sent; status, `Content-Range`, `Content-Length`, `Content-Type` and + `Accept-Ranges` come back as Jellyfin sent them. Errors: 401 bad token, + 403 kicked or not the current item, 404 no relay for the current media, + 410 unknown session, 502 Jellyfin unreachable or rejecting the server's + credential. Bytes count toward `bytesServed`. +- Pasted Jellyfin links sent as plain `url` stay plain links. + ## Join and reconnect `hello` then `welcome`, then the app opens `mediaUrl`, seeks to the server diff --git a/packages/core/lib/src/config.dart b/packages/core/lib/src/config.dart index 4fa4c4c..b4f0dd9 100644 --- a/packages/core/lib/src/config.dart +++ b/packages/core/lib/src/config.dart @@ -5,6 +5,91 @@ import 'package:path/path.dart' as p; import 'package:yaml/yaml.dart'; import 'errors.dart'; +import 'jellyfin/client_info.dart'; + +/// A Jellyfin login the server uses to relay the current Jellyfin media to +/// viewers without their own access (`GET /j/...`). Primitives only (it is +/// part of [AppConfig.toMap]). [client], [device], [deviceId] and [version] +/// must be the ones [token] was issued with: Jellyfin overwrites a token's +/// stored device fields with whatever the request header says. +class JellyfinCredential { + /// Server base URL, no trailing slash needed. + final String baseUrl; + /// The server's `ServerId`; only refs with this id are relayed. Empty: + /// the server looks it up at start (`/System/Info/Public`). + final String serverId; + final String token; + final String client; + final String device; + final String deviceId; + final String version; + const JellyfinCredential({ + required this.baseUrl, + this.serverId = '', + required this.token, + required this.client, + required this.device, + required this.deviceId, + required this.version, + }); + + JellyfinClientInfo get clientInfo => + JellyfinClientInfo(client: client, device: device, deviceId: deviceId, version: version); + + JellyfinCredential withServerId(String id) => JellyfinCredential( + baseUrl: baseUrl, + serverId: id, + token: token, + client: client, + device: device, + deviceId: deviceId, + version: version); + + /// Null when [value] is absent; throws [ConfigException] (without the + /// values: the map holds a token) when it is not a usable credential. + static JellyfinCredential? fromMap(Object? value) { + if (value == null) return null; + if (value is! Map) throw const ConfigException('jellyfin must be a map'); + String field(String key, {bool required = true}) { + final v = value[key]; + if (v == null && !required) return ''; + if (v is! String || (required && v.trim().isEmpty)) { + throw ConfigException('jellyfin.$key must be a non-empty string'); + } + return v.trim(); + } + + return JellyfinCredential( + baseUrl: field('baseUrl'), + serverId: field('serverId', required: false), + token: field('token'), + client: field('client'), + device: field('device'), + deviceId: field('deviceId'), + version: field('version'), + ); + } + + Map toMap() => { + 'baseUrl': baseUrl, + 'serverId': serverId, + 'token': token, + 'client': client, + 'device': device, + 'deviceId': deviceId, + 'version': version, + }; + + /// A file (YAML, or JSON by `.json` extension) holding just these keys: + /// the dedicated server's `--jellyfin` flag, keeping the token off the + /// command line. + static Future loadFile(String path) async => + JellyfinCredential.fromMap(await readConfigMapFile(path))!; + + /// Never prints the token. + @override + String toString() => 'JellyfinCredential($baseUrl, serverId: $serverId)'; +} class AppConfig { final int port; @@ -18,6 +103,8 @@ class AppConfig { /// Secret the hosting app's own client sends in hello to be marked host /// (unkickable). Empty means no member is the host (dedicated server). final String hostKey; + /// Optional login for relaying Jellyfin media to guests; null = no relay. + final JellyfinCredential? jellyfin; const AppConfig({ required this.port, @@ -27,6 +114,7 @@ class AppConfig { this.logLevel = 'INFO', this.advertiseHost = '', this.hostKey = '', + this.jellyfin, }); factory AppConfig.defaults() => @@ -42,7 +130,10 @@ class AppConfig { logLevel: (map['logLevel'] as String?) ?? 'INFO', advertiseHost: (map['advertiseHost'] as String?) ?? '', hostKey: (map['hostKey'] as String?) ?? '', + jellyfin: JellyfinCredential.fromMap(map['jellyfin']), ); + } on ConfigException { + rethrow; } catch (e) { throw ConfigException('invalid config map: $e'); } @@ -56,29 +147,11 @@ class AppConfig { 'logLevel': logLevel, 'advertiseHost': advertiseHost, 'hostKey': hostKey, + if (jellyfin != null) 'jellyfin': jellyfin!.toMap(), }; - static Future loadFile(String path) async { - final file = File(path); - if (!await file.exists()) { - throw ConfigException('config file not found: $path'); - } - final text = await file.readAsString(); - try { - if (path.endsWith('.json')) { - final decoded = jsonDecode(text); - if (decoded is! Map) throw const ConfigException('root must be a map'); - return AppConfig.fromMap(Map.from(decoded)); - } - final yamlMap = loadYaml(text); - if (yamlMap is! YamlMap) throw const ConfigException('root must be a map'); - return AppConfig.fromMap( - Map.from(yamlMap.map((k, v) => MapEntry('$k', v)))); - } catch (e) { - if (e is ConfigException) rethrow; - throw ConfigException('failed to parse $path: $e'); - } - } + static Future loadFile(String path) async => + AppConfig.fromMap(await readConfigMapFile(path)); static String defaultPath(String appName) { final home = @@ -86,3 +159,24 @@ class AppConfig { return p.join(home, '.config', appName, 'config.yaml'); } } + +/// Reads a YAML or JSON (by `.json` extension) file whose root is a map. +/// Parse errors are reported without the source excerpt: the file may hold +/// a token. +Future> readConfigMapFile(String path) async { + final file = File(path); + if (!await file.exists()) { + throw ConfigException('config file not found: $path'); + } + final text = await file.readAsString(); + Object? root; + try { + root = path.endsWith('.json') ? jsonDecode(text) : loadYaml(text); + } on YamlException catch (e) { + throw ConfigException('failed to parse $path: ${e.message} (line ${(e.span?.start.line ?? -1) + 1})'); + } on FormatException catch (e) { + throw ConfigException('failed to parse $path: ${e.message}'); + } + if (root is! Map) throw const ConfigException('root must be a map'); + return {for (final e in root.entries) '${e.key}': e.value}; +} diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index 71e6102..0a920f4 100644 --- a/packages/core/lib/src/constants.dart +++ b/packages/core/lib/src/constants.dart @@ -1,5 +1,5 @@ abstract class RelayConstants { - static const int protoVersion = 6; + static const int protoVersion = 7; static const int chunkBytes = 256 * 1024; static const int creditWindowChunks = 8; diff --git a/packages/core/lib/src/protocol.dart b/packages/core/lib/src/protocol.dart index 08b06a3..7e1c1b2 100644 --- a/packages/core/lib/src/protocol.dart +++ b/packages/core/lib/src/protocol.dart @@ -45,21 +45,96 @@ class HelloUser { HelloUser(name: (json['name'] as String?) ?? '', avatar: (json['avatar'] as String?) ?? ''); } +/// A Jellyfin item the session plays, by reference: each viewer resolves it +/// with its own account (or through the server's relay). Never carries a URL +/// or a token. +class JellyfinRef { + /// Jellyfin `ServerId` (`/System/Info/Public`), the key accounts are stored by. + final String serverId; + final String itemId; + final String? mediaSourceId; + /// Display text: item name, and e.g. "Show - S01E02" for episodes. + final String title; + final String? subtitle; + /// Length from Jellyfin's RunTimeTicks, informational; the sync clock + /// still uses the length a client reports. + final double? runtimeSecs; + const JellyfinRef({ + required this.serverId, + required this.itemId, + this.mediaSourceId, + this.title = '', + this.subtitle, + this.runtimeSecs, + }); + + Map toJson() => { + 'serverId': serverId, + 'itemId': itemId, + if (mediaSourceId != null) 'mediaSourceId': mediaSourceId, + 'title': title, + if (subtitle != null) 'subtitle': subtitle, + if (runtimeSecs != null) 'runtimeSecs': runtimeSecs, + }; + + factory JellyfinRef.fromJson(Map json) { + final runtime = json['runtimeSecs']; + return JellyfinRef( + serverId: (json['serverId'] as String?) ?? '', + itemId: (json['itemId'] as String?) ?? '', + mediaSourceId: json['mediaSourceId'] as String?, + title: (json['title'] as String?) ?? '', + subtitle: json['subtitle'] as String?, + runtimeSecs: runtime == null ? null : finiteDouble(runtime, 'runtimeSecs'), + ); + } + + @override + bool operator ==(Object other) => + other is JellyfinRef && + other.serverId == serverId && + other.itemId == itemId && + other.mediaSourceId == mediaSourceId && + other.title == title && + other.subtitle == subtitle && + other.runtimeSecs == runtimeSecs; + + @override + int get hashCode => + Object.hash(serverId, itemId, mediaSourceId, title, subtitle, runtimeSecs); +} + class HelloMedia { final String name; final int size; final double duration; final String? url; - const HelloMedia({required this.name, required this.size, required this.duration, this.url}); + /// Set for Jellyfin media; then there is no [url] (the server drops one). + final JellyfinRef? jellyfin; + const HelloMedia( + {required this.name, required this.size, required this.duration, this.url, this.jellyfin}); - Map toJson() => {'name': name, 'size': size, 'duration': duration, if (url != null) 'url': url}; + HelloMedia copyWith({double? duration}) => HelloMedia( + name: name, size: size, duration: duration ?? this.duration, url: url, jellyfin: jellyfin); - factory HelloMedia.fromJson(Map json) => HelloMedia( - name: (json['name'] as String?) ?? '', - size: finiteInt(json['size'], 'size', fallback: 0), - duration: finiteDouble(json['duration'], 'duration'), - url: json['url'] as String?, - ); + Map toJson() => { + 'name': name, + 'size': size, + 'duration': duration, + if (url != null) 'url': url, + if (jellyfin != null) 'jellyfin': jellyfin!.toJson(), + }; + + factory HelloMedia.fromJson(Map json) { + final jf = json['jellyfin']; + return HelloMedia( + name: (json['name'] as String?) ?? '', + size: finiteInt(json['size'], 'size', fallback: 0), + duration: finiteDouble(json['duration'], 'duration'), + url: json['url'] as String?, + jellyfin: jf is Map ? JellyfinRef.fromJson(Map.from(jf)) : null, + ); + } } class HelloMessage { @@ -134,14 +209,19 @@ class WelcomeMessage { final String token; final String memberId; final String? mediaUrl; + /// This member's `/j/...` URL for the current Jellyfin media when the + /// server can relay it (`state.relay`); private like [token]. + final String? relayUrl; final Map? state; - const WelcomeMessage({required this.token, this.memberId = '', this.mediaUrl, this.state}); + const WelcomeMessage( + {required this.token, this.memberId = '', this.mediaUrl, this.relayUrl, this.state}); Map toJson() => { 't': 'welcome', 'token': token, 'memberId': memberId, if (mediaUrl != null) 'mediaUrl': mediaUrl, + if (relayUrl != null) 'relayUrl': relayUrl, if (state != null) 'state': state, }; @@ -152,6 +232,7 @@ class WelcomeMessage { token: (json['token'] as String?) ?? '', memberId: (json['memberId'] as String?) ?? '', mediaUrl: json['mediaUrl'] as String?, + relayUrl: json['relayUrl'] as String?, state: state is Map ? Map.from(state) : null, ); } @@ -257,12 +338,16 @@ class ReportMessage { final double duration; /// The `mediaId` the client has open; [duration] belongs to it. final int mediaId; + /// Jellyfin access by `ServerId`: true = plays with its own account, + /// false = through the server's relay. Null leaves the last one in place. + final Map? jf; const ReportMessage( {required this.position, required this.buffered, required this.eventId, this.duration = 0, - this.mediaId = 0}); + this.mediaId = 0, + this.jf}); Map toJson() => { 't': 'report', @@ -271,6 +356,7 @@ class ReportMessage { 'eventId': eventId, if (duration > 0) 'duration': duration, if (mediaId != 0) 'mediaId': mediaId, + if (jf != null) 'jf': jf, }; factory ReportMessage.fromJson(Map json) { @@ -281,10 +367,25 @@ class ReportMessage { eventId: finiteInt(json['eventId'], 'eventId', fallback: 0), duration: finiteDouble(json['duration'], 'duration'), mediaId: finiteInt(json['mediaId'], 'mediaId'), + jf: jfAccessFromJson(json['jf']), ); } } +/// Reads a `jf` map (`{serverId: bool}`), dropping non-bool entries and +/// keys over 64 characters, at most 8 entries (it rides in every `state`); +/// null when absent or not a map. +Map? jfAccessFromJson(Object? value) { + if (value is! Map) return null; + final out = {}; + for (final e in value.entries) { + if (out.length >= 8) break; + final k = e.key; + if (k is String && k.length <= 64 && e.value is bool) out[k] = e.value as bool; + } + return out; +} + /// Clock probe: the client sends its own clock in [clientMs]; the server /// answers with a [PongMessage] carrying it back plus its clock. class PingMessage { @@ -333,6 +434,8 @@ class MemberView { final bool inSet; final double buffered; final bool stale; + /// Last `report.jf` of this member; empty when never reported. + final Map jf; const MemberView({ required this.id, required this.name, @@ -341,6 +444,7 @@ class MemberView { required this.inSet, required this.buffered, required this.stale, + this.jf = const {}, }); Map toJson() => { @@ -351,6 +455,7 @@ class MemberView { 'inSet': inSet, 'buffered': buffered, 'stale': stale, + if (jf.isNotEmpty) 'jf': jf, }; factory MemberView.fromJson(Map json) => MemberView( @@ -361,6 +466,7 @@ class MemberView { inSet: (json['inSet'] as bool?) ?? false, buffered: finiteDouble(json['buffered'], 'buffered'), stale: (json['stale'] as bool?) ?? false, + jf: jfAccessFromJson(json['jf']) ?? const {}, ); } @@ -376,6 +482,9 @@ class StateBroadcast { /// current event is not a seek. final String? seekBy; final int? seekSeq; + /// The server can relay the current Jellyfin media (`/j/...`) for viewers + /// without their own access to it. + final bool relay; const StateBroadcast({ required this.eventId, required this.transport, @@ -386,6 +495,7 @@ class StateBroadcast { required this.members, this.seekBy, this.seekSeq, + this.relay = false, }); Map toJson() => { @@ -399,6 +509,7 @@ class StateBroadcast { 'members': members.map((m) => m.toJson()).toList(), if (seekBy != null) 'seekBy': seekBy, if (seekSeq != null) 'seekSeq': seekSeq, + if (relay) 'relay': true, }; factory StateBroadcast.fromJson(Map json) { @@ -420,6 +531,7 @@ class StateBroadcast { : const [], seekBy: json['seekBy'] as String?, seekSeq: json['seekSeq'] == null ? null : finiteInt(json['seekSeq'], 'seekSeq'), + relay: (json['relay'] as bool?) ?? false, ); } } diff --git a/packages/core/lib/src/sync_engine.dart b/packages/core/lib/src/sync_engine.dart index 1bda9de..c912f56 100644 --- a/packages/core/lib/src/sync_engine.dart +++ b/packages/core/lib/src/sync_engine.dart @@ -22,6 +22,8 @@ class EngineMember { bool ready = false; bool inSet = false; double buffered = 0; + /// Last reported Jellyfin access by ServerId (`report.jf`). + Map jf = const {}; DateTime lastSeen; EngineMember({ required this.id, @@ -39,6 +41,7 @@ class EngineMember { inSet: inSet, buffered: buffered, stale: stale, + jf: jf, ); } @@ -234,17 +237,18 @@ class SyncEngine { } void applyReport(String token, double reportPosition, double buffered, int reportEventId, - {double duration = 0, int mediaId = 0}) { + {double duration = 0, int mediaId = 0, Map? jf}) { final member = members[token]; if (member == null) return; if (!reportPosition.isFinite) return; _seen(member); + if (jf != null) member.jf = Map.unmodifiable(jf); member.buffered = !buffered.isFinite || buffered < 0 ? 0 : buffered; final m = media; // First report of the length wins; only for the media the client has open. if (m != null && m.duration <= 0 && mediaId == this.mediaId && duration.isFinite && duration > 0) { - media = HelloMedia(name: m.name, size: m.size, duration: duration, url: m.url); + media = m.copyWith(duration: duration); } if (reportEventId < eventId) return; evaluate(); @@ -317,7 +321,9 @@ class SyncEngine { } } - StateBroadcast broadcast() { + /// [relay]: whether the server can relay the current Jellyfin media; the + /// engine does not know the server's credential. + StateBroadcast broadcast({bool relay = false}) { final bySeek = _seekEventId == eventId; return StateBroadcast( eventId: eventId, @@ -329,6 +335,7 @@ class SyncEngine { members: members.values.map((m) => m.view(stale: isStale(m))).toList(), seekBy: bySeek ? _seekBy : null, seekSeq: bySeek ? _seekSeq : null, + relay: relay, ); } } diff --git a/packages/core/test/jellyfin_ref_test.dart b/packages/core/test/jellyfin_ref_test.dart new file mode 100644 index 0000000..7d63875 --- /dev/null +++ b/packages/core/test/jellyfin_ref_test.dart @@ -0,0 +1,220 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:sync_relay_core/core.dart'; +import 'package:test/test.dart'; + +Map wire(Map json) => + Map.from(jsonDecode(jsonEncode(json)) as Map); + +void main() { + const ref = JellyfinRef( + serverId: 'abc123', + itemId: '0123456789abcdef0123456789abcdef', + mediaSourceId: 'fedcba9876543210fedcba9876543210', + title: 'Pilot', + subtitle: 'Show - S01E01', + runtimeSecs: 1440.5, + ); + + group('protocol', () { + test('swap with a Jellyfin ref round-trips, no url', () { + const swap = SwapIntent(media: HelloMedia(name: 'Pilot', size: 0, duration: 0, jellyfin: ref)); + final json = wire(swap.toJson()); + expect(json['media'], { + 'name': 'Pilot', + 'size': 0, + 'duration': 0.0, + 'jellyfin': { + 'serverId': 'abc123', + 'itemId': '0123456789abcdef0123456789abcdef', + 'mediaSourceId': 'fedcba9876543210fedcba9876543210', + 'title': 'Pilot', + 'subtitle': 'Show - S01E01', + 'runtimeSecs': 1440.5, + }, + }); + final back = SwapIntent.fromJson(json).media; + expect(back.jellyfin, ref); + expect(back.url, isNull); + }); + + test('minimal ref omits optional fields; url media has no jellyfin', () { + const minimal = JellyfinRef(serverId: 's', itemId: 'i'); + expect(minimal.toJson(), {'serverId': 's', 'itemId': 'i', 'title': ''}); + expect(JellyfinRef.fromJson(wire(minimal.toJson())), minimal); + final url = HelloMedia.fromJson(wire(const HelloMedia(name: 'a', size: 1, duration: 2, url: 'http://x/a').toJson())); + expect(url.jellyfin, isNull); + expect(url.toJson().containsKey('jellyfin'), isFalse); + }); + + test('non-finite runtimeSecs is rejected', () { + expect(() => JellyfinRef.fromJson({'serverId': 's', 'itemId': 'i', 'runtimeSecs': double.infinity}), + throwsA(isA())); + }); + + test('state relay flag and member jf round-trip, absent by default', () { + final state = StateBroadcast( + eventId: 1, + transport: 'held', + position: 0, + serverTime: DateTime.utc(2026), + mediaId: 2, + media: const HelloMedia(name: 'Pilot', size: 0, duration: 0, jellyfin: ref).toJson(), + members: const [ + MemberView(id: 'm1', name: 'a', ready: false, inSet: false, buffered: 0, stale: false, jf: {'abc123': false}), + MemberView(id: 'm2', name: 'b', ready: false, inSet: false, buffered: 0, stale: false), + ], + relay: true, + ); + final json = wire(state.toJson()); + expect(json['relay'], true); + expect((json['members'] as List)[0], containsPair('jf', {'abc123': false})); + expect((json['members'] as List)[1], isNot(contains('jf'))); + final back = StateBroadcast.fromJson(json); + expect(back.relay, isTrue); + expect(back.members[0].jf, {'abc123': false}); + expect(back.members[1].jf, isEmpty); + expect(HelloMedia.fromJson(back.media!).jellyfin, ref); + + final plain = StateBroadcast( + eventId: 1, transport: 'held', position: 0, serverTime: DateTime.utc(2026), mediaId: 1, members: const []); + expect(plain.toJson().containsKey('relay'), isFalse); + expect(StateBroadcast.fromJson(wire(plain.toJson())).relay, isFalse); + }); + + test('welcome relayUrl round-trips, omitted when null', () { + const w = WelcomeMessage(token: 't', memberId: 'm1', relayUrl: 'http://h:1/j/default/t/i'); + expect(WelcomeMessage.fromJson(wire(w.toJson())).relayUrl, 'http://h:1/j/default/t/i'); + expect(const WelcomeMessage(token: 't').toJson().containsKey('relayUrl'), isFalse); + }); + + test('report jf: round-trip, absent = null, junk dropped and capped', () { + const r = ReportMessage(position: 1, buffered: 2, eventId: 3, jf: {'abc123': true}); + expect(wire(r.toJson())['jf'], {'abc123': true}); + expect(ReportMessage.fromJson(wire(r.toJson())).jf, {'abc123': true}); + expect(ReportMessage.fromJson(wire(const ReportMessage(position: 1, buffered: 2, eventId: 3).toJson())).jf, + isNull); + final junk = ReportMessage.fromJson({ + 't': 'report', + 'position': 0, + 'buffered': 0, + 'eventId': 0, + 'jf': { + 'ok': true, + 'bad': 'yes', + 'x' * 65: true, + for (var i = 0; i < 20; i++) 's$i': false, + }, + }).jf!; + expect(junk['ok'], isTrue); + expect(junk.containsKey('bad'), isFalse); + expect(junk.length, 8); + }); + }); + + group('engine', () { + late DateTime t; + late SyncEngine engine; + setUp(() { + t = DateTime.utc(2026); + engine = SyncEngine(clock: () => t); + engine.applyJoin('a', 'alice', ''); + }); + + test('swap with a ref broadcasts the ref, and a learned duration keeps it', () { + expect(engine.applySwap('a', const HelloMedia(name: 'Pilot', size: 0, duration: 0, jellyfin: ref)), isTrue); + expect(engine.mediaId, 2); + engine.applyReport('a', 0, 0, engine.eventId, duration: 1441, mediaId: 2); + final media = HelloMedia.fromJson(engine.broadcast().media!); + expect(media.duration, 1441); + expect(media.jellyfin, ref); + expect(media.url, isNull); + }); + + test('broadcast carries relay only when asked', () { + expect(engine.broadcast().relay, isFalse); + expect(engine.broadcast(relay: true).relay, isTrue); + }); + + test('report jf is sticky per member and shows in the member view', () { + engine.applyReport('a', 0, 0, engine.eventId, jf: {'abc123': true}); + engine.applyReport('a', 0, 0, engine.eventId); + expect(engine.broadcast().members.single.jf, {'abc123': true}); + engine.applyReport('a', 0, 0, engine.eventId, jf: {'abc123': false}); + expect(engine.broadcast().members.single.jf, {'abc123': false}); + }); + }); + + group('config', () { + const cred = JellyfinCredential( + baseUrl: 'http://jf.local:8096', + serverId: 'abc123', + token: 'SECRET-TOKEN', + client: 'Watch Together', + device: 'host-pc', + deviceId: 'dev-1', + version: '1.2.3', + ); + + test('AppConfig map round-trips the credential as primitives', () { + final map = const AppConfig(port: 1, secret: '', name: 'n', jellyfin: cred).toMap(); + expect(map['jellyfin'], { + 'baseUrl': 'http://jf.local:8096', + 'serverId': 'abc123', + 'token': 'SECRET-TOKEN', + 'client': 'Watch Together', + 'device': 'host-pc', + 'deviceId': 'dev-1', + 'version': '1.2.3', + }); + final back = AppConfig.fromMap(wire(map)).jellyfin!; + expect(back.toMap(), cred.toMap()); + expect(const AppConfig(port: 1, secret: '', name: 'n').toMap().containsKey('jellyfin'), isFalse); + expect(AppConfig.fromMap(const {}).jellyfin, isNull); + }); + + test('serverId optional; missing fields fail without echoing values', () { + final m = cred.toMap()..remove('serverId'); + expect(JellyfinCredential.fromMap(m)!.serverId, ''); + final bad = cred.toMap()..remove('device'); + expect( + () => AppConfig.fromMap({'jellyfin': bad}), + throwsA(isA() + .having((e) => e.message, 'message', allOf(contains('jellyfin.device'), isNot(contains('SECRET')))))); + }); + + test('toString never prints the token', () { + expect('$cred', isNot(contains('SECRET'))); + }); + + test('clientInfo sends exactly the configured fields', () { + expect(cred.clientInfo.authorizationHeader(cred.token), + 'MediaBrowser Client="Watch%20Together", Device="host-pc", DeviceId="dev-1", Version="1.2.3", Token="SECRET-TOKEN"'); + }); + + test('credential file (YAML) loads; parse errors do not echo the file', () async { + final dir = await Directory.systemTemp.createTemp('jfcred'); + try { + final ok = File('${dir.path}/jf.yaml'); + await ok.writeAsString('baseUrl: http://jf.local:8096\ntoken: SECRET-TOKEN\nclient: Watch Together\n' + 'device: host-pc\ndeviceId: dev-1\nversion: 1.2.3\n'); + final loaded = await JellyfinCredential.loadFile(ok.path); + expect(loaded.token, 'SECRET-TOKEN'); + expect(loaded.serverId, ''); + final cfg = File('${dir.path}/cfg.yaml'); + await cfg.writeAsString('port: 9\njellyfin:\n baseUrl: http://x\n token: SECRET-TOKEN\n' + ' client: c\n device: d\n deviceId: i\n version: v\n'); + expect((await AppConfig.loadFile(cfg.path)).jellyfin!.device, 'd'); + final broken = File('${dir.path}/bad.yaml'); + await broken.writeAsString('token: SECRET-TOKEN\n : : [\n'); + await expectLater( + JellyfinCredential.loadFile(broken.path), + throwsA(isA() + .having((e) => e.message, 'message', isNot(contains('SECRET'))))); + } finally { + await dir.delete(recursive: true); + } + }); + }); +} diff --git a/packages/server/bin/server.dart b/packages/server/bin/server.dart index 6fd8b86..ef19c7f 100644 --- a/packages/server/bin/server.dart +++ b/packages/server/bin/server.dart @@ -16,6 +16,9 @@ Future main(List args) async { ..addOption('name', defaultsTo: 'server', help: 'Server name, reported by /health.') ..addOption('media', defaultsTo: '', help: 'Local file to serve as the starting media.') ..addOption('advertise-host', defaultsTo: '', help: 'Host put in file-media URLs instead of the one each client dialled.') + ..addOption('jellyfin', + help: 'YAML or JSON file with a Jellyfin login (baseUrl, token, client, device, deviceId, version[, serverId]) ' + 'to relay Jellyfin media to viewers without an account. The device fields must be the ones the token was issued with.') ..addOption('config', help: 'YAML or JSON config file (keys as in AppConfig); overrides the other flags.') ..addFlag('help', abbr: 'h', negatable: false, help: 'Show this help.'); final ArgResults parsed; @@ -51,6 +54,20 @@ Future main(List args) async { advertiseHost: '${parsed['advertise-host']}', ); } + final jellyfinPath = parsed['jellyfin'] as String?; + if (jellyfinPath != null) { + final c = config; + config = AppConfig( + port: c.port, + secret: c.secret, + name: c.name, + mediaPath: c.mediaPath, + logLevel: c.logLevel, + advertiseHost: c.advertiseHost, + hostKey: c.hostKey, + jellyfin: await JellyfinCredential.loadFile(jellyfinPath), + ); + } } on ConfigException catch (e) { stderr.writeln(e.message); exitCode = 64; diff --git a/packages/server/lib/server.dart b/packages/server/lib/server.dart index 92640cf..ddefd60 100644 --- a/packages/server/lib/server.dart +++ b/packages/server/lib/server.dart @@ -3,3 +3,4 @@ export 'src/serve.dart'; export 'src/server_hub.dart'; export 'src/session.dart'; export 'src/sync_server.dart'; +export 'src/jellyfin_relay.dart'; diff --git a/packages/server/lib/src/jellyfin_relay.dart b/packages/server/lib/src/jellyfin_relay.dart new file mode 100644 index 0000000..24b1d23 --- /dev/null +++ b/packages/server/lib/src/jellyfin_relay.dart @@ -0,0 +1,160 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:relic/relic.dart'; +import 'package:sync_relay_core/core.dart'; + +import 'media_endpoint.dart'; +import 'session.dart'; + +final _relayLog = createLogger('jellyfin_relay'); + +/// Upstream request for one relayed range: the original file +/// (`/Videos/{id}/stream?static=true`) with the server's credential in the +/// `Authorization` header (never in the URL, so nothing to leak in logs). +Uri jellyfinStaticStreamUri(JellyfinCredential cred, JellyfinRef ref) { + final base = normalizeServerUrl(cred.baseUrl) ?? cred.baseUrl; + return Uri.parse('$base/Videos/${Uri.encodeComponent(ref.itemId)}/stream').replace( + queryParameters: {'static': 'true', 'mediaSourceId': ref.mediaSourceId ?? ref.itemId}); +} + +bool _sameItemId(String a, String b) => (normalizeItemId(a) ?? a) == (normalizeItemId(b) ?? b); + +Response _plain(int status) => Response(status, body: Body.fromString('status $status')); + +/// `GET /j///`: relays byte ranges of the current +/// Jellyfin media from [credential]'s server to a viewer, for guests without +/// their own access. Only the active ref's item on the credential's server, +/// only for a valid viewer token: no open proxy. Range goes up as sent; +/// status, Content-Range, Content-Length, Content-Type and Accept-Ranges +/// come back as Jellyfin sent them (a 401/403 from Jellyfin is the server's +/// credential failing, so the viewer gets 502). Bytes stream through without +/// buffering, are counted as sent, and a viewer disconnect aborts upstream. +/// +/// [client] gives the HTTP client for upstream requests (the caller owns +/// and closes it). +void mountJellyfinRelayRoute( + RelicApp app, + SessionMap sessions, + JellyfinCredential? Function() credential, + HttpClient Function() client, { + String pattern = '/j/:sid/:token/:itemId', + void Function(String token, MediaLog log)? onLog, + Duration timeout = const Duration(seconds: 20), +}) { + app.get(pattern, (final req) async { + final stopwatch = Stopwatch()..start(); + final sid = req.pathParameters.raw[#sid] ?? ''; + final token = req.pathParameters.raw[#token] ?? ''; + final itemId = req.pathParameters.raw[#itemId] ?? ''; + final session = sessions.get(sid); + if (session == null) return _plain(410); + switch (session.gate(token)) { + case MediaGate.ok: + break; + case MediaGate.forbidden: + return _plain(403); + default: + return _plain(401); + } + final ref = session.media?.jellyfin; + final cred = credential(); + if (ref == null || cred == null || cred.serverId.isEmpty || cred.serverId != ref.serverId) { + return _plain(404); + } + if (!_sameItemId(itemId, ref.itemId)) return _plain(403); + + final range = req.headers['range']?.join(','); + final HttpClientRequest upReq; + final HttpClientResponse upRes; + try { + upReq = await client().getUrl(jellyfinStaticStreamUri(cred, ref)).timeout(timeout); + upReq.followRedirects = false; + upReq.headers + ..set(HttpHeaders.authorizationHeader, cred.clientInfo.authorizationHeader(cred.token)) + ..set(HttpHeaders.acceptEncodingHeader, 'identity'); + if (range != null && range.isNotEmpty) upReq.headers.set(HttpHeaders.rangeHeader, range); + upRes = await upReq.close().timeout(timeout); + } catch (e) { + _relayLog.warning('upstream ${ref.itemId}: $e'); + return _plain(502); + } + + final shortToken = token.length > 4 ? '…${token.substring(token.length - 4)}' : token; + void log(int status, int bytes) => onLog?.call( + token, + MediaLog( + method: req.method.name, + path: '/j/$sid/$shortToken/$itemId', + rangeHeader: range, + status: status, + bytesWritten: bytes, + durationMs: stopwatch.elapsedMilliseconds, + )); + + if (upRes.statusCode == 401 || upRes.statusCode == 403) { + _relayLog.warning('upstream ${ref.itemId}: ${upRes.statusCode} (server credential rejected)'); + await upRes.listen(null).cancel(); + log(502, 0); + return _plain(502); + } + + final headers = Headers.build((h) { + for (final name in const ['content-range', 'accept-ranges']) { + final v = upRes.headers.value(name); + if (v != null) h[name] = [v]; + } + }); + MimeType mime; + try { + mime = MimeType.parse(upRes.headers.value(HttpHeaders.contentTypeHeader) ?? ''); + } catch (_) { + mime = MimeType.octetStream; + } + var sent = 0; + late final StreamController body; + StreamSubscription>? sub; + body = StreamController( + onListen: () { + sub = upRes.listen( + (chunk) { + session.addBytes(token, chunk.length); + sent += chunk.length; + body.add(chunk is Uint8List ? chunk : Uint8List.fromList(chunk)); + }, + onError: (Object e) { + _relayLog.warning('upstream ${ref.itemId} broke: $e'); + sub = null; + log(upRes.statusCode, sent); + body.addError(e); + unawaited(body.close()); + }, + onDone: () { + sub = null; + log(upRes.statusCode, sent); + unawaited(body.close()); + }, + cancelOnError: true, + ); + }, + onPause: () => sub?.pause(), + onResume: () => sub?.resume(), + onCancel: () async { + // Viewer went away (mpv drops `bytes=N-` on every seek): cancelling + // an unfinished HttpClientResponse closes its connection, so the + // upstream transfer stops instead of draining the rest of the file. + final s = sub; + sub = null; + await s?.cancel(); + if (s != null) log(upRes.statusCode, sent); + }, + ); + return Response( + upRes.statusCode, + body: Body.fromDataStream(body.stream, + mimeType: mime, contentLength: upRes.contentLength >= 0 ? upRes.contentLength : null), + headers: headers, + ); + }); +} diff --git a/packages/server/lib/src/media_endpoint.dart b/packages/server/lib/src/media_endpoint.dart index 68bb57a..b29a861 100644 --- a/packages/server/lib/src/media_endpoint.dart +++ b/packages/server/lib/src/media_endpoint.dart @@ -136,7 +136,9 @@ void mountMediaRoute(RelicApp app, File file, log(byteCount); return Response( res.status, - body: Body.fromDataStream(stream, contentLength: byteCount), + // Relic replaces Content-Type with the body's mime type, so set it here. + body: Body.fromDataStream(stream, + mimeType: _mimeOf(res.headers['content-type']), contentLength: byteCount), headers: Headers.build((h) { res.headers.forEach((k, v) => h[k] = [v]); }), @@ -208,10 +210,20 @@ void mountSessionMediaRoute( }); return Response( res.status, - body: Body.fromDataStream(stream, contentLength: byteCount), + // Relic replaces Content-Type with the body's mime type, so set it here. + body: Body.fromDataStream(stream, + mimeType: _mimeOf(res.headers['content-type']), contentLength: byteCount), headers: Headers.build((h) { res.headers.forEach((k, v) => h[k] = [v]); }), ); }); } + +MimeType _mimeOf(String? contentType) { + try { + return MimeType.parse(contentType ?? ''); + } catch (_) { + return MimeType.octetStream; + } +} diff --git a/packages/server/lib/src/server_hub.dart b/packages/server/lib/src/server_hub.dart index 34233e7..ed8f300 100644 --- a/packages/server/lib/src/server_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -16,11 +16,22 @@ class ServerHub { final String hostKey; final SyncEngine engine; final Map sockets = {}; + /// The server's Jellyfin login for the `/j/...` relay; null = no relay. + JellyfinCredential? jellyfin; ServerHub(this.sessions, this.expectedSecret, - {SyncEngine? engine, this.hostKey = ''}) + {SyncEngine? engine, this.hostKey = '', this.jellyfin}) : engine = engine ?? SyncEngine(); + /// The current media is a Jellyfin ref on the server [jellyfin] logs in to. + bool get relayAvailable { + final ref = session.media?.jellyfin; + final cred = jellyfin; + return ref != null && cred != null && cred.serverId.isNotEmpty && cred.serverId == ref.serverId; + } + + StateBroadcast stateBroadcast() => engine.broadcast(relay: relayAvailable); + bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey; /// The member a reconnecting client may take over with its old token: @@ -35,7 +46,7 @@ class ServerHub { Session get session => sessions.getOrCreate('default', expectedSecret); void broadcastState() { - final raw = jsonEncode(engine.broadcast().toJson()); + final raw = jsonEncode(stateBroadcast().toJson()); for (final ws in sockets.values.toList()) { ws.trySendText(raw); } @@ -139,12 +150,34 @@ void wsErr(RelicWebSocket ws, String message) { ws.trySendText(jsonEncode({'t': 'err', 'message': message})); } +/// Jellyfin ids are path segments of upstream requests: keep them plain. +final _jellyfinId = RegExp(r'^[A-Za-z0-9-]{1,64}$'); + +/// The swap's Jellyfin ref cleaned for broadcast (names like +/// [sanitizeMediaName]), or null when its ids are unusable. +JellyfinRef? sanitizeJellyfinRef(JellyfinRef ref) { + if (!_jellyfinId.hasMatch(ref.serverId) || !_jellyfinId.hasMatch(ref.itemId)) return null; + final source = ref.mediaSourceId; + if (source != null && !_jellyfinId.hasMatch(source)) return null; + final runtime = ref.runtimeSecs; + return JellyfinRef( + serverId: ref.serverId, + itemId: ref.itemId, + mediaSourceId: source, + title: sanitizeMediaName(ref.title), + subtitle: ref.subtitle == null ? null : sanitizeMediaName(ref.subtitle!), + runtimeSecs: runtime == null || runtime < 0 ? null : runtime, + ); +} + +/// [relayUrlBuilder] builds a member's `/j/...` URL for `welcome.relayUrl`. void mountSyncRoute( RelicApp app, ServerHub hub, String Function(String sid, String token, SessionMedia? media, String host) - urlBuilder, -) { + urlBuilder, { + String Function(String sid, String token, String itemId, String host)? relayUrlBuilder, +}) { app.get('/sync', (final req) { // Host the client dialled (Host header), so file URLs work for remote joiners. final dialledHost = req.url.hasPort @@ -196,6 +229,30 @@ void mountSyncRoute( case 'swap': if (token == null) return; final parsed = SwapIntent.fromJson(decoded).media; + if (parsed.jellyfin != null) { + // By reference only: no URL (it could carry a token) is + // stored or broadcast for Jellyfin media. + final ref = sanitizeJellyfinRef(parsed.jellyfin!); + final media = ref == null + ? null + : HelloMedia( + name: sanitizeMediaName(parsed.name.isEmpty ? ref.title : parsed.name), + size: parsed.size, + duration: parsed.duration, + jellyfin: ref); + if (media == null || !hub.engine.applySwap(token, media)) { + wsErr(ws, 'bad swap media'); + return; + } + session.media = SessionMedia( + name: media.name, + size: media.size, + duration: media.duration, + jellyfin: ref, + mediaId: hub.engine.mediaId); + hub.changed(); + return; + } final intent = SwapIntent( media: HelloMedia( name: sanitizeMediaName(parsed.name), @@ -236,7 +293,7 @@ void mountSyncRoute( final report = ReportMessage.fromJson(decoded); hub.engine.applyReport( token, report.position, report.buffered, report.eventId, - duration: report.duration, mediaId: report.mediaId); + duration: report.duration, mediaId: report.mediaId, jf: report.jf); hub.changed(); case 'ping': final ping = PingMessage.fromJson(decoded); @@ -355,11 +412,15 @@ void mountSyncRoute( hub.forgetSocketOf(activeToken, ws); }; final media = session.media; + final relay = hub.relayAvailable; final welcome = WelcomeMessage( token: activeToken, memberId: memberId, mediaUrl: media?.file != null ? urlBuilder('default', activeToken, media, dialledHost) : media?.url, - state: hub.engine.broadcast().toJson(), + relayUrl: relay && relayUrlBuilder != null + ? relayUrlBuilder('default', activeToken, media!.jellyfin!.itemId, dialledHost) + : null, + state: hub.stateBroadcast().toJson(), ); try { ws.sendText(jsonEncode(welcome.toJson())); diff --git a/packages/server/lib/src/session.dart b/packages/server/lib/src/session.dart index 92cc668..b12f236 100644 --- a/packages/server/lib/src/session.dart +++ b/packages/server/lib/src/session.dart @@ -30,6 +30,8 @@ class SessionMedia { final double duration; final String? url; final File? file; + /// Jellyfin media by reference; never together with [url] or [file]. + final JellyfinRef? jellyfin; final int mediaId; const SessionMedia({ required this.name, @@ -37,6 +39,7 @@ class SessionMedia { required this.duration, this.url, this.file, + this.jellyfin, this.mediaId = 1, }); } diff --git a/packages/server/lib/src/sync_server.dart b/packages/server/lib/src/sync_server.dart index 833a2c1..de29207 100644 --- a/packages/server/lib/src/sync_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -6,6 +6,7 @@ import 'package:logging/logging.dart' as pkg_logging; import 'package:relic/relic.dart'; import 'package:sync_relay_core/core.dart'; +import 'jellyfin_relay.dart'; import 'media_endpoint.dart'; import 'server_hub.dart'; import 'session.dart'; @@ -20,9 +21,11 @@ class SyncServer { DateTime? _startedAt; Timer? _heartbeat; bool _routesMounted = false; + /// Upstream connections of the Jellyfin relay; created on first use. + HttpClient? _relayClient; SyncServer(this.config) { - hub = ServerHub(sessions, config.secret, hostKey: config.hostKey); + hub = ServerHub(sessions, config.secret, hostKey: config.hostKey, jellyfin: config.jellyfin); } bool get isRunning => _server != null; @@ -59,14 +62,17 @@ class SyncServer { ); }); mountSessionMediaRoute(_app, sessions); + mountJellyfinRelayRoute( + _app, sessions, () => hub.jellyfin, () => _relayClient ??= HttpClient()); + // advertiseHost is only an explicit override; otherwise echo what the client dialled. + String authority(String dialledHost) => config.advertiseHost.isNotEmpty + ? '${config.advertiseHost}:$boundPort' + : dialledHost; mountSyncRoute(_app, hub, (sid, token, media, dialledHost) { final name = Uri.encodeComponent(media?.name ?? 'file'); - // advertiseHost is only an explicit override; otherwise echo what the client dialled. - final authority = config.advertiseHost.isNotEmpty - ? '${config.advertiseHost}:$boundPort' - : dialledHost; - return 'http://$authority/m/$sid/$token/$name'; - }); + return 'http://${authority(dialledHost)}/m/$sid/$token/$name'; + }, relayUrlBuilder: (sid, token, itemId, dialledHost) => + 'http://${authority(dialledHost)}/j/$sid/$token/${Uri.encodeComponent(itemId)}'); _app.fallback = (final req) => Response.notFound( body: Body.fromString('not found'), ); @@ -89,6 +95,7 @@ class SyncServer { if (config.secret.isEmpty) { _log.warning('no password set: open server, anyone who can reach it can join'); } + await _resolveJellyfinServerId(); _mountRoutes(); _server = await _app.serve( address: InternetAddress.anyIPv4, @@ -110,10 +117,32 @@ class SyncServer { _log.info('server "${config.name}" listening on $boundPort'); } + /// A configured credential without a ServerId gets it from + /// `/System/Info/Public` (unauthenticated). On failure the relay stays off. + Future _resolveJellyfinServerId() async { + final cred = hub.jellyfin; + if (cred == null) return; + if (cred.serverId.isEmpty) { + final client = JellyfinClient(baseUrl: cred.baseUrl, clientInfo: cred.clientInfo); + try { + final info = await client.publicInfo().timeout(const Duration(seconds: 10)); + hub.jellyfin = cred.withServerId(info.id); + } catch (e) { + _log.warning('jellyfin relay off: server id lookup failed: $e'); + return; + } finally { + client.close(); + } + } + _log.info('jellyfin relay on for server ${hub.jellyfin!.serverId}'); + } + Future stop() async { if (_server == null) return; _heartbeat?.cancel(); _heartbeat = null; + _relayClient?.close(force: true); + _relayClient = null; await hub.closeAllSockets(); await _app.close(); _server = null; diff --git a/packages/server/test/jellyfin_relay_live_test.dart b/packages/server/test/jellyfin_relay_live_test.dart new file mode 100644 index 0000000..7843f20 --- /dev/null +++ b/packages/server/test/jellyfin_relay_live_test.dart @@ -0,0 +1,91 @@ +// Live relay test against a real Jellyfin server. Skipped unless the same +// variables as packages/core/test/jellyfin_live_test.dart are set: +// +// JELLYFIN_TEST_URL, JELLYFIN_TEST_TOKEN, JELLYFIN_TEST_CLIENT, +// JELLYFIN_TEST_DEVICE, JELLYFIN_TEST_DEVICE_ID, JELLYFIN_TEST_VERSION +// +// The device fields must be the token's own (Jellyfin overwrites a token's +// stored device fields with the header's). Read-only: it relays 1 KiB of one +// episode and compares it with the same range fetched directly. +library; + +import 'dart:convert'; +import 'dart:io'; + +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:test/test.dart'; + +void main() { + final env = Platform.environment; + const keys = [ + 'JELLYFIN_TEST_URL', + 'JELLYFIN_TEST_TOKEN', + 'JELLYFIN_TEST_CLIENT', + 'JELLYFIN_TEST_DEVICE', + 'JELLYFIN_TEST_DEVICE_ID', + 'JELLYFIN_TEST_VERSION', + ]; + final skip = keys.every(env.containsKey) ? null : 'JELLYFIN_TEST_* not set'; + + test('relays a 1 KiB range byte-exact', () async { + final cred = JellyfinCredential( + baseUrl: env['JELLYFIN_TEST_URL']!, + token: env['JELLYFIN_TEST_TOKEN']!, + client: env['JELLYFIN_TEST_CLIENT']!, + device: env['JELLYFIN_TEST_DEVICE']!, + deviceId: env['JELLYFIN_TEST_DEVICE_ID']!, + version: env['JELLYFIN_TEST_VERSION']!, + ); + final jf = JellyfinClient(baseUrl: cred.baseUrl, clientInfo: cred.clientInfo, accessToken: cred.token); + await jf.validateToken(); + final episode = (await jf.items(const JellyfinItemQuery( + includeItemTypes: [JellyfinItemType.episode], recursive: true, limit: 1))) + .items + .single; + + final server = SyncServer(AppConfig(port: 0, secret: '', name: 'live', jellyfin: cred)); + await server.start(); + final http = HttpClient(); + final ws = await WebSocket.connect('ws://127.0.0.1:${server.boundPort}/sync'); + try { + expect(server.hub.jellyfin!.serverId, isNotEmpty, reason: 'server id looked up'); + final frames = ws.map((m) => jsonDecode('$m') as Map).asBroadcastStream(); + ws.add(jsonEncode(HelloMessage( + proto: RelayConstants.protoVersion, + secret: '', + user: const HelloUser(name: 'live'), + media: const HelloMedia(name: '', size: 0, duration: 0)) + .toJson())); + final welcome = await frames.firstWhere((f) => f['t'] == 'welcome').timeout(const Duration(seconds: 5)); + ws.add(jsonEncode(SwapIntent( + media: HelloMedia( + name: episode.name, + size: 0, + duration: 0, + jellyfin: JellyfinRef( + serverId: server.hub.jellyfin!.serverId, itemId: episode.id, title: episode.name))).toJson())); + await frames.firstWhere((f) => f['t'] == 'state' && f['relay'] == true).timeout(const Duration(seconds: 5)); + + final req = await http.get('127.0.0.1', server.boundPort, '/j/default/${welcome['token']}/${episode.id}'); + req.headers.set('range', 'bytes=0-1023'); + final res = await req.close(); + expect(res.statusCode, 206); + final relayed = [for (final c in await res.toList()) ...c]; + expect(relayed, hasLength(1024)); + + final direct = await http.getUrl(jf.shareableStreamUrl(episode.id)); + jf.authHeaders.forEach(direct.headers.set); + direct.headers.set('range', 'bytes=0-1023'); + final directRes = await direct.close(); + expect(res.headers.value('content-range'), directRes.headers.value('content-range')); + expect(relayed, [for (final c in await directRes.toList()) ...c]); + expect(server.session.totalBytesServed, 1024); + } finally { + await ws.close(); + http.close(force: true); + jf.close(); + await server.stop(); + } + }, skip: skip, timeout: const Timeout(Duration(seconds: 60))); +} diff --git a/packages/server/test/jellyfin_relay_test.dart b/packages/server/test/jellyfin_relay_test.dart new file mode 100644 index 0000000..83cb176 --- /dev/null +++ b/packages/server/test/jellyfin_relay_test.dart @@ -0,0 +1,408 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:test/test.dart'; + +const _serverId = 'fakeserver01'; +const _item = '0123456789abcdef0123456789abcdef'; +const _source = 'fedcba9876543210fedcba9876543210'; +const _slowItem = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; +const _token = 'JF-SECRET-TOKEN'; + +const _cred = JellyfinCredential( + baseUrl: 'http://127.0.0.1:1', // port filled in per test + serverId: _serverId, + token: _token, + client: 'Watch Together', + device: 'host pc "1"', + deviceId: 'dev-42', + version: '0.9.0', +); + +/// Jellyfin stand-in: `/Videos/{id}/stream` serves [bytes] with ranges (the +/// same table as `/m`), `_slowItem` streams forever until the reader goes +/// away, `/System/Info/Public` gives the server id. +class _FakeJellyfin { + final HttpServer _server; + final Uint8List bytes; + final requests = []; + final auth = []; + final uris = []; + final ranges = []; + final slowAborted = Completer(); + int status401 = 0; + _FakeJellyfin._(this._server, this.bytes); + + int get port => _server.port; + + static Future<_FakeJellyfin> start(Uint8List bytes) async { + final s = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + final f = _FakeJellyfin._(s, bytes); + s.listen(f._handle); + return f; + } + + Future _handle(HttpRequest req) async { + final res = req.response; + if (req.uri.path == '/System/Info/Public') { + res.headers.contentType = ContentType.json; + res.write(jsonEncode({'Id': _serverId, 'ServerName': 'fake', 'Version': '10.11.0'})); + await res.close(); + return; + } + auth.add(req.headers.value(HttpHeaders.authorizationHeader)); + uris.add(req.uri); + ranges.add(req.headers.value(HttpHeaders.rangeHeader)); + if (status401 > 0) { + status401--; + res.statusCode = 401; + await res.close(); + return; + } + final m = RegExp(r'^/Videos/([^/]+)/stream$').firstMatch(req.uri.path); + if (m == null) { + res.statusCode = 404; + await res.close(); + return; + } + if (m.group(1) == _slowItem) { + // Raw socket: dart:io's HttpResponse.flush does not notice a peer that + // went away, the socket's input closing does. + final socket = await res.detachSocket(writeHeaders: false); + socket.listen((_) {}, onError: (_) {}, onDone: () { + if (!slowAborted.isCompleted) slowAborted.complete(); + }); + socket.write('HTTP/1.1 206 Partial Content\r\nContent-Type: video/x-matroska\r\n' + 'Content-Range: bytes 0-999999999/1000000000\r\nContent-Length: 1000000000\r\n\r\n'); + final chunk = Uint8List(16 * 1024); + try { + while (!slowAborted.isCompleted) { + socket.add(chunk); + await socket.flush(); + await Future.delayed(const Duration(milliseconds: 5)); + } + } catch (_) {} + socket.destroy(); + return; + } + final r = resolveMediaRequest( + rangeHeader: req.headers.value(HttpHeaders.rangeHeader), size: bytes.length, filename: 'x.mkv'); + res.statusCode = r.status; + res.headers.clear(); + r.headers.forEach(res.headers.set); + if (r.status == 200 || r.status == 206) { + final end = r.end ?? bytes.length - 1; + res.contentLength = end - r.start + 1; + res.add(bytes.sublist(r.start, end + 1)); + } + await res.close(); + } + + Future close() => _server.close(force: true); +} + +class _Client { + final WebSocket ws; + final raw = []; + final _welcome = Completer(); + _Client(this.ws) { + ws.listen((m) { + raw.add('$m'); + if ('$m'.contains('"t":"welcome"') && !_welcome.isCompleted) _welcome.complete('$m'); + }, onError: (_) {}, onDone: () {}); + } + + void send(Map j) => ws.add(jsonEncode(j)); + + Map? get lastState { + for (final f in raw.reversed) { + final d = jsonDecode(f) as Map; + if (d['t'] == 'state') return Map.from(d); + } + return null; + } + + static Future<(_Client, String)> join(int port, String name) async { + final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync'); + final c = _Client(ws); + c.send(HelloMessage( + proto: RelayConstants.protoVersion, + secret: '', + user: HelloUser(name: name), + media: const HelloMedia(name: '', size: 0, duration: 0), + ).toJson()); + return (c, await c._welcome.future.timeout(const Duration(seconds: 5))); + } +} + +Future _until(bool Function() cond, String what) async { + final deadline = DateTime.now().add(const Duration(seconds: 5)); + while (!cond()) { + if (DateTime.now().isAfter(deadline)) fail('timed out waiting for $what'); + await Future.delayed(const Duration(milliseconds: 20)); + } +} + +void main() { + late Uint8List bytes; + late _FakeJellyfin jf; + late SyncServer server; + late HttpClient http; + final clients = <_Client>[]; + + JellyfinCredential cred({String serverId = _serverId}) => JellyfinCredential( + baseUrl: 'http://127.0.0.1:${jf.port}/', + serverId: serverId, + token: _cred.token, + client: _cred.client, + device: _cred.device, + deviceId: _cred.deviceId, + version: _cred.version); + + Future startServer(JellyfinCredential? c) async { + server = SyncServer(AppConfig(port: 0, secret: '', name: 't', jellyfin: c)); + await server.start(); + } + + setUp(() async { + bytes = Uint8List.fromList(List.generate(100000, (i) => (i * 7) % 256)); + jf = await _FakeJellyfin.start(bytes); + http = HttpClient(); + }); + + tearDown(() async { + for (final c in clients) { + await c.ws.close(); + } + clients.clear(); + http.close(force: true); + await server.stop(); + await jf.close(); + }); + + Future<(_Client, WelcomeMessage)> join(String name) async { + final (c, raw) = await _Client.join(server.boundPort, name); + clients.add(c); + return (c, WelcomeMessage.fromJson(Map.from(jsonDecode(raw) as Map))); + } + + Map refSwap(String itemId, {String? url, String serverId = _serverId}) => { + 't': 'swap', + 'media': { + 'name': 'Pilot', + 'size': 0, + 'duration': 0, + if (url != null) 'url': url, + 'jellyfin': { + 'serverId': serverId, + 'itemId': itemId, + 'mediaSourceId': _source, + 'title': 'Pilot', + 'subtitle': 'Show - S01E01', + 'runtimeSecs': 1440.0, + }, + }, + }; + + Future swapToRef(_Client c, String itemId, {String? url}) async { + final before = c.raw.length; + c.send(refSwap(itemId, url: url)); + late String state; + await _until(() { + for (final f in c.raw.skip(before)) { + if (f.contains('"t":"state"') && f.contains(itemId)) { + state = f; + return true; + } + } + return false; + }, 'state with the ref'); + return state; + } + + Future relayGet(String token, String itemId, {String? range}) async { + final req = await http.get('127.0.0.1', server.boundPort, '/j/default/$token/$itemId'); + if (range != null) req.headers.set('range', range); + return req.close(); + } + + Future> readAll(HttpClientResponse r) async => [for (final c in await r.toList()) ...c]; + + group('wire', () { + test('ref swap: state carries the ref and relay, never a URL or token', () async { + await startServer(cred()); + final (a, _) = await join('alice'); + final state = await swapToRef(a, _item, url: 'http://jf/Videos/x/stream?api_key=LEAK'); + final decoded = jsonDecode(state) as Map; + expect(decoded['relay'], true); + final media = decoded['media'] as Map; + expect(media.containsKey('url'), isFalse); + expect(media['jellyfin'], { + 'serverId': _serverId, + 'itemId': _item, + 'mediaSourceId': _source, + 'title': 'Pilot', + 'subtitle': 'Show - S01E01', + 'runtimeSecs': 1440.0, + }); + expect(state, isNot(contains('LEAK'))); + expect(state, isNot(contains(_token))); + expect(state, isNot(contains('http'))); + expect(server.session.media!.url, isNull); + + // A later joiner gets its own relay URL in welcome, no mediaUrl. + final (_, w) = await join('bob'); + expect(w.mediaUrl, isNull); + expect(w.relayUrl, 'http://127.0.0.1:${server.boundPort}/j/default/${w.token}/$_item'); + expect(w.state!['relay'], true); + expect(jsonEncode(w.state), isNot(contains(_token))); + }); + + test('no relay without a credential, or for another Jellyfin server', () async { + await startServer(cred(serverId: 'otherserver')); + final (a, _) = await join('alice'); + final state = jsonDecode(await swapToRef(a, _item)) as Map; + expect(state.containsKey('relay'), isFalse); + final (_, w) = await join('bob'); + expect(w.relayUrl, isNull); + expect((await relayGet(w.token, _item)).statusCode, 404); + }); + + test('bad ref ids are rejected', () async { + await startServer(cred()); + final (a, _) = await join('alice'); + a.send(refSwap('../../Users')); + await _until(() => a.raw.any((f) => f.contains('bad swap media')), 'err'); + expect(server.session.media, isNull); + }); + + test('report jf shows on the member', () async { + await startServer(cred()); + final (a, w) = await join('alice'); + a.send(const ReportMessage(position: 0, buffered: 0, eventId: 0, jf: {_serverId: false}).toJson()); + await _until(() { + final members = a.lastState?['members'] as List?; + return members != null && members.any((m) => (m as Map)['jf'] != null); + }, 'member jf'); + final m = (a.lastState!['members'] as List).cast().firstWhere((m) => m['id'] == w.memberId); + expect(m['jf'], {_serverId: false}); + }); + }); + + group('relay endpoint', () { + late String token; + + Future setUpRef(String itemId) async { + await startServer(cred()); + final (a, w) = await join('alice'); + token = w.token; + await swapToRef(a, itemId); + } + + test('sends exactly the configured Authorization fields, static original stream', () async { + await setUpRef(_item); + final r = await relayGet(token, _item, range: 'bytes=0-9'); + await r.drain(); + expect(jf.auth.single, + 'MediaBrowser Client="Watch%20Together", Device="host%20pc%20%221%22", DeviceId="dev-42", Version="0.9.0", Token="JF-SECRET-TOKEN"'); + expect(jf.uris.single.path, '/Videos/$_item/stream'); + expect(jf.uris.single.queryParameters, {'static': 'true', 'mediaSourceId': _source}); + }); + + test('no Range: 200 full body, headers passed through', () async { + await setUpRef(_item); + final r = await relayGet(token, _item); + expect(r.statusCode, 200); + expect(r.headers.value('content-length'), '${bytes.length}'); + expect(r.headers.value('accept-ranges'), 'bytes'); + expect(r.headers.contentType?.mimeType, 'video/x-matroska'); + expect(await readAll(r), bytes); + expect(jf.ranges.single, isNull); + }); + + test('closed, open-ended and suffix ranges forward and come back byte-exact', () async { + await setUpRef(_item); + var r = await relayGet(token, _item, range: 'bytes=100-199'); + expect(r.statusCode, 206); + expect(r.headers.value('content-range'), 'bytes 100-199/${bytes.length}'); + expect(r.headers.value('content-length'), '100'); + expect(await readAll(r), bytes.sublist(100, 200)); + + r = await relayGet(token, _item, range: 'bytes=99000-'); + expect(r.statusCode, 206); + expect(r.headers.value('content-range'), 'bytes 99000-99999/${bytes.length}'); + expect(await readAll(r), bytes.sublist(99000)); + + r = await relayGet(token, _item, range: 'bytes=-256'); + expect(r.statusCode, 206); + expect(r.headers.value('content-range'), 'bytes 99744-99999/${bytes.length}'); + expect(await readAll(r), bytes.sublist(bytes.length - 256)); + expect(jf.ranges, ['bytes=100-199', 'bytes=99000-', 'bytes=-256']); + }); + + test('unsatisfiable range: 416 with Content-Range passed through', () async { + await setUpRef(_item); + final r = await relayGet(token, _item, range: 'bytes=500000-'); + expect(r.statusCode, 416); + expect(r.headers.value('content-range'), 'bytes */${bytes.length}'); + await r.drain(); + }); + + test('bytes are counted as sent', () async { + await setUpRef(_item); + await readAll(await relayGet(token, _item, range: 'bytes=0-999')); + await readAll(await relayGet(token, _item, range: 'bytes=-24')); + expect(server.session.tokens[token]!.bytesServed, 1024); + }); + + test('gates: bad token 401, other item 403, unknown session 410', () async { + await setUpRef(_item); + expect((await relayGet('nope', _item)).statusCode, 401); + expect((await relayGet(token, _slowItem)).statusCode, 403); + final req = await http.get('127.0.0.1', server.boundPort, '/j/other/$token/$_item'); + expect((await req.close()).statusCode, 410); + expect(jf.uris, isEmpty, reason: 'nothing reached Jellyfin'); + }); + + test('non-Jellyfin media: 404', () async { + await startServer(cred()); + final (a, w) = await join('alice'); + a.send(const SwapIntent(media: HelloMedia(name: 'x', size: 0, duration: 0, url: 'http://x/a.mkv')).toJson()); + await _until(() => server.session.media?.url != null, 'url swap'); + expect((await relayGet(w.token, _item)).statusCode, 404); + }); + + test('Jellyfin rejecting the credential is a 502, not the viewer\'s 401', () async { + await setUpRef(_item); + jf.status401 = 1; + expect((await relayGet(token, _item)).statusCode, 502); + }); + + test('viewer disconnect aborts the upstream transfer', () async { + await setUpRef(_slowItem); + final socket = await Socket.connect(InternetAddress.loopbackIPv4, server.boundPort); + socket.write('GET /j/default/$token/$_slowItem HTTP/1.1\r\nHost: 127.0.0.1\r\nRange: bytes=0-\r\n\r\n'); + var got = 0; + final enough = Completer(); + socket.listen((d) { + got += d.length; + if (got > 64 * 1024 && !enough.isCompleted) enough.complete(); + }, onError: (_) {}, onDone: () {}); + await enough.future.timeout(const Duration(seconds: 5)); + socket.destroy(); + await jf.slowAborted.future.timeout(const Duration(seconds: 5)); + final served = server.session.tokens[token]!.bytesServed; + expect(served, greaterThan(0)); + expect(served, lessThan(1000000000)); + }); + }); + + test('a credential without serverId looks it up at start', () async { + await startServer(cred(serverId: '')); + expect(server.hub.jellyfin!.serverId, _serverId); + }); +} diff --git a/packages/server/test/server_auth_test.dart b/packages/server/test/server_auth_test.dart index 1456143..2addafe 100644 --- a/packages/server/test/server_auth_test.dart +++ b/packages/server/test/server_auth_test.dart @@ -18,7 +18,7 @@ void main() { setUp(() async { bytes = Uint8List.fromList(List.generate(8192, (i) => i % 251)); file = File( - '${Directory.systemTemp.path}/server_test_${DateTime.now().microsecondsSinceEpoch}.bin'); + '${Directory.systemTemp.path}/server_test_${DateTime.now().microsecondsSinceEpoch}.mkv'); await file.writeAsBytes(bytes); server = SyncServer(AppConfig( port: 0, secret: 'shh', name: 'test', mediaPath: file.path)); @@ -117,6 +117,8 @@ void main() { resp = await mediaGet('/m/default/$token/$name', range: 'bytes=0-99'); expect(resp.statusCode, 206); expect(resp.headers.value('content-range'), 'bytes 0-99/8192'); + expect(resp.headers.contentType?.mimeType, contentTypeForFile(name)); + expect(contentTypeForFile(name), isNot('application/octet-stream')); final body = []; await for (final c in resp) { body.addAll(c); diff --git a/packages/server/test/session_test.dart b/packages/server/test/session_test.dart index 0c3c03b..a9737c9 100644 --- a/packages/server/test/session_test.dart +++ b/packages/server/test/session_test.dart @@ -80,8 +80,8 @@ void main() { }); group('core welcome + proto', () { - test('proto 6: state broadcast + hello hostKey', () { - expect(RelayConstants.protoVersion, 6); + test('proto 7: Jellyfin refs and relay', () { + expect(RelayConstants.protoVersion, 7); }); test('welcome round-trips with token + member id', () {