ruff: add Pylint (PL) and auto-run

nxc/connection.py

@@ -16,6 +16,7 @@
 from nxc.context import Context
 
 from impacket.dcerpc.v5 import transport
+import sys
 
 sem = BoundedSemaphore(1)
 global_failed_logins = 0
@@ -341,7 +342,7 @@ def parse_credentials(self):
                 except UnicodeDecodeError as e:
                     self.logger.error(f"{type(e).__name__}: Could not decode password file. Make sure the file only contains UTF-8 characters.")
                     self.logger.error("You can ignore non UTF-8 characters with the option '--ignore-pw-decoding'")
-                    exit(1)
+                    sys.exit(1)
             else:
                 secret.append(password)
                 cred_type.append("plaintext")

nxc/modules/add_computer.py

@@ -3,6 +3,7 @@
 import ssl
 import ldap3
 from impacket.dcerpc.v5 import samr, epm, transport
+import sys
 
 
 class NXCModule:
@@ -51,13 +52,13 @@ def options(self, context, module_options):
                 self.__computerName += "$"
         else:
             context.log.error("NAME option is required!")
-            exit(1)
+            sys.exit(1)
 
         if "PASSWORD" in module_options:
             self.__computerPassword = module_options["PASSWORD"]
         elif "PASSWORD" not in module_options and not self.__delete:
             context.log.error("PASSWORD option is required!")
-            exit(1)
+            sys.exit(1)
 
     def on_login(self, context, connection):
         self.__domain = connection.domain
@@ -89,7 +90,7 @@ def on_login(self, context, connection):
         if not self.noLDAPRequired:
             self.do_ldaps_add(connection, context)
         else:
-            exit(1)
+            sys.exit(1)
 
     def do_samr_add(self, context):
         """

nxc/modules/daclread.py

@@ -10,6 +10,7 @@
 from ldap3.protocol.formatters.formatters import format_sid
 from ldap3.utils.conv import escape_filter_chars
 from ldap3.protocol.microsoft import security_descriptor_control
+import sys
 
 OBJECT_TYPES_GUID = {}
 OBJECT_TYPES_GUID.update(SCHEMA_OBJECTS)
@@ -220,7 +221,7 @@ def options(self, context, module_options):
 
         if not module_options:
             context.log.fail("Select an option, example: -M daclread -o TARGET=Administrator ACTION=read")
-            exit(1)
+            sys.exit(1)
 
         if module_options and "TARGET" in module_options:
             if re.search(r"^(.+)\/([^\/]+)$", module_options["TARGET"]) is not None:
@@ -283,7 +284,7 @@ def on_login(self, context, connection):
                 context.log.highlight(f"Found principal SID to filter on: {self.principal_sid}")
             except Exception:
                 context.log.fail(f"Principal SID not found in LDAP ({_lookedup_principal})")
-                exit(1)
+                sys.exit(1)
 
         # Searching for the targets SID and their Security Decriptors
         # If there is only one target
@@ -298,7 +299,7 @@ def on_login(self, context, connection):
                 context.log.highlight(f"Target principal found in LDAP ({self.target_principal[0]})")
             except Exception:
                 context.log.fail(f"Target SID not found in LDAP ({self.target_sAMAccountName})")
-                exit(1)
+                sys.exit(1)
 
             if self.action == "read":
                 self.read(context)
@@ -332,7 +333,6 @@ def on_login(self, context, connection):
     def read(self, context):
         parsed_dacl = self.parse_dacl(context, self.principal_security_descriptor["Dacl"])
         self.print_parsed_dacl(context, parsed_dacl)
-        return
 
     # Permits to export the DACL of the targets
     # This function is called before any writing action (write, remove or restore)
@@ -375,7 +375,7 @@ def search_target_principal_security_descriptor(self, context, connection):
             self.target_principal = target[0]
         except Exception:
             context.log.fail(f"Principal not found in LDAP ({_lookedup_principal}), probably an LDAP session issue.")
-            exit(0)
+            sys.exit(0)
 
     # Attempts to retieve the SID and Distinguisehd Name from the sAMAccountName
     # Not used for the moment

nxc/modules/find-computer.py

@@ -3,6 +3,7 @@
 from nxc.logger import nxc_logger
 from impacket.ldap.ldap import LDAPSearchError
 from impacket.ldap.ldapasn1 import SearchResultEntry
+import sys
 
 
 class NXCModule:
@@ -32,7 +33,7 @@ def options(self, context, module_options):
             self.TEXT = module_options["TEXT"]
         else:
             context.log.error("TEXT option is required!")
-            exit(1)
+            sys.exit(1)
 
     def on_login(self, context, connection):
         search_filter = f"(&(objectCategory=computer)(&(|(operatingSystem=*{self.TEXT}*))(name=*{self.TEXT}*)))"

nxc/modules/group_members.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 from impacket.ldap import ldapasn1 as ldapasn1_impacket
+import sys
 
 
 class NXCModule:
@@ -31,7 +32,7 @@ def options(self, context, module_options):
             self.GROUP = module_options["GROUP"]
         else:
             context.log.error("GROUP option is required!")
-            exit(1)
+            sys.exit(1)
 
     def on_login(self, context, connection):
         # First look up the SID of the group passed in

nxc/modules/groupmembership.py

@@ -2,6 +2,7 @@
 
 from impacket.ldap import ldapasn1 as ldapasn1_impacket
 from impacket.ldap import ldap as ldap_impacket
+import sys
 
 
 class NXCModule:
@@ -25,11 +26,11 @@ def options(self, context, module_options):
         if "USER" in module_options:
             if module_options["USER"] == "":
                 context.log.fail("Invalid value for USER option!")
-                exit(1)
+                sys.exit(1)
             self.user = module_options["USER"]
         else:
             context.log.fail("Missing USER option, use --options to list available parameters")
-            exit(1)
+            sys.exit(1)
 
     def on_login(self, context, connection):
         """Concurrent. Required if on_admin_login is not present. This gets called on each authenticated connection"""

nxc/modules/keepass_trigger.py

@@ -97,12 +97,12 @@ def options(self, context, module_options):
                 "ALL",
             ]:
                 context.log.fail("Unrecognized action, use --options to list available parameters")
-                exit(1)
+                sys.exit(1)
             else:
                 self.action = module_options["ACTION"]
         else:
             context.log.fail("Missing ACTION option, use --options to list available parameters")
-            exit(1)
+            sys.exit(1)
 
         if "KEEPASS_CONFIG_PATH" in module_options:
             self.keepass_config_path = module_options["KEEPASS_CONFIG_PATH"]
@@ -119,7 +119,7 @@ def options(self, context, module_options):
         if "PSH_EXEC_METHOD" in module_options:
             if module_options["PSH_EXEC_METHOD"] not in ["ENCODE", "PS1"]:
                 context.log.fail("Unrecognized powershell execution method, use --options to list available parameters")
-                exit(1)
+                sys.exit(1)
             else:
                 self.powershell_exec_method = module_options["PSH_EXEC_METHOD"]
 

nxc/modules/ldap-checker.py

@@ -11,6 +11,7 @@
 from asyauth.common.credentials.kerberos import KerberosCredential
 
 from asysocks.unicomm.common.target import UniTarget, UniProto
+import sys
 
 
 class NXCModule:
@@ -41,7 +42,7 @@ async def run_ldaps_noEPA(target, credential):
             _, err = await ldapsClientConn.connect()
             if err is not None:
                 context.log.fail("ERROR while connecting to " + str(connection.domain) + ": " + str(err))
-                exit()
+                sys.exit()
             _, err = await ldapsClientConn.bind()
             if "data 80090346" in str(err):
                 return True  # channel binding IS enforced
@@ -64,7 +65,7 @@ async def run_ldaps_withEPA(target, credential):
             _, err = await ldapsClientConn.connect()
             if err is not None:
                 context.log.fail("ERROR while connecting to " + str(connection.domain) + ": " + str(err))
-                exit()
+                sys.exit()
             # forcing a miscalculation of the "Channel Bindings" av pair in Type 3 NTLM message
             ldapsClientConn.cb_data = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
             _, err = await ldapsClientConn.bind()
@@ -125,7 +126,7 @@ async def run_ldap(target, credential):
                     return True  # because LDAP server signing requirements ARE enforced
                 elif ("data 52e") in str(err):
                     context.log.fail("Not connected... exiting")
-                    exit()
+                    sys.exit()
                 elif err is None:
                     return False
                 return None
@@ -162,7 +163,7 @@ async def run_ldap(target, credential):
             context.log.fail("LDAP Signing IS Enforced")
         else:
             context.log.fail("Connection fail, exiting now")
-            exit()
+            sys.exit()
 
         if DoesLdapsCompleteHandshake(connection.host) is True:
             target = MSLDAPTarget(connection.host, 636, UniProto.CLIENT_SSL_TCP, hostname=connection.hostname, domain=connection.domain, dc_ip=connection.domain)
@@ -177,6 +178,6 @@ async def run_ldap(target, credential):
                 context.log.fail('LDAPS Channel Binding is set to "Required"')
             else:
                 context.log.fail("\nSomething went wrong...")
-                exit()
+                sys.exit()
         else:
             context.log.fail(connection.domain + " - cannot complete TLS handshake, cert likely not configured")

nxc/modules/procdump.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 # prdocdump module for nxc python3
-# author: github.com/mpgn
 # thanks to pixis (@HackAndDo) for making it pretty l33t :)
 # v0.4
 

nxc/modules/subnets.py

@@ -2,6 +2,7 @@
 
 from impacket.ldap import ldapasn1 as ldapasn1_impacket
 from impacket.ldap.ldap import LDAPSearchError
+import sys
 
 
 def search_res_entry_to_dict(results):
@@ -56,7 +57,7 @@ def on_login(self, context, connection):
             )
         except LDAPSearchError as e:
             context.log.fail(str(e))
-            exit()
+            sys.exit()
 
         for site in list_sites:
             if isinstance(site, ldapasn1_impacket.SearchResultEntry) is not True:

nxc/netexec.py

@@ -16,7 +16,7 @@
 from nxc.config import nxc_config, nxc_workspace, config_log, ignore_opsec
 from concurrent.futures import ThreadPoolExecutor, as_completed
 import asyncio
-import nxc.helpers.powershell as powershell
+from nxc.helpers import powershell
 import shutil
 import os
 from os.path import exists

nxc/nxcdb.py

@@ -57,7 +57,6 @@ def write_list(filename, entries):
     with open(os.path.expanduser(filename), "w") as export_file:
         for line in entries:
             export_file.write(line + "\n")
-    return
 
 
 def complete_import(text, line):

nxc/protocols/ftp/database.py

@@ -10,6 +10,7 @@
     NoSuchTableError,
 )
 from nxc.logger import nxc_logger
+import sys
 
 
 class database:
@@ -79,7 +80,7 @@ def reflect_tables(self):
                     [-] Optionally save the old DB data (`cp {self.db_path} ~/nxc_{self.protocol.lower()}.bak`)
                     [-] Then remove the {self.protocol} DB (`rm -f {self.db_path}`) and run nxc to initialize the new DB"""
                 )
-                exit()
+                sys.exit()
 
     def shutdown_db(self):
         try:

nxc/protocols/ldap.py

@@ -1071,7 +1071,6 @@ def trusted_for_delegation(self):
                 self.logger.highlight(value[0])
         else:
             self.logger.fail("No entries found!")
-        return
 
     def password_not_required(self):
         # Building the search filter
@@ -1205,7 +1204,6 @@ def admin_count(self):
                 self.logger.highlight(value[0])
         else:
             self.logger.fail("No entries found!")
-        return
 
     def gmsa(self):
         self.logger.display("Getting GMSA Passwords")

nxc/protocols/ldap/database.py

@@ -9,6 +9,7 @@
     NoSuchTableError,
 )
 from nxc.logger import nxc_logger
+import sys
 
 
 class database:
@@ -59,7 +60,7 @@ def reflect_tables(self):
                     [-] Optionally save the old DB data (`cp {self.db_path} ~/nxc_{self.protocol.lower()}.bak`)
                     [-] Then remove the nxc {self.protocol} DB (`rm -f {self.db_path}`) and run nxc to initialize the new DB"""
                 )
-                exit()
+                sys.exit()
 
     def shutdown_db(self):
         try:

nxc/protocols/mssql/database.py

@@ -12,6 +12,7 @@
 from sqlalchemy.exc import SAWarning
 import warnings
 from nxc.logger import nxc_logger
+import sys
 
 # if there is an issue with SQLAlchemy and a connection cannot be cleaned up properly it spews out annoying warnings
 warnings.filterwarnings("ignore", category=SAWarning)
@@ -82,7 +83,7 @@ def reflect_tables(self):
                     [-] Optionally save the old DB data (`cp {self.db_path} ~/nxc_{self.protocol.lower()}.bak`)
                     [-] Then remove the {self.protocol} DB (`rm -f {self.db_path}`) and run nxc to initialize the new DB"""
                 )
-                exit()
+                sys.exit()
 
     def shutdown_db(self):
         try:

nxc/protocols/rdp/database.py

@@ -9,6 +9,7 @@
     NoSuchTableError,
 )
 from nxc.logger import nxc_logger
+import sys
 
 
 class database:
@@ -61,7 +62,7 @@ def reflect_tables(self):
                     [-] Optionally save the old DB data (`cp {self.db_path} ~/nxc_{self.protocol.lower()}.bak`)
                     [-] Then remove the {self.protocol} DB (`rm -f {self.db_path}`) and run nxc to initialize the new DB"""
                 )
-                exit()
+                sys.exit()
 
     def shutdown_db(self):
         try:

nxc/protocols/smb.py

@@ -628,7 +628,6 @@ def check_if_admin(self):
             except scmr.DCERPCException:
                 self.admin_privs = False
                 pass
-        return
 
     def gen_relay_list(self):
         if self.server_os.lower().find("windows") != -1 and self.signing is False:

nxc/protocols/smb/database.py

@@ -15,6 +15,7 @@
 from sqlalchemy.orm import sessionmaker, scoped_session
 
 from nxc.logger import nxc_logger
+import sys
 
 # if there is an issue with SQLAlchemy and a connection cannot be cleaned up properly it spews out annoying warnings
 warnings.filterwarnings("ignore", category=SAWarning)
@@ -196,7 +197,7 @@ def reflect_tables(self):
                     [-] Optionally save the old DB data (`cp {self.db_path} ~/nxc_{self.protocol.lower()}.bak`)
                     [-] Then remove the {self.protocol} DB (`rm -f {self.db_path}`) and run nxc to initialize the new DB"""
                 )
-                exit()
+                sys.exit()
 
     def shutdown_db(self):
         try:

nxc/protocols/smb/smbspider.py

@@ -159,7 +159,6 @@ def dir_list(self, files, path):
             if self.content and not result.is_directory():
                 self.search_content(path, result)
 
-        return
 
     def search_content(self, path, result):
         path = path.replace("*", "")