fixing LM hash parsing issue for win2016, +tests

aesedb/_version.py

@@ -1,4 +1,4 @@
-__version__ = "0.1.0"
+__version__ = "0.1.1"
 __banner__ = \
 """
 # aesedb %s 

aesedb/security/ntds.py

@@ -172,7 +172,7 @@ def __decrypt_hash(self, secret: UserSecrets, record, db_cursor, with_history =
 				encryptedLMHash = CRYPTED_HASH.from_bytes(record[NAME_TO_INTERNAL['dBCSPwd']])
 				if encryptedLMHash.Header[:4] == b'\x13\x00\x00\x00':
 					# Win2016 TP4 decryption is different
-					encryptedLMHash = CRYPTED_HASHW16(record[NAME_TO_INTERNAL['dBCSPwd']])
+					encryptedLMHash = CRYPTED_HASHW16.from_bytes(record[NAME_TO_INTERNAL['dBCSPwd']])
 					pekIndex = encryptedLMHash.Header[4]
 					ctx = AES(self.__PEK[pekIndex], MODE_CBC, IV=encryptedLMHash.KeyMaterial)
 					dec_hash_temp = ctx.decrypt(encryptedLMHash.EncryptedHash[:16])

coverage.sh

@@ -0,0 +1 @@
+pytest --cov-config=.coveragerc --cov-report html --cov=aesedb tests/

setup.py

@@ -46,7 +46,7 @@
 	## these are only necessary for the command line tool
 	## lib can work without additional deps
 	install_requires=[
-		'unicrypto>=0.0.5',
+		'unicrypto>=0.0.9',
 		'aiowinreg>=0.0.7',
 		'tqdm',
 		'colorama',

tests/__init__.py

@@ -1,99 +0,0 @@
-
-import unittest
-import asyncio
-import pathlib
-
-
-class NTDSParse:
-	async def runner(self, parser, out_q):
-		_, err = await parser.run()
-		if err is not None:
-			await out_q.put((None, None, err, True))
-
-	async def parser_test(self, crypolibname):
-		try:
-			import unicrypto
-			unicrypto.use_library(crypolibname)
-			from aesedb.examples.ntdsparse import NTDSParserConsole
-			from .knownvalues import knowngood
-			basepath = pathlib.Path(__file__).parent.resolve()
-			ntdsfile = str(basepath.joinpath('ntds.dit'))
-			systemfile = str(basepath.joinpath('SYSTEM'))
-
-			lm_deleted = {}
-			nt_deleted = {}
-
-			out_q = asyncio.Queue()
-			parser = NTDSParserConsole(systemfile, ntdsfile, show_progress = True, outfile = None, ext_result_q=out_q, count_total = False)
-			x = asyncio.create_task(self.runner(parser, out_q))
-			while True:
-				usersecret, total, err, eof = await out_q.get()
-
-				if err is not None:
-					raise err
-				if eof is True:
-					break
-				if usersecret is None:
-					continue
-
-				for secret in str(usersecret).split('\r\n'):
-					ht, *t = secret.split(':')
-					if ht == 'ntlm':
-						domain, user, rid, sid, lm, nt, lastset_or_history = t
-						if lm not in lm_deleted:
-							del knowngood['lmhashes'][lm]
-							lm_deleted[lm] = 1
-						if nt not in nt_deleted:
-							del knowngood['nthashes'][nt]
-							nt_deleted[nt] = 1
-					elif ht == 'kerberos':
-						domain, user, sid, kt, secret = tuple(t)
-					elif ht == 'cleartext':
-						domain, user, sid, secret = tuple(t)
-
-			if len(knowngood['lmhashes']) > 0:
-				raise Exception('LM hashes remaining: %s' % len(knowngood['lmhashes']))
-
-			if len(knowngood['nthashes']) > 0:
-				raise Exception('NT hashes remaining: %s' % len(knowngood['nthashes']))
-			with open('remaining_nt.txt', 'w', newline = '') as f:
-				for x in knowngood['nthashes']:
-					f.write(x + '\r\n')
-			return True, None
-		except Exception as e:
-			print(e)
-			return False, e
-
-
-#class pure(NTDSParse, unittest.IsolatedAsyncioTestCase):	
-#	async def test_parser(self):
-#		_, err = await self.parser_test('pure')
-#		if err is not None:
-#			raise err
-#
-#class Crypto(NTDSParse, unittest.IsolatedAsyncioTestCase):
-#	async def test_parser(self):
-#		_, err = await self.parser_test('pure')
-#		if err is not None:
-#			raise err
-#
-#class pyCryptodome(NTDSParse, unittest.IsolatedAsyncioTestCase):
-#	async def test_parser(self):
-#		_, err = await self.parser_test('pure')
-#		if err is not None:
-#			raise err
-#
-#class MBEDTLS(NTDSParse, unittest.IsolatedAsyncioTestCase):
-#	async def test_parser(self):
-#		_, err = await self.parser_test('pure')
-#		if err is not None:
-#			raise err
-
-class cryptography(NTDSParse, unittest.IsolatedAsyncioTestCase):
-	async def test_parser(self):
-		_, err = await self.parser_test('pure')
-		if err is not None:
-			raise err
-
-if __name__ == '__main__':
-	unittest.main()

tests/test_parsing.py

@@ -0,0 +1,126 @@
+import pathlib
+from aesedb.security.ntds import NTDS
+from aesedb.eesentdb import ESENT_DB
+from aesedb.utils.afile import AFile
+from aesedb.security.ntds import NTDS
+from aesedb.utils.systemhive import SYSTEM
+from aiowinreg.ahive import AIOWinRegHive
+import pytest
+import asyncio
+
+
+
+CURRENT_FILE_PATH = pathlib.Path(__file__).parent.absolute()
+TESTDATA_DIR = CURRENT_FILE_PATH.joinpath('testdata')
+
+async def parse_ntds(system_fname, ntds_fname):
+	try:
+		regfile = AFile(system_fname)
+		reghive = AIOWinRegHive(regfile)
+		syshive = SYSTEM(reghive)
+		print('Getting bootkey...')
+		bootkey = await syshive.get_bootkey()
+		print('Bootkey: %s' % bootkey.hex())
+		file = AFile(ntds_fname)
+
+		print('Parsing ESENT DB...')
+		db = ESENT_DB(file)
+		_, err = await db.parse()
+		if err is not None:
+			raise err
+
+		print('Getting total row count...')
+		total, err = await db.get_rowcnt('datatable')
+		if err is not None:
+			raise err
+
+		print('Dumping secrets...')
+		secrets = []
+		ntds = NTDS(db, bootkey)
+
+		found_admin = False
+		found_test2 = False
+		found_krbtgt = False
+		async for secret, err in ntds.dump_secrets(with_history=True, ignore_errors = False):
+			if err is not None:
+				raise err
+			if secret is None:
+				continue
+			secrets.append(secret)
+			if found_admin is False:
+				if str(secret).find('Administrator') != -1:
+					found_admin = True
+			if found_test2 is False:
+				if str(secret).find('test2') != -1:
+					found_test2 = True
+			if found_krbtgt is False:
+				if str(secret).find('krbtgt') != -1:
+					found_krbtgt = True
+
+		if found_admin is False:
+			raise Exception('Administrator not found')
+		if found_test2 is False:
+			raise Exception('test2 not found')
+		if found_krbtgt is False:
+			raise Exception('krbtgt not found')
+		return secrets, None
+	except Exception as e:
+		return None, e
+
+#@pytest.mark.asyncio
+#async def test_2008_64():
+#	ntds_fname = TESTDATA_DIR.joinpath('win2008_32', 'ntds.dit')
+#	system_fname = TESTDATA_DIR.joinpath('win2008_32', 'SYSTEM')
+#	res, err = await parse_ntds(system_fname, ntds_fname)
+#	assert res is None
+
+@pytest.mark.asyncio
+async def test_2008r2_64():
+	ntds_fname = TESTDATA_DIR.joinpath('win2008r2_64', 'ntds.dit')
+	system_fname = TESTDATA_DIR.joinpath('win2008r2_64', 'SYSTEM')
+	secrets, err = await parse_ntds(system_fname, ntds_fname)
+	assert err is None
+	assert len(secrets) > 0
+
+@pytest.mark.asyncio
+async def test_2012_64():
+	ntds_fname = TESTDATA_DIR.joinpath('win2012_64', 'ntds.dit')
+	system_fname = TESTDATA_DIR.joinpath('win2012_64', 'SYSTEM')
+	secrets, err = await parse_ntds(system_fname, ntds_fname)
+	assert err is None
+	assert len(secrets) > 0
+
+@pytest.mark.asyncio
+async def test_2016_64():
+	ntds_fname = TESTDATA_DIR.joinpath('win2016_64', 'ntds.dit')
+	system_fname = TESTDATA_DIR.joinpath('win2016_64', 'SYSTEM')
+	secrets, err = await parse_ntds(system_fname, ntds_fname)
+	assert err is None
+	assert len(secrets) > 0
+
+@pytest.mark.asyncio
+async def test_2019_64():
+	ntds_fname = TESTDATA_DIR.joinpath('win2019_64', 'ntds.dit')
+	system_fname = TESTDATA_DIR.joinpath('win2019_64', 'SYSTEM')
+	secrets, err = await parse_ntds(system_fname, ntds_fname)
+	assert err is None
+	assert len(secrets) > 0
+
+@pytest.mark.asyncio
+async def test_2022_64():
+	ntds_fname = TESTDATA_DIR.joinpath('win2022_64', 'ntds.dit')
+	system_fname = TESTDATA_DIR.joinpath('win2022_64', 'SYSTEM')
+	secrets, err = await parse_ntds(system_fname, ntds_fname)
+	assert err is None
+	assert len(secrets) > 0
+
+@pytest.mark.asyncio
+async def test_2019_64_clear():
+	ntds_fname = TESTDATA_DIR.joinpath('win2019_64_clear', 'ntds.dit')
+	system_fname = TESTDATA_DIR.joinpath('win2019_64_clear', 'SYSTEM')
+	secrets, err = await parse_ntds(system_fname, ntds_fname)
+	assert err is None
+	assert len(secrets) > 0
+
+if __name__ == '__main__':
+	asyncio.run(test_2012_64())