continuum-proxy/src/uplink/webrtc.rs

//! 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)
}

/// Flattens an error's `source` chain into one line.
///
/// Needed because reqwest's own `Display` is just "error sending request for
/// url (...)" - the reason lives one or two sources down, and "connection
/// refused" versus "timed out" is the whole content of the message. This
/// error travels to the cloud and then to a viewer, so it is the only
/// explanation anyone gets.
fn with_causes(err: &(dyn std::error::Error + 'static)) -> String {
    let mut out = err.to_string();
    let mut source = err.source();
    while let Some(cause) = source {
        let text = cause.to_string();
        // Skip a cause that only repeats what we already said, which hyper's
        // wrappers do.
        if !out.contains(&text) {
            out.push_str(": ");
            out.push_str(&text);
        }
        source = cause.source();
    }
    out
}

/// 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
        .map_err(|err| {
            // Names go2rtc explicitly. The most likely cause by far is that
            // it is not running, and an error mentioning only a port sends
            // whoever reads it looking at the network.
            anyhow::anyhow!("could not reach go2rtc at {base}: {}", with_causes(&err))
        })?;

    // 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);
        }
    }
}