dmx_interface/net/
http.rs

1//! Minimal HTTP/1.1 server with fixed connection slots.
2//!
3//! Reference infrastructure for the web contributor. Serves the embedded
4//! single-file UI (`web/dist`), the config REST surface
5//! (`GET`/`POST /api/config`) and hands `/ws` upgrades to [`super::ws`].
6//!
7//! Design: [`SLOTS`] long-lived tasks, each owning one [`TcpSocket`] for
8//! its entire life (accept → handle → close → accept). No socket is ever
9//! created or dropped after boot — `TcpSocket::new` allocates a slot from
10//! the shared pool and re-creating sockets per connection proved fatal on
11//! hardware (unlogged reboot). smoltcp accepts one pending connection per
12//! listening socket, so [`SLOTS`] tasks accept [`SLOTS`] connections in
13//! parallel; SYNs beyond that are RST and the browser retries.
14
15use core::fmt::Write as _;
16use core::str;
17
18use embassy_executor::Spawner;
19use embassy_net::Stack;
20use embassy_net::tcp::{State, TcpSocket};
21use embassy_time::{Duration, Instant, Timer, with_timeout};
22use embedded_io_async::Write as _;
23use heapless::String;
24use static_cell::StaticCell;
25
26use crate::config::ConfigPatch;
27use crate::storage;
28
29use super::ws;
30
31/// TCP listen port.
32pub const HTTP_PORT: u16 = 80;
33
34/// Parallel connection slots. The UI is a single inlined HTML file, so a
35/// page load plus font, config fetch and WebSocket stays well below this.
36const SLOTS: usize = 4;
37
38/// Per-slot receive buffer (headers plus a config-sized body).
39const RX_SIZE: usize = 1536;
40
41/// Per-slot transmit buffer.
42const TX_SIZE: usize = 1024;
43
44/// Budget for reading and serving one plain request.
45const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
46
47/// Built UI (run `inv web:build` first — `include_bytes!` tracks the file).
48static INDEX_HTML: &[u8] =
49    include_bytes!(concat!(env!("CARGO_MANIFEST_DIR"), "/web/dist/index.html"));
50
51/// UI font (`web/src/style.scss` loads `/fonts/Fredoka.ttf`).
52static FREDOKA_TTF: &[u8] = include_bytes!(concat!(
53    env!("CARGO_MANIFEST_DIR"),
54    "/web/dist/fonts/Fredoka.ttf"
55));
56
57macro_rules! slot_task {
58    ($(#[$meta:meta])* $task_name:ident, $rx_name:ident, $tx_name:ident) => {
59        $(#[$meta])*
60        #[embassy_executor::task]
61        pub async fn $task_name(stack: Stack<'static>) {
62            static $rx_name: StaticCell<[u8; RX_SIZE]> = StaticCell::new();
63            static $tx_name: StaticCell<[u8; TX_SIZE]> = StaticCell::new();
64            serve(
65                stack,
66                $rx_name.init([0; RX_SIZE]),
67                $tx_name.init([0; TX_SIZE]),
68            )
69            .await;
70        }
71    };
72}
73
74slot_task!(
75    /// HTTP connection slot 0: accept → route → recycle, permanently.
76    slot0,
77    SLOT0_RX,
78    SLOT0_TX
79);
80slot_task!(
81    /// HTTP connection slot 1: accept → route → recycle, permanently.
82    slot1,
83    SLOT1_RX,
84    SLOT1_TX
85);
86slot_task!(
87    /// HTTP connection slot 2: accept → route → recycle, permanently.
88    slot2,
89    SLOT2_RX,
90    SLOT2_TX
91);
92slot_task!(
93    /// HTTP connection slot 3: accept → route → recycle, permanently.
94    slot3,
95    SLOT3_RX,
96    SLOT3_TX
97);
98
99/// Spawn all HTTP slots against the given stack.
100///
101/// Works in AP and station mode — the stack is the same either way.
102pub fn spawn_all(spawner: Spawner, stack: Stack<'static>) {
103    macro_rules! spawn_slot {
104        ($result:expr) => {
105            match $result {
106                Ok(token) => spawner.spawn(token),
107                Err(err) => log::error!("http: slot task build failed: {err:?}"),
108            }
109        };
110    }
111    spawn_slot!(slot0(stack));
112    spawn_slot!(slot1(stack));
113    spawn_slot!(slot2(stack));
114    spawn_slot!(slot3(stack));
115}
116
117/// Connection-slot loop: accept → route (or WS session) → recycle, forever.
118async fn serve(stack: Stack<'static>, rx: &'static mut [u8], tx: &'static mut [u8]) {
119    let mut socket = TcpSocket::new(stack, rx, tx);
120    socket.set_timeout(Some(REQUEST_TIMEOUT));
121    socket.set_nagle_enabled(false);
122    loop {
123        if let Err(err) = socket.accept(HTTP_PORT).await {
124            log::warn!("http: accept error: {err:?}");
125            Timer::after(Duration::from_millis(100)).await;
126            continue;
127        }
128        match with_timeout(REQUEST_TIMEOUT, route(&mut socket)).await {
129            Ok(Ok(Outcome::Upgrade(key))) => {
130                // The WebSocket session runs outside the request budget;
131                // `ws` applies its own per-frame idle timeout.
132                socket.set_timeout(None);
133                if ws::serve(&mut socket, &key).await.is_err() {
134                    log::debug!("http: websocket session ended abnormally");
135                }
136            }
137            Ok(Ok(Outcome::Handled)) => {}
138            Ok(Err(())) => log::debug!("http: request failed"),
139            Err(_) => log::debug!("http: request timed out"),
140        }
141        socket.set_timeout(Some(REQUEST_TIMEOUT));
142        recycle(&mut socket).await;
143    }
144}
145
146/// Close gracefully, then wait for `Closed` so `listen()` is legal again.
147async fn recycle(socket: &mut TcpSocket<'_>) {
148    socket.close();
149    let deadline = Instant::now().saturating_add(Duration::from_secs(3));
150    while socket.state() != State::Closed && Instant::now() < deadline {
151        Timer::after(Duration::from_millis(50)).await;
152    }
153    if socket.state() != State::Closed {
154        socket.abort();
155        Timer::after(Duration::from_millis(20)).await;
156    }
157}
158
159/// What happened to a connection after the request was read.
160enum Outcome {
161    /// A plain HTTP request was answered.
162    Handled,
163    /// A WebSocket upgrade was detected; the session key is returned.
164    Upgrade(String<32>),
165}
166
167/// Why a request could not be read — drives the error response.
168enum ReadError {
169    /// Transport failure (EOF/reset/timeout); nothing sensible to answer.
170    Unreadable,
171    /// Headers complete but `Content-Length` is unparseable → answer 400.
172    BadLength,
173    /// Announced body larger than the request buffer → answer 413.
174    TooLarge,
175}
176
177/// Read one request and either answer it or flag the upgrade.
178async fn route(socket: &mut TcpSocket<'_>) -> Result<Outcome, ()> {
179    let mut buf = [0u8; RX_SIZE];
180    let n = match read_request(socket, &mut buf).await {
181        Ok(n) => n,
182        Err(ReadError::Unreadable) => return Err(()),
183        Err(ReadError::BadLength) => {
184            log::warn!("http: malformed Content-Length");
185            send_status(socket, 400).await?;
186            return Ok(Outcome::Handled);
187        }
188        Err(ReadError::TooLarge) => {
189            log::warn!("http: payload exceeds request buffer");
190            send_status(socket, 413).await?;
191            return Ok(Outcome::Handled);
192        }
193    };
194    let head_end = find_headers_end(buf.get(..n).ok_or(())?).ok_or(())?;
195    let headers = buf.get(..head_end).ok_or(())?;
196    let Ok(parsed) = parse_request_line(headers) else {
197        log::warn!("http: malformed request line");
198        send_status(socket, 400).await?;
199        return Ok(Outcome::Handled);
200    };
201    let (method, path) = parsed;
202
203    // WebSocket upgrade takes over this connection entirely.
204    if method == "GET"
205        && path == "/ws"
206        && let Some(key) = ws_key(headers)
207    {
208        return Ok(Outcome::Upgrade(key));
209    }
210
211    let Ok(body) = parse_body(buf.get(..n).ok_or(())?, head_end) else {
212        log::warn!("http: malformed request body");
213        send_status(socket, 400).await?;
214        return Ok(Outcome::Handled);
215    };
216    let status: u16 = match (method, path) {
217        ("GET", "/" | "/index.html") => {
218            send_response(socket, 200, "text/html; charset=utf-8", INDEX_HTML).await?;
219            200
220        }
221        ("GET", "/fonts/Fredoka.ttf") => {
222            send_response(socket, 200, "font/ttf", FREDOKA_TTF).await?;
223            200
224        }
225        ("GET", "/api/config") => {
226            get_config(socket).await?;
227            200
228        }
229        ("POST", "/api/config") => post_config(socket, body).await?,
230        _ => {
231            send_status(socket, 404).await?;
232            404
233        }
234    };
235    if path.starts_with("/api/") {
236        log::info!("http: {method} {path} -> {status}");
237    } else {
238        log::debug!("http: {method} {path} -> {status}");
239    }
240    Ok(Outcome::Handled)
241}
242
243/// Read until `\r\n\r\n` plus the announced body; returns total length.
244async fn read_request(socket: &mut TcpSocket<'_>, buf: &mut [u8]) -> Result<usize, ReadError> {
245    let mut filled = 0usize;
246    loop {
247        let Some(chunk) = buf.get_mut(filled..) else {
248            return Err(ReadError::TooLarge);
249        };
250        let read = socket
251            .read(chunk)
252            .await
253            .map_err(|_| ReadError::Unreadable)?;
254        if read == 0 {
255            return Err(ReadError::Unreadable);
256        }
257        filled = filled.checked_add(read).ok_or(ReadError::Unreadable)?;
258        if let Some(head_end) = find_headers_end(buf.get(..filled).ok_or(ReadError::Unreadable)?) {
259            let announced = content_length(buf.get(..head_end).ok_or(ReadError::Unreadable)?)
260                .map_err(|()| ReadError::BadLength)?;
261            let total = head_end.checked_add(announced).ok_or(ReadError::TooLarge)?;
262            if total > buf.len() {
263                return Err(ReadError::TooLarge);
264            }
265            if filled >= total {
266                return Ok(filled);
267            }
268        }
269    }
270}
271
272/// Offset of the `\r\n\r\n` terminator, if present.
273fn find_headers_end(buf: &[u8]) -> Option<usize> {
274    buf.windows(4).position(|w| w == b"\r\n\r\n")
275}
276
277/// Parsed `Content-Length` (0 when absent).
278fn content_length(headers: &[u8]) -> Result<usize, ()> {
279    /// Lowercase header name searched for (match is case-insensitive).
280    const NEEDLE: &[u8] = b"content-length:";
281    for line in headers.split(|b| *b == b'\n') {
282        let line = trim(line);
283        if starts_with_ci(line, NEEDLE) {
284            let value = trim(line.get(NEEDLE.len()..).ok_or(())?);
285            let text = str::from_utf8(value).map_err(|_| ())?;
286            return text.parse().map_err(|_| ());
287        }
288    }
289    Ok(0)
290}
291
292/// Method and path of the request line. A block without `\r\n` (header-less
293/// minimal request) is treated as the line itself.
294fn parse_request_line(headers: &[u8]) -> Result<(&str, &str), ()> {
295    let line_end = headers
296        .windows(2)
297        .position(|w| w == b"\r\n")
298        .unwrap_or(headers.len());
299    let line = str::from_utf8(headers.get(..line_end).ok_or(())?).map_err(|_| ())?;
300    let mut parts = line.split(' ');
301    let method = parts.next().ok_or(())?;
302    let path = parts.next().ok_or(())?;
303    if method.is_empty() || path.is_empty() {
304        return Err(());
305    }
306    Ok((method, path))
307}
308
309/// Body slice following the header terminator (clamped to announced length).
310fn parse_body(buf: &[u8], head_end: usize) -> Result<&[u8], ()> {
311    let start = head_end.checked_add(4).ok_or(())?;
312    let announced = content_length(buf.get(..head_end).ok_or(())?)?;
313    let end = start.checked_add(announced).ok_or(())?;
314    buf.get(start..end).ok_or(())
315}
316
317/// Case-insensitive prefix test.
318fn starts_with_ci(line: &[u8], prefix: &[u8]) -> bool {
319    line.len() >= prefix.len()
320        && line
321            .get(..prefix.len())
322            .is_some_and(|head| head.eq_ignore_ascii_case(prefix))
323}
324
325/// Trim ASCII whitespace/CR.
326fn trim(mut bytes: &[u8]) -> &[u8] {
327    while bytes.first().is_some_and(u8::is_ascii_whitespace) {
328        bytes = bytes.get(1..).unwrap_or(bytes);
329    }
330    while bytes.last().is_some_and(u8::is_ascii_whitespace) {
331        bytes = bytes.get(..bytes.len().saturating_sub(1)).unwrap_or(bytes);
332    }
333    bytes
334}
335
336/// Extract `Sec-WebSocket-Key` from the header block.
337fn ws_key(headers: &[u8]) -> Option<String<32>> {
338    /// Lowercase header name searched for (match is case-insensitive).
339    const NEEDLE: &[u8] = b"sec-websocket-key:";
340    for line in headers.split(|b| *b == b'\n') {
341        let line = trim(line);
342        if starts_with_ci(line, NEEDLE) {
343            let value = trim(line.get(NEEDLE.len()..)?);
344            let key = str::from_utf8(value).ok()?;
345            return key.trim().parse().ok();
346        }
347    }
348    None
349}
350
351/// `GET /api/config` → full config as JSON.
352async fn get_config(socket: &mut TcpSocket<'_>) -> Result<(), ()> {
353    let config = storage::load();
354    let mut body = [0u8; 768];
355    let n = serde_json_core::to_slice(&config, &mut body).map_err(|_| ())?;
356    send_response(socket, 200, "application/json", body.get(..n).ok_or(())?).await
357}
358
359/// `POST /api/config` → merge patch, persist, apply; returns the status.
360async fn post_config(socket: &mut TcpSocket<'_>, body: &[u8]) -> Result<u16, ()> {
361    let (patch, _used): (ConfigPatch, usize) = match serde_json_core::from_slice(body) {
362        Ok(parsed) => parsed,
363        Err(err) => {
364            log::warn!("http: bad config patch: {err}");
365            send_status(socket, 400).await?;
366            return Ok(400);
367        }
368    };
369    let mut config = storage::load();
370    if !config.merge_patch(patch) {
371        log::warn!("http: config patch rejected");
372        send_status(socket, 400).await?;
373        return Ok(400);
374    }
375    if !storage::save(&config) {
376        send_status(socket, 500).await?;
377        return Ok(500);
378    }
379    config.apply();
380    log::info!("http: config saved via REST");
381    send_response(socket, 200, "application/json", b"{}").await?;
382    Ok(200)
383}
384
385/// Status-only response (e.g. 404).
386async fn send_status(socket: &mut TcpSocket<'_>, status: u16) -> Result<(), ()> {
387    let reason = match status {
388        400 => "Bad Request",
389        404 => "Not Found",
390        413 => "Payload Too Large",
391        500 => "Internal Server Error",
392        _ => "OK",
393    };
394    // 76 bytes worst case ("500 Internal Server Error" head); 404 fits in 64.
395    let mut head = String::<96>::new();
396    write!(
397        &mut head,
398        "HTTP/1.1 {status} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
399    )
400    .map_err(|_| ())?;
401    socket.write_all(head.as_bytes()).await.map_err(|_| ())
402}
403
404/// Status + headers + body; bodies larger than the TX buffer are streamed.
405async fn send_response(
406    socket: &mut TcpSocket<'_>,
407    status: u16,
408    content_type: &str,
409    body: &[u8],
410) -> Result<(), ()> {
411    let mut head = String::<128>::new();
412    write!(
413        &mut head,
414        "HTTP/1.1 {status} OK\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
415        body.len()
416    )
417    .map_err(|_| ())?;
418    socket.write_all(head.as_bytes()).await.map_err(|_| ())?;
419    for chunk in body.chunks(TX_SIZE) {
420        socket.write_all(chunk).await.map_err(|_| ())?;
421    }
422    Ok(())
423}
dmx_interface/net/http.rs

dmx_interface/net/
http.rs
