Compare commits

...
25 Commits
Author SHA1 Message Date
doloro a9bd890d16 cargo clippy fix 2026-09-17 18:57:35 +01:00
doloro 21a20b898c added custom url label and is protected fields to catalog listing 2026-09-17 18:57:00 +01:00
doloro 6ac7e60a34 Bump : 0.6.0 : Data channel pipeline, currently used for sending viewers data 2026-09-10 13:59:09 +01:00
doloro 94daa39d87 nix flake update 2026-09-10 13:58:09 +01:00
doloro c4a7e622b7 fmt all 2026-08-29 15:15:28 +01:00
doloro 98dbb3d36d Cleanup: over-engineering pass
- merge /api/uptime + /api/version into /api/health and /api/stats
- collapse RTMP per-codec dispatch into a single CodecParser path
- dedupe STUN attribute walking (ufrag_pair / parse_xor_mapped_address)
- dedupe StreamCodec <-> str0m Codec mapping via StreamCodec::from_str0m
- dedupe bearer token extraction in WHIP handlers
- replace regex charset validation with stdlib checks
- flatten one-field wrappers (WebRtcProxyConfig, HttpServerConfig, ServerInfo)
- remove SessionCookie extractor (only fed debug logs)
- delete dead code: entity ActiveModel helpers, get_stream_session,
  WebrtcProxy::local_addr, _connected flag, commented-out blocks
- drop unused deps: futures, rand, regex, serde_json
2026-08-21 11:39:02 +01:00
doloro 0a039a97c4 Bump : 0.5.4 : WHIP cleanup guarded by session id, stale detach task can't kill a newer stream on the same key 2026-08-20 19:08:12 +01:00
doloro f855e0e471 Bump : 0.5.3 : Fix login cookie being a chud FUCK and not persisting + support for new login system 2026-08-17 15:32:43 +01:00
doloro 72cfb26e38 Bump: 0.5.2 : label charset tightened; password/custom-id validation 2026-08-16 15:25:06 +01:00
doloro 80368b3ea5 Bump: 0.5.1 : persistent session cookie 2026-08-16 15:06:55 +01:00
doloro 0c6705badd Bump: 0.5.0 : unlisted, password and locked stream support 2026-08-16 14:56:29 +01:00
doloro 9960434c1c chore: add audit TODO notes 2026-08-08 13:45:37 +01:00
doloro c58ca27490 feat(whip): close publishes idle for 30s 2026-08-08 13:42:08 +01:00
doloro 98febe4bf3 chore: demote per-packet debug logs to trace 2026-08-08 13:42:08 +01:00
doloro f0a40efc25 chore(docker): default RUST_LOG to info, fix aarch64 build layer copy 2026-08-08 13:42:08 +01:00
doloro 7d9b3b6285 fix(http): reject unsupported codecs with 415 and the codec as body 2026-08-08 13:42:08 +01:00
doloro aa1d4e8f67 fix: WHIP not cleaning up 2026-08-07 17:21:15 +01:00
doloro e00541c07d bump: 0.4.0 (WHIP support) 2026-08-07 16:08:24 +01:00
doloro 3287a0ed7c chore: clean up and optimisation 2026-08-07 16:06:01 +01:00
doloro 5f152e9db7 feat: WHIP support 2026-08-07 11:48:31 +01:00
doloro c1435a70fb slop 2026-07-31 20:20:48 +01:00
doloro d4b77bd717 feat: High quality AGENTS.md file 2026-07-19 16:59:00 +01:00
doloro 4bb103e9b2 fix(http): use manual cookie headers with SameSite attribute and dev flag for cross-origin support 2026-07-19 01:02:55 +01:00
doloro fd1e59267d chore: add axum macros feature and remove unused tracing import 2026-07-19 01:02:52 +01:00
doloro a2c74a6e81 fix(rtmp): replace unwrap() calls with proper error handling and add handshake timeout 2026-07-19 01:02:40 +01:00
28 changed files with 2773 additions and 945 deletions
+642
View File
@@ -0,0 +1,642 @@
# AGENTS.md
This file provides comprehensive guidance for AI agents working with this repository.
It is the canonical reference; CLAUDE.md may be a subset of this.
---
## Project Summary
**RTMP-to-WHIP/WHEP bridge.** Accepts RTMP video+audio publish streams and re-streams to
browsers via WebRTC (WHEP). Also has a stub for WHIP ingest.
**Binary name:** `rtmp-to-whip`
**Workspace:** Rust 2024 edition, 3 crates (`server`, `entity`, `migration`)
**No automated tests exist** (except one ignored FLV replay test).
---
## Commands
```bash
# Build (all crates)
cargo build
# Run (server binary)
cargo run
# Build (Nix)
nix build
# Dev shell (provides clang + mold linker + sea-orm-cli + opus + just)
nix develop
```
---
## Workspace Layout
```
crates/
server/ — main binary (RTMP, HTTP, WebRTC, codecs, audio)
entity/ — SeaORM entity definitions (users, stream_key, stream_session, auth_session)
migration/ — SeaORM migrations (4 files)
target/ — build output (gitignored)
db/ — SQLite database at runtime (db/db.sqlite, gitignored)
index.html — WHEP test page (browser-based viewer)
```
---
## Ports & Networking
| Port | Protocol | Purpose |
|-------|----------|----------------------------------|
| 1935 | TCP | RTMP ingest (OBS/encoder) |
| 3000 | TCP | HTTP API (axum) |
| 6969 | UDP | WebRTC media (configurable via RTC_PORT env) |
---
## Environment Variables
| Variable | Default | Purpose |
|-----------------|---------|------------------------------------------------|
| `RTC_PORT` | `6969` | UDP port for WebRTC media traffic |
| `PUBLIC_DOMAIN` | — | Domain for ICE candidates (DNS-resolved); falls back to STUN discovery if unset |
| `SIGNUP_CODE` | — | Required token for `/api/user` signup; empty = disabled |
| `RUST_LOG` | — | Tracing filter (e.g. `info,warn`) |
---
## Entry Point — `crates/server/src/main.rs`
Initializes all components, spawns 4 workers in a `JoinSet`:
```rust
workers.spawn(http.run());
workers.spawn(proxy.run());
workers.spawn(webrtc.run());
workers.spawn(rtmp.run());
```
Shuts down on Ctrl-C or any worker exit (`workers.abort_all()`).
**`AppState`** — `Arc<Mutex<AppState>>` wrapping `DashMap<i32, StreamSession>`.
Keyed by `stream_key.id`.
**`StreamSession`** fields:
- `stream_key_id: i32`
- `stream_key_label: String`
- `frame_channel: async_broadcast::Sender<Arc<VideoFrame>>`
- `audio_channel: async_broadcast::Sender<Arc<OpusAudioFrame>>`
- `codec: Option<StreamCodec>``H264 | H265 | AV1`
- `started_at: DateTime<Utc>`
- `viewers: AtomicU32`
**Shared channels:**
- `offer_tx/offer_rx: mpsc::channel<(i32, i32, String)>(64)` — HTTP → WebRTC (request_id, stream_key_id, sdp_offer)
- `answer_tx/answer_rx: broadcast::<(i32, Option<String>)>(64)` — WebRTC → HTTP (request_id, answer_or_none)
**DB:** `sqlite://./db/db.sqlite?mode=rwc`. Migrations run at startup via `Migrator::up`.
`stream_session::Model::clean_unended_streams()` repairs crash-residual sessions.
---
## Architecture — Signal Flow
```
OBS/encoder → RTMP (port 1935) → H264Parser / H265Parser / Av1Parser → async_broadcast frame channel
Browser ← WebRTC/UDP ← str0m Rtc ← WHEP HTTP POST /api/stream/{slug}
WebrtcProxy (UDP port from RTC_PORT env, default 6969)
WHIP ingest: Browser/encoder → WHIP POST /api/whip → (stub: parses SDP, logs media)
```
---
## RTMP Ingestion — `crates/server/src/rtmp.rs`
`pub struct Rtmp { listener, stream_sessions, db }`
`async fn run(self)` — accepts TCP connections on port 1935, spawns one task per connection:
1. **Handshake**`rml_rtmp::Handshake` (C0+C1 → S0+S1+S2 → C2), 30s timeout.
2. **Session setup**`rml_rtmp::ServerSession`.
3. **Event loop** — 4096-byte reads, 15s idle timeout:
- `ConnectionRequested` → accepted unconditionally.
- `PublishStreamRequested` → DB lookup of stream key; rejects if not found or already live.
Inserts `StreamSession` into `AppState` + DB row.
- `VideoDataReceived` → dispatched to `CodecParser::parse` (H.264, H.265, or AV1).
Frame broadcast on `frame_channel`.
- `AudioDataReceived``AACParser` decodes AAC → PCM, `AudioProcesser` resamples + encodes Opus.
`OpusAudioFrame` broadcast on `audio_channel`.
- `PublishStreamFinished` → removes `StreamSession` from `AppState`, sets `ended_at` in DB.
**Codec detection** (`parse_video_codec`):
- Enhanced RTMP (byte 0 bit 7 = 1): bytes 14 = FourCC (`hvc1`→H.265, `avc1`→H.264, `av01`→AV1).
- Legacy RTMP (byte 0 bit 7 = 0): nibble `& 0x0F == 7` → H.264.
- `StreamSession.codec` is stamped on first `VideoDataReceived` and never changed.
---
## HTTP API — `crates/server/src/http.rs`
`pub struct HttpServer` fields: `offer_tx`, `accept_rx`, `appstate`, `request_count: AtomicI32`, `db`, `config`.
Serves on `0.0.0.0:3000` with CORS for `http://localhost:5173` and `https://stream.h.doloro.co.uk`.
### Routes
| Method | Path | Handler | Auth? | Notes |
|--------|-------------------------|--------------------------------|-------|-------------------------------------------------|
| GET | `/api/catalog` | `catalog_handler` | No | Returns `{ active_streams: [...] }` from DB + AppState |
| POST | `/api/user` | `create_user_handler` | No | Signup; requires `SIGNUP_CODE` header |
| POST | `/api/login` | `login_handler` | No | Verifies password, creates auth_session, sets cookie |
| GET | `/api/stream-key` | `get_all_stream_keys` | Yes | Returns user's stream keys |
| POST | `/api/stream-key` | `create_stream_key_handler` | Yes | Creates key with UUID value (`stream-key-<uuid>`) |
| PATCH | `/api/stream-key` | `edit_stream_key` | Yes | Edits label (validated by `KEY_RE` regex) |
| POST | `/api/whip` | `handle_whip_injest` | No | **Stub** — parses SDP, logs media lines |
| POST | `/api/stream/{slug}` | `stream_handler` | No | WHEP offer; forwards to WebRTC, waits for answer |
| GET | `/api/meow` | `meow_handler` | No | Returns `"meow"` |
### Auth
- Session token sent as `session` cookie or header.
- `AuthUser` extractor: looks up `auth_session` by token value, then joins to `users`.
- Protected routes require `AuthUser`.
- Cookie flags: `HttpOnly`, `Path=/`, `SameSite=None` (dev) or `Lax` (prod).
- `DevFlag` extractor: reads `?dev=1` query param.
### Validation
- Username: max 32 chars, non-empty.
- Stream key label: 164 chars, regex `^[A-Za-z0-9 _-]{1,67}$`, must contain at least one letter.
- User has `stream_key_limit` (default 3) enforced at creation time.
---
## WebRTC Proxy — `crates/server/src/webrtc_proxy.rs`
`pub struct WebrtcProxy` — all fields `Arc`-wrapped, `Clone`-able.
- `clients_ufrag: DashMap<String, mpsc::Sender<(Bytes, SocketAddr)>>` — pending ICE ufrag → per-client channel
- `clients_addr: DashMap<SocketAddr, mpsc::Sender<(Bytes, SocketAddr)>>` — established addr → per-client channel
- `socket: Arc<UdpSocket>` — shared UDP socket bound to `0.0.0.0:{RTC_PORT}`
- `public_addr: SocketAddr` — from `PUBLIC_DOMAIN` DNS or STUN discovery
**`async fn run(self)`** — UDP receive loop:
1. Receives datagrams on shared socket.
2. If source addr in `clients_addr`, forwards to that client.
3. Otherwise parses STUN binding request to extract ufrag (username attribute, split on `:`),
looks up `clients_ufrag`, promotes to `clients_addr`, forwards.
**`fn add_client(ufrag)`** — called by Webrtc when setting up peer connection.
Returns `(Arc<UdpSocket>, Receiver<(Bytes, SocketAddr)>)`.
**Public IP discovery:**
- If `PUBLIC_DOMAIN` env set: DNS-resolve for IPv4.
- Else if `cfg!(debug_assertions)`: returns `127.0.0.1` (for testing).
- Else: sends STUN Binding Request to `stun.l.google.com:19302`, parses XOR-MAPPED-ADDRESS.
---
## WebRTC Negotiation — `crates/server/src/webrtc.rs`
`pub struct Webrtc { offer_rx, accept_tx, sessions_ref, db, proxy }`
`async fn run(mut self)` — receives `(request_id, stream_id, sdp_body)` from `offer_rx`:
1. Looks up `stream_key` in DB; rejects if not found.
2. Gets codec from `StreamSession.codec` (determines which str0m codec config to enable).
3. Configures str0m `Rtc::builder`:
- Opus always enabled.
- H.265: `cc.enable_h265(true)`.
- AV1: `cc.enable_av1(true)`.
- H.264 (default): adds 3 profiles — Constrained Baseline (0x42e01f), Main (0x4d001f), High (0x64001f).
Firefox only offers Constrained Baseline — without it, no codec matches.
4. Accepts SDP offer → produces SDP answer.
5. Validates answer has non-empty video PT list (str0m returns empty list if no codec matched).
6. Extracts ICE ufrag from answer, registers with proxy.
7. Broadcasts answer on `accept_tx`.
8. Spawns `detach_connection` per peer.
**`detach_connection`** per-peer loop:
- Drains `rtc.poll_output()` — transmits via proxy UDP, handles events.
- `Event::MediaAdded` — selects best video PT by profile_level_id (H.264),
h265_profile_tier_level (H.265), or level_idx (AV1).
- `Event::Connected` — starts subscribing to frame/audio channels from AppState.
- `tokio::select!` on: str0m deadline, incoming UDP from proxy, video frame, audio frame.
- **Keyframe gating:** drops non-keyframe frames until first keyframe is seen.
- **Overflow handling:** on ring buffer overflow, resets `saw_keyframe` to wait for next keyframe.
- Video RTP timestamp: `MediaTime::from_90khz(ts * 90)` where `ts` is VideoFrame timestamp in ms.
- Audio RTP timestamp: `MediaTime::new(ts * 48, Frequency::FORTY_EIGHT_KHZ)`.
- Drains up to 7 pending frames per tick to catch up.
---
## Codecs — `crates/server/src/codec/`
### Trait — `CodecParser`
```rust
pub trait CodecParser: Send {
fn parse(&mut self, data: &[u8], timestamp_ms: u32) -> Option<VideoFrame>;
}
```
### `VideoFrame`
```rust
pub struct VideoFrame {
pub data: Bytes, // Annex-B elementary stream
pub is_keyframe: bool,
pub timestamp_ms: u32, // Presentation timestamp (DTS + CTS for B-frame streams)
}
```
### H.264 — `h264.rs`
Handles both legacy and enhanced RTMP.
**Legacy RTMP:** byte 0 = `(frame_type << 4) | codec_id`, byte 1 = packet type, bytes 24 = CTS.
**Enhanced RTMP:** byte 0 = `0x80 | (frame_type << 4) | packet_type`, bytes 14 = FourCC.
- Packet type `0`: parses `AVCDecoderConfigurationRecord`, caches SPS+PPS.
- Packet type `1`: AVCC → Annex-B conversion, prepends SPS+PPS before keyframes.
Uses `PTS = DTS + CTS` to avoid B-frame stuttering.
- Packet type `3`: same as type 1 but no CTS field.
### H.265 — `h265.rs`
Enhanced RTMP only (FourCC `hvc1`).
- Packet type `0`: bytes[5..] = `HEVCDecoderConfigurationRecord`.
- Packet type `1`: bytes 57 = CTS, bytes 8+ = HVCC NALUs. Uses `PTS = DTS + CTS`.
- Packet type `3`: bytes 5+ = HVCC NALUs, no CTS.
`HEVCDecoderConfigurationRecord` parsing: skips first 22 bytes (profile/level/tier),
parses arrays at byte 22, caches VPS (NAL type 32), SPS (33), PPS (34).
### AV1 — `av1.rs`
Enhanced RTMP only (FourCC `av01`).
- Packet type `0`: bytes[5..] = `AV1CodecConfigurationRecord`, extracts `configOBUs` from `payload[4..]`.
- Packet type `1`/`3`: bytes[5..] = OBU stream.
**Keyframe detection:**
1. RTMP FrameType=1.
2. Scan OBUs for `OBU_SEQUENCE_HEADER` (obu_type=1).
3. First coded frame after config (bootstrap fallback — OBS may never set FrameType=1 for AV1).
Config OBUs prepended to keyframe payloads so str0m's Av1Packetizer sees a Sequence Header OBU.
### FLV Replay Test — `flv_replay_test.rs`
Ignored test harness for codec validation:
```bash
FLV_IN=test.flv ES_OUT=output.es cargo test -p server flv_replay -- --ignored --nocapture
```
Parses FLV video tags, runs through codec parser, dumps Annex-B elementary stream with
temporal delimiter OBUs for ffprobe validation. Checks for duplicate PTS values.
---
## Audio — `crates/server/src/audio.rs`
### AACParser
Parses RTMP audio payloads:
- Byte 0: `>> 4 == 10` → AAC codec.
- Byte 1: `0` = AudioSpecificConfig → initializes Symphonia AAC decoder with `extra_data`.
- Byte 1: `1` = raw AAC frame → decodes via Symphonia → interleaved f32 PCM.
### AudioFrame
```rust
pub struct AudioFrame {
pub data: Bytes, // interleaved f32 PCM, little-endian
pub timestamp_ms: u32,
pub sample_rate: u32, // typically 44100 Hz from AAC-LC
}
```
### AudioProcesser
AAC → Opus transcoding pipeline:
- **Resampler:** `rubato::Fft` from 44100 Hz → 48000 Hz (if needed).
- **Encoder:** `opus::Encoder` — 48kHz stereo, `LowDelay` mode.
- **Frame size:** 960 samples per channel (20ms), 1920 interleaved.
- **Timestamp:** monotonic 48kHz counter: `timestamp_ms = samples_emitted / 48` (independent of RTMP timestamps).
- **Buffer:** accumulates resampled PCM until full Opus frame, then encodes and emits `OpusAudioFrame`.
### OpusAudioFrame
```rust
pub struct OpusAudioFrame {
pub data: Bytes,
pub timestamp_ms: u32,
}
```
---
## Password Hashing — `crates/server/src/hash.rs`
- `hash_password(password)` — Argon2id with random `OsRng` salt, returns PHC-format string.
- `verify_password(password, hash)` — parses PHC string, verifies with Argon2.
**Panics** if `hash` is not valid PHC format (uses `.unwrap()` on `PasswordHash::new`).
---
## Error Handling — `crates/server/src/http_error.rs`
`#[derive(Error, Debug)] pub enum HttpError { ... }`
Maps to HTTP status codes:
- `DbErr`/`Hash`/`Internal` → 500
- `NotFound` → 404
- `Unauthorized` → 401
- `Forbidden` → 403
- `Conflict` → 409
- `BadRequest` → 400
- `Unprocessable` → 422
- `NotAcceptable` → 406
4xx errors return message body; 5xx returns empty body (no internal details leaked).
---
## WebRTC Ingest (WHIP) — `crates/server/src/webrtc_ingest.rs`
**STUB — not implemented.** Currently:
- Accepts `POST /api/whip` with SDP body.
- Parses SDP via `str0m::change::SdpOffer`.
- Logs media lines.
---
## Database (SeaORM + SQLite)
**DB file:** `./db/db.sqlite`
**Migrations:** `crates/migration/` (4 migrations)
### Tables
#### `users`
| Column | Type | Constraints |
|---------------------|----------|--------------------------|
| `id` | INTEGER | PK, autoincrement |
| `username` | TEXT | NOT NULL, UNIQUE |
| `hashed_password` | TEXT | NOT NULL |
| `stream_key_limit` | INTEGER | NOT NULL, default 3 |
Relations: `has_many``stream_key`, `auth_session`
Custom methods:
- `Entity::create(db, username, password_hash)` — inserts user
- `Entity::find_by_username(db, username)` — lookup
- `Entity::find_by_auth_session(db, token)` — join auth_session → user
- `ActiveModel::update_username`, `update_password`, `change_stream_key_limit`
#### `stream_key`
| Column | Type | Constraints |
|--------------|----------|--------------------------|
| `id` | INTEGER | PK, autoincrement |
| `key_value` | TEXT | NOT NULL, UNIQUE |
| `user_id` | INTEGER | NOT NULL, FK → users |
| `label` | TEXT | NOT NULL |
| `is_active` | BOOLEAN | NOT NULL, default true |
| `is_unlisted`| BOOLEAN | NOT NULL, default true |
| `created_at` | DATETIME | NOT NULL |
Relations: `belongs_to``users`, `has_many``stream_session`
Custom methods:
- `Entity::create(db, user_id, key_value, label, is_unlisted)` — generates `stream-key-<uuid>` as key_value
- `Entity::find_by_key(db, key_value)` — lookup by raw key string
- `Entity::find_by_user(db, user_id)` — list user's keys
- `ActiveModel::change_label_value`
#### `stream_session`
| Column | Type | Constraints |
|----------------|----------|--------------------------|
| `id` | INTEGER | PK, autoincrement |
| `stream_key_id`| INTEGER | NOT NULL, FK → stream_key|
| `started_at` | DATETIME | NOT NULL |
| `ended_at` | DATETIME | nullable |
Relations: `belongs_to``stream_key`
Custom methods:
- `Model::create_stream_session(db, stream_key_id, started_at)` — creates open session
- `Model::get_stream_session(db, id)` — get by ID
- `Model::get_all_active_sessions(db)` — all sessions where `ended_at IS NULL`
- `Model::get_active_by_stream_key_id(db, key_id)` — single active session
- `Model::clean_unended_streams(db)` — sets `ended_at = started_at` for crash recovery
- `ActiveModel::finish_stream_session(db, ended_at)` — marks session ended
#### `auth_session`
| Column | Type | Constraints |
|-----------|----------|--------------------------|
| `id` | INTEGER | PK, autoincrement |
| `id_user` | INTEGER | NOT NULL, FK → users |
| `value` | TEXT | UUID v4 string |
Relations: `belongs_to``users`
Custom methods:
- `Entity::create(db, user_id)` — generates UUID token, creates session
- `Entity::find_by_user_id(db, user_id)` — lookup by user
### Conventions
- Query methods on `Entity` (e.g. `Entity::find_by_x`).
- Mutation helpers on `ActiveModel`.
- For destructive schema changes: expand-contract pattern (add → backfill → switch code → drop old).
---
## Dependencies
### Server (`crates/server/Cargo.toml`)
| Crate | Version | Purpose |
|---------------|---------|-----------------------------|
| `tokio` | 1 | Async runtime (full) |
| `str0m` | 0.20.0 | WebRTC |
| `rml_rtmp` | 0.8.0 | RTMP server |
| `axum` | 0.8 | HTTP framework |
| `axum-extra` | 0.12.6 | Cookie extraction |
| `sea-orm` | 1 | ORM (SQLite, tokio-rustls) |
| `symphonia` | 0.5 | Audio decode (AAC) |
| `opus` | 0.3.1 | Audio encode (Opus) |
| `rubato` | 3.0.0 | Audio resampling |
| `argon2` | 0.5.3 | Password hashing |
| `uuid` | 1.23.3 | UUID v4 |
| `dashmap` | 6.2.1 | Concurrent map |
| `async-broadcast` | 0.7.2 | Ring buffer channels |
| `tower-http` | 0.6 | CORS |
| `regex` | 1 | Stream key label validation |
| `thiserror` | 2.0.18 | Error types |
| `tracing` | 0.1 | Structured logging |
| `chrono` | 0.4.45 | Date/time |
| `bytes` | 1 | Byte buffers |
| `futures` | 0.3.32 | async block_on |
| `serde` | 1.0.228 | Serialization |
### Entity (`crates/entity/Cargo.toml`)
| Crate | Version | Purpose |
|---------------|---------|----------------|
| `sea-orm` | 1 | ORM macros |
| `serde` | 1 | Serialization |
| `chrono` | 0.4 | DateTime |
| `rand` | 0.10.1 | Random salts |
| `argon2` | 0.5.3 | Password hash |
| `uuid` | 1.23.3 | UUID v4 |
### Migration (`crates/migration/Cargo.toml`)
| Crate | Version | Purpose |
|--------------------|---------|----------------|
| `sea-orm-migration`| 1 | Migration CLI |
| `tokio` | 1 | Async runtime |
---
## Build & Release
### Profiles (root `Cargo.toml`)
- **dev:** debug = true
- **release:** lto = true, codegen-units = 1, panic = "abort"
- **flamegraph:** inherits release + debug + force-frame-pointers
### Cross-compilation
- x86_64: clang linker + mold (via rustflags).
- aarch64: cross-toolchain (`aarch64-linux-gnu-gcc`), no mold.
### Docker
- x86_64: multi-stage, strips binary, copies only needed .so deps.
- aarch64: QEMU cross-build from x86_64 builder, copies arm64 .so deps.
- Both use `scratch` final stage.
- Exposes: 1935/tcp, 3000/tcp, 6969/udp.
### Nix
- Uses `crane` for cargo builds.
- Dev shell: clang, mold, sea-orm-cli, cmake, opus, pkgconf, just.
- ADMIN_REF_CODE set in dev shell: `meowmeowpurrrmeow`.
---
## WHEP Test Page — `index.html`
- Connects to `http://localhost:5000/whep/test` (note: hardcoded to port 5000, not 3000).
- Creates RTCPeerConnection with video recvonly transceiver.
- Sends WHEP offer POST, receives SDP answer.
- Streams video to `<video>` element.
- Polls `pc.getStats()` every 500ms, displays:
- Server-to-client latency (network + jitter + decode)
- Network one-way RTT (from ICE candidate pair)
- Jitter buffer delay
- Decode latency
- FPS
- Packet loss ratio
- Jitter
---
## Key Implementation Details & Gotchas
1. **H.264 PT negotiation:** str0m uses `profile_level_id` to match H.264 profiles. Firefox only offers Constrained Baseline (0x42e01f). All three profiles must be configured or Firefox sees no codec.
2. **PTS vs DTS:** For both H.264 and H.265, presentation timestamp = DTS + CTS. B-frame streams stutter if DTS is used.
3. **AV1 keyframe detection:** OBS may never set RTMP FrameType=1 for AV1. The parser uses triple fallback: RTMP flag, OBU scan, first-frame bootstrap.
4. **H.265 support marked "very shity" (sic):** See `StreamCodec` enum doc comment in `main.rs:56`.
5. **No auth on WHEP endpoint:** `/api/stream/{slug}` is publicly accessible — any client who knows the stream key slug can get a WHEP answer.
6. **Request ID tracking:** `AtomicI32::fetch_add(1, Relaxed)` for WHEP request IDs. Monotonic, wraps at i32::MAX.
7. **Channel sizes:** broadcast channels (frame/audio) = 32, offer channel = 64, answer channel = 64, proxy client channel = 256.
8. **RTMP 4096-byte read buffer:** This is a fixed-size buffer. Large RTMP messages are split across multiple reads — `rml_rtmp` handles reassembly internally.
9. **Audio timestamp independence:** Opus RTP timestamps use a monotonic 48kHz counter, not RTMP timestamps. This ensures stable playback even if source timestamps are irregular.
10. **Stream session cleanup:** On any disconnect (error, timeout, finish), the cleanup closure removes from `AppState` and sets `ended_at` in DB. The `clean_unended_streams` at startup repairs crash-residual sessions.
---
## README.md Todo List
Completed:
- [X] axum cookie jar support
- [X] H.265 codec support
- [X] AV1 codec support
- [X] Custom error types (`thiserror` + `IntoResponse`)
Incomplete:
- [ ] Stream status via RTC data channel (frame drops, codec errors)
- [ ] WHEP → WHIP support
- [ ] Replace `async_broadcast` with circular buffer for frame → RTC management
QoL:
- [ ] Stream key renaming (partially implemented via `edit_stream_key`)
- [ ] Admin panel (username lookup, disable stream keys, change limits, CPU usage)
---
## File Index
### Server source (`crates/server/src/`)
- `main.rs` — entry point, AppState, StreamSession, worker spawning
- `http.rs` — HTTP server, all route handlers, auth extractors
- `http_error.rs` — HttpError enum + IntoResponse impl
- `rtmp.rs` — RTMP listener, handshake, codec dispatch, frame broadcasting
- `webrtc.rs` — WebRTC negotiation, str0m Rtc, RTP sending, keyframe gating
- `webrtc_proxy.rs` — UDP proxy, STUN parsing, public IP discovery
- `webrtc_ingest.rs` — WHIP stub handler
- `audio.rs` — AACParser, AudioProcesser (resample + Opus encode)
- `hash.rs` — Argon2id password hashing
- `codec/mod.rs` — CodecParser trait, VideoFrame struct
- `codec/h264.rs` — H.264 AVCC/legacy → Annex-B
- `codec/h265.rs` — H.265 HVCC → Annex-B (enhanced RTMP only)
- `codec/av1.rs` — AV1 OBU parsing (enhanced RTMP only)
- `codec/flv_replay_test.rs` — ignored test for codec validation
### Entity source (`crates/entity/src/`)
- `lib.rs` — module exports
- `prelude.rs` — entity type aliases
- `users.rs` — User entity + methods
- `stream_key.rs` — StreamKey entity + methods
- `stream_session.rs` — StreamSession entity + methods
- `auth_session.rs` — AuthSession entity + methods
### Migration source (`crates/migration/src/`)
- `lib.rs` — Migrator trait, migration registry
- `main.rs` — migration binary entry
- `m20260616_000001_create_users.rs` — users table
- `m20260616_000002_create_stream_key.rs` — stream_key table
- `m20260616_000003_create_stream_session.rs` — stream_session table
- `m20260616_000004_create_auth_session.rs` — auth_session table
### Config
- `Cargo.toml` (root) — workspace, profiles
- `crates/server/Cargo.toml` — server dependencies
- `crates/entity/Cargo.toml` — entity dependencies
- `crates/migration/Cargo.toml` — migration dependencies
- `Dockerfile` — x86_64 multi-stage build
- `Dockerfile.aarch64` — arm64 cross-build
- `compose.yaml` — docker-compose deployment
- `flake.nix` — Nix build + dev shell
- `Justfile` — Docker build/push shortcuts
- `.gitignore` — target, db, devenv files
Generated
+340 -377
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -43,6 +43,6 @@ EXPOSE 1935
EXPOSE 3000
EXPOSE 6969/udp
ENV RUST_LOG="info,warn"
ENV RUST_LOG="info"
CMD ["/rtmp-to-whip"]
+2
View File
@@ -55,4 +55,6 @@ EXPOSE 1935
EXPOSE 3000
EXPOSE 6969/udp
ENV RUST_LOG="info"
CMD ["/rtmp-to-whip"]
+36
View File
@@ -0,0 +1,36 @@
# TODO
Audit findings from ponytail-audit (ranked biggest cut first). Net: ~-300 lines, -5 deps possible.
## Deletions
- [ ] Delete `Dockerfile.aarch64` — merge into `Dockerfile` with `ARG TARGETARCH` (buildx sets it); both files are 90% identical except the dep-copy block. [Dockerfile.aarch64]
- [ ] Delete entity dead methods: `update_username`, `update_password`, `change_stream_key_limit`, `find_by_user_id`, `get_stream_session`, `get_all_active_sessions`. Zero callers; scaffolding for the unbuilt admin panel. [crates/entity/src/users.rs:77, crates/entity/src/auth_session.rs:43, crates/entity/src/stream_session.rs:53]
- [ ] Delete `/api/health`, `/api/uptime`, `/api/version` + `HealthResponse`/`UptimeResponse` — all subsets of `/api/stats`; also kills `serde_json` (only used by `version_handler`). [crates/server/src/http.rs:497]
- [ ] Delete `SessionCookie` extractor — its value is only debug-logged, never used; kill the extractor + 2 logs. [crates/server/src/http.rs:213]
- [ ] Delete `WebrtcProxy::local_addr()` (no callers), dead `let _ = x_port ^ …` stmt, and single-field `WebRtcProxyConfig` struct → pass `u16` (also fixes i32 port type). [crates/server/src/webrtc_proxy.rs:167, crates/server/src/webrtc_proxy.rs:139]
- [ ] Delete `webrtc.rs` `mid` binding from `add_media` + `_hint_mid` param — passed straight into an underscore. [crates/server/src/webrtc.rs:142]
- [ ] Delete `stream_key` `is_active`/`is_unlisted` columns — never read anywhere; `create()` hardcodes `is_active=false` and `is_unlisted` is always passed `false`. Drop param + column (migration, expand-contract). [crates/entity/src/stream_key.rs:45]
- [ ] Delete `rtmp.rs` redundant `stream_id` var (warn can use `current_stream_key_id`), `warn!("")` empty-log on parse error, and `parse_video_codec`'s `Result<_, Box<dyn Error>>``Option`. [crates/server/src/rtmp.rs:206, crates/server/src/rtmp.rs:369]
- [ ] Delete `catalog_handler`'s `get_all_active_sessions` query — result only feeds a `debug!`; kills the entity method too. [crates/server/src/http.rs:185]
- [ ] Delete commented-out routes/code (`admin/server_stats`, duplicate whip route, `.max_age`, `fs::File`) + `#[axum::debug_handler]`. [crates/server/src/http.rs:109]
- [ ] Delete `StreamSession.active_clients` — written 0, never read; drops `AtomicU32` import. [crates/server/src/main.rs:65]
- [ ] Delete `webrtc_ingest` `_connected` flag — set true, never read. [crates/server/src/webrtc_ingest.rs:363]
- [ ] Delete `users.rs` `let user = …; user` pointless binding in `find_by_username`. [crates/entity/src/users.rs:70]
- [ ] Delete `meow_handler` + route — joke endpoint, zero consumers. [crates/server/src/http.rs:492]
- [ ] Delete `index.html` `prevStats` — assigned, never read. [index.html:57]
- [ ] Delete deps — server: `futures`, `rand`; entity: `rand`, `argon2` (hashing lives in server crate now); `serde_json` (with health/uptime/version cut). [crates/server/Cargo.toml, crates/entity/Cargo.toml]
## Shrinks
- [ ] Extract shared "remove session + finish_stream_session" fn — closure duplicated 3× (rtmp.rs cleanup, webrtc_ingest detach cleanup, whip DELETE handler). Also lets `handle_whip_injest` drop its manual `err()` closure for `Result<HttpError>` like its siblings. [crates/server/src/rtmp.rs:141, crates/server/src/webrtc_ingest.rs:338, crates/server/src/webrtc_ingest.rs:34]
- [ ] Extract `fixup_answer_sdp()``whip_sdp_probe.rs` re-encodes the SDP string-fixups verbatim; probe drops ~30 lines. [crates/server/tests/whip_sdp_probe.rs:66]
- [ ] Derive `thiserror` on `AudioParseError` instead of hand-rolled `Display`+`Error` impls (~15 lines). [crates/server/src/audio.rs:66]
- [ ] Collapse `webrtc.rs` four near-identical channel re-subscribe blocks (`is_none`/`is_closed` × video/audio) → one helper. [crates/server/src/webrtc.rs:336]
## Out of scope (correctness — route to normal review)
- `Local::now()` vs `Utc::now()` inconsistency in session creation
- `KEY_RE` `{1,67}` vs `MAX_LABEL_LEN = 64` mismatch
- `index.html` hardcoded port 5000
- `/api/stream/{slug}` unauthenticated
+9 -8
View File
@@ -12,6 +12,8 @@ pub struct Model {
pub label: String,
pub is_active: bool,
pub is_unlisted: bool,
pub password: Option<String>,
pub custom_id: Option<String>,
pub created_at: DateTimeUtc,
}
@@ -78,15 +80,14 @@ impl Entity {
.all(db)
.await
}
}
impl ActiveModel {
pub async fn change_label_value(
mut self,
pub async fn find_by_custom_id(
db: &DatabaseConnection,
value: String,
) -> Result<Model, DbErr> {
self.label = Set(value);
self.update(db).await
custom_id: String,
) -> Result<Option<Model>, DbErr> {
Entity::find()
.filter(Column::CustomId.eq(custom_id))
.one(db)
.await
}
}
+4 -15
View File
@@ -50,32 +50,21 @@ impl Model {
.insert(db)
.await
}
pub async fn get_stream_session(
db: &DatabaseConnection,
stream_session_id: i32,
) -> Result<Model, DbErr> {
Entity::find_by_id(stream_session_id)
.one(db)
.await?
.ok_or(DbErr::RecordNotFound(format!(
"stream_session {stream_session_id}"
)))
}
pub async fn get_all_active_sessions(db: &DatabaseConnection) -> Result<Vec<Model>, DbErr> {
Ok(Entity::find()
Entity::find()
.filter(Column::EndedAt.is_null())
.all(db)
.await?)
.await
}
pub async fn get_active_by_stream_key_id(
db: &DatabaseConnection,
stream_key_id: i32,
) -> Result<Option<Model>, DbErr> {
Ok(Entity::find()
Entity::find()
.filter(Column::StreamKeyId.eq(stream_key_id))
.filter(Column::EndedAt.is_null())
.one(db)
.await?)
.await
}
pub async fn clean_unended_streams(db: &DatabaseConnection) -> Result<u64, DbErr> {
+4 -4
View File
@@ -58,18 +58,18 @@ impl Entity {
if let Some(x) = sessions {
Entity::find_by_id(x.id_user).one(db).await
} else {
return Ok(None);
Ok(None)
}
}
pub async fn find_by_username(
db: &DatabaseConnection,
username: String,
) -> Result<Option<Model>, DbErr> {
let user = Entity::find()
Entity::find()
.filter(Column::Username.eq(username))
.one(db)
.await;
user
.await
}
}
+4
View File
@@ -4,6 +4,8 @@ mod m20260616_000001_create_users;
mod m20260616_000002_create_stream_key;
mod m20260616_000003_create_stream_session;
mod m20260616_000004_create_auth_session;
mod m20260815_000005_add_password_to_stream_key;
mod m20260815_000006_set_stream_keys_unlisted;
pub struct Migrator;
@@ -15,6 +17,8 @@ impl MigratorTrait for Migrator {
Box::new(m20260616_000002_create_stream_key::Migration),
Box::new(m20260616_000003_create_stream_session::Migration),
Box::new(m20260616_000004_create_auth_session::Migration),
Box::new(m20260815_000005_add_password_to_stream_key::Migration),
Box::new(m20260815_000006_set_stream_keys_unlisted::Migration),
]
}
}
@@ -38,7 +38,7 @@ impl MigrationTrait for Migration {
ColumnDef::new(StreamKey::IsUnlisted)
.boolean()
.not_null()
.default(true),
.default(false),
)
.col(ColumnDef::new(StreamKey::CreatedAt).date_time().not_null())
.foreign_key(
@@ -67,5 +67,7 @@ pub enum StreamKey {
Label,
IsActive,
IsUnlisted,
Password,
CustomId,
CreatedAt,
}
@@ -0,0 +1,47 @@
use sea_orm_migration::prelude::*;
use super::m20260616_000002_create_stream_key::StreamKey;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.alter_table(
Table::alter()
.table(StreamKey::Table)
.add_column(ColumnDef::new(StreamKey::Password).string().null())
.to_owned(),
)
.await?;
manager
.alter_table(
Table::alter()
.table(StreamKey::Table)
.add_column(ColumnDef::new(StreamKey::CustomId).string().null())
.to_owned(),
)
.await
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.alter_table(
Table::alter()
.table(StreamKey::Table)
.drop_column(StreamKey::Password)
.to_owned(),
)
.await?;
manager
.alter_table(
Table::alter()
.table(StreamKey::Table)
.drop_column(StreamKey::CustomId)
.to_owned(),
)
.await
}
}
@@ -0,0 +1,26 @@
use sea_orm_migration::prelude::*;
use super::m20260616_000002_create_stream_key::StreamKey;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// By accident the is_unlisted column defaulted to true; existing stream
// keys were meant to be listed. Reset all rows to false here.
manager
.exec_stmt(
Query::update()
.table(StreamKey::Table)
.values([(StreamKey::IsUnlisted, false.into())])
.to_owned(),
)
.await
}
async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
Ok(())
}
}
+5 -6
View File
@@ -1,6 +1,6 @@
[package]
name = "server"
version = "0.3.0"
version = "0.6.0"
edition = "2024"
[target.x86_64-unknown-linux-gnu]
@@ -15,26 +15,25 @@ path = "src/main.rs"
async-broadcast = "0.7.2"
bytes = "1"
dashmap = "6.2.1"
rand = "0.10.1"
rml_rtmp = "0.8.0"
str0m = "0.20.0"
tokio = { version = "1", features = ["full"] }
axum = "0.8"
axum = { version = "0.8", features = ["macros"] }
serde = { version = "1.0.228", features = ["serde_derive"] }
serde_json = "1.0.150"
sea-orm = { version = "1", features = [ "sqlx-sqlite", "runtime-tokio-rustls", "macros" ] }
entity = {path = "../entity"}
migration = {path = "../migration"}
argon2 = "0.5.3"
uuid = { version = "1.23.3", features = ["v4"] }
tower-http = { version = "0.6", features = ["cors"] }
futures = "0.3.32"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
symphonia = { version = "0.5", features = ["aac"] }
opus = "0.3.1"
rubato = "3.0.0"
chrono = "0.4.45"
regex = "1"
time = "0.3"
thiserror = "2.0.18"
sysinfo = "0.36"
axum-extra = { version = "0.12.6", features = ["cookie"] }
serde_json = "1.0.151"
+5 -27
View File
@@ -1,4 +1,4 @@
use std::{error::Error, fmt::Display};
use std::error::Error;
use bytes::Bytes;
use rubato::{Fft, Resampler, audioadapter_buffers::direct::InterleavedSlice};
@@ -44,7 +44,7 @@ impl AudioProcesser {
pub fn encode(&mut self, frame: AudioFrame) -> Vec<OpusAudioFrame> {
let mut samples: Vec<f32> = frame
.data
.chunks_exact(4)
.as_chunks::<4>().0.iter()
.map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
.collect();
@@ -83,7 +83,6 @@ impl AudioProcesser {
pub struct AudioFrame {
pub data: Bytes, // interleaved f32 PCM, little-endian
pub timestamp_ms: u32,
pub sample_rate: u32,
}
@@ -96,23 +95,6 @@ pub struct AACParser {
decoder: Option<Box<dyn Decoder>>,
}
#[derive(Debug)]
enum AudioParseError {
InvalidCodec,
NoConfigPacket,
}
impl Display for AudioParseError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::InvalidCodec => write!(f, "Not the right codec provided"),
Self::NoConfigPacket => write!(f, "No config packet cached"),
}
}
}
impl Error for AudioParseError {}
// Byte 0: upper nibble = sound format (10 = AAC)
// Byte 1: 0 = AudioSpecificConfig, 1 = raw AAC frame
impl AACParser {
@@ -123,14 +105,14 @@ impl AACParser {
pub fn parse(
&mut self,
bytes: &[u8],
timestamp_ms: u32,
_timestamp_ms: u32,
) -> Result<Option<AudioFrame>, Box<dyn Error>> {
if bytes.len() < 2 {
return Ok(None);
}
if (bytes[0] >> 4) != 10 {
return Err(Box::new(AudioParseError::InvalidCodec));
return Err("not the right codec provided".into());
}
if bytes[1] == 0 {
@@ -143,10 +125,7 @@ impl AACParser {
return Ok(None);
}
let decoder = self
.decoder
.as_mut()
.ok_or(AudioParseError::NoConfigPacket)?;
let decoder = self.decoder.as_mut().ok_or("no config packet cached")?;
let packet = Packet::new_from_boxed_slice(
0,
@@ -169,7 +148,6 @@ impl AACParser {
Ok(Some(AudioFrame {
data: Bytes::from(pcm_bytes),
timestamp_ms,
sample_rate: spec.rate,
}))
}
+27 -13
View File
@@ -50,7 +50,9 @@ impl Av1CodecParser {
let has_size = (header >> 1) & 1 != 0;
i += 1;
if has_extension {
if i >= data.len() { return false; }
if i >= data.len() {
return false;
}
i += 1;
}
if obu_type == 1 {
@@ -61,13 +63,19 @@ impl Av1CodecParser {
let mut size: usize = 0;
let mut shift = 0;
loop {
if i >= data.len() { return false; }
if i >= data.len() {
return false;
}
let b = data[i] as usize;
i += 1;
size |= (b & 0x7F) << shift;
shift += 7;
if b & 0x80 == 0 { break; }
if shift > 32 { return false; }
if b & 0x80 == 0 {
break;
}
if shift > 32 {
return false;
}
}
i += size;
} else {
@@ -78,7 +86,11 @@ impl Av1CodecParser {
false
}
fn obus_for_frame(&mut self, payload: &[u8], rtmp_is_keyframe: bool) -> Option<(Vec<u8>, bool)> {
fn obus_for_frame(
&mut self,
payload: &[u8],
rtmp_is_keyframe: bool,
) -> Option<(Vec<u8>, bool)> {
if payload.is_empty() {
return None;
}
@@ -95,13 +107,11 @@ impl Av1CodecParser {
self.first_coded_frame = false;
if is_keyframe {
if let Some(config) = &self.config_obus {
let mut out = Vec::with_capacity(config.len() + payload.len());
out.extend_from_slice(config);
out.extend_from_slice(payload);
return Some((out, true));
}
if is_keyframe && let Some(config) = &self.config_obus {
let mut out = Vec::with_capacity(config.len() + payload.len());
out.extend_from_slice(config);
out.extend_from_slice(payload);
return Some((out, true));
}
Some((payload.to_vec(), is_keyframe))
@@ -130,7 +140,11 @@ impl CodecParser for Av1CodecParser {
// FourCC — enhanced RTMP only defines CTS for hvc1 CodedFrames.
let payload = data.get(5..)?;
let (obus, is_keyframe) = self.obus_for_frame(payload, rtmp_is_keyframe)?;
Some(VideoFrame { data: Bytes::from(obus), is_keyframe, timestamp_ms })
Some(VideoFrame {
data: Bytes::from(obus),
is_keyframe,
timestamp_ms,
})
}
_ => None,
}
+20 -10
View File
@@ -53,12 +53,20 @@ impl H264CodecParser {
}
let pts_ms = Self::pts_ms(timestamp_ms, &bytes[5..8]);
let data = self.avcc_to_annexb(bytes.get(8..)?, is_keyframe)?;
Some(VideoFrame { data: Bytes::from(data), is_keyframe, timestamp_ms: pts_ms })
Some(VideoFrame {
data: Bytes::from(data),
is_keyframe,
timestamp_ms: pts_ms,
})
}
3 => {
// CodedFramesX: no CTS field, bytes 5+ = AVCC NALUs
let data = self.avcc_to_annexb(bytes.get(5..)?, is_keyframe)?;
Some(VideoFrame { data: Bytes::from(data), is_keyframe, timestamp_ms })
Some(VideoFrame {
data: Bytes::from(data),
is_keyframe,
timestamp_ms,
})
}
_ => None,
}
@@ -77,7 +85,11 @@ impl H264CodecParser {
// PTS = DTS + CTS — see note on CodedFrames above.
let pts_ms = Self::pts_ms(timestamp_ms, &bytes[2..5]);
let data = self.avcc_to_annexb(&bytes[5..], is_keyframe)?;
Some(VideoFrame { data: Bytes::from(data), is_keyframe, timestamp_ms: pts_ms })
Some(VideoFrame {
data: Bytes::from(data),
is_keyframe,
timestamp_ms: pts_ms,
})
}
_ => None,
}
@@ -148,13 +160,11 @@ impl H264CodecParser {
// Prepend SPS+PPS before every keyframe so str0m's packetizer
// can bundle them into a STAP-A alongside the IDR NALU.
if is_keyframe {
if let (Some(sps), Some(pps)) = (&self.sps, &self.pps) {
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(sps);
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(pps);
}
if is_keyframe && let (Some(sps), Some(pps)) = (&self.sps, &self.pps) {
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(sps);
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(pps);
}
// Convert each length-prefixed NALU to an Annex B start-code NALU.
+18 -11
View File
@@ -74,15 +74,14 @@ impl H265CodecParser {
fn hvcc_to_annexb(&self, payload: &[u8], is_keyframe: bool) -> Option<Vec<u8>> {
let mut out = Vec::with_capacity(payload.len());
if is_keyframe {
if let (Some(vps), Some(sps), Some(pps)) = (&self.vps, &self.sps, &self.pps) {
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(vps);
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(sps);
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(pps);
}
if is_keyframe && let (Some(vps), Some(sps), Some(pps)) = (&self.vps, &self.sps, &self.pps)
{
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(vps);
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(sps);
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(pps);
}
let mut i = 0;
@@ -129,12 +128,20 @@ impl CodecParser for H265CodecParser {
let cts = i32::from_be_bytes([0, data[5], data[6], data[7]]) << 8 >> 8;
let pts_ms = (timestamp_ms as i64 + cts as i64).max(0) as u32;
let annexb = self.hvcc_to_annexb(data.get(8..)?, is_keyframe)?;
Some(VideoFrame { data: Bytes::from(annexb), is_keyframe, timestamp_ms: pts_ms })
Some(VideoFrame {
data: Bytes::from(annexb),
is_keyframe,
timestamp_ms: pts_ms,
})
}
3 => {
// CodedFramesX: no CTS, bytes 5+ = HVCC
let annexb = self.hvcc_to_annexb(data.get(5..)?, is_keyframe)?;
Some(VideoFrame { data: Bytes::from(annexb), is_keyframe, timestamp_ms })
Some(VideoFrame {
data: Bytes::from(annexb),
is_keyframe,
timestamp_ms,
})
}
_ => None,
}
+308 -136
View File
@@ -1,46 +1,44 @@
use std::{
net::SocketAddr,
sync::{
Arc, LazyLock,
Arc,
atomic::{AtomicI32, Ordering},
},
time::{Duration, Instant},
};
use axum_extra::extract::{CookieJar, cookie::Cookie};
use regex::Regex;
// The Rust `regex` crate is guaranteed linear-time and therefore does NOT
// support lookaround, so the JS source pattern
// /^(?=.*[A-Za-z])[A-Za-z0-9_-]{1,67}$/
// cannot be ported verbatim. The `(?=.*[A-Za-z])` lookahead only means
// "must contain at least one letter" — we drop it from the pattern and
// enforce that condition with a separate `.chars().any(..)` check below.
static KEY_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"^[A-Za-z0-9 _-]{1,67}$").unwrap());
/// Allowed charset for labels, passwords and custom IDs: letters, numbers,
/// dashes and apostrophes — no spaces. Mirrors the frontend
/// `/^[A-Za-z0-9'-]{0,67}$/` used by the keys-page popups.
fn valid_charset(s: &str) -> bool {
s.len() <= 67
&& s.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '\'')
}
use axum::{
Json, Router,
body::{Body, Bytes},
extract::{Form, FromRequestParts, Path, Query, State},
extract::{FromRequestParts, Path, State},
http::{
HeaderMap, HeaderName, Method, StatusCode,
header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, SET_COOKIE},
header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE},
request::Parts,
},
response::{IntoResponse, Response},
routing::{get, post},
routing::{delete, get, post},
};
use chrono::{DateTime, Utc};
use entity::{auth_session, stream_key, stream_session, users};
use sea_orm::{
DatabaseConnection, EntityTrait, IntoActiveModel, QueryFilter, prelude::DateTimeUtc,
};
use sea_orm::{ActiveModelTrait, DatabaseConnection, EntityTrait, IntoActiveModel, Set};
use serde::{Deserialize, Serialize};
use sysinfo::System;
use tokio::{
net::TcpListener,
sync::{Mutex, mpsc::Sender},
};
use tower_http::cors::CorsLayer;
use uuid::Uuid;
use tracing::{debug, info, warn};
@@ -48,23 +46,21 @@ use crate::{
AppState,
hash::{hash_password, verify_password},
http_error::HttpError,
webrtc_ingest::handle_whip_injest,
webrtc_ingest::{handle_whip_injest, handle_whip_injest_delete, handle_whip_injest_patch},
};
const MAX_USERNAME_LEN: usize = 32;
const MAX_LABEL_LEN: usize = 64;
pub struct HttpServerConfig {
pub signup_code: String,
}
pub struct HttpServer {
pub offer_tx: Sender<(i32, i32, String)>,
pub accept_rx: async_broadcast::InactiveReceiver<(i32, Option<String>)>,
pub accept_rx: async_broadcast::InactiveReceiver<(i32, Result<String, String>)>,
pub appstate: Arc<Mutex<AppState>>,
pub request_count: AtomicI32,
pub db: DatabaseConnection,
pub config: Arc<HttpServerConfig>,
pub signup_code: String,
pub version: &'static str,
pub start_time: Instant,
}
impl HttpServer {
@@ -77,7 +73,13 @@ impl HttpServer {
];
let cors = CorsLayer::new()
.allow_methods([Method::GET, Method::POST, Method::OPTIONS])
.allow_methods([
Method::GET,
Method::POST,
Method::PATCH,
Method::DELETE,
Method::OPTIONS,
])
.allow_headers([
AUTHORIZATION,
ACCEPT,
@@ -96,37 +98,51 @@ impl HttpServer {
.get(get_all_stream_keys)
.patch(edit_stream_key),
)
// .route("/api/admin/server_stats", get(todo!()))
.route("/api/whip", post(handle_whip_injest))
.route("/api/login", post(login_handler))
.route("/api/stream/{slug}", post(stream_handler))
.route("/api/whip", post(handle_whip_injest))
.route(
"/api/whip/{slug}",
delete(handle_whip_injest_delete).patch(handle_whip_injest_patch),
)
.route("/api/meow", get(meow_handler))
.route("/api/health", get(health_handler))
.route("/api/stats", get(stats_handler))
.layer(cors)
.with_state(state);
let addr = SocketAddr::from(([0, 0, 0, 0], 3000));
let listener = TcpListener::bind(addr).await.unwrap();
axum::serve(listener, app).await.unwrap();
axum::serve(
listener,
app.into_make_service_with_connect_info::<SocketAddr>(),
)
.await
.unwrap();
}
}
#[derive(Serialize)]
struct StreamCatalog {
active_streams: Vec<StreamListing>,
}
#[derive(Serialize)]
struct StreamListing {
label: String,
custom_url_label: String,
id: i32,
user: String,
started_at: DateTime<Utc>, //UNIX TIMESTAMP
started_at: DateTime<Utc>,
is_password_protected: bool,
}
#[derive(Deserialize)]
struct EditStreamKeyRequest {
id: i32,
new: String,
#[serde(default)]
new: Option<String>,
#[serde(default)]
password: Option<String>,
#[serde(default)]
unlisted: Option<bool>,
#[serde(default)]
custom_id: Option<String>,
}
async fn edit_stream_key(
@@ -135,15 +151,25 @@ async fn edit_stream_key(
Json(payload): Json<EditStreamKeyRequest>,
) -> Result<impl IntoResponse, HttpError> {
// Trim surrounding whitespace so labels aren't stored with leading/trailing spaces.
let new_label = payload.new.trim();
// Length (1..=67) and allowed charset ([A-Za-z0-9 _-], space included).
if !KEY_RE.is_match(new_label) {
return Err(HttpError::BadRequest("invalid label".into()));
}
// Replaces the JS lookahead: label must contain at least one letter.
if !new_label.chars().any(|c| c.is_ascii_alphabetic()) {
return Err(HttpError::BadRequest("label must contain a letter".into()));
let new_label = payload.new.as_deref().map(str::trim);
// Length (1..=67), allowed charset, and at least one letter.
if let Some(label) = new_label {
if label.is_empty() || !valid_charset(label) {
return Err(HttpError::BadRequest("invalid label".into()));
}
if !label.chars().any(|c| c.is_ascii_alphabetic()) {
return Err(HttpError::BadRequest("label must contain a letter".into()));
}
}
// Empty values are allowed (they clear the field).
if let Some(custom_id) = payload.custom_id.as_deref()
&& !custom_id.is_empty() && !valid_charset(custom_id) {
return Err(HttpError::BadRequest("invalid custom id".into()));
}
if let Some(pwd) = payload.password.as_deref()
&& !pwd.is_empty() && !valid_charset(pwd) {
return Err(HttpError::BadRequest("invalid password".into()));
}
let stream_key = stream_key::Entity::find_by_id(payload.id)
.one(&state.db)
@@ -153,10 +179,60 @@ async fn edit_stream_key(
if stream_key.user_id != auth.0.id {
return Err(HttpError::Forbidden);
}
stream_key
.into_active_model()
.change_label_value(&state.db, new_label.to_string())
.await?;
let lock = state.appstate.lock().await;
let mut ses = { lock.stream_sessions.get_mut(&payload.id) };
// drop(lock);
let mut am = stream_key.into_active_model();
if let Some(custom_id) = payload.custom_id {
am.custom_id = if custom_id.is_empty() {
if let Some(ref mut ses) = ses {
ses.custom_id = None;
}
Set(None)
} else {
// Check for conflict.
if stream_key::Entity::find_by_custom_id(&state.db, custom_id.clone())
.await?
.is_some()
{
return Err(HttpError::Conflict);
}
if let Some(ref mut ses) = ses {
ses.custom_id = Some(custom_id.clone());
}
Set(Some(custom_id))
};
}
if let Some(label) = new_label {
if let Some(ref mut ses) = ses {
ses.stream_key_label = label.to_string();
}
am.label = Set(label.to_string());
}
if let Some(pwd) = payload.password {
am.password = if pwd.is_empty() {
if let Some(ref mut ses) = ses {
ses.password = None;
}
Set(None)
} else {
if let Some(ref mut ses) = ses {
ses.password = Some(pwd.clone());
}
Set(Some(pwd))
};
}
if let Some(unlisted) = payload.unlisted {
if let Some(ref mut ses) = ses {
ses.is_unlisted = unlisted;
}
am.is_unlisted = Set(unlisted);
}
// This hopefully will not fail, if it does, our values for the stream key will be mismatched,
// that would be bad
am.update(&state.db).await?;
Ok(StatusCode::OK)
}
@@ -166,39 +242,22 @@ async fn catalog_handler(
) -> Result<impl IntoResponse, HttpError> {
let streams = stream_session::Model::get_all_active_sessions(&state.db).await?;
debug!("{:#?}", streams);
let catalog: Vec<StreamListing> = futures::future::join_all(streams.iter().map(|listing| {
let db = state.db.clone();
let stream_key_id = listing.stream_key_id;
let state2 = state.clone();
async move {
let key_info = stream_key::Entity::find_by_id(stream_key_id)
.one(&db)
.await?
.ok_or(HttpError::NotFound)?;
let user = users::Entity::find_by_id(key_info.user_id)
.one(&db)
.await?
.ok_or(HttpError::NotFound)?;
let meow = state2
.appstate
.clone()
.lock()
.await
.stream_sessions
.get(&key_info.id)
.ok_or(HttpError::NotFound)?
.started_at;
Ok::<_, HttpError>(StreamListing {
id: stream_key_id,
label: key_info.label,
user: user.username,
started_at: meow,
})
}
}))
.await
.into_iter()
.collect::<Result<Vec<_>, HttpError>>()?;
let catalog: Vec<StreamListing> = state
.appstate
.lock()
.await
.stream_sessions
.iter()
.filter(|x| !x.is_unlisted)
.map(|x| StreamListing {
id: x.stream_key_id,
label: x.stream_key_label.clone(),
custom_url_label: x.custom_id.clone().unwrap_or_default(),
user: x.stream_key_user.clone(),
started_at: x.started_at,
is_password_protected: x.password.is_some(),
})
.collect::<Vec<_>>();
Ok(Json(catalog))
}
@@ -216,12 +275,18 @@ impl FromRequestParts<Arc<HttpServer>> for AuthUser {
parts: &mut Parts,
state: &Arc<HttpServer>,
) -> Result<Self, HttpError> {
let jar = CookieJar::from_request_parts(parts, state).await.unwrap();
let jar = CookieJar::from_headers(&parts.headers);
let session = jar.get("session").ok_or(HttpError::Unauthorized)?;
let session_val = session.value().to_string();
tracing::debug!(?session_val, "AuthUser: extracted session cookie");
let user = users::Entity::find_by_auth_session(&state.db, session.to_string())
.await?
.ok_or(HttpError::Unauthorized)?;
let user = match users::Entity::find_by_auth_session(&state.db, session_val.clone()).await {
Ok(u) => u.ok_or(HttpError::Unauthorized)?,
Err(e) => {
tracing::error!(?session_val, error = %e, "AuthUser: db query failed");
return Err(HttpError::DbErr(e));
}
};
Ok(AuthUser(user))
}
@@ -260,16 +325,6 @@ async fn create_stream_key_handler(
Ok(StatusCode::CREATED)
}
struct StreamKeys {
keys: Vec<StreamKey>,
}
struct StreamKey {
id: i32,
label: String,
value: String,
}
async fn get_all_stream_keys(
State(state): State<Arc<HttpServer>>,
auth: AuthUser,
@@ -284,15 +339,41 @@ struct LoginForm {
password: String,
}
#[derive(Serialize)]
struct LoginResponse {
session_token: String,
struct DevFlag(bool);
impl<S> FromRequestParts<S> for DevFlag
where
S: Send + Sync,
{
type Rejection = std::convert::Infallible;
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
let query = parts.uri.query().unwrap_or("");
Ok(DevFlag(query.contains("dev=1")))
}
}
fn cookie_for_token(token: &str, dev: bool) -> Cookie<'static> {
let token = token.to_owned();
let mut cookie = Cookie::build(("session", token))
.path("/")
.max_age(time::Duration::days(30))
.same_site(if dev {
axum_extra::extract::cookie::SameSite::None
} else {
axum_extra::extract::cookie::SameSite::Lax
});
if dev {
cookie = cookie.secure(true);
}
cookie.build()
}
#[axum::debug_handler]
async fn login_handler(
State(state): State<Arc<HttpServer>>,
DevFlag(dev): DevFlag,
Json(payload): Json<LoginForm>,
jar: CookieJar,
) -> Result<impl IntoResponse, HttpError> {
let user = users::Entity::find_by_username(&state.db, payload.username.clone())
.await?
@@ -310,8 +391,11 @@ async fn login_handler(
let auth = auth_session::Entity::create(&state.db, user.id).await?;
let token = auth.value;
let jar = jar.add(Cookie::new("session", token));
Ok((StatusCode::OK, jar))
let cookie = cookie_for_token(&token, dev);
Ok((
StatusCode::OK,
[(axum::http::header::SET_COOKIE, cookie.to_string())],
))
}
#[derive(Deserialize)]
@@ -323,10 +407,10 @@ struct CreateUserForm {
async fn create_user_handler(
State(state): State<Arc<HttpServer>>,
DevFlag(dev): DevFlag,
Json(payload): Json<CreateUserForm>,
jar: CookieJar,
) -> Result<impl IntoResponse, HttpError> {
if state.config.signup_code.is_empty() || payload.ref_token != state.config.signup_code {
if state.signup_code.is_empty() || payload.ref_token != state.signup_code {
warn!(username = %payload.username, "signup rejected: invalid signup code");
return Err(HttpError::Unauthorized);
}
@@ -352,36 +436,58 @@ async fn create_user_handler(
let session = auth_session::Entity::create(&state.db, user.id).await?;
let token = session.value;
let jar = jar.add(Cookie::new("session", token));
Ok((jar, StatusCode::OK))
let cookie = cookie_for_token(&token, dev);
Ok((
StatusCode::OK,
[(axum::http::header::SET_COOKIE, cookie.to_string())],
))
}
async fn stream_handler(
State(state): State<Arc<HttpServer>>,
Path(slug): Path<String>,
headers: HeaderMap,
body: String,
) -> Result<impl IntoResponse, HttpError> {
let request_id_clone = state.request_count.fetch_add(1, Ordering::Relaxed);
let app = state.appstate.lock().await;
let stream_key_id = {
let app = state.appstate.lock().await;
app.stream_sessions
.iter()
.find(|e| e.value().stream_key_id.to_string() == slug)
.map(|e| *e.key())
app.stream_sessions.iter().find(|e| {
e.value().stream_key_id.to_string() == slug
|| e.value().custom_id.as_deref() == Some(slug.as_str())
})
// .map(|e| *e.key())
};
let stream_key_id = stream_key_id.ok_or_else(|| {
warn!(slug = %slug, "WHEP request for unknown or inactive stream");
// warn!(slug = %slug, "WHEP request for unknown or inactive stream");
HttpError::NotFound
})?;
info!(request_id = request_id_clone, slug = %slug, stream_key_id, "WHEP offer received");
let mut accept_rx = state.accept_rx.activate_cloned();
// Dont return stream by its id in the db if its unlisted
if stream_key_id.is_unlisted && stream_key_id.custom_id.clone().ok_or("") != Ok(slug.clone()) {
return Err(HttpError::NotFound);
}
let auth_header = headers.get("auth");
if let Some(password) = &stream_key_id.password {
if let Some(auth) = auth_header {
if auth.to_str().unwrap() != password {
return Err(HttpError::Unauthorized);
}
} else {
return Err(HttpError::Unauthorized);
};
};
info!(request_id = request_id_clone, slug = %slug, stream_key_id.stream_key_id, "WHEP offer received");
let accept_rx = state.accept_rx.activate_cloned();
// The webrtc worker owning the receiver died if this fails.
state
.offer_tx
.send((request_id_clone, stream_key_id, body))
.send((request_id_clone, stream_key_id.stream_key_id, body))
.await
.map_err(|_| HttpError::Internal)?;
debug!(
@@ -389,38 +495,104 @@ async fn stream_handler(
"offer sent, waiting for answer"
);
let reply_body = String::new();
while let Ok(answer) = accept_rx.recv().await {
debug!(request_id = request_id_clone, "received answer candidate");
if let Some(reply) = answer.1 {
if answer.0 == request_id_clone {
return Ok(Response::builder()
.status(StatusCode::CREATED)
.header("content-type", "application/sdp")
.body(reply)
.unwrap());
} else {
continue;
// Bound the wait: a WebRTC setup failure (malformed SDP, codec mismatch)
// would otherwise leave this HTTP request hanging forever.
match tokio::time::timeout(Duration::from_secs(10), async {
let mut accept_rx = accept_rx;
loop {
match accept_rx.recv().await {
Ok(answer) if answer.0 == request_id_clone => return Some(answer.1),
Ok(_) => continue,
Err(_) => return None,
}
} else if answer.0 == request_id_clone {
}
})
.await
{
Ok(Some(Ok(reply))) => {
return Ok(Response::builder()
.status(StatusCode::UNSUPPORTED_MEDIA_TYPE)
.body(String::new())
.status(StatusCode::CREATED)
.header("content-type", "application/sdp")
.body(reply)
.unwrap());
}
}
info!(
request_id = request_id_clone,
"answer channel closed without a match"
);
Ok(Some(Err(codec))) => {
return Err(HttpError::WhepCodecError(codec));
}
Ok(None) => {
info!(
request_id = request_id_clone,
"answer channel closed without a match"
);
}
Err(_) => {
warn!(
request_id = request_id_clone,
"timed out waiting for WHEP answer"
);
}
};
Ok(Response::builder()
.status(StatusCode::CREATED)
.header("content-type", "application/sdp")
.body(reply_body)
.status(StatusCode::GATEWAY_TIMEOUT)
.body(String::new())
.unwrap())
}
async fn meow_handler() -> &'static str {
"meow"
}
#[derive(Serialize)]
struct HealthResponse {
status: &'static str,
uptime_seconds: u64,
version: &'static str,
}
async fn health_handler(
State(state): State<Arc<HttpServer>>,
) -> Result<impl IntoResponse, HttpError> {
let uptime = state.start_time.elapsed().as_secs();
Ok((
StatusCode::OK,
Json(HealthResponse {
status: "ok",
uptime_seconds: uptime,
version: state.version,
}),
))
}
#[derive(Serialize)]
struct StatsResponse {
version: &'static str,
uptime_seconds: u64,
cpu_usage_percent: f32,
active_streams: usize,
request_count: i32,
}
async fn stats_handler(
State(state): State<Arc<HttpServer>>,
) -> Result<impl IntoResponse, HttpError> {
let uptime = state.start_time.elapsed().as_secs();
let mut system = System::new();
system.refresh_cpu_all();
let cpu_usage = system.global_cpu_usage();
let active_streams = {
let app = state.appstate.lock().await;
app.stream_sessions.len()
};
let request_count = state.request_count.load(Ordering::Relaxed);
Ok((
StatusCode::OK,
Json(StatsResponse {
version: state.version,
uptime_seconds: uptime,
cpu_usage_percent: cpu_usage,
active_streams,
request_count,
}),
))
}
+11 -6
View File
@@ -25,6 +25,8 @@ pub enum HttpError {
Unprocessable(String),
#[error("not acceptable: {0}")]
NotAcceptable(String),
#[error("unsupported codec: {0}")]
WhepCodecError(String),
#[error("internal error")]
Internal,
}
@@ -40,6 +42,7 @@ impl HttpError {
Self::BadRequest(_) => StatusCode::BAD_REQUEST,
Self::Unprocessable(_) => StatusCode::UNPROCESSABLE_ENTITY,
Self::NotAcceptable(_) => StatusCode::NOT_ACCEPTABLE,
Self::WhepCodecError(_) => StatusCode::UNSUPPORTED_MEDIA_TYPE,
}
}
}
@@ -47,12 +50,14 @@ impl HttpError {
impl IntoResponse for HttpError {
fn into_response(self) -> Response<Body> {
let status = self.status();
// Only surface a message body for client (4xx) errors; keep an empty
// body for 5xx so internal details aren't leaked.
let body = if status.is_client_error() {
self.to_string()
} else {
String::new()
let body = match self {
// The WHEP client (frontend) reads this body as the rejected
// codec, so send it bare rather than the full error string.
Self::WhepCodecError(codec) => codec,
// Only surface a message body for client (4xx) errors; keep an
// empty body for 5xx so internal details aren't leaked.
e if status.is_client_error() => e.to_string(),
_ => String::new(),
};
(status, body).into_response()
}
+59 -41
View File
@@ -1,49 +1,40 @@
use ::chrono::{DateTime, Utc};
use std::{env, error::Error, sync::Arc};
use tracing::{info, level_filters::LevelFilter, warn};
use std::{
env,
error::Error,
sync::{Arc, atomic::AtomicU32},
};
use tracing::{info, warn};
use async_broadcast::broadcast;
use dashmap::DashMap;
use entity::stream_session;
use migration::{Migrator, MigratorTrait};
use rml_rtmp::{
handshake::{Handshake, HandshakeProcessResult, PeerType},
sessions::{ServerSession, ServerSessionConfig, ServerSessionEvent, ServerSessionResult},
};
use sea_orm::{
Database, IntoActiveModel,
sqlx::types::chrono::{self, Local},
};
use tokio::{
fs,
io::{AsyncReadExt, AsyncWriteExt},
net::TcpListener,
sync::Mutex,
task::JoinSet,
time::Instant,
};
use sea_orm::Database;
use tokio::{net::TcpListener, sync::Mutex, task::JoinSet};
use tracing_subscriber::EnvFilter;
use uuid::Uuid;
use crate::{
audio::OpusAudioFrame,
codec::VideoFrame,
http::{HttpServer, HttpServerConfig},
webrtc_proxy::WebRtcProxyConfig,
audio::OpusAudioFrame, codec::VideoFrame, http::HttpServer, webrtc_proxy::WebrtcProxy,
};
const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION");
mod audio;
mod codec;
mod hash;
mod http;
mod http_error;
mod rtmp;
mod stream_session_2;
mod webrtc;
mod webrtc_ingest;
mod webrtc_proxy;
// #[derive(Debug)]
pub struct AppState {
pub stream_sessions: Arc<DashMap<i32, StreamSession>>,
pub webrtc_proxy: WebrtcProxy,
}
#[derive(Clone, Debug)]
@@ -53,14 +44,42 @@ pub enum StreamCodec {
AV1,
}
impl StreamCodec {
/// Map str0m's codec enum to ours; None for non-video codecs.
pub fn from_str0m(c: str0m::format::Codec) -> Option<Self> {
use str0m::format::Codec;
match c {
Codec::H264 => Some(Self::H264),
Codec::H265 => Some(Self::H265),
Codec::Av1 => Some(Self::AV1),
_ => None,
}
}
}
pub struct StreamSession {
pub stream_key_id: i32,
pub stream_key_label: String,
pub stream_key_user: String,
pub custom_id: Option<String>,
pub is_unlisted: bool,
pub password: Option<String>,
pub frame_channel: async_broadcast::Sender<Arc<VideoFrame>>,
pub audio_channel: async_broadcast::Sender<Arc<OpusAudioFrame>>,
pub codec: Option<StreamCodec>,
//
// ---- WHIP-negotiated video codec details (populated by whip ingest) ----
/// Payload type from WHIP SDP negotiation.
pub video_pt: Option<u8>,
/// H.264 profile_level_id from WHIP SDP negotiation (e.g. 0x42e01f for CBP).
/// When set alongside `video_pt`, the WHEP answer only advertises this exact
/// profile instead of all three H.264 profiles.
pub video_profile_level_id: Option<u32>,
// ----
/// Unique id per publish. Lets a session's cleanup verify it is still the
/// live session for its key instead of clobbering a newer one.
pub session_id: Uuid,
pub started_at: DateTime<Utc>,
pub active_clients: AtomicU32,
}
#[tokio::main]
@@ -83,16 +102,23 @@ async fn main() -> Result<(), Box<dyn Error>> {
stream_session::Model::clean_unended_streams(&db)
.await
.unwrap();
let proxyconfig = env::var("RTC_PORT")
.unwrap_or("6969".into())
.parse()
.expect("RTC_PORT needs to be a number (i32)");
let proxy = webrtc_proxy::WebrtcProxy::new(proxyconfig).await.unwrap();
let appstate = Arc::new(Mutex::new(AppState {
stream_sessions: Arc::new(DashMap::new()),
webrtc_proxy: proxy.clone(),
}));
let (offer_tx, offer_rx) = tokio::sync::mpsc::channel::<(i32, i32, String)>(64);
// Request_Id,
// String_Label,
// Offer_body
let (answer_tx, answer_rx) = broadcast::<(i32, Option<String>)>(64);
let (answer_tx, answer_rx) = broadcast::<(i32, Result<String, String>)>(64);
// Request_Id,
// Answer_body
@@ -102,21 +128,13 @@ async fn main() -> Result<(), Box<dyn Error>> {
appstate: appstate.clone(),
request_count: std::sync::atomic::AtomicI32::new(0),
db: db.clone(),
config: HttpServerConfig {
signup_code: env::var("SIGNUP_CODE").unwrap_or_else(|_| {
warn!("SIGNUP_CODE not set; signup will be disabled");
String::new()
}),
}
.into(),
signup_code: env::var("SIGNUP_CODE").unwrap_or_else(|_| {
warn!("SIGNUP_CODE not set; signup will be disabled");
String::new()
}),
version: SERVER_VERSION,
start_time: std::time::Instant::now(),
};
let proxyconfig = WebRtcProxyConfig {
proxy_port: env::var("RTC_PORT")
.unwrap_or("6969".into())
.parse()
.expect("RTC_PORT needs to be a number (i32)"),
};
let proxy = webrtc_proxy::WebrtcProxy::new(proxyconfig).await.unwrap();
let app = appstate.lock().await;
let webrtc = webrtc::Webrtc {
@@ -124,7 +142,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
accept_tx: answer_tx,
sessions_ref: app.stream_sessions.clone(),
db: db.clone(),
proxy: proxy.clone().into(),
proxy: proxy.clone().into(), // The entire of the proxy's internal are within an arc, cheap
// to clone
};
let rtmp = rtmp::Rtmp {
@@ -148,7 +167,6 @@ async fn main() -> Result<(), Box<dyn Error>> {
res = workers.join_next() => {
if let Some(Err(e)) = res {
tracing::error!("worker panicked: {:?}", e);
// fs::File::
} else {
tracing::error!("a worker exited unexpectedly");
}
+124 -82
View File
@@ -3,18 +3,19 @@ use std::{error::Error, sync::Arc, time::Duration};
use async_broadcast::broadcast;
use chrono::Utc;
use dashmap::DashMap;
use entity::stream_session;
use entity::{stream_session, users};
use rml_rtmp::{
handshake::{Handshake, HandshakeProcessResult, PeerType},
sessions::{ServerSession, ServerSessionConfig, ServerSessionEvent, ServerSessionResult},
};
use sea_orm::{DatabaseConnection, IntoActiveModel, sqlx::types::chrono::Local};
use sea_orm::{DatabaseConnection, EntityTrait, IntoActiveModel, sqlx::types::chrono::Local};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::{TcpListener, TcpStream},
time::{Instant, timeout},
time::timeout,
};
use tracing::{debug, info, warn};
use uuid::Uuid;
use crate::{
StreamCodec, StreamSession,
@@ -24,6 +25,8 @@ use crate::{
},
};
const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30);
pub struct Rtmp {
pub listener: TcpListener,
pub stream_sessions: Arc<DashMap<i32, StreamSession>>,
@@ -32,8 +35,11 @@ pub struct Rtmp {
async fn write_outbound(socket: &mut TcpStream, results: Vec<ServerSessionResult>) {
for r in results {
if let ServerSessionResult::OutboundResponse(p) = r {
socket.write_all(&p.bytes).await.unwrap();
if let ServerSessionResult::OutboundResponse(p) = r
&& let Err(e) = socket.write_all(&p.bytes).await
{
warn!("RTMP write error: {e}");
return;
}
}
}
@@ -69,7 +75,10 @@ impl Rtmp {
let mut server = Handshake::new(PeerType::Server);
let mut c0_c1 = [0u8; 1537];
socket.read_exact(&mut c0_c1).await?;
let n = timeout(HANDSHAKE_TIMEOUT, socket.read_exact(&mut c0_c1))
.await
.map_err(|_| "handshake C0+C1 timeout")??;
debug!("handshake C0+C1 read {} bytes", n);
let s0_s1_s2 = match server.process_bytes(&c0_c1) {
Ok(HandshakeProcessResult::InProgress { response_bytes }) => response_bytes,
x => return Err(format!("unexpected handshake state: {:?}", x).into()),
@@ -77,7 +86,9 @@ impl Rtmp {
socket.write_all(&s0_s1_s2).await?;
let mut c2 = [0u8; 1536];
socket.read_exact(&mut c2).await?;
timeout(HANDSHAKE_TIMEOUT, socket.read_exact(&mut c2))
.await
.map_err(|_| "handshake C2 timeout")??;
match server.process_bytes(&c2) {
Ok(HandshakeProcessResult::Completed { .. }) => {}
Ok(HandshakeProcessResult::InProgress { response_bytes }) => {
@@ -86,7 +97,8 @@ impl Rtmp {
x => return Err(format!("unexpected handshake state: {:?}", x).into()),
}
let (rtmp_session, init_bytes) = ServerSession::new(ServerSessionConfig::new()).unwrap();
let (rtmp_session, init_bytes) = ServerSession::new(ServerSessionConfig::new())
.map_err(|e| format!("ServerSession::new failed: {e}"))?;
write_outbound(&mut socket, init_bytes).await;
Ok((rtmp_session, socket))
@@ -116,9 +128,12 @@ impl Rtmp {
}
};
info!(%peer_addr, "RTMP handshake complete");
let (mut video_tx, mut video_rx) = broadcast::<Arc<VideoFrame>>(32);
let (mut audio_tx, mut audio_rx) = broadcast::<Arc<OpusAudioFrame>>(32);
// video_rx.cycle
// Ring sized by TIME, not frame count: 512 frames ≈ 5.7s @ 90fps,
// so transient ingest/send stalls can't overflow it (overflow drops
// the oldest frames and forces a freeze until the next keyframe).
// ponytail: fixed 512; per-stream dynamic sizing if memory ever matters.
let (mut video_tx, video_rx) = broadcast::<Arc<VideoFrame>>(512);
let (mut audio_tx, audio_rx) = broadcast::<Arc<OpusAudioFrame>>(512);
video_tx.set_overflow(true);
audio_tx.set_overflow(true);
@@ -193,12 +208,23 @@ impl Rtmp {
for event in events {
match event {
ServerSessionResult::OutboundResponse(p) => {
socket.write_all(&p.bytes).await.unwrap();
if let Err(e) = socket.write_all(&p.bytes).await {
warn!("RTMP outbound write error: {e}");
return;
}
}
ServerSessionResult::RaisedEvent(e) => match e {
ServerSessionEvent::ConnectionRequested { request_id, .. } => {
debug!("RTMP ConnectionRequested, accepting");
let reply = session.accept_request(request_id).unwrap();
let reply = match session.accept_request(request_id) {
Ok(r) => r,
Err(e) => {
warn!(
"Failed to accept connection request {request_id}: {e}"
);
return;
}
};
write_outbound(&mut socket, reply).await;
}
ServerSessionEvent::PublishStreamRequested {
@@ -215,54 +241,113 @@ impl Rtmp {
key
} else {
warn!(stream_key = %stream_key, "stream key not found, rejecting");
let reply = session
.reject_request(request_id, "", "Stream key invalid")
.unwrap();
let reply = match session.reject_request(
request_id,
"",
"Stream key invalid",
) {
Ok(r) => r,
Err(e) => {
warn!("Failed to reject stream key request: {e}");
return;
}
};
write_outbound(&mut socket, reply).await;
break;
};
let already_live =
stream_session::Model::get_active_by_stream_key_id(
match stream_session::Model::get_active_by_stream_key_id(
&db, key.id,
)
.await
.unwrap();
{
Ok(s) => s,
Err(e) => {
warn!("DB error checking active stream: {e}");
let reply = match session.reject_request(
request_id,
"",
"Internal error",
) {
Ok(r) => r,
Err(e2) => {
warn!(
"Failed to reject request on DB error: {e2}"
);
return;
}
};
write_outbound(&mut socket, reply).await;
return;
}
};
if already_live.is_some() {
warn!(stream_key_id = key.id, label = %key.label, "stream key already live, rejecting duplicate publish");
let reply = session
.reject_request(
request_id,
"",
"You're already streaming...",
)
.unwrap();
let reply = match session.reject_request(
request_id,
"",
"You're already streaming...",
) {
Ok(r) => r,
Err(e) => {
warn!("Failed to reject duplicate publish: {e}");
return;
}
};
write_outbound(&mut socket, reply).await;
break;
}
info!(stream_key_id = key.id, label = %key.label, "stream started");
let user = users::Entity::find_by_id(key.user_id)
.one(&db)
.await
.unwrap()
.unwrap();
current_stream_key_id = Some(key.id);
stream_sessions.insert(
key.id,
StreamSession {
stream_key_id: key.id,
stream_key_label: key.label,
stream_key_user: user.username,
custom_id: key.custom_id,
is_unlisted: key.is_unlisted,
password: key.password,
frame_channel: video_tx.clone(),
audio_channel: audio_tx.clone(),
codec: None,
started_at: Utc::now(),
active_clients: 0.into(),
session_id: Uuid::new_v4(),
video_pt: None,
video_profile_level_id: None,
},
);
let reply = session.accept_request(request_id).unwrap();
stream_session::Model::create_stream_session(
let reply = match session.accept_request(request_id) {
Ok(r) => r,
Err(e) => {
warn!("Failed to accept publish request: {e}");
return;
}
};
if let Err(e) = stream_session::Model::create_stream_session(
&db,
key.id,
Local::now().into(),
)
.await
.unwrap();
{
warn!("Failed to create stream session record: {e}");
let _ = session.reject_request(
request_id,
"",
"Internal error",
);
return;
}
stream_id = Some(key.id);
write_outbound(&mut socket, reply).await;
}
@@ -283,69 +368,26 @@ impl Rtmp {
}
current_stream_key_id = None;
}
// TODO: We can totally replace the broadcast with a
// circular_buff
// Arc<Vec<ArcSwap<Frame>>>
ServerSessionEvent::VideoDataReceived {
data, timestamp, ..
} => match Self::parse_video_codec(&data) {
Ok(codec) => {
if !codec_stamped {
if let Some(id) = current_stream_key_id {
if let Some(mut session) =
if let Some(id) = current_stream_key_id
&& let Some(mut session) =
stream_sessions.get_mut(&id)
{
session.codec = Some(codec.clone());
}
{
session.codec = Some(codec.clone());
}
codec_stamped = true;
}
match codec {
StreamCodec::H264 => {
let p = parser.get_or_insert_with(|| {
Box::new(H264CodecParser::new())
});
if let Some(frame) = p.parse(&data, timestamp.value)
{
video_tx.broadcast(Arc::new(frame)).await.ok();
}
}
StreamCodec::H265 => {
let p = parser.get_or_insert_with(|| {
Box::new(H265CodecParser::new())
});
if let Some(frame) = p.parse(&data, timestamp.value)
{
video_tx.broadcast(Arc::new(frame)).await.ok();
}
}
StreamCodec::AV1 => {
let p = parser.get_or_insert_with(|| {
Box::new(Av1CodecParser::new())
});
let pkt_type = data[0] & 0x0F;
match p.parse(&data, timestamp.value) {
Some(frame) => {
debug!(
pkt_type,
is_keyframe = frame.is_keyframe,
ts = frame.timestamp_ms,
bytes = frame.data.len(),
"AV1 frame → broadcast"
);
video_tx
.broadcast(Arc::new(frame))
.await
.ok();
}
None => {
debug!(
pkt_type,
"AV1 packet produced no frame (seq header or unknown type)"
);
}
}
}
let p = parser.get_or_insert_with(|| match codec {
StreamCodec::H264 => Box::new(H264CodecParser::new()),
StreamCodec::H265 => Box::new(H265CodecParser::new()),
StreamCodec::AV1 => Box::new(Av1CodecParser::new()),
});
if let Some(frame) = p.parse(&data, timestamp.value) {
video_tx.broadcast(Arc::new(frame)).await.ok();
}
}
Err(_err) => {
+18
View File
@@ -0,0 +1,18 @@
use serde::Serialize;
use crate::StreamSession;
#[derive(Serialize)]
pub struct StreamUpdateData {
viewers: u32,
}
impl StreamSession {
pub fn stream_update_data(&self) -> StreamUpdateData {
StreamUpdateData {
viewers: self
.active_clients
.load(std::sync::atomic::Ordering::Relaxed),
}
}
}
+168 -125
View File
@@ -2,13 +2,18 @@ use bytes::Bytes;
use dashmap::DashMap;
use entity::stream_key;
use sea_orm::{DatabaseConnection, EntityTrait};
use std::{net::SocketAddr, sync::Arc, time::Instant};
use std::{
net::SocketAddr,
sync::Arc,
time::{Duration, Instant},
};
use tokio::{net::UdpSocket, sync::mpsc::Receiver};
use tracing::{debug, error, info, warn};
use tracing::{debug, error, info, trace, warn};
use str0m::{
Candidate, Event, Input, Output, Rtc,
change::SdpOffer,
channel::ChannelId,
media::{Frequency, MediaKind, MediaTime, Mid, Pt},
net::{Protocol, Receive},
};
@@ -19,14 +24,12 @@ use crate::{
pub struct Webrtc {
pub offer_rx: Receiver<(i32, i32, String)>,
pub accept_tx: async_broadcast::Sender<(i32, Option<String>)>,
pub accept_tx: async_broadcast::Sender<(i32, Result<String, String>)>,
pub sessions_ref: Arc<DashMap<i32, StreamSession>>,
pub proxy: Arc<WebrtcProxy>,
pub db: DatabaseConnection,
}
const PER_CLIENT_CONNECTION_BUF: usize = 65535;
impl Webrtc {
pub async fn run(mut self) {
while let Some(offer) = self.offer_rx.recv().await {
@@ -42,7 +45,10 @@ impl Webrtc {
request_id,
stream_id, "stream key not found in DB, rejecting offer"
);
self.accept_tx.broadcast((request_id, None)).await.unwrap();
self.accept_tx
.broadcast((request_id, Err(String::new())))
.await
.unwrap();
continue;
}
if let Err(ref e) = stream_key {
@@ -50,48 +56,77 @@ impl Webrtc {
request_id,
stream_id, "DB error looking up stream key: {:?}", e
);
self.accept_tx.broadcast((request_id, None)).await.unwrap();
self.accept_tx
.broadcast((request_id, Err(String::new())))
.await
.unwrap();
continue;
}
// let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
// let local_addr = socket.local_addr().unwrap();
let local_addr = self.proxy.public_addr();
let stream_codec = self
.sessions_ref
.get(&stream_id)
.and_then(|s| s.codec.clone());
let session = self.sessions_ref.get(&stream_id);
let stream_codec = session.as_ref().and_then(|s| s.codec.clone());
let video_pt = session.as_ref().and_then(|s| s.video_pt);
let video_profile = session.as_ref().and_then(|s| s.video_profile_level_id);
drop(session);
info!(request_id, stream_id, codec = ?match &stream_codec {
Some(StreamCodec::H264) => "H264",
Some(StreamCodec::H265) => "H265",
Some(StreamCodec::AV1) => "AV1",
None => "unknown",
}, "configuring RTC codec");
}, video_pt, video_profile, "configuring RTC codec");
let mut builder = Rtc::builder();
{
use str0m::format::{Codec, FormatParams};
use str0m::media::Frequency;
let cc = builder.codec_config();
cc.clear();
cc.enable_opus(true);
match stream_codec {
Some(StreamCodec::H265) => {
// When video_pt is set (WHIP-negotiated), enable only that
// exact codec+PT+profile. Otherwise fall back to enabling all
// common profiles for the codec family (RTMP ingest doesn't
// carry profile info, so we cast a wide net for browser compat).
match (stream_codec.clone(), video_pt, video_profile) {
// --- Specific PT + profile (WHIP path) ---
(Some(StreamCodec::H264), Some(pt), Some(profile)) => {
cc.add_h264(pt.into(), None, true, profile);
}
(Some(StreamCodec::H264), Some(pt), None) => {
// PT known, profile unknown — default to Constrained
// Baseline (works everywhere but may mismatch for
// Main/High streams from OBS).
cc.add_h264(pt.into(), None, true, 0x42e01f);
}
(Some(StreamCodec::H265), Some(pt), _) => {
cc.add_h265(pt.into(), None, 1, 0, 180);
}
(Some(StreamCodec::AV1), Some(pt), _) => {
cc.add_config(
pt.into(),
None,
Codec::Av1,
Frequency::NINETY_KHZ,
None,
FormatParams::default(),
);
}
// --- Broad codec family (RTMP path, no WHIP info) ---
(Some(StreamCodec::H265), None, _) => {
cc.enable_h265(true);
}
Some(StreamCodec::AV1) => {
(Some(StreamCodec::AV1), None, _) => {
cc.enable_av1(true);
}
_ => {
// Advertise the common H.264 profiles. str0m matches an
// incoming offer's payload by profile-level-id, so we must
// list every profile a browser might offer or negotiation
// fails and the m=video line comes back with an empty PT
// list. Firefox in particular only ever offers Constrained
// Baseline (0x42e01f) — without it Firefox sees no codec.
cc.add_h264(102.into(), None, true, 0x42e01f); // Constrained Baseline
cc.add_h264(104.into(), None, true, 0x4d001f); // Main
cc.add_h264(106.into(), None, true, 0x64001f); // High
cc.add_h264(102.into(), None, true, 0x42e01f);
cc.add_h264(104.into(), None, true, 0x4d001f);
cc.add_h264(106.into(), None, true, 0x64001f);
}
}
}
@@ -100,7 +135,20 @@ impl Webrtc {
let candidate = Candidate::host(local_addr, Protocol::Udp).unwrap();
rtc.add_local_candidate(candidate);
let offer_sdp = SdpOffer::from_sdp_string(&sdp_body).unwrap();
// Malformed SDP from an unauthenticated client must not panic here:
// a panic kills the webrtc worker and main() aborts every other
// worker, taking the whole server down.
let offer_sdp = match SdpOffer::from_sdp_string(&sdp_body) {
Ok(sdp) => sdp,
Err(e) => {
warn!(request_id, stream_id, "malformed SDP offer: {:?}", e);
self.accept_tx
.broadcast((request_id, Err(String::new())))
.await
.unwrap();
continue;
}
};
let mut changes = rtc.sdp_api();
let mid = changes.add_media(
MediaKind::Video,
@@ -125,7 +173,7 @@ impl Webrtc {
}
};
let answer_sdp = offer_answer.to_sdp_string();
info!(request_id, "SDP answer:\n{}", answer_sdp);
trace!(request_id, "SDP answer:\n{}", answer_sdp);
// Detect the case where str0m couldn't match any video codec.
// str0m serialises the m-line with an empty PT list, which is invalid SDP
@@ -142,7 +190,10 @@ impl Webrtc {
"no video codec negotiated — browser likely doesn't support {:?}; rejecting offer",
stream_codec
);
self.accept_tx.broadcast((request_id, None)).await.unwrap();
self.accept_tx
.broadcast((request_id, Err(format!("{:?}", stream_codec))))
.await
.unwrap();
continue;
}
@@ -156,7 +207,7 @@ impl Webrtc {
debug!(request_id, "sending answer back");
self.accept_tx
.broadcast((request_id, Some(answer_sdp)))
.broadcast((request_id, Ok(answer_sdp)))
.await
.unwrap();
@@ -190,10 +241,34 @@ impl Webrtc {
let mut video_pt = None;
let mut audio_mid: Option<Mid> = None;
let mut audio_pt = None;
let mut channel_id = None;
let mut connected = false;
let mut video_stream: Option<async_broadcast::Receiver<Arc<VideoFrame>>> = None;
let mut audio_stream: Option<async_broadcast::Receiver<Arc<OpusAudioFrame>>> = None;
let mut saw_keyframe = false;
// Update client with stream info through webrtc data channel ()
let mut tick = tokio::time::interval(Duration::from_millis(2000));
if let Some(ses) = sessions_ref.get(&stream_id) {
ses.active_clients
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
struct ActiveClientGuard {
sessions: Arc<DashMap<i32, StreamSession>>,
id: i32,
}
impl Drop for ActiveClientGuard {
fn drop(&mut self) {
if let Some(ses) = self.sessions.get(&self.id) {
ses.active_clients
.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
}
}
}
let _active_client_guard = ActiveClientGuard {
sessions: sessions_ref.clone(),
id: stream_id,
};
loop {
let deadline = loop {
@@ -206,6 +281,9 @@ impl Webrtc {
}
}
Ok(Output::Event(e)) => match e {
Event::ChannelOpen(channelId, _name) => {
channel_id = Some(channelId);
}
Event::MediaAdded(ma) => {
info!(stream_id, kind = ?ma.kind, mid = ?ma.mid, "MediaAdded");
if ma.kind == MediaKind::Video {
@@ -246,16 +324,16 @@ impl Webrtc {
warn!(stream_id, mid = ?ma.mid, "no writer for video mid");
}
}
if ma.kind == MediaKind::Audio {
if let Some(writer) = rtc.writer(ma.mid) {
let best = writer.payload_params().max_by_key(|p| {
p.spec().format.profile_level_id.unwrap_or(0)
});
if let Some(params) = best {
info!(stream_id, pt = ?params.pt(), "selected audio PT");
audio_pt = Some(params.pt());
audio_mid = Some(ma.mid);
}
if ma.kind == MediaKind::Audio
&& let Some(writer) = rtc.writer(ma.mid)
{
let best = writer
.payload_params()
.max_by_key(|p| p.spec().format.profile_level_id.unwrap_or(0));
if let Some(params) = best {
info!(stream_id, pt = ?params.pt(), "selected audio PT");
audio_pt = Some(params.pt());
audio_mid = Some(ma.mid);
}
}
}
@@ -273,8 +351,8 @@ impl Webrtc {
}
_ => {}
},
Err(e) => {
error!("poll_output error (connection closing): {:?}", e);
Err(_e) => {
// error!("poll_output error (connection closing): {:?}", e);
return;
}
}
@@ -292,27 +370,25 @@ impl Webrtc {
);
}
}
if let Some(video) = &video_stream {
if video.is_closed() {
if let Some(session) = sessions_ref.get(&stream_id) {
video_stream = Some(session.frame_channel.new_receiver());
debug!(stream_id, "subscribed to video channel");
}
}
if let Some(video) = &video_stream
&& video.is_closed()
&& let Some(session) = sessions_ref.get(&stream_id)
{
video_stream = Some(session.frame_channel.new_receiver());
debug!(stream_id, "subscribed to video channel");
}
if audio_stream.is_none() {
if let Some(session) = sessions_ref.get(&stream_id) {
audio_stream = Some(session.audio_channel.new_receiver());
debug!(stream_id, "subscribed to audio channel");
}
if audio_stream.is_none()
&& let Some(session) = sessions_ref.get(&stream_id)
{
audio_stream = Some(session.audio_channel.new_receiver());
debug!(stream_id, "subscribed to audio channel");
}
if let Some(audio) = &audio_stream {
if audio.is_closed() {
if let Some(session) = sessions_ref.get(&stream_id) {
audio_stream = Some(session.audio_channel.new_receiver());
debug!(stream_id, "subscribed to audio channel");
}
}
if let Some(audio) = &audio_stream
&& audio.is_closed()
&& let Some(session) = sessions_ref.get(&stream_id)
{
audio_stream = Some(session.audio_channel.new_receiver());
debug!(stream_id, "subscribed to audio channel");
}
}
@@ -333,9 +409,9 @@ impl Webrtc {
}
}
result = rx.recv() => {
if let Some((data, from)) = result {
if let Ok(contents) = (&data[..]).try_into() {
if let Err(e) = rtc.handle_input(Input::Receive(
if let Some((data, from)) = result
&& let Ok(contents) = (&data[..]).try_into()
&& let Err(e) = rtc.handle_input(Input::Receive(
Instant::now(),
Receive {
proto: Protocol::Udp,
@@ -347,21 +423,25 @@ impl Webrtc {
error!("handle_input(Receive) error: {:?}", e);
return;
}
}
}
}
res = Webrtc::recv_video(&mut video_stream), if video_stream.is_some() => {
match res {
Ok(frame) => {
// One frame per wake: the select! re-arms immediately while
// more frames are pending, so backlog catch-up is paced
// frame-by-frame instead of an 8-frame burst (which overflows
// the browser jitter buffer and self-reinforces the backlog).
Webrtc::write_video_frame(
frame, &mut saw_keyframe, video_pt, video_mid, &mut rtc, stream_id,
);
Webrtc::drain_video(
&mut video_stream, &mut saw_keyframe, video_pt, video_mid, &mut rtc, stream_id,
);
}
Err(async_broadcast::RecvError::Closed) => {
debug!(stream_id, "video channel closed, stream ended");
// Drop the dead receiver so the select! branch disarms
// instead of spinning on Closed every iteration. The
// subscribe block above re-arms it if the same stream
// key is re-published.
video_stream = None;
}
Err(async_broadcast::RecvError::Overflowed(_)) => {
// Frames were dropped from the ring buffer; resuming mid-GOP
@@ -375,14 +455,22 @@ impl Webrtc {
match res {
Ok(frame) => {
Webrtc::write_audio_frame(frame, audio_pt, audio_mid, &mut rtc);
Webrtc::drain_audio(&mut audio_stream, audio_pt, audio_mid, &mut rtc);
}
Err(async_broadcast::RecvError::Closed) => {
debug!("audio channel closed, stream ended");
audio_stream = None;
}
Err(async_broadcast::RecvError::Overflowed(_)) => {}
}
}
_interval = tick.tick() => {
if let Some(id) = channel_id
&& let Some(session) = sessions_ref.get(&stream_id)
&& let Ok(json) = serde_json::to_vec(&session.stream_update_data())
&& let Some(mut ch) = rtc.channel(id) {
let _ = ch.write(false, &json);
}
}
}
}
}
@@ -435,38 +523,6 @@ impl Webrtc {
}
}
fn drain_video(
stream: &mut Option<async_broadcast::Receiver<Arc<VideoFrame>>>,
saw_keyframe: &mut bool,
video_pt: Option<Pt>,
video_mid: Option<Mid>,
rtc: &mut Rtc,
stream_id: i32,
) {
let Some(s) = stream.as_mut() else { return };
for _ in 0..7 {
match s.try_recv() {
Ok(frame) => Webrtc::write_video_frame(
frame,
saw_keyframe,
video_pt,
video_mid,
rtc,
stream_id,
),
Err(async_broadcast::TryRecvError::Empty) => break,
Err(async_broadcast::TryRecvError::Closed) => {
warn!(stream_id, "video channel closed, stream ended");
*stream = None;
break;
}
Err(async_broadcast::TryRecvError::Overflowed(_)) => {
*saw_keyframe = false;
}
}
}
}
fn write_audio_frame(
frame: Arc<OpusAudioFrame>,
audio_pt: Option<Pt>,
@@ -475,31 +531,18 @@ impl Webrtc {
) {
let now = Instant::now();
let rtp_time = MediaTime::new(frame.timestamp_ms as u64 * 48, Frequency::FORTY_EIGHT_KHZ);
if let (Some(pt), Some(writer)) = (audio_pt, audio_mid.and_then(|m| rtc.writer(m))) {
if let Err(e) = writer.write(pt, now, rtp_time, frame.data.to_vec()) {
warn!("RTP write error: {:?}", e);
}
if let (Some(pt), Some(writer)) = (audio_pt, audio_mid.and_then(|m| rtc.writer(m)))
&& let Err(e) = writer.write(pt, now, rtp_time, frame.data.to_vec())
{
warn!("RTP write error: {:?}", e);
}
}
fn drain_audio(
stream: &mut Option<async_broadcast::Receiver<Arc<OpusAudioFrame>>>,
audio_pt: Option<Pt>,
audio_mid: Option<Mid>,
rtc: &mut Rtc,
) {
let Some(s) = stream.as_mut() else { return };
for _ in 0..7 {
match s.try_recv() {
Ok(frame) => Webrtc::write_audio_frame(frame, audio_pt, audio_mid, rtc),
Err(async_broadcast::TryRecvError::Empty) => break,
Err(async_broadcast::TryRecvError::Closed) => {
info!("audio channel closed, stream ended");
*stream = None;
break;
}
Err(async_broadcast::TryRecvError::Overflowed(_)) => {}
}
fn write_channel_data(rtc: &mut Rtc, channel_id: ChannelId, data: &[u8]) {
if let Some(mut channel) = rtc.channel(channel_id)
&& let Err(e) = channel.write(false, data)
{
warn!("Channel write error: {:?}", e)
}
}
}
+624 -8
View File
@@ -1,17 +1,633 @@
use std::sync::Arc;
use std::{
net::SocketAddr,
sync::Arc,
time::{Duration, Instant},
};
use axum::{extract::State, response::IntoResponse};
use str0m::change::SdpOffer;
use tracing::info;
use async_broadcast::broadcast;
use axum::{
extract::{ConnectInfo, Path, State},
http::{HeaderMap, StatusCode, header},
response::IntoResponse,
};
use bytes::Bytes;
use chrono::{Local, Utc};
use dashmap::DashMap;
use entity::{stream_key, stream_session, users};
use sea_orm::{DatabaseConnection, EntityTrait, IntoActiveModel};
use str0m::{
Candidate, Event, Input, Output, Rtc,
change::SdpOffer,
media::{MediaKind, Mid},
net::{Protocol, Receive},
};
use tokio::{net::UdpSocket, sync::mpsc::Receiver};
use tracing::{debug, error, info, trace, warn};
use uuid::Uuid;
use crate::http::HttpServer;
use crate::{
StreamSession, audio::OpusAudioFrame, codec::VideoFrame, http::HttpServer,
http_error::HttpError,
};
/// Kill a WHIP publish if it delivers no media (video or audio) this long.
const NO_MEDIA_TIMEOUT: Duration = Duration::from_secs(30);
fn bearer_token(headers: &HeaderMap) -> Option<String> {
headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.map(|s| s.trim().to_string())
}
pub async fn handle_whip_injest_delete(
State(state): State<Arc<HttpServer>>,
ConnectInfo(remote): ConnectInfo<SocketAddr>,
Path(_slug): Path<String>,
headers: HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
let Some(token) = bearer_token(&headers) else {
warn!("Whip: missing bearer token from {}", remote);
return Err(HttpError::Unauthorized);
};
let key = stream_key::Entity::find_by_key(&state.db, &token)
.await?
.ok_or(HttpError::Unauthorized)?;
// The detach task may have already cleaned up the session when ICE
// disconnected (OBS closes the PeerConnection before sending DELETE).
// Deleting an already-gone session is still a successful delete.
if let Some(active_session) =
stream_session::Model::get_active_by_stream_key_id(&state.db, key.id).await?
{
active_session
.into_active_model()
.finish_stream_session(&state.db, Utc::now())
.await?;
}
state.appstate.lock().await.stream_sessions.remove(&key.id);
// Drop the trickle channel too, so PATCHes to a deleted session 404 and
// the lingering detach task's own cleanup can't reach a newer session.
state
.appstate
.lock()
.await
.webrtc_proxy
.trickle_tx
.remove(&key.id);
Ok(StatusCode::OK)
}
/// PATCH /api/whip/{id} — Trickle ICE candidate delivery.
pub async fn handle_whip_injest_patch(
State(state): State<Arc<HttpServer>>,
Path(slug): Path<String>,
headers: HeaderMap,
body: String,
) -> Result<impl IntoResponse, HttpError> {
// The slug is the stream_key_id (set by the POST handler's Location header).
let stream_key_id: i32 = slug.parse().map_err(|e| {
warn!(%slug, "Whip PATCH: bad slug: {:?}", e);
HttpError::NotFound
})?;
info!(stream_key_id, ct = ?headers.get(header::CONTENT_TYPE), "Whip PATCH: trickle candidate");
// Look up the trickle sender for this session.
let trickle_map = state.appstate.lock().await.webrtc_proxy.trickle_tx.clone();
let tx = trickle_map
.get(&stream_key_id)
.map(|e| e.value().1.clone())
.ok_or_else(|| {
warn!(stream_key_id, "Whip PATCH: no trickle channel for session");
HttpError::NotFound
})?;
debug!(stream_key_id, %body, "Whip PATCH: forwarding trickle candidate");
tx.send(body).ok();
Ok(StatusCode::NO_CONTENT)
}
pub async fn handle_whip_injest(
State(state): State<Arc<HttpServer>>,
ConnectInfo(remote): ConnectInfo<SocketAddr>,
headers: HeaderMap,
offer: String,
) -> impl IntoResponse {
let sdp_offer = SdpOffer::from_sdp_string(&offer).unwrap();
) -> axum::response::Response {
let err = |status, msg: &str| -> axum::response::Response {
(
status,
[(header::CONTENT_TYPE, "text/plain")],
msg.to_string(),
)
.into_response()
};
let public_addr = state.appstate.lock().await.webrtc_proxy.public_addr();
let Some(token) = bearer_token(&headers) else {
warn!("Whip: missing bearer token from {}", remote);
return err(StatusCode::UNAUTHORIZED, "missing bearer token");
};
let key = match stream_key::Entity::find_by_key(&state.db, &token).await {
Ok(Some(key)) => key,
Ok(None) => {
warn!("Whip: stream key not found, rejecting {}", remote);
return err(StatusCode::UNAUTHORIZED, "invalid stream key");
}
Err(e) => {
error!("Whip: DB error looking up stream key: {:?}", e);
return err(StatusCode::INTERNAL_SERVER_ERROR, "db error");
}
};
match stream_session::Model::get_active_by_stream_key_id(&state.db, key.id).await {
Ok(Some(_)) => {
warn!(stream_key_id = key.id, label = %key.label, "Whip: stream key already live, rejecting duplicate publish from {}", remote);
return err(StatusCode::CONFLICT, "stream already live");
}
Ok(None) => {}
Err(e) => {
error!("Whip: DB error checking active stream: {:?}", e);
return err(StatusCode::INTERNAL_SERVER_ERROR, "db error");
}
};
info!(stream_key_id = key.id, label = %key.label, "Whip authenticated stream key from {}", remote);
// Parse the SDP offer first so we can detect the codec and
// configure the Rtc before accepting.
let sdp_offer = match SdpOffer::from_sdp_string(&offer) {
Ok(o) => o,
Err(e) => {
error!("Whip: cant parse offer from {}: {:?}", remote, e);
return err(StatusCode::BAD_REQUEST, "invalid SDP offer");
}
};
info!(
"Whip offer from {} — {} media line(s):",
remote,
sdp_offer.media_lines.len()
);
for x in &sdp_offer.media_lines {
info!("{}", x);
info!(" {}", x);
}
// Detect codec from the offer so we can enable matching codecs.
let stream_codec = video_codec_from_sdp_offer(&sdp_offer);
info!(?stream_codec, "detected codec from WHIP offer");
// Build Rtc with ICE-Lite (required by WHIP RFC 9728 §4.1) and
// matching codecs enabled.
// Not using ICE-Lite: full ICE lets the server initiate checks
// when OBS hasn't sent its candidates yet (Trickle ICE without PATCH).
let mut builder = Rtc::builder();
{
let cc = builder.codec_config();
cc.clear();
cc.enable_opus(true);
match &stream_codec {
Some(codec) => {
info!("Whip: enabling {codec:?} codec");
match codec {
crate::StreamCodec::H264 => cc.enable_h264(true),
crate::StreamCodec::H265 => cc.enable_h265(true),
crate::StreamCodec::AV1 => cc.enable_av1(true),
}
}
None => {
warn!("Whip: no video codec detected in offer, enabling H.264 as fallback");
cc.enable_h264(true);
}
}
}
let mut rtc = builder.build(Instant::now());
let candidate = Candidate::host(public_addr, Protocol::Udp).unwrap();
rtc.add_local_candidate(candidate);
info!(%public_addr, "Whip: added local ICE candidate, accepting offer…");
let offer_answe = match rtc.sdp_api().accept_offer(sdp_offer) {
Ok(a) => a,
Err(e) => {
error!("cant accept inject offer: {:?}", e);
return err(StatusCode::BAD_REQUEST, "could not accept offer");
}
};
// OBS sends no a=candidate: lines (disableAutoGathering). Derive a
// remote host candidate from the HTTP source address so the server
// has somewhere to send STUN checks.
if let Ok(c) = Candidate::host(remote, Protocol::Udp) {
info!(%remote, "Whip: no candidates in offer, adding HTTP-derived remote host candidate");
rtc.add_remote_candidate(c);
}
// Extract negotiated codec, PT, and profile from the answer
// so WHEP viewers can use the exact same codec config.
let (negotiated_codec, video_pt, video_profile) = extract_negotiated_codec_info(&offer_answe);
info!(
?negotiated_codec,
video_pt,
?video_profile,
"negotiated codec from WHIP answer"
);
if negotiated_codec.is_none() || video_pt.is_none() {
warn!("Whip: no common video codec negotiated, rejecting");
return err(StatusCode::NOT_ACCEPTABLE, "no common video codec");
}
let answer_sdp = offer_answe
.to_sdp_string()
// Strip a=ice-options:trickle so OBS starts ICE immediately.
.replace("a=ice-options:trickle\r\n", "")
.replace("a=ice-options:trickle\n", "");
// Fix up the answer for libdatachannel (OBS WHIP):
// 1. Strip a=group:BUNDLE — OBS doesn't negotiate it.
// 2. Add the host candidate to the video m= line.
let answer_sdp = answer_sdp
.replace("a=group:BUNDLE 0 1\r\n", "")
.replace("a=group:BUNDLE 0 1\n", "");
let answer_sdp =
if let Some(cand_line) = answer_sdp.lines().find(|l| l.starts_with("a=candidate:")) {
// Insert the candidate line after the video m= line.
let cand_replacement = format!(
"m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 {}\r\n{}\r\n",
public_addr.ip(),
cand_line
);
answer_sdp.replace(
"m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 0.0.0.0\r\n",
&cand_replacement,
)
} else {
answer_sdp
};
info!("Serving Whip SDP answer to {}:\n{}", remote, answer_sdp);
let ufrag = answer_sdp
.lines()
.find(|l| l.starts_with("a=ice-ufrag:"))
.and_then(|l| l.strip_prefix("a=ice-ufrag:"))
.map(|s| s.trim().to_string());
let Some(ufrag) = ufrag else {
error!("Whip: answer has no a=ice-ufrag");
return err(StatusCode::INTERNAL_SERVER_ERROR, "no ice-ufrag in answer");
};
let ice_pwd = answer_sdp
.lines()
.find(|l| l.starts_with("a=ice-pwd:"))
.and_then(|l| l.strip_prefix("a=ice-pwd:"))
.map(|s| s.trim().to_string());
info!(
?ufrag,
?ice_pwd,
"Whip: registering ICE credentials with proxy"
);
let (socket, rx) = state.appstate.lock().await.webrtc_proxy.add_client(ufrag);
let stream_sessions = state.appstate.lock().await.stream_sessions.clone();
// Unique per publish: cleanup only removes the session this task created,
// never a newer one that re-published on the same stream key.
let session_id = Uuid::new_v4();
let (mut video_tx, video_rx) = broadcast::<Arc<VideoFrame>>(32);
let (mut audio_tx, audio_rx) = broadcast::<Arc<OpusAudioFrame>>(32);
// Never block the ingest loop on slow/missing viewers: overwrite the
// oldest frame instead (same as the RTMP path; the viewer re-waits for a
// keyframe on Overflowed). The undrained rx below must not stall sends.
video_tx.set_overflow(true);
audio_tx.set_overflow(true);
let user = users::Entity::find_by_id(key.user_id)
.one(&state.db)
.await
.unwrap()
.unwrap();
stream_sessions.insert(
key.id,
StreamSession {
stream_key_id: key.id,
stream_key_label: key.label,
stream_key_user: user.username,
custom_id: key.custom_id,
is_unlisted: key.is_unlisted,
password: key.password,
frame_channel: video_tx,
audio_channel: audio_tx,
codec: negotiated_codec,
session_id,
started_at: Utc::now(),
active_clients: 0.into(),
video_pt,
video_profile_level_id: video_profile,
},
);
info!(
stream_key_id = key.id,
"Whip: StreamSession inserted, spawning detach task"
);
// Trickle-ICE channel: OBS can send candidates via PATCH after the
// initial offer. We forward them to the Rtc task for add_remote_candidate.
let (trickle_tx, trickle_rx) = tokio::sync::mpsc::unbounded_channel();
let trickle_map = state.appstate.lock().await.webrtc_proxy.trickle_tx.clone();
trickle_map.insert(key.id, (session_id, trickle_tx));
let db = state.db.clone();
tokio::spawn(async move {
detach_inject_rtc(
stream_sessions,
key.id,
db,
socket,
rx,
trickle_rx,
rtc,
public_addr,
video_rx,
audio_rx,
session_id,
)
.await;
// Drop only our own trickle channel; a newer session on the same key
// must keep its own.
trickle_map.remove_if(&key.id, |_, (sid, _)| *sid == session_id);
});
stream_session::Model::create_stream_session(&state.db, key.id, Local::now().into())
.await
.ok();
let location = format!("/api/whip/{}", key.id);
axum::response::Response::builder()
.status(StatusCode::CREATED)
.header(header::CONTENT_TYPE, "application/sdp")
.header(header::LOCATION, &location)
.body(axum::body::Body::from(answer_sdp))
.unwrap()
}
async fn detach_inject_rtc(
sessions_ref: Arc<DashMap<i32, StreamSession>>,
stream_key_id: i32,
db: DatabaseConnection,
socket: Arc<UdpSocket>,
mut rx: Receiver<(Bytes, SocketAddr)>,
mut trickle_rx: tokio::sync::mpsc::UnboundedReceiver<String>,
mut rtc: Rtc,
local_addr: SocketAddr,
video_rx: async_broadcast::Receiver<Arc<VideoFrame>>,
audio_rx: async_broadcast::Receiver<Arc<OpusAudioFrame>>,
session_id: Uuid,
) {
let cleanup = || {
let sessions_ref = sessions_ref.clone();
let db = db.clone();
async move {
// Only clean up the session this task created. If the user has
// already opened another stream on the same key (e.g. DELETE then
// immediate republish), that newer session must not be touched.
let is_ours = sessions_ref
.get(&stream_key_id)
.is_some_and(|s| s.session_id == session_id);
if !is_ours {
debug!(
stream_key_id,
"skipping cleanup: session replaced by a newer stream on this key"
);
return;
}
debug!("Cleaning up {:?}", &stream_key_id);
sessions_ref.remove(&stream_key_id);
if let Ok(Some(s)) =
stream_session::Model::get_active_by_stream_key_id(&db, stream_key_id).await
{
s.into_active_model()
.finish_stream_session(&db, Local::now().into())
.await
.ok();
}
}
};
let mut video_mid: Option<Mid> = None;
let mut audio_mid: Option<Mid> = None;
let mut video_tx: Option<async_broadcast::Sender<Arc<VideoFrame>>> = None;
let mut audio_tx: Option<async_broadcast::Sender<Arc<OpusAudioFrame>>> = None;
let mut disconnect_timer: Option<Instant> = None;
loop {
// Blankly using them so they don't drop (like RTMP).
let _ = video_rx.is_closed();
let _ = audio_rx.is_closed();
let deadline = loop {
match rtc.poll_output() {
Ok(Output::Timeout(t)) => break t,
Ok(Output::Transmit(t)) => {
// The keep alive loop, by default is every 1 second.
trace!(
"Whip TX: {} bytes → {}:{}",
t.contents.len(),
t.destination.ip(),
t.destination.port()
);
if let Err(e) = socket.send_to(&t.contents, t.destination).await {
warn!("Whip UDP send error: {:?}", e);
cleanup().await;
return;
}
// 30 Sec time out if disconnected, then we clean up and disconnect.
if let Some(instant_since_disconnet) = disconnect_timer
&& Instant::now()
.duration_since(instant_since_disconnet)
.as_secs()
> NO_MEDIA_TIMEOUT.as_secs()
{
info!(
"WHIP connection on stream_key_id: {:?}, has been disconnected for {:?} seconds. Cleaning up and destroying the connection,",
stream_key_id,
NO_MEDIA_TIMEOUT.as_secs()
);
cleanup().await;
rtc.disconnect();
return;
}
}
Ok(Output::Event(e)) => match e {
Event::MediaAdded(ma) => {
info!(stream_key_id, kind = ?ma.kind, mid = ?ma.mid, "Whip MediaAdded");
if ma.kind == MediaKind::Video {
video_mid = Some(ma.mid);
}
if ma.kind == MediaKind::Audio {
audio_mid = Some(ma.mid);
}
}
Event::MediaData(md) => {
// str0m depacketizes RTP into full codec frames (no decode).
// H.264/H.265 → Annex-B, AV1 → OBU, Opus → raw Opus packets.
// Same format as our RTMP codec parsers produce — push directly.
if Some(md.mid) == video_mid {
let ts_ms = (md.time.as_seconds() * 1000.0) as u32;
let frame = VideoFrame {
data: Bytes::copy_from_slice(&md.data),
is_keyframe: md.is_keyframe(),
timestamp_ms: ts_ms,
};
if let Some(tx) = &video_tx {
tx.broadcast(Arc::new(frame)).await.ok();
}
}
if Some(md.mid) == audio_mid {
let ts_ms = (md.time.as_seconds() * 1000.0) as u32;
let frame = OpusAudioFrame {
data: Bytes::copy_from_slice(&md.data),
timestamp_ms: ts_ms,
};
if let Some(tx) = &audio_tx {
tx.broadcast(Arc::new(frame)).await.ok();
}
}
}
Event::IceConnectionStateChange(state) => {
info!(stream_key_id, ?state, "Whip ICE state change");
match state {
str0m::IceConnectionState::Disconnected => {
info!(
"Whip ICE disconnected... (State changed to Disconnected for {:?}) ((This is usually due to network jitter))",
&stream_key_id
);
disconnect_timer = Some(Instant::now());
}
str0m::IceConnectionState::Connected
| str0m::IceConnectionState::Completed => {
disconnect_timer = None;
}
_ => {}
}
}
Event::Connected => {
info!(
stream_key_id,
"Whip DTLS+ICE connected, wiring broadcast channels"
);
if let Some(session) = sessions_ref.get(&stream_key_id) {
video_tx = Some(session.frame_channel.clone());
audio_tx = Some(session.audio_channel.clone());
}
}
_ => {}
},
Err(e) => {
error!("Whip poll_output error (closing): {:?}", e);
cleanup().await;
return;
}
}
};
let sleep = tokio::time::sleep_until(deadline.max(Instant::now()).into());
tokio::select! {
_ = sleep => {
if let Err(e) = rtc.handle_input(Input::Timeout(Instant::now())) {
error!(stream_key_id, "Whip handle_input(Timeout) error: {:?}", e);
cleanup().await;
return;
}
}
// Trickle-ICE candidates from PATCH /api/whip/{id}
Some(candidate_line) = trickle_rx.recv() => {
info!(stream_key_id, %candidate_line, "Whip: received trickle candidate");
// Parse the candidate string (without "a=candidate:" prefix if present).
let cand_str = candidate_line
.strip_prefix("a=candidate:")
.unwrap_or(&candidate_line);
match str0m::Candidate::from_sdp_string(cand_str) {
Ok(c) => {
info!(stream_key_id, "Whip: adding remote candidate");
rtc.add_remote_candidate(c);
}
Err(e) => {
warn!(stream_key_id, %cand_str, "Whip: bad trickle candidate: {:?}", e);
}
}
}
// No UDP input at all for 30s: closing connection.
// (ICE keepalives still arrive when only the encoder is stalled,
// so this fires on a dead peer, not a paused one.)
result = tokio::time::timeout(NO_MEDIA_TIMEOUT, rx.recv()) => {
let Ok(Some((data, from))) = result else {
if result.is_err() {
warn!(stream_key_id, "Whip: no input for {NO_MEDIA_TIMEOUT:?}, closing connection");
rtc.disconnect();
} else {
info!(stream_key_id, "Whip proxy channel closed, cleaning up");
}
cleanup().await;
return;
};
trace!(
stream_key_id,
len = data.len(),
%from,
"Whip RX: {} bytes",
data.len()
);
if let Ok(contents) = (&data[..]).try_into()
&& let Err(e) = rtc.handle_input(Input::Receive(
Instant::now(),
Receive {
proto: Protocol::Udp,
source: from,
destination: local_addr,
contents,
},
)) {
error!("Whip handle_input(Receive) error: {:?}", e);
cleanup().await;
return;
}
}
}
}
}
/// Extract the negotiated video codec, PT, and profile_level_id from an SDP answer.
///
/// Walks the answer's media lines, finds the first video m-line with
/// negotiated rtp_params, and returns the codec + PT + H.264 profile.
pub fn extract_negotiated_codec_info(
answer: &str0m::change::SdpAnswer,
) -> (Option<crate::StreamCodec>, Option<u8>, Option<u32>) {
for line in answer.media_lines.iter() {
for p in line.rtp_params() {
if p.spec().codec.is_video() {
return (
crate::StreamCodec::from_str0m(p.spec().codec),
Some(*p.pt()),
p.spec().format.profile_level_id,
);
}
}
}
(None, None, None)
}
/// Extract the video codec from an SDP offer's media lines.
///
/// Walks every m= line's `rtp_params()`; returns the first video codec found.
pub fn video_codec_from_sdp_offer(
sdp_offer: &str0m::change::SdpOffer,
) -> Option<crate::StreamCodec> {
for line in sdp_offer.media_lines.iter() {
for p in line.rtp_params() {
if p.spec().codec.is_video() {
return crate::StreamCodec::from_str0m(p.spec().codec);
}
}
}
None
}
+99 -63
View File
@@ -7,16 +7,12 @@ use std::{
use bytes::Bytes;
use dashmap::DashMap;
use str0m::net::DatagramRecv;
use tokio::{
net::UdpSocket,
sync::mpsc::{self, Receiver},
};
use tracing::{debug, error, info, warn};
pub struct WebRtcProxyConfig {
pub proxy_port: i32,
}
use tracing::{debug, error, info, trace, warn};
use uuid::Uuid;
#[derive(Clone)]
pub struct WebrtcProxy {
@@ -24,28 +20,47 @@ pub struct WebrtcProxy {
clients_addr: Arc<DashMap<SocketAddr, tokio::sync::mpsc::Sender<(Bytes, SocketAddr)>>>,
socket: Arc<UdpSocket>,
public_addr: SocketAddr,
/// Trickle-ICE candidate channels for WHIP ingest.
/// Keyed by stream_key_id; each entry is tagged with the owning session's
/// id so cleanup can remove only its own entry and never a newer session
/// that re-published on the same key.
pub trickle_tx: Arc<DashMap<i32, (Uuid, tokio::sync::mpsc::UnboundedSender<String>)>>,
}
const STUN_MAGIC: u32 = 0x2112A442;
impl WebrtcProxy {
pub async fn new(config: WebRtcProxyConfig) -> Result<Self, Box<dyn Error>> {
let sock = UdpSocket::bind(format!("0.0.0.0:{}", config.proxy_port)).await?;
pub async fn new(proxy_port: i32) -> Result<Self, Box<dyn Error>> {
let sock = UdpSocket::bind(format!("0.0.0.0:{}", proxy_port)).await?;
let port = sock.local_addr()?.port();
let public_ip = match env::var("PUBLIC_DOMAIN") {
Ok(domain) => {
let public_ip = match env::var("PUBLIC_DOMAIN")
.ok()
.filter(|s| !s.trim().is_empty())
{
Some(domain) => {
let ip = resolve_domain(&domain).await?;
info!(%domain, %ip, "resolved PUBLIC_DOMAIN for WebRTC candidates");
ip
}
Err(_) => {
None => {
if cfg!(debug_assertions) {
// For testing
info!(
"using ip 0.0.0.0 for WebRTC candidates (because we are in a debug build)"
);
IpAddr::from([127, 0, 0, 1])
// For testing — advertise a real interface IP. Loopback is
// unreachable from clients whose ICE stack pins its UDP
// sockets to a specific interface (OBS sets
// IP_UNICAST_IF), which makes checks to 127.0.0.1 vanish.
match default_iface_ipv4() {
Some(ip) => {
info!(%ip, "using default interface IP for WebRTC candidates (debug build)");
ip
}
None => {
warn!(
"no non-loopback IPv4 interface found, falling back to 127.0.0.1"
);
IpAddr::from([127, 0, 0, 1])
}
}
} else {
let ip = stun_public_ip().await?;
info!(%ip, "discovered public IP via STUN for WebRTC candidates");
@@ -61,6 +76,7 @@ impl WebrtcProxy {
clients_ufrag: Arc::new(DashMap::new()),
clients_addr: Arc::new(DashMap::new()),
public_addr,
trickle_tx: Arc::new(DashMap::new()),
})
}
pub async fn run(self) {
@@ -81,6 +97,12 @@ impl WebrtcProxy {
// By addr
if let Some(tx) = by_addr.get(&from) {
trace!(
"proxy: routing {} bytes by addr {}:{} → channel",
b,
from.ip(),
from.port()
);
match tx.try_send((data, from)) {
Ok(_) => continue,
Err(e) => {
@@ -97,18 +119,30 @@ impl WebrtcProxy {
};
};
let Some(ufrag) = self::WebrtcProxy::ufrag(&data) else {
debug!("huh, packet isnt stun or added as client.");
let Some((part1, part2)) = self::WebrtcProxy::ufrag_pair(&data) else {
trace!("huh, packet isnt stun or added as client.");
continue;
};
let Some((_, tx)) = by_ufrag.remove(&ufrag) else {
// warn!("STUN packet ({}), isnt registored", ufrag);
// Try both parts of the STUN username — the first packet
// might be a response to OUR STUN request (remote:local)
// or an incoming request from the remote peer (local:remote).
let part2_lookup = part2.clone();
let entry = by_ufrag
.remove(&part1)
.or_else(|| part2_lookup.and_then(|p2| by_ufrag.remove(&p2)));
let Some((_, tx)) = entry else {
warn!("STUN packet ({}/{:?}), isnt registored", part1, part2);
continue;
};
by_addr.insert(from, tx.clone());
debug!("got ufrag {}", ufrag);
info!(
"proxy: STUN match → promoted {} → ufrag={} (match was {}/{})",
from,
part1,
part1,
part2.as_deref().unwrap_or("-")
);
debug!("sending data");
if let Err(e) = tx.try_send((data, from)) {
match e {
@@ -129,14 +163,10 @@ impl WebrtcProxy {
self.clients_ufrag.insert(ufrag, tx);
(self.socket.clone(), rx)
}
pub fn local_addr(&self) -> SocketAddr {
self.socket.local_addr().unwrap()
}
pub fn public_addr(&self) -> SocketAddr {
self.public_addr
}
pub fn ufrag(b: &Bytes) -> Option<String> {
pub fn ufrag_pair(b: &Bytes) -> Option<(String, Option<String>)> {
if b.len() <= 20 {
return None;
}
@@ -144,23 +174,39 @@ impl WebrtcProxy {
if magic != STUN_MAGIC {
return None;
}
// attribies start at 20
let mut pos = 20usize;
while (pos + 4) <= b.len() {
let attr_type: u16 = u16::from_be_bytes(b[pos..pos + 2].try_into().ok()?);
let attr_len: u16 = u16::from_be_bytes(b[pos + 2..pos + 4].try_into().ok()?);
pos = pos + 4;
if attr_type == 0x0006 {
let value = std::str::from_utf8(b[pos..pos + (attr_len as usize)].try_into().ok()?);
let local = value.unwrap().split(":").next();
return Some(local.unwrap().to_string());
}
pos += (attr_len as usize + 3) & !3;
}
None
let value = stun_attributes(b).find(|(t, _)| *t == 0x0006)?.1;
let value = std::str::from_utf8(value).ok()?;
let mut parts = value.split(':');
Some((
parts.next()?.to_string(),
parts.next().map(|s| s.to_string()),
))
}
}
/// Iterate `(attr_type, attr_value)` pairs over a STUN message's attributes
/// (RFC 5389 §15: header is 20 bytes, each attribute padded to a 4-byte boundary).
fn stun_attributes(data: &[u8]) -> impl Iterator<Item = (u16, &[u8])> {
let mut pos = 20usize;
std::iter::from_fn(move || {
let attr_type = u16::from_be_bytes(data.get(pos..pos + 2)?.try_into().ok()?);
let attr_len = u16::from_be_bytes(data.get(pos + 2..pos + 4)?.try_into().ok()?) as usize;
pos += 4;
let value = data.get(pos..pos + attr_len)?;
pos += (attr_len + 3) & !3;
Some((attr_type, value))
})
}
/// IP of the interface holding the default route, via a UDP connect() trick:
/// connect() only does a route lookup (no packets sent), so the kernel binds
/// the source address the OS would use for outbound traffic.
fn default_iface_ipv4() -> Option<IpAddr> {
let sock = std::net::UdpSocket::bind("0.0.0.0:0").ok()?;
sock.connect("8.8.8.8:9").ok()?;
sock.local_addr().ok().map(|a| a.ip())
}
async fn resolve_domain(domain: &str) -> Result<std::net::IpAddr, Box<dyn Error>> {
let addr = tokio::net::lookup_host(format!("{}:0", domain))
.await?
@@ -203,27 +249,17 @@ fn parse_xor_mapped_address(data: &[u8]) -> Option<std::net::IpAddr> {
return None;
}
let mut pos = 20usize;
while pos + 4 <= data.len() {
let attr_type = u16::from_be_bytes(data[pos..pos + 2].try_into().ok()?);
let attr_len = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?) as usize;
pos += 4;
if pos + attr_len > data.len() {
break;
}
if attr_type == 0x0020 && attr_len >= 8 {
// byte 0: reserved, byte 1: family (0x01=IPv4, 0x02=IPv6)
let family = data[pos + 1];
let x_port = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?);
let _ = x_port ^ (STUN_MAGIC >> 16) as u16; // port (unused here)
if family == 0x01 {
let x_addr = u32::from_be_bytes(data[pos + 4..pos + 8].try_into().ok()?);
let addr = x_addr ^ STUN_MAGIC;
return Some(std::net::IpAddr::V4(std::net::Ipv4Addr::from(addr)));
}
}
pos += (attr_len + 3) & !3;
let value = stun_attributes(data).find(|(t, _)| *t == 0x0020)?.1;
if value.len() < 8 {
return None;
}
// byte 0: reserved, byte 1: family (0x01=IPv4, 0x02=IPv6)
if value[1] == 0x01 {
let x_addr = u32::from_be_bytes(value[4..8].try_into().ok()?);
Some(std::net::IpAddr::V4(std::net::Ipv4Addr::from(
x_addr ^ magic,
)))
} else {
None
}
None
}
+118
View File
@@ -0,0 +1,118 @@
//! Temporary probe: replicate handle_whip_injest's Rtc setup, feed it a
//! realistic OBS/libdatachannel WHIP offer, print the answer SDP and check
//! whether the string-fixups in webrtc_ingest.rs actually match.
use std::{net::SocketAddr, time::Instant};
use str0m::{Candidate, Rtc, change::SdpOffer, net::Protocol};
fn obs_like_offer() -> String {
let mut s = String::new();
s.push_str("v=0\r\n");
s.push_str("o=- 4527835755568137757 2 IN IP4 127.0.0.1\r\n");
s.push_str("s=-\r\n");
s.push_str("t=0 0\r\n");
s.push_str("a=group:BUNDLE 0 1\r\n");
s.push_str("a=msid-semantic: WMS *\r\n");
s.push_str("m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n");
s.push_str("c=IN IP4 0.0.0.0\r\n");
s.push_str("a=rtcp:9 IN IP4 0.0.0.0\r\n");
s.push_str("a=ice-ufrag:obs_ufrag_audio\r\n");
s.push_str("a=ice-pwd:obs_pwd_audio\r\n");
s.push_str("a=ice-options:trickle\r\n");
s.push_str("a=fingerprint:sha-256 5A:6B:7C:8D:9E:AF:B0:C1:D2:E3:F4:05:16:27:38:49:5A:6B:7C:8D:9E:AF:B0:C1:D2:E3:F4:05:16:27:38:49:5A\r\n");
s.push_str("a=setup:actpass\r\n");
s.push_str("a=mid:0\r\n");
s.push_str("a=sendrecv\r\n");
s.push_str("a=rtcp-mux\r\n");
s.push_str("a=rtpmap:111 opus/48000/2\r\n");
s.push_str("a=rtcp-fb:111 transport-cc\r\n");
s.push_str("a=fmtp:111 minptime=10;useinbandfec=1\r\n");
s.push_str("m=video 9 UDP/TLS/RTP/SAVPF 96\r\n");
s.push_str("c=IN IP4 0.0.0.0\r\n");
s.push_str("a=rtcp:9 IN IP4 0.0.0.0\r\n");
s.push_str("a=ice-ufrag:obs_ufrag_video\r\n");
s.push_str("a=ice-pwd:obs_pwd_video\r\n");
s.push_str("a=ice-options:trickle\r\n");
s.push_str("a=fingerprint:sha-256 5A:6B:7C:8D:9E:AF:B0:C1:D2:E3:F4:05:16:27:38:49:5A:6B:7C:8D:9E:AF:B0:C1:D2:E3:F4:05:16:27:38:49:5A\r\n");
s.push_str("a=setup:actpass\r\n");
s.push_str("a=mid:1\r\n");
s.push_str("a=sendrecv\r\n");
s.push_str("a=rtcp-mux\r\n");
s.push_str("a=rtpmap:96 H264/90000\r\n");
s.push_str("a=rtcp-fb:96 nack\r\n");
s.push_str("a=rtcp-fb:96 nack pli\r\n");
s.push_str("a=rtcp-fb:96 transport-cc\r\n");
s.push_str(
"a=fmtp:96 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f\r\n",
);
s
}
#[test]
fn probe_whip_answer() {
let offer_sdp = obs_like_offer();
let sdp_offer = SdpOffer::from_sdp_string(&offer_sdp).expect("parse offer");
// Mirror handle_whip_injest's builder setup.
let mut builder = Rtc::builder();
{
let cc = builder.codec_config();
cc.clear();
cc.enable_opus(true);
cc.enable_h264(true);
}
let mut rtc = builder.build(Instant::now());
let public_addr: SocketAddr = "203.0.113.7:6969".parse().unwrap();
let candidate = Candidate::host(public_addr, Protocol::Udp).unwrap();
rtc.add_local_candidate(candidate);
let answer = rtc.sdp_api().accept_offer(sdp_offer).expect("accept offer");
let answer_sdp = answer.to_sdp_string();
println!("=== RAW ANSWER ===");
println!("{answer_sdp}");
println!("=== END RAW ANSWER ===");
// Now replicate the fixups from webrtc_ingest.rs verbatim.
let answer_sdp = answer_sdp
.replace("a=ice-options:trickle\r\n", "")
.replace("a=ice-options:trickle\n", "");
let answer_sdp = answer_sdp
.replace("a=group:BUNDLE 0 1\r\n", "")
.replace("a=group:BUNDLE 0 1\n", "");
let answer_sdp =
if let Some(cand_line) = answer_sdp.lines().find(|l| l.starts_with("a=candidate:")) {
let cand_replacement = format!(
"m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 127.0.0.1\r\n{}\r\n",
cand_line
);
answer_sdp.replace(
"m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 0.0.0.0\r\n",
&cand_replacement,
)
} else {
answer_sdp
};
println!("=== FIXED ANSWER ===");
println!("{answer_sdp}");
println!("=== END FIXED ANSWER ===");
// Diagnostics
let video_mline = answer_sdp
.lines()
.find(|l| l.starts_with("m=video"))
.unwrap();
println!("video m-line after fixup: {video_mline}");
println!(
"has a=candidate after fixup: {}",
answer_sdp.lines().any(|l| l.starts_with("a=candidate:"))
);
println!(
"a=candidate lines: {:?}",
answer_sdp
.lines()
.filter(|l| l.starts_with("a=candidate:"))
.collect::<Vec<_>>()
);
}
Generated
+28 -7
View File
@@ -2,11 +2,11 @@
"nodes": {
"crane": {
"locked": {
"lastModified": 1780532242,
"narHash": "sha256-D+BsdpxmtUwtqGoY0IXPhHgTlmqgcZKCEo1oMyn7ep0=",
"lastModified": 1788465171,
"narHash": "sha256-Y1/TTVXjYXGF068IThQH9fPSZ0SIE74PABlUxnWTUH0=",
"owner": "ipetkov",
"repo": "crane",
"rev": "59a82a1222dd3b2080b5cc52a1a2e8d5f1b77f37",
"rev": "eb35abda9f232cc6610b1d1e3200d15c49b7ac54",
"type": "github"
},
"original": {
@@ -35,11 +35,11 @@
},
"nixpkgs": {
"locked": {
"lastModified": 1781454065,
"narHash": "sha256-d2xfDjnfRuf/xYGdu9VVRHiav/2w5hDL/5cw2TuVAXw=",
"lastModified": 1789012029,
"narHash": "sha256-1CBkBf+Nhggykzlx0jvXrj5rk20btl+tK8Fa8Ml/BL4=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "9eac87a12312b8f60dd52e1c6e1a265f6fc7f5fc",
"rev": "d5dfd8e6716dde34398bc14bc87c10dece9c8c68",
"type": "github"
},
"original": {
@@ -53,7 +53,28 @@
"inputs": {
"crane": "crane",
"flake-utils": "flake-utils",
"nixpkgs": "nixpkgs"
"nixpkgs": "nixpkgs",
"rust-overlay": "rust-overlay"
}
},
"rust-overlay": {
"inputs": {
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1789024335,
"narHash": "sha256-kCy/MVLRIr95DJ4vspVzWj+kO/x+JuwSHmYZCvShg8w=",
"owner": "oxalica",
"repo": "rust-overlay",
"rev": "577bb1e1fc5af0713169176c5c76622c21fa3ec0",
"type": "github"
},
"original": {
"owner": "oxalica",
"repo": "rust-overlay",
"type": "github"
}
},
"systems": {
+23 -4
View File
@@ -7,6 +7,12 @@
crane.url = "github:ipetkov/crane";
flake-utils.url = "github:numtide/flake-utils";
rust-overlay = {
url = "github:oxalica/rust-overlay";
inputs.nixpkgs.follows = "nixpkgs";
};
};
outputs =
@@ -16,14 +22,27 @@
crane,
flake-utils,
...
}:
}@inputs:
flake-utils.lib.eachDefaultSystem (
system:
let
pkgs = nixpkgs.legacyPackages.${system};
pkgs = import inputs.nixpkgs {
inherit system;
overlays = [ (import inputs.rust-overlay) ];
};
inherit (pkgs) lib;
craneLib = crane.mkLib pkgs;
craneLib = (inputs.crane.mkLib pkgs).overrideToolchain (
p:
p.rust-bin.nightly.latest.default.override {
extensions = [
"rustc-codegen-cranelift-preview"
"rust-analyzer"
"rust-src"
];
}
);
# Common arguments can be set here to avoid repeating them later
# Note: changes here will rebuild all dependency crates
@@ -56,7 +75,7 @@
commonArgs
// {
pname = "rtmp-to-whip";
version = "0.1.0";
version = "0.4.0";
cargoArtifacts = craneLib.buildDepsOnly commonArgs;
cargoExtraArgs = "-p server";
src = fileSetForCrate ./crates/server;