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) + }) +}