From c6d52f6bf71d495cd4e31137f1b0b1af12d08db2 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Tue, 18 Feb 2020 19:56:03 +0000 Subject: [PATCH 1/3] avoid duplicating ldaptor code in LDAPServerWithUPNBind --- docs/source/examples/ldaptor_with_upn_bind.py | 97 ++++++------------- 1 file changed, 28 insertions(+), 69 deletions(-) diff --git a/docs/source/examples/ldaptor_with_upn_bind.py b/docs/source/examples/ldaptor_with_upn_bind.py index dd859675..d9560d2f 100644 --- a/docs/source/examples/ldaptor_with_upn_bind.py +++ b/docs/source/examples/ldaptor_with_upn_bind.py @@ -17,86 +17,45 @@ userPrincipalName: bob@ad.example.org """ -from ldaptor import interfaces from ldaptor.protocols import pureldap -from ldaptor.protocols.ldap import distinguishedname, ldaperrors from twisted.internet import defer from ldaptor.protocols.ldap.ldapserver import LDAPServer -class LDAPServerWithUPNBind(LDAPServer): +class LDAPServerWithUPNBind(LDAPServer, object): """ An LDAP server which support BIND using UPN similar to AD. """ _loginAttribute = b"userPrincipalName" - def handle_LDAPBindRequest(self, request, controls, reply): - if request.version != 3: - raise ldaperrors.LDAPProtocolError( - "Version %u not supported" % request.version - ) - - self.checkControls(controls) - - if request.dn == b"": - # anonymous bind - self.boundUser = None - return pureldap.LDAPBindResponse(resultCode=0) - - root = interfaces.IConnectedLDAPEntry(self.factory) - - def _gotUPNResult(results): - if len(results) != 1: - # Not exactly one result, so this might not be an UNP. - return distinguishedname.DistinguishedName(request.dn) - - # A single result, so the UPN might exist. - return results[0].dn - - if b"@" in request.dn and b"," not in request.dn: + @defer.inlineCallbacks + def handle_LDAPBindRequest(self, request, *args, **kwargs): + @defer.inlineCallbacks + def _request(): + if not (b"@" in request.dn and b"," not in request.dn): + defer.returnValue(request) + root = interfaces.IConnectedLDAPEntry(self.factory) # This might be an UPN request. - filterText = b"(" + self._loginAttribute + b"=" + request.dn + b")" - d = root.search(filterText=filterText) - d.addCallback(_gotUPNResult) - else: - d = defer.succeed(distinguishedname.DistinguishedName(request.dn)) - - # Once the BIND DN is known, search for the LDAP entry. - d.addCallback(lambda dn: root.lookup(dn)) - - def _noEntry(fail): - """ - Called when the requested BIND DN was not found. - """ - fail.trap(ldaperrors.LDAPNoSuchObject) - return None - - d.addErrback(_noEntry) - - def _gotEntry(entry, auth): - """ - Called when the requested BIND DN was found. - """ - if entry is None: - raise ldaperrors.LDAPInvalidCredentials() - - d = entry.bind(auth) - - def _cb(entry): - """ - Called when BIND operation was successful. - """ - self.boundUser = entry - msg = pureldap.LDAPBindResponse( - resultCode=ldaperrors.Success.resultCode, - matchedDN=entry.dn.getText(), + filter_text = b"(" + self._loginAttribute + b"=" + request.dn + b")" + results = yield root.search(filterText=filter_text) + if len(results) != 1: + defer.returnValue(request) + + defer.returnValue( + pureldap.LDAPBindRequest( + version=request.version, + dn=results[0].dn.getText(), + auth=request.auth, + tag=request.tag, + sasl=request.sasl, ) - return msg - - d.addCallback(_cb) - return d - - d.addCallback(_gotEntry, request.auth) + ) - return d + defer.returnValue( + ( + yield super(LDAPServerWithUPNBind, self).handle_LDAPBindRequest( + (yield _request()), *args, **kwargs + ) + ) + ) From b246e2d8c0200c8736b138721d6b0082dd3eb3b5 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Tue, 18 Feb 2020 19:57:01 +0000 Subject: [PATCH 2/3] restore import --- docs/source/examples/ldaptor_with_upn_bind.py | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/source/examples/ldaptor_with_upn_bind.py b/docs/source/examples/ldaptor_with_upn_bind.py index d9560d2f..1462d0c0 100644 --- a/docs/source/examples/ldaptor_with_upn_bind.py +++ b/docs/source/examples/ldaptor_with_upn_bind.py @@ -17,6 +17,7 @@ userPrincipalName: bob@ad.example.org """ +from ldaptor import interfaces from ldaptor.protocols import pureldap from twisted.internet import defer from ldaptor.protocols.ldap.ldapserver import LDAPServer From d1449d4e90ca65a9b388bdc086e2c841d99876fd Mon Sep 17 00:00:00 2001 From: Bret Curtis Date: Wed, 26 Aug 2026 10:49:12 +0200 Subject: [PATCH 3/3] UPN bind example: docstrings + light modernization (fixes #161) Applies the review comments left on #161 (adiroiban) by extracting the UPN resolution into a well-documented ``_resolveUPNBindDN`` helper, explaining the ``@`` vs ``,`` shape check that identifies a UPN, and naming the two fall-through cases (not-a-UPN, no-unambiguous-match). Also drops a couple of Python 2 leftovers now that we require 3.10+: the explicit ``object`` base and ``super(Cls, self)``. Behaviour is unchanged; this is a docs-only example file, still exercised only by being importable. Co-authored-with: Thomas Grainger --- docs/source/NEWS.rst | 9 ++ docs/source/examples/ldaptor_with_upn_bind.py | 103 ++++++++++-------- 2 files changed, 69 insertions(+), 43 deletions(-) diff --git a/docs/source/NEWS.rst b/docs/source/NEWS.rst index 3ee28aa4..b5e93548 100644 --- a/docs/source/NEWS.rst +++ b/docs/source/NEWS.rst @@ -101,6 +101,15 @@ Fixes ``LDAPClientTestDriver`` in ``ldaptor.testutil`` grew a matching ``notifyOnDisconnect`` so tests can exercise this path (#231). +Docs +~~~~ + +- Rewrote the ``ldaptor_with_upn_bind`` example: the custom + ``handle_LDAPBindRequest`` now delegates to the stock + ``LDAPServer.handle_LDAPBindRequest`` after resolving a UPN-style + BIND DN, instead of copy-pasting the base-class logic. Docstrings + explain the UPN shape and the fall-through cases (#161). + 21.2.0 (2021-02-28) ------------------- diff --git a/docs/source/examples/ldaptor_with_upn_bind.py b/docs/source/examples/ldaptor_with_upn_bind.py index 1462d0c0..93bd4de9 100644 --- a/docs/source/examples/ldaptor_with_upn_bind.py +++ b/docs/source/examples/ldaptor_with_upn_bind.py @@ -1,20 +1,27 @@ """ An ldaptor LDAP server which can authenticate based on UPN, as AD does. -The LDAP entry needs to have the `userPrincipalName` attribute set. - -dn: uid=bob,ou=people,dc=example,dc=org -objectclass: top -objectclass: person -objectClass: inetOrgPerson -uid: bob -cn: bobby -gn: Bob -sn: Roberts -mail: bob@example.org -homeDirectory: e:\\Users\\bob -userPassword: pass -userPrincipalName: bob@ad.example.org +The LDAP entry needs to have the ``userPrincipalName`` attribute set: + + dn: uid=bob,ou=people,dc=example,dc=org + objectclass: top + objectclass: person + objectClass: inetOrgPerson + uid: bob + cn: bobby + gn: Bob + sn: Roberts + mail: bob@example.org + homeDirectory: e:\\Users\\bob + userPassword: pass + userPrincipalName: bob@ad.example.org + +A UPN bind arrives as ``User.Name@ad.example.tld`` in the BIND DN slot +rather than a normal distinguished name. This server intercepts the BIND +request, looks up the entry whose ``userPrincipalName`` matches, and +rewrites the request's DN to that entry's real DN before delegating to +the stock :class:`LDAPServer` bind handler. Non-UPN BIND requests are +forwarded unchanged. """ from ldaptor import interfaces @@ -23,40 +30,50 @@ from ldaptor.protocols.ldap.ldapserver import LDAPServer -class LDAPServerWithUPNBind(LDAPServer, object): +class LDAPServerWithUPNBind(LDAPServer): """ - An LDAP server which support BIND using UPN similar to AD. + An LDAP server which supports BIND using a UPN (User Principal Name), + similar to Active Directory. """ _loginAttribute = b"userPrincipalName" @defer.inlineCallbacks def handle_LDAPBindRequest(self, request, *args, **kwargs): - @defer.inlineCallbacks - def _request(): - if not (b"@" in request.dn and b"," not in request.dn): - defer.returnValue(request) - root = interfaces.IConnectedLDAPEntry(self.factory) - # This might be an UPN request. - filter_text = b"(" + self._loginAttribute + b"=" + request.dn + b")" - results = yield root.search(filterText=filter_text) - if len(results) != 1: - defer.returnValue(request) - - defer.returnValue( - pureldap.LDAPBindRequest( - version=request.version, - dn=results[0].dn.getText(), - auth=request.auth, - tag=request.tag, - sasl=request.sasl, - ) - ) - - defer.returnValue( - ( - yield super(LDAPServerWithUPNBind, self).handle_LDAPBindRequest( - (yield _request()), *args, **kwargs - ) - ) + resolved = yield self._resolveUPNBindDN(request) + result = yield super().handle_LDAPBindRequest(resolved, *args, **kwargs) + return result + + @defer.inlineCallbacks + def _resolveUPNBindDN(self, request): + """ + If ``request`` looks like a UPN bind, resolve it to a real BIND DN. + + A UPN takes the form ``User.Name@ad.example.tld``: it contains an + ``@`` but no ``,``, so it is not a valid distinguished name. When + the shape matches, search the directory for the entry whose + ``userPrincipalName`` matches and return a rewritten + :class:`LDAPBindRequest` targeting that entry's DN. Otherwise + return ``request`` unchanged so the caller can fall through to the + normal DN-based bind path. + """ + if b"@" not in request.dn or b"," in request.dn: + # Not a UPN request; leave the DN alone. + return request + + root = interfaces.IConnectedLDAPEntry(self.factory) + filter_text = b"(" + self._loginAttribute + b"=" + request.dn + b")" + results = yield root.search(filterText=filter_text) + + if len(results) != 1: + # No unambiguous UPN match; fall through to the requested BIND + # DN and let the stock handler reject it as usual. + return request + + return pureldap.LDAPBindRequest( + version=request.version, + dn=results[0].dn.getText(), + auth=request.auth, + tag=request.tag, + sasl=request.sasl, )