1use std::time::Duration;23use iroh::Endpoint;4use sigil_protocol::{5 CONTROL_ALPN_V1, Capability, ClientHello, MEDIA_ALPN_V3, PointerSurfaceDimensions,6 read_host_hello, write_client_hello,7};89pub const MEDIA_TRANSPORT_NAMES: [&str; 2] = ["iroh-moq", "grouped-v3"];1011pub(crate) const CLIENT_ENDPOINT_CLOSE_TIMEOUT: Duration = Duration::from_secs(2);1213pub(crate) struct NegotiatedV1Stream {14 pub(crate) session_id: u64,15 pub(crate) capabilities: Vec<Capability>,16 pub(crate) pointer_surface_dimensions: Option<PointerSurfaceDimensions>,17}181920pub(crate) enum MediaTransport {21 UpstreamMoq,22 GroupedObjectsV3,23}2425impl MediaTransport {26 pub(crate) const fn diagnostic_name(self) -> &'static str {27 match self {28 Self::UpstreamMoq => MEDIA_TRANSPORT_NAMES[0],29 Self::GroupedObjectsV3 => MEDIA_TRANSPORT_NAMES[1],30 }31 }32}3334pub(crate) async fn negotiate_v1(35 send: &mut iroh::endpoint::SendStream,36 recv: &mut iroh::endpoint::RecvStream,37 nonce: [u8; 16],38 capabilities: Vec<Capability>,39 required: Option<Capability>,40 stream_name: &str,41 invitation: Option<&str>,42) -> Result<NegotiatedV1Stream, String> {43 let mut hello = ClientHello::new("portal/0.1.0", nonce, capabilities.clone());44 if let Some(invitation) = invitation {45 hello = hello.with_invitation(invitation);46 }47 write_client_hello(send, &hello)48 .await49 .map_err(|e| format!("Failed to send {stream_name} handshake: {e}"))?;50 let response = tokio::time::timeout(Duration::from_secs(10), read_host_hello(recv))51 .await52 .map_err(|_| format!("Timed out waiting for {stream_name} handshake"))?53 .map_err(|e| format!("Invalid {stream_name} handshake: {e}"))?54 .ok_or_else(|| format!("Host closed during {stream_name} handshake"))?;55 if !response.accepted {56 return Err(format!(57 "Host rejected {stream_name} stream: {}",58 response.message.as_deref().unwrap_or("unspecified reason")59 ));60 }61 if let Some(required) = required62 && !response.capabilities.contains(&required)63 {64 return Err(format!(65 "Host accepted {stream_name} without required capability {required:?}"66 ));67 }68 if let Some(unoffered) = response69 .capabilities70 .iter()71 .find(|capability| !capabilities.contains(capability))72 {73 return Err(format!(74 "Host accepted unoffered {stream_name} capability {unoffered:?}"75 ));76 }77 let session_id = response78 .session_id79 .ok_or_else(|| format!("Host omitted {stream_name} session ID"))?;80 Ok(NegotiatedV1Stream {81 session_id,82 capabilities: response.capabilities,83 pointer_surface_dimensions: response.pointer_surface_dimensions,84 })85}8687fn connection_error_is_unsupported_alpn(error: &iroh::endpoint::ConnectionError) -> bool {88 matches!(89 error,90 iroh::endpoint::ConnectionError::ConnectionClosed(close)91 if close.error_code == iroh::endpoint::TransportErrorCode::crypto(0x78)92 )93}9495pub(crate) fn connect_error_is_unsupported_alpn(error: &iroh::endpoint::ConnectError) -> bool {96 match error {97 iroh::endpoint::ConnectError::Connecting {98 source: iroh::endpoint::ConnectingError::ConnectionError { source, .. },99 ..100 }101 | iroh::endpoint::ConnectError::Connection { source, .. } => {102 connection_error_is_unsupported_alpn(source)103 }104 _ => false,105 }106}107108pub(crate) async fn open_negotiated_media_stream(109 endpoint: &Endpoint,110 address: &iroh::EndpointAddr,111 nonce: [u8; 16],112 invitation: Option<&str>,113) -> Result<114 (115 iroh::endpoint::Connection,116 iroh::endpoint::RecvStream,117 iroh::endpoint::SendStream,118 NegotiatedV1Stream,119 MediaTransport,120 ),121 String,122> {123 match endpoint.connect(address.clone(), CONTROL_ALPN_V1).await {124 Ok(connection) => {125 let (mut send, mut recv) = connection126 .open_bi()127 .await128 .map_err(|error| format!("Failed to open control handshake: {error}"))?;129 let negotiation = negotiate_v1(130 &mut send,131 &mut recv,132 nonce,133 vec![Capability::VideoH264],134 Some(Capability::VideoH264),135 "control",136 invitation,137 )138 .await?;139 // CONTROL owns the authenticated host lease. Keep both the140 // connection and the client->host send leg alive for keyframe141 // requests while media uses a separate upstream MoQ session.142 Ok((143 connection,144 recv,145 send,146 negotiation,147 MediaTransport::UpstreamMoq,148 ))149 }150 Err(control_error) if connect_error_is_unsupported_alpn(&control_error) => {151 // Grouped-v3 is the sole compatibility tier below upstream MoQ.152 let connection = endpoint153 .connect(address.clone(), MEDIA_ALPN_V3)154 .await155 .map_err(|v3_error| {156 format!(157 "Control connection failed ({control_error}); media v3 compatibility connection also failed ({v3_error})"158 )159 })?;160 let (mut send, mut recv) = connection161 .open_bi()162 .await163 .map_err(|error| format!("Failed to open media v3 handshake: {error}"))?;164 let negotiation = negotiate_v1(165 &mut send,166 &mut recv,167 nonce,168 vec![Capability::VideoH264],169 Some(Capability::VideoH264),170 "media v3",171 invitation,172 )173 .await?;174 Ok((175 connection,176 recv,177 send,178 negotiation,179 MediaTransport::GroupedObjectsV3,180 ))181 }182 Err(control_error) => Err(format!(183 "Control connection failed without an explicit unsupported-ALPN signal; refusing an unsafe media downgrade: {control_error}"184 )),185 }186}187188189mod tests {190 use super::*;191192 193 fn upstream_moq_transport_is_distinct_from_legacy_compatibility() {194 assert_eq!(MediaTransport::UpstreamMoq.diagnostic_name(), "iroh-moq");195 assert_ne!(196 MediaTransport::UpstreamMoq,197 MediaTransport::GroupedObjectsV3198 );199 }200201 202 fn compatibility_downgrade_requires_tls_no_application_protocol() {203 let unsupported = iroh::endpoint::ConnectionError::ConnectionClosed(204 iroh::endpoint::TransportError::new(205 iroh::endpoint::TransportErrorCode::crypto(0x78),206 "no application protocol".to_string(),207 )208 .into(),209 );210 assert!(connection_error_is_unsupported_alpn(&unsupported));211 assert!(!connection_error_is_unsupported_alpn(212 &iroh::endpoint::ConnectionError::TimedOut213 ));214 assert!(!connection_error_is_unsupported_alpn(215 &iroh::endpoint::ConnectionError::Reset216 ));217 }218219 220 fn media_transport_names_match_diagnostic_mapping() {221 assert_eq!(MEDIA_TRANSPORT_NAMES, ["iroh-moq", "grouped-v3"]);222 assert_eq!(223 [224 MediaTransport::UpstreamMoq.diagnostic_name(),225 MediaTransport::GroupedObjectsV3.diagnostic_name(),226 ],227 MEDIA_TRANSPORT_NAMES,228 );229 }230}