Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584use std::path::Path;use std::time::Instant;
use knot_cob::{CobHome, CobStore};use knot_cobs::{Grant, MembersChange};use knot_git::{Layout, Repo};use knot_pack::{ MaxWireBytes, PackLimits, PackReceiver, ReceiveCommand, ReceiveFramer, ReceiveGuard, RefDecision, receive_pack_guarded, receive_request_complete, sweep_incoming,};use knot_runtime::{K256Signer, SeededEntropy, Signer};use knot_types::{AccountDid, ActorId, ObjectFormat, Oid, RepoDid, UnixSeconds};
const SHA1: ObjectFormat = ObjectFormat::SHA1;
mod common;use common::{commit, must, pack_objects, receive_request};
fn limits() -> PackLimits { PackLimits::default()}
fn seeded_pack(layout: &Layout, did: &RepoDid) -> (Repo, String, Vec<u8>) { let bare = layout.create(did).unwrap(); let work_dir = tempfile::tempdir().unwrap(); let work = work_dir.path(); must(work, &["init", "-q", "-b", "main"]); commit(work, "a.txt", "x\n", "c1"); let c1 = must(work, &["rev-parse", "HEAD"]); let oids: Vec<String> = must(work, &["rev-list", "--objects", &c1]) .lines() .map(|line| line.split_whitespace().next().unwrap().to_string()) .collect(); let pack = pack_objects(work, &oids); (bare, c1, pack)}
fn report_text(report: &[u8]) -> String { String::from_utf8_lossy(report).replace('\0', "")}
fn dir_count(path: &Path) -> usize { std::fs::read_dir(path) .map(|entries| entries.filter_map(Result::ok).count()) .unwrap_or(0)}
fn live_object_count(repo: &Repo) -> usize { let objects = repo.objects_dir(); let loose: usize = std::fs::read_dir(&objects) .map(|entries| { entries .filter_map(Result::ok) .filter(|entry| { entry .file_name() .to_str() .is_some_and(|name| name.len() == 2) && entry.path().is_dir() }) .map(|shard| dir_count(&shard.path())) .sum() }) .unwrap_or(0); let packs = std::fs::read_dir(objects.join("pack")) .map(|entries| { entries .filter_map(Result::ok) .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "pack")) .count() }) .unwrap_or(0); loose + packs}
struct AllowPublic;
impl ReceiveGuard for AllowPublic { fn authorize(&self, _staged: &Repo, commands: &[ReceiveCommand]) -> Vec<RefDecision> { commands .iter() .map(|command| { if command.name().is_some_and(knot_git::is_public_ref) { RefDecision::Allow } else { RefDecision::Reject("not public ref".to_string()) } }) .collect() }}
struct DenyAll;
impl ReceiveGuard for DenyAll { fn authorize(&self, _staged: &Repo, commands: &[ReceiveCommand]) -> Vec<RefDecision> { commands .iter() .map(|_| RefDecision::Reject("unauthorized".to_string())) .collect() }}
struct AllowAll;
impl ReceiveGuard for AllowAll { fn authorize(&self, _staged: &Repo, commands: &[ReceiveCommand]) -> Vec<RefDecision> { commands.iter().map(|_| RefDecision::Allow).collect() }}
struct CobVerify { owner: ActorId, home: CobHome,}
impl ReceiveGuard for CobVerify { fn authorize(&self, staged: &Repo, commands: &[ReceiveCommand]) -> Vec<RefDecision> { commands .iter() .map(|command| { let store = CobStore::new(staged); let Ok(name) = knot_types::RefName::new(command.refname()) else { return RefDecision::Reject("invalid ref name".to_string()); }; match knot_cobs::verify_cob_ref(&store, &self.home, &name, &self.owner) { Ok(_) => RefDecision::Allow, Err(error) => RefDecision::Reject(error.to_string()), } }) .collect() }}
#[test]fn a_denied_push_leaves_no_objects_in_the_live_odb() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let did = RepoDid::new("did:plc:squid").unwrap(); let (bare, c1, pack) = seeded_pack(&layout, &did);
assert_eq!( live_object_count(&bare), 0, "fresh bare repo has no objects" ); let request = receive_request("refs/heads/main", &Oid::null().to_hex(), &c1, &pack); let report = receive_pack_guarded( &bare, &request, &limits(), &DenyAll, &|_| {}, &knot_pack::default_catalog().reject, ) .unwrap() .report; let report = report_text(&report); assert!( report.contains("ng refs/heads/main unauthorized"), "{report}" );
assert!( bare.references().unwrap().is_empty(), "denied push mustn't create the ref" ); assert_eq!( live_object_count(&bare), 0, "denied push must leave no objects in the live odb" ); assert!(!bare.contains(Oid::from_hex(&c1).unwrap()));}
#[test]fn an_authorized_push_migrates_objects_then_a_stale_push_is_rejected() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let did = RepoDid::new("did:plc:squid").unwrap(); let (bare, c1, pack) = seeded_pack(&layout, &did);
let report = receive_pack_guarded( &bare, &receive_request("refs/heads/main", &Oid::null().to_hex(), &c1, &pack), &limits(), &AllowPublic, &|_| {}, &knot_pack::default_catalog().reject, ) .unwrap() .report; let report = report_text(&report); assert!(report.contains("unpack ok"), "{report}"); assert!(report.contains("ok refs/heads/main"), "{report}"); let refs = bare.references().unwrap(); assert_eq!(refs.len(), 1); assert_eq!(refs[0].name.as_str(), "refs/heads/main"); assert_eq!(refs[0].target, Oid::from_hex(&c1).unwrap()); assert!(bare.contains(Oid::from_hex(&c1).unwrap())); let after_first = live_object_count(&bare);
let wrong_old = "1".repeat(40); let fresh = "2".repeat(40); let report = receive_pack_guarded( &bare, &receive_request("refs/heads/main", &wrong_old, &fresh, b""), &limits(), &AllowPublic, &|_| {}, &knot_pack::default_catalog().reject, ) .unwrap() .report; let report = report_text(&report); assert!(report.contains("ng refs/heads/main"), "{report}"); assert_eq!( bare.find_ref(&knot_types::RefName::new("refs/heads/main").unwrap()) .unwrap(), Some(Oid::from_hex(&c1).unwrap()), "stale push mustn't move the ref" ); assert_eq!( live_object_count(&bare), after_first, "stale push mustn't add objects to the live odb" );}
#[test]fn a_cob_ref_verifies_against_the_owner_key_and_is_refused_for_a_stranger() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let signer = K256Signer::generate(&SeededEntropy::new(7));
let source_did = RepoDid::new("did:plc:source").unwrap(); let home = CobHome::from(&source_did); let source = layout.create(&source_did).unwrap(); let store = CobStore::new(&source); let grant = Grant { subject: AccountDid::new("did:plc:nel").unwrap(), added_by: AccountDid::new("did:plc:nel").unwrap(), created_at: UnixSeconds::new(1), }; let created = store .create( &home, &MembersChange::Add(grant), &signer, UnixSeconds::new(1), ) .unwrap(); let tip = created.tip.oid().to_hex(); let refname = format!( "refs/cobs/sh.tangled.knot.member/{}", created.object.oid().to_hex() );
let reachable: Vec<String> = must(source.path(), &["rev-list", "--objects", &tip]) .lines() .map(|line| line.split_whitespace().next().unwrap().to_string()) .collect(); let pack = pack_objects(source.path(), &reachable); let owner = ActorId::from_secp256k1(signer.public_key().as_bytes());
let dest = layout .create(&RepoDid::new("did:plc:dest").unwrap()) .unwrap(); let report = receive_pack_guarded( &dest, &receive_request(&refname, &Oid::null().to_hex(), &tip, &pack), &limits(), &CobVerify { owner: owner.clone(), home: home.clone(), }, &|_| {}, &knot_pack::default_catalog().reject, ) .unwrap() .report; assert!( report_text(&report).contains(&format!("ok {refname}")), "owner-signed COB ref must be accepted at the receive boundary: {}", report_text(&report) );
let stranger_key = K256Signer::generate(&SeededEntropy::new(9)); let stranger = ActorId::from_secp256k1(stranger_key.public_key().as_bytes()); let dest2 = layout .create(&RepoDid::new("did:plc:dest2").unwrap()) .unwrap(); let report = receive_pack_guarded( &dest2, &receive_request(&refname, &Oid::null().to_hex(), &tip, &pack), &limits(), &CobVerify { owner: stranger, home: home.clone(), }, &|_| {}, &knot_pack::default_catalog().reject, ) .unwrap() .report; assert!( report_text(&report).contains(&format!("ng {refname}")), "COB ref not signed by the resolved owner key must be refused: {}", report_text(&report) ); assert_eq!( live_object_count(&dest2), 0, "refused COB ref push leaves no objects behind" );}
#[test]fn a_reserved_ref_update_is_refused_even_when_the_guard_allows_it() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let did = RepoDid::new("did:plc:squid").unwrap(); let bare = layout.create(&did).unwrap();
let cob_ref = format!("refs/cobs/sh.tangled.knot.member/{}", "a".repeat(40)); let old = "1".repeat(40); let new = "2".repeat(40); let report = receive_pack_guarded( &bare, &receive_request(&cob_ref, &old, &new, b""), &limits(), &AllowAll, &|_| {}, &knot_pack::default_catalog().reject, ) .unwrap() .report; let report = report_text(&report);
assert!( report.contains(&format!("ng {cob_ref}")), "non-create update to a reserved ref must be refused even under an allow-all guard: {report}" ); assert!( report.contains("cannot be modified"), "refusal must name the create-only rule, not connectivity or compare-and-swap: {report}" ); assert!( bare.references().unwrap().is_empty(), "refused reserved-ref update must land nothing" ); assert_eq!( live_object_count(&bare), 0, "refused reserved-ref update must migrate no objects" );}
fn pseudo_random(seed: u64, len: usize) -> Vec<u8> { let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15).wrapping_add(1); (0..len) .map(|_| { state ^= state << 13; state ^= state >> 7; state ^= state << 17; (state >> 24) as u8 }) .collect()}
fn incompressible_pack(megabytes: usize) -> Vec<u8> { let work_dir = tempfile::tempdir().unwrap(); let work = work_dir.path(); must(work, &["init", "-q", "-b", "main"]); (0..megabytes).for_each(|i| { let blob = pseudo_random(i as u64 + 1, 1024 * 1024); std::fs::write(work.join(format!("blob-{i:04}.bin")), blob).unwrap(); }); must(work, &["add", "-A"]); must(work, &["commit", "-q", "-m", "bulk"]); let head = must(work, &["rev-parse", "HEAD"]); let oids: Vec<String> = must(work, &["rev-list", "--objects", &head]) .lines() .map(|line| line.split_whitespace().next().unwrap().to_string()) .collect(); pack_objects(work, &oids)}
#[test]fn the_receive_framer_scans_incrementally_in_linear_time() { const READ_CHUNK: usize = 64 * 1024; let pack = incompressible_pack(24); let body = receive_request( "refs/heads/main", &Oid::null().to_hex(), &"1".repeat(40), &pack, ); assert!( pack.len() > 8 * 1024 * 1024, "pack must be large enough to expose any quadratic scaling: {} bytes", pack.len() );
let single_start = Instant::now(); let complete = ReceiveFramer::new(limits(), SHA1.kind()) .advance_bytes(&body) .unwrap(); let single = single_start.elapsed(); assert_eq!( complete, Some(body.len()), "one advance over the full body frames whole request" );
let chunks = body.len().div_ceil(READ_CHUNK); let chunked_start = Instant::now(); let mut framer = ReceiveFramer::new(limits(), SHA1.kind()); let mut framed_at = None; (1..=chunks).for_each(|n| { let end = (n * READ_CHUNK).min(body.len()); if framed_at.is_none() && let Some(total) = framer.advance_bytes(&body[..end]).unwrap() { framed_at = Some(total); } }); let chunked = chunked_start.elapsed(); assert_eq!( framed_at, Some(body.len()), "feeding the same body in 64KB chunks frames it at identical length" );
let ratio = chunked.as_secs_f64() / single.as_secs_f64().max(1e-6); assert!( ratio < 5.0, "resumable framer never re-inflates settled objects, so the {chunks}-chunk feed \ must stay within a small constant of one pass; got {ratio:.2}x" );}
#[test]fn receive_request_completion_framing_and_the_per_object_limit() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let did = RepoDid::new("did:plc:squid").unwrap(); let (_bare, c1, pack) = seeded_pack(&layout, &did); let request = receive_request("refs/heads/main", &Oid::null().to_hex(), &c1, &pack);
assert_eq!( receive_request_complete(&request[..request.len() / 2], &limits(), SHA1.kind()).unwrap(), None, "half-delivered request isn't yet complete" ); assert_eq!( receive_request_complete(&request, &limits(), SHA1.kind()).unwrap(), Some(request.len()), "whole request reports its exact length" ); let mut trailing = request.clone(); trailing.extend_from_slice(b"junk-after-the-pack"); assert_eq!( receive_request_complete(&trailing, &limits(), SHA1.kind()).unwrap(), Some(request.len()), "framer stops at the pack trailer, ignoring trailing bytes" );
let tight = PackLimits { max_object_bytes: knot_pack::MaxObjectBytes::new(1), ..PackLimits::default() }; assert!( receive_request_complete(&request, &tight, SHA1.kind()).is_err(), "object beyond the per-object limit is refused by the framer" );}
#[test]fn sweep_incoming_removes_abandoned_staging_directories() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let did = RepoDid::new("did:plc:squid").unwrap(); let bare = layout.create(&did).unwrap();
let staging = bare.path().join(".knot-incoming.4242.0"); std::fs::create_dir_all(staging.join("objects")).unwrap(); std::fs::write(staging.join("objects").join("leftover"), b"x").unwrap(); assert!(staging.exists());
let swept = sweep_incoming(scan.path()); assert_eq!(swept, 1, "exactly one staging directory is swept"); assert!(!staging.exists(), "abandoned staging directory is gone"); assert!( bare.path().join("objects").exists(), "live repository is untouched" );}
#[test]fn streamed_receipt_reassembles_a_request_byte_for_byte() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let did = RepoDid::new("did:plc:squid").unwrap(); let (_bare, c1, pack) = seeded_pack(&layout, &did); let request = receive_request("refs/heads/main", &Oid::null().to_hex(), &c1, &pack);
let dir = tempfile::tempdir().unwrap(); let mut receiver = PackReceiver::new( dir.path(), MaxWireBytes::new(1 << 30), limits(), SHA1.kind(), ) .unwrap(); let completed = request .chunks(7) .try_fold(false, |_, chunk| receiver.write(chunk)) .unwrap(); assert!( completed, "the framer must detect completion within the streamed request" );
let body = receiver.finish().unwrap(); let mut reconstructed = body.preamble().to_vec(); if let Some(pack) = body.open_pack().unwrap() { reconstructed.extend_from_slice(&std::fs::read(pack.path()).unwrap()); } assert_eq!( &reconstructed[..], &request[..], "the streamed body must be byte-identical to the buffered request" ); assert_eq!( receive_request_complete(&request, &limits(), SHA1.kind()).unwrap(), Some(request.len()), "streamed total must agree with the buffered framer" );}
#[test]fn streamed_receipt_handles_empty_and_delete_only_requests() { let scan = tempfile::tempdir().unwrap(); let layout = Layout::new(scan.path()); let did = RepoDid::new("did:plc:squid").unwrap(); let (_bare, c1, _pack) = seeded_pack(&layout, &did); let dir = tempfile::tempdir().unwrap();
let empty = PackReceiver::new( dir.path(), MaxWireBytes::new(1 << 30), limits(), SHA1.kind(), ) .unwrap() .finish() .unwrap(); assert!( empty.is_empty(), "a stream with no bytes yields an empty body" );
let delete = receive_request("refs/heads/main", &c1, &Oid::null().to_hex(), &[]); let mut receiver = PackReceiver::new( dir.path(), MaxWireBytes::new(1 << 30), limits(), SHA1.kind(), ) .unwrap(); let completed = delete .chunks(5) .try_fold(false, |_, chunk| receiver.write(chunk)) .unwrap(); assert!(completed, "a delete-only push completes without a pack"); let body = receiver.finish().unwrap(); assert_eq!(body.preamble(), &delete[..]); assert!( body.open_pack().unwrap().is_none(), "a delete-only push has no pack" );}