use std::io::Read;
use std::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::metrics::Metrics;
use crate::repo;
const MAX_BODY: usize = 64 * 1024 * 1024;
const UPLOAD_PACK: &str = "git-upload-pack";
const RECEIVE_PACK: &str = "git-receive-pack";
#[derive(Clone)]
pub struct AppState {
pub cache: Arc<GitCache>,
pub upstream_base: String,
pub cache_root: PathBuf,
pub serve_token: Option<String>,
pub max_decoded_body: usize,
pub max_concurrent: usize,
pub metrics: Arc<Metrics>,
}
pub fn router(state: AppState) -> Router {
let max_concurrent = state.max_concurrent;
let app = Router::new()
.route("/healthz", get(|| async { "ok" }))
.route("/readyz", get(|| async { "ok" }))
.route("/metrics", get(metrics_handler))
.fallback(handle_git)
.with_state(state);
if max_concurrent == 0 {
app
} else {
app.layer(GlobalConcurrencyLimitLayer::new(max_concurrent))
}
}
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")
}
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("auth", "unauthorized", "-");
return resp;
}
if path.ends_with(&format!("/{RECEIVE_PACK}"))
|| query.contains(&format!("service={RECEIVE_PACK}"))
{
st.metrics.record_request("receive_pack", "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("info_refs", "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"),
};
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;
}
err(StatusCode::NOT_FOUND, "not a git smart-http endpoint")
}
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("info_refs", "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("info_refs", "error", "-");
return err(StatusCode::BAD_REQUEST, &e.to_string());
}
};
if let Err(e) = st.cache.ensure_fresh(&repo, true).await {
st.metrics
.record_request("info_refs", "upstream_error", "-");
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("info_refs", "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("info_refs", "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("upload_pack", "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("upload_pack", "error", "-");
return err(StatusCode::BAD_REQUEST, &e.to_string());
}
};
if let Err(e) = st.cache.ensure_fresh(&repo, false).await {
st.metrics
.record_request("upload_pack", "upstream_error", "-");
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("upload_pack", "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("upload_pack", "error", "-");
tracing::warn!(repo = %name, error = %e, "upload_pack_rpc failed");
err(StatusCode::INTERNAL_SERVER_ERROR, "upload-pack failed")
}
}
}
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",
))
}
}
fn token_matches(provided: &str, expected: &str) -> bool {
provided.as_bytes().ct_eq(expected.as_bytes()).into()
}
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)?;
if out.len() > max_decoded {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"decoded request body exceeds limit",
));
}
Ok(Bytes::from(out))
}
None => Ok(body),
Some(enc) if enc.is_empty() || enc.eq_ignore_ascii_case("identity") => Ok(body),
Some(other) => Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("unsupported content-encoding: {other}"),
)),
}
}
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 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);
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() {
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")); assert!(!token_matches("s3cre", "s3cret")); assert!(!token_matches("s3cret-extra", "s3cret")); assert!(!token_matches("", "s3cret"));
assert!(token_matches("", "")); }
}