diff options
Diffstat (limited to 'python/samba/tests/password_hash.py')
| -rw-r--r-- | python/samba/tests/password_hash.py | 335 |
1 files changed, 335 insertions, 0 deletions
diff --git a/python/samba/tests/password_hash.py b/python/samba/tests/password_hash.py new file mode 100644 index 0000000..1b7af7d --- /dev/null +++ b/python/samba/tests/password_hash.py @@ -0,0 +1,335 @@ +# Tests for Tests for source4/dsdb/samdb/ldb_modules/password_hash.c +# +# Copyright (C) Catalyst IT Ltd. 2017 +# +# 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/>. +# + +""" +Base class for tests for source4/dsdb/samdb/ldb_modules/password_hash.c +""" + +from samba.credentials import Credentials +from samba.samdb import SamDB +from samba.auth import system_session +from samba.tests import TestCase +from samba.ndr import ndr_unpack +from samba.dcerpc import drsblobs +from samba.dcerpc.samr import DOMAIN_PASSWORD_STORE_CLEARTEXT +from samba.dsdb import UF_ENCRYPTED_TEXT_PASSWORD_ALLOWED +from samba.tests import delete_force +from samba.tests.password_test import PasswordCommon +import ldb +import samba +import binascii +from hashlib import md5 +import crypt + + +USER_NAME = "PasswordHashTestUser" +USER_PASS = samba.generate_random_password(32, 32) +UPN = "PWHash@User.Principle" + +# Get named package from the passed supplemental credentials +# +# returns the package and it's position within the supplemental credentials + + +def get_package(sc, name): + if sc is None: + return None + + idx = 0 + for p in sc.sub.packages: + idx += 1 + if name == p.name: + return (idx, p) + + return None + +# Calculate the MD5 password digest from the supplied user, realm and password +# + + +def calc_digest(user, realm, password): + + data = "%s:%s:%s" % (user, realm, password) + if isinstance(data, str): + data = data.encode('utf8') + + return md5(data).hexdigest() + + +class PassWordHashTests(TestCase): + + def setUp(self): + self.lp = samba.tests.env_loadparm() + super().setUp() + + def set_store_cleartext(self, cleartext): + # get the current pwdProperties + pwdProperties = self.ldb.get_pwdProperties() + # update the clear-text properties flag + props = int(pwdProperties) + if cleartext: + props |= DOMAIN_PASSWORD_STORE_CLEARTEXT + else: + props &= ~DOMAIN_PASSWORD_STORE_CLEARTEXT + self.ldb.set_pwdProperties(str(props)) + + # Add a user to ldb, this will exercise the password_hash code + # and calculate the appropriate supplemental credentials + def add_user(self, options=None, clear_text=False, ldb=None): + # set any needed options + if options is not None: + for (option, value) in options: + self.lp.set(option, value) + + if ldb is None: + self.creds = Credentials() + self.session = system_session() + self.creds.guess(self.lp) + self.session = system_session() + self.ldb = SamDB(session_info=self.session, + credentials=self.creds, + lp=self.lp) + else: + self.ldb = ldb + + res = self.ldb.search(base=self.ldb.get_config_basedn(), + expression="ncName=%s" % self.ldb.get_default_basedn(), + attrs=["nETBIOSName"]) + self.netbios_domain = str(res[0]["nETBIOSName"][0]) + self.dns_domain = self.ldb.domain_dns_name() + + # Gets back the basedn + base_dn = self.ldb.domain_dn() + + # Gets back the configuration basedn + configuration_dn = self.ldb.get_config_basedn().get_linearized() + + # permit password changes during this test + PasswordCommon.allow_password_changes(self, self.ldb) + + self.base_dn = self.ldb.domain_dn() + + account_control = 0 + if clear_text: + # Restore the current domain setting on exit. + pwdProperties = self.ldb.get_pwdProperties() + self.addCleanup(self.ldb.set_pwdProperties, pwdProperties) + # Update the domain setting + self.set_store_cleartext(clear_text) + account_control |= UF_ENCRYPTED_TEXT_PASSWORD_ALLOWED + + # (Re)adds the test user USER_NAME with password USER_PASS + # and userPrincipalName UPN + delete_force(self.ldb, "cn=" + USER_NAME + ",cn=users," + self.base_dn) + self.ldb.add({ + "dn": "cn=" + USER_NAME + ",cn=users," + self.base_dn, + "objectclass": "user", + "sAMAccountName": USER_NAME, + "userPassword": USER_PASS, + "userPrincipalName": UPN, + "userAccountControl": str(account_control) + }) + + # Get the supplemental credentials for the user under test + def get_supplemental_creds(self): + base = "cn=" + USER_NAME + ",cn=users," + self.base_dn + res = self.ldb.search(scope=ldb.SCOPE_BASE, + base=base, + attrs=["supplementalCredentials"]) + self.assertIs(True, len(res) > 0) + obj = res[0] + sc_blob = obj["supplementalCredentials"][0] + sc = ndr_unpack(drsblobs.supplementalCredentialsBlob, sc_blob) + return sc + + # Calculate and validate a Wdigest value + def check_digest(self, user, realm, password, digest): + expected = calc_digest(user, realm, password) + actual = binascii.hexlify(bytearray(digest)).decode('utf8') + error = "Digest expected[%s], actual[%s], " \ + "user[%s], realm[%s], pass[%s]" % \ + (expected, actual, user, realm, password) + self.assertEqual(expected, actual, error) + + # Check all of the 29 expected WDigest values + # + def check_wdigests(self, digests): + + self.assertEqual(29, digests.num_hashes) + + # Using the n-1 pattern in the array indexes to make it easier + # to check the tests against the spec and the samba-tool user tests. + self.check_digest(USER_NAME, + self.netbios_domain, + USER_PASS, + digests.hashes[1 - 1].hash) + self.check_digest(USER_NAME.lower(), + self.netbios_domain.lower(), + USER_PASS, + digests.hashes[2 - 1].hash) + self.check_digest(USER_NAME.upper(), + self.netbios_domain.upper(), + USER_PASS, + digests.hashes[3 - 1].hash) + self.check_digest(USER_NAME, + self.netbios_domain.upper(), + USER_PASS, + digests.hashes[4 - 1].hash) + self.check_digest(USER_NAME, + self.netbios_domain.lower(), + USER_PASS, + digests.hashes[5 - 1].hash) + self.check_digest(USER_NAME.upper(), + self.netbios_domain.lower(), + USER_PASS, + digests.hashes[6 - 1].hash) + self.check_digest(USER_NAME.lower(), + self.netbios_domain.upper(), + USER_PASS, + digests.hashes[7 - 1].hash) + self.check_digest(USER_NAME, + self.dns_domain, + USER_PASS, + digests.hashes[8 - 1].hash) + self.check_digest(USER_NAME.lower(), + self.dns_domain.lower(), + USER_PASS, + digests.hashes[9 - 1].hash) + self.check_digest(USER_NAME.upper(), + self.dns_domain.upper(), + USER_PASS, + digests.hashes[10 - 1].hash) + self.check_digest(USER_NAME, + self.dns_domain.upper(), + USER_PASS, + digests.hashes[11 - 1].hash) + self.check_digest(USER_NAME, + self.dns_domain.lower(), + USER_PASS, + digests.hashes[12 - 1].hash) + self.check_digest(USER_NAME.upper(), + self.dns_domain.lower(), + USER_PASS, + digests.hashes[13 - 1].hash) + self.check_digest(USER_NAME.lower(), + self.dns_domain.upper(), + USER_PASS, + digests.hashes[14 - 1].hash) + self.check_digest(UPN, + "", + USER_PASS, + digests.hashes[15 - 1].hash) + self.check_digest(UPN.lower(), + "", + USER_PASS, + digests.hashes[16 - 1].hash) + self.check_digest(UPN.upper(), + "", + USER_PASS, + digests.hashes[17 - 1].hash) + + name = "%s\\%s" % (self.netbios_domain, USER_NAME) + self.check_digest(name, + "", + USER_PASS, + digests.hashes[18 - 1].hash) + + name = "%s\\%s" % (self.netbios_domain.lower(), USER_NAME.lower()) + self.check_digest(name, + "", + USER_PASS, + digests.hashes[19 - 1].hash) + + name = "%s\\%s" % (self.netbios_domain.upper(), USER_NAME.upper()) + self.check_digest(name, + "", + USER_PASS, + digests.hashes[20 - 1].hash) + self.check_digest(USER_NAME, + "Digest", + USER_PASS, + digests.hashes[21 - 1].hash) + self.check_digest(USER_NAME.lower(), + "Digest", + USER_PASS, + digests.hashes[22 - 1].hash) + self.check_digest(USER_NAME.upper(), + "Digest", + USER_PASS, + digests.hashes[23 - 1].hash) + self.check_digest(UPN, + "Digest", + USER_PASS, + digests.hashes[24 - 1].hash) + self.check_digest(UPN.lower(), + "Digest", + USER_PASS, + digests.hashes[25 - 1].hash) + self.check_digest(UPN.upper(), + "Digest", + USER_PASS, + digests.hashes[26 - 1].hash) + name = "%s\\%s" % (self.netbios_domain, USER_NAME) + self.check_digest(name, + "Digest", + USER_PASS, + digests.hashes[27 - 1].hash) + + name = "%s\\%s" % (self.netbios_domain.lower(), USER_NAME.lower()) + self.check_digest(name, + "Digest", + USER_PASS, + digests.hashes[28 - 1].hash) + + name = "%s\\%s" % (self.netbios_domain.upper(), USER_NAME.upper()) + self.check_digest(name, + "Digest", + USER_PASS, + digests.hashes[29 - 1].hash) + + def checkUserPassword(self, up, expected): + + # Check we've received the correct number of hashes + self.assertEqual(len(expected), up.num_hashes) + + i = 0 + for (tag, alg, rounds) in expected: + self.assertEqual(tag, up.hashes[i].scheme) + + data = up.hashes[i].value.decode('utf8').split("$") + # Check we got the expected crypt algorithm + self.assertEqual(alg, data[1]) + + if rounds is None: + cmd = "$%s$%s" % (alg, data[2]) + else: + cmd = "$%s$rounds=%d$%s" % (alg, rounds, data[3]) + + # Calculate the expected hash value + expected = crypt.crypt(USER_PASS, cmd) + self.assertEqual(expected, up.hashes[i].value.decode('utf8')) + i += 1 + + # Check that the correct nt_hash was stored for userPassword + def checkNtHash(self, password, nt_hash): + creds = Credentials() + creds.set_anonymous() + creds.set_password(password) + expected = creds.get_nt_hash() + actual = bytearray(nt_hash) + self.assertEqual(expected, actual) |