feat(ldap): add ADCS collection support via CertiHound integration

nxc/protocols/ldap.py

@@ -82,6 +82,7 @@ def resolve_collection_methods(methods):
         "psremote",
         "dconly",
         "container",
+        "adcs",
     ]
     default_methods = ["group", "localadmin", "session", "trusts"]
     # Similar to SharpHound, All is not really all, it excludes loggedon
@@ -96,6 +97,7 @@ def resolve_collection_methods(methods):
         "rdp",
         "psremote",
         "container",
+        "adcs",
     ]
     # DC only, does not collect to computers
     dconly_methods = ["group", "trusts", "objectprops", "acl", "container"]
@@ -1626,93 +1628,101 @@ def ldap_to_filetime(ldap_time):
             break  # Only process first policy result
 
     def bloodhound(self):
-        # Check which version is desired
-        use_bhce = self.config.getboolean("BloodHound-CE", "bhce_enabled", fallback=False)
-        package_name, version, is_ce = get_bloodhound_info()
-
-        if use_bhce and not is_ce:
-            self.logger.fail("⚠️  Configuration Issue Detected ⚠️")
-            self.logger.fail(f"Your configuration has BloodHound-CE enabled, but the regular BloodHound package is installed. Modify your {CONFIG_PATH} config file or follow the instructions:")
-            self.logger.fail("Please run the following commands to fix this:")
-            self.logger.fail("poetry remove bloodhound-ce   # poetry falsely recognizes bloodhound-ce as a the old bloodhound package")
-            self.logger.fail("poetry add bloodhound-ce")
-            self.logger.fail("")
-
-            # If using pipx
-            self.logger.fail("Or if you installed with pipx:")
-            self.logger.fail("pipx runpip netexec uninstall -y bloodhound")
-            self.logger.fail("pipx inject netexec bloodhound-ce --force")
-            return
-
-        elif not use_bhce and is_ce:
-            self.logger.fail("⚠️  Configuration Issue Detected ⚠️")
-            self.logger.fail("Your configuration has regular BloodHound enabled, but the BloodHound-CE package is installed.")
-            self.logger.fail("Please run the following commands to fix this:")
-            self.logger.fail("poetry remove bloodhound-ce")
-            self.logger.fail("poetry add bloodhound")
-            self.logger.fail("")
-
-            # If using pipx
-            self.logger.fail("Or if you installed with pipx:")
-            self.logger.fail("pipx runpip netexec uninstall -y bloodhound-ce")
-            self.logger.fail("pipx inject netexec bloodhound --force")
-            return
-
-        auth = ADAuthentication(
-            username=self.username,
-            password=self.password,
-            domain=self.domain,
-            lm_hash=self.nthash,
-            nt_hash=self.nthash,
-            aeskey=self.aesKey,
-            kdc=self.kdcHost,
-            auth_method="auto",
-        )
-        ad = AD(
-            auth=auth,
-            domain=self.domain,
-            nameserver=self.args.dns_server,
-            dns_tcp=self.args.dns_tcp,
-            dns_timeout=self.args.dns_timeout,
-        )
         collect = resolve_collection_methods("Default" if not self.args.collection else self.args.collection)
         if not collect:
             return
         self.logger.highlight("Resolved collection methods: " + ", ".join(list(collect)))
 
-        self.logger.debug("Using DNS to retrieve domain information")
-        try:
-            ad.dns_resolve(domain=self.domain)
-        except (resolver.LifetimeTimeout, resolver.NoNameservers):
-            self.logger.fail("Bloodhound-python failed to resolve domain information, try specifying the DNS server.")
-            return
-
-        if self.args.kerberos:
-            self.logger.highlight("Using kerberos auth without ccache, getting TGT")
-            auth.get_tgt()
-        if self.args.use_kcache:
-            self.logger.highlight("Using kerberos auth from ccache")
-            auth.load_ccache()
+        # Separate ADCS from bloodhound-python methods
+        bh_collect = {m for m in collect if m != "adcs"}
+        need_bloodhound_python = len(bh_collect) > 0
 
         timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S") + "_"
-        bloodhound = BloodHound(ad, self.hostname, self.host, self.port)
-        bloodhound.connect()
+        adcs_files = []
+
+        # Run bloodhound-python collection if needed
+        if need_bloodhound_python:
+            use_bhce = self.config.getboolean("BloodHound-CE", "bhce_enabled", fallback=False)
+            package_name, version, is_ce = get_bloodhound_info()
+
+            if use_bhce and not is_ce:
+                self.logger.fail("Configuration Issue Detected")
+                self.logger.fail(f"Your configuration has BloodHound-CE enabled, but the regular BloodHound package is installed. Modify your {CONFIG_PATH} config file or follow the instructions:")
+                self.logger.fail("Please run the following commands to fix this:")
+                self.logger.fail("poetry remove bloodhound-ce   # poetry falsely recognizes bloodhound-ce as a the old bloodhound package")
+                self.logger.fail("poetry add bloodhound-ce")
+                self.logger.fail("")
+                self.logger.fail("Or if you installed with pipx:")
+                self.logger.fail("pipx runpip netexec uninstall -y bloodhound")
+                self.logger.fail("pipx inject netexec bloodhound-ce --force")
+                return
 
-        try:
-            bloodhound.run(
-                collect=collect,
-                num_workers=10,
-                disable_pooling=False,
-                timestamp=timestamp,
-                fileNamePrefix=self.output_filename.split("/")[-1],
-                computerfile=None,
-                cachefile=None,
-                exclude_dcs=False,
+            elif not use_bhce and is_ce:
+                self.logger.fail("Configuration Issue Detected")
+                self.logger.fail("Your configuration has regular BloodHound enabled, but the BloodHound-CE package is installed.")
+                self.logger.fail("Please run the following commands to fix this:")
+                self.logger.fail("poetry remove bloodhound-ce")
+                self.logger.fail("poetry add bloodhound")
+                self.logger.fail("")
+                self.logger.fail("Or if you installed with pipx:")
+                self.logger.fail("pipx runpip netexec uninstall -y bloodhound-ce")
+                self.logger.fail("pipx inject netexec bloodhound --force")
+                return
+
+            auth = ADAuthentication(
+                username=self.username,
+                password=self.password,
+                domain=self.domain,
+                lm_hash=self.nthash,
+                nt_hash=self.nthash,
+                aeskey=self.aesKey,
+                kdc=self.kdcHost,
+                auth_method="auto",
             )
-        except Exception as e:
-            self.logger.fail(f"BloodHound collection failed: {e.__class__.__name__} - {e}")
-            self.logger.debug(f"BloodHound collection failed: {e.__class__.__name__} - {e}", exc_info=True)
-            return
+            ad = AD(
+                auth=auth,
+                domain=self.domain,
+                nameserver=self.args.dns_server,
+                dns_tcp=self.args.dns_tcp,
+                dns_timeout=self.args.dns_timeout,
+            )
+
+            self.logger.debug("Using DNS to retrieve domain information")
+            try:
+                ad.dns_resolve(domain=self.domain)
+            except (resolver.LifetimeTimeout, resolver.NoNameservers):
+                self.logger.fail("Bloodhound-python failed to resolve domain information, try specifying the DNS server.")
+                return
+
+            if self.args.kerberos:
+                self.logger.highlight("Using kerberos auth without ccache, getting TGT")
+                auth.get_tgt()
+            if self.args.use_kcache:
+                self.logger.highlight("Using kerberos auth from ccache")
+                auth.load_ccache()
+
+            bloodhound = BloodHound(ad, self.hostname, self.host, self.port)
+            bloodhound.connect()
+
+            try:
+                bloodhound.run(
+                    collect=bh_collect,
+                    num_workers=10,
+                    disable_pooling=False,
+                    timestamp=timestamp,
+                    fileNamePrefix=self.output_filename.split("/")[-1],
+                    computerfile=None,
+                    cachefile=None,
+                    exclude_dcs=False,
+                )
+            except Exception as e:
+                self.logger.fail(f"BloodHound collection failed: {e.__class__.__name__} - {e}")
+                self.logger.debug(f"BloodHound collection failed: {e.__class__.__name__} - {e}", exc_info=True)
+                return
+
+        # Collect ADCS data using CertiHound if requested
+        if "adcs" in collect:
+            adcs_files = self._collect_adcs_for_bloodhound(timestamp)
 
         self.logger.highlight(f"Compressing output into {self.output_filename}_bloodhound.zip")
         list_of_files = os.listdir(os.getcwd())
@@ -1721,3 +1731,81 @@ def bloodhound(self):
                 if each_file.startswith(self.output_filename.split("/")[-1]) and each_file.endswith("json"):
                     z.write(each_file)
                     os.remove(each_file)
+            # Add ADCS files to the zip
+            for adcs_file in adcs_files:
+                if os.path.exists(adcs_file):
+                    z.write(adcs_file, os.path.basename(adcs_file))
+                    os.remove(adcs_file)
+
+    def _collect_adcs_for_bloodhound(self, timestamp):
+        """Collect ADCS data using CertiHound for BloodHound CE integration.
+
+        Args:
+            timestamp: Timestamp prefix for output files.
+
+        Returns:
+            List of file paths to include in the BloodHound zip.
+        """
+        from certihound import ADCSCollector, BloodHoundCEExporter, ImpacketLDAPAdapter
+
+        self.logger.highlight("Collecting ADCS data (CertiHound)...")
+
+        try:
+            # Get domain SID if not already set
+            if not self.sid_domain:
+                search_filter = "(objectClass=domain)"
+                attributes = ["objectSid"]
+                resp = self.search(search_filter, attributes, sizeLimit=1, baseDN=self.baseDN)
+                if resp:
+                    for item in resp:
+                        if isinstance(item, ldapasn1_impacket.SearchResultEntry):
+                            for attr in item["attributes"]:
+                                if str(attr["type"]) == "objectSid":
+                                    from impacket.ldap.ldaptypes import LDAP_SID
+                                    sid = LDAP_SID(bytes(attr["vals"][0]))
+                                    self.sid_domain = sid.formatCanonical()
+                                    break
+
+            if not self.sid_domain:
+                self.logger.fail("Could not retrieve domain SID for ADCS collection")
+                return []
+
+            # Create CertiHound adapter and collector
+            adapter = ImpacketLDAPAdapter(
+                search_func=self.search,
+                domain=self.domain,
+                domain_sid=self.sid_domain,
+            )
+
+            collector = ADCSCollector.from_external(
+                ldap_connection=adapter,
+                domain=self.domain,
+                domain_sid=self.sid_domain,
+            )
+            data = collector.collect_all()
+
+            self.logger.highlight(f"Found {len(data.templates)} certificate templates")
+            self.logger.highlight(f"Found {len(data.enterprise_cas)} Enterprise CAs")
+
+            # Export to BloodHound CE format
+            exporter = BloodHoundCEExporter(data.domain, data.domain_sid)
+            result = exporter.export(data)
+
+            # Write individual JSON files
+            adcs_files = []
+            output = result.to_dict()
+
+            for node_type, content in output.items():
+                filename = f"{timestamp}{node_type}.json"
+                with open(filename, "w") as f:
+                    import json
+                    json.dump(content, f)
+                adcs_files.append(filename)
+                self.logger.debug(f"Wrote ADCS file: {filename}")
+
+            return adcs_files
+
+        except Exception as e:
+            self.logger.fail(f"ADCS collection failed: {e}")
+            self.logger.debug(f"ADCS collection error: {e}", exc_info=True)
+            return []

nxc/protocols/ldap/proto_args.py

@@ -43,6 +43,6 @@ def proto_args(parser, parents):
 
     bgroup = ldap_parser.add_argument_group("Bloodhound Scan", "Options to play with Bloodhoud")
     bgroup.add_argument("--bloodhound", action="store_true", help="Perform a Bloodhound scan")
-    bgroup.add_argument("-c", "--collection", default="Default", help="Which information to collect. Supported: Group, LocalAdmin, Session, Trusts, Default, DCOnly, DCOM, RDP, PSRemote, LoggedOn, Container, ObjectProps, ACL, All. You can specify more than one by separating them with a comma")
+    bgroup.add_argument("-c", "--collection", default="Default", help="Which information to collect. Supported: Group, LocalAdmin, Session, Trusts, Default, DCOnly, DCOM, RDP, PSRemote, LoggedOn, Container, ObjectProps, ACL, ADCS, All. You can specify more than one by separating them with a comma. ADCS requires: pip install netexec[adcs]")
 
     return parser

pyproject.toml

@@ -48,6 +48,7 @@ dependencies = [
     "impacket @ git+https://github.com/fortra/impacket",
     "oscrypto @ git+https://github.com/wbond/oscrypto",
     "pynfsclient @ git+https://github.com/Pennyw0rth/NfsClient",
+    "certihound>=0.1.1",
 ]
 
 [project.urls]