Python ldap3 GSS-SPNEGO NTLM authentication

ldap-auth-ntlm.md

Example of how to glue pyspengo to ldap3 to pull off a GSS-SPNEGO authentication.

. /path/to/your/env_configuration
export LDAP_HOST LDAP_HOST_CA LDAP_USERNAME LDAP_PASSWORD
python ldap-auth-ntlm.py

ldap-auth-ntlm.py

#!/usr/bin/env python2.7

# reference
#  https://gist.github.com/pefoley2/c6991488e303cc46f8fa
#  https://gist.github.com/sigmaris/f4bf307e97b449148d78

# deps
#  apt-get -y install --no-install-recommends python-cryptography python-enum34 python-ldap3 python-pip python-pkg-resources
#  pip install pyspnego

# monkey patching guards
import pkg_resources
pkg_resources.require('ldap3==2.4.1')
pkg_resources.require('pyspnego==0.1.5')

import ldap3
from ldap3 import Connection, Server, ServerPool, Tls
from ldap3.protocol.sasl.sasl import send_sasl_negotiation, abort_sasl_negotiation
from ldap3.utils.log import log, log_enabled, set_library_log_detail_level, OFF, BASIC, NETWORK, EXTENDED
import logging
import os
import spnego
import ssl

logging.basicConfig(level=logging.DEBUG)
set_library_log_detail_level(NETWORK)	# EXTENDED trips the ASN.1 parser

# monkey patch out sign/seal flags to avoid:
# 00002029: LdapErr: DSID-0C09054C, comment: Cannot bind using sign/seal on a connection on which TLS or SSL is in effect, data 0, v4563
from spnego.ntlm import NTLMProxy
from spnego._ntlm_raw.messages import NegotiateFlags
if os.environ.get('LDAP_HOST_CA') != None:
    NTLMProxy.__init_orig__ = NTLMProxy.__init__
    def NTLMProxy_init_fixup(self, *args):
        NTLMProxy.__init_orig__(self, *args)
        self._context_req &= ~NegotiateFlags.sign
        self._context_req &= ~NegotiateFlags.seal
    NTLMProxy.__init__ = NTLMProxy_init_fixup

GSS_SPNEGO = 'GSS-SPNEGO'

ldap3.core.connection.SASL_AVAILABLE_MECHANISMS = [ GSS_SPNEGO ]

class GSSSPNEGOConnection(ldap3.Connection):
    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 == GSS_SPNEGO:
                        result = sasl_gss_spnego(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_gss_spnego(connection, controls):
    client = spnego.client(username=connection.sasl_credentials[0], password=connection.sasl_credentials[1], protocol='ntlm')

    in_token = None
    while not client.complete:
        out_token = client.step(in_token)
        if not out_token:
            break

        result = send_sasl_negotiation(connection, controls, out_token)
        if 'saslCreds' in result and result['saslCreds'] is not None:
            in_token = result['saslCreds']
        else:
            return None

if __name__ == '__main__':
    server_pool = ServerPool(None, ldap3.RANDOM, active=True, exhaust=10)

    server_options = {}
    server_options['get_info'] = ldap3.NONE

    LDAP_HOST_CA = os.environ.get('LDAP_HOST_CA')
    if LDAP_HOST_CA != None:
        tls_options = {}
        tls_options['version'] = ssl.PROTOCOL_TLSv1_2
        tls_options['validate'] = ssl.CERT_REQUIRED
        # convert to unicode https://bugs.python.org/issue37079
        tls_options['ca_certs_data'] = unicode(LDAP_HOST_CA.decode('string_escape'))
        server_options['tls'] = Tls(**tls_options)

    for host in os.environ.get('LDAP_HOST').split():
        if LDAP_HOST_CA != None:
            host = 'ldaps://' + host
        server = Server(host, **server_options)
        server_pool.add(server)

    connection_options = {}
    connection_options['read_only'] = True
    connection_options['authentication'] = ldap3.SASL
    connection_options['sasl_mechanism'] = GSS_SPNEGO

    connection = GSSSPNEGOConnection(server_pool, **connection_options)

    rebind_options = {}
    rebind_options['sasl_credentials'] = (os.environ.get('LDAP_USERNAME'),os.environ.get('LDAP_PASSWORD').decode('string_escape'))

    connection.rebind(**rebind_options)
    connection.unbind()