Skip to content

Close the review's protocol-parity and input-correctness findings #28

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
whoisaldo merged 1 commit into from
Aug 27, 2026
Merged

Close the review's protocol-parity and input-correctness findings #28

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 36 additions & 8 deletions host/src/input/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,24 +110,32 @@ pub fn scroll_to_wheel(delta_px: i16) -> i32 {
i32::from(delta_px) * 3
}

/// Drops duplicate began/ended events (the client sends edges twice for loss
/// tolerance) while letting distinct events through in order.
/// Number of recent edge ids remembered. The client sends each press/release
/// twice for loss tolerance, and Wi-Fi aggregation reorders freely, so the two
/// copies of one edge routinely arrive with other edges between them. A single
/// remembered id could not survive that: began(N), ended(N+1), began(N) walked
/// the slot N -> N+1 -> N and let the replayed press through, injecting a
/// double click where the user tapped once.
const RECENT_EDGE_IDS: usize = 8;

/// Drops duplicate began/ended events while letting distinct events through.
#[derive(Debug, Default)]
pub struct EventDeduper {
last_edge_id: Option<u32>,
recent: [Option<u32>; RECENT_EDGE_IDS],
next: usize,
}

impl EventDeduper {
/// True if the event should be processed.
pub fn accept(&mut self, event: &InputEvent) -> bool {
match event.phase {
PHASE_BEGAN | PHASE_ENDED | PHASE_CANCELLED => {
if self.last_edge_id == Some(event.event_id) {
false
} else {
self.last_edge_id = Some(event.event_id);
true
if self.recent.contains(&Some(event.event_id)) {
return false;
}
self.recent[self.next] = Some(event.event_id);
self.next = (self.next + 1) % RECENT_EDGE_IDS;
true
}
// Moves are idempotent-ish; duplicates are harmless.
_ => true,
Expand Down Expand Up @@ -388,6 +396,26 @@ mod tests {
assert_eq!(recorder::take().len(), 4);
}

#[test]
fn reordered_duplicate_edges_still_dedupe() {
// The wire sends each edge twice; the network may interleave them.
// began(7), ended(8), began(7), ended(8) must inject one press and
// one release, not two of each.
let mut dedupe = EventDeduper::default();
let mut began = event(KIND_TOUCH, PHASE_BEGAN, 0, 0);
began.event_id = 7;
let mut ended = event(KIND_TOUCH, PHASE_ENDED, 0, 0);
ended.event_id = 8;

assert!(dedupe.accept(&began));
assert!(dedupe.accept(&ended));
assert!(
!dedupe.accept(&began),
"the replayed press must not click again"
);
assert!(!dedupe.accept(&ended));
}

#[test]
fn edge_duplicates_are_dropped_moves_pass() {
let mut dedupe = EventDeduper::default();
Expand Down
8 changes: 6 additions & 2 deletions host/src/transport/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,9 +150,13 @@ pub async fn start_sender(
match classify(datagram) {
Classified::Control(_) => {
match eternal_wire::v2::control::parse_control(datagram) {
Ok((_, message)) => {
Ok((header, message)) => {
let mut actions = shared.session.lock().handle_control(
src, message, &config, Instant::now(),
src,
header.session_id,
message,
&config,
Instant::now(),
);
if let Some((session_id, event)) = actions.input.take() {
if let Some(geometry) = *shared.capture_geometry.lock()
Expand Down
139 changes: 112 additions & 27 deletions host/src/transport/session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,14 +132,31 @@ impl Session {
}
}

/// Feed one inbound control message. `now` is injected for testability.
/// Feed one inbound control message. `header_session_id` is the id the
/// datagram claims; every message except HELLO2 must match the live
/// session's. `now` is injected for testability.
pub fn handle_control(
&mut self,
source: SocketAddr,
header_session_id: u32,
message: ControlMessage,
config: &impl ConfigSource,
now: Instant,
) -> Actions {
// HELLO2 is the only message that legitimately carries no session id
// (it is asking for one). Everything else is authenticated by the id
// the handshake minted, not just by source IP: an IP is shared by
// every process on the device and by anything behind the same NAT, and
// is trivially spoofed on the local link. Without this, a stray
// INPUT_EVENT moved the mouse and clicked, a stray RECEIVER_REPORT
// steered the bitrate, and a late BYE from a superseded session tore
// down the one that replaced it.
if !matches!(message, ControlMessage::Hello2(_)) {
match self.active.as_ref() {
Some(session) if session.session_id == header_session_id => {}
_ => return Actions::default(),
}
}
match message {
ControlMessage::Hello2(hello) => self.handle_hello(source, hello, config, now),
ControlMessage::Bye(reason) => self.handle_bye(source, reason),
Expand All @@ -152,6 +169,19 @@ impl Session {
}
}

/// Test helper: send as the currently-connected client.
#[cfg(test)]
fn handle_control_authed(
&mut self,
source: SocketAddr,
message: ControlMessage,
config: &impl ConfigSource,
now: Instant,
) -> Actions {
let id = self.session_id().unwrap_or(0);
self.handle_control(source, id, message, config, now)
}

fn handle_input(&mut self, source: SocketAddr, event: InputEvent, now: Instant) -> Actions {
let mut actions = Actions::default();
let Some(session) = self.active.as_mut() else {
Expand Down Expand Up @@ -510,7 +540,7 @@ mod tests {
let now = Instant::now();
let peer = addr([10, 0, 0, 5], 50000);

let actions = session.handle_control(peer, hello(1, 50000), &TestConfig, now);
let actions = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);

assert_eq!(actions.replies.len(), 1);
let ack = parse_ack(&actions.replies[0].1);
Expand All @@ -535,10 +565,10 @@ mod tests {
let now = Instant::now();
let peer = addr([10, 0, 0, 5], 50000);

let first = session.handle_control(peer, hello(1, 50000), &TestConfig, now);
let first = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);
let first_id = parse_ack(&first.replies[0].1).session_id;

let dup = session.handle_control(peer, hello(1, 50000), &TestConfig, now);
let dup = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);
assert_eq!(parse_ack(&dup.replies[0].1).session_id, first_id);
assert!(dup.new_target.is_none(), "retransmit must not re-target");
assert!(!dup.force_idr);
Expand All @@ -549,12 +579,12 @@ mod tests {
let mut session = Session::new(1234);
let now = Instant::now();
let peer = addr([10, 0, 0, 5], 50000);
let first = session.handle_control(peer, hello(1, 50000), &TestConfig, now);
let first = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);
let first_id = parse_ack(&first.replies[0].1).session_id;

// App relaunched: new ephemeral source port, new nonce.
let relaunched = addr([10, 0, 0, 5], 50101);
let second = session.handle_control(relaunched, hello(2, 50101), &TestConfig, now);
let second = session.handle_control_authed(relaunched, hello(2, 50101), &TestConfig, now);
let second_id = parse_ack(&second.replies[0].1).session_id;

assert_ne!(
Expand All @@ -569,15 +599,15 @@ mod tests {
fn different_ip_is_rejected_busy_while_active() {
let mut session = Session::new(1234);
let now = Instant::now();
session.handle_control(
session.handle_control_authed(
addr([10, 0, 0, 5], 50000),
hello(1, 50000),
&TestConfig,
now,
);

let intruder = addr([10, 0, 0, 9], 40000);
let actions = session.handle_control(intruder, hello(9, 40000), &TestConfig, now);
let actions = session.handle_control_authed(intruder, hello(9, 40000), &TestConfig, now);
let ack = parse_ack(&actions.replies[0].1);
assert_eq!(ack.status, HelloStatus::Busy);
assert_eq!(ack.session_id, 0);
Expand All @@ -604,7 +634,7 @@ mod tests {
refresh_hz: 60,
device_name: String::new(),
});
let actions = session.handle_control(peer, msg, &TestConfig, now);
let actions = session.handle_control_authed(peer, msg, &TestConfig, now);
assert_eq!(
parse_ack(&actions.replies[0].1).status,
HelloStatus::VersionUnsupported
Expand All @@ -617,9 +647,9 @@ mod tests {
let mut session = Session::new(1234);
let now = Instant::now();
let peer = addr([10, 0, 0, 5], 50000);
session.handle_control(peer, hello(1, 50000), &TestConfig, now);
session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);

let bye = session.handle_control(
let bye = session.handle_control_authed(
peer,
ControlMessage::Bye(ByeReason::UserDisconnect),
&TestConfig,
Expand All @@ -629,7 +659,7 @@ mod tests {
assert!(!session.is_active());

// Re-establish, then let liveness lapse.
session.handle_control(peer, hello(2, 50000), &TestConfig, now);
session.handle_control_authed(peer, hello(2, 50000), &TestConfig, now);
let expired = session.tick(&TestConfig, false, now + LIVENESS_TIMEOUT);
assert!(expired.client_lost);
assert!(!session.is_active());
Expand All @@ -640,12 +670,12 @@ mod tests {
let mut session = Session::new(1234);
let now = Instant::now();
let peer = addr([10, 0, 0, 5], 50000);
session.handle_control(peer, hello(1, 50000), &TestConfig, now);
session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);

let later = now + Duration::from_millis(2500);
let mut report = ReceiverReport::default();
report.frames_complete = 99;
session.handle_control(
session.handle_control_authed(
peer,
ControlMessage::ReceiverReport(report),
&TestConfig,
Expand All @@ -667,18 +697,18 @@ mod tests {
let mut session = Session::new(1234);
let now = Instant::now();
let peer = addr([10, 0, 0, 5], 50000);
session.handle_control(peer, hello(1, 50000), &TestConfig, now);
session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);

let request = ControlMessage::KeyframeRequest(KeyframeRequest {
stream_epoch: 7,
last_complete_seq: 10,
reason: KeyframeReason::GapLoss,
});

let first = session.handle_control(peer, request.clone(), &TestConfig, now);
let first = session.handle_control_authed(peer, request.clone(), &TestConfig, now);
assert!(first.force_idr);

let spammed = session.handle_control(
let spammed = session.handle_control_authed(
peer,
request.clone(),
&TestConfig,
Expand All @@ -689,8 +719,12 @@ mod tests {
"requests inside the window are coalesced"
);

let granted_again =
session.handle_control(peer, request, &TestConfig, now + Duration::from_millis(600));
let granted_again = session.handle_control_authed(
peer,
request,
&TestConfig,
now + Duration::from_millis(600),
);
assert!(granted_again.force_idr);
}

Expand Down Expand Up @@ -719,8 +753,8 @@ mod tests {
let peer = addr([10, 0, 0, 5], 50000);

// View-only session: input is dropped.
session.handle_control(peer, hello(1, 50000), &TestConfig, now);
let dropped = session.handle_control(peer, input_event(1), &TestConfig, now);
session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);
let dropped = session.handle_control_authed(peer, input_event(1), &TestConfig, now);
assert!(
dropped.input.is_none(),
"view-only sessions must not inject"
Expand All @@ -732,17 +766,68 @@ mod tests {
_ => unreachable!(),
};
wants.feature_caps = FEATURE_WANTS_INPUT;
session.handle_control(peer, ControlMessage::Hello2(wants), &TestConfig, now);
session.handle_control_authed(peer, ControlMessage::Hello2(wants), &TestConfig, now);

let relayed = session.handle_control(peer, input_event(2), &TestConfig, now);
let relayed = session.handle_control_authed(peer, input_event(2), &TestConfig, now);
assert!(relayed.input.is_some());

// A different IP can't inject into this session.
let intruder = addr([10, 0, 0, 9], 40000);
let foreign = session.handle_control(intruder, input_event(3), &TestConfig, now);
let foreign = session.handle_control_authed(intruder, input_event(3), &TestConfig, now);
assert!(foreign.input.is_none());
}

#[test]
fn control_messages_need_the_negotiated_session_id() {
let mut session = Session::new(1234);
let now = Instant::now();
let peer = addr([10, 0, 0, 5], 50000);
let mut wants = match hello(1, 50000) {
ControlMessage::Hello2(h) => h,
_ => unreachable!(),
};
wants.feature_caps = FEATURE_WANTS_INPUT;
session.handle_control(peer, 0, ControlMessage::Hello2(wants), &TestConfig, now);
let live = session.session_id().unwrap();

// Same source address, wrong session id: another process on that
// device, or a spoofed packet on the local link. It must not move the
// mouse, steer the bitrate, or end the session.
let wrong = live.wrapping_add(1);
assert!(session
.handle_control(peer, wrong, input_event(1), &TestConfig, now)
.input
.is_none());
assert!(session
.handle_control(
peer,
wrong,
ControlMessage::ReceiverReport(ReceiverReport::default()),
&TestConfig,
now
)
.report
.is_none());
let bye = session.handle_control(
peer,
wrong,
ControlMessage::Bye(ByeReason::UserDisconnect),
&TestConfig,
now,
);
assert!(
!bye.client_lost,
"a foreign BYE must not tear the session down"
);
assert!(session.is_active());

// The real client still works.
assert!(session
.handle_control(peer, live, input_event(2), &TestConfig, now)
.input
.is_some());
}

#[test]
fn input_extends_liveness() {
let mut session = Session::new(1234);
Expand All @@ -753,11 +838,11 @@ mod tests {
_ => unreachable!(),
};
wants.feature_caps = FEATURE_WANTS_INPUT;
session.handle_control(peer, ControlMessage::Hello2(wants), &TestConfig, now);
session.handle_control_authed(peer, ControlMessage::Hello2(wants), &TestConfig, now);

// A drag mid-window keeps the session alive past the original deadline.
let later = now + Duration::from_millis(2500);
session.handle_control(peer, input_event(1), &TestConfig, later);
session.handle_control_authed(peer, input_event(1), &TestConfig, later);
let ticked = session.tick(&TestConfig, false, now + Duration::from_millis(4000));
assert!(!ticked.client_lost);
}
Expand All @@ -769,7 +854,7 @@ mod tests {
assert!(session.tick(&TestConfig, true, now).replies.is_empty());

let peer = addr([10, 0, 0, 5], 50000);
session.handle_control(peer, hello(1, 50000), &TestConfig, now);
session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now);
let ticked = session.tick(&TestConfig, true, now + Duration::from_millis(100));
assert_eq!(ticked.replies.len(), 1);
let (_, message) = eternal_wire::v2::control::parse_control(&ticked.replies[0].1).unwrap();
Expand Down
Loading
Loading