| 1 | //! HTTP server: router, middleware, listeners, signals and shutdown. |
| 2 | |
| 3 | pub mod assets; |
| 4 | pub mod clone; |
| 5 | pub mod dispatch; |
| 6 | pub mod headers; |
| 7 | |
| 8 | use std::future::{Future, IntoFuture}; |
| 9 | use std::sync::Arc; |
| 10 | use std::time::Duration; |
| 11 | |
| 12 | use anyhow::Context; |
| 13 | use axum::Router; |
| 14 | use axum::extract::{Request, State}; |
| 15 | use axum::http::{HeaderValue, StatusCode, Uri, header}; |
| 16 | use axum::middleware::{self, Next}; |
| 17 | use axum::response::{IntoResponse, Redirect, Response}; |
| 18 | use axum::routing::{any, get}; |
| 19 | use sd_notify::NotifyState; |
| 20 | use tokio::signal::unix::{SignalKind, signal}; |
| 21 | use tokio::sync::watch; |
| 22 | use tower::{Layer, ServiceBuilder}; |
| 23 | use tower_http::CompressionLevel; |
| 24 | use tower_http::catch_panic::CatchPanicLayer; |
| 25 | use tower_http::compression::predicate::{NotForContentType, Predicate}; |
| 26 | use tower_http::compression::{CompressionLayer, DefaultPredicate}; |
| 27 | use tower_http::trace::{DefaultOnResponse, TraceLayer}; |
| 28 | use tracing::Level; |
| 29 | |
| 30 | use crate::config::Listen; |
| 31 | use crate::state::AppState; |
| 32 | use crate::{handlers, index}; |
| 33 | |
| 34 | /// How long in-flight requests may run after SIGTERM before we exit anyway. |
| 35 | const DRAIN_DEADLINE: Duration = Duration::from_secs(10); |
| 36 | |
| 37 | /// Build the full application (routes + middleware) for the current config. |
| 38 | pub fn app(state: Arc<AppState>) -> Router { |
| 39 | let config = state.config.load(); |
| 40 | |
| 41 | let routes = Router::new() |
| 42 | .route("/", get(handlers::index::index)) |
| 43 | .route("/-/health", get(health)) |
| 44 | .route("/-/site.css", get(assets::site_css)) |
| 45 | .route("/-/favicon", get(assets::favicon)) |
| 46 | .route("/-/logo", get(assets::logo)) |
| 47 | .route("/-/site/{*path}", get(assets::site_asset)) |
| 48 | .route("/-/static/{hash}/{*path}", get(assets::serve)) |
| 49 | .route("/robots.txt", get(robots)) |
| 50 | .route("/{*path}", any(dispatch::dispatch)) |
| 51 | .fallback(handlers::error::not_found) |
| 52 | .method_not_allowed_fallback(method_not_allowed) |
| 53 | .with_state(state.clone()); |
| 54 | |
| 55 | // The prefix must be stripped *before* routing, so wrap the whole router |
| 56 | // (Router::layer would run after route matching). |
| 57 | let routes = match config.server.base_path.as_str() { |
| 58 | "/" => routes, |
| 59 | base => Router::new().fallback_service( |
| 60 | middleware::from_fn_with_state( |
| 61 | Arc::<str>::from(base.trim_end_matches('/')), |
| 62 | strip_base_path, |
| 63 | ) |
| 64 | .layer(routes), |
| 65 | ), |
| 66 | }; |
| 67 | |
| 68 | let trusted_proxy = config.server.trusted_proxy; |
| 69 | // gzip only: brotli's window costs megabytes per response for a ~2% |
| 70 | // smaller page. |
| 71 | let compression = CompressionLayer::new() |
| 72 | .no_deflate() |
| 73 | .no_zstd() |
| 74 | .quality(CompressionLevel::Precise(4)); |
| 75 | // Always installed (so the body type is fixed); disabling turns off every |
| 76 | // encoding instead. |
| 77 | let compression = if config.server.compression { |
| 78 | compression |
| 79 | } else { |
| 80 | compression.no_gzip() |
| 81 | } |
| 82 | // Fonts, archives and packfiles are already compressed. |
| 83 | .compress_when( |
| 84 | DefaultPredicate::new() |
| 85 | .and(NotForContentType::const_new("font/")) |
| 86 | .and(NotForContentType::const_new("application/zip")) |
| 87 | .and(NotForContentType::const_new("application/gzip")) |
| 88 | .and(NotForContentType::const_new("application/x-git-")), |
| 89 | ); |
| 90 | |
| 91 | let [csp, nosniff, referrer, frame, coop] = headers::layers(); |
| 92 | let panic_state = state.clone(); |
| 93 | routes.layer( |
| 94 | ServiceBuilder::new() |
| 95 | .layer( |
| 96 | TraceLayer::new_for_http() |
| 97 | .make_span_with(move |req: &Request| { |
| 98 | let client = trusted_proxy |
| 99 | .then(|| forwarded_for(req)) |
| 100 | .flatten() |
| 101 | .unwrap_or("-"); |
| 102 | tracing::info_span!( |
| 103 | "request", |
| 104 | method = %req.method(), |
| 105 | uri = %req.uri(), |
| 106 | client = %client, |
| 107 | ) |
| 108 | }) |
| 109 | .on_request(()) |
| 110 | .on_response(DefaultOnResponse::new().level(Level::INFO)), |
| 111 | ) |
| 112 | // Outside the panic and timeout handlers, so their pages get |
| 113 | // the headers too. |
| 114 | .layer(csp) |
| 115 | .layer(nosniff) |
| 116 | .layer(referrer) |
| 117 | .layer(frame) |
| 118 | .layer(coop) |
| 119 | .layer(CatchPanicLayer::custom(move |_| { |
| 120 | tracing::error!("request handler panicked"); |
| 121 | handlers::error::page( |
| 122 | &panic_state, |
| 123 | StatusCode::INTERNAL_SERVER_ERROR, |
| 124 | "INTERNAL ERROR", |
| 125 | ) |
| 126 | })) |
| 127 | .layer(middleware::from_fn_with_state(state, shield)) |
| 128 | .layer(compression), |
| 129 | ) |
| 130 | } |
| 131 | |
| 132 | /// Refuse requests beyond `limits.max_requests` at once, and answer with a |
| 133 | /// 503 page when a response hasn't started within |
| 134 | /// `limits.request_timeout_secs`. Streamed bodies (archives, clones) are not |
| 135 | /// cut off once they've started. |
| 136 | async fn shield(State(state): State<Arc<AppState>>, req: Request, next: Next) -> Response { |
| 137 | let Ok(_permit) = state.request_permits.clone().try_acquire_owned() else { |
| 138 | tracing::warn!("too many requests in flight; refusing {}", req.uri()); |
| 139 | return handlers::error::busy(&state); |
| 140 | }; |
| 141 | let timeout = Duration::from_secs(state.config.load().limits.request_timeout_secs); |
| 142 | match tokio::time::timeout(timeout, next.run(req)).await { |
| 143 | Ok(res) => res, |
| 144 | Err(_) => { |
| 145 | handlers::error::page(&state, StatusCode::SERVICE_UNAVAILABLE, "REQUEST TIMED OUT") |
| 146 | } |
| 147 | } |
| 148 | } |
| 149 | |
| 150 | async fn method_not_allowed(State(state): State<Arc<AppState>>) -> Response { |
| 151 | handlers::error::method_not_allowed(&state) |
| 152 | } |
| 153 | |
| 154 | /// First address in `X-Forwarded-For` (the original client, as seen by the |
| 155 | /// trusted proxy). |
| 156 | fn forwarded_for(req: &Request) -> Option<&str> { |
| 157 | let value = req.headers().get("x-forwarded-for")?.to_str().ok()?; |
| 158 | value |
| 159 | .split(',') |
| 160 | .next() |
| 161 | .map(str::trim) |
| 162 | .filter(|s| !s.is_empty()) |
| 163 | } |
| 164 | |
| 165 | /// Serve under `server.base_path`: strip the prefix before routing, redirect |
| 166 | /// the bare prefix to its slash form, and 404 everything outside it. |
| 167 | async fn strip_base_path(State(base): State<Arc<str>>, mut req: Request, next: Next) -> Response { |
| 168 | let path = req.uri().path(); |
| 169 | if path == &*base { |
| 170 | return Redirect::permanent(&format!("{base}/")).into_response(); |
| 171 | } |
| 172 | let Some(rest) = path.strip_prefix(&*base).filter(|r| r.starts_with('/')) else { |
| 173 | return StatusCode::NOT_FOUND.into_response(); |
| 174 | }; |
| 175 | let new_uri = match req.uri().query() { |
| 176 | Some(q) => format!("{rest}?{q}"), |
| 177 | None => rest.to_string(), |
| 178 | }; |
| 179 | match new_uri.parse::<Uri>() { |
| 180 | Ok(uri) => { |
| 181 | *req.uri_mut() = uri; |
| 182 | next.run(req).await |
| 183 | } |
| 184 | Err(_) => StatusCode::BAD_REQUEST.into_response(), |
| 185 | } |
| 186 | } |
| 187 | |
| 188 | /// Keep crawlers off the expensive pages: blame, archives, patches, |
| 189 | /// searches, path histories, deep log pages, diff options and clones. |
| 190 | async fn robots(State(state): State<Arc<AppState>>) -> impl IntoResponse { |
| 191 | let base = state.config.load().server.base_path.clone(); |
| 192 | let mut text = String::from("User-agent: *\n"); |
| 193 | for rule in [ |
| 194 | "*/blame/", |
| 195 | "*/archive/", |
| 196 | "*.patch$", |
| 197 | "*/find/", |
| 198 | "*/log/*/*", |
| 199 | "*?q=", |
| 200 | "*&q=", |
| 201 | "*?after=", |
| 202 | "*&after=", |
| 203 | "*?context=", |
| 204 | "*?ws=", |
| 205 | "*/info/refs", |
| 206 | "*/git-upload-pack", |
| 207 | ] { |
| 208 | text.push_str(&format!("Disallow: {base}{rule}\n")); |
| 209 | } |
| 210 | ( |
| 211 | [( |
| 212 | header::CACHE_CONTROL, |
| 213 | HeaderValue::from_static("public, max-age=86400"), |
| 214 | )], |
| 215 | text, |
| 216 | ) |
| 217 | } |
| 218 | |
| 219 | async fn health() -> impl IntoResponse { |
| 220 | ( |
| 221 | [(header::CACHE_CONTROL, HeaderValue::from_static("no-store"))], |
| 222 | "ok\n", |
| 223 | ) |
| 224 | } |
| 225 | |
| 226 | /// Run the server until SIGTERM/SIGINT. SIGHUP reloads the config. |
| 227 | pub async fn run(state: Arc<AppState>) -> anyhow::Result<()> { |
| 228 | let listen = state.config.load().server.listen.clone(); |
| 229 | let (shutdown_tx, shutdown_rx) = watch::channel(false); |
| 230 | tokio::spawn(handle_signals(state.clone(), shutdown_tx)?); |
| 231 | |
| 232 | // The index is complete before we report READY. |
| 233 | let st = state.clone(); |
| 234 | let stats = tokio::task::spawn_blocking(move || { |
| 235 | let config = st.config.load_full(); |
| 236 | st.index.full_scan(&config) |
| 237 | }) |
| 238 | .await |
| 239 | .context("initial scan failed")?; |
| 240 | let mut seen = Default::default(); |
| 241 | index::watch::log_scan(&stats, &state.index.load(), &mut seen, true); |
| 242 | tokio::spawn(index::watch::run(state.clone(), seen)); |
| 243 | tokio::spawn(index::languages::warm(state.clone())); |
| 244 | // Parse the embedded grammars now rather than on the first blob view. |
| 245 | tokio::task::spawn_blocking(|| { |
| 246 | let start = std::time::Instant::now(); |
| 247 | crate::render::highlight::syntaxes(); |
| 248 | tracing::debug!("syntaxes loaded in {:?}", start.elapsed()); |
| 249 | }); |
| 250 | |
| 251 | let app = app(state.clone()); |
| 252 | |
| 253 | match &listen { |
| 254 | Listen::Tcp(addr) => { |
| 255 | let listener = tokio::net::TcpListener::bind(addr) |
| 256 | .await |
| 257 | .with_context(|| format!("cannot listen on {addr}"))?; |
| 258 | serve(listener, app, &listen, shutdown_rx).await |
| 259 | } |
| 260 | Listen::Unix(path) => { |
| 261 | let listener = bind_unix(path)?; |
| 262 | let result = serve(listener, app, &listen, shutdown_rx).await; |
| 263 | let _ = std::fs::remove_file(path); |
| 264 | result |
| 265 | } |
| 266 | } |
| 267 | } |
| 268 | |
| 269 | async fn serve<L>( |
| 270 | listener: L, |
| 271 | app: Router, |
| 272 | listen: &Listen, |
| 273 | mut shutdown_rx: watch::Receiver<bool>, |
| 274 | ) -> anyhow::Result<()> |
| 275 | where |
| 276 | L: axum::serve::Listener, |
| 277 | L::Addr: std::fmt::Debug, |
| 278 | { |
| 279 | tracing::info!("listening on {listen}"); |
| 280 | notify(&[NotifyState::Ready, NotifyState::Status("serving")]); |
| 281 | |
| 282 | let mut drain_rx = shutdown_rx.clone(); |
| 283 | let server = axum::serve(listener, app) |
| 284 | .with_graceful_shutdown(async move { |
| 285 | let _ = shutdown_rx.wait_for(|stop| *stop).await; |
| 286 | }) |
| 287 | .into_future(); |
| 288 | |
| 289 | tokio::select! { |
| 290 | result = server => result.context("server error")?, |
| 291 | _ = async { |
| 292 | let _ = drain_rx.wait_for(|stop| *stop).await; |
| 293 | tokio::time::sleep(DRAIN_DEADLINE).await; |
| 294 | } => { |
| 295 | tracing::warn!( |
| 296 | "connections still open after {}s; exiting anyway", |
| 297 | DRAIN_DEADLINE.as_secs() |
| 298 | ); |
| 299 | } |
| 300 | } |
| 301 | tracing::info!("stopped"); |
| 302 | Ok(()) |
| 303 | } |
| 304 | |
| 305 | /// Bind a unix socket, replacing a stale socket file from a previous run. |
| 306 | fn bind_unix(path: &std::path::Path) -> anyhow::Result<tokio::net::UnixListener> { |
| 307 | use std::os::unix::fs::{FileTypeExt, PermissionsExt}; |
| 308 | |
| 309 | if let Ok(meta) = std::fs::symlink_metadata(path) { |
| 310 | anyhow::ensure!( |
| 311 | meta.file_type().is_socket(), |
| 312 | "{} exists and is not a socket", |
| 313 | path.display() |
| 314 | ); |
| 315 | std::fs::remove_file(path) |
| 316 | .with_context(|| format!("cannot remove stale socket {}", path.display()))?; |
| 317 | } |
| 318 | let listener = tokio::net::UnixListener::bind(path) |
| 319 | .with_context(|| format!("cannot listen on unix:{}", path.display()))?; |
| 320 | // Same reachability as a loopback TCP port: any local user (i.e. the |
| 321 | // reverse proxy) may connect. All content served is public anyway. |
| 322 | std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666)) |
| 323 | .with_context(|| format!("cannot set permissions on {}", path.display()))?; |
| 324 | Ok(listener) |
| 325 | } |
| 326 | |
| 327 | /// Install signal handlers (synchronously, so a failure aborts startup) and |
| 328 | /// return the task that services them. |
| 329 | fn handle_signals( |
| 330 | state: Arc<AppState>, |
| 331 | shutdown_tx: watch::Sender<bool>, |
| 332 | ) -> anyhow::Result<impl Future<Output = ()> + Send + 'static> { |
| 333 | let mut term = signal(SignalKind::terminate()).context("cannot install SIGTERM handler")?; |
| 334 | let mut int = signal(SignalKind::interrupt()).context("cannot install SIGINT handler")?; |
| 335 | let mut hup = signal(SignalKind::hangup()).context("cannot install SIGHUP handler")?; |
| 336 | |
| 337 | Ok(async move { |
| 338 | loop { |
| 339 | tokio::select! { |
| 340 | _ = term.recv() => { tracing::info!("SIGTERM received; draining"); break; } |
| 341 | _ = int.recv() => { tracing::info!("SIGINT received; draining"); break; } |
| 342 | _ = hup.recv() => reload(&state), |
| 343 | } |
| 344 | } |
| 345 | notify(&[NotifyState::Stopping]); |
| 346 | let _ = shutdown_tx.send(true); |
| 347 | }) |
| 348 | } |
| 349 | |
| 350 | fn reload(state: &AppState) { |
| 351 | tracing::info!("SIGHUP received; reloading configuration"); |
| 352 | let mut msg = vec![NotifyState::Reloading]; |
| 353 | if let Ok(now) = NotifyState::monotonic_usec_now() { |
| 354 | msg.push(now); |
| 355 | } |
| 356 | notify(&msg); |
| 357 | match state.reload() { |
| 358 | Ok(config) => { |
| 359 | crate::logging::apply(&config); |
| 360 | tracing::info!("configuration reloaded"); |
| 361 | state.rescan.notify_one(); |
| 362 | } |
| 363 | Err(e) => tracing::error!("reload failed, keeping current configuration: {e}"), |
| 364 | } |
| 365 | notify(&[NotifyState::Ready]); |
| 366 | } |
| 367 | |
| 368 | fn notify(states: &[NotifyState]) { |
| 369 | if let Err(e) = sd_notify::notify(states) { |
| 370 | tracing::debug!("sd_notify failed: {e}"); |
| 371 | } |
| 372 | } |