oxedyne/fe2o3 — fe2o3_net/src/ws/core.rs

65.2 KiB, 215 runs

created by r1870400018:605, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1	use crate::{
2	constant,
3	http::{
4	fields::{
5	HeaderFieldValue,
6	HeaderName,
7	},
8	header::HttpHeader,
9	msg::HttpMessage,
10	},
11	ws::{
12	handler::WebSocketHandler,
13	status::WebSocketStatusCode,
14	},
15	};
16
17	use oxedyne_fe2o3_core::{
18	prelude::*,
19	byte::ToBytes,
20	rand::Rand,
21	};
22	use oxedyne_fe2o3_data::{
23	ring::RingBuffer,
24	time::Timestamp,
25	};
26	use oxedyne_fe2o3_iop_crypto::enc::Encrypter;
27	use oxedyne_fe2o3_iop_db::api::Database;
28	use oxedyne_fe2o3_iop_hash::api::Hasher;
29	use oxedyne_fe2o3_jdat::id::NumIdDat;
30	use oxedyne_fe2o3_syntax::SyntaxRef;
31
32	use std::{
33	convert::TryFrom,
34	marker::PhantomData,
35	pin::Pin,
36	sync::{
37	Arc,
38	RwLock,
39	},
40	time::Instant,
41	};
42
43	use base64;
44	use sha1::{
45	Digest,
46	Sha1,
47	};
48	use tokio::{
49	self,
50	io::{
51	AsyncRead,
52	AsyncWrite,
53	AsyncReadExt,
54	AsyncWriteExt,
55	},
56	};
57
58
59	/// Bounds on what a peer may make a websocket reader allocate.
60	///
61	/// Both numbers are needed. A frame declares its own payload length, so the frame bound stops one
62	/// frame naming a size no machine can honour; but a message may arrive as any number of
63	/// continuation frames, each of them under the frame bound, so the message bound stops a peer
64	/// reaching the same total in instalments.
65	///
66	/// The defaults are [`constant::WEBSOCKET_MAX_FRAME_BYTES`] and
67	/// [`constant::WEBSOCKET_MAX_MESSAGE_BYTES`]. An application that knows its own traffic can raise
68	/// or lower them; there is deliberately no way to express "no bound".
69	#[derive(Clone, Copy, Debug, Eq, PartialEq)]
70	pub struct WebSocketLimits {
71	/// Most a single frame may declare, in bytes.
72	pub max_frame: usize,
73	/// Most an assembled message may reach, in bytes.
74	pub max_msg: usize,
75	}
76
77	impl Default for WebSocketLimits {
78	fn default() -> Self {
79	Self {
80	max_frame: constant::WEBSOCKET_MAX_FRAME_BYTES,
81	max_msg: constant::WEBSOCKET_MAX_MESSAGE_BYTES,
82	}
83	}
84	}
85
86	impl WebSocketLimits {
87
88	/// Bounds a message at `max_msg` bytes, and each of its frames at the same number.
89	///
90	/// The common case: an application knows how big its largest message is and does not care how
91	/// the peer chooses to fragment it.
92	pub fn new(max_msg: usize) -> Self {
93	Self {
94	max_frame: max_msg,
95	max_msg,
96	}
97	}
98
99	/// Refuses a frame whose declared payload length is over the frame bound.
100	///
101	/// The length is checked while it is still the 64-bit number that came off the wire, because
102	/// narrowing it first would truncate on a 32-bit target: a declared 2^32 + 1 bytes becomes one
103	/// byte, and the frame passes a bound it is vastly over. Nothing is allocated until this
104	/// returns.
105	pub fn check_frame(&self, declared: u64) -> Outcome<()> {
106	if declared > self.max_frame as u64 {
107	return Err(err!(
108	"A websocket frame declares a payload of {} bytes, over the {} byte frame limit.",
109	declared, self.max_frame;
110	IO, Network, Invalid, Input, Wire, TooBig));
111	}
112	Ok(())
113	}
114
115	/// Refuses a message whose assembled payload would be over the message bound.
116	///
117	/// Called with what is already buffered plus what the next frame declares, so a run of legal
118	/// continuation frames is stopped at the one that would take the total past the bound, before
119	/// that frame is allocated for.
120	pub fn check_msg(&self, total: usize) -> Outcome<()> {
121	if total > self.max_msg {
122	return Err(err!(
123	"A websocket message would reach {} bytes, over the {} byte message limit.",
124	total, self.max_msg;
125	IO, Network, Invalid, Input, Wire, TooBig));
126	}
127	Ok(())
128	}
129	}
130
131	#[derive(Debug)]
132	pub enum WebSocketMessage {
133	Text(String),
134	Binary(Vec<u8>),
135	Ping(Vec<u8>),
136	Pong(Vec<u8>),
137	Close(Option<WebSocketStatusCode>, Option<String>),
138	}
139
140	/// Builds a client's upgrade request for `path` on `host`, returning it with the random 16-byte
141	/// `Sec-WebSocket-Key` it carries, from which the server's accept key is checked.
142	///
143	/// `host` is the authority -- a name or address, with a port where it is not the default -- and
144	/// goes in the `Host` field alone; the path belongs to the request line. `origin` is sent as
145	/// `Origin` when given, since a server that checks origins refuses a request without one.
146	pub fn connect_request(
147	host: &str,
148	path: &str,
149	origin: Option<&str>,
150	)
151	-> Outcome<(HttpMessage, String)>
152	{
153	if !path.starts_with('/') {
154	return Err(err!(
155	"A websocket request path must begin with '/', found '{}'.", path;
156	Invalid, Input, String));
157	}
158	let mut key = [0u8; 16];
159	Rand::fill_u8(&mut key);
160	let key_str = base64::encode(&key);
161	let origin_line = match origin {
162	Some(origin) => fmt!("Origin: {}\r\n", origin),
163	None => String::new(),
164	};
165	let msg = fmt!(
166	"GET {} HTTP/1.1\r\n\
167	Host: {}\r\n\
168	Upgrade: websocket\r\n\
169	Connection: Upgrade\r\n\
170	Sec-WebSocket-Key: {}\r\n\
171	Sec-WebSocket-Version: 13\r\n\
172	{}\r\n",
173	path, host, key_str.clone(), origin_line,
174	);
175	Ok((
176	HttpMessage {
177	header: res!(HttpHeader::parse(msg, Some(true))),
178	body: Vec::new(),
179	head_only: false,
180	file: None,
181	},
182	key_str,
183	))
184	}
185
186	/// Derives the `Sec-WebSocket-Accept` value from the client's `Sec-WebSocket-Key`, as RFC 6455
187	/// §4.2.2 step 5.4 defines it: the key and the WebSocket GUID are concatenated as US-ASCII, hashed
188	/// with SHA-1, and the digest is base64-encoded with the standard alphabet.
189	///
190	/// Every browser recomputes this value and refuses the handshake if it disagrees, so the algorithm
191	/// is fixed by the peer, not by us. It is a free function because it needs none of `WebSocket`'s
192	/// generic parameters, and because a value a third party verifies must be testable on its own.
193	pub fn accept_key(key: &str) -> String {
194	let concatenated = fmt!("{}{}", key, constant::WEBSOCKET_GUID);
195	let mut hasher = Sha1::new();
196	hasher.update(concatenated.as_bytes());
197	let hash = hasher.finalize();
198	base64::encode(&hash)
199	}
200
201	/// Builds the `101 Switching Protocols` response that answers a client's upgrade request.
202	///
203	/// The request's `Sec-WebSocket-Key` is validated (present, valid base64, 16 bytes once decoded)
204	/// before the accept value is derived from it, so a malformed handshake is refused here rather
205	/// than half-completed.
206	///
207	/// A free function, because a server that owns its own socket -- one that reads the upgrade
208	/// request itself and then splits the stream into halves for two tasks -- needs the response text
209	/// without needing a [`WebSocket`], whose type parameters describe a database it has no use for.
210	pub fn accept_response(request: &HttpMessage) -> Outcome<String> {
211	let key = match request.header.get_the_field_value(&HeaderName::SecWebSocketKey) {
212	Ok(HeaderFieldValue::SecWebSocketKey(key)) => {
213	let key_byts = match base64::decode(&key) {
214	Ok(byts) => byts,
215	Err(e) => return Err(err!(e,
216	"The websocket key provided is not valid base64.";
217	IO, Network, Invalid, Input, String, Conversion)),
218	};
219	if key_byts.len() != 16 {
220	return Err(err!(
221	"The websocket key is {} bytes long, expected 16.", key_byts.len();
222	IO, Network, Invalid, Input, Mismatch, Size));
223	}
224	key
225	},
226	_ => return Err(err!(
227	"The websocket key string is missing.";
228	IO, Network, Input, Missing)),
229	};
230	Ok(fmt!(
231	"HTTP/1.1 101 Switching Protocols\r\n\
232	Upgrade: websocket\r\n\
233	Connection: Upgrade\r\n\
234	Sec-WebSocket-Accept: {}\r\n\r\n",
235	accept_key(&key),
236	))
237	}
238
239	/// Encodes a message as the bytes of one or more RFC 6455 frames.
240	///
241	/// `mask` is set by a client and clear by a server: RFC 6455 §5.3 requires client-to-server frames
242	/// to be masked and forbids the mask on server-to-client frames. A payload longer than
243	/// `chunk_thresh` is fragmented into frames of at most `chunk_size` bytes, the first carrying the
244	/// message's opcode and the rest the continuation opcode, with `FIN` on the last.
245	///
246	/// A free function so that the write half of a split stream can be framed by a task that never
247	/// touches the read half.
248	pub fn encode_message(
249	message: &WebSocketMessage,
250	mask: bool,
251	chunk_size: usize,
252	chunk_thresh: usize,
253	)
254	-> Outcome<Vec<u8>>
255	{
256	if chunk_size == 0 {
257	return Err(err!(
258	"A websocket chunk size of zero cannot make progress.";
259	Invalid, Input, Size));
260	}
261
262	// Determine the opcode based on the message type.
263	let initial_opcode = match message {
264	WebSocketMessage::Text(_) => 0x1,
265	WebSocketMessage::Binary(_) => 0x2,
266	WebSocketMessage::Ping(_) => 0x9,
267	WebSocketMessage::Pong(_) => 0xA,
268	WebSocketMessage::Close(_, _) => 0x8,
269	};
270
271	// Get the payload data.
272	let payload = match message {
273	WebSocketMessage::Text(text) => text.as_bytes().to_vec(),
274	WebSocketMessage::Binary(data) => data.clone(),
275	WebSocketMessage::Ping(data) => data.clone(),
276	WebSocketMessage::Pong(data) => data.clone(),
277	WebSocketMessage::Close(status_code, reason) => {
278	let mut data = Vec::new();
279	if let Some(code) = status_code {
280	data.extend_from_slice(&code.to_bytes());
281	}
282	if let Some(reason_str) = reason {
283	data.extend_from_slice(reason_str.as_bytes());
284	}
285	data
286	}
287	};
288	let payload_length = payload.len();
289
290	// Generate the masking key for client-side masking.
291	let mut masking_key = [0u8; 4];
292	if mask {
293	Rand::fill_u8(&mut masking_key);
294	}
295	let mask_bit = if mask { 0x80 } else { 0x00 };
296
297	let mut out = Vec::with_capacity(payload_length + 16);
298	if payload_length > chunk_thresh {
299	// Send the message in chunks.
300	let mut bytes_sent = 0;
301	while bytes_sent < payload_length {
302	let remaining_bytes = payload_length - bytes_sent;
303	let chunk = std::cmp::min(remaining_bytes, chunk_size);
304	let is_final = remaining_bytes <= chunk_size;
305
306	// Use the initial opcode for the first frame, continuation (0x0) for the others.
307	let opcode = if bytes_sent == 0 { initial_opcode } else { 0x0 };
308
309	// First byte: FIN bit and opcode.
310	out.push(if is_final { 0x80 \| opcode } else { opcode });
311
312	// Second byte: mask bit and payload length.
313	if chunk <= 125 {
314	out.push(mask_bit \| chunk as u8);
315	} else if chunk <= 65535 {
316	out.push(mask_bit \| 126);
317	out.extend_from_slice(&(chunk as u16).to_be_bytes());
318	} else {
319	out.push(mask_bit \| 127);
320	out.extend_from_slice(&(chunk as u64).to_be_bytes());
321	}
322	if mask {
323	out.extend_from_slice(&masking_key);
324	}
325	for i in 0..chunk {
326	let b = payload[bytes_sent + i];
327	out.push(if mask { b ^ masking_key[i % 4] } else { b });
328	}
329	bytes_sent += chunk;
330	}
331	} else {
332	// Send the message as a single frame.
333	out.push(0x80 \| initial_opcode);
334	if payload_length <= 125 {
335	out.push(mask_bit \| payload_length as u8);
336	} else if payload_length <= 65535 {
337	out.push(mask_bit \| 126);
338	out.extend_from_slice(&(payload_length as u16).to_be_bytes());
339	} else {
340	out.push(mask_bit \| 127);
341	out.extend_from_slice(&(payload_length as u64).to_be_bytes());
342	}
343	if mask {
344	out.extend_from_slice(&masking_key);
345	}
346	for (i, b) in payload.iter().enumerate() {
347	out.push(if mask { b ^ masking_key[i % 4] } else { *b });
348	}
349	}
350	Ok(out)
351	}
352
353	/// One RFC 6455 frame as it came off the wire, its payload already unmasked.
354	#[derive(Clone, Debug, Eq, PartialEq)]
355	pub struct WebSocketFrame {
356	pub fin: bool,
357	pub opcode: u8,
358	pub masked: bool, // whether the sender masked it, as a client must
359	pub payload: Vec<u8>,
360	}
361
362	impl WebSocketFrame {
363	/// Is this a control frame (close, ping or pong), which may arrive between the fragments of a
364	/// message without belonging to it?
365	pub fn is_control(&self) -> bool { (self.opcode & 0x08) != 0 }
366	}
367
368	/// Reads one frame from `stream`. `Ok(None)` means the peer closed the connection.
369	///
370	/// `buffered` is the payload already gathered of the message this frame may continue, so that a
371	/// data frame is refused as soon as its header shows it would take the message over
372	/// [`WebSocketLimits::max_msg`]. Every declared length is checked before a byte is reserved for
373	/// it, and an error from either bound carries the `TooBig` tag.
374	///
375	/// Cancelling the future part way through a frame loses the stream's place, so a caller wanting a
376	/// timeout waits for the first byte to arrive before calling this, as
377	/// [`crate::ws::client::WsClient::recv`] does.
378	pub async fn read_frame<R: AsyncRead + Unpin>(
379	stream: &mut R,
380	chunk_size: usize,
381	limits: WebSocketLimits,
382	buffered: usize,
383	)
384	-> Outcome<Option<WebSocketFrame>>
385	{
386	if chunk_size == 0 {
387	return Err(err!(
388	"A websocket chunk size of zero cannot make progress.";
389	Invalid, Input, Size));
390	}
391
392	// Read the first byte of the frame header.
393	let mut header_byte = [0u8; 1];
394	match stream.read_exact(&mut header_byte).await {
395	Ok(_n) => (),
396	Err(e) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
397	return Ok(None);
398	}
399	Err(e) => return Err(err!(e,
400	"While trying to read first byte of the frame header.";
401	IO, Network, Read, Wire)),
402	}
403	let fin = (header_byte[0] & 0x80) != 0;
404	let opcode = header_byte[0] & 0x0F;
405	let is_control = (opcode & 0x08) != 0;
406
407	// A control frame is never fragmented (RFC 6455 §5.5).
408	if is_control && !fin {
409	return Err(err!(
410	"A websocket control frame with opcode {:#x} is fragmented, which RFC 6455 §5.5 \
411	forbids.", opcode;
412	IO, Network, Invalid, Input, Wire));
413	}
414
415	// Read the second byte of the frame header.
416	let mut length_byte = [0u8; 1];
417	match stream.read_exact(&mut length_byte).await {
418	Ok(_n) => (),
419	Err(e) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
420	return Ok(None);
421	}
422	Err(e) => return Err(err!(e,
423	"While trying to read second byte of the frame header.";
424	IO, Network, Read, Wire)),
425	}
426
427	// The length stays 64 bits wide until it has been checked, since narrowing it first would
428	// truncate on a 32-bit target and let a huge declaration through as a small one.
429	let masked = (length_byte[0] & 0x80) != 0;
430	let declared: u64 = match length_byte[0] & 0x7F {
431	127 => {
432	let mut extended_length_bytes = [0u8; 8];
433	match stream.read_exact(&mut extended_length_bytes).await {
434	Ok(_n) => (),
435	Err(e) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
436	return Ok(None);
437	}
438	Err(e) => return Err(err!(e,
439	"While trying to read the 64-bit extended payload length.";
440	IO, Network, Read, Wire)),
441	}
442	u64::from_be_bytes(extended_length_bytes)
443	}
444	126 => {
445	let mut extended_length_bytes = [0u8; 2];
446	match stream.read_exact(&mut extended_length_bytes).await {
447	Ok(_n) => (),
448	Err(e) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
449	return Ok(None);
450	}
451	Err(e) => return Err(err!(e,
452	"While trying to read the 16-bit extended payload length.";
453	IO, Network, Read, Wire)),
454	}
455	u16::from_be_bytes(extended_length_bytes) as u64
456	}
457	len => len as u64,
458	};
459
460	// A control frame carries at most 125 bytes (RFC 6455 §5.5), so a control opcode declaring
461	// more than that is malformed however generous the limits are.
462	if is_control && declared > constant::WEBSOCKET_MAX_CONTROL_FRAME_BYTES {
463	return Err(err!(
464	"A websocket control frame with opcode {:#x} declares a payload of {} bytes; \
465	RFC 6455 §5.5 allows at most {}.",
466	opcode, declared, constant::WEBSOCKET_MAX_CONTROL_FRAME_BYTES;
467	IO, Network, Invalid, Input, Wire, TooBig));
468	}
469
470	// Bound the frame, and then the message a data frame would join, before anything is allocated
471	// to hold either. The peer's number is not believed until it has been agreed to. The bound's own
472	// error carries `TooBig`, and `Error::tags` reads every frame of the chain, so the 1009 close finds
473	// it through any frame added on the way out.
474	res!(limits.check_frame(declared));
475	let payload_length = declared as usize; // Narrowing is safe: `check_frame` bounded it.
476	if !is_control {
477	res!(limits.check_msg(buffered.saturating_add(payload_length)));
478	}
479
480	let mut masking_key = [0u8; 4];
481	if masked {
482	match stream.read_exact(&mut masking_key).await {
483	Ok(_n) => (),
484	Err(e) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
485	return Ok(None);
486	}
487	Err(e) => return Err(err!(e,
488	"While trying to read the frame masking key.";
489	IO, Network, Read, Wire)),
490	}
491	}
492
493	// Read the payload, unmasking each chunk as it lands.
494	let mut payload = vec![0u8; payload_length];
495	let mut bytes_read = 0;
496	while bytes_read < payload_length {
497	let chunk = std::cmp::min(chunk_size, payload_length - bytes_read);
498	match stream.read_exact(&mut payload[bytes_read..bytes_read + chunk]).await {
499	Ok(_n) => {
500	if masked {
501	for i in bytes_read..bytes_read + chunk {
502	payload[i] ^= masking_key[i % 4];
503	}
504	}
505	bytes_read += chunk;
506	}
507	Err(e) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
508	return Ok(None);
509	}
510	Err(e) => return Err(err!(e,
511	"While trying to read payload chunk.";
512	IO, Network, Read, Wire)),
513	}
514	}
515
516	Ok(Some(WebSocketFrame { fin, opcode, masked, payload }))
517	}
518
519	/// What one frame did to the message being assembled.
520	pub(crate) enum Assembled {
521	Pending, // a fragment, more to come
522	Message(WebSocketMessage), // the last fragment, or an unfragmented message
523	Control(WebSocketMessage), // a control frame, which leaves any message in progress intact
524	}
525
526	/// Folds `frame` into the message in progress, whose opcode is `opcode` (`None` between messages)
527	/// and whose payload so far is `buffer`. The one reassembly rule for both [`read_message`] and
528	/// [`crate::ws::client::WsClient`].
529	pub(crate) fn assemble(
530	opcode: &mut Option<u8>,
531	buffer: &mut Vec<u8>,
532	frame: WebSocketFrame,
533	)
534	-> Outcome<Assembled>
535	{
536	if frame.is_control() {
537	return Ok(Assembled::Control(res!(decode_message(frame.opcode, &frame.payload))));
538	}
539	match (*opcode, frame.opcode) {
540	(None, 0x0) => return Err(err!(
541	// The peer either lost track of the message it was sending or is probing for a panic.
542	"The first frame of a message carries the continuation opcode, so there is \
543	no message for it to continue.";
544	IO, Network, Invalid, Input, Wire)),
545	(None, 0x1) \| (None, 0x2) => *opcode = Some(frame.opcode),
546	(None, op) => return Err(err!(
547	"Unknown websocket data opcode {:#x}.", op;
548	IO, Network, Invalid, Input, Wire)),
549	(Some(_), 0x0) => (),
550	(Some(first), op) => return Err(err!(
551	"A websocket frame with opcode {:#x} arrived while a message with opcode {:#x} was \
552	still being continued; RFC 6455 §5.4 allows only continuation frames there.",
553	op, first;
554	IO, Network, Invalid, Input, Wire)),
555	}
556	buffer.extend_from_slice(&frame.payload);
557	if !frame.fin {
558	return Ok(Assembled::Pending);
559	}
560	let first = match opcode.take() {
561	Some(first) => first,
562	None => return Err(err!(
563	"No opcode was recorded for a finished websocket message."; Bug)),
564	};
565	// The buffer is emptied whether or not the payload decodes, so nothing of a bad message is
566	// left to join the front of the next.
567	let result = decode_message(first, buffer);
568	buffer.clear();
569	Ok(Assembled::Message(res!(result)))
570	}
571
572	/// Turns an assembled payload into the message its opcode names.
573	fn decode_message(opcode: u8, payload: &[u8]) -> Outcome<WebSocketMessage> {
574	Ok(match opcode {
575	0x1 => WebSocketMessage::Text(res!(std::str::from_utf8(payload)).to_string()),
576	0x2 => WebSocketMessage::Binary(payload.to_vec()),
577	0x8 => {
578	let status_code = if payload.len() >= 2 {
579	let nu16 = u16::from_be_bytes([payload[0], payload[1]]);
580	Some(res!(WebSocketStatusCode::try_from(nu16)))
581	} else {
582	None
583	};
584	let reason = if payload.len() > 2 {
585	Some(res!(std::str::from_utf8(&payload[2..])).to_string())
586	} else {
587	None
588	};
589	WebSocketMessage::Close(status_code, reason)
590	}
591	0x9 => WebSocketMessage::Ping(payload.to_vec()),
592	0xA => WebSocketMessage::Pong(payload.to_vec()),
593	_ => return Err(err!("Unknown opcode: {}", opcode; IO, Network, Invalid, Input)),
594	})
595	}
596
597	/// Reads one message -- however many frames it arrives in -- from `stream`.
598	///
599	/// `buffer` accumulates the payload across the frames of a fragmented message and is cleared
600	/// before the message is returned, so the same buffer can be reused for the next call. `Ok(None)`
601	/// means the peer closed the connection.
602	///
603	/// `limits` bounds what the peer can make this allocate; see [`read_frame`]. An error from either
604	/// bound carries the `TooBig` tag, which is how [`WebSocket::read`] knows to answer with a 1009
605	/// close.
606	///
607	/// A control frame arriving between the fragments of a message (RFC 6455 §5.4) cannot be handed
608	/// back without losing the message, since nothing here outlives the call. A close ends the
609	/// message and is returned; a ping or pong is consumed unanswered. A caller that must answer
610	/// such a ping owns both halves of the stream and assembles frames itself, as
611	/// [`crate::ws::client::WsClient`] does.
612	///
613	/// A free function so that the read half of a split stream can be decoded by a task that never
614	/// touches the write half.
615	pub async fn read_message<R: AsyncRead + Unpin>(
616	stream: &mut R,
617	buffer: &mut Vec<u8>,
618	chunk_size: usize,
619	limits: WebSocketLimits,
620	)
621	-> Outcome<Option<WebSocketMessage>>
622	{
623	let mut opcode = None;
624	loop {
625	let frame = match res!(read_frame(stream, chunk_size, limits, buffer.len()).await) {
626	Some(frame) => frame,
627	None => return Ok(None),
628	};
629	let mid_message = opcode.is_some();
630	match res!(assemble(&mut opcode, buffer, frame)) {
631	Assembled::Pending => (),
632	Assembled::Message(msg) => return Ok(Some(msg)),
633	Assembled::Control(msg) => match msg {
634	_ if !mid_message => return Ok(Some(msg)),
635	WebSocketMessage::Close(..) => {
636	buffer.clear();
637	return Ok(Some(msg));
638	}
639	_ => (), // interleaved ping or pong, see above
640	},
641	}
642	}
643	}
644
645	pub struct WebSocket<
646	'a,
647	const UIDL: usize,
648	UID: NumIdDat<UIDL> + 'static,
649	ENC: Encrypter,
650	KH: Hasher,
651	DB: Database<UIDL, UID, ENC, KH>,
652	S: AsyncRead + AsyncWrite + Unpin,
653	WSH: WebSocketHandler,
654	> {
655	stream: Pin<&'a mut S>,
656	is_server: bool,
657	buffer: Vec<u8>,
658	pub latency: RingBuffer<{ constant::WEBSOCKET_LATENCY_HISTORY_SIZE }, Option<u16>>,
659	pub handler: WSH,
660	chunk_size: usize,
661	chunk_thresh: usize,
662	limits: WebSocketLimits,
663	phantom1: PhantomData<UID>,
664	phantom2: PhantomData<ENC>,
665	phantom3: PhantomData<KH>,
666	phantom4: PhantomData<DB>,
667	}
668
669	impl<
670	'a,
671	const UIDL: usize,
672	UID: NumIdDat<UIDL> + 'static,
673	ENC: Encrypter + 'static,
674	KH: Hasher + 'static,
675	DB: Database<UIDL, UID, ENC, KH> + 'static,
676	S: AsyncRead + AsyncWrite + Unpin,
677	WSH: WebSocketHandler,
678	>
679	WebSocket<'a, UIDL, UID, ENC, KH, DB, S, WSH>
680	{
681	pub fn new_client(
682	stream: &'a mut S,
683	handler: WSH,
684	chunk_size: usize,
685	chunk_thresh: usize,
686	)
687	-> Self
688	{
689	Self {
690	stream: Pin::new(stream),
691	is_server: false,
692	buffer: Vec::new(),
693	latency: RingBuffer::default(),
694	handler,
695	chunk_size,
696	chunk_thresh,
697	limits: WebSocketLimits::default(),
698	phantom1: PhantomData,
699	phantom2: PhantomData,
700	phantom3: PhantomData,
701	phantom4: PhantomData,
702	}
703	}
704
705	pub fn new_server(
706	stream: &'a mut S,
707	handler: WSH,
708	chunk_size: usize,
709	chunk_thresh: usize,
710	)
711	-> Self
712	{
713	Self {
714	stream: Pin::new(stream),
715	is_server: true,
716	buffer: Vec::new(),
717	latency: RingBuffer::default(),
718	handler,
719	chunk_size,
720	chunk_thresh,
721	limits: WebSocketLimits::default(),
722	phantom1: PhantomData,
723	phantom2: PhantomData,
724	phantom3: PhantomData,
725	phantom4: PhantomData,
726	}
727	}
728
729	pub fn is_server(&self) -> bool { self.is_server }
730	pub fn is_client(&self) -> bool { !self.is_server }
731
732	/// The bounds applied to incoming messages.
733	pub fn limits(&self) -> WebSocketLimits { self.limits }
734
735	/// Replaces the bounds applied to incoming messages, for an application whose traffic differs
736	/// from the defaults in [`WebSocketLimits`].
737	pub fn with_limits(mut self, limits: WebSocketLimits) -> Self {
738	self.limits = limits;
739	self
740	}
741
742	pub async fn connect(
743	&mut self,
744	request: HttpMessage,
745	key: Option<String>,
746	)
747	-> Outcome<()>
748	{
749	if self.is_client() {
750	match key {
751	Some(key) => {
752	self.connect_as_client(request, key).await
753	}
754	None => Err(err!(
755	"Expected a key string, received: {:?}", key;
756	Input, Missing)),
757	}
758	} else {
759	self.connect_as_server(request).await
760	}
761	}
762
763	pub async fn connect_as_client(
764	&mut self,
765	request: HttpMessage,
766	key: String,
767	)
768	-> Outcome<()>
769	{
770	let result = request.write_all(&mut self.stream).await;
771	res!(result);
772	let result = HttpMessage::read::<
773	{ constant::HTTP_DEFAULT_HEADER_CHUNK_SIZE },
774	{ constant::HTTP_DEFAULT_BODY_CHUNK_SIZE },
775	_,
776	>(Pin::new(&mut self.stream), &Vec::new(), Some(false), None).await;
777	match result {
778	Ok((Some(response), _)) => {
779	res!(response.check_websocket_handshake(&accept_key(&key)));
780	info!("Client connection successfully upgraded to a websocket.");
781	},
782	Err(e) => return Err(err!(e,
783	"While checking server websocket upgrade response.";
784	IO, Network, Wire, Read)),
785	Ok((None, _)) => return Err(err!(
786	"UnexpectedEof indicates connection closure.";
787	IO, Network, Wire, Read)),
788	}
789
790	Ok(())
791	}
792
793	/// The HTTP(S) server has detected a websocket upgrade request message and passes it to this
794	/// method to complete the handshake.
795	pub async fn connect_as_server(
796	&mut self,
797	request: HttpMessage,
798	)
799	-> Outcome<()>
800	{
801	let response = res!(accept_response(&request));
802
803	match self.stream.write_all(response.as_bytes()).await {
804	Ok(()) => (),
805	Err(e) => return Err(err!(e,
806	"Could not send websocket handshake response.";
807	IO, Network, Wire, Write)),
808	}
809
810	info!("Server connection successfully upgraded to a websocket.");
811
812	Ok(())
813	}
814
815	/// Reads one message from the stream, however many frames it arrives in. `Ok(None)` means the
816	/// peer closed the connection.
817	///
818	/// A message that breaches this socket's [`WebSocketLimits`] is answered with a close frame
819	/// carrying status 1009, which is what RFC 6455 §7.4.1 reserves for a message too big to
820	/// process, and is the only notice the peer gets that its message was refused rather than
821	/// lost. The refusal leaves the stream part way through a frame whose payload was never read,
822	/// so there is nothing to resynchronise to and the connection ends: the close frame is the
823	/// last thing sent on it.
824	pub async fn read(&mut self) -> Outcome<Option<WebSocketMessage>> {
825	let chunk_size = self.chunk_size;
826	let limits = self.limits;
827	let mut buffer = std::mem::take(&mut self.buffer);
828	let result = read_message(
829	self.stream.as_mut().get_mut(),
830	&mut buffer,
831	chunk_size,
832	limits,
833	).await;
834	self.buffer = buffer;
835	if let Err(e) = &result {
836	if e.tags().contains(&ErrTag::TooBig) {
837	// Whatever was gathered of the refused message is dropped here rather than left to
838	// join the front of whatever is read next.
839	self.buffer.clear();
840	let close = WebSocketMessage::Close(
841	Some(WebSocketStatusCode::MessageTooBig),
842	Some(fmt!("Message too big")),
843	);
844	if let Err(e) = self.send(&close).await {
845	// The error being returned is the peer's, and stands whether or not it heard
846	// about it.
847	error!(err!(e,
848	"While sending a 1009 close frame to a peer whose message was over the \
849	limit.";
850	IO, Network, Wire, Write));
851	}
852	}
853	}
854	result
855	}
856
857	/// Frames `message` and writes it to the stream, masking it when this end is the client.
858	pub async fn send(
859	&mut self,
860	message: &WebSocketMessage,
861	)
862	-> Outcome<()>
863	{
864	let byts = res!(encode_message(
865	message,
866	self.is_client(),
867	self.chunk_size,
868	self.chunk_thresh,
869	));
870	let result = self.stream.write_all(&byts).await;
871	res!(result);
872	let result = self.stream.flush().await;
873	res!(result);
874	Ok(())
875	}
876
877
878	pub async fn close(
879	&mut self,
880	status_code: Option<WebSocketStatusCode>,
881	reason: Option<String>,
882	)
883	-> Outcome<()>
884	{
885	// Construct the close frame payload
886	let mut payload = Vec::new();
887	if let Some(code) = status_code {
888	let code_u16: u16 = code.into();
889	payload.extend_from_slice(&code_u16.to_be_bytes());
890	}
891	if let Some(reason_str) = reason.clone() {
892	payload.extend_from_slice(reason_str.as_bytes());
893	}
894
895	// Send the close frame
896	let close_frame = WebSocketMessage::Close(status_code, reason.clone());
897	let result = self.send(&close_frame).await;
898	res!(result);
899
900	if self.is_server() {
901	// Server-side: Wait for the client to send a close frame
902	let close_response;
903	loop {
904	let result = self.read().await;
905	match result {
906	Ok(Some(message)) => match message {
907	WebSocketMessage::Close(_, _) => {
908	close_response = Some(message);
909	break;
910	}
911	_ => {
912	// Ignore any other messages until we receive a close frame
913	continue;
914	}
915	},
916	Ok(None) => {
917	info!("The client has closed the connection.");
918	return Ok(());
919	}
920	Err(e) => return Err(e.into()),
921	}
922	}
923
924	// Verify the close response from the client
925	if let Some(WebSocketMessage::Close(client_status_code, client_reason)) = close_response {
926	if let Some(code) = status_code {
927	if client_status_code != Some(code) {
928	return Err(err!(
929	"Received unexpected close status code from client: {:?}", client_status_code;
930	IO, Network, Invalid, Input));
931	}
932	}
933	if let Some(reason_str) = reason {
934	if client_reason != Some(reason_str) {
935	return Err(err!(
936	"Received unexpected close reason from client: {:?}", client_reason;
937	IO, Network, Invalid, Input));
938	}
939	}
940	} else {
941	return Err(err!(
942	"Expected close frame response from client, but received: {:?}", close_response;
943	IO, Network, Invalid, Input));
944	}
945	} else {
946	// Client-side: Read the close frame response from the server
947	let close_response;
948	loop {
949	let result = self.read().await;
950
951	match result {
952	Ok(Some(msg)) => match msg {
953	WebSocketMessage::Close(_, _) => {
954	close_response = Some(msg);
955	break;
956	}
957	_ => {
958	// Ignore any other messages until we receive a close frame
959	continue;
960	}
961	}
962	Ok(None) => {
963	info!("The server has closed the connection.");
964	return Ok(());
965	}
966	Err(e) => return Err(e.into()),
967	}
968	}
969
970	// Verify the close response from the server
971	if let Some(WebSocketMessage::Close(server_status_code, server_reason)) = close_response {
972	if let Some(code) = status_code {
973	if server_status_code != Some(code) {
974	return Err(err!(
975	"Received unexpected close status code from server: {:?}", server_status_code;
976	IO, Network, Invalid, Input));
977	}
978	}
979	if let Some(reason_str) = reason {
980	if server_reason != Some(reason_str) {
981	return Err(err!(
982	"Received unexpected close reason from server: {:?}", server_reason;
983	IO, Network, Invalid, Input));
984	}
985	}
986	} else {
987	return Err(err!(
988	"Expected close frame response from server, but received: {:?}", close_response;
989	IO, Network, Invalid, Input));
990	}
991	}
992
993	// Close the underlying TCP connection
994	let result = self.stream.shutdown().await;
995	res!(result);
996
997	Ok(())
998	}
999
1000	async fn response_handler(
1001	&mut self,
1002	result: Outcome<Option<WebSocketMessage>>,
1003	err_count: &mut usize,
1004	max_errors: usize,
1005	in_typ: &str,
1006	id: &String,
1007	)
1008	-> Outcome<()>
1009	{
1010	match result {
1011	Ok(response_opt) => {
1012	if let Some(response) = response_opt {
1013	let result = self.send(&response).await;
1014	if let Err(e) = result {
1015	*err_count += 1;
1016	if *err_count > max_errors {
1017	let e = err!(e,
1018	"{}: The number of websocket handler errors has exceeded the limit of {}, \
1019	the connection will now be terminated.", id, max_errors;
1020	IO, Network, Wire, Excessive);
1021	error!(e.clone());
1022	return Err(e);
1023	} else {
1024	error!(e, "{}: While trying to send response to an incoming {} message. This \
1025	websocket handler error leaves {} more before connection termination.",
1026	id, in_typ, max_errors - *err_count,
1027	);
1028	}
1029	}
1030	}
1031	}
1032	Err(e) => {
1033	*err_count += 1;
1034	if *err_count > max_errors {
1035	let e = err!(e,
1036	"{}: The number of websocket handler errors has exceeded the limit of {}, \
1037	the connection will now be terminated.", id, max_errors;
1038	IO, Network, Wire, Excessive);
1039	error!(e.clone());
1040	return Err(e);
1041	} else {
1042	error!(e, "{}: This websocket handler error leaves {} more before connection \
1043	termination.", id, max_errors - *err_count,
1044	);
1045	}
1046	}
1047	}
1048	Ok(())
1049	}
1050
1051	pub async fn listen(
1052	&mut self,
1053	db: Option<(Arc<RwLock<DB>>, UID)>,
1054	syntax: SyntaxRef,
1055	ping_interval: Option<u8>,
1056	max_errors: u8,
1057	id: &String,
1058	)
1059	-> Outcome<()>
1060	{
1061	let mut err_count = 0;
1062	let max_errors = max_errors as usize;
1063	let mut ping_timestamp: Option<Instant> = None;
1064
1065	// Get dev_receiver if available for development refresh messages.
1066	let mut dev_receiver = res!(self.handler.dev_receiver(id));
1067
1068	let mut interval = ping_interval.map(\|interval\| {
1069	let duration = tokio::time::Duration::from_secs(interval as u64);
1070	tokio::time::interval(duration)
1071	});
1072
1073	// Start pinging at t = dt not t = 0.
1074	if let Some(interval) = &mut interval {
1075	interval.tick().await;
1076	}
1077
1078	loop {
1079	tokio::select! {
1080	result = self.read() => {
1081	match result {
1082	Ok(Some(msg)) => {
1083	match msg {
1084	WebSocketMessage::Text(txt) => {
1085	let result = self.handler.handle_text(
1086	txt,
1087	db.clone(),
1088	syntax.clone(),
1089	id,
1090	);
1091	let result = self.response_handler(
1092	result,
1093	&mut err_count,
1094	max_errors,
1095	"text",
1096	id,
1097	).await;
1098	//res!(result);
1099	if let Err(e) = result {
1100	error!(e);
1101	continue;
1102	}
1103	}
1104	WebSocketMessage::Binary(byts) => {
1105	let result = self.handler.handle_binary(
1106	byts,
1107	db.clone(),
1108	syntax.clone(),
1109	id,
1110	);
1111	let result = self.response_handler(
1112	result,
1113	&mut err_count,
1114	max_errors,
1115	"binary",
1116	id,
1117	).await;
1118	res!(result);
1119	}
1120	WebSocketMessage::Ping(byts) => {
1121	let result = self.response_handler(
1122	Ok(Some(WebSocketMessage::Pong(byts))),
1123	&mut err_count,
1124	max_errors,
1125	"ping",
1126	id,
1127	).await;
1128	res!(result);
1129	}
1130	WebSocketMessage::Pong(_byts) => {
1131	if let Some(timestamp) = ping_timestamp {
1132	let latency = timestamp.elapsed().as_millis();
1133	self.latency.set_and_adv(match u16::try_from(latency) {
1134	Ok(nu16) => Some(nu16),
1135	Err(_) => None,
1136	});
1137	ping_timestamp = None;
1138	} else {
1139	warn!("{}: Received unsolicited pong message.", id);
1140	}
1141	}
1142	WebSocketMessage::Close(status_code, reason) => {
1143	let result = self.close(status_code, reason).await;
1144	if let Err(e) = result {
1145	error!(err!(e,
1146	"{}: Error during WebSocket close:", id;
1147	IO, Network, Wire, Write));
1148	}
1149	break;
1150	}
1151	}
1152	}
1153	Ok(None) => {
1154	break;
1155	}
1156	Err(e) => {
1157	let too_big = e.tags().contains(&ErrTag::TooBig);
1158	let e = err!(e,
1159	"{}: Error reading websocket message:", id;
1160	IO, Network, Wire, Read);
1161	// A message over the limits was refused part way through a frame, so
1162	// the stream no longer starts on a frame boundary and every further
1163	// read would be of payload mistaken for a header. `read` has already
1164	// sent the 1009 close; the connection ends here rather than counting
1165	// this as one error among a permitted few.
1166	if too_big {
1167	error!(e);
1168	break;
1169	}
1170	let result = self.response_handler(
1171	Err(e),
1172	&mut err_count,
1173	max_errors,
1174	"",
1175	id,
1176	).await;
1177	//res!(result);
1178	if let Err(e) = result {
1179	error!(e);
1180	continue;
1181	}
1182	}
1183	}
1184	}
1185	// Development refresh notifications.
1186	_ = async {
1187	if let Some(receiver) = &mut dev_receiver {
1188	// If a () message is received here...
1189	receiver.recv().await.ok()
1190	} else {
1191	std::future::pending().await
1192	}
1193	} => {
1194	// ... a refresh message will be sent to the client here.
1195	let refresh = WebSocketMessage::Text(WSH::DEV_REFRESH_MSG.to_string());
1196	debug!("{}: POO Sending {:?}", id, refresh);
1197	if let Err(e) = self.send(&refresh).await {
1198	error!(err!(e,
1199	"{}: Error sending refresh message:", id;
1200	IO, Network, Wire, Write));
1201	}
1202	}
1203	// Pings.
1204	_ = async {
1205	if let Some(interval) = &mut interval {
1206	interval.tick().await;
1207	} else {
1208	tokio::time::sleep(std::time::Duration::from_secs(std::u64::MAX)).await;
1209	}
1210	} => {
1211	if let Some(_) = &interval {
1212	// Send a ping message.
1213	let now = res!(Timestamp::now());
1214	let ping_data = res!(now.to_bytes(Vec::new()));
1215	let ping = WebSocketMessage::Ping(ping_data);
1216	let result = self.send(&ping).await;
1217	if let Err(e) = result {
1218	let e = err!(e,
1219	"{}: Error sending ping message:", id;
1220	IO, Network, Wire, Write);
1221	let result = self.response_handler(Err(e), &mut err_count, max_errors, "", id).await;
1222	if let Err(e) = result {
1223	error!(e);
1224	continue;
1225	}
1226	} else {
1227	ping_timestamp = Some(Instant::now());
1228	}
1229	}
1230	}
1231	}
1232	}
1233
1234	Ok(())
1235	}
1236
1237	}
1238
1239	#[cfg(test)]
1240	mod tests {
1241	use super::*;
1242
1243	/// RFC 6455 §1.3 publishes a worked example of the handshake, and this is its key pair. The
1244	/// expected value comes from the RFC, not from us, which is the whole point: a browser
1245	/// recomputes it independently, so agreeing with ourselves proves nothing.
1246	#[test]
1247	fn test_accept_key_rfc6455_vector_00() {
1248	assert_eq!(
1249	accept_key("dGhlIHNhbXBsZSBub25jZQ=="),
1250	"s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",
1251	);
1252	}
1253
1254	/// The GUID is byte-exact per RFC 6455 §4.2.2, and appending it is what distinguishes the
1255	/// accept value from a bare hash of the key. Pinned so that a "tidy-up" cannot silently
1256	/// break every handshake.
1257	#[test]
1258	fn test_accept_key_guid_is_appended_00() {
1259	assert_eq!(constant::WEBSOCKET_GUID, "258EAFA5-E914-47DA-95CA-C5AB0DC85B11");
1260	// Without the GUID the digest differs, so this must not equal the vector above.
1261	let mut hasher = Sha1::new();
1262	hasher.update(b"dGhlIHNhbXBsZSBub25jZQ==");
1263	assert_ne!(base64::encode(&hasher.finalize()), "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=");
1264	}
1265
1266	/// The upgrade request RFC 6455 §1.2 prints, answered. The accept value is the RFC's, and the
1267	/// three header lines are the ones a browser insists on before it will call the socket open.
1268	#[test]
1269	fn test_accept_response_rfc6455_request_00() -> Outcome<()> {
1270	let req = HttpMessage {
1271	header: res!(HttpHeader::parse(fmt!(
1272	"GET /chat HTTP/1.1\r\n\
1273	Host: server.example.com\r\n\
1274	Upgrade: websocket\r\n\
1275	Connection: Upgrade\r\n\
1276	Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
1277	Sec-WebSocket-Version: 13\r\n\r\n"), Some(true))),
1278	body: Vec::new(),
1279	head_only: false,
1280	file: None,
1281	};
1282	let response = res!(accept_response(&req));
1283	assert_eq!(
1284	response,
1285	"HTTP/1.1 101 Switching Protocols\r\n\
1286	Upgrade: websocket\r\n\
1287	Connection: Upgrade\r\n\
1288	Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n",
1289	);
1290	Ok(())
1291	}
1292
1293	/// A request with no key at all cannot be answered, and must say so rather than answer with a
1294	/// digest of nothing.
1295	#[test]
1296	fn test_accept_response_rejects_missing_key_00() -> Outcome<()> {
1297	let req = HttpMessage {
1298	header: res!(HttpHeader::parse(fmt!(
1299	"GET /chat HTTP/1.1\r\n\
1300	Host: server.example.com\r\n\
1301	Upgrade: websocket\r\n\
1302	Connection: Upgrade\r\n\
1303	Sec-WebSocket-Version: 13\r\n\r\n"), Some(true))),
1304	body: Vec::new(),
1305	head_only: false,
1306	file: None,
1307	};
1308	assert!(accept_response(&req).is_err(),
1309	"a handshake with no Sec-WebSocket-Key must be refused");
1310	Ok(())
1311	}
1312
1313	/// A server frame is unmasked and, for a short payload, exactly two header bytes: RFC 6455
1314	/// §5.2 fixes FIN\|opcode then the length. The bytes below are what a browser's own decoder
1315	/// expects to see, so they are written out rather than recomputed.
1316	#[test]
1317	fn test_encode_message_server_text_frame_00() -> Outcome<()> {
1318	let byts = res!(encode_message(
1319	&WebSocketMessage::Text("hello".to_string()), false, 1024, 4096,
1320	));
1321	assert_eq!(byts, vec![0x81, 0x05, b'h', b'e', b'l', b'l', b'o']);
1322	Ok(())
1323	}
1324
1325	/// A client frame carries the mask bit and a four-byte key, and the payload is the plaintext
1326	/// XORed with it. RFC 6455 §5.3 requires this of every client-to-server frame.
1327	#[test]
1328	fn test_encode_message_client_masks_payload_00() -> Outcome<()> {
1329	let byts = res!(encode_message(
1330	&WebSocketMessage::Text("hello".to_string()), true, 1024, 4096,
1331	));
1332	assert_eq!(byts[0], 0x81);
1333	assert_eq!(byts[1], 0x80 \| 0x05, "the mask bit must be set on a client frame");
1334	assert_eq!(byts.len(), 2 + 4 + 5);
1335	let key = &byts[2..6];
1336	let unmasked: Vec<u8> = byts[6..].iter().enumerate()
1337	.map(\|(i, b)\| b ^ key[i % 4])
1338	.collect();
1339	assert_eq!(unmasked, b"hello".to_vec());
1340	Ok(())
1341	}
1342
1343	/// What one end frames, the other end reads: a masked client message decodes back to the same
1344	/// text on the server side.
1345	#[tokio::test]
1346	async fn test_read_message_round_trip_masked_00() -> Outcome<()> {
1347	let byts = res!(encode_message(
1348	&WebSocketMessage::Text("the ceremony begins".to_string()), true, 1024, 4096,
1349	));
1350	let mut src = &byts[..];
1351	let mut buffer = Vec::new();
1352	match res!(read_message(&mut src, &mut buffer, 1024, WebSocketLimits::default()).await) {
1353	Some(WebSocketMessage::Text(txt)) => assert_eq!(txt, "the ceremony begins"),
1354	other => return Err(err!(
1355	"Expected a text message, got {:?}.", other; Test, Mismatch)),
1356	}
1357	assert!(buffer.is_empty(), "the buffer must be left ready for the next message");
1358	Ok(())
1359	}
1360
1361	/// A payload past the chunking threshold goes out as several frames -- first opcode, then
1362	/// continuations, `FIN` only on the last -- and arrives as one message.
1363	#[tokio::test]
1364	async fn test_read_message_reassembles_fragments_00() -> Outcome<()> {
1365	let payload: String = std::iter::repeat('x').take(300).collect();
1366	let byts = res!(encode_message(
1367	&WebSocketMessage::Binary(payload.as_bytes().to_vec()), false, 100, 100,
1368	));
1369	// Four frames of 100 bytes' payload, each with a two-byte header.
1370	assert_eq!(byts.len(), 3 * (2 + 100));
1371	assert_eq!(byts[0], 0x02, "the first frame carries the opcode and no FIN");
1372	assert_eq!(byts[102], 0x00, "a middle frame carries the continuation opcode");
1373	assert_eq!(byts[204], 0x80, "the last frame carries FIN and the continuation opcode");
1374	let mut src = &byts[..];
1375	let mut buffer = Vec::new();
1376	match res!(read_message(&mut src, &mut buffer, 64, WebSocketLimits::default()).await) {
1377	Some(WebSocketMessage::Binary(got)) => {
1378	assert_eq!(got.len(), 300);
1379	assert_eq!(got, payload.as_bytes().to_vec());
1380	},
1381	other => return Err(err!(
1382	"Expected a binary message, got {:?}.", other; Test, Mismatch)),
1383	}
1384	Ok(())
1385	}
1386
1387	/// A ping between the fragments of a message (RFC 6455 §5.4) belongs to no message. Before
1388	/// `read_frame` its payload was appended to the text and its FIN ended the message early.
1389	#[tokio::test]
1390	async fn test_read_message_keeps_interleaved_ping_out_of_message_00() -> Outcome<()> {
1391	let frags = res!(encode_message(
1392	&WebSocketMessage::Text(fmt!("abcdefghij")), false, 5, 5,
1393	));
1394	let ping = res!(encode_message(&WebSocketMessage::Ping(b"hb".to_vec()), false, 125, 125));
1395	let mut byts = frags[..7].to_vec();
1396	byts.extend_from_slice(&ping);
1397	byts.extend_from_slice(&frags[7..]);
1398	let mut src = &byts[..];
1399	let mut buffer = Vec::new();
1400	match res!(read_message(&mut src, &mut buffer, 64, WebSocketLimits::default()).await) {
1401	Some(WebSocketMessage::Text(got)) => assert_eq!(got, "abcdefghij"),
1402	other => return Err(err!(
1403	"Expected the whole text message, got {:?}.", other; Test, Mismatch)),
1404	}
1405	assert!(buffer.is_empty());
1406	Ok(())
1407	}
1408
1409	/// A new data frame while a message is still being continued is a protocol error, not the
1410	/// start of a merged message.
1411	#[tokio::test]
1412	async fn test_read_message_refuses_data_frame_mid_message_00() -> Outcome<()> {
1413	// Text "ab" without FIN, then a whole binary frame.
1414	let byts = vec![0x01, 0x02, b'a', b'b', 0x82, 0x01, b'c'];
1415	let mut src = &byts[..];
1416	let mut buffer = Vec::new();
1417	assert!(read_message(&mut src, &mut buffer, 64, WebSocketLimits::default()).await.is_err(),
1418	"a data frame interrupting a fragmented message must be refused");
1419	Ok(())
1420	}
1421
1422	/// A control frame is never fragmented (RFC 6455 §5.5).
1423	#[tokio::test]
1424	async fn test_read_frame_refuses_fragmented_control_frame_00() -> Outcome<()> {
1425	let byts = vec![0x09, 0x00];
1426	let mut src = &byts[..];
1427	assert!(read_frame(&mut src, 64, WebSocketLimits::default(), 0).await.is_err());
1428	Ok(())
1429	}
1430
1431	/// A first frame carrying the continuation opcode continues nothing. It must be an error: this
1432	/// arrives from whoever is on the other end of the socket, and a panic there is a server a
1433	/// stranger can stop.
1434	#[tokio::test]
1435	async fn test_read_message_rejects_lone_continuation_00() -> Outcome<()> {
1436	let byts = vec![0x80, 0x02, b'h', b'i'];
1437	let mut src = &byts[..];
1438	let mut buffer = Vec::new();
1439	assert!(read_message(&mut src, &mut buffer, 1024, WebSocketLimits::default()).await.is_err(),
1440	"a lone continuation frame must be refused, not unwound");
1441	Ok(())
1442	}
1443
1444	/// A closed connection reads as the end of the stream, not as an error.
1445	#[tokio::test]
1446	async fn test_read_message_eof_is_none_00() -> Outcome<()> {
1447	let byts: Vec<u8> = Vec::new();
1448	let mut src = &byts[..];
1449	let mut buffer = Vec::new();
1450	assert!(res!(read_message(&mut src, &mut buffer, 1024, WebSocketLimits::default()).await)
1451	.is_none(),
1452	"an immediately-closed stream must read as None");
1453	Ok(())
1454	}
1455
1456	/// The header of a frame that declares more than the connection accepts, with as much of the
1457	/// payload as `tail` says following it. Written by hand rather than by `encode_message`,
1458	/// because a length no honest sender would write is the whole point.
1459	fn oversize_frame_header(declared: u64, tail: &[u8]) -> Vec<u8> {
1460	let mut byts = vec![0x82]; // FIN, binary.
1461	byts.push(127); // 64-bit extended length follows.
1462	byts.extend_from_slice(&declared.to_be_bytes());
1463	byts.extend_from_slice(tail);
1464	byts
1465	}
1466
1467	/// A frame declaring more than the limit allows is refused, and refused on the strength of the
1468	/// declaration alone: the reader stops at the header, so the bytes that followed it are still
1469	/// unread when the error comes back.
1470	///
1471	/// Before the bound existed this returned `Ok(None)` -- the reader allocated the megabyte the
1472	/// frame asked for, found the stream ended, and reported a closed connection.
1473	#[tokio::test]
1474	async fn test_read_message_refuses_oversize_frame_00() -> Outcome<()> {
1475	let limits = WebSocketLimits::new(1_024);
1476	let byts = oversize_frame_header(1_025, b"payload");
1477	let mut src = &byts[..];
1478	let mut buffer = Vec::new();
1479	let result = read_message(&mut src, &mut buffer, 256, limits).await;
1480	match result {
1481	// Tagged once: `Error::tags` reads the whole chain, so no frame on the way out repeats it.
1482	Err(e) => assert_eq!(e.tags().iter().filter(\|t\| **t == ErrTag::TooBig).count(), 1,
1483	"an over-limit frame must be tagged TooBig, once, so that a 1009 close can answer it; \
1484	got tags {:?}", e.tags()),
1485	Ok(other) => return Err(err!(
1486	"Expected a frame of 1025 bytes to be refused against a 1024 byte limit, got \
1487	{:?}.", other; Test, Mismatch)),
1488	}
1489	assert_eq!(src.len(), 7,
1490	"the seven payload bytes must be left unread: the frame was refused on its declared \
1491	length, before anything was reserved or read for it");
1492	assert!(buffer.is_empty(), "nothing of a refused frame belongs in the message buffer");
1493	Ok(())
1494	}
1495
1496	/// A frame declaring more bytes than the machine has memory for is refused just the same, and
1497	/// the run of this test is itself the proof that the check precedes the allocation: a reader
1498	/// that allocated first would abort the process here, and an abort cannot be caught or
1499	/// reported. Nothing but a check before the `vec!` can make this test pass.
1500	#[tokio::test]
1501	async fn test_read_message_refuses_impossible_length_00() -> Outcome<()> {
1502	// The largest length RFC 6455 §5.2 permits: the most significant bit must be clear.
1503	let byts = oversize_frame_header(0x7FFF_FFFF_FFFF_FFFF, &[]);
1504	let mut src = &byts[..];
1505	let mut buffer = Vec::new();
1506	assert!(read_message(&mut src, &mut buffer, 1_024, WebSocketLimits::default()).await
1507	.is_err(),
1508	"a frame declaring eight exabytes must be refused, not reserved for");
1509	Ok(())
1510	}
1511
1512	/// A large frame that is nonetheless within the limits still arrives whole. A bound that is
1513	/// only ever proved by what it rejects could be a reader that rejects everything.
1514	#[tokio::test]
1515	async fn test_read_message_allows_large_frame_within_limits_00() -> Outcome<()> {
1516	let payload: Vec<u8> = (0..200_000u32).map(\|i\| (i % 251) as u8).collect();
1517	// One frame, so the 64-bit length branch is the one exercised.
1518	let byts = res!(encode_message(
1519	&WebSocketMessage::Binary(payload.clone()), false, 1_000_000, 1_000_000,
1520	));
1521	assert_eq!(byts[1] & 0x7F, 127, "a payload this size is framed with a 64-bit length");
1522	let limits = WebSocketLimits::new(256 * 1_024);
1523	let mut src = &byts[..];
1524	let mut buffer = Vec::new();
1525	match res!(read_message(&mut src, &mut buffer, 4_096, limits).await) {
1526	Some(WebSocketMessage::Binary(got)) => assert_eq!(got, payload),
1527	other => return Err(err!(
1528	"Expected a 200,000 byte binary message, got {:?}.", other; Test, Mismatch)),
1529	}
1530	Ok(())
1531	}
1532
1533	/// Frames that each pass the frame bound can still add up past the message bound, since a
1534	/// message may be fragmented into as many continuations as the peer likes. The message bound
1535	/// is what stops the total, and it stops it at the frame that would breach it rather than
1536	/// after.
1537	#[tokio::test]
1538	async fn test_read_message_refuses_oversize_message_00() -> Outcome<()> {
1539	// Four frames of 1,000 bytes each: every one of them inside the 1,024 byte frame bound,
1540	// and together over the 2,048 byte message bound.
1541	let payload = vec![b'x'; 4_000];
1542	let byts = res!(encode_message(
1543	&WebSocketMessage::Binary(payload), false, 1_000, 1_000,
1544	));
1545	assert_eq!(byts[0], 0x02, "the first of several frames carries the opcode and no FIN");
1546	let limits = WebSocketLimits {
1547	max_frame: 1_024,
1548	max_msg: 2_048,
1549	};
1550	let mut src = &byts[..];
1551	let mut buffer = Vec::new();
1552	match read_message(&mut src, &mut buffer, 256, limits).await {
1553	Err(e) => assert_eq!(e.tags().iter().filter(\|t\| **t == ErrTag::TooBig).count(), 1,
1554	"an over-limit message must be tagged TooBig, once; got tags {:?}", e.tags()),
1555	Ok(other) => return Err(err!(
1556	"Expected four 1,000 byte frames to breach a 2,048 byte message limit, got {:?}.",
1557	other; Test, Mismatch)),
1558	}
1559	// Two frames were taken; the third was refused on its four-byte header, so the rest of
1560	// that frame and the whole of the fourth are still unread.
1561	assert_eq!(src.len(), 2 * (4 + 1_000) - 4,
1562	"the reader must stop at the header of the frame that would breach the limit");
1563	Ok(())
1564	}
1565
1566	/// Every frame of a fragmented message is bounded, not merely the first: a peer that opens
1567	/// with a modest frame and continues with an enormous one is refused on the continuation.
1568	#[tokio::test]
1569	async fn test_read_message_bounds_continuation_frames_00() -> Outcome<()> {
1570	let mut byts = vec![0x02, 0x02, b'h', b'i']; // Binary, no FIN, two bytes.
1571	// A continuation frame with FIN set, declaring far more than the bound.
1572	byts.push(0x80);
1573	byts.push(127);
1574	byts.extend_from_slice(&(1u64 << 40).to_be_bytes());
1575	let mut src = &byts[..];
1576	let mut buffer = Vec::new();
1577	assert!(read_message(&mut src, &mut buffer, 256, WebSocketLimits::new(4_096)).await.is_err(),
1578	"a continuation frame is as capable of declaring a huge payload as a first frame");
1579	Ok(())
1580	}
1581
1582	/// RFC 6455 §5.5 caps a control frame at 125 bytes, so a ping declaring more than that is
1583	/// malformed whatever the connection's own limits say. Checked separately because a generous
1584	/// limit would otherwise let a peer make the reader hold megabytes for a frame the protocol
1585	/// says is small.
1586	#[tokio::test]
1587	async fn test_read_message_refuses_oversize_control_frame_00() -> Outcome<()> {
1588	// FIN, ping, 16-bit length of 126 -- one byte over what a control frame may carry.
1589	let mut byts = vec![0x89, 126];
1590	byts.extend_from_slice(&126u16.to_be_bytes());
1591	byts.extend_from_slice(&vec![0u8; 126]);
1592	let mut src = &byts[..];
1593	let mut buffer = Vec::new();
1594	assert!(read_message(&mut src, &mut buffer, 256, WebSocketLimits::default()).await.is_err(),
1595	"a control frame over 125 bytes must be refused");
1596	Ok(())
1597	}
1598
1599	/// The defaults bound both dimensions, and the message bound is the looser of the two -- a
1600	/// message bound below the frame bound would make the frame bound unreachable and the pair
1601	/// misleading.
1602	#[test]
1603	fn test_websocket_limits_defaults_00() {
1604	let limits = WebSocketLimits::default();
1605	assert_eq!(limits.max_frame, constant::WEBSOCKET_MAX_FRAME_BYTES);
1606	assert_eq!(limits.max_msg, constant::WEBSOCKET_MAX_MESSAGE_BYTES);
1607	assert!(limits.max_msg >= limits.max_frame,
1608	"a message must be allowed to be at least as large as one frame of it");
1609	}
1610
1611	/// The declared length is checked as the 64-bit number it arrived as. Narrowed to a `usize`
1612	/// first, this length would be one byte on a 32-bit target and would pass any bound at all.
1613	#[test]
1614	fn test_websocket_limits_check_frame_is_64_bit_00() -> Outcome<()> {
1615	let limits = WebSocketLimits::new(1_024);
1616	assert!(limits.check_frame((1u64 << 32) + 1).is_err(),
1617	"a length that truncates to 1 in 32 bits must still be refused");
1618	assert!(limits.check_frame(1_024).is_ok(), "a length exactly at the bound is allowed");
1619	assert!(limits.check_frame(1_025).is_err(), "a length one over the bound is refused");
1620	Ok(())
1621	}
1622	}