Sign in

flagon-io/g1t

public

g1t/services/repos/src/pack_limits.rs

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