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