//! Module exposing structures relating to a reactor's view of a circuit hop.

// TODO(relay): don't import from the client module

use crate::client::circuit::handshake::RelayCryptLayerProtocol;

use crate::ccparams::CongestionControlParams;

use crate::circuit::CircParameters;

use crate::congestion::{CongestionControl, sendme};

use crate::memquota::{SpecificAccount, StreamAccount};

use crate::stream::CloseStreamBehavior;

use crate::stream::SEND_WINDOW_INIT;

use crate::stream::StreamMpscSender;

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

use crate::stream::flow_ctrl::params::FlowCtrlParameters;

use crate::stream::flow_ctrl::state::{

    FlowCtrlHooks, StreamFlowCtrl, StreamRateLimit, WithSidechannelMitigations,

};

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

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

use crate::streammap::{

    self, EndSentStreamEnt, OpenStreamEnt, ShouldSendEnd, StreamEntMut, StreamMap,

};

use crate::util::notify::{NotifyReceiver, NotifySender};

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

use derive_deftly::Deftly;

use postage::watch;

use safelog::sensitive as sv;

use tracing::{debug, trace};

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

use tor_cell::relaycell::extend::{CcRequest, CircRequestExt};

use tor_cell::relaycell::flow_ctrl::{Xoff, Xon, XonKBpsEwma};

use tor_cell::relaycell::msg::AnyRelayMsg;

use tor_cell::relaycell::{

    AnyRelayMsgOuter, RelayCellDecoder, RelayCellDecoderResult, RelayCellFormat, RelayCmd,

    StreamId, UnparsedRelayMsg,

};

use tor_error::{Bug, ErrorKind, HasKind, internal, into_internal};

use tor_memquota::derive_deftly_template_HasMemoryCost;

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

use tor_protover::named;

use tor_rtcompat::DynTimeProvider;

use std::num::NonZeroU32;

use std::pin::Pin;

use std::result::Result as StdResult;

use std::sync::{Arc, Mutex};

use web_time_compat::Instant;

#[cfg(test)]

use tor_cell::relaycell::msg::SendmeTag;

#[cfg(feature = "relay")]

use {

    crate::ccparams::{Algorithm, AlgorithmDiscriminants},

    crate::circuit::HandshakeSubprotocols,

    crate::relay::{CircNetParameters, CongestionControlNetParams},

};

use cfg_if::cfg_if;

/// The size of the stream's outbound RELAY message queue.

// TODO(tuning): figure out if this is a good size for this buffer

const CIRCUIT_BUFFER_SIZE: usize = 128;

/// Type of negotiation that we'll be performing as we establish a hop.

///

/// Determines what flavor of extensions we can send and receive, which in turn

/// limits the hop settings we can negotiate.

///

// TODO-CGO: This is likely to be refactored when we finally add support for

// HsV3+CGO, which will require refactoring

#[derive(Debug, Clone, Copy, Eq, PartialEq)]

pub(crate) enum HopNegotiationType {

    /// We're using a handshake in which extension-based negotiation cannot occur.

    None,

    /// We're using the HsV3-ntor handshake, in which the client can send extensions,

    /// but the server cannot.

    ///

    /// As a special case, the default relay encryption protocol is the hsv3

    /// variant of Tor1.

    //

    // We would call this "HalfDuplex" or something, but we do not expect to add

    // any more handshakes of this type.

    HsV3,

    /// We're using a handshake in which both client and relay can send extensions.

    Full,

}

/// The settings we use for single hop of a circuit.

///

/// Unlike [`CircParameters`], this type is crate-internal.

/// We construct it based on our settings from the circuit,

/// and from the hop's actual capabilities.

/// Then, we negotiate with the hop as part of circuit

/// creation/extension to determine the actual settings that will be in use.

/// Finally, we use those settings to construct the negotiated circuit hop.

//

// TODO: Relays should probably derive an instance of this type too, as

// part of the circuit creation handshake.

#[derive(Clone, Debug)]

pub(crate) struct HopSettings {

    /// The negotiated congestion control settings for this hop .

    pub(crate) ccontrol: CongestionControlParams,

    /// Flow control parameters that will be used for streams on this hop.

    pub(crate) flow_ctrl_params: FlowCtrlParameters,

    /// Maximum number of permitted incoming relay cells for this hop.

    pub(crate) n_incoming_cells_permitted: Option<u32>,

    /// Maximum number of permitted outgoing relay cells for this hop.

    pub(crate) n_outgoing_cells_permitted: Option<u32>,

    /// The relay cell encryption algorithm and cell format for this hop.

    relay_crypt_protocol: RelayCryptLayerProtocol,

}

impl HopSettings {

    /// Construct a new `HopSettings` based on `params` (a set of circuit parameters)

    /// and `caps` (a set of protocol capabilities for a circuit target).

    ///

    /// The resulting settings will represent what the client would prefer to negotiate

    /// (determined by `params`),

    /// as modified by what the target relay is believed to support (represented by `caps`).

    ///

    /// This represents the `HopSettings` in a pre-negotiation state:

    /// the circuit negotiation process will modify it.

    #[allow(clippy::unnecessary_wraps)] // likely to become fallible in the future.

            crate::ccparams::Algorithm::Vegas(_) => {

                // If the target doesn't support FLOWCTRL_CC, we can't use Vegas.

            }

        };

        // Negotiate CGO if it is supported, if CC is also supported,

        // and if CGO is available on this relay.

            HopNegotiationType::HsV3 => {

                // TODO-CGO: Support CGO when available.

                cfg_if! {

                    if #[cfg(all(feature = "hs-common", feature = "flowctl-cc", feature = "counter-galois-onion"))] {

                        } else {

                        }

                    } else if #[cfg(feature = "hs-common")] {

                            RelayCryptLayerProtocol::HsV3(RelayCellFormat::V0)

                    } else {

                        return Err(

                            tor_error::internal!("Unexpectedly tried to negotiate HsV3 without support!").into(),

                        );

                    }

                }

            }

            HopNegotiationType::Full => {

                cfg_if! {

                    if #[cfg(all(feature = "flowctl-cc", feature = "counter-galois-onion"))] {

                        #[allow(clippy::overly_complex_bool_expr)]

                        {

                        } else {

                        }

                    } else {

                        RelayCryptLayerProtocol::Tor1(RelayCellFormat::V0)

                    }

                }

            }

        };

    /// Build a [`HopSettings`] from the parameters requested during a circuit handshake.

    //

    // We disable `unused` warnings at the root of tor-proto,

    // but it's nice to have here so we re-enable it.

    #[warn(unused)]

    #[cfg(feature = "relay")]

        // Unpack everything to make sure that we aren't missing anything

        // (otherwise clippy would warn).

        let CircNetParameters {

            cc:

                CongestionControlNetParams {

                },

        // TODO: We have similar logic in and around `HopSettings` that deals with determining the

        // crypt protocol and cc algorithm to use. We might want to try to dedup some of this, or

        // make it more self-contained. This is a bit tricky though since the code is used in

        // different situations and the inputs are not the same.

            (AlgorithmDiscriminants::FixedWindow, true) => {

            }

            (AlgorithmDiscriminants::Vegas, true) => {

            }

        };

        // TODO(arti#2442): The builder pattern here seems like a footgun.

    /// Return the negotiated relay crypto protocol.

    /// Return the client circuit-creation extensions that we should use in order to negotiate

    /// these circuit hop parameters.

    #[allow(clippy::unnecessary_wraps)]

        // allow 'unused_mut' because of the combinations of `cfg` conditions below

        #[allow(unused_mut)]

        #[allow(unused, unused_mut)]

        // See whether we need to send a list of required protocol capabilities.

        // These aren't "negotiated" per se; they're simply demanded.

        // The relay will refuse the circuit if it doesn't support all of them,

        // and if any of them isn't supported in the SubprotocolRequest extension.

        //

        // (In other words, don't add capabilities here just because you want the

        // relay to have them! They must be explicitly listed as supported for use

        // with this extension. For the current list, see

        // https://spec.torproject.org/tor-spec/create-created-cells.html#subproto-request)

        //

        // TODO: Should this use `HandshakeSubprotocols` so that the above comment has some

        // compile-time checks?

        #[allow(unused_mut)]

        #[cfg(feature = "counter-galois-onion")]

}

#[cfg(test)]

impl std::default::Default for CircParameters {

}

/// An error that can occur when building a [`HopSettings`] using parameters requested during a

/// circuit handshake.

#[derive(Clone, Debug, thiserror::Error)]

pub(crate) enum HandshakeParamsError {

    /// The provided parameters are incompatible with each other.

    #[error("The provided handshake parameters are incompatible with each other: {0}")]

    IncompatibleParams(&'static str),

    /// An internal error.

    #[error("Internal error")]

    Internal(#[from] tor_error::Bug),

}

impl HasKind for HandshakeParamsError {

        }

}

impl CircParameters {

    /// Constructor

}

/// Instructions for sending a RELAY cell.

///

/// This instructs a circuit reactor to send a RELAY cell to a given target

/// (a hop, if we are a client, or the client, if we are a relay).

#[derive(educe::Educe)]

#[educe(Debug)]

pub(crate) struct SendRelayCell {

    /// The hop number, or `None` if we are a relay.

    pub(crate) hop: Option<HopNum>,

    /// Whether to use a RELAY_EARLY cell.

    pub(crate) early: bool,

    /// The cell to send.

    pub(crate) cell: AnyRelayMsgOuter,

}

/// The inbound state of a hop.

pub(crate) struct CircHopInbound {

    /// Decodes relay cells received from this hop.

    decoder: RelayCellDecoder,

    /// Remaining permitted incoming relay cells from this hop, plus 1.

    ///

    /// (In other words, `None` represents no limit,

    /// `Some(1)` represents an exhausted limit,

    /// and `Some(n)` means that n-1 more cells may be received.)

    ///

    /// If this ever decrements from Some(1), then the circuit must be torn down with an error.

    n_incoming_cells_permitted: Option<NonZeroU32>,

}

/// The outbound state of a hop.

pub(crate) struct CircHopOutbound {

    /// Congestion control object.

    ///

    /// This object is also in charge of handling circuit level SENDME logic for this hop.

    ccontrol: Arc<Mutex<CongestionControl>>,

    /// Map from stream IDs to streams.

    ///

    /// We store this with the reactor instead of the circuit, since the

    /// reactor needs it for every incoming cell on a stream, whereas

    /// the circuit only needs it when allocating new streams.

    ///

    /// NOTE: this is behind a mutex because the client reactor polls the `StreamMap`s

    /// of all hops concurrently, in a `FuturesUnordered`. Without the mutex,

    /// this wouldn't be possible, because it would mean holding multiple

    /// mutable references to `self` (the reactor). Note, however,

    /// that there should never be any contention on this mutex:

    /// we never create more than one

    /// `CircHopList::ready_streams_iterator()` stream

    /// at a time, and we never clone/lock the hop's `StreamMap` outside of it.

    ///

    /// Additionally, the stream map of the last hop (join point) of a conflux tunnel

    /// is shared with all the circuits in the tunnel.

    map: Arc<Mutex<StreamMap>>,

    /// Format to use for relay cells.

    //

    // When we have packed/fragmented cells, this may be replaced by a RelayCellEncoder.

    relay_format: RelayCellFormat,

    /// Flow control parameters for new streams.

    flow_ctrl_params: Arc<FlowCtrlParameters>,

    /// Remaining permitted outgoing relay cells from this hop, plus 1.

    ///

    /// If this ever decrements from Some(1), then the circuit must be torn down with an error.

    n_outgoing_cells_permitted: Option<NonZeroU32>,

}

impl CircHopInbound {

    /// Create a new [`CircHopInbound`].

    /// Parse a RELAY or RELAY_EARLY cell body.

    ///

    /// Requires that the cryptographic checks on the message have already been

    /// performed

    /// Decrement the limit of inbound cells that may be received from this hop; give

    /// an error if it would reach zero.

}

impl CircHopOutbound {

    /// Create a new [`CircHopOutbound`].

    /// Start a stream. Creates an entry in the stream map with the given channels, and sends the

    /// `message` to the provided hop.

        // TODO: This has a lot of duplicated code with `Self::add_ent_with_id()`.

        // A channel for the reactor to inform the writer of a new rate limit.

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

            // We are starting the stream,

            // so we're a client and want flow control sidechannel mitigations.

        // A queue for inbound RELAY messages.

        // A queue for outbound RELAY messages.

    /// Close the stream associated with `id` because the stream was dropped.

    ///

    /// If we have not already received an END cell on this stream, send one.

    /// If no END cell is specified, an END cell with the reason byte set to

    /// REASON_MISC will be sent.

    ///

    // Note(relay): `circ_uniq_id` is an opaque displayable type

    // because relays use a different circuit ID type

    // than clients. Eventually, we should probably make

    // them both use the same ID type, or have a nicer approach here

    #[allow(clippy::too_many_arguments)]

            circ_uniq_id = %circ_uniq_id,

            circ_id = %circ_id,

            stream_id = %id,

            should_send_end = ?should_send_end,

            "Ending stream",

        );

        // TODO: I am about 80% sure that we only send an END cell if

        // we didn't already get an END cell.  But I should double-check!

        {

    /// Check if we should send an XON message.

    ///

    /// If we should, then returns the XON message that should be sent.

        // the call below will return an error if XON/XOFF aren't supported,

        // so we check for support here

        {

            // stream went away

        };

    /// Check if we should send an XOFF message.

    ///

    /// If we should, then returns the XOFF message that should be sent.

        // the call below will return an error if XON/XOFF aren't supported,

        // so we check for support here

        {

            // stream went away

        };

    /// Return the format that is used for relay cells sent to this hop.

    ///

    /// For the most part, this format isn't necessary to interact with a CircHop;

    /// it becomes relevant when we are deciding _what_ we can encode for the hop.

    /// Delegate to CongestionControl, for testing purposes

    #[cfg(test)]

    /// Return the number of open streams on this hop.

    ///

    /// WARNING: because this locks the stream map mutex,

    /// it should never be called from a context where that mutex is already locked.

    /// Return a reference to our CongestionControl object.

    /// We're about to send `msg`.

    ///

    /// See [`OpenStreamEnt::about_to_send`](crate::streammap::OpenStreamEnt::about_to_send).

    //

    // TODO prop340: This should take a cell or similar, not a message.

    //

    // Note(relay): `circ_uniq_id` is an opaque displayable type

    // because relays use a different circuit ID type

    // than clients. Eventually, we should probably make

    // them both use the same ID type, or have a nicer approach here

            // This can happen when we have outgoing data queued when we received an END.

            // We shouldn't return an error here since it would close the circuit along with all

            // other streams, and instead we just let the caller send this message anyways.

            // Also the caller only calls `about_to_send()` for DATA cells,

            // which means that other non-DATA cells don't hit this code path and are always sent,

            // and so we should handle all cell types consistently.

            // TODO: We should drop the message and not send it,

            // but the caller of `about_to_send()` isn't designed to handle fallible sends

            // so it would need some refactoring to handle this.

                circ_uniq_id = %circ_uniq_id,

                circ_id = %circ_id,

                stream_id = %stream_id,

                "sending a relay cell for non-existent or non-open stream!",

            );

        };

    /// Add an entry to this map using the specified StreamId.

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

        // TODO: This has a lot of duplicated code with `Self::begin_stream()`.

        // A channel for the reactor to inform the writer of a new rate limit.

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

        // A queue for inbound RELAY messages.

        // A queue for outbound RELAY messages.

    /// Builds the reactor's flow control handler for a new stream.

    // TODO: remove the `Result` once we remove the "flowctl-cc" feature

    #[cfg_attr(feature = "flowctl-cc", expect(clippy::unnecessary_wraps))]

        {

        } else {

            cfg_if::cfg_if! {

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

                } else {

                    drop(params);

                    drop(rate_limit_updater);

                    drop(drain_rate_requester);

                    Err(internal!(

                        "`CongestionControl` doesn't use sendmes, but 'flowctl-cc' feature not enabled",

                    ).into())

                }

            }

        }

    /// Deliver `msg` to the specified open stream entry `ent`.

        use tor_async_utils::SinkTrySend as _;

        use tor_async_utils::SinkTrySendError as _;

        // The stream for this message exists, and is open.

        // We need to handle SENDME/XON/XOFF messages here, not in the stream's recv() method, or

        // else we'd never notice them if the stream isn't reading.

            RelayCmd::SENDME => {

            }

            RelayCmd::XON => {

            }

            RelayCmd::XOFF => {

            }

        }

                cfg_if::cfg_if! {

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

                        // If we get here, we either have a logic bug (!), or an attacker

                        // is sending us more cells than we asked for via congestion control.

                        return Err(Error::CircProto(format!(

                            "Stream sink would block; received too many cells on stream ID {}",

                            sv(streamid),

                        )));

                    } else {

                    }

                }

    /// Note that we received an END message (or other message indicating the end of

    /// the stream) on the stream with `id`.

    ///

    /// See [`StreamMap::ending_msg_received`](crate::streammap::StreamMap::ending_msg_received).

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

    /// Handle `msg`, delivering it to the stream with the specified `streamid` if appropriate.

    ///

    /// Returns back the provided `msg`, if the message is an incoming stream request

    /// that needs to be handled by the calling code.

    ///

    // TODO: the above is a bit of a code smell -- we should try to avoid passing the msg

    // back and forth like this.

    {

                // Can't have a stream level SENDME when congestion control is enabled.

            }

            }

            Some(StreamEntMut::EndSent(_))

                    RelayCmd::BEGIN | RelayCmd::BEGIN_DIR | RelayCmd::RESOLVE

                ) =>

            {

                // If the other side is sending us a BEGIN but hasn't yet acknowledged our END

                // message, just remove the old stream from the map and stop waiting for a

                // response

            }

                // We sent an end but maybe the other side hasn't heard.

                    StreamStatus::Closed => {

                    }

                }

            }

                RelayCmd::BEGIN | RelayCmd::BEGIN_DIR | RelayCmd::RESOLVE

            ) =>

            {

            }

            _ => {

                // No stream wants this message, or ever did.

            }

        }

    /// Get the stream map of this hop.

    /// Set the stream map of this hop to `map`.

    ///

    /// Returns an error if the existing stream map of the hop has any open stream.

    /// Decrement the limit of outbound cells that may be sent to this hop; give

    /// an error if it would reach zero.

}

/// If `val` is `Some(1)`, return Err(());

/// otherwise decrement it (if it is Some) and return Ok(()).

#[inline]

    // This is a bit verbose, but I've confirmed that it optimizes nicely.

            } else {

            }

        }

    }

/// Convert a limit from the form used in a HopSettings to that used here.

/// (The format we use here is more compact.)

    // See "known limitations" comment on n_incoming_cells_permitted.

/// A collection of components that can be used to interact with the reactor's view of a Tor stream.

//

// TODO: We also have a `StreamComponents` type that is used and built outside of the reactor.

// It's maybe confusing to have these similar type names, so a better name would be nice.

//

// TODO(arti#2068): The components we return should maybe depend on what type of flow control is

// used, so in the future we might want to make some of these fields optional.

#[derive(Debug, Deftly)]

#[derive_deftly(HasMemoryCost)]

pub(crate) struct ReactorStreamComponents {

    /// An MPSC receiver for inbound messages that arrive on the stream.

    #[deftly(has_memory_cost(indirect_size = "0"))] // estimate

    pub(crate) stream_inbound_rx: StreamQueueReceiver,

    /// An MPSC sender for outbound messages to be sent on the stream.

    #[deftly(has_memory_cost(indirect_size = "size_of::<AnyRelayMsg>()"))] // estimate

    pub(crate) stream_outbound_tx: StreamMpscSender<AnyRelayMsg>,

    /// A mechanism to allow the stream's writer to receive rate limit updates from the reactor.

    // The `watch::Sender` owns the indirect data.

    #[deftly(has_memory_cost(indirect_size = "0"))]

    pub(crate) rate_limit_rx: watch::Receiver<StreamRateLimit>,

    /// A mechanism to allow the stream's reader to receive drain rate update requests from the

    /// reactor.

    #[deftly(has_memory_cost(indirect_size = "0"))]

    pub(crate) drain_rate_request_rx: NotifyReceiver<DrainRateRequest>,

}