Simplify ccache SPN matching (#2318)

* Simplify ccache SPN matching

* tests: cover two-component ccache SPN matching

---------

Co-authored-by: Gabi Gonzalez <[email protected]>

impacket/krb5/ccache.py

@@ -426,8 +426,8 @@ class CCache:
                 # Let's search for any TGT/TGS that matches the server w/o the SPN's service type/port, returns
                 # the first one
                 # If server has no '/' we assume it's a ST from S4U2Self without a service type
-                if c['server'].prettyPrint().count(b'/') >= 2:
-                    # 3-part SPN: service/host/domain@REALM
+                if c['server'].prettyPrint().find(b'/') >= 0:
+                    # Match both service/host@REALM and service/host/domain@REALM.
                     # Seen when a service ticket is dumped from LSASS memory (e.g. via
                     # Rubeus' `dump` command) rather than requested fresh - Windows caches
                     # such tickets with the domain suffix appended as an extra SPN
@@ -450,14 +450,6 @@ class CCache:
                     if hostsMatch and cachedRealm == searchRealm:
                         LOG.debug('Returning cached credential for %s' % c['server'].prettyPrint().upper().decode('utf-8'))
                         return c
-                elif c['server'].prettyPrint().find(b'/') >=0:
-                    # Let's take the port out for comparison
-                    cachedSPN = (c['server'].prettyPrint().upper().split(b'/')[1].split(b'@')[0].split(b':')[0] + b'@' + c['server'].prettyPrint().upper().split(b'/')[1].split(b'@')[1])
-                    searchSPN = '%s@%s' % (server.upper().split('/')[1].split('@')[0].split(':')[0],
-                                               server.upper().split('/')[1].split('@')[1])
-                    if cachedSPN == b(searchSPN):
-                        LOG.debug('Returning cached credential for %s' % c['server'].prettyPrint().upper().decode('utf-8'))
-                        return c
                 else:
                     # Should be of form 'hostname$@REALM'
                     cachedSPN = (c['server'].prettyPrint().upper().split(b'@')[0].split(b':')[0] + b'@' + c['server'].prettyPrint().upper().split(b'@')[1])

tests/misc/test_ccache.py

@@ -161,6 +161,58 @@ class CCACHETests(unittest.TestCase):
         # Same short hostname, different child domain -> must not match
         self.assertIsNone(ccache.getCredential("LDAP/DC01.CHILD-B.LOCAL/CHILD-B.LOCAL@{}".format(realm)))
 
+    def test_ccache_getCredential_two_part_spn_short_fqdn_matching(self):
+        realm = "EXAMPLE.TEST"
+
+        cases = (
+            (
+                "HOST/FILESERVER@{}".format(realm),
+                "CIFS/FILESERVER.EXAMPLE.TEST@{}".format(realm),
+            ),
+            (
+                "HOST/FILESERVER.EXAMPLE.TEST@{}".format(realm),
+                "CIFS/FILESERVER@{}".format(realm),
+            ),
+            (
+                "HOST/FILESERVER@{}".format(realm),
+                "CIFS/FILESERVER.EXAMPLE.TEST:445@{}".format(realm),
+            ),
+        )
+
+        for cached_spn, requested_spn in cases:
+            with self.subTest(cached_spn=cached_spn, requested_spn=requested_spn):
+                ccache = self._ccache_with(cached_spn)
+                credential = ccache.credentials[0]
+                self.assertIs(ccache.getCredential(requested_spn), credential)
+
+    def test_ccache_getCredential_two_part_spn_matching_boundaries(self):
+        realm = "FOREST.LOCAL"
+        cached_spn = "HOST/SERVER.CHILD-A.LOCAL@{}".format(realm)
+        ccache = self._ccache_with(cached_spn)
+        credential = ccache.credentials[0]
+
+        # Relaxed matching must remain bounded by the Kerberos realm.
+        self.assertIsNone(
+            ccache.getCredential("CIFS/[email protected]")
+        )
+
+        # Two qualified names must not match only because their first DNS
+        # label is the same.
+        self.assertIsNone(
+            ccache.getCredential("CIFS/SERVER.CHILD-B.LOCAL@{}".format(realm))
+        )
+
+        # Disabling AnySPN must retain exact service and hostname matching.
+        self.assertIsNone(
+            ccache.getCredential(
+                "CIFS/SERVER@{}".format(realm),
+                anySPN=False,
+            )
+        )
+
+        # The negative lookups above must not alter the cached credential.
+        self.assertIs(ccache.getCredential(cached_spn, anySPN=False), credential)
+
     @staticmethod
     def _ccache_with(*spns):
         ccache = CCache()