alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::path::Path;
2use std::process::Command;
3use std::sync::Arc;
4
5use futures::stream::StreamExt;
6use knot_atproto::Atproto;
7use knot_cob::{CobHome, CobStore};
8use knot_cobs::{Registration, RegistryChange};
9use knot_git::{Layout, Repo};
10use knot_runtime::{
11 FakeHttp, HttpResponse, K256Signer, ManualClock, SeededEntropy, Signer, UnixMicros,
12};
13use knot_types::{KnotId, OwnerDid, RepoDid, RepoName, RepoRkey, UnixSeconds};
14use tempfile::TempDir;
15use tokio::net::TcpListener;
16use url::Url;
17
18const REPO_DID: &str = "did:plc:squid";
19const REPO_NAME: &str = "anemone";
20const OWNER_DID: &str = "did:plc:nel";
21const PDS_HOST: &str = "pds.oyster.cafe";
22const KNOT_DID: &str = "did:web:nel.pet";
23const PINNED_DATE: &str = "2026-06-20T12:00:00+00:00";
24
25fn git(cwd: &Path, env: &[(&str, &str)], args: &[&str]) -> (bool, String) {
26 let mut command = knot_fixtures::command_at(cwd, PINNED_DATE);
27 command.args(args);
28 env.iter().for_each(|(key, value)| {
29 command.env(key, value);
30 });
31 let out = command.output().expect("git runs");
32 (
33 out.status.success(),
34 format!(
35 "{}{}",
36 String::from_utf8_lossy(&out.stdout),
37 String::from_utf8_lossy(&out.stderr)
38 ),
39 )
40}
41
42fn keygen(dir: &Path) -> (String, String) {
43 let path = dir.join("client");
44 let out = Command::new("ssh-keygen")
45 .args([
46 "-t",
47 "ed25519",
48 "-N",
49 "",
50 "-C",
51 "nel@oyster.cafe",
52 "-f",
53 path.to_str().unwrap(),
54 ])
55 .output()
56 .expect("ssh-keygen runs");
57 assert!(out.status.success());
58 let public_line = std::fs::read_to_string(dir.join("client.pub"))
59 .unwrap()
60 .trim()
61 .to_string();
62 (path.to_str().unwrap().to_string(), public_line)
63}
64
65fn did_document(signer: &K256Signer, did: &str) -> Vec<u8> {
66 let multikey = knot_types::crypto::multikey(0xe7, signer.public_key().as_bytes());
67 serde_json::to_vec(&serde_json::json!({
68 "id": did,
69 "alsoKnownAs": ["at://nel.pet"],
70 "verificationMethod": [{
71 "id": format!("{did}#atproto"),
72 "type": "Multikey",
73 "controller": did,
74 "publicKeyMultibase": multikey
75 }],
76 "service": [{
77 "id": "#atproto_pds",
78 "type": "AtprotoPersonalDataServer",
79 "serviceEndpoint": format!("https://{PDS_HOST}")
80 }]
81 }))
82 .unwrap()
83}
84
85fn list_records_body(public_line: &str) -> Vec<u8> {
86 serde_json::to_vec(&serde_json::json!({
87 "records": [{
88 "uri": format!("at://{OWNER_DID}/sh.tangled.publicKey/1"),
89 "value": {
90 "$type": "sh.tangled.publicKey",
91 "key": public_line,
92 "name": "laptop",
93 "createdAt": "2026-06-08T00:00:00Z"
94 }
95 }]
96 }))
97 .unwrap()
98}
99
100fn fake_http(published_line: String) -> impl knot_runtime::HttpTransport {
101 let signer = K256Signer::generate(&SeededEntropy::new(1));
102 FakeHttp::new(move |request| {
103 let host = request.url.host_str().unwrap_or_default().to_string();
104 let path = request.url.path().to_string();
105 let body = if host == PDS_HOST {
106 list_records_body(&published_line)
107 } else if path.ends_with(OWNER_DID) {
108 did_document(&signer, OWNER_DID)
109 } else if path.ends_with(REPO_DID) {
110 did_document(&signer, REPO_DID)
111 } else {
112 return Ok(HttpResponse {
113 status: http::StatusCode::NOT_FOUND,
114 headers: http::HeaderMap::new(),
115 body: bytes::Bytes::new(),
116 });
117 };
118 Ok(HttpResponse {
119 status: http::StatusCode::OK,
120 headers: http::HeaderMap::new(),
121 body: bytes::Bytes::from(body),
122 })
123 })
124}
125
126fn actor_for_seed(seed: u64) -> knot_types::ActorId {
127 knot_types::ActorId::from_secp256k1(
128 K256Signer::generate(&SeededEntropy::new(seed))
129 .public_key()
130 .as_bytes(),
131 )
132}
133
134struct Server {
135 _scan: TempDir,
136 layout: Layout,
137 repo_did: RepoDid,
138 port: u16,
139 events: Arc<knot_events::EventLog<ManualClock>>,
140}
141
142async fn spawn(published_line: String) -> Server {
143 let scan = tempfile::tempdir().unwrap();
144 let meta_path = scan.path().join("meta");
145 Repo::create(&meta_path).unwrap();
146 let layout = Layout::new(scan.path().join("repos"));
147 let repo_did = RepoDid::new(REPO_DID).unwrap();
148 layout.create(&repo_did).unwrap();
149
150 let signer = K256Signer::generate(&SeededEntropy::new(2));
151 let meta = Repo::open(&meta_path).unwrap();
152 CobStore::new(&meta)
153 .create(
154 &CobHome::from(&KnotId::new(KNOT_DID).unwrap()),
155 &RegistryChange::Register(Registration {
156 owner: OwnerDid::new(OWNER_DID).unwrap(),
157 rkey: RepoRkey::new(REPO_NAME).unwrap(),
158 name: RepoName::new(REPO_NAME).unwrap(),
159 repo: repo_did.clone(),
160 created_at: UnixSeconds::new(1),
161 }),
162 &signer,
163 UnixSeconds::new(1),
164 )
165 .unwrap();
166
167 let index = Arc::new(knot_index::Index::new(meta_path, layout.clone()));
168 index.rebuild().unwrap();
169
170 let atproto = Arc::new(Atproto::new(
171 fake_http(published_line),
172 ManualClock::new(UnixMicros::new(1_000_000_000)),
173 KnotId::new(KNOT_DID).unwrap(),
174 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
175 ));
176 let key_dir = scan.path().join("hostkey");
177 std::fs::create_dir_all(&key_dir).unwrap();
178 let host_key = knot_ssh::load_or_create_host_key(&key_dir.join("host")).unwrap();
179 let events = Arc::new(knot_events::EventLog::new(
180 ManualClock::new(UnixMicros::new(1_000_000_000)),
181 knot_events::ReplayBounds::new(
182 knot_events::ReplayEvents::new(64).unwrap(),
183 knot_events::ReplayBytes::new(16 << 20).unwrap(),
184 ),
185 ));
186 let state = Arc::new(knot_ssh::SshState::new(knot_ssh::SshConfig {
187 layout: layout.clone(),
188 index,
189 atproto,
190 knot_actor: actor_for_seed(1),
191 events: Arc::clone(&events),
192 hostname: knot_types::KnotHostname::new("knot.test").unwrap(),
193 appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(),
194 admins: std::collections::BTreeSet::new(),
195 admission: knot_types::AdmissionPolicy::Closed,
196 max_pack_bytes: knot_xrpc::MaxWireBytes::new(1 << 30),
197 archive_limit: knot_git::ArchiveLimit::default(),
198 languages_push_budget: knot_xrpc::LanguagesPushBudget::new(std::time::Duration::from_secs(
199 2,
200 )),
201 ci_logs: None,
202 }));
203 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
204 let port = listener.local_addr().unwrap().port();
205 tokio::spawn(async move {
206 let _ = knot_ssh::serve_on_socket(listener, host_key, state).await;
207 });
208 Server {
209 _scan: scan,
210 layout,
211 repo_did,
212 port,
213 events,
214 }
215}
216
217fn ssh_command(key_path: &str) -> String {
218 format!(
219 "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
220 -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
221 )
222}
223
224fn seed_work(work: &Path) -> String {
225 std::fs::create_dir_all(work).unwrap();
226 git(work, &[], &["init", "-q", "-b", "main"]);
227 std::fs::write(work.join("README.md"), "hello over the simulated ssh\n").unwrap();
228 git(work, &[], &["add", "-A"]);
229 git(work, &[], &["commit", "-q", "-m", "initial"]);
230 let (ok, head) = git(work, &[], &["rev-parse", "HEAD"]);
231 assert!(ok);
232 head.trim().to_string()
233}
234
235async fn push_once(scratch: &Path) -> (String, Option<knot_types::Oid>, serde_json::Value) {
236 let (key_path, public_line) = keygen(scratch);
237 let server = spawn(public_line).await;
238 let url = format!(
239 "ssh://git@127.0.0.1:{}/{OWNER_DID}/{REPO_NAME}",
240 server.port
241 );
242 let ssh = ssh_command(&key_path);
243 let work = scratch.join("work");
244 let head = seed_work(&work);
245
246 let (ok, out) = tokio::task::spawn_blocking(move || {
247 git(
248 &work,
249 &[("GIT_SSH_COMMAND", &ssh)],
250 &["push", "-q", &url, "main"],
251 )
252 })
253 .await
254 .unwrap();
255 assert!(ok, "simulated ssh server must accept push:\n{out}");
256
257 let stored = server
258 .layout
259 .open(&server.repo_did)
260 .unwrap()
261 .find_ref(&knot_types::RefName::new("refs/heads/main").unwrap())
262 .unwrap();
263 let event = poll_for_event(&server.events).await;
264 drop(server);
265 (head, stored, event)
266}
267
268fn replay_bounds() -> knot_events::ReplayBounds {
269 knot_events::ReplayBounds::new(
270 knot_events::ReplayEvents::new(32).unwrap(),
271 knot_events::ReplayBytes::new(16 << 20).unwrap(),
272 )
273}
274
275async fn poll_for_event(events: &knot_events::EventLog<ManualClock>) -> serde_json::Value {
276 futures::stream::iter(0..100)
277 .then(|_| async {
278 let hit = events
279 .replay(knot_events::EventCursor::START, replay_bounds())
280 .events
281 .into_iter()
282 .find(|event| event.nsid == "sh.tangled.git.refUpdate")
283 .map(|event| serde_json::to_value(&*event).unwrap()["event"].clone());
284 if hit.is_none() {
285 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
286 }
287 hit
288 })
289 .filter_map(|hit| async move { hit })
290 .boxed()
291 .next()
292 .await
293 .expect("refUpdate event must be published within polling window")
294}
295
296#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
297async fn the_simulated_ssh_write_path_lands_a_seed_deterministic_tip() {
298 let first_dir = tempfile::tempdir().unwrap();
299 let (first_head, first_stored, first_event) = push_once(first_dir.path()).await;
300
301 let second_dir = tempfile::tempdir().unwrap();
302 let (second_head, second_stored, second_event) = push_once(second_dir.path()).await;
303
304 let tip = knot_types::Oid::from_hex(&first_head).unwrap();
305 assert_eq!(
306 first_stored,
307 Some(tip),
308 "pushed commit must be the repository's main tip"
309 );
310 assert_eq!(
311 first_head, second_head,
312 "two independent runs of the simulated ssh push must produce same commit oid"
313 );
314 assert_eq!(
315 first_stored, second_stored,
316 "assembled-against-doubles ssh write path is logically reproducible"
317 );
318
319 assert_eq!(first_event["ref"], "refs/heads/main");
320 assert_eq!(first_event["newSha"], first_head);
321 assert_eq!(
322 first_event["newSha"], second_event["newSha"],
323 "ref-update event the push emits is seed-stable too"
324 );
325}