use crate::{
consts::{
BTRFS_SEND_A_ATIME, BTRFS_SEND_A_CLONE_CTRANSID,
BTRFS_SEND_A_CLONE_LEN, BTRFS_SEND_A_CLONE_OFFSET,
BTRFS_SEND_A_CLONE_PATH, BTRFS_SEND_A_CLONE_UUID,
BTRFS_SEND_A_COMPRESSION, BTRFS_SEND_A_CTIME, BTRFS_SEND_A_CTRANSID,
BTRFS_SEND_A_DATA, BTRFS_SEND_A_ENCRYPTION,
BTRFS_SEND_A_FALLOCATE_MODE, BTRFS_SEND_A_FILE_OFFSET,
BTRFS_SEND_A_FILEATTR, BTRFS_SEND_A_GID, BTRFS_SEND_A_MODE,
BTRFS_SEND_A_MTIME, BTRFS_SEND_A_PATH, BTRFS_SEND_A_PATH_LINK,
BTRFS_SEND_A_PATH_TO, BTRFS_SEND_A_RDEV, BTRFS_SEND_A_SIZE,
BTRFS_SEND_A_UID, BTRFS_SEND_A_UNENCODED_FILE_LEN,
BTRFS_SEND_A_UNENCODED_LEN, BTRFS_SEND_A_UNENCODED_OFFSET,
BTRFS_SEND_A_UUID, BTRFS_SEND_A_VERITY_ALGORITHM,
BTRFS_SEND_A_VERITY_BLOCK_SIZE, BTRFS_SEND_A_VERITY_SALT_DATA,
BTRFS_SEND_A_VERITY_SIG_DATA, BTRFS_SEND_A_XATTR_DATA,
BTRFS_SEND_A_XATTR_NAME, BTRFS_SEND_C_CHMOD, BTRFS_SEND_C_CHOWN,
BTRFS_SEND_C_CLONE, BTRFS_SEND_C_ENABLE_VERITY,
BTRFS_SEND_C_ENCODED_WRITE, BTRFS_SEND_C_END, BTRFS_SEND_C_FALLOCATE,
BTRFS_SEND_C_FILEATTR, BTRFS_SEND_C_LINK, BTRFS_SEND_C_MKDIR,
BTRFS_SEND_C_MKFIFO, BTRFS_SEND_C_MKFILE, BTRFS_SEND_C_MKNOD,
BTRFS_SEND_C_MKSOCK, BTRFS_SEND_C_REMOVE_XATTR, BTRFS_SEND_C_RENAME,
BTRFS_SEND_C_RMDIR, BTRFS_SEND_C_SET_XATTR, BTRFS_SEND_C_SNAPSHOT,
BTRFS_SEND_C_SUBVOL, BTRFS_SEND_C_SYMLINK, BTRFS_SEND_C_TRUNCATE,
BTRFS_SEND_C_UNLINK, BTRFS_SEND_C_UPDATE_EXTENT, BTRFS_SEND_C_UTIMES,
BTRFS_SEND_C_WRITE, CMD_HEADER_LEN, SEND_STREAM_MAGIC,
SEND_STREAM_MAGIC_LEN, STREAM_HEADER_LEN,
},
stream::{StreamCommand, Timespec},
};
use std::io::{self, Write};
use uuid::Uuid;
#[derive(Debug)]
pub struct StreamWriter<W: Write> {
inner: W,
version: u32,
}
impl<W: Write> StreamWriter<W> {
pub fn new(mut inner: W, version: u32) -> io::Result<Self> {
if version == 0 || version > 3 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"unsupported send stream version {version} (supported: 1-3)"
),
));
}
let mut header = [0u8; STREAM_HEADER_LEN];
header[..SEND_STREAM_MAGIC_LEN].copy_from_slice(SEND_STREAM_MAGIC);
header[SEND_STREAM_MAGIC_LEN..STREAM_HEADER_LEN]
.copy_from_slice(&version.to_le_bytes());
inner.write_all(&header)?;
Ok(Self { inner, version })
}
#[must_use]
pub fn version(&self) -> u32 {
self.version
}
pub fn finish(mut self) -> io::Result<W> {
self.inner.flush()?;
Ok(self.inner)
}
pub fn write_command(&mut self, cmd: &StreamCommand) -> io::Result<()> {
let mut payload = Vec::new();
let cmd_id = encode_command(cmd, &mut payload, self.version)?;
self.write_framed(cmd_id, &payload)
}
fn write_framed(&mut self, cmd_id: u16, payload: &[u8]) -> io::Result<()> {
let payload_len = u32::try_from(payload.len()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"command {cmd_id} payload exceeds u32 length field: {} bytes",
payload.len(),
),
)
})?;
let mut header = [0u8; CMD_HEADER_LEN];
header[0..4].copy_from_slice(&payload_len.to_le_bytes());
header[4..6].copy_from_slice(&cmd_id.to_le_bytes());
let crc = crc32c::crc32c_append(!0, &header[0..6]);
let crc = crc32c::crc32c_append(crc, &[0u8; 4]);
let crc = !crc32c::crc32c_append(crc, payload);
header[6..10].copy_from_slice(&crc.to_le_bytes());
self.inner.write_all(&header)?;
self.inner.write_all(payload)?;
Ok(())
}
}
#[allow(clippy::too_many_lines)]
fn encode_command(
cmd: &StreamCommand,
out: &mut Vec<u8>,
version: u32,
) -> io::Result<u16> {
match cmd {
StreamCommand::Subvol {
path,
uuid,
ctransid,
} => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_uuid(out, BTRFS_SEND_A_UUID, uuid);
put_attr_u64(out, BTRFS_SEND_A_CTRANSID, *ctransid);
Ok(BTRFS_SEND_C_SUBVOL)
}
StreamCommand::Snapshot {
path,
uuid,
ctransid,
clone_uuid,
clone_ctransid,
} => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_uuid(out, BTRFS_SEND_A_UUID, uuid);
put_attr_u64(out, BTRFS_SEND_A_CTRANSID, *ctransid);
put_attr_uuid(out, BTRFS_SEND_A_CLONE_UUID, clone_uuid);
put_attr_u64(out, BTRFS_SEND_A_CLONE_CTRANSID, *clone_ctransid);
Ok(BTRFS_SEND_C_SNAPSHOT)
}
StreamCommand::Mkfile { path } => {
path_only(out, path, BTRFS_SEND_C_MKFILE)
}
StreamCommand::Mkdir { path } => {
path_only(out, path, BTRFS_SEND_C_MKDIR)
}
StreamCommand::Mknod { path, mode, rdev } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_MODE, *mode);
put_attr_u64(out, BTRFS_SEND_A_RDEV, *rdev);
Ok(BTRFS_SEND_C_MKNOD)
}
StreamCommand::Mkfifo { path } => {
path_only(out, path, BTRFS_SEND_C_MKFIFO)
}
StreamCommand::Mksock { path } => {
path_only(out, path, BTRFS_SEND_C_MKSOCK)
}
StreamCommand::Symlink { path, target } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_str(out, BTRFS_SEND_A_PATH_LINK, target)?;
Ok(BTRFS_SEND_C_SYMLINK)
}
StreamCommand::Rename { from, to } => {
put_attr_str(out, BTRFS_SEND_A_PATH, from)?;
put_attr_str(out, BTRFS_SEND_A_PATH_TO, to)?;
Ok(BTRFS_SEND_C_RENAME)
}
StreamCommand::Link { path, target } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_str(out, BTRFS_SEND_A_PATH_LINK, target)?;
Ok(BTRFS_SEND_C_LINK)
}
StreamCommand::Unlink { path } => {
path_only(out, path, BTRFS_SEND_C_UNLINK)
}
StreamCommand::Rmdir { path } => {
path_only(out, path, BTRFS_SEND_C_RMDIR)
}
StreamCommand::Write { path, offset, data } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_FILE_OFFSET, *offset);
put_attr_data(out, data, version)?;
Ok(BTRFS_SEND_C_WRITE)
}
StreamCommand::Clone {
path,
offset,
len,
clone_uuid,
clone_ctransid,
clone_path,
clone_offset,
} => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_FILE_OFFSET, *offset);
put_attr_u64(out, BTRFS_SEND_A_CLONE_LEN, *len);
put_attr_uuid(out, BTRFS_SEND_A_CLONE_UUID, clone_uuid);
put_attr_u64(out, BTRFS_SEND_A_CLONE_CTRANSID, *clone_ctransid);
put_attr_str(out, BTRFS_SEND_A_CLONE_PATH, clone_path)?;
put_attr_u64(out, BTRFS_SEND_A_CLONE_OFFSET, *clone_offset);
Ok(BTRFS_SEND_C_CLONE)
}
StreamCommand::SetXattr { path, name, data } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_str(out, BTRFS_SEND_A_XATTR_NAME, name)?;
put_attr_bytes(out, BTRFS_SEND_A_XATTR_DATA, data)?;
Ok(BTRFS_SEND_C_SET_XATTR)
}
StreamCommand::RemoveXattr { path, name } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_str(out, BTRFS_SEND_A_XATTR_NAME, name)?;
Ok(BTRFS_SEND_C_REMOVE_XATTR)
}
StreamCommand::Truncate { path, size } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_SIZE, *size);
Ok(BTRFS_SEND_C_TRUNCATE)
}
StreamCommand::Chmod { path, mode } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_MODE, *mode);
Ok(BTRFS_SEND_C_CHMOD)
}
StreamCommand::Chown { path, uid, gid } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_UID, *uid);
put_attr_u64(out, BTRFS_SEND_A_GID, *gid);
Ok(BTRFS_SEND_C_CHOWN)
}
StreamCommand::Utimes {
path,
atime,
mtime,
ctime,
} => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_timespec(out, BTRFS_SEND_A_ATIME, atime);
put_attr_timespec(out, BTRFS_SEND_A_MTIME, mtime);
put_attr_timespec(out, BTRFS_SEND_A_CTIME, ctime);
Ok(BTRFS_SEND_C_UTIMES)
}
StreamCommand::UpdateExtent { path, offset, len } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_FILE_OFFSET, *offset);
put_attr_u64(out, BTRFS_SEND_A_SIZE, *len);
Ok(BTRFS_SEND_C_UPDATE_EXTENT)
}
StreamCommand::EncodedWrite {
path,
offset,
unencoded_file_len,
unencoded_len,
unencoded_offset,
compression,
encryption,
data,
} => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_FILE_OFFSET, *offset);
put_attr_u64(
out,
BTRFS_SEND_A_UNENCODED_FILE_LEN,
*unencoded_file_len,
);
put_attr_u64(out, BTRFS_SEND_A_UNENCODED_LEN, *unencoded_len);
put_attr_u64(out, BTRFS_SEND_A_UNENCODED_OFFSET, *unencoded_offset);
put_attr_u32(out, BTRFS_SEND_A_COMPRESSION, *compression);
put_attr_u32(out, BTRFS_SEND_A_ENCRYPTION, *encryption);
put_attr_data(out, data, version)?;
Ok(BTRFS_SEND_C_ENCODED_WRITE)
}
StreamCommand::Fallocate {
path,
mode,
offset,
len,
} => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u32(out, BTRFS_SEND_A_FALLOCATE_MODE, *mode);
put_attr_u64(out, BTRFS_SEND_A_FILE_OFFSET, *offset);
put_attr_u64(out, BTRFS_SEND_A_SIZE, *len);
Ok(BTRFS_SEND_C_FALLOCATE)
}
StreamCommand::Fileattr { path, attr } => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u64(out, BTRFS_SEND_A_FILEATTR, *attr);
Ok(BTRFS_SEND_C_FILEATTR)
}
StreamCommand::EnableVerity {
path,
algorithm,
block_size,
salt,
sig,
} => {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
put_attr_u8(out, BTRFS_SEND_A_VERITY_ALGORITHM, *algorithm);
put_attr_u32(out, BTRFS_SEND_A_VERITY_BLOCK_SIZE, *block_size);
put_attr_bytes(out, BTRFS_SEND_A_VERITY_SALT_DATA, salt)?;
put_attr_bytes(out, BTRFS_SEND_A_VERITY_SIG_DATA, sig)?;
Ok(BTRFS_SEND_C_ENABLE_VERITY)
}
StreamCommand::End => Ok(BTRFS_SEND_C_END),
}
}
fn path_only(out: &mut Vec<u8>, path: &str, cmd_id: u16) -> io::Result<u16> {
put_attr_str(out, BTRFS_SEND_A_PATH, path)?;
Ok(cmd_id)
}
fn put_attr_bytes(out: &mut Vec<u8>, attr: u16, data: &[u8]) -> io::Result<()> {
let len = u16::try_from(data.len()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"attribute {attr} payload exceeds u16 length field: {} bytes",
data.len(),
),
)
})?;
out.extend_from_slice(&attr.to_le_bytes());
out.extend_from_slice(&len.to_le_bytes());
out.extend_from_slice(data);
Ok(())
}
fn put_attr_str(out: &mut Vec<u8>, attr: u16, s: &str) -> io::Result<()> {
put_attr_bytes(out, attr, s.as_bytes())
}
fn put_attr_data(
out: &mut Vec<u8>,
data: &[u8],
version: u32,
) -> io::Result<()> {
if version >= 2 {
out.extend_from_slice(&BTRFS_SEND_A_DATA.to_le_bytes());
out.extend_from_slice(data);
Ok(())
} else {
put_attr_bytes(out, BTRFS_SEND_A_DATA, data)
}
}
fn put_attr_u64(out: &mut Vec<u8>, attr: u16, v: u64) {
put_attr_bytes(out, attr, &v.to_le_bytes())
.expect("8-byte payload always fits");
}
fn put_attr_u32(out: &mut Vec<u8>, attr: u16, v: u32) {
put_attr_bytes(out, attr, &v.to_le_bytes())
.expect("4-byte payload always fits");
}
fn put_attr_u8(out: &mut Vec<u8>, attr: u16, v: u8) {
put_attr_bytes(out, attr, &[v]).expect("1-byte payload always fits");
}
fn put_attr_uuid(out: &mut Vec<u8>, attr: u16, uuid: &Uuid) {
put_attr_bytes(out, attr, uuid.as_bytes())
.expect("16-byte UUID always fits");
}
fn put_attr_timespec(out: &mut Vec<u8>, attr: u16, t: &Timespec) {
let mut buf = [0u8; 12];
buf[0..8].copy_from_slice(&t.sec.to_le_bytes());
buf[8..12].copy_from_slice(&t.nsec.to_le_bytes());
put_attr_bytes(out, attr, &buf).expect("12-byte timespec always fits");
}
#[cfg(test)]
mod tests {
use super::*;
use crate::StreamReader;
fn roundtrip(version: u32, cmds: &[StreamCommand]) {
let mut buf: Vec<u8> = Vec::new();
let mut writer = StreamWriter::new(&mut buf, version).unwrap();
for cmd in cmds {
writer.write_command(cmd).unwrap();
}
writer.finish().unwrap();
let mut reader = StreamReader::new(buf.as_slice()).unwrap();
assert_eq!(reader.version(), version);
for expected in cmds {
let got = reader.next_command().expect("parse").expect("not eof");
assert_eq!(format!("{got:?}"), format!("{expected:?}"));
}
assert!(reader.next_command().expect("eof check").is_none());
}
#[test]
fn header_only_roundtrips_each_version() {
for v in [1, 2, 3] {
let mut buf = Vec::new();
let writer = StreamWriter::new(&mut buf, v).unwrap();
writer.finish().unwrap();
let reader = StreamReader::new(buf.as_slice()).unwrap();
assert_eq!(reader.version(), v);
}
}
#[test]
fn invalid_versions_rejected() {
let mut buf = Vec::new();
assert!(StreamWriter::new(&mut buf, 0).is_err());
let mut buf = Vec::new();
assert!(StreamWriter::new(&mut buf, 4).is_err());
}
#[test]
fn subvol_and_snapshot_roundtrip() {
let uuid = Uuid::from_u128(0x0011_2233_4455_6677_8899_aabb_ccdd_eeff);
let clone_uuid =
Uuid::from_u128(0xdead_beef_dead_beef_dead_beef_dead_beef);
roundtrip(
1,
&[
StreamCommand::Subvol {
path: "snap1".into(),
uuid,
ctransid: 42,
},
StreamCommand::Snapshot {
path: "snap2".into(),
uuid,
ctransid: 100,
clone_uuid,
clone_ctransid: 50,
},
StreamCommand::End,
],
);
}
#[test]
fn filesystem_object_creation_roundtrips() {
roundtrip(
1,
&[
StreamCommand::Mkfile {
path: "f.txt".into(),
},
StreamCommand::Mkdir { path: "d".into() },
StreamCommand::Mknod {
path: "n".into(),
mode: 0o600,
rdev: 0x102,
},
StreamCommand::Mkfifo { path: "p".into() },
StreamCommand::Mksock { path: "s".into() },
StreamCommand::Symlink {
path: "l".into(),
target: "f.txt".into(),
},
],
);
}
#[test]
fn rename_link_unlink_rmdir_roundtrip() {
roundtrip(
1,
&[
StreamCommand::Rename {
from: "a".into(),
to: "b".into(),
},
StreamCommand::Link {
path: "alias".into(),
target: "real".into(),
},
StreamCommand::Unlink { path: "old".into() },
StreamCommand::Rmdir {
path: "empty".into(),
},
],
);
}
#[test]
fn write_clone_truncate_roundtrip() {
let clone_uuid =
Uuid::from_u128(0x0123_4567_89ab_cdef_0123_4567_89ab_cdef);
roundtrip(
1,
&[
StreamCommand::Write {
path: "f".into(),
offset: 4096,
data: vec![0x42; 1024],
},
StreamCommand::Clone {
path: "dst".into(),
offset: 8192,
len: 4096,
clone_uuid,
clone_ctransid: 7,
clone_path: "src".into(),
clone_offset: 0,
},
StreamCommand::Truncate {
path: "f".into(),
size: 65536,
},
],
);
}
#[test]
fn xattr_perms_times_roundtrip() {
roundtrip(
1,
&[
StreamCommand::SetXattr {
path: "f".into(),
name: "user.greeting".into(),
data: b"hello".to_vec(),
},
StreamCommand::RemoveXattr {
path: "f".into(),
name: "user.gone".into(),
},
StreamCommand::Chmod {
path: "f".into(),
mode: 0o644,
},
StreamCommand::Chown {
path: "f".into(),
uid: 1000,
gid: 1001,
},
StreamCommand::Utimes {
path: "f".into(),
atime: Timespec { sec: 100, nsec: 1 },
mtime: Timespec { sec: 200, nsec: 2 },
ctime: Timespec { sec: 300, nsec: 3 },
},
StreamCommand::UpdateExtent {
path: "f".into(),
offset: 0,
len: 4096,
},
],
);
}
#[test]
fn v2_commands_roundtrip() {
roundtrip(
2,
&[
StreamCommand::EncodedWrite {
path: "f".into(),
offset: 0,
unencoded_file_len: 4096,
unencoded_len: 4096,
unencoded_offset: 0,
compression: 2, encryption: 0,
data: vec![0xab; 256],
},
StreamCommand::Fallocate {
path: "f".into(),
mode: 3, offset: 4096,
len: 4096,
},
StreamCommand::Fileattr {
path: "f".into(),
attr: 0x40, },
],
);
}
#[test]
fn v3_enable_verity_roundtrips() {
roundtrip(
3,
&[StreamCommand::EnableVerity {
path: "f".into(),
algorithm: 1,
block_size: 4096,
salt: vec![1, 2, 3, 4],
sig: vec![],
}],
);
}
#[test]
fn end_command_roundtrips() {
roundtrip(1, &[StreamCommand::End]);
}
#[test]
fn corrupted_payload_fails_crc_check() {
let mut buf = Vec::new();
let mut writer = StreamWriter::new(&mut buf, 1).unwrap();
writer
.write_command(&StreamCommand::Mkfile { path: "f".into() })
.unwrap();
writer.finish().unwrap();
let payload_start = STREAM_HEADER_LEN + CMD_HEADER_LEN;
buf[payload_start + 4] ^= 0x01;
let mut reader = StreamReader::new(buf.as_slice()).unwrap();
let err = reader.next_command().unwrap_err();
assert!(
matches!(err, crate::StreamError::CrcMismatch { .. }),
"expected CrcMismatch, got {err:?}",
);
}
}