Krolyxon
Commit 18ca21c

Branches

Tags

Refs →

ferreniumsrcservermod.rs

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