Merge pull request Pennyw0rth#629 from Pennyw0rth/neff-update-ruff

Update Ruff and Fix Linting

Author: NeffIsBack

nxc/connection.py

@@ -293,8 +293,7 @@ def call_modules(self):
                 module.on_admin_login(context, self)
 
     def inc_failed_login(self, username):
-        global global_failed_logins
-        global user_failed_logins
+        global global_failed_logins, user_failed_logins
 
         if username not in user_failed_logins:
             user_failed_logins[username] = 0
@@ -304,16 +303,15 @@ def inc_failed_login(self, username):
         self.failed_logins += 1
 
     def over_fail_limit(self, username):
-        global global_failed_logins
-        global user_failed_logins
+        global global_failed_logins, user_failed_logins
 
         if global_failed_logins == self.args.gfail_limit:
             return True
 
         if self.failed_logins == self.args.fail_limit:
             return True
 
-        if username in user_failed_logins and self.args.ufail_limit == user_failed_logins[username]:
+        if username in user_failed_logins and self.args.ufail_limit == user_failed_logins[username]:  # noqa: SIM103
             return True
 
         return False

nxc/database.py

@@ -111,6 +111,7 @@ def initialize_db():
     # Even if the default workspace exists, we still need to check if every protocol has a database (in case of a new protocol)
     init_protocol_dbs("default")
 
+
 def format_host_query(q, filter_term, HostsTable):
     """One annoying thing is that if you search for an ip such as '10.10.10.5',
     it will return 10.10.10.5 and 10.10.10.52, so we have to check if its an ip address first
@@ -141,6 +142,7 @@ def format_host_query(q, filter_term, HostsTable):
 
     return q
 
+
 class BaseDB:
     def __init__(self, db_engine):
         self.db_engine = db_engine

nxc/helpers/args.py

@@ -1,6 +1,7 @@
 from argparse import ArgumentDefaultsHelpFormatter, SUPPRESS, OPTIONAL, ZERO_OR_MORE
 from argparse import Action
 
+
 class DisplayDefaultsNotNone(ArgumentDefaultsHelpFormatter):
     def _get_help_string(self, action):
         help_string = action.help

nxc/helpers/even6_parser.py

@@ -5,6 +5,7 @@
 
 from datetime import datetime
 
+
 class Substitution:
     def __init__(self, buf, offset):
         (sub_token, sub_id, sub_type) = struct.unpack_from("<BHB", buf, offset)
@@ -46,6 +47,7 @@ def xml(self, template=None):
         else:
             print("Unknown value type", hex(value.type))
 
+
 class Value:
     def __init__(self, buf, offset):
         token, string_type, length = struct.unpack_from("<BBH", buf, offset)
@@ -56,6 +58,7 @@ def __init__(self, buf, offset):
     def xml(self, template=None):
         return self._val
 
+
 class Attribute:
     def __init__(self, buf, offset):
         struct.unpack_from("<B", buf, offset)
@@ -75,13 +78,15 @@ def xml(self, template=None):
         val = self._value.xml(template)
         return None if val is None else f'{self._name.val}="{val}"'
 
+
 class Name:
     def __init__(self, buf, offset):
         hashs, length = struct.unpack_from("<HH", buf, offset)
 
         self.val = buf[offset + 4:offset + 4 + length * 2].decode("utf16")
         self.length = 4 + (length + 1) * 2
 
+
 class Element:
     def __init__(self, buf, offset):
         token, dependency_id, length = struct.unpack_from("<BHI", buf, offset)
@@ -151,6 +156,7 @@ def xml(self, template=None):
             children = (x.xml(template) for x in self._children)
             return "<{}{}>{}</{}>".format(self._name.val, attrs, "".join(children), self._name.val)
 
+
 class ValueSpec:
     def __init__(self, buf, offset, value_offset):
         self.length, self.type, value_eof = struct.unpack_from("<HBB", buf, offset)
@@ -159,6 +165,7 @@ def __init__(self, buf, offset, value_offset):
         if self.type == 0x21:
             self.template = BinXML(buf, value_offset)
 
+
 class TemplateInstance:
     def __init__(self, buf, offset):
         token, unknown0, guid, length, next_token = struct.unpack_from("<BB16sIB", buf, offset)
@@ -179,6 +186,7 @@ def __init__(self, buf, offset):
     def xml(self, template=None):
         return self._xml.xml(self)
 
+
 class BinXML:
     def __init__(self, buf, offset):
         header_token, major_version, minor_version, flags, next_token = struct.unpack_from("<BBBBB", buf, offset)
@@ -195,6 +203,7 @@ def __init__(self, buf, offset):
     def xml(self, template=None):
         return self._element.xml(template)
 
+
 class ResultSet:
     def __init__(self, buf):
         total_size, header_size, event_offset, bookmark_offset, binxml_size = struct.unpack_from("<IIIII", buf)

nxc/helpers/misc.py

@@ -6,6 +6,7 @@
 
 from ipaddress import ip_address
 
+
 def identify_target_file(target_file):
     with open(target_file) as target_file_handle:
         for i, line in enumerate(target_file_handle):
@@ -23,7 +24,7 @@ def gen_random_string(length=10):
 
 
 def validate_ntlm(data):
-    allowed = re.compile("^[0-9a-f]{32}", re.IGNORECASE)
+    allowed = re.compile(r"^[0-9a-f]{32}", re.IGNORECASE)
     return bool(allowed.match(data))
 
 
@@ -79,6 +80,7 @@ def _access_check(fn, mode):
                 if _access_check(name, mode):
                     return name
 
+
 def get_bloodhound_info():
     """
     Detect which BloodHound package is installed (regular or CE) and its version.
@@ -136,6 +138,7 @@ def get_bloodhound_info():
 
     return package_name, version, is_ce
 
+
 def detect_if_ip(target):
     try:
         ip_address(target)

nxc/helpers/pfx.py

@@ -30,7 +30,6 @@
 import secrets
 import hashlib
 import datetime
-import logging
 import random
 import base64
 
@@ -47,8 +46,7 @@
 from minikerberos.pkinit import PKINIT, DirtyDH
 from minikerberos.protocol.constants import NAME_TYPE, PaDataType
 from minikerberos.protocol.encryption import Enctype, _enctype_table, Key
-from minikerberos.protocol.asn1_structs import KDC_REQ_BODY, PrincipalName, KDCOptions, EncASRepPart, AS_REQ, PADATA_TYPE, \
-    PA_PAC_REQUEST
+from minikerberos.protocol.asn1_structs import KDC_REQ_BODY, PrincipalName, KDCOptions, EncASRepPart, AS_REQ, PADATA_TYPE, PA_PAC_REQUEST
 from minikerberos.protocol.rfc4556 import PKAuthenticator, AuthPack, PA_PK_AS_REP, KDCDHKeyInfo, PA_PK_AS_REQ
 
 from pyasn1.codec.der import decoder, encoder
@@ -70,6 +68,7 @@
 from impacket.krb5.ccache import CCache as impacket_CCache
 
 from nxc.paths import NXC_PATH
+from nxc.logger import nxc_logger
 
 
 class myPKINIT(PKINIT):
@@ -304,8 +303,8 @@ def truncate_key(value, keysize):
 
         key = Key(cipher.enctype, t_key)
         enc_data = as_rep["enc-part"]["cipher"]
-        logging.info("AS-REP encryption key (you might need this later):")
-        logging.info(hexlify(t_key).decode("utf-8"))
+        nxc_logger.info("AS-REP encryption key (you might need this later):")
+        nxc_logger.info(hexlify(t_key).decode("utf-8"))
         dec_data = cipher.decrypt(key, 3, enc_data)
         encasrep = EncASRepPart.load(dec_data).native
         cipher = _enctype_table[int(encasrep["key"]["keytype"])]
@@ -327,34 +326,27 @@ def printPac(self, data, key=None):
         for _bufferN in range(pacType["cBuffers"]):
             infoBuffer = PAC_INFO_BUFFER(buff)
             data = pacType["Buffers"][infoBuffer["Offset"] - 8:][:infoBuffer["cbBufferSize"]]
-            if logging.getLogger().level == logging.DEBUG:
-                print("TYPE 0x%x" % infoBuffer["ulType"])
+            nxc_logger.debug(f"TYPE 0x{infoBuffer['ulType']}")
             if infoBuffer["ulType"] == 2:
                 found = True
                 credinfo = PAC_CREDENTIAL_INFO(data)
-                if logging.getLogger().level == logging.DEBUG:
-                    credinfo.dump()
                 newCipher = _enctype_table[credinfo["EncryptionType"]]
                 out = newCipher.decrypt(key, 16, credinfo["SerializedData"])
                 type1 = TypeSerialization1(out)
                 # I'm skipping here 4 bytes with its the ReferentID for the pointer
                 newdata = out[len(type1) + 4:]
                 pcc = PAC_CREDENTIAL_DATA(newdata)
-                if logging.getLogger().level == logging.DEBUG:
-                    pcc.dump()
                 for cred in pcc["Credentials"]:
                     credstruct = NTLM_SUPPLEMENTAL_CREDENTIAL(b"".join(cred["Credentials"]))
-                    if logging.getLogger().level == logging.DEBUG:
-                        credstruct.dump()
 
-                    logging.info("Recovered NT Hash")
-                    logging.info(hexlify(credstruct["NtPassword"]).decode("utf-8"))
+                    nxc_logger.info("Recovered NT Hash")
+                    nxc_logger.info(hexlify(credstruct["NtPassword"]).decode("utf-8"))
                     nthash = hexlify(credstruct["NtPassword"]).decode("utf-8")
 
             buff = buff[len(infoBuffer):]
 
         if not found:
-            logging.info("Did not find the PAC_CREDENTIAL_INFO in the PAC. Are you sure your TGT originated from a PKINIT operation?")
+            nxc_logger.info("Did not find the PAC_CREDENTIAL_INFO in the PAC. Are you sure your TGT originated from a PKINIT operation?")
         return nthash
 
     def __init__(self, username, domain, kdcHost, key, tgt):
@@ -399,10 +391,8 @@ def dump(self):
         authenticator["cusec"] = now.microsecond
         authenticator["ctime"] = KerberosTime.to_asn1(now)
 
-        if logging.getLogger().level == logging.DEBUG:
-            logging.debug("AUTHENTICATOR")
-            print(authenticator.prettyPrint())
-            print("\n")
+        nxc_logger.debug("AUTHENTICATOR")
+        nxc_logger.debug(authenticator.prettyPrint() + "\n")
 
         encodedAuthenticator = encoder.encode(authenticator)
 
@@ -452,23 +442,18 @@ def dump(self):
 
         myTicket = ticket.to_asn1(TicketAsn1())
         seq_set_iter(reqBody, "additional-tickets", (myTicket,))
-        if logging.getLogger().level == logging.DEBUG:
-            logging.debug("Final TGS")
-            print(tgsReq.prettyPrint())
-        if logging.getLogger().level == logging.DEBUG:
-            logging.debug("Final TGS")
-            print(tgsReq.prettyPrint())
+        nxc_logger.debug("Final TGS")
+        nxc_logger.debug(tgsReq.prettyPrint())
 
         message = encoder.encode(tgsReq)
-        logging.info("Requesting ticket to self with PAC")
+        nxc_logger.info("Requesting ticket to self with PAC")
 
         r = sendReceive(message, self.__domain, self.__kdcHost)
 
         tgs = decoder.decode(r, asn1Spec=TGS_REP())[0]
 
-        if logging.getLogger().level == logging.DEBUG:
-            logging.debug("TGS_REP")
-            print(tgs.prettyPrint())
+        nxc_logger.debug("TGS_REP")
+        nxc_logger.debug(tgs.prettyPrint())
 
         cipherText = tgs["ticket"]["enc-part"]["cipher"]
 

nxc/helpers/powershell.py

@@ -12,6 +12,7 @@
 
 obfuscate_ps_scripts = False
 
+
 def replace_singles(s):
     """Replaces single quotes with a double quote
     We do this because quoting is very important in PowerShell, and we are doing multiple layers:
@@ -27,6 +28,7 @@ def replace_singles(s):
     """
     return s.replace("'", r"\"")
 
+
 def get_ps_script(path):
     """Generates a full path to a PowerShell script given a relative path.
 
@@ -56,19 +58,6 @@ def encode_ps_command(command):
     return b64encode(command.encode("UTF-16LE")).decode()
 
 
-def is_powershell_installed():
-    """
-    Check if PowerShell is installed.
-
-    Returns
-    -------
-        bool: True if PowerShell is installed, False otherwise.
-    """
-    if which("powershell"):
-        return True
-    return False
-
-
 def obfs_ps_script(path_to_script):
     """
     Obfuscates a PowerShell script.
@@ -90,7 +79,7 @@ def obfs_ps_script(path_to_script):
     obfs_script_dir = os.path.join(NXC_PATH, "obfuscated_scripts")
     obfs_ps_script = os.path.join(obfs_script_dir, ps_script)
 
-    if is_powershell_installed() and obfuscate_ps_scripts:
+    if bool(which("powershell")) and obfuscate_ps_scripts:
         if os.path.exists(obfs_ps_script):
             nxc_logger.display("Using cached obfuscated Powershell script")
             with open(obfs_ps_script) as script:
@@ -116,12 +105,11 @@ def obfs_ps_script(path_to_script):
             and debug statements from a PowerShell source file.
             """
             # strip block comments
-            stripped_code = re.sub(re.compile("<#.*?#>", re.DOTALL), "", script.read())
+            stripped_code = re.sub(re.compile(r"<#.*?#>", re.DOTALL), "", script.read())
             # strip blank lines, lines starting with #, and verbose/debug statements
             return "\n".join([line for line in stripped_code.split("\n") if ((line.strip() != "") and (not line.strip().startswith("#")) and (not line.strip().lower().startswith("write-verbose ")) and (not line.strip().lower().startswith("write-debug ")))])
 
 
-
 def create_ps_command(ps_command, force_ps32=False, obfs=False, custom_amsi=None, encode=True):
     """
     Generates a PowerShell command based on the provided `ps_command` parameter.
@@ -139,7 +127,7 @@ def create_ps_command(ps_command, force_ps32=False, obfs=False, custom_amsi=None
         str: The generated PowerShell command.
     """
     nxc_logger.debug(f"Creating PS command parameters: {ps_command=}, {force_ps32=}, {obfs=}, {custom_amsi=}, {encode=}")
-    
+
     if custom_amsi:
         nxc_logger.debug(f"Using custom AMSI bypass script: {custom_amsi}")
         with open(custom_amsi) as file_in:
@@ -154,7 +142,7 @@ def create_ps_command(ps_command, force_ps32=False, obfs=False, custom_amsi=None
         command = amsi_bypass + f"$functions = {{function Command-ToExecute{{{amsi_bypass + ps_command}}}}}; if ($Env:PROCESSOR_ARCHITECTURE -eq 'AMD64'){{$job = Start-Job -InitializationScript $functions -ScriptBlock {{Command-ToExecute}} -RunAs32; $job | Wait-Job | Receive-Job }} else {{IEX '$functions'; Command-ToExecute}}"
     else:
         command = f"{amsi_bypass} {ps_command}"
-    
+
     nxc_logger.debug(f"Generated PS command:\n {command}\n")
 
     if obfs:
@@ -163,7 +151,7 @@ def create_ps_command(ps_command, force_ps32=False, obfs=False, custom_amsi=None
         while True:
             nxc_logger.debug(f"Obfuscation attempt: {obfs_attempts + 1}")
             obfs_command = invoke_obfuscation(command)
-            
+
             command = f'powershell.exe -exec bypass -noni -nop -w 1 -C "{replace_singles(obfs_command)}"'
             if len(command) <= 8191:
                 break
@@ -176,11 +164,11 @@ def create_ps_command(ps_command, force_ps32=False, obfs=False, custom_amsi=None
         # if we arent encoding or obfuscating anything, we quote the entire powershell in double quotes, otherwise the final powershell command will syntax error
         command = f"-enc {encode_ps_command(command)}" if encode else f'"{command}"'
         command = f"powershell.exe -noni -nop -w 1 {command}"
-        
+
         if len(command) > 8191:
             nxc_logger.error(f"Command exceeds maximum length of 8191 chars (was {len(command)}). exiting.")
             exit(1)
-            
+
     nxc_logger.debug(f"Final command: {command}")
     return command
 
@@ -429,4 +417,3 @@ def invoke_obfuscation(script_string):
     obfuscated_script = choice(invoke_options)
     nxc_logger.debug(f"Script after obfuscation: {obfuscated_script}")
     return obfuscated_script
-

nxc/loaders/moduleloader.py

@@ -46,9 +46,7 @@ def module_is_sane(self, module, module_path):
             self.logger.fail(f"{module_path} missing the on_login/on_admin_login function(s)")
             module_error = True
 
-        if module_error:
-            return False
-        return True
+        return not module_error
 
     def load_module(self, module_path):
         """Load a module, initializing it and checking that it has the proper attributes"""