use async_compression::tokio::bufread::GzipDecoder;
use std::future::poll_fn;
use std::io;
use std::path::{Component, Path, PathBuf};
use std::pin::Pin;
use std::time::{Duration, SystemTime};
use tokio::io::{AsyncRead, AsyncWriteExt, BufReader};
use tokio_tar::{Archive, Entry, EntryType};
use tracing::{debug, error, info};
use crate::errors::ExtractError;
pub async fn extract<R: AsyncRead + Unpin + Send>(
reader: R,
dest_dir: &Path,
) -> Result<(), ExtractError> {
let mut decoder = GzipDecoder::new(BufReader::new(reader));
decoder.multiple_members(true);
let mut archive = Archive::new(decoder);
let mut entries = archive.entries().map_err(unreadable)?;
let mut entry_count = 0u64;
while let Some(entry) =
poll_fn(|cx| futures_core::Stream::poll_next(Pin::new(&mut entries), cx)).await
{
let mut entry = entry.map_err(unreadable)?;
entry_count = entry_count.saturating_add(1);
let path = entry.path().map_err(unreadable)?.into_owned();
if entry_count <= 10 {
info!(
name = %path.display(),
kind = entry.header().entry_type().as_byte(),
size = entry.effective_size(),
"tar entry"
);
}
if !is_safe(&path, entry.header().entry_type()) {
debug!(name = %path.display(), "skipping tar entry");
continue;
}
write_entry(&mut entry, &dest_dir.join(clean_path(&path))).await?;
}
drop(entries);
tokio::io::copy(&mut archive, &mut tokio::io::sink())
.await
.map_err(unreadable)?;
info!(total_entries = entry_count, "tar extraction complete");
Ok(())
}
fn unreadable(error: io::Error) -> ExtractError {
if error.kind() == io::ErrorKind::UnexpectedEof {
ExtractError::Truncated(error)
} else {
ExtractError::Corrupt(error)
}
}
async fn write_entry<R: AsyncRead + Unpin>(
entry: &mut Entry<Archive<R>>,
target: &Path,
) -> Result<(), ExtractError> {
match entry.header().entry_type() {
EntryType::Directory => tokio::fs::create_dir_all(target)
.await
.map_err(ExtractError::Io),
EntryType::Regular | EntryType::Continuous => write_file(entry, target).await,
_ => Ok(()),
}
}
async fn write_file<R: AsyncRead + Unpin>(
entry: &mut Entry<Archive<R>>,
target: &Path,
) -> Result<(), ExtractError> {
if let Some(parent) = target.parent() {
tokio::fs::create_dir_all(parent)
.await
.map_err(ExtractError::Io)?;
}
let mode = (entry.header().mode().unwrap_or(0o644) | 0o600) & 0o755;
let mtime = entry.header().mtime().ok();
let mut file = tokio::fs::File::create(target)
.await
.map_err(ExtractError::Io)?;
copy_member(entry, &mut file).await?;
file.flush().await.map_err(ExtractError::Io)?;
let file = file.into_std().await;
if let Some(modified) =
mtime.and_then(|secs| SystemTime::UNIX_EPOCH.checked_add(Duration::from_secs(secs)))
{
file.set_modified(modified).ok();
}
drop(file);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
tokio::fs::set_permissions(target, std::fs::Permissions::from_mode(mode))
.await
.ok();
}
Ok(())
}
async fn copy_member<R: AsyncRead + Unpin>(
entry: &mut Entry<Archive<R>>,
file: &mut tokio::fs::File,
) -> Result<(), ExtractError> {
use tokio::io::AsyncReadExt;
let mut buffer = vec![0u8; 64 * 1024];
loop {
let read = entry.read(&mut buffer).await.map_err(unreadable)?;
if read == 0 {
return Ok(());
}
let chunk = buffer.get(..read).unwrap_or_default();
file.write_all(chunk).await.map_err(ExtractError::Io)?;
}
}
fn clean_path(path: &Path) -> PathBuf {
path.components()
.filter_map(|component| match component {
Component::Normal(part) => Some(part),
_ => None,
})
.collect()
}
fn is_safe(path: &Path, kind: EntryType) -> bool {
if kind == EntryType::Symlink || kind == EntryType::Link {
return false;
}
if path.is_absolute()
|| path
.components()
.any(|component| matches!(component, Component::ParentDir))
{
error!(name = %path.display(), "unsafe path detected");
return false;
}
true
}
#[cfg(test)]
mod tests {
use super::extract;
use crate::errors::ExtractError;
use flate2::write::GzEncoder;
use flate2::Compression;
use std::path::Path;
use std::time::{Duration, SystemTime};
const MTIME: u64 = 1_700_000_000;
fn tar_gz(entries: &[(&str, &[u8])]) -> Vec<u8> {
let mut builder = tar::Builder::new(GzEncoder::new(Vec::new(), Compression::default()));
builder.preserve_absolute(true);
for (name, contents) in entries {
let mut header = tar::Header::new_gnu();
header.set_size(u64::try_from(contents.len()).unwrap());
header.set_mode(0o644);
header.set_mtime(MTIME);
builder.append_data(&mut header, name, *contents).unwrap();
}
builder.into_inner().unwrap().finish().unwrap()
}
fn hostile_tar_gz(kind: tar::EntryType, name: &str, link: &str, contents: &[u8]) -> Vec<u8> {
let mut builder = tar::Builder::new(GzEncoder::new(Vec::new(), Compression::default()));
let mut header = tar::Header::new_old();
header
.as_old_mut()
.name
.get_mut(..name.len())
.unwrap()
.copy_from_slice(name.as_bytes());
header
.as_old_mut()
.linkname
.get_mut(..link.len())
.unwrap()
.copy_from_slice(link.as_bytes());
header.set_entry_type(kind);
header.set_size(u64::try_from(contents.len()).unwrap());
header.set_mode(0o644);
header.set_cksum();
builder.append(&header, contents).unwrap();
builder.into_inner().unwrap().finish().unwrap()
}
async fn extract_into(archive: &[u8], dest: &Path) {
extract(archive, dest).await.unwrap();
}
async fn extract_to_temp(archive: &[u8]) -> tempfile::TempDir {
let dir = tempfile::tempdir().unwrap();
extract_into(archive, dir.path()).await;
dir
}
fn entries_under(dir: &Path) -> Vec<String> {
let mut found: Vec<String> = walkdir::WalkDir::new(dir)
.min_depth(1)
.into_iter()
.map(|entry| {
entry
.unwrap()
.path()
.strip_prefix(dir)
.unwrap()
.display()
.to_string()
})
.collect();
found.sort();
found
}
#[tokio::test]
async fn an_archive_is_extracted_from_a_reader() {
let dir = extract_to_temp(&tar_gz(&[("repo/HEAD", b"ref: refs/heads/trunk\n")])).await;
assert_eq!(
std::fs::read_to_string(dir.path().join("repo/HEAD")).unwrap(),
"ref: refs/heads/trunk\n"
);
}
#[tokio::test]
async fn an_absolute_entry_is_skipped_and_the_rest_is_extracted() {
let dir = extract_to_temp(&tar_gz(&[
("/etc/passwd", b"root::0:0:::\n"),
("repo/HEAD", b"ref: refs/heads/trunk\n"),
]))
.await;
assert_eq!(entries_under(dir.path()), ["repo", "repo/HEAD"]);
}
#[tokio::test]
async fn a_name_holding_two_dots_is_not_a_traversal() {
let dir = extract_to_temp(&tar_gz(&[("repo/v1../a..b", b"kept\n")])).await;
assert_eq!(
std::fs::read_to_string(dir.path().join("repo/v1../a..b")).unwrap(),
"kept\n"
);
}
#[tokio::test]
async fn an_entry_that_climbs_out_writes_nothing_anywhere() {
for name in ["../x", "repo/../../x", "repo/.."] {
let root = tempfile::tempdir().unwrap();
let dest = root.path().join("a/b");
std::fs::create_dir_all(&dest).unwrap();
extract_into(
&hostile_tar_gz(tar::EntryType::Regular, name, "", b"escaped\n"),
&dest,
)
.await;
assert_eq!(entries_under(root.path()), ["a", "a/b"], "{name}");
}
}
#[tokio::test]
async fn a_link_entry_writes_nothing_anywhere() {
let outside = tempfile::tempdir().unwrap();
let target = outside.path().join("target");
std::fs::write(&target, b"outside\n").unwrap();
let target = target.to_string_lossy();
for kind in [tar::EntryType::Symlink, tar::EntryType::Link] {
let dest = tempfile::tempdir().unwrap();
extract_into(
&hostile_tar_gz(kind, "repo/link", &target, b""),
dest.path(),
)
.await;
assert!(entries_under(dest.path()).is_empty(), "{kind:?}");
assert_eq!(
std::fs::read_to_string(outside.path().join("target")).unwrap(),
"outside\n"
);
}
}
#[tokio::test]
async fn a_checksum_that_does_not_match_is_a_corrupt_archive() {
let mut archive = tar_gz(&[("repo/HEAD", b"ref: refs/heads/trunk\n")]);
let crc = archive.len() - 8;
for byte in archive.get_mut(crc..crc + 4).unwrap() {
*byte ^= 0xFF;
}
let dir = tempfile::tempdir().unwrap();
let outcome = extract(archive.as_slice(), dir.path()).await;
assert!(
matches!(outcome, Err(ExtractError::Corrupt(_))),
"{outcome:?}"
);
}
#[tokio::test]
async fn a_stream_cut_short_is_a_truncated_archive() {
let archive = tar_gz(&[("repo/HEAD", b"ref: refs/heads/trunk\n")]);
let dir = tempfile::tempdir().unwrap();
let outcome = extract(archive.get(..archive.len() - 4).unwrap(), dir.path()).await;
assert!(
matches!(outcome, Err(ExtractError::Truncated(_))),
"{outcome:?}"
);
}
#[tokio::test]
async fn a_member_under_a_file_is_an_io_failure() {
let archive = tar_gz(&[
("repo", b"a file"),
("repo/HEAD", b"ref: refs/heads/trunk\n"),
]);
let dir = tempfile::tempdir().unwrap();
let outcome = extract(archive.as_slice(), dir.path()).await;
assert!(matches!(outcome, Err(ExtractError::Io(_))), "{outcome:?}");
}
#[cfg(unix)]
#[tokio::test]
async fn an_entry_keeps_no_setuid_bit_nor_write_for_others() {
use std::os::unix::fs::PermissionsExt;
let mut builder = tar::Builder::new(GzEncoder::new(Vec::new(), Compression::default()));
let mut header = tar::Header::new_gnu();
header.set_size(0);
header.set_mode(0o4777);
builder
.append_data(&mut header, "repo/run", std::io::empty())
.unwrap();
let dir = extract_to_temp(&builder.into_inner().unwrap().finish().unwrap()).await;
let mode = std::fs::metadata(dir.path().join("repo/run"))
.unwrap()
.permissions()
.mode();
assert_eq!(mode & 0o7777, 0o755);
}
#[tokio::test]
async fn the_entry_mtime_is_preserved() {
let dir = extract_to_temp(&tar_gz(&[("repo/HEAD", b"ref: refs/heads/trunk\n")])).await;
let modified = std::fs::metadata(dir.path().join("repo/HEAD"))
.unwrap()
.modified()
.unwrap();
assert_eq!(
modified,
SystemTime::UNIX_EPOCH
.checked_add(Duration::from_secs(MTIME))
.unwrap()
);
}
}