diff --git a/js/components/src/components/mobile-player/player.tsx b/js/components/src/components/mobile-player/player.tsx index 71a81a01d..b5bdcc5e2 100644 --- a/js/components/src/components/mobile-player/player.tsx +++ b/js/components/src/components/mobile-player/player.tsx @@ -58,35 +58,38 @@ export function Player( } }, [streamHasBFrames, streamForcesHLS, protocol, setStreamForcesHLS]); - // Pre-live over HLS: the streamer previewing their own stream needs a - // playback token (the HLS requests carry no session). It only opens the - // caller's own stream, so ask when this is that viewer, and again before - // it expires. Lives here rather than in the app's wrapper so every place - // that mounts a player (the live dashboard included) gets it. + // Pre-live over HLS: the streamer previewing their own stream needs their + // own playback session (an anonymous one opens published streams only). + // It only opens the caller's own stream, so ask when this is that viewer, + // and again before it expires. Lives here rather than in the app's + // wrapper so every place that mounts a player (the live dashboard + // included) gets it. const myDid = useDID(); const streamerDid = useLivestreamStoreOptional((x) => x.profile?.did); const agent = usePDSAgent(); - const setLiveToken = usePlayerStore((x) => x.setLiveToken); + const setPlaybackSession = usePlayerStore((x) => x.setPlaybackSession); const isMine = !!myDid && (props.src === myDid || (!!streamerDid && streamerDid === myDid)); useEffect(() => { if (!isMine || !agent) { - setLiveToken(undefined); + setPlaybackSession(undefined); return; } let cancelled = false; let timer: ReturnType | null = null; const fetchToken = async () => { try { - const res = await agent.client.call(place.stream.playback.getLiveToken); + const res = await agent.client.call( + place.stream.playback.getPlaybackSession, + ); if (cancelled) return; - setLiveToken(res.token); + setPlaybackSession(res.sid); const msLeft = new Date(res.expiresAt).getTime() - Date.now(); timer = setTimeout(fetchToken, Math.max(60_000, msLeft - 5 * 60_000)); } catch (e) { if (cancelled) return; - console.warn("could not fetch a live playback token", e); + console.warn("could not fetch a playback session", e); timer = setTimeout(fetchToken, 60_000); } }; @@ -94,9 +97,9 @@ export function Player( return () => { cancelled = true; if (timer) clearTimeout(timer); - setLiveToken(undefined); + setPlaybackSession(undefined); }; - }, [isMine, agent, setLiveToken]); + }, [isMine, agent, setPlaybackSession]); // if we set muted, set it and restore after useEffect(() => { diff --git a/js/components/src/components/mobile-player/shared.tsx b/js/components/src/components/mobile-player/shared.tsx index afde0776b..6c63a89ec 100644 --- a/js/components/src/components/mobile-player/shared.tsx +++ b/js/components/src/components/mobile-player/shared.tsx @@ -22,8 +22,9 @@ export function srcToUrl( protocol: string; } { const url = useStreamplaceStore((x) => x.url); - // The streamer's pre-live playback token, when their player holds one. - const liveToken = usePlayerStore((x) => x.liveToken); + // The streamer's own playback session, when their player holds one; it + // opens their pre-live stream. Anyone else gets a session from the node. + const playbackSession = usePlayerStore((x) => x.playbackSession); return useMemo(() => { if (props.src.startsWith("at://")) { const aturi = new AtUri(props.src); @@ -57,8 +58,8 @@ export function srcToUrl( } else { outUrl = `${url}/xrpc/place.stream.playback.getLivePlaylist?streamer=${props.src}`; } - if (liveToken) { - outUrl += `&token=${encodeURIComponent(liveToken)}`; + if (playbackSession) { + outUrl += `&sid=${encodeURIComponent(playbackSession)}`; } } else if (protocol === PlayerProtocol.PROGRESSIVE_MP4) { outUrl = `${url}/api/playback/${props.src}/stream.mp4`; @@ -73,5 +74,5 @@ export function srcToUrl( protocol: protocol, url: outUrl, }; - }, [props.src, props.selectedRendition, protocol, url, liveToken]); + }, [props.src, props.selectedRendition, protocol, url, playbackSession]); } diff --git a/js/components/src/player-store/player-state.tsx b/js/components/src/player-store/player-state.tsx index ace90f8d2..0259e9830 100644 --- a/js/components/src/player-store/player-state.tsx +++ b/js/components/src/player-store/player-state.tsx @@ -55,10 +55,10 @@ export interface PlayerState { /** Source (streamer did) */ src: string; - /** A pre-live playback token (getLiveToken) for the streamer's own + /** The streamer's own playback session (getPlaybackSession) for their * unpublished stream, threaded onto the HLS playlist URL. */ - liveToken?: string; - setLiveToken: (token: string | undefined) => void; + playbackSession?: string; + setPlaybackSession: (sid: string | undefined) => void; /** Function to set the source URL */ setSrc: (src: string) => void; diff --git a/js/components/src/player-store/player-store.tsx b/js/components/src/player-store/player-store.tsx index 1245c9269..d7888785e 100644 --- a/js/components/src/player-store/player-store.tsx +++ b/js/components/src/player-store/player-store.tsx @@ -84,8 +84,9 @@ export const makePlayerStore = (id?: string): StoreApi => { src: "", setSrc: (src: string) => set(() => ({ src })), - liveToken: undefined, - setLiveToken: (liveToken: string | undefined) => set(() => ({ liveToken })), + playbackSession: undefined, + setPlaybackSession: (playbackSession: string | undefined) => + set(() => ({ playbackSession })), ingestMediaSource: undefined, setIngestMediaSource: (ingestMediaSource: IngestMediaSource | undefined) => diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 16b11f08d..3e4268ca5 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -1277,16 +1277,6 @@ "type": "string", "description": "Opaque playback session identifier. Omit on the master playlist request; the server generates one and threads it through the sub-playlist + segment URLs it returns, for view-count correlation." } - }, - { - "name": "token", - "in": "query", - "required": false, - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live.", - "schema": { - "type": "string", - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live." - } } ] } @@ -1373,51 +1363,10 @@ "type": "string", "description": "Opaque playback session identifier, propagated from the media playlist that referenced this segment. Logged for view-count correlation; not used for access control." } - }, - { - "name": "token", - "in": "query", - "required": false, - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live.", - "schema": { - "type": "string", - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live." - } } ] } }, - "/xrpc/place.stream.playback.getLiveToken": { - "get": { - "summary": "Mint a short-lived playback token for the caller's own live stream, so their player can watch it over HLS before it is published (pre-live). HLS requests carry no session; the token, passed as `token` on getLivePlaylist and getLiveSegment, stands in for one. Requires an authenticated session; the token is bound to the caller's DID.", - "operationId": "place.stream.playback.getLiveToken", - "tags": ["place.stream.playback"], - "responses": { - "200": { - "description": "Success", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "token": { - "type": "string", - "description": "Opaque token to pass as the `token` query parameter." - }, - "expiresAt": { - "type": "string", - "description": "When the token stops working; fetch a new one before then.", - "format": "date-time" - } - }, - "required": ["token", "expiresAt"] - } - } - } - } - } - } - }, "/xrpc/place.stream.playback.getPlaybackServer": { "get": { "summary": "Get available playback servers for a livestream.", @@ -1459,6 +1408,37 @@ ] } }, + "/xrpc/place.stream.playback.getPlaybackSession": { + "get": { + "summary": "Mint the caller's own playback session for their live stream. Every playback URL carries a signed session as `sid`; the anonymous one a playlist hands out opens published streams only, while the owner session this returns also opens the caller's unpublished (pre-live) stream to their own player. Requires an authenticated session; the result is bound to the caller's DID. Pass it as `sid` on getLivePlaylist.", + "operationId": "place.stream.playback.getPlaybackSession", + "tags": ["place.stream.playback"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "sid": { + "type": "string", + "description": "The signed session to pass as the `sid` query parameter." + }, + "expiresAt": { + "type": "string", + "description": "When the session stops opening the stream unless renewed; fetch a new one before then.", + "format": "date-time" + } + }, + "required": ["sid", "expiresAt"] + } + } + } + } + } + } + }, "/xrpc/place.stream.playback.getVideoBlob": { "get": { "summary": "Fetch a content-addressed MUXL playback blob by CID. Serves either a primary fMP4 archive or a per-track init segment — both are CID-addressed and indistinguishable from the endpoint's point of view. HTTP `Range` headers are honored; HLS media playlists generated by getVideoPlaylist reference this endpoint with `EXT-X-BYTERANGE` to serve segments out of the primary blob. The `did` identifies the owning account for egress accounting: the blob itself is content-addressed and owner-agnostic, so the caller carries the attribution.", diff --git a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getliveplaylist.md b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getliveplaylist.md index c1a8e7afd..a8d7faf5b 100644 --- a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getliveplaylist.md +++ b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getliveplaylist.md @@ -22,7 +22,6 @@ Get an HLS CMAF playlist for a live stream. Returns a master playlist when `trac | `streamer` | `string` | ✅ | The streamer to play back: a DID (did:plc/did:web/did:key) or a Bluesky handle, which is resolved to its DID. | | | `track` | `string` | ❌ | Track ID (stringified u32 matching the MUXL container) for a single-track media playlist. Omit for the master playlist. | | | `sid` | `string` | ❌ | Opaque playback session identifier. Omit on the master playlist request; the server generates one and threads it through the sub-playlist + segment URLs it returns, for view-count correlation. | | -| `token` | `string` | ❌ | A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live. | | **Output:** @@ -63,10 +62,6 @@ _Schema not defined._ "sid": { "type": "string", "description": "Opaque playback session identifier. Omit on the master playlist request; the server generates one and threads it through the sub-playlist + segment URLs it returns, for view-count correlation." - }, - "token": { - "type": "string", - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live." } } }, diff --git a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getlivesegment.md b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getlivesegment.md index efdb1ed05..ce0de5aa4 100644 --- a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getlivesegment.md +++ b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getlivesegment.md @@ -23,7 +23,6 @@ Fetch a single live HLS segment, or a track's init segment, from the in-memory l | `track` | `string` | ✅ | Track ID (stringified u32 matching the MUXL container). | | | `seg` | `string` | ✅ | `init` for the track's init segment, or the segment's media-sequence number. A trailing `.m4s` is accepted and ignored. | | | `sid` | `string` | ❌ | Opaque playback session identifier, propagated from the media playlist that referenced this segment. Logged for view-count correlation; not used for access control. | | -| `token` | `string` | ❌ | A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live. | | **Output:** @@ -67,10 +66,6 @@ _Schema not defined._ "sid": { "type": "string", "description": "Opaque playback session identifier, propagated from the media playlist that referenced this segment. Logged for view-count correlation; not used for access control." - }, - "token": { - "type": "string", - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live." } } }, diff --git a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getlivetoken.md b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getlivetoken.md deleted file mode 100644 index 2d966d4cd..000000000 --- a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getlivetoken.md +++ /dev/null @@ -1,65 +0,0 @@ ---- -title: place.stream.playback.getLiveToken -description: Reference for the place.stream.playback.getLiveToken lexicon ---- - -**Lexicon Version:** 1 - -## Definitions - - - -### `main` - -**Type:** `query` - -Mint a short-lived playback token for the caller's own live stream, so their player can watch it over HLS before it is published (pre-live). HLS requests carry no session; the token, passed as `token` on getLivePlaylist and getLiveSegment, stands in for one. Requires an authenticated session; the token is bound to the caller's DID. - -**Parameters:** _(None defined)_ - -**Output:** - -- **Encoding:** `application/json` -- **Schema:** - -**Schema Type:** `object` - -| Name | Type | Req'd | Description | Constraints | -| ----------- | -------- | ----- | ---------------------------------------------------------- | ------------------ | -| `token` | `string` | ✅ | Opaque token to pass as the `token` query parameter. | | -| `expiresAt` | `string` | ✅ | When the token stops working; fetch a new one before then. | Format: `datetime` | - ---- - -## Lexicon Source - -```json -{ - "lexicon": 1, - "id": "place.stream.playback.getLiveToken", - "defs": { - "main": { - "type": "query", - "description": "Mint a short-lived playback token for the caller's own live stream, so their player can watch it over HLS before it is published (pre-live). HLS requests carry no session; the token, passed as `token` on getLivePlaylist and getLiveSegment, stands in for one. Requires an authenticated session; the token is bound to the caller's DID.", - "output": { - "encoding": "application/json", - "schema": { - "type": "object", - "required": ["token", "expiresAt"], - "properties": { - "token": { - "type": "string", - "description": "Opaque token to pass as the `token` query parameter." - }, - "expiresAt": { - "type": "string", - "format": "datetime", - "description": "When the token stops working; fetch a new one before then." - } - } - } - } - } - } -} -``` diff --git a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getplaybacksession.md b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getplaybacksession.md new file mode 100644 index 000000000..6fddcdaae --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-getplaybacksession.md @@ -0,0 +1,65 @@ +--- +title: place.stream.playback.getPlaybackSession +description: Reference for the place.stream.playback.getPlaybackSession lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `query` + +Mint the caller's own playback session for their live stream. Every playback URL carries a signed session as `sid`; the anonymous one a playlist hands out opens published streams only, while the owner session this returns also opens the caller's unpublished (pre-live) stream to their own player. Requires an authenticated session; the result is bound to the caller's DID. Pass it as `sid` on getLivePlaylist. + +**Parameters:** _(None defined)_ + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| ----------- | -------- | ----- | -------------------------------------------------------------------------------------- | ------------------ | +| `sid` | `string` | ✅ | The signed session to pass as the `sid` query parameter. | | +| `expiresAt` | `string` | ✅ | When the session stops opening the stream unless renewed; fetch a new one before then. | Format: `datetime` | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.playback.getPlaybackSession", + "defs": { + "main": { + "type": "query", + "description": "Mint the caller's own playback session for their live stream. Every playback URL carries a signed session as `sid`; the anonymous one a playlist hands out opens published streams only, while the owner session this returns also opens the caller's unpublished (pre-live) stream to their own player. Requires an authenticated session; the result is bound to the caller's DID. Pass it as `sid` on getLivePlaylist.", + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["sid", "expiresAt"], + "properties": { + "sid": { + "type": "string", + "description": "The signed session to pass as the `sid` query parameter." + }, + "expiresAt": { + "type": "string", + "format": "datetime", + "description": "When the session stops opening the stream unless renewed; fetch a new one before then." + } + } + } + } + } + } +} +``` diff --git a/lexicons/place/stream/playback/getLivePlaylist.json b/lexicons/place/stream/playback/getLivePlaylist.json index 2eae6364b..594c14bd5 100644 --- a/lexicons/place/stream/playback/getLivePlaylist.json +++ b/lexicons/place/stream/playback/getLivePlaylist.json @@ -20,10 +20,6 @@ "sid": { "type": "string", "description": "Opaque playback session identifier. Omit on the master playlist request; the server generates one and threads it through the sub-playlist + segment URLs it returns, for view-count correlation." - }, - "token": { - "type": "string", - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live." } } }, diff --git a/lexicons/place/stream/playback/getLiveSegment.json b/lexicons/place/stream/playback/getLiveSegment.json index b64dcefa3..48519db8e 100644 --- a/lexicons/place/stream/playback/getLiveSegment.json +++ b/lexicons/place/stream/playback/getLiveSegment.json @@ -24,10 +24,6 @@ "sid": { "type": "string", "description": "Opaque playback session identifier, propagated from the media playlist that referenced this segment. Logged for view-count correlation; not used for access control." - }, - "token": { - "type": "string", - "description": "A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live." } } }, diff --git a/lexicons/place/stream/playback/getLiveToken.json b/lexicons/place/stream/playback/getLiveToken.json deleted file mode 100644 index 4c21b92a9..000000000 --- a/lexicons/place/stream/playback/getLiveToken.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "lexicon": 1, - "id": "place.stream.playback.getLiveToken", - "defs": { - "main": { - "type": "query", - "description": "Mint a short-lived playback token for the caller's own live stream, so their player can watch it over HLS before it is published (pre-live). HLS requests carry no session; the token, passed as `token` on getLivePlaylist and getLiveSegment, stands in for one. Requires an authenticated session; the token is bound to the caller's DID.", - "output": { - "encoding": "application/json", - "schema": { - "type": "object", - "required": ["token", "expiresAt"], - "properties": { - "token": { - "type": "string", - "description": "Opaque token to pass as the `token` query parameter." - }, - "expiresAt": { - "type": "string", - "format": "datetime", - "description": "When the token stops working; fetch a new one before then." - } - } - } - } - } - } -} diff --git a/lexicons/place/stream/playback/getPlaybackSession.json b/lexicons/place/stream/playback/getPlaybackSession.json new file mode 100644 index 000000000..3a1ea626e --- /dev/null +++ b/lexicons/place/stream/playback/getPlaybackSession.json @@ -0,0 +1,28 @@ +{ + "lexicon": 1, + "id": "place.stream.playback.getPlaybackSession", + "defs": { + "main": { + "type": "query", + "description": "Mint the caller's own playback session for their live stream. Every playback URL carries a signed session as `sid`; the anonymous one a playlist hands out opens published streams only, while the owner session this returns also opens the caller's unpublished (pre-live) stream to their own player. Requires an authenticated session; the result is bound to the caller's DID. Pass it as `sid` on getLivePlaylist.", + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["sid", "expiresAt"], + "properties": { + "sid": { + "type": "string", + "description": "The signed session to pass as the `sid` query parameter." + }, + "expiresAt": { + "type": "string", + "format": "datetime", + "description": "When the session stops opening the stream unless renewed; fetch a new one before then." + } + } + } + } + } + } +} diff --git a/pkg/placestream/playbackgetliveplaylist.go b/pkg/placestream/playbackgetliveplaylist.go index ac2ebaef1..39d73c5a4 100644 --- a/pkg/placestream/playbackgetliveplaylist.go +++ b/pkg/placestream/playbackgetliveplaylist.go @@ -17,18 +17,14 @@ import ( // // sid: Opaque playback session identifier. Omit on the master playlist request; the server generates one and threads it through the sub-playlist + segment URLs it returns, for view-count correlation. // streamer: The streamer to play back: a DID (did:plc/did:web/did:key) or a Bluesky handle, which is resolved to its DID. -// token: A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live. // track: Track ID (stringified u32 matching the MUXL container) for a single-track media playlist. Omit for the master playlist. -func PlaybackGetLivePlaylist(ctx context.Context, c glex.LexClient, sid string, streamer string, token string, track string) ([]byte, error) { +func PlaybackGetLivePlaylist(ctx context.Context, c glex.LexClient, sid string, streamer string, track string) ([]byte, error) { buf := new(bytes.Buffer) params := map[string]interface{}{} if sid != "" { params["sid"] = sid } - if token != "" { - params["token"] = token - } if track != "" { params["track"] = track } diff --git a/pkg/placestream/playbackgetlivesegment.go b/pkg/placestream/playbackgetlivesegment.go index f92836514..2f5dc1715 100644 --- a/pkg/placestream/playbackgetlivesegment.go +++ b/pkg/placestream/playbackgetlivesegment.go @@ -18,18 +18,14 @@ import ( // seg: `init` for the track's init segment, or the segment's media-sequence number. A trailing `.m4s` is accepted and ignored. // sid: Opaque playback session identifier, propagated from the media playlist that referenced this segment. Logged for view-count correlation; not used for access control. // streamer: The streamer: a DID or a Bluesky handle (resolved to its DID). -// token: A playback token from getLiveToken. Lets the streamer watch their own stream before it is published; ignored once the stream is live. // track: Track ID (stringified u32 matching the MUXL container). -func PlaybackGetLiveSegment(ctx context.Context, c glex.LexClient, seg string, sid string, streamer string, token string, track string) ([]byte, error) { +func PlaybackGetLiveSegment(ctx context.Context, c glex.LexClient, seg string, sid string, streamer string, track string) ([]byte, error) { buf := new(bytes.Buffer) params := map[string]interface{}{} if sid != "" { params["sid"] = sid } - if token != "" { - params["token"] = token - } params["seg"] = seg params["streamer"] = streamer params["track"] = track diff --git a/pkg/placestream/playbackgetlivetoken.go b/pkg/placestream/playbackgetlivetoken.go deleted file mode 100644 index 496171315..000000000 --- a/pkg/placestream/playbackgetlivetoken.go +++ /dev/null @@ -1,53 +0,0 @@ -// Code generated by glex; DO NOT EDIT. - -// Lexicon schema: place.stream.playback.getLiveToken - -package placestream - -import ( - "context" - "io" - - glex "github.com/streamplace/glex/runtime" - cbg "github.com/whyrusleeping/cbor-gen" -) - -type PlaybackGetLiveToken_Output struct { - LexiconTypeID string `json:"$type,omitempty"` - // expiresAt: When the token stops working; fetch a new one before then. - ExpiresAt string `json:"expiresAt"` - // token: Opaque token to pass as the `token` query parameter. - Token string `json:"token"` -} - -// RecordTypeID implements glex.Record. -func (t *PlaybackGetLiveToken_Output) RecordTypeID() string { - return "place.stream.playback.getLiveToken" -} - -func (t *PlaybackGetLiveToken_Output) MarshalCBOR(w io.Writer) error { - if t == nil { - _, err := w.Write(cbg.CborNull) - return err - } - // stamp $type on a copy so marshal never mutates the record - cp := *t - cp.LexiconTypeID = "place.stream.playback.getLiveToken" - return glex.MarshalCBOR(w, &cp) -} - -func (t *PlaybackGetLiveToken_Output) UnmarshalCBOR(r io.Reader) error { - return glex.UnmarshalCBOR(r, t) -} - -// PlaybackGetLiveToken calls the XRPC method "place.stream.playback.getLiveToken". -// -// Mint a short-lived playback token for the caller's own live stream, so their player can watch it over HLS before it is published (pre-live). HLS requests carry no session; the token, passed as `token` on getLivePlaylist and getLiveSegment, stands in for one. Requires an authenticated session; the token is bound to the caller's DID. -func PlaybackGetLiveToken(ctx context.Context, c glex.LexClient) (*PlaybackGetLiveToken_Output, error) { - var out PlaybackGetLiveToken_Output - - if err := c.LexDo(ctx, glex.Query, "", "place.stream.playback.getLiveToken", nil, nil, &out); err != nil { - return nil, err - } - return &out, nil -} diff --git a/pkg/placestream/playbackgetplaybacksession.go b/pkg/placestream/playbackgetplaybacksession.go new file mode 100644 index 000000000..22359a10d --- /dev/null +++ b/pkg/placestream/playbackgetplaybacksession.go @@ -0,0 +1,53 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.playback.getPlaybackSession + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type PlaybackGetPlaybackSession_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + // expiresAt: When the session stops opening the stream unless renewed; fetch a new one before then. + ExpiresAt string `json:"expiresAt"` + // sid: The signed session to pass as the `sid` query parameter. + Sid string `json:"sid"` +} + +// RecordTypeID implements glex.Record. +func (t *PlaybackGetPlaybackSession_Output) RecordTypeID() string { + return "place.stream.playback.getPlaybackSession" +} + +func (t *PlaybackGetPlaybackSession_Output) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.playback.getPlaybackSession" + return glex.MarshalCBOR(w, &cp) +} + +func (t *PlaybackGetPlaybackSession_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// PlaybackGetPlaybackSession calls the XRPC method "place.stream.playback.getPlaybackSession". +// +// Mint the caller's own playback session for their live stream. Every playback URL carries a signed session as `sid`; the anonymous one a playlist hands out opens published streams only, while the owner session this returns also opens the caller's unpublished (pre-live) stream to their own player. Requires an authenticated session; the result is bound to the caller's DID. Pass it as `sid` on getLivePlaylist. +func PlaybackGetPlaybackSession(ctx context.Context, c glex.LexClient) (*PlaybackGetPlaybackSession_Output, error) { + var out PlaybackGetPlaybackSession_Output + + if err := c.LexDo(ctx, glex.Query, "", "place.stream.playback.getPlaybackSession", nil, nil, &out); err != nil { + return nil, err + } + return &out, nil +} diff --git a/pkg/psession/psession.go b/pkg/psession/psession.go new file mode 100644 index 000000000..9f4c0457d --- /dev/null +++ b/pkg/psession/psession.go @@ -0,0 +1,144 @@ +// Package psession is the signed playback session identifier: the one +// credential a playback URL carries. HLS clients (native players above all) +// can't sign requests, so whatever a playback URL is allowed to fetch has to +// be readable from the URL itself. A session names a playback (its TID, the +// key viewer counts and view logs group by), a scope (what it may fetch), an +// expiry, and is HMAC-signed together with the streamer it belongs to, so a +// session minted for one stream opens no other and a forged one opens +// nothing. +// +// Wire form, carried as the sid query parameter: +// +// 1.... +// +// The signature is the first 16 bytes of HMAC-SHA256 over +// "1|tid|scope|expiry|streamer", base64url without padding. +package psession + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/base64" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + "stream.place/streamplace/pkg/spid" +) + +// Scope is what a session may fetch. +type Scope string + +const ( + // ScopePublic is an ordinary viewer: published live streams and videos. + ScopePublic Scope = "p" + // ScopeOwner is the streamer's own session: it also opens their + // unpublished (pre-live) stream. + ScopeOwner Scope = "o" + + version = "1" + // sigBytes of the HMAC are kept: 128 bits is plenty for a token whose + // worst-case leak opens one stream to one more viewer. + sigBytes = 16 +) + +var ( + ErrMalformed = errors.New("playback session: malformed") + ErrSignature = errors.New("playback session: bad signature") + ErrExpired = errors.New("playback session: expired") +) + +// Session is a parsed, verified playback session. +type Session struct { + // ID is the session's TID: what viewer counts and view logs key by, + // stable across renewals. + ID string + Scope Scope + Expires time.Time +} + +func sign(key []byte, id string, scope Scope, exp int64, streamer string) string { + mac := hmac.New(sha256.New, key) + fmt.Fprintf(mac, "%s|%s|%s|%d|%s", version, id, scope, exp, streamer) + return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)[:sigBytes]) +} + +func encode(key []byte, id string, scope Scope, exp time.Time, streamer string) string { + e := exp.Unix() + return strings.Join([]string{version, id, string(scope), strconv.FormatInt(e, 10), sign(key, id, scope, e, streamer)}, ".") +} + +// Mint issues a new session for streamer, valid for ttl from now. +func Mint(key []byte, streamer string, scope Scope, ttl time.Duration, now time.Time) (Session, string) { + s := Session{ID: spid.TID(), Scope: scope, Expires: now.Add(ttl).Truncate(time.Second).UTC()} + return s, encode(key, s.ID, s.Scope, s.Expires, streamer) +} + +// Renew re-issues s for streamer with a fresh expiry, keeping its ID so the +// playback keeps counting as one session. +func Renew(key []byte, s Session, streamer string, ttl time.Duration, now time.Time) (Session, string) { + s.Expires = now.Add(ttl).Truncate(time.Second).UTC() + return s, encode(key, s.ID, s.Scope, s.Expires, streamer) +} + +// WellFormed reports whether value has the shape of a session, without +// checking its signature: what a handler can refuse before it has a key. +func WellFormed(value string) bool { + parts := strings.Split(value, ".") + if len(parts) != 5 || parts[0] != version { + return false + } + if _, err := syntax.ParseTID(parts[1]); err != nil { + return false + } + if scope := Scope(parts[2]); scope != ScopePublic && scope != ScopeOwner { + return false + } + _, err := strconv.ParseInt(parts[3], 10, 64) + return err == nil && parts[4] != "" +} + +// Parse verifies value as a session for streamer at now. An expired session +// comes back with ErrExpired and the parsed session, so a caller can renew +// it in place. +func Parse(key []byte, value, streamer string, now time.Time) (Session, error) { + parts := strings.Split(value, ".") + if len(parts) != 5 || parts[0] != version { + return Session{}, ErrMalformed + } + if _, err := syntax.ParseTID(parts[1]); err != nil { + return Session{}, ErrMalformed + } + scope := Scope(parts[2]) + if scope != ScopePublic && scope != ScopeOwner { + return Session{}, ErrMalformed + } + exp, err := strconv.ParseInt(parts[3], 10, 64) + if err != nil { + return Session{}, ErrMalformed + } + if !hmac.Equal([]byte(parts[4]), []byte(sign(key, parts[1], scope, exp, streamer))) { + return Session{}, ErrSignature + } + s := Session{ID: parts[1], Scope: scope, Expires: time.Unix(exp, 0).UTC()} + if now.After(s.Expires) { + return s, ErrExpired + } + return s, nil +} + +// ID returns the session id a sid value names, signed or not: the TID that +// viewer counts and view logs group by. Renewals of one session share it. +func ID(value string) string { + if i := strings.Index(value, "."); i > 0 && strings.HasPrefix(value, version+".") { + rest := value[i+1:] + if j := strings.Index(rest, "."); j > 0 { + return rest[:j] + } + return rest + } + return value +} diff --git a/pkg/psession/psession_test.go b/pkg/psession/psession_test.go new file mode 100644 index 000000000..6275066f2 --- /dev/null +++ b/pkg/psession/psession_test.go @@ -0,0 +1,56 @@ +package psession + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestSessionRoundTrip(t *testing.T) { + key := []byte("0123456789abcdef0123456789abcdef") + now := time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC) + s, sid := Mint(key, "did:plc:streamer", ScopePublic, time.Hour, now) + require.Equal(t, s.ID, ID(sid)) + require.Len(t, s.ID, 13, "a TID") + + got, err := Parse(key, sid, "did:plc:streamer", now.Add(30*time.Minute)) + require.NoError(t, err) + require.Equal(t, s, got) + + // Another stream, another key, a tampered scope: all refused. + _, err = Parse(key, sid, "did:plc:other", now) + require.ErrorIs(t, err, ErrSignature) + _, err = Parse([]byte("another key another key another"), sid, "did:plc:streamer", now) + require.ErrorIs(t, err, ErrSignature) + tampered := s.ID + ".o." + sid[len(s.ID)+5:] + _, err = Parse(key, "1."+tampered, "did:plc:streamer", now) + require.ErrorIs(t, err, ErrSignature) + + // Expired: reported as such, with the session so it can be renewed. + late, err := Parse(key, sid, "did:plc:streamer", now.Add(2*time.Hour)) + require.ErrorIs(t, err, ErrExpired) + require.Equal(t, s.ID, late.ID) + renewed, sid2 := Renew(key, late, "did:plc:streamer", time.Hour, now.Add(2*time.Hour)) + require.Equal(t, s.ID, renewed.ID, "renewal keeps the id") + require.NotEqual(t, sid, sid2) + _, err = Parse(key, sid2, "did:plc:streamer", now.Add(2*time.Hour+time.Minute)) + require.NoError(t, err) + require.Equal(t, s.ID, ID(sid2)) + + // Garbage and legacy bare TIDs are malformed, not signed. + for _, bad := range []string{"", "nope", s.ID, "2." + sid[2:], "1.x.p.1.sig"} { + _, err := Parse(key, bad, "did:plc:streamer", now) + require.ErrorIs(t, err, ErrMalformed, bad) + } + require.Equal(t, s.ID, ID(s.ID), "a bare id is its own id") +} + +func TestOwnerScope(t *testing.T) { + key := []byte("0123456789abcdef0123456789abcdef") + now := time.Now() + _, sid := Mint(key, "did:plc:streamer", ScopeOwner, time.Hour, now) + s, err := Parse(key, sid, "did:plc:streamer", now) + require.NoError(t, err) + require.Equal(t, ScopeOwner, s.Scope) +} diff --git a/pkg/spxrpc/live_token.go b/pkg/spxrpc/live_token.go deleted file mode 100644 index b556bc031..000000000 --- a/pkg/spxrpc/live_token.go +++ /dev/null @@ -1,81 +0,0 @@ -package spxrpc - -import ( - "context" - "crypto/hmac" - "crypto/sha256" - "encoding/base64" - "fmt" - "strconv" - "strings" - "time" -) - -// Pre-live HLS playback tokens. HLS requests carry no session, so a streamer -// previewing their own unpublished stream over HLS proves who they are with -// a token minted through the authenticated getLiveToken query: the DID and -// an expiry, HMAC-signed with a key shared through statedb (so -// every node in a station honours it). - -const liveTokenTTL = time.Hour - -// liveTokenKey is the station's token key, read from statedb once: token -// checks happen on every pre-live playlist and segment request, and the key -// never changes once it exists. -func (s *Server) liveTokenKey(ctx context.Context) ([]byte, error) { - if key := s.liveTokenKeyCache.Load(); key != nil { - return *key, nil - } - if s.statefulDB == nil { - return nil, fmt.Errorf("no statedb") - } - key, err := s.statefulDB.EnsureLiveTokenKey(ctx) - if err != nil { - return nil, err - } - s.liveTokenKeyCache.Store(&key) - return key, nil -} - -func liveTokenSig(key []byte, did string, exp int64) string { - mac := hmac.New(sha256.New, key) - mac.Write([]byte(did + "\x00" + strconv.FormatInt(exp, 10))) - return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)) -} - -// mintLiveToken returns a token for did valid until the returned time. -func mintLiveToken(key []byte, did string, now time.Time) (string, time.Time) { - exp := now.Add(liveTokenTTL).Unix() - tok := "v1." + base64.RawURLEncoding.EncodeToString([]byte(did)) + "." + strconv.FormatInt(exp, 10) + "." + liveTokenSig(key, did, exp) - return tok, time.Unix(exp, 0) -} - -// verifyLiveToken reports whether token was minted for did and is unexpired. -func verifyLiveToken(key []byte, token, did string, now time.Time) bool { - parts := strings.Split(token, ".") - if len(parts) != 4 || parts[0] != "v1" { - return false - } - rawDID, err := base64.RawURLEncoding.DecodeString(parts[1]) - if err != nil || string(rawDID) != did { - return false - } - exp, err := strconv.ParseInt(parts[2], 10, 64) - if err != nil || now.Unix() > exp { - return false - } - return hmac.Equal([]byte(parts[3]), []byte(liveTokenSig(key, did, exp))) -} - -// liveTokenAllows reports whether token lets the caller watch did's -// unpublished stream. -func (s *Server) liveTokenAllows(ctx context.Context, token, did string) bool { - if token == "" { - return false - } - key, err := s.liveTokenKey(ctx) - if err != nil { - return false - } - return verifyLiveToken(key, token, did, time.Now()) -} diff --git a/pkg/spxrpc/live_token_test.go b/pkg/spxrpc/live_token_test.go deleted file mode 100644 index 895348949..000000000 --- a/pkg/spxrpc/live_token_test.go +++ /dev/null @@ -1,22 +0,0 @@ -package spxrpc - -import ( - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -func TestLiveToken(t *testing.T) { - key := []byte("0123456789abcdef0123456789abcdef") - now := time.Date(2026, 9, 14, 12, 0, 0, 0, time.UTC) - tok, exp := mintLiveToken(key, "did:plc:abc", now) - require.Equal(t, now.Add(liveTokenTTL).Unix(), exp.Unix()) - require.True(t, verifyLiveToken(key, tok, "did:plc:abc", now)) - require.True(t, verifyLiveToken(key, tok, "did:plc:abc", now.Add(59*time.Minute))) - require.False(t, verifyLiveToken(key, tok, "did:plc:abc", now.Add(61*time.Minute)), "expired") - require.False(t, verifyLiveToken(key, tok, "did:plc:other", now), "bound to the DID") - require.False(t, verifyLiveToken([]byte("another key............................"), tok, "did:plc:abc", now), "wrong key") - require.False(t, verifyLiveToken(key, tok+"x", "did:plc:abc", now), "tampered") - require.False(t, verifyLiveToken(key, "", "did:plc:abc", now)) -} diff --git a/pkg/spxrpc/place_stream_playback_getlive.go b/pkg/spxrpc/place_stream_playback_getlive.go index 00f543f85..fa06099db 100644 --- a/pkg/spxrpc/place_stream_playback_getlive.go +++ b/pkg/spxrpc/place_stream_playback_getlive.go @@ -3,13 +3,12 @@ package spxrpc import ( "bytes" "context" - "github.com/streamplace/oatproxy/pkg/oatproxy" "io" "net/http" "net/url" "strconv" "stream.place/streamplace/pkg/cdn" - "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/psession" "strings" "time" @@ -23,11 +22,11 @@ import ( // getLiveSegment, vnd.apple.mpegurl on getLivePlaylist). NewServer registers // custom echo routes that override them; these exist only to satisfy the build. -func (s *Server) handlePlaceStreamPlaybackGetLivePlaylist(ctx context.Context, sid string, streamer string, token string, track string) (io.Reader, error) { +func (s *Server) handlePlaceStreamPlaybackGetLivePlaylist(ctx context.Context, sid string, streamer string, track string) (io.Reader, error) { return nil, stubMisrouted("getLivePlaylist") } -func (s *Server) handlePlaceStreamPlaybackGetLiveSegment(ctx context.Context, seg string, sid string, streamer string, token string, track string) (io.Reader, error) { +func (s *Server) handlePlaceStreamPlaybackGetLiveSegment(ctx context.Context, seg string, sid string, streamer string, track string) (io.Reader, error) { return nil, stubMisrouted("getLiveSegment") } @@ -85,30 +84,34 @@ func (s *Server) HandleGetLivePlaylist(c echo.Context) error { if nocdn { live = liveCDN{} } - // A pre-live (unpublished) window is the streamer's own preview: only a - // playback token they minted opens it, and it reads as not live to - // everyone else. The token rides along on every URL the playlist emits, - // all of them self-hosted — a CDN must not see or cache a preview. - token := c.QueryParam("token") - if s.mm.LiveWindowPublished(did) { - token = "" - } else if s.cli.WideOpen || s.liveTokenAllows(ctx, token, did) { - live = liveCDN{} - } else { - return echo.NewHTTPError(http.StatusNotFound, "StreamNotLive") - } - // Reuse the caller's sid when present so master/media/segment requests of - // one playback session share an identifier; mint one otherwise. - sid, err := sessionIDOrNew(c.QueryParam("sid")) - if err != nil { - return echo.NewHTTPError(http.StatusBadRequest, err.Error()) - } - // Sub-playlist + segment URLs carry the resolved DID, so follow-up requests // skip handle resolution and stay stable across a session. track := c.QueryParam("track") rendition := c.QueryParam("rendition") + // The viewer's playback session: verified against this stream, renewed + // while they keep watching, minted for a first request. A media + // playlist is re-fetched every few seconds, so one that arrives without + // a session is redirected to carry one rather than counted afresh on + // every poll. A pre-live (unpublished) window is the streamer's own + // preview: only their own session (from getPlaybackSession) opens it, + // and it reads as not live to everyone else; its URLs stay self-hosted, + // a CDN must not see or cache a preview. + ps, err := s.resolveSession(ctx, c.QueryParam("sid"), did, track != "" || rendition == "audio") + if err != nil { + return err + } + if !s.mm.LiveWindowPublished(did) { + if !s.cli.WideOpen && ps.Scope != psession.ScopeOwner { + return echo.NewHTTPError(http.StatusNotFound, "StreamNotLive") + } + live = liveCDN{} + } + if ps.Redirect { + return redirectWithSession(c, ps.SID) + } + sid := ps.SID + // rendition=audio requests the primary audio track's media playlist // directly, skipping the master playlist so the player never loads video. if track == "" && rendition == "audio" { @@ -121,15 +124,15 @@ func (s *Server) HandleGetLivePlaylist(c echo.Context) error { var body string if track == "" { body = w.MasterPlaylist(func(tid string) string { - return liveTrackPlaylistURL(did, tid, sid, nocdn, token) + return liveTrackPlaylistURL(did, tid, sid, nocdn) }) } else { // The init always comes from the node: it can change mid-stream and // is one small fetch per session. Numbered segments go to the CDN // when one is configured. - initURL := liveSegmentURL(did, track, "init", sid, token) + initURL := liveSegmentURL(did, track, "init", sid) segURI := func(seq uint64) string { - return live.segmentURL(did, track, strconv.FormatUint(seq, 10), sid, token) + return live.segmentURL(did, track, strconv.FormatUint(seq, 10), sid) } body = w.MediaPlaylist(track, initURL, segURI) if body == "" { @@ -139,7 +142,7 @@ func (s *Server) HandleGetLivePlaylist(c echo.Context) error { // so this is the session's heartbeat into the viewer count. Master // requests don't count — one-shot fetchers (preview cards, health // checks) aren't viewers. - s.mm.TouchHLSSession(did, sid) + s.mm.TouchHLSSession(did, ps.ID) } h := c.Response().Header() @@ -158,7 +161,7 @@ func (s *Server) HandleGetLivePlaylist(c echo.Context) error { // The bytes are the verbatim signed segment, so provenance travels with // playback. func (s *Server) HandleGetLiveSegment(c echo.Context) error { - return s.serveLiveSegment(c, c.QueryParam("streamer"), c.QueryParam("track"), c.QueryParam("seg"), c.QueryParam("sid"), c.QueryParam("token")) + return s.serveLiveSegment(c, c.QueryParam("streamer"), c.QueryParam("track"), c.QueryParam("seg"), c.QueryParam("sid")) } // liveSegmentPathRoute is the path-shaped form of getLiveSegment, @@ -172,11 +175,11 @@ const liveSegmentPathRoute = "/live/:streamer/:track/:seg" // HandleGetLiveSegmentPath serves liveSegmentPathRoute. func (s *Server) HandleGetLiveSegmentPath(c echo.Context) error { - return s.serveLiveSegment(c, c.Param("streamer"), c.Param("track"), c.Param("seg"), c.QueryParam("sid"), c.QueryParam("token")) + return s.serveLiveSegment(c, c.Param("streamer"), c.Param("track"), c.Param("seg"), c.QueryParam("sid")) } // serveLiveSegment is the body of both segment routes. -func (s *Server) serveLiveSegment(c echo.Context, streamer, track, segParam, sid, token string) error { +func (s *Server) serveLiveSegment(c echo.Context, streamer, track, segParam, sid string) error { ctx := c.Request().Context() did, err := s.resolveStreamer(ctx, streamer) if err != nil { @@ -195,8 +198,17 @@ func (s *Server) serveLiveSegment(c echo.Context, streamer, track, segParam, sid if w == nil { return echo.NewHTTPError(http.StatusNotFound, "StreamNotLive") } - // Pre-live segments only with the streamer's own playback token. - if !s.mm.LiveWindowPublished(did) && !s.cli.WideOpen && !s.liveTokenAllows(ctx, token, did) { + // A session on the URL must be this stream's (the playlist handler put + // it there); CDN-served segment URLs carry none, the CDN's own signing + // covers those. Pre-live segments open only to the streamer's own + // session. + var sess psession.Session + if sid != "" { + if sess, err = s.verifySession(ctx, sid, did); err != nil { + return err + } + } + if !s.mm.LiveWindowPublished(did) && !s.cli.WideOpen && sess.Scope != psession.ScopeOwner { return echo.NewHTTPError(http.StatusNotFound, "StreamNotLive") } @@ -204,7 +216,7 @@ func (s *Server) serveLiveSegment(c echo.Context, streamer, track, segParam, sid // (sid is threaded through every self-hosted segment URL by the // playlist handler; CDN-served segments carry none, and the playlist // requests that keep coming to the node are the heartbeat then). - s.mm.TouchHLSSession(did, sid) + s.mm.TouchHLSSession(did, sess.ID) var data []byte isInit := seg == "init" @@ -243,17 +255,11 @@ func (s *Server) serveLiveSegment(c echo.Context, streamer, track, segParam, sid // liveTrackPlaylistURL is the URL to a single-track live media playlist served // by this same handler. did is the resolved streamer DID; sid is propagated so // a player's playlist + segment requests share an identifier. -func liveTrackPlaylistURL(did, track, sid string, nocdn bool, token string) string { - q := url.Values{"streamer": {did}, "track": {track}} - if sid != "" { - q.Set("sid", sid) - } +func liveTrackPlaylistURL(did, track, sid string, nocdn bool) string { + q := withSID(url.Values{"streamer": {did}, "track": {track}}, sid) if nocdn { q.Set("nocdn", "1") } - if token != "" { - q.Set("token", token) - } return "/xrpc/place.stream.playback.getLivePlaylist?" + q.Encode() } @@ -295,9 +301,9 @@ func (l liveCDN) tokenExpiry() time.Time { // the init, which never goes through the CDN — see HandleGetLivePlaylist). // Self-hosted, it is liveSegmentURL; behind a CDN it is the path-shaped // route under the CDN host, signed when the provider signs. -func (l liveCDN) segmentURL(did, track, seg, sid, token string) string { - if l.URL == "" || seg == "init" || token != "" { - return liveSegmentURL(did, track, seg, sid, token) +func (l liveCDN) segmentURL(did, track, seg, sid string) string { + if l.URL == "" || seg == "init" { + return liveSegmentURL(did, track, seg, sid) } base, err := url.Parse(l.URL) if err != nil || base.Host == "" { @@ -323,29 +329,8 @@ func liveSegmentPath(did, track, seg string) string { // sequence number; the trailing ".m4s" is appended last (so seg is the final // query token) to satisfy ffmpeg's segment-extension allowlist — the handler // strips it back off. -func liveSegmentURL(did, track, seg, sid, token string) string { - q := url.Values{"streamer": {did}, "track": {track}} - if sid != "" { - q.Set("sid", sid) - } - if token != "" { - q.Set("token", token) - } +func liveSegmentURL(did, track, seg, sid string) string { + q := withSID(url.Values{"streamer": {did}, "track": {track}}, sid) return "/xrpc/place.stream.playback.getLiveSegment?" + q.Encode() + "&seg=" + url.QueryEscape(seg) + ".m4s" } - -// handlePlaceStreamPlaybackGetLiveToken mints a pre-live playback token for -// the caller's own stream (see live_token.go). -func (s *Server) handlePlaceStreamPlaybackGetLiveToken(ctx context.Context) (*placestream.PlaybackGetLiveToken_Output, error) { - session, _ := oatproxy.GetOAuthSession(ctx) - if session == nil { - return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") - } - key, err := s.liveTokenKey(ctx) - if err != nil { - return nil, err - } - tok, exp := mintLiveToken(key, session.DID, time.Now()) - return &placestream.PlaybackGetLiveToken_Output{Token: tok, ExpiresAt: exp.UTC().Format(time.RFC3339)}, nil -} diff --git a/pkg/spxrpc/place_stream_playback_getlive_test.go b/pkg/spxrpc/place_stream_playback_getlive_test.go index 31c22553d..d5a30ce3d 100644 --- a/pkg/spxrpc/place_stream_playback_getlive_test.go +++ b/pkg/spxrpc/place_stream_playback_getlive_test.go @@ -1,6 +1,7 @@ package spxrpc import ( + "context" "net/http" "net/http/httptest" "net/url" @@ -11,6 +12,7 @@ import ( "github.com/labstack/echo/v4" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/cdn/bunny" + "stream.place/streamplace/pkg/psession" ) // TestLiveSegmentURL pins the ffmpeg-compatibility contract: the segment URL @@ -18,7 +20,7 @@ import ( // its allowlist before fetching), and the handler must be able to recover the // did/track/seg from it. func TestLiveSegmentURL(t *testing.T) { - u := liveSegmentURL("did:plc:abc123", "1", "42", "3kabc", "") + u := liveSegmentURL("did:plc:abc123", "1", "42", "3kabc") require.True(t, strings.HasSuffix(u, ".m4s"), "segment URL must end in .m4s, got %q", u) require.True(t, strings.HasPrefix(u, "/xrpc/place.stream.playback.getLiveSegment?")) require.Contains(t, u, "track=1") @@ -28,7 +30,7 @@ func TestLiveSegmentURL(t *testing.T) { require.Contains(t, u, "streamer=did%3Aplc%3Aabc123") // init segment URL (no sid). - ui := liveSegmentURL("did:plc:abc123", "2", "init", "", "") + ui := liveSegmentURL("did:plc:abc123", "2", "init", "") require.True(t, strings.HasSuffix(ui, "seg=init.m4s"), "init URL must end in seg=init.m4s, got %q", ui) require.NotContains(t, ui, "sid=") } @@ -36,14 +38,14 @@ func TestLiveSegmentURL(t *testing.T) { // TestLiveTrackPlaylistURL confirms the master playlist's per-track sub-URLs // carry did + track + sid back to getLivePlaylist. func TestLiveTrackPlaylistURL(t *testing.T) { - u := liveTrackPlaylistURL("did:plc:xyz", "1", "3ksid", false, "") + u := liveTrackPlaylistURL("did:plc:xyz", "1", "3ksid", false) require.True(t, strings.HasPrefix(u, "/xrpc/place.stream.playback.getLivePlaylist?")) require.Contains(t, u, "track=1") require.Contains(t, u, "sid=3ksid") require.Contains(t, u, "streamer=did%3Aplc%3Axyz") require.NotContains(t, u, "nocdn") // The escape hatch follows the master into its sub-playlists. - require.Contains(t, liveTrackPlaylistURL("did:plc:xyz", "1", "3ksid", true, ""), "nocdn=1") + require.Contains(t, liveTrackPlaylistURL("did:plc:xyz", "1", "3ksid", true), "nocdn=1") } // TestLiveCDNSegmentURL pins the CDN contract: numbered segments move to the @@ -51,19 +53,19 @@ func TestLiveTrackPlaylistURL(t *testing.T) { // init stays on the node, and a zero liveCDN is exactly the self-hosted form. func TestLiveCDNSegmentURL(t *testing.T) { var self liveCDN - require.Equal(t, liveSegmentURL("did:plc:abc", "1", "42", "sid1", ""), self.segmentURL("did:plc:abc", "1", "42", "sid1", "")) + require.Equal(t, liveSegmentURL("did:plc:abc", "1", "42", "sid1"), self.segmentURL("did:plc:abc", "1", "42", "sid1")) l := liveCDN{URL: "https://live.b-cdn.net", TTL: time.Minute} - u := l.segmentURL("did:plc:abc", "1", "42", "sid1", "") + u := l.segmentURL("did:plc:abc", "1", "42", "sid1") require.Equal(t, "https://live.b-cdn.net/live/did:plc:abc/1/42.m4s", u) require.NotContains(t, u, "sid", "CDN URLs carry no session so every viewer shares the cache entry") require.True(t, strings.HasSuffix(u, ".m4s"), "ffmpeg's extension allowlist") // The init is one small per-session fetch and can change mid-stream: // it stays on the node, session and all. - require.Equal(t, liveSegmentURL("did:plc:abc", "1", "init", "sid1", ""), l.segmentURL("did:plc:abc", "1", "init", "sid1", "")) + require.Equal(t, liveSegmentURL("did:plc:abc", "1", "init", "sid1"), l.segmentURL("did:plc:abc", "1", "init", "sid1")) // A base URL with a path prefix keeps it. require.Equal(t, "https://cdn.example.com/edge/live/did:plc:abc/1/42.m4s", - liveCDN{URL: "https://cdn.example.com/edge/"}.segmentURL("did:plc:abc", "1", "42", "", "")) + liveCDN{URL: "https://cdn.example.com/edge/"}.segmentURL("did:plc:abc", "1", "42", "")) // The route pattern and the emitted path agree. require.Equal(t, "/live/:streamer/:track/:seg", liveSegmentPathRoute) pu, err := url.Parse(u) @@ -80,15 +82,15 @@ func TestLiveCDNSignedURLStableWithinBucket(t *testing.T) { base := time.Date(2026, 9, 10, 12, 0, 7, 0, time.UTC) clock := base l := liveCDN{URL: "https://live.b-cdn.net", Signer: bunny.Signer{Key: "k"}, TTL: time.Minute, now: func() time.Time { return clock }} - u1 := l.segmentURL("did:plc:abc", "1", "42", "", "") + u1 := l.segmentURL("did:plc:abc", "1", "42", "") clock = base.Add(40 * time.Second) - u2 := l.segmentURL("did:plc:abc", "1", "42", "", "") + u2 := l.segmentURL("did:plc:abc", "1", "42", "") require.Equal(t, u1, u2, "same bucket, same URL") require.Contains(t, u1, "token=") require.Contains(t, u1, "expires=") require.NotContains(t, u1, "sid=") clock = base.Add(61 * time.Second) - u3 := l.segmentURL("did:plc:abc", "1", "42", "", "") + u3 := l.segmentURL("did:plc:abc", "1", "42", "") require.NotEqual(t, u1, u3, "next bucket, fresh token") // expiry = end of the bucket after the current one. exp := l.tokenExpiry() @@ -107,7 +109,7 @@ func TestLiveSegmentPathRouteParams(t *testing.T) { got = []string{c.Param("streamer"), c.Param("track"), c.Param("seg")} return c.NoContent(http.StatusOK) }) - u := liveCDN{URL: "https://live.b-cdn.net"}.segmentURL("did:plc:abc", "1", "42", "", "") + u := liveCDN{URL: "https://live.b-cdn.net"}.segmentURL("did:plc:abc", "1", "42", "") pu, err := url.Parse(u) require.NoError(t, err) rec := httptest.NewRecorder() @@ -117,12 +119,79 @@ func TestLiveSegmentPathRouteParams(t *testing.T) { require.Equal(t, "42", strings.TrimSuffix(got[2], ".m4s")) } -// A pre-live token keeps every URL on the node: the CDN must not see (or -// cache) a streamer's private preview. -func TestLiveTokenKeepsURLsSelfHosted(t *testing.T) { - l := liveCDN{URL: "https://live.b-cdn.net", TTL: time.Minute} - u := l.segmentURL("did:plc:abc", "1", "42", "", "tok") - require.True(t, strings.HasPrefix(u, "/xrpc/place.stream.playback.getLiveSegment?")) - require.Contains(t, u, "token=tok") - require.Contains(t, liveTrackPlaylistURL("did:plc:abc", "1", "", false, "tok"), "token=tok") +// TestResolveSession pins the session contract on playback requests: a +// first request gets a fresh public session (and a media playlist is +// redirected to carry it), a valid one is kept, one past half its life or +// expired is renewed under the same id, and a forged one is refused. +func TestResolveSession(t *testing.T) { + key := []byte("0123456789abcdef0123456789abcdef") + s := &Server{} + s.sessionKeyCache.Store(&key) + ctx := context.Background() + const streamer = "did:plc:abc" + + fresh, err := s.resolveSession(ctx, "", streamer, true) + require.NoError(t, err) + require.True(t, fresh.Redirect, "a media playlist without a session is redirected to one") + require.Equal(t, psession.ScopePublic, fresh.Scope) + require.Equal(t, fresh.ID, psession.ID(fresh.SID)) + + master, err := s.resolveSession(ctx, "", streamer, false) + require.NoError(t, err) + require.False(t, master.Redirect, "a master playlist embeds the session it mints") + + kept, err := s.resolveSession(ctx, fresh.SID, streamer, true) + require.NoError(t, err) + require.False(t, kept.Redirect) + require.Equal(t, fresh.SID, kept.SID, "a young session is emitted as is") + + _, err = s.resolveSession(ctx, fresh.SID, "did:plc:other", true) + require.ErrorIs(t, err, errBadSession, "bound to the stream it was minted for") + _, err = s.resolveSession(ctx, "3kabc", streamer, true) + require.ErrorIs(t, err, errBadSession, "a bare id is not a session") + + // Expired: renewed under the same id; a media playlist is redirected to + // the renewal so the player's follow-ups carry it. + old, _ := psession.Mint(key, streamer, psession.ScopePublic, -time.Hour, time.Now()) + _, oldSID := psession.Renew(key, old, streamer, -time.Hour, time.Now()) + renewed, err := s.resolveSession(ctx, oldSID, streamer, true) + require.NoError(t, err) + require.True(t, renewed.Redirect) + require.Equal(t, old.ID, renewed.ID) + require.NotEqual(t, oldSID, renewed.SID) + + // Segments never mint: a URL out of a playlist carries a session or is + // refused. + sess, err := s.verifySession(ctx, renewed.SID, streamer) + require.NoError(t, err) + require.Equal(t, old.ID, sess.ID) + _, err = s.verifySession(ctx, "", streamer) + require.Error(t, err) + _, err = s.verifySession(ctx, oldSID, streamer) + require.Error(t, err, "expired") + + // The owner session: minted by getPlaybackSession, it is what opens a + // pre-live stream; a public one does not (the handler checks Scope). + _, ownerSID := psession.Mint(key, streamer, psession.ScopeOwner, ownerSessionTTL, time.Now()) + owner, err := s.resolveSession(ctx, ownerSID, streamer, true) + require.NoError(t, err) + require.Equal(t, psession.ScopeOwner, owner.Scope) +} + +// TestRedirectWithSession: the redirect keeps the request's URL and adds the +// session, and is never cached. +func TestRedirectWithSession(t *testing.T) { + e := echo.New() + req := httptest.NewRequest(http.MethodGet, "/xrpc/place.stream.playback.getLivePlaylist?streamer=did%3Aplc%3Aabc&track=1", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + require.NoError(t, redirectWithSession(c, "1.3kabc.p.1.sig")) + require.Equal(t, http.StatusFound, rec.Code) + loc, err := url.Parse(rec.Header().Get("Location")) + require.NoError(t, err) + require.Equal(t, "/xrpc/place.stream.playback.getLivePlaylist", loc.Path) + require.Equal(t, "did:plc:abc", loc.Query().Get("streamer")) + require.Equal(t, "1", loc.Query().Get("track")) + require.Equal(t, "1.3kabc.p.1.sig", loc.Query().Get("sid")) + require.Equal(t, "no-store", rec.Header().Get("Cache-Control")) } diff --git a/pkg/spxrpc/place_stream_playback_getvideo.go b/pkg/spxrpc/place_stream_playback_getvideo.go index 8627fee83..26c2cb5f1 100644 --- a/pkg/spxrpc/place_stream_playback_getvideo.go +++ b/pkg/spxrpc/place_stream_playback_getvideo.go @@ -20,23 +20,10 @@ import ( "stream.place/streamplace/pkg/cdn" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/placestream" - "stream.place/streamplace/pkg/spid" + "stream.place/streamplace/pkg/psession" "stream.place/streamplace/pkg/vod" ) -// sessionIDOrNew returns the caller's sid if it's well-formed, otherwise -// gives you a fresh tid -func sessionIDOrNew(supplied string) (string, error) { - if supplied != "" { - _, err := syntax.ParseTID(supplied) - if err != nil { - return "", fmt.Errorf("invalid session ID: %w", err) - } - return supplied, nil - } - return spid.TID(), nil -} - // videoCollection is the only collection getVideoPlaylist currently // knows how to dispatch on. The lexicon accepts any AT-URI so we have // room to grow into other playable record types without a wire break. @@ -139,7 +126,7 @@ func (s *Server) HandleGetVideoBlob(c echo.Context) error { } defer r.Close() rangeStart, rangeEnd := parseRangeForLog(c.Request().Header.Get("Range"), r.Size()) - s.logSegmentRequest(c, cid, did, c.QueryParam("sid"), rangeStart, rangeEnd) + s.logSegmentRequest(c, cid, did, psession.ID(c.QueryParam("sid")), rangeStart, rangeEnd) return serveBlobRange(c, r, "video/mp4") } @@ -290,9 +277,11 @@ func (s *Server) HandleGetVideoPlaylist(c echo.Context) error { // URL we previously handed it) so the master/media/segment requests // of one playback session share an identifier. Generate a fresh one // when the player is making its first request. - sid, err := sessionIDOrNew(c.QueryParam("sid")) - if err != nil { - return echo.NewHTTPError(http.StatusBadRequest, err.Error()) + // A sid that is not even the shape of a session is refused up front, + // before the lookups below; the signature is checked once the video is + // known to be servable. + if sid := c.QueryParam("sid"); sid != "" && !psession.WellFormed(sid) { + return errBadSession } uri := aturi.String() @@ -312,6 +301,19 @@ func (s *Server) HandleGetVideoPlaylist(c echo.Context) error { return echo.NewHTTPError(http.StatusForbidden, "VideoUnavailable") } + // The viewer's playback session, once the video is known to be + // servable: verified against the owner, renewed while they watch, + // minted for a first request (a media playlist without one is + // redirected to carry it, so the player's follow-ups share a session). + ps, err := s.resolveSession(ctx, c.QueryParam("sid"), aturi.Authority().String(), track != "") + if err != nil { + return err + } + if ps.Redirect { + return redirectWithSession(c, ps.SID) + } + sid := ps.SID + resolved, err := s.resolveVideoBlob(ctx, uri) if err != nil { return err @@ -343,7 +345,7 @@ func (s *Server) HandleGetVideoPlaylist(c echo.Context) error { } kind = "media" } - s.logManifestRequest(c, uri, sid, track, kind) + s.logManifestRequest(c, uri, ps.ID, track, kind) c.Response().Header().Set("Content-Type", "application/vnd.apple.mpegurl") c.Response().Header().Set("Cache-Control", "public, max-age=60") diff --git a/pkg/spxrpc/place_stream_playback_getvideo_test.go b/pkg/spxrpc/place_stream_playback_getvideo_test.go index 10ccf33d2..cbd5046ef 100644 --- a/pkg/spxrpc/place_stream_playback_getvideo_test.go +++ b/pkg/spxrpc/place_stream_playback_getvideo_test.go @@ -19,7 +19,6 @@ import ( "stream.place/streamplace/pkg/cdn/bunny" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/placestream" - "stream.place/streamplace/pkg/spid" "stream.place/streamplace/pkg/vod" ) @@ -316,48 +315,6 @@ func TestTicksToDuration(t *testing.T) { require.Equal(t, 29*time.Hour, ticksToDuration(29*3600*90000, 90000)) } -func TestSessionIDOrNew(t *testing.T) { - t.Run("supplied is reused", func(t *testing.T) { - // A well-formed TID like the player would have gotten from a - // prior master-playlist response. - valid := spid.TID() - got, err := sessionIDOrNew(valid) - require.NoError(t, err) - require.Equal(t, valid, got) - }) - t.Run("empty mints a fresh tid", func(t *testing.T) { - got, err := sessionIDOrNew("") - require.NoError(t, err) - // The fresh sid is itself a valid TID — players that round-trip - // it through ParseTID won't reject our output. - _, err = syntax.ParseTID(got) - require.NoError(t, err) - // TIDs embed a timestamp + random clock id, so back-to-back - // calls never collide. - got2, err := sessionIDOrNew("") - require.NoError(t, err) - require.NotEqual(t, got, got2) - }) - t.Run("invalid is rejected", func(t *testing.T) { - for _, bad := range []string{ - "has space", // whitespace - "has/slash", // URL-illegal char - "has?question", // ditto - "tooshort", // < 13 chars - "toolongtoolong", // > 13 chars - "1111111111111", // 13 chars but '1' isn't in the TID alphabet - "zabcdefghijkl", // 13 chars but 'z' isn't a valid first char - } { - _, err := sessionIDOrNew(bad) - require.Error(t, err, "expected %q to be rejected", bad) - } - }) -} - -// TestResolveVideoBlob_SourceClip walks the playback-resolve path for -// a clip record: the clip's `Video` field points at a parent video -// whose sourceTracks lead to a real MediaTrack + blob CID. resolve -// should return the parent's blob CID and surface the clip's bounds. func TestResolveVideoBlob_SourceClip(t *testing.T) { ctx := context.Background() m, err := model.MakeDB(":memory:") @@ -584,9 +541,9 @@ func TestParseSingleRange(t *testing.T) { } // TestHandleGetVideoPlaylist_RejectsBadSID exercises the handler-side -// validation. Validation runs before any DB / blob lookups, so the -// test doesn't need a populated model or playbackStore content — just -// the playbackStore presence check the handler does up front. +// validation: a sid that is not even the shape of a signed session is +// refused before any key, DB or blob lookup, so the test needs none of +// them — just the playbackStore presence check the handler does up front. func TestHandleGetVideoPlaylist_RejectsBadSID(t *testing.T) { store, err := blob.NewFileStore(t.TempDir()) require.NoError(t, err) @@ -600,6 +557,8 @@ func TestHandleGetVideoPlaylist_RejectsBadSID(t *testing.T) { {"contains slash", "has/slash"}, {"too long", strings.Repeat("x", 65)}, {"wrong tid alphabet", "1111111111111"}, + {"a bare session id, unsigned", "3kabcdefghijk"}, + {"unknown scope", "1.3kabcdefghijk.x.1.sig"}, } { t.Run(tc.name, func(t *testing.T) { req := httptest.NewRequest(http.MethodGet, diff --git a/pkg/spxrpc/playback_session.go b/pkg/spxrpc/playback_session.go new file mode 100644 index 000000000..10837c438 --- /dev/null +++ b/pkg/spxrpc/playback_session.go @@ -0,0 +1,158 @@ +package spxrpc + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" + "time" + + "github.com/labstack/echo/v4" + "github.com/streamplace/oatproxy/pkg/oatproxy" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/psession" +) + +// Playback sessions (pkg/psession) on the playback handlers. Every playlist +// and segment URL a handler emits carries the viewer's signed session as sid; +// the handlers verify it against the stream it was minted for, renew it while +// the viewer keeps watching, and hand a viewer without one a fresh one. + +const ( + // publicSessionTTL is how long a viewer's session lasts without being + // renewed. A live player refreshes its playlist every few seconds and + // every playlist rendered renews a session past half its life, so an + // attentive viewer's session never lapses; a URL copied out of a + // playlist stops working after this. + publicSessionTTL = 24 * time.Hour + // ownerSessionTTL is the streamer's own session (pre-live preview): + // minted through an authenticated call, renewed the same way. + ownerSessionTTL = time.Hour +) + +// sessionKey is the station's signing key, read from statedb once: session +// checks happen on every playlist and segment request, and the key never +// changes once it exists. +func (s *Server) sessionKey(ctx context.Context) ([]byte, error) { + if key := s.sessionKeyCache.Load(); key != nil { + return *key, nil + } + if s.statefulDB == nil { + return nil, fmt.Errorf("no statedb") + } + key, err := s.statefulDB.EnsurePlaybackSessionKey(ctx) + if err != nil { + return nil, err + } + s.sessionKeyCache.Store(&key) + return key, nil +} + +func sessionTTL(scope psession.Scope) time.Duration { + if scope == psession.ScopeOwner { + return ownerSessionTTL + } + return publicSessionTTL +} + +// playbackSession is what a playback request resolved to: the session to +// count by and the sid to emit in every URL of the response. +type playbackSession struct { + psession.Session + // SID is the value to put in emitted URLs: the caller's, or a renewal + // of it (same ID, fresh expiry), or a fresh session. + SID string + // Redirect is set when the caller sent no usable session for a request + // whose URL the player will keep re-fetching (a media playlist): the + // handler redirects to the same URL carrying SID, so the player's + // follow-ups share one session instead of minting one per poll. + Redirect bool +} + +var errBadSession = echo.NewHTTPError(http.StatusBadRequest, "InvalidSession") + +// resolveSession verifies the sid a playback request carries against +// streamer, or mints one. A malformed or forged sid is refused; an expired +// one is renewed in place, so a viewer who paused for a day resumes with +// their own session id. When needsRedirect is set and the caller had no +// valid session, the result asks for a redirect. +func (s *Server) resolveSession(ctx context.Context, sid, streamer string, needsRedirect bool) (playbackSession, error) { + if sid != "" && !psession.WellFormed(sid) { + return playbackSession{}, errBadSession + } + key, err := s.sessionKey(ctx) + if err != nil { + return playbackSession{}, err + } + now := time.Now() + if sid == "" { + sess, value := psession.Mint(key, streamer, psession.ScopePublic, publicSessionTTL, now) + return playbackSession{Session: sess, SID: value, Redirect: needsRedirect}, nil + } + sess, err := psession.Parse(key, sid, streamer, now) + switch { + case err == nil: + out := playbackSession{Session: sess, SID: sid} + if sess.Expires.Sub(now) < sessionTTL(sess.Scope)/2 { + out.Session, out.SID = psession.Renew(key, sess, streamer, sessionTTL(sess.Scope), now) + } + return out, nil + case errors.Is(err, psession.ErrExpired): + sess, value := psession.Renew(key, sess, streamer, sessionTTL(sess.Scope), now) + return playbackSession{Session: sess, SID: value, Redirect: needsRedirect}, nil + default: + return playbackSession{}, errBadSession + } +} + +// verifySession is resolveSession for a segment request: the URL came out +// of a playlist this node rendered, so it must carry a valid session or it +// is refused, nothing is minted. +func (s *Server) verifySession(ctx context.Context, sid, streamer string) (psession.Session, error) { + key, err := s.sessionKey(ctx) + if err != nil { + return psession.Session{}, err + } + sess, err := psession.Parse(key, sid, streamer, time.Now()) + if err != nil { + return psession.Session{}, echo.NewHTTPError(http.StatusForbidden, "InvalidSession") + } + return sess, nil +} + +// redirectWithSession answers a request that arrived without a usable +// session: the same URL, with the session attached. +func redirectWithSession(c echo.Context, sid string) error { + u := *c.Request().URL + q := u.Query() + q.Set("sid", sid) + u.RawQuery = q.Encode() + c.Response().Header().Set("Cache-Control", "no-store") + return c.Redirect(http.StatusFound, u.RequestURI()) +} + +// withSID adds the session to a set of URL query values. +func withSID(q url.Values, sid string) url.Values { + if sid != "" { + q.Set("sid", sid) + } + return q +} + +// handlePlaceStreamPlaybackGetPlaybackSession mints the caller's own +// session for their stream: it opens their unpublished (pre-live) stream to +// their own player, which the anonymous session a playlist hands out does +// not. Bound to the caller's DID, so it opens nobody else's stream. +func (s *Server) handlePlaceStreamPlaybackGetPlaybackSession(ctx context.Context) (*placestream.PlaybackGetPlaybackSession_Output, error) { + session, _ := oatproxy.GetOAuthSession(ctx) + if session == nil { + return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") + } + key, err := s.sessionKey(ctx) + if err != nil { + return nil, err + } + sess, sid := psession.Mint(key, session.DID, psession.ScopeOwner, ownerSessionTTL, time.Now()) + return &placestream.PlaybackGetPlaybackSession_Output{Sid: sid, ExpiresAt: sess.Expires.Format(time.RFC3339)}, nil +} diff --git a/pkg/spxrpc/spxrpc.go b/pkg/spxrpc/spxrpc.go index 6c1c4834a..770a1f833 100644 --- a/pkg/spxrpc/spxrpc.go +++ b/pkg/spxrpc/spxrpc.go @@ -32,21 +32,21 @@ import ( ) type Server struct { - e *echo.Echo - cli *config.CLI - model model.Model - OGImageCache *cache.Cache - LiveUsersCache *cache.Cache - GameSearchCache *cache.Cache - ScoreCache *cache.Cache - ATSync *atproto.ATProtoSynchronizer - statefulDB *statedb.StatefulDB - liveTokenKeyCache atomic.Pointer[[]byte] - bus *bus.Bus - op *oatproxy.OATProxy - localDB localdb.LocalDB - mm *media.MediaManager - uploadManager *upload.Manager + e *echo.Echo + cli *config.CLI + model model.Model + OGImageCache *cache.Cache + LiveUsersCache *cache.Cache + GameSearchCache *cache.Cache + ScoreCache *cache.Cache + ATSync *atproto.ATProtoSynchronizer + statefulDB *statedb.StatefulDB + sessionKeyCache atomic.Pointer[[]byte] + bus *bus.Bus + op *oatproxy.OATProxy + localDB localdb.LocalDB + mm *media.MediaManager + uploadManager *upload.Manager // playbackStore is where VOD blobs + metafiles + per-track init // segments live. Matches the blob.Store the VOD processor writes // into (vod.BlobsPrefix + .{mp4,json}). diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index 7cff04604..11f8c795e 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -467,8 +467,8 @@ func (s *Server) RegisterHandlersPlacestream(e *echo.Echo) error { e.POST("/xrpc/place.stream.multistream.putTarget", s.HandlePlaceStreamMultistreamPutTarget) e.GET("/xrpc/place.stream.playback.getLivePlaylist", s.HandlePlaceStreamPlaybackGetLivePlaylist) e.GET("/xrpc/place.stream.playback.getLiveSegment", s.HandlePlaceStreamPlaybackGetLiveSegment) - e.GET("/xrpc/place.stream.playback.getLiveToken", s.HandlePlaceStreamPlaybackGetLiveToken) e.GET("/xrpc/place.stream.playback.getPlaybackServer", s.HandlePlaceStreamPlaybackGetPlaybackServer) + e.GET("/xrpc/place.stream.playback.getPlaybackSession", s.HandlePlaceStreamPlaybackGetPlaybackSession) e.GET("/xrpc/place.stream.playback.getVideoBlob", s.HandlePlaceStreamPlaybackGetVideoBlob) e.GET("/xrpc/place.stream.playback.getVideoPlaylist", s.HandlePlaceStreamPlaybackGetVideoPlaylist) e.POST("/xrpc/place.stream.playback.whep", s.HandlePlaceStreamPlaybackWhep) @@ -1284,12 +1284,11 @@ func (s *Server) HandlePlaceStreamPlaybackGetLivePlaylist(c echo.Context) error defer span.End() sid := c.QueryParam("sid") streamer := c.QueryParam("streamer") - token := c.QueryParam("token") track := c.QueryParam("track") var out io.Reader var handleErr error - // func (s *Server) handlePlaceStreamPlaybackGetLivePlaylist(ctx context.Context,sid string,streamer string,token string,track string) (io.Reader, error) - out, handleErr = s.handlePlaceStreamPlaybackGetLivePlaylist(ctx, sid, streamer, token, track) + // func (s *Server) handlePlaceStreamPlaybackGetLivePlaylist(ctx context.Context,sid string,streamer string,track string) (io.Reader, error) + out, handleErr = s.handlePlaceStreamPlaybackGetLivePlaylist(ctx, sid, streamer, track) if handleErr != nil { return handleErr } @@ -1302,39 +1301,38 @@ func (s *Server) HandlePlaceStreamPlaybackGetLiveSegment(c echo.Context) error { seg := c.QueryParam("seg") sid := c.QueryParam("sid") streamer := c.QueryParam("streamer") - token := c.QueryParam("token") track := c.QueryParam("track") var out io.Reader var handleErr error - // func (s *Server) handlePlaceStreamPlaybackGetLiveSegment(ctx context.Context,seg string,sid string,streamer string,token string,track string) (io.Reader, error) - out, handleErr = s.handlePlaceStreamPlaybackGetLiveSegment(ctx, seg, sid, streamer, token, track) + // func (s *Server) handlePlaceStreamPlaybackGetLiveSegment(ctx context.Context,seg string,sid string,streamer string,track string) (io.Reader, error) + out, handleErr = s.handlePlaceStreamPlaybackGetLiveSegment(ctx, seg, sid, streamer, track) if handleErr != nil { return handleErr } return c.Stream(200, "video/mp4", out) } -func (s *Server) HandlePlaceStreamPlaybackGetLiveToken(c echo.Context) error { - ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamPlaybackGetLiveToken") +func (s *Server) HandlePlaceStreamPlaybackGetPlaybackServer(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamPlaybackGetPlaybackServer") defer span.End() - var out *placestream.PlaybackGetLiveToken_Output + stream := c.QueryParam("stream") + var out *placestream.PlaybackGetPlaybackServer_Output var handleErr error - // func (s *Server) handlePlaceStreamPlaybackGetLiveToken(ctx context.Context) (*placestream.PlaybackGetLiveToken_Output, error) - out, handleErr = s.handlePlaceStreamPlaybackGetLiveToken(ctx) + // func (s *Server) handlePlaceStreamPlaybackGetPlaybackServer(ctx context.Context,stream string) (*placestream.PlaybackGetPlaybackServer_Output, error) + out, handleErr = s.handlePlaceStreamPlaybackGetPlaybackServer(ctx, stream) if handleErr != nil { return handleErr } return c.JSON(200, out) } -func (s *Server) HandlePlaceStreamPlaybackGetPlaybackServer(c echo.Context) error { - ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamPlaybackGetPlaybackServer") +func (s *Server) HandlePlaceStreamPlaybackGetPlaybackSession(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamPlaybackGetPlaybackSession") defer span.End() - stream := c.QueryParam("stream") - var out *placestream.PlaybackGetPlaybackServer_Output + var out *placestream.PlaybackGetPlaybackSession_Output var handleErr error - // func (s *Server) handlePlaceStreamPlaybackGetPlaybackServer(ctx context.Context,stream string) (*placestream.PlaybackGetPlaybackServer_Output, error) - out, handleErr = s.handlePlaceStreamPlaybackGetPlaybackServer(ctx, stream) + // func (s *Server) handlePlaceStreamPlaybackGetPlaybackSession(ctx context.Context) (*placestream.PlaybackGetPlaybackSession_Output, error) + out, handleErr = s.handlePlaceStreamPlaybackGetPlaybackSession(ctx) if handleErr != nil { return handleErr } diff --git a/pkg/statedb/live_token_key.go b/pkg/statedb/live_token_key.go deleted file mode 100644 index e0709d95f..000000000 --- a/pkg/statedb/live_token_key.go +++ /dev/null @@ -1,29 +0,0 @@ -package statedb - -import ( - "context" - "crypto/rand" - "fmt" -) - -// EnsureLiveTokenKey returns the HMAC key that signs pre-live playback -// tokens, generating and storing one on first use. Shared through statedb -// so every node in a station honours tokens minted by any other. -func (state *StatefulDB) EnsureLiveTokenKey(ctx context.Context) ([]byte, error) { - const key = "live-token-key" - conf, err := state.GetConfig(key) - if err != nil { - return nil, fmt.Errorf("get live token key: %w", err) - } - if conf != nil && len(conf.Value) >= 32 { - return conf.Value, nil - } - secret := make([]byte, 32) - if _, err := rand.Read(secret); err != nil { - return nil, fmt.Errorf("generate live token key: %w", err) - } - if err := state.PutConfig(key, secret); err != nil { - return nil, fmt.Errorf("store live token key: %w", err) - } - return secret, nil -} diff --git a/pkg/statedb/playback_session_key.go b/pkg/statedb/playback_session_key.go new file mode 100644 index 000000000..79550540a --- /dev/null +++ b/pkg/statedb/playback_session_key.go @@ -0,0 +1,29 @@ +package statedb + +import ( + "context" + "crypto/rand" + "fmt" +) + +// EnsurePlaybackSessionKey returns the HMAC key that signs playback sessions +// (pkg/psession), generating and storing one on first use. Shared through +// statedb so every node in a station honours sessions minted by any other. +func (state *StatefulDB) EnsurePlaybackSessionKey(ctx context.Context) ([]byte, error) { + const key = "playback-session-key" + conf, err := state.GetConfig(key) + if err != nil { + return nil, fmt.Errorf("get playback session key: %w", err) + } + if conf != nil && len(conf.Value) >= 32 { + return conf.Value, nil + } + secret := make([]byte, 32) + if _, err := rand.Read(secret); err != nil { + return nil, fmt.Errorf("generate playback session key: %w", err) + } + if err := state.PutConfig(key, secret); err != nil { + return nil, fmt.Errorf("store playback session key: %w", err) + } + return secret, nil +} diff --git a/pkg/viewlog/ingest.go b/pkg/viewlog/ingest.go index 2538213ac..0f142e4e1 100644 --- a/pkg/viewlog/ingest.go +++ b/pkg/viewlog/ingest.go @@ -16,6 +16,7 @@ import ( "stream.place/streamplace/pkg/blob" "stream.place/streamplace/pkg/cdn" "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/psession" "stream.place/streamplace/pkg/statedb" "stream.place/streamplace/pkg/vod" ) @@ -296,7 +297,7 @@ func segmentEventFromRequest(r cdn.Request, salts *SaltManager) (Event, bool) { return Event{ Ts: ts, Type: EventTypeSegmentRequest, - SID: sid, + SID: psession.ID(sid), IPHash: ipHash, CID: cid, OwnerDID: q.Get("did"),