Krolyxon
Commit e758199

Branches

Tags

Refs →

ferreniumsrcservermod.rs

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