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::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    let app = Router::new()
49        .route("/healthz", get(|| async { "ok" }))
50        .route("/readyz", get(|| async { "ok" }))
51        .route("/metrics", get(metrics_handler))
52        .fallback(handle_git)
53        .with_state(state);
54    // One global semaphore shared across every per-connection clone of the
55    // service (axum clones it per connection), so the cap is process-wide rather
56    // than per-connection. `0` disables the limit entirely.
57    if max_concurrent == 0 {
58        app
59    } else {
60        app.layer(GlobalConcurrencyLimitLayer::new(max_concurrent))
61    }
62}
63
64async fn metrics_handler(State(st): State<AppState>) -> Response {
65    Response::builder()
66        .header(header::CONTENT_TYPE, "text/plain; version=0.0.4")
67        .body(Body::from(st.metrics.gather()))
68        .expect("valid response")
69}
70
71async fn handle_git(State(st): State<AppState>, req: Request<Body>) -> Response {
72    let (parts, body) = req.into_parts();
73    let path = parts.uri.path().to_string();
74    let query = parts.uri.query().unwrap_or("").to_string();
75    let git_protocol = parts
76        .headers
77        .get("git-protocol")
78        .and_then(|v| v.to_str().ok())
79        .map(str::to_string);
80
81    if let Some(resp) = check_auth(&st, &parts.headers) {
82        st.metrics.record_request("auth", "unauthorized", "-");
83        return resp;
84    }
85
86    // Read-only: refuse anything that would write upstream.
87    if path.ends_with(&format!("/{RECEIVE_PACK}"))
88        || query.contains(&format!("service={RECEIVE_PACK}"))
89    {
90        st.metrics.record_request("receive_pack", "rejected", "-");
91        return err(
92            StatusCode::FORBIDDEN,
93            "read-only proxy: pushes are not allowed",
94        );
95    }
96
97    if parts.method == Method::GET && path.ends_with("/info/refs") {
98        if !query.contains(&format!("service={UPLOAD_PACK}")) {
99            st.metrics.record_request("info_refs", "error", "-");
100            return err(
101                StatusCode::BAD_REQUEST,
102                "only smart-http git-upload-pack is supported",
103            );
104        }
105        return info_refs(st, &path, git_protocol.as_deref()).await;
106    }
107
108    if parts.method == Method::POST && path.ends_with(&format!("/{UPLOAD_PACK}")) {
109        let body = match axum::body::to_bytes(body, MAX_BODY).await {
110            Ok(b) => b,
111            Err(_) => return err(StatusCode::BAD_REQUEST, "failed to read request body"),
112        };
113        // Git's smart-HTTP client gzips the upload-pack request body (any
114        // normally-sized want/have negotiation) and sends `Content-Encoding:
115        // gzip`. `git upload-pack` reads its stdin as raw pkt-lines, so we must
116        // undo the transport encoding before handing the body over - otherwise
117        // it chokes on the gzip magic with "bad line length character".
118        let content_encoding = parts
119            .headers
120            .get(header::CONTENT_ENCODING)
121            .and_then(|v| v.to_str().ok());
122        let body = match decode_body(content_encoding, body, st.max_decoded_body) {
123            Ok(b) => b,
124            Err(_) => return err(StatusCode::BAD_REQUEST, "failed to decode request body"),
125        };
126        return upload_pack(st, &path, git_protocol.as_deref(), body).await;
127    }
128
129    err(StatusCode::NOT_FOUND, "not a git smart-http endpoint")
130}
131
132async fn info_refs(st: AppState, path: &str, git_protocol: Option<&str>) -> Response {
133    let Some(name) = repo::repo_name_from_path(path, "/info/refs") else {
134        st.metrics.record_request("info_refs", "error", "-");
135        return err(StatusCode::NOT_FOUND, "bad path");
136    };
137    let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) {
138        Ok(r) => r,
139        Err(e) => {
140            st.metrics.record_request("info_refs", "error", "-");
141            return err(StatusCode::BAD_REQUEST, &e.to_string());
142        }
143    };
144
145    // The upstream clone/fetch counters (per repo) are recorded inside `GitCache`;
146    // here we only account for the client request. The `repo` label is emitted only
147    // once a request is served; failures use `-` so a flood of distinct but doomed
148    // repo paths cannot inflate label cardinality (see `metrics`).
149    if let Err(e) = st.cache.ensure_fresh(&repo, true).await {
150        st.metrics
151            .record_request("info_refs", "upstream_error", "-");
152        tracing::warn!(repo = %name, error = %e, "ensure_fresh failed");
153        return err(StatusCode::BAD_GATEWAY, "upstream fetch failed");
154    }
155
156    match st.cache.advertise_refs(&repo, git_protocol).await {
157        Ok(body) => {
158            st.metrics.record_request("info_refs", "ok", &name);
159            Response::builder()
160                .header(
161                    header::CONTENT_TYPE,
162                    "application/x-git-upload-pack-advertisement",
163                )
164                .header(header::CACHE_CONTROL, "no-cache")
165                .body(Body::from(body))
166                .expect("valid response")
167        }
168        Err(e) => {
169            st.metrics.record_request("info_refs", "error", "-");
170            tracing::warn!(repo = %name, error = %e, "advertise_refs failed");
171            err(StatusCode::INTERNAL_SERVER_ERROR, "advertise-refs failed")
172        }
173    }
174}
175
176async fn upload_pack(
177    st: AppState,
178    path: &str,
179    git_protocol: Option<&str>,
180    body: Bytes,
181) -> Response {
182    let Some(name) = repo::repo_name_from_path(path, &format!("/{UPLOAD_PACK}")) else {
183        st.metrics.record_request("upload_pack", "error", "-");
184        return err(StatusCode::NOT_FOUND, "bad path");
185    };
186    let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) {
187        Ok(r) => r,
188        Err(e) => {
189            st.metrics.record_request("upload_pack", "error", "-");
190            return err(StatusCode::BAD_REQUEST, &e.to_string());
191        }
192    };
193
194    // The preceding info/refs already refreshed; here just ensure the mirror is
195    // present (a client could POST against a not-yet-cloned repo).
196    if let Err(e) = st.cache.ensure_fresh(&repo, false).await {
197        st.metrics
198            .record_request("upload_pack", "upstream_error", "-");
199        tracing::warn!(repo = %name, error = %e, "ensure mirror exists failed");
200        return err(StatusCode::BAD_GATEWAY, "upstream unavailable");
201    }
202
203    match st.cache.upload_pack_rpc(&repo, git_protocol, body).await {
204        Ok(stream) => {
205            st.metrics.record_request("upload_pack", "ok", &name);
206            Response::builder()
207                .header(header::CONTENT_TYPE, "application/x-git-upload-pack-result")
208                .header(header::CACHE_CONTROL, "no-cache")
209                .body(Body::from_stream(stream))
210                .expect("valid response")
211        }
212        Err(e) => {
213            st.metrics.record_request("upload_pack", "error", "-");
214            tracing::warn!(repo = %name, error = %e, "upload_pack_rpc failed");
215            err(StatusCode::INTERNAL_SERVER_ERROR, "upload-pack failed")
216        }
217    }
218}
219
220/// When a serve token is configured, require `Authorization: Bearer <token>`.
221/// Returns `Some(401)` to short-circuit, `None` to allow.
222fn check_auth(st: &AppState, headers: &HeaderMap) -> Option<Response> {
223    let expected = st.serve_token.as_ref()?;
224    let provided = headers
225        .get(header::AUTHORIZATION)
226        .and_then(|v| v.to_str().ok())
227        .and_then(|v| v.strip_prefix("Bearer "));
228    if provided.is_some_and(|t| token_matches(t, expected)) {
229        None
230    } else {
231        Some(err(
232            StatusCode::UNAUTHORIZED,
233            "missing or invalid bearer token",
234        ))
235    }
236}
237
238/// Compare a client-supplied bearer token against the expected one in constant
239/// time, so response latency does not leak how many leading bytes matched. Only
240/// the length can differ observably, which is not sensitive for a shared secret.
241fn token_matches(provided: &str, expected: &str) -> bool {
242    provided.as_bytes().ct_eq(expected.as_bytes()).into()
243}
244
245/// Undo the request's `Content-Encoding`. Git only ever gzips, so that is the
246/// single encoding we decode; an absent/`identity` header passes through
247/// untouched, and any other encoding is rejected by the caller as a bad body.
248///
249/// The decoded size is capped at `max_decoded`: DEFLATE reaches ~1000:1, so an
250/// unbounded read here would let a small compressed body expand into an
251/// out-of-memory kill (a decompression bomb). We read at most `max_decoded + 1`
252/// bytes so we can tell "exactly at the limit" from "over it" and reject the
253/// latter.
254fn decode_body(
255    content_encoding: Option<&str>,
256    body: Bytes,
257    max_decoded: usize,
258) -> std::io::Result<Bytes> {
259    match content_encoding.map(str::trim) {
260        Some(enc) if enc.eq_ignore_ascii_case("gzip") || enc.eq_ignore_ascii_case("x-gzip") => {
261            let mut out = Vec::new();
262            let limit = max_decoded as u64 + 1;
263            flate2::read::GzDecoder::new(&body[..])
264                .take(limit)
265                .read_to_end(&mut out)?;
266            within_limit(Bytes::from(out), max_decoded)
267        }
268        // Uncompressed: no expansion risk, but still enforce the cap so
269        // `--max-decoded-body-mb` bounds an identity request the same as a gzipped
270        // one - otherwise only the coarser transport cap (`MAX_BODY`) would apply.
271        None => within_limit(body, max_decoded),
272        Some(enc) if enc.is_empty() || enc.eq_ignore_ascii_case("identity") => {
273            within_limit(body, max_decoded)
274        }
275        Some(other) => Err(std::io::Error::new(
276            std::io::ErrorKind::InvalidData,
277            format!("unsupported content-encoding: {other}"),
278        )),
279    }
280}
281
282/// Reject a decoded body that exceeds the configured cap. Shared by every encoding
283/// branch so the limit is enforced uniformly, not just when decoding gzip.
284fn within_limit(body: Bytes, max_decoded: usize) -> std::io::Result<Bytes> {
285    if body.len() > max_decoded {
286        return Err(std::io::Error::new(
287            std::io::ErrorKind::InvalidData,
288            "decoded request body exceeds limit",
289        ));
290    }
291    Ok(body)
292}
293
294fn err(status: StatusCode, msg: &str) -> Response {
295    (status, format!("{msg}\n")).into_response()
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301    use std::io::Write;
302
303    fn gzip(bytes: &[u8]) -> Bytes {
304        let mut enc = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::best());
305        enc.write_all(bytes).unwrap();
306        Bytes::from(enc.finish().unwrap())
307    }
308
309    #[test]
310    fn identity_and_absent_encoding_pass_through() {
311        let raw = Bytes::from_static(b"want ...\n");
312        assert_eq!(decode_body(None, raw.clone(), 1024).unwrap(), raw);
313        assert_eq!(
314            decode_body(Some("identity"), raw.clone(), 1024).unwrap(),
315            raw
316        );
317        assert_eq!(decode_body(Some(""), raw.clone(), 1024).unwrap(), raw);
318    }
319
320    #[test]
321    fn identity_body_over_limit_is_rejected() {
322        // An uncompressed body must honor the decoded-body cap too, not just the
323        // gzip path. Exactly at the limit is accepted; one byte over is rejected.
324        let body = Bytes::from(vec![b'x'; 2048]);
325        for enc in [None, Some("identity"), Some("")] {
326            assert!(decode_body(enc, body.clone(), 2048).is_ok());
327            assert!(decode_body(enc, body.clone(), 2047).is_err());
328        }
329    }
330
331    #[test]
332    fn gzip_within_limit_decodes() {
333        let payload = b"command=ls-refs\n";
334        let decoded = decode_body(Some("gzip"), gzip(payload), 1024).unwrap();
335        assert_eq!(&decoded[..], payload);
336        // Case-insensitive and the `x-gzip` alias both decode.
337        assert_eq!(
338            &decode_body(Some("GZIP"), gzip(payload), 1024).unwrap()[..],
339            payload
340        );
341        assert_eq!(
342            &decode_body(Some("x-gzip"), gzip(payload), 1024).unwrap()[..],
343            payload
344        );
345    }
346
347    #[test]
348    fn gzip_decompression_bomb_is_rejected() {
349        // 1 MiB of zeros compresses to ~1 KiB but must not be allowed to expand
350        // past the cap. Exactly-at-limit is accepted; one byte over is rejected.
351        let big = vec![0u8; 1024 * 1024];
352        assert!(decode_body(Some("gzip"), gzip(&big), 1024).is_err());
353        assert!(decode_body(Some("gzip"), gzip(&big), big.len()).is_ok());
354        assert!(decode_body(Some("gzip"), gzip(&big), big.len() - 1).is_err());
355    }
356
357    #[test]
358    fn unsupported_encoding_is_rejected() {
359        assert!(decode_body(Some("br"), Bytes::from_static(b"x"), 1024).is_err());
360        assert!(decode_body(Some("deflate"), Bytes::from_static(b"x"), 1024).is_err());
361    }
362
363    #[test]
364    fn token_matches_only_the_exact_token() {
365        assert!(token_matches("s3cret", "s3cret"));
366        assert!(!token_matches("s3creT", "s3cret")); // last byte differs
367        assert!(!token_matches("s3cre", "s3cret")); // prefix, shorter
368        assert!(!token_matches("s3cret-extra", "s3cret")); // longer
369        assert!(!token_matches("", "s3cret"));
370        assert!(token_matches("", "")); // degenerate empty token
371    }
372}
git_cache_proxy/server.rs

git_cache_proxy/
server.rs
