diff --git a/cmd/oceanmedia/auth.go b/cmd/oceanmedia/auth.go new file mode 100644 index 00000000..0b4ab0f4 --- /dev/null +++ b/cmd/oceanmedia/auth.go @@ -0,0 +1,72 @@ +/* +DESCRIPTION + OceanMedia JWT authentication. + +AUTHORS + Elliot Shine + +LICENSE + Copyright (C) 2026 the Australian Ocean Lab (AusOcean). + + This is free software: you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This is distributed in the hope that it will be useful, but WITHOUT + ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public + License for more details. + + You should have received a copy of the GNU General Public License in + gpl.txt. If not, see http://www.gnu.org/licenses/. +*/ + +package main + +import ( + "errors" + "fmt" + + "github.com/ausocean/cloud/gauth" + "github.com/gofiber/fiber/v2" +) + +var ( + errMissingToken = errors.New("missing token") + errInvalidToken = errors.New("invalid token") + 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 { + token := c.Query("token") + if token == "" { + token = c.Get(fiber.HeaderAuthorization) + } + if token == "" { + return errMissingToken + } + + claims, err := gauth.GetClaims(token, s.jwtSecret) + if err != nil { + return fmt.Errorf("%w: %v", errInvalidToken, err) + } + if _, ok := claims["exp"]; !ok { + return 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 != 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 new file mode 100644 index 00000000..ecbe6879 --- /dev/null +++ b/cmd/oceanmedia/main.go @@ -0,0 +1,180 @@ +/* +DESCRIPTION + OceanMedia generates HLS playlists for OceanMedia streams. + +AUTHORS + Elliot Shine + +LICENSE + Copyright (C) 2026 the Australian Ocean Lab (AusOcean). + + This is free software: you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This is distributed in the hope that it will be useful, but WITHOUT + ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public + License for more details. + + You should have received a copy of the GNU General Public License in + gpl.txt. If not, see http://www.gnu.org/licenses/. +*/ + +// OceanMedia serves HLS playlists for OceanMedia broadcasts. A +// broadcast is made up of a BroadcastEvent and a series of MtsMediaV2 +// segments whose media is stored in an object store bucket. Playlists +// are generated on demand with pre-signed segment URLs so that players +// fetch media directly from the bucket. +// +// Playlists support full refreshes, LL-HLS blocking reloads +// (_HLS_msn), delta updates (_HLS_skip) and DVR of the whole stream. +package main + +import ( + _ "github.com/joho/godotenv/autoload" + + "context" + "flag" + "fmt" + "log" + "os" + rtdebug "runtime/debug" + "strconv" + "sync" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/gauth" + "github.com/ausocean/cloud/model" + "github.com/ausocean/cloud/storage" + "github.com/gofiber/fiber/v2" +) + +const ( + projectID = "oceanmedia" + version = "v0.1.0" +) + +// Secret names within OCEANMEDIA_SECRETS. +const ( + secretJWT = "jwtSecret" // JWT signing secret shared with referring services. + secretCloudflareAccount = "cloudflareAccountID" // Cloudflare R2 account ID. + secretCloudflareAccess = "cloudflareAccessKey" // Cloudflare R2 access key ID. + secretCloudflareSecret = "cloudflareSecretKey" // Cloudflare R2 secret access key. +) + +var ( + setupMutex sync.Mutex + mediaStore datastore.Store + srv *server + debug bool + standalone bool + storePath string + commitHash string +) + +func init() { + if info, ok := rtdebug.ReadBuildInfo(); ok { + for _, setting := range info.Settings { + if setting.Key == "vcs.revision" { + commitHash = setting.Value + if len(commitHash) > 7 { + commitHash = commitHash[:7] + } + break + } + } + } + if commitHash == "" { + commitHash = os.Getenv("COMMIT_HASH") + } +} + +func main() { + defaultPort := 8085 + v := os.Getenv("PORT") + if v != "" { + i, err := strconv.Atoi(v) + if err == nil { + defaultPort = i + } + } + + var ( + host string + port int + signedURLExpiry time.Duration + ) + flag.BoolVar(&debug, "debug", false, "Run in debug mode.") + flag.BoolVar(&standalone, "standalone", false, "Run in standalone mode.") + flag.StringVar(&host, "host", "localhost", "Host we run on in standalone mode") + flag.IntVar(&port, "port", defaultPort, "Port we listen on in standalone mode") + flag.StringVar(&storePath, "filestore", "store", "File store path") + flag.DurationVar(&signedURLExpiry, "signedurlttl", 12*time.Hour, "Pre-signed segment URL time to live") + flag.Parse() + + // Perform one-time setup or bail. + setup(context.Background(), signedURLExpiry) + + app := fiber.New(fiber.Config{DisableStartupMessage: true}) + + app.Get("/", indexHandler) + app.Get("/health", indexHandler) + app.Get("/_ah/warmup", indexHandler) + + // Playlists are requested as /.m3u8. + app.Get("/:file", srv.handlePlaylist) + + log.Printf("Listening on %s:%d", host, port) + log.Fatal(app.Listen(fmt.Sprintf("%s:%d", host, port))) +} + +// indexHandler handles health and liveness requests. +func indexHandler(c *fiber.Ctx) error { + if commitHash != "" { + return c.SendString(projectID + " " + version + " (" + commitHash + ")") + } + return c.SendString(projectID + " " + version) +} + +// setup executes per-instance one-time warmup and is used to +// initialize the datastores, secrets and signer. +func setup(ctx context.Context, signedURLExpiry time.Duration) { + setupMutex.Lock() + defer setupMutex.Unlock() + + if srv != nil { + return + } + + var err error + _, mediaStore, err = model.SetupDatastore(standalone, false, storePath, ctx) + if err != nil { + log.Fatalf("could not setup datastores: %v", err) + } + + secrets, err := gauth.GetSecrets(ctx, projectID, nil) + if err != nil { + log.Fatalf("could not get secrets: %v", err) + } + + jwtSecret := []byte(secrets[secretJWT]) + if len(jwtSecret) == 0 { + log.Fatalf("could not get %s, cannot validate tokens", secretJWT) + } + if secrets[secretCloudflareAccount] == "" { + log.Fatalf("could not get %s, cannot sign segment URLs", secretCloudflareAccount) + } + if secrets[secretCloudflareAccess] == "" || secrets[secretCloudflareSecret] == "" { + log.Fatalf("could not get %s/%s, cannot sign segment URLs", secretCloudflareAccess, secretCloudflareSecret) + } + + // The provider's account ID is used to reject segment URIs belonging to + // another account. The bucket is taken from each segment's StorageURI. + provider := storage.NewCloudflare(secrets[secretCloudflareAccount], secrets[secretCloudflareAccess], secrets[secretCloudflareSecret], "") + + // Both BroadcastEvent and MtsMediaV2 live in the media (vidgrind) datastore. + srv = newServer(mediaStore, provider, jwtSecret, signedURLExpiry) +} diff --git a/cmd/oceanmedia/playlist.go b/cmd/oceanmedia/playlist.go new file mode 100644 index 00000000..9d6cbc5c --- /dev/null +++ b/cmd/oceanmedia/playlist.go @@ -0,0 +1,248 @@ +/* +DESCRIPTION + OceanMedia HLS playlist generation. + +AUTHORS + Elliot Shine + +LICENSE + Copyright (C) 2026 the Australian Ocean Lab (AusOcean). + + This is free software: you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This is distributed in the hope that it will be useful, but WITHOUT + ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public + License for more details. + + You should have received a copy of the GNU General Public License in + gpl.txt. If not, see http://www.gnu.org/licenses/. +*/ + +package main + +import ( + "context" + "errors" + "fmt" + "math" + "strings" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/model" + "github.com/gofiber/fiber/v2" +) + +// playlistMIME is the MIME type used for HLS playlists. +const playlistMIME = "application/vnd.apple.mpegurl" + +// skipUntilTargets is the minimum value of EXT-X-SERVER-CONTROL +// CAN-SKIP-UNTIL expressed in target durations. +const skipUntilTargets = 6 + +// handlePlaylist serves an HLS media playlist for a broadcast. +// +// The broadcast ID is taken from the request path, e.g. +// GET /.m3u8?token=. +// +// LL-HLS query parameters: +// +// _HLS_msn= - block until media sequence number n is available. +// _HLS_skip=YES - return a delta playlist (EXT-X-SKIP). +// _HLS_part= - unsupported; whole segments are served. +func (s *server) handlePlaylist(c *fiber.Ctx) error { + file := c.Params("file") + if !strings.HasSuffix(file, ".m3u8") { + return fiber.ErrNotFound + } + broadcastID := strings.TrimSuffix(file, ".m3u8") + if broadcastID == "" { + return fiber.ErrNotFound + } + + 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()}) + } + + ctx := c.UserContext() + + var query struct { + MSN *int `query:"_HLS_msn"` + Part string `query:"_HLS_part"` + Skip string `query:"_HLS_skip"` + } + if err := c.QueryParser(&query); err != nil { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid query parameters: " + err.Error()}) + } + if query.Part != "" { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "_HLS_part is not supported"}) + } + hasMSN := query.MSN != nil + msn := 0 + if hasMSN { + msn = *query.MSN + if msn < 0 { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid _HLS_msn"}) + } + } + skip := strings.EqualFold(query.Skip, "YES") + + event, err := s.loadEvent(ctx, broadcastID) + if err != nil { + if errors.Is(err, datastore.ErrNoSuchEntity) { + return fiber.ErrNotFound + } + s.log("could not get broadcast %s: %v", broadcastID, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not get broadcast"}) + } + + segments, err := s.loadSegments(ctx, broadcastID) + if err != nil { + s.log("could not get segments for broadcast %s: %v", broadcastID, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not get segments"}) + } + + if hasMSN { + segments, err = s.blockForMSN(ctx, broadcastID, msn, segments, event.EndTime.IsZero()) + if err != nil { + if errors.Is(err, errMSNTooFarAhead) { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": err.Error()}) + } + s.log("blocking reload for broadcast %s failed: %v", broadcastID, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not get segments"}) + } + } + + 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"}) + } + + c.Set(fiber.HeaderContentType, playlistMIME) + c.Set(fiber.HeaderCacheControl, "no-cache") + c.Set("Access-Control-Allow-Origin", "*") + return c.SendString(playlist) +} + +// 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") + +// blockForMSN waits until the segment with the given media sequence +// number is available, or the block timeout elapses. It returns the +// (possibly refreshed) segments and an error if the requested MSN is +// too far ahead. A non-live broadcast will never receive new segments, +// so blocking is skipped for it. +func (s *server) blockForMSN(ctx context.Context, broadcastID string, msn int, segments []model.MtsMediaV2, live bool) ([]model.MtsMediaV2, error) { + if msn < 0 { + return nil, errMSNTooFarAhead + } + // The client may request at most the next segment after the last + // available one. + if msn > len(segments) { + return nil, errMSNTooFarAhead + } + if msn < len(segments) { + return segments, nil + } + if !live { + // The broadcast has ended and the segment does not exist. + return nil, errMSNTooFarAhead + } + + deadline := time.Now().Add(s.blockTimeout) + for msn >= len(segments) && time.Now().Before(deadline) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(s.blockPoll): + } + var err error + segments, err = s.loadSegments(ctx, broadcastID) + if err != nil { + return nil, err + } + } + 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) { + 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 := 0 + if hasMSN && skip && msn > 0 && msn <= len(segments) { + first = msn + } + + // The target duration is the longest segment duration, rounded up. + var ( + target int64 = 1 + totalSecs float64 + ) + for _, seg := range segments { + secs := float64(seg.Duration) / 1000 + totalSecs += secs + if d := int64(math.Ceil(secs)); d > target { + target = d + } + } + + // The server can skip as far back as the whole playlist. + canSkipUntil := int64(math.Ceil(totalSecs)) + if min := skipUntilTargets * target; canSkipUntil < min { + canSkipUntil = min + } + + var b strings.Builder + b.WriteString("#EXTM3U\n") + // Version 9 is required for the LL-HLS tags used below + // (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". + 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) + } + + 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") + } + + if !live { + b.WriteString("#EXT-X-ENDLIST\n") + } + + return b.String(), nil +} diff --git a/cmd/oceanmedia/playlist_test.go b/cmd/oceanmedia/playlist_test.go new file mode 100644 index 00000000..9ceb4c16 --- /dev/null +++ b/cmd/oceanmedia/playlist_test.go @@ -0,0 +1,381 @@ +/* +DESCRIPTION + Tests for OceanMedia playlist handling. + +AUTHORS + Elliot Shine + +LICENSE + Copyright (C) 2026 the Australian Ocean Lab (AusOcean). + + This is free software: you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This is distributed in the hope that it will be useful, but WITHOUT + ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public + License for more details. + + You should have received a copy of the GNU General Public License in + gpl.txt. If not, see http://www.gnu.org/licenses/. +*/ + +package main + +import ( + "context" + "fmt" + "io" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/gauth" + "github.com/ausocean/cloud/model" + "github.com/gofiber/fiber/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + testBroadcastID = "c7cfd21b-d69d-437f-969c-34cd4ac15bd5" + testSegmentURL = "https://acct.r2.cloudflarestorage.com/bucket/" + testBroadcastID + "/seg%d.ts" +) + +var testJWTSecret = []byte("test-jwt-secret") + +// fakeSigner returns deterministic signed URLs, rejecting invalid URIs. +type fakeSigner struct{} + +func (fakeSigner) SignURL(_ context.Context, rawURI string, _ time.Duration) (string, error) { + if !strings.HasPrefix(rawURI, "https://") { + return "", fmt.Errorf("invalid storage URI %q", rawURI) + } + return "https://signed.example.com/" + rawURI + "?sig=x", nil +} + +// countingSigner counts SignURL calls for cache tests. +type countingSigner struct { + mu sync.Mutex + calls int +} + +func (c *countingSigner) SignURL(_ context.Context, _ string, _ time.Duration) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls++ + return "https://signed.example.com/" + strconv.Itoa(c.calls), nil +} + +func (c *countingSigner) callCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + +func testSegments(n int, discontinuityAt int) []model.MtsMediaV2 { + segs := make([]model.MtsMediaV2, n) + for i := range segs { + segs[i] = model.MtsMediaV2{ + MID: 16, + Timestamp: 1790596912 + int64(i*3), + Duration: 2500, + StorageURI: fmt.Sprintf(testSegmentURL, i), + BroadcastID: testBroadcastID, + Discontinuity: discontinuityAt != -1 && i == discontinuityAt, + Type: "video/h264", + } + } + return segs +} + +func testToken(t *testing.T, sub string, exp time.Time) string { + t.Helper() + tok, err := gauth.PutClaims(map[string]interface{}{ + "sub": sub, + "exp": exp.Unix(), + }, testJWTSecret) + require.NoError(t, err) + return tok +} + +func newTestServer() *server { + s := newServer(nil, fakeSigner{}, testJWTSecret, time.Hour) + s.log = func(string, ...any) {} + 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()) +} + +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) + + assert.Contains(t, pl, "#EXTM3U") + assert.Contains(t, pl, "#EXT-X-VERSION:9") + assert.Contains(t, pl, "#EXT-X-TARGETDURATION:3") + assert.Contains(t, pl, "#EXT-X-MEDIA-SEQUENCE:0") + assert.Contains(t, pl, "#EXT-X-PLAYLIST-TYPE:EVENT") + 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.NotContains(t, pl, "#EXT-X-SKIP") + assert.NotContains(t, pl, "#EXT-X-DISCONTINUITY") +} + +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) + assert.NotContains(t, pl, "#EXT-X-ENDLIST") + assert.Contains(t, pl, "#EXT-X-PLAYLIST-TYPE:EVENT") +} + +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) + assert.Contains(t, pl, "#EXTM3U") + assert.Contains(t, pl, "#EXT-X-TARGETDURATION:1") + assert.NotContains(t, pl, "#EXTINF") + assert.NotContains(t, pl, "#EXT-X-ENDLIST") +} + +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) + + 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") + assert.Greater(t, disc, seg0) + assert.Less(t, disc, seg1) +} + +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) + + assert.Contains(t, pl, "#EXT-X-MEDIA-SEQUENCE:0") + assert.Contains(t, pl, "#EXT-X-SKIP:SKIPPED-SEGMENTS:2") + 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") +} + +func TestRenderPlaylistSignError(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) +} + +func newTestApp(s *server) *fiber.App { + app := fiber.New() + app.Get("/:file", s.handlePlaylist) + return app +} + +func doRequest(t *testing.T, app *fiber.App, url string) (int, string) { + t.Helper() + resp, err := app.Test(httptest.NewRequest("GET", url, nil)) + require.NoError(t, err) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + return resp.StatusCode, string(body) +} + +func TestHandlePlaylistSuccess(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()}, nil + } + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(2, -1), nil + } + app := newTestApp(s) + + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + status, body := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + + assert.Equal(t, fiber.StatusOK, status) + assert.Contains(t, body, "#EXTM3U") + assert.Contains(t, body, "#EXT-X-ENDLIST") +} + +func TestHandlePlaylistAuth(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID}, nil + } + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(1, -1), nil + } + app := newTestApp(s) + + t.Run("Missing", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8") + assert.Equal(t, fiber.StatusUnauthorized, status) + }) + + t.Run("Expired", func(t *testing.T) { + token := testToken(t, testBroadcastID, time.Now().Add(-time.Hour)) + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + assert.Equal(t, fiber.StatusUnauthorized, status) + }) + + t.Run("WrongBroadcast", func(t *testing.T) { + token := testToken(t, "some-other-broadcast", time.Now().Add(time.Hour)) + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + assert.Equal(t, fiber.StatusForbidden, status) + }) + + t.Run("BearerHeader", func(t *testing.T) { + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + req := httptest.NewRequest("GET", "/"+testBroadcastID+".m3u8", nil) + req.Header.Set("Authorization", "Bearer "+token) + resp, err := app.Test(req) + require.NoError(t, err) + assert.Equal(t, fiber.StatusOK, resp.StatusCode) + }) +} + +func TestHandlePlaylistNotFound(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return nil, fmt.Errorf("failed to get BroadcastEvent: %w", datastore.ErrNoSuchEntity) + } + app := newTestApp(s) + + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + assert.Equal(t, fiber.StatusNotFound, status) +} + +func TestHandlePlaylistBadRequest(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID}, nil + } + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(2, -1), nil + } + app := newTestApp(s) + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + + t.Run("InvalidMSN", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=abc") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("PartUnsupported", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_part=0") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("TooFarAhead", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=5") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("TooFarAheadEmpty", func(t *testing.T) { + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { return nil, nil } + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=1") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("EndedNoBlocking", func(t *testing.T) { + // An ended broadcast will never receive new segments, so requesting + // the next media sequence number must not block; it is an error. + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(2, -1), nil + } + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()}, nil + } + s.blockTimeout = time.Hour // Would hang if blocking were attempted. + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=2") + assert.Equal(t, fiber.StatusBadRequest, status) + }) +} + +func TestHandlePlaylistBlockingReload(t *testing.T) { + s := newTestServer() + s.blockPoll = time.Millisecond + s.blockTimeout = 2 * time.Second + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID}, nil + } + var calls int + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + calls++ + n := 1 + if calls > 1 { + n = 2 + } + return testSegments(n, -1), nil + } + app := newTestApp(s) + + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + status, body := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=1") + + 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") +} + +func TestHandlePlaylistNotM3U8(t *testing.T) { + s := newTestServer() + app := newTestApp(s) + status, _ := doRequest(t, app, "/"+testBroadcastID) + assert.Equal(t, fiber.StatusNotFound, status) +} diff --git a/cmd/oceanmedia/server.go b/cmd/oceanmedia/server.go new file mode 100644 index 00000000..c5c36788 --- /dev/null +++ b/cmd/oceanmedia/server.go @@ -0,0 +1,131 @@ +/* +DESCRIPTION + OceanMedia server and dependencies. + +AUTHORS + Elliot Shine + +LICENSE + Copyright (C) 2026 the Australian Ocean Lab (AusOcean). + + This is free software: you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This is distributed in the hope that it will be useful, but WITHOUT + ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public + License for more details. + + You should have received a copy of the GNU General Public License in + gpl.txt. If not, see http://www.gnu.org/licenses/. +*/ + +package main + +import ( + "context" + "log" + "sync" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/model" + "github.com/ausocean/cloud/storage" +) + +const ( + // defaultBlockPoll is how often the datastore is polled while + // waiting for a segment during a blocking reload. + defaultBlockPoll = 500 * time.Millisecond + + // defaultBlockTimeout is how long a blocking reload will wait for a + // segment before returning the current playlist. + defaultBlockTimeout = 15 * time.Second +) + +// urlSigner signs object storage URIs. It is satisfied by storage.Provider. +type urlSigner interface { + SignURL(ctx context.Context, rawURI string, ttl time.Duration) (string, error) +} + +type cachedURL struct { + url string + expires time.Time +} + +// server holds the dependencies used by OceanMedia's HTTP handlers. +type server struct { + store datastore.Store + signer urlSigner + jwtSecret []byte + signedURLTTL time.Duration + + blockPoll time.Duration + blockTimeout time.Duration + + // now is a field so that time can be controlled in tests. + now func() time.Time + + // Signed URLs are cached until half their lifetime has elapsed, so a + // URL handed to a client always has at least half its lifetime left. + signMu sync.Mutex + signCache map[string]cachedURL + + // loadEvent and loadSegments are fields so that they can be + // overridden in tests. + loadEvent func(ctx context.Context, broadcastID string) (*model.BroadcastEvent, error) + loadSegments func(ctx context.Context, broadcastID string) ([]model.MtsMediaV2, error) + + log func(string, ...any) +} + +// newServer returns a new server with the supplied dependencies. +func newServer(store datastore.Store, signer urlSigner, jwtSecret []byte, signedURLTTL time.Duration) *server { + if signedURLTTL <= 0 { + signedURLTTL = storage.DefaultSignedURLExpiry + } + s := &server{ + store: store, + signer: signer, + jwtSecret: jwtSecret, + signedURLTTL: signedURLTTL, + blockPoll: defaultBlockPoll, + blockTimeout: defaultBlockTimeout, + now: time.Now, + signCache: make(map[string]cachedURL), + log: log.Printf, + } + s.loadEvent = func(ctx context.Context, broadcastID string) (*model.BroadcastEvent, error) { + return model.GetBroadcastEvent(ctx, s.store, broadcastID) + } + s.loadSegments = func(ctx context.Context, broadcastID string) ([]model.MtsMediaV2, error) { + return model.GetMtsMediaV2ByBroadcast(ctx, s.store, broadcastID, nil, 0) + } + return s +} + +// signURL returns a pre-signed URL for the given storage URI, caching the +// result until half the signed URL time to live has elapsed. +func (s *server) signURL(ctx context.Context, rawURI string) (string, error) { + now := s.now() + + s.signMu.Lock() + if c, ok := s.signCache[rawURI]; ok && now.Before(c.expires) { + s.signMu.Unlock() + return c.url, nil + } + s.signMu.Unlock() + + signed, err := s.signer.SignURL(ctx, rawURI, s.signedURLTTL) + if err != nil { + return "", err + } + + s.signMu.Lock() + s.signCache[rawURI] = cachedURL{url: signed, expires: now.Add(s.signedURLTTL / 2)} + s.signMu.Unlock() + + return signed, nil +} diff --git a/oceanmedia.yaml b/oceanmedia.yaml new file mode 100644 index 00000000..3b2d4525 --- /dev/null +++ b/oceanmedia.yaml @@ -0,0 +1,20 @@ +# gcloud app deploy --version V --project ausoceanmedia oceanmedia.yaml +runtime: go126 + +env_variables: + AUSOCEAN_CREDENTIALS: gs://ausocean/AusOcean-ef9cd79f81a8.json + VIDGRIND_CREDENTIALS: gs://ausocean/VideoGrinder-b0ad82abac05.json + OCEANMEDIA_SECRETS: gs://ausocean/OceanMedia-secrets.txt + +main: cmd/oceanmedia + +handlers: + - url: /.* + secure: always + script: auto + +automatic_scaling: + min_instances: 1 + +inbound_services: + - warmup diff --git a/scripts/run_all_standalone.sh b/scripts/run_all_standalone.sh index 8a9c8b41..ee4c03c1 100755 --- a/scripts/run_all_standalone.sh +++ b/scripts/run_all_standalone.sh @@ -24,6 +24,7 @@ # Set the paths to your services OCEANBENCH_PATH="../cmd/oceanbench" OCEANTV_PATH="../cmd/oceantv" +OCEANMEDIA_PATH="../cmd/oceanmedia" OCEANCRON_PATH="../cmd/oceancron" DATABLUE_PATH="../cmd/datablue" @@ -79,6 +80,7 @@ trap cleanup SIGINT SIGTERM run_service "OceanBench" "$OCEANBENCH_PATH" "go run . --standalone --tvurl $TV_URL --cronurl $CRON_URL" run_service "OceanTV" "$OCEANTV_PATH" "go run . --standalone --filestore $FILESTORE_PATH" run_service "OceanCron" "$OCEANCRON_PATH" "go run . --standalone --filestore $FILESTORE_PATH" +run_service "OceanMedia" "$OCEANMEDIA_PATH" "go run . --standalone --filestore $FILESTORE_PATH" run_service "DataBlue" "$DATABLUE_PATH" "go run . --standalone --filestore $FILESTORE_PATH" # Wait for all background processes.