From 844f13f9e45cd2c110e5874a0141bb5289f559aa Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Thu, 28 May 2026 10:31:42 -0700 Subject: [PATCH] media: refresh the streaming signer's manifest per GoP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-live → live got stuck in production: a viewer waited for the publisher to go live and the signed segments still carried the "unpublished" manifest (no c2pa.published action). The only thing that unstuck it was terminating the RTMP connection so the reconnect built a fresh manifest. SignSegmentStream was the culprit. The v0.10.34 MUXL refactor (commit 928a51c3) consolidated signing into one streaming wasm call per RTMP session. We built the manifest exactly once at SignSegmentStream entry and passed it as the static SignerInput.TrackManifest / WrapperManifest — so livestream-record state (EndedAt-driven c2pa.published, title, metadata config) froze at connection time and only updated on reconnect. Switch to the dynamic-manifest callback muxl just gained: a closure that calls ms.buildManifest fresh on every GoP. Track and wrapper get the same closure (Streamplace passes the same JSON to both). buildManifest already does the right DB lookups, so a livestream record flipping mid- stream is reflected in the next signed segment — no reconnect needed. The cost is one model.GetLatestLivestreamForRepo + one GetMetadataConfiguration per signed segment (~1 / second / live stream). Small PK lookups; we'll layer a cache here if it ever bites. manifestBuilder on MediaSignerLocal becomes a `Manifester` interface (*ManifestBuilder still satisfies it) so a test can plug in a stub. New TestSignSegmentStreamRefreshesManifestPerSegment confirms BuildManifest is invoked once per GoP, and that switching its return value from "pre- live" to "live" mid-stream lands the c2pa.published action in every later segment's verify output — the exact path the production bug exercised. Co-Authored-By: Claude Opus 4.7 (1M context) --- pkg/media/dynamic_manifest_test.go | 150 +++++++++++++++++++++++++++++ pkg/media/media_signer.go | 42 +++++--- 2 files changed, 179 insertions(+), 13 deletions(-) create mode 100644 pkg/media/dynamic_manifest_test.go 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) -- 2.51.2