taler-rust

GNU Taler code in Rust. Largely core banking integrations.
Log | Files | Refs | Submodules | README | LICENSE

api.rs (9133B)

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