1
//! Module exposing types for representing circuits in the tunnel reactor.
2

            
3
pub(crate) mod circhop;
4
pub(super) mod extender;
5

            
6
use crate::channel::Channel;
7
use crate::circuit::cell_sender::CircuitCellSender;
8
use crate::circuit::celltypes::CreateResponse;
9
use crate::circuit::circhop::{HopSettings, ReactorStreamComponents};
10
use crate::circuit::create::{Create2Wrap, CreateFastWrap, CreateHandshakeWrap};
11
use crate::circuit::padding::CircPaddingDisposition;
12
use crate::circuit::{CircuitRxReceiver, UniqId};
13
use crate::client::circuit::handshake::{BoxedClientLayer, HandshakeRole};
14
use crate::client::circuit::padding::{
15
    self, PaddingController, PaddingEventStream, QueuedCellPaddingInfo,
16
};
17
use crate::client::circuit::{ClientCircChanMsg, MutableState, path};
18
use crate::client::reactor::MetaCellDisposition;
19
use crate::congestion::CongestionSignals;
20
use crate::congestion::sendme;
21
use crate::crypto::binding::CircuitBinding;
22
use crate::crypto::cell::{
23
    HopNum, InboundClientCrypt, InboundClientLayer, OutboundClientCrypt, OutboundClientLayer,
24
    RelayCellBody,
25
};
26
use crate::crypto::handshake::fast::CreateFastClient;
27
use crate::crypto::handshake::ntor::{NtorClient, NtorPublicKey};
28
use crate::crypto::handshake::ntor_v3::{NtorV3Client, NtorV3PublicKey};
29
use crate::crypto::handshake::{ClientHandshake, KeyGenerator};
30
use crate::memquota::{CircuitAccount, SpecificAccount as _, StreamAccount};
31
use crate::stream::cmdcheck::{AnyCmdChecker, StreamStatus};
32
use crate::stream::flow_ctrl::state::WithSidechannelMitigations;
33
use crate::stream::msg_streamid;
34
use crate::streammap;
35
use crate::tunnel::TunnelScopedCircId;
36
use crate::util::err::ReactorError;
37
use crate::util::timeout::TimeoutEstimator;
38
use crate::{ClockSkew, Error, Result};
39

            
40
use tor_async_utils::{SinkTrySend as _, SinkTrySendError as _};
41
use tor_cell::chancell::msg::{AnyChanMsg, HandshakeType, Relay};
42
use tor_cell::chancell::{AnyChanCell, ChanCmd, CircId};
43
use tor_cell::chancell::{BoxedCellBody, ChanMsg};
44
use tor_cell::relaycell::msg::{AnyRelayMsg, End, Sendme, SendmeTag, Truncated};
45
use tor_cell::relaycell::{
46
    AnyRelayMsgOuter, RelayCellDecoderResult, RelayCellFormat, RelayCmd, StreamId, UnparsedRelayMsg,
47
};
48
use tor_error::{Bug, internal};
49
use tor_linkspec::RelayIds;
50
use tor_llcrypto::pk;
51
use web_time_compat::{Duration, Instant, SystemTime};
52

            
53
use futures::SinkExt as _;
54
use oneshot_fused_workaround as oneshot;
55
use tor_rtcompat::{DynTimeProvider, SleepProvider as _};
56
use tracing::{debug, instrument, trace, warn};
57

            
58
use super::{
59
    CellHandlers, CircuitHandshake, CloseStreamBehavior, ReactorResultChannel, SendRelayCell,
60
};
61

            
62
use crate::conflux::msghandler::ConfluxStatus;
63

            
64
use std::borrow::Borrow;
65
use std::pin::Pin;
66
use std::result::Result as StdResult;
67
use std::sync::Arc;
68

            
69
use extender::HandshakeAuxDataHandler;
70

            
71
#[cfg(feature = "hs-service")]
72
use {
73
    crate::circuit::CircHopSyncView,
74
    crate::stream::{InboundDataCmdChecker, IncomingStreamRequest},
75
    tor_cell::relaycell::msg::Begin,
76
};
77

            
78
#[cfg(feature = "conflux")]
79
use {
80
    crate::conflux::msghandler::{ConfluxAction, ConfluxCmd, ConfluxMsgHandler, OooRelayMsg},
81
    crate::tunnel::TunnelId,
82
};
83

            
84
#[cfg(not(feature = "flowctl-cc"))]
85
use crate::stream::STREAM_READER_BUFFER;
86

            
87
pub(super) use circhop::{CircHop, CircHopList};
88

            
89
/// A circuit "leg" from a tunnel.
90
///
91
/// Regular (non-multipath) circuits have a single leg.
92
/// Conflux (multipath) circuits have `N` (usually, `N = 2`).
93
pub(crate) struct Circuit {
94
    /// The time provider.
95
    runtime: DynTimeProvider,
96
    /// The channel this circuit is attached to.
97
    channel: Arc<Channel>,
98
    /// Sender object used to actually send cells.
99
    ///
100
    /// NOTE: Control messages could potentially add unboundedly to this, although that's
101
    ///       not likely to happen (and isn't triggereable from the network, either).
102
    pub(super) chan_sender: CircuitCellSender,
103
    /// Input stream, on which we receive ChanMsg objects from this circuit's
104
    /// channel.
105
    ///
106
    // TODO: could use a SPSC channel here instead.
107
    pub(super) input: CircuitRxReceiver,
108
    /// The cryptographic state for this circuit for inbound cells.
109
    /// This object is divided into multiple layers, each of which is
110
    /// shared with one hop of the circuit.
111
    crypto_in: InboundClientCrypt,
112
    /// The cryptographic state for this circuit for outbound cells.
113
    crypto_out: OutboundClientCrypt,
114
    /// List of hops state objects used by the reactor
115
    pub(super) hops: CircHopList,
116
    /// Mutable information about this circuit,
117
    /// shared with the reactor's `ConfluxSet`.
118
    mutable: Arc<MutableState>,
119
    /// This circuit's identifier.
120
    circ_id: CircId,
121
    /// An identifier for logging about this reactor's circuit.
122
    unique_id: TunnelScopedCircId,
123
    /// A handler for conflux cells.
124
    ///
125
    /// Set once the conflux handshake is initiated by the reactor
126
    /// using [`Reactor::handle_link_circuits`](super::Reactor::handle_link_circuits).
127
    #[cfg(feature = "conflux")]
128
    conflux_handler: Option<ConfluxMsgHandler>,
129
    /// A padding controller to which padding-related events should be reported.
130
    padding_ctrl: PaddingController,
131
    /// An event stream telling us about padding-related events.
132
    //
133
    // TODO: it would be nice to have all of these streams wrapped in a single
134
    // SelectAll, but we can't really do that, since we need the ability to move them
135
    // from one conflux set to another, and a SelectAll doesn't let you actually
136
    // remove one of its constituent streams.  This issue might get solved along
137
    // with the rest of the next reactor refactoring.
138
    pub(super) padding_event_stream: PaddingEventStream,
139
    /// Current rules for blocking traffic, according to the padding controller.
140
    #[cfg(feature = "circ-padding")]
141
    padding_block: Option<padding::StartBlocking>,
142
    /// The circuit timeout estimator.
143
    ///
144
    /// Used for computing half-stream expiration.
145
    timeouts: Arc<dyn TimeoutEstimator>,
146
    /// Memory quota account
147
    #[allow(dead_code)] // Partly here to keep it alive as long as the circuit
148
    memquota: CircuitAccount,
149
}
150

            
151
/// A command to run in response to a circuit event.
152
///
153
/// Unlike `RunOnceCmdInner`, doesn't know anything about `UniqId`s.
154
/// The user of the `CircuitCmd`s is supposed to know the `UniqId`
155
/// of the circuit the `CircuitCmd` came from.
156
///
157
/// This type gets mapped to a `RunOnceCmdInner` in the circuit reactor.
158
#[derive(Debug, derive_more::From)]
159
pub(super) enum CircuitCmd {
160
    /// Send a RELAY cell on the circuit leg this command originates from.
161
    Send(SendRelayCell),
162
    /// Handle a SENDME message received on the circuit leg this command originates from.
163
    HandleSendMe {
164
        /// The hop number.
165
        hop: HopNum,
166
        /// The SENDME message to handle.
167
        sendme: Sendme,
168
    },
169
    /// Close the specified stream on the circuit leg this command originates from.
170
    CloseStream {
171
        /// The hop number.
172
        hop: HopNum,
173
        /// The ID of the stream to close.
174
        sid: StreamId,
175
        /// The stream-closing behavior.
176
        behav: CloseStreamBehavior,
177
        /// The reason for closing the stream.
178
        reason: streammap::TerminateReason,
179
    },
180
    /// Perform an action resulting from handling a conflux cell.
181
    #[cfg(feature = "conflux")]
182
    Conflux(ConfluxCmd),
183
    /// Perform a clean shutdown on this circuit.
184
    CleanShutdown,
185
    /// Enqueue an out-of-order cell in the reactor.
186
    #[cfg(feature = "conflux")]
187
    Enqueue(OooRelayMsg),
188
}
189

            
190
/// Return a `CircProto` error for the specified unsupported cell.
191
///
192
/// This error will shut down the reactor.
193
///
194
/// Note: this is a macro to simplify usage (this way the caller doesn't
195
/// need to .map() the result to the appropriate type)
196
macro_rules! unsupported_client_cell {
197
    ($msg:expr) => {{
198
        unsupported_client_cell!(@ $msg, "")
199
    }};
200

            
201
    ($msg:expr, $hopnum:expr) => {{
202
        let hop: HopNum = $hopnum;
203
        let hop_display = format!(" from hop {}", hop.display());
204
        unsupported_client_cell!(@ $msg, hop_display)
205
    }};
206

            
207
    (@ $msg:expr, $hopnum_display:expr) => {
208
        Err(crate::Error::CircProto(format!(
209
            "Unexpected {} cell{} on client circuit",
210
            $msg.cmd(),
211
            $hopnum_display,
212
        )))
213
    };
214
}
215

            
216
pub(super) use unsupported_client_cell;
217

            
218
impl Circuit {
219
    /// Create a new non-multipath circuit.
220
    #[allow(clippy::too_many_arguments)]
221
406
    pub(super) fn new(
222
406
        runtime: DynTimeProvider,
223
406
        channel: Arc<Channel>,
224
406
        circ_id: CircId,
225
406
        unique_id: TunnelScopedCircId,
226
406
        input: CircuitRxReceiver,
227
406
        memquota: CircuitAccount,
228
406
        mutable: Arc<MutableState>,
229
406
        padding_ctrl: PaddingController,
230
406
        padding_event_stream: PaddingEventStream,
231
406
        timeouts: Arc<dyn TimeoutEstimator>,
232
406
    ) -> Self {
233
406
        let chan_sender = CircuitCellSender::from_channel_sender(channel.sender());
234

            
235
406
        let crypto_out = OutboundClientCrypt::new();
236
406
        Circuit {
237
406
            runtime,
238
406
            channel,
239
406
            chan_sender,
240
406
            input,
241
406
            crypto_in: InboundClientCrypt::new(),
242
406
            hops: CircHopList::default(),
243
406
            unique_id,
244
406
            circ_id,
245
406
            crypto_out,
246
406
            mutable,
247
406
            #[cfg(feature = "conflux")]
248
406
            conflux_handler: None,
249
406
            padding_ctrl,
250
406
            padding_event_stream,
251
406
            #[cfg(feature = "circ-padding")]
252
406
            padding_block: None,
253
406
            timeouts,
254
406
            memquota,
255
406
        }
256
406
    }
257

            
258
    /// Return the process-unique identifier of this circuit.
259
18522
    pub(super) fn unique_id(&self) -> UniqId {
260
18522
        self.unique_id.unique_id()
261
18522
    }
262

            
263
    /// Return this circuit's identifier.
264
7158
    pub(super) fn circ_id(&self) -> CircId {
265
7158
        self.circ_id
266
7158
    }
267

            
268
    /// Return the shared mutable state of this circuit.
269
458
    pub(super) fn mutable(&self) -> &Arc<MutableState> {
270
458
        &self.mutable
271
458
    }
272

            
273
    /// Add this circuit to a multipath tunnel, by associating it with a new [`TunnelId`],
274
    /// and installing a [`ConfluxMsgHandler`] on this circuit.
275
    ///
276
    /// Once this is called, the circuit will be able to handle conflux cells.
277
    #[cfg(feature = "conflux")]
278
104
    pub(super) fn add_to_conflux_tunnel(
279
104
        &mut self,
280
104
        tunnel_id: TunnelId,
281
104
        conflux_handler: ConfluxMsgHandler,
282
104
    ) {
283
104
        self.unique_id = TunnelScopedCircId::new(tunnel_id, self.unique_id.unique_id());
284
104
        self.conflux_handler = Some(conflux_handler);
285
104
    }
286

            
287
    /// Send a LINK cell to the specified hop.
288
    ///
289
    /// This must be called *after* a [`ConfluxMsgHandler`] is installed
290
    /// on the circuit with [`add_to_conflux_tunnel`](Self::add_to_conflux_tunnel).
291
    #[cfg(feature = "conflux")]
292
104
    pub(super) async fn begin_conflux_link(
293
104
        &mut self,
294
104
        hop: HopNum,
295
104
        cell: AnyRelayMsgOuter,
296
104
        runtime: &tor_rtcompat::DynTimeProvider,
297
156
    ) -> Result<()> {
298
        use tor_rtcompat::SleepProvider as _;
299

            
300
104
        if self.conflux_handler.is_none() {
301
            return Err(internal!(
302
                "tried to send LINK cell before installing a ConfluxMsgHandler?!"
303
            )
304
            .into());
305
104
        }
306

            
307
104
        let cell = SendRelayCell {
308
104
            hop: Some(hop),
309
104
            early: false,
310
104
            cell,
311
104
        };
312
104
        self.send_relay_cell(cell).await?;
313

            
314
104
        let Some(conflux_handler) = self.conflux_handler.as_mut() else {
315
            return Err(internal!("ConfluxMsgHandler disappeared?!").into());
316
        };
317

            
318
104
        Ok(conflux_handler.note_link_sent(runtime.wallclock())?)
319
104
    }
320

            
321
    /// Get the wallclock time when the handshake on this circuit is supposed to time out.
322
    ///
323
    /// Returns `None` if the handshake is not currently in progress.
324
7106
    pub(super) fn conflux_hs_timeout(&self) -> Option<SystemTime> {
325
        cfg_if::cfg_if! {
326
            if #[cfg(feature = "conflux")] {
327
8558
                self.conflux_handler.as_ref().map(|handler| handler.handshake_timeout())?
328
            } else {
329
                None
330
            }
331
        }
332
7106
    }
333

            
334
    /// Handle a [`CtrlMsg::AddFakeHop`](super::CtrlMsg::AddFakeHop) message.
335
    #[cfg(test)]
336
948
    pub(super) fn handle_add_fake_hop(
337
948
        &mut self,
338
948
        format: RelayCellFormat,
339
948
        fwd_lasthop: bool,
340
948
        rev_lasthop: bool,
341
948
        dummy_peer_id: path::HopDetail,
342
948
        // TODO-CGO: Take HopSettings instead of CircParams.
343
948
        // (Do this after we've got the virtual-hop refactorings done for
344
948
        // virtual extending.)
345
948
        params: &crate::client::circuit::CircParameters,
346
948
        done: ReactorResultChannel<()>,
347
948
    ) {
348
        use tor_protover::{Protocols, named};
349

            
350
        use crate::client::circuit::test::DummyCrypto;
351

            
352
948
        assert!(matches!(format, RelayCellFormat::V0));
353
948
        let _ = format; // TODO-CGO: remove this once we have CGO+hs implemented.
354

            
355
948
        let fwd = Box::new(DummyCrypto::new(fwd_lasthop));
356
948
        let rev = Box::new(DummyCrypto::new(rev_lasthop));
357
948
        let binding = None;
358

            
359
948
        let settings = HopSettings::from_params_and_caps(
360
            // This is for testing only, so we'll assume full negotiation took place.
361
948
            crate::circuit::circhop::HopNegotiationType::Full,
362
948
            params,
363
948
            &[named::FLOWCTRL_CC].into_iter().collect::<Protocols>(),
364
        )
365
948
        .expect("Can't construct HopSettings");
366
948
        self.add_hop(dummy_peer_id, fwd, rev, binding, &settings)
367
948
            .expect("could not add hop to circuit");
368
948
        let _ = done.send(Ok(()));
369
948
    }
370

            
371
    /// Encode `msg` and encrypt it, returning the resulting cell
372
    /// and tag that should be expected for an authenticated SENDME sent
373
    /// in response to that cell.
374
4588
    fn encode_relay_cell(
375
4588
        crypto_out: &mut OutboundClientCrypt,
376
4588
        relay_format: RelayCellFormat,
377
4588
        hop: HopNum,
378
4588
        early: bool,
379
4588
        msg: AnyRelayMsgOuter,
380
4588
    ) -> Result<(AnyChanMsg, SendmeTag)> {
381
4588
        let mut body: RelayCellBody = msg
382
4588
            .encode(relay_format, &mut rand::rng())
383
4588
            .map_err(|e| Error::from_cell_enc(e, "relay cell body"))?
384
4588
            .into();
385
4588
        let cmd = if early {
386
72
            ChanCmd::RELAY_EARLY
387
        } else {
388
4516
            ChanCmd::RELAY
389
        };
390
4588
        let tag = crypto_out.encrypt(cmd, &mut body, hop)?;
391
4588
        let msg = Relay::from(BoxedCellBody::from(body));
392
4588
        let msg = if early {
393
72
            AnyChanMsg::RelayEarly(msg.into())
394
        } else {
395
4516
            AnyChanMsg::Relay(msg)
396
        };
397

            
398
4588
        Ok((msg, tag))
399
4588
    }
400

            
401
    /// Encode `msg`, encrypt it, and send it to the 'hop'th hop.
402
    ///
403
    /// If there is insufficient outgoing *circuit-level* or *stream-level*
404
    /// SENDME window, an error is returned instead.
405
    ///
406
    /// Does not check whether the cell is well-formed or reasonable.
407
    ///
408
    /// NOTE: the reactor should not call this function directly, only via
409
    /// [`ConfluxSet::send_relay_cell_on_leg`](super::conflux::ConfluxSet::send_relay_cell_on_leg),
410
    /// which will reroute the message, if necessary to the primary leg.
411
    #[instrument(level = "trace", skip_all)]
412
6882
    pub(super) async fn send_relay_cell(&mut self, msg: SendRelayCell) -> Result<()> {
413
        self.send_relay_cell_inner(msg, None).await
414
4588
    }
415

            
416
    /// As [`send_relay_cell`](Self::send_relay_cell), but takes an optional
417
    /// [`QueuedCellPaddingInfo`] in `padding_info`.
418
    ///
419
    /// If `padding_info` is None, `msg` must be non-padding: we report it as such to the
420
    /// padding controller.
421
    #[instrument(level = "trace", skip_all)]
422
4588
    async fn send_relay_cell_inner(
423
4588
        &mut self,
424
4588
        msg: SendRelayCell,
425
4588
        padding_info: Option<QueuedCellPaddingInfo>,
426
6882
    ) -> Result<()> {
427
        let SendRelayCell {
428
            hop,
429
            early,
430
            cell: msg,
431
4588
        } = msg;
432

            
433
        let is_conflux_link = msg.cmd() == RelayCmd::CONFLUX_LINK;
434
        if !is_conflux_link && self.is_conflux_pending() {
435
            // Note: it is the responsibility of the reactor user to wait until
436
            // at least one of the legs completes the handshake.
437
            return Err(internal!("tried to send cell on unlinked circuit").into());
438
        }
439

            
440
        trace!(
441
            circ_uniq_id = %self.unique_id,
442
            forward_circ_id = %self.circ_id,
443
            cell = ?msg,
444
            "sending relay cell"
445
        );
446

            
447
        // Cloned, because we borrow mutably from self when we get the circhop.
448
        let runtime = self.runtime.clone();
449
        let c_t_w = sendme::cmd_counts_towards_windows(msg.cmd());
450
        let stream_id = msg.stream_id();
451
        let hop = hop.expect("missing hop in client SendRelayCell?!");
452
        let circhop = self.hops.get_mut(hop).ok_or(Error::NoSuchHop)?;
453

            
454
        // We might be out of capacity entirely; see if we are about to hit a limit.
455
        //
456
        // TODO: If we ever add a notion of _recoverable_ errors below, we'll
457
        // need a way to restore this limit, and similarly for about_to_send().
458
        circhop.decrement_outbound_cell_limit()?;
459

            
460
        // We need to apply stream-level flow control *before* encoding the message.
461
        if c_t_w {
462
            if let Some(stream_id) = stream_id {
463
                circhop.about_to_send(stream_id, msg.msg())?;
464
            }
465
        }
466

            
467
        // Save the RelayCmd of the message before it gets consumed below.
468
        // We need this to tell our ConfluxMsgHandler about the cell we've just sent,
469
        // so that it can update its counters.
470
        let relay_cmd = msg.cmd();
471

            
472
        // NOTE(eta): Now that we've encrypted the cell, we *must* either send it or abort
473
        //            the whole circuit (e.g. by returning an error).
474
        let (msg, tag) = Self::encode_relay_cell(
475
            &mut self.crypto_out,
476
            circhop.relay_cell_format(),
477
            hop,
478
            early,
479
            msg,
480
        )?;
481
        // The cell counted for congestion control, inform our algorithm of such and pass down the
482
        // tag for authenticated SENDMEs.
483
        if c_t_w {
484
            circhop.ccontrol().note_data_sent(&runtime, &tag)?;
485
        }
486

            
487
        // Remember that we've enqueued this cell.
488
4588
        let padding_info = padding_info.or_else(|| self.padding_ctrl.queued_data(hop));
489

            
490
        self.send_msg(msg, padding_info).await?;
491

            
492
        #[cfg(feature = "conflux")]
493
        if let Some(conflux) = self.conflux_handler.as_mut() {
494
            conflux.note_cell_sent(relay_cmd);
495
        }
496

            
497
        Ok(())
498
4588
    }
499

            
500
    /// Helper: process a cell on a channel.  Most cells get ignored
501
    /// or rejected; a few get delivered to circuits.
502
    ///
503
    /// Return `CellStatus::CleanShutdown` if we should exit.
504
    ///
505
    // TODO: returning `Vec<CircuitCmd>` means we're unnecessarily
506
    // allocating a `Vec` here. Generally, the number of commands is going to be small
507
    // (usually 1, but > 1 when we start supporting packed cells).
508
    //
509
    // We should consider using smallvec instead. It might also be a good idea to have a
510
    // separate higher-level type splitting this out into Single(CircuitCmd),
511
    // and Multiple(SmallVec<[CircuitCmd; <capacity>]>).
512
512
    pub(super) fn handle_cell(
513
512
        &mut self,
514
512
        handlers: &mut CellHandlers,
515
512
        leg: UniqId,
516
512
        cell: ClientCircChanMsg,
517
512
    ) -> Result<Vec<CircuitCmd>> {
518
512
        trace!(
519
            circ_uniq_id = %self.unique_id,
520
            forward_circ_id = %self.circ_id,
521
            cell = ?cell,
522
            "handling cell"
523
        );
524
        use ClientCircChanMsg::*;
525
512
        match cell {
526
500
            Relay(r) => self.handle_relay_cell(handlers, leg, r),
527
12
            Destroy(d) => {
528
12
                let reason = d.reason();
529
12
                debug!(
530
                    circ_uniq_id = %self.unique_id,
531
                    forward_circ_id = %self.circ_id,
532
                    "Received DESTROY cell. Reason: {} [{}]",
533
                    reason.human_str(),
534
                    reason
535
                );
536

            
537
18
                self.handle_destroy_cell().map(|c| vec![c])
538
            }
539
        }
540
512
    }
541

            
542
    /// Decode `cell`, returning its corresponding hop number, tag,
543
    /// and decoded body.
544
500
    fn decode_relay_cell(
545
500
        &mut self,
546
500
        cell: Relay,
547
500
    ) -> Result<(HopNum, SendmeTag, RelayCellDecoderResult)> {
548
        // This is always RELAY, not RELAY_EARLY, so long as this code is client-only.
549
500
        let cmd = cell.cmd();
550
500
        let mut body = cell.into_relay_body().into();
551

            
552
        // Decrypt the cell. If it's recognized, then find the
553
        // corresponding hop.
554
500
        let (hopnum, tag) = self.crypto_in.decrypt(cmd, &mut body)?;
555

            
556
        // Decode the cell.
557
500
        let decode_res = self
558
500
            .hop_mut(hopnum)
559
500
            .ok_or_else(|| {
560
                Error::from(internal!(
561
                    "Trying to decode cell from nonexistent hop {:?}",
562
                    hopnum
563
                ))
564
            })?
565
500
            .decode(body.into())?;
566

            
567
500
        Ok((hopnum, tag, decode_res))
568
500
    }
569

            
570
    /// React to a Relay or RelayEarly cell.
571
500
    fn handle_relay_cell(
572
500
        &mut self,
573
500
        handlers: &mut CellHandlers,
574
500
        leg: UniqId,
575
500
        cell: Relay,
576
500
    ) -> Result<Vec<CircuitCmd>> {
577
500
        let (hopnum, tag, decode_res) = self.decode_relay_cell(cell)?;
578

            
579
500
        if decode_res.is_padding() {
580
            self.padding_ctrl.decrypted_padding(hopnum)?;
581
500
        } else {
582
500
            self.padding_ctrl.decrypted_data(hopnum);
583
500
        }
584

            
585
        // Check whether we are allowed to receive more data for this circuit hop.
586
500
        self.hop_mut(hopnum)
587
500
            .ok_or_else(|| internal!("nonexistent hop {:?}", hopnum))?
588
500
            .decrement_inbound_cell_limit()?;
589

            
590
500
        let c_t_w = decode_res.cmds().any(sendme::cmd_counts_towards_windows);
591

            
592
        // Decrement the circuit sendme windows, and see if we need to
593
        // send a sendme cell.
594
500
        let send_circ_sendme = if c_t_w {
595
96
            self.hop_mut(hopnum)
596
96
                .ok_or_else(|| Error::CircProto("Sendme from nonexistent hop".into()))?
597
96
                .ccontrol()
598
96
                .note_data_received()?
599
        } else {
600
404
            false
601
        };
602

            
603
500
        let mut circ_cmds = vec![];
604
        // If we do need to send a circuit-level SENDME cell, do so.
605
500
        if send_circ_sendme {
606
            // This always sends a V1 (tagged) sendme cell, and thereby assumes
607
            // that SendmeEmitMinVersion is no more than 1.  If the authorities
608
            // every increase that parameter to a higher number, this will
609
            // become incorrect.  (Higher numbers are not currently defined.)
610
            let sendme = Sendme::from(tag);
611
            let cell = AnyRelayMsgOuter::new(None, sendme.into());
612
            circ_cmds.push(CircuitCmd::Send(SendRelayCell {
613
                hop: Some(hopnum),
614
                early: false,
615
                cell,
616
            }));
617

            
618
            // Inform congestion control of the SENDME we are sending. This is a circuit level one.
619
            self.hop_mut(hopnum)
620
                .ok_or_else(|| {
621
                    Error::from(internal!(
622
                        "Trying to send SENDME to nonexistent hop {:?}",
623
                        hopnum
624
                    ))
625
                })?
626
                .ccontrol()
627
                .note_sendme_sent()?;
628
500
        }
629

            
630
500
        let (mut msgs, incomplete) = decode_res.into_parts();
631
932
        while let Some(msg) = msgs.next() {
632
500
            let msg_status = self.handle_relay_msg(handlers, hopnum, leg, c_t_w, msg)?;
633

            
634
152
            match msg_status {
635
280
                None => continue,
636
                Some(msg @ CircuitCmd::CleanShutdown) => {
637
                    for m in msgs {
638
                        debug!(
639
                            "{id}: Ignoring relay msg received after triggering shutdown: {m:?}",
640
                            id = self.unique_id
641
                        );
642
                    }
643
                    if let Some(incomplete) = incomplete {
644
                        debug!(
645
                            "{id}: Ignoring partial relay msg received after triggering shutdown: {:?}",
646
                            incomplete,
647
                            id = self.unique_id,
648
                        );
649
                    }
650
                    circ_cmds.push(msg);
651
                    return Ok(circ_cmds);
652
                }
653
152
                Some(msg) => {
654
152
                    circ_cmds.push(msg);
655
152
                }
656
            }
657
        }
658

            
659
432
        Ok(circ_cmds)
660
500
    }
661

            
662
    /// Handle a single incoming relay message.
663
500
    fn handle_relay_msg(
664
500
        &mut self,
665
500
        handlers: &mut CellHandlers,
666
500
        hopnum: HopNum,
667
500
        leg: UniqId,
668
500
        cell_counts_toward_windows: bool,
669
500
        msg: UnparsedRelayMsg,
670
500
    ) -> Result<Option<CircuitCmd>> {
671
        // If this msg wants/refuses to have a Stream ID, does it
672
        // have/not have one?
673
500
        let streamid = msg_streamid(&msg)?;
674

            
675
        // If this doesn't have a StreamId, it's a meta cell,
676
        // not meant for a particular stream.
677
500
        let Some(streamid) = streamid else {
678
232
            return self.handle_meta_cell(handlers, hopnum, msg);
679
        };
680

            
681
        #[cfg(feature = "conflux")]
682
268
        let msg = if let Some(conflux) = self.conflux_handler.as_mut() {
683
76
            match conflux.action_for_msg(hopnum, cell_counts_toward_windows, streamid, msg)? {
684
60
                ConfluxAction::Deliver(msg) => {
685
                    // The message either doesn't count towards the sequence numbers
686
                    // or is already well-ordered, so we're ready to handle it.
687

            
688
                    // It's possible that some of our buffered messages are now ready to be
689
                    // handled. We don't check that here, however, because that's handled
690
                    // by the reactor main loop.
691
60
                    msg
692
                }
693
16
                ConfluxAction::Enqueue(msg) => {
694
                    // Tell the reactor to enqueue this msg
695
16
                    return Ok(Some(CircuitCmd::Enqueue(msg)));
696
                }
697
            }
698
        } else {
699
            // If we don't have a conflux_handler, it means this circuit is not part of
700
            // a conflux tunnel, so we can just process the message.
701
192
            msg
702
        };
703

            
704
252
        self.handle_in_order_relay_msg(
705
252
            handlers,
706
252
            hopnum,
707
252
            leg,
708
252
            cell_counts_toward_windows,
709
252
            streamid,
710
252
            msg,
711
        )
712
500
    }
713

            
714
    /// Handle a single incoming relay message that is known to be in order.
715
268
    pub(super) fn handle_in_order_relay_msg(
716
268
        &mut self,
717
268
        handlers: &mut CellHandlers,
718
268
        hopnum: HopNum,
719
268
        leg: UniqId,
720
268
        cell_counts_toward_windows: bool,
721
268
        streamid: StreamId,
722
268
        msg: UnparsedRelayMsg,
723
268
    ) -> Result<Option<CircuitCmd>> {
724
268
        let now = self.runtime.now();
725

            
726
        #[cfg(feature = "conflux")]
727
268
        if let Some(conflux) = self.conflux_handler.as_mut() {
728
76
            conflux.inc_last_seq_delivered(&msg);
729
192
        }
730

            
731
268
        let path = self.mutable.path();
732

            
733
268
        let nonexistent_hop_err = || Error::CircProto("Cell from nonexistent hop!".into());
734
268
        let hop = self.hop_mut(hopnum).ok_or_else(nonexistent_hop_err)?;
735

            
736
268
        let hop_detail = path
737
268
            .iter()
738
268
            .nth(usize::from(hopnum))
739
268
            .ok_or_else(nonexistent_hop_err)?;
740

            
741
        // Returns the original message if it's an incoming stream request
742
        // that we need to handle.
743
268
        let res = hop.handle_msg(hop_detail, cell_counts_toward_windows, streamid, msg, now)?;
744

            
745
        // If it was an incoming stream request, we don't need to worry about
746
        // sending an XOFF as there's no stream data within this message.
747
264
        if let Some(msg) = res {
748
            cfg_if::cfg_if! {
749
                if #[cfg(feature = "hs-service")] {
750
48
                    return self.handle_incoming_stream_request(
751
48
                        handlers,
752
48
                        msg,
753
48
                        streamid,
754
48
                        hopnum,
755
48
                        leg,
756
                        // This is an onion service stream,
757
                        // so we want sidechannel mitigations for flow control.
758
48
                        WithSidechannelMitigations::Enabled,
759
                    );
760
                } else {
761
                    return Err(
762
                        Error::CircProto(format!("Cannot handle {} cells on this circuit", msg.cmd())),
763
                    );
764
                }
765
            }
766
216
        }
767

            
768
        // We may want to send an XOFF if the incoming buffer is too large.
769
216
        if let Some(cell) = hop.maybe_send_xoff(streamid)? {
770
            let cell = AnyRelayMsgOuter::new(Some(streamid), cell.into());
771
            let cell = SendRelayCell {
772
                hop: Some(hopnum),
773
                early: false,
774
                cell,
775
            };
776
            return Ok(Some(CircuitCmd::Send(cell)));
777
216
        }
778

            
779
216
        Ok(None)
780
268
    }
781

            
782
    /// Handle a conflux message coming from the specified hop.
783
    ///
784
    /// Returns an error if
785
    ///
786
    ///   * this is not a conflux circuit (i.e. it doesn't have a [`ConfluxMsgHandler`])
787
    ///   * this is a client circuit and the conflux message originated an unexpected hop
788
    ///   * the cell was sent in violation of the handshake protocol
789
    #[cfg(feature = "conflux")]
790
128
    fn handle_conflux_msg(
791
128
        &mut self,
792
128
        hop: HopNum,
793
128
        msg: UnparsedRelayMsg,
794
128
    ) -> Result<Option<ConfluxCmd>> {
795
128
        let Some(conflux_handler) = self.conflux_handler.as_mut() else {
796
            // If conflux is not enabled, tear down the circuit
797
            // (see 4.2.1. Cell Injection Side Channel Mitigations in prop329)
798
16
            return Err(Error::CircProto(format!(
799
16
                "Received {} cell from hop {} on non-conflux client circuit?!",
800
16
                msg.cmd(),
801
16
                hop.display(),
802
16
            )));
803
        };
804

            
805
112
        Ok(conflux_handler.handle_conflux_msg(msg, hop))
806
128
    }
807

            
808
    /// For conflux: return the sequence number of the last cell sent on this leg.
809
    ///
810
    /// Returns an error if this circuit is not part of a conflux set.
811
    #[cfg(feature = "conflux")]
812
48
    pub(super) fn last_seq_sent(&self) -> Result<u64> {
813
48
        let handler = self
814
48
            .conflux_handler
815
48
            .as_ref()
816
48
            .ok_or_else(|| internal!("tried to get last_seq_sent of non-conflux circ"))?;
817

            
818
48
        Ok(handler.last_seq_sent())
819
48
    }
820

            
821
    /// For conflux: set the sequence number of the last cell sent on this leg.
822
    ///
823
    /// Returns an error if this circuit is not part of a conflux set.
824
    #[cfg(feature = "conflux")]
825
8
    pub(super) fn set_last_seq_sent(&mut self, n: u64) -> Result<()> {
826
8
        let handler = self
827
8
            .conflux_handler
828
8
            .as_mut()
829
8
            .ok_or_else(|| internal!("tried to get last_seq_sent of non-conflux circ"))?;
830

            
831
8
        handler.set_last_seq_sent(n);
832
8
        Ok(())
833
8
    }
834

            
835
    /// For conflux: return the sequence number of the last cell received on this leg.
836
    ///
837
    /// Returns an error if this circuit is not part of a conflux set.
838
    #[cfg(feature = "conflux")]
839
32
    pub(super) fn last_seq_recv(&self) -> Result<u64> {
840
32
        let handler = self
841
32
            .conflux_handler
842
32
            .as_ref()
843
32
            .ok_or_else(|| internal!("tried to get last_seq_recv of non-conflux circ"))?;
844

            
845
32
        Ok(handler.last_seq_recv())
846
32
    }
847

            
848
    /// A helper for handling incoming stream requests.
849
    ///
850
    // TODO: can we make this a method on CircHop to avoid the double HopNum lookup?
851
    #[cfg(feature = "hs-service")]
852
48
    fn handle_incoming_stream_request(
853
48
        &mut self,
854
48
        handlers: &mut CellHandlers,
855
48
        msg: UnparsedRelayMsg,
856
48
        stream_id: StreamId,
857
48
        hop_num: HopNum,
858
48
        leg: UniqId,
859
48
        with_sidechannel_mitigations: WithSidechannelMitigations,
860
48
    ) -> Result<Option<CircuitCmd>> {
861
        use tor_cell::relaycell::msg::EndReason;
862
        use tor_error::into_internal;
863
        use tor_log_ratelim::log_ratelim;
864

            
865
        use crate::stream::incoming::StreamReqInfo;
866

            
867
        // We need to construct this early so that we don't double-borrow &mut self
868

            
869
48
        let Some(handler) = handlers.incoming_stream_req_handler.as_mut() else {
870
            return Err(Error::CircProto(
871
                "Cannot handle BEGIN cells on this circuit".into(),
872
            ));
873
        };
874

            
875
        // The handler's hop_num is only ever set to None for relays.
876
48
        let expected_hop_num = handler
877
48
            .hop_num
878
48
            .ok_or_else(|| internal!("Handler HopNum is None in client impl?!"))?;
879

            
880
48
        if hop_num != expected_hop_num {
881
12
            return Err(Error::CircProto(format!(
882
12
                "Expecting incoming streams from {}, but received {} cell from unexpected hop {}",
883
12
                expected_hop_num.display(),
884
12
                msg.cmd(),
885
12
                hop_num.display()
886
12
            )));
887
36
        }
888

            
889
36
        let message_closes_stream = handler.cmd_checker.check_msg(&msg)? == StreamStatus::Closed;
890

            
891
        // TODO: we've already looked up the `hop` in handle_relay_cell, so we shouldn't
892
        // have to look it up again! However, we can't pass the `&mut hop` reference from
893
        // `handle_relay_cell` to this function, because that makes Rust angry (we'd be
894
        // borrowing self as mutable more than once).
895
        //
896
        // TODO: we _could_ use self.hops.get_mut(..) instead self.hop_mut(..) inside
897
        // handle_relay_cell to work around the problem described above
898
36
        let hop = self.hops.get_mut(hop_num).ok_or(Error::CircuitClosed)?;
899

            
900
36
        if message_closes_stream {
901
            hop.ending_msg_received(stream_id)?;
902

            
903
            return Ok(None);
904
36
        }
905

            
906
36
        let begin = msg
907
36
            .decode::<Begin>()
908
36
            .map_err(|e| Error::from_bytes_err(e, "Invalid Begin message"))?
909
36
            .into_msg();
910

            
911
36
        let req = IncomingStreamRequest::Begin(begin);
912

            
913
        {
914
            use crate::stream::IncomingStreamRequestDisposition::*;
915

            
916
36
            let ctx = crate::stream::IncomingStreamRequestContext { request: &req };
917
            // IMPORTANT: super::syncview::CircHopSyncView::n_open_streams() (called via disposition() below)
918
            // accesses the stream map mutexes!
919
            //
920
            // This means it's very important not to call this function while any of the hop's
921
            // stream map mutex is held.
922
36
            let view = CircHopSyncView::new(hop.outbound());
923

            
924
36
            match handler.filter.as_mut().disposition(&ctx, &view)? {
925
36
                Accept => {}
926
                CloseCircuit => return Ok(Some(CircuitCmd::CleanShutdown)),
927
                RejectRequest(end) => {
928
                    let end_msg = AnyRelayMsgOuter::new(Some(stream_id), end.into());
929
                    let cell = SendRelayCell {
930
                        hop: Some(hop_num),
931
                        early: false,
932
                        cell: end_msg,
933
                    };
934
                    return Ok(Some(CircuitCmd::Send(cell)));
935
                }
936
            }
937
        }
938

            
939
        // TODO: Sadly, we need to look up `&mut hop` yet again,
940
        // since we needed to pass `&self.hops` by reference to our filter above. :(
941
36
        let hop = self.hops.get_mut(hop_num).ok_or(Error::CircuitClosed)?;
942
36
        let relay_cell_format = hop.relay_cell_format();
943

            
944
36
        let memquota = StreamAccount::new(&self.memquota)?;
945

            
946
36
        let cmd_checker = InboundDataCmdChecker::new_connected();
947
36
        let stream_components = hop.add_ent_with_id(
948
36
            self.chan_sender.time_provider(),
949
36
            stream_id,
950
36
            cmd_checker,
951
36
            with_sidechannel_mitigations,
952
36
            &memquota,
953
        )?;
954

            
955
36
        let outcome = Pin::new(&mut handler.incoming_sender).try_send(StreamReqInfo {
956
36
            req,
957
36
            stream_id,
958
36
            hop: Some((leg, hop_num).into()),
959
36
            stream_components,
960
36
            memquota,
961
36
            relay_cell_format,
962
36
        });
963

            
964
36
        log_ratelim!("Delivering message to incoming stream handler"; outcome);
965

            
966
36
        if let Err(e) = outcome {
967
            if e.is_full() {
968
                // The IncomingStreamRequestHandler's stream is full; it isn't
969
                // handling requests fast enough. So instead, we reply with an
970
                // END cell.
971
                let end_msg = AnyRelayMsgOuter::new(
972
                    Some(stream_id),
973
                    End::new_with_reason(EndReason::RESOURCELIMIT).into(),
974
                );
975

            
976
                let cell = SendRelayCell {
977
                    hop: Some(hop_num),
978
                    early: false,
979
                    cell: end_msg,
980
                };
981
                return Ok(Some(CircuitCmd::Send(cell)));
982
            } else if e.is_disconnected() {
983
                // The IncomingStreamRequestHandler's stream has been dropped.
984
                // In the Tor protocol as it stands, this always means that the
985
                // circuit itself is out-of-use and should be closed. (See notes
986
                // on `allow_stream_requests.`)
987
                //
988
                // Note that we will _not_ reach this point immediately after
989
                // the IncomingStreamRequestHandler is dropped; we won't hit it
990
                // until we next get an incoming request.  Thus, if we do later
991
                // want to add early detection for a dropped
992
                // IncomingStreamRequestHandler, we need to do it elsewhere, in
993
                // a different way.
994
                debug!(
995
                    circ_uniq_id = %self.unique_id,
996
                    forward_circ_id = %self.circ_id,
997
                    "Incoming stream request receiver dropped",
998
                );
999
                // This will _cause_ the circuit to get closed.
                return Err(Error::CircuitClosed);
            } else {
                // There are no errors like this with the current design of
                // futures::mpsc, but we shouldn't just ignore the possibility
                // that they'll be added later.
                return Err(Error::from((into_internal!(
                    "try_send failed unexpectedly"
                ))(e)));
            }
36
        }
36
        Ok(None)
48
    }
    /// Helper: process a destroy cell.
    #[allow(clippy::unnecessary_wraps)]
12
    fn handle_destroy_cell(&mut self) -> Result<CircuitCmd> {
        // I think there is nothing more to do here.
12
        Ok(CircuitCmd::CleanShutdown)
12
    }
    /// Handle a [`CtrlMsg::Create`](super::CtrlMsg::Create) message.
78
    pub(super) async fn handle_create(
78
        &mut self,
78
        recv_created: oneshot::Receiver<CreateResponse>,
78
        handshake: CircuitHandshake,
78
        settings: HopSettings,
78
        done: ReactorResultChannel<()>,
117
    ) -> StdResult<(), ReactorError> {
78
        let ret = match handshake {
16
            CircuitHandshake::CreateFast => self.create_firsthop_fast(recv_created, settings).await,
            CircuitHandshake::Ntor {
24
                public_key,
24
                ed_identity,
            } => {
24
                self.create_firsthop_ntor(recv_created, ed_identity, public_key, settings)
24
                    .await
            }
38
            CircuitHandshake::NtorV3 { public_key } => {
38
                self.create_firsthop_ntor_v3(recv_created, public_key, settings)
38
                    .await
            }
        };
78
        let _ = done.send(ret); // don't care if sender goes away
        // TODO: maybe we don't need to flush here?
        // (we could let run_once() handle all the flushing)
78
        self.chan_sender.flush().await?;
78
        Ok(())
78
    }
    /// Helper: create the first hop of a circuit.
    ///
    /// This is parameterized not just on the RNG, but a wrapper object to
    /// build the right kind of create cell, and a handshake object to perform
    /// the cryptographic handshake.
78
    async fn create_impl<H, W, M>(
78
        &mut self,
78
        recvcreated: oneshot::Receiver<CreateResponse>,
78
        wrap: &W,
78
        key: &H::KeyType,
78
        mut settings: HopSettings,
78
        msg: &M,
78
    ) -> Result<()>
78
    where
78
        H: ClientHandshake + HandshakeAuxDataHandler,
78
        W: CreateHandshakeWrap,
78
        H::KeyGen: KeyGenerator,
78
        M: Borrow<H::ClientAuxData>,
78
    {
        // We don't need to shut down the circuit on failure here, since this
        // function consumes the PendingClientCirc and only returns
        // a ClientCirc on success.
78
        let (state, msg) = H::client1(&mut rand::rng(), key, msg)?;
78
        let create_cell = wrap.to_chanmsg(msg);
78
        trace!(
            circ_uniq_id = %self.unique_id,
            forward_circ_id = %self.circ_id,
            create = %create_cell.cmd(),
            "Extending to hop 1",
        );
78
        self.send_msg(create_cell, None).await?;
78
        let reply = recvcreated
78
            .await
78
            .map_err(|_| Error::CircProto("Circuit closed while waiting".into()))?;
78
        let relay_handshake = wrap.decode_chanmsg(reply)?;
64
        let (server_msg, keygen) = H::client2(state, relay_handshake)?;
62
        H::handle_server_aux_data(&mut settings, &server_msg)?;
62
        let BoxedClientLayer { fwd, back, binding } = settings
62
            .relay_crypt_protocol()
62
            .construct_client_layers(HandshakeRole::Initiator, keygen)?;
62
        trace!(
            circ_uniq_id = %self.unique_id,
            forward_circ_id = %self.circ_id,
            "Handshake complete; circuit created."
        );
62
        let peer_id = self.channel.target().clone();
62
        self.add_hop(
62
            path::HopDetail::Relay(peer_id),
62
            fwd,
62
            back,
62
            binding,
62
            &settings,
        )?;
62
        Ok(())
78
    }
    /// Use the (questionable!) CREATE_FAST handshake to connect to the
    /// first hop of this circuit.
    ///
    /// There's no authentication in CREATE_FAST,
    /// so we don't need to know whom we're connecting to: we're just
    /// connecting to whichever relay the channel is for.
16
    async fn create_firsthop_fast(
16
        &mut self,
16
        recvcreated: oneshot::Receiver<CreateResponse>,
16
        settings: HopSettings,
24
    ) -> Result<()> {
        // In a CREATE_FAST handshake, we can't negotiate a format other than this.
16
        let wrap = CreateFastWrap;
16
        self.create_impl::<CreateFastClient, _, _>(recvcreated, &wrap, &(), settings, &())
16
            .await
16
    }
    /// Use the ntor handshake to connect to the first hop of this circuit.
    ///
    /// Note that the provided keys must match the channel's target,
    /// or the handshake will fail.
24
    async fn create_firsthop_ntor(
24
        &mut self,
24
        recvcreated: oneshot::Receiver<CreateResponse>,
24
        ed_identity: pk::ed25519::Ed25519Identity,
24
        pubkey: NtorPublicKey,
24
        settings: HopSettings,
36
    ) -> Result<()> {
        // Exit now if we have an Ed25519 or RSA identity mismatch.
24
        let target = RelayIds::builder()
24
            .ed_identity(ed_identity)
24
            .rsa_identity(pubkey.id)
24
            .build()
24
            .expect("Unable to build RelayIds");
24
        self.channel.check_match(&target)?;
24
        let wrap = Create2Wrap {
24
            handshake_type: HandshakeType::NTOR,
24
        };
24
        self.create_impl::<NtorClient, _, _>(recvcreated, &wrap, &pubkey, settings, &())
24
            .await
24
    }
    /// Use the ntor-v3 handshake to connect to the first hop of this circuit.
    ///
    /// Note that the provided key must match the channel's target,
    /// or the handshake will fail.
38
    async fn create_firsthop_ntor_v3(
38
        &mut self,
38
        recvcreated: oneshot::Receiver<CreateResponse>,
38
        pubkey: NtorV3PublicKey,
38
        settings: HopSettings,
57
    ) -> Result<()> {
        // Exit now if we have a mismatched key.
38
        let target = RelayIds::builder()
38
            .ed_identity(pubkey.id)
38
            .build()
38
            .expect("Unable to build RelayIds");
38
        self.channel.check_match(&target)?;
        // Set the client extensions.
38
        let client_extensions = settings.circuit_request_extensions()?;
38
        let wrap = Create2Wrap {
38
            handshake_type: HandshakeType::NTOR_V3,
38
        };
38
        self.create_impl::<NtorV3Client, _, _>(
38
            recvcreated,
38
            &wrap,
38
            &pubkey,
38
            settings,
38
            &client_extensions,
38
        )
38
        .await
38
    }
    /// Add a hop to the end of this circuit.
    ///
    /// Will return an error if the circuit already has [`u8::MAX`] hops.
1034
    pub(super) fn add_hop(
1034
        &mut self,
1034
        peer_id: path::HopDetail,
1034
        fwd: Box<dyn OutboundClientLayer + 'static + Send>,
1034
        rev: Box<dyn InboundClientLayer + 'static + Send>,
1034
        binding: Option<CircuitBinding>,
1034
        settings: &HopSettings,
1034
    ) -> StdResult<(), Bug> {
1034
        let hop_num = self.hops.len();
1034
        debug_assert_eq!(hop_num, usize::from(self.num_hops()));
        // There are several places in the code that assume that a `usize` hop number
        // can be cast or converted to a `u8` hop number,
        // so this check is important to prevent panics or incorrect behaviour.
1034
        if hop_num == usize::from(u8::MAX) {
            return Err(internal!(
                "cannot add more hops to a circuit with `u8::MAX` hops"
            ));
1034
        }
1034
        let hop_num = (hop_num as u8).into();
1034
        let hop = CircHop::new(self.unique_id, self.circ_id, hop_num, settings);
1034
        self.hops.push(hop);
1034
        self.crypto_in.add_layer(rev);
1034
        self.crypto_out.add_layer(fwd);
1034
        self.mutable.add_hop(peer_id, binding);
1034
        Ok(())
1034
    }
    /// Handle a RELAY cell on this circuit with stream ID 0.
    ///
    /// NOTE(prop349): this is part of Arti's "Base Circuit Hop Handler".
    /// This function returns a `CircProto` error if `msg` is an unsupported,
    /// unexpected, or otherwise invalid message:
    ///
    ///   * unexpected messages are rejected by returning an error using
    ///     [`unsupported_client_cell`]
    ///   * SENDME/TRUNCATED messages are rejected if they don't parse
    ///   * SENDME authentication tags are validated inside [`Circuit::handle_sendme`]
    ///   * conflux cells are handled in the client [`ConfluxMsgHandler`]
    ///
    /// The error is propagated all the way up to [`Circuit::handle_cell`],
    /// and eventually ends up being returned from the reactor's `run_once` function,
    /// causing it to shut down.
232
    fn handle_meta_cell(
232
        &mut self,
232
        handlers: &mut CellHandlers,
232
        hopnum: HopNum,
232
        msg: UnparsedRelayMsg,
232
    ) -> Result<Option<CircuitCmd>> {
        // SENDME cells and TRUNCATED get handled internally by the circuit.
        // TODO: This pattern (Check command, try to decode, map error) occurs
        // several times, and would be good to extract simplify. Such
        // simplification is obstructed by a couple of factors: First, that
        // there is not currently a good way to get the RelayCmd from _type_ of
        // a RelayMsg.  Second, that decode() [correctly] consumes the
        // UnparsedRelayMsg.  I tried a macro-based approach, and didn't care
        // for it. -nickm
232
        if msg.cmd() == RelayCmd::SENDME {
44
            let sendme = msg
44
                .decode::<Sendme>()
44
                .map_err(|e| Error::from_bytes_err(e, "sendme message"))?
44
                .into_msg();
44
            return Ok(Some(CircuitCmd::HandleSendMe {
44
                hop: hopnum,
44
                sendme,
44
            }));
188
        }
188
        if msg.cmd() == RelayCmd::TRUNCATED {
            let truncated = msg
                .decode::<Truncated>()
                .map_err(|e| Error::from_bytes_err(e, "truncated message"))?
                .into_msg();
            let reason = truncated.reason();
            debug!(
                circ_uniq_id = %self.unique_id,
                forward_circ_id = %self.circ_id,
                "Truncated from hop {}. Reason: {} [{}]",
                hopnum.display(),
                reason.human_str(),
                reason
            );
            return Ok(Some(CircuitCmd::CleanShutdown));
188
        }
188
        if msg.cmd() == RelayCmd::DROP {
            cfg_if::cfg_if! {
                if #[cfg(feature = "circ-padding")] {
                    return Ok(None);
                } else {
                    use crate::util::err::ExcessPadding;
                    return Err(Error::ExcessPadding(ExcessPadding::NoPaddingNegotiated, hopnum));
                }
            }
188
        }
188
        trace!(
            circ_uniq_id = %self.unique_id,
            forward_circ_id = %self.circ_id,
            cell = ?msg,
            "Received meta-cell"
        );
        #[cfg(feature = "conflux")]
60
        if matches!(
188
            msg.cmd(),
            RelayCmd::CONFLUX_LINK
                | RelayCmd::CONFLUX_LINKED
                | RelayCmd::CONFLUX_LINKED_ACK
                | RelayCmd::CONFLUX_SWITCH
        ) {
128
            let cmd = self.handle_conflux_msg(hopnum, msg)?;
112
            return Ok(cmd.map(CircuitCmd::from));
60
        }
60
        if self.is_conflux_pending() {
            warn!(
                circ_uniq_id = %self.unique_id,
                forward_circ_id = %self.circ_id,
                "received unexpected cell {msg:?} on unlinked conflux circuit",
            );
            return Err(Error::CircProto(
                "Received unexpected cell on unlinked circuit".into(),
            ));
60
        }
        // For all other command types, we'll only get them in response
        // to another command, which should have registered a responder.
        //
        // TODO: should the conflux state machine be a meta cell handler?
        // We'd need to add support for multiple meta handlers, and change the
        // MetaCellHandler API to support returning Option<RunOnceCmdInner>
        // (because some cells will require sending a response)
60
        if let Some(mut handler) = handlers.meta_handler.take() {
            // The handler has a TargetHop so we do a quick convert for equality check.
60
            if handler.expected_hop() == (self.unique_id(), hopnum).into() {
                // Somebody was waiting for a message -- maybe this message
48
                let ret = handler.handle_msg(msg, self);
48
                trace!(
                    circ_uniq_id = %self.unique_id,
                    forward_circ_id = %self.circ_id,
                    result = ?ret,
                    "meta handler completed",
                );
24
                match ret {
                    #[cfg(feature = "send-control-msg")]
                    Ok(MetaCellDisposition::Consumed) => {
                        handlers.meta_handler = Some(handler);
                        Ok(None)
                    }
24
                    Ok(MetaCellDisposition::ConversationFinished) => Ok(None),
                    #[cfg(feature = "send-control-msg")]
                    Ok(MetaCellDisposition::CloseCirc) => Ok(Some(CircuitCmd::CleanShutdown)),
24
                    Err(e) => Err(e),
                }
            } else {
                // Somebody wanted a message from a different hop!  Put this
                // one back.
12
                handlers.meta_handler = Some(handler);
12
                unsupported_client_cell!(msg, hopnum)
            }
        } else {
            // No need to call shutdown here, since this error will
            // propagate to the reactor shut it down.
            unsupported_client_cell!(msg)
        }
232
    }
    /// Handle a RELAY_SENDME cell on this circuit with stream ID 0.
    #[instrument(level = "trace", skip_all)]
44
    pub(super) fn handle_sendme(
44
        &mut self,
44
        hopnum: HopNum,
44
        msg: Sendme,
44
        signals: CongestionSignals,
44
    ) -> Result<Option<CircuitCmd>> {
        // Cloned, because we borrow mutably from self when we get the circhop.
44
        let runtime = self.runtime.clone();
        // No need to call "shutdown" on errors in this function;
        // it's called from the reactor task and errors will propagate there.
44
        let hop = self
44
            .hop_mut(hopnum)
44
            .ok_or_else(|| Error::CircProto(format!("Couldn't find hop {}", hopnum.display())))?;
44
        let tag = msg.into_sendme_tag().ok_or_else(||
                // Versions of Tor <=0.3.5 would omit a SENDME tag in this case;
                // but we don't support those any longer.
                 Error::CircProto("missing tag on circuit sendme".into()))?;
        // Update the CC object that we received a SENDME along with possible congestion signals.
44
        hop.ccontrol()
44
            .note_sendme_received(&runtime, tag, signals)?;
40
        Ok(None)
44
    }
    /// Send a message onto the circuit's channel.
    ///
    /// If the channel is ready to accept messages, it will be sent immediately. If not, the message
    /// will be enqueued for sending at a later iteration of the reactor loop.
    ///
    /// `info` is the status returned from the padding controller when we told it we were queueing
    /// this data.  It should be provided whenever possible.
    ///
    /// # Note
    ///
    /// Making use of the enqueuing capabilities of this function is discouraged! You should first
    /// check whether the channel is ready to receive messages (`self.channel.poll_ready`), and
    /// ideally use this to implement backpressure (such that you do not read from other sources
    /// that would send here while you know you're unable to forward the messages on).
    #[instrument(level = "trace", skip_all)]
4666
    async fn send_msg(
4666
        &mut self,
4666
        msg: AnyChanMsg,
4666
        info: Option<QueuedCellPaddingInfo>,
6999
    ) -> Result<()> {
        let cell = AnyChanCell::new(Some(self.circ_id), msg);
        // Note: this future is always `Ready`, so await won't block.
        Pin::new(&mut self.chan_sender)
            .send_unbounded((cell, info))
            .await?;
        Ok(())
4666
    }
    /// Remove all halfstreams that are expired at `now`.
7106
    pub(super) fn remove_expired_halfstreams(&mut self, now: Instant) {
7106
        self.hops.remove_expired_halfstreams(now);
7106
    }
    /// Return a reference to the hop corresponding to `hopnum`, if there is one.
4106
    pub(super) fn hop(&self, hopnum: HopNum) -> Option<&CircHop> {
4106
        self.hops.hop(hopnum)
4106
    }
    /// Return a mutable reference to the hop corresponding to `hopnum`, if there is one.
1788
    pub(super) fn hop_mut(&mut self, hopnum: HopNum) -> Option<&mut CircHop> {
1788
        self.hops.get_mut(hopnum)
1788
    }
    /// Begin a stream with the provided hop in this circuit.
    // TODO: see if there's a way that we can clean this up
    #[allow(clippy::too_many_arguments)]
96
    pub(super) fn begin_stream(
96
        &mut self,
96
        hop_num: HopNum,
96
        message: AnyRelayMsg,
96
        time_prov: &DynTimeProvider,
96
        cmd_checker: AnyCmdChecker,
96
        memquota: &StreamAccount,
96
    ) -> Result<(SendRelayCell, StreamId, ReactorStreamComponents)> {
96
        let Some(hop) = self.hop_mut(hop_num) else {
            return Err(internal!(
                "{}: Attempting to send a BEGIN cell to an unknown hop {hop_num:?}",
                self.unique_id,
            )
            .into());
        };
96
        hop.begin_stream(message, time_prov, cmd_checker, memquota)
96
    }
    /// Close the specified stream
    #[instrument(level = "trace", skip_all)]
64
    pub(super) async fn close_stream(
64
        &mut self,
64
        hop_num: HopNum,
64
        sid: StreamId,
64
        behav: CloseStreamBehavior,
64
        reason: streammap::TerminateReason,
64
        expiry: Instant,
96
    ) -> Result<()> {
        if let Some(hop) = self.hop_mut(hop_num) {
            let res = hop.close_stream(sid, behav, reason, expiry)?;
            if let Some(cell) = res {
                self.send_relay_cell(cell).await?;
            }
        }
        Ok(())
64
    }
    /// Returns true if there are any streams on this circuit
    ///
    /// Important: this function locks the stream map of its each of the [`CircHop`]s
    /// in this circuit, so it must **not** be called from any function where the
    /// stream map lock is held.
52
    pub(super) fn has_streams(&self) -> bool {
52
        self.hops.has_streams()
52
    }
    /// The number of hops in this circuit.
1286
    pub(super) fn num_hops(&self) -> u8 {
        // `Circuit::add_hop` checks to make sure that we never have more than `u8::MAX` hops,
        // so `self.hops.len()` should be safe to cast to a `u8`.
        // If that assumption is violated,
        // we choose to panic rather than silently use the wrong hop due to an `as` cast.
1286
        self.hops
1286
            .len()
1286
            .try_into()
1286
            .expect("`hops.len()` has more than `u8::MAX` hops")
1286
    }
    /// Check whether this circuit has any hops.
4946
    pub(super) fn has_hops(&self) -> bool {
4946
        !self.hops.is_empty()
4946
    }
    /// Get the `HopNum` of the last hop, if this circuit is non-empty.
    ///
    /// Returns `None` if the circuit has no hops.
252
    pub(super) fn last_hop_num(&self) -> Option<HopNum> {
252
        let num_hops = self.num_hops();
252
        if num_hops == 0 {
            // asked for the last hop, but there are no hops
            return None;
252
        }
252
        Some(HopNum::from(num_hops - 1))
252
    }
    /// Get the path of the circuit.
    ///
    /// **Warning:** Do not call while already holding the [`Self::mutable`] lock.
120
    pub(super) fn path(&self) -> Arc<path::Path> {
120
        self.mutable.path()
120
    }
    /// Return a ClockSkew declaring how much clock skew the other side of this channel
    /// claimed that we had when we negotiated the connection.
    pub(super) fn clock_skew(&self) -> ClockSkew {
        self.channel.clock_skew()
    }
    /// Does congestion control use stream SENDMEs for the given `hop`?
    ///
    /// Returns `None` if `hop` doesn't exist.
    pub(super) fn uses_stream_sendme(&self, hop: HopNum) -> Option<bool> {
        let hop = self.hop(hop)?;
        Some(hop.ccontrol().uses_stream_sendme())
    }
    /// Returns whether this is a conflux circuit that is not linked yet.
4560
    pub(super) fn is_conflux_pending(&self) -> bool {
4560
        let Some(status) = self.conflux_status() else {
3260
            return false;
        };
1300
        status != ConfluxStatus::Linked
4560
    }
    /// Returns the conflux status of this circuit.
    ///
    /// Returns `None` if this is not a conflux circuit.
4896
    pub(super) fn conflux_status(&self) -> Option<ConfluxStatus> {
        cfg_if::cfg_if! {
            if #[cfg(feature = "conflux")] {
4896
                self.conflux_handler
4896
                    .as_ref()
5606
                    .map(|handler| handler.status())
            } else {
                None
            }
        }
4896
    }
    /// Returns initial RTT on this leg, measured in the conflux handshake.
    #[cfg(feature = "conflux")]
1512
    pub(super) fn init_rtt(&self) -> Option<Duration> {
1512
        self.conflux_handler
1512
            .as_ref()
2268
            .map(|handler| handler.init_rtt())?
1512
    }
    /// Start or stop padding at the given hop.
    ///
    /// Replaces any previous padder at that hop.
    ///
    /// Return an error if that hop doesn't exist.
    #[cfg(feature = "circ-padding-manual")]
    pub(super) fn set_padding_at_hop(
        &self,
        hop: HopNum,
        padder: Option<padding::CircuitPadder>,
    ) -> Result<()> {
        if self.hop(hop).is_none() {
            return Err(Error::NoSuchHop);
        }
        self.padding_ctrl.install_padder_padding_at_hop(hop, padder);
        Ok(())
    }
    /// Determine how exactly to handle a request to handle padding.
    ///
    /// This is fairly complicated; see the maybenot documentation for more information.
    ///
    /// ## Limitations
    ///
    /// In our current padding implementation, a circuit is either blocked or not blocked:
    /// we do not keep track of which hop is actually doing the blocking.
    #[cfg(feature = "circ-padding")]
    fn padding_disposition(&self, send_padding: &padding::SendPadding) -> CircPaddingDisposition {
        crate::circuit::padding::padding_disposition(
            send_padding,
            &self.chan_sender,
            self.padding_block.as_ref(),
        )
    }
    /// Handle a request from our padding subsystem to send a padding packet.
    #[cfg(feature = "circ-padding")]
    pub(super) async fn send_padding(&mut self, send_padding: padding::SendPadding) -> Result<()> {
        use CircPaddingDisposition::*;
        let target_hop = send_padding.hop;
        match self.padding_disposition(&send_padding) {
            QueuePaddingNormally => {
                let queue_info = self.padding_ctrl.queued_padding(target_hop, send_padding);
                self.queue_padding_cell_for_hop(target_hop, queue_info)
                    .await?;
            }
            QueuePaddingAndBypass => {
                let queue_info = self.padding_ctrl.queued_padding(target_hop, send_padding);
                self.queue_padding_cell_for_hop(target_hop, queue_info)
                    .await?;
            }
            TreatQueuedCellAsPadding => {
                self.padding_ctrl
                    .replaceable_padding_already_queued(target_hop, send_padding);
            }
        }
        Ok(())
    }
    /// Generate and encrypt a padding cell, and send it to a targeted hop.
    ///
    /// Ignores any padding-based blocking.
    #[cfg(feature = "circ-padding")]
    async fn queue_padding_cell_for_hop(
        &mut self,
        target_hop: HopNum,
        queue_info: Option<QueuedCellPaddingInfo>,
    ) -> Result<()> {
        use tor_cell::relaycell::msg::Drop as DropMsg;
        let msg = SendRelayCell {
            hop: Some(target_hop),
            // TODO circpad: we will probably want padding machines that can send EARLY cells.
            early: false,
            cell: AnyRelayMsgOuter::new(None, DropMsg::default().into()),
        };
        self.send_relay_cell_inner(msg, queue_info).await
    }
    /// Enable padding-based blocking,
    /// or change the rule for padding-based blocking to the one in `block`.
    #[cfg(feature = "circ-padding")]
    pub(super) fn start_blocking_for_padding(&mut self, block: padding::StartBlocking) {
        self.chan_sender.start_blocking();
        self.padding_block = Some(block);
    }
    /// Disable padding-based blocking.
    #[cfg(feature = "circ-padding")]
    pub(super) fn stop_blocking_for_padding(&mut self) {
        self.chan_sender.stop_blocking();
        self.padding_block = None;
    }
    /// The estimated circuit build timeout for a circuit of the specified length.
64
    pub(super) fn estimate_cbt(&self, length: usize) -> Duration {
64
        self.timeouts.circuit_build_timeout(length)
64
    }
}
impl Drop for Circuit {
354
    fn drop(&mut self) {
354
        let _ = self.channel.close_circuit(self.circ_id);
354
    }
}