git_zero_core/pack/
receive.rs

1//! Streaming pack decoding for intake/indexing, independent of receive-pack control.
2//! Server push negotiation and status live in git-zero-server's `receive` module.
3//! Incremental pack intake. All emitted bytes and metadata are provisional
4//! until the pack trailer is verified. The host owns quarantine and publication.
5use super::{EntryHeader, EntryKind, crc::Crc32, decode::parse_entry_header};
6use crate::{
7    Error, ObjectDigest, ObjectFormat, ObjectId, ObjectInfo, Result,
8    object::CheckedHash as Hash,
9    zlib::{InflateLimits, InflateStatus, Inflater, WINDOW_SIZE},
10};
11
12/// Independent wire, object-count, and per-entry expansion budgets.
13#[derive(Clone, Copy, Debug, Eq, PartialEq)]
14pub struct DecodeLimits {
15    pub max_pack_bytes: u64,
16    pub max_objects: u32,
17    /// Bounds inflated object bodies or delta instruction streams, not the
18    /// eventual result of applying a delta. Hosts must bound delta resolution.
19    pub inflate: InflateLimits,
20}
21
22/// Events stop decoding before the next entry. Hosts can stage body fragments
23/// and persist metadata without retaining an object or an object-count array.
24#[derive(Clone, Copy, Debug, Eq, PartialEq)]
25pub enum DecodeStatus {
26    NeedInput,
27    NeedOutput,
28    Header {
29        version: u32,
30        objects: u32,
31    },
32    EntryStart(EntryHeader),
33    EntryEnd {
34        end_offset: u64,
35        crc32: u32,
36        /// Canonical ID for a plain object. Delta IDs require base resolution.
37        object_id: Option<ObjectId>,
38    },
39    Finished(ObjectId),
40}
41
42#[derive(Clone, Copy, Debug, Eq, PartialEq)]
43pub struct DecodeProgress {
44    pub consumed: usize,
45    pub written: usize,
46    pub status: DecodeStatus,
47}
48
49#[derive(Clone, Copy)]
50enum Phase {
51    Header,
52    Entry,
53    Body(EntryHeader),
54    Trailer,
55    Done(ObjectId),
56}
57
58/// Push-driven SHA-1/SHA-256 pack decoder with fixed inline state and one
59/// caller-owned 32 KiB inflate window. Neither input nor output is retained.
60/// Supports plain, offset-delta and reference-delta entries, including thin
61/// packs; checking base existence and graph connectivity is a host operation.
62///
63/// Preserve the unconsumed input suffix between calls. `EntryEnd` can include
64/// the final body fragment in `written`; persist it before handling the event.
65/// Every error is terminal, and all previously staged data must be discarded.
66/// No ref may be published before `Finished` and host object/graph validation.
67pub struct Decoder<'a> {
68    inflater: Inflater<'a>,
69    format: ObjectFormat,
70    limits: DecodeLimits,
71    phase: Phase,
72    header: [u8; 64],
73    header_len: usize,
74    offset: u64,
75    entry_offset: u64,
76    remaining: u32,
77    hash: Hash,
78    crc: Crc32,
79    object_hash: Option<ObjectDigest>,
80    failure: Option<Error>,
81}
82
83/// Stream one entry from a trusted pack offset, without loading the pack or
84/// its index. This shares framing, CRC, inflation, size and object-ID handling
85/// with `Decoder`. It does not verify the whole pack's checksum or graph.
86/// Compare the final CRC/ID with authenticated index metadata before use.
87///
88/// Events are `EntryStart`, `NeedInput`, `NeedOutput` and `EntryEnd`. At entry
89/// completion the following bytes stay unconsumed, even if `end_of_input` is
90/// true. Completed updates repeat `EntryEnd` with zero bytes consumed/written.
91/// The host may await storage between updates, retaining only the inflate
92/// window and this fixed state. Delta bodies contain instructions, not objects.
93pub struct EntryDecoder<'a> {
94    decoder: Decoder<'a>,
95    completion: Option<DecodeStatus>,
96}
97
98impl<'a> EntryDecoder<'a> {
99    pub fn new(
100        format: ObjectFormat,
101        offset: u64,
102        window: &'a mut [u8; WINDOW_SIZE],
103        max_entry_bytes: u64,
104        inflate: InflateLimits,
105    ) -> Result<Self> {
106        if offset < 12 {
107            return Err(Error::CorruptPack);
108        }
109        let end = offset
110            .checked_add(max_entry_bytes)
111            .ok_or(Error::LimitExceeded)?;
112        let mut decoder = Decoder::new(
113            format,
114            window,
115            DecodeLimits {
116                max_pack_bytes: end,
117                max_objects: 1,
118                inflate,
119            },
120        );
121        decoder.phase = Phase::Entry;
122        decoder.offset = offset;
123        decoder.entry_offset = offset;
124        decoder.remaining = 1;
125        Ok(Self {
126            decoder,
127            completion: None,
128        })
129    }
130
131    pub fn position(&self) -> u64 {
132        self.decoder.position()
133    }
134
135    pub fn update(
136        &mut self,
137        input: &[u8],
138        output: &mut [u8],
139        end_of_input: bool,
140    ) -> Result<DecodeProgress> {
141        if let Some(status) = self.completion {
142            return Ok(DecodeProgress {
143                consumed: 0,
144                written: 0,
145                status,
146            });
147        }
148        let progress = self.decoder.update(input, output, end_of_input)?;
149        if matches!(progress.status, DecodeStatus::EntryEnd { .. }) {
150            self.completion = Some(progress.status);
151        }
152        Ok(progress)
153    }
154
155    /// Require a complete entry. Calling early leaves the decoder resumable.
156    pub fn finish(&self) -> Result<()> {
157        if let Some(error) = self.decoder.failure {
158            return Err(error);
159        }
160        if self.completion.is_some() {
161            Ok(())
162        } else {
163            Err(Error::CorruptPack)
164        }
165    }
166}
167
168impl<'a> Decoder<'a> {
169    pub fn new(
170        format: ObjectFormat,
171        window: &'a mut [u8; WINDOW_SIZE],
172        limits: DecodeLimits,
173    ) -> Self {
174        Self {
175            inflater: Inflater::new(window, limits.inflate),
176            format,
177            limits,
178            phase: Phase::Header,
179            header: [0; 64],
180            header_len: 0,
181            offset: 0,
182            entry_offset: 12,
183            remaining: 0,
184            hash: format.checked_hash(),
185            crc: Crc32::new(),
186            object_hash: None,
187            failure: None,
188        }
189    }
190
191    pub fn position(&self) -> u64 {
192        self.offset
193    }
194
195    /// `end_of_input` asserts that this slice contains the final available
196    /// bytes. Events and output exhaustion can return before consuming it all;
197    /// resubmit the remaining suffix with the same flag. A request for more
198    /// input at EOF is an error. Bytes after the trailer are always rejected.
199    pub fn update(
200        &mut self,
201        input: &[u8],
202        output: &mut [u8],
203        end_of_input: bool,
204    ) -> Result<DecodeProgress> {
205        if let Some(error) = self.failure {
206            return Err(error);
207        }
208        let result = self.step(input, output).and_then(|progress| {
209            if end_of_input && progress.status == DecodeStatus::NeedInput {
210                Err(Error::CorruptPack)
211            } else {
212                Ok(progress)
213            }
214        });
215        if let Err(error) = result {
216            self.failure = Some(error);
217        }
218        result
219    }
220
221    /// Require a verified trailer. An early call does not invalidate decoding;
222    /// the host can continue supplying data. Transport EOF remains the host's
223    /// responsibility, including checking for data after a verified trailer.
224    pub fn finish(&self) -> Result<ObjectId> {
225        if let Some(error) = self.failure {
226            return Err(error);
227        }
228        match self.phase {
229            Phase::Done(id) => Ok(id),
230            _ => Err(Error::CorruptPack),
231        }
232    }
233
234    fn step(&mut self, input: &[u8], output: &mut [u8]) -> Result<DecodeProgress> {
235        let mut consumed = 0;
236        let status = loop {
237            match self.phase {
238                Phase::Done(id) => {
239                    if consumed != input.len() {
240                        return Err(Error::CorruptPack);
241                    }
242                    break DecodeStatus::Finished(id);
243                }
244                Phase::Body(entry) => return self.body(input, output, entry),
245                Phase::Header | Phase::Entry | Phase::Trailer => {
246                    let Some(&byte) = input.get(consumed) else {
247                        break DecodeStatus::NeedInput;
248                    };
249                    self.account(&[byte], !matches!(self.phase, Phase::Trailer))?;
250                    consumed += 1;
251                    *self
252                        .header
253                        .get_mut(self.header_len)
254                        .ok_or(Error::CorruptPack)? = byte;
255                    self.header_len += 1;
256                    if let Some(status) = self.header_byte()? {
257                        break status;
258                    }
259                }
260            }
261        };
262        Ok(DecodeProgress {
263            consumed,
264            written: 0,
265            status,
266        })
267    }
268
269    fn account(&mut self, bytes: &[u8], hashed: bool) -> Result<()> {
270        let end = self
271            .offset
272            .checked_add(bytes.len() as u64)
273            .ok_or(Error::LimitExceeded)?;
274        if end > self.limits.max_pack_bytes {
275            return Err(Error::LimitExceeded);
276        }
277        self.offset = end;
278        if hashed {
279            self.hash.update(bytes);
280        }
281        if matches!(self.phase, Phase::Entry | Phase::Body(_)) {
282            self.crc.update(bytes);
283        }
284        Ok(())
285    }
286
287    fn header_byte(&mut self) -> Result<Option<DecodeStatus>> {
288        match self.phase {
289            Phase::Header if self.header_len == 12 => self.pack_header().map(Some),
290            Phase::Entry => self.start_entry(),
291            Phase::Trailer if self.header_len == self.format.id_length() => {
292                self.verify_trailer()?;
293                Ok(None)
294            }
295            _ => Ok(None),
296        }
297    }
298
299    fn pack_header(&mut self) -> Result<DecodeStatus> {
300        let version = u32::from_be_bytes(self.header[4..8].try_into().unwrap());
301        if &self.header[..4] != b"PACK" || !matches!(version, 2 | 3) {
302            return Err(Error::CorruptPack);
303        }
304        self.remaining = u32::from_be_bytes(self.header[8..12].try_into().unwrap());
305        if self.remaining > self.limits.max_objects {
306            return Err(Error::LimitExceeded);
307        }
308        self.next();
309        Ok(DecodeStatus::Header {
310            version,
311            objects: self.remaining,
312        })
313    }
314
315    fn start_entry(&mut self) -> Result<Option<DecodeStatus>> {
316        let Some(entry) = entry_header(
317            &self.header[..self.header_len],
318            self.entry_offset,
319            self.format,
320        )?
321        else {
322            return Ok(None);
323        };
324        if entry.size > self.limits.inflate.max_output {
325            return Err(Error::LimitExceeded);
326        }
327        self.inflater.reset(InflateLimits {
328            max_output: entry.size,
329            ..self.limits.inflate
330        });
331        self.object_hash = match entry.kind {
332            EntryKind::Object(kind) => Some(ObjectDigest::new(
333                ObjectInfo {
334                    kind,
335                    size: entry.size,
336                },
337                self.format,
338            )),
339            _ => None,
340        };
341        self.header_len = 0;
342        self.phase = Phase::Body(entry);
343        Ok(Some(DecodeStatus::EntryStart(entry)))
344    }
345
346    fn verify_trailer(&mut self) -> Result<()> {
347        let id = self.hash.clone().finish().map_err(|_| Error::CorruptPack)?;
348        if id.as_bytes() != &self.header[..self.header_len] {
349            return Err(Error::CorruptPack);
350        }
351        self.phase = Phase::Done(id);
352        Ok(())
353    }
354
355    fn next(&mut self) {
356        self.phase = if self.remaining == 0 {
357            Phase::Trailer
358        } else {
359            Phase::Entry
360        };
361        self.header_len = 0;
362        self.entry_offset = self.offset;
363        self.crc = Crc32::new();
364    }
365
366    fn body(
367        &mut self,
368        input: &[u8],
369        output: &mut [u8],
370        entry: EntryHeader,
371    ) -> Result<DecodeProgress> {
372        let allowance = (self.limits.max_pack_bytes - self.offset).min(input.len() as u64) as usize;
373        let input = &input[..allowance];
374        let progress = self
375            .inflater
376            .update(input, output, false)
377            .map_err(|error| {
378                if error == Error::LimitExceeded {
379                    error
380                } else {
381                    Error::CorruptPack
382                }
383            })?;
384        self.account(&input[..progress.consumed], true)?;
385        if progress.status == InflateStatus::NeedInput && self.offset == self.limits.max_pack_bytes
386        {
387            return Err(Error::LimitExceeded);
388        }
389        if let Some(hash) = &mut self.object_hash {
390            hash.update(&output[..progress.written])
391                .map_err(|_| Error::CorruptPack)?;
392        }
393        let status = match progress.status {
394            InflateStatus::NeedInput => DecodeStatus::NeedInput,
395            InflateStatus::NeedOutput => DecodeStatus::NeedOutput,
396            InflateStatus::Finished => {
397                if self.inflater.total_output() != entry.size {
398                    return Err(Error::CorruptPack);
399                }
400                let event = DecodeStatus::EntryEnd {
401                    end_offset: self.offset,
402                    crc32: self.crc.value(),
403                    object_id: self
404                        .object_hash
405                        .take()
406                        .map(|hash| hash.finish())
407                        .transpose()
408                        .map_err(|_| Error::CorruptPack)?,
409                };
410                self.remaining -= 1;
411                self.next();
412                event
413            }
414        };
415        Ok(DecodeProgress {
416            consumed: progress.consumed,
417            written: progress.written,
418            status,
419        })
420    }
421}
422
423fn entry_header(bytes: &[u8], offset: u64, format: ObjectFormat) -> Result<Option<EntryHeader>> {
424    match parse_entry_header(bytes, offset, format) {
425        Ok((_, header)) => Ok(Some(header)),
426        Err(Error::BufferTooSmall) => Ok(None),
427        Err(error) => Err(error),
428    }
429}
430
431#[cfg(test)]
432mod tests {
433    use super::*;
434
435    #[test]
436    fn wire_budget_bounds_inflater_work_on_empty_nonfinal_blocks() {
437        let mut window = [0; WINDOW_SIZE];
438        let limits = DecodeLimits {
439            max_pack_bytes: 16,
440            max_objects: 1,
441            inflate: InflateLimits {
442                max_output: 0,
443                max_ratio: 1,
444                ratio_slack: 0,
445            },
446        };
447        let mut decoder = Decoder::new(ObjectFormat::Sha1, &mut window, limits);
448        decoder
449            .update(b"PACK\0\0\0\x02\0\0\0\x01", &mut [], false)
450            .unwrap();
451        decoder.update(&[0x30], &mut [], false).unwrap();
452        let mut body = [0; 511];
453        body[..2].copy_from_slice(&[0x78, 0x01]);
454        for block in body[2..502].chunks_exact_mut(5) {
455            block.copy_from_slice(&[0, 0, 0, 255, 255]);
456        }
457        body[502..].copy_from_slice(&[1, 0, 0, 255, 255, 0, 0, 0, 1]);
458        assert_eq!(
459            decoder.update(&body, &mut [], false),
460            Err(Error::LimitExceeded)
461        );
462        assert!(decoder.inflater.total_input() <= 3);
463        assert_eq!(
464            decoder.update(&[], &mut [], true),
465            Err(Error::LimitExceeded)
466        );
467    }
468
469    #[test]
470    fn entry_sizes_and_delta_offsets_are_not_limited_to_32_bits() {
471        let header = [0xbf, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x0f];
472        let entry = entry_header(&header, 12, ObjectFormat::Sha1)
473            .unwrap()
474            .unwrap();
475        assert_eq!(entry.size, u64::MAX);
476        let distance = u64::from(u32::MAX) + 129;
477        let mut encoded = [0; 11];
478        let mut cursor = encoded.len() - 1;
479        let mut value = distance;
480        encoded[cursor] = (value & 127) as u8;
481        value >>= 7;
482        while value != 0 {
483            value -= 1;
484            cursor -= 1;
485            encoded[cursor] = 128 | (value & 127) as u8;
486            value >>= 7;
487        }
488        cursor -= 1;
489        encoded[cursor] = 0x60;
490        let offset = distance + 12;
491        let entry = entry_header(&encoded[cursor..], offset, ObjectFormat::Sha1)
492            .unwrap()
493            .unwrap();
494        assert_eq!(entry.offset, offset);
495        assert_eq!(entry.kind, EntryKind::OffsetDelta { base_offset: 12 });
496        assert_eq!(
497            entry_header(&encoded[cursor..], distance + 11, ObjectFormat::Sha1),
498            Err(Error::CorruptPack)
499        );
500    }
501}
git_zero_core/pack/receive.rs

git_zero_core/pack/
receive.rs
