tenant.openagents/omega

1//! End-to-end tests for the proxy crate.
2//!
3//! Each test spawns a real proxy on `127.0.0.1:0` and makes real TCP
4//! connections to it, optionally also spawning a tiny stub origin server
5//! (or stub upstream proxy) to act as the destination. Everything is sync —
6//! std::net + threads + std::time::Duration timeouts.
7
8use futures::channel::mpsc;
9use futures::stream::StreamExt;
10use http_proxy::{
11    Allowlist, DenyReason, HostPattern, ProxyConfig, ProxyEvent, ProxyHandle, RequestMethod,
12    RequestOutcome, UpstreamProxy,
13};
14use std::io::{Read, Write};
15use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream};
16#[cfg(unix)]
17use std::os::unix::net::UnixStream;
18use std::thread;
19use std::time::Duration;
20
21const TEST_TIMEOUT: Duration = Duration::from_secs(5);
22
23/// Spin up a tiny TCP server that serves one connection: it reads the
24/// client's first request (until `\r\n\r\n`), echoes a fixed HTTP
25/// response, and returns the request bytes it saw.
26fn spawn_echo_origin(response: &'static [u8]) -> (SocketAddr, thread::JoinHandle<Vec<u8>>) {
27    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
28    let addr = listener.local_addr().unwrap();
29    let join = thread::spawn(move || {
30        let (mut sock, _) = listener.accept().unwrap();
31        sock.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
32        let buf = read_until_double_crlf(&mut sock);
33        sock.write_all(response).unwrap();
34        sock.shutdown(std::net::Shutdown::Write).unwrap();
35        buf
36    });
37    (addr, join)
38}
39
40/// Spin up a stub upstream HTTP proxy that serves one connection: it reads
41/// a CONNECT request, replies `200`, then acts like the requested origin —
42/// reading one more request through the "tunnel" and echoing a fixed
43/// response. Returns the CONNECT headers and the tunneled request bytes.
44fn spawn_stub_upstream_proxy(
45    tunnel_response: &'static [u8],
46) -> (SocketAddr, thread::JoinHandle<(Vec<u8>, Vec<u8>)>) {
47    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
48    let addr = listener.local_addr().unwrap();
49    let join = thread::spawn(move || {
50        let (mut sock, _) = listener.accept().unwrap();
51        sock.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
52        let connect_request = read_until_double_crlf(&mut sock);
53        assert!(
54            connect_request.starts_with(b"CONNECT "),
55            "expected CONNECT, got: {:?}",
56            String::from_utf8_lossy(&connect_request)
57        );
58        sock.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n")
59            .unwrap();
60        let tunneled_request = read_until_double_crlf(&mut sock);
61        sock.write_all(tunnel_response).unwrap();
62        sock.shutdown(std::net::Shutdown::Write).unwrap();
63        (connect_request, tunneled_request)
64    });
65    (addr, join)
66}
67
68fn read_until_double_crlf(sock: &mut TcpStream) -> Vec<u8> {
69    let mut buf = Vec::with_capacity(4096);
70    let mut tmp = [0u8; 4096];
71    loop {
72        let n = sock.read(&mut tmp).unwrap_or(0);
73        if n == 0 {
74            break;
75        }
76        buf.extend_from_slice(&tmp[..n]);
77        if buf.windows(4).any(|w| w == b"\r\n\r\n") {
78            break;
79        }
80    }
81    buf
82}
83
84fn spawn_proxy(allowlist: Allowlist) -> (ProxyHandle, mpsc::UnboundedReceiver<ProxyEvent>) {
85    spawn_proxy_with_upstream(allowlist, None)
86}
87
88fn spawn_proxy_with_upstream(
89    allowlist: Allowlist,
90    upstream: Option<UpstreamProxy>,
91) -> (ProxyHandle, mpsc::UnboundedReceiver<ProxyEvent>) {
92    let (events_tx, mut events_rx) = mpsc::unbounded();
93    let proxy = ProxyHandle::spawn(ProxyConfig {
94        allowlist,
95        upstream,
96        events: events_tx,
97    })
98    .expect("proxy spawn");
99
100    drain_ready(&proxy, &mut events_rx);
101    (proxy, events_rx)
102}
103
104#[cfg(unix)]
105fn spawn_unix_proxy(allowlist: Allowlist) -> (ProxyHandle, mpsc::UnboundedReceiver<ProxyEvent>) {
106    let (events_tx, mut events_rx) = mpsc::unbounded();
107    let proxy = ProxyHandle::spawn_unix_temp(ProxyConfig {
108        allowlist,
109        upstream: None,
110        events: events_tx,
111    })
112    .expect("proxy spawn");
113
114    drain_ready(&proxy, &mut events_rx);
115    (proxy, events_rx)
116}
117
118fn drain_ready(proxy: &ProxyHandle, events_rx: &mut mpsc::UnboundedReceiver<ProxyEvent>) {
119    let ready = futures::executor::block_on(events_rx.next());
120    match ready {
121        Some(ProxyEvent::Ready { port }) => assert_eq!(port, proxy.port()),
122        other => panic!("expected Ready event first, got {other:?}"),
123    }
124}
125
126fn next_event(events: &mut mpsc::UnboundedReceiver<ProxyEvent>) -> ProxyEvent {
127    futures::executor::block_on(events.next()).expect("events channel closed")
128}
129
130#[test]
131fn connect_allowed_host_completes_tunnel_and_emits_events() {
132    let (origin_addr, origin_join) = spawn_echo_origin(
133        b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello",
134    );
135
136    // `localhost` isn't a permitted allowlist *pattern*, but
137    // `Allowlist::any()` skips all policy checks (including the
138    // forbidden-resolved-IP filter), which is what lets this test reach a
139    // loopback origin. Denied paths are exercised in separate tests.
140    let (proxy, mut events) = spawn_proxy(Allowlist::any());
141    let target = format!("localhost:{}", origin_addr.port());
142
143    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
144    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
145    client.set_write_timeout(Some(TEST_TIMEOUT)).unwrap();
146
147    let req = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n");
148    client.write_all(req.as_bytes()).unwrap();
149
150    // Read the proxy's CONNECT response headers.
151    let response = read_until_double_crlf(&mut client);
152    let resp_text = String::from_utf8_lossy(&response);
153    assert!(
154        resp_text.starts_with("HTTP/1.1 200"),
155        "expected 200 from proxy, got: {resp_text}"
156    );
157
158    // Send tunnel payload.
159    client.write_all(b"hi origin\r\n\r\n").unwrap();
160    client.shutdown(std::net::Shutdown::Write).unwrap();
161    let _ = client.read_to_end(&mut Vec::new());
162
163    let origin_received = origin_join.join().unwrap();
164    assert!(
165        String::from_utf8_lossy(&origin_received).contains("hi origin"),
166        "origin saw: {:?}",
167        String::from_utf8_lossy(&origin_received)
168    );
169
170    match next_event(&mut events) {
171        ProxyEvent::RequestAttempt {
172            host,
173            port,
174            method,
175            outcome,
176        } => {
177            assert_eq!(host, "localhost");
178            assert_eq!(port, origin_addr.port());
179            assert_eq!(method, RequestMethod::Connect);
180            assert!(matches!(outcome, RequestOutcome::Allowed));
181        }
182        other => panic!("expected RequestAttempt, got {other:?}"),
183    }
184    match next_event(&mut events) {
185        ProxyEvent::RequestCompleted { host, .. } => {
186            assert_eq!(host, "localhost");
187        }
188        other => panic!("expected RequestCompleted, got {other:?}"),
189    }
190}
191
192#[cfg(unix)]
193#[test]
194fn unix_listener_denied_host_returns_511() {
195    let allowlist = Allowlist::from_patterns([HostPattern::parse("github.com").unwrap()]);
196    let (proxy, mut events) = spawn_unix_proxy(allowlist);
197    let socket_path = proxy
198        .socket_path()
199        .expect("Unix proxy socket path")
200        .to_path_buf();
201
202    let mut client = UnixStream::connect(socket_path).unwrap();
203    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
204    client
205        .write_all(b"CONNECT denied.example:443 HTTP/1.1\r\nHost: denied.example:443\r\n\r\n")
206        .unwrap();
207
208    let mut response = String::new();
209    client.read_to_string(&mut response).unwrap();
210    assert!(
211        response.starts_with("HTTP/1.1 511 "),
212        "expected 511, got: {response}"
213    );
214    assert!(response.contains("denied.example"));
215
216    match next_event(&mut events) {
217        ProxyEvent::RequestAttempt {
218            host,
219            port,
220            method,
221            outcome,
222        } => {
223            assert_eq!(host, "denied.example");
224            assert_eq!(port, 443);
225            assert_eq!(method, RequestMethod::Connect);
226            assert!(matches!(outcome, RequestOutcome::Denied { .. }));
227        }
228        other => panic!("expected RequestAttempt(Denied), got {other:?}"),
229    }
230}
231
232#[test]
233fn connect_denied_host_returns_511_with_via_header() {
234    let allowlist = Allowlist::from_patterns([HostPattern::parse("github.com").unwrap()]);
235    let (proxy, mut events) = spawn_proxy(allowlist);
236
237    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
238    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
239    client
240        .write_all(b"CONNECT denied.example:443 HTTP/1.1\r\nHost: denied.example:443\r\n\r\n")
241        .unwrap();
242
243    let mut response = String::new();
244    client.read_to_string(&mut response).unwrap();
245    assert!(
246        response.starts_with("HTTP/1.1 511 "),
247        "expected 511, got: {response}"
248    );
249    // OMEGA-DELTA-0031. The refusal a blocked request reads names Omega's
250    // proxy, not the upstream fork's. This assertion carried the upstream
251    // value; the delta is the policy record, so it is updated rather than
252    // deleted.
253    assert!(response.contains("Via: 1.1 omega-sandbox-proxy"));
254    assert!(response.contains("Proxy-Status: omega-sandbox-proxy"));
255    assert!(response.contains("denied.example"));
256    assert!(response.contains("not in this conversation's network allowlist"));
257
258    match next_event(&mut events) {
259        ProxyEvent::RequestAttempt {
260            host,
261            port,
262            method,
263            outcome,
264        } => {
265            assert_eq!(host, "denied.example");
266            assert_eq!(port, 443);
267            assert_eq!(method, RequestMethod::Connect);
268            assert!(
269                matches!(
270                    outcome,
271                    RequestOutcome::Denied {
272                        reason: DenyReason::HostNotInAllowlist { .. }
273                    }
274                ),
275                "outcome was {outcome:?}"
276            );
277        }
278        other => panic!("expected RequestAttempt(Denied), got {other:?}"),
279    }
280}
281
282#[test]
283fn http_forward_denied_host_returns_511() {
284    let allowlist = Allowlist::from_patterns([HostPattern::parse("github.com").unwrap()]);
285    let (proxy, mut events) = spawn_proxy(allowlist);
286
287    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
288    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
289    client
290        .write_all(b"GET http://denied.example/ HTTP/1.1\r\nHost: denied.example\r\n\r\n")
291        .unwrap();
292
293    let mut response = String::new();
294    client.read_to_string(&mut response).unwrap();
295    assert!(
296        response.starts_with("HTTP/1.1 511 "),
297        "expected 511, got: {response}"
298    );
299    assert!(response.contains("not in this conversation's network allowlist"));
300
301    match next_event(&mut events) {
302        ProxyEvent::RequestAttempt {
303            host,
304            method: RequestMethod::Http(method),
305            outcome:
306                RequestOutcome::Denied {
307                    reason: DenyReason::HostNotInAllowlist { .. },
308                },
309            ..
310        } => {
311            assert_eq!(host, "denied.example");
312            assert_eq!(method, "GET");
313        }
314        other => panic!("expected RequestAttempt(Denied), got {other:?}"),
315    }
316}
317
318#[test]
319fn ip_literal_connect_is_denied_for_pattern_allowlists() {
320    let allowlist = Allowlist::from_patterns([HostPattern::parse("github.com").unwrap()]);
321    let (proxy, mut events) = spawn_proxy(allowlist);
322
323    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
324    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
325    client
326        .write_all(b"CONNECT 1.2.3.4:443 HTTP/1.1\r\nHost: 1.2.3.4:443\r\n\r\n")
327        .unwrap();
328
329    let mut response = String::new();
330    client.read_to_string(&mut response).unwrap();
331    assert!(response.starts_with("HTTP/1.1 511 "));
332    assert!(response.contains("IP literal"));
333
334    match next_event(&mut events) {
335        ProxyEvent::RequestAttempt {
336            outcome:
337                RequestOutcome::Denied {
338                    reason: DenyReason::IpLiteralRejected { target },
339                },
340            ..
341        } => {
342            assert_eq!(target, "1.2.3.4:443");
343        }
344        other => panic!("expected IpLiteralRejected, got {other:?}"),
345    }
346}
347
348#[test]
349fn ip_literal_connect_is_allowed_when_allowlist_allows_any() {
350    // `Allowlist::any()` preserves unrestricted proxy semantics, IP literals included.
351    let (origin_addr, origin_join) = spawn_echo_origin(b"HTTP/1.1 200 OK\r\n\r\n");
352    let (proxy, mut events) = spawn_proxy(Allowlist::any());
353
354    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
355    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
356    let target = format!("127.0.0.1:{}", origin_addr.port());
357    let req = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n");
358    client.write_all(req.as_bytes()).unwrap();
359
360    let response = read_until_double_crlf(&mut client);
361    let resp_text = String::from_utf8_lossy(&response);
362    assert!(
363        resp_text.starts_with("HTTP/1.1 200"),
364        "expected 200 from proxy, got: {resp_text}"
365    );
366
367    client.write_all(b"ping\r\n\r\n").unwrap();
368    client.shutdown(std::net::Shutdown::Write).unwrap();
369    let _ = client.read_to_end(&mut Vec::new());
370    origin_join.join().unwrap();
371
372    match next_event(&mut events) {
373        ProxyEvent::RequestAttempt { host, outcome, .. } => {
374            assert_eq!(host, "127.0.0.1");
375            assert!(matches!(outcome, RequestOutcome::Allowed));
376        }
377        other => panic!("expected RequestAttempt(Allowed), got {other:?}"),
378    }
379}
380
381#[test]
382fn host_resolving_to_loopback_is_denied_for_pattern_allowlists() {
383    // DNS-rebinding protection end-to-end: even when a hostname slips past
384    // the allowlist (here `localhost`, which patterns can't express but the
385    // wire can carry), the proxy must deny anything that resolves into the
386    // local machine rather than connect. The denial surfaces at the
387    // allowlist layer for this probe; the resolved-IP layer behind it is
388    // unit-tested in `connection.rs` (`plan_route` tests), since
389    // exercising it end-to-end would require an allowlisted hostname whose
390    // real DNS points at loopback.
391    let allowlist = Allowlist::from_patterns([HostPattern::parse("github.com").unwrap()]);
392    let (proxy, _events) = spawn_proxy(allowlist);
393
394    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
395    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
396    client
397        .write_all(b"CONNECT localhost:80 HTTP/1.1\r\nHost: localhost:80\r\n\r\n")
398        .unwrap();
399
400    let mut response = String::new();
401    client.read_to_string(&mut response).unwrap();
402    assert!(
403        response.starts_with("HTTP/1.1 511 "),
404        "expected 511, got: {response}"
405    );
406}
407
408#[test]
409fn http_forward_allowed_rewrites_to_origin_form() {
410    let (origin_addr, origin_join) = spawn_echo_origin(
411        b"HTTP/1.1 200 OK\r\nContent-Length: 11\r\nConnection: close\r\n\r\nhello world",
412    );
413
414    let (proxy, mut events) = spawn_proxy(Allowlist::any());
415
416    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
417    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
418    let req = format!(
419        "GET http://localhost:{}/foo HTTP/1.1\r\nHost: localhost:{}\r\n\r\n",
420        origin_addr.port(),
421        origin_addr.port()
422    );
423    client.write_all(req.as_bytes()).unwrap();
424
425    let mut response = Vec::new();
426    client.read_to_end(&mut response).unwrap();
427    let response_str = String::from_utf8_lossy(&response);
428    assert!(response_str.contains("HTTP/1.1 200"));
429    assert!(response_str.contains("hello world"));
430    // Mirror what real HTTP clients do after a `Connection: close`
431    // response: close the socket so the proxy's bidir-pump finishes the
432    // client→upstream half. (Threads handle this fine — when the client's
433    // socket goes out of scope the pump's read returns EOF.)
434    drop(client);
435
436    let origin_saw = origin_join.join().unwrap();
437    let origin_saw_str = String::from_utf8_lossy(&origin_saw);
438    // The absolute-form proxy request must reach the origin rewritten to
439    // origin-form — many real servers don't accept absolute-form.
440    assert!(
441        origin_saw_str.starts_with("GET /foo HTTP/1.1\r\n"),
442        "origin saw: {origin_saw_str}"
443    );
444    assert!(
445        origin_saw_str.contains(&format!("Host: localhost:{}\r\n", origin_addr.port())),
446        "origin saw: {origin_saw_str}"
447    );
448
449    match next_event(&mut events) {
450        ProxyEvent::RequestAttempt {
451            method: RequestMethod::Http(m),
452            outcome: RequestOutcome::Allowed,
453            ..
454        } => {
455            assert_eq!(m, "GET");
456        }
457        other => panic!("expected RequestAttempt(Allowed), got {other:?}"),
458    }
459    match next_event(&mut events) {
460        ProxyEvent::RequestCompleted { .. } => {}
461        other => panic!("expected RequestCompleted, got {other:?}"),
462    }
463}
464
465#[test]
466fn connect_chains_through_upstream_proxy_with_auth() {
467    let (upstream_addr, upstream_join) = spawn_stub_upstream_proxy(b"tunnel says hi");
468    let upstream = UpstreamProxy::parse(
469        Some(&format!(
470            "http://alice:s3cret@127.0.0.1:{}",
471            upstream_addr.port()
472        )),
473        // The stub upstream is on loopback, which `proxyvars` implicitly
474        // bypasses; the destination decides bypassing, and `proxied.example`
475        // isn't local, so the chain is used.
476        None,
477    )
478    .unwrap()
479    .unwrap();
480
481    let allowlist = Allowlist::from_patterns([HostPattern::parse("proxied.example").unwrap()]);
482    let (proxy, mut events) = spawn_proxy_with_upstream(allowlist, Some(upstream));
483
484    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
485    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
486    client
487        .write_all(b"CONNECT proxied.example:443 HTTP/1.1\r\nHost: proxied.example:443\r\n\r\n")
488        .unwrap();
489
490    let response = read_until_double_crlf(&mut client);
491    let resp_text = String::from_utf8_lossy(&response);
492    assert!(
493        resp_text.starts_with("HTTP/1.1 200"),
494        "expected 200 from proxy, got: {resp_text}"
495    );
496
497    client.write_all(b"client tunnel bytes\r\n\r\n").unwrap();
498    client.shutdown(std::net::Shutdown::Write).unwrap();
499    let mut tunneled_back = Vec::new();
500    let _ = client.read_to_end(&mut tunneled_back);
501    assert_eq!(tunneled_back, b"tunnel says hi");
502
503    let (connect_request, tunneled_request) = upstream_join.join().unwrap();
504    let connect_text = String::from_utf8_lossy(&connect_request);
505    assert!(
506        connect_text.starts_with("CONNECT proxied.example:443 HTTP/1.1\r\n"),
507        "upstream saw: {connect_text}"
508    );
509    // Basic auth from the upstream URL is injected on the CONNECT.
510    assert!(
511        connect_text.contains("Proxy-Authorization: Basic YWxpY2U6czNjcmV0\r\n"),
512        "upstream saw: {connect_text}"
513    );
514    assert!(
515        String::from_utf8_lossy(&tunneled_request).contains("client tunnel bytes"),
516        "tunneled: {:?}",
517        String::from_utf8_lossy(&tunneled_request)
518    );
519
520    match next_event(&mut events) {
521        ProxyEvent::RequestAttempt {
522            host,
523            outcome: RequestOutcome::Allowed,
524            ..
525        } => assert_eq!(host, "proxied.example"),
526        other => panic!("expected RequestAttempt(Allowed), got {other:?}"),
527    }
528}
529
530#[test]
531fn http_forward_chains_through_upstream_via_connect_tunnel() {
532    // Plain HTTP through the upstream must also use a CONNECT tunnel pinned
533    // to the approved host. Handing the upstream a routable absolute-form
534    // stream would let keep-alive requests after the first reach hosts the
535    // allowlist never approved.
536    let (upstream_addr, upstream_join) =
537        spawn_stub_upstream_proxy(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok");
538    let upstream = UpstreamProxy::parse(
539        Some(&format!("http://127.0.0.1:{}", upstream_addr.port())),
540        None,
541    )
542    .unwrap()
543    .unwrap();
544
545    let allowlist = Allowlist::from_patterns([HostPattern::parse("proxied.example").unwrap()]);
546    let (proxy, _events) = spawn_proxy_with_upstream(allowlist, Some(upstream));
547
548    let mut client = TcpStream::connect((Ipv4Addr::LOCALHOST, proxy.port())).unwrap();
549    client.set_read_timeout(Some(TEST_TIMEOUT)).unwrap();
550    client
551        .write_all(b"GET http://proxied.example/data HTTP/1.1\r\nHost: proxied.example\r\n\r\n")
552        .unwrap();
553
554    let mut response = Vec::new();
555    client.read_to_end(&mut response).unwrap();
556    let response_str = String::from_utf8_lossy(&response);
557    assert!(
558        response_str.contains("HTTP/1.1 200") && response_str.contains("ok"),
559        "client got: {response_str}"
560    );
561    drop(client);
562
563    let (connect_request, tunneled_request) = upstream_join.join().unwrap();
564    let connect_text = String::from_utf8_lossy(&connect_request);
565    assert!(
566        connect_text.starts_with("CONNECT proxied.example:80 HTTP/1.1\r\n"),
567        "upstream saw: {connect_text}"
568    );
569    // Inside the tunnel, the origin sees an origin-form request.
570    let tunneled_text = String::from_utf8_lossy(&tunneled_request);
571    assert!(
572        tunneled_text.starts_with("GET /data HTTP/1.1\r\n"),
573        "tunneled: {tunneled_text}"
574    );
575    assert!(
576        tunneled_text.contains("Host: proxied.example\r\n"),
577        "tunneled: {tunneled_text}"
578    );
579}
580
581#[test]
582fn dropping_handle_stops_listener() {
583    let (proxy, _events) = spawn_proxy(Allowlist::any());
584    let port = proxy.port();
585    drop(proxy);
586
587    // After Drop, the listener has been signaled and woken; new
588    // connections should be refused.
589    let result = TcpStream::connect_timeout(
590        &SocketAddr::from((Ipv4Addr::LOCALHOST, port)),
591        Duration::from_millis(500),
592    );
593    assert!(
594        result.is_err(),
595        "listener should have stopped after ProxyHandle drop, but new connection succeeded"
596    );
597}
598