summaryrefslogtreecommitdiffstats
path: root/third_party/pam_wrapper/python/pypamtest.c
diff options
context:
space:
mode:
Diffstat (limited to 'third_party/pam_wrapper/python/pypamtest.c')
-rw-r--r--third_party/pam_wrapper/python/pypamtest.c1307
1 files changed, 1307 insertions, 0 deletions
diff --git a/third_party/pam_wrapper/python/pypamtest.c b/third_party/pam_wrapper/python/pypamtest.c
new file mode 100644
index 0000000..008a85f
--- /dev/null
+++ b/third_party/pam_wrapper/python/pypamtest.c
@@ -0,0 +1,1307 @@
+/*
+ * Copyright (c) 2015 Andreas Schneider <asn@samba.org>
+ * Copyright (c) 2015 Jakub Hrozek <jakub.hrozek@posteo.se>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "config.h"
+
+#include <Python.h>
+#include <structmember.h>
+
+#include "libpamtest.h"
+
+#define PYTHON_MODULE_NAME "pypamtest"
+
+#ifndef discard_const_p
+#if defined(__intptr_t_defined) || defined(HAVE_UINTPTR_T)
+# define discard_const_p(type, ptr) ((type *)((uintptr_t)(ptr)))
+#else
+# define discard_const_p(type, ptr) ((type *)(ptr))
+#endif
+#endif
+
+#define __unused __attribute__((__unused__))
+
+#if PY_MAJOR_VERSION >= 3
+#define IS_PYTHON3 1
+#define RETURN_ON_ERROR return NULL
+#else
+#define IS_PYTHON3 0
+#define RETURN_ON_ERROR return
+#endif /* PY_MAJOR_VERSION */
+
+/* We only return up to 16 messages from the PAM conversation */
+#define PAM_CONV_MSG_MAX 16
+
+#if IS_PYTHON3
+PyMODINIT_FUNC PyInit_pypamtest(void);
+#else
+PyMODINIT_FUNC initpypamtest(void);
+#endif
+
+typedef struct {
+ PyObject_HEAD
+
+ enum pamtest_ops pam_operation;
+ int expected_rv;
+ int flags;
+
+ PyObject *pam_handle;
+ PyObject *pam_env;
+} TestCaseObject;
+
+#define PyTestCase_AsTestCaseObject(py_obj) \
+ (TestCaseObject *)(py_obj)
+
+/**********************************************************
+ *** module-specific exceptions
+ **********************************************************/
+static PyObject *PyExc_PamTestError;
+
+/**********************************************************
+ *** helper functions
+ **********************************************************/
+
+#define REPR_FMT "{ pam_operation [%d] " \
+ "expected_rv [%d] " \
+ "flags [%d] }"
+
+static char *py_strdup(const char *string)
+{
+ char *copy;
+
+ copy = PyMem_New(char, strlen(string) + 1);
+ if (copy == NULL) {
+ PyErr_NoMemory();
+ return NULL;
+ }
+
+ return strcpy(copy, string);
+}
+
+static PyObject *get_utf8_string(PyObject *obj,
+ const char *attrname)
+{
+ const char *a = attrname ? attrname : "attribute";
+ PyObject *obj_utf8 = NULL;
+
+ if (PyBytes_Check(obj)) {
+ obj_utf8 = obj;
+ Py_INCREF(obj_utf8); /* Make sure we can DECREF later */
+ } else if (PyUnicode_Check(obj)) {
+ if ((obj_utf8 = PyUnicode_AsUTF8String(obj)) == NULL) {
+ return NULL;
+ }
+ } else {
+ PyErr_Format(PyExc_TypeError, "%s must be a string", a);
+ return NULL;
+ }
+
+ return obj_utf8;
+}
+
+static void free_cstring_list(const char **list)
+{
+ int i;
+
+ if (list == NULL) {
+ return;
+ }
+
+ for (i=0; list[i]; i++) {
+ PyMem_Free(discard_const_p(char, list[i]));
+ }
+ PyMem_Free(list);
+}
+
+static void free_string_list(char **list)
+{
+ int i;
+
+ if (list == NULL) {
+ return;
+ }
+
+ for (i=0; list[i]; i++) {
+ PyMem_Free(list[i]);
+ }
+ PyMem_Free(list);
+}
+
+static char **new_conv_list(const size_t list_size)
+{
+ char **list;
+ size_t i;
+
+ if (list_size == 0) {
+ return NULL;
+ }
+
+ if (list_size + 1 < list_size) {
+ return NULL;
+ }
+
+ list = PyMem_New(char *, list_size + 1);
+ if (list == NULL) {
+ return NULL;
+ }
+ list[list_size] = NULL;
+
+ for (i = 0; i < list_size; i++) {
+ list[i] = PyMem_New(char, PAM_MAX_MSG_SIZE);
+ if (list[i] == NULL) {
+ PyMem_Free(list);
+ return NULL;
+ }
+ memset(list[i], 0, PAM_MAX_MSG_SIZE);
+ }
+
+ return list;
+}
+
+static int sequence_as_string_list(PyObject *seq,
+ const char *paramname,
+ const char **str_list[],
+ size_t *num_str_list)
+{
+ const char *p = paramname ? paramname : "attribute values";
+ const char **result;
+ PyObject *utf_item;
+ int i;
+ Py_ssize_t len;
+ PyObject *item;
+
+ if (!PySequence_Check(seq)) {
+ PyErr_Format(PyExc_TypeError,
+ "The object must be a sequence\n");
+ return -1;
+ }
+
+ len = PySequence_Size(seq);
+ if (len == -1) {
+ return -1;
+ }
+
+ result = PyMem_New(const char *, (len + 1));
+ if (result == NULL) {
+ PyErr_NoMemory();
+ return -1;
+ }
+
+ for (i = 0; i < len; i++) {
+ item = PySequence_GetItem(seq, i);
+ if (item == NULL) {
+ break;
+ }
+
+ utf_item = get_utf8_string(item, p);
+ if (utf_item == NULL) {
+ Py_DECREF(item);
+ return -1;
+ }
+
+ result[i] = py_strdup(PyBytes_AsString(utf_item));
+ Py_DECREF(utf_item);
+ if (result[i] == NULL) {
+ Py_DECREF(item);
+ return -1;
+ }
+ Py_DECREF(item);
+ }
+
+ result[i] = NULL;
+
+ *str_list = result;
+ *num_str_list = (size_t)len;
+
+ return 0;
+}
+
+static PyObject *string_list_as_tuple(char **str_list)
+{
+ int rc;
+ size_t len, i;
+ PyObject *tup;
+ PyObject *py_str;
+
+ for (len=0; str_list[len] != NULL; len++) {
+ if (str_list[len][0] == '\0') {
+ /* unused string, stop counting */
+ break;
+ }
+ }
+
+ tup = PyTuple_New(len);
+ if (tup == NULL) {
+ PyErr_NoMemory();
+ return NULL;
+ }
+
+ for (i = 0; i < len; i++) {
+ py_str = PyUnicode_FromString(str_list[i]);
+ if (py_str == NULL) {
+ Py_DECREF(tup);
+ PyErr_NoMemory();
+ return NULL;
+ }
+
+ /* PyTuple_SetItem() steals the reference to
+ * py_str, so it's enough to decref the tuple
+ * pointer afterwards */
+ rc = PyTuple_SetItem(tup, i, py_str);
+ if (rc != 0) {
+ /* cleanup */
+ Py_DECREF(py_str);
+ Py_DECREF(tup);
+ PyErr_NoMemory();
+ return NULL;
+ }
+ }
+
+ return tup;
+}
+
+static void
+set_pypamtest_exception(PyObject *exc,
+ enum pamtest_err perr,
+ struct pam_testcase *tests,
+ size_t num_tests)
+{
+ PyObject *obj = NULL;
+ /* REPR_FMT contains just %d expansions, so this is safe */
+ char test_repr[256] = { '\0' };
+ union {
+ char *str;
+ PyObject *obj;
+ } pypam_str_object;
+ const char *strerr;
+ const struct pam_testcase *failed = NULL;
+
+ if (exc == NULL) {
+ PyErr_BadArgument();
+ return;
+ }
+
+ strerr = pamtest_strerror(perr);
+
+ if (perr == PAMTEST_ERR_CASE) {
+ failed = _pamtest_failed_case(tests, num_tests);
+ if (failed) {
+ snprintf(test_repr, sizeof(test_repr), REPR_FMT,
+ failed->pam_operation,
+ failed->expected_rv,
+ failed->flags);
+ }
+ }
+
+ if (test_repr[0] != '\0' && failed != NULL) {
+ PyErr_Format(exc,
+ "Error [%d]: Test case %s returned [%d]",
+ perr, test_repr, failed->op_rv);
+ } else {
+ obj = Py_BuildValue(discard_const_p(char, "(i,s)"),
+ perr,
+ strerr ? strerr : "Unknown error");
+ PyErr_SetObject(exc, obj);
+ }
+
+ pypam_str_object.str = test_repr;
+ Py_XDECREF(pypam_str_object.obj);
+ Py_XDECREF(obj);
+}
+
+/* Returned when doc(test_case) is invoked */
+PyDoc_STRVAR(TestCaseObject__doc__,
+"pamtest test case\n\n"
+"Represents one operation in PAM transaction. An example is authentication, "
+"opening a session or password change. Each operation has an expected error "
+"code. The run_pamtest() function accepts a list of these test case objects\n"
+"Params:\n\n"
+"pam_operation: - the PAM operation to run. Use constants from pypamtest "
+"such as pypamtest.PAMTEST_AUTHENTICATE. This argument is required.\n"
+"expected_rv: - The PAM return value we expect the operation to return. "
+"Defaults to 0 (PAM_SUCCESS)\n"
+"flags: - Additional flags to pass to the PAM operation. Defaults to 0.\n"
+);
+
+static PyObject *
+TestCase_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
+{
+ TestCaseObject *self;
+
+ (void) args; /* unused */
+ (void) kwds; /* unused */
+
+ self = (TestCaseObject *)type->tp_alloc(type, 0);
+ if (self == NULL) {
+ PyErr_NoMemory();
+ return NULL;
+ }
+
+ return (PyObject *) self;
+}
+
+/* The traverse and clear methods must be defined even though they do nothing
+ * otherwise Garbage Collector is not happy
+ */
+static int TestCase_clear(TestCaseObject *self)
+{
+ (void) self; /* unused */
+
+ return 0;
+}
+
+static void TestCase_dealloc(TestCaseObject *self)
+{
+ Py_TYPE(self)->tp_free((PyObject *)self);
+}
+
+static int TestCase_traverse(TestCaseObject *self,
+ visitproc visit,
+ void *arg)
+{
+ (void) self; /* unused */
+ (void) visit; /* unused */
+ (void) arg; /* unused */
+
+ return 0;
+}
+
+static int TestCase_init(TestCaseObject *self,
+ PyObject *args,
+ PyObject *kwargs)
+{
+ const char * const kwlist[] = { "pam_operation",
+ "expected_rv",
+ "flags",
+ NULL };
+ int pam_operation = -1;
+ int expected_rv = PAM_SUCCESS;
+ int flags = 0;
+ int ok;
+
+ ok = PyArg_ParseTupleAndKeywords(args,
+ kwargs,
+ "i|ii",
+ discard_const_p(char *, kwlist),
+ &pam_operation,
+ &expected_rv,
+ &flags);
+ if (!ok) {
+ return -1;
+ }
+
+ switch (pam_operation) {
+ case PAMTEST_AUTHENTICATE:
+ case PAMTEST_SETCRED:
+ case PAMTEST_ACCOUNT:
+ case PAMTEST_OPEN_SESSION:
+ case PAMTEST_CLOSE_SESSION:
+ case PAMTEST_CHAUTHTOK:
+ case PAMTEST_GETENVLIST:
+ case PAMTEST_KEEPHANDLE:
+ break;
+ default:
+ PyErr_Format(PyExc_ValueError,
+ "Unsupported PAM operation %d",
+ pam_operation);
+ return -1;
+ }
+
+ self->flags = flags;
+ self->expected_rv = expected_rv;
+ self->pam_operation = pam_operation;
+
+ return 0;
+}
+
+/*
+ * This function returns string representation of the object, but one that
+ * can be parsed by a machine.
+ *
+ * str() is also string represtentation, but just human-readable.
+ */
+static PyObject *TestCase_repr(TestCaseObject *self)
+{
+ return PyUnicode_FromFormat(REPR_FMT,
+ self->pam_operation,
+ self->expected_rv,
+ self->flags);
+}
+
+static PyMemberDef pypamtest_test_case_members[] = {
+ {
+ discard_const_p(char, "pam_operation"),
+ T_INT,
+ offsetof(TestCaseObject, pam_operation),
+ READONLY,
+ discard_const_p(char, "The PAM operation to run"),
+ },
+
+ {
+ discard_const_p(char, "expected_rv"),
+ T_INT,
+ offsetof(TestCaseObject, expected_rv),
+ READONLY,
+ discard_const_p(char, "The expected PAM return code"),
+ },
+
+ {
+ discard_const_p(char, "flags"),
+ T_INT,
+ offsetof(TestCaseObject, flags),
+ READONLY,
+ discard_const_p(char, "Additional flags for the PAM operation"),
+ },
+
+ {
+ discard_const_p(char, "pam_handle"),
+ T_OBJECT_EX,
+ offsetof(TestCaseObject, pam_handle),
+ READONLY,
+ discard_const_p(char, "Pam handle"),
+ },
+
+ {
+ discard_const_p(char, "pam_env"),
+ T_OBJECT_EX,
+ offsetof(TestCaseObject, pam_env),
+ READONLY,
+ discard_const_p(char, "Pam env"),
+ },
+
+ { NULL, 0, 0, 0, NULL } /* Sentinel */
+};
+
+static PyTypeObject pypamtest_test_case = {
+ PyVarObject_HEAD_INIT(NULL, 0)
+ .tp_name = "pypamtest.TestCase",
+ .tp_basicsize = sizeof(TestCaseObject),
+ .tp_new = TestCase_new,
+ .tp_dealloc = (destructor) TestCase_dealloc,
+ .tp_traverse = (traverseproc) TestCase_traverse,
+ .tp_clear = (inquiry) TestCase_clear,
+ .tp_init = (initproc) TestCase_init,
+ .tp_repr = (reprfunc) TestCase_repr,
+ .tp_members = pypamtest_test_case_members,
+ .tp_flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_GC,
+ .tp_doc = TestCaseObject__doc__
+};
+
+PyDoc_STRVAR(TestResultObject__doc__,
+"pamtest test result\n\n"
+"The test result object is returned from run_pamtest on success. It contains"
+"two lists of strings (up to 16 strings each) which contain the info and error"
+"messages the PAM conversation printed\n\n"
+"Attributes:\n"
+"errors: PAM_ERROR_MSG-level messages printed during the PAM conversation\n"
+"info: PAM_TEXT_INFO-level messages printed during the PAM conversation\n"
+);
+
+typedef struct {
+ PyObject_HEAD
+
+ PyObject *info_msg_list;
+ PyObject *error_msg_list;
+} TestResultObject;
+
+static PyObject *
+TestResult_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
+{
+ TestResultObject *self;
+
+ (void) args; /* unused */
+ (void) kwds; /* unused */
+
+ self = (TestResultObject *)type->tp_alloc(type, 0);
+ if (self == NULL) {
+ PyErr_NoMemory();
+ return NULL;
+ }
+
+ return (PyObject *) self;
+}
+
+static int TestResult_clear(TestResultObject *self)
+{
+ (void) self; /* unused */
+
+ return 0;
+}
+
+static void TestResult_dealloc(TestResultObject *self)
+{
+ Py_TYPE(self)->tp_free((PyObject *)self);
+}
+
+static int TestResult_traverse(TestResultObject *self,
+ visitproc visit,
+ void *arg)
+{
+ (void) self; /* unused */
+ (void) visit; /* unused */
+ (void) arg; /* unused */
+
+ return 0;
+}
+
+static int TestResult_init(TestResultObject *self,
+ PyObject *args,
+ PyObject *kwargs)
+{
+ const char * const kwlist[] = { "info_msg_list",
+ "error_msg_list",
+ NULL };
+ int ok;
+ PyObject *py_info_list = NULL;
+ PyObject *py_err_list = NULL;
+
+ ok = PyArg_ParseTupleAndKeywords(args,
+ kwargs,
+ "|OO",
+ discard_const_p(char *, kwlist),
+ &py_info_list,
+ &py_err_list);
+ if (!ok) {
+ return -1;
+ }
+
+ if (py_info_list) {
+ ok = PySequence_Check(py_info_list);
+ if (!ok) {
+ PyErr_Format(PyExc_TypeError,
+ "List of info messages must be a sequence\n");
+ return -1;
+ }
+
+ self->info_msg_list = py_info_list;
+ Py_XINCREF(py_info_list);
+ } else {
+ self->info_msg_list = PyList_New(0);
+ if (self->info_msg_list == NULL) {
+ PyErr_NoMemory();
+ return -1;
+ }
+ }
+
+ if (py_err_list) {
+ ok = PySequence_Check(py_err_list);
+ if (!ok) {
+ PyErr_Format(PyExc_TypeError,
+ "List of error messages must be a sequence\n");
+ return -1;
+ }
+
+ self->error_msg_list = py_err_list;
+ Py_XINCREF(py_err_list);
+ } else {
+ self->error_msg_list = PyList_New(0);
+ if (self->error_msg_list == NULL) {
+ PyErr_NoMemory();
+ return -1;
+ }
+ }
+
+ return 0;
+}
+
+static PyObject *test_result_list_concat(PyObject *list,
+ const char delim_pre,
+ const char delim_post)
+{
+ PyObject *res;
+ PyObject *item;
+ Py_ssize_t size;
+ Py_ssize_t i;
+
+ res = PyUnicode_FromString("");
+ if (res == NULL) {
+ return NULL;
+ }
+
+ size = PySequence_Size(list);
+
+ for (i=0; i < size; i++) {
+ item = PySequence_GetItem(list, i);
+ if (item == NULL) {
+ PyMem_Free(res);
+ return NULL;
+ }
+
+#if IS_PYTHON3
+ res = PyUnicode_FromFormat("%U%c%U%c",
+ res, delim_pre, item, delim_post);
+#else
+ res = PyUnicode_FromFormat("%U%c%s%c",
+ res,
+ delim_pre,
+ PyString_AsString(item),
+ delim_post);
+#endif
+ Py_XDECREF(item);
+ }
+
+ return res;
+}
+
+static PyObject *TestResult_repr(TestResultObject *self)
+{
+ PyObject *u_info = NULL;
+ PyObject *u_error = NULL;
+ PyObject *res = NULL;
+
+ u_info = test_result_list_concat(self->info_msg_list, '{', '}');
+ u_error = test_result_list_concat(self->info_msg_list, '{', '}');
+ if (u_info == NULL || u_error == NULL) {
+ Py_XDECREF(u_error);
+ Py_XDECREF(u_info);
+ return NULL;
+ }
+
+ res = PyUnicode_FromFormat("{ errors: { %U } infos: { %U } }",
+ u_info, u_error);
+ Py_DECREF(u_error);
+ Py_DECREF(u_info);
+ return res;
+}
+
+static PyMemberDef pypamtest_test_result_members[] = {
+ {
+ discard_const_p(char, "errors"),
+ T_OBJECT_EX,
+ offsetof(TestResultObject, error_msg_list),
+ READONLY,
+ discard_const_p(char,
+ "List of error messages from PAM conversation"),
+ },
+
+ {
+ discard_const_p(char, "info"),
+ T_OBJECT_EX,
+ offsetof(TestResultObject, info_msg_list),
+ READONLY,
+ discard_const_p(char,
+ "List of info messages from PAM conversation"),
+ },
+
+ { NULL, 0, 0, 0, NULL } /* Sentinel */
+};
+
+static PyTypeObject pypamtest_test_result = {
+ PyVarObject_HEAD_INIT(NULL, 0)
+ .tp_name = "pypamtest.TestResult",
+ .tp_basicsize = sizeof(TestResultObject),
+ .tp_new = TestResult_new,
+ .tp_dealloc = (destructor) TestResult_dealloc,
+ .tp_traverse = (traverseproc) TestResult_traverse,
+ .tp_clear = (inquiry) TestResult_clear,
+ .tp_init = (initproc) TestResult_init,
+ .tp_repr = (reprfunc) TestResult_repr,
+ .tp_members = pypamtest_test_result_members,
+ .tp_flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_GC,
+ .tp_doc = TestResultObject__doc__
+};
+
+/**********************************************************
+ *** Methods of the module
+ **********************************************************/
+
+static TestResultObject *construct_test_conv_result(char **msg_info, char **msg_err)
+{
+ PyObject *py_msg_info = NULL;
+ PyObject *py_msg_err = NULL;
+ TestResultObject *result = NULL;
+ PyObject *result_args = NULL;
+ int rc;
+
+ py_msg_info = string_list_as_tuple(msg_info);
+ py_msg_err = string_list_as_tuple(msg_err);
+ if (py_msg_info == NULL || py_msg_err == NULL) {
+ /* The exception is raised in string_list_as_tuple() */
+ Py_XDECREF(py_msg_err);
+ Py_XDECREF(py_msg_info);
+ return NULL;
+ }
+
+ result = (TestResultObject *) TestResult_new(&pypamtest_test_result,
+ NULL,
+ NULL);
+ if (result == NULL) {
+ /* The exception is raised in TestResult_new */
+ Py_XDECREF(py_msg_err);
+ Py_XDECREF(py_msg_info);
+ return NULL;
+ }
+
+ result_args = PyTuple_New(2);
+ if (result_args == NULL) {
+ /* The exception is raised in TestResult_new */
+ Py_XDECREF(result);
+ Py_XDECREF(py_msg_err);
+ Py_XDECREF(py_msg_info);
+ return NULL;
+ }
+
+ /* Brand new tuples with fixed size don't need error checking */
+ PyTuple_SET_ITEM(result_args, 0, py_msg_info);
+ PyTuple_SET_ITEM(result_args, 1, py_msg_err);
+
+ rc = TestResult_init(result, result_args, NULL);
+ Py_XDECREF(result_args);
+ if (rc != 0) {
+ Py_XDECREF(result);
+ return NULL;
+ }
+
+ return result;
+}
+
+static int py_testcase_get(PyObject *py_test,
+ const char *member_name,
+ long *_value)
+{
+ PyObject* item = NULL;
+
+ /*
+ * PyPyObject_GetAttrString() increases the refcount on the
+ * returned value.
+ */
+ item = PyObject_GetAttrString(py_test, member_name);
+ if (item == NULL) {
+ return EINVAL;
+ }
+
+ *_value = PyLong_AsLong(item);
+ Py_DECREF(item);
+
+ return 0;
+}
+
+static int py_testcase_to_cstruct(PyObject *py_test, struct pam_testcase *test)
+{
+ int rc;
+ long value;
+
+ memset(test, 0, sizeof(struct pam_testcase));
+
+ rc = py_testcase_get(py_test, "pam_operation", &value);
+ if (rc != 0) {
+ return rc;
+ }
+ test->pam_operation = value;
+
+ rc = py_testcase_get(py_test, "expected_rv", &value);
+ if (rc != 0) {
+ return rc;
+ }
+ test->expected_rv = value;
+
+ rc = py_testcase_get(py_test, "flags", &value);
+ if (rc != 0) {
+ return rc;
+ }
+ test->flags = value;
+
+ return 0;
+}
+
+static void free_conv_data(struct pamtest_conv_data *conv_data)
+{
+ if (conv_data == NULL) {
+ return;
+ }
+
+ free_string_list(conv_data->out_err);
+ free_string_list(conv_data->out_info);
+ free_cstring_list(conv_data->in_echo_on);
+ free_cstring_list(conv_data->in_echo_off);
+}
+
+/* conv_data must be a pointer to allocated conv_data structure.
+ *
+ * Use free_conv_data() to free the contents.
+ */
+static int fill_conv_data(PyObject *py_echo_off,
+ PyObject *py_echo_on,
+ struct pamtest_conv_data *conv_data)
+{
+ size_t conv_count = 0;
+ size_t count = 0;
+ int rc;
+
+ conv_data->in_echo_on = NULL;
+ conv_data->in_echo_off = NULL;
+ conv_data->out_err = NULL;
+ conv_data->out_info = NULL;
+
+ if (py_echo_off != NULL) {
+ rc = sequence_as_string_list(py_echo_off,
+ "echo_off",
+ &conv_data->in_echo_off,
+ &count);
+ if (rc != 0) {
+ free_conv_data(conv_data);
+ return ENOMEM;
+ }
+ conv_count += count;
+ }
+
+ if (py_echo_on != NULL) {
+ rc = sequence_as_string_list(py_echo_on,
+ "echo_on",
+ &conv_data->in_echo_on,
+ &count);
+ if (rc != 0) {
+ free_conv_data(conv_data);
+ return ENOMEM;
+ }
+ conv_count += count;
+ }
+
+ if (conv_count > PAM_CONV_MSG_MAX) {
+ free_conv_data(conv_data);
+ return ENOMEM;
+ }
+
+ conv_data->out_info = new_conv_list(PAM_CONV_MSG_MAX);
+ conv_data->out_err = new_conv_list(PAM_CONV_MSG_MAX);
+ if (conv_data->out_info == NULL || conv_data->out_err == NULL) {
+ free_conv_data(conv_data);
+ return ENOMEM;
+ }
+
+ return 0;
+}
+
+/* test_list is allocated using PyMem_New and must be freed accordingly.
+ * Returns errno that should be handled into exception in the caller
+ */
+static int py_tc_list_to_cstruct_list(PyObject *py_test_list,
+ Py_ssize_t num_tests,
+ struct pam_testcase **_test_list)
+{
+ Py_ssize_t i;
+ PyObject *py_test;
+ int rc;
+ struct pam_testcase *test_list;
+
+ test_list = PyMem_New(struct pam_testcase,
+ num_tests * sizeof(struct pam_testcase));
+ if (test_list == NULL) {
+ return ENOMEM;
+ }
+
+ for (i = 0; i < num_tests; i++) {
+ /*
+ * PySequence_GetItem() increases the refcount on the
+ * returned value
+ */
+ py_test = PySequence_GetItem(py_test_list, i);
+ if (py_test == NULL) {
+ PyMem_Free(test_list);
+ return EIO;
+ }
+
+ rc = py_testcase_to_cstruct(py_test, &test_list[i]);
+ Py_DECREF(py_test);
+ if (rc != 0) {
+ PyMem_Free(test_list);
+ return EIO;
+ }
+ }
+
+ *_test_list = test_list;
+ return 0;
+}
+
+static int cstruct_to_py_testcase(PyObject *pytest, struct pam_testcase *ctest)
+{
+ TestCaseObject *t = PyTestCase_AsTestCaseObject(pytest);
+ size_t i;
+ int rc;
+
+ switch (t->pam_operation) {
+ case PAMTEST_GETENVLIST:
+ if (ctest->case_out.envlist == NULL) {
+ break;
+ }
+
+ t->pam_env = PyDict_New();
+ if (t->pam_env == NULL) {
+ return ENOMEM;
+ }
+ for (i = 0; ctest->case_out.envlist[i] != NULL; i++) {
+ char *key = NULL;
+ char *val = NULL;
+ key = strdup(ctest->case_out.envlist[i]);
+ if (key == NULL) {
+ return ENOMEM;
+ }
+ val = strrchr(key, '=');
+ if (val == NULL) {
+ PyErr_Format(PyExc_IOError,
+ "Failed to parse PAM environment "
+ "variable");
+ free(key);
+ return EINVAL;
+ }
+ *val = '\0';
+ rc = PyDict_SetItem(t->pam_env,
+ PyUnicode_FromString(key),
+ PyUnicode_FromString(val + 1));
+ free(key);
+ if (rc == -1) {
+ return rc;
+ }
+ }
+ break;
+ case PAMTEST_KEEPHANDLE:
+ t->pam_handle = PyCapsule_New(ctest->case_out.ph, NULL, NULL);
+ if (t->pam_handle == NULL) {
+ return ENOMEM;
+ }
+ break;
+ default:
+ break;
+ }
+
+ return 0;
+}
+
+static int cstruct_list_to_py_tc_list(PyObject *py_test_list,
+ Py_ssize_t num_tests,
+ struct pam_testcase *test_list)
+{
+ Py_ssize_t i;
+ PyObject *py_test = NULL;
+ int rc;
+
+ for (i = 0; i < num_tests; i++) {
+ py_test = PySequence_GetItem(py_test_list, i);
+ if (py_test == NULL) {
+ return EIO;
+ }
+
+ rc = cstruct_to_py_testcase(py_test, &test_list[i]);
+ Py_DECREF(py_test);
+ if (rc != 0) {
+ return EIO;
+ }
+ }
+
+ return 0;
+}
+
+PyDoc_STRVAR(RunPamTest__doc__,
+"Run PAM tests\n\n"
+"This function runs PAM test cases and reports result\n"
+"Parameters:\n"
+"service: The PAM service to use in the conversation (string)\n"
+"username: The user to run PAM conversation as\n"
+"test_list: Sequence of pypamtest.TestCase objects\n"
+"echo_off_list: Sequence of strings that will be used by PAM "
+"conversation for PAM_PROMPT_ECHO_OFF input. These are typically "
+"passwords.\n"
+"echo_on_list: Sequence of strings that will be used by PAM "
+"conversation for PAM_PROMPT_ECHO_ON input.\n"
+);
+
+static PyObject *pypamtest_run_pamtest(PyObject *module,
+ PyObject *args,
+ PyObject *kwargs)
+{
+ int ok;
+ int rc;
+ char *username = NULL;
+ char *service = NULL;
+ PyObject *py_test_list;
+ PyObject *py_echo_off = NULL;
+ PyObject *py_echo_on = NULL;
+ PyObject *py_pam_handle = NULL;
+ Py_ssize_t num_tests;
+ struct pam_testcase *test_list;
+ enum pamtest_err perr;
+ struct pamtest_conv_data conv_data;
+ pam_handle_t *pam_handle = NULL;
+ TestResultObject *result = NULL;
+ const char * const kwnames[] = { "username",
+ "service",
+ "tests",
+ "echo_off",
+ "echo_on",
+ "handle",
+ NULL };
+
+ (void) module; /* unused */
+
+ ok = PyArg_ParseTupleAndKeywords(args,
+ kwargs,
+ discard_const_p(char, "ssO|OOO"),
+ discard_const_p(char *, kwnames),
+ &username,
+ &service,
+ &py_test_list,
+ &py_echo_off,
+ &py_echo_on,
+ &py_pam_handle);
+ if (!ok) {
+ return NULL;
+ }
+
+ ok = PySequence_Check(py_test_list);
+ if (!ok) {
+ PyErr_Format(PyExc_TypeError, "tests must be a sequence");
+ return NULL;
+ }
+
+ num_tests = PySequence_Size(py_test_list);
+ if (num_tests == -1) {
+ PyErr_Format(PyExc_IOError, "Cannot get sequence length");
+ return NULL;
+ }
+
+ rc = py_tc_list_to_cstruct_list(py_test_list, num_tests, &test_list);
+ if (rc != 0) {
+ if (rc == ENOMEM) {
+ PyErr_NoMemory();
+ return NULL;
+ } else {
+ PyErr_Format(PyExc_IOError,
+ "Cannot convert test to C structure");
+ return NULL;
+ }
+ }
+
+ rc = fill_conv_data(py_echo_off, py_echo_on, &conv_data);
+ if (rc != 0) {
+ PyMem_Free(test_list);
+ PyErr_NoMemory();
+ return NULL;
+ }
+
+ if (py_pam_handle != NULL) {
+ pam_handle = (pam_handle_t *)PyCapsule_GetPointer(py_pam_handle,
+ NULL);
+ if (pam_handle == NULL) {
+ PyMem_Free(test_list);
+ PyErr_Format(PyExc_IOError,
+ "Failed to get the pam handle pointer");
+ return NULL;
+ }
+ }
+
+ perr = _pamtest(service,
+ username,
+ &conv_data,
+ test_list,
+ num_tests,
+ pam_handle);
+ if (perr != PAMTEST_ERR_OK) {
+ free_conv_data(&conv_data);
+ set_pypamtest_exception(PyExc_PamTestError,
+ perr,
+ test_list,
+ num_tests);
+ PyMem_Free(test_list);
+ return NULL;
+ }
+
+ rc = cstruct_list_to_py_tc_list(py_test_list, num_tests, test_list);
+ if (rc != 0) {
+ if (rc == ENOMEM) {
+ PyErr_NoMemory();
+ return NULL;
+ } else {
+ PyErr_Format(PyExc_IOError,
+ "Cannot convert C structure to python");
+ return NULL;
+ }
+ }
+ PyMem_Free(test_list);
+
+ result = construct_test_conv_result(conv_data.out_info,
+ conv_data.out_err);
+ free_conv_data(&conv_data);
+ if (result == NULL) {
+ PyMem_Free(test_list);
+ return NULL;
+ }
+
+ return (PyObject *)result;
+}
+
+static PyMethodDef pypamtest_module_methods[] = {
+ {
+ discard_const_p(char, "run_pamtest"),
+ (PyCFunction) pypamtest_run_pamtest,
+ METH_VARARGS | METH_KEYWORDS,
+ RunPamTest__doc__,
+ },
+
+ { NULL, NULL, 0, NULL } /* Sentinel */
+};
+
+/*
+ * This is the module structure describing the module and
+ * to define methods
+ */
+#if IS_PYTHON3
+static struct PyModuleDef pypamtestdef = {
+ .m_base = PyModuleDef_HEAD_INIT,
+ .m_name = PYTHON_MODULE_NAME,
+ .m_size = -1,
+ .m_methods = pypamtest_module_methods,
+};
+#endif
+
+/**********************************************************
+ *** Initialize the module
+ **********************************************************/
+
+#if PY_VERSION_HEX >= 0x02070000 /* >= 2.7.0 */
+PyDoc_STRVAR(PamTestError__doc__,
+"pypamtest specific exception\n\n"
+"This exception is raised if the _pamtest() function fails. If _pamtest() "
+"returns PAMTEST_ERR_CASE (a test case returns unexpected error code), then "
+"the exception also details which test case failed."
+);
+#endif
+
+#if IS_PYTHON3
+PyMODINIT_FUNC PyInit_pypamtest(void)
+#else
+PyMODINIT_FUNC initpypamtest(void)
+#endif
+{
+ PyObject *m;
+ union {
+ PyTypeObject *type_obj;
+ PyObject *obj;
+ } pypam_object;
+ int ret;
+
+#if IS_PYTHON3
+ m = PyModule_Create(&pypamtestdef);
+ if (m == NULL) {
+ RETURN_ON_ERROR;
+ }
+#else
+ m = Py_InitModule(discard_const_p(char, PYTHON_MODULE_NAME),
+ pypamtest_module_methods);
+#endif
+
+#if PY_VERSION_HEX >= 0x02070000 /* >= 2.7.0 */
+ PyExc_PamTestError = PyErr_NewExceptionWithDoc(discard_const_p(char, "pypamtest.PamTestError"),
+ PamTestError__doc__,
+ PyExc_EnvironmentError,
+ NULL);
+#else /* < 2.7.0 */
+ PyExc_PamTestError = PyErr_NewException(discard_const_p(char, "pypamtest.PamTestError"),
+ PyExc_EnvironmentError,
+ NULL);
+#endif
+
+ if (PyExc_PamTestError == NULL) {
+ RETURN_ON_ERROR;
+ }
+
+ Py_INCREF(PyExc_PamTestError);
+ ret = PyModule_AddObject(m, discard_const_p(char, "PamTestError"),
+ PyExc_PamTestError);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+
+ ret = PyModule_AddIntMacro(m, PAMTEST_AUTHENTICATE);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+ ret = PyModule_AddIntMacro(m, PAMTEST_SETCRED);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+ ret = PyModule_AddIntMacro(m, PAMTEST_ACCOUNT);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+ ret = PyModule_AddIntMacro(m, PAMTEST_OPEN_SESSION);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+ ret = PyModule_AddIntMacro(m, PAMTEST_CLOSE_SESSION);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+ ret = PyModule_AddIntMacro(m, PAMTEST_CHAUTHTOK);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+
+ ret = PyModule_AddIntMacro(m, PAMTEST_GETENVLIST);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+ ret = PyModule_AddIntMacro(m, PAMTEST_KEEPHANDLE);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+
+ ret = PyModule_AddIntConstant(m,
+ "PAMTEST_FLAG_DELETE_CRED",
+ PAM_DELETE_CRED);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+
+ ret = PyModule_AddIntConstant(m,
+ "PAMTEST_FLAG_ESTABLISH_CRED",
+ PAM_ESTABLISH_CRED);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+
+ ret = PyModule_AddIntConstant(m,
+ "PAMTEST_FLAG_REINITIALIZE_CRED",
+ PAM_REINITIALIZE_CRED);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+
+ ret = PyModule_AddIntConstant(m,
+ "PAMTEST_FLAG_REFRESH_CRED",
+ PAM_REFRESH_CRED);
+ if (ret == -1) {
+ RETURN_ON_ERROR;
+ }
+
+ pypam_object.type_obj = &pypamtest_test_case;
+ if (PyType_Ready(pypam_object.type_obj) < 0) {
+ RETURN_ON_ERROR;
+ }
+ Py_INCREF(pypam_object.obj);
+ PyModule_AddObject(m, "TestCase", pypam_object.obj);
+
+ pypam_object.type_obj = &pypamtest_test_result;
+ if (PyType_Ready(pypam_object.type_obj) < 0) {
+ RETURN_ON_ERROR;
+ }
+ Py_INCREF(pypam_object.obj);
+ PyModule_AddObject(m, "TestResult", pypam_object.obj);
+
+#if IS_PYTHON3
+ return m;
+#endif
+}