forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
17.5 KiB
1//! Git LFS routes: the batch API, object GET and PUT, and verify.
2//!
3//! Each route accepts either a `RemoteAuth <token>` action token (issued by the batch API or by
4//! the SSH `git-lfs-authenticate` command, scoped to one repo) or HTTP Basic with a personal access
5//! token, or no credentials for public reads. The `{repo}` segment may carry the `.git` suffix,
6//! which `find_repo` strips, so both the `.git` and the bare paths reach these handlers.
7
8use std::net::IpAddr;
9use std::path::Path;
10use std::sync::atomic::{AtomicU64, Ordering};
11use std::time::{SystemTime, UNIX_EPOCH};
12
13use forge_core::Action;
14use forge_core::lfs_token::{self, LfsOp};
15use futures_util::{StreamExt, TryStreamExt};
16use http::header::{AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, HeaderValue};
17use http_body::Frame;
18use http_body_util::StreamBody;
19use serde::Deserialize;
20use serde_json::{Value, json};
21use sha2::{Digest, Sha256};
22use tokio::io::AsyncWriteExt;
23use tokio_util::io::ReaderStream;
24use topcoat::context::{Cx, app_context};
25use topcoat::router::request::headers;
26use topcoat::router::response::Response;
27use topcoat::router::{Body, HeaderMap, StatusCode, path_param, route, to_bytes};
28
29use super::{is_valid_oid, object_path, tmp_dir};
30use crate::access::{AccessError, find_repo, resolve_repo_for};
31use crate::client_ip::client_ip;
32use crate::git_http::{access_error, authenticate};
33use crate::state::AppState;
34
35path_param!(owner);
36path_param!(repo);
37path_param!(oid);
38
39/// Media type of LFS JSON bodies.
40const LFS_JSON: &str = "application/vnd.git-lfs+json";
41/// Lifetime of action tokens, in seconds.
42const TOKEN_TTL_SECS: i64 = 3600;
43/// Largest batch request body accepted.
44const MAX_BATCH_BYTES: usize = 4 * 1024 * 1024;
45/// Most objects one batch may name.
46const MAX_BATCH_OBJECTS: usize = 1000;
47/// Largest verify request body accepted.
48const MAX_VERIFY_BYTES: usize = 4096;
49
50/// Distinguishes the temp files of concurrent uploads in one process.
51static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
52
53#[derive(Deserialize)]
54struct BatchRequest {
55 operation: String,
56 #[serde(default)]
57 transfers: Vec<String>,
58 #[serde(default)]
59 hash_algo: Option<String>,
60 objects: Vec<BatchObject>,
61}
62
63#[derive(Deserialize)]
64struct BatchObject {
65 oid: String,
66 size: u64,
67}
68
69#[derive(Deserialize)]
70struct VerifyRequest {
71 oid: String,
72 size: u64,
73}
74
75#[route(POST "/{owner}/{repo}/info/lfs/objects/batch")]
76async fn batch(cx: &Cx, body: Body) -> topcoat::Result<Response> {
77 let state = app_context::<AppState>(cx);
78 let headers = headers(cx);
79 let owner = path_param::<Owner>(cx);
80 let name = path_param::<Repo>(cx);
81
82 let bytes = to_bytes(body, MAX_BATCH_BYTES).await?;
83 let BatchPlan { request, action } = match parse_batch_operation(&bytes) {
84 Ok(plan) => plan,
85 Err(response) => return Ok(*response),
86 };
87 let max = state.config.lfs.max_object_bytes();
88 for object in &request.objects {
89 if !is_valid_oid(&object.oid) {
90 return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
91 }
92 if object.size > max {
93 return Ok(message(
94 StatusCode::UNPROCESSABLE_ENTITY,
95 "an object exceeds the size limit",
96 ));
97 }
98 }
99
100 let repo = match authorize(state, client_ip(cx), headers, owner, name, action).await {
101 Ok(repo) => repo,
102 Err(response) => return Ok(*response),
103 };
104 let writing = matches!(action, Action::Write);
105 let writable = if writing {
106 ensure_writable(state, &repo).await
107 } else {
108 Ok(())
109 };
110 if let Err(response) = writable {
111 return Ok(*response);
112 }
113
114 let token_op = if writing {
115 LfsOp::Upload
116 } else {
117 LfsOp::Download
118 };
119 let base = batch_base(state, &repo);
120 let token = lfs_token::issue(&state.key, &repo.id, token_op, now_unix() + TOKEN_TTL_SECS);
121
122 let mut objects = Vec::with_capacity(request.objects.len());
123 for object in &request.objects {
124 let stored = match state.store.lfs_object_size(&repo.id, &object.oid).await {
125 Ok(stored) => stored,
126 Err(e) => return Ok(server_error(e)),
127 };
128 let href = format!("{base}/{}", object.oid);
129 let entry = match (writing, stored) {
130 (true, Some(_)) => json!({ "oid": object.oid, "size": object.size }),
131 (true, None) => json!({
132 "oid": object.oid,
133 "size": object.size,
134 "actions": {
135 "upload": action_json(&href, &token),
136 "verify": action_json(&format!("{base}/verify"), &token),
137 },
138 }),
139 (false, Some(size)) => json!({
140 "oid": object.oid,
141 "size": size,
142 "actions": { "download": action_json(&href, &token) },
143 }),
144 (false, None) => json!({
145 "oid": object.oid,
146 "size": object.size,
147 "error": { "code": 404, "message": "Object does not exist" },
148 }),
149 };
150 objects.push(entry);
151 }
152 Ok(lfs_json(
153 StatusCode::OK,
154 json!({ "transfer": "basic", "objects": objects }),
155 ))
156}
157
158/// A batch request that passed its own checks, with the action it asks for.
159struct BatchPlan {
160 request: BatchRequest,
161 action: Action,
162}
163
164/// Parses and checks a batch body. The error is the response to send.
165fn parse_batch_operation(bytes: &[u8]) -> Result<BatchPlan, Box<Response>> {
166 let request: BatchRequest = serde_json::from_slice(bytes).map_err(|e| {
167 Box::new(message(
168 StatusCode::UNPROCESSABLE_ENTITY,
169 format!("invalid batch request: {e}"),
170 ))
171 })?;
172 let action = match request.operation.as_str() {
173 "download" => Action::Read,
174 "upload" => Action::Write,
175 _ => return Err(unprocessable("unsupported operation")),
176 };
177 if !request.transfers.is_empty() && !request.transfers.iter().any(|t| t == "basic") {
178 return Err(unprocessable("only the basic transfer is supported"));
179 }
180 if request
181 .hash_algo
182 .as_deref()
183 .is_some_and(|algo| algo != "sha256")
184 {
185 return Err(unprocessable("unsupported hash algorithm"));
186 }
187 if request.objects.len() > MAX_BATCH_OBJECTS {
188 return Err(unprocessable(format!(
189 "a batch may name at most {MAX_BATCH_OBJECTS} objects"
190 )));
191 }
192 Ok(BatchPlan { request, action })
193}
194
195#[route(GET "/{owner}/{repo}/info/lfs/objects/{oid}")]
196async fn get_object(cx: &Cx) -> topcoat::Result<Response> {
197 let state = app_context::<AppState>(cx);
198 let headers = headers(cx);
199 let owner = path_param::<Owner>(cx);
200 let name = path_param::<Repo>(cx);
201 let oid = path_param::<Oid>(cx);
202 if !is_valid_oid(oid) {
203 return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
204 }
205
206 let repo = match authorize(state, client_ip(cx), headers, owner, name, Action::Read).await {
207 Ok(repo) => repo,
208 Err(response) => return Ok(*response),
209 };
210 let size = match state.store.lfs_object_size(&repo.id, oid).await {
211 Ok(Some(size)) => size,
212 Ok(None) => return Ok(message(StatusCode::NOT_FOUND, "Object does not exist")),
213 Err(e) => return Ok(server_error(e)),
214 };
215 let file = match tokio::fs::File::open(object_path(&state.config.data_dir, oid)).await {
216 Ok(file) => file,
217 Err(e) => {
218 // The link exists but the bytes do not: the store is damaged, not the request.
219 tracing::error!(oid = %oid, error = %e, "lfs object file is missing");
220 return Ok(server_error(e));
221 }
222 };
223
224 let chunks = ReaderStream::new(file).map_ok(Frame::data);
225 let mut res = Response::new(Body::new(StreamBody::new(chunks)));
226 res.headers_mut().insert(
227 CONTENT_TYPE,
228 HeaderValue::from_static("application/octet-stream"),
229 );
230 res.headers_mut()
231 .insert(CONTENT_LENGTH, HeaderValue::from(size));
232 Ok(res)
233}
234
235#[route(PUT "/{owner}/{repo}/info/lfs/objects/{oid}")]
236async fn put_object(cx: &Cx, body: Body) -> topcoat::Result<Response> {
237 let state = app_context::<AppState>(cx);
238 let headers = headers(cx);
239 let owner = path_param::<Owner>(cx);
240 let name = path_param::<Repo>(cx);
241 let oid = path_param::<Oid>(cx);
242 if !is_valid_oid(oid) {
243 return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
244 }
245
246 let repo = match authorize(state, client_ip(cx), headers, owner, name, Action::Write).await {
247 Ok(repo) => repo,
248 Err(response) => return Ok(*response),
249 };
250 if let Err(response) = ensure_writable(state, &repo).await {
251 return Ok(*response);
252 }
253
254 let max = state.config.lfs.max_object_bytes();
255 let declared = headers
256 .get(CONTENT_LENGTH)
257 .and_then(|v| v.to_str().ok())
258 .and_then(|v| v.parse::<u64>().ok());
259 if declared.is_some_and(|len| len > max) {
260 return Ok(message(
261 StatusCode::PAYLOAD_TOO_LARGE,
262 "the object exceeds the size limit",
263 ));
264 }
265
266 let size = match receive_object(&state.config.data_dir, body, oid, max).await {
267 Ok(size) => size,
268 Err(response) => return Ok(*response),
269 };
270 if let Err(e) = state.store.insert_lfs_object(&repo.id, oid, size).await {
271 return Ok(server_error(e));
272 }
273 Ok(Response::new(Body::empty()))
274}
275
276#[route(POST "/{owner}/{repo}/info/lfs/objects/verify")]
277async fn verify(cx: &Cx, body: Body) -> topcoat::Result<Response> {
278 let state = app_context::<AppState>(cx);
279 let headers = headers(cx);
280 let owner = path_param::<Owner>(cx);
281 let name = path_param::<Repo>(cx);
282
283 let repo = match authorize(state, client_ip(cx), headers, owner, name, Action::Write).await {
284 Ok(repo) => repo,
285 Err(response) => return Ok(*response),
286 };
287 if let Err(response) = ensure_writable(state, &repo).await {
288 return Ok(*response);
289 }
290
291 let bytes = to_bytes(body, MAX_VERIFY_BYTES).await?;
292 let request: VerifyRequest = match serde_json::from_slice(&bytes) {
293 Ok(request) => request,
294 Err(e) => {
295 return Ok(message(
296 StatusCode::UNPROCESSABLE_ENTITY,
297 format!("invalid verify request: {e}"),
298 ));
299 }
300 };
301 if !is_valid_oid(&request.oid) {
302 return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
303 }
304 Ok(
305 match state.store.lfs_object_size(&repo.id, &request.oid).await {
306 Ok(Some(size)) if size == request.size => Response::new(Body::empty()),
307 Ok(Some(_)) => message(StatusCode::UNPROCESSABLE_ENTITY, "the size does not match"),
308 Ok(None) => message(StatusCode::NOT_FOUND, "Object does not exist"),
309 Err(e) => server_error(e),
310 },
311 )
312}
313
314/// Who may act on a repo for `action`. A `RemoteAuth` token is checked against the repo and the
315/// operation it grants, and skips the user check; otherwise HTTP Basic (a PAT) or anonymous access
316/// goes through the usual repo authorization. The error is the response to send.
317async fn authorize(
318 state: &AppState,
319 ip: IpAddr,
320 headers: &HeaderMap,
321 owner: &str,
322 name: &str,
323 action: Action,
324) -> Result<forge_core::Repo, Box<Response>> {
325 if let Some(token) = remote_auth_token(headers) {
326 let repo = match find_repo(state, owner, name).await {
327 Ok(Some(repo)) => repo,
328 Ok(None) => return Err(Box::new(access_failure(AccessError::NotFound))),
329 Err(e) => return Err(Box::new(access_failure(e))),
330 };
331 let wanted = match action {
332 Action::Write => LfsOp::Upload,
333 _ => LfsOp::Download,
334 };
335 if !lfs_token::verify(&state.key, token, &repo.id, wanted, now_unix()) {
336 return Err(Box::new(access_failure(AccessError::Unauthorized)));
337 }
338 return Ok(repo);
339 }
340
341 let actor = authenticate(state, ip, headers).await?;
342 resolve_repo_for(state, actor.as_ref(), owner, name, action)
343 .await
344 .map_err(|e| Box::new(access_failure(e)))
345}
346
347/// The token of a `Authorization: RemoteAuth <token>` header, if there is one.
348fn remote_auth_token(headers: &HeaderMap) -> Option<&str> {
349 let value = headers.get(AUTHORIZATION)?.to_str().ok()?;
350 let (scheme, token) = value.split_once(' ')?;
351 let token = token.trim();
352 (scheme.eq_ignore_ascii_case("remoteauth") && !token.is_empty()).then_some(token)
353}
354
355/// Refuses writes to an archived repo or to a pull mirror, which is read-only by design.
356async fn ensure_writable(state: &AppState, repo: &forge_core::Repo) -> Result<(), Box<Response>> {
357 if repo.archived {
358 return Err(Box::new(message(
359 StatusCode::FORBIDDEN,
360 "the repository is archived",
361 )));
362 }
363 match state.store.pull_mirror(&repo.id).await {
364 Ok(Some(_)) => Err(Box::new(message(
365 StatusCode::FORBIDDEN,
366 "the repository is a pull mirror",
367 ))),
368 Ok(None) => Ok(()),
369 Err(e) => Err(Box::new(server_error(e))),
370 }
371}
372
373/// Streams an upload into a temp file, checks its SHA-256 against `oid` and its size against
374/// `max`, then renames it into the object store. Returns the object size. A failed upload leaves
375/// no file behind.
376async fn receive_object(
377 data_dir: &Path,
378 body: Body,
379 oid: &str,
380 max: u64,
381) -> Result<u64, Box<Response>> {
382 let dir = tmp_dir(data_dir);
383 tokio::fs::create_dir_all(&dir)
384 .await
385 .map_err(|e| Box::new(server_error(e)))?;
386 let seq = TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
387 let temp = dir.join(format!("{oid}.{}.{seq}.part", std::process::id()));
388
389 let (size, digest) = match write_and_hash(&temp, body, max).await {
390 Ok(written) => written,
391 Err(response) => {
392 let _ = tokio::fs::remove_file(&temp).await;
393 return Err(response);
394 }
395 };
396 if digest != oid {
397 let _ = tokio::fs::remove_file(&temp).await;
398 return Err(unprocessable("the content does not match the oid"));
399 }
400
401 let target = object_path(data_dir, oid);
402 if let Some(parent) = target.parent() {
403 tokio::fs::create_dir_all(parent)
404 .await
405 .map_err(|e| Box::new(server_error(e)))?;
406 }
407 if let Err(e) = tokio::fs::rename(&temp, &target).await {
408 let _ = tokio::fs::remove_file(&temp).await;
409 return Err(Box::new(server_error(e)));
410 }
411 Ok(size)
412}
413
414/// Writes the request body to `temp` and returns its size and hex SHA-256.
415async fn write_and_hash(temp: &Path, body: Body, max: u64) -> Result<(u64, String), Box<Response>> {
416 let mut file = tokio::fs::OpenOptions::new()
417 .write(true)
418 .create_new(true)
419 .open(temp)
420 .await
421 .map_err(|e| Box::new(server_error(e)))?;
422 let mut hasher = Sha256::new();
423 let mut size: u64 = 0;
424 let mut stream = std::pin::pin!(body.into_data_stream());
425 while let Some(chunk) = stream.next().await {
426 let Ok(chunk) = chunk else {
427 return Err(Box::new(message(
428 StatusCode::BAD_REQUEST,
429 "the request body could not be read",
430 )));
431 };
432 size = size.saturating_add(u64::try_from(chunk.len()).unwrap_or(u64::MAX));
433 if size > max {
434 return Err(Box::new(message(
435 StatusCode::PAYLOAD_TOO_LARGE,
436 "the object exceeds the size limit",
437 )));
438 }
439 hasher.update(&chunk);
440 file.write_all(&chunk)
441 .await
442 .map_err(|e| Box::new(server_error(e)))?;
443 }
444 file.sync_all()
445 .await
446 .map_err(|e| Box::new(server_error(e)))?;
447 Ok((size, hex::encode(hasher.finalize())))
448}
449
450/// The `href` prefix of LFS actions for `repo`, always in the `.git` form.
451fn batch_base(state: &AppState, repo: &forge_core::Repo) -> String {
452 format!(
453 "{}/{}/{}.git/info/lfs/objects",
454 state.config.base_url.trim_end_matches('/'),
455 repo.owner_name,
456 repo.name
457 )
458}
459
460/// One action of a batch response.
461fn action_json(href: &str, token: &str) -> Value {
462 json!({
463 "href": href,
464 "header": { "Authorization": format!("RemoteAuth {token}") },
465 "expires_in": TOKEN_TTL_SECS,
466 })
467}
468
469/// Maps an access failure to a response. A 401 carries the LFS challenge, so git-lfs asks for
470/// credentials.
471fn access_failure(error: AccessError) -> Response {
472 let unauthorized = error == AccessError::Unauthorized;
473 let mut res = access_error(error);
474 if unauthorized {
475 res.headers_mut().insert(
476 "LFS-Authenticate",
477 HeaderValue::from_static("Basic realm=\"forge\""),
478 );
479 }
480 res
481}
482
483/// A 422 response, boxed so it can be the error of a `Result`.
484fn unprocessable(text: impl Into<String>) -> Box<Response> {
485 Box::new(message(StatusCode::UNPROCESSABLE_ENTITY, text))
486}
487
488/// Logs an internal failure and answers 500.
489fn server_error(error: impl std::fmt::Display) -> Response {
490 tracing::error!(error = %error, "lfs request failed");
491 message(StatusCode::INTERNAL_SERVER_ERROR, "internal server error")
492}
493
494/// A JSON answer in the LFS media type.
495fn lfs_json(status: StatusCode, value: Value) -> Response {
496 let mut res = Response::new(Body::from(value.to_string()));
497 *res.status_mut() = status;
498 res.headers_mut()
499 .insert(CONTENT_TYPE, HeaderValue::from_static(LFS_JSON));
500 res
501}
502
503/// An LFS JSON error with a `message` field.
504fn message(status: StatusCode, text: impl Into<String>) -> Response {
505 lfs_json(status, json!({ "message": text.into() }))
506}
507
508fn now_unix() -> i64 {
509 SystemTime::now()
510 .duration_since(UNIX_EPOCH)
511 .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
512}