use crypto::{Ed25519Signer, Signer};
use objects::object::{StateId, thread_replication::ThreadGenesis};
use super::*;
#[test]
fn partial_fetch_requires_explicit_selection_and_matching_completion() {
let (mut open, mut ready, endpoint, _) = fixture();
ready.full_closure_available = false;
assert!(
Validation::new(
open.clone(),
ready.clone(),
Some(&endpoint),
Limits::default()
)
.is_err(),
"an endpoint cannot silently weaken a complete request"
);
open.selection.as_mut().expect("selection").allow_partial = true;
let mut validation = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
.expect("explicit partial request");
validation.artifact = ready.packs.len();
let complete = |coverage| FetchServerFrame {
body: Some(fetch_server_frame::Body::Complete(FetchComplete {
revision: ready.current.clone(),
checkpoint: ready.checkpoint.clone(),
closure: coverage,
..Default::default()
})),
};
assert!(
validation
.accept(complete(Coverage::Complete as i32))
.is_err(),
"partial Ready cannot produce a full availability receipt"
);
assert!(matches!(
validation
.accept(complete(Coverage::Partial as i32))
.expect("matching partial completion"),
Item::Complete(_)
));
}
pub(super) fn fixture() -> (FetchOpen, TransferReady, EndpointRef, [Vec<u8>; 2]) {
let signer = Ed25519Signer::from_seed(&[61; 32]).expect("test creator");
let spool_id = uuid::Uuid::from_u128(0x01980000000070008000000000000001);
let genesis = ThreadGenesis {
owner: objects::object::thread_replication::GenesisOwner::LocalKey(
signer.public_key().try_into().expect("public key"),
),
version: 1,
spool: spool_id.to_string(),
parent: None,
base: StateId::from_bytes([17; 32]),
name: "download".into(),
intent: "source".into(),
creator: signer.public_key().try_into().expect("public key"),
nonce: vec![],
};
let thread = ThreadRef {
spool: Some(SpoolRef {
id: spool_id.to_string(),
}),
id: Some(ThreadId {
value: genesis.id().expect("identity").as_bytes().to_vec(),
}),
};
let revision = RevisionRef {
spool: thread.spool.clone(),
revision: Some(revision_ref::Revision::State(
api::heddle::api::common::StateId {
value: [31; 32].to_vec(),
},
)),
};
let endpoint = EndpointRef {
public_key: vec![42; 32],
kind: EndpointKind::Weft as i32,
};
let artifacts = [
b"independently verified pack bytes".to_vec(),
b"independently verified index bytes".to_vec(),
];
let packs = artifacts
.iter()
.enumerate()
.map(|(i, bytes)| {
let address = ObjectAddress {
algorithm: "blake3".into(),
digest: blake3::hash(bytes).as_bytes().to_vec(),
};
PackExtent {
pack: Some(address.clone()),
kind: if i == 0 {
pack_extent::Kind::NativePack
} else {
pack_extent::Kind::NativeIndex
} as i32,
offset: 0,
length: bytes.len() as u64,
extent_digest: Some(address),
}
})
.collect();
let open = FetchOpen {
thread: Some(thread.clone()),
revision: Some(revision.clone()),
selection: Some(TransferSelection {
facets: vec![SharedFacet::Source as i32],
..Default::default()
}),
..Default::default()
};
let ready = TransferReady {
endpoint: Some(endpoint.clone()),
thread: Some(thread),
current: Some(revision),
owner_genesis: Some(
repo::sign_spool_owner_genesis(&signer, *spool_id.as_bytes()).expect("owner signature"),
),
ownership: Some(OwnerState::default()),
thread_genesis: Some(ThreadGenesisRecord {
native_genesis_authority: None,
boundary_acceptances: Vec::new(),
ownership_claims: vec![],
ownership_claim_admissions: vec![],
ownership_resolutions: vec![],
ownership_resolution_admissions: vec![],
genesis: Some(
replication::opening::sign_genesis(&genesis, &signer).expect("creator signature"),
),
creator_authority: vec![],
admission: None,
}),
packs,
checkpoint: Some(TransferCheckpoint {
transfer_id: vec![9; 16],
plan_digest: vec![5; 32],
..Default::default()
}),
budget: Some(ReadBudget {
max_items: 128,
max_frame_bytes: 16 * 1024,
max_snapshot_bytes: 1024 * 1024,
}),
full_closure_available: true,
..Default::default()
};
(open, ready, endpoint, artifacts)
}
fn chunk(ready: &TransferReady, artifact: usize, data: Vec<u8>) -> FetchServerFrame {
let mut extent = ready.packs[artifact].clone();
extent.length = data.len() as u64;
extent.extent_digest = Some(ObjectAddress {
algorithm: "blake3".into(),
digest: blake3::hash(&data).as_bytes().to_vec(),
});
FetchServerFrame {
body: Some(fetch_server_frame::Body::Pack(PackChunk {
extent: Some(extent),
data,
})),
}
}
fn complete(ready: &TransferReady) -> FetchServerFrame {
let mut checkpoint = ready.checkpoint.clone().expect("checkpoint");
checkpoint.committed_bytes = ready.packs.iter().map(|p| p.length).sum();
FetchServerFrame {
body: Some(fetch_server_frame::Body::Complete(FetchComplete {
revision: ready.current.clone(),
checkpoint: Some(checkpoint),
closure: Coverage::Complete as i32,
missing: vec![],
})),
}
}
#[test]
fn native_fetch_commits_only_after_both_hashed_artifacts_and_exact_checkpoint() {
let (open, ready, endpoint, artifacts) = fixture();
let mut download = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
.expect("admission");
assert!(matches!(
download.accept(complete(&ready)),
Err(Error::Invalid(
"download does not match its exact declared source coverage"
))
));
assert!(!download.done);
for (i, data) in artifacts.into_iter().enumerate() {
assert!(matches!(
download
.accept(chunk(&ready, i, data))
.expect("verified chunk"),
Item::Pack(_)
));
}
assert!(matches!(
download.accept(complete(&ready)).expect("complete"),
Item::Complete(_)
));
assert!(download.done);
assert!(
download.accept(complete(&ready)).is_err(),
"terminal checkpoint cannot replay as a new commit"
);
}
#[test]
fn native_fetch_rejects_rehashed_corruption_against_original_whole_artifact() {
let (open, ready, endpoint, mut artifacts) = fixture();
let mut download = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
.expect("admission");
artifacts[0][0] ^= 1;
assert!(matches!(
download.accept(chunk(&ready, 0, artifacts[0].clone())),
Err(Error::Invalid("whole artifact hash mismatch"))
));
assert!(!download.done);
}
#[test]
fn native_fetch_rejects_different_endpoint_and_truncated_index() {
let (open, ready, endpoint, artifacts) = fixture();
let wrong = EndpointRef {
public_key: vec![1; 32],
..endpoint.clone()
};
assert!(matches!(
Validation::new(open.clone(), ready.clone(), Some(&wrong), Limits::default()),
Err(Error::Invalid(
"admission does not match requested endpoint and revision"
))
));
let mut download = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
.expect("admission");
download
.accept(chunk(&ready, 0, artifacts[0].clone()))
.expect("pack");
download
.accept(chunk(&ready, 1, artifacts[1][..3].to_vec()))
.expect("index prefix");
assert!(matches!(
download.accept(complete(&ready)),
Err(Error::Invalid(
"download does not match its exact declared source coverage"
))
));
}
#[test]
fn native_fetch_rejects_cross_spool_genesis_before_source() {
let (open, mut ready, endpoint, _) = fixture();
ready
.owner_genesis
.as_mut()
.expect("owner")
.genesis
.as_mut()
.expect("genesis")
.spool_uuid = uuid::Uuid::now_v7().as_bytes().to_vec();
assert!(matches!(
Validation::new(open, ready, Some(&endpoint), Limits::default()),
Err(Error::Invalid(
"original owner genesis must bind this spool"
))
));
}
#[test]
fn native_fetch_rejects_hybrid_ready_and_operations_before_staging() {
use api::heddle::api::common::ProtocolCompatibility;
let refused = |result: Result<(), Error>| {
assert!(
matches!(result, Err(Error::Invalid(message)) if message.contains("api#307")),
"HYBRID fields must be refused, never ignored"
);
};
let (open, ready, endpoint, _) = fixture();
let mut hybrid = ready.clone();
hybrid.import_authority = Some(ImportPublicProofBundleV1::default());
refused(Validation::new(open.clone(), hybrid, Some(&endpoint), Limits::default()).map(|_| ()));
let mut hybrid = ready.clone();
hybrid.protocol = Some(ProtocolCompatibility::default());
refused(Validation::new(open.clone(), hybrid, Some(&endpoint), Limits::default()).map(|_| ()));
let mut download =
Validation::new(open, ready, Some(&endpoint), Limits::default()).expect("admission");
refused(
download
.accept(FetchServerFrame {
body: Some(fetch_server_frame::Body::Operations(
ReplicationOperations {
import_authority: Some(ImportPublicProofBundleV1::default()),
..Default::default()
},
)),
})
.map(|_| ()),
);
}
#[test]
fn import_ancestry_pages_need_import_authority_and_stay_within_bounds() {
use api::heddle::api::common::StateId as WireStateId;
let (open, ready, endpoint, _) = fixture();
let thread = ready.thread.clone();
let page = |states: usize| ImportAncestryPage {
floor_tiers: Some(ImportFloorTierSummary::default()),
thread: thread.clone(),
tip: Some(WireStateId { value: vec![9; 32] }),
signed_operation_digest: vec![3; 32],
coverage: import_ancestry_page::Coverage::Floor as i32,
page_index: 0,
page_count: 1,
member_count: states as u32,
states: (0..states)
.map(|index| ImportAncestorState {
id: Some(WireStateId {
value: vec![index as u8; 32],
}),
canonical_state: vec![1],
})
.collect(),
};
let frame = |page| FetchServerFrame {
body: Some(fetch_server_frame::Body::ImportAncestry(page)),
};
let mut download = Validation::new(
open.clone(),
ready.clone(),
Some(&endpoint),
Limits::default(),
)
.expect("admission");
assert!(matches!(
download.accept(frame(page(1))),
Err(Error::Invalid(
"import ancestry requires import authority on Ready"
))
));
let mut validation = Validation {
import_operations: std::collections::BTreeSet::from([vec![3; 32]]),
frame_bytes: usize::MAX,
..Validation::new(open, ready, Some(&endpoint), Limits::default()).expect("admission")
};
let mut oversized = page(ANCESTRY_PAGE_STATES + 1);
oversized.member_count = oversized.states.len() as u32;
assert!(matches!(
validation.accept(frame(oversized)),
Err(Error::Invalid("import ancestry page bounds"))
));
let mut unspecified = page(1);
unspecified.coverage = 0;
assert!(matches!(
validation.accept(frame(unspecified)),
Err(Error::Invalid("import ancestry coverage unspecified"))
));
let mut foreign = page(1);
foreign.signed_operation_digest = vec![4; 32];
assert!(matches!(
validation.accept(frame(foreign)),
Err(Error::Invalid(
"import ancestry names an operation outside the carried authority"
))
));
let mut narrow = page(1);
narrow.states[0].id = Some(WireStateId { value: vec![1; 31] });
assert!(matches!(
validation.accept(frame(narrow)),
Err(Error::Invalid("import ancestor State framing"))
));
assert!(matches!(
validation.accept(frame(page(2))).expect("well-formed page"),
Item::ImportAncestry(_)
));
assert_eq!(validation.ancestry_states, 2);
}