git_cache_proxy/
server.rs

1// SPDX-License-Identifier: Apache-2.0
2//! HTTP surface: the git smart-HTTP endpoints plus health/metrics.
3//!
4//! Routing is method + path suffix based (git paths have arbitrary depth), so
5//! the git handler is registered as the router fallback and dispatches:
6//!   GET  <repo>/info/refs?service=git-upload-pack  -> ref advertisement
7//!   POST <repo>/git-upload-pack                    -> packfile (streamed)
8//!   anything git-receive-pack                      -> 403 (read-only)
9
10use std::io::Read;
11use std::path::{Path, PathBuf};
12use std::sync::Arc;
13
14use axum::Router;
15use axum::body::{Body, Bytes};
16use axum::extract::State;
17use axum::http::{HeaderMap, Method, Request, StatusCode, header};
18use axum::response::{IntoResponse, Response};
19use axum::routing::get;
20use subtle::ConstantTimeEq;
21use tower::limit::GlobalConcurrencyLimitLayer;
22
23use crate::git::GitCache;
24use crate::metrics::Metrics;
25use crate::repo;
26
27const MAX_BODY: usize = 64 * 1024 * 1024;
28const UPLOAD_PACK: &str = "git-upload-pack";
29const RECEIVE_PACK: &str = "git-receive-pack";
30
31#[derive(Clone)]
32pub struct AppState {
33    pub cache: Arc<GitCache>,
34    pub upstream_base: String,
35    pub cache_root: PathBuf,
36    pub serve_token: Option<String>,
37    /// Upper bound (bytes) on a decoded upload-pack request body. See
38    /// `Config::max_decoded_body_mb`.
39    pub max_decoded_body: usize,
40    /// Max concurrent in-flight requests (`0` = unlimited). See
41    /// `Config::max_concurrent_requests`.
42    pub max_concurrent: usize,
43    pub metrics: Arc<Metrics>,
44}
45
46pub fn router(state: AppState) -> Router {
47    let max_concurrent = state.max_concurrent;
48
49    // Observability endpoints are deliberately kept *out* of the concurrency
50    // limit below: a liveness/readiness probe or a metrics scrape must stay
51    // responsive even when a burst of clones has saturated the semaphore -
52    // otherwise a healthy-but-busy pod fails its probes and gets restarted.
53    let observability = Router::new()
54        .route("/healthz", get(|| async { "ok" }))
55        .route("/readyz", get(readyz))
56        .route("/metrics", get(metrics_handler))
57        .with_state(state.clone());
58
59    // The git smart-HTTP surface has arbitrary-depth paths, so it is the router
60    // fallback, and it is the only thing the concurrency limit wraps (`0`
61    // disables it). One global semaphore shared across every per-connection
62    // clone of the service (axum clones it per connection) makes the cap
63    // process-wide rather than per-connection. `merge` keeps the (limited)
64    // fallback as the merged fallback, while the observability routes above are
65    // matched first and bypass it.
66    let mut git = Router::new().fallback(handle_git).with_state(state);
67    if max_concurrent != 0 {
68        git = git.layer(GlobalConcurrencyLimitLayer::new(max_concurrent));
69    }
70
71    observability.merge(git)
72}
73
74async fn metrics_handler(State(st): State<AppState>) -> Response {
75    Response::builder()
76        .header(header::CONTENT_TYPE, "text/plain; version=0.0.4")
77        .body(Body::from(st.metrics.gather()))
78        .expect("valid response")
79}
80
81/// Readiness probe. Liveness (`/healthz`) only says the process is up; readiness
82/// additionally verifies the proxy can do its one job - write bare mirrors into
83/// the cache root - so a detached, unmounted, read-only, or unwritable cache
84/// volume surfaces as `503 Service Unavailable` here instead of a later flood of
85/// upstream `502`s. Kept out of the concurrency limit (see `router`).
86async fn readyz(State(st): State<AppState>) -> Response {
87    match cache_writable(&st.cache_root).await {
88        Ok(()) => (StatusCode::OK, "ok").into_response(),
89        Err(e) => {
90            tracing::warn!(
91                cache_root = %st.cache_root.display(),
92                error = %e,
93                "readiness check failed"
94            );
95            err(StatusCode::SERVICE_UNAVAILABLE, "cache root not writable")
96        }
97    }
98}
99
100/// Confirm the cache root exists and is writable by creating and removing a probe
101/// file - the same directory the mirrors live in, so it tests the real target
102/// rather than a proxy for it. A single create + unlink is cheap enough to run
103/// per probe and catches the failure modes that make "ready" a lie: a missing
104/// directory, a read-only remount, or a permissions problem.
105async fn cache_writable(cache_root: &Path) -> std::io::Result<()> {
106    tokio::fs::create_dir_all(cache_root).await?;
107    let probe = cache_root.join(".readyz-probe");
108    tokio::fs::write(&probe, b"").await?;
109    let _ = tokio::fs::remove_file(&probe).await;
110    Ok(())
111}
112
113async fn handle_git(State(st): State<AppState>, req: Request<Body>) -> Response {
114    let (parts, body) = req.into_parts();
115    let path = parts.uri.path().to_string();
116    let query = parts.uri.query().unwrap_or("").to_string();
117    let git_protocol = parts
118        .headers
119        .get("git-protocol")
120        .and_then(|v| v.to_str().ok())
121        .map(str::to_string);
122
123    if let Some(resp) = check_auth(&st, &parts.headers) {
124        st.metrics.record_request("auth", "unauthorized", "-");
125        return resp;
126    }
127
128    // Read-only: refuse anything that would write upstream.
129    if path.ends_with(&format!("/{RECEIVE_PACK}"))
130        || query.contains(&format!("service={RECEIVE_PACK}"))
131    {
132        st.metrics.record_request("receive_pack", "rejected", "-");
133        return err(
134            StatusCode::FORBIDDEN,
135            "read-only proxy: pushes are not allowed",
136        );
137    }
138
139    if parts.method == Method::GET && path.ends_with("/info/refs") {
140        if !query.contains(&format!("service={UPLOAD_PACK}")) {
141            st.metrics.record_request("info_refs", "error", "-");
142            return err(
143                StatusCode::BAD_REQUEST,
144                "only smart-http git-upload-pack is supported",
145            );
146        }
147        return info_refs(st, &path, git_protocol.as_deref()).await;
148    }
149
150    if parts.method == Method::POST && path.ends_with(&format!("/{UPLOAD_PACK}")) {
151        let body = match axum::body::to_bytes(body, MAX_BODY).await {
152            Ok(b) => b,
153            Err(_) => return err(StatusCode::BAD_REQUEST, "failed to read request body"),
154        };
155        // Git's smart-HTTP client gzips the upload-pack request body (any
156        // normally-sized want/have negotiation) and sends `Content-Encoding:
157        // gzip`. `git upload-pack` reads its stdin as raw pkt-lines, so we must
158        // undo the transport encoding before handing the body over - otherwise
159        // it chokes on the gzip magic with "bad line length character".
160        let content_encoding = parts
161            .headers
162            .get(header::CONTENT_ENCODING)
163            .and_then(|v| v.to_str().ok());
164        let body = match decode_body(content_encoding, body, st.max_decoded_body) {
165            Ok(b) => b,
166            Err(_) => return err(StatusCode::BAD_REQUEST, "failed to decode request body"),
167        };
168        return upload_pack(st, &path, git_protocol.as_deref(), body).await;
169    }
170
171    err(StatusCode::NOT_FOUND, "not a git smart-http endpoint")
172}
173
174async fn info_refs(st: AppState, path: &str, git_protocol: Option<&str>) -> Response {
175    let Some(name) = repo::repo_name_from_path(path, "/info/refs") else {
176        st.metrics.record_request("info_refs", "error", "-");
177        return err(StatusCode::NOT_FOUND, "bad path");
178    };
179    let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) {
180        Ok(r) => r,
181        Err(e) => {
182            st.metrics.record_request("info_refs", "error", "-");
183            return err(StatusCode::BAD_REQUEST, &e.to_string());
184        }
185    };
186
187    // The upstream clone/fetch counters (per repo) are recorded inside `GitCache`;
188    // here we only account for the client request. The `repo` label is emitted only
189    // once a request is served; failures use `-` so a flood of distinct but doomed
190    // repo paths cannot inflate label cardinality (see `metrics`).
191    if let Err(e) = st.cache.ensure_fresh(&repo, true).await {
192        st.metrics
193            .record_request("info_refs", "upstream_error", "-");
194        tracing::warn!(repo = %name, error = %e, "ensure_fresh failed");
195        return err(StatusCode::BAD_GATEWAY, "upstream fetch failed");
196    }
197
198    match st.cache.advertise_refs(&repo, git_protocol).await {
199        Ok(body) => {
200            st.metrics.record_request("info_refs", "ok", &name);
201            Response::builder()
202                .header(
203                    header::CONTENT_TYPE,
204                    "application/x-git-upload-pack-advertisement",
205                )
206                .header(header::CACHE_CONTROL, "no-cache")
207                .body(Body::from(body))
208                .expect("valid response")
209        }
210        Err(e) => {
211            st.metrics.record_request("info_refs", "error", "-");
212            tracing::warn!(repo = %name, error = %e, "advertise_refs failed");
213            err(StatusCode::INTERNAL_SERVER_ERROR, "advertise-refs failed")
214        }
215    }
216}
217
218async fn upload_pack(
219    st: AppState,
220    path: &str,
221    git_protocol: Option<&str>,
222    body: Bytes,
223) -> Response {
224    let Some(name) = repo::repo_name_from_path(path, &format!("/{UPLOAD_PACK}")) else {
225        st.metrics.record_request("upload_pack", "error", "-");
226        return err(StatusCode::NOT_FOUND, "bad path");
227    };
228    let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) {
229        Ok(r) => r,
230        Err(e) => {
231            st.metrics.record_request("upload_pack", "error", "-");
232            return err(StatusCode::BAD_REQUEST, &e.to_string());
233        }
234    };
235
236    // The preceding info/refs already refreshed; here just ensure the mirror is
237    // present (a client could POST against a not-yet-cloned repo).
238    if let Err(e) = st.cache.ensure_fresh(&repo, false).await {
239        st.metrics
240            .record_request("upload_pack", "upstream_error", "-");
241        tracing::warn!(repo = %name, error = %e, "ensure mirror exists failed");
242        return err(StatusCode::BAD_GATEWAY, "upstream unavailable");
243    }
244
245    match st.cache.upload_pack_rpc(&repo, git_protocol, body).await {
246        Ok(stream) => {
247            st.metrics.record_request("upload_pack", "ok", &name);
248            Response::builder()
249                .header(header::CONTENT_TYPE, "application/x-git-upload-pack-result")
250                .header(header::CACHE_CONTROL, "no-cache")
251                .body(Body::from_stream(stream))
252                .expect("valid response")
253        }
254        Err(e) => {
255            st.metrics.record_request("upload_pack", "error", "-");
256            tracing::warn!(repo = %name, error = %e, "upload_pack_rpc failed");
257            err(StatusCode::INTERNAL_SERVER_ERROR, "upload-pack failed")
258        }
259    }
260}
261
262/// When a serve token is configured, require `Authorization: Bearer <token>`.
263/// Returns `Some(401)` to short-circuit, `None` to allow.
264fn check_auth(st: &AppState, headers: &HeaderMap) -> Option<Response> {
265    let expected = st.serve_token.as_ref()?;
266    let provided = headers
267        .get(header::AUTHORIZATION)
268        .and_then(|v| v.to_str().ok())
269        .and_then(|v| v.strip_prefix("Bearer "));
270    if provided.is_some_and(|t| token_matches(t, expected)) {
271        None
272    } else {
273        Some(err(
274            StatusCode::UNAUTHORIZED,
275            "missing or invalid bearer token",
276        ))
277    }
278}
279
280/// Compare a client-supplied bearer token against the expected one in constant
281/// time, so response latency does not leak how many leading bytes matched. Only
282/// the length can differ observably, which is not sensitive for a shared secret.
283fn token_matches(provided: &str, expected: &str) -> bool {
284    provided.as_bytes().ct_eq(expected.as_bytes()).into()
285}
286
287/// Undo the request's `Content-Encoding`. Git only ever gzips, so that is the
288/// single encoding we decode; an absent/`identity` header passes through
289/// untouched, and any other encoding is rejected by the caller as a bad body.
290///
291/// The decoded size is capped at `max_decoded`: DEFLATE reaches ~1000:1, so an
292/// unbounded read here would let a small compressed body expand into an
293/// out-of-memory kill (a decompression bomb). We read at most `max_decoded + 1`
294/// bytes so we can tell "exactly at the limit" from "over it" and reject the
295/// latter.
296fn decode_body(
297    content_encoding: Option<&str>,
298    body: Bytes,
299    max_decoded: usize,
300) -> std::io::Result<Bytes> {
301    match content_encoding.map(str::trim) {
302        Some(enc) if enc.eq_ignore_ascii_case("gzip") || enc.eq_ignore_ascii_case("x-gzip") => {
303            let mut out = Vec::new();
304            let limit = max_decoded as u64 + 1;
305            flate2::read::GzDecoder::new(&body[..])
306                .take(limit)
307                .read_to_end(&mut out)?;
308            within_limit(Bytes::from(out), max_decoded)
309        }
310        // Uncompressed: no expansion risk, but still enforce the cap so
311        // `--max-decoded-body-mb` bounds an identity request the same as a gzipped
312        // one - otherwise only the coarser transport cap (`MAX_BODY`) would apply.
313        None => within_limit(body, max_decoded),
314        Some(enc) if enc.is_empty() || enc.eq_ignore_ascii_case("identity") => {
315            within_limit(body, max_decoded)
316        }
317        Some(other) => Err(std::io::Error::new(
318            std::io::ErrorKind::InvalidData,
319            format!("unsupported content-encoding: {other}"),
320        )),
321    }
322}
323
324/// Reject a decoded body that exceeds the configured cap. Shared by every encoding
325/// branch so the limit is enforced uniformly, not just when decoding gzip.
326fn within_limit(body: Bytes, max_decoded: usize) -> std::io::Result<Bytes> {
327    if body.len() > max_decoded {
328        return Err(std::io::Error::new(
329            std::io::ErrorKind::InvalidData,
330            "decoded request body exceeds limit",
331        ));
332    }
333    Ok(body)
334}
335
336fn err(status: StatusCode, msg: &str) -> Response {
337    (status, format!("{msg}\n")).into_response()
338}
339
340#[cfg(test)]
341mod tests {
342    use super::*;
343    use std::io::Write;
344
345    fn gzip(bytes: &[u8]) -> Bytes {
346        let mut enc = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::best());
347        enc.write_all(bytes).unwrap();
348        Bytes::from(enc.finish().unwrap())
349    }
350
351    #[test]
352    fn identity_and_absent_encoding_pass_through() {
353        let raw = Bytes::from_static(b"want ...\n");
354        assert_eq!(decode_body(None, raw.clone(), 1024).unwrap(), raw);
355        assert_eq!(
356            decode_body(Some("identity"), raw.clone(), 1024).unwrap(),
357            raw
358        );
359        assert_eq!(decode_body(Some(""), raw.clone(), 1024).unwrap(), raw);
360    }
361
362    #[test]
363    fn identity_body_over_limit_is_rejected() {
364        // An uncompressed body must honor the decoded-body cap too, not just the
365        // gzip path. Exactly at the limit is accepted; one byte over is rejected.
366        let body = Bytes::from(vec![b'x'; 2048]);
367        for enc in [None, Some("identity"), Some("")] {
368            assert!(decode_body(enc, body.clone(), 2048).is_ok());
369            assert!(decode_body(enc, body.clone(), 2047).is_err());
370        }
371    }
372
373    #[test]
374    fn gzip_within_limit_decodes() {
375        let payload = b"command=ls-refs\n";
376        let decoded = decode_body(Some("gzip"), gzip(payload), 1024).unwrap();
377        assert_eq!(&decoded[..], payload);
378        // Case-insensitive and the `x-gzip` alias both decode.
379        assert_eq!(
380            &decode_body(Some("GZIP"), gzip(payload), 1024).unwrap()[..],
381            payload
382        );
383        assert_eq!(
384            &decode_body(Some("x-gzip"), gzip(payload), 1024).unwrap()[..],
385            payload
386        );
387    }
388
389    #[test]
390    fn gzip_decompression_bomb_is_rejected() {
391        // 1 MiB of zeros compresses to ~1 KiB but must not be allowed to expand
392        // past the cap. Exactly-at-limit is accepted; one byte over is rejected.
393        let big = vec![0u8; 1024 * 1024];
394        assert!(decode_body(Some("gzip"), gzip(&big), 1024).is_err());
395        assert!(decode_body(Some("gzip"), gzip(&big), big.len()).is_ok());
396        assert!(decode_body(Some("gzip"), gzip(&big), big.len() - 1).is_err());
397    }
398
399    #[test]
400    fn unsupported_encoding_is_rejected() {
401        assert!(decode_body(Some("br"), Bytes::from_static(b"x"), 1024).is_err());
402        assert!(decode_body(Some("deflate"), Bytes::from_static(b"x"), 1024).is_err());
403    }
404
405    #[test]
406    fn token_matches_only_the_exact_token() {
407        assert!(token_matches("s3cret", "s3cret"));
408        assert!(!token_matches("s3creT", "s3cret")); // last byte differs
409        assert!(!token_matches("s3cre", "s3cret")); // prefix, shorter
410        assert!(!token_matches("s3cret-extra", "s3cret")); // longer
411        assert!(!token_matches("", "s3cret"));
412        assert!(token_matches("", "")); // degenerate empty token
413    }
414}
git_cache_proxy/server.rs

git_cache_proxy/
server.rs
