Krolyxon
Commit 4030c43

Branches

Tags

Refs →

ferreniumsrcservermod.rs

372 lines · 13.0 KB · Rust

Raw Blame History Download
1//! HTTP server: router, middleware, listeners, signals and shutdown.
2
3pub mod assets;
4pub mod clone;
5pub mod dispatch;
6pub mod headers;
7
8use std::future::{Future, IntoFuture};
9use std::sync::Arc;
10use std::time::Duration;
11
12use anyhow::Context;
13use axum::Router;
14use axum::extract::{Request, State};
15use axum::http::{HeaderValue, StatusCode, Uri, header};
16use axum::middleware::{self, Next};
17use axum::response::{IntoResponse, Redirect, Response};
18use axum::routing::{any, get};
19use sd_notify::NotifyState;
20use tokio::signal::unix::{SignalKind, signal};
21use tokio::sync::watch;
22use tower::{Layer, ServiceBuilder};
23use tower_http::CompressionLevel;
24use tower_http::catch_panic::CatchPanicLayer;
25use tower_http::compression::predicate::{NotForContentType, Predicate};
26use tower_http::compression::{CompressionLayer, DefaultPredicate};
27use tower_http::trace::{DefaultOnResponse, TraceLayer};
28use tracing::Level;
29
30use crate::config::Listen;
31use crate::state::AppState;
32use crate::{handlers, index};
33
34/// How long in-flight requests may run after SIGTERM before we exit anyway.
35const DRAIN_DEADLINE: Duration = Duration::from_secs(10);
36
37/// Build the full application (routes + middleware) for the current config.
38pub 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.
136async 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
150async 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).
156fn 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.
167async 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.
190async 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
219async 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.
227pub 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
269async fn serve<L>(
270 listener: L,
271 app: Router,
272 listen: &Listen,
273 mut shutdown_rx: watch::Receiver<bool>,
274) -> anyhow::Result<()>
275where
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.
306fn 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.
329fn 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
350fn 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
368fn notify(states: &[NotifyState]) {
369 if let Err(e) = sd_notify::notify(states) {
370 tracing::debug!("sd_notify failed: {e}");
371 }
372}