Passwordless Python LDAP3 authentication from Windows client

Question

I am using the great ldap3 package and I am trying to connect with a active directory server but without requiring to provide actual credentials in plain text.

Following SASL mechanisms are supported. ['GSSAPI', 'GSS-SPNEGO', 'EXTERNAL', 'DIGEST-MD5']

I tried to install the package GSSAPI but that doesn't work on my Windows machine. Error on pip install gssapi was: subprocess.CalledProcessError: Command 'krb5-config --libs gssapi' returned non-zero exit status 1.

Can anybody provide a simple example for that? I believe GSS-SPNEGO could be the solution but I did't find any comprehensible example in the internet.

Answer 1

I think the answer in 2024 is a little bit easier. At least for my use case.

According to the documentation you can now "just" use Kerberos as authentication method: https://ldap3.readthedocs.io/en/latest/bind.html#kerberos without patching any file like the answer from @dima-lituiev suggested.

I just installed winkerberos.

pip install winkerberos

And wrote the following script:

from ldap3 import Server, Connection, SASL, KERBEROS, ALL
server = Server("YOUR_DOMAIN_CONTROLLER_URL", get_info=ALL)
c = Connection(
    server, authentication=SASL, sasl_mechanism=KERBEROS)
c.bind()

print(c.extend.standard.who_am_i())

Answer 2

Thank you for asking this. I gave it one last shot today and got it to work.

See Davide's answer

It requires you to have the ldap3 package and to install the winkerberos package:

pip install winkerberos

Then you need to replace the kerberos.py file in your site-packages (PYTHON_HOME\Lib\site-packages\ldap3\protocol\sasl\kerberos.py) with the one he links to replacement kerberos.py.

You need to change the following line in the replacement kerberos.py file:

from treadmill import kerberoswrapper as kerberos 

Change to

import winkerberos as kerberos

Then you can connect like this:

from ldap3 import Server, Connection, Tls, SASL, GSSAPI
import ssl

tls = Tls(validate=ssl.CERT_NONE, version=ssl.PROTOCOL_TLSv1)
server = Server('server_fqdn', use_ssl=True, tls=tls)
c = Connection(server, authentication=SASL, sasl_mechanism=GSSAPI)
c.bind()
print(c.extend.standard.who_am_i())
c.unbind()

Replace server_fqdn with the fully qualified domain name of your AD server.

You may want to change the version value to whatever protocol your AD server uses.

If someone has a less messy method to accomplish this please chime in!

Answer 3

Using the initial answer, and to avoid monkey-patching, one could utilize the following code, based on file provided there and on the ldap3\core\connection.py module.

ldap3kerberos.py

"""Replaces the use of python-gssapi with kerberos in ldap3.
"""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import base64
import socket

import ldap3
from ldap3.core.exceptions import LDAPCommunicationError
from ldap3.protocol.sasl.sasl import send_sasl_negotiation
from ldap3.protocol.sasl.sasl import abort_sasl_negotiation

from ldap3.protocol.sasl.external import sasl_external
from ldap3.protocol.sasl.digestMd5 import sasl_digest_md5
from ldap3.protocol.sasl.plain import sasl_plain
from ldap3.utils.log import log, log_enabled, BASIC
from ldap3 import EXTERNAL, DIGEST_MD5, GSSAPI


import winkerberos as kerberos

NO_SECURITY_LAYER = 1
INTEGRITY_PROTECTION = 2
CONFIDENTIALITY_PROTECTION = 4


class Connection(ldap3.Connection):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def do_sasl_bind(self,
                     controls):
        if log_enabled(BASIC):
            log(BASIC, 'start SASL BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            result = None

            if not self.sasl_in_progress:
                self.sasl_in_progress = True
                try:
                    if self.sasl_mechanism == EXTERNAL:
                        result = sasl_external(self, controls)
                    elif self.sasl_mechanism == DIGEST_MD5:
                        result = sasl_digest_md5(self, controls)
                    elif self.sasl_mechanism == GSSAPI:
                        result = sasl_gssapi(self, controls)
                    elif self.sasl_mechanism == 'PLAIN':
                        result = sasl_plain(self, controls)
                finally:
                    self.sasl_in_progress = False

            if log_enabled(BASIC):
                log(BASIC, 'done SASL BIND operation, result <%s>', result)

            return result


def sasl_gssapi(connection, controls):
    """
    Performs a bind using the Kerberos v5 ("GSSAPI") SASL mechanism
    from RFC 4752. Does not support any security layers, only authentication!
    sasl_credentials can be empty or a tuple with one or two elements.
    The first element determines which service principal to request a ticket
    for and can be one of the following:
    - None or False, to use the hostname from the Server object
    - True to perform a reverse DNS lookup to retrieve the canonical hostname
      for the hosts IP address
    - A string containing the hostname
    The optional second element is what authorization ID to request.
    - If omitted or None, the authentication ID is used as the authorization ID
    - If a string, the authorization ID to use. Should start with "dn:" or
      "user:".
    """
    # pylint: disable=too-many-branches
    target_name = None
    authz_id = b''
    if connection.sasl_credentials:
        if (len(connection.sasl_credentials) >= 1 and
                connection.sasl_credentials[0]):
            if connection.sasl_credentials[0] is True:
                hostname = \
                    socket.gethostbyaddr(connection.socket.getpeername()[0])[0]
                target_name = 'ldap@' + hostname

            else:
                target_name = 'ldap@' + connection.sasl_credentials[0]
        if (len(connection.sasl_credentials) >= 2 and
                connection.sasl_credentials[1]):
            authz_id = connection.sasl_credentials[1].encode("utf-8")
    if target_name is None:
        target_name = 'ldap@' + connection.server.host

    gssflags = (
        kerberos.GSS_C_MUTUAL_FLAG |
        kerberos.GSS_C_SEQUENCE_FLAG |
        kerberos.GSS_C_INTEG_FLAG |
        kerberos.GSS_C_CONF_FLAG
    )

    _, ctx = kerberos.authGSSClientInit(target_name, gssflags=gssflags)

    in_token = b''
    try:
        while True:
            status = kerberos.authGSSClientStep(
                ctx,
                base64.b64encode(in_token).decode('ascii')
            )
            out_token = kerberos.authGSSClientResponse(ctx) or ''
            result = send_sasl_negotiation(
                connection,
                controls,
                base64.b64decode(out_token)
            )
            in_token = result['saslCreds'] or b''
            if status == kerberos.AUTH_GSS_COMPLETE:
                break

        kerberos.authGSSClientUnwrap(
            ctx,
            base64.b64encode(in_token).decode('ascii')
        )
        unwrapped_token = base64.b64decode(
            kerberos.authGSSClientResponse(ctx) or ''
        )

        if len(unwrapped_token) != 4:
            raise LDAPCommunicationError('Incorrect response from server')

        server_security_layers = unwrapped_token[0]
        if not isinstance(server_security_layers, int):
            server_security_layers = ord(server_security_layers)
        if server_security_layers in (0, NO_SECURITY_LAYER):
            if unwrapped_token.message[1:] != '\x00\x00\x00':
                raise LDAPCommunicationError(
                    'Server max buffer size must be 0 if no security layer'
                )
        if not server_security_layers & NO_SECURITY_LAYER:
            raise LDAPCommunicationError(
                'Server requires a security layer, but this is not implemented'
            )

        client_security_layers = bytearray([NO_SECURITY_LAYER, 0, 0, 0])
        kerberos.authGSSClientWrap(
            ctx,
            base64.b64encode(
                bytes(client_security_layers) + authz_id
            ).decode('ascii')
        )
        out_token = kerberos.authGSSClientResponse(ctx) or ''

        return send_sasl_negotiation(
            connection,
            controls,
            base64.b64decode(out_token)
        )
    except (kerberos.GSSError, LDAPCommunicationError):
        abort_sasl_negotiation(connection, controls)
        raise

Install winkerberos: pip install winkerberos

In your script, use the following code (connect_timeout, mode and receive_timeout parameters are for example only and could be omitted or changed):

import ldap
import ldap3kerberos

server = ldap3.Server(fqdn, connect_timeout=10, mode=ldap3.IP_V4_ONLY)
conn = ldap3kerberos.Connection(
    server, authentication=ldap3.SASL, sasl_mechanism=ldap3.GSSAPI,
    auto_bind=True, receive_timeout=10
)

If you have several domain controller servers for an AD domain, ensure you are connecting to some specific server, otherwise you will get the exception:

winkerberos.GSSError: SSPI: InitializeSecurityContext: The specified target is unknown or unreachable