Files
sausage/userland/box/src/transfer.rs
T

380 lines
13 KiB
Rust

//! Bounded streaming transfers. Payloads stay in HTTP/VFS, not RPC responses.
use super::*;
pub fn file_path(id: &str, path: &str) -> Result<String> {
if !path.starts_with('/') || path == "/" {
return Err(error("expected an absolute file path inside the box"));
}
let mut u = url::Url::parse("http://box.invalid").map_err(|e| error(e.to_string()))?;
{
let mut segments = u.path_segments_mut().map_err(|_| error("bad file path"))?;
segments.clear().push("containers").push(id).push("files");
for s in path[1..].split('/') {
segments.push(s);
}
}
Ok(u.path().into())
}
pub fn put(h: &str, id: &str, path: &str, source: &mut dyn Read, len: u64) -> Result<u64> {
if len > MAX_FILE {
return Err(error("service file limit is 64 MiB; split larger files"));
}
let mut body = http::request(
h,
Method::Put,
&file_path(id, path)?,
Some((source, len)),
300,
)?;
let ok: OkResult = serde_json::from_reader(&mut body)?;
if !ok.ok {
return Err(error("service file write did not succeed"));
}
Ok(len)
}
fn size(h: &str, id: &str, path: &str) -> Result<u64> {
// stat only regular files: service otherwise returns directory listings or
// can silently truncate at 64 MiB. Keep flags and user paths separate.
let out = checked(
h,
id,
json!([
"sh",
"-c",
"[ -f \"$1\" ] && stat -Lc %s -- \"$1\"",
"sh",
path
]),
)?;
let n: u64 = out
.trim()
.parse()
.map_err(|_| error("invalid file size from box"))?;
if n > MAX_FILE {
return Err(error("service file limit is 64 MiB; split larger files"));
}
Ok(n)
}
fn download(h: &str, id: &str, path: &str, out: &mut dyn Write, expected: u64) -> Result<u64> {
let mut body = http::request(h, Method::Get, &file_path(id, path)?, None, 300)?;
let n = std::io::copy(&mut (&mut body).take(MAX_FILE + 1), out)?;
if n != expected {
return Err(error(format!(
"file changed or response truncated: expected {expected} bytes, got {n}"
)));
}
Ok(n)
}
/// Private disk spools, removed on both success and error. Not persistent state.
struct Spool {
dir: std::path::PathBuf,
}
impl Spool {
fn new() -> Result<Self> {
static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
let dir = std::path::PathBuf::from(format!(
"/var/lib/box/transfers/{}-{}",
crate::pid(),
NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
));
std::fs::create_dir_all(dir.parent().ok_or_else(|| error("no spool parent"))?)?;
std::fs::create_dir(&dir)?;
Ok(Self { dir })
}
fn path(&self) -> std::path::PathBuf {
self.dir.join("archive.tar")
}
}
impl Drop for Spool {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.dir);
}
}
pub fn get_file(h: &str, p: GetFile) -> Result<FileResult> {
let len = size(h, &p.id, &p.path)?;
if let Some(dest) = &p.to_vfs {
let target = Path::new(dest);
let parent = target
.parent()
.filter(|p| !p.as_os_str().is_empty())
.unwrap_or(Path::new("."));
std::fs::create_dir_all(parent)?;
let temp = parent.join(format!(".box-download-{}", crate::pid()));
// Use create_new: never truncate an unrelated temporary file.
let mut f = File::options().write(true).create_new(true).open(&temp)?;
let result = (|| -> Result<()> {
download(h, &p.id, &p.path, &mut f, len)?;
f.flush()?;
drop(f);
std::fs::rename(&temp, target)?;
Ok(())
})();
if result.is_err() {
let _ = std::fs::remove_file(&temp);
}
result?;
Ok(FileResult {
bytes: len,
text: None,
to_vfs: p.to_vfs,
})
} else {
if len > 65536 {
return Err(error(
"inline file limit is 64 KiB; use to_vfs to stream larger files",
));
}
let mut data = Vec::new();
download(h, &p.id, &p.path, &mut data, len)?;
Ok(FileResult {
bytes: len,
text: Some(
String::from_utf8(data).map_err(|_| error("file is not UTF-8; use to_vfs"))?,
),
to_vfs: None,
})
}
}
fn quote(s: &str) -> String {
format!("'{}'", s.replace('\'', "'\\''"))
}
fn dir(s: &str) -> Result<()> {
if !s.starts_with('/') {
return Err(error("box directory must be absolute"));
}
Ok(())
}
fn remote_tmp() -> String {
format!("/tmp/sausage-box-{}.tar", crate::pid())
}
fn cleanup(h: &str, id: &str, path: &str) {
if let Err(e) = checked(h, id, json!(["rm", "-f", "--", path])) {
eprintln!("box transfer cleanup: {e}");
}
}
// tar::Header::set_metadata panics on wasm32-wasip2. Construct portable
// headers explicitly; WASI does not expose Unix executable mode bits.
fn append_tree(tar: &mut tar::Builder<File>, source: &Path, name: &Path) -> Result<()> {
let metadata = std::fs::symlink_metadata(source)?;
let payload = if metadata.is_file() {
metadata.len()
} else {
0
};
if tar
.get_ref()
.metadata()?
.len()
.saturating_add(payload)
.saturating_add(4096)
> MAX_FILE
{
return Err(error("tree tar exceeds 64 MiB service limit"));
}
let mut header = tar::Header::new_gnu();
header.set_mtime(
metadata
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs())
.unwrap_or(0),
);
header.set_uid(0);
header.set_gid(0);
if metadata.is_dir() {
header.set_entry_type(tar::EntryType::Directory);
header.set_size(0);
header.set_mode(0o755);
header.set_cksum();
tar.append_data(&mut header, name, std::io::empty())?;
let mut entries = std::fs::read_dir(source)?.collect::<std::io::Result<Vec<_>>>()?;
entries.sort_by_key(|e| e.file_name());
for entry in entries {
append_tree(tar, &entry.path(), &name.join(entry.file_name()))?;
}
} else if metadata.is_file() {
if metadata.len() > MAX_FILE {
return Err(error("tree exceeds 64 MiB service limit"));
}
header.set_entry_type(tar::EntryType::Regular);
header.set_size(metadata.len());
header.set_mode(0o644);
header.set_cksum();
tar.append_data(&mut header, name, File::open(source)?)?;
} else if metadata.file_type().is_symlink() {
header.set_entry_type(tar::EntryType::Symlink);
header.set_size(0);
header.set_mode(0o777);
header.set_cksum();
tar.append_link(&mut header, name, std::fs::read_link(source)?)?;
} else {
return Err(error(format!(
"cannot copy special file {}",
source.display()
)));
}
Ok(())
}
pub fn put_tree(h: &str, p: PutTree) -> Result<OkResult> {
dir(&p.to_dir)?;
if !Path::new(&p.from_vfs_dir).is_dir() {
return Err(error("from_vfs_dir is not a directory"));
}
let spool = Spool::new()?;
let f = File::create(spool.path())?;
let mut tar = tar::Builder::new(f);
tar.follow_symlinks(false);
append_tree(&mut tar, Path::new(&p.from_vfs_dir), Path::new("."))?;
let mut output = tar.into_inner()?;
output.flush()?;
drop(output);
let mut f = File::open(spool.path())?;
let len = f.metadata()?.len();
let remote = remote_tmp();
let result = (|| -> Result<()> {
put(h, &p.id, &remote, &mut f, len)?;
checked(
h,
&p.id,
json!([
"sh",
"-c",
format!(
"mkdir -p -- {0} && tar -xf {1} -C {0}",
quote(&p.to_dir),
quote(&remote)
)
]),
)?;
Ok(())
})();
cleanup(h, &p.id, &remote);
result?;
Ok(OkResult { ok: true })
}
// tar's generic unpack attempts unsupported Unix filesystem operations on
// WASI. Copy contents explicitly, avoiding chmod/utime. Symlink creation is
// unsupported by the world VFS and must fail without publishing a partial tree.
fn unpack_tree(file: File, root: &Path) -> Result<()> {
let mut archive = tar::Archive::new(file);
let mut hardlinks = Vec::new();
let mut expanded = 0u64;
for entry in archive.entries()? {
let mut entry = entry?;
let name = entry.path()?.into_owned();
let safe = |p: &Path| {
p.components().all(|c| {
matches!(
c,
std::path::Component::Normal(_) | std::path::Component::CurDir
)
})
};
if !safe(&name) {
return Err(error("archive has a path outside the destination"));
}
let path = root.join(&name);
// Do not accidentally write through an earlier archive symlink.
for parent in name.ancestors().skip(1) {
if std::fs::symlink_metadata(root.join(parent))
.is_ok_and(|m| m.file_type().is_symlink())
{
return Err(error("archive writes through a symlink directory"));
}
}
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let kind = entry.header().entry_type();
if kind.is_dir() {
std::fs::create_dir_all(&path)?;
} else if kind.is_file() {
expanded = expanded.saturating_add(entry.size());
if expanded > MAX_FILE {
return Err(error("expanded tree exceeds 64 MiB transfer limit"));
}
let mut out = File::options().write(true).create_new(true).open(&path)?;
std::io::copy(&mut entry, &mut out)?;
} else if kind.is_symlink() {
return Err(error(
"world VFS does not support creating symlinks; export a dereferenced tar/tree from the box instead",
));
} else if kind.is_hard_link() {
let target = entry
.link_name()?
.ok_or_else(|| error("archive hardlink has no target"))?
.into_owned();
if !safe(&target) {
return Err(error("archive hardlink target is outside destination"));
}
hardlinks.push((path, root.join(target)));
} else {
return Err(error("archive contains an unsupported special file"));
}
}
// Materialize hardlinks as independent files: VFS has no hardlink contract.
for (path, target) in hardlinks {
let metadata = std::fs::symlink_metadata(&target)?;
if !metadata.is_file() {
return Err(error("archive hardlink target is not a regular file"));
}
expanded = expanded.saturating_add(metadata.len());
if expanded > MAX_FILE {
return Err(error("expanded tree exceeds 64 MiB transfer limit"));
}
let mut out = File::options().write(true).create_new(true).open(path)?;
std::io::copy(&mut File::open(target)?, &mut out)?;
}
Ok(())
}
pub fn get_tree(h: &str, p: GetTree) -> Result<OkResult> {
dir(&p.from_dir)?;
let dest = Path::new(&p.to_vfs_dir);
if dest.try_exists()? {
return Err(error("to_vfs_dir already exists; choose a new directory"));
}
let spool = Spool::new()?;
let remote = remote_tmp();
let result = (|| -> Result<()> {
checked(
h,
&p.id,
json!(["tar", "-cf", remote, "-C", p.from_dir, "."]),
)?;
let len = size(h, &p.id, &remote)?;
let mut f = File::create(spool.path())?;
download(h, &p.id, &remote, &mut f, len)?;
drop(f);
// Stage alongside destination so rename is atomic and cannot cross mounts.
let parent = dest
.parent()
.filter(|p| !p.as_os_str().is_empty())
.unwrap_or(Path::new("."));
std::fs::create_dir_all(parent)?;
let staging = parent.join(format!(".box-tree-{}", crate::pid()));
std::fs::create_dir(&staging)?;
let unpack = (|| -> Result<()> {
unpack_tree(File::open(spool.path())?, &staging)?;
// Recheck to avoid accidentally replacing an existing empty directory.
if dest.try_exists()? {
return Err(error("destination appeared during transfer"));
}
std::fs::rename(&staging, dest)?;
Ok(())
})();
if unpack.is_err() {
let _ = std::fs::remove_dir_all(staging);
}
unpack
})();
cleanup(h, &p.id, &remote);
result?;
Ok(OkResult { ok: true })
}