JezK
Edit File: lockout_tests.py
#!/usr/bin/env python3 # Unix SMB/CIFS implementation. # Copyright (C) Stefan Metzmacher 2020 # Copyright (C) Catalyst.Net Ltd # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # import sys import os sys.path.insert(0, 'bin/python') os.environ['PYTHONUNBUFFERED'] = '1' from concurrent import futures from enum import Enum from functools import partial from multiprocessing import Pipe import time from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers.base import Cipher from cryptography.hazmat.primitives.ciphers import algorithms import ldb from samba import ( NTSTATUSError, dsdb, generate_random_bytes, generate_random_password, ntstatus, unix2nttime, werror, ) from samba.credentials import DONT_USE_KERBEROS, MUST_USE_KERBEROS from samba.crypto import ( aead_aes_256_cbc_hmac_sha512_blob, des_crypt_blob_16, md4_hash_blob, sha512_pbkdf2, ) from samba.dcerpc import lsa, samr from samba.samdb import SamDB from samba.tests import connect_samdb, env_get_var_value, env_loadparm from samba.tests.krb5.as_req_tests import AsReqBaseTest from samba.tests.krb5 import kcrypto from samba.tests.krb5.kdc_base_test import KDCBaseTest from samba.tests.krb5.raw_testcase import KerberosCredentials import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1 from samba.tests.krb5.rfc4120_constants import ( KDC_ERR_CLIENT_REVOKED, KDC_ERR_PREAUTH_FAILED, KRB_AS_REP, KRB_ERROR, NT_PRINCIPAL, NT_SRV_INST, ) global_asn1_print = False global_hexdump = False class ConnectionResult(Enum): LOCKED_OUT = 1 WRONG_PASSWORD = 2 SUCCESS = 3 def connect_kdc(pipe, url, hostname, username, password, domain, realm, workstation, dn, expect_error=True): AsReqBaseTest.setUpClass() as_req_base = AsReqBaseTest() as_req_base.setUp() user_creds = KerberosCredentials() user_creds.set_username(username) user_creds.set_password(password) user_creds.set_domain(domain) user_creds.set_realm(realm) user_creds.set_workstation(workstation) user_creds.set_kerberos_state(DONT_USE_KERBEROS) user_name = user_creds.get_username() cname = as_req_base.PrincipalName_create(name_type=NT_PRINCIPAL, names=user_name.split('/')) krbtgt_creds = as_req_base.get_krbtgt_creds() krbtgt_supported_etypes = krbtgt_creds.tgs_supported_enctypes realm = krbtgt_creds.get_realm() krbtgt_account = krbtgt_creds.get_username() sname = as_req_base.PrincipalName_create(name_type=NT_SRV_INST, names=[krbtgt_account, realm]) expected_salt = user_creds.get_salt() till = as_req_base.get_KerberosTime(offset=36000) kdc_options = krb5_asn1.KDCOptions('postdated') preauth_key = as_req_base.PasswordKey_from_creds(user_creds, kcrypto.Enctype.AES256) ts_enc_padata = as_req_base.get_enc_timestamp_pa_data_from_key(preauth_key) padata = [ts_enc_padata] krbtgt_decryption_key = ( as_req_base.TicketDecryptionKey_from_creds(krbtgt_creds)) etypes = as_req_base.get_default_enctypes() if expect_error: expected_error_modes = (KDC_ERR_CLIENT_REVOKED, KDC_ERR_PREAUTH_FAILED) else: expected_error_modes = 0 # Remove the LDAP connection. del type(as_req_base)._ldb # Indicate that we're ready. This ensures we hit the right transaction # lock. pipe.send_bytes(b'0') # Wait for the main process to take out a transaction lock. if not pipe.poll(timeout=5): raise AssertionError('main process failed to indicate readiness') # Try making a Kerberos AS-REQ to the KDC. This should fail, either due to # the user's account being locked out or due to using the wrong password. as_rep, kdc_exchange_dict = as_req_base._test_as_exchange( cname=cname, realm=realm, sname=sname, till=till, client_as_etypes=etypes, expected_error_mode=expected_error_modes, expected_crealm=realm, expected_cname=cname, expected_srealm=realm, expected_sname=sname, expected_salt=expected_salt, etypes=etypes, padata=padata, kdc_options=kdc_options, expected_supported_etypes=krbtgt_supported_etypes, expected_account_name=user_name, preauth_key=preauth_key, ticket_decryption_key=krbtgt_decryption_key, pac_request=True) as_req_base.assertIsNotNone(as_rep) msg_type = as_rep['msg-type'] if expect_error and msg_type != KRB_ERROR or ( not expect_error and msg_type != KRB_AS_REP): raise AssertionError(f'wrong message type {msg_type}') if not expect_error: return ConnectionResult.SUCCESS error_code = as_rep['error-code'] if error_code == KDC_ERR_CLIENT_REVOKED: return ConnectionResult.LOCKED_OUT elif error_code == KDC_ERR_PREAUTH_FAILED: return ConnectionResult.WRONG_PASSWORD else: raise AssertionError(f'wrong error code {error_code}') def connect_ntlm(pipe, url, hostname, username, password, domain, realm, workstation, dn): user_creds = KerberosCredentials() user_creds.set_username(username) user_creds.set_password(password) user_creds.set_domain(domain) user_creds.set_workstation(workstation) user_creds.set_kerberos_state(DONT_USE_KERBEROS) # Indicate that we're ready. This ensures we hit the right transaction # lock. pipe.send_bytes(b'0') # Wait for the main process to take out a transaction lock. if not pipe.poll(timeout=5): raise AssertionError('main process failed to indicate readiness') try: # Try connecting to SamDB. This should fail, either due to our # account being locked out or due to using the wrong password. SamDB(url=url, credentials=user_creds, lp=env_loadparm()) except ldb.LdbError as err: num, estr = err.args if num != ldb.ERR_INVALID_CREDENTIALS: raise AssertionError(f'connection raised wrong error code ' f'({err})') if f'data {werror.WERR_ACCOUNT_LOCKED_OUT:x},' in estr: return ConnectionResult.LOCKED_OUT elif f'data {werror.WERR_LOGON_FAILURE:x},' in estr: return ConnectionResult.WRONG_PASSWORD else: raise AssertionError(f'connection raised wrong error code ' f'({estr})') else: return ConnectionResult.SUCCESS def connect_samr(pipe, url, hostname, username, password, domain, realm, workstation, dn): # Get the user's NT hash. user_creds = KerberosCredentials() user_creds.set_password(password) nt_hash = user_creds.get_nt_hash() # Generate a new UTF-16 password. new_password = generate_random_password(32, 32) new_password = new_password.encode('utf-16le') # Generate the MD4 hash of the password. new_password_md4 = md4_hash_blob(new_password) # Prefix the password with padding so it is 512 bytes long. new_password_len = len(new_password) remaining_len = 512 - new_password_len new_password = bytes(remaining_len) + new_password # Append the 32-bit length of the password.. new_password += int.to_bytes(new_password_len, length=4, byteorder='little') # Encrypt the password with RC4 and the existing NT hash. encryptor = Cipher(algorithms.ARC4(nt_hash), None, default_backend()).encryptor() new_password = encryptor.update(new_password) # Create a key from the MD4 hash of the new password. key = new_password_md4[:14] # Encrypt the old NT hash with DES to obtain the verifier. verifier = des_crypt_blob_16(nt_hash, key) server = lsa.String() server.string = hostname account = lsa.String() account.string = username nt_password = samr.CryptPassword() nt_password.data = list(new_password) nt_verifier = samr.Password() nt_verifier.hash = list(verifier) conn = samr.samr(f'ncacn_np:{hostname}[krb5,seal,smb2]') # Indicate that we're ready. This ensures we hit the right transaction # lock. pipe.send_bytes(b'0') # Wait for the main process to take out a transaction lock. if not pipe.poll(timeout=5): raise AssertionError('main process failed to indicate readiness') try: # Try changing the password. This should fail, either due to our # account being locked out or due to using the wrong password. conn.ChangePasswordUser3(server=server, account=account, nt_password=nt_password, nt_verifier=nt_verifier, lm_change=True, lm_password=None, lm_verifier=None, password3=None) except NTSTATUSError as err: num, estr = err.args if num == ntstatus.NT_STATUS_ACCOUNT_LOCKED_OUT: return ConnectionResult.LOCKED_OUT elif num == ntstatus.NT_STATUS_WRONG_PASSWORD: return ConnectionResult.WRONG_PASSWORD else: raise AssertionError(f'pwd change raised wrong error code ' f'({num:08X})') else: return ConnectionResult.SUCCESS def connect_samr_aes(pipe, url, hostname, username, password, domain, realm, workstation, dn): # Get the user's NT hash. user_creds = KerberosCredentials() user_creds.set_password(password) nt_hash = user_creds.get_nt_hash() # Generate a new UTF-16 password. new_password = generate_random_password(32, 32) new_password = new_password.encode('utf-16le') # Prepend the 16-bit length of the password.. new_password_len = int.to_bytes(len(new_password), length=2, byteorder='little') new_password = new_password_len + new_password server = lsa.String() server.string = hostname account = lsa.String() account.string = username # Derive a key from the user's NT hash. iv = generate_random_bytes(16) iterations = 5555 cek = sha512_pbkdf2(nt_hash, iv, iterations) enc_key_salt = (b'Microsoft SAM encryption key ' b'AEAD-AES-256-CBC-HMAC-SHA512 16\0') mac_key_salt = (b'Microsoft SAM MAC key ' b'AEAD-AES-256-CBC-HMAC-SHA512 16\0') # Encrypt the new password. ciphertext, auth_data = aead_aes_256_cbc_hmac_sha512_blob(new_password, cek, enc_key_salt, mac_key_salt, iv) # Create the new password structure pwd_buf = samr.EncryptedPasswordAES() pwd_buf.auth_data = list(auth_data) pwd_buf.salt = list(iv) pwd_buf.cipher_len = len(ciphertext) pwd_buf.cipher = list(ciphertext) pwd_buf.PBKDF2Iterations = iterations conn = samr.samr(f'ncacn_np:{hostname}[krb5,seal,smb2]') # Indicate that we're ready. This ensures we hit the right transaction # lock. pipe.send_bytes(b'0') # Wait for the main process to take out a transaction lock. if not pipe.poll(timeout=5): raise AssertionError('main process failed to indicate readiness') try: # Try changing the password. This should fail, either due to our # account being locked out or due to using the wrong password. conn.ChangePasswordUser4(server=server, account=account, password=pwd_buf) except NTSTATUSError as err: num, estr = err.args if num == ntstatus.NT_STATUS_ACCOUNT_LOCKED_OUT: return ConnectionResult.LOCKED_OUT elif num == ntstatus.NT_STATUS_WRONG_PASSWORD: return ConnectionResult.WRONG_PASSWORD else: raise AssertionError(f'pwd change raised wrong error code ' f'({num:08X})') else: return ConnectionResult.SUCCESS def ldap_pwd_change(pipe, url, hostname, username, password, domain, realm, workstation, dn): lp = env_loadparm() admin_creds = KerberosCredentials() admin_creds.guess(lp) admin_creds.set_username(env_get_var_value('ADMIN_USERNAME')) admin_creds.set_password(env_get_var_value('ADMIN_PASSWORD')) admin_creds.set_kerberos_state(MUST_USE_KERBEROS) samdb = SamDB(url=url, credentials=admin_creds, lp=lp) old_utf16pw = f'"{password}"'.encode('utf-16le') new_password = generate_random_password(32, 32) new_utf16pw = f'"{new_password}"'.encode('utf-16le') msg = ldb.Message(ldb.Dn(samdb, dn)) msg['0'] = ldb.MessageElement(old_utf16pw, ldb.FLAG_MOD_DELETE, 'unicodePwd') msg['1'] = ldb.MessageElement(new_utf16pw, ldb.FLAG_MOD_ADD, 'unicodePwd') # Indicate that we're ready. This ensures we hit the right transaction # lock. pipe.send_bytes(b'0') # Wait for the main process to take out a transaction lock. if not pipe.poll(timeout=5): raise AssertionError('main process failed to indicate readiness') # Try changing the user's password. This should fail, either due to the # user's account being locked out or due to specifying the wrong password. try: samdb.modify(msg) except ldb.LdbError as err: num, estr = err.args if num != ldb.ERR_CONSTRAINT_VIOLATION: raise AssertionError(f'pwd change raised wrong error code ({err})') if f'<{werror.WERR_ACCOUNT_LOCKED_OUT:08X}:' in estr: return ConnectionResult.LOCKED_OUT elif f'<{werror.WERR_INVALID_PASSWORD:08X}:' in estr: return ConnectionResult.WRONG_PASSWORD else: raise AssertionError(f'pwd change raised wrong error code ' f'({estr})') else: return ConnectionResult.SUCCESS class LockoutTests(KDCBaseTest): def setUp(self): super().setUp() self.do_asn1_print = global_asn1_print self.do_hexdump = global_hexdump samdb = self.get_samdb() base_dn = ldb.Dn(samdb, samdb.domain_dn()) def modify_attr(attr, value): if value is None: value = [] flag = ldb.FLAG_MOD_DELETE else: value = str(value) flag = ldb.FLAG_MOD_REPLACE msg = ldb.Message(base_dn) msg[attr] = ldb.MessageElement( value, flag, attr) samdb.modify(msg) res = samdb.search(base_dn, scope=ldb.SCOPE_BASE, attrs=['lockoutDuration', 'lockoutThreshold', 'msDS-LogonTimeSyncInterval']) self.assertEqual(1, len(res)) # Reset the lockout duration as it was before. lockout_duration = res[0].get('lockoutDuration', idx=0) self.addCleanup(modify_attr, 'lockoutDuration', lockout_duration) # Set the new lockout duration: locked out accounts now stay locked # out. modify_attr('lockoutDuration', 0) # Reset the lockout threshold as it was before. lockout_threshold = res[0].get('lockoutThreshold', idx=0) self.addCleanup(modify_attr, 'lockoutThreshold', lockout_threshold) # Set the new lockout threshold. self.lockout_threshold = 3 modify_attr('lockoutThreshold', self.lockout_threshold) # Reset the logon time sync interval as it was before. sync_interval = res[0].get('msDS-LogonTimeSyncInterval', idx=0) self.addCleanup(modify_attr, 'msDS-LogonTimeSyncInterval', sync_interval) # Set the new logon time sync interval. Setting it to 0 eliminates the # need for this attribute to be updated on logon, and thus the # requirement to take out a transaction. modify_attr('msDS-LogonTimeSyncInterval', 0) # Get the old 'minPwdAge'. minPwdAge = samdb.get_minPwdAge() # Reset the 'minPwdAge' as it was before. self.addCleanup(samdb.set_minPwdAge, minPwdAge) # Set it temporarily to '0'. samdb.set_minPwdAge('0') def assertLocalSamDB(self, samdb): if samdb.url.startswith('tdb://'): return if samdb.url.startswith('mdb://'): return self.fail(f'connection to {samdb.url} is not local!') def wait_for_ready(self, pipe, future): if pipe.poll(timeout=5): return # We failed to read a response from the pipe, so see if the test raised # an exception with more information. if future.done(): exception = future.exception(timeout=0) if exception is not None: raise exception self.fail('test failed to indicate readiness') def test_lockout_transaction_kdc(self): self.do_lockout_transaction(connect_kdc) def test_lockout_transaction_ntlm(self): self.do_lockout_transaction(connect_ntlm) def test_lockout_transaction_samr(self): self.do_lockout_transaction(connect_samr) def test_lockout_transaction_samr_aes(self): if not self.gnutls_pbkdf2_support: self.skipTest('gnutls_pbkdf2() is not available') self.do_lockout_transaction(connect_samr_aes) def test_lockout_transaction_ldap_pw_change(self): self.do_lockout_transaction(ldap_pwd_change) # Tests to ensure we can handle the account being renamed. We do not test # renames with SAMR password changes, because in that case the entire # process happens inside a transaction, and the password change method only # receives the account username. By the time it searches for the account, # it will have already been renamed, and so it will always fail to find the # account. def test_lockout_transaction_rename_kdc(self): self.do_lockout_transaction(connect_kdc, rename=True) def test_lockout_transaction_rename_ntlm(self): self.do_lockout_transaction(connect_ntlm, rename=True) def test_lockout_transaction_rename_ldap_pw_change(self): self.do_lockout_transaction(ldap_pwd_change, rename=True) def test_lockout_transaction_bad_pwd_kdc(self): self.do_lockout_transaction(connect_kdc, correct_pw=False) def test_lockout_transaction_bad_pwd_ntlm(self): self.do_lockout_transaction(connect_ntlm, correct_pw=False) def test_lockout_transaction_bad_pwd_samr(self): self.do_lockout_transaction(connect_samr, correct_pw=False) def test_lockout_transaction_bad_pwd_samr_aes(self): if not self.gnutls_pbkdf2_support: self.skipTest('gnutls_pbkdf2() is not available') self.do_lockout_transaction(connect_samr_aes, correct_pw=False) def test_lockout_transaction_bad_pwd_ldap_pw_change(self): self.do_lockout_transaction(ldap_pwd_change, correct_pw=False) def test_bad_pwd_count_transaction_kdc(self): self.do_bad_pwd_count_transaction(connect_kdc) def test_bad_pwd_count_transaction_ntlm(self): self.do_bad_pwd_count_transaction(connect_ntlm) def test_bad_pwd_count_transaction_samr(self): self.do_bad_pwd_count_transaction(connect_samr) def test_bad_pwd_count_transaction_samr_aes(self): if not self.gnutls_pbkdf2_support: self.skipTest('gnutls_pbkdf2() is not available') self.do_bad_pwd_count_transaction(connect_samr_aes) def test_bad_pwd_count_transaction_ldap_pw_change(self): self.do_bad_pwd_count_transaction(ldap_pwd_change) def test_bad_pwd_count_transaction_rename_kdc(self): self.do_bad_pwd_count_transaction(connect_kdc, rename=True) def test_bad_pwd_count_transaction_rename_ntlm(self): self.do_bad_pwd_count_transaction(connect_ntlm, rename=True) def test_bad_pwd_count_transaction_rename_ldap_pw_change(self): self.do_bad_pwd_count_transaction(ldap_pwd_change, rename=True) def test_lockout_race_kdc(self): self.do_lockout_race(connect_kdc) def test_lockout_race_ntlm(self): self.do_lockout_race(connect_ntlm) def test_lockout_race_samr(self): self.do_lockout_race(connect_samr) def test_lockout_race_samr_aes(self): if not self.gnutls_pbkdf2_support: self.skipTest('gnutls_pbkdf2() is not available') self.do_lockout_race(connect_samr_aes) def test_lockout_race_ldap_pw_change(self): self.do_lockout_race(ldap_pwd_change) def test_logon_without_transaction_ntlm(self): self.do_logon_without_transaction(connect_ntlm) # Tests to ensure that the connection functions work correctly in the happy # path. def test_logon_kdc(self): self.do_logon(partial(connect_kdc, expect_error=False)) def test_logon_ntlm(self): self.do_logon(connect_ntlm) def test_logon_samr(self): self.do_logon(connect_samr) def test_logon_samr_aes(self): if not self.gnutls_pbkdf2_support: self.skipTest('gnutls_pbkdf2() is not available') self.do_logon(connect_samr_aes) def test_logon_ldap_pw_change(self): self.do_logon(ldap_pwd_change) # Test that connection without a correct password works. def do_logon(self, connect_fn): # Create the user account for testing. user_creds = self.get_cached_creds(account_type=self.AccountType.USER, use_cache=False) user_dn = user_creds.get_dn() admin_creds = self.get_admin_creds() lp = self.get_lp() # Get a connection to our local SamDB. samdb = connect_samdb(samdb_url=lp.samdb_url(), lp=lp, credentials=admin_creds) self.assertLocalSamDB(samdb) password = user_creds.get_password() # Prepare to connect to the server with a valid password. our_pipe, their_pipe = Pipe(duplex=True) # Inform the test function that it may proceed. our_pipe.send_bytes(b'0') result = connect_fn(pipe=their_pipe, url=f'ldap://{samdb.host_dns_name()}', hostname=samdb.host_dns_name(), username=user_creds.get_username(), password=password, domain=user_creds.get_domain(), realm=user_creds.get_realm(), workstation=user_creds.get_workstation(), dn=str(user_dn)) # The connection should succeed. self.assertEqual(result, ConnectionResult.SUCCESS) # Lock out the account while holding a transaction lock, then release the # lock. A logon attempt already in progress should reread the account # details and recognise the account is locked out. The account can # additionally be renamed within the transaction to ensure that, by using # the GUID, rereading the account's details still succeeds. def do_lockout_transaction(self, connect_fn, rename=False, correct_pw=True): # Create the user account for testing. user_creds = self.get_cached_creds(account_type=self.AccountType.USER, use_cache=False) user_dn = user_creds.get_dn() admin_creds = self.get_admin_creds() lp = self.get_lp() # Get a connection to our local SamDB. samdb = connect_samdb(samdb_url=lp.samdb_url(), lp=lp, credentials=admin_creds) self.assertLocalSamDB(samdb) password = user_creds.get_password() if not correct_pw: password = password[:-1] # Prepare to connect to the server. with futures.ProcessPoolExecutor(max_workers=1) as executor: our_pipe, their_pipe = Pipe(duplex=True) connect_future = executor.submit( connect_fn, pipe=their_pipe, url=f'ldap://{samdb.host_dns_name()}', hostname=samdb.host_dns_name(), username=user_creds.get_username(), password=password, domain=user_creds.get_domain(), realm=user_creds.get_realm(), workstation=user_creds.get_workstation(), dn=str(user_dn)) # Wait until the test process indicates it's ready. self.wait_for_ready(our_pipe, connect_future) # Take out a transaction. samdb.transaction_start() try: # Lock out the account. We must do it using an actual password # check like so, rather than directly with a database # modification, so that the account is also added to the # auxiliary bad password database. old_utf16pw = f'"Secret007"'.encode('utf-16le') # invalid pwd new_utf16pw = f'"Secret008"'.encode('utf-16le') msg = ldb.Message(user_dn) msg['0'] = ldb.MessageElement(old_utf16pw, ldb.FLAG_MOD_DELETE, 'unicodePwd') msg['1'] = ldb.MessageElement(new_utf16pw, ldb.FLAG_MOD_ADD, 'unicodePwd') for i in range(self.lockout_threshold): try: samdb.modify(msg) except ldb.LdbError as err: num, estr = err.args # We get an error, but the bad password count should # still be updated. self.assertEqual(num, ldb.ERR_OPERATIONS_ERROR) self.assertEqual('Failed to obtain remote address for ' 'the LDAP client while changing the ' 'password', estr) else: self.fail('pwd change should have failed') # Ensure the account is locked out. res = samdb.search( user_dn, scope=ldb.SCOPE_BASE, attrs=['msDS-User-Account-Control-Computed']) self.assertEqual(1, len(res)) uac = int(res[0].get('msDS-User-Account-Control-Computed', idx=0)) self.assertTrue(uac & dsdb.UF_LOCKOUT) # Now the bad password database has been updated, inform the # test process that it may proceed. our_pipe.send_bytes(b'0') # Wait one second to ensure the test process hits the # transaction lock. time.sleep(1) if rename: # While we're at it, rename the account to ensure that is # also safe if a race occurs. msg = ldb.Message(user_dn) new_username = self.get_new_username() msg['sAMAccountName'] = ldb.MessageElement( new_username, ldb.FLAG_MOD_REPLACE, 'sAMAccountName') samdb.modify(msg) except Exception: samdb.transaction_cancel() raise # Commit the local transaction. samdb.transaction_commit() result = connect_future.result(timeout=5) self.assertEqual(result, ConnectionResult.LOCKED_OUT) # Update the bad password count while holding a transaction lock, then # release the lock. A logon attempt already in progress should reread the # account details and ensure the bad password count is atomically # updated. The account can additionally be renamed within the transaction # to ensure that, by using the GUID, rereading the account's details still # succeeds. def do_bad_pwd_count_transaction(self, connect_fn, rename=False): # Create the user account for testing. user_creds = self.get_cached_creds(account_type=self.AccountType.USER, use_cache=False) user_dn = user_creds.get_dn() admin_creds = self.get_admin_creds() lp = self.get_lp() # Get a connection to our local SamDB. samdb = connect_samdb(samdb_url=lp.samdb_url(), lp=lp, credentials=admin_creds) self.assertLocalSamDB(samdb) # Prepare to connect to the server with an invalid password. with futures.ProcessPoolExecutor(max_workers=1) as executor: our_pipe, their_pipe = Pipe(duplex=True) connect_future = executor.submit( connect_fn, pipe=their_pipe, url=f'ldap://{samdb.host_dns_name()}', hostname=samdb.host_dns_name(), username=user_creds.get_username(), password=user_creds.get_password()[:-1], # invalid password domain=user_creds.get_domain(), realm=user_creds.get_realm(), workstation=user_creds.get_workstation(), dn=str(user_dn)) # Wait until the test process indicates it's ready. self.wait_for_ready(our_pipe, connect_future) # Take out a transaction. samdb.transaction_start() try: # Inform the test process that it may proceed. our_pipe.send_bytes(b'0') # Wait one second to ensure the test process hits the # transaction lock. time.sleep(1) # Set badPwdCount to 1. msg = ldb.Message(user_dn) now = int(time.time()) bad_pwd_time = unix2nttime(now) msg['badPwdCount'] = ldb.MessageElement( '1', ldb.FLAG_MOD_REPLACE, 'badPwdCount') msg['badPasswordTime'] = ldb.MessageElement( str(bad_pwd_time), ldb.FLAG_MOD_REPLACE, 'badPasswordTime') if rename: # While we're at it, rename the account to ensure that is # also safe if a race occurs. new_username = self.get_new_username() msg['sAMAccountName'] = ldb.MessageElement( new_username, ldb.FLAG_MOD_REPLACE, 'sAMAccountName') samdb.modify(msg) # Ensure the account is not yet locked out. res = samdb.search( user_dn, scope=ldb.SCOPE_BASE, attrs=['msDS-User-Account-Control-Computed']) self.assertEqual(1, len(res)) uac = int(res[0].get('msDS-User-Account-Control-Computed', idx=0)) self.assertFalse(uac & dsdb.UF_LOCKOUT) except Exception: samdb.transaction_cancel() raise # Commit the local transaction. samdb.transaction_commit() result = connect_future.result(timeout=5) self.assertEqual(result, ConnectionResult.WRONG_PASSWORD, result) # Check that badPwdCount has now increased to 2. res = samdb.search(user_dn, scope=ldb.SCOPE_BASE, attrs=['badPwdCount']) self.assertEqual(1, len(res)) bad_pwd_count = int(res[0].get('badPwdCount', idx=0)) self.assertEqual(2, bad_pwd_count) # Attempt to log in to the account with an incorrect password, using # lockoutThreshold+1 simultaneous attempts. We should get three 'wrong # password' errors and one 'locked out' error, showing that the bad # password count is checked and incremented atomically. def do_lockout_race(self, connect_fn): # Create the user account for testing. user_creds = self.get_cached_creds(account_type=self.AccountType.USER, use_cache=False) user_dn = user_creds.get_dn() admin_creds = self.get_admin_creds() lp = self.get_lp() # Get a connection to our local SamDB. samdb = connect_samdb(samdb_url=lp.samdb_url(), lp=lp, credentials=admin_creds) self.assertLocalSamDB(samdb) # Prepare to connect to the server with an invalid password, using four # simultaneous requests. Only three of those attempts should get # through before the account is locked out. num_attempts = self.lockout_threshold + 1 with futures.ProcessPoolExecutor(max_workers=num_attempts) as executor: connect_futures = [] our_pipes = [] for i in range(num_attempts): our_pipe, their_pipe = Pipe(duplex=True) our_pipes.append(our_pipe) connect_future = executor.submit( connect_fn, pipe=their_pipe, url=f'ldap://{samdb.host_dns_name()}', hostname=samdb.host_dns_name(), username=user_creds.get_username(), password=user_creds.get_password()[:-1], # invalid pw domain=user_creds.get_domain(), realm=user_creds.get_realm(), workstation=user_creds.get_workstation(), dn=str(user_dn)) connect_futures.append(connect_future) # Wait until the test process indicates it's ready. self.wait_for_ready(our_pipe, connect_future) # Take out a transaction. samdb.transaction_start() try: # Inform the test processes that they may proceed. for our_pipe in our_pipes: our_pipe.send_bytes(b'0') # Wait one second to ensure the test processes hit the # transaction lock. time.sleep(1) except Exception: samdb.transaction_cancel() raise # Commit the local transaction. samdb.transaction_commit() lockouts = 0 wrong_passwords = 0 for i, connect_future in enumerate(connect_futures): result = connect_future.result(timeout=5) if result == ConnectionResult.LOCKED_OUT: lockouts += 1 elif result == ConnectionResult.WRONG_PASSWORD: wrong_passwords += 1 else: self.fail(f'process {i} gave an unexpected result ' f'{result}') self.assertEqual(wrong_passwords, self.lockout_threshold) self.assertEqual(lockouts, num_attempts - self.lockout_threshold) # Ensure the account is now locked out. res = samdb.search( user_dn, scope=ldb.SCOPE_BASE, attrs=['badPwdCount', 'msDS-User-Account-Control-Computed']) self.assertEqual(1, len(res)) bad_pwd_count = int(res[0].get('badPwdCount', idx=0)) self.assertEqual(self.lockout_threshold, bad_pwd_count) uac = int(res[0].get('msDS-User-Account-Control-Computed', idx=0)) self.assertTrue(uac & dsdb.UF_LOCKOUT) # Test that logon is possible even while we locally hold a transaction # lock. This test only works with NTLM authentication; Kerberos # authentication must take out a transaction to update the logonCount # attribute, and LDAP and SAMR password changes both take out a transaction # to effect the password change. NTLM is the only logon method that does # not require a transaction, and can thus be performed while we're holding # the lock. def do_logon_without_transaction(self, connect_fn): # Create the user account for testing. user_creds = self.get_cached_creds(account_type=self.AccountType.USER, use_cache=False) user_dn = user_creds.get_dn() admin_creds = self.get_admin_creds() lp = self.get_lp() # Get a connection to our local SamDB. samdb = connect_samdb(samdb_url=lp.samdb_url(), lp=lp, credentials=admin_creds) self.assertLocalSamDB(samdb) password = user_creds.get_password() # Prepare to connect to the server with a valid password. with futures.ProcessPoolExecutor(max_workers=1) as executor: our_pipe, their_pipe = Pipe(duplex=True) connect_future = executor.submit( connect_fn, pipe=their_pipe, url=f'ldap://{samdb.host_dns_name()}', hostname=samdb.host_dns_name(), username=user_creds.get_username(), password=password, domain=user_creds.get_domain(), realm=user_creds.get_realm(), workstation=user_creds.get_workstation(), dn=str(user_dn)) # Wait until the test process indicates it's ready. self.wait_for_ready(our_pipe, connect_future) # Take out a transaction. samdb.transaction_start() try: # Inform the test process that it may proceed. our_pipe.send_bytes(b'0') # The connection should succeed, despite our holding a # transaction. result = connect_future.result(timeout=5) self.assertEqual(result, ConnectionResult.SUCCESS) except Exception: samdb.transaction_cancel() raise # Commit the local transaction. samdb.transaction_commit() if __name__ == '__main__': global_asn1_print = False global_hexdump = False import unittest unittest.main()