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