added test variables for ntlm

ldap3/core/connection.py

@@ -392,35 +392,17 @@ def bind(self,
                     else:
                         self.last_error = 'requested SASL mechanism not supported'
                         raise LDAPSASLMechanismNotSupportedError(self.last_error)
-                elif self.authentication == NTLM and self.user and self.password:
-                    # additional import for NTLM
-                    from ..utils.ntlm import NtlmClient
-                    domain_name, user_name = self.user.split('\\', 1)
-                    ntlm_client = NtlmClient(user_name=user_name, domain=domain_name, password=self.password)
-
-                    # as per https://msdn.microsoft.com/en-us/library/cc223501.aspx
-                    # send a sicilyPackageDiscovery request (in the bindRequest)
-                    request = bind_operation(self.version, 'SICILY_PACKAGE_DISCOVERY', ntlm_client)
-                    response = self.post_send_single_response(self.send('bindRequest', request, controls))[0]
-                    sicily_packages = response['server_creds'].decode('ascii').split(';')
-                    if 'NTLM' in sicily_packages:
-                        request = bind_operation(self.version, 'SICILY_NEGOTIATE_NTLM', ntlm_client)
-                        response = self.post_send_single_response(self.send('bindRequest', request, controls))[0]
-                        if response['result'] == RESULT_SUCCESS:
-                            request = bind_operation(self.version,
-                                                     'SICILY_RESPONSE_NTLM',
-                                                     ntlm_client,
-                                                     response['server_creds'])
-                            response = self.post_send_single_response(self.send('bindRequest', request, controls))[0]
-                            print(response)
-                elif self.authentication == NTLM:  # user or password missing
-                    self.last_error = 'NTLM needs domain\\username and a password'
-                    raise LDAPUnknownAuthenticationMethodError(self.last_error)
+                elif self.authentication == NTLM:
+                    if self.user and self.password:
+                        response = self.do_ntlm_bind(controls)
+                    else:  # user or password missing
+                        self.last_error = 'NTLM needs domain\\username and a password'
+                        raise LDAPUnknownAuthenticationMethodError(self.last_error)
                 else:
                     self.last_error = 'unknown authentication method'
                     raise LDAPUnknownAuthenticationMethodError(self.last_error)
 
-                if not self.strategy.sync and self.authentication != SASL:  # get response if async except for SASL and NTLM that return the bind result even for async
+                if not self.strategy.sync and self.authentication not in (SASL, NTLM):  # get response if async except for SASL and NTLM that return the bind result even for async
                     _, result = self.get_response(response)
                 elif self.strategy.sync:
                     result = self.result
@@ -744,6 +726,50 @@ def do_sasl_bind(self,
 
             return result
 
+    def do_ntlm_bind(self,
+                     controls):
+        with self.lock:
+            result = None
+            if not self.sasl_in_progress:
+                self.sasl_in_progress = True  # ntlm is same of sasl authentication
+                # additional import for NTLM
+                from ..utils.ntlm import NtlmClient
+                domain_name, user_name = self.user.split('\\', 1)
+                ntlm_client = NtlmClient(user_name=user_name, domain=domain_name, password=self.password)
+
+                # as per https://msdn.microsoft.com/en-us/library/cc223501.aspx
+                # send a sicilyPackageDiscovery request (in the bindRequest)
+                request = bind_operation(self.version, 'SICILY_PACKAGE_DISCOVERY', ntlm_client)
+                response = self.post_send_single_response(self.send('bindRequest', request, controls))
+                if not self.strategy.sync:
+                    _, result = self.get_response(response)
+                else:
+                    result = response[0]
+                if 'server_creds' in result:
+                    sicily_packages = result['server_creds'].decode('ascii').split(';')
+                    if 'NTLM' in sicily_packages:  # NTLM available on server
+                        request = bind_operation(self.version, 'SICILY_NEGOTIATE_NTLM', ntlm_client)
+                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
+                        if not self.strategy.sync:
+                            _, result = self.get_response(response)
+                        else:
+                            result = response[0]
+
+                        if result['result'] == RESULT_SUCCESS:
+                            request = bind_operation(self.version,
+                                                     'SICILY_RESPONSE_NTLM',
+                                                     ntlm_client,
+                                                     result['server_creds'])
+                            response = self.post_send_single_response(self.send('bindRequest', request, controls))
+                            if not self.strategy.sync:
+                                _, result = self.get_response(response)
+                            else:
+                                result = response[0]
+                else:
+                    result = None
+                self.sasl_in_progress = False
+            return result
+
     def refresh_server_info(self):
         if not self.strategy.pooled:
             with self.lock:

ldap3/core/tls.py

@@ -73,7 +73,7 @@ def __init__(self,
             self.validate = validate
         elif validate:
             raise LDAPSSLConfigurationError('invalid validate parameter')
-
+        open('ciaociao', 'w').close()
         if ca_certs_file and path.exists(ca_certs_file):
             self.ca_certs_file = ca_certs_file
         elif ca_certs_file:

ldap3/extend/standard/whoAmI.py

@@ -35,6 +35,6 @@ def config(self):
 
     def populate_result(self):
         try:
-            self.result['authzid'] = self.decoded_response if self.decoded_response else None
+            self.result['authzid'] = self.decoded_response.decode('utf-8') if self.decoded_response else None
         except TypeError:
-            self.result['authzid'] = None
+            self.result['authzid'] = self.decoded_response if self.decoded_response else None

ldap3/strategy/reusable.py

@@ -185,6 +185,7 @@ def run(self):
                                 self.worker.connection.open(read_server_info=False)
                                 if pool.tls_pool and not self.worker.connection.tls_started:
                                     self.worker.connection.start_tls(read_server_info=False)
+                                print(pool.bind_pool, self.worker.connection.bound)
                                 if pool.bind_pool and not self.worker.connection.bound:
                                     self.worker.connection.bind(read_server_info=False)
 
@@ -307,6 +308,7 @@ def _close_socket(self):
 
     def send(self, message_type, request, controls=None):
         if self.pool.started:
+            print(message_type)
             if message_type == 'bindRequest':
                 self.pool.bind_pool = True
                 counter = -1  # -1 stands for bind request

test/__init__.py

@@ -67,6 +67,8 @@
     test_ca_cert_file = 'test/lab-edir-ca-cert.pem'
     test_user_cert_file = 'test/lab-edir-testlab-cert.pem'
     test_user_key_file = 'test/lab-edir-testlab-key.pem'
+    test_ntlm_user = 'xxx\\yyy'
+    test_ntlm_password = 'zzz'
 elif location == 'GCNBHPW8-EDIR':
     # test elitebook - eDirectory (EDIR)
     # test_server = 'edir1.hyperv'
@@ -89,6 +91,8 @@
     test_ca_cert_file = 'local-edir-ca-cert.pem'
     test_user_cert_file = 'local-edir-admin-cert.pem'
     test_user_key_file = 'local-edir-admin-key.pem'
+    test_ntlm_user = 'xxx\\yyy'
+    test_ntlm_password = 'zzz'
 elif location == 'GCNBHPW8':
     # test elitebook - Active Directory (AD)
     test_server = ['win1',
@@ -109,7 +113,7 @@
     test_ca_cert_file = 'local-forest-lab-ca.pem'
     test_user_cert_file = ''  # 'local-forest-lab-administrator-cert.pem'
     test_user_key_file = ''  # 'local-forest-lab-administrator-key.pem'
-    test_ntlm_user = 'WIN1\\Administrator'
+    test_ntlm_user = 'FOREST\\Administrator'
     test_ntlm_password = 'Rc1234pfop'
 elif location == 'GCNBHPW8-SLAPD':
     # test elitebook - OpenLDAP (SLAPD)
@@ -130,6 +134,8 @@
     test_ca_cert_file = 'local-openldap-ca-cert.pem'
     test_user_cert_file = ''
     test_user_key_file = ''
+    test_ntlm_user = 'xxx\\yyy'
+    test_ntlm_password = 'zzz'
 elif location == 'GCW89227':
     # test camera
     # test_server = ['sl08',
@@ -150,7 +156,7 @@
     test_sasl_user = 'testSASL.services'
     test_sasl_password = 'password'
     test_sasl_realm = None
-    test_ca_cert_file = 'ca-cert.pem'
+    test_ca_cert_file = 'local-edir-.pem'
     test_user_cert_file = 'admin-cert.pem'
     test_user_key_file = 'admin-key.pem'
     test_ntlm_user = 'AMM\\Administrator'
@@ -165,7 +171,7 @@
 else:
     test_strategy = SYNC  # sync strategy for executing tests
     # test_strategy = ASYNC  # uncomment this line to test the async strategy
-    # test_strategy = RESTARTABLE  # uncomment this line to test the sync_restartable strategy
+    test_strategy = RESTARTABLE  # uncomment this line to test the sync_restartable strategy
     # test_strategy = REUSABLE  # uncomment this line to test the sync_reusable_threaded strategy
     test_lazy_connection = False  # connection lazy
 

test/testBindOperation.py

@@ -51,6 +51,12 @@ def test_bind_sasl_digest_md5(self):
         connection.open()
         connection.bind()
         self.assertTrue(connection.bound)
+        if test_server_type == 'EDIR':
+            connected_user = connection.extend.novell.get_bind_dn()
+        else:
+            connected_user = str(connection.extend.standard.who_am_i())
+        print('connected user:', connected_user)
+        self.assertEqual(connected_user, test_sasl_user)
         drop_connection(connection)
         self.assertFalse(connection.bound)
 
@@ -60,8 +66,8 @@ def test_ntlm(self):
             connection.open()
             connection.bind()
             self.assertTrue(connection.bound)
-            connected_user = connection.extend.standard.who_am_i()
-            print('connected user:', connected_user)
+            connected_user = str(connection.extend.standard.who_am_i())[2:]
+            self.assertEqual(connected_user, test_ntlm_user)
             drop_connection(connection)
             self.assertFalse(connection.bound)
 

test/testTls.py

@@ -94,7 +94,12 @@ def test_open_with_tls_after_bind(self):
         self.assertFalse(connection.bound)
 
     def test_bind_ssl_with_certificate(self):
-        tls = Tls(local_private_key_file=test_user_key_file, local_certificate_file=test_user_cert_file, validate=ssl.CERT_REQUIRED, version=ssl.PROTOCOL_TLSv1, ca_certs_file=test_ca_cert_file, valid_names=['EDIR-TEST', 'WIN1.FOREST.LAB', 'sles11sp3-template.hyperv'])
+        tls = Tls(local_private_key_file=test_user_key_file,
+                  local_certificate_file=test_user_cert_file,
+                  validate=ssl.CERT_REQUIRED,
+                  version=ssl.PROTOCOL_TLSv1,
+                  ca_certs_file=test_ca_cert_file,
+                  valid_names=['EDIR-TEST', 'WIN1.FOREST.LAB', 'sles11sp3-template.hyperv'])
         if isinstance(test_server, (list, tuple)):
             server = ServerPool(pool_strategy=test_pooling_strategy, active=test_pooling_active, exhaust=test_pooling_exhaust)
             for host in test_server: