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::{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 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 && contains(body, b"command=fetch")
377 })
378 .cloned()
379 .unwrap_or_else(|| {
380 let summary: Vec<String> = captured
381 .iter()
382 .map(|(method, uri, headers, body)| {
383 format!(
384 "{method} {uri} git-protocol={:?} body[..64]={:?}",
385 headers.get("git-protocol"),
386 String::from_utf8_lossy(&body[..body.len().min(64)])
387 )
388 })
389 .collect();
390 panic!("git issued no protocol-v2 fetch over the TLS edge, captured: {summary:#?}")
391 })
392}
393
394fn extract_pack(response: &[u8]) -> Vec<u8> {
395 let mut channel = Vec::new();
396 let mut pos = 0usize;
397 while pos + 4 <= response.len() {
398 let len = std::str::from_utf8(&response[pos..pos + 4])
399 .ok()
400 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
401 .unwrap_or(0);
402 pos += 4;
403 if len < 4 {
404 continue;
405 }
406 let end = (pos + len - 4).min(response.len());
407 let payload = &response[pos..end];
408 pos = end;
409 if payload.first() == Some(&1) {
410 channel.extend_from_slice(&payload[1..]);
411 }
412 }
413 match channel.windows(4).position(|window| window == b"PACK") {
414 Some(start) => channel.split_off(start),
415 None => channel,
416 }
417}
418
419fn index_pack(repo: &Path, pack: &[u8]) {
420 let (indexed, report) =
421 knot_fixtures::feed(repo, &["index-pack", "--stdin", "--fix-thin"], pack);
422 assert!(indexed, "index-pack failed: {report}");
423}
424
425fn advance_knot(bare: &Path, work: &Path, old: &str, new: &str) {
426 let oids: Vec<String> = must(work, &["rev-list", "--objects", new, "--not", old])
427 .lines()
428 .filter_map(|line| line.split_whitespace().next())
429 .map(str::to_string)
430 .collect();
431 let request = receive_request("refs/heads/main", old, new, &pack_objects(work, &oids));
432 let repo = knot_git::Repo::open(bare).expect("open knot bare");
433 let report = knot_pack::receive_pack(&repo, &request).expect("knot receive");
434 assert!(
435 String::from_utf8_lossy(&report).contains("ok refs/heads/main"),
436 "knot must accept a receive that advances main"
437 );
438}
439
440fn init(dir: &Path, format: ObjectFormat) {
441 std::fs::create_dir_all(dir).unwrap();
442 let fmt = format!("--object-format={}", format.capability());
443 must(dir, &["init", &fmt, "-q", dir.to_str().unwrap()]);
444}
445
446fn seed(work: &Path, bares: [&Path; 2], format: ObjectFormat) {
447 let fmt = format!("--object-format={}", format.capability());
448 std::fs::create_dir_all(work).unwrap();
449 must(work, &["init", &fmt, "-q", "-b", "main"]);
450 std::fs::write(work.join("README.md"), "h3 conformance\n").unwrap();
451 must(work, &["add", "-A"]);
452 must(work, &["commit", "-q", "-m", "c1"]);
453 let c1 = must(work, &["rev-parse", "HEAD"]);
454 std::fs::write(work.join("src.txt"), "more\n").unwrap();
455 must(work, &["add", "-A"]);
456 must(work, &["commit", "-q", "-m", "c2"]);
457 must(work, &["checkout", "-q", "-b", "dev", &c1]);
458 std::fs::write(work.join("dev.txt"), "branch\n").unwrap();
459 must(work, &["add", "-A"]);
460 must(work, &["commit", "-q", "-m", "c3"]);
461 must(work, &["checkout", "-q", "main"]);
462 must(work, &["tag", "-a", "v1", "-m", "release"]);
463 bares.into_iter().for_each(|bare| {
464 must(
465 work,
466 &["push", "-q", bare.to_str().unwrap(), "main", "dev", "v1"],
467 );
468 must(bare, &["symbolic-ref", "HEAD", "refs/heads/main"]);
469 });
470}
471
472async fn h3_serves_the_canonical_object_set(format: ObjectFormat, did_str: &str) {
473 let scan = tempfile::tempdir().unwrap();
474 let certdir = tempfile::tempdir().unwrap();
475 let scratch = tempfile::tempdir().unwrap();
476 let canon_root = tempfile::tempdir().unwrap();
477
478 let did = RepoDid::new(did_str).unwrap();
479 let layout = Layout::new(scan.path()).with_object_format(format);
480 layout.create(&did).unwrap();
481 let knot_bare = layout.repo_path(&did).unwrap();
482 let canon_bare = canon_root.path().join("canon.git");
483 let fmt = format!("--object-format={}", format.capability());
484 must(
485 canon_root.path(),
486 &["init", "--bare", &fmt, "-q", canon_bare.to_str().unwrap()],
487 );
488
489 let work = scratch.path().join("work");
490 seed(&work, [&knot_bare, &canon_bare], format);
491
492 let canon_url = format!("file://{}", canon_bare.to_str().unwrap());
493 let canon_clone = scratch.path().join("canon-clone");
494 must(
495 scratch.path(),
496 &["clone", "-q", &canon_url, canon_clone.to_str().unwrap()],
497 );
498 let canonical = object_set(&canon_clone);
499
500 let edge = stand_up(layout, certdir.path()).await;
501 let url = format!("https://{}/{}", edge.addr, did.as_str());
502
503 let h1_clone = scratch.path().join("h1-clone");
504 must(
505 scratch.path(),
506 &[
507 "-c",
508 "http.sslVerify=false",
509 "clone",
510 "-q",
511 &url,
512 h1_clone.to_str().unwrap(),
513 ],
514 );
515 must(&h1_clone, &["config", "http.sslVerify", "false"]);
516 assert_eq!(
517 must(&canon_clone, &["rev-parse", "HEAD^{tree}"]),
518 must(&h1_clone, &["rev-parse", "HEAD^{tree}"]),
519 "{format:?} h1/h2 TLS clone checks out the canonical tree"
520 );
521 assert_eq!(
522 canonical,
523 object_set(&h1_clone),
524 "{format:?} h1/h2 TLS clone transfers the canonical object set"
525 );
526
527 let h3_clone = scratch.path().join("h3-clone");
528 init(&h3_clone, format);
529 let pack = extract_pack(&replay_over_h3(&edge, did.as_str(), &fetch_request(&edge.log)).await);
530 index_pack(&h3_clone, &pack);
531 assert_eq!(
532 canonical,
533 object_set(&h3_clone),
534 "{format:?} h3 clone over QUIC transfers the canonical object set"
535 );
536
537 let old = must(&work, &["rev-parse", "HEAD"]);
538 std::fs::write(work.join("incremental.txt"), "fetch me\n").unwrap();
539 must(&work, &["add", "-A"]);
540 must(&work, &["commit", "-q", "-m", "c4"]);
541 let new = must(&work, &["rev-parse", "HEAD"]);
542 advance_knot(&knot_bare, &work, &old, &new);
543 must(&work, &["push", "-q", canon_bare.to_str().unwrap(), "main"]);
544
545 let canon_after = scratch.path().join("canon-after");
546 must(
547 scratch.path(),
548 &["clone", "-q", &canon_url, canon_after.to_str().unwrap()],
549 );
550 let canonical_after = object_set(&canon_after);
551
552 edge.log.lock().unwrap().clear();
553 must(&h1_clone, &["fetch", "-q", "origin"]);
554 assert_eq!(
555 canonical_after,
556 object_set(&h1_clone),
557 "{format:?} h1/h2 TLS fetch advances to the canonical object set"
558 );
559
560 let fetch_pack =
561 extract_pack(&replay_over_h3(&edge, did.as_str(), &fetch_request(&edge.log)).await);
562 index_pack(&h3_clone, &fetch_pack);
563 assert_eq!(
564 canonical_after,
565 object_set(&h3_clone),
566 "{format:?} h3 incremental fetch over QUIC advances to the canonical object set"
567 );
568
569 edge.shutdown.cancel();
570 edge.task.abort();
571}
572
573#[tokio::test(flavor = "multi_thread")]
574async fn the_git_superset_guarantee_holds_over_h3_in_both_object_formats() {
575 h3_serves_the_canonical_object_set(ObjectFormat::SHA1, "did:plc:squid").await;
576 h3_serves_the_canonical_object_set(ObjectFormat::SHA256, "did:plc:cuttle").await;
577}