alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod acme;
2mod altsvc;
3mod compression;
4mod limits;
5mod peer;
6mod protocol;
7mod quic;
8mod robustness;
9mod tcp;
10mod tls;
11mod zerortt;
12
13use std::future::Future;
14use std::net::SocketAddr;
15use std::path::{Path, PathBuf};
16use std::pin::Pin;
17use std::sync::Arc;
18
19use axum::Router;
20use axum::middleware::from_fn;
21use rustls::server::ResolvesServerCert;
22use tokio::net::TcpListener;
23use tokio_util::sync::CancellationToken;
24
25pub use acme::{AcmeCacheDir, AcmeContact, AcmeContactError, AcmeError, AcmeParams};
26pub use limits::{
27 ConnectionBudget, HeaderTimeout, IdleTimeout, ListenLimits, MaxConcurrentStreams,
28};
29pub use peer::SocketPeer;
30pub use protocol::NegotiatedProtocol;
31pub use quic::EndpointError;
32pub use robustness::{
33 BodyInactivityTimeout, BurstSize, EdgeGuards, MaxInflightRequests, RequestTimeout,
34 RequestsPerSecond, WriteRequestTimeout,
35};
36pub use tls::{ReloadableCertResolver, SpkiPin, TlsError, load_certified_key};
37pub use zerortt::{EarlyData, RequiresFullHandshake, ZeroRttRoutes, ZeroRttSafe};
38
39pub mod fuzz {
40 pub fn spki_of_certificate(data: &[u8]) {
41 crate::tls::fuzz_of_certificate(data);
42 }
43
44 pub fn spki_pin(data: &[u8]) {
45 let _ = crate::SpkiPin::from_base64(&String::from_utf8_lossy(data));
46 }
47}
48
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct CertChainPath(PathBuf);
51
52impl CertChainPath {
53 pub fn new(path: impl Into<PathBuf>) -> Self {
54 Self(path.into())
55 }
56
57 pub fn as_path(&self) -> &Path {
58 &self.0
59 }
60}
61
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct PrivateKeyPath(PathBuf);
64
65impl PrivateKeyPath {
66 pub fn new(path: impl Into<PathBuf>) -> Self {
67 Self(path.into())
68 }
69
70 pub fn as_path(&self) -> &Path {
71 &self.0
72 }
73}
74
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub struct ClientCaPath(PathBuf);
77
78impl ClientCaPath {
79 pub fn new(path: impl Into<PathBuf>) -> Self {
80 Self(path.into())
81 }
82
83 pub fn as_path(&self) -> &Path {
84 &self.0
85 }
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub struct PublicBind(SocketAddr);
90
91impl PublicBind {
92 pub const fn new(addr: SocketAddr) -> Self {
93 Self(addr)
94 }
95
96 pub const fn get(self) -> SocketAddr {
97 self.0
98 }
99}
100
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
102pub struct InternalBind(SocketAddr);
103
104impl InternalBind {
105 pub const fn new(addr: SocketAddr) -> Self {
106 Self(addr)
107 }
108
109 pub const fn get(self) -> SocketAddr {
110 self.0
111 }
112}
113
114pub struct StaticCertPaths {
115 pub cert_path: CertChainPath,
116 pub key_path: PrivateKeyPath,
117}
118
119pub enum CertSource {
120 Static(StaticCertPaths),
121 Acme(AcmeParams),
122}
123
124pub struct InternalTls {
125 pub addr: InternalBind,
126 pub client_ca_path: ClientCaPath,
127 pub spki_pin: SpkiPin,
128}
129
130pub struct TlsSetup {
131 pub source: CertSource,
132 pub http3: bool,
133 pub internal: Option<InternalTls>,
134}
135
136pub struct EdgeConfig {
137 pub http_addr: PublicBind,
138 pub limits: ListenLimits,
139 pub guards: EdgeGuards,
140 pub tls: Option<TlsSetup>,
141}
142
143#[derive(Debug, thiserror::Error)]
144pub enum EdgeError {
145 #[error(transparent)]
146 Io(#[from] std::io::Error),
147 #[error(transparent)]
148 Tls(#[from] TlsError),
149 #[error(transparent)]
150 Acme(#[from] AcmeError),
151 #[error(transparent)]
152 Endpoint(#[from] EndpointError),
153}
154
155type Served = Pin<Box<dyn Future<Output = Result<(), EdgeError>> + Send>>;
156
157fn base_router(app: RequiresFullHandshake, early_data_safe: ZeroRttRoutes) -> Router {
158 early_data_safe
159 .into_router()
160 .merge(app.into_router())
161 .layer(compression::layer())
162 .layer(from_fn(zerortt::tag_from_header))
163}
164
165fn finish(router: Router) -> Router {
166 altsvc::with_host_from_authority(router.layer(from_fn(protocol::tag)))
167}
168
169pub async fn serve(
170 config: EdgeConfig,
171 app: RequiresFullHandshake,
172 early_data_safe: ZeroRttRoutes,
173 shutdown: CancellationToken,
174) -> Result<(), EdgeError> {
175 let EdgeConfig {
176 http_addr,
177 limits,
178 guards,
179 tls,
180 } = config;
181 let layers = guards.prepare(&shutdown);
182 let early_data = early_data_safe.early_data_policy();
183 let wants_internal = tls
184 .as_ref()
185 .and_then(|setup| setup.internal.as_ref())
186 .is_some();
187 let base = base_router(app, early_data_safe);
188 let internal_router = wants_internal.then(|| finish(base.clone()));
189 let router = finish(robustness::apply(base, layers));
190 let listener = TcpListener::bind(http_addr.get()).await?;
191
192 let Some(setup) = tls else {
193 return Ok(tcp::serve_plaintext(listener, router, limits, shutdown).await?);
194 };
195
196 let (resolver, acme): (Arc<dyn ResolvesServerCert>, bool) = match setup.source {
197 CertSource::Static(paths) => {
198 let certified = tls::load_certified_key(&paths)?;
199 let reloadable = Arc::new(ReloadableCertResolver::new(certified));
200 tls::spawn_cert_reload(Arc::clone(&reloadable), paths, shutdown.clone());
201 (reloadable, false)
202 }
203 CertSource::Acme(params) => (acme::start(params, shutdown.clone())?, true),
204 };
205
206 let extra_alpn: &[&[u8]] = if acme { &[tls::ACME_TLS_ALPN] } else { &[] };
207 let tcp_config = Arc::new(tls::build_tls_server_config(
208 Arc::clone(&resolver),
209 extra_alpn,
210 )?);
211 let port = http_addr.get().port();
212
213 let mut servers: Vec<Served> = Vec::new();
214
215 servers.push({
216 let app = match setup.http3 {
217 true => altsvc::with_alt_svc(router.clone(), altsvc::Port::new(port)),
218 false => router.clone(),
219 };
220 let shutdown = shutdown.clone();
221 Box::pin(async move {
222 let result = tcp::serve_tls(listener, app, tcp_config, limits, shutdown.clone()).await;
223 shutdown.cancel();
224 Ok(result?)
225 })
226 });
227
228 if setup.http3 {
229 let endpoint =
230 quic::build_endpoint(http_addr.get(), Arc::clone(&resolver), limits, early_data)?;
231 let app = router.clone();
232 let shutdown = shutdown.clone();
233 servers.push(Box::pin(async move {
234 quic::serve_http3(endpoint, app, limits, shutdown.clone()).await;
235 shutdown.cancel();
236 Ok(())
237 }));
238 }
239
240 if let Some(internal) = setup.internal {
241 let internal_listener = TcpListener::bind(internal.addr.get()).await?;
242 let mtls_config = Arc::new(tls::build_mtls_server_config(
243 Arc::clone(&resolver),
244 &internal.client_ca_path,
245 internal.spki_pin,
246 )?);
247 let app = internal_router
248 .expect("an internal router is built whenever an internal bind is configured");
249 let shutdown = shutdown.clone();
250 servers.push(Box::pin(async move {
251 let result = tcp::serve_tls(
252 internal_listener,
253 app,
254 mtls_config,
255 limits,
256 shutdown.clone(),
257 )
258 .await;
259 shutdown.cancel();
260 Ok(result?)
261 }));
262 }
263
264 futures::future::join_all(servers)
265 .await
266 .into_iter()
267 .collect::<Result<Vec<()>, EdgeError>>()
268 .map(drop)
269}
270
271#[cfg(test)]
272mod tests {
273 use super::*;
274
275 use axum::body::Body;
276 use axum::extract::ConnectInfo;
277 use axum::routing::post;
278 use http::{Request, StatusCode};
279 use tower::ServiceExt;
280
281 use std::num::{NonZeroU32, NonZeroU64};
282
283 use zerortt::ZeroRttSafe;
284
285 fn test_layers() -> robustness::GuardLayers {
286 EdgeGuards::new(
287 RequestsPerSecond::new(NonZeroU32::new(10_000).unwrap()),
288 BurstSize::new(NonZeroU32::new(10_000).unwrap()),
289 MaxInflightRequests::new(NonZeroU32::new(1_024).unwrap()),
290 RequestTimeout::from_millis(NonZeroU64::new(30_000).unwrap()),
291 BodyInactivityTimeout::from_millis(NonZeroU64::new(30_000).unwrap()),
292 WriteRequestTimeout::from_millis(NonZeroU64::new(1_800_000).unwrap()),
293 knot_types::ProxyTrust::default(),
294 )
295 .prepare(&CancellationToken::new())
296 }
297
298 fn wired() -> Router {
299 let safe =
300 ZeroRttRoutes::new().get("/info/refs", ZeroRttSafe::new(|| async { "advertisement" }));
301 let full = RequiresFullHandshake::new(
302 Router::new().route("/git-upload-pack", post(|| async { "pack" })),
303 );
304 finish(robustness::apply(base_router(full, safe), test_layers()))
305 }
306
307 fn tight_layers() -> robustness::GuardLayers {
308 EdgeGuards::new(
309 RequestsPerSecond::new(NonZeroU32::new(1).unwrap()),
310 BurstSize::new(NonZeroU32::new(2).unwrap()),
311 MaxInflightRequests::new(NonZeroU32::new(1_024).unwrap()),
312 RequestTimeout::from_millis(NonZeroU64::new(30_000).unwrap()),
313 BodyInactivityTimeout::from_millis(NonZeroU64::new(30_000).unwrap()),
314 WriteRequestTimeout::from_millis(NonZeroU64::new(1_800_000).unwrap()),
315 knot_types::ProxyTrust::default(),
316 )
317 .prepare(&CancellationToken::new())
318 }
319
320 async fn status_of(mut request: Request<Body>) -> StatusCode {
321 request
322 .extensions_mut()
323 .insert(ConnectInfo(SocketAddr::from(([127, 0, 0, 1], 41001))));
324 wired().oneshot(request).await.unwrap().status()
325 }
326
327 #[tokio::test]
328 async fn a_write_in_early_data_is_refused_with_425() {
329 let request = Request::post("/git-upload-pack")
330 .header("early-data", "1")
331 .body(Body::empty())
332 .unwrap();
333 assert_eq!(status_of(request).await, StatusCode::TOO_EARLY);
334 }
335
336 #[tokio::test]
337 async fn a_write_after_the_handshake_is_served() {
338 let request = Request::post("/git-upload-pack")
339 .body(Body::empty())
340 .unwrap();
341 assert_eq!(status_of(request).await, StatusCode::OK);
342 }
343
344 #[tokio::test]
345 async fn the_advertisement_is_served_even_in_early_data() {
346 let request = Request::get("/info/refs")
347 .header("early-data", "1")
348 .body(Body::empty())
349 .unwrap();
350 assert_eq!(status_of(request).await, StatusCode::OK);
351 }
352
353 #[tokio::test]
354 async fn the_internal_admin_router_shares_no_rate_limit_budget_with_the_data_plane() {
355 let safe = ZeroRttRoutes::new().get("/info/refs", ZeroRttSafe::new(|| async { "ok" }));
356 let full = RequiresFullHandshake::new(
357 Router::new().route("/git-upload-pack", post(|| async { "pack" })),
358 );
359 let base = base_router(full, safe);
360 let public = finish(robustness::apply(base.clone(), tight_layers()));
361 let internal = finish(base);
362
363 let request = || {
364 let mut request = Request::get("/info/refs").body(Body::empty()).unwrap();
365 request
366 .extensions_mut()
367 .insert(ConnectInfo(SocketAddr::from(([127, 0, 0, 1], 41001))));
368 request
369 };
370
371 let p1 = public.clone().oneshot(request()).await.unwrap().status();
372 let p2 = public.clone().oneshot(request()).await.unwrap().status();
373 let p3 = public.clone().oneshot(request()).await.unwrap().status();
374 assert_eq!([p1, p2], [StatusCode::OK, StatusCode::OK]);
375 assert_eq!(
376 p3,
377 StatusCode::TOO_MANY_REQUESTS,
378 "the public edge still enforces the per-IP burst"
379 );
380
381 let i1 = internal.clone().oneshot(request()).await.unwrap().status();
382 let i2 = internal.clone().oneshot(request()).await.unwrap().status();
383 let i3 = internal.clone().oneshot(request()).await.unwrap().status();
384 let i4 = internal.clone().oneshot(request()).await.unwrap().status();
385 assert_eq!(
386 [i1, i2, i3, i4],
387 [StatusCode::OK; 4],
388 "the internal admin bind is unguarded, so a public flood never sheds admin requests"
389 );
390 }
391}