netget/client/fastcgi/
mod.rs

1//! FastCGI 1.0 client in the web-server role.
2pub mod actions;
3use crate::client::llm_budget::call_llm_for_client;
4use crate::llm::actions::client_trait::{Client, ClientActionResult};
5use crate::logging::emit::Log;
6use crate::protocol::{ConnectContext, Event};
7use crate::server::fastcgi::record;
8use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
9use crate::state::{AccessLogOwner, ClientStatus};
10pub use actions::FastcgiClientProtocol;
11use anyhow::{Context, Result};
12use serde_json::{json, Map, Value};
13use std::net::SocketAddr;
14use std::time::Duration;
15use tokio::io::AsyncWriteExt;
16use tokio::net::TcpStream;
17use tokio::sync::mpsc;
18
19pub const DEFAULT_DOCUMENT_ROOT: &str = "/var/www/html";
20pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
21const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
22/// How long an aborted request may take to send its END_REQUEST.
23const ABORT_GRACE: Duration = Duration::from_secs(5);
24
25struct Settings {
26    remote: String,
27    document_root: String,
28    timeout: Duration,
29}
30
31async fn dial(remote: &str) -> Result<TcpStream> {
32    tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(remote))
33        .await
34        .context("FastCGI connect timed out")?
35        .with_context(|| format!("FastCGI connect to {remote}"))
36}
37
38pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
39    let p = ctx.startup_params.as_ref();
40    let document_root = p
41        .map(|p| p.get_optional_string("document_root"))
42        .transpose()?
43        .flatten()
44        .unwrap_or_else(|| DEFAULT_DOCUMENT_ROOT.to_owned());
45    anyhow::ensure!(
46        document_root.starts_with('/') && document_root.len() <= 1024,
47        "document_root must be an absolute path"
48    );
49    let timeout = p
50        .map(|p| p.get_optional_u64("request_timeout_secs"))
51        .transpose()?
52        .flatten()
53        .unwrap_or(DEFAULT_REQUEST_TIMEOUT.as_secs());
54    anyhow::ensure!(
55        (1..=600).contains(&timeout),
56        "request_timeout_secs must be 1..=600"
57    );
58    let stream = dial(&ctx.remote_addr).await?;
59    let local = stream.local_addr()?;
60    let settings = Settings {
61        remote: ctx.remote_addr.clone(),
62        document_root: document_root.trim_end_matches('/').to_owned(),
63        timeout: Duration::from_secs(timeout),
64    };
65    ctx.state
66        .update_client_status(ctx.client_id, ClientStatus::Connected)
67        .await;
68    let external =
69        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
70    let (internal_tx, internal_rx) = mpsc::channel::<Value>(16);
71    let (event_tx, mut event_rx) = mpsc::channel::<Event>(16);
72    event_tx.try_send(Event::new(
73        &actions::CONNECTED_EVENT,
74        json!({"remote_addr": ctx.remote_addr}),
75    ))?;
76    let events_ctx = ctx.clone();
77    let dispatcher = tokio::spawn(async move {
78        let protocol = FastcgiClientProtocol;
79        while let Some(event) = event_rx.recv().await {
80            let instruction = events_ctx
81                .state
82                .get_instruction_for_client(events_ctx.client_id)
83                .await
84                .unwrap_or_default();
85            let memory = events_ctx
86                .state
87                .get_memory_for_client(events_ctx.client_id)
88                .await
89                .unwrap_or_default();
90            match call_llm_for_client(
91                &events_ctx.llm_client,
92                &events_ctx.state,
93                events_ctx.client_id.to_string(),
94                &instruction,
95                &memory,
96                Some(&event),
97                &protocol,
98                &events_ctx.status_tx,
99            )
100            .await
101            {
102                Ok(result) => {
103                    if let Some(memory) = result.memory_updates {
104                        events_ctx
105                            .state
106                            .set_memory_for_client(events_ctx.client_id, memory)
107                            .await;
108                    }
109                    for action in result.actions {
110                        if internal_tx.send(action).await.is_err() {
111                            return;
112                        }
113                    }
114                }
115                Err(e) => Log::new(Some(&events_ctx.status_tx))
116                    .warn(format!("FastCGI client handler: {e}")),
117            }
118        }
119    });
120    ctx.state
121        .register_client_task(ctx.client_id, dispatcher)
122        .await;
123    let session_ctx = ctx.clone();
124    let task = tokio::spawn(async move {
125        let result = session(
126            &session_ctx,
127            &settings,
128            Some(stream),
129            external,
130            internal_rx,
131            event_tx,
132        )
133        .await;
134        let status = match result {
135            Ok(()) => ClientStatus::Disconnected,
136            Err(e) => {
137                Log::new(Some(&session_ctx.status_tx)).warn(format!("FastCGI client ended: {e}"));
138                ClientStatus::Error(e.to_string())
139            }
140        };
141        session_ctx
142            .state
143            .update_client_status(session_ctx.client_id, status)
144            .await;
145        session_ctx
146            .state
147            .remove_client_handle(session_ctx.client_id)
148            .await;
149        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
150    });
151    ctx.state.register_client_task(ctx.client_id, task).await;
152    Ok(local)
153}
154
155fn reply(command: Option<ClientCommand>, outcome: Result<ClientSendOutcome>) {
156    if let Some(c) = command {
157        crate::client::command_support::reply(c, outcome);
158    }
159}
160
161async fn session(
162    ctx: &ConnectContext,
163    settings: &Settings,
164    mut stream: Option<TcpStream>,
165    mut external: mpsc::Receiver<ClientCommand>,
166    mut internal: mpsc::Receiver<Value>,
167    events: mpsc::Sender<Event>,
168) -> Result<()> {
169    let mut next_id: u16 = 1;
170    loop {
171        let (action, command) = tokio::select! {
172            c = external.recv() => match c { Some(c) => (c.action.clone(), Some(c)), None => return Ok(()) },
173            a = internal.recv() => match a { Some(a) => (a, None), None => return Ok(()) },
174        };
175        match FastcgiClientProtocol.execute_action(action.clone()) {
176            Ok(ClientActionResult::Disconnect) => {
177                reply(command, Ok(ClientSendOutcome::Disconnected));
178                return Ok(());
179            }
180            Ok(_) => {}
181            Err(e) => {
182                reply(
183                    command,
184                    Ok(ClientSendOutcome::Rejected {
185                        error: e.to_string(),
186                    }),
187                );
188                continue;
189            }
190        }
191        let is_request = action["type"] == "fastcgi_request";
192        let request_id = next_id;
193        if is_request {
194            next_id = next_id.checked_add(1).unwrap_or(1);
195        }
196        // An application may close an idle kept connection; one reconnect per action.
197        let mut outcome = None;
198        for attempt in 0..2 {
199            if stream.is_none() {
200                match dial(&settings.remote).await {
201                    Ok(s) => stream = Some(s),
202                    Err(e) => {
203                        outcome = Some(Err(e));
204                        break;
205                    }
206                }
207            }
208            let s = stream.as_mut().expect("dialled above");
209            let r = if is_request {
210                request(s, settings, request_id, &action).await
211            } else {
212                get_values(s, &action).await
213            };
214            match r {
215                Err(Failure::Transport(e)) if attempt == 0 => {
216                    Log::new(Some(&ctx.status_tx)).debug(format!("FastCGI reconnecting: {e}"));
217                    stream = None;
218                }
219                Err(Failure::Transport(e)) => {
220                    stream = None;
221                    outcome = Some(Err(e));
222                    break;
223                }
224                Ok((data, keep)) => {
225                    if !keep {
226                        stream = None;
227                    }
228                    outcome = Some(Ok(data));
229                    break;
230                }
231            }
232        }
233        let outcome = outcome.unwrap_or_else(|| Err(anyhow::anyhow!("no attempt made")));
234        if command.is_some() {
235            ctx.state
236                .record_access_log(
237                    AccessLogOwner::Client(ctx.client_id.as_u32()),
238                    "FastCGI",
239                    None,
240                    "injected_action",
241                    json!({"type": action["type"], "path": action["path"]}),
242                    vec![json!({"ok": outcome.is_ok()})],
243                )
244                .await;
245        }
246        match outcome {
247            Ok(data) => {
248                reply(command, Ok(ClientSendOutcome::Sent { bytes_sent: 0 }));
249                let kind = if is_request {
250                    &*actions::RESPONSE_EVENT
251                } else {
252                    &*actions::VALUES_EVENT
253                };
254                events
255                    .send(Event::new(kind, data))
256                    .await
257                    .context("FastCGI event consumer stopped")?;
258            }
259            Err(e) => {
260                Log::new(Some(&ctx.status_tx)).warn(format!("FastCGI request failed: {e}"));
261                reply(command, Err(e));
262            }
263        }
264    }
265}
266
267enum Failure {
268    /// The connection failed before any answer: worth one reconnect.
269    Transport(anyhow::Error),
270}
271
272fn params_for(
273    settings: &Settings,
274    action: &Value,
275    body_len: usize,
276) -> Result<Vec<(String, String)>> {
277    let method = action["method"].as_str().unwrap_or("GET");
278    let path = action["path"].as_str().unwrap_or("/");
279    let query = action["query"].as_str().unwrap_or("");
280    let (host, port) = settings
281        .remote
282        .rsplit_once(':')
283        .unwrap_or((settings.remote.as_str(), ""));
284    let mut params: Vec<(String, String)> = vec![
285        ("GATEWAY_INTERFACE".into(), "CGI/1.1".into()),
286        ("SERVER_SOFTWARE".into(), "netget".into()),
287        ("SERVER_PROTOCOL".into(), "HTTP/1.1".into()),
288        ("REQUEST_METHOD".into(), method.into()),
289        (
290            "REQUEST_URI".into(),
291            if query.is_empty() {
292                path.into()
293            } else {
294                format!("{path}?{query}")
295            },
296        ),
297        ("SCRIPT_NAME".into(), path.into()),
298        (
299            "SCRIPT_FILENAME".into(),
300            format!("{}{path}", settings.document_root),
301        ),
302        ("DOCUMENT_ROOT".into(), settings.document_root.clone()),
303        ("PATH_INFO".into(), String::new()),
304        ("QUERY_STRING".into(), query.into()),
305        ("SERVER_NAME".into(), host.trim_matches(['[', ']']).into()),
306        ("SERVER_PORT".into(), port.into()),
307        ("REMOTE_ADDR".into(), "127.0.0.1".into()),
308        (
309            "CONTENT_LENGTH".into(),
310            if body_len > 0 {
311                body_len.to_string()
312            } else {
313                String::new()
314            },
315        ),
316    ];
317    for (k, v) in record::check_headers(action.get("headers"))? {
318        if k.eq_ignore_ascii_case("content-type") {
319            params.push(("CONTENT_TYPE".into(), v));
320        } else if !k.eq_ignore_ascii_case("content-length") {
321            params.push((
322                format!("HTTP_{}", k.to_ascii_uppercase().replace('-', "_")),
323                v,
324            ));
325        }
326    }
327    for (k, v) in record::check_headers(action.get("params"))? {
328        params.retain(|(name, _)| *name != k);
329        params.push((k, v));
330    }
331    Ok(params)
332}
333
334async fn send(s: &mut TcpStream, bytes: &[u8]) -> Result<(), Failure> {
335    s.write_all(bytes)
336        .await
337        .map_err(|e| Failure::Transport(e.into()))?;
338    s.flush().await.map_err(|e| Failure::Transport(e.into()))
339}
340
341/// One Responder request. `Ok` carries the event data and whether the connection stays usable.
342async fn request(
343    s: &mut TcpStream,
344    settings: &Settings,
345    id: u16,
346    action: &Value,
347) -> Result<(Value, bool), Failure> {
348    let failed = |e: anyhow::Error| Ok((json!({"request_id": id, "error": e.to_string()}), false));
349    let body = match record::decode_body(action) {
350        Ok(b) => b,
351        Err(e) => return failed(e),
352    };
353    let params = match params_for(settings, action, body.len()) {
354        Ok(p) => p,
355        Err(e) => return failed(e),
356    };
357    let pairs = record::encode_pairs(params.iter().map(|(k, v)| (k.as_str(), v.as_str())));
358    if pairs.len() > record::MAX_PARAMS_BYTES {
359        return failed(anyhow::anyhow!("params exceed 64 KiB"));
360    }
361    let mut out = record::encode(
362        record::BEGIN_REQUEST,
363        id,
364        &record::begin_request(record::ROLE_RESPONDER, record::KEEP_CONN),
365    );
366    out.extend(record::encode_stream(record::PARAMS, id, &pairs));
367    out.extend(record::encode_stream(record::STDIN, id, &body));
368    send(s, &out).await?;
369    let mut stdout = Vec::new();
370    let mut stderr = Vec::new();
371    let mut aborted = false;
372    let mut deadline = tokio::time::Instant::now() + settings.timeout;
373    let mut first = true;
374    let (app_status, protocol_status) = loop {
375        let rec = match tokio::time::timeout_at(deadline, record::read_record(s)).await {
376            Err(_) if !aborted => {
377                aborted = true;
378                send(s, &record::encode(record::ABORT_REQUEST, id, &[])).await?;
379                deadline = tokio::time::Instant::now() + ABORT_GRACE;
380                continue;
381            }
382            Err(_) => {
383                return Ok((
384                    json!({"request_id": id, "aborted": true, "error": "no END_REQUEST after ABORT_REQUEST"}),
385                    false,
386                ));
387            }
388            Ok(Ok(Some(r))) => r,
389            Ok(Ok(None)) if first => {
390                return Err(Failure::Transport(anyhow::anyhow!(
391                    "application closed the connection"
392                )))
393            }
394            Ok(Ok(None)) => {
395                return failed(anyhow::anyhow!(
396                    "application closed the connection mid-response"
397                ))
398            }
399            Ok(Err(e)) if first => return Err(Failure::Transport(e)),
400            Ok(Err(e)) => return failed(e),
401        };
402        first = false;
403        if rec.request_id != id {
404            continue;
405        }
406        match rec.kind {
407            record::STDOUT => {
408                if stdout.len() + rec.content.len() > record::MAX_BODY_BYTES + 64 * 1024 {
409                    return failed(anyhow::anyhow!("response over 1 MiB"));
410                }
411                stdout.extend_from_slice(&rec.content);
412            }
413            record::STDERR => {
414                let room = record::MAX_STDERR_BYTES.saturating_sub(stderr.len());
415                stderr.extend_from_slice(&rec.content[..rec.content.len().min(room)]);
416            }
417            record::END_REQUEST => {
418                if rec.content.len() < 8 {
419                    return failed(anyhow::anyhow!("END_REQUEST body is 8 bytes"));
420                }
421                let c = &rec.content;
422                break (u32::from_be_bytes([c[0], c[1], c[2], c[3]]), c[4]);
423            }
424            _ => {}
425        }
426    };
427    let mut data = json!({
428        "request_id": id,
429        "app_status": app_status,
430        "protocol_status": record::protocol_status_name(protocol_status),
431        "stderr": String::from_utf8_lossy(&stderr),
432    });
433    if aborted {
434        data["aborted"] = json!(true);
435    }
436    if protocol_status == record::REQUEST_COMPLETE && !stdout.is_empty() {
437        match record::parse_cgi_response(&stdout) {
438            Ok((status, headers, body)) => {
439                let (text, encoding) = record::body_text(&body);
440                data["status"] = json!(status);
441                data["headers"] = Value::Object(headers);
442                data["body"] = json!(text);
443                data["body_encoding"] = json!(encoding);
444            }
445            Err(e) => data["error"] = json!(format!("not a CGI response: {e}")),
446        }
447    }
448    Ok((data, true))
449}
450
451async fn get_values(s: &mut TcpStream, action: &Value) -> Result<(Value, bool), Failure> {
452    let names: Vec<String> = match action["names"].as_array() {
453        Some(list) => list
454            .iter()
455            .filter_map(|n| n.as_str().map(str::to_owned))
456            .collect(),
457        None => ["FCGI_MAX_CONNS", "FCGI_MAX_REQS", "FCGI_MPXS_CONNS"]
458            .map(String::from)
459            .to_vec(),
460    };
461    let body = record::encode_pairs(names.iter().map(|n| (n.as_str(), "")));
462    send(s, &record::encode(record::GET_VALUES, 0, &body)).await?;
463    let deadline = tokio::time::Instant::now() + ABORT_GRACE;
464    loop {
465        let rec = match tokio::time::timeout_at(deadline, record::read_record(s)).await {
466            Err(_) => {
467                return Ok((
468                    json!({"values": {}, "error": "no FCGI_GET_VALUES_RESULT"}),
469                    true,
470                ))
471            }
472            Ok(Ok(Some(r))) => r,
473            Ok(Ok(None)) => {
474                return Err(Failure::Transport(anyhow::anyhow!(
475                    "application closed the connection"
476                )))
477            }
478            Ok(Err(e)) => return Err(Failure::Transport(e)),
479        };
480        if rec.request_id == 0 && rec.kind == record::GET_VALUES_RESULT {
481            let values: Map<String, Value> = match record::decode_pairs(&rec.content) {
482                Ok(pairs) => pairs.into_iter().map(|(k, v)| (k, json!(v))).collect(),
483                Err(e) => return Ok((json!({"values": {}, "error": e.to_string()}), false)),
484            };
485            return Ok((json!({"values": values}), true));
486        }
487        if rec.request_id == 0 && rec.kind == record::UNKNOWN_TYPE {
488            return Ok((
489                json!({"values": {}, "error": "the application does not know FCGI_GET_VALUES"}),
490                true,
491            ));
492        }
493    }
494}
netget/client/fastcgi/mod.rs

netget/client/fastcgi/
mod.rs
