alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::HashMap;
2use std::path::{Path, PathBuf};
3use std::process::Command;
4use std::sync::Arc;
5
6use futures::stream::StreamExt;
7use knot_atproto::Atproto;
8use knot_cob::{CobHome, CobStore};
9use knot_cobs::{CollaboratorsChange, Grant, MembersChange, Registration, RegistryChange};
10use knot_git::{Layout, Repo};
11use knot_index::Index;
12use knot_pack::MaxWireBytes;
13use knot_postreceive::LanguagesPushBudget;
14use knot_runtime::{
15 FakeDns, FakeHttp, HttpResponse, K256Signer, ManualClock, SeededEntropy, Signer, UnixMicros,
16};
17use knot_types::{
18 AccountDid, KnotId, Oid, OwnerDid, RefName, RepoDid, RepoName, RepoRkey, UnixSeconds,
19};
20use tempfile::TempDir;
21use tokio::net::TcpListener;
22use url::Url;
23
24const REPO_DID: &str = "did:plc:squid";
25const REPO_NAME: &str = "anemone";
26const OWNER_DID: &str = "did:plc:nel";
27const PDS_HOST: &str = "pds.oyster.cafe";
28
29fn git(cwd: &Path, env: &[(&str, &str)], args: &[&str]) -> (bool, String) {
30 let mut command = knot_fixtures::command(cwd);
31 command.args(args);
32 env.iter().for_each(|(key, value)| {
33 command.env(key, value);
34 });
35 let out = command.output().expect("git runs");
36 let combined = format!(
37 "{}{}",
38 String::from_utf8_lossy(&out.stdout),
39 String::from_utf8_lossy(&out.stderr)
40 );
41 (out.status.success(), combined)
42}
43
44fn keygen(dir: &Path, name: &str) -> (String, String) {
45 let path = dir.join(name);
46 let out = Command::new("ssh-keygen")
47 .args([
48 "-t",
49 "ed25519",
50 "-N",
51 "",
52 "-C",
53 "nel@oyster.cafe",
54 "-f",
55 path.to_str().unwrap(),
56 ])
57 .output()
58 .expect("ssh-keygen runs");
59 assert!(
60 out.status.success(),
61 "ssh-keygen failed: {}",
62 String::from_utf8_lossy(&out.stderr)
63 );
64 let public_line = std::fs::read_to_string(dir.join(format!("{name}.pub")))
65 .unwrap()
66 .trim()
67 .to_string();
68 (path.to_str().unwrap().to_string(), public_line)
69}
70
71fn did_document(signer: &K256Signer, did: &str, pds: &str) -> Vec<u8> {
72 let multikey = knot_types::crypto::multikey(0xe7, signer.public_key().as_bytes());
73 serde_json::to_vec(&serde_json::json!({
74 "id": did,
75 "alsoKnownAs": ["at://nel.pet"],
76 "verificationMethod": [{
77 "id": format!("{did}#atproto"),
78 "type": "Multikey",
79 "controller": did,
80 "publicKeyMultibase": multikey
81 }],
82 "service": [{
83 "id": "#atproto_pds",
84 "type": "AtprotoPersonalDataServer",
85 "serviceEndpoint": pds
86 }]
87 }))
88 .unwrap()
89}
90
91fn list_records_body(lines: &[&str]) -> Vec<u8> {
92 let records: Vec<_> = lines
93 .iter()
94 .map(|line| {
95 serde_json::json!({
96 "value": {
97 "$type": "sh.tangled.publicKey",
98 "key": line,
99 "name": "laptop",
100 "createdAt": "2026-06-08T00:00:00Z"
101 }
102 })
103 })
104 .collect();
105 serde_json::to_vec(&serde_json::json!({ "records": records })).unwrap()
106}
107
108fn ok_body(body: Vec<u8>) -> HttpResponse {
109 HttpResponse {
110 status: http::StatusCode::OK,
111 headers: http::HeaderMap::new(),
112 body: bytes::Bytes::from(body),
113 }
114}
115
116fn fake_dns() -> impl knot_runtime::DnsTxtResolver {
117 FakeDns::new(|name: &str| {
118 Ok(match name {
119 "_atproto.nel.pet" => vec![format!("did={OWNER_DID}")],
120 _ => Vec::new(),
121 })
122 })
123}
124
125fn not_found() -> HttpResponse {
126 HttpResponse {
127 status: http::StatusCode::NOT_FOUND,
128 headers: http::HeaderMap::new(),
129 body: bytes::Bytes::new(),
130 }
131}
132
133fn fake_http(published_line: String) -> impl knot_runtime::HttpTransport {
134 let signer = K256Signer::generate(&SeededEntropy::new(1));
135 let pds = format!("https://{PDS_HOST}");
136 FakeHttp::new(move |request| {
137 let host = request.url.host_str().unwrap_or_default().to_string();
138 let path = request.url.path().to_string();
139 let body = if host == PDS_HOST {
140 list_records_body(&[&published_line])
141 } else if path.ends_with(REPO_DID) {
142 did_document(&signer, REPO_DID, &pds)
143 } else if path.ends_with(OWNER_DID) {
144 did_document(&signer, OWNER_DID, &pds)
145 } else {
146 return Ok(not_found());
147 };
148 Ok(ok_body(body))
149 })
150}
151
152fn multi_http(identities: HashMap<String, Vec<String>>) -> impl knot_runtime::HttpTransport {
153 let signer = K256Signer::generate(&SeededEntropy::new(77));
154 FakeHttp::new(move |request| {
155 let host = request.url.host_str().unwrap_or_default().to_string();
156 if host == "plc.directory" {
157 let did = request.url.path().trim_start_matches('/').to_string();
158 return Ok(ok_body(did_document(&signer, &did, "https://pds.test")));
159 }
160 if host == "pds.test" {
161 let repo = request
162 .url
163 .query_pairs()
164 .find(|(key, _)| key == "repo")
165 .map(|(_, value)| value.into_owned())
166 .unwrap_or_default();
167 let lines = identities.get(&repo).cloned().unwrap_or_default();
168 let refs: Vec<&str> = lines.iter().map(String::as_str).collect();
169 return Ok(ok_body(list_records_body(&refs)));
170 }
171 Ok(not_found())
172 })
173}
174
175fn actor_for_seed(seed: u64) -> knot_types::ActorId {
176 knot_types::ActorId::from_secp256k1(
177 K256Signer::generate(&SeededEntropy::new(seed))
178 .public_key()
179 .as_bytes(),
180 )
181}
182
183struct Server {
184 _scan: TempDir,
185 layout: Layout,
186 repo_did: RepoDid,
187 port: u16,
188 events: Arc<knot_events::EventLog<ManualClock>>,
189}
190
191async fn spawn_server(
192 published_line: String,
193 max_pack_bytes: MaxWireBytes,
194) -> (Server, Arc<Index>) {
195 spawn_server_with(published_line, max_pack_bytes, true).await
196}
197
198async fn spawn_server_with(
199 published_line: String,
200 max_pack_bytes: MaxWireBytes,
201 warm: bool,
202) -> (Server, Arc<Index>) {
203 let (server, index, _, _) = spawn_server_core(published_line, max_pack_bytes, warm, None).await;
204 (server, index)
205}
206
207async fn spawn_server_core(
208 published_line: String,
209 max_pack_bytes: MaxWireBytes,
210 warm: bool,
211 lfs: Option<knot_lfs::LfsHandle>,
212) -> (
213 Server,
214 Arc<Index>,
215 tokio_util::sync::CancellationToken,
216 tokio::task::JoinHandle<()>,
217) {
218 let scan = tempfile::tempdir().unwrap();
219 let meta_path = scan.path().join("meta");
220 Repo::create(&meta_path).unwrap();
221 let layout = Layout::new(scan.path().join("repos"));
222 let repo_did = RepoDid::new(REPO_DID).unwrap();
223 layout.create(&repo_did).unwrap();
224
225 let signer = K256Signer::generate(&SeededEntropy::new(2));
226 let meta = Repo::open(&meta_path).unwrap();
227 let store = CobStore::new(&meta);
228 store
229 .create(
230 &CobHome::from(&KnotId::new("did:web:nel.pet").unwrap()),
231 &RegistryChange::Register(Registration {
232 owner: OwnerDid::new(OWNER_DID).unwrap(),
233 rkey: RepoRkey::new(REPO_NAME).unwrap(),
234 name: RepoName::new(REPO_NAME).unwrap(),
235 repo: repo_did.clone(),
236 created_at: UnixSeconds::new(1),
237 }),
238 &signer,
239 UnixSeconds::new(1),
240 )
241 .unwrap();
242
243 let index = Arc::new(Index::new(meta_path, layout.clone()));
244 if warm {
245 index.rebuild().unwrap();
246 }
247
248 let atproto = Arc::new(
249 Atproto::new(
250 fake_http(published_line),
251 ManualClock::new(UnixMicros::new(1_000_000_000)),
252 KnotId::new("did:web:nel.pet").unwrap(),
253 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
254 )
255 .with_dns(Arc::new(fake_dns())),
256 );
257
258 let key_dir = scan.path().join("hostkey");
259 std::fs::create_dir_all(&key_dir).unwrap();
260 let host_key = knot_ssh::load_or_create_host_key(&key_dir.join("host")).unwrap();
261
262 let events = Arc::new(knot_events::EventLog::new(
263 ManualClock::new(UnixMicros::new(1_000_000_000)),
264 knot_events::ReplayBounds::new(
265 knot_events::ReplayEvents::new(64).unwrap(),
266 knot_events::ReplayBytes::new(16 << 20).unwrap(),
267 ),
268 ));
269 let base = knot_ssh::SshState::new(
270 layout.clone(),
271 Arc::clone(&index),
272 atproto,
273 actor_for_seed(1),
274 Arc::clone(&events),
275 knot_types::KnotHostname::new("knot.test").unwrap(),
276 knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(),
277 std::collections::BTreeSet::new(),
278 knot_types::AdmissionPolicy::Closed,
279 max_pack_bytes,
280 LanguagesPushBudget::new(std::time::Duration::from_secs(2)),
281 None,
282 );
283 let state = Arc::new(match lfs {
284 Some(handle) => base.with_lfs(handle, 2),
285 None => base,
286 });
287
288 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
289 let port = listener.local_addr().unwrap().port();
290 let shutdown = tokio_util::sync::CancellationToken::new();
291 let serve_task = tokio::spawn({
292 let shutdown = shutdown.clone();
293 async move {
294 let _ = knot_ssh::serve_drained(listener, host_key, state, shutdown).await;
295 }
296 });
297
298 (
299 Server {
300 _scan: scan,
301 layout,
302 repo_did,
303 port,
304 events,
305 },
306 index,
307 shutdown,
308 serve_task,
309 )
310}
311
312fn ssh_command(key_path: &str) -> String {
313 format!(
314 "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
315 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
316 )
317}
318
319async fn git_ssh(cwd: &Path, key: &str, args: &[&str]) -> (bool, String) {
320 let ssh = ssh_command(key);
321 let cwd = cwd.to_path_buf();
322 let owned: Vec<String> = args.iter().map(|arg| arg.to_string()).collect();
323 tokio::task::spawn_blocking(move || {
324 let argv: Vec<&str> = owned.iter().map(String::as_str).collect();
325 git(&cwd, &[("GIT_SSH_COMMAND", &ssh)], &argv)
326 })
327 .await
328 .unwrap()
329}
330
331async fn push(work: &Path, url: &str, key: &str, refspecs: &[&str]) -> (bool, String) {
332 let args: Vec<&str> = std::iter::once("push")
333 .chain(std::iter::once(url))
334 .chain(refspecs.iter().copied())
335 .collect();
336 git_ssh(work, key, &args).await
337}
338
339async fn clone(url: &str, key: &str, dest: &Path) -> (bool, String) {
340 git_ssh(
341 Path::new("/tmp"),
342 key,
343 &["clone", "-q", url, dest.to_str().unwrap()],
344 )
345 .await
346}
347
348fn seed_work(work: &Path) -> String {
349 std::fs::create_dir_all(work).unwrap();
350 git(work, &[], &["init", "-q", "-b", "main"]);
351 std::fs::write(work.join("README.md"), "hello over ssh\n").unwrap();
352 git(work, &[], &["add", "-A"]);
353 git(work, &[], &["commit", "-q", "-m", "initial"]);
354 let (ok, head) = git(work, &[], &["rev-parse", "HEAD"]);
355 assert!(ok);
356 head.trim().to_string()
357}
358
359fn seed_commits(work: &Path, count: usize) {
360 std::fs::create_dir_all(work).unwrap();
361 git(work, &[], &["init", "-q", "-b", "main"]);
362 (0..count).for_each(|i| {
363 std::fs::write(work.join("log.txt"), format!("line {i}\n")).unwrap();
364 git(work, &[], &["add", "-A"]);
365 git(work, &[], &["commit", "-q", "-m", &format!("c{i}")]);
366 });
367}
368
369fn seed_cob(work: &Path, signer_seed: u64, subject: &str, home: &CobHome) -> (Oid, String, String) {
370 let repo = Repo::open(work).unwrap();
371 let signer = K256Signer::generate(&SeededEntropy::new(signer_seed));
372 let created = CobStore::new(&repo)
373 .create(
374 home,
375 &MembersChange::Add(Grant {
376 subject: AccountDid::new(subject).unwrap(),
377 added_by: AccountDid::new(OWNER_DID).unwrap(),
378 created_at: UnixSeconds::new(1),
379 }),
380 &signer,
381 UnixSeconds::new(1),
382 )
383 .unwrap();
384 let cob_ref = format!(
385 "refs/cobs/sh.tangled.knot.member/{}",
386 created.object.oid().to_hex()
387 );
388 let spec = format!("{cob_ref}:{cob_ref}");
389 (created.tip.oid(), cob_ref, spec)
390}
391
392fn main_tip(layout: &Layout, repo: &RepoDid) -> Option<Oid> {
393 layout
394 .open(repo)
395 .unwrap()
396 .find_ref(&RefName::new("refs/heads/main").unwrap())
397 .unwrap()
398}
399
400fn ref_names(server: &Server) -> Vec<String> {
401 server
402 .layout
403 .open(&server.repo_did)
404 .unwrap()
405 .references()
406 .unwrap()
407 .iter()
408 .map(|record| record.name.as_str().to_string())
409 .collect()
410}
411
412fn replay_bounds() -> knot_events::ReplayBounds {
413 knot_events::ReplayBounds::new(
414 knot_events::ReplayEvents::new(32).unwrap(),
415 knot_events::ReplayBytes::new(16 << 20).unwrap(),
416 )
417}
418
419async fn poll_for_event(
420 events: &knot_events::EventLog<ManualClock>,
421 nsid: &str,
422) -> serde_json::Value {
423 for _ in 0..50 {
424 if let Some(payload) = events
425 .replay(knot_events::EventCursor::START, replay_bounds())
426 .events
427 .into_iter()
428 .find(|event| event.nsid == nsid)
429 .map(|event| serde_json::to_value(&*event).unwrap()["event"].clone())
430 {
431 return payload;
432 }
433 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
434 }
435 panic!("no {nsid} event was published within the polling window");
436}
437
438struct Fixture {
439 scratch: TempDir,
440 server: Server,
441 index: Arc<Index>,
442 key_path: String,
443 url: String,
444 work: PathBuf,
445}
446
447async fn fixture() -> Fixture {
448 let scratch = tempfile::tempdir().unwrap();
449 let (key_path, public_line) = keygen(scratch.path(), "client");
450 let (server, index) = spawn_server(public_line, MaxWireBytes::new(1 << 30)).await;
451 let url = format!(
452 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
453 server.port
454 );
455 let work = scratch.path().join("work");
456 Fixture {
457 scratch,
458 server,
459 index,
460 key_path,
461 url,
462 work,
463 }
464}
465
466fn fetch_main_exit(clone_dir: &Path, ssh: &str, extra_git: &[&str]) -> Option<i32> {
467 let mut args = vec!["-k", "3", "20", "git"];
468 args.extend_from_slice(extra_git);
469 args.extend_from_slice(&["fetch", "origin", "main"]);
470 Command::new("timeout")
471 .args(&args)
472 .current_dir(clone_dir)
473 .env("GIT_SSH_COMMAND", ssh)
474 .status()
475 .expect("timeout/git runs")
476 .code()
477}
478
479async fn incremental_fetch_exit(
480 seed_count: usize,
481 extra_git: &'static [&'static str],
482) -> Option<i32> {
483 let scratch = tempfile::tempdir().unwrap();
484 let (key_path, public_line) = keygen(scratch.path(), "client");
485 let (server, _index) = spawn_server(public_line, MaxWireBytes::new(1 << 30)).await;
486 let url = format!(
487 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
488 server.port
489 );
490
491 let work = scratch.path().join("work");
492 seed_commits(&work, seed_count);
493 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
494 assert!(ok, "seeding push must land:\n{out}");
495
496 let clone_dir = scratch.path().join("clone");
497 let (ok, out) = clone(&url, &key_path, &clone_dir).await;
498 assert!(ok, "clone over ssh must succeed:\n{out}");
499
500 git(
501 &work,
502 &[],
503 &["commit", "-q", "--allow-empty", "-m", "advance"],
504 );
505 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
506 assert!(ok, "advancing server tip must succeed:\n{out}");
507
508 let ssh = ssh_command(&key_path);
509 let exit = tokio::task::spawn_blocking(move || fetch_main_exit(&clone_dir, &ssh, extra_git))
510 .await
511 .unwrap();
512 drop(server);
513 exit
514}
515
516#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
517async fn incremental_fetch_over_ssh_completes() {
518 let cases: [(&'static [&'static str], &str); 2] = [
519 (
520 &["-c", "protocol.version=0"],
521 "diverged v0 fetch sends more than 32 haves and blocks on an ACK/NAK. Upload loop \
522 answers each have-batch flush with a NAK instead of waiting for done, so it never \
523 hangs",
524 ),
525 (
526 &[],
527 "git forwards GIT_PROTOCOL over ssh, so default fetch path negotiates with the v2 loop",
528 ),
529 ];
530 futures::stream::iter(cases)
531 .for_each(|(extra, rationale)| async move {
532 assert_eq!(
533 incremental_fetch_exit(50, extra).await,
534 Some(0),
535 "{rationale}"
536 );
537 })
538 .await;
539}
540
541#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
542async fn addressing_variants_land() {
543 let fx = fixture().await;
544 let head = seed_work(&fx.work);
545 let head_oid = Oid::from_hex(&head).unwrap();
546 let port = fx.server.port;
547 let variants = [
548 format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/{REPO_NAME}"),
549 format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/{REPO_NAME}.git"),
550 format!("ssh://git@127.0.0.1:{port}/nel.pet/{REPO_NAME}"),
551 format!("ssh://git@127.0.0.1:{port}/{REPO_DID}"),
552 ];
553 let fx = &fx;
554 futures::stream::iter(variants)
555 .for_each(|url| async move {
556 let (ok, out) = push(&fx.work, &url, &fx.key_path, &["main"]).await;
557 assert!(ok, "addressing {url} must resolve and push:\n{out}");
558 assert_eq!(
559 main_tip(&fx.server.layout, &fx.server.repo_did),
560 Some(head_oid),
561 "{url}: pushed commit must be the repository's main tip"
562 );
563 })
564 .await;
565}
566
567#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
568async fn a_push_while_the_index_is_warming_is_refused() {
569 let scratch = tempfile::tempdir().unwrap();
570 let (key_path, public_line) = keygen(scratch.path(), "client");
571 let (server, _index) = spawn_server_with(public_line, MaxWireBytes::new(1 << 30), false).await;
572 let url = format!(
573 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
574 server.port
575 );
576
577 let work = scratch.path().join("work");
578 seed_work(&work);
579 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
580 assert!(
581 !ok,
582 "warming index must fail closed at the SSH boundary:\n{out}"
583 );
584 assert_eq!(
585 main_tip(&server.layout, &server.repo_did),
586 None,
587 "no ref lands while index is warming"
588 );
589}
590
591#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
592async fn unresolvable_targets_refused() {
593 let fx = fixture().await;
594 seed_work(&fx.work);
595 let port = fx.server.port;
596
597 let bad_name = format!("ssh://git@127.0.0.1:{port}/{OWNER_DID}/conch");
598 let (ok, out) = push(&fx.work, &bad_name, &fx.key_path, &["main"]).await;
599 assert!(
600 !ok,
601 "owner/name with no registry entry must be rejected, not silently routed:\n{out}"
602 );
603
604 fx.server
605 .layout
606 .create(&RepoDid::new("did:plc:clam").unwrap())
607 .unwrap();
608 let ghost_url = format!("ssh://git@127.0.0.1:{port}/did:plc:clam");
609 let dest = fx.scratch.path().join("ghost");
610 let (ok, out) = clone(&ghost_url, &fx.key_path, &dest).await;
611 assert!(
612 !ok,
613 "repo present on disk but absent from registry mustn't be served by bare DID:\n{out}"
614 );
615}
616
617#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
618async fn an_authorized_push_over_ssh_succeeds_and_a_clone_reads_it_back() {
619 let fx = fixture().await;
620 let head = seed_work(&fx.work);
621
622 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
623 assert!(ok, "authorized push over ssh must succeed:\n{out}");
624 assert_eq!(
625 main_tip(&fx.server.layout, &fx.server.repo_did),
626 Some(Oid::from_hex(&head).unwrap()),
627 "pushed commit must be the repository's main tip"
628 );
629
630 let clone_dir = fx.scratch.path().join("clone");
631 let (ok, out) = clone(&fx.url, &fx.key_path, &clone_dir).await;
632 assert!(ok, "clone over ssh must succeed:\n{out}");
633 assert!(
634 clone_dir.join("README.md").exists(),
635 "clone must check out the pushed file"
636 );
637}
638
639#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
640async fn an_authorized_push_emits_a_ref_update_event() {
641 let fx = fixture().await;
642 let head = seed_work(&fx.work);
643
644 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
645 assert!(ok, "authorized push over ssh must succeed:\n{out}");
646
647 let event = poll_for_event(&fx.server.events, "sh.tangled.git.refUpdate").await;
648 assert_eq!(event["ref"], "refs/heads/main");
649 assert_eq!(event["newSha"], head);
650 assert_eq!(event["committerDid"], OWNER_DID);
651 assert_eq!(event["ownerDid"], OWNER_DID);
652 assert_eq!(event["meta"]["isDefaultRef"], true);
653}
654
655#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
656async fn an_oversized_push_is_refused_at_the_ssh_boundary() {
657 let scratch = tempfile::tempdir().unwrap();
658 let (key_path, public_line) = keygen(scratch.path(), "client");
659 let (server, _index) = spawn_server(public_line, MaxWireBytes::new(64)).await;
660 let url = format!(
661 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
662 server.port
663 );
664
665 let work = scratch.path().join("work");
666 seed_work(&work);
667 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
668 assert!(
669 !ok,
670 "push larger than the configured limit must be refused:\n{out}"
671 );
672 assert!(
673 ref_names(&server).is_empty(),
674 "oversized push mustn't land any ref"
675 );
676}
677
678#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
679async fn an_up_to_date_push_over_ssh_is_accepted() {
680 let fx = fixture().await;
681 seed_work(&fx.work);
682
683 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
684 assert!(ok, "first push must land:\n{out}");
685
686 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
687 assert!(
688 ok,
689 "up-to-date no-op push must succeed instead of failing with a stream error:\n{out}"
690 );
691 assert!(
692 out.contains("up-to-date") || out.contains("up to date"),
693 "git must report branch is up to date:\n{out}"
694 );
695}
696
697#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
698async fn a_denied_push_over_ssh_leaves_no_objects_in_the_live_odb() {
699 let scratch = tempfile::tempdir().unwrap();
700 let (_registered_path, registered_line) = keygen(scratch.path(), "registered");
701 let (attacker_path, _attacker_line) = keygen(scratch.path(), "attacker");
702 let (server, _index) = spawn_server(registered_line, MaxWireBytes::new(1 << 30)).await;
703 let url = format!(
704 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
705 server.port
706 );
707
708 let work = scratch.path().join("work");
709 let head = seed_work(&work);
710 let (ok, out) = push(&work, &url, &attacker_path, &["main"]).await;
711 assert!(!ok, "unauthorized push must be rejected:\n{out}");
712
713 let repo = server.layout.open(&server.repo_did).unwrap();
714 assert!(
715 repo.references().unwrap().is_empty(),
716 "denied push must create no ref"
717 );
718 assert!(
719 !repo.contains(Oid::from_hex(&head).unwrap()),
720 "denied push must migrate no objects into the live odb"
721 );
722}
723
724#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
725async fn ref_namespace_policy() {
726 let fx = fixture().await;
727 seed_work(&fx.work);
728
729 let (ok, out) = push(
730 &fx.work,
731 &fx.url,
732 &fx.key_path,
733 &["main:refs/hidden/feature/main"],
734 )
735 .await;
736 assert!(!ok, "push to refs/hidden/* must be rejected:\n{out}");
737 assert!(
738 ref_names(&fx.server).is_empty(),
739 "forbidden-ref push must land nothing"
740 );
741
742 let (ok, out) = push(
743 &fx.work,
744 &fx.url,
745 &fx.key_path,
746 &["main:refs/notes/commits"],
747 )
748 .await;
749 assert!(
750 ok,
751 "push to any non-reserved namespace must be accepted:\n{out}"
752 );
753 assert!(
754 ref_names(&fx.server)
755 .iter()
756 .any(|name| name == "refs/notes/commits"),
757 "pushed ref must land"
758 );
759}
760
761#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
762async fn cob_ref_guard_lifecycle() {
763 let fx = fixture().await;
764 seed_work(&fx.work);
765 let home = CobHome::from(&RepoDid::new(REPO_DID).unwrap());
766 let foreign = CobHome::from(&RepoDid::new("did:plc:whelk").unwrap());
767
768 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
769 assert!(ok, "head must land for the advertisement check:\n{out}");
770
771 let (owned_tip, owned_ref, owned_spec) = seed_cob(&fx.work, 1, "did:plc:limpet", &home);
772 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[owned_spec.as_str()]).await;
773 assert!(
774 ok,
775 "COB ref signed by the repository key must verify and land over ssh:\n{out}"
776 );
777
778 let (_forged_tip, forged_ref, forged_spec) = seed_cob(&fx.work, 9, "did:plc:whelk", &home);
779 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[forged_spec.as_str()]).await;
780 assert!(
781 !ok,
782 "COB ref signed by a stranger must be refused at the receive boundary:\n{out}"
783 );
784
785 let (_transplant_tip, transplant_ref, transplant_spec) =
786 seed_cob(&fx.work, 1, "did:plc:mussel", &foreign);
787 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[transplant_spec.as_str()]).await;
788 assert!(
789 !ok,
790 "same key signing for another repo's home must be refused on transplant:\n{out}"
791 );
792
793 let landed = ref_names(&fx.server);
794 assert!(
795 landed.contains(&owned_ref),
796 "owner-signed COB ref must be stored: {landed:?}"
797 );
798 assert!(
799 !landed.contains(&forged_ref),
800 "stranger-signed COB ref must be absent: {landed:?}"
801 );
802 assert!(
803 !landed.contains(&transplant_ref),
804 "transplanted COB ref must be absent: {landed:?}"
805 );
806
807 let cob_name = RefName::new(&owned_ref).unwrap();
808 let del = format!(":{owned_ref}");
809 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[del.as_str()]).await;
810 assert!(!ok, "deleting a COB ref must be refused:\n{out}");
811 assert!(
812 out.contains("append-only"),
813 "rejection must name the append-only rule:\n{out}"
814 );
815 assert!(
816 ref_names(&fx.server).contains(&owned_ref),
817 "COB ref must survive the refused delete"
818 );
819
820 let repo = Repo::open(&fx.work).unwrap();
821 CobStore::new(&repo)
822 .update(
823 &home,
824 knot_types::CobId::new(owned_tip),
825 &MembersChange::Add(Grant {
826 subject: AccountDid::new("did:plc:bailey").unwrap(),
827 added_by: AccountDid::new(OWNER_DID).unwrap(),
828 created_at: UnixSeconds::new(2),
829 }),
830 &K256Signer::generate(&SeededEntropy::new(1)),
831 UnixSeconds::new(2),
832 )
833 .unwrap();
834 assert_ne!(
835 repo.find_ref(&cob_name).unwrap(),
836 Some(owned_tip),
837 "local COB ref now points at a new, equally valid tip"
838 );
839 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[owned_spec.as_str()]).await;
840 assert!(
841 !ok,
842 "re-pushing a moved COB ref must be refused instead of silently clobbered:\n{out}"
843 );
844 assert_eq!(
845 fx.server
846 .layout
847 .open(&fx.server.repo_did)
848 .unwrap()
849 .find_ref(&cob_name)
850 .unwrap(),
851 Some(owned_tip),
852 "live COB ref must still point at the original tip"
853 );
854
855 let (ok, advert) = git_ssh(Path::new("/tmp"), &fx.key_path, &["ls-remote", &fx.url]).await;
856 assert!(ok, "ls-remote over ssh must succeed:\n{advert}");
857 assert!(
858 advert.contains("refs/heads/main"),
859 "head must be advertised:\n{advert}"
860 );
861 assert!(
862 !advert.contains("refs/cobs/"),
863 "no refs/cobs/* may leak into the ssh advertisement:\n{advert}"
864 );
865}
866
867#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
868async fn key_recognition_edge_cases() {
869 let fx = fixture().await;
870 let head = seed_work(&fx.work);
871 let head_oid = Oid::from_hex(&head).unwrap();
872
873 let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
874 let two_ids = format!(
875 "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
876 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
877 fx.key_path
878 );
879 let (ok, out) = {
880 let (work, url) = (fx.work.clone(), fx.url.clone());
881 tokio::task::spawn_blocking(move || {
882 git(
883 &work,
884 &[("GIT_SSH_COMMAND", &two_ids)],
885 &["push", "-q", &url, "main"],
886 )
887 })
888 .await
889 .unwrap()
890 };
891 assert!(
892 ok,
893 "rejecting unregistered key must let client cycle to the registered one:\n{out}"
894 );
895 assert_eq!(
896 main_tip(&fx.server.layout, &fx.server.repo_did),
897 Some(head_oid)
898 );
899
900 let blob = russh::keys::ssh_key::PublicKey::from_openssh(
901 &std::fs::read_to_string(fx.scratch.path().join("client.pub")).unwrap(),
902 )
903 .unwrap()
904 .to_bytes()
905 .unwrap();
906 fx.index.cache_key(
907 knot_types::OfferedKey::from_bytes(blob),
908 &AccountDid::new("did:plc:whelk").unwrap(),
909 );
910 let (ok, out) = push(
911 &fx.work,
912 &fx.url,
913 &fx.key_path,
914 &["main:refs/heads/squat-check"],
915 )
916 .await;
917 assert!(
918 ok,
919 "stranger who published the owner's key mustn't deny the owner's push:\n{out}"
920 );
921 assert_eq!(
922 fx.server
923 .layout
924 .open(&fx.server.repo_did)
925 .unwrap()
926 .find_ref(&RefName::new("refs/heads/squat-check").unwrap())
927 .unwrap(),
928 Some(head_oid)
929 );
930}
931
932#[test]
933fn a_group_or_other_readable_host_key_is_refused_on_load() {
934 use std::os::unix::fs::PermissionsExt;
935 let dir = tempfile::tempdir().unwrap();
936 let path = dir.path().join("host");
937 knot_ssh::load_or_create_host_key(&path).unwrap();
938 assert_eq!(
939 std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
940 0o600,
941 "freshly created host key is 0600"
942 );
943
944 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
945 let refused = knot_ssh::load_or_create_host_key(&path);
946 assert!(
947 matches!(refused, Err(knot_ssh::SshError::HostKey { .. })),
948 "world-readable existing host key must be refused on load: {refused:?}"
949 );
950}
951
952async fn launch(
953 host_key_dir: &Path,
954 layout: Layout,
955 index: Arc<Index>,
956 identities: HashMap<String, Vec<String>>,
957) -> u16 {
958 let atproto = Arc::new(Atproto::new(
959 multi_http(identities),
960 ManualClock::new(UnixMicros::new(1_000_000_000)),
961 KnotId::new("did:web:nel.pet").unwrap(),
962 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
963 ));
964 std::fs::create_dir_all(host_key_dir).unwrap();
965 let host_key = knot_ssh::load_or_create_host_key(&host_key_dir.join("host")).unwrap();
966 let events = Arc::new(knot_events::EventLog::new(
967 ManualClock::new(UnixMicros::new(1_000_000_000)),
968 knot_events::ReplayBounds::new(
969 knot_events::ReplayEvents::new(64).unwrap(),
970 knot_events::ReplayBytes::new(16 << 20).unwrap(),
971 ),
972 ));
973 let state = Arc::new(knot_ssh::SshState::new(
974 layout,
975 index,
976 atproto,
977 actor_for_seed(77),
978 events,
979 knot_types::KnotHostname::new("knot.test").unwrap(),
980 knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(),
981 std::collections::BTreeSet::new(),
982 knot_types::AdmissionPolicy::Closed,
983 MaxWireBytes::new(1 << 30),
984 LanguagesPushBudget::new(std::time::Duration::from_secs(2)),
985 None,
986 ));
987 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
988 let port = listener.local_addr().unwrap().port();
989 tokio::spawn(async move {
990 let _ = knot_ssh::serve_on_socket(listener, host_key, state).await;
991 });
992 port
993}
994
995#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
996async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo_it_has_no_grant_on()
997 {
998 const REPO_A: &str = "did:plc:squid";
999 const REPO_B: &str = "did:plc:clam";
1000 const OWNER: &str = "did:plc:nel";
1001 const COLLAB: &str = "did:plc:olaren";
1002
1003 let scratch = tempfile::tempdir().unwrap();
1004 let (owner_key, owner_line) = keygen(scratch.path(), "owner");
1005 let (collab_key, collab_line) = keygen(scratch.path(), "collab");
1006
1007 let meta_path = scratch.path().join("meta");
1008 Repo::create(&meta_path).unwrap();
1009 let layout = Layout::new(scratch.path().join("repos"));
1010 let repo_a = RepoDid::new(REPO_A).unwrap();
1011 let repo_b = RepoDid::new(REPO_B).unwrap();
1012 let git_a = layout.create(&repo_a).unwrap();
1013 layout.create(&repo_b).unwrap();
1014
1015 let signer = K256Signer::generate(&SeededEntropy::new(2));
1016 let meta = Repo::open(&meta_path).unwrap();
1017 let store = CobStore::new(&meta);
1018 let knot_home = CobHome::from(&KnotId::new("did:web:nel.pet").unwrap());
1019 let reg = store
1020 .create(
1021 &knot_home,
1022 &RegistryChange::Register(Registration {
1023 owner: OwnerDid::new(OWNER).unwrap(),
1024 rkey: RepoRkey::new("anemone").unwrap(),
1025 name: RepoName::new("anemone").unwrap(),
1026 repo: repo_a.clone(),
1027 created_at: UnixSeconds::new(1),
1028 }),
1029 &signer,
1030 UnixSeconds::new(1),
1031 )
1032 .unwrap();
1033 store
1034 .update(
1035 &knot_home,
1036 reg.object,
1037 &RegistryChange::Register(Registration {
1038 owner: OwnerDid::new(OWNER).unwrap(),
1039 rkey: RepoRkey::new("barnacle").unwrap(),
1040 name: RepoName::new("barnacle").unwrap(),
1041 repo: repo_b.clone(),
1042 created_at: UnixSeconds::new(2),
1043 }),
1044 &signer,
1045 UnixSeconds::new(2),
1046 )
1047 .unwrap();
1048 store
1049 .create(
1050 &knot_home,
1051 &MembersChange::Add(Grant {
1052 subject: AccountDid::new(COLLAB).unwrap(),
1053 added_by: AccountDid::new(OWNER).unwrap(),
1054 created_at: UnixSeconds::new(1),
1055 }),
1056 &signer,
1057 UnixSeconds::new(1),
1058 )
1059 .unwrap();
1060 CobStore::new(&git_a)
1061 .create(
1062 &CobHome::from(&repo_a),
1063 &CollaboratorsChange::Add(Grant {
1064 subject: AccountDid::new(COLLAB).unwrap(),
1065 added_by: AccountDid::new(OWNER).unwrap(),
1066 created_at: UnixSeconds::new(1),
1067 }),
1068 &signer,
1069 UnixSeconds::new(1),
1070 )
1071 .unwrap();
1072
1073 let index = Arc::new(Index::new(meta_path, layout.clone()));
1074 index.rebuild().unwrap();
1075 index.warm_collaborators();
1076
1077 let identities = HashMap::from([
1078 (OWNER.to_string(), vec![owner_line]),
1079 (COLLAB.to_string(), vec![collab_line]),
1080 ]);
1081 let port = launch(
1082 &scratch.path().join("hostkey"),
1083 layout.clone(),
1084 Arc::clone(&index),
1085 identities,
1086 )
1087 .await;
1088
1089 let work_a = scratch.path().join("work_a");
1090 let head_a = seed_work(&work_a);
1091 let url_a = format!("ssh://git@127.0.0.1:{port}/{REPO_A}");
1092 let (ok, out) = push(&work_a, &url_a, &collab_key, &["main"]).await;
1093 assert!(
1094 ok,
1095 "collaborator must push the repo it collaborates on:\n{out}"
1096 );
1097 assert_eq!(
1098 main_tip(&layout, &repo_a),
1099 Some(Oid::from_hex(&head_a).unwrap()),
1100 "collaborator's commit must be repo A's main tip"
1101 );
1102
1103 let work_b = scratch.path().join("work_b");
1104 seed_work(&work_b);
1105 let url_b = format!("ssh://git@127.0.0.1:{port}/{REPO_B}");
1106 let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await;
1107 assert!(
1108 !denied,
1109 "key recognized via repo A but with no grant on repo B must be denied, recognition is \
1110 not authorization:\n{out}"
1111 );
1112 assert!(
1113 main_tip(&layout, &repo_b).is_none(),
1114 "denied cross-repo push must land nothing on repo B"
1115 );
1116
1117 let work_owner = scratch.path().join("work_owner_b");
1118 let head_owner = seed_work(&work_owner);
1119 let (ok, out) = push(&work_owner, &url_b, &owner_key, &["main"]).await;
1120 assert!(ok, "owner must push to repo B:\n{out}");
1121 assert_eq!(
1122 main_tip(&layout, &repo_b),
1123 Some(Oid::from_hex(&head_owner).unwrap()),
1124 "owner's push to repo B must land, isolating the collaborator's denial as authorization"
1125 );
1126}
1127
1128fn ssh_bare(key_path: &str, port: u16) -> (bool, String) {
1129 let out = Command::new("ssh")
1130 .args([
1131 "-i",
1132 key_path,
1133 "-o",
1134 "IdentitiesOnly=yes",
1135 "-o",
1136 "StrictHostKeyChecking=no",
1137 "-o",
1138 "UserKnownHostsFile=/dev/null",
1139 "-o",
1140 "PreferredAuthentications=publickey",
1141 "-o",
1142 "BatchMode=yes",
1143 "-p",
1144 &port.to_string(),
1145 "git@127.0.0.1",
1146 ])
1147 .output()
1148 .expect("ssh runs");
1149 (
1150 out.status.success(),
1151 format!(
1152 "{}{}",
1153 String::from_utf8_lossy(&out.stdout),
1154 String::from_utf8_lossy(&out.stderr)
1155 ),
1156 )
1157}
1158
1159#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1160async fn a_bare_ssh_session_greets_the_recognized_user() {
1161 let fx = fixture().await;
1162 let port = fx.server.port;
1163 let key_path = fx.key_path.clone();
1164 let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1165 .await
1166 .unwrap();
1167 assert!(
1168 out.contains("@nel.pet"),
1169 "greeting resolves and addresses the user by handle:\n{out}"
1170 );
1171 assert!(out.contains("knot.test"), "greeting names the knot:\n{out}");
1172}
1173
1174#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1175async fn a_push_to_a_new_branch_offers_a_pull_request_link() {
1176 let fx = fixture().await;
1177 seed_work(&fx.work);
1178
1179 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1180 assert!(ok, "seeding main must land:\n{out}");
1181
1182 git(&fx.work, &[], &["checkout", "-q", "-b", "feature"]);
1183 std::fs::write(fx.work.join("feature.txt"), "work\n").unwrap();
1184 git(&fx.work, &[], &["add", "-A"]);
1185 git(&fx.work, &[], &["commit", "-q", "-m", "feature work"]);
1186
1187 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["feature"]).await;
1188 assert!(ok, "feature-branch push must land:\n{out}");
1189 assert!(
1190 out.contains("https://tangled.test/nel.pet/anemone/pulls/new"),
1191 "new non-default branch is answered with a pull-request link:\n{out}"
1192 );
1193 assert!(
1194 out.contains("sourceBranch=feature") && out.contains("targetBranch=main"),
1195 "link points the new branch at the default:\n{out}"
1196 );
1197}
1198
1199#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1200async fn a_verbose_ci_push_option_reports_a_clean_pipeline() {
1201 let fx = fixture().await;
1202 std::fs::create_dir_all(fx.work.join(".tangled/workflows")).unwrap();
1203 git(&fx.work, &[], &["init", "-q", "-b", "main"]);
1204 std::fs::write(
1205 fx.work.join(".tangled/workflows/ci.yml"),
1206 "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n",
1207 )
1208 .unwrap();
1209 git(&fx.work, &[], &["add", "-A"]);
1210 git(&fx.work, &[], &["commit", "-q", "-m", "add ci"]);
1211
1212 let (ok, out) = push(
1213 &fx.work,
1214 &fx.url,
1215 &fx.key_path,
1216 &["--push-option=verbose-ci", "main"],
1217 )
1218 .await;
1219 assert!(ok, "push with a push option must land:\n{out}");
1220 assert!(
1221 out.contains("no diagnostics"),
1222 "verbose-ci reports clean compile over the sideband:\n{out}"
1223 );
1224}
1225
1226#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1227async fn git_archive_remote_over_ssh_streams_a_tar_of_the_tree() {
1228 let fx = fixture().await;
1229 seed_work(&fx.work);
1230 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1231 assert!(ok, "seeding push must land before archiving:\n{out}");
1232
1233 let out_tar = fx.scratch.path().join("archive.tar");
1234 let (ok, out) = git_ssh(
1235 &fx.work,
1236 &fx.key_path,
1237 &[
1238 "archive",
1239 "--format=tar",
1240 "--remote",
1241 &fx.url,
1242 "-o",
1243 out_tar.to_str().unwrap(),
1244 "HEAD",
1245 ],
1246 )
1247 .await;
1248 assert!(ok, "git archive --remote over ssh must succeed:\n{out}");
1249
1250 let tar = std::fs::read(&out_tar).unwrap();
1251 assert!(
1252 tar.windows(b"README.md".len()).any(|w| w == b"README.md"),
1253 "archived tar must contain the README.md entry"
1254 );
1255}
1256
1257fn pkt(payload: &[u8]) -> Vec<u8> {
1258 let mut framed = format!("{:04x}", payload.len() + 4).into_bytes();
1259 framed.extend_from_slice(payload);
1260 framed
1261}
1262
1263fn pkt_text(line: &str) -> Vec<u8> {
1264 pkt(format!("{line}\n").as_bytes())
1265}
1266
1267fn read_until(reader: &mut impl std::io::Read, needle: &[u8], buffer: &mut Vec<u8>) {
1268 std::iter::from_fn(|| {
1269 let mut byte = [0u8; 1];
1270 match reader.read(&mut byte) {
1271 Ok(0) | Err(_) => None,
1272 Ok(_) => {
1273 buffer.push(byte[0]);
1274 Some(buffer.ends_with(needle))
1275 }
1276 }
1277 })
1278 .find(|done| *done)
1279 .expect("the session must answer before closing the stream");
1280}
1281
1282fn trickled_lfs_upload(
1283 key_path: &str,
1284 port: u16,
1285 body: &[u8],
1286 oid: &str,
1287 midway: std::sync::mpsc::Sender<()>,
1288) -> (bool, String) {
1289 use std::io::Write;
1290 let mut child = Command::new("ssh")
1291 .args([
1292 "-i",
1293 key_path,
1294 "-o",
1295 "IdentitiesOnly=yes",
1296 "-o",
1297 "StrictHostKeyChecking=no",
1298 "-o",
1299 "UserKnownHostsFile=/dev/null",
1300 "-o",
1301 "PreferredAuthentications=publickey",
1302 "-o",
1303 "BatchMode=yes",
1304 "-p",
1305 &port.to_string(),
1306 "git@127.0.0.1",
1307 &format!("git-lfs-transfer '{OWNER_DID}/{REPO_NAME}' upload"),
1308 ])
1309 .stdin(std::process::Stdio::piped())
1310 .stdout(std::process::Stdio::piped())
1311 .stderr(std::process::Stdio::null())
1312 .spawn()
1313 .expect("ssh runs");
1314 let mut stdin = child.stdin.take().unwrap();
1315 let mut stdout = child.stdout.take().unwrap();
1316 let mut transcript = Vec::new();
1317
1318 read_until(&mut stdout, b"version=1\n0000", &mut transcript);
1319
1320 let (first, second) = body.split_at(body.len() / 2);
1321 stdin
1322 .write_all(&pkt_text(&format!("put-object {oid}")))
1323 .unwrap();
1324 stdin
1325 .write_all(&pkt_text(&format!("size={}", body.len())))
1326 .unwrap();
1327 stdin.write_all(b"0001").unwrap();
1328 first.chunks(32 * 1024).for_each(|chunk| {
1329 stdin.write_all(&pkt(chunk)).unwrap();
1330 });
1331 stdin.flush().unwrap();
1332 midway.send(()).unwrap();
1333 std::thread::sleep(std::time::Duration::from_millis(900));
1334
1335 second.chunks(32 * 1024).for_each(|chunk| {
1336 stdin.write_all(&pkt(chunk)).unwrap();
1337 });
1338 stdin.write_all(b"0000").unwrap();
1339 stdin.flush().unwrap();
1340 read_until(&mut stdout, b"status 200\n0000", &mut transcript);
1341
1342 stdin.write_all(&pkt_text("quit")).unwrap();
1343 stdin.write_all(b"0000").unwrap();
1344 stdin.flush().unwrap();
1345 drop(stdin);
1346 use std::io::Read;
1347 let _ = stdout.read_to_end(&mut transcript);
1348 let status = child.wait().expect("ssh exits");
1349 (
1350 status.success(),
1351 String::from_utf8_lossy(&transcript).into_owned(),
1352 )
1353}
1354
1355#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1356async fn shutdown_drains_an_in_flight_lfs_transfer_before_exit() {
1357 use knot_lfs::LfsStore;
1358 use sha2::Digest;
1359 let scratch = tempfile::tempdir().unwrap();
1360 let (key_path, public_line) = keygen(scratch.path(), "drain");
1361 let lfs_dir = scratch.path().join("lfs");
1362 std::fs::create_dir_all(&lfs_dir).unwrap();
1363 let handle = knot_lfs::LfsHandle::open(
1364 knot_lfs::LfsStorePath::new(&lfs_dir),
1365 knot_lfs::LfsSize::new(1 << 30),
1366 knot_lfs::FreeSpaceFloor::new(0),
1367 )
1368 .unwrap();
1369 let (server, _index, shutdown, serve_task) = spawn_server_core(
1370 public_line,
1371 MaxWireBytes::new(1 << 20),
1372 true,
1373 Some(handle.clone()),
1374 )
1375 .await;
1376
1377 let body: Vec<u8> = (0..1_048_576u32).map(|n| (n % 251) as u8).collect();
1378 let oid = knot_lfs::LfsOid::from_digest(sha2::Sha256::digest(&body).into());
1379 let (midway_tx, midway_rx) = std::sync::mpsc::channel();
1380
1381 let client = {
1382 let key_path = key_path.clone();
1383 let oid = oid.clone();
1384 let port = server.port;
1385 tokio::task::spawn_blocking(move || {
1386 trickled_lfs_upload(&key_path, port, &body, oid.as_str(), midway_tx)
1387 })
1388 };
1389
1390 tokio::task::spawn_blocking(move || {
1391 midway_rx
1392 .recv_timeout(std::time::Duration::from_secs(20))
1393 .expect("the upload must reach its midway point")
1394 })
1395 .await
1396 .unwrap();
1397
1398 shutdown.cancel();
1399 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
1400 assert!(
1401 !serve_task.is_finished(),
1402 "the listener must keep draining while a transfer is in flight"
1403 );
1404
1405 let (ok, transcript) = client.await.unwrap();
1406 assert!(
1407 ok,
1408 "the in-flight upload must finish cleanly across the shutdown:\n{transcript}"
1409 );
1410 assert!(
1411 transcript.contains("status 200"),
1412 "the server must acknowledge the drained upload:\n{transcript}"
1413 );
1414
1415 tokio::time::timeout(std::time::Duration::from_secs(10), serve_task)
1416 .await
1417 .expect("the drained listener must exit promptly once transfers finish")
1418 .unwrap();
1419
1420 let repo_did = RepoDid::new(REPO_DID).unwrap();
1421 assert_eq!(
1422 handle
1423 .store
1424 .probe(&repo_did, &oid)
1425 .unwrap()
1426 .map(|size| size.get()),
1427 Some(1_048_576),
1428 "the drained upload must be durable"
1429 );
1430
1431 let (connected, _) = {
1432 let key_path = key_path.clone();
1433 let port = server.port;
1434 tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1435 .await
1436 .unwrap()
1437 };
1438 assert!(
1439 !connected,
1440 "a connection after shutdown must be refused, the drain only covers in-flight work"
1441 );
1442}