1//! The relay as shipped, run as its own process and spoken to as a WebSocket.
2
3use relay::ws::{self, Message};
4use std::{
5 io::{BufRead, BufReader, Read, Write},
6 net::{SocketAddr, TcpStream},
7 process::{Child, Command, Stdio},
8 thread,
9 time::Duration,
10};
11
12/// A relay process, killed when dropped.
13struct Relay {
14 child: Child,
15 address: SocketAddr,
16}
17
18impl Relay {
19 fn start(options: &[&str]) -> Relay {
20 let mut child = Command::new(env!("CARGO_BIN_EXE_snowbound-relay"))
21 .args(["--listen", "127.0.0.1:0"])
22 .args(options)
23 .env_clear()
24 .stdout(Stdio::piped())
25 .spawn()
26 .unwrap();
27 let mut line = String::new();
28 BufReader::new(child.stdout.take().unwrap())
29 .read_line(&mut line)
30 .unwrap();
31 let address = line.trim().rsplit(' ').next().unwrap().parse().unwrap();
32 Relay { child, address }
33 }
34
35 fn get(&self, path: &str) -> String {
36 let mut stream = TcpStream::connect(self.address).unwrap();
37 write!(stream, "GET {path} HTTP/1.1\r\nHost: relay\r\n\r\n").unwrap();
38 let mut response = String::new();
39 stream.read_to_string(&mut response).unwrap();
40 response
41 }
42
43 /// Asks `method` on `path` with `body` on a connection of its own: the status line and
44 /// the answer.
45 fn ask(&self, method: &str, path: &str, body: &[u8]) -> (String, Vec<u8>) {
46 let mut stream = TcpStream::connect(self.address).unwrap();
47 write!(
48 stream,
49 "{method} {path} HTTP/1.1\r\nHost: relay\r\nContent-Length: {}\r\n\r\n",
50 body.len()
51 )
52 .unwrap();
53 stream.write_all(body).unwrap();
54 let head = ws::head(&mut stream).unwrap();
55 let length = ws::header(&head, "Content-Length")
56 .unwrap()
57 .parse()
58 .unwrap();
59 let mut answer = vec![0; length];
60 stream.read_exact(&mut answer).unwrap();
61 (head.lines().next().unwrap().to_owned(), answer)
62 }
63
64 fn join(&self, path: &str) -> Result<Peer, String> {
65 self.join_from(path, None)
66 }
67
68 /// Joins as if from `address`, as a proxy would say.
69 fn join_from(&self, path: &str, address: Option<&str>) -> Result<Peer, String> {
70 let mut stream = TcpStream::connect(self.address).unwrap();
71 stream
72 .set_read_timeout(Some(Duration::from_secs(10)))
73 .unwrap();
74 let forwarded = address
75 .map(|address| format!("X-Forwarded-For: {address}\r\n"))
76 .unwrap_or_default();
77 write!(
78 stream,
79 "GET {path} HTTP/1.1\r\nHost: relay\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
80 Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n{forwarded}\r\n"
81 )
82 .unwrap();
83 let mut reader = BufReader::new(stream.try_clone().unwrap());
84 let head = ws::head(&mut reader).map_err(|error| error.to_string())?;
85 if !head.starts_with("HTTP/1.1 101") {
86 return Err(head);
87 }
88 assert!(head.contains("s3pPLMBiTxaQ9kYGzzhZRbK+xOo="));
89 Ok(Peer {
90 stream,
91 reader: ws::Reader::new(reader, 1 << 20, false),
92 })
93 }
94}
95
96impl Drop for Relay {
97 fn drop(&mut self) {
98 let _ = self.child.kill();
99 let _ = self.child.wait();
100 }
101}
102
103struct Peer {
104 stream: TcpStream,
105 reader: ws::Reader<BufReader<TcpStream>>,
106}
107
108impl Peer {
109 fn send(&mut self, opcode: u8, payload: &[u8]) {
110 self.stream
111 .write_all(&ws::frame(opcode, payload, Some([1, 2, 3, 4])))
112 .unwrap();
113 }
114
115 fn say(&mut self, text: &str) {
116 self.send(ws::TEXT, text.as_bytes());
117 }
118
119 fn send_to(&mut self, slot: u32, bytes: &[u8]) {
120 self.send(ws::BINARY, &[&slot.to_be_bytes()[..], bytes].concat());
121 }
122
123 fn hear(&mut self) -> Message {
124 self.reader.read().unwrap()
125 }
126
127 fn text(&mut self) -> String {
128 match self.hear() {
129 Message::Text(text) => text,
130 other => panic!("heard {other:?}"),
131 }
132 }
133
134 /// Whether the relay has hung up, waiting at most `patience`.
135 fn closed(&mut self, patience: Duration) -> bool {
136 self.stream.set_read_timeout(Some(patience)).unwrap();
137 loop {
138 match self.reader.read() {
139 Ok(_) => continue,
140 Err(error) => {
141 return !matches!(
142 error.kind(),
143 std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
144 );
145 }
146 }
147 }
148 }
149}
150
151fn status(refusal: Result<Peer, String>) -> String {
152 match refusal {
153 Ok(_) => panic!("joined"),
154 Err(head) => head.lines().next().unwrap().to_owned(),
155 }
156}
157
158/// The number of the code a claim was given.
159fn claim(relay: &Relay) -> (Peer, u32) {
160 let mut owner = relay.join("/v1/claim").unwrap();
161 let number = owner
162 .text()
163 .strip_prefix("nameplate ")
164 .unwrap()
165 .parse()
166 .unwrap();
167 assert_eq!(owner.text(), "welcome 1");
168 (owner, number)
169}
170
171#[test]
172fn health_answers() {
173 let relay = Relay::start(&[]);
174 let response = relay.get("/health");
175 assert!(response.starts_with("HTTP/1.1 200"), "{response}");
176 assert!(response.contains("\"rooms\":0"), "{response}");
177 assert!(relay.get("/nothing").starts_with("HTTP/1.1 404"));
178 assert!(relay.get("/v1/room/abc").starts_with("HTTP/1.1 400"));
179}
180
181/// Peers in a room hear who comes and goes and pass each other messages, which arrive
182/// labelled with the sender's slot.
183#[test]
184fn peers_in_a_room_pass_messages() {
185 let relay = Relay::start(&[]);
186 let mut ada = relay.join("/v1/room/0123abcd").unwrap();
187 assert_eq!(ada.text(), "welcome 1");
188 let mut grace = relay.join("/v1/room/0123abcd").unwrap();
189 assert_eq!(grace.text(), "welcome 2 1");
190 assert_eq!(ada.text(), "joined 2");
191 grace.send_to(1, b"hello");
192 assert_eq!(
193 ada.hear(),
194 Message::Binary([&2u32.to_be_bytes()[..], b"hello"].concat())
195 );
196 ada.send_to(2, b"hi");
197 assert_eq!(
198 grace.hear(),
199 Message::Binary([&1u32.to_be_bytes()[..], b"hi"].concat())
200 );
201 ada.send(ws::PING, b"p");
202 assert_eq!(ada.hear(), Message::Pong);
203 let health = relay.get("/health");
204 assert!(health.contains("\"rooms\":1,\"peers\":2"), "{health}");
205 drop(grace);
206 assert_eq!(ada.text(), "left 2");
207}
208
209/// A group message goes once to the relay, which copies it to everyone else in the room or
210/// to the slots it names, marked as the group's; the relay counts what it passed on.
211#[test]
212fn group_messages_are_copied() {
213 let relay = Relay::start(&[]);
214 let mut peers: Vec<Peer> = (1..=3)
215 .map(|_| relay.join("/v1/room/0123abcd").unwrap())
216 .collect();
217 assert_eq!(peers[0].text(), "welcome 1");
218 assert_eq!(peers[0].text(), "joined 2");
219 assert_eq!(peers[0].text(), "joined 3");
220 assert_eq!(peers[1].text(), "welcome 2 1");
221 assert_eq!(peers[1].text(), "joined 3");
222 assert_eq!(peers[2].text(), "welcome 3 1 2");
223 let from = |slot: u32, bytes: &[u8]| {
224 Message::Binary([&(slot | relay::GROUP).to_be_bytes()[..], bytes].concat())
225 };
226 peers[0].send_to(relay::BROADCAST, b"everyone");
227 assert_eq!(peers[1].hear(), from(1, b"everyone"));
228 assert_eq!(peers[2].hear(), from(1, b"everyone"));
229 let named = [&3u32.to_be_bytes()[..], b"just you"].concat();
230 peers[1].send_to(relay::GROUP | 1, &named);
231 assert_eq!(peers[2].hear(), from(2, b"just you"));
232 peers[0].send_to(2, b"stream");
233 assert_eq!(
234 peers[1].hear(),
235 Message::Binary([&1u32.to_be_bytes()[..], b"stream"].concat())
236 );
237 let health = relay.get("/health");
238 let given = 2 * (4 + 8) + (4 + 8) + (4 + 6);
239 assert!(
240 health.contains(&format!("\"bytes_out\":{given}")),
241 "{health}"
242 );
243}
244
245/// One joining a code's room hears only its owner until the owner says it met it; then the
246/// others in the room, and they it.
247#[test]
248fn a_code_admits_whom_its_owner_met() {
249 let relay = Relay::start(&[]);
250 let (mut owner, number) = claim(&relay);
251 let path = format!("/v1/room/code-{number}");
252 let mut ada = relay.join(&path).unwrap();
253 assert_eq!(ada.text(), "welcome 2 1");
254 assert_eq!(owner.text(), "joined 2");
255 owner.say("met 2");
256 let mut grace = relay.join(&path).unwrap();
257 assert_eq!(grace.text(), "welcome 3 1");
258 assert_eq!(owner.text(), "joined 3");
259 // Ada can't reach Grace before the owner meets her.
260 ada.send_to(3, b"psst");
261 grace.send_to(1, b"hello");
262 assert_eq!(
263 owner.hear(),
264 Message::Binary([&3u32.to_be_bytes()[..], b"hello"].concat())
265 );
266 owner.say("met 3");
267 assert_eq!(grace.text(), "joined 2");
268 assert_eq!(ada.text(), "joined 3");
269 ada.send_to(3, b"hi");
270 assert_eq!(
271 grace.hear(),
272 Message::Binary([&2u32.to_be_bytes()[..], b"hi"].concat())
273 );
274 // Only the owner's word counts.
275 let mut mallory = relay.join(&path).unwrap();
276 assert_eq!(mallory.text(), "welcome 4 1");
277 ada.say("met 4");
278 grace.send_to(4, b"anyone?");
279 owner.say("failed 4");
280 assert!(mallory.closed(Duration::from_secs(5)));
281 assert_eq!(owner.text(), "joined 4");
282 assert_eq!(owner.text(), "left 4");
283}
284
285#[test]
286fn an_unknown_code_or_room_is_refused() {
287 let relay = Relay::start(&[]);
288 assert_eq!(
289 status(relay.join("/v1/room/code-5")),
290 "HTTP/1.1 404 Not Found"
291 );
292 assert_eq!(
293 status(relay.join("/v1/room/UPPER")),
294 "HTTP/1.1 404 Not Found"
295 );
296 assert_eq!(status(relay.join("/v1/room/")), "HTTP/1.1 404 Not Found");
297}
298
299/// Five wrong codes burn a code: it admits no one new, and its owner hears so. A wrong guess
300/// still hears what the owner said before refusing it.
301#[test]
302fn wrong_codes_burn_a_code() {
303 let relay = Relay::start(&["--failures-per-minute", "100"]);
304 let (mut owner, number) = claim(&relay);
305 let path = format!("/v1/room/code-{number}");
306 for slot in 2..7 {
307 let mut guess = relay.join(&path).unwrap();
308 assert_eq!(guess.text(), format!("welcome {slot} 1"));
309 assert_eq!(owner.text(), format!("joined {slot}"));
310 owner.send_to(slot, b"hello");
311 owner.say(&format!("failed {slot}"));
312 assert_eq!(
313 guess.hear(),
314 Message::Binary([&1u32.to_be_bytes()[..], b"hello"].concat())
315 );
316 assert!(guess.closed(Duration::from_secs(5)));
317 assert_eq!(owner.text(), format!("left {slot}"));
318 }
319 assert_eq!(owner.text(), "burned");
320 assert_eq!(status(relay.join(&path)), "HTTP/1.1 410 Gone");
321}
322
323/// A burned code's number stays out of use once its owner leaves, so the old code can't lead
324/// to the owner's next one.
325#[test]
326fn a_burned_number_is_not_given_out_again() {
327 let relay = Relay::start(&["--burn-after", "1"]);
328 // A polled owner, whose leaving is answered once it has left.
329 let (_, session) = relay.ask("GET", "/v1/claim?poll=1", b"");
330 let session = String::from_utf8(session).unwrap();
331 let session = format!("/v1/poll/{}", session.strip_prefix("session ").unwrap());
332 let say = |opcode: u8, text: &str| {
333 let frame = ws::frame(opcode, text.as_bytes(), Some([1, 2, 3, 4]));
334 relay.ask("POST", &session, &frame).0
335 };
336 let (_, heard) = relay.ask("GET", &session, b"");
337 let Ok(Message::Text(nameplate)) = ws::Reader::new(&heard[..], 1 << 20, false).read() else {
338 panic!("heard {heard:?}");
339 };
340 let number: u32 = nameplate
341 .strip_prefix("nameplate ")
342 .unwrap()
343 .parse()
344 .unwrap();
345 let path = format!("/v1/room/code-{number}");
346 let mut guess = relay.join(&path).unwrap();
347 assert_eq!(guess.text(), "welcome 2 1");
348 assert_eq!(say(ws::TEXT, "failed 2"), "HTTP/1.1 200 OK");
349 assert!(guess.closed(Duration::from_secs(5)));
350 assert_eq!(say(ws::CLOSE, ""), "HTTP/1.1 410 Gone");
351 let mut next = relay
352 .join(&format!("/v1/claim?nameplate={number}"))
353 .unwrap();
354 assert_ne!(next.text(), format!("nameplate {number}"));
355 assert_eq!(status(relay.join(&path)), "HTTP/1.1 410 Gone");
356}
357
358/// An owner coming back asks for its code's number again, and gets it while no one else
359/// has claimed it.
360#[test]
361fn an_owner_comes_back_to_its_number() {
362 let relay = Relay::start(&[]);
363 let (owner, number) = claim(&relay);
364 let mut ada = relay.join("/v1/room/0123abcd").unwrap();
365 assert_eq!(ada.text(), "welcome 1");
366 let mut taken = relay
367 .join(&format!("/v1/claim?nameplate={number}"))
368 .unwrap();
369 assert_ne!(taken.text(), format!("nameplate {number}"));
370 drop(owner);
371 thread::sleep(Duration::from_millis(200));
372 let mut again = relay
373 .join(&format!("/v1/claim?nameplate={number}"))
374 .unwrap();
375 assert_eq!(again.text(), format!("nameplate {number}"));
376}
377
378/// Silence, or leaving before the owner says it met, counts as a wrong code; too many from
379/// an address in a minute lock it out, while another address is still let in.
380#[test]
381fn an_address_trying_too_many_codes_is_locked_out() {
382 let relay = Relay::start(&[
383 "--trust-forwarded",
384 "true",
385 "--failures-per-minute",
386 "3",
387 "--pending",
388 "1",
389 "--burn-after",
390 "100",
391 ]);
392 let (mut owner, number) = claim(&relay);
393 let path = format!("/v1/room/code-{number}");
394 let mallory = Some("203.0.113.9");
395 // Silent: the relay gives up on it.
396 let mut silent = relay.join_from(&path, mallory).unwrap();
397 assert!(silent.closed(Duration::from_secs(5)));
398 // Gone before the owner's word.
399 drop(relay.join_from(&path, mallory).unwrap());
400 // Refused by the owner.
401 let mut refused = relay.join_from(&path, mallory).unwrap();
402 assert_eq!(refused.text(), "welcome 4 1");
403 owner.say("failed 4");
404 assert!(refused.closed(Duration::from_secs(5)));
405 let locked = status(relay.join_from(&path, mallory));
406 assert_eq!(locked, "HTTP/1.1 429 Too Many Requests");
407 let Err(again) = relay.join_from(&path, mallory) else {
408 panic!("let in while locked out");
409 };
410 assert!(
411 again.contains("Retry-After: 60"),
412 "a minute's lockout: {again}"
413 );
414 // Its /64 neighbour is the same subscriber; another address isn't.
415 let mut ada = relay.join_from(&path, Some("198.51.100.1")).unwrap();
416 assert_eq!(ada.text(), "welcome 5 1");
417}
418
419/// At most three waiting at once from an address allowed three wrong codes a minute: it
420/// can't run many guesses side by side.
421#[test]
422fn guesses_waiting_at_once_count_against_the_limit() {
423 let relay = Relay::start(&["--failures-per-minute", "3"]);
424 let (_owner, number) = claim(&relay);
425 let path = format!("/v1/room/code-{number}");
426 let _waiting: Vec<Peer> = (0..3).map(|_| relay.join(&path).unwrap()).collect();
427 assert_eq!(status(relay.join(&path)), "HTTP/1.1 429 Too Many Requests");
428}
429
430#[test]
431fn joins_rooms_and_messages_are_capped() {
432 let relay = Relay::start(&[
433 "--max-room-peers",
434 "2",
435 "--joins-per-minute",
436 "5",
437 "--max-message",
438 "1000",
439 ]);
440 let mut ada = relay.join("/v1/room/aaaa").unwrap();
441 let _grace = relay.join("/v1/room/aaaa").unwrap();
442 assert_eq!(
443 status(relay.join("/v1/room/aaaa")),
444 "HTTP/1.1 503 Service Unavailable"
445 );
446 let _bbbb = relay.join("/v1/room/bbbb").unwrap();
447 let _fifth = relay.join("/v1/room/cccc").unwrap();
448 assert_eq!(
449 status(relay.join("/v1/room/dddd")),
450 "HTTP/1.1 429 Too Many Requests"
451 );
452 assert_eq!(ada.text(), "welcome 1");
453 ada.send_to(2, &[0; 2000]);
454 assert!(ada.closed(Duration::from_secs(5)));
455}
456
457#[test]
458fn a_silent_connection_is_closed() {
459 let relay = Relay::start(&["--idle", "1"]);
460 let mut quiet = relay.join("/v1/room/bbbb").unwrap();
461 assert_eq!(quiet.text(), "welcome 1");
462 assert!(quiet.closed(Duration::from_secs(5)));
463}