Merge pull request Pennyw0rth#473 from Pennyw0rth/neff-code-improvements

nxc/connection.py

@@ -229,7 +229,8 @@ def proto_flow(self):
         else:
             self.logger.debug("Created connection object")
             self.enum_host_info()
-            if self.print_host_info() and (self.login() or (self.username == "" and self.password == "")):
+            self.print_host_info()
+            if self.login() or (self.username == "" and self.password == ""):
                 if hasattr(self.args, "module") and self.args.module:
                     self.load_modules()
                     self.logger.debug("Calling modules")

nxc/protocols/ftp.py

@@ -24,7 +24,7 @@ def proto_logger(self):
 
     def proto_flow(self):
         self.proto_logger()
-        if self.create_conn_obj() and self.enum_host_info() and self.print_host_info() and self.login():
+        if self.create_conn_obj() and self.login():
             if hasattr(self.args, "module") and self.args.module:
                 self.load_modules()
                 self.logger.debug("Calling modules")
@@ -38,11 +38,9 @@ def enum_host_info(self):
         self.logger.debug(f"Welcome result: {welcome}")
         self.remote_version = welcome.split("220", 1)[1].strip()  # strip out the extra space in the front
         self.logger.debug(f"Remote version: {self.remote_version}")
-        return True
 
     def print_host_info(self):
         self.logger.display(f"Banner: {self.remote_version}")
-        return True
 
     def create_conn_obj(self):
         self.conn = FTP()

nxc/protocols/ldap.py

@@ -312,7 +312,6 @@ def print_host_info(self):
             smbv1 = colored(f"SMBv1:{self.smbv1}", host_info_colors[2], attrs=["bold"]) if self.smbv1 else colored(f"SMBv1:{self.smbv1}", host_info_colors[3], attrs=["bold"])
             self.logger.display(f"{self.server_os}{f' x{self.os_arch}' if self.os_arch else ''} (name:{self.hostname}) (domain:{self.targetDomain}) ({signing}) ({smbv1})")
             self.logger.extra["protocol"] = "LDAP"
-        return True
 
     def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="", kdcHost="", useCache=False):
         self.username = username
@@ -694,7 +693,7 @@ def getUnixTime(self, t):
         t /= 10000000
         return t
 
-    def search(self, searchFilter, attributes, sizeLimit=0):
+    def search(self, searchFilter, attributes, sizeLimit=0) -> list:
         try:
             if self.ldapConnection:
                 self.logger.debug(f"Search Filter={searchFilter}")
@@ -714,8 +713,8 @@ def search(self, searchFilter, attributes, sizeLimit=0):
                 e.getAnswers()
             else:
                 self.logger.fail(e)
-                return False
-        return False
+                return []
+        return []
 
     def users(self):
         """

nxc/protocols/mssql.py

@@ -141,7 +141,6 @@ def enum_host_info(self):
 
     def print_host_info(self):
         self.logger.display(f"{self.server_os} (name:{self.hostname}) (domain:{self.targetDomain})")
-        return True
 
     @reconnect_mssql
     def kerberos_login(

nxc/protocols/nfs.py

@@ -69,7 +69,6 @@ def enum_host_info(self):
 
     def print_host_info(self):
         self.logger.display(f"Target supported NFS versions: ({', '.join(str(x) for x in self.nfs_versions)})")
-        return True
 
     def disconnect(self):
         """Disconnect mount and portmap if they are connected"""

nxc/protocols/rdp.py

@@ -81,11 +81,6 @@ def __init__(self, args, db, host):
 
         connection.__init__(self, args, db, host)
 
-    # def proto_flow(self):
-    #     if self.create_conn_obj():
-    #         if self.login() or (self.username == '' and self.password == ''):
-    #             if hasattr(self.args, 'module') and self.args.module:
-
     def proto_logger(self):
         import platform
         if platform.python_version() in ["3.11.5", "3.11.6", "3.12.0"]:
@@ -112,7 +107,6 @@ def print_host_info(self):
             self.logger.display(f"Probably old, doesn't not support HYBRID or HYBRID_EX ({nla})")
         else:
             self.logger.display(f"{self.server_os} (name:{self.hostname}) (domain:{self.domain}) ({nla})")
-        return True
 
     def create_conn_obj(self):
         self.target = RDPTarget(ip=self.host, domain="FAKE", port=self.port, timeout=self.args.rdp_timeout)

nxc/protocols/smb.py

@@ -312,7 +312,6 @@ def print_host_info(self):
         signing = colored(f"signing:{self.signing}", host_info_colors[0], attrs=["bold"]) if self.signing else colored(f"signing:{self.signing}", host_info_colors[1], attrs=["bold"])
         smbv1 = colored(f"SMBv1:{self.smbv1}", host_info_colors[2], attrs=["bold"]) if self.smbv1 else colored(f"SMBv1:{self.smbv1}", host_info_colors[3], attrs=["bold"])
         self.logger.display(f"{self.server_os}{f' x{self.os_arch}' if self.os_arch else ''} (name:{self.hostname}) (domain:{self.targetDomain}) ({signing}) ({smbv1})")
-        return True
 
     def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="", kdcHost="", useCache=False):
         self.logger.debug(f"KDC set to: {kdcHost}")
@@ -619,7 +618,20 @@ def gen_relay_list(self):
                     relay_list.write(self.host + "\n")
 
     @requires_admin
-    def execute(self, payload=None, get_output=False, methods=None):
+    def execute(self, payload=None, get_output=False, methods=None) -> str:
+        """
+        Executes a command on the target host using CMD.exe and the specified method(s).
+
+        Args:
+        ----
+            payload (str): The command to execute
+            get_output (bool): Whether to get the output of the command (can be useful for AV evasion)
+            methods (list): The method(s) to use for command execution
+
+        Returns:
+        -------
+            str: The output of the command
+        """
         if self.args.exec_method:
             methods = [self.args.exec_method]
         if not methods:
@@ -753,7 +765,7 @@ def execute(self, payload=None, get_output=False, methods=None):
 
             if "This script contains malicious content" in output:
                 self.logger.fail("Command execution blocked by AMSI")
-                return None
+                return ""
 
             if (self.args.execute or self.args.ps_execute):
                 self.logger.success(f"Executed command via {current_method}")
@@ -764,14 +776,29 @@ def execute(self, payload=None, get_output=False, methods=None):
             return output
         else:
             self.logger.fail(f"Execute command failed with {current_method}")
-            return False
+            return ""
 
     @requires_admin
-    def ps_execute(self, payload=None, get_output=False, methods=None, force_ps32=False, obfs=False, encode=False):
+    def ps_execute(self, payload=None, get_output=False, methods=None, force_ps32=False, obfs=False, encode=False) -> list:
+        """
+        Wrapper for executing a PowerShell command on the target host. This still uses the execute() method internally, but
+        creates a PowerShell command together with possible AMSI bypasses and other options.
+
+        Args:
+        ----
+            payload (str): The PowerShell command to execute OR the path to a file containing PowerShell commands
+            get_output (bool): Whether to get the output of the command (can be useful for AV evasion)
+            methods (list): The method(s) to use for command execution
+            force_ps32 (bool): Whether to force 32-bit PowerShell
+
+        Returns:
+        -------
+            list: A list containing the lines of the output of the command
+        """
         payload = self.args.ps_execute if not payload and self.args.ps_execute else payload
         if not payload:
             self.logger.error("No command to execute specified!")
-            return None
+            return []
 
         response = []
         obfs = obfs if obfs else self.args.obfs

nxc/protocols/ssh.py

@@ -55,7 +55,6 @@ def proto_logger(self):
 
     def print_host_info(self):
         self.logger.display(self.remote_version if self.remote_version != "Unknown SSH Version" else f"{self.remote_version}, skipping...")
-        return True
 
     def enum_host_info(self):
         if self.conn._transport.remote_version:

nxc/protocols/winrm.py

@@ -72,8 +72,6 @@ def print_host_info(self):
         self.logger.extra["port"] = self.port
         self.logger.display(f"{self.server_os} (name:{self.hostname}) (domain:{self.targetDomain})")
 
-        return True
-
     def create_conn_obj(self):
         if self.is_link_local_ipv6:
             self.logger.fail("winrm not support link-local ipv6, exiting...")

nxc/protocols/wmi.py

@@ -146,7 +146,6 @@ def print_host_info(self):
         self.logger.extra["protocol"] = "RPC"
         self.logger.extra["port"] = "135"
         self.logger.display(f"{self.server_os} (name:{self.hostname}) (domain:{self.targetDomain})")
-        return True
 
     def check_if_admin(self):
         try: