python/samba/tests/krb5/authn_policy_tests.py

   1 #!/usr/bin/env python3
   2 # Unix SMB/CIFS implementation.
   3 # Copyright (C) Stefan Metzmacher 2020
   4 # Copyright (C) Catalyst.Net Ltd 2023
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 3 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19
  20 import sys
  21 import os
  22
  23 sys.path.insert(0, 'bin/python')
  24 os.environ['PYTHONUNBUFFERED'] = '1'
  25
  26 import random
  27
  28 import ldb
  29
  30 from samba import dsdb, ntstatus
  31 from samba.dcerpc import netlogon, security
  32 from samba.ndr import ndr_pack
  33
  34 import samba.tests.krb5.kcrypto as kcrypto
  35 from samba.tests.krb5.kdc_base_test import GroupType
  36 from samba.tests.krb5.kdc_tgs_tests import KdcTgsBaseTests
  37 from samba.tests.krb5.rfc4120_constants import (
  38     FX_FAST_ARMOR_AP_REQUEST,
  39     KDC_ERR_BADOPTION,
  40     KDC_ERR_GENERIC,
  41     KDC_ERR_NEVER_VALID,
  42     KDC_ERR_POLICY,
  43     NT_PRINCIPAL,
  44     NT_SRV_INST,
  45 )
  46 import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
  47
  48 global_asn1_print = False
  49 global_hexdump = False
  50
  51 HRES_SEC_E_INVALID_TOKEN = 0x80090308
  52 HRES_SEC_E_LOGON_DENIED = 0x8009030C
  53
  54
  55 class AuthnPolicyTests(KdcTgsBaseTests):
  56     @classmethod
  57     def setUpClass(cls):
  58         super().setUpClass()
  59
  60         cls._max_ticket_life = None
  61         cls._max_renew_life = None
  62
  63     def setUp(self):
  64         super().setUp()
  65         self.do_asn1_print = global_asn1_print
  66         self.do_hexdump = global_hexdump
  67
  68     def get_max_ticket_life(self):
  69         if self._max_ticket_life is None:
  70             self._fetch_default_lifetimes()
  71
  72         return self._max_ticket_life
  73
  74     def get_max_renew_life(self):
  75         if self._max_renew_life is None:
  76             self._fetch_default_lifetimes()
  77
  78         return self._max_renew_life
  79
  80     def _fetch_default_lifetimes(self):
  81         samdb = self.get_samdb()
  82
  83         domain_policy_dn = samdb.get_default_basedn()
  84         domain_policy_dn.add_child('CN=Default Domain Policy,CN=System')
  85
  86         res = samdb.search(domain_policy_dn,
  87                            scope=ldb.SCOPE_BASE,
  88                            attrs=['maxTicketAge', 'maxRenewAge'])
  89         self.assertEqual(1, len(res))
  90
  91         max_ticket_age = res[0].get('maxTicketAge', idx=0)
  92         max_renew_age = res[0].get('maxRenewAge', idx=0)
  93
  94         if max_ticket_age is not None:
  95             max_ticket_age = int(max_ticket_age.decode('utf-8'))
  96         else:
  97             max_ticket_age = 10
  98
  99         if max_renew_age is not None:
 100             max_renew_age = int(max_renew_age.decode('utf-8'))
 101         else:
 102             max_renew_age = 7
 103
 104         type(self)._max_ticket_life = max_ticket_age * 60 * 60
 105         type(self)._max_renew_life = max_renew_age * 24 * 60 * 60
 106
 107     # Get account credentials for testing.
 108     def _get_creds(self,
 109                    account_type=KdcTgsBaseTests.AccountType.USER,
 110                    member_of=None,
 111                    protected=False,
 112                    assigned_policy=None,
 113                    assigned_silo=None,
 114                    ntlm=False,
 115                    spn=None,
 116                    allowed_rodc=None,
 117                    cached=True):
 118         opts = {
 119             'kerberos_enabled': not ntlm,
 120             'spn': spn,
 121         }
 122
 123         members = ()
 124         if protected:
 125             samdb = self.get_samdb()
 126             protected_users_group = (f'<SID={samdb.get_domain_sid()}-'
 127                                      f'{security.DOMAIN_RID_PROTECTED_USERS}>')
 128             members += (protected_users_group,)
 129         if member_of is not None:
 130             members += (member_of,)
 131         if assigned_policy is not None:
 132             opts['assigned_policy'] = str(assigned_policy)
 133             cached = False   # Policies are rarely reused between accounts.
 134         if assigned_silo is not None:
 135             opts['assigned_silo'] = str(assigned_silo)
 136             cached = False   # Silos are rarely reused between accounts.
 137         if allowed_rodc:
 138             opts['allowed_replication_mock'] = True
 139             opts['revealed_to_mock_rodc'] = True
 140
 141         if members:
 142             opts['member_of'] = members
 143
 144         return self.get_cached_creds(account_type=account_type,
 145                                      opts=opts,
 146                                      use_cache=cached)
 147
 148     def test_authn_policy_tgt_lifetime_user(self):
 149         # Create an authentication policy with certain TGT lifetimes set.
 150         user_life = 111
 151         computer_life = 222
 152         service_life = 333
 153         policy_id = self.get_new_username()
 154         policy = self.create_authn_policy(policy_id,
 155                                           enforced=True,
 156                                           user_tgt_lifetime=user_life,
 157                                           computer_tgt_lifetime=computer_life,
 158                                           service_tgt_lifetime=service_life)
 159
 160         # Create a user account with the assigned policy.
 161         client_creds = self._get_creds(account_type=self.AccountType.USER,
 162                                        assigned_policy=policy)
 163
 164         # Request a Kerberos ticket with a lifetime of two hours, and assert
 165         # that the actual lifetime matches the user lifetime set in the policy.
 166         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 167         tgt = self._get_tgt(client_creds, till=till)
 168         self.check_ticket_times(tgt, expected_life=user_life,
 169                                 expected_renew_life=user_life)
 170
 171     def test_authn_policy_tgt_lifetime_computer(self):
 172         user_life = 111
 173         computer_life = 222
 174         service_life = 333
 175         policy_id = self.get_new_username()
 176         policy = self.create_authn_policy(policy_id,
 177                                           enforced=True,
 178                                           user_tgt_lifetime=user_life,
 179                                           computer_tgt_lifetime=computer_life,
 180                                           service_tgt_lifetime=service_life)
 181
 182         # Create a computer account with the assigned policy.
 183         client_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
 184                                        assigned_policy=policy)
 185
 186         # Request a Kerberos ticket with a lifetime of two hours, and assert
 187         # that the actual lifetime matches the computer lifetime set in the
 188         # policy.
 189         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 190         tgt = self._get_tgt(client_creds, till=till)
 191         self.check_ticket_times(tgt, expected_life=computer_life,
 192                                 expected_renew_life=computer_life)
 193
 194     def test_authn_policy_tgt_lifetime_service(self):
 195         user_life = 111
 196         computer_life = 222
 197         service_life = 333
 198         policy_id = self.get_new_username()
 199         policy = self.create_authn_policy(policy_id,
 200                                           enforced=True,
 201                                           user_tgt_lifetime=user_life,
 202                                           computer_tgt_lifetime=computer_life,
 203                                           service_tgt_lifetime=service_life)
 204
 205         # Create a managed service account with the assigned policy.
 206         client_creds = self._get_creds(
 207             account_type=self.AccountType.MANAGED_SERVICE,
 208             assigned_policy=policy)
 209
 210         # Request a Kerberos ticket with a lifetime of two hours, and assert
 211         # that the actual lifetime matches the service lifetime set in the
 212         # policy.
 213         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 214         tgt = self._get_tgt(client_creds, till=till)
 215         self.check_ticket_times(tgt, expected_life=service_life,
 216                                 expected_renew_life=service_life)
 217
 218     def test_authn_silo_tgt_lifetime_user(self):
 219         # Create an authentication policy with certain TGT lifetimes set.
 220         user_life = 111
 221         computer_life = 222
 222         service_life = 333
 223         policy_id = self.get_new_username()
 224         policy = self.create_authn_policy(policy_id,
 225                                           enforced=True,
 226                                           user_tgt_lifetime=user_life,
 227                                           computer_tgt_lifetime=computer_life,
 228                                           service_tgt_lifetime=service_life)
 229
 230         # Create a second policy with different lifetimes, so we can verify the
 231         # correct policy is enforced.
 232         wrong_policy_id = self.get_new_username()
 233         wrong_policy = self.create_authn_policy(wrong_policy_id,
 234                                                 enforced=True,
 235                                                 user_tgt_lifetime=444,
 236                                                 computer_tgt_lifetime=555,
 237                                                 service_tgt_lifetime=666)
 238
 239         # Create an authentication silo with our existing policies.
 240         silo_id = self.get_new_username()
 241         silo = self.create_authn_silo(silo_id,
 242                                       user_policy=str(policy),
 243                                       computer_policy=str(wrong_policy),
 244                                       service_policy=str(wrong_policy),
 245                                       enforced=True)
 246
 247         # Create a user account assigned to the silo.
 248         client_creds = self._get_creds(account_type=self.AccountType.USER,
 249                                        assigned_silo=silo)
 250         client_dn_str = str(client_creds.get_dn())
 251
 252         # Add the user to the silo as a member.
 253         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 254                           expect_attr=False)
 255
 256         # Request a Kerberos ticket with a lifetime of two hours, and assert
 257         # that the actual lifetime matches the user lifetime set in the
 258         # appropriate policy.
 259         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 260         tgt = self._get_tgt(client_creds, till=till)
 261         self.check_ticket_times(tgt, expected_life=user_life,
 262                                 expected_renew_life=user_life)
 263
 264     def test_authn_silo_tgt_lifetime_computer(self):
 265         user_life = 111
 266         computer_life = 222
 267         service_life = 333
 268         policy_id = self.get_new_username()
 269         policy = self.create_authn_policy(policy_id,
 270                                           enforced=True,
 271                                           user_tgt_lifetime=user_life,
 272                                           computer_tgt_lifetime=computer_life,
 273                                           service_tgt_lifetime=service_life)
 274
 275         wrong_policy_id = self.get_new_username()
 276         wrong_policy = self.create_authn_policy(wrong_policy_id,
 277                                                 enforced=True,
 278                                                 user_tgt_lifetime=444,
 279                                                 computer_tgt_lifetime=555,
 280                                                 service_tgt_lifetime=666)
 281
 282         # Create an authentication silo with our existing policies.
 283         silo_id = self.get_new_username()
 284         silo = self.create_authn_silo(silo_id,
 285                                       user_policy=str(wrong_policy),
 286                                       computer_policy=str(policy),
 287                                       service_policy=str(wrong_policy),
 288                                       enforced=True)
 289
 290         # Create a computer account assigned to the silo.
 291         client_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
 292                                        assigned_silo=silo)
 293         client_dn_str = str(client_creds.get_dn())
 294
 295         # Add the computer to the silo as a member.
 296         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 297                           expect_attr=False)
 298
 299         # Request a Kerberos ticket with a lifetime of two hours, and assert
 300         # that the actual lifetime matches the computer lifetime set in the
 301         # appropriate policy.
 302         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 303         tgt = self._get_tgt(client_creds, till=till)
 304         self.check_ticket_times(tgt, expected_life=computer_life,
 305                                 expected_renew_life=computer_life)
 306
 307     def test_authn_silo_tgt_lifetime_service(self):
 308         user_life = 111
 309         computer_life = 222
 310         service_life = 333
 311         policy_id = self.get_new_username()
 312         policy = self.create_authn_policy(policy_id,
 313                                           enforced=True,
 314                                           user_tgt_lifetime=user_life,
 315                                           computer_tgt_lifetime=computer_life,
 316                                           service_tgt_lifetime=service_life)
 317
 318         wrong_policy_id = self.get_new_username()
 319         wrong_policy = self.create_authn_policy(wrong_policy_id,
 320                                                 enforced=True,
 321                                                 user_tgt_lifetime=444,
 322                                                 computer_tgt_lifetime=555,
 323                                                 service_tgt_lifetime=666)
 324
 325         # Create an authentication silo with our existing policies.
 326         silo_id = self.get_new_username()
 327         silo = self.create_authn_silo(silo_id,
 328                                       user_policy=str(wrong_policy),
 329                                       computer_policy=str(wrong_policy),
 330                                       service_policy=str(policy),
 331                                       enforced=True)
 332
 333         # Create a managed service account assigned to the silo.
 334         client_creds = self._get_creds(
 335             account_type=self.AccountType.MANAGED_SERVICE,
 336             assigned_silo=silo)
 337         client_dn_str = str(client_creds.get_dn())
 338
 339         # Add the managed service account to the silo as a member.
 340         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 341                           expect_attr=False)
 342
 343         # Request a Kerberos ticket with a lifetime of two hours, and assert
 344         # that the actual lifetime matches the service lifetime set in the
 345         # appropriate policy.
 346         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 347         tgt = self._get_tgt(client_creds, till=till)
 348         self.check_ticket_times(tgt, expected_life=service_life,
 349                                 expected_renew_life=service_life)
 350
 351     # Test that an authentication silo takes priority over a policy assigned
 352     # directly.
 353     def test_authn_silo_and_policy_tgt_lifetime_user(self):
 354         # Create an authentication policy with certain TGT lifetimes set.
 355         user_life = 111
 356         computer_life = 222
 357         service_life = 333
 358         policy_id = self.get_new_username()
 359         policy = self.create_authn_policy(policy_id,
 360                                           enforced=True,
 361                                           user_tgt_lifetime=user_life,
 362                                           computer_tgt_lifetime=computer_life,
 363                                           service_tgt_lifetime=service_life)
 364
 365         # Create a second policy with different lifetimes, so we can verify the
 366         # correct policy is enforced.
 367         wrong_policy_id = self.get_new_username()
 368         wrong_policy = self.create_authn_policy(wrong_policy_id,
 369                                                 enforced=True,
 370                                                 user_tgt_lifetime=444,
 371                                                 computer_tgt_lifetime=555,
 372                                                 service_tgt_lifetime=666)
 373
 374         # Create an authentication silo with our existing policies.
 375         silo_id = self.get_new_username()
 376         silo = self.create_authn_silo(silo_id,
 377                                       user_policy=str(policy),
 378                                       computer_policy=str(wrong_policy),
 379                                       service_policy=str(wrong_policy),
 380                                       enforced=True)
 381
 382         # Create a user account assigned to the silo, and also to a policy.
 383         client_creds = self._get_creds(account_type=self.AccountType.USER,
 384                                        assigned_silo=silo,
 385                                        assigned_policy=wrong_policy)
 386         client_dn_str = str(client_creds.get_dn())
 387
 388         # Add the user to the silo as a member.
 389         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 390                           expect_attr=False)
 391
 392         # Request a Kerberos ticket with a lifetime of two hours, and assert
 393         # that the actual lifetime matches the user lifetime set in the
 394         # appropriate policy.
 395         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 396         tgt = self._get_tgt(client_creds, till=till)
 397         self.check_ticket_times(tgt, expected_life=user_life,
 398                                 expected_renew_life=user_life)
 399
 400     def test_authn_silo_and_policy_tgt_lifetime_computer(self):
 401         user_life = 111
 402         computer_life = 222
 403         service_life = 333
 404         policy_id = self.get_new_username()
 405         policy = self.create_authn_policy(policy_id,
 406                                           enforced=True,
 407                                           user_tgt_lifetime=user_life,
 408                                           computer_tgt_lifetime=computer_life,
 409                                           service_tgt_lifetime=service_life)
 410
 411         wrong_policy_id = self.get_new_username()
 412         wrong_policy = self.create_authn_policy(wrong_policy_id,
 413                                                 enforced=True,
 414                                                 user_tgt_lifetime=444,
 415                                                 computer_tgt_lifetime=555,
 416                                                 service_tgt_lifetime=666)
 417
 418         # Create an authentication silo with our existing policies.
 419         silo_id = self.get_new_username()
 420         silo = self.create_authn_silo(silo_id,
 421                                       user_policy=str(wrong_policy),
 422                                       computer_policy=str(policy),
 423                                       service_policy=str(wrong_policy),
 424                                       enforced=True)
 425
 426         # Create a computer account assigned to the silo, and also to a policy.
 427         client_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
 428                                        assigned_silo=silo,
 429                                        assigned_policy=wrong_policy)
 430         client_dn_str = str(client_creds.get_dn())
 431
 432         # Add the computer to the silo as a member.
 433         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 434                           expect_attr=False)
 435
 436         # Request a Kerberos ticket with a lifetime of two hours, and assert
 437         # that the actual lifetime matches the computer lifetime set in the
 438         # appropriate policy.
 439         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 440         tgt = self._get_tgt(client_creds, till=till)
 441         self.check_ticket_times(tgt, expected_life=computer_life,
 442                                 expected_renew_life=computer_life)
 443
 444     def test_authn_silo_and_policy_tgt_lifetime_service(self):
 445         user_life = 111
 446         computer_life = 222
 447         service_life = 333
 448         policy_id = self.get_new_username()
 449         policy = self.create_authn_policy(policy_id,
 450                                           enforced=True,
 451                                           user_tgt_lifetime=user_life,
 452                                           computer_tgt_lifetime=computer_life,
 453                                           service_tgt_lifetime=service_life)
 454
 455         wrong_policy_id = self.get_new_username()
 456         wrong_policy = self.create_authn_policy(wrong_policy_id,
 457                                                 enforced=True,
 458                                                 user_tgt_lifetime=444,
 459                                                 computer_tgt_lifetime=555,
 460                                                 service_tgt_lifetime=666)
 461
 462         # Create an authentication silo with our existing policies.
 463         silo_id = self.get_new_username()
 464         silo = self.create_authn_silo(silo_id,
 465                                       user_policy=str(wrong_policy),
 466                                       computer_policy=str(wrong_policy),
 467                                       service_policy=str(policy),
 468                                       enforced=True)
 469
 470         # Create a managed service account assigned to the silo, and also to a
 471         # policy.
 472         client_creds = self._get_creds(
 473             account_type=self.AccountType.MANAGED_SERVICE,
 474             assigned_silo=silo,
 475             assigned_policy=wrong_policy)
 476         client_dn_str = str(client_creds.get_dn())
 477
 478         # Add the managed service account to the silo as a member.
 479         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 480                           expect_attr=False)
 481
 482         # Request a Kerberos ticket with a lifetime of two hours, and assert
 483         # that the actual lifetime matches the service lifetime set in the
 484         # appropriate policy.
 485         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 486         tgt = self._get_tgt(client_creds, till=till)
 487         self.check_ticket_times(tgt, expected_life=service_life,
 488                                 expected_renew_life=service_life)
 489
 490     def test_authn_policy_tgt_lifetime_max(self):
 491         # Create an authentication policy with the maximum allowable TGT
 492         # lifetime set.
 493         INT64_MAX = 0x7fff_ffff_ffff_ffff
 494         max_lifetime = INT64_MAX // 10_000_000
 495         policy_id = self.get_new_username()
 496         policy = self.create_authn_policy(policy_id,
 497                                           enforced=True,
 498                                           user_tgt_lifetime=max_lifetime)
 499
 500         # Create a user account with the assigned policy.
 501         client_creds = self._get_creds(account_type=self.AccountType.USER,
 502                                        assigned_policy=policy)
 503
 504         # Request a Kerberos ticket with a ‘till’ time far in the
 505         # future, and assert that the actual lifetime is the maximum
 506         # allowed by the Default Domain policy.
 507         till = '99991231235959Z'
 508         expected_lifetime = self.get_max_ticket_life()
 509         tgt = self._get_tgt(client_creds, till=till)
 510         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 511                                 expected_renew_life=expected_lifetime)
 512
 513     def test_authn_policy_tgt_lifetime_min(self):
 514         # Create an authentication policy with the minimum allowable TGT
 515         # lifetime set.
 516         INT64_MIN = -0x8000_0000_0000_0000
 517         min_lifetime = round(INT64_MIN / 10_000_000)
 518         policy_id = self.get_new_username()
 519         policy = self.create_authn_policy(policy_id,
 520                                           enforced=True,
 521                                           user_tgt_lifetime=min_lifetime)
 522
 523         # Create a user account with the assigned policy.
 524         client_creds = self._get_creds(account_type=self.AccountType.USER,
 525                                        assigned_policy=policy)
 526
 527         # Request a Kerberos ticket with a lifetime of two hours. The request
 528         # should fail with a NEVER_VALID error.
 529         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 530         self._get_tgt(client_creds, till=till,
 531                       expected_error=KDC_ERR_NEVER_VALID,
 532                       expect_status=True,
 533                       expected_status=ntstatus.NT_STATUS_TIME_DIFFERENCE_AT_DC)
 534
 535     def test_authn_policy_tgt_lifetime_zero(self):
 536         # Create an authentication policy with the TGT lifetime set to zero.
 537         lifetime = 0
 538         policy_id = self.get_new_username()
 539         policy = self.create_authn_policy(policy_id,
 540                                           enforced=True,
 541                                           user_tgt_lifetime=lifetime)
 542
 543         # Create a user account with the assigned policy.
 544         client_creds = self._get_creds(account_type=self.AccountType.USER,
 545                                        assigned_policy=policy)
 546
 547         # Request a Kerberos ticket with a ‘till’ time far in the
 548         # future. Assert that the actual lifetime is the maximum
 549         # allowed by the Default Domain Policy
 550         till = '99991231235959Z'
 551         expected_lifetime = self.get_max_ticket_life()
 552         expected_renew_life = self.get_max_renew_life()
 553         tgt = self._get_tgt(client_creds, till=till)
 554         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 555                                 expected_renew_life=expected_renew_life)
 556
 557     def test_authn_policy_tgt_lifetime_one_second(self):
 558         # Create an authentication policy with the TGT lifetime set to one
 559         # second.
 560         lifetime = 1
 561         policy_id = self.get_new_username()
 562         policy = self.create_authn_policy(policy_id,
 563                                           enforced=True,
 564                                           user_tgt_lifetime=lifetime)
 565
 566         # Create a user account with the assigned policy.
 567         client_creds = self._get_creds(account_type=self.AccountType.USER,
 568                                        assigned_policy=policy)
 569
 570         # Request a Kerberos ticket with a lifetime of two hours, and assert
 571         # that the actual lifetime matches the user lifetime set in the
 572         # appropriate policy.
 573         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 574         tgt = self._get_tgt(client_creds, till=till)
 575         self.check_ticket_times(tgt, expected_life=lifetime,
 576                                 expected_renew_life=lifetime)
 577
 578     def test_authn_policy_tgt_lifetime_kpasswd_lifetime(self):
 579         # Create an authentication policy with the TGT lifetime set to two
 580         # minutes (the lifetime of a kpasswd ticket).
 581         lifetime = 2 * 60
 582         policy_id = self.get_new_username()
 583         policy = self.create_authn_policy(policy_id,
 584                                           enforced=True,
 585                                           user_tgt_lifetime=lifetime)
 586
 587         # Create a user account with the assigned policy.
 588         client_creds = self._get_creds(account_type=self.AccountType.USER,
 589                                        assigned_policy=policy)
 590
 591         # Request a Kerberos ticket with a lifetime of two hours, and assert
 592         # that the actual lifetime matches the user lifetime set in the
 593         # appropriate policy.
 594         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 595         tgt = self._get_tgt(client_creds, till=till)
 596         self.check_ticket_times(tgt, expected_life=lifetime,
 597                                 expected_renew_life=lifetime)
 598
 599     def test_authn_policy_tgt_lifetime_short_protected(self):
 600         # Create an authentication policy with a short TGT lifetime set.
 601         lifetime = 111
 602         policy_id = self.get_new_username()
 603         policy = self.create_authn_policy(policy_id,
 604                                           enforced=True,
 605                                           user_tgt_lifetime=lifetime)
 606
 607         # Create a user account with the assigned policy, belonging to the
 608         # Protected Users group.
 609         client_creds = self._get_creds(account_type=self.AccountType.USER,
 610                                        protected=True,
 611                                        assigned_policy=policy)
 612
 613         # Request a Kerberos ticket with a lifetime of two hours, and assert
 614         # that the actual lifetime matches the user lifetime set in the policy.
 615         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 616         tgt = self._get_tgt(client_creds, till=till)
 617         self.check_ticket_times(tgt, expected_life=lifetime,
 618                                 expected_renew_life=lifetime)
 619
 620     def test_authn_policy_tgt_lifetime_long_protected(self):
 621         # Create an authentication policy with a long TGT lifetime set. This
 622         # exceeds the lifetime of four hours enforced by Protected Users.
 623         lifetime = 6 * 60 * 60  # 6 hours
 624         policy_id = self.get_new_username()
 625         policy = self.create_authn_policy(policy_id,
 626                                           enforced=True,
 627                                           user_tgt_lifetime=lifetime)
 628
 629         # Create a user account with the assigned policy, belonging to the
 630         # Protected Users group.
 631         client_creds = self._get_creds(account_type=self.AccountType.USER,
 632                                        protected=True,
 633                                        assigned_policy=policy)
 634
 635         # Request a Kerberos ticket with a lifetime of eight hours, and assert
 636         # that the actual lifetime matches the user lifetime set in the policy,
 637         # taking precedence over the lifetime enforced by Protected Users.
 638         till = self.get_KerberosTime(offset=8 * 60 * 60)  # 8 hours
 639         tgt = self._get_tgt(client_creds, till=till)
 640         self.check_ticket_times(tgt, expected_life=lifetime,
 641                                 expected_renew_life=lifetime)
 642
 643     def test_authn_policy_tgt_lifetime_zero_protected(self):
 644         # Create an authentication policy with the TGT lifetime set to zero.
 645         policy_id = self.get_new_username()
 646         policy = self.create_authn_policy(policy_id,
 647                                           enforced=True,
 648                                           user_tgt_lifetime=0)
 649
 650         # Create a user account with the assigned policy, belonging to the
 651         # Protected Users group.
 652         client_creds = self._get_creds(account_type=self.AccountType.USER,
 653                                        protected=True,
 654                                        assigned_policy=policy)
 655
 656         # Request a Kerberos ticket with a lifetime of six hours, and assert
 657         # that the actual lifetime is the four hours enforced by Protected
 658         # Users.
 659         till = self.get_KerberosTime(offset=6 * 60 * 60)  # 6 hours
 660         tgt = self._get_tgt(client_creds, till=till)
 661         self.check_ticket_times(tgt, expected_life=4 * 60 * 60,
 662                                 expected_renew_life=4 * 60 * 60)
 663
 664     def test_authn_policy_tgt_lifetime_none_protected(self):
 665         # Create an authentication policy with no TGT lifetime set.
 666         policy_id = self.get_new_username()
 667         policy = self.create_authn_policy(policy_id,
 668                                           enforced=True)
 669
 670         # Create a user account with the assigned policy, belonging to the
 671         # Protected Users group.
 672         client_creds = self._get_creds(account_type=self.AccountType.USER,
 673                                        protected=True,
 674                                        assigned_policy=policy)
 675
 676         # Request a Kerberos ticket with a lifetime of six hours, and assert
 677         # that the actual lifetime is the four hours enforced by Protected
 678         # Users.
 679         till = self.get_KerberosTime(offset=6 * 60 * 60)  # 6 hours
 680         tgt = self._get_tgt(client_creds, till=till)
 681         self.check_ticket_times(tgt, expected_life=4 * 60 * 60,
 682                                 expected_renew_life=4 * 60 * 60)
 683
 684     def test_authn_policy_tgt_lifetime_unenforced_protected(self):
 685         # Create an unenforced authentication policy with a TGT lifetime set.
 686         lifetime = 123
 687         policy_id = self.get_new_username()
 688         policy = self.create_authn_policy(policy_id,
 689                                           enforced=False,
 690                                           user_tgt_lifetime=lifetime)
 691
 692         # Create a user account with the assigned policy, belonging to the
 693         # Protected Users group.
 694         client_creds = self._get_creds(account_type=self.AccountType.USER,
 695                                        protected=True,
 696                                        assigned_policy=policy)
 697
 698         # Request a Kerberos ticket with a lifetime of six hours, and assert
 699         # that the actual lifetime is the four hours enforced by Protected
 700         # Users.
 701         till = self.get_KerberosTime(offset=6 * 60 * 60)  # 6 hours
 702         tgt = self._get_tgt(client_creds, till=till)
 703         self.check_ticket_times(tgt, expected_life=4 * 60 * 60,
 704                                 expected_renew_life=4 * 60 * 60)
 705
 706     def test_authn_policy_not_enforced(self):
 707         # Create an authentication policy with the TGT lifetime set. The policy
 708         # is not enforced.
 709         lifetime = 123
 710         policy_id = self.get_new_username()
 711         policy = self.create_authn_policy(policy_id,
 712                                           user_tgt_lifetime=lifetime)
 713
 714         # Create a user account with the assigned policy.
 715         client_creds = self._get_creds(account_type=self.AccountType.USER,
 716                                        assigned_policy=policy)
 717
 718         # Request a Kerberos ticket with a ‘till’ time far in the
 719         # future. Assert that the actual lifetime is the maximum allowed by
 720         # the Default Domain Policy.
 721         till = '99991231235959Z'
 722         expected_lifetime = self.get_max_ticket_life()
 723         expected_renew_life = self.get_max_renew_life()
 724         tgt = self._get_tgt(client_creds, till=till)
 725         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 726                                 expected_renew_life=expected_renew_life)
 727
 728     def test_authn_policy_unenforced(self):
 729         # Create an authentication policy with the TGT lifetime set. The policy
 730         # is set to be unenforced.
 731         lifetime = 123
 732         policy_id = self.get_new_username()
 733         policy = self.create_authn_policy(policy_id,
 734                                           enforced=False,
 735                                           user_tgt_lifetime=lifetime)
 736
 737         # Create a user account with the assigned policy.
 738         client_creds = self._get_creds(account_type=self.AccountType.USER,
 739                                        assigned_policy=policy)
 740
 741         # Request a Kerberos ticket with a ‘till’ time far in the
 742         # future. Assert that the actual lifetime is the maximum allowed by
 743         # the Default Domain Policy.
 744         till = '99991231235959Z'
 745         expected_lifetime = self.get_max_ticket_life()
 746         expected_renew_life = self.get_max_renew_life()
 747         tgt = self._get_tgt(client_creds, till=till)
 748         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 749                                 expected_renew_life=expected_renew_life)
 750
 751     def test_authn_silo_not_enforced(self):
 752         # Create an authentication policy with the TGT lifetime set.
 753         lifetime = 123
 754         policy_id = self.get_new_username()
 755         policy = self.create_authn_policy(policy_id,
 756                                           enforced=True,
 757                                           user_tgt_lifetime=lifetime)
 758
 759         # Create an authentication silo with our existing policy. The silo is
 760         # not enforced.
 761         silo_id = self.get_new_username()
 762         silo = self.create_authn_silo(silo_id,
 763                                       user_policy=str(policy))
 764
 765         # Create a user account assigned to the silo.
 766         client_creds = self._get_creds(account_type=self.AccountType.USER,
 767                                        assigned_silo=silo)
 768         client_dn_str = str(client_creds.get_dn())
 769
 770         # Add the user to the silo as a member.
 771         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 772                           expect_attr=False)
 773
 774         # Request a Kerberos ticket with a ‘till’ time far in the
 775         # future. Assert that the actual lifetime is the maximum allowed by
 776         # the Default Domain Policy.
 777         till = '99991231235959Z'
 778         expected_lifetime = self.get_max_ticket_life()
 779         expected_renew_life = self.get_max_renew_life()
 780         tgt = self._get_tgt(client_creds, till=till)
 781         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 782                                 expected_renew_life=expected_renew_life)
 783
 784     def test_authn_silo_unenforced(self):
 785         # Create an authentication policy with the TGT lifetime set.
 786         lifetime = 123
 787         policy_id = self.get_new_username()
 788         policy = self.create_authn_policy(policy_id,
 789                                           enforced=True,
 790                                           user_tgt_lifetime=lifetime)
 791
 792         # Create an authentication silo with our existing policy. The silo is
 793         # set to be unenforced.
 794         silo_id = self.get_new_username()
 795         silo = self.create_authn_silo(silo_id,
 796                                       user_policy=str(policy),
 797                                       enforced=False)
 798
 799         # Create a user account assigned to the silo.
 800         client_creds = self._get_creds(account_type=self.AccountType.USER,
 801                                        assigned_silo=silo)
 802         client_dn_str = str(client_creds.get_dn())
 803
 804         # Add the user to the silo as a member.
 805         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 806                           expect_attr=False)
 807
 808         # Request a Kerberos ticket with a ‘till’ time far in the
 809         # future. Assert that the actual lifetime is the maximum allowed by
 810         # the Default Domain Policy.
 811         till = '99991231235959Z'
 812         expected_lifetime = self.get_max_ticket_life()
 813         expected_renew_life = self.get_max_renew_life()
 814         tgt = self._get_tgt(client_creds, till=till)
 815         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 816                                 expected_renew_life=expected_renew_life)
 817
 818     def test_authn_silo_not_enforced_policy(self):
 819         # Create an authentication policy with the TGT lifetime set. The policy
 820         # is not enforced.
 821         lifetime = 123
 822         policy_id = self.get_new_username()
 823         policy = self.create_authn_policy(policy_id,
 824                                           user_tgt_lifetime=lifetime)
 825
 826         # Create an authentication silo with our existing policy.
 827         silo_id = self.get_new_username()
 828         silo = self.create_authn_silo(silo_id,
 829                                       user_policy=str(policy),
 830                                       enforced=True)
 831
 832         # Create a user account assigned to the silo.
 833         client_creds = self._get_creds(account_type=self.AccountType.USER,
 834                                        assigned_silo=silo)
 835         client_dn_str = str(client_creds.get_dn())
 836
 837         # Add the user to the silo as a member.
 838         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 839                           expect_attr=False)
 840
 841         # Request a Kerberos ticket with a lifetime of two hours. Despite the
 842         # fact that the policy is unenforced, the actual lifetime matches the
 843         # user lifetime set in the appropriate policy.
 844         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 845         tgt = self._get_tgt(client_creds, till=till)
 846         self.check_ticket_times(tgt, expected_life=lifetime,
 847                                 expected_renew_life=lifetime)
 848
 849     def test_authn_silo_unenforced_policy(self):
 850         # Create an authentication policy with the TGT lifetime set. The policy
 851         # is set to be unenforced.
 852         lifetime = 123
 853         policy_id = self.get_new_username()
 854         policy = self.create_authn_policy(policy_id,
 855                                           enforced=False,
 856                                           user_tgt_lifetime=lifetime)
 857
 858         # Create an authentication silo with our existing policy.
 859         silo_id = self.get_new_username()
 860         silo = self.create_authn_silo(silo_id,
 861                                       user_policy=str(policy),
 862                                       enforced=True)
 863
 864         # Create a user account assigned to the silo.
 865         client_creds = self._get_creds(account_type=self.AccountType.USER,
 866                                        assigned_silo=silo)
 867         client_dn_str = str(client_creds.get_dn())
 868
 869         # Add the user to the silo as a member.
 870         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 871                           expect_attr=False)
 872
 873         # Request a Kerberos ticket with a lifetime of two hours. Despite the
 874         # fact that the policy is unenforced, the actual lifetime matches the
 875         # user lifetime set in the appropriate policy.
 876         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
 877         tgt = self._get_tgt(client_creds, till=till)
 878         self.check_ticket_times(tgt, expected_life=lifetime,
 879                                 expected_renew_life=lifetime)
 880
 881     def test_authn_silo_not_enforced_and_assigned_policy(self):
 882         # Create an authentication policy with the TGT lifetime set.
 883         silo_lifetime = 123
 884         silo_policy_id = self.get_new_username()
 885         silo_policy = self.create_authn_policy(silo_policy_id,
 886                                                enforced=True,
 887                                                user_tgt_lifetime=silo_lifetime)
 888
 889         # Create an authentication silo with our existing policy. The silo is
 890         # not enforced.
 891         silo_id = self.get_new_username()
 892         silo = self.create_authn_silo(silo_id,
 893                                       user_policy=str(silo_policy))
 894
 895         # Create a second policy with a different lifetime, so we can verify
 896         # the correct policy is enforced.
 897         lifetime = 456
 898         policy_id = self.get_new_username()
 899         policy = self.create_authn_policy(policy_id,
 900                                           enforced=True,
 901                                           user_tgt_lifetime=lifetime)
 902
 903         # Create a user account assigned to the silo, and also to the policy.
 904         client_creds = self._get_creds(account_type=self.AccountType.USER,
 905                                        assigned_silo=silo,
 906                                        assigned_policy=policy)
 907         client_dn_str = str(client_creds.get_dn())
 908
 909         # Add the user to the silo as a member.
 910         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 911                           expect_attr=False)
 912
 913         # Request a Kerberos ticket with a ‘till’ time far in the
 914         # future. Assert that the actual lifetime is the maximum
 915         # allowed by the Default Domain Policy. The directly-assigned
 916         # policy is not enforced.
 917         till = '99991231235959Z'
 918         expected_lifetime = self.get_max_ticket_life()
 919         expected_renew_life = self.get_max_renew_life()
 920         tgt = self._get_tgt(client_creds, till=till)
 921         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 922                                 expected_renew_life=expected_renew_life)
 923
 924     def test_authn_silo_unenforced_and_assigned_policy(self):
 925         # Create an authentication policy with the TGT lifetime set.
 926         silo_lifetime = 123
 927         silo_policy_id = self.get_new_username()
 928         silo_policy = self.create_authn_policy(silo_policy_id,
 929                                                enforced=True,
 930                                                user_tgt_lifetime=silo_lifetime)
 931
 932         # Create an authentication silo with our existing policy. The silo is
 933         # set to be unenforced.
 934         silo_id = self.get_new_username()
 935         silo = self.create_authn_silo(silo_id,
 936                                       user_policy=str(silo_policy),
 937                                       enforced=False)
 938
 939         # Create a second policy with a different lifetime, so we can verify
 940         # the correct policy is enforced.
 941         lifetime = 456
 942         policy_id = self.get_new_username()
 943         policy = self.create_authn_policy(policy_id,
 944                                           enforced=True,
 945                                           user_tgt_lifetime=lifetime)
 946
 947         # Create a user account assigned to the silo, and also to the policy.
 948         client_creds = self._get_creds(account_type=self.AccountType.USER,
 949                                        assigned_silo=silo,
 950                                        assigned_policy=policy)
 951         client_dn_str = str(client_creds.get_dn())
 952
 953         # Add the user to the silo as a member.
 954         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 955                           expect_attr=False)
 956
 957         # Request a Kerberos ticket with a ‘till’ time far in the
 958         # future. Assert that the actual lifetime is the maximum
 959         # allowed by the Default Domain Policy. The directly-assigned
 960         # policy is not enforced.
 961         till = '99991231235959Z'
 962         expected_lifetime = self.get_max_ticket_life()
 963         expected_renew_life = self.get_max_renew_life()
 964         tgt = self._get_tgt(client_creds, till=till)
 965         self.check_ticket_times(tgt, expected_life=expected_lifetime,
 966                                 expected_renew_life=expected_renew_life)
 967
 968     def test_authn_silo_not_enforced_policy_and_assigned_policy(self):
 969         # Create an authentication policy with the TGT lifetime set. The policy
 970         # is not enforced.
 971         silo_lifetime = 123
 972         silo_policy_id = self.get_new_username()
 973         silo_policy = self.create_authn_policy(silo_policy_id,
 974                                                user_tgt_lifetime=silo_lifetime)
 975
 976         # Create an authentication silo with our existing policy.
 977         silo_id = self.get_new_username()
 978         silo = self.create_authn_silo(silo_id,
 979                                       user_policy=str(silo_policy),
 980                                       enforced=True)
 981
 982         # Create a second policy with a different lifetime, so we can verify
 983         # the correct policy is enforced.
 984         lifetime = 456
 985         policy_id = self.get_new_username()
 986         policy = self.create_authn_policy(policy_id,
 987                                           enforced=True,
 988                                           user_tgt_lifetime=lifetime)
 989
 990         # Create a user account assigned to the silo, and also to the policy.
 991         client_creds = self._get_creds(account_type=self.AccountType.USER,
 992                                        assigned_silo=silo,
 993                                        assigned_policy=policy)
 994         client_dn_str = str(client_creds.get_dn())
 995
 996         # Add the user to the silo as a member.
 997         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
 998                           expect_attr=False)
 999
1000         # Request a Kerberos ticket with a lifetime of two hours. Despite the
1001         # fact that the policy is unenforced, the actual lifetime matches the
1002         # user lifetime set in the appropriate policy. The directly-assigned
1003         # policy is not enforced.
1004         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
1005         tgt = self._get_tgt(client_creds, till=till)
1006         self.check_ticket_times(tgt, expected_life=silo_lifetime,
1007                                 expected_renew_life=silo_lifetime)
1008
1009     def test_authn_silo_unenforced_policy_and_assigned_policy(self):
1010         # Create an authentication policy with the TGT lifetime set. The policy
1011         # is set to be unenforced.
1012         silo_lifetime = 123
1013         silo_policy_id = self.get_new_username()
1014         silo_policy = self.create_authn_policy(silo_policy_id,
1015                                                enforced=False,
1016                                                user_tgt_lifetime=silo_lifetime)
1017
1018         # Create an authentication silo with our existing policy.
1019         silo_id = self.get_new_username()
1020         silo = self.create_authn_silo(silo_id,
1021                                       user_policy=str(silo_policy),
1022                                       enforced=True)
1023
1024         # Create a second policy with a different lifetime, so we can verify
1025         # the correct policy is enforced.
1026         lifetime = 456
1027         policy_id = self.get_new_username()
1028         policy = self.create_authn_policy(policy_id,
1029                                           enforced=True,
1030                                           user_tgt_lifetime=lifetime)
1031
1032         # Create a user account assigned to the silo, and also to the policy.
1033         client_creds = self._get_creds(account_type=self.AccountType.USER,
1034                                        assigned_silo=silo,
1035                                        assigned_policy=policy)
1036         client_dn_str = str(client_creds.get_dn())
1037
1038         # Add the user to the silo as a member.
1039         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
1040                           expect_attr=False)
1041
1042         # Request a Kerberos ticket with a lifetime of two hours. Despite the
1043         # fact that the policy is unenforced, the actual lifetime matches the
1044         # user lifetime set in the appropriate policy. The directly-assigned
1045         # policy is not enforced.
1046         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
1047         tgt = self._get_tgt(client_creds, till=till)
1048         self.check_ticket_times(tgt, expected_life=silo_lifetime,
1049                                 expected_renew_life=silo_lifetime)
1050
1051     def test_authn_silo_not_a_member(self):
1052         # Create an authentication policy with the TGT lifetime set.
1053         lifetime = 123
1054         policy_id = self.get_new_username()
1055         policy = self.create_authn_policy(policy_id,
1056                                           enforced=True,
1057                                           user_tgt_lifetime=lifetime)
1058
1059         # Create an authentication silo with our existing policy.
1060         silo_id = self.get_new_username()
1061         silo = self.create_authn_silo(silo_id,
1062                                       user_policy=str(policy),
1063                                       enforced=True)
1064
1065         # Create a user account assigned to the silo.
1066         client_creds = self._get_creds(account_type=self.AccountType.USER,
1067                                        assigned_silo=silo)
1068
1069         # Do not add the user to the silo as a member.
1070
1071         # Request a Kerberos ticket with a ‘till’ time far in the
1072         # future. Assert that the actual lifetime is the maximum allowed by
1073         # the Default Domain Policy.
1074         till = '99991231235959Z'
1075         expected_lifetime = self.get_max_ticket_life()
1076         expected_renew_life = self.get_max_renew_life()
1077         tgt = self._get_tgt(client_creds, till=till)
1078         self.check_ticket_times(tgt, expected_life=expected_lifetime,
1079                                 expected_renew_life=expected_renew_life)
1080
1081     def test_authn_silo_not_a_member_and_assigned_policy(self):
1082         # Create an authentication policy with the TGT lifetime set.
1083         silo_lifetime = 123
1084         silo_policy_id = self.get_new_username()
1085         silo_policy = self.create_authn_policy(silo_policy_id,
1086                                                enforced=True,
1087                                                user_tgt_lifetime=silo_lifetime)
1088
1089         # Create an authentication silo with our existing policy.
1090         silo_id = self.get_new_username()
1091         silo = self.create_authn_silo(silo_id,
1092                                       user_policy=str(silo_policy),
1093                                       enforced=True)
1094
1095         # Create a second policy with a different lifetime, so we can verify
1096         # the correct policy is enforced.
1097         lifetime = 456
1098         policy_id = self.get_new_username()
1099         policy = self.create_authn_policy(policy_id,
1100                                           enforced=True,
1101                                           user_tgt_lifetime=lifetime)
1102
1103         # Create a user account assigned to the silo, and also to the policy.
1104         client_creds = self._get_creds(account_type=self.AccountType.USER,
1105                                        assigned_silo=silo,
1106                                        assigned_policy=policy)
1107
1108         # Do not add the user to the silo as a member.
1109
1110         # Request a Kerberos ticket with a lifetime of two hours, and assert
1111         # that the actual lifetime matches the user lifetime set in the
1112         # directly-assigned policy.
1113         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
1114         tgt = self._get_tgt(client_creds, till=till)
1115         self.check_ticket_times(tgt, expected_life=lifetime,
1116                                 expected_renew_life=lifetime)
1117
1118     def test_authn_silo_not_assigned(self):
1119         # Create an authentication policy with the TGT lifetime set.
1120         lifetime = 123
1121         policy_id = self.get_new_username()
1122         policy = self.create_authn_policy(policy_id,
1123                                           enforced=True,
1124                                           user_tgt_lifetime=lifetime)
1125
1126         # Create an authentication silo with our existing policies.
1127         silo_id = self.get_new_username()
1128         silo = self.create_authn_silo(silo_id,
1129                                       user_policy=str(policy),
1130                                       enforced=True)
1131
1132         # Create a user account, but don’t assign it to the silo.
1133         client_creds = self.get_cached_creds(
1134             account_type=self.AccountType.USER)
1135         client_dn_str = str(client_creds.get_dn())
1136
1137         # Add the user to the silo as a member.
1138         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
1139                           expect_attr=False)
1140
1141         # Request a Kerberos ticket with a ‘till’ time far in the
1142         # future. Assert that the actual lifetime is the maximum allowed by
1143         # the Default Domain Policy.
1144         till = '99991231235959Z'
1145         expected_lifetime = self.get_max_ticket_life()
1146         expected_renew_life = self.get_max_renew_life()
1147         tgt = self._get_tgt(client_creds, till=till)
1148         self.check_ticket_times(tgt, expected_life=expected_lifetime,
1149                                 expected_renew_life=expected_renew_life)
1150
1151     def test_authn_silo_not_assigned_and_assigned_policy(self):
1152         # Create an authentication policy with the TGT lifetime set.
1153         lifetime = 123
1154         policy_id = self.get_new_username()
1155         policy = self.create_authn_policy(policy_id,
1156                                           enforced=True,
1157                                           user_tgt_lifetime=lifetime)
1158
1159         # Create an authentication silo with our existing policies.
1160         silo_id = self.get_new_username()
1161         silo = self.create_authn_silo(silo_id,
1162                                       user_policy=str(policy),
1163                                       enforced=True)
1164
1165         # Create a second policy with a different lifetime, so we can verify
1166         # the correct policy is enforced.
1167         lifetime = 456
1168         policy_id = self.get_new_username()
1169         policy = self.create_authn_policy(policy_id,
1170                                           enforced=True,
1171                                           user_tgt_lifetime=lifetime)
1172
1173         # Create a user account assigned to the policy, but not to the silo.
1174         client_creds = self._get_creds(account_type=self.AccountType.USER,
1175                                        assigned_policy=policy)
1176         client_dn_str = str(client_creds.get_dn())
1177
1178         # Add the user to the silo as a member.
1179         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
1180                           expect_attr=False)
1181
1182         # Request a Kerberos ticket with a lifetime of two hours, and assert
1183         # that the actual lifetime matches the user lifetime set in the
1184         # directly-assigned policy.
1185         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
1186         tgt = self._get_tgt(client_creds, till=till)
1187         self.check_ticket_times(tgt, expected_life=lifetime,
1188                                 expected_renew_life=lifetime)
1189
1190     def test_authn_silo_no_applicable_policy(self):
1191         # Create an authentication policy with the TGT lifetime set.
1192         user_life = 111
1193         policy_id = self.get_new_username()
1194         policy = self.create_authn_policy(policy_id,
1195                                           enforced=True,
1196                                           user_tgt_lifetime=user_life)
1197
1198         # Create an authentication silo containing no policies.
1199         silo_id = self.get_new_username()
1200         silo = self.create_authn_silo(silo_id,
1201                                       enforced=True)
1202
1203         # Create a user account assigned to the silo, and also to a policy.
1204         client_creds = self._get_creds(account_type=self.AccountType.USER,
1205                                        assigned_silo=silo,
1206                                        assigned_policy=policy)
1207         client_dn_str = str(client_creds.get_dn())
1208
1209         # Add the user to the silo as a member.
1210         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
1211                           expect_attr=False)
1212
1213         # Request a Kerberos ticket with a ‘till’ time far in the
1214         # future, and assert that the actual lifetime is the maximum
1215         # allowed by the Default Domain Policy.
1216         till = '99991231235959Z'
1217         expected_lifetime = self.get_max_ticket_life()
1218         expected_renew_life = self.get_max_renew_life()
1219         tgt = self._get_tgt(client_creds, till=till)
1220         self.check_ticket_times(tgt, expected_life=expected_lifetime,
1221                                 expected_renew_life=expected_renew_life)
1222
1223     def test_authn_silo_no_tgt_lifetime(self):
1224         # Create an authentication policy with no TGT lifetime set.
1225         silo_policy_id = self.get_new_username()
1226         silo_policy = self.create_authn_policy(silo_policy_id,
1227                                                enforced=True)
1228
1229         # Create a second policy with a lifetime set, so we can verify the
1230         # correct policy is enforced.
1231         policy_id = self.get_new_username()
1232         policy = self.create_authn_policy(policy_id,
1233                                           enforced=True,
1234                                           user_tgt_lifetime=456)
1235
1236         # Create an authentication silo with our existing policy.
1237         silo_id = self.get_new_username()
1238         silo = self.create_authn_silo(silo_id,
1239                                       user_policy=str(silo_policy),
1240                                       enforced=True)
1241
1242         # Create a user account assigned to the silo, and also to a policy.
1243         client_creds = self._get_creds(account_type=self.AccountType.USER,
1244                                        assigned_silo=silo,
1245                                        assigned_policy=policy)
1246         client_dn_str = str(client_creds.get_dn())
1247
1248         # Add the user to the silo as a member.
1249         self.add_to_group(client_dn_str, silo, 'msDS-AuthNPolicySiloMembers',
1250                           expect_attr=False)
1251
1252         # Request a Kerberos ticket with a ‘till’ time far in the
1253         # future, and assert that the actual lifetime is the maximum
1254         # allowed by the Default Domain Policy.
1255         till = '99991231235959Z'
1256         expected_lifetime = self.get_max_ticket_life()
1257         expected_renew_life = self.get_max_renew_life()
1258         tgt = self._get_tgt(client_creds, till=till)
1259         self.check_ticket_times(tgt, expected_life=expected_lifetime,
1260                                 expected_renew_life=expected_renew_life)
1261
1262     def test_not_a_policy(self):
1263         # Create a user account with the assigned policy set to something that
1264         # isn’t a policy.
1265         samdb = self.get_samdb()
1266         client_creds = self._get_creds(
1267             account_type=self.AccountType.USER,
1268             assigned_policy=samdb.get_default_basedn())
1269
1270         # Request a Kerberos ticket with a ‘till’ time far in the
1271         # future, and assert that the actual lifetime is the maximum
1272         # allowed by the Default Domain Policy.
1273         till = '99991231235959Z'
1274         expected_lifetime = self.get_max_ticket_life()
1275         expected_renew_life = self.get_max_renew_life()
1276         tgt = self._get_tgt(client_creds, till=till)
1277         self.check_ticket_times(tgt, expected_life=expected_lifetime,
1278                                 expected_renew_life=expected_renew_life)
1279
1280     def test_not_a_silo(self):
1281         # Create a user account assigned to a silo that isn’t a silo.
1282         samdb = self.get_samdb()
1283         client_creds = self._get_creds(
1284             account_type=self.AccountType.USER,
1285             assigned_silo=samdb.get_default_basedn())
1286
1287         # Request a Kerberos ticket with a ‘till’ time far in the
1288         # future, and assert that the actual lifetime is the maximum
1289         # allowed by the Default Domain Policy.
1290         till = '99991231235959Z'
1291         expected_lifetime = self.get_max_ticket_life()
1292         expected_renew_life = self.get_max_renew_life()
1293         tgt = self._get_tgt(client_creds, till=till)
1294         self.check_ticket_times(tgt, expected_life=expected_lifetime,
1295                                 expected_renew_life=expected_renew_life)
1296
1297     def test_not_a_silo_and_policy(self):
1298         # Create an authentication policy with the TGT lifetime set.
1299         user_life = 123
1300         policy_id = self.get_new_username()
1301         policy = self.create_authn_policy(policy_id,
1302                                           enforced=True,
1303                                           user_tgt_lifetime=user_life)
1304
1305         # Create a user account assigned to a silo that isn’t a silo, and also
1306         # to a policy.
1307         samdb = self.get_samdb()
1308         client_creds = self._get_creds(
1309             account_type=self.AccountType.USER,
1310             assigned_silo=samdb.get_default_basedn(),
1311             assigned_policy=policy)
1312
1313         # Request a Kerberos ticket with a lifetime of two hours, and assert
1314         # that the actual lifetime matches the user lifetime set in the
1315         # directly-assigned policy.
1316         till = self.get_KerberosTime(offset=2 * 60 * 60)  # 2 hours
1317         tgt = self._get_tgt(client_creds, till=till)
1318         self.check_ticket_times(tgt, expected_life=user_life,
1319                                 expected_renew_life=user_life)
1320
1321     def test_authn_policy_allowed_from_empty(self):
1322         # Create a machine account with which to perform FAST.
1323         mach_creds = self.get_cached_creds(
1324             account_type=self.AccountType.COMPUTER)
1325         mach_tgt = self.get_tgt(mach_creds)
1326
1327         # Create an authentication policy with no DACL in the security
1328         # descriptor.
1329         allowed_from = 'O:SY'
1330         policy_id = self.get_new_username()
1331         policy = self.create_authn_policy(policy_id,
1332                                           enforced=True,
1333                                           user_allowed_from=allowed_from)
1334
1335         # Create a user account with the assigned policy.
1336         client_creds = self._get_creds(account_type=self.AccountType.USER,
1337                                        assigned_policy=policy)
1338
1339         # Show that we can authenticate using an armor ticket.
1340         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1341
1342     def test_authn_policy_allowed_from_user_allow(self):
1343         samdb = self.get_samdb()
1344
1345         # Create a machine account with which to perform FAST.
1346         mach_creds = self.get_cached_creds(
1347             account_type=self.AccountType.COMPUTER)
1348         mach_tgt = self.get_tgt(mach_creds)
1349         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1350
1351         # Create an authentication policy that explicitly allows the machine
1352         # account for a user. Include some different TGT lifetimes for testing
1353         # what gets logged.
1354         allowed = f'O:SYD:(A;;CR;;;{mach_sid})'
1355         denied = 'O:SYD:(D;;CR;;;WD)'
1356         policy_id = self.get_new_username()
1357         policy = self.create_authn_policy(policy_id,
1358                                           enforced=True,
1359                                           user_allowed_from=allowed,
1360                                           user_tgt_lifetime=120,
1361                                           computer_tgt_lifetime=240,
1362                                           service_allowed_from=denied,
1363                                           service_tgt_lifetime=360)
1364
1365         # Create a user account with the assigned policy.
1366         client_creds = self._get_creds(account_type=self.AccountType.USER,
1367                                        assigned_policy=policy)
1368
1369         # Show that we can authenticate using an armor ticket.
1370         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1371
1372     def test_authn_policy_allowed_from_user_deny(self):
1373         samdb = self.get_samdb()
1374
1375         # Create a machine account with which to perform FAST.
1376         mach_creds = self.get_cached_creds(
1377             account_type=self.AccountType.COMPUTER)
1378         mach_tgt = self.get_tgt(mach_creds)
1379         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1380
1381         # Create an authentication policy that explicitly denies the machine
1382         # account for a user. Include some different TGT lifetimes for testing
1383         # what gets logged.
1384         allowed = 'O:SYD:(A;;CR;;;WD)'
1385         denied = f'O:SYD:(D;;CR;;;{mach_sid})'
1386         policy_id = self.get_new_username()
1387         policy = self.create_authn_policy(policy_id,
1388                                           enforced=True,
1389                                           user_allowed_from=denied,
1390                                           user_tgt_lifetime=120,
1391                                           computer_tgt_lifetime=240,
1392                                           service_allowed_from=allowed,
1393                                           service_tgt_lifetime=360)
1394
1395         # Create a user account with the assigned policy.
1396         client_creds = self._get_creds(account_type=self.AccountType.USER,
1397                                        assigned_policy=policy)
1398
1399         # Show that we get a policy error when trying to authenticate.
1400         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1401                       expected_error=KDC_ERR_POLICY)
1402
1403     def test_authn_policy_allowed_from_service_allow(self):
1404         samdb = self.get_samdb()
1405
1406         # Create a machine account with which to perform FAST.
1407         mach_creds = self.get_cached_creds(
1408             account_type=self.AccountType.COMPUTER)
1409         mach_tgt = self.get_tgt(mach_creds)
1410         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1411
1412         # Create an authentication policy that explicitly allows the machine
1413         # account for a service.
1414         allowed = f'O:SYD:(A;;CR;;;{mach_sid})'
1415         denied = 'O:SYD:(D;;CR;;;WD)'
1416         policy_id = self.get_new_username()
1417         policy = self.create_authn_policy(policy_id,
1418                                           enforced=True,
1419                                           user_allowed_from=denied,
1420                                           service_allowed_from=allowed)
1421
1422         # Create a managed service account with the assigned policy.
1423         client_creds = self._get_creds(
1424             account_type=self.AccountType.MANAGED_SERVICE,
1425             assigned_policy=policy)
1426
1427         # Show that we can authenticate using an armor ticket.
1428         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1429
1430     def test_authn_policy_allowed_from_service_deny(self):
1431         samdb = self.get_samdb()
1432
1433         # Create a machine account with which to perform FAST.
1434         mach_creds = self.get_cached_creds(
1435             account_type=self.AccountType.COMPUTER)
1436         mach_tgt = self.get_tgt(mach_creds)
1437         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1438
1439         # Create an authentication policy that explicitly denies the machine
1440         # account for a service.
1441         allowed = 'O:SYD:(A;;CR;;;WD)'
1442         denied = f'O:SYD:(D;;CR;;;{mach_sid})'
1443         policy_id = self.get_new_username()
1444         policy = self.create_authn_policy(policy_id,
1445                                           enforced=True,
1446                                           user_allowed_from=allowed,
1447                                           service_allowed_from=denied)
1448
1449         # Create a managed service account with the assigned policy.
1450         client_creds = self._get_creds(
1451             account_type=self.AccountType.MANAGED_SERVICE,
1452             assigned_policy=policy)
1453
1454         # Show that we get a policy error when trying to authenticate.
1455         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1456                       expected_error=KDC_ERR_POLICY)
1457
1458     def test_authn_policy_allowed_from_no_owner(self):
1459         # Create a machine account with which to perform FAST.
1460         mach_creds = self.get_cached_creds(
1461             account_type=self.AccountType.COMPUTER)
1462         mach_tgt = self.get_tgt(mach_creds)
1463
1464         # Create an authentication policy that explicitly allows the machine
1465         # account for a user. Omit the owner (O:SY) from the SDDL.
1466         allowed = 'D:(A;;CR;;;WD)'
1467         policy_id = self.get_new_username()
1468         policy = self.create_authn_policy(policy_id,
1469                                           enforced=True,
1470                                           user_allowed_from=allowed)
1471
1472         # Create a user account with the assigned policy.
1473         client_creds = self._get_creds(account_type=self.AccountType.USER,
1474                                        assigned_policy=policy)
1475
1476         # Show that we get a generic error if the security descriptor lacks an
1477         # owner.
1478         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1479                       expected_error=KDC_ERR_GENERIC)
1480
1481     def test_authn_policy_allowed_from_no_owner_unenforced(self):
1482         # Create a machine account with which to perform FAST.
1483         mach_creds = self.get_cached_creds(
1484             account_type=self.AccountType.COMPUTER)
1485         mach_tgt = self.get_tgt(mach_creds)
1486
1487         # Create an unenforced authentication policy that explicitly allows the
1488         # machine account for a user. Omit the owner (O:SY) from the SDDL.
1489         allowed = 'D:(A;;CR;;;WD)'
1490         policy_id = self.get_new_username()
1491         policy = self.create_authn_policy(policy_id,
1492                                           enforced=False,
1493                                           user_allowed_from=allowed)
1494
1495         # Create a user account with the assigned policy.
1496         client_creds = self._get_creds(account_type=self.AccountType.USER,
1497                                        assigned_policy=policy)
1498
1499         # Show that we don’t get an error if the policy is unenforced.
1500         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1501
1502     def test_authn_policy_allowed_from_owner_self(self):
1503         samdb = self.get_samdb()
1504
1505         # Create a machine account with which to perform FAST.
1506         mach_creds = self.get_cached_creds(
1507             account_type=self.AccountType.COMPUTER)
1508         mach_tgt = self.get_tgt(mach_creds)
1509         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1510
1511         # Create an authentication policy that explicitly allows the machine
1512         # account for a user. Set the owner to the machine account.
1513         allowed = f'O:{mach_sid}D:(A;;CR;;;WD)'
1514         policy_id = self.get_new_username()
1515         policy = self.create_authn_policy(policy_id,
1516                                           enforced=True,
1517                                           user_allowed_from=allowed)
1518
1519         # Create a user account with the assigned policy.
1520         client_creds = self._get_creds(account_type=self.AccountType.USER,
1521                                        assigned_policy=policy)
1522
1523         # Show that we can authenticate using an armor ticket.
1524         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1525
1526     def test_authn_policy_allowed_from_owner_anon(self):
1527         # Create a machine account with which to perform FAST.
1528         mach_creds = self.get_cached_creds(
1529             account_type=self.AccountType.COMPUTER)
1530         mach_tgt = self.get_tgt(mach_creds)
1531
1532         # Create an authentication policy that explicitly allows the machine
1533         # account for a user. Set the owner to be anonymous.
1534         allowed = 'O:AND:(A;;CR;;;WD)'
1535         policy_id = self.get_new_username()
1536         policy = self.create_authn_policy(policy_id,
1537                                           enforced=True,
1538                                           user_allowed_from=allowed)
1539
1540         # Create a user account with the assigned policy.
1541         client_creds = self._get_creds(account_type=self.AccountType.USER,
1542                                        assigned_policy=policy)
1543
1544         # Show that we can authenticate using an armor ticket.
1545         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1546
1547     def test_authn_policy_allowed_from_no_fast(self):
1548         # Create an authentication policy that restricts authentication.
1549         # Include some different TGT lifetimes for testing what gets logged.
1550         allowed_from = 'O:SY'
1551         policy_id = self.get_new_username()
1552         policy = self.create_authn_policy(policy_id,
1553                                           enforced=True,
1554                                           user_allowed_from=allowed_from,
1555                                           user_tgt_lifetime=115,
1556                                           computer_tgt_lifetime=235,
1557                                           service_tgt_lifetime=355)
1558
1559         # Create a user account with the assigned policy.
1560         client_creds = self._get_creds(account_type=self.AccountType.USER,
1561                                        assigned_policy=policy)
1562
1563         # Show that we cannot authenticate without using an armor ticket.
1564         self._get_tgt(client_creds, expected_error=KDC_ERR_POLICY,
1565                       expect_status=True,
1566                       expected_status=ntstatus.NT_STATUS_INVALID_WORKSTATION)
1567
1568     def test_authn_policy_allowed_from_user_allow_group_not_a_member(self):
1569         samdb = self.get_samdb()
1570
1571         # Create a new group.
1572         group_name = self.get_new_username()
1573         group_dn = self.create_group(samdb, group_name)
1574         group_sid = self.get_objectSid(samdb, group_dn)
1575
1576         # Create a machine account with which to perform FAST and which does
1577         # not belong to the group.
1578         mach_creds = self.get_cached_creds(
1579             account_type=self.AccountType.COMPUTER)
1580         mach_tgt = self.get_tgt(mach_creds)
1581
1582         # Create an authentication policy that allows accounts belonging to the
1583         # group.
1584         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
1585         policy_id = self.get_new_username()
1586         policy = self.create_authn_policy(policy_id,
1587                                           enforced=True,
1588                                           user_allowed_from=allowed)
1589
1590         # Create a user account with the assigned policy.
1591         client_creds = self._get_creds(account_type=self.AccountType.USER,
1592                                        assigned_policy=policy)
1593
1594         # Show that we get a policy error, as the machine account does not
1595         # belong to the group.
1596         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1597                       expected_error=KDC_ERR_POLICY)
1598
1599     def test_authn_policy_allowed_from_user_allow_group_member(self):
1600         samdb = self.get_samdb()
1601
1602         # Create a new group.
1603         group_name = self.get_new_username()
1604         group_dn = self.create_group(samdb, group_name)
1605         group_sid = self.get_objectSid(samdb, group_dn)
1606
1607         # Create a machine account with which to perform FAST that belongs to
1608         # the group.
1609         mach_creds = self.get_cached_creds(
1610             account_type=self.AccountType.COMPUTER,
1611             opts={'member_of': (group_dn,)})
1612         mach_tgt = self.get_tgt(mach_creds)
1613
1614         # Create an authentication policy that allows accounts belonging to the
1615         # group.
1616         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
1617         policy_id = self.get_new_username()
1618         policy = self.create_authn_policy(policy_id,
1619                                           enforced=True,
1620                                           user_allowed_from=allowed)
1621
1622         # Create a user account with the assigned policy.
1623         client_creds = self._get_creds(account_type=self.AccountType.USER,
1624                                        assigned_policy=policy)
1625
1626         # Show that we can authenticate using an armor ticket, since the
1627         # machine account belongs to the group.
1628         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1629
1630     def test_authn_policy_allowed_from_user_allow_domain_local_group(self):
1631         samdb = self.get_samdb()
1632
1633         # Create a new domain-local group.
1634         group_name = self.get_new_username()
1635         group_dn = self.create_group(samdb, group_name,
1636                                      gtype=GroupType.DOMAIN_LOCAL.value)
1637         group_sid = self.get_objectSid(samdb, group_dn)
1638
1639         # Create a machine account with which to perform FAST that belongs to
1640         # the group.
1641         mach_creds = self.get_cached_creds(
1642             account_type=self.AccountType.COMPUTER,
1643             opts={'member_of': (group_dn,)})
1644         mach_tgt = self.get_tgt(mach_creds)
1645
1646         # Create an authentication policy that allows accounts belonging to the
1647         # group.
1648         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
1649         policy_id = self.get_new_username()
1650         policy = self.create_authn_policy(policy_id,
1651                                           enforced=True,
1652                                           user_allowed_from=allowed)
1653
1654         # Create a user account with the assigned policy.
1655         client_creds = self._get_creds(account_type=self.AccountType.USER,
1656                                        assigned_policy=policy)
1657
1658         # Show that the groups in the armor ticket are expanded to include the
1659         # domain-local group.
1660         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1661
1662     def test_authn_policy_allowed_from_user_allow_asserted_identity(self):
1663         # Create a machine account with which to perform FAST.
1664         mach_creds = self.get_cached_creds(
1665             account_type=self.AccountType.COMPUTER)
1666         mach_tgt = self.get_tgt(mach_creds)
1667
1668         # Create an authentication policy that allows accounts with the
1669         # Authentication Authority Asserted Identity SID.
1670         allowed = (
1671             f'O:SYD:(A;;CR;;;'
1672             f'{security.SID_AUTHENTICATION_AUTHORITY_ASSERTED_IDENTITY})'
1673         )
1674         policy_id = self.get_new_username()
1675         policy = self.create_authn_policy(policy_id,
1676                                           enforced=True,
1677                                           user_allowed_from=allowed)
1678
1679         # Create a user account with the assigned policy.
1680         client_creds = self._get_creds(account_type=self.AccountType.USER,
1681                                        assigned_policy=policy)
1682
1683         # Show that authentication is allowed.
1684         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1685
1686     def test_authn_policy_allowed_from_user_allow_claims_valid(self):
1687         # Create a machine account with which to perform FAST.
1688         mach_creds = self.get_cached_creds(
1689             account_type=self.AccountType.COMPUTER)
1690         mach_tgt = self.get_tgt(mach_creds)
1691
1692         # Create an authentication policy that allows accounts with the
1693         # Claims Valid SID.
1694         allowed = f'O:SYD:(A;;CR;;;{security.SID_CLAIMS_VALID})'
1695         policy_id = self.get_new_username()
1696         policy = self.create_authn_policy(policy_id,
1697                                           enforced=True,
1698                                           user_allowed_from=allowed)
1699
1700         # Create a user account with the assigned policy.
1701         client_creds = self._get_creds(account_type=self.AccountType.USER,
1702                                        assigned_policy=policy)
1703
1704         # Show that authentication is allowed.
1705         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1706
1707     def test_authn_policy_allowed_from_user_allow_compounded_auth(self):
1708         # Create a machine account with which to perform FAST.
1709         mach_creds = self.get_cached_creds(
1710             account_type=self.AccountType.COMPUTER)
1711         mach_tgt = self.get_tgt(mach_creds)
1712
1713         # Create an authentication policy that allows accounts with the
1714         # Compounded Authentication SID.
1715         allowed = f'O:SYD:(A;;CR;;;{security.SID_COMPOUNDED_AUTHENTICATION})'
1716         policy_id = self.get_new_username()
1717         policy = self.create_authn_policy(policy_id,
1718                                           enforced=True,
1719                                           user_allowed_from=allowed)
1720
1721         # Create a user account with the assigned policy.
1722         client_creds = self._get_creds(account_type=self.AccountType.USER,
1723                                        assigned_policy=policy)
1724
1725         # Show that authentication is denied.
1726         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1727                       expected_error=KDC_ERR_POLICY)
1728
1729     def test_authn_policy_allowed_from_user_allow_authenticated_users(self):
1730         # Create a machine account with which to perform FAST.
1731         mach_creds = self.get_cached_creds(
1732             account_type=self.AccountType.COMPUTER)
1733         mach_tgt = self.get_tgt(mach_creds)
1734
1735         # Create an authentication policy that allows accounts with the
1736         # Authenticated Users SID.
1737         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_AUTHENTICATED_USERS})'
1738         policy_id = self.get_new_username()
1739         policy = self.create_authn_policy(policy_id,
1740                                           enforced=True,
1741                                           user_allowed_from=allowed)
1742
1743         # Create a user account with the assigned policy.
1744         client_creds = self._get_creds(account_type=self.AccountType.USER,
1745                                        assigned_policy=policy)
1746
1747         # Show that authentication is allowed.
1748         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1749
1750     def test_authn_policy_allowed_from_user_allow_ntlm_authn(self):
1751         # Create a machine account with which to perform FAST.
1752         mach_creds = self.get_cached_creds(
1753             account_type=self.AccountType.COMPUTER)
1754         mach_tgt = self.get_tgt(mach_creds)
1755
1756         # Create an authentication policy that allows accounts with the NTLM
1757         # Authentication SID.
1758         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_NTLM_AUTHENTICATION})'
1759         policy_id = self.get_new_username()
1760         policy = self.create_authn_policy(policy_id,
1761                                           enforced=True,
1762                                           user_allowed_from=allowed)
1763
1764         # Create a user account with the assigned policy.
1765         client_creds = self._get_creds(account_type=self.AccountType.USER,
1766                                        assigned_policy=policy)
1767
1768         # Show that authentication is denied.
1769         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1770                       expected_error=KDC_ERR_POLICY)
1771
1772     def test_authn_policy_allowed_from_user_allow_from_rodc(self):
1773         samdb = self.get_samdb()
1774
1775         # Create a machine account with which to perform FAST.
1776         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
1777                                      allowed_rodc=True)
1778         # Modify the TGT to be issued by an RODC.
1779         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1780         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1781
1782         # Create an authentication policy that explicitly allows the machine
1783         # account for a user.
1784         allowed = f'O:SYD:(A;;CR;;;{mach_sid})'
1785         denied = 'O:SYD:(D;;CR;;;WD)'
1786         policy_id = self.get_new_username()
1787         policy = self.create_authn_policy(policy_id,
1788                                           enforced=True,
1789                                           user_allowed_from=allowed,
1790                                           service_allowed_from=denied)
1791
1792         # Create a user account with the assigned policy.
1793         client_creds = self._get_creds(account_type=self.AccountType.USER,
1794                                        assigned_policy=policy)
1795
1796         # Show that we can authenticate using an armor ticket.
1797         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1798
1799     def test_authn_policy_allowed_from_user_deny_from_rodc(self):
1800         samdb = self.get_samdb()
1801
1802         # Create a machine account with which to perform FAST.
1803         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
1804                                      allowed_rodc=True)
1805         # Modify the TGT to be issued by an RODC.
1806         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1807         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1808
1809         # Create an authentication policy that explicitly denies the machine
1810         # account for a user.
1811         allowed = 'O:SYD:(A;;CR;;;WD)'
1812         denied = f'O:SYD:(D;;CR;;;{mach_sid})'
1813         policy_id = self.get_new_username()
1814         policy = self.create_authn_policy(policy_id,
1815                                           enforced=True,
1816                                           user_allowed_from=denied,
1817                                           service_allowed_from=allowed)
1818
1819         # Create a user account with the assigned policy.
1820         client_creds = self._get_creds(account_type=self.AccountType.USER,
1821                                        assigned_policy=policy)
1822
1823         # Show that we get a policy error when trying to authenticate.
1824         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1825                       expected_error=KDC_ERR_POLICY)
1826
1827     def test_authn_policy_allowed_from_service_allow_from_rodc(self):
1828         samdb = self.get_samdb()
1829
1830         # Create a machine account with which to perform FAST.
1831         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
1832                                      allowed_rodc=True)
1833         # Modify the TGT to be issued by an RODC.
1834         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1835         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1836
1837         # Create an authentication policy that explicitly allows the machine
1838         # account for a service.
1839         allowed = f'O:SYD:(A;;CR;;;{mach_sid})'
1840         denied = 'O:SYD:(D;;CR;;;WD)'
1841         policy_id = self.get_new_username()
1842         policy = self.create_authn_policy(policy_id,
1843                                           enforced=True,
1844                                           user_allowed_from=denied,
1845                                           service_allowed_from=allowed)
1846
1847         # Create a managed service account with the assigned policy.
1848         client_creds = self._get_creds(
1849             account_type=self.AccountType.MANAGED_SERVICE,
1850             assigned_policy=policy)
1851
1852         # Show that we can authenticate using an armor ticket.
1853         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1854
1855     def test_authn_policy_allowed_from_service_deny_from_rodc(self):
1856         samdb = self.get_samdb()
1857
1858         # Create a machine account with which to perform FAST.
1859         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
1860                                      allowed_rodc=True)
1861         # Modify the TGT to be issued by an RODC.
1862         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1863         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
1864
1865         # Create an authentication policy that explicitly denies the machine
1866         # account for a service.
1867         allowed = 'O:SYD:(A;;CR;;;WD)'
1868         denied = f'O:SYD:(D;;CR;;;{mach_sid})'
1869         policy_id = self.get_new_username()
1870         policy = self.create_authn_policy(policy_id,
1871                                           enforced=True,
1872                                           user_allowed_from=allowed,
1873                                           service_allowed_from=denied)
1874
1875         # Create a managed service account with the assigned policy.
1876         client_creds = self._get_creds(
1877             account_type=self.AccountType.MANAGED_SERVICE,
1878             assigned_policy=policy)
1879
1880         # Show that we get a policy error when trying to authenticate.
1881         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1882                       expected_error=KDC_ERR_POLICY)
1883
1884     def test_authn_policy_allowed_from_user_allow_group_not_a_member_from_rodc(self):
1885         samdb = self.get_samdb()
1886
1887         # Create a new group.
1888         group_name = self.get_new_username()
1889         group_dn = self.create_group(samdb, group_name)
1890         group_sid = self.get_objectSid(samdb, group_dn)
1891
1892         # Create a machine account with which to perform FAST and which does
1893         # not belong to the group.
1894         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
1895                                      allowed_rodc=True)
1896         # Modify the TGT to be issued by an RODC.
1897         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1898
1899         # Create an authentication policy that allows accounts belonging to the
1900         # group.
1901         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
1902         policy_id = self.get_new_username()
1903         policy = self.create_authn_policy(policy_id,
1904                                           enforced=True,
1905                                           user_allowed_from=allowed)
1906
1907         # Create a user account with the assigned policy.
1908         client_creds = self._get_creds(account_type=self.AccountType.USER,
1909                                        assigned_policy=policy)
1910
1911         # Show that we get a policy error, as the machine account does not
1912         # belong to the group.
1913         self._get_tgt(client_creds, armor_tgt=mach_tgt,
1914                       expected_error=KDC_ERR_POLICY)
1915
1916     def test_authn_policy_allowed_from_user_allow_group_member_from_rodc(self):
1917         samdb = self.get_samdb()
1918
1919         # Create a new group.
1920         group_name = self.get_new_username()
1921         group_dn = self.create_group(samdb, group_name)
1922         group_sid = self.get_objectSid(samdb, group_dn)
1923
1924         # Create a machine account with which to perform FAST that belongs to
1925         # the group.
1926         mach_creds = self.get_cached_creds(
1927             account_type=self.AccountType.COMPUTER,
1928             opts={'member_of': (group_dn,),
1929                   'allowed_replication_mock': True,
1930                   'revealed_to_mock_rodc': True})
1931         # Modify the TGT to be issued by an RODC.
1932         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1933
1934         # Create an authentication policy that allows accounts belonging to the
1935         # group.
1936         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
1937         policy_id = self.get_new_username()
1938         policy = self.create_authn_policy(policy_id,
1939                                           enforced=True,
1940                                           user_allowed_from=allowed)
1941
1942         # Create a user account with the assigned policy.
1943         client_creds = self._get_creds(account_type=self.AccountType.USER,
1944                                        assigned_policy=policy)
1945
1946         # Show that we can authenticate using an armor ticket, since the
1947         # machine account belongs to the group.
1948         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1949
1950     def test_authn_policy_allowed_from_user_allow_domain_local_group_from_rodc(self):
1951         samdb = self.get_samdb()
1952
1953         # Create a new domain-local group.
1954         group_name = self.get_new_username()
1955         group_dn = self.create_group(samdb, group_name,
1956                                      gtype=GroupType.DOMAIN_LOCAL.value)
1957         group_sid = self.get_objectSid(samdb, group_dn)
1958
1959         # Create a machine account with which to perform FAST that belongs to
1960         # the group.
1961         mach_creds = self.get_cached_creds(
1962             account_type=self.AccountType.COMPUTER,
1963             opts={'member_of': (group_dn,),
1964                   'allowed_replication_mock': True,
1965                   'revealed_to_mock_rodc': True})
1966         # Modify the TGT to be issued by an RODC.
1967         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1968
1969         # Create an authentication policy that allows accounts belonging to the
1970         # group.
1971         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
1972         policy_id = self.get_new_username()
1973         policy = self.create_authn_policy(policy_id,
1974                                           enforced=True,
1975                                           user_allowed_from=allowed)
1976
1977         # Create a user account with the assigned policy.
1978         client_creds = self._get_creds(account_type=self.AccountType.USER,
1979                                        assigned_policy=policy)
1980
1981         # Show that the groups in the armor ticket are expanded to include the
1982         # domain-local group.
1983         self._get_tgt(client_creds, armor_tgt=mach_tgt)
1984
1985     def test_authn_policy_allowed_from_user_allow_asserted_identity_from_rodc(self):
1986         # Create a machine account with which to perform FAST.
1987         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
1988                                      allowed_rodc=True)
1989         # Modify the TGT to be issued by an RODC.
1990         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
1991
1992         # Create an authentication policy that allows accounts with the
1993         # Authentication Authority Asserted Identity SID.
1994         allowed = (
1995             f'O:SYD:(A;;CR;;;'
1996             f'{security.SID_AUTHENTICATION_AUTHORITY_ASSERTED_IDENTITY})'
1997         )
1998         policy_id = self.get_new_username()
1999         policy = self.create_authn_policy(policy_id,
2000                                           enforced=True,
2001                                           user_allowed_from=allowed)
2002
2003         # Create a user account with the assigned policy.
2004         client_creds = self._get_creds(account_type=self.AccountType.USER,
2005                                        assigned_policy=policy)
2006
2007         # Show that authentication is allowed.
2008         self._get_tgt(client_creds, armor_tgt=mach_tgt)
2009
2010     def test_authn_policy_allowed_from_user_allow_claims_valid_from_rodc(self):
2011         # Create a machine account with which to perform FAST.
2012         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2013                                      allowed_rodc=True)
2014         # Modify the TGT to be issued by an RODC.
2015         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
2016
2017         # Create an authentication policy that allows accounts with the
2018         # Claims Valid SID.
2019         allowed = f'O:SYD:(A;;CR;;;{security.SID_CLAIMS_VALID})'
2020         policy_id = self.get_new_username()
2021         policy = self.create_authn_policy(policy_id,
2022                                           enforced=True,
2023                                           user_allowed_from=allowed)
2024
2025         # Create a user account with the assigned policy.
2026         client_creds = self._get_creds(account_type=self.AccountType.USER,
2027                                        assigned_policy=policy)
2028
2029         # Show that authentication is allowed.
2030         self._get_tgt(client_creds, armor_tgt=mach_tgt)
2031
2032     def test_authn_policy_allowed_from_user_allow_compounded_authn_from_rodc(self):
2033         # Create a machine account with which to perform FAST.
2034         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2035                                      allowed_rodc=True)
2036         # Modify the TGT to be issued by an RODC.
2037         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
2038
2039         # Create an authentication policy that allows accounts with the
2040         # Compounded Authentication SID.
2041         allowed = f'O:SYD:(A;;CR;;;{security.SID_COMPOUNDED_AUTHENTICATION})'
2042         policy_id = self.get_new_username()
2043         policy = self.create_authn_policy(policy_id,
2044                                           enforced=True,
2045                                           user_allowed_from=allowed)
2046
2047         # Create a user account with the assigned policy.
2048         client_creds = self._get_creds(account_type=self.AccountType.USER,
2049                                        assigned_policy=policy)
2050
2051         # Show that authentication is denied.
2052         self._get_tgt(client_creds, armor_tgt=mach_tgt,
2053                       expected_error=KDC_ERR_POLICY)
2054
2055     def test_authn_policy_allowed_from_user_allow_authenticated_users_from_rodc(self):
2056         # Create a machine account with which to perform FAST.
2057         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2058                                      allowed_rodc=True)
2059         # Modify the TGT to be issued by an RODC.
2060         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
2061
2062         # Create an authentication policy that allows accounts with the
2063         # Authenticated Users SID.
2064         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_AUTHENTICATED_USERS})'
2065         policy_id = self.get_new_username()
2066         policy = self.create_authn_policy(policy_id,
2067                                           enforced=True,
2068                                           user_allowed_from=allowed)
2069
2070         # Create a user account with the assigned policy.
2071         client_creds = self._get_creds(account_type=self.AccountType.USER,
2072                                        assigned_policy=policy)
2073
2074         # Show that authentication is allowed.
2075         self._get_tgt(client_creds, armor_tgt=mach_tgt)
2076
2077     def test_authn_policy_allowed_from_user_allow_ntlm_authn_from_rodc(self):
2078         # Create a machine account with which to perform FAST.
2079         mach_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2080                                      allowed_rodc=True)
2081         # Modify the TGT to be issued by an RODC.
2082         mach_tgt = self.issued_by_rodc(self.get_tgt(mach_creds))
2083
2084         # Create an authentication policy that allows accounts with the NTLM
2085         # Authentication SID.
2086         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_NTLM_AUTHENTICATION})'
2087         policy_id = self.get_new_username()
2088         policy = self.create_authn_policy(policy_id,
2089                                           enforced=True,
2090                                           user_allowed_from=allowed)
2091
2092         # Create a user account with the assigned policy.
2093         client_creds = self._get_creds(account_type=self.AccountType.USER,
2094                                        assigned_policy=policy)
2095
2096         # Show that authentication is denied.
2097         self._get_tgt(client_creds, armor_tgt=mach_tgt,
2098                       expected_error=KDC_ERR_POLICY)
2099
2100     def test_authn_policy_allowed_from_user_deny_user(self):
2101         samdb = self.get_samdb()
2102
2103         # Create a machine account with which to perform FAST.
2104         mach_creds = self.get_cached_creds(
2105             account_type=self.AccountType.COMPUTER)
2106         mach_tgt = self.get_tgt(mach_creds)
2107         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
2108
2109         # Create a user account.
2110         client_creds = self.get_cached_creds(account_type=self.AccountType.USER,
2111                                              use_cache=False)
2112         client_dn = client_creds.get_dn()
2113         client_sid = self.get_objectSid(samdb, client_dn)
2114
2115         # Create an authentication policy that explicitly allows the machine
2116         # account for a user, while denying the user account itself.
2117         allowed = f'O:SYD:(A;;CR;;;{mach_sid})(D;;CR;;;{client_sid})'
2118         denied = 'O:SYD:(D;;CR;;;WD)'
2119         policy_id = self.get_new_username()
2120         policy = self.create_authn_policy(policy_id,
2121                                           enforced=True,
2122                                           user_allowed_from=allowed,
2123                                           service_allowed_from=denied)
2124
2125         # Assign the policy to the user account.
2126         self.add_attribute(samdb, str(client_dn),
2127                            'msDS-AssignedAuthNPolicy', str(policy))
2128
2129         # Show that authentication is allowed.
2130         self._get_tgt(client_creds, armor_tgt=mach_tgt)
2131
2132     def test_authn_policy_allowed_to_empty(self):
2133         # Create a machine account with which to perform FAST.
2134         mach_creds = self.get_cached_creds(
2135             account_type=self.AccountType.COMPUTER)
2136         mach_tgt = self.get_tgt(mach_creds)
2137
2138         # Create a user account.
2139         client_creds = self.get_cached_creds(
2140             account_type=self.AccountType.USER)
2141         tgt = self.get_tgt(client_creds)
2142
2143         # Create an authentication policy with no DACL in the security
2144         # descriptor.
2145         allowed_to = 'O:SY'
2146         policy_id = self.get_new_username()
2147         policy = self.create_authn_policy(policy_id,
2148                                           enforced=True,
2149                                           computer_allowed_to=allowed_to)
2150
2151         # Create a computer account with the assigned policy.
2152         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2153                                        assigned_policy=policy)
2154
2155         # Show that authentication is allowed.
2156         self._tgs_req(tgt, 0, client_creds, target_creds,
2157                       armor_tgt=mach_tgt)
2158
2159     def test_authn_policy_allowed_to_computer_allow(self):
2160         samdb = self.get_samdb()
2161
2162         # Create a machine account with which to perform FAST.
2163         mach_creds = self.get_cached_creds(
2164             account_type=self.AccountType.COMPUTER)
2165         mach_tgt = self.get_tgt(mach_creds)
2166
2167         # Create a user account.
2168         client_creds = self.get_cached_creds(
2169             account_type=self.AccountType.USER)
2170         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2171         tgt = self.get_tgt(client_creds)
2172
2173         # Create an authentication policy that applies to a computer and
2174         # explicitly allows the user account to obtain a service ticket.
2175         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
2176         denied = 'O:SYD:(D;;CR;;;WD)'
2177         policy_id = self.get_new_username()
2178         policy = self.create_authn_policy(policy_id,
2179                                           enforced=True,
2180                                           user_allowed_to=denied,
2181                                           computer_allowed_to=allowed,
2182                                           service_allowed_to=denied)
2183
2184         # Create a computer account with the assigned policy.
2185         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2186                                        assigned_policy=policy)
2187
2188         # Show that obtaining a service ticket is allowed.
2189         self._tgs_req(tgt, 0, client_creds, target_creds,
2190                       armor_tgt=mach_tgt)
2191
2192     def test_authn_policy_allowed_to_computer_deny(self):
2193         samdb = self.get_samdb()
2194
2195         # Create a machine account with which to perform FAST.
2196         mach_creds = self.get_cached_creds(
2197             account_type=self.AccountType.COMPUTER)
2198         mach_tgt = self.get_tgt(mach_creds)
2199
2200         # Create a user account.
2201         client_creds = self.get_cached_creds(
2202             account_type=self.AccountType.USER)
2203         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2204         tgt = self.get_tgt(client_creds)
2205
2206         # Create an authentication policy that applies to a computer and
2207         # explicitly denies the user account to obtain a service ticket.
2208         denied = f'O:SYD:(D;;CR;;;{client_sid})'
2209         allowed = 'O:SYD:(A;;CR;;;WD)'
2210         policy_id = self.get_new_username()
2211         policy = self.create_authn_policy(policy_id,
2212                                           enforced=True,
2213                                           user_allowed_to=allowed,
2214                                           computer_allowed_to=denied,
2215                                           service_allowed_to=allowed)
2216
2217         # Create a computer account with the assigned policy.
2218         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2219                                        assigned_policy=policy)
2220
2221         # Show that obtaining a service ticket is denied.
2222         self._tgs_req(
2223             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2224             armor_tgt=mach_tgt,
2225             expect_edata=self.expect_padata_outer,
2226             # We aren’t particular about whether or not we get an NTSTATUS.
2227             expect_status=None,
2228             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
2229             check_patypes=False)
2230
2231     def test_authn_policy_allowed_to_computer_allow_but_deny_mach(self):
2232         samdb = self.get_samdb()
2233
2234         # Create a machine account with which to perform FAST.
2235         mach_creds = self.get_cached_creds(
2236             account_type=self.AccountType.COMPUTER)
2237         mach_tgt = self.get_tgt(mach_creds)
2238         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
2239
2240         # Create a user account.
2241         client_creds = self.get_cached_creds(
2242             account_type=self.AccountType.USER)
2243         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2244         tgt = self.get_tgt(client_creds)
2245
2246         # Create an authentication policy that applies to a computer and
2247         # explicitly allows the user account to obtain a service ticket, while
2248         # explicitly denying the machine account.
2249         allowed = f'O:SYD:(A;;CR;;;{client_sid})(D;;CR;;;{mach_sid})'
2250         denied = 'O:SYD:(D;;CR;;;WD)'
2251         policy_id = self.get_new_username()
2252         policy = self.create_authn_policy(policy_id,
2253                                           enforced=True,
2254                                           user_allowed_to=denied,
2255                                           computer_allowed_to=allowed,
2256                                           service_allowed_to=denied)
2257
2258         # Create a computer account with the assigned policy.
2259         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2260                                        assigned_policy=policy)
2261
2262         # Despite the documentation’s claims that the machine account is also
2263         # access-checked, obtaining a service ticket is allowed.
2264         self._tgs_req(tgt, 0, client_creds, target_creds,
2265                       armor_tgt=mach_tgt)
2266
2267     def test_authn_policy_allowed_to_computer_allow_mach(self):
2268         samdb = self.get_samdb()
2269
2270         # Create a machine account with which to perform FAST.
2271         mach_creds = self.get_cached_creds(
2272             account_type=self.AccountType.COMPUTER)
2273         mach_tgt = self.get_tgt(mach_creds)
2274         mach_sid = self.get_objectSid(samdb, mach_creds.get_dn())
2275
2276         # Create a user account.
2277         client_creds = self.get_cached_creds(
2278             account_type=self.AccountType.USER)
2279         tgt = self.get_tgt(client_creds)
2280
2281         # Create an authentication policy that applies to a computer and
2282         # explicitly allows the machine account to obtain a service ticket.
2283         allowed = f'O:SYD:(A;;CR;;;{mach_sid})'
2284         denied = 'O:SYD:(D;;CR;;;WD)'
2285         policy_id = self.get_new_username()
2286         policy = self.create_authn_policy(policy_id,
2287                                           enforced=True,
2288                                           user_allowed_to=denied,
2289                                           computer_allowed_to=allowed,
2290                                           service_allowed_to=denied)
2291
2292         # Create a computer account with the assigned policy.
2293         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2294                                        assigned_policy=policy)
2295
2296         # Show that obtaining a service ticket is denied.
2297         self._tgs_req(
2298             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2299             armor_tgt=mach_tgt,
2300             expect_edata=self.expect_padata_outer,
2301             # We aren’t particular about whether or not we get an NTSTATUS.
2302             expect_status=None,
2303             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
2304             check_patypes=False)
2305
2306     def test_authn_policy_allowed_no_fast(self):
2307         samdb = self.get_samdb()
2308
2309         # Create a user account.
2310         client_creds = self.get_cached_creds(
2311             account_type=self.AccountType.USER)
2312         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2313         tgt = self.get_tgt(client_creds)
2314
2315         # Create an authentication policy that applies to a computer and
2316         # explicitly allows the user account to obtain a service ticket.
2317         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
2318         denied = 'O:SYD:(D;;CR;;;WD)'
2319         policy_id = self.get_new_username()
2320         policy = self.create_authn_policy(policy_id,
2321                                           enforced=True,
2322                                           user_allowed_to=denied,
2323                                           computer_allowed_to=allowed,
2324                                           service_allowed_to=denied)
2325
2326         # Create a computer account with the assigned policy.
2327         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2328                                        assigned_policy=policy)
2329
2330         # Show that obtaining a service ticket is allowed without an armor TGT.
2331         self._tgs_req(tgt, 0, client_creds, target_creds)
2332
2333     def test_authn_policy_denied_no_fast(self):
2334         samdb = self.get_samdb()
2335
2336         # Create a user account.
2337         client_creds = self.get_cached_creds(
2338             account_type=self.AccountType.USER)
2339         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2340         tgt = self.get_tgt(client_creds)
2341
2342         # Create an authentication policy that applies to a computer and
2343         # explicitly disallows the user account to obtain a service ticket.
2344         denied = f'O:SYD:(D;;CR;;;{client_sid})'
2345         allowed = 'O:SYD:(A;;CR;;;WD)'
2346         policy_id = self.get_new_username()
2347         policy = self.create_authn_policy(policy_id,
2348                                           enforced=True,
2349                                           user_allowed_to=allowed,
2350                                           computer_allowed_to=denied,
2351                                           service_allowed_to=allowed)
2352
2353         # Create a computer account with the assigned policy.
2354         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2355                                        assigned_policy=policy)
2356
2357         # Show that obtaining a service ticket is not allowed.
2358         self._tgs_req(
2359             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2360             expect_edata=self.expect_padata_outer,
2361             expect_status=True,
2362             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
2363
2364     def test_authn_policy_allowed_to_computer_allow_asserted_identity(self):
2365         # Create a machine account with which to perform FAST.
2366         mach_creds = self.get_cached_creds(
2367             account_type=self.AccountType.COMPUTER)
2368         mach_tgt = self.get_tgt(mach_creds)
2369
2370         # Create a user account.
2371         client_creds = self.get_cached_creds(
2372             account_type=self.AccountType.USER)
2373         tgt = self.get_tgt(client_creds)
2374
2375         # Create an authentication policy that allows accounts with the
2376         # Authentication Authority Asserted Identity SID to obtain a service
2377         # ticket.
2378         allowed = (
2379             f'O:SYD:(A;;CR;;;'
2380             f'{security.SID_AUTHENTICATION_AUTHORITY_ASSERTED_IDENTITY})'
2381         )
2382         denied = 'O:SYD:(D;;CR;;;WD)'
2383         policy_id = self.get_new_username()
2384         policy = self.create_authn_policy(policy_id,
2385                                           enforced=True,
2386                                           user_allowed_to=denied,
2387                                           computer_allowed_to=allowed,
2388                                           service_allowed_to=denied)
2389
2390         # Create a computer account with the assigned policy.
2391         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2392                                        assigned_policy=policy)
2393
2394         # Show that obtaining a service ticket is allowed.
2395         self._tgs_req(tgt, 0, client_creds, target_creds,
2396                       armor_tgt=mach_tgt)
2397
2398     def test_authn_policy_allowed_to_computer_allow_claims_valid(self):
2399         # Create a machine account with which to perform FAST.
2400         mach_creds = self.get_cached_creds(
2401             account_type=self.AccountType.COMPUTER)
2402         mach_tgt = self.get_tgt(mach_creds)
2403
2404         # Create a user account.
2405         client_creds = self.get_cached_creds(
2406             account_type=self.AccountType.USER)
2407         tgt = self.get_tgt(client_creds)
2408
2409         # Create an authentication policy that allows accounts with the Claims
2410         # Valid SID to obtain a service ticket.
2411         allowed = f'O:SYD:(A;;CR;;;{security.SID_CLAIMS_VALID})'
2412         denied = 'O:SYD:(D;;CR;;;WD)'
2413         policy_id = self.get_new_username()
2414         policy = self.create_authn_policy(policy_id,
2415                                           enforced=True,
2416                                           user_allowed_to=denied,
2417                                           computer_allowed_to=allowed,
2418                                           service_allowed_to=denied)
2419
2420         # Create a computer account with the assigned policy.
2421         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2422                                        assigned_policy=policy)
2423
2424         # Show that obtaining a service ticket is allowed.
2425         self._tgs_req(tgt, 0, client_creds, target_creds,
2426                       armor_tgt=mach_tgt)
2427
2428     def test_authn_policy_allowed_to_computer_allow_compounded_auth(self):
2429         # Create a machine account with which to perform FAST.
2430         mach_creds = self.get_cached_creds(
2431             account_type=self.AccountType.COMPUTER)
2432         mach_tgt = self.get_tgt(mach_creds)
2433
2434         # Create a user account.
2435         client_creds = self.get_cached_creds(
2436             account_type=self.AccountType.USER)
2437         tgt = self.get_tgt(client_creds)
2438
2439         # Create an authentication policy that allows accounts with the
2440         # Compounded Authentication SID to obtain a service ticket.
2441         allowed = f'O:SYD:(A;;CR;;;{security.SID_COMPOUNDED_AUTHENTICATION})'
2442         denied = 'O:SYD:(D;;CR;;;WD)'
2443         policy_id = self.get_new_username()
2444         policy = self.create_authn_policy(policy_id,
2445                                           enforced=True,
2446                                           user_allowed_to=denied,
2447                                           computer_allowed_to=allowed,
2448                                           service_allowed_to=denied)
2449
2450         # Create a computer account with the assigned policy.
2451         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2452                                        assigned_policy=policy)
2453
2454         # Show that obtaining a service ticket is denied.
2455         self._tgs_req(
2456             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2457             armor_tgt=mach_tgt,
2458             expect_edata=self.expect_padata_outer,
2459             # We aren’t particular about whether or not we get an NTSTATUS.
2460             expect_status=None,
2461             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
2462             check_patypes=False)
2463
2464     def test_authn_policy_allowed_to_computer_allow_authenticated_users(self):
2465         # Create a machine account with which to perform FAST.
2466         mach_creds = self.get_cached_creds(
2467             account_type=self.AccountType.COMPUTER)
2468         mach_tgt = self.get_tgt(mach_creds)
2469
2470         # Create a user account.
2471         client_creds = self.get_cached_creds(
2472             account_type=self.AccountType.USER)
2473         tgt = self.get_tgt(client_creds)
2474
2475         # Create an authentication policy that allows accounts with the
2476         # Authenticated Users SID to obtain a service ticket.
2477         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_AUTHENTICATED_USERS})'
2478         denied = 'O:SYD:(D;;CR;;;WD)'
2479         policy_id = self.get_new_username()
2480         policy = self.create_authn_policy(policy_id,
2481                                           enforced=True,
2482                                           user_allowed_to=denied,
2483                                           computer_allowed_to=allowed,
2484                                           service_allowed_to=denied)
2485
2486         # Create a computer account with the assigned policy.
2487         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2488                                        assigned_policy=policy)
2489
2490         # Show that obtaining a service ticket is allowed.
2491         self._tgs_req(tgt, 0, client_creds, target_creds,
2492                       armor_tgt=mach_tgt)
2493
2494     def test_authn_policy_allowed_to_computer_allow_ntlm_authn(self):
2495         # Create a machine account with which to perform FAST.
2496         mach_creds = self.get_cached_creds(
2497             account_type=self.AccountType.COMPUTER)
2498         mach_tgt = self.get_tgt(mach_creds)
2499
2500         # Create a user account.
2501         client_creds = self.get_cached_creds(
2502             account_type=self.AccountType.USER)
2503         tgt = self.get_tgt(client_creds)
2504
2505         # Create an authentication policy that allows accounts with the NTLM
2506         # Authentication SID to obtain a service ticket.
2507         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_NTLM_AUTHENTICATION})'
2508         denied = 'O:SYD:(D;;CR;;;WD)'
2509         policy_id = self.get_new_username()
2510         policy = self.create_authn_policy(policy_id,
2511                                           enforced=True,
2512                                           user_allowed_to=denied,
2513                                           computer_allowed_to=allowed,
2514                                           service_allowed_to=denied)
2515
2516         # Create a computer account with the assigned policy.
2517         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2518                                        assigned_policy=policy)
2519
2520         # Show that obtaining a service ticket is denied.
2521         self._tgs_req(
2522             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2523             armor_tgt=mach_tgt,
2524             expect_edata=self.expect_padata_outer,
2525             # We aren’t particular about whether or not we get an NTSTATUS.
2526             expect_status=None,
2527             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
2528             check_patypes=False)
2529
2530     def test_authn_policy_allowed_to_no_owner(self):
2531         samdb = self.get_samdb()
2532
2533         # Create a machine account with which to perform FAST.
2534         mach_creds = self.get_cached_creds(
2535             account_type=self.AccountType.COMPUTER)
2536         mach_tgt = self.get_tgt(mach_creds)
2537
2538         # Create a user account.
2539         client_creds = self.get_cached_creds(
2540             account_type=self.AccountType.USER)
2541         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2542         tgt = self.get_tgt(client_creds)
2543
2544         # Create an authentication policy that applies to a computer and
2545         # explicitly allows the user account to obtain a service ticket. Omit
2546         # the owner (O:SY) from the SDDL.
2547         allowed = f'D:(A;;CR;;;{client_sid})'
2548         policy_id = self.get_new_username()
2549         policy = self.create_authn_policy(policy_id,
2550                                           enforced=True,
2551                                           computer_allowed_to=allowed)
2552
2553         # Create a computer account with the assigned policy.
2554         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2555                                        assigned_policy=policy)
2556
2557         # Show that obtaining a service ticket is denied.
2558         self._tgs_req(tgt, KDC_ERR_POLICY, client_creds, target_creds,
2559                       armor_tgt=mach_tgt,
2560                       expect_edata=self.expect_padata_outer,
2561                       # We aren’t particular about whether or not we get an NTSTATUS.
2562                       expect_status=None,
2563                       expected_status=ntstatus.NT_STATUS_INVALID_PARAMETER,
2564                       check_patypes=False)
2565
2566     def test_authn_policy_allowed_to_no_owner_unenforced(self):
2567         samdb = self.get_samdb()
2568
2569         # Create a machine account with which to perform FAST.
2570         mach_creds = self.get_cached_creds(
2571             account_type=self.AccountType.COMPUTER)
2572         mach_tgt = self.get_tgt(mach_creds)
2573
2574         # Create a user account.
2575         client_creds = self.get_cached_creds(
2576             account_type=self.AccountType.USER)
2577         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2578         tgt = self.get_tgt(client_creds)
2579
2580         # Create an unenforced authentication policy that applies to a computer
2581         # and explicitly allows the user account to obtain a service
2582         # ticket. Omit the owner (O:SY) from the SDDL.
2583         allowed = f'D:(A;;CR;;;{client_sid})'
2584         policy_id = self.get_new_username()
2585         policy = self.create_authn_policy(policy_id,
2586                                           enforced=False,
2587                                           computer_allowed_to=allowed)
2588
2589         # Create a computer account with the assigned policy.
2590         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2591                                        assigned_policy=policy)
2592
2593         # Show that obtaining a service ticket is allowed.
2594         self._tgs_req(tgt, 0, client_creds, target_creds,
2595                       armor_tgt=mach_tgt)
2596
2597     def test_authn_policy_allowed_to_owner_self(self):
2598         samdb = self.get_samdb()
2599
2600         # Create a machine account with which to perform FAST.
2601         mach_creds = self.get_cached_creds(
2602             account_type=self.AccountType.COMPUTER)
2603         mach_tgt = self.get_tgt(mach_creds)
2604
2605         # Create a user account.
2606         client_creds = self.get_cached_creds(
2607             account_type=self.AccountType.USER)
2608         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2609         tgt = self.get_tgt(client_creds)
2610
2611         # Create an authentication policy that applies to a computer and
2612         # explicitly allows the user account to obtain a service ticket. Set
2613         # the owner to the user account.
2614         allowed = f'O:{client_sid}D:(A;;CR;;;{client_sid})'
2615         policy_id = self.get_new_username()
2616         policy = self.create_authn_policy(policy_id,
2617                                           enforced=True,
2618                                           computer_allowed_to=allowed)
2619
2620         # Create a computer account with the assigned policy.
2621         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2622                                        assigned_policy=policy)
2623
2624         # Show that obtaining a service ticket is allowed.
2625         self._tgs_req(tgt, 0, client_creds, target_creds,
2626                       armor_tgt=mach_tgt)
2627
2628     def test_authn_policy_allowed_to_owner_anon(self):
2629         samdb = self.get_samdb()
2630
2631         # Create a machine account with which to perform FAST.
2632         mach_creds = self.get_cached_creds(
2633             account_type=self.AccountType.COMPUTER)
2634         mach_tgt = self.get_tgt(mach_creds)
2635
2636         # Create a user account.
2637         client_creds = self.get_cached_creds(
2638             account_type=self.AccountType.USER)
2639         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2640         tgt = self.get_tgt(client_creds)
2641
2642         # Create an authentication policy that applies to a computer and
2643         # explicitly allows the user account to obtain a service ticket. Set
2644         # the owner to be anonymous.
2645         allowed = f'O:AND:(A;;CR;;;{client_sid})'
2646         policy_id = self.get_new_username()
2647         policy = self.create_authn_policy(policy_id,
2648                                           enforced=True,
2649                                           computer_allowed_to=allowed)
2650
2651         # Create a computer account with the assigned policy.
2652         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2653                                        assigned_policy=policy)
2654
2655         # Show that obtaining a service ticket is allowed.
2656         self._tgs_req(tgt, 0, client_creds, target_creds,
2657                       armor_tgt=mach_tgt)
2658
2659     def test_authn_policy_allowed_to_user_allow(self):
2660         samdb = self.get_samdb()
2661
2662         # Create a machine account with which to perform FAST.
2663         mach_creds = self.get_cached_creds(
2664             account_type=self.AccountType.COMPUTER)
2665         mach_tgt = self.get_tgt(mach_creds)
2666
2667         # Create a user account.
2668         client_creds = self.get_cached_creds(
2669             account_type=self.AccountType.USER)
2670         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2671         tgt = self.get_tgt(client_creds)
2672
2673         # Create an authentication policy that applies to a user and explicitly
2674         # allows the user account to obtain a service ticket.
2675         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
2676         denied = 'O:SYD:(D;;CR;;;WD)'
2677         policy_id = self.get_new_username()
2678         policy = self.create_authn_policy(policy_id,
2679                                           enforced=True,
2680                                           user_allowed_to=allowed,
2681                                           computer_allowed_to=denied,
2682                                           service_allowed_to=denied)
2683
2684         # Create a user account with the assigned policy.
2685         target_creds = self._get_creds(account_type=self.AccountType.USER,
2686                                        assigned_policy=policy,
2687                                        spn='host/{account}')
2688
2689         # Show that obtaining a service ticket is allowed.
2690         self._tgs_req(tgt, 0, client_creds, target_creds,
2691                       armor_tgt=mach_tgt)
2692
2693     def test_authn_policy_allowed_to_user_deny(self):
2694         samdb = self.get_samdb()
2695
2696         # Create a machine account with which to perform FAST.
2697         mach_creds = self.get_cached_creds(
2698             account_type=self.AccountType.COMPUTER)
2699         mach_tgt = self.get_tgt(mach_creds)
2700
2701         # Create a user account.
2702         client_creds = self.get_cached_creds(
2703             account_type=self.AccountType.USER)
2704         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2705         tgt = self.get_tgt(client_creds)
2706
2707         # Create an authentication policy that applies to a user and
2708         # explicitly denies the user account to obtain a service ticket.
2709         denied = f'O:SYD:(D;;CR;;;{client_sid})'
2710         allowed = 'O:SYD:(A;;CR;;;WD)'
2711         policy_id = self.get_new_username()
2712         policy = self.create_authn_policy(policy_id,
2713                                           enforced=True,
2714                                           user_allowed_to=denied,
2715                                           computer_allowed_to=allowed,
2716                                           service_allowed_to=allowed)
2717
2718         # Create a user account with the assigned policy.
2719         target_creds = self._get_creds(account_type=self.AccountType.USER,
2720                                        assigned_policy=policy,
2721                                        spn='host/{account}')
2722
2723         # Show that obtaining a service ticket is denied.
2724         self._tgs_req(
2725             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2726             armor_tgt=mach_tgt,
2727             expect_edata=self.expect_padata_outer,
2728             # We aren’t particular about whether or not we get an NTSTATUS.
2729             expect_status=None,
2730             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
2731             check_patypes=False)
2732
2733     def test_authn_policy_allowed_to_service_allow(self):
2734         samdb = self.get_samdb()
2735
2736         # Create a machine account with which to perform FAST.
2737         mach_creds = self.get_cached_creds(
2738             account_type=self.AccountType.COMPUTER)
2739         mach_tgt = self.get_tgt(mach_creds)
2740
2741         # Create a user account.
2742         client_creds = self.get_cached_creds(
2743             account_type=self.AccountType.USER)
2744         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2745         tgt = self.get_tgt(client_creds)
2746
2747         # Create an authentication policy that applies to a managed service and
2748         # explicitly allows the user account to obtain a service ticket.
2749         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
2750         denied = 'O:SYD:(D;;CR;;;WD)'
2751         policy_id = self.get_new_username()
2752         policy = self.create_authn_policy(policy_id,
2753                                           enforced=True,
2754                                           user_allowed_to=denied,
2755                                           computer_allowed_to=denied,
2756                                           service_allowed_to=allowed)
2757
2758         # Create a managed service account with the assigned policy.
2759         target_creds = self._get_creds(
2760             account_type=self.AccountType.MANAGED_SERVICE,
2761             assigned_policy=policy)
2762
2763         # Show that obtaining a service ticket is allowed.
2764         self._tgs_req(tgt, 0, client_creds, target_creds,
2765                       armor_tgt=mach_tgt)
2766
2767     def test_authn_policy_allowed_to_service_deny(self):
2768         samdb = self.get_samdb()
2769
2770         # Create a machine account with which to perform FAST.
2771         mach_creds = self.get_cached_creds(
2772             account_type=self.AccountType.COMPUTER)
2773         mach_tgt = self.get_tgt(mach_creds)
2774
2775         # Create a user account.
2776         client_creds = self.get_cached_creds(
2777             account_type=self.AccountType.USER)
2778         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2779         tgt = self.get_tgt(client_creds)
2780
2781         # Create an authentication policy that applies to a managed service and
2782         # explicitly denies the user account to obtain a service ticket.
2783         denied = f'O:SYD:(D;;CR;;;{client_sid})'
2784         allowed = 'O:SYD:(A;;CR;;;WD)'
2785         policy_id = self.get_new_username()
2786         policy = self.create_authn_policy(policy_id,
2787                                           enforced=True,
2788                                           user_allowed_to=allowed,
2789                                           computer_allowed_to=allowed,
2790                                           service_allowed_to=denied)
2791
2792         # Create a managed service account with the assigned policy.
2793         target_creds = self._get_creds(
2794             account_type=self.AccountType.MANAGED_SERVICE,
2795             assigned_policy=policy)
2796
2797         # Show that obtaining a service ticket is denied.
2798         self._tgs_req(
2799             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2800             armor_tgt=mach_tgt,
2801             expect_edata=self.expect_padata_outer,
2802             # We aren’t particular about whether or not we get an NTSTATUS.
2803             expect_status=None,
2804             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
2805             check_patypes=False)
2806
2807     def test_authn_policy_allowed_to_user_allow_from_rodc(self):
2808         samdb = self.get_samdb()
2809
2810         # Create a machine account with which to perform FAST.
2811         mach_creds = self.get_cached_creds(
2812             account_type=self.AccountType.COMPUTER)
2813         mach_tgt = self.get_tgt(mach_creds)
2814
2815         # Create a user account.
2816         client_creds = self._get_creds(account_type=self.AccountType.USER,
2817                                        allowed_rodc=True)
2818         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2819         # Modify the TGT to be issued by an RODC.
2820         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
2821
2822         # Create an authentication policy that applies to a user and explicitly
2823         # allows the user account to obtain a service ticket.
2824         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
2825         denied = 'O:SYD:(D;;CR;;;WD)'
2826         policy_id = self.get_new_username()
2827         policy = self.create_authn_policy(policy_id,
2828                                           enforced=True,
2829                                           user_allowed_to=allowed,
2830                                           computer_allowed_to=denied,
2831                                           service_allowed_to=denied)
2832
2833         # Create a user account with the assigned policy.
2834         target_creds = self._get_creds(account_type=self.AccountType.USER,
2835                                        assigned_policy=policy,
2836                                        spn='host/{account}')
2837
2838         # Show that obtaining a service ticket is allowed.
2839         self._tgs_req(tgt, 0, client_creds, target_creds,
2840                       armor_tgt=mach_tgt)
2841
2842     def test_authn_policy_allowed_to_user_deny_from_rodc(self):
2843         samdb = self.get_samdb()
2844
2845         # Create a machine account with which to perform FAST.
2846         mach_creds = self.get_cached_creds(
2847             account_type=self.AccountType.COMPUTER)
2848         mach_tgt = self.get_tgt(mach_creds)
2849
2850         # Create a user account.
2851         client_creds = self._get_creds(account_type=self.AccountType.USER,
2852                                        allowed_rodc=True)
2853         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2854         # Modify the TGT to be issued by an RODC.
2855         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
2856
2857         # Create an authentication policy that applies to a user and
2858         # explicitly denies the user account to obtain a service ticket.
2859         denied = f'O:SYD:(D;;CR;;;{client_sid})'
2860         allowed = 'O:SYD:(A;;CR;;;WD)'
2861         policy_id = self.get_new_username()
2862         policy = self.create_authn_policy(policy_id,
2863                                           enforced=True,
2864                                           user_allowed_to=denied,
2865                                           computer_allowed_to=allowed,
2866                                           service_allowed_to=allowed)
2867
2868         # Create a user account with the assigned policy.
2869         target_creds = self._get_creds(account_type=self.AccountType.USER,
2870                                        assigned_policy=policy,
2871                                        spn='host/{account}')
2872
2873         # Show that obtaining a service ticket is denied.
2874         self._tgs_req(
2875             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2876             armor_tgt=mach_tgt,
2877             expect_edata=self.expect_padata_outer,
2878             # We aren’t particular about whether or not we get an NTSTATUS.
2879             expect_status=None,
2880             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
2881
2882     def test_authn_policy_allowed_to_computer_allow_from_rodc(self):
2883         samdb = self.get_samdb()
2884
2885         # Create a machine account with which to perform FAST.
2886         mach_creds = self.get_cached_creds(
2887             account_type=self.AccountType.COMPUTER)
2888         mach_tgt = self.get_tgt(mach_creds)
2889
2890         # Create a user account.
2891         client_creds = self._get_creds(account_type=self.AccountType.USER,
2892                                        allowed_rodc=True)
2893         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2894         # Modify the TGT to be issued by an RODC.
2895         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
2896
2897         # Create an authentication policy that applies to a computer and
2898         # explicitly allows the user account to obtain a service ticket.
2899         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
2900         denied = 'O:SYD:(D;;CR;;;WD)'
2901         policy_id = self.get_new_username()
2902         policy = self.create_authn_policy(policy_id,
2903                                           enforced=True,
2904                                           user_allowed_to=denied,
2905                                           computer_allowed_to=allowed,
2906                                           service_allowed_to=denied)
2907
2908         # Create a computer account with the assigned policy.
2909         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2910                                        assigned_policy=policy)
2911
2912         # Show that obtaining a service ticket is allowed.
2913         self._tgs_req(tgt, 0, client_creds, target_creds,
2914                       armor_tgt=mach_tgt)
2915
2916     def test_authn_policy_allowed_to_computer_deny_from_rodc(self):
2917         samdb = self.get_samdb()
2918
2919         # Create a machine account with which to perform FAST.
2920         mach_creds = self.get_cached_creds(
2921             account_type=self.AccountType.COMPUTER)
2922         mach_tgt = self.get_tgt(mach_creds)
2923
2924         # Create a user account.
2925         client_creds = self._get_creds(account_type=self.AccountType.USER,
2926                                        allowed_rodc=True)
2927         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2928         # Modify the TGT to be issued by an RODC.
2929         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
2930
2931         # Create an authentication policy that applies to a computer and
2932         # explicitly denies the user account to obtain a service ticket.
2933         denied = f'O:SYD:(D;;CR;;;{client_sid})'
2934         allowed = 'O:SYD:(A;;CR;;;WD)'
2935         policy_id = self.get_new_username()
2936         policy = self.create_authn_policy(policy_id,
2937                                           enforced=True,
2938                                           user_allowed_to=allowed,
2939                                           computer_allowed_to=denied,
2940                                           service_allowed_to=allowed)
2941
2942         # Create a computer account with the assigned policy.
2943         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
2944                                        assigned_policy=policy)
2945
2946         # Show that obtaining a service ticket is denied.
2947         self._tgs_req(
2948             tgt, KDC_ERR_POLICY, client_creds, target_creds,
2949             armor_tgt=mach_tgt,
2950             expect_edata=self.expect_padata_outer,
2951             # We aren’t particular about whether or not we get an NTSTATUS.
2952             expect_status=None,
2953             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,)
2954
2955     def test_authn_policy_allowed_to_service_allow_from_rodc(self):
2956         samdb = self.get_samdb()
2957
2958         # Create a machine account with which to perform FAST.
2959         mach_creds = self.get_cached_creds(
2960             account_type=self.AccountType.COMPUTER)
2961         mach_tgt = self.get_tgt(mach_creds)
2962
2963         # Create a user account.
2964         client_creds = self._get_creds(account_type=self.AccountType.USER,
2965                                        allowed_rodc=True)
2966         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
2967         # Modify the TGT to be issued by an RODC.
2968         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
2969
2970         # Create an authentication policy that applies to a managed service and
2971         # explicitly allows the user account to obtain a service ticket.
2972         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
2973         denied = 'O:SYD:(D;;CR;;;WD)'
2974         policy_id = self.get_new_username()
2975         policy = self.create_authn_policy(policy_id,
2976                                           enforced=True,
2977                                           user_allowed_to=denied,
2978                                           computer_allowed_to=denied,
2979                                           service_allowed_to=allowed)
2980
2981         # Create a managed service account with the assigned policy.
2982         target_creds = self._get_creds(
2983             account_type=self.AccountType.MANAGED_SERVICE,
2984             assigned_policy=policy)
2985
2986         # Show that obtaining a service ticket is allowed.
2987         self._tgs_req(tgt, 0, client_creds, target_creds,
2988                       armor_tgt=mach_tgt)
2989
2990     def test_authn_policy_allowed_to_service_deny_from_rodc(self):
2991         samdb = self.get_samdb()
2992
2993         # Create a machine account with which to perform FAST.
2994         mach_creds = self.get_cached_creds(
2995             account_type=self.AccountType.COMPUTER)
2996         mach_tgt = self.get_tgt(mach_creds)
2997
2998         # Create a user account.
2999         client_creds = self._get_creds(account_type=self.AccountType.USER,
3000                                        allowed_rodc=True)
3001         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
3002         # Modify the TGT to be issued by an RODC.
3003         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3004
3005         # Create an authentication policy that applies to a managed service and
3006         # explicitly denies the user account to obtain a service ticket.
3007         denied = f'O:SYD:(D;;CR;;;{client_sid})'
3008         allowed = 'O:SYD:(A;;CR;;;WD)'
3009         policy_id = self.get_new_username()
3010         policy = self.create_authn_policy(policy_id,
3011                                           enforced=True,
3012                                           user_allowed_to=allowed,
3013                                           computer_allowed_to=allowed,
3014                                           service_allowed_to=denied)
3015
3016         # Create a managed service account with the assigned policy.
3017         target_creds = self._get_creds(
3018             account_type=self.AccountType.MANAGED_SERVICE,
3019             assigned_policy=policy)
3020
3021         # Show that obtaining a service ticket is denied.
3022         self._tgs_req(
3023             tgt, KDC_ERR_POLICY, client_creds, target_creds,
3024             armor_tgt=mach_tgt,
3025             expect_edata=self.expect_padata_outer,
3026             # We aren’t particular about whether or not we get an NTSTATUS.
3027             expect_status=None,
3028             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
3029
3030     def test_authn_policy_allowed_to_user_allow_group_not_a_member(self):
3031         samdb = self.get_samdb()
3032
3033         # Create a machine account with which to perform FAST.
3034         mach_creds = self.get_cached_creds(
3035             account_type=self.AccountType.COMPUTER)
3036         mach_tgt = self.get_tgt(mach_creds)
3037
3038         # Create a new group.
3039         group_name = self.get_new_username()
3040         group_dn = self.create_group(samdb, group_name)
3041         group_sid = self.get_objectSid(samdb, group_dn)
3042
3043         # Create a user account which does not belong to the group.
3044         client_creds = self.get_cached_creds(
3045             account_type=self.AccountType.USER)
3046         tgt = self.get_tgt(client_creds)
3047
3048         # Create an authentication policy that allows accounts belonging to the
3049         # group.
3050         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
3051         policy_id = self.get_new_username()
3052         policy = self.create_authn_policy(policy_id,
3053                                           enforced=True,
3054                                           user_allowed_to=allowed)
3055
3056         # Create a user account with the assigned policy.
3057         target_creds = self._get_creds(account_type=self.AccountType.USER,
3058                                        assigned_policy=policy,
3059                                        spn='host/{account}')
3060
3061         # Show that we get a policy error, as the user account does not belong
3062         # to the group.
3063         self._tgs_req(
3064             tgt, KDC_ERR_POLICY, client_creds, target_creds,
3065             armor_tgt=mach_tgt,
3066             expect_edata=self.expect_padata_outer,
3067             # We aren’t particular about whether or not we get an NTSTATUS.
3068             expect_status=None,
3069             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
3070             check_patypes=False)
3071
3072     def test_authn_policy_allowed_to_user_allow_group_member(self):
3073         samdb = self.get_samdb()
3074
3075         # Create a machine account with which to perform FAST.
3076         mach_creds = self.get_cached_creds(
3077             account_type=self.AccountType.COMPUTER)
3078         mach_tgt = self.get_tgt(mach_creds)
3079
3080         # Create a new group.
3081         group_name = self.get_new_username()
3082         group_dn = self.create_group(samdb, group_name)
3083         group_sid = self.get_objectSid(samdb, group_dn)
3084
3085         # Create a user account that belongs to the group.
3086         client_creds = self.get_cached_creds(
3087             account_type=self.AccountType.USER,
3088             opts={'member_of': (group_dn,)})
3089         tgt = self.get_tgt(client_creds)
3090
3091         # Create an authentication policy that allows accounts belonging to the
3092         # group.
3093         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
3094         policy_id = self.get_new_username()
3095         policy = self.create_authn_policy(policy_id,
3096                                           enforced=True,
3097                                           user_allowed_to=allowed)
3098
3099         # Create a user account with the assigned policy.
3100         target_creds = self._get_creds(account_type=self.AccountType.USER,
3101                                        assigned_policy=policy,
3102                                        spn='host/{account}')
3103
3104         # Show that we can get a service ticket, since the user account belongs
3105         # to the group.
3106         self._tgs_req(tgt, 0, client_creds, target_creds,
3107                       armor_tgt=mach_tgt)
3108
3109     def test_authn_policy_allowed_to_user_allow_domain_local_group(self):
3110         samdb = self.get_samdb()
3111
3112         # Create a machine account with which to perform FAST.
3113         mach_creds = self.get_cached_creds(
3114             account_type=self.AccountType.COMPUTER)
3115         mach_tgt = self.get_tgt(mach_creds)
3116
3117         # Create a new domain-local group.
3118         group_name = self.get_new_username()
3119         group_dn = self.create_group(samdb, group_name,
3120                                      gtype=GroupType.DOMAIN_LOCAL.value)
3121         group_sid = self.get_objectSid(samdb, group_dn)
3122
3123         # Create a user account that belongs to the group.
3124         client_creds = self.get_cached_creds(
3125             account_type=self.AccountType.USER,
3126             opts={'member_of': (group_dn,)})
3127         tgt = self.get_tgt(client_creds)
3128
3129         # Create an authentication policy that allows accounts belonging to the
3130         # group.
3131         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
3132         policy_id = self.get_new_username()
3133         policy = self.create_authn_policy(policy_id,
3134                                           enforced=True,
3135                                           user_allowed_to=allowed)
3136
3137         # Create a user account with the assigned policy.
3138         target_creds = self._get_creds(account_type=self.AccountType.USER,
3139                                        assigned_policy=policy,
3140                                        spn='host/{account}')
3141
3142         # Show that the groups in the TGT are expanded to include the
3143         # domain-local group.
3144         self._tgs_req(tgt, 0, client_creds, target_creds,
3145                       armor_tgt=mach_tgt)
3146
3147     def test_authn_policy_allowed_to_computer_allow_asserted_identity_from_rodc(self):
3148         # Create a machine account with which to perform FAST.
3149         mach_creds = self.get_cached_creds(
3150             account_type=self.AccountType.COMPUTER)
3151         mach_tgt = self.get_tgt(mach_creds)
3152
3153         # Create a user account.
3154         client_creds = self._get_creds(account_type=self.AccountType.USER,
3155                                        allowed_rodc=True)
3156         # Modify the TGT to be issued by an RODC.
3157         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3158
3159         # Create an authentication policy that allows accounts with the
3160         # Authentication Authority Asserted Identity SID to obtain a service
3161         # ticket.
3162         allowed = (
3163             f'O:SYD:(A;;CR;;;'
3164             f'{security.SID_AUTHENTICATION_AUTHORITY_ASSERTED_IDENTITY})'
3165         )
3166         denied = 'O:SYD:(D;;CR;;;WD)'
3167         policy_id = self.get_new_username()
3168         policy = self.create_authn_policy(policy_id,
3169                                           enforced=True,
3170                                           user_allowed_to=denied,
3171                                           computer_allowed_to=allowed,
3172                                           service_allowed_to=denied)
3173
3174         # Create a computer account with the assigned policy.
3175         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
3176                                        assigned_policy=policy)
3177
3178         # Show that obtaining a service ticket is allowed.
3179         self._tgs_req(tgt, 0, client_creds, target_creds,
3180                       armor_tgt=mach_tgt)
3181
3182     def test_authn_policy_allowed_to_computer_allow_claims_valid_from_rodc(self):
3183         # Create a machine account with which to perform FAST.
3184         mach_creds = self.get_cached_creds(
3185             account_type=self.AccountType.COMPUTER)
3186         mach_tgt = self.get_tgt(mach_creds)
3187
3188         # Create a user account.
3189         client_creds = self._get_creds(account_type=self.AccountType.USER,
3190                                        allowed_rodc=True)
3191         # Modify the TGT to be issued by an RODC.
3192         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3193
3194         # Create an authentication policy that allows accounts with the Claims
3195         # Valid SID to obtain a service ticket.
3196         allowed = f'O:SYD:(A;;CR;;;{security.SID_CLAIMS_VALID})'
3197         denied = 'O:SYD:(D;;CR;;;WD)'
3198         policy_id = self.get_new_username()
3199         policy = self.create_authn_policy(policy_id,
3200                                           enforced=True,
3201                                           user_allowed_to=denied,
3202                                           computer_allowed_to=allowed,
3203                                           service_allowed_to=denied)
3204
3205         # Create a computer account with the assigned policy.
3206         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
3207                                        assigned_policy=policy)
3208
3209         # Show that obtaining a service ticket is not allowed.
3210         self._tgs_req(tgt, KDC_ERR_POLICY, client_creds, target_creds,
3211                       armor_tgt=mach_tgt)
3212
3213     def test_authn_policy_allowed_to_computer_allow_compounded_authn_from_rodc(self):
3214         # Create a machine account with which to perform FAST.
3215         mach_creds = self.get_cached_creds(
3216             account_type=self.AccountType.COMPUTER)
3217         mach_tgt = self.get_tgt(mach_creds)
3218
3219         # Create a user account.
3220         client_creds = self._get_creds(account_type=self.AccountType.USER,
3221                                        allowed_rodc=True)
3222         # Modify the TGT to be issued by an RODC.
3223         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3224
3225         # Create an authentication policy that allows accounts with the
3226         # Compounded Authentication SID to obtain a service ticket.
3227         allowed = f'O:SYD:(A;;CR;;;{security.SID_COMPOUNDED_AUTHENTICATION})'
3228         denied = 'O:SYD:(D;;CR;;;WD)'
3229         policy_id = self.get_new_username()
3230         policy = self.create_authn_policy(policy_id,
3231                                           enforced=True,
3232                                           user_allowed_to=denied,
3233                                           computer_allowed_to=allowed,
3234                                           service_allowed_to=denied)
3235
3236         # Create a computer account with the assigned policy.
3237         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
3238                                        assigned_policy=policy)
3239
3240         # Show that obtaining a service ticket is denied.
3241         self._tgs_req(
3242             tgt, KDC_ERR_POLICY, client_creds, target_creds,
3243             armor_tgt=mach_tgt,
3244             expect_edata=self.expect_padata_outer,
3245             # We aren’t particular about whether or not we get an NTSTATUS.
3246             expect_status=None,
3247             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
3248
3249     def test_authn_policy_allowed_to_computer_allow_authenticated_users_from_rodc(self):
3250         # Create a machine account with which to perform FAST.
3251         mach_creds = self.get_cached_creds(
3252             account_type=self.AccountType.COMPUTER)
3253         mach_tgt = self.get_tgt(mach_creds)
3254
3255         # Create a user account.
3256         client_creds = self._get_creds(account_type=self.AccountType.USER,
3257                                        allowed_rodc=True)
3258         # Modify the TGT to be issued by an RODC.
3259         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3260
3261         # Create an authentication policy that allows accounts with the
3262         # Authenticated Users SID to obtain a service ticket.
3263         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_AUTHENTICATED_USERS})'
3264         denied = 'O:SYD:(D;;CR;;;WD)'
3265         policy_id = self.get_new_username()
3266         policy = self.create_authn_policy(policy_id,
3267                                           enforced=True,
3268                                           user_allowed_to=denied,
3269                                           computer_allowed_to=allowed,
3270                                           service_allowed_to=denied)
3271
3272         # Create a computer account with the assigned policy.
3273         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
3274                                        assigned_policy=policy)
3275
3276         # Show that obtaining a service ticket is allowed.
3277         self._tgs_req(tgt, 0, client_creds, target_creds,
3278                       armor_tgt=mach_tgt)
3279
3280     def test_authn_policy_allowed_to_computer_allow_ntlm_authn_from_rodc(self):
3281         # Create a machine account with which to perform FAST.
3282         mach_creds = self.get_cached_creds(
3283             account_type=self.AccountType.COMPUTER)
3284         mach_tgt = self.get_tgt(mach_creds)
3285
3286         # Create a user account.
3287         client_creds = self._get_creds(account_type=self.AccountType.USER,
3288                                        allowed_rodc=True)
3289         # Modify the TGT to be issued by an RODC.
3290         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3291
3292         # Create an authentication policy that allows accounts with the NTLM
3293         # Authentication SID to obtain a service ticket.
3294         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_NTLM_AUTHENTICATION})'
3295         denied = 'O:SYD:(D;;CR;;;WD)'
3296         policy_id = self.get_new_username()
3297         policy = self.create_authn_policy(policy_id,
3298                                           enforced=True,
3299                                           user_allowed_to=denied,
3300                                           computer_allowed_to=allowed,
3301                                           service_allowed_to=denied)
3302
3303         # Create a computer account with the assigned policy.
3304         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
3305                                        assigned_policy=policy)
3306
3307         # Show that obtaining a service ticket is denied.
3308         self._tgs_req(
3309             tgt, KDC_ERR_POLICY, client_creds, target_creds,
3310             armor_tgt=mach_tgt,
3311             expect_edata=self.expect_padata_outer,
3312             # We aren’t particular about whether or not we get an NTSTATUS.
3313             expect_status=None,
3314             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
3315
3316     def test_authn_policy_allowed_to_user_allow_group_not_a_member_from_rodc(self):
3317         samdb = self.get_samdb()
3318
3319         # Create a machine account with which to perform FAST.
3320         mach_creds = self.get_cached_creds(
3321             account_type=self.AccountType.COMPUTER)
3322         mach_tgt = self.get_tgt(mach_creds)
3323
3324         # Create a new group.
3325         group_name = self.get_new_username()
3326         group_dn = self.create_group(samdb, group_name)
3327         group_sid = self.get_objectSid(samdb, group_dn)
3328
3329         # Create a user account which does not belong to the group.
3330         client_creds = self._get_creds(account_type=self.AccountType.USER,
3331                                        allowed_rodc=True)
3332         # Modify the TGT to be issued by an RODC.
3333         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3334
3335         # Create an authentication policy that allows accounts belonging to the
3336         # group.
3337         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
3338         policy_id = self.get_new_username()
3339         policy = self.create_authn_policy(policy_id,
3340                                           enforced=True,
3341                                           user_allowed_to=allowed)
3342
3343         # Create a user account with the assigned policy.
3344         target_creds = self._get_creds(account_type=self.AccountType.USER,
3345                                        assigned_policy=policy,
3346                                        spn='host/{account}')
3347
3348         # Show that we get a policy error, as the user account does not belong
3349         # to the group.
3350         self._tgs_req(
3351             tgt, KDC_ERR_POLICY, client_creds, target_creds,
3352             armor_tgt=mach_tgt,
3353             expect_edata=self.expect_padata_outer,
3354             # We aren’t particular about whether or not we get an NTSTATUS.
3355             expect_status=None,
3356             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
3357
3358     def test_authn_policy_allowed_to_user_allow_group_member_from_rodc(self):
3359         samdb = self.get_samdb()
3360
3361         # Create a machine account with which to perform FAST.
3362         mach_creds = self.get_cached_creds(
3363             account_type=self.AccountType.COMPUTER)
3364         mach_tgt = self.get_tgt(mach_creds)
3365
3366         # Create a new group.
3367         group_name = self.get_new_username()
3368         group_dn = self.create_group(samdb, group_name)
3369         group_sid = self.get_objectSid(samdb, group_dn)
3370
3371         # Create a user account that belongs to the group.
3372         client_creds = self.get_cached_creds(
3373             account_type=self.AccountType.USER,
3374             opts={'member_of': (group_dn,),
3375                   'allowed_replication_mock': True,
3376                   'revealed_to_mock_rodc': True})
3377         # Modify the TGT to be issued by an RODC.
3378         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3379
3380         # Create an authentication policy that allows accounts belonging to the
3381         # group.
3382         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
3383         policy_id = self.get_new_username()
3384         policy = self.create_authn_policy(policy_id,
3385                                           enforced=True,
3386                                           user_allowed_to=allowed)
3387
3388         # Create a user account with the assigned policy.
3389         target_creds = self._get_creds(account_type=self.AccountType.USER,
3390                                        assigned_policy=policy,
3391                                        spn='host/{account}')
3392
3393         # Show that we can get a service ticket, since the user account belongs
3394         # to the group.
3395         self._tgs_req(tgt, 0, client_creds, target_creds,
3396                       armor_tgt=mach_tgt)
3397
3398     def test_authn_policy_allowed_to_user_allow_domain_local_group_from_rodc(self):
3399         samdb = self.get_samdb()
3400
3401         # Create a machine account with which to perform FAST.
3402         mach_creds = self.get_cached_creds(
3403             account_type=self.AccountType.COMPUTER)
3404         mach_tgt = self.get_tgt(mach_creds)
3405
3406         # Create a new domain-local group.
3407         group_name = self.get_new_username()
3408         group_dn = self.create_group(samdb, group_name,
3409                                      gtype=GroupType.DOMAIN_LOCAL.value)
3410         group_sid = self.get_objectSid(samdb, group_dn)
3411
3412         # Create a user account that belongs to the group.
3413         client_creds = self.get_cached_creds(
3414             account_type=self.AccountType.USER,
3415             opts={'member_of': (group_dn,),
3416                   'allowed_replication_mock': True,
3417                   'revealed_to_mock_rodc': True})
3418         # Modify the TGT to be issued by an RODC.
3419         tgt = self.issued_by_rodc(self.get_tgt(client_creds))
3420
3421         # Create an authentication policy that allows accounts belonging to the
3422         # group.
3423         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
3424         policy_id = self.get_new_username()
3425         policy = self.create_authn_policy(policy_id,
3426                                           enforced=True,
3427                                           user_allowed_to=allowed)
3428
3429         # Create a user account with the assigned policy.
3430         target_creds = self._get_creds(account_type=self.AccountType.USER,
3431                                        assigned_policy=policy,
3432                                        spn='host/{account}')
3433
3434         # Show that the groups in the TGT are expanded to include the
3435         # domain-local group.
3436         self._tgs_req(tgt, 0, client_creds, target_creds,
3437                       armor_tgt=mach_tgt)
3438
3439     def test_authn_policy_allowed_to_computer_allow_to_self(self):
3440         samdb = self.get_samdb()
3441
3442         # Create a machine account with which to perform FAST.
3443         mach_creds = self.get_cached_creds(
3444             account_type=self.AccountType.COMPUTER)
3445         mach_tgt = self.get_tgt(mach_creds)
3446
3447         # Create a computer account.
3448         client_creds = self.get_cached_creds(
3449             account_type=self.AccountType.COMPUTER,
3450             opts={'id': 1},
3451             use_cache=False)
3452         client_dn = client_creds.get_dn()
3453         client_sid = self.get_objectSid(samdb, client_dn)
3454         tgt = self.get_tgt(client_creds)
3455
3456         # Create an authentication policy that applies to a computer and
3457         # explicitly allows the user account to obtain a service ticket.
3458         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
3459         denied = 'O:SYD:(D;;CR;;;WD)'
3460         policy_id = self.get_new_username()
3461         policy = self.create_authn_policy(policy_id,
3462                                           enforced=True,
3463                                           user_allowed_to=denied,
3464                                           computer_allowed_to=allowed,
3465                                           service_allowed_to=denied)
3466
3467         # Assign the policy to the account.
3468         self.add_attribute(samdb, str(client_dn),
3469                            'msDS-AssignedAuthNPolicy', str(policy))
3470
3471         # Show that obtaining a service ticket to ourselves is allowed.
3472         self._tgs_req(tgt, 0, client_creds, client_creds,
3473                       armor_tgt=mach_tgt)
3474
3475     def test_authn_policy_allowed_to_computer_deny_to_self(self):
3476         samdb = self.get_samdb()
3477
3478         # Create a machine account with which to perform FAST.
3479         mach_creds = self.get_cached_creds(
3480             account_type=self.AccountType.COMPUTER)
3481         mach_tgt = self.get_tgt(mach_creds)
3482
3483         # Create a computer account.
3484         client_creds = self.get_cached_creds(
3485             account_type=self.AccountType.COMPUTER,
3486             opts={'id': 1},
3487             use_cache=False)
3488         client_dn = client_creds.get_dn()
3489         client_sid = self.get_objectSid(samdb, client_dn)
3490         tgt = self.get_tgt(client_creds)
3491
3492         # Create an authentication policy that applies to a computer and
3493         # explicitly denies the user account to obtain a service ticket.
3494         denied = f'O:SYD:(D;;CR;;;{client_sid})'
3495         allowed = 'O:SYD:(A;;CR;;;WD)'
3496         policy_id = self.get_new_username()
3497         policy = self.create_authn_policy(policy_id,
3498                                           enforced=True,
3499                                           user_allowed_to=allowed,
3500                                           computer_allowed_to=denied,
3501                                           service_allowed_to=allowed)
3502
3503         # Assign the policy to the account.
3504         self.add_attribute(samdb, str(client_dn),
3505                            'msDS-AssignedAuthNPolicy', str(policy))
3506
3507         # Show that obtaining a service ticket to ourselves is allowed, despite
3508         # the policy disallowing it.
3509         self._tgs_req(tgt, 0, client_creds, client_creds,
3510                       armor_tgt=mach_tgt)
3511
3512     def test_authn_policy_allowed_to_computer_allow_to_self_with_self(self):
3513         samdb = self.get_samdb()
3514
3515         # Create a computer account.
3516         client_creds = self.get_cached_creds(
3517             account_type=self.AccountType.COMPUTER,
3518             use_cache=False)
3519         client_dn = client_creds.get_dn()
3520         client_sid = self.get_objectSid(samdb, client_dn)
3521         tgt = self.get_tgt(client_creds)
3522
3523         # Create an authentication policy that applies to a computer and
3524         # explicitly allows the account to obtain a service ticket.
3525         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
3526         denied = 'O:SYD:(D;;CR;;;WD)'
3527         policy_id = self.get_new_username()
3528         policy = self.create_authn_policy(policy_id,
3529                                           enforced=True,
3530                                           user_allowed_to=denied,
3531                                           computer_allowed_to=allowed,
3532                                           service_allowed_to=denied)
3533
3534         # Assign the policy to the account.
3535         self.add_attribute(samdb, str(client_dn),
3536                            'msDS-AssignedAuthNPolicy', str(policy))
3537
3538         # Show that obtaining a service ticket to ourselves armored with our
3539         # own TGT is allowed.
3540         self._tgs_req(tgt, 0, client_creds, client_creds,
3541                       armor_tgt=tgt)
3542
3543     def test_authn_policy_allowed_to_computer_deny_to_self_with_self(self):
3544         samdb = self.get_samdb()
3545
3546         # Create a computer account.
3547         client_creds = self.get_cached_creds(
3548             account_type=self.AccountType.COMPUTER,
3549             use_cache=False)
3550         client_dn = client_creds.get_dn()
3551         client_sid = self.get_objectSid(samdb, client_dn)
3552         tgt = self.get_tgt(client_creds)
3553
3554         # Create an authentication policy that applies to a computer and
3555         # explicitly denies the account to obtain a service ticket.
3556         denied = f'O:SYD:(D;;CR;;;{client_sid})'
3557         allowed = 'O:SYD:(A;;CR;;;WD)'
3558         policy_id = self.get_new_username()
3559         policy = self.create_authn_policy(policy_id,
3560                                           enforced=True,
3561                                           user_allowed_to=allowed,
3562                                           computer_allowed_to=denied,
3563                                           service_allowed_to=allowed)
3564
3565         # Assign the policy to the account.
3566         self.add_attribute(samdb, str(client_dn),
3567                            'msDS-AssignedAuthNPolicy', str(policy))
3568
3569         # Show that obtaining a service ticket to ourselves armored with our
3570         # own TGT is allowed, despite the policy’s disallowing it.
3571         self._tgs_req(tgt, 0, client_creds, client_creds,
3572                       armor_tgt=tgt)
3573
3574     def test_authn_policy_allowed_to_user_allow_s4u2self(self):
3575         samdb = self.get_samdb()
3576
3577         # Create a machine account with which to perform FAST.
3578         mach_creds = self.get_cached_creds(
3579             account_type=self.AccountType.COMPUTER)
3580         mach_tgt = self.get_tgt(mach_creds)
3581
3582         # Create a user account.
3583         client_creds = self.get_cached_creds(
3584             account_type=self.AccountType.USER)
3585         client_cname = self.PrincipalName_create(
3586             name_type=NT_PRINCIPAL,
3587             names=[client_creds.get_username()])
3588         client_realm = client_creds.get_realm()
3589         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
3590
3591         # Create an authentication policy that applies to a computer and
3592         # explicitly allows the user account to obtain a service ticket.
3593         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
3594         policy_id = self.get_new_username()
3595         policy = self.create_authn_policy(policy_id,
3596                                           enforced=True,
3597                                           computer_allowed_to=allowed)
3598
3599         # Create a computer account with the assigned policy.
3600         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
3601                                        assigned_policy=policy)
3602         target_tgt = self.get_tgt(target_creds)
3603
3604         def generate_s4u2self_padata(_kdc_exchange_dict,
3605                                      _callback_dict,
3606                                      req_body):
3607             padata = self.PA_S4U2Self_create(
3608                 name=client_cname,
3609                 realm=client_realm,
3610                 tgt_session_key=target_tgt.session_key,
3611                 ctype=None)
3612
3613             return [padata], req_body
3614
3615         # Show that obtaining a service ticket with S4U2Self is allowed.
3616         self._tgs_req(target_tgt, 0, target_creds, target_creds,
3617                       expected_cname=client_cname,
3618                       generate_fast_padata_fn=generate_s4u2self_padata,
3619                       armor_tgt=mach_tgt)
3620
3621     def test_authn_policy_allowed_to_user_deny_s4u2self(self):
3622         samdb = self.get_samdb()
3623
3624         # Create a machine account with which to perform FAST.
3625         mach_creds = self.get_cached_creds(
3626             account_type=self.AccountType.COMPUTER)
3627         mach_tgt = self.get_tgt(mach_creds)
3628
3629         # Create a user account.
3630         client_creds = self.get_cached_creds(
3631             account_type=self.AccountType.USER)
3632         client_cname = self.PrincipalName_create(
3633             name_type=NT_PRINCIPAL,
3634             names=[client_creds.get_username()])
3635         client_realm = client_creds.get_realm()
3636         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
3637
3638         # Create an authentication policy that applies to a computer and
3639         # explicitly denies the user account to obtain a service ticket.
3640         denied = f'O:SYD:(D;;CR;;;{client_sid})'
3641         policy_id = self.get_new_username()
3642         policy = self.create_authn_policy(policy_id,
3643                                           enforced=True,
3644                                           computer_allowed_to=denied)
3645
3646         # Create a computer account with the assigned policy.
3647         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
3648                                        assigned_policy=policy)
3649         target_tgt = self.get_tgt(target_creds)
3650
3651         def generate_s4u2self_padata(_kdc_exchange_dict,
3652                                      _callback_dict,
3653                                      req_body):
3654             padata = self.PA_S4U2Self_create(
3655                 name=client_cname,
3656                 realm=client_realm,
3657                 tgt_session_key=target_tgt.session_key,
3658                 ctype=None)
3659
3660             return [padata], req_body
3661
3662         # Show that obtaining a service ticket with S4U2Self is allowed,
3663         # despite the policy.
3664         self._tgs_req(target_tgt, 0, target_creds, target_creds,
3665                       expected_cname=client_cname,
3666                       generate_fast_padata_fn=generate_s4u2self_padata,
3667                       armor_tgt=mach_tgt)
3668
3669     def test_authn_policy_allowed_to_user_allow_constrained_delegation(self):
3670         samdb = self.get_samdb()
3671
3672         client_creds = self.get_cached_creds(
3673             account_type=self.AccountType.USER)
3674         client_dn = client_creds.get_dn()
3675         client_sid = self.get_objectSid(samdb, client_dn)
3676
3677         client_username = client_creds.get_username()
3678         client_cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
3679                                                  names=[client_username])
3680
3681         client_tkt_options = 'forwardable'
3682         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
3683
3684         client_tgt = self.get_tgt(client_creds,
3685                                   kdc_options=client_tkt_options,
3686                                   expected_flags=expected_flags)
3687
3688         # Create a machine account with which to perform FAST.
3689         mach_creds = self.get_cached_creds(
3690             account_type=self.AccountType.COMPUTER)
3691         mach_tgt = self.get_tgt(mach_creds)
3692
3693         # Create a target account.
3694         target_creds = self.get_cached_creds(
3695             account_type=self.AccountType.COMPUTER,
3696             opts={'id': 1},
3697             use_cache=False)
3698         target_spn = target_creds.get_spn()
3699
3700         service_creds = self.get_cached_creds(
3701             account_type=self.AccountType.COMPUTER,
3702             opts={
3703                 'delegation_to_spn': target_spn,
3704             })
3705         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
3706         service_tgt = self.get_tgt(service_creds)
3707
3708         # Create an authentication policy that applies to a computer and
3709         # explicitly allows the service account to obtain a service ticket,
3710         # while denying the user.
3711         allowed = f'O:SYD:(A;;CR;;;{service_sid})(D;;CR;;;{client_sid})'
3712         policy_id = self.get_new_username()
3713         policy = self.create_authn_policy(policy_id,
3714                                           enforced=True,
3715                                           computer_allowed_to=allowed)
3716
3717         # Assign the policy to the target account.
3718         self.add_attribute(samdb, str(target_creds.get_dn()),
3719                            'msDS-AssignedAuthNPolicy', str(policy))
3720
3721         client_service_tkt = self.get_service_ticket(
3722             client_tgt,
3723             service_creds,
3724             kdc_options=client_tkt_options,
3725             expected_flags=expected_flags)
3726
3727         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
3728
3729         target_decryption_key = self.TicketDecryptionKey_from_creds(
3730             target_creds)
3731         target_etypes = target_creds.tgs_supported_enctypes
3732
3733         service_name = service_creds.get_username()
3734         if service_name[-1] == '$':
3735             service_name = service_name[:-1]
3736         expected_transited_services = [
3737             f'host/{service_name}@{service_creds.get_realm()}'
3738         ]
3739
3740         # Show that obtaining a service ticket with constrained delegation is
3741         # allowed.
3742         self._tgs_req(service_tgt, 0, service_creds, target_creds,
3743                       armor_tgt=mach_tgt,
3744                       kdc_options=kdc_options,
3745                       expected_cname=client_cname,
3746                       expected_account_name=client_username,
3747                       additional_ticket=client_service_tkt,
3748                       decryption_key=target_decryption_key,
3749                       expected_sid=client_sid,
3750                       expected_supported_etypes=target_etypes,
3751                       expected_proxy_target=target_spn,
3752                       expected_transited_services=expected_transited_services)
3753
3754     def test_authn_policy_allowed_to_user_deny_constrained_delegation(self):
3755         samdb = self.get_samdb()
3756
3757         client_creds = self.get_cached_creds(
3758             account_type=self.AccountType.USER)
3759         client_dn = client_creds.get_dn()
3760         client_sid = self.get_objectSid(samdb, client_dn)
3761
3762         client_tkt_options = 'forwardable'
3763         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
3764
3765         client_tgt = self.get_tgt(client_creds,
3766                                   kdc_options=client_tkt_options,
3767                                   expected_flags=expected_flags)
3768
3769         # Create a machine account with which to perform FAST.
3770         mach_creds = self.get_cached_creds(
3771             account_type=self.AccountType.COMPUTER)
3772         mach_tgt = self.get_tgt(mach_creds)
3773
3774         # Create a target account.
3775         target_creds = self.get_cached_creds(
3776             account_type=self.AccountType.COMPUTER,
3777             opts={'id': 1},
3778             use_cache=False)
3779         target_spn = target_creds.get_spn()
3780
3781         service_creds = self.get_cached_creds(
3782             account_type=self.AccountType.COMPUTER,
3783             opts={
3784                 'delegation_to_spn': target_spn,
3785             })
3786         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
3787         service_tgt = self.get_tgt(service_creds)
3788
3789         # Create an authentication policy that applies to a computer and
3790         # explicitly denies the service account to obtain a service ticket,
3791         # while allowing the user.
3792         denied = f'O:SYD:(D;;CR;;;{service_sid})(A;;CR;;;{client_sid})'
3793         policy_id = self.get_new_username()
3794         policy = self.create_authn_policy(policy_id,
3795                                           enforced=True,
3796                                           computer_allowed_to=denied)
3797
3798         # Assign the policy to the target account.
3799         self.add_attribute(samdb, str(target_creds.get_dn()),
3800                            'msDS-AssignedAuthNPolicy', str(policy))
3801
3802         client_service_tkt = self.get_service_ticket(
3803             client_tgt,
3804             service_creds,
3805             kdc_options=client_tkt_options,
3806             expected_flags=expected_flags)
3807
3808         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
3809
3810         target_decryption_key = self.TicketDecryptionKey_from_creds(
3811             target_creds)
3812
3813         # Show that obtaining a service ticket with constrained delegation is
3814         # not allowed.
3815         self._tgs_req(
3816             service_tgt, KDC_ERR_POLICY, service_creds, target_creds,
3817             armor_tgt=mach_tgt,
3818             kdc_options=kdc_options,
3819             additional_ticket=client_service_tkt,
3820             decryption_key=target_decryption_key,
3821             expect_edata=self.expect_padata_outer,
3822             # We aren’t particular about whether or not we get an NTSTATUS.
3823             expect_status=None,
3824             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
3825             check_patypes=False)
3826
3827     def test_authn_policy_allowed_to_user_allow_constrained_delegation_wrong_sname(self):
3828         client_creds = self.get_cached_creds(
3829             account_type=self.AccountType.USER,
3830             use_cache=False)
3831
3832         client_tkt_options = 'forwardable'
3833         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
3834
3835         client_tgt = self.get_tgt(client_creds,
3836                                   kdc_options=client_tkt_options,
3837                                   expected_flags=expected_flags)
3838
3839         # Create a machine account with which to perform FAST.
3840         mach_creds = self.get_cached_creds(
3841             account_type=self.AccountType.COMPUTER)
3842         mach_tgt = self.get_tgt(mach_creds)
3843
3844         # Create a target account.
3845         target_creds = self.get_cached_creds(
3846             account_type=self.AccountType.COMPUTER,
3847             opts={'id': 1})
3848         target_spn = target_creds.get_spn()
3849
3850         service_creds = self.get_cached_creds(
3851             account_type=self.AccountType.COMPUTER,
3852             opts={'delegation_to_spn': target_spn})
3853         service_tgt = self.get_tgt(service_creds)
3854
3855         client_service_tkt = self.get_service_ticket(
3856             client_tgt,
3857             service_creds,
3858             kdc_options=client_tkt_options,
3859             expected_flags=expected_flags,
3860             fresh=True)
3861         # Change the ‘sname’ of the ticket to an incorrect value.
3862         client_service_tkt.set_sname(self.get_krbtgt_sname())
3863
3864         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
3865
3866         target_decryption_key = self.TicketDecryptionKey_from_creds(
3867             target_creds)
3868
3869         # Show that obtaining a service ticket with constrained delegation
3870         # fails if the sname doesn’t match.
3871         self._tgs_req(service_tgt, KDC_ERR_BADOPTION,
3872                       service_creds, target_creds,
3873                       armor_tgt=mach_tgt,
3874                       kdc_options=kdc_options,
3875                       additional_ticket=client_service_tkt,
3876                       decryption_key=target_decryption_key,
3877                       expect_edata=self.expect_padata_outer,
3878                       check_patypes=False)
3879
3880     def test_authn_policy_allowed_to_user_allow_rbcd(self):
3881         samdb = self.get_samdb()
3882         functional_level = self.get_domain_functional_level(samdb)
3883
3884         if functional_level < dsdb.DS_DOMAIN_FUNCTION_2008:
3885             self.skipTest('RBCD requires FL2008')
3886
3887         client_creds = self.get_cached_creds(
3888             account_type=self.AccountType.USER)
3889         client_dn = client_creds.get_dn()
3890         client_sid = self.get_objectSid(samdb, client_dn)
3891
3892         client_username = client_creds.get_username()
3893         client_cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
3894                                                  names=[client_username])
3895
3896         client_tkt_options = 'forwardable'
3897         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
3898
3899         client_tgt = self.get_tgt(client_creds,
3900                                   kdc_options=client_tkt_options,
3901                                   expected_flags=expected_flags)
3902
3903         # Create a machine account with which to perform FAST.
3904         mach_creds = self.get_cached_creds(
3905             account_type=self.AccountType.COMPUTER)
3906         mach_tgt = self.get_tgt(mach_creds)
3907
3908         service_creds = self.get_cached_creds(
3909             account_type=self.AccountType.COMPUTER,
3910             opts={'id': 1})
3911         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
3912         service_tgt = self.get_tgt(service_creds)
3913
3914         # Create an authentication policy that applies to a computer and
3915         # explicitly allows the service account to obtain a service ticket,
3916         # while denying the user.
3917         allowed = f'O:SYD:(A;;CR;;;{service_sid})(D;;CR;;;{client_sid})'
3918         policy_id = self.get_new_username()
3919         policy = self.create_authn_policy(policy_id,
3920                                           enforced=True,
3921                                           computer_allowed_to=allowed)
3922
3923         # Create a target account with the assigned policy.
3924         target_creds = self.get_cached_creds(
3925             account_type=self.AccountType.COMPUTER,
3926             opts={
3927                 'assigned_policy': str(policy),
3928                 'delegation_from_dn': str(service_creds.get_dn()),
3929             })
3930
3931         client_service_tkt = self.get_service_ticket(
3932             client_tgt,
3933             service_creds,
3934             kdc_options=client_tkt_options,
3935             expected_flags=expected_flags)
3936
3937         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
3938
3939         target_decryption_key = self.TicketDecryptionKey_from_creds(
3940             target_creds)
3941         target_etypes = target_creds.tgs_supported_enctypes
3942
3943         service_name = service_creds.get_username()
3944         if service_name[-1] == '$':
3945             service_name = service_name[:-1]
3946         expected_transited_services = [
3947             f'host/{service_name}@{service_creds.get_realm()}'
3948         ]
3949
3950         # Show that obtaining a service ticket with RBCD is allowed.
3951         self._tgs_req(service_tgt, 0, service_creds, target_creds,
3952                       armor_tgt=mach_tgt,
3953                       kdc_options=kdc_options,
3954                       pac_options='1001',  # supports claims, RBCD
3955                       expected_cname=client_cname,
3956                       expected_account_name=client_username,
3957                       additional_ticket=client_service_tkt,
3958                       decryption_key=target_decryption_key,
3959                       expected_sid=client_sid,
3960                       expected_supported_etypes=target_etypes,
3961                       expected_proxy_target=target_creds.get_spn(),
3962                       expected_transited_services=expected_transited_services)
3963
3964     def test_authn_policy_allowed_to_user_deny_rbcd(self):
3965         samdb = self.get_samdb()
3966         functional_level = self.get_domain_functional_level(samdb)
3967
3968         if functional_level < dsdb.DS_DOMAIN_FUNCTION_2008:
3969             self.skipTest('RBCD requires FL2008')
3970
3971         client_creds = self.get_cached_creds(
3972             account_type=self.AccountType.USER)
3973         client_dn = client_creds.get_dn()
3974         client_sid = self.get_objectSid(samdb, client_dn)
3975
3976         client_tkt_options = 'forwardable'
3977         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
3978
3979         client_tgt = self.get_tgt(client_creds,
3980                                   kdc_options=client_tkt_options,
3981                                   expected_flags=expected_flags)
3982
3983         # Create a machine account with which to perform FAST.
3984         mach_creds = self.get_cached_creds(
3985             account_type=self.AccountType.COMPUTER)
3986         mach_tgt = self.get_tgt(mach_creds)
3987
3988         service_creds = self.get_cached_creds(
3989             account_type=self.AccountType.COMPUTER,
3990             opts={'id': 1})
3991         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
3992         service_tgt = self.get_tgt(service_creds)
3993
3994         # Create an authentication policy that applies to a computer and
3995         # explicitly denies the service account to obtain a service ticket,
3996         # while allowing the user.
3997         denied = f'O:SYD:(D;;CR;;;{service_sid})(A;;CR;;;{client_sid})'
3998         policy_id = self.get_new_username()
3999         policy = self.create_authn_policy(policy_id,
4000                                           enforced=True,
4001                                           computer_allowed_to=denied)
4002
4003         # Create a target account with the assigned policy.
4004         target_creds = self.get_cached_creds(
4005             account_type=self.AccountType.COMPUTER,
4006             opts={
4007                 'assigned_policy': str(policy),
4008                 'delegation_from_dn': str(service_creds.get_dn()),
4009             })
4010
4011         client_service_tkt = self.get_service_ticket(
4012             client_tgt,
4013             service_creds,
4014             kdc_options=client_tkt_options,
4015             expected_flags=expected_flags)
4016
4017         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4018
4019         target_decryption_key = self.TicketDecryptionKey_from_creds(
4020             target_creds)
4021
4022         # Show that obtaining a service ticket with RBCD is not allowed.
4023         self._tgs_req(service_tgt, KDC_ERR_POLICY, service_creds, target_creds,
4024                       armor_tgt=mach_tgt,
4025                       kdc_options=kdc_options,
4026                       pac_options='1001',  # supports claims, RBCD
4027                       additional_ticket=client_service_tkt,
4028                       decryption_key=target_decryption_key,
4029                       expect_edata=self.expect_padata_outer,
4030                       check_patypes=False)
4031
4032     def test_authn_policy_allowed_to_user_allow_rbcd_wrong_sname(self):
4033         samdb = self.get_samdb()
4034         functional_level = self.get_domain_functional_level(samdb)
4035
4036         if functional_level < dsdb.DS_DOMAIN_FUNCTION_2008:
4037             self.skipTest('RBCD requires FL2008')
4038
4039         client_creds = self.get_cached_creds(
4040             account_type=self.AccountType.USER,
4041             use_cache=False)
4042
4043         client_tkt_options = 'forwardable'
4044         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
4045
4046         client_tgt = self.get_tgt(client_creds,
4047                                   kdc_options=client_tkt_options,
4048                                   expected_flags=expected_flags)
4049
4050         # Create a machine account with which to perform FAST.
4051         mach_creds = self.get_cached_creds(
4052             account_type=self.AccountType.COMPUTER)
4053         mach_tgt = self.get_tgt(mach_creds)
4054
4055         service_creds = self.get_cached_creds(
4056             account_type=self.AccountType.COMPUTER,
4057             opts={'id': 1})
4058         service_tgt = self.get_tgt(service_creds)
4059
4060         # Create a target account with the assigned policy.
4061         target_creds = self.get_cached_creds(
4062             account_type=self.AccountType.COMPUTER,
4063             opts={
4064                 'delegation_from_dn': str(service_creds.get_dn()),
4065             })
4066
4067         client_service_tkt = self.get_service_ticket(
4068             client_tgt,
4069             service_creds,
4070             kdc_options=client_tkt_options,
4071             expected_flags=expected_flags,
4072             fresh=True)
4073         # Change the ‘sname’ of the ticket to an incorrect value.
4074         client_service_tkt.set_sname(self.get_krbtgt_sname())
4075
4076         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4077
4078         target_decryption_key = self.TicketDecryptionKey_from_creds(
4079             target_creds)
4080
4081         # Show that obtaining a service ticket with RBCD fails if the sname
4082         # doesn’t match.
4083         self._tgs_req(service_tgt, KDC_ERR_BADOPTION,
4084                       service_creds, target_creds,
4085                       armor_tgt=mach_tgt,
4086                       kdc_options=kdc_options,
4087                       pac_options='1001',  # supports claims, RBCD
4088                       additional_ticket=client_service_tkt,
4089                       decryption_key=target_decryption_key,
4090                       expect_edata=self.expect_padata_outer,
4091                       check_patypes=False)
4092
4093     def test_authn_policy_allowed_to_user_allow_constrained_delegation_to_self(self):
4094         samdb = self.get_samdb()
4095
4096         client_creds = self.get_cached_creds(
4097             account_type=self.AccountType.USER)
4098         client_dn = client_creds.get_dn()
4099         client_sid = self.get_objectSid(samdb, client_dn)
4100
4101         client_username = client_creds.get_username()
4102         client_cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
4103                                                  names=[client_username])
4104
4105         client_tkt_options = 'forwardable'
4106         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
4107
4108         client_tgt = self.get_tgt(client_creds,
4109                                   kdc_options=client_tkt_options,
4110                                   expected_flags=expected_flags)
4111
4112         # Create a machine account with which to perform FAST.
4113         mach_creds = self.get_cached_creds(
4114             account_type=self.AccountType.COMPUTER)
4115         mach_tgt = self.get_tgt(mach_creds)
4116
4117         # Create a service account.
4118         service_creds = self.get_cached_creds(
4119             account_type=self.AccountType.COMPUTER,
4120             opts={'id': 1},
4121             use_cache=False)
4122         service_dn_str = str(service_creds.get_dn())
4123         service_spn = service_creds.get_spn()
4124         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
4125         service_tgt = self.get_tgt(service_creds)
4126
4127         # Allow delegation to ourselves.
4128         self.add_attribute(samdb, service_dn_str,
4129                            'msDS-AllowedToDelegateTo', service_spn)
4130
4131         # Create an authentication policy that applies to a computer and
4132         # explicitly allows the client account to obtain a service ticket,
4133         # while denying the service.
4134         allowed = f'O:SYD:(A;;CR;;;{client_sid})(D;;CR;;;{service_sid})'
4135         policy_id = self.get_new_username()
4136         policy = self.create_authn_policy(policy_id,
4137                                           enforced=True,
4138                                           computer_allowed_to=allowed)
4139
4140         # Assign the policy to the service account.
4141         self.add_attribute(samdb, service_dn_str,
4142                            'msDS-AssignedAuthNPolicy', str(policy))
4143
4144         client_service_tkt = self.get_service_ticket(
4145             client_tgt,
4146             service_creds,
4147             kdc_options=client_tkt_options,
4148             expected_flags=expected_flags)
4149
4150         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4151
4152         target_decryption_key = self.TicketDecryptionKey_from_creds(
4153             service_creds)
4154         target_etypes = service_creds.tgs_supported_enctypes
4155
4156         service_name = service_creds.get_username()
4157         if service_name[-1] == '$':
4158             service_name = service_name[:-1]
4159         expected_transited_services = [
4160             f'host/{service_name}@{service_creds.get_realm()}'
4161         ]
4162
4163         # Show that obtaining a service ticket to ourselves with constrained
4164         # delegation is allowed.
4165         self._tgs_req(service_tgt, 0, service_creds, service_creds,
4166                       armor_tgt=mach_tgt,
4167                       kdc_options=kdc_options,
4168                       expected_cname=client_cname,
4169                       expected_account_name=client_username,
4170                       additional_ticket=client_service_tkt,
4171                       decryption_key=target_decryption_key,
4172                       expected_sid=client_sid,
4173                       expected_supported_etypes=target_etypes,
4174                       expected_proxy_target=service_spn,
4175                       expected_transited_services=expected_transited_services)
4176
4177     def test_authn_policy_allowed_to_user_deny_constrained_delegation_to_self(self):
4178         samdb = self.get_samdb()
4179
4180         client_creds = self.get_cached_creds(
4181             account_type=self.AccountType.USER)
4182         client_dn = client_creds.get_dn()
4183         client_sid = self.get_objectSid(samdb, client_dn)
4184
4185         client_username = client_creds.get_username()
4186         client_cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
4187                                                  names=[client_username])
4188
4189         client_tkt_options = 'forwardable'
4190         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
4191
4192         client_tgt = self.get_tgt(client_creds,
4193                                   kdc_options=client_tkt_options,
4194                                   expected_flags=expected_flags)
4195
4196         # Create a machine account with which to perform FAST.
4197         mach_creds = self.get_cached_creds(
4198             account_type=self.AccountType.COMPUTER)
4199         mach_tgt = self.get_tgt(mach_creds)
4200
4201         # Create a service account.
4202         service_creds = self.get_cached_creds(
4203             account_type=self.AccountType.COMPUTER,
4204             opts={'id': 1},
4205             use_cache=False)
4206         service_dn_str = str(service_creds.get_dn())
4207         service_spn = service_creds.get_spn()
4208         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
4209         service_tgt = self.get_tgt(service_creds)
4210
4211         # Allow delegation to ourselves.
4212         self.add_attribute(samdb, service_dn_str,
4213                            'msDS-AllowedToDelegateTo', service_spn)
4214
4215         client_service_tkt = self.get_service_ticket(
4216             client_tgt,
4217             service_creds,
4218             kdc_options=client_tkt_options,
4219             expected_flags=expected_flags)
4220
4221         # Create an authentication policy that applies to a computer and
4222         # explicitly denies the client account to obtain a service ticket,
4223         # while allowing the service.
4224         allowed = f'O:SYD:(D;;CR;;;{client_sid})(A;;CR;;;{service_sid})'
4225         policy_id = self.get_new_username()
4226         policy = self.create_authn_policy(policy_id,
4227                                           enforced=True,
4228                                           computer_allowed_to=allowed)
4229
4230         # Assign the policy to the service account.
4231         self.add_attribute(samdb, service_dn_str,
4232                            'msDS-AssignedAuthNPolicy', str(policy))
4233
4234         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4235
4236         target_decryption_key = self.TicketDecryptionKey_from_creds(
4237             service_creds)
4238         target_etypes = service_creds.tgs_supported_enctypes
4239
4240         service_name = service_creds.get_username()
4241         if service_name[-1] == '$':
4242             service_name = service_name[:-1]
4243         expected_transited_services = [
4244             f'host/{service_name}@{service_creds.get_realm()}'
4245         ]
4246
4247         # Show that obtaining a service ticket to ourselves with constrained
4248         # delegation is allowed, despite the policy’s disallowing it.
4249         self._tgs_req(service_tgt, 0, service_creds, service_creds,
4250                       armor_tgt=mach_tgt,
4251                       kdc_options=kdc_options,
4252                       expected_cname=client_cname,
4253                       expected_account_name=client_username,
4254                       additional_ticket=client_service_tkt,
4255                       decryption_key=target_decryption_key,
4256                       expected_sid=client_sid,
4257                       expected_supported_etypes=target_etypes,
4258                       expected_proxy_target=service_spn,
4259                       expected_transited_services=expected_transited_services)
4260
4261     def test_authn_policy_allowed_to_user_not_allowed_constrained_delegation_to_self(self):
4262         samdb = self.get_samdb()
4263
4264         client_creds = self.get_cached_creds(
4265             account_type=self.AccountType.USER)
4266         client_dn = client_creds.get_dn()
4267         client_sid = self.get_objectSid(samdb, client_dn)
4268
4269         client_tkt_options = 'forwardable'
4270         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
4271
4272         client_tgt = self.get_tgt(client_creds,
4273                                   kdc_options=client_tkt_options,
4274                                   expected_flags=expected_flags)
4275
4276         # Create a machine account with which to perform FAST.
4277         mach_creds = self.get_cached_creds(
4278             account_type=self.AccountType.COMPUTER)
4279         mach_tgt = self.get_tgt(mach_creds)
4280
4281         # Create a service account.
4282         service_creds = self.get_cached_creds(
4283             account_type=self.AccountType.COMPUTER,
4284             opts={'id': 1},
4285             use_cache=False)
4286         service_dn_str = str(service_creds.get_dn())
4287         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
4288         service_tgt = self.get_tgt(service_creds)
4289
4290         # Don’t set msDS-AllowedToDelegateTo.
4291
4292         # Create an authentication policy that applies to a computer and
4293         # explicitly allows the client account to obtain a service ticket,
4294         # while denying the service.
4295         allowed = f'O:SYD:(A;;CR;;;{client_sid})(D;;CR;;;{service_sid})'
4296         policy_id = self.get_new_username()
4297         policy = self.create_authn_policy(policy_id,
4298                                           enforced=True,
4299                                           computer_allowed_to=allowed)
4300
4301         # Assign the policy to the service account.
4302         self.add_attribute(samdb, service_dn_str,
4303                            'msDS-AssignedAuthNPolicy', str(policy))
4304
4305         client_service_tkt = self.get_service_ticket(
4306             client_tgt,
4307             service_creds,
4308             kdc_options=client_tkt_options,
4309             expected_flags=expected_flags)
4310
4311         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4312
4313         target_decryption_key = self.TicketDecryptionKey_from_creds(
4314             service_creds)
4315
4316         # Show that obtaining a service ticket to ourselves with constrained
4317         # delegation is not allowed without msDS-AllowedToDelegateTo.
4318         self._tgs_req(service_tgt, KDC_ERR_BADOPTION,
4319                       service_creds, service_creds,
4320                       armor_tgt=mach_tgt,
4321                       kdc_options=kdc_options,
4322                       additional_ticket=client_service_tkt,
4323                       decryption_key=target_decryption_key,
4324                       expect_edata=self.expect_padata_outer,
4325                       check_patypes=False)
4326
4327     def test_authn_policy_allowed_to_user_allow_rbcd_to_self(self):
4328         samdb = self.get_samdb()
4329         functional_level = self.get_domain_functional_level(samdb)
4330
4331         if functional_level < dsdb.DS_DOMAIN_FUNCTION_2008:
4332             self.skipTest('RBCD requires FL2008')
4333
4334         client_creds = self.get_cached_creds(
4335             account_type=self.AccountType.USER)
4336         client_dn = client_creds.get_dn()
4337         client_sid = self.get_objectSid(samdb, client_dn)
4338
4339         client_username = client_creds.get_username()
4340         client_cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
4341                                                  names=[client_username])
4342
4343         client_tkt_options = 'forwardable'
4344         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
4345
4346         client_tgt = self.get_tgt(client_creds,
4347                                   kdc_options=client_tkt_options,
4348                                   expected_flags=expected_flags)
4349
4350         # Create a machine account with which to perform FAST.
4351         mach_creds = self.get_cached_creds(
4352             account_type=self.AccountType.COMPUTER)
4353         mach_tgt = self.get_tgt(mach_creds)
4354
4355         # Create a service account allowed to delegate to itself. We can’t use
4356         # a more specific ACE containing the account’s SID (obtained
4357         # post-creation) as Samba (unlike Windows) won’t let us modify
4358         # msDS-AllowedToActOnBehalfOfOtherIdentity without being System.
4359         domain_sid = security.dom_sid(samdb.get_domain_sid())
4360         security_descriptor = security.descriptor.from_sddl(
4361             'O:BAD:(A;;CR;;;WD)', domain_sid)
4362         service_creds = self.get_cached_creds(
4363             account_type=self.AccountType.COMPUTER,
4364             opts={'delegation_from_dn': ndr_pack(security_descriptor)},
4365             use_cache=False)
4366         service_dn_str = str(service_creds.get_dn())
4367         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
4368         service_tgt = self.get_tgt(service_creds)
4369
4370         # Create an authentication policy that applies to a computer and
4371         # explicitly allows the client account to obtain a service ticket,
4372         # while denying the service.
4373         allowed = f'O:SYD:(A;;CR;;;{client_sid})(D;;CR;;;{service_sid})'
4374         policy_id = self.get_new_username()
4375         policy = self.create_authn_policy(policy_id,
4376                                           enforced=True,
4377                                           computer_allowed_to=allowed)
4378
4379         # Assign the policy to the service account.
4380         self.add_attribute(samdb, service_dn_str,
4381                            'msDS-AssignedAuthNPolicy', str(policy))
4382
4383         client_service_tkt = self.get_service_ticket(
4384             client_tgt,
4385             service_creds,
4386             kdc_options=client_tkt_options,
4387             expected_flags=expected_flags)
4388
4389         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4390
4391         service_decryption_key = self.TicketDecryptionKey_from_creds(
4392             service_creds)
4393         service_etypes = service_creds.tgs_supported_enctypes
4394
4395         service_name = service_creds.get_username()
4396         if service_name[-1] == '$':
4397             service_name = service_name[:-1]
4398         expected_transited_services = [
4399             f'host/{service_name}@{service_creds.get_realm()}'
4400         ]
4401
4402         # Show that obtaining a service ticket to ourselves with RBCD is
4403         # allowed.
4404         self._tgs_req(service_tgt, 0, service_creds, service_creds,
4405                       armor_tgt=mach_tgt,
4406                       kdc_options=kdc_options,
4407                       pac_options='1001',  # supports claims, RBCD
4408                       expected_cname=client_cname,
4409                       expected_account_name=client_username,
4410                       additional_ticket=client_service_tkt,
4411                       decryption_key=service_decryption_key,
4412                       expected_sid=client_sid,
4413                       expected_supported_etypes=service_etypes,
4414                       expected_proxy_target=service_creds.get_spn(),
4415                       expected_transited_services=expected_transited_services)
4416
4417     def test_authn_policy_allowed_to_user_deny_rbcd_to_self(self):
4418         samdb = self.get_samdb()
4419         functional_level = self.get_domain_functional_level(samdb)
4420
4421         if functional_level < dsdb.DS_DOMAIN_FUNCTION_2008:
4422             self.skipTest('RBCD requires FL2008')
4423
4424         client_creds = self.get_cached_creds(
4425             account_type=self.AccountType.USER)
4426         client_dn = client_creds.get_dn()
4427         client_sid = self.get_objectSid(samdb, client_dn)
4428
4429         client_username = client_creds.get_username()
4430         client_cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
4431                                                  names=[client_username])
4432
4433         client_tkt_options = 'forwardable'
4434         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
4435
4436         client_tgt = self.get_tgt(client_creds,
4437                                   kdc_options=client_tkt_options,
4438                                   expected_flags=expected_flags)
4439
4440         # Create a machine account with which to perform FAST.
4441         mach_creds = self.get_cached_creds(
4442             account_type=self.AccountType.COMPUTER)
4443         mach_tgt = self.get_tgt(mach_creds)
4444
4445         # Create a service account allowed to delegate to itself. We can’t use
4446         # a more specific ACE containing the account’s SID (obtained
4447         # post-creation) as Samba (unlike Windows) won’t let us modify
4448         # msDS-AllowedToActOnBehalfOfOtherIdentity without being System.
4449         domain_sid = security.dom_sid(samdb.get_domain_sid())
4450         security_descriptor = security.descriptor.from_sddl(
4451             'O:BAD:(A;;CR;;;WD)', domain_sid)
4452         service_creds = self.get_cached_creds(
4453             account_type=self.AccountType.COMPUTER,
4454             opts={'delegation_from_dn': ndr_pack(security_descriptor)},
4455             use_cache=False)
4456         service_dn_str = str(service_creds.get_dn())
4457         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
4458         service_tgt = self.get_tgt(service_creds)
4459
4460         client_service_tkt = self.get_service_ticket(
4461             client_tgt,
4462             service_creds,
4463             kdc_options=client_tkt_options,
4464             expected_flags=expected_flags)
4465
4466         # Create an authentication policy that applies to a computer and
4467         # explicitly denies the client account to obtain a service ticket,
4468         # while allowing the service.
4469         allowed = f'O:SYD:(D;;CR;;;{client_sid})(A;;CR;;;{service_sid})'
4470         policy_id = self.get_new_username()
4471         policy = self.create_authn_policy(policy_id,
4472                                           enforced=True,
4473                                           computer_allowed_to=allowed)
4474
4475         # Assign the policy to the service account.
4476         self.add_attribute(samdb, service_dn_str,
4477                            'msDS-AssignedAuthNPolicy', str(policy))
4478
4479         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4480
4481         service_decryption_key = self.TicketDecryptionKey_from_creds(
4482             service_creds)
4483         service_etypes = service_creds.tgs_supported_enctypes
4484
4485         service_name = service_creds.get_username()
4486         if service_name[-1] == '$':
4487             service_name = service_name[:-1]
4488         expected_transited_services = [
4489             f'host/{service_name}@{service_creds.get_realm()}'
4490         ]
4491
4492         # Show that obtaining a service ticket to ourselves with RBCD is
4493         # allowed, despite the policy’s disallowing it.
4494         self._tgs_req(service_tgt, 0, service_creds, service_creds,
4495                       armor_tgt=mach_tgt,
4496                       kdc_options=kdc_options,
4497                       pac_options='1001',  # supports claims, RBCD
4498                       expected_cname=client_cname,
4499                       expected_account_name=client_username,
4500                       additional_ticket=client_service_tkt,
4501                       decryption_key=service_decryption_key,
4502                       expected_sid=client_sid,
4503                       expected_supported_etypes=service_etypes,
4504                       expected_proxy_target=service_creds.get_spn(),
4505                       expected_transited_services=expected_transited_services)
4506
4507     def test_authn_policy_allowed_to_user_not_allowed_rbcd_to_self(self):
4508         samdb = self.get_samdb()
4509         functional_level = self.get_domain_functional_level(samdb)
4510
4511         if functional_level < dsdb.DS_DOMAIN_FUNCTION_2008:
4512             self.skipTest('RBCD requires FL2008')
4513
4514         client_creds = self.get_cached_creds(
4515             account_type=self.AccountType.USER)
4516         client_dn = client_creds.get_dn()
4517         client_sid = self.get_objectSid(samdb, client_dn)
4518
4519         client_tkt_options = 'forwardable'
4520         expected_flags = krb5_asn1.TicketFlags(client_tkt_options)
4521
4522         client_tgt = self.get_tgt(client_creds,
4523                                   kdc_options=client_tkt_options,
4524                                   expected_flags=expected_flags)
4525
4526         # Create a machine account with which to perform FAST.
4527         mach_creds = self.get_cached_creds(
4528             account_type=self.AccountType.COMPUTER)
4529         mach_tgt = self.get_tgt(mach_creds)
4530
4531         # Create a service account.
4532         service_creds = self.get_cached_creds(
4533             account_type=self.AccountType.COMPUTER,
4534             opts={'id': 1},
4535             use_cache=False)
4536         service_dn_str = str(service_creds.get_dn())
4537         service_sid = self.get_objectSid(samdb, service_creds.get_dn())
4538         service_tgt = self.get_tgt(service_creds)
4539
4540         # Don’t set msDS-AllowedToActOnBehalfOfOtherIdentity.
4541
4542         # Create an authentication policy that applies to a computer and
4543         # explicitly allows the client account to obtain a service ticket,
4544         # while denying the service.
4545         allowed = f'O:SYD:(A;;CR;;;{client_sid})(D;;CR;;;{service_sid})'
4546         policy_id = self.get_new_username()
4547         policy = self.create_authn_policy(policy_id,
4548                                           enforced=True,
4549                                           computer_allowed_to=allowed)
4550
4551         # Assign the policy to the service account.
4552         self.add_attribute(samdb, service_dn_str,
4553                            'msDS-AssignedAuthNPolicy', str(policy))
4554
4555         client_service_tkt = self.get_service_ticket(
4556             client_tgt,
4557             service_creds,
4558             kdc_options=client_tkt_options,
4559             expected_flags=expected_flags)
4560
4561         kdc_options = str(krb5_asn1.KDCOptions('cname-in-addl-tkt'))
4562
4563         service_decryption_key = self.TicketDecryptionKey_from_creds(
4564             service_creds)
4565
4566         # Show that obtaining a service ticket to ourselves with RBCD
4567         # is not allowed without msDS-AllowedToActOnBehalfOfOtherIdentity.
4568         self._tgs_req(service_tgt, KDC_ERR_BADOPTION,
4569                       service_creds, service_creds,
4570                       armor_tgt=mach_tgt,
4571                       kdc_options=kdc_options,
4572                       pac_options='1001',  # supports claims, RBCD
4573                       additional_ticket=client_service_tkt,
4574                       decryption_key=service_decryption_key,
4575                       expect_edata=self.expect_padata_outer,
4576                       check_patypes=False)
4577
4578     def test_authn_policy_allowed_to_computer_allow_user2user(self):
4579         samdb = self.get_samdb()
4580
4581         # Create a machine account with which to perform FAST.
4582         mach_creds = self.get_cached_creds(
4583             account_type=self.AccountType.COMPUTER)
4584         mach_tgt = self.get_tgt(mach_creds)
4585
4586         client_creds = self.get_mach_creds()
4587         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
4588         client_tgt = self.get_tgt(client_creds)
4589
4590         # Create an authentication policy that applies to a computer and
4591         # explicitly allows the user account to obtain a service ticket.
4592         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
4593         policy_id = self.get_new_username()
4594         policy = self.create_authn_policy(policy_id,
4595                                           enforced=True,
4596                                           computer_allowed_to=allowed)
4597
4598         # Create a computer account with the assigned policy.
4599         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
4600                                        assigned_policy=policy)
4601         target_tgt = self._get_tgt(target_creds)
4602
4603         kdc_options = str(krb5_asn1.KDCOptions('enc-tkt-in-skey'))
4604
4605         # Show that obtaining a service ticket with user-to-user is allowed.
4606         self._tgs_req(client_tgt, 0, client_creds, target_creds,
4607                       armor_tgt=mach_tgt,
4608                       kdc_options=kdc_options,
4609                       additional_ticket=target_tgt)
4610
4611     def test_authn_policy_allowed_to_computer_deny_user2user(self):
4612         samdb = self.get_samdb()
4613
4614         # Create a machine account with which to perform FAST.
4615         mach_creds = self.get_cached_creds(
4616             account_type=self.AccountType.COMPUTER)
4617         mach_tgt = self.get_tgt(mach_creds)
4618
4619         client_creds = self.get_mach_creds()
4620         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
4621         client_tgt = self.get_tgt(client_creds)
4622
4623         # Create an authentication policy that applies to a computer and
4624         # explicitly denies the user account to obtain a service ticket.
4625         denied = f'O:SYD:(D;;CR;;;{client_sid})'
4626         policy_id = self.get_new_username()
4627         policy = self.create_authn_policy(policy_id,
4628                                           enforced=True,
4629                                           computer_allowed_to=denied)
4630
4631         # Create a computer account with the assigned policy.
4632         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
4633                                        assigned_policy=policy)
4634         target_tgt = self._get_tgt(target_creds)
4635
4636         kdc_options = str(krb5_asn1.KDCOptions('enc-tkt-in-skey'))
4637
4638         # Show that obtaining a service ticket with user-to-user is not
4639         # allowed.
4640         self._tgs_req(
4641             client_tgt, KDC_ERR_POLICY, client_creds, target_creds,
4642             armor_tgt=mach_tgt,
4643             kdc_options=kdc_options,
4644             additional_ticket=target_tgt,
4645             expect_edata=self.expect_padata_outer,
4646             # We aren’t particular about whether or not we get an NTSTATUS.
4647             expect_status=None,
4648             expected_status=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED,
4649             check_patypes=False)
4650
4651     def test_authn_policy_allowed_to_user_derived_class_allow(self):
4652         samdb = self.get_samdb()
4653
4654         # Create a machine account with which to perform FAST.
4655         mach_creds = self.get_cached_creds(
4656             account_type=self.AccountType.COMPUTER)
4657         mach_tgt = self.get_tgt(mach_creds)
4658
4659         # Create a user account.
4660         client_creds = self.get_cached_creds(
4661             account_type=self.AccountType.USER)
4662         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
4663         tgt = self.get_tgt(client_creds)
4664
4665         # Create an authentication policy that applies to a user and explicitly
4666         # allows the user account to obtain a service ticket.
4667         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
4668         denied = 'O:SYD:(D;;CR;;;WD)'
4669         policy_id = self.get_new_username()
4670         policy = self.create_authn_policy(policy_id,
4671                                           enforced=True,
4672                                           user_allowed_to=allowed,
4673                                           computer_allowed_to=denied,
4674                                           service_allowed_to=denied)
4675
4676         # Create a schema class derived from ‘user’.
4677         class_id = random.randint(0, 100000000)
4678         user_class_cn = f'my-User-Class-{class_id}'
4679         user_class = user_class_cn.replace('-', '')
4680         class_dn = samdb.get_schema_basedn()
4681         class_dn.add_child(f'CN={user_class_cn}')
4682         governs_id = f'1.3.6.1.4.1.7165.4.6.2.9.{class_id}'
4683
4684         samdb.add({
4685             'dn': class_dn,
4686             'objectClass': 'classSchema',
4687             'subClassOf': 'user',
4688             'governsId': governs_id,
4689             'lDAPDisplayName': user_class,
4690         })
4691
4692         # Create an account derived from ‘user’ with the assigned policy.
4693         target_name = self.get_new_username()
4694         target_creds, target_dn = self.create_account(
4695             samdb, target_name,
4696             account_type=self.AccountType.USER,
4697             spn='host/{account}',
4698             additional_details={
4699                 'msDS-AssignedAuthNPolicy': str(policy),
4700                 'objectClass': user_class,
4701             })
4702
4703         keys = self.get_keys(target_creds)
4704         self.creds_set_keys(target_creds, keys)
4705
4706         # Show that obtaining a service ticket is allowed.
4707         self._tgs_req(tgt, 0, client_creds, target_creds,
4708                       armor_tgt=mach_tgt)
4709
4710     def test_authn_policy_allowed_to_computer_derived_class_allow(self):
4711         samdb = self.get_samdb()
4712
4713         # Create a machine account with which to perform FAST.
4714         mach_creds = self.get_cached_creds(
4715             account_type=self.AccountType.COMPUTER)
4716         mach_tgt = self.get_tgt(mach_creds)
4717
4718         # Create a user account.
4719         client_creds = self.get_cached_creds(
4720             account_type=self.AccountType.USER)
4721         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
4722         tgt = self.get_tgt(client_creds)
4723
4724         # Create an authentication policy that applies to a computer and
4725         # explicitly allows the user account to obtain a service ticket.
4726         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
4727         denied = 'O:SYD:(D;;CR;;;WD)'
4728         policy_id = self.get_new_username()
4729         policy = self.create_authn_policy(policy_id,
4730                                           enforced=True,
4731                                           user_allowed_to=denied,
4732                                           computer_allowed_to=allowed,
4733                                           service_allowed_to=denied)
4734
4735         # Create a schema class derived from ‘computer’.
4736         class_id = random.randint(0, 100000000)
4737         computer_class_cn = f'my-Computer-Class-{class_id}'
4738         computer_class = computer_class_cn.replace('-', '')
4739         class_dn = samdb.get_schema_basedn()
4740         class_dn.add_child(f'CN={computer_class_cn}')
4741         governs_id = f'1.3.6.1.4.1.7165.4.6.2.9.{class_id}'
4742
4743         samdb.add({
4744             'dn': class_dn,
4745             'objectClass': 'classSchema',
4746             'subClassOf': 'computer',
4747             'governsId': governs_id,
4748             'lDAPDisplayName': computer_class,
4749         })
4750
4751         # Create an account derived from ‘computer’ with the assigned policy.
4752         target_name = self.get_new_username()
4753         target_creds, target_dn = self.create_account(
4754             samdb, target_name,
4755             account_type=self.AccountType.COMPUTER,
4756             spn=f'host/{target_name}',
4757             additional_details={
4758                 'msDS-AssignedAuthNPolicy': str(policy),
4759                 'objectClass': computer_class,
4760             })
4761
4762         keys = self.get_keys(target_creds)
4763         self.creds_set_keys(target_creds, keys)
4764
4765         # Show that obtaining a service ticket is allowed.
4766         self._tgs_req(tgt, 0, client_creds, target_creds,
4767                       armor_tgt=mach_tgt)
4768
4769     def test_authn_policy_allowed_to_service_derived_class_allow(self):
4770         samdb = self.get_samdb()
4771
4772         # Create a machine account with which to perform FAST.
4773         mach_creds = self.get_cached_creds(
4774             account_type=self.AccountType.COMPUTER)
4775         mach_tgt = self.get_tgt(mach_creds)
4776
4777         # Create a user account.
4778         client_creds = self.get_cached_creds(
4779             account_type=self.AccountType.USER)
4780         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
4781         tgt = self.get_tgt(client_creds)
4782
4783         # Create an authentication policy that applies to a managed service and
4784         # explicitly allows the user account to obtain a service ticket.
4785         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
4786         denied = 'O:SYD:(D;;CR;;;WD)'
4787         policy_id = self.get_new_username()
4788         policy = self.create_authn_policy(policy_id,
4789                                           enforced=True,
4790                                           user_allowed_to=denied,
4791                                           computer_allowed_to=denied,
4792                                           service_allowed_to=allowed)
4793
4794         # Create a schema class derived from ‘msDS-ManagedServiceAccount’.
4795         class_id = random.randint(0, 100000000)
4796         service_class_cn = f'my-Managed-Service-Class-{class_id}'
4797         service_class = service_class_cn.replace('-', '')
4798         class_dn = samdb.get_schema_basedn()
4799         class_dn.add_child(f'CN={service_class_cn}')
4800         governs_id = f'1.3.6.1.4.1.7165.4.6.2.9.{class_id}'
4801
4802         samdb.add({
4803             'dn': class_dn,
4804             'objectClass': 'classSchema',
4805             'subClassOf': 'msDS-ManagedServiceAccount',
4806             'governsId': governs_id,
4807             'lDAPDisplayName': service_class,
4808         })
4809
4810         # Create an account derived from ‘msDS-ManagedServiceAccount’ with the
4811         # assigned policy.
4812         target_name = self.get_new_username()
4813         target_creds, target_dn = self.create_account(
4814             samdb, target_name,
4815             account_type=self.AccountType.MANAGED_SERVICE,
4816             spn=f'host/{target_name}',
4817             additional_details={
4818                 'msDS-AssignedAuthNPolicy': str(policy),
4819                 'objectClass': service_class,
4820             })
4821
4822         # Show that obtaining a service ticket is allowed.
4823         self._tgs_req(tgt, 0, client_creds, target_creds,
4824                       armor_tgt=mach_tgt)
4825
4826     def test_authn_policy_ntlm_allow_user(self):
4827         # Create an authentication policy allowing NTLM authentication for
4828         # users.
4829         allowed = 'O:SYD:(A;;CR;;;WD)'
4830         policy_id = self.get_new_username()
4831         policy = self.create_authn_policy(policy_id,
4832                                           enforced=True,
4833                                           user_allowed_ntlm=True,
4834                                           user_allowed_from=allowed,
4835                                           service_allowed_ntlm=False,
4836                                           service_allowed_from=allowed)
4837
4838         # Create a user account with the assigned policy.
4839         client_creds = self._get_creds(account_type=self.AccountType.USER,
4840                                        assigned_policy=policy,
4841                                        ntlm=True)
4842
4843         # Show that NTLM authentication succeeds.
4844         self._connect(client_creds, simple_bind=False)
4845
4846     def test_authn_policy_ntlm_deny_user(self):
4847         # Create an authentication policy denying NTLM authentication for
4848         # users.
4849         allowed = 'O:SYD:(A;;CR;;;WD)'
4850         policy_id = self.get_new_username()
4851         policy = self.create_authn_policy(policy_id,
4852                                           enforced=True,
4853                                           user_allowed_ntlm=False,
4854                                           user_allowed_from=allowed,
4855                                           service_allowed_ntlm=True,
4856                                           service_allowed_from=allowed)
4857
4858         # Create a user account with the assigned policy.
4859         client_creds = self._get_creds(account_type=self.AccountType.USER,
4860                                        assigned_policy=policy,
4861                                        ntlm=True)
4862
4863         # Show that NTLM authentication fails.
4864         self._connect(client_creds, simple_bind=False,
4865                       expect_error=f'{HRES_SEC_E_LOGON_DENIED:08X}')
4866
4867     def test_authn_policy_ntlm_computer(self):
4868         # Create an authentication policy denying NTLM authentication.
4869         denied = 'O:SYD:(D;;CR;;;WD)'
4870         policy_id = self.get_new_username()
4871         policy = self.create_authn_policy(policy_id,
4872                                           enforced=True,
4873                                           user_allowed_ntlm=False,
4874                                           user_allowed_from=denied,
4875                                           service_allowed_ntlm=False,
4876                                           service_allowed_from=denied)
4877
4878         # Create a computer account with the assigned policy.
4879         client_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
4880                                        assigned_policy=policy,
4881                                        ntlm=True)
4882
4883         # Show that NTLM authentication succeeds.
4884         self._connect(client_creds, simple_bind=False)
4885
4886     def test_authn_policy_ntlm_allow_service(self):
4887         # Create an authentication policy allowing NTLM authentication for
4888         # services.
4889         allowed = 'O:SYD:(A;;CR;;;WD)'
4890         policy_id = self.get_new_username()
4891         policy = self.create_authn_policy(policy_id,
4892                                           enforced=True,
4893                                           user_allowed_ntlm=False,
4894                                           user_allowed_from=allowed,
4895                                           service_allowed_ntlm=True,
4896                                           service_allowed_from=allowed)
4897
4898         # Create a managed service account with the assigned policy.
4899         client_creds = self._get_creds(
4900             account_type=self.AccountType.MANAGED_SERVICE,
4901             assigned_policy=policy,
4902             ntlm=True)
4903
4904         # Show that NTLM authentication succeeds.
4905         self._connect(client_creds, simple_bind=False)
4906
4907     def test_authn_policy_ntlm_deny_service(self):
4908         # Create an authentication policy denying NTLM authentication for
4909         # services.
4910         allowed = 'O:SYD:(A;;CR;;;WD)'
4911         policy_id = self.get_new_username()
4912         policy = self.create_authn_policy(policy_id,
4913                                           enforced=True,
4914                                           user_allowed_ntlm=True,
4915                                           user_allowed_from=allowed,
4916                                           service_allowed_ntlm=False,
4917                                           service_allowed_from=allowed)
4918
4919         # Create a managed service account with the assigned policy.
4920         client_creds = self._get_creds(
4921             account_type=self.AccountType.MANAGED_SERVICE,
4922             assigned_policy=policy,
4923             ntlm=True)
4924
4925         # Show that NTLM authentication fails.
4926         self._connect(client_creds, simple_bind=False,
4927                       expect_error=f'{HRES_SEC_E_LOGON_DENIED:08X}')
4928
4929     def test_authn_policy_ntlm_deny_no_device_restrictions(self):
4930         # Create an authentication policy denying NTLM authentication for
4931         # users.
4932         policy_id = self.get_new_username()
4933         policy = self.create_authn_policy(policy_id,
4934                                           enforced=True,
4935                                           user_allowed_ntlm=False,
4936                                           service_allowed_ntlm=True)
4937
4938         # Create a user account with the assigned policy.
4939         client_creds = self._get_creds(account_type=self.AccountType.USER,
4940                                        assigned_policy=policy,
4941                                        ntlm=True)
4942
4943         # Show that without AllowedToAuthenticateFrom set in the policy, NTLM
4944         # authentication succeeds.
4945         self._connect(client_creds, simple_bind=False)
4946
4947     def test_authn_policy_simple_bind_allow_user(self):
4948         # Create an authentication policy allowing NTLM authentication for
4949         # users.
4950         allowed = 'O:SYD:(A;;CR;;;WD)'
4951         policy_id = self.get_new_username()
4952         policy = self.create_authn_policy(policy_id,
4953                                           enforced=True,
4954                                           user_allowed_ntlm=True,
4955                                           user_allowed_from=allowed,
4956                                           service_allowed_ntlm=False,
4957                                           service_allowed_from=allowed)
4958
4959         # Create a user account with the assigned policy.
4960         client_creds = self._get_creds(account_type=self.AccountType.USER,
4961                                        assigned_policy=policy,
4962                                        ntlm=True)
4963
4964         # Show that a simple bind succeeds.
4965         self._connect(client_creds, simple_bind=True)
4966
4967     def test_authn_policy_simple_bind_deny_user(self):
4968         # Create an authentication policy denying NTLM authentication for
4969         # users.
4970         allowed = 'O:SYD:(A;;CR;;;WD)'
4971         policy_id = self.get_new_username()
4972         policy = self.create_authn_policy(policy_id,
4973                                           enforced=True,
4974                                           user_allowed_ntlm=False,
4975                                           user_allowed_from=allowed,
4976                                           service_allowed_ntlm=True,
4977                                           service_allowed_from=allowed)
4978
4979         # Create a user account with the assigned policy.
4980         client_creds = self._get_creds(account_type=self.AccountType.USER,
4981                                        assigned_policy=policy,
4982                                        ntlm=True)
4983
4984         # Show that a simple bind fails.
4985         self._connect(client_creds, simple_bind=True,
4986                       expect_error=f'{HRES_SEC_E_INVALID_TOKEN:08X}')
4987
4988     def test_authn_policy_simple_bind_deny_no_device_restrictions(self):
4989         # Create an authentication policy denying NTLM authentication for
4990         # users.
4991         policy_id = self.get_new_username()
4992         policy = self.create_authn_policy(policy_id,
4993                                           enforced=True,
4994                                           user_allowed_ntlm=False,
4995                                           service_allowed_ntlm=True)
4996
4997         # Create a user account with the assigned policy.
4998         client_creds = self._get_creds(account_type=self.AccountType.USER,
4999                                        assigned_policy=policy,
5000                                        ntlm=True)
5001
5002         # Show that without AllowedToAuthenticateFrom set in the policy, a
5003         # simple bind succeeds.
5004         self._connect(client_creds, simple_bind=True)
5005
5006     def test_authn_policy_samr_pwd_change_allow_service_allowed_from(self):
5007         # Create an authentication policy allowing NTLM authentication for
5008         # managed service accounts.
5009         allowed = 'O:SYD:(A;;CR;;;WD)'
5010         policy_id = self.get_new_username()
5011         policy = self.create_authn_policy(policy_id,
5012                                           enforced=True,
5013                                           service_allowed_ntlm=True,
5014                                           service_allowed_from=allowed)
5015
5016         # Create a managed service account with the assigned policy.
5017         client_creds = self._get_creds(
5018             account_type=self.AccountType.MANAGED_SERVICE,
5019             assigned_policy=policy,
5020             ntlm=True)
5021
5022         # Show that a SAMR password change is allowed.
5023         self._test_samr_change_password(client_creds, expect_error=None)
5024
5025     def test_authn_policy_samr_pwd_change_allow_service_not_allowed_from(self):
5026         # Create an authentication policy allowing NTLM authentication for
5027         # managed service accounts.
5028         denied = 'O:SYD:(D;;CR;;;WD)'
5029         policy_id = self.get_new_username()
5030         policy = self.create_authn_policy(policy_id,
5031                                           enforced=True,
5032                                           service_allowed_ntlm=True,
5033                                           service_allowed_from=denied)
5034
5035         # Create a managed service account with the assigned policy.
5036         client_creds = self._get_creds(
5037             account_type=self.AccountType.MANAGED_SERVICE,
5038             assigned_policy=policy,
5039             ntlm=True)
5040
5041         # Show that a SAMR password change is allowed.
5042         self._test_samr_change_password(client_creds, expect_error=None)
5043
5044     def test_authn_policy_samr_pwd_change_allow_service_no_allowed_from(self):
5045         # Create an authentication policy allowing NTLM authentication for
5046         # managed service accounts.
5047         policy_id = self.get_new_username()
5048         policy = self.create_authn_policy(policy_id,
5049                                           enforced=True,
5050                                           service_allowed_ntlm=True)
5051
5052         # Create a managed service account with the assigned policy.
5053         client_creds = self._get_creds(
5054             account_type=self.AccountType.MANAGED_SERVICE,
5055             assigned_policy=policy,
5056             ntlm=True)
5057
5058         # Show that a SAMR password change is allowed.
5059         self._test_samr_change_password(client_creds, expect_error=None)
5060
5061     def test_authn_policy_samr_pwd_change_deny_service_allowed_from(self):
5062         # Create an authentication policy denying NTLM authentication for
5063         # managed service accounts.
5064         allowed = 'O:SYD:(A;;CR;;;WD)'
5065         policy_id = self.get_new_username()
5066         policy = self.create_authn_policy(policy_id,
5067                                           enforced=True,
5068                                           service_allowed_ntlm=False,
5069                                           service_allowed_from=allowed)
5070
5071         # Create a managed service account with the assigned policy.
5072         client_creds = self._get_creds(
5073             account_type=self.AccountType.MANAGED_SERVICE,
5074             assigned_policy=policy,
5075             ntlm=True)
5076
5077         # Show that the SAMR connection fails.
5078         self._test_samr_change_password(
5079             client_creds, expect_error=None,
5080             connect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5081
5082     def test_authn_policy_samr_pwd_change_deny_service_not_allowed_from(self):
5083         # Create an authentication policy denying NTLM authentication for
5084         # managed service accounts.
5085         denied = 'O:SYD:(D;;CR;;;WD)'
5086         policy_id = self.get_new_username()
5087         policy = self.create_authn_policy(policy_id,
5088                                           enforced=True,
5089                                           service_allowed_ntlm=False,
5090                                           service_allowed_from=denied)
5091
5092         # Create a managed service account with the assigned policy.
5093         client_creds = self._get_creds(
5094             account_type=self.AccountType.MANAGED_SERVICE,
5095             assigned_policy=policy,
5096             ntlm=True)
5097
5098         # Show that the SAMR connection fails.
5099         self._test_samr_change_password(
5100             client_creds, expect_error=None,
5101             connect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5102
5103     def test_authn_policy_samr_pwd_change_deny_service_no_allowed_from(self):
5104         # Create an authentication policy denying NTLM authentication for
5105         # managed service accounts.
5106         policy_id = self.get_new_username()
5107         policy = self.create_authn_policy(policy_id,
5108                                           enforced=True,
5109                                           service_allowed_ntlm=False)
5110
5111         # Create a managed service account with the assigned policy.
5112         client_creds = self._get_creds(
5113             account_type=self.AccountType.MANAGED_SERVICE,
5114             assigned_policy=policy,
5115             ntlm=True)
5116
5117         # Show that a SAMR password change is allowed.
5118         self._test_samr_change_password(client_creds, expect_error=None)
5119
5120     def test_authn_policy_samlogon_allow_user(self):
5121         # Create an authentication policy allowing NTLM authentication for
5122         # users.
5123         allowed = 'O:SYD:(A;;CR;;;WD)'
5124         policy_id = self.get_new_username()
5125         policy = self.create_authn_policy(policy_id,
5126                                           enforced=True,
5127                                           user_allowed_ntlm=True,
5128                                           user_allowed_from=allowed,
5129                                           service_allowed_ntlm=False,
5130                                           service_allowed_from=allowed)
5131
5132         # Create a user account with the assigned policy.
5133         client_creds = self._get_creds(account_type=self.AccountType.USER,
5134                                        assigned_policy=policy,
5135                                        ntlm=True)
5136
5137         # Show that a network SamLogon succeeds.
5138         self._test_samlogon(creds=client_creds,
5139                             logon_type=netlogon.NetlogonNetworkInformation)
5140
5141         # Show that an interactive SamLogon succeeds. Although MS-APDS doesn’t
5142         # state it, AllowedNTLMNetworkAuthentication applies to interactive
5143         # logons too.
5144         self._test_samlogon(creds=client_creds,
5145                             logon_type=netlogon.NetlogonInteractiveInformation)
5146
5147     def test_authn_policy_samlogon_deny_user(self):
5148         # Create an authentication policy denying NTLM authentication for
5149         # users.
5150         allowed = 'O:SYD:(A;;CR;;;WD)'
5151         policy_id = self.get_new_username()
5152         policy = self.create_authn_policy(policy_id,
5153                                           enforced=True,
5154                                           user_allowed_ntlm=False,
5155                                           user_allowed_from=allowed,
5156                                           service_allowed_ntlm=True,
5157                                           service_allowed_from=allowed)
5158
5159         # Create a user account with the assigned policy.
5160         client_creds = self._get_creds(account_type=self.AccountType.USER,
5161                                        assigned_policy=policy,
5162                                        ntlm=True)
5163
5164         # Show that a network SamLogon fails.
5165         self._test_samlogon(
5166             creds=client_creds,
5167             logon_type=netlogon.NetlogonNetworkInformation,
5168             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5169
5170         # Show that an interactive SamLogon fails.
5171         self._test_samlogon(
5172             creds=client_creds,
5173             logon_type=netlogon.NetlogonInteractiveInformation,
5174             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5175
5176     def test_authn_policy_samlogon_network_computer(self):
5177         # Create an authentication policy denying NTLM authentication.
5178         denied = 'O:SYD:(D;;CR;;;WD)'
5179         policy_id = self.get_new_username()
5180         policy = self.create_authn_policy(policy_id,
5181                                           enforced=True,
5182                                           user_allowed_ntlm=False,
5183                                           user_allowed_from=denied,
5184                                           service_allowed_ntlm=False,
5185                                           service_allowed_from=denied)
5186
5187         # Create a computer account with the assigned policy.
5188         client_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5189                                        assigned_policy=policy,
5190                                        ntlm=True)
5191
5192         # Show that a network SamLogon succeeds.
5193         self._test_samlogon(
5194             creds=client_creds,
5195             logon_type=netlogon.NetlogonNetworkInformation)
5196
5197     def test_authn_policy_samlogon_interactive_allow_user_allowed_from(self):
5198         # Create an authentication policy allowing NTLM authentication for
5199         # users.
5200         allowed = 'O:SYD:(A;;CR;;;WD)'
5201         policy_id = self.get_new_username()
5202         policy = self.create_authn_policy(policy_id,
5203                                           enforced=True,
5204                                           user_allowed_ntlm=True,
5205                                           user_allowed_from=allowed)
5206
5207         # Create a user account with the assigned policy.
5208         client_creds = self._get_creds(account_type=self.AccountType.USER,
5209                                        assigned_policy=policy,
5210                                        ntlm=True,
5211                                        cached=False)
5212
5213         # Show that an interactive SamLogon succeeds. Although MS-APDS doesn’t
5214         # state it, AllowedNTLMNetworkAuthentication applies to interactive
5215         # logons too.
5216         self._test_samlogon(creds=client_creds,
5217                             logon_type=netlogon.NetlogonInteractiveInformation)
5218
5219     def test_authn_policy_samlogon_interactive_allow_user_not_allowed_from(self):
5220         # Create an authentication policy allowing NTLM authentication for
5221         # users.
5222         denied = 'O:SYD:(D;;CR;;;WD)'
5223         policy_id = self.get_new_username()
5224         policy = self.create_authn_policy(policy_id,
5225                                           enforced=True,
5226                                           user_allowed_ntlm=True,
5227                                           user_allowed_from=denied)
5228
5229         # Create a user account with the assigned policy.
5230         client_creds = self._get_creds(account_type=self.AccountType.USER,
5231                                        assigned_policy=policy,
5232                                        ntlm=True,
5233                                        cached=False)
5234
5235         # Show that an interactive SamLogon succeeds. Although MS-APDS doesn’t
5236         # state it, AllowedNTLMNetworkAuthentication applies to interactive
5237         # logons too.
5238         self._test_samlogon(creds=client_creds,
5239                             logon_type=netlogon.NetlogonInteractiveInformation)
5240
5241     def test_authn_policy_samlogon_interactive_allow_user_no_allowed_from(self):
5242         # Create an authentication policy allowing NTLM authentication for
5243         # users.
5244         policy_id = self.get_new_username()
5245         policy = self.create_authn_policy(policy_id,
5246                                           enforced=True,
5247                                           user_allowed_ntlm=True)
5248
5249         # Create a user account with the assigned policy.
5250         client_creds = self._get_creds(account_type=self.AccountType.USER,
5251                                        assigned_policy=policy,
5252                                        ntlm=True,
5253                                        cached=False)
5254
5255         # Show that an interactive SamLogon succeeds.
5256         self._test_samlogon(creds=client_creds,
5257                             logon_type=netlogon.NetlogonInteractiveInformation)
5258
5259     def test_authn_policy_samlogon_interactive_deny_user_allowed_from(self):
5260         # Create an authentication policy disallowing NTLM authentication for
5261         # users.
5262         allowed = 'O:SYD:(A;;CR;;;WD)'
5263         policy_id = self.get_new_username()
5264         policy = self.create_authn_policy(policy_id,
5265                                           enforced=True,
5266                                           user_allowed_ntlm=False,
5267                                           user_allowed_from=allowed)
5268
5269         # Create a user account with the assigned policy.
5270         client_creds = self._get_creds(account_type=self.AccountType.USER,
5271                                        assigned_policy=policy,
5272                                        ntlm=True,
5273                                        cached=False)
5274
5275         # Show that an interactive SamLogon fails.
5276         self._test_samlogon(
5277             creds=client_creds,
5278             logon_type=netlogon.NetlogonInteractiveInformation,
5279             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5280
5281     def test_authn_policy_samlogon_interactive_deny_user_not_allowed_from(self):
5282         # Create an authentication policy disallowing NTLM authentication for
5283         # users.
5284         denied = 'O:SYD:(D;;CR;;;WD)'
5285         policy_id = self.get_new_username()
5286         policy = self.create_authn_policy(policy_id,
5287                                           enforced=True,
5288                                           user_allowed_ntlm=False,
5289                                           user_allowed_from=denied)
5290
5291         # Create a user account with the assigned policy.
5292         client_creds = self._get_creds(account_type=self.AccountType.USER,
5293                                        assigned_policy=policy,
5294                                        ntlm=True,
5295                                        cached=False)
5296
5297         # Show that an interactive SamLogon fails.
5298         self._test_samlogon(
5299             creds=client_creds,
5300             logon_type=netlogon.NetlogonInteractiveInformation,
5301             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5302
5303     def test_authn_policy_samlogon_interactive_deny_user_no_allowed_from(self):
5304         # Create an authentication policy disallowing NTLM authentication for
5305         # users.
5306         policy_id = self.get_new_username()
5307         policy = self.create_authn_policy(policy_id,
5308                                           enforced=True,
5309                                           user_allowed_ntlm=False)
5310
5311         # Create a user account with the assigned policy.
5312         client_creds = self._get_creds(account_type=self.AccountType.USER,
5313                                        assigned_policy=policy,
5314                                        ntlm=True,
5315                                        cached=False)
5316
5317         # Show that an interactive SamLogon succeeds.
5318         self._test_samlogon(creds=client_creds,
5319                             logon_type=netlogon.NetlogonInteractiveInformation)
5320
5321     def test_authn_policy_samlogon_interactive_user_allowed_from(self):
5322         # Create an authentication policy not specifying whether NTLM
5323         # authentication is allowed or not.
5324         allowed = 'O:SYD:(A;;CR;;;WD)'
5325         policy_id = self.get_new_username()
5326         policy = self.create_authn_policy(policy_id,
5327                                           enforced=True,
5328                                           user_allowed_from=allowed)
5329
5330         # Create a user account with the assigned policy.
5331         client_creds = self._get_creds(account_type=self.AccountType.USER,
5332                                        assigned_policy=policy,
5333                                        ntlm=True,
5334                                        cached=False)
5335
5336         # Show that an interactive SamLogon fails.
5337         self._test_samlogon(
5338             creds=client_creds,
5339             logon_type=netlogon.NetlogonInteractiveInformation,
5340             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5341
5342     def test_authn_policy_samlogon_network_user_allowed_from(self):
5343         # Create an authentication policy not specifying whether NTLM
5344         # authentication is allowed or not.
5345         allowed = 'O:SYD:(A;;CR;;;WD)'
5346         policy_id = self.get_new_username()
5347         policy = self.create_authn_policy(policy_id,
5348                                           enforced=True,
5349                                           user_allowed_from=allowed)
5350
5351         # Create a user account with the assigned policy.
5352         client_creds = self._get_creds(account_type=self.AccountType.USER,
5353                                        assigned_policy=policy,
5354                                        ntlm=True,
5355                                        cached=False)
5356
5357         # Show that a network SamLogon fails.
5358         self._test_samlogon(
5359             creds=client_creds,
5360             logon_type=netlogon.NetlogonNetworkInformation,
5361             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5362
5363     def test_authn_policy_samlogon_network_allow_service_allowed_from(self):
5364         # Create an authentication policy allowing NTLM authentication for
5365         # services.
5366         allowed = 'O:SYD:(A;;CR;;;WD)'
5367         policy_id = self.get_new_username()
5368         policy = self.create_authn_policy(policy_id,
5369                                           enforced=True,
5370                                           service_allowed_ntlm=True,
5371                                           service_allowed_from=allowed)
5372
5373         # Create a managed service account with the assigned policy.
5374         client_creds = self._get_creds(
5375             account_type=self.AccountType.MANAGED_SERVICE,
5376             assigned_policy=policy,
5377             ntlm=True,
5378             cached=False)
5379
5380         # Show that a network SamLogon succeeds.
5381         self._test_samlogon(creds=client_creds,
5382                             logon_type=netlogon.NetlogonNetworkInformation)
5383
5384     def test_authn_policy_samlogon_network_allow_service_not_allowed_from(self):
5385         # Create an authentication policy allowing NTLM authentication for
5386         # services.
5387         denied = 'O:SYD:(D;;CR;;;WD)'
5388         policy_id = self.get_new_username()
5389         policy = self.create_authn_policy(policy_id,
5390                                           enforced=True,
5391                                           service_allowed_ntlm=True,
5392                                           service_allowed_from=denied)
5393
5394         # Create a managed service account with the assigned policy.
5395         client_creds = self._get_creds(
5396             account_type=self.AccountType.MANAGED_SERVICE,
5397             assigned_policy=policy,
5398             ntlm=True,
5399             cached=False)
5400
5401         # Show that a network SamLogon succeeds.
5402         self._test_samlogon(creds=client_creds,
5403                             logon_type=netlogon.NetlogonNetworkInformation)
5404
5405     def test_authn_policy_samlogon_network_allow_service_no_allowed_from(self):
5406         # Create an authentication policy allowing NTLM authentication for
5407         # services.
5408         policy_id = self.get_new_username()
5409         policy = self.create_authn_policy(policy_id,
5410                                           enforced=True,
5411                                           service_allowed_ntlm=True)
5412
5413         # Create a managed service account with the assigned policy.
5414         client_creds = self._get_creds(
5415             account_type=self.AccountType.MANAGED_SERVICE,
5416             assigned_policy=policy,
5417             ntlm=True,
5418             cached=False)
5419
5420         # Show that a network SamLogon succeeds.
5421         self._test_samlogon(creds=client_creds,
5422                             logon_type=netlogon.NetlogonNetworkInformation)
5423
5424     def test_authn_policy_samlogon_network_deny_service_allowed_from(self):
5425         # Create an authentication policy disallowing NTLM authentication for
5426         # services.
5427         allowed = 'O:SYD:(A;;CR;;;WD)'
5428         policy_id = self.get_new_username()
5429         policy = self.create_authn_policy(policy_id,
5430                                           enforced=True,
5431                                           service_allowed_ntlm=False,
5432                                           service_allowed_from=allowed)
5433
5434         # Create a managed service account with the assigned policy.
5435         client_creds = self._get_creds(
5436             account_type=self.AccountType.MANAGED_SERVICE,
5437             assigned_policy=policy,
5438             ntlm=True,
5439             cached=False)
5440
5441         # Show that a network SamLogon fails.
5442         self._test_samlogon(
5443             creds=client_creds,
5444             logon_type=netlogon.NetlogonNetworkInformation,
5445             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5446
5447     def test_authn_policy_samlogon_network_deny_service_not_allowed_from(self):
5448         # Create an authentication policy disallowing NTLM authentication for
5449         # services.
5450         denied = 'O:SYD:(D;;CR;;;WD)'
5451         policy_id = self.get_new_username()
5452         policy = self.create_authn_policy(policy_id,
5453                                           enforced=True,
5454                                           service_allowed_ntlm=False,
5455                                           service_allowed_from=denied)
5456
5457         # Create a managed service account with the assigned policy.
5458         client_creds = self._get_creds(
5459             account_type=self.AccountType.MANAGED_SERVICE,
5460             assigned_policy=policy,
5461             ntlm=True,
5462             cached=False)
5463
5464         # Show that a network SamLogon fails.
5465         self._test_samlogon(
5466             creds=client_creds,
5467             logon_type=netlogon.NetlogonNetworkInformation,
5468             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5469
5470     def test_authn_policy_samlogon_network_deny_service_no_allowed_from(self):
5471         # Create an authentication policy disallowing NTLM authentication for
5472         # services.
5473         policy_id = self.get_new_username()
5474         policy = self.create_authn_policy(policy_id,
5475                                           enforced=True,
5476                                           service_allowed_ntlm=False)
5477
5478         # Create a managed service account with the assigned policy.
5479         client_creds = self._get_creds(
5480             account_type=self.AccountType.MANAGED_SERVICE,
5481             assigned_policy=policy,
5482             ntlm=True,
5483             cached=False)
5484
5485         # Show that a network SamLogon succeeds.
5486         self._test_samlogon(creds=client_creds,
5487                             logon_type=netlogon.NetlogonNetworkInformation)
5488
5489     def test_authn_policy_samlogon_network_allow_service_allowed_from_to_self(self):
5490         # Create an authentication policy allowing NTLM authentication for
5491         # services.
5492         allowed = 'O:SYD:(A;;CR;;;WD)'
5493         policy_id = self.get_new_username()
5494         policy = self.create_authn_policy(policy_id,
5495                                           enforced=True,
5496                                           service_allowed_ntlm=True,
5497                                           service_allowed_from=allowed)
5498
5499         # Create a managed service account with the assigned policy.
5500         client_creds = self._get_creds(
5501             account_type=self.AccountType.MANAGED_SERVICE,
5502             assigned_policy=policy,
5503             ntlm=True,
5504             cached=False)
5505
5506         # Show that a network SamLogon to ourselves succeeds.
5507         self._test_samlogon(creds=client_creds,
5508                             domain_joined_mach_creds=client_creds,
5509                             logon_type=netlogon.NetlogonNetworkInformation)
5510
5511     def test_authn_policy_samlogon_network_allow_service_not_allowed_from_to_self(self):
5512         # Create an authentication policy allowing NTLM authentication for
5513         # services.
5514         denied = 'O:SYD:(D;;CR;;;WD)'
5515         policy_id = self.get_new_username()
5516         policy = self.create_authn_policy(policy_id,
5517                                           enforced=True,
5518                                           service_allowed_ntlm=True,
5519                                           service_allowed_from=denied)
5520
5521         # Create a managed service account with the assigned policy.
5522         client_creds = self._get_creds(
5523             account_type=self.AccountType.MANAGED_SERVICE,
5524             assigned_policy=policy,
5525             ntlm=True,
5526             cached=False)
5527
5528         # Show that a network SamLogon to ourselves succeeds.
5529         self._test_samlogon(creds=client_creds,
5530                             domain_joined_mach_creds=client_creds,
5531                             logon_type=netlogon.NetlogonNetworkInformation)
5532
5533     def test_authn_policy_samlogon_network_allow_service_no_allowed_from_to_self(self):
5534         # Create an authentication policy allowing NTLM authentication for
5535         # services.
5536         policy_id = self.get_new_username()
5537         policy = self.create_authn_policy(policy_id,
5538                                           enforced=True,
5539                                           service_allowed_ntlm=True)
5540
5541         # Create a managed service account with the assigned policy.
5542         client_creds = self._get_creds(
5543             account_type=self.AccountType.MANAGED_SERVICE,
5544             assigned_policy=policy,
5545             ntlm=True,
5546             cached=False)
5547
5548         # Show that a network SamLogon to ourselves succeeds.
5549         self._test_samlogon(creds=client_creds,
5550                             domain_joined_mach_creds=client_creds,
5551                             logon_type=netlogon.NetlogonNetworkInformation)
5552
5553     def test_authn_policy_samlogon_network_deny_service_allowed_from_to_self(self):
5554         # Create an authentication policy disallowing NTLM authentication for
5555         # services.
5556         allowed = 'O:SYD:(A;;CR;;;WD)'
5557         policy_id = self.get_new_username()
5558         policy = self.create_authn_policy(policy_id,
5559                                           enforced=True,
5560                                           service_allowed_ntlm=False,
5561                                           service_allowed_from=allowed)
5562
5563         # Create a managed service account with the assigned policy.
5564         client_creds = self._get_creds(
5565             account_type=self.AccountType.MANAGED_SERVICE,
5566             assigned_policy=policy,
5567             ntlm=True,
5568             cached=False)
5569
5570         # Show that a network SamLogon to ourselves fails.
5571         self._test_samlogon(
5572             creds=client_creds,
5573             domain_joined_mach_creds=client_creds,
5574             logon_type=netlogon.NetlogonNetworkInformation,
5575             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5576
5577     def test_authn_policy_samlogon_network_deny_service_not_allowed_from_to_self(self):
5578         # Create an authentication policy disallowing NTLM authentication for
5579         # services.
5580         denied = 'O:SYD:(D;;CR;;;WD)'
5581         policy_id = self.get_new_username()
5582         policy = self.create_authn_policy(policy_id,
5583                                           enforced=True,
5584                                           service_allowed_ntlm=False,
5585                                           service_allowed_from=denied)
5586
5587         # Create a managed service account with the assigned policy.
5588         client_creds = self._get_creds(
5589             account_type=self.AccountType.MANAGED_SERVICE,
5590             assigned_policy=policy,
5591             ntlm=True,
5592             cached=False)
5593
5594         # Show that a network SamLogon to ourselves fails.
5595         self._test_samlogon(
5596             creds=client_creds,
5597             domain_joined_mach_creds=client_creds,
5598             logon_type=netlogon.NetlogonNetworkInformation,
5599             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5600
5601     def test_authn_policy_samlogon_network_deny_service_no_allowed_from_to_self(self):
5602         # Create an authentication policy disallowing NTLM authentication for
5603         # services.
5604         policy_id = self.get_new_username()
5605         policy = self.create_authn_policy(policy_id,
5606                                           enforced=True,
5607                                           service_allowed_ntlm=False)
5608
5609         # Create a managed service account with the assigned policy.
5610         client_creds = self._get_creds(
5611             account_type=self.AccountType.MANAGED_SERVICE,
5612             assigned_policy=policy,
5613             ntlm=True,
5614             cached=False)
5615
5616         # Show that a network SamLogon to ourselves succeeds.
5617         self._test_samlogon(creds=client_creds,
5618                             domain_joined_mach_creds=client_creds,
5619                             logon_type=netlogon.NetlogonNetworkInformation)
5620
5621     def test_authn_policy_samlogon_interactive_deny_no_device_restrictions(self):
5622         # Create an authentication policy denying NTLM authentication for
5623         # users.
5624         policy_id = self.get_new_username()
5625         policy = self.create_authn_policy(policy_id,
5626                                           enforced=True,
5627                                           user_allowed_ntlm=False,
5628                                           service_allowed_ntlm=True)
5629
5630         # Create a user account with the assigned policy.
5631         client_creds = self._get_creds(account_type=self.AccountType.USER,
5632                                        assigned_policy=policy,
5633                                        ntlm=True)
5634
5635         # Show that without AllowedToAuthenticateFrom set in the policy, an
5636         # interactive SamLogon succeeds.
5637         self._test_samlogon(creds=client_creds,
5638                             logon_type=netlogon.NetlogonInteractiveInformation)
5639
5640     def test_authn_policy_samlogon_network_deny_no_device_restrictions(self):
5641         # Create an authentication policy denying NTLM authentication for
5642         # users.
5643         policy_id = self.get_new_username()
5644         policy = self.create_authn_policy(policy_id,
5645                                           enforced=True,
5646                                           user_allowed_ntlm=False,
5647                                           service_allowed_ntlm=True)
5648
5649         # Create a user account with the assigned policy.
5650         client_creds = self._get_creds(account_type=self.AccountType.USER,
5651                                        assigned_policy=policy,
5652                                        ntlm=True)
5653
5654         # Show that without AllowedToAuthenticateFrom set in the policy, a
5655         # network SamLogon succeeds.
5656         self._test_samlogon(creds=client_creds,
5657                             logon_type=netlogon.NetlogonNetworkInformation)
5658
5659     def test_samlogon_allowed_to_computer_allow(self):
5660         samdb = self.get_samdb()
5661
5662         # Create a user account.
5663         client_creds = self._get_creds(account_type=self.AccountType.USER,
5664                                        ntlm=True)
5665         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
5666
5667         # Create an authentication policy that applies to a computer and
5668         # explicitly allows the user account to obtain a service ticket.
5669         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
5670         denied = 'O:SYD:(D;;CR;;;WD)'
5671         policy_id = self.get_new_username()
5672         policy = self.create_authn_policy(policy_id,
5673                                           enforced=True,
5674                                           user_allowed_to=denied,
5675                                           computer_allowed_to=allowed,
5676                                           service_allowed_to=denied)
5677
5678         # Create a computer account with the assigned policy.
5679         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5680                                        assigned_policy=policy)
5681
5682         # Show that a network SamLogon succeeds.
5683         self._test_samlogon(creds=client_creds,
5684                             domain_joined_mach_creds=target_creds,
5685                             logon_type=netlogon.NetlogonNetworkInformation)
5686
5687         # Show that an interactive SamLogon succeeds.
5688         self._test_samlogon(creds=client_creds,
5689                             domain_joined_mach_creds=target_creds,
5690                             logon_type=netlogon.NetlogonInteractiveInformation)
5691
5692     def test_samlogon_allowed_to_computer_deny(self):
5693         samdb = self.get_samdb()
5694
5695         # Create a user account.
5696         client_creds = self._get_creds(account_type=self.AccountType.USER,
5697                                        ntlm=True)
5698         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
5699
5700         # Create an authentication policy that applies to a computer and
5701         # explicitly denies the user account to obtain a service ticket.
5702         denied = f'O:SYD:(D;;CR;;;{client_sid})'
5703         allowed = 'O:SYD:(A;;CR;;;WD)'
5704         policy_id = self.get_new_username()
5705         policy = self.create_authn_policy(policy_id,
5706                                           enforced=True,
5707                                           user_allowed_to=allowed,
5708                                           computer_allowed_to=denied,
5709                                           service_allowed_to=allowed)
5710
5711         # Create a computer account with the assigned policy.
5712         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5713                                        assigned_policy=policy)
5714
5715         # Show that a network SamLogon fails.
5716         self._test_samlogon(
5717             creds=client_creds,
5718             domain_joined_mach_creds=target_creds,
5719             logon_type=netlogon.NetlogonNetworkInformation,
5720             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5721
5722         # Show that an interactive SamLogon fails.
5723         self._test_samlogon(
5724             creds=client_creds,
5725             domain_joined_mach_creds=target_creds,
5726             logon_type=netlogon.NetlogonInteractiveInformation,
5727             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5728
5729     def test_samlogon_allowed_to_computer_deny_protected(self):
5730         samdb = self.get_samdb()
5731
5732         # Create a protected user account.
5733         client_creds = self._get_creds(account_type=self.AccountType.USER,
5734                                        protected=True,
5735                                        ntlm=True)
5736         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
5737
5738         # Create an authentication policy that applies to a computer and
5739         # explicitly denies the user account to obtain a service ticket.
5740         denied = f'O:SYD:(D;;CR;;;{client_sid})'
5741         allowed = 'O:SYD:(A;;CR;;;WD)'
5742         policy_id = self.get_new_username()
5743         policy = self.create_authn_policy(policy_id,
5744                                           enforced=True,
5745                                           user_allowed_to=allowed,
5746                                           computer_allowed_to=denied,
5747                                           service_allowed_to=allowed)
5748
5749         # Create a computer account with the assigned policy.
5750         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5751                                        assigned_policy=policy)
5752
5753         # Show that a network SamLogon fails.
5754         self._test_samlogon(
5755             creds=client_creds,
5756             domain_joined_mach_creds=target_creds,
5757             logon_type=netlogon.NetlogonNetworkInformation,
5758             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5759
5760         # Show that an interactive SamLogon fails.
5761         self._test_samlogon(
5762             creds=client_creds,
5763             domain_joined_mach_creds=target_creds,
5764             logon_type=netlogon.NetlogonInteractiveInformation,
5765             expect_error=ntstatus.NT_STATUS_ACCOUNT_RESTRICTION)
5766
5767     def test_samlogon_allowed_to_computer_allow_asserted_identity(self):
5768         # Create a user account.
5769         client_creds = self._get_creds(account_type=self.AccountType.USER,
5770                                        ntlm=True)
5771
5772         # Create an authentication policy that allows accounts with the
5773         # Authentication Authority Asserted Identity SID to obtain a service
5774         # ticket.
5775         allowed = (
5776             f'O:SYD:(A;;CR;;;'
5777             f'{security.SID_AUTHENTICATION_AUTHORITY_ASSERTED_IDENTITY})'
5778         )
5779         denied = 'O:SYD:(D;;CR;;;WD)'
5780         policy_id = self.get_new_username()
5781         policy = self.create_authn_policy(policy_id,
5782                                           enforced=True,
5783                                           user_allowed_to=denied,
5784                                           computer_allowed_to=allowed,
5785                                           service_allowed_to=denied)
5786
5787         # Create a computer account with the assigned policy.
5788         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5789                                        assigned_policy=policy)
5790
5791         # Show that a network SamLogon fails.
5792         self._test_samlogon(
5793             creds=client_creds,
5794             domain_joined_mach_creds=target_creds,
5795             logon_type=netlogon.NetlogonNetworkInformation,
5796             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5797
5798         # Show that an interactive SamLogon fails.
5799         self._test_samlogon(
5800             creds=client_creds,
5801             domain_joined_mach_creds=target_creds,
5802             logon_type=netlogon.NetlogonInteractiveInformation,
5803             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5804
5805     def test_samlogon_allowed_to_computer_allow_claims_valid(self):
5806         # Create a user account.
5807         client_creds = self._get_creds(account_type=self.AccountType.USER,
5808                                        ntlm=True)
5809
5810         # Create an authentication policy that allows accounts with the Claims
5811         # Valid SID to obtain a service ticket.
5812         allowed = f'O:SYD:(A;;CR;;;{security.SID_CLAIMS_VALID})'
5813         denied = 'O:SYD:(D;;CR;;;WD)'
5814         policy_id = self.get_new_username()
5815         policy = self.create_authn_policy(policy_id,
5816                                           enforced=True,
5817                                           user_allowed_to=denied,
5818                                           computer_allowed_to=allowed,
5819                                           service_allowed_to=denied)
5820
5821         # Create a computer account with the assigned policy.
5822         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5823                                        assigned_policy=policy)
5824
5825         # Show that a network SamLogon fails.
5826         self._test_samlogon(
5827             creds=client_creds,
5828             domain_joined_mach_creds=target_creds,
5829             logon_type=netlogon.NetlogonNetworkInformation,
5830             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5831
5832         # Show that an interactive SamLogon fails.
5833         self._test_samlogon(
5834             creds=client_creds,
5835             domain_joined_mach_creds=target_creds,
5836             logon_type=netlogon.NetlogonInteractiveInformation,
5837             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5838
5839     def test_samlogon_allowed_to_computer_allow_compounded_auth(self):
5840         # Create a user account.
5841         client_creds = self._get_creds(account_type=self.AccountType.USER,
5842                                        ntlm=True)
5843
5844         # Create an authentication policy that allows accounts with the
5845         # Compounded Authentication SID to obtain a service ticket.
5846         allowed = f'O:SYD:(A;;CR;;;{security.SID_COMPOUNDED_AUTHENTICATION})'
5847         denied = 'O:SYD:(D;;CR;;;WD)'
5848         policy_id = self.get_new_username()
5849         policy = self.create_authn_policy(policy_id,
5850                                           enforced=True,
5851                                           user_allowed_to=denied,
5852                                           computer_allowed_to=allowed,
5853                                           service_allowed_to=denied)
5854
5855         # Create a computer account with the assigned policy.
5856         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5857                                        assigned_policy=policy)
5858
5859         # Show that a network SamLogon fails.
5860         self._test_samlogon(
5861             creds=client_creds,
5862             domain_joined_mach_creds=target_creds,
5863             logon_type=netlogon.NetlogonNetworkInformation,
5864             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5865
5866         # Show that an interactive SamLogon fails.
5867         self._test_samlogon(
5868             creds=client_creds,
5869             domain_joined_mach_creds=target_creds,
5870             logon_type=netlogon.NetlogonInteractiveInformation,
5871             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5872
5873     def test_samlogon_allowed_to_computer_allow_authenticated_users(self):
5874         # Create a user account.
5875         client_creds = self._get_creds(account_type=self.AccountType.USER,
5876                                        ntlm=True)
5877
5878         # Create an authentication policy that allows accounts with the
5879         # Authenticated Users SID to obtain a service ticket.
5880         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_AUTHENTICATED_USERS})'
5881         denied = 'O:SYD:(D;;CR;;;WD)'
5882         policy_id = self.get_new_username()
5883         policy = self.create_authn_policy(policy_id,
5884                                           enforced=True,
5885                                           user_allowed_to=denied,
5886                                           computer_allowed_to=allowed,
5887                                           service_allowed_to=denied)
5888
5889         # Create a computer account with the assigned policy.
5890         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5891                                        assigned_policy=policy)
5892
5893         # Show that a network SamLogon succeeds.
5894         self._test_samlogon(creds=client_creds,
5895                             domain_joined_mach_creds=target_creds,
5896                             logon_type=netlogon.NetlogonNetworkInformation)
5897
5898         # Show that an interactive SamLogon succeeds.
5899         self._test_samlogon(creds=client_creds,
5900                             domain_joined_mach_creds=target_creds,
5901                             logon_type=netlogon.NetlogonInteractiveInformation)
5902
5903     def test_samlogon_allowed_to_computer_allow_ntlm_authn(self):
5904         # Create a user account.
5905         client_creds = self._get_creds(account_type=self.AccountType.USER,
5906                                        ntlm=True)
5907
5908         # Create an authentication policy that allows accounts with the NTLM
5909         # Authentication SID to obtain a service ticket.
5910         allowed = f'O:SYD:(A;;CR;;;{security.SID_NT_NTLM_AUTHENTICATION})'
5911         denied = 'O:SYD:(D;;CR;;;WD)'
5912         policy_id = self.get_new_username()
5913         policy = self.create_authn_policy(policy_id,
5914                                           enforced=True,
5915                                           user_allowed_to=denied,
5916                                           computer_allowed_to=allowed,
5917                                           service_allowed_to=denied)
5918
5919         # Create a computer account with the assigned policy.
5920         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5921                                        assigned_policy=policy)
5922
5923         # Show that a network SamLogon fails.
5924         self._test_samlogon(
5925             creds=client_creds,
5926             domain_joined_mach_creds=target_creds,
5927             logon_type=netlogon.NetlogonNetworkInformation,
5928             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5929
5930         # Show that an interactive SamLogon fails.
5931         self._test_samlogon(
5932             creds=client_creds,
5933             domain_joined_mach_creds=target_creds,
5934             logon_type=netlogon.NetlogonInteractiveInformation,
5935             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
5936
5937     def test_samlogon_allowed_to_no_owner(self):
5938         samdb = self.get_samdb()
5939
5940         # Create a user account.
5941         client_creds = self._get_creds(account_type=self.AccountType.USER,
5942                                        ntlm=True)
5943         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
5944
5945         # Create an authentication policy that applies to a computer and
5946         # explicitly allows the user account to obtain a service ticket. Omit
5947         # the owner (O:SY) from the SDDL.
5948         allowed = f'D:(A;;CR;;;{client_sid})'
5949         policy_id = self.get_new_username()
5950         policy = self.create_authn_policy(policy_id,
5951                                           enforced=True,
5952                                           computer_allowed_to=allowed)
5953
5954         # Create a computer account with the assigned policy.
5955         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5956                                        assigned_policy=policy)
5957
5958         # Show that a network SamLogon fails.
5959         self._test_samlogon(
5960             creds=client_creds,
5961             domain_joined_mach_creds=target_creds,
5962             logon_type=netlogon.NetlogonNetworkInformation,
5963             expect_error=ntstatus.NT_STATUS_INVALID_PARAMETER)
5964
5965         # Show that an interactive SamLogon fails.
5966         self._test_samlogon(
5967             creds=client_creds,
5968             domain_joined_mach_creds=target_creds,
5969             logon_type=netlogon.NetlogonInteractiveInformation,
5970             expect_error=ntstatus.NT_STATUS_INVALID_PARAMETER)
5971
5972     def test_samlogon_allowed_to_no_owner_unenforced(self):
5973         samdb = self.get_samdb()
5974
5975         # Create a user account.
5976         client_creds = self._get_creds(account_type=self.AccountType.USER,
5977                                        ntlm=True)
5978         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
5979
5980         # Create an unenforced authentication policy that applies to a computer
5981         # and explicitly allows the user account to obtain a service
5982         # ticket. Omit the owner (O:SY) from the SDDL.
5983         allowed = f'D:(A;;CR;;;{client_sid})'
5984         policy_id = self.get_new_username()
5985         policy = self.create_authn_policy(policy_id,
5986                                           enforced=False,
5987                                           computer_allowed_to=allowed)
5988
5989         # Create a computer account with the assigned policy.
5990         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
5991                                        assigned_policy=policy)
5992
5993         # Show that a network SamLogon succeeds.
5994         self._test_samlogon(creds=client_creds,
5995                             domain_joined_mach_creds=target_creds,
5996                             logon_type=netlogon.NetlogonNetworkInformation)
5997
5998         # Show that an interactive SamLogon succeeds.
5999         self._test_samlogon(creds=client_creds,
6000                             domain_joined_mach_creds=target_creds,
6001                             logon_type=netlogon.NetlogonInteractiveInformation)
6002
6003     def test_samlogon_allowed_to_service_allow(self):
6004         samdb = self.get_samdb()
6005
6006         # Create a user account.
6007         client_creds = self._get_creds(account_type=self.AccountType.USER,
6008                                        ntlm=True)
6009         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
6010
6011         # Create an authentication policy that applies to a managed service and
6012         # explicitly allows the user account to obtain a service ticket.
6013         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
6014         denied = 'O:SYD:(D;;CR;;;WD)'
6015         policy_id = self.get_new_username()
6016         policy = self.create_authn_policy(policy_id,
6017                                           enforced=True,
6018                                           user_allowed_to=denied,
6019                                           computer_allowed_to=denied,
6020                                           service_allowed_to=allowed)
6021
6022         # Create a managed service account with the assigned policy.
6023         target_creds = self._get_creds(
6024             account_type=self.AccountType.MANAGED_SERVICE,
6025             assigned_policy=policy)
6026
6027         # Show that a network SamLogon succeeds.
6028         self._test_samlogon(creds=client_creds,
6029                             domain_joined_mach_creds=target_creds,
6030                             logon_type=netlogon.NetlogonNetworkInformation)
6031
6032         # Show that an interactive SamLogon succeeds.
6033         self._test_samlogon(creds=client_creds,
6034                             domain_joined_mach_creds=target_creds,
6035                             logon_type=netlogon.NetlogonInteractiveInformation)
6036
6037     def test_samlogon_allowed_to_service_deny(self):
6038         samdb = self.get_samdb()
6039
6040         # Create a user account.
6041         client_creds = self._get_creds(account_type=self.AccountType.USER,
6042                                        ntlm=True)
6043         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
6044
6045         # Create an authentication policy that applies to a managed service and
6046         # explicitly denies the user account to obtain a service ticket.
6047         denied = f'O:SYD:(D;;CR;;;{client_sid})'
6048         allowed = 'O:SYD:(A;;CR;;;WD)'
6049         policy_id = self.get_new_username()
6050         policy = self.create_authn_policy(policy_id,
6051                                           enforced=True,
6052                                           user_allowed_to=allowed,
6053                                           computer_allowed_to=allowed,
6054                                           service_allowed_to=denied)
6055
6056         # Create a managed service account with the assigned policy.
6057         target_creds = self._get_creds(
6058             account_type=self.AccountType.MANAGED_SERVICE,
6059             assigned_policy=policy)
6060
6061         # Show that a network SamLogon fails.
6062         self._test_samlogon(
6063             creds=client_creds,
6064             domain_joined_mach_creds=target_creds,
6065             logon_type=netlogon.NetlogonNetworkInformation,
6066             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
6067
6068         # Show that an interactive SamLogon fails.
6069         self._test_samlogon(
6070             creds=client_creds,
6071             domain_joined_mach_creds=target_creds,
6072             logon_type=netlogon.NetlogonInteractiveInformation,
6073             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
6074
6075     def test_samlogon_allowed_to_computer_allow_group_not_a_member(self):
6076         samdb = self.get_samdb()
6077
6078         # Create a new group.
6079         group_name = self.get_new_username()
6080         group_dn = self.create_group(samdb, group_name)
6081         group_sid = self.get_objectSid(samdb, group_dn)
6082
6083         # Create a user account which does not belong to the group.
6084         client_creds = self._get_creds(account_type=self.AccountType.USER,
6085                                        ntlm=True)
6086
6087         # Create an authentication policy that allows accounts belonging to the
6088         # group.
6089         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
6090         policy_id = self.get_new_username()
6091         policy = self.create_authn_policy(policy_id,
6092                                           enforced=True,
6093                                           computer_allowed_to=allowed)
6094
6095         # Create a computer account with the assigned policy.
6096         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
6097                                        assigned_policy=policy)
6098
6099         # Show that a network SamLogon fails, as the user account does not
6100         # belong to the group.
6101         self._test_samlogon(
6102             creds=client_creds,
6103             domain_joined_mach_creds=target_creds,
6104             logon_type=netlogon.NetlogonNetworkInformation,
6105             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
6106
6107         # Show that an interactive SamLogon fails, as the user account does not
6108         # belong to the group.
6109         self._test_samlogon(
6110             creds=client_creds,
6111             domain_joined_mach_creds=target_creds,
6112             logon_type=netlogon.NetlogonInteractiveInformation,
6113             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
6114
6115     def test_samlogon_allowed_to_computer_allow_group_member(self):
6116         samdb = self.get_samdb()
6117
6118         # Create a new group.
6119         group_name = self.get_new_username()
6120         group_dn = self.create_group(samdb, group_name)
6121         group_sid = self.get_objectSid(samdb, group_dn)
6122
6123         # Create a user account that belongs to the group.
6124         client_creds = self._get_creds(account_type=self.AccountType.USER,
6125                                        member_of=group_dn,
6126                                        ntlm=True)
6127
6128         # Create an authentication policy that allows accounts belonging to the
6129         # group.
6130         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
6131         policy_id = self.get_new_username()
6132         policy = self.create_authn_policy(policy_id,
6133                                           enforced=True,
6134                                           computer_allowed_to=allowed)
6135
6136         # Create a computer account with the assigned policy.
6137         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
6138                                        assigned_policy=policy)
6139
6140         # Show that a network SamLogon succeeds, since the user account belongs
6141         # to the group.
6142         self._test_samlogon(creds=client_creds,
6143                             domain_joined_mach_creds=target_creds,
6144                             logon_type=netlogon.NetlogonNetworkInformation)
6145
6146         # Show that an interactive SamLogon succeeds, since the user account
6147         # belongs to the group.
6148         self._test_samlogon(creds=client_creds,
6149                             domain_joined_mach_creds=target_creds,
6150                             logon_type=netlogon.NetlogonInteractiveInformation)
6151
6152     def test_samlogon_allowed_to_computer_allow_domain_local_group(self):
6153         samdb = self.get_samdb()
6154
6155         # Create a new domain-local group.
6156         group_name = self.get_new_username()
6157         group_dn = self.create_group(samdb, group_name,
6158                                      gtype=GroupType.DOMAIN_LOCAL.value)
6159         group_sid = self.get_objectSid(samdb, group_dn)
6160
6161         # Create a user account that belongs to the group.
6162         client_creds = self._get_creds(account_type=self.AccountType.USER,
6163                                        member_of=group_dn,
6164                                        ntlm=True)
6165
6166         # Create an authentication policy that allows accounts belonging to the
6167         # group.
6168         allowed = f'O:SYD:(A;;CR;;;{group_sid})'
6169         policy_id = self.get_new_username()
6170         policy = self.create_authn_policy(policy_id,
6171                                           enforced=True,
6172                                           computer_allowed_to=allowed)
6173
6174         # Create a computer account with the assigned policy.
6175         target_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
6176                                        assigned_policy=policy)
6177
6178         # Show that a network SamLogon succeeds, since the user account belongs
6179         # to the group.
6180         self._test_samlogon(creds=client_creds,
6181                             domain_joined_mach_creds=target_creds,
6182                             logon_type=netlogon.NetlogonNetworkInformation)
6183
6184         # Show that an interactive SamLogon succeeds, since the user account
6185         # belongs to the group.
6186         self._test_samlogon(creds=client_creds,
6187                             domain_joined_mach_creds=target_creds,
6188                             logon_type=netlogon.NetlogonInteractiveInformation)
6189
6190     def test_samlogon_allowed_to_computer_allow_to_self(self):
6191         samdb = self.get_samdb()
6192
6193         # Create a computer account.
6194         client_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
6195                                        ntlm=True,
6196                                        cached=False)
6197         client_dn = client_creds.get_dn()
6198         client_sid = self.get_objectSid(samdb, client_dn)
6199
6200         # Create an authentication policy that applies to a computer and
6201         # explicitly allows the user account to obtain a service ticket.
6202         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
6203         denied = 'O:SYD:(D;;CR;;;WD)'
6204         policy_id = self.get_new_username()
6205         policy = self.create_authn_policy(policy_id,
6206                                           enforced=True,
6207                                           user_allowed_to=denied,
6208                                           computer_allowed_to=allowed,
6209                                           service_allowed_to=denied)
6210
6211         # Assign the policy to the account.
6212         self.add_attribute(samdb, str(client_dn),
6213                            'msDS-AssignedAuthNPolicy', str(policy))
6214
6215         # Show that a network SamLogon to ourselves succeeds.
6216         self._test_samlogon(creds=client_creds,
6217                             domain_joined_mach_creds=client_creds,
6218                             logon_type=netlogon.NetlogonNetworkInformation)
6219
6220     def test_samlogon_allowed_to_computer_deny_to_self(self):
6221         samdb = self.get_samdb()
6222
6223         # Create a computer account.
6224         client_creds = self._get_creds(account_type=self.AccountType.COMPUTER,
6225                                        ntlm=True,
6226                                        cached=False)
6227         client_dn = client_creds.get_dn()
6228         client_sid = self.get_objectSid(samdb, client_dn)
6229
6230         # Create an authentication policy that applies to a computer and
6231         # explicitly denies the user account to obtain a service ticket.
6232         denied = f'O:SYD:(D;;CR;;;{client_sid})'
6233         allowed = 'O:SYD:(A;;CR;;;WD)'
6234         policy_id = self.get_new_username()
6235         policy = self.create_authn_policy(policy_id,
6236                                           enforced=True,
6237                                           user_allowed_to=allowed,
6238                                           computer_allowed_to=denied,
6239                                           service_allowed_to=allowed)
6240
6241         # Assign the policy to the account.
6242         self.add_attribute(samdb, str(client_dn),
6243                            'msDS-AssignedAuthNPolicy', str(policy))
6244
6245         # Show that a network SamLogon to ourselves fails, despite
6246         # authentication being allowed in the Kerberos case.
6247         self._test_samlogon(
6248             creds=client_creds,
6249             domain_joined_mach_creds=client_creds,
6250             logon_type=netlogon.NetlogonNetworkInformation,
6251             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
6252
6253     def test_samlogon_allowed_to_service_allow_to_self(self):
6254         samdb = self.get_samdb()
6255
6256         # Create a managed service account.
6257         client_creds = self._get_creds(
6258             account_type=self.AccountType.MANAGED_SERVICE,
6259             ntlm=True,
6260             cached=False)
6261         client_dn = client_creds.get_dn()
6262         client_sid = self.get_objectSid(samdb, client_dn)
6263
6264         # Create an authentication policy that applies to a managed service and
6265         # explicitly allows the user account to obtain a service ticket.
6266         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
6267         denied = 'O:SYD:(D;;CR;;;WD)'
6268         policy_id = self.get_new_username()
6269         policy = self.create_authn_policy(policy_id,
6270                                           enforced=True,
6271                                           user_allowed_to=denied,
6272                                           computer_allowed_to=denied,
6273                                           service_allowed_to=allowed)
6274
6275         # Assign the policy to the account.
6276         self.add_attribute(samdb, str(client_dn),
6277                            'msDS-AssignedAuthNPolicy', str(policy))
6278
6279         # Show that a network SamLogon to ourselves succeeds.
6280         self._test_samlogon(creds=client_creds,
6281                             domain_joined_mach_creds=client_creds,
6282                             logon_type=netlogon.NetlogonNetworkInformation)
6283
6284     def test_samlogon_allowed_to_service_deny_to_self(self):
6285         samdb = self.get_samdb()
6286
6287         # Create a managed service account.
6288         client_creds = self._get_creds(
6289             account_type=self.AccountType.MANAGED_SERVICE,
6290             ntlm=True,
6291             cached=False)
6292         client_dn = client_creds.get_dn()
6293         client_sid = self.get_objectSid(samdb, client_dn)
6294
6295         # Create an authentication policy that applies to a managed service and
6296         # explicitly denies the user account to obtain a service ticket.
6297         denied = f'O:SYD:(D;;CR;;;{client_sid})'
6298         allowed = 'O:SYD:(A;;CR;;;WD)'
6299         policy_id = self.get_new_username()
6300         policy = self.create_authn_policy(policy_id,
6301                                           enforced=True,
6302                                           user_allowed_to=allowed,
6303                                           computer_allowed_to=allowed,
6304                                           service_allowed_to=denied)
6305
6306         # Assign the policy to the account.
6307         self.add_attribute(samdb, str(client_dn),
6308                            'msDS-AssignedAuthNPolicy', str(policy))
6309
6310         # Show that a network SamLogon to ourselves fails, despite
6311         # authentication being allowed in the Kerberos case.
6312         self._test_samlogon(
6313             creds=client_creds,
6314             domain_joined_mach_creds=client_creds,
6315             logon_type=netlogon.NetlogonNetworkInformation,
6316             expect_error=ntstatus.NT_STATUS_AUTHENTICATION_FIREWALL_FAILED)
6317
6318     def test_samlogon_allowed_to_computer_derived_class_allow(self):
6319         samdb = self.get_samdb()
6320
6321         # Create a user account.
6322         client_creds = self._get_creds(account_type=self.AccountType.USER,
6323                                        ntlm=True)
6324         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
6325
6326         # Create an authentication policy that applies to a computer and
6327         # explicitly allows the user account to obtain a service ticket.
6328         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
6329         denied = 'O:SYD:(D;;CR;;;WD)'
6330         policy_id = self.get_new_username()
6331         policy = self.create_authn_policy(policy_id,
6332                                           enforced=True,
6333                                           user_allowed_to=denied,
6334                                           computer_allowed_to=allowed,
6335                                           service_allowed_to=denied)
6336
6337         # Create a schema class derived from ‘computer’.
6338         class_id = random.randint(0, 100000000)
6339         computer_class_cn = f'my-Computer-Class-{class_id}'
6340         computer_class = computer_class_cn.replace('-', '')
6341         class_dn = samdb.get_schema_basedn()
6342         class_dn.add_child(f'CN={computer_class_cn}')
6343         governs_id = f'1.3.6.1.4.1.7165.4.6.2.9.{class_id}'
6344
6345         samdb.add({
6346             'dn': class_dn,
6347             'objectClass': 'classSchema',
6348             'subClassOf': 'computer',
6349             'governsId': governs_id,
6350             'lDAPDisplayName': computer_class,
6351         })
6352
6353         # Create an account derived from ‘computer’ with the assigned policy.
6354         target_name = self.get_new_username()
6355         target_creds, target_dn = self.create_account(
6356             samdb, target_name,
6357             account_type=self.AccountType.COMPUTER,
6358             spn=f'host/{target_name}',
6359             additional_details={
6360                 'msDS-AssignedAuthNPolicy': str(policy),
6361                 'objectClass': computer_class,
6362             })
6363
6364         keys = self.get_keys(target_creds)
6365         self.creds_set_keys(target_creds, keys)
6366
6367         # Show that a network SamLogon succeeds.
6368         self._test_samlogon(creds=client_creds,
6369                             domain_joined_mach_creds=target_creds,
6370                             logon_type=netlogon.NetlogonNetworkInformation)
6371
6372         # Show that an interactive SamLogon succeeds.
6373         self._test_samlogon(creds=client_creds,
6374                             domain_joined_mach_creds=target_creds,
6375                             logon_type=netlogon.NetlogonInteractiveInformation)
6376
6377     def test_samlogon_allowed_to_service_derived_class_allow(self):
6378         samdb = self.get_samdb()
6379
6380         # Create a user account.
6381         client_creds = self._get_creds(account_type=self.AccountType.USER,
6382                                        ntlm=True)
6383         client_sid = self.get_objectSid(samdb, client_creds.get_dn())
6384
6385         # Create an authentication policy that applies to a managed service and
6386         # explicitly allows the user account to obtain a service ticket.
6387         allowed = f'O:SYD:(A;;CR;;;{client_sid})'
6388         denied = 'O:SYD:(D;;CR;;;WD)'
6389         policy_id = self.get_new_username()
6390         policy = self.create_authn_policy(policy_id,
6391                                           enforced=True,
6392                                           user_allowed_to=denied,
6393                                           computer_allowed_to=denied,
6394                                           service_allowed_to=allowed)
6395
6396         # Create a schema class derived from ‘msDS-ManagedServiceAccount’.
6397         class_id = random.randint(0, 100000000)
6398         service_class_cn = f'my-Managed-Service-Class-{class_id}'
6399         service_class = service_class_cn.replace('-', '')
6400         class_dn = samdb.get_schema_basedn()
6401         class_dn.add_child(f'CN={service_class_cn}')
6402         governs_id = f'1.3.6.1.4.1.7165.4.6.2.9.{class_id}'
6403
6404         samdb.add({
6405             'dn': class_dn,
6406             'objectClass': 'classSchema',
6407             'subClassOf': 'msDS-ManagedServiceAccount',
6408             'governsId': governs_id,
6409             'lDAPDisplayName': service_class,
6410         })
6411
6412         # Create an account derived from ‘msDS-ManagedServiceAccount’ with the
6413         # assigned policy.
6414         target_name = self.get_new_username()
6415         target_creds, target_dn = self.create_account(
6416             samdb, target_name,
6417             account_type=self.AccountType.MANAGED_SERVICE,
6418             spn=f'host/{target_name}',
6419             additional_details={
6420                 'msDS-AssignedAuthNPolicy': str(policy),
6421                 'objectClass': service_class,
6422             })
6423
6424         # Show that a network SamLogon succeeds.
6425         self._test_samlogon(creds=client_creds,
6426                             domain_joined_mach_creds=target_creds,
6427                             logon_type=netlogon.NetlogonNetworkInformation)
6428
6429         # Show that an interactive SamLogon succeeds.
6430         self._test_samlogon(creds=client_creds,
6431                             domain_joined_mach_creds=target_creds,
6432                             logon_type=netlogon.NetlogonInteractiveInformation)
6433
6434     def check_ticket_times(self,
6435                            ticket_creds,
6436                            expected_life=None,
6437                            expected_renew_life=None):
6438         ticket = ticket_creds.ticket_private
6439
6440         authtime = ticket['authtime']
6441         starttime = ticket.get('starttime', authtime)
6442         endtime = ticket['endtime']
6443         renew_till = ticket.get('renew-till', None)
6444
6445         starttime = self.get_EpochFromKerberosTime(starttime)
6446
6447         if expected_life is not None:
6448             actual_end = self.get_EpochFromKerberosTime(
6449                 endtime.decode('ascii'))
6450             actual_lifetime = actual_end - starttime
6451
6452             self.assertEqual(expected_life, actual_lifetime)
6453
6454         if renew_till is None:
6455             self.assertIsNone(expected_renew_life)
6456         else:
6457             if expected_renew_life is not None:
6458                 actual_renew_till = self.get_EpochFromKerberosTime(
6459                     renew_till.decode('ascii'))
6460                 actual_renew_life = actual_renew_till - starttime
6461
6462                 self.assertEqual(expected_renew_life, actual_renew_life)
6463
6464     def _get_tgt(self, creds, *,
6465                  armor_tgt=None,
6466                  till=None,
6467                  expected_error=0,
6468                  expect_status=None,
6469                  expected_status=None):
6470         user_name = creds.get_username()
6471         realm = creds.get_realm()
6472         salt = creds.get_salt()
6473
6474         cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
6475                                           names=user_name.split('/'))
6476         sname = self.PrincipalName_create(name_type=NT_SRV_INST,
6477                                           names=['krbtgt', realm])
6478         expected_sname = self.PrincipalName_create(
6479             name_type=NT_SRV_INST, names=['krbtgt', realm.upper()])
6480
6481         expected_cname = cname
6482
6483         if till is None:
6484             till = self.get_KerberosTime(offset=36000)
6485
6486         renew_time = till
6487
6488         krbtgt_creds = self.get_krbtgt_creds()
6489         ticket_decryption_key = (
6490             self.TicketDecryptionKey_from_creds(krbtgt_creds))
6491
6492         expected_etypes = krbtgt_creds.tgs_supported_enctypes
6493
6494         kdc_options = str(krb5_asn1.KDCOptions('renewable'))
6495         # Contrary to Microsoft’s documentation, the returned ticket is
6496         # renewable.
6497         expected_flags = krb5_asn1.TicketFlags('renewable')
6498
6499         preauth_key = self.PasswordKey_from_creds(creds,
6500                                                   kcrypto.Enctype.AES256)
6501
6502         expected_realm = realm.upper()
6503
6504         etypes = kcrypto.Enctype.AES256, kcrypto.Enctype.RC4
6505
6506         if armor_tgt is not None:
6507             authenticator_subkey = self.RandomKey(kcrypto.Enctype.AES256)
6508             armor_key = self.generate_armor_key(authenticator_subkey,
6509                                                 armor_tgt.session_key)
6510             armor_subkey = authenticator_subkey
6511
6512             client_challenge_key = self.generate_client_challenge_key(
6513                 armor_key, preauth_key)
6514             enc_challenge_padata = self.get_challenge_pa_data(
6515                 client_challenge_key)
6516
6517             def generate_fast_padata_fn(kdc_exchange_dict,
6518                                         _callback_dict,
6519                                         req_body):
6520                 return [enc_challenge_padata], req_body
6521
6522             generate_fast_fn = self.generate_simple_fast
6523             generate_fast_armor_fn = self.generate_ap_req
6524             generate_padata_fn = None
6525
6526             fast_armor_type = FX_FAST_ARMOR_AP_REQUEST
6527         else:
6528             ts_enc_padata = self.get_enc_timestamp_pa_data_from_key(
6529                 preauth_key)
6530
6531             def generate_padata_fn(kdc_exchange_dict,
6532                                    _callback_dict,
6533                                    req_body):
6534                 return [ts_enc_padata], req_body
6535
6536             generate_fast_fn = None
6537             generate_fast_padata_fn = None
6538             generate_fast_armor_fn = None
6539
6540             armor_key = None
6541             armor_subkey = None
6542
6543             fast_armor_type = None
6544
6545         if not expected_error:
6546             check_error_fn = None
6547             check_rep_fn = self.generic_check_kdc_rep
6548         else:
6549             check_error_fn = self.generic_check_kdc_error
6550             check_rep_fn = None
6551
6552         kdc_exchange_dict = self.as_exchange_dict(
6553             creds=creds,
6554             expected_error_mode=expected_error,
6555             expect_status=expect_status,
6556             expected_status=expected_status,
6557             expected_crealm=expected_realm,
6558             expected_cname=expected_cname,
6559             expected_srealm=expected_realm,
6560             expected_sname=expected_sname,
6561             expected_salt=salt,
6562             expected_flags=expected_flags,
6563             expected_supported_etypes=expected_etypes,
6564             generate_padata_fn=generate_padata_fn,
6565             generate_fast_padata_fn=generate_fast_padata_fn,
6566             generate_fast_fn=generate_fast_fn,
6567             generate_fast_armor_fn=generate_fast_armor_fn,
6568             fast_armor_type=fast_armor_type,
6569             check_error_fn=check_error_fn,
6570             check_rep_fn=check_rep_fn,
6571             check_kdc_private_fn=self.generic_check_kdc_private,
6572             armor_key=armor_key,
6573             armor_tgt=armor_tgt,
6574             armor_subkey=armor_subkey,
6575             kdc_options=kdc_options,
6576             preauth_key=preauth_key,
6577             ticket_decryption_key=ticket_decryption_key,
6578             # PA-DATA types are not important for these tests.
6579             check_patypes=False)
6580
6581         rep = self._generic_kdc_exchange(kdc_exchange_dict,
6582                                          cname=cname,
6583                                          realm=realm,
6584                                          sname=sname,
6585                                          till_time=till,
6586                                          renew_time=renew_time,
6587                                          etypes=etypes)
6588         if expected_error:
6589             self.check_error_rep(rep, expected_error)
6590
6591             return None
6592
6593         self.check_as_reply(rep)
6594
6595         ticket_creds = kdc_exchange_dict['rep_ticket_creds']
6596         return ticket_creds
6597
6598
6599 if __name__ == '__main__':
6600     global_asn1_print = False
6601     global_hexdump = False
6602     import unittest
6603     unittest.main()