serve.rs (9484B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2024, 2025, 2026 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 17 //! Listening socket selection: TCP, a UNIX domain socket, or a socket 18 //! inherited from systemd. 19 20 use std::{ 21 fs::Permissions, 22 io::ErrorKind, 23 net::{IpAddr, SocketAddr}, 24 os::unix::fs::PermissionsExt as _, 25 time::Instant, 26 }; 27 28 use axum::{ 29 Router, 30 extract::Request, 31 middleware::{self, Next}, 32 response::Response, 33 }; 34 use compact_str::CompactString; 35 use listenfd::ListenFd; 36 use rand::Rng as _; 37 use tokio::{ 38 net::{TcpListener, UnixListener}, 39 signal, 40 }; 41 use tracing::{Level, debug, info}; 42 43 use crate::{ 44 config::{Section, ValueErr}, 45 log::LOG_TASK_ID, 46 map_config, 47 }; 48 49 #[derive(Debug, Clone)] 50 pub enum Serve { 51 Tcp(SocketAddr), 52 Unix { 53 path: String, 54 permission: Permissions, 55 }, 56 Systemd, 57 } 58 59 impl Serve { 60 /// Read the `SERVE` option and whatever else the chosen mode needs 61 pub fn parse(s: &Section) -> Result<Self, ValueErr> { 62 map_config!(s, "serve", "SERVE", 63 "tcp" => { 64 let port = s.number("PORT").require()?; 65 let ip: IpAddr = s.parse("IP addr", "BIND_TO").require()?; 66 Serve::Tcp(SocketAddr::new(ip, port)) 67 }, 68 "unix" => { 69 let path = s.path("UNIXPATH").require()?; 70 let permission = s.unix_mode("UNIXPATH_MODE").require()?; 71 Serve::Unix { path, permission } 72 }, 73 "systemd" => { Serve::Systemd } 74 ) 75 .require() 76 } 77 78 /// Resolve listener from a config and environment 79 fn resolve(&self) -> Result<Listener, std::io::Error> { 80 match self { 81 Serve::Tcp(socket_addr) => { 82 info!(target: "api", "Server listening on {socket_addr}"); 83 let listener = std::net::TcpListener::bind(socket_addr)?; 84 listener.set_nonblocking(true)?; 85 Ok(Listener::Tcp(TcpListener::from_std(listener)?)) 86 } 87 Serve::Unix { path, permission } => { 88 info!(target: "api", 89 "Server listening on unix domain socket {path} {:o}", 90 permission.mode() 91 ); 92 if let Err(e) = std::fs::remove_file(path) { 93 let kind = e.kind(); 94 if kind != ErrorKind::NotFound { 95 return Err(e); 96 } 97 } 98 let listener = std::os::unix::net::UnixListener::bind(path)?; 99 std::fs::set_permissions(path, permission.clone())?; 100 listener.set_nonblocking(true)?; 101 Ok(Listener::Unix(UnixListener::from_std(listener)?)) 102 } 103 Serve::Systemd => { 104 let mut listenfd = ListenFd::from_env(); 105 if let Ok(Some(unix)) = listenfd.take_unix_listener(0) { 106 info!(target: "api", 107 "Server listening on activated unix socket {:?}", 108 unix.local_addr()? 109 ); 110 unix.set_nonblocking(true)?; 111 Ok(Listener::Unix(UnixListener::from_std(unix)?)) 112 } else if let Ok(Some(tcp)) = listenfd.take_tcp_listener(0) { 113 info!(target: "api", 114 "Server listening on activated TCP socket {:?}", 115 tcp.local_addr()? 116 ); 117 tcp.set_nonblocking(true)?; 118 Ok(Listener::Tcp(TcpListener::from_std(tcp)?)) 119 } else { 120 Err(std::io::Error::other("Missing systemd activated socket")) 121 } 122 } 123 } 124 } 125 } 126 127 enum Listener { 128 Tcp(TcpListener), 129 Unix(UnixListener), 130 } 131 132 /// Bind the socket described by `serve`, add the request logger and serve 133 /// `router` until SIGINT or SIGTERM, letting in-flight requests finish. 134 pub async fn serve(router: Router, serve: &Serve) -> std::io::Result<()> { 135 let listener = serve.resolve()?; 136 let router = router.layer(middleware::from_fn(logger_middleware)); 137 match listener { 138 Listener::Tcp(tcp_listener) => { 139 axum::serve(tcp_listener, router) 140 .with_graceful_shutdown(shutdown_signal()) 141 .await?; 142 } 143 Listener::Unix(unix_listener) => { 144 axum::serve(unix_listener, router) 145 .with_graceful_shutdown(shutdown_signal()) 146 .await?; 147 } 148 } 149 150 info!(target: "api", "Server stopped"); 151 Ok(()) 152 } 153 154 /** Wait for a system signal shutdown */ 155 async fn shutdown_signal() { 156 let ctrl_c = async { 157 signal::ctrl_c() 158 .await 159 .expect("failed to install Ctrl+C handler"); 160 }; 161 162 let terminate = async { 163 signal::unix::signal(signal::unix::SignalKind::terminate()) 164 .expect("failed to install signal handler") 165 .recv() 166 .await; 167 }; 168 169 tokio::select! { 170 _ = ctrl_c => {}, 171 _ = terminate => {}, 172 } 173 } 174 175 #[macro_export] 176 macro_rules! dyn_event { 177 ($lvl:ident, $($arg:tt)+) => { 178 match $lvl { 179 ::tracing::Level::TRACE => ::tracing::trace!($($arg)+), 180 ::tracing::Level::DEBUG => ::tracing::debug!($($arg)+), 181 ::tracing::Level::INFO => ::tracing::info!($($arg)+), 182 ::tracing::Level::WARN => ::tracing::warn!($($arg)+), 183 ::tracing::Level::ERROR => ::tracing::error!($($arg)+), 184 } 185 }; 186 } 187 188 /** Taler API logger */ 189 async fn logger_middleware(request: Request, next: Next) -> Response { 190 let now = Instant::now(); 191 let request_id: CompactString = { 192 let charset = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; 193 let mut rng = rand::thread_rng(); 194 195 let mut ansi = [0u8; 10]; 196 for c in ansi.iter_mut() { 197 let idx = rng.gen_range(0..charset.len()); 198 *c = charset[idx]; 199 } 200 unsafe { CompactString::from_utf8_unchecked(ansi) } 201 }; 202 let method = request.method().clone(); 203 let path_and_query = request.uri().path_and_query().cloned(); 204 let path_and_query = path_and_query 205 .as_ref() 206 .map(|it| it.as_str()) 207 .unwrap_or_default(); 208 LOG_TASK_ID 209 .scope(request_id, async { 210 debug!(target: "api", "{method} {path_and_query}"); 211 let response = next.run(request).await; 212 let elapsed = now.elapsed(); 213 let status = response.status(); 214 let level = match status.as_u16() { 215 400..500 => Level::WARN, 216 500..600 => Level::ERROR, 217 _ => Level::INFO, 218 }; 219 dyn_event!(level, target: "api", 220 "{} {method} {path_and_query} {}ms", 221 response.status(), 222 elapsed.as_millis() 223 ); 224 response 225 }) 226 .await 227 } 228 229 #[cfg(test)] 230 mod test { 231 use std::os::unix::fs::PermissionsExt as _; 232 233 use crate::config::Config; 234 235 use super::Serve; 236 237 fn parse(content: &str) -> Result<Serve, String> { 238 let cfg = Config::from_mem(content).unwrap(); 239 Serve::parse(&cfg.section("test")).map_err(|e| e.to_string()) 240 } 241 242 #[test] 243 fn tcp() { 244 let Serve::Tcp(addr) = parse("[test]\nSERVE=tcp\nPORT=8080\nBIND_TO=127.0.0.1").unwrap() 245 else { 246 panic!("expected a TCP socket") 247 }; 248 assert_eq!("127.0.0.1:8080", addr.to_string()); 249 } 250 251 #[test] 252 fn unix() { 253 let Serve::Unix { path, permission } = 254 parse("[test]\nSERVE=unix\nUNIXPATH=/run/kych/kych.sock\nUNIXPATH_MODE=660").unwrap() 255 else { 256 panic!("expected a UNIX domain socket") 257 }; 258 assert_eq!("/run/kych/kych.sock", path); 259 assert_eq!(0o660, permission.mode()); 260 } 261 262 #[test] 263 fn systemd() { 264 assert!(matches!( 265 parse("[test]\nSERVE=systemd").unwrap(), 266 Serve::Systemd 267 )); 268 } 269 270 #[test] 271 fn errors() { 272 assert_eq!( 273 "Missing serve option SERVE in section [test]", 274 parse("[test]\nPORT=8080").unwrap_err() 275 ); 276 assert_eq!( 277 "Invalid serve option SERVE in section [test]: \ 278 expected 'tcp', 'unix' or 'systemd' got 'http'", 279 parse("[test]\nSERVE=http").unwrap_err() 280 ); 281 // An option the chosen mode needs but does not have is reported on its 282 // own terms, without a "SERVE is invalid" prefix in front of it. 283 assert_eq!( 284 "Missing path option UNIXPATH in section [test]", 285 parse("[test]\nSERVE=unix").unwrap_err() 286 ); 287 assert_eq!( 288 "Missing IP addr option BIND_TO in section [test]", 289 parse("[test]\nSERVE=tcp\nPORT=8080").unwrap_err() 290 ); 291 } 292 }