tests/krb5: Add tests for enterprise principals with canonicalization

Signed-off-by: Joseph Sutton <josephsutton@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>

3 files changed, 394 insertions, 10 deletions

diff --git a/python/samba/tests/krb5/as_req_tests.py b/python/samba/tests/krb5/as_req_tests.py
index 5aa6fcb3881..263e77d4812 100755
--- a/python/samba/tests/krb5/as_req_tests.py
+++ b/python/samba/tests/krb5/as_req_tests.py
@@ -344,6 +344,146 @@ class AsReqKerberosTests(AsReqBaseTest):
             expected_error=KDC_ERR_C_PRINCIPAL_UNKNOWN,
             expect_edata=False)
 
+    def test_as_req_enterprise_canon(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'krb5_enterprise0'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            expected_cname=expected_cname,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=krb5_asn1.KDCOptions('canonicalize'))
+
+    def test_as_req_enterprise_canon_case(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'krb5_enterprise1'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            expected_cname=expected_cname,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=krb5_asn1.KDCOptions('canonicalize'))
+
+    def test_as_req_enterprise_canon_mac(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'krb5_enterprise2'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            expected_cname=expected_cname,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=krb5_asn1.KDCOptions('canonicalize'))
+
+    def test_as_req_enterprise_canon_mac_case(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'krb5_enterprise3'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            expected_cname=expected_cname,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=krb5_asn1.KDCOptions('canonicalize'))
+
+    def test_as_req_enterprise_no_canon(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'krb5_enterprise4'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=0)
+
+    def test_as_req_enterprise_no_canon_case(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'krb5_enterprise5'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=0)
+
+    def test_as_req_enterprise_no_canon_mac(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'krb5_enterprise6'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=0)
+
+    def test_as_req_enterprise_no_canon_mac_case(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'krb5_enterprise7'})
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        self._run_as_req_enc_timestamp(
+            client_creds,
+            client_account=client_account,
+            name_type=NT_ENTERPRISE_PRINCIPAL,
+            kdc_options=0)
+
 
 if __name__ == "__main__":
     global_asn1_print = False
diff --git a/python/samba/tests/krb5/kdc_base_test.py b/python/samba/tests/krb5/kdc_base_test.py
index a23fef67b53..aada0457461 100644
--- a/python/samba/tests/krb5/kdc_base_test.py
+++ b/python/samba/tests/krb5/kdc_base_test.py
@@ -1405,17 +1405,25 @@ class KDCBaseTest(RawKerberosTest):
         return service_ticket_creds
 
     def get_tgt(self, creds, to_rodc=False, kdc_options=None,
+                client_account=None, client_name_type=NT_PRINCIPAL,
                 expected_flags=None, unexpected_flags=None,
                 expected_account_name=None, expected_upn_name=None,
+                expected_cname=None,
                 expected_sid=None,
                 pac_request=True, expect_pac=True,
                 expect_pac_attrs=None, expect_pac_attrs_pac_request=None,
                 expect_requester_sid=None,
                 fresh=False):
-        user_name = creds.get_username()
+        if client_account is not None:
+            user_name = client_account
+        else:
+            user_name = creds.get_username()
+
         cache_key = (user_name, to_rodc, kdc_options, pac_request,
+                     client_name_type,
                      str(expected_flags), str(unexpected_flags),
                      expected_account_name, expected_upn_name, expected_sid,
+                     str(expected_cname),
                      expect_pac, expect_pac_attrs,
                      expect_pac_attrs_pac_request, expect_requester_sid)
 
@@ -1430,11 +1438,14 @@ class KDCBaseTest(RawKerberosTest):
         salt = creds.get_salt()
 
         etype = (AES256_CTS_HMAC_SHA1_96, ARCFOUR_HMAC_MD5)
-        cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
-                                          names=[user_name])
+        cname = self.PrincipalName_create(name_type=client_name_type,
+                                          names=user_name.split('/'))
         sname = self.PrincipalName_create(name_type=NT_SRV_INST,
                                           names=['krbtgt', realm])
 
+        if expected_cname is None:
+            expected_cname = cname
+
         till = self.get_KerberosTime(offset=36000)
 
         if to_rodc:
@@ -1463,7 +1474,7 @@ class KDCBaseTest(RawKerberosTest):
             client_as_etypes=etype,
             expected_error_mode=KDC_ERR_PREAUTH_REQUIRED,
             expected_crealm=realm,
-            expected_cname=cname,
+            expected_cname=expected_cname,
             expected_srealm=realm,
             expected_sname=sname,
             expected_account_name=expected_account_name,
@@ -1510,7 +1521,7 @@ class KDCBaseTest(RawKerberosTest):
             client_as_etypes=etype,
             expected_error_mode=0,
             expected_crealm=expected_realm,
-            expected_cname=cname,
+            expected_cname=expected_cname,
             expected_srealm=expected_realm,
             expected_sname=expected_sname,
             expected_account_name=expected_account_name,
diff --git a/python/samba/tests/krb5/kdc_tgs_tests.py b/python/samba/tests/krb5/kdc_tgs_tests.py
index 2923d53772a..740dd43f34d 100755
--- a/python/samba/tests/krb5/kdc_tgs_tests.py
+++ b/python/samba/tests/krb5/kdc_tgs_tests.py
@@ -48,6 +48,7 @@ from samba.tests.krb5.rfc4120_constants import (
     KDC_ERR_TGT_REVOKED,
     KRB_ERR_TKT_NYV,
     KDC_ERR_WRONG_REALM,
+    NT_ENTERPRISE_PRINCIPAL,
     NT_PRINCIPAL,
     NT_SRV_INST,
 )
@@ -231,14 +232,19 @@ class KdcTgsTests(KDCBaseTest):
             "rep = {%s},%s" % (rep, pac_data))
 
     def _make_tgs_request(self, client_creds, service_creds, tgt,
+                          client_account=None,
+                          client_name_type=NT_PRINCIPAL,
+                          kdc_options=None,
                           pac_request=None, expect_pac=True,
                           expect_error=False,
+                          expected_cname=None,
                           expected_account_name=None,
                           expected_upn_name=None,
                           expected_sid=None):
-        client_account = client_creds.get_username()
-        cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
-                                          names=[client_account])
+        if client_account is None:
+            client_account = client_creds.get_username()
+        cname = self.PrincipalName_create(name_type=client_name_type,
+                                          names=client_account.split('/'))
 
         service_account = service_creds.get_username()
         sname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
@@ -247,7 +253,8 @@ class KdcTgsTests(KDCBaseTest):
         realm = service_creds.get_realm()
 
         expected_crealm = realm
-        expected_cname = cname
+        if expected_cname is None:
+            expected_cname = cname
         expected_srealm = realm
         expected_sname = sname
 
@@ -255,7 +262,9 @@ class KdcTgsTests(KDCBaseTest):
 
         etypes = (AES256_CTS_HMAC_SHA1_96, ARCFOUR_HMAC_MD5)
 
-        kdc_options = str(krb5_asn1.KDCOptions('canonicalize'))
+        if kdc_options is None:
+            kdc_options = 'canonicalize'
+        kdc_options = str(krb5_asn1.KDCOptions(kdc_options))
 
         target_decryption_key = self.TicketDecryptionKey_from_creds(
             service_creds)
@@ -335,6 +344,230 @@ class KdcTgsTests(KDCBaseTest):
         pac = self.get_ticket_pac(ticket, expect_pac=False)
         self.assertIsNone(pac)
 
+    def test_request_enterprise_canon(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'tgs_enterprise0'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        kdc_options = 'canonicalize'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_cname=expected_cname,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_cname=expected_cname,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
+    def test_request_enterprise_canon_case(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'tgs_enterprise1'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        kdc_options = 'canonicalize'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_cname=expected_cname,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_cname=expected_cname,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
+    def test_request_enterprise_canon_mac(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'tgs_enterprise2'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        kdc_options = 'canonicalize'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_cname=expected_cname,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_cname=expected_cname,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
+    def test_request_enterprise_canon_case_mac(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'tgs_enterprise3'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        expected_cname = self.PrincipalName_create(
+            name_type=NT_PRINCIPAL,
+            names=[user_name])
+
+        kdc_options = 'canonicalize'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_cname=expected_cname,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_cname=expected_cname,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
+    def test_request_enterprise_no_canon(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'tgs_enterprise4'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        kdc_options = '0'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
+    def test_request_enterprise_no_canon_case(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.USER,
+            opts={'upn': 'tgs_enterprise5'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        kdc_options = '0'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
+    def test_request_enterprise_no_canon_mac(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'tgs_enterprise6'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm()
+        client_account = f'{user_name}@{realm}'
+
+        kdc_options = '0'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
+    def test_request_enterprise_no_canon_case_mac(self):
+        client_creds = self.get_cached_creds(
+            account_type=self.AccountType.COMPUTER,
+            opts={'upn': 'tgs_enterprise7'})
+        service_creds = self.get_service_creds()
+
+        user_name = client_creds.get_username()
+        realm = client_creds.get_realm().lower()
+        client_account = f'{user_name}@{realm}'
+
+        kdc_options = '0'
+
+        tgt = self.get_tgt(client_creds,
+                           client_account=client_account,
+                           client_name_type=NT_ENTERPRISE_PRINCIPAL,
+                           expected_account_name=user_name,
+                           kdc_options=kdc_options)
+
+        self._make_tgs_request(
+            client_creds, service_creds, tgt,
+            client_account=client_account,
+            client_name_type=NT_ENTERPRISE_PRINCIPAL,
+            expected_account_name=user_name,
+            kdc_options=kdc_options)
+
     def test_client_no_auth_data_required(self):
         client_creds = self.get_cached_creds(
             account_type=self.AccountType.USER,