ws.rs (14151B)
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 use std::time::Duration; 21 22 use compact_str::CompactString; 23 use futures_util::TryStreamExt as _; 24 use reqwest::{Client, StatusCode}; 25 use reqwest_websocket::{Message, Upgrade}; 26 use serde::{Deserialize, Serialize}; 27 use sqlx::PgPool; 28 use taler_common::ExpoBackoffDecorr; 29 use thiserror::Error; 30 use tracing::{debug, error, info, trace}; 31 32 use crate::{ 33 ebics::{ 34 EbicsClient, EbicsErrKind, 35 ebics_code::EbicsReturnCode, 36 order::{BTF, Order}, 37 }, 38 keys::{BankKeys, ClientKeys}, 39 }; 40 41 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] 42 #[serde(rename_all = "UPPERCASE")] 43 pub struct WssParams { 44 pub url: String, 45 pub token: String, 46 pub ott: String, 47 pub validity: String, 48 pub partnerid: String, 49 pub userid: Option<String>, 50 } 51 52 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] 53 #[serde(rename_all = "UPPERCASE")] 54 pub struct WssNotificationClass { 55 pub name: String, 56 pub vers: String, 57 pub timestamp: String, 58 } 59 60 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] 61 #[serde(rename_all = "UPPERCASE")] 62 pub struct WssNotificationBTF { 63 pub service: CompactString, 64 pub scope: Option<CompactString>, 65 pub option: Option<CompactString>, 66 pub conttype: Option<CompactString>, 67 pub msgname: CompactString, 68 pub variant: Option<CompactString>, 69 pub version: Option<CompactString>, 70 pub format: Option<CompactString>, 71 } 72 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] 73 #[serde(rename_all = "UPPERCASE")] 74 pub struct WssInfo { 75 pub lang: String, 76 pub free: String, 77 } 78 79 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] 80 #[serde(untagged)] 81 pub enum WssNotification { 82 // INFO 83 #[serde(rename_all = "UPPERCASE")] 84 GeneralInfo { 85 mclass: Vec<WssNotificationClass>, 86 info: Vec<WssInfo>, 87 }, 88 #[serde(rename_all = "UPPERCASE")] 89 NewData { 90 mclass: Vec<WssNotificationClass>, 91 partnerid: String, 92 userid: Option<String>, 93 btf: Vec<WssNotificationBTF>, 94 #[serde(default)] 95 ordertype: Vec<String>, 96 }, 97 } 98 99 impl WssParams { 100 async fn connect( 101 &self, 102 client: &Client, 103 mut lambda: impl AsyncFnMut(WssNotification), 104 ) -> Result<(), WssError> { 105 let Self { 106 url, 107 token, 108 partnerid, 109 userid, 110 .. 111 } = self; 112 let username = format!( 113 "{partnerid}{}", 114 std::fmt::from_fn(|f| if let Some(userid) = userid { 115 write!(f, "_{userid}") 116 } else { 117 Ok(()) 118 }) 119 ); 120 121 let mut ws = client 122 .get( 123 url.replace("https://", "wss://") 124 .replace("http://", "ws://"), 125 ) 126 .basic_auth(username, Some(&token)) 127 .upgrade() 128 .send() 129 .await? 130 .into_websocket() 131 .await?; 132 trace!(target: "wss", "wait for ws msg"); 133 while let Some(msg) = ws.try_next().await? { 134 match msg { 135 Message::Text(str) => { 136 // TODO handle error 137 let msg: WssNotification = serde_json::from_str(&str)?; 138 trace!(target: "wss", "received: {msg:?}"); 139 lambda(msg).await; 140 } 141 Message::Binary(_) => { 142 // TODO what should we do ? 143 } 144 Message::Ping(_) | Message::Pong(_) => { 145 // Handled by tungstenite 146 } 147 Message::Close { code, reason } => { 148 debug!(target: "wss", "closed {code} {reason}"); 149 break; 150 } 151 } 152 trace!(target: "wss", "wait for ws msg"); 153 } 154 Ok(()) 155 } 156 } 157 158 #[derive(Error, Debug)] 159 pub enum WssError { 160 #[error("ws: {0}")] 161 Ws(#[from] reqwest_websocket::Error), 162 #[error("ws JSON msg: {0}")] 163 ReqJson(#[from] serde_json::Error), 164 } 165 166 pub async fn listen_for_notification( 167 ebics: &EbicsClient<'_>, 168 db: &PgPool, 169 client: &ClientKeys, 170 bank: &BankKeys, 171 sender: tokio::sync::mpsc::Sender<Vec<Order>>, 172 ) { 173 let mut backoff = ExpoBackoffDecorr::new(Duration::from_secs(30), Duration::from_mins(30), 2.5); 174 loop { 175 let res: Result<bool, anyhow::Error> = async { 176 let res = ebics 177 .download( 178 db, 179 client, 180 bank, 181 &Order::WSS_PARAMS, 182 &None, 183 false, 184 async |content| { 185 serde_json::from_slice::<WssParams>(&content) 186 .map_err(|e| EbicsErrKind::Custom(e.to_string().into())) 187 }, 188 ) 189 .await; 190 let params = match res { 191 Ok(params) => params, 192 Err(e) => { 193 if matches!( 194 e.kind, 195 // Expected EBICS error 196 EbicsErrKind::Code { 197 technical: EbicsReturnCode::EBICS_INVALID_ORDER_TYPE, 198 .. 199 } | 200 // Netzbon HTTP error 201 EbicsErrKind::HTTP(StatusCode::BAD_REQUEST) 202 ) { 203 // Failure is expected if this wss is not supported 204 info!(target: "ws", "Real-time EBICS notifications is not supported"); 205 return Ok(true); 206 } else { 207 return Err(e.into()); 208 } 209 } 210 }; 211 info!(target: "ws", "Listening to real-time EBICS notifications"); 212 trace!(target: "ws", "{params:?}"); 213 214 params 215 .connect(&ebics.http, async |msg| { 216 backoff.reset(); 217 match msg { 218 WssNotification::GeneralInfo { info, .. } => { 219 for info in info { 220 info!(target: "ws", "info: {}", info.free); 221 } 222 } 223 WssNotification::NewData { btf, .. } => { 224 let orders = btf 225 .into_iter() 226 .map(|it| { 227 Order::BTD(BTF { 228 service: it.service, 229 scope: it.scope, 230 option: it.option, 231 container: it.conttype, 232 msg: it.msgname, 233 version: it.version, 234 }) 235 }) 236 .collect(); 237 sender.send(orders).await.ok(); 238 } 239 } 240 }) 241 .await?; 242 Ok(false) 243 } 244 .await; 245 match res { 246 Ok(stop) => { 247 if stop { 248 return; 249 } 250 } 251 Err(e) => { 252 error!(target: "ws", "{e}"); 253 tokio::time::sleep(backoff.backoff()).await; 254 } 255 } 256 } 257 } 258 259 #[cfg(test)] 260 mod test { 261 use std::{fmt::Debug, fs::Permissions, os::unix::fs::PermissionsExt as _}; 262 263 use axum::{ 264 extract::{ 265 WebSocketUpgrade, 266 ws::{CloseFrame, Message, Utf8Bytes}, 267 }, 268 http::HeaderMap, 269 routing::get, 270 }; 271 use reqwest::header::AUTHORIZATION; 272 use serde::{Serialize, de::DeserializeOwned}; 273 use taler_api::api::TalerRouter as _; 274 275 use crate::{ 276 test::wait_for_unix_socket, 277 ws::{WssNotification, WssParams}, 278 }; 279 280 // WSS params example from the spec 281 const PARAMS_EXAMPLE: &str = r#" 282 { 283 "URL": "http://bankmitwebsocket.de", 284 "TOKEN": "550e8400-e29b-11d4-a716-446655440000", 285 "OTT": "N", 286 "VALIDITY": "2019-03-21T10:35:22Z", 287 "PARTNERID": "K1234567", 288 "USERID": "USER4711" 289 } 290 "#; 291 // Authorization header example from the spec 292 const AUTH_EXAMPLE: &str = 293 "Basic SzEyMzQ1NjdfVVNFUjQ3MTE6NTUwZTg0MDAtZTI5Yi0xMWQ0LWE3MTYtNDQ2NjU1NDQwMDAw"; 294 // Notifications examples from the spec 295 const NOTIFICATION_EXAMPLES: [&str; 3] = [ 296 r#" 297 { 298 "MCLASS": [ 299 { 300 "NAME": "EBICS-HAA", 301 "VERS": "1.0", 302 "TIMESTAMP": "2019-05-13T12:21:50Z" 303 } 304 ], 305 "PARTNERID": "K1234567", 306 "USERID": "USER471", 307 "BTF": [ 308 { 309 "SERVICE": "REP", 310 "SCOPE": "DE", 311 "CONTTYPE": "ZIP", 312 "MSGNAME": "camt.054" 313 } 314 ], 315 "ORDERTYPE": [ 316 "C5N" 317 ] 318 } 319 "#, 320 r#" 321 { 322 "MCLASS": [ 323 { 324 "NAME": "EBICS-HAA", 325 "VERS": "1.0", 326 "TIMESTAMP": "2019-05-13T12:21:53Z" 327 } 328 ], 329 "PARTNERID": "K1234567", 330 "USERID": "USER471", 331 "BTF": [ 332 { 333 "SERVICE": "REP", 334 "SCOPE": "DE", 335 "CONTTYPE": "ZIP", 336 "MSGNAME": "camt.052" 337 }, 338 { 339 "SERVICE": "REP", 340 "SCOPE": "DE", 341 "OPTION": "SCI", 342 "CONTTYPE": "ZIP", 343 "MSGNAME": "pain.002" 344 } 345 ], 346 "ORDERTYPE": [ 347 "C52", 348 "CIZ" 349 ] 350 } 351 "#, 352 r#" 353 { 354 "MCLASS": [ 355 { 356 "NAME": "INFO", 357 "VERS": "1.0", 358 "TIMESTAMP": "2019-03-25T12:25:34Z" 359 } 360 ], 361 "INFO": [ 362 { 363 "LANG": "EN", 364 "FREE": " The EBICS-Service is limited on 30.03.2019 from 10:00 a.m. - 11:00a.m. due to maintenance work " 365 } 366 ] 367 } 368 "#, 369 ]; 370 371 #[test] 372 pub fn serialization() { 373 fn roundrip<T: Serialize + DeserializeOwned + Eq + Debug>(src: &str) { 374 let it: T = serde_json::from_str(src).unwrap(); 375 let roundrip: T = serde_json::from_str(&serde_json::to_string(&it).unwrap()).unwrap(); 376 assert_eq!(it, roundrip); 377 } 378 roundrip::<WssParams>(PARAMS_EXAMPLE); 379 for ex in NOTIFICATION_EXAMPLES { 380 roundrip::<WssNotification>(ex); 381 } 382 } 383 384 #[tokio::test] 385 pub async fn params() { 386 let path = "/tmp/libeufin_nexus_wss_test.sock"; 387 std::fs::remove_file(path).ok(); 388 389 tokio::spawn(async move { 390 axum::Router::new() 391 .route( 392 "/", 393 get(async |headers: HeaderMap, ws: WebSocketUpgrade| { 394 assert_eq!( 395 headers.get(AUTHORIZATION).map(|it| it.as_bytes()), 396 Some(AUTH_EXAMPLE.as_bytes()) 397 ); 398 ws.on_upgrade(async |mut it| { 399 for ex in NOTIFICATION_EXAMPLES { 400 it.send(Message::Text(Utf8Bytes::from_static(ex))) 401 .await 402 .unwrap(); 403 } 404 it.send(Message::Close(Some(CloseFrame { 405 code: 1000, 406 reason: Utf8Bytes::from_static("Test done"), 407 }))) 408 .await 409 .unwrap(); 410 }) 411 }), 412 ) 413 .serve( 414 &taler_api::Serve::Unix { 415 path: path.into(), 416 permission: Permissions::from_mode(0o660), 417 }, 418 None, 419 ) 420 .await 421 .unwrap() 422 }); 423 wait_for_unix_socket(path).await; 424 let client = reqwest::ClientBuilder::new() 425 .unix_socket(path) 426 .build() 427 .unwrap(); 428 let params: WssParams = serde_json::from_str(PARAMS_EXAMPLE).unwrap(); 429 let mut count = 0; 430 params 431 .connect(&client, async |msg| { 432 count += 1; 433 // Check message number and type 434 assert!(count <= 3); 435 if count == 3 { 436 assert!(matches!(msg, WssNotification::GeneralInfo { .. })) 437 } else { 438 assert!(matches!(msg, WssNotification::NewData { .. })) 439 } 440 }) 441 .await 442 .unwrap(); 443 // Check receive all messages 444 assert_eq!(3, count); 445 } 446 }