taler-helper-auditor-transfer.c (41005B)
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 * 22 * Three checks, each with its own cursor over the exchange's tables: 23 * 24 * 1) every batch deposit must eventually be aggregated (`total_amount_lag`), 25 * 2) nothing may be aggregated before there is a deposit to justify it 26 * (`total_early_aggregation`), 27 * 3) every aggregation must eventually turn into a wire transfer. 28 * 29 * The third is what keeps "the exchange paid the merchant" apart from "the 30 * exchange decided what it owes and then sat on the money": an 31 * `aggregation_tracking` row is not proof of payment, a `wire_out` row is. 32 * What is still owed is split by the reason the exchange gave for not paying 33 * -- `total_aml_hold` for an open KYC requirement, `total_small_aggregate` for 34 * an aggregate that does not yet cover its wire fee, and `total_transfer_lag` 35 * for a transfer the exchange never gave any reason for -- and each held 36 * transfer is listed in `auditor_aml_holds`. 37 * 38 * @author Christian Grothoff 39 */ 40 #include "platform.h" 41 #include <gnunet/gnunet_util_lib.h> 42 #include <gnunet/gnunet_curl_lib.h> 43 #include "auditordb_lib.h" 44 #include "exchangedb_lib.h" 45 #include "taler/taler_json_lib.h" 46 #include "report-lib.h" 47 #include "taler/taler_dbevents.h" 48 #include "auditor-database/delete_aml_hold.h" 49 #include "auditor-database/delete_early_aggregation.h" 50 #include "auditor-database/delete_pending_deposit.h" 51 #include "auditor-database/event_listen.h" 52 #include "auditor-database/get_auditor_progress.h" 53 #include "auditor-database/get_balance.h" 54 #include "auditor-database/insert_aml_hold.h" 55 #include "auditor-database/insert_amount_arithmetic_inconsistency.h" 56 #include "auditor-database/insert_auditor_progress.h" 57 #include "auditor-database/insert_balance.h" 58 #include "auditor-database/insert_early_aggregation.h" 59 #include "auditor-database/insert_pending_deposit.h" 60 #include "auditor-database/insert_row_inconsistency.h" 61 #include "auditor-database/preflight.h" 62 #include "auditor-database/start.h" 63 #include "auditor-database/update_aml_hold.h" 64 #include "auditor-database/update_auditor_progress.h" 65 #include "auditor-database/update_balance.h" 66 #include "exchange-database/get_aggregation_deferral_by_wtid.h" 67 #include "exchange-database/get_aggregation_transient_by_wtid.h" 68 #include "exchange-database/get_open_legitimization_measure.h" 69 #include "exchange-database/get_pending_aggregation.h" 70 #include "exchange-database/preflight.h" 71 #include "exchange-database/rollback.h" 72 struct AggregationContext; 73 #define TALER_EXCHANGEDB_AGGREGATION_RESULT_CLOSURE struct AggregationContext 74 #include "exchange-database/iterate_aggregations_above_serial_id.h" 75 struct ImportMissingWireContext; 76 #define TALER_EXCHANGEDB_WIRE_MISSING_RESULT_CLOSURE struct \ 77 ImportMissingWireContext 78 #include "exchange-database/iterate_batch_deposits_missing_wire.h" 79 struct HoldContext; 80 #define TALER_EXCHANGEDB_AGGREGATION_WTID_RESULT_CLOSURE struct HoldContext 81 #include "exchange-database/iterate_aggregation_wtids_above_serial_id.h" 82 #define TALER_AUDITORDB_AML_HOLD_RESULT_CLOSURE struct HoldContext 83 #include "auditor-database/iterate_aml_holds.h" 84 #include "exchange-database/start_read_only.h" 85 86 87 /** 88 * Run in test mode. Exit when idle instead of 89 * going to sleep and waiting for more work. 90 */ 91 static int test_mode; 92 93 /** 94 * Return value from main(). 95 */ 96 static int global_ret; 97 98 /** 99 * Last reserve_out / wire_out serial IDs seen. 100 */ 101 static TALER_ARL_DEF_PP (wire_batch_deposit_id); 102 static TALER_ARL_DEF_PP (wire_aggregation_id); 103 104 /** 105 * Row of `aggregation_tracking' up to which we have looked for wire transfers 106 * the exchange aggregated but did not execute. 107 */ 108 static TALER_ARL_DEF_PP (wire_hold_aggregation_id); 109 110 /** 111 * Total amount which the exchange did not aggregate/transfer in time. 112 */ 113 static TALER_ARL_DEF_AB (total_amount_lag); 114 115 /** 116 * Total amount which the exchange did aggregate/transfer too early. 117 */ 118 static TALER_ARL_DEF_AB (total_early_aggregation); 119 120 /** 121 * Total amount the exchange aggregated but did not wire because it says a 122 * legitimization requirement against the recipient is still open. Money the 123 * exchange is withholding for legal reasons. 124 */ 125 static TALER_ARL_DEF_AB (total_aml_hold); 126 127 /** 128 * Total amount the exchange aggregated but did not wire because what it has 129 * collected for the account so far does not cover the wire fee. Money that is 130 * waiting for the next deposit rather than for anyone's decision, and thus not 131 * an irregularity at all -- it is kept apart so that it does not inflate the 132 * two balances that are. 133 */ 134 static TALER_ARL_DEF_AB (total_small_aggregate); 135 136 /** 137 * Total amount the exchange aggregated but did not wire and gave no reason 138 * for. Money the exchange simply has not paid out. 139 */ 140 static TALER_ARL_DEF_AB (total_transfer_lag); 141 142 /** 143 * Should we run checks that only work for exchange-internal audits? 144 */ 145 static int internal_checks; 146 147 /** 148 * Database event handler to wake us up again. 149 */ 150 static struct GNUNET_DB_EventHandler *eh; 151 152 /** 153 * The auditors's configuration. 154 */ 155 static const struct GNUNET_CONFIGURATION_Handle *cfg; 156 157 158 /** 159 * Task run on shutdown. 160 * 161 * @param cls NULL 162 */ 163 static void 164 do_shutdown (void *cls) 165 { 166 (void) cls; 167 if (NULL != eh) 168 { 169 TALER_AUDITORDB_event_listen_cancel (eh); 170 eh = NULL; 171 } 172 TALER_ARL_done (); 173 TALER_EXCHANGEDB_unload_accounts (); 174 TALER_ARL_cfg = NULL; 175 } 176 177 178 /** 179 * Closure for import_wire_missing_cb(). 180 */ 181 struct ImportMissingWireContext 182 { 183 /** 184 * Set to maximum row ID encountered. 185 */ 186 uint64_t max_batch_deposit_uuid; 187 188 /** 189 * Set to database errors in callback. 190 */ 191 enum GNUNET_DB_QueryStatus err; 192 }; 193 194 195 /** 196 * Function called on deposits that need to be checked for their 197 * wire transfer. 198 * 199 * @param wc closure, points to a `struct ImportMissingWireContext` 200 * @param batch_deposit_serial_id serial of the entry in the batch deposits table 201 * @param total_amount value of the missing deposits, including fee 202 * @param wire_target_h_payto where should the funds be wired 203 * @param deadline what was the earliest requested wire transfer deadline 204 */ 205 static void 206 import_wire_missing_cb ( 207 struct ImportMissingWireContext *wc, 208 uint64_t batch_deposit_serial_id, 209 const struct TALER_Amount *total_amount, 210 const struct TALER_FullPaytoHashP *wire_target_h_payto, 211 struct GNUNET_TIME_Timestamp deadline) 212 { 213 enum GNUNET_DB_QueryStatus qs; 214 215 if (wc->err < 0) 216 return; /* already failed */ 217 GNUNET_assert (batch_deposit_serial_id >= wc->max_batch_deposit_uuid); 218 wc->max_batch_deposit_uuid = batch_deposit_serial_id + 1; 219 qs = TALER_AUDITORDB_delete_early_aggregation ( 220 TALER_ARL_adb, 221 batch_deposit_serial_id); 222 switch (qs) 223 { 224 case GNUNET_DB_STATUS_SOFT_ERROR: 225 GNUNET_break (0); 226 wc->err = qs; 227 return; 228 case GNUNET_DB_STATUS_HARD_ERROR: 229 GNUNET_break (0); 230 wc->err = qs; 231 return; 232 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 233 qs = TALER_AUDITORDB_insert_pending_deposit ( 234 TALER_ARL_adb, 235 batch_deposit_serial_id, 236 wire_target_h_payto, 237 total_amount, 238 deadline); 239 if (0 > qs) 240 { 241 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 242 wc->err = qs; 243 return; 244 } 245 TALER_ARL_amount_add (&TALER_ARL_USE_AB (total_amount_lag), 246 &TALER_ARL_USE_AB (total_amount_lag), 247 total_amount); 248 break; 249 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 250 TALER_ARL_amount_subtract (&TALER_ARL_USE_AB (total_early_aggregation), 251 &TALER_ARL_USE_AB (total_early_aggregation), 252 total_amount); 253 break; 254 default: 255 GNUNET_assert (0); 256 } 257 } 258 259 260 /** 261 * Checks for wire transfers that should have happened. 262 * 263 * @return transaction status 264 */ 265 static enum GNUNET_DB_QueryStatus 266 check_for_required_transfers (void) 267 { 268 enum GNUNET_DB_QueryStatus qs; 269 struct ImportMissingWireContext wc = { 270 .max_batch_deposit_uuid = TALER_ARL_USE_PP (wire_batch_deposit_id), 271 .err = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 272 }; 273 274 qs = TALER_EXCHANGEDB_iterate_batch_deposits_missing_wire ( 275 TALER_ARL_edb, 276 TALER_ARL_USE_PP (wire_batch_deposit_id), 277 &import_wire_missing_cb, 278 &wc); 279 if (0 > qs) 280 { 281 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 282 return qs; 283 } 284 if (0 > wc.err) 285 { 286 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == wc.err); 287 return wc.err; 288 } 289 TALER_ARL_USE_PP (wire_batch_deposit_id) = wc.max_batch_deposit_uuid; 290 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 291 } 292 293 294 /** 295 * Closure for #clear_finished_transfer_cb(). 296 */ 297 struct AggregationContext 298 { 299 /** 300 * Set to maximum row ID encountered. 301 */ 302 uint64_t max_aggregation_serial; 303 304 /** 305 * Set to database errors in callback. 306 */ 307 enum GNUNET_DB_QueryStatus err; 308 }; 309 310 311 /** 312 * Function called on aggregations that were done for 313 * a (batch) deposit. 314 * 315 * @param ac closure 316 * @param amount affected amount 317 * @param tracking_serial_id where in the table are we 318 * @param batch_deposit_serial_id which batch deposit was aggregated 319 */ 320 static void 321 clear_finished_transfer_cb ( 322 struct AggregationContext *ac, 323 const struct TALER_Amount *amount, 324 uint64_t tracking_serial_id, 325 uint64_t batch_deposit_serial_id) 326 { 327 enum GNUNET_DB_QueryStatus qs; 328 329 if (0 > ac->err) 330 return; /* already failed */ 331 GNUNET_assert (ac->max_aggregation_serial <= tracking_serial_id); 332 ac->max_aggregation_serial = tracking_serial_id + 1; 333 qs = TALER_AUDITORDB_delete_pending_deposit ( 334 TALER_ARL_adb, 335 batch_deposit_serial_id); 336 switch (qs) 337 { 338 case GNUNET_DB_STATUS_SOFT_ERROR: 339 GNUNET_break (0); 340 ac->err = qs; 341 return; 342 case GNUNET_DB_STATUS_HARD_ERROR: 343 GNUNET_break (0); 344 ac->err = qs; 345 return; 346 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 347 qs = TALER_AUDITORDB_insert_early_aggregation ( 348 TALER_ARL_adb, 349 batch_deposit_serial_id, 350 tracking_serial_id, 351 amount); 352 if (0 > qs) 353 { 354 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 355 ac->err = qs; 356 return; 357 } 358 TALER_ARL_amount_add (&TALER_ARL_USE_AB (total_early_aggregation), 359 &TALER_ARL_USE_AB (total_early_aggregation), 360 amount); 361 break; 362 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 363 TALER_ARL_amount_subtract (&TALER_ARL_USE_AB (total_amount_lag), 364 &TALER_ARL_USE_AB (total_amount_lag), 365 amount); 366 break; 367 default: 368 GNUNET_assert (0); 369 } 370 } 371 372 373 /** 374 * Checks that all wire transfers that should have happened 375 * (based on deposits) have indeed happened. 376 * 377 * @return transaction status 378 */ 379 static enum GNUNET_DB_QueryStatus 380 check_for_completed_transfers (void) 381 { 382 struct AggregationContext ac = { 383 .max_aggregation_serial = TALER_ARL_USE_PP (wire_aggregation_id), 384 .err = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 385 }; 386 enum GNUNET_DB_QueryStatus qs; 387 388 qs = TALER_EXCHANGEDB_iterate_aggregations_above_serial_id ( 389 TALER_ARL_edb, 390 TALER_ARL_USE_PP (wire_aggregation_id), 391 &clear_finished_transfer_cb, 392 &ac); 393 if (0 > qs) 394 { 395 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 396 return qs; 397 } 398 if (0 > ac.err) 399 { 400 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == ac.err); 401 return ac.err; 402 } 403 TALER_ARL_USE_PP (wire_aggregation_id) = ac.max_aggregation_serial; 404 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 405 } 406 407 408 /** 409 * Closure for #note_held_transfer_cb() and #evaluate_hold_cb(). 410 */ 411 struct HoldContext 412 { 413 /** 414 * Wire transfers already evaluated in this round, so that the three 415 * `aggregation_tracking` rows of one transfer cost one evaluation, and so 416 * that the pass over the holds already on file does not redo them. 417 */ 418 struct GNUNET_CONTAINER_MultiHashMap *seen; 419 420 /** 421 * Sum of the holds the exchange attributes to an open KYC requirement. 422 */ 423 struct TALER_Amount aml_hold; 424 425 /** 426 * Sum of the holds the exchange attributes to the aggregate being too small 427 * to be worth its wire fee. 428 */ 429 struct TALER_Amount small_aggregate; 430 431 /** 432 * Sum of the holds the exchange gave no reason for. 433 */ 434 struct TALER_Amount transfer_lag; 435 436 /** 437 * Set to maximum row ID encountered. 438 */ 439 uint64_t max_aggregation_serial; 440 441 /** 442 * Set to database errors in the callback. 443 */ 444 enum GNUNET_DB_QueryStatus err; 445 }; 446 447 448 /** 449 * Report a row inconsistency. 450 * 451 * @param[in,out] hc our state, to flag database failures in 452 * @param table name of the exchange table the bad row is in 453 * @param rowid row that is bad, 0 if the finding is not about one row 454 * @param diagnostic what is wrong with it 455 */ 456 static void 457 report_row (struct HoldContext *hc, 458 const char *table, 459 uint64_t rowid, 460 const char *diagnostic) 461 { 462 struct TALER_AUDITORDB_RowInconsistency ri = { 463 .row_id = rowid, 464 .row_table = (char *) table, 465 .diagnostic = (char *) diagnostic 466 }; 467 enum GNUNET_DB_QueryStatus qs; 468 469 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 470 "Row %llu of `%s' is bad: %s\n", 471 (unsigned long long) rowid, 472 table, 473 diagnostic); 474 qs = TALER_AUDITORDB_insert_row_inconsistency (TALER_ARL_adb, 475 &ri); 476 if (0 > qs) 477 { 478 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 479 hc->err = qs; 480 } 481 } 482 483 484 /** 485 * Work out whether the exchange still owes the wire transfer @a wtid, how much 486 * it owes and what it says is keeping it from paying; then record the result 487 * and count it towards the running totals. 488 * 489 * The classification is the exchange's own, read out of `aggregation_deferrals` 490 * -- append-only, and therefore the one statement of its reasons that survives 491 * replication to an external auditor. Taking it at face value is deliberate: 492 * whether a hold is *justified* is a question for the people reading 493 * `/monitoring/aml-holds`, not for the auditor. What the auditor does check is 494 * that the claim is consistent with everything else it can see: the amount 495 * against its own recomputation, a KYC claim against the legitimization tables, 496 * and under `-i` the whole thing against the live `aggregation_transient`. A 497 * disagreement is reported rather than resolved in either side's favour. 498 * 499 * A transfer with no `aggregation_deferrals` row at all is a transfer the 500 * exchange decided on, did not make, and never explained; that is 501 * `total_transfer_lag`. 502 * 503 * @param[in,out] hc our state 504 * @param wtid wire transfer to examine 505 */ 506 static void 507 evaluate_hold (struct HoldContext *hc, 508 const struct TALER_WireTransferIdentifierRawP *wtid) 509 { 510 struct TALER_FullPaytoHashP h_payto; 511 struct TALER_FullPayto payto = { NULL }; 512 struct TALER_NormalizedPaytoHashP h_normalized_payto; 513 struct TALER_Amount deposited; 514 struct TALER_Amount refunded; 515 struct TALER_Amount deposit_fee; 516 struct TALER_Amount deductions; 517 struct TALER_Amount amount; 518 struct TALER_Amount claimed; 519 struct GNUNET_TIME_Timestamp deferral_time; 520 struct GNUNET_TIME_Absolute measure_start; 521 enum TALER_EXCHANGEDB_DeferralReason reason = TALER_EXCHANGEDB_DR_NONE; 522 uint64_t claimed_measure = 0; 523 uint64_t measure = 0; 524 bool have_claim; 525 bool open_measure; 526 enum GNUNET_DB_QueryStatus qs; 527 528 qs = TALER_EXCHANGEDB_get_pending_aggregation (TALER_ARL_edb, 529 wtid, 530 &h_payto, 531 &payto, 532 &deposited, 533 &refunded, 534 &deposit_fee); 535 if (0 > qs) 536 { 537 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 538 hc->err = qs; 539 return; 540 } 541 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 542 { 543 /* The `wire_out' row has appeared: the money left the exchange, so the 544 hold ends and stops counting towards either total. This is also the 545 only place a hold is ever cleared -- there is no separate pass over 546 `wire_out`, because a transfer that was executed simply stops being a 547 pending aggregation. */ 548 qs = TALER_AUDITORDB_delete_aml_hold (TALER_ARL_adb, 549 wtid); 550 if (0 > qs) 551 { 552 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 553 hc->err = qs; 554 } 555 return; 556 } 557 558 /* What the exchange still owes, computed the way `exchange_do_aggregate' 559 computes it, so that it is comparable with what the exchange itself 560 recorded. */ 561 TALER_ARL_amount_add (&deductions, 562 &refunded, 563 &deposit_fee); 564 if (TALER_ARL_SR_INVALID_NEGATIVE == 565 TALER_ARL_amount_subtract_neg (&amount, 566 &deposited, 567 &deductions)) 568 { 569 char *diag; 570 571 GNUNET_asprintf (&diag, 572 "refunds and deposit fees exceed the deposits aggregated" 573 " into wire transfer %s", 574 TALER_B2S (wtid)); 575 report_row (hc, 576 "aggregation_tracking", 577 0, 578 diag); 579 GNUNET_free (diag); 580 GNUNET_assert (GNUNET_OK == 581 TALER_amount_set_zero (TALER_ARL_currency, 582 &amount)); 583 } 584 585 /* What the exchange says about it. Absent for a transfer it never 586 explained, and for one deferred by a version of the exchange from before 587 `aggregation_deferrals' existed. */ 588 qs = TALER_EXCHANGEDB_get_aggregation_deferral_by_wtid (TALER_ARL_edb, 589 wtid, 590 &claimed, 591 &reason, 592 &claimed_measure, 593 &deferral_time); 594 if (0 > qs) 595 { 596 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 597 hc->err = qs; 598 GNUNET_free (payto.full_payto); 599 return; 600 } 601 have_claim = (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs); 602 if (! have_claim) 603 { 604 reason = TALER_EXCHANGEDB_DR_NONE; 605 claimed_measure = 0; 606 } 607 608 /* `legitimization_measures' is reached through the account's *normalized* 609 payto hash, and `wire_targets.h_normalized_payto' is not replicated, so 610 the normalization has to happen here rather than in the query. */ 611 TALER_full_payto_normalize_and_hash (payto, 612 &h_normalized_payto); 613 qs = TALER_EXCHANGEDB_get_open_legitimization_measure ( 614 TALER_ARL_edb, 615 &h_normalized_payto, 616 &measure, 617 &measure_start); 618 if (0 > qs) 619 { 620 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 621 hc->err = qs; 622 GNUNET_free (payto.full_payto); 623 return; 624 } 625 open_measure = (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs); 626 if (! open_measure) 627 measure = 0; 628 switch (reason) 629 { 630 case TALER_EXCHANGEDB_DR_NONE: 631 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 632 "Exchange holds %s for %s and says nothing about why\n", 633 TALER_amount2s (&amount), 634 payto.full_payto); 635 break; 636 case TALER_EXCHANGEDB_DR_AMOUNT_TOO_SMALL: 637 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 638 "Exchange holds %s for %s, too little to cover the wire fee\n", 639 TALER_amount2s (&amount), 640 payto.full_payto); 641 break; 642 case TALER_EXCHANGEDB_DR_KYC: 643 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 644 "Exchange holds %s for %s awaiting KYC measure %llu\n", 645 TALER_amount2s (&amount), 646 payto.full_payto, 647 (unsigned long long) claimed_measure); 648 break; 649 } 650 651 if (have_claim && 652 (0 != TALER_amount_cmp (&claimed, 653 &amount)) ) 654 { 655 struct TALER_AUDITORDB_AmountArithmeticInconsistency aai = { 656 .problem_row_id = 0, 657 .operation = (char *) "aggregation deferral", 658 .exchange_amount = claimed, 659 .auditor_amount = amount, 660 /* claiming to hold less than it owes understates the exchange's 661 liabilities, which is the direction that favours the exchange */ 662 .profitable = (0 > TALER_amount_cmp (&claimed, 663 &amount)) 664 }; 665 666 qs = TALER_AUDITORDB_insert_amount_arithmetic_inconsistency ( 667 TALER_ARL_adb, 668 &aai); 669 if (0 > qs) 670 { 671 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 672 hc->err = qs; 673 } 674 } 675 676 if ( (TALER_EXCHANGEDB_DR_KYC == reason) && 677 (! open_measure) ) 678 { 679 char *diag; 680 681 /* The one claim the auditor can check on its own: a legitimization 682 requirement is a row in `legitimization_measures' that no 683 `legitimization_processes' row carried through to a decision, and that 684 is replicated. Claiming KYC without one is how an exchange would dress 685 up simply keeping the money. */ 686 GNUNET_asprintf (&diag, 687 "exchange withholds wire transfer %s for legitimization" 688 " measure %llu, which is not open", 689 TALER_B2S (wtid), 690 (unsigned long long) claimed_measure); 691 report_row (hc, 692 "aggregation_deferrals", 693 claimed_measure, 694 diag); 695 GNUNET_free (diag); 696 } 697 698 if (internal_checks) 699 { 700 struct TALER_Amount transient; 701 uint64_t transient_measure; 702 703 /* `aggregation_transient' is the live version of what the deferral row 704 claims: updated in place, deleted on payout, and unreplicable. On an 705 internal audit we can hold the two against each other, which is what 706 catches a `aggregation_deferrals' row that was written once and then 707 left behind by an aggregate that has moved on. */ 708 qs = TALER_EXCHANGEDB_get_aggregation_transient_by_wtid ( 709 TALER_ARL_edb, 710 &h_payto, 711 wtid, 712 &transient, 713 &transient_measure); 714 if (0 > qs) 715 { 716 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 717 hc->err = qs; 718 GNUNET_free (payto.full_payto); 719 return; 720 } 721 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 722 { 723 char *diag; 724 725 /* The exchange marked the deposits as aggregated, did not transfer the 726 money, and is not tracking that it owes it either. */ 727 GNUNET_asprintf (&diag, 728 "wire transfer %s was aggregated but is neither in" 729 " wire_out nor in aggregation_transient", 730 TALER_B2S (wtid)); 731 report_row (hc, 732 "aggregation_transient", 733 0, 734 diag); 735 GNUNET_free (diag); 736 } 737 else 738 { 739 if (0 != 740 TALER_amount_cmp (&transient, 741 &amount)) 742 { 743 struct TALER_AUDITORDB_AmountArithmeticInconsistency aai = { 744 .problem_row_id = 0, 745 .operation = (char *) "aggregation transient", 746 .exchange_amount = transient, 747 .auditor_amount = amount, 748 /* claiming to hold less than it owes understates the exchange's 749 liabilities, which is the direction that favours the exchange */ 750 .profitable = (0 > TALER_amount_cmp (&transient, 751 &amount)) 752 }; 753 754 qs = TALER_AUDITORDB_insert_amount_arithmetic_inconsistency ( 755 TALER_ARL_adb, 756 &aai); 757 if (0 > qs) 758 { 759 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 760 hc->err = qs; 761 } 762 } 763 if ( (TALER_EXCHANGEDB_DR_KYC == reason) != 764 (0 != transient_measure) ) 765 { 766 char *diag; 767 768 GNUNET_asprintf (&diag, 769 "exchange is tracking wire transfer %s as %s but its" 770 " last aggregation_deferrals row says %s", 771 TALER_B2S (wtid), 772 (0 != transient_measure) 773 ? "awaiting legitimization" 774 : "not awaiting legitimization", 775 (TALER_EXCHANGEDB_DR_KYC == reason) 776 ? "it is" 777 : "it is not"); 778 report_row (hc, 779 "aggregation_transient", 780 transient_measure, 781 diag); 782 GNUNET_free (diag); 783 } 784 } 785 } 786 787 qs = TALER_AUDITORDB_insert_aml_hold (TALER_ARL_adb, 788 wtid, 789 &h_payto, 790 payto, 791 &amount, 792 (uint32_t) reason, 793 claimed_measure); 794 GNUNET_free (payto.full_payto); 795 if (0 > qs) 796 { 797 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 798 hc->err = qs; 799 return; 800 } 801 /* Insert-then-update, not "insert if new": the amount grows as further 802 deposits join the transfer, and the reason the exchange gives for holding 803 it can change long after the hold began. */ 804 qs = TALER_AUDITORDB_update_aml_hold (TALER_ARL_adb, 805 wtid, 806 &amount, 807 (uint32_t) reason, 808 claimed_measure); 809 if (0 > qs) 810 { 811 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 812 hc->err = qs; 813 return; 814 } 815 { 816 struct TALER_Amount *bucket; 817 818 switch (reason) 819 { 820 case TALER_EXCHANGEDB_DR_AMOUNT_TOO_SMALL: 821 bucket = &hc->small_aggregate; 822 break; 823 case TALER_EXCHANGEDB_DR_KYC: 824 bucket = &hc->aml_hold; 825 break; 826 case TALER_EXCHANGEDB_DR_NONE: 827 default: 828 bucket = &hc->transfer_lag; 829 break; 830 } 831 TALER_ARL_amount_add (bucket, 832 bucket, 833 &amount); 834 } 835 } 836 837 838 /** 839 * Function called on each row of the aggregation tracking table. 840 * 841 * @param hc closure 842 * @param rowid row of the entry in the aggregation tracking table 843 * @param wtid wire transfer the deposit was aggregated into 844 * @param wire_target_h_payto account the wire transfer should go to 845 * @param pending false if the exchange really did make the transfer 846 * @return #GNUNET_OK to continue to iterate 847 */ 848 static enum GNUNET_GenericReturnValue 849 note_held_transfer_cb ( 850 struct HoldContext *hc, 851 uint64_t rowid, 852 const struct TALER_WireTransferIdentifierRawP *wtid, 853 const struct TALER_FullPaytoHashP *wire_target_h_payto, 854 bool pending) 855 { 856 struct GNUNET_HashCode key; 857 858 (void) wire_target_h_payto; 859 if (0 > hc->err) 860 return GNUNET_SYSERR; /* already failed */ 861 GNUNET_assert (hc->max_aggregation_serial <= rowid); 862 hc->max_aggregation_serial = rowid + 1; 863 if (! pending) 864 { 865 /* Aggregated and paid, the ordinary case. It cannot have a hold on file 866 either: a wtid is only ever wired once, so the rows we are walking here 867 are the ones that created it, and a hold from an earlier run is caught 868 by the pass over the holds instead. */ 869 return GNUNET_OK; 870 } 871 GNUNET_CRYPTO_hash (wtid, 872 sizeof (*wtid), 873 &key); 874 if (GNUNET_OK != 875 GNUNET_CONTAINER_multihashmap_put ( 876 hc->seen, 877 &key, 878 hc->seen, 879 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)) 880 return GNUNET_OK; /* another row of the same transfer, already evaluated */ 881 evaluate_hold (hc, 882 wtid); 883 if (0 > hc->err) 884 return GNUNET_SYSERR; 885 return GNUNET_OK; 886 } 887 888 889 /** 890 * Function called on each hold we already had on file. 891 * 892 * @param hc closure 893 * @param ah the hold 894 * @return #GNUNET_OK to continue to iterate 895 */ 896 static enum GNUNET_GenericReturnValue 897 evaluate_hold_cb (struct HoldContext *hc, 898 const struct TALER_AUDITORDB_AmlHold *ah) 899 { 900 struct GNUNET_HashCode key; 901 902 if (0 > hc->err) 903 return GNUNET_SYSERR; /* already failed */ 904 GNUNET_CRYPTO_hash (&ah->wtid, 905 sizeof (ah->wtid), 906 &key); 907 if (GNUNET_YES == 908 GNUNET_CONTAINER_multihashmap_contains (hc->seen, 909 &key)) 910 return GNUNET_OK; /* fresh rows arrived for it, already re-evaluated */ 911 evaluate_hold (hc, 912 &ah->wtid); 913 if (0 > hc->err) 914 return GNUNET_SYSERR; 915 return GNUNET_OK; 916 } 917 918 919 /** 920 * Checks which wire transfers the exchange aggregated but did not execute, and 921 * what it says is keeping it from making them. 922 * 923 * Unlike the two checks above, the three balances this maintains are current 924 * state and not a running tally: they are recomputed from the holds on file on 925 * every run. That is what keeps them convergent, because a hold changes in 926 * three independent ways -- it grows as deposits are added to the transfer, it 927 * is reclassified when the exchange states a different reason, and it ends when 928 * the transfer is finally made -- and only the first of those is announced by 929 * a new row that a cursor could walk. 930 * 931 * @return transaction status 932 */ 933 static enum GNUNET_DB_QueryStatus 934 check_for_held_transfers (void) 935 { 936 struct HoldContext hc = { 937 .max_aggregation_serial = TALER_ARL_USE_PP (wire_hold_aggregation_id), 938 .err = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 939 }; 940 enum GNUNET_DB_QueryStatus qs; 941 942 GNUNET_assert (GNUNET_OK == 943 TALER_amount_set_zero (TALER_ARL_currency, 944 &hc.aml_hold)); 945 GNUNET_assert (GNUNET_OK == 946 TALER_amount_set_zero (TALER_ARL_currency, 947 &hc.small_aggregate)); 948 GNUNET_assert (GNUNET_OK == 949 TALER_amount_set_zero (TALER_ARL_currency, 950 &hc.transfer_lag)); 951 hc.seen = GNUNET_CONTAINER_multihashmap_create (128, 952 GNUNET_NO); 953 /* First the aggregations we have not seen before, then the holds we already 954 knew about; the map keeps the two from doing each other's work. */ 955 qs = TALER_EXCHANGEDB_iterate_aggregation_wtids_above_serial_id ( 956 TALER_ARL_edb, 957 TALER_ARL_USE_PP (wire_hold_aggregation_id), 958 ¬e_held_transfer_cb, 959 &hc); 960 if (0 > qs) 961 { 962 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 963 goto cleanup; 964 } 965 if (0 > hc.err) 966 { 967 qs = hc.err; 968 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 969 goto cleanup; 970 } 971 qs = TALER_AUDITORDB_iterate_aml_holds (TALER_ARL_adb, 972 &evaluate_hold_cb, 973 &hc); 974 if (0 > qs) 975 { 976 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 977 goto cleanup; 978 } 979 if (0 > hc.err) 980 { 981 qs = hc.err; 982 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 983 goto cleanup; 984 } 985 TALER_ARL_USE_PP (wire_hold_aggregation_id) = hc.max_aggregation_serial; 986 TALER_ARL_USE_AB (total_aml_hold) = hc.aml_hold; 987 TALER_ARL_USE_AB (total_small_aggregate) = hc.small_aggregate; 988 TALER_ARL_USE_AB (total_transfer_lag) = hc.transfer_lag; 989 /* One TALER_amount2s() per statement: it returns a single static buffer. */ 990 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 991 "Exchange is holding %s awaiting KYC\n", 992 TALER_amount2s (&hc.aml_hold)); 993 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 994 "Exchange is holding %s in aggregates below the wire fee\n", 995 TALER_amount2s (&hc.small_aggregate)); 996 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 997 "Exchange is holding %s for no stated reason\n", 998 TALER_amount2s (&hc.transfer_lag)); 999 qs = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 1000 cleanup: 1001 GNUNET_CONTAINER_multihashmap_destroy (hc.seen); 1002 return qs; 1003 } 1004 1005 1006 /** 1007 * Start the database transactions and begin the audit. 1008 * 1009 * @return transaction status 1010 */ 1011 static enum GNUNET_DB_QueryStatus 1012 begin_transaction (void) 1013 { 1014 enum GNUNET_DB_QueryStatus qs; 1015 1016 if (GNUNET_SYSERR == 1017 TALER_EXCHANGEDB_preflight (TALER_ARL_edb)) 1018 { 1019 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1020 "Failed to initialize exchange database connection.\n"); 1021 return GNUNET_DB_STATUS_HARD_ERROR; 1022 } 1023 if (GNUNET_SYSERR == 1024 TALER_AUDITORDB_preflight (TALER_ARL_adb)) 1025 { 1026 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1027 "Failed to initialize auditor database session.\n"); 1028 return GNUNET_DB_STATUS_HARD_ERROR; 1029 } 1030 if (GNUNET_OK != 1031 TALER_AUDITORDB_start (TALER_ARL_adb, 1032 "auditor-transfer")) 1033 { 1034 GNUNET_break (0); 1035 return GNUNET_DB_STATUS_HARD_ERROR; 1036 } 1037 if (GNUNET_OK != 1038 TALER_EXCHANGEDB_start_read_only (TALER_ARL_edb, 1039 "transfer auditor")) 1040 { 1041 GNUNET_break (0); 1042 TALER_AUDITORDB_rollback (TALER_ARL_adb); 1043 return GNUNET_DB_STATUS_HARD_ERROR; 1044 } 1045 qs = TALER_AUDITORDB_get_auditor_progress ( 1046 TALER_ARL_adb, 1047 TALER_ARL_GET_PP (wire_batch_deposit_id), 1048 TALER_ARL_GET_PP (wire_aggregation_id), 1049 TALER_ARL_GET_PP (wire_hold_aggregation_id), 1050 NULL); 1051 if (0 > qs) 1052 goto handle_db_error; 1053 1054 qs = TALER_AUDITORDB_get_balance ( 1055 TALER_ARL_adb, 1056 TALER_ARL_GET_AB (total_amount_lag), 1057 TALER_ARL_GET_AB (total_early_aggregation), 1058 TALER_ARL_GET_AB (total_aml_hold), 1059 TALER_ARL_GET_AB (total_small_aggregate), 1060 TALER_ARL_GET_AB (total_transfer_lag), 1061 NULL); 1062 if (0 > qs) 1063 goto handle_db_error; 1064 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 1065 { 1066 GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE, 1067 "First analysis of with transfer auditor, starting audit from scratch\n"); 1068 } 1069 else 1070 { 1071 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1072 "Resuming transfer audit at %llu / %llu / %llu\n", 1073 (unsigned long long) TALER_ARL_USE_PP (wire_batch_deposit_id), 1074 (unsigned long long) TALER_ARL_USE_PP (wire_aggregation_id), 1075 (unsigned long long) TALER_ARL_USE_PP ( 1076 wire_hold_aggregation_id)); 1077 } 1078 1079 qs = check_for_required_transfers (); 1080 if (0 > qs) 1081 goto handle_db_error; 1082 qs = check_for_completed_transfers (); 1083 if (0 > qs) 1084 goto handle_db_error; 1085 qs = check_for_held_transfers (); 1086 if (0 > qs) 1087 goto handle_db_error; 1088 1089 qs = TALER_AUDITORDB_update_auditor_progress ( 1090 TALER_ARL_adb, 1091 TALER_ARL_SET_PP (wire_batch_deposit_id), 1092 TALER_ARL_SET_PP (wire_aggregation_id), 1093 TALER_ARL_SET_PP (wire_hold_aggregation_id), 1094 NULL); 1095 if (0 > qs) 1096 goto handle_db_error; 1097 qs = TALER_AUDITORDB_insert_auditor_progress ( 1098 TALER_ARL_adb, 1099 TALER_ARL_SET_PP (wire_batch_deposit_id), 1100 TALER_ARL_SET_PP (wire_aggregation_id), 1101 TALER_ARL_SET_PP (wire_hold_aggregation_id), 1102 NULL); 1103 if (0 > qs) 1104 goto handle_db_error; 1105 qs = TALER_AUDITORDB_update_balance ( 1106 TALER_ARL_adb, 1107 TALER_ARL_SET_AB (total_amount_lag), 1108 TALER_ARL_SET_AB (total_early_aggregation), 1109 TALER_ARL_SET_AB (total_aml_hold), 1110 TALER_ARL_SET_AB (total_small_aggregate), 1111 TALER_ARL_SET_AB (total_transfer_lag), 1112 NULL); 1113 if (0 > qs) 1114 goto handle_db_error; 1115 qs = TALER_AUDITORDB_insert_balance ( 1116 TALER_ARL_adb, 1117 TALER_ARL_SET_AB (total_amount_lag), 1118 TALER_ARL_SET_AB (total_early_aggregation), 1119 TALER_ARL_SET_AB (total_aml_hold), 1120 TALER_ARL_SET_AB (total_small_aggregate), 1121 TALER_ARL_SET_AB (total_transfer_lag), 1122 NULL); 1123 if (0 > qs) 1124 goto handle_db_error; 1125 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1126 "Concluded audit step at %llu/%llu/%llu\n", 1127 (unsigned long long) TALER_ARL_USE_PP (wire_aggregation_id), 1128 (unsigned long long) TALER_ARL_USE_PP (wire_batch_deposit_id), 1129 (unsigned long long) TALER_ARL_USE_PP ( 1130 wire_hold_aggregation_id)); 1131 TALER_EXCHANGEDB_rollback (TALER_ARL_edb); 1132 qs = TALER_AUDITORDB_commit (TALER_ARL_adb); 1133 if (0 > qs) 1134 goto handle_db_error; 1135 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1136 "Transaction concluded!\n"); 1137 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 1138 handle_db_error: 1139 TALER_AUDITORDB_rollback (TALER_ARL_adb); 1140 TALER_EXCHANGEDB_rollback (TALER_ARL_edb); 1141 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 1142 return qs; 1143 } 1144 1145 1146 /** 1147 * Start auditor process. 1148 */ 1149 static void 1150 start (void) 1151 { 1152 enum GNUNET_DB_QueryStatus qs; 1153 1154 for (unsigned int max_retries = 3; max_retries>0; max_retries--) 1155 { 1156 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1157 "Trying again (%u attempts left)\n", 1158 max_retries); 1159 qs = begin_transaction (); 1160 if (GNUNET_DB_STATUS_SOFT_ERROR != qs) 1161 break; 1162 } 1163 if (0 > qs) 1164 { 1165 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1166 "Audit failed\n"); 1167 GNUNET_break (0); 1168 global_ret = EXIT_FAILURE; 1169 GNUNET_SCHEDULER_shutdown (); 1170 return; 1171 } 1172 } 1173 1174 1175 /** 1176 * Function called on events received from Postgres. 1177 * 1178 * @param cls closure, NULL 1179 * @param extra additional event data provided 1180 * @param extra_size number of bytes in @a extra 1181 */ 1182 static void 1183 db_notify (void *cls, 1184 const void *extra, 1185 size_t extra_size) 1186 { 1187 (void) cls; 1188 (void) extra; 1189 (void) extra_size; 1190 1191 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1192 "Received notification to wake transfer helper\n"); 1193 start (); 1194 } 1195 1196 1197 /** 1198 * Main function that will be run. 1199 * 1200 * @param cls closure 1201 * @param args remaining command-line arguments 1202 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 1203 * @param c configuration 1204 */ 1205 static void 1206 run (void *cls, 1207 char *const *args, 1208 const char *cfgfile, 1209 const struct GNUNET_CONFIGURATION_Handle *c) 1210 { 1211 (void) cls; 1212 (void) args; 1213 (void) cfgfile; 1214 cfg = c; 1215 if (GNUNET_OK != 1216 TALER_ARL_init (c)) 1217 { 1218 global_ret = EXIT_FAILURE; 1219 return; 1220 } 1221 GNUNET_SCHEDULER_add_shutdown (&do_shutdown, 1222 NULL); 1223 if (GNUNET_OK != 1224 TALER_EXCHANGEDB_load_accounts (TALER_ARL_cfg, 1225 TALER_EXCHANGEDB_ALO_DEBIT 1226 | TALER_EXCHANGEDB_ALO_CREDIT 1227 | TALER_EXCHANGEDB_ALO_AUTHDATA)) 1228 { 1229 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1230 "No bank accounts configured\n"); 1231 global_ret = EXIT_NOTCONFIGURED; 1232 GNUNET_SCHEDULER_shutdown (); 1233 return; 1234 } 1235 if (0 == test_mode) 1236 { 1237 // FIXME-Optimization: use different event type in the future! 1238 struct GNUNET_DB_EventHeaderP es = { 1239 .size = htons (sizeof (es)), 1240 .type = htons (TALER_DBEVENT_EXCHANGE_AUDITOR_WAKE_HELPER_WIRE) 1241 }; 1242 1243 eh = TALER_AUDITORDB_event_listen (TALER_ARL_adb, 1244 &es, 1245 GNUNET_TIME_UNIT_FOREVER_REL, 1246 &db_notify, 1247 NULL); 1248 GNUNET_assert (NULL != eh); 1249 } 1250 start (); 1251 } 1252 1253 1254 /** 1255 * The main function of the wire auditing tool. Checks that 1256 * the exchange's records of wire transfers match that of 1257 * the wire gateway. 1258 * 1259 * @param argc number of arguments from the command line 1260 * @param argv command line arguments 1261 * @return 0 ok, 1 on error 1262 */ 1263 int 1264 main (int argc, 1265 char *const *argv) 1266 { 1267 const struct GNUNET_GETOPT_CommandLineOption options[] = { 1268 GNUNET_GETOPT_option_flag ('i', 1269 "internal", 1270 "perform checks only applicable for exchange-internal audits", 1271 &internal_checks), 1272 GNUNET_GETOPT_option_flag ('t', 1273 "test", 1274 "run in test mode and exit when idle", 1275 &test_mode), 1276 GNUNET_GETOPT_option_timetravel ('T', 1277 "timetravel"), 1278 GNUNET_GETOPT_OPTION_END 1279 }; 1280 enum GNUNET_GenericReturnValue ret; 1281 1282 ret = GNUNET_PROGRAM_run ( 1283 TALER_AUDITOR_project_data (), 1284 argc, 1285 argv, 1286 "taler-helper-auditor-transfer", 1287 gettext_noop ( 1288 "Audit exchange database for consistency of aggregations/transfers with respect to deposit deadlines"), 1289 options, 1290 &run, 1291 NULL); 1292 if (GNUNET_SYSERR == ret) 1293 return EXIT_INVALIDARGUMENT; 1294 if (GNUNET_NO == ret) 1295 return EXIT_SUCCESS; 1296 return global_ret; 1297 } 1298 1299 1300 /* end of taler-helper-auditor-transfer.c */