1use std::sync::Mutex;23use super::*;45pub(super) const MEDIA_CAPABILITIES: &[Capability] = &[Capability::VideoH264];6const AUDIO_CAPABILITIES: &[Capability] = &[Capability::AudioOpus];7pub(super) const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);8const INPUT_ACK_TIMEOUT: Duration = Duration::from_secs(1);9const REJECTION_DRAIN_TIMEOUT: Duration = Duration::from_secs(1);101112struct InputSessionGuard<L, F>13where14 F: FnOnce() -> Result<()>,15{16 _lease: L,17 reset: Option<F>,18}1920impl<L, F> InputSessionGuard<L, F>21where22 F: FnOnce() -> Result<()>,23{24 fn new(lease: L, reset: F) -> Self {25 Self {26 _lease: lease,27 reset: Some(reset),28 }29 }3031 fn finish(mut self) -> Result<()> {32 self.run()33 }3435 fn run(&mut self) -> Result<()> {36 let Some(reset) = self.reset.take() else {37 return Ok(());38 };39 reset()40 }41}4243impl<L, F> Drop for InputSessionGuard<L, F>44where45 F: FnOnce() -> Result<()>,46{47 fn drop(&mut self) {48 if let Err(error) = self.run() {49 error!(%error, "failed to release held input transitions while dropping session");50 }51 }52}535455pub struct MediaV3Handler {56 pub config: HostConfig,57 pub sessions: Arc<SessionRegistry>,58 pub authorization: AuthorizationPolicy,59}606162pub struct ControlHandler {63 pub config: HostConfig,64 pub sessions: Arc<SessionRegistry>,65 pub authorization: AuthorizationPolicy,66}676869pub struct ControlV2Handler {70 pub sessions: Arc<SessionRegistry>,71 pub generations: Arc<MediaGenerationManager>,72 pub authorization: AuthorizationPolicy,73 pub input_operations: Arc<InputOperations>,74 pub host_secret: [u8; 32],75}767778pub struct MediaFeedbackHandler {79 pub config: HostConfig,80 pub sessions: Arc<SessionRegistry>,81 pub generations: Arc<MediaGenerationManager>,82 pub authorization: AuthorizationPolicy,83 pub input_operations: Arc<InputOperations>,84}8586/// Upstream MoQ admission guarded by an already-authenticated control lease.87///88/// This deliberately does not use `iroh_moq::Moq::protocol_handler`: that89/// actor makes a completed session globally visible before application-level90/// acceptance. Consuming the exact pending attachment first prevents MoQ from91/// bypassing Sigil's invitation, enrollment, and one-client gate.9293pub struct AuthorizedMoqHandler {94 pub sessions: Arc<SessionRegistry>,95 pub origin: Origin,96}9798impl ProtocolHandler for MediaV3Handler {99 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {100 let remote = connection.remote_id();101 if let Err(error) = serve_media_v3(102 connection,103 self.config.clone(),104 &self.sessions,105 &self.authorization,106 )107 .await108 {109 warn!(%remote, %error, "media v3 connection ended");110 }111 Ok(())112 }113}114115impl ProtocolHandler for ControlHandler {116 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {117 let remote = connection.remote_id();118 if let Err(error) = serve_control_moq(119 connection,120 self.config.clone(),121 &self.sessions,122 &self.authorization,123 )124 .await125 {126 warn!(%remote, %error, "MoQ control connection ended");127 }128 Ok(())129 }130}131132impl ProtocolHandler for MediaFeedbackHandler {133 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {134 let remote = connection.remote_id();135 if let Err(error) = serve_media_feedback(136 connection,137 &self.config,138 &self.sessions,139 &self.generations,140 &self.authorization,141 &self.input_operations,142 )143 .await144 {145 warn!(%remote, %error, "media feedback connection ended");146 }147 Ok(())148 }149}150151impl ProtocolHandler for AuthorizedMoqHandler {152 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {153 let remote = connection.remote_id();154 let attachment = match self.sessions.claim_moq(remote) {155 Ok(attachment) => attachment,156 Err(error) => {157 connection.close(MOQ_REJECT_CODE.into(), b"unauthorized MoQ attachment");158 warn!(%remote, %error, "rejected unsolicited MoQ connection");159 return Ok(());160 }161 };162 if let Err(error) = serve_authorized_moq(connection, self.origin, attachment).await {163 warn!(%remote, %error, "authorized MoQ connection ended");164 }165 Ok(())166 }167}168169170pub struct InputHandler {171 pub backend: InputBackend,172 pub pointer_positions: Option<PointerPositionTracker>,173 pub sessions: Arc<SessionRegistry>,174}175176177pub struct InputOperations {178 backend: InputBackend,179 operation: Mutex<()>,180}181182impl InputOperations {183 pub fn new(backend: InputBackend) -> Self {184 Self {185 backend,186 operation: Mutex::new(()),187 }188 }189190 fn capabilities(&self) -> &'static [Capability] {191 self.backend.capabilities()192 }193194 fn apply_v2(195 &self,196 sessions: &SessionRegistry,197 remote: EndpointId,198 binding: &InputEventV2,199 negotiated: &[Capability],200 ) -> Result<InputDisposition> {201 let _operation = self202 .operation203 .lock()204 .map_err(|_| anyhow::anyhow!("input operation lock poisoned"))?;205 ensure!(206 sessions.is_v2_focus_owner(207 remote,208 binding.session_id,209 binding.slot,210 binding.focus_generation,211 ),212 "input focus changed before backend application"213 );214 self.backend.apply(&binding.event, negotiated)215 }216217 pub(crate) fn reset(&self) -> Result<()> {218 let _operation = match self.operation.lock() {219 Ok(operation) => operation,220 Err(poisoned) => {221 warn!("recovering poisoned input operation lock for reset");222 poisoned.into_inner()223 }224 };225 self.backend.reset_session()226 }227228 pub(crate) fn neutralize_focus_transition(229 &self,230 sessions: &SessionRegistry,231 transition_id: u64,232 reason: FocusTransitionReasonV2,233 ) -> Result<()> {234 info!(235 transition_id,236 ?reason,237 "focus transition entered no-inject neutralizing state"238 );239 self.reset()?;240 info!(241 transition_id,242 ?reason,243 "focus transition virtual input neutralized"244 );245 sessions.complete_v2_focus_transition(transition_id)?;246 info!(247 transition_id,248 ?reason,249 "focus transition successor committed after neutralization"250 );251 Ok(())252 }253}254255256pub struct InputV2Handler {257 pub operations: Arc<InputOperations>,258 pub pointer_positions: Option<PointerPositionTracker>,259 pub sessions: Arc<SessionRegistry>,260}261262263pub struct AudioHandler {264 pub config: HostConfig,265 pub sessions: Arc<SessionRegistry>,266}267268impl ProtocolHandler for AudioHandler {269 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {270 let remote = connection.remote_id();271 if let Err(error) = serve_audio(connection, self.config.clone(), &self.sessions).await {272 warn!(%remote, %error, "audio connection ended");273 }274 Ok(())275 }276}277278impl ProtocolHandler for InputHandler {279 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {280 let remote = connection.remote_id();281 if let Err(error) = serve_input(282 connection,283 &self.backend,284 self.pointer_positions.as_ref(),285 &self.sessions,286 )287 .await288 {289 warn!(%remote, %error, "input connection ended");290 }291 Ok(())292 }293}294295impl ProtocolHandler for ControlV2Handler {296 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {297 let remote = connection.remote_id();298 if let Err(error) = serve_control_moq_v2(299 connection,300 &self.sessions,301 &self.generations,302 &self.authorization,303 &self.input_operations,304 self.host_secret,305 )306 .await307 {308 warn!(%remote, %error, "MoQ control v2 connection ended");309 }310 Ok(())311 }312}313314impl ProtocolHandler for InputV2Handler {315 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {316 let remote = connection.remote_id();317 if let Err(error) = serve_input_v2(318 connection,319 &self.operations,320 self.pointer_positions.as_ref(),321 &self.sessions,322 )323 .await324 {325 warn!(%remote, %error, "input v2 connection ended");326 }327 Ok(())328 }329}330331async fn serve_input(332 connection: Connection,333 backend: &InputBackend,334 pointer_positions: Option<&PointerPositionTracker>,335 sessions: &Arc<SessionRegistry>,336) -> Result<()> {337 let remote = connection.remote_id();338 let handshake_permit = sessions339 .pending_handshakes340 .try_acquire()341 .context("too many pending handshakes")?;342 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())343 .await344 .context("timed out accepting input stream")?345 .context("accepting input stream")?;346 let hello = receive_hello_unconstrained(&mut recv).await?;347 drop(handshake_permit);348 debug!(%remote, agent = %hello.agent, "input hello received");349 ensure!(350 hello.invitation.is_none(),351 "invitations are accepted only on the first media handshake"352 );353 let lease = match sessions.claim_input(remote, hello.nonce) {354 Ok(lease) => lease,355 Err(error) => {356 send_rejection(&mut send, error.to_string()).await?;357 return Err(error);358 }359 };360 let session_id = lease.session_id;361 let grants = lease.grants;362 let authorization_revision = lease.authorization_revision;363 // Owning the lease guarantees cancellation or unwinding neutralizes the364 // virtual devices before another input stream can be admitted.365 let session_guard = InputSessionGuard::new(lease, || backend.reset_session());366 let supported =367 supported_input_capabilities(backend.capabilities(), pointer_positions.is_some());368 let negotiated = negotiated_input_capabilities(&hello, &supported, grants);369370 let ack_enabled = negotiated.contains(&Capability::InputAck);371 let feedback_enabled = negotiated.contains(&Capability::PointerPositionFeedback);372 let visibility_feedback_enabled = negotiated.contains(&Capability::PointerVisibilityFeedback);373 let mut pointer_positions = pointer_positions374 .filter(|_| feedback_enabled)375 .map(PointerPositionTracker::subscribe);376 write_host_hello(377 &mut send,378 &HostHello::accepted(session_id, negotiated.clone()),379 )380 .await?;381 info!(%remote, session_id, authorization_revision, "input client accepted");382383 let session_result: Result<()> = async {384 let mut received_events = 0_u64;385 if let Some(pointer_positions) = pointer_positions.as_ref() {386 let pointer_state = *pointer_positions.borrow();387 let (pointer_position, pointer_visible) =388 pointer_feedback_fields(Some(pointer_state), visibility_feedback_enabled);389 tokio::time::timeout(390 INPUT_ACK_TIMEOUT,391 write_input_ack(392 &mut send,393 &InputAck {394 sequence: received_events,395 pointer_position,396 pointer_visible,397 },398 ),399 )400 .await401 .context("timed out writing initial pointer position")??;402 }403 loop {404 if !sessions.is_active(remote, session_id) {405 debug!(%remote, session_id, "media session ended; closing input");406 break;407 }408 tokio::select! {409 _ = sessions.session_changed.notified() => continue,410 changed = async {411 pointer_positions412 .as_mut()413 .expect("feedback branch is guarded")414 .changed()415 .await416 }, if pointer_positions.is_some() => {417 changed.context("Xwayland pointer tracker stopped")?;418 let pointer_state = {419 let receiver = pointer_positions420 .as_mut()421 .expect("feedback branch is guarded");422 *receiver.borrow_and_update()423 };424 let (pointer_position, pointer_visible) =425 pointer_feedback_fields(Some(pointer_state), visibility_feedback_enabled);426 tokio::time::timeout(427 INPUT_ACK_TIMEOUT,428 write_input_ack(429 &mut send,430 &InputAck {431 sequence: received_events,432 pointer_position,433 pointer_visible,434 },435 ),436 )437 .await438 .context("timed out writing pointer position feedback")??;439 }440 event = read_input_event(&mut recv) => {441 let Some(event) = event? else {442 break;443 };444 if !sessions.is_active(remote, session_id) {445 debug!(%remote, session_id, "discarding input after media ended");446 break;447 }448 match backend.apply(&event, &negotiated)? {449 InputDisposition::Probed => {450 debug!(%remote, "input liveness probe acknowledged");451 }452 InputDisposition::Observed => {453 info!(%remote, event_type = input_event_type(&event), "input event observed");454 }455 InputDisposition::Disabled => {456 debug!(%remote, event_type = input_event_type(&event), "input event ignored because injection is disabled");457 }458 459 InputDisposition::Injected => {460 debug!(%remote, event_type = input_event_type(&event), "input event injected");461 }462 InputDisposition::TextIgnored => {463 debug!(%remote, event_type = "text", "text input is unsupported and was ignored");464 }465 }466 received_events = received_events.saturating_add(1);467 if ack_enabled {468 let pointer_state = pointer_positions469 .as_ref()470 .map(|positions| *positions.borrow());471 let (pointer_position, pointer_visible) =472 pointer_feedback_fields(pointer_state, visibility_feedback_enabled);473 tokio::time::timeout(474 INPUT_ACK_TIMEOUT,475 write_input_ack(476 &mut send,477 &InputAck {478 sequence: received_events,479 pointer_position,480 pointer_visible,481 },482 ),483 )484 .await485 .context("timed out writing input acknowledgment")??;486 }487 }488 }489 }490 Ok(())491 }492 .await;493 let reset_result = session_guard494 .finish()495 .context("releasing held input transitions at session end");496 if session_result.is_err()497 && let Err(error) = &reset_result498 {499 error!(%error, "input session and held-transition release both failed");500 }501 let result = session_result.and(reset_result);502 info!(%remote, "input client released");503 result504}505506async fn serve_input_v2(507 connection: Connection,508 operations: &InputOperations,509 pointer_positions: Option<&PointerPositionTracker>,510 sessions: &Arc<SessionRegistry>,511) -> Result<()> {512 let remote = connection.remote_id();513 let handshake_permit = sessions514 .pending_handshakes515 .try_acquire()516 .context("too many pending handshakes")?;517 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())518 .await519 .context("timed out accepting input v2 stream")?520 .context("accepting input v2 stream")?;521 let hello = tokio::time::timeout(HANDSHAKE_TIMEOUT, read_input_client_hello_v2(&mut recv))522 .await523 .context("timed out waiting for input v2 hello")??524 .context("client closed before input v2 hello")?;525 drop(handshake_permit);526 debug!(%remote, agent = %hello.agent, session_id = hello.session_id, focus_generation = hello.focus_generation, "input v2 hello received");527 let lease = match sessions.claim_input_v2(528 remote,529 hello.nonce,530 hello.session_id,531 hello.slot,532 hello.focus_generation,533 ) {534 Ok(lease) => lease,535 Err(error) => {536 write_input_host_hello_v2(&mut send, &InputHostHelloV2::rejected(error.to_string()))537 .await?;538 send.finish()?;539 return Err(error);540 }541 };542 let supported =543 supported_input_capabilities(operations.capabilities(), pointer_positions.is_some());544 let negotiated = negotiated_input_capabilities_v2(&hello, &supported, lease.grants);545 let ack_enabled = negotiated.contains(&Capability::InputAck);546 let feedback_enabled = negotiated.contains(&Capability::PointerPositionFeedback);547 let visibility_feedback_enabled = negotiated.contains(&Capability::PointerVisibilityFeedback);548 let mut pointer_positions = pointer_positions549 .filter(|_| feedback_enabled)550 .map(PointerPositionTracker::subscribe);551 write_input_host_hello_v2(552 &mut send,553 &InputHostHelloV2::accepted(554 lease.session_id,555 lease.slot,556 lease.focus_generation,557 negotiated.clone(),558 ),559 )560 .await?;561 info!(%remote, session_id = lease.session_id, focus_generation = lease.focus_generation, authorization_revision = lease.authorization_revision, "input v2 client accepted");562563 let session_id = lease.session_id;564 let slot = lease.slot;565 let focus_generation = lease.focus_generation;566 let guard = InputSessionGuard::new(lease, || operations.reset());567 let session_result: Result<()> = async {568 let mut received_events = 0_u64;569 if let Some(pointer_positions) = pointer_positions.as_ref() {570 let (pointer_position, pointer_visible) = pointer_feedback_fields(571 Some(*pointer_positions.borrow()),572 visibility_feedback_enabled,573 );574 write_input_ack_v2(575 &mut send,576 &InputAckV2 {577 session_id,578 slot,579 focus_generation,580 ack: InputAck {581 sequence: 0,582 pointer_position,583 pointer_visible,584 },585 },586 )587 .await?;588 }589 loop {590 if !sessions.is_v2_focus_owner(remote, session_id, slot, focus_generation) {591 break;592 }593 tokio::select! {594 _ = sessions.session_changed.notified() => continue,595 changed = async {596 pointer_positions597 .as_mut()598 .expect("feedback branch is guarded")599 .changed()600 .await601 }, if pointer_positions.is_some() => {602 changed.context("Xwayland pointer tracker stopped")?;603 let state = *pointer_positions604 .as_mut()605 .expect("feedback branch is guarded")606 .borrow_and_update();607 let (pointer_position, pointer_visible) =608 pointer_feedback_fields(Some(state), visibility_feedback_enabled);609 write_input_ack_v2(610 &mut send,611 &InputAckV2 {612 session_id,613 slot,614 focus_generation,615 ack: InputAck {616 sequence: received_events,617 pointer_position,618 pointer_visible,619 },620 },621 )622 .await?;623 }624 event = read_input_event_v2(&mut recv) => {625 let Some(event) = event? else { break; };626 ensure!(627 event.session_id == session_id628 && event.slot == slot629 && event.focus_generation == focus_generation,630 "input event changed its accepted focus binding"631 );632 let disposition = operations.apply_v2(sessions, remote, &event, &negotiated)?;633 match disposition {634 InputDisposition::Probed => debug!(%remote, "input v2 liveness probe acknowledged"),635 InputDisposition::Observed => info!(%remote, event_type = input_event_type(&event.event), "input v2 event observed"),636 InputDisposition::Disabled => debug!(%remote, event_type = input_event_type(&event.event), "input v2 event ignored because injection is disabled"),637 638 InputDisposition::Injected => debug!(%remote, event_type = input_event_type(&event.event), "input v2 event injected"),639 InputDisposition::TextIgnored => debug!(%remote, "input v2 text event ignored"),640 }641 received_events = received_events.saturating_add(1);642 if ack_enabled {643 let pointer_state = pointer_positions.as_ref().map(|positions| *positions.borrow());644 let (pointer_position, pointer_visible) =645 pointer_feedback_fields(pointer_state, visibility_feedback_enabled);646 write_input_ack_v2(647 &mut send,648 &InputAckV2 {649 session_id,650 slot,651 focus_generation,652 ack: InputAck {653 sequence: received_events,654 pointer_position,655 pointer_visible,656 },657 },658 )659 .await?;660 }661 }662 }663 }664 Ok(())665 }666 .await;667 let reset_result = guard.finish().context("neutralizing input v2 session");668 let result = session_result.and(reset_result);669 info!(%remote, session_id, focus_generation, "input v2 client released");670 result671}672673async fn serve_audio(674 connection: Connection,675 config: HostConfig,676 sessions: &Arc<SessionRegistry>,677) -> Result<()> {678 let remote = connection.remote_id();679 let handshake_permit = sessions680 .pending_handshakes681 .try_acquire()682 .context("too many pending handshakes")?;683 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())684 .await685 .context("timed out accepting audio handshake stream")?686 .context("accepting audio handshake stream")?;687 let hello = receive_hello(&mut recv, Capability::AudioOpus).await?;688 drop(handshake_permit);689 debug!(%remote, agent = %hello.agent, "audio hello received");690 ensure!(691 hello.invitation.is_none(),692 "invitations are accepted only on the first media handshake"693 );694695 if config.audio.is_none() {696 send_rejection(&mut send, "audio is unavailable").await?;697 bail!("audio is not configured");698 }699 let maximum_datagram = connection.max_datagram_size();700 if maximum_datagram.is_none_or(|maximum| maximum < AUDIO_HEADER_LEN + MAX_AUDIO_PAYLOAD_LEN) {701 send_rejection(&mut send, "peer cannot carry v1 audio datagrams").await?;702 bail!(703 "peer audio datagram limit {:?} is below {}",704 maximum_datagram,705 AUDIO_HEADER_LEN + MAX_AUDIO_PAYLOAD_LEN706 );707 }708 let lease = match sessions.claim_audio(remote, hello.nonce) {709 Ok(lease) => lease,710 Err(error) => {711 send_rejection(&mut send, error.to_string()).await?;712 return Err(error);713 }714 };715 ensure!(716 lease.grants.contains(InvitationGrants::VIEW),717 "active Portal session lacks audio view permission"718 );719 let (mut audio_receiver, audio_task) =720 match spawn_pipewire_audio(config, lease.session_clock).await {721 Ok(source) => source,722 Err(error) => {723 send_rejection(&mut send, "audio source is unavailable").await?;724 return Err(error);725 }726 };727 let audio_task = SourceTaskGuard::new(audio_task);728 write_host_hello(729 &mut send,730 &HostHello::accepted(731 lease.session_id,732 negotiated_capabilities(&hello, AUDIO_CAPABILITIES),733 ),734 )735 .await?;736 send.finish()?;737 info!(%remote, session_id = lease.session_id, authorization_revision = lease.authorization_revision, "audio client accepted");738739 let session_result: Result<()> = async {740 loop {741 if !sessions.is_active(remote, lease.session_id) {742 break;743 }744 tokio::select! {745 _ = sessions.session_changed.notified() => continue,746 packet = audio_receiver.recv() => {747 let packet = packet.context("audio source stopped")?;748 let flags = if packet.discontinuity {749 AudioFlags::DISCONTINUITY750 } else {751 AudioFlags::NONE752 };753 let header = AudioPacketHeader::opus(754 packet.payload.len(),755 packet.sequence,756 packet.capture_timestamp_us,757 packet.pts_us,758 flags,759 )?;760 let datagram = AudioPacket::new(header, packet.payload.as_ref().to_vec())?761 .encode_datagram()?;762 match connection.send_datagram(datagram.into()) {763 Ok(()) => {}764 Err(error) => {765 // The non-waiting API bounds the QUIC datagram buffer by766 // evicting stale datagrams. Its errors mean the negotiated767 // path cannot carry the fixed v1 packet and are terminal.768 return Err(error).context("sending bounded audio datagram");769 }770 }771 }772 result = connection.closed() => {773 debug!(%remote, ?result, "audio connection closed");774 break;775 }776 }777 }778 Ok(())779 }780 .await;781 audio_task.abort_and_wait().await;782 drop(lease);783 info!(%remote, "audio client released");784 session_result785}786787pub(super) async fn send_rejection(788 send: &mut iroh::endpoint::SendStream,789 message: impl Into<String>,790) -> Result<()> {791 write_host_hello(send, &HostHello::rejected(message)).await?;792 send.finish()?;793 if tokio::time::timeout(REJECTION_DRAIN_TIMEOUT, send.stopped())794 .await795 .is_err()796 {797 debug!("timed out waiting for peer to acknowledge handshake rejection");798 }799 Ok(())800}801802pub(super) async fn send_rejection_v2(803 send: &mut iroh::endpoint::SendStream,804 message: impl Into<String>,805) -> Result<()> {806 write_host_hello_v2(send, &HostHelloV2::rejected(message)).await?;807 send.finish()?;808 if tokio::time::timeout(REJECTION_DRAIN_TIMEOUT, send.stopped())809 .await810 .is_err()811 {812 debug!("timed out waiting for peer to acknowledge control v2 rejection");813 }814 Ok(())815}816817pub(super) fn negotiated_capabilities(818 hello: &ClientHello,819 supported: &[Capability],820) -> Vec<Capability> {821 supported822 .iter()823 .copied()824 .filter(|capability| hello.capabilities.contains(capability))825 .collect()826}827828fn negotiated_input_capabilities(829 hello: &ClientHello,830 supported: &[Capability],831 grants: InvitationGrants,832) -> Vec<Capability> {833 let mut negotiated = negotiated_capabilities(hello, supported);834 negotiated.retain(|capability| input_capability_authorized(*capability, grants));835 if !negotiated.contains(&Capability::PointerPositionFeedback) {836 negotiated.retain(|capability| *capability != Capability::PointerVisibilityFeedback);837 }838 negotiated839}840841fn negotiated_input_capabilities_v2(842 hello: &InputClientHelloV2,843 supported: &[Capability],844 grants: InvitationGrants,845) -> Vec<Capability> {846 let mut negotiated = supported847 .iter()848 .copied()849 .filter(|capability| hello.capabilities.contains(capability))850 .collect::<Vec<_>>();851 negotiated.retain(|capability| input_capability_authorized(*capability, grants));852 if !negotiated.contains(&Capability::PointerPositionFeedback) {853 negotiated.retain(|capability| *capability != Capability::PointerVisibilityFeedback);854 }855 negotiated856}857858fn input_capability_authorized(capability: Capability, grants: InvitationGrants) -> bool {859 match capability {860 Capability::AbsolutePointer861 | Capability::RelativePointer862 | Capability::Keyboard863 | Capability::Text864 | Capability::PointerPositionFeedback865 | Capability::PointerVisibilityFeedback => {866 grants.contains(InvitationGrants::POINTER_KEYBOARD)867 }868 Capability::Gamepad => grants.contains(InvitationGrants::GAMEPAD),869 Capability::InputAck => {870 grants.contains(InvitationGrants::POINTER_KEYBOARD)871 || grants.contains(InvitationGrants::GAMEPAD)872 }873 Capability::VideoH264 | Capability::AudioOpus => false,874 }875}876877fn pointer_feedback_fields(878 pointer_state: Option<PointerState>,879 visibility_feedback_enabled: bool,880) -> (Option<sigil_protocol::PointerPosition>, Option<bool>) {881 match pointer_state {882 Some(state) if visibility_feedback_enabled => (state.position, Some(state.visible)),883 Some(state) => (state.position.filter(|_| state.visible), None),884 None => (None, None),885 }886}887888fn supported_input_capabilities(889 backend: &[Capability],890 pointer_feedback_available: bool,891) -> Vec<Capability> {892 let mut supported = backend.to_vec();893 if pointer_feedback_available && supported.contains(&Capability::RelativePointer) {894 supported.push(Capability::PointerPositionFeedback);895 supported.push(Capability::PointerVisibilityFeedback);896 }897 supported898}899900fn input_event_type(event: &sigil_protocol::InputEvent) -> &'static str {901 match event {902 sigil_protocol::InputEvent::Probe => "probe",903 sigil_protocol::InputEvent::MouseMove { .. } => "mouse-move",904 sigil_protocol::InputEvent::MouseMoveRelative { .. } => "mouse-move-relative",905 sigil_protocol::InputEvent::MousePositionSync { .. } => "mouse-position-sync",906 sigil_protocol::InputEvent::MouseClick { .. } => "mouse-click",907 sigil_protocol::InputEvent::MouseDown { .. } => "mouse-down",908 sigil_protocol::InputEvent::MouseUp { .. } => "mouse-up",909 sigil_protocol::InputEvent::MouseScroll { .. } => "mouse-scroll",910 sigil_protocol::InputEvent::KeyDown { .. } => "key-down",911 sigil_protocol::InputEvent::KeyUp { .. } => "key-up",912 sigil_protocol::InputEvent::KeyClick { .. } => "key-click",913 sigil_protocol::InputEvent::Text { .. } => "text",914 sigil_protocol::InputEvent::Gamepad { .. } => "gamepad-snapshot",915 }916}917918pub(super) async fn receive_hello<R>(reader: &mut R, required: Capability) -> Result<ClientHello>919where920 R: tokio::io::AsyncRead + Unpin,921{922 let hello = tokio::time::timeout(HANDSHAKE_TIMEOUT, read_client_hello(reader))923 .await924 .context("timed out waiting for client hello")??925 .context("client closed before hello")?;926 ensure!(927 hello.capabilities.contains(&required),928 "client did not offer required capability {required:?}"929 );930 Ok(hello)931}932933async fn receive_hello_unconstrained<R>(reader: &mut R) -> Result<ClientHello>934where935 R: tokio::io::AsyncRead + Unpin,936{937 tokio::time::timeout(HANDSHAKE_TIMEOUT, read_client_hello(reader))938 .await939 .context("timed out waiting for client hello")??940 .context("client closed before hello")941}942943944mod tests {945 use std::cell::Cell;946 use std::panic::{AssertUnwindSafe, catch_unwind};947 use std::sync::atomic::{AtomicBool, Ordering};948949 use super::*;950951 952 fn input_session_guard_resets_before_releasing_lease_on_drop() {953 let sessions = Arc::new(SessionRegistry::default());954 let remote = endpoint(1);955 let nonce = [1; 16];956 let _media = sessions957 .claim(remote, nonce, InvitationGrants::ALL)958 .unwrap();959 let input = sessions.claim_input(remote, nonce).unwrap();960 let resets = Cell::new(0);961 let reset_saw_lease_held = Cell::new(false);962963 {964 let _guard = InputSessionGuard::new(input, || {965 resets.set(resets.get() + 1);966 reset_saw_lease_held.set(sessions.claim_input(remote, nonce).is_err());967 Ok(())968 });969 }970971 assert_eq!(resets.get(), 1);972 assert!(reset_saw_lease_held.get());973 assert!(sessions.claim_input(remote, nonce).is_ok());974 }975976 977 fn input_session_guard_finish_propagates_reset_error_once() {978 let sessions = Arc::new(SessionRegistry::default());979 let remote = endpoint(1);980 let nonce = [2; 16];981 let _media = sessions982 .claim(remote, nonce, InvitationGrants::ALL)983 .unwrap();984 let input = sessions.claim_input(remote, nonce).unwrap();985 let resets = Cell::new(0);986987 let error = InputSessionGuard::new(input, || {988 resets.set(resets.get() + 1);989 Err(anyhow::anyhow!("reset failed"))990 })991 .finish()992 .unwrap_err();993994 assert_eq!(error.to_string(), "reset failed");995 assert_eq!(resets.get(), 1);996 assert!(sessions.claim_input(remote, nonce).is_ok());997 }998999 1000 fn input_session_guard_resets_before_releasing_lease_during_unwind() {1001 let sessions = Arc::new(SessionRegistry::default());1002 let remote = endpoint(1);1003 let nonce = [3; 16];1004 let _media = sessions1005 .claim(remote, nonce, InvitationGrants::ALL)1006 .unwrap();1007 let input = sessions.claim_input(remote, nonce).unwrap();1008 let reset_saw_lease_held = Cell::new(false);10091010 let panic_result = catch_unwind(AssertUnwindSafe(|| {1011 let _guard = InputSessionGuard::new(input, || {1012 reset_saw_lease_held.set(sessions.claim_input(remote, nonce).is_err());1013 Ok(())1014 });1015 panic!("simulate input task panic");1016 }));10171018 assert!(panic_result.is_err());1019 assert!(reset_saw_lease_held.get());1020 assert!(sessions.claim_input(remote, nonce).is_ok());1021 }10221023 1024 async fn input_session_guard_resets_before_releasing_lease_when_task_is_cancelled() {1025 let sessions = Arc::new(SessionRegistry::default());1026 let remote = endpoint(1);1027 let nonce = [4; 16];1028 let _media = sessions1029 .claim(remote, nonce, InvitationGrants::ALL)1030 .unwrap();1031 let input = sessions.claim_input(remote, nonce).unwrap();1032 let reset_saw_lease_held = Arc::new(AtomicBool::new(false));1033 let task_sessions = Arc::clone(&sessions);1034 let task_reset_saw_lease_held = Arc::clone(&reset_saw_lease_held);1035 let (armed_tx, armed_rx) = tokio::sync::oneshot::channel();10361037 let task = tokio::spawn(async move {1038 let _guard = InputSessionGuard::new(input, move || {1039 task_reset_saw_lease_held.store(1040 task_sessions.claim_input(remote, nonce).is_err(),1041 Ordering::SeqCst,1042 );1043 Ok(())1044 });1045 armed_tx.send(()).unwrap();1046 std::future::pending::<()>().await;1047 });10481049 armed_rx.await.unwrap();1050 task.abort();1051 assert!(task.await.unwrap_err().is_cancelled());1052 assert!(reset_saw_lease_held.load(Ordering::SeqCst));1053 assert!(sessions.claim_input(remote, nonce).is_ok());1054 }10551056 1057 fn capability_negotiation_is_an_exact_intersection() {1058 let hello = ClientHello::new(1059 "test",1060 [0; 16],1061 vec![1062 Capability::AbsolutePointer,1063 Capability::RelativePointer,1064 Capability::Keyboard,1065 Capability::Text,1066 Capability::Gamepad,1067 Capability::VideoH264,1068 Capability::AudioOpus,1069 ],1070 );1071 assert_eq!(1072 negotiated_capabilities(1073 &hello,1074 &[1075 Capability::RelativePointer,1076 Capability::Keyboard,1077 Capability::Gamepad,1078 ]1079 ),1080 vec![1081 Capability::RelativePointer,1082 Capability::Keyboard,1083 Capability::Gamepad,1084 ]1085 );1086 assert!(negotiated_capabilities(&hello, &[Capability::InputAck]).is_empty());1087 assert_eq!(1088 negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1089 vec![Capability::VideoH264]1090 );1091 assert_eq!(1092 negotiated_capabilities(&hello, AUDIO_CAPABILITIES),1093 vec![Capability::AudioOpus]1094 );1095 }10961097 1098 fn enrollment_grants_are_a_strict_input_capability_ceiling() {1099 let hello = ClientHello::new(1100 "test",1101 [0; 16],1102 vec![1103 Capability::RelativePointer,1104 Capability::AbsolutePointer,1105 Capability::Keyboard,1106 Capability::Text,1107 Capability::Gamepad,1108 Capability::InputAck,1109 Capability::PointerPositionFeedback,1110 Capability::PointerVisibilityFeedback,1111 ],1112 );1113 let supported = hello.capabilities.clone();11141115 assert!(1116 negotiated_input_capabilities(&hello, &supported, InvitationGrants::VIEW).is_empty()1117 );1118 assert_eq!(1119 negotiated_input_capabilities(1120 &hello,1121 &supported,1122 InvitationGrants::VIEW.union(InvitationGrants::POINTER_KEYBOARD),1123 ),1124 vec![1125 Capability::RelativePointer,1126 Capability::AbsolutePointer,1127 Capability::Keyboard,1128 Capability::Text,1129 Capability::InputAck,1130 Capability::PointerPositionFeedback,1131 Capability::PointerVisibilityFeedback,1132 ]1133 );1134 assert_eq!(1135 negotiated_input_capabilities(1136 &hello,1137 &supported,1138 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),1139 ),1140 vec![Capability::Gamepad, Capability::InputAck]1141 );1142 }11431144 1145 fn pointer_feedback_is_advertised_only_with_tracker_and_relative_input() {1146 assert_eq!(1147 supported_input_capabilities(&[Capability::RelativePointer], false),1148 vec![Capability::RelativePointer]1149 );1150 assert_eq!(1151 supported_input_capabilities(&[Capability::RelativePointer], true),1152 vec![1153 Capability::RelativePointer,1154 Capability::PointerPositionFeedback,1155 Capability::PointerVisibilityFeedback,1156 ]1157 );1158 assert_eq!(1159 supported_input_capabilities(&[Capability::InputAck], true),1160 vec![Capability::InputAck]1161 );1162 }11631164 1165 fn old_pointer_feedback_client_gets_legacy_host_hello_and_ack_shape() {1166 let hello = ClientHello::new(1167 "old-client",1168 [0; 16],1169 vec![1170 Capability::RelativePointer,1171 Capability::PointerPositionFeedback,1172 ],1173 );1174 let supported = supported_input_capabilities(&[Capability::RelativePointer], true);1175 let negotiated = negotiated_input_capabilities(&hello, &supported, InvitationGrants::ALL);1176 assert_eq!(1177 negotiated,1178 vec![1179 Capability::RelativePointer,1180 Capability::PointerPositionFeedback,1181 ]1182 );1183 assert_eq!(1184 serde_json::to_string(&HostHello::accepted(7, negotiated)).unwrap(),1185 r#"{"version":1,"accepted":true,"session_id":7,"capabilities":["relative_pointer","pointer_position_feedback"],"message":null}"#1186 );11871188 let position = sigil_protocol::PointerPosition { x: 320, y: 200 };1189 let (pointer_position, pointer_visible) = pointer_feedback_fields(1190 Some(PointerState {1191 position: Some(position),1192 visible: true,1193 }),1194 false,1195 );1196 assert_eq!(1197 serde_json::to_string(&InputAck {1198 sequence: 1,1199 pointer_position,1200 pointer_visible,1201 })1202 .unwrap(),1203 r#"{"sequence":1,"pointer_position":{"x":320,"y":200}}"#1204 );12051206 let (pointer_position, pointer_visible) = pointer_feedback_fields(1207 Some(PointerState {1208 position: Some(position),1209 visible: false,1210 }),1211 false,1212 );1213 assert_eq!(1214 serde_json::to_string(&InputAck {1215 sequence: 1,1216 pointer_position,1217 pointer_visible,1218 })1219 .unwrap(),1220 r#"{"sequence":1}"#1221 );1222 }12231224 1225 fn pointer_visibility_feedback_requires_position_feedback() {1226 let visibility_only = ClientHello::new(1227 "invalid-client",1228 [0; 16],1229 vec![Capability::PointerVisibilityFeedback],1230 );1231 let supported = supported_input_capabilities(&[Capability::RelativePointer], true);1232 assert!(1233 negotiated_input_capabilities(&visibility_only, &supported, InvitationGrants::ALL)1234 .is_empty()1235 );12361237 let upgraded = ClientHello::new(1238 "upgraded-client",1239 [0; 16],1240 vec![1241 Capability::RelativePointer,1242 Capability::PointerPositionFeedback,1243 Capability::PointerVisibilityFeedback,1244 ],1245 );1246 assert_eq!(1247 negotiated_input_capabilities(&upgraded, &supported, InvitationGrants::ALL),1248 supported1249 );12501251 let position = sigil_protocol::PointerPosition { x: 320, y: 200 };1252 assert_eq!(1253 pointer_feedback_fields(1254 Some(PointerState {1255 position: Some(position),1256 visible: false,1257 }),1258 true,1259 ),1260 (Some(position), Some(false))1261 );1262 }1263}