alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::HashMap;
2
3use flate2::{Decompress, FlushDecompress, Status};
4use gix_pack::data::{Entry, entry::Header};
5
6use knot_types::ObjectCount;
7
8use crate::error::{PackError, PackLimit};
9use crate::ids::PackOffset;
10use crate::meter::{PackLimits, check_depth, malformed, pack_object_count};
11use crate::pkt::{self, Frame};
12
13const HEADER_SLACK: usize = 64;
14
15fn pack_start(buf: &[u8]) -> Option<usize> {
16 let caps = pkt::first_command(buf)
17 .map(pkt::parse_caps)
18 .unwrap_or_default();
19 let boundary = if caps.push_options { 2 } else { 1 };
20 pkt::frames(buf, Some(boundary))
21 .filter_map(|item| match item {
22 Ok((Frame::Flush, at)) => Some(at),
23 _ => None,
24 })
25 .nth(boundary - 1)
26}
27
28fn new_oid_field(line: &[u8]) -> Option<&[u8]> {
29 let line = line.split(|byte| *byte == 0).next().unwrap_or(line);
30 line.split(|byte| *byte == b' ').nth(1)
31}
32
33fn no_pack_needed(buf: &[u8]) -> bool {
34 pkt::frames(buf, Some(1))
35 .filter_map(|item| match item {
36 Ok((Frame::Data(payload), _)) => Some(payload),
37 _ => None,
38 })
39 .all(|line| {
40 new_oid_field(line)
41 .map(|oid| oid.iter().all(|byte| *byte == b'0'))
42 .unwrap_or(false)
43 })
44}
45
46const PREAMBLE_SCAN_LIMIT: usize = 16 * 1024 * 1024;
47
48trait PackSource {
49 fn len(&self) -> u64;
50 fn read_at(&self, offset: u64, buf: &mut [u8]) -> std::io::Result<usize>;
51}
52
53impl PackSource for [u8] {
54 fn len(&self) -> u64 {
55 <[u8]>::len(self) as u64
56 }
57 fn read_at(&self, offset: u64, buf: &mut [u8]) -> std::io::Result<usize> {
58 let start = usize::try_from(offset)
59 .unwrap_or(usize::MAX)
60 .min(<[u8]>::len(self));
61 let read = (<[u8]>::len(self) - start).min(buf.len());
62 buf[..read].copy_from_slice(&self[start..start + read]);
63 Ok(read)
64 }
65}
66
67struct FileSource<'a> {
68 file: &'a std::fs::File,
69 len: u64,
70}
71
72impl PackSource for FileSource<'_> {
73 fn len(&self) -> u64 {
74 self.len
75 }
76 fn read_at(&self, offset: u64, buf: &mut [u8]) -> std::io::Result<usize> {
77 use std::os::unix::fs::FileExt;
78 let read = self.len.saturating_sub(offset).min(buf.len() as u64) as usize;
79 self.file.read_exact_at(&mut buf[..read], offset)?;
80 Ok(read)
81 }
82}
83
84fn read_head<S: PackSource + ?Sized>(source: &S, upto: u64) -> Result<Vec<u8>, PackError> {
85 let limit = upto.min(PREAMBLE_SCAN_LIMIT as u64) as usize;
86 let mut head = vec![0u8; limit];
87 let read = source
88 .read_at(0, &mut head)
89 .map_err(|error| PackError::Pack(format!("pack read: {error}")))?;
90 head.truncate(read);
91 Ok(head)
92}
93
94struct EntryInflate {
95 data_offset: PackOffset,
96 decompressed_size: u64,
97 decompress: Decompress,
98 produced: u64,
99}
100
101impl EntryInflate {
102 fn feed<S: PackSource + ?Sized>(
103 &mut self,
104 pack: &S,
105 scratch: &mut [u8],
106 chunk: &mut [u8],
107 ) -> Result<Option<PackOffset>, PackError> {
108 loop {
109 let consumed = self.decompress.total_in();
110 let out_before = self.decompress.total_out();
111 let read = pack
112 .read_at(self.data_offset.get() + consumed, chunk)
113 .map_err(|error| PackError::Pack(format!("pack read: {error}")))?;
114 let status = self
115 .decompress
116 .decompress(&chunk[..read], scratch, FlushDecompress::None)
117 .map_err(|error| PackError::Pack(format!("inflate: {error}")))?;
118 self.produced += self.decompress.total_out() - out_before;
119 if self.produced > self.decompressed_size {
120 return Err(malformed("object inflates beyond its declared size"));
121 }
122 match status {
123 Status::StreamEnd => {
124 return if self.produced == self.decompressed_size {
125 self.data_offset
126 .get()
127 .checked_add(self.decompress.total_in())
128 .map(|offset| Some(PackOffset::new(offset)))
129 .ok_or_else(|| malformed("pack offset overflow"))
130 } else {
131 Err(malformed("object decompressed size mismatch"))
132 };
133 }
134 Status::Ok | Status::BufError => {
135 if self.decompress.total_in() == consumed
136 && self.decompress.total_out() == out_before
137 {
138 return Ok(None);
139 }
140 }
141 }
142 }
143 }
144}
145
146#[derive(Default)]
147struct PackProgress {
148 num_objects: ObjectCount,
149 objects_done: ObjectCount,
150 next_offset: PackOffset,
151 total_decompressed: u64,
152 base_of: HashMap<PackOffset, PackOffset>,
153 current: Option<EntryInflate>,
154 depth_checked: bool,
155}
156
157impl PackProgress {
158 fn scan<S: PackSource + ?Sized>(
159 &mut self,
160 pack: &S,
161 limits: &PackLimits,
162 kind: gix::hash::Kind,
163 ) -> Result<Option<usize>, PackError> {
164 let hash_len = kind.len_in_bytes();
165 let len = pack.len();
166 let mut scratch = [0u8; 8192];
167 let mut chunk = [0u8; 8192];
168 let mut header = [0u8; HEADER_SLACK];
169 loop {
170 if self.objects_done == self.num_objects {
171 if !self.depth_checked {
172 check_depth(&self.base_of, limits.max_delta_depth)?;
173 self.depth_checked = true;
174 }
175 let total_len = (self.next_offset.get() as usize)
176 .checked_add(hash_len)
177 .ok_or_else(|| malformed("pack length overflow"))?;
178 return Ok((len as usize >= total_len).then_some(total_len));
179 }
180 match self.current.as_mut() {
181 Some(entry) => match entry.feed(pack, &mut scratch, &mut chunk)? {
182 Some(next_offset) => {
183 self.next_offset = next_offset;
184 self.objects_done = self.objects_done.succ();
185 self.current = None;
186 }
187 None => return Ok(None),
188 },
189 None => {
190 let start = self.next_offset;
191 if start.get() >= len {
192 return Ok(None);
193 }
194 let read = pack
195 .read_at(start.get(), &mut header)
196 .map_err(|error| PackError::Pack(format!("pack read: {error}")))?;
197 let mut reader: &[u8] = &header[..read];
198 let entry = match Entry::from_read(&mut reader, start.get(), hash_len) {
199 Ok(entry) => entry,
200 Err(error) => {
201 return if len.saturating_sub(start.get()) < HEADER_SLACK as u64 {
202 Ok(None)
203 } else {
204 Err(PackError::Pack(error.to_string()))
205 };
206 }
207 };
208 if limits.max_object_bytes.exceeded_by(entry.decompressed_size) {
209 return Err(PackError::LimitExceeded(PackLimit::ObjectBytes));
210 }
211 self.total_decompressed = self
212 .total_decompressed
213 .checked_add(entry.decompressed_size)
214 .ok_or_else(|| malformed("decompressed size overflow"))?;
215 if limits.max_total_bytes.exceeded_by(self.total_decompressed) {
216 return Err(PackError::LimitExceeded(PackLimit::TotalBytes));
217 }
218 if let Header::OfsDelta { base_distance } = entry.header {
219 let base = entry
220 .checked_base_pack_offset(base_distance)
221 .ok_or_else(|| malformed("ofs-delta base out of range"))?;
222 self.base_of.insert(self.next_offset, PackOffset::new(base));
223 }
224 self.current = Some(EntryInflate {
225 data_offset: PackOffset::new(entry.data_offset),
226 decompressed_size: entry.decompressed_size,
227 decompress: Decompress::new(true),
228 produced: 0,
229 });
230 }
231 }
232 }
233 }
234}
235
236pub struct ReceiveFramer {
237 limits: PackLimits,
238 kind: gix::hash::Kind,
239 pack_start: Option<usize>,
240 pack: Option<PackProgress>,
241}
242
243impl ReceiveFramer {
244 pub fn new(limits: PackLimits, kind: gix::hash::Kind) -> Self {
245 Self {
246 limits,
247 kind,
248 pack_start: None,
249 pack: None,
250 }
251 }
252
253 pub fn pack_start(&self) -> Option<usize> {
254 self.pack_start
255 }
256
257 pub fn advance_bytes(&mut self, buf: &[u8]) -> Result<Option<usize>, PackError> {
258 self.advance(buf)
259 }
260
261 pub fn advance_file(
262 &mut self,
263 file: &std::fs::File,
264 len: u64,
265 ) -> Result<Option<usize>, PackError> {
266 self.advance(&FileSource { file, len })
267 }
268
269 fn advance<S: PackSource + ?Sized>(&mut self, source: &S) -> Result<Option<usize>, PackError> {
270 let len = source.len();
271 let pack_start = match self.pack_start {
272 Some(start) => start,
273 None => {
274 let head = read_head(source, len)?;
275 match pack_start(&head) {
276 Some(start) => {
277 self.pack_start = Some(start);
278 start
279 }
280 None => return Ok(None),
281 }
282 }
283 };
284 if self.pack.is_none() {
285 let pack_len = len.saturating_sub(pack_start as u64);
286 if pack_len == 0 {
287 let head = read_head(source, pack_start as u64)?;
288 return Ok(no_pack_needed(&head).then_some(pack_start));
289 }
290 if pack_len < 12 {
291 return Ok(None);
292 }
293 let mut header = [0u8; 12];
294 source
295 .read_at(pack_start as u64, &mut header)
296 .map_err(|error| PackError::Pack(format!("pack read: {error}")))?;
297 if &header[..4] != b"PACK" {
298 return Err(malformed("packfile is missing its PACK signature"));
299 }
300 let num_objects = pack_object_count(&header)?;
301 if num_objects > self.limits.max_objects {
302 return Err(PackError::LimitExceeded(PackLimit::Objects));
303 }
304 self.pack = Some(PackProgress {
305 num_objects,
306 next_offset: PackOffset::new(pack_start as u64 + 12),
307 ..PackProgress::default()
308 });
309 }
310 let kind = self.kind;
311 self.pack
312 .as_mut()
313 .expect("pack progress initialized")
314 .scan(source, &self.limits, kind)
315 }
316}
317
318pub fn receive_request_complete(
319 buf: &[u8],
320 limits: &PackLimits,
321 kind: gix::hash::Kind,
322) -> Result<Option<usize>, PackError> {
323 ReceiveFramer::new(*limits, kind).advance_bytes(buf)
324}
325
326pub fn archive_request_complete(buf: &[u8]) -> Option<usize> {
327 pkt::frames(buf, Some(1)).find_map(|item| match item {
328 Ok((Frame::Flush, end)) => Some(end),
329 _ => None,
330 })
331}
332
333#[derive(Default)]
334pub struct UploadFramer {
335 scanned: usize,
336 v2: bool,
337 flushes: usize,
338 complete: Option<usize>,
339}
340
341impl UploadFramer {
342 pub fn new() -> Self {
343 Self::default()
344 }
345
346 pub fn advance(&mut self, buf: &[u8]) -> Option<usize> {
347 if self.complete.is_some() {
348 return self.complete;
349 }
350 let base = self.scanned;
351 for item in pkt::frames(&buf[base..], None) {
352 let Ok((frame, at)) = item else { break };
353 let boundary = base + at;
354 match frame {
355 Frame::Data(payload) => {
356 if payload.starts_with(b"command=") {
357 self.v2 = true;
358 }
359 let trimmed = payload
360 .iter()
361 .rposition(|byte| !byte.is_ascii_whitespace())
362 .map(|end| &payload[..=end])
363 .unwrap_or(payload);
364 if !self.v2 && trimmed == b"done" {
365 self.complete = Some(boundary);
366 return self.complete;
367 }
368 }
369 Frame::Flush => {
370 self.flushes += 1;
371 if self.v2 {
372 self.complete = Some(boundary);
373 return self.complete;
374 }
375 }
376 _ => {}
377 }
378 self.scanned = boundary;
379 }
380 None
381 }
382
383 pub fn unanswered_flushes(&self) -> usize {
384 self.flushes.saturating_sub(1)
385 }
386}
387
388pub fn upload_v0_nak() -> Vec<u8> {
389 let mut buf = Vec::new();
390 pkt::write_data(&mut buf, b"NAK\n").expect("write to in-memory buffer never fails");
391 buf
392}
393
394#[cfg(test)]
395mod tests {
396 use super::*;
397
398 fn v2_fetch_request() -> Vec<u8> {
399 let mut buf = Vec::new();
400 pkt::write_data(&mut buf, b"command=fetch\n").unwrap();
401 pkt::write_delim(&mut buf).unwrap();
402 pkt::write_data(&mut buf, b"want 1111111111111111111111111111111111111111\n").unwrap();
403 pkt::write_data(&mut buf, b"want 2222222222222222222222222222222222222222\n").unwrap();
404 pkt::write_data(&mut buf, b"done\n").unwrap();
405 pkt::write_flush(&mut buf).unwrap();
406 buf
407 }
408
409 #[test]
410 fn upload_framer_completes_a_v2_request_at_the_terminating_flush() {
411 let request = v2_fetch_request();
412 assert_eq!(UploadFramer::new().advance(&request), Some(request.len()));
413 }
414
415 #[test]
416 fn upload_framer_fed_one_byte_at_a_time_never_overruns_the_buffer() {
417 let request = v2_fetch_request();
418 let mut framer = UploadFramer::new();
419 let mut buf = Vec::new();
420 let mut completed = None;
421 for byte in &request {
422 buf.push(*byte);
423 if let Some(len) = framer.advance(&buf) {
424 assert!(
425 len <= buf.len(),
426 "advance returned {len} past buffer of {}",
427 buf.len()
428 );
429 completed = Some(len);
430 break;
431 }
432 }
433 assert_eq!(
434 completed,
435 Some(request.len()),
436 "incrementally fed request completes exactly once the whole buffer has arrived"
437 );
438 }
439
440 #[test]
441 fn upload_framer_counts_v0_have_batch_flushes_without_completing() {
442 let mut buf = Vec::new();
443 pkt::write_data(&mut buf, b"want 1111111111111111111111111111111111111111\n").unwrap();
444 pkt::write_flush(&mut buf).unwrap();
445 pkt::write_data(&mut buf, b"have 2222222222222222222222222222222222222222\n").unwrap();
446 pkt::write_flush(&mut buf).unwrap();
447 let mut framer = UploadFramer::new();
448 assert_eq!(framer.advance(&buf), None, "v0 request is open until done");
449 assert_eq!(
450 framer.unanswered_flushes(),
451 1,
452 "two flushes seen, one have-batch awaits NAK"
453 );
454 }
455}