microsandbox_network/proxy/
http_connect.rs

1//! HTTP CONNECT outbound proxy builders and transport.
2
3use std::fmt;
4use std::net::AddrParseError;
5#[cfg(feature = "engine")]
6use std::{io, net::SocketAddr, time::Duration};
7
8#[cfg(feature = "engine")]
9use base64::Engine as _;
10use microsandbox_types::SecretSource;
11use serde::{Deserialize, Serialize};
12#[cfg(feature = "engine")]
13use tokio::{
14    io::{AsyncReadExt, AsyncWriteExt},
15    net::TcpStream,
16    time::timeout,
17};
18use zeroize::Zeroizing;
19
20use super::types::{
21    OutboundProxy, OutboundProxyBuildError, OutboundProxyConfig, OutboundProxyProtocol,
22};
23
24//--------------------------------------------------------------------------------------------------
25// Constants
26//--------------------------------------------------------------------------------------------------
27
28#[cfg(feature = "engine")]
29const CONNECT_RESPONSE_HEADER_LIMIT: usize = 8192;
30#[cfg(feature = "engine")]
31const CONNECT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);
32
33//--------------------------------------------------------------------------------------------------
34// Types
35//--------------------------------------------------------------------------------------------------
36
37/// Environment-backed username/password credentials for an HTTP CONNECT proxy.
38#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
39pub struct HttpConnectCredentials {
40    username: String,
41    password: SecretSource,
42}
43
44/// Resolved HTTP CONNECT credentials held only by the network runtime.
45#[doc(hidden)]
46#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
47pub struct ResolvedHttpConnectCredentials {
48    username: String,
49    password: Zeroizing<String>,
50}
51
52/// Builds an HTTP CONNECT outbound proxy.
53#[derive(Debug, Clone)]
54pub struct HttpConnectProxyBuilder {
55    address: String,
56    credentials: Option<HttpConnectCredentials>,
57}
58
59/// HTTP CONNECT wire protocol operations.
60#[cfg(feature = "engine")]
61pub(super) struct HttpConnectProtocol;
62
63//--------------------------------------------------------------------------------------------------
64// Methods
65//--------------------------------------------------------------------------------------------------
66
67impl HttpConnectProxyBuilder {
68    /// Creates a builder for the configured proxy address.
69    pub(super) fn new(address: impl Into<String>) -> Self {
70        Self {
71            address: address.into(),
72            credentials: None,
73        }
74    }
75
76    /// Sets username authentication and a host-side password source.
77    ///
78    /// The username may be empty, for proxies that authenticate with a token
79    /// alone, but must not contain `:` or control characters. The password
80    /// source must resolve to a non-empty value at launch.
81    pub fn credentials(mut self, username: impl Into<String>, password: SecretSource) -> Self {
82        self.credentials = Some(HttpConnectCredentials {
83            username: username.into(),
84            password,
85        });
86        self
87    }
88}
89
90impl HttpConnectCredentials {
91    /// Returns the durable authentication username.
92    pub(crate) fn username(&self) -> &str {
93        &self.username
94    }
95
96    /// Returns the host-side password source.
97    pub(crate) fn password_source(&self) -> &SecretSource {
98        &self.password
99    }
100
101    /// Validates the durable username and the host-side password source.
102    pub(super) fn validate(&self) -> Result<(), OutboundProxyBuildError> {
103        let reason = if let Err(reason) = validate_username(&self.username) {
104            reason
105        } else {
106            match &self.password {
107                SecretSource::Env { var } if var.is_empty() => {
108                    "password environment variable must not be empty"
109                }
110                SecretSource::Env { .. } => return Ok(()),
111                SecretSource::Store { .. } => "store-backed password sources are not supported yet",
112            }
113        };
114        Err(OutboundProxyBuildError::InvalidHttpConnectCredentials { reason })
115    }
116}
117
118impl ResolvedHttpConnectCredentials {
119    /// Creates credentials for the private launch contract.
120    #[doc(hidden)]
121    pub fn new(username: impl Into<String>, password: impl Into<String>) -> Self {
122        Self {
123            username: username.into(),
124            password: Zeroizing::new(password.into()),
125        }
126    }
127
128    /// Returns the resolved authentication username.
129    pub(super) fn username(&self) -> &str {
130        &self.username
131    }
132
133    /// Validates the resolved username and password.
134    pub(super) fn validate(&self) -> Result<(), OutboundProxyBuildError> {
135        let reason = if let Err(reason) = validate_username(&self.username) {
136            reason
137        } else if self.password.is_empty() {
138            "password must not be empty"
139        } else {
140            return Ok(());
141        };
142        Err(OutboundProxyBuildError::InvalidHttpConnectCredentials { reason })
143    }
144}
145
146#[cfg(feature = "engine")]
147impl HttpConnectProtocol {
148    /// Opens a TCP tunnel through the configured HTTP proxy.
149    pub(super) async fn connect(
150        address: SocketAddr,
151        destination: SocketAddr,
152        credentials: Option<&ResolvedHttpConnectCredentials>,
153    ) -> io::Result<TcpStream> {
154        let mut stream = TcpStream::connect(address).await?;
155        let authority = destination.to_string();
156        let request = Self::build_connect_request(&authority, credentials);
157
158        stream.write_all(request.as_bytes()).await?;
159        stream.flush().await?;
160
161        let status = timeout(
162            CONNECT_RESPONSE_TIMEOUT,
163            Self::read_connect_response_status(&mut stream),
164        )
165        .await
166        .map_err(|_| {
167            io::Error::new(
168                io::ErrorKind::TimedOut,
169                "HTTP CONNECT proxy did not complete its response headers within 10 seconds",
170            )
171        })??;
172
173        if (200..300).contains(&status) {
174            return Ok(stream);
175        }
176
177        if status == 407 {
178            let message = if credentials.is_some() {
179                "HTTP CONNECT proxy rejected the supplied credentials (status 407)"
180            } else {
181                "HTTP CONNECT proxy requires authentication (status 407)"
182            };
183            return Err(io::Error::new(io::ErrorKind::PermissionDenied, message));
184        }
185
186        Err(io::Error::new(
187            io::ErrorKind::ConnectionRefused,
188            format!("HTTP CONNECT proxy rejected the tunnel with status {status}"),
189        ))
190    }
191
192    /// Builds the CONNECT request.
193    ///
194    /// Exactly one `Proxy-Authorization: Basic` header is added when credentials
195    /// are configured. The password is base64-encoded (standard alphabet, UTF-8
196    /// bytes) into the header, and the whole request is zeroized on drop so the
197    /// token does not linger in host memory.
198    ///
199    /// The request is sized exactly and written in a single pass. Growing the
200    /// buffer would reallocate and free a block that already holds the token, and
201    /// `Zeroizing` only wipes the final allocation.
202    fn build_connect_request(
203        authority: &str,
204        credentials: Option<&ResolvedHttpConnectCredentials>,
205    ) -> Zeroizing<String> {
206        let encoded = credentials.map(|credentials| {
207            let userinfo = Zeroizing::new(format!(
208                "{}:{}",
209                credentials.username,
210                credentials.password.as_str()
211            ));
212            Zeroizing::new(base64::engine::general_purpose::STANDARD.encode(userinfo.as_bytes()))
213        });
214        let token = encoded.as_ref().map(|encoded| encoded.as_str());
215
216        let mut capacity = "CONNECT ".len()
217            + authority.len()
218            + " HTTP/1.1\r\nHost: ".len()
219            + authority.len()
220            + "\r\n".len();
221        if let Some(token) = token {
222            capacity += "Proxy-Authorization: Basic ".len() + token.len() + "\r\n".len();
223        }
224        capacity += "\r\n".len();
225
226        let mut request = Zeroizing::new(String::with_capacity(capacity));
227        request.push_str("CONNECT ");
228        request.push_str(authority);
229        request.push_str(" HTTP/1.1\r\nHost: ");
230        request.push_str(authority);
231        request.push_str("\r\n");
232        if let Some(token) = token {
233            request.push_str("Proxy-Authorization: Basic ");
234            request.push_str(token);
235            request.push_str("\r\n");
236        }
237        request.push_str("\r\n");
238        request
239    }
240
241    async fn read_connect_response_status(stream: &mut TcpStream) -> io::Result<u16> {
242        let mut total_header_bytes = 0;
243
244        loop {
245            let mut header = Vec::with_capacity(256);
246            loop {
247                if total_header_bytes >= CONNECT_RESPONSE_HEADER_LIMIT {
248                    return Err(io::Error::new(
249                        io::ErrorKind::InvalidData,
250                        "HTTP CONNECT proxy response headers exceeded 8192 bytes",
251                    ));
252                }
253
254                header.push(stream.read_u8().await?);
255                total_header_bytes += 1;
256                if header.ends_with(b"\r\n\r\n") {
257                    break;
258                }
259            }
260
261            let status_end = header
262                .windows(2)
263                .position(|window| window == b"\r\n")
264                .ok_or_else(|| {
265                    io::Error::new(
266                        io::ErrorKind::InvalidData,
267                        "HTTP CONNECT proxy returned an invalid status line",
268                    )
269                })?;
270            let status = Self::parse_status_line(&header[..status_end])?;
271
272            if (100..200).contains(&status) && status != 101 {
273                continue;
274            }
275
276            return Ok(status);
277        }
278    }
279
280    /// Parses an HTTP/1.x status line and returns its status code.
281    ///
282    /// Fields are split on ASCII whitespace, so leading whitespace is tolerated
283    /// and one or more SP/HTAB separators are accepted. The version token must be
284    /// exactly `HTTP/1.0` or `HTTP/1.1` and the status code must be exactly three
285    /// ASCII digits; a trailing separator after the code is optional. The reason
286    /// phrase is ignored and may contain spaces, tabs, and valid-UTF-8 non-ASCII
287    /// text, but every control byte other than HTAB is rejected anywhere in the
288    /// line and the line must be valid UTF-8.
289    fn parse_status_line(status_line: &[u8]) -> io::Result<u16> {
290        if status_line
291            .iter()
292            .any(|b| b.is_ascii_control() && *b != b'\t')
293        {
294            return Err(io::Error::new(
295                io::ErrorKind::InvalidData,
296                "HTTP CONNECT proxy returned a status line containing control characters",
297            ));
298        }
299        let status_line = std::str::from_utf8(status_line).map_err(|_| {
300            io::Error::new(
301                io::ErrorKind::InvalidData,
302                "HTTP CONNECT proxy returned a non-ASCII status line",
303            )
304        })?;
305        let mut fields = status_line.split_ascii_whitespace();
306        let version = fields.next().unwrap_or_default();
307        if version != "HTTP/1.0" && version != "HTTP/1.1" {
308            return Err(io::Error::new(
309                io::ErrorKind::InvalidData,
310                "HTTP CONNECT proxy returned an invalid HTTP status line",
311            ));
312        }
313        let status = fields.next().ok_or_else(|| {
314            io::Error::new(
315                io::ErrorKind::InvalidData,
316                "HTTP CONNECT proxy returned a status line without a status code",
317            )
318        })?;
319        if status.len() != 3 {
320            return Err(io::Error::new(
321                io::ErrorKind::InvalidData,
322                "HTTP CONNECT proxy returned a status code that is not three digits",
323            ));
324        }
325        if !status.bytes().all(|byte| byte.is_ascii_digit()) {
326            return Err(io::Error::new(
327                io::ErrorKind::InvalidData,
328                "HTTP CONNECT proxy returned a status code containing non-digit characters",
329            ));
330        }
331        status.parse::<u16>().map_err(|_| {
332            io::Error::new(
333                io::ErrorKind::InvalidData,
334                "HTTP CONNECT proxy returned an invalid status code",
335            )
336        })
337    }
338}
339
340//--------------------------------------------------------------------------------------------------
341// Trait Implementations
342//--------------------------------------------------------------------------------------------------
343
344impl OutboundProxyConfig for HttpConnectProxyBuilder {
345    fn build(self) -> Result<OutboundProxy, OutboundProxyBuildError> {
346        let address = self.address.parse().map_err(|source: AddrParseError| {
347            OutboundProxyBuildError::InvalidAddress {
348                protocol: OutboundProxyProtocol::HttpConnect,
349                address: self.address,
350                source,
351            }
352        })?;
353
354        if let Some(credentials) = &self.credentials {
355            credentials.validate()?;
356        }
357
358        Ok(OutboundProxy::HttpConnect {
359            address,
360            credentials: self.credentials,
361        })
362    }
363}
364
365impl fmt::Debug for ResolvedHttpConnectCredentials {
366    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
367        f.debug_struct("ResolvedHttpConnectCredentials")
368            .field("username", &self.username)
369            .field("password", &"[REDACTED]")
370            .finish()
371    }
372}
373
374//--------------------------------------------------------------------------------------------------
375// Functions: Helpers
376//--------------------------------------------------------------------------------------------------
377
378/// Validates a Basic authentication username shared by the durable and resolved
379/// credentials.
380///
381/// RFC 7617 permits an empty user-id, which some proxies use for token-only
382/// authentication, but forbids `:` and control characters.
383fn validate_username(username: &str) -> Result<(), &'static str> {
384    if username.contains(':') {
385        Err("username must not contain ':'")
386    } else if username.chars().any(char::is_control) {
387        Err("username must not contain control characters")
388    } else {
389        Ok(())
390    }
391}
392
393//--------------------------------------------------------------------------------------------------
394// Tests
395//--------------------------------------------------------------------------------------------------
396
397#[cfg(all(test, feature = "engine"))]
398mod tests {
399    use std::{future::Future, io, net::SocketAddr, time::Duration};
400
401    use base64::Engine as _;
402    use microsandbox_types::SecretSource;
403    use tokio::{
404        io::{AsyncReadExt, AsyncWriteExt},
405        net::{TcpListener, TcpStream},
406        sync::oneshot,
407        task::JoinHandle,
408        time::{Instant, timeout},
409    };
410
411    use super::{
412        CONNECT_RESPONSE_HEADER_LIMIT, CONNECT_RESPONSE_TIMEOUT, HttpConnectProtocol,
413        ResolvedHttpConnectCredentials,
414    };
415    use crate::proxy::{
416        OutboundProxy, OutboundProxyBuildError, OutboundProxyBuilder, OutboundProxyConfig,
417        ResolvedOutboundProxy,
418    };
419
420    const TEST_TIMEOUT: Duration = Duration::from_secs(5);
421
422    // Exact rejection diagnostics, one per rejection class.
423    const CONTROL: &str = "HTTP CONNECT proxy returned a status line containing control characters";
424    const NON_ASCII: &str = "HTTP CONNECT proxy returned a non-ASCII status line";
425    const INVALID_LINE: &str = "HTTP CONNECT proxy returned an invalid HTTP status line";
426    const NO_CODE: &str = "HTTP CONNECT proxy returned a status line without a status code";
427    const CODE_LENGTH: &str = "HTTP CONNECT proxy returned a status code that is not three digits";
428    const CODE_DIGITS: &str =
429        "HTTP CONNECT proxy returned a status code containing non-digit characters";
430
431    #[tokio::test]
432    async fn connect_rejects_malformed_status_lines() {
433        let cases: &[(&str, &[u8])] = &[
434            ("signed", b"HTTP/1.1 +200 OK"),
435            ("signed two-digit", b"HTTP/1.1 +20 OK"),
436            ("signed two-digit 99", b"HTTP/1.1 +99 OK"),
437            ("padded", b"HTTP/1.1 0200 OK"),
438            ("short", b"HTTP/1.1 20 OK"),
439            ("long", b"HTTP/1.1 2000 OK"),
440            ("FF", b"HTTP/1.1\x0c200 OK"),
441            ("VT", b"HTTP/1.1\x0b200 OK"),
442            ("bare CR", b"HTTP/1.1\r200 OK"),
443            ("NUL", b"HTTP/1.1 200 O\0K"),
444            ("DEL", b"HTTP/1.1 200 O\x7fK"),
445            ("bare LF", b"HTTP/1.1 200 O\nK"),
446            ("trailing CR", b"HTTP/1.1 200 OK\r"),
447            ("HTTP/2", b"HTTP/2 200 OK"),
448            ("HTTP/1.2", b"HTTP/1.2 200 OK"),
449            ("lowercase", b"http/1.1 200 OK"),
450            ("letter code", b"HTTP/1.1 2O0 OK"),
451            ("joined reason", b"HTTP/1.1 200OK"),
452            ("missing code", b"HTTP/1.1 OK"),
453            ("NBSP", b"HTTP/1.1\xa0200 OK"),
454            ("non-ASCII code", b"HTTP/1.1 2\xc30 OK"),
455        ];
456        // Collect failures so every regression is exercised in one run.
457        let mut failures = Vec::new();
458
459        for (label, line) in cases {
460            let response = [*line, b"\r\n\r\n"].concat();
461            let (result, request) = exchange(response, destination()).await;
462            assert_eq!(
463                request,
464                expected_request(destination()),
465                "{label}: {line:?}"
466            );
467            if !matches!(&result, Err(error) if error.kind() == io::ErrorKind::InvalidData) {
468                failures.push(format!("{label}: {line:?}: {result:?}"));
469            }
470        }
471
472        assert!(failures.is_empty(), "{}", failures.join("\n"));
473
474        // Pin one exact diagnostic per rejection class on the wire.
475        for (label, line, message) in [
476            ("control byte", &b"HTTP/1.1 200 O\0K"[..], CONTROL),
477            ("bad version", b"HTTP/2 200 OK", INVALID_LINE),
478            ("missing code", b"HTTP/1.1", NO_CODE),
479            ("wrong length", b"HTTP/1.1 +200 OK", CODE_LENGTH),
480            ("non-digit", b"HTTP/1.1 2O0 OK", CODE_DIGITS),
481            ("non-ASCII status line", b"HTTP/1.1 200 \xff", NON_ASCII),
482        ] {
483            let response = [line, b"\r\n\r\n"].concat();
484            let (result, _) = exchange(response, destination()).await;
485            let error = result.expect_err(label);
486            assert_eq!(error.kind(), io::ErrorKind::InvalidData, "{label}");
487            assert_eq!(error.to_string(), message, "{label}");
488        }
489    }
490
491    #[tokio::test]
492    async fn connect_accepts_sp_htab_separator_runs() {
493        for line in [
494            &b"HTTP/1.1\t200 OK"[..],
495            b"HTTP/1.1  200 OK",
496            b"HTTP/1.1 \t 200\t\tOK",
497            b"HTTP/1.1 204\t",
498            b"HTTP/1.0 200 \tleading tab reason",
499        ] {
500            assert_success([line, b"\r\n\r\n"].concat(), &format!("{line:?}"), b"").await;
501        }
502    }
503
504    #[tokio::test]
505    async fn connect_accepts_utf8_non_ascii_reason_phrase() {
506        for line in [
507            "HTTP/1.1 200 café".as_bytes(),
508            b"HTTP/1.1 200 \xc2\x85obs",
509            "HTTP/1.0 201 résumé".as_bytes(),
510        ] {
511            let tail = b"\0\xfftail";
512            assert_success(
513                [line, b"\r\n\r\n", tail].concat(),
514                &format!("{line:?}"),
515                tail,
516            )
517            .await;
518        }
519    }
520
521    #[tokio::test]
522    async fn connect_rejects_invalid_utf8_reason_phrase() {
523        for (label, line) in [
524            ("obs-text FF/80", &b"HTTP/1.1 200 \xff\x80"[..]),
525            ("Latin-1 e9", b"HTTP/1.0 201 \xe9t\xe9"),
526        ] {
527            let response = [line, b"\r\n\r\n"].concat();
528            let (result, request) = exchange(response, destination()).await;
529            assert_eq!(request, expected_request(destination()), "{label}");
530            let error = result.expect_err(label);
531            assert_eq!(error.kind(), io::ErrorKind::InvalidData, "{label}");
532            assert_eq!(error.to_string(), NON_ASCII, "{label}: {line:?}");
533        }
534    }
535
536    #[tokio::test]
537    async fn connect_waits_for_final_response_after_interim_headers() {
538        let tail = b"\0\xffinterim tail";
539        let cases: &[(&str, &[u8], &str, u16)] = &[
540            (
541                "100 then 200",
542                b"HTTP/1.1 100 Continue\r\n\r\n",
543                "HTTP/1.1",
544                200,
545            ),
546            (
547                "103 then 201",
548                b"HTTP/1.1 103 Early Hints\r\n\r\n",
549                "HTTP/1.1",
550                201,
551            ),
552            (
553                "100 then 103 then 299",
554                b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 103 Early Hints\r\n\r\n",
555                "HTTP/1.0",
556                299,
557            ),
558            (
559                "HTAB interim",
560                b"HTTP/1.1\t103\tEarly Hints\r\n\r\n",
561                "HTTP/1.1",
562                200,
563            ),
564        ];
565
566        for (label, interim, version, status) in cases {
567            let final_header = format!("{version} {status} OK\r\n\r\n");
568            assert_success(
569                [*interim, final_header.as_bytes(), tail].concat(),
570                &format!("{label}: {interim:?}"),
571                tail,
572            )
573            .await;
574        }
575        let prefix = b"HTTP/1.1 103 Early Hints\r\n\r\n";
576        let error = assert_error(
577            [prefix.as_slice(), b"HTTP/1.1 503 Unavailable\r\n\r\n"].concat(),
578            "103 then 503",
579            io::ErrorKind::ConnectionRefused,
580        )
581        .await;
582        assert_eq!(
583            error.to_string(),
584            "HTTP CONNECT proxy rejected the tunnel with status 503",
585            "103 then 503"
586        );
587        assert_error(
588            [prefix.as_slice(), b"HTTP/1.1 +200 OK\r\n\r\n"].concat(),
589            "malformed final after 103",
590            io::ErrorKind::InvalidData,
591        )
592        .await;
593        assert_error(
594            b"HTTP/1.1 +100 Continue\r\n\r\nHTTP/1.1 200 OK\r\n\r\n".to_vec(),
595            "malformed interim",
596            io::ErrorKind::InvalidData,
597        )
598        .await;
599    }
600
601    #[test]
602    fn status_line_accepts_rfc_aligned_framing() {
603        for version in ["HTTP/1.0", "HTTP/1.1"] {
604            for status in [0, 100, 101, 103, 199, 200, 201, 204, 299, 407, 503, 999] {
605                for separator in [" ", "\t", "  ", " \t "] {
606                    for reason in [
607                        &b"OK"[..],
608                        b"Connection Established",
609                        b"",
610                        b"OK ",
611                        b"A\tB",
612                        "café".as_bytes(),
613                        b"\xc2\x85",
614                    ] {
615                        let mut line =
616                            format!("{version}{separator}{status:03}{separator}").into_bytes();
617                        line.extend_from_slice(reason);
618                        assert_eq!(
619                            HttpConnectProtocol::parse_status_line(&line)
620                                .map_err(|error| error.kind()),
621                            Ok(status),
622                            "{line:?}"
623                        );
624                    }
625                }
626            }
627        }
628    }
629
630    #[test]
631    fn status_line_accepts_mixed_separator_runs() {
632        // A different SP/HTAB run in each separator position must be accepted;
633        // these also fail if the second separator is required to match the first.
634        for (line, status) in [
635            (&b"HTTP/1.1 \t 407  Proxy Auth "[..], 407),
636            (b"HTTP/1.1\t407  Proxy Auth ", 407),
637            (b"HTTP/1.1 \t200 OK", 200),
638            (b"HTTP/1.1\t\t200  OK", 200),
639            (b"HTTP/1.0 204\t", 204),
640            // No trailing separator after the code is required.
641            (b"HTTP/1.1 200", 200),
642            (b"HTTP/1.0 204", 204),
643            // Leading whitespace is tolerated by ASCII-whitespace splitting.
644            (b" HTTP/1.1 200 OK", 200),
645            (b"\tHTTP/1.1\t200\tOK", 200),
646        ] {
647            assert_eq!(
648                HttpConnectProtocol::parse_status_line(line).map_err(|error| error.kind()),
649                Ok(status),
650                "{line:?}"
651            );
652        }
653    }
654
655    #[test]
656    fn status_line_rejects_malformed_framing() {
657        for line in [
658            &b""[..],
659            b"HTTP/",
660            b"HTTP/1.1",
661            b"HTTP/1.1 ",
662            b"HTTP/1.1  ",
663            b"HTTP/2 200 OK",
664            b"HTTP/2.0 200 OK",
665            b"HTTP/1.2 200 OK",
666            b"HTTP/1.10 200 OK",
667            b"http/1.1 200 OK",
668            b"Http/1.1 200 OK",
669            b"HTTP/1.1200 OK",
670            b"HTTP/1.1 200OK",
671            b"HTTP/1.1 +200 OK",
672            b"HTTP/1.1 +20 OK",
673            b"HTTP/1.1 +99 OK",
674            b"HTTP/1.1 -200 OK",
675            b"HTTP/1.1 0200 OK",
676            b"HTTP/1.1 2O0 OK",
677            b"HTTP/1.1 20 OK",
678            b"HTTP/1.1 20",
679            b"HTTP/1.1 2000 OK",
680            b"HTTP/1.1 OK",
681            "HTTP/1.1 ٢٠٠ OK".as_bytes(),
682            b"HTTP/1.1\xa0200 OK",
683            b"HTTP/1.1\x0c200 OK",
684            b"HTTP/1.1\x0b200 OK",
685            b"HTTP/1.1\r200 OK",
686            b"HTTP/1.1 200 OK\r",
687            b"HTTP/1.1 200 O\nK",
688        ] {
689            assert_eq!(
690                HttpConnectProtocol::parse_status_line(line).map_err(|error| error.kind()),
691                Err(io::ErrorKind::InvalidData),
692                "{line:?}"
693            );
694        }
695
696        // Pin the exact diagnostic for each rejection class. Each class gets a
697        // second, structurally different input so the classes cannot collapse or
698        // swap messages; `+20` in particular records why the code must be three
699        // ASCII digits rather than merely parseable as a `u16`.
700        for (line, message) in [
701            (&b"HTTP/1.1 200 O\0K"[..], CONTROL),
702            (b"HTTP/2 200 OK", INVALID_LINE),
703            (b"http/1.1 200 OK", INVALID_LINE),
704            (b"HTTP/1.1", NO_CODE),
705            (b"HTTP/1.1 ", NO_CODE),
706            (b"HTTP/1.1 +200 OK", CODE_LENGTH),
707            (b"HTTP/1.1 0200 OK", CODE_LENGTH),
708            (b"HTTP/1.1 +20 OK", CODE_DIGITS),
709            (b"HTTP/1.1 2O0 OK", CODE_DIGITS),
710            (b"HTTP/1.1 200 \xff", NON_ASCII),
711        ] {
712            assert_eq!(
713                HttpConnectProtocol::parse_status_line(line)
714                    .unwrap_err()
715                    .to_string(),
716                message,
717                "{line:?}"
718            );
719        }
720    }
721
722    #[test]
723    fn status_line_byte_class_sweep() {
724        for byte in 0..=u8::MAX {
725            // The oracle is the grammar's explicit byte ranges, not the parser's predicates.
726            let digit = b"0123456789".iter().position(|&digit| byte == digit);
727            let control = byte.is_ascii_control() && byte != b'\t';
728            let non_ascii = byte >= 0x80;
729            let mut cases: Vec<(&str, Vec<u8>, Result<u16, String>)> = vec![
730                (
731                    "reason",
732                    [b"HTTP/1.1 200 A".as_slice(), &[byte], b"B"].concat(),
733                    if matches!(byte, 0x09 | 0x20..=0x7e) {
734                        Ok(200)
735                    } else if control {
736                        Err(CONTROL.to_string())
737                    } else {
738                        Err(NON_ASCII.to_string())
739                    },
740                ),
741                (
742                    "version separator",
743                    [b"HTTP/1.1".as_slice(), &[byte], b"200 OK"].concat(),
744                    if matches!(byte, 0x09 | 0x20) {
745                        Ok(200)
746                    } else if control {
747                        Err(CONTROL.to_string())
748                    } else if non_ascii {
749                        Err(NON_ASCII.to_string())
750                    } else {
751                        Err(INVALID_LINE.to_string())
752                    },
753                ),
754                (
755                    "reason separator",
756                    [b"HTTP/1.1 200".as_slice(), &[byte], b"OK"].concat(),
757                    if matches!(byte, 0x09 | 0x20) {
758                        Ok(200)
759                    } else if control {
760                        Err(CONTROL.to_string())
761                    } else if non_ascii {
762                        Err(NON_ASCII.to_string())
763                    } else {
764                        Err(CODE_LENGTH.to_string())
765                    },
766                ),
767                (
768                    "version digit",
769                    [b"HTTP/1.".as_slice(), &[byte], b" 200 OK"].concat(),
770                    if matches!(byte, b'0' | b'1') {
771                        Ok(200)
772                    } else if control {
773                        Err(CONTROL.to_string())
774                    } else if non_ascii {
775                        Err(NON_ASCII.to_string())
776                    } else {
777                        Err(INVALID_LINE.to_string())
778                    },
779                ),
780            ];
781            // Sweep each of the three code digits independently so a sign or a
782            // non-ASCII digit in any position is rejected, not just the middle one.
783            for (position, weight) in [(0usize, 100u16), (1, 10), (2, 1)] {
784                let mut code = *b"000";
785                code[position] = byte;
786                cases.push((
787                    "code digit",
788                    [b"HTTP/1.1 ".as_slice(), code.as_slice(), b" OK"].concat(),
789                    if let Some(digit) = digit {
790                        Ok(weight * digit as u16)
791                    } else if control {
792                        Err(CONTROL.to_string())
793                    } else if non_ascii {
794                        Err(NON_ASCII.to_string())
795                    } else if matches!(byte, 0x09 | 0x20) {
796                        Err(CODE_LENGTH.to_string())
797                    } else {
798                        Err(CODE_DIGITS.to_string())
799                    },
800                ));
801            }
802            for (label, line, expected) in cases {
803                assert_eq!(
804                    HttpConnectProtocol::parse_status_line(&line)
805                        .map_err(|error| error.to_string()),
806                    expected,
807                    "{label}: {line:?}"
808                );
809            }
810        }
811    }
812
813    #[tokio::test]
814    async fn connect_accepts_all_2xx_statuses() {
815        for (index, status) in [200, 201, 204, 299].into_iter().enumerate() {
816            let version = if index % 2 == 0 {
817                "HTTP/1.0"
818            } else {
819                "HTTP/1.1"
820            };
821            let reason = if status == 204 { "" } else { "OK" };
822            let response = format!("{version} {status} {reason}\r\n\r\n");
823            assert_success(response.into_bytes(), &format!("status {status}"), b"").await;
824        }
825    }
826
827    #[tokio::test]
828    async fn connect_rejects_non_2xx_statuses() {
829        // Expected kinds are an independent oracle, not re-derived from production.
830        for (status, kind) in [
831            (101, io::ErrorKind::ConnectionRefused),
832            (300, io::ErrorKind::ConnectionRefused),
833            (403, io::ErrorKind::ConnectionRefused),
834            (407, io::ErrorKind::PermissionDenied),
835            (503, io::ErrorKind::ConnectionRefused),
836            (0, io::ErrorKind::ConnectionRefused),
837            (999, io::ErrorKind::ConnectionRefused),
838        ] {
839            let response = format!("HTTP/1.1 {status:03} Rejected\r\n\r\n");
840            let (result, request) = exchange(response.into_bytes(), destination()).await;
841            assert_eq!(request, expected_request(destination()), "status {status}");
842            let error = result.unwrap_err();
843            assert_eq!(error.kind(), kind, "status {status}");
844            // 407 has a dedicated authentication diagnostic; every other non-2xx
845            // status keeps the generic rejection message.
846            let expected = if status == 407 {
847                "HTTP CONNECT proxy requires authentication (status 407)".to_string()
848            } else {
849                format!("HTTP CONNECT proxy rejected the tunnel with status {status}")
850            };
851            assert_eq!(error.to_string(), expected, "status {status}");
852        }
853    }
854
855    #[tokio::test]
856    async fn connect_sends_canonical_ipv6_authority() {
857        let destination = "[2001:db8::9]:443".parse().unwrap();
858        let (result, request) = exchange(b"HTTP/1.1 200 OK\r\n\r\n".to_vec(), destination).await;
859        result.unwrap();
860        assert_eq!(
861            request,
862            b"CONNECT [2001:db8::9]:443 HTTP/1.1\r\nHost: [2001:db8::9]:443\r\n\r\n"
863        );
864    }
865
866    #[tokio::test]
867    async fn connect_enforces_response_header_limit() {
868        assert_eq!(CONNECT_RESPONSE_HEADER_LIMIT, 8192);
869        for len in [8191, 8192] {
870            assert_success(
871                response_headers_with_len(b"HTTP/1.1 200 OK", len),
872                &format!("length {len}"),
873                b"",
874            )
875            .await;
876        }
877        let oversized = response_headers_with_len(b"HTTP/1.1 200 OK", 8193);
878        let (result, request) = exchange(oversized, destination()).await;
879        assert_eq!(request, expected_request(destination()));
880        let error = result.unwrap_err();
881        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
882        assert_eq!(
883            error.to_string(),
884            "HTTP CONNECT proxy response headers exceeded 8192 bytes"
885        );
886
887        for (len, kind) in [
888            (8192, io::ErrorKind::InvalidData),
889            (8191, io::ErrorKind::UnexpectedEof),
890        ] {
891            let mut response = b"HTTP/1.1 200 OK\r\nX-Pad: ".to_vec();
892            response.resize(len, b'a');
893            assert_eq!(response.len(), len, "unterminated length {len}");
894            assert_error(response, &format!("unterminated length {len}"), kind).await;
895        }
896
897        let tail = b"\0\xfflimit\r\n\r\n";
898        let mut response = response_headers_with_len(b"HTTP/1.1 200 OK", 8192);
899        response.extend_from_slice(tail);
900        assert_success(response, "exact limit with tail", tail).await;
901    }
902
903    #[tokio::test]
904    async fn connect_counts_interim_headers_toward_total_limit() {
905        for (len, accepted) in [(4096, true), (4097, false)] {
906            let mut response = response_headers_with_len(b"HTTP/1.1 103 Early Hints", 4096);
907            response.extend(response_headers_with_len(b"HTTP/1.1 200 OK", len));
908            if accepted {
909                assert_success(response, "4096 + 4096", b"").await;
910            } else {
911                assert_error(response, "4096 + 4097", io::ErrorKind::InvalidData).await;
912            }
913        }
914    }
915
916    #[tokio::test]
917    async fn connect_rejects_early_eof() {
918        for response in [
919            &b""[..],
920            b"HTTP/1.1 20",
921            b"HTTP/1.1 200 OK\r\nX-Test: value\r\n",
922            b"HTTP/1.1 103 Early Hints\r\n\r\n",
923        ] {
924            assert_error(
925                response.to_vec(),
926                &format!("{response:?}"),
927                io::ErrorKind::UnexpectedEof,
928            )
929            .await;
930        }
931        assert_success(
932            b"HTTP/1.1 200 OK\r\n\r\n".to_vec(),
933            "complete before EOF",
934            b"",
935        )
936        .await;
937    }
938
939    #[tokio::test]
940    async fn connect_rejects_101_without_waiting_for_another_response() {
941        let (listener, address, request_tx, request_rx) = bind_mock_proxy().await;
942        let (release_tx, release_rx) = oneshot::channel();
943        let task = tokio::spawn(async move {
944            let mut stream = accept_and_capture(listener, request_tx).await;
945            bounded(stream.write_all(b"HTTP/1.1 101 Switching Protocols\r\n\r\n"))
946                .await
947                .unwrap();
948            bounded(release_rx).await.unwrap();
949        });
950
951        let result = bounded(HttpConnectProtocol::connect(address, destination(), None)).await;
952
953        let request = bounded(request_rx).await.unwrap();
954        release_tx.send(()).unwrap();
955        join_proxy(task).await;
956
957        assert_eq!(request, expected_request(destination()));
958        let error = result.expect_err("101 must be terminal while proxy stays open");
959        assert_eq!(error.kind(), io::ErrorKind::ConnectionRefused);
960        assert_eq!(
961            error.to_string(),
962            "HTTP CONNECT proxy rejected the tunnel with status 101"
963        );
964    }
965
966    #[tokio::test]
967    async fn connect_preserves_tunnel_bytes_and_supports_bidirectional_io() {
968        let (listener, address, request_tx, request_rx) = bind_mock_proxy().await;
969        let tail = b"\0\xffearly\r\n\r\ndata";
970        let task = tokio::spawn(async move {
971            let mut stream = accept_and_capture(listener, request_tx).await;
972            let response = [b"HTTP/1.1 201 Created\r\n\r\n".as_slice(), tail].concat();
973            bounded(stream.write_all(&response)).await.unwrap();
974            let mut ping = [0; 4];
975            bounded(stream.read_exact(&mut ping)).await.unwrap();
976            assert_eq!(&ping, b"ping");
977            bounded(stream.write_all(b"pong")).await.unwrap();
978        });
979
980        let mut stream = bounded(HttpConnectProtocol::connect(address, destination(), None))
981            .await
982            .unwrap();
983        assert_eq!(stream.peer_addr().unwrap(), address);
984        assert_eq!(
985            bounded(request_rx).await.unwrap(),
986            expected_request(destination())
987        );
988        let mut received = vec![0; tail.len()];
989        bounded(stream.read_exact(&mut received)).await.unwrap();
990        assert_eq!(received, tail);
991        bounded(stream.write_all(b"ping")).await.unwrap();
992        let mut pong = [0; 4];
993        bounded(stream.read_exact(&mut pong)).await.unwrap();
994        assert_eq!(&pong, b"pong");
995        join_proxy(task).await;
996    }
997
998    #[tokio::test]
999    async fn connect_waits_for_all_fragments_and_preserves_tail() {
1000        // Each server phase is released with an explicit ack, so this proves the
1001        // handshake does not complete before the last fragment arrives and that
1002        // bytes queued after the terminator are preserved. Polling the handshake
1003        // once per phase does not prove it consumed fragment k before k + 1 was
1004        // written (bytes may still be in flight), so it is not an incremental
1005        // parsing proof.
1006        let cases: &[(&str, &[&[u8]])] = &[
1007            ("status code", &[b"HTTP/1.1 2", b"00 OK\r\n", b"\r", b"\n"]),
1008            (
1009                "UTF-8 reason",
1010                &[b"HTTP/1.1 200 caf\xc3", b"\xa9\r", b"\n\r", b"\n"],
1011            ),
1012        ];
1013        for (label, fragments) in cases {
1014            let label = format!("{label}: {fragments:?}");
1015            let (listener, address, request_tx, request_rx) = bind_mock_proxy().await;
1016            let tail = b"\0\xfffragmented tail";
1017            let fragments: Vec<Vec<u8>> =
1018                fragments.iter().map(|fragment| fragment.to_vec()).collect();
1019            let mut phases = Vec::new();
1020            let mut server_phases = Vec::new();
1021            for _ in &fragments {
1022                let (written_tx, written_rx) = oneshot::channel();
1023                let (ack_tx, ack_rx) = oneshot::channel();
1024                phases.push((written_rx, ack_tx));
1025                server_phases.push((written_tx, ack_rx));
1026            }
1027            let task = tokio::spawn(async move {
1028                let mut stream = accept_and_capture(listener, request_tx).await;
1029                let last = fragments.len() - 1;
1030                for (index, (mut fragment, (written_tx, ack_rx))) in
1031                    fragments.into_iter().zip(server_phases).enumerate()
1032                {
1033                    if index == last {
1034                        fragment.extend_from_slice(tail);
1035                    }
1036                    bounded(stream.write_all(&fragment)).await.unwrap();
1037                    written_tx.send(()).unwrap();
1038                    bounded(ack_rx).await.unwrap();
1039                }
1040            });
1041
1042            let mut handshake =
1043                Box::pin(HttpConnectProtocol::connect(address, destination(), None));
1044
1045            let request = tokio::select! {
1046                biased;
1047                result = &mut handshake => panic!("{label}: completed before fragments: {result:?}"),
1048                captured = bounded(request_rx) => captured.unwrap(),
1049            };
1050            assert_eq!(request, expected_request(destination()), "{label}");
1051            let last = phases.len() - 1;
1052            for (index, (written_rx, ack_tx)) in phases.into_iter().enumerate() {
1053                // Await the server phase without awaiting a pending handshake: no implicit sleeps.
1054                bounded(written_rx).await.unwrap();
1055                if index < last {
1056                    assert!(
1057                        futures::poll!(&mut handshake).is_pending(),
1058                        "{label}: fragment {index}"
1059                    );
1060                }
1061                ack_tx.send(()).unwrap();
1062            }
1063            let mut stream = bounded(&mut handshake).await.unwrap();
1064            assert_eq!(stream.peer_addr().unwrap(), address, "{label}");
1065            let mut received = vec![0; tail.len()];
1066            bounded(stream.read_exact(&mut received)).await.unwrap();
1067            assert_eq!(received, tail, "{label}");
1068            join_proxy(task).await;
1069        }
1070    }
1071
1072    #[tokio::test]
1073    async fn connect_times_out_stalled_response() {
1074        assert_eq!(CONNECT_RESPONSE_TIMEOUT, Duration::from_secs(10));
1075        for (label, response) in [
1076            ("no bytes", &b""[..]),
1077            ("partial final", b"HTTP/1.1 200 OK\r\nX-Test: partial"),
1078            ("interim then stall", b"HTTP/1.1 103 Early Hints\r\n\r\n"),
1079        ] {
1080            let (address, request_rx, written_rx, task) =
1081                spawn_stalled_proxy(response.to_vec()).await;
1082            let started = Instant::now();
1083            let mut handshake =
1084                Box::pin(HttpConnectProtocol::connect(address, destination(), None));
1085
1086            let request = tokio::select! {
1087                biased;
1088                result = &mut handshake => panic!("{label}: stalled proxy completed early: {result:?}"),
1089                captured = bounded(request_rx) => captured.unwrap(),
1090            };
1091            assert_eq!(request, expected_request(destination()), "{label}");
1092            tokio::select! {
1093                biased;
1094                result = &mut handshake => panic!("{label}: stalled proxy completed early: {result:?}"),
1095                written = bounded(written_rx) => written.unwrap(),
1096            }
1097            assert!(futures::poll!(&mut handshake).is_pending(), "{label}");
1098
1099            // On macOS, advancing virtual time can outrun socket readiness for the
1100            // written variants: closing with unread bytes produces a reset instead of EOF.
1101            // Use the plan's real-clock fallback, retaining the production 10s deadline.
1102            let outcome = timeout(Duration::from_secs(15), &mut handshake).await;
1103            let error = outcome
1104                .expect("response deadline was not enforced")
1105                .unwrap_err();
1106            assert_eq!(error.kind(), io::ErrorKind::TimedOut, "{label}");
1107            let elapsed = Instant::now().duration_since(started);
1108            assert!(
1109                elapsed >= CONNECT_RESPONSE_TIMEOUT,
1110                "{label}: expired early"
1111            );
1112            assert!(elapsed < Duration::from_secs(15), "{label}: expired late");
1113            join_proxy(task).await;
1114        }
1115    }
1116
1117    #[tokio::test]
1118    async fn cancelling_connect_closes_proxy_socket() {
1119        let (address, request_rx, written_rx, task) = spawn_stalled_proxy(Vec::new()).await;
1120        let handshake = tokio::spawn(HttpConnectProtocol::connect(address, destination(), None));
1121
1122        let request = bounded(request_rx).await.unwrap();
1123        assert_eq!(request, expected_request(destination()));
1124        bounded(written_rx).await.unwrap();
1125
1126        handshake.abort();
1127        let error = bounded(handshake).await.unwrap_err();
1128        assert!(error.is_cancelled());
1129        // The proxy must observe EOF, not be aborted to hide a leaked client socket.
1130        join_proxy(task).await;
1131    }
1132
1133    #[tokio::test]
1134    async fn connect_sends_exactly_one_basic_authorization_header() {
1135        let credentials = ResolvedHttpConnectCredentials::new("user", "pass");
1136        let (result, request) = exchange_authenticated(
1137            b"HTTP/1.1 200 OK\r\n\r\n".to_vec(),
1138            destination(),
1139            &credentials,
1140        )
1141        .await;
1142        result.expect("authenticated tunnel");
1143
1144        let token = base64::engine::general_purpose::STANDARD.encode("user:pass");
1145        let expected = format!(
1146            "CONNECT {0} HTTP/1.1\r\nHost: {0}\r\nProxy-Authorization: Basic {token}\r\n\r\n",
1147            destination()
1148        )
1149        .into_bytes();
1150        assert_eq!(request, expected);
1151        // Regression guard: exactly one authorization header, never a duplicate.
1152        assert_eq!(
1153            request
1154                .split(|byte| *byte == b'\n')
1155                .filter(|line| line.starts_with(b"Proxy-Authorization:"))
1156                .count(),
1157            1
1158        );
1159    }
1160
1161    #[tokio::test]
1162    async fn connect_sends_empty_username_as_leading_colon() {
1163        let credentials = ResolvedHttpConnectCredentials::new("", "token");
1164        let (result, request) = exchange_authenticated(
1165            b"HTTP/1.1 200 OK\r\n\r\n".to_vec(),
1166            destination(),
1167            &credentials,
1168        )
1169        .await;
1170        result.expect("token-only tunnel");
1171
1172        // base64(":token"), written out so the encoding is checked independently.
1173        let expected = b"Proxy-Authorization: Basic OnRva2Vu\r\n";
1174        assert!(
1175            request
1176                .windows(expected.len())
1177                .any(|window| window == expected),
1178            "{request:?}"
1179        );
1180    }
1181
1182    #[tokio::test]
1183    async fn connect_base64_encodes_password_bytes_verbatim() {
1184        // Only the first ':' separates the userinfo, and the whole pair is
1185        // base64-encoded, so a password may contain ':', space, tab, and non-ASCII.
1186        // The expected tokens are written out in full rather than re-encoded with
1187        // the same library, so a wrong alphabet or padding choice cannot mask
1188        // itself.
1189        for (password, token) in [
1190            // Identical under the URL-safe alphabet.
1191            ("p:a ss\t\u{e9}", "dXNlcjpwOmEgc3MJw6k="),
1192            // Contains '/' and '=', so the URL-safe and unpadded variants differ.
1193            ("a\u{ff}\u{e9}", "dXNlcjphw7/DqQ=="),
1194        ] {
1195            let credentials = ResolvedHttpConnectCredentials::new("user", password);
1196            let (result, request) = exchange_authenticated(
1197                b"HTTP/1.1 200 OK\r\n\r\n".to_vec(),
1198                destination(),
1199                &credentials,
1200            )
1201            .await;
1202            result.expect("authenticated tunnel");
1203
1204            let expected = format!("Proxy-Authorization: Basic {token}").into_bytes();
1205            assert!(
1206                request
1207                    .windows(expected.len())
1208                    .any(|window| window == expected),
1209                "password {password:?}: {request:?}"
1210            );
1211        }
1212    }
1213
1214    #[test]
1215    fn build_connect_request_allocates_the_buffer_once() {
1216        // Growing the request buffer would reallocate and free a block that already
1217        // holds the base64 token; `Zeroizing` only wipes the final allocation.
1218        let credentials = ResolvedHttpConnectCredentials::new("user", "a\u{ff}\u{e9}");
1219        for authority in ["192.0.2.10:443", "[2001:db8::1]:443"] {
1220            for credentials in [None, Some(&credentials)] {
1221                let request = HttpConnectProtocol::build_connect_request(authority, credentials);
1222                assert_eq!(
1223                    request.capacity(),
1224                    request.len(),
1225                    "request for {authority} was not allocated in one pass"
1226                );
1227            }
1228        }
1229    }
1230
1231    #[tokio::test]
1232    async fn connect_reports_407_as_an_authentication_failure() {
1233        let response =
1234            b"HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic\r\n\r\n"
1235                .to_vec();
1236
1237        let (result, request) = exchange(response.clone(), destination()).await;
1238        assert_eq!(request, expected_request(destination()));
1239        let error = result.expect_err("407 without credentials");
1240        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
1241        assert!(
1242            error.to_string().contains("requires authentication"),
1243            "{error}"
1244        );
1245
1246        let credentials = ResolvedHttpConnectCredentials::new("user", "pass");
1247        let (result, _) = exchange_authenticated(response, destination(), &credentials).await;
1248        let error = result.expect_err("407 with credentials");
1249        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
1250        assert!(
1251            error
1252                .to_string()
1253                .contains("rejected the supplied credentials"),
1254            "{error}"
1255        );
1256    }
1257
1258    #[test]
1259    fn resolved_credentials_never_render_the_password() {
1260        let credentials = ResolvedHttpConnectCredentials::new("user", "hunter2");
1261        let debug = format!("{credentials:?}");
1262        assert!(debug.contains("user"), "{debug}");
1263        assert!(debug.contains("[REDACTED]"), "{debug}");
1264        assert!(!debug.contains("hunter2"), "{debug}");
1265    }
1266
1267    #[test]
1268    fn builder_rejects_invalid_http_connect_credentials() {
1269        for (username, var) in [
1270            ("user:name", "PROXY_PASSWORD"),
1271            ("user\nname", "PROXY_PASSWORD"),
1272            ("user", ""),
1273        ] {
1274            let error = OutboundProxyBuilder::new()
1275                .http_connect("127.0.0.1:3128")
1276                .credentials(username, SecretSource::env(var))
1277                .build()
1278                .expect_err("invalid credentials must not build");
1279            assert!(
1280                matches!(
1281                    error,
1282                    OutboundProxyBuildError::InvalidHttpConnectCredentials { .. }
1283                ),
1284                "{username:?}/{var:?}: {error}"
1285            );
1286        }
1287
1288        // RFC 7617 permits an empty user-id for token-only proxies.
1289        OutboundProxyBuilder::new()
1290            .http_connect("127.0.0.1:3128")
1291            .credentials("", SecretSource::env("PROXY_PASSWORD"))
1292            .build()
1293            .unwrap();
1294
1295        let proxy = OutboundProxyBuilder::new()
1296            .http_connect("127.0.0.1:3128")
1297            .credentials("user", SecretSource::env("PROXY_PASSWORD"))
1298            .build()
1299            .unwrap();
1300        assert!(matches!(
1301            proxy,
1302            OutboundProxy::HttpConnect {
1303                credentials: Some(_),
1304                ..
1305            }
1306        ));
1307    }
1308
1309    #[test]
1310    fn resolved_unauthenticated_http_connect_omits_credentials() {
1311        let resolved = ResolvedOutboundProxy::HttpConnect {
1312            address: "127.0.0.1:3128".parse().unwrap(),
1313            credentials: None,
1314        };
1315
1316        // Older runtimes decode this variant without a credentials member, so an
1317        // unauthenticated proxy keeps its previous launch JSON shape.
1318        assert_eq!(
1319            serde_json::to_value(&resolved).unwrap(),
1320            serde_json::json!({"http_connect": {"address": "127.0.0.1:3128"}})
1321        );
1322        let restored: ResolvedOutboundProxy = serde_json::from_value(
1323            serde_json::json!({"http_connect": {"address": "127.0.0.1:3128"}}),
1324        )
1325        .unwrap();
1326        assert_eq!(restored, resolved);
1327    }
1328
1329    #[test]
1330    fn durable_http_connect_credentials_round_trip_without_the_secret() {
1331        let proxy = OutboundProxyBuilder::new()
1332            .http_connect("127.0.0.1:3128")
1333            .credentials("user", SecretSource::env("PROXY_PASSWORD"))
1334            .build()
1335            .unwrap();
1336
1337        let json = serde_json::to_value(&proxy).unwrap();
1338        assert_eq!(
1339            json,
1340            serde_json::json!({
1341                "protocol": "http_connect",
1342                "address": "127.0.0.1:3128",
1343                "credentials": {
1344                    "username": "user",
1345                    "password": {"kind": "env", "var": "PROXY_PASSWORD"}
1346                }
1347            })
1348        );
1349        let restored: OutboundProxy = serde_json::from_value(json).unwrap();
1350        assert_eq!(restored, proxy);
1351
1352        // The field is additive: an unauthenticated proxy omits the key entirely.
1353        let plain = OutboundProxyBuilder::new()
1354            .http_connect("127.0.0.1:3128")
1355            .build()
1356            .unwrap();
1357        assert!(
1358            serde_json::to_value(&plain)
1359                .unwrap()
1360                .get("credentials")
1361                .is_none()
1362        );
1363    }
1364
1365    async fn assert_success(response: Vec<u8>, label: &str, tail: &[u8]) {
1366        let (result, request) = exchange(response, destination()).await;
1367
1368        assert_eq!(request, expected_request(destination()), "{label}");
1369        let mut stream = result.unwrap_or_else(|error| panic!("{label}: {error}"));
1370        let mut received = vec![0; tail.len()];
1371        bounded(stream.read_exact(&mut received)).await.unwrap();
1372        assert_eq!(received, tail, "{label}");
1373    }
1374
1375    async fn assert_error(response: Vec<u8>, label: &str, kind: io::ErrorKind) -> io::Error {
1376        let (result, request) = exchange(response, destination()).await;
1377
1378        assert_eq!(request, expected_request(destination()), "{label}");
1379        let error = result.expect_err(label);
1380        assert_eq!(error.kind(), kind, "{label}: {error}");
1381        error
1382    }
1383
1384    async fn exchange(
1385        response: Vec<u8>,
1386        destination: SocketAddr,
1387    ) -> (io::Result<TcpStream>, Vec<u8>) {
1388        let (address, request_rx, task) = spawn_mock_proxy(response).await;
1389        let result = bounded(HttpConnectProtocol::connect(address, destination, None)).await;
1390
1391        let request = bounded(request_rx).await.expect("request capture stalled");
1392        join_proxy(task).await;
1393        if let Ok(stream) = &result {
1394            assert_eq!(stream.peer_addr().unwrap(), address);
1395        }
1396        (result, request)
1397    }
1398
1399    async fn exchange_authenticated(
1400        response: Vec<u8>,
1401        destination: SocketAddr,
1402        credentials: &ResolvedHttpConnectCredentials,
1403    ) -> (io::Result<TcpStream>, Vec<u8>) {
1404        let (address, request_rx, task) = spawn_mock_proxy(response).await;
1405        let result = bounded(HttpConnectProtocol::connect(
1406            address,
1407            destination,
1408            Some(credentials),
1409        ))
1410        .await;
1411
1412        let request = bounded(request_rx).await.expect("request capture stalled");
1413        join_proxy(task).await;
1414        if let Ok(stream) = &result {
1415            assert_eq!(stream.peer_addr().unwrap(), address);
1416        }
1417        (result, request)
1418    }
1419
1420    async fn spawn_mock_proxy(
1421        response: Vec<u8>,
1422    ) -> (SocketAddr, oneshot::Receiver<Vec<u8>>, JoinHandle<()>) {
1423        let (listener, address, request_tx, request_rx) = bind_mock_proxy().await;
1424        let task = tokio::spawn(async move {
1425            let mut stream = accept_and_capture(listener, request_tx).await;
1426            bounded(stream.write_all(&response))
1427                .await
1428                .expect("proxy response write stalled");
1429            // Returning drops the stream, producing EOF for empty/truncated responses.
1430        });
1431        (address, request_rx, task)
1432    }
1433
1434    async fn spawn_stalled_proxy(
1435        response: Vec<u8>,
1436    ) -> (
1437        SocketAddr,
1438        oneshot::Receiver<Vec<u8>>,
1439        oneshot::Receiver<()>,
1440        JoinHandle<()>,
1441    ) {
1442        let (listener, address, request_tx, request_rx) = bind_mock_proxy().await;
1443        let (written_tx, written_rx) = oneshot::channel();
1444        let task = tokio::spawn(async move {
1445            let mut stream = accept_and_capture(listener, request_tx).await;
1446            bounded(stream.write_all(&response)).await.unwrap();
1447            written_tx.send(()).unwrap();
1448            let mut byte = [0];
1449            // A 5s EOF watchdog would fire before the real 10s handshake deadline.
1450            // The caller bounds the fixture join after timeout or cancellation.
1451            assert_eq!(
1452                stream.read(&mut byte).await.unwrap(),
1453                0,
1454                "client did not close the proxy socket"
1455            );
1456        });
1457        (address, request_rx, written_rx, task)
1458    }
1459
1460    async fn read_request_headers(stream: &mut TcpStream) -> Vec<u8> {
1461        bounded(async {
1462            let mut request = Vec::new();
1463            loop {
1464                assert!(request.len() < 8192, "unterminated CONNECT request");
1465                request.push(stream.read_u8().await.unwrap());
1466                if request.ends_with(b"\r\n\r\n") {
1467                    return request;
1468                }
1469            }
1470        })
1471        .await
1472    }
1473
1474    fn response_headers_with_len(status_line: &[u8], total_len: usize) -> Vec<u8> {
1475        let mut response = status_line.to_vec();
1476        response.extend_from_slice(b"\r\nX-Pad: ");
1477        let fixed_len = response.len() + b"\r\n\r\n".len();
1478        assert!(total_len >= fixed_len);
1479        response.resize(response.len() + total_len - fixed_len, b'a');
1480        response.extend_from_slice(b"\r\n\r\n");
1481        assert_eq!(response.len(), total_len);
1482        response
1483    }
1484
1485    fn expected_request(destination: SocketAddr) -> Vec<u8> {
1486        format!("CONNECT {destination} HTTP/1.1\r\nHost: {destination}\r\n\r\n").into_bytes()
1487    }
1488
1489    fn destination() -> SocketAddr {
1490        "203.0.113.9:443".parse().unwrap()
1491    }
1492
1493    async fn join_proxy(task: JoinHandle<()>) {
1494        bounded(task).await.expect("proxy fixture did not finish");
1495    }
1496
1497    /// Awaits a fixture future under `TEST_TIMEOUT`, panicking if it stalls.
1498    async fn bounded<F: Future>(future: F) -> F::Output {
1499        timeout(TEST_TIMEOUT, future)
1500            .await
1501            .expect("operation did not complete before TEST_TIMEOUT")
1502    }
1503
1504    /// Binds a loopback listener for a mock proxy and returns its address plus
1505    /// the channel used to capture the CONNECT request the client sends.
1506    async fn bind_mock_proxy() -> (
1507        TcpListener,
1508        SocketAddr,
1509        oneshot::Sender<Vec<u8>>,
1510        oneshot::Receiver<Vec<u8>>,
1511    ) {
1512        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1513        let address = listener.local_addr().unwrap();
1514        let (request_tx, request_rx) = oneshot::channel();
1515        (listener, address, request_tx, request_rx)
1516    }
1517
1518    /// Accepts one proxy connection and captures the client's CONNECT request.
1519    async fn accept_and_capture(
1520        listener: TcpListener,
1521        request_tx: oneshot::Sender<Vec<u8>>,
1522    ) -> TcpStream {
1523        let (mut stream, _) = bounded(listener.accept())
1524            .await
1525            .expect("proxy was not contacted");
1526        request_tx
1527            .send(read_request_headers(&mut stream).await)
1528            .unwrap();
1529        stream
1530    }
1531
1532    // --- Independent verification additions (verifier) ---
1533
1534    fn xorshift(state: &mut u64) -> u64 {
1535        let mut value = *state;
1536        value ^= value << 13;
1537        value ^= value >> 7;
1538        value ^= value << 17;
1539        *state = value;
1540        value
1541    }
1542
1543    /// Fuzz over the deterministic xorshift generator: the parser consumes
1544    /// untrusted bytes, so it must always return `Ok`/`Err` and never panic.
1545    #[test]
1546    fn parse_status_line_never_panics_on_arbitrary_bytes() {
1547        let mut state = 0x2545_F491_4F6C_DD1Du64;
1548        let pool: &[u8] = b"HTTP/1.1HTTP/1.0200 \t+-OK\r\n";
1549        let extras: &[u8] = &[
1550            0x00, 0x08, 0x0a, 0x0b, 0x0c, 0x1f, 0x7f, 0x80, 0x85, 0xc3, 0xff, b'a', b'0', b'1',
1551            b'.', b'/',
1552        ];
1553        let seeds: &[&[u8]] = &[
1554            b"",
1555            b"HTTP/1.1 200 OK",
1556            b"HTTP/1.0 204 ",
1557            b"HTTP/1.1\t200 OK",
1558            b"HTTP/1.1  200 OK",
1559            b"HTTP/1.1 200",
1560            b"HTTP/1.1 +20 OK",
1561            b"HTTP/2 200 OK",
1562            b"HTTP/1.1 200 O\x00K",
1563            b"HTTP/1.1 \t 407  Proxy Auth ",
1564        ];
1565        for iteration in 0..200_000u64 {
1566            let mut line: Vec<u8> = match xorshift(&mut state) % 3 {
1567                0 => {
1568                    let len = (xorshift(&mut state) % 24) as usize;
1569                    (0..len)
1570                        .map(|_| {
1571                            if xorshift(&mut state).is_multiple_of(4) {
1572                                extras[(xorshift(&mut state) as usize) % extras.len()]
1573                            } else {
1574                                pool[(xorshift(&mut state) as usize) % pool.len()]
1575                            }
1576                        })
1577                        .collect()
1578                }
1579                _ => {
1580                    let mut line = seeds[(xorshift(&mut state) as usize) % seeds.len()].to_vec();
1581                    if !line.is_empty() {
1582                        for _ in 0..(xorshift(&mut state) % 4) {
1583                            let pos = (xorshift(&mut state) as usize) % line.len();
1584                            line[pos] = if xorshift(&mut state).is_multiple_of(2) {
1585                                extras[(xorshift(&mut state) as usize) % extras.len()]
1586                            } else {
1587                                pool[(xorshift(&mut state) as usize) % pool.len()]
1588                            };
1589                        }
1590                    }
1591                    match xorshift(&mut state) % 4 {
1592                        0 => line.push(extras[(xorshift(&mut state) as usize) % extras.len()]),
1593                        1 if line.len() > 1 => {
1594                            let keep = line.len() - ((xorshift(&mut state) as usize) % 3) - 1;
1595                            line.truncate(keep);
1596                        }
1597                        _ => {}
1598                    }
1599                    line
1600                }
1601            };
1602            if iteration % 7 == 0 {
1603                let mut prefixed = b"HTTP/1.1 200 OK".to_vec();
1604                prefixed.extend_from_slice(&line);
1605                line = prefixed;
1606            }
1607            match HttpConnectProtocol::parse_status_line(&line) {
1608                Ok(status) => assert!(status <= 999, "iteration {iteration}: {line:?}"),
1609                Err(error) => assert_eq!(
1610                    error.kind(),
1611                    io::ErrorKind::InvalidData,
1612                    "iteration {iteration}: {line:?}"
1613                ),
1614            }
1615        }
1616    }
1617}
microsandbox_network/proxy/http_connect.rs

microsandbox_network/proxy/
http_connect.rs
