1//! The connection to a relay, wherever the network lets one through: straight to it, or
2//! through the proxy `proxy` names by `CONNECT`, over TLS that trusts what the system trusts
3//! (so a proxy that inspects HTTPS with its own authority works once the system trusts it);
4//! as a WebSocket, or where something on the way refuses WebSockets, as HTTPS requests the
5//! relay answers the same messages over (`GET` waits for what is to come, `POST` sends). What
6//! fails is told apart, so a person can be told what to try.
7
8use super::proxy::{self, Proxy};
9use ::relay::ws;
10use base64::Engine;
11use rustls::{ClientConfig, ClientConnection, RootCertStore, pki_types::ServerName};
12use std::{
13 io::{self, BufReader, Read, Write},
14 net::{Shutdown, SocketAddr, TcpStream, ToSocketAddrs},
15 sync::{
16 Arc, Condvar, Mutex, OnceLock,
17 atomic::{AtomicBool, Ordering},
18 },
19 thread,
20 time::{Duration, Instant},
21};
22
23const CONNECT: Duration = Duration::from_secs(10);
24/// How long a connection may hear nothing: the relay hears a ping every 30 s and answers.
25pub(super) const QUIET: Duration = Duration::from_secs(75);
26/// The most a poll's answer holds.
27const MOST: usize = 4 << 20;
28/// Set once WebSockets were refused, so later connections go straight to polling.
29static POLLING: AtomicBool = AtomicBool::new(false);
30
31pub use super::model::Trouble;
32
33pub(super) enum Failure {
34 Trouble(Trouble, io::Error),
35 /// The relay answered, with this HTTP status and how long to wait.
36 Refused(u16, Option<Duration>),
37}
38
39impl Failure {
40 fn other(error: io::Error) -> Self {
41 Failure::Trouble(Trouble::Other, error)
42 }
43}
44
45impl From<io::Error> for Failure {
46 fn from(error: io::Error) -> Self {
47 Failure::other(error)
48 }
49}
50
51/// A relay's address: `ws://` or `wss://`, a host, and the path the relay's `/v1/` follows.
52#[derive(Debug, PartialEq)]
53pub(super) struct Address {
54 pub tls: bool,
55 /// The host and port as the URL gave them, for the `Host` header.
56 pub authority: String,
57 pub host: String,
58 pub port: u16,
59 pub path: String,
60}
61
62pub(super) fn parse(url: &str) -> io::Result<Address> {
63 let bad = || io::Error::new(io::ErrorKind::InvalidInput, format!("Not a relay: {url}"));
64 let (tls, rest) = match url.split_once("://") {
65 Some(("wss", rest)) => (true, rest),
66 Some(("ws", rest)) => (false, rest),
67 _ => return Err(bad()),
68 };
69 let (authority, path) = rest.split_at(rest.find('/').unwrap_or(rest.len()));
70 let (host, port) = match authority.rsplit_once(':') {
71 Some((host, port)) if !port.contains(']') => (host, port.parse().map_err(|_| bad())?),
72 _ => (authority, if tls { 443 } else { 80 }),
73 };
74 let host = host.trim_start_matches('[').trim_end_matches(']');
75 if host.is_empty() {
76 return Err(bad());
77 }
78 Ok(Address {
79 tls,
80 authority: authority.into(),
81 host: host.into(),
82 port,
83 path: path.trim_end_matches('/').into(),
84 })
85}
86
87/// A connection to the relay: what it writes the relay's messages into, what it reads them
88/// from, and how to hang up.
89pub(super) struct Connection {
90 pub writer: Box<dyn Write + Send>,
91 pub reader: Box<dyn Read + Send>,
92 pub close: Box<dyn Fn() + Send + Sync>,
93}
94
95/// Opens the relay's `path` at `address`, as a WebSocket, else as HTTPS requests where
96/// something on the way refused the WebSocket.
97pub(super) fn connect(address: &Address, path: &str) -> Result<Connection, Failure> {
98 if !POLLING.load(Ordering::Acquire) {
99 match websocket(address, path) {
100 Err(Failure::Refused(status, _)) if !relay_status(status) => {
101 POLLING.store(true, Ordering::Release);
102 }
103 Err(Failure::Trouble(Trouble::Other, _)) => {
104 POLLING.store(true, Ordering::Release);
105 }
106 connected => return connected,
107 }
108 }
109 poll(address, path).map_err(|failure| match failure {
110 Failure::Refused(status, _) if !relay_status(status) => Failure::Trouble(
111 Trouble::Blocked,
112 io::Error::other(format!("Refused with {status}")),
113 ),
114 failure => failure,
115 })
116}
117
118/// Whether the relay itself answers with `status`, rather than something on the way.
119fn relay_status(status: u16) -> bool {
120 matches!(status, 404 | 410 | 429 | 503)
121}
122
123/// A stream to the relay, through the proxy where there is one, over TLS where its address
124/// asks.
125struct Stream {
126 reader: Box<dyn Read + Send>,
127 writer: Box<dyn Write + Send>,
128 tcp: TcpStream,
129}
130
131fn connect_resolved(
132 targets: impl Iterator<Item = SocketAddr>,
133 deadline: Instant,
134 mut connect: impl FnMut(&SocketAddr, Duration) -> io::Result<TcpStream>,
135) -> Result<TcpStream, (Trouble, io::Error)> {
136 let targets: Vec<_> = targets.collect();
137 let mut failure = (Trouble::Dns, io::ErrorKind::NotFound.into());
138 for (at, target) in targets.iter().enumerate() {
139 let remaining = deadline.saturating_duration_since(Instant::now());
140 let attempts = u32::try_from(targets.len() - at).unwrap_or(u32::MAX);
141 let timeout = remaining / attempts;
142 if timeout.is_zero() {
143 return Err((Trouble::TimedOut, io::ErrorKind::TimedOut.into()));
144 }
145 let connected = connect(target, timeout);
146 if Instant::now() >= deadline {
147 return Err((Trouble::TimedOut, io::ErrorKind::TimedOut.into()));
148 }
149 match connected {
150 Ok(tcp) => return Ok(tcp),
151 Err(error) => {
152 let trouble = match error.kind() {
153 io::ErrorKind::TimedOut => Trouble::TimedOut,
154 _ => Trouble::Unreachable,
155 };
156 failure = (trouble, error);
157 }
158 }
159 }
160 Err(failure)
161}
162
163fn open(address: &Address) -> Result<Stream, Failure> {
164 let proxy = proxy::for_host(&address.host, address.tls);
165 let (host, port) = match &proxy {
166 Some(proxy) => (proxy.host.as_str(), proxy.port),
167 None => (address.host.as_str(), address.port),
168 };
169 let reached = |trouble| match proxy {
170 Some(_) => Trouble::ProxyUnreachable,
171 None => trouble,
172 };
173 let targets = (host, port)
174 .to_socket_addrs()
175 .map_err(|error| Failure::Trouble(reached(Trouble::Dns), error))?;
176 let tcp = connect_resolved(
177 targets,
178 Instant::now() + CONNECT,
179 TcpStream::connect_timeout,
180 )
181 .map_err(|(trouble, error)| Failure::Trouble(reached(trouble), error))?;
182 tcp.set_nodelay(true)?;
183 tcp.set_read_timeout(Some(QUIET))?;
184 tcp.set_write_timeout(Some(QUIET))?;
185 if let Some(proxy) = &proxy {
186 tunnel(&tcp, proxy, address)?;
187 }
188 if !address.tls {
189 return Ok(Stream {
190 reader: Box::new(tcp.try_clone()?),
191 writer: Box::new(tcp.try_clone()?),
192 tcp,
193 });
194 }
195 let name = ServerName::try_from(address.host.clone())
196 .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
197 let mut connection = ClientConnection::new(tls(), name).map_err(io::Error::other)?;
198 while connection.is_handshaking() {
199 connection.complete_io(&mut &tcp).map_err(|error| {
200 let rejected = error
201 .get_ref()
202 .and_then(|inner| inner.downcast_ref::<rustls::Error>())
203 .is_some_and(|tls| matches!(tls, rustls::Error::InvalidCertificate(_)));
204 match rejected {
205 true => Failure::Trouble(Trouble::Certificate, error),
206 false => Failure::other(error),
207 }
208 })?;
209 }
210 let connection = Arc::new(Mutex::new(connection));
211 Ok(Stream {
212 reader: Box::new(TlsReader {
213 tcp: tcp.try_clone()?,
214 tls: Arc::clone(&connection),
215 plain: Vec::new(),
216 at: 0,
217 }),
218 writer: Box::new(TlsWriter {
219 tcp: tcp.try_clone()?,
220 tls: connection,
221 }),
222 tcp,
223 })
224}
225
226/// Asks `proxy` on `tcp` to connect through to the relay.
227fn tunnel(tcp: &TcpStream, proxy: &Proxy, address: &Address) -> Result<(), Failure> {
228 let target = match address.host.contains(':') {
229 true => format!("[{}]:{}", address.host, address.port),
230 false => format!("{}:{}", address.host, address.port),
231 };
232 let mut request = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n");
233 if let Some(authorization) = proxy.authorization() {
234 request += &format!("Proxy-Authorization: {authorization}\r\n");
235 }
236 request += "\r\n";
237 (&mut &*tcp)
238 .write_all(request.as_bytes())
239 .map_err(|error| Failure::Trouble(Trouble::ProxyUnreachable, error))?;
240 let head =
241 ws::head(&mut &*tcp).map_err(|error| Failure::Trouble(Trouble::ProxyUnreachable, error))?;
242 match status(&head) {
243 200 => Ok(()),
244 407 => Err(Failure::Trouble(
245 Trouble::ProxyAuthentication,
246 io::Error::new(io::ErrorKind::PermissionDenied, "The proxy asks for a name"),
247 )),
248 status => Err(Failure::Trouble(
249 Trouble::ProxyRefused(status),
250 io::Error::other(format!("The proxy answered {status}")),
251 )),
252 }
253}
254
255fn status(head: &str) -> u16 {
256 head.split(' ')
257 .nth(1)
258 .and_then(|status| status.parse().ok())
259 .unwrap_or(0)
260}
261
262fn retry(head: &str) -> Option<Duration> {
263 ws::header(head, "Retry-After")
264 .and_then(|seconds| seconds.parse().ok())
265 .map(Duration::from_secs)
266}
267
268/// Opens a WebSocket to `path` at `address`.
269fn websocket(address: &Address, path: &str) -> Result<Connection, Failure> {
270 let Stream {
271 reader,
272 mut writer,
273 tcp,
274 } = open(address)?;
275 let mut key = [0; 16];
276 getrandom::fill(&mut key).map_err(|_| io::Error::other("System random source failed"))?;
277 let key = base64::engine::general_purpose::STANDARD.encode(key);
278 write!(
279 writer,
280 "GET {path} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
281 Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\nUser-Agent: Snowbound/{}\r\n\r\n",
282 address.authority,
283 env!("CARGO_PKG_VERSION"),
284 )?;
285 let mut reader = BufReader::new(reader);
286 let head = ws::head(&mut reader)?;
287 if status(&head) != 101 {
288 return Err(Failure::Refused(status(&head), retry(&head)));
289 }
290 if ws::header(&head, "Sec-WebSocket-Accept") != Some(ws::accept(&key).as_str()) {
291 return Err(io::Error::new(io::ErrorKind::InvalidData, "Not a relay").into());
292 }
293 Ok(Connection {
294 writer,
295 reader: Box::new(reader),
296 close: Box::new(move || {
297 let _ = tcp.shutdown(Shutdown::Both);
298 }),
299 })
300}
301
302/// An HTTP/1.1 connection that answers one request after another.
303struct Http {
304 reader: BufReader<Box<dyn Read + Send>>,
305 writer: Box<dyn Write + Send>,
306 tcp: TcpStream,
307}
308
309impl Http {
310 fn open(address: &Address) -> Result<Self, Failure> {
311 let Stream {
312 reader,
313 writer,
314 tcp,
315 } = open(address)?;
316 Ok(Self {
317 reader: BufReader::new(reader),
318 writer,
319 tcp,
320 })
321 }
322
323 /// Sends `method` on `path` with `body`: the status, the head and the body answered.
324 fn ask(
325 &mut self,
326 address: &Address,
327 method: &str,
328 path: &str,
329 body: &[u8],
330 ) -> io::Result<(u16, String, Vec<u8>)> {
331 write!(
332 self.writer,
333 "{method} {path} HTTP/1.1\r\nHost: {}\r\nUser-Agent: Snowbound/{}\r\n\
334 Content-Length: {}\r\nContent-Type: application/octet-stream\r\n\r\n",
335 address.authority,
336 env!("CARGO_PKG_VERSION"),
337 body.len()
338 )?;
339 self.writer.write_all(body)?;
340 let head = ws::head(&mut self.reader)?;
341 let length: usize = ws::header(&head, "Content-Length")
342 .and_then(|length| length.parse().ok())
343 .unwrap_or(0);
344 if length > MOST {
345 return Err(io::Error::new(
346 io::ErrorKind::InvalidData,
347 "An answer too large",
348 ));
349 }
350 let mut answer = vec![0; length];
351 self.reader.read_exact(&mut answer)?;
352 Ok((status(&head), head, answer))
353 }
354}
355
356/// What a polled session shares between its reading, its sending and hanging up.
357struct Session {
358 /// Messages written and not yet sent.
359 pending: Mutex<Vec<u8>>,
360 ready: Condvar,
361 closed: AtomicBool,
362 /// The connections open now, to shut on hanging up.
363 open: Mutex<Vec<TcpStream>>,
364}
365
366impl Session {
367 fn close(&self) {
368 self.closed.store(true, Ordering::Release);
369 self.ready.notify_all();
370 for tcp in self.open.lock().unwrap().drain(..) {
371 let _ = tcp.shutdown(Shutdown::Both);
372 }
373 }
374}
375
376/// Joins `path` as HTTPS requests: the relay's answer names a session, whose messages a
377/// `GET` waits for and a `POST` sends, each a run of WebSocket frames as on a WebSocket.
378fn poll(address: &Address, path: &str) -> Result<Connection, Failure> {
379 let mut http = Http::open(address)?;
380 let joined = match path.contains('?') {
381 true => format!("{path}&poll=1"),
382 false => format!("{path}?poll=1"),
383 };
384 let (status, head, body) = http.ask(address, "GET", &joined, &[])?;
385 if status != 200 {
386 return Err(Failure::Refused(status, retry(&head)));
387 }
388 let session_id = String::from_utf8_lossy(&body)
389 .strip_prefix("session ")
390 .map(|id| id.trim().to_owned())
391 .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Not a relay"))?;
392 let at = format!("{}/v1/poll/{session_id}", address.path);
393 let session = Arc::new(Session {
394 pending: Mutex::default(),
395 ready: Condvar::new(),
396 closed: AtomicBool::new(false),
397 open: Mutex::new(vec![http.tcp.try_clone()?]),
398 });
399 let address = Arc::new(Address {
400 tls: address.tls,
401 authority: address.authority.clone(),
402 host: address.host.clone(),
403 port: address.port,
404 path: address.path.clone(),
405 });
406 // Sends what is written, a batch a request, on a connection of its own.
407 let (sending, to, posting) = (Arc::clone(&session), at.clone(), Arc::clone(&address));
408 thread::Builder::new()
409 .name("live relay post".into())
410 .spawn(move || post(&sending, &posting, &to))?;
411 let closing = Arc::clone(&session);
412 Ok(Connection {
413 writer: Box::new(PollWriter(Arc::clone(&session))),
414 reader: Box::new(PollReader {
415 session,
416 address,
417 at,
418 http,
419 arrived: Vec::new(),
420 read: 0,
421 }),
422 close: Box::new(move || closing.close()),
423 })
424}
425
426fn post(session: &Session, address: &Address, at: &str) {
427 let mut http: Option<Http> = None;
428 loop {
429 let batch = {
430 let mut pending = session.pending.lock().unwrap();
431 while pending.is_empty() && !session.closed.load(Ordering::Acquire) {
432 pending = session.ready.wait(pending).unwrap();
433 }
434 if session.closed.load(Ordering::Acquire) {
435 return;
436 }
437 std::mem::take(&mut *pending)
438 };
439 if http.is_none() {
440 http = Http::open(address).ok();
441 if let Some(opened) = &http {
442 let Ok(tcp) = opened.tcp.try_clone() else {
443 session.close();
444 return;
445 };
446 let mut open = session.open.lock().unwrap();
447 if session.closed.load(Ordering::Acquire) {
448 let _ = tcp.shutdown(Shutdown::Both);
449 return;
450 }
451 open.push(tcp);
452 }
453 }
454 let answered = http
455 .as_mut()
456 .map(|connection| connection.ask(address, "POST", at, &batch));
457 // A lost HTTP answer may follow delivery; replaying its bytes corrupts the stream.
458 if !matches!(answered, Some(Ok((200 | 204, ..)))) {
459 session.close();
460 return;
461 }
462 }
463}
464
465struct PollWriter(Arc<Session>);
466
467impl Write for PollWriter {
468 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
469 if self.0.closed.load(Ordering::Acquire) {
470 return Err(io::ErrorKind::BrokenPipe.into());
471 }
472 self.0.pending.lock().unwrap().extend_from_slice(bytes);
473 self.0.ready.notify_all();
474 Ok(bytes.len())
475 }
476
477 fn flush(&mut self) -> io::Result<()> {
478 Ok(())
479 }
480}
481
482/// Reads what each `GET` brings, waiting for the next when it is all read.
483struct PollReader {
484 session: Arc<Session>,
485 address: Arc<Address>,
486 at: String,
487 http: Http,
488 arrived: Vec<u8>,
489 read: usize,
490}
491
492impl Read for PollReader {
493 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
494 while self.read == self.arrived.len() {
495 if self.session.closed.load(Ordering::Acquire) {
496 return Ok(0);
497 }
498 let answered = self.http.ask(&self.address, "GET", &self.at, &[]);
499 match answered {
500 Ok((200, _, body)) => {
501 self.arrived = body;
502 self.read = 0;
503 }
504 Ok((410, ..)) => {
505 self.session.close();
506 return Ok(0);
507 }
508 Ok((status, ..)) => {
509 self.session.close();
510 return Err(io::Error::other(format!("The relay answered {status}")));
511 }
512 Err(error) => {
513 self.session.close();
514 return Err(error);
515 }
516 }
517 }
518 let length = buffer.len().min(self.arrived.len() - self.read);
519 buffer[..length].copy_from_slice(&self.arrived[self.read..self.read + length]);
520 self.read += length;
521 Ok(length)
522 }
523}
524
525/// The certificate authorities the system trusts, then Mozilla's for a system whose store is
526/// missing or stale, as updates trust them.
527fn tls() -> Arc<ClientConfig> {
528 static CONFIG: OnceLock<Arc<ClientConfig>> = OnceLock::new();
529 Arc::clone(CONFIG.get_or_init(|| {
530 let mut roots = RootCertStore::empty();
531 roots.add_parsable_certificates(rustls_native_certs::load_native_certs().certs);
532 roots.add_parsable_certificates(webpki_root_certs::TLS_SERVER_ROOT_CERTS.iter().cloned());
533 let provider = Arc::new(rustls::crypto::ring::default_provider());
534 Arc::new(
535 ClientConfig::builder_with_provider(provider)
536 .with_safe_default_protocol_versions()
537 .expect("ring speaks TLS 1.2 and 1.3")
538 .with_root_certificates(roots)
539 .with_no_client_auth(),
540 )
541 }))
542}
543
544/// TLS read on one thread while another writes: the socket is read without the lock, and
545/// what arrives is decrypted under it.
546struct TlsReader {
547 tcp: TcpStream,
548 tls: Arc<Mutex<ClientConnection>>,
549 plain: Vec<u8>,
550 at: usize,
551}
552
553impl Read for TlsReader {
554 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
555 while self.at == self.plain.len() {
556 self.plain.clear();
557 self.at = 0;
558 let mut raw = vec![0; 16 << 10];
559 let length = self.tcp.read(&mut raw)?;
560 if length == 0 {
561 return Ok(0);
562 }
563 let mut tls = self.tls.lock().unwrap();
564 let mut arrived = &raw[..length];
565 let mut closed = false;
566 while !arrived.is_empty() {
567 tls.read_tls(&mut arrived)?;
568 tls.process_new_packets()
569 .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
570 let mut chunk = [0; 4096];
571 loop {
572 match tls.reader().read(&mut chunk) {
573 Ok(0) => {
574 closed = true;
575 break;
576 }
577 Ok(length) => self.plain.extend_from_slice(&chunk[..length]),
578 Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
579 Err(error) => return Err(error),
580 }
581 }
582 }
583 while tls.wants_write() {
584 tls.write_tls(&mut &self.tcp)?;
585 }
586 if closed && self.plain.is_empty() {
587 return Ok(0);
588 }
589 }
590 let length = buffer.len().min(self.plain.len() - self.at);
591 buffer[..length].copy_from_slice(&self.plain[self.at..self.at + length]);
592 self.at += length;
593 Ok(length)
594 }
595}
596
597struct TlsWriter {
598 tcp: TcpStream,
599 tls: Arc<Mutex<ClientConnection>>,
600}
601
602impl Write for TlsWriter {
603 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
604 let mut tls = self.tls.lock().unwrap();
605 let length = tls.writer().write(bytes)?;
606 while tls.wants_write() {
607 tls.write_tls(&mut &self.tcp)?;
608 }
609 Ok(length)
610 }
611
612 fn flush(&mut self) -> io::Result<()> {
613 Ok(())
614 }
615}
616
617#[cfg(test)]
618#[path = "transport_tests.rs"]
619mod tests;