Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
8.2 kB · 205 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206package media
import ( "bytes" "context" "fmt" "os" "sort" "testing" "time"
"github.com/stretchr/testify/require" "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/livehls" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/muxl")
// newBareSegmentSigner is an ephemeral signer with the test streamer name,// enough to drive SignSegmentStream in tests.func newBareSegmentSigner(t *testing.T) *MediaSignerLocal { t.Helper() ms, err := NewEphemeralMediaSigner("test-streamer") require.NoError(t, err) return ms}
// TestValidateMP4MediaBareSegment exercises the full .m4s-native validate// path: sign the fragmented fixture per-segment (the live ingest shape),// reassemble one GoP's bare canonical .m4s, and run ValidateMP4Media over it.// That wraps the bare segment to a flat MP4 for gstreamer (codec/dimensions)// and verifies the signatures in-wasm — proving a signed bare .m4s parses// through qtdemux with the correct co64 offsets wrap-flat synthesizes.func TestValidateMP4MediaBareSegment(t *testing.T) { ctx := context.Background() ms := newBareSegmentSigner(t)
frag, err := os.ReadFile(getFixture("h264-opus-frag.mp4")) require.NoError(t, err)
// Sign per-segment; keep the first GoP's bare .m4s (what ValidateMP4 gets // per call in the live path). eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() var m4s []byte for ev := range eventCh { if ev.Type == "signed-segment" && m4s == nil { m4s = concatTracksByID(ev.Tracks) } } require.NoError(t, <-errCh) require.NotEmpty(t, m4s, "expected at least one signed GoP")
res, err := ValidateMP4Media(ctx, m4s) require.NoError(t, err) require.NotNil(t, res.MediaData) require.NotEmpty(t, res.MediaData.Video, "should parse a video track") require.Greater(t, res.MediaData.Video[0].Width, 0, "video width from qtdemux") require.Greater(t, res.MediaData.Video[0].Height, 0, "video height from qtdemux") require.NotEmpty(t, res.MediaData.Audio, "should parse an audio track")}
// A peer naming one streamer must not get another creator's otherwise valid// segment ingested as theirs.func TestValidateReplicatedMP4BindsStreamer(t *testing.T) { ctx := context.Background() segment := sourceSegmentForTest(t) mm, _ := getStaticTestMediaManager(t) err := mm.ValidateReplicatedMP4(ctx, bytes.NewReader(segment), "did:plc:different-streamer") require.ErrorContains(t, err, "not did:plc:different-streamer") require.Nil(t, mm.GetLiveWindow("did:plc:different-streamer")) require.ErrorContains(t, mm.ValidateReplicatedMP4(ctx, bytes.NewReader(segment), ""), "no named streamer")}
// TestFeedLiveWindow signs the fixture per-segment, folds the resulting bare// .m4s into a MediaManager live-HLS window via feedLiveWindow, and confirms the// window is populated per track with retrievable signed segments + valid// playlists — the feed path ValidateMP4 drives for every validated segment.func TestFeedLiveWindow(t *testing.T) { ctx := context.Background() ms := newBareSegmentSigner(t) frag, err := os.ReadFile(getFixture("h264-opus-frag.mp4")) require.NoError(t, err)
eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() var m4s []byte for ev := range eventCh { if ev.Type != "signed-segment" { continue } tids := make([]string, 0, len(ev.Tracks)) for tid := range ev.Tracks { tids = append(tids, tid) } sort.Strings(tids) for _, tid := range tids { m4s = append(m4s, ev.Tracks[tid]...) } } require.NoError(t, <-errCh) require.NotEmpty(t, m4s)
mm := &MediaManager{liveWindows: map[string]*livehls.Writer{}}
// Pre-live (unpublished) segments are folded in for the streamer's own // preview, and the window remembers that its latest segment is not // public — the getLive* handlers keep it to holders of a playback token. t0 := time.Now() mm.feedLiveWindow(ctx, "did:test:streamer", m4s, t0, false) require.NotNil(t, mm.GetLiveWindow("did:test:streamer"), "pre-live segments make a window") require.False(t, mm.LiveWindowPublished("did:test:streamer"), "…but it is not public")
preLive := mm.GetLiveWindow("did:test:streamer") mm.feedLiveWindow(ctx, "did:test:streamer", m4s, t0.Add(2*time.Second), true) require.True(t, mm.LiveWindowPublished("did:test:streamer"))
w := mm.GetLiveWindow("did:test:streamer") require.NotNil(t, w, "window created on feed") require.NotSame(t, preLive, w, "going public starts the window over: the preview segments are not served to the public") tids := w.TrackIDs() require.NotEmpty(t, tids, "window has tracks") for _, tid := range tids { require.Len(t, w.Track(tid).Segments, len(preLive.Track(tid).Segments), "track %s: only the published segment, none of the pre-live ones", tid) } // Once public, further segments extend the same window. mm.feedLiveWindow(ctx, "did:test:streamer", m4s, t0.Add(4*time.Second), true) require.Same(t, w, mm.GetLiveWindow("did:test:streamer")) for _, tid := range tids { require.NotEmpty(t, w.InitSegment(tid), "track %s has an init segment", tid) tr := w.Track(tid) require.NotEmpty(t, tr.Segments, "track %s has segments", tid) data := w.SegmentData(tid, tr.Segments[0].Seq) require.GreaterOrEqual(t, len(data), 8) require.Equal(t, "uuid", string(data[4:8]), "track %s segment is the signed .m4s", tid) pl := w.MediaPlaylist(tid, "init.mp4", func(seq uint64) string { return fmt.Sprintf("seg%d.m4s", seq) }) require.Contains(t, pl, "#EXT-X-MAP") require.Contains(t, pl, "#EXTINF") } require.Contains(t, w.MasterPlaylist(func(tid string) string { return tid + ".m3u8" }), "#EXTM3U")
// Segments are fed concurrently, so a pre-live segment can finish // validating after the stream went public. It is older than the window, // so it is dropped: the public window neither restarts nor turns back // into a preview under its viewers. segs := len(w.Track(tids[0]).Segments) mm.feedLiveWindow(ctx, "did:test:streamer", m4s, t0.Add(time.Second), false) require.Same(t, w, mm.GetLiveWindow("did:test:streamer"), "a late pre-live segment does not touch the public window") require.True(t, mm.LiveWindowPublished("did:test:streamer"), "…and does not make it a preview") require.Len(t, w.Track(tids[0]).Segments, segs, "…nor does it add to it")
// A pre-live segment newer than the window is the stream going back to // preview (the streamer ended it but is still sending), and does flip it. mm.feedLiveWindow(ctx, "did:test:streamer", m4s, t0.Add(6*time.Second), false) require.False(t, mm.LiveWindowPublished("did:test:streamer"), "a newer pre-live segment takes the stream back to preview")}
// seedBanLabel writes an active ban label for did into the model.func seedBanLabel(t *testing.T, mod model.Model, did string) { t.Helper() lex := &comatproto.LabelDefs_Label{ Cts: time.Now().UTC().Format(time.RFC3339), Src: "did:plc:test-labeler", Uri: did, Val: atproto.LabelDMCAViolation, } var buf bytes.Buffer require.NoError(t, lex.MarshalCBOR(&buf)) require.NoError(t, mod.CreateLabel(&model.Label{ Src: lex.Src, Uri: did, Val: atproto.LabelDMCAViolation, Record: buf.Bytes(), }))}
// TestStreamerIsBanned exercises the defense-in-depth gate validateSource applies// at the ingest chokepoint: a streamer with an active ban label is rejected// regardless of whether their ingest worker was torn down. A clean streamer// passes; the ONLY change between the two checks is the ban label.func TestStreamerIsBanned(t *testing.T) { mm, _ := getStaticTestMediaManager(t) did := "did:plc:bannedstreamer"
banned, err := mm.streamerIsBanned(did) require.NoError(t, err) require.False(t, banned, "a clean streamer is not banned")
seedBanLabel(t, mm.model, did)
banned, err = mm.streamerIsBanned(did) require.NoError(t, err) require.True(t, banned, "an active ban label is detected at the validate chokepoint")}