use std::collections::{BTreeMap, HashMap};
use std::io::Cursor;
use bytes::{Buf, Bytes, BytesMut};
use cid::Cid as IpldCid;
use jacquard_common::types::crypto::{DAG_CBOR, SHA2_256};
use miette::{IntoDiagnostic, Result};
use sha2::{Digest, Sha256};
/// CIDv1 (dag-cbor, sha2-256) of a raw DAG-CBOR block.
pub(crate) fn compute_cid(bytes: &[u8]) -> IpldCid {
let digest = Sha256::digest(bytes);
let mh = cid::multihash::Multihash::<64>::wrap(SHA2_256, &digest)
.expect("sha2-256 digest fits a 64-byte multihash");
IpldCid::new_v1(DAG_CBOR, mh)
}
/// blocks of a whole-repo CAR. hashed rather than ordered: the MST walk looks up every
/// block, and cids are chosen by the PDS, so the hasher is randomly seeded.
pub(crate) type CarBlocks = HashMap<IpldCid, Bytes, ahash::RandomState>;
/// blocks of a commit-sized car (firehose `#commit` and `#sync`, `getRecord` proofs,
/// `getBlocks`), ordered: the shape jacquard's `MemoryBlockStore` takes.
pub(crate) type CommitBlocks = BTreeMap<IpldCid, Bytes>;
pub(crate) type CommitCar = ParsedCar<CommitBlocks>;
/// repo blocks average a few hundred bytes; sizing for that avoids most rehashing.
const EXPECTED_SECTION_BYTES: usize = 256;
/// the biggest block a streamed car may hold. a stream keeps a block whole until it ends, so
/// without a cap a pds could park the whole car in memory as one block, past a spill. the
/// firehose stops records at 1 MiB but sends bigger updates as a #sync, which ends up here, so
/// this leaves some room past that and still turns away a broken car
const MAX_STREAMED_SECTION_BYTES: usize = 2 * 1024 * 1024;
#[derive(Debug)]
pub(crate) struct ParsedCar<M> {
pub root: IpldCid,
pub blocks: Blocks<M>,
}
/// a car's blocks by cid, and whether the parse hashed every one of them against its cid. only
/// a verifying parse sets that, and only other `Blocks` can be added afterwards, so a block
/// handed out as checked was checked.
#[derive(Debug)]
pub(crate) struct Blocks<M> {
map: M,
cids_checked: bool,
}
/// the lookup [`Blocks`] needs from the map behind it.
pub(crate) trait BlockMap {
fn get(&self, cid: &IpldCid) -> Option<&Bytes>;
}
impl<S: std::hash::BuildHasher> BlockMap for HashMap<IpldCid, Bytes, S> {
fn get(&self, cid: &IpldCid) -> Option<&Bytes> {
HashMap::get(self, cid)
}
}
impl BlockMap for CommitBlocks {
fn get(&self, cid: &IpldCid) -> Option<&Bytes> {
BTreeMap::get(self, cid)
}
}
impl<M: BlockMap> Blocks<M> {
/// blocks nothing has checked yet; each one is hashed when it is verified.
#[cfg(test)]
pub(crate) fn unverified(map: M) -> Self {
Self {
map,
cids_checked: false,
}
}
/// the bytes under `cid`, for reading them; storing them takes [`Blocks::block`].
pub(crate) fn get(&self, cid: &IpldCid) -> Option<&Bytes> {
self.map.get(cid)
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub(crate) fn block(&self, cid: &IpldCid) -> Option<CarBlock> {
self.map.get(cid).map(|bytes| CarBlock {
cid: *cid,
bytes: bytes.clone(),
cid_checked: self.cids_checked,
})
}
pub(crate) fn map(&self) -> &M {
&self.map
}
}
/// no blocks, so every one of them is checked.
impl<M: Default> Default for Blocks<M> {
fn default() -> Self {
Self {
map: M::default(),
cids_checked: true,
}
}
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
impl Blocks<CommitBlocks> {
/// adds `other`'s blocks, which replace any with the same cid. the result is checked only
/// if both sides were.
pub(crate) fn extend(&mut self, other: Self) {
self.map.extend(other.map);
self.cids_checked &= other.cids_checked;
}
}
/// a block as a car delivered it. its bytes are only reachable through [`CarBlock::verify`],
/// which costs a hash only if the parse did not check it already.
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
#[derive(Debug)]
pub(crate) struct CarBlock {
cid: IpldCid,
bytes: Bytes,
cid_checked: bool,
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
impl CarBlock {
/// a block for `bytes`, under the cid they hash to.
#[cfg(test)]
pub(crate) fn from_body(bytes: impl Into<Bytes>) -> Self {
let bytes = bytes.into();
let cid = compute_cid(&bytes);
Self {
cid,
bytes,
cid_checked: true,
}
}
/// a block whose bytes were just hashed against `cid`
pub(crate) fn hashed(cid: IpldCid, bytes: Bytes) -> Self {
Self {
cid,
bytes,
cid_checked: true,
}
}
/// how many bytes the block holds, which needs no check
pub(crate) fn size(&self) -> usize {
self.bytes.len()
}
/// the cid the car claims for this block.
#[cfg(test)]
pub(crate) fn cid(&self) -> &IpldCid {
&self.cid
}
/// the same block, hashed now if the parse didn't, so verifying it later costs nothing
pub(crate) fn checked(self) -> Result<Self> {
self.verify()?;
Ok(Self {
cid_checked: true,
..self
})
}
pub(crate) fn verify(&self) -> Result<VerifiedBlock<'_>> {
if !self.cid_checked {
validate_block_cid(&self.cid, &self.bytes)?;
}
Ok(VerifiedBlock {
cid: &self.cid,
bytes: &self.bytes,
})
}
}
/// a block whose bytes hash to its cid (sha2-256, dag-cbor). record storage reads bodies only
/// through one: events and history find bodies by cid, so a body kept under a cid it does not
/// hash to would be served as that record to every reader.
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
#[derive(Debug)]
pub(crate) struct VerifiedBlock<'a> {
cid: &'a IpldCid,
bytes: &'a Bytes,
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
impl<'a> VerifiedBlock<'a> {
pub(crate) fn cid(&self) -> &'a IpldCid {
self.cid
}
pub(crate) fn bytes(&self) -> &'a Bytes {
self.bytes
}
}
// backfill streams whole cars with WholeCar, so only tests and the benches parse one in a piece
#[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn parse_car(data: Bytes, verify_cids: bool) -> Result<ParsedCar<CarBlocks>> {
let (root, blocks_start) = car_root(&data)?;
let mut map = CarBlocks::with_capacity_and_hasher(
data.len() / EXPECTED_SECTION_BYTES,
ahash::RandomState::new(),
);
for_each_block(&data, blocks_start, verify_cids, |cid, payload| {
map.insert(cid, payload);
})?;
let blocks = Blocks {
map,
cids_checked: verify_cids,
};
Ok(ParsedCar { root, blocks })
}
pub(crate) fn parse_commit_car(data: Bytes, verify_cids: bool) -> Result<CommitCar> {
let (root, blocks_start) = car_root(&data)?;
let blocks = ordered_blocks(&data, blocks_start, verify_cids)?;
Ok(ParsedCar { root, blocks })
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub(crate) fn parse_car_blocks(data: Bytes, verify_cids: bool) -> Result<Blocks<CommitBlocks>> {
let (_, blocks_start) = car_header_bounds(&data)?;
ordered_blocks(&data, blocks_start, verify_cids)
}
fn ordered_blocks(
data: &Bytes,
blocks_start: usize,
verify_cids: bool,
) -> Result<Blocks<CommitBlocks>> {
let mut map = BTreeMap::new();
for_each_block(data, blocks_start, verify_cids, |cid, payload| {
map.insert(cid, payload);
})?;
Ok(Blocks {
map,
cids_checked: verify_cids,
})
}
fn car_root(data: &[u8]) -> Result<(IpldCid, usize)> {
let (header_start, blocks_start) = car_header_bounds(data)?;
Ok((
header_root(&data[header_start..blocks_start])?,
blocks_start,
))
}
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct CarHeader {
roots: Vec<IpldCid>,
version: u64,
}
impl CarHeader {
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub(crate) fn new_v1(roots: Vec<IpldCid>) -> Self {
Self { roots, version: 1 }
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub(crate) fn encode(&self) -> std::io::Result<Vec<u8>> {
serde_ipld_dagcbor::to_vec(self).map_err(std::io::Error::other)
}
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub(crate) struct CarWriter<W> {
header: Option<CarHeader>,
writer: W,
cid_buf: Vec<u8>,
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
impl<W: tokio::io::AsyncWrite + Unpin> CarWriter<W> {
pub(crate) fn new(header: CarHeader, writer: W) -> Self {
Self {
header: Some(header),
writer,
cid_buf: Vec::new(),
}
}
pub(crate) async fn write(&mut self, cid: IpldCid, data: impl AsRef<[u8]>) -> std::io::Result<()> {
use tokio::io::AsyncWriteExt as _;
if let Some(header) = self.header.take() {
let header_bytes = header.encode()?;
let mut varint = [0u8; 10];
let n = encode_uvarint(header_bytes.len(), &mut varint);
self.writer.write_all(&varint[..n]).await?;
self.writer.write_all(&header_bytes).await?;
}
self.cid_buf.clear();
cid.write_bytes(&mut self.cid_buf)
.map_err(std::io::Error::other)?;
let data = data.as_ref();
let mut varint = [0u8; 10];
let n = encode_uvarint(self.cid_buf.len() + data.len(), &mut varint);
self.writer.write_all(&varint[..n]).await?;
self.writer.write_all(&self.cid_buf).await?;
self.writer.write_all(data).await?;
Ok(())
}
pub(crate) async fn finish(mut self) -> std::io::Result<W> {
use tokio::io::AsyncWriteExt as _;
self.writer.flush().await?;
Ok(self.writer)
}
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
fn encode_uvarint(mut value: usize, buf: &mut [u8; 10]) -> usize {
let mut i = 0;
while value >= 0x80 {
buf[i] = (value as u8) | 0x80;
value >>= 7;
i += 1;
}
buf[i] = value as u8;
i + 1
}
fn header_root(header: &[u8]) -> Result<IpldCid> {
let header: CarHeader = serde_ipld_dagcbor::from_slice(header).into_diagnostic()?;
if header.version != 1 {
miette::bail!("Only CAR file version 1 is supported");
}
header
.roots
.first()
.copied()
.ok_or_else(|| miette::miette!("CAR file has no roots"))
}
fn car_header_bounds(data: &[u8]) -> Result<(usize, usize)> {
let mut offset = 0;
let Some(header_len) = read_uvarint(data, &mut offset)? else {
return Err(miette::miette!("empty CAR file"));
};
let header_start = offset;
let header_end = header_start
.checked_add(header_len)
.ok_or_else(|| miette::miette!("CAR header length overflow"))?;
if header_end > data.len() {
return Err(miette::miette!("truncated CAR header"));
}
Ok((header_start, header_end))
}
/// yields each section's cid and zero-copy payload in file order. with `verify_cids`, every
/// section is hashed as it is read, while its bytes are still in cache; checking a finished
/// map afterwards would walk memory in cid order.
fn for_each_block(
data: &Bytes,
mut offset: usize,
verify_cids: bool,
mut on_block: impl FnMut(IpldCid, Bytes),
) -> Result<()> {
while let Some(section_len) = read_uvarint(data, &mut offset)? {
let section_end = offset
.checked_add(section_len)
.ok_or_else(|| miette::miette!("CAR block length overflow"))?;
if section_end > data.len() {
return Err(miette::miette!("truncated CAR block"));
}
let section = data.slice(offset..section_end);
offset = section_end;
let (cid, block_start) = split_section(§ion)?;
let payload = section.slice(block_start..);
if verify_cids {
validate_block_cid(&cid, &payload)?;
}
on_block(cid, payload);
}
Ok(())
}
/// a section's cid, and where its payload starts
fn split_section(section: &[u8]) -> Result<(IpldCid, usize)> {
let mut cursor = Cursor::new(section);
let cid = IpldCid::read_bytes(&mut cursor).into_diagnostic()?;
let block_start = cursor.position() as usize;
if block_start >= section.len() {
return Err(miette::miette!("CAR block has no payload for {cid}"));
}
Ok((cid, block_start))
}
/// a car read as its chunks arrive, so a caller can copy out the blocks it wants without the
/// whole file ever being in memory as one piece. besides what the caller copies out it holds
/// one unfinished section at most. blocks come out in file order, hashed against their cids when
/// `verify_cids` is set, the same as [`parse_car`].
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub(crate) struct CarStream {
pending: BytesMut,
root: Option<IpldCid>,
verify_cids: bool,
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
impl CarStream {
pub(crate) fn new(verify_cids: bool) -> Self {
Self {
pending: BytesMut::new(),
root: None,
verify_cids,
}
}
/// adds the next chunk and hands each block it completes to `on_block`, with the car's
/// root, since the header always comes first
pub(crate) fn push(
&mut self,
chunk: &[u8],
mut on_block: impl FnMut(&IpldCid, IpldCid, &[u8]) -> Result<()>,
) -> Result<()> {
self.pending.extend_from_slice(chunk);
let mut consumed = 0;
while let Some((len, len_bytes)) = try_read_uvarint(&self.pending[consumed..])? {
if len > MAX_STREAMED_SECTION_BYTES {
miette::bail!("CAR section of {len} bytes is over {MAX_STREAMED_SECTION_BYTES}");
}
let start = consumed + len_bytes;
let end = start
.checked_add(len)
.ok_or_else(|| miette::miette!("CAR section length overflow"))?;
if end > self.pending.len() {
break;
}
let section = &self.pending[start..end];
match self.root {
None => self.root = Some(header_root(section)?),
Some(root) => {
let (cid, block_start) = split_section(section)?;
let payload = §ion[block_start..];
if self.verify_cids {
validate_block_cid(&cid, payload)?;
}
on_block(&root, cid, payload)?;
}
}
consumed = end;
}
self.pending.advance(consumed);
Ok(())
}
/// hashes every block from here on against its cid, whatever the car was made with
pub(crate) fn check_cids(&mut self) {
self.verify_cids = true;
}
/// ends the car, with `kept` holding the blocks the caller copied out of it. a car that
/// stops partway through its header or a block is truncated, not just short
pub(crate) fn finish_with<M>(self, kept: M) -> Result<ParsedCar<M>> {
let root = match (self.root, self.pending.is_empty()) {
(None, true) => return Err(miette::miette!("empty CAR file")),
(None, false) => return Err(miette::miette!("truncated CAR header")),
(Some(_), false) => return Err(miette::miette!("truncated CAR block")),
(Some(root), true) => root,
};
Ok(ParsedCar {
root,
blocks: Blocks {
map: kept,
cids_checked: self.verify_cids,
},
})
}
}
/// a whole car read as its chunks arrive, every block copied out on its own. the body is never
/// held in one piece, and blocks the mst walk leaves behind are freed with the map instead of
/// pinning the body until the last record is stored
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub(crate) struct WholeCar {
stream: CarStream,
blocks: CarBlocks,
}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
impl WholeCar {
pub(crate) fn new(verify_cids: bool) -> Self {
Self {
stream: CarStream::new(verify_cids),
blocks: CarBlocks::default(),
}
}
/// makes room for a car of about `car_bytes`, so the map isn't rehashed as it fills
pub(crate) fn reserve(&mut self, car_bytes: usize) {
self.blocks.reserve(car_bytes / EXPECTED_SECTION_BYTES);
}
pub(crate) fn push(&mut self, chunk: &[u8]) -> Result<()> {
let blocks = &mut self.blocks;
self.stream.push(chunk, |_, cid, bytes| {
blocks.insert(cid, Bytes::copy_from_slice(bytes));
Ok(())
})
}
pub(crate) fn finish(self) -> Result<ParsedCar<CarBlocks>> {
self.stream.finish_with(self.blocks)
}
/// the stream partway through the car, and the blocks it has read so far
pub(crate) fn into_parts(self) -> (CarStream, CarBlocks) {
(self.stream, self.blocks)
}
}
/// checks that a block's payload hashes to its claimed CID (sha2-256, dag-cbor).
pub(crate) fn validate_block_cid(claimed_cid: &IpldCid, bytes: &[u8]) -> Result<()> {
let computed_cid = compute_cid(bytes);
if computed_cid != *claimed_cid {
return Err(miette::miette!(
"CAR block CID mismatch: claimed {claimed_cid}, computed {computed_cid}"
));
}
Ok(())
}
/// a uvarint at the start of `data` and how many bytes it took, or `None` while `data` ends
/// before it does
fn try_read_uvarint(data: &[u8]) -> Result<Option<(usize, usize)>> {
let mut value = 0u64;
for (idx, byte) in data.iter().take(10).enumerate() {
value |= u64::from(byte & 0x7f) << (7 * idx);
if byte & 0x80 == 0 {
return Ok(Some((usize::try_from(value).into_diagnostic()?, idx + 1)));
}
}
if data.len() >= 10 {
return Err(miette::miette!("uvarint overflow in CAR file"));
}
Ok(None)
}
fn read_uvarint(data: &[u8], offset: &mut usize) -> Result<Option<usize>> {
let rest = &data[(*offset).min(data.len())..];
if rest.is_empty() {
return Ok(None);
}
let (value, len) =
try_read_uvarint(rest)?.ok_or_else(|| miette::miette!("truncated uvarint in CAR file"))?;
*offset += len;
Ok(Some(value))
}
#[cfg(test)]
mod tests {
use super::*;
use cid::multihash::Multihash;
use jacquard_common::types::crypto::{DAG_CBOR, SHA2_256};
fn cid(byte: u8) -> IpldCid {
let hash = [byte; 32];
let mh = Multihash::<64>::wrap(SHA2_256, &hash).unwrap();
IpldCid::new_v1(DAG_CBOR, mh)
}
#[tokio::test]
async fn parses_rootless_block_car() {
let cid = cid(1);
let mut buf = Vec::new();
let header = CarHeader::new_v1(Vec::new());
let mut writer = CarWriter::new(header, &mut buf);
writer.write(cid, b"block".to_vec()).await.unwrap();
writer.finish().await.unwrap();
let blocks = parse_car_blocks(buf.into(), false).unwrap();
assert_eq!(blocks.get(&cid).unwrap().as_ref(), b"block");
}
#[tokio::test]
async fn parses_rooted_car_with_zero_copy_payloads() {
let root = cid(1);
let block = cid(2);
let mut buf = Vec::new();
let header = CarHeader::new_v1(vec![root]);
let mut writer = CarWriter::new(header, &mut buf);
writer.write(block, b"record".to_vec()).await.unwrap();
writer.finish().await.unwrap();
let data = Bytes::from(buf);
let data_range = data.as_ptr() as usize..data.as_ptr() as usize + data.len();
let parsed = parse_car(data, false).unwrap();
let payload = parsed.blocks.get(&block).unwrap();
let payload_range = payload.as_ptr() as usize..payload.as_ptr() as usize + payload.len();
assert_eq!(parsed.root, root);
assert_eq!(payload.as_ref(), b"record");
assert!(data_range.start <= payload_range.start);
assert!(payload_range.end <= data_range.end);
}
async fn car_with_blocks(blocks: &[(IpldCid, &[u8])]) -> Bytes {
let mut buf = Vec::new();
let header = CarHeader::new_v1(vec![blocks[0].0]);
let mut writer = CarWriter::new(header, &mut buf);
for (cid, bytes) in blocks {
writer.write(*cid, bytes.to_vec()).await.unwrap();
}
writer.finish().await.unwrap();
buf.into()
}
/// a rooted car whose blocks are big enough for some sections to need a two-byte length
async fn stream_fixture() -> (IpldCid, Vec<(IpldCid, Vec<u8>)>, Bytes) {
let blocks = [
b"a".to_vec(),
vec![7u8; 300],
b"record".to_vec(),
vec![1u8; 129],
]
.into_iter()
.map(|bytes| (compute_cid(&bytes), bytes))
.collect::<Vec<_>>();
let root = blocks[0].0;
let mut buf = Vec::new();
let mut writer = CarWriter::new(CarHeader::new_v1(vec![root]), &mut buf);
for (cid, bytes) in &blocks {
writer.write(*cid, bytes).await.unwrap();
}
writer.finish().await.unwrap();
(root, blocks, buf.into())
}
fn stream_in_chunks(
car: &[u8],
chunk: usize,
verify_cids: bool,
) -> Result<(CommitCar, Vec<IpldCid>)> {
let mut stream = CarStream::new(verify_cids);
let mut kept = CommitBlocks::new();
let mut order = Vec::new();
for piece in car.chunks(chunk) {
stream.push(piece, |_, cid, bytes| {
order.push(cid);
kept.insert(cid, Bytes::copy_from_slice(bytes));
Ok(())
})?;
}
Ok((stream.finish_with(kept)?, order))
}
#[tokio::test]
async fn stream_yields_what_a_whole_parse_does_however_the_car_is_chunked() {
let (root, blocks, car) = stream_fixture().await;
let whole = parse_car(car.clone(), true).unwrap();
for chunk in [1, 2, 3, 7, 64, 333, car.len()] {
let (streamed, order) = stream_in_chunks(&car, chunk, true).unwrap();
assert_eq!(streamed.root, root, "chunk {chunk}");
assert_eq!(
order,
blocks.iter().map(|(cid, _)| *cid).collect::<Vec<_>>(),
"chunk {chunk}"
);
for (cid, bytes) in &blocks {
assert_eq!(
streamed.blocks.get(cid).unwrap().as_ref(),
bytes.as_slice(),
"chunk {chunk}"
);
assert_eq!(
whole.blocks.get(cid),
streamed.blocks.get(cid),
"chunk {chunk}"
);
}
assert!(streamed.blocks.block(&root).unwrap().verify().is_ok());
}
}
#[tokio::test]
async fn a_stream_turns_a_huge_block_away_before_it_arrives() {
let (_, _, car) = stream_fixture().await;
let header_len = {
let mut offset = 0;
let len = read_uvarint(&car, &mut offset).unwrap().unwrap();
offset + len
};
// the header, then a section that only says how long it'll be
let mut start = car[..header_len].to_vec();
let mut len = MAX_STREAMED_SECTION_BYTES + 1;
while len >= 0x80 {
start.push((len as u8 & 0x7f) | 0x80);
len >>= 7;
}
start.push(len as u8);
let mut stream = CarStream::new(true);
let err = stream.push(&start, |_, _, _| Ok(())).unwrap_err();
assert!(err.to_string().contains("CAR section of"), "{err}");
}
#[tokio::test]
async fn whole_car_keeps_every_block_a_whole_parse_does() {
let (root, _, car) = stream_fixture().await;
let parsed = parse_car(car.clone(), true).unwrap();
for chunk in [1, 3, 64, car.len()] {
let mut whole = WholeCar::new(true);
whole.reserve(car.len());
for piece in car.chunks(chunk) {
whole.push(piece).unwrap();
}
let streamed = whole.finish().unwrap();
assert_eq!(streamed.root, root, "chunk {chunk}");
assert_eq!(streamed.blocks.map(), parsed.blocks.map(), "chunk {chunk}");
assert!(streamed.blocks.block(&root).unwrap().cid_checked);
}
let real = compute_cid(b"trusted");
let forged = car_with_blocks(&[(real, b"trusted"), (cid(1), b"forged")]).await;
let mut whole = WholeCar::new(true);
let err = whole.push(&forged).unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
}
#[tokio::test]
async fn stream_cut_short_is_truncated() {
let (_, blocks, car) = stream_fixture().await;
let err = stream_in_chunks(&[], 1, true).unwrap_err();
assert!(err.to_string().contains("empty CAR file"), "{err}");
// cutting exactly between sections leaves a complete, shorter car
let mut boundaries = Vec::new();
let mut offset = 0;
while let Some(len) = read_uvarint(&car, &mut offset).unwrap() {
offset += len;
boundaries.push(offset);
}
assert_eq!(boundaries.len(), blocks.len() + 1);
for cut in 1..car.len() {
let result = stream_in_chunks(&car[..cut], 5, true);
if let Some(sections) = boundaries.iter().position(|&end| end == cut) {
let (streamed, order) = result.unwrap();
assert_eq!(order.len(), sections, "cut at {cut}");
assert_eq!(streamed.blocks.map().len(), sections, "cut at {cut}");
continue;
}
let err = result
.err()
.unwrap_or_else(|| panic!("a car cut at {cut} parsed"));
let want = if cut < boundaries[0] {
"truncated CAR header"
} else {
"truncated CAR block"
};
assert!(err.to_string().contains(want), "cut at {cut}: {err}");
}
}
#[tokio::test]
async fn stream_rejects_a_forged_block_when_verifying() {
let real = compute_cid(b"trusted");
let car = car_with_blocks(&[(real, b"trusted"), (cid(1), b"forged")]).await;
let err = stream_in_chunks(&car, 3, true).unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
let (unverified, _) = stream_in_chunks(&car, 3, false).unwrap();
assert_eq!(unverified.blocks.get(&cid(1)).unwrap().as_ref(), b"forged");
assert!(unverified.blocks.block(&cid(1)).unwrap().verify().is_err());
}
#[tokio::test]
async fn stream_rejects_a_rootless_car() {
let mut buf = Vec::new();
let mut writer = CarWriter::new(CarHeader::new_v1(Vec::new()), &mut buf);
writer.write(cid(1), b"block".to_vec()).await.unwrap();
writer.finish().await.unwrap();
let streamed = stream_in_chunks(&buf, 4, false).unwrap_err();
let whole = parse_car(buf.into(), false).unwrap_err();
assert_eq!(streamed.to_string(), whole.to_string());
}
#[test]
fn partial_uvarint_waits_for_more_bytes() {
assert_eq!(try_read_uvarint(&[]).unwrap(), None);
assert_eq!(try_read_uvarint(&[0x80]).unwrap(), None);
assert_eq!(try_read_uvarint(&[0xac, 0x02]).unwrap(), Some((300, 2)));
assert_eq!(try_read_uvarint(&[0x05, 0xff]).unwrap(), Some((5, 1)));
assert!(try_read_uvarint(&[0x80; 10]).is_err());
}
#[tokio::test]
async fn verifying_parse_rejects_forged_block() {
let real = compute_cid(b"trusted");
let car = car_with_blocks(&[(real, b"trusted"), (cid(1), b"forged")]).await;
assert!(parse_car(car.clone(), false).is_ok());
let err = parse_car(car.clone(), true).unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"));
let err = parse_commit_car(car.clone(), true).unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"));
let err = parse_car_blocks(car, true).unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"));
}
#[tokio::test]
async fn verifying_parse_accepts_matching_blocks() {
let root = compute_cid(b"root");
let leaf = compute_cid(b"leaf");
let car = car_with_blocks(&[(root, b"root"), (leaf, b"leaf")]).await;
let parsed = parse_car(car.clone(), true).unwrap();
assert_eq!(parsed.root, root);
assert_eq!(parsed.blocks.get(&leaf).unwrap().as_ref(), b"leaf");
let commit = parse_commit_car(car, true).unwrap();
assert_eq!(commit.root, root);
assert_eq!(commit.blocks.get(&leaf).unwrap().as_ref(), b"leaf");
}
#[tokio::test]
async fn verifying_parse_rejects_forged_copy_shadowed_by_valid_duplicate() {
// the map keeps only the last copy, so the forged one is caught only while parsing;
// no honest pds sends it, so the whole car is rejected
let real = compute_cid(b"trusted");
let car = car_with_blocks(&[(real, b"forged"), (real, b"trusted")]).await;
let err = parse_car(car.clone(), true).unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"));
let err = parse_commit_car(car, true).unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"));
}
#[tokio::test]
async fn every_parse_keeps_the_last_duplicate_copy() {
let car = car_with_blocks(&[(cid(1), b"first"), (cid(1), b"second")]).await;
let repo = parse_car(car.clone(), false).unwrap();
assert_eq!(repo.blocks.get(&cid(1)).unwrap().as_ref(), b"second");
let commit = parse_commit_car(car.clone(), false).unwrap();
assert_eq!(commit.blocks.get(&cid(1)).unwrap().as_ref(), b"second");
let blocks = parse_car_blocks(car, false).unwrap();
assert_eq!(blocks.get(&cid(1)).unwrap().as_ref(), b"second");
}
// iroh-car's reader, behind jacquard's `parse_car_bytes`, accepts each of the next three
#[tokio::test]
async fn rejects_a_trailing_partial_varint() {
// iroh-car takes a varint cut off by the end of the car for a clean end
let mut car = car_with_blocks(&[(cid(1), b"block")]).await.to_vec();
car.push(0x80);
let err = parse_commit_car(car.into(), false).unwrap_err();
assert!(err.to_string().contains("truncated uvarint"));
}
#[tokio::test]
async fn rejects_a_section_longer_than_the_car() {
// iroh-car zero-fills a buffer of the claimed length before reading into it
let mut car = car_with_blocks(&[(cid(1), b"block")]).await.to_vec();
car.extend_from_slice(&[0xff, 0xff, 0xff, 0x01, 0x00]);
let err = parse_commit_car(car.into(), false).unwrap_err();
assert!(err.to_string().contains("truncated CAR block"));
}
#[tokio::test]
async fn rejects_a_block_without_payload() {
// iroh-car accepts it, and the cid of empty bytes even passes verification
let empty = compute_cid(b"");
let car = car_with_blocks(&[(empty, b"")]).await;
let err = parse_commit_car(car, true).unwrap_err();
assert!(err.to_string().contains("has no payload"));
}
fn forged(claimed: IpldCid) -> Blocks<CommitBlocks> {
Blocks::unverified(BTreeMap::from([(claimed, Bytes::from_static(b"forged"))]))
}
#[test]
fn unverified_blocks_are_hashed_when_verified() {
let block = forged(cid(1)).block(&cid(1)).unwrap();
let err = block.verify().unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"));
let real = compute_cid(b"forged");
let block = forged(real).block(&real).unwrap();
let verified = block.verify().unwrap();
assert_eq!(verified.cid(), &real);
assert_eq!(verified.bytes().as_ref(), b"forged");
}
#[test]
fn checking_a_block_hashes_it_once() {
let block = forged(cid(1)).block(&cid(1)).unwrap();
let err = block.checked().unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"));
let real = compute_cid(b"forged");
let checked = forged(real).block(&real).unwrap().checked().unwrap();
assert!(checked.cid_checked);
assert_eq!(checked.verify().unwrap().bytes().as_ref(), b"forged");
}
#[tokio::test]
async fn blocks_from_a_non_verifying_parse_are_hashed_when_verified() {
let car = car_with_blocks(&[(cid(1), b"forged")]).await;
let repo = parse_car(car.clone(), false).unwrap();
assert!(repo.blocks.block(&cid(1)).unwrap().verify().is_err());
let commit = parse_commit_car(car.clone(), false).unwrap();
assert!(commit.blocks.block(&cid(1)).unwrap().verify().is_err());
let blocks = parse_car_blocks(car, false).unwrap();
assert!(blocks.block(&cid(1)).unwrap().verify().is_err());
}
#[tokio::test]
async fn blocks_from_a_verifying_parse_verify_as_parsed() {
let leaf = compute_cid(b"leaf");
let car = car_with_blocks(&[(leaf, b"leaf")]).await;
let parsed = parse_commit_car(car, true).unwrap();
let block = parsed.blocks.block(&leaf).unwrap();
let verified = block.verify().unwrap();
assert_eq!(verified.cid(), &leaf);
assert_eq!(verified.bytes().as_ref(), b"leaf");
}
#[tokio::test]
async fn merged_blocks_stay_checked_only_if_both_sides_were() {
let leaf = compute_cid(b"leaf");
let car = car_with_blocks(&[(leaf, b"leaf")]).await;
let mut checked = parse_car_blocks(car.clone(), true).unwrap();
checked.extend(forged(cid(1)));
assert!(checked.block(&cid(1)).unwrap().verify().is_err());
let mut unchecked = forged(cid(1));
unchecked.extend(parse_car_blocks(car, true).unwrap());
assert!(unchecked.block(&cid(1)).unwrap().verify().is_err());
let mut empty = Blocks::default();
empty.extend(forged(cid(1)));
assert!(empty.block(&cid(1)).unwrap().verify().is_err());
}
#[tokio::test]
async fn an_empty_default_keeps_merged_blocks_checked() {
// sparse backfill folds verified fetches into a default, so a default
// counted as unchecked would hash every one of them again
let leaf = compute_cid(b"leaf");
let car = car_with_blocks(&[(leaf, b"leaf")]).await;
let mut merged = Blocks::default();
merged.extend(parse_car_blocks(car, true).unwrap());
let block = merged.block(&leaf).unwrap();
assert!(block.cid_checked);
assert_eq!(block.cid(), &leaf);
}
}