diff --git a/pkg/media/dynamic_manifest_test.go b/pkg/media/dynamic_manifest_test.go new file mode 100644 index 00000000..3f387ada --- /dev/null +++ b/pkg/media/dynamic_manifest_test.go @@ -0,0 +1,150 @@ +package media + +import ( + "bytes" + "context" + "encoding/json" + "os" + "strings" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/muxl" +) + +// stubManifester returns one manifest for the first BuildManifest call and a +// different one for every call after — the smallest possible model of a +// pre-live → live transition for SignSegmentStream to react to. +type stubManifester struct { + calls atomic.Int32 + preLive []byte + live []byte +} + +func (s *stubManifester) BuildManifest(_ context.Context, _ string, _ int64) ([]byte, error) { + if s.calls.Add(1) == 1 { + return s.preLive, nil + } + return s.live, nil +} + +// TestSignSegmentStreamRefreshesManifestPerSegment is the pre-live → live +// regression test: a long-lived SignSegmentStream call must consult the +// manifest builder for EVERY GoP, so a livestream record flipping from +// EndedAt-set (no c2pa.published) to EndedAt-nil (c2pa.published) lands in +// subsequent segments without restarting the wasm signer. +// +// Pre-fix, manifestBs was built once at SignSegmentStream entry and frozen for +// the whole RTMP session: this test would observe BuildManifest fire exactly +// once, and every signed segment would carry the pre-live manifest. +func TestSignSegmentStreamRefreshesManifestPerSegment(t *testing.T) { + ctx := context.Background() + ms := newBareSegmentSigner(t) + // Drop PrebuiltManifest so the stub manifestBuilder wins the dispatch in + // buildManifest (the production code path the fix targets). + ms.PrebuiltManifest = nil + stub := &stubManifester{ + preLive: []byte(`{ + "title":"pre-live", + "assertions":[ + {"label":"c2pa.actions","data":{"actions":[{"action":"c2pa.created"}]}}, + {"label":"cawg.metadata","data":{ + "@context":{"dc":"http://purl.org/dc/elements/1.1/"}, + "dc:creator":"did:example","dc:title":"pre-live", + "dc:date":"1970-01-01T00:00:00.000Z" + }} + ] + }`), + live: []byte(`{ + "title":"live", + "assertions":[ + {"label":"c2pa.actions","data":{"actions":[ + {"action":"c2pa.created"}, + {"action":"c2pa.published"} + ]}}, + {"label":"cawg.metadata","data":{ + "@context":{"dc":"http://purl.org/dc/elements/1.1/"}, + "dc:creator":"did:example","dc:title":"live", + "dc:date":"1970-01-01T00:00:00.000Z" + }} + ] + }`), + } + ms.manifestBuilder = stub + + 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) + close(eventCh) + errCh <- err + }() + + var gops [][]byte // per-GoP bare .m4s + for ev := range eventCh { + if ev.Type != "signed-segment" { + continue + } + gops = append(gops, concatTracksSorted(ev.Tracks)) + } + require.NoError(t, <-errCh) + require.GreaterOrEqual(t, len(gops), 2, + "fixture must produce at least two GoPs to exercise the per-segment refresh") + + // Two of the three buildManifest paths matter here: + // - one call per GoP (the property the fix introduces); + // - the WrapperManifestFn is wired in muxl too, but the wasm signer + // currently doesn't ask the host for kind=1 (wrapper). So we expect + // exactly len(gops) calls — same as the GoP count. + require.Equal(t, int32(len(gops)), stub.calls.Load(), + "BuildManifest must be called once per GoP") + + // First GoP: pre-live (c2pa.created only). Later GoPs: live (c2pa.published + // added). The actions live inside the SIGNED claim, so we re-verify each + // bare .m4s in-wasm and read them back from the verify JSON. + for i, m4s := range gops { + out, err := muxl.RunMuxlVerify(ctx, bytes.NewReader(m4s)) + require.NoError(t, err, "GoP %d verify", i) + require.NotContains(t, out, `"validation_state":"Invalid"`, + "GoP %d must validate (verify JSON: %s)", i, out) + + var doc struct { + Segments []struct { + Manifest struct { + Assertions []struct { + Label string `json:"label"` + Data struct { + Actions []struct { + Action string `json:"action"` + } `json:"actions"` + } `json:"data"` + } `json:"assertions"` + } `json:"manifest"` + } `json:"segments"` + } + require.NoError(t, json.Unmarshal([]byte(out), &doc), "GoP %d JSON", i) + require.NotEmpty(t, doc.Segments, "GoP %d has segments", i) + + hasPublished := false + for _, seg := range doc.Segments { + for _, a := range seg.Manifest.Assertions { + if !strings.HasPrefix(a.Label, "c2pa.actions") { + continue + } + for _, act := range a.Data.Actions { + if act.Action == "c2pa.published" { + hasPublished = true + } + } + } + } + wantPublished := i > 0 + require.Equal(t, wantPublished, hasPublished, + "GoP %d: published=%t expected %t — manifest did not refresh between segments", + i, hasPublished, wantPublished) + } +} diff --git a/pkg/media/media_signer.go b/pkg/media/media_signer.go index f7e45aa5..e94e5dad 100644 --- a/pkg/media/media_signer.go +++ b/pkg/media/media_signer.go @@ -39,6 +39,13 @@ type MediaSigner interface { var DoReplay = false +// Manifester builds the C2PA manifest JSON SignSegmentStream embeds in each +// signed GoP. Production uses *ManifestBuilder (model-driven); tests can plug +// in a stub to drive mid-stream manifest changes. +type Manifester interface { + BuildManifest(ctx context.Context, streamerName string, start int64) ([]byte, error) +} + type MediaSignerLocal struct { StreamerName string Signer crypto.Signer @@ -46,7 +53,7 @@ type MediaSignerLocal struct { Cert []byte TAURL string did string - manifestBuilder *ManifestBuilder + manifestBuilder Manifester PrebuiltManifest []byte // Optional: use this manifest instead of building one sigs [][]byte } @@ -131,26 +138,35 @@ func (ms *MediaSignerLocal) buildManifest(ctx context.Context, start int64) ([]b } // SignSegmentStream streams an fMP4 input through muxl-sign's per-segment -// signer, emitting one signed-segment event per GoP on eventCh. The manifest -// is built once for the stream; muxl-sign stamps each segment's signing time -// into cawg.metadata/dc:date as it signs. Signing backend: an -// *ecdsa.PrivateKey is marshaled to PEM and signed in-wasm, otherwise the -// host-callback path keeps the key out of the sandbox. +// signer, emitting one signed-segment event per GoP on eventCh. +// +// The manifest is fetched FRESH from buildManifest once per GoP via muxl's +// host_get_manifest callback, so connection-lifetime fields (livestream +// EndedAt → c2pa.published, title, metadata config) update mid-stream without +// needing to terminate the RTMP session. Before this, the manifest was sealed +// at connection start and a pre-live → live transition stayed invisible until +// the streamer reconnected. +// +// muxl-sign stamps each segment's signing time into cawg.metadata/dc:date as +// it signs. Signing backend: an *ecdsa.PrivateKey is marshaled to PEM and +// signed in-wasm, otherwise the host-callback path keeps the key out of the +// sandbox. func (ms *MediaSignerLocal) SignSegmentStream(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { ctx, span := signerTracer.Start(ctx, "SignSegmentStream", trace.WithAttributes( attribute.String("streamer", ms.StreamerName), )) defer span.End() - manifestBs, err := ms.buildManifest(ctx, time.Now().UnixMilli()) - if err != nil { - return fmt.Errorf("failed to build manifest: %w", err) + // One callback shared by both kinds — track and wrapper manifests are the + // same JSON in Streamplace today, so a single buildManifest call per GoP + // covers both. If they ever diverge we split this in two. + fetchManifest := func() ([]byte, error) { + return ms.buildManifest(ctx, time.Now().UnixMilli()) } - in := muxl.SignerInput{ - CertPEM: ms.Cert, - TrackManifest: manifestBs, - WrapperManifest: manifestBs, + CertPEM: ms.Cert, + TrackManifestFn: fetchManifest, + WrapperManifestFn: fetchManifest, } if _, ok := ms.Signer.(*ecdsa.PrivateKey); ok { keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer)