zuka
zuka/src/http/mod.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
branchbc985d6bbd321be340e3750b3da4734ee1553cbd ▾
main All branches →
zuka/src/http/mod.rs
RSmod.rs11.9 KBDownload
1// Routing and the shared application state.
2//
3// `handle` is infallible and owns response defaults; `route` returns `Result` so
4// every handler can use `?` and have one conversion point to an error response.
5
6pub mod auth;
7pub mod ratelimit;
8pub mod request;
9pub mod response;
10
11use crate::account::assertion::Verifier_;
12use crate::account::key::KeyFile;
13use crate::account::token::TokenFile;
14use crate::api;
15use crate::ci::run::RunStore;
16use crate::config::Config;
17use crate::error::{Error, Result};
18use crate::git::store::GitStore;
19use crate::git::transport;
20use crate::store::cached::Cached;
21use crate::store::fs::FsStore;
22use crate::store::quota;
23use hyper::body::Incoming;
24use hyper::{Method, Request, Response, StatusCode};
25use ratelimit::RateLimiter;
26use response::Body;
27use std::collections::HashSet;
28use std::convert::Infallible;
29use std::net::SocketAddr;
30use std::sync::atomic::{AtomicU64, Ordering};
31use std::sync::{Arc, Mutex};
32use std::time::Instant;
33use tokio::sync::Semaphore;
34
35pub struct AppState {
36 pub config: Config,
37 pub git: GitStore,
38 pub meta: FsStore,
39 /// Bounds concurrent `git` subprocesses. Without it, N simultaneous clones each
40 /// spawn a `pack-objects` and the host runs out of memory — measured at 1.5 GB
41 /// for four clones of a 257 MB repository.
42 pub git_slots: Arc<Semaphore>,
43 pub limiter: RateLimiter,
44 pub runs: RunStore,
45 /// Present only in tenant mode: verifies the control plane's assertions.
46 pub verifier: Option<Verifier_>,
47 /// Ids of runs this process is currently executing.
48 ///
49 /// The claim is in memory because it only has to be unique within the executor;
50 /// a restart re-derives the queue from the records themselves.
51 in_flight: Mutex<HashSet<String>>,
52 /// Parsed once and re-parsed only when the file changes, so a credential added
53 /// by the CLI works without a restart and a stable file is not re-read per
54 /// request (see `store::cached`).
55 tokens: Cached<TokenFile>,
56 ssh_keys: Cached<KeyFile>,
57 requests: AtomicU64,
58}
59
60impl AppState {
61 pub fn new(config: Config) -> Result<Self> {
62 let git = GitStore::open(&config.data_dir, config.max_pack_bytes)?;
63 let meta = FsStore::open(config.meta_dir())?;
64 let tokens = Cached::load(config.tokens_file(), TokenFile::load)?;
65 let ssh_keys = Cached::load(config.keys_file(), KeyFile::load)?;
66 let git_slots = Arc::new(Semaphore::new(config.max_concurrent_git));
67 let limiter = RateLimiter::new(config.rate_per_minute);
68 let runs = RunStore::open(config.runs_dir())?;
69 let verifier = if config.mode == crate::config::Mode::Tenant {
70 Some(Verifier_::new(&config.control_public_keys)?)
71 } else {
72 None
73 };
74
75 Ok(AppState {
76 config,
77 git,
78 meta,
79 git_slots,
80 limiter,
81 runs,
82 verifier,
83 in_flight: Mutex::new(HashSet::new()),
84 tokens,
85 ssh_keys,
86 requests: AtomicU64::new(0),
87 })
88 }
89
90 pub fn tokens(&self) -> TokenFile {
91 self.tokens.get()
92 }
93
94 pub fn ssh_keys(&self) -> KeyFile {
95 self.ssh_keys.get()
96 }
97
98 /// Replace the SSH key set, persisting it first.
99 pub fn put_ssh_keys(&self, keys: KeyFile) -> Result<()> {
100 let to_write = keys.clone();
101 self.ssh_keys.put(keys, |path| to_write.save(path))
102 }
103
104 pub fn next_request_id(&self) -> u64 {
105 self.requests.fetch_add(1, Ordering::Relaxed)
106 }
107
108 /// Storage limits, as configured.
109 pub fn quota_limits(&self) -> quota::Limits {
110 quota::Limits {
111 max_repos: self.config.max_repos_per_account,
112 max_repo_bytes: self.config.max_repo_bytes,
113 max_account_bytes: self.config.max_account_bytes,
114 }
115 }
116
117 /// Current storage usage for an account.
118 pub fn usage(&self, account: &crate::git::validate::Name) -> quota::Usage {
119 quota::account_usage(self.git.git_root(), account)
120 }
121
122 /// Whether an account may create another repository.
123 pub fn check_repo_quota(&self, account: &crate::git::validate::Name) -> Result<()> {
124 let limits = self.quota_limits();
125 if limits.max_repos == 0 {
126 return Ok(());
127 }
128 quota::check_repo_count(&self.usage(account), limits)
129 }
130
131 /// Whether a repository may accept a push.
132 ///
133 /// Over-quota blocks writes only: reads and clones keep working, because taking
134 /// someone's data away is the wrong answer to them having too much of it.
135 pub fn check_write_quota(
136 &self,
137 account: &crate::git::validate::Name,
138 repo: &crate::git::validate::Name,
139 ) -> Result<()> {
140 let limits = self.quota_limits();
141 if limits.disabled() {
142 return Ok(());
143 }
144 let path = self.git.repo_path(account, repo)?;
145 quota::check_writable(quota::repo_bytes(&path), &self.usage(account), limits)
146 }
147
148 /// Resolve the caller for a request.
149 ///
150 /// One place, so the two modes cannot drift: standalone checks the local token
151 /// file, tenant checks the control plane's signed assertion and ignores any
152 /// token the client sent.
153 pub fn identify(
154 &self,
155 headers: &hyper::HeaderMap,
156 method: &str,
157 path: &str,
158 ) -> Result<crate::account::Identity> {
159 match &self.verifier {
160 Some(verifier) => auth::resolve_assertion(verifier, headers, method, path),
161 None => {
162 let tokens = self.tokens();
163 auth::resolve_local(&tokens, &auth::extract(headers))
164 }
165 }
166 }
167
168 /// Number of runs this process is executing.
169 pub fn running_runs(&self) -> usize {
170 self.in_flight.lock().expect("run lock").len()
171 }
172
173 /// Claim a queued run. Returns false when it is already being executed.
174 pub fn claim_run(&self, id: &str) -> bool {
175 self.in_flight
176 .lock()
177 .expect("run lock")
178 .insert(id.to_string())
179 }
180
181 pub fn release_run(&self, id: &str) {
182 self.in_flight.lock().expect("run lock").remove(id);
183 }
184
185 /// Take one request's allowance for `key`.
186 pub fn check_rate(&self, key: &str) -> Result<()> {
187 match self.limiter.check(key) {
188 Ok(()) => Ok(()),
189 Err(retry_after) => Err(Error::RateLimited { retry_after }),
190 }
191 }
192
193 /// Refuse a write while free space is below the reserve.
194 ///
195 /// Checked before the write: `git` interrupted by ENOSPC leaves a partial pack,
196 /// and a repository created without room for its object store is corrupt rather
197 /// than empty (SPEC §4.9).
198 pub fn ensure_space(&self) -> Result<()> {
199 match available_bytes(&self.config.data_dir) {
200 Some(free) if free < self.config.disk_reserve_bytes => Err(Error::StorageFull),
201 _ => Ok(()),
202 }
203 }
204}
205
206/// Entry point. Never fails: an error becomes a response.
207pub async fn handle(
208 request: Request<Incoming>,
209 state: Arc<AppState>,
210 peer: SocketAddr,
211) -> std::result::Result<Response<Body>, Infallible> {
212 let method = request.method().clone();
213 let path = request.uri().path().to_string();
214 let request_id = state.next_request_id();
215 let started = Instant::now();
216
217 // Parsed once and handed to `route`: it allocates three strings, and doing it
218 // twice per request was pure waste.
219 let git_route = transport::parse_route(&path, request.uri().query().unwrap_or(""));
220 let is_git_path = git_route.is_some();
221
222 let (mut result, account) = match route(request, Arc::clone(&state), peer, git_route).await {
223 Ok((response, account)) => (response, account),
224 Err(error) => {
225 if let Some(cause) = error.cause() {
226 eprintln!(
227 "[http] req={request_id} {method} {path} status={} error={cause:#}",
228 error.status()
229 );
230 }
231 // git renders a JSON error body as noise at the user's terminal, so the
232 // wire paths get plain text across the whole error range — not just 401
233 // (SPEC §4.7).
234 let response = if is_git_path {
235 response::git_error(&error)
236 } else {
237 response::problem(&error)
238 };
239 (response, None)
240 }
241 };
242
243 response::apply_defaults(&mut result);
244 eprintln!(
245 "[http] req={request_id} method={method} path={path} status={} account={} dur={:?}",
246 result.status().as_u16(),
247 account.as_deref().unwrap_or("-"),
248 started.elapsed()
249 );
250 Ok(result)
251}
252
253/// Returns the response plus the account it was served for, for the access log.
254async fn route(
255 request: Request<Incoming>,
256 state: Arc<AppState>,
257 peer: SocketAddr,
258 git_route: Option<transport::GitRoute>,
259) -> Result<(Response<Body>, Option<String>)> {
260 let path = request.uri().path().to_string();
261 let method = request.method().clone();
262
263 if path == "/healthz" {
264 return Ok((health(&state)?, None));
265 }
266 if path == "/openapi.json" {
267 return Ok((
268 response::json_str(StatusCode::OK, crate::api::openapi::DOCUMENT),
269 None,
270 ));
271 }
272
273 if path == "/mcp" {
274 return crate::mcp::handle(request, state).await;
275 }
276
277 if let Some(git_route) = git_route {
278 return api::git_http::handle(request, state, git_route, peer).await;
279 }
280
281 let segments = request::segments(&path);
282 if segments.first() == Some(&"v1") {
283 return api::route(request, state, &segments[1..]).await;
284 }
285
286 if method == Method::OPTIONS {
287 // CORS stays off. The viewer is same-origin server-rendered HTML and needs
288 // none of it, and CORS plus bearer auth plus attacker-controlled bytes is
289 // the standard cross-origin theft recipe (SPEC §4.8).
290 return Ok((response::no_content(), None));
291 }
292
293 // Last, so it can never shadow an API or transport route: a repository named
294 // `v1` must not be able to take over the API prefix by existing.
295 if crate::web::owns(&path) {
296 return Ok((crate::web::handle(request, state).await?, None));
297 }
298
299 Err(Error::NotFound("route"))
300}
301
302/// Liveness plus the disk headroom writes depend on.
303fn health(state: &AppState) -> Result<Response<Body>> {
304 let free = available_bytes(&state.config.data_dir);
305 let degraded = free.is_some_and(|f| f < state.config.disk_reserve_bytes);
306
307 let body = serde_json::json!({
308 "status": if degraded { "degraded" } else { "ok" },
309 "mode": state.config.mode.as_str(),
310 "version": env!("CARGO_PKG_VERSION"),
311 "disk_free_bytes": free,
312 "disk_reserve_bytes": state.config.disk_reserve_bytes,
313 "git_slots_available": state.git_slots.available_permits(),
314 "rate_limit_per_minute": state.config.rate_per_minute,
315 "ci_enabled": state.config.ci_enabled,
316 "ci_running": state.running_runs(),
317
318 });
319
320 let status = if degraded {
321 StatusCode::SERVICE_UNAVAILABLE
322 } else {
323 StatusCode::OK
324 };
325 Ok(response::json(status, &body))
326}
327
328/// Free bytes on the filesystem holding `path`, or `None` where unavailable.
329///
330/// Uses `libc`'s `statvfs` rather than a hand-declared struct: the layout differs
331/// between macOS and Linux, and getting it wrong reads garbage rather than failing.
332#[cfg(unix)]
333pub fn available_bytes(path: &std::path::Path) -> Option<u64> {
334 use std::ffi::CString;
335 use std::os::unix::ffi::OsStrExt;
336
337 let c_path = CString::new(path.as_os_str().as_bytes()).ok()?;
338 // SAFETY: `c_path` is a valid NUL-terminated path; `stat` is only read after
339 // statvfs reports success.
340 unsafe {
341 let mut stat: libc::statvfs = std::mem::zeroed();
342 if libc::statvfs(c_path.as_ptr(), &mut stat) != 0 {
343 return None;
344 }
345 Some((stat.f_bavail as u64).saturating_mul(stat.f_frsize as u64))
346 }
347}
348
349#[cfg(not(unix))]
350pub fn available_bytes(_path: &std::path::Path) -> Option<u64> {
351 None
352}