tests/krb5: Allow changing the SID of a user's PAC

This lets us simulate a ticket of a user from another domain.

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

diff --git a/python/samba/tests/krb5/group_tests.py b/python/samba/tests/krb5/group_tests.py
index 0044203e81de05533b8c6cae8195c9e122784d75..12bc3eff6b1234b558c281b84245c63ae785377f 100755 (executable)
--- a/python/samba/tests/krb5/group_tests.py
+++ b/python/samba/tests/krb5/group_tests.py
@@ -107,7 +107,11 @@ class GroupTests(KDCBaseTest):
     # Get a ticket with the SIDs in the PAC replaced with ones we specify. This
     # is useful for creating arbitrary tickets that can be used to perform a
     # TGS-REQ.
-    def ticket_with_sids(self, ticket, new_sids, domain_sid):
+    def ticket_with_sids(self,
+                         ticket,
+                         new_sids,
+                         domain_sid,
+                         user_rid):
         krbtgt_creds = self.get_krbtgt_creds()
         krbtgt_key = self.TicketDecryptionKey_from_creds(krbtgt_creds)
 
@@ -117,14 +121,19 @@ class GroupTests(KDCBaseTest):
 
         modify_pac_fn = partial(self.set_pac_sids,
                                 new_sids=new_sids,
-                                domain_sid=domain_sid)
+                                domain_sid=domain_sid,
+                                user_rid=user_rid)
 
         return self.modified_ticket(ticket,
                                     modify_pac_fn=modify_pac_fn,
                                     checksum_keys=checksum_keys)
 
     # Replace the SIDs in a PAC with 'new_sids'.
-    def set_pac_sids(self, pac, new_sids, domain_sid):
+    def set_pac_sids(self,
+                     pac,
+                     new_sids,
+                     domain_sid,
+                     user_rid):
         base_sids = []
         extra_sids = []
         resource_sids = []
@@ -167,6 +176,10 @@ class GroupTests(KDCBaseTest):
             else:
                 self.fail(f'invalid SID type {sid_type}')
 
+        found_logon_info = True
+
+        user_sid = security.dom_sid(f'{domain_sid}-{user_rid}')
+
         pac_buffers = pac.buffers
         for pac_buffer in pac_buffers:
             # Find the LOGON_INFO PAC buffer.
@@ -191,6 +204,9 @@ class GroupTests(KDCBaseTest):
                 else:
                     logon_info.info3.base.groups.rids = None
 
+                logon_info.info3.base.domain_sid = security.dom_sid(domain_sid)
+                logon_info.info3.base.rid = int(user_rid)
+
                 # Add Resource SIDs and set the RESOURCE_GROUPS flag as needed.
                 logon_info.resource_groups.groups.count = len(resource_sids)
                 if resource_sids:
@@ -205,9 +221,18 @@ class GroupTests(KDCBaseTest):
                     logon_info.info3.base.user_flags &= ~(
                         netlogon.NETLOGON_RESOURCE_GROUPS)
 
-                break
-        else:
-            self.fail('no LOGON_INFO PAC buffer')
+                found_logon_info = True
+
+            # Also replace the user's SID in the UPN DNS buffer.
+            elif pac_buffer.type == krb5pac.PAC_TYPE_UPN_DNS_INFO:
+                upn_dns_info_ex = pac_buffer.info.ex
+
+                upn_dns_info_ex.objectsid = user_sid
+
+            # But don't replace the user's SID in the Requester SID buffer, or
+            # we'll get a SID mismatch.
+
+        self.assertTrue(found_logon_info, 'no LOGON_INFO PAC buffer')
 
         pac.buffers = pac_buffers
 
@@ -1019,6 +1044,9 @@ class GroupTests(KDCBaseTest):
         # Optional SIDs to replace those in the PAC prior to a TGS-REQ.
         tgs_sids = case.pop('tgs:sids', None)
 
+        # Optional user SID to replace that in the PAC prior to a TGS-REQ.
+        tgs_user_sid = case.pop('tgs:user_sid', None)
+
         # The SIDs we expect to see in the PAC after a AS-REQ or a TGS-REQ.
         as_expected = case.pop('as:expected', None)
         tgs_expected = case.pop('tgs:expected', None)
@@ -1050,6 +1078,11 @@ class GroupTests(KDCBaseTest):
                                  'specified compression for TGS request, but '
                                  'no expected SIDs provided')
 
+        if tgs_user_sid is not None:
+            self.assertIsNotNone(tgs_sids,
+                                 'specified TGS-REQ user SID, but no '
+                                 'accompanying SIDs provided')
+
         samdb = self.get_samdb()
 
         domain_sid = samdb.get_domain_sid()
@@ -1083,9 +1116,19 @@ class GroupTests(KDCBaseTest):
         groups = self.setup_groups(samdb, group_setup, user_principal)
         del group_setup
 
-        expected_groups = self.map_sids(as_expected, groups, domain_sid)
-        tgs_sids_mapped = self.map_sids(tgs_sids, groups, domain_sid)
-        tgs_expected_mapped = self.map_sids(tgs_expected, groups, domain_sid)
+        if tgs_user_sid is None:
+            tgs_user_sid = user_sid
+        elif tgs_user_sid in groups:
+            tgs_user_sid = groups[tgs_user_sid].sid
+
+        tgs_domain_sid, tgs_user_rid = tgs_user_sid.rsplit('-', 1)
+
+        expected_groups = self.map_sids(as_expected, groups,
+                                        domain_sid)
+        tgs_sids_mapped = self.map_sids(tgs_sids, groups,
+                                        tgs_domain_sid)
+        tgs_expected_mapped = self.map_sids(tgs_expected, groups,
+                                            tgs_domain_sid)
 
         till = self.get_KerberosTime(offset=36000)
         kdc_options = '0'
@@ -1128,13 +1171,20 @@ class GroupTests(KDCBaseTest):
 
         if tgs_sids is not None:
             # Replace the SIDs in the PAC with the ones provided by the test.
-            ticket = self.ticket_with_sids(ticket, tgs_sids_mapped, domain_sid)
+            ticket = self.ticket_with_sids(ticket,
+                                           tgs_sids_mapped,
+                                           tgs_domain_sid,
+                                           tgs_user_rid)
 
         target_creds, sname = self.get_target(tgs_to_krbtgt, tgs_compression)
         decryption_key = self.TicketDecryptionKey_from_creds(target_creds)
 
         subkey = self.RandomKey(ticket.session_key.etype)
 
+        requester_sid = None
+        if tgs_to_krbtgt:
+            requester_sid = user_sid
+
         # Perform a TGS-REQ with the user account.
 
         kdc_exchange_dict = self.tgs_exchange_dict(
@@ -1144,8 +1194,9 @@ class GroupTests(KDCBaseTest):
             expected_sname=sname,
             expected_account_name=user_name,
             expected_groups=tgs_expected_mapped,
-            expected_sid=user_sid,
-            expected_domain_sid=domain_sid,
+            expected_sid=tgs_user_sid,
+            expected_requester_sid=requester_sid,
+            expected_domain_sid=tgs_domain_sid,
             expected_supported_etypes=target_supported_etypes,
             ticket_decryption_key=decryption_key,
             check_rep_fn=self.generic_check_kdc_rep,

diff --git a/python/samba/tests/krb5/raw_testcase.py b/python/samba/tests/krb5/raw_testcase.py
index a3fcc701ab000af4163f0b926c6d837a8fea170d..beae57d07dc2ffab5e81582e07c07516016aee3d 100644 (file)
--- a/python/samba/tests/krb5/raw_testcase.py
+++ b/python/samba/tests/krb5/raw_testcase.py
@@ -2456,6 +2456,7 @@ class RawKerberosTest(TestCaseInTempDir):
                          unexpected_groups=None,
                          expected_upn_name=None,
                          expected_sid=None,
+                         expected_requester_sid=None,
                          expected_domain_sid=None,
                          expected_supported_etypes=None,
                          expected_flags=None,
@@ -2526,6 +2527,7 @@ class RawKerberosTest(TestCaseInTempDir):
             'unexpected_groups': unexpected_groups,
             'expected_upn_name': expected_upn_name,
             'expected_sid': expected_sid,
+            'expected_requester_sid': expected_requester_sid,
             'expected_domain_sid': expected_domain_sid,
             'expected_supported_etypes': expected_supported_etypes,
             'expected_flags': expected_flags,
@@ -2592,6 +2594,7 @@ class RawKerberosTest(TestCaseInTempDir):
                           unexpected_groups=None,
                           expected_upn_name=None,
                           expected_sid=None,
+                          expected_requester_sid=None,
                           expected_domain_sid=None,
                           expected_supported_etypes=None,
                           expected_flags=None,
@@ -2663,6 +2666,7 @@ class RawKerberosTest(TestCaseInTempDir):
             'unexpected_groups': unexpected_groups,
             'expected_upn_name': expected_upn_name,
             'expected_sid': expected_sid,
+            'expected_requester_sid': expected_requester_sid,
             'expected_domain_sid': expected_domain_sid,
             'expected_supported_etypes': expected_supported_etypes,
             'expected_flags': expected_flags,
@@ -3378,12 +3382,15 @@ class RawKerberosTest(TestCaseInTempDir):
             expected_types.append(krb5pac.PAC_TYPE_ATTRIBUTES_INFO)
 
         expect_requester_sid = kdc_exchange_dict['expect_requester_sid']
+        expected_requester_sid = kdc_exchange_dict['expected_requester_sid']
 
         if expect_requester_sid is None:
             if self.expect_extra_pac_buffers:
                 expect_requester_sid = expect_extra_pac_buffers
             else:
                 require_strict.add(krb5pac.PAC_TYPE_REQUESTER_SID)
+        if expected_requester_sid is not None:
+            expect_requester_sid = True
         if expect_requester_sid:
             expected_types.append(krb5pac.PAC_TYPE_REQUESTER_SID)
 
@@ -3482,8 +3489,11 @@ class RawKerberosTest(TestCaseInTempDir):
                       and expect_requester_sid):
                 requester_sid = pac_buffer.info.sid
 
+                if expected_requester_sid is None:
+                    expected_requester_sid = expected_sid
                 if expected_sid is not None:
-                    self.assertEqual(expected_sid, str(requester_sid))
+                    self.assertEqual(expected_requester_sid,
+                                     str(requester_sid))
 
             elif pac_buffer.type in {krb5pac.PAC_TYPE_CLIENT_CLAIMS_INFO,
                                      krb5pac.PAC_TYPE_DEVICE_CLAIMS_INFO}: