commit 390f0850310573649b7b33bcb1112b6cbe5ff6ee
parent 9ace1bc7f330c800c5e6e03835ede87a924f5311
Author: Christian Grothoff <christian@grothoff.org>
Date: Sun, 5 Jan 2025 14:20:22 +0100
library function to wrap sanction list evaluation program
Diffstat:
3 files changed, 609 insertions(+), 1 deletion(-)
diff --git a/src/include/taler_kyclogic_lib.h b/src/include/taler_kyclogic_lib.h
@@ -1131,4 +1131,72 @@ void
TALER_KYCLOGIC_run_aml_program_cancel (
struct TALER_KYCLOGIC_AmlProgramRunnerHandle *aprh);
+
+/**
+ * Handle to a sanction list evaluation helper process.
+ */
+struct TALER_KYCLOGIC_SanctionRater;
+
+/**
+ * Function called with the result of a sanction evaluation.
+ *
+ * @param cls closure
+ * @param ec error code, #TALER_EC_NONE on success
+ * @param best_match identifies the sanction list entry with the best match
+ * @param rating likelihood of the match, from 0 (none) to 1 (perfect)
+ * @param confidence confidence in the evaluation, from 0 (none) to 1 (perfect)
+ */
+typedef void
+(*TALER_KYCLOGIC_SanctionResultCallback)(
+ void *cls,
+ enum TALER_ErrorCode ec,
+ const char *best_match,
+ double rating,
+ double confidence);
+
+
+/**
+ * Launch sanction rating helper process
+ *
+ * @param binary program name
+ * @param argv argument to give to the process
+ * @return handle for sanction list rating
+ */
+struct TALER_KYCLOGIC_SanctionRater *
+TALER_KYCLOGIC_saction_rater_start (
+ const char *binary,
+ const char **argv);
+
+
+/**
+ * KYC evaluation.
+ */
+struct TALER_KYCLOGIC_EvaluationEntry;
+
+/**
+ * Evaluate KYC attributes against sacntions list using @a sr
+ *
+ * @param[in,out] sr santion list evaluator
+ * @param attributes KYC attributes to evaluate
+ * @param cb function to call with the results
+ * @param cb_cls closure
+ * @return NULL on error
+ */
+struct TALER_KYCLOGIC_EvaluationEntry *
+TALER_KYCLOGIC_saction_rater_eval (
+ struct TALER_KYCLOGIC_SanctionRater *sr,
+ const json_t *attributes,
+ TALER_KYCLOGIC_SanctionResultCallback cb,
+ void *cb_cls);
+
+
+/**
+ * Stop sanction rating helper process.
+ *
+ * @param[in] sr process to stop
+ */
+void
+TALER_KYCLOGIC_saction_rater_stop (
+ struct TALER_KYCLOGIC_SanctionRater *sr);
+
#endif
diff --git a/src/kyclogic/Makefile.am b/src/kyclogic/Makefile.am
@@ -45,7 +45,8 @@ lib_LTLIBRARIES = \
libtalerkyclogic.la
libtalerkyclogic_la_SOURCES = \
- kyclogic_api.c
+ kyclogic_api.c \
+ kyclogic_sanctions.c
libtalerkyclogic_la_LIBADD = \
$(top_builddir)/src/json/libtalerjson.la \
$(top_builddir)/src/util/libtalerutil.la \
diff --git a/src/kyclogic/kyclogic_sanctions.c b/src/kyclogic/kyclogic_sanctions.c
@@ -0,0 +1,539 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2025 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 published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
+
+ You should have received a copy of the GNU Affero General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+*/
+/**
+ * @file kyclogic_sanctions.c
+ * @brief wrapper around sanction list evaluator
+ * @author Christian Grothoff
+ */
+#include "platform.h"
+#include "taler_json_lib.h"
+#include "taler_kyclogic_lib.h"
+
+
+/**
+ * Entry in the ordered list of pending evaluations.
+ */
+struct TALER_KYCLOGIC_EvaluationEntry
+{
+ /**
+ * Kept in a DLL.
+ */
+ struct TALER_KYCLOGIC_EvaluationEntry *prev;
+
+ /**
+ * Kept in a DLL.
+ */
+ struct TALER_KYCLOGIC_EvaluationEntry *next;
+
+ /**
+ * Callback to call with the result.
+ */
+ TALER_KYCLOGIC_SanctionResultCallback cb;
+
+ /**
+ * Closure for @e cb.
+ */
+ void *cb_cls;
+
+ /**
+ * Buffer with data we need to send to the helper.
+ */
+ void *write_buf;
+
+ /**
+ * Total length of @e write_buf.
+ */
+ size_t write_size;
+
+ /**
+ * Current write position in @e write_buf.
+ */
+ size_t write_pos;
+
+};
+
+
+/**
+ * Handle to a sanction list evaluation helper process.
+ */
+struct TALER_KYCLOGIC_SanctionRater
+{
+
+ /**
+ * Kept in a DLL.
+ */
+ struct TALER_KYCLOGIC_EvaluationEntry *ee_head;
+
+ /**
+ * Kept in a DLL.
+ */
+ struct TALER_KYCLOGIC_EvaluationEntry *ee_tail;
+
+ /**
+ * Handle to the helper process.
+ */
+ struct GNUNET_OS_Process *helper;
+
+ /**
+ * Pipe for the stdin of the @e helper.
+ */
+ struct GNUNET_DISK_FileHandle *chld_stdin;
+
+ /**
+ * Pipe for the stdout of the @e helper.
+ */
+ struct GNUNET_DISK_FileHandle *chld_stdout;
+
+ /**
+ * Handle to wait on the child to terminate.
+ */
+ struct GNUNET_ChildWaitHandle *cwh;
+
+ /**
+ * Task to read JSON output from the child.
+ */
+ struct GNUNET_SCHEDULER_Task *read_task;
+
+ /**
+ * Task to send JSON input to the child.
+ */
+ struct GNUNET_SCHEDULER_Task *write_task;
+
+ /**
+ * Buffer for reading data from the helper.
+ */
+ void *read_buf;
+
+ /**
+ * Current size of @a read_buf.
+ */
+ size_t read_size;
+
+ /**
+ * Current offset in @a read_buf.
+ */
+ size_t read_pos;
+
+};
+
+
+/**
+ * We encountered a hard error (or explicit stop) of @a sr.
+ * Shut down processing (but do not yet free @a sr).
+ *
+ * @param[in,out] sr sanction rater to fail
+ */
+static void
+fail_hard (struct TALER_KYCLOGIC_SanctionRater *sr)
+{
+ struct TALER_KYCLOGIC_EvaluationEntry *ee;
+
+ if (NULL != sr->chld_stdin)
+ {
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_file_close (sr->chld_stdin));
+ sr->chld_stdin = NULL;
+ }
+ if (NULL != sr->read_task)
+ {
+ GNUNET_SCHEDULER_cancel (sr->read_task);
+ sr->read_task = NULL;
+ }
+ if (NULL != sr->helper)
+ {
+ GNUNET_OS_process_destroy (sr->helper);
+ sr->helper = NULL;
+ }
+ while (NULL != (ee = sr->ee_tail))
+ {
+ GNUNET_CONTAINER_DLL_remove (sr->ee_head,
+ sr->ee_tail,
+ ee);
+ ee->cb (ee->cb_cls,
+ /* FIXME: better EC? */
+ TALER_EC_GENERIC_INTERNAL_INVARIANT_FAILURE,
+ NULL,
+ 1.0,
+ 0.0);
+ free (ee->write_buf);
+ GNUNET_free (ee);
+ }
+}
+
+
+/**
+ * Parse data from input buffer.
+ *
+ * @param[in,out] sr sanction rater to process data from
+ * @return true if everything is fine, false on failure
+ */
+static bool
+process_buffer (struct TALER_KYCLOGIC_SanctionRater *sr)
+{
+ const char *buf = sr->read_buf;
+ size_t buf_len;
+ void *end;
+
+ end = memrchr (sr->read_buf,
+ '\n',
+ sr->read_pos);
+ if ( (NULL == end) &&
+ (sr->read_pos < 2048) )
+ return true;
+ buf_len = end - sr->read_buf;
+ while (0 != buf_len)
+ {
+ char *nl;
+ double rating;
+ double confidence;
+ char best_match[1024];
+ size_t line_len;
+
+ nl = memchr (buf,
+ '\n',
+ buf_len);
+ GNUNET_assert (NULL != nl);
+ *nl = '\0';
+ line_len = nl - buf;
+ if (3 !=
+ sscanf (buf,
+ "%lf %lf %1023s",
+
+ &rating,
+ &confidence,
+ best_match))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Malformed input line `%s'\n",
+ buf);
+ GNUNET_break (0);
+ return false;
+ }
+ {
+ struct TALER_KYCLOGIC_EvaluationEntry *ee = sr->ee_tail;
+
+ GNUNET_CONTAINER_DLL_remove (sr->ee_head,
+ sr->ee_tail,
+ ee);
+ ee->cb (ee->cb_cls,
+ TALER_EC_NONE,
+ best_match,
+ rating,
+ confidence);
+ free (ee->write_buf);
+ GNUNET_free (ee);
+ }
+ buf += line_len;
+ buf_len -= line_len;
+ }
+ buf_len = end - sr->read_buf;
+ memmove (sr->read_buf,
+ end,
+ sr->read_pos - buf_len);
+ sr->read_pos -= buf_len;
+ return true;
+}
+
+
+/**
+ * Function called when we can read more data from
+ * the child process.
+ *
+ * @param cls our `struct TALER_KYCLOGIC_SanctionRater *`
+ */
+static void
+read_cb (void *cls)
+{
+ struct TALER_KYCLOGIC_SanctionRater *sr = cls;
+
+ sr->read_task = NULL;
+ while (1)
+ {
+ ssize_t ret;
+
+ if (sr->read_size == sr->read_pos)
+ {
+ /* Grow input buffer */
+ size_t ns;
+ void *tmp;
+
+ ns = GNUNET_MAX (2 * sr->read_size,
+ 1024);
+ if (ns > GNUNET_MAX_MALLOC_CHECKED)
+ ns = GNUNET_MAX_MALLOC_CHECKED;
+ if (sr->read_size == ns)
+ {
+ /* Helper returned more than 40 MB of data! Stop reading! */
+ GNUNET_break (0);
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_file_close (sr->chld_stdin));
+ return;
+ }
+ tmp = GNUNET_malloc_large (ns);
+ if (NULL == tmp)
+ {
+ /* out of memory, also stop reading */
+ GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR,
+ "malloc");
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_file_close (sr->chld_stdin));
+ return;
+ }
+ GNUNET_memcpy (tmp,
+ sr->read_buf,
+ sr->read_pos);
+ GNUNET_free (sr->read_buf);
+ sr->read_buf = tmp;
+ sr->read_size = ns;
+ }
+ ret = GNUNET_DISK_file_read (sr->chld_stdout,
+ sr->read_buf + sr->read_pos,
+ sr->read_size - sr->read_pos);
+ if (ret < 0)
+ {
+ if ( (EAGAIN != errno) &&
+ (EWOULDBLOCK != errno) &&
+ (EINTR != errno) )
+ {
+ GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
+ "read");
+ return;
+ }
+ /* Continue later */
+ break;
+ }
+ if (0 == ret)
+ {
+ /* regular end of stream, odd! */
+ GNUNET_break (0);
+ fail_hard (sr);
+ return;
+ }
+ GNUNET_assert (sr->read_size >= sr->read_pos + ret);
+ sr->read_pos += ret;
+ if (! process_buffer (sr))
+ return;
+ }
+ sr->read_task
+ = GNUNET_SCHEDULER_add_read_file (
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ sr->chld_stdout,
+ &read_cb,
+ sr);
+}
+
+
+/**
+ * Function called when we can write more data to
+ * the child process.
+ *
+ * @param cls our `struct SanctionRater *`
+ */
+static void
+write_cb (void *cls)
+{
+ struct TALER_KYCLOGIC_SanctionRater *sr = cls;
+ struct TALER_KYCLOGIC_EvaluationEntry *ee = sr->ee_tail;
+ ssize_t ret;
+
+ sr->write_task = NULL;
+ while (ee->write_size > ee->write_pos)
+ {
+ ret = GNUNET_DISK_file_write (sr->chld_stdin,
+ ee->write_buf + ee->write_pos,
+ ee->write_size - ee->write_pos);
+ if (ret < 0)
+ {
+ if ( (EAGAIN != errno) &&
+ (EINTR != errno) )
+ GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
+ "write");
+ break;
+ }
+ if (0 == ret)
+ {
+ GNUNET_break (0);
+ break;
+ }
+ GNUNET_assert (ee->write_size >= ee->write_pos + ret);
+ ee->write_pos += ret;
+ }
+ if (ee->write_size == ee->write_pos)
+ {
+ free (ee->write_buf);
+ GNUNET_CONTAINER_DLL_remove (sr->ee_head,
+ sr->ee_tail,
+ ee);
+ GNUNET_free (ee);
+ ee = sr->ee_tail;
+ if (NULL == ee)
+ return;
+ }
+ if ( (ee->write_size > ee->write_pos) &&
+ ( (EAGAIN == errno) ||
+ (EWOULDBLOCK == errno) ||
+ (EINTR == errno) ) )
+ {
+ sr->write_task
+ = GNUNET_SCHEDULER_add_write_file (
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ sr->chld_stdin,
+ &write_cb,
+ sr);
+ return;
+ }
+ /* helper must have died */
+ GNUNET_break (0);
+ fail_hard (sr);
+}
+
+
+/**
+ * Defines a GNUNET_ChildCompletedCallback which is sent back
+ * upon death or completion of a child process.
+ *
+ * @param cls handle for the callback
+ * @param type type of the process
+ * @param exit_code status code of the process
+ *
+ */
+static void
+child_done_cb (void *cls,
+ enum GNUNET_OS_ProcessStatusType type,
+ long unsigned int exit_code)
+{
+ struct TALER_KYCLOGIC_SanctionRater *sr = cls;
+
+ sr->cwh = NULL;
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Conversion helper exited with status %d and code %llu after outputting %llu bytes of data\n",
+ (int) type,
+ (unsigned long long) exit_code,
+ (unsigned long long) sr->read_pos);
+ fail_hard (sr);
+}
+
+
+struct TALER_KYCLOGIC_SanctionRater *
+TALER_KYCLOGIC_saction_rater_start (const char *binary,
+ const char **argv)
+{
+ struct TALER_KYCLOGIC_SanctionRater *sr;
+ struct GNUNET_DISK_PipeHandle *pipe_stdin;
+ struct GNUNET_DISK_PipeHandle *pipe_stdout;
+
+ sr = GNUNET_new (struct TALER_KYCLOGIC_SanctionRater);
+ pipe_stdin = GNUNET_DISK_pipe (GNUNET_DISK_PF_BLOCKING_READ);
+ GNUNET_assert (NULL != pipe_stdin);
+ pipe_stdout = GNUNET_DISK_pipe (GNUNET_DISK_PF_BLOCKING_WRITE);
+ GNUNET_assert (NULL != pipe_stdout);
+ sr->helper = GNUNET_OS_start_process_vap (GNUNET_OS_INHERIT_STD_ERR,
+ pipe_stdin,
+ pipe_stdout,
+ NULL,
+ binary,
+ (char *const *) argv);
+ if (NULL == sr->helper)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Failed to run conversion helper `%s'\n",
+ binary);
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_pipe_close (pipe_stdin));
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_pipe_close (pipe_stdout));
+ GNUNET_free (sr);
+ return NULL;
+ }
+ sr->chld_stdin =
+ GNUNET_DISK_pipe_detach_end (pipe_stdin,
+ GNUNET_DISK_PIPE_END_WRITE);
+ sr->chld_stdout =
+ GNUNET_DISK_pipe_detach_end (pipe_stdout,
+ GNUNET_DISK_PIPE_END_READ);
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_pipe_close (pipe_stdin));
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_pipe_close (pipe_stdout));
+
+ sr->read_task
+ = GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL,
+ sr->chld_stdout,
+ &read_cb,
+ sr);
+ sr->cwh = GNUNET_wait_child (sr->helper,
+ &child_done_cb,
+ sr);
+ return sr;
+}
+
+
+struct TALER_KYCLOGIC_EvaluationEntry *
+TALER_KYCLOGIC_saction_rater_eval (struct TALER_KYCLOGIC_SanctionRater *sr,
+ const json_t *attributes,
+ TALER_KYCLOGIC_SanctionResultCallback cb,
+ void *cb_cls)
+{
+ struct TALER_KYCLOGIC_EvaluationEntry *ee;
+
+ if (NULL == sr->read_task)
+ return NULL;
+ ee = GNUNET_new (struct TALER_KYCLOGIC_EvaluationEntry);
+ ee->cb = cb;
+ ee->cb_cls = cb_cls;
+ GNUNET_CONTAINER_DLL_insert (sr->ee_head,
+ sr->ee_tail,
+ ee);
+ ee->write_buf = json_dumps (attributes,
+ JSON_COMPACT);
+ ee->write_size = strlen (ee->write_buf);
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Passing %llu bytes to JSON conversion tool\n",
+ (unsigned long long) ee->write_size);
+ if (NULL == sr->write_task)
+ sr->write_task
+ = GNUNET_SCHEDULER_add_write_file (GNUNET_TIME_UNIT_FOREVER_REL,
+ sr->chld_stdin,
+ &write_cb,
+ sr);
+ return ee;
+}
+
+
+void
+TALER_KYCLOGIC_saction_rater_stop (
+ struct TALER_KYCLOGIC_SanctionRater *sr)
+{
+ fail_hard (sr);
+ if (NULL != sr->cwh)
+ {
+ GNUNET_wait_child_cancel (sr->cwh);
+ sr->cwh = NULL;
+ }
+ if (NULL != sr->write_task)
+ {
+ GNUNET_SCHEDULER_cancel (sr->write_task);
+ sr->write_task = NULL;
+ }
+ if (NULL != sr->chld_stdout)
+ {
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_DISK_file_close (sr->chld_stdout));
+ sr->chld_stdout = NULL;
+ }
+ GNUNET_free (sr->read_buf);
+ GNUNET_free (sr);
+}
