Visitar URL original
feat(ldapobject): add asynchronous whoami() extended operation by spaceone · Pull Request #680 · python-ldap/python-ldap · GitHub
Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions Doc/reference/ldap.rst
Original file line number Diff line number Diff line change
Expand Up @@ -1404,6 +1404,15 @@ and wait for and return with the server's result, or with
These methods are all synchronous in nature.


.. py:method:: LDAPObject.whoami([serverctrls=None [, clientctrls=None]]) -> int

Send the LDAP "Who Am I?" extended operation asynchronously and return
its message ID. Use :py:meth:`extop_result()` with the message ID to
retrieve the response.
Comment thread
spaceone marked this conversation as resolved.
Resolved

*serverctrls* and *clientctrls* like described in section :ref:`ldap-controls`.


.. py:method:: LDAPObject.whoami_s() -> string

This synchronous method implements the LDAP "Who Am I?"
Expand Down
34 changes: 34 additions & 0 deletions Lib/ldap/extop/whoami.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
"""
Classes for "Who am I?" extended operation (see RFC 4532)

See https://www.python-ldap.org/ for details.
"""

from __future__ import annotations

from . import ExtendedRequest, ExtendedResponse


class WhoAmIRequest(ExtendedRequest):
"""
Who Am I? extended request
"""

requestName: str = '1.3.6.1.4.1.4203.1.11.3'


class WhoAmIResponse(ExtendedResponse):
"""
Who Am I? extended response
"""

responseName = None

def decodeResponseValue(self, value: bytes | None) -> str | None:
self.dn = None
raw = value.decode('UTF-8') if value is not None else None
if raw and raw.startswith('dn:'):
self.dn = raw.removeprefix('dn:')
elif raw and raw.startswith('u:'):
self.userid = raw.removeprefix('u:')
return raw
17 changes: 17 additions & 0 deletions Lib/ldap/ldapobject.py
Original file line number Diff line number Diff line change
Expand Up @@ -1031,6 +1031,23 @@ def whoami_s(
) -> str:
return self._ldap_call(self._l.whoami_s, serverctrls, clientctrls) # type: ignore[no-any-return]

def whoami(
self,
serverctrls: list[RequestControl] | None = None,
clientctrls: list[RequestControl] | None = None,
) -> int:
"""
whoami([serverctrls=None[,clientctrls=None]]) -> int

Send a Who Am I? extended operation asynchronously. Use the returned
message ID with :meth:`extop_result` to retrieve the response.
"""
return self._ldap_call( # type: ignore[no-any-return]
self._l.whoami,
RequestControlTuples(serverctrls),
RequestControlTuples(clientctrls),
)

def get_option(self, option: int) -> Any:
result = self._ldap_call(self._l.get_option, option)
if option == ldap.OPT_SERVER_CONTROLS or option == ldap.OPT_CLIENT_CONTROLS:
Expand Down
43 changes: 43 additions & 0 deletions Modules/LDAPObject.c
Original file line number Diff line number Diff line change
Expand Up @@ -1313,6 +1313,48 @@ l_ldap_whoami_s(LDAPObject *self, PyObject *args)
return result;
}

static PyObject *
l_ldap_whoami(LDAPObject *self, PyObject *args)
{
PyObject *serverctrls = Py_None;
PyObject *clientctrls = Py_None;
LDAPControl **server_ldcs = NULL;
LDAPControl **client_ldcs = NULL;
int msgid;
int ldaperror;

if (!PyArg_ParseTuple(args, "|OO:whoami", &serverctrls, &clientctrls))
return NULL;
if (not_valid(self))
return NULL;

if (!Py_IsNone(serverctrls)) {
if (!LDAPControls_from_object(serverctrls, &server_ldcs))
return NULL;
}

if (!Py_IsNone(clientctrls)) {
if (!LDAPControls_from_object(clientctrls, &client_ldcs)) {
LDAPControl_List_DEL(server_ldcs);
return NULL;
}
}

LDAP_BEGIN_ALLOW_THREADS(self);
ldaperror = ldap_whoami(self->ldap, server_ldcs, client_ldcs, &msgid);
LDAP_END_ALLOW_THREADS(self);

LDAPControl_List_DEL(server_ldcs);
LDAPControl_List_DEL(client_ldcs);

if (ldaperror != LDAP_SUCCESS) {
PyObject *module = PyType_GetModuleByDef(Py_TYPE((PyObject *)self), LDAPMod_moduledef);
return LDAPerror(module, self->ldap);
}

return PyLong_FromLong(msgid);
}

#ifdef HAVE_TLS
/* ldap_start_tls_s */

Expand Down Expand Up @@ -1530,6 +1572,7 @@ static PyMethodDef l_ldap_methods[] = {
#ifdef HAVE_TLS
{"start_tls_s", (PyCFunction)l_ldap_start_tls_s, METH_VARARGS},
#endif
{"whoami", (PyCFunction)l_ldap_whoami, METH_VARARGS},
{"whoami_s", (PyCFunction)l_ldap_whoami_s, METH_VARARGS},
{"passwd", (PyCFunction)l_ldap_passwd, METH_VARARGS},
{"set_option", (PyCFunction)l_ldap_set_option, METH_VARARGS},
Expand Down
9 changes: 8 additions & 1 deletion Tests/t_cext.py
Original file line number Diff line number Diff line change
Expand Up @@ -687,9 +687,15 @@ def test_rename(self):
self.assertEqual(pmsg[0][1]['cn'], [b'IAmRenamedAgain'])

def test_whoami(self):
authzid = 'dn:' + self.server.root_dn
l = self._open_conn()
r = l.whoami_s()
self.assertEqual('dn:' + self.server.root_dn, r)
self.assertEqual(authzid, r)

m = l.whoami()
_type, _data, _msgid, _ctrls, oid, value = l.result4(m, _ldap.MSG_ALL, self.timeout, 0, 0, 1)
self.assertIsNone(oid)
self.assertEqual(value, authzid.encode('UTF-8'))

def test_whoami_unbound(self):
l = self._open_conn(bind=False)
Expand Down Expand Up @@ -863,6 +869,7 @@ def test_invalid_controls(self):
l = self._open_conn()
self.assertInvalidControls(l.simple_bind, '', '')
self.assertInvalidControls(l.whoami_s)
self.assertInvalidControls(l.whoami)
self.assertInvalidControls(l.passwd, 'dn', 'initial', 'changed')
self.assertInvalidControls(l.add_ext, 'dn', [('cn', b'cn')])
self.assertInvalidControls(l.modify_ext, 'dn', [(_ldap.MOD_ADD, 'attr', [b'value'])])
Expand Down
14 changes: 14 additions & 0 deletions Tests/t_ldapobject.py
Original file line number Diff line number Diff line change
Expand Up @@ -458,6 +458,20 @@ def test_async_search_no_such_object_exception_contains_message_id(self):
self._ldap_conn.result()
self.assertEqual(cm.exception.args[0]['msgid'], msgid)

def test_async_whoami(self):
from ldap.extop.whoami import WhoAmIResponse

authz_id = 'dn:' + self.server.root_dn

l = self.ldap_object_class(self.server.ldap_uri)
l.simple_bind_s(self.server.root_dn, self.server.root_pw)
msgid = l.whoami()
_oid, value = l.extop_result(msgid)
self.assertEqual(value.decode('UTF-8').lower(), authz_id.lower())

res = WhoAmIResponse(_oid, value)
self.assertEqual(res.dn, authz_id[3:])

def test_passwd_s(self):
l = self._ldap_conn

Expand Down
Loading