//! Launching pluggable transport binaries and communicating with them.

//!

//! This module contains utilities to launch pluggable transports supporting pt-spec.txt

//! version 1, and communicate with them in order to specify configuration parameters and

//! receive updates as to the current state of the PT.

use crate::err;

use crate::err::PtError;

use crate::PtClientMethod;

use futures::channel::mpsc::Receiver;

use futures::StreamExt;

use itertools::Itertools;

use std::borrow::Cow;

use std::collections::HashMap;

use std::ffi::OsString;

use std::io::{BufRead, BufReader};

use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};

use std::path::PathBuf;

use std::process::{Child, Command, Stdio};

use std::str::FromStr;

use std::sync::Arc;

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

use std::{io, thread};

use tor_basic_utils::PathExt as _;

use tor_error::{internal, warn_report};

use tor_linkspec::PtTransportName;

use tor_rtcompat::{Runtime, SleepProviderExt};

use tor_socksproto::SocksVersion;

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

/// Amount of time we give a pluggable transport child process to exit gracefully.

const GRACEFUL_EXIT_TIME: Duration = Duration::from_secs(5);

/// Default timeout for PT binary startup.

const PT_START_TIMEOUT: Duration = Duration::from_secs(30);

/// Size for the buffer storing pluggable transport stdout lines.

const PT_STDIO_BUFFER: usize = 64;

/// An arbitrary key/value status update from a pluggable transport.

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

pub struct PtStatus {

    /// Arbitrary key-value data about the state of this transport, from the binary running

    /// said transport.

    // NOTE(eta): This is assumed to not have duplicate keys.

    data: HashMap<String, String>,

}

/// A message sent from a pluggable transport child process.

///

/// For more in-depth information about these messages, consult pt-spec.txt.

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

#[non_exhaustive]

#[cfg_attr(feature = "experimental-api", visibility::make(pub))]

pub enum PtMessage {

    /// `VERSION-ERROR`: No compatible pluggable transport specification version was provided.

    VersionError(String),

    /// `VERSION`: Specifies the version the binary is using for the IPC protocol.

    Version(String),

    /// `ENV-ERROR`: Reports an error with the provided environment variables.

    EnvError(String),

    /// `PROXY DONE`: The configured proxy was correctly initialised.

    ProxyDone,

    /// `PROXY-ERROR`: An error was encountered setting up the configured proxy.

    ProxyError(String),

    /// `CMETHOD`: A client transport has been launched.

    ClientTransportLaunched {

        /// The name of the launched transport.

        transport: PtTransportName,

        /// The protocol used ('socks4' or 'socks5').

        protocol: String,

        /// An address to connect via this transport.

        /// (This should be localhost.)

        endpoint: SocketAddr,

    },

    /// `CMETHOD-ERROR`: An error was encountered setting up a client transport.

    ClientTransportFailed {

        /// The name of the transport.

        transport: PtTransportName,

        /// The error message.

        message: String,

    },

    /// `CMETHODS DONE`: All client transports that are supported have been launched.

    ClientTransportsDone,

    /// `SMETHOD`: A server transport has been launched.

    ServerTransportLaunched {

        /// The name of the launched transport.

        transport: PtTransportName,

        /// The endpoint clients should use the reach the transport.

        endpoint: SocketAddr,

        /// Additional per-transport information.

        // NOTE(eta): This assumes it actually is k/v and repeated keys aren't allowed...

        options: HashMap<String, String>,

    },

    /// `SMETHOD-ERROR`: An error was encountered setting up a server transport.

    ServerTransportFailed {

        /// The name of the transport.

        transport: PtTransportName,

        /// The error message.

        message: String,

    },

    /// `SMETHODS DONE`: All server transports that are supported have been launched.

    ServerTransportsDone,

    /// `LOG`: A log message.

    Log {

        /// The severity (one of 'error', 'warning', 'notice', 'info', 'debug').

        severity: String,

        /// The log message.

        message: String,

    },

    /// `STATUS`: Arbitrary key/value status messages.

    Status(PtStatus),

    /// A line containing an unknown command.

    Unknown(String),

}

/// Parse a value (something on the RHS of an =), which could be a CString as defined by

/// control-spec.txt §2. Returns (value, unparsed rest of string).

        // This is a CString, so we're going to need to parse it char-by-char.

        // FIXME(eta): This currently doesn't parse octal escape codes, even though the spec says

        //             we should. That's finicky, though, and probably not used.

        loop {

                {

                },

            }

        }

    } else {

        // Simple: just find the space

    })

/// Chomp one key/value pair off a list of smethod args.

/// Returns (k, v, unparsed rest of string).

/// Will also chomp the comma at the end, if there is one.

            '\\' => {

            }

            '=' => {

            }

        }

    }

impl FromStr for PtMessage {

    type Err = Cow<'static, str>;

    // NOTE(eta): This, of course, implies that the PT IPC communications are valid UTF-8.

    //            This assumption might turn out to be false.

    #[allow(clippy::cognitive_complexity)]

            }

            }

            }

            },

            }

                }

            }

                }

            }

            },

                // The SMETHOD endpoint is the place where _clients_ connect, and it shouldn't be localhost.

                // NOTE(eta): pt-spec.txt seems to imply these options can't contain spaces, so

                //            we work under that assumption.

                //            It also doesn't actually parse them out -- but seeing as the API to

                //            feed these back in will want them as separated k/v pairs, I think

                //            it makes sense to here.

                                // At least check our assumption that this is actually k/v

                                // and not Vec<(String, String)>.

                        }

                }

                Self::ServerTransportLaunched {

                }

            }

                }

            }

            },

                )

                .0;

            }

                        // At least check our assumption that this is actually k/v

                        // and not Vec<(String, String)>.

                }

            }

        })

}

use sealed::*;

/// Sealed trait to protect private types and default trait implementations

pub(crate) mod sealed {

    use super::*;

    /// A handle to receive lines from a pluggable transport process' stdout asynchronously.

    //

    // FIXME(eta): This currently spawns an OS thread, since there's no other way to do this without

    //             being async-runtime dependent (or adding process spawning to tor-rtcompat).

    #[derive(Debug)]

    pub struct AsyncPtChild {

        /// Channel to receive lines from the child process stdout.

        stdout: Receiver<io::Result<String>>,

        /// Identifier to put in logging messages.

        pub identifier: String,

    }

    impl AsyncPtChild {

        /// Wrap an OS child process by spawning a worker thread to forward output from the child

        /// to the asynchronous runtime via use of a channel.

            );

            // TODO RELAY #1649 We don't use a tor_memquota::mq_queue here yet

                // Forward lines from the blocking reader to the async channel.

                    }

                            // Channel dropped, so shut down the pluggable transport process.

                                ident

                            );

                        // Encountered an error reading, so ensure the process is shut down (it's

                        // probably "broken pipe" anyway, so this is slightly redundant, but the

                        // rest of the code assumes errors are nonrecoverable).

                }

                // Has it already quit? If so, just exit now.

                    // FIXME(eta): We currently throw away the exit code, which might be useful

                    //             for debugging purposes!

                    Ok(None) => {

                        // Kill it.

                    }

                    Ok(Some(_)) => {

                    } // It exited.

                    }

                }

        /// Receive a message from the pluggable transport binary asynchronously.

        ///

        /// Note: This will convert `PtMessage::Log` into a tracing log call automatically.

        #[allow(clippy::cognitive_complexity)] // due to tracing

            loop {

                            // FIXME(eta): I wanted to make this integrate with `tracing` more nicely,

                            //             but gave up after 15 minutes of clicking through spaghetti.

                            }

                        } else {

                        }

                    }

                    }

                }

            }

    }

    /// Defines some helper methods that are required later on

    #[async_trait::async_trait]

    pub trait PluggableTransportPrivate {

        /// Return the [`AsyncPtChild`] if it exists

        fn inner(&mut self) -> Result<&mut AsyncPtChild, PtError>;

        /// Set the [`AsyncPtChild`]

        fn set_inner(&mut self, newval: Option<AsyncPtChild>);

        /// Return a loggable identifier for this transport.

        fn identifier(&self) -> &str;

        /// Checks whether a transport is specified in our specific parameters

        fn specific_params_contains(&self, transport: &PtTransportName) -> bool;

        /// Common handler for `ClientTransportLaunched` and `ServerTransportLaunched`

                    }

                },

            };

        /// Attempt to launch the PT and return the corresponding `[AsyncPtChild]`

                // WARN: this may not be the correct error to throw here

                transports

            );

        /// Consolidates some of the [`PtMessage`] potential matches to

        /// deduplicate code

        ///

        /// Note that getting a [`PtMessage`] from this method implies that

        /// the method was unable to match it and thus you should continue handling

        /// the message. Getting [`None`] after error handling means that a match

        /// was found and the appropriate action was successfully taken, and you don't

        /// need to worry about it.

        async fn try_match_common_messages<R: Runtime>(

            &self,

            rt: &R,

            deadline: Instant,

            async_child: &mut AsyncPtChild,

            {

                        async_child.identifier, transport, message

                    );

                }

                }

                }

                // TODO(eta): We don't do anything with these right now!

                }

                // Return the PtMessage as it is for further processing

                // TODO: handle [`PtError::ProtocolViolation`] here somehow

                }

            }

    }

}

/// Common parameters passed to a pluggable transport.

pub struct PtCommonParameters {

    /// A path where the launched PT can store state.

    state_location: PathBuf,

    /// An IPv4 address to bind outgoing connections to (if specified).

    ///

    /// Leaving this out will mean the PT uses a sane default.

    #[builder(default)]

    outbound_bind_v4: Option<Ipv4Addr>,

    /// An IPv6 address to bind outgoing connections to (if specified).

    ///

    /// Leaving this out will mean the PT uses a sane default.

    #[builder(default)]

    outbound_bind_v6: Option<Ipv6Addr>,

    /// The maximum time we should wait for a pluggable transport binary to report successful

    /// initialization. If `None`, a default value is used.

    #[builder(default)]

    timeout: Option<Duration>,

}

impl PtCommonParameters {

    /// Return a new `PtCommonParametersBuilder` for constructing a set of parameters.

    /// Convert these parameters into a set of environment variables to be passed to the PT binary

    /// in accordance with the specification.

}

/// Parameters passed only to a pluggable transport client.

pub struct PtClientParameters {

    /// A SOCKS URI specifying a proxy to use.

    #[builder(default)]

    proxy_uri: Option<String>,

    /// A list of transports to initialise.

    ///

    /// The PT launch will fail if all transports are not successfully initialised.

    transports: Vec<PtTransportName>,

}

impl PtClientParameters {

    /// Return a new `PtClientParametersBuilder` for constructing a set of parameters.

    /// Convert these parameters into a set of environment variables to be passed to the PT binary

    /// in accordance with the specification.

}

/// Parameters passed only to a pluggable transport server.

pub struct PtServerParameters {

    /// A list of transports to initialise.

    ///

    /// The PT launch will fail if all transports are not successfully initialised.

    transports: Vec<PtTransportName>,

    /// Transport options for each server transport

    #[builder(default)]

    server_transport_options: String,

    /// Set host:port on which the server transport should listen for connections

    #[builder(default)]

    server_bindaddr: String,

    /// Set host:port on which the server transport should forward requests

    #[builder(default)]

    server_orport: Option<String>,

    /// Set host:port on which the server transport should forward requests (extended ORPORT)

    #[builder(default)]

    server_extended_orport: Option<String>,

}

impl PtServerParameters {

    /// Return a new `PtServerParametersBuilder` for constructing a set of parameters.

    /// Convert these parameters into a set of environment variables to be passed to the PT binary

    /// in accordance with the specification.

}

/// Common functionality implemented to allow code reuse

#[async_trait::async_trait]

#[cfg_attr(feature = "experimental-api", visibility::make(pub))]

pub trait PluggableTransport: PluggableTransportPrivate {

    /// Get all client methods returned by the binary, if it has been launched.

    ///

    /// If it hasn't been launched, the returned map will be empty.

    // TODO(eta): Actually figure out a way to expose this more stably.

    fn transport_methods(&self) -> &HashMap<PtTransportName, PtClientMethod>;

    /// Get the next [`PtMessage`] from the running transport. It is recommended to call this

    /// in a loop once a PT has been launched, in order to forward log messages and find out about

    /// status updates.

    //

    // FIXME(eta): This API will probably go away and get replaced with something better.

    //             In particular, we'd want to cache `Status` messages from before this method

    //             was called.

            // FIXME(eta): Currently this lets the caller still think the methods work by calling

            //             transport_methods.

                ret

            );

}

/// A pluggable transport binary in a child process.

///

/// These start out inert, and must be launched with [`PluggableClientTransport::launch`] in order

/// to be useful.

#[derive(Debug)]

pub struct PluggableClientTransport {

    /// The currently running child, if there is one.

    inner: Option<AsyncPtChild>,

    /// The path to the binary to run.

    pub(crate) binary_path: PathBuf,

    /// Arguments to pass to the binary.

    arguments: Vec<String>,

    /// Configured parameters.

    common_params: PtCommonParameters,

    /// Configured client-only parameters.

    client_params: PtClientParameters,

    /// Information about client methods obtained from the PT.

    cmethods: HashMap<PtTransportName, PtClientMethod>,

}

impl PluggableTransport for PluggableClientTransport {

}

impl PluggableTransportPrivate for PluggableClientTransport {

        }

}

impl PluggableClientTransport {

    /// Create a new pluggable transport wrapper, wrapping the binary at `binary_path` and passing

    /// the `params` to it.

    ///

    /// You must call [`PluggableClientTransport::launch`] to actually run the PT.

    /// Launch the pluggable transport, executing the binary.

    ///

    /// Will return an error if the launch fails, one of the transports fail, not all transports

    /// were launched, or the launch times out.

        loop {

            {

                            PtMessage::ClientTransportLaunched {

                            }

                            PtMessage::ProxyDone => {

                            }

                            // TODO: unify most of the handling of ClientTransportsDone with ServerTransportsDone

                            PtMessage::ClientTransportsDone => {

                                        unsupported

                                    );

                            }

                            }

                        }

                }

            }

        }

}

/// A pluggable transport server binary in a child process.

///

/// These start out inert, and must be launched with [`PluggableServerTransport::launch`] in order

/// to be useful.

#[derive(Debug)]

pub struct PluggableServerTransport {

    /// The currently running child, if there is one.

    inner: Option<AsyncPtChild>,

    /// The path to the binary to run.

    pub(crate) binary_path: PathBuf,

    /// Arguments to pass to the binary.

    arguments: Vec<String>,

    /// Configured parameters.

    common_params: PtCommonParameters,

    /// Configured server-only parameters.

    server_params: PtServerParameters,

    /// Information about server methods obtained from the PT.

    smethods: HashMap<PtTransportName, PtClientMethod>,

}

impl PluggableTransportPrivate for PluggableServerTransport {

        }

}

impl PluggableTransport for PluggableServerTransport {

}

impl PluggableServerTransport {

    /// Create a new pluggable transport wrapper, wrapping the binary at `binary_path` and passing

    /// the `params` to it.

    ///

    /// You must call [`PluggableServerTransport::launch`] to actually run the PT.

    /// Launch the pluggable transport, executing the binary.

    ///

    /// Will return an error if the launch fails, one of the transports fail, not all transports

    /// were launched, or the launch times out.

        loop {

            {

                            PtMessage::ServerTransportLaunched {

                            }

                            PtMessage::ServerTransportsDone => {

                                        unsupported

                                    );

                            }

                            }

                        }

                }

            }

        }

}

#[cfg(test)]

mod test {

    // @@ begin test lint list maintained by maint/add_warning @@

    #![allow(clippy::bool_assert_comparison)]

    #![allow(clippy::clone_on_copy)]

    #![allow(clippy::dbg_macro)]

    #![allow(clippy::mixed_attributes_style)]

    #![allow(clippy::print_stderr)]

    #![allow(clippy::print_stdout)]

    #![allow(clippy::single_char_pattern)]

    #![allow(clippy::unwrap_used)]

    #![allow(clippy::unchecked_duration_subtraction)]

    #![allow(clippy::useless_vec)]

    #![allow(clippy::needless_pass_by_value)]

    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->

    use crate::ipc::{PtMessage, PtStatus};

    use std::borrow::Cow;

    use std::collections::HashMap;

    use std::net::{IpAddr, Ipv4Addr, SocketAddr};

    #[test]

    fn it_parses_spec_examples() {

        assert_eq!(

            "VERSION-ERROR no-version".parse(),

            Ok(PtMessage::VersionError("no-version".into()))

        );

        assert_eq!("VERSION 1".parse(), Ok(PtMessage::Version("1".into())));

        assert_eq!(

            "ENV-ERROR No TOR_PT_AUTH_COOKIE_FILE when TOR_PT_EXTENDED_SERVER_PORT set".parse(),

            Ok(PtMessage::EnvError(

                "No TOR_PT_AUTH_COOKIE_FILE when TOR_PT_EXTENDED_SERVER_PORT set".into()

            ))

        );

        assert_eq!("PROXY DONE".parse(), Ok(PtMessage::ProxyDone));

        assert_eq!(

            "PROXY-ERROR SOCKS 4 upstream proxies unsupported".parse(),

            Ok(PtMessage::ProxyError(

                "SOCKS 4 upstream proxies unsupported".into()

            ))

        );

        assert_eq!(

            "CMETHOD trebuchet socks5 127.0.0.1:19999".parse(),

            Ok(PtMessage::ClientTransportLaunched {

                transport: "trebuchet".parse().unwrap(),

                protocol: "socks5".to_string(),

                endpoint: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 19999)

            })

        );

        assert_eq!(

            "CMETHOD-ERROR trebuchet no rocks available".parse(),

            Ok(PtMessage::ClientTransportFailed {

                transport: "trebuchet".parse().unwrap(),

                message: "no rocks available".to_string()

            })

        );

        assert_eq!("CMETHODS DONE".parse(), Ok(PtMessage::ClientTransportsDone));

        assert_eq!(

            "SMETHOD trebuchet 198.51.100.1:19999".parse(),

            Ok(PtMessage::ServerTransportLaunched {

                transport: "trebuchet".parse().unwrap(),

                endpoint: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1)), 19999),

                options: Default::default()

            })

        );

        let mut map = HashMap::new();

        map.insert("N".to_string(), "13".to_string());

        assert_eq!(

            "SMETHOD rot_by_N 198.51.100.1:2323 ARGS:N=13".parse(),

            Ok(PtMessage::ServerTransportLaunched {

                transport: "rot_by_N".parse().unwrap(),

                endpoint: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1)), 2323),

                options: map

            })

        );

        let mut map = HashMap::new();

        map.insert(

            "cert".to_string(),

            "HszPy3vWfjsESCEOo9ZBkRv6zQ/1mGHzc8arF0y2SpwFr3WhsMu8rK0zyaoyERfbz3ddFw".to_string(),

        );

        map.insert("iat-mode".to_string(), "0".to_string());

        assert_eq!(

            "SMETHOD obfs4 198.51.100.1:43734 ARGS:cert=HszPy3vWfjsESCEOo9ZBkRv6zQ/1mGHzc8arF0y2SpwFr3WhsMu8rK0zyaoyERfbz3ddFw,iat-mode=0".parse(),

            Ok(PtMessage::ServerTransportLaunched {

                transport: "obfs4".parse().unwrap(),

                endpoint: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1)), 43734),

                options: map

            })

        );

        assert_eq!(

            "SMETHOD-ERROR trebuchet no cows available".parse(),

            Ok(PtMessage::ServerTransportFailed {

                transport: "trebuchet".parse().unwrap(),

                message: "no cows available".to_string()

            })

        );

        assert_eq!(

            "LOG SEVERITY=debug MESSAGE=\"Connected to bridge A\"".parse(),

            Ok(PtMessage::Log {

                severity: "debug".to_string(),

                message: "Connected to bridge A".to_string()

            })

        );

        assert_eq!(

            "LOG SEVERITY=debug MESSAGE=\"\\r\\n\\t\"".parse(),

            Ok(PtMessage::Log {

                severity: "debug".to_string(),

                message: "\r\n\t".to_string()

            })

        );

        assert_eq!(

            "LOG SEVERITY=debug MESSAGE=".parse(),

            Ok(PtMessage::Log {

                severity: "debug".to_string(),

                message: "".to_string()

            })

        );

        assert_eq!(

            "LOG SEVERITY=debug MESSAGE=\"\\a\"".parse::<PtMessage>(),

            Ok(PtMessage::Log {

                severity: "debug".to_string(),

                message: "a".to_string()

            })

        );

        for i in 0..9 {

            let msg = format!("LOG SEVERITY=debug MESSAGE=\"\\{i}\"");

            assert_eq!(

                msg.parse::<PtMessage>(),

                Err(Cow::from("attempted unsupported octal escape code"))

            );

        }

        assert_eq!(

            "SMETHOD obfs4 198.51.100.1:43734 ARGS:iat-mode=0\\".parse::<PtMessage>(),

            Err(Cow::from(

                "failed to parse SMETHOD ARGS: smethod arg terminates with backslash"

            ))

        );

        assert_eq!(

            "SMETHOD obfs4 198.51.100.1:43734 ARGS:iat-mode=fo=o".parse::<PtMessage>(),

            Err(Cow::from(

                "failed to parse SMETHOD ARGS: encountered = while parsing value"

            ))

        );

        assert_eq!(

            "SMETHOD obfs4 198.51.100.1:43734 ARGS:iat-mode".parse::<PtMessage>(),

            Err(Cow::from(

                "failed to parse SMETHOD ARGS: ran out of chars parsing smethod arg"

            ))

        );

        let mut map = HashMap::new();

        map.insert("ADDRESS".to_string(), "198.51.100.123:1234".to_string());

        map.insert("CONNECT".to_string(), "Success".to_string());

        assert_eq!(

            "STATUS ADDRESS=198.51.100.123:1234 CONNECT=Success".parse(),

            Ok(PtMessage::Status(PtStatus { data: map }))

        );

        let mut map = HashMap::new();

        map.insert("ADDRESS".to_string(), "198.51.100.123:1234".to_string());

        map.insert("CONNECT".to_string(), "Success".to_string());

        map.insert("TRANSPORT".to_string(), "obfs4".to_string());

        assert_eq!(

            "STATUS TRANSPORT=obfs4 ADDRESS=198.51.100.123:1234 CONNECT=Success".parse(),

            Ok(PtMessage::Status(PtStatus { data: map }))

        );

        let mut map = HashMap::new();

        map.insert("ADDRESS".to_string(), "198.51.100.222:2222".to_string());

        map.insert("CONNECT".to_string(), "Failed".to_string());

        map.insert("FINGERPRINT".to_string(), "<Fingerprint>".to_string());

        map.insert("ERRSTR".to_string(), "Connection refused".to_string());

        assert_eq!(

            "STATUS ADDRESS=198.51.100.222:2222 CONNECT=Failed FINGERPRINT=<Fingerprint> ERRSTR=\"Connection refused\"".parse(),

            Ok(PtMessage::Status(PtStatus {

                data: map

            }))

        );

    }

}