taler-helper-auditor-transfer.c (17125B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2017-2024 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU 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 General Public License for more details. 12 13 You should have received a copy of the GNU General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 /** 17 * @file auditor/taler-helper-auditor-transfer.c 18 * @brief audits that deposits past due date are 19 * aggregated and have a matching wire transfer 20 * database. 21 * @author Christian Grothoff 22 */ 23 #include "platform.h" 24 #include <gnunet/gnunet_util_lib.h> 25 #include <gnunet/gnunet_curl_lib.h> 26 #include "auditordb_lib.h" 27 #include "exchangedb_lib.h" 28 #include "taler/taler_json_lib.h" 29 #include "report-lib.h" 30 #include "taler/taler_dbevents.h" 31 #include "auditor-database/delete_early_aggregation.h" 32 #include "auditor-database/delete_pending_deposit.h" 33 #include "auditor-database/event_listen.h" 34 #include "auditor-database/get_auditor_progress.h" 35 #include "auditor-database/get_balance.h" 36 #include "auditor-database/insert_auditor_progress.h" 37 #include "auditor-database/insert_balance.h" 38 #include "auditor-database/insert_early_aggregation.h" 39 #include "auditor-database/insert_pending_deposit.h" 40 #include "auditor-database/preflight.h" 41 #include "auditor-database/start.h" 42 #include "auditor-database/update_auditor_progress.h" 43 #include "auditor-database/update_balance.h" 44 #include "exchange-database/preflight.h" 45 #include "exchange-database/rollback.h" 46 struct AggregationContext; 47 #define TALER_EXCHANGEDB_AGGREGATION_RESULT_CLOSURE struct AggregationContext 48 #include "exchange-database/select_aggregations_above_serial.h" 49 struct ImportMissingWireContext; 50 #define TALER_EXCHANGEDB_WIRE_MISSING_RESULT_CLOSURE struct \ 51 ImportMissingWireContext 52 #include "exchange-database/select_batch_deposits_missing_wire.h" 53 #include "exchange-database/start_read_only.h" 54 55 56 /** 57 * Run in test mode. Exit when idle instead of 58 * going to sleep and waiting for more work. 59 */ 60 static int test_mode; 61 62 /** 63 * Return value from main(). 64 */ 65 static int global_ret; 66 67 /** 68 * Last reserve_out / wire_out serial IDs seen. 69 */ 70 static TALER_ARL_DEF_PP (wire_batch_deposit_id); 71 static TALER_ARL_DEF_PP (wire_aggregation_id); 72 73 /** 74 * Total amount which the exchange did not aggregate/transfer in time. 75 */ 76 static TALER_ARL_DEF_AB (total_amount_lag); 77 78 /** 79 * Total amount which the exchange did aggregate/transfer too early. 80 */ 81 static TALER_ARL_DEF_AB (total_early_aggregation); 82 83 /** 84 * Should we run checks that only work for exchange-internal audits? 85 */ 86 static int internal_checks; 87 88 /** 89 * Database event handler to wake us up again. 90 */ 91 static struct GNUNET_DB_EventHandler *eh; 92 93 /** 94 * The auditors's configuration. 95 */ 96 static const struct GNUNET_CONFIGURATION_Handle *cfg; 97 98 99 /** 100 * Task run on shutdown. 101 * 102 * @param cls NULL 103 */ 104 static void 105 do_shutdown (void *cls) 106 { 107 (void) cls; 108 if (NULL != eh) 109 { 110 TALER_AUDITORDB_event_listen_cancel (eh); 111 eh = NULL; 112 } 113 TALER_ARL_done (); 114 TALER_EXCHANGEDB_unload_accounts (); 115 TALER_ARL_cfg = NULL; 116 } 117 118 119 /** 120 * Closure for import_wire_missing_cb(). 121 */ 122 struct ImportMissingWireContext 123 { 124 /** 125 * Set to maximum row ID encountered. 126 */ 127 uint64_t max_batch_deposit_uuid; 128 129 /** 130 * Set to database errors in callback. 131 */ 132 enum GNUNET_DB_QueryStatus err; 133 }; 134 135 136 /** 137 * Function called on deposits that need to be checked for their 138 * wire transfer. 139 * 140 * @param wc closure, points to a `struct ImportMissingWireContext` 141 * @param batch_deposit_serial_id serial of the entry in the batch deposits table 142 * @param total_amount value of the missing deposits, including fee 143 * @param wire_target_h_payto where should the funds be wired 144 * @param deadline what was the earliest requested wire transfer deadline 145 */ 146 static void 147 import_wire_missing_cb ( 148 struct ImportMissingWireContext *wc, 149 uint64_t batch_deposit_serial_id, 150 const struct TALER_Amount *total_amount, 151 const struct TALER_FullPaytoHashP *wire_target_h_payto, 152 struct GNUNET_TIME_Timestamp deadline) 153 { 154 enum GNUNET_DB_QueryStatus qs; 155 156 if (wc->err < 0) 157 return; /* already failed */ 158 GNUNET_assert (batch_deposit_serial_id >= wc->max_batch_deposit_uuid); 159 wc->max_batch_deposit_uuid = batch_deposit_serial_id + 1; 160 qs = TALER_AUDITORDB_delete_early_aggregation ( 161 TALER_ARL_adb, 162 batch_deposit_serial_id); 163 switch (qs) 164 { 165 case GNUNET_DB_STATUS_SOFT_ERROR: 166 GNUNET_break (0); 167 wc->err = qs; 168 return; 169 case GNUNET_DB_STATUS_HARD_ERROR: 170 GNUNET_break (0); 171 wc->err = qs; 172 return; 173 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 174 qs = TALER_AUDITORDB_insert_pending_deposit ( 175 TALER_ARL_adb, 176 batch_deposit_serial_id, 177 wire_target_h_payto, 178 total_amount, 179 deadline); 180 if (0 > qs) 181 { 182 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 183 wc->err = qs; 184 return; 185 } 186 TALER_ARL_amount_add (&TALER_ARL_USE_AB (total_amount_lag), 187 &TALER_ARL_USE_AB (total_amount_lag), 188 total_amount); 189 break; 190 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 191 TALER_ARL_amount_subtract (&TALER_ARL_USE_AB (total_early_aggregation), 192 &TALER_ARL_USE_AB (total_early_aggregation), 193 total_amount); 194 break; 195 default: 196 GNUNET_assert (0); 197 } 198 } 199 200 201 /** 202 * Checks for wire transfers that should have happened. 203 * 204 * @return transaction status 205 */ 206 static enum GNUNET_DB_QueryStatus 207 check_for_required_transfers (void) 208 { 209 enum GNUNET_DB_QueryStatus qs; 210 struct ImportMissingWireContext wc = { 211 .max_batch_deposit_uuid = TALER_ARL_USE_PP (wire_batch_deposit_id), 212 .err = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 213 }; 214 215 qs = TALER_EXCHANGEDB_select_batch_deposits_missing_wire ( 216 TALER_ARL_edb, 217 TALER_ARL_USE_PP (wire_batch_deposit_id), 218 &import_wire_missing_cb, 219 &wc); 220 if (0 > qs) 221 { 222 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 223 return qs; 224 } 225 if (0 > wc.err) 226 { 227 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == wc.err); 228 return wc.err; 229 } 230 TALER_ARL_USE_PP (wire_batch_deposit_id) = wc.max_batch_deposit_uuid; 231 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 232 } 233 234 235 /** 236 * Closure for #clear_finished_transfer_cb(). 237 */ 238 struct AggregationContext 239 { 240 /** 241 * Set to maximum row ID encountered. 242 */ 243 uint64_t max_aggregation_serial; 244 245 /** 246 * Set to database errors in callback. 247 */ 248 enum GNUNET_DB_QueryStatus err; 249 }; 250 251 252 /** 253 * Function called on aggregations that were done for 254 * a (batch) deposit. 255 * 256 * @param ac closure 257 * @param amount affected amount 258 * @param tracking_serial_id where in the table are we 259 * @param batch_deposit_serial_id which batch deposit was aggregated 260 */ 261 static void 262 clear_finished_transfer_cb ( 263 struct AggregationContext *ac, 264 const struct TALER_Amount *amount, 265 uint64_t tracking_serial_id, 266 uint64_t batch_deposit_serial_id) 267 { 268 enum GNUNET_DB_QueryStatus qs; 269 270 if (0 > ac->err) 271 return; /* already failed */ 272 GNUNET_assert (ac->max_aggregation_serial <= tracking_serial_id); 273 ac->max_aggregation_serial = tracking_serial_id + 1; 274 qs = TALER_AUDITORDB_delete_pending_deposit ( 275 TALER_ARL_adb, 276 batch_deposit_serial_id); 277 switch (qs) 278 { 279 case GNUNET_DB_STATUS_SOFT_ERROR: 280 GNUNET_break (0); 281 ac->err = qs; 282 return; 283 case GNUNET_DB_STATUS_HARD_ERROR: 284 GNUNET_break (0); 285 ac->err = qs; 286 return; 287 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 288 qs = TALER_AUDITORDB_insert_early_aggregation ( 289 TALER_ARL_adb, 290 batch_deposit_serial_id, 291 tracking_serial_id, 292 amount); 293 if (0 > qs) 294 { 295 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 296 ac->err = qs; 297 return; 298 } 299 TALER_ARL_amount_add (&TALER_ARL_USE_AB (total_early_aggregation), 300 &TALER_ARL_USE_AB (total_early_aggregation), 301 amount); 302 break; 303 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 304 TALER_ARL_amount_subtract (&TALER_ARL_USE_AB (total_amount_lag), 305 &TALER_ARL_USE_AB (total_amount_lag), 306 amount); 307 break; 308 default: 309 GNUNET_assert (0); 310 } 311 } 312 313 314 /** 315 * Checks that all wire transfers that should have happened 316 * (based on deposits) have indeed happened. 317 * 318 * @return transaction status 319 */ 320 static enum GNUNET_DB_QueryStatus 321 check_for_completed_transfers (void) 322 { 323 struct AggregationContext ac = { 324 .max_aggregation_serial = TALER_ARL_USE_PP (wire_aggregation_id), 325 .err = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 326 }; 327 enum GNUNET_DB_QueryStatus qs; 328 329 qs = TALER_EXCHANGEDB_select_aggregations_above_serial ( 330 TALER_ARL_edb, 331 TALER_ARL_USE_PP (wire_aggregation_id), 332 &clear_finished_transfer_cb, 333 &ac); 334 if (0 > qs) 335 { 336 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 337 return qs; 338 } 339 if (0 > ac.err) 340 { 341 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == ac.err); 342 return ac.err; 343 } 344 TALER_ARL_USE_PP (wire_aggregation_id) = ac.max_aggregation_serial; 345 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 346 } 347 348 349 /** 350 * Start the database transactions and begin the audit. 351 * 352 * @return transaction status 353 */ 354 static enum GNUNET_DB_QueryStatus 355 begin_transaction (void) 356 { 357 enum GNUNET_DB_QueryStatus qs; 358 359 if (GNUNET_SYSERR == 360 TALER_EXCHANGEDB_preflight (TALER_ARL_edb)) 361 { 362 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 363 "Failed to initialize exchange database connection.\n"); 364 return GNUNET_DB_STATUS_HARD_ERROR; 365 } 366 if (GNUNET_SYSERR == 367 TALER_AUDITORDB_preflight (TALER_ARL_adb)) 368 { 369 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 370 "Failed to initialize auditor database session.\n"); 371 return GNUNET_DB_STATUS_HARD_ERROR; 372 } 373 if (GNUNET_OK != 374 TALER_AUDITORDB_start (TALER_ARL_adb, 375 "auditor-transfer")) 376 { 377 GNUNET_break (0); 378 return GNUNET_DB_STATUS_HARD_ERROR; 379 } 380 if (GNUNET_OK != 381 TALER_TALER_EXCHANGEDB_start_read_only (TALER_ARL_edb, 382 "transfer auditor")) 383 { 384 GNUNET_break (0); 385 TALER_AUDITORDB_rollback (TALER_ARL_adb); 386 return GNUNET_DB_STATUS_HARD_ERROR; 387 } 388 qs = TALER_AUDITORDB_get_auditor_progress ( 389 TALER_ARL_adb, 390 TALER_ARL_GET_PP (wire_batch_deposit_id), 391 TALER_ARL_GET_PP (wire_aggregation_id), 392 NULL); 393 if (0 > qs) 394 goto handle_db_error; 395 396 qs = TALER_AUDITORDB_get_balance ( 397 TALER_ARL_adb, 398 TALER_ARL_GET_AB (total_amount_lag), 399 TALER_ARL_GET_AB (total_early_aggregation), 400 NULL); 401 if (0 > qs) 402 goto handle_db_error; 403 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 404 { 405 GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE, 406 "First analysis of with transfer auditor, starting audit from scratch\n"); 407 } 408 else 409 { 410 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 411 "Resuming transfer audit at %llu / %llu\n", 412 (unsigned long long) TALER_ARL_USE_PP (wire_batch_deposit_id), 413 (unsigned long long) TALER_ARL_USE_PP (wire_aggregation_id)); 414 } 415 416 qs = check_for_required_transfers (); 417 if (0 > qs) 418 goto handle_db_error; 419 qs = check_for_completed_transfers (); 420 if (0 > qs) 421 goto handle_db_error; 422 423 qs = TALER_AUDITORDB_update_auditor_progress ( 424 TALER_ARL_adb, 425 TALER_ARL_SET_PP (wire_batch_deposit_id), 426 TALER_ARL_SET_PP (wire_aggregation_id), 427 NULL); 428 if (0 > qs) 429 goto handle_db_error; 430 qs = TALER_AUDITORDB_insert_auditor_progress ( 431 TALER_ARL_adb, 432 TALER_ARL_SET_PP (wire_batch_deposit_id), 433 TALER_ARL_SET_PP (wire_aggregation_id), 434 NULL); 435 if (0 > qs) 436 goto handle_db_error; 437 qs = TALER_AUDITORDB_update_balance ( 438 TALER_ARL_adb, 439 TALER_ARL_SET_AB (total_amount_lag), 440 TALER_ARL_SET_AB (total_early_aggregation), 441 NULL); 442 if (0 > qs) 443 goto handle_db_error; 444 qs = TALER_AUDITORDB_insert_balance ( 445 TALER_ARL_adb, 446 TALER_ARL_SET_AB (total_amount_lag), 447 TALER_ARL_SET_AB (total_early_aggregation), 448 NULL); 449 if (0 > qs) 450 goto handle_db_error; 451 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 452 "Concluded audit step at %llu/%llu\n", 453 (unsigned long long) TALER_ARL_USE_PP (wire_aggregation_id), 454 (unsigned long long) TALER_ARL_USE_PP (wire_batch_deposit_id)); 455 TALER_EXCHANGEDB_rollback (TALER_ARL_edb); 456 qs = TALER_AUDITORDB_commit (TALER_ARL_adb); 457 if (0 > qs) 458 goto handle_db_error; 459 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 460 "Transaction concluded!\n"); 461 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 462 handle_db_error: 463 TALER_AUDITORDB_rollback (TALER_ARL_adb); 464 TALER_EXCHANGEDB_rollback (TALER_ARL_edb); 465 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 466 return qs; 467 } 468 469 470 /** 471 * Start auditor process. 472 */ 473 static void 474 start (void) 475 { 476 enum GNUNET_DB_QueryStatus qs; 477 478 for (unsigned int max_retries = 3; max_retries>0; max_retries--) 479 { 480 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 481 "Trying again (%u attempts left)\n", 482 max_retries); 483 qs = begin_transaction (); 484 if (GNUNET_DB_STATUS_SOFT_ERROR != qs) 485 break; 486 } 487 if (0 > qs) 488 { 489 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 490 "Audit failed\n"); 491 GNUNET_break (0); 492 global_ret = EXIT_FAILURE; 493 GNUNET_SCHEDULER_shutdown (); 494 return; 495 } 496 } 497 498 499 /** 500 * Function called on events received from Postgres. 501 * 502 * @param cls closure, NULL 503 * @param extra additional event data provided 504 * @param extra_size number of bytes in @a extra 505 */ 506 static void 507 db_notify (void *cls, 508 const void *extra, 509 size_t extra_size) 510 { 511 (void) cls; 512 (void) extra; 513 (void) extra_size; 514 515 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 516 "Received notification to wake transfer helper\n"); 517 start (); 518 } 519 520 521 /** 522 * Main function that will be run. 523 * 524 * @param cls closure 525 * @param args remaining command-line arguments 526 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 527 * @param c configuration 528 */ 529 static void 530 run (void *cls, 531 char *const *args, 532 const char *cfgfile, 533 const struct GNUNET_CONFIGURATION_Handle *c) 534 { 535 (void) cls; 536 (void) args; 537 (void) cfgfile; 538 cfg = c; 539 if (GNUNET_OK != 540 TALER_ARL_init (c)) 541 { 542 global_ret = EXIT_FAILURE; 543 return; 544 } 545 GNUNET_SCHEDULER_add_shutdown (&do_shutdown, 546 NULL); 547 if (GNUNET_OK != 548 TALER_EXCHANGEDB_load_accounts (TALER_ARL_cfg, 549 TALER_EXCHANGEDB_ALO_DEBIT 550 | TALER_EXCHANGEDB_ALO_CREDIT 551 | TALER_EXCHANGEDB_ALO_AUTHDATA)) 552 { 553 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 554 "No bank accounts configured\n"); 555 global_ret = EXIT_NOTCONFIGURED; 556 GNUNET_SCHEDULER_shutdown (); 557 return; 558 } 559 if (0 == test_mode) 560 { 561 // FIXME-Optimization: use different event type in the future! 562 struct GNUNET_DB_EventHeaderP es = { 563 .size = htons (sizeof (es)), 564 .type = htons (TALER_DBEVENT_EXCHANGE_AUDITOR_WAKE_HELPER_WIRE) 565 }; 566 567 eh = TALER_AUDITORDB_event_listen (TALER_ARL_adb, 568 &es, 569 GNUNET_TIME_UNIT_FOREVER_REL, 570 &db_notify, 571 NULL); 572 GNUNET_assert (NULL != eh); 573 } 574 start (); 575 } 576 577 578 /** 579 * The main function of the wire auditing tool. Checks that 580 * the exchange's records of wire transfers match that of 581 * the wire gateway. 582 * 583 * @param argc number of arguments from the command line 584 * @param argv command line arguments 585 * @return 0 ok, 1 on error 586 */ 587 int 588 main (int argc, 589 char *const *argv) 590 { 591 const struct GNUNET_GETOPT_CommandLineOption options[] = { 592 GNUNET_GETOPT_option_flag ('i', 593 "internal", 594 "perform checks only applicable for exchange-internal audits", 595 &internal_checks), 596 GNUNET_GETOPT_option_flag ('t', 597 "test", 598 "run in test mode and exit when idle", 599 &test_mode), 600 GNUNET_GETOPT_option_timetravel ('T', 601 "timetravel"), 602 GNUNET_GETOPT_OPTION_END 603 }; 604 enum GNUNET_GenericReturnValue ret; 605 606 ret = GNUNET_PROGRAM_run ( 607 TALER_AUDITOR_project_data (), 608 argc, 609 argv, 610 "taler-helper-auditor-transfer", 611 gettext_noop ( 612 "Audit exchange database for consistency of aggregations/transfers with respect to deposit deadlines"), 613 options, 614 &run, 615 NULL); 616 if (GNUNET_SYSERR == ret) 617 return EXIT_INVALIDARGUMENT; 618 if (GNUNET_NO == ret) 619 return EXIT_SUCCESS; 620 return global_ret; 621 } 622 623 624 /* end of taler-helper-auditor-transfer.c */