Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_net/tests/ws_client.rs

13.4 KiB, 1 run

created by r1870400018:59661, 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#![cfg(feature = "async")]
2//! `WsClient` against a hand-written tokio server on loopback. The server side is built from the
3//! crate's framing functions plus raw bytes, and every client frame it reads is checked for the
4//! mask RFC 6455 §5.3 requires.
5
6use oxedyne_fe2o3_core::prelude::*;
7use oxedyne_fe2o3_net::ws::{
8 accept_key,
9 encode_message,
10 read_frame,
11 status::WebSocketStatusCode,
12 WebSocketFrame,
13 WebSocketLimits,
14 WebSocketMessage,
15 WsClient,
16};
17
18use std::time::Duration;
19
20use tokio::{
21 io::{
22 AsyncReadExt,
23 AsyncWriteExt,
24 },
25 net::{
26 TcpListener,
27 TcpStream,
28 },
29 task::JoinHandle,
30};
31
32
33const WAIT: Duration = Duration::from_secs(5);
34
35/// Binds a loopback listener and runs `serve` on the first connection it accepts, returning the
36/// address to connect to and the server task.
37async fn spawn_server<F, Fut>(serve: F) -> Outcome<(String, JoinHandle<Outcome<()>>)>
38where
39 F: FnOnce(TcpStream) -> Fut + Send + 'static,
40 Fut: std::future::Future<Output = Outcome<()>> + Send + 'static,
41{
42 let listener = res!(TcpListener::bind("127.0.0.1:0").await);
43 let addr = res!(listener.local_addr());
44 let task = tokio::spawn(async move {
45 let (stream, _) = res!(listener.accept().await);
46 serve(stream).await
47 });
48 Ok((addr.to_string(), task))
49}
50
51/// Reads the upgrade request up to its blank line, byte by byte so that nothing after it is
52/// consumed.
53async fn read_request(stream: &mut TcpStream) -> Outcome<String> {
54 let mut head = Vec::new();
55 while !head.ends_with(b"\r\n\r\n") {
56 let mut b = [0u8; 1];
57 res!(stream.read_exact(&mut b).await);
58 head.push(b[0]);
59 }
60 Ok(res!(String::from_utf8(head)))
61}
62
63fn request_key(request: &str) -> Outcome<String> {
64 for line in request.lines() {
65 if let Some((name, value)) = line.split_once(':') {
66 if name.trim().eq_ignore_ascii_case("sec-websocket-key") {
67 return Ok(value.trim().to_string());
68 }
69 }
70 }
71 Err(err!("The upgrade request has no Sec-WebSocket-Key:\n{}", request; Test, Missing))
72}
73
74fn switching(accept: &str) -> String {
75 fmt!("HTTP/1.1 101 Switching Protocols\r\n\
76 Upgrade: websocket\r\n\
77 Connection: Upgrade\r\n\
78 Sec-WebSocket-Accept: {}\r\n\r\n", accept)
79}
80
81/// Completes the handshake properly, returning the request text.
82async fn accept(stream: &mut TcpStream) -> Outcome<String> {
83 let request = res!(read_request(stream).await);
84 let key = res!(request_key(&request));
85 res!(stream.write_all(switching(&accept_key(&key)).as_bytes()).await);
86 Ok(request)
87}
88
89/// Reads one client frame, insisting that it is masked.
90async fn read_masked(stream: &mut TcpStream) -> Outcome<WebSocketFrame> {
91 match res!(read_frame(stream, 1_024, WebSocketLimits::default(), 0).await) {
92 Some(frame) => {
93 if !frame.masked {
94 return Err(err!(
95 "The client sent an unmasked frame with opcode {:#x}.", frame.opcode;
96 Test, Invalid));
97 }
98 Ok(frame)
99 }
100 None => Err(err!("The client closed before sending a frame."; Test, Missing)),
101 }
102}
103
104async fn join(task: JoinHandle<Outcome<()>>) -> Outcome<()> {
105 match task.await {
106 Ok(outcome) => outcome,
107 Err(e) => Err(err!(e, "The test server task failed."; Test)),
108 }
109}
110
111#[tokio::test]
112async fn test_ws_client_handshake_and_echo_00() -> Outcome<()> {
113 let (addr, server) = res!(spawn_server(|mut stream| async move {
114 let request = res!(accept(&mut stream).await);
115 let addr = res!(stream.local_addr());
116 // The path goes on the request line and the Host field carries the authority alone.
117 // Field names are case-insensitive (RFC 9110 §5.1).
118 let lower = request.to_lowercase();
119 if !request.starts_with("GET /peer HTTP/1.1\r\n")
120 || !lower.contains(&fmt!("\r\nhost: {}\r\n", addr))
121 || !lower.contains("\r\norigin: https://test.example\r\n")
122 {
123 return Err(err!("Unexpected upgrade request:\n{}", request; Test, Mismatch));
124 }
125 let frame = res!(read_masked(&mut stream).await);
126 let echo = res!(encode_message(
127 &WebSocketMessage::Text(res!(String::from_utf8(frame.payload))), false, 1_024, 1_024));
128 res!(stream.write_all(&echo).await);
129 Ok(())
130 }).await);
131
132 let mut client = res!(WsClient::connect(&addr, "/peer", Some("https://test.example")).await);
133 res!(client.send_text("hello peer").await);
134 match res!(client.recv(WAIT).await) {
135 Some(WebSocketMessage::Text(txt)) => req!(txt, fmt!("hello peer")),
136 other => return Err(err!("Expected the echo, got {:?}.", other; Test, Mismatch)),
137 }
138 join(server).await
139}
140
141#[tokio::test]
142async fn test_ws_client_refusal_names_status_00() -> Outcome<()> {
143 let (addr, server) = res!(spawn_server(|mut stream| async move {
144 res!(read_request(&mut stream).await);
145 res!(stream.write_all(
146 b"HTTP/1.1 400 Bad Request\r\nContent-Length: 4\r\n\r\nnope").await);
147 Ok(())
148 }).await);
149
150 match WsClient::connect(&addr, "/peer", None).await {
151 Ok(_) => return Err(err!("A 400 refusal was taken as a handshake."; Test, Unexpected)),
152 Err(e) => {
153 let msg = fmt!("{}", e);
154 if !msg.contains("400") {
155 return Err(err!("The refusal error does not name the status: {}", msg;
156 Test, Mismatch));
157 }
158 }
159 }
160 join(server).await
161}
162
163#[tokio::test]
164async fn test_ws_client_wrong_accept_key_refused_00() -> Outcome<()> {
165 let wrong = accept_key("AAAAAAAAAAAAAAAAAAAAAA==");
166 let wrong_srv = wrong.clone();
167 let (addr, server) = res!(spawn_server(|mut stream| async move {
168 res!(read_request(&mut stream).await);
169 res!(stream.write_all(switching(&wrong_srv).as_bytes()).await);
170 Ok(())
171 }).await);
172
173 match WsClient::connect(&addr, "/peer", None).await {
174 Ok(_) => return Err(err!(
175 "A 101 carrying the wrong Sec-WebSocket-Accept was taken as a handshake.";
176 Test, Unexpected)),
177 Err(e) => {
178 let msg = fmt!("{}", e);
179 if !msg.contains(&wrong) {
180 return Err(err!("The key error does not name the key received: {}", msg;
181 Test, Mismatch));
182 }
183 }
184 }
185 join(server).await
186}
187
188/// A text message in three fragments with a ping between the first two: the client must answer
189/// the ping at once and still deliver the text whole.
190#[tokio::test]
191async fn test_ws_client_reassembles_fragments_and_answers_ping_00() -> Outcome<()> {
192 let text: String = "0123456789abcdefghij".to_string();
193 let text_srv = text.clone();
194 let (addr, server) = res!(spawn_server(|mut stream| async move {
195 res!(accept(&mut stream).await);
196 // Frames of 8 payload bytes, each with a two-byte header: 10 + 10 + 6.
197 let frags = res!(encode_message(&WebSocketMessage::Text(text_srv), false, 8, 8));
198 if frags.len() != 26 || frags[0] != 0x01 || frags[10] != 0x00 || frags[20] != 0x80 {
199 return Err(err!("Unexpected fragmentation: {:02x?}", frags; Test, Mismatch));
200 }
201 let ping = res!(encode_message(&WebSocketMessage::Ping(b"hb".to_vec()), false, 125, 125));
202 res!(stream.write_all(&frags[..10]).await);
203 res!(stream.write_all(&ping).await);
204 // The pong must come back while the message is still incomplete.
205 let pong = res!(read_masked(&mut stream).await);
206 if pong.opcode != 0xA || pong.payload != b"hb" {
207 return Err(err!("Expected a pong echoing 'hb', got {:?}.", pong; Test, Mismatch));
208 }
209 res!(stream.write_all(&frags[10..]).await);
210 Ok(())
211 }).await);
212
213 let mut client = res!(WsClient::connect(&addr, "/", None).await);
214 match res!(client.recv(WAIT).await) {
215 Some(WebSocketMessage::Text(got)) => req!(got, text),
216 other => return Err(err!("Expected the reassembled text, got {:?}.", other;
217 Test, Mismatch)),
218 }
219 join(server).await
220}
221
222/// A frame over the client's limit is refused before its payload is read, the server hears a
223/// 1009 close, and the client is left unusable.
224#[tokio::test]
225async fn test_ws_client_refuses_over_limit_frame_00() -> Outcome<()> {
226 let (addr, server) = res!(spawn_server(|mut stream| async move {
227 res!(accept(&mut stream).await);
228 let big = res!(encode_message(
229 &WebSocketMessage::Binary(vec![7u8; 1_000]), false, 4_096, 4_096));
230 res!(stream.write_all(&big).await);
231 let close = res!(read_masked(&mut stream).await);
232 let code = u16::from_be_bytes([close.payload[0], close.payload[1]]);
233 if close.opcode != 0x8 || code != 1009 {
234 return Err(err!("Expected a 1009 close, got {:?}.", close; Test, Mismatch));
235 }
236 Ok(())
237 }).await);
238
239 let mut client = res!(WsClient::connect(&addr, "/", None).await)
240 .with_limits(WebSocketLimits::new(64));
241 match client.recv(WAIT).await {
242 Ok(msg) => return Err(err!("An over-limit frame was accepted: {:?}", msg;
243 Test, Unexpected)),
244 Err(e) => if !e.tags().contains(&ErrTag::TooBig) {
245 return Err(err!(e, "The refusal lacks the TooBig tag."; Test, Mismatch));
246 },
247 }
248 if client.is_open() {
249 return Err(err!("The client stayed open after a refused frame."; Test, Unexpected));
250 }
251 join(server).await
252}
253
254/// Text, binary, ping-answering pong and close all leave the client masked, and the close
255/// handshake completes when the server echoes.
256#[tokio::test]
257async fn test_ws_client_masks_every_frame_00() -> Outcome<()> {
258 let (addr, server) = res!(spawn_server(|mut stream| async move {
259 res!(accept(&mut stream).await);
260 let ping = res!(encode_message(&WebSocketMessage::Ping(vec![1, 2]), false, 125, 125));
261 res!(stream.write_all(&ping).await);
262 let mut opcodes = Vec::new();
263 loop {
264 let frame = res!(read_masked(&mut stream).await);
265 opcodes.push(frame.opcode);
266 if frame.opcode == 0x8 {
267 let echo = res!(encode_message(
268 &WebSocketMessage::Close(Some(WebSocketStatusCode::NormalClosure), None),
269 false, 125, 125));
270 res!(stream.write_all(&echo).await);
271 break;
272 }
273 }
274 req!(opcodes, vec![0xA, 0x1, 0x2, 0x8]);
275 Ok(())
276 }).await);
277
278 let mut client = res!(WsClient::connect(&addr, "/", None).await);
279 // Nothing but the ping arrives, so this times out having answered it.
280 match res!(client.recv(Duration::from_millis(300)).await) {
281 None => (),
282 Some(msg) => return Err(err!("Expected nothing, got {:?}.", msg; Test, Unexpected)),
283 }
284 res!(client.send_text("t").await);
285 res!(client.send(&WebSocketMessage::Binary(vec![9; 200])).await);
286 res!(client.close(WAIT).await);
287 if client.send_text("late").await.is_ok() {
288 return Err(err!("A closed client sent a message."; Test, Unexpected));
289 }
290 join(server).await
291}
292
293/// A deadline passing between two fragments loses nothing: the next call completes the message.
294#[tokio::test]
295async fn test_ws_client_timeout_keeps_partial_message_00() -> Outcome<()> {
296 let (addr, server) = res!(spawn_server(|mut stream| async move {
297 res!(accept(&mut stream).await);
298 let frags = res!(encode_message(
299 &WebSocketMessage::Binary(b"first-second".to_vec()), false, 6, 6));
300 res!(stream.write_all(&frags[..8]).await);
301 tokio::time::sleep(Duration::from_millis(400)).await;
302 res!(stream.write_all(&frags[8..]).await);
303 Ok(())
304 }).await);
305
306 let mut client = res!(WsClient::connect(&addr, "/", None).await);
307 match res!(client.recv(Duration::from_millis(100)).await) {
308 None => (),
309 Some(msg) => return Err(err!("Expected a timeout, got {:?}.", msg; Test, Unexpected)),
310 }
311 match res!(client.recv(WAIT).await) {
312 Some(WebSocketMessage::Binary(got)) => req!(got, b"first-second".to_vec()),
313 other => return Err(err!("Expected the whole message, got {:?}.", other;
314 Test, Mismatch)),
315 }
316 join(server).await
317}
318
319/// A server-initiated close is echoed with the same status and returned, and the client is then
320/// closed.
321#[tokio::test]
322async fn test_ws_client_echoes_server_close_00() -> Outcome<()> {
323 let (addr, server) = res!(spawn_server(|mut stream| async move {
324 res!(accept(&mut stream).await);
325 let close = res!(encode_message(
326 &WebSocketMessage::Close(Some(WebSocketStatusCode::GoingAway), Some(fmt!("bye"))),
327 false, 125, 125));
328 res!(stream.write_all(&close).await);
329 let echo = res!(read_masked(&mut stream).await);
330 let code = u16::from_be_bytes([echo.payload[0], echo.payload[1]]);
331 if echo.opcode != 0x8 || code != 1001 {
332 return Err(err!("Expected a 1001 close echo, got {:?}.", echo; Test, Mismatch));
333 }
334 Ok(())
335 }).await);
336
337 let mut client = res!(WsClient::connect(&addr, "/", None).await);
338 match res!(client.recv(WAIT).await) {
339 Some(WebSocketMessage::Close(Some(WebSocketStatusCode::GoingAway), Some(reason))) =>
340 req!(reason, fmt!("bye")),
341 other => return Err(err!("Expected the server's close, got {:?}.", other;
342 Test, Mismatch)),
343 }
344 if client.is_open() {
345 return Err(err!("The client stayed open after the server closed."; Test, Unexpected));
346 }
347 join(server).await
348}