taba_client/transport/
mod.rs

1//! The transport: carries the request bytes to the server and brings the response back. The core (building and reading
2//! HTTP) is in `http`; here there is only TCP and TLS (`rustls`, TLS 1.2 or later).
3
4use std::io::{self, BufReader, Read, Write};
5use std::net::{TcpStream, ToSocketAddrs};
6use std::sync::{Arc, Mutex};
7use std::time::{Duration, Instant};
8
9use rustls::pki_types::pem::PemObject;
10use rustls::pki_types::{CertificateDer, ServerName};
11use rustls::{ClientConfig, ClientConnection, RootCertStore, StreamOwned};
12
13use crate::http::{self, ReadError, Request, Response, Url};
14
15#[derive(Debug, Clone, Copy)]
16pub struct Timeouts {
17    /// To open the connection.
18    pub connect: Duration,
19    /// Total of the call, from the open connection until the response is read.
20    pub total: Duration,
21}
22
23/// Why the call failed, and whether the server may have received the request.
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum TransportError {
26    /// Failure before sending any byte of the request (DNS, refusal, TLS handshake).
27    Connect(String),
28    /// Time ran out; `sent` says whether the request had already been written.
29    Timeout { sent: bool },
30    /// Network failure after starting to send.
31    Io(String),
32    /// The response is not valid HTTP.
33    Protocol(String),
34}
35
36pub trait Transport: Send + Sync {
37    fn send(
38        &self,
39        url: &Url,
40        request: &Request,
41        timeouts: &Timeouts,
42    ) -> Result<Response, TransportError>;
43}
44
45/// Plain TCP or with `rustls`. One connection per call.
46pub struct TcpTransport {
47    ca_pem: Option<Vec<u8>>,
48    tls: Mutex<Option<Arc<ClientConfig>>>,
49}
50
51impl TcpTransport {
52    /// `ca_pem` is a CA certificate (PEM) to trust, besides the system roots.
53    pub fn new(ca_pem: Option<Vec<u8>>) -> Self {
54        TcpTransport {
55            ca_pem,
56            tls: Mutex::new(None),
57        }
58    }
59
60    fn tls_config(&self) -> Result<Arc<ClientConfig>, TransportError> {
61        let mut cached = self.tls.lock().unwrap_or_else(|e| e.into_inner());
62        if let Some(config) = cached.as_ref() {
63            return Ok(config.clone());
64        }
65        let config =
66            Arc::new(build_tls_config(self.ca_pem.as_deref()).map_err(TransportError::Connect)?);
67        *cached = Some(config.clone());
68        Ok(config)
69    }
70}
71
72/// The TLS configuration: the system roots plus the given CA. It never turns certificate
73/// verification off (UC-CNT-030).
74pub fn build_tls_config(ca_pem: Option<&[u8]>) -> Result<ClientConfig, String> {
75    let mut roots = RootCertStore::empty();
76    for cert in rustls_native_certs::load_native_certs().certs {
77        let _ = roots.add(cert);
78    }
79    if let Some(pem) = ca_pem {
80        let mut found = 0;
81        for cert in CertificateDer::pem_slice_iter(pem) {
82            let cert = cert.map_err(|e| format!("invalid CA certificate: {e}"))?;
83            roots
84                .add(cert)
85                .map_err(|e| format!("CA certificate refused: {e}"))?;
86            found += 1;
87        }
88        if found == 0 {
89            return Err("the CA file has no PEM certificate".into());
90        }
91    }
92    let provider = Arc::new(rustls::crypto::ring::default_provider());
93    let builder = ClientConfig::builder_with_provider(provider)
94        .with_protocol_versions(&[&rustls::version::TLS13, &rustls::version::TLS12])
95        .map_err(|e| e.to_string())?;
96    Ok(builder.with_root_certificates(roots).with_no_client_auth())
97}
98
99/// Reads and writes with a single deadline for the whole call.
100struct Deadlined {
101    sock: TcpStream,
102    deadline: Instant,
103}
104
105impl Deadlined {
106    fn remaining(&self) -> io::Result<Duration> {
107        let left = self.deadline.saturating_duration_since(Instant::now());
108        if left.is_zero() {
109            Err(io::ErrorKind::TimedOut.into())
110        } else {
111            Ok(left)
112        }
113    }
114}
115
116impl Read for Deadlined {
117    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
118        self.sock.set_read_timeout(Some(self.remaining()?))?;
119        self.sock.read(buf)
120    }
121}
122
123impl Write for Deadlined {
124    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
125        self.sock.set_write_timeout(Some(self.remaining()?))?;
126        self.sock.write(buf)
127    }
128
129    fn flush(&mut self) -> io::Result<()> {
130        self.sock.flush()
131    }
132}
133
134fn is_timeout(e: &io::Error) -> bool {
135    matches!(
136        e.kind(),
137        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
138    )
139}
140
141fn connect(url: &Url, timeouts: &Timeouts) -> Result<TcpStream, TransportError> {
142    let addrs = (url.host.as_str(), url.port)
143        .to_socket_addrs()
144        .map_err(|e| TransportError::Connect(format!("could not find {}: {e}", url.host)))?;
145    let mut last = None;
146    for addr in addrs {
147        match TcpStream::connect_timeout(&addr, timeouts.connect) {
148            Ok(sock) => {
149                let _ = sock.set_nodelay(true);
150                return Ok(sock);
151            }
152            Err(e) if is_timeout(&e) => return Err(TransportError::Timeout { sent: false }),
153            Err(e) => last = Some(e),
154        }
155    }
156    Err(TransportError::Connect(match last {
157        Some(e) => format!("could not connect to {}:{}: {e}", url.host, url.port),
158        None => format!("{} has no address", url.host),
159    }))
160}
161
162impl Transport for TcpTransport {
163    fn send(
164        &self,
165        url: &Url,
166        request: &Request,
167        timeouts: &Timeouts,
168    ) -> Result<Response, TransportError> {
169        let sock = connect(url, timeouts)?;
170        let mut stream = Deadlined {
171            sock,
172            deadline: Instant::now() + timeouts.total,
173        };
174        let bytes = http::encode_request(&url.host_header(), request);
175        if url.https {
176            let name = ServerName::try_from(url.host.clone()).map_err(|_| {
177                TransportError::Connect(format!("`{}` is not a valid server name", url.host))
178            })?;
179            let mut conn = ClientConnection::new(self.tls_config()?, name)
180                .map_err(|e| TransportError::Connect(e.to_string()))?;
181            // The handshake ends before any byte of the request: failing here is "did not send".
182            while conn.is_handshaking() {
183                conn.complete_io(&mut stream).map_err(|e| {
184                    if is_timeout(&e) {
185                        TransportError::Timeout { sent: false }
186                    } else {
187                        TransportError::Connect(format!("TLS failure: {e}"))
188                    }
189                })?;
190            }
191            exchange(StreamOwned::new(conn, stream), &bytes)
192        } else {
193            exchange(stream, &bytes)
194        }
195    }
196}
197
198fn exchange<S: Read + Write>(mut stream: S, bytes: &[u8]) -> Result<Response, TransportError> {
199    if let Err(e) = stream.write_all(bytes).and_then(|_| stream.flush()) {
200        return Err(if is_timeout(&e) {
201            TransportError::Timeout { sent: true }
202        } else {
203            TransportError::Io(format!("failed to send: {e}"))
204        });
205    }
206    let mut reader = BufReader::new(stream);
207    match http::read_response(&mut reader) {
208        Ok(response) => Ok(response),
209        Err(ReadError::Io { error, .. }) if is_timeout(&error) => {
210            Err(TransportError::Timeout { sent: true })
211        }
212        Err(ReadError::Io { error, .. }) => Err(TransportError::Io(format!(
213            "failed to read the response: {error}"
214        ))),
215        Err(ReadError::Malformed(why)) => Err(TransportError::Protocol(why.to_owned())),
216    }
217}
218
219#[cfg(test)]
220mod tests;
taba_client/transport/mod.rs

taba_client/transport/
mod.rs
