sausage/userland/git/src/http.rs

//! Synchronous wasi:http backend. Response packs stream; replayable request bodies spool to the current directory.
//! No executor/block_on, worker threads, signal handling, or secret-bearing diagnostics.
use gix_error::ErrorExt;
use gix_transport::client::blocking_io::http::{GetResponse, Http, PostBodyDataKind, PostResponse};
use std::{
    cell::RefCell,
    fs::File,
    io::{self, BufRead, BufReader, Cursor, Read, Seek, Write},
    rc::Rc,
};
use wasi::http::{outgoing_handler, types::*};

type Shared = Rc<RefCell<Pending>>;
pub struct WasiHttp {
    timeout: Option<u64>,
    redirect: Rc<RefCell<Option<String>>>,
}
impl Default for WasiHttp {
    fn default() -> Self {
        Self {
            timeout: None,
            redirect: Default::default(),
        }
    }
}
fn err(text: impl Into<String>) -> io::Error {
    io::Error::other(text.into())
}

/// The host injects credentials. Git accepts credential-free HTTP and HTTPS URLs.
pub fn clean_url(input: &str) -> io::Result<url::Url> {
    let u = url::Url::parse(input).map_err(|_| err("invalid HTTP URL"))?;
    if !matches!(u.scheme(), "http" | "https")
        || u.host_str().is_none()
        || !u.username().is_empty()
        || u.password().is_some()
        || u.query().is_some()
        || u.fragment().is_some()
    {
        return Err(err(
            "use an HTTP or HTTPS repository URL without credentials, query, or fragment",
        ));
    }
    Ok(u)
}

fn status_error(status: u16, host: &str) -> io::Error {
    // Other rather than PermissionDenied prevents external credential helpers.
    err(format!("HTTP {status} from {host}"))
}

struct Incoming {
    stream: Option<wasi::io::streams::InputStream>,
    body: Option<IncomingBody>,
}
impl Read for Incoming {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }
        let Some(stream) = self.stream.as_ref() else {
            return Ok(0);
        };
        loop {
            match stream.blocking_read(buf.len().min(64 * 1024) as u64) {
                Ok(bytes) if bytes.is_empty() => {
                    stream.subscribe().block();
                    continue;
                }
                Ok(bytes) => {
                    buf[..bytes.len()].copy_from_slice(&bytes);
                    return Ok(bytes.len());
                }
                Err(wasi::io::streams::StreamError::Closed) => {
                    self.stream.take();
                    if let Some(body) = self.body.take() {
                        let trailers = IncomingBody::finish(body);
                        trailers.subscribe().block();
                        match trailers.get() {
                            Some(Ok(Ok(_))) => {}
                            _ => return Err(err("HTTP response body failed")),
                        }
                    }
                    return Ok(0);
                }
                Err(_) => return Err(err("HTTP response read failed or timed out")),
            }
        }
    }
}
impl Drop for Incoming {
    fn drop(&mut self) {
        self.stream.take();
        self.body.take();
    }
}

// State is shared only within one synchronous request, not across handlers/processes.
struct Pending {
    timeout: Option<u64>,
    url: String,
    base: String,
    headers: Vec<String>,
    spool: Option<File>,
    path: Option<std::path::PathBuf>,
    writer_open: bool,
    response_headers: Option<Cursor<Vec<u8>>>,
    response: Option<BufReader<Box<dyn Read>>>,
    redirect: Rc<RefCell<Option<String>>>,
}
impl Drop for Pending {
    fn drop(&mut self) {
        self.spool.take();
        if let Some(path) = self.path.take() {
            let _ = std::fs::remove_file(&path);
        }
    }
}
impl Pending {
    fn start(&mut self) -> io::Result<()> {
        if self.response.is_some() {
            return Ok(());
        }
        if self.writer_open {
            return Err(err(
                "HTTP request body must be closed before reading response",
            ));
        }
        let mut target = url::Url::parse(&self.url).map_err(|_| err("invalid HTTP request URL"))?;
        for attempt in 0..2 {
            let response = self.send(&target)?;
            let status = response.status();
            if matches!(status, 301 | 302 | 303 | 307 | 308) {
                if attempt == 1 {
                    return Err(err("HTTP redirect limit (one) exceeded"));
                }
                let locations = response.headers().get(&"location".to_string());
                let location = locations
                    .first()
                    .and_then(|v| std::str::from_utf8(v).ok())
                    .ok_or_else(|| err("HTTP redirect has no valid Location"))?;
                let next = target
                    .join(location)
                    .map_err(|_| err("invalid HTTP redirect"))?;
                if !matches!(next.scheme(), "http" | "https")
                    || (target.scheme() == "https" && next.scheme() != "https")
                    || !next.username().is_empty()
                    || next.password().is_some()
                {
                    return Err(err("redirect requires an HTTPS URL without credentials"));
                }
                // Preserve the git endpoint suffix so subsequent POSTs use the new repository base.
                let tail = self
                    .url
                    .strip_prefix(&self.base)
                    .ok_or_else(|| err("invalid HTTP base"))?;
                let next_text = next.as_str();
                let base = next_text
                    .strip_suffix(tail)
                    .ok_or_else(|| err("redirect changed the Git endpoint suffix"))?;
                *self.redirect.borrow_mut() = Some(base.to_owned());
                target = next;
                continue;
            }
            if !(200..300).contains(&status) {
                return Err(status_error(status, target.host_str().unwrap_or("unknown")));
            }
            let mut headers = Vec::new();
            for (name, value) in response.headers().entries() {
                headers.extend_from_slice(name.as_bytes());
                headers.extend_from_slice(b": ");
                headers.extend_from_slice(&value);
                headers.push(b'\n');
            }
            let body = response
                .consume()
                .map_err(|_| err("could not consume HTTP response"))?;
            let stream = body
                .stream()
                .map_err(|_| err("could not open HTTP response stream"))?;
            self.response_headers = Some(Cursor::new(headers));
            self.response = Some(BufReader::new(Box::new(Incoming {
                stream: Some(stream),
                body: Some(body),
            })));
            return Ok(());
        }
        Err(err("HTTP redirect failed"))
    }
    fn send(&mut self, target: &url::Url) -> io::Result<IncomingResponse> {
        let fields = Fields::new();
        for header in &self.headers {
            let (name, value) = header
                .split_once(':')
                .ok_or_else(|| err("invalid HTTP header"))?;
            // Never accept Authorization from Git config or URL.
            if name.eq_ignore_ascii_case("authorization") {
                continue;
            }
            fields
                .append(&name.to_string(), &value.trim().as_bytes().to_vec())
                .map_err(|_| err("invalid HTTP header"))?;
        }
        if let Some(file) = self.spool.as_ref() {
            fields
                .append(
                    "content-length",
                    &file.metadata()?.len().to_string().into_bytes(),
                )
                .map_err(|_| err("invalid content length"))?;
        }
        let request = OutgoingRequest::new(fields);
        request
            .set_method(&if self.spool.is_some() {
                Method::Post
            } else {
                Method::Get
            })
            .map_err(|_| err("invalid HTTP method"))?;
        let scheme = if target.scheme() == "https" {
            Scheme::Https
        } else {
            Scheme::Http
        };
        request
            .set_scheme(Some(&scheme))
            .map_err(|_| err("invalid HTTP scheme"))?;
        let authority = &target[url::Position::BeforeHost..url::Position::AfterPort];
        request
            .set_authority(Some(authority))
            .map_err(|_| err("invalid HTTP authority"))?;
        request
            .set_path_with_query(Some(
                &target[url::Position::BeforePath..url::Position::AfterQuery],
            ))
            .map_err(|_| err("invalid HTTP path"))?;
        let body = request
            .body()
            .map_err(|_| err("could not open HTTP request body"))?;
        let options = RequestOptions::new();
        options
            .set_connect_timeout(Some(
                self.timeout.map_or(120_000_000_000, |s| s * 1_000_000_000),
            ))
            .map_err(|_| err("connect timeout unsupported"))?;
        options
            .set_first_byte_timeout(Some(
                self.timeout.map_or(120_000_000_000, |s| s * 1_000_000_000),
            ))
            .map_err(|_| err("read timeout unsupported"))?;
        options
            .set_between_bytes_timeout(Some(
                self.timeout.map_or(120_000_000_000, |s| s * 1_000_000_000),
            ))
            .map_err(|_| err("read timeout unsupported"))?;
        let future = outgoing_handler::handle(request, Some(options))
            .map_err(|_| err("HTTP connection failed"))?;
        if let Some(file) = self.spool.as_mut() {
            file.rewind()?;
            let stream = body.write().map_err(|_| err("could not write HTTP body"))?;
            let mut bytes = [0; 4096]; // WASI blocking_write_and_flush accepts at most 4096 bytes
            loop {
                let n = file.read(&mut bytes)?;
                if n == 0 {
                    break;
                }
                stream
                    .blocking_write_and_flush(&bytes[..n])
                    .map_err(|_| err("HTTP request write failed or timed out"))?;
            }
            drop(stream);
        }
        OutgoingBody::finish(body, None).map_err(|_| err("HTTP request finish failed"))?;
        future.subscribe().block();
        future
            .get()
            .ok_or_else(|| err("HTTP response not ready"))?
            .map_err(|_| err("HTTP response already consumed"))?
            .map_err(|code| err(format!("HTTP request failed: {code:?}")))
    }
}

pub struct Headers(Shared);
pub struct Response(Shared, Vec<u8>, usize);
pub struct PostBody(Shared);
impl Write for PostBody {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.0
            .borrow_mut()
            .spool
            .as_mut()
            .ok_or_else(|| err("request closed"))?
            .write(bytes)
    }
    fn flush(&mut self) -> io::Result<()> {
        self.0
            .borrow_mut()
            .spool
            .as_mut()
            .ok_or_else(|| err("request closed"))?
            .flush()
    }
}
impl Drop for PostBody {
    fn drop(&mut self) {
        self.0.borrow_mut().writer_open = false;
    }
}
impl Read for Headers {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let mut p = self.0.borrow_mut();
        p.start()?;
        p.response_headers
            .as_mut()
            .ok_or_else(|| err("missing headers"))?
            .read(buf)
    }
}
impl Read for Response {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let available = self.fill_buf()?;
        let n = available.len().min(buf.len());
        buf[..n].copy_from_slice(&available[..n]);
        self.consume(n);
        Ok(n)
    }
}
impl BufRead for Response {
    fn fill_buf(&mut self) -> io::Result<&[u8]> {
        if self.2 == self.1.len() {
            let mut p = self.0.borrow_mut();
            p.start()?;
            self.1.resize(16 * 1024, 0);
            let n = p
                .response
                .as_mut()
                .ok_or_else(|| err("missing response"))?
                .read(&mut self.1)?;
            self.1.truncate(n);
            self.2 = 0;
        }
        Ok(&self.1[self.2..])
    }
    fn consume(&mut self, n: usize) {
        self.2 = (self.2 + n).min(self.1.len());
    }
}
impl WasiHttp {
    fn pending(
        &self,
        url: &str,
        base: &str,
        headers: impl IntoIterator<Item = impl AsRef<str>>,
        post: bool,
    ) -> io::Result<Shared> {
        let (spool, path) = if post {
            static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
            let dir = ".";
            let path = std::path::PathBuf::from(format!(
                "{dir}/git-http-{}",
                NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
            ));
            (
                Some(
                    File::options()
                        .read(true)
                        .write(true)
                        .create_new(true)
                        .open(&path)?,
                ),
                Some(path),
            )
        } else {
            (None, None)
        };
        Ok(Rc::new(RefCell::new(Pending {
            timeout: self.timeout,
            url: url.into(),
            base: base.into(),
            headers: headers.into_iter().map(|s| s.as_ref().to_owned()).collect(),
            spool,
            path,
            writer_open: post,
            response_headers: None,
            response: None,
            redirect: self.redirect.clone(),
        })))
    }
}
impl Http for WasiHttp {
    type Headers = BufReader<Headers>;
    type ResponseBody = Response;
    type PostBody = PostBody;
    fn get(
        &mut self,
        url: &str,
        base: &str,
        headers: impl IntoIterator<Item = impl AsRef<str>>,
    ) -> gix_error::Result<GetResponse<Self::Headers, Response>> {
        let p = self
            .pending(url, base, headers, false)
            .map_err(|e| e.raise())?;
        p.borrow_mut().start().map_err(|e| e.raise())?;
        Ok(GetResponse {
            headers: BufReader::new(Headers(p.clone())),
            body: Response(p, vec![], 0),
        })
    }
    fn post(
        &mut self,
        url: &str,
        base: &str,
        headers: impl IntoIterator<Item = impl AsRef<str>>,
        _: PostBodyDataKind,
    ) -> gix_error::Result<PostResponse<Self::Headers, Response, PostBody>> {
        let p = self
            .pending(url, base, headers, true)
            .map_err(|e| e.raise())?;
        Ok(PostResponse {
            headers: BufReader::new(Headers(p.clone())),
            body: Response(p.clone(), vec![], 0),
            post_body: PostBody(p),
        })
    }
    fn configure(&mut self, _: &dyn std::any::Any) -> gix_error::Result<()> {
        Ok(())
    }
    fn redirected_base_url(&self) -> Option<String> {
        self.redirect.borrow().clone()
    }
}