Merge pull request #1526 from NtAlexio2/ldap-sasl

support sasl authentication in ldap protocol
fortra · Jun 29, 2023 · b5dab2d · b5dab2d
2 parents 24cb0a9 + fbd3a8c
commit b5dab2d
Show file tree

Hide file tree

Showing 2 changed files with 54 additions and 2 deletions.
diff --git a/impacket/ldap/ldap.py b/impacket/ldap/ldap.py
@@ -35,7 +35,7 @@
     KNOWN_CONTROLS, CONTROL_PAGEDRESULTS, NOTIFICATION_DISCONNECT, KNOWN_NOTIFICATIONS, BindRequest, SearchRequest, \
     SearchResultDone, LDAPMessage
 from impacket.ntlm import getNTLMSSPType1, getNTLMSSPType3
-from impacket.spnego import SPNEGO_NegTokenInit, TypesMech
+from impacket.spnego import SPNEGO_NegTokenInit, SPNEGO_NegTokenResp, TypesMech
 
 try:
     import OpenSSL
@@ -312,6 +312,50 @@ def login(self, user='', password='', domain='', lmhash='', nthash='', authentic
             type3, exportedSessionKey = getNTLMSSPType3(negotiate, bytes(type2), user, password, domain, lmhash, nthash)
             bindRequest['authentication']['sicilyResponse'] = type3.getData()
             response = self.sendReceive(bindRequest)[0]['protocolOp']
+        elif authenticationChoice == 'sasl':
+            if lmhash != '' or nthash != '':
+                if len(lmhash) % 2:
+                    lmhash = '0' + lmhash
+                if len(nthash) % 2:
+                    nthash = '0' + nthash
+                try:
+                    lmhash = unhexlify(lmhash)
+                    nthash = unhexlify(nthash)
+                except TypeError:
+                    pass
+
+            bindRequest['name'] = user
+
+            # NTLM Negotiate
+            negotiate = getNTLMSSPType1('', domain)
+
+            blob = SPNEGO_NegTokenInit()
+            blob['MechTypes'] = [TypesMech['NTLMSSP - Microsoft NTLM Security Support Provider']]
+            blob['MechToken'] = negotiate.getData()
+
+            bindRequest['authentication']['sasl']['mechanism'] = 'GSS-SPNEGO'
+            bindRequest['authentication']['sasl']['credentials'] = blob.getData()
+            response = self.sendReceive(bindRequest)[0]['protocolOp']
+            if response['bindResponse']['resultCode'] != ResultCode('saslBindInProgress'):
+                raise LDAPSessionError(
+                    errorString='Error in bindRequest during the NTLMAuthNegotiate request -> %s: %s' %
+                                (response['bindResponse']['resultCode'].prettyPrint(),
+                                 response['bindResponse']['diagnosticMessage'])
+                )
+
+            # NTLM Challenge
+            serverSaslCreds = response['bindResponse']['serverSaslCreds']
+            spnegoTokenResp = SPNEGO_NegTokenResp(serverSaslCreds.asOctets())
+            type2 = spnegoTokenResp['ResponseToken']
+
+            # NTLM Auth
+            type3, exportedSessionKey = getNTLMSSPType3(negotiate, type2, user, password, domain, lmhash, nthash)
+            blob = SPNEGO_NegTokenResp()
+            blob['ResponseToken'] = type3.getData()
+
+            bindRequest['authentication']['sasl']['mechanism'] = 'GSS-SPNEGO'
+            bindRequest['authentication']['sasl']['credentials'] = blob.getData()
+            response = self.sendReceive(bindRequest)[0]['protocolOp']
         else:
             raise LDAPSessionError(errorString="Unknown authenticationChoice: '%s'" % authenticationChoice)
 

diff --git a/tests/SMB_RPC/test_ldap.py b/tests/SMB_RPC/test_ldap.py
@@ -19,7 +19,7 @@
 
 class LDAPTests(RemoteTestCase):
     def connect(self, login=True):
-        self.ldapConnection = ldap.LDAPConnection(self.url, self.baseDN)
+        self.ldapConnection = ldap.LDAPConnection(self.url, self.baseDN, self.machine)
         if login:
             self.ldapConnection.login(self.username, self.password)
         return self.ldapConnection
@@ -82,6 +82,14 @@ def test_sicilyNtlm(self):
 
         self.dummySearch(ldapConnection)
 
+    def test_saslNtlm(self):
+        ldapConnection = self.connect(False)
+        ldapConnection.login(
+            user=self.username, password=self.password, domain=self.domain,
+            authenticationChoice="sasl"
+        )
+        self.dummySearch(ldapConnection)
+
     def test_kerberosLogin(self):
         ldapConnection = self.connect(False)
         ldapConnection.kerberosLogin(self.username, self.password, self.domain)