taler-helper-auditor-transfer.c (40534B)
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 "exchange-database/get_aggregation_deferral_by_wtid.h" 65 #include "exchange-database/get_aggregation_transient_by_wtid.h" 66 #include "exchange-database/get_open_legitimization_measure.h" 67 #include "exchange-database/get_pending_aggregation.h" 68 #include "exchange-database/preflight.h" 69 #include "exchange-database/rollback.h" 70 struct AggregationContext; 71 #define TALER_EXCHANGEDB_AGGREGATION_RESULT_CLOSURE struct AggregationContext 72 #include "exchange-database/iterate_aggregations_above_serial_id.h" 73 struct ImportMissingWireContext; 74 #define TALER_EXCHANGEDB_WIRE_MISSING_RESULT_CLOSURE struct \ 75 ImportMissingWireContext 76 #include "exchange-database/iterate_batch_deposits_missing_wire.h" 77 struct HoldContext; 78 #define TALER_EXCHANGEDB_AGGREGATION_WTID_RESULT_CLOSURE struct HoldContext 79 #include "exchange-database/iterate_aggregation_wtids_above_serial_id.h" 80 #define TALER_AUDITORDB_AML_HOLD_RESULT_CLOSURE struct HoldContext 81 #include "auditor-database/iterate_aml_holds.h" 82 #include "exchange-database/start_read_only.h" 83 84 85 /** 86 * Run in test mode. Exit when idle instead of 87 * going to sleep and waiting for more work. 88 */ 89 static int test_mode; 90 91 /** 92 * Return value from main(). 93 */ 94 static int global_ret; 95 96 /** 97 * Last reserve_out / wire_out serial IDs seen. 98 */ 99 static TALER_ARL_DEF_PP (wire_batch_deposit_id); 100 static TALER_ARL_DEF_PP (wire_aggregation_id); 101 102 /** 103 * Row of `aggregation_tracking` up to which we have looked for wire transfers 104 * the exchange aggregated but did not execute. 105 */ 106 static TALER_ARL_DEF_PP (wire_hold_aggregation_id); 107 108 /** 109 * Total amount which the exchange did not aggregate/transfer in time. 110 */ 111 static TALER_ARL_DEF_AB (total_amount_lag); 112 113 /** 114 * Total amount which the exchange did aggregate/transfer too early. 115 */ 116 static TALER_ARL_DEF_AB (total_early_aggregation); 117 118 /** 119 * Total amount the exchange aggregated but did not wire because it says a 120 * legitimization requirement against the recipient is still open. Money the 121 * exchange is withholding for legal reasons. 122 */ 123 static TALER_ARL_DEF_AB (total_aml_hold); 124 125 /** 126 * Total amount the exchange aggregated but did not wire because what it has 127 * collected for the account so far does not cover the wire fee. Money that is 128 * waiting for the next deposit rather than for anyone's decision, and thus not 129 * an irregularity at all -- it is kept apart so that it does not inflate the 130 * two balances that are. 131 */ 132 static TALER_ARL_DEF_AB (total_small_aggregate); 133 134 /** 135 * Total amount the exchange aggregated but did not wire and gave no reason 136 * for. Money the exchange simply has not paid out. 137 */ 138 static TALER_ARL_DEF_AB (total_transfer_lag); 139 140 /** 141 * Should we run checks that only work for exchange-internal audits? 142 */ 143 static int internal_checks; 144 145 /** 146 * Database event handler to wake us up again. 147 */ 148 static struct GNUNET_DB_EventHandler *eh; 149 150 /** 151 * The auditors's configuration. 152 */ 153 static const struct GNUNET_CONFIGURATION_Handle *cfg; 154 155 156 /** 157 * Task run on shutdown. 158 * 159 * @param cls NULL 160 */ 161 static void 162 do_shutdown (void *cls) 163 { 164 (void) cls; 165 if (NULL != eh) 166 { 167 TALER_AUDITORDB_event_listen_cancel (eh); 168 eh = NULL; 169 } 170 TALER_ARL_done (); 171 TALER_EXCHANGEDB_unload_accounts (); 172 TALER_ARL_cfg = NULL; 173 } 174 175 176 /** 177 * Closure for import_wire_missing_cb(). 178 */ 179 struct ImportMissingWireContext 180 { 181 /** 182 * Set to maximum row ID encountered. 183 */ 184 uint64_t max_batch_deposit_uuid; 185 186 /** 187 * Set to database errors in callback. 188 */ 189 enum GNUNET_DB_QueryStatus err; 190 }; 191 192 193 /** 194 * Function called on deposits that need to be checked for their 195 * wire transfer. 196 * 197 * @param wc closure, points to a `struct ImportMissingWireContext` 198 * @param batch_deposit_serial_id serial of the entry in the batch deposits table 199 * @param total_amount value of the missing deposits, including fee 200 * @param wire_target_h_payto where should the funds be wired 201 * @param deadline what was the earliest requested wire transfer deadline 202 */ 203 static void 204 import_wire_missing_cb ( 205 struct ImportMissingWireContext *wc, 206 uint64_t batch_deposit_serial_id, 207 const struct TALER_Amount *total_amount, 208 const struct TALER_FullPaytoHashP *wire_target_h_payto, 209 struct GNUNET_TIME_Timestamp deadline) 210 { 211 enum GNUNET_DB_QueryStatus qs; 212 213 if (wc->err < 0) 214 return; /* already failed */ 215 GNUNET_assert (batch_deposit_serial_id >= wc->max_batch_deposit_uuid); 216 wc->max_batch_deposit_uuid = batch_deposit_serial_id + 1; 217 qs = TALER_AUDITORDB_delete_early_aggregation ( 218 TALER_ARL_adb, 219 batch_deposit_serial_id); 220 switch (qs) 221 { 222 case GNUNET_DB_STATUS_SOFT_ERROR: 223 GNUNET_break (0); 224 wc->err = qs; 225 return; 226 case GNUNET_DB_STATUS_HARD_ERROR: 227 GNUNET_break (0); 228 wc->err = qs; 229 return; 230 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 231 qs = TALER_AUDITORDB_insert_pending_deposit ( 232 TALER_ARL_adb, 233 batch_deposit_serial_id, 234 wire_target_h_payto, 235 total_amount, 236 deadline); 237 if (0 > qs) 238 { 239 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 240 wc->err = qs; 241 return; 242 } 243 if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs) 244 TALER_ARL_amount_add (&TALER_ARL_USE_AB (total_amount_lag), 245 &TALER_ARL_USE_AB (total_amount_lag), 246 total_amount); 247 break; 248 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 249 TALER_ARL_amount_subtract (&TALER_ARL_USE_AB (total_early_aggregation), 250 &TALER_ARL_USE_AB (total_early_aggregation), 251 total_amount); 252 break; 253 default: 254 GNUNET_assert (0); 255 } 256 } 257 258 259 /** 260 * Checks for wire transfers that should have happened. 261 * 262 * @return transaction status 263 */ 264 static enum GNUNET_DB_QueryStatus 265 check_for_required_transfers (void) 266 { 267 enum GNUNET_DB_QueryStatus qs; 268 struct ImportMissingWireContext wc = { 269 .max_batch_deposit_uuid = TALER_ARL_USE_PP (wire_batch_deposit_id), 270 .err = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 271 }; 272 273 qs = TALER_EXCHANGEDB_iterate_batch_deposits_missing_wire ( 274 TALER_ARL_edb, 275 TALER_ARL_USE_PP (wire_batch_deposit_id), 276 &import_wire_missing_cb, 277 &wc); 278 if (0 > qs) 279 { 280 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 281 return qs; 282 } 283 if (0 > wc.err) 284 { 285 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == wc.err); 286 return wc.err; 287 } 288 TALER_ARL_USE_PP (wire_batch_deposit_id) = wc.max_batch_deposit_uuid; 289 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 290 } 291 292 293 /** 294 * Closure for #clear_finished_transfer_cb(). 295 */ 296 struct AggregationContext 297 { 298 /** 299 * Set to maximum row ID encountered. 300 */ 301 uint64_t max_aggregation_serial; 302 303 /** 304 * Set to database errors in callback. 305 */ 306 enum GNUNET_DB_QueryStatus err; 307 }; 308 309 310 /** 311 * Function called on aggregations that were done for 312 * a (batch) deposit. 313 * 314 * @param ac closure 315 * @param amount affected amount 316 * @param tracking_serial_id where in the table are we 317 * @param batch_deposit_serial_id which batch deposit was aggregated 318 */ 319 static void 320 clear_finished_transfer_cb ( 321 struct AggregationContext *ac, 322 const struct TALER_Amount *amount, 323 uint64_t tracking_serial_id, 324 uint64_t batch_deposit_serial_id) 325 { 326 enum GNUNET_DB_QueryStatus qs; 327 328 if (0 > ac->err) 329 return; /* already failed */ 330 GNUNET_assert (ac->max_aggregation_serial <= tracking_serial_id); 331 ac->max_aggregation_serial = tracking_serial_id + 1; 332 qs = TALER_AUDITORDB_delete_pending_deposit ( 333 TALER_ARL_adb, 334 batch_deposit_serial_id); 335 switch (qs) 336 { 337 case GNUNET_DB_STATUS_SOFT_ERROR: 338 GNUNET_break (0); 339 ac->err = qs; 340 return; 341 case GNUNET_DB_STATUS_HARD_ERROR: 342 GNUNET_break (0); 343 ac->err = qs; 344 return; 345 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 346 qs = TALER_AUDITORDB_insert_early_aggregation ( 347 TALER_ARL_adb, 348 batch_deposit_serial_id, 349 tracking_serial_id, 350 amount); 351 if (0 > qs) 352 { 353 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 354 ac->err = qs; 355 return; 356 } 357 if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs) 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 .problem_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 ? 1 665 : -1 666 }; 667 668 qs = TALER_AUDITORDB_insert_amount_arithmetic_inconsistency ( 669 TALER_ARL_adb, 670 &aai); 671 if (0 > qs) 672 { 673 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 674 hc->err = qs; 675 } 676 } 677 678 if ( (TALER_EXCHANGEDB_DR_KYC == reason) && 679 (! open_measure) ) 680 { 681 char *diag; 682 683 /* The one claim the auditor can check on its own: a legitimization 684 requirement is a row in `legitimization_measures' that no 685 `legitimization_processes' row carried through to a decision, and that 686 is replicated. Claiming KYC without one is how an exchange would dress 687 up simply keeping the money. */ 688 GNUNET_asprintf (&diag, 689 "exchange withholds wire transfer %s for legitimization" 690 " measure %llu, which is not open", 691 TALER_B2S (wtid), 692 (unsigned long long) claimed_measure); 693 report_row (hc, 694 "aggregation_deferrals", 695 claimed_measure, 696 diag); 697 GNUNET_free (diag); 698 } 699 700 if (internal_checks) 701 { 702 struct TALER_Amount transient; 703 uint64_t transient_measure; 704 705 /* `aggregation_transient' is the live version of what the deferral row 706 claims: updated in place, deleted on payout, and unreplicable. On an 707 internal audit we can hold the two against each other, which is what 708 catches a `aggregation_deferrals' row that was written once and then 709 left behind by an aggregate that has moved on. */ 710 qs = TALER_EXCHANGEDB_get_aggregation_transient_by_wtid ( 711 TALER_ARL_edb, 712 &h_payto, 713 wtid, 714 &transient, 715 &transient_measure); 716 if (0 > qs) 717 { 718 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 719 hc->err = qs; 720 GNUNET_free (payto.full_payto); 721 return; 722 } 723 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 724 { 725 char *diag; 726 727 /* The exchange marked the deposits as aggregated, did not transfer the 728 money, and is not tracking that it owes it either. */ 729 GNUNET_asprintf (&diag, 730 "wire transfer %s was aggregated but is neither in" 731 " wire_out nor in aggregation_transient", 732 TALER_B2S (wtid)); 733 report_row (hc, 734 "aggregation_transient", 735 0, 736 diag); 737 GNUNET_free (diag); 738 } 739 else 740 { 741 if (0 != 742 TALER_amount_cmp (&transient, 743 &amount)) 744 { 745 struct TALER_AUDITORDB_AmountArithmeticInconsistency aai = { 746 .problem_row_id = 0, 747 .operation = (char *) "aggregation transient", 748 .exchange_amount = transient, 749 .auditor_amount = amount, 750 /* claiming to hold less than it owes understates the exchange's 751 liabilities, which is the direction that favours the exchange */ 752 .profitable = (0 > TALER_amount_cmp (&transient, 753 &amount)) 754 ? 1 755 : -1 756 }; 757 758 qs = TALER_AUDITORDB_insert_amount_arithmetic_inconsistency ( 759 TALER_ARL_adb, 760 &aai); 761 if (0 > qs) 762 { 763 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 764 hc->err = qs; 765 } 766 } 767 if ( (TALER_EXCHANGEDB_DR_KYC == reason) != 768 (0 != transient_measure) ) 769 { 770 char *diag; 771 772 GNUNET_asprintf (&diag, 773 "exchange is tracking wire transfer %s as %s but its" 774 " last aggregation_deferrals row says %s", 775 TALER_B2S (wtid), 776 (0 != transient_measure) 777 ? "awaiting legitimization" 778 : "not awaiting legitimization", 779 (TALER_EXCHANGEDB_DR_KYC == reason) 780 ? "it is" 781 : "it is not"); 782 report_row (hc, 783 "aggregation_transient", 784 transient_measure, 785 diag); 786 GNUNET_free (diag); 787 } 788 } 789 } 790 791 qs = TALER_AUDITORDB_insert_aml_hold (TALER_ARL_adb, 792 wtid, 793 &h_payto, 794 payto, 795 &amount, 796 (uint32_t) reason, 797 claimed_measure); 798 GNUNET_free (payto.full_payto); 799 if (0 > qs) 800 { 801 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 802 hc->err = qs; 803 return; 804 } 805 /* Insert-then-update, not "insert if new": the amount grows as further 806 deposits join the transfer, and the reason the exchange gives for holding 807 it can change long after the hold began. */ 808 qs = TALER_AUDITORDB_update_aml_hold (TALER_ARL_adb, 809 wtid, 810 &amount, 811 (uint32_t) reason, 812 claimed_measure); 813 if (0 > qs) 814 { 815 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 816 hc->err = qs; 817 return; 818 } 819 { 820 struct TALER_Amount *bucket; 821 822 switch (reason) 823 { 824 case TALER_EXCHANGEDB_DR_AMOUNT_TOO_SMALL: 825 bucket = &hc->small_aggregate; 826 break; 827 case TALER_EXCHANGEDB_DR_KYC: 828 bucket = &hc->aml_hold; 829 break; 830 case TALER_EXCHANGEDB_DR_NONE: 831 default: 832 bucket = &hc->transfer_lag; 833 break; 834 } 835 TALER_ARL_amount_add (bucket, 836 bucket, 837 &amount); 838 } 839 } 840 841 842 /** 843 * Function called on each row of the aggregation tracking table. 844 * 845 * @param hc closure 846 * @param rowid row of the entry in the aggregation tracking table 847 * @param wtid wire transfer the deposit was aggregated into 848 * @param wire_target_h_payto account the wire transfer should go to 849 * @param pending false if the exchange really did make the transfer 850 * @return #GNUNET_OK to continue to iterate 851 */ 852 static enum GNUNET_GenericReturnValue 853 note_held_transfer_cb ( 854 struct HoldContext *hc, 855 uint64_t rowid, 856 const struct TALER_WireTransferIdentifierRawP *wtid, 857 const struct TALER_FullPaytoHashP *wire_target_h_payto, 858 bool pending) 859 { 860 struct GNUNET_HashCode key; 861 862 (void) wire_target_h_payto; 863 if (0 > hc->err) 864 return GNUNET_SYSERR; /* already failed */ 865 GNUNET_assert (hc->max_aggregation_serial <= rowid); 866 hc->max_aggregation_serial = rowid + 1; 867 if (! pending) 868 { 869 /* Aggregated and paid, the ordinary case. It cannot have a hold on file 870 either: a wtid is only ever wired once, so the rows we are walking here 871 are the ones that created it, and a hold from an earlier run is caught 872 by the pass over the holds instead. */ 873 return GNUNET_OK; 874 } 875 GNUNET_CRYPTO_hash (wtid, 876 sizeof (*wtid), 877 &key); 878 if (GNUNET_OK != 879 GNUNET_CONTAINER_multihashmap_put ( 880 hc->seen, 881 &key, 882 hc->seen, 883 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)) 884 return GNUNET_OK; /* another row of the same transfer, already evaluated */ 885 evaluate_hold (hc, 886 wtid); 887 if (0 > hc->err) 888 return GNUNET_SYSERR; 889 return GNUNET_OK; 890 } 891 892 893 /** 894 * Function called on each hold we already had on file. 895 * 896 * @param hc closure 897 * @param ah the hold 898 * @return #GNUNET_OK to continue to iterate 899 */ 900 static enum GNUNET_GenericReturnValue 901 evaluate_hold_cb (struct HoldContext *hc, 902 const struct TALER_AUDITORDB_AmlHold *ah) 903 { 904 struct GNUNET_HashCode key; 905 906 if (0 > hc->err) 907 return GNUNET_SYSERR; /* already failed */ 908 GNUNET_CRYPTO_hash (&ah->wtid, 909 sizeof (ah->wtid), 910 &key); 911 if (GNUNET_YES == 912 GNUNET_CONTAINER_multihashmap_contains (hc->seen, 913 &key)) 914 return GNUNET_OK; /* fresh rows arrived for it, already re-evaluated */ 915 evaluate_hold (hc, 916 &ah->wtid); 917 if (0 > hc->err) 918 return GNUNET_SYSERR; 919 return GNUNET_OK; 920 } 921 922 923 /** 924 * Checks which wire transfers the exchange aggregated but did not execute, and 925 * what it says is keeping it from making them. 926 * 927 * Unlike the two checks above, the three balances this maintains are current 928 * state and not a running tally: they are recomputed from the holds on file on 929 * every run. That is what keeps them convergent, because a hold changes in 930 * three independent ways -- it grows as deposits are added to the transfer, it 931 * is reclassified when the exchange states a different reason, and it ends when 932 * the transfer is finally made -- and only the first of those is announced by 933 * a new row that a cursor could walk. 934 * 935 * @return transaction status 936 */ 937 static enum GNUNET_DB_QueryStatus 938 check_for_held_transfers (void) 939 { 940 struct HoldContext hc = { 941 .max_aggregation_serial = TALER_ARL_USE_PP (wire_hold_aggregation_id), 942 .err = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 943 }; 944 enum GNUNET_DB_QueryStatus qs; 945 946 GNUNET_assert (GNUNET_OK == 947 TALER_amount_set_zero (TALER_ARL_currency, 948 &hc.aml_hold)); 949 GNUNET_assert (GNUNET_OK == 950 TALER_amount_set_zero (TALER_ARL_currency, 951 &hc.small_aggregate)); 952 GNUNET_assert (GNUNET_OK == 953 TALER_amount_set_zero (TALER_ARL_currency, 954 &hc.transfer_lag)); 955 hc.seen = GNUNET_CONTAINER_multihashmap_create (128, 956 GNUNET_NO); 957 /* First the aggregations we have not seen before, then the holds we already 958 knew about; the map keeps the two from doing each other's work. */ 959 qs = TALER_EXCHANGEDB_iterate_aggregation_wtids_above_serial_id ( 960 TALER_ARL_edb, 961 TALER_ARL_USE_PP (wire_hold_aggregation_id), 962 ¬e_held_transfer_cb, 963 &hc); 964 if (0 > qs) 965 { 966 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 967 goto cleanup; 968 } 969 if (0 > hc.err) 970 { 971 qs = hc.err; 972 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 973 goto cleanup; 974 } 975 qs = TALER_AUDITORDB_iterate_aml_holds (TALER_ARL_adb, 976 &evaluate_hold_cb, 977 &hc); 978 if (0 > qs) 979 { 980 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 981 goto cleanup; 982 } 983 if (0 > hc.err) 984 { 985 qs = hc.err; 986 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 987 goto cleanup; 988 } 989 TALER_ARL_USE_PP (wire_hold_aggregation_id) = hc.max_aggregation_serial; 990 TALER_ARL_USE_AB (total_aml_hold) = hc.aml_hold; 991 TALER_ARL_USE_AB (total_small_aggregate) = hc.small_aggregate; 992 TALER_ARL_USE_AB (total_transfer_lag) = hc.transfer_lag; 993 /* One TALER_amount2s() per statement: it returns a single static buffer. */ 994 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 995 "Exchange is holding %s awaiting KYC\n", 996 TALER_amount2s (&hc.aml_hold)); 997 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 998 "Exchange is holding %s in aggregates below the wire fee\n", 999 TALER_amount2s (&hc.small_aggregate)); 1000 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1001 "Exchange is holding %s for no stated reason\n", 1002 TALER_amount2s (&hc.transfer_lag)); 1003 qs = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 1004 cleanup: 1005 GNUNET_CONTAINER_multihashmap_destroy (hc.seen); 1006 return qs; 1007 } 1008 1009 1010 /** 1011 * Start the database transactions and begin the audit. 1012 * 1013 * @return transaction status 1014 */ 1015 static enum GNUNET_DB_QueryStatus 1016 begin_transaction (void) 1017 { 1018 enum GNUNET_DB_QueryStatus qs; 1019 1020 if (GNUNET_SYSERR == 1021 TALER_EXCHANGEDB_preflight (TALER_ARL_edb)) 1022 { 1023 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1024 "Failed to initialize exchange database connection.\n"); 1025 return GNUNET_DB_STATUS_HARD_ERROR; 1026 } 1027 if (GNUNET_SYSERR == 1028 TALER_AUDITORDB_preflight (TALER_ARL_adb)) 1029 { 1030 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1031 "Failed to initialize auditor database session.\n"); 1032 return GNUNET_DB_STATUS_HARD_ERROR; 1033 } 1034 if (GNUNET_OK != 1035 TALER_AUDITORDB_start (TALER_ARL_adb, 1036 "auditor-transfer")) 1037 { 1038 GNUNET_break (0); 1039 return GNUNET_DB_STATUS_HARD_ERROR; 1040 } 1041 if (GNUNET_OK != 1042 TALER_EXCHANGEDB_start_read_only (TALER_ARL_edb, 1043 "transfer auditor")) 1044 { 1045 GNUNET_break (0); 1046 TALER_AUDITORDB_rollback (TALER_ARL_adb); 1047 return GNUNET_DB_STATUS_HARD_ERROR; 1048 } 1049 qs = TALER_AUDITORDB_get_auditor_progress ( 1050 TALER_ARL_adb, 1051 TALER_ARL_GET_PP (wire_batch_deposit_id), 1052 TALER_ARL_GET_PP (wire_aggregation_id), 1053 TALER_ARL_GET_PP (wire_hold_aggregation_id), 1054 NULL); 1055 if (0 > qs) 1056 goto handle_db_error; 1057 1058 qs = TALER_AUDITORDB_get_balance ( 1059 TALER_ARL_adb, 1060 TALER_ARL_GET_AB (total_amount_lag), 1061 TALER_ARL_GET_AB (total_early_aggregation), 1062 TALER_ARL_GET_AB (total_aml_hold), 1063 TALER_ARL_GET_AB (total_small_aggregate), 1064 TALER_ARL_GET_AB (total_transfer_lag), 1065 NULL); 1066 if (0 > qs) 1067 goto handle_db_error; 1068 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 1069 { 1070 GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE, 1071 "First analysis of with transfer auditor, starting audit from scratch\n"); 1072 } 1073 else 1074 { 1075 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1076 "Resuming transfer audit at %llu / %llu / %llu\n", 1077 (unsigned long long) TALER_ARL_USE_PP (wire_batch_deposit_id), 1078 (unsigned long long) TALER_ARL_USE_PP (wire_aggregation_id), 1079 (unsigned long long) TALER_ARL_USE_PP ( 1080 wire_hold_aggregation_id)); 1081 } 1082 1083 qs = check_for_required_transfers (); 1084 if (0 > qs) 1085 goto handle_db_error; 1086 qs = check_for_completed_transfers (); 1087 if (0 > qs) 1088 goto handle_db_error; 1089 qs = check_for_held_transfers (); 1090 if (0 > qs) 1091 goto handle_db_error; 1092 1093 qs = TALER_AUDITORDB_insert_auditor_progress ( 1094 TALER_ARL_adb, 1095 TALER_ARL_SET_PP (wire_batch_deposit_id), 1096 TALER_ARL_SET_PP (wire_aggregation_id), 1097 TALER_ARL_SET_PP (wire_hold_aggregation_id), 1098 NULL); 1099 if (0 > qs) 1100 goto handle_db_error; 1101 qs = TALER_AUDITORDB_insert_balance ( 1102 TALER_ARL_adb, 1103 TALER_ARL_SET_AB (total_amount_lag), 1104 TALER_ARL_SET_AB (total_early_aggregation), 1105 TALER_ARL_SET_AB (total_aml_hold), 1106 TALER_ARL_SET_AB (total_small_aggregate), 1107 TALER_ARL_SET_AB (total_transfer_lag), 1108 NULL); 1109 if (0 > qs) 1110 goto handle_db_error; 1111 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1112 "Concluded audit step at %llu/%llu/%llu\n", 1113 (unsigned long long) TALER_ARL_USE_PP (wire_aggregation_id), 1114 (unsigned long long) TALER_ARL_USE_PP (wire_batch_deposit_id), 1115 (unsigned long long) TALER_ARL_USE_PP ( 1116 wire_hold_aggregation_id)); 1117 TALER_EXCHANGEDB_rollback (TALER_ARL_edb); 1118 qs = TALER_AUDITORDB_commit (TALER_ARL_adb); 1119 if (0 > qs) 1120 goto handle_db_error; 1121 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1122 "Transaction concluded!\n"); 1123 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 1124 handle_db_error: 1125 TALER_AUDITORDB_rollback (TALER_ARL_adb); 1126 TALER_EXCHANGEDB_rollback (TALER_ARL_edb); 1127 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 1128 return qs; 1129 } 1130 1131 1132 /** 1133 * Start auditor process. 1134 */ 1135 static void 1136 start (void) 1137 { 1138 enum GNUNET_DB_QueryStatus qs; 1139 1140 for (unsigned int max_retries = 3; max_retries>0; max_retries--) 1141 { 1142 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1143 "Trying again (%u attempts left)\n", 1144 max_retries); 1145 qs = begin_transaction (); 1146 if (GNUNET_DB_STATUS_SOFT_ERROR != qs) 1147 break; 1148 } 1149 if (0 > qs) 1150 { 1151 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1152 "Audit failed\n"); 1153 GNUNET_break (0); 1154 global_ret = EXIT_FAILURE; 1155 GNUNET_SCHEDULER_shutdown (); 1156 return; 1157 } 1158 } 1159 1160 1161 /** 1162 * Function called on events received from Postgres. 1163 * 1164 * @param cls closure, NULL 1165 * @param extra additional event data provided 1166 * @param extra_size number of bytes in @a extra 1167 */ 1168 static void 1169 db_notify (void *cls, 1170 const void *extra, 1171 size_t extra_size) 1172 { 1173 (void) cls; 1174 (void) extra; 1175 (void) extra_size; 1176 1177 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1178 "Received notification to wake transfer helper\n"); 1179 start (); 1180 } 1181 1182 1183 /** 1184 * Main function that will be run. 1185 * 1186 * @param cls closure 1187 * @param args remaining command-line arguments 1188 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 1189 * @param c configuration 1190 */ 1191 static void 1192 run (void *cls, 1193 char *const *args, 1194 const char *cfgfile, 1195 const struct GNUNET_CONFIGURATION_Handle *c) 1196 { 1197 (void) cls; 1198 (void) args; 1199 (void) cfgfile; 1200 cfg = c; 1201 if (EXIT_SUCCESS != 1202 (global_ret = TALER_ARL_init (c))) 1203 { 1204 return; 1205 } 1206 GNUNET_SCHEDULER_add_shutdown (&do_shutdown, 1207 NULL); 1208 if (GNUNET_OK != 1209 TALER_EXCHANGEDB_load_accounts (TALER_ARL_cfg, 1210 TALER_EXCHANGEDB_ALO_DEBIT 1211 | TALER_EXCHANGEDB_ALO_CREDIT 1212 | TALER_EXCHANGEDB_ALO_AUTHDATA)) 1213 { 1214 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1215 "No bank accounts configured\n"); 1216 global_ret = EXIT_NOTCONFIGURED; 1217 GNUNET_SCHEDULER_shutdown (); 1218 return; 1219 } 1220 if (0 == test_mode) 1221 { 1222 // FIXME-Optimization: use different event type in the future! 1223 struct GNUNET_DB_EventHeaderP es = { 1224 .size = htons (sizeof (es)), 1225 .type = htons (TALER_DBEVENT_EXCHANGE_AUDITOR_WAKE_HELPER_WIRE) 1226 }; 1227 1228 eh = TALER_AUDITORDB_event_listen (TALER_ARL_adb, 1229 &es, 1230 GNUNET_TIME_UNIT_FOREVER_REL, 1231 &db_notify, 1232 NULL); 1233 GNUNET_assert (NULL != eh); 1234 } 1235 start (); 1236 } 1237 1238 1239 /** 1240 * The main function of the wire auditing tool. Checks that 1241 * the exchange's records of wire transfers match that of 1242 * the wire gateway. 1243 * 1244 * @param argc number of arguments from the command line 1245 * @param argv command line arguments 1246 * @return 0 ok, 1 on error 1247 */ 1248 int 1249 main (int argc, 1250 char *const *argv) 1251 { 1252 const struct GNUNET_GETOPT_CommandLineOption options[] = { 1253 GNUNET_GETOPT_option_flag ('i', 1254 "internal", 1255 "perform checks only applicable for exchange-internal audits", 1256 &internal_checks), 1257 GNUNET_GETOPT_option_flag ('t', 1258 "test", 1259 "run in test mode and exit when idle", 1260 &test_mode), 1261 GNUNET_GETOPT_option_timetravel ('T', 1262 "timetravel"), 1263 GNUNET_GETOPT_OPTION_END 1264 }; 1265 enum GNUNET_GenericReturnValue ret; 1266 1267 ret = GNUNET_PROGRAM_run ( 1268 TALER_AUDITOR_project_data (), 1269 argc, 1270 argv, 1271 "taler-helper-auditor-transfer", 1272 gettext_noop ( 1273 "Audit exchange database for consistency of aggregations/transfers with respect to deposit deadlines"), 1274 options, 1275 &run, 1276 NULL); 1277 if (GNUNET_SYSERR == ret) 1278 return EXIT_NOTCONFIGURED; 1279 if (GNUNET_NO == ret) 1280 return EXIT_SUCCESS; 1281 return global_ret; 1282 } 1283 1284 1285 /* end of taler-helper-auditor-transfer.c */