alice.pds.demo.boltless.dev
This repository has no description
29 kB
752 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod allocator;
2
3#[global_allocator]
4static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
5
6#[allow(non_upper_case_globals)]
7#[unsafe(export_name = "_rjem_malloc_conf")]
8pub static malloc_conf: &[u8] =
9 b"background_thread:true,retain:false,dirty_decay_ms:0,muzzy_decay_ms:0\0";
10
11use std::collections::BTreeSet;
12use std::num::{NonZeroU32, NonZeroU64};
13use std::path::PathBuf;
14use std::sync::Arc;
15use std::time::Duration;
16
17use tokio_util::sync::CancellationToken;
18
19use anyhow::Context;
20use axum::Json;
21use axum::response::Html;
22use axum::routing::get;
23use base64::Engine;
24use knot_atproto::Atproto;
25use knot_config::HomepageSource;
26use knot_index::Index;
27use knot_runtime::{Clock, HttpTransport, OsEntropy, ReqwestHttp, SystemClock};
28use knot_secrets::{MasterKey, SealedStore};
29use knot_types::{ActorId, AuthorName, BranchName, CiLogsAddr, Email, KnotHostname, ObjectCount};
30use knot_xrpc::XrpcState;
31use tower_http::services::ServeFile;
32
33const MAINTENANCE_SHUTDOWN_DRAIN: Duration = Duration::from_secs(30);
34const EDGE_SHUTDOWN_DRAIN: Duration = Duration::from_secs(40);
35
36const DEFAULT_HOMEPAGE: &str = include_str!("homepage.html");
37
38struct IndexRepos(Arc<Index>);
39
40impl knot_maintenance::RepoSource for IndexRepos {
41 fn repos(&self) -> Vec<knot_types::RepoDid> {
42 self.0.hosted_repos()
43 }
44
45 fn ready_repos(&self) -> Option<Vec<knot_types::RepoDid>> {
46 match self.0.coverage().registry {
47 knot_index::Coverage::Ready => Some(self.0.hosted_repos()),
48 knot_index::Coverage::Warming => None,
49 }
50 }
51}
52
53struct AtprotoHandleResolver<H, C> {
54 atproto: Arc<Atproto<H, C>>,
55}
56
57impl<H: HttpTransport, C: Clock> knot_pack::HandleResolver for AtprotoHandleResolver<H, C> {
58 fn resolve(
59 &self,
60 handle: knot_types::Handle,
61 ) -> std::pin::Pin<
62 Box<dyn std::future::Future<Output = Option<knot_types::AccountDid>> + Send + '_>,
63 > {
64 Box::pin(async move { self.atproto.resolve_handle_to_did(&handle).await.ok() })
65 }
66}
67
68fn init_tracing() {
69 let filter = tracing_subscriber::EnvFilter::try_from_default_env()
70 .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
71 tracing_subscriber::fmt()
72 .with_env_filter(filter)
73 .with_writer(std::io::stderr)
74 .init();
75}
76
77fn subcommand(name: &str) -> Option<anyhow::Result<()>> {
78 match name {
79 "config-template" => {
80 print!("{}", knot_config::template());
81 Some(Ok(()))
82 }
83 "validate" => Some(
84 knot_config::load(std::env::args().nth(2).map(PathBuf::from).as_deref())
85 .context("load configuration")
86 .and_then(|config| {
87 config
88 .verify_environment()
89 .context("verify runtime environment")
90 })
91 .map(|()| println!("configuration is valid")),
92 ),
93 _ => None,
94 }
95}
96
97#[tokio::main]
98async fn main() -> anyhow::Result<()> {
99 #[cfg(target_os = "linux")]
100 rustix::process::set_dumpable_behavior(rustix::process::DumpableBehavior::NotDumpable)
101 .context("disable core dumps and ptrace attachment")?;
102
103 if let Some(result) = std::env::args().nth(1).as_deref().and_then(subcommand) {
104 return result;
105 }
106
107 init_tracing();
108
109 tracing::info!("!");
110 tracing::info!("!");
111 tracing::info!("!");
112 tracing::info!("> If knot1 was so good then why isn't there a... ( ˶°ㅁ°)");
113 tracing::info!("...");
114 tracing::info!("Welcome to knot2!");
115 tracing::info!("This code was made with love.");
116 tracing::info!("Hachapuri is sho tasty, definitely worth a try. Better than pizza tbh.");
117 tracing::info!("!");
118 tracing::info!("!");
119 tracing::info!("!");
120
121 let config_path = std::env::args().nth(1).map(PathBuf::from);
122 let config = knot_config::load(config_path.as_deref()).context("load configuration")?;
123 config
124 .verify_environment()
125 .context("verify runtime environment")?;
126
127 let resources = knot_resource::init(knot_resource::Ceilings {
128 max_threads: match config.resources.max_threads {
129 0 => None,
130 n => Some(knot_resource::ThreadCount::new(n as usize)),
131 },
132 max_memory: match config.resources.max_memory_bytes {
133 0 => None,
134 n => Some(knot_resource::MemoryBudget::new(n)),
135 },
136 });
137 tracing::info!(
138 threads = resources.threads.get(),
139 memory_bytes = resources.memory.map(knot_resource::MemoryBudget::get),
140 memory_source = ?resources.memory_source,
141 memory_high_bytes = resources
142 .memory_high_bytes
143 .map(knot_resource::MemoryHighBytes::get),
144 "resource governor initialized"
145 );
146 allocator::install();
147
148 // I made the hostname stay aa string in the config,
149 // so that confique can layer env-over-file.
150 // Here's where it becomes a real type,
151 // and the DID + the service url stack on this.
152 let hostname = KnotHostname::new(config.server.hostname.clone())
153 .context("server.hostname isn't a valid knot hostname")?;
154 let knot_did = hostname.knot_did();
155 let object_format = config.object_format().context("parse git.object_format")?;
156 let default_branch =
157 BranchName::new(config.repo.default_branch.as_str()).context("parse default branch")?;
158 let layout = knot_git::Layout::new(&config.repo.scan_path)
159 .with_default_branch(default_branch)
160 .with_object_format(object_format)
161 .reserving_meta(&knot_did)
162 .context("reserve meta-repo path")?;
163
164 let swept = knot_pack::sweep_incoming(&config.repo.scan_path);
165 if swept > 0 {
166 tracing::info!(swept, "swept abandoned receive staging directories");
167 }
168
169 layout
170 .bootstrap_meta(&knot_did)
171 .context("bootstrap meta-repo")?;
172 let meta_path = layout
173 .meta_path(&knot_did)
174 .context("resolve meta-repo path")?;
175
176 let index = Arc::new(Index::new(meta_path.clone(), layout.clone()));
177 index.rebuild().context("rebuild index from meta-repo")?;
178 tracing::info!(coverage = ?index.coverage(), "index ready");
179
180 let warm = Arc::clone(&index);
181 tokio::task::spawn_blocking(move || warm.warm_collaborators());
182
183 let http = ReqwestHttp::new(config.http_limits()).context("build outbound HTTP client")?;
184 let git_http: Arc<dyn knot_runtime::HttpTransport> = Arc::new(
185 ReqwestHttp::new(config.fork_http_limits()).context("build outbound git fetch client")?,
186 );
187 let atproto = Arc::new(Atproto::new(
188 http,
189 SystemClock,
190 knot_did.clone(),
191 knot_atproto::PlcDirectory::new(config.atproto.plc_directory.clone())
192 .context("atproto.plc_directory isn't a valid PLC base URL")?,
193 ));
194 let admins: BTreeSet<_> = config.server.admins.iter().cloned().collect();
195 let admission = config.acl.admission;
196 let service_owner = config
197 .server
198 .admins
199 .first()
200 .cloned()
201 .context("at least one admin is configured")?;
202
203 let master_key = MasterKey::new(
204 base64::engine::general_purpose::STANDARD
205 .decode(
206 std::env::var(&config.secrets.master_key_env)
207 .context("read master key from environment")?
208 .trim(),
209 )
210 .context("decode master key as base64")?,
211 )
212 .context("master key from environment")?;
213 let secrets = Arc::new(
214 SealedStore::open(
215 &config.secrets.sealed_key_file,
216 &master_key,
217 Box::new(OsEntropy),
218 )
219 .context("open sealed key store")?,
220 );
221 let knot_signing_key = secrets
222 .ensure(&knot_did)
223 .context("seal knot's own signing key")?;
224 let knot_actor = ActorId::from_secp256k1(knot_signing_key.as_bytes());
225 let appview_endpoint = config.server.appview_endpoint.clone();
226 let knot_service_url = knot_types::KnotServiceUrl::new(format!("https://{hostname}"))
227 .context("server.hostname doesn't form a valid knot service URL")?;
228 let did_document =
229 knot_atproto::knot_did_document(&knot_did, &knot_signing_key, &knot_service_url);
230
231 let http_addr = config.server.listen_addr;
232 let listen_limits = knot_edge::ListenLimits::new(
233 knot_edge::HeaderTimeout::from_millis(
234 NonZeroU64::new(config.server.listen_header_timeout_ms)
235 .context("server.listen_header_timeout_ms must be greater than zero")?,
236 ),
237 knot_edge::IdleTimeout::from_millis(
238 NonZeroU64::new(config.server.listen_idle_timeout_ms)
239 .context("server.listen_idle_timeout_ms must be greater than zero")?,
240 ),
241 NonZeroU32::new(config.server.listen_max_connections)
242 .context("server.listen_max_connections must be greater than zero")?,
243 );
244 // A header name that doesn't parse will never match,
245 // `effective_peer` falls back to socket,
246 // and every request in the world shares
247 // the proxy's address + its one ratelimit bucket.
248 // So... better to refuse to start.
249 let trusted_proxy_header = config
250 .xrpc
251 .trusted_proxy_header
252 .as_deref()
253 .map(|header| axum::http::HeaderName::from_bytes(header.as_bytes()))
254 .transpose()
255 .context("xrpc.trusted_proxy_header isn't a valid HTTP header name")?;
256 let edge_guards = knot_edge::EdgeGuards::new(
257 knot_edge::RequestsPerSecond::new(
258 NonZeroU32::new(config.server.listen_rate_limit_per_second)
259 .context("server.listen_rate_limit_per_second must be greater than zero")?,
260 ),
261 knot_edge::BurstSize::new(
262 NonZeroU32::new(config.server.listen_rate_limit_burst)
263 .context("server.listen_rate_limit_burst must be greater than zero")?,
264 ),
265 knot_edge::MaxInflightRequests::new(
266 NonZeroU32::new(config.server.listen_max_inflight_requests)
267 .context("server.listen_max_inflight_requests must be greater than zero")?,
268 ),
269 knot_edge::RequestTimeout::from_millis(
270 NonZeroU64::new(config.server.listen_request_timeout_ms)
271 .context("server.listen_request_timeout_ms must be greater than zero")?,
272 ),
273 knot_edge::BodyInactivityTimeout::from_millis(
274 NonZeroU64::new(config.server.listen_body_timeout_ms)
275 .context("server.listen_body_timeout_ms must be greater than zero")?,
276 ),
277 knot_edge::WriteRequestTimeout::from_millis(
278 NonZeroU64::new(config.server.listen_write_request_timeout_ms)
279 .context("server.listen_write_request_timeout_ms must be greater than zero")?,
280 ),
281 trusted_proxy_header.clone(),
282 );
283 let tls_setup = build_tls_setup(&config, &hostname).context("assemble TLS configuration")?;
284 if config.tls.http3 && tls_setup.is_none() {
285 tracing::warn!(
286 "tls.http3 is set without any TLS certificate, so HTTP/3 won't start. Configure a static cert or ACME to serve h3."
287 );
288 }
289 if tls_setup.is_none() && config.xrpc.trusted_proxy_header.is_none() {
290 tracing::warn!(
291 "running plaintext behind a reverse proxy without xrpc.trusted_proxy_header. Per-IP rate limiting will key on the proxy socket address, throttling all clients as one. Set xrpc.trusted_proxy_header to the header your proxy appends."
292 );
293 }
294 if config.tls.acme_enabled && http_addr.port() != 443 {
295 tracing::warn!(
296 listen_port = http_addr.port(),
297 "ACME validation over TLS-ALPN-01 needs the certificate authority to reach this host on TCP 443. Map 443 to the listen port if it differs."
298 );
299 }
300 if config.tls.acme_enabled && config.tls.acme_staging {
301 tracing::warn!(
302 "ACME is using the Let's Encrypt staging directory. Its certificates aren't browser-trusted. Unset tls.acme_staging for real certificates."
303 );
304 }
305 let ssh_addr = config.server.ssh_listen_addr;
306 let ssh_max_pack_bytes = config.server.ssh_max_pack_bytes as usize;
307 let pack_limits = knot_pack::PackLimits {
308 max_objects: ObjectCount::from(config.pack.max_objects),
309 max_total_bytes: knot_pack::MaxTotalBytes::new(config.pack.max_total_bytes),
310 ..knot_pack::PackLimits::default()
311 };
312 knot_pack::init_selection_limits(knot_pack::SelectionLimits {
313 max_objects: ObjectCount::from(config.pack.selection_max_objects),
314 time_budget: Duration::from_secs(config.pack.selection_time_budget_secs),
315 });
316 let host_key = knot_ssh::load_or_create_host_key(&config.server.ssh_host_key_file)
317 .context("load or create SSH host key")?;
318
319 let xrpc_limits = knot_xrpc::LimitConfig {
320 rate: Some(knot_xrpc::RateLimit {
321 burst: knot_xrpc::Burst::new(config.xrpc.preauth_burst),
322 refill: knot_xrpc::RefillMicros::new(
323 config.xrpc.preauth_refill_ms.saturating_mul(1_000),
324 ),
325 }),
326 per_peer_inflight: Some(knot_xrpc::PerPeerInflight::new(
327 config.xrpc.per_peer_inflight as usize,
328 )),
329 global_inflight: Some(knot_xrpc::GlobalInflight::new(
330 config.xrpc.global_inflight as usize,
331 )),
332 };
333 let byte_limits = knot_xrpc::ByteLimits {
334 body: knot_xrpc::BodyLimit::new(config.xrpc.max_body_bytes as usize),
335 patch: knot_xrpc::PatchLimit::new(config.xrpc.max_patch_bytes as usize),
336 patch_decompressed: knot_xrpc::PatchDecompressedLimit::new(
337 config.xrpc.max_patch_decompressed_bytes,
338 ),
339 response: knot_xrpc::ResponseLimit::new(config.xrpc.max_response_bytes as usize),
340 archive: knot_xrpc::ArchiveLimit::new(config.xrpc.max_archive_bytes),
341 fork_pack: knot_xrpc::ForkPackLimit::new(config.xrpc.fork_max_pack_bytes),
342 pack: knot_xrpc::MaxWireBytes::new(ssh_max_pack_bytes),
343 };
344 let budgets = knot_xrpc::Budgets {
345 tree_last_commit: knot_xrpc::TreeReadBudget::new(knot_xrpc::ReadBudget::Within(
346 Duration::from_millis(config.xrpc.tree_last_commit_budget_ms),
347 )),
348 blob_last_commit: knot_xrpc::BlobReadBudget::new(knot_xrpc::ReadBudget::Within(
349 Duration::from_millis(config.xrpc.blob_last_commit_budget_ms),
350 )),
351 languages: knot_xrpc::LanguagesReadBudget::new(knot_xrpc::ReadBudget::Within(
352 Duration::from_millis(config.xrpc.languages_budget_ms),
353 )),
354 languages_push: knot_xrpc::LanguagesPushBudget::new(Duration::from_millis(
355 config.xrpc.languages_push_budget_ms,
356 )),
357 };
358 let committer = knot_xrpc::Committer {
359 name: AuthorName::new(config.git.user_name.clone()),
360 email: Email::new(config.git.user_email.clone()),
361 };
362 let reservations = Arc::new(knot_xrpc::Reservations::new(
363 knot_xrpc::ReservationTtl::new(config.xrpc.reservation_ttl_secs as i64),
364 knot_xrpc::PerActorQuota::new(config.xrpc.per_actor_reservations as usize),
365 knot_xrpc::GlobalQuota::new(config.xrpc.max_pending_reservations as usize),
366 ));
367 let replay_bounds = knot_events::ReplayBounds::new(
368 knot_events::ReplayEvents::new(config.xrpc.events_replay_buffer as usize)
369 .context("xrpc.events_replay_buffer must be greater than zero")?,
370 knot_events::ReplayBytes::new(config.xrpc.events_replay_bytes as usize)
371 .context("xrpc.events_replay_bytes must be greater than zero")?,
372 );
373 let events = Arc::new(knot_events::EventLog::new(SystemClock, replay_bounds));
374 let subscriber_gate = Arc::new(knot_events::SubscriberGate::new(
375 knot_events::GlobalSubscriberLimit::new(config.xrpc.events_max_subscribers as usize),
376 knot_events::PerPeerSubscriberLimit::new(config.xrpc.events_max_per_peer as usize),
377 ));
378
379 let maintenance_enabled = config.maintenance.enabled;
380 let maintenance_options = knot_maintenance::Options::from_config(&config.maintenance);
381 let maintenance_interval = Duration::from_secs(config.maintenance.interval_secs);
382 let maintenance_large_push =
383 knot_maintenance::PushBytes::new(config.maintenance.large_push_bytes);
384
385 let lfs_handle = config
386 .lfs
387 .store_path
388 .as_ref()
389 .map(|path| {
390 knot_lfs::LfsHandle::open(
391 knot_lfs::LfsStorePath::new(path),
392 knot_lfs::LfsSize::new(config.lfs.max_object_bytes),
393 knot_lfs::FreeSpaceFloor::new(config.lfs.free_space_floor_bytes),
394 )
395 .inspect(|_| {
396 tracing::info!(store = %path.display(), "git-lfs capability enabled");
397 })
398 })
399 .transpose()
400 .context("open LFS object store")?;
401 let lfs_max_ssh_transfers = config.lfs.max_ssh_transfers as usize;
402 let lfs_max_http_downloads = config.lfs.max_http_downloads as usize;
403 let lfs_gc_grace = knot_maintenance::lfs_grace(
404 knot_maintenance::GcGrace::from_secs(config.lfs.gc_grace_secs),
405 knot_maintenance::ReflogRetention::from_secs(config.maintenance.reflog_expire_secs),
406 );
407 let lfs_gc_interval =
408 knot_maintenance::SweepInterval::new(Duration::from_secs(config.lfs.gc_interval_secs));
409 if lfs_handle.is_some() && !maintenance_enabled {
410 tracing::warn!(
411 "LFS store configured but maintenance is disabled. Unreferenced LFS objects will accumulate with no garbage collection or orphan sweep."
412 );
413 }
414
415 let pack_cache_config = knot_pack::CacheConfig {
416 enabled: config.pack_cache.enabled,
417 ttl: Duration::from_secs(config.pack_cache.ttl_secs),
418 max_entry_bytes: knot_pack::MaxEntryBytes::new(config.pack_cache.max_entry_bytes as usize),
419 max_total_bytes: knot_pack::MaxCacheBytes::new(knot_resource::pack_cache_bytes(
420 config.pack_cache.max_total_bytes,
421 ) as usize),
422 };
423
424 let ci_logs = config
425 .ci
426 .logs_addr
427 .as_deref()
428 .map(CiLogsAddr::new)
429 .transpose()
430 .context("ci.logs_addr must be host:port")?;
431
432 let homepage = config.homepage.source();
433 let catalog = Arc::new(
434 knot_messages::Catalog::parse(&config.messages).context("parse message templates")?,
435 );
436 let legacy_admin = legacy_admin_secret(&config);
437
438 knot_config::init(config);
439
440 let (maintenance_handle, maintenance_shutdown, maintenance_task) = if maintenance_enabled {
441 let (scheduler, handle) = knot_maintenance::Scheduler::new(
442 layout.clone(),
443 Arc::new(IndexRepos(Arc::clone(&index))),
444 SystemClock,
445 maintenance_options,
446 maintenance_interval,
447 maintenance_large_push,
448 );
449 let scheduler = match &lfs_handle {
450 Some(lfs) => {
451 scheduler.with_lfs_gc(Arc::clone(&lfs.store), lfs_gc_grace, lfs_gc_interval)
452 }
453 None => scheduler,
454 };
455 let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
456 let task = tokio::spawn(scheduler.run(shutdown_rx));
457 tracing::info!(
458 interval_secs = maintenance_interval.as_secs(),
459 "maintenance scheduler running"
460 );
461 (handle, Some(shutdown_tx), Some(task))
462 } else {
463 (knot_maintenance::MaintenanceHandle::disabled(), None, None)
464 };
465
466 let slots = knot_resource::Slots::for_machine();
467
468 let ssh_base = knot_ssh::SshState::new(
469 layout.clone(),
470 Arc::clone(&index),
471 Arc::clone(&atproto),
472 knot_actor,
473 Arc::clone(&events),
474 hostname.clone(),
475 appview_endpoint.clone(),
476 admins.clone(),
477 admission,
478 byte_limits.pack,
479 budgets.languages_push,
480 ci_logs.clone(),
481 )
482 .with_maintenance(maintenance_handle.clone())
483 .with_limits(pack_limits)
484 .with_slots(slots.clone())
485 .with_catalog(Arc::clone(&catalog));
486 let ssh_state = Arc::new(match &lfs_handle {
487 Some(handle) => ssh_base.with_lfs(handle.clone(), lfs_max_ssh_transfers),
488 None => ssh_base,
489 });
490
491 let xrpc_state = Arc::new(XrpcState {
492 layout: layout.clone(),
493 index: Arc::clone(&index),
494 atproto: Arc::clone(&atproto),
495 secrets,
496 entropy: Arc::new(OsEntropy),
497 ci_logs,
498 admins,
499 admission,
500 knot_did,
501 knot_hostname: hostname,
502 meta_path,
503 knot_service_url,
504 limiter: Arc::new(knot_xrpc::PreAuthLimiter::with_config(xrpc_limits)),
505 cob_locks: Arc::new(knot_xrpc::CobLocks::default()),
506 reservations,
507 trusted_proxy_header,
508 committer,
509 byte_limits,
510 budgets,
511 git_http,
512 pack_limits,
513 service_owner,
514 subscriber_gate,
515 maintenance: maintenance_handle,
516 appview: appview_endpoint,
517 slots: slots.clone(),
518 events,
519 lfs: lfs_handle.map(|handle| knot_xrpc::LfsWeb::new(handle, lfs_max_http_downloads)),
520 catalog: Arc::clone(&catalog),
521 });
522
523 let resolver: Arc<dyn knot_pack::RepoResolver> = {
524 let index = Arc::clone(&index);
525 Arc::new(move |target: &knot_pack::RepoTarget| match target {
526 knot_pack::RepoTarget::Did(did) => {
527 knot_pack::RepoLookup::from_resolved(index.owner_of(did), |_| did.clone())
528 }
529 knot_pack::RepoTarget::OwnerRkey(owner, rkey) => {
530 knot_pack::RepoLookup::from_resolved(index.resolve_repo(owner, rkey), |found| found)
531 }
532 })
533 };
534 let receive_advertiser = knot_xrpc::receive_advertiser(Arc::clone(&xrpc_state));
535 let handle_resolver: Arc<dyn knot_pack::HandleResolver> = Arc::new(AtprotoHandleResolver {
536 atproto: Arc::clone(&atproto),
537 });
538 let (write_routes, early_data_safe) = knot_pack::edge_routes(
539 layout,
540 resolver,
541 Some(receive_advertiser),
542 Some(handle_resolver),
543 slots.pack.clone(),
544 pack_cache_config,
545 Arc::clone(&catalog),
546 xrpc_state.knot_hostname.clone(),
547 Arc::new(SystemClock),
548 );
549 let legacy_admin_routes = legacy_admin.map(|secret| {
550 tracing::warn!(
551 route = knot_xrpc::legacy_admin::ADD_MEMBER_ROUTE,
552 "serving the legacy basic-auth admin route"
553 );
554 knot_xrpc::legacy_admin::router(Arc::clone(&xrpc_state), secret)
555 });
556 let base_router = write_routes.merge(knot_xrpc::router(xrpc_state)).route(
557 "/.well-known/did.json",
558 get(move || {
559 let document = did_document.clone();
560 async move { Json(document) }
561 }),
562 );
563 let base_router = match legacy_admin_routes {
564 Some(routes) => base_router.merge(routes),
565 None => base_router,
566 };
567 let base_router = match homepage {
568 HomepageSource::Disabled => base_router,
569 HomepageSource::Default => base_router.route("/", get(|| async { Html(DEFAULT_HOMEPAGE) })),
570 HomepageSource::File(path) => base_router.route_service("/", ServeFile::new(path)),
571 };
572 let app = knot_edge::RequiresFullHandshake::new(base_router);
573 let scheme = if tls_setup.is_some() { "https" } else { "http" };
574 let edge_config = knot_edge::EdgeConfig {
575 http_addr: knot_edge::PublicBind::new(http_addr),
576 limits: listen_limits,
577 guards: edge_guards,
578 tls: tls_setup,
579 };
580 tracing::info!("listening on {scheme}://{http_addr} and ssh://{ssh_addr}");
581
582 let shutdown = CancellationToken::new();
583 tokio::spawn(allocator::govern_decay(shutdown.clone()));
584 let mut edge_task = tokio::spawn(knot_edge::serve(
585 edge_config,
586 app,
587 early_data_safe,
588 shutdown.clone(),
589 ));
590 let mut ssh_task = {
591 let shutdown = shutdown.clone();
592 tokio::spawn(async move { knot_ssh::serve(ssh_addr, host_key, ssh_state, shutdown).await })
593 };
594
595 let exit = tokio::select! {
596 result = &mut edge_task => FirstExit::Edge(result),
597 result = &mut ssh_task => FirstExit::Ssh(result),
598 () = shutdown_signal() => {
599 tracing::info!("shutdown signal received");
600 FirstExit::Signal
601 }
602 };
603 shutdown.cancel();
604 let drain = async {
605 match &exit {
606 FirstExit::Edge(_) => {
607 let _ = (&mut ssh_task).await;
608 }
609 FirstExit::Ssh(_) => {
610 let _ = (&mut edge_task).await;
611 }
612 FirstExit::Signal => {
613 let _ = (&mut edge_task).await;
614 let _ = (&mut ssh_task).await;
615 }
616 }
617 };
618 if tokio::time::timeout(EDGE_SHUTDOWN_DRAIN, drain)
619 .await
620 .is_err()
621 {
622 tracing::warn!(
623 timeout_secs = EDGE_SHUTDOWN_DRAIN.as_secs(),
624 "aborting edge drain after timeout"
625 );
626 }
627 if let Some(shutdown) = maintenance_shutdown {
628 let _ = shutdown.send(true);
629 }
630 if let Some(task) = maintenance_task
631 && tokio::time::timeout(MAINTENANCE_SHUTDOWN_DRAIN, task)
632 .await
633 .is_err()
634 {
635 tracing::warn!(
636 timeout_secs = MAINTENANCE_SHUTDOWN_DRAIN.as_secs(),
637 "aborting maintenance drain after timeout :3"
638 );
639 }
640 match exit {
641 FirstExit::Edge(result) => result
642 .context("edge server task panicked")?
643 .context("serve edge")?,
644 FirstExit::Ssh(result) => result
645 .context("ssh server task panicked")?
646 .context("serve ssh")?,
647 FirstExit::Signal => {}
648 }
649 Ok(())
650}
651
652enum FirstExit {
653 Edge(Result<Result<(), knot_edge::EdgeError>, tokio::task::JoinError>),
654 Ssh(Result<Result<(), knot_ssh::SshError>, tokio::task::JoinError>),
655 Signal,
656}
657
658fn legacy_admin_secret(
659 config: &knot_config::Validated,
660) -> Option<knot_xrpc::legacy_admin::LegacyAdminSecret> {
661 let name = config.acl.legacy_admin_secret_env.as_deref()?;
662 let value = zeroize::Zeroizing::new(std::env::var(name).unwrap_or_default());
663 match knot_xrpc::legacy_admin::LegacyAdminSecret::new(&value) {
664 Ok(secret) => Some(secret),
665 Err(_) => {
666 tracing::warn!(
667 secret_env = name,
668 "the environment variable in acl.legacy_admin_secret_env is unset or empty, so we won't serve the admin route"
669 );
670 None
671 }
672 }
673}
674
675fn build_tls_setup(
676 config: &knot_config::KnotConfig,
677 hostname: &knot_types::KnotHostname,
678) -> anyhow::Result<Option<knot_edge::TlsSetup>> {
679 let tls = &config.tls;
680 let source = if tls.acme_enabled {
681 knot_edge::CertSource::Acme(knot_edge::AcmeParams {
682 domains: vec![hostname.clone()],
683 contact: knot_edge::AcmeContact::new(
684 tls.acme_contact
685 .clone()
686 .context("tls.acme_contact is required when ACME is enabled")?,
687 )?,
688 cache_dir: knot_edge::AcmeCacheDir::new(
689 tls.acme_cache_dir
690 .clone()
691 .context("tls.acme_cache_dir is required when ACME is enabled")?,
692 ),
693 production: !tls.acme_staging,
694 })
695 } else {
696 match (&tls.cert_path, &tls.key_path) {
697 (Some(cert_path), Some(key_path)) => {
698 knot_edge::CertSource::Static(knot_edge::StaticCertPaths {
699 cert_path: knot_edge::CertChainPath::new(cert_path.clone()),
700 key_path: knot_edge::PrivateKeyPath::new(key_path.clone()),
701 })
702 }
703 _ => return Ok(None),
704 }
705 };
706
707 let internal = match tls.mtls_enabled {
708 true => Some(knot_edge::InternalTls {
709 addr: knot_edge::InternalBind::new(config.server.internal_listen_addr),
710 client_ca_path: knot_edge::ClientCaPath::new(
711 tls.mtls_client_ca_path
712 .clone()
713 .context("tls.mtls_client_ca_path is required when mTLS is enabled")?,
714 ),
715 spki_pin: knot_edge::SpkiPin::from_base64(
716 tls.mtls_admin_spki_pin
717 .as_deref()
718 .context("tls.mtls_admin_spki_pin is required when mTLS is enabled")?,
719 )
720 .context("parse tls.mtls_admin_spki_pin")?,
721 }),
722 false => None,
723 };
724
725 Ok(Some(knot_edge::TlsSetup {
726 source,
727 http3: tls.http3,
728 internal,
729 }))
730}
731
732async fn shutdown_signal() {
733 let interrupt = async {
734 let _ = tokio::signal::ctrl_c().await;
735 };
736 #[cfg(unix)]
737 let terminate = async {
738 match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
739 Ok(mut stream) => {
740 stream.recv().await;
741 }
742 Err(_) => std::future::pending::<()>().await,
743 }
744 };
745 #[cfg(not(unix))]
746 let terminate = std::future::pending::<()>();
747
748 tokio::select! {
749 () = interrupt => {}
750 () = terminate => {}
751 }
752}