--- /dev/null
+# Unix SMB/CIFS implementation. Tests for dsdb
+# Copyright (C) Andrew Bartlett <abartlet@samba.org> 2024
+#
+# 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/>.
+#
+
+"""These tests want to be run on a freshly provisioned domain that has
+not been greatly modified by other tests (which at the time of writing
+probably means 'chgdcpass').
+
+Tests here should only read the database.
+
+This is to avoid flapping tests.
+"""
+
+from samba.credentials import Credentials
+from samba.samdb import SamDB
+from samba.auth import system_session
+from samba.tests import TestCase
+import ldb
+import samba
+
+class DsdbQuietEnvTests(TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls.lp = samba.tests.env_loadparm()
+ cls.creds = Credentials()
+ cls.creds.guess(cls.lp)
+ cls.session = system_session()
+ cls.samdb = SamDB(session_info=cls.session,
+ credentials=cls.creds,
+ lp=cls.lp)
+
+ def test_gkdi_create_root_key_wrong_version(self):
+
+ server_config_dn = self.samdb.get_config_basedn()
+ server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
+ "CN=Server Configuration," +
+ "CN=Group Key Distribution Service," +
+ "CN=Services")
+ res = self.samdb.search(base=server_config_dn,
+ scope=ldb.SCOPE_BASE,
+ attrs=["msKds-Version"])
+
+ self.assertEqual(len(res), 1)
+
+ msg = res[0]
+ version = int(msg["msKds-Version"][0])
+ self.assertEqual(version, 1)
+
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-Version": [str(version)]},
+ ldb.FLAG_MOD_REPLACE))
+ self.samdb.modify(ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-Version": ["2"]},
+ ldb.FLAG_MOD_REPLACE))
+
+ try:
+ self.samdb.new_gkdi_root_key()
+ self.fail("Creating key with invalid version should fail")
+ except ldb.LdbError as e:
+ (enum, estr) = e.args
+ self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
+
+ def test_gkdi_create_root_key_4096(self):
+
+ server_config_dn = self.samdb.get_config_basedn()
+ server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
+ "CN=Server Configuration," +
+ "CN=Group Key Distribution Service," +
+ "CN=Services")
+ res = self.samdb.search(base=server_config_dn,
+ scope=ldb.SCOPE_BASE,
+ attrs=["msKds-PublicKeyLength"])
+
+ self.assertEqual(len(res), 1)
+
+ msg = res[0]
+ if "msKds-PublicKeyLength" in msg:
+ keylen = msg[0]["msKds-PublicKeyLength"]
+ # Ensure test still tests something in the future, if the default changes
+ self.assertNotEqual(keylen, 4096)
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-PublicKeyLength": [str(keylen)]},
+ ldb.FLAG_MOD_REPLACE))
+ else:
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-PublicKeyLength": []},
+ ldb.FLAG_MOD_DELETE))
+
+ self.samdb.modify(ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-PublicKeyLength": ["4096"]},
+ ldb.FLAG_MOD_REPLACE))
+
+ dn = self.samdb.new_gkdi_root_key()
+
+ root_key_res = self.samdb.search(base=dn,
+ scope=ldb.SCOPE_BASE)
+ self.assertEqual(len(root_key_res), 1)
+ root_key = root_key_res[0]
+
+ self.assertEqual(int(root_key["msKds-PublicKeyLength"][0]), 4096)
+ self.assertEqual(str(root_key["msKds-KDFAlgorithmID"][0]), "SP800_108_CTR_HMAC")
+ self.assertEqual(str(root_key["msKds-SecretAgreementAlgorithmID"][0]), "DH")
+ self.assertEqual(int(root_key["msKds-Version"][0]), 1)
+
+ def test_gkdi_create_root_key_priv_1024(self):
+
+ server_config_dn = self.samdb.get_config_basedn()
+ server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
+ "CN=Server Configuration," +
+ "CN=Group Key Distribution Service," +
+ "CN=Services")
+ res = self.samdb.search(base=server_config_dn,
+ scope=ldb.SCOPE_BASE,
+ attrs=["msKds-PrivateKeyLength"])
+
+ self.assertEqual(len(res), 1)
+
+ msg = res[0]
+ if "msKds-PrivateKeyLength" in msg:
+ keylen = msg["msKds-PrivateKeyLength"]
+ # Ensure test still tests something in the future, if the default changes
+ self.assertNotEqual(keylen, 1024)
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-PrivateKeyLength": [str(keylen)]},
+ ldb.FLAG_MOD_REPLACE))
+ else:
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-PrivateKeyLength": []},
+ ldb.FLAG_MOD_DELETE))
+
+ self.samdb.modify(ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-PrivateKeyLength": ["1024"]},
+ ldb.FLAG_MOD_REPLACE))
+
+ dn = self.samdb.new_gkdi_root_key()
+
+ root_key_res = self.samdb.search(base=dn,
+ scope=ldb.SCOPE_BASE)
+ self.assertEqual(len(root_key_res), 1)
+ root_key = root_key_res[0]
+
+ self.assertEqual(int(root_key["msKds-PrivateKeyLength"][0]), 1024)
+ self.assertEqual(str(root_key["msKds-KDFAlgorithmID"][0]), "SP800_108_CTR_HMAC")
+ self.assertEqual(str(root_key["msKds-SecretAgreementAlgorithmID"][0]), "DH")
+ self.assertEqual(int(root_key["msKds-Version"][0]), 1)
+
+ def test_gkdi_create_root_key_bad_alg(self):
+ server_config_dn = self.samdb.get_config_basedn()
+ server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
+ "CN=Server Configuration," +
+ "CN=Group Key Distribution Service," +
+ "CN=Services")
+ res = self.samdb.search(base=server_config_dn,
+ scope=ldb.SCOPE_BASE,
+ attrs=["msKds-KDFAlgorithmID"])
+
+ self.assertEqual(len(res), 1)
+
+ msg = res[0]
+ if "msKds-KDFAlgorithmID" in msg:
+ alg = msg["msKds-KDFAlgorithmID"][0]
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-KDFAlgorithmID": [alg]},
+ ldb.FLAG_MOD_REPLACE))
+ else:
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-KDFAlgorithmID": []},
+ ldb.FLAG_MOD_DELETE))
+
+ self.samdb.modify(ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-KDFAlgorithmID": ["NO_AN_ALG"]},
+ ldb.FLAG_MOD_REPLACE))
+
+ try:
+ self.samdb.new_gkdi_root_key()
+ self.fail("Creating key with invalid algorithm should fail")
+ except ldb.LdbError as e:
+ (enum, estr) = e.args
+ self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
+
+ def test_gkdi_create_root_key_good_alg(self):
+ server_config_dn = self.samdb.get_config_basedn()
+ server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
+ "CN=Server Configuration," +
+ "CN=Group Key Distribution Service," +
+ "CN=Services")
+ res = self.samdb.search(base=server_config_dn,
+ scope=ldb.SCOPE_BASE,
+ attrs=["msKds-KDFAlgorithmID"])
+
+ self.assertEqual(len(res), 1)
+
+ msg = res[0]
+ if "msKds-KDFAlgorithmID" in msg:
+ alg = msg["msKds-KDFAlgorithmID"][0]
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-KDFAlgorithmID": [alg]},
+ ldb.FLAG_MOD_REPLACE))
+ else:
+ self.addCleanup(self.samdb.modify,
+ ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-KDFAlgorithmID": []},
+ ldb.FLAG_MOD_DELETE))
+
+ self.samdb.modify(ldb.Message.from_dict(self.samdb,
+ {"dn": msg["dn"],
+ "msKds-KDFAlgorithmID": ["SP800_108_CTR_HMAC"]},
+ ldb.FLAG_MOD_REPLACE))
+
+ dn = self.samdb.new_gkdi_root_key()
+
+ root_key_res = self.samdb.search(base=dn,
+ scope=ldb.SCOPE_BASE)
+ self.assertEqual(len(root_key_res), 1)
+ root_key = root_key_res[0]
+
+ self.assertEqual(int(root_key["msKds-PublicKeyLength"][0]), 2048)
+ self.assertEqual(str(root_key["msKds-KDFAlgorithmID"][0]), "SP800_108_CTR_HMAC")
+ self.assertEqual(str(root_key["msKds-SecretAgreementAlgorithmID"][0]), "DH")
+ self.assertEqual(int(root_key["msKds-Version"][0]), 1)
expression=f"(&(objectClass = msKds-ProvRootKey)(msKds-UseStartTime<={min_use_start_time}))")
self.assertGreater(len(res), 0)
-
- def test_gkdi_create_root_key_wrong_version(self):
-
- server_config_dn = self.samdb.get_config_basedn()
- server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
- "CN=Server Configuration," +
- "CN=Group Key Distribution Service," +
- "CN=Services")
- res = self.samdb.search(base=server_config_dn,
- scope=ldb.SCOPE_BASE,
- attrs=["msKds-Version"])
-
- self.assertEqual(len(res), 1)
-
- msg = res[0]
- version = int(msg["msKds-Version"][0])
- self.assertEqual(version, 1)
-
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-Version": [str(version)]},
- ldb.FLAG_MOD_REPLACE))
- self.samdb.modify(ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-Version": ["2"]},
- ldb.FLAG_MOD_REPLACE))
-
- try:
- self.samdb.new_gkdi_root_key()
- self.fail("Creating key with invalid version should fail")
- except ldb.LdbError as e:
- (enum, estr) = e.args
- self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
-
- def test_gkdi_create_root_key_4096(self):
-
- server_config_dn = self.samdb.get_config_basedn()
- server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
- "CN=Server Configuration," +
- "CN=Group Key Distribution Service," +
- "CN=Services")
- res = self.samdb.search(base=server_config_dn,
- scope=ldb.SCOPE_BASE,
- attrs=["msKds-PublicKeyLength"])
-
- self.assertEqual(len(res), 1)
-
- msg = res[0]
- if "msKds-PublicKeyLength" in msg:
- keylen = msg[0]["msKds-PublicKeyLength"]
- # Ensure test still tests something in the future, if the default changes
- self.assertNotEqual(keylen, 4096)
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-PublicKeyLength": [str(keylen)]},
- ldb.FLAG_MOD_REPLACE))
- else:
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-PublicKeyLength": []},
- ldb.FLAG_MOD_DELETE))
-
- self.samdb.modify(ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-PublicKeyLength": ["4096"]},
- ldb.FLAG_MOD_REPLACE))
-
- dn = self.samdb.new_gkdi_root_key()
-
- root_key_res = self.samdb.search(base=dn,
- scope=ldb.SCOPE_BASE)
- self.assertEqual(len(root_key_res), 1)
- root_key = root_key_res[0]
-
- self.assertEqual(int(root_key["msKds-PublicKeyLength"][0]), 4096)
- self.assertEqual(str(root_key["msKds-KDFAlgorithmID"][0]), "SP800_108_CTR_HMAC")
- self.assertEqual(str(root_key["msKds-SecretAgreementAlgorithmID"][0]), "DH")
- self.assertEqual(int(root_key["msKds-Version"][0]), 1)
-
- def test_gkdi_create_root_key_priv_1024(self):
-
- server_config_dn = self.samdb.get_config_basedn()
- server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
- "CN=Server Configuration," +
- "CN=Group Key Distribution Service," +
- "CN=Services")
- res = self.samdb.search(base=server_config_dn,
- scope=ldb.SCOPE_BASE,
- attrs=["msKds-PrivateKeyLength"])
-
- self.assertEqual(len(res), 1)
-
- msg = res[0]
- if "msKds-PrivateKeyLength" in msg:
- keylen = msg["msKds-PrivateKeyLength"]
- # Ensure test still tests something in the future, if the default changes
- self.assertNotEqual(keylen, 1024)
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-PrivateKeyLength": [str(keylen)]},
- ldb.FLAG_MOD_REPLACE))
- else:
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-PrivateKeyLength": []},
- ldb.FLAG_MOD_DELETE))
-
- self.samdb.modify(ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-PrivateKeyLength": ["1024"]},
- ldb.FLAG_MOD_REPLACE))
-
- dn = self.samdb.new_gkdi_root_key()
-
- root_key_res = self.samdb.search(base=dn,
- scope=ldb.SCOPE_BASE)
- self.assertEqual(len(root_key_res), 1)
- root_key = root_key_res[0]
-
- self.assertEqual(int(root_key["msKds-PrivateKeyLength"][0]), 1024)
- self.assertEqual(str(root_key["msKds-KDFAlgorithmID"][0]), "SP800_108_CTR_HMAC")
- self.assertEqual(str(root_key["msKds-SecretAgreementAlgorithmID"][0]), "DH")
- self.assertEqual(int(root_key["msKds-Version"][0]), 1)
-
- def test_gkdi_create_root_key_bad_alg(self):
- server_config_dn = self.samdb.get_config_basedn()
- server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
- "CN=Server Configuration," +
- "CN=Group Key Distribution Service," +
- "CN=Services")
- res = self.samdb.search(base=server_config_dn,
- scope=ldb.SCOPE_BASE,
- attrs=["msKds-KDFAlgorithmID"])
-
- self.assertEqual(len(res), 1)
-
- msg = res[0]
- if "msKds-KDFAlgorithmID" in msg:
- alg = msg["msKds-KDFAlgorithmID"][0]
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-KDFAlgorithmID": [alg]},
- ldb.FLAG_MOD_REPLACE))
- else:
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-KDFAlgorithmID": []},
- ldb.FLAG_MOD_DELETE))
-
- self.samdb.modify(ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-KDFAlgorithmID": ["NO_AN_ALG"]},
- ldb.FLAG_MOD_REPLACE))
-
- try:
- self.samdb.new_gkdi_root_key()
- self.fail("Creating key with invalid algorithm should fail")
- except ldb.LdbError as e:
- (enum, estr) = e.args
- self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
-
- def test_gkdi_create_root_key_good_alg(self):
- server_config_dn = self.samdb.get_config_basedn()
- server_config_dn.add_child("CN=Group Key Distribution Service Server Configuration," +
- "CN=Server Configuration," +
- "CN=Group Key Distribution Service," +
- "CN=Services")
- res = self.samdb.search(base=server_config_dn,
- scope=ldb.SCOPE_BASE,
- attrs=["msKds-KDFAlgorithmID"])
-
- self.assertEqual(len(res), 1)
-
- msg = res[0]
- if "msKds-KDFAlgorithmID" in msg:
- alg = msg["msKds-KDFAlgorithmID"][0]
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-KDFAlgorithmID": [alg]},
- ldb.FLAG_MOD_REPLACE))
- else:
- self.addCleanup(self.samdb.modify,
- ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-KDFAlgorithmID": []},
- ldb.FLAG_MOD_DELETE))
-
- self.samdb.modify(ldb.Message.from_dict(self.samdb,
- {"dn": msg["dn"],
- "msKds-KDFAlgorithmID": ["SP800_108_CTR_HMAC"]},
- ldb.FLAG_MOD_REPLACE))
-
- dn = self.samdb.new_gkdi_root_key()
-
- root_key_res = self.samdb.search(base=dn,
- scope=ldb.SCOPE_BASE)
- self.assertEqual(len(root_key_res), 1)
- root_key = root_key_res[0]
-
- self.assertEqual(int(root_key["msKds-PublicKeyLength"][0]), 2048)
- self.assertEqual(str(root_key["msKds-KDFAlgorithmID"][0]), "SP800_108_CTR_HMAC")
- self.assertEqual(str(root_key["msKds-SecretAgreementAlgorithmID"][0]), "DH")
- self.assertEqual(int(root_key["msKds-Version"][0]), 1)
plansmbtorture4testsuite(t, "ad_dc_default:local", '//$SERVER/whavever')
# These tests want to run on a barely changed fresh provision, before
-# too much happens to this environment.
-planpythontestsuite("chgdcpass:local", "samba.tests.dsdb_quiet_provision_tests")
+# too much happens to this environment, it is read only and local
+# (direct to the DB) so we use proclimitdc as it is otherwise empty
+# bar a test for process limits.
+planpythontestsuite("proclimitdc:local", "samba.tests.dsdb_quiet_provision_tests")
+
+# We want this local test to run in an environment where not much is happening that could use root keys
+planpythontestsuite("chgdcpass:local", "samba.tests.dsdb_quiet_env_tests")
# Local tests
for t in smbtorture4_testsuites("dlz_bind9."):
planpythontestsuite("ad_dc_default", "samba.tests.samba_tool.domain_claim")
planpythontestsuite("ad_dc_default", "samba.tests.samba_tool.domain_auth_policy")
planpythontestsuite("ad_dc_default", "samba.tests.samba_tool.domain_auth_silo")
-planpythontestsuite("ad_dc_default", "samba.tests.samba_tool.domain_kds_root_key")
+
+# This test needs to be run in an environment well apart from most
+# other tests as it deletes root keys and we don't want this to happen
+# where a gMSA account might be live.
+planpythontestsuite("chgdcpass", "samba.tests.samba_tool.domain_kds_root_key")
+
planpythontestsuite("ad_dc_default", "samba.tests.samba_tool.domain_models")
planpythontestsuite("ad_dc_default", "samba.tests.samba_tool.service_account")
planpythontestsuite("schema_dc:local", "samba.tests.samba_tool.schema")