1use std::time::Duration;23use iroh::Endpoint;4use sigil_protocol::{5 CONTROL_ALPN_V1, CONTROL_ALPN_V2, Capability, ClientHello, ClientHelloV2, MEDIA_ALPN_V3,6 PointerSurfaceDimensions, SessionSnapshotV2, read_host_hello, read_host_hello_v2,7 write_client_hello, write_client_hello_v2,8};910pub const MEDIA_TRANSPORT_NAMES: [&str; 2] = ["iroh-moq", "grouped-v3"];1112pub(crate) const CLIENT_ENDPOINT_CLOSE_TIMEOUT: Duration = Duration::from_secs(2);1314pub(crate) struct NegotiatedV1Stream {15 pub(crate) session_id: u64,16 pub(crate) capabilities: Vec<Capability>,17 pub(crate) pointer_surface_dimensions: Option<PointerSurfaceDimensions>,18}1920pub(crate) struct NegotiatedV2Stream {21 pub(crate) session_id: u64,22 pub(crate) pointer_surface_dimensions: Option<PointerSurfaceDimensions>,23 pub(crate) initial_snapshot: SessionSnapshotV2,24 pub(crate) media_subscription_capability: String,25 pub(crate) media_authorization_revision: u64,26}272829pub(crate) enum NegotiatedMediaStream {30 V2(NegotiatedV2Stream),31 V1(NegotiatedV1Stream),32}3334impl NegotiatedMediaStream {35 pub(crate) fn session_id(&self) -> u64 {36 match self {37 Self::V2(stream) => stream.session_id,38 Self::V1(stream) => stream.session_id,39 }40 }4142 pub(crate) fn pointer_surface_dimensions(&self) -> Option<PointerSurfaceDimensions> {43 match self {44 Self::V2(stream) => stream.pointer_surface_dimensions,45 Self::V1(stream) => stream.pointer_surface_dimensions,46 }47 }48}495051pub(crate) enum ControlProtocol {52 V2,53 V1,54}5556impl ControlProtocol {57 pub(crate) const fn diagnostic_name(self) -> &'static str {58 match self {59 Self::V2 => "control-v2",60 Self::V1 => "legacy-v1",61 }62 }63}646566pub(crate) enum MediaTransport {67 UpstreamMoq,68 GroupedObjectsV3,69}7071impl MediaTransport {72 pub(crate) const fn diagnostic_name(self) -> &'static str {73 match self {74 Self::UpstreamMoq => MEDIA_TRANSPORT_NAMES[0],75 Self::GroupedObjectsV3 => MEDIA_TRANSPORT_NAMES[1],76 }77 }78}7980pub(crate) async fn negotiate_v1(81 send: &mut iroh::endpoint::SendStream,82 recv: &mut iroh::endpoint::RecvStream,83 nonce: [u8; 16],84 capabilities: Vec<Capability>,85 required: Option<Capability>,86 stream_name: &str,87 invitation: Option<&str>,88) -> Result<NegotiatedV1Stream, String> {89 let mut hello = ClientHello::new("portal/0.1.0", nonce, capabilities.clone());90 if let Some(invitation) = invitation {91 hello = hello.with_invitation(invitation);92 }93 write_client_hello(send, &hello)94 .await95 .map_err(|e| format!("Failed to send {stream_name} handshake: {e}"))?;96 let response = tokio::time::timeout(Duration::from_secs(10), read_host_hello(recv))97 .await98 .map_err(|_| format!("Timed out waiting for {stream_name} handshake"))?99 .map_err(|e| format!("Invalid {stream_name} handshake: {e}"))?100 .ok_or_else(|| format!("Host closed during {stream_name} handshake"))?;101 if !response.accepted {102 return Err(format!(103 "Host rejected {stream_name} stream: {}",104 response.message.as_deref().unwrap_or("unspecified reason")105 ));106 }107 if let Some(required) = required108 && !response.capabilities.contains(&required)109 {110 return Err(format!(111 "Host accepted {stream_name} without required capability {required:?}"112 ));113 }114 if let Some(unoffered) = response115 .capabilities116 .iter()117 .find(|capability| !capabilities.contains(capability))118 {119 return Err(format!(120 "Host accepted unoffered {stream_name} capability {unoffered:?}"121 ));122 }123 let session_id = response124 .session_id125 .ok_or_else(|| format!("Host omitted {stream_name} session ID"))?;126 Ok(NegotiatedV1Stream {127 session_id,128 capabilities: response.capabilities,129 pointer_surface_dimensions: response.pointer_surface_dimensions,130 })131}132133async fn negotiate_v2(134 send: &mut iroh::endpoint::SendStream,135 recv: &mut iroh::endpoint::RecvStream,136 nonce: [u8; 16],137 invitation: Option<&str>,138) -> Result<NegotiatedV2Stream, String> {139 let capabilities = vec![Capability::VideoH264, Capability::AudioOpus];140 let mut hello = ClientHelloV2::new("portal/0.1.0", nonce, capabilities.clone());141 if let Some(invitation) = invitation {142 hello = hello.with_invitation(invitation);143 }144 write_client_hello_v2(send, &hello)145 .await146 .map_err(|error| format!("Failed to send control v2 handshake: {error}"))?;147 let response = tokio::time::timeout(Duration::from_secs(10), read_host_hello_v2(recv))148 .await149 .map_err(|_| "Timed out waiting for control v2 handshake".to_string())?150 .map_err(|error| format!("Invalid control v2 handshake: {error}"))?151 .ok_or_else(|| "Host closed during control v2 handshake".to_string())?;152 if !response.accepted {153 return Err(format!(154 "Host rejected control v2 stream: {}",155 response.message.as_deref().unwrap_or("unspecified reason")156 ));157 }158 if !response.capabilities.contains(&Capability::VideoH264) {159 return Err("Host accepted control v2 without required VideoH264 capability".to_string());160 }161 if let Some(unoffered) = response162 .capabilities163 .iter()164 .find(|capability| !capabilities.contains(capability))165 {166 return Err(format!(167 "Host accepted unoffered control v2 capability {unoffered:?}"168 ));169 }170 Ok(NegotiatedV2Stream {171 session_id: response172 .session_id173 .ok_or_else(|| "Host omitted control v2 session ID".to_string())?,174 pointer_surface_dimensions: response.pointer_surface_dimensions,175 initial_snapshot: response176 .snapshot177 .ok_or_else(|| "Host omitted initial control v2 snapshot".to_string())?,178 media_subscription_capability: response179 .media_subscription_capability180 .ok_or_else(|| "Host omitted control v2 media subscription capability".to_string())?,181 media_authorization_revision: response182 .media_authorization_revision183 .ok_or_else(|| "Host omitted control v2 media authorization revision".to_string())?,184 })185}186187fn connection_error_is_unsupported_alpn(error: &iroh::endpoint::ConnectionError) -> bool {188 matches!(189 error,190 iroh::endpoint::ConnectionError::ConnectionClosed(close)191 if close.error_code == iroh::endpoint::TransportErrorCode::crypto(0x78)192 )193}194195pub(crate) fn connect_error_is_unsupported_alpn(error: &iroh::endpoint::ConnectError) -> bool {196 match error {197 iroh::endpoint::ConnectError::Connecting {198 source: iroh::endpoint::ConnectingError::ConnectionError { source, .. },199 ..200 }201 | iroh::endpoint::ConnectError::Connection { source, .. } => {202 connection_error_is_unsupported_alpn(source)203 }204 _ => false,205 }206}207208pub(crate) async fn open_negotiated_media_stream(209 endpoint: &Endpoint,210 address: &iroh::EndpointAddr,211 nonce: [u8; 16],212 invitation: Option<&str>,213) -> Result<214 (215 iroh::endpoint::Connection,216 iroh::endpoint::RecvStream,217 iroh::endpoint::SendStream,218 NegotiatedMediaStream,219 MediaTransport,220 ControlProtocol,221 ),222 String,223> {224 match endpoint.connect(address.clone(), CONTROL_ALPN_V2).await {225 Ok(connection) => {226 let (mut send, mut recv) = connection227 .open_bi()228 .await229 .map_err(|error| format!("Failed to open control v2 handshake: {error}"))?;230 let negotiation = negotiate_v2(&mut send, &mut recv, nonce, invitation).await?;231 Ok((232 connection,233 recv,234 send,235 NegotiatedMediaStream::V2(negotiation),236 MediaTransport::UpstreamMoq,237 ControlProtocol::V2,238 ))239 }240 Err(v2_error) if connect_error_is_unsupported_alpn(&v2_error) => {241 open_negotiated_v1_or_v3(endpoint, address, nonce, invitation, v2_error).await242 }243 Err(v2_error) => Err(format!(244 "Control v2 connection failed without an explicit unsupported-ALPN signal; refusing an unsafe downgrade: {v2_error}"245 )),246 }247}248249async fn open_negotiated_v1_or_v3(250 endpoint: &Endpoint,251 address: &iroh::EndpointAddr,252 nonce: [u8; 16],253 invitation: Option<&str>,254 v2_error: iroh::endpoint::ConnectError,255) -> Result<256 (257 iroh::endpoint::Connection,258 iroh::endpoint::RecvStream,259 iroh::endpoint::SendStream,260 NegotiatedMediaStream,261 MediaTransport,262 ControlProtocol,263 ),264 String,265> {266 match endpoint.connect(address.clone(), CONTROL_ALPN_V1).await {267 Ok(connection) => {268 let (mut send, mut recv) = connection269 .open_bi()270 .await271 .map_err(|error| format!("Failed to open control handshake: {error}"))?;272 let negotiation = negotiate_v1(273 &mut send,274 &mut recv,275 nonce,276 vec![Capability::VideoH264],277 Some(Capability::VideoH264),278 "control",279 invitation,280 )281 .await?;282 // CONTROL owns the authenticated host lease. Keep both the283 // connection and the client->host send leg alive for keyframe284 // requests while media uses a separate upstream MoQ session.285 Ok((286 connection,287 recv,288 send,289 NegotiatedMediaStream::V1(negotiation),290 MediaTransport::UpstreamMoq,291 ControlProtocol::V1,292 ))293 }294 Err(control_error) if connect_error_is_unsupported_alpn(&control_error) => {295 // Grouped-v3 is the sole compatibility tier below upstream MoQ.296 let connection = endpoint297 .connect(address.clone(), MEDIA_ALPN_V3)298 .await299 .map_err(|v3_error| {300 format!(301 "Control v2 was unsupported ({v2_error}); control v1 failed ({control_error}); media v3 compatibility connection also failed ({v3_error})"302 )303 })?;304 let (mut send, mut recv) = connection305 .open_bi()306 .await307 .map_err(|error| format!("Failed to open media v3 handshake: {error}"))?;308 let negotiation = negotiate_v1(309 &mut send,310 &mut recv,311 nonce,312 vec![Capability::VideoH264],313 Some(Capability::VideoH264),314 "media v3",315 invitation,316 )317 .await?;318 Ok((319 connection,320 recv,321 send,322 NegotiatedMediaStream::V1(negotiation),323 MediaTransport::GroupedObjectsV3,324 ControlProtocol::V1,325 ))326 }327 Err(control_error) => Err(format!(328 "Control v2 was unsupported ({v2_error}); control v1 failed without an explicit unsupported-ALPN signal; refusing an unsafe media downgrade: {control_error}"329 )),330 }331}332333334mod tests {335 use super::*;336337 338 fn upstream_moq_transport_is_distinct_from_legacy_compatibility() {339 assert_eq!(MediaTransport::UpstreamMoq.diagnostic_name(), "iroh-moq");340 assert_ne!(341 MediaTransport::UpstreamMoq,342 MediaTransport::GroupedObjectsV3343 );344 }345346 347 fn compatibility_downgrade_requires_tls_no_application_protocol() {348 let unsupported = iroh::endpoint::ConnectionError::ConnectionClosed(349 iroh::endpoint::TransportError::new(350 iroh::endpoint::TransportErrorCode::crypto(0x78),351 "no application protocol".to_string(),352 )353 .into(),354 );355 assert!(connection_error_is_unsupported_alpn(&unsupported));356 assert!(!connection_error_is_unsupported_alpn(357 &iroh::endpoint::ConnectionError::TimedOut358 ));359 assert!(!connection_error_is_unsupported_alpn(360 &iroh::endpoint::ConnectionError::Reset361 ));362 }363364 365 fn media_transport_names_match_diagnostic_mapping() {366 assert_eq!(MEDIA_TRANSPORT_NAMES, ["iroh-moq", "grouped-v3"]);367 assert_eq!(368 [369 MediaTransport::UpstreamMoq.diagnostic_name(),370 MediaTransport::GroupedObjectsV3.diagnostic_name(),371 ],372 MEDIA_TRANSPORT_NAMES,373 );374 }375}