Answer relayed WebRTC offers from the gateway's own go2rtc
Adds src/uplink/webrtc.rs: on a webrtc_offer from the cloud, POST the offer to go2rtc on 127.0.0.1 and send the answer back as webrtc_answer, or the reason as webrtc_error. No credential is involved, because the request goes to loopback and go2rtc exempts loopback from its API password. That is measured, and it is also why a probe from the gateway cannot be used to check the password is working - test from another host for that. Spawned rather than awaited in the select! loop: go2rtc gathers ICE candidates before answering, which includes a STUN round trip, so awaiting inline would stall telemetry and heartbeats for a whole farm while one person opened one camera. It rides the outbound queue added earlier for exactly this kind of out-of-band reply. The bind address is turned into a loopback URL by taking its port and refusing anything it cannot read a port out of - guessing 1984 would produce a gateway that silently signals to the wrong place when someone has moved it. Stream names are validated rather than percent-encoded, since a real one is a printer id and anything else is a bug or an attempt at one. camera.rs now states why there are no ice_servers and no TURN credentials on this side: the viewer's browser is pinned to relay-only, so it allocates the relay and go2rtc just sends to it. Only one side of a session needs one. That asymmetry is load-bearing, not tidy - go2rtc reads its config once at startup, so rotating short-lived credentials here would mean restarting it and dropping every viewer on a timer.
This commit is contained in:
+20
-8
@@ -256,15 +256,27 @@ pub fn render_config(api_listen: &str, sources: &[CameraSource], auth: &ApiAuth)
|
||||
// listener above. An empty `listen` is how go2rtc is told not to bind.
|
||||
out.push_str("rtsp:\n listen: \"\"\n");
|
||||
|
||||
// `webrtc:` is deliberately left alone, so go2rtc keeps its default
|
||||
// media port. It is not a second open stream server - a WebRTC session
|
||||
// only exists after an SDP offer through the API - and the browser needs
|
||||
// those candidates to receive anything.
|
||||
// `webrtc:` is deliberately left alone, and with TURN in the picture that
|
||||
// is now a decision worth stating rather than an omission.
|
||||
//
|
||||
// The API listener itself is unauthenticated, which is a real limit
|
||||
// rather than an oversight: the browser reaches it directly, there is no
|
||||
// relay, and the frontend is handed only a URL and a stream name. So
|
||||
// anyone who can reach the gateway's LAN can watch. See PLAN.md.
|
||||
// go2rtc keeps its default media port and its default candidate
|
||||
// gathering, which produces a host candidate and - measured against
|
||||
// 1.9.14 - a server-reflexive one it discovers by STUN on its own. That
|
||||
// is everything this end needs. No `ice_servers` here, and in particular
|
||||
// **no TURN credentials on the gateway**: the viewer's browser is pinned
|
||||
// to relay-only, so the browser allocates the relay and go2rtc simply
|
||||
// sends to it from its own candidate. Only one side of a WebRTC session
|
||||
// has to hold a relay for the media to traverse one.
|
||||
//
|
||||
// That asymmetry is what makes this tractable at all, because go2rtc
|
||||
// reads its config exactly once at startup: TURN credentials are
|
||||
// short-lived, so rotating them here would mean restarting go2rtc and
|
||||
// dropping every viewer on a timer. Keeping them browser-side moves the
|
||||
// rotation to a request that is already per-viewer.
|
||||
//
|
||||
// This is not a second open stream server either way: a WebRTC session
|
||||
// exists only after an SDP offer through the API, and that API now
|
||||
// requires a password.
|
||||
out
|
||||
}
|
||||
|
||||
|
||||
@@ -27,6 +27,7 @@
|
||||
//! buffering.
|
||||
|
||||
mod download;
|
||||
mod webrtc;
|
||||
mod protocol;
|
||||
|
||||
pub use protocol::{ClientMessage, PrinterCommand, PrinterTelemetry, ServerMessage};
|
||||
@@ -274,6 +275,21 @@ async fn serve(
|
||||
tracing::error!(%message, "cloud rejected this gateway's handshake - check CONTINUUM_GATEWAY_TOKEN");
|
||||
return Ok(());
|
||||
}
|
||||
// Spawned, never awaited inline: go2rtc gathers
|
||||
// ICE candidates before answering, which
|
||||
// includes a STUN round trip, so this takes
|
||||
// seconds. Awaiting it here would stall
|
||||
// telemetry and heartbeats for the whole farm
|
||||
// while one person opened one camera.
|
||||
Ok(ServerMessage::WebrtcOffer { session_id, camera_stream, sdp }) => {
|
||||
spawn_webrtc_negotiation(
|
||||
config.go2rtc_api_listen.clone(),
|
||||
session_id,
|
||||
camera_stream,
|
||||
sdp,
|
||||
outbound.clone(),
|
||||
);
|
||||
}
|
||||
Ok(other) => info!(?other, "received from cloud"),
|
||||
// Loudly. This used to be an `if let Ok(..)` with
|
||||
// no else, which is the same silent-drop bug that
|
||||
@@ -424,6 +440,35 @@ fn spawn_long_command(
|
||||
});
|
||||
}
|
||||
|
||||
/// Answers one WebRTC offer, off this connection's task.
|
||||
///
|
||||
/// Always answers, success or failure: the cloud is holding a browser's
|
||||
/// request open waiting for this, and a silent failure there is a viewer
|
||||
/// watching a video element that never loads until something times out.
|
||||
fn spawn_webrtc_negotiation(
|
||||
api_listen: String,
|
||||
session_id: String,
|
||||
camera_stream: String,
|
||||
offer_sdp: String,
|
||||
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
let reply = match webrtc::negotiate(&api_listen, &camera_stream, &offer_sdp).await {
|
||||
Ok(sdp) => ClientMessage::WebrtcAnswer { session_id, sdp },
|
||||
Err(err) => {
|
||||
// Logged as well as returned. The cloud gets the reason, but
|
||||
// this is the copy that survives a gateway whose uplink is
|
||||
// about to drop.
|
||||
warn!(%camera_stream, %err, "could not negotiate a camera stream");
|
||||
ClientMessage::WebrtcError { session_id, message: err.to_string() }
|
||||
}
|
||||
};
|
||||
if outbound.send(reply).await.is_err() {
|
||||
warn!(%camera_stream, "uplink is shutting down; the viewer gets no answer");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Finds the printer by our own id and runs the command. The error is a
|
||||
/// `String` rather than a typed error because its only destination is a
|
||||
/// `CommandResult` going back to the cloud as text.
|
||||
|
||||
@@ -47,6 +47,15 @@ pub enum ClientMessage {
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
error: Option<String>,
|
||||
},
|
||||
/// go2rtc's answer SDP for a `WebrtcOffer`, passed through verbatim.
|
||||
WebrtcAnswer { session_id: String, sdp: String },
|
||||
/// No answer was obtainable: no such camera here, go2rtc not running, or
|
||||
/// it refused the offer.
|
||||
///
|
||||
/// Its own variant rather than an empty answer, because the cloud is
|
||||
/// waiting on this and a viewer otherwise watches a video element that
|
||||
/// never loads with nothing anywhere saying why.
|
||||
WebrtcError { session_id: String, message: String },
|
||||
}
|
||||
|
||||
/// A flattened snapshot of one printer, in the terms an operator console
|
||||
@@ -149,6 +158,21 @@ pub enum ServerMessage {
|
||||
/// Do something to one printer. `printer_id` is our own id from
|
||||
/// printers.toml.
|
||||
Command { command_id: String, printer_id: String, command: PrinterCommand },
|
||||
/// Relay a browser's WebRTC offer to this gateway's go2rtc and answer
|
||||
/// with the SDP it returns. **Signalling only - no video on this
|
||||
/// socket.**
|
||||
///
|
||||
/// Media goes browser-to-go2rtc directly over ICE, which is the whole
|
||||
/// reason only the signalling is relayed: on a workshop LAN the video
|
||||
/// never leaves the building, and it costs the cloud no bandwidth at
|
||||
/// all. go2rtc's answer carries a server-reflexive candidate alongside
|
||||
/// its host ones, so even a remote viewer can often reach the gateway
|
||||
/// without a relay.
|
||||
///
|
||||
/// The browser cannot do this part itself. go2rtc sends no CORS headers
|
||||
/// and checks its password before CORS, so the unauthenticated preflight
|
||||
/// a browser always sends is refused - measured, not assumed.
|
||||
WebrtcOffer { session_id: String, camera_stream: String, sdp: String },
|
||||
}
|
||||
|
||||
/// The closed set of things the cloud is allowed to ask a printer to do.
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
//! WebRTC signalling, relayed from the cloud to this gateway's own go2rtc.
|
||||
//!
|
||||
//! Only the signalling. The video never touches this process or the cloud:
|
||||
//! once the browser has go2rtc's answer, the two negotiate media directly
|
||||
//! over ICE. On a workshop LAN that means the stream never leaves the
|
||||
//! building and costs the cloud nothing, which is the entire reason the
|
||||
//! relay carries SDP rather than bytes.
|
||||
//!
|
||||
//! ## Why the cloud has to be in the middle at all
|
||||
//!
|
||||
//! A browser cannot signal to go2rtc itself. Measured against go2rtc 1.9.14:
|
||||
//! it returns no `Access-Control-Allow-Origin` on any response, and it checks
|
||||
//! its API password *before* CORS, so the unauthenticated preflight a browser
|
||||
//! always sends gets a 401. A gateway is also behind a workshop NAT, so the
|
||||
//! cloud cannot dial in either - the only way through is the connection this
|
||||
//! gateway already opened outbound.
|
||||
//!
|
||||
//! ## Why no credential appears here
|
||||
//!
|
||||
//! The request goes to `127.0.0.1`, and go2rtc exempts loopback from its API
|
||||
//! password. That is a measured property, not an assumption, and it is also
|
||||
//! why a probe from the gateway itself cannot be used to check that the
|
||||
//! password is working - test from another host for that.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
/// go2rtc answers an offer after gathering candidates, which includes a STUN
|
||||
/// round trip to discover its public address, so this is not instant. Still
|
||||
/// bounded: a viewer waiting on a dead go2rtc should be told, and the cloud
|
||||
/// is holding a request open for this.
|
||||
const NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(8);
|
||||
|
||||
/// Turns go2rtc's *bind* address into a loopback URL we can call it on.
|
||||
///
|
||||
/// `CONTINUUM_GO2RTC_API_LISTEN` is a listen address - `:1984`,
|
||||
/// `0.0.0.0:1984` - and neither is something to send a request to. Only the
|
||||
/// port is portable, and loopback is the address we want regardless: it is
|
||||
/// the one go2rtc exempts from its password.
|
||||
fn local_api_base(api_listen: &str) -> anyhow::Result<String> {
|
||||
let port = api_listen
|
||||
.rsplit_once(':')
|
||||
.map(|(_, port)| port)
|
||||
.unwrap_or(api_listen)
|
||||
.trim();
|
||||
|
||||
if port.is_empty() || !port.chars().all(|c| c.is_ascii_digit()) {
|
||||
anyhow::bail!("cannot tell which port go2rtc's API is on from '{api_listen}'");
|
||||
}
|
||||
Ok(format!("http://127.0.0.1:{port}"))
|
||||
}
|
||||
|
||||
/// Rejects a stream name that could not be a go2rtc stream name.
|
||||
///
|
||||
/// It goes into a query string, and this is cheaper and clearer than
|
||||
/// percent-encoding: a real name is a printer id from `printers.toml`, so
|
||||
/// anything outside this set is a bug or an attempt at one, and refusing it
|
||||
/// says so instead of forwarding something odd to go2rtc.
|
||||
fn checked_stream_name(name: &str) -> anyhow::Result<&str> {
|
||||
if name.is_empty() || !name.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) {
|
||||
anyhow::bail!("'{name}' is not a usable go2rtc stream name");
|
||||
}
|
||||
Ok(name)
|
||||
}
|
||||
|
||||
/// Posts the browser's offer to go2rtc and returns its answer SDP.
|
||||
pub async fn negotiate(api_listen: &str, camera_stream: &str, offer_sdp: &str) -> anyhow::Result<String> {
|
||||
let base = local_api_base(api_listen)?;
|
||||
let stream = checked_stream_name(camera_stream)?;
|
||||
|
||||
let response = reqwest::Client::new()
|
||||
.post(format!("{base}/api/webrtc?src={stream}"))
|
||||
.header("Content-Type", "application/sdp")
|
||||
.body(offer_sdp.to_string())
|
||||
.timeout(NEGOTIATION_TIMEOUT)
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
// go2rtc answers 201 rather than 200 here, so this checks the class and
|
||||
// not a single code.
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
// Its body is the useful half - "streams: wrong url" for a camera it
|
||||
// does not have, say - and a bare status would send whoever is
|
||||
// reading the logs looking at the network instead.
|
||||
let body = response.text().await.unwrap_or_default();
|
||||
let detail = body.trim();
|
||||
anyhow::bail!(
|
||||
"go2rtc refused the offer for '{stream}' (HTTP {}){}",
|
||||
status.as_u16(),
|
||||
if detail.is_empty() { String::new() } else { format!(": {detail}") }
|
||||
);
|
||||
}
|
||||
|
||||
let answer = response.text().await?;
|
||||
if answer.trim().is_empty() {
|
||||
anyhow::bail!("go2rtc accepted the offer for '{stream}' but returned an empty answer");
|
||||
}
|
||||
Ok(answer)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_bind_address_becomes_a_loopback_url() {
|
||||
// The three shapes an operator actually writes, all of which have to
|
||||
// end up at loopback - that is the address go2rtc does not demand a
|
||||
// password on.
|
||||
for listen in [":1984", "0.0.0.0:1984", "127.0.0.1:1984"] {
|
||||
assert_eq!(local_api_base(listen).unwrap(), "http://127.0.0.1:1984");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_bind_address_with_no_port_is_refused_rather_than_guessed() {
|
||||
// Guessing 1984 would produce a gateway that silently signals to the
|
||||
// wrong place when someone has moved the port.
|
||||
for listen in ["", "localhost", ":", "1984a", ":http"] {
|
||||
assert!(local_api_base(listen).is_err(), "{listen:?} should be refused");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_stream_name_that_could_break_out_of_the_query_is_refused() {
|
||||
for name in ["", "h2c&src=other", "h2c?x", "h2c/../y", "h2c one"] {
|
||||
assert!(checked_stream_name(name).is_err(), "{name:?} should be refused");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_real_printer_id_is_accepted() {
|
||||
for name in ["h2c", "p1p", "printer-2", "bay_3", "x1.c"] {
|
||||
assert_eq!(checked_stream_name(name).unwrap(), name);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user