use std::{
collections::{BTreeMap, BTreeSet},
io::{Read, Seek, Write},
};
use super::{ObjectType, PackObjectId, PackReader, PackStats, StreamingPackBuilder, SyncData};
use crate::{
object::{
ContentHash, EntryRedactions, ObjectSource, PartialTree, State, Tree, TreeEntryTarget,
},
store::{Result, StoreError},
};
pub(super) fn validate(
reader: &PackReader<'_>,
selected: &State,
max_decoded_bytes: u64,
references: &[crate::object::source_target::capture::ReferenceProof],
visibility: Option<&crate::object::thread_replication::CaptureVisibility>,
) -> Result<Vec<PackObjectId>> {
Ok(validate_disclosure(
reader,
selected,
max_decoded_bytes,
references,
visibility,
false,
)?
.objects)
}
pub struct VisibleSourceClosure {
pub objects: Vec<PackObjectId>,
pub partial_trees: Vec<PartialTree>,
}
pub(super) fn validate_disclosure(
reader: &PackReader<'_>,
selected: &State,
max_decoded_bytes: u64,
references: &[crate::object::source_target::capture::ReferenceProof],
visibility: Option<&crate::object::thread_replication::CaptureVisibility>,
allow_partial: bool,
) -> Result<VisibleSourceClosure> {
let canonical = selected.encode_current_msgpack()?;
let mut available = BTreeMap::new();
let mut trees = BTreeMap::new();
let mut partial_trees = Vec::new();
let mut decoded = 0_u64;
reader.visit_objects(|id, kind, data| {
decoded = decoded
.checked_add(data.len() as u64)
.ok_or_else(|| invalid("source pack size overflow"))?;
if decoded > max_decoded_bytes {
return Err(invalid("source pack decoded byte budget exceeded"));
}
if available.insert(id, kind).is_some() {
return Err(invalid("duplicate source pack object"));
}
match (id, kind) {
(PackObjectId::StateId(id), ObjectType::State)
if id == selected.id() && data == canonical => {}
(PackObjectId::Hash(hash), ObjectType::Blob)
if ContentHash::compute_typed("blob", data) == hash => {}
(PackObjectId::Hash(hash), ObjectType::Tree) => {
if allow_partial && crate::object::is_redacted_tree(data) {
let partial = crate::object::decode_redacted_projection(data)?;
partial.verify()?;
if partial.declared_root() != hash || !partial.has_redactions() {
return Err(invalid(
"partial source tree differs from its address or is complete",
));
}
trees.insert(hash, partial.visible_tree()?);
partial_trees.push(partial);
return Ok(());
}
let tree = Tree::decode_canonical(data)
.map_err(|_| invalid("source pack requires complete canonical tree anchors"))?;
if tree.hash() != hash {
return Err(invalid("source tree hash differs from its address"));
}
trees.insert(hash, tree);
}
_ => {
return Err(invalid(
"source pack contains an unselected or incorrectly addressed object",
));
}
}
Ok(())
})?;
let mut visited = BTreeSet::new();
let mut pending = vec![
(PackObjectId::StateId(selected.id()), ObjectType::State),
(PackObjectId::Hash(selected.tree), ObjectType::Tree),
];
while let Some((id, kind)) = pending.pop() {
if available.get(&id) != Some(&kind) {
return Err(invalid("source closure is incomplete"));
}
if !visited.insert(id) {
continue;
}
if kind == ObjectType::Tree {
let PackObjectId::Hash(hash) = id else {
return Err(invalid("tree address is not a content hash"));
};
let tree = trees
.get(&hash)
.ok_or_else(|| invalid("source tree unavailable"))?;
for (hash, kind) in children(tree) {
pending.push((PackObjectId::Hash(hash), kind));
}
}
}
if let Some(visibility) = visibility {
visibility.validate(selected)?;
let mut leaves = BTreeMap::<ContentHash, BTreeSet<ContentHash>>::new();
for entry in &visibility.entries {
if !visited.contains(&PackObjectId::Hash(entry.tree_id)) {
return Err(invalid("entry visibility tree is outside selected source"));
}
let tree = trees
.get(&entry.tree_id)
.ok_or_else(|| invalid("entry visibility subject is not a tree"))?;
let actual = leaves.entry(entry.tree_id).or_insert_with(|| {
(0..tree.entries().len())
.filter_map(|index| tree.v4_leaf_hash_at(index))
.collect()
});
if !actual.contains(&entry.leaf_hash) {
return Err(invalid(
"entry visibility leaf is absent from selected salted tree",
));
}
}
}
for reference in references {
let closure = crate::object::source_target::capture::closure(
&ReferenceReader(reader),
reference.descriptor,
&reference.scope,
reference.state,
)?;
for hash in closure.blobs.keys() {
let id = PackObjectId::Hash(*hash);
if available.get(&id) != Some(&ObjectType::Blob) {
return Err(invalid("reference closure is incomplete"));
}
visited.insert(id);
}
}
if visited.len() != available.len() {
return Err(invalid(
"source pack contains objects outside the selected revision",
));
}
Ok(VisibleSourceClosure {
objects: visited.into_iter().collect(),
partial_trees,
})
}
fn invalid(message: &str) -> StoreError {
StoreError::InvalidObject(message.into())
}
pub fn build_source_pack<W: Write + Read + Seek + SyncData>(
builder: StreamingPackBuilder<W>,
source: &impl ObjectSource,
selected: &State,
max_objects: usize,
max_decoded_bytes: u64,
) -> Result<(W, PackStats)> {
build_source_pack_with_references(
builder,
source,
selected,
&[],
max_objects,
max_decoded_bytes,
)
}
pub fn build_source_pack_with_references<W: Write + Read + Seek + SyncData>(
builder: StreamingPackBuilder<W>,
source: &impl ObjectSource,
selected: &State,
references: &[crate::object::source_target::capture::ReferenceProof],
max_objects: usize,
max_decoded_bytes: u64,
) -> Result<(W, PackStats)> {
build_disclosure(
builder,
source,
selected,
references,
None,
max_objects,
max_decoded_bytes,
)
.map(|(output, stats, _)| (output, stats))
}
pub fn build_visible_source_pack<W: Write + Read + Seek + SyncData>(
builder: StreamingPackBuilder<W>,
source: &impl ObjectSource,
selected: &State,
references: &[crate::object::source_target::capture::ReferenceProof],
redactions: &EntryRedactions,
max_objects: usize,
max_decoded_bytes: u64,
) -> Result<(W, PackStats, bool)> {
build_disclosure(
builder,
source,
selected,
references,
Some(redactions),
max_objects,
max_decoded_bytes,
)
}
fn build_disclosure<W: Write + Read + Seek + SyncData>(
mut builder: StreamingPackBuilder<W>,
source: &impl ObjectSource,
selected: &State,
references: &[crate::object::source_target::capture::ReferenceProof],
redactions: Option<&EntryRedactions>,
max_objects: usize,
max_decoded_bytes: u64,
) -> Result<(W, PackStats, bool)> {
if max_objects < 2 {
return Err(invalid("source pack object budget exceeded"));
}
let canonical = selected.encode_current_msgpack()?;
let mut decoded = 0_u64;
charge_bytes(&mut decoded, canonical.len() as u64, max_decoded_bytes)?;
builder.add_id(
PackObjectId::StateId(selected.id()),
ObjectType::State,
&canonical,
)?;
let mut discovered = BTreeMap::from([(selected.tree, ObjectType::Tree)]);
let mut partial = false;
let mut pending = discovered.clone();
while let Some((hash, kind)) = pending.pop_first() {
match kind {
ObjectType::Tree => {
let tree = source
.get_tree(&hash)?
.ok_or_else(|| invalid("selected source tree is missing"))?;
if tree.hash() != hash {
return Err(invalid("source tree differs from its address"));
}
let full = tree.encode_canonical()?;
charge_bytes(&mut decoded, full.len() as u64, max_decoded_bytes)?;
let (canonical, visible_tree) = match redactions {
Some(redactions)
if (0..tree.entries().len())
.any(|index| !redactions.entry_visible(&tree, index)) =>
{
partial = true;
let partial = PartialTree::project(&tree, redactions.leaves())?;
(
crate::object::encode_redacted_projection(&partial)?,
partial.visible_tree()?,
)
}
_ => (full, tree),
};
for (hash, kind) in children(&visible_tree) {
if let Some(expected) = discovered.get(&hash) {
if *expected != kind {
return Err(invalid(
"source object is referenced with conflicting types",
));
}
continue;
}
if discovered.len().saturating_add(1) >= max_objects {
return Err(invalid("source pack object budget exceeded"));
}
discovered.insert(hash, kind);
pending.insert(hash, kind);
}
builder.add_id(PackObjectId::Hash(hash), ObjectType::Tree, &canonical)?;
}
ObjectType::Blob => {
let length = source
.decoded_blob_len(&hash)?
.ok_or_else(|| invalid("selected source blob is missing"))?;
if length > max_decoded_bytes.saturating_sub(decoded) {
return Err(invalid("source pack decoded byte budget exceeded"));
}
let bytes = source
.get_blob_bytes(&hash)?
.ok_or_else(|| invalid("selected source blob is missing"))?;
if bytes.len() as u64 != length
|| ContentHash::compute_typed("blob", &bytes) != hash
{
return Err(invalid(
"source blob differs from its address or declared size",
));
}
charge_bytes(&mut decoded, length, max_decoded_bytes)?;
builder.add_id(PackObjectId::Hash(hash), ObjectType::Blob, bytes)?;
}
_ => return Err(invalid("unexpected source object type")),
}
}
if !partial {
for reference in references {
let closure = crate::object::source_target::capture::closure(
source,
reference.descriptor,
&reference.scope,
reference.state,
)?;
for (hash, bytes) in closure.blobs {
if let Some(kind) = discovered.get(&hash) {
if *kind != ObjectType::Blob {
return Err(invalid("reference object type conflict"));
}
continue;
}
if discovered.len().saturating_add(1) >= max_objects {
return Err(invalid("reference pack object budget exceeded"));
}
charge_bytes(&mut decoded, bytes.len() as u64, max_decoded_bytes)?;
discovered.insert(hash, ObjectType::Blob);
builder.add_id(PackObjectId::Hash(hash), ObjectType::Blob, bytes)?;
}
}
}
let (output, stats) = builder.finalize()?;
Ok((output, stats, !partial))
}
fn charge_bytes(decoded: &mut u64, length: u64, limit: u64) -> Result<()> {
*decoded = decoded
.checked_add(length)
.ok_or_else(|| invalid("source pack size overflow"))?;
if *decoded > limit {
return Err(invalid("source pack decoded byte budget exceeded"));
}
Ok(())
}
fn children(tree: &Tree) -> impl Iterator<Item = (ContentHash, ObjectType)> + '_ {
tree.entries()
.iter()
.filter_map(|entry| match entry.target() {
TreeEntryTarget::Tree { hash } => Some((*hash, ObjectType::Tree)),
TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
Some((*hash, ObjectType::Blob))
}
TreeEntryTarget::Gitlink { .. } | TreeEntryTarget::Spoollink { .. } => None,
})
}
struct ReferenceReader<'a, 'b>(&'a PackReader<'b>);
impl ObjectSource for ReferenceReader<'_, '_> {
fn get_tree(&self, _: &ContentHash) -> Result<Option<Tree>> {
Err(invalid("reference closure must not read source trees"))
}
fn get_state(&self, _: &crate::object::StateId) -> Result<Option<State>> {
Err(invalid("reference closure must not read source history"))
}
fn get_blob(&self, hash: &ContentHash) -> Result<Option<crate::object::Blob>> {
match self.0.get_hashed_object(hash)? {
Some((ObjectType::Blob, bytes)) => Ok(Some(crate::object::Blob::new(bytes))),
None => Ok(None),
_ => Err(invalid("reference object must be a blob")),
}
}
fn decoded_blob_len(&self, hash: &ContentHash) -> Result<Option<u64>> {
self.0.get_hashed_object_size(hash)
}
}