Coverage / crates/hub/src/lib.rs 87.2%

    pub fn images(&self) -> images::Store {

        images::Store::new(&self.config.data)

    }

    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 async fn build(config: Config) -> anyhow::Result<Shared> {

    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;

    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)

}

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);

    leader::run(app, "diffusion.backfill", |app| async move { diffusion::backfill(&app).await });

    leader::run(app, "diffusion.reconciler", diffusion::reconciler);

    leader::run(app, "secrets.history_scan", history_scan::worker);

    app.leader.note("queue.worker");

    tokio::spawn(queue::worker(app.clone()));

    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(_) => {}

        }

}

pub fn router(app: Shared) -> axum::Router {

    if !app.roles.web {

        return worker_router(app);

    }

    full_router(app)

}

fn worker_router(app: Shared) -> axum::Router {

    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))

}

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))

}

fn app_https(app: &Shared) -> Option<String> {

    let base = app.config.base_url();

    base.starts_with("https://").then(|| base.trim_end_matches('/').to_string())

}

fn forwarded_scheme(headers: &axum::http::HeaderMap) -> Option<String> {

    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::<serde_json::Value>(visitor).ok()?["scheme"].as_str().map(str::to_ascii_lowercase)

}

async fn https_only(

    axum::extract::State(https): axum::extract::State<Option<String>>,

    req: axum::extract::Request,

    next: axum::middleware::Next,

) -> axum::response::Response {

    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()

}

async fn agent_requests(

    axum::extract::State(app): axum::extract::State<Shared>,

    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| t.trim().to_string());

    let agent = match token {

        Some(t) if t.starts_with(keyless::GRANT_TOKEN_PREFIX) => db::accounts::agent_grant_by_token(&app.db, &auth::token_hash(&t))

            .await

            .ok()

            .flatten()

            .filter(|g| g.live(db::now_ms()))

            .map(|g| g.agent),

        Some(t) => db::accounts::token_by_hash(&app.db, &auth::token_hash(&t)).await.ok().flatten().filter(|t| t.agent).map(|t| t.name),

        None => None,

    notify::AGENT.scope(agent, next.run(req)).await

}

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

}

async fn private_pages(req: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response {

    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

}

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 {

    }

    tracing::info!("roles: {}", app.roles.names().join(", "));

    let signal = shutdown::on_signal()?;

    let mut tasks = tokio::task::JoinSet::new();

    if let Some(addr) = http {

        let listener = match shutdown::systemd_listener() {

                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::<std::net::SocketAddr>())

                .with_graceful_shutdown(shutdown::stopping())

                .await

                .map_err(anyhow::Error::from)

        });

    if let Some(addr) = https {

    }

    tokio::select! {

        r = tasks.join_next(), if !tasks.is_empty() => { r.transpose()?.transpose()?; Ok(()) }

        _ = signal => {

}