use rustix::fs::{renameat_with, RenameFlags, CWD};
use std::io;
use std::num::NonZeroUsize;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::io::{AsyncWriteExt, BufWriter};
use tracing::warn;
use crate::errors::{DownloadError, ExtractError, FetchError, PruneError};
const TIMEOUTS: Timeouts = Timeouts {
connect: Duration::from_secs(30),
total: Duration::from_hours(1),
};
const DEFAULT_DOWNLOAD_BUFFER_SIZE: usize = 64 * 1024;
const MAX_DOWNLOAD_BUFFER_SIZE: usize = 16 * 1024 * 1024;
const WORK_PREFIX: &str = ".fluidattacks_";
const ARCHIVE_NAME: &str = "repo.tar.gz";
const EXTRACTED_NAME: &str = "extracted";
const GIT_DIR_DATA: [&str; 6] = ["HEAD", "index", "objects", "packed-refs", "refs", "shallow"];
const OBJECT_STORE_POINTERS: [&str; 2] = ["alternates", "http-alternates"];
const TRUSTED_CONFIG: &str = "[core]\nrepositoryformatversion = 0\nbare = false\n";
struct Timeouts {
connect: Duration,
total: Duration,
}
fn resolve_buffer_size(buffer_size: Option<NonZeroUsize>) -> usize {
buffer_size.map_or(DEFAULT_DOWNLOAD_BUFFER_SIZE, |size| {
size.get().min(MAX_DOWNLOAD_BUFFER_SIZE)
})
}
fn parse_url(url: &str) -> Result<url::Url, FetchError> {
let parsed = url::Url::parse(url).map_err(|error| FetchError::InvalidUrl {
url: url.to_owned(),
reason: error.to_string(),
})?;
if !matches!(parsed.scheme(), "http" | "https") {
return Err(FetchError::NonHttpUrl(url.to_owned()));
}
Ok(parsed)
}
fn request_error(error: reqwest::Error) -> FetchError {
let error = error.without_url();
if error.is_connect() {
return if error.is_timeout() {
FetchError::ConnectTimeout(error)
} else {
FetchError::Connect(error)
};
}
if error.is_timeout() {
return FetchError::Timeout(error);
}
let mut cause: Option<&(dyn std::error::Error + 'static)> = Some(&error);
let mut socket = None;
while let Some(current) = cause {
if let Some(hyper) = current.downcast_ref::<hyper::Error>() {
if hyper.is_incomplete_message() {
return FetchError::Disconnected(error);
}
if hyper.is_parse() {
return FetchError::Malformed(error);
}
}
if let Some(io) = current.downcast_ref::<io::Error>() {
socket = Some(io);
}
cause = current.source();
}
match socket {
Some(io) => FetchError::Socket {
reset: io.kind() == io::ErrorKind::ConnectionReset,
errno: io.raw_os_error(),
source: error,
},
None => FetchError::Other(error),
}
}
fn body_error(error: reqwest::Error) -> FetchError {
let error = error.without_url();
if error.is_timeout() {
FetchError::BodyTimeout(error)
} else {
FetchError::BodyCut(error)
}
}
async fn fetch_to(
url: &str,
path: &Path,
buffer_size: usize,
timeouts: Timeouts,
) -> Result<(), DownloadError> {
let url = parse_url(url)?;
let client = reqwest::Client::builder()
.no_proxy()
.timeout(timeouts.total)
.connect_timeout(timeouts.connect)
.build()
.map_err(FetchError::Other)?;
let mut response = client.get(url).send().await.map_err(request_error)?;
let status = response.status();
if status != reqwest::StatusCode::OK {
return Err(FetchError::Status(status).into());
}
let file = tokio::fs::File::create(path)
.await
.map_err(DownloadError::Write)?;
let mut writer = BufWriter::with_capacity(buffer_size, file);
while let Some(chunk) = response.chunk().await.map_err(body_error)? {
for slice in chunk.chunks(buffer_size) {
writer
.write_all(slice)
.await
.map_err(DownloadError::Write)?;
}
}
writer.flush().await.map_err(DownloadError::Write)?;
Ok(())
}
async fn extracted_dir(extracted: &Path) -> Result<PathBuf, ExtractError> {
let mut entries = tokio::fs::read_dir(extracted)
.await
.map_err(ExtractError::Io)?;
let mut dirs = Vec::new();
while let Some(entry) = entries.next_entry().await.map_err(ExtractError::Io)? {
if entry.file_type().await.is_ok_and(|kind| kind.is_dir()) {
dirs.push(entry.path());
}
}
dirs.sort();
if dirs.len() > 1 {
warn!(
count = dirs.len(),
"multiple directories found in archive, using the first one"
);
}
dirs.into_iter().next().ok_or_else(|| {
ExtractError::NoRepository("no directory found in the extracted archive".to_owned())
})
}
async fn keep_only_history(repo: &Path) -> Result<(), ExtractError> {
let git_dir = repo.join(".git");
let is_dir = tokio::fs::symlink_metadata(&git_dir)
.await
.is_ok_and(|meta| meta.is_dir());
if !is_dir {
return Err(ExtractError::NoRepository(
"no .git directory found in the extracted archive".to_owned(),
));
}
let mut entries = tokio::fs::read_dir(&git_dir)
.await
.map_err(ExtractError::Io)?;
while let Some(entry) = entries.next_entry().await.map_err(ExtractError::Io)? {
if GIT_DIR_DATA.iter().any(|kept| entry.file_name() == *kept) {
continue;
}
let path = entry.path();
let removed = if entry.file_type().await.is_ok_and(|kind| kind.is_dir()) {
tokio::fs::remove_dir_all(&path).await
} else {
tokio::fs::remove_file(&path).await
};
removed.map_err(ExtractError::Io)?;
}
for pointer in OBJECT_STORE_POINTERS {
match tokio::fs::remove_file(git_dir.join("objects/info").join(pointer)).await {
Err(error) if error.kind() != io::ErrorKind::NotFound => {
return Err(ExtractError::Io(error));
}
_ => {}
}
}
tokio::fs::write(git_dir.join("config"), TRUSTED_CONFIG)
.await
.map_err(ExtractError::Io)
}
fn prune(repo: &Path, git_ignore: &[String]) -> Result<(), DownloadError> {
super::fs::remove_symlinks(repo).map_err(PruneError::from)?;
super::fs::delete_gitignored_files(repo, git_ignore)?;
Ok(())
}
fn place(repo: &Path, dest: &Path) -> Result<(), DownloadError> {
match std::fs::symlink_metadata(dest) {
Ok(meta) if meta.is_dir() => renameat_with(CWD, repo, CWD, dest, RenameFlags::EXCHANGE)
.map_err(|errno| DownloadError::Place(errno.into())),
Ok(_) => Err(DownloadError::Place(io::Error::from(
io::ErrorKind::NotADirectory,
))),
Err(error) if error.kind() == io::ErrorKind::NotFound => {
std::fs::rename(repo, dest).map_err(DownloadError::Place)
}
Err(error) => Err(DownloadError::Place(error)),
}
}
struct Request<'a> {
url: &'a str,
git_ignore: &'a [String],
buffer_size: usize,
checkout: bool,
timeouts: Timeouts,
}
async fn build_repo(request: Request<'_>, work: &Path) -> Result<PathBuf, DownloadError> {
let archive = work.join(ARCHIVE_NAME);
fetch_to(request.url, &archive, request.buffer_size, request.timeouts).await?;
let extracted = work.join(EXTRACTED_NAME);
tokio::fs::create_dir(&extracted)
.await
.map_err(ExtractError::Io)?;
let file = tokio::fs::File::open(&archive)
.await
.map_err(ExtractError::Io)?;
super::tar::extract(file, &extracted).await?;
let _ = tokio::fs::remove_file(&archive).await;
let repo = extracted_dir(&extracted).await?;
keep_only_history(&repo).await?;
if request.checkout {
super::cli::reset_hard(&repo).await?;
prune(&repo, request.git_ignore)?;
}
Ok(repo)
}
async fn download_with(request: Request<'_>, dest: &Path) -> Result<(), DownloadError> {
let dest = &std::path::absolute(dest).map_err(DownloadError::Write)?;
let parent = dest.parent().ok_or_else(|| {
DownloadError::Write(io::Error::new(
io::ErrorKind::InvalidInput,
"the destination has no parent directory",
))
})?;
tokio::fs::create_dir_all(parent)
.await
.map_err(DownloadError::Write)?;
let work = tempfile::Builder::new()
.prefix(WORK_PREFIX)
.tempdir_in(parent)
.map_err(DownloadError::Place)?;
let repo = build_repo(request, work.path()).await?;
place(&repo, dest)
}
pub async fn download_and_extract_repo(
url: &str,
dest: &Path,
git_ignore: &[String],
buffer_size: Option<NonZeroUsize>,
checkout: bool,
) -> Result<(), DownloadError> {
let request = Request {
url,
git_ignore,
buffer_size: resolve_buffer_size(buffer_size),
checkout,
timeouts: TIMEOUTS,
};
download_with(request, dest).await
}
#[cfg(test)]
mod tests {
use super::{
download_and_extract_repo, extracted_dir, fetch_to, place, resolve_buffer_size, Timeouts,
DEFAULT_DOWNLOAD_BUFFER_SIZE, MAX_DOWNLOAD_BUFFER_SIZE, TRUSTED_CONFIG,
};
use crate::errors::{DownloadError, ExtractError, FetchError};
use flate2::write::GzEncoder;
use flate2::Compression;
use std::num::NonZeroUsize;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
async fn serve(archive: Vec<u8>) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
tokio::spawn(async move {
while let Ok((mut stream, _)) = listener.accept().await {
let mut request = [0u8; 1024];
let _ = stream.read(&mut request).await;
let head = format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
archive.len()
);
let _ = stream.write_all(head.as_bytes()).await;
let _ = stream.write_all(&archive).await;
}
});
format!("http://{address}/repo.tar.gz")
}
async fn download(archive: Vec<u8>, dest: &Path, checkout: bool) -> Result<(), DownloadError> {
let url = serve(archive).await;
download_and_extract_repo(&url, dest, &[], None, checkout).await
}
fn tar_gz(name: &str, contents: &[u8]) -> Vec<u8> {
let mut builder = tar::Builder::new(GzEncoder::new(Vec::new(), Compression::default()));
let mut header = tar::Header::new_gnu();
header.set_size(u64::try_from(contents.len()).unwrap());
header.set_mode(0o644);
builder.append_data(&mut header, name, contents).unwrap();
builder.into_inner().unwrap().finish().unwrap()
}
#[tokio::test]
async fn the_archive_replaces_an_existing_destination_and_leaves_nothing_beside_it() {
let tmp = tempfile::tempdir().unwrap();
let dest = tmp.path().join("dest");
std::fs::create_dir_all(&dest).unwrap();
std::fs::write(dest.join("stale"), b"old").unwrap();
download(
tar_gz("repo/.git/HEAD", b"ref: refs/heads/trunk\n"),
&dest,
false,
)
.await
.unwrap();
assert_eq!(
std::fs::read_to_string(dest.join(".git/HEAD")).unwrap(),
"ref: refs/heads/trunk\n"
);
assert!(!dest.join("stale").exists());
assert_eq!(std::fs::read_dir(tmp.path()).unwrap().count(), 1);
}
#[tokio::test]
async fn an_archive_without_a_directory_is_rejected() {
let tmp = tempfile::tempdir().unwrap();
let dest = tmp.path().join("dest");
let outcome = download(tar_gz("HEAD", b"ref: refs/heads/trunk\n"), &dest, false).await;
assert!(
matches!(
outcome,
Err(DownloadError::Extract(ExtractError::NoRepository(_)))
),
"{outcome:?}"
);
assert!(!dest.exists());
}
const SHORT: Duration = Duration::from_secs(1);
const LONG: Duration = Duration::from_secs(30);
async fn answer_and_stall(mut stream: tokio::net::TcpStream, raw: &'static [u8]) {
let mut request = [0u8; 1024];
let _ = stream.read(&mut request).await;
let _ = stream.write_all(raw).await;
while stream.read(&mut request).await.is_ok_and(|read| read > 0) {}
}
async fn serve_and_stall(raw: &'static [u8]) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
tokio::spawn(async move {
while let Ok((stream, _)) = listener.accept().await {
tokio::spawn(answer_and_stall(stream, raw));
}
});
format!("http://{address}/repo.tar.gz")
}
async fn fetch_with(url: &str, timeouts: Timeouts) -> Result<(), DownloadError> {
let dir = tempfile::tempdir().unwrap();
fetch_to(
url,
&dir.path().join("repo.tar.gz"),
DEFAULT_DOWNLOAD_BUFFER_SIZE,
timeouts,
)
.await
}
#[tokio::test]
async fn a_connect_that_never_completes_is_a_connect_timeout() {
let url = serve_and_stall(b"")
.await
.replacen("http://", "https://", 1);
let outcome = fetch_with(
&url,
Timeouts {
connect: SHORT,
total: LONG,
},
)
.await;
assert!(
matches!(
outcome,
Err(DownloadError::Fetch(FetchError::ConnectTimeout(_)))
),
"{outcome:?}"
);
}
#[tokio::test]
async fn a_response_that_never_starts_is_a_timeout_before_the_headers() {
let url = serve_and_stall(b"").await;
let outcome = fetch_with(
&url,
Timeouts {
connect: LONG,
total: SHORT,
},
)
.await;
assert!(
matches!(outcome, Err(DownloadError::Fetch(FetchError::Timeout(_)))),
"{outcome:?}"
);
}
#[tokio::test]
async fn a_body_that_stops_arriving_is_a_timeout_reading_the_body() {
let url =
serve_and_stall(b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\n\r\n\x1f\x8b\x08\x00")
.await;
let outcome = fetch_with(
&url,
Timeouts {
connect: LONG,
total: SHORT,
},
)
.await;
assert!(
matches!(
outcome,
Err(DownloadError::Fetch(FetchError::BodyTimeout(_)))
),
"{outcome:?}"
);
}
#[tokio::test]
async fn a_failed_fetch_never_carries_the_query_of_the_url() {
let refused = {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
format!("http://{}/repo.tar.gz", listener.local_addr().unwrap())
};
let silent = serve_and_stall(b"").await;
let cut = serve_and_stall(b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\n\r\nx").await;
for url in [refused, silent, cut] {
let outcome = fetch_with(
&format!("{url}?X-Amz-Signature=secret"),
Timeouts {
connect: SHORT,
total: SHORT,
},
)
.await;
assert!(!format!("{outcome:?}").contains("secret"), "{outcome:?}");
let error = outcome.unwrap_err();
assert!(!error.to_string().contains("secret"), "{error}");
}
}
#[test]
fn an_absent_buffer_size_takes_the_default_of_the_python() {
assert_eq!(resolve_buffer_size(None), DEFAULT_DOWNLOAD_BUFFER_SIZE);
assert_eq!(resolve_buffer_size(NonZeroUsize::new(8 * 1024)), 8 * 1024);
}
#[test]
fn a_buffer_size_above_the_maximum_is_clamped() {
assert_eq!(
resolve_buffer_size(NonZeroUsize::new(MAX_DOWNLOAD_BUFFER_SIZE + 1)),
MAX_DOWNLOAD_BUFFER_SIZE
);
}
fn git(repo: &Path, args: &[&str]) {
let status = std::process::Command::new("git")
.arg("-C")
.arg(repo)
.args([
"-c",
"user.email=dev@fluidattacks.com",
"-c",
"user.name=dev",
])
.args(args)
.status()
.unwrap();
assert!(status.success());
}
fn hostile_archive(dir: &Path, key: &str) -> (Vec<u8>, PathBuf) {
let marker = dir.join("pwned");
let payload = dir.join("payload.sh");
std::fs::write(
&payload,
format!("#!/bin/sh\ntouch {}\ncat\n", marker.display()),
)
.unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&payload, std::fs::Permissions::from_mode(0o755)).unwrap();
}
let source = dir.join("source");
std::fs::create_dir_all(&source).unwrap();
git(&source, &["init", "-q"]);
std::fs::write(source.join(".gitattributes"), b"* filter=pwn\n").unwrap();
std::fs::write(source.join("tracked.txt"), b"content\n").unwrap();
git(&source, &["add", "-A"]);
git(&source, &["commit", "-q", "-m", "add tracked file"]);
git(&source, &["config", key, &payload.to_string_lossy()]);
std::fs::create_dir_all(source.join(".git/objects/info")).unwrap();
std::fs::write(
source.join(".git/objects/info/alternates"),
format!("{}\n", dir.display()),
)
.unwrap();
let mut builder = tar::Builder::new(GzEncoder::new(Vec::new(), Compression::default()));
builder
.append_dir_all("source/.git", source.join(".git"))
.unwrap();
let archive = builder.into_inner().unwrap().finish().unwrap();
std::fs::remove_dir_all(&source).unwrap();
(archive, marker)
}
#[tokio::test]
async fn a_restored_archive_runs_nothing_it_configures() {
for key in ["filter.pwn.smudge", "filter.pwn.process", "core.fsmonitor"] {
let tmp = tempfile::tempdir().unwrap();
let (archive, marker) = hostile_archive(tmp.path(), key);
let dest = tmp.path().join("dest");
download(archive, &dest, true).await.unwrap();
assert!(!marker.exists(), "{key} ran");
assert_eq!(
std::fs::read_to_string(dest.join("tracked.txt")).unwrap(),
"content\n"
);
}
}
#[tokio::test]
async fn only_the_history_of_the_archive_is_kept() {
let tmp = tempfile::tempdir().unwrap();
let (archive, _) = hostile_archive(tmp.path(), "core.fsmonitor");
let dest = tmp.path().join("dest");
download(archive, &dest, false).await.unwrap();
let git_dir = dest.join(".git");
assert_eq!(
std::fs::read_to_string(git_dir.join("config")).unwrap(),
TRUSTED_CONFIG
);
assert!(!git_dir.join("hooks").exists());
assert!(!git_dir.join("objects/info/alternates").exists());
assert!(git_dir.join("HEAD").is_file());
}
#[tokio::test]
async fn a_git_file_in_the_archive_is_rejected() {
let tmp = tempfile::tempdir().unwrap();
let dest = tmp.path().join("dest");
let outcome = download(
tar_gz("repo/.git", b"gitdir: /elsewhere/.git\n"),
&dest,
false,
)
.await;
assert!(
matches!(
outcome,
Err(DownloadError::Extract(ExtractError::NoRepository(_)))
),
"{outcome:?}"
);
assert!(!dest.exists());
}
#[tokio::test]
async fn a_failed_reset_keeps_the_previous_destination() {
let tmp = tempfile::tempdir().unwrap();
let dest = tmp.path().join("dest");
std::fs::create_dir_all(&dest).unwrap();
std::fs::write(dest.join("previous"), b"old").unwrap();
let outcome = download(
tar_gz("repo/.git/HEAD", b"ref: refs/heads/trunk\n"),
&dest,
true,
)
.await;
assert!(
matches!(outcome, Err(DownloadError::Reset(_))),
"{outcome:?}"
);
assert_eq!(std::fs::read(dest.join("previous")).unwrap(), b"old");
assert_eq!(std::fs::read_dir(tmp.path()).unwrap().count(), 1);
}
#[test]
fn a_failed_swap_keeps_the_previous_destination() {
let tmp = tempfile::tempdir().unwrap();
let dest = tmp.path().join("dest");
std::fs::create_dir_all(&dest).unwrap();
std::fs::write(dest.join("previous"), b"old").unwrap();
let outcome = place(&tmp.path().join("missing"), &dest);
assert!(
matches!(outcome, Err(DownloadError::Place(_))),
"{outcome:?}"
);
assert_eq!(std::fs::read(dest.join("previous")).unwrap(), b"old");
}
#[tokio::test]
async fn a_work_directory_that_cannot_be_made_is_a_place_failure() {
use std::os::unix::fs::PermissionsExt;
let tmp = tempfile::tempdir().unwrap();
let parent = tmp.path().join("read-only");
std::fs::create_dir(&parent).unwrap();
std::fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o500)).unwrap();
if std::fs::create_dir(parent.join("probe")).is_ok() {
return;
}
let outcome =
download_and_extract_repo("http://127.0.0.1:9/", &parent.join("root"), &[], None, true)
.await;
std::fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o700)).unwrap();
assert!(
matches!(outcome, Err(DownloadError::Place(_))),
"{outcome:?}"
);
}
#[tokio::test]
async fn the_first_directory_is_taken_in_a_stable_order() {
let tmp = tempfile::tempdir().unwrap();
std::fs::write(tmp.path().join("stray"), b"").unwrap();
std::fs::create_dir_all(tmp.path().join("b_repo")).unwrap();
std::fs::create_dir_all(tmp.path().join("a_repo")).unwrap();
assert_eq!(
extracted_dir(tmp.path()).await.unwrap(),
tmp.path().join("a_repo")
);
}
}