git_cache_proxy/

server.rs

// SPDX-License-Identifier: Apache-2.0
//! HTTP surface: the git smart-HTTP endpoints plus health/metrics.
//!
//! Routing is method + path suffix based (git paths have arbitrary depth), so
//! the git handler is registered as the router fallback and dispatches:
//!   GET  <repo>/info/refs?service=git-upload-pack  -> ref advertisement
//!   POST <repo>/git-upload-pack                    -> packfile (streamed)
//!   anything git-receive-pack                      -> 403 (read-only)

use std::io::Read;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use axum::Router;
use axum::body::{Body, Bytes};
use axum::extract::State;
use axum::http::{HeaderMap, Method, Request, StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use subtle::ConstantTimeEq;
use tower::limit::GlobalConcurrencyLimitLayer;

use crate::git::GitCache;
use crate::lfs::{Lfs, Outcome};
use crate::metrics::{LfsResult, Metrics, RequestKind, Status};
use crate::repo;

const MAX_BODY: usize = 64 * 1024 * 1024;
const UPLOAD_PACK: &str = "git-upload-pack";
const RECEIVE_PACK: &str = "git-receive-pack";
const LFS_CONTENT_TYPE: &str = "application/vnd.git-lfs+json";
const LFS_BATCH_SUFFIX: &str = "/info/lfs/objects/batch";

#[derive(Clone)]
pub struct AppState {
    pub cache: Arc<GitCache>,
    pub lfs: Arc<Lfs>,
    pub upstream_base: String,
    pub cache_root: PathBuf,
    pub serve_token: Option<String>,
    /// Upper bound (bytes) on a decoded upload-pack request body. See
    /// `Config::max_decoded_body_mb`.
    pub max_decoded_body: usize,
    /// Max concurrent in-flight requests (`0` = unlimited). See
    /// `Config::max_concurrent_requests`.
    pub max_concurrent: usize,
    pub metrics: Arc<Metrics>,
}

pub fn router(state: AppState) -> Router {
    let max_concurrent = state.max_concurrent;

    // Observability endpoints are deliberately kept *out* of the concurrency
    // limit below: a liveness/readiness probe or a metrics scrape must stay
    // responsive even when a burst of clones has saturated the semaphore -
    // otherwise a healthy-but-busy pod fails its probes and gets restarted.
    let observability = Router::new()
        .route("/healthz", get(|| async { "ok" }))
        .route("/readyz", get(readyz))
        .route("/metrics", get(metrics_handler))
        .with_state(state.clone());

    // The git smart-HTTP surface has arbitrary-depth paths, so it is the router
    // fallback, and it is the only thing the concurrency limit wraps (`0`
    // disables it). One global semaphore shared across every per-connection
    // clone of the service (axum clones it per connection) makes the cap
    // process-wide rather than per-connection. `merge` keeps the (limited)
    // fallback as the merged fallback, while the observability routes above are
    // matched first and bypass it.
    let mut git = Router::new().fallback(handle_git).with_state(state);
    if max_concurrent != 0 {
        git = git.layer(GlobalConcurrencyLimitLayer::new(max_concurrent));
    }

    observability.merge(git)
}

async fn metrics_handler(State(st): State<AppState>) -> Response {
    Response::builder()
        .header(header::CONTENT_TYPE, "text/plain; version=0.0.4")
        .body(Body::from(st.metrics.gather()))
        .expect("valid response")
}

/// Readiness probe. Liveness (`/healthz`) only says the process is up; readiness
/// additionally verifies the proxy can do its one job - write bare mirrors into
/// the cache root - so a detached, unmounted, read-only, or unwritable cache
/// volume surfaces as `503 Service Unavailable` here instead of a later flood of
/// upstream `502`s. Kept out of the concurrency limit (see `router`).
async fn readyz(State(st): State<AppState>) -> Response {
    match cache_writable(&st.cache_root).await {
        Ok(()) => (StatusCode::OK, "ok").into_response(),
        Err(e) => {
            tracing::warn!(
                cache_root = %st.cache_root.display(),
                error = %e,
                "readiness check failed"
            );
            err(StatusCode::SERVICE_UNAVAILABLE, "cache root not writable")
        }
    }
}

/// Confirm the cache root exists and is writable by creating and removing a probe
/// file - the same directory the mirrors live in, so it tests the real target
/// rather than a proxy for it. A single create + unlink is cheap enough to run
/// per probe and catches the failure modes that make "ready" a lie: a missing
/// directory, a read-only remount, or a permissions problem.
async fn cache_writable(cache_root: &Path) -> std::io::Result<()> {
    tokio::fs::create_dir_all(cache_root).await?;
    let probe = cache_root.join(".readyz-probe");
    tokio::fs::write(&probe, b"").await?;
    let _ = tokio::fs::remove_file(&probe).await;
    Ok(())
}

async fn handle_git(State(st): State<AppState>, req: Request<Body>) -> Response {
    let (parts, body) = req.into_parts();
    let path = parts.uri.path().to_string();
    let query = parts.uri.query().unwrap_or("").to_string();
    let git_protocol = parts
        .headers
        .get("git-protocol")
        .and_then(|v| v.to_str().ok())
        .map(str::to_string);

    if let Some(resp) = check_auth(&st, &parts.headers) {
        st.metrics
            .record_request(RequestKind::Auth, Status::Unauthorized, "-");
        return resp;
    }

    // Read-only: refuse anything that would write upstream.
    if path.ends_with(&format!("/{RECEIVE_PACK}"))
        || query.contains(&format!("service={RECEIVE_PACK}"))
    {
        st.metrics
            .record_request(RequestKind::ReceivePack, Status::Rejected, "-");
        return err(
            StatusCode::FORBIDDEN,
            "read-only proxy: pushes are not allowed",
        );
    }

    if parts.method == Method::GET && path.ends_with("/info/refs") {
        if !query.contains(&format!("service={UPLOAD_PACK}")) {
            st.metrics
                .record_request(RequestKind::InfoRefs, Status::Error, "-");
            return err(
                StatusCode::BAD_REQUEST,
                "only smart-http git-upload-pack is supported",
            );
        }
        return info_refs(st, &path, git_protocol.as_deref()).await;
    }

    if parts.method == Method::POST && path.ends_with(&format!("/{UPLOAD_PACK}")) {
        let body = match axum::body::to_bytes(body, MAX_BODY).await {
            Ok(b) => b,
            Err(_) => return err(StatusCode::BAD_REQUEST, "failed to read request body"),
        };
        // Git's smart-HTTP client gzips the upload-pack request body (any
        // normally-sized want/have negotiation) and sends `Content-Encoding:
        // gzip`. `git upload-pack` reads its stdin as raw pkt-lines, so we must
        // undo the transport encoding before handing the body over - otherwise
        // it chokes on the gzip magic with "bad line length character".
        let content_encoding = parts
            .headers
            .get(header::CONTENT_ENCODING)
            .and_then(|v| v.to_str().ok());
        let body = match decode_body(content_encoding, body, st.max_decoded_body) {
            Ok(b) => b,
            Err(_) => return err(StatusCode::BAD_REQUEST, "failed to decode request body"),
        };
        return upload_pack(st, &path, git_protocol.as_deref(), body).await;
    }

    // git-LFS uses a different HTTP API alongside the git endpoints.
    if parts.method == Method::POST && path.ends_with(LFS_BATCH_SUFFIX) {
        let body = match axum::body::to_bytes(body, MAX_BODY).await {
            Ok(b) => b,
            Err(_) => return err(StatusCode::BAD_REQUEST, "failed to read request body"),
        };
        return lfs_batch(st, &path, &parts.headers, body).await;
    }
    if parts.method == Method::GET
        && let Some((repo_name, oid)) = repo::lfs_object_from_path(&path)
    {
        return lfs_object(st, repo_name, oid, &query).await;
    }

    err(StatusCode::NOT_FOUND, "not a git smart-http endpoint")
}

/// Proxy an LFS batch request to upstream, rewriting each object's download URL back
/// to this proxy so the object fetch is cached here. An `upload` batch is refused:
/// the proxy is read-only, like `git-receive-pack`.
async fn lfs_batch(st: AppState, path: &str, headers: &HeaderMap, body: Bytes) -> Response {
    let Some(name) = repo::lfs_batch_repo(path) else {
        st.metrics
            .record_request(RequestKind::LfsBatch, Status::Error, "-");
        return err(StatusCode::NOT_FOUND, "bad lfs batch path");
    };
    if is_lfs_upload(&body) {
        st.metrics
            .record_request(RequestKind::LfsBatch, Status::Rejected, "-");
        return err(
            StatusCode::FORBIDDEN,
            "read-only proxy: lfs upload is not allowed",
        );
    }
    let advertise = advertise_base(headers);
    match st.lfs.batch(&name, &body, &advertise).await {
        Ok(json) => {
            st.metrics
                .record_request(RequestKind::LfsBatch, Status::Ok, &name);
            Response::builder()
                .header(header::CONTENT_TYPE, LFS_CONTENT_TYPE)
                .header(header::CACHE_CONTROL, "no-cache")
                .body(Body::from(json))
                .expect("valid response")
        }
        Err(e) => {
            st.metrics
                .record_request(RequestKind::LfsBatch, Status::UpstreamError, "-");
            tracing::warn!(repo = %name, error = %e, "lfs batch failed");
            err(StatusCode::BAD_GATEWAY, "upstream lfs batch failed")
        }
    }
}

/// Serve a cached LFS object, fetching and caching it from upstream on a miss. The
/// `repo` label is `-` because objects are content-addressed and shared across repos
/// (the cache hit/miss is tracked separately in `lfs_objects_total`).
async fn lfs_object(st: AppState, repo_name: String, oid: String, query: &str) -> Response {
    let size = size_from_query(query);
    match st.lfs.ensure_object(&repo_name, &oid, size).await {
        Ok((path, outcome)) => {
            st.metrics.record_lfs(match outcome {
                Outcome::Hit => LfsResult::Hit,
                Outcome::Miss => LfsResult::Miss,
            });
            match lfs_file_response(&path).await {
                Ok(resp) => {
                    st.metrics
                        .record_request(RequestKind::LfsObject, Status::Ok, "-");
                    resp
                }
                Err(e) => {
                    st.metrics
                        .record_request(RequestKind::LfsObject, Status::Error, "-");
                    tracing::warn!(oid = %oid, error = %e, "serve cached lfs object failed");
                    err(StatusCode::INTERNAL_SERVER_ERROR, "serve lfs object failed")
                }
            }
        }
        Err(e) => {
            st.metrics.record_lfs(LfsResult::Error);
            st.metrics
                .record_request(RequestKind::LfsObject, Status::UpstreamError, "-");
            tracing::warn!(oid = %oid, error = %e, "lfs object fetch failed");
            err(StatusCode::BAD_GATEWAY, "upstream lfs object fetch failed")
        }
    }
}

/// Stream a cached LFS object file to the client with its length.
async fn lfs_file_response(path: &Path) -> std::io::Result<Response> {
    let file = tokio::fs::File::open(path).await?;
    let len = file.metadata().await?.len();
    let stream = tokio_util::io::ReaderStream::new(file);
    Ok(Response::builder()
        .header(header::CONTENT_TYPE, "application/octet-stream")
        .header(header::CONTENT_LENGTH, len)
        .body(Body::from_stream(stream))
        .expect("valid response"))
}

/// Whether an LFS batch request asks to upload (write). Best-effort: an unparsable
/// body is treated as not-upload and forwarded, letting upstream decide.
fn is_lfs_upload(body: &[u8]) -> bool {
    let Ok(v) = serde_json::from_slice::<serde_json::Value>(body) else {
        return false;
    };
    v.get("operation").and_then(serde_json::Value::as_str) == Some("upload")
}

/// The `size` query parameter of an object request, which the proxy embeds in the
/// download URLs it advertises (the upstream batch API needs it on a cache miss).
fn size_from_query(query: &str) -> Option<u64> {
    query
        .split('&')
        .find_map(|kv| kv.strip_prefix("size="))
        .and_then(|v| v.parse().ok())
}

/// The proxy's own base URL (`scheme://host`) as the client reached it, used to
/// rewrite LFS object download URLs back to this proxy. Honors `X-Forwarded-Proto` /
/// `X-Forwarded-Host` from a TLS-terminating ingress and otherwise falls back to the
/// request `Host` over plain http, which is what the proxy itself speaks.
fn advertise_base(headers: &HeaderMap) -> String {
    let first = |v: &axum::http::HeaderValue| {
        v.to_str()
            .ok()
            .map(|s| s.split(',').next().unwrap_or(s).trim().to_string())
    };
    let scheme = headers
        .get("x-forwarded-proto")
        .and_then(first)
        .filter(|s| !s.is_empty())
        .unwrap_or_else(|| "http".to_string());
    let host = headers
        .get("x-forwarded-host")
        .or_else(|| headers.get(header::HOST))
        .and_then(first)
        .unwrap_or_default();
    format!("{scheme}://{host}")
}

async fn info_refs(st: AppState, path: &str, git_protocol: Option<&str>) -> Response {
    let Some(name) = repo::repo_name_from_path(path, "/info/refs") else {
        st.metrics
            .record_request(RequestKind::InfoRefs, Status::Error, "-");
        return err(StatusCode::NOT_FOUND, "bad path");
    };
    let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) {
        Ok(r) => r,
        Err(e) => {
            st.metrics
                .record_request(RequestKind::InfoRefs, Status::Error, "-");
            return err(StatusCode::BAD_REQUEST, &e.to_string());
        }
    };

    // The upstream clone/fetch counters (per repo) are recorded inside `GitCache`;
    // here we only account for the client request. The `repo` label is emitted only
    // once a request is served; failures use `-` so a flood of distinct but doomed
    // repo paths cannot inflate label cardinality (see `metrics`).
    if let Err(e) = st.cache.ensure_fresh(&repo, true).await {
        st.metrics
            .record_request(RequestKind::InfoRefs, Status::UpstreamError, "-");
        tracing::warn!(repo = %name, error = %e, "ensure_fresh failed");
        return err(StatusCode::BAD_GATEWAY, "upstream fetch failed");
    }

    match st.cache.advertise_refs(&repo, git_protocol).await {
        Ok(body) => {
            st.metrics
                .record_request(RequestKind::InfoRefs, Status::Ok, &name);
            Response::builder()
                .header(
                    header::CONTENT_TYPE,
                    "application/x-git-upload-pack-advertisement",
                )
                .header(header::CACHE_CONTROL, "no-cache")
                .body(Body::from(body))
                .expect("valid response")
        }
        Err(e) => {
            st.metrics
                .record_request(RequestKind::InfoRefs, Status::Error, "-");
            tracing::warn!(repo = %name, error = %e, "advertise_refs failed");
            err(StatusCode::INTERNAL_SERVER_ERROR, "advertise-refs failed")
        }
    }
}

async fn upload_pack(
    st: AppState,
    path: &str,
    git_protocol: Option<&str>,
    body: Bytes,
) -> Response {
    let Some(name) = repo::repo_name_from_path(path, &format!("/{UPLOAD_PACK}")) else {
        st.metrics
            .record_request(RequestKind::UploadPack, Status::Error, "-");
        return err(StatusCode::NOT_FOUND, "bad path");
    };
    let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) {
        Ok(r) => r,
        Err(e) => {
            st.metrics
                .record_request(RequestKind::UploadPack, Status::Error, "-");
            return err(StatusCode::BAD_REQUEST, &e.to_string());
        }
    };

    // The preceding info/refs already refreshed; here just ensure the mirror is
    // present (a client could POST against a not-yet-cloned repo).
    if let Err(e) = st.cache.ensure_fresh(&repo, false).await {
        st.metrics
            .record_request(RequestKind::UploadPack, Status::UpstreamError, "-");
        tracing::warn!(repo = %name, error = %e, "ensure mirror exists failed");
        return err(StatusCode::BAD_GATEWAY, "upstream unavailable");
    }

    match st.cache.upload_pack_rpc(&repo, git_protocol, body).await {
        Ok(stream) => {
            st.metrics
                .record_request(RequestKind::UploadPack, Status::Ok, &name);
            Response::builder()
                .header(header::CONTENT_TYPE, "application/x-git-upload-pack-result")
                .header(header::CACHE_CONTROL, "no-cache")
                .body(Body::from_stream(stream))
                .expect("valid response")
        }
        Err(e) => {
            st.metrics
                .record_request(RequestKind::UploadPack, Status::Error, "-");
            tracing::warn!(repo = %name, error = %e, "upload_pack_rpc failed");
            err(StatusCode::INTERNAL_SERVER_ERROR, "upload-pack failed")
        }
    }
}

/// When a serve token is configured, require `Authorization: Bearer <token>`.
/// Returns `Some(401)` to short-circuit, `None` to allow.
fn check_auth(st: &AppState, headers: &HeaderMap) -> Option<Response> {
    let expected = st.serve_token.as_ref()?;
    let provided = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "));
    if provided.is_some_and(|t| token_matches(t, expected)) {
        None
    } else {
        Some(err(
            StatusCode::UNAUTHORIZED,
            "missing or invalid bearer token",
        ))
    }
}

/// Compare a client-supplied bearer token against the expected one in constant
/// time, so response latency does not leak how many leading bytes matched. Only
/// the length can differ observably, which is not sensitive for a shared secret.
fn token_matches(provided: &str, expected: &str) -> bool {
    provided.as_bytes().ct_eq(expected.as_bytes()).into()
}

/// Undo the request's `Content-Encoding`. Git only ever gzips, so that is the
/// single encoding we decode; an absent/`identity` header passes through
/// untouched, and any other encoding is rejected by the caller as a bad body.
///
/// The decoded size is capped at `max_decoded`: DEFLATE reaches ~1000:1, so an
/// unbounded read here would let a small compressed body expand into an
/// out-of-memory kill (a decompression bomb). We read at most `max_decoded + 1`
/// bytes so we can tell "exactly at the limit" from "over it" and reject the
/// latter.
fn decode_body(
    content_encoding: Option<&str>,
    body: Bytes,
    max_decoded: usize,
) -> std::io::Result<Bytes> {
    match content_encoding.map(str::trim) {
        Some(enc) if enc.eq_ignore_ascii_case("gzip") || enc.eq_ignore_ascii_case("x-gzip") => {
            let mut out = Vec::new();
            let limit = max_decoded as u64 + 1;
            flate2::read::GzDecoder::new(&body[..])
                .take(limit)
                .read_to_end(&mut out)?;
            within_limit(Bytes::from(out), max_decoded)
        }
        // Uncompressed: no expansion risk, but still enforce the cap so
        // `--max-decoded-body-mb` bounds an identity request the same as a gzipped
        // one - otherwise only the coarser transport cap (`MAX_BODY`) would apply.
        None => within_limit(body, max_decoded),
        Some(enc) if enc.is_empty() || enc.eq_ignore_ascii_case("identity") => {
            within_limit(body, max_decoded)
        }
        Some(other) => Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("unsupported content-encoding: {other}"),
        )),
    }
}

/// Reject a decoded body that exceeds the configured cap. Shared by every encoding
/// branch so the limit is enforced uniformly, not just when decoding gzip.
fn within_limit(body: Bytes, max_decoded: usize) -> std::io::Result<Bytes> {
    if body.len() > max_decoded {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "decoded request body exceeds limit",
        ));
    }
    Ok(body)
}

fn err(status: StatusCode, msg: &str) -> Response {
    (status, format!("{msg}\n")).into_response()
}

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

    fn gzip(bytes: &[u8]) -> Bytes {
        let mut enc = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::best());
        enc.write_all(bytes).unwrap();
        Bytes::from(enc.finish().unwrap())
    }

    #[test]
    fn identity_and_absent_encoding_pass_through() {
        let raw = Bytes::from_static(b"want ...\n");
        assert_eq!(decode_body(None, raw.clone(), 1024).unwrap(), raw);
        assert_eq!(
            decode_body(Some("identity"), raw.clone(), 1024).unwrap(),
            raw
        );
        assert_eq!(decode_body(Some(""), raw.clone(), 1024).unwrap(), raw);
    }

    #[test]
    fn identity_body_over_limit_is_rejected() {
        // An uncompressed body must honor the decoded-body cap too, not just the
        // gzip path. Exactly at the limit is accepted; one byte over is rejected.
        let body = Bytes::from(vec![b'x'; 2048]);
        for enc in [None, Some("identity"), Some("")] {
            assert!(decode_body(enc, body.clone(), 2048).is_ok());
            assert!(decode_body(enc, body.clone(), 2047).is_err());
        }
    }

    #[test]
    fn gzip_within_limit_decodes() {
        let payload = b"command=ls-refs\n";
        let decoded = decode_body(Some("gzip"), gzip(payload), 1024).unwrap();
        assert_eq!(&decoded[..], payload);
        // Case-insensitive and the `x-gzip` alias both decode.
        assert_eq!(
            &decode_body(Some("GZIP"), gzip(payload), 1024).unwrap()[..],
            payload
        );
        assert_eq!(
            &decode_body(Some("x-gzip"), gzip(payload), 1024).unwrap()[..],
            payload
        );
    }

    #[test]
    fn gzip_decompression_bomb_is_rejected() {
        // 1 MiB of zeros compresses to ~1 KiB but must not be allowed to expand
        // past the cap. Exactly-at-limit is accepted; one byte over is rejected.
        let big = vec![0u8; 1024 * 1024];
        assert!(decode_body(Some("gzip"), gzip(&big), 1024).is_err());
        assert!(decode_body(Some("gzip"), gzip(&big), big.len()).is_ok());
        assert!(decode_body(Some("gzip"), gzip(&big), big.len() - 1).is_err());
    }

    #[test]
    fn unsupported_encoding_is_rejected() {
        assert!(decode_body(Some("br"), Bytes::from_static(b"x"), 1024).is_err());
        assert!(decode_body(Some("deflate"), Bytes::from_static(b"x"), 1024).is_err());
    }

    #[test]
    fn token_matches_only_the_exact_token() {
        assert!(token_matches("s3cret", "s3cret"));
        assert!(!token_matches("s3creT", "s3cret")); // last byte differs
        assert!(!token_matches("s3cre", "s3cret")); // prefix, shorter
        assert!(!token_matches("s3cret-extra", "s3cret")); // longer
        assert!(!token_matches("", "s3cret"));
        assert!(token_matches("", "")); // degenerate empty token
    }

    #[test]
    fn advertise_base_uses_forwarded_headers_then_host() {
        use axum::http::HeaderValue;

        let mut h = HeaderMap::new();
        h.insert(header::HOST, HeaderValue::from_static("svc.local:8080"));
        assert_eq!(advertise_base(&h), "http://svc.local:8080");
        // A TLS-terminating ingress advertises via X-Forwarded-*.
        h.insert("x-forwarded-proto", HeaderValue::from_static("https"));
        h.insert(
            "x-forwarded-host",
            HeaderValue::from_static("proxy.example"),
        );
        assert_eq!(advertise_base(&h), "https://proxy.example");
        // A comma-listed forwarded chain uses the first hop.
        h.insert("x-forwarded-proto", HeaderValue::from_static("https, http"));
        assert_eq!(advertise_base(&h), "https://proxy.example");
    }

    #[test]
    fn size_from_query_parses_only_a_valid_size() {
        assert_eq!(size_from_query("size=42"), Some(42));
        assert_eq!(size_from_query("a=1&size=7&b=2"), Some(7));
        assert_eq!(size_from_query(""), None);
        assert_eq!(size_from_query("size=notanumber"), None);
    }

    #[test]
    fn is_lfs_upload_detects_the_operation() {
        assert!(is_lfs_upload(br#"{"operation":"upload","objects":[]}"#));
        assert!(!is_lfs_upload(br#"{"operation":"download"}"#));
        assert!(!is_lfs_upload(b"not json")); // unparsable -> forwarded, not upload
        assert!(!is_lfs_upload(b"{}"));
    }
}