netget/client/ninep/
mod.rs

1//! 9P2000 client. One connection, attached once; each action walks from the root fid to a
2//! fresh fid, does its work and clunks it.
3pub mod actions;
4use crate::client::llm_budget::call_llm_for_client;
5use crate::llm::actions::client_trait::{Client, ClientActionResult};
6use crate::logging::emit::Log;
7use crate::protocol::{ConnectContext, Event};
8use crate::server::ninep::wire::{self, Reader, Stat, Writer};
9use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
10use crate::state::{AccessLogOwner, ClientStatus};
11pub use actions::NinepClientProtocol;
12use anyhow::{Context, Result};
13use serde_json::{json, Value};
14use std::net::SocketAddr;
15use tokio::{
16    io::{AsyncWriteExt, ReadHalf, WriteHalf},
17    net::TcpStream,
18    sync::mpsc,
19};
20
21pub const DEFAULT_UNAME: &str = "netget";
22const ROOT_FID: u32 = 0;
23
24/// An error the server answered with (Rerror): the operation failed, the session goes on.
25#[derive(Debug)]
26pub struct Remote(pub String);
27impl std::fmt::Display for Remote {
28    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        f.write_str(&self.0)
30    }
31}
32impl std::error::Error for Remote {}
33
34fn remote(text: impl Into<String>) -> anyhow::Error {
35    anyhow::Error::new(Remote(text.into()))
36}
37
38pub struct Conn {
39    reader: ReadHalf<TcpStream>,
40    writer: WriteHalf<TcpStream>,
41    msize: u32,
42    next_tag: u16,
43    next_fid: u32,
44}
45
46impl Conn {
47    async fn rpc(&mut self, kind: u8, body: Writer) -> Result<(u8, Vec<u8>)> {
48        let tag = if kind == wire::TVERSION {
49            wire::NOTAG
50        } else {
51            self.next_tag = self.next_tag.wrapping_add(1) % wire::NOTAG;
52            self.next_tag
53        };
54        let message = body.finish(kind, tag);
55        anyhow::ensure!(
56            message.len() as u32 <= self.msize,
57            "9P request exceeds msize"
58        );
59        tokio::time::timeout(wire::IO_TIMEOUT, self.writer.write_all(&message))
60            .await
61            .context("9P write deadline")??;
62        let (rkind, rtag, rbody) =
63            wire::read_message(&mut self.reader, self.msize, wire::IO_TIMEOUT)
64                .await?
65                .context("9P server closed the connection")?;
66        anyhow::ensure!(rtag == tag, "9P reply tag {rtag} does not answer {tag}");
67        if rkind == wire::RERROR {
68            return Err(remote(Reader::new(&rbody).string()?));
69        }
70        anyhow::ensure!(
71            rkind == kind + 1,
72            "9P reply type {rkind} does not answer {kind}"
73        );
74        Ok((rkind, rbody))
75    }
76
77    fn fid(&mut self) -> u32 {
78        self.next_fid = if self.next_fid >= 0x7fff_ffff {
79            1
80        } else {
81            self.next_fid + 1
82        };
83        self.next_fid
84    }
85
86    async fn clunk(&mut self, fid: u32) {
87        let _ = self.rpc(wire::TCLUNK, Writer::new().u32(fid)).await;
88    }
89
90    /// Walk from the root to `path`, in steps of at most 16 elements, onto a new fid.
91    async fn walk(&mut self, path: &str) -> Result<u32> {
92        let elements = wire::elements(path)?;
93        let fid = self.fid();
94        let mut from = ROOT_FID;
95        let mut chunks: Vec<&[String]> = elements.chunks(wire::MAX_WELEM).collect();
96        if chunks.is_empty() {
97            chunks.push(&[]);
98        }
99        for chunk in chunks {
100            let mut w = Writer::new().u32(from).u32(fid).u16(chunk.len() as u16);
101            for name in chunk {
102                w = w.string(name);
103            }
104            let result = self.rpc(wire::TWALK, w).await;
105            let walked = match result {
106                Ok((_, body)) => Reader::new(&body).u16()? as usize,
107                Err(e) => {
108                    if from == fid {
109                        self.clunk(fid).await;
110                    }
111                    return Err(e);
112                }
113            };
114            if walked != chunk.len() {
115                if from == fid {
116                    self.clunk(fid).await;
117                }
118                return Err(remote("file does not exist"));
119            }
120            from = fid;
121        }
122        Ok(fid)
123    }
124
125    /// Open `fid` and return the I/O unit to read or write in.
126    async fn open(&mut self, fid: u32, mode: u8) -> Result<u32> {
127        let (_, body) = self
128            .rpc(wire::TOPEN, Writer::new().u32(fid).u8(mode))
129            .await?;
130        let mut r = Reader::new(&body);
131        r.qid()?;
132        Ok(self.iounit(r.u32()?))
133    }
134
135    fn iounit(&self, offered: u32) -> u32 {
136        let max = self.msize - wire::IOHDRSZ;
137        if offered == 0 {
138            max
139        } else {
140            offered.min(max)
141        }
142    }
143
144    async fn read_all(&mut self, fid: u32, iounit: u32) -> Result<Vec<u8>> {
145        let mut data = Vec::new();
146        loop {
147            let (_, body) = self
148                .rpc(
149                    wire::TREAD,
150                    Writer::new().u32(fid).u64(data.len() as u64).u32(iounit),
151                )
152                .await?;
153            let mut r = Reader::new(&body);
154            let count = r.u32()? as usize;
155            anyhow::ensure!(
156                count <= iounit as usize,
157                "9P server returned more than asked"
158            );
159            if count == 0 {
160                return Ok(data);
161            }
162            data.extend_from_slice(r.take(count)?);
163            anyhow::ensure!(data.len() <= wire::MAX_CONTENT, "9P content exceeds 1 MiB");
164        }
165    }
166
167    async fn stat(&mut self, fid: u32) -> Result<Stat> {
168        let (_, body) = self.rpc(wire::TSTAT, Writer::new().u32(fid)).await?;
169        let mut r = Reader::new(&body);
170        r.u16()?;
171        r.stat()
172    }
173
174    /// Run one action and return its result fields (merged into ninep_result).
175    async fn run(&mut self, action: &Value) -> Result<Value> {
176        let path = action["path"].as_str().unwrap_or("/");
177        match action["type"].as_str().unwrap_or_default() {
178            "ninep_ls" => {
179                let fid = self.walk(path).await?;
180                let result = async {
181                    let iounit = self.open(fid, wire::OREAD).await?;
182                    let data = self.read_all(fid, iounit).await?;
183                    let mut r = Reader::new(&data);
184                    let mut entries = Vec::new();
185                    while !r.is_empty() {
186                        anyhow::ensure!(
187                            entries.len() < wire::MAX_ENTRIES,
188                            "listing exceeds 1024 entries"
189                        );
190                        entries.push(stat_json(&r.stat()?));
191                    }
192                    Ok(json!({"entries": entries}))
193                }
194                .await;
195                self.clunk(fid).await;
196                result
197            }
198            "ninep_cat" => {
199                let fid = self.walk(path).await?;
200                let result = async {
201                    let iounit = self.open(fid, wire::OREAD).await?;
202                    let data = self.read_all(fid, iounit).await?;
203                    let (text, encoding) = wire::to_text(&data);
204                    Ok(json!({"data": text, "encoding": encoding}))
205                }
206                .await;
207                self.clunk(fid).await;
208                result
209            }
210            "ninep_stat" => {
211                let fid = self.walk(path).await?;
212                let result = self.stat(fid).await;
213                self.clunk(fid).await;
214                Ok(json!({"stat": stat_json(&result?)}))
215            }
216            "ninep_write" => self.write(path, action).await,
217            "ninep_mkdir" => {
218                let fid = self.walk(&wire::parent(path)).await?;
219                let result = self
220                    .rpc(
221                        wire::TCREATE,
222                        Writer::new()
223                            .u32(fid)
224                            .string(&wire::base(path))
225                            .u32(wire::DMDIR | 0o755)
226                            .u8(wire::OREAD),
227                    )
228                    .await;
229                self.clunk(fid).await;
230                result.map(|_| json!({}))
231            }
232            "ninep_remove" => {
233                let fid = self.walk(path).await?;
234                // Tremove clunks the fid whatever the answer.
235                self.rpc(wire::TREMOVE, Writer::new().u32(fid)).await?;
236                Ok(json!({}))
237            }
238            "ninep_rename" => {
239                let fid = self.walk(path).await?;
240                let stat = Stat {
241                    name: action["name"].as_str().unwrap_or_default().to_string(),
242                    ..Stat::dont_touch()
243                };
244                let encoded = wire::encode_stat(&stat);
245                let result = self
246                    .rpc(
247                        wire::TWSTAT,
248                        Writer::new()
249                            .u32(fid)
250                            .u16(encoded.len() as u16)
251                            .bytes(&encoded),
252                    )
253                    .await;
254                self.clunk(fid).await;
255                result.map(|_| json!({}))
256            }
257            other => anyhow::bail!("Unknown 9P client action {other}"),
258        }
259    }
260
261    async fn write(&mut self, path: &str, action: &Value) -> Result<Value> {
262        let data = wire::from_text(
263            action["data"].as_str().unwrap_or_default(),
264            action["encoding"].as_str(),
265        )?;
266        let append = action["append"] == true;
267        let (fid, iounit) = match self.walk(path).await {
268            Ok(fid) => {
269                let mode = if append {
270                    wire::OWRITE
271                } else {
272                    wire::OWRITE | wire::OTRUNC
273                };
274                match self.open(fid, mode).await {
275                    Ok(iounit) => (fid, iounit),
276                    Err(e) => {
277                        self.clunk(fid).await;
278                        return Err(e);
279                    }
280                }
281            }
282            Err(e) if action["create"] == true && e.downcast_ref::<Remote>().is_some() => {
283                let fid = self.walk(&wire::parent(path)).await?;
284                match self
285                    .rpc(
286                        wire::TCREATE,
287                        Writer::new()
288                            .u32(fid)
289                            .string(&wire::base(path))
290                            .u32(0o644)
291                            .u8(wire::OWRITE),
292                    )
293                    .await
294                {
295                    Ok((_, body)) => {
296                        let mut r = Reader::new(&body);
297                        r.qid()?;
298                        let iounit = self.iounit(r.u32()?);
299                        (fid, iounit)
300                    }
301                    Err(e) => {
302                        self.clunk(fid).await;
303                        return Err(e);
304                    }
305                }
306            }
307            Err(e) => return Err(e),
308        };
309        let result = async {
310            let mut offset = if append {
311                self.stat(fid).await?.length
312            } else {
313                0
314            };
315            let mut written = 0usize;
316            for chunk in data
317                .chunks(iounit as usize)
318                .chain(data.is_empty().then_some(&[][..]))
319            {
320                let (_, body) = self
321                    .rpc(
322                        wire::TWRITE,
323                        Writer::new()
324                            .u32(fid)
325                            .u64(offset)
326                            .u32(chunk.len() as u32)
327                            .bytes(chunk),
328                    )
329                    .await?;
330                let count = Reader::new(&body).u32()? as usize;
331                anyhow::ensure!(
332                    count <= chunk.len(),
333                    "9P server wrote more than it was sent"
334                );
335                written += count;
336                offset += count as u64;
337                if count < chunk.len() {
338                    break;
339                }
340            }
341            Ok(json!({"bytes_written": written}))
342        }
343        .await;
344        self.clunk(fid).await;
345        result
346    }
347}
348
349fn stat_json(s: &Stat) -> Value {
350    json!({
351        "name": s.name,
352        "kind": if s.mode & wire::DMDIR != 0 { "dir" } else { "file" },
353        "size": s.length,
354        "mode": s.mode & 0o777,
355        "owner": s.uid,
356        "mtime": s.mtime,
357    })
358}
359
360pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
361    let params = ctx.startup_params.as_ref();
362    let uname = params
363        .map(|p| p.get_optional_string("uname"))
364        .transpose()?
365        .flatten()
366        .unwrap_or_else(|| DEFAULT_UNAME.to_string());
367    let aname = params
368        .map(|p| p.get_optional_string("aname"))
369        .transpose()?
370        .flatten()
371        .unwrap_or_default();
372    let stream = tokio::time::timeout(wire::IO_TIMEOUT, TcpStream::connect(&ctx.remote_addr))
373        .await
374        .context("9P connect deadline")??;
375    let local = stream.local_addr()?;
376    let (reader, writer) = tokio::io::split(stream);
377    let mut conn = Conn {
378        reader,
379        writer,
380        msize: wire::MAX_MSIZE,
381        next_tag: 0,
382        next_fid: ROOT_FID,
383    };
384    let (_, body) = conn
385        .rpc(
386            wire::TVERSION,
387            Writer::new().u32(wire::MAX_MSIZE).string(wire::VERSION),
388        )
389        .await?;
390    let mut r = Reader::new(&body);
391    let msize = r.u32()?;
392    let version = r.string()?;
393    anyhow::ensure!(
394        version == wire::VERSION,
395        "9P server offered version {version:?}"
396    );
397    anyhow::ensure!(
398        (wire::MIN_MSIZE..=wire::MAX_MSIZE).contains(&msize),
399        "9P server offered msize {msize}"
400    );
401    conn.msize = msize;
402    conn.rpc(
403        wire::TATTACH,
404        Writer::new()
405            .u32(ROOT_FID)
406            .u32(wire::NOFID)
407            .string(&uname)
408            .string(&aname),
409    )
410    .await
411    .map_err(|e| anyhow::anyhow!("9P attach refused: {e}"))?;
412    ctx.state
413        .update_client_status(ctx.client_id, ClientStatus::Connected)
414        .await;
415    let external =
416        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
417    let (internal_tx, internal_rx) = mpsc::channel::<Value>(16);
418    let (event_tx, mut event_rx) = mpsc::channel::<Event>(16);
419    event_tx.try_send(Event::new(
420        &actions::CONNECTED_EVENT,
421        json!({"version": version, "msize": msize, "uname": uname}),
422    ))?;
423    let events_ctx = ctx.clone();
424    let dispatcher = tokio::spawn(async move {
425        let protocol = NinepClientProtocol;
426        while let Some(event) = event_rx.recv().await {
427            let instruction = events_ctx
428                .state
429                .get_instruction_for_client(events_ctx.client_id)
430                .await
431                .unwrap_or_default();
432            let memory = events_ctx
433                .state
434                .get_memory_for_client(events_ctx.client_id)
435                .await
436                .unwrap_or_default();
437            match call_llm_for_client(
438                &events_ctx.llm_client,
439                &events_ctx.state,
440                events_ctx.client_id.to_string(),
441                &instruction,
442                &memory,
443                Some(&event),
444                &protocol,
445                &events_ctx.status_tx,
446            )
447            .await
448            {
449                Ok(result) => {
450                    if let Some(memory) = result.memory_updates {
451                        events_ctx
452                            .state
453                            .set_memory_for_client(events_ctx.client_id, memory)
454                            .await;
455                    }
456                    for action in result.actions {
457                        if internal_tx.send(action).await.is_err() {
458                            return;
459                        }
460                    }
461                }
462                Err(e) => {
463                    Log::new(Some(&events_ctx.status_tx)).warn(format!("9P client handler: {e}"))
464                }
465            }
466        }
467    });
468    let dispatcher_abort = dispatcher.abort_handle();
469    ctx.state
470        .register_client_task(ctx.client_id, dispatcher)
471        .await;
472    let session_ctx = ctx.clone();
473    let task = tokio::spawn(async move {
474        let result = session(&session_ctx, conn, external, internal_rx, event_tx).await;
475        if result.is_err() {
476            dispatcher_abort.abort();
477        }
478        let status = match result {
479            Ok(()) => ClientStatus::Disconnected,
480            Err(e) => {
481                Log::new(Some(&session_ctx.status_tx)).warn(format!("9P client ended: {e}"));
482                ClientStatus::Error(e.to_string())
483            }
484        };
485        session_ctx
486            .state
487            .update_client_status(session_ctx.client_id, status)
488            .await;
489        session_ctx
490            .state
491            .remove_client_handle(session_ctx.client_id)
492            .await;
493        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
494    });
495    ctx.state.register_client_task(ctx.client_id, task).await;
496    Ok(local)
497}
498
499async fn session(
500    ctx: &ConnectContext,
501    mut conn: Conn,
502    mut external: mpsc::Receiver<ClientCommand>,
503    mut internal: mpsc::Receiver<Value>,
504    events: mpsc::Sender<Event>,
505) -> Result<()> {
506    loop {
507        let (action, mut injected) = tokio::select! {
508            command = external.recv() => match command {
509                Some(c) => (c.action.clone(), Some(c)),
510                None => return Ok(()),
511            },
512            action = internal.recv() => match action {
513                Some(a) => (a, None),
514                None => return Ok(()),
515            },
516        };
517        match NinepClientProtocol.execute_action(action.clone()) {
518            Ok(ClientActionResult::Disconnect) => {
519                conn.clunk(ROOT_FID).await;
520                if let Some(command) = injected.take() {
521                    crate::client::command_support::reply(
522                        command,
523                        Ok(ClientSendOutcome::Disconnected),
524                    );
525                }
526                return Ok(());
527            }
528            Ok(_) => {}
529            Err(e) => {
530                if let Some(command) = injected.take() {
531                    crate::client::command_support::reply(
532                        command,
533                        Ok(ClientSendOutcome::Rejected {
534                            error: e.to_string(),
535                        }),
536                    );
537                }
538                continue;
539            }
540        }
541        let op = action["type"]
542            .as_str()
543            .unwrap_or_default()
544            .trim_start_matches("ninep_")
545            .to_string();
546        let path = action["path"].as_str().unwrap_or("/").to_string();
547        if injected.is_some() {
548            ctx.state
549                .record_access_log(
550                    AccessLogOwner::Client(ctx.client_id.as_u32()),
551                    "9P",
552                    None,
553                    "injected_action",
554                    json!({"op": op, "path": path}),
555                    vec![],
556                )
557                .await;
558        }
559        let mut event = json!({"op": op, "path": path});
560        match conn.run(&action).await {
561            Ok(fields) => {
562                event["ok"] = json!(true);
563                if let Some(map) = fields.as_object() {
564                    for (k, v) in map {
565                        event[k] = v.clone();
566                    }
567                }
568            }
569            Err(e) => match e.downcast_ref::<Remote>() {
570                Some(Remote(text)) => {
571                    event["ok"] = json!(false);
572                    event["error"] = json!(text);
573                }
574                // A transport or framing failure ends the session.
575                None => {
576                    if let Some(command) = injected.take() {
577                        crate::client::command_support::reply(
578                            command,
579                            Err(anyhow::anyhow!(e.to_string())),
580                        );
581                    }
582                    return Err(e);
583                }
584            },
585        }
586        if let Some(command) = injected.take() {
587            crate::client::command_support::reply(
588                command,
589                Ok(ClientSendOutcome::Executed {
590                    detail: event.to_string(),
591                }),
592            );
593        }
594        events
595            .try_send(Event::new(&actions::RESULT_EVENT, event))
596            .context("9P event queue full; consumer stalled")?;
597    }
598}
netget/client/ninep/mod.rs

netget/client/ninep/
mod.rs
