Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/repos/src/pack_limits.rs

598 lines23,605 bytesCodeBlame
1//! What the git store will not hold, checked before it is asked to: no
2//! file over 32 MB, and no repository over about 1 GB
3//! (docs/ARTIFACTS.md, "Limits"). A push that would cross either is
4//! declined with a reason git prints, instead of failing inside the store.
5//!
6//! [`PackSizer`] walks a receive-pack body as it arrives, a chunk at a
7//! time, without keeping it: the commands, then each object of the pack,
8//! inflated into a small window and thrown away, only to learn its size
9//! and where the next one starts. A delta's size is the size of the object
10//! it makes, read from the start of the delta. Memory stays at a few tens
11//! of kilobytes however large the push.
12//!
13//! [`Sideband`] does the same for the other direction: it takes an
14//! upload-pack answer that used side-band framing a chunk at a time and
15//! gives back the pack's bytes, so a pack can go from one repository to
16//! another without being held whole (land.rs).
17
18use miniz_oxide::inflate::TINFLStatus;
19use miniz_oxide::inflate::core::{DecompressorOxide, decompress, inflate_flags};
20
21/// The largest file or blob the git store holds. Cloudflare documents
22/// "32 MB"; decimal megabytes are assumed, so nothing it would refuse gets
23/// through.
24pub const MAX_OBJECT_BYTES: u64 = 32_000_000;
25/// The largest repository the git store holds is 1 GB. Pushes stop a little
26/// before it: what g1t counts (`stored_bytes`) is a lower bound.
27pub const DEFAULT_REPO_LIMIT_BYTES: u64 = 950_000_000;
28/// The largest request body Cloudflare passes on for g1t.sh's plan. A
29/// larger push is refused by the network with HTTP 413 before g1t sees it.
30pub const PLATFORM_BODY_LIMIT_BYTES: u64 = 100_000_000;
31
32/// Inflate's window must be a power of two of at least 32 KiB.
33const WINDOW: usize = 32 * 1024;
34/// A delta starts with two sizes, each at most ten bytes.
35const DELTA_HEAD: usize = 20;
36
37/// Why a push cannot be stored.
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub enum Violation {
40 /// An object larger than the store holds.
41 ObjectTooLarge { size: u64 },
42 /// The body is not a receive-pack request g1t can read.
43 Malformed(String),
44}
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47enum Phase {
48 /// pkt-line commands, until a flush packet.
49 Commands,
50 /// The rest of one command's payload.
51 SkipLine(usize),
52 /// After the commands: `PACK`, or push options before it.
53 PackHeader,
54 /// The pack's version and object count.
55 PackCount,
56 /// An object's type and size.
57 EntryHeader,
58 /// The offset that names an ofs-delta's base.
59 OffsetBase,
60 /// The id that names a ref-delta's base.
61 RefBase(usize),
62 /// The object's deflated data.
63 Inflate,
64 /// The pack's checksum, and anything after it.
65 Trailer,
66}
67
68/// Walks a receive-pack body a chunk at a time; see the module docs.
69pub struct PackSizer {
70 max_object: u64,
71 phase: Phase,
72 /// Header bytes not yet complete.
73 pending: Vec<u8>,
74 objects_left: u32,
75 /// The object being inflated: whether it is a delta, and its size.
76 delta: bool,
77 size: u64,
78 inflater: Box<DecompressorOxide>,
79 window: Vec<u8>,
80 window_at: usize,
81 /// The first bytes a delta inflated to.
82 head: Vec<u8>,
83 /// Objects read in full.
84 pub objects: u32,
85 /// The largest object seen, inflated.
86 pub largest: u64,
87 /// Bytes fed so far.
88 pub fed: u64,
89 /// Where the pack began in the body, once it has.
90 pack_start: Option<u64>,
91}
92
93impl PackSizer {
94 pub fn new(max_object: u64) -> Self {
95 PackSizer {
96 max_object,
97 phase: Phase::Commands,
98 pending: Vec::with_capacity(32),
99 objects_left: 0,
100 delta: false,
101 size: 0,
102 inflater: Box::default(),
103 window: Vec::new(),
104 window_at: 0,
105 head: Vec::with_capacity(DELTA_HEAD),
106 objects: 0,
107 largest: 0,
108 fed: 0,
109 pack_start: None,
110 }
111 }
112
113 /// Bytes of the pack itself, after the commands, so far.
114 pub fn pack_bytes(&self) -> u64 {
115 self.pack_start.map_or(0, |start| self.fed - start)
116 }
117
118 /// Whether every object the pack said it holds has been read, or the
119 /// push carries no pack (it only deletes).
120 #[cfg(test)]
121 pub fn complete(&self) -> bool {
122 match self.phase {
123 Phase::Trailer => true,
124 Phase::Commands | Phase::PackHeader => self.pack_start.is_none() && self.pending.is_empty(),
125 _ => false,
126 }
127 }
128
129 /// Reads the next chunk of the body.
130 pub fn feed(&mut self, mut input: &[u8]) -> Result<(), Violation> {
131 let base = self.fed;
132 let length = input.len() as u64;
133 self.fed += length;
134 while !input.is_empty() {
135 match self.phase {
136 Phase::Commands => {
137 let taken = self.take(&mut input, 4);
138 if !taken {
139 return Ok(());
140 }
141 let length = std::str::from_utf8(&self.pending)
142 .ok()
143 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
144 .ok_or_else(|| Violation::Malformed("a command is not a pkt-line".into()))?;
145 self.pending.clear();
146 match length {
147 0 => self.phase = Phase::PackHeader,
148 1..=3 => {}
149 _ => self.phase = Phase::SkipLine(length - 4),
150 }
151 }
152 Phase::SkipLine(left) => {
153 let skipped = left.min(input.len());
154 input = &input[skipped..];
155 self.phase = if skipped == left { Phase::Commands } else { Phase::SkipLine(left - skipped) };
156 }
157 Phase::PackHeader => {
158 if !self.take(&mut input, 4) {
159 return Ok(());
160 }
161 if self.pending == b"PACK" {
162 self.pack_start = Some(base + length - input.len() as u64 - 4);
163 self.phase = Phase::PackCount;
164 continue;
165 }
166 // Push options come as pkt-lines between the commands
167 // and the pack, ended by another flush.
168 let line = std::str::from_utf8(&self.pending)
169 .ok()
170 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
171 .ok_or_else(|| Violation::Malformed("the push does not carry a pack".into()))?;
172 self.pending.clear();
173 if line >= 4 {
174 self.phase = Phase::SkipLine(line - 4);
175 }
176 }
177 Phase::PackCount => {
178 if !self.take(&mut input, 12) {
179 return Ok(());
180 }
181 let p = &self.pending;
182 self.objects_left = u32::from_be_bytes([p[8], p[9], p[10], p[11]]);
183 self.pending.clear();
184 self.phase = if self.objects_left == 0 { Phase::Trailer } else { Phase::EntryHeader };
185 }
186 Phase::EntryHeader => {
187 let byte = input[0];
188 input = &input[1..];
189 self.pending.push(byte);
190 if byte & 0x80 != 0 {
191 if self.pending.len() > 10 {
192 return Err(Violation::Malformed("an object's size is too long".into()));
193 }
194 continue;
195 }
196 let first = self.pending[0];
197 let code = (first >> 4) & 7;
198 let mut size = u64::from(first & 15);
199 for (n, byte) in self.pending[1..].iter().enumerate() {
200 size |= u64::from(byte & 0x7f) << (4 + 7 * n);
201 }
202 self.pending.clear();
203 self.size = size;
204 self.delta = matches!(code, 6 | 7);
205 if !self.delta && size > self.max_object {
206 return Err(Violation::ObjectTooLarge { size });
207 }
208 self.phase = match code {
209 1..=4 => self.start_inflate(),
210 6 => Phase::OffsetBase,
211 7 => Phase::RefBase(20),
212 _ => return Err(Violation::Malformed(format!("an object has an unknown type {code}"))),
213 };
214 }
215 Phase::OffsetBase => {
216 let byte = input[0];
217 input = &input[1..];
218 if byte & 0x80 == 0 {
219 self.phase = self.start_inflate();
220 }
221 }
222 Phase::RefBase(left) => {
223 let skipped = left.min(input.len());
224 input = &input[skipped..];
225 self.phase = if skipped == left { self.start_inflate() } else { Phase::RefBase(left - skipped) };
226 }
227 Phase::Inflate => {
228 let flags = inflate_flags::TINFL_FLAG_PARSE_ZLIB_HEADER
229 | inflate_flags::TINFL_FLAG_HAS_MORE_INPUT
230 | inflate_flags::TINFL_FLAG_IGNORE_ADLER32;
231 let (status, consumed, written) =
232 decompress(&mut self.inflater, input, &mut self.window, self.window_at, flags);
233 if self.delta && self.head.len() < DELTA_HEAD {
234 let wanted = (DELTA_HEAD - self.head.len()).min(written);
235 for n in 0..wanted {
236 self.head.push(self.window[(self.window_at + n) & (WINDOW - 1)]);
237 }
238 }
239 self.window_at = (self.window_at + written) & (WINDOW - 1);
240 input = &input[consumed..];
241 match status {
242 TINFLStatus::Done => self.finish_object()?,
243 TINFLStatus::NeedsMoreInput | TINFLStatus::HasMoreOutput => {
244 if consumed == 0 && written == 0 && !input.is_empty() {
245 return Err(Violation::Malformed("an object stopped inflating".into()));
246 }
247 }
248 other => return Err(Violation::Malformed(format!("an object could not be inflated: {other:?}"))),
249 }
250 }
251 Phase::Trailer => return Ok(()),
252 }
253 }
254 Ok(())
255 }
256
257 /// Moves up to `wanted` bytes in all into `pending`; whether it has them.
258 fn take(&mut self, input: &mut &[u8], wanted: usize) -> bool {
259 let missing = wanted - self.pending.len();
260 let taken = missing.min(input.len());
261 self.pending.extend_from_slice(&input[..taken]);
262 *input = &input[taken..];
263 self.pending.len() == wanted
264 }
265
266 fn start_inflate(&mut self) -> Phase {
267 self.inflater.init();
268 if self.window.is_empty() {
269 self.window = vec![0; WINDOW];
270 }
271 self.window_at = 0;
272 self.head.clear();
273 Phase::Inflate
274 }
275
276 fn finish_object(&mut self) -> Result<(), Violation> {
277 let size = if self.delta {
278 let mut at = 0;
279 let _base = varint(&self.head, &mut at);
280 varint(&self.head, &mut at).ok_or_else(|| Violation::Malformed("a delta is too short".into()))?
281 } else {
282 self.size
283 };
284 if size > self.max_object {
285 return Err(Violation::ObjectTooLarge { size });
286 }
287 self.largest = self.largest.max(size);
288 self.objects += 1;
289 self.objects_left -= 1;
290 self.phase = if self.objects_left == 0 { Phase::Trailer } else { Phase::EntryHeader };
291 Ok(())
292 }
293}
294
295fn varint(data: &[u8], at: &mut usize) -> Option<u64> {
296 let mut value = 0u64;
297 let mut shift = 0;
298 loop {
299 let byte = *data.get(*at)?;
300 *at += 1;
301 value |= u64::from(byte & 0x7f) << shift;
302 if byte & 0x80 == 0 {
303 return Some(value);
304 }
305 shift += 7;
306 if shift > 63 {
307 return None;
308 }
309 }
310}
311
312/// Bytes as people read them: "31.2 MB".
313pub fn megabytes(bytes: u64) -> String {
314 format!("{:.1} MB", bytes as f64 / 1_000_000.0)
315}
316
317/// Whether a push of `incoming` bytes would take a repository holding
318/// `held` past `limit`.
319pub fn over_repo_limit(held: u64, incoming: u64, limit: u64) -> bool {
320 held.saturating_add(incoming) > limit
321}
322
323/// The upload-pack answer's side-band channels: the pack, progress, and a
324/// fatal error.
325const PACK_BAND: u8 = 1;
326const ERROR_BAND: u8 = 3;
327
328/// Takes an upload-pack answer that used side-band framing a chunk at a
329/// time and gives back the pack's bytes as they arrive.
330#[derive(Default)]
331pub struct Sideband {
332 /// A packet not yet complete: its length line, then its payload.
333 pending: Vec<u8>,
334 /// Whether the first pack bytes were `PACK`, once known.
335 started: bool,
336 /// Bytes of pack given back so far.
337 pub pack_bytes: u64,
338 done: bool,
339}
340
341impl Sideband {
342 /// The pack bytes in the next chunk of the answer, or why it failed.
343 pub fn feed(&mut self, input: &[u8]) -> Result<Vec<u8>, String> {
344 let mut out = Vec::new();
345 self.pending.extend_from_slice(input);
346 let mut at = 0;
347 while !self.done && self.pending.len() >= at + 4 {
348 let length = std::str::from_utf8(&self.pending[at..at + 4])
349 .ok()
350 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
351 .ok_or_else(|| "the source did not answer in pkt-lines".to_owned())?;
352 if length < 4 {
353 // Flush and delimiter packets carry nothing. A flush after
354 // the pack ends the answer.
355 at += 4;
356 if length == 0 && self.started {
357 self.done = true;
358 }
359 continue;
360 }
361 if self.pending.len() < at + length {
362 break;
363 }
364 let payload = &self.pending[at + 4..at + length];
365 match payload.first() {
366 Some(&PACK_BAND) => {
367 let data = &payload[1..];
368 if !self.started && !data.is_empty() {
369 if !(data.starts_with(b"PACK") || (self.pack_bytes == 0 && b"PACK".starts_with(data))) {
370 return Err(format!("the source did not send a pack: {}", String::from_utf8_lossy(data)));
371 }
372 self.started = true;
373 }
374 self.pack_bytes += data.len() as u64;
375 out.extend_from_slice(data);
376 }
377 Some(&ERROR_BAND) => {
378 return Err(format!("the source refused the fetch: {}", String::from_utf8_lossy(&payload[1..])));
379 }
380 // ACK and NAK lines, and progress.
381 _ => {}
382 }
383 at += length;
384 }
385 self.pending.drain(..at);
386 Ok(out)
387 }
388
389 /// Whether a pack arrived.
390 pub fn finish(&self) -> Result<(), String> {
391 if self.started { Ok(()) } else { Err("the source did not send a pack".to_owned()) }
392 }
393}
394
395#[cfg(test)]
396mod tests {
397 use super::*;
398 use g1t_scan::pack::{ObjectKind, write_pack};
399 use miniz_oxide::deflate::compress_to_vec_zlib;
400
401 fn pkt(payload: &str) -> Vec<u8> {
402 format!("{:04x}{payload}", payload.len() + 4).into_bytes()
403 }
404
405 fn push(pack: &[u8]) -> Vec<u8> {
406 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
407 let new = "4807077b296e6edbf410d55e72749d3e1170c291";
408 [pkt(&format!("{old} {new} refs/heads/main\0 report-status side-band-64k\n")), b"0000".to_vec(), pack.to_vec()].concat()
409 }
410
411 /// Feeds `body` in chunks of `size` bytes.
412 fn walk(body: &[u8], size: usize, max: u64) -> Result<PackSizer, Violation> {
413 let mut sizer = PackSizer::new(max);
414 for chunk in body.chunks(size) {
415 sizer.feed(chunk)?;
416 }
417 Ok(sizer)
418 }
419
420 fn noise(len: usize, seed: u32) -> Vec<u8> {
421 let mut state = seed;
422 (0..len)
423 .map(|_| {
424 state = state.wrapping_mul(1_103_515_245).wrapping_add(12_345);
425 (state >> 16) as u8
426 })
427 .collect()
428 }
429
430 #[test]
431 fn every_object_is_walked_whatever_the_chunks() {
432 let objects = vec![
433 (ObjectKind::Blob, noise(70_000, 1)),
434 (ObjectKind::Blob, b"small\n".to_vec()),
435 (ObjectKind::Blob, vec![b'a'; 200_000]),
436 (ObjectKind::Tree, Vec::new()),
437 ];
438 let body = push(&write_pack(&objects));
439 for size in [1, 3, 7, 64, 1000, 65_536, body.len()] {
440 let sizer = walk(&body, size, MAX_OBJECT_BYTES).unwrap();
441 assert_eq!(sizer.objects, 4, "chunks of {size}");
442 assert_eq!(sizer.largest, 200_000);
443 assert!(sizer.complete());
444 assert_eq!(sizer.fed, body.len() as u64);
445 assert_eq!(sizer.pack_bytes(), body.len() as u64 - (body.windows(4).position(|w| w == b"PACK").unwrap() as u64));
446 }
447 }
448
449 #[test]
450 fn an_object_over_the_limit_is_found_from_its_header() {
451 let objects = vec![(ObjectKind::Blob, vec![b'x'; 5_000]), (ObjectKind::Blob, vec![b'y'; 50])];
452 let body = push(&write_pack(&objects));
453 assert_eq!(walk(&body, 13, 4_999).err(), Some(Violation::ObjectTooLarge { size: 5_000 }));
454 assert!(walk(&body, 13, 5_000).is_ok());
455 }
456
457 /// A pack entry for a ref-delta against `base` that makes an object of
458 /// `target` bytes.
459 fn ref_delta(base_len: usize, target: usize) -> Vec<u8> {
460 fn put(mut value: usize, out: &mut Vec<u8>) {
461 loop {
462 let byte = (value & 0x7f) as u8;
463 value >>= 7;
464 if value == 0 {
465 out.push(byte);
466 return;
467 }
468 out.push(byte | 0x80);
469 }
470 }
471 let mut delta = Vec::new();
472 put(base_len, &mut delta);
473 put(target, &mut delta);
474 // Copy the base's first `target` bytes, in one instruction of up to
475 // 0x10000 per copy.
476 let mut copied = 0;
477 while copied < target {
478 let size = (target - copied).min(0xffff);
479 delta.extend_from_slice(&[0x80 | 0x01 | 0x02 | 0x10 | 0x20, (copied & 0xff) as u8, ((copied >> 8) & 0xff) as u8, (size & 0xff) as u8, ((size >> 8) & 0xff) as u8]);
480 copied += size;
481 }
482 let mut entry = Vec::new();
483 let mut size = delta.len();
484 let mut byte = (7u8 << 4) | (size & 15) as u8;
485 size >>= 4;
486 while size > 0 {
487 entry.push(byte | 0x80);
488 byte = (size & 0x7f) as u8;
489 size >>= 7;
490 }
491 entry.push(byte);
492 entry.extend_from_slice(&[0xab; 20]);
493 entry.extend(compress_to_vec_zlib(&delta, 6));
494 entry
495 }
496
497 #[test]
498 fn a_delta_is_measured_by_the_object_it_makes() {
499 let mut pack = b"PACK".to_vec();
500 pack.extend_from_slice(&2u32.to_be_bytes());
501 pack.extend_from_slice(&1u32.to_be_bytes());
502 pack.extend(ref_delta(60_000, 60_000));
503 pack.extend_from_slice(&[0u8; 20]);
504 let body = push(&pack);
505 for size in [1, 5, 4096] {
506 let sizer = walk(&body, size, MAX_OBJECT_BYTES).unwrap();
507 assert_eq!((sizer.objects, sizer.largest), (1, 60_000));
508 }
509 // The delta itself is a few bytes; the object it makes is not.
510 assert_eq!(walk(&body, 7, 59_999).err(), Some(Violation::ObjectTooLarge { size: 60_000 }));
511 }
512
513 #[test]
514 fn a_push_that_only_deletes_has_no_pack() {
515 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
516 let body = [pkt(&format!("{old} 0000000000000000000000000000000000000000 refs/heads/gone\0 report-status delete-refs\n")), b"0000".to_vec()].concat();
517 let sizer = walk(&body, 5, MAX_OBJECT_BYTES).unwrap();
518 assert_eq!(sizer.objects, 0);
519 assert!(sizer.complete());
520 assert_eq!(sizer.pack_bytes(), 0);
521 }
522
523 #[test]
524 fn push_options_before_the_pack_are_skipped() {
525 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
526 let new = "4807077b296e6edbf410d55e72749d3e1170c291";
527 let pack = write_pack(&[(ObjectKind::Blob, b"hi\n".to_vec())]);
528 let body = [
529 pkt(&format!("{old} {new} refs/heads/main\0 report-status push-options\n")),
530 b"0000".to_vec(),
531 pkt("ci.skip\n"),
532 b"0000".to_vec(),
533 pack.clone(),
534 ]
535 .concat();
536 for size in [1, 6, body.len()] {
537 let sizer = walk(&body, size, MAX_OBJECT_BYTES).unwrap();
538 assert_eq!(sizer.objects, 1);
539 assert_eq!(sizer.pack_bytes(), pack.len() as u64);
540 }
541 }
542
543 #[test]
544 fn a_body_that_is_not_a_push_is_malformed() {
545 assert!(matches!(walk(b"zzzz", 4, 10), Err(Violation::Malformed(_))));
546 assert!(matches!(walk(&push(b"NOPE\0\0\0\x02\0\0\0\x01"), 4, 10), Err(Violation::Malformed(_))));
547 // A pack cut short is not complete.
548 let body = push(&write_pack(&[(ObjectKind::Blob, noise(10_000, 3))]));
549 let cut = walk(&body[..body.len() / 2], 100, MAX_OBJECT_BYTES).unwrap();
550 assert!(!cut.complete());
551 }
552
553 #[test]
554 fn the_repository_limit_counts_what_it_holds_and_what_arrives() {
555 assert!(!over_repo_limit(900_000_000, 50_000_000, DEFAULT_REPO_LIMIT_BYTES));
556 assert!(over_repo_limit(900_000_000, 50_000_001, DEFAULT_REPO_LIMIT_BYTES));
557 assert!(!over_repo_limit(0, 0, 0));
558 assert_eq!(megabytes(31_200_000), "31.2 MB");
559 }
560
561 fn band(channel: u8, data: &[u8]) -> Vec<u8> {
562 let mut out = format!("{:04x}", data.len() + 5).into_bytes();
563 out.push(channel);
564 out.extend_from_slice(data);
565 out
566 }
567
568 #[test]
569 fn a_side_band_answer_gives_back_its_pack_whatever_the_chunks() {
570 let pack = write_pack(&[(ObjectKind::Blob, noise(30_000, 9))]);
571 let mut answer = pkt("NAK\n");
572 answer.extend(band(2, b"Counting objects\n"));
573 for part in pack.chunks(65_515) {
574 answer.extend(band(1, part));
575 }
576 answer.extend_from_slice(b"0000");
577 for size in [1, 3, 1000, answer.len()] {
578 let mut demux = Sideband::default();
579 let mut out = Vec::new();
580 for chunk in answer.chunks(size) {
581 out.extend(demux.feed(chunk).unwrap());
582 }
583 demux.finish().unwrap();
584 assert_eq!(out, pack, "chunks of {size}");
585 assert_eq!(demux.pack_bytes, pack.len() as u64);
586 }
587 }
588
589 #[test]
590 fn a_side_band_error_or_no_pack_fails() {
591 let mut demux = Sideband::default();
592 let answer = [pkt("NAK\n"), band(3, b"upload-pack: not our ref")].concat();
593 assert!(demux.feed(&answer).unwrap_err().contains("not our ref"));
594 let mut empty = Sideband::default();
595 empty.feed(&[pkt("NAK\n"), b"0000".to_vec()].concat()).unwrap();
596 assert!(empty.finish().is_err());
597 }
598}