python/samba/tests/krb5/alias_tests.py

   1 #!/usr/bin/env python3
   2 # Unix SMB/CIFS implementation.
   3 # Copyright (C) Stefan Metzmacher 2020
   4 # Copyright (C) 2021 Catalyst.Net Ltd
   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 import ldb
  24
  25 from samba.tests import delete_force
  26 import samba.tests.krb5.kcrypto as kcrypto
  27 from samba.tests.krb5.kdc_base_test import KDCBaseTest
  28 from samba.tests.krb5.rfc4120_constants import (
  29     AES256_CTS_HMAC_SHA1_96,
  30     ARCFOUR_HMAC_MD5,
  31     KDC_ERR_CLIENT_NAME_MISMATCH,
  32     NT_PRINCIPAL,
  33 )
  34
  35 sys.path.insert(0, 'bin/python')
  36 os.environ['PYTHONUNBUFFERED'] = '1'
  37
  38 global_asn1_print = False
  39 global_hexdump = False
  40
  41
  42 class AliasTests(KDCBaseTest):
  43     def test_dc_alias_rename(self):
  44         self._run_dc_alias(action='rename')
  45
  46     def test_dc_alias_delete(self):
  47         self._run_dc_alias(action='delete')
  48
  49     def _run_dc_alias(self, action=None):
  50         target_creds = self.get_dc_creds()
  51         target_name = target_creds.get_username()[:-1]
  52
  53         self._run_alias(target_name, lambda: target_creds, action=action)
  54
  55     def test_create_alias_rename(self):
  56         self._run_create_alias(action='rename')
  57
  58     def test_create_alias_delete(self):
  59         self._run_create_alias(action='delete')
  60
  61     def _run_create_alias(self, action=None):
  62         target_name = self.get_new_username()
  63
  64         def create_target():
  65             samdb = self.get_samdb()
  66
  67             realm = samdb.domain_dns_name().lower()
  68
  69             hostname = f'{target_name}.{realm}'
  70             spn = f'ldap/{hostname}'
  71
  72             details = {
  73                 'dNSHostName': hostname
  74             }
  75
  76             creds, fn = self.create_account(
  77                 samdb,
  78                 target_name,
  79                 account_type=self.AccountType.COMPUTER,
  80                 spn=spn,
  81                 additional_details=details)
  82
  83             return creds
  84
  85         self._run_alias(target_name, create_target, action=action)
  86
  87     def _run_alias(self, target_name, target_creds_fn, action=None):
  88         samdb = self.get_samdb()
  89
  90         mach_name = self.get_new_username()
  91
  92         # Create a machine account.
  93         mach_creds, mach_dn = self.create_account(
  94             samdb, mach_name, account_type=self.AccountType.COMPUTER)
  95         self.addCleanup(delete_force, samdb, mach_dn)
  96
  97         mach_sid = self.get_objectSid(samdb, mach_dn)
  98         realm = mach_creds.get_realm()
  99
 100         # The account salt doesn't change when the account is renamed.
 101         old_salt = mach_creds.get_salt()
 102         mach_creds.set_forced_salt(old_salt)
 103
 104         # Rename the account to alias with the target account.
 105         msg = ldb.Message(ldb.Dn(samdb, mach_dn))
 106         msg['sAMAccountName'] = ldb.MessageElement(target_name,
 107                                                    ldb.FLAG_MOD_REPLACE,
 108                                                    'sAMAccountName')
 109         samdb.modify(msg)
 110         mach_creds.set_username(target_name)
 111
 112         # Get a TGT for the machine account.
 113         tgt = self.get_tgt(mach_creds, kdc_options='0', fresh=True)
 114
 115         # Check the PAC.
 116         pac_data = self.get_pac_data(tgt.ticket_private['authorization-data'])
 117
 118         upn = f'{target_name}@{realm.lower()}'
 119
 120         self.assertEqual(target_name, str(pac_data.account_name))
 121         self.assertEqual(mach_sid, pac_data.account_sid)
 122         self.assertEqual(target_name, pac_data.logon_name)
 123         self.assertEqual(upn, pac_data.upn)
 124         self.assertEqual(realm, pac_data.domain_name)
 125
 126         # Rename or delete the machine account.
 127         if action == 'rename':
 128             mach_name2 = self.get_new_username()
 129
 130             msg = ldb.Message(ldb.Dn(samdb, mach_dn))
 131             msg['sAMAccountName'] = ldb.MessageElement(mach_name2,
 132                                                        ldb.FLAG_MOD_REPLACE,
 133                                                        'sAMAccountName')
 134             samdb.modify(msg)
 135         elif action == 'delete':
 136             samdb.delete(mach_dn)
 137         else:
 138             self.fail(action)
 139
 140         # Get the credentials for the target account.
 141         target_creds = target_creds_fn()
 142
 143         # Look up the DNS host name of the target account.
 144         target_dn = target_creds.get_dn()
 145         res = samdb.search(target_dn,
 146                            scope=ldb.SCOPE_BASE,
 147                            attrs=['dNSHostName'])
 148         target_hostname = str(res[0].get('dNSHostName', idx=0))
 149
 150         sname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
 151                                           names=['ldap', target_hostname])
 152         target_cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
 153                                                  names=[target_name])
 154
 155         target_decryption_key = self.TicketDecryptionKey_from_creds(
 156             target_creds)
 157
 158         authenticator_subkey = self.RandomKey(kcrypto.Enctype.AES256)
 159
 160         etypes = (AES256_CTS_HMAC_SHA1_96, ARCFOUR_HMAC_MD5)
 161
 162         def generate_s4u2self_padata(_kdc_exchange_dict,
 163                                      _callback_dict,
 164                                      req_body):
 165             padata = self.PA_S4U2Self_create(name=target_cname,
 166                                              realm=realm,
 167                                              tgt_session_key=tgt.session_key,
 168                                              ctype=None)
 169             return [padata], req_body
 170
 171         expected_error_mode = KDC_ERR_CLIENT_NAME_MISMATCH
 172
 173         # Make a request using S4U2Self. The request should fail.
 174         kdc_exchange_dict = self.tgs_exchange_dict(
 175             expected_crealm=realm,
 176             expected_cname=target_cname,
 177             expected_srealm=realm,
 178             expected_sname=sname,
 179             ticket_decryption_key=target_decryption_key,
 180             generate_padata_fn=generate_s4u2self_padata,
 181             expected_error_mode=expected_error_mode,
 182             check_error_fn=self.generic_check_kdc_error,
 183             check_kdc_private_fn=self.generic_check_kdc_private,
 184             tgt=tgt,
 185             authenticator_subkey=authenticator_subkey,
 186             kdc_options='0',
 187             expect_pac=True)
 188
 189         rep = self._generic_kdc_exchange(kdc_exchange_dict,
 190                                          cname=None,
 191                                          realm=realm,
 192                                          sname=sname,
 193                                          etypes=etypes)
 194         self.check_error_rep(rep, expected_error_mode)
 195
 196
 197 if __name__ == '__main__':
 198     global_asn1_print = False
 199     global_hexdump = False
 200     import unittest
 201     unittest.main()