Adds support for ESC13

certipy/commands/find.py

@@ -12,6 +12,7 @@
 
 from asn1crypto import x509
 from certipy.lib.constants import (
+    ISSUANCE_POLICY_RIGHTS,
     CERTIFICATE_RIGHTS,
     CERTIFICATION_AUTHORITY_RIGHTS,
     EXTENDED_RIGHTS_MAP,
@@ -29,6 +30,7 @@
 from certipy.lib.security import (
     ActiveDirectorySecurity,
     CertifcateSecurity,
+    IssuancePolicySecurity,
     is_admin_sid,
 )
 from certipy.lib.target import Target
@@ -90,6 +92,7 @@ def __init__(
         stdout: bool = False,
         output: str = None,
         enabled: bool = False,
+        oids: bool = False,
         vulnerable: bool = False,
         hide_admins: bool = False,
         dc_only: bool = False,
@@ -106,6 +109,7 @@ def __init__(
         self.stdout = stdout
         self.output = output
         self.enabled = enabled
+        self.oids = oids
         self.vuln = vulnerable
         self.hide_admins = hide_admins
         self.dc_only = dc_only
@@ -231,6 +235,55 @@ def find(self):
             )
         )
 
+        logging.info("Finding issuance policies")
+
+        oids = self.get_issuance_policies()
+
+        logging.info(
+            "Found %d issuance polic%s"
+            % (
+                len(cas),
+                "ies" if len(cas) != 1 else "y",
+            )
+        )
+
+        no_enabled_oids = 0
+        for template in templates:
+            object_id = template.get("objectGUID").lstrip("{").rstrip("}")
+            issuance_policies = template.get("msPKI-Certificate-Policy")
+            if not isinstance(issuance_policies, list):
+                if issuance_policies is None:
+                    issuance_policies = []
+                else:
+                    issuance_policies = [issuance_policies]
+            template.set("issuance_policies", issuance_policies)
+
+            for oid in oids:
+                if oid.get("msPKI-Cert-Template-OID") in issuance_policies:
+                    no_enabled_oids += 1
+                    linked_group = b''.join(oid.get_raw("msDS-OIDToGroupLink")).decode()
+                    if "templates" in oid["attributes"].keys():
+                        oid.get("templates").append(template.get("name"))
+                        oid.get("templates_ids").append(object_id)
+                    else:
+                        oid.set("templates", [template.get("name")])
+                        oid.set("templates_ids", [object_id])
+                    if linked_group:
+                        oid.set("linked_group", linked_group)
+                        if "issuance_policies_linked_groups" in template["attributes"].keys():
+                            template.get("issuance_policies_linked_groups").append(linked_group)
+                        else:
+                            template.set("issuance_policies_linked_groups", [linked_group])
+
+        logging.info(
+            "Found %d OID%s linked to %s"
+            % (
+                no_enabled_oids,
+                "s" if no_enabled_oids != 1 else "",
+                "templates" if no_enabled_oids != 1 else "a template",
+            )
+        )
+
         for ca in cas:
 
             if self.dc_only:
@@ -455,7 +508,7 @@ def find(self):
             self.output_bloodhound_data(prefix, templates, cas)
 
         if self.text or self.json or not_specified:
-            output = self.get_output_for_text_and_json(templates, cas)
+            output = self.get_output_for_text_and_json(templates, cas, oids)
 
             if self.text or not_specified:
                 if self.stdout:
@@ -477,10 +530,11 @@ def find(self):
                 )
 
     def get_output_for_text_and_json(
-        self, templates: List[LDAPEntry], cas: List[LDAPEntry]
+        self, templates: List[LDAPEntry], cas: List[LDAPEntry], oids: List[LDAPEntry]
     ):
         ca_entries = {}
         template_entries = {}
+        oids_entries ={} 
 
         for template in templates:
             if self.enabled and template.get("enabled") is not True:
@@ -516,6 +570,24 @@ def get_output_for_text_and_json(
 
             ca_entries[len(ca_entries)] = entry
 
+        if self.oids is True:
+            for oid in oids:
+                vulnerabilities = self.get_oid_vulnerabilities(oid)
+                if self.vuln and len(vulnerabilities) == 0:
+                    continue
+
+                entry = OrderedDict()
+                entry = self.get_oid_properties(oid, entry)
+
+                permissions = self.get_oid_permissions(oid)
+                if len(permissions) > 0:
+                    entry["Permissions"] = permissions
+
+                if len(vulnerabilities) > 0:
+                    entry["[!] Vulnerabilities"] = vulnerabilities
+
+                oids_entries[len(oids_entries)] = entry
+
         output = {}
 
         if len(ca_entries) == 0:
@@ -530,6 +602,14 @@ def get_output_for_text_and_json(
         else:
             output["Certificate Templates"] = template_entries
 
+        if self.oids is True:
+            if len(oids_entries) == 0:
+                output[
+                    "Issuance Policies"
+                ] = "[!] Could not find any issuance policy"
+            else:
+                output["Issuance Policies"] = oids_entries
+
         return output
 
     def output_bloodhound_data(
@@ -731,6 +811,7 @@ def get_certificate_templates(self) -> List[LDAPEntry]:
                 "msPKI-Enrollment-Flag",
                 "msPKI-Private-Key-Flag",
                 "msPKI-Certificate-Name-Flag",
+                "msPKI-Certificate-Policy",
                 "msPKI-Minimal-Key-Size",
                 "msPKI-RA-Signature",
                 "pKIExtendedKeyUsage",
@@ -742,6 +823,25 @@ def get_certificate_templates(self) -> List[LDAPEntry]:
 
         return templates
 
+    def get_issuance_policies(self) -> List[LDAPEntry]:
+        templates = self.connection.search(
+            "(objectclass=msPKI-Enterprise-Oid)",
+            search_base="CN=OID,CN=Public Key Services,CN=Services,%s"
+            % self.connection.configuration_path,
+            attributes=[
+                "cn",
+                "name",
+                "displayName",
+                "msDS-OIDToGroupLink",
+                "msPKI-Cert-Template-OID",
+                "nTSecurityDescriptor",
+                "objectGUID",
+            ],
+            query_sd=True,
+        )
+
+        return templates
+
     def get_certificate_authorities(self) -> List[LDAPEntry]:
         cas = self.connection.search(
             "(&(objectClass=pKIEnrollmentService))",
@@ -830,7 +930,9 @@ def get_template_properties(
             "authorized_signatures_required": "Authorized Signatures Required",
             "validity_period": "Validity Period",
             "renewal_period": "Renewal Period",
-            "msPKI-Minimal-Key-Size": "Minimum RSA Key Length"
+            "msPKI-Minimal-Key-Size": "Minimum RSA Key Length",
+            "msPKI-Certificate-Policy": "Issuance Policies",
+            "issuance_policies_linked_groups" : "Linked Groups"
         }
 
         if template_properties is None:
@@ -967,6 +1069,19 @@ def list_sids(sids: List[str]):
                     enrollable_sids
                 )
 
+            # ESC13
+            if (
+                user_can_enroll
+                and template.get("client_authentication")
+                and template.get("msPKI-Certificate-Policy")
+                and template.get("issuance_policies_linked_groups")
+            ):
+                vulnerabilities[
+                    "ESC13"
+                ] = "%s can enroll, template allows client authentication and issuance policy is linked to group %s" % (list_sids(
+                    enrollable_sids
+                ), template.get("issuance_policies_linked_groups"))
+
         # ESC4
         security = CertifcateSecurity(template.get("nTSecurityDescriptor"))
         owner_sid = security.owner
@@ -1170,6 +1285,120 @@ def ca_has_vulnerable_acl(self, ca: LDAPEntry):
 
         return has_vulnerable_acl, vulnerable_acl_sids
 
+    def get_oid_properties(
+            self, oid: LDAPEntry, oid_properties: dict = None
+        ) -> dict:
+            properties_map = {
+                "cn": "Issuance Policy Name",
+                "displayName": "Display Name",
+                "templates": "Certificate Template(s)",
+                "linked_group" : "Linked Group"
+            }
+
+            if oid_properties is None:
+                oid_properties = OrderedDict()
+
+            for property_key, property_display in properties_map.items():
+                property_value = oid.get(property_key)
+                if property_value is None:
+                    continue
+                oid_properties[property_display] = property_value
+
+            return oid_properties
+
+    def get_oid_permissions(self, oid: LDAPEntry):
+        security = IssuancePolicySecurity(oid.get("nTSecurityDescriptor"))
+        oid_permissions = {}
+        access_rights = {}
+        if security is not None:
+            if not self.hide_admins or not is_admin_sid(security.owner):
+                oid_permissions["Owner"] = self.connection.lookup_sid(
+                    security.owner
+                ).get("name")
+
+            for sid, rights in security.aces.items():
+                if self.hide_admins and is_admin_sid(sid):
+                    continue
+                oid_rights = rights["rights"].to_list()
+                for oid_right in oid_rights:
+                    if oid_right not in access_rights:
+                        access_rights[oid_right] = [
+                            self.connection.lookup_sid(sid).get("name")
+                        ]
+                    else:
+                        access_rights[oid_right].append(
+                            self.connection.lookup_sid(sid).get("name")
+                        )
+
+            oid_permissions["Access Rights"] = access_rights
+
+        return oid_permissions
+
+
+    def get_oid_vulnerabilities(self, oid: LDAPEntry):
+        def list_sids(sids: List[str]):
+            sids_mapping = list(
+                map(
+                    lambda sid: repr(self.connection.lookup_sid(sid).get("name")),
+                    sids,
+                )
+            )
+            if len(sids_mapping) == 1:
+                return sids_mapping[0]
+
+            return ", ".join(sids_mapping[:-1]) + " and " + sids_mapping[-1]
+
+        if oid.get("vulnerabilities"):
+            return oid.get("vulnerabilities")
+
+        vulnerabilities = {}
+
+        # ESC13
+        security = IssuancePolicySecurity(oid.get("nTSecurityDescriptor"))
+        owner_sid = security.owner
+
+        if owner_sid in self.connection.get_user_sids(self.target.username):
+            vulnerabilities[
+                "ESC13"
+            ] = "Issuance Policy OID is owned by %s" % self.connection.lookup_sid(owner_sid).get(
+                "name"
+            )
+        else:
+            has_vulnerable_acl, vulnerable_acl_sids = self.oid_has_vulnerable_acl(
+                oid
+            )
+            if has_vulnerable_acl:
+                vulnerabilities["ESC13"] = "%s has dangerous permissions" % list_sids(
+                    vulnerable_acl_sids
+                )
+
+        return vulnerabilities
+
+    def oid_has_vulnerable_acl(self, oid: LDAPEntry):
+        has_vulnerable_acl = False
+
+        security = IssuancePolicySecurity(oid.get("nTSecurityDescriptor"))
+        aces = security.aces
+        vulnerable_acl_sids = []
+        for sid, rights in aces.items():
+            if sid not in self.connection.get_user_sids(self.target.username):
+                continue
+
+            ad_rights = rights["rights"]
+            if any(
+                right in ad_rights
+                for right in [
+                    ISSUANCE_POLICY_RIGHTS.GENERIC_ALL,
+                    ISSUANCE_POLICY_RIGHTS.WRITE_OWNER,
+                    ISSUANCE_POLICY_RIGHTS.WRITE_DACL,
+                    ISSUANCE_POLICY_RIGHTS.WRITE_PROPERTY,
+                ]
+            ):
+                vulnerable_acl_sids.append(sid)
+                has_vulnerable_acl = True
+
+        return has_vulnerable_acl, vulnerable_acl_sids
+
 
 def entry(options: argparse.Namespace) -> None:
     target = Target.from_options(options, dc_as_target=True)

certipy/commands/parsers/find.py

@@ -65,6 +65,11 @@ def add_subparser(subparsers: argparse._SubParsersAction) -> Tuple[str, Callable
         action="store_true",
         help="Show only vulnerable certificate templates based on nested group memberships. Does not affect BloodHound output",
     )
+    group.add_argument(
+        "-oids",
+        action="store_true",
+        help="Show OIDs (Issuance Policies) and their properties.",
+    )
     group.add_argument(
         "-hide-admins",
         action="store_true",

certipy/lib/constants.py

@@ -268,6 +268,12 @@ def to_list(self):
             filtered_members.append(member)
         return filtered_members
 
+class ISSUANCE_POLICY_RIGHTS(IntFlag):
+    GENERIC_READ = 131220
+    GENERIC_ALL = 983551
+    WRITE_OWNER = 524288
+    WRITE_DACL = 262144
+    WRITE_PROPERTY = 32
 
 # https://github.com/GhostPack/Certify/blob/2b1530309c0c5eaf41b2505dfd5a68c83403d031/Certify/Domain/CertificateAuthority.cs#L11
 class CERTIFICATION_AUTHORITY_RIGHTS(IntFlag):

certipy/lib/security.py

@@ -10,6 +10,7 @@
     ACTIVE_DIRECTORY_RIGHTS,
     CERTIFICATE_RIGHTS,
     CERTIFICATION_AUTHORITY_RIGHTS,
+    ISSUANCE_POLICY_RIGHTS,
 )
 
 
@@ -59,6 +60,9 @@ class CASecurity(ActiveDirectorySecurity):
 class CertifcateSecurity(ActiveDirectorySecurity):
     RIGHTS_TYPE = CERTIFICATE_RIGHTS
 
+class IssuancePolicySecurity(ActiveDirectorySecurity):
+    RIGHTS_TYPE = ISSUANCE_POLICY_RIGHTS
+
 
 def is_admin_sid(sid: str):
     return (