determinism: introduce deterministic muxing

Cargo.lock

···
 ]
 
 [[package]]
+name = "c2pa"
+version = "0.58.0"
+source = "git+https://github.com/streamplace/c2pa-rs.git?rev=544825abfaf9e588813e18dba70c8d5afd039d46#544825abfaf9e588813e18dba70c8d5afd039d46"
+dependencies = [
+ "asn1-rs",
+ "async-generic",
+ "async-recursion",
+ "async-trait",
+ "atree",
+ "base64 0.22.1",
+ "bcder",
+ "byteorder",
+ "byteordered",
+ "bytes",
+ "chrono",
+ "ciborium",
+ "config",
+ "console_log",
+ "const-hex",
+ "const-oid 0.9.6",
+ "conv",
+ "coset",
+ "der 0.7.10",
+ "ed25519-dalek 2.2.0",
+ "env_logger",
+ "extfmt",
+ "getrandom 0.2.16",
+ "hex",
+ "hex-literal",
+ "http",
+ "id3",
+ "img-parts",
+ "iref",
+ "jfifdump",
+ "js-sys",
+ "lazy_static",
+ "log",
+ "memchr",
+ "mp4",
+ "nom",
+ "non-empty-string",
+ "nonempty-collections",
+ "num-bigint-dig",
+ "openssl",
+ "pem",
+ "pkcs1",
+ "pkcs8 0.10.2",
+ "png_pong",
+ "quick-xml",
+ "rand 0.8.5",
+ "rand_chacha 0.3.1",
+ "rand_core 0.9.3",
+ "range-set",
+ "rasn",
+ "rasn-cms",
+ "rasn-ocsp",
+ "rasn-pkix",
+ "regex",
+ "reqwest",
+ "riff",
+ "ring",
+ "rsa",
+ "serde",
+ "serde-transcode",
+ "serde-wasm-bindgen",
+ "serde_bytes",
+ "serde_cbor",
+ "serde_derive",
+ "serde_json",
+ "serde_with",
+ "sha1",
+ "sha2 0.10.9",
+ "spki 0.7.3",
+ "static-iref",
+ "tempfile",
+ "thiserror 2.0.16",
+ "toml 0.8.23",
+ "treeline",
+ "ureq",
+ "url",
+ "uuid",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+ "web-time",
+ "windows-core",
+ "wstd",
+ "x509-certificate",
+ "x509-parser",
+ "zeroize",
+ "zip",
+]
+
+[[package]]
 name = "camino"
 version = "1.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 version = "0.0.0"
 dependencies = [
  "anyhow",
- "c2pa",
+ "c2pa 0.58.0 (git+https://github.com/hyphacoop/c2pa-rs.git?rev=1b84d40219b27340a30fc309250e774e8a7b7761)",
  "schemars 0.8.22",
  "serde",
  "serde_json",
···
  "anyhow",
  "async-trait",
  "bytes",
- "c2pa",
+ "c2pa 0.58.0 (git+https://github.com/streamplace/c2pa-rs.git?rev=544825abfaf9e588813e18dba70c8d5afd039d46)",
  "hex",
  "iroh",
  "iroh-base",

hack/deterministic-mux.sh

···
+#!/bin/bash
+
+set -euo pipefail
+
+TMPDIR=$(mktemp -d)
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+$SCRIPT_DIR/../build-linux-amd64/streamplace clip --out "$TMPDIR/combined1.mp4" "$@"
+sleep 2
+$SCRIPT_DIR/../build-linux-amd64/streamplace clip --out "$TMPDIR/combined2.mp4" "$@"
+xxd "$TMPDIR/combined1.mp4" > "$TMPDIR/combined1.mp4.xxd"
+xxd "$TMPDIR/combined2.mp4" > "$TMPDIR/combined2.mp4.xxd"
+diff --color=always "$TMPDIR/combined1.mp4.xxd" "$TMPDIR/combined2.mp4.xxd"

pkg/api/stream_key.go

···
 		return nil, err
 	}
 
-	var mediaSigner media.MediaSigner
-	if a.CLI.ExternalSigning {
-		mediaSigner, err = media.MakeMediaSignerExt(ctx, a.CLI, did, addrBytes, a.Model)
-	} else {
-		mediaSigner, err = media.MakeMediaSigner(ctx, a.CLI, did, signer, a.Model)
-	}
+	mediaSigner, err := media.MakeMediaSigner(ctx, a.CLI, did, signer, a.Model)
 	if err != nil {
 		return nil, fmt.Errorf("invalid authorization key (not valid secp256k1): %w", err)
 	}

pkg/cmd/streamplace.go

···
 		return Clip(ctx, fs.Args(), *out)
 	}
 
+	if len(os.Args) > 1 && os.Args[1] == "segment" {
+		cli := config.CLI{Build: build}
+		fs := cli.NewFlagSet("streamplace clip")
+		out := fs.String("out-dir", "", "output directory")
+
+		err := cli.Parse(fs, os.Args[2:])
+		if err != nil {
+			return err
+		}
+		ctx := context.Background()
+		ctx = log.WithDebugValue(ctx, cli.Debug)
+		return media.SegmentFile(ctx, fs.Args()[0], *out)
+	}
+
 	if len(os.Args) > 1 && os.Args[1] == "self-test" {
 		err := media.RunSelfTest(context.Background())
 		if err != nil {

pkg/config/config.go

···
 	fs.BoolVar(&cli.PrintChat, "print-chat", false, "print chat messages to stdout")
 	fs.StringVar(&cli.WHIPTest, "whip-test", "", "run a WHIP self-test with the given parameters")
 	fs.StringVar(&cli.RelayHost, "relay-host", "wss://bsky.network", "websocket url for relay firehose")
-	fs.Bool("insecure", false, "DEPRECATED, does nothing.")
 	fs.StringVar(&cli.Color, "color", "", "'true' to enable colorized logging, 'false' to disable")
 	fs.StringVar(&cli.BroadcasterHost, "broadcaster-host", "", "public host for the broadcaster group that this node is a part of (excluding https:// e.g. stream.place)")
 	fs.StringVar(&cli.XXDeprecatedPublicHost, "public-host", "", "deprecated, use broadcaster-host or server-host instead as appropriate")
 	fs.StringVar(&cli.ServerHost, "server-host", "", "public host for this particular physical streamplace node. defaults to broadcaster-host and only must be set for multi-node broadcasters")
 	fs.BoolVar(&cli.Thumbnail, "thumbnail", true, "enable thumbnail generation")
 	fs.BoolVar(&cli.SmearAudio, "smear-audio", false, "enable audio smearing to create 'perfect' segment timestamps")
-	fs.BoolVar(&cli.ExternalSigning, "external-signing", false, "enable external signing via exec (prevents potential memory leak)")
 	fs.StringVar(&cli.TracingEndpoint, "tracing-endpoint", "", "gRPC endpoint to send traces to")
 	fs.IntVar(&cli.RateLimitPerSecond, "rate-limit-per-second", 0, "rate limit for requests per second per ip")
 	fs.IntVar(&cli.RateLimitBurst, "rate-limit-burst", 0, "rate limit burst for requests per ip")
···
 	cli.StringSliceFlag(fs, &cli.Replicators, "replicators", []string{ReplicatorWebsocket}, "list of replication protocols to use (http, iroh)")
 	fs.StringVar(&cli.WebsocketURL, "websocket-url", "", "override the websocket (ws:// or wss://) url to use for replication (normally not necessary, used for testing)")
 	fs.BoolVar(&cli.BehindHTTPSProxy, "behind-https-proxy", false, "set to true if this node is behind an https proxy and we should report https URLs even though the node isn't serving HTTPS")
+
+	fs.Bool("external-signing", true, "DEPRECATED, does nothing.")
+	fs.Bool("insecure", false, "DEPRECATED, does nothing.")
 
 	lpFlags := flag.NewFlagSet("livepeer", flag.ContinueOnError)
 	_ = starter.NewLivepeerConfig(lpFlags)

pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go

···
 	}
 	{
 		checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t {
+			return C.uniffi_iroh_streamplace_checksum_func_get_manifests()
+		})
+		if checksum != 17 {
+			// If this happens try cleaning and rebuilding your project
+			panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_get_manifests: UniFFI API checksum mismatch")
+		}
+	}
+	{
+		checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t {
 			return C.uniffi_iroh_streamplace_checksum_func_init_logging()
 		})
 		if checksum != 40911 {
···
 	}
 	{
 		checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t {
+			return C.uniffi_iroh_streamplace_checksum_func_resign()
+		})
+		if checksum != 33363 {
+			// If this happens try cleaning and rebuilding your project
+			panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_resign: UniFFI API checksum mismatch")
+		}
+	}
+	{
+		checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t {
 			return C.uniffi_iroh_streamplace_checksum_func_sign()
 		})
 		if checksum != 23786 {
 			// If this happens try cleaning and rebuilding your project
 			panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_sign: UniFFI API checksum mismatch")
+		}
+	}
+	{
+		checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t {
+			return C.uniffi_iroh_streamplace_checksum_func_sign_with_ingredients()
+		})
+		if checksum != 63680 {
+			// If this happens try cleaning and rebuilding your project
+			panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_sign_with_ingredients: UniFFI API checksum mismatch")
 		}
 	}
 	{
···
 	}
 }
 
+type FfiConverterSequenceBytes struct{}
+
+var FfiConverterSequenceBytesINSTANCE = FfiConverterSequenceBytes{}
+
+func (c FfiConverterSequenceBytes) Lift(rb RustBufferI) [][]byte {
+	return LiftFromRustBuffer[[][]byte](c, rb)
+}
+
+func (c FfiConverterSequenceBytes) Read(reader io.Reader) [][]byte {
+	length := readInt32(reader)
+	if length == 0 {
+		return nil
+	}
+	result := make([][]byte, 0, length)
+	for i := int32(0); i < length; i++ {
+		result = append(result, FfiConverterBytesINSTANCE.Read(reader))
+	}
+	return result
+}
+
+func (c FfiConverterSequenceBytes) Lower(value [][]byte) C.RustBuffer {
+	return LowerIntoRustBuffer[[][]byte](c, value)
+}
+
+func (c FfiConverterSequenceBytes) Write(writer io.Writer, value [][]byte) {
+	if len(value) > math.MaxInt32 {
+		panic("[][]byte is too large to fit into Int32")
+	}
+
+	writeInt32(writer, int32(len(value)))
+	for _, item := range value {
+		FfiConverterBytesINSTANCE.Write(writer, item)
+	}
+}
+
+type FfiDestroyerSequenceBytes struct{}
+
+func (FfiDestroyerSequenceBytes) Destroy(sequence [][]byte) {
+	for _, value := range sequence {
+		FfiDestroyerBytes{}.Destroy(value)
+	}
+}
+
 type FfiConverterSequencePublicKey struct{}
 
 var FfiConverterSequencePublicKeyINSTANCE = FfiConverterSequencePublicKey{}
···
 	}
 }
 
+func GetManifests(data []byte) (string, error) {
+	_uniffiRV, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) RustBufferI {
+		return GoRustBuffer{
+			inner: C.uniffi_iroh_streamplace_fn_func_get_manifests(FfiConverterBytesINSTANCE.Lower(data), _uniffiStatus),
+		}
+	})
+	if _uniffiErr != nil {
+		var _uniffiDefaultValue string
+		return _uniffiDefaultValue, _uniffiErr
+	} else {
+		return FfiConverterStringINSTANCE.Lift(_uniffiRV), nil
+	}
+}
+
 // Initialize logging with the default subscriber that respects RUST_LOG environment variable.
 // This function is safe to call multiple times - it will only initialize logging once.
 func InitLogging() {
···
 	}
 }
 
+func Resign(unsignedSegLabel string, unsignedSegData []byte, signedConcatData []byte, certs []byte, gosigner GoSigner) ([]byte, error) {
+	_uniffiRV, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) RustBufferI {
+		return GoRustBuffer{
+			inner: C.uniffi_iroh_streamplace_fn_func_resign(FfiConverterStringINSTANCE.Lower(unsignedSegLabel), FfiConverterBytesINSTANCE.Lower(unsignedSegData), FfiConverterBytesINSTANCE.Lower(signedConcatData), FfiConverterBytesINSTANCE.Lower(certs), FfiConverterGoSignerINSTANCE.Lower(gosigner), _uniffiStatus),
+		}
+	})
+	if _uniffiErr != nil {
+		var _uniffiDefaultValue []byte
+		return _uniffiDefaultValue, _uniffiErr
+	} else {
+		return FfiConverterBytesINSTANCE.Lift(_uniffiRV), nil
+	}
+}
+
 func Sign(manifest string, data []byte, certs []byte, gosigner GoSigner) ([]byte, error) {
 	_uniffiRV, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) RustBufferI {
 		return GoRustBuffer{
 			inner: C.uniffi_iroh_streamplace_fn_func_sign(FfiConverterStringINSTANCE.Lower(manifest), FfiConverterBytesINSTANCE.Lower(data), FfiConverterBytesINSTANCE.Lower(certs), FfiConverterGoSignerINSTANCE.Lower(gosigner), _uniffiStatus),
+		}
+	})
+	if _uniffiErr != nil {
+		var _uniffiDefaultValue []byte
+		return _uniffiDefaultValue, _uniffiErr
+	} else {
+		return FfiConverterBytesINSTANCE.Lift(_uniffiRV), nil
+	}
+}
+
+func SignWithIngredients(manifest string, data []byte, certs []byte, ingredients [][]byte, gosigner GoSigner) ([]byte, error) {
+	_uniffiRV, _uniffiErr := rustCallWithError[SpError](FfiConverterSpError{}, func(_uniffiStatus *C.RustCallStatus) RustBufferI {
+		return GoRustBuffer{
+			inner: C.uniffi_iroh_streamplace_fn_func_sign_with_ingredients(FfiConverterStringINSTANCE.Lower(manifest), FfiConverterBytesINSTANCE.Lower(data), FfiConverterBytesINSTANCE.Lower(certs), FfiConverterSequenceBytesINSTANCE.Lower(ingredients), FfiConverterGoSignerINSTANCE.Lower(gosigner), _uniffiStatus),
 		}
 	})
 	if _uniffiErr != nil {

pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h

···
 RustBuffer uniffi_iroh_streamplace_fn_func_get_manifest_and_cert(RustBuffer data, RustCallStatus *out_status
 );
 #endif
+#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_GET_MANIFESTS
+#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_GET_MANIFESTS
+RustBuffer uniffi_iroh_streamplace_fn_func_get_manifests(RustBuffer data, RustCallStatus *out_status
+);
+#endif
 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_INIT_LOGGING
 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_INIT_LOGGING
 void uniffi_iroh_streamplace_fn_func_init_logging(RustCallStatus *out_status
···
 void* uniffi_iroh_streamplace_fn_func_node_id_from_ticket(RustBuffer ticket_str, RustCallStatus *out_status
 );
 #endif
+#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_RESIGN
+#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_RESIGN
+RustBuffer uniffi_iroh_streamplace_fn_func_resign(RustBuffer unsigned_seg_label, RustBuffer unsigned_seg_data, RustBuffer signed_concat_data, RustBuffer certs, void* gosigner, RustCallStatus *out_status
+);
+#endif
 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SIGN
 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SIGN
 RustBuffer uniffi_iroh_streamplace_fn_func_sign(RustBuffer manifest, RustBuffer data, RustBuffer certs, void* gosigner, RustCallStatus *out_status
+);
+#endif
+#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SIGN_WITH_INGREDIENTS
+#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SIGN_WITH_INGREDIENTS
+RustBuffer uniffi_iroh_streamplace_fn_func_sign_with_ingredients(RustBuffer manifest, RustBuffer data, RustBuffer certs, RustBuffer ingredients, void* gosigner, RustCallStatus *out_status
 );
 #endif
 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SUBSCRIBE_ITEM_DEBUG
···
     
 );
 #endif
+#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_GET_MANIFESTS
+#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_GET_MANIFESTS
+uint16_t uniffi_iroh_streamplace_checksum_func_get_manifests(void
+    
+);
+#endif
 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_INIT_LOGGING
 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_INIT_LOGGING
 uint16_t uniffi_iroh_streamplace_checksum_func_init_logging(void
···
     
 );
 #endif
+#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_RESIGN
+#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_RESIGN
+uint16_t uniffi_iroh_streamplace_checksum_func_resign(void
+    
+);
+#endif
 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_SIGN
 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_SIGN
 uint16_t uniffi_iroh_streamplace_checksum_func_sign(void
+    
+);
+#endif
+#ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_SIGN_WITH_INGREDIENTS
+#define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_SIGN_WITH_INGREDIENTS
+uint16_t uniffi_iroh_streamplace_checksum_func_sign_with_ingredients(void
     
 );
 #endif

pkg/media/clip.go

···
 	defer cancel()
 
 	pipelineSlice := []string{
-		"mp4mux faststart=true name=muxer ! appsink sync=false name=mp4sink",
-		"h264parse name=videoparse ! h264timestamper ! muxer.video_0",
+		"mp4mux name=muxer ! appsink sync=false name=mp4sink",
+		"h264parse name=videoparse ! muxer.video_0",
 		"opusparse name=audioparse ! muxer.audio_0",
 	}
 

pkg/media/deterministic_muxing_test.go

···
+package media
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"stream.place/streamplace/test/remote"
+)
+
+var muxTestCount = 2
+
+func TestDeterministicMuxing(t *testing.T) {
+	withNoGSTLeaks(t, func() {
+		tempDir, err := os.MkdirTemp("", "deterministic_muxing_test")
+		require.NoError(t, err)
+
+		startFile := remote.RemoteArchive("14ba49843a56c0510e2b5059123abd2f98a502b1f4c7d706b0ae1066d438468c/BigBuckBunny_1sGOP_4kp60_NoBframes.1min.tar.gz")
+
+		require.NoError(t, err)
+		for i := 0; i < muxTestCount; i++ {
+			splitAndCombineTest(t, tempDir, startFile)
+			require.NoError(t, err)
+		}
+	})
+}
+
+func splitAndCombineTest(t *testing.T, tempDir string, inputDir string) string {
+	var err error
+	tempDir, err = os.MkdirTemp(tempDir, "splitAndCombineTest")
+	require.NoError(t, err)
+
+	firstReport, err := makeSegDirReport(t, inputDir)
+	require.NoError(t, err)
+
+	combinedHashes := []string{}
+	combinedFiles := []string{}
+	for i := 0; i < muxTestCount; i++ {
+		outFilePath := filepath.Join(tempDir, fmt.Sprintf("combined_%d.mp4", i))
+		combinedFiles = append(combinedFiles, outFilePath)
+		outFile, err := os.Create(outFilePath)
+		require.NoError(t, err)
+		defer outFile.Close()
+		err = Clip(context.Background(), firstReport.Segs, outFile)
+		require.NoError(t, err)
+		hash, err := hashFile(outFilePath)
+		require.NoError(t, err)
+		combinedHashes = append(combinedHashes, hash)
+	}
+
+	for _, hash := range combinedHashes {
+		require.Equal(t, hash, combinedHashes[0])
+	}
+
+	for i := 0; i < muxTestCount; i++ {
+		segDir, err := os.MkdirTemp(tempDir, "segs")
+		require.NoError(t, err)
+		err = SegmentFile(context.Background(), combinedFiles[0], segDir)
+		require.NoError(t, err)
+		report, err := makeSegDirReport(t, segDir)
+		require.NoError(t, err)
+		require.NoError(t, report.CheckEquals(firstReport), "round-trip muxing is not deterministic")
+	}
+
+	return combinedFiles[0]
+}
+
+type SegDirReport struct {
+	Dir    string
+	Segs   []string
+	Hashes []string
+}
+
+func makeSegDirReport(t *testing.T, segDir string) (*SegDirReport, error) {
+	segs := []string{}
+	segEntries, err := os.ReadDir(segDir)
+	require.NoError(t, err)
+	for _, segEntry := range segEntries {
+		if segEntry.Type().IsRegular() {
+			segPath := filepath.Join(segDir, segEntry.Name())
+			segs = append(segs, segPath)
+		}
+	}
+	sort.Strings(segs)
+	hashes := make([]string, len(segs))
+	for i, segPath := range segs {
+		hash, err := hashFile(segPath)
+		if err != nil {
+			return nil, err
+		}
+		hashes[i] = hash
+	}
+
+	return &SegDirReport{
+		Dir:    segDir,
+		Segs:   segs,
+		Hashes: hashes,
+	}, nil
+}
+
+func (s *SegDirReport) Equals(other *SegDirReport) bool {
+	if len(s.Segs) != len(other.Segs) {
+		return false
+	}
+	if len(s.Hashes) != len(other.Hashes) {
+		return false
+	}
+	return reflect.DeepEqual(s.Hashes, other.Hashes)
+}
+
+func hashFile(path string) (string, error) {
+	bs, err := os.ReadFile(path)
+	if err != nil {
+		return "", err
+	}
+	hash := sha256.Sum256(bs)
+	return fmt.Sprintf("%x", hash), nil
+}
+
+func (s *SegDirReport) CheckEquals(other *SegDirReport) error {
+	if !s.Equals(other) {
+		str1 := s.ToString()
+		str2 := other.ToString()
+		return fmt.Errorf("files should be equal: %s\n%s", str1, str2)
+	}
+	return nil
+}
+
+func (s *SegDirReport) ToString() string {
+	bs, err := json.MarshalIndent(s, "", "  ")
+	if err != nil {
+		panic(err)
+	}
+	return string(bs)
+}

pkg/media/ingredient_test.go

···
+package media
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/bluesky-social/indigo/util"
+	"github.com/stretchr/testify/require"
+	c2patypes "stream.place/streamplace/pkg/c2patypes"
+	"stream.place/streamplace/pkg/config"
+	ct "stream.place/streamplace/pkg/config/configtesting"
+	"stream.place/streamplace/pkg/crypto/spkey"
+	"stream.place/streamplace/pkg/iroh/generated/iroh_streamplace"
+	"stream.place/streamplace/test/remote"
+)
+
+var testTimestamp = "2025-01-01T00:00:00.000Z"
+
+type ManifestResult struct {
+	Manifests map[string]c2patypes.Manifest `json:"manifests"`
+}
+
+type ManifestAndMetadata struct {
+	Manifest        c2patypes.Manifest
+	SegmentMetadata *SegmentMetadata
+}
+
+func TestIngredientConcat(t *testing.T) {
+	withNoGSTLeaks(t, func() {
+		tempDir, err := os.MkdirTemp("", "ingredient_test")
+		require.NoError(t, err)
+		// defer os.RemoveAll(tempDir)
+		segments := remote.RemoteArchive("14ba49843a56c0510e2b5059123abd2f98a502b1f4c7d706b0ae1066d438468c/BigBuckBunny_1sGOP_4kp60_NoBframes.1min.tar.gz")
+		// segments := "/home/iameli/testvids/three"
+		testVids := []string{}
+		sort.Strings(testVids)
+		segEntries, err := os.ReadDir(segments)
+		require.NoError(t, err)
+		for _, segEntry := range segEntries {
+			if segEntry.Type().IsRegular() {
+				testVids = append(testVids, filepath.Join(segments, segEntry.Name()))
+			}
+		}
+		firstReport, err := makeSegDirReport(t, segments)
+		require.NoError(t, err)
+		priv, _, err := spkey.GenerateStreamKey()
+		require.NoError(t, err)
+		signer, err := spkey.KeyToSigner(priv)
+		require.NoError(t, err)
+		cli := ct.CLI(t, &config.CLI{
+			TAURL:    "http://timestamp.digicert.com",
+			WideOpen: true,
+		})
+		msInterface, err := MakeMediaSigner(context.Background(), cli, "test-person", signer, nil)
+		require.NoError(t, err)
+		ms := msInterface.(*MediaSignerLocal)
+		buf := bytes.Buffer{}
+		err = Clip(context.Background(), testVids, &buf)
+		require.NoError(t, err)
+		ingredients := [][]byte{}
+		startTS, err := time.Parse(util.ISO8601, testTimestamp)
+		require.NoError(t, err)
+		signedSegDir := makeTestSubdir(t, tempDir, "signed-segments")
+		for i, vid := range testVids {
+			ts := startTS.Add(time.Duration(i) * time.Second)
+			bs, err := os.ReadFile(vid)
+			require.NoError(t, err)
+			signedBS, err := ms.SignMP4(context.Background(), bytes.NewReader(bs), ts.UnixMilli())
+			require.NoError(t, err)
+			ingredients = append(ingredients, signedBS)
+			err = os.WriteFile(filepath.Join(signedSegDir, fmt.Sprintf("signed_%06d.mp4", i)), signedBS, 0644)
+			require.NoError(t, err)
+		}
+		signedReport, err := makeSegDirReport(t, signedSegDir)
+		require.NoError(t, err)
+		signedConcatBS, err := ms.SignConcatMP4(context.Background(), bytes.NewReader(buf.Bytes()), ingredients)
+		require.NoError(t, err)
+		require.Greater(t, len(signedConcatBS), 0)
+		concatSegment := filepath.Join(tempDir, "ingredient-concat.mp4")
+		err = os.WriteFile(concatSegment, signedConcatBS, 0644)
+		require.NoError(t, err)
+		splitSegDir := makeTestSubdir(t, tempDir, "split-segments")
+		err = SegmentFile(context.Background(), concatSegment, splitSegDir)
+		require.NoError(t, err)
+		splitReport, err := makeSegDirReport(t, splitSegDir)
+		require.NoError(t, err)
+		require.NoError(t, splitReport.CheckEquals(firstReport), "split segments are not equal to original segments")
+		DoReplay = true
+		// splitSignedSegDir := makeTestSubdir(t, tempDir, "split-signed-segments")
+		// for i, vid := range splitReport.Segs {
+		// 	ts := startTS.Add(time.Duration(i) * time.Second)
+		// 	bs, err := os.ReadFile(vid)
+		// 	require.NoError(t, err)
+		// 	signedBS, err := ms.SignMP4(context.Background(), bytes.NewReader(bs), ts.UnixMilli())
+		// 	require.NoError(t, err)
+		// 	err = os.WriteFile(filepath.Join(splitSignedSegDir, fmt.Sprintf("signed_%06d.mp4", i)), signedBS, 0644)
+		// 	require.NoError(t, err)
+		// }
+		// signedSplitReport, err := makeSegDirReport(t, splitSignedSegDir)
+		// require.NoError(t, err)
+		// require.NoError(t, signedSplitReport.CheckEquals(signedReport), "split signed segments are not equal to signed segments")
+		manifestsStr, err := iroh_streamplace.GetManifests(signedConcatBS)
+		require.NoError(t, err)
+		var manifests ManifestResult
+		err = json.Unmarshal([]byte(manifestsStr), &manifests)
+		require.NoError(t, err)
+		require.Greater(t, len(manifests.Manifests), 0)
+		manifestList := []ManifestAndMetadata{}
+		for _, manifest := range manifests.Manifests {
+			metadata, err := ParseSegmentAssertions(context.Background(), &manifest)
+			if err == ErrMissingMetadata {
+				continue
+			}
+			require.NoError(t, err)
+			manifestList = append(manifestList, ManifestAndMetadata{
+				Manifest:        manifest,
+				SegmentMetadata: metadata,
+			})
+		}
+		sort.Slice(manifestList, func(i, j int) bool {
+			m1 := manifestList[i]
+			m2 := manifestList[j]
+			return m1.SegmentMetadata.StartTime.Time().Before(m2.SegmentMetadata.StartTime.Time())
+		})
+		signedSplitSegDir := makeTestSubdir(t, tempDir, "signed-split-segments")
+		for i, vid := range testVids {
+			bs, err := os.ReadFile(vid)
+			require.NoError(t, err)
+			rustCallbackSigner := &RustCallbackSigner{
+				Signer: ms.Signer,
+			}
+			fmt.Println("resigning", *manifestList[i].Manifest.Label)
+			signedBS, err := iroh_streamplace.Resign(*manifestList[i].Manifest.Label, bs, signedConcatBS, ms.Cert, rustCallbackSigner)
+			require.NoError(t, err)
+			err = os.WriteFile(filepath.Join(signedSplitSegDir, fmt.Sprintf("signed_%06d.mp4", i)), signedBS, 0644)
+			require.NoError(t, err)
+		}
+		signedSplitReport, err := makeSegDirReport(t, signedSplitSegDir)
+		require.NoError(t, err)
+		require.NoError(t, signedSplitReport.CheckEquals(signedReport), "split signed segments are not equal to signed segments")
+	})
+}
+
+func makeTestSubdir(t *testing.T, tempDir, subdir string) string {
+	subDir := filepath.Join(tempDir, subdir)
+	err := os.MkdirAll(subDir, 0755)
+	require.NoError(t, err)
+	return subDir
+}

pkg/media/media.go

···
 	Livestream            *streamplace.Livestream
 }
 
+var ErrMissingMetadata = errors.New("missing segment metadata")
 var ErrInvalidMetadata = errors.New("invalid segment metadata")
 
 func ParseSegmentAssertions(ctx context.Context, mani *c2patypes.Manifest) (*SegmentMetadata, error) {
···
 		}
 	}
 	if ass == nil {
-		return nil, fmt.Errorf("couldn't find %s assertions", StreamplaceMetadata)
+		return nil, ErrMissingMetadata
 	}
 	proc := ld.NewJsonLdProcessor()
 	options := ld.NewJsonLdOptions("")

pkg/media/media_signer.go

···
 	Pub() aqpub.Pub
 	Streamer() string
 	DID() string
+	SignConcatMP4(ctx context.Context, input io.ReadSeeker, ingredients [][]byte) ([]byte, error)
 }
+
+var DoReplay = false
 
 type MediaSignerLocal struct {
 	StreamerName     string
···
 	did              string
 	manifestBuilder  *ManifestBuilder
 	PrebuiltManifest []byte // Optional: use this manifest instead of building one
+	sigs             [][]byte
 }
 
 func prepareCert(ctx context.Context, cli *config.CLI, signer crypto.Signer) ([]byte, error) {
···
 	spmetrics.SigningDuration.WithLabelValues(ms.StreamerName).Observe(float64(time.Since(startTime).Milliseconds()))
 	return bs, nil
 }
+
+func (ms *MediaSignerLocal) SignConcatMP4(ctx context.Context, input io.ReadSeeker, ingredients [][]byte) ([]byte, error) {
+	startTime := time.Now()
+	ctx, span := otel.Tracer("signer").Start(ctx, "SignMP4")
+	defer span.End()
+	for _, ingredient := range ingredients {
+		_, err := iroh_streamplace.GetManifestAndCert(ingredient)
+		if err != nil {
+			return nil, err
+		}
+	}
+	// title := "livestream"
+	mani := obj{
+		"title": "Livestream Clip",
+		// "assertions": []obj{
+		// 	{
+		// 		"label": "c2pa.actions",
+		// 		"data": obj{
+		// 			"actions": []obj{
+		// 				{"action": "c2pa.created"},
+		// 				{"action": "c2pa.published"},
+		// 			},
+		// 		},
+		// 	},
+		// 	{
+		// 		"label": StreamplaceMetadata,
+		// 		"data": obj{
+		// 			"@context": obj{
+		// 				"dc": "http://purl.org/dc/elements/1.1/",
+		// 			},
+		// 			"dc:creator": ms.StreamerName,
+		// 			"dc:title":   []string{title},
+		// 			"dc:date":    []string{aqtime.FromMillis(start).String()},
+		// 		},
+		// 	},
+		// },
+	}
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_MarshalManifest")
+	manifestBs, err := json.Marshal(mani)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal manifest: %w", err)
+	}
+	var manifest c2patypes.ManifestDefinition
+	err = json.Unmarshal(manifestBs, &manifest)
+	if err != nil {
+		return nil, fmt.Errorf("failed to unmarshal manifest: %w", err)
+	}
+	span.End()
+
+	bs, err := io.ReadAll(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read input: %w", err)
+	}
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_Sign")
+	rustCallbackSigner := &RustCallbackSigner{
+		Signer: ms.Signer,
+	}
+	bs, err = iroh_streamplace.SignWithIngredients(string(manifestBs), bs, ms.Cert, ingredients, rustCallbackSigner)
+	if err != nil {
+		return nil, err
+	}
+	span.End()
+
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_OutputBytes")
+	defer ctx.Done()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get output bytes: %w", err)
+	}
+	span.End()
+	spmetrics.SigningDuration.WithLabelValues(ms.StreamerName).Observe(float64(time.Since(startTime).Milliseconds()))
+	return bs, nil
+}
+
+// don't call externally! this is used as a callback for the rust library
 
 func (ms *MediaSignerLocal) Pub() aqpub.Pub {
 	return ms.AQPub

pkg/media/media_signer_ext.go

···
-package media
-
-import (
-	"bytes"
-	"context"
-	"crypto"
-	"crypto/ecdsa"
-	"fmt"
-	"io"
-	"os"
-	"os/exec"
-	"time"
-
-	"github.com/decred/dcrd/dcrec/secp256k1"
-	"github.com/mr-tron/base58"
-	"go.opentelemetry.io/otel"
-	"stream.place/streamplace/pkg/atproto"
-	"stream.place/streamplace/pkg/config"
-	"stream.place/streamplace/pkg/crypto/aqpub"
-	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/model"
-	"stream.place/streamplace/pkg/spmetrics"
-)
-
-type MediaSignerExt struct {
-	cli             *config.CLI
-	signer          crypto.Signer
-	pub             aqpub.Pub
-	certPath        string
-	streamer        string
-	keyBs           []byte
-	taURL           string
-	did             string
-	manifestBuilder *ManifestBuilder
-}
-
-func MakeMediaSignerExt(ctx context.Context, cli *config.CLI, streamer string, keyBs []byte, model model.Model) (MediaSigner, error) {
-	key, _ := secp256k1.PrivKeyFromBytes(keyBs)
-	if key == nil {
-		return nil, fmt.Errorf("invalid authorization key (not valid secp256k1)")
-	}
-	var signer crypto.Signer = key.ToECDSA()
-	certBs, err := prepareCert(ctx, cli, signer)
-	if err != nil {
-		return nil, err
-	}
-	// Write certificate to a temporary file
-	certFile, err := os.CreateTemp("", "cert-*.pem")
-	if err != nil {
-		return nil, fmt.Errorf("failed to create temp cert file: %w", err)
-	}
-	defer certFile.Close()
-
-	if _, err := certFile.Write(certBs); err != nil {
-		return nil, fmt.Errorf("failed to write cert to temp file: %w", err)
-	}
-
-	certPath := certFile.Name()
-	pub, err := aqpub.FromPublicKey(signer.Public().(*ecdsa.PublicKey))
-	if err != nil {
-		return nil, err
-	}
-	did, err := atproto.ParsePubKey(signer.Public().(*ecdsa.PublicKey))
-	if err != nil {
-		return nil, err
-	}
-	return &MediaSignerExt{
-		signer:          signer,
-		certPath:        certPath,
-		streamer:        streamer,
-		pub:             pub,
-		keyBs:           keyBs,
-		taURL:           cli.TAURL,
-		did:             did.DIDKey(),
-		manifestBuilder: NewManifestBuilder(model),
-	}, nil
-}
-
-func (ms *MediaSignerExt) SignMP4(ctx context.Context, input io.ReadSeeker, start int64) ([]byte, error) {
-	startTime := time.Now()
-	_, span := otel.Tracer("signer").Start(ctx, "SignMP4_Ext")
-	defer span.End()
-
-	// Build manifest with metadata from database
-	manifestBs, err := ms.manifestBuilder.BuildManifest(ctx, ms.streamer, start)
-	if err != nil {
-		log.Error(ctx, "MediaSignerExt: failed to build manifest", "error", err)
-		return nil, fmt.Errorf("failed to build manifest: %w", err)
-	}
-
-	// Get the path to the current executable
-	execPath, err := os.Executable()
-	if err != nil {
-		return nil, fmt.Errorf("failed to get executable path: %w", err)
-	}
-
-	enc := base58.Encode(ms.keyBs)
-
-	// Prepare command with manifest JSON
-	cmd := exec.Command(execPath, "sign",
-		"--key", enc,
-		"--cert", ms.certPath,
-		"--ta-url", ms.taURL,
-		"--streamer", ms.streamer,
-		"--start-time", fmt.Sprintf("%d", start),
-		"--manifest", string(manifestBs))
-
-	// overwrite so that our subprocesses don't do their own leak checking
-	cmd.Env = append(os.Environ(), "LD_PRELOAD=")
-
-	// Set up pipes for stdin and stdout
-	stdin, err := cmd.StdinPipe()
-	if err != nil {
-		return nil, fmt.Errorf("failed to create stdin pipe: %w", err)
-	}
-
-	stdout := &bytes.Buffer{}
-	cmd.Stdout = stdout
-	stderr := &bytes.Buffer{}
-	cmd.Stderr = stderr
-
-	// Start the command
-	if err := cmd.Start(); err != nil {
-		return nil, fmt.Errorf("failed to start command: %w", err)
-	}
-
-	// Copy input to stdin
-	_, err = io.Copy(stdin, input)
-	if err != nil {
-		return nil, fmt.Errorf("failed to write to stdin: %w stderr=%s", err, stderr.String())
-	}
-	stdin.Close()
-
-	// Wait for the command to complete
-	if err := cmd.Wait(); err != nil {
-		log.Error(ctx, "MediaSignerExt: subprocess failed", "error", err, "stderr", stderr.String())
-		return nil, fmt.Errorf("command failed: %w, stderr: %s", err, stderr.String())
-	}
-
-	spmetrics.SigningDuration.WithLabelValues(ms.streamer).Observe(float64(time.Since(startTime).Milliseconds()))
-	return stdout.Bytes(), nil
-}
-
-func (ms *MediaSignerExt) Pub() aqpub.Pub {
-	return ms.pub
-}
-
-func (ms *MediaSignerExt) Streamer() string {
-	return ms.streamer
-}
-
-func (ms *MediaSignerExt) DID() string {
-	return ms.did
-}

pkg/media/segmenter.go

···
 	"bytes"
 	"context"
 	"fmt"
+	"os"
+	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/go-gst/go-gst/gst"
 	"github.com/go-gst/go-gst/gst/app"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
-	"stream.place/streamplace/pkg/globalerror"
+	"golang.org/x/sync/errgroup"
 	"stream.place/streamplace/pkg/log"
 )
 
 // element that takes the input stream, muxes to mp4, and signs the result
-func (mm *MediaManager) SegmentAndSignElem(ctx context.Context, ms MediaSigner) (*gst.Element, error) {
+func SegmentElem(ctx context.Context, cb func(ctx context.Context, buf []byte, now int64) error) (*gst.Element, error) {
 	// elem, err := gst.NewElement("splitmuxsink name=splitter async-finalize=true sink-factory=appsink muxer-factory=matroskamux max-size-bytes=1")
 	elem, err := gst.NewElementWithProperties("splitmuxsink", map[string]any{
 		"name":           "signer",
···
 		}
 	}()
 
+	// we didn't need faststart but i'm leaving this commented here in case
+	// you want to change any other muxer properties in the future
+
+	_, err = elem.Connect("muxer-added", func(split, muxEle *gst.Element) {
+		err := muxEle.SetProperty("presentation-time", false)
+		if err != nil {
+			panic("error setting interleave-bytes to 4000: " + err.Error())
+		}
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect muxer-added handler: %w", err)
+	}
+
+	// channel to make sure data is emitted in order
+	var ch chan struct{}
+
 	_, err = elem.Connect("sink-added", func(split, sinkEle *gst.Element) {
+		previousSegCh := ch
+		mySegCh := make(chan struct{}, 1)
+		ch = mySegCh
 		buf := &bytes.Buffer{}
+		err := sinkEle.SetProperty("sync", false)
+		if err != nil {
+			panic("error setting sync to false: " + err.Error())
+		}
 		appsink := app.SinkFromElement(sinkEle)
 		if appsink == nil {
 			panic("appsink should not be nil")
···
 		appsink.SetCallbacks(&app.SinkCallbacks{
 			NewSampleFunc: WriterNewSample(ctx, buf),
 			EOSFunc: func(sink *app.Sink) {
-				ctx, span := otel.Tracer("signer").Start(ctx, "SegmentAndSignElem", trace.WithAttributes(
-					attribute.String("streamer", ms.Streamer()),
-				))
-				defer span.End()
+				// ctx, span := otel.Tracer("signer").Start(ctx, "SegmentAndSignElem", trace.WithAttributes(
+				// 	attribute.String("streamer", ms.Streamer()),
+				// ))
+				// defer span.End()
+				now := time.Now().UnixMilli()
 				resetTimer <- struct{}{}
-				now := time.Now().UnixMilli()
 				bs := buf.Bytes()
-				if mm.cli.SmearAudio {
-					smearedBuf := &bytes.Buffer{}
-					err := SmearAudioTimestamps(ctx, bytes.NewReader(buf.Bytes()), smearedBuf)
-					if err != nil {
-						log.Error(ctx, "error smearing audio timestamps", "error", err)
-						return
-					}
-					bs = smearedBuf.Bytes()
+
+				if previousSegCh != nil {
+					<-previousSegCh
 				}
-				bs, err := ms.SignMP4(ctx, bytes.NewReader(bs), now)
+				err := cb(ctx, bs, now)
 				if err != nil {
 					log.Error(ctx, "error signing segment", "error", err)
 					return
 				}
+				close(mySegCh)
 
-				err = mm.ValidateMP4(ctx, bytes.NewReader(bs), true)
-				if err != nil {
-					log.Error(ctx, "error validating segment", "error", err)
-					globalerror.GlobalError(err)
-					// We don't want to stop the pipeline here because we want to keep the stream
-					// alive. Lots of weird invalid data coming in from WebRTC connections on
-					// phones. Better we drop one weird segment than force the stream to restart
-					return
-				}
 			},
 		})
 	})
···
 
 	return elem, nil
 }
+
+func (mm *MediaManager) SegmentAndSignElem(ctx context.Context, ms MediaSigner) (*gst.Element, error) {
+	return SegmentElem(ctx, func(ctx context.Context, bs []byte, now int64) error {
+		if mm.cli.SmearAudio {
+			smearedBuf := &bytes.Buffer{}
+			err := SmearAudioTimestamps(ctx, bytes.NewReader(bs), smearedBuf)
+			if err != nil {
+				return fmt.Errorf("error smearing audio timestamps: %w", err)
+			}
+			bs = smearedBuf.Bytes()
+		}
+		signedBs, err := ms.SignMP4(ctx, bytes.NewReader(bs), now)
+		if err != nil {
+			return err
+		}
+		return mm.ValidateMP4(ctx, bytes.NewReader(signedBs), true)
+	})
+}
+
+func SegmentFile(ctx context.Context, input string, outDir string) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	g, ctx := errgroup.WithContext(ctx)
+	pipelineSlice := []string{
+		"filesrc name=filesrc ! qtdemux name=demux",
+		"demux. ! queue ! h264parse ! rtph264pay ! rtph264depay ! h264parse name=videoparse",
+		"demux. ! queue ! opusparse name=audioparse",
+	}
+	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))
+	if err != nil {
+		return fmt.Errorf("error creating MKVIngest pipeline: %w", err)
+	}
+
+	srcele, err := pipeline.GetElementByName("filesrc")
+	if err != nil {
+		return err
+	}
+	if err := srcele.Set("location", input); err != nil {
+		return err
+	}
+
+	videoParseEle, err := pipeline.GetElementByName("videoparse")
+	if err != nil {
+		return err
+	}
+
+	var segCount atomic.Int64
+
+	segmenter, err := SegmentElem(ctx, func(ctx context.Context, buf []byte, now int64) error {
+		seg := segCount.Load()
+		segCount.Add(1)
+		g.Go(func() error {
+			fpath := fmt.Sprintf("%s/%06d.mp4", outDir, seg)
+			log.Log(ctx, "writing segment", "path", fpath)
+			fd, err := os.Create(fpath)
+			if err != nil {
+				return err
+			}
+			defer fd.Close()
+			_, err = fd.Write(buf)
+			return err
+		})
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	err = pipeline.Add(segmenter)
+	if err != nil {
+		return err
+	}
+	err = videoParseEle.Link(segmenter)
+	if err != nil {
+		return err
+	}
+	audioparse, err := pipeline.GetElementByName("audioparse")
+	if err != nil {
+		return err
+	}
+	err = audioparse.Link(segmenter)
+	if err != nil {
+		return err
+	}
+
+	busErr := make(chan error)
+	go func() {
+		err := HandleBusMessages(ctx, pipeline)
+		cancel()
+		busErr <- err
+	}()
+
+	err = pipeline.SetState(gst.StatePlaying)
+	if err != nil {
+		return err
+	}
+
+	defer func() {
+		err := pipeline.SetState(gst.StateNull)
+		if err != nil {
+			log.Error(ctx, "error setting pipeline to null state", "error", err)
+		}
+	}()
+
+	<-busErr
+	err = g.Wait()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/media/validate.go

···
 	"strings"
 	"time"
 
+	"github.com/bluesky-social/indigo/atproto/crypto"
 	"go.opentelemetry.io/otel"
 	"stream.place/streamplace/pkg/aqtime"
 	c2patypes "stream.place/streamplace/pkg/c2patypes"
···
 	if err != nil {
 		return err
 	}
-	var maniCert ManifestAndCert
-	maniStr, err := iroh_streamplace.GetManifestAndCert(buf)
+
+	valid, err := ValidateMP4Media(ctx, buf)
 	if err != nil {
 		return err
 	}
-	err = json.Unmarshal([]byte(maniStr), &maniCert)
-	if err != nil {
-		return err
-	}
-	label := maniCert.Manifest.Label
+	meta := valid.Meta
+	pub := valid.Pub
+	mediaData := valid.MediaData
+	manifest := valid.Manifest
+
+	label := manifest.Label
 	if label != nil && mm.model != nil {
 		oldSeg, err := mm.model.GetSegment(*label)
 		if err != nil {
···
 			return nil
 		}
 	}
-	pub, err := signers.ParseES256KCert([]byte(maniCert.Cert))
-	if err != nil {
-		return err
-	}
-	meta, err := ParseSegmentAssertions(ctx, &maniCert.Manifest)
-	if err != nil {
-		return err
-	}
+
 	if meta.MetadataConfiguration != nil {
 		if meta.MetadataConfiguration.DistributionPolicy != nil {
 			allowedBroadcasters := meta.MetadataConfiguration.DistributionPolicy.AllowedBroadcasters
···
 			}
 		}
 	}
-	mediaData, err := ParseSegmentMediaData(ctx, buf)
-	if err != nil {
-		return err
-	}
-	// special case for test signers that are only signed with a key
+
 	var repoDID string
 	var signingKeyDID string
+	// special case for test signers that are only signed with a key
 	if strings.HasPrefix(meta.Creator, constants.DID_KEY_PREFIX) {
 		signingKeyDID = meta.Creator
 		repoDID = meta.Creator
···
 		deleteAfter = meta.DistributionPolicy.ExpiresAt
 	}
 	seg := &model.Segment{
-		ID:                 *maniCert.Manifest.Label,
+		ID:                 *label,
 		SigningKeyDID:      signingKeyDID,
 		RepoDID:            repoDID,
 		StartTime:          meta.StartTime.Time(),
···
 			case <-ctx.Done():
 				return
 			case <-time.After(1 * time.Minute):
-				log.Warn(ctx, "failed to send segment to channel, timing out", "streamer", repoDID, "signingKey", signingKeyDID, "segmentID", *maniCert.Manifest.Label)
+				log.Warn(ctx, "failed to send segment to channel, timing out", "streamer", repoDID, "signingKey", signingKeyDID, "segmentID", *label)
 				return
 			}
 		}()
 	}
 	aqt := aqtime.FromTime(meta.StartTime.Time())
-	log.Log(ctx, "successfully ingested segment", "user", repoDID, "signingKey", signingKeyDID, "timestamp", aqt.FileSafeString(), "segmentID", *maniCert.Manifest.Label)
+	log.Log(ctx, "successfully ingested segment", "user", repoDID, "signingKey", signingKeyDID, "timestamp", aqt.FileSafeString(), "segmentID", *label)
 	return nil
 }
 
···
 	}
 	return false
 }
+
+type ValidationResult struct {
+	Pub       *crypto.PublicKeyK256
+	Meta      *SegmentMetadata
+	MediaData *model.SegmentMediaData
+	Manifest  *c2patypes.Manifest
+	Cert      string
+}
+
+// validate a signed mp4 file unto itself, ignoring whether this user is allowed and whatnot
+func ValidateMP4Media(ctx context.Context, buf []byte) (*ValidationResult, error) {
+	var maniCert ManifestAndCert
+	maniStr, err := iroh_streamplace.GetManifestAndCert(buf)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal([]byte(maniStr), &maniCert)
+	if err != nil {
+		return nil, err
+	}
+	pub, err := signers.ParseES256KCert([]byte(maniCert.Cert))
+	if err != nil {
+		return nil, err
+	}
+	meta, err := ParseSegmentAssertions(ctx, &maniCert.Manifest)
+	if err != nil {
+		return nil, err
+	}
+	mediaData, err := ParseSegmentMediaData(ctx, buf)
+	if err != nil {
+		return nil, err
+	}
+	return &ValidationResult{
+		Pub:       pub,
+		Meta:      meta,
+		MediaData: mediaData,
+		Manifest:  &maniCert.Manifest,
+		Cert:      maniCert.Cert,
+	}, nil
+}

rust/iroh-streamplace/Cargo.toml

···
 postcard = { version = "1.1.3", features = ["use-std"] }
 irpc = "0.9.0"
 irpc-iroh = "0.9.0"
-c2pa = { git = "https://github.com/hyphacoop/c2pa-rs.git", rev = "1b84d40219b27340a30fc309250e774e8a7b7761", features = [
+c2pa = { git = "https://github.com/streamplace/c2pa-rs.git", rev = "544825abfaf9e588813e18dba70c8d5afd039d46", features = [
   "openssl",
   "file_io",
 ] }

rust/iroh-streamplace/src/c2pa.rs

···
 use std::{io::Cursor, sync::Arc};
 
-use c2pa::{Builder, CallbackSigner, Reader, settings::Settings};
+use c2pa::Builder;
+use c2pa::CallbackSigner;
+use c2pa::Reader;
+use c2pa::settings::Settings;
+
+use c2pa::Ingredient;
+use c2pa::jumbf_io;
+use c2pa::status_tracker::StatusTracker;
+use c2pa::store::Store;
+
 use serde_json;
 
 #[derive(Debug, thiserror::Error, uniffi::Error)]
···
 pub fn get_manifest_and_cert(data: Vec<u8>) -> Result<String, SPError> {
     let reader = Reader::from_stream("video/mp4", Cursor::new(data))
         .map_err(|e| SPError::C2paError(e.to_string()))?;
+
     if let Some(manifest) = reader.active_manifest() {
         let cert_chain = if let Some(si) = manifest.signature_info() {
             si.cert_chain()
···
     Settings::from_toml(TOML_SETTINGS).map_err(|e| SPError::C2paError(e.to_string()))?;
     let callback_signer = CallbackSigner::new(
         move |_context: *const (), data: &[u8]| {
+            let signature = gosigner
+                .sign(data.to_vec())
+                .map_err(|e| c2pa::Error::BadParam(e.to_string()))?;
+            Ok(signature)
+        },
+        c2pa::SigningAlg::Es256K,
+        certs,
+    );
+    let mut builder =
+        Builder::from_json(&manifest).map_err(|e| SPError::C2paError(e.to_string()))?;
+    let mut output = Vec::new();
+    let mut input_cursor = Cursor::new(data);
+    let mut output_cursor = Cursor::new(&mut output);
+    builder
+        .sign(
+            &callback_signer,
+            "video/mp4",
+            &mut input_cursor,
+            &mut output_cursor,
+        )
+        .map_err(|e| SPError::C2paError(e.to_string()))?;
+    Ok(output)
+}
+
+#[uniffi::export]
+pub fn get_manifests(data: Vec<u8>) -> Result<String, SPError> {
+    let store = Reader::from_stream("video/mp4", Cursor::new(data))
+        .map_err(|e| SPError::C2paError(e.to_string()))?;
+    let result = serde_json::json!({
+        "manifests": store.manifests()
+    });
+    Ok(result.to_string())
+}
+
+#[uniffi::export]
+pub fn resign(
+    unsigned_seg_label: String,
+    unsigned_seg_data: Vec<u8>,
+    signed_concat_data: Vec<u8>,
+    certs: Vec<u8>,
+    gosigner: Arc<dyn GoSigner>,
+) -> Result<Vec<u8>, SPError> {
+    let callback_signer = CallbackSigner::new(
+        move |_context: *const (), data: &[u8]| {
+            gosigner
+                .sign(data.to_vec())
+                .map_err(|e| c2pa::Error::BadParam(e.to_string()))
+        },
+        c2pa::SigningAlg::Es256K,
+        certs,
+    );
+
+    let mut validation_log = StatusTracker::default();
+
+    let combined_store = Store::from_stream(
+        "video/mp4",
+        Cursor::new(signed_concat_data),
+        true,
+        &mut validation_log,
+    )
+    .map_err(|e| SPError::C2paError(format!("from_stream failed: {}", e)))?;
+
+    let seg_claim = combined_store
+        .get_claim(unsigned_seg_label.as_str())
+        .ok_or(SPError::C2paError(format!(
+            "Segment claim not found: {}",
+            unsigned_seg_label
+        )))?;
+
+    let seg_claim_clone = seg_claim.clone();
+
+    let mut seg_store = Store::new();
+    let _provenance = seg_store
+        .commit_claim(seg_claim_clone)
+        .map_err(|e| SPError::C2paError(format!("commit_claim failed: {}", e)))?;
+
+    let mut output = Vec::new();
+    let mut output_cursor = Cursor::new(&mut output);
+    let mut input_cursor = Cursor::new(unsigned_seg_data);
+
+    let jumbf_bytes = seg_store
+        .to_jumbf(&callback_signer)
+        .map_err(|e| SPError::C2paError(format!("to_jumbf failed: {}", e)))?;
+
+    jumbf_io::save_jumbf_to_stream(
+        "video/mp4",
+        &mut input_cursor,
+        &mut output_cursor,
+        &jumbf_bytes,
+    )
+    .map_err(|e| SPError::C2paError(format!("save_jumbf_to_stream failed: {}", e)))?;
+
+    // seg_store
+    //     .save_to_stream(
+    //         "video/mp4",
+    //         &mut input_cursor,
+    //         &mut output_cursor,
+    //         &callback_signer,
+    //     )
+    //     .map_err(|e| SPError::C2paError(format!("save_to_stream failed: {}", e)))?;
+    Ok(output)
+}
+
+#[uniffi::export]
+pub fn sign_with_ingredients(
+    manifest: String,
+    data: Vec<u8>,
+    certs: Vec<u8>,
+    ingredients: Vec<Vec<u8>>,
+    gosigner: Arc<dyn GoSigner>,
+) -> Result<Vec<u8>, SPError> {
+    Settings::from_toml(TOML_SETTINGS).map_err(|e| SPError::C2paError(e.to_string()))?;
+    let callback_signer = CallbackSigner::new(
+        move |_context: *const (), data: &[u8]| {
             gosigner
                 .sign(data.to_vec())
                 .map_err(|e| c2pa::Error::BadParam(e.to_string()))
···
     );
     let mut builder =
         Builder::from_json(&manifest).map_err(|e| SPError::C2paError(e.to_string()))?;
+    for ingredient in ingredients {
+        let mut cursor = Cursor::new(ingredient);
+        let ingredient = Ingredient::from_stream("video/mp4", &mut cursor)
+            .map_err(|e| SPError::C2paError(e.to_string()))?;
+        builder.add_ingredient(ingredient);
+    }
     let mut output = Vec::new();
     let mut input_cursor = Cursor::new(data);
     let mut output_cursor = Cursor::new(&mut output);

subprojects/gstreamer-full.wrap

···
 [wrap-git]
 url = https://gitlab.freedesktop.org/iameli/gstreamer.git
-revision = 5c372f9595b7024d1074e45ee50b68b7a9a3f5e3
+revision = 587f2f9d8678df804515dac40aaac0a732c32c00
 depth = 1

test/remote/remote-fixtures.go

···
 package remote
 
 import (
+	"archive/tar"
+	"compress/gzip"
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
···
 
 	return finalPath
 }
+
+// takes a tarball, returns a directory with the contents
+func RemoteArchive(name string) string {
+	fpath := RemoteFixture(name)
+
+	// Create extracted directory adjacent to the archive file
+	dir := filepath.Dir(fpath)
+	extractedDir := filepath.Join(dir, "extracted")
+
+	if err := os.MkdirAll(extractedDir, 0755); err != nil {
+		panic(err)
+	}
+
+	// Extract the tarball contents into the directory
+	file, err := os.Open(fpath)
+	if err != nil {
+		panic(err)
+	}
+	defer file.Close()
+
+	// Create gzip reader
+	gzr, err := gzip.NewReader(file)
+	if err != nil {
+		panic(err)
+	}
+	defer gzr.Close()
+
+	tr := tar.NewReader(gzr)
+	for {
+		header, err := tr.Next()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			panic(err)
+		}
+
+		target := filepath.Join(extractedDir, header.Name)
+
+		switch header.Typeflag {
+		case tar.TypeDir:
+			if err := os.MkdirAll(target, 0755); err != nil {
+				panic(err)
+			}
+		case tar.TypeReg:
+			// Create parent directories if needed
+			if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
+				panic(err)
+			}
+
+			outFile, err := os.Create(target)
+			if err != nil {
+				panic(err)
+			}
+
+			if _, err := io.Copy(outFile, tr); err != nil {
+				outFile.Close()
+				panic(err)
+			}
+			outFile.Close()
+		}
+	}
+
+	return extractedDir
+}