diff --git a/apricot/ldap/read_only_ldap_server.py b/apricot/ldap/read_only_ldap_server.py index e745a8c..17081e1 100644 --- a/apricot/ldap/read_only_ldap_server.py +++ b/apricot/ldap/read_only_ldap_server.py @@ -10,9 +10,14 @@ LDAPSearchResultEntry, ) from twisted.internet import defer +from twisted.python import log class ReadOnlyLDAPServer(LDAPServer): + def __init__(self): + super().__init__() + self.debug = True + def getRootDSE( # noqa: N802 self, request: LDAPBindRequest, @@ -21,17 +26,21 @@ def getRootDSE( # noqa: N802 """ Handle an LDAP Root RSE request """ + log.msg(f"Types: request '{type(request)}'; reply '{type(reply)}'") + log.msg(f"Calling getRootDSE with '{request.toWire() if request else 'NoRequest'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") return super().getRootDSE(request, reply) def handle_LDAPAddRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Refuse to handle an LDAP add request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP add with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") id((request, controls, reply)) # ignore unused arguments msg = "ReadOnlyLDAPServer will not handle LDAP add requests" raise LDAPProtocolError(msg) @@ -39,34 +48,40 @@ def handle_LDAPAddRequest( # noqa: N802 def handle_LDAPBindRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Handle an LDAP bind request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP bind with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") return super().handle_LDAPBindRequest(request, controls, reply) def handle_LDAPCompareRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Handle an LDAP compare request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP compare with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") return super().handle_LDAPCompareRequest(request, controls, reply) def handle_LDAPDelRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Refuse to handle an LDAP delete request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP del with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") id((request, controls, reply)) # ignore unused arguments msg = "ReadOnlyLDAPServer will not handle LDAP delete requests" raise LDAPProtocolError(msg) @@ -74,23 +89,27 @@ def handle_LDAPDelRequest( # noqa: N802 def handle_LDAPExtendedRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Handle an LDAP extended request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP extended with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") return super().handle_LDAPExtendedRequest(request, controls, reply) def handle_LDAPModifyDNRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Refuse to handle an LDAP modify DN request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP modify DN with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") id((request, controls, reply)) # ignore unused arguments msg = "ReadOnlyLDAPServer will not handle LDAP modify DN requests" raise LDAPProtocolError(msg) @@ -98,12 +117,14 @@ def handle_LDAPModifyDNRequest( # noqa: N802 def handle_LDAPModifyRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Refuse to handle an LDAP modify request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP modify with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") id((request, controls, reply)) # ignore unused arguments msg = "ReadOnlyLDAPServer will not handle LDAP modify requests" raise LDAPProtocolError(msg) @@ -111,21 +132,25 @@ def handle_LDAPModifyRequest( # noqa: N802 def handle_LDAPUnbindRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[..., None] | None, ) -> None: """ Handle an LDAP unbind request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP unbind with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") super().handle_LDAPUnbindRequest(request, controls, reply) def handle_LDAPSearchRequest( # noqa: N802 self, request: LDAPBindRequest, - controls: LDAPControl | None, + controls: list[LDAPControl] | None, reply: Callable[[LDAPSearchResultEntry], None] | None, ) -> defer.Deferred[ILDAPEntry]: """ Handle an LDAP search request """ + log.msg(f"Types: request '{type(request)}'; controls '{type(controls)}'; reply '{type(reply)}'") + log.msg(f"Calling LDAP search with '{request.toWire() if request else 'NoRequest'}' '{[c.toWire() for c in controls] if controls else 'NoControls'}' (DN='{request.dn if hasattr(request, 'dn') else 'NoDN'}')") return super().handle_LDAPSearchRequest(request, controls, reply)