fabrica

hanna/fabrica

main
Switch branch
mainView all branches
Clone
HTTPS
SSH
16604 bytes
Raw
1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! Git-LFS: the [Batch API] and basic transfer (upload / download) over HTTP.
6//!
7//! Objects are content-addressed by their sha256 `oid` and stored on disk under
8//! `storage.lfs_dir`, sharded `{oid[0..2]}/{oid[2..4]}/{oid}`; the `lfs_objects`
9//! table records which repository holds which object (for access control and the
10//! batch "do you have this?" answer). Uploads stream to a temp file, are verified
11//! against the claimed oid, and only then moved into place — a corrupt or
12//! mislabelled upload never becomes a servable object.
13//!
14//! Authorization runs before any bytes move: download needs [`Access::Read`],
15//! upload needs [`Access::Write`]. HTTPS clients authenticate with Basic
16//! credentials; SSH clients present the bearer token minted by
17//! `git-lfs-authenticate` (see [`crate::git_http::authenticate`]).
18//!
19//! [Batch API]: https://github.com/git-lfs/git-lfs/blob/main/docs/api/batch.md
20
21use std::collections::HashMap;
22use std::sync::atomic::{AtomicU64, Ordering};
23
24use axum::body::Body;
25use axum::extract::{Path, State};
26use axum::http::{HeaderMap, StatusCode, header};
27use axum::response::{IntoResponse, Response};
28use axum_extra::extract::cookie::CookieJar;
29use blob::Namespace;
30use futures_util::TryStreamExt as _;
31use serde::{Deserialize, Serialize};
32use sha2::{Digest, Sha256};
33use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
34use tokio_util::io::{ReaderStream, StreamReader};
35
36use crate::AppState;
37use crate::git_http::authorize;
38
39/// The media type every LFS JSON request and response uses.
40const LFS_JSON: &str = "application/vnd.git-lfs+json";
41
42/// Per-process counter for unique upload temp-file names.
43static TMP_SEQ: AtomicU64 = AtomicU64::new(0);
44
45/// Split a repo-relative path on `/info/lfs/`, returning `(repo_path, lfs_tail)`
46/// with any trailing `.git` stripped from the repo path. `lfs_tail` is e.g.
47/// `objects/batch` or `objects/<oid>`.
48#[must_use]
49pub(crate) fn split_lfs(rest: &str) -> Option<(&str, &str)> {
50 let idx = rest.find("/info/lfs/")?;
51 let repo = rest[..idx].strip_suffix(".git").unwrap_or(&rest[..idx]);
52 Some((repo, &rest[idx + "/info/lfs/".len()..]))
53}
54
55/// A 64-char lowercase-hex sha256 is the only shape we treat as an oid; anything
56/// else is rejected before it can touch the filesystem.
57fn valid_oid(oid: &str) -> bool {
58 oid.len() == 64
59 && oid
60 .bytes()
61 .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
62}
63
64/// The blob key for an object, sharded by the first two byte-pairs of the oid.
65/// Callers pass only validated 64-char oids, so the slices are in bounds.
66fn object_key(oid: &str) -> String {
67 format!("{}/{}/{}", &oid[0..2], &oid[2..4], oid)
68}
69
70/// Hex-encode a byte slice (lowercase), for comparing a computed digest to an oid.
71fn hex(bytes: &[u8]) -> String {
72 let mut s = String::with_capacity(bytes.len() * 2);
73 for b in bytes {
74 s.push(char::from_digit(u32::from(b >> 4), 16).unwrap_or('0'));
75 s.push(char::from_digit(u32::from(b & 0xf), 16).unwrap_or('0'));
76 }
77 s
78}
79
80/// `PUT /{owner}/{*rest}` — the object-upload route. Recognizes only the LFS
81/// `…/info/lfs/objects/{oid}` shape and rejects everything else.
82pub(crate) async fn put(
83 State(state): State<AppState>,
84 jar: CookieJar,
85 headers: HeaderMap,
86 Path((owner, rest)): Path<(String, String)>,
87 body: Body,
88) -> Response {
89 let Some((repo_path, tail)) = split_lfs(&rest) else {
90 return StatusCode::NOT_FOUND.into_response();
91 };
92 let Some(oid) = tail.strip_prefix("objects/") else {
93 return StatusCode::NOT_FOUND.into_response();
94 };
95 if oid.contains('/') {
96 return StatusCode::NOT_FOUND.into_response();
97 }
98 upload(
99 state,
100 jar,
101 headers,
102 owner,
103 repo_path.to_string(),
104 oid.to_string(),
105 body,
106 )
107 .await
108}
109
110// ---- Batch API ----
111
112/// One `{oid, size}` pair in a batch request.
113#[derive(Debug, Deserialize)]
114struct ObjectId {
115 oid: String,
116 #[serde(default)]
117 size: i64,
118}
119
120/// A `POST objects/batch` request body.
121#[derive(Debug, Deserialize)]
122struct BatchRequest {
123 operation: String,
124 objects: Vec<ObjectId>,
125}
126
127/// A single transfer action (an href the client PUTs to or GETs from).
128#[derive(Debug, Serialize)]
129struct Action {
130 href: String,
131 #[serde(skip_serializing_if = "Option::is_none")]
132 header: Option<HashMap<String, String>>,
133}
134
135/// The actions available for one object (only the relevant ones are set).
136#[derive(Debug, Default, Serialize)]
137struct Actions {
138 #[serde(skip_serializing_if = "Option::is_none")]
139 download: Option<Action>,
140 #[serde(skip_serializing_if = "Option::is_none")]
141 upload: Option<Action>,
142 #[serde(skip_serializing_if = "Option::is_none")]
143 verify: Option<Action>,
144}
145
146/// A per-object error in the batch response.
147#[derive(Debug, Serialize)]
148struct ObjectError {
149 code: u16,
150 message: String,
151}
152
153/// One object in the batch response.
154#[derive(Debug, Serialize)]
155struct BatchObject {
156 oid: String,
157 size: i64,
158 #[serde(skip_serializing_if = "Option::is_none")]
159 actions: Option<Actions>,
160 #[serde(skip_serializing_if = "Option::is_none")]
161 error: Option<ObjectError>,
162}
163
164/// The `POST objects/batch` response body.
165#[derive(Debug, Serialize)]
166struct BatchResponse {
167 transfer: String,
168 objects: Vec<BatchObject>,
169}
170
171/// `POST .../info/lfs/objects/batch` — negotiate which objects the client may
172/// upload or download and where.
173pub(crate) async fn batch(
174 state: AppState,
175 jar: CookieJar,
176 headers: HeaderMap,
177 owner: String,
178 repo_path: String,
179 body: Body,
180) -> Response {
181 let Ok(raw) = axum::body::to_bytes(body, 1024 * 1024).await else {
182 return lfs_error(StatusCode::BAD_REQUEST, "request too large");
183 };
184 let Ok(req) = serde_json::from_slice::<BatchRequest>(&raw) else {
185 return lfs_error(StatusCode::BAD_REQUEST, "malformed batch request");
186 };
187 let uploading = req.operation == "upload";
188 let required = if uploading {
189 auth::Access::Write
190 } else {
191 auth::Access::Read
192 };
193 let repo = match authorize(&state, &jar, &headers, &owner, &repo_path, required).await {
194 Ok(repo) => repo,
195 Err(resp) => return resp,
196 };
197
198 let base = format!(
199 "{}/{}/{}.git/info/lfs/objects",
200 state.config.instance.url.trim_end_matches('/'),
201 owner,
202 repo_path
203 );
204 let mut objects = Vec::with_capacity(req.objects.len());
205 for obj in &req.objects {
206 objects.push(if uploading {
207 batch_upload(&state, &repo.id, &base, obj).await
208 } else {
209 batch_download(&state, &repo.id, &base, obj).await
210 });
211 }
212 let payload = BatchResponse {
213 transfer: "basic".to_string(),
214 objects,
215 };
216 lfs_json(StatusCode::OK, &payload)
217}
218
219/// Resolve one object for a download batch: offer a download action if we hold it.
220async fn batch_download(
221 state: &AppState,
222 repo_id: &str,
223 base: &str,
224 obj: &ObjectId,
225) -> BatchObject {
226 if valid_oid(&obj.oid)
227 && state
228 .store
229 .lfs_object_exists(repo_id, &obj.oid)
230 .await
231 .unwrap_or(false)
232 {
233 BatchObject {
234 oid: obj.oid.clone(),
235 size: obj.size,
236 actions: Some(Actions {
237 download: Some(Action {
238 href: format!("{base}/{}", obj.oid),
239 header: None,
240 }),
241 ..Actions::default()
242 }),
243 error: None,
244 }
245 } else {
246 BatchObject {
247 oid: obj.oid.clone(),
248 size: obj.size,
249 actions: None,
250 error: Some(ObjectError {
251 code: 404,
252 message: "object does not exist".to_string(),
253 }),
254 }
255 }
256}
257
258/// Resolve one object for an upload batch: offer upload+verify unless we have it.
259async fn batch_upload(state: &AppState, repo_id: &str, base: &str, obj: &ObjectId) -> BatchObject {
260 if !valid_oid(&obj.oid) {
261 return BatchObject {
262 oid: obj.oid.clone(),
263 size: obj.size,
264 actions: None,
265 error: Some(ObjectError {
266 code: 422,
267 message: "invalid oid".to_string(),
268 }),
269 };
270 }
271 // Already stored (in this repo): no action needed, the client skips it.
272 if state
273 .store
274 .lfs_object_exists(repo_id, &obj.oid)
275 .await
276 .unwrap_or(false)
277 {
278 return BatchObject {
279 oid: obj.oid.clone(),
280 size: obj.size,
281 actions: None,
282 error: None,
283 };
284 }
285 BatchObject {
286 oid: obj.oid.clone(),
287 size: obj.size,
288 actions: Some(Actions {
289 upload: Some(Action {
290 href: format!("{base}/{}", obj.oid),
291 header: None,
292 }),
293 verify: Some(Action {
294 href: format!("{base}/{}/verify", obj.oid),
295 header: None,
296 }),
297 ..Actions::default()
298 }),
299 error: None,
300 }
301}
302
303// ---- Basic transfer ----
304
305/// `GET .../info/lfs/objects/{oid}` — stream a stored object.
306pub(crate) async fn download(
307 state: AppState,
308 jar: CookieJar,
309 headers: HeaderMap,
310 owner: String,
311 repo_path: String,
312 oid: String,
313) -> Response {
314 let repo = match authorize(
315 &state,
316 &jar,
317 &headers,
318 &owner,
319 &repo_path,
320 auth::Access::Read,
321 )
322 .await
323 {
324 Ok(repo) => repo,
325 Err(resp) => return resp,
326 };
327 if !valid_oid(&oid)
328 || !state
329 .store
330 .lfs_object_exists(&repo.id, &oid)
331 .await
332 .unwrap_or(false)
333 {
334 return StatusCode::NOT_FOUND.into_response();
335 }
336 let Ok(Some(read)) = state.blobs.open(Namespace::Lfs, &object_key(&oid)).await else {
337 return StatusCode::NOT_FOUND.into_response();
338 };
339 let body = Body::from_stream(ReaderStream::new(read.reader));
340 (
341 [
342 (header::CONTENT_TYPE, "application/octet-stream".to_string()),
343 (header::CONTENT_LENGTH, read.size.to_string()),
344 ],
345 body,
346 )
347 .into_response()
348}
349
350/// `PUT .../info/lfs/objects/{oid}` — receive an object, verifying its sha256
351/// matches the oid before it becomes servable.
352pub(crate) async fn upload(
353 state: AppState,
354 jar: CookieJar,
355 headers: HeaderMap,
356 owner: String,
357 repo_path: String,
358 oid: String,
359 body: Body,
360) -> Response {
361 let repo = match authorize(
362 &state,
363 &jar,
364 &headers,
365 &owner,
366 &repo_path,
367 auth::Access::Write,
368 )
369 .await
370 {
371 Ok(repo) => repo,
372 Err(resp) => return resp,
373 };
374 if !valid_oid(&oid) {
375 return lfs_error(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid");
376 }
377 // Idempotent: if we already hold it, drain the body and succeed.
378 if state
379 .store
380 .lfs_object_exists(&repo.id, &oid)
381 .await
382 .unwrap_or(false)
383 {
384 return StatusCode::OK.into_response();
385 }
386
387 // Stream to a local scratch file to hash and size it before it becomes
388 // servable; `data_dir` is always local, whatever the blob backend is.
389 let tmp_dir = state.config.storage.data_dir.join("tmp");
390 if tokio::fs::create_dir_all(&tmp_dir).await.is_err() {
391 return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "storage error");
392 }
393 let tmp = tmp_dir.join(format!(
394 "lfs-{}.{}.tmp",
395 std::process::id(),
396 TMP_SEQ.fetch_add(1, Ordering::Relaxed)
397 ));
398
399 let Ok((digest, size)) = stream_to_file(body, &tmp).await else {
400 let _ = tokio::fs::remove_file(&tmp).await;
401 return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "write failed");
402 };
403 if digest != oid {
404 let _ = tokio::fs::remove_file(&tmp).await;
405 return lfs_error(StatusCode::UNPROCESSABLE_ENTITY, "oid mismatch");
406 }
407
408 // Move the verified object into the blob store (local rename or S3 upload).
409 if state
410 .blobs
411 .put_file(
412 Namespace::Lfs,
413 &object_key(&oid),
414 &tmp,
415 "application/octet-stream",
416 )
417 .await
418 .is_err()
419 {
420 let _ = tokio::fs::remove_file(&tmp).await;
421 return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "storage error");
422 }
423 if state
424 .store
425 .lfs_object_add(&repo.id, &oid, size)
426 .await
427 .is_err()
428 {
429 return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "storage error");
430 }
431 StatusCode::OK.into_response()
432}
433
434/// `POST .../info/lfs/objects/{oid}/verify` — confirm an object landed at the
435/// expected size (git-lfs calls this after an upload).
436pub(crate) async fn verify(
437 state: AppState,
438 jar: CookieJar,
439 headers: HeaderMap,
440 owner: String,
441 repo_path: String,
442 body: Body,
443) -> Response {
444 let repo = match authorize(
445 &state,
446 &jar,
447 &headers,
448 &owner,
449 &repo_path,
450 auth::Access::Read,
451 )
452 .await
453 {
454 Ok(repo) => repo,
455 Err(resp) => return resp,
456 };
457 let Ok(raw) = axum::body::to_bytes(body, 64 * 1024).await else {
458 return lfs_error(StatusCode::BAD_REQUEST, "request too large");
459 };
460 let Ok(obj) = serde_json::from_slice::<ObjectId>(&raw) else {
461 return lfs_error(StatusCode::BAD_REQUEST, "malformed request");
462 };
463 match state.store.lfs_object_size(&repo.id, &obj.oid).await {
464 Ok(Some(size)) if size == obj.size => StatusCode::OK.into_response(),
465 _ => lfs_error(StatusCode::NOT_FOUND, "object not found"),
466 }
467}
468
469/// Stream `body` into `path`, returning `(hex_sha256, size)`.
470async fn stream_to_file(body: Body, path: &std::path::Path) -> std::io::Result<(String, i64)> {
471 let stream = body.into_data_stream().map_err(std::io::Error::other);
472 let reader = StreamReader::new(stream);
473 tokio::pin!(reader);
474 let mut file = tokio::fs::File::create(path).await?;
475 let mut hasher = Sha256::new();
476 let mut buf = vec![0u8; 64 * 1024];
477 let mut size: i64 = 0;
478 loop {
479 let n = reader.read(&mut buf).await?;
480 if n == 0 {
481 break;
482 }
483 hasher.update(&buf[..n]);
484 file.write_all(&buf[..n]).await?;
485 size += i64::try_from(n).unwrap_or(0);
486 }
487 file.flush().await?;
488 Ok((hex(&hasher.finalize()), size))
489}
490
491/// Render a value as an LFS JSON response with the required content type.
492fn lfs_json<T: Serialize>(status: StatusCode, value: &T) -> Response {
493 match serde_json::to_vec(value) {
494 Ok(bytes) => (status, [(header::CONTENT_TYPE, LFS_JSON)], bytes).into_response(),
495 Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
496 }
497}
498
499/// Render an LFS error body (`{"message": "..."}`) with the given status.
500fn lfs_error(status: StatusCode, message: &str) -> Response {
501 let body = serde_json::json!({ "message": message });
502 (status, [(header::CONTENT_TYPE, LFS_JSON)], body.to_string()).into_response()
503}
504
505#[cfg(test)]
506mod tests {
507 use super::*;
508
509 #[test]
510 fn split_lfs_finds_the_repo_and_tail() {
511 assert_eq!(
512 split_lfs("owner-scope/repo.git/info/lfs/objects/batch"),
513 Some(("owner-scope/repo", "objects/batch"))
514 );
515 // The `.git` is optional (git-lfs derives the endpoint from the remote).
516 assert_eq!(
517 split_lfs("grp/sub/repo/info/lfs/objects/abc"),
518 Some(("grp/sub/repo", "objects/abc"))
519 );
520 assert_eq!(split_lfs("owner/repo.git/info/refs"), None);
521 }
522
523 #[test]
524 fn valid_oid_accepts_only_lowercase_hex_sha256() {
525 assert!(valid_oid(&"a".repeat(64)));
526 assert!(valid_oid(&"0123456789abcdef".repeat(4)));
527 assert!(!valid_oid(&"a".repeat(63)), "too short");
528 assert!(!valid_oid(&"A".repeat(64)), "uppercase rejected");
529 assert!(!valid_oid(&"g".repeat(64)), "non-hex rejected");
530 }
531
532 #[test]
533 fn hex_matches_a_known_sha256() {
534 // sha256("") = e3b0c442...
535 let digest = Sha256::digest(b"");
536 assert_eq!(
537 hex(&digest),
538 "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
539 );
540 }
541}