lib.rs (20026B)
1 /* 2 * This file is part of LibEuFin. 3 * Copyright (C) 2026 Taler Systems S.A. 4 5 * LibEuFin is free software; you can redistribute it and/or modify 6 * it under the terms of the GNU Affero General Public License as 7 * published by the Free Software Foundation; either version 3, or 8 * (at your option) any later version. 9 10 * LibEuFin is distributed in the hope that it will be useful, but 11 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY 12 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General 13 * Public License for more details. 14 15 * You should have received a copy of the GNU Affero General Public 16 * License along with LibEuFin; see the file COPYING. If not, see 17 * <http://www.gnu.org/licenses/> 18 */ 19 20 #![allow(clippy::too_many_arguments)] 21 22 use std::{fmt::Write, sync::Arc}; 23 24 use anyhow::{anyhow, bail}; 25 use compact_str::{CompactString, CompactStringExt, ToCompactString}; 26 use jiff::{SignedDuration, Timestamp, civil::Date}; 27 use libeufin_ebics::{ 28 cli::{EbicsArgs, EbicsLogs}, 29 db::update_task_status, 30 ebics::{ 31 EbicsClient, EbicsCtx, EbicsErrKind, EbicsError, EbicsErrorHelper as _, 32 administrative::{AccountInfo, HKD, OrderInfo}, 33 order::{Order, OrderDoc}, 34 rand_ebics_id, 35 }, 36 iso20022::pain001::{Pain001Msg, Pain001Tx, create_pain001}, 37 keys::{BankKeys, ClientKeys}, 38 }; 39 use sqlx::PgPool; 40 use taler_api::api::TalerRouter; 41 use taler_build::long_version; 42 use taler_common::{ 43 CommonArgs, 44 cli::ConfigCmd, 45 config::Config, 46 types::{amount::Amount, payto::TransferIbanPayto, utils::date_to_utc_ts}, 47 }; 48 use taler_test_utils::Router; 49 use tracing::{debug, error, info, warn}; 50 51 use crate::{ 52 api::NexusApi, 53 config::{NexusCfg, NexusEbicsConfig, parse_db_cfg}, 54 constants::SUBMIT_TASK_KEY, 55 db::{ 56 dbinit, 57 initiated::{ 58 batch_initiated, batch_sub_failure, batch_sub_success, initiate, initiated_submittable, 59 }, 60 pool, 61 }, 62 fetch::ebics_fetch, 63 list::ListCmd, 64 manual::ManualCmd, 65 model::PaymentBatch, 66 testing::TestingCmd, 67 }; 68 69 pub mod api; 70 #[cfg(test)] 71 pub mod bench; 72 pub mod config; 73 pub mod constants; 74 pub mod db; 75 pub mod fetch; 76 pub mod list; 77 pub mod manual; 78 pub mod model; 79 #[cfg(test)] 80 pub mod test; 81 pub mod testing; 82 83 #[derive(clap::Subcommand, Debug)] 84 pub enum Cmd { 85 /// Initialize libeufin-nexus database 86 Dbinit { 87 /// Reset database (DANGEROUS: All existing data is lost) 88 #[arg(short, long)] 89 reset: bool, 90 }, 91 /// Set up the EBICS subscriber 92 EbicsSetup { 93 #[command(flatten)] 94 ebics_logs: EbicsLogs, 95 96 /// Resubmits all the keys to the bank 97 #[arg(long)] 98 force_keys_resubmission: bool, 99 100 /// Accepts the bank keys without interactively asking the user 101 #[arg(long)] 102 auto_accept_keys: bool, 103 104 /// Generates the PDF with the client public keys to send to the bank 105 #[arg(long)] 106 generate_registration_pdf: bool, 107 }, 108 /// Submits pending initiated payments found in the database 109 EbicsSubmit { 110 #[clap(flatten)] 111 ebics: EbicsArgs, 112 }, 113 /// Downloads and parse EBICS files from the bank and register them into the database 114 EbicsFetch { 115 #[clap(flatten)] 116 ebics: EbicsArgs, 117 118 /// Only supported in --transient mode, this option lets specify the earliest timestamp of the downloaded documents 119 #[arg(long, value_name = "YYYY-MM-DD", requires = "transient")] 120 pinned_start: Option<Date>, 121 122 /// Only supported in --transient mode, do not consume fetched documents 123 #[arg(long, requires = "transient")] 124 peek: bool, 125 126 /// Only supported in --transient mode, run a checkpoint 127 #[arg(long, requires = "transient")] 128 checkpoint: bool, 129 130 documents: Vec<OrderDoc>, 131 }, 132 /// Run libeufin-nexus HTTP server 133 Serve { 134 /// 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 135 #[arg(long)] 136 check: bool, 137 }, 138 /// Initiate an outgoing payment 139 InitiatePayment { 140 /// The amount to transfer, payto 'amount' parameter takes the precedence 141 #[arg(long)] 142 amount: Option<Amount>, 143 144 /// The payment subject, payto 'message' parameter takes the precedence 145 #[arg(long)] 146 subject: Option<CompactString>, 147 148 /// The payment end-to-end UID 149 #[arg(long, alias("request-uid"))] 150 end_to_end_id: Option<CompactString>, 151 152 /// The credited account IBAN payto UR 153 payto: TransferIbanPayto, 154 }, 155 #[command(subcommand)] 156 Manual(ManualCmd), 157 #[command(subcommand)] 158 List(ListCmd), 159 #[command(subcommand)] 160 Config(ConfigCmd), 161 #[command(subcommand)] 162 Testing(Box<TestingCmd>), 163 } 164 165 #[derive(clap::Parser, Debug)] 166 #[command(long_version = long_version(), about, long_about = None)] 167 pub struct Args { 168 #[clap(flatten)] 169 pub common: CommonArgs, 170 171 #[command(subcommand)] 172 pub cmd: Cmd, 173 } 174 175 pub fn batch_pain001( 176 batch: &PaymentBatch, 177 cfg: &NexusEbicsConfig, 178 now: Timestamp, 179 instant: bool, 180 ) -> Result<String, EbicsErrKind> { 181 let msg = Pain001Msg { 182 msg_id: &batch.msg_id, 183 timestamp: &now, 184 debtor: &cfg.account, 185 sum: batch.sum, 186 txs: batch 187 .payments 188 .iter() 189 .map(|tx| { 190 // TODO handle missing name ? 191 Pain001Tx { 192 creditor: &tx.creditor, 193 amount: tx.amount, 194 subject: &tx.subject, 195 e2e_id: &tx.e2e_id, 196 } 197 }) 198 .collect(), 199 }; 200 create_pain001(&msg, &cfg.dialect, instant) 201 } 202 203 pub async fn ebics_submit( 204 ebics: &EbicsClient<'_>, 205 cfg: &NexusCfg, 206 client: &ClientKeys, 207 bank: &BankKeys, 208 db: &PgPool, 209 transient: bool, 210 ) -> anyhow::Result<()> { 211 let ebics_cfg = cfg.ebics()?; 212 let submit_cfg = cfg.submit()?; 213 214 let submit_batch = async |order: &Order, 215 batch: &PaymentBatch, 216 instant: bool| 217 -> Result<CompactString, EbicsError> { 218 let ctx = EbicsCtx::new(order); 219 let xml = batch_pain001(batch, ebics_cfg, Timestamp::now(), instant).ctx(&ctx)?; 220 ebics.upload(client, bank, order, xml.as_bytes()).await 221 }; 222 223 let submit_all = async || -> anyhow::Result<()> { 224 let standard = cfg.ebics()?.dialect.standard(); 225 226 // Find a supported debit order 227 let mut instant_order = standard.instant_direct_debit(); 228 let debit_order = standard.direct_debit(); 229 230 // Create batch if necessary 231 batch_initiated( 232 db, 233 &Timestamp::now(), 234 &rand_ebics_id(), 235 submit_cfg.require_ack, 236 ) 237 .await?; 238 239 // Send submittable batches 240 for batch in initiated_submittable(db, &cfg.currency).await? { 241 debug!(target: "submit", "Submitting batch {}", batch.msg_id); 242 let res = async { 243 if let Some(instant) = instant_order.clone() { 244 match submit_batch(&instant, &batch, true).await { 245 Ok(id) => return Ok(id), 246 Err(e) => if let EbicsErrKind::Code { .. } = e.kind { 247 // No longer try to submit using the instant method for now 248 debug!(target: "submit", "Failed to submit using instant credit order {e}"); 249 instant_order = None; 250 } else { 251 return Err(e) 252 }, 253 } 254 } 255 submit_batch(&debit_order, &batch, false).await 256 }.await; 257 match res { 258 Ok(order_id) => { 259 batch_sub_success(db, batch.id, &Timestamp::now(), &order_id).await?; 260 let txs = batch 261 .payments 262 .iter() 263 .map(|it| &it.e2e_id) 264 .collect::<Vec<_>>() 265 .join_compact(","); 266 if instant_order.is_some() { 267 info!(target: "submit", "Instant batch {} submitted as order {order_id}: {txs}", batch.msg_id); 268 } else { 269 info!(target: "submit", "Batch {} submitted as order {order_id}: {txs}", batch.msg_id); 270 } 271 } 272 Err(e) => { 273 batch_sub_failure(db, batch.id, &Timestamp::now(), &e.to_string()).await?; 274 error!(target: "submit", "Batch {} submission failure: {e}", batch.msg_id); 275 return Err(e.into()); 276 } 277 } 278 } 279 280 Ok(()) 281 }; 282 if transient { 283 debug!(target: "submit", "Transient mode: submitting what found and returning"); 284 submit_all().await 285 } else { 286 debug!(target: "submit", "Running with a frequency of {}", submit_cfg.frequency_raw); 287 loop { 288 let now = Timestamp::now(); 289 let success = match submit_all().await { 290 Ok(_) => true, 291 Err(e) => { 292 error!(target: "submit", "{e}"); 293 false 294 } 295 }; 296 if let Err(e) = update_task_status(db, SUBMIT_TASK_KEY, &now, success).await { 297 warn!(target: "submit", "{e}"); 298 } 299 tokio::time::sleep( 300 Timestamp::now() 301 .duration_until(now + submit_cfg.frequency) 302 .max(SignedDuration::ZERO) 303 .unsigned_abs(), 304 ) 305 .await; 306 } 307 } 308 } 309 310 pub async fn ebics_setup( 311 ebics: &EbicsClient<'_>, 312 cfg: &NexusCfg, 313 db: &PgPool, 314 force_keys_resubmission: bool, 315 generate_registration_pdf: bool, 316 auto_accept_keys: bool, 317 ) -> anyhow::Result<()> { 318 let (client, bank) = libeufin_ebics::setup::ebics_setup( 319 ebics, 320 &cfg.ebics_setup()?, 321 force_keys_resubmission, 322 generate_registration_pdf, 323 auto_accept_keys, 324 ) 325 .await?; 326 327 // Check account information 328 info!(target: "setup", "Doing administrative request HKD"); 329 let HKD { partner, users } = ebics.hkd(db, &client, &bank, false).await?; 330 let host = cfg.host()?; 331 let user = users.iter().find(|it| it.id == host.user_id); 332 // Debug logging 333 let fmt = std::fmt::from_fn(|f| { 334 if partner.name.is_some() || !partner.accounts.is_empty() { 335 f.write_str("Partner Info: ")?; 336 if let Some(name) = &partner.name { 337 write!(f, "'{name}'")?; 338 } 339 for AccountInfo { 340 currency, 341 iban, 342 bic, 343 } in &partner.accounts 344 { 345 write!(f, "{currency}-{iban}{bic}")?; 346 } 347 f.write_char('\n')?; 348 } 349 writeln!(f, "Supported orders:")?; 350 for OrderInfo { order, description } in &partner.orders { 351 writeln!(f, "- {order}: {description}")?; 352 } 353 if let Some(user) = user { 354 writeln!(f, "Authorized orders:")?; 355 for order in &user.permissions { 356 writeln!(f, "- {order}")?; 357 } 358 } 359 Ok(()) 360 }); 361 debug!(target: "setup", "{fmt}"); 362 363 // Check partner info match config 364 let ebics = cfg.ebics()?; 365 if let Some(name) = &partner.name 366 && name != ebics.account.name 367 { 368 warn!(target: "setup", "Expected NAME '{}' from config got '{name}' from bank", ebics.account.name); 369 } 370 if let Some(account) = partner 371 .accounts 372 .iter() 373 .find(|it| it.iban == ebics.account.iban) 374 { 375 if account.currency != cfg.currency { 376 error!(target:"setup", "Expected CURRENCY '{}' from config got '{}' from bank", cfg.currency, account.currency); 377 } 378 if let Some(bic) = ebics.account.bic 379 && bic != account.bic 380 { 381 error!(target:"setup", "Expected BIC '{bic}' from config got '{}' from bank", account.bic); 382 } 383 } else if !partner.accounts.is_empty() { 384 let ibans = partner.accounts.iter().map(|it| it.iban).join_compact(" "); 385 error!(target: "setup", "Expected IBAN {} from config got {ibans} from bank", ebics.account.iban); 386 } 387 388 let std = ebics.dialect.standard(); 389 let instant_debit_order = std.instant_direct_debit(); 390 let debit_order = std.direct_debit(); 391 let required_orders = std.required(); 392 let partner_orders: Vec<_> = partner.orders.iter().map(|it| &it.order).collect(); 393 394 // Check partner support for direct debit orders 395 if let Some(o) = &instant_debit_order 396 && !partner_orders.contains(&o) 397 { 398 warn!(target: "setup", "Unsupported instant debit order: {o}"); 399 } 400 if !partner_orders.contains(&&debit_order) { 401 warn!(target: "setup", "Unsupported debit order: {debit_order}"); 402 } 403 404 // Check partner support required orders 405 let unsupported = required_orders 406 .iter() 407 .filter(|o| !partner_orders.contains(o)) 408 .map(|o| o.to_compact_string()) 409 .join_compact(" "); 410 if !unsupported.is_empty() { 411 warn!(target: "setup", "Unsupported orders: {unsupported}") 412 } 413 414 if let Some(user) = user { 415 // Check user is authorized for direct debit orders 416 if let Some(o) = &instant_debit_order 417 && partner_orders.contains(&o) 418 && !user.permissions.contains(o) 419 { 420 warn!(target: "setup", "Unauthorized instant debit order: {o}"); 421 } 422 if partner_orders.contains(&&debit_order) && !user.permissions.contains(&debit_order) { 423 warn!(target: "setup", "Unauthorized debit order: {debit_order}"); 424 } 425 426 // Check user is authorized for required orders 427 let unsupported = required_orders 428 .iter() 429 .filter(|o| partner_orders.contains(o) && !user.permissions.contains(o)) 430 .map(|o| o.to_compact_string()) 431 .join_compact(" "); 432 if !unsupported.is_empty() { 433 warn!(target: "setup", "Unauthorized orders: {unsupported}") 434 } 435 436 // Check user is authorized for required orders 437 info!(target: "setup", "Subscriber status: {}", user.status.description()) 438 } 439 440 println!("setup ready"); 441 Ok(()) 442 } 443 444 pub async fn run(cfg: &Config, cmd: Cmd) -> anyhow::Result<()> { 445 match cmd { 446 Cmd::Dbinit { reset } => { 447 let cfg = parse_db_cfg(cfg)?; 448 dbinit(&cfg, reset).await?; 449 } 450 Cmd::EbicsSetup { 451 ebics_logs, 452 force_keys_resubmission, 453 auto_accept_keys, 454 generate_registration_pdf, 455 } => { 456 let cfg = NexusCfg::parse(cfg)?; 457 let pool = pool(&cfg.db_cfg).await?; 458 let ebics = EbicsClient::new(cfg.host()?.ebics(), ebics_logs)?; 459 ebics_setup( 460 &ebics, 461 &cfg, 462 &pool, 463 force_keys_resubmission, 464 generate_registration_pdf, 465 auto_accept_keys, 466 ) 467 .await?; 468 } 469 Cmd::EbicsFetch { 470 pinned_start, 471 peek, 472 checkpoint, 473 ebics: EbicsArgs { logs, transient }, 474 documents, 475 } => { 476 let cfg = NexusCfg::parse(cfg)?; 477 let pool = pool(&cfg.db_cfg).await?; 478 let ebics = EbicsClient::new(cfg.host()?.ebics(), logs)?; 479 let (client, bank) = cfg.expect_full_keys()?; 480 ebics_fetch( 481 &ebics, 482 &cfg, 483 &client, 484 &bank, 485 &pool, 486 &documents, 487 &pinned_start.map(|it| date_to_utc_ts(&it)), 488 peek, 489 transient, 490 transient && checkpoint, 491 ) 492 .await? 493 } 494 Cmd::EbicsSubmit { 495 ebics: EbicsArgs { logs, transient }, 496 } => { 497 let cfg = NexusCfg::parse(cfg)?; 498 let pool = pool(&cfg.db_cfg).await?; 499 let ebics = EbicsClient::new(cfg.host()?.ebics(), logs)?; 500 let (client, bank) = cfg.expect_full_keys()?; 501 ebics_submit(&ebics, &cfg, &client, &bank, &pool, transient).await? 502 } 503 Cmd::InitiatePayment { 504 amount, 505 subject, 506 end_to_end_id, 507 payto, 508 } => { 509 let cfg = NexusCfg::parse(cfg)?; 510 let pool = pool(&cfg.db_cfg).await?; 511 512 let subject = payto 513 .subject 514 .as_ref() 515 .or(subject.as_ref()) 516 .ok_or(anyhow!("Missing subject"))?; 517 let amount = payto 518 .amount 519 .as_ref() 520 .or(amount.as_ref()) 521 .ok_or(anyhow!("Missing amount"))?; 522 523 if cfg.currency != amount.currency { 524 bail!( 525 "Wrong currency: expected {} got {}", 526 cfg.currency, 527 amount.currency 528 ); 529 } 530 initiate( 531 &pool, 532 amount, 533 subject, 534 &payto.clone().into(), 535 &Timestamp::now(), 536 &end_to_end_id 537 .as_ref() 538 .cloned() 539 .unwrap_or_else(rand_ebics_id), 540 ) 541 .await?; 542 } 543 Cmd::Serve { check } => { 544 let cfg = NexusCfg::parse(cfg)?; 545 if check { 546 let mut start_server = false; 547 for (name, api) in [ 548 ("Wire Gateway API", cfg.wire_cfg), 549 ("Revenue API", cfg.revenue_cfg), 550 ("Observability API", cfg.observability_cfg), 551 ] { 552 if api.is_some() { 553 start_server = true; 554 info!("{name} is enabled: starting the server") 555 } 556 } 557 558 if !start_server { 559 info!("All APIs are disabled: not starting the server"); 560 std::process::exit(1); 561 } 562 } else { 563 let pool = pool(&cfg.db_cfg).await?; 564 let ebics_cfg = cfg.ebics()?; 565 let api = Arc::new( 566 NexusApi::start( 567 pool, 568 cfg.currency, 569 ebics_cfg.account.clone(), 570 ebics_cfg.qr_iban, 571 ) 572 .await, 573 ); 574 let mut server = Router::new(); 575 if let Some(it) = cfg.wire_cfg { 576 server = server 577 .wire_gateway(api.clone(), it.auth.method()) 578 .prepared_transfer(api.clone()); 579 } 580 if let Some(it) = cfg.revenue_cfg { 581 server = server.revenue(api.clone(), it.auth.method()) 582 } 583 if let Some(it) = cfg.observability_cfg { 584 server = server.observability(api.clone(), it.auth.method()) 585 } 586 server.serve(&cfg.serve_cfg, None).await?; 587 } 588 } 589 Cmd::Manual(cmd) => { 590 let cfg = NexusCfg::parse(cfg)?; 591 let pool = pool(&cfg.db_cfg).await?; 592 cmd.run(&pool, &cfg).await?; 593 } 594 Cmd::List(cmd) => { 595 let cfg = NexusCfg::parse(cfg)?; 596 let pool = pool(&cfg.db_cfg).await?; 597 cmd.run(&pool, &cfg.currency).await?; 598 } 599 Cmd::Config(cmd) => cmd.run(cfg)?, 600 Cmd::Testing(cmd) => cmd.run(cfg).await?, 601 } 602 Ok(()) 603 }