-more work on AML triggers for P2P transfers

9 files changed, 337 insertions, 32 deletions

diff --git a/src/exchange/taler-exchange-httpd.c b/src/exchange/taler-exchange-httpd.c
index 62bd9a9dc..0c5d36e0f 100644
--- a/src/exchange/taler-exchange-httpd.c
+++ b/src/exchange/taler-exchange-httpd.c
@@ -148,6 +148,13 @@ struct TALER_EXCHANGEDB_Plugin *TEH_plugin;
 char *TEH_currency;
 
 /**
+ * What is the largest amount we allow a peer to
+ * merge into a reserve before always triggering
+ * an AML check?
+ */
+struct TALER_Amount TEH_aml_threshold;
+
+/**
  * Our base URL.
  */
 char *TEH_base_url;
@@ -1861,6 +1868,16 @@ exchange_serve_process_config (void)
     return GNUNET_SYSERR;
   }
   if (GNUNET_OK !=
+      TALER_config_get_amount (TEH_cfg,
+                               "taler",
+                               "AML_THRESHOLD",
+                               &TEH_aml_threshold))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "Need amount in section `TALER' under `AML_THRESHOLD'\n");
+    return GNUNET_SYSERR;
+  }
+  if (GNUNET_OK !=
       GNUNET_CONFIGURATION_get_value_string (TEH_cfg,
                                              "exchange",
                                              "BASE_URL",
diff --git a/src/exchange/taler-exchange-httpd.h b/src/exchange/taler-exchange-httpd.h
index 0c2bd9e81..960036265 100644
--- a/src/exchange/taler-exchange-httpd.h
+++ b/src/exchange/taler-exchange-httpd.h
@@ -98,6 +98,13 @@ extern struct TALER_EXCHANGEDB_Plugin *TEH_plugin;
 extern char *TEH_currency;
 
 /**
+ * What is the largest amount we allow a peer to
+ * merge into a reserve before always triggering
+ * an AML check?
+ */
+extern struct TALER_Amount TEH_aml_threshold;
+
+/**
  * Our (externally visible) base URL.
  */
 extern char *TEH_base_url;
diff --git a/src/exchange/taler-exchange-httpd_batch-withdraw.c b/src/exchange/taler-exchange-httpd_batch-withdraw.c
index b2f35b568..e6be54a58 100644
--- a/src/exchange/taler-exchange-httpd_batch-withdraw.c
+++ b/src/exchange/taler-exchange-httpd_batch-withdraw.c
@@ -1,6 +1,6 @@
 /*
   This file is part of TALER
-  Copyright (C) 2014-2022 Taler Systems SA
+  Copyright (C) 2014-2023 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as
@@ -108,6 +108,11 @@ struct BatchWithdrawContext
    */
   unsigned int planchets_length;
 
+  /**
+   * AML decision, #TALER_AML_NORMAL if we may proceed.
+   */
+  enum TALER_AmlDecisionState aml_decision;
+
 };
 
 
@@ -151,6 +156,34 @@ batch_withdraw_amount_cb (void *cls,
 
 
 /**
+ * Function called on each @a amount that was found to
+ * be relevant for the AML check as it was merged into
+ * the reserve.
+ *
+ * @param cls `struct TALER_Amount *` to total up the amounts
+ * @param amount encountered transaction amount
+ * @param date when was the amount encountered
+ * @return #GNUNET_OK to continue to iterate,
+ *         #GNUNET_NO to abort iteration
+ *         #GNUNET_SYSERR on internal error (also abort itaration)
+ */
+static enum GNUNET_GenericReturnValue
+aml_amount_cb (
+  void *cls,
+  const struct TALER_Amount *amount,
+  struct GNUNET_TIME_Absolute date)
+{
+  struct TALER_Amount *total = cls;
+
+  GNUNET_assert (0 <=
+                 TALER_amount_add (total,
+                                   total,
+                                   amount));
+  return GNUNET_OK;
+}
+
+
+/**
  * Function implementing withdraw transaction.  Runs the
  * transaction logic; IF it returns a non-error code, the transaction
  * logic MUST NOT queue a MHD response.  IF it returns an hard error,
@@ -178,8 +211,102 @@ batch_withdraw_transaction (void *cls,
   bool balance_ok = false;
   bool found = false;
   const char *kyc_required;
+  struct TALER_PaytoHashP reserve_h_payto;
 
   wc->now = GNUNET_TIME_timestamp_get ();
+  /* Do AML check: compute total merged amount and check
+     against applicable AML threshold */
+  {
+    char *reserve_payto;
+
+    reserve_payto = TALER_reserve_make_payto (TEH_base_url,
+                                              wc->reserve_pub);
+    TALER_payto_hash (reserve_payto,
+                      &reserve_h_payto);
+    GNUNET_free (reserve_payto);
+  }
+  {
+    struct TALER_Amount merge_amount;
+    struct TALER_Amount threshold;
+    struct GNUNET_TIME_Absolute now_minus_one_month;
+
+    now_minus_one_month
+      = GNUNET_TIME_absolute_subtract (wc->now.abs_time,
+                                       GNUNET_TIME_UNIT_MONTHS);
+    GNUNET_assert (GNUNET_OK ==
+                   TALER_amount_set_zero (TEH_currency,
+                                          &merge_amount));
+    qs = TEH_plugin->select_merge_amounts_for_kyc_check (TEH_plugin->cls,
+                                                         &reserve_h_payto,
+                                                         now_minus_one_month,
+                                                         &aml_amount_cb,
+                                                         &merge_amount);
+    if (qs < 0)
+    {
+      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+      if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+        *mhd_ret = TALER_MHD_reply_with_error (connection,
+                                               MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                               TALER_EC_GENERIC_DB_FETCH_FAILED,
+                                               "select_merge_amounts_for_kyc_check");
+      return qs;
+    }
+    qs = TEH_plugin->select_aml_threshold (TEH_plugin->cls,
+                                           &reserve_h_payto,
+                                           &wc->aml_decision,
+                                           &threshold);
+    if (qs < 0)
+    {
+      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+      if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+        *mhd_ret = TALER_MHD_reply_with_error (connection,
+                                               MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                               TALER_EC_GENERIC_DB_FETCH_FAILED,
+                                               "select_aml_threshold");
+      return qs;
+    }
+    if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
+    {
+      threshold = TEH_aml_threshold; /* use default */
+      wc->aml_decision = TALER_AML_NORMAL;
+    }
+
+    switch (wc->aml_decision)
+    {
+    case TALER_AML_NORMAL:
+      if (0 >= TALER_amount_cmp (&merge_amount,
+                                 &threshold))
+      {
+        /* merge_amount <= threshold, continue withdraw below */
+        break;
+      }
+      wc->aml_decision = TALER_AML_PENDING;
+      qs = TEH_plugin->trigger_aml_process (TEH_plugin->cls,
+                                            &reserve_h_payto,
+                                            &merge_amount);
+      if (qs <= 0)
+      {
+        GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+        if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+          *mhd_ret = TALER_MHD_reply_with_error (connection,
+                                                 MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                                 TALER_EC_GENERIC_DB_STORE_FAILED,
+                                                 "trigger_aml_process");
+        return qs;
+      }
+      return qs;
+    case TALER_AML_PENDING:
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "AML already pending, doing nothing\n");
+      return qs;
+    case TALER_AML_FROZEN:
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Account frozen, doing nothing\n");
+      return qs;
+    }
+  }
+
+  /* Check if the money came from a wire transfer */
   qs = TEH_plugin->reserves_get_origin (TEH_plugin->cls,
                                         wc->reserve_pub,
                                         &wc->h_payto);
@@ -352,6 +479,9 @@ generate_reply_success (const struct TEH_RequestContext *rc,
                                             &wc->h_payto,
                                             &wc->kyc);
   }
+  if (TALER_AML_NORMAL != wc->aml_decision)
+    return TEH_RESPONSE_reply_aml_blocked (rc->connection,
+                                           wc->aml_decision);
 
   sigs = json_array ();
   GNUNET_assert (NULL != sigs);
diff --git a/src/exchange/taler-exchange-httpd_responses.c b/src/exchange/taler-exchange-httpd_responses.c
index 33bc13985..5d9dfc3aa 100644
--- a/src/exchange/taler-exchange-httpd_responses.c
+++ b/src/exchange/taler-exchange-httpd_responses.c
@@ -1142,4 +1142,29 @@ TEH_RESPONSE_reply_kyc_required (struct MHD_Connection *connection,
 }
 
 
+MHD_RESULT
+TEH_RESPONSE_reply_aml_blocked (struct MHD_Connection *connection,
+                                enum TALER_AmlDecisionState status)
+{
+  enum TALER_ErrorCode ec = TALER_EC_GENERIC_INTERNAL_INVARIANT_FAILURE;
+
+  switch (status)
+  {
+  case TALER_AML_NORMAL:
+    GNUNET_break (0);
+    return MHD_NO;
+  case TALER_AML_PENDING:
+    ec = TALER_EC_EXCHANGE_GENERIC_AML_PENDING;
+    break;
+  case TALER_AML_FROZEN:
+    ec = TALER_EC_EXCHANGE_GENERIC_AML_FROZEN;
+    break;
+  }
+  return TALER_MHD_REPLY_JSON_PACK (
+    connection,
+    MHD_HTTP_UNAVAILABLE_FOR_LEGAL_REASONS,
+    TALER_JSON_pack_ec (ec));
+}
+
+
 /* end of taler-exchange-httpd_responses.c */
diff --git a/src/exchange/taler-exchange-httpd_responses.h b/src/exchange/taler-exchange-httpd_responses.h
index ba6577b29..0db6968f8 100644
--- a/src/exchange/taler-exchange-httpd_responses.h
+++ b/src/exchange/taler-exchange-httpd_responses.h
@@ -92,6 +92,19 @@ TEH_RESPONSE_reply_kyc_required (struct MHD_Connection *connection,
 
 
 /**
+ * Send information that an AML process is blocking
+ * the operation right now.
+ *
+ * @param connection connection to the client
+ * @param status current AML status
+ * @return MHD result code
+ */
+MHD_RESULT
+TEH_RESPONSE_reply_aml_blocked (struct MHD_Connection *connection,
+                                enum TALER_AmlDecisionState status);
+
+
+/**
  * Send assertion that the given denomination key hash
  * is not usable (typically expired) at this time.
  *
diff --git a/src/exchange/taler-exchange-httpd_withdraw.c b/src/exchange/taler-exchange-httpd_withdraw.c
index 567cad5a9..40cefc7dc 100644
--- a/src/exchange/taler-exchange-httpd_withdraw.c
+++ b/src/exchange/taler-exchange-httpd_withdraw.c
@@ -61,16 +61,21 @@ struct WithdrawContext
   struct TALER_EXCHANGEDB_KycStatus kyc;
 
   /**
-   * Hash of the payto-URI representing the reserve
-   * from which we are withdrawing.
+   * Hash of the payto-URI representing the account
+   * from which the money was put into the reserve.
    */
-  struct TALER_PaytoHashP h_payto;
+  struct TALER_PaytoHashP h_account_payto;
 
   /**
    * Current time for the DB transaction.
    */
   struct GNUNET_TIME_Timestamp now;
 
+  /**
+   * AML decision, #TALER_AML_NORMAL if we may proceed.
+   */
+  enum TALER_AmlDecisionState aml_decision;
+
 };
 
 
@@ -108,7 +113,7 @@ withdraw_amount_cb (void *cls,
     return;
   qs = TEH_plugin->select_withdraw_amounts_for_kyc_check (
     TEH_plugin->cls,
-    &wc->h_payto,
+    &wc->h_account_payto,
     limit,
     cb,
     cb_cls);
@@ -121,6 +126,34 @@ withdraw_amount_cb (void *cls,
 
 
 /**
+ * Function called on each @a amount that was found to
+ * be relevant for the AML check as it was merged into
+ * the reserve.
+ *
+ * @param cls `struct TALER_Amount *` to total up the amounts
+ * @param amount encountered transaction amount
+ * @param date when was the amount encountered
+ * @return #GNUNET_OK to continue to iterate,
+ *         #GNUNET_NO to abort iteration
+ *         #GNUNET_SYSERR on internal error (also abort itaration)
+ */
+static enum GNUNET_GenericReturnValue
+aml_amount_cb (
+  void *cls,
+  const struct TALER_Amount *amount,
+  struct GNUNET_TIME_Absolute date)
+{
+  struct TALER_Amount *total = cls;
+
+  GNUNET_assert (0 <=
+                 TALER_amount_add (total,
+                                   total,
+                                   amount));
+  return GNUNET_OK;
+}
+
+
+/**
  * Function implementing withdraw transaction.  Runs the
  * transaction logic; IF it returns a non-error code, the transaction
  * logic MUST NOT queue a MHD response.  IF it returns an hard error,
@@ -150,23 +183,116 @@ withdraw_transaction (void *cls,
   uint64_t ruuid;
   const struct TALER_CsNonce *nonce;
   const struct TALER_BlindedPlanchet *bp;
+  struct TALER_PaytoHashP reserve_h_payto;
 
   wc->now = GNUNET_TIME_timestamp_get ();
+  /* Do AML check: compute total merged amount and check
+     against applicable AML threshold */
+  {
+    char *reserve_payto;
+
+    reserve_payto = TALER_reserve_make_payto (TEH_base_url,
+                                              &wc->collectable.reserve_pub);
+    TALER_payto_hash (reserve_payto,
+                      &reserve_h_payto);
+    GNUNET_free (reserve_payto);
+  }
+  {
+    struct TALER_Amount merge_amount;
+    struct TALER_Amount threshold;
+    struct GNUNET_TIME_Absolute now_minus_one_month;
+
+    now_minus_one_month
+      = GNUNET_TIME_absolute_subtract (wc->now.abs_time,
+                                       GNUNET_TIME_UNIT_MONTHS);
+    GNUNET_assert (GNUNET_OK ==
+                   TALER_amount_set_zero (TEH_currency,
+                                          &merge_amount));
+    qs = TEH_plugin->select_merge_amounts_for_kyc_check (TEH_plugin->cls,
+                                                         &reserve_h_payto,
+                                                         now_minus_one_month,
+                                                         &aml_amount_cb,
+                                                         &merge_amount);
+    if (qs < 0)
+    {
+      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+      if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+        *mhd_ret = TALER_MHD_reply_with_error (connection,
+                                               MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                               TALER_EC_GENERIC_DB_FETCH_FAILED,
+                                               "select_merge_amounts_for_kyc_check");
+      return qs;
+    }
+    qs = TEH_plugin->select_aml_threshold (TEH_plugin->cls,
+                                           &reserve_h_payto,
+                                           &wc->aml_decision,
+                                           &threshold);
+    if (qs < 0)
+    {
+      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+      if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+        *mhd_ret = TALER_MHD_reply_with_error (connection,
+                                               MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                               TALER_EC_GENERIC_DB_FETCH_FAILED,
+                                               "select_aml_threshold");
+      return qs;
+    }
+    if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
+    {
+      threshold = TEH_aml_threshold; /* use default */
+      wc->aml_decision = TALER_AML_NORMAL;
+    }
+
+    switch (wc->aml_decision)
+    {
+    case TALER_AML_NORMAL:
+      if (0 >= TALER_amount_cmp (&merge_amount,
+                                 &threshold))
+      {
+        /* merge_amount <= threshold, continue withdraw below */
+        break;
+      }
+      wc->aml_decision = TALER_AML_PENDING;
+      qs = TEH_plugin->trigger_aml_process (TEH_plugin->cls,
+                                            &reserve_h_payto,
+                                            &merge_amount);
+      if (qs <= 0)
+      {
+        GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+        if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+          *mhd_ret = TALER_MHD_reply_with_error (connection,
+                                                 MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                                 TALER_EC_GENERIC_DB_STORE_FAILED,
+                                                 "trigger_aml_process");
+        return qs;
+      }
+      return qs;
+    case TALER_AML_PENDING:
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "AML already pending, doing nothing\n");
+      return qs;
+    case TALER_AML_FROZEN:
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Account frozen, doing nothing\n");
+      return qs;
+    }
+  }
+
+  /* Check if the money came from a wire transfer */
   qs = TEH_plugin->reserves_get_origin (TEH_plugin->cls,
                                         &wc->collectable.reserve_pub,
-                                        &wc->h_payto);
+                                        &wc->h_account_payto);
   if (qs < 0)
     return qs;
-  /* If no results, reserve was created by merge,
-     in which case no KYC check is required as the
-     merge already did that. */
+  /* If no results, reserve was created by merge, in which case no KYC check
+     is required as the merge already did that. */
   if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs)
   {
     const char *kyc_required;
 
     qs = TALER_KYCLOGIC_kyc_test_required (
       TALER_KYCLOGIC_KYC_TRIGGER_WITHDRAW,
-      &wc->h_payto,
+      &wc->h_account_payto,
       TEH_plugin->select_satisfied_kyc_processes,
       TEH_plugin->cls,
       &withdraw_amount_cb,
@@ -191,7 +317,7 @@ withdraw_transaction (void *cls,
       return TEH_plugin->insert_kyc_requirement_for_account (
         TEH_plugin->cls,
         kyc_required,
-        &wc->h_payto,
+        &wc->h_account_payto,
         &wc->kyc.requirement_row);
     }
   }
@@ -515,8 +641,13 @@ TEH_handler_withdraw (struct TEH_RequestContext *rc,
 
   if (! wc.kyc.ok)
     return TEH_RESPONSE_reply_kyc_required (rc->connection,
-                                            &wc.h_payto,
+                                            &wc.h_account_payto,
                                             &wc.kyc);
+
+  if (TALER_AML_NORMAL != wc.aml_decision)
+    return TEH_RESPONSE_reply_aml_blocked (rc->connection,
+                                           wc.aml_decision);
+
   {
     MHD_RESULT ret;
 
diff --git a/src/exchange/test_taler_exchange_httpd.conf b/src/exchange/test_taler_exchange_httpd.conf
index 9bd4851fb..0af23b9df 100644
--- a/src/exchange/test_taler_exchange_httpd.conf
+++ b/src/exchange/test_taler_exchange_httpd.conf
@@ -7,6 +7,7 @@ TALER_RUNTIME_DIR = ${TMPDIR:-${TMP:-/tmp}}/${USER:-}/taler-system-runtime/
 # Currency supported by the exchange (can only be one)
 CURRENCY = EUR
 CURRENCY_ROUND_UNIT = EUR:0.01
+AML_THRESHOLD = EUR:1000000
 
 [auditor]
 TINY_AMOUNT = EUR:0.01
diff --git a/src/exchangedb/pg_reserves_in_insert.c b/src/exchangedb/pg_reserves_in_insert.c
index ac14cb567..da21a9063 100644
--- a/src/exchangedb/pg_reserves_in_insert.c
+++ b/src/exchangedb/pg_reserves_in_insert.c
@@ -54,25 +54,6 @@ compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
 }
 
 
-static void
-notify_on_reserve (struct PostgresClosure *pg,
-                   const struct TALER_ReservePublicKeyP *reserve_pub)
-{
-  struct TALER_ReserveEventP rep = {
-    .header.size = htons (sizeof (rep)),
-    .header.type = htons (TALER_DBEVENT_EXCHANGE_RESERVE_INCOMING),
-    .reserve_pub = *reserve_pub
-  };
-
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-              "Notifying on reserve!\n");
-  TEH_PG_event_notify (pg,
-                       &rep.header,
-                       NULL,
-                       0);
-}
-
-
 static enum GNUNET_DB_QueryStatus
 insert1 (struct PostgresClosure *pg,
          const struct TALER_EXCHANGEDB_ReserveInInfo reserves[1],
diff --git a/src/exchangedb/pg_select_aml_threshold.c b/src/exchangedb/pg_select_aml_threshold.c
index 723524adf..e67a57a39 100644
--- a/src/exchangedb/pg_select_aml_threshold.c
+++ b/src/exchangedb/pg_select_aml_threshold.c
@@ -41,7 +41,7 @@ TEH_PG_select_aml_threshold (
   uint32_t status32 = TALER_AML_NORMAL;
   struct GNUNET_PQ_ResultSpec rs[] = {
     TALER_PQ_RESULT_SPEC_AMOUNT ("threshold",
-                                 &threshold),
+                                 threshold),
     GNUNET_PQ_result_spec_uint32 ("status",
                                   &status32),
     GNUNET_PQ_result_spec_end