//! OxygenHub: a forge for Oxygen repositories (`docs/spec.md`). //! //! One process, or several sharing a database: an axum router serving //! pages, the JSON API and the wire protocol (the web role); the background //! tasks (the worker role, singletons under a leader lock: docs/roles.md); //! SQLite or PostgreSQL for hub state; repositories as bare Oxygen stores. pub mod api; pub mod apt; pub mod assistants; pub mod auth; pub mod autoland; pub mod backups; pub mod bus; pub mod config; pub mod coverage; pub mod db; pub mod diffusion; pub mod drops; pub mod events; pub mod flow; pub mod headers; pub mod history_scan; pub mod hooks; pub mod images; pub mod insights; pub mod junit; pub mod keyless; pub mod keys; pub mod leader; pub mod levels; pub mod live; pub mod mail; pub mod maintenance; pub mod metrics; pub mod mirror; pub mod net; pub mod notify; pub mod oidc; pub mod protocol; pub mod provenance; pub mod push; pub mod push_protection; pub mod queue; pub mod render; pub mod reports; pub mod repos; pub mod runner_updates; pub mod secrets; pub mod setup; pub mod shutdown; pub mod signin; pub mod social; pub mod storage; pub mod swarm; pub mod tags; pub mod terraform; pub mod throttle; pub mod tlog; pub mod variables; pub mod web; pub mod wiki; use crate::config::Config; use crate::repos::Repos; use std::sync::Arc; /// Everything a request handler can reach. pub struct App { pub config: Config, pub db: db::Pool, pub repos: Arc, /// Wikis, one Oxygen repository per repository (hub spec §6.11). pub wikis: Arc, pub server: Arc, pub hub_key: keys::HubKey, pub cookie_key: auth::CookieKey, /// Seals AI assistants' tokens at rest (hub spec §13.4). pub assistant_key: assistants::SecretKey, /// Seals Flow secrets at rest (hub spec §13.2, #107). pub secret_key: assistants::SecretKey, /// The OIDC issuer for jobs (hub spec §13.5), made on first use. pub oidc: tokio::sync::OnceCell, /// Which assistants are answering which proposals, and recent calls /// per repository (hub spec §13.4). pub assistant_gate: std::sync::Mutex, /// Failed sign-ins and the like, per account and address (oxygenhub#221). pub throttle: std::sync::Mutex, /// Events for live pages and workers, here and in the other processes /// sharing the database (docs/roles.md, #273). pub bus: bus::Bus, /// Who is resolving which conflict together (hub spec §6.10). pub rooms: Arc, /// Outgoing mail: hub.toml's `[email]`, overridden by what the admin /// page saved (#101). pub mail: std::sync::RwLock, /// The image swarm's tracker (oxygenhub#144). pub swarm: swarm::Tracker, pub started: std::time::Instant, /// The transparency log (#238). pub tlog: Arc, /// What this process does: web, worker or both (docs/roles.md, #272). pub roles: config::Roles, /// The background tasks started here and the leader locks held (#272). pub leader: leader::Leader, } impl App { /// The rootfs images this hub keeps for runners (oxygenflow#39). pub fn images(&self) -> images::Store { images::Store::new(&self.config.data) } /// The OIDC issuer, its key loaded or made the first time it is asked for. pub async fn oidc(&self) -> anyhow::Result<&oidc::Issuer> { let keys = self.config.keys_dir(); self.oidc.get_or_try_init(|| async move { tokio::task::spawn_blocking(move || oidc::Issuer::load_or_create(&keys)).await? }).await } } pub type Shared = Arc; /// Open the data directory and build the application state. `init` must /// have run once; `serve` and the tests call this. pub async fn build(config: Config) -> anyhow::Result { let roles = config.roles()?; std::fs::create_dir_all(config.repos_dir())?; let db = db::open(&config).await?; let repos = Arc::new(Repos::new(config.repos_dir())); let wikis = Arc::new(Repos::new(config.data.join("wikis"))); let hub_key = keys::ensure_identity(&config.keys_dir(), &config.site.name, config.base_url())?; let cookie_key = auth::CookieKey::load_or_create(&config.keys_dir())?; let assistant_key = assistants::SecretKey::load_or_create(&config.keys_dir())?; let secret_key = assistants::SecretKey::load_or_create_named(&config.keys_dir(), "secrets.key")?; let backend = protocol::HubBackend::new( repos.clone(), wikis.clone(), db.clone(), config.auth.push_tokens.clone(), tokio::runtime::Handle::current(), ) .with_acceptor(hub_key.identity.clone(), keys::load_secret(&config.keys_dir())?.1) .with_scan_budget(push_protection::Budget { bytes: config.limits.scan_limit.0, file: config.limits.scan_file_limit.0, time: std::time::Duration::from_millis(config.limits.scan_time_ms), }); let slot = backend.app_slot(); let server = oxy_server::Server::new(backend); let mail = mail::load(&db, &config.email).await; // The web role appends to the log; a worker alone only reads it (#272). let (tlog_dir, origin) = (config.data.join("tlog"), tlog::origin_of(config.base_url())); let tlog = Arc::new(if roles.web { tlog::Log::open(&tlog_dir, &origin)? } else { tlog::Log::open_reader(&tlog_dir, &origin)? }); tlog.configure(&config.tlog)?; let app = Arc::new(App { mail: std::sync::RwLock::new(mail), config, db, repos, wikis, server, hub_key, cookie_key, assistant_key, secret_key, oidc: tokio::sync::OnceCell::new(), assistant_gate: std::sync::Mutex::new(assistants::Gate::default()), throttle: std::sync::Mutex::new(throttle::Throttle::default()), bus: bus::Bus::default(), rooms: Arc::new(live::Rooms::default()), swarm: Default::default(), started: std::time::Instant::now(), tlog, roles, leader: leader::Leader::default(), }); *slot.lock().unwrap() = Some(Arc::downgrade(&app)); bus::start(&app).await?; storage::load(&app).await?; coverage::load(&app).await?; junit::load(&app).await?; reports::load(&app).await?; start_background(&app); setup::ensure(&app).await?; Ok(app) } /// Start this process's background tasks (docs/roles.md, #272). The /// worker role's are singletons across every process sharing the database, /// each run under its leader lock; the transparency log's checkpointer goes /// with the web role, which appends to the log, and is a singleton too. fn start_background(app: &Shared) { if app.roles.web { // The checkpointer signs the log this process appends to (#238). leader::run(app, "tlog.checkpointer", tlog::checkpointer); } if !app.roles.worker { return; } leader::run(app, "hooks.worker", hooks::worker); leader::run(app, "flow.reaper", flow::reaper); leader::run(app, "flow.scheduler", flow::scheduler); leader::run(app, "flow.placement", flow::placement); leader::run(app, "maintenance.nightly", maintenance::nightly); leader::run(app, "projects.archiver", maintenance::archiver); leader::run(app, "backups.scheduler", backups::scheduler); // The reverse pin index for repositories pushed before it existed (#148): // once, by whichever worker leads it first (idempotent if run again). leader::run(app, "diffusion.backfill", |app| async move { diffusion::backfill(&app).await }); // Mend the pin index and the bumps it drives (oxygenhub#210). leader::run(app, "diffusion.reconciler", diffusion::reconciler); // Each repository's history scanned for secrets (#277), a budget at a // time, at low priority. leader::run(app, "secrets.history_scan", history_scan::worker); // The durable work queue (#276): every worker takes its due items. app.leader.note("queue.worker"); tokio::spawn(queue::worker(app.clone())); // What each image is for, read or learned once (oxygenhub#173): safe // in every worker, so it takes no lock. app.leader.note("images.learn_archs"); tokio::spawn({ let app = app.clone(); async move { match images::learn_archs(&app).await { Ok(n) if n > 0 => tracing::info!(images = n, "read the architecture of kept images"), Ok(_) => {} Err(e) => tracing::warn!(error = %e, "learning image architectures"), } } }); } /// The router this process serves: everything for the web role; for a /// worker alone, `/health` and `/metrics`, so it can be watched. pub fn router(app: Shared) -> axum::Router { if !app.roles.web { return worker_router(app); } full_router(app) } /// What a worker-only process answers: its health and its metrics. fn worker_router(app: Shared) -> axum::Router { use axum::routing::get; let hsts = headers::hsts(app.config.base_url(), app.config.server.hsts_max_age); axum::Router::new() .route("/health", get(web::home::health)) .route("/metrics", get(metrics::metrics)) .with_state(app) .layer(axum::middleware::from_fn_with_state(hsts, headers::security_headers)) } /// The full router: pages, API, protocol and static assets. fn full_router(app: Shared) -> axum::Router { let https = app_https(&app); let hsts = headers::hsts(app.config.base_url(), app.config.server.hsts_max_age); web::router(app.clone()) .layer(axum::middleware::from_fn_with_state(app, agent_requests)) .layer(axum::middleware::from_fn(private_pages)) .layer(axum::middleware::from_fn(slow_requests)) .layer(axum::middleware::from_fn_with_state(https, https_only)) .layer(axum::middleware::from_fn_with_state(hsts, headers::security_headers)) } /// The site's https address, when it has one: what [`https_only`] redirects to. fn app_https(app: &Shared) -> Option { let base = app.config.base_url(); base.starts_with("https://").then(|| base.trim_end_matches('/').to_string()) } /// The scheme a proxy in front of the hub says the visitor used /// (`X-Forwarded-Proto`, else Cloudflare's `CF-Visitor`); `None` for a /// request made to the hub directly (runners and tools on its host). fn forwarded_scheme(headers: &axum::http::HeaderMap) -> Option { if let Some(p) = headers.get("x-forwarded-proto").and_then(|v| v.to_str().ok()) { return p.split(',').next().map(|s| s.trim().to_ascii_lowercase()); } let visitor = headers.get("cf-visitor").and_then(|v| v.to_str().ok())?; serde_json::from_str::(visitor).ok()?["scheme"].as_str().map(str::to_ascii_lowercase) } /// HTTPS only (oxygenhub#218): on a site whose address is https, a request /// a proxy reports as plain http is sent to the https address (GET and /// HEAD) or refused (anything else, so a form or a password is never taken /// over http). Requests made to the hub directly carry no such report and /// are untouched. async fn https_only( axum::extract::State(https): axum::extract::State>, req: axum::extract::Request, next: axum::middleware::Next, ) -> axum::response::Response { use axum::response::IntoResponse; let Some(base) = https else { return next.run(req).await }; if forwarded_scheme(req.headers()).as_deref() != Some("http") { return next.run(req).await; } let to = format!("{base}{}", req.uri().path_and_query().map(|p| p.as_str()).unwrap_or("/")); if matches!(*req.method(), axum::http::Method::GET | axum::http::Method::HEAD) { return axum::response::Redirect::permanent(&to).into_response(); } (axum::http::StatusCode::BAD_REQUEST, format!("This site takes requests over HTTPS only: send it to {to}\n")).into_response() } /// A request made with a token its owner marked as an agent's runs with /// that token's name in [`notify::AGENT`], so what the agent does is told /// to its owner (oxygenhub#183). async fn agent_requests( axum::extract::State(app): axum::extract::State, req: axum::extract::Request, next: axum::middleware::Next, ) -> axum::response::Response { let token = req .headers() .get(axum::http::header::AUTHORIZATION) .and_then(|v| v.to_str().ok()) .and_then(|v| v.strip_prefix("Bearer ")) .map(|t| auth::token_hash(t.trim())); let agent = match token { Some(hash) => db::accounts::token_by_hash(&app.db, &hash).await.ok().flatten().filter(|t| t.agent).map(|t| t.name), None => None, }; notify::AGENT.scope(agent, next.run(req)).await } /// Requests slower than this are logged, with their path (#145). const SLOW: std::time::Duration = std::time::Duration::from_secs(2); /// Log a slow request's method, path and time, so a slow page can be /// found afterwards. Never the query, which may hold a token. async fn slow_requests(req: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response { let (method, path) = (req.method().clone(), req.uri().path().to_string()); let start = std::time::Instant::now(); let res = next.run(req).await; let took = start.elapsed(); metrics::observe(method.as_str(), &path, res.status().as_u16(), took.as_secs_f64()); if took >= SLOW { tracing::warn!("slow request: {method} {path} took {:.1} s ({})", took.as_secs_f64(), res.status().as_u16()); } res } /// A page for a signed-in viewer is theirs alone: it may list private /// repositories, and the dashboard's sky carries their activity (#157). An /// HTML response to a request with a session cookie, which has not said /// how to cache itself, says `private, no-store`, so no shared cache (a /// proxy or a CDN in front of the hub) ever serves it to anyone else. async fn private_pages(req: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response { use axum::http::header; let signed_in = req .headers() .get_all(header::COOKIE) .iter() .filter_map(|v| v.to_str().ok()) .any(|c| c.split(';').any(|kv| kv.trim_start().starts_with("oxyhub="))); let mut res = next.run(req).await; let html = res.headers().get(header::CONTENT_TYPE).and_then(|v| v.to_str().ok()).is_some_and(|t| t.starts_with("text/html")); if signed_in && html && !res.headers().contains_key(header::CACHE_CONTROL) { res.headers_mut().insert(header::CACHE_CONTROL, header::HeaderValue::from_static("private, no-store")); } res } /// Serve on the configured addresses until the process is told to stop. pub async fn serve(app: Shared) -> anyhow::Result<()> { let router = router(app.clone()); let http = app.config.bind_addr()?; let https = app.config.tls_bind_addr()?; if http.is_none() && https.is_none() && app.roles.web { anyhow::bail!("nothing to bind: set server.bind or server.tls_bind"); } tracing::info!("roles: {}", app.roles.names().join(", ")); // Installed first: from here a SIGTERM drains rather than kills. let signal = shutdown::on_signal()?; let mut tasks = tokio::task::JoinSet::new(); if let Some(addr) = http { let listener = match shutdown::systemd_listener() { Some(l) => { tracing::info!("http on the socket systemd holds ({})", l.local_addr().map(|a| a.to_string()).unwrap_or_default()); tokio::net::TcpListener::from_std(l)? } None => { tracing::info!("http on {addr}"); tokio::net::TcpListener::bind(addr).await? } }; let r = router.clone(); tasks.spawn(async move { axum::serve(listener, r.into_make_service_with_connect_info::()) .with_graceful_shutdown(shutdown::stopping()) .await .map_err(anyhow::Error::from) }); } if let Some(addr) = https { let _ = rustls::crypto::ring::default_provider().install_default(); let tls = axum_server::tls_rustls::RustlsConfig::from_pem_file(app.config.tls_cert(), app.config.tls_key()) .await .map_err(|e| anyhow::anyhow!("loading TLS files: {e}"))?; tracing::info!("https on {addr}"); let r = router.clone(); let handle = axum_server::Handle::new(); let draining = handle.clone(); tokio::spawn(async move { shutdown::stopping().await; draining.graceful_shutdown(Some(shutdown::DRAIN)); }); tasks.spawn(async move { axum_server::bind_rustls(addr, tls) .handle(handle) .serve(r.into_make_service_with_connect_info::()) .await .map_err(anyhow::Error::from) }); } tokio::select! { // A worker with no address to serve on runs until it is stopped. r = tasks.join_next(), if !tasks.is_empty() => { r.transpose()?.transpose()?; Ok(()) } _ = signal => { // The servers stop taking connections and finish what they // have; whatever is still running after DRAIN is cut off. let drained = tokio::time::timeout(shutdown::DRAIN, async { while tasks.join_next().await.is_some() {} }).await; if drained.is_err() { tracing::warn!("requests still running after {} s: stopping anyway", shutdown::DRAIN.as_secs()); } tracing::info!("stopped"); Ok(()) } } }