add Schannel certificate authentication via --schannel

nxc/cli.py

@@ -107,6 +107,7 @@ def gen_cli_args():
     certificate_group.add_argument("--pfx-pass", metavar="PFXPASS", help="Password of the pfx certificate")
     certificate_group.add_argument("--pem-cert", metavar="PEMCERT", help="Use certificate authentication from PEM file")
     certificate_group.add_argument("--pem-key", metavar="PEMKEY", help="Private key for the PEM format")
+    certificate_group.add_argument("--schannel", action="store_true", help="Authenticate with the certificate using Schannel instead of PKINIT (LDAP protocol only)")
 
     p_loader = ProtocolLoader()
     protocols = p_loader.get_protocols()

nxc/connection.py

@@ -567,6 +567,12 @@ def login(self):
             if not self.args.username:
                 self.logger.fail("You must specify a username when using certificate authentication")
                 return False
+            if self.args.schannel:
+                if self.args.protocol != "ldap":
+                    self.logger.fail("Schannel authentication is only supported for the LDAP protocol")
+                    return False
+                with sem:
+                    return self.schannel_login()
             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
 
 
@@ -526,3 +530,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 = f"{self.hostname}_{self.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}"
+    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
@@ -434,6 +435,48 @@ def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="",
                 )
                 return False
 
+    def schannel_login(self):
+        cert_file, key_file = pfx_to_pem_files(self)
+        if not cert_file:
+            return False
+
+        self.username = self.args.username[0]
+        try:
+            # Schannel needs TLS: over LDAPS the certificate is sent during the handshake, over LDAP it is sent during StartTLS
+            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} using Schannel")
+            self.ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, certfile=cert_file, keyfile=key_file)
+            self.ldap_connection.login(authenticationChoice="external")
+
+            mapped_user = self.get_ldap_username()
+            if mapped_user:
+                self.username = mapped_user
+
+            self.check_if_admin()
+            self.logger.debug(f"Adding credential: {self.domain}/{self.username} from pfx")
+            self.db.add_credential("certificate", self.domain, self.username, "")
+
+            self.logger.success(f"{self.domain}\\{self.username} from pfx {self.mark_pwned()}")
+
+            if self.username != "":
+                add_user_bh(self.username, self.domain, self.logger, self.config)
+            if self.admin_privs:
+                add_user_bh(f"{self.hostname}$", self.domain, self.logger, self.config)
+            return True
+        except ldap_impacket.LDAPSessionError as e:
+            self.logger.fail(f"{self.domain}\\{self.username} {e!s}", color="red")
+            return False
+        except OSError as e:
+            self.logger.fail(f"{self.domain}\\{self.username} Error connecting to the domain, are you sure the 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 plaintext_login(self, domain, username, password):
         self.username = username
         self.password = password
@@ -618,7 +661,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])