1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
mod dump;
use crate::{RunContext, Runnable};
use anyhow::{Context, Result, bail};
use btrfs_stream::{ReceiveContext, StreamCommand, StreamReader};
use clap::Parser;
use std::{fs::File, io, path::PathBuf};
/// Receive subvolumes from a stream.
///
/// Read a btrfs send stream and recreate subvolumes on the destination filesystem.
/// Streams can be received incrementally based on a parent subvolume to only
/// apply changes. Multiple streams can be received in sequence. The destination
/// filesystem must be mounted and writable. Requires CAP_SYS_ADMIN.
#[derive(Parser, Debug)]
#[allow(clippy::doc_markdown, clippy::struct_excessive_bools)]
pub struct ReceiveCommand {
/// Mount point of the destination filesystem (not required with --dump)
mount: Option<PathBuf>,
/// Read the stream from FILE instead of stdin
#[clap(short = 'f', long = "file")]
file: Option<PathBuf>,
/// Terminate after receiving an end-cmd marker
#[clap(short = 'e', long)]
terminate_on_end: bool,
/// Confine the process to directory using chroot
#[clap(short = 'C', long)]
chroot: bool,
/// Terminate after NERR errors (0 means unlimited)
#[clap(short = 'E', long)]
max_errors: Option<u64>,
/// The root mount point of the destination filesystem
#[clap(short = 'm', long = "root-mount")]
root_mount: Option<PathBuf>,
/// Always decompress instead of using encoded I/O
#[clap(long)]
force_decompress: bool,
/// Dump stream metadata without requiring the mount parameter
#[clap(long)]
dump: bool,
}
impl Runnable for ReceiveCommand {
fn run(&self, _ctx: &RunContext) -> Result<()> {
let input: Box<dyn io::Read> = match &self.file {
Some(path) => Box::new(File::open(path).with_context(|| {
format!("cannot open '{}'", path.display())
})?),
None => Box::new(io::stdin()),
};
if self.dump {
return dump::dump_stream(input);
}
let mount = self.mount.as_ref().ok_or_else(|| {
anyhow::anyhow!("mount point is required (unless --dump)")
})?;
if !mount.is_dir() {
bail!("'{}' is not a directory", mount.display());
}
// The input file must be opened before chroot (it may be outside
// the mount point). The stream reader consumes the input.
let mut reader = StreamReader::new(input)?;
let dest = if self.chroot {
// Confine the process to the mount point. After this, all paths
// in the stream are resolved relative to "/".
let mount_cstr =
std::ffi::CString::new(mount.to_str().ok_or_else(|| {
anyhow::anyhow!("mount path is not valid UTF-8")
})?)
.context("mount path contains null byte")?;
if unsafe { nix::libc::chroot(mount_cstr.as_ptr()) } != 0 {
return Err(std::io::Error::last_os_error()).context(format!(
"failed to chroot to '{}'",
mount.display()
));
}
if unsafe { nix::libc::chdir(c"/".as_ptr()) } != 0 {
return Err(std::io::Error::last_os_error())
.context("failed to chdir to / after chroot");
}
eprintln!("Chroot to {}", mount.display());
PathBuf::from("/")
} else {
mount.clone()
};
let mut ctx = ReceiveContext::new(&dest)?;
let max_errors = self.max_errors.unwrap_or(0);
let mut error_count = 0u64;
let mut received_subvol = false;
loop {
match reader.next_command() {
Err(e) => {
error_count += 1;
eprintln!("ERROR: {e:#}");
if max_errors > 0 && error_count >= max_errors {
bail!("too many errors ({error_count}), aborting");
}
}
Ok(None) => {
// EOF — finalize and exit.
break;
}
Ok(Some(StreamCommand::End)) => {
ctx.close_write_fd();
ctx.finish_subvol()?;
received_subvol = false;
if self.terminate_on_end {
return Ok(());
}
// Try to read the next stream header for multi-stream input.
// If there's more data, the next call to next_command() on a
// new reader will pick it up. We re-create the reader with the
// remaining input.
let inner = reader.into_inner();
match StreamReader::new(inner) {
Ok(new_reader) => {
reader = new_reader;
}
Err(_) => {
// No more streams.
return Ok(());
}
}
}
Ok(Some(cmd)) => {
if matches!(
&cmd,
StreamCommand::Subvol { .. }
| StreamCommand::Snapshot { .. }
) {
received_subvol = true;
}
if let Err(e) = ctx.process_command(&cmd) {
error_count += 1;
eprintln!("ERROR: {e:#}");
if max_errors > 0 && error_count >= max_errors {
bail!("too many errors ({error_count}), aborting");
}
}
}
}
}
// Finalize the last subvolume if we received one.
if received_subvol {
ctx.finish_subvol()?;
}
Ok(())
}
}