zuka
zuka/src/api/git_http.rs

worklyn / zukapublic

Agent-first git hosting. One Rust binary: git over HTTP and SSH, a REST API, MCP, CI, and multi-tenant isolation.

Get a copy: git clone https://zuka.worklyn.com/worklyn/zuka.git
branchf609407da207bcd3baf025777727ebdce314a69d ▾
main All branches →
zuka/src/api/git_http.rs
RSgit_http.rs5.8 KBDownload
1// The git wire protocol endpoints.
2//
3// These are not REST and do not use problem+json: git prints the response body at
4// the user, and a 401 without `WWW-Authenticate` makes a clone fail rather than
5// prompt for credentials (SPEC §4.7).
6//
7// Nothing here buffers a pack. The request body is handed to the CGI as a reader
8// and the CGI's stdout becomes the response body, so memory is one chunk rather
9// than two copies of the pack.
10
11use crate::error::{Error, Result};
12use crate::git::transport::{self, CgiRequest, CgiResponse, GitRoute};
13use crate::git::validate;
14use crate::http::response::{self, Body};
15use crate::http::AppState;
16use bytes::Bytes;
17use futures::TryStreamExt;
18use http_body_util::{BodyExt, StreamBody};
19use hyper::body::{Frame, Incoming};
20use hyper::{Request, Response, StatusCode};
21use std::net::SocketAddr;
22use std::sync::Arc;
23use tokio_util::io::{ReaderStream, StreamReader};
24
25pub async fn handle(
26 req: Request<Incoming>,
27 state: Arc<AppState>,
28 route: GitRoute,
29 peer: SocketAddr,
30) -> Result<(Response<Body>, Option<String>)> {
31 // Validate before anything reaches GIT_PROJECT_ROOT + PATH_INFO. Relying on a
32 // later metadata lookup to reject traversal makes the control incidental.
33 let account = validate::name(&route.account)?;
34 let repo = validate::name(&route.repo)?;
35
36 let identity = state.identify(req.headers(), req.method().as_str(), req.uri().path())?;
37
38 // Advertising refs for receive-pack is itself a write signal, so the same check
39 // covers the advertisement and the push.
40 if route.writes {
41 identity.require_write(&account, &repo)?;
42 } else {
43 identity.require_read(&account, &repo)?;
44 }
45
46 if state.meta.get_ready_repo(&account, &repo)?.is_none() {
47 return Err(Error::NotFound("repository"));
48 }
49
50 // receive-pack interrupted by ENOSPC leaves a partial pack, so the reserve is
51 // checked before the child is spawned, not after.
52 if route.writes {
53 state.ensure_space()?;
54 state.check_write_quota(&account, &repo)?;
55 }
56
57 let method = req.method().as_str().to_string();
58 let headers = req.headers().clone();
59 let content_type = header(&headers, "content-type");
60 let content_encoding = header(&headers, "content-encoding");
61 let git_protocol = header(&headers, "git-protocol");
62 let content_length = headers
63 .get("content-length")
64 .and_then(|v| v.to_str().ok())
65 .and_then(|v| v.parse::<u64>().ok());
66
67 let peer_addr = peer.ip().to_string();
68
69 // Stream the request body into the child rather than collecting it. git switches
70 // to chunked for anything over http.postBuffer (1 MiB by default), so the
71 // buffered path was the normal path for pushes.
72 let reader = StreamReader::new(
73 req.into_body()
74 .into_data_stream()
75 .map_err(|e| std::io::Error::other(e.to_string())),
76 );
77
78 let cgi = CgiRequest {
79 method: &method,
80 path_info: &route.path_info,
81 query: &route.query,
82 content_type: content_type.as_deref(),
83 content_encoding: content_encoding.as_deref(),
84 git_protocol: git_protocol.as_deref(),
85 remote_user: account.as_str(),
86 remote_addr: &peer_addr,
87 content_length,
88 hook_env: &state.config.hook_env(),
89 };
90
91 let produced = transport::spawn(
92 state.git.git_root(),
93 cgi,
94 reader,
95 state.config.max_pack_bytes,
96 Arc::clone(&state.git_slots),
97 )
98 .await?;
99
100 let response = build_response(produced, &route)?;
101 Ok((response, Some(account.as_str().to_string())))
102}
103
104/// Turn a running CGI into a streaming response.
105///
106/// The guard rides inside the body, so the child and its concurrency permit are
107/// released exactly when the client stops reading — including on disconnect.
108fn build_response(produced: CgiResponse, route: &GitRoute) -> Result<Response<Body>> {
109 let CgiResponse {
110 status,
111 headers,
112 prefix,
113 stdout,
114 guard,
115 } = produced;
116
117 let status = StatusCode::from_u16(status).unwrap_or(StatusCode::OK);
118
119 let forwarded: Vec<(String, String)> = headers
120 .into_iter()
121 .filter(|(name, _)| !transport::is_suppressed(name))
122 .collect();
123
124 let tail = ReaderStream::new(stdout).map_ok(Frame::data);
125 let head = futures::stream::once(async move { Ok(Frame::data(prefix)) });
126 let joined = futures::StreamExt::chain(head, tail);
127
128 // Holding the guard in the stream's scope is what ties the child's lifetime to
129 // the response body rather than to this function.
130 let body = StreamBody::new(GuardedStream {
131 inner: Box::pin(joined),
132 _guard: guard,
133 })
134 .boxed();
135
136 // A stale ref advertisement makes a client negotiate against objects the server
137 // no longer has, so it must never be cached.
138 let extra: &[(&str, &str)] = if route.path_info.ends_with("/info/refs") {
139 transport::NO_CACHE
140 } else {
141 &[]
142 };
143
144 response::stream(status, forwarded, extra, body)
145 .map_err(|e| Error::Internal(anyhow::Error::from(e).context("build git response")))
146}
147
148/// A stream that keeps a value alive for as long as it is being polled.
149struct GuardedStream<S> {
150 inner: std::pin::Pin<Box<S>>,
151 _guard: transport::CgiGuard,
152}
153
154impl<S> futures::Stream for GuardedStream<S>
155where
156 S: futures::Stream<Item = std::result::Result<Frame<Bytes>, std::io::Error>>,
157{
158 type Item = std::result::Result<Frame<Bytes>, std::io::Error>;
159
160 fn poll_next(
161 mut self: std::pin::Pin<&mut Self>,
162 cx: &mut std::task::Context<'_>,
163 ) -> std::task::Poll<Option<Self::Item>> {
164 self.inner.as_mut().poll_next(cx)
165 }
166}
167
168fn header(headers: &hyper::HeaderMap, name: &str) -> Option<String> {
169 headers
170 .get(name)
171 .and_then(|v| v.to_str().ok())
172 .map(String::from)
173}