add Schannel certificate authentication via --schannel

nxc/connection.py

@@ -250,6 +250,7 @@ def proto_flow(self):
             # Construct the output file template using os.path.join for OS compatibility
             base_log_dir = os.path.join(NXC_PATH, "logs")
             filename_pattern = f"{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}".replace(":", "-")
+            self.filename_pattern = filename_pattern
             self.output_file_template = os.path.join(base_log_dir, "{output_folder}", filename_pattern)
             # Default output filename for logs
             self.output_filename = os.path.join(base_log_dir, filename_pattern)
@@ -568,6 +569,9 @@ def login(self):
             if not self.args.username:
                 self.logger.fail("You must specify a username when using certificate authentication")
                 return False
+            if hasattr(self.args, "schannel") and self.args.schannel:
+                with sem:
+                    return self.plaintext_login(self.domain, self.args.username[0], "")
             with sem:
                 return pfx_auth(self)
 

nxc/helpers/pfx.py

@@ -65,9 +65,13 @@
 from minikerberos.common.target import KerberosTarget
 from minikerberos.common.ccache import CCACHE
 
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.serialization import pkcs12
+from cryptography.x509 import load_pem_x509_certificate
+
 from impacket.krb5.ccache import CCache as impacket_CCache
 
-from nxc.paths import NXC_PATH
+from nxc.paths import NXC_PATH, TMP_PATH
 from nxc.logger import nxc_logger
 
 
@@ -533,3 +537,40 @@ def pfx_auth(self):
 
     self.logger.info("Successfully authenticated using Certificate")
     return True
+
+
+def pfx_to_pem_files(self):
+    """Convert the provided certificate material (PFX or PEM) into PEM cert and key files in TMP_PATH."""
+    if self.args.pfx_cert or self.args.pfx_base64:
+        if self.args.pfx_base64:
+            with open(self.args.pfx_base64, "rb") as f:
+                pfx_data = base64.b64decode(f.read())
+        else:
+            with open(self.args.pfx_cert, "rb") as f:
+                pfx_data = f.read()
+        pfx_pass = self.args.pfx_pass.encode() if self.args.pfx_pass else None
+        key, cert, _ = pkcs12.load_key_and_certificates(pfx_data, pfx_pass)
+    elif self.args.pem_cert and self.args.pem_key:
+        with open(self.args.pem_cert, "rb") as f:
+            cert = load_pem_x509_certificate(f.read())
+        with open(self.args.pem_key, "rb") as f:
+            key = serialization.load_pem_private_key(f.read(), password=None)
+    else:
+        self.logger.fail("You must either specify a PFX file + optional password or a combination of Cert PEM file and Private key PEM file")
+        return None, None
+
+    basename = self.filename_pattern
+    cert_path = os.path.normpath(os.path.expanduser(f"{TMP_PATH}/{basename}_cert.pem"))
+    key_path = os.path.normpath(os.path.expanduser(f"{TMP_PATH}/{basename}_key.pem"))
+
+    with open(cert_path, "wb") as cert_file:
+        cert_file.write(cert.public_bytes(serialization.Encoding.PEM))
+
+    with open(key_path, "wb") as key_file:
+        key_file.write(key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            encryption_algorithm=serialization.NoEncryption(),
+        ))
+
+    return cert_path, key_path

nxc/protocols/ldap.py

@@ -41,6 +41,7 @@
 from nxc.config import process_secret, host_info_colors
 from nxc.connection import connection
 from nxc.helpers.bloodhound import add_user_bh
+from nxc.helpers.pfx import pfx_to_pem_files
 from nxc.helpers.misc import get_bloodhound_info, convert, d2b, parse_argument
 from nxc.logger import NXCAdapter
 from nxc.protocols.ldap.bloodhound import BloodHound, resolve_collection_methods
@@ -451,21 +452,44 @@ def plaintext_login(self, domain, username, password):
                     hash_asreproast.write(f"{hash_tgt}\n")
             return False
 
+        cert_file = key_file = None
+        if self.args.schannel:
+            cert_file, key_file = pfx_to_pem_files(self)
+            if not cert_file:
+                return False
+
         try:
             # Connect to LDAP
             self.logger.extra["protocol"] = "LDAPS" if self.port == 636 else "LDAP"
             self.logger.extra["port"] = "636" if self.port == 636 else "389"
             proto = "ldaps" if self.port == 636 else "ldap"
             ldap_url = f"{proto}://{self.target}"
-            self.logger.info(f"Connecting to {ldap_url} - {self.baseDN} - {self.host} [3]")
-            self.ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, signing=self.auth_choice != "simple", timeout=self.args.ldap_timeout)
-            self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=self.auth_choice)
+            authentication_choice = "external" if self.args.schannel else self.auth_choice
+            conn_kwargs = {"url": ldap_url, "baseDN": self.baseDN, "dstIp": self.host, "signing": self.auth_choice != "simple", "timeout": self.args.ldap_timeout}
+            if self.args.schannel:
+                conn_kwargs["certfile"] = cert_file
+                conn_kwargs["keyfile"] = key_file
+
+            self.logger.info(f"Connecting to {ldap_url} using Schannel" if self.args.schannel else f"Connecting to {ldap_url} - {self.baseDN} - {self.host} [3]")
+            self.ldap_connection = ldap_impacket.LDAPConnection(**conn_kwargs)
+            self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=authentication_choice)
+
+            if self.args.schannel:
+                mapped_user = self.get_ldap_username()
+                if mapped_user:
+                    self.username = mapped_user
+
             self.check_if_admin()
-            self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.password}")
-            self.db.add_credential("plaintext", domain, self.username, self.password)
 
-            # Prepare success credential text
-            self.logger.success(f"{domain}\\{self.username}:{process_secret(self.password)} {self.mark_pwned()}")
+            if self.args.schannel:
+                self.logger.debug(f"Adding credential: {self.domain}/{self.username} from certificate")
+                self.db.add_credential("certificate", self.domain, self.username, "")
+                self.logger.success(f"{self.domain}\\{self.username} from certificate {self.mark_pwned()}")
+            else:
+                self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.password}")
+                self.db.add_credential("plaintext", domain, self.username, self.password)
+                # Prepare success credential text
+                self.logger.success(f"{domain}\\{self.username}:{process_secret(self.password)} {self.mark_pwned()}")
 
             if self.username != "":
                 add_user_bh(self.username, self.domain, self.logger, self.config)
@@ -517,6 +541,10 @@ def plaintext_login(self, domain, username, password):
         except OSError as e:
             self.logger.fail(f"{self.domain}\\{self.username}:{process_secret(self.password)} {'Error connecting to the domain, are you sure LDAP service is running on the target?'} \nError: {e}")
             return False
+        finally:
+            for tmp_file in (cert_file, key_file):
+                if tmp_file and os.path.exists(tmp_file):
+                    os.remove(tmp_file)
 
     def hash_login(self, domain, username, ntlm_hash):
         self.logger.extra["protocol"] = "LDAP"
@@ -622,7 +650,7 @@ def check_if_admin(self):
         resp = self.search(search_filter, attributes, baseDN=self.baseDN)
         resp_parsed = parse_result_attributes(resp)
 
-        if resp and (self.password != "" or self.lmhash != "" or self.nthash != "" or self.aesKey != "" or self.use_kcache) and self.username != "":
+        if resp and (self.password != "" or self.lmhash != "" or self.nthash != "" or self.aesKey != "" or self.use_kcache or self.args.schannel) and self.username != "":
             for item in resp_parsed:
                 self.sid_domain = "-".join(item["objectSid"].split("-")[:-1])
 

nxc/protocols/ldap/proto_args.py

@@ -10,6 +10,7 @@ def proto_args(parser, parents):
     ldap_parser.add_argument("--port", type=int, default=389, action=DefaultTrackingAction, help="LDAP port")
     ldap_parser.add_argument("--ldap-timeout", type=int, default=3, help="LDAP connection timeout")
     ldap_parser.add_argument("-d", metavar="DOMAIN", dest="domain", type=str, default=None, help="domain to authenticate to")
+    ldap_parser.add_argument("--schannel", action="store_true", help="Authenticate with the certificate using Schannel instead of PKINIT")
 
     egroup = ldap_parser.add_argument_group("Retrieve hash on the remote DC", "Options to get hashes from Kerberos")
     egroup.add_argument("--asreproast", help="Output AS_REP response to crack with hashcat to file")

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "beautifulsoup4>=4.11,<5",
     "bloodhound-ce>=1.8.0",
     "certihound>=0.1.1",
+    "cryptography>=42.0.8",
     "dploot>=3.2.2",
     "dsinternals>=1.2.4",
     "jwt>=1.3.1",