tor_proto/channel/
circmap.rs

1//! Types and code to map circuit IDs to circuits.
2
3// NOTE: This is a work in progress and I bet I'll refactor it a lot;
4// it needs to stay opaque!
5
6use crate::circuit::CircuitRxSender;
7use crate::client::circuit::padding::{PaddingController, QueuedCellPaddingInfo};
8use crate::{Error, Result};
9use tor_basic_utils::RngExt;
10use tor_cell::chancell::CircId;
11use tor_cell::chancell::msg::DestroyReason;
12
13use crate::circuit::celltypes::CreateResponse;
14use crate::client::circuit::halfcirc::HalfCirc;
15
16use oneshot_fused_workaround as oneshot;
17
18use rand::Rng;
19use rand::distr::Distribution;
20use std::collections::{HashMap, hash_map::Entry};
21use std::ops::{Deref, DerefMut};
22use std::result::Result as StdResult;
23use std::sync::Arc;
24
25#[cfg(feature = "relay")]
26use crate::relay::RelayCirc;
27
28/// Which group of circuit IDs are we allowed to allocate in this map?
29///
30/// If we initiated the channel, we use High circuit ids.  If we're the
31/// responder, we use low circuit ids.
32#[derive(Copy, Clone)]
33pub(crate) enum CircIdRange {
34    /// Only use circuit IDs with the MSB cleared.
35    #[allow(dead_code)] // Relays will need this.
36    Low,
37    /// Only use circuit IDs with the MSB set.
38    High,
39    // Historical note: There used to be an "All" range of circuit IDs
40    // available to clients only.  We stopped using "All" when we moved to link
41    // protocol version 4.
42}
43
44impl CircIdRange {
45    /// The range of integer circuit IDs that we are allowed to allocate.
46    /// Prefer using other more specific methods over this one.
47    const fn integer_range(&self) -> std::ops::RangeInclusive<u32> {
48        const MIDPOINT: u32 = 0x8000_0000;
49
50        match self {
51            // 0 is an invalid value
52            Self::Low => 1..=(MIDPOINT - 1),
53            Self::High => MIDPOINT..=u32::MAX,
54        }
55    }
56
57    /// Is this circuit ID allowed to be allocated by the channel's peer?
58    pub(crate) fn is_allowed_for_peer(&self, id: CircId) -> bool {
59        // If our range does not contain it, then it is allowed.
60        // Note that a `CircId` never contains a value of zero,
61        // so no need to consider it here.
62        !self.integer_range().contains(&id.into())
63    }
64}
65
66impl rand::distr::Distribution<CircId> for CircIdRange {
67    /// Return a random circuit ID in the appropriate range.
68    fn sample<R: Rng + ?Sized>(&self, mut rng: &mut R) -> CircId {
69        let v = rng.gen_range_checked(self.integer_range());
70        let v = v.expect("Unexpected empty range passed to gen_range_checked");
71        CircId::new(v).expect("Unexpected zero value")
72    }
73}
74
75/// An entry in the circuit map.  Right now, we only have "here's the
76/// way to send cells to a given circuit", but that's likely to
77/// change.
78#[derive(Debug)]
79pub(super) enum CircEnt {
80    /// An origin circuit that has not yet received a CREATED cell.
81    ///
82    /// For this circuit, the CREATED* cell or DESTROY cell gets sent
83    /// to the oneshot sender to tell the corresponding
84    /// PendingClientCirc that the handshake is done.
85    ///
86    /// Once that's done, the `CircuitRxSender` mpsc sender will be used to send subsequent
87    /// cells to the circuit.
88    Opening {
89        /// The oneshot sender on which to report a create response
90        create_response_sender: oneshot::Sender<CreateResponse>,
91        /// A sink which should receive all the relay cells for this circuit
92        /// from this channel
93        cell_sender: CircuitRxSender,
94        /// A padding controller we should use when reporting flushed cells.
95        padding_ctrl: PaddingController,
96    },
97
98    /// An origin circuit (a circuit which originated here)
99    /// that is open and can be given relay cells.
100    OpenOrigin {
101        /// A sink which should receive all the relay cells for this circuit
102        /// from this channel
103        cell_sender: CircuitRxSender,
104        /// A padding controller we should use when reporting flushed cells.
105        padding_ctrl: PaddingController,
106    },
107
108    /// A relay circuit (a circuit in which we are a hop on the path)
109    /// that is open and can be given relay cells.
110    #[cfg(feature = "relay")]
111    OpenRelay {
112        /// A handle to the circuit.
113        /// TODO(relay): We need to store the `Arc<RelayCirc>` somewhere
114        /// and currently this seems like the best place to store it.
115        /// As we implement more functionality maybe we'll find a better place to store it,
116        /// in which case we should consider combining the `OpenOrigin` and `OpenRelay` variants.
117        _circ: Arc<RelayCirc>,
118        /// A sink which should receive all the relay cells for this circuit
119        /// from this channel
120        cell_sender: CircuitRxSender,
121        /// A padding controller we should use when reporting flushed cells.
122        padding_ctrl: PaddingController,
123    },
124
125    /// A circuit where we have sent a DESTROY, but the other end might
126    /// not have gotten a DESTROY yet.
127    DestroySent(HalfCirc),
128}
129
130/// An "smart pointer" that wraps an exclusive reference
131/// of a `CircEnt`.
132///
133/// When being dropped, this object updates the open or opening entries
134/// counter of the `CircMap`.
135pub(super) struct MutCircEnt<'a> {
136    /// An exclusive reference to the `CircEnt`.
137    value: &'a mut CircEnt,
138    /// An exclusive reference to the open or opening
139    ///  entries counter.
140    open_count: &'a mut usize,
141    /// True if the entry was open or opening when borrowed.
142    was_open: bool,
143}
144
145impl<'a> Drop for MutCircEnt<'a> {
146    fn drop(&mut self) {
147        let is_open = !matches!(self.value, CircEnt::DestroySent(_));
148        match (self.was_open, is_open) {
149            (false, true) => *self.open_count = self.open_count.saturating_add(1),
150            (true, false) => *self.open_count = self.open_count.saturating_sub(1),
151            (_, _) => (),
152        };
153    }
154}
155
156impl<'a> Deref for MutCircEnt<'a> {
157    type Target = CircEnt;
158    fn deref(&self) -> &Self::Target {
159        self.value
160    }
161}
162
163impl<'a> DerefMut for MutCircEnt<'a> {
164    fn deref_mut(&mut self) -> &mut Self::Target {
165        self.value
166    }
167}
168
169/// A map from circuit IDs to circuit entries. Each channel has one.
170pub(super) struct CircMap {
171    /// Map from circuit IDs to entries
172    m: HashMap<CircId, CircEnt>,
173    /// Rule for allocating new circuit IDs.
174    range: CircIdRange,
175    /// Number of open or opening entry in this map.
176    open_count: usize,
177}
178
179impl CircMap {
180    /// Make a new empty CircMap
181    pub(super) fn new(idrange: CircIdRange) -> Self {
182        CircMap {
183            m: HashMap::new(),
184            range: idrange,
185            open_count: 0,
186        }
187    }
188
189    /// Add a new set of elements (corresponding to a
190    /// [`PendingClientTunnel`](crate::client::circuit::PendingClientTunnel))
191    /// as an entry to this map.
192    ///
193    /// On success return the allocated circuit ID.
194    pub(super) fn add_origin_ent<R: Rng>(
195        &mut self,
196        rng: &mut R,
197        createdsink: oneshot::Sender<CreateResponse>,
198        sink: CircuitRxSender,
199        padding_ctrl: PaddingController,
200    ) -> Result<CircId> {
201        /// How many times do we probe for a random circuit ID before
202        /// we assume that the range is fully populated?
203        ///
204        /// TODO: C tor does 64, but that is probably overkill with 4-byte circuit IDs.
205        const N_ATTEMPTS: usize = 16;
206        let iter = self.range.sample_iter(rng).take(N_ATTEMPTS);
207        let circ_ent = CircEnt::Opening {
208            create_response_sender: createdsink,
209            cell_sender: sink,
210            padding_ctrl,
211        };
212        for id in iter {
213            let ent = self.m.entry(id);
214            if let Entry::Vacant(_) = &ent {
215                ent.or_insert(circ_ent);
216                self.open_count += 1;
217                return Ok(id);
218            }
219        }
220        Err(Error::IdRangeFull)
221    }
222
223    /// Add a new set of elements (corresponding to a [`RelayCirc`]) as an entry to this map.
224    ///
225    /// We use [`DestroyReason`] as the return type since we very likely want to destroy the circuit
226    /// if this fails, and not return an error and destroy the entire channel.
227    #[cfg(feature = "relay")]
228    pub(super) fn add_relay_ent(
229        &mut self,
230        circ_id: CircId,
231        circ: Arc<RelayCirc>,
232        sink: CircuitRxSender,
233        padding_ctrl: PaddingController,
234    ) -> StdResult<(), DestroyReason> {
235        // The peer is only allowed to use a subset of the ID range.
236        if !self.range.is_allowed_for_peer(circ_id) {
237            return Err(DestroyReason::NONE);
238        }
239
240        let circ_ent = CircEnt::OpenRelay {
241            _circ: circ,
242            cell_sender: sink,
243            padding_ctrl,
244        };
245
246        if let Entry::Vacant(ent) = self.m.entry(circ_id) {
247            ent.insert(circ_ent);
248            self.open_count += 1;
249            Ok(())
250        } else {
251            Err(DestroyReason::NONE)
252        }
253    }
254
255    /// Testing only: install an entry in this circuit map without regard
256    /// for consistency.
257    #[cfg(test)]
258    pub(super) fn put_unchecked(&mut self, id: CircId, ent: CircEnt) {
259        self.m.insert(id, ent);
260    }
261
262    /// Return the entry for `id` in this map, if any.
263    pub(super) fn get_mut(&mut self, id: CircId) -> Option<MutCircEnt> {
264        let open_count = &mut self.open_count;
265        self.m.get_mut(&id).map(move |ent| MutCircEnt {
266            open_count,
267            was_open: !matches!(ent, CircEnt::DestroySent(_)),
268            value: ent,
269        })
270    }
271
272    /// Returns `true` if the circuit with the specified `id`
273    /// is open or opening.
274    ///
275    /// Returns `false` if the circuit is not in the circuit map,
276    /// or if we have already sent a DESTROY on it.
277    pub(super) fn is_open(&self, id: CircId) -> bool {
278        let Some(entry) = self.m.get(&id) else {
279            return false;
280        };
281
282        match entry {
283            CircEnt::Opening { .. } | CircEnt::OpenOrigin { .. } => true,
284            #[cfg(feature = "relay")]
285            CircEnt::OpenRelay { .. } => true,
286            CircEnt::DestroySent(..) => false,
287        }
288    }
289
290    /// Inform the relevant circuit's padding subsystem that a given cell has been flushed.
291    pub(super) fn note_cell_flushed(&mut self, id: CircId, info: QueuedCellPaddingInfo) {
292        let padding_ctrl = match self.m.get(&id) {
293            Some(CircEnt::Opening { padding_ctrl, .. }) => padding_ctrl,
294            Some(CircEnt::OpenOrigin { padding_ctrl, .. }) => padding_ctrl,
295            #[cfg(feature = "relay")]
296            Some(CircEnt::OpenRelay { padding_ctrl, .. }) => padding_ctrl,
297            Some(CircEnt::DestroySent(..)) | None => return,
298        };
299        padding_ctrl.flushed_relay_cell(info);
300    }
301
302    /// See whether 'id' is an opening circuit.  If so, mark it "open" and
303    /// return a oneshot::Sender that is waiting for its create cell.
304    ///
305    /// Returns `None` if `id` is not in our circuit map,
306    /// or if it is not an opening circuit.
307    pub(super) fn advance_from_opening(
308        &mut self,
309        id: CircId,
310    ) -> Option<oneshot::Sender<CreateResponse>> {
311        // TODO: there should be a better way to do
312        // this. hash_map::Entry seems like it could be better, but
313        // there seems to be no way to replace the object in-place as
314        // a consuming function of itself.
315        let ok = matches!(self.m.get(&id), Some(CircEnt::Opening { .. }));
316        if ok {
317            if let Some(CircEnt::Opening {
318                create_response_sender: oneshot,
319                cell_sender: sink,
320                padding_ctrl,
321            }) = self.m.remove(&id)
322            {
323                self.m.insert(
324                    id,
325                    CircEnt::OpenOrigin {
326                        cell_sender: sink,
327                        padding_ctrl,
328                    },
329                );
330                Some(oneshot)
331            } else {
332                panic!("internal error: inconsistent circuit state");
333            }
334        } else {
335            None
336        }
337    }
338
339    /// Called when we have sent a DESTROY on a circuit.  Configures
340    /// a "HalfCirc" object to track how many cells we get on this
341    /// circuit, and to prevent us from reusing it immediately.
342    pub(super) fn destroy_sent(&mut self, id: CircId, hc: HalfCirc) {
343        if let Some(replaced) = self.m.insert(id, CircEnt::DestroySent(hc)) {
344            if !matches!(replaced, CircEnt::DestroySent(_)) {
345                // replaced an Open/Opening entry with DestroySent
346                self.open_count = self.open_count.saturating_sub(1);
347            }
348        }
349    }
350
351    /// Extract the value from this map with 'id' if any
352    pub(super) fn remove(&mut self, id: CircId) -> Option<CircEnt> {
353        self.m.remove(&id).map(|removed| {
354            if !matches!(removed, CircEnt::DestroySent(_)) {
355                self.open_count = self.open_count.saturating_sub(1);
356            }
357            removed
358        })
359    }
360
361    /// Return the total number of open and opening entries in the map
362    pub(super) fn open_ent_count(&self) -> usize {
363        self.open_count
364    }
365}
366
367#[cfg(test)]
368mod test {
369    // @@ begin test lint list maintained by maint/add_warning @@
370    #![allow(clippy::bool_assert_comparison)]
371    #![allow(clippy::clone_on_copy)]
372    #![allow(clippy::dbg_macro)]
373    #![allow(clippy::mixed_attributes_style)]
374    #![allow(clippy::print_stderr)]
375    #![allow(clippy::print_stdout)]
376    #![allow(clippy::single_char_pattern)]
377    #![allow(clippy::unwrap_used)]
378    #![allow(clippy::unchecked_time_subtraction)]
379    #![allow(clippy::useless_vec)]
380    #![allow(clippy::needless_pass_by_value)]
381    #![allow(clippy::string_slice)] // See arti#2571
382    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
383    use super::*;
384    use crate::circuit::test::fake_mpsc;
385    use crate::client::circuit::padding::new_padding;
386    use tor_basic_utils::test_rng::testing_rng;
387    use tor_rtcompat::DynTimeProvider;
388
389    #[test]
390    fn circmap_basics() {
391        let mut map_low = CircMap::new(CircIdRange::Low);
392        let mut map_high = CircMap::new(CircIdRange::High);
393        let mut ids_low: Vec<CircId> = Vec::new();
394        let mut ids_high: Vec<CircId> = Vec::new();
395        let mut rng = testing_rng();
396        tor_rtcompat::test_with_one_runtime!(|runtime| async {
397            let (padding_ctrl, _padding_stream) = new_padding(DynTimeProvider::new(runtime));
398
399            assert!(map_low.get_mut(CircId::new(77).unwrap()).is_none());
400
401            for _ in 0..128 {
402                let (csnd, _) = oneshot::channel();
403                let (snd, _) = fake_mpsc(8);
404                let id_low = map_low
405                    .add_origin_ent(&mut rng, csnd, snd, padding_ctrl.clone())
406                    .unwrap();
407                assert!(u32::from(id_low) > 0);
408                assert!(u32::from(id_low) < 0x80000000);
409                assert!(!ids_low.contains(&id_low));
410                ids_low.push(id_low);
411
412                assert!(matches!(
413                    *map_low.get_mut(id_low).unwrap(),
414                    CircEnt::Opening { .. }
415                ));
416
417                let (csnd, _) = oneshot::channel();
418                let (snd, _) = fake_mpsc(8);
419                let id_high = map_high
420                    .add_origin_ent(&mut rng, csnd, snd, padding_ctrl.clone())
421                    .unwrap();
422                assert!(u32::from(id_high) >= 0x80000000);
423                assert!(!ids_high.contains(&id_high));
424                ids_high.push(id_high);
425            }
426
427            // Test open / opening entry counting
428            assert_eq!(128, map_low.open_ent_count());
429            assert_eq!(128, map_high.open_ent_count());
430
431            // Test remove
432            assert!(map_low.get_mut(ids_low[0]).is_some());
433            map_low.remove(ids_low[0]);
434            assert!(map_low.get_mut(ids_low[0]).is_none());
435            assert_eq!(127, map_low.open_ent_count());
436
437            // Test DestroySent doesn't count
438            map_low.destroy_sent(CircId::new(256).unwrap(), HalfCirc::new(1));
439            assert_eq!(127, map_low.open_ent_count());
440
441            // Test advance_from_opening.
442
443            // Good case.
444            assert!(map_high.get_mut(ids_high[0]).is_some());
445            assert!(matches!(
446                *map_high.get_mut(ids_high[0]).unwrap(),
447                CircEnt::Opening { .. }
448            ));
449            let adv = map_high.advance_from_opening(ids_high[0]);
450            assert!(adv.is_some());
451            assert!(matches!(
452                *map_high.get_mut(ids_high[0]).unwrap(),
453                CircEnt::OpenOrigin { .. }
454            ));
455
456            // Can't double-advance.
457            let adv = map_high.advance_from_opening(ids_high[0]);
458            assert!(adv.is_none());
459
460            // Can't advance an entry that is not there.  We know "77"
461            // can't be in map_high, since we only added high circids to
462            // it.
463            let adv = map_high.advance_from_opening(CircId::new(77).unwrap());
464            assert!(adv.is_none());
465        });
466    }
467}
tor_proto/channel/circmap.rs

tor_proto/channel/
circmap.rs
