Move cert auth logic to helpers function

Author: NeffIsBack

nxc/connection.py

@@ -1,13 +1,11 @@
 import random
-import os
 import sys
 import contextlib
 
 from os.path import isfile
 from threading import BoundedSemaphore
 from functools import wraps
 from time import sleep
-from datetime import datetime
 from ipaddress import ip_address
 from dns import resolver, rdatatype
 from socket import AF_UNSPEC, SOCK_DGRAM, IPPROTO_IP, AI_CANONNAME, getaddrinfo
@@ -18,13 +16,8 @@
 from nxc.logger import nxc_logger, NXCAdapter
 from nxc.context import Context
 from nxc.protocols.ldap.laps import laps_search
-from nxc.helpers.pfx import myPKINIT, GETPAC
+from nxc.helpers.pfx import pfx_auth
 
-from minikerberos.network.clientsocket import KerberosClientSocket
-from minikerberos.common.target import KerberosTarget
-from minikerberos.common.ccache import CCACHE
-
-from impacket.krb5.ccache import CCache
 from impacket.dcerpc.v5 import transport
 
 sem = BoundedSemaphore(1)
@@ -563,57 +556,7 @@ def login(self):
                 self.logger.fail("You must specify a username when using certificate authentication")
                 return False
             with sem:
-                # Static DH params because the ones generated by cryptography are considered unsafe by AD for some weird reason
-                dhparams = {
-                    "p": int("00ffffffffffffffffc90fdaa22168c234c4c6628b80dc1cd129024e088a67cc74020bbea63b139b22514a08798e3404ddef9519b3cd3a431b302b0a6df25f14374fe1356d6d51c245e485b576625e7ec6f44c42e9a637ed6b0bff5cb6f406b7edee386bfb5a899fa5ae9f24117c4b1fe649286651ece65381ffffffffffffffff", 16),
-                    "g": 2
-                }
-                self.logger.info("Loading certificate and key from file")
-
-                # Load the certificate and key from file
-                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.cert_pem and self.args.key_pem:
-                    ini = myPKINIT.from_pem(self.args.cert_pem, self.args.key_pem, 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
-
-                username = self.args.username[0]
-                log_ccache = os.path.expanduser(f"~/.nxc/logs/{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}-{username}.ccache".replace(":", "-"))
-
-                # Request a TGT with the cert data
-                req = ini.build_asreq(self.domain, username)
-                self.logger.info("Requesting TGT")
-
-                sock = KerberosClientSocket(KerberosTarget(self.host))
-                try:
-                    res = sock.sendrecv(req)
-                except Exception as e:
-                    self.logger.fail(str(e))
-                    return False
-
-                encasrep, session_key, cipher, key = ini.decrypt_asrep(res.native)
-                ccache_minikerberos = CCACHE()
-                ccache_minikerberos.add_tgt(res.native, encasrep)
-                ccache_minikerberos.to_file(log_ccache)
-                self.logger.info(f"Saved TGT to file { log_ccache }")
-                self.logger.info(f"Using Kerberos Cache { log_ccache }")
-                ccache = CCache.loadFile(log_ccache)
-                principal = f"krbtgt/{self.domain.upper()}@{self.domain.upper()}"
-                creds = ccache.getCredential(principal)
-                if creds is not None:
-                    tgt = creds.toTGT()
-                    dumper = GETPAC(username, self.domain, self.host, key, tgt)
-                    nthash = dumper.dump()
-                    if not self.kerberos:
-                        self.hash_login(self.domain, username, nthash)
-                    else:
-                        self.kerberos_login(self.domain, username, "", nthash, "", self.kdcHost, False)
-
-                self.logger.info("Successfully authenticated using Certificate")
-                return True
+                return pfx_auth(self)
 
         if hasattr(self.args, "laps") and self.args.laps:
             self.logger.debug("Trying to authenticate using LAPS")

nxc/helpers/pfx.py

@@ -26,6 +26,7 @@
 #  Dirk-jan Mollema (@_dirkjan)
 #
 
+import os
 import secrets
 import hashlib
 import datetime
@@ -61,6 +62,14 @@
     PAC_CREDENTIAL_DATA, NTLM_SUPPLEMENTAL_CREDENTIAL
 from impacket.krb5.types import Principal, KerberosTime, Ticket
 
+# Imports for pfx_auth
+from minikerberos.network.clientsocket import KerberosClientSocket
+from minikerberos.common.target import KerberosTarget
+from minikerberos.common.ccache import CCACHE
+
+from impacket.krb5.ccache import CCache
+
+
 class myPKINIT(PKINIT):
     """
     Copy of minikerberos PKINIT
@@ -475,4 +484,59 @@ def dump(self):
         # S4USelf + U2U uses this other key
         plainText = cipher.decrypt(sessionKey, 2, cipherText)
         specialkey = Key(18, unhexlify(self.__asrep_key))
-        return self.printPac(plainText, specialkey)
+        return self.printPac(plainText, specialkey)
+    
+
+def pfx_auth(self):
+    """Handles the authentication using a PFX or PEM file"""
+    # Static DH params because the ones generated by cryptography are considered unsafe by AD for some weird reason
+    dhparams = {
+        "p": int("00ffffffffffffffffc90fdaa22168c234c4c6628b80dc1cd129024e088a67cc74020bbea63b139b22514a08798e3404ddef9519b3cd3a431b302b0a6df25f14374fe1356d6d51c245e485b576625e7ec6f44c42e9a637ed6b0bff5cb6f406b7edee386bfb5a899fa5ae9f24117c4b1fe649286651ece65381ffffffffffffffff", 16),
+        "g": 2
+    }
+    self.logger.info("Loading certificate and key from file")
+
+    # Load the certificate and key from file
+    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.cert_pem and self.args.key_pem:
+        ini = myPKINIT.from_pem(self.args.cert_pem, self.args.key_pem, 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
+
+    username = self.args.username[0]
+    log_ccache = os.path.expanduser(f"~/.nxc/logs/{self.hostname}_{self.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}-{username}.ccache".replace(":", "-"))
+
+    # Request a TGT with the cert data
+    req = ini.build_asreq(self.domain, username)
+    self.logger.info("Requesting TGT")
+
+    sock = KerberosClientSocket(KerberosTarget(self.host))
+    try:
+        res = sock.sendrecv(req)
+    except Exception as e:
+        self.logger.fail(str(e))
+        return False
+
+    encasrep, session_key, cipher, key = ini.decrypt_asrep(res.native)
+    ccache_minikerberos = CCACHE()
+    ccache_minikerberos.add_tgt(res.native, encasrep)
+    ccache_minikerberos.to_file(log_ccache)
+    self.logger.info(f"Saved TGT to file { log_ccache }")
+    self.logger.info(f"Using Kerberos Cache { log_ccache }")
+    ccache = CCache.loadFile(log_ccache)
+    principal = f"krbtgt/{self.domain.upper()}@{self.domain.upper()}"
+    creds = ccache.getCredential(principal)
+    if creds is not None:
+        tgt = creds.toTGT()
+        dumper = GETPAC(username, self.domain, self.host, key, tgt)
+        nthash = dumper.dump()
+        if not self.kerberos:
+            self.hash_login(self.domain, username, nthash)
+        else:
+            self.kerberos_login(self.domain, username, "", nthash, "", self.kdcHost, False)
+
+    self.logger.info("Successfully authenticated using Certificate")
+    return True