use super::draft14::{Draft14RelayAccess, NativeDraft14Publisher};
use super::sfu::{NativeSfuPublisher, SfuControlAccess};
use super::{BroadcastPublisherSigner, BroadcastSession, Broadcasts};
use anyhow::{anyhow, ensure, Context, Result};
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
use serde::Deserialize;
use std::time::{SystemTime, UNIX_EPOCH};
#[derive(Deserialize)]
struct GrantResponse {
grant: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct AccessResponse {
issuer_public_key: String,
relay: RelayResponse,
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
enum RelayResponse {
Moq {
#[serde(rename = "relayUrl")]
relay_url: String,
token: String,
namespace: String,
},
Sfu {
#[serde(rename = "controlUrl")]
control_url: String,
},
}
pub enum NativeCarrier {
Moq(NativeDraft14Publisher),
Sfu(NativeSfuPublisher),
}
impl NativeCarrier {
pub async fn flush(&self) -> Result<()> {
match self {
Self::Moq(carrier) => carrier.flush().await,
Self::Sfu(carrier) => carrier.flush().await,
}
}
pub async fn close(self) {
match self {
Self::Moq(carrier) => carrier.close(),
Self::Sfu(carrier) => carrier.close().await,
}
}
}
fn key(value: &str) -> Result<[u8; 32]> {
URL_SAFE_NO_PAD
.decode(value)
.context("invalid issuer key")?
.try_into()
.map_err(|_| anyhow!("issuer key must be 32 bytes"))
}
pub fn checked_api_origin(value: &str, paired: bool) -> Result<url::Url> {
let api = url::Url::parse(value).context("invalid broadcast API URL")?;
let loopback_http = api.scheme() == "http"
&& matches!(api.host_str(), Some("localhost" | "127.0.0.1" | "[::1]"));
ensure!(
(loopback_http || (paired && api.scheme() == "https"))
&& api.username().is_empty()
&& api.password().is_none()
&& api.query().is_none()
&& api.fragment().is_none()
&& (value == api.origin().ascii_serialization()
|| value == format!("{}/", api.origin().ascii_serialization())),
"broadcast API must be loopback HTTP or a paired exact HTTPS origin"
);
Ok(api)
}
pub async fn open_broadcast(
harness_api: &str,
origin: &str,
pairing_code: Option<&str>,
plugin_token: Option<&str>,
) -> Result<(BroadcastSession, NativeCarrier)> {
let api = checked_api_origin(
harness_api,
pairing_code.is_some() || plugin_token.is_some(),
)?;
let origin_url = url::Url::parse(&origin).context("invalid Live public origin")?;
ensure!(
matches!(origin_url.scheme(), "http" | "https")
&& origin_url.origin().ascii_serialization() == origin
&& origin_url.username().is_empty()
&& origin_url.password().is_none(),
"OPENRTC_LIVE_PUBLIC_ORIGIN must be an exact HTTP(S) origin"
);
if api.scheme() == "https" {
ensure!(
origin == api.origin().ascii_serialization(),
"paired broadcast origin must match the HTTPS API"
);
}
let mut seed = [0_u8; 32];
getrandom::getrandom(&mut seed).context("native binding key entropy failed")?;
let binding = SigningKey::from_bytes(&seed);
seed.fill(0);
let binding_public_jwk = serde_json::json!({
"kty": "OKP", "crv": "Ed25519",
"x": URL_SAFE_NO_PAD.encode(binding.verifying_key().to_bytes())
});
let publisher_signer = BroadcastPublisherSigner::generate()?;
let publication_key = URL_SAFE_NO_PAD.encode(publisher_signer.verifying_key());
let http = reqwest::Client::new();
let grant_path = if plugin_token.is_some() {
"/api/broadcast/obs/plugin/start"
} else if pairing_code.is_some() {
"/api/broadcast/obs/redeem"
} else {
"/api/broadcast/publisher"
};
let publisher: GrantResponse = http
.post(api.join(grant_path)?)
.header("Origin", origin)
.json(&serde_json::json!({
"code": pairing_code,
"pluginToken": plugin_token,
"bindingPublicJwk": binding_public_jwk,
"publicationVerifyingKey": publication_key
}))
.send()
.await?
.error_for_status()?
.json()
.await?;
let access: AccessResponse = http
.post(api.join("/api/broadcast/access")?)
.header("Origin", origin)
.json(&serde_json::json!({
"grant": publisher.grant,
"bindingPublicJwk": binding_public_jwk,
"publicationVerifyingKey": publication_key
}))
.send()
.await?
.error_for_status()?
.json()
.await?;
let issuer = VerifyingKey::from_bytes(&key(&access.issuer_public_key)?)
.context("issuer key was rejected")?;
let now_ms = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64;
let broadcasts = Broadcasts::default();
let challenge = broadcasts.prepare_grant_verification(&publisher.grant, &issuer, now_ms)?;
let proof = binding.sign(&challenge.signing_bytes);
let grant = broadcasts.complete_grant_verification(
&publisher.grant,
&issuer,
&challenge.handle,
&proof.to_bytes(),
now_ms,
)?;
let session = broadcasts.open_publisher(grant, &publisher_signer, now_ms)?;
let carrier = match access.relay {
RelayResponse::Moq {
relay_url,
token,
namespace,
} => {
let relay = Draft14RelayAccess::new(&relay_url, token, namespace)?;
NativeCarrier::Moq(NativeDraft14Publisher::connect_live(session.clone(), relay).await?)
}
RelayResponse::Sfu { control_url } => {
let control_url = api.join(&control_url)?;
ensure!(
control_url.origin() == api.origin() && control_url.scheme() == api.scheme(),
"SFU control must stay on the broadcast API origin"
);
NativeCarrier::Sfu(
NativeSfuPublisher::connect(
session.clone(),
SfuControlAccess {
control_url,
origin: origin.to_owned(),
grant: publisher.grant,
binding_public_jwk,
publication_verifying_key: publication_key,
},
)
.await?,
)
}
};
Ok((session, carrier))
}