netget/server/fastcgi/
mod.rs

1//! FastCGI 1.0 application in the Responder role. Rust owns records, request ids, streams,
2//! management records and bounds; the handler owns every response.
3pub mod actions;
4pub mod record;
5
6use crate::llm::action_helper::call_llm;
7use crate::llm::actions::protocol_trait::ActionResult;
8use crate::logging::emit::Log;
9use crate::protocol::{Event, SpawnContext};
10use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
11use crate::server::connection::ConnectionId;
12use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
13use anyhow::{bail, Context, Result};
14use record::Record;
15use serde_json::{json, Map, Value};
16use std::{net::SocketAddr, sync::Arc, time::Duration};
17use tokio::io::{AsyncWrite, AsyncWriteExt};
18
19pub const IDLE_TIMEOUT: Duration = Duration::from_secs(60);
20
21struct Shared {
22    ctx: SpawnContext,
23    idle: Duration,
24}
25
26pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
27    let idle = ctx
28        .startup_params
29        .as_ref()
30        .map(|p| p.get_optional_u64("idle_timeout_secs"))
31        .transpose()?
32        .flatten()
33        .unwrap_or(IDLE_TIMEOUT.as_secs());
34    anyhow::ensure!(
35        (1..=3600).contains(&idle),
36        "idle_timeout_secs must be 1..=3600"
37    );
38    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
39    let local = listener.local_addr()?;
40    Log::new(Some(&ctx.status_tx)).info(format!("FastCGI responder on {local}"));
41    let shared = Arc::new(Shared {
42        ctx: ctx.clone(),
43        idle: Duration::from_secs(idle),
44    });
45    let server_id = ctx.server_id;
46    let accept = tokio::spawn(async move {
47        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
48        loop {
49            // A web server reading a 503 on a FastCGI socket gets nothing it understands, so a
50            // refused connection is simply closed.
51            let (stream, peer, permit) = match accept_bounded(
52                &listener,
53                &limiter,
54                b"",
55                "FastCGI",
56                Some(&shared.ctx.status_tx),
57            )
58            .await
59            {
60                Ok(v) => v,
61                Err(_) => break,
62            };
63            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
64            let now = crate::utils::clock::Instant::now();
65            shared
66                .ctx
67                .state
68                .add_connection_to_server(
69                    server_id,
70                    ConnectionState {
71                        id,
72                        remote_addr: peer,
73                        local_addr: local,
74                        bytes_sent: 0,
75                        bytes_received: 0,
76                        packets_sent: 0,
77                        packets_received: 0,
78                        last_activity: now,
79                        status: ConnectionStatus::Active,
80                        status_changed_at: now,
81                        protocol_info: ProtocolConnectionInfo::empty(),
82                    },
83                )
84                .await;
85            let child = shared.clone();
86            shared
87                .ctx
88                .state
89                .spawn_server_task(server_id, async move {
90                    let _permit = permit;
91                    if let Err(e) = connection(&child, id, stream).await {
92                        Log::new(Some(&child.ctx.status_tx))
93                            .debug(format!("FastCGI connection {id}: {e}"));
94                    }
95                    child
96                        .ctx
97                        .state
98                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
99                        .await;
100                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
101                })
102                .await;
103        }
104    });
105    ctx.state.register_server_task(server_id, accept).await;
106    Ok(local)
107}
108
109fn outcome(ctx: &SpawnContext, id: ConnectionId, request: u16, decision: &str) {
110    let summary = format!("FastCGI connection {id} request={request} decision={decision}");
111    let log = Log::new(Some(&ctx.status_tx));
112    if decision.starts_with("fail_closed_") || decision == "model_silent" {
113        log.error(summary);
114    } else {
115        log.info(summary);
116    }
117}
118
119#[derive(Default)]
120struct Pending {
121    id: u16,
122    keep_conn: bool,
123    params: Vec<u8>,
124    params_done: bool,
125    stdin: Vec<u8>,
126}
127
128async fn write<W: AsyncWrite + Unpin>(
129    shared: &Shared,
130    id: ConnectionId,
131    w: &mut W,
132    bytes: &[u8],
133) -> Result<()> {
134    w.write_all(bytes).await?;
135    w.flush().await?;
136    shared
137        .ctx
138        .state
139        .update_connection_stats(
140            shared.ctx.server_id,
141            id,
142            None,
143            Some(bytes.len() as u64),
144            None,
145            Some(1),
146        )
147        .await;
148    Ok(())
149}
150
151fn end(request: u16, app_status: u32, protocol_status: u8) -> Vec<u8> {
152    record::encode(
153        record::END_REQUEST,
154        request,
155        &record::end_request(app_status, protocol_status),
156    )
157}
158
159/// The records answering one request: STDOUT, optional STDERR, END_REQUEST.
160fn reply(request: u16, stdout: &[u8], stderr: Option<&[u8]>, app_status: u32) -> Vec<u8> {
161    let mut out = record::encode_stream(record::STDOUT, request, stdout);
162    if let Some(e) = stderr.filter(|e| !e.is_empty()) {
163        out.extend(record::encode_stream(record::STDERR, request, e));
164    }
165    out.extend(end(request, app_status, record::REQUEST_COMPLETE));
166    out
167}
168
169fn error_reply(request: u16, status: u16, text: &str) -> Vec<u8> {
170    let mut headers = vec![(
171        "Content-Type".to_owned(),
172        "text/plain; charset=utf-8".to_owned(),
173    )];
174    if status == 503 {
175        headers.push(("Retry-After".to_owned(), "5".to_owned()));
176    }
177    reply(
178        request,
179        &record::build_cgi_response(status, &headers, text.as_bytes()),
180        None,
181        1,
182    )
183}
184
185async fn connection(
186    shared: &Shared,
187    id: ConnectionId,
188    stream: tokio::net::TcpStream,
189) -> Result<()> {
190    let ctx = &shared.ctx;
191    let (mut reader, mut writer) = tokio::io::split(stream);
192    let mut pending: Option<Pending> = None;
193    loop {
194        let rec: Record =
195            match tokio::time::timeout(shared.idle, record::read_record(&mut reader)).await {
196                Err(_) => bail!("idle for {}s", shared.idle.as_secs()),
197                Ok(r) => match r? {
198                    Some(r) => r,
199                    None => return Ok(()),
200                },
201            };
202        ctx.state
203            .update_connection_stats(
204                ctx.server_id,
205                id,
206                Some((record::HEADER_LEN + rec.content.len()) as u64),
207                None,
208                Some(1),
209                None,
210            )
211            .await;
212        if rec.request_id == 0 {
213            match rec.kind {
214                record::GET_VALUES => {
215                    let asked = record::decode_pairs(&rec.content)?;
216                    let max = DEFAULT_MAX_CONNECTIONS.to_string();
217                    let known: Vec<(&str, &str)> = asked
218                        .iter()
219                        .filter_map(|(k, _)| match k.as_str() {
220                            "FCGI_MAX_CONNS" => Some(("FCGI_MAX_CONNS", max.as_str())),
221                            "FCGI_MAX_REQS" => Some(("FCGI_MAX_REQS", max.as_str())),
222                            "FCGI_MPXS_CONNS" => Some(("FCGI_MPXS_CONNS", "0")),
223                            _ => None,
224                        })
225                        .collect();
226                    let body = record::encode_pairs(known);
227                    write(
228                        shared,
229                        id,
230                        &mut writer,
231                        &record::encode(record::GET_VALUES_RESULT, 0, &body),
232                    )
233                    .await?;
234                }
235                kind => {
236                    write(
237                        shared,
238                        id,
239                        &mut writer,
240                        &record::encode(record::UNKNOWN_TYPE, 0, &record::unknown_type(kind)),
241                    )
242                    .await?;
243                }
244            }
245            continue;
246        }
247        match rec.kind {
248            record::BEGIN_REQUEST => {
249                anyhow::ensure!(rec.content.len() >= 8, "BEGIN_REQUEST body is 8 bytes");
250                let role = u16::from_be_bytes([rec.content[0], rec.content[1]]);
251                let keep_conn = rec.content[2] & record::KEEP_CONN != 0;
252                if pending.is_some() {
253                    write(
254                        shared,
255                        id,
256                        &mut writer,
257                        &end(rec.request_id, 0, record::CANT_MPX_CONN),
258                    )
259                    .await?;
260                    continue;
261                }
262                if role != record::ROLE_RESPONDER {
263                    outcome(ctx, id, rec.request_id, "protocol_refusal");
264                    write(
265                        shared,
266                        id,
267                        &mut writer,
268                        &end(rec.request_id, 0, record::UNKNOWN_ROLE),
269                    )
270                    .await?;
271                    if !keep_conn {
272                        return Ok(());
273                    }
274                    continue;
275                }
276                pending = Some(Pending {
277                    id: rec.request_id,
278                    keep_conn,
279                    ..Default::default()
280                });
281            }
282            record::ABORT_REQUEST => {
283                if let Some(p) = pending.take_if(|p| p.id == rec.request_id) {
284                    outcome(ctx, id, p.id, "peer_abort");
285                    write(
286                        shared,
287                        id,
288                        &mut writer,
289                        &end(p.id, 0, record::REQUEST_COMPLETE),
290                    )
291                    .await?;
292                    if !p.keep_conn {
293                        return Ok(());
294                    }
295                }
296            }
297            record::PARAMS => {
298                let Some(p) = pending.as_mut().filter(|p| p.id == rec.request_id) else {
299                    continue;
300                };
301                anyhow::ensure!(!p.params_done, "PARAMS after the end of the PARAMS stream");
302                if rec.content.is_empty() {
303                    p.params_done = true;
304                } else if p.params.len() + rec.content.len() > record::MAX_PARAMS_BYTES {
305                    let p = pending.take().expect("checked");
306                    outcome(ctx, id, p.id, "protocol_refusal");
307                    write(
308                        shared,
309                        id,
310                        &mut writer,
311                        &error_reply(p.id, 431, "request parameters too large"),
312                    )
313                    .await?;
314                    return Ok(());
315                } else {
316                    p.params.extend_from_slice(&rec.content);
317                }
318            }
319            record::STDIN => {
320                let Some(p) = pending.as_mut().filter(|p| p.id == rec.request_id) else {
321                    continue;
322                };
323                anyhow::ensure!(p.params_done, "STDIN before the end of the PARAMS stream");
324                if !rec.content.is_empty() {
325                    if p.stdin.len() + rec.content.len() > record::MAX_BODY_BYTES {
326                        let p = pending.take().expect("checked");
327                        outcome(ctx, id, p.id, "protocol_refusal");
328                        write(
329                            shared,
330                            id,
331                            &mut writer,
332                            &error_reply(p.id, 413, "request body too large"),
333                        )
334                        .await?;
335                        return Ok(());
336                    }
337                    p.stdin.extend_from_slice(&rec.content);
338                    continue;
339                }
340                let p = pending.take().expect("checked");
341                let answer = respond(shared, id, &p).await;
342                write(shared, id, &mut writer, &answer).await?;
343                if !p.keep_conn {
344                    return Ok(());
345                }
346            }
347            record::DATA => {}
348            kind @ (record::STDOUT
349            | record::STDERR
350            | record::END_REQUEST
351            | record::GET_VALUES_RESULT
352            | record::UNKNOWN_TYPE) => bail!("record type {kind} only flows from an application"),
353            kind => {
354                write(
355                    shared,
356                    id,
357                    &mut writer,
358                    &record::encode(record::UNKNOWN_TYPE, 0, &record::unknown_type(kind)),
359                )
360                .await?;
361            }
362        }
363    }
364}
365
366/// Ask the handler about one complete request and build the records that answer it. Every
367/// failure is a 5xx response with a category message, never one the handler did not give.
368async fn respond(shared: &Shared, id: ConnectionId, p: &Pending) -> Vec<u8> {
369    let ctx = &shared.ctx;
370    let pairs = match record::decode_pairs(&p.params) {
371        Ok(pairs) => pairs,
372        Err(_) => {
373            outcome(ctx, id, p.id, "protocol_refusal");
374            return error_reply(p.id, 400, "malformed request parameters");
375        }
376    };
377    let mut params = Map::new();
378    let mut headers = Map::new();
379    for (k, v) in &pairs {
380        if let Some(h) = k.strip_prefix("HTTP_") {
381            headers.insert(h.to_ascii_lowercase().replace('_', "-"), json!(v));
382        }
383        params.insert(k.clone(), json!(v));
384    }
385    for (param, header) in [
386        ("CONTENT_TYPE", "content-type"),
387        ("CONTENT_LENGTH", "content-length"),
388    ] {
389        if let Some(v) = params.get(param).filter(|v| v.as_str() != Some("")) {
390            headers.insert(header.into(), v.clone());
391        }
392    }
393    let get = |k: &str| params.get(k).cloned().unwrap_or(Value::Null);
394    let (body, encoding) = record::body_text(&p.stdin);
395    let event = Event::new(
396        &actions::REQUEST_EVENT,
397        json!({
398            "request_id": p.id,
399            "method": params.get("REQUEST_METHOD").cloned().unwrap_or(json!("GET")),
400            "request_uri": get("REQUEST_URI"),
401            "script_name": get("SCRIPT_NAME"),
402            "path_info": get("PATH_INFO"),
403            "query_string": get("QUERY_STRING"),
404            "content_type": get("CONTENT_TYPE"),
405            "headers": headers,
406            "params": params,
407            "body": body,
408            "body_encoding": encoding,
409            "keep_conn": p.keep_conn,
410        }),
411    );
412    let failed = |decision: &str, e: anyhow::Error| {
413        outcome(ctx, id, p.id, decision);
414        let failure = crate::utils::wire_failure::WireFailure::classify(&e);
415        error_reply(
416            p.id,
417            if failure.is_overloaded() { 503 } else { 500 },
418            failure.text(),
419        )
420    };
421    let result = match call_llm(
422        &ctx.llm_client,
423        &ctx.state,
424        ctx.server_id,
425        Some(id),
426        &event,
427        &actions::FastcgiProtocol,
428    )
429    .await
430    {
431        Ok(r) => r,
432        Err(e) => return failed("fail_closed_llm_error", e),
433    };
434    if !result.failures.is_empty() {
435        return failed(
436            "fail_closed_invalid_reply",
437            anyhow::anyhow!("invalid handler answer"),
438        );
439    }
440    let answers: Vec<Value> = result
441        .protocol_results
442        .into_iter()
443        .flat_map(|r| match r {
444            ActionResult::Multiple(items) => items,
445            other => vec![other],
446        })
447        .filter_map(|r| match r {
448            ActionResult::Custom { name, data } if name == "fastcgi_respond" => Some(data),
449            _ => None,
450        })
451        .collect();
452    let answer = match answers.as_slice() {
453        [one] => one,
454        [] => return failed("model_silent", anyhow::anyhow!("no handler answer")),
455        _ => {
456            return failed(
457                "fail_closed_invalid_reply",
458                anyhow::anyhow!("more than one answer"),
459            )
460        }
461    };
462    let built = (|| -> Result<Vec<u8>> {
463        actions::check_respond(answer)?;
464        let status = answer["status"]
465            .as_u64()
466            .map(|s| s as u16)
467            .context("fastcgi_respond requires an explicit status")?;
468        let headers = record::check_headers(answer.get("headers"))?;
469        let body = record::decode_body(answer)?;
470        Ok(reply(
471            p.id,
472            &record::build_cgi_response(status, &headers, &body),
473            answer["stderr"].as_str().map(str::as_bytes),
474            0,
475        ))
476    })();
477    match built {
478        Ok(records) => {
479            outcome(ctx, id, p.id, "model_answer");
480            records
481        }
482        Err(e) => failed("fail_closed_invalid_reply", e),
483    }
484}
netget/server/fastcgi/mod.rs

netget/server/fastcgi/
mod.rs
