fluidattacks_core_git/

download.rs

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};

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"];
// Inside `objects/`, these point git at object stores elsewhere on disk.
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 {
    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 {
    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)?;
    // As the python's aiohttp, the environment's proxy is ignored.
    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());
        }
    }
    // The producers pack a single top-level directory.
    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())
    })
}

// A `.git` file or link would redirect git to a repository outside the archive.
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)
        .and_then(|()| super::fs::delete_gitignored_files(repo, git_ignore))
        .map_err(|error| DownloadError::Prune {
            message: format!("{error:#}"),
            errno: error
                .chain()
                .find_map(|cause| cause.downcast_ref::<io::Error>()?.raw_os_error()),
        })
}

// One swap with no await inside, so `dest` holds the previous tree or the
// new one, never neither.
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())),
        // As the python, which could not remove a file or a link there.
        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)),
    }
}

// What `download_and_extract_repo` was asked for, with the buffer size resolved.
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)?;
    // Beside the destination, so the repository swaps into place, and
    // whatever fails or is cancelled leaves the destination as it was.
    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)
}

/// Leaves the repository archived at `url` at `dest`, replacing what was there.
///
/// The archive holds only `.git`, so the working tree is restored with a
/// `git reset --hard`. Pass `checkout = false` to keep just the history: peak
/// disk stays at the size of the archive plus `.git`, which is what a consumer
/// that walks commits needs. `git_ignore` prunes working-tree files and has no
/// effect on a history-only download. `buffer_size` bounds a single write of
/// the archive to disk; `None` takes 64 KiB, and anything above 16 MiB is
/// taken as 16 MiB.
///
/// The archive is untrusted, so only its history is kept and its config is
/// replaced. Everything is built in a directory beside `dest`, so a failed or
/// cancelled call leaves `dest` as it was. Nothing is retried. A relative
/// `dest` is resolved against the working directory once, when the call
/// starts, and the working directory is never changed.
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;

    // Answers every request with `archive`, as a bucket would.
    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) {}
    }

    // Answers every connection with `raw` and keeps it open, so only a
    // timeout of the client ends the request.
    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() {
        // The connect covers the TLS handshake, which this server never answers.
        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:?}"
        );
    }

    #[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();

        // A HEAD with no objects behind it, which git cannot reset to.
        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 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")
        );
    }
}