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.keys().record(
1038 &AccountDid::new("did:plc:whelk").unwrap(),
1039 vec![knot_types::OfferedKey::from_bytes(blob)],
1040 knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(knot_types::UnixSeconds::new(0)),
1041 );
1042 let (ok, out) = push(
1043 &fx.work,
1044 &fx.url,
1045 &fx.key_path,
1046 &["main:refs/heads/squat-check"],
1047 )
1048 .await;
1049 assert!(
1050 ok,
1051 "stranger who published the owner's key mustn't deny the owner's push:\n{out}"
1052 );
1053 assert_eq!(
1054 fx.server
1055 .layout
1056 .open(&fx.server.repo_did)
1057 .unwrap()
1058 .find_ref(&RefName::new("refs/heads/squat-check").unwrap())
1059 .unwrap(),
1060 Some(head_oid)
1061 );
1062}
1063
1064#[test]
1065fn a_group_or_other_readable_host_key_is_refused_on_load() {
1066 use std::os::unix::fs::PermissionsExt;
1067 let dir = tempfile::tempdir().unwrap();
1068 let path = dir.path().join("host");
1069 knot_ssh::load_or_create_host_key(&path).unwrap();
1070 assert_eq!(
1071 std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
1072 0o600,
1073 "freshly created host key is 0600"
1074 );
1075
1076 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
1077 let refused = knot_ssh::load_or_create_host_key(&path);
1078 assert!(
1079 matches!(refused, Err(knot_ssh::SshError::HostKey { .. })),
1080 "world-readable existing host key must be refused on load: {refused:?}"
1081 );
1082}
1083
1084async fn launch(
1085 host_key_dir: &Path,
1086 layout: Layout,
1087 index: Arc<Index>,
1088 identities: HashMap<String, Vec<String>>,
1089) -> u16 {
1090 let atproto = Arc::new(Atproto::new(
1091 multi_http(identities),
1092 ManualClock::new(UnixMicros::new(1_000_000_000)),
1093 KnotId::new("did:web:nel.pet").unwrap(),
1094 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
1095 ));
1096 std::fs::create_dir_all(host_key_dir).unwrap();
1097 let host_key = knot_ssh::load_or_create_host_key(&host_key_dir.join("host")).unwrap();
1098 let events = Arc::new(knot_events::EventLog::new(
1099 ManualClock::new(UnixMicros::new(1_000_000_000)),
1100 knot_events::ReplayBounds::new(
1101 knot_events::ReplayEvents::new(64).unwrap(),
1102 knot_events::ReplayBytes::new(16 << 20).unwrap(),
1103 ),
1104 ));
1105 let state = Arc::new(knot_ssh::SshState::new(knot_ssh::SshConfig {
1106 layout,
1107 index,
1108 atproto,
1109 knot_actor: actor_for_seed(77),
1110 events,
1111 hostname: knot_types::KnotHostname::new("knot.test").unwrap(),
1112 appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(),
1113 admins: std::collections::BTreeSet::new(),
1114 admission: knot_types::AdmissionPolicy::Closed,
1115 max_pack_bytes: MaxWireBytes::new(1 << 30),
1116 archive_limit: ArchiveLimit::default(),
1117 languages_push_budget: LanguagesPushBudget::new(std::time::Duration::from_secs(2)),
1118 ci_logs: None,
1119 }));
1120 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1121 let port = listener.local_addr().unwrap().port();
1122 tokio::spawn(async move {
1123 let _ = knot_ssh::serve_on_socket(listener, host_key, state).await;
1124 });
1125 port
1126}
1127
1128#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1129async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo_it_has_no_grant_on()
1130 {
1131 const REPO_A: &str = "did:plc:squid";
1132 const REPO_B: &str = "did:plc:clam";
1133 const OWNER: &str = "did:plc:nel";
1134 const COLLAB: &str = "did:plc:olaren";
1135
1136 let scratch = tempfile::tempdir().unwrap();
1137 let (owner_key, owner_line) = keygen(scratch.path(), "owner");
1138 let (collab_key, collab_line) = keygen(scratch.path(), "collab");
1139
1140 let meta_path = scratch.path().join("meta");
1141 Repo::create(&meta_path).unwrap();
1142 let layout = Layout::new(scratch.path().join("repos"));
1143 let repo_a = RepoDid::new(REPO_A).unwrap();
1144 let repo_b = RepoDid::new(REPO_B).unwrap();
1145 let git_a = layout.create(&repo_a).unwrap();
1146 layout.create(&repo_b).unwrap();
1147
1148 let signer = K256Signer::generate(&SeededEntropy::new(2));
1149 let meta = Repo::open(&meta_path).unwrap();
1150 let store = CobStore::new(&meta);
1151 let knot_home = CobHome::from(&KnotId::new("did:web:nel.pet").unwrap());
1152 let reg = store
1153 .create(
1154 &knot_home,
1155 &RegistryChange::Register(Registration {
1156 owner: OwnerDid::new(OWNER).unwrap(),
1157 rkey: RepoRkey::new("anemone").unwrap(),
1158 name: RepoName::new("anemone").unwrap(),
1159 repo: repo_a.clone(),
1160 created_at: UnixSeconds::new(1),
1161 }),
1162 &signer,
1163 UnixSeconds::new(1),
1164 )
1165 .unwrap();
1166 store
1167 .update(
1168 &knot_home,
1169 reg.object,
1170 &RegistryChange::Register(Registration {
1171 owner: OwnerDid::new(OWNER).unwrap(),
1172 rkey: RepoRkey::new("barnacle").unwrap(),
1173 name: RepoName::new("barnacle").unwrap(),
1174 repo: repo_b.clone(),
1175 created_at: UnixSeconds::new(2),
1176 }),
1177 &signer,
1178 UnixSeconds::new(2),
1179 )
1180 .unwrap();
1181 store
1182 .create(
1183 &knot_home,
1184 &MembersChange::Add(Grant {
1185 subject: AccountDid::new(COLLAB).unwrap(),
1186 added_by: AccountDid::new(OWNER).unwrap(),
1187 created_at: UnixSeconds::new(1),
1188 }),
1189 &signer,
1190 UnixSeconds::new(1),
1191 )
1192 .unwrap();
1193 CobStore::new(&git_a)
1194 .create(
1195 &CobHome::from(&repo_a),
1196 &CollaboratorsChange::Add(Grant {
1197 subject: AccountDid::new(COLLAB).unwrap(),
1198 added_by: AccountDid::new(OWNER).unwrap(),
1199 created_at: UnixSeconds::new(1),
1200 }),
1201 &signer,
1202 UnixSeconds::new(1),
1203 )
1204 .unwrap();
1205
1206 let index = Arc::new(Index::new(meta_path, layout.clone()));
1207 index.rebuild().unwrap();
1208 index.warm_collaborators();
1209
1210 let identities = HashMap::from([
1211 (OWNER.to_string(), vec![owner_line]),
1212 (COLLAB.to_string(), vec![collab_line]),
1213 ]);
1214 let port = launch(
1215 &scratch.path().join("hostkey"),
1216 layout.clone(),
1217 Arc::clone(&index),
1218 identities,
1219 )
1220 .await;
1221
1222 let work_a = scratch.path().join("work_a");
1223 let head_a = seed_work(&work_a);
1224 let url_a = format!("ssh://git@127.0.0.1:{port}/{REPO_A}");
1225 let (ok, out) = push(&work_a, &url_a, &collab_key, &["main"]).await;
1226 assert!(
1227 ok,
1228 "collaborator must push the repo it collaborates on:\n{out}"
1229 );
1230 assert_eq!(
1231 main_tip(&layout, &repo_a),
1232 Some(Oid::from_hex(&head_a).unwrap()),
1233 "collaborator's commit must be repo A's main tip"
1234 );
1235
1236 let work_b = scratch.path().join("work_b");
1237 seed_work(&work_b);
1238 let url_b = format!("ssh://git@127.0.0.1:{port}/{REPO_B}");
1239 let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await;
1240 assert!(
1241 !denied,
1242 "key recognized via repo A but with no grant on repo B must be denied, recognition is \
1243 not authorization:\n{out}"
1244 );
1245 assert!(
1246 main_tip(&layout, &repo_b).is_none(),
1247 "denied cross-repo push must land nothing on repo B"
1248 );
1249
1250 let work_owner = scratch.path().join("work_owner_b");
1251 let head_owner = seed_work(&work_owner);
1252 let (ok, out) = push(&work_owner, &url_b, &owner_key, &["main"]).await;
1253 assert!(ok, "owner must push to repo B:\n{out}");
1254 assert_eq!(
1255 main_tip(&layout, &repo_b),
1256 Some(Oid::from_hex(&head_owner).unwrap()),
1257 "owner's push to repo B must land, isolating the collaborator's denial as authorization"
1258 );
1259}
1260
1261fn ssh_bare(key_path: &str, port: u16) -> (bool, String) {
1262 let out = Command::new("ssh")
1263 .args([
1264 "-i",
1265 key_path,
1266 "-o",
1267 "IdentitiesOnly=yes",
1268 "-o",
1269 "StrictHostKeyChecking=no",
1270 "-o",
1271 "UserKnownHostsFile=/dev/null",
1272 "-o",
1273 "PreferredAuthentications=publickey",
1274 "-o",
1275 "BatchMode=yes",
1276 "-p",
1277 &port.to_string(),
1278 "git@127.0.0.1",
1279 ])
1280 .output()
1281 .expect("ssh runs");
1282 (
1283 out.status.success(),
1284 format!(
1285 "{}{}",
1286 String::from_utf8_lossy(&out.stdout),
1287 String::from_utf8_lossy(&out.stderr)
1288 ),
1289 )
1290}
1291
1292#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1293async fn a_bare_ssh_session_greets_the_recognized_user() {
1294 let fx = fixture().await;
1295 let port = fx.server.port;
1296 let key_path = fx.key_path.clone();
1297 let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1298 .await
1299 .unwrap();
1300 assert!(
1301 out.contains("@nel.pet"),
1302 "greeting resolves and addresses the user by handle:\n{out}"
1303 );
1304 assert!(out.contains("knot.test"), "greeting names the knot:\n{out}");
1305}
1306
1307#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1308async fn a_push_to_a_new_branch_offers_a_pull_request_link() {
1309 let fx = fixture().await;
1310 seed_work(&fx.work);
1311
1312 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1313 assert!(ok, "seeding main must land:\n{out}");
1314
1315 git(&fx.work, &[], &["checkout", "-q", "-b", "feature"]);
1316 std::fs::write(fx.work.join("feature.txt"), "work\n").unwrap();
1317 git(&fx.work, &[], &["add", "-A"]);
1318 git(&fx.work, &[], &["commit", "-q", "-m", "feature work"]);
1319
1320 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["feature"]).await;
1321 assert!(ok, "feature-branch push must land:\n{out}");
1322 assert!(
1323 out.contains("https://tangled.test/nel.pet/anemone/pulls/new"),
1324 "new non-default branch is answered with a pull-request link:\n{out}"
1325 );
1326 assert!(
1327 out.contains("sourceBranch=feature") && out.contains("targetBranch=main"),
1328 "link points the new branch at the default:\n{out}"
1329 );
1330}
1331
1332#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1333async fn a_verbose_ci_push_option_reports_a_clean_pipeline() {
1334 let fx = fixture().await;
1335 std::fs::create_dir_all(fx.work.join(".tangled/workflows")).unwrap();
1336 git(&fx.work, &[], &["init", "-q", "-b", "main"]);
1337 std::fs::write(
1338 fx.work.join(".tangled/workflows/ci.yml"),
1339 "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n",
1340 )
1341 .unwrap();
1342 git(&fx.work, &[], &["add", "-A"]);
1343 git(&fx.work, &[], &["commit", "-q", "-m", "add ci"]);
1344
1345 let (ok, out) = push(
1346 &fx.work,
1347 &fx.url,
1348 &fx.key_path,
1349 &["--push-option=verbose-ci", "main"],
1350 )
1351 .await;
1352 assert!(ok, "push with a push option must land:\n{out}");
1353 assert!(
1354 out.contains("no diagnostics"),
1355 "verbose-ci reports clean compile over the sideband:\n{out}"
1356 );
1357}
1358
1359#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1360async fn git_archive_remote_over_ssh_streams_a_tar_of_the_tree() {
1361 let fx = fixture().await;
1362 seed_work(&fx.work);
1363 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1364 assert!(ok, "the seeding push must succeed before archiving:\n{out}");
1365
1366 let out_tar = fx.scratch.path().join("archive.tar");
1367 let (ok, out) = git_ssh(
1368 &fx.work,
1369 &fx.key_path,
1370 &[
1371 "archive",
1372 "--format=tar",
1373 "--remote",
1374 &fx.url,
1375 "-o",
1376 out_tar.to_str().unwrap(),
1377 "HEAD",
1378 ],
1379 )
1380 .await;
1381 assert!(ok, "git archive --remote over ssh must succeed:\n{out}");
1382
1383 let tar = std::fs::read(&out_tar).unwrap();
1384 assert!(
1385 knot_fixtures::contains(&tar, b"README.md"),
1386 "archived tar must contain the README.md entry"
1387 );
1388}
1389
1390#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1391async fn git_archive_remote_over_ssh_honors_the_configured_archive_limit() {
1392 let fx = fixture_with_archive_limit(ArchiveLimit::new(512)).await;
1393 seed_work(&fx.work);
1394 let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await;
1395 assert!(ok, "the seeding push must succeed before archiving:\n{out}");
1396
1397 let out_tar = fx.scratch.path().join("archive.tar");
1398 let (ok, out) = git_ssh(
1399 &fx.work,
1400 &fx.key_path,
1401 &[
1402 "archive",
1403 "--format=tar",
1404 "--remote",
1405 &fx.url,
1406 "-o",
1407 out_tar.to_str().unwrap(),
1408 "HEAD",
1409 ],
1410 )
1411 .await;
1412 assert!(!ok, "git archive --remote past the limit must fail:\n{out}");
1413 assert!(
1414 out.contains("archive exceeds the 512 byte limit"),
1415 "the refusal must reach the client over the ssh channel:\n{out}"
1416 );
1417}
1418
1419fn pkt(payload: &[u8]) -> Vec<u8> {
1420 let mut framed = format!("{:04x}", payload.len() + 4).into_bytes();
1421 framed.extend_from_slice(payload);
1422 framed
1423}
1424
1425fn pkt_text(line: &str) -> Vec<u8> {
1426 pkt(format!("{line}\n").as_bytes())
1427}
1428
1429fn read_until(reader: &mut impl std::io::Read, needle: &[u8], buffer: &mut Vec<u8>) {
1430 std::iter::from_fn(|| {
1431 let mut byte = [0u8; 1];
1432 match reader.read(&mut byte) {
1433 Ok(0) | Err(_) => None,
1434 Ok(_) => {
1435 buffer.push(byte[0]);
1436 Some(buffer.ends_with(needle))
1437 }
1438 }
1439 })
1440 .find(|done| *done)
1441 .expect("the session must answer before closing the stream");
1442}
1443
1444fn trickled_lfs_upload(
1445 key_path: &str,
1446 port: u16,
1447 body: &[u8],
1448 oid: &str,
1449 midway: std::sync::mpsc::Sender<()>,
1450) -> (bool, String) {
1451 use std::io::Write;
1452 let mut child = Command::new("ssh")
1453 .args([
1454 "-i",
1455 key_path,
1456 "-o",
1457 "IdentitiesOnly=yes",
1458 "-o",
1459 "StrictHostKeyChecking=no",
1460 "-o",
1461 "UserKnownHostsFile=/dev/null",
1462 "-o",
1463 "PreferredAuthentications=publickey",
1464 "-o",
1465 "BatchMode=yes",
1466 "-p",
1467 &port.to_string(),
1468 "git@127.0.0.1",
1469 &format!("git-lfs-transfer '{OWNER_DID}/{REPO_NAME}' upload"),
1470 ])
1471 .stdin(std::process::Stdio::piped())
1472 .stdout(std::process::Stdio::piped())
1473 .stderr(std::process::Stdio::null())
1474 .spawn()
1475 .expect("ssh runs");
1476 let mut stdin = child.stdin.take().unwrap();
1477 let mut stdout = child.stdout.take().unwrap();
1478 let mut transcript = Vec::new();
1479
1480 read_until(&mut stdout, b"version=1\n0000", &mut transcript);
1481
1482 let (first, second) = body.split_at(body.len() / 2);
1483 stdin
1484 .write_all(&pkt_text(&format!("put-object {oid}")))
1485 .unwrap();
1486 stdin
1487 .write_all(&pkt_text(&format!("size={}", body.len())))
1488 .unwrap();
1489 stdin.write_all(b"0001").unwrap();
1490 first.chunks(32 * 1024).for_each(|chunk| {
1491 stdin.write_all(&pkt(chunk)).unwrap();
1492 });
1493 stdin.flush().unwrap();
1494 midway.send(()).unwrap();
1495 std::thread::sleep(std::time::Duration::from_millis(900));
1496
1497 second.chunks(32 * 1024).for_each(|chunk| {
1498 stdin.write_all(&pkt(chunk)).unwrap();
1499 });
1500 stdin.write_all(b"0000").unwrap();
1501 stdin.flush().unwrap();
1502 read_until(&mut stdout, b"status 200\n0000", &mut transcript);
1503
1504 stdin.write_all(&pkt_text("quit")).unwrap();
1505 stdin.write_all(b"0000").unwrap();
1506 stdin.flush().unwrap();
1507 drop(stdin);
1508 use std::io::Read;
1509 let _ = stdout.read_to_end(&mut transcript);
1510 let status = child.wait().expect("ssh exits");
1511 (
1512 status.success(),
1513 String::from_utf8_lossy(&transcript).into_owned(),
1514 )
1515}
1516
1517#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1518async fn shutdown_drains_an_in_flight_lfs_transfer_before_exit() {
1519 use knot_lfs::LfsStore;
1520 use sha2::Digest;
1521 let scratch = tempfile::tempdir().unwrap();
1522 let (key_path, public_line) = keygen(scratch.path(), "drain");
1523 let lfs_dir = scratch.path().join("lfs");
1524 std::fs::create_dir_all(&lfs_dir).unwrap();
1525 let handle = knot_lfs::LfsHandle::open(
1526 knot_lfs::LfsStorePath::new(&lfs_dir),
1527 knot_lfs::LfsSize::new(1 << 30),
1528 knot_lfs::FreeSpaceFloor::new(0),
1529 )
1530 .unwrap();
1531 let (server, _index, shutdown, serve_task) = spawn_server_core(
1532 public_line,
1533 MaxWireBytes::new(1 << 20),
1534 ArchiveLimit::default(),
1535 true,
1536 Some(handle.clone()),
1537 )
1538 .await;
1539
1540 let body: Vec<u8> = (0..1_048_576u32).map(|n| (n % 251) as u8).collect();
1541 let oid = knot_lfs::LfsOid::from_digest(sha2::Sha256::digest(&body).into());
1542 let (midway_tx, midway_rx) = std::sync::mpsc::channel();
1543
1544 let client = {
1545 let key_path = key_path.clone();
1546 let oid = oid.clone();
1547 let port = server.port;
1548 tokio::task::spawn_blocking(move || {
1549 trickled_lfs_upload(&key_path, port, &body, oid.as_str(), midway_tx)
1550 })
1551 };
1552
1553 tokio::task::spawn_blocking(move || {
1554 midway_rx
1555 .recv_timeout(std::time::Duration::from_secs(20))
1556 .expect("the upload must reach its midway point")
1557 })
1558 .await
1559 .unwrap();
1560
1561 shutdown.cancel();
1562 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
1563 assert!(
1564 !serve_task.is_finished(),
1565 "the listener must keep draining while a transfer is in flight"
1566 );
1567
1568 let (ok, transcript) = client.await.unwrap();
1569 assert!(
1570 ok,
1571 "the in-flight upload must finish cleanly across the shutdown:\n{transcript}"
1572 );
1573 assert!(
1574 transcript.contains("status 200"),
1575 "the server must acknowledge the drained upload:\n{transcript}"
1576 );
1577
1578 tokio::time::timeout(std::time::Duration::from_secs(10), serve_task)
1579 .await
1580 .expect("the drained listener must exit promptly once transfers finish")
1581 .unwrap();
1582
1583 let repo_did = RepoDid::new(REPO_DID).unwrap();
1584 assert_eq!(
1585 handle
1586 .store
1587 .probe(&repo_did, &oid)
1588 .unwrap()
1589 .map(|size| size.get()),
1590 Some(1_048_576),
1591 "the drained upload must be durable"
1592 );
1593
1594 let (connected, _) = {
1595 let key_path = key_path.clone();
1596 let port = server.port;
1597 tokio::task::spawn_blocking(move || ssh_bare(&key_path, port))
1598 .await
1599 .unwrap()
1600 };
1601 assert!(
1602 !connected,
1603 "a connection after shutdown must be refused, the drain only covers in-flight work"
1604 );
1605}