`claim` holds the legacy registry lock while checking `live_control_leases`, but `claim_v2_authorized` released it before taking the v2 lock. Interleaved, a legacy exclusive client and a v2 viewer both cleared their mutual-exclusion check and were admitted together, which breaks the one-generation invariant the whole design rests on. Hold `active` across the v2 mutation so both paths take the same locks in the same order. `FocusArbiter::command_times` was reclaimed only by `release`, which cannot find a viewer that revocation or adaptive detachment already removed from the roster. Those timestamps then outlived the host, against the boundedness the arbiter documents and unlike `admission_rates`, which is swept on every admission. Retire the candidate explicitly on both exits. `claim_moq` passed the capability's own host field as `verify_binding`'s expected host, so that leg was a no-op. It is unexploitable today because the capability is host-issued and stored server-side, but it is exactly the check that matters once a capability can arrive over the relay path this work is building toward. Carry the issuing host identity on the pending attachment. The admission interleaving is too narrow to reproduce by racing threads, so the regression test asserts the lock discipline that closes it directly.
- commit
- dc297376ba062212bfaae208d2711e45473cf225
- parent
- 005d35a4afd3e5cfa0b7dc86c5c0f16b5969f64a
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 5 | + |
| M | crates/ | 263 | +++++++++++++++++++++++++++++- |
@@ -116,6 +116,11 @@ impl FocusArbiter {
116116 Ok(())117117 }118118 119+ 120+ pub(super) fn tracked_command_candidates(&self) -> usize {121+ self.command_times.len()122+ }123+119124 pub(super) fn retire_candidate(&mut self, candidate: &FocusCandidate) {120125 self.command_times121126 .remove(&(candidate.presence_id.clone(), candidate.session_id));@@ -58,6 +58,7 @@ struct PendingMoqAttachment {
5858 closed: tokio::sync::oneshot::Sender<()>,5959 telemetry: Arc<MediaV3Telemetry>,6060 subscription_capability: Option<SignedSubscriptionCapability>,61+ expected_host: Option<[u8; 32]>,6162 }6263 6364 impl std::fmt::Debug for PendingMoqAttachment {@@ -646,13 +647,16 @@ impl SessionRegistry {
646647 == authorized.committed_revision,647648 "viewer admission used a stale authorization revision"648649 );649- ensure!(650- self.active651- .lock()652- .expect("session registry poisoned")653- .is_none(),654- "host already has an active legacy client"655- );650+ // Hold the legacy registry lock across the whole v2 admission. `claim`651+ // takes `active` and then `v2_state`; releasing `active` here first652+ // would let a legacy client and a v2 viewer both pass their mutual653+ // exclusion check and be admitted together.654+ // Hold the legacy registry lock across the whole v2 admission. `claim`655+ // takes `active` and then `v2_state`; releasing `active` here first656+ // would let a legacy client and a v2 viewer both pass their mutual657+ // exclusion check and be admitted together.658+ let active = self.active.lock().expect("session registry poisoned");659+ ensure!(active.is_none(), "host already has an active legacy client");656660 let mut state = self.v2_state.lock().expect("v2 session state poisoned");657661 let replacing = state.viewers.contains_key(&remote);658662 ensure!(@@ -745,6 +749,7 @@ impl SessionRegistry {
745749 replacement_transition,746750 };747751 drop(state);752+ drop(active);748753 self.session_changed.notify_waiters();749754 info!(750755 viewer_handle,@@ -1085,6 +1090,10 @@ impl SessionRegistry {
10851090 .remove(&mutation.peer)10861091 .expect("validated authorization viewer disappeared");10871092 viewer.snapshots.send_replace(None);1093+ // `release` reclaims focus rate-limit state only for a viewer it1094+ // still finds in the roster. Revocation removes it here, so retire1095+ // the candidate now or its timestamps outlive the host.1096+ state.focus.retire_candidate(&focus_candidate);10881097 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");10891098 pending.remove(&(mutation.peer, viewer.session.session_id));10901099 } else if let Some(grants) = mutation.current_grants {@@ -1482,6 +1491,7 @@ impl SessionRegistry {
14821491 closed,14831492 telemetry,14841493 subscription_capability: None,1494+ expected_host: None,14851495 },14861496 );14871497 Ok(MoqAttachmentWait {@@ -1497,6 +1507,7 @@ impl SessionRegistry {
14971507 broadcast_name: String,14981508 broadcast: BroadcastConsumer,14991509 subscription_capability: SignedSubscriptionCapability,1510+ expected_host: [u8; 32],15001511 ) -> Result<MoqAttachmentWait> {15011512 let state = self.v2_state.lock().expect("v2 session state poisoned");15021513 let session = state@@ -1506,7 +1517,9 @@ impl SessionRegistry {
15061517 .map(|viewer| &viewer.session)15071518 .context("authenticated MoQ expectation does not match the active v2 session")?;15081519 ensure!(1509- subscription_capability.claims.media_generation_id == session.media_generation_id1520+ subscription_capability.claims.host_node_id == expected_host1521+ && subscription_capability.claims.media_generation_id1522+ == session.media_generation_id15101523 && subscription_capability.claims.subscriber_endpoint_id == *remote.as_bytes()15111524 && subscription_capability15121525 .claims@@ -1535,6 +1548,7 @@ impl SessionRegistry {
15351548 closed,15361549 telemetry,15371550 subscription_capability: Some(subscription_capability),1551+ expected_host: Some(expected_host),15381552 },15391553 );15401554 Ok(MoqAttachmentWait {@@ -1588,9 +1602,12 @@ impl SessionRegistry {
15881602 .subscription_capability15891603 .as_ref()15901604 .context("v2 MoQ attachment has no subscription capability")?;1605+ let expected_host = attachment1606+ .expected_host1607+ .context("v2 MoQ attachment has no expected host identity")?;15911608 capability15921609 .verify_binding(1593- capability.claims.host_node_id,1610+ expected_host,15941611 session.media_generation_id,15951612 *remote.as_bytes(),15961613 capability.claims.tracks,@@ -1637,6 +1654,15 @@ impl SessionRegistry {
16371654 )16381655 }16391656 1657+ 1658+ pub(super) fn focus_rate_limit_entries(&self) -> usize {1659+ self.v2_state1660+ .lock()1661+ .expect("v2 session state poisoned")1662+ .focus1663+ .tracked_command_candidates()1664+ }1665+16401666 fn release(&self, remote: EndpointId, session_id: u64) {16411667 let mut active = self.active.lock().expect("session registry poisoned");16421668 if let Some(session) = active.as_mut()@@ -1762,8 +1788,15 @@ impl SessionRegistry {
17621788 ),17631789 "focused viewer must be neutralized before adaptive detachment"17641790 );1791+ let retired = FocusCandidate {1792+ presence_id: viewer.presence_id.clone(),1793+ session_id,1794+ };17651795 viewer.snapshots.send_replace(None);17661796 state.viewers.remove(&remote);1797+ // Detaching bypasses `release`'s roster lookup, so reclaim this1798+ // viewer's focus rate-limit timestamps explicitly.1799+ state.focus.retire_candidate(&retired);17671800 if state.viewers.is_empty() {17681801 state.media = None;17691802 }@@ -2288,6 +2321,186 @@ mod tests {
22882321 }22892322 }22902323 2324+ // Regression: `claim_v2_authorized` used to drop the legacy registry lock2325+ // before taking the v2 lock, so a legacy client and a v2 viewer could both2326+ // clear their mutual-exclusion check and be admitted at the same time.2327+ //2328+ // The interleaving is too narrow to reproduce by racing threads, so assert2329+ // the lock discipline that closes it directly: while v2 admission is parked2330+ // on the v2 lock it must still be holding the legacy registry lock. Under2331+ // the old ordering it had already released it, and `claim` could walk in.2332+ 2333+ fn v2_admission_holds_the_legacy_registry_lock_across_the_v2_mutation() {2334+ let sessions = Arc::new(SessionRegistry::default());2335+ let blocked = Arc::clone(&sessions);2336+ let state_guard = sessions.v2_state.lock().expect("v2 session state poisoned");2337+2338+ let admission = std::thread::spawn(move || {2339+ blocked.claim_v2_authorized(2340+ endpoint(2),2341+ [2; 16],2342+ authorized_viewer("viewer-0000000000000002", InvitationGrants::ALL),2343+ )2344+ });2345+ // Generous settle time: the thread is now parked on `v2_state`.2346+ std::thread::sleep(Duration::from_millis(500));2347+ assert!(2348+ sessions.active.try_lock().is_err(),2349+ "v2 admission released the legacy registry lock before mutating v2 state, \2350+ which lets `claim` admit a legacy client concurrently"2351+ );2352+2353+ drop(state_guard);2354+ // Hold the lease: dropping it would release the viewer and make the2355+ // exclusion assertion below vacuous.2356+ let _viewer = admission2357+ .join()2358+ .expect("v2 admission thread panicked")2359+ .unwrap_or_else(|error| {2360+ panic!("v2 admission must succeed once the v2 lock is free: {error:#}")2361+ });2362+ assert!(2363+ sessions2364+ .claim(endpoint(1), [1; 16], InvitationGrants::ALL)2365+ .is_err(),2366+ "a legacy client must be refused while a v2 viewer holds a control lease"2367+ );2368+ }2369+2370+ 2371+ fn legacy_and_v2_admission_stay_mutually_exclusive_under_contention() {2372+ for round in 0..300_u16 {2373+ let sessions = Arc::new(SessionRegistry::default());2374+ let legacy_registry = Arc::clone(&sessions);2375+ let v2_registry = Arc::clone(&sessions);2376+ let gate = Arc::new(std::sync::Barrier::new(2));2377+ let legacy_gate = Arc::clone(&gate);2378+ let v2_gate = Arc::clone(&gate);2379+2380+ let legacy = std::thread::spawn(move || {2381+ legacy_gate.wait();2382+ legacy_registry2383+ .claim(endpoint(1), [1; 16], InvitationGrants::ALL)2384+ .ok()2385+ });2386+ let v2 = std::thread::spawn(move || {2387+ v2_gate.wait();2388+ v2_registry2389+ .claim_v2_authorized(2390+ endpoint(2),2391+ [2; 16],2392+ authorized_viewer("viewer-0000000000000002", InvitationGrants::ALL),2393+ )2394+ .ok()2395+ });2396+ let legacy = legacy.join().expect("legacy admission thread panicked");2397+ let v2 = v2.join().expect("v2 admission thread panicked");2398+ assert!(2399+ legacy.is_none() || v2.is_none(),2400+ "round {round} admitted a legacy client and a v2 viewer at once"2401+ );2402+ assert!(2403+ legacy.is_some() || v2.is_some(),2404+ "round {round} admitted neither client"2405+ );2406+ }2407+ }2408+2409+ // Regression: focus rate-limit timestamps were reclaimed only by `release`,2410+ // which cannot find a viewer that revocation or adaptive detachment already2411+ // removed from the roster.2412+ 2413+ fn focus_rate_limit_state_is_reclaimed_on_every_viewer_exit() {2414+ let request = FocusCommandV2 {2415+ request_id: 1,2416+ action: FocusCommandActionV2::Request,2417+ slot: ControllerSlot::ZERO,2418+ expected_revision: 1,2419+ expected_focus_generation: None,2420+ expected_proposal_id: None,2421+ };2422+2423+ // Ordinary disconnect.2424+ let sessions = Arc::new(SessionRegistry::default());2425+ let remote = endpoint(1);2426+ let lease = sessions2427+ .claim_v2_authorized(2428+ remote,2429+ [1; 16],2430+ authorized_viewer("viewer-0000000000000001", InvitationGrants::ALL),2431+ )2432+ .unwrap();2433+ sessions2434+ .apply_focus_command(remote, lease.session_id, &request)2435+ .unwrap();2436+ assert_eq!(sessions.focus_rate_limit_entries(), 1);2437+ drop(lease);2438+ assert_eq!(2439+ sessions.focus_rate_limit_entries(),2440+ 0,2441+ "disconnect must reclaim focus rate-limit state"2442+ );2443+2444+ // Adaptive slow-viewer detachment.2445+ let sessions = Arc::new(SessionRegistry::default());2446+ let holder = endpoint(1);2447+ let spectator = endpoint(2);2448+ let holder_lease = sessions2449+ .claim_v2_authorized(2450+ holder,2451+ [1; 16],2452+ authorized_viewer("viewer-0000000000000001", InvitationGrants::ALL),2453+ )2454+ .unwrap();2455+ let spectator_lease = sessions2456+ .claim_v2_authorized(2457+ spectator,2458+ [2; 16],2459+ authorized_viewer("viewer-0000000000000002", InvitationGrants::ALL),2460+ )2461+ .unwrap();2462+ let mut spectator_request = request.clone();2463+ spectator_request.expected_revision = sessions2464+ .v2_revision(spectator, spectator_lease.session_id)2465+ .unwrap();2466+ sessions2467+ .apply_focus_command(spectator, spectator_lease.session_id, &spectator_request)2468+ .unwrap();2469+ assert_eq!(sessions.focus_rate_limit_entries(), 1);2470+ // The spectator now holds focus, so release it before detaching.2471+ let mut release = spectator_request.clone();2472+ release.action = FocusCommandActionV2::Release;2473+ release.expected_revision = sessions2474+ .v2_revision(spectator, spectator_lease.session_id)2475+ .unwrap();2476+ release.expected_focus_generation = match sessions.v2_state.lock().unwrap().focus.state() {2477+ FocusStateV2::Held {2478+ focus_generation, ..2479+ } => Some(*focus_generation),2480+ other => panic!("expected held focus, got {other:?}"),2481+ };2482+ let effect = sessions2483+ .apply_focus_command(spectator, spectator_lease.session_id, &release)2484+ .unwrap();2485+ if let Some(transition) = effect.neutralization {2486+ sessions2487+ .complete_v2_focus_transition(transition.transition_id)2488+ .unwrap();2489+ }2490+ assert!(2491+ sessions2492+ .disconnect_v2_viewer(spectator, spectator_lease.session_id)2493+ .unwrap()2494+ );2495+ assert_eq!(2496+ sessions.focus_rate_limit_entries(),2497+ 0,2498+ "adaptive detachment must reclaim focus rate-limit state"2499+ );2500+ drop(spectator_lease);2501+ drop(holder_lease);2502+ }2503+22912504 22922505 fn v2_admits_bounded_viewers_and_publishes_personalized_complete_rosters() {22932506 let sessions = Arc::new(SessionRegistry::new(3));@@ -2649,6 +2862,36 @@ mod tests {
26492862 assert!(sessions.claim_moq(endpoint(1)).is_err());26502863 }26512864 2865+ // Regression: `claim_moq` compared the capability's host field against2866+ // itself, so the host-binding leg of `verify_binding` was a no-op. It only2867+ // matters once a capability can arrive from anywhere but the host's own2868+ // issuance path, which is exactly what the relay design contemplates.2869+ 2870+ fn v2_moq_attachment_rejects_a_capability_minted_by_another_host() {2871+ let host = iroh::SecretKey::from_bytes(&[9; 32]);2872+ let impostor = iroh::SecretKey::from_bytes(&[31; 32]);2873+ let remote = endpoint(1);2874+ let sessions = Arc::new(SessionRegistry::default());2875+ let lease = sessions2876+ .claim_v2(remote, [1; 16], InvitationGrants::VIEW)2877+ .unwrap();2878+ let forged = test_subscription(&impostor, remote, lease.session_id, 100, 200);2879+ let (_producer, consumer) = test_moq_broadcast();2880+ assert!(2881+ sessions2882+ .expect_moq_v2(2883+ remote,2884+ lease.session_id,2885+ media_moq_broadcast_name(lease.session_id).unwrap(),2886+ consumer,2887+ forged,2888+ *host.public().as_bytes(),2889+ )2890+ .is_err(),2891+ "a capability signed by another host must never be attached"2892+ );2893+ }2894+26522895 26532896 fn v2_moq_attachment_requires_a_fresh_endpoint_bound_subscription() {26542897 let host = iroh::SecretKey::from_bytes(&[9; 32]);@@ -2666,6 +2909,7 @@ mod tests {
26662909 media_moq_broadcast_name(lease.session_id).unwrap(),26672910 consumer,26682911 capability,2912+ *host.public().as_bytes(),26692913 )26702914 .unwrap();26712915 assert!(sessions.claim_moq_at(endpoint(2), 150).is_err());@@ -2685,6 +2929,7 @@ mod tests {
26852929 media_moq_broadcast_name(expired_lease.session_id).unwrap(),26862930 consumer,26872931 expired,2932+ *host.public().as_bytes(),26882933 )26892934 .unwrap();26902935 assert!(expired_sessions.claim_moq_at(remote, 201).is_err());