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