diff --git a/Cargo.lock b/Cargo.lock index ad9251a5f7..a6f50aa5f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4314,20 +4314,14 @@ version = "0.0.1" dependencies = [ "anyhow", "axum", - "axum-server", "bytes", - "clap", "hang", "moq-mux", - "moq-native", "moq-net", "reqwest", - "rustls", - "sd-notify", "str0m", "thiserror 2.0.18", "tokio", - "tower-http 0.7.0", "tracing", "url", "uuid", @@ -4338,24 +4332,18 @@ name = "moq-rtmp" version = "0.0.1" dependencies = [ "anyhow", - "axum", - "axum-server", "bytes", - "clap", "futures", "moq-mux", "moq-native", "moq-net", "rml_rtmp", "rustls", - "sd-notify", "socket2 0.6.4", "thiserror 2.0.18", "tokio", "tokio-rustls", - "tower-http 0.7.0", "tracing", - "url", ] [[package]] @@ -4363,23 +4351,14 @@ name = "moq-srt" version = "0.0.1" dependencies = [ "anyhow", - "axum", - "axum-server", "bytes", - "clap", "futures", - "humantime", "moq-mux", - "moq-native", "moq-net", - "rustls", - "sd-notify", "srt-tokio", "thiserror 2.0.18", "tokio", - "tower-http 0.7.0", "tracing", - "url", ] [[package]] diff --git a/doc/.vitepress/config.ts b/doc/.vitepress/config.ts index 440e64e5d7..cac2c84510 100644 --- a/doc/.vitepress/config.ts +++ b/doc/.vitepress/config.ts @@ -137,8 +137,6 @@ export default defineConfig({ ], }, { text: "CLI", link: "/bin/cli" }, - { text: "WebRTC", link: "/bin/rtc" }, - { text: "RTMP", link: "/bin/rtmp" }, { text: "OBS", link: "/bin/obs" }, { text: "GStreamer", link: "/bin/gstreamer" }, { text: "Web", link: "/bin/web" }, diff --git a/doc/bin/index.md b/doc/bin/index.md index 0aa79c9b7b..233b9c1515 100644 --- a/doc/bin/index.md +++ b/doc/bin/index.md @@ -29,18 +29,6 @@ Another tool does the encoding (ex. ffmpeg), making it easy to pipe any media in ffmpeg -f avfoundation -i "0" -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts ``` -## [moq-rtc](/bin/rtc) - -A WebRTC <-> MoQ gateway. Speaks WHIP (publish) and WHEP (subscribe) in either -HTTP role, so it can accept incoming peers (OBS, browsers) or dial out to a -remote WebRTC server. Ingest and egress both work for H.264, VP8, VP9, and Opus. - -## [moq-rtmp](/bin/rtmp) - -An RTMP / enhanced-RTMP -> MoQ ingest gateway. Accepts RTMP from any encoder -(OBS, ffmpeg) and publishes it into MoQ, supporting H.264/HEVC/AV1/VP9 and -AAC/Opus/AC-3. - ## [OBS Plugin](/bin/obs) Real-time latency with the familiar OBS interface. diff --git a/doc/bin/rtc.md b/doc/bin/rtc.md deleted file mode 100644 index c764c84c2c..0000000000 --- a/doc/bin/rtc.md +++ /dev/null @@ -1,120 +0,0 @@ ---- -title: moq-rtc -description: WebRTC <-> MoQ gateway (WHIP/WHEP) ---- - -# moq-rtc - -`moq-rtc` bridges WebRTC and Media over QUIC. It speaks -[WHIP](https://datatracker.ietf.org/doc/html/rfc9725) (publish) and WHEP -(subscribe) in **either HTTP role**, so it can either accept incoming peers -or dial out to a remote WebRTC server. - -## The 2x2 - -| Subcommand | WebRTC role | Direction | Status | -|---|---|---|---| -| `server publish` | accept WHIP publishes | RTP into MoQ | working | -| `client subscribe` | dial a remote WHEP URL | RTP into MoQ | working | -| `server subscribe` | serve WHEP subscriptions | MoQ -> RTP | working | -| `client publish` | dial a remote WHIP URL | MoQ -> RTP | working | - -All four paths work. The egress paths use str0m's Frame API to packetize -MoQ frames back into RTP; the per-codec adapters live in `codec::Track` -and are the same shape regardless of HTTP role. - -### Keyframe latency on the egress side - -A freshly-connected WHEP / WHIP-out peer subscribes at the *current* -(in-progress) MoQ group, which begins at a keyframe, so it gets a -decodable start without waiting for the next GOP boundary. If the peer -loses keyframe packets, str0m fulfils its NACK retransmissions from the -video send buffer, which is sized to cover a large keyframe plus the rest -of the current group. MoQ has no PLI path back to the publisher, so -`KeyframeRequest` (PLI/FIR) events from the peer are logged but not -propagated upstream. - -The egress paths (WHEP server, WHIP client) negotiate H.264, H.265, VP8, -VP9, AV1, and Opus. The ingest paths (WHIP server, WHEP client) currently -accept H.264, VP8, VP9, and Opus; H.265 / AV1 ingest is a follow-up. - -## CLI shape - -Mirrors `moq-cli`: globals first, then HTTP role, then direction. - -```bash -# server publish (WHIP server): accept publishes into MoQ -moq-rtc --relay https://relay.example.com --broadcast my-stream \ - server --listen 0.0.0.0:8088 publish - -# client subscribe (WHEP client): pull from a remote WHEP source -moq-rtc --relay https://relay.example.com --broadcast cam0 \ - client --url https://camera.example.com/whep/cam0 subscribe - -# server subscribe (WHEP server): serve a MoQ broadcast over WHEP -moq-rtc --relay https://relay.example.com --broadcast my-stream \ - server --listen 0.0.0.0:8088 subscribe - -# client publish (WHIP client): push a MoQ broadcast to a remote WHIP endpoint -moq-rtc --relay https://relay.example.com --broadcast my-stream \ - client --url https://twitch.tv/whip publish -``` - -### Global flags - -- `--relay`: upstream MoQ relay to publish to / subscribe from. -- `--broadcast`: MoQ broadcast name this gateway binds to. -- `--public-addr`: optional public UDP socket address(es) to advertise as - ICE host candidates. Repeat the flag (or comma-separate) for dual-stack - IPv4 + IPv6 deployments. When empty, str0m discovers peer-reflexive - candidates via STUN binding requests, which works for most NAT - scenarios. - -### Server flags - -- `--listen`: HTTP bind address (default `[::]:8088`). -- `--udp-bind`: UDP address the shared WebRTC media socket binds to - (default `0.0.0.0:0`, i.e. an OS-picked port for dev/loopback). Every - WHIP/WHEP session shares this one port (demuxed by ICE ufrag), so a - deployment behind a firewall pins it (e.g. `0.0.0.0:8089`) and opens just - that one media port. Pair it with `--public-addr` so the advertised ICE - candidate uses the pinned port. -- `--tls-cert` / `--tls-key`: serve HTTPS instead. Most WHIP clients - require it in practice. - -### Client flags - -- `--url`: remote WHIP or WHEP resource URL. - -### Session teardown - -The bundled WHIP/WHEP servers honor an HTTP `DELETE` to the resource URL -returned in the `Location` header (`//`), per -RFC 9725. It ends the session promptly, releasing its broadcast -announcement and shared-media-port registration instead of waiting for the -ICE disconnect timeout. Embedders that own their own routing can call -`Server::terminate(resource_id)` to do the same. - -## Codec mapping - -| WebRTC codec | MoQ catalog | Egress | Ingest | -|--------------|-------------|--------|--------| -| Opus | `AudioCodec::Opus`, 48 kHz / stereo | yes | yes | -| H.264 | `VideoCodec::H264` (avc3 inline or avc1 + `avcC`) | yes | yes (avc3) | -| H.265 | `VideoCodec::H265` (hev1 inline or hvc1 + `hvcC`) | yes | no | -| VP8 | `VideoCodec::VP8` | yes | yes | -| VP9 | `VideoCodec::VP9` | yes | yes | -| AV1 | `VideoCodec::AV1` | yes | no | - -On egress, `codec::Track` reshapes each rendition into what str0m's Frame -API expects. Opus / VP8 / VP9 / AV1 and inline-parameter H.264 (avc3) / -H.265 (hev1) pass through untouched. Out-of-band-parameter H.264 (avc1) and -H.265 (hvc1) are rewritten from length-prefixed NALU to Annex-B with the -parameter sets (SPS/PPS, plus VPS for H.265) prepended to each keyframe. -This reuses `moq-mux`'s `h264::Avcc::parse` / `h265::Hvcc::parse` and `annexb` -helpers, the same logic its own avc1/hvc1 transmuxers use. - -On ingest, H.264 is reassembled by str0m as Annex-B; `moq-mux`'s H.264 -importer in `Avc3` mode publishes the inline-parameter shape directly, which -lines up with what the WebCodecs decoder in `@moq/watch` already expects. No -extra conversion needed in the gateway. diff --git a/doc/bin/rtmp.md b/doc/bin/rtmp.md deleted file mode 100644 index 9ba666fba0..0000000000 --- a/doc/bin/rtmp.md +++ /dev/null @@ -1,136 +0,0 @@ ---- -title: moq-rtmp -description: RTMP / enhanced-RTMP <-> MoQ gateway (ingest and egress) ---- - -# moq-rtmp - -`moq-rtmp` bridges [RTMP](https://en.wikipedia.org/wiki/Real-Time_Messaging_Protocol) -(the protocol OBS, ffmpeg, and most hardware encoders and players speak) and -Media over QUIC, in **both directions**: - -- **Publish (ingest):** an encoder pushes a stream in, and `moq-rtmp` publishes it - into MoQ as an ordinary broadcast. -- **Play (egress):** a player pulls `rtmp://host//`, and `moq-rtmp` - subscribes to that broadcast from MoQ and streams it back down. VLC, ffplay, and - mpv can play it (browsers can't -- Flash is dead). - -RTMP carries media as FLV-format audio/video messages. `moq-rtmp` runs the RTMP -handshake and chunk/AMF session (via the pure-Rust -[`rml_rtmp`](https://crates.io/crates/rml_rtmp), no librtmp). On ingest it re-wraps -each message as an FLV tag and feeds it to `moq-mux`'s FLV demuxer; on egress it -muxes the broadcast back to FLV with `moq-mux` and sends the tags out as RTMP -messages. It's the sibling of `moq-srt` (SRT/MPEG-TS) and `moq-rtc` (WHIP/WHEP). - -Both **legacy RTMP** (H.264 + AAC, plus MP3) and **enhanced RTMP** (E-RTMP: the -HEVC, AV1, VP9, Opus, AC-3, and MP3 FourCC payloads) are supported in each -direction, because all codec handling lives in the `moq-mux` FLV demuxer/muxer. -Legacy players that speak only H.264 + AAC will reject the E-RTMP codecs on the -play path. - -## CLI shape - -The binary has two modes, mirroring `moq-srt`: - -```bash -# serve: ingest RTMP and serve it directly as a local relay -moq-rtmp serve --server-bind [::]:443 --tls-generate localhost \ - --rtmp-listen 0.0.0.0:1935 --rtmp-prefix live/ - -# publish: ingest RTMP and forward broadcasts to a remote relay -moq-rtmp publish --relay https://relay.example.com \ - --rtmp-listen 0.0.0.0:1935 --rtmp-prefix live/ -``` - -Point any encoder at it. In OBS, set the server to `rtmp://127.0.0.1:1935/live` -and the stream key to `cam0`; with ffmpeg: - -```bash -# Lands at broadcast `live/cam0`. -ffmpeg -re -i input.mp4 -c copy -f flv rtmp://127.0.0.1:1935/live/cam0 -``` - -Then play the same broadcast back out over RTMP from any player: - -```bash -# Pulls broadcast `live/cam0` (the same URL it was pushed to). -ffplay rtmp://127.0.0.1:1935/live/cam0 -mpv rtmp://127.0.0.1:1935/live/cam0 -vlc rtmp://127.0.0.1:1935/live/cam0 -``` - -### `serve` flags - -- `--server-bind`: QUIC/WebTransport bind address (default `[::]:443`). Also - serves the `/certificate.sha256` endpoint browsers need for self-signed - `http://` origins, and a static player directory with `--dir`. -- `--tls-generate ` / `--tls-cert` / `--tls-key`: server TLS. - -### `publish` flags - -- `--relay`: upstream MoQ relay to publish every ingested broadcast into. - -### RTMP flags - -- `--rtmp-listen`: TCP bind address for the RTMP server (default `[::]:1935`). -- `--rtmp-prefix`: prepended to every broadcast path, to namespace a listener's - streams (e.g. `live/`). - -### RTMPS flags - -RTMPS (RTMP over TLS, `rtmps://`) is served on a second listener alongside -plaintext RTMP, sharing the same `--rtmp-prefix`: - -- `--rtmps-listen`: TCP bind address for the RTMPS server (off unless set). RTMPS - has no well-known port; 443 or a custom one are common. -- `--rtmps-tls-cert` / `--rtmps-tls-key`: PEM certificate chain and key. -- `--rtmps-tls-generate `: or generate a throwaway self-signed cert - (testing only; clients must disable verification). - -```bash -moq-rtmp serve --server-bind [::]:443 --tls-generate localhost \ - --rtmp-listen 0.0.0.0:1935 \ - --rtmps-listen 0.0.0.0:1936 --rtmps-tls-cert cert.pem --rtmps-tls-key key.pem \ - --rtmp-prefix live/ -``` - -## Routing - -Each connection's broadcast path is `/` from the RTMP app and stream -key (`rtmp://host//`), falling back to just the app when the key is -empty, with `--rtmp-prefix` prepended. The same routing applies to both -directions, so the URL round-trips: push to `rtmp://host/live/cam0`, then pull it -back from `rtmp://host/live/cam0`. - -A play waits for the broadcast to be announced, so a player can connect slightly -before the publisher. First **publisher** on a path wins (a second publish to a -live path is rejected); **plays** don't claim a path, so any number of players can -pull the same broadcast at once. In `serve` mode plays are served from the same -origin the server exposes, so anything in it -- RTMP ingests and otherwise -- can -be pulled back out over RTMP. - -## Notes and limitations - -- **Auth.** The binary (and the `moq_rtmp::run` convenience) is unauthenticated: - anyone who can reach the TCP port can publish or play. Gate it with a host - firewall or a private network. To authenticate, embed the library and drive its - `Server` / `Request` API: `Server::accept` yields a `Request` that is either a - `Publish` or a `Play`, and you verify the app / stream key (e.g. the stream key - as a moq-token JWT) before accepting it into / out of an origin at a path of your - choosing, or rejecting it -- no callback, the policy lives in your loop. -- **Embedding.** A relay can run the gateway in-process by depending on the - `moq-rtmp` library (`default-features = false`). Call `moq_rtmp::run` against its - own origin for the unauthenticated case (publishers ingest into it, players are - served out of it), or use `Server` / `Request` to plug in the relay's existing - JWT/path auth and scope the origin per token. Either way the media stays local - with no extra hop. -- **RTMPS.** Embedders can terminate TLS themselves: set `Config::tls` (or - `Server::with_tls`) with a `rustls::ServerConfig`, or accept the connection and - finish the TLS handshake by hand and hand the stream to `moq_rtmp::accept_stream` - (which works over any `AsyncRead + AsyncWrite` transport). -- **Codecs.** FLAC enhanced-audio payloads are dropped (no MoQ catalog codec); - everything else (H.264/HEVC/AV1/VP9 video, AAC/MP3/Opus/AC-3/E-AC-3 audio) is - supported. MP3 is accepted as both the legacy SoundFormat 2 tag and the E-RTMP - `.mp3` FourCC, and re-muxed as the legacy tag on export. - -(Written by Claude) diff --git a/rs/CLAUDE.md b/rs/CLAUDE.md index 58da45f356..b573d4a764 100644 --- a/rs/CLAUDE.md +++ b/rs/CLAUDE.md @@ -31,8 +31,10 @@ Layered roughly transport -> container/format -> media -> apps/bindings. **Apps / binaries** - `moq-relay` (lib+bin): clusterable, media-agnostic relay. axum HTTP API, JWT auth, WebSocket fallback, clustering. Config/TOML merge pattern lives here (see below). -- `moq-cli` (lib+bin, `moq`): serve/accept/publish/subscribe; stdin/stdout media piping. -- `moq-rtc` (lib+bin): WebRTC (WHIP/WHEP) gateway. Bridges browser WebRTC ingest/playback to MoQ broadcasts (str0m ICE/DTLS, A/V sync, NACK). Embeddable lib (`default-features = false`) + standalone binary (`server` feature). +- `moq-cli` (lib+bin, `moq`): serve/accept/publish/subscribe + `hls`; stdin/stdout media piping. The CLI surface for the gateway library crates below lives here (a `moq-cli`-driven command-line interface). +- `moq-rtc` (lib): WebRTC (WHIP/WHEP) gateway. Bridges browser WebRTC ingest/playback to MoQ broadcasts (str0m ICE/DTLS, A/V sync, NACK). Embed via its axum routers / `Client`. +- `moq-rtmp` (lib): RTMP / enhanced-RTMP gateway (ingest + egress, `rml_rtmp`, FLV via `moq-mux`). RTMPS (rustls + tokio-rustls) is the optional `tls` feature. +- `moq-srt` (lib): bidirectional SRT gateway (MPEG-TS via `srt-tokio` + `moq-mux`). - `moq-bench` (bin): relay load generator. `JoinSet`-spawned staggered connections, rand sampling. - `moq-boy` (bin): crowd-controlled Game Boy emulator publisher (blocking emulator thread + async monitor tasks). - `moq-token` (lib) / `moq-token-cli` (bin): JWT auth. `Claims`, `Algorithm`, `KeyType` (EC/RSA/OCT/OKP), JWKS. CLI does generate/sign/verify. diff --git a/rs/moq-rtc/Cargo.toml b/rs/moq-rtc/Cargo.toml index 1305b19866..50e9c08df6 100644 --- a/rs/moq-rtc/Cargo.toml +++ b/rs/moq-rtc/Cargo.toml @@ -15,53 +15,23 @@ categories = ["multimedia", "network-programming", "web-programming"] [lib] doctest = false -[[bin]] -name = "moq-rtc" -path = "bin/moq-rtc.rs" -doc = false -# The binary (TLS listener, CLI, relay client) needs the `server` feature; the -# library can be depended on with `default-features = false` to embed the -# WHIP/WHEP gateway (the axum routers and WHEP/WHIP client) into another process. -required-features = ["server"] - -[features] -default = ["server", "iroh", "noq", "websocket"] -# Standalone binary: TLS-terminating listener, CORS, CLI parsing, and the -# moq-native relay client. Pulls in axum-server / clap / moq-native / rustls / -# sd-notify / tower-http. -server = ["dep:axum-server", "dep:clap", "dep:moq-native", "dep:rustls", "dep:sd-notify", "dep:tower-http"] -iroh = ["server", "moq-native/iroh"] -noq = ["server", "moq-native/noq"] -quinn = ["server", "moq-native/quinn"] -quiche = ["server", "moq-native/quiche"] -websocket = ["server", "moq-native/websocket"] - -# The embeddable gateway library needs only anyhow/axum/bytes/hang/moq-mux/ -# moq-net/reqwest/str0m/thiserror/tokio/tracing/url/uuid. The rest -# (axum-server/clap/moq-native/rustls/sd-notify/tower-http) are `optional` and -# pulled in by the `server` feature for the binary. +# Pure library: the WHIP/WHEP <-> MoQ gateway (axum routers + WHEP/WHIP client). +# The CLI lives in `moq-cli` (the `rtc` subcommand); embedders mount the routers +# / dial with `Client` against their own origin. [dependencies] anyhow = { version = "1", features = ["backtrace"] } axum = { version = "0.8", features = ["tokio"] } -axum-server = { version = "0.8", features = ["tls-rustls"], optional = true } bytes = "1" -clap = { version = "4", features = ["derive"], optional = true } hang = { workspace = true } moq-mux = { workspace = true } -moq-native = { workspace = true, default-features = false, features = ["aws-lc-rs"], optional = true } moq-net = { workspace = true } reqwest = { version = "0.13", default-features = false, features = ["rustls"] } -rustls = { version = "0.23", features = ["aws-lc-rs"], default-features = false, optional = true } str0m = "0.20" thiserror = "2" tokio = { workspace = true, features = ["full"] } -tower-http = { version = "0.7", features = ["cors"], optional = true } tracing = "0.1" url = "2" uuid = { version = "1", features = ["v4"] } -[target.'cfg(unix)'.dependencies] -sd-notify = { version = "0.5", optional = true } - [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } diff --git a/rs/moq-rtc/bin/moq-rtc.rs b/rs/moq-rtc/bin/moq-rtc.rs deleted file mode 100644 index 88a01ec59e..0000000000 --- a/rs/moq-rtc/bin/moq-rtc.rs +++ /dev/null @@ -1,263 +0,0 @@ -//! `moq-rtc` binary. -//! -//! Subcommand structure mirrors `moq-cli`: globals first -//! (`--relay`, `--broadcast`), then HTTP role (`client` / `server`), then -//! direction (`publish` / `subscribe`). The 2x2 is: -//! -//! - `server publish` -- WHIP server, ingest from a remote publisher into MoQ. -//! - `server subscribe` -- WHEP server, egress a MoQ broadcast to remote subscribers. -//! - `client subscribe` -- WHEP client, pull a remote WHEP feed into MoQ. -//! - `client publish` -- WHIP client, push a MoQ broadcast to a remote WHIP endpoint. - -use std::net::SocketAddr; -use std::path::PathBuf; - -use anyhow::Context; -use axum::Router; -use clap::{Parser, Subcommand}; -use moq_rtc::{Client, Server}; -use tower_http::cors::{Any, CorsLayer}; -use url::Url; - -#[derive(Parser, Clone)] -#[command(version)] -struct Cli { - #[command(flatten)] - log: moq_native::Log, - - /// MoQ client configuration for dialing the upstream relay. - #[command(flatten)] - moq_client: moq_native::ClientConfig, - - /// URL of the upstream MoQ relay where ingested broadcasts get - /// published and from which egressed broadcasts are pulled. - #[arg(long, env = "MOQ_RTC_RELAY")] - relay: Url, - - /// Broadcast name on the MoQ relay this gateway binds to. For - /// `server publish` this is the path that arrives at the WHIP - /// endpoint; for the other three it's a single name pinned to the - /// gateway. - #[arg(long, alias = "name", env = "MOQ_RTC_BROADCAST")] - broadcast: String, - - /// Public UDP socket address(es) to advertise as ICE host candidates. - /// Repeat the flag (or comma-separate) to advertise both an IPv4 and - /// an IPv6 candidate on a dual-stack deployment. When empty, the - /// session relies on str0m discovering peer-reflexive candidates via - /// STUN binding requests, which is enough for most NAT scenarios. - #[arg(long, env = "MOQ_RTC_PUBLIC_ADDR", value_delimiter = ',')] - public_addr: Vec, - - #[command(subcommand)] - role: Role, -} - -#[derive(Subcommand, Clone)] -enum Role { - /// Dial out: act as an HTTP client and POST an SDP offer to a remote - /// WHIP or WHEP URL. - Client { - /// Remote WHIP or WHEP resource URL. - #[arg(long, env = "MOQ_RTC_URL")] - url: Url, - - #[command(subcommand)] - direction: Direction, - }, - /// Listen: accept incoming WHIP or WHEP HTTP requests. - Server { - /// HTTP listener for the WHIP/WHEP endpoints. - #[arg(long, env = "MOQ_RTC_LISTEN", default_value = "[::]:8088")] - listen: SocketAddr, - - /// UDP socket the shared WebRTC media mux binds to. All WHIP/WHEP - /// sessions share this one port (demuxed by ICE ufrag), so open just - /// this in the firewall. `0.0.0.0:0` lets the OS pick (loopback/dev). - #[arg(long, env = "MOQ_RTC_UDP_BIND", default_value = "0.0.0.0:0")] - udp_bind: SocketAddr, - - /// Optional TLS cert (PEM). Requires `--tls-key`. - #[arg(long, env = "MOQ_RTC_TLS_CERT", requires = "tls_key")] - tls_cert: Option, - - #[arg(long, env = "MOQ_RTC_TLS_KEY", requires = "tls_cert")] - tls_key: Option, - - #[command(subcommand)] - direction: Direction, - }, -} - -#[derive(Subcommand, Clone, Copy, Debug)] -enum Direction { - /// WHIP (publish protocol): RTP flows toward the WHIP-server endpoint. - Publish, - /// WHEP (subscribe protocol): RTP flows away from the WHEP-server endpoint. - Subscribe, -} - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - rustls::crypto::aws_lc_rs::default_provider() - .install_default() - .expect("failed to install default crypto provider"); - - let Cli { - log, - moq_client, - relay, - broadcast, - public_addr, - role, - } = Cli::parse(); - log.init()?; - - let moq_client = moq_client.init().context("failed to init moq client")?; - - // Two origins so ingest and egress see independent views of the - // upstream relay. The publisher feeds the relay; the subscriber - // reads from it. - let publisher = moq_net::Origin::random().produce(); - let subscriber = moq_net::Origin::random().produce(); - let subscriber_consumer = subscriber.consume(); - - let reconnect = moq_client - .with_publish(publisher.consume()) - .with_consume(subscriber) - .reconnect(relay.clone()); - - tracing::info!(%relay, %broadcast, "starting moq-rtc"); - - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - let driver = run_role(role, &broadcast, public_addr, publisher, subscriber_consumer); - - tokio::select! { - res = driver => res, - res = reconnect.closed() => res.map_err(Into::into), - _ = tokio::signal::ctrl_c() => Ok(()), - } -} - -async fn run_role( - role: Role, - broadcast: &str, - public_addr: Vec, - publisher: moq_net::OriginProducer, - subscriber: moq_net::OriginConsumer, -) -> anyhow::Result<()> { - match role { - Role::Server { - listen, - udp_bind, - tls_cert, - tls_key, - direction, - } => { - run_server( - public_addr, - udp_bind, - publisher, - subscriber, - listen, - tls_cert, - tls_key, - direction, - ) - .await - } - Role::Client { url, direction } => { - run_client(broadcast, public_addr, publisher, subscriber, url, direction).await - } - } -} - -#[allow(clippy::too_many_arguments)] -async fn run_server( - public_addr: Vec, - udp_bind: SocketAddr, - publisher: moq_net::OriginProducer, - subscriber: moq_net::OriginConsumer, - listen: SocketAddr, - tls_cert: Option, - tls_key: Option, - direction: Direction, -) -> anyhow::Result<()> { - let mut config = moq_rtc::server::Config::default(); - config.ice_candidates = public_addr; - config.udp_bind = udp_bind; - let server = Server::new(config, publisher, subscriber); - - let app = match direction { - Direction::Publish => Router::new().merge(server.publish_router()), - Direction::Subscribe => Router::new().merge(server.subscribe_router()), - }; - let app = app.layer(CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any)); - - tracing::info!(%listen, ?direction, "moq-rtc server listening"); - serve(app, listen, tls_cert, tls_key).await -} - -async fn run_client( - broadcast_name: &str, - public_addr: Vec, - publisher: moq_net::OriginProducer, - subscriber: moq_net::OriginConsumer, - url: Url, - direction: Direction, -) -> anyhow::Result<()> { - let mut config = moq_rtc::client::Config::default(); - config.ice_candidates = public_addr; - let client = Client::new(config); - - match direction { - Direction::Subscribe => { - // WHEP client: receive remote RTP, publish it as `broadcast_name`. - // The announcement lives as long as `broadcast` (moved into the client - // subscribe task below) and is withdrawn when that producer closes. - let broadcast = moq_net::Broadcast::new().produce(); - let consumer = broadcast.consume(); - if !publisher.publish_broadcast(broadcast_name, consumer) { - anyhow::bail!("failed to publish broadcast {broadcast_name}"); - } - client.subscribe(url, broadcast).await?; - } - Direction::Publish => { - // WHIP client: read the local broadcast, push as RTP to remote. - // Once the per-codec re-packetizer lands, this should poll - // `subscriber.announced()` to await the broadcast rather than - // erroring on first-miss. - let broadcast = subscriber - .request_broadcast(broadcast_name) - .await - .map_err(|_| anyhow::anyhow!("broadcast {} not announced", broadcast_name))?; - client.publish(url, broadcast).await?; - } - } - - // `client.subscribe` spawns the session in the background; block on - // ctrl-c instead of returning so the binary stays up. - tokio::signal::ctrl_c().await?; - Ok(()) -} - -async fn serve(app: Router, bind: SocketAddr, cert: Option, key: Option) -> anyhow::Result<()> { - let service = app.into_make_service(); - match (cert, key) { - (Some(cert), Some(key)) => { - let config = axum_server::tls_rustls::RustlsConfig::from_pem_file(cert, key) - .await - .context("failed to load TLS cert/key")?; - axum_server::bind_rustls(bind, config).serve(service).await?; - } - (None, None) => { - axum_server::bind(bind).serve(service).await?; - } - // clap's `requires` already gates this at parse time; the explicit - // arm is belt-and-suspenders in case someone strips the attribute. - (Some(_), None) | (None, Some(_)) => anyhow::bail!("--tls-cert and --tls-key must be set together"), - } - Ok(()) -} diff --git a/rs/moq-rtc/src/lib.rs b/rs/moq-rtc/src/lib.rs index 2bb66626be..937f322dfb 100644 --- a/rs/moq-rtc/src/lib.rs +++ b/rs/moq-rtc/src/lib.rs @@ -16,15 +16,12 @@ //! //! ## Embedding //! -//! Depend on this crate with `default-features = false` to embed the gateway -//! in another process: build a [`Server`] over your own +//! Build a [`Server`] over your own //! [`OriginProducer`](moq_net::OriginProducer) / //! [`OriginConsumer`](moq_net::OriginConsumer) and merge //! [`Server::publish_router`] / [`Server::subscribe_router`] into your own axum -//! app, or dial out with [`Client`]. That lean build skips the standalone -//! binary's deps (axum-server, clap, moq-native, rustls, sd-notify, tower-http), -//! which the `server` feature (on by default) pulls back in for the `moq-rtc` -//! binary. +//! app, or dial out with [`Client`]. A command-line interface is provided by the +//! `moq-cli` binary, on top of this library. //! //! The bundled routers are unauthenticated: they derive the broadcast name from //! the request path. To own the HTTP route and authorize requests yourself diff --git a/rs/moq-rtmp/Cargo.toml b/rs/moq-rtmp/Cargo.toml index dace71ae31..9486e328ed 100644 --- a/rs/moq-rtmp/Cargo.toml +++ b/rs/moq-rtmp/Cargo.toml @@ -15,40 +15,21 @@ categories = ["multimedia", "network-programming", "web-programming"] [lib] doctest = false -[[bin]] -name = "moq-rtmp" -path = "bin/moq-rtmp.rs" -doc = false -# The binary (relay client/server, CLI) needs the `server` feature; the library -# can be depended on with `default-features = false` for ingest only (e.g. a -# relay that embeds the RTMP listener and publishes into its own origin). -required-features = ["server"] - [features] -default = ["server", "noq", "websocket"] -# Relay client/server transports + the moq-rtmp binary. Pulls in moq-native / -# axum / clap / rustls. -server = ["dep:axum", "dep:axum-server", "dep:clap", "dep:moq-native", "dep:rustls", "dep:sd-notify", "dep:tokio-rustls", "dep:tower-http", "dep:url"] -noq = ["server", "moq-native/noq"] -quinn = ["server", "moq-native/quinn"] -quiche = ["server", "moq-native/quiche"] -websocket = ["server", "moq-native/websocket"] -iroh = ["server", "moq-native/iroh"] +default = ["tls"] +# RTMPS (RTMP over TLS): the `Conn::Tls` variant, `Server::with_tls`, +# `Config::tls`, and the `pub use rustls`. Plaintext RTMP needs none of this, so +# an embedder can drop TLS with `default-features = false`. +tls = ["dep:rustls", "dep:tokio-rustls"] -# The ingest library needs only anyhow/bytes/moq-mux/moq-net/rml_rtmp/thiserror/ -# tokio/tracing; the rest (axum/clap/moq-native/rustls/tokio-rustls/tower-http/ -# url/sd-notify) are `optional` and pulled in by the `server` feature. RTMPS -# (Config::tls / Server::with_tls) needs rustls + tokio-rustls, so it's -# server-gated too. +# Pure library: the RTMP <-> MoQ gateway. The CLI lives in `moq-cli` (the `rtmp` +# subcommand); a relay can also embed `moq_rtmp::run` / `Server` against its own +# origin. Only rustls + tokio-rustls (RTMPS, the `tls` feature) are optional. [dependencies] anyhow = { version = "1", features = ["backtrace"] } -axum = { version = "0.8", features = ["tokio"], optional = true } -axum-server = { version = "0.8", features = ["tls-rustls"], optional = true } bytes = "1" -clap = { version = "4", features = ["derive"], optional = true } futures = "0.3" moq-mux = { workspace = true } -moq-native = { workspace = true, default-features = false, features = ["aws-lc-rs"], optional = true } moq-net = { workspace = true } # Pure-Rust RTMP (no librtmp / ffmpeg): handshake + chunk codec + the # server-session state machine. RTMP carries FLV, demuxed by moq-mux. @@ -64,9 +45,9 @@ tokio = { workspace = true, features = ["full"] } # workspace shares one tokio-rustls; the crypto provider comes from the supplied # `rustls::ServerConfig`, so no default crypto feature is needed here. tokio-rustls = { version = "0.26", default-features = false, optional = true } -tower-http = { version = "0.7", features = ["cors", "fs"], optional = true } tracing = "0.1" -url = { version = "2", optional = true } -[target.'cfg(unix)'.dependencies] -sd-notify = { version = "0.5", optional = true } +[dev-dependencies] +# The RTMPS test borrows moq-native's self-signed cert generation to build a +# `rustls::ServerConfig`; `quinn` pulls in a backend so `tls::Server` is built. +moq-native = { workspace = true, default-features = false, features = ["aws-lc-rs", "quinn"] } diff --git a/rs/moq-rtmp/README.md b/rs/moq-rtmp/README.md index aafb0fee7b..f6e244872a 100644 --- a/rs/moq-rtmp/README.md +++ b/rs/moq-rtmp/README.md @@ -15,8 +15,8 @@ or ffmpeg dependency. ## Library -Depend on it with `default-features = false` to skip the binary's relay -client/server and CLI dependencies: +RTMPS (`Config::tls` / `Server::with_tls`) is the only optional piece; drop it +with `default-features = false` for a plaintext-only build: ```toml moq-rtmp = { version = "0.0.1", default-features = false } @@ -111,41 +111,12 @@ Two ways to serve `rtmps://`: } ``` -## Binary +## CLI -The `moq-rtmp` binary (needs the default `server` feature) has two modes. +A command-line interface is provided by the [`moq-cli`](../moq-cli) binary, on +top of this library. -`serve` runs RTMP directly as a local relay, so MoQ subscribers (native or -browser) connect straight to this binary; RTMP players can also pull the same -broadcasts back out. It also exposes the `/certificate.sha256` endpoint browsers -need for self-signed `http://` origins, and can serve a static player directory -with `--dir`: - -```bash -moq-rtmp serve --server-bind [::]:443 --tls-generate localhost \ - --rtmp-listen 0.0.0.0:1935 --rtmp-prefix live/ -``` - -`publish` instead forwards every ingested broadcast out to a remote relay over -WebTransport (like `moq-srt publish` / `moq-cli hls import`): - -```bash -moq-rtmp publish --relay https://relay.example.com \ - --rtmp-listen 0.0.0.0:1935 --rtmp-prefix live/ -``` - -Either mode also accepts RTMPS alongside plaintext RTMP. Add `--rtmps-listen` -with a cert (`--rtmps-tls-cert` / `--rtmps-tls-key`, or `--rtmps-tls-generate` -for a throwaway self-signed cert), and OBS/ffmpeg can publish to `rtmps://`: - -```bash -moq-rtmp serve --server-bind [::]:443 --tls-generate localhost \ - --rtmp-listen 0.0.0.0:1935 \ - --rtmps-listen 0.0.0.0:1936 --rtmps-tls-cert cert.pem --rtmps-tls-key key.pem \ - --rtmp-prefix live/ -``` - -Feed either mode with any RTMP source. OBS: set the server to +Point any RTMP source at the gateway. OBS: set the server to `rtmp://127.0.0.1:1935/live` and the stream key to `cam0`. ffmpeg: ```bash @@ -175,8 +146,8 @@ the broadcast to be announced. ## Auth -The `run` entry point and the `moq-rtmp` binary are unauthenticated: anyone who -can reach the TCP port can publish or play, so gate them with a host firewall or a +The `run` entry point (and the `moq-cli` CLI built on it) is unauthenticated: +anyone who can reach the TCP port can publish or play, so gate them with a host firewall or a private network. To authenticate, use the `Server` / `Request` API above and verify each request in your accept loop (e.g. the stream key as a moq-token JWT, the app as the broadcast path) before accepting it. That is the intended diff --git a/rs/moq-rtmp/bin/moq-rtmp.rs b/rs/moq-rtmp/bin/moq-rtmp.rs deleted file mode 100644 index a5838ca632..0000000000 --- a/rs/moq-rtmp/bin/moq-rtmp.rs +++ /dev/null @@ -1,223 +0,0 @@ -//! `moq-rtmp` binary. -//! -//! Bridges RTMP / enhanced-RTMP (OBS, ffmpeg, hardware encoders, and players like -//! VLC / ffplay / mpv) to MoQ. Each listener does both directions: a `publish` -//! ingests a stream in, a `play` pulls one back out (`rtmp://host//` -//! round-trips to the same path). Two binary modes wire up the MoQ side: -//! -//! - `serve` runs a local QUIC/WebTransport server so MoQ subscribers connect -//! straight to this binary (no separate relay needed). RTMP players can also -//! pull the same broadcasts back out. -//! - `publish` forwards every ingested broadcast out to a remote relay over -//! WebTransport, like `moq-srt publish` / `moq-cli hls import` / `moq-rtc` WHIP. -//! -//! A relay that wants in-process ingest/egress should instead depend on the -//! `moq-rtmp` library and call `moq_rtmp::run` against its own origin. - -mod serve; -mod web; - -use std::net::SocketAddr; -use std::path::PathBuf; - -use anyhow::Context; -use clap::{Args, Parser, Subcommand}; -use url::Url; - -#[derive(Parser, Clone)] -#[command(name = "moq-rtmp", version)] -struct Cli { - #[command(flatten)] - log: moq_native::Log, - - #[command(subcommand)] - command: Command, -} - -#[derive(Subcommand, Clone)] -enum Command { - /// Ingest RTMP and serve it directly as a local relay. - Serve { - /// The QUIC/WebTransport server configuration. - #[command(flatten)] - config: moq_native::ServerConfig, - - /// Optionally serve static files (e.g. a web player) from this directory. - #[arg(long)] - dir: Option, - - #[command(flatten)] - rtmp: RtmpArgs, - }, - /// Ingest RTMP and publish the broadcasts to a remote MoQ relay. - Publish { - /// The MoQ client configuration. - #[command(flatten)] - config: moq_native::ClientConfig, - - /// URL of the MoQ relay to publish into (e.g. `https://relay.example.com`). - /// - /// `https://` makes a WebTransport connection over QUIC. `http://` first - /// fetches `/certificate.sha256` for the TLS fingerprint (insecure). The - /// `?jwt=` query parameter supplies a moq-token-cli JWT; otherwise the - /// public path (if any) is used. - #[arg(long, env = "MOQ_RTMP_RELAY")] - relay: Url, - - #[command(flatten)] - rtmp: RtmpArgs, - }, -} - -/// CLI flags for the RTMP listener(s), converted into [`moq_rtmp::Config`]s. -#[derive(Args, Clone)] -struct RtmpArgs { - /// Address to listen on for plaintext RTMP ingest (`rtmp://`). RTMP's - /// well-known port is 1935. - #[arg(long = "rtmp-listen", env = "MOQ_RTMP_LISTEN", default_value = "[::]:1935")] - listen: SocketAddr, - - /// Prefix prepended to every ingested broadcast path (e.g. `live/`). - #[arg(long = "rtmp-prefix", env = "MOQ_RTMP_PREFIX", default_value = "")] - prefix: String, - - /// Also listen for RTMPS (RTMP over TLS, `rtmps://`) on this address, in - /// addition to plaintext RTMP. Requires a certificate via `--rtmps-tls-cert` - /// /`--rtmps-tls-key` or a self-signed `--rtmps-tls-generate`. RTMPS has no - /// well-known port; common choices are 443 or a custom one. - /// - /// The bundled RTMPS terminator reuses moq-native's certificate loader, which - /// is only built with the `noq` or `quinn` backend; these flags are absent in - /// other builds. - #[cfg(any(feature = "noq", feature = "quinn"))] - #[arg(long = "rtmps-listen", env = "MOQ_RTMPS_LISTEN")] - rtmps_listen: Option, - - /// PEM certificate chain for RTMPS. - #[cfg(any(feature = "noq", feature = "quinn"))] - #[arg(long = "rtmps-tls-cert", env = "MOQ_RTMPS_TLS_CERT")] - rtmps_cert: Option, - - /// PEM private key for RTMPS. - #[cfg(any(feature = "noq", feature = "quinn"))] - #[arg(long = "rtmps-tls-key", env = "MOQ_RTMPS_TLS_KEY")] - rtmps_key: Option, - - /// Or generate a self-signed RTMPS certificate for these hostnames (testing - /// only; clients must disable verification or pin the fingerprint). - #[cfg(any(feature = "noq", feature = "quinn"))] - #[arg(long = "rtmps-tls-generate", env = "MOQ_RTMPS_TLS_GENERATE", value_delimiter = ',')] - rtmps_generate: Vec, -} - -impl RtmpArgs { - /// Build the listener configs: always plaintext RTMP, plus RTMPS when - /// `--rtmps-listen` is set. Both share the `--rtmp-prefix`. - fn configs(&self) -> anyhow::Result> { - let mut plain = moq_rtmp::Config::default(); - plain.listen = Some(self.listen); - plain.prefix = self.prefix.clone(); - #[cfg_attr(not(any(feature = "noq", feature = "quinn")), allow(unused_mut))] - let mut configs = vec![plain]; - - #[cfg(any(feature = "noq", feature = "quinn"))] - if let Some(addr) = self.rtmps_listen { - // Reuse moq-native's cert loader (on-disk pair or self-signed). RTMPS - // has no ALPN convention, so advertise none. - let mut tls = moq_native::tls::Server::default(); - tls.cert = self.rtmps_cert.clone().into_iter().collect(); - tls.key = self.rtmps_key.clone().into_iter().collect(); - tls.generate = self.rtmps_generate.clone(); - let server_config = tls.server_config(vec![]).context("build RTMPS TLS config")?; - - let mut rtmps = moq_rtmp::Config::default(); - rtmps.listen = Some(addr); - rtmps.prefix = self.prefix.clone(); - rtmps.tls = Some(server_config); - configs.push(rtmps); - } - - Ok(configs) - } -} - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - // moq-native pulls in `ring` somewhere transitively, so install the - // aws-lc-rs provider explicitly (mirrors moq-cli's main). - rustls::crypto::aws_lc_rs::default_provider() - .install_default() - .expect("failed to install default crypto provider"); - - let cli = Cli::parse(); - cli.log.init()?; - - match cli.command { - Command::Serve { config, dir, rtmp } => run_serve(config, dir, rtmp.configs()?).await, - Command::Publish { config, relay, rtmp } => run_publish(config, relay, rtmp.configs()?).await, - } -} - -/// Run every configured RTMP/RTMPS listener against `origin`, failing if any of -/// them does. Stays pending while at least one listener is alive. -async fn run_ingest(origin: moq_net::OriginProducer, configs: Vec) -> anyhow::Result<()> { - use futures::StreamExt; - - let mut listeners: futures::stream::FuturesUnordered<_> = configs - .into_iter() - .map(|config| moq_rtmp::run(origin.clone(), config)) - .collect(); - - while let Some(res) = listeners.next().await { - res.context("rtmp ingest failed")?; - } - - Ok(()) -} - -/// Run a local QUIC/WebTransport server and ingest RTMP directly into it. -async fn run_serve( - config: moq_native::ServerConfig, - dir: Option, - rtmp: Vec, -) -> anyhow::Result<()> { - let web_bind = config.bind.clone().unwrap_or_else(|| "[::]:443".to_string()); - - let server = config.init().context("init moq server")?; - let web_tls = server.tls_info(); - - // RTMP publishes broadcasts into this origin; the server serves them out. - let origin = moq_net::Origin::random().produce(); - - tracing::info!(%web_bind, "moq-rtmp serving"); - - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - tokio::select! { - res = serve::run(server, origin.clone()) => res.context("moq server failed"), - res = web::run(&web_bind, web_tls, dir) => res.context("web server failed"), - res = run_ingest(origin, rtmp) => res, - _ = tokio::signal::ctrl_c() => Ok(()), - } -} - -/// Ingest RTMP and forward every broadcast to a remote relay at `relay`. -async fn run_publish(config: moq_native::ClientConfig, relay: Url, rtmp: Vec) -> anyhow::Result<()> { - let client = config.init().context("init moq client")?; - - // RTMP publishes broadcasts into this origin; the client forwards them on. - let origin = moq_net::Origin::random().produce(); - let reconnect = client.with_publish(origin.consume()).reconnect(relay.clone()); - - tracing::info!(%relay, "moq-rtmp publishing"); - - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - tokio::select! { - res = run_ingest(origin, rtmp) => res, - res = reconnect.closed() => res.context("relay connection failed"), - _ = tokio::signal::ctrl_c() => Ok(()), - } -} diff --git a/rs/moq-rtmp/bin/serve.rs b/rs/moq-rtmp/bin/serve.rs deleted file mode 100644 index 2b135791d3..0000000000 --- a/rs/moq-rtmp/bin/serve.rs +++ /dev/null @@ -1,38 +0,0 @@ -//! Local QUIC/WebTransport server for `serve` mode. -//! -//! Accepts MoQ sessions and serves every broadcast the RTMP listener has -//! published into `origin`, so subscribers connect straight to this binary -//! instead of a separate relay. - -use moq_net::{OriginConsumer, OriginProducer}; - -/// Accept sessions and publish `origin`'s broadcasts to each subscriber. -pub async fn run(mut server: moq_native::Server, origin: OriginProducer) -> anyhow::Result<()> { - tracing::info!(addr = ?server.local_addr(), "listening"); - - let mut conn_id = 0; - while let Some(request) = server.accept().await { - let id = conn_id; - conn_id += 1; - - let consumer = origin.consume(); - tokio::spawn(async move { - if let Err(err) = serve_session(id, request, consumer).await { - tracing::warn!(%err, "session ended"); - } - }); - } - - anyhow::bail!("server stopped accepting connections") -} - -#[tracing::instrument("session", skip_all, fields(id))] -async fn serve_session(id: u64, request: moq_native::Request, consumer: OriginConsumer) -> anyhow::Result<()> { - // Blindly accept the session (WebTransport or QUIC), serving every ingested - // broadcast to the subscriber. - let session = request.with_publish(consumer).ok().await?; - - tracing::info!(id, "accepted session"); - - session.closed().await.map_err(Into::into) -} diff --git a/rs/moq-rtmp/bin/web.rs b/rs/moq-rtmp/bin/web.rs deleted file mode 100644 index 9492c91131..0000000000 --- a/rs/moq-rtmp/bin/web.rs +++ /dev/null @@ -1,64 +0,0 @@ -//! HTTP sidecar for `serve` mode. -//! -//! Serves `/certificate.sha256` (the TLS fingerprint browsers fetch when -//! connecting to a self-signed `http://` origin) and, optionally, a static -//! directory for local development. Mirrors `moq-cli`'s web server. - -use std::path::PathBuf; -use std::sync::{Arc, RwLock}; - -use anyhow::Context; -use axum::handler::HandlerWithoutStateExt; -use axum::http::StatusCode; -use axum::response::IntoResponse; -use axum::{Router, http::Method, routing::get}; -use tower_http::cors::{Any, CorsLayer}; -use tower_http::services::ServeDir; - -/// Serve the cert-fingerprint endpoint (and optional static `public` dir) on `bind`. -pub async fn run( - bind: &str, - tls_info: Arc>, - public: Option, -) -> anyhow::Result<()> { - let listen = tokio::net::lookup_host(bind) - .await - .context("invalid listen address")? - .next() - .context("invalid listen address")?; - - async fn handle_404() -> impl IntoResponse { - (StatusCode::NOT_FOUND, "Not found") - } - - let fingerprint_handler = move || async move { - tls_info - .read() - .expect("tls_info read lock poisoned") - .fingerprints - .first() - .expect("missing certificate") - .clone() - }; - - let mut app = Router::new() - .route("/certificate.sha256", get(fingerprint_handler)) - .layer(CorsLayer::new().allow_origin(Any).allow_methods([Method::GET])); - - if let Some(public) = public.as_ref() { - tracing::info!(public = %public.display(), "serving directory"); - - let public = ServeDir::new(public).not_found_service(handle_404.into_service()); - app = app.fallback_service(public); - } else { - app = app.fallback_service(handle_404.into_service()); - } - - // Dual-stack so the cert endpoint answers over IPv4 too, even on Windows - // where `[::]` is IPv6-only by default. - let listener = moq_native::bind::tcp(listen).context("failed to bind web listener")?; - let server = axum_server::from_tcp(listener)?; - server.serve(app.into_make_service()).await?; - - Ok(()) -} diff --git a/rs/moq-rtmp/src/lib.rs b/rs/moq-rtmp/src/lib.rs index 37c3144525..e28e31e2ca 100644 --- a/rs/moq-rtmp/src/lib.rs +++ b/rs/moq-rtmp/src/lib.rs @@ -34,8 +34,8 @@ //! a JWT and scoping the origin per token) plugs its policy in, with no //! callback. It mirrors `moq-native`'s `Server` / `Request`. //! -//! The bundled `moq-rtmp` binary serves the origin locally or forwards it to a -//! remote relay (those paths need the `server` feature). +//! A command-line interface is provided by the `moq-cli` binary, on top of this +//! library. //! //! RTMPS (RTMP over TLS) is supported two ways: //! @@ -62,6 +62,6 @@ pub use server::{Conn, Play, Publish, Request, Server, Stream, accept_stream}; /// Re-export of the `rustls` version this crate builds [`Config::tls`] against, /// so consumers construct a matching [`rustls::ServerConfig`] (a major `rustls` -/// bump is a breaking change). Only available with the `server` feature. -#[cfg(feature = "server")] +/// bump is a breaking change). Only available with the `tls` feature. +#[cfg(feature = "tls")] pub use rustls; diff --git a/rs/moq-rtmp/src/listen.rs b/rs/moq-rtmp/src/listen.rs index 0a70db92bb..31baa1489c 100644 --- a/rs/moq-rtmp/src/listen.rs +++ b/rs/moq-rtmp/src/listen.rs @@ -47,12 +47,12 @@ pub struct Config { /// TLS configuration for RTMPS (RTMP over TLS). When set, the /// [`listen`](Self::listen) address speaks RTMPS instead of plaintext RTMP, /// so clients connect with `rtmps://`. Build it with - /// [`moq_native::tls::Server::server_config`] (pass an empty ALPN list) or + /// `moq_native::tls::Server::server_config` (pass an empty ALPN list) or /// any [`rustls::ServerConfig`]. Leave `None` for plaintext. /// /// To serve both RTMP and RTMPS, run two listeners: call [`run`] once per /// config (one with `tls`, one without) against a cloned origin. - #[cfg(feature = "server")] + #[cfg(feature = "tls")] pub tls: Option>, } @@ -77,15 +77,15 @@ pub async fn run(origin: OriginProducer, config: Config) -> Result<()> { unreachable!("pending future never resolves"); }; - #[cfg_attr(not(feature = "server"), allow(unused_mut))] + #[cfg_attr(not(feature = "tls"), allow(unused_mut))] let mut server = Server::bind(listen).await?; - #[cfg(feature = "server")] + #[cfg(feature = "tls")] let tls = config.tls.is_some(); - #[cfg(not(feature = "server"))] + #[cfg(not(feature = "tls"))] let tls = false; - #[cfg(feature = "server")] + #[cfg(feature = "tls")] if let Some(tls) = config.tls.clone() { server = server.with_tls(tls); } diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 13dfdfab63..10bace8dfc 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -87,7 +87,7 @@ pub enum Conn { /// A TLS connection (`rtmps://`), established by [`Server::with_tls`]. Boxed /// because a `TlsStream` is large relative to a bare `TcpStream`. - #[cfg(feature = "server")] + #[cfg(feature = "tls")] Tls(Box>), } @@ -95,7 +95,7 @@ impl AsyncRead for Conn { fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { match self.get_mut() { Conn::Plain(s) => Pin::new(s).poll_read(cx, buf), - #[cfg(feature = "server")] + #[cfg(feature = "tls")] Conn::Tls(s) => Pin::new(s).poll_read(cx, buf), } } @@ -105,7 +105,7 @@ impl AsyncWrite for Conn { fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { match self.get_mut() { Conn::Plain(s) => Pin::new(s).poll_write(cx, buf), - #[cfg(feature = "server")] + #[cfg(feature = "tls")] Conn::Tls(s) => Pin::new(s).poll_write(cx, buf), } } @@ -113,7 +113,7 @@ impl AsyncWrite for Conn { fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match self.get_mut() { Conn::Plain(s) => Pin::new(s).poll_flush(cx), - #[cfg(feature = "server")] + #[cfg(feature = "tls")] Conn::Tls(s) => Pin::new(s).poll_flush(cx), } } @@ -121,7 +121,7 @@ impl AsyncWrite for Conn { fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match self.get_mut() { Conn::Plain(s) => Pin::new(s).poll_shutdown(cx), - #[cfg(feature = "server")] + #[cfg(feature = "tls")] Conn::Tls(s) => Pin::new(s).poll_shutdown(cx), } } @@ -138,7 +138,7 @@ pub struct Server { /// When set, each accepted connection is TLS-terminated (RTMPS) before the /// RTMP handshake. - #[cfg(feature = "server")] + #[cfg(feature = "tls")] tls: Option, /// In-flight handshakes; each resolves to a ready [`Request`], or `None` if @@ -152,7 +152,7 @@ impl Server { let listener = TcpListener::bind(addr).await?; Ok(Self { listener, - #[cfg(feature = "server")] + #[cfg(feature = "tls")] tls: None, pending: FuturesUnordered::new(), }) @@ -160,9 +160,9 @@ impl Server { /// Terminate TLS on every accepted connection, turning this into an RTMPS /// listener (`rtmps://`). Pass a `rustls::ServerConfig` (e.g. from - /// [`moq_native::tls::Server::server_config`] with an empty ALPN list), or + /// `moq_native::tls::Server::server_config` with an empty ALPN list), or /// `None` to leave it plaintext. - #[cfg(feature = "server")] + #[cfg(feature = "tls")] pub fn with_tls(mut self, tls: impl Into>>) -> Self { self.tls = tls.into().map(tokio_rustls::TlsAcceptor::from); self @@ -192,13 +192,13 @@ impl Server { res = self.listener.accept() => match res { Ok((stream, peer)) => { configure_socket(&stream, peer); - #[cfg(feature = "server")] + #[cfg(feature = "tls")] let tls = self.tls.clone(); self.pending.push(Box::pin(async move { // The TLS handshake (if any) and the RTMP handshake share one // budget, so a client that stalls either is dropped. let outcome = tokio::time::timeout(REQUEST_TIMEOUT, async move { - #[cfg(feature = "server")] + #[cfg(feature = "tls")] let conn = match tls { Some(acceptor) => Conn::Tls(Box::new( acceptor @@ -208,7 +208,7 @@ impl Server { )), None => Conn::Plain(stream), }; - #[cfg(not(feature = "server"))] + #[cfg(not(feature = "tls"))] let conn = Conn::Plain(stream); accept_until_request(conn, peer).await }) @@ -1172,10 +1172,9 @@ mod tests { } /// The same publish flow, but over TLS: prove [`Server::with_tls`] terminates - /// RTMPS and yields an identical [`Request`]. Gated on `quinn` because it - /// borrows moq-native's cert generation (`server_config`), which needs a - /// moq-native backend feature. - #[cfg(feature = "quinn")] + /// RTMPS and yields an identical [`Request`]. Gated on `tls` (RTMPS support); + /// the cert is generated by the `moq-native` dev-dependency. + #[cfg(feature = "tls")] #[tokio::test] async fn rtmps_accept_yields_publish_request() { use std::sync::Arc; diff --git a/rs/moq-srt/Cargo.toml b/rs/moq-srt/Cargo.toml index f55d6367db..42ae12a421 100644 --- a/rs/moq-srt/Cargo.toml +++ b/rs/moq-srt/Cargo.toml @@ -15,48 +15,17 @@ categories = ["multimedia", "network-programming", "web-programming"] [lib] doctest = false -[[bin]] -name = "moq-srt" -path = "bin/moq-srt.rs" -doc = false -# The binary (relay client/server, CLI) needs the `server` feature; the library -# can be depended on with `default-features = false` for ingest only (e.g. a -# relay that embeds the SRT listener and publishes into its own origin). -required-features = ["server"] - -[features] -default = ["server", "noq", "websocket"] -# Relay client/server transports + the moq-srt binary. Pulls in moq-native / -# axum / clap / rustls. -server = ["dep:axum", "dep:axum-server", "dep:clap", "dep:humantime", "dep:moq-native", "dep:rustls", "dep:sd-notify", "dep:tower-http", "dep:url"] -noq = ["server", "moq-native/noq"] -quinn = ["server", "moq-native/quinn"] -quiche = ["server", "moq-native/quiche"] -websocket = ["server", "moq-native/websocket"] -iroh = ["server", "moq-native/iroh"] - -# The ingest library needs only anyhow/bytes/futures/moq-mux/moq-net/srt-tokio/ -# thiserror/tokio/tracing; the rest (axum/clap/moq-native/rustls/tower-http/url/ -# sd-notify) are `optional` and pulled in by the `server` feature for the binary. +# Pure library: the SRT <-> MoQ gateway. The CLI lives in `moq-cli` (the `srt` +# subcommand); a relay can also embed `moq_srt::run` / `Server` against its own +# origin. [dependencies] anyhow = { version = "1", features = ["backtrace"] } -axum = { version = "0.8", features = ["tokio"], optional = true } -axum-server = { version = "0.8", features = ["tls-rustls"], optional = true } bytes = "1" -clap = { version = "4", features = ["derive"], optional = true } futures = "0.3" -humantime = { version = "2.3", optional = true } moq-mux = { workspace = true } -moq-native = { workspace = true, default-features = false, features = ["aws-lc-rs"], optional = true } moq-net = { workspace = true } -rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs"], optional = true } # Pure-Rust SRT (no libsrt / ffmpeg). SRT carries MPEG-TS, demuxed by moq-mux. srt-tokio = "0.4" thiserror = "2" tokio = { workspace = true, features = ["full"] } -tower-http = { version = "0.7", features = ["cors", "fs"], optional = true } tracing = "0.1" -url = { version = "2", optional = true } - -[target.'cfg(unix)'.dependencies] -sd-notify = { version = "0.5", optional = true } diff --git a/rs/moq-srt/README.md b/rs/moq-srt/README.md index 5bd12c3e1a..97f067633a 100644 --- a/rs/moq-srt/README.md +++ b/rs/moq-srt/README.md @@ -20,12 +20,7 @@ Pure Rust: SRT is provided by `srt-tokio`, with no libsrt or ffmpeg dependency. Two entry points. `Config` + `run` is the unauthenticated convenience: a relay embeds ingest by calling `run` against its own origin, so the ingested media is published locally with no extra hop. For auth, drive `Server` / `Request` -directly (see [Auth](#auth) below). Depend on it with `default-features = false` -to skip the binary's relay client/server and CLI dependencies: - -```toml -moq-srt = { version = "0.0.1", default-features = false } -``` +directly (see [Auth](#auth) below). ```rust let mut srt = moq_srt::Config::default(); @@ -39,29 +34,12 @@ tokio::select! { } ``` -## Binary - -The `moq-srt` binary (needs the default `server` feature) has two modes. - -`serve` ingests SRT and serves it directly as a local relay, so MoQ subscribers -(native or browser) connect straight to this binary. It also exposes the -`/certificate.sha256` endpoint browsers need for self-signed `http://` origins, -and can serve a static player directory with `--dir`: - -```bash -moq-srt serve --server-bind [::]:443 --tls-generate localhost \ - --srt-listen 0.0.0.0:9000 --srt-prefix live/ -``` - -`publish` instead forwards every ingested broadcast out to a remote relay over -WebTransport (like `moq-cli hls import`): +## CLI -```bash -moq-srt publish --relay https://relay.example.com \ - --srt-listen 0.0.0.0:9000 --srt-prefix live/ -``` +A command-line interface is provided by the [`moq-cli`](../moq-cli) binary, on +top of this library. -Feed either mode with any SRT source: +Feed any SRT source: ```bash # Publish: lands at broadcast `live/cam0`. diff --git a/rs/moq-srt/bin/moq-srt.rs b/rs/moq-srt/bin/moq-srt.rs deleted file mode 100644 index 82c525042a..0000000000 --- a/rs/moq-srt/bin/moq-srt.rs +++ /dev/null @@ -1,157 +0,0 @@ -//! `moq-srt` binary. -//! -//! Runs an SRT gateway (MPEG-TS in via `m=publish`, out via `m=request`) and -//! connects it to MoQ two ways: -//! -//! - `serve` runs a local QUIC/WebTransport server so subscribers connect -//! straight to this binary (no separate relay needed). Ingested broadcasts are -//! also requestable back out over SRT. -//! - `publish` forwards every ingested broadcast out to a remote relay over -//! WebTransport, like `moq-cli hls import` / `moq-rtc` WHIP. SRT requests are -//! served from the local origin (broadcasts ingested by this same process). -//! -//! A relay that wants an in-process gateway should instead depend on the -//! `moq-srt` library and call `moq_srt::run` against its own origin. - -mod serve; -mod web; - -use std::net::SocketAddr; -use std::path::PathBuf; -use std::time::Duration; - -use anyhow::Context; -use clap::{Args, Parser, Subcommand}; -use url::Url; - -#[derive(Parser, Clone)] -#[command(name = "moq-srt", version)] -struct Cli { - #[command(flatten)] - log: moq_native::Log, - - #[command(subcommand)] - command: Command, -} - -#[derive(Subcommand, Clone)] -enum Command { - /// Ingest SRT and serve it directly as a local relay. - Serve { - /// The QUIC/WebTransport server configuration. - #[command(flatten)] - config: moq_native::ServerConfig, - - /// Optionally serve static files (e.g. a web player) from this directory. - #[arg(long)] - dir: Option, - - #[command(flatten)] - srt: SrtArgs, - }, - /// Ingest SRT and publish the broadcasts to a remote MoQ relay. - Publish { - /// The MoQ client configuration. - #[command(flatten)] - config: moq_native::ClientConfig, - - /// URL of the MoQ relay to publish into (e.g. `https://relay.example.com`). - /// - /// `https://` makes a WebTransport connection over QUIC. `http://` first - /// fetches `/certificate.sha256` for the TLS fingerprint (insecure). The - /// `?jwt=` query parameter supplies a moq-token-cli JWT; otherwise the - /// public path (if any) is used. - #[arg(long, env = "MOQ_SRT_RELAY")] - relay: Url, - - #[command(flatten)] - srt: SrtArgs, - }, -} - -/// CLI flags for the SRT listener, converted into a [`moq_srt::Config`]. -#[derive(Args, Clone)] -struct SrtArgs { - /// Address to listen on for SRT ingest (e.g. `0.0.0.0:9000`). - #[arg(long = "srt-listen", env = "MOQ_SRT_LISTEN")] - listen: SocketAddr, - - /// Prefix prepended to every ingested broadcast path (e.g. `live/`). - #[arg(long = "srt-prefix", env = "MOQ_SRT_PREFIX", default_value = "")] - prefix: String, - - /// SRT receive latency: the negotiated buffer that trades delay for loss recovery. - #[arg(long = "srt-latency", env = "MOQ_SRT_LATENCY", default_value = "200ms", value_parser = humantime::parse_duration)] - latency: Duration, -} - -impl From for moq_srt::Config { - fn from(args: SrtArgs) -> Self { - let mut config = moq_srt::Config::default(); - config.listen = Some(args.listen); - config.prefix = args.prefix; - config.latency = args.latency; - config - } -} - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - // moq-native pulls in `ring` somewhere transitively, so install the - // aws-lc-rs provider explicitly (mirrors moq-cli's main). - rustls::crypto::aws_lc_rs::default_provider() - .install_default() - .expect("failed to install default crypto provider"); - - let cli = Cli::parse(); - cli.log.init()?; - - match cli.command { - Command::Serve { config, dir, srt } => run_serve(config, dir, srt.into()).await, - Command::Publish { config, relay, srt } => run_publish(config, relay, srt.into()).await, - } -} - -/// Run a local QUIC/WebTransport server and ingest SRT directly into it. -async fn run_serve(config: moq_native::ServerConfig, dir: Option, srt: moq_srt::Config) -> anyhow::Result<()> { - let server = config.init().context("init moq server")?; - // Derive the HTTP sidecar bind from the actual listener, so a port-0 or - // hostname-resolved server still serves /certificate.sha256 on the real socket. - let web_bind = server.local_addr().context("server local addr")?.to_string(); - let web_tls = server.tls_info(); - - // SRT publishes broadcasts into this origin; the server serves them out. - let origin = moq_net::Origin::random().produce(); - - tracing::info!(%web_bind, "moq-srt serving"); - - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - tokio::select! { - res = serve::run(server, origin.clone()) => res.context("moq server failed"), - res = web::run(&web_bind, web_tls, dir) => res.context("web server failed"), - res = moq_srt::run(origin, srt) => res.context("srt ingest failed"), - _ = tokio::signal::ctrl_c() => Ok(()), - } -} - -/// Ingest SRT and forward every broadcast to a remote relay at `relay`. -async fn run_publish(config: moq_native::ClientConfig, relay: Url, srt: moq_srt::Config) -> anyhow::Result<()> { - let client = config.init().context("init moq client")?; - - // SRT publishes broadcasts into this origin; the client forwards them on. - let origin = moq_net::Origin::random().produce(); - let reconnect = client.with_publish(origin.consume()).reconnect(relay.clone()); - - tracing::info!(%relay, "moq-srt publishing"); - - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - tokio::select! { - res = moq_srt::run(origin, srt) => res.context("srt ingest failed"), - res = reconnect.closed() => res.context("relay connection failed"), - _ = tokio::signal::ctrl_c() => Ok(()), - } -} diff --git a/rs/moq-srt/bin/serve.rs b/rs/moq-srt/bin/serve.rs deleted file mode 100644 index d49e476b81..0000000000 --- a/rs/moq-srt/bin/serve.rs +++ /dev/null @@ -1,38 +0,0 @@ -//! Local QUIC/WebTransport server for `serve` mode. -//! -//! Accepts MoQ sessions and serves every broadcast the SRT listener has -//! published into `origin`, so subscribers connect straight to this binary -//! instead of a separate relay. - -use moq_net::{OriginConsumer, OriginProducer}; - -/// Accept sessions and publish `origin`'s broadcasts to each subscriber. -pub async fn run(mut server: moq_native::Server, origin: OriginProducer) -> anyhow::Result<()> { - tracing::info!(addr = ?server.local_addr(), "listening"); - - let mut conn_id = 0; - while let Some(request) = server.accept().await { - let id = conn_id; - conn_id += 1; - - let consumer = origin.consume(); - tokio::spawn(async move { - if let Err(err) = serve_session(id, request, consumer).await { - tracing::warn!(%err, "session ended"); - } - }); - } - - anyhow::bail!("server stopped accepting connections") -} - -#[tracing::instrument("session", skip_all, fields(id))] -async fn serve_session(id: u64, request: moq_native::Request, consumer: OriginConsumer) -> anyhow::Result<()> { - // Blindly accept the session (WebTransport or QUIC), serving every ingested - // broadcast to the subscriber. - let session = request.with_publish(consumer).ok().await?; - - tracing::info!(id, "accepted session"); - - session.closed().await.map_err(Into::into) -} diff --git a/rs/moq-srt/bin/web.rs b/rs/moq-srt/bin/web.rs deleted file mode 100644 index 9492c91131..0000000000 --- a/rs/moq-srt/bin/web.rs +++ /dev/null @@ -1,64 +0,0 @@ -//! HTTP sidecar for `serve` mode. -//! -//! Serves `/certificate.sha256` (the TLS fingerprint browsers fetch when -//! connecting to a self-signed `http://` origin) and, optionally, a static -//! directory for local development. Mirrors `moq-cli`'s web server. - -use std::path::PathBuf; -use std::sync::{Arc, RwLock}; - -use anyhow::Context; -use axum::handler::HandlerWithoutStateExt; -use axum::http::StatusCode; -use axum::response::IntoResponse; -use axum::{Router, http::Method, routing::get}; -use tower_http::cors::{Any, CorsLayer}; -use tower_http::services::ServeDir; - -/// Serve the cert-fingerprint endpoint (and optional static `public` dir) on `bind`. -pub async fn run( - bind: &str, - tls_info: Arc>, - public: Option, -) -> anyhow::Result<()> { - let listen = tokio::net::lookup_host(bind) - .await - .context("invalid listen address")? - .next() - .context("invalid listen address")?; - - async fn handle_404() -> impl IntoResponse { - (StatusCode::NOT_FOUND, "Not found") - } - - let fingerprint_handler = move || async move { - tls_info - .read() - .expect("tls_info read lock poisoned") - .fingerprints - .first() - .expect("missing certificate") - .clone() - }; - - let mut app = Router::new() - .route("/certificate.sha256", get(fingerprint_handler)) - .layer(CorsLayer::new().allow_origin(Any).allow_methods([Method::GET])); - - if let Some(public) = public.as_ref() { - tracing::info!(public = %public.display(), "serving directory"); - - let public = ServeDir::new(public).not_found_service(handle_404.into_service()); - app = app.fallback_service(public); - } else { - app = app.fallback_service(handle_404.into_service()); - } - - // Dual-stack so the cert endpoint answers over IPv4 too, even on Windows - // where `[::]` is IPv6-only by default. - let listener = moq_native::bind::tcp(listen).context("failed to bind web listener")?; - let server = axum_server::from_tcp(listener)?; - server.serve(app.into_make_service()).await?; - - Ok(()) -} diff --git a/rs/moq-srt/src/lib.rs b/rs/moq-srt/src/lib.rs index e70fc7d1b9..250ca60a8e 100644 --- a/rs/moq-srt/src/lib.rs +++ b/rs/moq-srt/src/lib.rs @@ -24,8 +24,8 @@ //! JWT and scoping the origin per token) plugs its policy in. It mirrors //! `moq-native`'s `Server` / `Request`. //! -//! The bundled `moq-srt` binary serves the origin locally or forwards it to a -//! remote relay (those paths need the `server` feature). +//! A command-line interface is provided by the `moq-cli` binary, on top of this +//! library. //! //! Pure Rust: SRT is provided by `srt-tokio`, with no libsrt or ffmpeg //! dependency.