blitz_net/
tls.rs

1//! Native trust anchors and nagoya socket transport.
2//!
3//! DNS always runs on a nagoya worker, including when the fetch future is
4//! polled directly by a caller or a local reactor.
5
6use super::{Error, pool::Origin};
7use blitz_traits::tls_policy::TlsPolicyError;
8use bytes::BytesMut;
9use nago_rustls::{
10    TlsSession,
11    rustls::{ClientConfig, ClientConnection, RootCertStore},
12    rustls_pki_types::ServerName,
13};
14use nagoya::{net::TcpStream, reactor::Handle};
15use ps_rustcrypto::policy::{self, ct::CtVerifier};
16use std::{
17    collections::HashMap,
18    sync::{Arc, Mutex, OnceLock},
19};
20
21const READ_CHUNK: usize = 16 * 1024;
22
23pub(super) fn default_config() -> Result<Arc<ClientConfig>, Error> {
24    static CONFIG: OnceLock<Result<Arc<ClientConfig>, String>> = OnceLock::new();
25    match CONFIG.get_or_init(build_config) {
26        Ok(config) => Ok(config.clone()),
27        Err(error) => Err(Error::Tls(error.clone())),
28    }
29}
30
31fn native_roots() -> Result<RootCertStore, String> {
32    let native = rustls_native_certs::load_native_certs();
33    let mut roots = RootCertStore::empty();
34    let (accepted, rejected) = roots.add_parsable_certificates(native.certs);
35    if accepted == 0 {
36        return Err(format!(
37            "no usable native TLS trust anchors: {} loading errors, {rejected} rejected certificates",
38            native.errors.len(),
39        ));
40    }
41    if !native.errors.is_empty() || rejected != 0 {
42        eprintln!(
43            "TLS trust store: {} loading errors, {rejected} rejected certificates",
44            native.errors.len(),
45        );
46    }
47    Ok(roots)
48}
49
50/// The native roots that are also built-in public roots, matched by subject
51/// and key. Like Chrome, CT is enforced only on chains ending in one of
52/// these; locally added anchors (enterprise or test CAs) skip it.
53fn public_roots(native: &RootCertStore) -> RootCertStore {
54    let mut public = RootCertStore::empty();
55    public.roots = native
56        .roots
57        .iter()
58        .filter(|root| {
59            webpki_roots::TLS_SERVER_ROOTS.iter().any(|built_in| {
60                root.subject.as_ref() == built_in.subject.as_ref()
61                    && root.subject_public_key_info.as_ref()
62                        == built_in.subject_public_key_info.as_ref()
63            })
64        })
65        .cloned()
66        .collect();
67    public
68}
69
70/// HTTP/2 preferred; HTTP/1.1 for servers that do not offer it.
71pub(super) fn alpn() -> Vec<Vec<u8>> {
72    vec![b"h2".to_vec(), b"http/1.1".to_vec()]
73}
74
75fn build_config() -> Result<Arc<ClientConfig>, String> {
76    let roots = native_roots()?;
77    let public = public_roots(&roots);
78    // Explicit even when an embedder installed another process default.
79    let mut provider = ps_rustcrypto::provider();
80    // Keep the wire offer explicit even if a later provider adds suites.
81    provider.cipher_suites = vec![
82        ps_rustcrypto::TLS13_CHACHA20_POLY1305_SHA256,
83        ps_rustcrypto::TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
84    ];
85    let provider = Arc::new(provider);
86    let mut config = ClientConfig::builder_with_provider(provider)
87        .with_safe_default_protocol_versions()
88        .map_err(|error| error.to_string())?
89        // ECC-only: a chain holding an RSA key anywhere, anchor included, is
90        // refused with unsupported_certificate before any signature check.
91        // A valid chain ending in a public root must then be publicly
92        // logged under the CT policy.
93        .dangerous()
94        .with_custom_certificate_verifier(
95            CtVerifier::ecc_only(roots, public)
96                .map_err(|error| format!("TLS trust store: {error}"))?,
97        )
98        .with_no_client_auth();
99    config.alpn_protocols = alpn();
100    Ok(Arc::new(config))
101}
102
103/// Connection counters and the latest TLS policy refusal for each host.
104static HOSTS: Mutex<Option<HashMap<String, HostStats>>> = Mutex::new(None);
105
106/// Connection counters for one host, since the process started.
107#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
108pub struct HostStats {
109    /// New TCP connections opened, TLS attempts included.
110    pub connections: u32,
111    /// Handshakes refused under the strict TLS policy.
112    pub refusals: u32,
113    /// TLS handshakes that completed under the strict policy.
114    pub handshakes: u32,
115    /// The latest TLS policy refusal, cleared by the next TLS attempt.
116    pub last_tls_policy_error: Option<TlsPolicyError>,
117    /// The protocol the latest connection to this host settled on: `h2`,
118    /// `h2 (coalesced)` onto another host's connection, or `http/1.1`.
119    pub last_alpn: Option<&'static str>,
120}
121
122fn update(host: &str, change: impl FnOnce(&mut HostStats)) -> HostStats {
123    let Ok(mut map) = HOSTS.lock() else {
124        return HostStats::default();
125    };
126    let stats = map
127        .get_or_insert_with(HashMap::new)
128        .entry(host.to_ascii_lowercase())
129        .or_default();
130    change(stats);
131    *stats
132}
133
134pub(super) fn record_alpn(host: &str, alpn: &'static str) {
135    update(host, |stats| stats.last_alpn = Some(alpn));
136}
137
138/// Connection counters for every host contacted so far, sorted by host.
139pub fn connection_stats() -> Vec<(String, HostStats)> {
140    let mut all: Vec<_> = HOSTS
141        .lock()
142        .ok()
143        .and_then(|map| {
144            map.as_ref()
145                .map(|map| map.iter().map(|(h, s)| (h.clone(), *s)).collect())
146        })
147        .unwrap_or_default();
148    all.sort_by(|a, b| a.0.cmp(&b.0));
149    all
150}
151
152pub(super) async fn open(
153    origin: &Origin,
154    handle: &Handle,
155    config: Arc<ClientConfig>,
156) -> Result<Transport, Error> {
157    if !origin.tls {
158        return Ok(Transport::Plain(connect(origin, handle).await?));
159    }
160    update(&origin.host, |stats| stats.last_tls_policy_error = None);
161    // One connection and one policy. A refusal is terminal for this request.
162    match handshake(origin, handle, config).await {
163        Ok(tls) => {
164            update(&origin.host, |stats| stats.handshakes += 1);
165            Ok(Transport::Tls(tls))
166        }
167        Err(error) => {
168            if let Error::TlsPolicy(reason) = &error {
169                update(&origin.host, |stats| {
170                    stats.refusals += 1;
171                    stats.last_tls_policy_error = Some(*reason);
172                });
173            }
174            // Resource loads may discard the returned error, so log it here.
175            info(format_args!("TLS {}: {error}", origin.host));
176            Err(error)
177        }
178    }
179}
180
181fn info(message: std::fmt::Arguments<'_>) {
182    #[cfg(feature = "tracing")]
183    tracing::info!("{message}");
184    #[cfg(not(feature = "tracing"))]
185    eprintln!("info: {message}");
186}
187
188async fn connect(origin: &Origin, handle: &Handle) -> Result<TcpStream, Error> {
189    let host = origin.host.clone();
190    let port = origin.port;
191    let addresses = nagoya::spawn(async move {
192        nagoya::net::resolve(&host, port).map_err(|error| Error::Dns(error.to_string()))
193    })
194    .await
195    .ok_or_else(|| Error::Dns("resolver task was cancelled".to_owned()))??;
196
197    let stats = update(&origin.host, |stats| stats.connections += 1);
198    info(format_args!(
199        "net {}: new connection #{} (tls {})",
200        origin.host, stats.connections, origin.tls
201    ));
202    nagoya::net::connect_any(&addresses, handle)
203        .await
204        .map_err(|error| Error::Io(error.to_string()))
205}
206
207/// One TLS connection attempt, retaining the reason for any policy refusal.
208async fn handshake(
209    origin: &Origin,
210    handle: &Handle,
211    config: Arc<ClientConfig>,
212) -> Result<Box<TlsSession<TcpStream>>, Error> {
213    let stream = connect(origin, handle).await?;
214    let name = ServerName::try_from(origin.host.to_string())
215        .map_err(|_| Error::InvalidRequest("invalid TLS server name"))?;
216    let session =
217        ClientConnection::new(config, name).map_err(|error| Error::Tls(error.to_string()))?;
218    let mut tls = TlsSession::client(stream, session);
219    if let Err(errno) = tls.handshake().await {
220        return Err(handshake_failure(tls.tls_error(), errno));
221    }
222    if tls
223        .alpn_protocol()
224        .is_some_and(|protocol| protocol != b"http/1.1" && protocol != b"h2")
225    {
226        return Err(Error::Tls(
227            "peer negotiated an unsupported ALPN protocol".to_owned(),
228        ));
229    }
230    Ok(Box::new(tls))
231}
232
233/// Preserve the observed reason without probing a second, weaker connection.
234///
235/// An alert 40 or 71 does not say whether the server needs RSA, AES, or some
236/// other algorithm. Report the rejected profile without inventing that fact.
237fn handshake_failure(
238    error: Option<&nago_rustls::rustls::Error>,
239    errno: impl std::fmt::Display,
240) -> Error {
241    use nago_rustls::rustls::Error as TlsError;
242    use nago_rustls::rustls::{
243        AlertDescription, CertificateError, PeerIncompatible, PeerMisbehaved,
244    };
245
246    let reason = match error {
247        Some(error) if policy::is_rsa_refusal(error) => TlsPolicyError::RsaCertificateChain,
248        Some(error) if policy::is_key_type_refusal(error) => TlsPolicyError::CertificateKey,
249        Some(TlsError::InvalidCertificate(
250            CertificateError::UnsupportedSignatureAlgorithmContext { .. }
251            | CertificateError::UnsupportedSignatureAlgorithmForPublicKeyContext { .. },
252        )) => TlsPolicyError::CertificateSignature,
253        Some(TlsError::PeerMisbehaved(PeerMisbehaved::SelectedUnofferedCipherSuite)) => {
254            TlsPolicyError::CipherSuite
255        }
256        Some(TlsError::PeerMisbehaved(
257            PeerMisbehaved::SignedHandshakeWithUnadvertisedSigScheme,
258        ))
259        | Some(TlsError::PeerIncompatible(PeerIncompatible::NoSignatureSchemesInCommon)) => {
260            TlsPolicyError::HandshakeSignature
261        }
262        Some(TlsError::AlertReceived(AlertDescription::HandshakeFailure)) => {
263            TlsPolicyError::PeerRejectedAlgorithms { alert: 40 }
264        }
265        Some(TlsError::AlertReceived(AlertDescription::InsufficientSecurity)) => {
266            TlsPolicyError::PeerRejectedAlgorithms { alert: 71 }
267        }
268        Some(error) if policy::ct::is_not_logged(error) => {
269            return Error::Tls("certificate not publicly logged".to_owned());
270        }
271        Some(error) => return Error::Tls(format!("handshake: {error}")),
272        None => return Error::Tls(format!("handshake: {errno}")),
273    };
274    Error::TlsPolicy(reason)
275}
276
277pub(super) enum Transport {
278    Plain(TcpStream),
279    Tls(Box<TlsSession<TcpStream>>),
280}
281
282impl Transport {
283    pub async fn read_into(&mut self, buffer: &mut BytesMut) -> Result<usize, Error> {
284        match self {
285            Self::Plain(stream) => {
286                // Read directly into the destination's spare capacity.
287                buffer.reserve(READ_CHUNK);
288                stream
289                    .read_buf(buffer)
290                    .await
291                    .map_err(|error| Error::Io(error.to_string()))
292            }
293            Self::Tls(stream) => {
294                // TlsSession exposes an initialised plaintext slice rather
295                // than a read_buf operation. Avoid a second staging buffer.
296                let start = buffer.len();
297                buffer.resize(start + READ_CHUNK, 0);
298                match stream.read(&mut buffer[start..]).await {
299                    Ok(read) => {
300                        buffer.truncate(start + read);
301                        Ok(read)
302                    }
303                    Err(error) => {
304                        buffer.truncate(start);
305                        Err(Error::Io(error.to_string()))
306                    }
307                }
308            }
309        }
310    }
311
312    pub async fn write_all(&mut self, bytes: &[u8]) -> Result<(), Error> {
313        match self {
314            Self::Plain(stream) => stream
315                .write_all(bytes)
316                .await
317                .map_err(|error| Error::Write(error.to_string())),
318            Self::Tls(stream) => stream
319                .write_all(bytes)
320                .await
321                .map_err(|error| Error::Write(error.to_string())),
322        }
323    }
324}
blitz_net/tls.rs

blitz_net/
tls.rs
