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

fuchsia_hyper/
fuchsia.rs
