libeufin

lib.rs (21547B)

---

 /*
 * This file is part of LibEuFin.
 * Copyright (C) 2026 Taler Systems S.A.
 
 * LibEuFin is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation; either version 3, or
 * (at your option) any later version.
 
 * LibEuFin is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
 * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Affero General
 * Public License for more details.
 
 * You should have received a copy of the GNU Affero General Public
 * License along with LibEuFin; see the file COPYING.  If not, see
 * <http://www.gnu.org/licenses/>
 */
 
 #![allow(clippy::too_many_arguments)]
 
 use std::{
     fmt::Write as _,
     io::{Cursor, Read as _},
     sync::Arc,
 };
 
 use anyhow::{anyhow, bail};
 use axum::{body::Bytes, http::HeaderValue};
 use compact_str::CompactStringExt;
 use http_client::Client;
 use jiff::{SignedDuration, Timestamp, Zoned, civil::Date, tz::TimeZone};
 use libeufin_ebics::{
     cli::{EbicsArgs, EbicsLogs},
     db::{get_task_status, update_task_status},
     ebics::{
         EbicsClient, EbicsErrKind,
         administrative::{AccountInfo, HKD, OrderInfo},
         ebics_code::EbicsReturnCode,
         order::{Order, OrderDoc},
     },
     iso20022::{HacAction, hac::parse_hac},
     keys::{BankKeys, ClientKeys},
     ws::listen_for_notification,
 };
 use sqlx::PgPool;
 use taler_api::api::TalerRouter;
 use taler_build::long_version;
 use taler_common::{CommonArgs, cli::ConfigCmd, config::Config, types::utils::date_to_utc_ts};
 use tokio::{time::timeout, try_join};
 use tracing::{debug, error, info, trace, warn};
 
 use crate::{
     api::{EbisyncState, sync_api},
     azure::AzureBlobStorage,
     config::{Destination, EbisyncCfg, Source, parse_db_cfg},
     constants::{CHECKPOINT_KEY, FETCH_TASK_KEY},
     db::{dbinit, pool},
 };
 
 pub mod api;
 pub mod azure;
 pub mod config;
 pub mod constants;
 pub mod db;
 
 /// Testing helper commands
 #[derive(clap::Subcommand, Debug)]
 pub enum TestingCmd {
     // Create the container
     CreateContainer,
 }
 
 #[derive(clap::Subcommand, Debug)]
 pub enum Cmd {
     /// Initialize libeufin-ebisync database
     Dbinit {
         /// Reset database (DANGEROUS: All existing data is lost)
         #[arg(short, long)]
         reset: bool,
     },
     /// Set up the EBICS subscriber
     Setup {
         #[command(flatten)]
         ebics_logs: EbicsLogs,
 
         /// Resubmits all the keys to the bank
         #[arg(long)]
         force_keys_resubmission: bool,
 
         /// Accepts the bank keys without interactively asking the user
         #[arg(long)]
         auto_accept_keys: bool,
 
         /// Generates the PDF with the client public keys to send to the bank
         #[arg(long)]
         generate_registration_pdf: bool,
     },
     /// Downloads EBICS files from the bank and store them in the configured destination
     Fetch {
         #[clap(flatten)]
         ebics: EbicsArgs,
 
         /// Only supported in --transient mode, this option lets specify the earliest timestamp of the downloaded documents
         #[arg(long, value_name = "YYYY-MM-DD", requires = "transient")]
         pinned_start: Option<Date>,
 
         /// Only supported in --transient mode, do not consume fetched documents
         #[arg(long, requires = "transient")]
         peek: bool,
 
         /// Only supported in --transient mode, run a checkpoint
         #[arg(long, requires = "transient")]
         checkpoint: bool,
         /// Check whether a destination is configured. Exit with 0 if at destination is configured, otherwise 1
         #[arg(long)]
         check: bool,
     },
     /// Run libeufin-ebisync HTTP server
     Serve {
         /// Check whether an API is in use (if it's useful to start the HTTP server). Exit with 0 if at least one API is enabled, otherwise 1
         #[arg(long)]
         check: bool,
     },
     #[command(subcommand)]
     Config(ConfigCmd),
     #[command(subcommand)]
     Testing(TestingCmd),
 }
 
 #[derive(clap::Parser, Debug)]
 #[command(long_version = long_version(), about, long_about = None)]
 pub struct Args {
     #[clap(flatten)]
     pub common: CommonArgs,
 
     #[command(subcommand)]
     pub cmd: Cmd,
 }
 
 pub async fn ebics_setup(
     ebics: &EbicsClient<'_>,
     cfg: &EbisyncCfg<'_>,
     db: &PgPool,
     force_keys_resubmission: bool,
     generate_registration_pdf: bool,
     auto_accept_keys: bool,
 ) -> anyhow::Result<()> {
     let (client, bank) = libeufin_ebics::setup::ebics_setup(
         ebics,
         &cfg.ebics_setup()?,
         force_keys_resubmission,
         generate_registration_pdf,
         auto_accept_keys,
     )
     .await?;
 
     // Check account information
     info!(target: "setup", "Doing administrative request HKD");
     let HKD { partner, .. } = ebics.hkd(db, &client, &bank, false).await?;
     // Debug logging
     let fmt = std::fmt::from_fn(|f| {
         if partner.name.is_some() || !partner.accounts.is_empty() {
             f.write_str("Partner Info: ")?;
             if let Some(name) = &partner.name {
                 write!(f, "'{name}'")?;
             }
             for AccountInfo {
                 currency,
                 iban,
                 bic,
             } in &partner.accounts
             {
                 write!(f, "{currency}-{iban}{bic}")?;
             }
             f.write_char('\n')?;
         }
         writeln!(f, "Supported orders:")?;
         for OrderInfo { order, description } in &partner.orders {
             writeln!(f, "- {order}: {description}")?;
         }
         Ok(())
     });
     debug!(target: "setup", "{fmt}");
     info!(target: "setup", "EBICS ready");
 
     info!(target: "setup", "Check fetch destination setup");
     if let Some(dest) =
         DestinationClient::prepare(&cfg.fetch()?.destination, &http_client::client())?
     {
         match dest {
             DestinationClient::AzureBlobStorage { client, container } => {
                 client.container_metadata(container).await?;
             }
         }
     } else {
         warn!(target: "setup", "No destination configured");
     }
     info!(target: "setup", "Fetch destination ready");
 
     println!("setup ready");
     Ok(())
 }
 
 pub enum DestinationClient<'a> {
     AzureBlobStorage {
         client: AzureBlobStorage<'a>,
         container: &'a str,
     },
 }
 
 impl<'a> DestinationClient<'a> {
     pub fn prepare(dest: &'a Destination, client: &'a Client) -> anyhow::Result<Option<Self>> {
         Ok(match dest {
             Destination::None => None,
             Destination::AzureBlobStorage {
                 api,
                 name: account,
                 key,
                 container,
             } => Some(DestinationClient::AzureBlobStorage {
                 client: AzureBlobStorage::new(api.as_str(), account, key, client)?,
                 container,
             }),
         })
     }
 
     pub async fn upload(&self, name: &str, xml: Bytes) -> anyhow::Result<()> {
         match self {
             DestinationClient::AzureBlobStorage { client, container } => {
                 client
                     .put_blob(
                         container,
                         name,
                         xml,
                         HeaderValue::from_static("application/xml"),
                     )
                     .await?;
             }
         }
         Ok(())
     }
 }
 
 pub async fn ebics_fetch(
     ebics: &EbicsClient<'_>,
     cfg: &EbisyncCfg<'_>,
     client: &ClientKeys,
     bank: &BankKeys,
     db: &PgPool,
     pinned_start: &Option<Timestamp>,
     peek: bool,
     check: bool,
     transient: bool,
     transient_checkpoint: bool,
 ) -> anyhow::Result<()> {
     let fetch_cfg = cfg.fetch()?;
     let (sender, mut receiver) = tokio::sync::mpsc::channel::<Vec<Order>>(10);
     let http = http_client::client();
     let dest = DestinationClient::prepare(&fetch_cfg.destination, &http)?;
 
     if check {
         if dest.is_none() {
             info!("No destination configured, not starting the fetcher");
             std::process::exit(1);
         }
     } else if let Some(dest) = dest {
         let upload = async |orders: &[Order], since: Option<Timestamp>| -> anyhow::Result<()> {
             for order in orders {
                 if let Err(e) = ebics
                     .download(
                         db,
                         client,
                         bank,
                         order,
                         &since.map(|it| (it, Timestamp::now())),
                         transient && peek,
                         async |content| {
                             if order.doc() == Some(OrderDoc::acknowledgement) {
                                 for ack in parse_hac(&content)? {
                                     debug!(target: "fetch", "{ack}");
                                     if let Some(order_id) = &ack.order_id {
                                         match ack.action {
                                             HacAction::ORDER_HAC_FINAL_POS => {
                                                 info!(target: "fetch", "Order {order_id} accepted at {}", ack.timestamp);
                                             }
                                             HacAction::ORDER_HAC_FINAL_NEG => {
                                                 info!(target: "fetch", "Order {order_id} refused at {}", ack.timestamp);
                                             }
                                             _ => {}
                                         }
                                     }
                                 }
                             } else {
                                 let mut z = zip::ZipArchive::new(Cursor::new(content))?;
                                 for i in 0..z.len() {
                                     let mut file = z.by_index(i)?;
                                     trace!(target: "fetch", "upload {}", file.name());
                                     let mut buf = Vec::new();
                                     file.read_to_end(&mut buf)?;
                                     dest.upload(file.name(), buf.into())
                                         .await
                                         .map_err(|e| EbicsErrKind::Custom(e.to_string().into()))?;
                                 }
                             }
                             Ok(())
                         },
                     )
                     .await
                 {
                     if let EbicsErrKind::Code {
                         bank: EbicsReturnCode::EBICS_NO_DOWNLOAD_DATA_AVAILABLE,
                         ..
                     } = e.kind
                     {
                         continue;
                     }
                     return Err(e.into());
                 }
             }
             Ok(())
         };
         let fetch = async {
             if transient {
                 info!(target: "fetch", "Transient mode: fetching once and returning");
             } else {
                 info!(target: "fetch", "Running with a frequency of {}", fetch_cfg.frequency_raw);
             }
 
             let mut last_fetch = Timestamp::UNIX_EPOCH;
             loop {
                 let now = Timestamp::now();
                 let checkpoint = get_task_status(db, CHECKPOINT_KEY)
                     .await?
                     .unwrap_or_default();
                 let next_fetch = last_fetch + fetch_cfg.frequency;
                 let next_checkpoint = {
                     if let Some(last_trial) = checkpoint.last_trial {
                         // We run today at checkpoint_time
                         let checkpoint_date = Zoned::new(now, TimeZone::UTC)
                             .with()
                             .time(fetch_cfg.checkpoint_time)
                             .build()
                             .unwrap();
                         // If we already ran today we ran tomorrow
                         if last_trial > checkpoint_date.timestamp() {
                             checkpoint_date.tomorrow().unwrap().timestamp()
                         } else {
                             checkpoint_date.timestamp()
                         }
                     } else {
                         // We never ran, we must checkpoint now
                         Timestamp::UNIX_EPOCH
                     }
                 };
 
                 let mut success = true;
                 if
                 // Run transient checkpoint at request
                 (transient && transient_checkpoint)
                     // Or run recurrent checkpoint
                     || (!transient && now > next_checkpoint)
                 {
                     info!(target: "fetch", "Running checkpoint");
 
                     let since = if let Some(pinned_start) = pinned_start
                         && transient
                         && checkpoint
                             .last_successfull
                             .map(|it| *pinned_start <= it)
                             .unwrap_or(true)
                     {
                         Some(*pinned_start)
                     } else {
                         checkpoint.last_successfull
                     };
                     let res = async {
                         // We fetch HKD to only fetch supported EBICS orders and get the document versions
                         let hkd = ebics.hkd(db, client, bank, false).await?;
                         let mut supported_orders = hkd
                             .partner
                             .orders
                             .into_iter()
                             .map(|it| it.order)
                             .collect::<Vec<_>>();
                         debug!(target: "fetch", "HKD: {}", supported_orders.iter().map(|it| it.to_string()).join_compact(", "));
                         supported_orders
                             .retain(|it| it.is_downloadable());
                         upload( &supported_orders, since).await
                     }
                     .await;
                     if let Err(e) = res {
                         success = false;
                         error!(target: "fetch", "{e}");
                     }
                     try_join!(
                         update_task_status(db, CHECKPOINT_KEY, &now, success),
                         update_task_status(db, FETCH_TASK_KEY, &now, success)
                     )?;
                     last_fetch = now;
                 } else if transient || now > next_fetch {
                     if !transient {
                         info!(target: "fetch", "Running at frequency");
                     }
                     let res = async {
                         // We fetch HAA to only fetch pending & supported EBICS orders and get the document versions
                         let mut haa = ebics.haa(db, client, bank, false).await?;
                         debug!(target: "fetch", "HAA: {}", haa.orders.iter().map(|it| it.to_string()).join_compact(", "));
                         haa.orders
                             .retain(|it| it.is_downloadable());
                         upload( &haa.orders, *pinned_start).await
                     }
                     .await;
                     if let Err(e) = res {
                         success = false;
                         error!(target: "fetch", "{e}");
                     }
                     update_task_status(db, FETCH_TASK_KEY, &now, success).await?;
                     last_fetch = now;
                 }
 
                 if transient {
                     if success {
                         return anyhow::Ok(());
                     } else {
                         return Err(anyhow!("fetch failed"));
                     }
                 }
 
                 let delay = now
                     .duration_until(next_fetch.min(next_checkpoint))
                     .max(SignedDuration::ZERO)
                     .unsigned_abs();
                 let tx = timeout(delay, receiver.recv()).await;
                 if let Ok(Some(notification)) = tx {
                     info!(target: "fetch", "Running at real-time notifications reception");
                     if let Err(e) = upload(&notification, None).await {
                         error!(target: "fetch", "{e}");
                     }
                 }
             }
         };
         if transient {
             fetch.await?;
         } else {
             tokio::try_join!(fetch, async {
                 listen_for_notification(ebics, db, client, bank, sender).await;
                 Ok(())
             })?;
         }
     }
 
     Ok(())
 }
 
 pub async fn run(cfg: &Config, cmd: Cmd) -> anyhow::Result<()> {
     match cmd {
         Cmd::Dbinit { reset } => {
             let cfg = parse_db_cfg(cfg)?;
             dbinit(&cfg, reset).await?;
         }
         Cmd::Setup {
             ebics_logs,
             force_keys_resubmission,
             auto_accept_keys,
             generate_registration_pdf,
         } => {
             let cfg = EbisyncCfg::parse(cfg)?;
             let pool = pool(&cfg.db_cfg).await?;
             let ebics = EbicsClient::new(cfg.host.ebics(), ebics_logs)?;
             ebics_setup(
                 &ebics,
                 &cfg,
                 &pool,
                 force_keys_resubmission,
                 generate_registration_pdf,
                 auto_accept_keys,
             )
             .await?;
         }
         Cmd::Fetch {
             pinned_start,
             peek,
             checkpoint,
             check,
             ebics: EbicsArgs { logs, transient },
         } => {
             let cfg = EbisyncCfg::parse(cfg)?;
             let pool = pool(&cfg.db_cfg).await?;
             let ebics = EbicsClient::new(cfg.host.ebics(), logs)?;
             let (client, bank) = cfg.expect_full_keys()?;
             ebics_fetch(
                 &ebics,
                 &cfg,
                 &client,
                 &bank,
                 &pool,
                 &pinned_start.map(|it| date_to_utc_ts(&it)),
                 peek,
                 check,
                 transient,
                 transient && checkpoint,
             )
             .await?
         }
         Cmd::Serve { check } => {
             let cfg = EbisyncCfg::parse(cfg)?;
             let auth = match &cfg.submit()?.source {
                 Source::None => None,
                 Source::SyncAPI(auth_cfg) => Some(auth_cfg.method()),
             };
             if check {
                 if auth.is_none() {
                     info!("No source api, not starting the server");
                     std::process::exit(1);
                 }
             } else if let Some(auth) = auth {
                 let (client, bank) = cfg.expect_full_keys()?;
                 let tmp = EbisyncState {
                     db: pool(&cfg.db_cfg).await?,
                     cfg: cfg.host.clone(),
                     client,
                     bank,
                 };
                 let cfg = cfg.serve()?;
 
                 sync_api(Arc::new(tmp), &cfg.spa_path, auth)
                     .serve(&cfg.serve, None)
                     .await?;
             }
         }
         Cmd::Config(cmd) => cmd.run(cfg)?,
         Cmd::Testing(cmd) => match cmd {
             TestingCmd::CreateContainer => {
                 let cfg = EbisyncCfg::parse(cfg)?;
                 let http = http_client::client();
                 let dest = DestinationClient::prepare(&cfg.fetch()?.destination, &http)?;
                 if let Some(DestinationClient::AzureBlobStorage { client, container }) = dest {
                     client.create_container(container).await?;
                     info!("Created container {container}");
                 } else {
                     bail!("Destination is not azure-blob-storage")
                 }
             }
         },
     }
     Ok(())
 }
 
 #[cfg(test)]
 mod ebics {
     use clap::Parser as _;
     use libeufin_ebics::test::{EbicsState, TestBank};
     use sqlx::PgPool;
     use taler_common::config::Config;
     use taler_macros::db_test;
 
     use crate::{Args, constants::CONFIG_SOURCE, run};
 
     pub async fn ebisync_cmd(cfg: &Config, cmd: &str) -> anyhow::Result<()> {
         let parts = shlex::split(cmd).unwrap();
         let args = std::iter::once("libeufin-ebisync").chain(parts.iter().map(|it| it.as_str()));
 
         let cmd = Args::try_parse_from(args).unwrap();
         run(cfg, cmd.cmd).await
     }
 
     async fn test_setup(db: PgPool) -> (TestBank, Config, PgPool) {
         let test = TestBank::new().await;
         let cfg = Config::from_mem_with_env(
             CONFIG_SOURCE,
             &format!(
                 "
                 [paths]
                 EBISYNC_HOME = {:?}
 
                 [ebisync]
                 HOST_BASE_URL = http://bank.example.com/
                 UNIXPATH = {}
                 HOST_ID = PFEBICS
                 USER_ID = PFC00563
                 PARTNER_ID = PFC00563
 
                 [ebisyncdb-postgres]
                 CONFIG = postgresql:///{}
                 ",
                 test.dir.path(),
                 test.sock_path,
                 db.connect_options().get_database().unwrap()
             ),
         )
         .unwrap();
         test.sequences(&[
             EbicsState::hev,
             EbicsState::ini,
             EbicsState::hia,
             EbicsState::hpb,
             EbicsState::hkd,
             EbicsState::receipt_ok,
         ]);
         ebisync_cmd(&cfg, "setup --auto-accept-keys").await.unwrap();
 
         (test, cfg, db)
     }
 
     #[db_test]
     async fn setup(db: PgPool) {
         test_setup(db).await;
     }
 }