From 5366fff5bfc73df46a7a27fd777fef9c0f642d22 Mon Sep 17 00:00:00 2001 From: Elliot Shine Date: Tue, 20 Oct 2026 13:19:20 +1030 Subject: [PATCH] oceanmedia: fix delta updates and make URLs stable This change fixes delta updates - namely changing skipped segments to use an equals instead of a colon and always providing a segment even if there isn't a new one ready yet. It also makes URLs stable by using a redirect to a signed URL instead of a signed URL directly. This is important as players need a stable URL to match up segments in delta requests. It also has the benefit in that we don't need to sign every URL which reduces the amount of processing we need to do. --- cmd/oceanmedia/auth.go | 62 ++++++++++---- cmd/oceanmedia/main.go | 4 + cmd/oceanmedia/playlist.go | 98 +++++++++++++++------- cmd/oceanmedia/playlist_test.go | 141 +++++++++++++++++++++----------- 4 files changed, 213 insertions(+), 92 deletions(-) diff --git a/cmd/oceanmedia/auth.go b/cmd/oceanmedia/auth.go index 0b4ab0f4..c35146f1 100644 --- a/cmd/oceanmedia/auth.go +++ b/cmd/oceanmedia/auth.go @@ -27,6 +27,7 @@ package main import ( "errors" "fmt" + "strings" "github.com/ausocean/cloud/gauth" "github.com/gofiber/fiber/v2" @@ -38,34 +39,65 @@ var ( errUnauthorized = errors.New("unauthorized") ) -// authenticate validates the request's JWT and ensures that it authorises -// access to the given broadcast. Tokens are minted by referring services -// (such as OceanBench and AusOceanTV) using the shared JWT secret and -// supplied to players as the "token" query parameter. A Bearer token in -// the Authorization header is also accepted for programmatic clients. -// -// Required claims are "sub" (the broadcast ID) and "exp". -func (s *server) authenticate(c *fiber.Ctx, broadcastID string) error { +// requestToken returns the JWT supplied with the request, either as the +// "token" query parameter or as a Bearer token in the Authorization header. +// The "Bearer " prefix, if present, is stripped. +func requestToken(c *fiber.Ctx) string { token := c.Query("token") if token == "" { token = c.Get(fiber.HeaderAuthorization) } + return strings.TrimPrefix(token, "Bearer ") +} + +// requestBaseURL returns the base URL (scheme://host) the request was made +// to, honouring the X-Forwarded-Proto header set by the App Engine proxy. +func requestBaseURL(c *fiber.Ctx) string { + scheme := c.Protocol() + if p := c.Get("X-Forwarded-Proto"); p != "" { + scheme = p + } + return scheme + "://" + c.Hostname() +} + +// writeAuthError writes an error response for an authentication failure. +func writeAuthError(c *fiber.Ctx, err error) error { + status := fiber.StatusForbidden + if errors.Is(err, errMissingToken) || errors.Is(err, errInvalidToken) { + status = fiber.StatusUnauthorized + } + return c.Status(status).JSON(fiber.Map{"error": err.Error()}) +} + +// tokenClaims validates the request's JWT and returns its claims. Required +// claims are "sub" (the broadcast ID) and "exp". +func (s *server) tokenClaims(c *fiber.Ctx) (map[string]interface{}, error) { + token := requestToken(c) if token == "" { - return errMissingToken + return nil, errMissingToken } claims, err := gauth.GetClaims(token, s.jwtSecret) if err != nil { - return fmt.Errorf("%w: %v", errInvalidToken, err) + return nil, fmt.Errorf("%w: %v", errInvalidToken, err) } if _, ok := claims["exp"]; !ok { - return fmt.Errorf("%w: missing exp claim", errInvalidToken) + return nil, fmt.Errorf("%w: missing exp claim", errInvalidToken) } - sub, ok := claims["sub"].(string) - if !ok || sub == "" { - return fmt.Errorf("%w: missing sub claim", errInvalidToken) + if sub, ok := claims["sub"].(string); !ok || sub == "" { + return nil, fmt.Errorf("%w: missing sub claim", errInvalidToken) + } + return claims, nil +} + +// authenticate validates the request's JWT and ensures that it authorises +// access to the given broadcast. +func (s *server) authenticate(c *fiber.Ctx, broadcastID string) error { + claims, err := s.tokenClaims(c) + if err != nil { + return err } - if sub != broadcastID { + if sub, _ := claims["sub"].(string); sub != broadcastID { return fmt.Errorf("%w: token is not valid for broadcast %s", errUnauthorized, broadcastID) } return nil diff --git a/cmd/oceanmedia/main.go b/cmd/oceanmedia/main.go index ecbe6879..19942121 100644 --- a/cmd/oceanmedia/main.go +++ b/cmd/oceanmedia/main.go @@ -124,6 +124,10 @@ func main() { app.Get("/health", indexHandler) app.Get("/_ah/warmup", indexHandler) + // Segments are requested as /segment?u=&token= and + // redirected to a pre-signed bucket URL. + app.Get("/segment", srv.handleSegment) + // Playlists are requested as /.m3u8. app.Get("/:file", srv.handlePlaylist) diff --git a/cmd/oceanmedia/playlist.go b/cmd/oceanmedia/playlist.go index 9d6cbc5c..f3c4d6e1 100644 --- a/cmd/oceanmedia/playlist.go +++ b/cmd/oceanmedia/playlist.go @@ -29,6 +29,7 @@ import ( "errors" "fmt" "math" + "net/url" "strings" "time" @@ -65,11 +66,7 @@ func (s *server) handlePlaylist(c *fiber.Ctx) error { } if err := s.authenticate(c, broadcastID); err != nil { - status := fiber.StatusForbidden - if errors.Is(err, errMissingToken) || errors.Is(err, errInvalidToken) { - status = fiber.StatusUnauthorized - } - return c.Status(status).JSON(fiber.Map{"error": err.Error()}) + return writeAuthError(c, err) } ctx := c.UserContext() @@ -121,11 +118,7 @@ func (s *server) handlePlaylist(c *fiber.Ctx) error { } } - playlist, err := s.renderPlaylist(ctx, event, segments, msn, hasMSN, skip) - if err != nil { - s.log("could not render playlist for broadcast %s: %v", broadcastID, err) - return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not render playlist"}) - } + playlist := s.renderPlaylist(event, segments, msn, hasMSN, skip, requestBaseURL(c), requestToken(c)) c.Set(fiber.HeaderContentType, playlistMIME) c.Set(fiber.HeaderCacheControl, "no-cache") @@ -133,6 +126,33 @@ func (s *server) handlePlaylist(c *fiber.Ctx) error { return c.SendString(playlist) } +// handleSegment redirects a segment URL to a pre-signed bucket URL. +func (s *server) handleSegment(c *fiber.Ctx) error { + rawURI := c.Query("u") + if rawURI == "" { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "missing segment URI"}) + } + + claims, err := s.tokenClaims(c) + if err != nil { + return writeAuthError(c, err) + } + broadcastID, _ := claims["sub"].(string) + if !uriBelongsToBroadcast(rawURI, broadcastID) { + return c.Status(fiber.StatusForbidden).JSON(fiber.Map{"error": "segment does not belong to this broadcast"}) + } + + signed, err := s.signURL(c.UserContext(), rawURI) + if err != nil { + s.log("could not sign segment %q: %v", rawURI, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not sign segment"}) + } + + c.Set(fiber.HeaderCacheControl, "no-store") + c.Set("Access-Control-Allow-Origin", "*") + return c.Redirect(signed, fiber.StatusFound) +} + // errMSNTooFarAhead is returned when a requested media sequence number // is beyond the next expected segment. var errMSNTooFarAhead = errors.New("requested media sequence number is too far ahead") @@ -175,17 +195,24 @@ func (s *server) blockForMSN(ctx context.Context, broadcastID string, msn int, s return segments, nil } -// renderPlaylist builds an HLS media playlist. msn/hasMSN and skip -// control delta updates. The whole stream is included so that viewers -// can seek within a live stream. -func (s *server) renderPlaylist(ctx context.Context, event *model.BroadcastEvent, segments []model.MtsMediaV2, msn int, hasMSN, skip bool) (string, error) { +// renderPlaylist builds an HLS EVENT playlist. When a client +// requests a delta update (_HLS_skip=YES with _HLS_msn) the segments +// before its requested media sequence number are omitted and indicated +// with EXT-X-SKIP. +func (s *server) renderPlaylist(event *model.BroadcastEvent, segments []model.MtsMediaV2, msn int, hasMSN, skip bool, baseURL, token string) string { live := event.EndTime.IsZero() - // When a client requests a delta we omit the segments it already - // has, i.e. those before the requested media sequence number. + // first is the index of the first segment to list. For a delta we skip + // the segments the client already has (those before its requested MSN). + // At least one segment is always listed, including at the live edge when + // the requested segment is not available yet, so the playlist is never + // empty. first := 0 - if hasMSN && skip && msn > 0 && msn <= len(segments) { + if hasMSN && skip && msn > 0 && len(segments) > 0 { first = msn + if first >= len(segments) { + first = len(segments) - 1 + } } // The target duration is the longest segment duration, rounded up. @@ -213,36 +240,47 @@ func (s *server) renderPlaylist(ctx context.Context, event *model.BroadcastEvent // (EXT-X-SERVER-CONTROL and EXT-X-SKIP). b.WriteString("#EXT-X-VERSION:9\n") fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", target) - // The playlist always begins at media sequence number 0 (the start of - // the broadcast). For a delta playlist, EXT-X-SKIP indicates how many - // leading segments have been omitted, so the first listed segment has - // media sequence number "first". + // The playlist begins at media sequence number 0 (the start of the + // broadcast). The media sequence number of the first listed segment is + // EXT-X-MEDIA-SEQUENCE plus SKIPPED-SEGMENTS. b.WriteString("#EXT-X-MEDIA-SEQUENCE:0\n") b.WriteString("#EXT-X-PLAYLIST-TYPE:EVENT\n") fmt.Fprintf(&b, "#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,CAN-SKIP-UNTIL=%d\n", canSkipUntil) if first > 0 { - fmt.Fprintf(&b, "#EXT-X-SKIP:SKIPPED-SEGMENTS:%d\n", first) + fmt.Fprintf(&b, "#EXT-X-SKIP:SKIPPED-SEGMENTS=%d\n", first) } for i := first; i < len(segments); i++ { seg := segments[i] - signed, err := s.signURL(ctx, seg.StorageURI) - if err != nil { - return "", fmt.Errorf("segment %d: %w", i, err) - } - if i > 0 && seg.Discontinuity { b.WriteString("#EXT-X-DISCONTINUITY\n") } fmt.Fprintf(&b, "#EXTINF:%.3f,\n", float64(seg.Duration)/1000) - b.WriteString(signed) - b.WriteString("\n") + fmt.Fprintf(&b, "%s/segment?u=%s&token=%s\n", baseURL, url.QueryEscape(seg.StorageURI), url.QueryEscape(token)) } if !live { b.WriteString("#EXT-X-ENDLIST\n") } - return b.String(), nil + return b.String() +} + +// uriBelongsToBroadcast returns true if broadcastID appears as a path +// segment of rawURI. +func uriBelongsToBroadcast(rawURI, broadcastID string) bool { + if broadcastID == "" { + return false + } + u, err := url.Parse(rawURI) + if err != nil { + return false + } + for _, p := range strings.Split(strings.Trim(u.Path, "/"), "/") { + if p == broadcastID { + return true + } + } + return false } diff --git a/cmd/oceanmedia/playlist_test.go b/cmd/oceanmedia/playlist_test.go index 9ceb4c16..2c2b1626 100644 --- a/cmd/oceanmedia/playlist_test.go +++ b/cmd/oceanmedia/playlist_test.go @@ -29,7 +29,7 @@ import ( "fmt" "io" "net/http/httptest" - "strconv" + "net/url" "strings" "sync" "testing" @@ -57,7 +57,7 @@ func (fakeSigner) SignURL(_ context.Context, rawURI string, _ time.Duration) (st if !strings.HasPrefix(rawURI, "https://") { return "", fmt.Errorf("invalid storage URI %q", rawURI) } - return "https://signed.example.com/" + rawURI + "?sig=x", nil + return "https://signed.example.com/?u=" + url.QueryEscape(rawURI), nil } // countingSigner counts SignURL calls for cache tests. @@ -70,7 +70,7 @@ func (c *countingSigner) SignURL(_ context.Context, _ string, _ time.Duration) ( c.mu.Lock() defer c.mu.Unlock() c.calls++ - return "https://signed.example.com/" + strconv.Itoa(c.calls), nil + return "https://signed.example.com/" + fmt.Sprint(c.calls), nil } func (c *countingSigner) callCount() int { @@ -111,35 +111,16 @@ func newTestServer() *server { return s } -func TestServerSignURLCache(t *testing.T) { - cs := &countingSigner{} - s := newServer(nil, cs, testJWTSecret, time.Hour) - - now := time.Now() - s.now = func() time.Time { return now } - - u1, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") - require.NoError(t, err) - u2, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") - require.NoError(t, err) - assert.Equal(t, u1, u2) - assert.Equal(t, 1, cs.callCount()) - - // Past half the time to live the URL should be re-signed. - now = now.Add(31 * time.Minute) - u3, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") - require.NoError(t, err) - assert.NotEqual(t, u1, u3) - assert.Equal(t, 2, cs.callCount()) +// render renders a playlist with test defaults. +func render(s *server, event *model.BroadcastEvent, segments []model.MtsMediaV2, msn int, hasMSN, skip bool) string { + return s.renderPlaylist(event, segments, msn, hasMSN, skip, "http://example.com", "tok") } func TestRenderPlaylistVOD(t *testing.T) { s := newTestServer() - ctx := context.Background() event := &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()} - pl, err := s.renderPlaylist(ctx, event, testSegments(3, -1), 0, false, false) - require.NoError(t, err) + pl := render(s, event, testSegments(3, -1), 0, false, false) assert.Contains(t, pl, "#EXTM3U") assert.Contains(t, pl, "#EXT-X-VERSION:9") @@ -149,7 +130,8 @@ func TestRenderPlaylistVOD(t *testing.T) { assert.Contains(t, pl, "#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES") assert.Contains(t, pl, "#EXT-X-ENDLIST") assert.Equal(t, 3, strings.Count(pl, "#EXTINF:2.500,")) - assert.Equal(t, 3, strings.Count(pl, "https://signed.example.com/")) + assert.Equal(t, 3, strings.Count(pl, "http://example.com/segment?u=")) + assert.Contains(t, pl, url.QueryEscape(testSegments(1, -1)[0].StorageURI)) assert.NotContains(t, pl, "#EXT-X-SKIP") assert.NotContains(t, pl, "#EXT-X-DISCONTINUITY") } @@ -158,8 +140,7 @@ func TestRenderPlaylistLive(t *testing.T) { s := newTestServer() event := &model.BroadcastEvent{ID: testBroadcastID} - pl, err := s.renderPlaylist(context.Background(), event, testSegments(2, -1), 0, false, false) - require.NoError(t, err) + pl := render(s, event, testSegments(2, -1), 0, false, false) assert.NotContains(t, pl, "#EXT-X-ENDLIST") assert.Contains(t, pl, "#EXT-X-PLAYLIST-TYPE:EVENT") } @@ -168,8 +149,7 @@ func TestRenderPlaylistLiveEmpty(t *testing.T) { s := newTestServer() event := &model.BroadcastEvent{ID: testBroadcastID} - pl, err := s.renderPlaylist(context.Background(), event, nil, 0, false, false) - require.NoError(t, err) + pl := render(s, event, nil, 0, false, false) assert.Contains(t, pl, "#EXTM3U") assert.Contains(t, pl, "#EXT-X-TARGETDURATION:1") assert.NotContains(t, pl, "#EXTINF") @@ -180,13 +160,12 @@ func TestRenderPlaylistDiscontinuity(t *testing.T) { s := newTestServer() event := &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()} - pl, err := s.renderPlaylist(context.Background(), event, testSegments(3, 1), 0, false, false) - require.NoError(t, err) + pl := render(s, event, testSegments(3, 1), 0, false, false) assert.Equal(t, 1, strings.Count(pl, "#EXT-X-DISCONTINUITY")) disc := strings.Index(pl, "#EXT-X-DISCONTINUITY") - seg0 := strings.Index(pl, "seg0.ts") - seg1 := strings.Index(pl, "seg1.ts") + seg0 := strings.Index(pl, url.QueryEscape(fmt.Sprintf(testSegmentURL, 0))) + seg1 := strings.Index(pl, url.QueryEscape(fmt.Sprintf(testSegmentURL, 1))) assert.Greater(t, disc, seg0) assert.Less(t, disc, seg1) } @@ -195,29 +174,56 @@ func TestRenderPlaylistDelta(t *testing.T) { s := newTestServer() event := &model.BroadcastEvent{ID: testBroadcastID} - pl, err := s.renderPlaylist(context.Background(), event, testSegments(3, -1), 2, true, true) - require.NoError(t, err) + pl := render(s, event, testSegments(3, -1), 2, true, true) assert.Contains(t, pl, "#EXT-X-MEDIA-SEQUENCE:0") - assert.Contains(t, pl, "#EXT-X-SKIP:SKIPPED-SEGMENTS:2") + assert.Contains(t, pl, "#EXT-X-SKIP:SKIPPED-SEGMENTS=2") + assert.Contains(t, pl, "CAN-SKIP-UNTIL=") assert.Equal(t, 1, strings.Count(pl, "#EXTINF:")) - assert.Contains(t, pl, "seg2.ts") - assert.NotContains(t, pl, "seg0.ts") - assert.NotContains(t, pl, "seg1.ts") + assert.Contains(t, pl, url.QueryEscape(fmt.Sprintf(testSegmentURL, 2))) + assert.NotContains(t, pl, url.QueryEscape(fmt.Sprintf(testSegmentURL, 0))) + assert.NotContains(t, pl, url.QueryEscape(fmt.Sprintf(testSegmentURL, 1))) } -func TestRenderPlaylistSignError(t *testing.T) { +// TestRenderPlaylistDeltaNoNewSegments checks that when a client requests a +// delta for a media sequence number that is not available yet, at least one +// segment is listed (the playlist is never empty). +func TestRenderPlaylistDeltaNoNewSegments(t *testing.T) { s := newTestServer() event := &model.BroadcastEvent{ID: testBroadcastID} - segs := testSegments(1, -1) - segs[0].StorageURI = "not a uri" - _, err := s.renderPlaylist(context.Background(), event, segs, 0, false, false) - assert.Error(t, err) + pl := render(s, event, testSegments(3, -1), 5, true, true) + + assert.Contains(t, pl, "#EXT-X-SKIP:SKIPPED-SEGMENTS=2") + assert.Equal(t, 1, strings.Count(pl, "#EXTINF:")) + assert.Contains(t, pl, url.QueryEscape(fmt.Sprintf(testSegmentURL, 2))) +} + +func TestServerSignURLCache(t *testing.T) { + cs := &countingSigner{} + s := newServer(nil, cs, testJWTSecret, time.Hour) + + now := time.Now() + s.now = func() time.Time { return now } + + u1, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") + require.NoError(t, err) + u2, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") + require.NoError(t, err) + assert.Equal(t, u1, u2) + assert.Equal(t, 1, cs.callCount()) + + // Once most of the time to live has elapsed the URL is re-signed. + now = now.Add(55 * time.Minute) + u3, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") + require.NoError(t, err) + assert.NotEqual(t, u1, u3) + assert.Equal(t, 2, cs.callCount()) } func newTestApp(s *server) *fiber.App { app := fiber.New() + app.Get("/segment", s.handleSegment) app.Get("/:file", s.handlePlaylist) return app } @@ -248,6 +254,7 @@ func TestHandlePlaylistSuccess(t *testing.T) { assert.Equal(t, fiber.StatusOK, status) assert.Contains(t, body, "#EXTM3U") assert.Contains(t, body, "#EXT-X-ENDLIST") + assert.Contains(t, body, "http://example.com/segment?u=") } func TestHandlePlaylistAuth(t *testing.T) { @@ -370,7 +377,7 @@ func TestHandlePlaylistBlockingReload(t *testing.T) { assert.Equal(t, fiber.StatusOK, status) assert.GreaterOrEqual(t, calls, 2) assert.Equal(t, 2, strings.Count(body, "#EXTINF:")) - assert.Contains(t, body, "seg1.ts") + assert.Contains(t, body, url.QueryEscape(fmt.Sprintf(testSegmentURL, 1))) } func TestHandlePlaylistNotM3U8(t *testing.T) { @@ -379,3 +386,43 @@ func TestHandlePlaylistNotM3U8(t *testing.T) { status, _ := doRequest(t, app, "/"+testBroadcastID) assert.Equal(t, fiber.StatusNotFound, status) } + +func TestHandleSegment(t *testing.T) { + s := newTestServer() + app := newTestApp(s) + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + uri := fmt.Sprintf(testSegmentURL, 0) + + segmentRequest := func(token, rawURI string) string { + return "/segment?u=" + url.QueryEscape(rawURI) + "&token=" + token + } + + t.Run("RedirectsToSignedURL", func(t *testing.T) { + resp, err := app.Test(httptest.NewRequest("GET", segmentRequest(token, uri), nil)) + require.NoError(t, err) + defer resp.Body.Close() + assert.Equal(t, fiber.StatusFound, resp.StatusCode) + assert.Contains(t, resp.Header.Get("Location"), "https://signed.example.com/") + }) + + t.Run("MissingURI", func(t *testing.T) { + status, _ := doRequest(t, app, "/segment?token="+token) + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("MissingToken", func(t *testing.T) { + status, _ := doRequest(t, app, segmentRequest("", uri)) + assert.Equal(t, fiber.StatusUnauthorized, status) + }) + + t.Run("OtherBroadcast", func(t *testing.T) { + otherToken := testToken(t, "some-other-broadcast", time.Now().Add(time.Hour)) + status, _ := doRequest(t, app, segmentRequest(otherToken, uri)) + assert.Equal(t, fiber.StatusForbidden, status) + }) + + t.Run("URINotMatchingBroadcast", func(t *testing.T) { + status, _ := doRequest(t, app, segmentRequest(token, "https://acct.r2.cloudflarestorage.com/bucket/other/seg.ts")) + assert.Equal(t, fiber.StatusForbidden, status) + }) +}