fuchsia_hyper/
fuchsia.rs

1// Copyright 2018 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::happy_eyeballs::{self, RealSocketConnector};
6use crate::{
7    HyperConnectorFuture, SocketOptions, TcpOptions, TcpStream, connect_and_bind_device,
8    parse_ip_addr,
9};
10use fidl::endpoints::{ClientEnd, create_endpoints};
11use fidl_connector::{Connect, ServiceReconnector};
12use fidl_fuchsia_net_name::{LookupIpOptions, LookupMarker, LookupProxy, LookupResult};
13use fidl_fuchsia_posix_socket::{Domain, ProviderMarker, StreamSocketProtocol};
14use fuchsia_async::net;
15
16use futures::future::{Future, FutureExt};
17use futures::io;
18use futures::task::{Context, Poll};
19use http::uri::{Scheme, Uri};
20use hyper_util::rt::TokioIo;
21use rustls::RootCertStore;
22use std::convert::TryFrom as _;
23use std::net::SocketAddr;
24use std::num::TryFromIntError;
25use std::sync::{Arc, LazyLock};
26use tower_service::Service;
27use zx::StatusExt;
28
29pub fn new_root_cert_store() -> Arc<RootCertStore> {
30    // It can be expensive to parse the certs, so cache them
31    static ROOT_STORE: LazyLock<Arc<RootCertStore>> = LazyLock::new(|| {
32        let mut root_store = rustls::RootCertStore::empty();
33
34        root_store.extend(webpki_roots_fuchsia::TLS_SERVER_ROOTS.iter().cloned());
35
36        Arc::new(root_store)
37    });
38
39    Arc::clone(&ROOT_STORE)
40}
41
42/// A Fuchsia-compatible implementation of hyper's `Connect` trait which allows
43/// creating a TcpStream to a particular destination.
44#[derive(Clone)]
45pub struct HyperConnector {
46    tcp_options: TcpOptions,
47    socket_options: SocketOptions,
48    provider: RealServiceConnector,
49}
50
51impl From<(TcpOptions, SocketOptions)> for HyperConnector {
52    fn from((tcp_options, socket_options): (TcpOptions, SocketOptions)) -> Self {
53        Self { tcp_options, socket_options, provider: RealServiceConnector::new() }
54    }
55}
56
57impl HyperConnector {
58    pub fn new() -> Self {
59        Self::from_tcp_options(TcpOptions::default())
60    }
61
62    pub fn from_tcp_options(tcp_options: TcpOptions) -> Self {
63        Self {
64            tcp_options,
65            socket_options: SocketOptions::default(),
66            provider: RealServiceConnector::new(),
67        }
68    }
69}
70
71impl Service<Uri> for HyperConnector {
72    type Response = TokioIo<TcpStream>;
73    type Error = std::io::Error;
74    type Future = HyperConnectorFuture;
75
76    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
77        // This connector is always ready, but others might not be.
78        Poll::Ready(Ok(()))
79    }
80
81    fn call(&mut self, dst: Uri) -> Self::Future {
82        let self_ = self.clone();
83        HyperConnectorFuture { fut: Box::pin(async move { self_.call_async(dst).await }) }
84    }
85}
86
87impl HyperConnector {
88    async fn call_async(&self, dst: Uri) -> Result<TokioIo<TcpStream>, io::Error> {
89        let host = dst.host().ok_or_else(|| io::Error::other("destination host is unspecified"))?;
90        let port = match dst.port() {
91            Some(port) => port.as_u16(),
92            None => {
93                if dst.scheme() == Some(&Scheme::HTTPS) {
94                    443
95                } else {
96                    80
97                }
98            }
99        };
100
101        let stream = connect_to_addr(
102            self.provider.clone(),
103            host,
104            port,
105            self.socket_options.bind_device.as_deref(),
106        )
107        .await?;
108        let () = self.tcp_options.apply(stream.std())?;
109
110        Ok(TokioIo::new(TcpStream { stream }))
111    }
112}
113
114#[allow(dead_code)] // TODO(https://fxbug.dev/421409340)
115#[derive(Clone)]
116pub struct Executor;
117
118impl<F: Future + Send + 'static> hyper::rt::Executor<F> for Executor {
119    fn execute(&self, fut: F) {
120        fuchsia_async::Task::spawn(fut.map(|_| ())).detach()
121    }
122}
123
124#[allow(dead_code)] // TODO(https://fxbug.dev/421409340)
125#[derive(Clone)]
126pub struct LocalExecutor;
127
128impl<F: Future + 'static> hyper::rt::Executor<F> for LocalExecutor {
129    fn execute(&self, fut: F) {
130        fuchsia_async::Task::local(fut.map(drop)).detach()
131    }
132}
133
134pub(crate) trait ProviderConnector {
135    fn connect(&self) -> Result<ClientEnd<ProviderMarker>, io::Error>;
136}
137
138trait LookupConnector {
139    fn connect(&self) -> Result<LookupProxy, io::Error>;
140}
141
142#[derive(Clone)]
143struct RealServiceConnector {
144    name_lookup_connector: ServiceReconnector<LookupMarker>,
145}
146
147impl RealServiceConnector {
148    fn new() -> Self {
149        RealServiceConnector { name_lookup_connector: ServiceReconnector::<LookupMarker>::new() }
150    }
151}
152
153impl ProviderConnector for RealServiceConnector {
154    fn connect(&self) -> Result<ClientEnd<ProviderMarker>, io::Error> {
155        let (client_end, server_end) = create_endpoints::<ProviderMarker>();
156        fuchsia_component::client::connect_channel_to_protocol::<ProviderMarker>(server_end.into())
157            .map_err(|err| {
158                io::Error::other(format!("failed to connect to socket provider service: {}", err))
159            })?;
160        Ok(client_end)
161    }
162}
163
164impl LookupConnector for RealServiceConnector {
165    fn connect(&self) -> Result<LookupProxy, io::Error> {
166        self.name_lookup_connector.connect().map_err(|err| {
167            io::Error::other(format!("failed to connect to name lookup service: {}", err))
168        })
169    }
170}
171
172async fn connect_to_addr<T: ProviderConnector + LookupConnector + 'static>(
173    provider: T,
174    host: &str,
175    port: u16,
176    bind_device: Option<&str>,
177) -> Result<net::TcpStream, io::Error> {
178    if let Some(addr) = parse_ip_addr_with_provider(&provider, host, port).await? {
179        return connect_and_bind_device(&provider, addr, bind_device)?.await;
180    }
181
182    happy_eyeballs::happy_eyeballs(
183        resolve_ip_addr(&provider, host, port).await?,
184        RealSocketConnector::new(provider),
185        happy_eyeballs::RECOMMENDED_MIN_CONN_ATT_DELAY,
186        happy_eyeballs::RECOMMENDED_CONN_ATT_DELAY,
187        bind_device,
188    )
189    .await
190}
191
192async fn resolve_ip_addr<T: LookupConnector>(
193    name_lookup: &T,
194    host: &str,
195    port: u16,
196) -> Result<impl Iterator<Item = SocketAddr> + use<T>, io::Error> {
197    let proxy = name_lookup.connect()?;
198    let LookupResult { addresses, .. } = proxy
199        .lookup_ip(
200            host,
201            &LookupIpOptions {
202                ipv4_lookup: Some(true),
203                ipv6_lookup: Some(true),
204                sort_addresses: Some(true),
205                ..Default::default()
206            },
207        )
208        .await
209        .map_err(|err| io::Error::other(format!("failed to call NameProvider.LookupIp: {}", err)))?
210        .map_err(|err| {
211            // Match stdlib's behavior, which maps all GAI errors but EAI_SYSTEM
212            // to io::ErrorKind::Other.
213            io::Error::other(format!("NameProvider.LookupIp failure: {:?}", err))
214        })?;
215
216    Ok(addresses
217        .ok_or_else(|| io::Error::other("addresses not provided in NameProvider response"))?
218        .into_iter()
219        .map(move |addr| {
220            let fidl_fuchsia_net_ext::IpAddress(addr) = addr.into();
221            SocketAddr::new(addr, port)
222        }))
223}
224
225async fn parse_ip_addr_with_provider(
226    provider: &impl ProviderConnector,
227    host: &str,
228    port: u16,
229) -> Result<Option<SocketAddr>, io::Error> {
230    parse_ip_addr(host, port, |zone_id| async {
231        let proxy = provider.connect()?.into_proxy();
232        let id = proxy
233            .interface_name_to_index(zone_id)
234            .await
235            .map_err(|err| {
236                io::Error::other(format!(
237                    "failed to communicate with socket provider while getting interface index: {}",
238                    err
239                ))
240            })?
241            .map_err(|status| {
242                let status = zx::Status::err_from_raw(status);
243                io::Error::new(
244                    status.into_io_error_kind(),
245                    format!("failed to get interface index from socket provider: {status}"),
246                )
247            })?;
248
249        // SocketAddrV6 only works with 32 bit scope ids.
250        u32::try_from(id).map_err(|TryFromIntError { .. }| {
251            io::Error::other("interface index too large to convert to scope_id")
252        })
253    })
254    .await
255}
256
257pub(crate) fn stream_socket<T: ProviderConnector>(
258    provider: &T,
259    domain: Domain,
260    proto: StreamSocketProtocol,
261) -> io::Result<socket2::Socket> {
262    let socket_provider = provider.connect()?.into_sync_proxy();
263    let sock = socket_provider
264        .stream_socket(domain, proto, zx::MonotonicInstant::INFINITE)
265        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?
266        .map_err(|e| std::io::Error::from_raw_os_error(e.into_primitive()))?;
267
268    Ok(fdio::create_fd(sock.into()).map_err(|s| s.into_io_error())?.into())
269}
270
271#[cfg(test)]
272mod test {
273    use super::*;
274    use crate::*;
275    use assert_matches::assert_matches;
276    use fidl::endpoints::{create_proxy_and_stream, create_request_stream};
277    use fidl_fuchsia_net_name::{LookupError, LookupRequest};
278    use fidl_fuchsia_posix_socket::ProviderRequest;
279    use fuchsia_async::net::TcpListener;
280    use fuchsia_async::{self as fasync, LocalExecutor};
281    use futures::prelude::*;
282    use std::cell::RefCell;
283
284    struct PanicConnector;
285
286    impl ProviderConnector for PanicConnector {
287        fn connect(&self) -> Result<ClientEnd<ProviderMarker>, io::Error> {
288            panic!("should not be trying to talk to the Provider service")
289        }
290    }
291
292    #[test]
293    fn can_create_client() {
294        let _exec = LocalExecutor::default();
295        let _client = new_client();
296    }
297
298    #[test]
299    fn can_create_https_client() {
300        let _exec = LocalExecutor::default();
301        let _client = new_https_client();
302    }
303
304    #[fasync::run_singlethreaded(test)]
305    async fn hyper_connector_sets_tcp_options() {
306        let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0);
307        let listener = TcpListener::bind(&addr).unwrap();
308        let addr = listener.local_addr().unwrap();
309
310        let idle = std::time::Duration::from_secs(36);
311        let interval = std::time::Duration::from_secs(47);
312        let count = 58;
313        let rcvbuf = 1 << 20;
314        let uri = format!("https://{}", addr).parse::<hyper::Uri>().unwrap();
315        let (io, _server) = future::try_join(
316            HyperConnector::from_tcp_options(TcpOptions {
317                keepalive_idle: Some(idle),
318                keepalive_interval: Some(interval),
319                keepalive_count: Some(count),
320                tcp_receive_buffer_size: Some(rcvbuf),
321            })
322            .call(uri),
323            listener.accept_stream().try_next(),
324        )
325        .await
326        .unwrap();
327
328        let tcp_stream = io.into_inner();
329        let stream = socket2::SockRef::from(tcp_stream.stream.std());
330
331        assert_matches!(stream.keepalive(), Ok(v) if v);
332        assert_matches!(stream.tcp_keepalive_time(), Ok(v) if v == idle);
333        assert_matches!(stream.tcp_keepalive_interval(), Ok(v) if v == interval);
334        assert_matches!(stream.tcp_keepalive_retries(), Ok(v) if v == count);
335        assert_matches!(stream.recv_buffer_size(), Ok(v) if v == rcvbuf);
336    }
337
338    #[fasync::run_singlethreaded(test)]
339    async fn test_parse_ipv6_addr_with_provider() {
340        let expected = "fe80::1:2:3:4".parse::<Ipv6Addr>().unwrap();
341
342        assert_matches!(
343            parse_ip_addr_with_provider(&PanicConnector, "[fe80::1:2:3:4%250]", 8080).await,
344            Ok(Some(addr)) if addr == SocketAddr::V6(SocketAddrV6::new(expected, 8080, 0, 0))
345        );
346
347        assert_matches!(
348            parse_ip_addr_with_provider(&PanicConnector, "[fe80::1:2:3:4%252]", 8080).await,
349            Ok(Some(addr)) if addr == SocketAddr::V6(SocketAddrV6::new(expected, 8080, 0, 2))
350        );
351    }
352
353    #[fasync::run_singlethreaded(test)]
354    async fn test_parse_ipv6_addr_with_provider_supports_interface_names() {
355        let connector = RealServiceConnector::new();
356        let expected = "fe80::1:2:3:4".parse::<Ipv6Addr>().unwrap();
357
358        assert_matches!(
359            parse_ip_addr_with_provider(&connector, "[fe80::1:2:3:4%25lo]", 8080).await,
360            Ok(Some(addr)) if addr == SocketAddr::V6(SocketAddrV6::new(expected, 8080, 0, 1))
361        );
362
363        assert_matches!(
364            parse_ip_addr_with_provider(&connector, "[fe80::1:2:3:4%25]", 8080).await,
365            Err(err) if err.kind() == io::ErrorKind::NotFound
366        );
367
368        assert_matches!(
369            parse_ip_addr_with_provider(&connector, "[fe80::1:2:3:4%25unknownif]", 8080).await,
370            Err(err) if err.kind() == io::ErrorKind::NotFound
371        );
372    }
373
374    #[fasync::run_singlethreaded(test)]
375    async fn test_parse_ipv6_addr_handles_connection_errors() {
376        struct ErrorConnector;
377
378        impl ProviderConnector for ErrorConnector {
379            fn connect(&self) -> Result<ClientEnd<ProviderMarker>, io::Error> {
380                Err(io::Error::other("something bad happened"))
381            }
382        }
383
384        assert_matches!(parse_ip_addr_with_provider(&ErrorConnector, "[fe80::1:2:3:4%25lo]", 8080).await,
385            Err(err) if err.kind() == io::ErrorKind::Other);
386    }
387
388    #[fasync::run_singlethreaded(test)]
389    async fn test_parse_ipv6_addr_handles_large_interface_indices() {
390        let (client_end, mut stream) = create_request_stream::<ProviderMarker>();
391
392        let provider_fut = async move {
393            while let Some(req) = stream.try_next().await.unwrap_or(None) {
394                match req {
395                    ProviderRequest::InterfaceNameToIndex { name: _, responder } => {
396                        responder.send(Ok(u64::MAX)).unwrap()
397                    }
398                    _ => panic!("unexpected request"),
399                }
400            }
401        };
402
403        struct ErrorConnector {
404            client_end: RefCell<Option<ClientEnd<ProviderMarker>>>,
405        }
406
407        impl ProviderConnector for ErrorConnector {
408            fn connect(&self) -> Result<ClientEnd<ProviderMarker>, io::Error> {
409                let client_end = self.client_end.borrow_mut().take().unwrap();
410                Ok(client_end)
411            }
412        }
413
414        let connector = ErrorConnector { client_end: RefCell::new(Some(client_end)) };
415
416        let parse_ip_fut = parse_ip_addr_with_provider(&connector, "[fe80::1:2:3:4%25lo]", 8080);
417
418        // Join the two futures to make sure they both complete.
419        let ((), res) = future::join(provider_fut, parse_ip_fut).await;
420
421        assert_matches!(res, Err(err) if err.kind() == io::ErrorKind::Other);
422    }
423
424    struct ProxyConnector<T> {
425        proxy: T,
426    }
427
428    impl LookupConnector for ProxyConnector<LookupProxy> {
429        fn connect(&self) -> Result<LookupProxy, io::Error> {
430            Ok(self.proxy.clone())
431        }
432    }
433
434    #[fasync::run_singlethreaded(test)]
435    async fn test_resolve_ip_addr() {
436        let (sender, receiver) =
437            futures::channel::mpsc::unbounded::<Result<LookupResult, LookupError>>();
438        let (proxy, stream) = create_proxy_and_stream::<LookupMarker>();
439        const TEST_HOSTNAME: &'static str = "foobar.com";
440        let name_lookup_fut = stream.zip(receiver).for_each(|(req, rsp)| match req {
441            Ok(LookupRequest::LookupIp { hostname, options, responder }) => {
442                assert_eq!(hostname.as_str(), TEST_HOSTNAME);
443                assert_eq!(
444                    options,
445                    LookupIpOptions {
446                        ipv4_lookup: Some(true),
447                        ipv6_lookup: Some(true),
448                        sort_addresses: Some(true),
449                        ..Default::default()
450                    }
451                );
452                let rsp = rsp.as_ref().map_err(|e| *e);
453                futures::future::ready(responder.send(rsp).expect("failed to send FIDL response"))
454            }
455            req => panic!("unexpected item in request stream {:?}", req),
456        });
457
458        let connector = ProxyConnector { proxy };
459
460        let ip_v4 = Ipv4Addr::LOCALHOST.into();
461        let ip_v6 = Ipv6Addr::LOCALHOST.into();
462        const PORT1: u16 = 1234;
463        const PORT2: u16 = 4321;
464
465        let test_fut = async move {
466            // Test expectation's error variant is a tuple of the lookup error
467            // to inject and the expected io error kind returned.
468            type Expectation = Result<Vec<std::net::IpAddr>, (LookupError, io::ErrorKind)>;
469            let test_resolve = |port, expect: Expectation| {
470                let fidl_response = expect
471                    .clone()
472                    .map(|addrs| LookupResult {
473                        addresses: Some(
474                            addrs
475                                .into_iter()
476                                .map(|std| fidl_fuchsia_net_ext::IpAddress(std).into())
477                                .collect(),
478                        ),
479                        ..Default::default()
480                    })
481                    .map_err(|(fidl_err, _io_err)| fidl_err);
482                let expect = expect
483                    .map(|addrs| {
484                        addrs.into_iter().map(|addr| SocketAddr::new(addr, port)).collect()
485                    })
486                    .map_err(|(_fidl_err, io_err)| io_err);
487                let () = sender.unbounded_send(fidl_response).expect("failed to send expectation");
488                resolve_ip_addr(&connector, TEST_HOSTNAME, port)
489                    .map_ok(Iterator::collect::<Vec<_>>)
490                    // Map IO error to kind so we can do equality.
491                    .map_err(|err| err.kind())
492                    .map(move |result| {
493                        assert_eq!(result, expect);
494                    })
495            };
496            let () = test_resolve(PORT1, Ok(vec![ip_v4])).await;
497            let () = test_resolve(PORT2, Ok(vec![ip_v6])).await;
498            let () = test_resolve(PORT1, Ok(vec![ip_v4, ip_v6])).await;
499            let () = test_resolve(PORT1, Err((LookupError::NotFound, io::ErrorKind::Other))).await;
500        };
501
502        let ((), ()) = futures::future::join(name_lookup_fut, test_fut).await;
503    }
504}
fuchsia_hyper/fuchsia.rs

fuchsia_hyper/
fuchsia.rs
