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::{CacheConfig, 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::{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 None,
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(
168 layout.clone(),
169 serve_dids(),
170 None,
171 None,
172 knot_resource::PackSlots::new(4),
173 CacheConfig::default(),
174 Arc::new(knot_messages::Catalog::defaults()),
175 knot_pack::default_hostname().clone(),
176 Arc::new(knot_runtime::SystemClock),
177 );
178 let log: Arc<Mutex<Vec<Captured>>> = Arc::new(Mutex::new(Vec::new()));
179 let sink = log.clone();
180 let recorded = write_routes.layer(axum::middleware::from_fn(
181 move |request: axum::extract::Request, next: axum::middleware::Next| {
182 let sink = sink.clone();
183 async move {
184 let (parts, body) = request.into_parts();
185 let bytes = axum::body::to_bytes(body, usize::MAX)
186 .await
187 .unwrap_or_default();
188 sink.lock().unwrap().push((
189 parts.method.clone(),
190 parts.uri.clone(),
191 parts.headers.clone(),
192 bytes.clone(),
193 ));
194 next.run(axum::extract::Request::from_parts(
195 parts,
196 axum::body::Body::from(bytes),
197 ))
198 .await
199 }
200 },
201 ));
202 let app = RequiresFullHandshake::new(recorded);
203 let shutdown = CancellationToken::new();
204 let mut task = tokio::spawn(knot_edge::serve(
205 edge_config(addr, cert, key),
206 app,
207 advertisement,
208 shutdown.clone(),
209 ));
210 let client = h3_client();
211 if await_ready(&client, addr, &mut task, &cert_der).await {
212 return Edge {
213 addr,
214 shutdown,
215 log,
216 task,
217 client,
218 };
219 }
220 client.close(0u32.into(), b"stand up retry");
221 shutdown.cancel();
222 let _ = task.await;
223 }
224 panic!("couldn't bind a free TCP+UDP port for the edge after several attempts");
225}
226
227#[derive(Debug)]
228struct AcceptAnyServerCert;
229
230impl ServerCertVerifier for AcceptAnyServerCert {
231 fn verify_server_cert(
232 &self,
233 _end_entity: &CertificateDer<'_>,
234 _intermediates: &[CertificateDer<'_>],
235 _server_name: &ServerName<'_>,
236 _ocsp_response: &[u8],
237 _now: UnixTime,
238 ) -> Result<ServerCertVerified, rustls::Error> {
239 Ok(ServerCertVerified::assertion())
240 }
241
242 fn verify_tls12_signature(
243 &self,
244 message: &[u8],
245 cert: &CertificateDer<'_>,
246 dss: &DigitallySignedStruct,
247 ) -> Result<HandshakeSignatureValid, rustls::Error> {
248 rustls::crypto::verify_tls12_signature(
249 message,
250 cert,
251 dss,
252 &aws_lc_rs::default_provider().signature_verification_algorithms,
253 )
254 }
255
256 fn verify_tls13_signature(
257 &self,
258 message: &[u8],
259 cert: &CertificateDer<'_>,
260 dss: &DigitallySignedStruct,
261 ) -> Result<HandshakeSignatureValid, rustls::Error> {
262 rustls::crypto::verify_tls13_signature(
263 message,
264 cert,
265 dss,
266 &aws_lc_rs::default_provider().signature_verification_algorithms,
267 )
268 }
269
270 fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
271 aws_lc_rs::default_provider()
272 .signature_verification_algorithms
273 .supported_schemes()
274 }
275}
276
277fn h3_client() -> quinn::Endpoint {
278 let mut crypto =
279 rustls::ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
280 .with_protocol_versions(&[&rustls::version::TLS13])
281 .unwrap()
282 .dangerous()
283 .with_custom_certificate_verifier(Arc::new(AcceptAnyServerCert))
284 .with_no_client_auth();
285 crypto.alpn_protocols = vec![b"h3".to_vec()];
286 let quic = quinn::crypto::rustls::QuicClientConfig::try_from(crypto).unwrap();
287 let mut endpoint = quinn::Endpoint::client("127.0.0.1:0".parse().unwrap()).unwrap();
288 endpoint.set_default_client_config(quinn::ClientConfig::new(Arc::new(quic)));
289 endpoint
290}
291
292async fn drain(
293 stream: &mut h3::client::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
294) -> Vec<u8> {
295 let mut out = Vec::new();
296 while let Some(mut chunk) = stream.recv_data().await.unwrap() {
297 out.extend_from_slice(&chunk.copy_to_bytes(chunk.remaining()));
298 }
299 out
300}
301
302async fn finish_request(
303 stream: &mut h3::client::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
304) {
305 match stream.finish().await {
306 Ok(()) => (),
307 Err(h3::error::StreamError::RemoteTerminate { code, .. })
308 if code == h3::error::Code::H3_NO_ERROR => {}
309 Err(error) => panic!("finishing the request stream failed: {error}"),
310 }
311}
312
313async fn replay_over_h3(edge: &Edge, did: &str, request: &Captured) -> Vec<u8> {
314 let (_, uri, headers, body) = request;
315 let connection = edge
316 .client
317 .connect(edge.addr, "localhost")
318 .unwrap()
319 .await
320 .unwrap();
321 let quic = connection.clone();
322 let (mut driver, mut sender) = h3::client::new(h3_quinn::Connection::new(connection))
323 .await
324 .unwrap();
325 let drive = tokio::spawn(async move {
326 let _ = std::future::poll_fn(|cx| driver.poll_close(cx)).await;
327 });
328
329 let warmup = http::Request::get(format!(
330 "https://localhost/{did}/info/refs?service=git-upload-pack"
331 ))
332 .header("git-protocol", "version=2")
333 .body(())
334 .unwrap();
335 let mut warm = sender.send_request(warmup).await.unwrap();
336 finish_request(&mut warm).await;
337 assert!(
338 warm.recv_response().await.unwrap().status().is_success(),
339 "h3 info/refs advertisement must serve over QUIC"
340 );
341 drain(&mut warm).await;
342
343 let mut builder = http::Request::builder()
344 .method(Method::POST)
345 .uri(format!("https://localhost{}", uri.path()));
346 for name in ["content-type", "content-encoding", "git-protocol", "accept"] {
347 if let Some(value) = headers.get(name) {
348 builder = builder.header(name, value);
349 }
350 }
351 let mut stream = sender
352 .send_request(builder.body(()).unwrap())
353 .await
354 .unwrap();
355 stream.send_data(body.clone()).await.unwrap();
356 finish_request(&mut stream).await;
357 let response = stream.recv_response().await.unwrap();
358 assert!(
359 response.status().is_success(),
360 "h3 upload-pack returned {}",
361 response.status()
362 );
363 let out = drain(&mut stream).await;
364 quic.close(0u32.into(), b"done");
365 drive.abort();
366 out
367}
368
369fn fetch_request(log: &Arc<Mutex<Vec<Captured>>>) -> Captured {
370 let captured = log.lock().unwrap();
371 captured
372 .iter()
373 .find(|(method, uri, _, body)| {
374 method == Method::POST
375 && uri.path().ends_with("/git-upload-pack")
376 && body
377 .windows(b"command=fetch".len())
378 .any(|window| window == b"command=fetch")
379 })
380 .cloned()
381 .unwrap_or_else(|| {
382 let summary: Vec<String> = captured
383 .iter()
384 .map(|(method, uri, headers, body)| {
385 format!(
386 "{method} {uri} git-protocol={:?} body[..64]={:?}",
387 headers.get("git-protocol"),
388 String::from_utf8_lossy(&body[..body.len().min(64)])
389 )
390 })
391 .collect();
392 panic!("git issued no protocol-v2 fetch over the TLS edge, captured: {summary:#?}")
393 })
394}
395
396fn extract_pack(response: &[u8]) -> Vec<u8> {
397 let mut channel = Vec::new();
398 let mut pos = 0usize;
399 while pos + 4 <= response.len() {
400 let len = std::str::from_utf8(&response[pos..pos + 4])
401 .ok()
402 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
403 .unwrap_or(0);
404 pos += 4;
405 if len < 4 {
406 continue;
407 }
408 let end = (pos + len - 4).min(response.len());
409 let payload = &response[pos..end];
410 pos = end;
411 if payload.first() == Some(&1) {
412 channel.extend_from_slice(&payload[1..]);
413 }
414 }
415 match channel.windows(4).position(|window| window == b"PACK") {
416 Some(start) => channel.split_off(start),
417 None => channel,
418 }
419}
420
421fn index_pack(repo: &Path, pack: &[u8]) {
422 let (indexed, report) =
423 knot_fixtures::feed(repo, &["index-pack", "--stdin", "--fix-thin"], pack);
424 assert!(indexed, "index-pack failed: {report}");
425}
426
427fn advance_knot(bare: &Path, work: &Path, old: &str, new: &str) {
428 let oids: Vec<String> = must(work, &["rev-list", "--objects", new, "--not", old])
429 .lines()
430 .filter_map(|line| line.split_whitespace().next())
431 .map(str::to_string)
432 .collect();
433 let request = receive_request("refs/heads/main", old, new, &pack_objects(work, &oids));
434 let repo = knot_git::Repo::open(bare).expect("open knot bare");
435 let report = knot_pack::receive_pack(&repo, &request).expect("knot receive");
436 assert!(
437 String::from_utf8_lossy(&report).contains("ok refs/heads/main"),
438 "knot must accept a receive that advances main"
439 );
440}
441
442fn init(dir: &Path, format: ObjectFormat) {
443 std::fs::create_dir_all(dir).unwrap();
444 let fmt = format!("--object-format={}", format.capability());
445 must(dir, &["init", &fmt, "-q", dir.to_str().unwrap()]);
446}
447
448fn seed(work: &Path, bares: [&Path; 2], format: ObjectFormat) {
449 let fmt = format!("--object-format={}", format.capability());
450 std::fs::create_dir_all(work).unwrap();
451 must(work, &["init", &fmt, "-q", "-b", "main"]);
452 std::fs::write(work.join("README.md"), "h3 conformance\n").unwrap();
453 must(work, &["add", "-A"]);
454 must(work, &["commit", "-q", "-m", "c1"]);
455 let c1 = must(work, &["rev-parse", "HEAD"]);
456 std::fs::write(work.join("src.txt"), "more\n").unwrap();
457 must(work, &["add", "-A"]);
458 must(work, &["commit", "-q", "-m", "c2"]);
459 must(work, &["checkout", "-q", "-b", "dev", &c1]);
460 std::fs::write(work.join("dev.txt"), "branch\n").unwrap();
461 must(work, &["add", "-A"]);
462 must(work, &["commit", "-q", "-m", "c3"]);
463 must(work, &["checkout", "-q", "main"]);
464 must(work, &["tag", "-a", "v1", "-m", "release"]);
465 bares.into_iter().for_each(|bare| {
466 must(
467 work,
468 &["push", "-q", bare.to_str().unwrap(), "main", "dev", "v1"],
469 );
470 must(bare, &["symbolic-ref", "HEAD", "refs/heads/main"]);
471 });
472}
473
474async fn h3_serves_the_canonical_object_set(format: ObjectFormat, did_str: &str) {
475 let scan = tempfile::tempdir().unwrap();
476 let certdir = tempfile::tempdir().unwrap();
477 let scratch = tempfile::tempdir().unwrap();
478 let canon_root = tempfile::tempdir().unwrap();
479
480 let did = RepoDid::new(did_str).unwrap();
481 let layout = Layout::new(scan.path()).with_object_format(format);
482 layout.create(&did).unwrap();
483 let knot_bare = layout.repo_path(&did).unwrap();
484 let canon_bare = canon_root.path().join("canon.git");
485 let fmt = format!("--object-format={}", format.capability());
486 must(
487 canon_root.path(),
488 &["init", "--bare", &fmt, "-q", canon_bare.to_str().unwrap()],
489 );
490
491 let work = scratch.path().join("work");
492 seed(&work, [&knot_bare, &canon_bare], format);
493
494 let canon_url = format!("file://{}", canon_bare.to_str().unwrap());
495 let canon_clone = scratch.path().join("canon-clone");
496 must(
497 scratch.path(),
498 &["clone", "-q", &canon_url, canon_clone.to_str().unwrap()],
499 );
500 let canonical = object_set(&canon_clone);
501
502 let edge = stand_up(layout, certdir.path()).await;
503 let url = format!("https://{}/{}", edge.addr, did.as_str());
504
505 let h1_clone = scratch.path().join("h1-clone");
506 must(
507 scratch.path(),
508 &[
509 "-c",
510 "http.sslVerify=false",
511 "clone",
512 "-q",
513 &url,
514 h1_clone.to_str().unwrap(),
515 ],
516 );
517 must(&h1_clone, &["config", "http.sslVerify", "false"]);
518 assert_eq!(
519 must(&canon_clone, &["rev-parse", "HEAD^{tree}"]),
520 must(&h1_clone, &["rev-parse", "HEAD^{tree}"]),
521 "{format:?} h1/h2 TLS clone checks out the canonical tree"
522 );
523 assert_eq!(
524 canonical,
525 object_set(&h1_clone),
526 "{format:?} h1/h2 TLS clone transfers the canonical object set"
527 );
528
529 let h3_clone = scratch.path().join("h3-clone");
530 init(&h3_clone, format);
531 let pack = extract_pack(&replay_over_h3(&edge, did.as_str(), &fetch_request(&edge.log)).await);
532 index_pack(&h3_clone, &pack);
533 assert_eq!(
534 canonical,
535 object_set(&h3_clone),
536 "{format:?} h3 clone over QUIC transfers the canonical object set"
537 );
538
539 let old = must(&work, &["rev-parse", "HEAD"]);
540 std::fs::write(work.join("incremental.txt"), "fetch me\n").unwrap();
541 must(&work, &["add", "-A"]);
542 must(&work, &["commit", "-q", "-m", "c4"]);
543 let new = must(&work, &["rev-parse", "HEAD"]);
544 advance_knot(&knot_bare, &work, &old, &new);
545 must(&work, &["push", "-q", canon_bare.to_str().unwrap(), "main"]);
546
547 let canon_after = scratch.path().join("canon-after");
548 must(
549 scratch.path(),
550 &["clone", "-q", &canon_url, canon_after.to_str().unwrap()],
551 );
552 let canonical_after = object_set(&canon_after);
553
554 edge.log.lock().unwrap().clear();
555 must(&h1_clone, &["fetch", "-q", "origin"]);
556 assert_eq!(
557 canonical_after,
558 object_set(&h1_clone),
559 "{format:?} h1/h2 TLS fetch advances to the canonical object set"
560 );
561
562 let fetch_pack =
563 extract_pack(&replay_over_h3(&edge, did.as_str(), &fetch_request(&edge.log)).await);
564 index_pack(&h3_clone, &fetch_pack);
565 assert_eq!(
566 canonical_after,
567 object_set(&h3_clone),
568 "{format:?} h3 incremental fetch over QUIC advances to the canonical object set"
569 );
570
571 edge.shutdown.cancel();
572 edge.task.abort();
573}
574
575#[tokio::test(flavor = "multi_thread")]
576async fn the_git_superset_guarantee_holds_over_h3_in_both_object_formats() {
577 h3_serves_the_canonical_object_set(ObjectFormat::SHA1, "did:plc:squid").await;
578 h3_serves_the_canonical_object_set(ObjectFormat::SHA256, "did:plc:cuttle").await;
579}