//! Code to handle incoming cells on a circuit.

use super::streammap::{ShouldSendEnd, StreamEnt};

use crate::circuit::celltypes::{ClientCircChanMsg, CreateResponse};

use crate::circuit::unique_id::UniqId;

use crate::circuit::{

    sendme, streammap, CircParameters, Create2Wrap, CreateFastWrap, CreateHandshakeWrap,

};

use crate::crypto::cell::{

    ClientLayer, CryptInit, HopNum, InboundClientCrypt, InboundClientLayer, OutboundClientCrypt,

    OutboundClientLayer, RelayCellBody, Tor1RelayCrypto,

};

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

use crate::{Error, Result};

use std::collections::VecDeque;

use std::convert::TryFrom;

use std::marker::PhantomData;

use std::pin::Pin;

use tor_cell::chancell::msg::{ChanMsg, Relay};

use tor_cell::relaycell::msg::{End, RelayMsg, Sendme};

use tor_cell::relaycell::{RelayCell, RelayCmd, StreamId};

use futures::channel::{mpsc, oneshot};

use futures::Sink;

use futures::Stream;

use tor_error::internal;

use std::sync::atomic::{AtomicU8, Ordering};

use std::sync::Arc;

use std::task::{Context, Poll};

use crate::channel::Channel;

#[cfg(test)]

use crate::circuit::sendme::CircTag;

use crate::circuit::sendme::StreamSendWindow;

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

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

use tor_cell::chancell;

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

use tor_linkspec::LinkSpec;

use tor_llcrypto::pk;

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

/// Initial value for outbound flow-control window on streams.

pub(super) const SEND_WINDOW_INIT: u16 = 500;

/// Initial value for inbound flow-control window on streams.

pub(super) const RECV_WINDOW_INIT: u16 = 500;

/// Size of the buffer used between the reactor and a `StreamReader`.

///

/// FIXME(eta): We pick 2× the receive window, which is very conservative (we arguably shouldn't

///             get sent more than the receive window anyway!). We might do due to things that

///             don't count towards the window though.

pub(super) const STREAM_READER_BUFFER: usize = (2 * RECV_WINDOW_INIT) as usize;

/// The type of a oneshot channel used to inform reactor users of the result of an operation.

pub(super) type ReactorResultChannel<T> = oneshot::Sender<Result<T>>;

/// A handshake type, to be used when creating circuit hops.

pub(super) enum CircuitHandshake {

    /// Use the CREATE_FAST handshake.

    CreateFast,

    /// Use the ntor handshake.

    Ntor {

        /// The public key of the relay.

        public_key: NtorPublicKey,

        /// The first hop's Ed25519 identity, which is verified against

        /// the identity held in the circuit's channel.

        ed_identity: pk::ed25519::Ed25519Identity,

    },

}

/// A message telling the reactor to do something.

pub(super) enum CtrlMsg {

    /// Create the first hop of this circuit.

    Create {

        /// A oneshot channel on which we'll receive the creation response.

        recv_created: oneshot::Receiver<CreateResponse>,

        /// The handshake type to use for the first hop.

        handshake: CircuitHandshake,

        /// Whether the hop supports authenticated SENDME cells.

        /// (And therefore, whether we should require them.)

        require_sendme_auth: RequireSendmeAuth,

        /// Other parameters relevant for circuit creation.

        params: CircParameters,

        /// Oneshot channel to notify on completion.

        done: ReactorResultChannel<()>,

    },

    /// Extend a circuit by one hop, using the ntor handshake.

    ExtendNtor {

        /// The handshake type to use for this hop.

        public_key: NtorPublicKey,

        /// Information about how to connect to the relay we're extending to.

        linkspecs: Vec<LinkSpec>,

        /// Whether the hop supports authenticated SENDME cells.

        /// (And therefore, whether we should require them.)

        require_sendme_auth: RequireSendmeAuth,

        /// Other parameters relevant for circuit extension.

        params: CircParameters,

        /// Oneshot channel to notify on completion.

        done: ReactorResultChannel<()>,

    },

    /// Begin a stream with the provided hop in this circuit.

    ///

    /// Allocates a stream ID, and sends the provided message to that hop.

    BeginStream {

        /// The hop number to begin the stream with.

        hop_num: HopNum,

        /// The message to send.

        message: RelayMsg,

        /// A channel to send messages on this stream down.

        ///

        /// This sender shouldn't ever block, because we use congestion control and only send

        /// SENDME cells once we've read enough out of the other end. If it *does* block, we

        /// can assume someone is trying to send us more cells than they should, and abort

        /// the stream.

        sender: mpsc::Sender<RelayMsg>,

        /// A channel to receive messages to send on this stream from.

        rx: mpsc::Receiver<RelayMsg>,

        /// Oneshot channel to notify on completion, with the allocated stream ID.

        done: ReactorResultChannel<StreamId>,

    },

    /// Send a SENDME cell (used to ask for more data to be sent) on the given stream.

    SendSendme {

        /// The stream ID to send a SENDME for.

        stream_id: StreamId,

        /// The hop number the stream is on.

        hop_num: HopNum,

    },

    /// Shut down the reactor.

    Shutdown,

    /// (tests only) Add a hop to the list of hops on this circuit, with dummy cryptography.

    #[cfg(test)]

    AddFakeHop {

        supports_flowctrl_1: bool,

        fwd_lasthop: bool,

        rev_lasthop: bool,

        params: CircParameters,

        done: ReactorResultChannel<()>,

    },

    /// (tests only) Get the send window and expected tags for a given hop.

    #[cfg(test)]

    QuerySendWindow {

        hop: HopNum,

        done: ReactorResultChannel<(u16, Vec<CircTag>)>,

    },

    /// (tests only) Send a raw relay cell with send_relay_cell().

    #[cfg(test)]

    SendRelayCell {

        hop: HopNum,

        early: bool,

        cell: RelayCell,

    },

}

/// Represents the reactor's view of a single hop.

pub(super) struct CircHop {

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

    map: streammap::StreamMap,

    /// Window used to say how many cells we can receive.

    recvwindow: sendme::CircRecvWindow,

    /// If true, this hop is using an older link protocol and we

    /// shouldn't expect good authenticated SENDMEs from it.

    auth_sendme_required: RequireSendmeAuth,

    /// Window used to say how many cells we can send.

    sendwindow: sendme::CircSendWindow,

    /// Buffer for messages we can't send to this hop yet due to congestion control.

    ///

    /// Contains the cell to send, and a boolean equivalent to the `early` parameter

    /// in `Reactor::send_relay_cell` (as in, whether to send the cell using `RELAY_EARLY`).

    ///

    /// This shouldn't grow unboundedly: we try and pop things off it first before

    /// doing things that would result in it growing (and stop before growing it

    /// if popping things off it can't be done).

    ///

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

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

    outbound: VecDeque<(bool, RelayCell)>,

}

/// Enumeration to determine whether we require circuit-level SENDME cells to be

/// authenticated.

///

/// (This is an enumeration rather than a boolean to prevent accidental sense

/// inversion.)

pub(super) enum RequireSendmeAuth {

    /// Sendme authentication is expected from this hop, and therefore is

    /// required.

    Yes,

    /// Sendme authentication is not expected from this hop, and therefore not

    /// required.

    No,

}

impl RequireSendmeAuth {

    /// Create an appropriate [`RequireSendmeAuth`] for a given set of relay

    /// subprotocol versions.

    //

    // TODO(nickm): At some point in the future, once there are no 0.3.5 relays

    // on the Tor network, we can safely require authenticated SENDMEs from all

    // relays.

    //

    // At that point, if we have a relay implementation in Rust, it should look

    // at the network parameter `SendmeAcceptMinVersion` when deciding whether

    // to require authenticated SENDMEs.

            // The relay supports FlowCtrl=1, and therefore will authenticate.

        } else {

        }

}

/// An indicator on what we should do when we receive a cell for a circuit.

enum CellStatus {

    /// The circuit should stay open.

    Continue,

    /// Perform a clean shutdown on this circuit.

    CleanShutdown,

}

impl CircHop {

    /// Create a new hop.

}

/// An object that's waiting for a meta cell (one not associated with a stream) in order to make

/// progress.

///

/// # Background

///

/// The `Reactor` can't have async functions that send and receive cells, because its job is to

/// send and receive cells: if one of its functions tried to do that, it would just hang forever.

///

/// To get around this problem, the reactor can send some cells, and then make one of these

/// `MetaCellHandler` objects, which will be run when the reply arrives.

pub(super) trait MetaCellHandler: Send {

    /// The hop we're expecting the message to come from. This is compared against the hop

    /// from which we actually receive messages, and an error is thrown if the two don't match.

    fn expected_hop(&self) -> HopNum;

    /// Called when the message we were waiting for arrives.

    ///

    /// Gets a copy of the `Reactor` in order to do anything it likes there.

    fn finish(&mut self, msg: RelayMsg, reactor: &mut Reactor) -> Result<()>;

}

/// An object that can extend a circuit by one hop, using the `MetaCellHandler` trait.

///

/// Yes, I know having trait bounds on structs is bad, but in this case it's necessary

/// since we want to be able to use `H::KeyType`.

struct CircuitExtender<H, L, FWD, REV>

where

    H: ClientHandshake,

{

    /// Handshake state.

    state: Option<H::StateType>,

    /// Whether the hop supports authenticated SENDME cells.

    /// (And therefore, whether we require them.)

    require_sendme_auth: RequireSendmeAuth,

    /// Parameters used for this extension.

    params: CircParameters,

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

    unique_id: UniqId,

    /// The hop we're expecting the EXTENDED2 cell to come back from.

    expected_hop: HopNum,

    /// `PhantomData` used to make the other type parameters required for a circuit extension

    /// part of the `struct`, instead of having them be provided during a function call.

    ///

    /// This is done this way so we can implement `MetaCellHandler` for this type, which

    /// doesn't include any generic type parameters; we need them to be part of the type

    /// so we know what they are for that `impl` block.

    phantom: PhantomData<(L, FWD, REV)>,

}

impl<H, L, FWD, REV> CircuitExtender<H, L, FWD, REV>

where

    H: ClientHandshake,

    H::KeyGen: KeyGenerator,

    L: CryptInit + ClientLayer<FWD, REV>,

    FWD: OutboundClientLayer + 'static + Send,

    REV: InboundClientLayer + 'static + Send,

{

    /// Start extending a circuit, sending the necessary EXTEND cell and returning a

    /// new `CircuitExtender` to be called when the reply arrives.

    ///

    /// The `handshake_id` is the numeric identifier for what kind of

    /// handshake we're doing.  The `key` is the relay's onion key that

    /// goes along with the handshake, and the `linkspecs` are the

    /// link specifiers to include in the EXTEND cell to tell the

    /// current last hop which relay to connect to.

        use tor_cell::relaycell::msg::{Body, Extend2};

        // Perform the first part of the cryptographic handshake

        // ... and now we wait for a response.

}

impl<H, L, FWD, REV> MetaCellHandler for CircuitExtender<H, L, FWD, REV>

where

    H: ClientHandshake,

    H::StateType: Send,

    H::KeyGen: KeyGenerator,

    L: CryptInit + ClientLayer<FWD, REV> + Send,

    FWD: OutboundClientLayer + 'static + Send,

    REV: InboundClientLayer + 'static + Send,

{

        // ???? Do we need to shutdown the circuit for the remaining error

        // ???? cases in this function?

            _ => {

            }

        };

        // Now perform the second part of the handshake, and see if it

        // succeeded.

        // If we get here, it succeeded.  Add a new hop to the circuit.

}

/// Object to handle incoming cells and background tasks on a circuit

///

/// This type is returned when you finish a circuit; you need to spawn a

/// new task that calls `run()` on it.

#[must_use = "If you don't call run() on a reactor, the circuit won't work."]

pub struct Reactor {

    /// Receiver for control messages for this reactor, sent by `ClientCirc` objects.

    pub(super) control: mpsc::UnboundedReceiver<CtrlMsg>,

    /// Buffer for cells we can't send out the channel yet due to it being full.

    ///

    /// We try and dequeue off this first before doing anything else, ensuring that

    /// it cannot grow unboundedly (and if we start having to enqueue things on here after

    /// the channel shows backpressure, we stop pulling from receivers that could send here).

    ///

    /// 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) outbound: VecDeque<ChanCell>,

    /// The channel this circuit is using to send cells through.

    pub(super) channel: Channel,

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

    /// channel.

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

    pub(super) input: mpsc::Receiver<ClientCircChanMsg>,

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

    pub(super) crypto_in: InboundClientCrypt,

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

    pub(super) crypto_out: OutboundClientCrypt,

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

    pub(super) hops: Vec<CircHop>,

    /// Shared atomic for the number of hops this circuit has.

    pub(super) num_hops: Arc<AtomicU8>,

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

    pub(super) unique_id: UniqId,

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

    pub(super) channel_id: CircId,

    /// A handler for a meta cell, together with a result channel to notify on completion.

    pub(super) meta_handler: Option<(Box<dyn MetaCellHandler>, ReactorResultChannel<()>)>,

}

impl Reactor {

    /// Launch the reactor, and run until the circuit closes or we

    /// encounter an error.

    ///

    /// Once this method returns, the circuit is dead and cannot be

    /// used again.

            }

        };

    /// Helper for run: doesn't mark the circuit closed on finish.  Only

    /// processes one cell or control message.

            // Check whether we've got a control message pending.

                    None => {

                    }

                    Some(CtrlMsg::Shutdown) => {

                        trace!(

                    }

                    // This message requires actually blocking, so we can't handle it inside

                    // this nonblocking poll_fn.

                    }

                }

            // Check whether we've got an input message pending.

                    None => {

                    }

                    }

                }

            // Now for the tricky part. We want to grab some relay cells from all of our streams

            // and forward them on to the channel, but we need to pay attention to both whether

            // the channel can accept cells right now, and whether congestion control allows us

            // to send them.

            //

            // We also have to do somewhat cursed things and call start_send inside this poll_fn,

            // since we need to check whether the channel can still receive cells after each one

            // that we send.

                // (using this as a named block for early returns; not actually a loop)

                #[allow(clippy::never_loop)]

                'outer: loop {

                    // First, drain our queue of things we tried to send earlier, but couldn't.

                        // `futures::Sink::start_send` dictates we need to call `poll_ready` before

                        // each `start_send` call.

                    }

                    // Let's look at our hops, and streams for each hop.

                            'hop: while let Some((early, cell)) = self.hops[i].outbound.pop_front()

                            {

                                trace!(

                            }

                        // Look at all of the streams on this hop.

                            if let StreamEnt::Open {

                            {

                                // Do the stream and hop send windows allow us to obtain and

                                // send something?

                                //

                                // FIXME(eta): not everything counts toward congestion control!

                                    }

                        }

                    }

                }

            // Close the streams we said we'd close.

            }

            // Send messages we said we'd send.

            }

            } else {

            }

        if let Some(CtrlMsg::Create {

        {

                CircuitHandshake::CreateFast => {

                }

                CircuitHandshake::Ntor {

                }

            };

    /// 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, a handshake object to perform

    /// the cryptographic cryptographic handshake, and a layer type to

    /// handle relay crypto after this hop is built.

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

        };

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

    /// Use the ntor handshake to connect to the first hop of this circuit.

    ///

    /// Note that the provided 'target' must match the channel's target,

    /// or the handshake will fail.

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

    /// Handle a RELAY cell on this circuit with stream ID 0.

    fn handle_meta_cell(&mut self, hopnum: HopNum, msg: RelayMsg) -> Result<CellStatus> {

        // SENDME cells and TRUNCATED get handled internally by the circuit.

        // For all other command types, we'll only get them in response

        // to another command, which should have registered a responder.

        //

        // TODO: that means that service-introduction circuits will need

        // a different implementation, but that should be okay. We'll work

        // something out.

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

            } else {

                // Somebody wanted a message from a different hop!  Put this

                // one back.

            }

        } else {

            // No need to call shutdown here, since this error will

            // propagate to the reactor shut it down.

        }

    /// Handle a RELAY_SENDME cell on this circuit with stream ID 0.

        // No need to call "shutdown" on errors in this function;

        // it's called from the reactor task and errors will propagate there.

                } else {

                }

            }

            None => {

                } else {

                }

            }

        };

    /// Send a message onto the circuit's channel (to be called with a `Context`)

    ///

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

    ///

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

        } else {

            // This has been observed to happen in code that doesn't have bugs in it, simply due

            // to the way `Channel`'s `poll_ready` implementation works (it can change due to

            // the actions of another thread in between callers of this function checking it,

            // and this function checking it).

            //

            // However, if it's happening a lot more than it used to, that probably indicates

            // some caller that's not checking whether the channel is full before calling

            // this function.

        }

    /// Wrapper around `send_msg_direct` that uses `futures::future::poll_fn` to get a `Context`.

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

    ///

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

                // Send window is empty! Push this cell onto the hop's outbound queue, and it'll

                // get sent later.

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

        } else {

        };

        // If the cell counted towards our sendme window, decrement

        // that window, and maybe remember the authentication tag.

                // We need to decrement the stream-level sendme window.

                // Stream data cells should only be dequeued and fed into this function if

                // the window is above zero, so we don't need to worry about enqueuing things.

                } else {

                }

    /// Try to install a given meta-cell handler to receive any unusual cells on

    /// this circuit, along with a result channel to notify on completion.

        } else {

        }

    /// Handle a CtrlMsg other than Shutdown.

            // This is handled earlier, since it requires blocking.

            // This is handled earlier, since it requires generating a ReactorError.

            CtrlMsg::ExtendNtor {

                    }

                };

            }

            CtrlMsg::BeginStream {

            }

            #[cfg(test)]

            CtrlMsg::AddFakeHop {

                } else {

                };

            }

            #[cfg(test)]

                } else {

                });

            }

            #[cfg(test)]

            }

        }

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

    /// `message` to the provided hop.

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

        // Mark the stream as closing.

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

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

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

    ///

    /// Return true if we should exit.

        use ClientCircChanMsg::*;

            }

        }

    /// React to a Relay or RelayEarly cell.

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

        // corresponding hop.

        // Make a copy of the authentication tag. TODO: I'd rather not

        // copy it, but I don't see a way around it right now.

        };

        // Decode the cell.

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

                .recvwindow

        // Break the message apart into its streamID and message.

            Some(StreamEnt::Open {

                    // We need to handle sendmes here, not in the stream's

                    // recv() method, or else we'd never notice them if the

                    // stream isn't reading.

                // Remember whether this was an end cell: if so we should

                // close the stream.

                // TODO: Add a wrapper type here to reject cells that should

                // never go to a client, like BEGIN.

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