python/samba/tests/samba_tool/domain_kds_root_key.py

   1 # Unix SMB/CIFS implementation.
   2 #
   3 # Tests for samba-tool commands for Key Distribution Services
   4 #
   5 # Copyright © Catalyst.Net Ltd. 2024
   6 #
   7 # This program is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # This program is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19
  20 import json
  21 import os
  22 import re
  23 from datetime import datetime, timezone
  24
  25 from .base import SambaToolCmdTest
  26 from samba.dcerpc import misc
  27
  28 from samba.nt_time import (nt_now,
  29                            NT_TICKS_PER_SEC,
  30                            nt_time_from_string,
  31                            string_from_nt_time)
  32
  33 from ldb import SCOPE_SUBTREE, Dn
  34
  35 from samba.tests.gkdi import create_root_key
  36
  37
  38 HOST = "ldap://{DC_SERVER}".format(**os.environ)
  39 CREDS = "-U{DC_USERNAME}%{DC_PASSWORD}".format(**os.environ)
  40 SMBCONF = os.environ['SERVERCONFFILE']
  41
  42 # alice%Secret007
  43 NON_ADMIN_CREDS = "-U{DOMAIN_USER}%{DOMAIN_USER_PASSWORD}".format(**os.environ)
  44
  45 TIMESTAMP_RE = r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}\+00:00'
  46
  47 NOWISH = 'about now'
  48
  49
  50 class KdsRootKeyTestsBase(SambaToolCmdTest):
  51     @classmethod
  52     def setUpClass(cls):
  53         cls.samdb = cls.getSamDB("-H", HOST, CREDS)
  54         dn = cls.samdb.get_config_basedn()
  55         dn.add_child("CN=Master Root Keys,CN=Group Key Distribution Service,CN=Services")
  56         cls.root_key_base_dn = dn
  57
  58         # we'll add one for all tests to rely on -- but most will add
  59         # their own.
  60         super().setUpClass()
  61
  62     @classmethod
  63     def _create_root_key_timediff(cls, create_diff=0, use_diff=0):
  64         now = nt_now()
  65         nt_create = now + create_diff * NT_TICKS_PER_SEC
  66         nt_use = now + use_diff * NT_TICKS_PER_SEC
  67         guid, dn = create_root_key(cls.samdb,
  68                                    cls.root_key_base_dn,
  69                                    current_nt_time=nt_create,
  70                                    use_start_time=nt_use)
  71
  72         return guid, dn, nt_create, nt_use
  73
  74     def _create_root_key_timediff_cleanup(self, create_diff=0, use_diff=0):
  75         """create a root key that will disappear when the test ends."""
  76         guid, dn, nt_create, nt_use = self._create_root_key_timediff(
  77             create_diff,
  78             use_diff)
  79         self.addCleanup(self.samdb.delete, dn)
  80         return guid, dn, nt_create, nt_use
  81
  82     def _check_timestamp(self, isotimestamp, expected, range=10000):
  83         """Check that a timestamp string matches an nt-time.
  84
  85         By default we give a millisecond of leeway, because the ISO
  86         timestamp has less resolution than NT time (at most 6 decimal
  87         digits for seconds).
  88         """
  89
  90         t = nt_time_from_string(isotimestamp)
  91
  92         if expected is None:
  93             # we don't know what we want, but at least it's a time!
  94             return
  95
  96         if expected is NOWISH:
  97             expected = nt_now()
  98             range = 2.0 * NT_TICKS_PER_SEC
  99
 100         self.assertGreaterEqual(t, expected - range)
 101         self.assertLessEqual(t, expected + range)
 102
 103     def _test_list_output_snippet(self, output,
 104                                   guid=r'\b[0-9a-fA-F-]{36}\b',
 105                                   created=None,
 106                                   used_from=None,
 107                                   verbose=False):
 108         # name 1146a853-b604-75ac-5acc-4ef4f0530584
 109         #    created        2024-02-15T22:55:47.865576+00:00 (about 4 days ago)
 110         #    usable from    2024-02-15T22:55:47.865576+00:00 (about 4 days ago)
 111         self.assertRegex(output, f"(?m)^name {guid}$")
 112
 113         m = re.search(f' created +({TIMESTAMP_RE})', output)
 114         self.assertIsNotNone(m, "create timestamp not found")
 115         create_timestamp = m.group(1)
 116         self._check_timestamp(create_timestamp, created)
 117
 118         m = re.search(f' usable from +({TIMESTAMP_RE})', output)
 119         self.assertIsNotNone(m, "usable from timestamp not found")
 120         used_from_timestamp = m.group(1)
 121         self._check_timestamp(used_from_timestamp, used_from)
 122
 123         if verbose:
 124             dn = f"CN={guid},{self.root_key_base_dn}"
 125             self.assertRegex(output, f"(?m)^ +dn +{dn}$")
 126             self.assertRegex(output, r"(?m)^ +whenCreated +\d{14}.0Z$")
 127             self.assertRegex(output, r"(?m)^ +whenChanged +\d{14}.0Z$")
 128             self.assertRegex(output, r"(?m)^ +objectGUID +[0-9a-fA-F-]{36}$")
 129             self.assertRegex(output, r"(?m)^ +msKds-KDFAlgorithmID \w+$")
 130             self.assertRegex(output, r"(?m)^ +msKds-KDFParam \w+$")
 131             self.assertRegex(output, r"(?m)^ +msKds-SecretAgreementAlgorithmID \w+$")
 132             self.assertRegex(output, r"(?m)^ +msKds-PublicKeyLength \d+$")
 133             self.assertRegex(output, r"(?m)^ +msKds-PrivateKeyLength \d+$")
 134             self.assertRegex(output, r"(?m)^ +msKds-Version  1$")
 135             self.assertRegex(output, rf"(?m)^ +msKds-DomainID [\w=, ]+{self.samdb.domain_dn()}$",
 136                              re.MULTILINE)
 137             self.assertRegex(output, f"(?m)^ +cn +{guid}$")  # same guid as name
 138
 139     def _test_list_output_json_snippet(self, snippet,
 140                                        guid=r'\b[0-9a-fA-F-]{36}\b',
 141                                        created=None,
 142                                        used_from=None,
 143                                        verbose=False):
 144
 145         _guid = lambda x: re.fullmatch(str(guid), x)
 146         _hexstr = lambda x: re.fullmatch('[0-9a-fA-F]+', x)
 147         _str = lambda x: isinstance(x, str)
 148         _int = lambda x: isinstance(x, int)
 149
 150         # these next 2 will raise an assertion error on failure
 151         def _used_from(x):
 152             self._check_timestamp(x, used_from)
 153             return True
 154
 155         def _created(x):
 156             self._check_timestamp(x, used_from)
 157             return True
 158
 159         validators = {
 160             "cn": _guid,
 161             "dn": _str,
 162             "msKds-CreateTime": _created,
 163             "msKds-DomainID": _str,
 164             "msKds-KDFAlgorithmID": _str,
 165             "msKds-KDFParam": _hexstr,
 166             "msKds-PrivateKeyLength": _int,
 167             "msKds-PublicKeyLength": _int,
 168             "msKds-SecretAgreementAlgorithmID": _str,
 169             "msKds-UseStartTime": _used_from,
 170             "msKds-Version": _int,
 171             "name": _guid,
 172             "objectGUID": _str,
 173             "whenChanged": _str,
 174             "whenCreated": _str,
 175         }
 176         if verbose:
 177             keys = validators
 178         else:
 179             keys = ["name", "msKds-UseStartTime", "msKds-CreateTime", "dn"]
 180
 181         self.assertEqual(len(keys), len(snippet), f"keys: {keys}, json: {snippet}")
 182
 183         for k in keys:
 184             f = validators.get(k)
 185             v = snippet.get(k)
 186             self.assertTrue(f(v), f"{k} value {v} is wrong or malformed")
 187
 188     def _get_root_key_guids(self):
 189         """Get the current list of GUIDs."""
 190         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 191                                        "-H", HOST, CREDS)
 192         return [x['name'] for x in json.loads(out)]
 193
 194     def _delete_root_key(self, guid):
 195         dn = Dn(self.samdb, str(self.root_key_base_dn))
 196         dn.add_child(f"CN={guid}")
 197         self.samdb.delete(dn)
 198
 199 class KdsRootKeyTests(KdsRootKeyTestsBase):
 200
 201     @classmethod
 202     def setUpClass(cls):
 203         super().setUpClass()
 204         # we'll add one for all tests to rely on.
 205         cls.common_guid, cls.common_dn, cls.common_time, _ = cls._create_root_key_timediff()
 206         cls.addClassCleanup(cls.samdb.delete, cls.common_dn)
 207
 208     def test_list(self):
 209
 210         """Do we list root keys with the expected info?"""
 211         # For this test we also need to create some root keys.
 212         guid, dn, _created, _used = self._create_root_key_timediff_cleanup()
 213
 214         result, out, err = self.runcmd("domain", "kds", "root-key", "list",
 215                                        "-H", HOST, CREDS)
 216         self.assertCmdSuccess(result, out, err)
 217         self.assertEqual(err, "", "not expecting error messages")
 218
 219         # the output looks something like
 220         #
 221         #------------------------------------------------------------------------
 222         # 2 root keys found.
 223         #
 224         # name d58e85d7-ffc4-d118-9c43-46fac38dea05
 225         #   created        2024-02-27T09:09:21.065486+00:00 (about 1 seconds ago)
 226         #   usable from    2024-02-27T09:09:21.065486+00:00 (about 1 seconds ago)
 227         #
 228         # name 8f3e6557-3ec9-cb84-2ecd-9e258df68e79
 229         #   created        2024-02-27T09:09:10.853494+00:00 (about 12 seconds ago)
 230         #   usable from    2024-02-27T09:09:10.853494+00:00 (about 12 seconds ago)
 231         #-------------------------------------------------------------------------
 232         #
 233         # we want to check the various bits.
 234
 235         parts = out.rstrip().split("\n\n")
 236
 237         self.assertEqual(parts[0], f"{len(parts) - 1} root keys found.")
 238
 239         self._test_list_output_snippet(parts[1], guid,
 240                                        created=NOWISH,
 241                                        used_from=NOWISH)
 242
 243         guid2, dn2, _created2, _used2 = self._create_root_key_timediff_cleanup()
 244
 245         result, out, err = self.runcmd("domain", "kds", "root-key", "list",
 246                                        "-H", HOST, CREDS)
 247         self.assertCmdSuccess(result, out, err)
 248         self.assertEqual(err, "", "not expecting error messages")
 249
 250         parts2 = out.rstrip().split("\n\n")
 251         self.assertEqual(parts2[0], f"{len(parts)} root keys found.")
 252         self.assertEqual(len(parts2), len(parts) + 1)
 253
 254         # we want to check that both of them are still there, in the
 255         # right order, which is newest first.
 256         self._test_list_output_snippet(parts2[1], guid2,
 257                                        created=_created2,
 258                                        used_from=_used2)
 259         self._test_list_output_snippet(parts2[2], guid,
 260                                        created=_created,
 261                                        used_from=_used)
 262
 263     def test_list_verbose(self):
 264         """Do we list root keys with the expected info?"""
 265         guid, dn, _created, _used = self._create_root_key_timediff_cleanup()
 266
 267         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "-v",
 268                                        "-H", HOST, CREDS)
 269
 270         self.assertCmdSuccess(result, out, err)
 271         self.assertEqual(err, "", "not expecting error messages")
 272
 273         self._test_list_output_snippet(out, guid, verbose=True)
 274
 275         guid2, dn2, _created2, _used2 = self._create_root_key_timediff_cleanup()
 276
 277         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "-v",
 278                                        "-H", HOST, CREDS)
 279         self.assertCmdSuccess(result, out, err)
 280         self.assertEqual(err, "", "not expecting error messages")
 281
 282         self._test_list_output_snippet(out, guid2, verbose=True)
 283
 284         # in case there are other root keys, we will test each piece
 285         # using the default '[0-9a-fA-F-]{36}' guid-ish assertion.
 286
 287         pieces = out.rstrip().split('\n\n')
 288         self.assertRegex(pieces[0], f'{len(pieces) - 1} root keys found.')
 289
 290         for piece in pieces[1:]:
 291             self._test_list_output_snippet(piece, verbose=True)
 292
 293     def test_list_json(self):
 294         """The JSON should be a list of dicts, containing the right things"""
 295         guid, dn, _created, _used = self._create_root_key_timediff_cleanup()
 296
 297         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "-v", "--json",
 298                                        "-H", HOST, CREDS)
 299         self.assertCmdSuccess(result, out, err)
 300         self.assertEqual(err, "", "not expecting error messages")
 301         data = json.loads(out)
 302         for snippet in data:
 303             self._test_list_output_json_snippet(snippet, verbose=True)
 304
 305         # non-verbose
 306         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 307                                        "-H", HOST, CREDS)
 308         self.assertCmdSuccess(result, out, err)
 309         self.assertEqual(err, "", "not expecting error messages")
 310         data = json.loads(out)
 311         for snippet in data:
 312             self._test_list_output_json_snippet(snippet)
 313
 314     def test_view_key_that_exists(self):
 315         guid, dn, _created, _used = self._create_root_key_timediff_cleanup()
 316         cmd = ["domain", "kds", "root-key", "view",
 317                "-H", HOST, CREDS,
 318                "--name", str(guid)]
 319
 320         result, out, err = self.runcmd(*cmd)
 321         self.assertCmdSuccess(result, out, err)
 322         self.assertEqual(err, "", "not expecting error messages")
 323
 324         self._test_list_output_snippet(out, guid,
 325                                        created=NOWISH,
 326                                        used_from=NOWISH,
 327                                        verbose=True)
 328
 329     def test_view_key_that_exists_json(self):
 330         guid, dn, _created, _used = self._create_root_key_timediff_cleanup()
 331
 332         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 333                                        "--json",
 334                                        "--name", str(guid),
 335                                        "-H", HOST, CREDS)
 336         self.assertCmdSuccess(result, out, err)
 337         self.assertEqual(err, "", "not expecting error messages")
 338         data = json.loads(out)
 339         self._test_list_output_json_snippet(data, guid,
 340                                             created=_created,
 341                                             used_from=_used,
 342                                             verbose=True)
 343
 344
 345     def test_view_key_latest_json(self):
 346         guid, dn, _created, _used = self._create_root_key_timediff_cleanup()
 347
 348         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 349                                        "--json",
 350                                        "--latest",
 351                                        "-H", HOST, CREDS)
 352         self.assertCmdSuccess(result, out, err)
 353         self.assertEqual(err, "", "not expecting error messages")
 354         data = json.loads(out)
 355         self._test_list_output_json_snippet(data, guid,
 356                                             created=_created,
 357                                             used_from=_used,
 358                                             verbose=True)
 359
 360         # if we make a new now-ish key, it will be shown with
 361         # --latest, forgetting the old one.
 362         guid2, dn2, _created2, _used2 = self._create_root_key_timediff_cleanup()
 363
 364         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 365                                        "--json",
 366                                        "--latest",
 367                                        "-H", HOST, CREDS)
 368         self.assertCmdSuccess(result, out, err)
 369         self.assertEqual(err, "", "not expecting error messages")
 370         data = json.loads(out)
 371         self._test_list_output_json_snippet(data, guid2,
 372                                             created=_created2,
 373                                             used_from=_used2,
 374                                             verbose=True)
 375
 376         # if we make a new backdated key, it will not be shown as
 377         # latest, even though it was the most recently created.
 378
 379         self._create_root_key_timediff_cleanup(use_diff=-600)
 380
 381         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 382                                        "--json",
 383                                        "--latest",
 384                                        "-H", HOST, CREDS)
 385         self.assertCmdSuccess(result, out, err)
 386         self.assertEqual(err, "", "not expecting error messages")
 387         data = json.loads(out)
 388         self._test_list_output_json_snippet(data, guid2,
 389                                             created=_created2,
 390                                             used_from=_used2,
 391                                             verbose=True)
 392
 393         # if we make a future-dated key, it will be shown as
 394         # latest, even though it doesn't work yet.
 395
 396         guid3, dn3, _created3, _used3 = self._create_root_key_timediff_cleanup()
 397
 398         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 399                                        "--json",
 400                                        "--latest",
 401                                        "-H", HOST, CREDS)
 402         self.assertCmdSuccess(result, out, err)
 403         self.assertEqual(err, "", "not expecting error messages")
 404         data = json.loads(out)
 405         self._test_list_output_json_snippet(data, guid3,
 406                                             created=_created3,
 407                                             used_from=_used3,
 408                                             verbose=True)
 409
 410     def test_view_non_existent(self):
 411         """Viewing a non-existent GUID should fail, regardless of what exists."""
 412         guid = misc.GUID(b'a' * 16)
 413
 414         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 415                                        "-H", HOST, CREDS,
 416                                        "--name", str(guid))
 417         self.assertCmdFail(result)
 418
 419         self.assertIn("ERROR: no such root key: 61616161-6161-6161-6161-616161616161",
 420                       err)
 421
 422     def test_view_non_existent_json(self):
 423         guid = misc.GUID(b'a' * 16)
 424
 425         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 426                                        "-H", HOST, CREDS,
 427                                        "--name", str(guid),
 428                                        "--json")
 429         self.assertCmdFail(result)
 430         data = json.loads(out)
 431         self.assertEqual(
 432             data,
 433             {
 434                 "message": f"no such root key: {guid}",
 435                 "status": "error"
 436             })
 437
 438     def test_delete_non_existent(self):
 439         """Deletion of non-existent guid should fail"""
 440         guid = 'eeeeeeee-1111-eeee-1111-000000000000'
 441         result, out, err = self.runcmd("domain", "kds", "root-key", "delete",
 442                                        "-H", HOST, CREDS,
 443                                        "--name", guid)
 444         self.assertCmdFail(result)
 445         self.assertIn(f"ERROR: no such root key: {guid}", err)
 446
 447     def test_delete_non_existent_json(self):
 448         """Deletion of non-existent guids should fail"""
 449         for guid in ('eeeeeeee-1111-eeee-1111-000000000000',
 450                      'foo',
 451                      ''):
 452             result, out, err = self.runcmd("domain", "kds", "root-key", "delete",
 453                                            "-H", HOST, CREDS,
 454                                            "--name", guid,
 455                                            "--json")
 456             self.assertCmdFail(result)
 457             data = json.loads(out)
 458             self.assertEqual(
 459                 data,
 460                 {
 461                     "message": f"no such root key: {guid}",
 462                     "status": "error"
 463                 })
 464
 465     def test_create(self):
 466         """does create work?"""
 467         pre_create = self._get_root_key_guids()
 468
 469         result, out, err = self.runcmd("domain", "kds", "root-key", "create",
 470                                        "-H", HOST, CREDS)
 471         self.assertCmdSuccess(result, out, err)
 472         self.assertEqual(err, "", "not expecting error messages")
 473
 474         post_create = self._get_root_key_guids()
 475
 476         new_guids = list(set(post_create) - set(pre_create))
 477         gone_guids = set(pre_create) - set(post_create)
 478         self.assertEqual(len(gone_guids), 0)
 479         self.assertEqual(len(new_guids), 1)
 480         self.assertRegex(out,
 481                          f"created root key {new_guids[0]}, usable from {TIMESTAMP_RE}")
 482         self._delete_root_key(new_guids[0])
 483
 484     def test_create_json(self):
 485         """does create work?"""
 486         pre_create = self._get_root_key_guids()
 487
 488         result, out, err = self.runcmd("domain", "kds", "root-key", "create",
 489                                        "-H", HOST, CREDS, "--json")
 490         self.assertCmdSuccess(result, out, err)
 491         self.assertEqual(err, "", "not expecting error messages")
 492
 493         post_create = self._get_root_key_guids()
 494
 495         new_guids = list(set(post_create) - set(pre_create))
 496         gone_guids = set(pre_create) - set(post_create)
 497         self.assertEqual(len(gone_guids), 0)
 498         self.assertEqual(len(new_guids), 1)
 499         data = json.loads(out)
 500         self.assertEqual(data['dn'], f"CN={new_guids[0]},{self.root_key_base_dn}")
 501         self.assertEqual(data['status'], 'OK')
 502         self.assertRegex(data['message'],
 503                          f"created root key {new_guids[0]}, usable from {TIMESTAMP_RE}")
 504         self._delete_root_key(new_guids[0])
 505
 506     def test_create_json_non_admin(self):
 507         """can you create a root-key without being admin?"""
 508         pre_create = self._get_root_key_guids()
 509
 510         result, out, err = self.runcmd("domain", "kds", "root-key", "create",
 511                                        "-H", HOST, NON_ADMIN_CREDS, "--json")
 512         self.assertCmdFail(result)
 513
 514         post_create = self._get_root_key_guids()
 515
 516         self.assertEqual(set(pre_create), set(post_create))
 517         data = json.loads(out)
 518         self.assertEqual(data['status'], 'error')
 519         self.assertEqual(data['message'], 'User has insufficient access rights')
 520         self.assertEqual(err, "", "not expecting stderr messages")
 521
 522     def test_create_json_1997(self):
 523         """does create work?"""
 524         pre_create = self._get_root_key_guids()
 525
 526         result, out, err = self.runcmd("domain", "kds", "root-key", "create",
 527                                        "-H", HOST, CREDS, "--json",
 528                                        "--use-start-time",
 529                                        "1997-11-11T23:18:00.259810+00:00")
 530         self.assertCmdSuccess(result, out, err)
 531         self.assertEqual(err, "", "not expecting error messages")
 532
 533         post_create = self._get_root_key_guids()
 534
 535         new_guids = list(set(post_create) - set(pre_create))
 536         gone_guids = set(pre_create) - set(post_create)
 537         self.assertEqual(len(gone_guids), 0)
 538         self.assertEqual(len(new_guids), 1)
 539         data = json.loads(out)
 540         self.assertEqual(data['dn'], f"CN={new_guids[0]},{self.root_key_base_dn}")
 541         self.assertEqual(data['status'], 'OK')
 542         self.assertRegex(data['message'],
 543                          f"created root key {new_guids[0]}, usable from 1997-11-1")
 544         self._delete_root_key(new_guids[0])
 545
 546     def test_create_json_2197(self):
 547         """does create work?"""
 548         pre_create = self._get_root_key_guids()
 549
 550         result, out, err = self.runcmd("domain", "kds", "root-key", "create",
 551                                        "-H", HOST, CREDS, "--json",
 552                                        "--use-start-time",
 553                                        "2197-11-11T23:18:00")
 554         self.assertCmdSuccess(result, out, err)
 555         self.assertEqual(err, "", "not expecting error messages")
 556
 557         post_create = self._get_root_key_guids()
 558
 559         new_guids = list(set(post_create) - set(pre_create))
 560         gone_guids = set(pre_create) - set(post_create)
 561         self.assertEqual(len(gone_guids), 0)
 562         self.assertEqual(len(new_guids), 1)
 563         data = json.loads(out)
 564         self.assertEqual(data['dn'], f"CN={new_guids[0]},{self.root_key_base_dn}")
 565         self.assertEqual(data['status'], 'OK')
 566         self.assertRegex(data['message'],
 567                          f"created root key {new_guids[0]}, usable from 2197-11-1")
 568         self._delete_root_key(new_guids[0])
 569
 570     def test_create_future(self):
 571         """does create work, with a use-start-time 500 seconds in the
 572         future?"""
 573         pre_create = self._get_root_key_guids()
 574         now = nt_now()
 575         later = now + 500 * NT_TICKS_PER_SEC
 576         timestamp = string_from_nt_time(later)
 577
 578         result, out, err = self.runcmd("domain", "kds", "root-key", "create",
 579                                        "-H", HOST, CREDS, "--json",
 580                                        "--use-start-time", timestamp)
 581
 582         self.assertCmdSuccess(result, out, err)
 583         self.assertEqual(err, "", "not expecting error messages")
 584
 585         post_create = self._get_root_key_guids()
 586
 587         new_guids = list(set(post_create) - set(pre_create))
 588         gone_guids = set(pre_create) - set(post_create)
 589         self.assertEqual(len(gone_guids), 0)
 590         self.assertEqual(len(new_guids), 1)
 591         data = json.loads(out)
 592         self.assertEqual(data['dn'], f"CN={new_guids[0]},{self.root_key_base_dn}")
 593         self.assertEqual(data['status'], 'OK')
 594         self.assertRegex(data['message'],
 595                          f"created root key {new_guids[0]}, usable from {timestamp[:-10]}")
 596         self._delete_root_key(new_guids[0])
 597
 598     def test_delete(self):
 599         """does delete work?"""
 600         # make one to delete, and get the list as JSON
 601         _guid, dn, _created, _used = self._create_root_key_timediff()
 602         guid = str(_guid)
 603
 604         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 605                                        "-H", HOST, CREDS)
 606         pre_delete = json.loads(out)
 607
 608         result, out, err = self.runcmd("domain", "kds", "root-key", "delete",
 609                                        "-H", HOST, CREDS,
 610                                        "--name", guid)
 611         self.assertCmdSuccess(result, out, err)
 612         self.assertEqual(err, "", "not expecting error messages")
 613         self.assertEqual(out, f"deleted root key {guid}\n")
 614
 615         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 616                                        "-H", HOST, CREDS)
 617         post_delete = json.loads(out)
 618
 619         self.assertEqual(len(pre_delete), len(post_delete) + 1)
 620
 621         post_names = [x['name'] for x in post_delete]
 622         pre_names = [x['name'] for x in pre_delete]
 623
 624         self.assertIn(guid, pre_names)
 625         self.assertNotIn(guid, post_names)
 626
 627     def test_delete_json(self):
 628         """does delete --json work?"""
 629         _guid, dn, _created, _used = self._create_root_key_timediff()
 630         guid = str(_guid)
 631
 632         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 633                                        "-H", HOST, CREDS)
 634         pre_delete = json.loads(out)
 635
 636         result, out, err = self.runcmd("domain", "kds", "root-key", "delete",
 637                                        "-H", HOST, CREDS, "--json",
 638                                        "--name", guid)
 639
 640         self.assertCmdSuccess(result, out, err)
 641         self.assertEqual(err, "", "not expecting error messages")
 642         data = json.loads(out)
 643         self.assertEqual(
 644             data,
 645             {
 646                 "message": f"deleted root key {guid}",
 647                 "status": "error"
 648             })
 649
 650         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 651                                        "-H", HOST, CREDS)
 652         post_delete = json.loads(out)
 653
 654         self.assertEqual(len(pre_delete), len(post_delete) + 1)
 655
 656         post_names = [x['name'] for x in post_delete]
 657         pre_names = [x['name'] for x in pre_delete]
 658
 659         self.assertIn(guid, pre_names)
 660         self.assertNotIn(guid, post_names)
 661
 662     def test_delete_non_admin(self):
 663         """does delete as non-admin fail?"""
 664         # make one to delete, and get the list as JSON
 665         _guid, dn, _created, _used = self._create_root_key_timediff()
 666         guid = str(_guid)
 667
 668         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 669                                        "-H", HOST, CREDS)
 670         pre_delete = json.loads(out)
 671
 672         result, out, err = self.runcmd("domain", "kds", "root-key", "delete",
 673                                        "-H", HOST, NON_ADMIN_CREDS,
 674                                        "--name", guid)
 675         self.assertCmdFail(result)
 676         self.assertIn(f"ERROR: no such root key: {guid}", err)
 677
 678         # a bad guid should be just like a good guid
 679         guid2 = 'eeeeeeee-1111-eeee-1111-000000000000'
 680         result, out2, err2 = self.runcmd("domain", "kds", "root-key", "delete",
 681                                          "-H", HOST, NON_ADMIN_CREDS,
 682                                          "--name", guid2)
 683         self.assertCmdFail(result)
 684         self.assertIn(f"ERROR: no such root key: {guid2}", err2)
 685
 686         result, out, err = self.runcmd("domain", "kds", "root-key", "list", "--json",
 687                                        "-H", HOST, CREDS)
 688         post_delete = json.loads(out)
 689
 690         self.assertEqual(len(pre_delete), len(post_delete))
 691
 692         post_names = [x['name'] for x in post_delete]
 693         pre_names = [x['name'] for x in pre_delete]
 694
 695         self.assertIn(guid, pre_names)
 696         self.assertIn(guid, post_names)
 697
 698     def test_list_non_admin(self):
 699         """There are root keys, but non-admins can't see them"""
 700         result, out, err = self.runcmd("domain", "kds", "root-key", "list",
 701                                        "-H", HOST, NON_ADMIN_CREDS)
 702         self.assertCmdSuccess(result, out, err)
 703         self.assertEqual(err, "", "not expecting error messages")
 704         self.assertEqual(out, "no root keys found.\n")
 705
 706     def test_list_json_non_admin(self):
 707         """Insufficient rights should look like an empty list."""
 708         # this is a copy of the KdsNoRootKeyTests test below --
 709         # non-admin should look exactly like an empty list.
 710         for extra in ([], ["-v"]):
 711             result, out, err = self.runcmd("domain", "kds", "root-key", "list",
 712                                            "-H", HOST, NON_ADMIN_CREDS, "--json", *extra)
 713             self.assertCmdSuccess(result, out, err)
 714             self.assertEqual(err, "", "not expecting error messages")
 715             data = json.loads(out)
 716             self.assertEqual(data, [])
 717
 718     def test_view_key_non_admin(self):
 719         """should not appear to non-admin"""
 720         guid, dn, _created, _used = self._create_root_key_timediff_cleanup()
 721
 722         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 723                                        "--json",
 724                                        "--name", str(guid),
 725                                        "-H", HOST, NON_ADMIN_CREDS)
 726         self.assertCmdFail(result)
 727         self.assertEqual(err, "", "not expecting error messages")
 728         data = json.loads(out)
 729         data = json.loads(out)
 730         self.assertEqual(
 731             data,
 732             {
 733                 "message": f"no such root key: {guid}",
 734                 "status": "error"
 735             })
 736
 737
 738 class KdsNoRootKeyTests(KdsRootKeyTestsBase):
 739     """Here we test the case were there are no root keys, which we need to
 740     ensure by deleting any that are there.
 741     """
 742
 743     @classmethod
 744     def setUpClass(cls):
 745         super().setUpClass()
 746         # We delete all the root keys, and add one back at the end,
 747         # in case other tests want there to be one.
 748         res = cls.samdb.search(cls.root_key_base_dn,
 749                                scope=SCOPE_SUBTREE,
 750                                expression="(objectClass = msKds-ProvRootKey)")
 751
 752         for msg in res:
 753             cls.samdb.delete(msg.dn)
 754
 755         cls.addClassCleanup(cls.samdb.new_gkdi_root_key)
 756
 757     def test_list_empty(self):
 758         """Check the message when there are no root keys"""
 759         result, out, err = self.runcmd("domain", "kds", "root-key", "list",
 760                                        "-H", HOST, CREDS)
 761         self.assertCmdSuccess(result, out, err)
 762         self.assertEqual(err, "", "not expecting error messages")
 763         self.assertEqual(out, "no root keys found.\n")
 764
 765     def test_list_empty_json(self):
 766         """The JSON should be an empty list when there are no root keys"""
 767         # verbose flag makes no difference here.
 768         for extra in ([], ["-v"]):
 769             result, out, err = self.runcmd("domain", "kds", "root-key", "list",
 770                                            "-H", HOST, CREDS, "--json", *extra)
 771             self.assertCmdSuccess(result, out, err)
 772             self.assertEqual(err, "", "not expecting error messages")
 773             data = json.loads(out)
 774             self.assertEqual(data, [])
 775
 776     def test_list_empty_json_non_admin(self):
 777         """Insufficient rights should look like an empty list."""
 778         # verbose flag makes no difference here.
 779         for extra in ([], ["-v"]):
 780             result, out, err = self.runcmd("domain", "kds", "root-key", "list",
 781                                            "-H", HOST, NON_ADMIN_CREDS, "--json", *extra)
 782             self.assertCmdSuccess(result, out, err)
 783             self.assertEqual(err, "", "not expecting error messages")
 784             data = json.loads(out)
 785             self.assertEqual(data, [])
 786
 787     def test_view_latest_non_existent(self):
 788         """With no root keys, --latest should return an error"""
 789
 790         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 791                                        "-H", HOST, CREDS,
 792                                        "--latest")
 793
 794         self.assertEqual(err, "ERROR: no root keys found\n")
 795         self.assertCmdFail(result)
 796
 797     def test_view_latest_non_existent_json(self):
 798         """With no root keys, --latest should return an error"""
 799
 800         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 801                                        "-H", HOST, CREDS,
 802                                        "--json", "--latest")
 803         self.assertCmdFail(result)
 804         data = json.loads(out)
 805         self.assertEqual(
 806             data,
 807             {
 808                 "message": "no root keys found",
 809                 "status": "error"
 810             })
 811
 812     def test_view_non_existent(self):
 813         """Viewing a non-existent GUID should fail, regardless of what exists."""
 814         guid = misc.GUID(b'b' * 16)
 815
 816         result, out, err = self.runcmd("domain", "kds", "root-key", "view",
 817                                        "-H", HOST, CREDS,
 818                                        "--name", str(guid))
 819         self.assertCmdFail(result)
 820
 821         self.assertIn("ERROR: no such root key: 62626262-6262-6262-6262-626262626262",
 822                       err)