alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::BTreeSet;
2use std::net::SocketAddr;
3use std::num::{NonZeroU32, NonZeroU64};
4use std::path::{Path, PathBuf};
5use std::sync::{Arc, Mutex};
6use std::time::Duration;
7
8use bytes::{Buf, Bytes};
9use http::{HeaderMap, Method, Uri};
10use knot_edge::{
11 BodyInactivityTimeout, BurstSize, CertSource, EdgeConfig, EdgeGuards, HeaderTimeout,
12 IdleTimeout, ListenLimits, MaxInflightRequests, RequestTimeout, RequestsPerSecond,
13 RequiresFullHandshake, StaticCertPaths, TlsSetup, WriteRequestTimeout,
14};
15use knot_git::Layout;
16use knot_pack::{RepoLookup, RepoResolver, RepoTarget};
17use knot_types::{ObjectFormat, RepoDid};
18use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
19use rustls::crypto::aws_lc_rs;
20use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
21use rustls::{DigitallySignedStruct, SignatureScheme};
22use tokio::task::JoinHandle;
23use tokio_util::sync::CancellationToken;
24
25mod common;
26use common::{contains, must, pack_objects, receive_request};
27
28type Captured = (Method, Uri, HeaderMap, Bytes);
29
30fn nz32(value: u32) -> NonZeroU32 {
31 NonZeroU32::new(value).unwrap()
32}
33
34fn nz64(value: u64) -> NonZeroU64 {
35 NonZeroU64::new(value).unwrap()
36}
37
38fn object_set(dir: &Path) -> BTreeSet<String> {
39 must(
40 dir,
41 &[
42 "cat-file",
43 "--batch-all-objects",
44 "--batch-check=%(objectname)",
45 ],
46 )
47 .lines()
48 .map(|line| line.trim().to_string())
49 .filter(|line| !line.is_empty())
50 .collect()
51}
52
53fn serve_dids() -> Arc<dyn RepoResolver> {
54 Arc::new(|target: &RepoTarget| match target {
55 RepoTarget::Did(did) => RepoLookup::Hosted(did.clone()),
56 RepoTarget::OwnerPath(_, _) => RepoLookup::Unhosted,
57 })
58}
59
60fn free_port() -> u16 {
61 std::net::TcpListener::bind("127.0.0.1:0")
62 .unwrap()
63 .local_addr()
64 .unwrap()
65 .port()
66}
67
68fn write_self_signed(dir: &Path) -> (PathBuf, PathBuf, Vec<u8>) {
69 let generated =
70 rcgen::generate_simple_self_signed(vec!["localhost".to_string(), "127.0.0.1".to_string()])
71 .unwrap();
72 let cert_path = dir.join("cert.pem");
73 let key_path = dir.join("key.pem");
74 std::fs::write(&cert_path, generated.cert.pem()).unwrap();
75 std::fs::write(&key_path, generated.signing_key.serialize_pem()).unwrap();
76 let der = generated.cert.der().as_ref().to_vec();
77 (cert_path, key_path, der)
78}
79
80fn edge_config(addr: SocketAddr, cert: PathBuf, key: PathBuf) -> EdgeConfig {
81 EdgeConfig {
82 http_addr: knot_edge::PublicBind::new(addr),
83 limits: ListenLimits::new(
84 HeaderTimeout::from_millis(nz64(30_000)),
85 IdleTimeout::from_millis(nz64(120_000)),
86 nz32(1024),
87 ),
88 guards: EdgeGuards::new(
89 RequestsPerSecond::new(nz32(1_000_000)),
90 BurstSize::new(nz32(1_000_000)),
91 MaxInflightRequests::new(nz32(10_000)),
92 RequestTimeout::from_millis(nz64(120_000)),
93 BodyInactivityTimeout::from_millis(nz64(120_000)),
94 WriteRequestTimeout::from_millis(nz64(1_800_000)),
95 knot_types::ProxyTrust::default(),
96 ),
97 tls: Some(TlsSetup {
98 source: CertSource::Static(StaticCertPaths {
99 cert_path: knot_edge::CertChainPath::new(cert),
100 key_path: knot_edge::PrivateKeyPath::new(key),
101 }),
102 http3: true,
103 internal: None,
104 }),
105 }
106}
107
108struct Edge {
109 addr: SocketAddr,
110 shutdown: CancellationToken,
111 log: Arc<Mutex<Vec<Captured>>>,
112 task: JoinHandle<Result<(), knot_edge::EdgeError>>,
113 client: quinn::Endpoint,
114}
115
116async fn probe_identity(
117 endpoint: &quinn::Endpoint,
118 addr: SocketAddr,
119 expected_cert: &[u8],
120) -> Option<bool> {
121 let connecting = endpoint.connect(addr, "localhost").ok()?;
122 let connection = tokio::time::timeout(Duration::from_millis(250), connecting)
123 .await
124 .ok()?
125 .ok()?;
126 let ours = connection
127 .peer_identity()
128 .and_then(|identity| identity.downcast::<Vec<CertificateDer<'static>>>().ok())
129 .map(|certs| {
130 certs
131 .first()
132 .is_some_and(|cert| cert.as_ref() == expected_cert)
133 })
134 .unwrap_or(false);
135 connection.close(0u32.into(), b"probe done");
136 Some(ours)
137}
138
139async fn await_ready(
140 endpoint: &quinn::Endpoint,
141 addr: SocketAddr,
142 task: &mut JoinHandle<Result<(), knot_edge::EdgeError>>,
143 expected_cert: &[u8],
144) -> bool {
145 for _ in 0..200 {
146 if task.is_finished() {
147 return false;
148 }
149 if let Some(ours) = probe_identity(endpoint, addr, expected_cert).await {
150 return ours;
151 }
152 tokio::time::sleep(Duration::from_millis(20)).await;
153 }
154 false
155}
156
157async fn stand_up(layout: Layout, certdir: &Path) -> Edge {
158 static TRACE: std::sync::Once = std::sync::Once::new();
159 TRACE.call_once(|| {
160 let _ = tracing_subscriber::fmt()
161 .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
162 .try_init();
163 });
164 for _ in 0..8 {
165 let addr: SocketAddr = format!("127.0.0.1:{}", free_port()).parse().unwrap();
166 let (cert, key, cert_der) = write_self_signed(certdir);
167 let (write_routes, advertisement) = knot_pack::edge_routes(knot_pack::EdgeConfig {
168 pack_slots: knot_resource::PackSlots::new(4),
169 ..knot_pack::EdgeConfig::serving(
170 layout.clone(),
171 serve_dids(),
172 Arc::new(knot_runtime::SystemClock),
173 )
174 });
175 let log: Arc<Mutex<Vec<Captured>>> = Arc::new(Mutex::new(Vec::new()));
176 let sink = log.clone();
177 let recorded = write_routes.layer(axum::middleware::from_fn(
178 move |request: axum::extract::Request, next: axum::middleware::Next| {
179 let sink = sink.clone();
180 async move {
181 let (parts, body) = request.into_parts();
182 let bytes = axum::body::to_bytes(body, usize::MAX)
183 .await
184 .unwrap_or_default();
185 sink.lock().unwrap().push((
186 parts.method.clone(),
187 parts.uri.clone(),
188 parts.headers.clone(),
189 bytes.clone(),
190 ));
191 next.run(axum::extract::Request::from_parts(
192 parts,
193 axum::body::Body::from(bytes),
194 ))
195 .await
196 }
197 },
198 ));
199 let app = RequiresFullHandshake::new(recorded);
200 let shutdown = CancellationToken::new();
201 let mut task = tokio::spawn(knot_edge::serve(
202 edge_config(addr, cert, key),
203 app,
204 advertisement,
205 shutdown.clone(),
206 ));
207 let client = h3_client();
208 if await_ready(&client, addr, &mut task, &cert_der).await {
209 return Edge {
210 addr,
211 shutdown,
212 log,
213 task,
214 client,
215 };
216 }
217 client.close(0u32.into(), b"stand up retry");
218 shutdown.cancel();
219 let _ = task.await;
220 }
221 panic!("couldn't bind a free TCP+UDP port for the edge after several attempts");
222}
223
224#[derive(Debug)]
225struct AcceptAnyServerCert;
226
227impl ServerCertVerifier for AcceptAnyServerCert {
228 fn verify_server_cert(
229 &self,
230 _end_entity: &CertificateDer<'_>,
231 _intermediates: &[CertificateDer<'_>],
232 _server_name: &ServerName<'_>,
233 _ocsp_response: &[u8],
234 _now: UnixTime,
235 ) -> Result<ServerCertVerified, rustls::Error> {
236 Ok(ServerCertVerified::assertion())
237 }
238
239 fn verify_tls12_signature(
240 &self,
241 message: &[u8],
242 cert: &CertificateDer<'_>,
243 dss: &DigitallySignedStruct,
244 ) -> Result<HandshakeSignatureValid, rustls::Error> {
245 rustls::crypto::verify_tls12_signature(
246 message,
247 cert,
248 dss,
249 &aws_lc_rs::default_provider().signature_verification_algorithms,
250 )
251 }
252
253 fn verify_tls13_signature(
254 &self,
255 message: &[u8],
256 cert: &CertificateDer<'_>,
257 dss: &DigitallySignedStruct,
258 ) -> Result<HandshakeSignatureValid, rustls::Error> {
259 rustls::crypto::verify_tls13_signature(
260 message,
261 cert,
262 dss,
263 &aws_lc_rs::default_provider().signature_verification_algorithms,
264 )
265 }
266
267 fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
268 aws_lc_rs::default_provider()
269 .signature_verification_algorithms
270 .supported_schemes()
271 }
272}
273
274fn h3_client() -> quinn::Endpoint {
275 let mut crypto =
276 rustls::ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
277 .with_protocol_versions(&[&rustls::version::TLS13])
278 .unwrap()
279 .dangerous()
280 .with_custom_certificate_verifier(Arc::new(AcceptAnyServerCert))
281 .with_no_client_auth();
282 crypto.alpn_protocols = vec![b"h3".to_vec()];
283 let quic = quinn::crypto::rustls::QuicClientConfig::try_from(crypto).unwrap();
284 let mut endpoint = quinn::Endpoint::client("127.0.0.1:0".parse().unwrap()).unwrap();
285 endpoint.set_default_client_config(quinn::ClientConfig::new(Arc::new(quic)));
286 endpoint
287}
288
289async fn drain(
290 stream: &mut h3::client::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
291) -> Vec<u8> {
292 let mut out = Vec::new();
293 while let Some(mut chunk) = stream.recv_data().await.unwrap() {
294 out.extend_from_slice(&chunk.copy_to_bytes(chunk.remaining()));
295 }
296 out
297}
298
299async fn finish_request(
300 stream: &mut h3::client::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
301) {
302 match stream.finish().await {
303 Ok(()) => (),
304 Err(h3::error::StreamError::RemoteTerminate { code, .. })
305 if code == h3::error::Code::H3_NO_ERROR => {}
306 Err(error) => panic!("finishing the request stream failed: {error}"),
307 }
308}
309
310async fn replay_over_h3(edge: &Edge, did: &str, request: &Captured) -> Vec<u8> {
311 let (_, uri, headers, body) = request;
312 let connection = edge
313 .client
314 .connect(edge.addr, "localhost")
315 .unwrap()
316 .await
317 .unwrap();
318 let quic = connection.clone();
319 let (mut driver, mut sender) = h3::client::new(h3_quinn::Connection::new(connection))
320 .await
321 .unwrap();
322 let drive = tokio::spawn(async move {
323 let _ = std::future::poll_fn(|cx| driver.poll_close(cx)).await;
324 });
325
326 let warmup = http::Request::get(format!(
327 "https://localhost/{did}/info/refs?service=git-upload-pack"
328 ))
329 .header("git-protocol", "version=2")
330 .body(())
331 .unwrap();
332 let mut warm = sender.send_request(warmup).await.unwrap();
333 finish_request(&mut warm).await;
334 assert!(
335 warm.recv_response().await.unwrap().status().is_success(),
336 "h3 info/refs advertisement must serve over QUIC"
337 );
338 drain(&mut warm).await;
339
340 let mut builder = http::Request::builder()
341 .method(Method::POST)
342 .uri(format!("https://localhost{}", uri.path()));
343 for name in ["content-type", "content-encoding", "git-protocol", "accept"] {
344 if let Some(value) = headers.get(name) {
345 builder = builder.header(name, value);
346 }
347 }
348 let mut stream = sender
349 .send_request(builder.body(()).unwrap())
350 .await
351 .unwrap();
352 stream.send_data(body.clone()).await.unwrap();
353 finish_request(&mut stream).await;
354 let response = stream.recv_response().await.unwrap();
355 assert!(
356 response.status().is_success(),
357 "h3 upload-pack returned {}",
358 response.status()
359 );
360 let out = drain(&mut stream).await;
361 quic.close(0u32.into(), b"done");
362 drive.abort();
363 out
364}
365
366fn fetch_request(log: &Arc<Mutex<Vec<Captured>>>) -> Captured {
367 let captured = log.lock().unwrap();
368 captured
369 .iter()
370 .find(|(method, uri, _, body)| {
371 method == Method::POST
372 && uri.path().ends_with("/git-upload-pack")
373 && contains(body, b"command=fetch")
374 })
375 .cloned()
376 .unwrap_or_else(|| {
377 let summary: Vec<String> = captured
378 .iter()
379 .map(|(method, uri, headers, body)| {
380 format!(
381 "{method} {uri} git-protocol={:?} body[..64]={:?}",
382 headers.get("git-protocol"),
383 String::from_utf8_lossy(&body[..body.len().min(64)])
384 )
385 })
386 .collect();
387 panic!("git issued no protocol-v2 fetch over the TLS edge, captured: {summary:#?}")
388 })
389}
390
391fn extract_pack(response: &[u8]) -> Vec<u8> {
392 let mut channel = Vec::new();
393 let mut pos = 0usize;
394 while pos + 4 <= response.len() {
395 let len = std::str::from_utf8(&response[pos..pos + 4])
396 .ok()
397 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
398 .unwrap_or(0);
399 pos += 4;
400 if len < 4 {
401 continue;
402 }
403 let end = (pos + len - 4).min(response.len());
404 let payload = &response[pos..end];
405 pos = end;
406 if payload.first() == Some(&1) {
407 channel.extend_from_slice(&payload[1..]);
408 }
409 }
410 match channel.windows(4).position(|window| window == b"PACK") {
411 Some(start) => channel.split_off(start),
412 None => channel,
413 }
414}
415
416fn index_pack(repo: &Path, pack: &[u8]) {
417 let (indexed, report) =
418 knot_fixtures::feed(repo, &["index-pack", "--stdin", "--fix-thin"], pack);
419 assert!(indexed, "index-pack failed: {report}");
420}
421
422fn advance_knot(bare: &Path, work: &Path, old: &str, new: &str) {
423 let oids: Vec<String> = must(work, &["rev-list", "--objects", new, "--not", old])
424 .lines()
425 .filter_map(|line| line.split_whitespace().next())
426 .map(str::to_string)
427 .collect();
428 let request = receive_request("refs/heads/main", old, new, &pack_objects(work, &oids));
429 let repo = knot_git::Repo::open(bare).expect("open knot bare");
430 let report = knot_pack::receive_pack(&repo, &request).expect("knot receive");
431 assert!(
432 String::from_utf8_lossy(&report).contains("ok refs/heads/main"),
433 "knot must accept a receive that advances main"
434 );
435}
436
437fn init(dir: &Path, format: ObjectFormat) {
438 std::fs::create_dir_all(dir).unwrap();
439 let fmt = format!("--object-format={}", format.capability());
440 must(dir, &["init", &fmt, "-q", dir.to_str().unwrap()]);
441}
442
443fn seed(work: &Path, bares: [&Path; 2], format: ObjectFormat) {
444 let fmt = format!("--object-format={}", format.capability());
445 std::fs::create_dir_all(work).unwrap();
446 must(work, &["init", &fmt, "-q", "-b", "main"]);
447 std::fs::write(work.join("README.md"), "h3 conformance\n").unwrap();
448 must(work, &["add", "-A"]);
449 must(work, &["commit", "-q", "-m", "c1"]);
450 let c1 = must(work, &["rev-parse", "HEAD"]);
451 std::fs::write(work.join("src.txt"), "more\n").unwrap();
452 must(work, &["add", "-A"]);
453 must(work, &["commit", "-q", "-m", "c2"]);
454 must(work, &["checkout", "-q", "-b", "dev", &c1]);
455 std::fs::write(work.join("dev.txt"), "branch\n").unwrap();
456 must(work, &["add", "-A"]);
457 must(work, &["commit", "-q", "-m", "c3"]);
458 must(work, &["checkout", "-q", "main"]);
459 must(work, &["tag", "-a", "v1", "-m", "release"]);
460 bares.into_iter().for_each(|bare| {
461 must(
462 work,
463 &["push", "-q", bare.to_str().unwrap(), "main", "dev", "v1"],
464 );
465 must(bare, &["symbolic-ref", "HEAD", "refs/heads/main"]);
466 });
467}
468
469async fn h3_serves_the_canonical_object_set(format: ObjectFormat, did_str: &str) {
470 let scan = tempfile::tempdir().unwrap();
471 let certdir = tempfile::tempdir().unwrap();
472 let scratch = tempfile::tempdir().unwrap();
473 let canon_root = tempfile::tempdir().unwrap();
474
475 let did = RepoDid::new(did_str).unwrap();
476 let layout = Layout::new(scan.path()).with_object_format(format);
477 layout.create(&did).unwrap();
478 let knot_bare = layout.repo_path(&did).unwrap();
479 let canon_bare = canon_root.path().join("canon.git");
480 let fmt = format!("--object-format={}", format.capability());
481 must(
482 canon_root.path(),
483 &["init", "--bare", &fmt, "-q", canon_bare.to_str().unwrap()],
484 );
485
486 let work = scratch.path().join("work");
487 seed(&work, [&knot_bare, &canon_bare], format);
488
489 let canon_url = format!("file://{}", canon_bare.to_str().unwrap());
490 let canon_clone = scratch.path().join("canon-clone");
491 must(
492 scratch.path(),
493 &["clone", "-q", &canon_url, canon_clone.to_str().unwrap()],
494 );
495 let canonical = object_set(&canon_clone);
496
497 let edge = stand_up(layout, certdir.path()).await;
498 let url = format!("https://{}/{}", edge.addr, did.as_str());
499
500 let h1_clone = scratch.path().join("h1-clone");
501 must(
502 scratch.path(),
503 &[
504 "-c",
505 "http.sslVerify=false",
506 "clone",
507 "-q",
508 &url,
509 h1_clone.to_str().unwrap(),
510 ],
511 );
512 must(&h1_clone, &["config", "http.sslVerify", "false"]);
513 assert_eq!(
514 must(&canon_clone, &["rev-parse", "HEAD^{tree}"]),
515 must(&h1_clone, &["rev-parse", "HEAD^{tree}"]),
516 "{format:?} h1/h2 TLS clone checks out the canonical tree"
517 );
518 assert_eq!(
519 canonical,
520 object_set(&h1_clone),
521 "{format:?} h1/h2 TLS clone transfers the canonical object set"
522 );
523
524 let h3_clone = scratch.path().join("h3-clone");
525 init(&h3_clone, format);
526 let pack = extract_pack(&replay_over_h3(&edge, did.as_str(), &fetch_request(&edge.log)).await);
527 index_pack(&h3_clone, &pack);
528 assert_eq!(
529 canonical,
530 object_set(&h3_clone),
531 "{format:?} h3 clone over QUIC transfers the canonical object set"
532 );
533
534 let old = must(&work, &["rev-parse", "HEAD"]);
535 std::fs::write(work.join("incremental.txt"), "fetch me\n").unwrap();
536 must(&work, &["add", "-A"]);
537 must(&work, &["commit", "-q", "-m", "c4"]);
538 let new = must(&work, &["rev-parse", "HEAD"]);
539 advance_knot(&knot_bare, &work, &old, &new);
540 must(&work, &["push", "-q", canon_bare.to_str().unwrap(), "main"]);
541
542 let canon_after = scratch.path().join("canon-after");
543 must(
544 scratch.path(),
545 &["clone", "-q", &canon_url, canon_after.to_str().unwrap()],
546 );
547 let canonical_after = object_set(&canon_after);
548
549 edge.log.lock().unwrap().clear();
550 must(&h1_clone, &["fetch", "-q", "origin"]);
551 assert_eq!(
552 canonical_after,
553 object_set(&h1_clone),
554 "{format:?} h1/h2 TLS fetch advances to the canonical object set"
555 );
556
557 let fetch_pack =
558 extract_pack(&replay_over_h3(&edge, did.as_str(), &fetch_request(&edge.log)).await);
559 index_pack(&h3_clone, &fetch_pack);
560 assert_eq!(
561 canonical_after,
562 object_set(&h3_clone),
563 "{format:?} h3 incremental fetch over QUIC advances to the canonical object set"
564 );
565
566 edge.shutdown.cancel();
567 edge.task.abort();
568}
569
570#[tokio::test(flavor = "multi_thread")]
571async fn the_git_superset_guarantee_holds_over_h3_in_both_object_formats() {
572 h3_serves_the_canonical_object_set(ObjectFormat::SHA1, "did:plc:squid").await;
573 h3_serves_the_canonical_object_set(ObjectFormat::SHA256, "did:plc:cuttle").await;
574}