netget/server/nbd/
mod.rs

1//! Network Block Device server. Rust owns the fixed newstyle handshake, option haggling and
2//! transmission; the handler describes each export once per connection and Rust serves every
3//! read from that description. Exports are read-only.
4pub mod actions;
5pub mod wire;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, SpawnContext};
11use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
12use crate::server::connection::ConnectionId;
13use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
14use crate::utils::clock::Instant;
15use anyhow::{bail, ensure, Context, Result};
16use serde_json::json;
17use std::collections::HashMap;
18use std::net::SocketAddr;
19use std::sync::Arc;
20use std::time::Duration;
21use tokio::io::{AsyncReadExt, AsyncWriteExt};
22use tokio::net::TcpStream;
23use wire::*;
24
25/// The whole handshake, and the rest of any message once its first byte has arrived.
26pub const HANDSHAKE_DEADLINE: Duration = Duration::from_secs(60);
27pub const MESSAGE_DEADLINE: Duration = Duration::from_secs(60);
28pub const MAX_OPTIONS: usize = 64;
29/// Block status descriptors in one reply.
30pub const MAX_DESCRIPTORS: usize = 1024;
31
32pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
33    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
34    let local = listener.local_addr()?;
35    Log::new(Some(&ctx.status_tx)).info(format!("NBD server on {local}"));
36    let shared = Arc::new(ctx.clone());
37    let server_id = ctx.server_id;
38    let accept =
39        tokio::spawn(async move {
40            let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
41            loop {
42                let (stream, peer, permit) =
43                    match accept_bounded(&listener, &limiter, b"", "NBD", Some(&shared.status_tx))
44                        .await
45                    {
46                        Ok(v) => v,
47                        Err(_) => break,
48                    };
49                let id = ConnectionId::new(shared.state.get_next_unified_id().await);
50                let now = Instant::now();
51                shared
52                    .state
53                    .add_connection_to_server(
54                        server_id,
55                        ConnectionState {
56                            id,
57                            remote_addr: peer,
58                            local_addr: local,
59                            bytes_sent: 0,
60                            bytes_received: 0,
61                            packets_sent: 0,
62                            packets_received: 0,
63                            last_activity: now,
64                            status: ConnectionStatus::Active,
65                            status_changed_at: now,
66                            protocol_info: ProtocolConnectionInfo::empty(),
67                        },
68                    )
69                    .await;
70                let child = shared.clone();
71                shared
72                    .state
73                    .spawn_server_task(server_id, async move {
74                        let _permit = permit;
75                        let mut c = Conn {
76                            ctx: &child,
77                            id,
78                            peer,
79                            stream,
80                            structured: false,
81                            meta: false,
82                            exports: HashMap::new(),
83                        };
84                        if let Err(e) = c.run().await {
85                            Log::new(Some(&child.status_tx))
86                                .debug(format!("NBD connection {id}: {e:#}"));
87                        }
88                        child
89                            .state
90                            .update_connection_status(server_id, id, ConnectionStatus::Closed)
91                            .await;
92                        let _ = child.status_tx.send("__UPDATE_UI__".into());
93                    })
94                    .await;
95            }
96        });
97    ctx.state.register_server_task(server_id, accept).await;
98    Ok(local)
99}
100
101fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
102    let summary = format!("NBD connection {id} operation={operation} decision={decision}");
103    let log = Log::new(Some(&ctx.status_tx));
104    if decision.starts_with("fail_closed_") || decision == "model_silent" {
105        log.error(summary);
106    } else {
107        log.info(summary);
108    }
109}
110
111/// What the handler said about an export or the list.
112enum Answer {
113    Export(Arc<Export>),
114    List(Vec<(String, Option<String>)>),
115    Refused(u32),
116}
117
118struct Conn<'a> {
119    ctx: &'a SpawnContext,
120    id: ConnectionId,
121    peer: SocketAddr,
122    stream: TcpStream,
123    structured: bool,
124    meta: bool,
125    exports: HashMap<String, Option<Arc<Export>>>,
126}
127
128fn opt_reply(option: u32, kind: u32, data: &[u8]) -> Vec<u8> {
129    let mut out = Vec::with_capacity(20 + data.len());
130    out.extend(REPLY_MAGIC.to_be_bytes());
131    out.extend(option.to_be_bytes());
132    out.extend(kind.to_be_bytes());
133    out.extend((data.len() as u32).to_be_bytes());
134    out.extend(data);
135    out
136}
137
138fn chunk(flags: u16, kind: u16, cookie: u64, payload: &[u8]) -> Vec<u8> {
139    let mut out = Vec::with_capacity(20 + payload.len());
140    out.extend(STRUCTURED_REPLY_MAGIC.to_be_bytes());
141    out.extend(flags.to_be_bytes());
142    out.extend(kind.to_be_bytes());
143    out.extend(cookie.to_be_bytes());
144    out.extend((payload.len() as u32).to_be_bytes());
145    out.extend(payload);
146    out
147}
148
149fn simple(error: u32, cookie: u64, data: &[u8]) -> Vec<u8> {
150    let mut out = Vec::with_capacity(16 + data.len());
151    out.extend(SIMPLE_REPLY_MAGIC.to_be_bytes());
152    out.extend(error.to_be_bytes());
153    out.extend(cookie.to_be_bytes());
154    out.extend(data);
155    out
156}
157
158fn transmission_flags(structured: bool) -> u16 {
159    let mut f = TFLAG_HAS_FLAGS
160        | TFLAG_READ_ONLY
161        | TFLAG_SEND_FLUSH
162        | TFLAG_CAN_MULTI_CONN
163        | TFLAG_SEND_CACHE;
164    if structured {
165        f |= TFLAG_SEND_DF;
166    }
167    f
168}
169
170impl Conn<'_> {
171    async fn read_exact(&mut self, buf: &mut [u8], deadline: tokio::time::Instant) -> Result<()> {
172        tokio::time::timeout_at(deadline, self.stream.read_exact(buf))
173            .await
174            .context("the client stalled")??;
175        Ok(())
176    }
177
178    async fn write(&mut self, bytes: &[u8]) -> Result<()> {
179        self.stream.write_all(bytes).await?;
180        self.ctx
181            .state
182            .update_connection_stats(
183                self.ctx.server_id,
184                self.id,
185                None,
186                Some(bytes.len() as u64),
187                None,
188                Some(1),
189            )
190            .await;
191        Ok(())
192    }
193
194    async fn ask(&mut self, event: Event, operation: &str) -> Answer {
195        let result = match call_llm(
196            &self.ctx.llm_client,
197            &self.ctx.state,
198            self.ctx.server_id,
199            Some(self.id),
200            &event,
201            &actions::NbdProtocol,
202        )
203        .await
204        {
205            Ok(r) => r,
206            Err(_) => {
207                outcome(self.ctx, self.id, operation, "fail_closed_llm_error");
208                return Answer::Refused(REP_ERR_POLICY);
209            }
210        };
211        let mut answers = Vec::new();
212        let mut pending = result.protocol_results;
213        while let Some(r) = pending.pop() {
214            match r {
215                ActionResult::Custom { data, .. } => answers.push(data),
216                ActionResult::Multiple(items) => pending.extend(items),
217                _ => {}
218            }
219        }
220        if !result.failures.is_empty() || answers.len() != 1 {
221            outcome(
222                self.ctx,
223                self.id,
224                operation,
225                if answers.is_empty() && result.failures.is_empty() {
226                    "model_silent"
227                } else {
228                    "fail_closed_invalid_reply"
229                },
230            );
231            return Answer::Refused(REP_ERR_POLICY);
232        }
233        let a = answers.remove(0);
234        match a["type"].as_str() {
235            Some("nbd_export") if operation == "export" => match Export::from_action(&a) {
236                Ok(e) => {
237                    outcome(self.ctx, self.id, operation, "model_answer");
238                    Answer::Export(Arc::new(e))
239                }
240                Err(_) => {
241                    outcome(self.ctx, self.id, operation, "fail_closed_invalid_reply");
242                    Answer::Refused(REP_ERR_POLICY)
243                }
244            },
245            Some("nbd_list_exports") if operation == "list" => {
246                outcome(self.ctx, self.id, operation, "model_answer");
247                Answer::List(
248                    a["exports"]
249                        .as_array()
250                        .into_iter()
251                        .flatten()
252                        .filter_map(|e| {
253                            Some((
254                                e["name"].as_str()?.to_owned(),
255                                e["description"].as_str().map(str::to_owned),
256                            ))
257                        })
258                        .collect(),
259                )
260            }
261            Some("nbd_reject") => {
262                outcome(self.ctx, self.id, operation, "model_reject");
263                Answer::Refused(if a["reason"] == "unknown" {
264                    REP_ERR_UNKNOWN
265                } else {
266                    REP_ERR_POLICY
267                })
268            }
269            _ => {
270                outcome(self.ctx, self.id, operation, "fail_closed_invalid_reply");
271                Answer::Refused(REP_ERR_POLICY)
272            }
273        }
274    }
275
276    /// The handler's description of an export, asked once per name per connection.
277    async fn export(&mut self, name: &str, option: &str) -> Result<Arc<Export>, u32> {
278        if let Some(known) = self.exports.get(name) {
279            return known.clone().ok_or(REP_ERR_UNKNOWN);
280        }
281        let event = Event::new(
282            &actions::EXPORT_EVENT,
283            json!({"export": name, "option": option}),
284        );
285        match self.ask(event, "export").await {
286            Answer::Export(e) => {
287                self.exports.insert(name.to_owned(), Some(e.clone()));
288                Ok(e)
289            }
290            Answer::Refused(code) => {
291                if code == REP_ERR_UNKNOWN {
292                    self.exports.insert(name.to_owned(), None);
293                }
294                Err(code)
295            }
296            Answer::List(_) => Err(REP_ERR_POLICY),
297        }
298    }
299
300    async fn run(&mut self) -> Result<()> {
301        let mut hello = Vec::with_capacity(18);
302        hello.extend(NBDMAGIC.to_be_bytes());
303        hello.extend(IHAVEOPT.to_be_bytes());
304        hello.extend((FLAG_FIXED_NEWSTYLE | FLAG_NO_ZEROES).to_be_bytes());
305        self.write(&hello).await?;
306        let deadline = tokio::time::Instant::now() + HANDSHAKE_DEADLINE;
307        let mut b4 = [0u8; 4];
308        self.read_exact(&mut b4, deadline).await?;
309        let client_flags = u32::from_be_bytes(b4);
310        let (fixed, no_zeroes) = (client_flags & 1 != 0, client_flags & 2 != 0);
311        ensure!(
312            client_flags & !3 == 0,
313            "unknown client flags {client_flags:#x}"
314        );
315        let mut options = 0;
316        let export = loop {
317            options += 1;
318            if options > MAX_OPTIONS {
319                outcome(self.ctx, self.id, "negotiate", "protocol_refusal");
320                bail!("more than {MAX_OPTIONS} options");
321            }
322            let mut head = [0u8; 16];
323            self.read_exact(&mut head, deadline).await?;
324            ensure!(
325                u64::from_be_bytes(head[..8].try_into()?) == IHAVEOPT,
326                "an option without IHAVEOPT"
327            );
328            let option = u32::from_be_bytes(head[8..12].try_into()?);
329            let len = u32::from_be_bytes(head[12..16].try_into()?) as usize;
330            if len > MAX_OPTION {
331                outcome(self.ctx, self.id, "negotiate", "protocol_refusal");
332                bail!("option {option} declares {len} bytes");
333            }
334            let mut data = vec![0u8; len];
335            self.read_exact(&mut data, deadline).await?;
336            self.ctx
337                .state
338                .update_connection_stats(
339                    self.ctx.server_id,
340                    self.id,
341                    Some(16 + len as u64),
342                    None,
343                    Some(1),
344                    None,
345                )
346                .await;
347            if !fixed && option != OPT_EXPORT_NAME {
348                bail!("a client without fixed newstyle may only send EXPORT_NAME");
349            }
350            match option {
351                OPT_EXPORT_NAME => {
352                    let name = String::from_utf8_lossy(&data).into_owned();
353                    match self.export(&name, "export_name").await {
354                        Ok(e) => {
355                            let mut out = e.size.to_be_bytes().to_vec();
356                            out.extend(transmission_flags(self.structured).to_be_bytes());
357                            if !no_zeroes {
358                                out.extend([0u8; 124]);
359                            }
360                            self.write(&out).await?;
361                            break e;
362                        }
363                        // EXPORT_NAME has no error reply: closing is the refusal.
364                        Err(_) => return Ok(()),
365                    }
366                }
367                OPT_ABORT => {
368                    self.write(&opt_reply(option, REP_ACK, &[])).await?;
369                    return Ok(());
370                }
371                OPT_LIST => {
372                    if len != 0 {
373                        self.write(&opt_reply(option, REP_ERR_INVALID, b"LIST takes no data"))
374                            .await?;
375                        continue;
376                    }
377                    match self
378                        .ask(
379                            Event::new(
380                                &actions::LIST_EVENT,
381                                json!({"client": self.peer.to_string()}),
382                            ),
383                            "list",
384                        )
385                        .await
386                    {
387                        Answer::List(list) => {
388                            for (name, description) in list {
389                                let mut d = (name.len() as u32).to_be_bytes().to_vec();
390                                d.extend(name.as_bytes());
391                                d.extend(description.unwrap_or_default().as_bytes());
392                                self.write(&opt_reply(option, REP_SERVER, &d)).await?;
393                            }
394                            self.write(&opt_reply(option, REP_ACK, &[])).await?;
395                        }
396                        _ => {
397                            self.write(&opt_reply(
398                                option,
399                                REP_ERR_POLICY,
400                                b"listing is not allowed",
401                            ))
402                            .await?
403                        }
404                    }
405                }
406                OPT_STRUCTURED_REPLY => {
407                    if len != 0 {
408                        self.write(&opt_reply(option, REP_ERR_INVALID, &[])).await?;
409                    } else {
410                        self.structured = true;
411                        self.write(&opt_reply(option, REP_ACK, &[])).await?;
412                    }
413                }
414                OPT_LIST_META_CONTEXT | OPT_SET_META_CONTEXT => {
415                    let Some(queries) = meta_queries(&data) else {
416                        self.write(&opt_reply(
417                            option,
418                            REP_ERR_INVALID,
419                            b"malformed meta context request",
420                        ))
421                        .await?;
422                        continue;
423                    };
424                    if option == OPT_SET_META_CONTEXT && !self.structured {
425                        self.write(&opt_reply(
426                            option,
427                            REP_ERR_INVALID,
428                            b"negotiate structured replies first",
429                        ))
430                        .await?;
431                        continue;
432                    }
433                    let wanted = if option == OPT_LIST_META_CONTEXT {
434                        queries.is_empty()
435                            || queries.iter().any(|q| q == "base:" || q == BASE_ALLOCATION)
436                    } else {
437                        queries.iter().any(|q| q == BASE_ALLOCATION)
438                    };
439                    if option == OPT_SET_META_CONTEXT {
440                        self.meta = wanted;
441                    }
442                    if wanted {
443                        let mut d = 1u32.to_be_bytes().to_vec();
444                        d.extend(BASE_ALLOCATION.as_bytes());
445                        self.write(&opt_reply(option, REP_META_CONTEXT, &d)).await?;
446                    }
447                    self.write(&opt_reply(option, REP_ACK, &[])).await?;
448                }
449                OPT_INFO | OPT_GO => {
450                    let Some((name, infos)) = info_request(&data) else {
451                        self.write(&opt_reply(
452                            option,
453                            REP_ERR_INVALID,
454                            b"malformed INFO/GO request",
455                        ))
456                        .await?;
457                        continue;
458                    };
459                    let which = if option == OPT_GO { "go" } else { "info" };
460                    let e = match self.export(&name, which).await {
461                        Ok(e) => e,
462                        Err(code) => {
463                            let text: &[u8] = if code == REP_ERR_UNKNOWN {
464                                b"no such export"
465                            } else {
466                                b"refused"
467                            };
468                            self.write(&opt_reply(option, code, text)).await?;
469                            continue;
470                        }
471                    };
472                    let mut info = INFO_EXPORT.to_be_bytes().to_vec();
473                    info.extend(e.size.to_be_bytes());
474                    info.extend(transmission_flags(self.structured).to_be_bytes());
475                    self.write(&opt_reply(option, REP_INFO, &info)).await?;
476                    if infos.contains(&INFO_NAME) {
477                        let mut d = INFO_NAME.to_be_bytes().to_vec();
478                        d.extend(name.as_bytes());
479                        self.write(&opt_reply(option, REP_INFO, &d)).await?;
480                    }
481                    if let (true, Some(desc)) = (infos.contains(&INFO_DESCRIPTION), &e.description)
482                    {
483                        let mut d = INFO_DESCRIPTION.to_be_bytes().to_vec();
484                        d.extend(desc.as_bytes());
485                        self.write(&opt_reply(option, REP_INFO, &d)).await?;
486                    }
487                    if infos.contains(&INFO_BLOCK_SIZE) {
488                        let mut d = INFO_BLOCK_SIZE.to_be_bytes().to_vec();
489                        for v in [e.block_min, e.block_preferred, e.block_max] {
490                            d.extend(v.to_be_bytes());
491                        }
492                        self.write(&opt_reply(option, REP_INFO, &d)).await?;
493                    }
494                    self.write(&opt_reply(option, REP_ACK, &[])).await?;
495                    if option == OPT_GO {
496                        break e;
497                    }
498                }
499                _ => {
500                    self.write(&opt_reply(option, REP_ERR_UNSUP, b"not supported"))
501                        .await?
502                }
503            }
504        };
505        self.transmission(&export).await
506    }
507
508    async fn transmission(&mut self, e: &Export) -> Result<()> {
509        loop {
510            let mut req = [0u8; 28];
511            let n = self.stream.read(&mut req[..1]).await?;
512            if n == 0 {
513                return Ok(());
514            }
515            let deadline = tokio::time::Instant::now() + MESSAGE_DEADLINE;
516            self.read_exact(&mut req[1..], deadline).await?;
517            ensure!(
518                u32::from_be_bytes(req[..4].try_into()?) == REQUEST_MAGIC,
519                "bad request magic"
520            );
521            let flags = u16::from_be_bytes(req[4..6].try_into()?);
522            let kind = u16::from_be_bytes(req[6..8].try_into()?);
523            let cookie = u64::from_be_bytes(req[8..16].try_into()?);
524            let offset = u64::from_be_bytes(req[16..24].try_into()?);
525            let length = u32::from_be_bytes(req[24..28].try_into()?);
526            self.ctx
527                .state
528                .update_connection_stats(self.ctx.server_id, self.id, Some(28), None, Some(1), None)
529                .await;
530            let in_range = offset
531                .checked_add(length as u64)
532                .is_some_and(|end| end <= e.size);
533            match kind {
534                CMD_DISC => return Ok(()),
535                CMD_WRITE => {
536                    // Drain the payload so the stream stays in step, then refuse.
537                    ensure!(length <= MAX_REQUEST, "a write of {length} bytes");
538                    let mut sink = vec![0u8; length as usize];
539                    self.read_exact(&mut sink, deadline).await?;
540                    self.fail(cookie, EPERM, None, "the export is read-only")
541                        .await?;
542                }
543                CMD_TRIM | CMD_WRITE_ZEROES => {
544                    self.fail(cookie, EPERM, None, "the export is read-only")
545                        .await?
546                }
547                CMD_FLUSH | CMD_CACHE => self.ok(cookie).await?,
548                CMD_READ => {
549                    if length == 0 || length > e.block_max || !in_range {
550                        self.fail(
551                            cookie,
552                            EINVAL,
553                            None,
554                            "the read is outside the export or over the maximum block size",
555                        )
556                        .await?;
557                    } else if let Some((at, err)) = e.error_in(offset, length as u64) {
558                        self.fail(cookie, err, Some(at), "this region fails reads")
559                            .await?;
560                    } else if !self.structured {
561                        self.write(&simple(0, cookie, &e.read(offset, length)))
562                            .await?;
563                    } else if flags & CMD_FLAG_DF != 0 {
564                        let mut p = offset.to_be_bytes().to_vec();
565                        p.extend(e.read(offset, length));
566                        self.write(&chunk(REPLY_FLAG_DONE, REPLY_OFFSET_DATA, cookie, &p))
567                            .await?;
568                    } else {
569                        let pieces = e.pieces(offset, length);
570                        let last = pieces.len() - 1;
571                        for (i, piece) in pieces.into_iter().enumerate() {
572                            let done = if i == last { REPLY_FLAG_DONE } else { 0 };
573                            let bytes = match piece {
574                                Piece::Data(o, data) => {
575                                    let mut p = o.to_be_bytes().to_vec();
576                                    p.extend(data);
577                                    chunk(done, REPLY_OFFSET_DATA, cookie, &p)
578                                }
579                                Piece::Hole(o, n) => {
580                                    let mut p = o.to_be_bytes().to_vec();
581                                    p.extend(n.to_be_bytes());
582                                    chunk(done, REPLY_OFFSET_HOLE, cookie, &p)
583                                }
584                            };
585                            self.write(&bytes).await?;
586                        }
587                    }
588                }
589                CMD_BLOCK_STATUS => {
590                    if !self.meta || length == 0 || !in_range {
591                        self.fail(
592                            cookie,
593                            EINVAL,
594                            None,
595                            "block status needs base:allocation and a range inside the export",
596                        )
597                        .await?;
598                    } else {
599                        let mut p = 1u32.to_be_bytes().to_vec();
600                        for (_, len, data) in e
601                            .runs(offset, length as u64)
602                            .into_iter()
603                            .take(MAX_DESCRIPTORS)
604                        {
605                            p.extend((len as u32).to_be_bytes());
606                            p.extend(
607                                (if data { 0 } else { STATE_HOLE | STATE_ZERO }).to_be_bytes(),
608                            );
609                        }
610                        self.write(&chunk(REPLY_FLAG_DONE, REPLY_BLOCK_STATUS, cookie, &p))
611                            .await?;
612                    }
613                }
614                _ => self.fail(cookie, EINVAL, None, "unknown command").await?,
615            }
616        }
617    }
618
619    async fn ok(&mut self, cookie: u64) -> Result<()> {
620        if self.structured {
621            self.write(&chunk(REPLY_FLAG_DONE, REPLY_NONE, cookie, &[]))
622                .await
623        } else {
624            self.write(&simple(0, cookie, &[])).await
625        }
626    }
627
628    async fn fail(
629        &mut self,
630        cookie: u64,
631        errno: u32,
632        at: Option<u64>,
633        message: &str,
634    ) -> Result<()> {
635        if !self.structured {
636            return self.write(&simple(errno, cookie, &[])).await;
637        }
638        let mut p = errno.to_be_bytes().to_vec();
639        p.extend((message.len() as u16).to_be_bytes());
640        p.extend(message.as_bytes());
641        let kind = match at {
642            Some(o) => {
643                p.extend(o.to_be_bytes());
644                REPLY_ERROR_OFFSET
645            }
646            None => REPLY_ERROR,
647        };
648        self.write(&chunk(REPLY_FLAG_DONE, kind, cookie, &p)).await
649    }
650}
651
652/// An INFO/GO request's export name and requested info types.
653fn info_request(d: &[u8]) -> Option<(String, Vec<u16>)> {
654    let n = u32::from_be_bytes(d.get(..4)?.try_into().ok()?) as usize;
655    let name = String::from_utf8(d.get(4..4 + n)?.to_vec()).ok()?;
656    let rest = d.get(4 + n..)?;
657    let count = u16::from_be_bytes(rest.get(..2)?.try_into().ok()?) as usize;
658    let list = rest.get(2..)?;
659    if list.len() != count * 2 {
660        return None;
661    }
662    Some((
663        name,
664        list.chunks(2)
665            .map(|c| u16::from_be_bytes([c[0], c[1]]))
666            .collect(),
667    ))
668}
669
670/// A meta context request's queries.
671fn meta_queries(d: &[u8]) -> Option<Vec<String>> {
672    let n = u32::from_be_bytes(d.get(..4)?.try_into().ok()?) as usize;
673    let mut at = 4 + n;
674    d.get(4..at)?;
675    let count = u32::from_be_bytes(d.get(at..at + 4)?.try_into().ok()?) as usize;
676    at += 4;
677    let mut out = Vec::new();
678    for _ in 0..count.min(256) {
679        let l = u32::from_be_bytes(d.get(at..at + 4)?.try_into().ok()?) as usize;
680        out.push(String::from_utf8(d.get(at + 4..at + 4 + l)?.to_vec()).ok()?);
681        at += 4 + l;
682    }
683    (at == d.len() && out.len() == count).then_some(out)
684}
netget/server/nbd/mod.rs

netget/server/nbd/
mod.rs
