diff --git a/knot2/crates/knot-xrpc/tests/private_stream.rs b/knot2/crates/knot-xrpc/tests/private_stream.rs new file mode 100644 index 000000000..8d866c750 --- /dev/null +++ b/knot2/crates/knot-xrpc/tests/private_stream.rs @@ -0,0 +1,449 @@ +mod common; + +use std::collections::BTreeSet; +use std::time::Duration; + +use futures::StreamExt; +use http::{HeaderMap, StatusCode}; +use ipld_core::ipld::Ipld; +use tokio_tungstenite::tungstenite::Message; + +use knot_git::Repo; +use knot_record::chain; +use knot_types::{AccountDid, RepoDid, UnixMicros, UnixSeconds}; + +use common::{ + Answer, NO_DATE_OR_LENGTH, World, bearer_headers, collaborator_collection, commit_file, + create_ticket, get_answer, ghost, grant_body, http_push, knot_service_did, nsid, owner, + post_as, post_authed, serve_pushes, sh_git, stranger, ticket_collection, +}; + +const PRIVATE_NAME: &str = "vault-kelp"; +const DOOMED_NAME: &str = "vault-doomed"; +const PRIVATE_TICKET: &str = "secret-ticket-title"; +const PRIVATE_COMMENT: &str = "secret-comment-text"; +const DOOMED_TICKET: &str = "doomed-ticket-title"; +const CANARY: &str = "canary-driftwood"; +const OUTSIDER: &str = "did:plc:outsider"; +const COMMENT: &str = "org.tangled.feed.comment"; + +struct Activity { + public: RepoDid, + private: RepoDid, + doomed: RepoDid, + private_ticket: String, + private_cids: BTreeSet>, + doomed_cids: BTreeSet>, +} + +impl Activity { + fn secrets(&self) -> Vec { + [ + self.private.as_str(), + self.doomed.as_str(), + PRIVATE_NAME, + DOOMED_NAME, + PRIVATE_TICKET, + PRIVATE_COMMENT, + DOOMED_TICKET, + stranger().as_str(), + ] + .map(str::to_owned) + .to_vec() + } +} + +async fn create_repo(world: &World, name: &str, private: bool) -> RepoDid { + let (status, created) = post_authed( + world, + &nsid("org.tangled.repo.createRepo"), + &owner(), + serde_json::json!({ "name": name, "private": private }), + ) + .await; + assert_eq!(status, StatusCode::OK, "creating {name}: {created}"); + RepoDid::new(created["did"].as_str().expect("create names the did")).unwrap() +} + +async fn push_commits(world: &World, addr: std::net::SocketAddr, did: &RepoDid, name: &str) { + let work = tempfile::tempdir().unwrap(); + sh_git(work.path(), &["init", "-q", "-b", "main"]); + for (n, when) in ["2026-06-01T12:30:00+02:00", "2026-06-01T12:31:00+02:00"] + .iter() + .enumerate() + { + commit_file( + work.path(), + "README.md", + format!("{name} {n}\n").as_bytes(), + &format!("{name} commit {n}"), + when, + ); + let (ok, log) = http_push(work.path(), addr, did, &owner(), &["main"]).await; + assert!(ok, "pushing {name}: {log}"); + } + assert!(world.layout.open(did).is_ok()); +} + +async fn comment_on(world: &World, repo: &RepoDid, ticket: &str, cid: &str, text: &str) { + let (status, written) = post_as( + world, + &nsid("com.atproto.repo.createRecord"), + &owner(), + &knot_types::ServiceDid::new(repo.as_str()).unwrap(), + serde_json::json!({ + "repo": repo, + "collection": COMMENT, + "record": { + "$type": COMMENT, + "subject": { "uri": ticket, "cid": cid }, + "body": { "$type": "org.tangled.markup.markdown", "text": text }, + "createdAt": "2026-09-11T00:00:00Z" + } + }), + ) + .await; + assert_eq!(status, StatusCode::OK, "commenting on {ticket}: {written}"); +} + +async fn open_ticket( + world: &World, + repo: &RepoDid, + title: &str, + comment: &str, +) -> (String, String) { + let (rkey, cid) = create_ticket(world, repo, title, "a body").await; + let uri = format!("at://{repo}/{}/{}", ticket_collection(), rkey); + comment_on(world, repo, &uri, &cid.cid().to_string(), comment).await; + (uri, cid.cid().to_string()) +} + +fn frame_parts(frame: &[u8]) -> (Ipld, Ipld) { + let mut read = frame; + let header = serde_ipld_dagcbor::de::from_reader_once(&mut read).unwrap(); + let payload = serde_ipld_dagcbor::de::from_reader_once(&mut read).unwrap(); + (header, payload) +} + +fn field<'a>(map: &'a Ipld, key: &str) -> Option<&'a Ipld> { + match map { + Ipld::Map(entries) => entries.get(key), + _ => None, + } +} + +async fn block_cids(frame: &[u8]) -> BTreeSet> { + let (_, payload) = frame_parts(frame); + let mut cids = BTreeSet::new(); + if let Some(Ipld::Link(commit)) = field(&payload, "commit") { + cids.insert(commit.to_bytes()); + } + if let Some(Ipld::Bytes(car)) = field(&payload, "blocks") { + let parsed = jacquard_repo::car::parse_car_bytes(car).await.unwrap(); + cids.extend(parsed.blocks.keys().map(|cid| cid.to_bytes())); + } + cids +} + +async fn chain_frames(world: &World, did: &RepoDid) -> Vec> { + frames_of(&world.layout.open(did).unwrap()) +} + +fn frames_of(repo: &Repo) -> Vec> { + let Some(tip) = chain::tip(repo).unwrap() else { + return Vec::new(); + }; + chain::frames_after(repo, tip.git, UnixMicros::new(0), usize::MAX) + .found + .into_iter() + .map(|(_, frame)| frame.to_vec()) + .collect() +} + +async fn chain_cids(world: &World, did: &RepoDid) -> BTreeSet> { + cids_of(chain_frames(world, did).await).await +} + +async fn cids_of(frames: Vec>) -> BTreeSet> { + let mut all = BTreeSet::new(); + for frame in frames { + all.extend(block_cids(&frame).await); + } + all +} + +async fn run_activity(world: &World, addr: std::net::SocketAddr) -> Activity { + let public = create_repo(world, "reef", false).await; + let private = create_repo(world, PRIVATE_NAME, true).await; + let doomed = create_repo(world, DOOMED_NAME, true).await; + for (did, name) in [ + (&public, "reef"), + (&private, PRIVATE_NAME), + (&doomed, DOOMED_NAME), + ] { + push_commits(world, addr, did, name).await; + } + open_ticket(world, &public, "public-ticket-title", "public-comment-text").await; + let (private_ticket, _) = open_ticket(world, &private, PRIVATE_TICKET, PRIVATE_COMMENT).await; + open_ticket(world, &doomed, DOOMED_TICKET, "doomed-comment-text").await; + + let offered = post_as( + world, + &nsid("com.atproto.repo.createRecord"), + &owner(), + &knot_types::ServiceDid::new(private.as_str()).unwrap(), + grant_body( + &private, + &collaborator_collection(), + &stranger(), + "2026-09-11T00:00:00Z", + ), + ) + .await; + assert_eq!(offered.0, StatusCode::OK, "{}", offered.1); + + let private_cids = chain_cids(world, &private).await; + let doomed_cids = chain_cids(world, &doomed).await; + let (status, deleted) = post_authed( + world, + &nsid("org.tangled.repo.deleteRepo"), + &owner(), + serde_json::json!({ "repo": doomed }), + ) + .await; + assert_eq!(status, StatusCode::OK, "{deleted}"); + + open_ticket(world, &public, CANARY, "after the private ones").await; + Activity { + public, + private, + doomed, + private_ticket, + private_cids, + doomed_cids, + } +} + +type Ws = + tokio_tungstenite::WebSocketStream>; + +async fn subscribe(addr: std::net::SocketAddr, query: &str) -> Ws { + let (ws, _) = tokio_tungstenite::connect_async(format!( + "ws://{addr}/xrpc/com.atproto.sync.subscribeRepos{query}" + )) + .await + .unwrap(); + ws +} + +fn contains(haystack: &[u8], needle: &[u8]) -> bool { + !needle.is_empty() + && haystack + .windows(needle.len()) + .any(|window| window == needle) +} + +async fn drain_until_public_is_caught_up( + ws: &mut Ws, + world: &World, + public: &RepoDid, +) -> Vec> { + let mut expected = BTreeSet::new(); + for frame in chain_frames(world, public).await { + let (_, payload) = frame_parts(&frame); + let Some(Ipld::Link(commit)) = field(&payload, "commit") else { + panic!("a chain frame names its commit"); + }; + expected.insert(commit.to_bytes()); + } + assert!( + !expected.is_empty(), + "the public chain has frames to wait for" + ); + let mut frames = Vec::new(); + let mut seen = BTreeSet::new(); + let mut canary = false; + while !(canary && expected.is_subset(&seen)) { + let next = tokio::time::timeout(Duration::from_secs(30), ws.next()) + .await + .unwrap_or_else(|_| { + panic!( + "the stream went quiet with {} of {} public commits delivered, canary {canary}", + seen.len(), + expected.len() + ) + }) + .expect("the stream stays open") + .expect("the frame is readable"); + let Message::Binary(bytes) = next else { + continue; + }; + canary |= contains(&bytes, CANARY.as_bytes()); + let (_, payload) = frame_parts(&bytes); + if let Some(Ipld::Link(commit)) = field(&payload, "commit") { + seen.insert(commit.to_bytes()); + } + frames.push(bytes.to_vec()); + } + frames +} + +async fn assert_nothing_private_in( + frames: &[Vec], + world: &World, + activity: &Activity, + via: &str, +) { + let public_cids = { + let mut all = chain_cids(world, &activity.public).await; + let meta = Repo::open(&world.state.meta_path).unwrap(); + all.extend(cids_of(frames_of(&meta)).await); + all + }; + let private_only: BTreeSet<_> = activity.private_cids.difference(&public_cids).collect(); + let doomed_only: BTreeSet<_> = activity.doomed_cids.difference(&public_cids).collect(); + assert!( + !private_only.is_empty() && !doomed_only.is_empty(), + "the private chains hold blocks of their own to look for" + ); + let secrets = activity.secrets(); + let mut public_frames = 0; + for frame in frames { + for secret in &secrets { + assert!( + !contains(frame, secret.as_bytes()), + "{via}: a frame carries {secret}" + ); + } + for cid in private_only.iter().chain(doomed_only.iter()) { + assert!( + !contains(frame, cid), + "{via}: a frame carries a block of a private chain" + ); + } + public_frames += contains(frame, activity.public.as_str().as_bytes()) as usize; + } + assert!( + public_frames >= chain_frames(world, &activity.public).await.len(), + "{via}: every public commit still flows ({public_frames} frames name the public repo)" + ); +} + +#[tokio::test] +async fn a_replayed_firehose_never_names_a_private_repo_or_carries_its_blocks() { + let world = World::administered(); + world.add_member(&owner(), &owner(), UnixSeconds::new(1_000)); + let addr = serve_pushes(&world).await; + let activity = run_activity(&world, addr).await; + + let mut ws = subscribe(addr, "?cursor=0").await; + let frames = drain_until_public_is_caught_up(&mut ws, &world, &activity.public).await; + assert_nothing_private_in(&frames, &world, &activity, "replay from cursor 0").await; +} + +#[tokio::test] +async fn a_live_firehose_never_names_a_private_repo_or_carries_its_blocks() { + let world = World::administered(); + world.add_member(&owner(), &owner(), UnixSeconds::new(1_000)); + let addr = serve_pushes(&world).await; + let mut ws = subscribe(addr, "").await; + let activity = run_activity(&world, addr).await; + + let frames = drain_until_public_is_caught_up(&mut ws, &world, &activity.public).await; + assert_nothing_private_in(&frames, &world, &activity, "live").await; +} + +async fn answer(world: &World, path: &str, headers: HeaderMap, did: &RepoDid) -> Answer { + get_answer(world, path, headers, did, NO_DATE_OR_LENGTH).await +} + +#[tokio::test] +async fn a_private_repo_answers_the_sync_reads_like_a_missing_did() { + let world = World::unshed(); + world.add_member(&owner(), &owner(), UnixSeconds::new(1_000)); + let addr = serve_pushes(&world).await; + let activity = run_activity(&world, addr).await; + let block = chain_frames(&world, &activity.private).await; + let cid = { + let (_, payload) = frame_parts(block.last().unwrap()); + match field(&payload, "commit") { + Some(Ipld::Link(cid)) => cid.to_string(), + other => panic!("the frame names its commit, not {other:?}"), + } + }; + let rkey = activity + .private_ticket + .rsplit('/') + .next() + .unwrap() + .to_owned(); + let collection = ticket_collection(); + let routes: Vec<(&str, String)> = vec![ + ("com.atproto.sync.getRepo", "did={did}".into()), + ("com.atproto.sync.getLatestCommit", "did={did}".into()), + ("com.atproto.sync.getRepoStatus", "did={did}".into()), + ( + "com.atproto.sync.getRecord", + format!("did={{did}}&collection={collection}&rkey={rkey}"), + ), + ( + "com.atproto.sync.getBlocks", + format!("did={{did}}&cids={cid}"), + ), + ( + "com.atproto.repo.getRecord", + format!("repo={{did}}&collection={collection}&rkey={rkey}"), + ), + ( + "com.atproto.repo.listRecords", + format!("repo={{did}}&collection={collection}"), + ), + ("com.atproto.repo.describeRepo", "repo={did}".into()), + ]; + let ghost = ghost(); + let outsider = AccountDid::new(OUTSIDER).unwrap(); + let (mut compared, mut members) = (0, 0); + for (method, query) in &routes { + let path = + |did: &RepoDid| format!("/xrpc/{method}?{}", query.replace("{did}", did.as_str())); + let viewers: [(&str, Option<&AccountDid>); 2] = + [("anonymous", None), ("an outsider", Some(&outsider))]; + for (who, actor) in viewers { + let headers = |token_for: Option<&AccountDid>| match token_for { + Some(actor) => bearer_headers(&nsid(method), actor, &knot_service_did()), + None => HeaderMap::new(), + }; + let hidden = answer( + &world, + &path(&activity.private), + headers(actor), + &activity.private, + ) + .await; + let missing = answer(&world, &path(&ghost), headers(actor), &ghost).await; + assert!( + missing.0 == StatusCode::BAD_REQUEST, + "{who} on {method}: a missing did answers 400 RepoNotFound, got {}", + missing.0 + ); + assert_eq!(hidden, missing, "{who} on {method}: private vs missing"); + compared += 1; + } + for member in [owner(), stranger()] { + let read = answer( + &world, + &path(&activity.private), + headers_of(method, &member), + &activity.private, + ) + .await; + assert_eq!(read.0, StatusCode::OK, "{member} on {method}: {}", read.2); + members += 1; + } + } + assert_eq!((compared, members), (routes.len() * 2, routes.len() * 2)); +} + +fn headers_of(method: &str, actor: &AccountDid) -> HeaderMap { + bearer_headers(&nsid(method), actor, &knot_service_did()) +}