api.rs (8824B)
1 /* 2 This file is part of TALER 3 Copyright (C) 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 use std::{ 18 sync::{ 19 Arc, 20 atomic::{AtomicU32, Ordering}, 21 }, 22 time::Instant, 23 }; 24 25 use axum::{ 26 extract::{Request, State}, 27 middleware::{self, Next}, 28 response::Response, 29 }; 30 use compact_str::CompactString; 31 use rand::RngExt as _; 32 use revenue::Revenue; 33 use taler_common::{ 34 error_code::ErrorCode, 35 log::LOG_TASK_ID, 36 types::amount::{Amount, Currency}, 37 }; 38 use tokio::signal; 39 use tower_http::cors::{Any, CorsLayer}; 40 use tracing::{Level, debug, info}; 41 use wire::WireGateway; 42 43 use crate::{ 44 Listener, Serve, 45 api::prepared::PreparedTransfer, 46 auth::{AuthMethod, AuthMiddlewareState}, 47 error::{ApiResult, LoggedError, failure, failure_code}, 48 }; 49 50 pub mod prepared; 51 pub mod revenue; 52 pub mod wire; 53 54 pub use axum::Router; 55 56 pub trait Validation { 57 fn check(&self, currency: &Currency) -> ApiResult<()>; 58 } 59 60 fn check_currency(currency: &Currency, amount: &Amount) -> ApiResult<()> { 61 if &amount.currency != currency { 62 Err(failure( 63 ErrorCode::GENERIC_CURRENCY_MISMATCH, 64 format!( 65 "wrong currency expected {} got {}", 66 currency, amount.currency 67 ), 68 )) 69 } else { 70 Ok(()) 71 } 72 } 73 74 pub trait TalerApi: Send + Sync + 'static { 75 fn currency(&self) -> Currency; 76 fn implementation(&self) -> &'static str; 77 } 78 79 pub trait RouterUtils { 80 fn auth(self, auth: AuthMethod, realm: &str) -> Self; 81 } 82 83 impl<S: Send + Clone + Sync + 'static> RouterUtils for Router<S> { 84 fn auth(self, auth: AuthMethod, realm: &str) -> Self { 85 self.route_layer(middleware::from_fn_with_state( 86 Arc::new(AuthMiddlewareState::new(auth, realm)), 87 crate::auth::auth_middleware, 88 )) 89 } 90 } 91 92 pub trait TalerRouter { 93 fn wire_gateway<T: WireGateway>(self, api: Arc<T>, auth: AuthMethod) -> Self; 94 fn prepared_transfer<T: PreparedTransfer>(self, api: Arc<T>) -> Self; 95 fn revenue<T: Revenue>(self, api: Arc<T>, auth: AuthMethod) -> Self; 96 fn finalize(self) -> Self; 97 fn serve( 98 self, 99 serve: &Serve, 100 lifetime: Option<u32>, 101 ) -> impl std::future::Future<Output = std::io::Result<()>> + Send; 102 } 103 104 impl TalerRouter for Router { 105 fn wire_gateway<T: WireGateway>(self, api: Arc<T>, auth: AuthMethod) -> Self { 106 self.nest("/taler-wire-gateway", wire::router(api, auth)) 107 } 108 109 fn prepared_transfer<T: PreparedTransfer>(self, api: Arc<T>) -> Self { 110 self.nest("/taler-prepared-transfer", prepared::router(api)) 111 } 112 113 fn revenue<T: Revenue>(self, api: Arc<T>, auth: AuthMethod) -> Self { 114 self.nest("/taler-revenue", revenue::router(api, auth)) 115 } 116 117 fn finalize(self) -> Router { 118 self.method_not_allowed_fallback(async || failure_code(ErrorCode::GENERIC_METHOD_INVALID)) 119 .fallback(async || failure_code(ErrorCode::GENERIC_ENDPOINT_UNKNOWN)) 120 .layer( 121 CorsLayer::new() 122 .allow_origin(Any) 123 .allow_methods(Any) 124 .allow_headers(Any), 125 ) 126 .layer(middleware::from_fn(logger_middleware)) 127 } 128 129 async fn serve(mut self, serve: &Serve, lifetime: Option<u32>) -> std::io::Result<()> { 130 let listener = serve.resolve()?; 131 132 let notify = Arc::new(tokio::sync::Notify::new()); 133 if let Some(lifetime) = lifetime { 134 self = self.layer(middleware::from_fn_with_state( 135 Arc::new(LifetimeMiddlewareState { 136 notify: notify.clone(), 137 lifetime: AtomicU32::new(lifetime), 138 }), 139 lifetime_middleware, 140 )) 141 } 142 let router = self.finalize(); 143 let signal = shutdown_signal(notify); 144 match listener { 145 Listener::Tcp(tcp_listener) => { 146 axum::serve(tcp_listener, router) 147 .with_graceful_shutdown(signal) 148 .await?; 149 } 150 Listener::Unix(unix_listener) => { 151 axum::serve(unix_listener, router) 152 .with_graceful_shutdown(signal) 153 .await?; 154 } 155 } 156 157 info!(target: "api", "Server stopped"); 158 Ok(()) 159 } 160 } 161 162 struct LifetimeMiddlewareState { 163 lifetime: AtomicU32, 164 notify: Arc<tokio::sync::Notify>, 165 } 166 167 async fn lifetime_middleware( 168 State(state): State<Arc<LifetimeMiddlewareState>>, 169 request: Request, 170 next: Next, 171 ) -> Response { 172 let mut current = state.lifetime.load(Ordering::Relaxed); 173 while current != 0 { 174 match state.lifetime.compare_exchange_weak( 175 current, 176 current - 1, 177 Ordering::Relaxed, 178 Ordering::Relaxed, 179 ) { 180 Ok(_) => break, 181 Err(new) => current = new, 182 } 183 } 184 if current == 0 { 185 state.notify.notify_one(); 186 } 187 next.run(request).await 188 } 189 190 /** Wait for manual shutdown or system signal shutdown */ 191 async fn shutdown_signal(manual_shutdown: Arc<tokio::sync::Notify>) { 192 let ctrl_c = async { 193 signal::ctrl_c() 194 .await 195 .expect("failed to install Ctrl+C handler"); 196 }; 197 198 #[cfg(unix)] 199 let terminate = async { 200 signal::unix::signal(signal::unix::SignalKind::terminate()) 201 .expect("failed to install signal handler") 202 .recv() 203 .await; 204 }; 205 206 #[cfg(not(unix))] 207 let terminate = std::future::pending::<()>(); 208 209 let manual = async { manual_shutdown.notified().await }; 210 211 tokio::select! { 212 _ = ctrl_c => {}, 213 _ = terminate => {}, 214 _ = manual => {} 215 } 216 } 217 218 #[macro_export] 219 macro_rules! dyn_event { 220 ($lvl:ident, $($arg:tt)+) => { 221 match $lvl { 222 ::tracing::Level::TRACE => ::tracing::trace!($($arg)+), 223 ::tracing::Level::DEBUG => ::tracing::debug!($($arg)+), 224 ::tracing::Level::INFO => ::tracing::info!($($arg)+), 225 ::tracing::Level::WARN => ::tracing::warn!($($arg)+), 226 ::tracing::Level::ERROR => ::tracing::error!($($arg)+), 227 } 228 }; 229 } 230 231 /** Taler API logger */ 232 async fn logger_middleware(request: Request, next: Next) -> Response { 233 let now = Instant::now(); 234 let request_id: compact_str::CompactString = { 235 let charset = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; 236 let mut rng = rand::rng(); 237 238 let mut ansi = [0u8; 10]; 239 for c in ansi.iter_mut() { 240 let idx = rng.random_range(0..charset.len()); 241 *c = charset[idx]; 242 } 243 unsafe { CompactString::from_utf8_unchecked(ansi) } 244 }; 245 let method = request.method().clone(); 246 let path_and_query = request.uri().path_and_query().cloned(); 247 let path_and_query = path_and_query 248 .as_ref() 249 .map(|it| it.as_str()) 250 .unwrap_or_default(); 251 LOG_TASK_ID 252 .scope(request_id, async { 253 debug!(target: "api", "{method} {path_and_query}"); 254 let response = next.run(request).await; 255 let elapsed = now.elapsed(); 256 let status = response.status(); 257 let level = match status.as_u16() { 258 400..500 => Level::WARN, 259 500..600 => Level::ERROR, 260 _ => Level::INFO, 261 }; 262 263 if let Some(log) = response.extensions().get::<LoggedError>() { 264 let LoggedError { code, info } = log; 265 dyn_event!(level, target: "api", 266 "{} {method} {path_and_query} {}ms: {code}{}", 267 response.status(), 268 elapsed.as_millis(), 269 std::fmt::from_fn(|f|{ 270 if let Some(info) = info { 271 write!(f, " {info}")?; 272 } 273 Ok(()) 274 }) 275 ); 276 } else { 277 dyn_event!(level, target: "api", 278 "{} {method} {path_and_query} {}ms", 279 response.status(), 280 elapsed.as_millis() 281 ); 282 } 283 response 284 }) 285 .await 286 }