api.rs (20999B)
1 /* 2 This file is part of TALER 3 Copyright (C) 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::{collections::BTreeMap, sync::Arc}; 18 19 use aws_lc_rs::signature::ParsedPublicKey; 20 use axum::{ 21 body::Bytes, 22 extract::State, 23 http::{HeaderMap, HeaderName, StatusCode}, 24 routing::post, 25 }; 26 use compact_str::{CompactString, format_compact}; 27 use http_client::headers::HeaderParser; 28 use jiff::Timestamp; 29 use prometheus_client::{ 30 encoding::EncodeLabelSet, 31 metrics::{family::Family, gauge::Gauge}, 32 registry::{Registry, Unit}, 33 }; 34 use serde::Deserialize; 35 use serde_json::value::RawValue; 36 use sqlx::PgPool; 37 use taler_api::{ 38 api::{ 39 TalerApi, TalerRouter as _, observability::Observability, prepared::PreparedTransfer, 40 revenue::Revenue, wire::WireGateway, 41 }, 42 config::ApiCfg, 43 error::{ApiResult, failure_code, not_implemented}, 44 subject::{IncomingKey, fmt_in_subject}, 45 }; 46 use taler_common::{ 47 api::{ 48 params::{History, Page}, 49 prepared::{ 50 RegistrationRequest, RegistrationResponse, SubjectFormat, TransferSubject, 51 Unregistration, 52 }, 53 revenue::RevenueIncomingHistory, 54 wire::{ 55 AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddMappedRequest, 56 IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse, 57 TransferState, TransferStatus, 58 }, 59 }, 60 db::IncomingType, 61 error_code::ErrorCode, 62 types::{amount::Currency, payto::PaytoImpl, time::TalerTimestamp}, 63 }; 64 use taler_test_utils::Router; 65 use tokio::sync::watch::Sender; 66 67 use crate::{ 68 config::WiseBalance, 69 db::{self, AddIncomingResult, TxIn}, 70 payto::FullWisePayto, 71 }; 72 73 #[allow(clippy::too_many_arguments)] 74 pub async fn start( 75 pool: PgPool, 76 name: &str, 77 balances: Vec<WiseBalance>, 78 key: ParsedPublicKey, 79 profile_id: u64, 80 wire_gateway: &Option<ApiCfg>, 81 revenue: &Option<ApiCfg>, 82 observability: &Option<ApiCfg>, 83 ) -> Router { 84 let mut main_router = Router::new() 85 .route("/wise-webhooks", post(receive_webhook)) 86 .with_state(Arc::new(WebhookState { 87 pool: pool.clone(), 88 key, 89 profile: profile_id, 90 balances: balances.iter().map(|b| (b.id, b.currency)).collect(), 91 })); 92 let apis: Vec<_> = balances 93 .into_iter() 94 .map(|b| { 95 let metrics = Metrics::default(); 96 WiseBalanceApi { 97 id: b.id, 98 currency: b.currency, 99 pool: pool.clone(), 100 payto: b.payto.full(name), 101 in_channel: Sender::new(0), 102 taler_in_channel: Sender::new(0), 103 registry: metrics.registry(), 104 metrics, 105 } 106 }) 107 .collect(); 108 let in_channels = Arc::new(BTreeMap::from_iter( 109 apis.iter().map(|b| (b.id, b.in_channel.clone())), 110 )); 111 let taler_in_channels = Arc::new(BTreeMap::from_iter( 112 apis.iter().map(|b| (b.id, b.taler_in_channel.clone())), 113 )); 114 tokio::spawn(db::notification_listener( 115 pool, 116 in_channels, 117 taler_in_channels, 118 )); 119 for api in apis { 120 let api = Arc::new(api); 121 let mut balance_router = Router::new(); 122 if let Some(cfg) = wire_gateway { 123 balance_router = balance_router 124 .wire_gateway(api.clone(), cfg.auth.method()) 125 .prepared_transfer(api.clone()); 126 } 127 if let Some(cfg) = revenue { 128 balance_router = balance_router.revenue(api.clone(), cfg.auth.method()); 129 } 130 if let Some(cfg) = observability { 131 balance_router = balance_router.observability(api.clone(), cfg.auth.method()); 132 } 133 main_router = main_router.nest(&format!("/balances/{}", api.id), balance_router); 134 } 135 main_router 136 } 137 138 pub struct WiseBalanceApi { 139 pub id: u32, 140 pub currency: Currency, 141 pub pool: sqlx::PgPool, 142 pub payto: FullWisePayto, 143 pub in_channel: Sender<i64>, 144 pub taler_in_channel: Sender<i64>, 145 metrics: Metrics, 146 registry: Registry, 147 } 148 149 #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)] 150 struct TaskLabel { 151 name: &'static str, 152 } 153 154 #[derive(Default)] 155 struct Metrics { 156 db_access: Gauge, 157 task_execution: Family<TaskLabel, Gauge>, 158 task_success: Family<TaskLabel, Gauge>, 159 } 160 161 impl Metrics { 162 pub fn registry(&self) -> Registry { 163 let mut registry = Registry::default(); 164 165 registry.register( 166 "db_access", 167 "Whether the last database metrics refresh succeeded", 168 self.db_access.clone(), 169 ); 170 registry.register_with_unit( 171 "task_execution_timestamp_seconds", 172 "Unix timestamp of the last task execution", 173 Unit::Seconds, 174 self.task_execution.clone(), 175 ); 176 registry.register_with_unit( 177 "task_success_timestamp_seconds", 178 "Unix timestamp of the last successful task execution", 179 Unit::Seconds, 180 self.task_success.clone(), 181 ); 182 registry 183 } 184 185 pub async fn sync(&self, db: &PgPool, balance_id: u32) { 186 let success = match db::load_sync_state(db, balance_id).await { 187 Ok(state) => { 188 let label = TaskLabel { name: "sync" }; 189 if let Some(time) = state.as_ref().and_then(|s| s.execution) { 190 self.task_execution 191 .get_or_create(&label) 192 .set(time.as_second()); 193 } else { 194 self.task_execution.remove(&label); 195 } 196 let success = state.and_then(|s| { 197 s.early_success 198 .max(s.sync_success) 199 .max(s.checkpoint_success) 200 }); 201 if let Some(time) = success { 202 self.task_success 203 .get_or_create(&label) 204 .set(time.as_second()); 205 } else { 206 self.task_success.remove(&label); 207 } 208 true 209 } 210 Err(_) => false, 211 }; 212 self.db_access.set(if success { 1 } else { 0 }); 213 } 214 } 215 216 impl TalerApi for WiseBalanceApi { 217 fn currency(&self) -> Currency { 218 self.currency 219 } 220 221 fn implementation(&self) -> &'static str { 222 "urn:net:taler:specs:taler-wise:taler-rust" 223 } 224 } 225 226 impl WiseBalanceApi { 227 async fn add_incoming( 228 &self, 229 tx: &TxIn, 230 subject: IncomingKey, 231 ) -> ApiResult<AddIncomingResponse> { 232 let now = Timestamp::now(); 233 match db::register_tx_in(&self.pool, tx, &Some(subject), &now).await? { 234 AddIncomingResult::Success { 235 row_id, valued_at, .. 236 } => Ok(AddIncomingResponse { 237 row_id, 238 timestamp: valued_at.into(), 239 }), 240 AddIncomingResult::ReservePubReuse => { 241 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 242 } 243 AddIncomingResult::UnknownMapping => { 244 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN)) 245 } 246 AddIncomingResult::MappingReuse => { 247 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED)) 248 } 249 } 250 } 251 } 252 253 impl WireGateway for WiseBalanceApi { 254 async fn transfer(&self, _: TransferRequest) -> ApiResult<TransferResponse> { 255 Err(not_implemented()) 256 } 257 258 async fn transfer_page(&self, _: Page, _: Option<TransferState>) -> ApiResult<TransferList> { 259 Ok(TransferList { 260 transfers: Vec::new(), 261 debit_account: self.payto.as_uri(), 262 }) 263 } 264 265 async fn transfer_by_id(&self, _: u64) -> ApiResult<Option<TransferStatus>> { 266 Ok(None) 267 } 268 269 async fn outgoing_history(&self, _: History) -> ApiResult<OutgoingHistory> { 270 Ok(OutgoingHistory { 271 outgoing_transactions: Vec::new(), 272 debit_account: self.payto.as_uri(), 273 }) 274 } 275 276 async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> { 277 Ok(IncomingHistory { 278 incoming_transactions: db::incoming_history( 279 &self.pool, 280 self.id, 281 &self.currency, 282 ¶ms, 283 || self.taler_in_channel.subscribe(), 284 ) 285 .await?, 286 credit_account: self.payto.as_uri(), 287 }) 288 } 289 290 async fn add_incoming_reserve( 291 &self, 292 req: AddIncomingRequest, 293 ) -> ApiResult<AddIncomingResponse> { 294 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 295 let subject = format_compact!("Admin incoming {}", req.reserve_pub); 296 self.add_incoming( 297 &TxIn { 298 balance_id: self.id, 299 wise_ref: None, 300 amount: req.amount, 301 subject, 302 name, 303 debtor: Some(account), 304 value_at: Timestamp::now(), 305 }, 306 IncomingKey::reserve(req.reserve_pub), 307 ) 308 .await 309 } 310 311 async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> { 312 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 313 let subject = format_compact!("Admin incoming KYC:{}", req.account_pub); 314 self.add_incoming( 315 &TxIn { 316 balance_id: self.id, 317 wise_ref: None, 318 amount: req.amount, 319 subject, 320 name, 321 debtor: Some(account), 322 value_at: Timestamp::now(), 323 }, 324 IncomingKey::kyc(req.account_pub), 325 ) 326 .await 327 } 328 329 async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> { 330 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 331 let subject = format_compact!("Admin incoming MAP:{}", req.authorization_pub); 332 self.add_incoming( 333 &TxIn { 334 balance_id: self.id, 335 wise_ref: None, 336 amount: req.amount, 337 subject, 338 name, 339 debtor: Some(account), 340 value_at: Timestamp::now(), 341 }, 342 IncomingKey::map(req.authorization_pub), 343 ) 344 .await 345 } 346 347 fn support_account_check(&self) -> bool { 348 false 349 } 350 } 351 352 impl Revenue for WiseBalanceApi { 353 async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> { 354 Ok(RevenueIncomingHistory { 355 incoming_transactions: db::revenue_history( 356 &self.pool, 357 self.id, 358 &self.currency, 359 ¶ms, 360 || self.in_channel.subscribe(), 361 ) 362 .await?, 363 credit_account: self.payto.as_uri(), 364 }) 365 } 366 } 367 368 impl PreparedTransfer for WiseBalanceApi { 369 fn supported_formats(&self) -> &[SubjectFormat] { 370 &[SubjectFormat::SIMPLE] 371 } 372 373 async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> { 374 let creditor = FullWisePayto::try_from(&req.credit_account)?; 375 if creditor != self.payto { 376 return Err(failure_code(ErrorCode::BANK_UNKNOWN_CREDITOR)); 377 } 378 match db::transfer_register(&self.pool, &req).await? { 379 db::RegistrationResult::Success => { 380 let simple = TransferSubject::Simple { 381 credit_amount: req.credit_amount, 382 subject: if req.authorization_pub == req.account_pub && !req.recurrent { 383 fmt_in_subject(req.r#type.into(), &req.account_pub).to_string() 384 } else { 385 fmt_in_subject(IncomingType::map, &req.authorization_pub).to_string() 386 }, 387 }; 388 ApiResult::Ok(RegistrationResponse { 389 subjects: vec![simple], 390 expiration: TalerTimestamp::Never, 391 }) 392 } 393 db::RegistrationResult::ReservePubReuse => { 394 ApiResult::Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 395 } 396 } 397 } 398 399 async fn unregistration(&self, req: Unregistration) -> ApiResult<bool> { 400 Ok(db::transfer_unregister(&self.pool, &req).await?) 401 } 402 } 403 404 impl Observability for WiseBalanceApi { 405 async fn metrics(&self) -> ApiResult<&Registry> { 406 self.metrics.sync(&self.pool, self.id).await; 407 Ok(&self.registry) 408 } 409 } 410 411 struct WebhookState { 412 pool: PgPool, 413 key: ParsedPublicKey, 414 profile: u64, 415 balances: BTreeMap<u32, Currency>, 416 } 417 418 #[derive(Deserialize)] 419 struct WebhookEvent<'a> { 420 event_type: CompactString, 421 schema_version: CompactString, 422 subscription_id: uuid::Uuid, 423 sent_at: Timestamp, 424 #[serde(borrow)] 425 data: &'a RawValue, 426 } 427 428 #[derive(Deserialize)] 429 struct BalanceEvent { 430 resource: Resource, 431 balance_id: u32, 432 currency: Currency, 433 occurred_at: Timestamp, 434 transfer_reference: Option<CompactString>, 435 transaction_type: CompactString, 436 } 437 438 #[derive(Deserialize)] 439 struct Resource { 440 profile_id: u64, 441 } 442 443 async fn receive_webhook( 444 State(state): State<Arc<WebhookState>>, 445 headers: HeaderMap, 446 body: Bytes, 447 ) -> Result<StatusCode, StatusCode> { 448 // Verify signature 449 let signature = headers 450 .b64_header(HeaderName::from_static("x-signature-sha256")) 451 .map_err(|_| StatusCode::UNAUTHORIZED)?; 452 state 453 .key 454 .verify_sig(&body, &signature) 455 .map_err(|_| StatusCode::UNAUTHORIZED)?; 456 457 // Parse webhook event 458 let event: WebhookEvent<'_> = 459 serde_json::from_slice(&body).map_err(|_| StatusCode::BAD_REQUEST)?; 460 tracing::debug!(target: "webhook", 461 event_type = %event.event_type, 462 schema_version = %event.schema_version, 463 subscription_id = %event.subscription_id, 464 sent_at = %event.sent_at, 465 "received Wise webhook" 466 ); 467 468 // We only handle balances update right now 469 if event.event_type != "balances#update" || event.schema_version != "4.0.0" { 470 return Ok(StatusCode::NO_CONTENT); 471 } 472 let balance: BalanceEvent = 473 serde_json::from_str(event.data.get()).map_err(|_| StatusCode::BAD_REQUEST)?; 474 if balance.transaction_type != "credit" 475 || balance.resource.profile_id != state.profile 476 || state.balances.get(&balance.balance_id) != Some(&balance.currency) 477 { 478 return Ok(StatusCode::NO_CONTENT); 479 } 480 sqlx::query("SELECT pg_notify('wise_balance_credit', $1)") 481 .bind(balance.balance_id.to_string()) 482 .execute(&state.pool) 483 .await 484 .map_err(|err| { 485 tracing::error!(target: "webhook", "cannot notify balance change: {err}"); 486 StatusCode::SERVICE_UNAVAILABLE 487 })?; 488 tracing::debug!(target: "webhook", balance_id = balance.balance_id, transfer_reference = ?balance.transfer_reference, occurred_at = %balance.occurred_at, "notified balance change"); 489 Ok(StatusCode::NO_CONTENT) 490 } 491 492 #[cfg(test)] 493 mod test { 494 495 use std::sync::LazyLock; 496 497 use compact_str::CompactString; 498 use jiff::Timestamp; 499 use sqlx::PgPool; 500 use taler_api::{ 501 api::TalerRouter as _, 502 config::{ApiCfg, AuthCfg}, 503 subject::IncomingKey, 504 }; 505 use taler_common::{ 506 api::{ 507 EddsaPublicKey, observability::Config, prepared::PreparedTransferConfig, 508 revenue::RevenueConfig, wire::WireConfig, 509 }, 510 types::{ 511 amount::{Currency, amount}, 512 iban::iban, 513 payto::{BankID, PaytoURI}, 514 }, 515 }; 516 use taler_macros::db_test; 517 use taler_test_utils::{ 518 Router, 519 routine::{admin_add_incoming_routine, in_history_routine, revenue_routine}, 520 server::TestServer, 521 tasks, 522 }; 523 524 use crate::{ 525 api, 526 config::{ApiEnv, WiseBalance}, 527 db::{TxIn, register_tx_in}, 528 payto::{FullWisePayto, WiseAccount}, 529 }; 530 531 static PAYTO: LazyLock<FullWisePayto> = LazyLock::new(|| { 532 "payto://ach/021000021/024030222?receiver-name=Exchange" 533 .parse() 534 .unwrap() 535 }); 536 static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri()); 537 538 async fn setup(pool: &PgPool) -> Router { 539 let balances = vec![ 540 WiseBalance { 541 id: 42, 542 currency: Currency::TEST, 543 payto: PAYTO.clone().into_inner().0, 544 }, 545 WiseBalance { 546 id: 34, 547 currency: Currency::KUDOS, 548 payto: PAYTO.clone().into_inner().0, 549 }, 550 ]; 551 api::start( 552 pool.clone(), 553 "Exchange", 554 balances, 555 ApiEnv::sandbox.public_key(), 556 7, 557 &Some(ApiCfg { 558 auth: AuthCfg::None, 559 }), 560 &Some(ApiCfg { 561 auth: AuthCfg::None, 562 }), 563 &Some(ApiCfg { 564 auth: AuthCfg::None, 565 }), 566 ) 567 .await 568 .finalize() 569 } 570 571 #[db_test] 572 async fn config(db: PgPool) { 573 let server = setup(&db).await; 574 for id in [42, 34] { 575 server 576 .get(format!("/balances/{id}/taler-wire-gateway/config")) 577 .await 578 .assert_ok_json::<WireConfig>(); 579 server 580 .get(format!("/balances/{id}/taler-prepared-transfer/config")) 581 .await 582 .assert_ok_json::<PreparedTransferConfig>(); 583 server 584 .get(format!("/balances/{id}/taler-revenue/config")) 585 .await 586 .assert_ok_json::<RevenueConfig>(); 587 server 588 .get(format!("/balances/{id}/taler-observability/config")) 589 .await 590 .assert_ok_json::<Config>(); 591 server 592 .get(format!("/balances/{id}/taler-observability/metrics")) 593 .await 594 .assert_ok(); 595 } 596 } 597 598 async fn r#in(db: &PgPool, subject: Option<IncomingKey>) { 599 register_tx_in( 600 db, 601 &TxIn { 602 balance_id: 42, 603 wise_ref: None, 604 amount: amount("EUR:10"), 605 subject: "subject".into(), 606 name: CompactString::const_new("Name"), 607 debtor: Some(WiseAccount::IBAN(BankID { 608 iban: iban("HU30162000031000163100000000"), 609 bic: None, 610 })), 611 value_at: Timestamp::now(), 612 }, 613 &subject, 614 &Timestamp::now(), 615 ) 616 .await 617 .unwrap(); 618 } 619 620 async fn in_malformed(db: &PgPool) { 621 r#in(db, None).await 622 } 623 624 async fn in_talerable(db: &PgPool) { 625 r#in(db, Some(IncomingKey::reserve(EddsaPublicKey::rand()))).await 626 } 627 628 #[db_test] 629 async fn admin_add_incoming(db: PgPool) { 630 let server = setup(&db).await; 631 admin_add_incoming_routine( 632 &server.prefix("/balances/42/taler-wire-gateway"), 633 &server.prefix("/balances/42/taler-prepared-transfer"), 634 &EXCHANGE, 635 &EXCHANGE, 636 ) 637 .await; 638 } 639 640 #[db_test] 641 async fn in_history(db: PgPool) { 642 let server = setup(&db).await; 643 let db = &db; 644 in_history_routine( 645 &server.prefix("/balances/42/taler-wire-gateway"), 646 &server.prefix("/balances/42/taler-prepared-transfer"), 647 &EXCHANGE, 648 &EXCHANGE, 649 tasks!({ in_talerable(db).await }), 650 tasks!({ in_malformed(db).await }), 651 ) 652 .await; 653 } 654 655 #[db_test] 656 async fn revenue(db: PgPool) { 657 let server = setup(&db).await; 658 let db = &db; 659 revenue_routine( 660 &server.prefix("/balances/42/taler-wire-gateway"), 661 &server.prefix("/balances/42/taler-revenue"), 662 &EXCHANGE, 663 tasks!({ in_malformed(db).await }, { in_talerable(db).await },), 664 tasks!(), 665 ) 666 .await; 667 } 668 }