use super::limits::{
check_manifest_json_depth, classify_member_ceiling, classify_source_ceiling,
ReplayContractError, ReplayIngestError, ReplayLimits,
};
use super::{paths, BundleEntry, ReadBundle};
use crate::replay::manifest::ReplayManifest;
use anyhow::{Context, Result};
use assay_common::limits::{LimitKind, LimitReader};
use flate2::read::GzDecoder;
use flate2::Compression;
use flate2::GzBuilder;
use std::collections::BTreeMap;
use std::io::{Read, Write};
use tar::{Archive, Builder, Header};
pub fn write_bundle_tar_gz<W: Write>(
w: W,
manifest: &ReplayManifest,
entries: &[BundleEntry],
) -> Result<()> {
let manifest_json = serde_json::to_vec(manifest).context("serialize manifest")?;
let gz = GzBuilder::new().mtime(0).write(w, Compression::default());
let mut tar = Builder::new(gz);
tar.mode(tar::HeaderMode::Deterministic);
write_tar_entry(&mut tar, paths::MANIFEST, &manifest_json)?;
let mut sorted: Vec<_> = entries.iter().collect();
sorted.sort_by(|a, b| a.path.as_str().cmp(b.path.as_str()));
for e in &sorted {
normalize_path_and_append(&mut tar, &e.path, &e.data)?;
}
let gz = tar.into_inner().context("finalize tar")?;
gz.finish().context("finish gzip")?;
Ok(())
}
fn write_tar_entry<T: Write>(tar: &mut Builder<T>, path: &str, data: &[u8]) -> Result<()> {
let mut header = Header::new_gnu();
header.set_path(path).context("set_path")?;
header.set_size(data.len() as u64);
header.set_mode(0o644);
header.set_uid(0);
header.set_gid(0);
header.set_mtime(0);
header.set_cksum();
tar.append(&header, data).context("append entry")?;
Ok(())
}
fn normalize_path_and_append<T: Write>(
tar: &mut Builder<T>,
path: &str,
data: &[u8],
) -> Result<()> {
let normalized = paths::validate_entry_path(path)?;
write_tar_entry(tar, &normalized, data)
}
pub fn read_bundle_tar_gz<R: Read>(r: R) -> Result<ReadBundle> {
read_bundle_tar_gz_with_limits(r, ReplayLimits::default())
}
pub fn read_bundle_tar_gz_with_limits<R: Read>(r: R, limits: ReplayLimits) -> Result<ReadBundle> {
let mut snapshot = Vec::new();
LimitReader::new(r, limits.max_source_bytes, LimitKind::SourceBytes)
.read_to_end(&mut snapshot)
.map_err(|err| {
classify_source_ceiling(&err)
.map(anyhow::Error::from)
.unwrap_or_else(|| anyhow::Error::from(err).context("read bundle source"))
})?;
let decoder = GzDecoder::new(std::io::Cursor::new(&snapshot));
let bounded_decoder =
LimitReader::new(decoder, limits.max_decoded_bytes, LimitKind::DecodedBytes);
let mut ar = Archive::new(bounded_decoder);
let mut manifest_data: Option<Vec<u8>> = None;
let mut seen: BTreeMap<String, Vec<u8>> = BTreeMap::new();
let mut entry_count: usize = 0;
let entries = ar.entries().map_err(|e| {
classify_source_ceiling(&e)
.map(anyhow::Error::from)
.unwrap_or_else(|| anyhow::Error::from(e).context("list tar entries"))
})?;
for entry in entries {
let mut e = entry.map_err(|err| {
classify_source_ceiling(&err)
.map(anyhow::Error::from)
.unwrap_or_else(|| anyhow::Error::from(err).context("read tar entry"))
})?;
entry_count += 1;
if entry_count > limits.max_entries {
return Err(anyhow::Error::from(ReplayIngestError::TooManyEntries {
limit: limits.max_entries,
}));
}
let raw_path_len = e.path_bytes().len();
if raw_path_len > limits.max_path_len {
return Err(anyhow::Error::from(ReplayIngestError::PathTooLong {
limit: limits.max_path_len,
}));
}
let path = e.path().context("entry path")?;
let path_str = path.to_string_lossy().replace('\\', "/");
if path_str == paths::MANIFEST {
if manifest_data.is_some() {
return Err(anyhow::Error::from(ReplayContractError::DuplicateManifest));
}
let mut data = Vec::new();
let mut bounded =
LimitReader::new(&mut e, limits.max_manifest_bytes, LimitKind::MemberBytes);
bounded.read_to_end(&mut data).map_err(|err| {
classify_member_ceiling(&err)
.map(anyhow::Error::from)
.unwrap_or_else(|| anyhow::Error::from(err).context("read manifest body"))
})?;
check_manifest_json_depth(&data, limits.max_manifest_json_depth)?;
manifest_data = Some(data);
continue;
}
let canonical = paths::validate_entry_path(&path_str)?;
let mut data = Vec::new();
let mut bounded = LimitReader::new(&mut e, limits.max_member_bytes, LimitKind::MemberBytes);
bounded.read_to_end(&mut data).map_err(|err| {
classify_member_ceiling(&err)
.map(anyhow::Error::from)
.unwrap_or_else(|| anyhow::Error::from(err).context("read entry body"))
})?;
if seen.insert(canonical, data).is_some() {
return Err(anyhow::Error::from(ReplayContractError::DuplicatePath));
}
}
let manifest_json = manifest_data.context("manifest.json missing in bundle")?;
let manifest: ReplayManifest =
serde_json::from_slice(&manifest_json).context("parse manifest.json")?;
let entries = seen.into_iter().collect();
Ok(ReadBundle { manifest, entries })
}