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 a_client_requesting_ssh_compression_clones_an_incompressible_pack() {
657 let fx = fixture().await;
658 std::fs::create_dir_all(&fx.work).unwrap();
659 git(&fx.work, &[], &["init", "-q", "-b", "main"]);
660 let mut state = 0x9e3779b97f4a7c15u64;
661 let payload: Vec<u8> = std::iter::repeat_with(|| {
662 state ^= state << 13;
663 state ^= state >> 7;
664 state ^= state << 17;
665 state.to_le_bytes()
666 })
667 .take(32 * 1024)
668 .flatten()
669 .collect();
670 std::fs::write(fx.work.join("noise.bin"), &payload).unwrap();
671 git(&fx.work, &[], &["add", "-A"]);
672 git(&fx.work, &[], &["commit", "-q", "-m", "noise"]);
673
674 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
675 assert!(ok, "push must succeed:\n{out}");
676
677 let dest = fx.scratch.path().join("compressed-clone");
678 let ssh = format!("{} -o Compression=yes", ssh_command(&fx.key_path));
679 let url = fx.url.clone();
680 let dest_arg = dest.to_str().unwrap().to_string();
681 let (ok, out) = tokio::task::spawn_blocking(move || {
682 git(
683 Path::new("/tmp"),
684 &[("GIT_SSH_COMMAND", &ssh)],
685 &["clone", "-q", &url, &dest_arg],
686 )
687 })
688 .await
689 .unwrap();
690 assert!(
691 ok,
692 "clone with ssh compression requested must succeed:\n{out}"
693 );
694 assert_eq!(std::fs::read(dest.join("noise.bin")).unwrap(), payload);
695}
696
697#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
698async fn an_oversized_push_is_refused_at_the_ssh_boundary() {
699 let scratch = tempfile::tempdir().unwrap();
700 let (key_path, public_line) = keygen(scratch.path(), "client");
701 let (server, _index) = spawn_server(public_line, MaxWireBytes::new(64)).await;
702 let url = format!(
703 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
704 server.port
705 );
706
707 let work = scratch.path().join("work");
708 seed_work(&work);
709 let (ok, out) = push(&work, &url, &key_path, &["main"]).await;
710 assert!(
711 !ok,
712 "push larger than the configured limit must be refused:\n{out}"
713 );
714 assert!(
715 ref_names(&server).is_empty(),
716 "oversized push mustn't land any ref"
717 );
718}
719
720#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
721async fn an_up_to_date_push_over_ssh_is_accepted() {
722 let fx = fixture().await;
723 seed_work(&fx.work);
724
725 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
726 assert!(ok, "first push must land:\n{out}");
727
728 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
729 assert!(
730 ok,
731 "up-to-date no-op push must succeed instead of failing with a stream error:\n{out}"
732 );
733 assert!(
734 out.contains("up-to-date") || out.contains("up to date"),
735 "git must report branch is up to date:\n{out}"
736 );
737}
738
739#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
740async fn a_denied_push_over_ssh_leaves_no_objects_in_the_live_odb() {
741 let scratch = tempfile::tempdir().unwrap();
742 let (_registered_path, registered_line) = keygen(scratch.path(), "registered");
743 let (attacker_path, _attacker_line) = keygen(scratch.path(), "attacker");
744 let (server, _index) = spawn_server(registered_line, MaxWireBytes::new(1 << 30)).await;
745 let url = format!(
746 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
747 server.port
748 );
749
750 let work = scratch.path().join("work");
751 let head = seed_work(&work);
752 let (ok, out) = push(&work, &url, &attacker_path, &["main"]).await;
753 assert!(!ok, "unauthorized push must be rejected:\n{out}");
754
755 let repo = server.layout.open(&server.repo_did).unwrap();
756 assert!(
757 repo.references().unwrap().is_empty(),
758 "denied push must create no ref"
759 );
760 assert!(
761 !repo.contains(Oid::from_hex(&head).unwrap()),
762 "denied push must migrate no objects into the live odb"
763 );
764}
765
766#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
767async fn ref_namespace_policy() {
768 let fx = fixture().await;
769 seed_work(&fx.work);
770
771 let (ok, out) = push(
772 &fx.work,
773 &fx.url,
774 &fx.key_path,
775 &["main:refs/hidden/feature/main"],
776 )
777 .await;
778 assert!(!ok, "push to refs/hidden/* must be rejected:\n{out}");
779 assert!(
780 ref_names(&fx.server).is_empty(),
781 "forbidden-ref push must land nothing"
782 );
783
784 let (ok, out) = push(
785 &fx.work,
786 &fx.url,
787 &fx.key_path,
788 &["main:refs/notes/commits"],
789 )
790 .await;
791 assert!(
792 ok,
793 "push to any non-reserved namespace must be accepted:\n{out}"
794 );
795 assert!(
796 ref_names(&fx.server)
797 .iter()
798 .any(|name| name == "refs/notes/commits"),
799 "pushed ref must land"
800 );
801}
802
803#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
804async fn cob_ref_guard_lifecycle() {
805 let fx = fixture().await;
806 seed_work(&fx.work);
807 let home = CobHome::from(&RepoDid::new(REPO_DID).unwrap());
808 let foreign = CobHome::from(&RepoDid::new("did:plc:whelk").unwrap());
809
810 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
811 assert!(ok, "head must land for the advertisement check:\n{out}");
812
813 let (owned_tip, owned_ref, owned_spec) = seed_cob(&fx.work, 1, "did:plc:limpet", &home);
814 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[owned_spec.as_str()]).await;
815 assert!(
816 ok,
817 "COB ref signed by the repository key must verify and land over ssh:\n{out}"
818 );
819
820 let (_forged_tip, forged_ref, forged_spec) = seed_cob(&fx.work, 9, "did:plc:whelk", &home);
821 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[forged_spec.as_str()]).await;
822 assert!(
823 !ok,
824 "COB ref signed by a stranger must be refused at the receive boundary:\n{out}"
825 );
826
827 let (_transplant_tip, transplant_ref, transplant_spec) =
828 seed_cob(&fx.work, 1, "did:plc:mussel", &foreign);
829 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[transplant_spec.as_str()]).await;
830 assert!(
831 !ok,
832 "same key signing for another repo's home must be refused on transplant:\n{out}"
833 );
834
835 let landed = ref_names(&fx.server);
836 assert!(
837 landed.contains(&owned_ref),
838 "owner-signed COB ref must be stored: {landed:?}"
839 );
840 assert!(
841 !landed.contains(&forged_ref),
842 "stranger-signed COB ref must be absent: {landed:?}"
843 );
844 assert!(
845 !landed.contains(&transplant_ref),
846 "transplanted COB ref must be absent: {landed:?}"
847 );
848
849 let cob_name = RefName::new(&owned_ref).unwrap();
850 let del = format!(":{owned_ref}");
851 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[del.as_str()]).await;
852 assert!(!ok, "deleting a COB ref must be refused:\n{out}");
853 assert!(
854 out.contains("append-only"),
855 "rejection must name the append-only rule:\n{out}"
856 );
857 assert!(
858 ref_names(&fx.server).contains(&owned_ref),
859 "COB ref must survive the refused delete"
860 );
861
862 let repo = Repo::open(&fx.work).unwrap();
863 CobStore::new(&repo)
864 .update(
865 &home,
866 knot_types::CobId::new(owned_tip),
867 &MembersChange::Add(Grant {
868 subject: AccountDid::new("did:plc:bailey").unwrap(),
869 added_by: AccountDid::new(OWNER_DID).unwrap(),
870 created_at: UnixSeconds::new(2),
871 }),
872 &K256Signer::generate(&SeededEntropy::new(1)),
873 UnixSeconds::new(2),
874 )
875 .unwrap();
876 assert_ne!(
877 repo.find_ref(&cob_name).unwrap(),
878 Some(owned_tip),
879 "local COB ref now points at a new, equally valid tip"
880 );
881 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &[owned_spec.as_str()]).await;
882 assert!(
883 !ok,
884 "re-pushing a moved COB ref must be refused instead of silently clobbered:\n{out}"
885 );
886 assert_eq!(
887 fx.server
888 .layout
889 .open(&fx.server.repo_did)
890 .unwrap()
891 .find_ref(&cob_name)
892 .unwrap(),
893 Some(owned_tip),
894 "live COB ref must still point at the original tip"
895 );
896
897 let (ok, advert) = git_ssh(Path::new("/tmp"), &fx.key_path, &["ls-remote", &fx.url]).await;
898 assert!(ok, "ls-remote over ssh must succeed:\n{advert}");
899 assert!(
900 advert.contains("refs/heads/main"),
901 "head must be advertised:\n{advert}"
902 );
903 assert!(
904 !advert.contains("refs/cobs/"),
905 "no refs/cobs/* may leak into the ssh advertisement:\n{advert}"
906 );
907}
908
909#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
910async fn key_recognition_edge_cases() {
911 let fx = fixture().await;
912 let head = seed_work(&fx.work);
913 let head_oid = Oid::from_hex(&head).unwrap();
914
915 let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
916 let two_ids = format!(
917 "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
918 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
919 fx.key_path
920 );
921 let (ok, out) = {
922 let (work, url) = (fx.work.clone(), fx.url.clone());
923 tokio::task::spawn_blocking(move || {
924 git(
925 &work,
926 &[("GIT_SSH_COMMAND", &two_ids)],
927 &["push", "-q", &url, "main"],
928 )
929 })
930 .await
931 .unwrap()
932 };
933 assert!(
934 ok,
935 "rejecting unregistered key must let client cycle to the registered one:\n{out}"
936 );
937 assert_eq!(
938 main_tip(&fx.server.layout, &fx.server.repo_did),
939 Some(head_oid)
940 );
941
942 let blob = russh::keys::ssh_key::PublicKey::from_openssh(
943 &std::fs::read_to_string(fx.scratch.path().join("client.pub")).unwrap(),
944 )
945 .unwrap()
946 .to_bytes()
947 .unwrap();
948 fx.index.cache_key(
949 knot_types::OfferedKey::from_bytes(blob),
950 &AccountDid::new("did:plc:whelk").unwrap(),
951 );
952 let (ok, out) = push(
953 &fx.work,
954 &fx.url,
955 &fx.key_path,
956 &["main:refs/heads/squat-check"],
957 )
958 .await;
959 assert!(
960 ok,
961 "stranger who published the owner's key mustn't deny the owner's push:\n{out}"
962 );
963 assert_eq!(
964 fx.server
965 .layout
966 .open(&fx.server.repo_did)
967 .unwrap()
968 .find_ref(&RefName::new("refs/heads/squat-check").unwrap())
969 .unwrap(),
970 Some(head_oid)
971 );
972}
973
974#[test]
975fn a_group_or_other_readable_host_key_is_refused_on_load() {
976 use std::os::unix::fs::PermissionsExt;
977 let dir = tempfile::tempdir().unwrap();
978 let path = dir.path().join("host");
979 knot_ssh::load_or_create_host_key(&path).unwrap();
980 assert_eq!(
981 std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
982 0o600,
983 "freshly created host key is 0600"
984 );
985
986 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
987 let refused = knot_ssh::load_or_create_host_key(&path);
988 assert!(
989 matches!(refused, Err(knot_ssh::SshError::HostKey { .. })),
990 "world-readable existing host key must be refused on load: {refused:?}"
991 );
992}
993
994async fn launch(
995 host_key_dir: &Path,
996 layout: Layout,
997 index: Arc<Index>,
998 identities: HashMap<String, Vec<String>>,
999) -> u16 {
1000 let atproto = Arc::new(Atproto::new(
1001 multi_http(identities),
1002 ManualClock::new(UnixMicros::new(1_000_000_000)),
1003 KnotId::new("did:web:nel.pet").unwrap(),
1004 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
1005 ));
1006 std::fs::create_dir_all(host_key_dir).unwrap();
1007 let host_key = knot_ssh::load_or_create_host_key(&host_key_dir.join("host")).unwrap();
1008 let events = Arc::new(knot_events::EventLog::new(
1009 ManualClock::new(UnixMicros::new(1_000_000_000)),
1010 knot_events::ReplayBounds::new(
1011 knot_events::ReplayEvents::new(64).unwrap(),
1012 knot_events::ReplayBytes::new(16 << 20).unwrap(),
1013 ),
1014 ));
1015 let state = Arc::new(knot_ssh::SshState::new(
1016 layout,
1017 index,
1018 atproto,
1019 actor_for_seed(77),
1020 events,
1021 knot_types::KnotHostname::new("knot.test").unwrap(),
1022 knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(),
1023 std::collections::BTreeSet::new(),
1024 knot_types::AdmissionPolicy::Closed,
1025 MaxWireBytes::new(1 << 30),
1026 LanguagesPushBudget::new(std::time::Duration::from_secs(2)),
1027 None,
1028 ));
1029 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1030 let port = listener.local_addr().unwrap().port();
1031 tokio::spawn(async move {
1032 let _ = knot_ssh::serve_on_socket(listener, host_key, state).await;
1033 });
1034 port
1035}
1036
1037#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1038async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo_it_has_no_grant_on()
1039 {
1040 const REPO_A: &str = "did:plc:squid";
1041 const REPO_B: &str = "did:plc:clam";
1042 const OWNER: &str = "did:plc:nel";
1043 const COLLAB: &str = "did:plc:olaren";
1044
1045 let scratch = tempfile::tempdir().unwrap();
1046 let (owner_key, owner_line) = keygen(scratch.path(), "owner");
1047 let (collab_key, collab_line) = keygen(scratch.path(), "collab");
1048
1049 let meta_path = scratch.path().join("meta");
1050 Repo::create(&meta_path).unwrap();
1051 let layout = Layout::new(scratch.path().join("repos"));
1052 let repo_a = RepoDid::new(REPO_A).unwrap();
1053 let repo_b = RepoDid::new(REPO_B).unwrap();
1054 let git_a = layout.create(&repo_a).unwrap();
1055 layout.create(&repo_b).unwrap();
1056
1057 let signer = K256Signer::generate(&SeededEntropy::new(2));
1058 let meta = Repo::open(&meta_path).unwrap();
1059 let store = CobStore::new(&meta);
1060 let knot_home = CobHome::from(&KnotId::new("did:web:nel.pet").unwrap());
1061 let reg = store
1062 .create(
1063 &knot_home,
1064 &RegistryChange::Register(Registration {
1065 owner: OwnerDid::new(OWNER).unwrap(),
1066 rkey: RepoRkey::new("anemone").unwrap(),
1067 name: RepoName::new("anemone").unwrap(),
1068 repo: repo_a.clone(),
1069 created_at: UnixSeconds::new(1),
1070 }),
1071 &signer,
1072 UnixSeconds::new(1),
1073 )
1074 .unwrap();
1075 store
1076 .update(
1077 &knot_home,
1078 reg.object,
1079 &RegistryChange::Register(Registration {
1080 owner: OwnerDid::new(OWNER).unwrap(),
1081 rkey: RepoRkey::new("barnacle").unwrap(),
1082 name: RepoName::new("barnacle").unwrap(),
1083 repo: repo_b.clone(),
1084 created_at: UnixSeconds::new(2),
1085 }),
1086 &signer,
1087 UnixSeconds::new(2),
1088 )
1089 .unwrap();
1090 store
1091 .create(
1092 &knot_home,
1093 &MembersChange::Add(Grant {
1094 subject: AccountDid::new(COLLAB).unwrap(),
1095 added_by: AccountDid::new(OWNER).unwrap(),
1096 created_at: UnixSeconds::new(1),
1097 }),
1098 &signer,
1099 UnixSeconds::new(1),
1100 )
1101 .unwrap();
1102 CobStore::new(&git_a)
1103 .create(
1104 &CobHome::from(&repo_a),
1105 &CollaboratorsChange::Add(Grant {
1106 subject: AccountDid::new(COLLAB).unwrap(),
1107 added_by: AccountDid::new(OWNER).unwrap(),
1108 created_at: UnixSeconds::new(1),
1109 }),
1110 &signer,
1111 UnixSeconds::new(1),
1112 )
1113 .unwrap();
1114
1115 let index = Arc::new(Index::new(meta_path, layout.clone()));
1116 index.rebuild().unwrap();
1117 index.warm_collaborators();
1118
1119 let identities = HashMap::from([
1120 (OWNER.to_string(), vec![owner_line]),
1121 (COLLAB.to_string(), vec![collab_line]),
1122 ]);
1123 let port = launch(
1124 &scratch.path().join("hostkey"),
1125 layout.clone(),
1126 Arc::clone(&index),
1127 identities,
1128 )
1129 .await;
1130
1131 let work_a = scratch.path().join("work_a");
1132 let head_a = seed_work(&work_a);
1133 let url_a = format!("ssh://git@127.0.0.1:{port}/{REPO_A}");
1134 let (ok, out) = push(&work_a, &url_a, &collab_key, &["main"]).await;
1135 assert!(
1136 ok,
1137 "collaborator must push the repo it collaborates on:\n{out}"
1138 );
1139 assert_eq!(
1140 main_tip(&layout, &repo_a),
1141 Some(Oid::from_hex(&head_a).unwrap()),
1142 "collaborator's commit must be repo A's main tip"
1143 );
1144
1145 let work_b = scratch.path().join("work_b");
1146 seed_work(&work_b);
1147 let url_b = format!("ssh://git@127.0.0.1:{port}/{REPO_B}");
1148 let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await;
1149 assert!(
1150 !denied,
1151 "key recognized via repo A but with no grant on repo B must be denied, recognition is \
1152 not authorization:\n{out}"
1153 );
1154 assert!(
1155 main_tip(&layout, &repo_b).is_none(),
1156 "denied cross-repo push must land nothing on repo B"
1157 );
1158
1159 let work_owner = scratch.path().join("work_owner_b");
1160 let head_owner = seed_work(&work_owner);
1161 let (ok, out) = push(&work_owner, &url_b, &owner_key, &["main"]).await;
1162 assert!(ok, "owner must push to repo B:\n{out}");
1163 assert_eq!(
1164 main_tip(&layout, &repo_b),
1165 Some(Oid::from_hex(&head_owner).unwrap()),
1166 "owner's push to repo B must land, isolating the collaborator's denial as authorization"
1167 );
1168}
1169
1170fn ssh_bare(key_path: &str, port: u16) -> (bool, String) {
1171 let out = Command::new("ssh")
1172 .args([
1173 "-i",
1174 key_path,
1175 "-o",
1176 "IdentitiesOnly=yes",
1177 "-o",
1178 "StrictHostKeyChecking=no",
1179 "-o",
1180 "UserKnownHostsFile=/dev/null",
1181 "-o",
1182 "PreferredAuthentications=publickey",
1183 "-o",
1184 "BatchMode=yes",
1185 "-p",
1186 &port.to_string(),
1187 "git@127.0.0.1",
1188 ])
1189 .output()
1190 .expect("ssh runs");
1191 (
1192 out.status.success(),
1193 format!(
1194 "{}{}",
1195 String::from_utf8_lossy(&out.stdout),
1196 String::from_utf8_lossy(&out.stderr)
1197 ),
1198 )
1199}
1200
1201#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1202async fn a_bare_ssh_session_greets_the_recognized_user() {
1203 let fx = fixture().await;
1204 let port = fx.server.port;
1205 let key_path = fx.key_path.clone();
1206 let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1207 .await
1208 .unwrap();
1209 assert!(
1210 out.contains("@nel.pet"),
1211 "greeting resolves and addresses the user by handle:\n{out}"
1212 );
1213 assert!(out.contains("knot.test"), "greeting names the knot:\n{out}");
1214}
1215
1216#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1217async fn a_push_to_a_new_branch_offers_a_pull_request_link() {
1218 let fx = fixture().await;
1219 seed_work(&fx.work);
1220
1221 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1222 assert!(ok, "seeding main must land:\n{out}");
1223
1224 git(&fx.work, &[], &["checkout", "-q", "-b", "feature"]);
1225 std::fs::write(fx.work.join("feature.txt"), "work\n").unwrap();
1226 git(&fx.work, &[], &["add", "-A"]);
1227 git(&fx.work, &[], &["commit", "-q", "-m", "feature work"]);
1228
1229 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["feature"]).await;
1230 assert!(ok, "feature-branch push must land:\n{out}");
1231 assert!(
1232 out.contains("https://tangled.test/nel.pet/anemone/pulls/new"),
1233 "new non-default branch is answered with a pull-request link:\n{out}"
1234 );
1235 assert!(
1236 out.contains("sourceBranch=feature") && out.contains("targetBranch=main"),
1237 "link points the new branch at the default:\n{out}"
1238 );
1239}
1240
1241#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1242async fn a_verbose_ci_push_option_reports_a_clean_pipeline() {
1243 let fx = fixture().await;
1244 std::fs::create_dir_all(fx.work.join(".tangled/workflows")).unwrap();
1245 git(&fx.work, &[], &["init", "-q", "-b", "main"]);
1246 std::fs::write(
1247 fx.work.join(".tangled/workflows/ci.yml"),
1248 "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n",
1249 )
1250 .unwrap();
1251 git(&fx.work, &[], &["add", "-A"]);
1252 git(&fx.work, &[], &["commit", "-q", "-m", "add ci"]);
1253
1254 let (ok, out) = push(
1255 &fx.work,
1256 &fx.url,
1257 &fx.key_path,
1258 &["--push-option=verbose-ci", "main"],
1259 )
1260 .await;
1261 assert!(ok, "push with a push option must land:\n{out}");
1262 assert!(
1263 out.contains("no diagnostics"),
1264 "verbose-ci reports clean compile over the sideband:\n{out}"
1265 );
1266}
1267
1268#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1269async fn git_archive_remote_over_ssh_streams_a_tar_of_the_tree() {
1270 let fx = fixture().await;
1271 seed_work(&fx.work);
1272 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1273 assert!(ok, "seeding push must land before archiving:\n{out}");
1274
1275 let out_tar = fx.scratch.path().join("archive.tar");
1276 let (ok, out) = git_ssh(
1277 &fx.work,
1278 &fx.key_path,
1279 &[
1280 "archive",
1281 "--format=tar",
1282 "--remote",
1283 &fx.url,
1284 "-o",
1285 out_tar.to_str().unwrap(),
1286 "HEAD",
1287 ],
1288 )
1289 .await;
1290 assert!(ok, "git archive --remote over ssh must succeed:\n{out}");
1291
1292 let tar = std::fs::read(&out_tar).unwrap();
1293 assert!(
1294 tar.windows(b"README.md".len()).any(|w| w == b"README.md"),
1295 "archived tar must contain the README.md entry"
1296 );
1297}
1298
1299fn pkt(payload: &[u8]) -> Vec<u8> {
1300 let mut framed = format!("{:04x}", payload.len() + 4).into_bytes();
1301 framed.extend_from_slice(payload);
1302 framed
1303}
1304
1305fn pkt_text(line: &str) -> Vec<u8> {
1306 pkt(format!("{line}\n").as_bytes())
1307}
1308
1309fn read_until(reader: &mut impl std::io::Read, needle: &[u8], buffer: &mut Vec<u8>) {
1310 std::iter::from_fn(|| {
1311 let mut byte = [0u8; 1];
1312 match reader.read(&mut byte) {
1313 Ok(0) | Err(_) => None,
1314 Ok(_) => {
1315 buffer.push(byte[0]);
1316 Some(buffer.ends_with(needle))
1317 }
1318 }
1319 })
1320 .find(|done| *done)
1321 .expect("the session must answer before closing the stream");
1322}
1323
1324fn trickled_lfs_upload(
1325 key_path: &str,
1326 port: u16,
1327 body: &[u8],
1328 oid: &str,
1329 midway: std::sync::mpsc::Sender<()>,
1330) -> (bool, String) {
1331 use std::io::Write;
1332 let mut child = Command::new("ssh")
1333 .args([
1334 "-i",
1335 key_path,
1336 "-o",
1337 "IdentitiesOnly=yes",
1338 "-o",
1339 "StrictHostKeyChecking=no",
1340 "-o",
1341 "UserKnownHostsFile=/dev/null",
1342 "-o",
1343 "PreferredAuthentications=publickey",
1344 "-o",
1345 "BatchMode=yes",
1346 "-p",
1347 &port.to_string(),
1348 "git@127.0.0.1",
1349 &format!("git-lfs-transfer '{OWNER_DID}/{REPO_NAME}' upload"),
1350 ])
1351 .stdin(std::process::Stdio::piped())
1352 .stdout(std::process::Stdio::piped())
1353 .stderr(std::process::Stdio::null())
1354 .spawn()
1355 .expect("ssh runs");
1356 let mut stdin = child.stdin.take().unwrap();
1357 let mut stdout = child.stdout.take().unwrap();
1358 let mut transcript = Vec::new();
1359
1360 read_until(&mut stdout, b"version=1\n0000", &mut transcript);
1361
1362 let (first, second) = body.split_at(body.len() / 2);
1363 stdin
1364 .write_all(&pkt_text(&format!("put-object {oid}")))
1365 .unwrap();
1366 stdin
1367 .write_all(&pkt_text(&format!("size={}", body.len())))
1368 .unwrap();
1369 stdin.write_all(b"0001").unwrap();
1370 first.chunks(32 * 1024).for_each(|chunk| {
1371 stdin.write_all(&pkt(chunk)).unwrap();
1372 });
1373 stdin.flush().unwrap();
1374 midway.send(()).unwrap();
1375 std::thread::sleep(std::time::Duration::from_millis(900));
1376
1377 second.chunks(32 * 1024).for_each(|chunk| {
1378 stdin.write_all(&pkt(chunk)).unwrap();
1379 });
1380 stdin.write_all(b"0000").unwrap();
1381 stdin.flush().unwrap();
1382 read_until(&mut stdout, b"status 200\n0000", &mut transcript);
1383
1384 stdin.write_all(&pkt_text("quit")).unwrap();
1385 stdin.write_all(b"0000").unwrap();
1386 stdin.flush().unwrap();
1387 drop(stdin);
1388 use std::io::Read;
1389 let _ = stdout.read_to_end(&mut transcript);
1390 let status = child.wait().expect("ssh exits");
1391 (
1392 status.success(),
1393 String::from_utf8_lossy(&transcript).into_owned(),
1394 )
1395}
1396
1397#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1398async fn shutdown_drains_an_in_flight_lfs_transfer_before_exit() {
1399 use knot_lfs::LfsStore;
1400 use sha2::Digest;
1401 let scratch = tempfile::tempdir().unwrap();
1402 let (key_path, public_line) = keygen(scratch.path(), "drain");
1403 let lfs_dir = scratch.path().join("lfs");
1404 std::fs::create_dir_all(&lfs_dir).unwrap();
1405 let handle = knot_lfs::LfsHandle::open(
1406 knot_lfs::LfsStorePath::new(&lfs_dir),
1407 knot_lfs::LfsSize::new(1 << 30),
1408 knot_lfs::FreeSpaceFloor::new(0),
1409 )
1410 .unwrap();
1411 let (server, _index, shutdown, serve_task) = spawn_server_core(
1412 public_line,
1413 MaxWireBytes::new(1 << 20),
1414 true,
1415 Some(handle.clone()),
1416 )
1417 .await;
1418
1419 let body: Vec<u8> = (0..1_048_576u32).map(|n| (n % 251) as u8).collect();
1420 let oid = knot_lfs::LfsOid::from_digest(sha2::Sha256::digest(&body).into());
1421 let (midway_tx, midway_rx) = std::sync::mpsc::channel();
1422
1423 let client = {
1424 let key_path = key_path.clone();
1425 let oid = oid.clone();
1426 let port = server.port;
1427 tokio::task::spawn_blocking(move || {
1428 trickled_lfs_upload(&key_path, port, &body, oid.as_str(), midway_tx)
1429 })
1430 };
1431
1432 tokio::task::spawn_blocking(move || {
1433 midway_rx
1434 .recv_timeout(std::time::Duration::from_secs(20))
1435 .expect("the upload must reach its midway point")
1436 })
1437 .await
1438 .unwrap();
1439
1440 shutdown.cancel();
1441 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
1442 assert!(
1443 !serve_task.is_finished(),
1444 "the listener must keep draining while a transfer is in flight"
1445 );
1446
1447 let (ok, transcript) = client.await.unwrap();
1448 assert!(
1449 ok,
1450 "the in-flight upload must finish cleanly across the shutdown:\n{transcript}"
1451 );
1452 assert!(
1453 transcript.contains("status 200"),
1454 "the server must acknowledge the drained upload:\n{transcript}"
1455 );
1456
1457 tokio::time::timeout(std::time::Duration::from_secs(10), serve_task)
1458 .await
1459 .expect("the drained listener must exit promptly once transfers finish")
1460 .unwrap();
1461
1462 let repo_did = RepoDid::new(REPO_DID).unwrap();
1463 assert_eq!(
1464 handle
1465 .store
1466 .probe(&repo_did, &oid)
1467 .unwrap()
1468 .map(|size| size.get()),
1469 Some(1_048_576),
1470 "the drained upload must be durable"
1471 );
1472
1473 let (connected, _) = {
1474 let key_path = key_path.clone();
1475 let port = server.port;
1476 tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1477 .await
1478 .unwrap()
1479 };
1480 assert!(
1481 !connected,
1482 "a connection after shutdown must be refused, the drain only covers in-flight work"
1483 );
1484}