mumble_server_runtime_gateway/
peer.rs

1//! What the voice plane needs to reach one connection, and how it finds it.
2//!
3//! This is the guide's `Bindings` table (9.7). A datagram arrives with nothing
4//! but a source address, and everything needed to answer it has to be reachable
5//! from that address alone: the OCB2 domain, the shard's routing table, the
6//! cursor to gate on, and a queue for the TCP fallback.
7//!
8//! # Locks
9//!
10//! There are several, and none of them is on a hot path in the sense that
11//! matters. Each [`Peer`] owns its own `Mutex`, so two connections never contend
12//! with each other; the registry's `RwLock`s are read once per datagram and
13//! written once per connection lifetime. No guard here is ever held across an
14//! `.await`: every method returns before its caller can suspend.
15//!
16//! The one lock that would be a design error is a shared lock over *live state*
17//! on the packet path. There is none: the routing table is an `Arc` that the
18//! shard replaces whole, and reading it copies a pointer.
19
20use std::collections::HashMap;
21use std::net::{IpAddr, SocketAddr};
22use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
23use std::sync::{Arc, Mutex, MutexGuard, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard};
24use std::time::{Duration, Instant};
25
26use mumble_server_runtime_crypto::CryptState;
27use mumble_server_runtime_shard::{AudioRouting, ConnectionId, OutboundQueue, SessionId, ShardId};
28use tokio::sync::watch;
29
30use crate::limits::{TextBudget, VoiceBudget};
31
32/// Which shard a connection currently belongs to, and where its routes come
33/// from.
34///
35/// Replaced wholesale by a migration. The address, the key and the session are
36/// deliberately *not* in here: a migration must not disturb the UDP plane, and
37/// the surest way to guarantee that is for the fields it rewrites to be
38/// disjoint from the fields the UDP plane depends on.
39#[derive(Debug, Clone)]
40pub struct ShardPlane {
41    pub shard: ShardId,
42    pub routing: watch::Receiver<Arc<AudioRouting>>,
43}
44
45/// One connection, as the voice plane sees it.
46pub struct Peer {
47    connection: ConnectionId,
48    /// Stable for the life of the connection, across migrations included.
49    session: SessionId,
50    /// The TCP peer's host address. The cold path uses it to narrow the
51    /// candidates for an unknown datagram.
52    host: IpAddr,
53    crypt: Mutex<CryptState>,
54    /// The address this connection has proven it owns, if any.
55    address: Mutex<Option<SocketAddr>>,
56    /// Murmur's `aiUdpFlag`: whether this peer's own audio last arrived over
57    /// UDP. It decides how *it* is reached, never how anyone else is.
58    udp_mode: AtomicBool,
59    budget: Mutex<VoiceBudget>,
60    /// What this connection may still type. Its own bucket rather than a share
61    /// of the voice one: a talkative user is not a flooding one.
62    text: Mutex<TextBudget>,
63    /// How far through its shard's journal the connection has been advanced.
64    /// Written by the shard, read here (guide 9.5).
65    cursor: Arc<AtomicU64>,
66    queue: Arc<OutboundQueue>,
67    plane: RwLock<ShardPlane>,
68    /// When this connection was registered, for the one statistic it may be
69    /// told about itself.
70    online_since: Instant,
71    /// The last numbers the client reported about its own side of the link.
72    reported: Mutex<ClientReport>,
73}
74
75/// What a client last told us about the connection, in its own `Ping`.
76///
77/// Every field is the client's claim, not a measurement of ours: the reference
78/// server stores them verbatim and hands them back in `UserStats`, which is what
79/// fills the "To Client" column and the ping statistics of the information
80/// window. Client-supplied numbers are only ever shown back to the client that
81/// supplied them, so a client that lies here lies to itself alone.
82///
83/// REF: references/mumble/src/murmur/Messages.cpp : `msgPing` assigns
84///   `uiRemoteGood/Late/Lost/Resync`, `dUDPPingAvg/Var`, `uiUDPPackets`,
85///   `dTCPPingAvg/Var` and `uiTCPPackets` straight from the message.
86#[derive(Debug, Clone, Copy, Default, PartialEq)]
87pub struct ClientReport {
88    pub good: u32,
89    pub late: u32,
90    pub lost: u32,
91    pub resync: u32,
92    pub udp_packets: u32,
93    pub tcp_packets: u32,
94    pub udp_ping_avg: f32,
95    pub udp_ping_var: f32,
96    pub tcp_ping_avg: f32,
97    pub tcp_ping_var: f32,
98}
99
100/// Hand-written so key material never reaches a log. Everything printed here is
101/// an identifier or a state flag; the OCB2 domain is named, not shown.
102impl std::fmt::Debug for Peer {
103    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
104        formatter
105            .debug_struct("Peer")
106            .field("connection", &self.connection)
107            .field("session", &self.session)
108            .field("host", &self.host)
109            .field("address", &self.proven_address())
110            .field("udp_mode", &self.udp_mode.load(Ordering::Relaxed))
111            .field("shard", &self.shard())
112            .field("cursor", &self.cursor())
113            .finish_non_exhaustive()
114    }
115}
116
117impl Peer {
118    #[must_use]
119    pub fn new(
120        connection: ConnectionId,
121        session: SessionId,
122        host: IpAddr,
123        crypt: CryptState,
124        queue: Arc<OutboundQueue>,
125        plane: ShardPlane,
126        now: Instant,
127    ) -> Peer {
128        Peer {
129            connection,
130            session,
131            host,
132            crypt: Mutex::new(crypt),
133            address: Mutex::new(None),
134            udp_mode: AtomicBool::new(true),
135            budget: Mutex::new(VoiceBudget::new(now)),
136            text: Mutex::new(TextBudget::new(now)),
137            // Created here rather than handed in: the shard writes it and the
138            // voice plane reads it, and neither of them exists yet.
139            cursor: Arc::new(AtomicU64::new(0)),
140            queue,
141            plane: RwLock::new(plane),
142            online_since: now,
143            reported: Mutex::new(ClientReport::default()),
144        }
145    }
146
147    /// When this connection was registered.
148    #[must_use]
149    pub fn online_since(&self) -> Instant {
150        self.online_since
151    }
152
153    /// Store what the client reported about its own side of the link.
154    pub fn record_report(&self, report: ClientReport) {
155        *lock(&self.reported) = report;
156    }
157
158    #[must_use]
159    pub fn reported(&self) -> ClientReport {
160        *lock(&self.reported)
161    }
162
163    /// Voice throughput over the last second, in bytes per second, and how long
164    /// this connection has been idle.
165    #[must_use]
166    pub fn traffic(&self, now: Instant) -> (u32, Duration) {
167        let mut budget = lock(&self.budget);
168        (budget.bandwidth(now), budget.idle(now))
169    }
170
171    /// Note a control message that is not a keepalive.
172    pub fn record_activity(&self, now: Instant) {
173        lock(&self.budget).touch(now);
174    }
175
176    #[must_use]
177    pub fn connection(&self) -> ConnectionId {
178        self.connection
179    }
180
181    #[must_use]
182    pub fn session(&self) -> SessionId {
183        self.session
184    }
185
186    #[must_use]
187    pub fn host(&self) -> IpAddr {
188        self.host
189    }
190
191    #[must_use]
192    pub fn queue(&self) -> Arc<OutboundQueue> {
193        Arc::clone(&self.queue)
194    }
195
196    #[must_use]
197    pub fn cursor_cell(&self) -> Arc<AtomicU64> {
198        Arc::clone(&self.cursor)
199    }
200
201    #[must_use]
202    pub fn cursor(&self) -> u64 {
203        self.cursor.load(Ordering::Relaxed)
204    }
205
206    #[must_use]
207    pub fn shard(&self) -> ShardId {
208        read(&self.plane).shard
209    }
210
211    /// The routing table of the shard this connection belongs to.
212    #[must_use]
213    pub fn routing(&self) -> Arc<AudioRouting> {
214        let plane = read(&self.plane);
215        Arc::clone(&plane.routing.borrow())
216    }
217
218    /// Point this connection at another shard. Called by a migration, once.
219    pub fn move_to(&self, plane: ShardPlane) {
220        *write(&self.plane) = plane;
221    }
222
223    /// Try to decrypt a datagram in this peer's OCB2 domain.
224    ///
225    /// A failure is side-effect free - the IV is restored and nothing is written
226    /// to the replay history - which is what makes the cold path's "try every
227    /// candidate" safe.
228    ///
229    /// REF: references/mumble/src/murmur/Server.cpp : `Server::run` binds an
230    ///   unknown peer to the first `checkDecrypt` that succeeds.
231    #[must_use]
232    pub fn decrypt(&self, datagram: &[u8]) -> Option<Vec<u8>> {
233        lock(&self.crypt).decrypt(datagram)
234    }
235
236    #[must_use]
237    pub fn encrypt(&self, plaintext: &[u8]) -> Option<Vec<u8>> {
238        lock(&self.crypt).encrypt(plaintext)
239    }
240
241    /// The OCB2 counters the TCP `Ping` reply must report.
242    #[must_use]
243    pub fn crypt_counters(&self) -> (u32, u32, u32) {
244        let state = lock(&self.crypt);
245        (state.good, state.late, state.lost)
246    }
247
248    /// Whether this peer may send one more voice packet right now.
249    #[must_use]
250    pub fn allow_voice(&self, now: Instant, bytes: usize) -> bool {
251        lock(&self.budget).allow(now, bytes)
252    }
253
254    /// Whether this peer may send one more text message right now.
255    #[must_use]
256    pub fn allow_text(&self, now: Instant) -> bool {
257        lock(&self.text).allow(now)
258    }
259
260    /// Record that this peer's audio is arriving over UDP again, or that it has
261    /// fallen back to the tunnel.
262    ///
263    /// REF: references/mumble/src/murmur/Server.cpp : `aiUdpFlag` goes to 0 on a
264    ///   `UDPTunnel` message and back to 1 when a datagram arrives.
265    pub fn set_udp_mode(&self, on: bool) {
266        self.udp_mode.store(on, Ordering::Relaxed);
267    }
268
269    /// Where audio for this peer goes: its proven address, or `None` meaning the
270    /// TCP tunnel.
271    ///
272    /// Both conditions are required, as in the real server: a proven address
273    /// *and* a peer whose own audio last came over UDP. A client whose UDP dies
274    /// mid-call keeps hearing everyone, through the tunnel.
275    #[must_use]
276    pub fn destination(&self) -> Option<SocketAddr> {
277        if !self.udp_mode.load(Ordering::Relaxed) {
278            return None;
279        }
280        *lock(&self.address)
281    }
282
283    /// The address this peer has proven, whatever transport it last used.
284    #[must_use]
285    pub fn proven_address(&self) -> Option<SocketAddr> {
286        *lock(&self.address)
287    }
288
289    fn bind(&self, address: SocketAddr) {
290        *lock(&self.address) = Some(address);
291    }
292
293    /// Mark the connection for teardown by its own task.
294    pub fn close(&self) {
295        self.queue.mark_fatal();
296    }
297}
298
299/// Every live connection, indexed the four ways the runtime needs.
300///
301/// Registration and removal are explicit rather than RAII on [`Peer`], because
302/// the voice plane holds `Arc<Peer>` clones across `.await` points: a peer must
303/// stop being *findable* the moment its connection ends, while the last
304/// in-flight packet is allowed to finish with the copy it already has.
305#[derive(Debug, Default)]
306pub struct Peers {
307    by_connection: RwLock<HashMap<ConnectionId, Arc<Peer>>>,
308    by_session: RwLock<HashMap<SessionId, Arc<Peer>>>,
309    by_address: RwLock<HashMap<SocketAddr, Arc<Peer>>>,
310    /// The guide's runtime-global "host IP to connections" index. It cannot be
311    /// per shard: an unknown datagram has not told us which shard it came from
312    /// yet, which is the whole reason the cold path exists.
313    by_host: RwLock<HashMap<IpAddr, Vec<Arc<Peer>>>>,
314}
315
316impl Peers {
317    #[must_use]
318    pub fn new() -> Peers {
319        Peers::default()
320    }
321
322    pub fn insert(&self, peer: Arc<Peer>) {
323        write(&self.by_connection).insert(peer.connection(), Arc::clone(&peer));
324        write(&self.by_session).insert(peer.session(), Arc::clone(&peer));
325        write(&self.by_host)
326            .entry(peer.host())
327            .or_default()
328            .push(peer);
329    }
330
331    /// Forget a connection. Any address it had proven is released with it, so a
332    /// later datagram from that address is re-proven rather than delivered to
333    /// whoever inherits the socket.
334    pub fn remove(&self, connection: ConnectionId) -> Option<Arc<Peer>> {
335        let peer = write(&self.by_connection).remove(&connection)?;
336        write(&self.by_session).remove(&peer.session());
337        if let Some(address) = peer.proven_address() {
338            write(&self.by_address).remove(&address);
339        }
340        let mut hosts = write(&self.by_host);
341        if let Some(list) = hosts.get_mut(&peer.host()) {
342            list.retain(|other| other.connection() != connection);
343            if list.is_empty() {
344                hosts.remove(&peer.host());
345            }
346        }
347        Some(peer)
348    }
349
350    #[must_use]
351    pub fn by_connection(&self, connection: ConnectionId) -> Option<Arc<Peer>> {
352        read(&self.by_connection).get(&connection).cloned()
353    }
354
355    #[must_use]
356    pub fn by_session(&self, session: SessionId) -> Option<Arc<Peer>> {
357        read(&self.by_session).get(&session).cloned()
358    }
359
360    #[must_use]
361    pub fn by_address(&self, address: SocketAddr) -> Option<Arc<Peer>> {
362        read(&self.by_address).get(&address).cloned()
363    }
364
365    /// The connections whose TCP peer shares this host address.
366    #[must_use]
367    pub fn candidates(&self, host: IpAddr) -> Vec<Arc<Peer>> {
368        read(&self.by_host).get(&host).cloned().unwrap_or_default()
369    }
370
371    /// Bind a proven address to a peer.
372    ///
373    /// The address is rebound, not merged: a client behind a NAT that reassigns
374    /// its port keeps working, and the old entry is dropped so it cannot deliver
375    /// to a peer that has moved.
376    pub fn bind(&self, address: SocketAddr, peer: &Arc<Peer>) {
377        if let Some(previous) = peer.proven_address()
378            && previous != address
379        {
380            write(&self.by_address).remove(&previous);
381        }
382        peer.bind(address);
383        write(&self.by_address).insert(address, Arc::clone(peer));
384    }
385
386    /// Every live connection attached to one shard.
387    #[must_use]
388    pub fn on_shard(&self, shard: ShardId) -> Vec<Arc<Peer>> {
389        read(&self.by_connection)
390            .values()
391            .filter(|peer| peer.shard() == shard)
392            .cloned()
393            .collect()
394    }
395
396    #[must_use]
397    pub fn len(&self) -> usize {
398        read(&self.by_connection).len()
399    }
400
401    #[must_use]
402    pub fn is_empty(&self) -> bool {
403        self.len() == 0
404    }
405}
406
407/// A poisoned lock means a panic unwound while a field was being read or
408/// written. Nothing guarded here can be left half-updated - each guard covers a
409/// single assignment or a single crypto call - so recovering is correct, whereas
410/// propagating would take the whole voice plane down over one connection.
411fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
412    mutex.lock().unwrap_or_else(PoisonError::into_inner)
413}
414
415fn read<T>(lock: &RwLock<T>) -> RwLockReadGuard<'_, T> {
416    lock.read().unwrap_or_else(PoisonError::into_inner)
417}
418
419fn write<T>(lock: &RwLock<T>) -> RwLockWriteGuard<'_, T> {
420    lock.write().unwrap_or_else(PoisonError::into_inner)
421}
422
423#[cfg(test)]
424mod tests {
425    #![allow(clippy::expect_used)]
426
427    use super::*;
428    use mumble_server_runtime_shard::AudioRouting;
429
430    fn plane() -> ShardPlane {
431        let (_sender, routing) = watch::channel(Arc::new(AudioRouting::default()));
432        ShardPlane {
433            shard: ShardId(1),
434            routing,
435        }
436    }
437
438    fn peer(connection: u64, session: u32, host: [u8; 4]) -> Arc<Peer> {
439        // Dropping the writer half closes the queue, which is harmless here:
440        // these tests exercise the registry and never push a message.
441        let (queue, _writer) = OutboundQueue::new();
442        Arc::new(Peer::new(
443            ConnectionId(connection),
444            SessionId(session),
445            IpAddr::from(host),
446            CryptState::new(&[0u8; 16], &[0u8; 16], &[1u8; 16]),
447            Arc::new(queue),
448            plane(),
449            Instant::now(),
450        ))
451    }
452
453    fn address(port: u16) -> SocketAddr {
454        SocketAddr::from(([127, 0, 0, 1], port))
455    }
456
457    #[test]
458    fn a_removed_connection_releases_its_address() {
459        let peers = Peers::new();
460        let alice = peer(1, 10, [127, 0, 0, 1]);
461        peers.insert(Arc::clone(&alice));
462        peers.bind(address(5000), &alice);
463
464        assert!(peers.by_address(address(5000)).is_some());
465        peers.remove(ConnectionId(1));
466
467        assert!(
468            peers.by_address(address(5000)).is_none(),
469            "an address left bound would deliver to whoever inherits the socket"
470        );
471        assert!(peers.by_session(SessionId(10)).is_none());
472        assert!(peers.candidates(IpAddr::from([127, 0, 0, 1])).is_empty());
473    }
474
475    #[test]
476    fn rebinding_an_address_drops_the_previous_one() {
477        let peers = Peers::new();
478        let alice = peer(1, 10, [127, 0, 0, 1]);
479        peers.insert(Arc::clone(&alice));
480
481        peers.bind(address(5000), &alice);
482        peers.bind(address(5001), &alice);
483
484        assert!(peers.by_address(address(5000)).is_none());
485        assert_eq!(
486            peers
487                .by_address(address(5001))
488                .map(|found| found.connection()),
489            Some(ConnectionId(1))
490        );
491    }
492
493    #[test]
494    fn the_host_index_narrows_the_cold_path() {
495        let peers = Peers::new();
496        peers.insert(peer(1, 10, [10, 0, 0, 1]));
497        peers.insert(peer(2, 20, [10, 0, 0, 1]));
498        peers.insert(peer(3, 30, [10, 0, 0, 2]));
499
500        assert_eq!(peers.candidates(IpAddr::from([10, 0, 0, 1])).len(), 2);
501        assert_eq!(peers.candidates(IpAddr::from([10, 0, 0, 2])).len(), 1);
502        assert!(peers.candidates(IpAddr::from([10, 0, 0, 3])).is_empty());
503    }
504
505    #[test]
506    fn a_peer_on_the_tunnel_has_no_udp_destination() {
507        let peers = Peers::new();
508        let alice = peer(1, 10, [127, 0, 0, 1]);
509        peers.insert(Arc::clone(&alice));
510        peers.bind(address(5000), &alice);
511        assert_eq!(alice.destination(), Some(address(5000)));
512
513        alice.set_udp_mode(false);
514        assert_eq!(
515            alice.destination(),
516            None,
517            "a client that tunnelled its own audio must be answered on the tunnel"
518        );
519        assert_eq!(
520            alice.proven_address(),
521            Some(address(5000)),
522            "falling back must not forget the proof, only stop using it"
523        );
524    }
525}
mumble_server_runtime_gateway/peer.rs

mumble_server_runtime_gateway/
peer.rs
