add Schannel certificate authentication via --schannel

nxc/connection.py

@@ -249,10 +249,10 @@ 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.output_file_template = os.path.join(base_log_dir, "{output_folder}", filename_pattern)
+            self.filename_pattern = f"{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}".replace(":", "-")
+            self.output_file_template = os.path.join(base_log_dir, "{output_folder}", self.filename_pattern)
             # Default output filename for logs
-            self.output_filename = os.path.join(base_log_dir, filename_pattern)
+            self.output_filename = os.path.join(base_log_dir, self.filename_pattern)
 
             self.print_host_info()
             if self.login() or (self.username == "" and self.password == "" and self.protocol != "mssql"):
@@ -565,6 +565,10 @@ def login(self):
 
         if self.args.pfx_cert or self.args.pfx_base64 or self.args.pem_cert:
             self.logger.debug("Trying to authenticate using Certificate pfx")
+            if self.args.protocol == "ldap" and self.args.schannel:
+                # Schannel maps the certificate to an account server-side, so the supplied username is ignored
+                with sem:
+                    return self.plaintext_login(self.domain, "", "")
             if not self.args.username:
                 self.logger.fail("You must specify a username when using certificate authentication")
                 return False

nxc/helpers/pfx.py

@@ -70,18 +70,54 @@
 
 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
 
 
-def _load_asn1_certificate(data):
-    """Load a certificate as an asn1crypto object, accepting both PEM and DER input
-
-    The certificate is parsed by cryptography and re-encoded, so that malformed input is
-    rejected here instead of failing later inside asn1crypto, which parses lazily
-    """
-    cert = load_pem_x509_certificate(data) if b"-----BEGIN" in data else load_der_x509_certificate(data)
-    return _to_asn1_certificate(cert)
+def load_pfx_data(pfxdata, pfxpass):
+    """Load the certificate and private key of a PFX as cryptography objects"""
+    if isinstance(pfxpass, str):
+        pfxpass = pfxpass.encode()
+    # cryptography requires None (not an empty password) for a pfx without password
+    privkey, cert, _ = pkcs12.load_key_and_certificates(pfxdata, pfxpass if pfxpass else None)
+    # a pfx holding only a certificate parses fine, so check the key before the certificate
+    if privkey is None:
+        raise Exception("No private key found in the PFX file")
+    if cert is None:
+        raise Exception("No certificate found in the PFX file")
+    return cert, privkey
+
+
+def load_pem_files(certfile, privkeyfile):
+    """Load a certificate and a private key from separate files as cryptography objects, accepting both PEM and DER input"""
+    with open(certfile, "rb") as f:
+        certdata = f.read()
+    cert = load_pem_x509_certificate(certdata) if b"-----BEGIN" in certdata else load_der_x509_certificate(certdata)
+
+    with open(privkeyfile, "rb") as f:
+        keydata = f.read()
+    try:
+        try:
+            privkey = serialization.load_pem_private_key(keydata, password=None)
+        except ValueError:
+            privkey = serialization.load_der_private_key(keydata, password=None)
+    except TypeError as e:
+        # raised by both loaders when the key is encrypted and no password was given
+        raise Exception(f"Private key {privkeyfile} is password protected, which is not supported. Decrypt it first with: openssl rsa -in {privkeyfile} -out decrypted.pem") from e
+    return cert, privkey
+
+
+def load_cert_and_key(args):
+    """Load the certificate material given on the command line (PFX, base64 encoded PFX or cert + key files)"""
+    if args.pfx_cert or args.pfx_base64:
+        with open(args.pfx_cert or args.pfx_base64, "rb") as f:
+            pfxdata = f.read()
+        if args.pfx_base64:
+            pfxdata = base64.b64decode(pfxdata)
+        return load_pfx_data(pfxdata, args.pfx_pass)
+    if args.pem_cert and args.pem_key:
+        return load_pem_files(args.pem_cert, args.pem_key)
+    raise Exception("You must either specify a PFX file + optional password or a combination of Cert PEM file and Private key PEM file")
 
 
 def _to_asn1_certificate(cert):
@@ -113,51 +149,13 @@ def __init__(self):
         self.diffie = None
 
     @staticmethod
-    def from_pfx(pfxfile, pfxpass, dh_params=None, b64=False):
-        with open(pfxfile, "rb") as f:
-            pfxdata = f.read()
-
-        if b64:
-            pfxdata = base64.b64decode(pfxdata)
-
-        return myPKINIT.from_pfx_data(pfxdata, pfxpass, dh_params)
-
-    @staticmethod
-    def from_pfx_data(pfxdata, pfxpass, dh_params=None):
+    def from_cert_and_key(cert, privkey, dh_params=None):
         pkinit = myPKINIT()
-        if isinstance(pfxpass, str):
-            pfxpass = pfxpass.encode()
-        # cryptography requires None (not an empty password) for a pfx without password
-        privkey, cert, _ = pkcs12.load_key_and_certificates(pfxdata, pfxpass if pfxpass else None)
-        # a pfx holding only a certificate parses fine, so check the key before the certificate
-        if privkey is None:
-            raise Exception("No private key found in the PFX file")
-        if cert is None:
-            raise Exception("No certificate found in the PFX file")
         pkinit.privkey = _check_rsa_privkey(privkey)
         pkinit.certificate = _to_asn1_certificate(cert)
         pkinit.setup(dh_params=dh_params)
         return pkinit
 
-    @staticmethod
-    def from_pem(certfile, privkeyfile, dh_params=None):
-        pkinit = myPKINIT()
-        with open(certfile, "rb") as f:
-            pkinit.certificate = _load_asn1_certificate(f.read())
-        with open(privkeyfile, "rb") as f:
-            keydata = f.read()
-        try:
-            try:
-                privkey = serialization.load_pem_private_key(keydata, password=None)
-            except ValueError:
-                privkey = serialization.load_der_private_key(keydata, password=None)
-        except TypeError as e:
-            # raised by both loaders when the key is encrypted and no password was given
-            raise Exception(f"Private key {privkeyfile} is password protected, which is not supported. Decrypt it first with: openssl rsa -in {privkeyfile} -out decrypted.pem") from e
-        pkinit.privkey = _check_rsa_privkey(privkey)
-        pkinit.setup(dh_params=dh_params)
-        return pkinit
-
     def sign_authpack(self, data, wrap_signed=False):
         return self.sign_authpack_native(data, wrap_signed)
 
@@ -514,14 +512,7 @@ def pfx_auth(self):
 
     # Load the certificate and key from file
     try:
-        if self.args.pfx_cert or self.args.pfx_base64:
-            pfx = self.args.pfx_cert if self.args.pfx_cert else self.args.pfx_base64
-            ini = myPKINIT.from_pfx(pfx, self.args.pfx_pass, dhparams, bool(self.args.pfx_base64))
-        elif self.args.pem_cert and self.args.pem_key:
-            ini = myPKINIT.from_pem(self.args.pem_cert, self.args.pem_key, dhparams)
-        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
+        ini = myPKINIT.from_cert_and_key(*load_cert_and_key(self.args), dhparams)
     except FileNotFoundError as e:
         self.logger.fail(f"Certificate or key file not found: {e.filename}")
         return False
@@ -568,3 +559,31 @@ 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."""
+    try:
+        cert, key = load_cert_and_key(self.args)
+    except FileNotFoundError as e:
+        self.logger.fail(f"Certificate or key file not found: {e.filename}")
+        return None, None
+    except Exception as e:
+        self.logger.fail(f"Failed to load certificate/key: {e}")
+        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
@@ -65,6 +66,9 @@
     "KDC_ERR_PREAUTH_FAILED": "KDC_ERR_PREAUTH_FAILED",
 }
 
+# LDAP authenticationChoice sent to the DC, as displayed to the operator
+auth_methods = {"sasl": "NTLM", "simple": "SIMPLE bind", "external": "Schannel"}
+
 
 class ldap(connection):
     def __init__(self, args, db, host):
@@ -334,7 +338,7 @@ def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="",
             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} [1]")
+            self.logger.info(f"Connecting to {ldap_url} - {self.baseDN} - {self.host} using Kerberos")
             self.ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout)
             self.ldap_connection.kerberosLogin(username, password, domain, self.lmhash, self.nthash, aesKey, kdcHost=kdcHost, useCache=useCache)
             if self.username == "":
@@ -393,7 +397,7 @@ def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="",
                     self.logger.extra["port"] = "636"
                     self.port = 636
                     ldaps_url = f"ldaps://{self.target}"
-                    self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} [2]")
+                    self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using Kerberos")
                     self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout)
                     self.ldap_connection.kerberosLogin(username, password, domain, self.lmhash, self.nthash, aesKey, kdcHost=kdcHost, useCache=useCache)
                     if self.username == "":
@@ -451,21 +455,41 @@ 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
+            self.logger.info(f"Connecting to {ldap_url} - {self.baseDN} - {self.host} using {auth_methods[authentication_choice]}")
+            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, certfile=cert_file, keyfile=key_file)
+            self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=authentication_choice)
+
+            if self.args.schannel:
+                # The certificate is mapped to an account server-side, so whoami is the only way to know who we are
+                self.username = self.get_ldap_username()
+                if not self.username:
+                    self.logger.fail("Authenticated with the certificate but the mapped user could not be retrieved with LDAP whoami")
+                    return False
+
             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)
@@ -486,7 +510,7 @@ def plaintext_login(self, domain, username, password):
                     self.logger.extra["port"] = "636"
                     self.port = 636
                     ldaps_url = f"ldaps://{self.target}"
-                    self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} [4]")
+                    self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using {auth_methods[self.auth_choice]}")
                     self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout)
                     self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=self.auth_choice)
                     self.check_if_admin()
@@ -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"
@@ -553,7 +581,7 @@ def hash_login(self, domain, username, ntlm_hash):
             self.logger.extra["port"] = "636" if self.port == 636 else "389"
             proto = "ldaps" if self.port == 636 else "ldap"
             ldaps_url = f"{proto}://{self.target}"
-            self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host}")
+            self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using NT hash")
             self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout)
             self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash)
             self.check_if_admin()
@@ -579,7 +607,7 @@ def hash_login(self, domain, username, ntlm_hash):
                     self.logger.extra["port"] = "636"
                     self.port = 636
                     ldaps_url = f"ldaps://{self.target}"
-                    self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host}")
+                    self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using NT hash")
                     self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout)
                     self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash)
                     self.check_if_admin()
@@ -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")