btrfs_stream/
stream.rs

1//! Binary send stream parser for the btrfs send/receive protocol.
2//!
3//! Parses the TLV-encoded command stream produced by `btrfs send` (or the
4//! kernel's `BTRFS_IOC_SEND` ioctl). The stream consists of a header followed
5//! by a sequence of commands, each with a CRC32C checksum and a set of
6//! typed TLV attributes.
7
8use crate::consts::{
9    BTRFS_SEND_A_ATIME, BTRFS_SEND_A_CLONE_CTRANSID, BTRFS_SEND_A_CLONE_LEN,
10    BTRFS_SEND_A_CLONE_OFFSET, BTRFS_SEND_A_CLONE_PATH,
11    BTRFS_SEND_A_CLONE_UUID, BTRFS_SEND_A_COMPRESSION, BTRFS_SEND_A_CTIME,
12    BTRFS_SEND_A_CTRANSID, BTRFS_SEND_A_DATA, BTRFS_SEND_A_ENCRYPTION,
13    BTRFS_SEND_A_FALLOCATE_MODE, BTRFS_SEND_A_FILE_OFFSET,
14    BTRFS_SEND_A_FILEATTR, BTRFS_SEND_A_GID, BTRFS_SEND_A_MODE,
15    BTRFS_SEND_A_MTIME, BTRFS_SEND_A_PATH, BTRFS_SEND_A_PATH_LINK,
16    BTRFS_SEND_A_PATH_TO, BTRFS_SEND_A_RDEV, BTRFS_SEND_A_SIZE,
17    BTRFS_SEND_A_UID, BTRFS_SEND_A_UNENCODED_FILE_LEN,
18    BTRFS_SEND_A_UNENCODED_LEN, BTRFS_SEND_A_UNENCODED_OFFSET,
19    BTRFS_SEND_A_UUID, BTRFS_SEND_A_VERITY_ALGORITHM,
20    BTRFS_SEND_A_VERITY_BLOCK_SIZE, BTRFS_SEND_A_VERITY_SALT_DATA,
21    BTRFS_SEND_A_VERITY_SIG_DATA, BTRFS_SEND_A_XATTR_DATA,
22    BTRFS_SEND_A_XATTR_NAME, BTRFS_SEND_C_CHMOD, BTRFS_SEND_C_CHOWN,
23    BTRFS_SEND_C_CLONE, BTRFS_SEND_C_ENABLE_VERITY, BTRFS_SEND_C_ENCODED_WRITE,
24    BTRFS_SEND_C_END, BTRFS_SEND_C_FALLOCATE, BTRFS_SEND_C_FILEATTR,
25    BTRFS_SEND_C_LINK, BTRFS_SEND_C_MKDIR, BTRFS_SEND_C_MKFIFO,
26    BTRFS_SEND_C_MKFILE, BTRFS_SEND_C_MKNOD, BTRFS_SEND_C_MKSOCK,
27    BTRFS_SEND_C_REMOVE_XATTR, BTRFS_SEND_C_RENAME, BTRFS_SEND_C_RMDIR,
28    BTRFS_SEND_C_SET_XATTR, BTRFS_SEND_C_SNAPSHOT, BTRFS_SEND_C_SUBVOL,
29    BTRFS_SEND_C_SYMLINK, BTRFS_SEND_C_TRUNCATE, BTRFS_SEND_C_UNLINK,
30    BTRFS_SEND_C_UPDATE_EXTENT, BTRFS_SEND_C_UTIMES, BTRFS_SEND_C_WRITE,
31    CMD_HEADER_LEN, SEND_STREAM_MAGIC, SEND_STREAM_MAGIC_LEN,
32    STREAM_HEADER_LEN,
33};
34use std::io::Read;
35use uuid::Uuid;
36
37/// Errors that can occur while parsing a btrfs send stream.
38#[derive(Debug, thiserror::Error)]
39pub enum StreamError {
40    /// The stream header does not start with the btrfs send magic.
41    #[error("invalid send stream: bad magic")]
42    BadMagic,
43    /// The stream version is not supported (must be 1-3).
44    #[error("unsupported send stream version {0} (supported: 1-3)")]
45    UnsupportedVersion(u32),
46    /// CRC32C checksum mismatch on a command.
47    #[error(
48        "CRC mismatch for command {cmd}: expected {expected:#010x}, got {computed:#010x}"
49    )]
50    CrcMismatch {
51        cmd: u16,
52        expected: u32,
53        computed: u32,
54    },
55    /// The stream header could not be read completely.
56    #[error("truncated send stream header: {0}")]
57    TruncatedHeader(std::io::Error),
58    /// A command payload could not be read completely.
59    #[error("truncated send stream payload: {0}")]
60    TruncatedPayload(std::io::Error),
61    /// A TLV attribute structure is incomplete.
62    #[error("truncated TLV: {detail}")]
63    TruncatedTlv { detail: String },
64    /// A TLV attribute type is out of range.
65    #[error("invalid TLV attribute type {0}")]
66    InvalidTlvType(u16),
67    /// A required TLV attribute is missing from a command.
68    #[error("missing required attribute: {0}")]
69    MissingAttribute(&'static str),
70    /// A TLV attribute value is malformed (wrong size, bad encoding, etc.).
71    #[error("attribute {name}: {detail}")]
72    InvalidAttribute { name: &'static str, detail: String },
73    /// An unknown command type was encountered.
74    #[error("unknown send stream command type {0}")]
75    UnknownCommand(u16),
76    /// An underlying I/O error.
77    #[error(transparent)]
78    Io(#[from] std::io::Error),
79}
80
81type Result<T> = std::result::Result<T, StreamError>;
82
83// Maximum number of TLV attribute types we track. (One past the
84// largest `BTRFS_SEND_A_*` constant; updates needed if v4 attrs land.)
85const MAX_ATTRS: usize = 36;
86
87/// A timestamp from the send stream (sec + nsec).
88#[derive(Debug, Clone, Copy)]
89pub struct Timespec {
90    pub sec: u64,
91    pub nsec: u32,
92}
93
94/// A parsed command from the send stream.
95#[derive(Debug)]
96pub enum StreamCommand {
97    Subvol {
98        path: String,
99        uuid: Uuid,
100        ctransid: u64,
101    },
102    Snapshot {
103        path: String,
104        uuid: Uuid,
105        ctransid: u64,
106        clone_uuid: Uuid,
107        clone_ctransid: u64,
108    },
109    Mkfile {
110        path: String,
111    },
112    Mkdir {
113        path: String,
114    },
115    Mknod {
116        path: String,
117        mode: u64,
118        rdev: u64,
119    },
120    Mkfifo {
121        path: String,
122    },
123    Mksock {
124        path: String,
125    },
126    Symlink {
127        path: String,
128        target: String,
129    },
130    Rename {
131        from: String,
132        to: String,
133    },
134    Link {
135        path: String,
136        target: String,
137    },
138    Unlink {
139        path: String,
140    },
141    Rmdir {
142        path: String,
143    },
144    Write {
145        path: String,
146        offset: u64,
147        data: Vec<u8>,
148    },
149    Clone {
150        path: String,
151        offset: u64,
152        len: u64,
153        clone_uuid: Uuid,
154        clone_ctransid: u64,
155        clone_path: String,
156        clone_offset: u64,
157    },
158    SetXattr {
159        path: String,
160        name: String,
161        data: Vec<u8>,
162    },
163    RemoveXattr {
164        path: String,
165        name: String,
166    },
167    Truncate {
168        path: String,
169        size: u64,
170    },
171    Chmod {
172        path: String,
173        mode: u64,
174    },
175    Chown {
176        path: String,
177        uid: u64,
178        gid: u64,
179    },
180    Utimes {
181        path: String,
182        atime: Timespec,
183        mtime: Timespec,
184        ctime: Timespec,
185    },
186    UpdateExtent {
187        path: String,
188        offset: u64,
189        len: u64,
190    },
191    /// v2: write pre-compressed data that can be passed through to the
192    /// filesystem without decompression via `BTRFS_IOC_ENCODED_WRITE`.
193    EncodedWrite {
194        path: String,
195        offset: u64,
196        /// Unencoded (decompressed) file length to write.
197        unencoded_file_len: u64,
198        /// Total unencoded length (may be larger due to sector alignment).
199        unencoded_len: u64,
200        /// Offset within the unencoded data where the file data starts.
201        unencoded_offset: u64,
202        /// Compression algorithm (0=none, 1=zlib, 2=zstd, 3-7=lzo with varying sector sizes).
203        compression: u32,
204        /// Encryption algorithm (currently always 0).
205        encryption: u32,
206        data: Vec<u8>,
207    },
208    /// v2: preallocate space or punch a hole.
209    Fallocate {
210        path: String,
211        /// `FALLOC_FL_*` flags (0=allocate, `1=KEEP_SIZE`, `3=PUNCH_HOLE|KEEP_SIZE`).
212        mode: u32,
213        offset: u64,
214        len: u64,
215    },
216    /// v2: set inode file attributes (chattr flags).
217    Fileattr {
218        path: String,
219        attr: u64,
220    },
221    /// v3: enable fs-verity on a file.
222    EnableVerity {
223        path: String,
224        algorithm: u8,
225        block_size: u32,
226        salt: Vec<u8>,
227        sig: Vec<u8>,
228    },
229    End,
230}
231
232/// Reads and parses a btrfs send stream.
233#[derive(Debug)]
234pub struct StreamReader<R: Read> {
235    reader: R,
236    version: u32,
237    buf: Vec<u8>,
238}
239
240impl<R: Read> StreamReader<R> {
241    /// Read and validate the stream header, returning a new reader.
242    ///
243    /// # Errors
244    ///
245    /// Returns an error if the header is truncated, the magic is invalid,
246    /// or the stream version is unsupported.
247    ///
248    /// # Panics
249    ///
250    /// Panics are unreachable: all `unwrap()` calls are on fixed-size
251    /// slice-to-array conversions that always succeed.
252    pub fn new(mut reader: R) -> Result<Self> {
253        let mut header = [0u8; STREAM_HEADER_LEN];
254        reader
255            .read_exact(&mut header)
256            .map_err(StreamError::TruncatedHeader)?;
257
258        if &header[..SEND_STREAM_MAGIC_LEN] != SEND_STREAM_MAGIC {
259            return Err(StreamError::BadMagic);
260        }
261
262        let version = u32::from_le_bytes(
263            header[SEND_STREAM_MAGIC_LEN..STREAM_HEADER_LEN]
264                .try_into()
265                .unwrap(),
266        );
267
268        if version == 0 || version > 3 {
269            return Err(StreamError::UnsupportedVersion(version));
270        }
271
272        Ok(Self {
273            reader,
274            version,
275            buf: Vec::with_capacity(64 * 1024),
276        })
277    }
278
279    /// Return the stream protocol version.
280    #[must_use]
281    pub fn version(&self) -> u32 {
282        self.version
283    }
284
285    /// Consume the underlying reader back out.
286    #[must_use]
287    pub fn into_inner(self) -> R {
288        self.reader
289    }
290
291    /// Read the next command from the stream.
292    ///
293    /// Returns `Ok(None)` on clean EOF (no more data), `Ok(Some(End))` when
294    /// the stream contains an explicit end-of-stream marker.
295    ///
296    /// # Errors
297    ///
298    /// Returns an error on I/O failure, CRC mismatch, malformed TLV
299    /// attributes, or unknown command types.
300    ///
301    /// # Panics
302    ///
303    /// Panics are unreachable: all `unwrap()` calls are on fixed-size
304    /// slice-to-array conversions that always succeed.
305    #[allow(clippy::too_many_lines)]
306    pub fn next_command(&mut self) -> Result<Option<StreamCommand>> {
307        // Read command header.
308        let mut cmd_hdr = [0u8; CMD_HEADER_LEN];
309        if !read_exact_or_eof(&mut self.reader, &mut cmd_hdr)? {
310            return Ok(None); // clean EOF
311        }
312
313        #[allow(clippy::cast_possible_truncation)]
314        // payload length fits in usize
315        let payload_len =
316            u32::from_le_bytes(cmd_hdr[0..4].try_into().unwrap()) as usize;
317        let cmd = u16::from_le_bytes(cmd_hdr[4..6].try_into().unwrap());
318        let expected_crc =
319            u32::from_le_bytes(cmd_hdr[6..10].try_into().unwrap());
320
321        // Read payload.
322        self.buf.resize(payload_len, 0);
323        self.reader
324            .read_exact(&mut self.buf)
325            .map_err(StreamError::TruncatedPayload)?;
326
327        // Validate CRC32C over header (with crc field zeroed) + payload.
328        // The btrfs send stream uses raw CRC-32C (init=0, xorout=0), not
329        // the standard ISO 3309 convention. We compute incrementally to
330        // avoid allocating a contiguous buffer for the entire command.
331        // raw_crc32c(seed, data) == !crc32c_append(!seed, data)
332        let crc = crc32c::crc32c_append(!0, &cmd_hdr[0..6]); // len + cmd
333        let crc = crc32c::crc32c_append(crc, &[0u8; 4]); // zeroed crc field
334        let crc = crc32c::crc32c_append(crc, &self.buf); // payload
335        let computed_crc = !crc;
336        if computed_crc != expected_crc {
337            return Err(StreamError::CrcMismatch {
338                cmd,
339                expected: expected_crc,
340                computed: computed_crc,
341            });
342        }
343
344        // Parse TLV attributes from payload.
345        let attrs = parse_tlv_attrs(&self.buf, self.version)?;
346
347        // Dispatch by command type.
348        match cmd {
349            BTRFS_SEND_C_SUBVOL => Ok(Some(StreamCommand::Subvol {
350                path: attr_string(
351                    &self.buf,
352                    &attrs,
353                    BTRFS_SEND_A_PATH,
354                    "path",
355                )?,
356                uuid: attr_uuid(&self.buf, &attrs, BTRFS_SEND_A_UUID, "uuid")?,
357                ctransid: attr_u64(
358                    &self.buf,
359                    &attrs,
360                    BTRFS_SEND_A_CTRANSID,
361                    "ctransid",
362                )?,
363            })),
364            BTRFS_SEND_C_SNAPSHOT => Ok(Some(StreamCommand::Snapshot {
365                path: attr_string(
366                    &self.buf,
367                    &attrs,
368                    BTRFS_SEND_A_PATH,
369                    "path",
370                )?,
371                uuid: attr_uuid(&self.buf, &attrs, BTRFS_SEND_A_UUID, "uuid")?,
372                ctransid: attr_u64(
373                    &self.buf,
374                    &attrs,
375                    BTRFS_SEND_A_CTRANSID,
376                    "ctransid",
377                )?,
378                clone_uuid: attr_uuid(
379                    &self.buf,
380                    &attrs,
381                    BTRFS_SEND_A_CLONE_UUID,
382                    "clone_uuid",
383                )?,
384                clone_ctransid: attr_u64(
385                    &self.buf,
386                    &attrs,
387                    BTRFS_SEND_A_CLONE_CTRANSID,
388                    "clone_ctransid",
389                )?,
390            })),
391            BTRFS_SEND_C_MKFILE => Ok(Some(StreamCommand::Mkfile {
392                path: attr_string(
393                    &self.buf,
394                    &attrs,
395                    BTRFS_SEND_A_PATH,
396                    "path",
397                )?,
398            })),
399            BTRFS_SEND_C_MKDIR => Ok(Some(StreamCommand::Mkdir {
400                path: attr_string(
401                    &self.buf,
402                    &attrs,
403                    BTRFS_SEND_A_PATH,
404                    "path",
405                )?,
406            })),
407            BTRFS_SEND_C_MKNOD => Ok(Some(StreamCommand::Mknod {
408                path: attr_string(
409                    &self.buf,
410                    &attrs,
411                    BTRFS_SEND_A_PATH,
412                    "path",
413                )?,
414                mode: attr_u64(&self.buf, &attrs, BTRFS_SEND_A_MODE, "mode")?,
415                rdev: attr_u64(&self.buf, &attrs, BTRFS_SEND_A_RDEV, "rdev")?,
416            })),
417            BTRFS_SEND_C_MKFIFO => Ok(Some(StreamCommand::Mkfifo {
418                path: attr_string(
419                    &self.buf,
420                    &attrs,
421                    BTRFS_SEND_A_PATH,
422                    "path",
423                )?,
424            })),
425            BTRFS_SEND_C_MKSOCK => Ok(Some(StreamCommand::Mksock {
426                path: attr_string(
427                    &self.buf,
428                    &attrs,
429                    BTRFS_SEND_A_PATH,
430                    "path",
431                )?,
432            })),
433            BTRFS_SEND_C_SYMLINK => Ok(Some(StreamCommand::Symlink {
434                path: attr_string(
435                    &self.buf,
436                    &attrs,
437                    BTRFS_SEND_A_PATH,
438                    "path",
439                )?,
440                target: attr_string(
441                    &self.buf,
442                    &attrs,
443                    BTRFS_SEND_A_PATH_LINK,
444                    "link_target",
445                )?,
446            })),
447            BTRFS_SEND_C_RENAME => Ok(Some(StreamCommand::Rename {
448                from: attr_string(
449                    &self.buf,
450                    &attrs,
451                    BTRFS_SEND_A_PATH,
452                    "path",
453                )?,
454                to: attr_string(
455                    &self.buf,
456                    &attrs,
457                    BTRFS_SEND_A_PATH_TO,
458                    "path_to",
459                )?,
460            })),
461            BTRFS_SEND_C_LINK => Ok(Some(StreamCommand::Link {
462                path: attr_string(
463                    &self.buf,
464                    &attrs,
465                    BTRFS_SEND_A_PATH,
466                    "path",
467                )?,
468                target: attr_string(
469                    &self.buf,
470                    &attrs,
471                    BTRFS_SEND_A_PATH_LINK,
472                    "link_target",
473                )?,
474            })),
475            BTRFS_SEND_C_UNLINK => Ok(Some(StreamCommand::Unlink {
476                path: attr_string(
477                    &self.buf,
478                    &attrs,
479                    BTRFS_SEND_A_PATH,
480                    "path",
481                )?,
482            })),
483            BTRFS_SEND_C_RMDIR => Ok(Some(StreamCommand::Rmdir {
484                path: attr_string(
485                    &self.buf,
486                    &attrs,
487                    BTRFS_SEND_A_PATH,
488                    "path",
489                )?,
490            })),
491            BTRFS_SEND_C_SET_XATTR => Ok(Some(StreamCommand::SetXattr {
492                path: attr_string(
493                    &self.buf,
494                    &attrs,
495                    BTRFS_SEND_A_PATH,
496                    "path",
497                )?,
498                name: attr_string(
499                    &self.buf,
500                    &attrs,
501                    BTRFS_SEND_A_XATTR_NAME,
502                    "xattr_name",
503                )?,
504                data: attr_data(
505                    &self.buf,
506                    &attrs,
507                    BTRFS_SEND_A_XATTR_DATA,
508                    "xattr_data",
509                )?,
510            })),
511            BTRFS_SEND_C_REMOVE_XATTR => Ok(Some(StreamCommand::RemoveXattr {
512                path: attr_string(
513                    &self.buf,
514                    &attrs,
515                    BTRFS_SEND_A_PATH,
516                    "path",
517                )?,
518                name: attr_string(
519                    &self.buf,
520                    &attrs,
521                    BTRFS_SEND_A_XATTR_NAME,
522                    "xattr_name",
523                )?,
524            })),
525            BTRFS_SEND_C_WRITE => Ok(Some(StreamCommand::Write {
526                path: attr_string(
527                    &self.buf,
528                    &attrs,
529                    BTRFS_SEND_A_PATH,
530                    "path",
531                )?,
532                offset: attr_u64(
533                    &self.buf,
534                    &attrs,
535                    BTRFS_SEND_A_FILE_OFFSET,
536                    "file_offset",
537                )?,
538                data: attr_data(&self.buf, &attrs, BTRFS_SEND_A_DATA, "data")?,
539            })),
540            BTRFS_SEND_C_CLONE => Ok(Some(StreamCommand::Clone {
541                path: attr_string(
542                    &self.buf,
543                    &attrs,
544                    BTRFS_SEND_A_PATH,
545                    "path",
546                )?,
547                offset: attr_u64(
548                    &self.buf,
549                    &attrs,
550                    BTRFS_SEND_A_FILE_OFFSET,
551                    "file_offset",
552                )?,
553                len: attr_u64(
554                    &self.buf,
555                    &attrs,
556                    BTRFS_SEND_A_CLONE_LEN,
557                    "clone_len",
558                )?,
559                clone_uuid: attr_uuid(
560                    &self.buf,
561                    &attrs,
562                    BTRFS_SEND_A_CLONE_UUID,
563                    "clone_uuid",
564                )?,
565                clone_ctransid: attr_u64(
566                    &self.buf,
567                    &attrs,
568                    BTRFS_SEND_A_CLONE_CTRANSID,
569                    "clone_ctransid",
570                )?,
571                clone_path: attr_string(
572                    &self.buf,
573                    &attrs,
574                    BTRFS_SEND_A_CLONE_PATH,
575                    "clone_path",
576                )?,
577                clone_offset: attr_u64(
578                    &self.buf,
579                    &attrs,
580                    BTRFS_SEND_A_CLONE_OFFSET,
581                    "clone_offset",
582                )?,
583            })),
584            BTRFS_SEND_C_TRUNCATE => Ok(Some(StreamCommand::Truncate {
585                path: attr_string(
586                    &self.buf,
587                    &attrs,
588                    BTRFS_SEND_A_PATH,
589                    "path",
590                )?,
591                size: attr_u64(&self.buf, &attrs, BTRFS_SEND_A_SIZE, "size")?,
592            })),
593            BTRFS_SEND_C_CHMOD => Ok(Some(StreamCommand::Chmod {
594                path: attr_string(
595                    &self.buf,
596                    &attrs,
597                    BTRFS_SEND_A_PATH,
598                    "path",
599                )?,
600                mode: attr_u64(&self.buf, &attrs, BTRFS_SEND_A_MODE, "mode")?,
601            })),
602            BTRFS_SEND_C_CHOWN => Ok(Some(StreamCommand::Chown {
603                path: attr_string(
604                    &self.buf,
605                    &attrs,
606                    BTRFS_SEND_A_PATH,
607                    "path",
608                )?,
609                uid: attr_u64(&self.buf, &attrs, BTRFS_SEND_A_UID, "uid")?,
610                gid: attr_u64(&self.buf, &attrs, BTRFS_SEND_A_GID, "gid")?,
611            })),
612            BTRFS_SEND_C_UTIMES => Ok(Some(StreamCommand::Utimes {
613                path: attr_string(
614                    &self.buf,
615                    &attrs,
616                    BTRFS_SEND_A_PATH,
617                    "path",
618                )?,
619                atime: attr_timespec(
620                    &self.buf,
621                    &attrs,
622                    BTRFS_SEND_A_ATIME,
623                    "atime",
624                )?,
625                mtime: attr_timespec(
626                    &self.buf,
627                    &attrs,
628                    BTRFS_SEND_A_MTIME,
629                    "mtime",
630                )?,
631                ctime: attr_timespec(
632                    &self.buf,
633                    &attrs,
634                    BTRFS_SEND_A_CTIME,
635                    "ctime",
636                )?,
637            })),
638            BTRFS_SEND_C_UPDATE_EXTENT => {
639                Ok(Some(StreamCommand::UpdateExtent {
640                    path: attr_string(
641                        &self.buf,
642                        &attrs,
643                        BTRFS_SEND_A_PATH,
644                        "path",
645                    )?,
646                    offset: attr_u64(
647                        &self.buf,
648                        &attrs,
649                        BTRFS_SEND_A_FILE_OFFSET,
650                        "file_offset",
651                    )?,
652                    len: attr_u64(
653                        &self.buf,
654                        &attrs,
655                        BTRFS_SEND_A_SIZE,
656                        "size",
657                    )?,
658                }))
659            }
660            BTRFS_SEND_C_FALLOCATE => Ok(Some(StreamCommand::Fallocate {
661                path: attr_string(
662                    &self.buf,
663                    &attrs,
664                    BTRFS_SEND_A_PATH,
665                    "path",
666                )?,
667                mode: attr_u32(
668                    &self.buf,
669                    &attrs,
670                    BTRFS_SEND_A_FALLOCATE_MODE,
671                    "fallocate_mode",
672                )?,
673                offset: attr_u64(
674                    &self.buf,
675                    &attrs,
676                    BTRFS_SEND_A_FILE_OFFSET,
677                    "file_offset",
678                )?,
679                len: attr_u64(&self.buf, &attrs, BTRFS_SEND_A_SIZE, "size")?,
680            })),
681            BTRFS_SEND_C_FILEATTR => Ok(Some(StreamCommand::Fileattr {
682                path: attr_string(
683                    &self.buf,
684                    &attrs,
685                    BTRFS_SEND_A_PATH,
686                    "path",
687                )?,
688                attr: attr_u64(
689                    &self.buf,
690                    &attrs,
691                    BTRFS_SEND_A_FILEATTR,
692                    "fileattr",
693                )?,
694            })),
695            BTRFS_SEND_C_ENCODED_WRITE => {
696                Ok(Some(StreamCommand::EncodedWrite {
697                    path: attr_string(
698                        &self.buf,
699                        &attrs,
700                        BTRFS_SEND_A_PATH,
701                        "path",
702                    )?,
703                    offset: attr_u64(
704                        &self.buf,
705                        &attrs,
706                        BTRFS_SEND_A_FILE_OFFSET,
707                        "file_offset",
708                    )?,
709                    unencoded_file_len: attr_u64(
710                        &self.buf,
711                        &attrs,
712                        BTRFS_SEND_A_UNENCODED_FILE_LEN,
713                        "unencoded_file_len",
714                    )?,
715                    unencoded_len: attr_u64(
716                        &self.buf,
717                        &attrs,
718                        BTRFS_SEND_A_UNENCODED_LEN,
719                        "unencoded_len",
720                    )?,
721                    unencoded_offset: attr_u64(
722                        &self.buf,
723                        &attrs,
724                        BTRFS_SEND_A_UNENCODED_OFFSET,
725                        "unencoded_offset",
726                    )?,
727                    // Compression and encryption default to 0 if absent.
728                    compression: attr_opt_u32(
729                        &self.buf,
730                        &attrs,
731                        BTRFS_SEND_A_COMPRESSION,
732                        0,
733                    ),
734                    encryption: attr_opt_u32(
735                        &self.buf,
736                        &attrs,
737                        BTRFS_SEND_A_ENCRYPTION,
738                        0,
739                    ),
740                    data: attr_data(
741                        &self.buf,
742                        &attrs,
743                        BTRFS_SEND_A_DATA,
744                        "data",
745                    )?,
746                }))
747            }
748            BTRFS_SEND_C_ENABLE_VERITY => {
749                Ok(Some(StreamCommand::EnableVerity {
750                    path: attr_string(
751                        &self.buf,
752                        &attrs,
753                        BTRFS_SEND_A_PATH,
754                        "path",
755                    )?,
756                    algorithm: attr_u8(
757                        &self.buf,
758                        &attrs,
759                        BTRFS_SEND_A_VERITY_ALGORITHM,
760                        "verity_algorithm",
761                    )?,
762                    block_size: attr_u32(
763                        &self.buf,
764                        &attrs,
765                        BTRFS_SEND_A_VERITY_BLOCK_SIZE,
766                        "verity_block_size",
767                    )?,
768                    salt: attr_data(
769                        &self.buf,
770                        &attrs,
771                        BTRFS_SEND_A_VERITY_SALT_DATA,
772                        "verity_salt",
773                    )?,
774                    sig: attr_data(
775                        &self.buf,
776                        &attrs,
777                        BTRFS_SEND_A_VERITY_SIG_DATA,
778                        "verity_sig",
779                    )?,
780                }))
781            }
782            BTRFS_SEND_C_END => Ok(Some(StreamCommand::End)),
783            _ => Err(StreamError::UnknownCommand(cmd)),
784        }
785    }
786}
787
788/// Read exactly `buf.len()` bytes, returning false on clean EOF (zero bytes read).
789fn read_exact_or_eof(reader: &mut impl Read, buf: &mut [u8]) -> Result<bool> {
790    let mut pos = 0;
791    while pos < buf.len() {
792        match reader.read(&mut buf[pos..]) {
793            Ok(0) => {
794                if pos == 0 {
795                    return Ok(false);
796                }
797                return Err(StreamError::TruncatedPayload(
798                    std::io::Error::new(
799                        std::io::ErrorKind::UnexpectedEof,
800                        format!("unexpected EOF after {pos} bytes"),
801                    ),
802                ));
803            }
804            Ok(n) => pos += n,
805            Err(e) => return Err(StreamError::TruncatedPayload(e)),
806        }
807    }
808    Ok(true)
809}
810
811/// Attribute storage: for each attribute type index, Option<(offset, len)> into
812/// the payload buffer.
813type AttrTable = [Option<(usize, usize)>; MAX_ATTRS];
814
815/// Parse TLV attributes from a command payload buffer into a lookup table.
816fn parse_tlv_attrs(payload: &[u8], version: u32) -> Result<AttrTable> {
817    let mut attrs: AttrTable = [None; MAX_ATTRS];
818    let mut pos = 0;
819
820    while pos < payload.len() {
821        if pos + 2 > payload.len() {
822            return Err(StreamError::TruncatedTlv {
823                detail: "not enough bytes for type field".into(),
824            });
825        }
826        let tlv_type =
827            u16::from_le_bytes(payload[pos..pos + 2].try_into().unwrap());
828        pos += 2;
829
830        if tlv_type == 0 || usize::from(tlv_type) > MAX_ATTRS {
831            return Err(StreamError::InvalidTlvType(tlv_type));
832        }
833
834        // In v2+, the DATA attribute has no length field — it extends to the
835        // end of the command payload.
836        let tlv_len = if version >= 2 && tlv_type == BTRFS_SEND_A_DATA {
837            payload.len() - pos
838        } else {
839            if pos + 2 > payload.len() {
840                return Err(StreamError::TruncatedTlv {
841                    detail: "not enough bytes for length field".into(),
842                });
843            }
844            let len = usize::from(u16::from_le_bytes(
845                payload[pos..pos + 2].try_into().unwrap(),
846            ));
847            pos += 2;
848            len
849        };
850
851        if pos + tlv_len > payload.len() {
852            return Err(StreamError::TruncatedTlv {
853                detail: format!(
854                    "attribute type {tlv_type} needs {tlv_len} bytes but only {} remain",
855                    payload.len() - pos
856                ),
857            });
858        }
859
860        attrs[usize::from(tlv_type - 1)] = Some((pos, tlv_len));
861        pos += tlv_len;
862    }
863
864    Ok(attrs)
865}
866
867fn get_attr<'a>(
868    buf: &'a [u8],
869    attrs: &AttrTable,
870    attr_type: u16,
871    name: &'static str,
872) -> Result<&'a [u8]> {
873    let (offset, len) = attrs[usize::from(attr_type - 1)]
874        .ok_or(StreamError::MissingAttribute(name))?;
875    Ok(&buf[offset..offset + len])
876}
877
878fn attr_u64(
879    buf: &[u8],
880    attrs: &AttrTable,
881    attr_type: u16,
882    name: &'static str,
883) -> Result<u64> {
884    let data = get_attr(buf, attrs, attr_type, name)?;
885    if data.len() < 8 {
886        return Err(StreamError::InvalidAttribute {
887            name,
888            detail: format!("too short for u64: {} bytes", data.len()),
889        });
890    }
891    Ok(u64::from_le_bytes(data[0..8].try_into().unwrap()))
892}
893
894fn attr_string(
895    buf: &[u8],
896    attrs: &AttrTable,
897    attr_type: u16,
898    name: &'static str,
899) -> Result<String> {
900    let data = get_attr(buf, attrs, attr_type, name)?;
901    // Strings in the stream are null-terminated; strip the trailing NUL.
902    let s = if data.last() == Some(&0) {
903        &data[..data.len() - 1]
904    } else {
905        data
906    };
907    String::from_utf8(s.to_vec()).map_err(|_| StreamError::InvalidAttribute {
908        name,
909        detail: "not valid UTF-8".into(),
910    })
911}
912
913fn attr_uuid(
914    buf: &[u8],
915    attrs: &AttrTable,
916    attr_type: u16,
917    name: &'static str,
918) -> Result<Uuid> {
919    let data = get_attr(buf, attrs, attr_type, name)?;
920    if data.len() < 16 {
921        return Err(StreamError::InvalidAttribute {
922            name,
923            detail: format!("too short for UUID: {} bytes", data.len()),
924        });
925    }
926    Ok(Uuid::from_bytes(data[0..16].try_into().unwrap()))
927}
928
929fn attr_timespec(
930    buf: &[u8],
931    attrs: &AttrTable,
932    attr_type: u16,
933    name: &'static str,
934) -> Result<Timespec> {
935    let data = get_attr(buf, attrs, attr_type, name)?;
936    if data.len() < 12 {
937        return Err(StreamError::InvalidAttribute {
938            name,
939            detail: format!("too short for timespec: {} bytes", data.len()),
940        });
941    }
942    Ok(Timespec {
943        sec: u64::from_le_bytes(data[0..8].try_into().unwrap()),
944        nsec: u32::from_le_bytes(data[8..12].try_into().unwrap()),
945    })
946}
947
948fn attr_u32(
949    buf: &[u8],
950    attrs: &AttrTable,
951    attr_type: u16,
952    name: &'static str,
953) -> Result<u32> {
954    let data = get_attr(buf, attrs, attr_type, name)?;
955    if data.len() < 4 {
956        return Err(StreamError::InvalidAttribute {
957            name,
958            detail: format!("too short for u32: {} bytes", data.len()),
959        });
960    }
961    Ok(u32::from_le_bytes(data[0..4].try_into().unwrap()))
962}
963
964fn attr_u8(
965    buf: &[u8],
966    attrs: &AttrTable,
967    attr_type: u16,
968    name: &'static str,
969) -> Result<u8> {
970    let data = get_attr(buf, attrs, attr_type, name)?;
971    if data.is_empty() {
972        return Err(StreamError::InvalidAttribute {
973            name,
974            detail: "is empty".into(),
975        });
976    }
977    Ok(data[0])
978}
979
980/// Like `get_attr` but returns `None` instead of an error when the attribute
981/// is absent. Used for optional attributes in v2 commands.
982fn get_attr_opt<'a>(
983    buf: &'a [u8],
984    attrs: &AttrTable,
985    attr_type: u16,
986) -> Option<&'a [u8]> {
987    let (offset, len) = attrs[usize::from(attr_type - 1)]?;
988    Some(&buf[offset..offset + len])
989}
990
991fn attr_opt_u32(
992    buf: &[u8],
993    attrs: &AttrTable,
994    attr_type: u16,
995    default: u32,
996) -> u32 {
997    match get_attr_opt(buf, attrs, attr_type) {
998        Some(data) if data.len() >= 4 => {
999            u32::from_le_bytes(data[0..4].try_into().unwrap())
1000        }
1001        _ => default,
1002    }
1003}
1004
1005fn attr_data(
1006    buf: &[u8],
1007    attrs: &AttrTable,
1008    attr_type: u16,
1009    name: &'static str,
1010) -> Result<Vec<u8>> {
1011    let data = get_attr(buf, attrs, attr_type, name)?;
1012    Ok(data.to_vec())
1013}
1014
1015#[cfg(test)]
1016mod tests {
1017    use super::*;
1018    use std::io::Cursor;
1019
1020    /// Build a valid v1 stream header.
1021    fn v1_header() -> Vec<u8> {
1022        let mut buf = Vec::new();
1023        buf.extend_from_slice(SEND_STREAM_MAGIC);
1024        buf.extend_from_slice(&1u32.to_le_bytes());
1025        buf
1026    }
1027
1028    /// Build a v1 command with proper CRC. `cmd` is the command type code,
1029    /// `payload` is the raw TLV payload bytes.
1030    fn build_command(cmd: u16, payload: &[u8]) -> Vec<u8> {
1031        let len = payload.len() as u32;
1032        // Build header with zeroed CRC for checksum computation.
1033        // Uses raw CRC-32C (init=0, xorout=0) matching the btrfs send format.
1034        let mut crc_input = Vec::new();
1035        crc_input.extend_from_slice(&len.to_le_bytes());
1036        crc_input.extend_from_slice(&cmd.to_le_bytes());
1037        crc_input.extend_from_slice(&[0u8; 4]); // zeroed crc
1038        crc_input.extend_from_slice(payload);
1039        let crc = !crc32c::crc32c_append(!0, &crc_input);
1040
1041        let mut out = Vec::new();
1042        out.extend_from_slice(&len.to_le_bytes());
1043        out.extend_from_slice(&cmd.to_le_bytes());
1044        out.extend_from_slice(&crc.to_le_bytes());
1045        out.extend_from_slice(payload);
1046        out
1047    }
1048
1049    /// Build a TLV attribute with a u16 type, u16 length, and raw data.
1050    fn tlv(attr_type: u16, data: &[u8]) -> Vec<u8> {
1051        let mut buf = Vec::new();
1052        buf.extend_from_slice(&attr_type.to_le_bytes());
1053        buf.extend_from_slice(&(data.len() as u16).to_le_bytes());
1054        buf.extend_from_slice(data);
1055        buf
1056    }
1057
1058    /// Build a TLV string attribute (null-terminated).
1059    fn tlv_string(attr_type: u16, s: &str) -> Vec<u8> {
1060        let mut data = s.as_bytes().to_vec();
1061        data.push(0); // null terminator
1062        tlv(attr_type, &data)
1063    }
1064
1065    /// Build a TLV u64 attribute (little-endian).
1066    fn tlv_u64(attr_type: u16, val: u64) -> Vec<u8> {
1067        tlv(attr_type, &val.to_le_bytes())
1068    }
1069
1070    /// Build a TLV UUID attribute.
1071    fn tlv_uuid(attr_type: u16, uuid: Uuid) -> Vec<u8> {
1072        tlv(attr_type, uuid.as_bytes())
1073    }
1074
1075    /// Build a TLV timespec attribute (8 bytes sec + 4 bytes nsec, LE).
1076    fn tlv_timespec(attr_type: u16, sec: u64, nsec: u32) -> Vec<u8> {
1077        let mut data = Vec::new();
1078        data.extend_from_slice(&sec.to_le_bytes());
1079        data.extend_from_slice(&nsec.to_le_bytes());
1080        tlv(attr_type, &data)
1081    }
1082
1083    // --- Header parsing ---
1084
1085    #[test]
1086    fn valid_v1_header() {
1087        let mut stream = v1_header();
1088        // Append an END command so we can actually read something.
1089        stream.extend_from_slice(&build_command(BTRFS_SEND_C_END, &[]));
1090        let reader = StreamReader::new(Cursor::new(stream)).unwrap();
1091        assert_eq!(reader.version(), 1);
1092    }
1093
1094    #[test]
1095    fn valid_v2_header() {
1096        let mut buf = Vec::new();
1097        buf.extend_from_slice(SEND_STREAM_MAGIC);
1098        buf.extend_from_slice(&2u32.to_le_bytes());
1099        buf.extend_from_slice(&build_command(BTRFS_SEND_C_END, &[]));
1100        let reader = StreamReader::new(Cursor::new(buf)).unwrap();
1101        assert_eq!(reader.version(), 2);
1102    }
1103
1104    #[test]
1105    fn bad_magic() {
1106        let buf = b"not-a-stream\0\x01\x00\x00\x00";
1107        let err = StreamReader::new(Cursor::new(buf.to_vec())).unwrap_err();
1108        assert!(
1109            format!("{err}").contains("bad magic"),
1110            "expected bad magic error, got: {err}"
1111        );
1112    }
1113
1114    #[test]
1115    fn unsupported_version_zero() {
1116        let mut buf = Vec::new();
1117        buf.extend_from_slice(SEND_STREAM_MAGIC);
1118        buf.extend_from_slice(&0u32.to_le_bytes());
1119        let err = StreamReader::new(Cursor::new(buf)).unwrap_err();
1120        assert!(format!("{err}").contains("unsupported"), "got: {err}");
1121    }
1122
1123    #[test]
1124    fn valid_v3_header() {
1125        let mut buf = Vec::new();
1126        buf.extend_from_slice(SEND_STREAM_MAGIC);
1127        buf.extend_from_slice(&3u32.to_le_bytes());
1128        buf.extend_from_slice(&build_command(BTRFS_SEND_C_END, &[]));
1129        let reader = StreamReader::new(Cursor::new(buf)).unwrap();
1130        assert_eq!(reader.version(), 3);
1131    }
1132
1133    #[test]
1134    fn unsupported_version_four() {
1135        let mut buf = Vec::new();
1136        buf.extend_from_slice(SEND_STREAM_MAGIC);
1137        buf.extend_from_slice(&4u32.to_le_bytes());
1138        let err = StreamReader::new(Cursor::new(buf)).unwrap_err();
1139        assert!(format!("{err}").contains("unsupported"), "got: {err}");
1140    }
1141
1142    #[test]
1143    fn truncated_header() {
1144        let buf = b"btrfs-str"; // too short
1145        let err = StreamReader::new(Cursor::new(buf.to_vec())).unwrap_err();
1146        assert!(matches!(err, StreamError::TruncatedHeader(_)), "got: {err}");
1147    }
1148
1149    // --- END command ---
1150
1151    #[test]
1152    fn end_command() {
1153        let mut stream = v1_header();
1154        stream.extend_from_slice(&build_command(BTRFS_SEND_C_END, &[]));
1155        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1156        let cmd = reader.next_command().unwrap().unwrap();
1157        assert!(matches!(cmd, StreamCommand::End));
1158    }
1159
1160    #[test]
1161    fn eof_after_end() {
1162        let mut stream = v1_header();
1163        stream.extend_from_slice(&build_command(BTRFS_SEND_C_END, &[]));
1164        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1165        reader.next_command().unwrap(); // END
1166        assert!(reader.next_command().unwrap().is_none()); // clean EOF
1167    }
1168
1169    // --- CRC validation ---
1170
1171    #[test]
1172    fn corrupted_crc() {
1173        let mut stream = v1_header();
1174        let mut cmd = build_command(BTRFS_SEND_C_END, &[]);
1175        cmd[8] ^= 0xFF; // flip a byte in the CRC field
1176        stream.extend_from_slice(&cmd);
1177        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1178        let err = reader.next_command().unwrap_err();
1179        assert!(format!("{err}").contains("CRC mismatch"), "got: {err}");
1180    }
1181
1182    #[test]
1183    fn corrupted_payload() {
1184        let mut payload = Vec::new();
1185        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "test"));
1186        let mut stream = v1_header();
1187        let mut cmd = build_command(BTRFS_SEND_C_MKFILE, &payload);
1188        // Corrupt a payload byte (after the 10-byte header).
1189        if cmd.len() > 11 {
1190            cmd[11] ^= 0xFF;
1191        }
1192        stream.extend_from_slice(&cmd);
1193        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1194        let err = reader.next_command().unwrap_err();
1195        assert!(format!("{err}").contains("CRC mismatch"), "got: {err}");
1196    }
1197
1198    // --- Simple commands ---
1199
1200    #[test]
1201    fn mkfile_command() {
1202        let payload = tlv_string(BTRFS_SEND_A_PATH, "hello.txt");
1203        let mut stream = v1_header();
1204        stream.extend_from_slice(&build_command(BTRFS_SEND_C_MKFILE, &payload));
1205        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1206        match reader.next_command().unwrap().unwrap() {
1207            StreamCommand::Mkfile { path } => assert_eq!(path, "hello.txt"),
1208            other => panic!("expected Mkfile, got {other:?}"),
1209        }
1210    }
1211
1212    #[test]
1213    fn mkdir_command() {
1214        let payload = tlv_string(BTRFS_SEND_A_PATH, "subdir");
1215        let mut stream = v1_header();
1216        stream.extend_from_slice(&build_command(BTRFS_SEND_C_MKDIR, &payload));
1217        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1218        match reader.next_command().unwrap().unwrap() {
1219            StreamCommand::Mkdir { path } => assert_eq!(path, "subdir"),
1220            other => panic!("expected Mkdir, got {other:?}"),
1221        }
1222    }
1223
1224    #[test]
1225    fn unlink_command() {
1226        let payload = tlv_string(BTRFS_SEND_A_PATH, "gone.txt");
1227        let mut stream = v1_header();
1228        stream.extend_from_slice(&build_command(BTRFS_SEND_C_UNLINK, &payload));
1229        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1230        match reader.next_command().unwrap().unwrap() {
1231            StreamCommand::Unlink { path } => assert_eq!(path, "gone.txt"),
1232            other => panic!("expected Unlink, got {other:?}"),
1233        }
1234    }
1235
1236    #[test]
1237    fn rmdir_command() {
1238        let payload = tlv_string(BTRFS_SEND_A_PATH, "old_dir");
1239        let mut stream = v1_header();
1240        stream.extend_from_slice(&build_command(BTRFS_SEND_C_RMDIR, &payload));
1241        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1242        match reader.next_command().unwrap().unwrap() {
1243            StreamCommand::Rmdir { path } => assert_eq!(path, "old_dir"),
1244            other => panic!("expected Rmdir, got {other:?}"),
1245        }
1246    }
1247
1248    // --- Commands with multiple attributes ---
1249
1250    #[test]
1251    fn rename_command() {
1252        let mut payload = Vec::new();
1253        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "old_name"));
1254        payload
1255            .extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH_TO, "new_name"));
1256        let mut stream = v1_header();
1257        stream.extend_from_slice(&build_command(BTRFS_SEND_C_RENAME, &payload));
1258        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1259        match reader.next_command().unwrap().unwrap() {
1260            StreamCommand::Rename { from, to } => {
1261                assert_eq!(from, "old_name");
1262                assert_eq!(to, "new_name");
1263            }
1264            other => panic!("expected Rename, got {other:?}"),
1265        }
1266    }
1267
1268    #[test]
1269    fn symlink_command() {
1270        let mut payload = Vec::new();
1271        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "link"));
1272        payload
1273            .extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH_LINK, "/target"));
1274        let mut stream = v1_header();
1275        stream
1276            .extend_from_slice(&build_command(BTRFS_SEND_C_SYMLINK, &payload));
1277        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1278        match reader.next_command().unwrap().unwrap() {
1279            StreamCommand::Symlink { path, target } => {
1280                assert_eq!(path, "link");
1281                assert_eq!(target, "/target");
1282            }
1283            other => panic!("expected Symlink, got {other:?}"),
1284        }
1285    }
1286
1287    #[test]
1288    fn chmod_command() {
1289        let mut payload = Vec::new();
1290        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "file.sh"));
1291        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_MODE, 0o755));
1292        let mut stream = v1_header();
1293        stream.extend_from_slice(&build_command(BTRFS_SEND_C_CHMOD, &payload));
1294        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1295        match reader.next_command().unwrap().unwrap() {
1296            StreamCommand::Chmod { path, mode } => {
1297                assert_eq!(path, "file.sh");
1298                assert_eq!(mode, 0o755);
1299            }
1300            other => panic!("expected Chmod, got {other:?}"),
1301        }
1302    }
1303
1304    #[test]
1305    fn chown_command() {
1306        let mut payload = Vec::new();
1307        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "owned"));
1308        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_UID, 1000));
1309        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_GID, 1000));
1310        let mut stream = v1_header();
1311        stream.extend_from_slice(&build_command(BTRFS_SEND_C_CHOWN, &payload));
1312        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1313        match reader.next_command().unwrap().unwrap() {
1314            StreamCommand::Chown { path, uid, gid } => {
1315                assert_eq!(path, "owned");
1316                assert_eq!(uid, 1000);
1317                assert_eq!(gid, 1000);
1318            }
1319            other => panic!("expected Chown, got {other:?}"),
1320        }
1321    }
1322
1323    #[test]
1324    fn truncate_command() {
1325        let mut payload = Vec::new();
1326        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "shrink.bin"));
1327        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_SIZE, 4096));
1328        let mut stream = v1_header();
1329        stream
1330            .extend_from_slice(&build_command(BTRFS_SEND_C_TRUNCATE, &payload));
1331        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1332        match reader.next_command().unwrap().unwrap() {
1333            StreamCommand::Truncate { path, size } => {
1334                assert_eq!(path, "shrink.bin");
1335                assert_eq!(size, 4096);
1336            }
1337            other => panic!("expected Truncate, got {other:?}"),
1338        }
1339    }
1340
1341    #[test]
1342    fn write_command() {
1343        let file_data = b"hello world";
1344        let mut payload = Vec::new();
1345        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "data.bin"));
1346        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_FILE_OFFSET, 512));
1347        payload.extend_from_slice(&tlv(BTRFS_SEND_A_DATA, file_data));
1348        let mut stream = v1_header();
1349        stream.extend_from_slice(&build_command(BTRFS_SEND_C_WRITE, &payload));
1350        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1351        match reader.next_command().unwrap().unwrap() {
1352            StreamCommand::Write { path, offset, data } => {
1353                assert_eq!(path, "data.bin");
1354                assert_eq!(offset, 512);
1355                assert_eq!(data, file_data);
1356            }
1357            other => panic!("expected Write, got {other:?}"),
1358        }
1359    }
1360
1361    #[test]
1362    fn set_xattr_command() {
1363        let xattr_value = b"\x01\x02\x03";
1364        let mut payload = Vec::new();
1365        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "file"));
1366        payload.extend_from_slice(&tlv_string(
1367            BTRFS_SEND_A_XATTR_NAME,
1368            "user.test",
1369        ));
1370        payload.extend_from_slice(&tlv(BTRFS_SEND_A_XATTR_DATA, xattr_value));
1371        let mut stream = v1_header();
1372        stream.extend_from_slice(&build_command(
1373            BTRFS_SEND_C_SET_XATTR,
1374            &payload,
1375        ));
1376        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1377        match reader.next_command().unwrap().unwrap() {
1378            StreamCommand::SetXattr { path, name, data } => {
1379                assert_eq!(path, "file");
1380                assert_eq!(name, "user.test");
1381                assert_eq!(data, xattr_value);
1382            }
1383            other => panic!("expected SetXattr, got {other:?}"),
1384        }
1385    }
1386
1387    #[test]
1388    fn utimes_command() {
1389        let mut payload = Vec::new();
1390        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "timed"));
1391        payload.extend_from_slice(&tlv_timespec(BTRFS_SEND_A_ATIME, 1000, 111));
1392        payload.extend_from_slice(&tlv_timespec(BTRFS_SEND_A_MTIME, 2000, 222));
1393        payload.extend_from_slice(&tlv_timespec(BTRFS_SEND_A_CTIME, 3000, 333));
1394        let mut stream = v1_header();
1395        stream.extend_from_slice(&build_command(BTRFS_SEND_C_UTIMES, &payload));
1396        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1397        match reader.next_command().unwrap().unwrap() {
1398            StreamCommand::Utimes {
1399                path,
1400                atime,
1401                mtime,
1402                ctime,
1403            } => {
1404                assert_eq!(path, "timed");
1405                assert_eq!(atime.sec, 1000);
1406                assert_eq!(atime.nsec, 111);
1407                assert_eq!(mtime.sec, 2000);
1408                assert_eq!(mtime.nsec, 222);
1409                assert_eq!(ctime.sec, 3000);
1410                assert_eq!(ctime.nsec, 333);
1411            }
1412            other => panic!("expected Utimes, got {other:?}"),
1413        }
1414    }
1415
1416    // --- Subvol and Snapshot ---
1417
1418    #[test]
1419    fn subvol_command() {
1420        let uuid =
1421            Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
1422        let mut payload = Vec::new();
1423        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "my_subvol"));
1424        payload.extend_from_slice(&tlv_uuid(BTRFS_SEND_A_UUID, uuid));
1425        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_CTRANSID, 42));
1426        let mut stream = v1_header();
1427        stream.extend_from_slice(&build_command(BTRFS_SEND_C_SUBVOL, &payload));
1428        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1429        match reader.next_command().unwrap().unwrap() {
1430            StreamCommand::Subvol {
1431                path,
1432                uuid: u,
1433                ctransid,
1434            } => {
1435                assert_eq!(path, "my_subvol");
1436                assert_eq!(u, uuid);
1437                assert_eq!(ctransid, 42);
1438            }
1439            other => panic!("expected Subvol, got {other:?}"),
1440        }
1441    }
1442
1443    #[test]
1444    fn snapshot_command() {
1445        let uuid =
1446            Uuid::parse_str("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee").unwrap();
1447        let clone_uuid =
1448            Uuid::parse_str("11111111-2222-3333-4444-555555555555").unwrap();
1449        let mut payload = Vec::new();
1450        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "my_snap"));
1451        payload.extend_from_slice(&tlv_uuid(BTRFS_SEND_A_UUID, uuid));
1452        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_CTRANSID, 100));
1453        payload
1454            .extend_from_slice(&tlv_uuid(BTRFS_SEND_A_CLONE_UUID, clone_uuid));
1455        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_CLONE_CTRANSID, 99));
1456        let mut stream = v1_header();
1457        stream
1458            .extend_from_slice(&build_command(BTRFS_SEND_C_SNAPSHOT, &payload));
1459        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1460        match reader.next_command().unwrap().unwrap() {
1461            StreamCommand::Snapshot {
1462                path,
1463                uuid: u,
1464                ctransid,
1465                clone_uuid: cu,
1466                clone_ctransid,
1467            } => {
1468                assert_eq!(path, "my_snap");
1469                assert_eq!(u, uuid);
1470                assert_eq!(ctransid, 100);
1471                assert_eq!(cu, clone_uuid);
1472                assert_eq!(clone_ctransid, 99);
1473            }
1474            other => panic!("expected Snapshot, got {other:?}"),
1475        }
1476    }
1477
1478    // --- Clone command ---
1479
1480    #[test]
1481    fn clone_command() {
1482        let clone_uuid =
1483            Uuid::parse_str("12345678-1234-1234-1234-123456789abc").unwrap();
1484        let mut payload = Vec::new();
1485        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "dest"));
1486        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_FILE_OFFSET, 0));
1487        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_CLONE_LEN, 4096));
1488        payload
1489            .extend_from_slice(&tlv_uuid(BTRFS_SEND_A_CLONE_UUID, clone_uuid));
1490        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_CLONE_CTRANSID, 7));
1491        payload
1492            .extend_from_slice(&tlv_string(BTRFS_SEND_A_CLONE_PATH, "source"));
1493        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_CLONE_OFFSET, 8192));
1494        let mut stream = v1_header();
1495        stream.extend_from_slice(&build_command(BTRFS_SEND_C_CLONE, &payload));
1496        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1497        match reader.next_command().unwrap().unwrap() {
1498            StreamCommand::Clone {
1499                path,
1500                offset,
1501                len,
1502                clone_uuid: cu,
1503                clone_ctransid,
1504                clone_path,
1505                clone_offset,
1506            } => {
1507                assert_eq!(path, "dest");
1508                assert_eq!(offset, 0);
1509                assert_eq!(len, 4096);
1510                assert_eq!(cu, clone_uuid);
1511                assert_eq!(clone_ctransid, 7);
1512                assert_eq!(clone_path, "source");
1513                assert_eq!(clone_offset, 8192);
1514            }
1515            other => panic!("expected Clone, got {other:?}"),
1516        }
1517    }
1518
1519    // --- Multi-command stream ---
1520
1521    #[test]
1522    fn multiple_commands_in_sequence() {
1523        let uuid = Uuid::nil();
1524        let mut stream = v1_header();
1525
1526        // SUBVOL
1527        let mut p = Vec::new();
1528        p.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "vol"));
1529        p.extend_from_slice(&tlv_uuid(BTRFS_SEND_A_UUID, uuid));
1530        p.extend_from_slice(&tlv_u64(BTRFS_SEND_A_CTRANSID, 1));
1531        stream.extend_from_slice(&build_command(BTRFS_SEND_C_SUBVOL, &p));
1532
1533        // MKFILE
1534        stream.extend_from_slice(&build_command(
1535            BTRFS_SEND_C_MKFILE,
1536            &tlv_string(BTRFS_SEND_A_PATH, "file.txt"),
1537        ));
1538
1539        // END
1540        stream.extend_from_slice(&build_command(BTRFS_SEND_C_END, &[]));
1541
1542        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1543
1544        assert!(matches!(
1545            reader.next_command().unwrap().unwrap(),
1546            StreamCommand::Subvol { .. }
1547        ));
1548        assert!(matches!(
1549            reader.next_command().unwrap().unwrap(),
1550            StreamCommand::Mkfile { .. }
1551        ));
1552        assert!(matches!(
1553            reader.next_command().unwrap().unwrap(),
1554            StreamCommand::End
1555        ));
1556        assert!(reader.next_command().unwrap().is_none());
1557    }
1558
1559    // --- Error cases ---
1560
1561    #[test]
1562    fn unknown_command_type() {
1563        let mut stream = v1_header();
1564        stream.extend_from_slice(&build_command(99, &[]));
1565        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1566        let err = reader.next_command().unwrap_err();
1567        assert!(
1568            format!("{err}").contains("unknown send stream command type 99"),
1569            "got: {err}"
1570        );
1571    }
1572
1573    #[test]
1574    fn missing_required_attribute() {
1575        // MKFILE with no PATH attribute.
1576        let mut stream = v1_header();
1577        stream.extend_from_slice(&build_command(BTRFS_SEND_C_MKFILE, &[]));
1578        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1579        let err = reader.next_command().unwrap_err();
1580        assert!(
1581            format!("{err}").contains("missing required attribute"),
1582            "got: {err}"
1583        );
1584    }
1585
1586    #[test]
1587    fn truncated_payload() {
1588        let mut stream = v1_header();
1589        // Write a command header claiming 100 bytes of payload, but don't
1590        // provide the payload data.
1591        let len = 100u32;
1592        let cmd = BTRFS_SEND_C_END;
1593        let mut crc_input = Vec::new();
1594        crc_input.extend_from_slice(&len.to_le_bytes());
1595        crc_input.extend_from_slice(&cmd.to_le_bytes());
1596        crc_input.extend_from_slice(&[0u8; 4]);
1597        let crc = crc32c::crc32c(&crc_input);
1598        stream.extend_from_slice(&len.to_le_bytes());
1599        stream.extend_from_slice(&cmd.to_le_bytes());
1600        stream.extend_from_slice(&crc.to_le_bytes());
1601        // No payload bytes follow.
1602        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1603        let err = reader.next_command().unwrap_err();
1604        assert!(format!("{err}").contains("truncated"), "got: {err}");
1605    }
1606
1607    #[test]
1608    fn into_inner_returns_reader() {
1609        let stream = v1_header();
1610        let cursor = Cursor::new(stream);
1611        let reader = StreamReader::new(cursor).unwrap();
1612        let _inner: Cursor<Vec<u8>> = reader.into_inner();
1613    }
1614
1615    // --- UpdateExtent ---
1616
1617    #[test]
1618    fn update_extent_command() {
1619        let mut payload = Vec::new();
1620        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "extent.dat"));
1621        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_FILE_OFFSET, 65536));
1622        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_SIZE, 131072));
1623        let mut stream = v1_header();
1624        stream.extend_from_slice(&build_command(
1625            BTRFS_SEND_C_UPDATE_EXTENT,
1626            &payload,
1627        ));
1628        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1629        match reader.next_command().unwrap().unwrap() {
1630            StreamCommand::UpdateExtent { path, offset, len } => {
1631                assert_eq!(path, "extent.dat");
1632                assert_eq!(offset, 65536);
1633                assert_eq!(len, 131072);
1634            }
1635            other => panic!("expected UpdateExtent, got {other:?}"),
1636        }
1637    }
1638
1639    // --- Mknod, Mkfifo, Mksock ---
1640
1641    #[test]
1642    fn mknod_command() {
1643        let mut payload = Vec::new();
1644        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "dev_node"));
1645        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_MODE, 0o660));
1646        payload.extend_from_slice(&tlv_u64(BTRFS_SEND_A_RDEV, 0x0801)); // 8:1
1647        let mut stream = v1_header();
1648        stream.extend_from_slice(&build_command(BTRFS_SEND_C_MKNOD, &payload));
1649        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1650        match reader.next_command().unwrap().unwrap() {
1651            StreamCommand::Mknod { path, mode, rdev } => {
1652                assert_eq!(path, "dev_node");
1653                assert_eq!(mode, 0o660);
1654                assert_eq!(rdev, 0x0801);
1655            }
1656            other => panic!("expected Mknod, got {other:?}"),
1657        }
1658    }
1659
1660    #[test]
1661    fn mkfifo_command() {
1662        let payload = tlv_string(BTRFS_SEND_A_PATH, "my_fifo");
1663        let mut stream = v1_header();
1664        stream.extend_from_slice(&build_command(BTRFS_SEND_C_MKFIFO, &payload));
1665        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1666        match reader.next_command().unwrap().unwrap() {
1667            StreamCommand::Mkfifo { path } => assert_eq!(path, "my_fifo"),
1668            other => panic!("expected Mkfifo, got {other:?}"),
1669        }
1670    }
1671
1672    #[test]
1673    fn mksock_command() {
1674        let payload = tlv_string(BTRFS_SEND_A_PATH, "my_sock");
1675        let mut stream = v1_header();
1676        stream.extend_from_slice(&build_command(BTRFS_SEND_C_MKSOCK, &payload));
1677        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1678        match reader.next_command().unwrap().unwrap() {
1679            StreamCommand::Mksock { path } => assert_eq!(path, "my_sock"),
1680            other => panic!("expected Mksock, got {other:?}"),
1681        }
1682    }
1683
1684    // --- Link and RemoveXattr ---
1685
1686    #[test]
1687    fn link_command() {
1688        let mut payload = Vec::new();
1689        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "hardlink"));
1690        payload
1691            .extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH_LINK, "original"));
1692        let mut stream = v1_header();
1693        stream.extend_from_slice(&build_command(BTRFS_SEND_C_LINK, &payload));
1694        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1695        match reader.next_command().unwrap().unwrap() {
1696            StreamCommand::Link { path, target } => {
1697                assert_eq!(path, "hardlink");
1698                assert_eq!(target, "original");
1699            }
1700            other => panic!("expected Link, got {other:?}"),
1701        }
1702    }
1703
1704    #[test]
1705    fn remove_xattr_command() {
1706        let mut payload = Vec::new();
1707        payload.extend_from_slice(&tlv_string(BTRFS_SEND_A_PATH, "file"));
1708        payload.extend_from_slice(&tlv_string(
1709            BTRFS_SEND_A_XATTR_NAME,
1710            "user.old",
1711        ));
1712        let mut stream = v1_header();
1713        stream.extend_from_slice(&build_command(
1714            BTRFS_SEND_C_REMOVE_XATTR,
1715            &payload,
1716        ));
1717        let mut reader = StreamReader::new(Cursor::new(stream)).unwrap();
1718        match reader.next_command().unwrap().unwrap() {
1719            StreamCommand::RemoveXattr { path, name } => {
1720                assert_eq!(path, "file");
1721                assert_eq!(name, "user.old");
1722            }
1723            other => panic!("expected RemoveXattr, got {other:?}"),
1724        }
1725    }
1726}
btrfs_stream/stream.rs

btrfs_stream/
stream.rs
