gunnar-server 2.0.0

//! **Smart HTTP for the FETCH side, as a `BufRead`/`Write` pair.**
//!
//! [`gunnar_wire::fetch_pack`] talks to a remote through two objects — a
//! `&mut dyn BufRead` and a `&mut dyn Write` — and it writes, then reads, then
//! writes, then reads. Over SSH that is literally what the channel is: one
//! duplex stream, and [`crate::transport`]'s bridge hands the two halves over
//! untouched.
//!
//! Smart HTTP is **not a stream**. It is `GET /info/refs` once and then one
//! POST per round, request body in, response body out, and the connection is
//! nobody's to hold open between them. So something has to sit between the
//! state machine and the transport and turn one shape into the other, and this
//! is it:
//!
//! ```text
//!   fetch_pack writes  ──▶ [pending]                       (nothing on the wire)
//!   fetch_pack reads   ──▶ pending is non-empty and the
//!                          current body is exhausted
//!                            ──▶ POST /git-upload-pack, body = pending
//!                            ──▶ the response body becomes what is read
//! ```
//!
//! **The trigger is the first read after a write, not the flush.** A `flush()`
//! is advisory — `fetch_pack` happens to call one after each request, but a
//! writer is entitled not to, and an adapter that fired on `flush` would
//! deadlock against any client that did not. "Somebody wants bytes and I have
//! an unsent request" is the condition that is true exactly when the POST is
//! due, and it is true whether or not anybody flushed.
//!
//! # What this copies, named
//!
//! The trick is `gix-transport`'s (`client::http`): its `RequestWriter` buffers
//! what the protocol writes and the POST is made when the response is first
//! asked for, so a stateless-rpc round trip looks like a stream to the code
//! above it. What is **not** copied is everything around it — gix's HTTP
//! transport carries its own `Http` trait with curl and reqwest backends, a
//! redirect follower, its own auth types and its own error tree. That is 128
//! packages this tree measured and declined (see `fetch_pack`'s module doc);
//! what is reused here is a five-line idea, written out in this file.
//!
//! # Why `ureq` and not the client that is already here
//!
//! `hyper` IS already in this crate's graph — for the gRPC server, the metrics
//! server and (under `webhooks`) an outbound client. It is an **async** client,
//! and this adapter is called from the job queue's blocking worker in the
//! middle of a `BufRead::fill_buf`. Reaching hyper from there means a
//! `block_on` inside a blocking read, which is the deadlock
//! [`crate::transport`]'s module doc already documents from the other
//! direction. `ureq` is blocking by construction, is rustls-only (no curl
//! anywhere in this constellation, and this keeps it so), and is the client
//! `gunnar-forge` already resolved — so the packages are shared rather than
//! added twice.
//!
//! # The packfile is never held
//!
//! [`ureq::Body::into_reader`] is a streaming reader over the response
//! body, and it is what [`HttpRead`] reads from. The pack goes from the socket
//! into the ingest one buffer at a time, exactly as it does over SSH; the most
//! this module ever holds of a repository is one request body, which is a list
//! of wants and haves.

use std::cell::RefCell;
use std::io::{self, BufRead, BufReader, Read, Write};
use std::rc::Rc;
use std::time::Duration;

/// How long one request may take to *connect*.
///
/// Short: a mirror upstream that is not answering should fail the job with a
/// name, not hold a queue worker for the kernel's connect timeout while every
/// other repository's sync waits behind it.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(20);

/// How long one request may take in total.
///
/// **Generous, because one of these requests is a whole clone.** The POST that
/// answers `command=fetch` streams the packfile, and a large mirror is minutes.
/// It is bounded rather than infinite so a stalled TLS session cannot park a
/// worker for ever.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(60 * 60);

/// What the two halves share: the request body that has been written and not
/// yet sent.
///
/// `Rc<RefCell<…>>` and not a lock, because both halves are used by one
/// `FetchPack::fetch` call on one thread — the job queue's worker — and a lock
/// here would be a claim about concurrency that is not true. It is not `Send`,
/// and that is the type saying so.
type Pending = Rc<RefCell<Vec<u8>>>;

/// **The write half**: everything `fetch_pack` writes, kept until somebody
/// reads.
pub(crate) struct HttpWrite {
    pending: Pending,
}

impl Write for HttpWrite {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.pending.borrow_mut().extend_from_slice(buf);
        Ok(buf.len())
    }

    /// **Deliberately not the trigger.** See the module doc: the POST is made
    /// on the first read that finds an unsent request, so a client that never
    /// flushes still works.
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

/// **The read half**: the advertisement, then one POST response per round.
pub(crate) struct HttpRead {
    agent: ureq::Agent,
    /// `{base}/git-upload-pack` — computed once so the two halves cannot
    /// disagree about which repository they are talking to.
    post_url: String,
    /// The `Authorization` header value, or `None` for a public remote.
    ///
    /// Held as the finished header value rather than as the token, so the
    /// secret exists in exactly one shape in this process and nothing has to
    /// re-derive it. It is never formatted into a log line or an error; see
    /// [`refuse`].
    auth: Option<String>,
    pending: Pending,
    /// The body currently being read: the advertisement at first, then each
    /// POST response in turn.
    reading: Box<dyn BufRead>,
}

/// **Open a smart-HTTP fetch against `base`.**
///
/// Performs the `GET {base}/info/refs?service=git-upload-pack` advertisement
/// immediately, because that is what the first read of a fetch consumes and
/// doing it lazily would mean a constructor that cannot report an unreachable
/// remote, a 401, or a remote that is not a git server at all.
///
/// `base` is the remote URL with any trailing `/` removed; a `.git` suffix is
/// the remote's business and is passed through untouched.
pub(crate) fn open(
    base: &str,
    auth: Option<String>,
) -> Result<(HttpRead, HttpWrite), String> {
    open_through(base, auth, nordisk_pairing::egress::proxy())
}

/// [`open`], leaving through `door` when there is one (the front's egress door a box behind
/// its front was fed, the lean estate): the remote is reached as `CONNECT host:port` through
/// it, end to end, the front naming the host.
pub(crate) fn open_through(
    base: &str,
    auth: Option<String>,
    door: Option<&str>,
) -> Result<(HttpRead, HttpWrite), String> {
    let base = base.trim_end_matches('/');
    let agent = ureq::Agent::config_builder()
        // ureq's own default, restated so that a future default cannot quietly
        // start forwarding a mirror's token to whatever a redirect names.
        .redirect_auth_headers(ureq::config::RedirectAuthHeaders::Never)
        .timeout_connect(Some(CONNECT_TIMEOUT))
        .timeout_global(Some(REQUEST_TIMEOUT))
        // ureq 3 (2026-10-02), restated to ureq 2's behaviour: no proxy taken from
        // the environment and at most 5 redirects. The one proxy is the fed door.
        .proxy(door.and_then(|d| ureq::Proxy::new(d).ok()))
        .max_redirects(5)
        .build()
        .new_agent();

    let advertisement = format!("{base}/info/refs?service=git-upload-pack");
    let mut request = agent
        .get(&advertisement)
        // Without this header the remote answers **v0**, and `fetch_pack` is a
        // v2-only client: the failure would be "the remote opened with
        // `<sha> HEAD…` and not `version 2`", which reads as a broken server
        // rather than as a header this side forgot to send.
        .header("Git-Protocol", "version=2")
        .header("Accept", "*/*");
    if let Some(value) = &auth {
        request = request.header("Authorization", value);
    }
    let response = request.call().map_err(|err| refuse("the advertisement", &advertisement, err))?;
    let reading = skip_service_announcement(Box::new(BufReader::new(
        response.into_body().into_reader(),
    )))
    .map_err(|err| format!("reading the advertisement from {}: {err}", scrub(&advertisement)))?;

    let pending: Pending = Rc::new(RefCell::new(Vec::new()));
    Ok((
        HttpRead {
            agent,
            post_url: format!("{base}/git-upload-pack"),
            auth,
            pending: Rc::clone(&pending),
            reading,
        },
        HttpWrite { pending },
    ))
}

/// **Step over the smart-HTTP service announcement, if the server sent one.**
///
/// The smart-HTTP protocol lets a server open the `info/refs` answer with
/// `# service=git-upload-pack` and a flush-pkt before anything else, in v2 as in
/// v0, and GitHub, Forgejo (Codeberg) and stock `git http-backend` all do.
/// gunnar's own server does not, which is why every test passed while every
/// import from GitHub and Codeberg failed with "the remote opened with
/// `# service=git-upload-pack` and not `version 2`" (audit 2026-10-18).
///
/// The first pkt-line is read; when it is the announcement it is dropped with
/// the flush after it, and otherwise it is put back in front of the rest so
/// the fetch reads exactly what the server sent.
fn skip_service_announcement(mut body: Box<dyn BufRead>) -> io::Result<Box<dyn BufRead>> {
    let mut length = [0u8; 4];
    if let Err(err) = body.read_exact(&mut length) {
        // Too short to hold a pkt-line: hand over what there is and let the
        // protocol layer name it.
        if err.kind() == io::ErrorKind::UnexpectedEof {
            return Ok(Box::new(io::Cursor::new(Vec::new())));
        }
        return Err(err);
    }
    let size = std::str::from_utf8(&length)
        .ok()
        .and_then(|hex| usize::from_str_radix(hex, 16).ok());
    let line = match size {
        Some(size) if size > 4 => {
            let mut line = vec![0u8; size - 4];
            body.read_exact(&mut line)?;
            line
        }
        // A flush, a delimiter or not a pkt-line at all: not the announcement.
        _ => Vec::new(),
    };
    if line.starts_with(b"# service=") {
        let mut flush = [0u8; 4];
        body.read_exact(&mut flush)?;
        if &flush != b"0000" {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "the service announcement was not followed by a flush-pkt",
            ));
        }
        return Ok(body);
    }
    let mut seen = length.to_vec();
    seen.extend_from_slice(&line);
    Ok(Box::new(BufReader::new(io::Cursor::new(seen).chain(body))))
}

impl HttpRead {
    /// Send what has been written and make the response the thing being read.
    fn post(&mut self) -> io::Result<()> {
        let body = std::mem::take(&mut *self.pending.borrow_mut());
        let mut request = self
            .agent
            .post(&self.post_url)
            .header("Content-Type", "application/x-git-upload-pack-request")
            .header("Accept", "application/x-git-upload-pack-result")
            .header("Git-Protocol", "version=2");
        if let Some(value) = &self.auth {
            request = request.header("Authorization", value);
        }
        let response = request
            .send(&body[..])
            .map_err(|err| io::Error::other(refuse("a fetch round", &self.post_url, err)))?;
        self.reading = Box::new(BufReader::new(response.into_body().into_reader()));
        Ok(())
    }
}

impl BufRead for HttpRead {
    fn fill_buf(&mut self) -> io::Result<&[u8]> {
        // The one condition in this file: somebody wants bytes, the body in
        // hand is spent, and there is a request that has not been sent. The
        // borrow from the first `fill_buf` ends inside the `if`, which is why
        // this can be written without a loop and without cloning a buffer.
        let exhausted = self.reading.fill_buf()?.is_empty();
        if exhausted && !self.pending.borrow().is_empty() {
            self.post()?;
        }
        self.reading.fill_buf()
    }

    fn consume(&mut self, amount: usize) {
        self.reading.consume(amount);
    }
}

impl Read for HttpRead {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let available = self.fill_buf()?;
        let taken = available.len().min(buf.len());
        buf[..taken].copy_from_slice(&available[..taken]);
        self.consume(taken);
        Ok(taken)
    }
}

/// **One `ureq::Error`, rendered without the URL's userinfo and without the
/// credential.**
///
/// `ureq`'s own `Display` prints the request URL. That is normally the most
/// useful half of an HTTP error and here it is the dangerous half: a remote
/// written as `https://user:token@host/repo.git` would put the token into every
/// log line, every FAILED job and every `Mirror.Status` response. So the URL is
/// **not** taken from the error — it is the caller's own `url` argument, which
/// this function scrubs — and the error contributes only its status or its
/// kind.
///
/// A 401 and a 403 are named as what they are, because "authentication failed"
/// and "authenticated, and not allowed" send an operator to two different
/// places and only one of them is the credential.
fn refuse(what: &str, url: &str, err: ureq::Error) -> String {
    let url = scrub(url);
    match err {
        // **Plain words, for the person who pressed Import** (owner order
        // 2026-10-18): what happened and what to do, with the status code at
        // the end for whoever reads it next. No verbs, no flags.
        ureq::Error::StatusCode(401) => format!(
            "This repository is private. Connect your GitHub or Codeberg account or add an \
             access token, then try again. ({url} answered {what} with 401.)"
        ),
        ureq::Error::StatusCode(403) => format!(
            "The access token was accepted but may not read this repository. Use a token \
             that can read it, then try again. ({url} answered {what} with 403.)"
        ),
        ureq::Error::StatusCode(404) => format!(
            "No repository was found at this address. Check the address; a private \
             repository can also look like this until an access token is added. ({url} \
             answered {what} with 404.)"
        ),
        ureq::Error::StatusCode(code) => format!("{url} answered {what} with HTTP {code}"),
        transport => {
            let (kind, detail) = transport_detail(&transport);
            format!("{url} could not be reached for {what}: {kind}{detail}")
        }
    }
}

/// **A ureq transport error as a kind word plus the one detail that is safe.**
///
/// ureq 3 (2026-10-02) has no `ErrorKind`/`message()`, and its `Display` is not
/// safe as a whole: `BadUri` and `RequireHttpsOnly` print the request URL, which
/// can carry `user:token@`. So each variant is rendered on purpose, and only an
/// I/O error's own sentence, a timeout's phase and a TLS reason are printed —
/// none of which carries the URL or the header.
///
/// A name that does not resolve arrives as an I/O error whose sentence is libc's
/// ("failed to lookup address information: Name or service not known" /
/// "Temporary failure in name resolution"). It is labelled `Dns`, as ureq 2's
/// kind was, because that sentence is the one fact that tells a missing resolver
/// file from a resolver nobody answers — measured 2026-09-08 on gunnar.rs, where
/// a whole re-image was spent finding out which.
fn transport_detail(err: &ureq::Error) -> (&'static str, String) {
    use ureq::Error::*;
    match err {
        Io(e) if e.to_string().starts_with("failed to lookup address information") => ("Dns", format!(" ({e})")),
        Io(e) => ("Io", format!(" ({e})")),
        HostNotFound => ("Dns", " (the name resolved to no address)".to_string()),
        Timeout(t) => ("Timeout", format!(" ({t})")),
        ConnectionFailed => ("ConnectionFailed", String::new()),
        TooManyRedirects => ("TooManyRedirects", String::new()),
        RedirectFailed => ("RedirectFailed", String::new()),
        BadUri(_) | Http(_) => ("InvalidUrl", String::new()),
        RequireHttpsOnly(_) | TlsRequired => ("InsecureRequestHttpsOnly", String::new()),
        Tls(reason) => ("Tls", format!(" ({reason})")),
        Rustls(e) => ("Tls", format!(" ({e})")),
        Pem(_) => ("Tls", String::new()),
        Protocol(_) | LargeResponseHeader(..) => ("BadResponse", String::new()),
        BodyExceedsLimit(_) | BodyStalled => ("Io", String::new()),
        InvalidProxyUrl | ConnectProxyFailed(_) => ("ProxyConnect", String::new()),
        _ => ("Other", String::new()),
    }
}

pub(crate) use crate::mirror_fetch::scrub;

#[cfg(test)]
mod tests {
    use super::*;

    /// **The smart-HTTP service announcement is stepped over** — what GitHub,
    /// Forgejo and `git http-backend` open with — and a body without one is
    /// handed on byte for byte.
    #[test]
    fn the_service_announcement_is_skipped_and_nothing_else_is() {
        let v2 = b"000eversion 2\n0013object-format=sha1\n0000".to_vec();
        let mut announced = b"001e# service=git-upload-pack\n0000".to_vec();
        announced.extend_from_slice(&v2);
        for body in [announced, v2.clone()] {
            let mut read = Vec::new();
            skip_service_announcement(Box::new(io::Cursor::new(body)))
                .expect("a well-formed advertisement")
                .read_to_end(&mut read)
                .expect("read the rest");
            assert_eq!(read, v2);
        }
        let mut short = Vec::new();
        skip_service_announcement(Box::new(io::Cursor::new(b"00".to_vec())))
            .expect("too short is left to the protocol layer")
            .read_to_end(&mut short)
            .expect("read");
        assert!(short.is_empty());
        assert!(
            skip_service_announcement(Box::new(io::Cursor::new(
                b"001e# service=git-upload-pack\n000eversion 2\n".to_vec()
            )))
            .is_err(),
            "an announcement must be followed by a flush"
        );
    }

    /// ★ **A mirror fetch of a box behind its front leaves through the door it was fed** (the
    /// lean estate, A19 row 14): the advertisement's GET goes to the front's egress door as
    /// `CONNECT remote:port`, never to the remote directly. Over loopback, the door recording
    /// what it was asked for.
    #[test]
    fn a_mirror_fetch_leaves_through_the_fed_door() {
        use std::net::{SocketAddr, TcpListener};
        static SEEN: std::sync::Mutex<Vec<(String, u16)>> = std::sync::Mutex::new(Vec::new());
        fn record(host: &str, port: u16, resolved: &[SocketAddr]) -> Result<(), String> {
            SEEN.lock().unwrap().push((host.to_string(), port));
            if resolved.is_empty() { Err("unresolved".into()) } else { Ok(()) }
        }
        let remote = TcpListener::bind("127.0.0.1:0").unwrap();
        let at = remote.local_addr().unwrap();
        std::thread::spawn(move || {
            for mut c in remote.incoming().flatten() {
                let mut head = Vec::new();
                let mut b = [0u8; 1];
                while !head.ends_with(b"\r\n\r\n") && c.read(&mut b).map(|n| n == 1).unwrap_or(false) {
                    head.push(b[0]);
                }
                let body = b"001e# service=git-upload-pack\n0000000eversion 2\n0000";
                let _ = write!(c, "HTTP/1.1 200 OK\r\nContent-Type: application/x-git-upload-pack-advertisement\r\nContent-Length: {}\r\n\r\n", body.len());
                let _ = c.write_all(body);
            }
        });
        let door = nordisk_pairing::egress::serve("127.0.0.1:0".parse().unwrap(), nordisk_pairing::egress::Policy { peer: |_| true, admits: record }).unwrap();
        let door = format!("http://{door}");
        let opened = open_through(&format!("http://127.0.0.1:{}/ns/repo.git", at.port()), None, Some(&door));
        assert!(opened.is_ok(), "the advertisement through the door: {:?}", opened.err());
        assert!(SEEN.lock().unwrap().iter().any(|(h, p)| h == "127.0.0.1" && *p == at.port()), "the door carried the fetch: {:?}", SEEN.lock().unwrap());
    }

    /// **The write half puts nothing on the wire.** A `flush` that sent would
    /// make the adapter POST an incomplete request the moment `fetch_pack`
    /// flushed mid-round; this asserts that a flush is inert and the bytes are
    /// still there for the read half to send.
    #[test]
    fn a_flush_sends_nothing_and_keeps_the_request() {
        let pending: Pending = Rc::new(RefCell::new(Vec::new()));
        let mut writer = HttpWrite {
            pending: Rc::clone(&pending),
        };
        writer.write_all(b"0014command=ls-refs\n").expect("buffered");
        writer.flush().expect("a flush is inert");
        assert_eq!(
            &*pending.borrow(),
            b"0014command=ls-refs\n",
            "a flush must not consume the request; the read half sends it"
        );
    }
}