alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::{HashMap, HashSet};
2use std::path::{Path, PathBuf};
3use std::process::Command;
4use std::sync::atomic::{AtomicUsize, Ordering};
5use std::sync::{Arc, Mutex};
6
7use futures::stream::StreamExt;
8use knot_atproto::Atproto;
9use knot_cob::{CobHome, CobStore};
10use knot_cobs::{CollaboratorsChange, Grant, MembersChange, Registration, RegistryChange};
11use knot_git::{ArchiveLimit, Layout, Repo};
12use knot_index::{Index, Resolved};
13use knot_pack::MaxWireBytes;
14use knot_postreceive::LanguagesPushBudget;
15use knot_runtime::{
16 FakeDns, FakeHttp, HttpResponse, K256Signer, ManualClock, SeededEntropy, Signer, UnixMicros,
17};
18use knot_types::{
19 AccountDid, KnotId, Oid, OwnerDid, RefName, RepoDid, RepoName, RepoRkey, UnixSeconds,
20};
21use tempfile::TempDir;
22use tokio::net::TcpListener;
23use url::Url;
24
25const REPO_DID: &str = "did:plc:squid";
26const REPO_NAME: &str = "anemone";
27const OWNER_DID: &str = "did:plc:nel";
28const TID_REPO_DID: &str = "did:plc:limpet";
29const TID_RKEY: &str = "3mizfnpxii522";
30const TID_REPO_NAME: &str = "periwinkle.cloud";
31const PDS_HOST: &str = "pds.oyster.cafe";
32
33fn git(cwd: &Path, env: &[(&str, &str)], args: &[&str]) -> (bool, String) {
34 let mut command = knot_fixtures::command(cwd);
35 command.args(args);
36 env.iter().for_each(|(key, value)| {
37 command.env(key, value);
38 });
39 let out = command.output().expect("git runs");
40 let combined = format!(
41 "{}{}",
42 String::from_utf8_lossy(&out.stdout),
43 String::from_utf8_lossy(&out.stderr)
44 );
45 (out.status.success(), combined)
46}
47
48fn keygen(dir: &Path, name: &str) -> (String, String) {
49 let path = dir.join(name);
50 let out = Command::new("ssh-keygen")
51 .args([
52 "-t",
53 "ed25519",
54 "-N",
55 "",
56 "-C",
57 "nel@oyster.cafe",
58 "-f",
59 path.to_str().unwrap(),
60 ])
61 .output()
62 .expect("ssh-keygen runs");
63 assert!(
64 out.status.success(),
65 "ssh-keygen failed: {}",
66 String::from_utf8_lossy(&out.stderr)
67 );
68 let public_line = std::fs::read_to_string(dir.join(format!("{name}.pub")))
69 .unwrap()
70 .trim()
71 .to_string();
72 (path.to_str().unwrap().to_string(), public_line)
73}
74
75fn did_document(signer: &K256Signer, did: &str, pds: &str) -> Vec<u8> {
76 let multikey = knot_types::crypto::multikey(0xe7, signer.public_key().as_bytes());
77 serde_json::to_vec(&serde_json::json!({
78 "id": did,
79 "alsoKnownAs": ["at://nel.pet"],
80 "verificationMethod": [{
81 "id": format!("{did}#atproto"),
82 "type": "Multikey",
83 "controller": did,
84 "publicKeyMultibase": multikey
85 }],
86 "service": [{
87 "id": "#atproto_pds",
88 "type": "AtprotoPersonalDataServer",
89 "serviceEndpoint": pds
90 }]
91 }))
92 .unwrap()
93}
94
95fn list_records_body(lines: &[&str]) -> Vec<u8> {
96 let records: Vec<_> = lines
97 .iter()
98 .map(|line| {
99 serde_json::json!({
100 "value": {
101 "$type": "sh.tangled.publicKey",
102 "key": line,
103 "name": "laptop",
104 "createdAt": "2026-06-08T00:00:00Z"
105 }
106 })
107 })
108 .collect();
109 serde_json::to_vec(&serde_json::json!({ "records": records })).unwrap()
110}
111
112fn ok_body(body: Vec<u8>) -> HttpResponse {
113 HttpResponse {
114 status: http::StatusCode::OK,
115 headers: http::HeaderMap::new(),
116 body: bytes::Bytes::from(body),
117 }
118}
119
120fn fake_dns() -> impl knot_runtime::DnsTxtResolver {
121 FakeDns::new(|name: &str| {
122 Ok(match name {
123 "_atproto.nel.pet" => vec![format!("did={OWNER_DID}")],
124 _ => Vec::new(),
125 })
126 })
127}
128
129fn not_found() -> HttpResponse {
130 HttpResponse {
131 status: http::StatusCode::NOT_FOUND,
132 headers: http::HeaderMap::new(),
133 body: bytes::Bytes::new(),
134 }
135}
136
137fn forever() -> knot_index::KeyLease {
138 knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(UnixSeconds::new(0))
139}
140
141fn server_error() -> HttpResponse {
142 HttpResponse {
143 status: http::StatusCode::INTERNAL_SERVER_ERROR,
144 headers: http::HeaderMap::new(),
145 body: bytes::Bytes::new(),
146 }
147}
148
149fn fake_http(published_line: String) -> impl knot_runtime::HttpTransport {
150 let signer = K256Signer::generate(&SeededEntropy::new(1));
151 let pds = format!("https://{PDS_HOST}");
152 FakeHttp::new(move |request| {
153 let host = request.url.host_str().unwrap_or_default().to_string();
154 let path = request.url.path().to_string();
155 let body = if host == PDS_HOST {
156 list_records_body(&[&published_line])
157 } else if path.ends_with(REPO_DID) {
158 did_document(&signer, REPO_DID, &pds)
159 } else if path.ends_with(TID_REPO_DID) {
160 did_document(&signer, TID_REPO_DID, &pds)
161 } else if path.ends_with(OWNER_DID) {
162 did_document(&signer, OWNER_DID, &pds)
163 } else {
164 return Ok(not_found());
165 };
166 Ok(ok_body(body))
167 })
168}
169
170#[derive(Default, Clone)]
171struct Accounts {
172 identities: Arc<Mutex<HashMap<String, Vec<String>>>>,
173 unreachable: Arc<Mutex<HashSet<String>>>,
174 listings: Arc<AtomicUsize>,
175}
176
177impl Accounts {
178 fn publishing(identities: HashMap<String, Vec<String>>) -> Self {
179 Self {
180 identities: Arc::new(Mutex::new(identities)),
181 ..Self::default()
182 }
183 }
184
185 fn unreachable(self, dids: HashSet<String>) -> Self {
186 *self.unreachable.lock().unwrap() = dids;
187 self
188 }
189
190 fn restore(&self, did: &str) {
191 self.unreachable.lock().unwrap().remove(did);
192 }
193
194 fn publish(&self, did: &str, line: String) {
195 self.identities
196 .lock()
197 .unwrap()
198 .entry(did.to_string())
199 .or_default()
200 .push(line);
201 }
202
203 fn published_by(&self, did: &str) -> Vec<String> {
204 self.identities
205 .lock()
206 .unwrap()
207 .get(did)
208 .cloned()
209 .unwrap_or_default()
210 }
211
212 fn listings(&self) -> usize {
213 self.listings.load(Ordering::SeqCst)
214 }
215}
216
217fn multi_http(accounts: Accounts) -> impl knot_runtime::HttpTransport {
218 let signer = K256Signer::generate(&SeededEntropy::new(77));
219 FakeHttp::new(move |request| {
220 let host = request.url.host_str().unwrap_or_default().to_string();
221 if host == "plc.directory" {
222 let did = request.url.path().trim_start_matches('/').to_string();
223 return Ok(ok_body(did_document(&signer, &did, "https://pds.test")));
224 }
225 if host == "pds.test" {
226 let repo = request
227 .url
228 .query_pairs()
229 .find(|(key, _)| key == "repo")
230 .map(|(_, value)| value.into_owned())
231 .unwrap_or_default();
232 accounts.listings.fetch_add(1, Ordering::SeqCst);
233 if accounts.unreachable.lock().unwrap().contains(&repo) {
234 return Ok(server_error());
235 }
236 let lines = accounts.published_by(&repo);
237 let refs: Vec<&str> = lines.iter().map(String::as_str).collect();
238 return Ok(ok_body(list_records_body(&refs)));
239 }
240 Ok(not_found())
241 })
242}
243
244fn actor_for_seed(seed: u64) -> knot_types::ActorId {
245 knot_types::ActorId::from_secp256k1(
246 K256Signer::generate(&SeededEntropy::new(seed))
247 .public_key()
248 .as_bytes(),
249 )
250}
251
252struct Server {
253 _scan: TempDir,
254 layout: Layout,
255 repo_did: RepoDid,
256 port: u16,
257 events: Arc<knot_events::EventLog<ManualClock>>,
258}
259
260async fn spawn_server(
261 published_line: String,
262 max_pack_bytes: MaxWireBytes,
263) -> (Server, Arc<Index>) {
264 spawn_server_with(published_line, max_pack_bytes, true).await
265}
266
267async fn spawn_server_with(
268 published_line: String,
269 max_pack_bytes: MaxWireBytes,
270 warm: bool,
271) -> (Server, Arc<Index>) {
272 let (server, index, _, _) = spawn_server_core(
273 published_line,
274 max_pack_bytes,
275 ArchiveLimit::default(),
276 warm,
277 None,
278 )
279 .await;
280 (server, index)
281}
282
283async fn spawn_server_core(
284 published_line: String,
285 max_pack_bytes: MaxWireBytes,
286 archive_limit: ArchiveLimit,
287 warm: bool,
288 lfs: Option<knot_lfs::LfsHandle>,
289) -> (
290 Server,
291 Arc<Index>,
292 tokio_util::sync::CancellationToken,
293 tokio::task::JoinHandle<()>,
294) {
295 let scan = tempfile::tempdir().unwrap();
296 let meta_path = scan.path().join("meta");
297 Repo::create(&meta_path).unwrap();
298 let layout = Layout::new(scan.path().join("repos"));
299 let repo_did = RepoDid::new(REPO_DID).unwrap();
300 layout.create(&repo_did).unwrap();
301
302 let signer = K256Signer::generate(&SeededEntropy::new(2));
303 let meta = Repo::open(&meta_path).unwrap();
304 let store = CobStore::new(&meta);
305 let home = CobHome::from(&KnotId::new("did:web:nel.pet").unwrap());
306 let registry = store
307 .create(
308 &home,
309 &RegistryChange::Register(Registration {
310 owner: OwnerDid::new(OWNER_DID).unwrap(),
311 rkey: RepoRkey::new(REPO_NAME).unwrap(),
312 name: RepoName::new(REPO_NAME).unwrap(),
313 repo: repo_did.clone(),
314 created_at: UnixSeconds::new(1),
315 }),
316 &signer,
317 UnixSeconds::new(1),
318 )
319 .unwrap()
320 .object;
321
322 let tid_repo_did = RepoDid::new(TID_REPO_DID).unwrap();
323 layout.create(&tid_repo_did).unwrap();
324 store
325 .update(
326 &home,
327 registry,
328 &RegistryChange::Register(Registration {
329 owner: OwnerDid::new(OWNER_DID).unwrap(),
330 rkey: RepoRkey::new(TID_RKEY).unwrap(),
331 name: RepoName::new(TID_REPO_NAME).unwrap(),
332 repo: tid_repo_did,
333 created_at: UnixSeconds::new(2),
334 }),
335 &signer,
336 UnixSeconds::new(2),
337 )
338 .unwrap();
339
340 let index = Arc::new(Index::new(meta_path, layout.clone()));
341 if warm {
342 index.rebuild().unwrap();
343 }
344
345 let atproto = Arc::new(
346 Atproto::new(
347 fake_http(published_line),
348 ManualClock::new(UnixMicros::new(1_000_000_000)),
349 KnotId::new("did:web:nel.pet").unwrap(),
350 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
351 )
352 .with_dns(Arc::new(fake_dns())),
353 );
354
355 let key_dir = scan.path().join("hostkey");
356 std::fs::create_dir_all(&key_dir).unwrap();
357 let host_key = knot_ssh::load_or_create_host_key(&key_dir.join("host")).unwrap();
358
359 let events = Arc::new(knot_events::EventLog::new(
360 ManualClock::new(UnixMicros::new(1_000_000_000)),
361 knot_events::ReplayBounds::new(
362 knot_events::ReplayEvents::new(64).unwrap(),
363 knot_events::ReplayBytes::new(16 << 20).unwrap(),
364 ),
365 ));
366 let base = knot_ssh::SshState::new(knot_ssh::SshConfig {
367 layout: layout.clone(),
368 index: Arc::clone(&index),
369 atproto,
370 knot_actor: actor_for_seed(1),
371 events: Arc::clone(&events),
372 hostname: knot_types::KnotHostname::new("knot.test").unwrap(),
373 appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(),
374 admins: std::collections::BTreeSet::new(),
375 admission: knot_types::AdmissionPolicy::Closed,
376 max_pack_bytes,
377 archive_limit,
378 languages_push_budget: LanguagesPushBudget::new(std::time::Duration::from_secs(2)),
379 ci_logs: None,
380 });
381 let state = Arc::new(match lfs {
382 Some(handle) => base.with_lfs(handle, 2),
383 None => base,
384 });
385
386 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
387 let port = listener.local_addr().unwrap().port();
388 let shutdown = tokio_util::sync::CancellationToken::new();
389 let serve_task = tokio::spawn({
390 let shutdown = shutdown.clone();
391 async move {
392 let _ = knot_ssh::serve_drained(listener, host_key, state, shutdown).await;
393 }
394 });
395
396 (
397 Server {
398 _scan: scan,
399 layout,
400 repo_did,
401 port,
402 events,
403 },
404 index,
405 shutdown,
406 serve_task,
407 )
408}
409
410fn ssh_command(key_path: &str) -> String {
411 format!(
412 "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
413 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
414 )
415}
416
417async fn git_ssh(cwd: &Path, key: &str, args: &[&str]) -> (bool, String) {
418 let ssh = ssh_command(key);
419 let cwd = cwd.to_path_buf();
420 let owned: Vec<String> = args.iter().map(|arg| arg.to_string()).collect();
421 tokio::task::spawn_blocking(move || {
422 let argv: Vec<&str> = owned.iter().map(String::as_str).collect();
423 git(&cwd, &[("GIT_SSH_COMMAND", &ssh)], &argv)
424 })
425 .await
426 .unwrap()
427}
428
429async fn push(work: &Path, url: &str, key: &str, refspecs: &[&str]) -> (bool, String) {
430 let args: Vec<&str> = std::iter::once("push")
431 .chain(std::iter::once(url))
432 .chain(refspecs.iter().copied())
433 .collect();
434 git_ssh(work, key, &args).await
435}
436
437async fn clone(url: &str, key: &str, dest: &Path) -> (bool, String) {
438 git_ssh(
439 Path::new("/tmp"),
440 key,
441 &["clone", "-q", url, dest.to_str().unwrap()],
442 )
443 .await
444}
445
446fn seed_work(work: &Path) -> String {
447 std::fs::create_dir_all(work).unwrap();
448 git(work, &[], &["init", "-q", "-b", "main"]);
449 std::fs::write(work.join("README.md"), "hello over ssh\n").unwrap();
450 git(work, &[], &["add", "-A"]);
451 git(work, &[], &["commit", "-q", "-m", "initial"]);
452 let (ok, head) = git(work, &[], &["rev-parse", "HEAD"]);
453 assert!(ok);
454 head.trim().to_string()
455}
456
457fn seed_commits(work: &Path, count: usize) {
458 std::fs::create_dir_all(work).unwrap();
459 git(work, &[], &["init", "-q", "-b", "main"]);
460 (0..count).for_each(|i| {
461 std::fs::write(work.join("log.txt"), format!("line {i}\n")).unwrap();
462 git(work, &[], &["add", "-A"]);
463 git(work, &[], &["commit", "-q", "-m", &format!("c{i}")]);
464 });
465}
466
467fn seed_cob(work: &Path, signer_seed: u64, subject: &str, home: &CobHome) -> (Oid, String, String) {
468 let repo = Repo::open(work).unwrap();
469 let signer = K256Signer::generate(&SeededEntropy::new(signer_seed));
470 let created = CobStore::new(&repo)
471 .create(
472 home,
473 &MembersChange::Add(Grant {
474 subject: AccountDid::new(subject).unwrap(),
475 added_by: AccountDid::new(OWNER_DID).unwrap(),
476 created_at: UnixSeconds::new(1),
477 }),
478 &signer,
479 UnixSeconds::new(1),
480 )
481 .unwrap();
482 let cob_ref = format!(
483 "refs/cobs/sh.tangled.knot.member/{}",
484 created.object.oid().to_hex()
485 );
486 let spec = format!("{cob_ref}:{cob_ref}");
487 (created.tip.oid(), cob_ref, spec)
488}
489
490fn main_tip(layout: &Layout, repo: &RepoDid) -> Option<Oid> {
491 layout
492 .open(repo)
493 .unwrap()
494 .find_ref(&RefName::new("refs/heads/main").unwrap())
495 .unwrap()
496}
497
498fn ref_names(server: &Server) -> Vec<String> {
499 server
500 .layout
501 .open(&server.repo_did)
502 .unwrap()
503 .references()
504 .unwrap()
505 .iter()
506 .map(|record| record.name.as_str().to_string())
507 .collect()
508}
509
510fn replay_bounds() -> knot_events::ReplayBounds {
511 knot_events::ReplayBounds::new(
512 knot_events::ReplayEvents::new(32).unwrap(),
513 knot_events::ReplayBytes::new(16 << 20).unwrap(),
514 )
515}
516
517async fn poll_for_event(
518 events: &knot_events::EventLog<ManualClock>,
519 nsid: &str,
520) -> serde_json::Value {
521 for _ in 0..50 {
522 if let Some(payload) = events
523 .replay(knot_events::EventCursor::START, replay_bounds())
524 .events
525 .into_iter()
526 .find(|event| event.nsid == nsid)
527 .map(|event| serde_json::to_value(&*event).unwrap()["event"].clone())
528 {
529 return payload;
530 }
531 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
532 }
533 panic!("no {nsid} event was published within the polling window");
534}
535
536struct Fixture {
537 scratch: TempDir,
538 server: Server,
539 index: Arc<Index>,
540 key_path: String,
541 url: String,
542 work: PathBuf,
543}
544
545async fn fixture() -> Fixture {
546 fixture_with_archive_limit(ArchiveLimit::default()).await
547}
548
549async fn fixture_with_archive_limit(archive_limit: ArchiveLimit) -> Fixture {
550 let scratch = tempfile::tempdir().unwrap();
551 let (key_path, public_line) = keygen(scratch.path(), "client");
552 let (server, index, _, _) = spawn_server_core(
553 public_line,
554 MaxWireBytes::new(1 << 30),
555 archive_limit,
556 true,
557 None,
558 )
559 .await;
560 let url = format!(
561 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
562 server.port
563 );
564 let work = scratch.path().join("work");
565 Fixture {
566 scratch,
567 server,
568 index,
569 key_path,
570 url,
571 work,
572 }
573}
574
575fn fetch_main_exit(clone_dir: &Path, ssh: &str, extra_git: &[&str]) -> Option<i32> {
576 let mut args = vec!["-k", "3", "20", "git"];
577 args.extend_from_slice(extra_git);
578 args.extend_from_slice(&["fetch", "origin", "main"]);
579 Command::new("timeout")
580 .args(&args)
581 .current_dir(clone_dir)
582 .env("GIT_SSH_COMMAND", ssh)
583 .status()
584 .expect("timeout/git runs")
585 .code()
586}
587
588async fn incremental_fetch_exit(
589 seed_count: usize,
590 extra_git: &'static [&'static str],
591) -> Option<i32> {
592 let scratch = tempfile::tempdir().unwrap();
593 let (key_path, public_line) = keygen(scratch.path(), "client");
594 let (server, _index) = spawn_server(public_line, MaxWireBytes::new(1 << 30)).await;
595 let url = format!(
596 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
597 server.port
598 );
599
600 let work = scratch.path().join("work");
601 seed_commits(&work, seed_count);
602 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
603 assert!(ok, "seeding push must land:\n{out}");
604
605 let clone_dir = scratch.path().join("clone");
606 let (ok, out) = clone(&url, &key_path, &clone_dir).await;
607 assert!(ok, "clone over ssh must succeed:\n{out}");
608
609 git(
610 &work,
611 &[],
612 &["commit", "-q", "--allow-empty", "-m", "advance"],
613 );
614 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
615 assert!(ok, "advancing server tip must succeed:\n{out}");
616
617 let ssh = ssh_command(&key_path);
618 let exit = tokio::task::spawn_blocking(move || fetch_main_exit(&clone_dir, &ssh, extra_git))
619 .await
620 .unwrap();
621 drop(server);
622 exit
623}
624
625#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
626async fn incremental_fetch_over_ssh_completes() {
627 let cases: [(&'static [&'static str], &str); 2] = [
628 (
629 &["-c", "protocol.version=0"],
630 "diverged v0 fetch sends more than 32 haves and blocks on an ACK/NAK. Upload loop \
631 answers each have-batch flush with a NAK instead of waiting for done, so it never \
632 hangs",
633 ),
634 (
635 &[],
636 "git forwards GIT_PROTOCOL over ssh, so default fetch path negotiates with the v2 loop",
637 ),
638 ];
639 futures::stream::iter(cases)
640 .for_each(|(extra, rationale)| async move {
641 assert_eq!(
642 incremental_fetch_exit(50, extra).await,
643 Some(0),
644 "{rationale}"
645 );
646 })
647 .await;
648}
649
650#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
651async fn a_display_name_addresses_a_repo_whose_record_key_is_a_tid() {
652 let fx = fixture().await;
653 let head = seed_work(&fx.work);
654 let head_oid = Oid::from_hex(&head).unwrap();
655 let port = fx.server.port;
656 let target = RepoDid::new(TID_REPO_DID).unwrap();
657 let variants = [
658 format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/{TID_REPO_NAME}"),
659 format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/{TID_REPO_NAME}.git"),
660 format!("ssh://git@127.0.0.1:{port}/nel.pet/{TID_REPO_NAME}"),
661 format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/{TID_RKEY}"),
662 ];
663 let fx = &fx;
664 let target = ⌖
665 futures::stream::iter(variants)
666 .for_each(|url| async move {
667 let (ok, out) = push(&fx.work, &url, &fx.key_path, &["main"]).await;
668 assert!(
669 ok,
670 "a PDS-minted record key leaves the display name as the only human \
671 path, so {url} must resolve and push:\n{out}"
672 );
673 assert_eq!(
674 main_tip(&fx.server.layout, target),
675 Some(head_oid),
676 "{url}: pushed commit must be the named repository's main tip"
677 );
678 })
679 .await;
680
681 let (ok, out) = push(
682 &fx.work,
683 &format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/whelk"),
684 &fx.key_path,
685 &["main"],
686 )
687 .await;
688 assert!(
689 !ok,
690 "a segment matching neither a record key nor a name stays unresolvable:\n{out}"
691 );
692}
693
694#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
695async fn addressing_variants_land() {
696 let fx = fixture().await;
697 let head = seed_work(&fx.work);
698 let head_oid = Oid::from_hex(&head).unwrap();
699 let port = fx.server.port;
700 let variants = [
701 format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/{REPO_NAME}"),
702 format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/{REPO_NAME}.git"),
703 format!("ssh://git@127.0.0.1:{port}/nel.pet/{REPO_NAME}"),
704 format!("ssh://git@127.0.0.1:{port}/{REPO_DID}"),
705 ];
706 let fx = &fx;
707 futures::stream::iter(variants)
708 .for_each(|url| async move {
709 let (ok, out) = push(&fx.work, &url, &fx.key_path, &["main"]).await;
710 assert!(ok, "addressing {url} must resolve and push:\n{out}");
711 assert_eq!(
712 main_tip(&fx.server.layout, &fx.server.repo_did),
713 Some(head_oid),
714 "{url}: pushed commit must be the repository's main tip"
715 );
716 })
717 .await;
718}
719
720#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
721async fn a_push_while_the_index_is_warming_is_refused() {
722 let scratch = tempfile::tempdir().unwrap();
723 let (key_path, public_line) = keygen(scratch.path(), "client");
724 let (server, _index) = spawn_server_with(public_line, MaxWireBytes::new(1 << 30), false).await;
725 let url = format!(
726 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
727 server.port
728 );
729
730 let work = scratch.path().join("work");
731 seed_work(&work);
732 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
733 assert!(
734 !ok,
735 "warming index must fail closed at the SSH boundary:\n{out}"
736 );
737 assert_eq!(
738 main_tip(&server.layout, &server.repo_did),
739 None,
740 "no ref lands while index is warming"
741 );
742}
743
744#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
745async fn unresolvable_targets_refused() {
746 let fx = fixture().await;
747 seed_work(&fx.work);
748 let port = fx.server.port;
749
750 let bad_name = format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/conch");
751 let (ok, out) = push(&fx.work, &bad_name, &fx.key_path, &["main"]).await;
752 assert!(
753 !ok,
754 "owner/name with no registry entry must be rejected, not silently routed:\n{out}"
755 );
756
757 fx.server
758 .layout
759 .create(&RepoDid::new("did:plc:clam").unwrap())
760 .unwrap();
761 let ghost_url = format!("ssh://git@127.0.0.1:{port}/did:plc:clam");
762 let dest = fx.scratch.path().join("ghost");
763 let (ok, out) = clone(&ghost_url, &fx.key_path, &dest).await;
764 assert!(
765 !ok,
766 "repo present on disk but absent from registry mustn't be served by bare DID:\n{out}"
767 );
768}
769
770#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
771async fn an_authorized_push_over_ssh_succeeds_and_a_clone_reads_it_back() {
772 let fx = fixture().await;
773 let head = seed_work(&fx.work);
774
775 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
776 assert!(ok, "authorized push over ssh must succeed:\n{out}");
777 assert_eq!(
778 main_tip(&fx.server.layout, &fx.server.repo_did),
779 Some(Oid::from_hex(&head).unwrap()),
780 "pushed commit must be the repository's main tip"
781 );
782
783 let clone_dir = fx.scratch.path().join("clone");
784 let (ok, out) = clone(&fx.url, &fx.key_path, &clone_dir).await;
785 assert!(ok, "clone over ssh must succeed:\n{out}");
786 assert!(
787 clone_dir.join("README.md").exists(),
788 "clone must check out the pushed file"
789 );
790}
791
792#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
793async fn an_authorized_push_emits_a_ref_update_event() {
794 let fx = fixture().await;
795 let head = seed_work(&fx.work);
796
797 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
798 assert!(ok, "authorized push over ssh must succeed:\n{out}");
799
800 let event = poll_for_event(&fx.server.events, "sh.tangled.git.refUpdate").await;
801 assert_eq!(event["ref"], "refs/heads/main");
802 assert_eq!(event["newSha"], head);
803 assert_eq!(event["committerDid"], OWNER_DID);
804 assert_eq!(event["ownerDid"], OWNER_DID);
805 assert_eq!(event["meta"]["isDefaultRef"], true);
806}
807
808#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
809async fn a_client_requesting_ssh_compression_clones_an_incompressible_pack() {
810 let fx = fixture().await;
811 std::fs::create_dir_all(&fx.work).unwrap();
812 git(&fx.work, &[], &["init", "-q", "-b", "main"]);
813 let mut state = 0x9e3779b97f4a7c15u64;
814 let payload: Vec<u8> = std::iter::repeat_with(|| {
815 state ^= state << 13;
816 state ^= state >> 7;
817 state ^= state << 17;
818 state.to_le_bytes()
819 })
820 .take(32 * 1024)
821 .flatten()
822 .collect();
823 std::fs::write(fx.work.join("noise.bin"), &payload).unwrap();
824 git(&fx.work, &[], &["add", "-A"]);
825 git(&fx.work, &[], &["commit", "-q", "-m", "noise"]);
826
827 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
828 assert!(ok, "push must succeed:\n{out}");
829
830 let dest = fx.scratch.path().join("compressed-clone");
831 let ssh = format!("{} -o Compression=yes", ssh_command(&fx.key_path));
832 let url = fx.url.clone();
833 let dest_arg = dest.to_str().unwrap().to_string();
834 let (ok, out) = tokio::task::spawn_blocking(move || {
835 git(
836 Path::new("/tmp"),
837 &[("GIT_SSH_COMMAND", &ssh)],
838 &["clone", "-q", &url, &dest_arg],
839 )
840 })
841 .await
842 .unwrap();
843 assert!(
844 ok,
845 "clone with ssh compression requested must succeed:\n{out}"
846 );
847 assert_eq!(std::fs::read(dest.join("noise.bin")).unwrap(), payload);
848}
849
850#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
851async fn an_oversized_push_is_refused_at_the_ssh_boundary() {
852 let scratch = tempfile::tempdir().unwrap();
853 let (key_path, public_line) = keygen(scratch.path(), "client");
854 let (server, _index) = spawn_server(public_line, MaxWireBytes::new(64)).await;
855 let url = format!(
856 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
857 server.port
858 );
859
860 let work = scratch.path().join("work");
861 seed_work(&work);
862 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
863 assert!(
864 !ok,
865 "push larger than the configured limit must be refused:\n{out}"
866 );
867 assert!(
868 ref_names(&server).is_empty(),
869 "oversized push mustn't land any ref"
870 );
871}
872
873#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
874async fn an_up_to_date_push_over_ssh_is_accepted() {
875 let fx = fixture().await;
876 seed_work(&fx.work);
877
878 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
879 assert!(ok, "first push must land:\n{out}");
880
881 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
882 assert!(
883 ok,
884 "up-to-date no-op push must succeed instead of failing with a stream error:\n{out}"
885 );
886 assert!(
887 out.contains("up-to-date") || out.contains("up to date"),
888 "git must report branch is up to date:\n{out}"
889 );
890}
891
892#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
893async fn a_denied_push_over_ssh_leaves_no_objects_in_the_live_odb() {
894 let scratch = tempfile::tempdir().unwrap();
895 let (_registered_path, registered_line) = keygen(scratch.path(), "registered");
896 let (attacker_path, _attacker_line) = keygen(scratch.path(), "attacker");
897 let (server, _index) = spawn_server(registered_line, MaxWireBytes::new(1 << 30)).await;
898 let url = format!(
899 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
900 server.port
901 );
902
903 let work = scratch.path().join("work");
904 let head = seed_work(&work);
905 let (ok, out) = push(&work, &url, &attacker_path, &["main"]).await;
906 assert!(!ok, "unauthorized push must be rejected:\n{out}");
907
908 let repo = server.layout.open(&server.repo_did).unwrap();
909 assert!(
910 repo.references().unwrap().is_empty(),
911 "denied push must create no ref"
912 );
913 assert!(
914 !repo.contains(Oid::from_hex(&head).unwrap()),
915 "denied push must migrate no objects into the live odb"
916 );
917}
918
919#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
920async fn ref_namespace_policy() {
921 let fx = fixture().await;
922 seed_work(&fx.work);
923
924 let (ok, out) = push(
925 &fx.work,
926 &fx.url,
927 &fx.key_path,
928 &["main:refs/hidden/feature/main"],
929 )
930 .await;
931 assert!(!ok, "push to refs/hidden/* must be rejected:\n{out}");
932 assert!(
933 ref_names(&fx.server).is_empty(),
934 "forbidden-ref push must land nothing"
935 );
936
937 let (ok, out) = push(
938 &fx.work,
939 &fx.url,
940 &fx.key_path,
941 &["main:refs/notes/commits"],
942 )
943 .await;
944 assert!(
945 ok,
946 "push to any non-reserved namespace must be accepted:\n{out}"
947 );
948 assert!(
949 ref_names(&fx.server)
950 .iter()
951 .any(|name| name == "refs/notes/commits"),
952 "pushed ref must land"
953 );
954}
955
956#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
957async fn cob_ref_guard_lifecycle() {
958 let fx = fixture().await;
959 seed_work(&fx.work);
960 let home = CobHome::from(&RepoDid::new(REPO_DID).unwrap());
961 let foreign = CobHome::from(&RepoDid::new("did:plc:whelk").unwrap());
962
963 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
964 assert!(ok, "head must land for the advertisement check:\n{out}");
965
966 let (owned_tip, owned_ref, owned_spec) = seed_cob(&fx.work, 1, "did:plc:limpet", &home);
967 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[owned_spec.as_str()]).await;
968 assert!(
969 ok,
970 "COB ref signed by the repository key must verify and land over ssh:\n{out}"
971 );
972
973 let (_forged_tip, forged_ref, forged_spec) = seed_cob(&fx.work, 9, "did:plc:whelk", &home);
974 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[forged_spec.as_str()]).await;
975 assert!(
976 !ok,
977 "COB ref signed by a stranger must be refused at the receive boundary:\n{out}"
978 );
979
980 let (_transplant_tip, transplant_ref, transplant_spec) =
981 seed_cob(&fx.work, 1, "did:plc:mussel", &foreign);
982 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[transplant_spec.as_str()]).await;
983 assert!(
984 !ok,
985 "same key signing for another repo's home must be refused on transplant:\n{out}"
986 );
987
988 let landed = ref_names(&fx.server);
989 assert!(
990 landed.contains(&owned_ref),
991 "owner-signed COB ref must be stored: {landed:?}"
992 );
993 assert!(
994 !landed.contains(&forged_ref),
995 "stranger-signed COB ref must be absent: {landed:?}"
996 );
997 assert!(
998 !landed.contains(&transplant_ref),
999 "transplanted COB ref must be absent: {landed:?}"
1000 );
1001
1002 let cob_name = RefName::new(&owned_ref).unwrap();
1003 let del = format!(":{owned_ref}");
1004 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[del.as_str()]).await;
1005 assert!(!ok, "deleting a COB ref must be refused:\n{out}");
1006 assert!(
1007 out.contains("append-only"),
1008 "rejection must name the append-only rule:\n{out}"
1009 );
1010 assert!(
1011 ref_names(&fx.server).contains(&owned_ref),
1012 "COB ref must survive the refused delete"
1013 );
1014
1015 let repo = Repo::open(&fx.work).unwrap();
1016 CobStore::new(&repo)
1017 .update(
1018 &home,
1019 knot_types::CobId::new(owned_tip),
1020 &MembersChange::Add(Grant {
1021 subject: AccountDid::new("did:plc:bailey").unwrap(),
1022 added_by: AccountDid::new(OWNER_DID).unwrap(),
1023 created_at: UnixSeconds::new(2),
1024 }),
1025 &K256Signer::generate(&SeededEntropy::new(1)),
1026 UnixSeconds::new(2),
1027 )
1028 .unwrap();
1029 assert_ne!(
1030 repo.find_ref(&cob_name).unwrap(),
1031 Some(owned_tip),
1032 "local COB ref now points at a new, equally valid tip"
1033 );
1034 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[owned_spec.as_str()]).await;
1035 assert!(
1036 !ok,
1037 "re-pushing a moved COB ref must be refused instead of silently clobbered:\n{out}"
1038 );
1039 assert_eq!(
1040 fx.server
1041 .layout
1042 .open(&fx.server.repo_did)
1043 .unwrap()
1044 .find_ref(&cob_name)
1045 .unwrap(),
1046 Some(owned_tip),
1047 "live COB ref must still point at the original tip"
1048 );
1049
1050 let (ok, advert) = git_ssh(Path::new("/tmp"), &fx.key_path, &["ls-remote", &fx.url]).await;
1051 assert!(ok, "ls-remote over ssh must succeed:\n{advert}");
1052 assert!(
1053 advert.contains("refs/heads/main"),
1054 "head must be advertised:\n{advert}"
1055 );
1056 assert!(
1057 !advert.contains("refs/cobs/"),
1058 "no refs/cobs/* may leak into the ssh advertisement:\n{advert}"
1059 );
1060}
1061
1062#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1063async fn a_filled_key_set_refuses_an_unregistered_key_and_an_acl_write_reopens_the_check() {
1064 let fx = fixture().await;
1065 let head = seed_work(&fx.work);
1066 let head_oid = Oid::from_hex(&head).unwrap();
1067
1068 fx.index.keys().mark_ready(fx.index.generation());
1069 fx.index.refresh_members().unwrap();
1070 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1071 assert!(
1072 ok,
1073 "a grant written after the key set was read must reopen the check, or whoever it \
1074 grants is refused at the handshake until the next fill pass:\n{out}"
1075 );
1076 assert_eq!(
1077 main_tip(&fx.server.layout, &fx.server.repo_did),
1078 Some(head_oid)
1079 );
1080
1081 fx.index.keys().record(
1082 &AccountDid::new(OWNER_DID).unwrap(),
1083 vec![knot_types::OfferedKey::from_bytes(registered_blob(&fx))],
1084 forever(),
1085 );
1086 fx.index.keys().mark_ready(fx.index.generation());
1087
1088 let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
1089 let two_ids = format!(
1090 "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
1091 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
1092 fx.key_path
1093 );
1094 let (ok, out) = {
1095 let (work, url) = (fx.work.clone(), fx.url.clone());
1096 tokio::task::spawn_blocking(move || {
1097 git(
1098 &work,
1099 &[("GIT_SSH_COMMAND", &two_ids)],
1100 &["push", "-q", &url, "main:refs/heads/second"],
1101 )
1102 })
1103 .await
1104 .unwrap()
1105 };
1106 assert!(
1107 ok,
1108 "a filled key set refuses the unregistered key, so the client offers its registered key \
1109 without the url identifying anybody:\n{out}"
1110 );
1111 assert_eq!(
1112 fx.server
1113 .layout
1114 .open(&fx.server.repo_did)
1115 .unwrap()
1116 .find_ref(&RefName::new("refs/heads/second").unwrap())
1117 .unwrap(),
1118 Some(head_oid)
1119 );
1120}
1121
1122#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1123async fn key_recognition_edge_cases() {
1124 let fx = fixture().await;
1125 let head = seed_work(&fx.work);
1126 let head_oid = Oid::from_hex(&head).unwrap();
1127
1128 let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
1129 let two_ids = format!(
1130 "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
1131 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
1132 fx.key_path
1133 );
1134 let (ok, out) = {
1135 let (work, url) = (fx.work.clone(), fx.url.clone());
1136 tokio::task::spawn_blocking(move || {
1137 git(
1138 &work,
1139 &[("GIT_SSH_COMMAND", &two_ids)],
1140 &["push", "-q", &url, "main"],
1141 )
1142 })
1143 .await
1144 .unwrap()
1145 };
1146 assert!(
1147 !ok,
1148 "with the key set still filling, the push is checked against whichever key the client \
1149 offers first:\n{out}"
1150 );
1151 assert!(
1152 out.contains("@nel.pet"),
1153 "refusal lists who may push, so the pusher knows which key to offer:\n{out}"
1154 );
1155 assert!(
1156 out.contains("IdentitiesOnly"),
1157 "refusal states how a multi-key client can offer its registered key:\n{out}"
1158 );
1159 assert_eq!(
1160 main_tip(&fx.server.layout, &fx.server.repo_did),
1161 None,
1162 "the refused push leaves the repo empty"
1163 );
1164
1165 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1166 assert!(ok, "offering only the registered key must succeed:\n{out}");
1167 assert_eq!(
1168 main_tip(&fx.server.layout, &fx.server.repo_did),
1169 Some(head_oid)
1170 );
1171
1172 let blob = russh::keys::ssh_key::PublicKey::from_openssh(
1173 &std::fs::read_to_string(fx.scratch.path().join("client.pub")).unwrap(),
1174 )
1175 .unwrap()
1176 .to_bytes()
1177 .unwrap();
1178 fx.index.keys().record(
1179 &AccountDid::new("did:plc:cuttle").unwrap(),
1180 vec![knot_types::OfferedKey::from_bytes(blob)],
1181 forever(),
1182 );
1183 let (ok, out) = push(
1184 &fx.work,
1185 &fx.url,
1186 &fx.key_path,
1187 &["main:refs/heads/squat-check"],
1188 )
1189 .await;
1190 assert!(
1191 ok,
1192 "stranger who published the owner's key mustn't deny the owner's push:\n{out}"
1193 );
1194 assert_eq!(
1195 fx.server
1196 .layout
1197 .open(&fx.server.repo_did)
1198 .unwrap()
1199 .find_ref(&RefName::new("refs/heads/squat-check").unwrap())
1200 .unwrap(),
1201 Some(head_oid)
1202 );
1203}
1204
1205#[test]
1206fn a_group_or_other_readable_host_key_is_refused_on_load() {
1207 use std::os::unix::fs::PermissionsExt;
1208 let dir = tempfile::tempdir().unwrap();
1209 let path = dir.path().join("host");
1210 knot_ssh::load_or_create_host_key(&path).unwrap();
1211 assert_eq!(
1212 std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
1213 0o600,
1214 "freshly created host key is 0600"
1215 );
1216
1217 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
1218 let refused = knot_ssh::load_or_create_host_key(&path);
1219 assert!(
1220 matches!(refused, Err(knot_ssh::SshError::HostKey { .. })),
1221 "world-readable existing host key must be refused on load: {refused:?}"
1222 );
1223}
1224
1225async fn launch(host_key_dir: &Path, layout: Layout, index: Arc<Index>, accounts: Accounts) -> u16 {
1226 let atproto = Arc::new(Atproto::new(
1227 multi_http(accounts),
1228 ManualClock::new(UnixMicros::new(1_000_000_000)),
1229 KnotId::new("did:web:nel.pet").unwrap(),
1230 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
1231 ));
1232 std::fs::create_dir_all(host_key_dir).unwrap();
1233 let host_key = knot_ssh::load_or_create_host_key(&host_key_dir.join("host")).unwrap();
1234 let events = Arc::new(knot_events::EventLog::new(
1235 ManualClock::new(UnixMicros::new(1_000_000_000)),
1236 knot_events::ReplayBounds::new(
1237 knot_events::ReplayEvents::new(64).unwrap(),
1238 knot_events::ReplayBytes::new(16 << 20).unwrap(),
1239 ),
1240 ));
1241 let state = Arc::new(knot_ssh::SshState::new(knot_ssh::SshConfig {
1242 layout,
1243 index,
1244 atproto,
1245 knot_actor: actor_for_seed(77),
1246 events,
1247 hostname: knot_types::KnotHostname::new("knot.test").unwrap(),
1248 appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(),
1249 admins: std::collections::BTreeSet::new(),
1250 admission: knot_types::AdmissionPolicy::Closed,
1251 max_pack_bytes: MaxWireBytes::new(1 << 30),
1252 archive_limit: ArchiveLimit::default(),
1253 languages_push_budget: LanguagesPushBudget::new(std::time::Duration::from_secs(2)),
1254 ci_logs: None,
1255 }));
1256 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1257 let port = listener.local_addr().unwrap().port();
1258 tokio::spawn(async move {
1259 let _ = knot_ssh::serve_on_socket(listener, host_key, state).await;
1260 });
1261 port
1262}
1263
1264#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1265async fn a_handle_in_the_url_identifies_a_visitor_and_lets_a_multi_key_client_find_its_key() {
1266 let fx = fixture().await;
1267 let head = seed_work(&fx.work);
1268 let head_oid = Oid::from_hex(&head).unwrap();
1269
1270 let port = fx.server.port;
1271 let greeted_key = fx.key_path.clone();
1272 let (_ok, out) =
1273 tokio::task::spawn_blocking(move || ssh_bare_as(&greeted_key, "nel.pet", port))
1274 .await
1275 .unwrap();
1276 assert!(
1277 out.contains("@nel.pet"),
1278 "an asserted handle identifies the visitor on first contact, with an empty cache:\n{out}"
1279 );
1280
1281 let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
1282 let two_ids = format!(
1283 "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
1284 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
1285 fx.key_path
1286 );
1287 let identified = format!("ssh://nel.pet@127.0.0.1:{}/{REPO_DID}", fx.server.port);
1288 let (ok, out) = {
1289 let (work, url) = (fx.work.clone(), identified.clone());
1290 tokio::task::spawn_blocking(move || {
1291 git(
1292 &work,
1293 &[("GIT_SSH_COMMAND", &two_ids)],
1294 &["push", "-q", &url, "main"],
1295 )
1296 })
1297 .await
1298 .unwrap()
1299 };
1300 assert!(
1301 ok,
1302 "a handle in the url lets the knot refuse the unregistered key so the client offers the \
1303 next key:\n{out}"
1304 );
1305 assert_eq!(
1306 main_tip(&fx.server.layout, &fx.server.repo_did),
1307 Some(head_oid)
1308 );
1309 assert_eq!(
1310 fx.index.owner_of_key(
1311 &knot_types::OfferedKey::from_bytes(registered_blob(&fx)),
1312 UnixSeconds::new(0),
1313 ),
1314 Resolved::Ready(None),
1315 "an asserted handle is whatever the client typed, so the keys read for it mustn't enter \
1316 the set, or anyone can fill the key budget by asserting handles"
1317 );
1318}
1319
1320fn registered_blob(fx: &Fixture) -> Vec<u8> {
1321 russh::keys::ssh_key::PublicKey::from_openssh(
1322 &std::fs::read_to_string(fx.scratch.path().join("client.pub")).unwrap(),
1323 )
1324 .unwrap()
1325 .to_bytes()
1326 .unwrap()
1327}
1328
1329fn registered_index(
1330 scratch: &TempDir,
1331 budget: knot_index::KeyBudget,
1332) -> (Layout, RepoDid, Arc<Index>) {
1333 let meta_path = scratch.path().join("meta");
1334 Repo::create(&meta_path).unwrap();
1335 let layout = Layout::new(scratch.path().join("repos"));
1336 let repo_did = RepoDid::new(REPO_DID).unwrap();
1337 layout.create(&repo_did).unwrap();
1338
1339 let signer = K256Signer::generate(&SeededEntropy::new(2));
1340 let meta = Repo::open(&meta_path).unwrap();
1341 CobStore::new(&meta)
1342 .create(
1343 &CobHome::from(&KnotId::new("did:web:nel.pet").unwrap()),
1344 &RegistryChange::Register(Registration {
1345 owner: OwnerDid::new(OWNER_DID).unwrap(),
1346 rkey: RepoRkey::new(REPO_NAME).unwrap(),
1347 name: RepoName::new(REPO_NAME).unwrap(),
1348 repo: repo_did.clone(),
1349 created_at: UnixSeconds::new(1),
1350 }),
1351 &signer,
1352 UnixSeconds::new(1),
1353 )
1354 .unwrap();
1355
1356 let index = Arc::new(Index::with_key_budget(meta_path, layout.clone(), budget));
1357 index.rebuild().unwrap();
1358 index.warm_collaborators();
1359 (layout, repo_did, index)
1360}
1361
1362#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1363async fn an_unreachable_pds_reads_as_transient_and_its_keys_get_in_once_it_recovers() {
1364 let scratch = tempfile::tempdir().unwrap();
1365 let (stale_key, stale_line) = keygen(scratch.path(), "stale");
1366 let (fresh_key, fresh_line) = keygen(scratch.path(), "fresh");
1367 let (layout, repo_did, index) = registered_index(&scratch, knot_index::KeyBudget::DEFAULT);
1368
1369 let accounts = Accounts::publishing(HashMap::from([(OWNER_DID.to_string(), vec![stale_line])]))
1370 .unreachable(HashSet::from([OWNER_DID.to_string()]));
1371 let port = launch(
1372 &scratch.path().join("hostkey"),
1373 layout.clone(),
1374 Arc::clone(&index),
1375 accounts.clone(),
1376 )
1377 .await;
1378 let url = format!("ssh://git@127.0.0.1:{port}/{REPO_DID}");
1379
1380 let work = scratch.path().join("work");
1381 let head = seed_work(&work);
1382 let (ok, out) = push(&work, &url, &stale_key, &["main"]).await;
1383 assert!(
1384 !ok,
1385 "a push mustn't be accepted while the owner's records are unreadable:\n{out}"
1386 );
1387 assert!(
1388 out.contains("retry shortly"),
1389 "an unreadable PDS must read as transient:\n{out}"
1390 );
1391 assert!(
1392 !out.contains("doesn't match"),
1393 "a transient failure mustn't be reported to the pusher as a wrong key:\n{out}"
1394 );
1395 assert_eq!(
1396 main_tip(&layout, &repo_did),
1397 None,
1398 "the refused push leaves the repo empty"
1399 );
1400
1401 accounts.restore(OWNER_DID);
1402 let (ok, out) = push(&work, &url, &stale_key, &["main"]).await;
1403 assert!(
1404 ok,
1405 "the key the owner publishes must push once its PDS answers again:\n{out}"
1406 );
1407
1408 accounts.publish(OWNER_DID, fresh_line);
1409 let (ok, out) = push(&work, &url, &fresh_key, &["main", "--force"]).await;
1410 assert!(
1411 ok,
1412 "a key the owner published after the knot last read the account must get in on the next \
1413 push, or publishing a second key locks its owner out until a fill pass catches up:\n{out}"
1414 );
1415 assert_eq!(
1416 main_tip(&layout, &repo_did),
1417 Some(Oid::from_hex(&head).unwrap())
1418 );
1419}
1420
1421#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1422async fn an_account_the_budget_couldnt_fit_still_clears_the_handshake_and_pushes() {
1423 let scratch = tempfile::tempdir().unwrap();
1424 let (owner_key, owner_line) = keygen(scratch.path(), "owner");
1425 let (layout, repo_did, index) =
1426 registered_index(&scratch, knot_index::KeyBudget::from_bytes(200));
1427
1428 let blob = russh::keys::ssh_key::PublicKey::from_openssh(&owner_line)
1429 .unwrap()
1430 .to_bytes()
1431 .unwrap();
1432 assert_eq!(
1433 index.keys().record(
1434 &AccountDid::new(OWNER_DID).unwrap(),
1435 vec![knot_types::OfferedKey::from_bytes(blob)],
1436 forever(),
1437 ),
1438 knot_index::KeyRecord::Unheld,
1439 "a 200-byte budget records the read without keeping the key"
1440 );
1441 index.keys().mark_ready(index.generation());
1442
1443 let port = launch(
1444 &scratch.path().join("hostkey"),
1445 layout.clone(),
1446 Arc::clone(&index),
1447 Accounts::publishing(HashMap::from([(OWNER_DID.to_string(), vec![owner_line])])),
1448 )
1449 .await;
1450 let url = format!("ssh://git@127.0.0.1:{port}/{REPO_DID}");
1451
1452 let work = scratch.path().join("work");
1453 let head = seed_work(&work);
1454 let (ok, out) = push(&work, &url, &owner_key, &["main"]).await;
1455 assert!(
1456 ok,
1457 "the set can't fit the owner's keys, so the handshake must defer to the push check \
1458 instead of refusing a key the accounts on file don't publish:\n{out}"
1459 );
1460 assert_eq!(
1461 main_tip(&layout, &repo_did),
1462 Some(Oid::from_hex(&head).unwrap())
1463 );
1464}
1465
1466#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1467async fn a_collaborator_pushes_its_repo_but_is_denied_on_a_repo_it_doesnt_collaborate_on() {
1468 const REPO_A: &str = "did:plc:squid";
1469 const REPO_B: &str = "did:plc:clam";
1470 const OWNER: &str = "did:plc:nel";
1471 const COLLAB: &str = "did:plc:olaren";
1472
1473 let scratch = tempfile::tempdir().unwrap();
1474 let (owner_key, owner_line) = keygen(scratch.path(), "owner");
1475 let (collab_key, collab_line) = keygen(scratch.path(), "collab");
1476
1477 let meta_path = scratch.path().join("meta");
1478 Repo::create(&meta_path).unwrap();
1479 let layout = Layout::new(scratch.path().join("repos"));
1480 let repo_a = RepoDid::new(REPO_A).unwrap();
1481 let repo_b = RepoDid::new(REPO_B).unwrap();
1482 let git_a = layout.create(&repo_a).unwrap();
1483 layout.create(&repo_b).unwrap();
1484
1485 let signer = K256Signer::generate(&SeededEntropy::new(2));
1486 let meta = Repo::open(&meta_path).unwrap();
1487 let store = CobStore::new(&meta);
1488 let knot_home = CobHome::from(&KnotId::new("did:web:nel.pet").unwrap());
1489 let reg = store
1490 .create(
1491 &knot_home,
1492 &RegistryChange::Register(Registration {
1493 owner: OwnerDid::new(OWNER).unwrap(),
1494 rkey: RepoRkey::new("anemone").unwrap(),
1495 name: RepoName::new("anemone").unwrap(),
1496 repo: repo_a.clone(),
1497 created_at: UnixSeconds::new(1),
1498 }),
1499 &signer,
1500 UnixSeconds::new(1),
1501 )
1502 .unwrap();
1503 store
1504 .update(
1505 &knot_home,
1506 reg.object,
1507 &RegistryChange::Register(Registration {
1508 owner: OwnerDid::new(OWNER).unwrap(),
1509 rkey: RepoRkey::new("barnacle").unwrap(),
1510 name: RepoName::new("barnacle").unwrap(),
1511 repo: repo_b.clone(),
1512 created_at: UnixSeconds::new(2),
1513 }),
1514 &signer,
1515 UnixSeconds::new(2),
1516 )
1517 .unwrap();
1518 store
1519 .create(
1520 &knot_home,
1521 &MembersChange::Add(Grant {
1522 subject: AccountDid::new(COLLAB).unwrap(),
1523 added_by: AccountDid::new(OWNER).unwrap(),
1524 created_at: UnixSeconds::new(1),
1525 }),
1526 &signer,
1527 UnixSeconds::new(1),
1528 )
1529 .unwrap();
1530 CobStore::new(&git_a)
1531 .create(
1532 &CobHome::from(&repo_a),
1533 &CollaboratorsChange::Add(Grant {
1534 subject: AccountDid::new(COLLAB).unwrap(),
1535 added_by: AccountDid::new(OWNER).unwrap(),
1536 created_at: UnixSeconds::new(1),
1537 }),
1538 &signer,
1539 UnixSeconds::new(1),
1540 )
1541 .unwrap();
1542
1543 let index = Arc::new(Index::new(meta_path, layout.clone()));
1544 index.rebuild().unwrap();
1545 index.warm_collaborators();
1546
1547 let identities = HashMap::from([
1548 (OWNER.to_string(), vec![owner_line]),
1549 (COLLAB.to_string(), vec![collab_line]),
1550 ]);
1551 let accounts = Accounts::publishing(identities);
1552 let port = launch(
1553 &scratch.path().join("hostkey"),
1554 layout.clone(),
1555 Arc::clone(&index),
1556 accounts.clone(),
1557 )
1558 .await;
1559
1560 let work_a = scratch.path().join("work_a");
1561 let head_a = seed_work(&work_a);
1562 let url_a = format!("ssh://git@127.0.0.1:{port}/{REPO_A}");
1563 let (ok, out) = push(&work_a, &url_a, &collab_key, &["main"]).await;
1564 assert!(
1565 ok,
1566 "collaborator must push the repo it collaborates on:\n{out}"
1567 );
1568 assert_eq!(
1569 main_tip(&layout, &repo_a),
1570 Some(Oid::from_hex(&head_a).unwrap()),
1571 "collaborator's commit must be repo A's main tip"
1572 );
1573
1574 let work_b = scratch.path().join("work_b");
1575 seed_work(&work_b);
1576 let url_b = format!("ssh://git@127.0.0.1:{port}/{REPO_B}");
1577 let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await;
1578 assert!(
1579 !denied,
1580 "a key that pushes repo A must be denied on repo B, where its owner was never granted:\n{out}"
1581 );
1582 assert!(
1583 main_tip(&layout, &repo_b).is_none(),
1584 "denied cross-repo push must land nothing on repo B"
1585 );
1586
1587 let after_first_denial = accounts.listings();
1588 let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await;
1589 assert!(!denied, "the second attempt is denied the same way:\n{out}");
1590 assert_eq!(
1591 accounts.listings(),
1592 after_first_denial,
1593 "repo B's owner was read during the first denial and is on file, so retrying mustn't \
1594 read that PDS again, or anyone with a key can make the knot fetch from a third party \
1595 at will:\n{out}"
1596 );
1597
1598 let work_owner = scratch.path().join("work_owner_b");
1599 let head_owner = seed_work(&work_owner);
1600 let (ok, out) = push(&work_owner, &url_b, &owner_key, &["main"]).await;
1601 assert!(ok, "owner must push to repo B:\n{out}");
1602 assert_eq!(
1603 main_tip(&layout, &repo_b),
1604 Some(Oid::from_hex(&head_owner).unwrap()),
1605 "owner's push to repo B must land, isolating the collaborator's denial as authorization"
1606 );
1607}
1608
1609fn ssh_bare(key_path: &str, port: u16) -> (bool, String) {
1610 ssh_bare_as(key_path, "git", port)
1611}
1612
1613fn ssh_bare_as(key_path: &str, user: &str, port: u16) -> (bool, String) {
1614 let out = Command::new("ssh")
1615 .args([
1616 "-i",
1617 key_path,
1618 "-o",
1619 "IdentitiesOnly=yes",
1620 "-o",
1621 "StrictHostKeyChecking=no",
1622 "-o",
1623 "UserKnownHostsFile=/dev/null",
1624 "-o",
1625 "PreferredAuthentications=publickey",
1626 "-o",
1627 "BatchMode=yes",
1628 "-p",
1629 &port.to_string(),
1630 &format!("{user}@127.0.0.1"),
1631 ])
1632 .output()
1633 .expect("ssh runs");
1634 (
1635 out.status.success(),
1636 format!(
1637 "{}{}",
1638 String::from_utf8_lossy(&out.stdout),
1639 String::from_utf8_lossy(&out.stderr)
1640 ),
1641 )
1642}
1643
1644#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1645async fn a_bare_ssh_session_greets_a_visitor_then_identifies_them_once_they_have_pushed() {
1646 let fx = fixture().await;
1647 let port = fx.server.port;
1648
1649 let key_path = fx.key_path.clone();
1650 let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1651 .await
1652 .unwrap();
1653 assert!(out.contains("knot.test"), "greeting names the knot:\n{out}");
1654 assert!(
1655 out.contains("ssh key"),
1656 "a visitor the knot can't identify yet learns what a push needs:\n{out}"
1657 );
1658
1659 seed_work(&fx.work);
1660 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1661 assert!(ok, "seeding main must succeed:\n{out}");
1662
1663 let key_path = fx.key_path.clone();
1664 let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1665 .await
1666 .unwrap();
1667 assert!(
1668 out.contains("@nel.pet"),
1669 "a push teaches the knot the key, so the next greeting uses the handle:\n{out}"
1670 );
1671}
1672
1673#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1674async fn a_push_to_a_new_branch_offers_a_pull_request_link() {
1675 let fx = fixture().await;
1676 seed_work(&fx.work);
1677
1678 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1679 assert!(ok, "seeding main must land:\n{out}");
1680
1681 git(&fx.work, &[], &["checkout", "-q", "-b", "feature"]);
1682 std::fs::write(fx.work.join("feature.txt"), "work\n").unwrap();
1683 git(&fx.work, &[], &["add", "-A"]);
1684 git(&fx.work, &[], &["commit", "-q", "-m", "feature work"]);
1685
1686 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["feature"]).await;
1687 assert!(ok, "feature-branch push must land:\n{out}");
1688 assert!(
1689 out.contains("https://tangled.test/nel.pet/anemone/pulls/new"),
1690 "new non-default branch is answered with a pull-request link:\n{out}"
1691 );
1692 assert!(
1693 out.contains("sourceBranch=feature") && out.contains("targetBranch=main"),
1694 "link points the new branch at the default:\n{out}"
1695 );
1696}
1697
1698#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1699async fn a_verbose_ci_push_option_reports_a_clean_pipeline() {
1700 let fx = fixture().await;
1701 std::fs::create_dir_all(fx.work.join(".tangled/workflows")).unwrap();
1702 git(&fx.work, &[], &["init", "-q", "-b", "main"]);
1703 std::fs::write(
1704 fx.work.join(".tangled/workflows/ci.yml"),
1705 "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n",
1706 )
1707 .unwrap();
1708 git(&fx.work, &[], &["add", "-A"]);
1709 git(&fx.work, &[], &["commit", "-q", "-m", "add ci"]);
1710
1711 let (ok, out) = push(
1712 &fx.work,
1713 &fx.url,
1714 &fx.key_path,
1715 &["--push-option=verbose-ci", "main"],
1716 )
1717 .await;
1718 assert!(ok, "push with a push option must land:\n{out}");
1719 assert!(
1720 out.contains("no diagnostics"),
1721 "verbose-ci reports clean compile over the sideband:\n{out}"
1722 );
1723}
1724
1725#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1726async fn git_archive_remote_over_ssh_streams_a_tar_of_the_tree() {
1727 let fx = fixture().await;
1728 seed_work(&fx.work);
1729 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1730 assert!(ok, "the seeding push must succeed before archiving:\n{out}");
1731
1732 let out_tar = fx.scratch.path().join("archive.tar");
1733 let (ok, out) = git_ssh(
1734 &fx.work,
1735 &fx.key_path,
1736 &[
1737 "archive",
1738 "--format=tar",
1739 "--remote",
1740 &fx.url,
1741 "-o",
1742 out_tar.to_str().unwrap(),
1743 "HEAD",
1744 ],
1745 )
1746 .await;
1747 assert!(ok, "git archive --remote over ssh must succeed:\n{out}");
1748
1749 let tar = std::fs::read(&out_tar).unwrap();
1750 assert!(
1751 knot_fixtures::contains(&tar, b"README.md"),
1752 "archived tar must contain the README.md entry"
1753 );
1754}
1755
1756#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1757async fn git_archive_remote_over_ssh_honors_the_configured_archive_limit() {
1758 let fx = fixture_with_archive_limit(ArchiveLimit::new(512)).await;
1759 seed_work(&fx.work);
1760 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1761 assert!(ok, "the seeding push must succeed before archiving:\n{out}");
1762
1763 let out_tar = fx.scratch.path().join("archive.tar");
1764 let (ok, out) = git_ssh(
1765 &fx.work,
1766 &fx.key_path,
1767 &[
1768 "archive",
1769 "--format=tar",
1770 "--remote",
1771 &fx.url,
1772 "-o",
1773 out_tar.to_str().unwrap(),
1774 "HEAD",
1775 ],
1776 )
1777 .await;
1778 assert!(!ok, "git archive --remote past the limit must fail:\n{out}");
1779 assert!(
1780 out.contains("archive exceeds the 512 byte limit"),
1781 "the refusal must reach the client over the ssh channel:\n{out}"
1782 );
1783}
1784
1785fn pkt(payload: &[u8]) -> Vec<u8> {
1786 let mut framed = format!("{:04x}", payload.len() + 4).into_bytes();
1787 framed.extend_from_slice(payload);
1788 framed
1789}
1790
1791fn pkt_text(line: &str) -> Vec<u8> {
1792 pkt(format!("{line}\n").as_bytes())
1793}
1794
1795fn read_until(reader: &mut impl std::io::Read, needle: &[u8], buffer: &mut Vec<u8>) {
1796 std::iter::from_fn(|| {
1797 let mut byte = [0u8; 1];
1798 match reader.read(&mut byte) {
1799 Ok(0) | Err(_) => None,
1800 Ok(_) => {
1801 buffer.push(byte[0]);
1802 Some(buffer.ends_with(needle))
1803 }
1804 }
1805 })
1806 .find(|done| *done)
1807 .expect("the session must answer before closing the stream");
1808}
1809
1810fn trickled_lfs_upload(
1811 key_path: &str,
1812 port: u16,
1813 body: &[u8],
1814 oid: &str,
1815 midway: std::sync::mpsc::Sender<()>,
1816) -> (bool, String) {
1817 use std::io::Write;
1818 let mut child = Command::new("ssh")
1819 .args([
1820 "-i",
1821 key_path,
1822 "-o",
1823 "IdentitiesOnly=yes",
1824 "-o",
1825 "StrictHostKeyChecking=no",
1826 "-o",
1827 "UserKnownHostsFile=/dev/null",
1828 "-o",
1829 "PreferredAuthentications=publickey",
1830 "-o",
1831 "BatchMode=yes",
1832 "-p",
1833 &port.to_string(),
1834 "git@127.0.0.1",
1835 &format!("git-lfs-transfer '{OWNER_DID}/{REPO_NAME}' upload"),
1836 ])
1837 .stdin(std::process::Stdio::piped())
1838 .stdout(std::process::Stdio::piped())
1839 .stderr(std::process::Stdio::null())
1840 .spawn()
1841 .expect("ssh runs");
1842 let mut stdin = child.stdin.take().unwrap();
1843 let mut stdout = child.stdout.take().unwrap();
1844 let mut transcript = Vec::new();
1845
1846 read_until(&mut stdout, b"version=1\n0000", &mut transcript);
1847
1848 let (first, second) = body.split_at(body.len() / 2);
1849 stdin
1850 .write_all(&pkt_text(&format!("put-object {oid}")))
1851 .unwrap();
1852 stdin
1853 .write_all(&pkt_text(&format!("size={}", body.len())))
1854 .unwrap();
1855 stdin.write_all(b"0001").unwrap();
1856 first.chunks(32 * 1024).for_each(|chunk| {
1857 stdin.write_all(&pkt(chunk)).unwrap();
1858 });
1859 stdin.flush().unwrap();
1860 midway.send(()).unwrap();
1861 std::thread::sleep(std::time::Duration::from_millis(900));
1862
1863 second.chunks(32 * 1024).for_each(|chunk| {
1864 stdin.write_all(&pkt(chunk)).unwrap();
1865 });
1866 stdin.write_all(b"0000").unwrap();
1867 stdin.flush().unwrap();
1868 read_until(&mut stdout, b"status 200\n0000", &mut transcript);
1869
1870 stdin.write_all(&pkt_text("quit")).unwrap();
1871 stdin.write_all(b"0000").unwrap();
1872 stdin.flush().unwrap();
1873 drop(stdin);
1874 use std::io::Read;
1875 let _ = stdout.read_to_end(&mut transcript);
1876 let status = child.wait().expect("ssh exits");
1877 (
1878 status.success(),
1879 String::from_utf8_lossy(&transcript).into_owned(),
1880 )
1881}
1882
1883#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1884async fn shutdown_drains_an_in_flight_lfs_transfer_before_exit() {
1885 use knot_lfs::LfsStore;
1886 use sha2::Digest;
1887 let scratch = tempfile::tempdir().unwrap();
1888 let (key_path, public_line) = keygen(scratch.path(), "drain");
1889 let lfs_dir = scratch.path().join("lfs");
1890 std::fs::create_dir_all(&lfs_dir).unwrap();
1891 let handle = knot_lfs::LfsHandle::open(
1892 knot_lfs::LfsStorePath::new(&lfs_dir),
1893 knot_lfs::LfsSize::new(1 << 30),
1894 knot_lfs::FreeSpaceFloor::new(0),
1895 )
1896 .unwrap();
1897 let (server, _index, shutdown, serve_task) = spawn_server_core(
1898 public_line,
1899 MaxWireBytes::new(1 << 20),
1900 ArchiveLimit::default(),
1901 true,
1902 Some(handle.clone()),
1903 )
1904 .await;
1905
1906 let body: Vec<u8> = (0..1_048_576u32).map(|n| (n % 251) as u8).collect();
1907 let oid = knot_lfs::LfsOid::from_digest(sha2::Sha256::digest(&body).into());
1908 let (midway_tx, midway_rx) = std::sync::mpsc::channel();
1909
1910 let client = {
1911 let key_path = key_path.clone();
1912 let oid = oid.clone();
1913 let port = server.port;
1914 tokio::task::spawn_blocking(move || {
1915 trickled_lfs_upload(&key_path, port, &body, oid.as_str(), midway_tx)
1916 })
1917 };
1918
1919 tokio::task::spawn_blocking(move || {
1920 midway_rx
1921 .recv_timeout(std::time::Duration::from_secs(20))
1922 .expect("the upload must reach its midway point")
1923 })
1924 .await
1925 .unwrap();
1926
1927 shutdown.cancel();
1928 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
1929 assert!(
1930 !serve_task.is_finished(),
1931 "the listener must keep draining while a transfer is in flight"
1932 );
1933
1934 let (ok, transcript) = client.await.unwrap();
1935 assert!(
1936 ok,
1937 "the in-flight upload must finish cleanly across the shutdown:\n{transcript}"
1938 );
1939 assert!(
1940 transcript.contains("status 200"),
1941 "the server must acknowledge the drained upload:\n{transcript}"
1942 );
1943
1944 tokio::time::timeout(std::time::Duration::from_secs(10), serve_task)
1945 .await
1946 .expect("the drained listener must exit promptly once transfers finish")
1947 .unwrap();
1948
1949 let repo_did = RepoDid::new(REPO_DID).unwrap();
1950 assert_eq!(
1951 handle
1952 .store
1953 .probe(&repo_did, &oid)
1954 .unwrap()
1955 .map(|size| size.get()),
1956 Some(1_048_576),
1957 "the drained upload must be durable"
1958 );
1959
1960 let (connected, _) = {
1961 let key_path = key_path.clone();
1962 let port = server.port;
1963 tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1964 .await
1965 .unwrap()
1966 };
1967 assert!(
1968 !connected,
1969 "a connection after shutdown must be refused, the drain only covers in-flight work"
1970 );
1971}