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