From 38b91c213d29fac83d5f2fcab7fafd120f0ffb59 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 29 Oct 2025 16:06:41 -0700 Subject: [PATCH] determinism: introduce deterministic muxing --- Cargo.lock | 98 +++++++++- hack/deterministic-mux.sh | 12 ++ pkg/api/stream_key.go | 7 +- pkg/cmd/streamplace.go | 14 ++ pkg/config/config.go | 5 +- .../iroh_streamplace/iroh_streamplace.go | 112 +++++++++++ .../iroh_streamplace/iroh_streamplace.h | 33 ++++ pkg/media/clip.go | 4 +- pkg/media/deterministic_muxing_test.go | 142 ++++++++++++++ pkg/media/ingredient_test.go | 156 +++++++++++++++ pkg/media/media.go | 3 +- pkg/media/media_signer.go | 78 ++++++++ pkg/media/media_signer_ext.go | 154 --------------- pkg/media/segmenter.go | 178 +++++++++++++++--- pkg/media/validate.go | 78 +++++--- rust/iroh-streamplace/Cargo.toml | 2 +- rust/iroh-streamplace/src/c2pa.rs | 132 ++++++++++++- subprojects/gstreamer-full.wrap | 2 +- test/remote/remote-fixtures.go | 67 +++++++ 19 files changed, 1056 insertions(+), 221 deletions(-) create mode 100755 hack/deterministic-mux.sh create mode 100644 pkg/media/deterministic_muxing_test.go create mode 100644 pkg/media/ingredient_test.go delete mode 100644 pkg/media/media_signer_ext.go diff --git a/Cargo.lock b/Cargo.lock index a08938880..1c25b7569 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -614,6 +614,100 @@ dependencies = [ "zip", ] +[[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" @@ -1429,7 +1523,7 @@ name = "export_c2pa_schema" 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", @@ -2641,7 +2735,7 @@ dependencies = [ "anyhow", "async-trait", "bytes", - "c2pa", + "c2pa 0.58.0 (git+https://github.com/streamplace/c2pa-rs.git?rev=544825abfaf9e588813e18dba70c8d5afd039d46)", "hex", "iroh", "iroh-base", diff --git a/hack/deterministic-mux.sh b/hack/deterministic-mux.sh new file mode 100755 index 000000000..4ea795010 --- /dev/null +++ b/hack/deterministic-mux.sh @@ -0,0 +1,12 @@ +#!/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" diff --git a/pkg/api/stream_key.go b/pkg/api/stream_key.go index 747c53a15..455a29af1 100644 --- a/pkg/api/stream_key.go +++ b/pkg/api/stream_key.go @@ -82,12 +82,7 @@ func (a *StreamplaceAPI) MakeMediaSigner(ctx context.Context, keyStr string) (me 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) } diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index a6599d2a0..adb0c98f8 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -141,6 +141,20 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { 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 { diff --git a/pkg/config/config.go b/pkg/config/config.go index 4b1314ce2..4f13b10a0 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -195,14 +195,12 @@ func (cli *CLI) NewFlagSet(name string) *flag.FlagSet { 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") @@ -230,6 +228,9 @@ func (cli *CLI) NewFlagSet(name string) *flag.FlagSet { 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) lpFlags.VisitAll(func(f *flag.Flag) { diff --git a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go index 07023f280..b7ed47379 100644 --- a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go +++ b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go @@ -361,6 +361,15 @@ func uniffiCheckChecksums() { panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_get_manifest_and_cert: UniFFI API checksum mismatch") } } + { + 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() @@ -388,6 +397,15 @@ func uniffiCheckChecksums() { panic("iroh_streamplace: uniffi_iroh_streamplace_checksum_func_node_id_from_ticket: UniFFI API checksum mismatch") } } + { + 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() @@ -397,6 +415,15 @@ func uniffiCheckChecksums() { 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") + } + } { checksum := rustCall(func(_uniffiStatus *C.RustCallStatus) C.uint16_t { return C.uniffi_iroh_streamplace_checksum_func_subscribe_item_debug() @@ -4540,6 +4567,49 @@ func (FfiDestroyerSequenceString) Destroy(sequence []string) { } } +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{} @@ -4703,6 +4773,20 @@ func GetManifestAndCert(data []byte) (string, error) { } } +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() { @@ -4737,6 +4821,20 @@ func NodeIdFromTicket(ticketStr string) (*PublicKey, error) { } } +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{ @@ -4751,6 +4849,20 @@ func Sign(manifest string, data []byte, certs []byte, gosigner GoSigner) ([]byte } } +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 { + var _uniffiDefaultValue []byte + return _uniffiDefaultValue, _uniffiErr + } else { + return FfiConverterBytesINSTANCE.Lift(_uniffiRV), nil + } +} + func SubscribeItemDebug(item SubscribeItem) string { return FfiConverterStringINSTANCE.Lift(rustCall(func(_uniffiStatus *C.RustCallStatus) RustBufferI { return GoRustBuffer{ diff --git a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h index 625810b5d..995bfb6bf 100644 --- a/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h +++ b/pkg/iroh/generated/iroh_streamplace/iroh_streamplace.h @@ -738,6 +738,11 @@ uint64_t uniffi_iroh_streamplace_fn_method_writescope_put(void* ptr, RustBuffer 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 @@ -754,11 +759,21 @@ void uniffi_iroh_streamplace_fn_func_init_logging_with_level(RustBuffer level, R 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 #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_FN_FUNC_SUBSCRIBE_ITEM_DEBUG RustBuffer uniffi_iroh_streamplace_fn_func_subscribe_item_debug(RustBuffer item, RustCallStatus *out_status @@ -1048,6 +1063,12 @@ void ffi_iroh_streamplace_rust_future_complete_void(uint64_t handle, RustCallSta #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_GET_MANIFEST_AND_CERT uint16_t uniffi_iroh_streamplace_checksum_func_get_manifest_and_cert(void +); +#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 @@ -1066,12 +1087,24 @@ uint16_t uniffi_iroh_streamplace_checksum_func_init_logging_with_level(void #define UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_NODE_ID_FROM_TICKET uint16_t uniffi_iroh_streamplace_checksum_func_node_id_from_ticket(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 #ifndef UNIFFI_FFIDEF_UNIFFI_IROH_STREAMPLACE_CHECKSUM_FUNC_SUBSCRIBE_ITEM_DEBUG diff --git a/pkg/media/clip.go b/pkg/media/clip.go index 4987148f2..86b91ea5e 100644 --- a/pkg/media/clip.go +++ b/pkg/media/clip.go @@ -43,8 +43,8 @@ func Clip(ctx context.Context, sources []string, w io.Writer) error { 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", } diff --git a/pkg/media/deterministic_muxing_test.go b/pkg/media/deterministic_muxing_test.go new file mode 100644 index 000000000..328d5ec33 --- /dev/null +++ b/pkg/media/deterministic_muxing_test.go @@ -0,0 +1,142 @@ +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) +} diff --git a/pkg/media/ingredient_test.go b/pkg/media/ingredient_test.go new file mode 100644 index 000000000..75c05a14f --- /dev/null +++ b/pkg/media/ingredient_test.go @@ -0,0 +1,156 @@ +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 +} diff --git a/pkg/media/media.go b/pkg/media/media.go index da0dabe45..e63ce2fbd 100644 --- a/pkg/media/media.go +++ b/pkg/media/media.go @@ -197,6 +197,7 @@ type SegmentMetadata struct { 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) { @@ -210,7 +211,7 @@ func ParseSegmentAssertions(ctx context.Context, mani *c2patypes.Manifest) (*Seg } } if ass == nil { - return nil, fmt.Errorf("couldn't find %s assertions", StreamplaceMetadata) + return nil, ErrMissingMetadata } proc := ld.NewJsonLdProcessor() options := ld.NewJsonLdOptions("") diff --git a/pkg/media/media_signer.go b/pkg/media/media_signer.go index 9ac449c31..24dedd96f 100644 --- a/pkg/media/media_signer.go +++ b/pkg/media/media_signer.go @@ -29,8 +29,11 @@ type MediaSigner interface { 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 Signer crypto.Signer @@ -40,6 +43,7 @@ type MediaSignerLocal struct { 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) { @@ -171,6 +175,80 @@ func (ms *MediaSignerLocal) SignMP4(ctx context.Context, input io.ReadSeeker, st 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 } diff --git a/pkg/media/media_signer_ext.go b/pkg/media/media_signer_ext.go deleted file mode 100644 index 15736fcfa..000000000 --- a/pkg/media/media_signer_ext.go +++ /dev/null @@ -1,154 +0,0 @@ -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 -} diff --git a/pkg/media/segmenter.go b/pkg/media/segmenter.go index 09d12af53..1cc16ba40 100644 --- a/pkg/media/segmenter.go +++ b/pkg/media/segmenter.go @@ -4,19 +4,19 @@ import ( "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", @@ -55,8 +55,31 @@ func (mm *MediaManager) SegmentAndSignElem(ctx context.Context, ms MediaSigner) } }() + // 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") @@ -65,37 +88,24 @@ func (mm *MediaManager) SegmentAndSignElem(ctx context.Context, ms MediaSigner) 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() - resetTimer <- struct{}{} + // ctx, span := otel.Tracer("signer").Start(ctx, "SegmentAndSignElem", trace.WithAttributes( + // attribute.String("streamer", ms.Streamer()), + // )) + // defer span.End() now := time.Now().UnixMilli() + resetTimer <- struct{}{} 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 - } }, }) }) @@ -105,3 +115,115 @@ func (mm *MediaManager) SegmentAndSignElem(ctx context.Context, ms MediaSigner) 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 +} diff --git a/pkg/media/validate.go b/pkg/media/validate.go index 4b8b849e7..14308318e 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/bluesky-social/indigo/atproto/crypto" "go.opentelemetry.io/otel" "stream.place/streamplace/pkg/aqtime" c2patypes "stream.place/streamplace/pkg/c2patypes" @@ -32,16 +33,17 @@ func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local if err != nil { return err } - var maniCert ManifestAndCert - maniStr, err := iroh_streamplace.GetManifestAndCert(buf) - if err != nil { - return err - } - err = json.Unmarshal([]byte(maniStr), &maniCert) + + valid, err := ValidateMP4Media(ctx, buf) 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 { @@ -52,14 +54,7 @@ func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local 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 @@ -70,13 +65,10 @@ func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local } } } - 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 @@ -123,7 +115,7 @@ func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local deleteAfter = meta.DistributionPolicy.ExpiresAt } seg := &model.Segment{ - ID: *maniCert.Manifest.Label, + ID: *label, SigningKeyDID: signingKeyDID, RepoDID: repoDID, StartTime: meta.StartTime.Time(), @@ -150,13 +142,13 @@ func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local 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 } @@ -205,3 +197,43 @@ func (mm *MediaManager) isWarningBlocked(warning string) bool { } 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 +} diff --git a/rust/iroh-streamplace/Cargo.toml b/rust/iroh-streamplace/Cargo.toml index 5976b2caf..ae375a0ea 100644 --- a/rust/iroh-streamplace/Cargo.toml +++ b/rust/iroh-streamplace/Cargo.toml @@ -25,7 +25,7 @@ serde = { version = "1.0.219", features = ["derive"] } 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", ] } diff --git a/rust/iroh-streamplace/src/c2pa.rs b/rust/iroh-streamplace/src/c2pa.rs index 289d267bc..04932c0a7 100644 --- a/rust/iroh-streamplace/src/c2pa.rs +++ b/rust/iroh-streamplace/src/c2pa.rs @@ -1,6 +1,15 @@ 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)] @@ -16,6 +25,7 @@ pub enum SPError { pub fn get_manifest_and_cert(data: Vec) -> Result { 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() @@ -106,6 +116,120 @@ pub fn sign( data: Vec, certs: Vec, gosigner: Arc, +) -> Result, SPError> { + 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) -> Result { + 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, + signed_concat_data: Vec, + certs: Vec, + gosigner: Arc, +) -> Result, 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, + certs: Vec, + ingredients: Vec>, + gosigner: Arc, ) -> Result, SPError> { Settings::from_toml(TOML_SETTINGS).map_err(|e| SPError::C2paError(e.to_string()))?; let callback_signer = CallbackSigner::new( @@ -119,6 +243,12 @@ pub fn sign( ); 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); diff --git a/subprojects/gstreamer-full.wrap b/subprojects/gstreamer-full.wrap index 962be5609..99b007593 100644 --- a/subprojects/gstreamer-full.wrap +++ b/subprojects/gstreamer-full.wrap @@ -1,4 +1,4 @@ [wrap-git] url = https://gitlab.freedesktop.org/iameli/gstreamer.git -revision = 5c372f9595b7024d1074e45ee50b68b7a9a3f5e3 +revision = 587f2f9d8678df804515dac40aaac0a732c32c00 depth = 1 diff --git a/test/remote/remote-fixtures.go b/test/remote/remote-fixtures.go index f7a8e1390..f1218a3a9 100644 --- a/test/remote/remote-fixtures.go +++ b/test/remote/remote-fixtures.go @@ -1,6 +1,8 @@ package remote import ( + "archive/tar" + "compress/gzip" "crypto/sha256" "encoding/hex" "fmt" @@ -80,3 +82,68 @@ func RemoteFixture(name string) string { 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 +} -- knotmirror 0.0.1