majordomus_cli/http/
ws.rs

1//! WebSocket, by hand: the handshake and the frames of RFC 6455, and nothing else.
2//!
3//! There is no dependency here for the same reason there is none for base64 or for
4//! percent-encoding: what this server needs of the protocol is a handshake, text frames it
5//! writes, a ping it writes and a close it writes, and that is less code than the seam an
6//! async runtime would introduce into a synchronous server. `Socket.IO` would be a second
7//! protocol on top of this one, with its own framing, its own reconnection semantics and
8//! its own client to ship, in exchange for nothing this control plane needs.
9//!
10//! What is deliberately absent is reading. The server this crate runs is
11//! [`tiny_http`], whose upgrade hands over one `Read + Write` value that cannot be split;
12//! a reader thread blocking on it would have to hold the same lock the writer needs. So
13//! the stream this module writes is one-way by design: everything a client wants to say —
14//! which executions to follow, and from which sequence — it says in the request target,
15//! and the server ignores whatever the client sends afterwards. A client that goes away is
16//! noticed by a non-blocking look at the connection's socket on each quiet interval, which
17//! drains and drops what the client sent (`super::events::LOOK`), and failing that by the
18//! next write failing, which the heartbeat guarantees will happen.
19
20use std::io::Write;
21
22/// The constant RFC 6455 appends to the client's key before hashing it.
23const GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
24
25/// The largest frame this server writes in one piece.
26pub const MAX_FRAME_BYTES: usize = 256 * 1024;
27
28/// Is this request asking to become a WebSocket?
29///
30/// ```
31/// use majordomus_cli::http::{ws, Request};
32/// let plain = Request::parse_target("GET", "/events", vec![]);
33/// assert!(!ws::is_upgrade(&plain));
34/// let asking = plain.clone().with_headers(vec![
35///     ("Upgrade".into(), "websocket".into()),
36///     ("Connection".into(), "Upgrade".into()),
37/// ]);
38/// assert!(ws::is_upgrade(&asking));
39/// ```
40pub fn is_upgrade(req: &super::Request) -> bool {
41    req.header("upgrade")
42        .is_some_and(|v| v.eq_ignore_ascii_case("websocket"))
43        && req
44            .header("connection")
45            .is_some_and(|v| v.to_ascii_lowercase().contains("upgrade"))
46}
47
48/// What a handshake needs from the request, or why it cannot be done.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub enum Handshake {
51    /// The value of `Sec-WebSocket-Accept` to answer with.
52    Accept(String),
53    /// The request cannot become a WebSocket, and this is what to tell the client.
54    Refuse {
55        /// The status to answer with.
56        status: u16,
57        /// Why, for a person reading a network panel.
58        reason: String,
59    },
60}
61
62/// Read the handshake headers and compute the answer.
63///
64/// ```
65/// use majordomus_cli::http::{ws::{handshake, Handshake}, Request};
66/// // the example key of RFC 6455 §1.3 and the accept value it specifies
67/// let req = Request::parse_target("GET", "/events", vec![]).with_headers(vec![
68///     ("Upgrade".into(), "websocket".into()),
69///     ("Connection".into(), "Upgrade".into()),
70///     ("Sec-WebSocket-Key".into(), "dGhlIHNhbXBsZSBub25jZQ==".into()),
71///     ("Sec-WebSocket-Version".into(), "13".into()),
72/// ]);
73/// assert_eq!(
74///     handshake(&req),
75///     Handshake::Accept("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=".into())
76/// );
77/// ```
78pub fn handshake(req: &super::Request) -> Handshake {
79    if req.method != "GET" {
80        return Handshake::Refuse {
81            status: 405,
82            reason: "a WebSocket is opened with GET".into(),
83        };
84    }
85    match req.header("sec-websocket-version") {
86        Some("13") => {}
87        Some(other) => {
88            return Handshake::Refuse {
89                status: 400,
90                reason: format!("Sec-WebSocket-Version {other} is not 13"),
91            }
92        }
93        None => {
94            return Handshake::Refuse {
95                status: 400,
96                reason: "no Sec-WebSocket-Version header".into(),
97            }
98        }
99    }
100    let Some(key) = req.header("sec-websocket-key") else {
101        return Handshake::Refuse {
102            status: 400,
103            reason: "no Sec-WebSocket-Key header".into(),
104        };
105    };
106    Handshake::Accept(accept_for(key))
107}
108
109/// The `Sec-WebSocket-Accept` value for a client key.
110pub fn accept_for(key: &str) -> String {
111    crate::cockpit::base64(&sha1(format!("{key}{GUID}").as_bytes()))
112}
113
114/// The opcodes this server writes.
115#[derive(Debug, Clone, Copy, PartialEq, Eq)]
116pub enum Opcode {
117    /// A UTF-8 payload.
118    Text,
119    /// A heartbeat the client answers automatically.
120    Ping,
121    /// The end of the conversation.
122    Close,
123}
124
125impl Opcode {
126    fn bits(self) -> u8 {
127        match self {
128            Opcode::Text => 0x1,
129            Opcode::Ping => 0x9,
130            Opcode::Close => 0x8,
131        }
132    }
133}
134
135/// One unfragmented frame, from the server, unmasked as RFC 6455 requires of a server.
136///
137/// ```
138/// use majordomus_cli::http::ws::{frame, Opcode};
139/// assert_eq!(frame(Opcode::Text, b"hi"), vec![0x81, 0x02, b'h', b'i']);
140/// assert_eq!(&frame(Opcode::Ping, b"")[..2], &[0x89, 0x00]);
141/// // a payload of 126 bytes or more carries an extended length
142/// assert_eq!(&frame(Opcode::Text, &[b'x'; 200])[..4], &[0x81, 126, 0, 200]);
143/// ```
144pub fn frame(opcode: Opcode, payload: &[u8]) -> Vec<u8> {
145    let mut out = Vec::with_capacity(payload.len() + 10);
146    out.push(0x80 | opcode.bits());
147    match payload.len() {
148        n if n < 126 => out.push(n as u8),
149        n if n <= u16::MAX as usize => {
150            out.push(126);
151            out.extend_from_slice(&(n as u16).to_be_bytes());
152        }
153        n => {
154            out.push(127);
155            out.extend_from_slice(&(n as u64).to_be_bytes());
156        }
157    }
158    out.extend_from_slice(payload);
159    out
160}
161
162/// Write one text frame, refusing a payload over [`MAX_FRAME_BYTES`].
163///
164/// The refusal is not a truncation: a client that received half a JSON document would
165/// have to guess. The store does not bound every event — a handler's output and a caller's
166/// input are unbounded values — so the live channel measures an event before it writes it
167/// and replaces such a value with a reference to where it is read
168/// ([`super::events::Withheld`]); this refusal is the last line, not the bound.
169pub fn write_text(stream: &mut dyn Write, text: &str) -> std::io::Result<()> {
170    if text.len() > MAX_FRAME_BYTES {
171        return Err(std::io::Error::other(format!(
172            "a frame of {} bytes is over the {MAX_FRAME_BYTES}-byte bound",
173            text.len()
174        )));
175    }
176    stream.write_all(&frame(Opcode::Text, text.as_bytes()))?;
177    stream.flush()
178}
179
180/// Write a ping with no payload.
181pub fn write_ping(stream: &mut dyn Write) -> std::io::Result<()> {
182    stream.write_all(&frame(Opcode::Ping, b""))?;
183    stream.flush()
184}
185
186/// Write a close frame carrying a status code and a reason, then flush.
187///
188/// 1000 is a normal close; 1001 is "going away", which is what a server that is shutting
189/// down means.
190pub fn write_close(stream: &mut dyn Write, code: u16, reason: &str) -> std::io::Result<()> {
191    let mut payload = code.to_be_bytes().to_vec();
192    payload.extend_from_slice(&reason.as_bytes()[..reason.len().min(120)]);
193    stream.write_all(&frame(Opcode::Close, &payload))?;
194    stream.flush()
195}
196
197/// SHA-1 of a byte string, for the one thing RFC 6455 specifies it for.
198///
199/// The handshake is the only place this algorithm appears in this crate, and it is not
200/// used for anything that needs to be hard to forge — the accept value proves the peer
201/// read the request, nothing more. Everything this executable hashes for real (the
202/// registry fingerprint, the index, the artifacts) is SHA-256.
203///
204/// ```
205/// use majordomus_cli::http::ws::sha1;
206/// let digest: String = sha1(b"abc").iter().map(|b| format!("{b:02x}")).collect();
207/// assert_eq!(digest, "a9993e364706816aba3e25717850c26c9cd0d89d");
208/// let empty: String = sha1(b"").iter().map(|b| format!("{b:02x}")).collect();
209/// assert_eq!(empty, "da39a3ee5e6b4b0d3255bfef95601890afd80709");
210/// ```
211pub fn sha1(message: &[u8]) -> [u8; 20] {
212    let mut h: [u32; 5] = [
213        0x6745_2301,
214        0xEFCD_AB89,
215        0x98BA_DCFE,
216        0x1032_5476,
217        0xC3D2_E1F0,
218    ];
219    let mut padded = message.to_vec();
220    padded.push(0x80);
221    while padded.len() % 64 != 56 {
222        padded.push(0);
223    }
224    padded.extend_from_slice(&((message.len() as u64) * 8).to_be_bytes());
225
226    for block in padded.chunks_exact(64) {
227        let mut w = [0u32; 80];
228        for (i, word) in block.chunks_exact(4).enumerate() {
229            w[i] = u32::from_be_bytes([word[0], word[1], word[2], word[3]]);
230        }
231        for i in 16..80 {
232            w[i] = (w[i - 3] ^ w[i - 8] ^ w[i - 14] ^ w[i - 16]).rotate_left(1);
233        }
234        let [mut a, mut b, mut c, mut d, mut e] = h;
235        for (i, word) in w.iter().enumerate() {
236            let (f, k) = match i {
237                0..=19 => ((b & c) | ((!b) & d), 0x5A82_7999),
238                20..=39 => (b ^ c ^ d, 0x6ED9_EBA1),
239                40..=59 => ((b & c) | (b & d) | (c & d), 0x8F1B_BCDC),
240                _ => (b ^ c ^ d, 0xCA62_C1D6),
241            };
242            let temp = a
243                .rotate_left(5)
244                .wrapping_add(f)
245                .wrapping_add(e)
246                .wrapping_add(k)
247                .wrapping_add(*word);
248            e = d;
249            d = c;
250            c = b.rotate_left(30);
251            b = a;
252            a = temp;
253        }
254        h[0] = h[0].wrapping_add(a);
255        h[1] = h[1].wrapping_add(b);
256        h[2] = h[2].wrapping_add(c);
257        h[3] = h[3].wrapping_add(d);
258        h[4] = h[4].wrapping_add(e);
259    }
260    let mut out = [0u8; 20];
261    for (i, word) in h.iter().enumerate() {
262        out[i * 4..i * 4 + 4].copy_from_slice(&word.to_be_bytes());
263    }
264    out
265}
266
267#[cfg(test)]
268mod tests {
269    use super::*;
270    use crate::http::Request;
271
272    fn hex(bytes: &[u8]) -> String {
273        bytes.iter().map(|b| format!("{b:02x}")).collect()
274    }
275
276    #[test]
277    fn sha1_matches_the_published_vectors() {
278        assert_eq!(hex(&sha1(b"")), "da39a3ee5e6b4b0d3255bfef95601890afd80709");
279        assert_eq!(
280            hex(&sha1(b"abc")),
281            "a9993e364706816aba3e25717850c26c9cd0d89d"
282        );
283        assert_eq!(
284            hex(&sha1(
285                b"abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq"
286            )),
287            "84983e441c3bd26ebaae4aa1f95129e5e54670f1",
288            "a message that crosses a block boundary"
289        );
290        assert_eq!(
291            hex(&sha1(&[b'a'; 1000])),
292            "291e9a6c66994949b57ba5e650361e98fc36b1ba"
293        );
294        // exactly one block of padding overflow: 56 bytes needs a second block
295        assert_eq!(sha1(&[b'x'; 56]).len(), 20);
296        assert_eq!(sha1(&[b'x'; 64]).len(), 20);
297    }
298
299    #[test]
300    fn the_accept_value_is_the_one_rfc_6455_specifies() {
301        assert_eq!(
302            accept_for("dGhlIHNhbXBsZSBub25jZQ=="),
303            "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
304        );
305    }
306
307    #[test]
308    fn a_handshake_is_refused_with_a_reason_a_person_can_read() {
309        let base = Request::parse_target("GET", "/events", vec![]);
310        let refuse =
311            |headers: Vec<(String, String)>| match handshake(&base.clone().with_headers(headers)) {
312                Handshake::Refuse { status, reason } => (status, reason),
313                Handshake::Accept(_) => panic!("accepted"),
314            };
315        assert_eq!(refuse(vec![]).0, 400);
316        assert!(refuse(vec![]).1.contains("Version"));
317        assert!(refuse(vec![("Sec-WebSocket-Version".into(), "8".into())])
318            .1
319            .contains("not 13"));
320        assert!(refuse(vec![("Sec-WebSocket-Version".into(), "13".into())])
321            .1
322            .contains("Key"));
323        let post = Request::parse_target("POST", "/events", vec![]);
324        assert_eq!(
325            match handshake(&post) {
326                Handshake::Refuse { status, .. } => status,
327                Handshake::Accept(_) => 0,
328            },
329            405
330        );
331    }
332
333    #[test]
334    fn only_a_request_that_asks_is_an_upgrade() {
335        let base = Request::parse_target("GET", "/events", vec![]);
336        assert!(!is_upgrade(&base));
337        assert!(!is_upgrade(
338            &base
339                .clone()
340                .with_headers(vec![("Upgrade".into(), "websocket".into())])
341        ));
342        assert!(is_upgrade(&base.clone().with_headers(vec![
343            ("upgrade".into(), "WebSocket".into()),
344            ("connection".into(), "keep-alive, Upgrade".into()),
345        ])));
346    }
347
348    #[test]
349    fn frames_carry_the_lengths_the_protocol_specifies() {
350        assert_eq!(frame(Opcode::Text, b""), vec![0x81, 0x00]);
351        assert_eq!(frame(Opcode::Close, b""), vec![0x88, 0x00]);
352        let medium = frame(Opcode::Text, &[b'x'; 65_535]);
353        assert_eq!(&medium[..4], &[0x81, 126, 0xff, 0xff]);
354        assert_eq!(medium.len(), 65_535 + 4);
355        let large = frame(Opcode::Text, &[b'x'; 70_000]);
356        assert_eq!(large[1], 127);
357        assert_eq!(
358            u64::from_be_bytes(large[2..10].try_into().unwrap()),
359            70_000,
360            "the 64-bit length"
361        );
362        // no mask bit is ever set: a server frame is never masked
363        for f in [
364            frame(Opcode::Text, b"a"),
365            frame(Opcode::Ping, b""),
366            frame(Opcode::Close, b"x"),
367        ] {
368            assert_eq!(f[1] & 0x80, 0, "a server frame carries no mask");
369            assert_eq!(f[0] & 0x80, 0x80, "and is never fragmented");
370        }
371    }
372
373    #[test]
374    fn writing_refuses_an_oversized_frame_rather_than_cutting_it() {
375        let mut sink: Vec<u8> = Vec::new();
376        let big = "x".repeat(MAX_FRAME_BYTES + 1);
377        assert!(write_text(&mut sink, &big).is_err());
378        assert!(sink.is_empty(), "nothing half-written reaches the wire");
379        assert!(write_text(&mut sink, "{\"a\":1}").is_ok());
380        assert_eq!(&sink[2..], b"{\"a\":1}");
381        sink.clear();
382        assert!(write_ping(&mut sink).is_ok());
383        assert_eq!(sink, vec![0x89, 0x00]);
384        sink.clear();
385        assert!(write_close(&mut sink, 1001, "going away").is_ok());
386        assert_eq!(sink[0], 0x88);
387        assert_eq!(u16::from_be_bytes([sink[2], sink[3]]), 1001);
388        assert_eq!(&sink[4..], b"going away");
389    }
390}
majordomus_cli/http/ws.rs

majordomus_cli/http/
ws.rs
