containers_api/conn/
transport.rs

1//! Transports for communicating with the Podman or Docker daemon
2
3use crate::conn::{Error, Headers, Payload, Result};
4
5use futures_util::{
6    stream::{self, Stream},
7    StreamExt,
8};
9use hyper::{
10    body::Bytes,
11    client::{Client, HttpConnector},
12    header, Body, Method, Request, Response,
13};
14#[cfg(feature = "tls")]
15use hyper_openssl::HttpsConnector;
16#[cfg(unix)]
17use hyperlocal::UnixConnector;
18#[cfg(unix)]
19use hyperlocal::Uri as DomainUri;
20use url::Url;
21
22use std::{iter::IntoIterator, path::PathBuf};
23
24/// Transports are types which define supported means of communication.
25#[derive(Clone, Debug)]
26pub enum Transport {
27    /// A network tcp interface
28    Tcp {
29        client: Client<HttpConnector>,
30        host: Url,
31    },
32    /// TCP/TLS
33    #[cfg(feature = "tls")]
34    #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
35    EncryptedTcp {
36        client: Client<HttpsConnector<HttpConnector>>,
37        host: Url,
38    },
39    /// A Unix domain socket
40    #[cfg(unix)]
41    Unix {
42        client: Client<UnixConnector>,
43        path: PathBuf,
44    },
45}
46
47impl Transport {
48    pub fn remote_addr(&self) -> &str {
49        match &self {
50            Self::Tcp { ref host, .. } => host.as_ref(),
51            #[cfg(feature = "tls")]
52            Self::EncryptedTcp { ref host, .. } => host.as_ref(),
53            #[cfg(unix)]
54            Self::Unix { ref path, .. } => path.to_str().unwrap_or_default(),
55        }
56    }
57
58    pub fn make_uri(&self, ep: &str) -> Result<hyper::Uri> {
59        match self {
60            Transport::Tcp { host, .. } => format!("{host}{ep}").parse().map_err(Error::InvalidUri),
61            #[cfg(feature = "tls")]
62            Transport::EncryptedTcp { host, .. } => {
63                format!("{host}{ep}").parse().map_err(Error::InvalidUri)
64            }
65            #[cfg(unix)]
66            Transport::Unix { path, .. } => Ok(DomainUri::new(path, ep).into()),
67        }
68    }
69
70    /// Send the given request and return a Future of the response.
71    pub async fn request(&self, req: Request<Body>) -> Result<Response<Body>> {
72        log::trace!("sending request {} {}", req.method(), req.uri());
73        match self {
74            Transport::Tcp { ref client, .. } => client.request(req),
75            #[cfg(feature = "tls")]
76            Transport::EncryptedTcp { ref client, .. } => client.request(req),
77            #[cfg(unix)]
78            Transport::Unix { ref client, .. } => client.request(req),
79        }
80        .await
81        .map_err(Error::from)
82    }
83
84    pub async fn request_string(&self, req: Request<Body>) -> Result<String> {
85        let body = self.request(req).await.map(|resp| resp.into_body())?;
86        body_to_string(body).await
87    }
88}
89
90pub(crate) async fn body_to_string(body: Body) -> Result<String> {
91    let bytes = hyper::body::to_bytes(body).await?;
92    String::from_utf8(bytes.to_vec()).map_err(Error::from)
93}
94
95/// Builds an HTTP request.
96pub(crate) fn build_request<B>(
97    method: Method,
98    uri: hyper::Uri,
99    body: Payload<B>,
100    headers: Option<Headers>,
101) -> Result<Request<Body>>
102where
103    B: Into<Body>,
104{
105    let builder = hyper::http::request::Builder::new();
106    let req = builder.method(method).uri(&uri);
107    let mut req = req.header(header::HOST, "");
108
109    if let Some(h) = headers {
110        for (k, v) in h.into_iter() {
111            req = req.header(k, v);
112        }
113    }
114
115    // early return
116    if body.is_none() {
117        return Ok(req.body(Body::empty())?);
118    }
119
120    let mime = body.mime_type();
121    if let Some(c) = mime {
122        req = req.header(header::CONTENT_TYPE, &c.to_string());
123    }
124
125    // it's ok to unwrap, we check that the body is not none
126    req.body(body.into_inner().unwrap().into())
127        .map_err(Error::from)
128}
129
130pub(crate) async fn get_response_string(response: Response<Body>) -> Result<String> {
131    body_to_string(response.into_body()).await
132}
133
134pub(crate) fn stream_response(response: Response<Body>) -> impl Stream<Item = Result<Bytes>> {
135    stream_body(response.into_body())
136}
137
138pub(crate) fn stream_json_response(response: Response<Body>) -> impl Stream<Item = Result<Bytes>> {
139    stream_json_body(response.into_body())
140}
141
142fn stream_body(body: Body) -> impl Stream<Item = Result<Bytes>> {
143    async fn unfold(mut body: Body) -> Option<(Result<Bytes>, Body)> {
144        body.next()
145            .await
146            .map(|chunk| (chunk.map_err(Error::from), body))
147    }
148
149    stream::unfold(body, unfold)
150}
151
152static JSON_WHITESPACE: &[u8] = b"\r\n";
153
154fn stream_json_body(body: Body) -> impl Stream<Item = Result<Bytes>> {
155    async fn unfold(mut body: Body) -> Option<(Result<Bytes>, Body)> {
156        let mut chunk = Vec::new();
157        while let Some(chnk) = body.next().await {
158            match chnk {
159                Ok(chnk) => {
160                    chunk.extend(chnk.to_vec());
161                    if chnk.ends_with(JSON_WHITESPACE) {
162                        break;
163                    }
164                }
165                Err(e) => {
166                    return Some((Err(Error::from(e)), body));
167                }
168            }
169        }
170
171        if chunk.is_empty() {
172            return None;
173        }
174
175        Some((Ok(Bytes::from(chunk)), body))
176    }
177
178    stream::unfold(body, unfold)
179}
containers_api/conn/transport.rs

containers_api/conn/
transport.rs
