diff --git a/Lib/ldap/dn.py b/Lib/ldap/dn.py index a9d96846..dd7278b6 100644 --- a/Lib/ldap/dn.py +++ b/Lib/ldap/dn.py @@ -48,12 +48,17 @@ def str2dn(dn,flags=0): return ldap.functions._ldap_function_call(None,_ldap.str2dn,dn,flags) -def dn2str(dn): +def dn2str(dn, flags=0): """ This function takes a decomposed DN as parameter and returns - a single string. It's the inverse to str2dn() but will always - return a DN in LDAPv3 format compliant to RFC 4514. + a single string. It's the inverse to str2dn() but will by default always + return a DN in LDAPv3 format compliant to RFC 4514 if not otherwise specified + via flags. + + See also the OpenLDAP man-page ldap_dn2str(3) """ + if flags: + return ldap.functions._ldap_function_call(None, _ldap.dn2str, dn, flags) return ','.join([ '+'.join([ '='.join((atype,escape_dn_chars(avalue or ''))) @@ -61,6 +66,7 @@ def dn2str(dn): for rdn in dn ]) + def explode_dn(dn, notypes=False, flags=0): """ explode_dn(dn [, notypes=False [, flags=0]]) -> list @@ -116,3 +122,8 @@ def is_dn(s,flags=0): return False else: return True + + +def normalize(s, flags=0): + """Returns a normalized distinguished name (DN)""" + return dn2str(str2dn(s, flags), flags) diff --git a/Modules/functions.c b/Modules/functions.c index 9a977ff7..3f7f7eca 100644 --- a/Modules/functions.c +++ b/Modules/functions.c @@ -155,6 +155,222 @@ l_ldap_str2dn(PyObject *unused, PyObject *args) return result; } +/* ldap_dn2str */ + +static void +_free_dn_structure(LDAPDN dn) +{ + if (dn == NULL) + return; + + for (LDAPRDN *rdn = dn; *rdn != NULL; rdn++) { + for (LDAPAVA **avap = *rdn; *avap != NULL; avap++) { + LDAPAVA *ava = *avap; + + if (ava->la_attr.bv_val) { + free(ava->la_attr.bv_val); + } + if (ava->la_value.bv_val) { + free(ava->la_value.bv_val); + } + free(ava); + } + free(*rdn); + } + free(dn); +} + +/* + * Convert a Python list-of-list-of-(str, str, int) into an LDAPDN and + * call ldap_dn2bv to build a DN string. + * + * Python signature: dn2str(dn: list[list[tuple[str, str, int]]], flags: int) -> str + * Returns the DN string on success, or raises TypeError or RuntimeError on error. + */ +static PyObject * +l_ldap_dn2str(PyObject *self, PyObject *args) +{ + PyObject *dn_list = NULL; + int flags = 0; + LDAPDN dn = NULL; + LDAPAVA *ava; + LDAPAVA **rdn; + BerValue str = { 0, NULL }; + PyObject *py_rdn_seq = NULL, *py_ava_item = NULL; + PyObject *py_name = NULL, *py_value = NULL, *py_encoding = NULL; + PyObject *result = NULL; + Py_ssize_t nrdns = 0, navas = 0, name_len = 0, value_len = 0; + int i = 0, j = 0; + int ldap_err; + const char *name_utf8, *value_utf8; + + const char *type_error_message = "expected list[list[tuple[str, str, int]]]"; + + if (!PyArg_ParseTuple(args, "Oi:dn2str", &dn_list, &flags)) { + return NULL; + } + + if (!PySequence_Check(dn_list)) { + PyErr_SetString(PyExc_TypeError, type_error_message); + return NULL; + } + + nrdns = PySequence_Size(dn_list); + if (nrdns < 0) { + PyErr_SetString(PyExc_TypeError, type_error_message); + return NULL; + } + + /* Allocate array of LDAPRDN pointers (+1 for NULL terminator) */ + dn = (LDAPRDN *) calloc((size_t)nrdns + 1, sizeof(LDAPRDN)); + if (dn == NULL) { + PyErr_NoMemory(); + return NULL; + } + + for (i = 0; i < nrdns; i++) { + py_rdn_seq = PySequence_GetItem(dn_list, i); /* New reference */ + if (py_rdn_seq == NULL) { + goto error_cleanup; + } + if (!PySequence_Check(py_rdn_seq)) { + PyErr_SetString(PyExc_TypeError, type_error_message); + goto error_cleanup; + } + + navas = PySequence_Size(py_rdn_seq); + if (navas < 0) { + PyErr_SetString(PyExc_TypeError, type_error_message); + goto error_cleanup; + } + + /* Allocate array of LDAPAVA* pointers (+1 for NULL terminator) */ + rdn = (LDAPAVA **)calloc((size_t)navas + 1, sizeof(LDAPAVA *)); + if (rdn == NULL) { + PyErr_NoMemory(); + goto error_cleanup; + } + + for (j = 0; j < navas; j++) { + py_ava_item = PySequence_GetItem(py_rdn_seq, j); /* New reference */ + if (py_ava_item == NULL) { + goto error_cleanup; + } + /* Expect a 3‐tuple: (name: str, value: str, encoding: int) */ + if (!PyTuple_Check(py_ava_item) || PyTuple_Size(py_ava_item) != 3) { + PyErr_SetString(PyExc_TypeError, type_error_message); + goto error_cleanup; + } + + py_name = PyTuple_GetItem(py_ava_item, 0); /* Borrowed reference */ + py_value = PyTuple_GetItem(py_ava_item, 1); /* Borrowed reference */ + py_encoding = PyTuple_GetItem(py_ava_item, 2); /* Borrowed reference */ + + if (!PyUnicode_Check(py_name) || !PyUnicode_Check(py_value) || !PyLong_Check(py_encoding)) { + PyErr_SetString(PyExc_TypeError, type_error_message); + goto error_cleanup; + } + + name_len = 0; + value_len = 0; + name_utf8 = PyUnicode_AsUTF8AndSize(py_name, &name_len); + value_utf8 = PyUnicode_AsUTF8AndSize(py_value, &value_len); + if (name_utf8 == NULL || value_utf8 == NULL) { + goto error_cleanup; + } + + ava = (LDAPAVA *) calloc(1, sizeof(LDAPAVA)); + + if (ava == NULL) { + PyErr_NoMemory(); + goto error_cleanup; + } + + ava->la_attr.bv_val = (char *)malloc((size_t)name_len + 1); + if (ava->la_attr.bv_val == NULL) { + free(ava); + PyErr_NoMemory(); + goto error_cleanup; + } + memcpy(ava->la_attr.bv_val, name_utf8, (size_t)name_len); + ava->la_attr.bv_val[name_len] = '\0'; + ava->la_attr.bv_len = (ber_len_t) name_len; + + ava->la_value.bv_val = (char *)malloc((size_t)value_len + 1); + if (ava->la_value.bv_val == NULL) { + free(ava->la_attr.bv_val); + free(ava); + PyErr_NoMemory(); + goto error_cleanup; + } + memcpy(ava->la_value.bv_val, value_utf8, (size_t)value_len); + ava->la_value.bv_val[value_len] = '\0'; + ava->la_value.bv_len = (ber_len_t) value_len; + + ava->la_flags = (int)PyLong_AsLong(py_encoding); + if (PyErr_Occurred()) { + /* Encoding conversion failed */ + free(ava->la_attr.bv_val); + free(ava->la_value.bv_val); + free(ava); + goto error_cleanup; + } + + rdn[j] = ava; + Py_DECREF(py_ava_item); + py_ava_item = NULL; + } + + /* Null‐terminate the RDN */ + rdn[navas] = NULL; + + dn[i] = rdn; + Py_DECREF(py_rdn_seq); + py_rdn_seq = NULL; + } + + /* Null‐terminate the DN */ + dn[nrdns] = NULL; + + /* Call ldap_dn2bv to build a DN string */ + ldap_err = ldap_dn2bv(dn, &str, flags); + if (ldap_err != LDAP_SUCCESS) { + PyErr_SetString(PyExc_RuntimeError, ldap_err2string(ldap_err)); + goto error_cleanup; + } + + result = PyUnicode_FromString(str.bv_val); + if (result == NULL) { + goto error_cleanup; + } + + /* Free the memory allocated by ldap_dn2bv */ + ldap_memfree(str.bv_val); + str.bv_val = NULL; + + /* Free our local DN structure */ + _free_dn_structure(dn); + dn = NULL; + + return result; + + error_cleanup: + /* Free any partially built DN structure */ + _free_dn_structure(dn); + dn = NULL; + + /* If ldap_dn2bv allocated something, free it */ + if (str.bv_val) { + ldap_memfree(str.bv_val); + str.bv_val = NULL; + } + + /* Cleanup Python temporaries */ + Py_XDECREF(py_ava_item); + Py_XDECREF(py_rdn_seq); + return NULL; +} + /* ldap_set_option (global options) */ static PyObject * @@ -191,6 +407,7 @@ static PyMethodDef methods[] = { {"initialize_fd", (PyCFunction)l_ldap_initialize_fd, METH_VARARGS}, #endif {"str2dn", (PyCFunction)l_ldap_str2dn, METH_VARARGS}, + {"dn2str", (PyCFunction)l_ldap_dn2str, METH_VARARGS}, {"set_option", (PyCFunction)l_ldap_set_option, METH_VARARGS}, {"get_option", (PyCFunction)l_ldap_get_option, METH_VARARGS}, {NULL, NULL} diff --git a/Tests/t_ldap_dn.py b/Tests/t_ldap_dn.py index 86d36403..eafa22b3 100644 --- a/Tests/t_ldap_dn.py +++ b/Tests/t_ldap_dn.py @@ -40,23 +40,24 @@ def test_escape_dn_chars(self): test function escape_dn_chars() """ self.assertEqual(ldap.dn.escape_dn_chars('foobar'), 'foobar') - self.assertEqual(ldap.dn.escape_dn_chars('foo,bar'), 'foo\\,bar') - self.assertEqual(ldap.dn.escape_dn_chars('foo=bar'), 'foo\\=bar') + self.assertEqual(ldap.dn.escape_dn_chars('foo,bar'), r'foo\,bar') + self.assertEqual(ldap.dn.escape_dn_chars('foo=bar'), r'foo\=bar') self.assertEqual(ldap.dn.escape_dn_chars('foo#bar'), 'foo#bar') - self.assertEqual(ldap.dn.escape_dn_chars('#foobar'), '\\#foobar') + self.assertEqual(ldap.dn.escape_dn_chars('#foobar'), r'\#foobar') self.assertEqual(ldap.dn.escape_dn_chars('foo bar'), 'foo bar') - self.assertEqual(ldap.dn.escape_dn_chars(' foobar'), '\\ foobar') - self.assertEqual(ldap.dn.escape_dn_chars(' '), '\\ ') - self.assertEqual(ldap.dn.escape_dn_chars(' '), '\\ \\ ') - self.assertEqual(ldap.dn.escape_dn_chars('foobar '), 'foobar\\ ') + self.assertEqual(ldap.dn.escape_dn_chars(' foobar'), r'\ foobar') + self.assertEqual(ldap.dn.escape_dn_chars(' '), r'\ ') + self.assertEqual(ldap.dn.escape_dn_chars(' '), r'\ \ ') + self.assertEqual(ldap.dn.escape_dn_chars('foobar '), r'foobar\ ') self.assertEqual(ldap.dn.escape_dn_chars('f+o>o,bo\\,b\\