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

pub(crate) mod circhop;

pub(super) mod extender;

use crate::channel::Channel;

use crate::circuit::cell_sender::CircuitCellSender;

use crate::circuit::celltypes::CreateResponse;

use crate::circuit::circhop::HopSettings;

use crate::circuit::create::{Create2Wrap, CreateFastWrap, CreateHandshakeWrap};

use crate::circuit::padding::CircPaddingDisposition;

use crate::circuit::{CircuitRxReceiver, UniqId};

use crate::client::circuit::handshake::{BoxedClientLayer, HandshakeRole};

use crate::client::circuit::padding::{

    self, PaddingController, PaddingEventStream, QueuedCellPaddingInfo,

};

use crate::client::circuit::{ClientCircChanMsg, MutableState, path};

use crate::client::reactor::MetaCellDisposition;

use crate::congestion::CongestionSignals;

use crate::congestion::sendme;

use crate::crypto::binding::CircuitBinding;

use crate::crypto::cell::{

    HopNum, InboundClientCrypt, InboundClientLayer, OutboundClientCrypt, OutboundClientLayer,

    RelayCellBody,

};

use crate::crypto::handshake::fast::CreateFastClient;

use crate::crypto::handshake::ntor::{NtorClient, NtorPublicKey};

use crate::crypto::handshake::ntor_v3::{NtorV3Client, NtorV3PublicKey};

use crate::crypto::handshake::{ClientHandshake, KeyGenerator};

use crate::memquota::{CircuitAccount, SpecificAccount as _, StreamAccount};

use crate::stream::cmdcheck::{AnyCmdChecker, StreamStatus};

use crate::stream::flow_ctrl::state::StreamRateLimit;

use crate::stream::flow_ctrl::xon_xoff::reader::DrainRateRequest;

use crate::stream::queue::{StreamQueueSender, stream_queue};

use crate::stream::{StreamMpscReceiver, msg_streamid};

use crate::streammap;

use crate::tunnel::TunnelScopedCircId;

use crate::util::err::ReactorError;

use crate::util::notify::NotifySender;

use crate::util::timeout::TimeoutEstimator;

use crate::{ClockSkew, Error, Result};

use tor_async_utils::{SinkTrySend as _, SinkTrySendError as _};

use tor_cell::chancell::msg::{AnyChanMsg, HandshakeType, Relay};

use tor_cell::chancell::{AnyChanCell, ChanCmd, CircId};

use tor_cell::chancell::{BoxedCellBody, ChanMsg};

use tor_cell::relaycell::msg::{AnyRelayMsg, End, Sendme, SendmeTag, Truncated};

use tor_cell::relaycell::{

    AnyRelayMsgOuter, RelayCellDecoderResult, RelayCellFormat, RelayCmd, StreamId, UnparsedRelayMsg,

};

use tor_error::{Bug, internal};

use tor_linkspec::RelayIds;

use tor_llcrypto::pk;

use tor_memquota::mq_queue::{ChannelSpec as _, MpscSpec};

use futures::SinkExt as _;

use oneshot_fused_workaround as oneshot;

use postage::watch;

use tor_rtcompat::{DynTimeProvider, SleepProvider as _};

use tracing::{debug, instrument, trace, warn};

use super::{

    CellHandlers, CircuitHandshake, CloseStreamBehavior, ReactorResultChannel, SendRelayCell,

};

use crate::conflux::msghandler::ConfluxStatus;

use std::borrow::Borrow;

use std::pin::Pin;

use std::result::Result as StdResult;

use std::sync::Arc;

use std::time::{Duration, Instant, SystemTime};

use extender::HandshakeAuxDataHandler;

#[cfg(feature = "hs-service")]

use {

    crate::circuit::CircHopSyncView,

    crate::client::stream::{InboundDataCmdChecker, IncomingStreamRequest},

    tor_cell::relaycell::msg::Begin,

};

#[cfg(feature = "conflux")]

use {

    crate::conflux::msghandler::{ConfluxAction, ConfluxCmd, ConfluxMsgHandler, OooRelayMsg},

    crate::tunnel::TunnelId,

};

#[cfg(not(feature = "flowctl-cc"))]

use crate::stream::STREAM_READER_BUFFER;

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

/// A circuit "leg" from a tunnel.

///

/// Regular (non-multipath) circuits have a single leg.

/// Conflux (multipath) circuits have `N` (usually, `N = 2`).

pub(crate) struct Circuit {

    /// The time provider.

    runtime: DynTimeProvider,

    /// The channel this circuit is attached to.

    channel: Arc<Channel>,

    /// Sender object used to actually send cells.

    ///

    /// NOTE: Control messages could potentially add unboundedly to this, although that's

    ///       not likely to happen (and isn't triggereable from the network, either).

    pub(super) chan_sender: CircuitCellSender,

    /// Input stream, on which we receive ChanMsg objects from this circuit's

    /// channel.

    ///

    // TODO: could use a SPSC channel here instead.

    pub(super) input: CircuitRxReceiver,

    /// The cryptographic state for this circuit for inbound cells.

    /// This object is divided into multiple layers, each of which is

    /// shared with one hop of the circuit.

    crypto_in: InboundClientCrypt,

    /// The cryptographic state for this circuit for outbound cells.

    crypto_out: OutboundClientCrypt,

    /// List of hops state objects used by the reactor

    pub(super) hops: CircHopList,

    /// Mutable information about this circuit,

    /// shared with the reactor's `ConfluxSet`.

    mutable: Arc<MutableState>,

    /// This circuit's identifier on the upstream channel.

    channel_id: CircId,

    /// An identifier for logging about this reactor's circuit.

    unique_id: TunnelScopedCircId,

    /// A handler for conflux cells.

    ///

    /// Set once the conflux handshake is initiated by the reactor

    /// using [`Reactor::handle_link_circuits`](super::Reactor::handle_link_circuits).

    #[cfg(feature = "conflux")]

    conflux_handler: Option<ConfluxMsgHandler>,

    /// A padding controller to which padding-related events should be reported.

    padding_ctrl: PaddingController,

    /// An event stream telling us about padding-related events.

    //

    // TODO: it would be nice to have all of these streams wrapped in a single

    // SelectAll, but we can't really do that, since we need the ability to move them

    // from one conflux set to another, and a SelectAll doesn't let you actually

    // remove one of its constituent streams.  This issue might get solved along

    // with the the rest of the next reactor refactoring.

    pub(super) padding_event_stream: PaddingEventStream,

    /// Current rules for blocking traffic, according to the padding controller.

    #[cfg(feature = "circ-padding")]

    padding_block: Option<padding::StartBlocking>,

    /// The circuit timeout estimator.

    ///

    /// Used for computing half-stream expiration.

    timeouts: Arc<dyn TimeoutEstimator>,

    /// Memory quota account

    #[allow(dead_code)] // Partly here to keep it alive as long as the circuit

    memquota: CircuitAccount,

}

/// A command to run in response to a circuit event.

///

/// Unlike `RunOnceCmdInner`, doesn't know anything about `UniqId`s.

/// The user of the `CircuitCmd`s is supposed to know the `UniqId`

/// of the circuit the `CircuitCmd` came from.

///

/// This type gets mapped to a `RunOnceCmdInner` in the circuit reactor.

#[derive(Debug, derive_more::From)]

pub(super) enum CircuitCmd {

    /// Send a RELAY cell on the circuit leg this command originates from.

    Send(SendRelayCell),

    /// Handle a SENDME message received on the circuit leg this command originates from.

    HandleSendMe {

        /// The hop number.

        hop: HopNum,

        /// The SENDME message to handle.

        sendme: Sendme,

    },

    /// Close the specified stream on the circuit leg this command originates from.

    CloseStream {

        /// The hop number.

        hop: HopNum,

        /// The ID of the stream to close.

        sid: StreamId,

        /// The stream-closing behavior.

        behav: CloseStreamBehavior,

        /// The reason for closing the stream.

        reason: streammap::TerminateReason,

    },

    /// Perform an action resulting from handling a conflux cell.

    #[cfg(feature = "conflux")]

    Conflux(ConfluxCmd),

    /// Perform a clean shutdown on this circuit.

    CleanShutdown,

    /// Enqueue an out-of-order cell in the reactor.

    #[cfg(feature = "conflux")]

    Enqueue(OooRelayMsg),

}

/// Return a `CircProto` error for the specified unsupported cell.

///

/// This error will shut down the reactor.

///

/// Note: this is a macro to simplify usage (this way the caller doesn't

/// need to .map() the result to the appropriate type)

macro_rules! unsupported_client_cell {

    ($msg:expr) => {{

        unsupported_client_cell!(@ $msg, "")

    }};

    ($msg:expr, $hopnum:expr) => {{

        let hop: HopNum = $hopnum;

        let hop_display = format!(" from hop {}", hop.display());

        unsupported_client_cell!(@ $msg, hop_display)

    }};

    (@ $msg:expr, $hopnum_display:expr) => {

        Err(crate::Error::CircProto(format!(

            "Unexpected {} cell{} on client circuit",

            $msg.cmd(),

            $hopnum_display,

        )))

    };

}

pub(super) use unsupported_client_cell;

impl Circuit {

    /// Create a new non-multipath circuit.

    #[allow(clippy::too_many_arguments)]

    /// Return the process-unique identifier of this circuit.

    /// Return the shared mutable state of this circuit.

    /// Add this circuit to a multipath tunnel, by associating it with a new [`TunnelId`],

    /// and installing a [`ConfluxMsgHandler`] on this circuit.

    ///

    /// Once this is called, the circuit will be able to handle conflux cells.

    #[cfg(feature = "conflux")]

    /// Send a LINK cell to the specified hop.

    ///

    /// This must be called *after* a [`ConfluxMsgHandler`] is installed

    /// on the circuit with [`add_to_conflux_tunnel`](Self::add_to_conflux_tunnel).

    #[cfg(feature = "conflux")]

        use tor_rtcompat::SleepProvider as _;

        };

    /// Get the wallclock time when the handshake on this circuit is supposed to time out.

    ///

    /// Returns `None` if the handshake is not currently in progress.

        cfg_if::cfg_if! {

            if #[cfg(feature = "conflux")] {

            } else {

                None

            }

        }

    /// Handle a [`CtrlMsg::AddFakeHop`](super::CtrlMsg::AddFakeHop) message.

    #[cfg(test)]

        use tor_protover::{Protocols, named};

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

            // This is for testing only, so we'll assume full negotiation took place.

        )

    /// Encode `msg` and encrypt it, returning the resulting cell

    /// and tag that should be expected for an authenticated SENDME sent

    /// in response to that cell.

        } else {

        };

        } else {

        };

    /// Encode `msg`, encrypt it, and send it to the 'hop'th hop.

    ///

    /// If there is insufficient outgoing *circuit-level* or *stream-level*

    /// SENDME window, an error is returned instead.

    ///

    /// Does not check whether the cell is well-formed or reasonable.

    ///

    /// NOTE: the reactor should not call this function directly, only via

    /// [`ConfluxSet::send_relay_cell_on_leg`](super::conflux::ConfluxSet::send_relay_cell_on_leg),

    /// which will reroute the message, if necessary to the primary leg.

    #[instrument(level = "trace", skip_all)]

        self.send_relay_cell_inner(msg, None).await

    /// As [`send_relay_cell`](Self::send_relay_cell), but takes an optional

    /// [`QueuedCellPaddingInfo`] in `padding_info`.

    ///

    /// If `padding_info` is None, `msg` must be non-padding: we report it as such to the

    /// padding controller.

    #[instrument(level = "trace", skip_all)]

        let SendRelayCell {

            hop,

            early,

            cell: msg,

        let is_conflux_link = msg.cmd() == RelayCmd::CONFLUX_LINK;

        if !is_conflux_link && self.is_conflux_pending() {

            // Note: it is the responsibility of the reactor user to wait until

            // at least one of the legs completes the handshake.

            return Err(internal!("tried to send cell on unlinked circuit").into());

        }

        trace!(circ_id = %self.unique_id, cell = ?msg, "sending relay cell");

        // Cloned, because we borrow mutably from self when we get the circhop.

        let runtime = self.runtime.clone();

        let c_t_w = sendme::cmd_counts_towards_windows(msg.cmd());

        let stream_id = msg.stream_id();

        let hop = hop.expect("missing hop in client SendRelayCell?!");

        let circhop = self.hops.get_mut(hop).ok_or(Error::NoSuchHop)?;

        // We might be out of capacity entirely; see if we are about to hit a limit.

        //

        // TODO: If we ever add a notion of _recoverable_ errors below, we'll

        // need a way to restore this limit, and similarly for about_to_send().

        circhop.decrement_outbound_cell_limit()?;

        // We need to apply stream-level flow control *before* encoding the message.

        if c_t_w {

            if let Some(stream_id) = stream_id {

                circhop.about_to_send(stream_id, msg.msg())?;

            }

        }

        // Save the RelayCmd of the message before it gets consumed below.

        // We need this to tell our ConfluxMsgHandler about the cell we've just sent,

        // so that it can update its counters.

        let relay_cmd = msg.cmd();

        // NOTE(eta): Now that we've encrypted the cell, we *must* either send it or abort

        //            the whole circuit (e.g. by returning an error).

        let (msg, tag) = Self::encode_relay_cell(

            &mut self.crypto_out,

            circhop.relay_cell_format(),

            hop,

            early,

            msg,

        )?;

        // The cell counted for congestion control, inform our algorithm of such and pass down the

        // tag for authenticated SENDMEs.

        if c_t_w {

            circhop.ccontrol().note_data_sent(&runtime, &tag)?;

        }

        // Remember that we've enqueued this cell.

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

        #[cfg(feature = "conflux")]

        if let Some(conflux) = self.conflux_handler.as_mut() {

            conflux.note_cell_sent(relay_cmd);

        }

        Ok(())

    /// Helper: process a cell on a channel.  Most cells get ignored

    /// or rejected; a few get delivered to circuits.

    ///

    /// Return `CellStatus::CleanShutdown` if we should exit.

    ///

    // TODO: returning `Vec<CircuitCmd>` means we're unnecessarily

    // allocating a `Vec` here. Generally, the number of commands is going to be small

    // (usually 1, but > 1 when we start supporting packed cells).

    //

    // We should consider using smallvec instead. It might also be a good idea to have a

    // separate higher-level type splitting this out into Single(CircuitCmd),

    // and Multiple(SmallVec<[CircuitCmd; <capacity>]>).

        use ClientCircChanMsg::*;

                    circ_id = %self.unique_id,

                    reason

                );

            }

        }

    /// Decode `cell`, returning its corresponding hop number, tag,

    /// and decoded body.

        // This is always RELAY, not RELAY_EARLY, so long as this code is client-only.

        // Decrypt the cell. If it's recognized, then find the

        // corresponding hop.

        // Decode the cell.

                    hopnum

                ))

    /// React to a Relay or RelayEarly cell.

        // Check whether we are allowed to receive more data for this circuit hop.

        // Decrement the circuit sendme windows, and see if we need to

        // send a sendme cell.

        } else {

        };

        // If we do need to send a circuit-level SENDME cell, do so.

            // This always sends a V1 (tagged) sendme cell, and thereby assumes

            // that SendmeEmitMinVersion is no more than 1.  If the authorities

            // every increase that parameter to a higher number, this will

            // become incorrect.  (Higher numbers are not currently defined.)

            // Inform congestion control of the SENDME we are sending. This is a circuit level one.

                        hopnum

                    ))

                            id = self.unique_id

                        );

                    }

                            incomplete,

                            id = self.unique_id,

                        );

                }

            }

        }

    /// Handle a single incoming relay message.

        // If this msg wants/refuses to have a Stream ID, does it

        // have/not have one?

        // If this doesn't have a StreamId, it's a meta cell,

        // not meant for a particular stream.

        };

        #[cfg(feature = "conflux")]

                    // The message either doesn't count towards the sequence numbers

                    // or is already well-ordered, so we're ready to handle it.

                    // It's possible that some of our buffered messages are now ready to be

                    // handled. We don't check that here, however, because that's handled

                    // by the reactor main loop.

                }

                    // Tell the reactor to enqueue this msg

                }

            }

        } else {

            // If we don't have a conflux_handler, it means this circuit is not part of

            // a conflux tunnel, so we can just process the message.

        };

        )

    /// Handle a single incoming relay message that is known to be in order.

        #[cfg(feature = "conflux")]

        // Returns the original message if it's an incoming stream request

        // that we need to handle.

        // If it was an incoming stream request, we don't need to worry about

        // sending an XOFF as there's no stream data within this message.

            cfg_if::cfg_if! {

                if #[cfg(feature = "hs-service")] {

                } else {

                    return Err(internal!("incoming stream not rejected, but hs-service feature is disabled?!").into());

                }

            }

        // We may want to send an XOFF if the incoming buffer is too large.

    /// Handle a conflux message coming from the specified hop.

    ///

    /// Returns an error if

    ///

    ///   * this is not a conflux circuit (i.e. it doesn't have a [`ConfluxMsgHandler`])

    ///   * this is a client circuit and the conflux message originated an unexpected hop

    ///   * the cell was sent in violation of the handshake protocol

    #[cfg(feature = "conflux")]

            // If conflux is not enabled, tear down the circuit

            // (see 4.2.1. Cell Injection Side Channel Mitigations in prop329)

        };

    /// For conflux: return the sequence number of the last cell sent on this leg.

    ///

    /// Returns an error if this circuit is not part of a conflux set.

    #[cfg(feature = "conflux")]

    /// For conflux: set the sequence number of the last cell sent on this leg.

    ///

    /// Returns an error if this circuit is not part of a conflux set.

    #[cfg(feature = "conflux")]

    /// For conflux: return the sequence number of the last cell received on this leg.

    ///

    /// Returns an error if this circuit is not part of a conflux set.

    #[cfg(feature = "conflux")]

    /// A helper for handling incoming stream requests.

    ///

    // TODO: can we make this a method on CircHop to avoid the double HopNum lookup?

    #[cfg(feature = "hs-service")]

        use tor_cell::relaycell::msg::EndReason;

        use tor_error::into_internal;

        use tor_log_ratelim::log_ratelim;

        use crate::client::circuit::CIRCUIT_BUFFER_SIZE;

        use crate::stream::incoming::StreamReqInfo;

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

        };

        // The handler's hop_num is only ever set to None for relays.

        // TODO: we've already looked up the `hop` in handle_relay_cell, so we shouldn't

        // have to look it up again! However, we can't pass the `&mut hop` reference from

        // `handle_relay_cell` to this function, because that makes Rust angry (we'd be

        // borrowing self as mutable more than once).

        //

        // TODO: we _could_ use self.hops.get_mut(..) instead self.hop_mut(..) inside

        // handle_relay_cell to work around the problem described above

        {

            use crate::client::stream::IncomingStreamRequestDisposition::*;

            // IMPORTANT: super::syncview::CircHopSyncView::n_open_streams() (called via disposition() below)

            // accesses the stream map mutexes!

            //

            // This means it's very important not to call this function while any of the hop's

            // stream map mutex is held.

                }

            }

        }

        // TODO: Sadly, we need to look up `&mut hop` yet again,

        // since we needed to pass `&self.hops` by reference to our filter above. :(

            #[cfg(not(feature = "flowctl-cc"))]

            STREAM_READER_BUFFER,

        // A channel for the reactor to request a new drain rate from the reader.

        // Typically this notification will be sent after an XOFF is sent so that the reader can

        // send us a new drain rate when the stream data queue becomes empty.

                // The IncomingStreamRequestHandler's stream is full; it isn't

                // handling requests fast enough. So instead, we reply with an

                // END cell.

                );

                // The IncomingStreamRequestHandler's stream has been dropped.

                // In the Tor protocol as it stands, this always means that the

                // circuit itself is out-of-use and should be closed. (See notes

                // on `allow_stream_requests.`)

                //

                // Note that we will _not_ reach this point immediately after

                // the IncomingStreamRequestHandler is dropped; we won't hit it

                // until we next get an incoming request.  Thus, if we do later

                // want to add early detection for a dropped

                // IncomingStreamRequestHandler, we need to do it elsewhere, in

                // a different way.

                    circ_id = %self.unique_id,

                );

                // This will _cause_ the circuit to get closed.

            } 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.

            }

    /// Helper: process a destroy cell.

    #[allow(clippy::unnecessary_wraps)]

        // I think there is nothing more to do here.

    /// Handle a [`CtrlMsg::Create`](super::CtrlMsg::Create) message.

            CircuitHandshake::Ntor {

            } => {

            }

            }

        };

        // TODO: maybe we don't need to flush here?

        // (we could let run_once() handle all the flushing)

    /// 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.

        // 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.

            // done like this because holding the RNG across an await boundary makes the future

            // non-Send

        };

            circ_id = %self.unique_id,

        );

    /// 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.

        // In a CREATE_FAST handshake, we can't negotiate a format other than this.

    /// 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.

        // Exit now if we have an Ed25519 or RSA identity mismatch.

    /// 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.

        // Exit now if we have a mismatched key.

        // Set the client extensions.

    /// Add a hop to the end of this circuit.

    ///

    /// Will return an error if the circuit already has [`u8::MAX`] 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.

    /// 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.

    #[allow(clippy::cognitive_complexity)]

        // 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

                circ_id = %self.unique_id,

                reason

            );

            cfg_if::cfg_if! {

                if #[cfg(feature = "circ-padding")] {

                } else {

                    use crate::util::err::ExcessPadding;

                    return Err(Error::ExcessPadding(ExcessPadding::NoPaddingNegotiated, hopnum));

                }

            }

        #[cfg(feature = "conflux")]

            RelayCmd::CONFLUX_LINK

                | RelayCmd::CONFLUX_LINKED

                | RelayCmd::CONFLUX_LINKED_ACK

                | RelayCmd::CONFLUX_SWITCH

        ) {

                circ_id = %self.unique_id,

            );

        // 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)

            // The handler has a TargetHop so we do a quick convert for equality check.

                // Somebody was waiting for a message -- maybe this message