diff --git a/Cargo.lock b/Cargo.lock index 0dde7e08..cde2d8d4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -85,56 +85,6 @@ dependencies = [ "libc", ] -[[package]] -name = "anstream" -version = "0.6.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43d5b281e737544384e969a5ccad3f1cdd24b48086a0fc1b2a5262a26b8f4f4a" -dependencies = [ - "anstyle", - "anstyle-parse", - "anstyle-query", - "anstyle-wincon", - "colorchoice", - "is_terminal_polyfill", - "utf8parse", -] - -[[package]] -name = "anstyle" -version = "1.0.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" - -[[package]] -name = "anstyle-parse" -version = "0.2.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e7644824f0aa2c7b9384579234ef10eb7efb6a0deb83f9630a49594dd9c15c2" -dependencies = [ - "utf8parse", -] - -[[package]] -name = "anstyle-query" -version = "1.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" -dependencies = [ - "windows-sys 0.61.1", -] - -[[package]] -name = "anstyle-wincon" -version = "3.0.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" -dependencies = [ - "anstyle", - "once_cell_polyfill", - "windows-sys 0.61.1", -] - [[package]] name = "anyhow" version = "1.0.99" @@ -575,104 +525,6 @@ dependencies = [ "serde", ] -[[package]] -name = "c2pa" -version = "0.58.0" -source = "git+https://github.com/streamplace/c2pa-rs.git?rev=81d737bd#81d737bd117743d8146cd8317604162af4df269f" -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", - "ecdsa", - "ed25519-dalek 2.2.0", - "env_logger", - "extfmt", - "getrandom 0.2.16", - "hex", - "hex-literal", - "http", - "id3", - "img-parts", - "iref", - "jfifdump", - "js-sys", - "k256", - "lazy_static", - "log", - "memchr", - "mp4", - "nom", - "non-empty-string", - "nonempty-collections", - "num-bigint-dig", - "p256", - "p384", - "p521", - "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 = "c2pa" version = "0.58.0" @@ -804,12 +656,6 @@ dependencies = [ "thiserror 1.0.69", ] -[[package]] -name = "cbor4ii" -version = "1.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "faed1a83001dc2c9201451030cc317e35bef36c84d3781d7c5bb9f343c397da8" - [[package]] name = "cc" version = "1.2.32" @@ -905,46 +751,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "clap" -version = "4.5.60" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2797f34da339ce31042b27d23607e051786132987f595b02ba4f6a6dffb7030a" -dependencies = [ - "clap_builder", - "clap_derive", -] - -[[package]] -name = "clap_builder" -version = "4.5.60" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24a241312cea5059b13574bb9b3861cabf758b879c15190b37b6d6fd63ab6876" -dependencies = [ - "anstream", - "anstyle", - "clap_lex", - "strsim", -] - -[[package]] -name = "clap_derive" -version = "4.5.55" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a92793da1a46a5f2a02a6f4c46c6496b28c43638adea8306fcb0caa1634f24e5" -dependencies = [ - "heck", - "proc-macro2", - "quote", - "syn 2.0.105", -] - -[[package]] -name = "clap_lex" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" - [[package]] name = "cobs" version = "0.3.0" @@ -965,12 +771,6 @@ dependencies = [ "termcolor", ] -[[package]] -name = "colorchoice" -version = "1.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" - [[package]] name = "config" version = "0.14.1" @@ -1312,49 +1112,12 @@ dependencies = [ "syn 2.0.105", ] -[[package]] -name = "dasl" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b59666035a4386b0fd272bd78da4cbc3ccb558941e97579ab00f0eb4639f2a49" -dependencies = [ - "blake3", - "cbor4ii", - "data-encoding", - "data-encoding-macro", - "scopeguard", - "serde", - "serde_bytes", - "sha2 0.10.9", - "thiserror 2.0.16", -] - [[package]] name = "data-encoding" version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476" -[[package]] -name = "data-encoding-macro" -version = "0.1.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "47ce6c96ea0102f01122a185683611bd5ac8d99e62bc59dd12e6bda344ee673d" -dependencies = [ - "data-encoding", - "data-encoding-macro-internal", -] - -[[package]] -name = "data-encoding-macro-internal" -version = "0.1.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8d162beedaa69905488a8da94f5ac3edb4dd4788b732fadb7bd120b2625c1976" -dependencies = [ - "data-encoding", - "syn 2.0.105", -] - [[package]] name = "delegate" version = "0.8.0" @@ -1373,8 +1136,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid 0.9.6", - "der_derive 0.7.3", - "flagset", "pem-rfc7468 0.7.0", "zeroize", ] @@ -1386,7 +1147,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e9d8dd2f26c86b27a2a8ea2767ec7f9df7a89516e4794e54ac01ee618dda3aa4" dependencies = [ "const-oid 0.10.1", - "der_derive 0.8.0-rc.6", + "der_derive", "pem-rfc7468 1.0.0-rc.3", "zeroize", ] @@ -1405,17 +1166,6 @@ dependencies = [ "rusticata-macros", ] -[[package]] -name = "der_derive" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.105", -] - [[package]] name = "der_derive" version = "0.8.0-rc.6" @@ -1737,7 +1487,7 @@ name = "export_c2pa_schema" version = "0.0.0" dependencies = [ "anyhow", - "c2pa 0.58.0 (git+https://github.com/streamplace/c2pa-rs.git?rev=f23cfd398aac79f06cc7114baa633096189d5a37)", + "c2pa", "schemars 0.8.22", "serde", "serde_json", @@ -1790,12 +1540,6 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" -[[package]] -name = "flagset" -version = "0.4.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" - [[package]] name = "flate2" version = "1.1.2" @@ -2972,7 +2716,7 @@ dependencies = [ "async-trait", "base64 0.22.1", "bytes", - "c2pa 0.58.0 (git+https://github.com/streamplace/c2pa-rs.git?rev=f23cfd398aac79f06cc7114baa633096189d5a37)", + "c2pa", "hex", "iroh", "iroh-base", @@ -3059,12 +2803,6 @@ dependencies = [ "windows-sys 0.59.0", ] -[[package]] -name = "is_terminal_polyfill" -version = "1.70.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" - [[package]] name = "itertools" version = "0.13.0" @@ -3307,63 +3045,6 @@ dependencies = [ "thiserror 1.0.69", ] -[[package]] -name = "mp4-atom" -version = "0.10.1" -source = "git+https://github.com/streamplace/mp4-atom.git?branch=streamplace#126c49b8e3e3f8810089ed5cb157da94ce4f04b7" -dependencies = [ - "derive_more 2.0.1", - "num", - "paste", - "thiserror 1.0.69", - "tracing", -] - -[[package]] -name = "muxl" -version = "0.1.0" -source = "git+https://github.com/streamplace/muxl.git?rev=a1cfe7a4250835a55fb440d02221ce6206d45408#a1cfe7a4250835a55fb440d02221ce6206d45408" -dependencies = [ - "blake3", - "clap", - "dasl", - "hex", - "mp4-atom", - "serde", - "serde_bytes", - "serde_json", - "tempfile", -] - -[[package]] -name = "muxl-sign" -version = "0.1.0" -source = "git+https://github.com/streamplace/muxl.git?rev=a1cfe7a4250835a55fb440d02221ce6206d45408#a1cfe7a4250835a55fb440d02221ce6206d45408" -dependencies = [ - "c2pa 0.58.0 (git+https://github.com/streamplace/c2pa-rs.git?rev=81d737bd)", - "clap", - "const-oid 0.9.6", - "dasl", - "der 0.7.10", - "k256", - "mp4-atom", - "muxl", - "serde", - "serde_bytes", - "serde_json", - "sha1", - "sha2 0.10.9", - "spki 0.7.3", - "x509-cert", -] - -[[package]] -name = "muxl-wasm" -version = "0.1.0" -dependencies = [ - "muxl-sign", -] - [[package]] name = "n0-future" version = "0.1.3" @@ -3651,20 +3332,6 @@ dependencies = [ "winapi", ] -[[package]] -name = "num" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23" -dependencies = [ - "num-bigint", - "num-complex", - "num-integer", - "num-iter", - "num-rational", - "num-traits", -] - [[package]] name = "num-bigint" version = "0.4.6" @@ -3693,15 +3360,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "num-complex" -version = "0.4.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" -dependencies = [ - "num-traits", -] - [[package]] name = "num-conv" version = "0.1.0" @@ -3800,12 +3458,6 @@ dependencies = [ "portable-atomic", ] -[[package]] -name = "once_cell_polyfill" -version = "1.70.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" - [[package]] name = "ordered-multimap" version = "0.7.3" @@ -5827,27 +5479,6 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" -[[package]] -name = "tls_codec" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0de2e01245e2bb89d6f05801c564fa27624dbd7b1846859876c7dad82e90bf6b" -dependencies = [ - "tls_codec_derive", - "zeroize", -] - -[[package]] -name = "tls_codec_derive" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d2e76690929402faae40aebdda620a2c0e25dd6d3b9afe48867dfd95991f4bd" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.105", -] - [[package]] name = "tokio" version = "1.47.1" @@ -6379,12 +6010,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" -[[package]] -name = "utf8parse" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" - [[package]] name = "uuid" version = "1.12.0" @@ -6987,20 +6612,6 @@ dependencies = [ "tap", ] -[[package]] -name = "x509-cert" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94" -dependencies = [ - "const-oid 0.9.6", - "der 0.7.10", - "sha1", - "signature 2.2.0", - "spki 0.7.3", - "tls_codec", -] - [[package]] name = "x509-certificate" version = "0.24.0" diff --git a/Cargo.toml b/Cargo.toml index 58ed954b..8c73d338 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,4 +1,4 @@ [workspace] resolver = "3" -members = ["rust/export-c2pa-schema", "rust/iroh-streamplace", "rust/muxl-wasm"] +members = ["rust/export-c2pa-schema", "rust/iroh-streamplace"] exclude = ["subprojects/c2pa_go"] diff --git a/Makefile b/Makefile index 0212bb69..e8c331f5 100644 --- a/Makefile +++ b/Makefile @@ -291,14 +291,13 @@ test-vod: .PHONY: dev-setup dev-setup: - $(MAKE) -j16 app-cached dev-setup-meson muxl-wasm + $(MAKE) -j16 app-cached dev-setup-meson .PHONY: dev dev: app-cached $(LEXICON_STAMP) if [ ! -d $(BUILDDIR) ]; then $(MAKE) dev-setup; fi cp ./util/streamplace-dev.sh $(BUILDDIR)/streamplace $(MAKE) dev-rust - $(MAKE) muxl-wasm PKG_CONFIG_PATH=$(PKG_CONFIG_PATH) \ CGO_LDFLAGS="$(MACOS_VERSION_FLAG)" \ LD_LIBRARY_PATH=$(BUILDDIR)/lib go build -tags mainnet -o $(BUILDDIR)/libstreamplace ./cmd/libstreamplace/... @@ -313,14 +312,6 @@ dev-setup-meson-configure: meson setup --default-library=shared $(BUILDDIR) $(SHARED_OPTS) meson configure --default-library=shared $(BUILDDIR) $(SHARED_OPTS) -.PHONY: muxl-wasm -muxl-wasm: - rustup target add wasm32-wasip1 - if [ "$(BUILDOS)" = "darwin" ]; then stat "$$(brew --prefix)/opt/llvm/bin/clang" >/dev/null 2>&1 || (echo "llvm not installed, run 'brew install llvm' and try again" && exit 1); fi \ - && export PATH="$$(brew --prefix)/opt/llvm/bin:$$PATH" \ - && CC="" cargo build -p muxl-wasm --target wasm32-wasip1 --release \ - && cp target/wasm32-wasip1/release/muxl-wasm.wasm pkg/muxl/muxl.wasm - .PHONY: dev-rust dev-rust: .build/bin/uniffi-bindgen-go-forked cargo build @@ -335,7 +326,7 @@ dev-rust: .build/bin/uniffi-bindgen-go-forked && mv $(BUILDDIR)/lib/libiroh_streamplace.$$EXT.tmp $(BUILDDIR)/lib/libiroh_streamplace.$$EXT .PHONY: dev-test -dev-test: muxl-wasm +dev-test: go install github.com/jstemmer/go-junit-report/v2@latest \ && PKG_CONFIG_PATH=$(PKG_CONFIG_PATH) \ LD_LIBRARY_PATH=$(shell realpath $(BUILDDIR))/lib \ @@ -634,11 +625,11 @@ desktop-windows-amd64: && mv "js/desktop/out/make/squirrel.windows/x64/Streamplace-$(VERSION_ELECTRON) Setup.exe" ./bin/streamplace-desktop-$(VERSION)-windows-amd64.exe .PHONY: streamplace -streamplace: app-cached meson-setup-static muxl-wasm +streamplace: app-cached meson-setup-static meson compile -C $(BUILDDIR) streamplace | grep -v drectve .PHONY: archive -archive: app-cached meson-setup-static godeps muxl-wasm +archive: app-cached meson-setup-static godeps meson compile -C $(BUILDDIR) archive | grep -v drectve .PHONY: linux-amd64 diff --git a/go.mod b/go.mod index 5af39ad7..65c55666 100644 --- a/go.mod +++ b/go.mod @@ -38,7 +38,6 @@ require ( github.com/golangci/golangci-lint/v2 v2.1.6 github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 - github.com/hyphacoop/go-dasl v0.8.0 github.com/ipfs/go-cid v0.5.0 github.com/ipfs/go-ipld-cbor v0.2.0 github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 @@ -65,10 +64,10 @@ require ( github.com/slok/go-http-metrics v0.13.0 github.com/starttoaster/prometheus-exporter-scraper v0.0.1 github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac + github.com/streamplace/muxl/go v0.0.0-00010101000000-000000000000 github.com/streamplace/oatproxy v0.0.0-20260508220721-f8852e8dbf44 github.com/stretchr/testify v1.11.1 github.com/tdewolff/canvas v0.0.0-20250728095813-50d4cb1eee71 - github.com/tetratelabs/wazero v1.11.0 github.com/tus/tusd/v2 v2.8.0 github.com/urfave/cli/v3 v3.6.2 github.com/whyrusleeping/cbor-gen v0.3.1 @@ -312,6 +311,7 @@ require ( github.com/hexops/gotextdiff v1.0.3 // indirect github.com/holiman/uint256 v1.3.2 // indirect github.com/hyphacoop/cbor/v2 v2.0.0-20251007204234-2a4fa83e606e // indirect + github.com/hyphacoop/go-dasl v0.8.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/invopop/yaml v0.3.1 // indirect github.com/ipfs/bbloom v0.0.4 // indirect @@ -496,6 +496,7 @@ require ( github.com/tdewolff/minify/v2 v2.23.4 // indirect github.com/tdewolff/parse/v2 v2.8.0 // indirect github.com/tetafro/godot v1.5.1 // indirect + github.com/tetratelabs/wazero v1.11.0 // indirect github.com/thales-e-security/pool v0.0.2 // indirect github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67 // indirect github.com/timonwong/loggercheck v0.11.0 // indirect @@ -571,3 +572,5 @@ require ( mvdan.cc/unparam v0.0.0-20250301125049-0df0534333a4 // indirect star-tex.org/x/tex v0.7.1 // indirect ) + +replace github.com/streamplace/muxl/go => ../muxl/go diff --git a/pkg/muxl/muxl.go b/pkg/muxl/muxl.go index af3955bc..544d76f4 100644 --- a/pkg/muxl/muxl.go +++ b/pkg/muxl/muxl.go @@ -1,938 +1,213 @@ +// Package muxl wraps github.com/streamplace/muxl/go — the upstream MUXL/S2PA +// Go library, with its muxl-sign toolchain embedded as wasm — behind the +// package-level API streamplace has historically used. The wazero runtime, +// host imports (host_sign/host_sha256), and linear-memory tuning all live +// upstream now; this package is a thin adapter that builds one process-wide +// Engine and delegates to it. There is no longer a local wasm build step. package muxl import ( - "bytes" "context" - "crypto/rand" - "crypto/sha256" - "errors" - "fmt" "io" - "sort" - "strings" "sync" - "sync/atomic" - "testing/fstest" - _ "embed" - - "github.com/hyphacoop/go-dasl/drisl" - "github.com/tetratelabs/wazero" - "github.com/tetratelabs/wazero/api" - "github.com/tetratelabs/wazero/experimental" - "github.com/tetratelabs/wazero/imports/wasi_snapshot_preview1" - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/trace" - "stream.place/streamplace/pkg/log" + upstream "github.com/streamplace/muxl/go" ) -var muxlTracer = otel.Tracer("muxl") - -var moduleCounter atomic.Uint64 - -// MuxlEvent represents one event from the muxl segmenter/concatenator -// stdout stream. The wire format is `crate::cbor::CborEvent` on the muxl -// Rust side — a `type`-tagged union with "init" and "segment" variants. -// muxl-sign's `sign-segment` subcommand emits the same shape with a -// "signed-segment" type tag (the per-track bytes carry a leading -// c2pa-uuid box); we decode it into this same struct. -// Fields not relevant to a given Type are left zero. -type MuxlEvent struct { - Type string `cbor:"type"` - Number uint32 `cbor:"number,omitempty"` - - // --- Init event --- - // Data is the full ftyp+moov for ALL tracks combined — same bytes - // the live ingest path writes straight to its output. - Data []byte `cbor:"data,omitempty"` - // Catalog describes per-track codec/dimensions/timescale/etc. - Catalog *MuxlCatalog `cbor:"catalog,omitempty"` - // TrackInits is per-track standalone ftyp+moov bytes (one per track) - // keyed by stringified track ID. Used to construct HLS per-track - // init segments addressable by their own BDASL CID. - TrackInits map[string][]byte `cbor:"track_inits,omitempty"` - - // --- Segment event --- - // Tracks is per-track moof+mdat bytes keyed by stringified track ID. - Tracks map[string][]byte `cbor:"tracks,omitempty"` - // Durations is per-track segment duration in timescale ticks. - Durations map[string]uint64 `cbor:"durations,omitempty"` - // SampleCounts is per-track sample (frame) count for this segment. - SampleCounts map[string]uint32 `cbor:"sample_counts,omitempty"` - // BodySize is the total bytes this segment contributes to the - // concatenated output (sum of len(Tracks[*])). - BodySize uint64 `cbor:"body_size,omitempty"` - // DurationUs is the segment's playable wall duration in microseconds. - DurationUs uint64 `cbor:"duration_us,omitempty"` -} - -// MuxlCatalog mirrors `crate::catalog::Catalog` from muxl (the Rust -// authoritative type). We mirror only the fields the metafile pipeline -// needs; unknown fields are ignored on decode. -type MuxlCatalog struct { - Video *MuxlCatalogVideo `cbor:"video,omitempty"` - Audio *MuxlCatalogAudio `cbor:"audio,omitempty"` -} - -type MuxlCatalogVideo struct { - Renditions map[string]MuxlVideoConfig `cbor:"renditions"` -} - -type MuxlCatalogAudio struct { - Renditions map[string]MuxlAudioConfig `cbor:"renditions"` -} - -type MuxlVideoConfig struct { - Codec string `cbor:"codec"` - Container MuxlContainer `cbor:"container"` - CodedWidth uint32 `cbor:"codedWidth"` - CodedHeight uint32 `cbor:"codedHeight"` -} - -type MuxlAudioConfig struct { - Codec string `cbor:"codec"` - Container MuxlContainer `cbor:"container"` - SampleRate uint32 `cbor:"sampleRate"` - NumberOfChannels uint32 `cbor:"numberOfChannels"` -} - -// MuxlContainer is the tagged-union container descriptor. Kind is -// either "cmaf" (the streamplace default) or "legacy". For "cmaf" -// the timescale/trackId fields are populated; for "legacy" they are -// zero. -type MuxlContainer struct { - Kind string `cbor:"kind"` - Timescale uint32 `cbor:"timescale,omitempty"` - TrackID uint32 `cbor:"trackId,omitempty"` -} - -// TrackID returns the configured CMAF track ID, or 0 for legacy. -func (c MuxlVideoConfig) TrackID() uint32 { return c.Container.TrackID } - -// TrackID returns the configured CMAF track ID, or 0 for legacy. -func (c MuxlAudioConfig) TrackID() uint32 { return c.Container.TrackID } - -// Timescale returns the media timescale (ticks per second), or 0 for legacy. -func (c MuxlVideoConfig) Timescale() uint32 { return c.Container.Timescale } - -// Timescale returns the media timescale (ticks per second), or 0 for legacy. -func (c MuxlAudioConfig) Timescale() uint32 { return c.Container.Timescale } - -// muxl.wasm is built from rust/muxl-wasm via `make muxl-wasm`. It bundles -// the full muxl-sign CLI — both unsigned subcommands (segment, concat, -// catalog, fmp4, mp4, hls) and the signing ones (sign-per-track, -// sign-segment) — so this package only needs one wasm artifact. -// -//go:embed muxl.wasm -var wasmBytes []byte - -var wasmRuntime wazero.Runtime - -// Compile the wasm module exactly once and reuse the result; instantiation -// is cheap, compilation is not. The signer runs for the length of a stream -// and the wrap/verify helpers run per segment, so the difference adds up fast. -var ( - compileOnce sync.Once - compiled wazero.CompiledModule - compileErr error +// Re-exported upstream types under the names streamplace callers use. These are +// aliases, so values (events, catalogs, signer inputs) flow between this package +// and upstream with no conversion and the existing call sites are unchanged. +type ( + MuxlEvent = upstream.Event + MuxlCatalog = upstream.Catalog + MuxlCatalogVideo = upstream.CatalogVideo + MuxlCatalogAudio = upstream.CatalogAudio + MuxlVideoConfig = upstream.VideoConfig + MuxlAudioConfig = upstream.AudioConfig + MuxlContainer = upstream.Container + SignerInput = upstream.SignerInput ) -// signerRegistry holds the per-instance host-sign closure used by -// muxl-sign's `--host-sign` mode. The wasm import looks the closure up by -// the instance name (which we make unique per call via moduleCounter), so -// concurrent signs don't collide. RunMuxlSignSegment registers a closure on -// entry and deletes it on return. -var signerRegistry sync.Map // string → func([]byte) ([]byte, error) +// SignerToCallback adapts a crypto.Signer into the host-sign callback that +// SignerInput.Sign expects (SHA-256 digest, ECDSA DER → fixed-width r‖s). See +// the upstream doc. +var SignerToCallback = upstream.SignerToCallback -// hostSignErr is the sentinel u32 muxl-sign's host_sign import returns to -// signal "the host couldn't sign this" — anything other than a real -// signature length. -const hostSignErr = ^uint32(0) +// --- engine singleton ------------------------------------------------------- -// memoryConfig holds the per-instance wasm linear memory tuning. wazero's -// default allocator reallocs+memcpys on every memory.grow page, so a -// module that ends up at 50MB allocates ~25GB of cumulative slices on its -// way there. The custom allocator below pre-allocates the backing buffer -// and grows geometrically, so a typical segment never reallocs at all. -// -// initial is the upfront capacity of the backing []byte; max is a hard -// ceiling — Reallocate returns nil past it, which surfaces to the wasm -// module as a memory.grow failure. Defaults are conservative; Configure -// overrides them from CLI flags. var ( - memoryConfigMu sync.RWMutex - memoryInitialBytes uint64 = 50 * 1024 * 1024 - memoryMaxBytes uint64 = 1024 * 1024 * 1024 + engineOnce sync.Once + engine *upstream.WASMEngine + engineErr error + + memMu sync.Mutex + memInitial = uint64(50 * 1024 * 1024) + memMax = uint64(1024 * 1024 * 1024) ) -// Configure sets the wasm linear memory tuning used by all subsequent -// RunMuxl* calls. Safe to call concurrently with in-flight calls (they'll -// keep their existing allocator) but typically called once at startup. +// Configure sets the wasm linear-memory tuning (initial backing capacity and +// hard ceiling) applied when the Engine is first built. Call it before the +// first muxl operation — streamplace does, from its CLI bootstrap. func Configure(initialBytes, maxBytes uint64) { - memoryConfigMu.Lock() - defer memoryConfigMu.Unlock() - memoryInitialBytes = initialBytes - memoryMaxBytes = maxBytes -} - -func memoryConfigSnapshot() (initial, max uint64) { - memoryConfigMu.RLock() - defer memoryConfigMu.RUnlock() - return memoryInitialBytes, memoryMaxBytes -} - -// muxlAllocator implements experimental.MemoryAllocator. Stateless apart -// from the configured ceilings and the per-call ctx/instance used to log -// cap-exceeded events; each Allocate call produces a fresh -// muxlLinearMemory. -type muxlAllocator struct { - ctx context.Context - instanceName string - initialBytes uint64 - maxBytes uint64 -} - -func (a *muxlAllocator) Allocate(capHint, wasmMax uint64) experimental.LinearMemory { - effectiveMax := a.maxBytes - if wasmMax > 0 && wasmMax < effectiveMax { - effectiveMax = wasmMax - } - initial := a.initialBytes - if initial < capHint { - initial = capHint - } - if initial > effectiveMax { - initial = effectiveMax - } - return &muxlLinearMemory{ - ctx: a.ctx, - instanceName: a.instanceName, - buf: make([]byte, 0, initial), - max: effectiveMax, - } -} - -// muxlLinearMemory is the per-instance backing buffer. Reallocate keeps -// the same slice (no copy) when the new size fits in the existing -// capacity; otherwise it doubles capacity (geometric growth) up to max, -// or returns nil if the requested size exceeds max. -type muxlLinearMemory struct { - ctx context.Context - instanceName string - buf []byte - max uint64 -} - -func (m *muxlLinearMemory) Reallocate(size uint64) []byte { - if size > m.max { - log.Error(m.ctx, "muxl memory cap exceeded", - "instance", m.instanceName, - "requested_bytes", size, - "max_bytes", m.max, - ) - return nil - } - if size <= uint64(cap(m.buf)) { - m.buf = m.buf[:size] - return m.buf - } - newCap := uint64(cap(m.buf)) * 2 - if newCap < size { - newCap = size - } - if newCap > m.max { - newCap = m.max - } - newBuf := make([]byte, size, newCap) - copy(newBuf, m.buf) - m.buf = newBuf - return m.buf -} - -func (m *muxlLinearMemory) Free() { - m.buf = nil -} - -func init() { - ctx := context.Background() - wasmRuntime = wazero.NewRuntime(ctx) - wasi_snapshot_preview1.MustInstantiate(ctx, wasmRuntime) - - // Register the `muxl` host module that muxl.wasm imports. The - // imports are declared unconditionally on the wasm side (they're part - // of the binary's import table whether or not the corresponding wasm - // path is exercised) so the host module must always exist. PEM-mode - // sign invocations simply never call into host_sign; in-wasm SHA-256 - // invocations never call into host_sha256. - _, err := wasmRuntime.NewHostModuleBuilder("muxl"). - NewFunctionBuilder(). - WithFunc(hostSign). - Export("host_sign"). - NewFunctionBuilder(). - WithFunc(hostSha256). - Export("host_sha256"). - Instantiate(ctx) - if err != nil { - panic(fmt.Errorf("registering muxl host module: %w", err)) - } -} - -// hostSha256 is the trampoline behind muxl-sign's -// `muxl.host_sha256` import. Reads the input from wasm linear -// memory at (dataPtr, dataLen), hashes it with native Go's SHA-256 -// (which has hardware-accelerated paths via the `crypto/sha256` package -// on amd64/arm64), and writes the 32-byte digest back at outPtr. -// -// Used today by the bench-sha256 subcommand to size the upper bound on -// what host SHA-256 saves vs in-wasm sha2; if the win is real and the -// patch story for c2pa-rs's sha2 dep gets settled, this becomes the -// hot-path implementation for all hashing too. -func hostSha256(ctx context.Context, mod api.Module, dataPtr, dataLen, outPtr uint32) { - _, span := muxlTracer.Start(ctx, "muxl.hostSha256", trace.WithAttributes( - attribute.String("instance", mod.Name()), - attribute.Int64("data_len", int64(dataLen)), - )) - defer span.End() - - data, ok := mod.Memory().Read(dataPtr, dataLen) - if !ok { - log.Error(ctx, "host_sha256: bad data pointer/length", "instance", mod.Name(), "ptr", dataPtr, "len", dataLen) - span.SetAttributes(attribute.String("error", "bad data pointer")) - return - } - span.AddEvent("read input bytes") - sum := sha256.Sum256(data) - span.AddEvent("hashed") - if !mod.Memory().Write(outPtr, sum[:]) { - log.Error(ctx, "host_sha256: bad output pointer", "instance", mod.Name(), "ptr", outPtr) - span.SetAttributes(attribute.String("error", "bad output pointer")) - } -} - -// hostSign is the trampoline behind muxl-sign's `muxl.host_sign` -// import. It looks up the per-instance closure registered by -// RunMuxlSignSegment, hands it the bytes to sign, and writes the signature -// back into wasm memory. Returns the signature length on success or -// hostSignErr on any failure. -func hostSign(ctx context.Context, mod api.Module, dataPtr, dataLen, outPtr, outMax uint32) uint32 { - ctx, span := muxlTracer.Start(ctx, "muxl.hostSign", trace.WithAttributes( - attribute.String("instance", mod.Name()), - attribute.Int64("data_len", int64(dataLen)), - )) - defer span.End() - - v, ok := signerRegistry.Load(mod.Name()) - if !ok { - log.Error(ctx, "host_sign called with no signer registered", "instance", mod.Name()) - span.SetAttributes(attribute.String("error", "no signer registered")) - return hostSignErr - } - signFn := v.(func([]byte) ([]byte, error)) - span.AddEvent("registry lookup ok") - - data, ok := mod.Memory().Read(dataPtr, dataLen) - if !ok { - log.Error(ctx, "host_sign: bad data pointer/length", "instance", mod.Name(), "ptr", dataPtr, "len", dataLen) - span.SetAttributes(attribute.String("error", "bad data pointer")) - return hostSignErr - } - span.AddEvent("read input bytes") - - signCtx, signSpan := muxlTracer.Start(ctx, "muxl.hostSign.signFn") - sig, err := signFn(data) - signSpan.End() - _ = signCtx - if err != nil { - log.Error(ctx, "host_sign: signer returned error", "instance", mod.Name(), "error", err) - span.SetAttributes(attribute.String("error", err.Error())) - return hostSignErr - } - span.SetAttributes(attribute.Int("sig_len", len(sig))) - - if uint32(len(sig)) > outMax { - log.Error(ctx, "host_sign: signature too long for output buffer", "instance", mod.Name(), "len", len(sig), "max", outMax) - span.SetAttributes(attribute.String("error", "signature too long")) - return hostSignErr - } - if !mod.Memory().Write(outPtr, sig) { - log.Error(ctx, "host_sign: bad output pointer", "instance", mod.Name(), "ptr", outPtr) - span.SetAttributes(attribute.String("error", "bad output pointer")) - return hostSignErr - } - span.AddEvent("wrote signature") - return uint32(len(sig)) -} - -func getModule(ctx context.Context) (wazero.CompiledModule, error) { - compileOnce.Do(func() { - _, span := muxlTracer.Start(ctx, "muxl.CompileModule", trace.WithAttributes( - attribute.Int("wasm_bytes", len(wasmBytes)), - )) - compiled, compileErr = wasmRuntime.CompileModule(ctx, wasmBytes) - span.End() + memMu.Lock() + defer memMu.Unlock() + memInitial, memMax = initialBytes, maxBytes +} + +// getEngine compiles the embedded muxl wasm once and returns the shared Engine. +// Compilation is the expensive step; each operation instantiates a fresh, +// isolated module internally, so the Engine is safe for concurrent use. +func getEngine() (*upstream.WASMEngine, error) { + engineOnce.Do(func() { + memMu.Lock() + initial, max := memInitial, memMax + memMu.Unlock() + // Process-lifetime resource — build under Background so a cancelled + // request context can't tear down the shared engine. + engine, engineErr = upstream.NewWASM(context.Background(), upstream.WithMemory(initial, max)) }) - if compileErr != nil { - return nil, fmt.Errorf("error compiling muxl wasm module: %w", compileErr) - } - return compiled, nil + return engine, engineErr } -// RunMuxlSegmenterEvents decodes the muxl segmenter's DRISL stream and -// delivers each *MuxlEvent (init + per-GoP segment, carrying the catalog, -// per-track init segments, durations, and sample counts) on eventCh. This is -// the unsigned counterpart of RunMuxlSignSegment, suitable for driving the -// live HLS writer (pkg/livehls) without a signing key. eventCh is NOT closed -// by this call; the caller closes it once the function returns. -func RunMuxlSegmenterEvents(ctx context.Context, input io.Reader, eventCh chan *MuxlEvent) error { - mod, err := getModule(ctx) - if err != nil { - return err - } - return runMuxlWith(ctx, mod, []string{"muxl-wasm", "segment", "-", "--stdout"}, nil, false, input, nil, nil, nil, nil, eventCh) -} - -// RunMuxlUnwrapEvents re-derives the per-track event stream from a STORED -// MUXL wrapper (bare .m4s, fMP4, or flat MP4) via muxl's `unwrap --events`. -// It decodes the same DRISL event stream as RunMuxlSegmenterEvents (init + -// per-GoP segment), but the per-track bytes are the stored canonical segments -// verbatim — any C2PA/S2PA signature is preserved — with durations and sample -// counts recomputed from the moofs. This is the unified entry point for -// driving live HLS / live-to-VOD / DVR off already-stored signed segments -// (e.g. from ValidateMP4, covering both local and replicated segments). -// -// Deterministic (fake clock): pure structural re-derivation, no signing. -// eventCh is NOT closed by this call; the caller closes it once the function -// returns. -func RunMuxlUnwrapEvents(ctx context.Context, input io.Reader, eventCh chan *MuxlEvent) error { - mod, err := getModule(ctx) - if err != nil { - return err - } - return runMuxlWith(ctx, mod, []string{"muxl-wasm", "unwrap", "--events", "-"}, nil, false, input, nil, nil, nil, nil, eventCh) -} +// --- delegating helpers ----------------------------------------------------- -// RunMuxlWrap synthesizes a presentation MP4 from a MUXL wrapper — a bare -// .m4s segment stream, a MUXL fMP4, or a flat MP4 — via muxl's `wrap` -// subcommand. format is "fmp4" (appendable: ftyp+moov(init) + verbatim -// segments) or "flat" (finalized faststart). The segment bytes, and any -// C2PA/S2PA signatures over them, pass through untouched; only the -// ftyp+moov header is synthesized from the segments' embedded catalogs. -// -// This is the inbound header-synthesis that makes a stored canonical .m4s -// understandable to gstreamer / transmux / players. Deterministic (fake -// clock — pure structural assembly, no signing). +// RunMuxlWrap synthesizes a presentation MP4 from a MUXL wrapper. format is +// "fmp4" or "flat" (default fmp4). func RunMuxlWrap(ctx context.Context, input io.Reader, format string, output io.Writer) error { - if format == "" { - format = "fmp4" - } - mod, err := getModule(ctx) + eng, err := getEngine() if err != nil { return err } - return runMuxlWith(ctx, mod, []string{"muxl-wasm", "wrap", "-", "-", "--format", format}, nil, false, input, output, nil, nil, nil, nil) + return eng.Wrap(ctx, input, format, output) } -// RunMuxlWrapInit synthesizes only the per-stream init segment (ftyp+moov) -// from a MUXL wrapper's embedded catalogs — the HLS EXT-X-MAP target. -// Equivalent to `wrap --format fmp4 --init-only`. +// RunMuxlWrapInit synthesizes only the per-stream init segment (ftyp+moov) — +// the HLS EXT-X-MAP target — from a MUXL wrapper's embedded catalogs. func RunMuxlWrapInit(ctx context.Context, input io.Reader, output io.Writer) error { - mod, err := getModule(ctx) + eng, err := getEngine() if err != nil { return err } - return runMuxlWith(ctx, mod, []string{"muxl-wasm", "wrap", "-", "-", "--format", "fmp4", "--init-only"}, nil, false, input, output, nil, nil, nil, nil) + return eng.WrapInit(ctx, input, output) } -// RunMuxlVerify validates the C2PA/S2PA signatures on a signed MUXL wrapper -// (bare .m4s stream, fMP4, or flat MP4) entirely inside the wasm sandbox via -// muxl-sign's `verify` subcommand, returning the manifest+cert+validation -// JSON document: -// -// {"segments":[{"track_id":N,"manifest":{..},"cert":"", -// "validation_results":{..},"validation_state":".."}, ..]} -// -// one entry per canonical segment. This replaces the iroh-streamplace c2pa -// uniffi binding (get_manifest_and_cert) for segment validation: each -// segment verifies standalone as the "m4s" asset it was signed as, so no -// synthesized header is involved in the hash. Runs against the real clock so -// any cert-validity-window checks observe wall time. +// RunMuxlVerify validates the C2PA/S2PA signatures on a signed MUXL wrapper and +// returns the per-segment manifest+cert+validation JSON document. func RunMuxlVerify(ctx context.Context, input io.Reader) (string, error) { - mod, err := getModule(ctx) + eng, err := getEngine() if err != nil { return "", err } - var out bytes.Buffer - if err := runMuxlWith(ctx, mod, []string{"muxl-wasm", "verify"}, nil, true, input, &out, nil, nil, nil, nil); err != nil { - return "", err + return eng.Verify(ctx, input) +} + +// RunMuxlUnwrapEvents re-derives the per-track event stream from a stored MUXL +// wrapper (bare .m4s, fMP4, or flat MP4), bytes and signatures verbatim. +func RunMuxlUnwrapEvents(ctx context.Context, input io.Reader, eventCh chan *MuxlEvent) error { + eng, err := getEngine() + if err != nil { + return err } - return out.String(), nil + return eng.UnwrapEvents(ctx, input, eventCh) } -// RunMuxlConcatenatorEvents concatenates MUXL-compatible fMP4 archives -// (init+segments) into a single fMP4 stream; if the init segment changes a -// new init is emitted. Bytes route to initCh/segCh, and each decoded -// *MuxlEvent is also sent on eventCh (if non-nil) so the caller can inspect -// per-segment metadata (durations, sample counts, per-track init segments) -// for sidecar metafile generation. -func RunMuxlConcatenatorEvents(ctx context.Context, input io.Reader, initCh chan []byte, segCh chan []byte, eventCh chan *MuxlEvent) error { - mod, err := getModule(ctx) +// RunMuxlSegmenterEvents segments an fMP4 stream into per-GoP canonical MUXL +// events (unsigned). +func RunMuxlSegmenterEvents(ctx context.Context, input io.Reader, eventCh chan *MuxlEvent) error { + eng, err := getEngine() if err != nil { return err } - return runMuxlWith(ctx, mod, []string{"muxl-wasm", "concat"}, nil, false, input, nil, nil, initCh, segCh, eventCh) + return eng.SegmentEvents(ctx, input, eventCh) } -// RunMuxlSignSegment streams an fMP4 input through muxl-sign's -// `sign-segment` subcommand: the muxl segmenter splits the input -// per-GoP and C2PA-signs each canonical segment in place, so the bytes -// routed to segCh are [c2pa-uuid][muxl-uuid][moof][mdat] per track. The -// init/segment/event channels behave exactly like -// RunMuxlConcatenatorEvents — the wire stream is the same DRISL event -// format, just with a "signed-segment" type tag in place of "segment". -// -// Signing needs the real wall clock (c2pa-rs checks cert validity at -// sign time and draws COSE nonces from real randomness), so unlike the -// plain segmenter this runs with realClock=true and its output is not -// byte-stable across runs. The segment bytes come from the input reader; the -// signing backend is selected like the rest of muxl-sign: -// exactly one of in.KeyPEM (in-wasm PEM signing) or in.Sign (host-callback, -// for hardware-backed keys via the wasm host_sign import) must be set. +// RunMuxlSignSegment segments an fMP4 stream and S2PA-signs each canonical +// segment in place. Exactly one of in.KeyPEM or in.Sign must be set. func RunMuxlSignSegment(ctx context.Context, input io.Reader, in SignerInput, initCh chan []byte, segCh chan []byte, eventCh chan *MuxlEvent) error { - hasKey := len(in.KeyPEM) > 0 - hasSign := in.Sign != nil - if hasKey == hasSign { - return fmt.Errorf("muxl: exactly one of SignerInput.KeyPEM or SignerInput.Sign must be set") - } - if in.Alg == "" { - in.Alg = "es256k" - } - mod, err := getModule(ctx) + eng, err := getEngine() if err != nil { return err } - keysFS := fstest.MapFS{ - "cert.pem": {Data: in.CertPEM}, - "track.json": {Data: in.TrackManifest}, - "wrapper.json": {Data: in.WrapperManifest}, - } - args := []string{ - "muxl-wasm", "sign-segment", - "--cert", "/keys/cert.pem", - "--alg", in.Alg, - "--track-manifest", "/keys/track.json", - "--wrapper-manifest", "/keys/wrapper.json", - } - if hasKey { - keysFS["key.pem"] = &fstest.MapFile{Data: in.KeyPEM} - args = append(args, "--key", "/keys/key.pem") - } else { - args = append(args, "--host-sign") - } - fsCfg := wazero.NewFSConfig().WithFSMount(keysFS, "/keys") - return runMuxlWith(ctx, mod, args, fsCfg, true, input, nil, in.Sign, initCh, segCh, eventCh) + return eng.SignSegment(ctx, input, in, initCh, segCh, eventCh) } -// SignerInput is the input bundle for RunMuxlSignSegment. Exactly one -// of KeyPEM or Sign must be set: -// -// - KeyPEM: the streamer's PKCS#8-PEM private key bytes. Sent into the -// wasm sandbox via a read-only FS mount; signing happens inside wasm -// using c2pa-rs. Use this for software keys where the bytes are -// readily available (e.g. atproto-derived stream keys). -// - Sign: a host-side closure that takes pre-hashed-or-not data (per -// c2pa's CallbackSigner contract: ECDSA receives the unhashed bytes -// and the closure is expected to do SHA-256 + sign + raw r||s) and -// returns the signature. Use this for hardware-backed signers -// (PKCS#11, EIP-712) whose key bytes never leave the host. Powered -// by the wasm `streamplace.host_sign` import — see hostSign. -// -// Cert chain is always PEM bytes, leaf first. Manifests are JSON bodies -// already substituted with per-segment values (timestamps etc.). -type SignerInput struct { - CertPEM []byte - KeyPEM []byte - Sign func(data []byte) ([]byte, error) - TrackManifest []byte - WrapperManifest []byte - // Alg defaults to "es256k" when empty. - Alg string -} +// --- push-style Concatenator ------------------------------------------------ -// Concatenator accepts full fMP4 archives (init+segments) and produces -// deduplicated output: init segments are emitted only when they change, -// and segment data is emitted without the init header, suitable for -// concatenation into a single fMP4 stream. -// -// Usage: +// Concatenator is a push wrapper over the Engine: Write whole fMP4 archives, +// receive processed output on the channels, Close to finish. It mirrors the +// upstream Concatenator but adds an engine-construction error path on Close — +// the upstream value can't be built without a ready Engine, and callers expect +// construction never to fail (the error surfaces on Close). // // cat := muxl.NewConcatenator(ctx) // go func() { cat.Write(fullFmp4Archive); cat.Close() }() // initSeg := <-cat.InitCh // for seg := range cat.SegCh { /* append to output */ } type Concatenator struct { - stdinWriter *io.PipeWriter - InitCh chan []byte - SegCh chan []byte - // EventCh emits the full *MuxlEvent for every wasm event in - // addition to the byte-level dispatch on InitCh/SegCh. Use this - // when you need per-segment metadata (durations, sample counts, - // per-track init segments) — for instance, building the HLS - // metafile sidecar. Closed alongside InitCh/SegCh on shutdown. + // InitCh receives an init segment only when the track configuration + // changes. SegCh receives concatenable segment bodies (signed, for the + // signing segmenter). EventCh receives the full *MuxlEvent with per-segment + // metadata. All three are closed when processing finishes. + InitCh chan []byte + SegCh chan []byte EventCh chan *MuxlEvent - done chan error -} -// NewConcatenator starts the WASM concat process in the background. -// Write full fMP4 archives via Write(), receive processed output on -// InitCh, SegCh, and EventCh. -// -// InitCh receives a new init segment only when the track configuration changes. -// SegCh receives raw segment data (moof+mdat) that can be concatenated after an init. -// EventCh receives the full *MuxlEvent — same data plus per-segment metadata -// (durations, sample counts, per-track init segments) needed by the HLS -// metafile builder. All three channels are closed when the concatenator -// finishes (after Close + WASM exit). -func NewConcatenator(ctx context.Context) *Concatenator { - initCh := make(chan []byte, 1) - segCh := make(chan []byte, 16) - eventCh := make(chan *MuxlEvent, 16) - stdinReader, stdinWriter := io.Pipe() - done := make(chan error, 1) - - c := &Concatenator{ - stdinWriter: stdinWriter, - InitCh: initCh, - SegCh: segCh, - EventCh: eventCh, - done: done, - } - - go func() { - err := RunMuxlConcatenatorEvents(ctx, stdinReader, initCh, segCh, eventCh) - close(initCh) - close(segCh) - close(eventCh) - done <- err - }() - - return c + write func([]byte) error + closeFn func() error } -// NewSigningSegmenter is the signing counterpart to NewConcatenator: it -// drives muxl-sign's `sign-segment` subcommand instead of `concat`, so -// the bytes emitted on SegCh are C2PA-signed canonical segments -// ([c2pa-uuid][muxl-uuid][moof][mdat] per track). The Concatenator -// plumbing (Write/Close + the three channels) is identical; only the -// underlying wasm subcommand differs. Feed full fMP4 archives via -// Write(); receive signed output on InitCh, SegCh, and EventCh. -func NewSigningSegmenter(ctx context.Context, in SignerInput) *Concatenator { - initCh := make(chan []byte, 1) - segCh := make(chan []byte, 16) - eventCh := make(chan *MuxlEvent, 16) - stdinReader, stdinWriter := io.Pipe() - done := make(chan error, 1) +// Write feeds a full fMP4 archive (init+segments) to the pipeline. +func (c *Concatenator) Write(data []byte) error { return c.write(data) } - c := &Concatenator{ - stdinWriter: stdinWriter, - InitCh: initCh, - SegCh: segCh, - EventCh: eventCh, - done: done, - } +// Close signals end of input and waits for the pipeline to finish; the output +// channels are closed by the time it returns. +func (c *Concatenator) Close() error { return c.closeFn() } - go func() { - err := RunMuxlSignSegment(ctx, stdinReader, in, initCh, segCh, eventCh) - close(initCh) - close(segCh) - close(eventCh) - done <- err - }() - - return c -} - -// Write feeds a full fMP4 archive (init+segments) to the concatenator. -func (c *Concatenator) Write(data []byte) error { - _, err := c.stdinWriter.Write(data) - return err -} - -// Close signals that no more data will be written. The WASM process will -// finish processing and the output channels will be closed. -func (c *Concatenator) Close() error { - c.stdinWriter.Close() - return <-c.done -} - -// logWriter adapts WASM stderr output to log calls, one message per line. -// Lines tagged "Error:" are surfaced at error level so signing failures -// aren't lost in debug noise. -type logWriter struct { - ctx context.Context - instanceID uint64 - buf []byte -} - -func (w *logWriter) Write(p []byte) (int, error) { - w.buf = append(w.buf, p...) - for { - i := 0 - for i < len(w.buf) && w.buf[i] != '\n' { - i++ - } - if i >= len(w.buf) { - break - } - line := string(w.buf[:i]) - w.buf = w.buf[i+1:] - if strings.HasPrefix(line, "Error:") || strings.HasPrefix(line, "thread '") { - log.Error(w.ctx, "muxl wasm error", "instance", w.instanceID, "msg", line) - } else { - log.Debug(w.ctx, "muxl wasm", "instance", w.instanceID, "msg", line) - } +// NewConcatenator drives Engine.ConcatEvents: dedup/concatenate MUXL fMP4 +// archives into one stream (a new init is emitted only when the catalog +// changes). +func NewConcatenator(ctx context.Context) *Concatenator { + eng, err := getEngine() + if err != nil { + return failedConcatenator(err) } - return len(p), nil -} - -// stderrWriter lets tests override how wasm stderr is captured. Production -// uses logWriter. Tests can swap in os.Stderr to surface clap/c2pa errors -// directly (slog's default config drops debug-level logs in tests). -var stderrWriter func(ctx context.Context, instanceID uint64) io.Writer = func(ctx context.Context, instanceID uint64) io.Writer { - return &logWriter{ctx: ctx, instanceID: instanceID} + return adopt(upstream.NewConcatenator(ctx, eng)) } -// runMuxlWith instantiates a precompiled wasm module with the given args and -// optional FS mount. If initCh+segCh are non-nil, stdout is parsed as DRISL -// events and routed to those channels; otherwise if stdout is non-nil the -// module's stdout writes go straight there. If input is non-nil, it's piped -// to the module's stdin. If signFn is non-nil it's registered against the -// instance's name for the duration of the call so that wasm calls into -// `streamplace.host_sign` route to it. realClock=true exposes the host's -// wall clock and real randomness — c2pa-rs needs both for cert validity -// checks and COSE sign nonces. The segmenter intentionally runs against -// wazero's fake clock so its output stays byte-stable across runs. -func runMuxlWith(ctx context.Context, mod wazero.CompiledModule, args []string, fsCfg wazero.FSConfig, realClock bool, input io.Reader, stdout io.Writer, signFn func([]byte) ([]byte, error), initCh chan []byte, segCh chan []byte, eventCh chan *MuxlEvent) error { - instanceID := moduleCounter.Add(1) - instanceName := fmt.Sprintf("muxl-%d", instanceID) - - ctx, span := muxlTracer.Start(ctx, "muxl.runMuxlWith", trace.WithAttributes( - attribute.String("instance", instanceName), - attribute.StringSlice("args", args), - attribute.Bool("real_clock", realClock), - attribute.Bool("has_input", input != nil), - attribute.Bool("has_stdout", stdout != nil), - attribute.Bool("has_sign_fn", signFn != nil), - attribute.Bool("parse_events", initCh != nil && segCh != nil), - )) - defer span.End() - - if signFn != nil { - signerRegistry.Store(instanceName, signFn) - defer signerRegistry.Delete(instanceName) - span.AddEvent("registered host signer") - } - - cfg := wazero.NewModuleConfig(). - WithName(instanceName). - WithStderr(stderrWriter(ctx, instanceID)). - WithArgs(args...) - if realClock { - cfg = cfg. - WithSysWalltime(). - WithSysNanotime(). - WithSysNanosleep(). - WithRandSource(rand.Reader) - } - if fsCfg != nil { - cfg = cfg.WithFSConfig(fsCfg) - } - - var stdinReader *io.PipeReader - var stdinWriter *io.PipeWriter - if input != nil { - stdinReader, stdinWriter = io.Pipe() - cfg = cfg.WithStdin(stdinReader) - } - - var stdoutReader *io.PipeReader - var stdoutWriter *io.PipeWriter - parseEvents := initCh != nil || segCh != nil || eventCh != nil - if parseEvents { - stdoutReader, stdoutWriter = io.Pipe() - cfg = cfg.WithStdout(stdoutWriter) - } else if stdout != nil { - cfg = cfg.WithStdout(stdout) - } - span.AddEvent("config built") - - initialBytes, maxBytes := memoryConfigSnapshot() - allocator := &muxlAllocator{ - ctx: ctx, - instanceName: instanceName, - initialBytes: initialBytes, - maxBytes: maxBytes, - } - - errCh := make(chan error, 1) - go func() { - // Span covers the wasm's entire run from instantiation through - // exit + cleanup. Host calls (host_sign, host_sha256) made - // during execution will be children of this span via the ctx - // wazero passes through. - instCtx, instSpan := muxlTracer.Start(ctx, "muxl.wasm.InstantiateModule", trace.WithAttributes( - attribute.String("instance", instanceName), - attribute.Int64("memory_initial_bytes", int64(initialBytes)), - attribute.Int64("memory_max_bytes", int64(maxBytes)), - )) - instCtx = experimental.WithMemoryAllocator(instCtx, allocator) - instance, err := wasmRuntime.InstantiateModule(instCtx, mod, cfg) - instSpan.End() - if err != nil { - log.Error(ctx, "error instantiating module", "error", err) - } - // wazero leaves the module registered on clean exit; close to free - // its WASM memory. Without this the signer leaks ~10MB per segment. - if instance != nil { - closeCtx, closeSpan := muxlTracer.Start(ctx, "muxl.wasm.Instance.Close", trace.WithAttributes( - attribute.String("instance", instanceName), - )) - closeErr := instance.Close(closeCtx) - closeSpan.End() - if closeErr != nil { - log.Error(ctx, "error closing wasm module", "error", closeErr) - } - } - if stdoutWriter != nil { - stdoutWriter.Close() - } - errCh <- err - }() - - if input != nil { - go func() { - _, copySpan := muxlTracer.Start(ctx, "muxl.wasm.stdinCopy", trace.WithAttributes( - attribute.String("instance", instanceName), - )) - n, err := io.Copy(stdinWriter, input) - copySpan.SetAttributes(attribute.Int64("bytes_copied", n)) - copySpan.End() - if err != nil && !errors.Is(err, io.ErrClosedPipe) { - log.Error(ctx, "error copying input to stdin", "error", err) - } - stdinWriter.Close() - }() - } - - if parseEvents { - _, parseSpan := muxlTracer.Start(ctx, "muxl.wasm.parseEvents") - err := ParseMuxlEvents(ctx, stdoutReader, initCh, segCh, eventCh) - parseSpan.End() - if err != nil { - return fmt.Errorf("parsing events: %w", err) - } - } - - span.AddEvent("waiting on wasm exit") - if wasmErr := <-errCh; wasmErr != nil { - return fmt.Errorf("wasm execution: %w", wasmErr) +// NewSigningSegmenter drives Engine.SignSegment: each canonical segment is +// S2PA-signed in place, so SegCh carries [c2pa-uuid][muxl-uuid][moof][mdat] per +// track. Exactly one of in.KeyPEM or in.Sign must be set. +func NewSigningSegmenter(ctx context.Context, in SignerInput) *Concatenator { + eng, err := getEngine() + if err != nil { + return failedConcatenator(err) } - span.AddEvent("wasm exited") - return nil + return adopt(upstream.NewSigningSegmenter(ctx, eng, in)) } -// SignerToCallback wraps a crypto.Signer for use as SignerInput.Sign. -// -// Hashes data with SHA-256 (matching c2pa's CallbackSigner contract for -// SHA-256-family algs — ECDSA P-256/secp256k1, RSA PS256), calls the -// signer, and converts ECDSA DER output to raw r||s. byteLen is the -// curve's coordinate byte size: 32 for ES256/ES256K, 48 for ES384, 66 -// for ES512. For RSA-PSS algs the byteLen argument is ignored and the -// raw signer output is passed through. -// -// The crypto.Signer can be a software ecdsa.PrivateKey, a PKCS#11 -// hardware signer, an EIP-712 wallet wrapper, etc. — anything implementing -// the standard interface. -func SignerToCallback(signer cryptoSigner, byteLen int) func([]byte) ([]byte, error) { - return func(data []byte) ([]byte, error) { - // Note: no ctx threading — the sync hostSign trampoline already - // holds a span open for "muxl.hostSign.signFn" that this work is - // running under. Sub-spans here would only show up if the - // closure was called directly from a Go context; not worth the - // allocation for the common (host_sign-driven) path. - digest := sha256Sum(data) - sig, err := signer.Sign(rand.Reader, digest[:], cryptoSHA256) - if err != nil { - return nil, fmt.Errorf("muxl host sign: %w", err) - } - raw, ok := derECDSAToRaw(sig, byteLen) - if !ok { - return sig, nil - } - return raw, nil +func adopt(c *upstream.Concatenator) *Concatenator { + return &Concatenator{ + InitCh: c.InitCh, + SegCh: c.SegCh, + EventCh: c.EventCh, + write: c.Write, + closeFn: c.Close, } } -// ParseMuxlEvents decodes the muxl wasm's DRISL event stream and -// dispatches each event. Bytes are routed to initCh/segCh (legacy API, -// used by live ingest). If eventCh is non-nil it ALSO receives the -// full *MuxlEvent — used by callers that need the per-segment metadata -// (durations, sample counts, per-track init segments) to build a -// metafile sidecar. Any/all of the three channels may be nil. -// -// Per-segment per-track bytes are concatenated for segCh in sorted -// track-ID order (lex on the stringified ID) so the byte layout of the -// concatenated output is deterministic. Same ordering is what the -// metafile builder uses to compute per-track byte offsets. -func ParseMuxlEvents(ctx context.Context, r io.Reader, initCh chan []byte, segCh chan []byte, eventCh chan *MuxlEvent) error { - decoder := drisl.NewDecoder(r) - - for { - var ev MuxlEvent - err := decoder.Decode(&ev) - if errors.Is(err, io.EOF) { - break - } - if err != nil { - return fmt.Errorf("decode muxl event: %w", err) - } - switch ev.Type { - case "init": - if initCh != nil { - select { - case <-ctx.Done(): - return nil - case initCh <- ev.Data: - } - } - case "segment", "signed-segment": - if segCh != nil { - keys := make([]string, 0, len(ev.Tracks)) - for k := range ev.Tracks { - keys = append(keys, k) - } - sort.Strings(keys) - combined := []byte{} - for _, k := range keys { - combined = append(combined, ev.Tracks[k]...) - } - select { - case <-ctx.Done(): - return nil - case segCh <- combined: - } - } - default: - return fmt.Errorf("unknown event type: %s", ev.Type) - } - if eventCh != nil { - select { - case <-ctx.Done(): - return nil - case eventCh <- &ev: - } - } +// failedConcatenator returns a Concatenator with closed output channels whose +// Write/Close report err, preserving the "engine error surfaces on Close" +// contract callers relied on. +func failedConcatenator(err error) *Concatenator { + initCh, segCh, eventCh := make(chan []byte), make(chan []byte), make(chan *MuxlEvent) + close(initCh) + close(segCh) + close(eventCh) + return &Concatenator{ + InitCh: initCh, + SegCh: segCh, + EventCh: eventCh, + write: func([]byte) error { return err }, + closeFn: func() error { return err }, } - - return nil } diff --git a/pkg/muxl/muxl_sign_test.go b/pkg/muxl/muxl_sign_test.go index b2dcec74..697d1e60 100644 --- a/pkg/muxl/muxl_sign_test.go +++ b/pkg/muxl/muxl_sign_test.go @@ -2,7 +2,6 @@ package muxl import ( "context" - "io" "os" "path/filepath" "sync" @@ -13,10 +12,6 @@ import ( "stream.place/streamplace/test/remote" ) -func init() { - stderrWriter = func(_ context.Context, _ uint64) io.Writer { return os.Stderr } -} - // TestConcatenatorRealSegments drives the wasm Concatenator with real // signed flat MP4 segments and confirms it // emits one segment event per input file (minus one held back until diff --git a/pkg/muxl/sign_adapter.go b/pkg/muxl/sign_adapter.go deleted file mode 100644 index 4eb7e42c..00000000 --- a/pkg/muxl/sign_adapter.go +++ /dev/null @@ -1,41 +0,0 @@ -package muxl - -import ( - "crypto" - "crypto/sha256" - "encoding/asn1" - "math/big" -) - -// cryptoSigner is the crypto.Signer interface, aliased so SignerToCallback -// reads tidily without dragging the crypto import into muxl.go. -type cryptoSigner = crypto.Signer - -// cryptoSHA256 is crypto.SHA256 as a SignerOpts value. -var cryptoSHA256 crypto.SignerOpts = crypto.SHA256 - -func sha256Sum(data []byte) [32]byte { - return sha256.Sum256(data) -} - -// derECDSAToRaw converts a DER-encoded ECDSA signature (the SEQUENCE { r, -// s } shape Go's crypto.Signer returns for ECDSA keys) to the fixed-width -// r||s format c2pa expects from CallbackSigner closures. -// -// byteLen is the curve's coordinate width: 32 for P-256/secp256k1, 48 for -// P-384, 66 for P-521. r and s are left-padded with zeros to that width. -// -// Returns ok=false if the input doesn't look like ASN.1 DER ECDSA — in -// that case the caller should pass the bytes through unmodified (e.g. -// RSA-PSS signatures aren't DER-shaped). -func derECDSAToRaw(der []byte, byteLen int) ([]byte, bool) { - var sig struct{ R, S *big.Int } - rest, err := asn1.Unmarshal(der, &sig) - if err != nil || len(rest) != 0 || sig.R == nil || sig.S == nil { - return nil, false - } - out := make([]byte, 2*byteLen) - sig.R.FillBytes(out[:byteLen]) - sig.S.FillBytes(out[byteLen:]) - return out, true -} diff --git a/pkg/muxl/stress_test.go b/pkg/muxl/stress_test.go deleted file mode 100644 index 32035596..00000000 --- a/pkg/muxl/stress_test.go +++ /dev/null @@ -1,64 +0,0 @@ -package muxl - -import ( - "bytes" - "context" - "os" - "strconv" - "strings" - "testing" - - "github.com/stretchr/testify/require" -) - -// TestSha256Bench drives muxl-wasm's `bench-sha256` subcommand once for -// each backend and prints the per-iter timings + throughput. This is the -// upper bound on what a host-SHA256 path could save us on the c2pa-rs -// hashing hot loop — the in-wasm baseline is sha2 running through wazero, -// the host path is the streamplace.host_sha256 import (Go's -// crypto/sha256, which uses CPU SHA-NI extensions where available). -// -// Skipped by default; opt in with STREAMPLACE_STRESS=1. Knobs: -// -// STREAMPLACE_STRESS_SHA256_SIZE bytes per iteration default 1 MB -// STREAMPLACE_STRESS_SHA256_ITERATIONS iteration count default 200 -func TestSha256Bench(t *testing.T) { - if os.Getenv("STREAMPLACE_STRESS") == "" { - t.Skip("set STREAMPLACE_STRESS=1 to run") - } - - size := envInt(t, "STREAMPLACE_STRESS_SHA256_SIZE", 1024*1024) - iters := envInt(t, "STREAMPLACE_STRESS_SHA256_ITERATIONS", 200) - ctx := context.Background() - mod, err := getModule(ctx) - require.NoError(t, err) - - runBench := func(mode string) string { - var stdout bytes.Buffer - args := []string{ - "muxl-wasm", "bench-sha256", - "--size", strconv.Itoa(size), - "--iterations", strconv.Itoa(iters), - "--mode", mode, - } - err := runMuxlWith(ctx, mod, args, nil, true, nil, &stdout, nil, nil, nil, nil) - require.NoError(t, err) - return strings.TrimSpace(stdout.String()) - } - - t.Logf("sha256 bench: size=%d iterations=%d", size, iters) - t.Log(runBench("wasm")) - t.Log(runBench("host")) -} - -func envInt(t *testing.T, key string, def int) int { - v := os.Getenv(key) - if v == "" { - return def - } - n, err := strconv.Atoi(v) - if err != nil { - t.Fatalf("invalid %s=%q: %v", key, v, err) - } - return n -} diff --git a/rust/muxl-wasm/Cargo.toml b/rust/muxl-wasm/Cargo.toml deleted file mode 100644 index fe61e7cc..00000000 --- a/rust/muxl-wasm/Cargo.toml +++ /dev/null @@ -1,15 +0,0 @@ -[package] -name = "muxl-wasm" -version = "0.1.0" -edition = "2024" -license = "Apache-2.0" - -[[bin]] -name = "muxl-wasm" -path = "src/main.rs" - -[dependencies] - -muxl-sign = { git = "https://github.com/streamplace/muxl.git", rev = "a1cfe7a4250835a55fb440d02221ce6206d45408", package = "muxl-sign" } -# temporary local version to test -# muxl-sign = { path = "../../../muxl/crates/muxl-sign" } diff --git a/rust/muxl-wasm/src/main.rs b/rust/muxl-wasm/src/main.rs deleted file mode 100644 index cf2a0786..00000000 --- a/rust/muxl-wasm/src/main.rs +++ /dev/null @@ -1,3 +0,0 @@ -fn main() { - muxl_sign::cli_main(); -}