//! Client-specific types and implementation.

pub mod channel;

pub mod circuit;

pub mod stream;

#[cfg(feature = "rpc")]

pub mod rpc;

#[cfg(feature = "send-control-msg")]

pub(crate) mod msghandler;

pub(crate) mod reactor;

use derive_deftly::Deftly;

use oneshot_fused_workaround as oneshot;

use std::net::IpAddr;

use std::sync::Arc;

use tracing::instrument;

use crate::circuit::UniqId;

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

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

    CircuitPadder, CircuitPadderConfig, CircuitPadderConfigError,

};

use crate::client::stream::{

    DataStream, OutboundDataCmdChecker, ResolveCmdChecker, ResolveStream, StreamParameters,

    StreamReceiver,

};

use crate::congestion::sendme::StreamRecvWindow;

use crate::crypto::cell::HopNum;

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

use crate::stream::STREAM_READER_BUFFER;

use crate::stream::cmdcheck::AnyCmdChecker;

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

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

use crate::stream::queue::stream_queue;

use crate::stream::{RECV_WINDOW_INIT, StreamComponents, StreamTarget, Tunnel};

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

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

use circuit::{CIRCUIT_BUFFER_SIZE, ClientCirc, Path};

use reactor::{CtrlCmd, CtrlMsg, FlowCtrlMsg, MetaCellHandler};

use postage::watch;

use tor_cell::relaycell::StreamId;

use tor_cell::relaycell::flow_ctrl::XonKbpsEwma;

use tor_cell::relaycell::msg::{AnyRelayMsg, Begin, Resolve, Resolved, ResolvedVal};

use tor_error::bad_api_usage;

use tor_linkspec::OwnedChanTarget;

use tor_memquota::derive_deftly_template_HasMemoryCost;

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

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

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

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

use crate::client::stream::{IncomingCmdChecker, IncomingStream};

#[cfg(feature = "send-control-msg")]

use msghandler::{MsgHandler, UserMsgHandler};

/// Handle to use during an ongoing protocol exchange with a circuit's last hop

///

/// This is obtained from [`ClientTunnel::start_conversation`],

/// and used to send messages to the last hop relay.

//

// TODO(conflux): this should use ClientTunnel, and it should be moved into

// the tunnel module.

#[cfg(feature = "send-control-msg")]

pub struct Conversation<'r>(&'r ClientTunnel);

#[cfg(feature = "send-control-msg")]

impl Conversation<'_> {

    /// Send a protocol message as part of an ad-hoc exchange

    ///

    /// Responses are handled by the `UserMsgHandler` set up

    /// when the `Conversation` was created.

    /// Send a `SendMsgAndInstallHandler` to the reactor and wait for the outcome

    ///

    /// The guts of `start_conversation` and `Conversation::send_msg`

}

/// A low-level client tunnel API.

///

/// This is a communication channel to the tunnel reactor, which manages 1 or more circuits.

///

/// Note: the tor-circmgr crates wrap this type in specialized *Tunnel types exposing only the

/// desired subset of functionality depending on purpose and path size.

///

/// Some API calls are for single path and some for multi path. A check with the underlying reactor

/// is done preventing for instance multi path calls to be used on a single path. Top level types

/// should prevent this and thus this object should never be used directly.

#[derive(Debug)]

#[cfg_attr(

    feature = "rpc",

    derive(derive_deftly::Deftly),

    derive_deftly(tor_rpcbase::templates::Object)

)]

#[allow(dead_code)] // TODO(conflux)

pub struct ClientTunnel {

    /// The underlying handle to the reactor.

    circ: ClientCirc,

}

impl ClientTunnel {

    /// Return a handle to the `ClientCirc` of this `ClientTunnel`, if the tunnel is a single

    /// circuit tunnel.

    ///

    /// Returns an error if the tunnel has more than one circuit.

    /// Return the channel target of the first hop.

    ///

    /// Can only be used for single path tunnel.

    /// Return true if the circuit reactor is closed meaning the circuit is unusable for both

    /// receiving or sending.

    /// Return a [`TargetHop`] representing precisely the last hop of the circuit as in set as a

    /// HopLocation with its id and hop number.

    ///

    /// Return an error if there is no last hop.

    /// Return a description of the last hop of the tunnel.

    ///

    /// Return None if the last hop is virtual; return an error

    /// if the tunnel has no circuits, or all of its circuits are zero length.

    ///

    ///

    /// # Panics

    ///

    /// Panics if there is no last hop.  (This should be impossible outside of

    /// the tor-proto crate, but within the crate it's possible to have a

    /// circuit with no hops.)

    /// Return the number of hops this tunnel as. Fail for a multi path.

    /// Return the [`Path`] objects describing all the hops

    /// of all the circuits in this tunnel.

    /// Return a representation of the Paths for all the circuits in this tunnel,

    /// as a map from each circuits' UniqId to its path.

    ///

    /// This is only exposed for the RPC subsystem, where it is documented that the

    /// format of `UniqId` is not stable.

    #[cfg(feature = "rpc")]

    /// Return a process-unique identifier for this tunnel.

    ///

    /// Returns the reactor unique ID of the main reactor.

    /// Return the time at which this tunnel last had any open streams.

    ///

    /// Returns `None` if this tunnel has never had any open streams,

    /// or if it currently has open streams.

    ///

    /// NOTE that the Instant returned by this method is not affected by

    /// any runtime mocking; it is the output of an ordinary call to

    /// `Instant::now()`.

    /// Return a future that will resolve once the underlying circuit reactor has closed.

    ///

    /// Note that this method does not itself cause the tunnel to shut down.

    /// Single-path tunnel only. Multi path onion service is not supported yet.

    ///

    /// Tell this tunnel to begin allowing the final hop of the tunnel to try

    /// to create new Tor streams, and to return those pending requests in an

    /// asynchronous stream.

    ///

    /// Ordinarily, these requests are rejected.

    ///

    /// There can only be one [`Stream`](futures::Stream) of this type created on a given tunnel.

    /// If a such a [`Stream`](futures::Stream) already exists, this method will return

    /// an error.

    ///

    /// After this method has been called on a tunnel, the tunnel is expected

    /// to receive requests of this type indefinitely, until it is finally closed.

    /// If the `Stream` is dropped, the next request on this tunnel will cause it to close.

    ///

    /// Only onion services (and eventually) exit relays should call this

    /// method.

    //

    // TODO: Someday, we might want to allow a stream request handler to be

    // un-registered.  However, nothing in the Tor protocol requires it.

    //

    // Any incoming request handlers installed on the other circuits

    // (which are shutdown using CtrlCmd::ShutdownAndReturnCircuit)

    // will be discarded (along with the reactor of that circuit)

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

    #[allow(unreachable_code, unused_variables)] // TODO(conflux)

        use futures::stream::StreamExt;

        /// The size of the channel receiving IncomingStreamRequestContexts.

        const INCOMING_BUFFER: usize = STREAM_READER_BUFFER;

        // TODO(#2002): support onion service conflux

            "Cannot allow stream requests on a multi-path tunnel"

        // Check whether the AwaitStreamRequest was processed successfully.

        }

            let StreamReqInfo {

            // We already enforce this in handle_incoming_stream_request; this

            // assertion is just here to make sure that we don't ever

            // accidentally remove or fail to enforce that check, since it is

            // security-critical.

            // TODO(#2002): figure out what this is going to look like

            // for onion services (perhaps we should forbid this function

            // from being called on a multipath circuit?)

            //

            // See also:

            // https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/3002#note_3200937

            // can be used to build a reader that supports XON/XOFF flow control

    /// Single and Multi path helper, used to begin a stream.

    ///

    /// This function allocates a stream ID, and sends the message

    /// (like a BEGIN or RESOLVE), but doesn't wait for a response.

    ///

    /// The caller will typically want to see the first cell in response,

    /// to see whether it is e.g. an END or a CONNECTED.

        // TODO: Possibly this should take a hop, rather than just

        // assuming it's the last hop.

            #[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.

        // can be used to build a reader that supports XON/XOFF flow control

    /// Install a [`CircuitPadder`] at the listed `hop`.

    ///

    /// Replaces any previous padder installed at that hop.

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

    /// Remove any [`CircuitPadder`] at the listed `hop`.

    ///

    /// Does nothing if there was not a padder installed there.

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

    /// Start a DataStream (anonymized connection) to the given

    /// address and port, using a BEGIN cell.

        let StreamComponents {

        );

    /// Single and multi path helper.

    ///

    /// Start a stream to the given address and port, using a BEGIN

    /// cell.

    ///

    /// The use of a string for the address is intentional: you should let

    /// the remote Tor relay do the hostname lookup for you.

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

        let parameters = parameters.unwrap_or_default();

        let begin_flags = parameters.begin_flags();

        let optimistic = parameters.is_optimistic();

        let target = if parameters.suppressing_hostname() {

            ""

        } else {

            target

        };

        let beginmsg = Begin::new(target, port, begin_flags)

        self.begin_data_stream(beginmsg.into(), optimistic).await

    /// Start a new stream to the last relay in the tunnel, using

    /// a BEGIN_DIR cell.

        // Note that we always open begindir connections optimistically.

        // Since they are local to a relay that we've already authenticated

        // with and built a tunnel to, there should be no additional checks

        // we need to perform to see whether the BEGINDIR will succeed.

    /// Perform a DNS lookup, using a RESOLVE cell with the last relay

    /// in this tunnel.

    ///

    /// Note that this function does not check for timeouts; that's

    /// the caller's responsibility.

    /// Perform a reverse DNS lookup, by sending a RESOLVE cell with

    /// the last relay on this tunnel.

    ///

    /// Note that this function does not check for timeouts; that's

    /// the caller's responsibility.

                ),

    /// Send an ad-hoc message to a given hop on the circuit, without expecting

    /// a reply.

    ///

    /// (If you want to handle one or more possible replies, see

    /// [`ClientTunnel::start_conversation`].)

    // TODO(conflux): Change this to use the ReactorHandle for the control commands.

    #[cfg(feature = "send-control-msg")]

    /// Start an ad-hoc protocol exchange to the specified hop on this tunnel.

    ///

    /// To use this:

    ///

    ///  0. Create an inter-task channel you'll use to receive

    ///     the outcome of your conversation,

    ///     and bundle it into a [`UserMsgHandler`].

    ///

    ///  1. Call `start_conversation`.

    ///     This will install a your handler, for incoming messages,

    ///     and send the outgoing message (if you provided one).

    ///     After that, each message on the circuit

    ///     that isn't handled by the core machinery

    ///     is passed to your provided `reply_handler`.

    ///

    ///  2. Possibly call `send_msg` on the [`Conversation`],

    ///     from the call site of `start_conversation`,

    ///     possibly multiple times, from time to time,

    ///     to send further desired messages to the peer.

    ///

    ///  3. In your [`UserMsgHandler`], process the incoming messages.

    ///     You may respond by

    ///     sending additional messages

    ///     When the protocol exchange is finished,

    ///     `UserMsgHandler::handle_msg` should return

    ///     [`ConversationFinished`](reactor::MetaCellDisposition::ConversationFinished).

    ///

    /// If you don't need the `Conversation` to send followup messages,

    /// you may simply drop it,

    /// and rely on the responses you get from your handler,

    /// on the channel from step 0 above.

    /// Your handler will remain installed and able to process incoming messages

    /// until it returns `ConversationFinished`.

    ///

    /// (If you don't want to accept any replies at all, it may be

    /// simpler to use [`ClientTunnel::send_raw_msg`].)

    ///

    /// Note that it is quite possible to use this function to violate the tor

    /// protocol; most users of this API will not need to call it.  It is used

    /// to implement most of the onion service handshake.

    ///

    /// # Limitations

    ///

    /// Only one conversation may be active at any one time,

    /// for any one circuit.

    /// This generally means that this function should not be called

    /// on a tunnel which might be shared with anyone else.

    ///

    /// Likewise, it is forbidden to try to extend the tunnel,

    /// while the conversation is in progress.

    ///

    /// After the conversation has finished, the tunnel may be extended.

    /// Or, `start_conversation` may be called again;

    /// but, in that case there will be a gap between the two conversations,

    /// during which no `UserMsgHandler` is installed,

    /// and unexpected incoming messages would close the tunnel.

    ///

    /// If these restrictions are violated, the tunnel will be closed with an error.

    ///

    /// ## Precise definition of the lifetime of a conversation

    ///

    /// A conversation is in progress from entry to `start_conversation`,

    /// until entry to the body of the [`UserMsgHandler::handle_msg`](MsgHandler::handle_msg)

    /// call which returns [`ConversationFinished`](reactor::MetaCellDisposition::ConversationFinished).

    /// (*Entry* since `handle_msg` is synchronously embedded

    /// into the incoming message processing.)

    /// So you may start a new conversation as soon as you have the final response

    /// via your inter-task channel from (0) above.

    ///

    /// The lifetime relationship of the [`Conversation`],

    /// vs the handler returning `ConversationFinished`

    /// is not enforced by the type system.

    // Doing so without still leaving plenty of scope for runtime errors doesn't seem possible,

    // at least while allowing sending followup messages from outside the handler.

    #[cfg(feature = "send-control-msg")]

        // We need to resolve the TargetHop into a precise HopLocation so our msg handler can match

        // the right Leg/Hop with inbound cell.

    /// Shut down this tunnel, along with all streams that are using it. Happens asynchronously

    /// (i.e. the tunnel won't necessarily be done shutting down immediately after this function

    /// returns!).

    ///

    /// Note that other references to this tunnel may exist. If they do, they will stop working

    /// after you call this function.

    ///

    /// It's not necessary to call this method if you're just done with a tunnel: the tunnel should

    /// close on its own once nothing is using it any more.

    // TODO(conflux): This should use the ReactorHandle instead.

    /// Helper: Send the resolve message, and read resolved message from

    /// resolve stream.

        let StreamComponents {

            target: _,

            xon_xoff_reader_ctrl: _,

    // TODO(conflux)

}

// TODO(conflux): We will likely need to enforce some invariants here, for example that the `circ`

// has the expected (non-zero) number of hops.

impl TryFrom<ClientCirc> for ClientTunnel {

    type Error = Error;

}

/// Convert a [`ResolvedVal`] into a Result, based on whether or not

/// it represents an error.

    }

/// A precise position in a tunnel.

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

#[derive_deftly(HasMemoryCost)]

#[non_exhaustive]

pub enum HopLocation {

    /// A specific position in a tunnel.

    Hop((UniqId, HopNum)),

    /// The join point of a multi-path tunnel.

    JoinPoint,

}

/// A position in a tunnel.

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

#[non_exhaustive]

pub enum TargetHop {

    /// A specific position in a tunnel.

    Hop(HopLocation),

    /// The last hop of a tunnel.

    ///

    /// This should be used only when you don't care about what specific hop is used.

    /// Some tunnels may be extended or truncated,

    /// which means that the "last hop" may change at any time.

    LastHop,

}

impl From<(UniqId, HopNum)> for HopLocation {

}

impl From<(UniqId, HopNum)> for TargetHop {

}

impl HopLocation {

    /// Return the hop number if not a JointPoint.

        }

}

impl ClientTunnel {

    /// Close the pending stream that owns this StreamTarget, delivering the specified

    /// END message (if any)

    ///

    /// See [`StreamTarget::close_pending`].

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

    /// Request to send a SENDME cell for this stream.

    ///

    /// See [`StreamTarget::send_sendme`].

    /// Inform the circuit reactor that there has been a change in the drain rate for this stream.

    ///

    /// See [`StreamTarget::drain_rate_update`].

}