1
//! Launching pluggable transport binaries and communicating with them.
2
//!
3
//! This module contains utilities to launch pluggable transports supporting pt-spec.txt
4
//! version 1, and communicate with them in order to specify configuration parameters and
5
//! receive updates as to the current state of the PT.
6

            
7
use crate::err;
8
use crate::err::PtError;
9
use crate::PtClientMethod;
10
use futures::channel::mpsc::Receiver;
11
use futures::StreamExt;
12
use itertools::Itertools;
13
use std::borrow::Cow;
14
use std::collections::HashMap;
15
use std::ffi::OsString;
16
use std::io::{BufRead, BufReader};
17
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
18
use std::path::PathBuf;
19
use std::process::{Child, Command, Stdio};
20
use std::str::FromStr;
21
use std::sync::Arc;
22
use std::time::{Duration, Instant};
23
use std::{io, thread};
24
use tor_basic_utils::PathExt as _;
25
use tor_error::{internal, warn_report};
26
use tor_linkspec::PtTransportName;
27
use tor_rtcompat::{Runtime, SleepProviderExt};
28
use tor_socksproto::SocksVersion;
29
use tracing::{debug, error, info, trace, warn};
30

            
31
/// Amount of time we give a pluggable transport child process to exit gracefully.
32
const GRACEFUL_EXIT_TIME: Duration = Duration::from_secs(5);
33
/// Default timeout for PT binary startup.
34
const PT_START_TIMEOUT: Duration = Duration::from_secs(30);
35
/// Size for the buffer storing pluggable transport stdout lines.
36
const PT_STDIO_BUFFER: usize = 64;
37

            
38
/// An arbitrary key/value status update from a pluggable transport.
39
#[derive(PartialEq, Eq, Debug, Clone)]
40
pub struct PtStatus {
41
    /// Arbitrary key-value data about the state of this transport, from the binary running
42
    /// said transport.
43
    // NOTE(eta): This is assumed to not have duplicate keys.
44
    data: HashMap<String, String>,
45
}
46

            
47
/// A message sent from a pluggable transport child process.
48
///
49
/// For more in-depth information about these messages, consult pt-spec.txt.
50
#[derive(PartialEq, Eq, Debug, Clone)]
51
#[non_exhaustive]
52
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
53
pub enum PtMessage {
54
    /// `VERSION-ERROR`: No compatible pluggable transport specification version was provided.
55
    VersionError(String),
56
    /// `VERSION`: Specifies the version the binary is using for the IPC protocol.
57
    Version(String),
58
    /// `ENV-ERROR`: Reports an error with the provided environment variables.
59
    EnvError(String),
60
    /// `PROXY DONE`: The configured proxy was correctly initialised.
61
    ProxyDone,
62
    /// `PROXY-ERROR`: An error was encountered setting up the configured proxy.
63
    ProxyError(String),
64
    /// `CMETHOD`: A client transport has been launched.
65
    ClientTransportLaunched {
66
        /// The name of the launched transport.
67
        transport: PtTransportName,
68
        /// The protocol used ('socks4' or 'socks5').
69
        protocol: String,
70
        /// An address to connect via this transport.
71
        /// (This should be localhost.)
72
        endpoint: SocketAddr,
73
    },
74
    /// `CMETHOD-ERROR`: An error was encountered setting up a client transport.
75
    ClientTransportFailed {
76
        /// The name of the transport.
77
        transport: PtTransportName,
78
        /// The error message.
79
        message: String,
80
    },
81
    /// `CMETHODS DONE`: All client transports that are supported have been launched.
82
    ClientTransportsDone,
83
    /// `SMETHOD`: A server transport has been launched.
84
    ServerTransportLaunched {
85
        /// The name of the launched transport.
86
        transport: PtTransportName,
87
        /// The endpoint clients should use the reach the transport.
88
        endpoint: SocketAddr,
89
        /// Additional per-transport information.
90
        // NOTE(eta): This assumes it actually is k/v and repeated keys aren't allowed...
91
        options: HashMap<String, String>,
92
    },
93
    /// `SMETHOD-ERROR`: An error was encountered setting up a server transport.
94
    ServerTransportFailed {
95
        /// The name of the transport.
96
        transport: PtTransportName,
97
        /// The error message.
98
        message: String,
99
    },
100
    /// `SMETHODS DONE`: All server transports that are supported have been launched.
101
    ServerTransportsDone,
102
    /// `LOG`: A log message.
103
    Log {
104
        /// The severity (one of 'error', 'warning', 'notice', 'info', 'debug').
105
        severity: String,
106
        /// The log message.
107
        message: String,
108
    },
109
    /// `STATUS`: Arbitrary key/value status messages.
110
    Status(PtStatus),
111
    /// A line containing an unknown command.
112
    Unknown(String),
113
}
114

            
115
/// Parse a value (something on the RHS of an =), which could be a CString as defined by
116
/// control-spec.txt §2. Returns (value, unparsed rest of string).
117
44
fn parse_one_value(from: &str) -> Result<(String, &str), &'static str> {
118
44
    let first_char = from.chars().next();
119
44
    Ok(if first_char.is_none() {
120
2
        (String::new(), "")
121
42
    } else if let Some('"') = first_char {
122
        // This is a CString, so we're going to need to parse it char-by-char.
123
        // FIXME(eta): This currently doesn't parse octal escape codes, even though the spec says
124
        //             we should. That's finicky, though, and probably not used.
125
26
        let mut ret = String::new();
126
26
        let mut chars = from.chars();
127
26
        assert_eq!(chars.next(), Some('"')); // discard "
128
        loop {
129
112
            let ch = chars.next().ok_or("ran out of input parsing CString")?;
130
112
            match ch {
131
26
                '\\' => match chars
132
26
                    .next()
133
26
                    .ok_or("encountered trailing backslash in CString")?
134
                {
135
2
                    'n' => ret.push('\n'),
136
2
                    'r' => ret.push('\r'),
137
2
                    't' => ret.push('\t'),
138
20
                    '0'..='8' => return Err("attempted unsupported octal escape code"),
139
2
                    ch2 => ret.push(ch2),
140
                },
141
8
                '"' => break,
142
78
                _ => ret.push(ch),
143
            }
144
        }
145
8
        (ret, chars.as_str())
146
    } else {
147
        // Simple: just find the space
148
16
        let space = from.find(' ').unwrap_or(from.len());
149
16
        (from[0..space].into(), &from[space..])
150
    })
151
44
}
152

            
153
/// Chomp one key/value pair off a list of smethod args.
154
/// Returns (k, v, unparsed rest of string).
155
/// Will also chomp the comma at the end, if there is one.
156
12
fn parse_one_smethod_arg(args: &str) -> Result<(String, String, &str), &'static str> {
157
12
    // NOTE(eta): Apologies for this looking a bit gnarly. Ideally, this is what you'd use
158
12
    //            something like `nom` for, but I didn't want to bring in a dep just for this.
159
12

            
160
12
    let mut key = String::new();
161
12
    let mut val = String::new();
162
12
    // If true, we're reading the value, not the key.
163
12
    let mut reading_val = false;
164
12
    let mut chars = args.chars();
165
248
    while let Some(c) = chars.next() {
166
242
        let target = if reading_val { &mut val } else { &mut key };
167
242
        match c {
168
            '\\' => {
169
2
                let c = chars
170
2
                    .next()
171
2
                    .ok_or("smethod arg terminates with backslash")?;
172
                target.push(c);
173
            }
174
            '=' => {
175
12
                if reading_val {
176
2
                    return Err("encountered = while parsing value");
177
10
                }
178
10
                reading_val = true;
179
            }
180
2
            ',' => break,
181
226
            c => target.push(c),
182
        }
183
    }
184
8
    if !reading_val {
185
2
        return Err("ran out of chars parsing smethod arg");
186
6
    }
187
6
    Ok((key, val, chars.as_str()))
188
12
}
189

            
190
impl FromStr for PtMessage {
191
    type Err = Cow<'static, str>;
192

            
193
    // NOTE(eta): This, of course, implies that the PT IPC communications are valid UTF-8.
194
    //            This assumption might turn out to be false.
195
    #[allow(clippy::cognitive_complexity)]
196
62
    fn from_str(s: &str) -> Result<Self, Self::Err> {
197
62
        // TODO(eta): Maybe tolerate additional whitespace (using `split_whitespace`)?.
198
62
        //            This requires modified words.join() logic, though.
199
62
        let mut words = s.split(' ');
200
62
        let first_word = words.next().ok_or_else(|| Cow::from("empty line"))?;
201
62
        Ok(match first_word {
202
62
            "VERSION-ERROR" => {
203
2
                let rest = words.join(" ");
204
2
                Self::VersionError(rest)
205
            }
206
60
            "VERSION" => {
207
2
                let vers = words.next().ok_or_else(|| Cow::from("no version"))?;
208
2
                Self::Version(vers.into())
209
            }
210
58
            "ENV-ERROR" => {
211
2
                let rest = words.join(" ");
212
2
                Self::EnvError(rest)
213
            }
214
56
            "PROXY" => match words.next() {
215
2
                Some("DONE") => Self::ProxyDone,
216
                _ => Self::Unknown(s.into()),
217
            },
218
54
            "PROXY-ERROR" => {
219
2
                let rest = words.join(" ");
220
2
                Self::ProxyError(rest)
221
            }
222
52
            "CMETHOD" => {
223
2
                let transport = words.next().ok_or_else(|| Cow::from("no transport"))?;
224
2
                let protocol = words.next().ok_or_else(|| Cow::from("no protocol"))?;
225
2
                let endpoint = words
226
2
                    .next()
227
2
                    .ok_or_else(|| Cow::from("no endpoint"))?
228
2
                    .parse::<SocketAddr>()
229
2
                    .map_err(|e| Cow::from(format!("failed to parse endpoint: {}", e)))?;
230
2
                if !endpoint.ip().is_loopback() {
231
                    return Err(Cow::from(format!(
232
                        "CMETHOD endpoint {endpoint} was not localhost"
233
                    )));
234
2
                }
235
2
                Self::ClientTransportLaunched {
236
2
                    transport: transport
237
2
                        .parse()
238
2
                        .map_err(|_| Cow::from("bad transport ID"))?,
239
2
                    protocol: protocol.to_string(),
240
2
                    endpoint,
241
                }
242
            }
243
50
            "CMETHOD-ERROR" => {
244
2
                let transport = words.next().ok_or_else(|| Cow::from("no transport"))?;
245
2
                let rest = words.join(" ");
246
2
                Self::ClientTransportFailed {
247
2
                    transport: transport
248
2
                        .parse()
249
2
                        .map_err(|_| Cow::from("bad transport ID"))?,
250
2
                    message: rest,
251
                }
252
            }
253
48
            "CMETHODS" => match words.next() {
254
2
                Some("DONE") => Self::ClientTransportsDone,
255
                _ => Self::Unknown(s.into()),
256
            },
257
46
            "SMETHOD" => {
258
12
                let transport = words.next().ok_or_else(|| Cow::from("no transport"))?;
259
12
                let endpoint = words
260
12
                    .next()
261
12
                    .ok_or_else(|| Cow::from("no endpoint"))?
262
12
                    .parse::<SocketAddr>()
263
12
                    .map_err(|e| Cow::from(format!("failed to parse endpoint: {}", e)))?;
264
                // The SMETHOD endpoint is the place where _clients_ connect, and it shouldn't be localhost.
265
12
                let mut parsed_args = HashMap::new();
266

            
267
                // NOTE(eta): pt-spec.txt seems to imply these options can't contain spaces, so
268
                //            we work under that assumption.
269
                //            It also doesn't actually parse them out -- but seeing as the API to
270
                //            feed these back in will want them as separated k/v pairs, I think
271
                //            it makes sense to here.
272
16
                for option in words {
273
10
                    if let Some(mut args) = option.strip_prefix("ARGS:") {
274
16
                        while !args.is_empty() {
275
15
                            let (k, v, rest) = parse_one_smethod_arg(args).map_err(|e| {
276
6
                                Cow::from(format!("failed to parse SMETHOD ARGS: {}", e))
277
15
                            })?;
278
6
                            if parsed_args.contains_key(&k) {
279
                                // At least check our assumption that this is actually k/v
280
                                // and not Vec<(String, String)>.
281
                                warn!("PT SMETHOD arguments contain repeated key {}!", k);
282
6
                            }
283
6
                            parsed_args.insert(k, v);
284
6
                            args = rest;
285
                        }
286
                    }
287
                }
288
                Self::ServerTransportLaunched {
289
6
                    transport: transport
290
6
                        .parse()
291
6
                        .map_err(|_| Cow::from("bad transport ID"))?,
292
6
                    endpoint,
293
6
                    options: parsed_args,
294
                }
295
            }
296
34
            "SMETHOD-ERROR" => {
297
2
                let transport = words.next().ok_or_else(|| Cow::from("no transport"))?;
298
2
                let rest = words.join(" ");
299
2
                Self::ServerTransportFailed {
300
2
                    transport: transport
301
2
                        .parse()
302
2
                        .map_err(|_| Cow::from("bad transport ID"))?,
303
2
                    message: rest,
304
                }
305
            }
306
32
            "SMETHODS" => match words.next() {
307
                Some("DONE") => Self::ServerTransportsDone,
308
                _ => Self::Unknown(s.into()),
309
            },
310
32
            "LOG" => {
311
26
                let severity = words
312
26
                    .next()
313
26
                    .ok_or_else(|| Cow::from("no severity"))?
314
26
                    .strip_prefix("SEVERITY=")
315
26
                    .ok_or_else(|| Cow::from("badly formatted severity"))?;
316
26
                let message = words.join(" ");
317
8
                let message = parse_one_value(
318
26
                    message
319
26
                        .strip_prefix("MESSAGE=")
320
26
                        .ok_or_else(|| Cow::from("no or badly formatted message"))?,
321
                )
322
26
                .map_err(Cow::from)?
323
                .0;
324
8
                Self::Log {
325
8
                    severity: severity.into(),
326
8
                    message,
327
8
                }
328
            }
329
6
            "STATUS" => {
330
6
                let mut ret = HashMap::new();
331
6
                let message = words.join(" ");
332
6
                let mut message = &message as &str;
333
24
                while !message.is_empty() {
334
18
                    let equals = message
335
18
                        .find('=')
336
18
                        .ok_or_else(|| Cow::from(format!("failed to find = in '{}'", message)))?;
337
18
                    let k = &message[..equals];
338
18
                    if equals + 1 == message.len() {
339
                        return Err(Cow::from("key with no value"));
340
18
                    }
341
18
                    let (v, rest) = parse_one_value(&message[(equals + 1)..]).map_err(Cow::from)?;
342
18
                    if ret.contains_key(k) {
343
                        // At least check our assumption that this is actually k/v
344
                        // and not Vec<(String, String)>.
345
                        warn!("STATUS contains repeated key {}!", k);
346
18
                    }
347
18
                    ret.insert(k.to_owned(), v);
348
18
                    message = rest;
349
18
                    if message.starts_with(' ') {
350
12
                        message = &message[1..];
351
12
                    }
352
                }
353
6
                Self::Status(PtStatus { data: ret })
354
            }
355
            _ => Self::Unknown(s.into()),
356
        })
357
62
    }
358
}
359

            
360
use sealed::*;
361
/// Sealed trait to protect private types and default trait implementations
362
pub(crate) mod sealed {
363
    use super::*;
364

            
365
    /// A handle to receive lines from a pluggable transport process' stdout asynchronously.
366
    //
367
    // FIXME(eta): This currently spawns an OS thread, since there's no other way to do this without
368
    //             being async-runtime dependent (or adding process spawning to tor-rtcompat).
369
    #[derive(Debug)]
370
    pub struct AsyncPtChild {
371
        /// Channel to receive lines from the child process stdout.
372
        stdout: Receiver<io::Result<String>>,
373
        /// Identifier to put in logging messages.
374
        pub identifier: String,
375
    }
376

            
377
    impl AsyncPtChild {
378
        /// Wrap an OS child process by spawning a worker thread to forward output from the child
379
        /// to the asynchronous runtime via use of a channel.
380
        pub fn new(mut child: Child, identifier: String) -> Result<Self, PtError> {
381
            let (stdin, stdout) = (
382
                child.stdin.take().ok_or_else(|| {
383
                    PtError::Internal(internal!("Created child process without stdin pipe"))
384
                })?,
385
                child.stdout.take().ok_or_else(|| {
386
                    PtError::Internal(internal!("Created child process without stdout pipe"))
387
                })?,
388
            );
389
            // TODO RELAY #1649 We don't use a tor_memquota::mq_queue here yet
390
            let (mut tx, rx) = tor_async_utils::mpsc_channel_no_memquota(PT_STDIO_BUFFER);
391
            let ident = identifier.clone();
392
            #[allow(clippy::cognitive_complexity)]
393
            thread::spawn(move || {
394
                let reader = BufReader::new(stdout);
395
                let _stdin = stdin;
396
                let mut noted_full = false;
397
                // Forward lines from the blocking reader to the async channel.
398
                for line in reader.lines() {
399
                    let err = line.is_err();
400
                    match &line {
401
                        Ok(l) => trace!("<-- PT {}: {:?}", ident, l),
402
                        Err(e) => trace!("<-- PT {}: Error: {:?}", ident, e),
403
                    }
404
                    if let Err(e) = tx.try_send(line) {
405
                        if e.is_disconnected() {
406
                            debug!("PT {} is disconnected; shutting it down.", ident);
407
                            // Channel dropped, so shut down the pluggable transport process.
408
                            break;
409
                        }
410
                        // The other kind of error is "full", which we can't do anything about.
411
                        // Just throw the line away.
412
                        if !noted_full {
413
                            noted_full = true; // warn only once per PT.
414
                            warn!(
415
                                "Bug: Message queue for PT {} became full; dropping message",
416
                                ident
417
                            );
418
                        }
419
                    }
420
                    if err {
421
                        // Encountered an error reading, so ensure the process is shut down (it's
422
                        // probably "broken pipe" anyway, so this is slightly redundant, but the
423
                        // rest of the code assumes errors are nonrecoverable).
424
                        break;
425
                    }
426
                }
427
                // Has it already quit? If so, just exit now.
428
                if let Ok(Some(_)) = child.try_wait() {
429
                    // FIXME(eta): We currently throw away the exit code, which might be useful
430
                    //             for debugging purposes!
431
                    debug!("PT {} has exited.", ident);
432
                    return;
433
                }
434
                // Otherwise, tell it to exit.
435
                // Dropping stdin should tell the PT to exit, since we set the correct environment
436
                // variable for that to happen.
437
                trace!("Asking PT {} to exit, nicely.", ident);
438
                drop(_stdin);
439
                // Give it some time to exit.
440
                thread::sleep(GRACEFUL_EXIT_TIME);
441
                match child.try_wait() {
442
                    Ok(None) => {
443
                        // Kill it.
444
                        debug!("Sending kill signal to PT {}", ident);
445
                        if let Err(e) = child.kill() {
446
                            warn_report!(e, "Failed to kill() spawned PT {}", ident);
447
                        }
448
                    }
449
                    Ok(Some(_)) => {
450
                        debug!("PT {} shut down successfully.", ident);
451
                    } // It exited.
452
                    Err(e) => {
453
                        warn_report!(e, "Failed to call try_wait() on spawned PT {}", ident);
454
                    }
455
                }
456
            });
457
            Ok(AsyncPtChild {
458
                stdout: rx,
459
                identifier,
460
            })
461
        }
462

            
463
        /// Receive a message from the pluggable transport binary asynchronously.
464
        ///
465
        /// Note: This will convert `PtMessage::Log` into a tracing log call automatically.
466
        pub async fn recv(&mut self) -> err::Result<PtMessage> {
467
            loop {
468
                match self.stdout.next().await {
469
                    None => return Err(PtError::ChildGone),
470
                    Some(Ok(line)) => {
471
                        let line =
472
                            line.parse::<PtMessage>()
473
                                .map_err(|e| PtError::IpcParseFailed {
474
                                    line,
475
                                    error: e.into(),
476
                                })?;
477
                        if let PtMessage::Log { severity, message } = line {
478
                            // FIXME(eta): I wanted to make this integrate with `tracing` more nicely,
479
                            //             but gave up after 15 minutes of clicking through spaghetti.
480
                            match &severity as &str {
481
                                "error" => error!("[pt {}] {}", self.identifier, message),
482
                                "warning" => warn!("[pt {}] {}", self.identifier, message),
483
                                "notice" => info!("[pt {}] {}", self.identifier, message),
484
                                "info" => debug!("[pt {}] {}", self.identifier, message),
485
                                "debug" => trace!("[pt {}] {}", self.identifier, message),
486
                                x => warn!("[pt] {} {} {}", self.identifier, x, message),
487
                            }
488
                        } else {
489
                            return Ok(line);
490
                        }
491
                    }
492
                    Some(Err(e)) => {
493
                        return Err(PtError::ChildReadFailed(Arc::new(e)));
494
                    }
495
                }
496
            }
497
        }
498
    }
499

            
500
    /// Defines some helper methods that are required later on
501
    #[async_trait::async_trait]
502
    pub trait PluggableTransportPrivate {
503
        /// Return the [`AsyncPtChild`] if it exists
504
        fn inner(&mut self) -> Result<&mut AsyncPtChild, PtError>;
505

            
506
        /// Set the [`AsyncPtChild`]
507
        fn set_inner(&mut self, newval: Option<AsyncPtChild>);
508

            
509
        /// Return a loggable identifier for this transport.
510
        fn identifier(&self) -> &str;
511

            
512
        /// Checks whether a transport is specified in our specific parameters
513
        fn specific_params_contains(&self, transport: &PtTransportName) -> bool;
514

            
515
        /// Common handler for `ClientTransportLaunched` and `ServerTransportLaunched`
516
        fn common_transport_launched_handler(
517
            &self,
518
            protocol: Option<String>,
519
            transport: PtTransportName,
520
            endpoint: SocketAddr,
521
            methods: &mut HashMap<PtTransportName, PtClientMethod>,
522
        ) -> Result<(), PtError> {
523
            if !self.specific_params_contains(&transport) {
524
                return Err(PtError::ProtocolViolation(format!(
525
                    "binary launched unwanted transport '{}'",
526
                    transport
527
                )));
528
            }
529
            let protocol = match protocol {
530
                Some(protocol_str) => match &protocol_str as &str {
531
                    "socks4" => SocksVersion::V4,
532
                    "socks5" => SocksVersion::V5,
533
                    x => {
534
                        return Err(PtError::ProtocolViolation(format!(
535
                            "unknown CMETHOD protocol '{}'",
536
                            x
537
                        )))
538
                    }
539
                },
540
                None => SocksVersion::V5,
541
            };
542
            let method = PtClientMethod {
543
                kind: protocol,
544
                endpoint,
545
            };
546
            info!("Transport '{}' uses method {:?}", transport, method);
547
            methods.insert(transport, method);
548
            Ok(())
549
        }
550

            
551
        /// Attempt to launch the PT and return the corresponding `[AsyncPtChild]`
552
        fn get_child_from_pt_launch(
553
            inner: &Option<AsyncPtChild>,
554
            transports: &Vec<PtTransportName>,
555
            binary_path: &PathBuf,
556
            arguments: &[String],
557
            all_env_vars: HashMap<OsString, OsString>,
558
        ) -> Result<AsyncPtChild, PtError> {
559
            if inner.is_some() {
560
                let warning_msg =
561
                    format!("Attempted to launch PT binary for {:?} twice.", transports);
562
                warn!("{warning_msg}");
563
                // WARN: this may not be the correct error to throw here
564
                return Err(PtError::ChildProtocolViolation(warning_msg));
565
            }
566
            info!(
567
                "Launching pluggable transport at {} for {:?}",
568
                binary_path.display_lossy(),
569
                transports
570
            );
571
            let child = Command::new(binary_path)
572
                .args(arguments.iter())
573
                .envs(all_env_vars)
574
                .stdout(Stdio::piped())
575
                .stdin(Stdio::piped())
576
                .spawn()
577
                .map_err(|e| PtError::ChildSpawnFailed {
578
                    path: binary_path.clone(),
579
                    error: Arc::new(e),
580
                })?;
581

            
582
            let identifier = crate::managed::pt_identifier(binary_path)?;
583
            AsyncPtChild::new(child, identifier)
584
        }
585

            
586
        /// Consolidates some of the [`PtMessage`] potential matches to
587
        /// deduplicate code
588
        ///
589
        /// Note that getting a [`PtMessage`] from this method implies that
590
        /// the method was unable to match it and thus you should continue handling
591
        /// the message. Getting [`None`] after error handling means that a match
592
        /// was found and the appropriate action was successfully taken, and you don't
593
        /// need to worry about it.
594
        async fn try_match_common_messages<R: Runtime>(
595
            &self,
596
            rt: &R,
597
            deadline: Instant,
598
            async_child: &mut AsyncPtChild,
599
        ) -> Result<Option<PtMessage>, PtError> {
600
            match rt
601
                .timeout(
602
                    // FIXME(eta): It'd be nice if SleepProviderExt took an `Instant` natively.
603
                    deadline.saturating_duration_since(Instant::now()),
604
                    async_child.recv(),
605
                )
606
                .await
607
                .map_err(|_| PtError::Timeout)??
608
            {
609
                PtMessage::ClientTransportFailed { transport, message }
610
                | PtMessage::ServerTransportFailed { transport, message } => {
611
                    warn!(
612
                        "PT {} unable to launch {}. It said: {:?}",
613
                        async_child.identifier, transport, message
614
                    );
615
                    return Err(PtError::TransportGaveError {
616
                        transport: transport.to_string(),
617
                        message,
618
                    });
619
                }
620
                PtMessage::VersionError(e) => {
621
                    if e != "no-version" {
622
                        warn!("weird VERSION-ERROR: {}", e);
623
                    }
624
                    return Err(PtError::UnsupportedVersion);
625
                }
626
                PtMessage::Version(vers) => {
627
                    if vers != "1" {
628
                        return Err(PtError::ProtocolViolation(format!(
629
                            "stated version is {}, asked for 1",
630
                            vers
631
                        )));
632
                    }
633
                    Ok(None)
634
                }
635
                PtMessage::EnvError(e) => return Err(PtError::ChildProtocolViolation(e)),
636
                PtMessage::ProxyError(e) => return Err(PtError::ProxyError(e)),
637
                // TODO(eta): We don't do anything with these right now!
638
                PtMessage::Status(_) => Ok(None),
639
                PtMessage::Unknown(x) => {
640
                    warn!("unknown PT line: {}", x);
641
                    Ok(None)
642
                }
643
                // Return the PtMessage as it is for further processing
644
                // TODO: handle [`PtError::ProtocolViolation`] here somehow
645
                x => {
646
                    return Ok(Some(x));
647
                }
648
            }
649
        }
650
    }
651
}
652

            
653
/// Common parameters passed to a pluggable transport.
654
#[derive(PartialEq, Eq, Clone, Debug, derive_builder::Builder)]
655
pub struct PtCommonParameters {
656
    /// A path where the launched PT can store state.
657
    state_location: PathBuf,
658
    /// An IPv4 address to bind outgoing connections to (if specified).
659
    ///
660
    /// Leaving this out will mean the PT uses a sane default.
661
    #[builder(default)]
662
    outbound_bind_v4: Option<Ipv4Addr>,
663
    /// An IPv6 address to bind outgoing connections to (if specified).
664
    ///
665
    /// Leaving this out will mean the PT uses a sane default.
666
    #[builder(default)]
667
    outbound_bind_v6: Option<Ipv6Addr>,
668
    /// The maximum time we should wait for a pluggable transport binary to report successful
669
    /// initialization. If `None`, a default value is used.
670
    #[builder(default)]
671
    timeout: Option<Duration>,
672
}
673

            
674
impl PtCommonParameters {
675
    /// Return a new `PtCommonParametersBuilder` for constructing a set of parameters.
676
    pub fn builder() -> PtCommonParametersBuilder {
677
        PtCommonParametersBuilder::default()
678
    }
679

            
680
    /// Convert these parameters into a set of environment variables to be passed to the PT binary
681
    /// in accordance with the specification.
682
    fn common_environment_variables(&self) -> HashMap<OsString, OsString> {
683
        let mut ret = HashMap::new();
684
        ret.insert("TOR_PT_MANAGED_TRANSPORT_VER".into(), "1".into());
685
        ret.insert(
686
            "TOR_PT_STATE_LOCATION".into(),
687
            self.state_location.clone().into_os_string(),
688
        );
689
        ret.insert("TOR_PT_EXIT_ON_STDIN_CLOSE".into(), "1".into());
690
        if let Some(v4) = self.outbound_bind_v4 {
691
            ret.insert(
692
                "TOR_PT_OUTBOUND_BIND_ADDRESS_V4".into(),
693
                v4.to_string().into(),
694
            );
695
        }
696
        if let Some(v6) = self.outbound_bind_v6 {
697
            // pt-spec.txt: "IPv6 addresses MUST always be wrapped in square brackets."
698
            ret.insert(
699
                "TOR_PT_OUTBOUND_BIND_ADDRESS_V6".into(),
700
                format!("[{}]", v6).into(),
701
            );
702
        }
703
        ret
704
    }
705
}
706

            
707
/// Parameters passed only to a pluggable transport client.
708
#[derive(PartialEq, Eq, Clone, Debug, derive_builder::Builder)]
709
pub struct PtClientParameters {
710
    /// A SOCKS URI specifying a proxy to use.
711
    #[builder(default)]
712
    proxy_uri: Option<String>,
713
    /// A list of transports to initialise.
714
    ///
715
    /// The PT launch will fail if all transports are not successfully initialised.
716
    transports: Vec<PtTransportName>,
717
}
718

            
719
impl PtClientParameters {
720
    /// Return a new `PtClientParametersBuilder` for constructing a set of parameters.
721
    pub fn builder() -> PtClientParametersBuilder {
722
        PtClientParametersBuilder::default()
723
    }
724

            
725
    /// Convert these parameters into a set of environment variables to be passed to the PT binary
726
    /// in accordance with the specification.
727
    fn environment_variables(
728
        &self,
729
        common_params: &PtCommonParameters,
730
    ) -> HashMap<OsString, OsString> {
731
        let mut ret = common_params.common_environment_variables();
732
        if let Some(ref proxy_uri) = self.proxy_uri {
733
            ret.insert("TOR_PT_PROXY".into(), proxy_uri.clone().into());
734
        }
735
        ret.insert(
736
            "TOR_PT_CLIENT_TRANSPORTS".into(),
737
            self.transports.iter().join(",").into(),
738
        );
739
        ret
740
    }
741
}
742

            
743
/// Parameters passed only to a pluggable transport server.
744
#[derive(PartialEq, Eq, Clone, Debug, derive_builder::Builder)]
745
pub struct PtServerParameters {
746
    /// A list of transports to initialise.
747
    ///
748
    /// The PT launch will fail if all transports are not successfully initialised.
749
    transports: Vec<PtTransportName>,
750
    /// Transport options for each server transport
751
    #[builder(default)]
752
    server_transport_options: String,
753
    /// Set host:port on which the server transport should listen for connections
754
    #[builder(default)]
755
    server_bindaddr: String,
756
    /// Set host:port on which the server transport should forward requests
757
    #[builder(default)]
758
    server_orport: Option<String>,
759
    /// Set host:port on which the server transport should forward requests (extended ORPORT)
760
    #[builder(default)]
761
    server_extended_orport: Option<String>,
762
}
763

            
764
impl PtServerParameters {
765
    /// Return a new `PtServerParametersBuilder` for constructing a set of parameters.
766
    pub fn builder() -> PtServerParametersBuilder {
767
        PtServerParametersBuilder::default()
768
    }
769

            
770
    /// Convert these parameters into a set of environment variables to be passed to the PT binary
771
    /// in accordance with the specification.
772
    fn environment_variables(
773
        &self,
774
        common_params: &PtCommonParameters,
775
    ) -> HashMap<OsString, OsString> {
776
        let mut ret = common_params.common_environment_variables();
777
        ret.insert(
778
            "TOR_PT_SERVER_TRANSPORTS".into(),
779
            self.transports.iter().join(",").into(),
780
        );
781
        ret.insert(
782
            "TOR_PT_SERVER_TRANSPORT_OPTIONS".into(),
783
            self.server_transport_options.clone().into(),
784
        );
785
        ret.insert(
786
            "TOR_PT_SERVER_BINDADDR".into(),
787
            self.server_bindaddr.clone().into(),
788
        );
789
        if let Some(ref server_orport) = self.server_orport {
790
            ret.insert("TOR_PT_ORPORT".into(), server_orport.into());
791
        }
792
        if let Some(ref server_extended_orport) = self.server_extended_orport {
793
            ret.insert(
794
                "TOR_PT_EXTENDED_SERVER_PORT".into(),
795
                server_extended_orport.into(),
796
            );
797
        }
798
        ret
799
    }
800
}
801

            
802
/// Common functionality implemented to allow code reuse
803
#[async_trait::async_trait]
804
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
805
pub trait PluggableTransport: PluggableTransportPrivate {
806
    /// Get all client methods returned by the binary, if it has been launched.
807
    ///
808
    /// If it hasn't been launched, the returned map will be empty.
809
    // TODO(eta): Actually figure out a way to expose this more stably.
810
    fn transport_methods(&self) -> &HashMap<PtTransportName, PtClientMethod>;
811

            
812
    /// Get the next [`PtMessage`] from the running transport. It is recommended to call this
813
    /// in a loop once a PT has been launched, in order to forward log messages and find out about
814
    /// status updates.
815
    //
816
    // FIXME(eta): This API will probably go away and get replaced with something better.
817
    //             In particular, we'd want to cache `Status` messages from before this method
818
    //             was called.
819
    async fn next_message(&mut self) -> err::Result<PtMessage> {
820
        let inner = self.inner()?;
821
        let ret = inner.recv().await;
822
        if let Err(PtError::ChildGone) | Err(PtError::ChildReadFailed { .. }) = &ret {
823
            // FIXME(eta): Currently this lets the caller still think the methods work by calling
824
            //             transport_methods.
825
            debug!(
826
                "PT {}: Received {:?}; shutting down.",
827
                self.identifier(),
828
                ret
829
            );
830
            self.set_inner(None);
831
        }
832
        ret
833
    }
834
}
835
/// A pluggable transport binary in a child process.
836
///
837
/// These start out inert, and must be launched with [`PluggableClientTransport::launch`] in order
838
/// to be useful.
839
#[derive(Debug)]
840
pub struct PluggableClientTransport {
841
    /// The currently running child, if there is one.
842
    inner: Option<AsyncPtChild>,
843
    /// The path to the binary to run.
844
    pub(crate) binary_path: PathBuf,
845
    /// Arguments to pass to the binary.
846
    arguments: Vec<String>,
847
    /// Configured parameters.
848
    common_params: PtCommonParameters,
849
    /// Configured client-only parameters.
850
    client_params: PtClientParameters,
851
    /// Information about client methods obtained from the PT.
852
    cmethods: HashMap<PtTransportName, PtClientMethod>,
853
}
854

            
855
impl PluggableTransport for PluggableClientTransport {
856
    fn transport_methods(&self) -> &HashMap<PtTransportName, PtClientMethod> {
857
        &self.cmethods
858
    }
859
}
860

            
861
impl PluggableTransportPrivate for PluggableClientTransport {
862
    fn inner(&mut self) -> Result<&mut AsyncPtChild, PtError> {
863
        self.inner.as_mut().ok_or(PtError::ChildGone)
864
    }
865
    fn set_inner(&mut self, newval: Option<AsyncPtChild>) {
866
        self.inner = newval;
867
    }
868
    fn identifier(&self) -> &str {
869
        match &self.inner {
870
            Some(child) => &child.identifier,
871
            None => "<not yet launched>",
872
        }
873
    }
874
    fn specific_params_contains(&self, transport: &PtTransportName) -> bool {
875
        self.client_params.transports.contains(transport)
876
    }
877
}
878

            
879
impl PluggableClientTransport {
880
    /// Create a new pluggable transport wrapper, wrapping the binary at `binary_path` and passing
881
    /// the `params` to it.
882
    ///
883
    /// You must call [`PluggableClientTransport::launch`] to actually run the PT.
884
    pub fn new(
885
        binary_path: PathBuf,
886
        arguments: Vec<String>,
887
        common_params: PtCommonParameters,
888
        client_params: PtClientParameters,
889
    ) -> Self {
890
        Self {
891
            common_params,
892
            client_params,
893
            arguments,
894
            binary_path,
895
            inner: None,
896
            cmethods: Default::default(),
897
        }
898
    }
899

            
900
    /// Launch the pluggable transport, executing the binary.
901
    ///
902
    /// Will return an error if the launch fails, one of the transports fail, not all transports
903
    /// were launched, or the launch times out.
904
    pub async fn launch<R: Runtime>(&mut self, rt: R) -> err::Result<()> {
905
        let all_env_vars = self
906
            .client_params
907
            .environment_variables(&self.common_params);
908

            
909
        let mut async_child =
910
            <PluggableClientTransport as PluggableTransportPrivate>::get_child_from_pt_launch(
911
                &self.inner,
912
                &self.client_params.transports,
913
                &self.binary_path,
914
                &self.arguments,
915
                all_env_vars,
916
            )?;
917

            
918
        let deadline = Instant::now() + self.common_params.timeout.unwrap_or(PT_START_TIMEOUT);
919
        let mut cmethods = HashMap::new();
920
        let mut proxy_done = self.client_params.proxy_uri.is_none();
921

            
922
        loop {
923
            match self
924
                .try_match_common_messages(&rt, deadline, &mut async_child)
925
                .await
926
            {
927
                Ok(maybe_message) => {
928
                    if let Some(message) = maybe_message {
929
                        match message {
930
                            PtMessage::ClientTransportLaunched {
931
                                transport,
932
                                protocol,
933
                                endpoint,
934
                            } => {
935
                                self.common_transport_launched_handler(
936
                                    Some(protocol),
937
                                    transport,
938
                                    endpoint,
939
                                    &mut cmethods,
940
                                )?;
941
                            }
942
                            PtMessage::ProxyDone => {
943
                                if proxy_done {
944
                                    return Err(PtError::ProtocolViolation(
945
                                        "binary initiated proxy when not asked (or twice)".into(),
946
                                    ));
947
                                }
948
                                info!("PT binary now proxying connections via supplied URI");
949
                                proxy_done = true;
950
                            }
951
                            // TODO: unify most of the handling of ClientTransportsDone with ServerTransportsDone
952
                            PtMessage::ClientTransportsDone => {
953
                                let unsupported = self
954
                                    .client_params
955
                                    .transports
956
                                    .iter()
957
                                    .filter(|&x| !cmethods.contains_key(x))
958
                                    .map(|x| x.to_string())
959
                                    .collect::<Vec<_>>();
960
                                if !unsupported.is_empty() {
961
                                    warn!(
962
                                        "PT binary failed to initialise transports: {:?}",
963
                                        unsupported
964
                                    );
965
                                    return Err(PtError::ClientTransportsUnsupported(unsupported));
966
                                }
967
                                info!("PT binary initialisation done");
968
                                break;
969
                            }
970
                            x => {
971
                                return Err(PtError::ProtocolViolation(format!(
972
                                    "received unexpected {:?}",
973
                                    x
974
                                )));
975
                            }
976
                        }
977
                    }
978
                }
979
                Err(e) => return Err(e),
980
            }
981
        }
982
        self.cmethods = cmethods;
983
        self.inner = Some(async_child);
984
        // TODO(eta): We need to expose the log and status messages after this function exits!
985
        Ok(())
986
    }
987
}
988

            
989
/// A pluggable transport server binary in a child process.
990
///
991
/// These start out inert, and must be launched with [`PluggableServerTransport::launch`] in order
992
/// to be useful.
993
#[derive(Debug)]
994
pub struct PluggableServerTransport {
995
    /// The currently running child, if there is one.
996
    inner: Option<AsyncPtChild>,
997
    /// The path to the binary to run.
998
    pub(crate) binary_path: PathBuf,
999
    /// 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 {
    fn inner(&mut self) -> Result<&mut AsyncPtChild, PtError> {
        self.inner.as_mut().ok_or(PtError::ChildGone)
    }
    fn set_inner(&mut self, newval: Option<AsyncPtChild>) {
        self.inner = newval;
    }
    fn identifier(&self) -> &str {
        match &self.inner {
            Some(child) => &child.identifier,
            None => "<not yet launched>",
        }
    }
    fn specific_params_contains(&self, transport: &PtTransportName) -> bool {
        self.server_params.transports.contains(transport)
    }
}
impl PluggableTransport for PluggableServerTransport {
    fn transport_methods(&self) -> &HashMap<PtTransportName, PtClientMethod> {
        &self.smethods
    }
}
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.
    pub fn new(
        binary_path: PathBuf,
        arguments: Vec<String>,
        common_params: PtCommonParameters,
        server_params: PtServerParameters,
    ) -> Self {
        Self {
            common_params,
            server_params,
            arguments,
            binary_path,
            inner: None,
            smethods: Default::default(),
        }
    }
    /// 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.
    pub async fn launch<R: Runtime>(&mut self, rt: R) -> err::Result<()> {
        let all_env_vars = self
            .server_params
            .environment_variables(&self.common_params);
        let mut async_child =
            <PluggableServerTransport as PluggableTransportPrivate>::get_child_from_pt_launch(
                &self.inner,
                &self.server_params.transports,
                &self.binary_path,
                &self.arguments,
                all_env_vars,
            )?;
        let deadline = Instant::now() + self.common_params.timeout.unwrap_or(PT_START_TIMEOUT);
        let mut smethods = HashMap::new();
        loop {
            match self
                .try_match_common_messages(&rt, deadline, &mut async_child)
                .await
            {
                Ok(maybe_message) => {
                    if let Some(message) = maybe_message {
                        match message {
                            PtMessage::ServerTransportLaunched {
                                transport,
                                endpoint,
                                options: _,
                            } => {
                                self.common_transport_launched_handler(
                                    None,
                                    transport,
                                    endpoint,
                                    &mut smethods,
                                )?;
                            }
                            PtMessage::ServerTransportsDone => {
                                let unsupported = self
                                    .server_params
                                    .transports
                                    .iter()
                                    .filter(|&x| !smethods.contains_key(x))
                                    .map(|x| x.to_string())
                                    .collect::<Vec<_>>();
                                if !unsupported.is_empty() {
                                    warn!(
                                        "PT binary failed to initialise transports: {:?}",
                                        unsupported
                                    );
                                    return Err(PtError::ClientTransportsUnsupported(unsupported));
                                }
                                info!("PT binary initialisation done");
                                break;
                            }
                            x => {
                                return Err(PtError::ProtocolViolation(format!(
                                    "received unexpected {:?}",
                                    x
                                )));
                            }
                        }
                    }
                }
                Err(e) => return Err(e),
            }
        }
        self.smethods = smethods;
        self.inner = Some(async_child);
        // TODO(eta): We need to expose the log and status messages after this function exits!
        Ok(())
    }
}
#[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
            }))
        );
    }
}