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, TcpStream, ToSocketAddrs},
15 sync::{
16 Arc, Condvar, Mutex, OnceLock,
17 atomic::{AtomicBool, Ordering},
18 },
19 thread,
20 time::Duration,
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 open(address: &Address) -> Result<Stream, Failure> {
132 let proxy = proxy::for_host(&address.host, address.tls);
133 let (host, port) = match &proxy {
134 Some(proxy) => (proxy.host.as_str(), proxy.port),
135 None => (address.host.as_str(), address.port),
136 };
137 let reached = |trouble| match proxy {
138 Some(_) => Trouble::ProxyUnreachable,
139 None => trouble,
140 };
141 let target = (host, port)
142 .to_socket_addrs()
143 .map_err(|error| Failure::Trouble(reached(Trouble::Dns), error))?
144 .next()
145 .ok_or_else(|| Failure::Trouble(reached(Trouble::Dns), io::ErrorKind::NotFound.into()))?;
146 let tcp = TcpStream::connect_timeout(&target, CONNECT).map_err(|error| {
147 let trouble = match error.kind() {
148 io::ErrorKind::TimedOut => Trouble::TimedOut,
149 _ => Trouble::Unreachable,
150 };
151 Failure::Trouble(reached(trouble), error)
152 })?;
153 tcp.set_nodelay(true)?;
154 tcp.set_read_timeout(Some(QUIET))?;
155 tcp.set_write_timeout(Some(QUIET))?;
156 if let Some(proxy) = &proxy {
157 tunnel(&tcp, proxy, address)?;
158 }
159 if !address.tls {
160 return Ok(Stream {
161 reader: Box::new(tcp.try_clone()?),
162 writer: Box::new(tcp.try_clone()?),
163 tcp,
164 });
165 }
166 let name = ServerName::try_from(address.host.clone())
167 .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
168 let mut connection = ClientConnection::new(tls(), name).map_err(io::Error::other)?;
169 while connection.is_handshaking() {
170 connection.complete_io(&mut &tcp).map_err(|error| {
171 let rejected = error
172 .get_ref()
173 .and_then(|inner| inner.downcast_ref::<rustls::Error>())
174 .is_some_and(|tls| matches!(tls, rustls::Error::InvalidCertificate(_)));
175 match rejected {
176 true => Failure::Trouble(Trouble::Certificate, error),
177 false => Failure::other(error),
178 }
179 })?;
180 }
181 let connection = Arc::new(Mutex::new(connection));
182 Ok(Stream {
183 reader: Box::new(TlsReader {
184 tcp: tcp.try_clone()?,
185 tls: Arc::clone(&connection),
186 plain: Vec::new(),
187 at: 0,
188 }),
189 writer: Box::new(TlsWriter {
190 tcp: tcp.try_clone()?,
191 tls: connection,
192 }),
193 tcp,
194 })
195}
196
197/// Asks `proxy` on `tcp` to connect through to the relay.
198fn tunnel(tcp: &TcpStream, proxy: &Proxy, address: &Address) -> Result<(), Failure> {
199 let target = format!("{}:{}", address.host, address.port);
200 let mut request = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n");
201 if let Some(authorization) = proxy.authorization() {
202 request += &format!("Proxy-Authorization: {authorization}\r\n");
203 }
204 request += "\r\n";
205 (&mut &*tcp)
206 .write_all(request.as_bytes())
207 .map_err(|error| Failure::Trouble(Trouble::ProxyUnreachable, error))?;
208 let head =
209 ws::head(&mut &*tcp).map_err(|error| Failure::Trouble(Trouble::ProxyUnreachable, error))?;
210 match status(&head) {
211 200 => Ok(()),
212 407 => Err(Failure::Trouble(
213 Trouble::ProxyAuthentication,
214 io::Error::new(io::ErrorKind::PermissionDenied, "The proxy asks for a name"),
215 )),
216 status => Err(Failure::Trouble(
217 Trouble::ProxyRefused(status),
218 io::Error::other(format!("The proxy answered {status}")),
219 )),
220 }
221}
222
223fn status(head: &str) -> u16 {
224 head.split(' ')
225 .nth(1)
226 .and_then(|status| status.parse().ok())
227 .unwrap_or(0)
228}
229
230fn retry(head: &str) -> Option<Duration> {
231 ws::header(head, "Retry-After")
232 .and_then(|seconds| seconds.parse().ok())
233 .map(Duration::from_secs)
234}
235
236/// Opens a WebSocket to `path` at `address`.
237fn websocket(address: &Address, path: &str) -> Result<Connection, Failure> {
238 let Stream {
239 reader,
240 mut writer,
241 tcp,
242 } = open(address)?;
243 let mut key = [0; 16];
244 getrandom::fill(&mut key).map_err(|_| io::Error::other("System random source failed"))?;
245 let key = base64::engine::general_purpose::STANDARD.encode(key);
246 write!(
247 writer,
248 "GET {path} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
249 Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\nUser-Agent: Snowbound/{}\r\n\r\n",
250 address.authority,
251 env!("CARGO_PKG_VERSION"),
252 )?;
253 let mut reader = BufReader::new(reader);
254 let head = ws::head(&mut reader)?;
255 if status(&head) != 101 {
256 return Err(Failure::Refused(status(&head), retry(&head)));
257 }
258 if ws::header(&head, "Sec-WebSocket-Accept") != Some(ws::accept(&key).as_str()) {
259 return Err(io::Error::new(io::ErrorKind::InvalidData, "Not a relay").into());
260 }
261 Ok(Connection {
262 writer,
263 reader: Box::new(reader),
264 close: Box::new(move || {
265 let _ = tcp.shutdown(Shutdown::Both);
266 }),
267 })
268}
269
270/// An HTTP/1.1 connection that answers one request after another.
271struct Http {
272 reader: BufReader<Box<dyn Read + Send>>,
273 writer: Box<dyn Write + Send>,
274 tcp: TcpStream,
275}
276
277impl Http {
278 fn open(address: &Address) -> Result<Self, Failure> {
279 let Stream {
280 reader,
281 writer,
282 tcp,
283 } = open(address)?;
284 Ok(Self {
285 reader: BufReader::new(reader),
286 writer,
287 tcp,
288 })
289 }
290
291 /// Sends `method` on `path` with `body`: the status, the head and the body answered.
292 fn ask(
293 &mut self,
294 address: &Address,
295 method: &str,
296 path: &str,
297 body: &[u8],
298 ) -> io::Result<(u16, String, Vec<u8>)> {
299 write!(
300 self.writer,
301 "{method} {path} HTTP/1.1\r\nHost: {}\r\nUser-Agent: Snowbound/{}\r\n\
302 Content-Length: {}\r\nContent-Type: application/octet-stream\r\n\r\n",
303 address.authority,
304 env!("CARGO_PKG_VERSION"),
305 body.len()
306 )?;
307 self.writer.write_all(body)?;
308 let head = ws::head(&mut self.reader)?;
309 let length: usize = ws::header(&head, "Content-Length")
310 .and_then(|length| length.parse().ok())
311 .unwrap_or(0);
312 if length > MOST {
313 return Err(io::Error::new(
314 io::ErrorKind::InvalidData,
315 "An answer too large",
316 ));
317 }
318 let mut answer = vec![0; length];
319 self.reader.read_exact(&mut answer)?;
320 Ok((status(&head), head, answer))
321 }
322}
323
324/// What a polled session shares between its reading, its sending and hanging up.
325struct Session {
326 /// Messages written and not yet sent.
327 pending: Mutex<Vec<u8>>,
328 ready: Condvar,
329 closed: AtomicBool,
330 /// The connections open now, to shut on hanging up.
331 open: Mutex<Vec<TcpStream>>,
332}
333
334impl Session {
335 fn close(&self) {
336 self.closed.store(true, Ordering::Release);
337 self.ready.notify_all();
338 for tcp in self.open.lock().unwrap().drain(..) {
339 let _ = tcp.shutdown(Shutdown::Both);
340 }
341 }
342}
343
344/// Joins `path` as HTTPS requests: the relay's answer names a session, whose messages a
345/// `GET` waits for and a `POST` sends, each a run of WebSocket frames as on a WebSocket.
346fn poll(address: &Address, path: &str) -> Result<Connection, Failure> {
347 let mut http = Http::open(address)?;
348 let joined = match path.contains('?') {
349 true => format!("{path}&poll=1"),
350 false => format!("{path}?poll=1"),
351 };
352 let (status, head, body) = http.ask(address, "GET", &joined, &[])?;
353 if status != 200 {
354 return Err(Failure::Refused(status, retry(&head)));
355 }
356 let session_id = String::from_utf8_lossy(&body)
357 .strip_prefix("session ")
358 .map(|id| id.trim().to_owned())
359 .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Not a relay"))?;
360 let at = format!("{}/v1/poll/{session_id}", address.path);
361 let session = Arc::new(Session {
362 pending: Mutex::default(),
363 ready: Condvar::new(),
364 closed: AtomicBool::new(false),
365 open: Mutex::new(vec![http.tcp.try_clone()?]),
366 });
367 let address = Arc::new(Address {
368 tls: address.tls,
369 authority: address.authority.clone(),
370 host: address.host.clone(),
371 port: address.port,
372 path: address.path.clone(),
373 });
374 // Sends what is written, a batch a request, on a connection of its own.
375 let (sending, to, posting) = (Arc::clone(&session), at.clone(), Arc::clone(&address));
376 thread::Builder::new()
377 .name("live relay post".into())
378 .spawn(move || post(&sending, &posting, &to))?;
379 let closing = Arc::clone(&session);
380 Ok(Connection {
381 writer: Box::new(PollWriter(Arc::clone(&session))),
382 reader: Box::new(PollReader {
383 session,
384 address,
385 at,
386 http,
387 arrived: Vec::new(),
388 read: 0,
389 }),
390 close: Box::new(move || closing.close()),
391 })
392}
393
394fn post(session: &Session, address: &Address, at: &str) {
395 let mut http: Option<Http> = None;
396 loop {
397 let batch = {
398 let mut pending = session.pending.lock().unwrap();
399 while pending.is_empty() && !session.closed.load(Ordering::Acquire) {
400 pending = session.ready.wait(pending).unwrap();
401 }
402 if session.closed.load(Ordering::Acquire) {
403 return;
404 }
405 std::mem::take(&mut *pending)
406 };
407 if http.is_none() {
408 http = Http::open(address).ok();
409 if let Some(opened) = &http {
410 let Ok(tcp) = opened.tcp.try_clone() else {
411 session.close();
412 return;
413 };
414 let mut open = session.open.lock().unwrap();
415 if session.closed.load(Ordering::Acquire) {
416 let _ = tcp.shutdown(Shutdown::Both);
417 return;
418 }
419 open.push(tcp);
420 }
421 }
422 let answered = http
423 .as_mut()
424 .map(|connection| connection.ask(address, "POST", at, &batch));
425 // A lost HTTP answer may follow delivery; replaying its bytes corrupts the stream.
426 if !matches!(answered, Some(Ok((200 | 204, ..)))) {
427 session.close();
428 return;
429 }
430 }
431}
432
433struct PollWriter(Arc<Session>);
434
435impl Write for PollWriter {
436 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
437 if self.0.closed.load(Ordering::Acquire) {
438 return Err(io::ErrorKind::BrokenPipe.into());
439 }
440 self.0.pending.lock().unwrap().extend_from_slice(bytes);
441 self.0.ready.notify_all();
442 Ok(bytes.len())
443 }
444
445 fn flush(&mut self) -> io::Result<()> {
446 Ok(())
447 }
448}
449
450/// Reads what each `GET` brings, waiting for the next when it is all read.
451struct PollReader {
452 session: Arc<Session>,
453 address: Arc<Address>,
454 at: String,
455 http: Http,
456 arrived: Vec<u8>,
457 read: usize,
458}
459
460impl Read for PollReader {
461 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
462 while self.read == self.arrived.len() {
463 if self.session.closed.load(Ordering::Acquire) {
464 return Ok(0);
465 }
466 let answered = self.http.ask(&self.address, "GET", &self.at, &[]);
467 match answered {
468 Ok((200, _, body)) => {
469 self.arrived = body;
470 self.read = 0;
471 }
472 Ok((410, ..)) => {
473 self.session.close();
474 return Ok(0);
475 }
476 Ok((status, ..)) => {
477 self.session.close();
478 return Err(io::Error::other(format!("The relay answered {status}")));
479 }
480 Err(error) => {
481 self.session.close();
482 return Err(error);
483 }
484 }
485 }
486 let length = buffer.len().min(self.arrived.len() - self.read);
487 buffer[..length].copy_from_slice(&self.arrived[self.read..self.read + length]);
488 self.read += length;
489 Ok(length)
490 }
491}
492
493/// The certificate authorities the system trusts, then Mozilla's for a system whose store is
494/// missing or stale, as updates trust them.
495fn tls() -> Arc<ClientConfig> {
496 static CONFIG: OnceLock<Arc<ClientConfig>> = OnceLock::new();
497 Arc::clone(CONFIG.get_or_init(|| {
498 let mut roots = RootCertStore::empty();
499 roots.add_parsable_certificates(rustls_native_certs::load_native_certs().certs);
500 roots.add_parsable_certificates(webpki_root_certs::TLS_SERVER_ROOT_CERTS.iter().cloned());
501 let provider = Arc::new(rustls::crypto::ring::default_provider());
502 Arc::new(
503 ClientConfig::builder_with_provider(provider)
504 .with_safe_default_protocol_versions()
505 .expect("ring speaks TLS 1.2 and 1.3")
506 .with_root_certificates(roots)
507 .with_no_client_auth(),
508 )
509 }))
510}
511
512/// TLS read on one thread while another writes: the socket is read without the lock, and
513/// what arrives is decrypted under it.
514struct TlsReader {
515 tcp: TcpStream,
516 tls: Arc<Mutex<ClientConnection>>,
517 plain: Vec<u8>,
518 at: usize,
519}
520
521impl Read for TlsReader {
522 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
523 while self.at == self.plain.len() {
524 self.plain.clear();
525 self.at = 0;
526 let mut raw = vec![0; 16 << 10];
527 let length = self.tcp.read(&mut raw)?;
528 if length == 0 {
529 return Ok(0);
530 }
531 let mut tls = self.tls.lock().unwrap();
532 let mut arrived = &raw[..length];
533 let mut closed = false;
534 while !arrived.is_empty() {
535 tls.read_tls(&mut arrived)?;
536 tls.process_new_packets()
537 .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
538 let mut chunk = [0; 4096];
539 loop {
540 match tls.reader().read(&mut chunk) {
541 Ok(0) => {
542 closed = true;
543 break;
544 }
545 Ok(length) => self.plain.extend_from_slice(&chunk[..length]),
546 Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
547 Err(error) => return Err(error),
548 }
549 }
550 }
551 while tls.wants_write() {
552 tls.write_tls(&mut &self.tcp)?;
553 }
554 if closed && self.plain.is_empty() {
555 return Ok(0);
556 }
557 }
558 let length = buffer.len().min(self.plain.len() - self.at);
559 buffer[..length].copy_from_slice(&self.plain[self.at..self.at + length]);
560 self.at += length;
561 Ok(length)
562 }
563}
564
565struct TlsWriter {
566 tcp: TcpStream,
567 tls: Arc<Mutex<ClientConnection>>,
568}
569
570impl Write for TlsWriter {
571 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
572 let mut tls = self.tls.lock().unwrap();
573 let length = tls.writer().write(bytes)?;
574 while tls.wants_write() {
575 tls.write_tls(&mut &self.tcp)?;
576 }
577 Ok(length)
578 }
579
580 fn flush(&mut self) -> io::Result<()> {
581 Ok(())
582 }
583}
584
585#[cfg(test)]
586#[path = "transport_tests.rs"]
587mod tests;